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.util;
  
  import java.util.Arrays;
  
  import org.apache.carbondata.core.metadata.datatype.DataTypes;
  import org.apache.carbondata.core.metadata.encoder.Encoding;
  import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
  import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
  import org.apache.carbondata.core.scan.expression.ColumnExpression;
  import org.apache.carbondata.core.scan.expression.Expression;
  import org.apache.carbondata.core.scan.expression.LiteralExpression;
  import org.apache.carbondata.core.scan.expression.conditional.GreaterThanEqualToExpression;
  import org.apache.carbondata.core.scan.expression.conditional.GreaterThanExpression;
  import org.apache.carbondata.core.scan.expression.conditional.LessThanEqualToExpression;
  import org.apache.carbondata.core.scan.expression.conditional.NotEqualsExpression;
  import org.apache.carbondata.core.scan.expression.logical.AndExpression;
  import org.apache.carbondata.core.scan.expression.logical.OrExpression;
  import org.apache.carbondata.core.scan.expression.logical.RangeExpression;
  import org.apache.carbondata.core.scan.expression.logical.TrueExpression;
  import org.apache.carbondata.core.scan.filter.executer.RangeValueFilterExecutorImpl;
  import org.apache.carbondata.core.scan.filter.intf.FilterOptimizer;
  import org.apache.carbondata.core.scan.filter.optimizer.RangeFilterOptimizer;
  import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.DimColumnResolvedFilterInfo;
  
  import mockit.Deencapsulation;
  import mockit.MockUp;
  import org.junit.AfterClass;
  import org.junit.Assert;
  import org.junit.BeforeClass;
  import org.junit.Test;
  
  /* Test Cases for Range Filter */
  public class RangeFilterProcessorTest {
  
    public static DimColumnResolvedFilterInfo dimColumnResolvedFilterInfo =
        new DimColumnResolvedFilterInfo();
  
    @BeforeClass public static void setUp() throws Exception {
      ColumnSchema columnSchema = new ColumnSchema();
      columnSchema.setDataType(DataTypes.STRING);
      CarbonDimension carbonDimension = new CarbonDimension(columnSchema, 0, 0, 0);
      dimColumnResolvedFilterInfo.setDimension(carbonDimension);
    }
  
    public boolean checkBothTrees(Expression a, Expression b) {
  
      if (null == a && null == b) {
        return true;
      }
  
      if (null == a) return false;
      if (null == b) return false;
  
      if ((a != null) && (b != null)) {
        return ((a.getClass() == b.getClass()) && (
            (a.getChildren().size() == 0 && b.getChildren().size() == 0) || (
                a.getChildren().size() == 2 && b.getChildren().size() == 2 && checkBothTrees(
                    a.getChildren().get(0), b.getChildren().get(0)) && checkBothTrees(
                    a.getChildren().get(1), b.getChildren().get(1)))));
      }
      return false;
    }
  
    @Test public void createFilterTree() {
  
      Expression inputFilter;
      boolean result = false;
      ColumnExpression cola = new ColumnExpression("a", DataTypes.STRING);
      cola.setDimension(true);
  
      ColumnSchema empColumnSchema = new ColumnSchema();
      empColumnSchema.setColumnName("empNameCol");
      empColumnSchema.setColumnUniqueId("empNameCol");
      empColumnSchema.setDimensionColumn(true);
      empColumnSchema.setEncodingList(Arrays.asList(Encoding.DICTIONARY));
      empColumnSchema.setDataType(DataTypes.STRING);
      CarbonDimension empDimension = new CarbonDimension(empColumnSchema, 0, 0, 0);
      cola.setDimension(empDimension);
  
      Expression greaterThan =
          new GreaterThanEqualToExpression(cola, new LiteralExpression("11", DataTypes.STRING));
  
      ColumnExpression colb = new ColumnExpression("a", DataTypes.STRING);
      colb.setDimension(true);
      colb.setDimension(empDimension);
      Expression lessThan =
          new LessThanEqualToExpression(colb, new LiteralExpression("20", DataTypes.STRING));
      inputFilter = new AndExpression(greaterThan, lessThan);
  
      Expression output = new AndExpression(new RangeExpression(
          new GreaterThanEqualToExpression(new ColumnExpression("a", DataTypes.STRING),
              new LiteralExpression("11", DataTypes.STRING)),
          new LessThanEqualToExpression(new ColumnExpression("a", DataTypes.STRING),
              new LiteralExpression("20", DataTypes.STRING))), new TrueExpression(null));
      FilterOptimizer rangeFilterOptimizer =
          new RangeFilterOptimizer(inputFilter);
      rangeFilterOptimizer.optimizeFilter();
      result = checkBothTrees(inputFilter, output);
      Assert.assertTrue(result);
  
    }
  
    @Test public void createFilterTree_noChange() {
  
      // Build 2nd Tree no change a < 5 and a > 20
  
      Expression inputFilter;
      boolean result = false;
      ColumnExpression cola = new ColumnExpression("a", DataTypes.STRING);
      cola.setDimension(true);
  
      ColumnSchema empColumnSchema = new ColumnSchema();
      empColumnSchema.setColumnName("empNameCol");
      empColumnSchema.setColumnUniqueId("empNameCol");
      empColumnSchema.setDimensionColumn(true);
      empColumnSchema.setEncodingList(Arrays.asList(Encoding.DICTIONARY));
      empColumnSchema.setDataType(DataTypes.STRING);
      CarbonDimension empDimension = new CarbonDimension(empColumnSchema, 0, 0, 0);
      cola.setDimension(empDimension);
  
      Expression greaterThan =
          new GreaterThanEqualToExpression(cola, new LiteralExpression("20", DataTypes.STRING));
  
      ColumnExpression colb = new ColumnExpression("a", DataTypes.STRING);
      colb.setDimension(true);
      colb.setDimension(empDimension);
      Expression lessThan =
          new LessThanEqualToExpression(colb, new LiteralExpression("05", DataTypes.STRING));
      inputFilter = new AndExpression(greaterThan, lessThan);
  
      Expression output = new AndExpression(
          new GreaterThanEqualToExpression(new ColumnExpression("a", DataTypes.STRING),
              new LiteralExpression("20", DataTypes.STRING)),
          new LessThanEqualToExpression(new ColumnExpression("a", DataTypes.STRING),
              new LiteralExpression("05", DataTypes.STRING)));
      FilterOptimizer rangeFilterOptimizer =
          new RangeFilterOptimizer(inputFilter);
      rangeFilterOptimizer.optimizeFilter();
      result = checkBothTrees(inputFilter, output);
      // no change
      Assert.assertTrue(result);
    }
  
    @Test public void createFilterTree_flavor1() {
  
      // Build 3rd BTree a >= '11' and a > '12' and a <= '20' and a <= '15'
  
      Expression inputFilter;
      boolean result = false;
  
      ColumnSchema empColumnSchema = new ColumnSchema();
      empColumnSchema.setColumnName("a");
      empColumnSchema.setColumnUniqueId("a");
      empColumnSchema.setDimensionColumn(true);
      empColumnSchema.setEncodingList(Arrays.asList(Encoding.DICTIONARY));
      empColumnSchema.setDataType(DataTypes.STRING);
      CarbonDimension empDimension = new CarbonDimension(empColumnSchema, 0, 0, 0);
  
      ColumnExpression cola1 = new ColumnExpression("a", DataTypes.STRING);
      cola1.setDimension(true);
      cola1.setDimension(empDimension);
  
      ColumnExpression cola2 = new ColumnExpression("a", DataTypes.STRING);
      cola2.setDimension(true);
      cola2.setDimension(empDimension);
  
      ColumnExpression cola3 = new ColumnExpression("a", DataTypes.STRING);
      cola3.setDimension(true);
      cola3.setDimension(empDimension);
  
      ColumnExpression cola4 = new ColumnExpression("a", DataTypes.STRING);
      cola4.setDimension(true);
      cola4.setDimension(empDimension);
  
      Expression lessThan1 =
          new LessThanEqualToExpression(cola1, new LiteralExpression("15", DataTypes.STRING));
      Expression lessThan2 =
          new LessThanEqualToExpression(cola2, new LiteralExpression("20", DataTypes.STRING));
      Expression greaterThan1 =
          new GreaterThanExpression(cola3, new LiteralExpression("12", DataTypes.STRING));
      Expression greaterThan2 =
          new GreaterThanEqualToExpression(cola4, new LiteralExpression("11", DataTypes.STRING));
  
      Expression And1 = new AndExpression(new NotEqualsExpression(null, null), greaterThan2);
      Expression And2 = new AndExpression(And1, greaterThan1);
      Expression And3 = new AndExpression(And2, lessThan2);
      inputFilter = new AndExpression(And3, lessThan1);
  
      // Build The output
  
      ColumnExpression colb1 = new ColumnExpression("a", DataTypes.STRING);
      cola1.setDimension(true);
      cola1.setDimension(empDimension);
  
      ColumnExpression colb2 = new ColumnExpression("a", DataTypes.STRING);
      cola2.setDimension(true);
      cola2.setDimension(empDimension);
  
      Expression greaterThanb1 =
          new GreaterThanExpression(cola3, new LiteralExpression("12", DataTypes.STRING));
  
      Expression lessThanb1 =
          new LessThanEqualToExpression(cola1, new LiteralExpression("15", DataTypes.STRING));
  
      Expression Andb1 =
          new AndExpression(new NotEqualsExpression(null, null), new TrueExpression(null));
  
      Expression Andb2 = new AndExpression(Andb1, new RangeExpression(greaterThanb1, lessThanb1));
      Expression Andb3 = new AndExpression(Andb2, new TrueExpression(null));
  
      FilterOptimizer rangeFilterOptimizer =
          new RangeFilterOptimizer(inputFilter);
      rangeFilterOptimizer.optimizeFilter();
      result = checkBothTrees(inputFilter, new AndExpression(Andb3, new TrueExpression(null)));
      // no change
      Assert.assertTrue(result);
    }
  
    @Test public void createFilterTree_flavor2() {
  
      // Build 3rd BTree a >= '11' or a > '12' or a <= '20' or a <= '15'
  
      Expression inputFilter;
      boolean result = false;
  
      ColumnSchema empColumnSchema = new ColumnSchema();
      empColumnSchema.setColumnName("a");
      empColumnSchema.setColumnUniqueId("a");
      empColumnSchema.setDimensionColumn(true);
      empColumnSchema.setEncodingList(Arrays.asList(Encoding.DICTIONARY));
      empColumnSchema.setDataType(DataTypes.STRING);
      CarbonDimension empDimension = new CarbonDimension(empColumnSchema, 0, 0, 0);
  
      ColumnExpression cola1 = new ColumnExpression("a", DataTypes.STRING);
      cola1.setDimension(true);
      cola1.setDimension(empDimension);
  
      ColumnExpression cola2 = new ColumnExpression("a", DataTypes.STRING);
      cola2.setDimension(true);
      cola2.setDimension(empDimension);
  
      ColumnExpression cola3 = new ColumnExpression("a", DataTypes.STRING);
      cola3.setDimension(true);
      cola3.setDimension(empDimension);
  
      ColumnExpression cola4 = new ColumnExpression("a", DataTypes.STRING);
      cola4.setDimension(true);
      cola4.setDimension(empDimension);
  
      Expression lessThan1 =
          new LessThanEqualToExpression(cola1, new LiteralExpression("15", DataTypes.STRING));
      Expression lessThan2 =
          new LessThanEqualToExpression(cola2, new LiteralExpression("20", DataTypes.STRING));
      Expression greaterThan1 =
          new GreaterThanExpression(cola3, new LiteralExpression("12", DataTypes.STRING));
      Expression greaterThan2 =
          new GreaterThanEqualToExpression(cola4, new LiteralExpression("11", DataTypes.STRING));
  
      Expression Or1 = new OrExpression(new NotEqualsExpression(null, null), greaterThan2);
      Expression Or2 = new OrExpression(Or1, greaterThan1);
      Expression Or3 = new OrExpression(Or2, lessThan2);
      inputFilter = new OrExpression(Or3, lessThan1);
  
      // Build The output
  
      ColumnExpression colb1 = new ColumnExpression("a", DataTypes.STRING);
      cola1.setDimension(true);
      cola1.setDimension(empDimension);
  
      ColumnExpression colb2 = new ColumnExpression("a", DataTypes.STRING);
      cola2.setDimension(true);
      cola2.setDimension(empDimension);
  
      ColumnExpression colb3 = new ColumnExpression("a", DataTypes.STRING);
      cola3.setDimension(true);
      cola3.setDimension(empDimension);
  
      ColumnExpression colb4 = new ColumnExpression("a", DataTypes.STRING);
      cola4.setDimension(true);
      cola4.setDimension(empDimension);
  
      Expression lessThanb1 =
          new LessThanEqualToExpression(colb1, new LiteralExpression("15", DataTypes.STRING));
      Expression lessThanb2 =
          new LessThanEqualToExpression(colb2, new LiteralExpression("20", DataTypes.STRING));
      Expression greaterThanb1 =
          new GreaterThanExpression(colb3, new LiteralExpression("12", DataTypes.STRING));
      Expression greaterThanb2 =
          new GreaterThanEqualToExpression(colb4, new LiteralExpression("11", DataTypes.STRING));
  
      Expression Orb1 = new OrExpression(new NotEqualsExpression(null, null), greaterThanb2);
      Expression Orb2 = new OrExpression(Orb1, greaterThanb1);
      Expression Orb3 = new OrExpression(Orb2, lessThanb2);
  
      FilterOptimizer rangeFilterOptimizer =
          new RangeFilterOptimizer(inputFilter);
      rangeFilterOptimizer.optimizeFilter();
      result = checkBothTrees(inputFilter, new OrExpression(Orb3, lessThanb1));
      // no change
      Assert.assertTrue(result);
    }
  
    @Test public void checkIsScanRequired1() {
      byte[] BlockMin = { 1 };
      byte[] BlockMax = { 2 };
      boolean result;
  
      byte[][] filterMinMax = { { (byte) 10 }, { (byte) 20 } };
  
      RangeValueFilterExecutorImpl range = new MockUp<RangeValueFilterExecutorImpl>() {
      }.getMockInstance();
      Deencapsulation.setField(range, "isDimensionPresentInCurrentBlock", true);
      Deencapsulation.setField(range, "lessThanExp", true);
      Deencapsulation.setField(range, "greaterThanExp", true);
      Deencapsulation.setField(range, "dimColEvaluatorInfo", dimColumnResolvedFilterInfo);
  
      result = range.isScanRequired(BlockMin, BlockMax, filterMinMax, true);
      Assert.assertFalse(result);
    }
  
    @Test public void checkIsScanRequired1_1() {
      byte[] BlockMin = { 21 };
      byte[] BlockMax = { 22 };
      boolean result;
  
      byte[][] filterMinMax = { { (byte) 10 }, { (byte) 20 } };
  
      RangeValueFilterExecutorImpl range = new MockUp<RangeValueFilterExecutorImpl>() {
      }.getMockInstance();
      Deencapsulation.setField(range, "isDimensionPresentInCurrentBlock", true);
      Deencapsulation.setField(range, "lessThanExp", true);
      Deencapsulation.setField(range, "greaterThanExp", true);
      Deencapsulation.setField(range, "dimColEvaluatorInfo", dimColumnResolvedFilterInfo);
      result = range.isScanRequired(BlockMin, BlockMax, filterMinMax, true);
      Assert.assertFalse(result);
    }
  
    @Test public void checkIsScanRequired2() {
      byte[] BlockMin = { 12 };
      byte[] BlockMax = { 16 };
      boolean result;
  
      byte[][] filterMinMax = { { (byte) 10 }, { (byte) 20 } };
  
      RangeValueFilterExecutorImpl range = new MockUp<RangeValueFilterExecutorImpl>() {
      }.getMockInstance();
      Deencapsulation.setField(range, "isDimensionPresentInCurrentBlock", true);
      Deencapsulation.setField(range, "lessThanExp", true);
      Deencapsulation.setField(range, "greaterThanExp", true);
      Deencapsulation.setField(range, "dimColEvaluatorInfo", dimColumnResolvedFilterInfo);
      result = range.isScanRequired(BlockMin, BlockMax, filterMinMax, true);
      Assert.assertTrue(result);
    }
  
    @Test public void checkIsScanRequired3() {
      byte[] BlockMin = { 12 };
      byte[] BlockMax = { 16 };
      boolean result;
      boolean rangeCovered;
  
      byte[][] filterMinMax = { { (byte) 10 }, { (byte) 20 } };
  
      RangeValueFilterExecutorImpl range = new MockUp<RangeValueFilterExecutorImpl>() {
      }.getMockInstance();
      Deencapsulation.setField(range, "isDimensionPresentInCurrentBlock", true);
      Deencapsulation.setField(range, "lessThanExp", true);
      Deencapsulation.setField(range, "greaterThanExp", true);
      Deencapsulation.setField(range, "dimColEvaluatorInfo", dimColumnResolvedFilterInfo);
  
      result = range.isScanRequired(BlockMin, BlockMax, filterMinMax, true);
      rangeCovered = Deencapsulation.getField(range, "isRangeFullyCoverBlock");
      Assert.assertTrue(result);
      Assert.assertTrue(rangeCovered);
    }
  
    @Test public void checkIsScanRequired4() {
      byte[] BlockMin = { 12 };
      byte[] BlockMax = { 30 };
      boolean result;
      boolean startBlockMinIsDefaultStart;
  
      byte[][] filterMinMax = { { (byte) 10 }, { (byte) 20 } };
  
      RangeValueFilterExecutorImpl range = new MockUp<RangeValueFilterExecutorImpl>() {
      }.getMockInstance();
      Deencapsulation.setField(range, "isDimensionPresentInCurrentBlock", true);
      Deencapsulation.setField(range, "lessThanExp", true);
      Deencapsulation.setField(range, "greaterThanExp", true);
      Deencapsulation.setField(range, "dimColEvaluatorInfo", dimColumnResolvedFilterInfo);
  
      result = range.isScanRequired(BlockMin, BlockMax, filterMinMax, true);
      startBlockMinIsDefaultStart = Deencapsulation.getField(range, "startBlockMinIsDefaultStart");
      Assert.assertTrue(result);
      Assert.assertTrue(startBlockMinIsDefaultStart);
    }
  
    @Test public void checkIsScanRequired5() {
      byte[] BlockMin = { 10 };
      byte[] BlockMax = { 16 };
      boolean result;
      boolean endBlockMaxisDefaultEnd;
  
      byte[][] filterMinMax = { { (byte) 15 }, { (byte) 20 } };
  
      RangeValueFilterExecutorImpl range = new MockUp<RangeValueFilterExecutorImpl>() {
      }.getMockInstance();
      Deencapsulation.setField(range, "isDimensionPresentInCurrentBlock", true);
      Deencapsulation.setField(range, "lessThanExp", true);
      Deencapsulation.setField(range, "greaterThanExp", true);
      Deencapsulation.setField(range, "dimColEvaluatorInfo", dimColumnResolvedFilterInfo);
  
      result = range.isScanRequired(BlockMin, BlockMax, filterMinMax, true);
      endBlockMaxisDefaultEnd = Deencapsulation.getField(range, "endBlockMaxisDefaultEnd");
      Assert.assertTrue(result);
      Assert.assertTrue(endBlockMaxisDefaultEnd);
    }
  
    @AfterClass public static void testcleanUp() {
    }
  
  }
  
  