content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.carbondata.core.range;
  
  import java.io.Serializable;
  import java.nio.charset.Charset;
  import java.util.Set;
  
  import org.apache.carbondata.core.metadata.datatype.DataType;
  import org.apache.carbondata.core.util.ByteUtil;
  import org.apache.carbondata.core.util.DataTypeUtil;
  import org.apache.carbondata.core.util.comparator.SerializableComparator;
  
  import static org.apache.carbondata.core.constants.CarbonCommonConstants.DEFAULT_CHARSET;
  
  /**
   * This class prepares a tree for pruning using min-max of block
   */
  public class BlockMinMaxTree implements Serializable {
  
    private MinMaxNode root;
  
    private final boolean isPrimitiveAndNotDate;
    private final boolean isDimensionColumn;
    private final DataType joinDataType;
    private final SerializableComparator comparator;
  
    public BlockMinMaxTree(boolean isPrimitiveAndNotDate, boolean isDimensionColumn,
        DataType joinDataType, SerializableComparator comparator) {
      this.isPrimitiveAndNotDate = isPrimitiveAndNotDate;
      this.isDimensionColumn = isDimensionColumn;
      this.joinDataType = joinDataType;
      this.comparator = comparator;
    }
  
    public MinMaxNode getRoot() {
      return root;
    }
  
    public void insert(MinMaxNode newMinMaxNode) {
      root = insert(getRoot(), newMinMaxNode);
    }
  
    private MinMaxNode insert(MinMaxNode root, MinMaxNode newMinMaxNode) {
      /* 1. check if the root null, then insert and make new node
       * 2. check if the new node completely overlaps with the root, where minCompare and maxCompare
       * both are zero, if yes add the filepaths and return
       * 3. if root is less than new node, check if the root has right subtree,
       *    if(yes) {
       *       replace the right node with the newnode's min and max based on comparison and then
       *        call insert with right node as new root and newnode
       *         insert(root.getRight, newnode)
       *     } else {
       *       make the new node as right node and set right node and return
       *     }
       * 4. if root is more than new node, check if the root has left subtree,
       *    if(yes) {
       *       replace the left node with the newnode's min and max based on comparison and then
       *        call insert with left node as new root and newnode
       *         insert(root.getLeft, newnode)
       *     } else {
       *       make the new node as left node and set left node and return
       *     }
       * */
      if (root == null) {
        root = newMinMaxNode;
        return root;
      }
  
      if (compareNodesBasedOnMinMax(root, newMinMaxNode) == 0) {
        root.addFilePats(newMinMaxNode.getFilePaths());
        return root;
      }
  
      if (compareNodesBasedOnMinMax(root, newMinMaxNode) < 0) {
        if (root.getRightSubTree() == null) {
          root.setRightSubTree(newMinMaxNode);
          root.setRightSubTreeMax(newMinMaxNode.getMax());
          root.setRightSubTreeMin(newMinMaxNode.getMin());
        } else {
          if (compareMinMax(root.getRightSubTreeMax(), newMinMaxNode.getMax()) < 0) {
            root.setRightSubTreeMax(newMinMaxNode.getMax());
          }
          if (compareMinMax(root.getRightSubTreeMin(), newMinMaxNode.getMin()) > 0) {
            root.setRightSubTreeMin(newMinMaxNode.getMin());
          }
          insert(root.getRightSubTree(), newMinMaxNode);
        }
      } else {
        if (root.getLeftSubTree() == null) {
          root.setLeftSubTree(newMinMaxNode);
          root.setLeftSubTreeMax(newMinMaxNode.getMax());
          root.setLeftSubTreeMin(newMinMaxNode.getMin());
        } else {
          if (compareMinMax(root.getLeftSubTreeMax(), newMinMaxNode.getMax()) < 0) {
            root.setLeftSubTreeMax(newMinMaxNode.getMax());
          }
          if (compareMinMax(root.getLeftSubTreeMin(), newMinMaxNode.getMin()) > 0) {
            root.setLeftSubTreeMin(newMinMaxNode.getMin());
          }
          insert(root.getLeftSubTree(), newMinMaxNode);
        }
      }
      return root;
    }
  
    private int compareNodesBasedOnMinMax(MinMaxNode root, MinMaxNode newMinMaxNode) {
      int minCompare = compareMinMax(root.getMin(), newMinMaxNode.getMin());
      int maxCompare = compareMinMax(root.getMax(), newMinMaxNode.getMax());
      if (minCompare == 0) {
        return maxCompare;
      } else {
        return minCompare;
      }
    }
  
    private int compareMinMax(Object key1, Object key2) {
      if (isDimensionColumn) {
        if (isPrimitiveAndNotDate) {
          return comparator.compare(key1, key2);
        } else {
          return ByteUtil.UnsafeComparer.INSTANCE
              .compareTo(key1.toString().getBytes(Charset.forName(DEFAULT_CHARSET)),
                  key2.toString().getBytes(Charset.forName(DEFAULT_CHARSET)));
        }
      } else {
        return comparator.compare(key1, key2);
      }
    }
  
    /**
     * This method returns the list of carbondata files where the input fieldValue might present
     */
    public Set<String> getMatchingFiles(byte[] fieldValue, Set<String> matchedFilesSet) {
      getMatchingFiles(getRoot(), fieldValue, matchedFilesSet);
      return matchedFilesSet;
    }
  
    private void getMatchingFiles(MinMaxNode root, byte[] fieldValue, Set<String> matchedFilesSet) {
      Object data;
      if (root == null) {
        return;
      }
      if (isDimensionColumn) {
        if (isPrimitiveAndNotDate) {
          data = DataTypeUtil.getDataBasedOnDataTypeForNoDictionaryColumn(fieldValue, joinDataType);
        } else {
          // for string comparator is Unsafe comparator, so cannot include in common code
          getMatchingFilesForString(root, fieldValue, matchedFilesSet);
          return;
        }
      } else {
        data = DataTypeUtil.getMeasureObjectFromDataType(fieldValue, joinDataType);
      }
  
      if (comparator.compare(data, root.getMax()) <= 0
          && comparator.compare(data, root.getMin()) >= 0) {
        matchedFilesSet.addAll(root.getFilePaths());
      }
  
      if (root.getLeftSubTree() != null && comparator.compare(data, root.getLeftSubTreeMax()) <= 0
          && comparator.compare(data, root.getLeftSubTreeMin()) >= 0) {
        getMatchingFiles(root.getLeftSubTree(), fieldValue, matchedFilesSet);
      }
  
      if (root.getRightSubTree() != null && comparator.compare(data, root.getRightSubTreeMax()) <= 0
          && comparator.compare(data, root.getRightSubTreeMin()) >= 0) {
        getMatchingFiles(root.getRightSubTree(), fieldValue, matchedFilesSet);
      }
    }
  
    private void getMatchingFilesForString(MinMaxNode root, byte[] fieldValue,
        Set<String> matchedFilesSet) {
      if (root == null) {
        return;
      }
      if (ByteUtil.UnsafeComparer.INSTANCE
          .compareTo(fieldValue, root.getMin().toString().getBytes(Charset.forName(DEFAULT_CHARSET)))
          >= 0 && ByteUtil.UnsafeComparer.INSTANCE
          .compareTo(fieldValue, root.getMax().toString().getBytes(Charset.forName(DEFAULT_CHARSET)))
          <= 0) {
        matchedFilesSet.addAll(root.getFilePaths());
      }
      if (root.getLeftSubTree() != null && ByteUtil.UnsafeComparer.INSTANCE.compareTo(fieldValue,
          root.getLeftSubTreeMin().toString().getBytes(Charset.forName(DEFAULT_CHARSET))) >= 0 &&
          ByteUtil.UnsafeComparer.INSTANCE.compareTo(fieldValue,
              root.getLeftSubTreeMax().toString().getBytes(Charset.forName(DEFAULT_CHARSET))) <= 0) {
        getMatchingFilesForString(root.getLeftSubTree(), fieldValue, matchedFilesSet);
      }
  
      if (root.getRightSubTree() != null && ByteUtil.UnsafeComparer.INSTANCE.compareTo(fieldValue,
          root.getRightSubTreeMin().toString().getBytes(Charset.forName(DEFAULT_CHARSET))) >= 0 &&
          ByteUtil.UnsafeComparer.INSTANCE.compareTo(fieldValue,
              root.getRightSubTreeMax().toString().getBytes(Charset.forName(DEFAULT_CHARSET))) <= 0) {
        getMatchingFilesForString(root.getRightSubTree(), fieldValue, matchedFilesSet);
      }
    }
  }
  