/*
 * 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.geo.scan.filter.executor;

import java.util.BitSet;
import java.util.List;
import java.util.Map;

import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.scan.expression.Expression;
import org.apache.carbondata.core.scan.filter.GenericQueryType;
import org.apache.carbondata.core.scan.filter.executer.RowLevelFilterExecutorImpl;
import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.DimColumnResolvedFilterInfo;
import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.MeasureColumnResolvedFilterInfo;
import org.apache.carbondata.core.util.DataTypeUtil;
import org.apache.carbondata.geo.scan.expression.PolygonExpression;

/**
 * Polygon filter executor. Prunes Blocks and Blocklets based on the selected ranges of polygon.
 */
public class PolygonFilterExecutorImpl extends RowLevelFilterExecutorImpl {
  public PolygonFilterExecutorImpl(List<DimColumnResolvedFilterInfo> dimColEvaluatorInfoList,
      List<MeasureColumnResolvedFilterInfo> msrColEvalutorInfoList, Expression exp,
      AbsoluteTableIdentifier tableIdentifier, SegmentProperties segmentProperties,
      Map<Integer, GenericQueryType> complexDimensionInfoMap) {
    super(dimColEvaluatorInfoList, msrColEvalutorInfoList, exp, tableIdentifier, segmentProperties,
        complexDimensionInfoMap);
  }

  private int getNearestRangeIndex(List<Long[]> ranges, long searchForNumber) {
    Long[] range;
    int low = 0, mid = 0, high = ranges.size() - 1;
    while (low <= high) {
      mid = low + ((high - low) / 2);
      range = ranges.get(mid);
      if (searchForNumber >= range[0]) {
        if (searchForNumber <= range[1]) {
          // Return the range index if the number is between min and max values of the range
          return mid;
        } else {
          // Number is bigger than this range's min and max. Search on the right side of the range
          low = mid + 1;
        }
      } else {
        // Number is smaller than this range's min and max. Search on the left side of the range
        high = mid - 1;
      }
    }
    return mid;
  }

  /**
   * Checks if the current block or blocklet needs to be scanned
   * @param maxValue Max value in the current block or blocklet
   * @param minValue Min value in te current block or blocklet
   * @return True or False  True if current block or blocket needs to be scanned. Otherwise False.
   */
  private boolean isScanRequired(byte[] maxValue, byte[] minValue) {
    PolygonExpression polygon = (PolygonExpression) exp;
    List<Long[]> ranges = polygon.getRanges();
    Long min =
        (Long) DataTypeUtil.getDataBasedOnDataTypeForNoDictionaryColumn(minValue, DataTypes.LONG);
    Long max =
        (Long) DataTypeUtil.getDataBasedOnDataTypeForNoDictionaryColumn(maxValue, DataTypes.LONG);

    // Find the nearest possible range index for both the min and max values. If a value do not
    // exist in the any of the range, get the preceding range index where it fits best
    int startIndex = getNearestRangeIndex(ranges, min);
    int endIndex = getNearestRangeIndex(ranges, max);
    if (endIndex > startIndex) {
       // Multiple ranges fall between min and max. Need to scan this block or blocklet
      return true;
    }
    // Got same index for both min and max values.
    Long[] oneRange = ranges.subList(startIndex, endIndex + 1).get(0);
    if ((min >= oneRange[0] && min <= oneRange[1]) || (max >= oneRange[0] && max <= oneRange[1])) {
      // Either min or max is within the range
      return true;
    }
    // No range between min and max values. Scan can be avoided for this block or blocklet
    return false;
  }

  @Override
  public BitSet isScanRequired(byte[][] blockMaxValue, byte[][] blockMinValue,
      boolean[] isMinMaxSet) {
    assert (exp instanceof PolygonExpression);
    int dimIndex = dimensionChunkIndex[0];
    BitSet bitSet = new BitSet(1);
    if (isMinMaxSet[dimIndex] && isScanRequired(blockMaxValue[dimIndex], blockMinValue[dimIndex])) {
      bitSet.set(0);
    }
    return bitSet;
  }
}