carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ravipes...@apache.org
Subject [3/5] carbondata git commit: Measure Filter implementation
Date Mon, 03 Jul 2017 09:50:25 GMT
http://git-wip-us.apache.org/repos/asf/carbondata/blob/17db292a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RestructureIncludeFilterExecutorImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RestructureIncludeFilterExecutorImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RestructureIncludeFilterExecutorImpl.java
index 03fe2b4..20889ea 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RestructureIncludeFilterExecutorImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RestructureIncludeFilterExecutorImpl.java
@@ -22,13 +22,16 @@ import java.util.BitSet;
 import org.apache.carbondata.core.datastore.block.SegmentProperties;
 import org.apache.carbondata.core.scan.filter.FilterUtil;
 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.scan.processor.BlocksChunkHolder;
 import org.apache.carbondata.core.util.BitSetGroup;
 
 public class RestructureIncludeFilterExecutorImpl extends RestructureEvaluatorImpl {
 
   protected DimColumnResolvedFilterInfo dimColumnEvaluatorInfo;
+  protected MeasureColumnResolvedFilterInfo measureColumnResolvedFilterInfo;
   protected SegmentProperties segmentProperties;
+  protected boolean isMeasure;
 
   /**
    * flag to check whether filter values contain the default value applied on the dimension column
@@ -37,11 +40,19 @@ public class RestructureIncludeFilterExecutorImpl extends RestructureEvaluatorIm
   protected boolean isDefaultValuePresentInFilterValues;
 
   public RestructureIncludeFilterExecutorImpl(DimColumnResolvedFilterInfo dimColumnEvaluatorInfo,
-      SegmentProperties segmentProperties) {
+      MeasureColumnResolvedFilterInfo measureColumnResolvedFilterInfo,
+      SegmentProperties segmentProperties, boolean isMeasure) {
     this.dimColumnEvaluatorInfo = dimColumnEvaluatorInfo;
+    this.measureColumnResolvedFilterInfo = measureColumnResolvedFilterInfo;
+    this.isMeasure = isMeasure;
     this.segmentProperties = segmentProperties;
-    isDefaultValuePresentInFilterValues =
-        isDimensionDefaultValuePresentInFilterValues(dimColumnEvaluatorInfo);
+    if (isMeasure) {
+      isDefaultValuePresentInFilterValues =
+          isMeasureDefaultValuePresentInFilterValues(measureColumnResolvedFilterInfo);
+    } else {
+      isDefaultValuePresentInFilterValues =
+          isDimensionDefaultValuePresentInFilterValues(dimColumnEvaluatorInfo);
+    }
   }
 
   @Override public BitSetGroup applyFilter(BlocksChunkHolder blockChunkHolder) throws IOException {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/17db292a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelFilterExecuterImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelFilterExecuterImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelFilterExecuterImpl.java
index a72d526..0cdb3c7 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelFilterExecuterImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelFilterExecuterImpl.java
@@ -104,17 +104,31 @@ public class RowLevelFilterExecuterImpl implements FilterExecuter {
       List<MeasureColumnResolvedFilterInfo> msrColEvalutorInfoList, Expression exp,
       AbsoluteTableIdentifier tableIdentifier, SegmentProperties segmentProperties,
       Map<Integer, GenericQueryType> complexDimensionInfoMap) {
-    this.dimColEvaluatorInfoList = dimColEvaluatorInfoList;
     this.segmentProperties = segmentProperties;
-    this.dimensionBlocksIndex = new int[dimColEvaluatorInfoList.size()];
-    this.isDimensionPresentInCurrentBlock = new boolean[dimColEvaluatorInfoList.size()];
+    if (null == dimColEvaluatorInfoList) {
+      this.dimColEvaluatorInfoList = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+    } else {
+      this.dimColEvaluatorInfoList = dimColEvaluatorInfoList;
+    }
+    if (dimColEvaluatorInfoList.size() > 0) {
+      this.isDimensionPresentInCurrentBlock = new boolean[dimColEvaluatorInfoList.size()];
+      this.dimensionBlocksIndex = new int[dimColEvaluatorInfoList.size()];
+    } else {
+      this.isDimensionPresentInCurrentBlock = new boolean[]{false};
+      this.dimensionBlocksIndex = new int[]{0};
+    }
     if (null == msrColEvalutorInfoList) {
       this.msrColEvalutorInfoList = new ArrayList<MeasureColumnResolvedFilterInfo>(20);
     } else {
       this.msrColEvalutorInfoList = msrColEvalutorInfoList;
     }
-    this.measureBlocksIndex = new int[msrColEvalutorInfoList.size()];
-    this.isMeasurePresentInCurrentBlock = new boolean[msrColEvalutorInfoList.size()];
+    if (msrColEvalutorInfoList.size() > 0) {
+      this.isMeasurePresentInCurrentBlock = new boolean[msrColEvalutorInfoList.size()];
+      this.measureBlocksIndex = new int[msrColEvalutorInfoList.size()];
+    } else {
+      this.isMeasurePresentInCurrentBlock = new boolean[]{false};
+      this.measureBlocksIndex = new int[] {0};
+    }
     this.exp = exp;
     this.tableIdentifier = tableIdentifier;
     this.complexDimensionInfoMap = complexDimensionInfoMap;

http://git-wip-us.apache.org/repos/asf/carbondata/blob/17db292a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeGrtThanFiterExecuterImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeGrtThanFiterExecuterImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeGrtThanFiterExecuterImpl.java
index be82be7..cbe59a1 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeGrtThanFiterExecuterImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeGrtThanFiterExecuterImpl.java
@@ -17,16 +17,23 @@
 package org.apache.carbondata.core.scan.filter.executer;
 
 import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
 import java.util.BitSet;
+import java.util.Comparator;
 import java.util.List;
 
 import org.apache.carbondata.core.constants.CarbonCommonConstants;
 import org.apache.carbondata.core.datastore.block.SegmentProperties;
 import org.apache.carbondata.core.datastore.chunk.DimensionColumnDataChunk;
+import org.apache.carbondata.core.datastore.chunk.MeasureColumnDataChunk;
 import org.apache.carbondata.core.datastore.chunk.impl.DimensionRawColumnChunk;
+import org.apache.carbondata.core.datastore.chunk.impl.MeasureRawColumnChunk;
 import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
+import org.apache.carbondata.core.metadata.datatype.DataType;
 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.CarbonMeasure;
 import org.apache.carbondata.core.scan.expression.Expression;
 import org.apache.carbondata.core.scan.expression.exception.FilterUnsupportedException;
 import org.apache.carbondata.core.scan.filter.FilterUtil;
@@ -36,6 +43,7 @@ import org.apache.carbondata.core.scan.processor.BlocksChunkHolder;
 import org.apache.carbondata.core.util.BitSetGroup;
 import org.apache.carbondata.core.util.ByteUtil;
 import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.DataTypeUtil;
 
 public class RowLevelRangeGrtThanFiterExecuterImpl extends RowLevelFilterExecuterImpl {
   private byte[][] filterRangeValues;
@@ -44,6 +52,7 @@ public class RowLevelRangeGrtThanFiterExecuterImpl extends RowLevelFilterExecute
    * flag to check whether default values is present in the filter value list
    */
   private boolean isDefaultValuePresentInFilter;
+  private int lastDimensionColOrdinal = 0;
   public RowLevelRangeGrtThanFiterExecuterImpl(
       List<DimColumnResolvedFilterInfo> dimColEvaluatorInfoList,
       List<MeasureColumnResolvedFilterInfo> msrColEvalutorInfoList, Expression exp,
@@ -52,8 +61,11 @@ public class RowLevelRangeGrtThanFiterExecuterImpl extends RowLevelFilterExecute
     super(dimColEvaluatorInfoList, msrColEvalutorInfoList, exp, tableIdentifier, segmentProperties,
         null);
     this.filterRangeValues = filterRangeValues;
-    isNaturalSorted = dimColEvaluatorInfoList.get(0).getDimension().isUseInvertedIndex()
-        && dimColEvaluatorInfoList.get(0).getDimension().isSortColumn();
+    lastDimensionColOrdinal = segmentProperties.getLastDimensionColOrdinal();
+    if (isDimensionPresentInCurrentBlock[0] == true) {
+      isNaturalSorted = dimColEvaluatorInfoList.get(0).getDimension().isUseInvertedIndex()
+          && dimColEvaluatorInfoList.get(0).getDimension().isSortColumn();
+    }
     ifDefaultValueMatchesFilter();
   }
 
@@ -61,7 +73,7 @@ public class RowLevelRangeGrtThanFiterExecuterImpl extends RowLevelFilterExecute
    * This method will check whether default value is present in the given filter values
    */
   private void ifDefaultValueMatchesFilter() {
-    if (!this.isDimensionPresentInCurrentBlock[0]) {
+    if (!dimColEvaluatorInfoList.isEmpty() && !isDimensionPresentInCurrentBlock[0]) {
       CarbonDimension dimension = this.dimColEvaluatorInfoList.get(0).getDimension();
       byte[] defaultValue = dimension.getDefaultValue();
       if (null != defaultValue) {
@@ -74,80 +86,205 @@ public class RowLevelRangeGrtThanFiterExecuterImpl extends RowLevelFilterExecute
           }
         }
       }
+    } else if (!msrColEvalutorInfoList.isEmpty() && !isMeasurePresentInCurrentBlock[0]) {
+      CarbonMeasure measure = this.msrColEvalutorInfoList.get(0).getMeasure();
+      byte[] defaultValue = measure.getDefaultValue();
+      if (null != defaultValue) {
+        for (int k = 0; k < filterRangeValues.length; k++) {
+          int maxCompare =
+              ByteUtil.UnsafeComparer.INSTANCE.compareTo(filterRangeValues[k], defaultValue);
+          if (maxCompare < 0) {
+            isDefaultValuePresentInFilter = true;
+            break;
+          }
+        }
+      }
     }
   }
 
   @Override public BitSet isScanRequired(byte[][] blockMaxValue, byte[][] blockMinValue) {
     BitSet bitSet = new BitSet(1);
-    boolean isScanRequired =
-        isScanRequired(blockMaxValue[dimensionBlocksIndex[0]], filterRangeValues);
+    boolean isScanRequired = false;
+    byte[] maxValue = null;
+    if (isMeasurePresentInCurrentBlock[0] || isDimensionPresentInCurrentBlock[0]) {
+      if (isMeasurePresentInCurrentBlock[0]) {
+        maxValue = blockMaxValue[measureBlocksIndex[0] + lastDimensionColOrdinal];
+        isScanRequired =
+            isScanRequired(maxValue, filterRangeValues, msrColEvalutorInfoList.get(0).getType());
+      } else {
+        maxValue = blockMaxValue[dimensionBlocksIndex[0]];
+        isScanRequired = isScanRequired(maxValue, filterRangeValues);
+      }
+    } else {
+      isScanRequired = isDefaultValuePresentInFilter;
+    }
+
     if (isScanRequired) {
       bitSet.set(0);
     }
     return bitSet;
   }
 
+
   private boolean isScanRequired(byte[] blockMaxValue, byte[][] filterValues) {
     boolean isScanRequired = false;
-    if (isDimensionPresentInCurrentBlock[0]) {
-      for (int k = 0; k < filterValues.length; k++) {
-        // filter value should be in range of max and min value i.e
-        // max>filtervalue>min
-        // so filter-max should be negative
-        int maxCompare = ByteUtil.UnsafeComparer.INSTANCE.compareTo(filterValues[k], blockMaxValue);
-        // if any filter value is in range than this block needs to be
-        // scanned means always less than block max range.
-        if (maxCompare < 0) {
-          isScanRequired = true;
-          break;
-        }
+    for (int k = 0; k < filterValues.length; k++) {
+      // filter value should be in range of max and min value i.e
+      // max>filtervalue>min
+      // so filter-max should be negative
+      int maxCompare = ByteUtil.UnsafeComparer.INSTANCE.compareTo(filterValues[k], blockMaxValue);
+      // if any filter value is in range than this block needs to be
+      // scanned less than equal to max range.
+      if (maxCompare < 0) {
+        isScanRequired = true;
+        break;
       }
-    } else {
-      isScanRequired = isDefaultValuePresentInFilter;
     }
     return isScanRequired;
   }
 
+  private boolean isScanRequired(byte[] maxValue, byte[][] filterValue,
+      DataType dataType) {
+    for (int i = 0; i < filterValue.length; i++) {
+      if (filterValue[i].length == 0 || maxValue.length == 0) {
+        return isScanRequired(maxValue, filterValue);
+      }
+      switch (dataType) {
+        case DOUBLE:
+          double maxValueDouble = ByteBuffer.wrap(maxValue).getDouble();
+          double filterValueDouble = ByteBuffer.wrap(filterValue[i]).getDouble();
+          if (filterValueDouble < maxValueDouble) {
+            return true;
+          }
+          break;
+        case INT:
+        case SHORT:
+        case LONG:
+          long maxValueLong = ByteBuffer.wrap(maxValue).getLong();
+          long filterValueLong = ByteBuffer.wrap(filterValue[i]).getLong();
+          if (filterValueLong < maxValueLong) {
+            return true;
+          }
+          break;
+        case DECIMAL:
+          BigDecimal maxDecimal = DataTypeUtil.byteToBigDecimal(maxValue);
+          BigDecimal filterDecimal = DataTypeUtil.byteToBigDecimal(filterValue[i]);
+          if (filterDecimal.compareTo(maxDecimal) < 0) {
+            return true;
+          }
+      }
+    }
+    return false;
+  }
+
   @Override public BitSetGroup applyFilter(BlocksChunkHolder blockChunkHolder)
       throws FilterUnsupportedException, IOException {
     // select all rows if dimension does not exists in the current block
-    if (!isDimensionPresentInCurrentBlock[0]) {
+    if (!isDimensionPresentInCurrentBlock[0] && !isMeasurePresentInCurrentBlock[0]) {
       int numberOfRows = blockChunkHolder.getDataBlock().nodeSize();
       return FilterUtil
           .createBitSetGroupWithDefaultValue(blockChunkHolder.getDataBlock().numberOfPages(),
               numberOfRows, true);
     }
-    int blockIndex =
-        segmentProperties.getDimensionOrdinalToBlockMapping().get(dimensionBlocksIndex[0]);
-    if (null == blockChunkHolder.getDimensionRawDataChunk()[blockIndex]) {
-      blockChunkHolder.getDimensionRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
-          .getDimensionChunk(blockChunkHolder.getFileReader(), blockIndex);
-    }
-    DimensionRawColumnChunk rawColumnChunk =
-        blockChunkHolder.getDimensionRawDataChunk()[blockIndex];
-    BitSetGroup bitSetGroup = new BitSetGroup(rawColumnChunk.getPagesCount());
-    for (int i = 0; i < rawColumnChunk.getPagesCount(); i++) {
-      if (rawColumnChunk.getMaxValues() != null) {
-        if (isScanRequired(rawColumnChunk.getMaxValues()[i], this.filterRangeValues)) {
-          int compare = ByteUtil.UnsafeComparer.INSTANCE
-              .compareTo(filterRangeValues[0], rawColumnChunk.getMinValues()[i]);
-          if (compare < 0) {
-            BitSet bitSet = new BitSet(rawColumnChunk.getRowCount()[i]);
-            bitSet.flip(0, rawColumnChunk.getRowCount()[i]);
-            bitSetGroup.setBitSet(bitSet, i);
-          } else {
-            BitSet bitSet = getFilteredIndexes(rawColumnChunk.convertToDimColDataChunk(i),
-                rawColumnChunk.getRowCount()[i]);
-            bitSetGroup.setBitSet(bitSet, i);
+    if (isDimensionPresentInCurrentBlock[0]) {
+      int blockIndex =
+          segmentProperties.getDimensionOrdinalToBlockMapping().get(dimensionBlocksIndex[0]);
+      if (null == blockChunkHolder.getDimensionRawDataChunk()[blockIndex]) {
+        blockChunkHolder.getDimensionRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
+            .getDimensionChunk(blockChunkHolder.getFileReader(), blockIndex);
+      }
+      DimensionRawColumnChunk rawColumnChunk =
+          blockChunkHolder.getDimensionRawDataChunk()[blockIndex];
+      BitSetGroup bitSetGroup = new BitSetGroup(rawColumnChunk.getPagesCount());
+      for (int i = 0; i < rawColumnChunk.getPagesCount(); i++) {
+        if (rawColumnChunk.getMaxValues() != null) {
+          if (isScanRequired(rawColumnChunk.getMaxValues()[i], this.filterRangeValues)) {
+            int compare = ByteUtil.UnsafeComparer.INSTANCE
+                .compareTo(filterRangeValues[0], rawColumnChunk.getMinValues()[i]);
+            if (compare < 0) {
+              BitSet bitSet = new BitSet(rawColumnChunk.getRowCount()[i]);
+              bitSet.flip(0, rawColumnChunk.getRowCount()[i]);
+              bitSetGroup.setBitSet(bitSet, i);
+            } else {
+              BitSet bitSet = getFilteredIndexes(rawColumnChunk.convertToDimColDataChunk(i),
+                  rawColumnChunk.getRowCount()[i]);
+              bitSetGroup.setBitSet(bitSet, i);
+            }
           }
+        } else {
+          BitSet bitSet = getFilteredIndexes(rawColumnChunk.convertToDimColDataChunk(i),
+              rawColumnChunk.getRowCount()[i]);
+          bitSetGroup.setBitSet(bitSet, i);
+        }
+      }
+      return bitSetGroup;
+    } else if (isMeasurePresentInCurrentBlock[0]) {
+      int blockIndex =
+          segmentProperties.getMeasuresOrdinalToBlockMapping().get(measureBlocksIndex[0]);
+      if (null == blockChunkHolder.getMeasureRawDataChunk()[blockIndex]) {
+        blockChunkHolder.getMeasureRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
+            .getMeasureChunk(blockChunkHolder.getFileReader(), blockIndex);
+      }
+      MeasureRawColumnChunk rawColumnChunk = blockChunkHolder.getMeasureRawDataChunk()[blockIndex];
+      BitSetGroup bitSetGroup = new BitSetGroup(rawColumnChunk.getPagesCount());
+      for (int i = 0; i < rawColumnChunk.getPagesCount(); i++) {
+        if (rawColumnChunk.getMaxValues() != null) {
+          if (isScanRequired(rawColumnChunk.getMaxValues()[i], this.filterRangeValues,
+              msrColEvalutorInfoList.get(0).getType())) {
+            int compare = ByteUtil.UnsafeComparer.INSTANCE
+                .compareTo(filterRangeValues[0], rawColumnChunk.getMinValues()[i]);
+            if (compare < 0) {
+              BitSet bitSet = new BitSet(rawColumnChunk.getRowCount()[i]);
+              bitSet.flip(0, rawColumnChunk.getRowCount()[i]);
+              bitSetGroup.setBitSet(bitSet, i);
+            } else {
+              BitSet bitSet =
+                  getFilteredIndexesForMeasures(rawColumnChunk.convertToMeasureColDataChunk(i),
+                      rawColumnChunk.getRowCount()[i]);
+              bitSetGroup.setBitSet(bitSet, i);
+            }
+          }
+        } else {
+          BitSet bitSet =
+              getFilteredIndexesForMeasures(rawColumnChunk.convertToMeasureColDataChunk(i),
+                  rawColumnChunk.getRowCount()[i]);
+          bitSetGroup.setBitSet(bitSet, i);
         }
-      } else {
-        BitSet bitSet = getFilteredIndexes(rawColumnChunk.convertToDimColDataChunk(i),
-            rawColumnChunk.getRowCount()[i]);
-        bitSetGroup.setBitSet(bitSet, i);
       }
+      return bitSetGroup;
     }
-    return bitSetGroup;
+    return null;
+  }
+
+  private BitSet getFilteredIndexesForMeasures(MeasureColumnDataChunk measureColumnDataChunk,
+      int numerOfRows) {
+    BitSet bitSet = new BitSet(numerOfRows);
+    byte[][] filterValues = this.filterRangeValues;
+    DataType msrType = msrColEvalutorInfoList.get(0).getType();
+    Comparator comparator = FilterUtil.getComparatorByDataTypeForMeasure(msrType);
+    for (int i = 0; i < filterValues.length; i++) {
+      if (filterValues[i].length == 0) {
+        BitSet nullBitSet = measureColumnDataChunk.getNullValueIndexHolder().getBitSet();
+        for (int j = nullBitSet.nextSetBit(0); j >= 0; j = nullBitSet.nextSetBit(j + 1)) {
+          bitSet.set(j);
+        }
+        continue;
+      }
+      Object filter = DataTypeUtil.getMeasureObjectFromDataType(filterValues[i], msrType);
+      for (int startIndex = 0; startIndex < numerOfRows; startIndex++) {
+        Object msrValue = DataTypeUtil
+            .getMeasureObjectBasedOnDataType(measureColumnDataChunk, startIndex,
+                 msrColEvalutorInfoList.get(0).getMeasure());
+
+        if (comparator.compare(msrValue, filter) > 0) {
+          // This is a match.
+          bitSet.set(startIndex);
+        }
+      }
+    }
+    CarbonUtil
+        .updateBitSetForNull(measureColumnDataChunk.getNullValueIndexHolder().getBitSet(), bitSet);
+    return bitSet;
   }
 
   private BitSet getFilteredIndexes(DimensionColumnDataChunk dimensionColumnDataChunk,
@@ -296,6 +433,12 @@ public class RowLevelRangeGrtThanFiterExecuterImpl extends RowLevelFilterExecute
         blockChunkHolder.getDimensionRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
             .getDimensionChunk(blockChunkHolder.getFileReader(), blockIndex);
       }
+    } else if (isMeasurePresentInCurrentBlock[0]) {
+      int blockIndex = measureBlocksIndex[0];
+      if (null == blockChunkHolder.getMeasureRawDataChunk()[blockIndex]) {
+        blockChunkHolder.getMeasureRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
+            .getMeasureChunk(blockChunkHolder.getFileReader(), blockIndex);
+      }
     }
   }
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/17db292a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeGrtrThanEquaToFilterExecuterImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeGrtrThanEquaToFilterExecuterImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeGrtrThanEquaToFilterExecuterImpl.java
index 53da6c5..5c7878d 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeGrtrThanEquaToFilterExecuterImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeGrtrThanEquaToFilterExecuterImpl.java
@@ -17,16 +17,23 @@
 package org.apache.carbondata.core.scan.filter.executer;
 
 import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
 import java.util.BitSet;
+import java.util.Comparator;
 import java.util.List;
 
 import org.apache.carbondata.core.constants.CarbonCommonConstants;
 import org.apache.carbondata.core.datastore.block.SegmentProperties;
 import org.apache.carbondata.core.datastore.chunk.DimensionColumnDataChunk;
+import org.apache.carbondata.core.datastore.chunk.MeasureColumnDataChunk;
 import org.apache.carbondata.core.datastore.chunk.impl.DimensionRawColumnChunk;
+import org.apache.carbondata.core.datastore.chunk.impl.MeasureRawColumnChunk;
 import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
+import org.apache.carbondata.core.metadata.datatype.DataType;
 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.CarbonMeasure;
 import org.apache.carbondata.core.scan.expression.Expression;
 import org.apache.carbondata.core.scan.expression.exception.FilterUnsupportedException;
 import org.apache.carbondata.core.scan.filter.FilterUtil;
@@ -36,6 +43,7 @@ import org.apache.carbondata.core.scan.processor.BlocksChunkHolder;
 import org.apache.carbondata.core.util.BitSetGroup;
 import org.apache.carbondata.core.util.ByteUtil;
 import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.DataTypeUtil;
 
 public class RowLevelRangeGrtrThanEquaToFilterExecuterImpl extends RowLevelFilterExecuterImpl {
 
@@ -45,6 +53,7 @@ public class RowLevelRangeGrtrThanEquaToFilterExecuterImpl extends RowLevelFilte
    * flag to check whether default values is present in the filter value list
    */
   private boolean isDefaultValuePresentInFilter;
+  private int lastDimensionColOrdinal = 0;
 
   public RowLevelRangeGrtrThanEquaToFilterExecuterImpl(
       List<DimColumnResolvedFilterInfo> dimColEvaluatorInfoList,
@@ -54,8 +63,12 @@ public class RowLevelRangeGrtrThanEquaToFilterExecuterImpl extends RowLevelFilte
     super(dimColEvaluatorInfoList, msrColEvalutorInfoList, exp, tableIdentifier, segmentProperties,
         null);
     this.filterRangeValues = filterRangeValues;
-    isNaturalSorted = dimColEvaluatorInfoList.get(0).getDimension().isUseInvertedIndex()
-        && dimColEvaluatorInfoList.get(0).getDimension().isSortColumn();
+    lastDimensionColOrdinal = segmentProperties.getLastDimensionColOrdinal();
+
+    if (isDimensionPresentInCurrentBlock[0] == true) {
+      isNaturalSorted = dimColEvaluatorInfoList.get(0).getDimension().isUseInvertedIndex()
+          && dimColEvaluatorInfoList.get(0).getDimension().isSortColumn();
+    }
     ifDefaultValueMatchesFilter();
   }
 
@@ -63,7 +76,7 @@ public class RowLevelRangeGrtrThanEquaToFilterExecuterImpl extends RowLevelFilte
    * This method will check whether default value is present in the given filter values
    */
   private void ifDefaultValueMatchesFilter() {
-    if (!this.isDimensionPresentInCurrentBlock[0]) {
+    if (!dimColEvaluatorInfoList.isEmpty() && !isDimensionPresentInCurrentBlock[0]) {
       CarbonDimension dimension = this.dimColEvaluatorInfoList.get(0).getDimension();
       byte[] defaultValue = dimension.getDefaultValue();
       if (null != defaultValue) {
@@ -76,13 +89,39 @@ public class RowLevelRangeGrtrThanEquaToFilterExecuterImpl extends RowLevelFilte
           }
         }
       }
+    } else if (!msrColEvalutorInfoList.isEmpty() && !isMeasurePresentInCurrentBlock[0]) {
+      CarbonMeasure measure = this.msrColEvalutorInfoList.get(0).getMeasure();
+      byte[] defaultValue = measure.getDefaultValue();
+      if (null != defaultValue) {
+        for (int k = 0; k < filterRangeValues.length; k++) {
+          int maxCompare =
+              ByteUtil.UnsafeComparer.INSTANCE.compareTo(filterRangeValues[k], defaultValue);
+          if (maxCompare <= 0) {
+            isDefaultValuePresentInFilter = true;
+            break;
+          }
+        }
+      }
     }
   }
 
   @Override public BitSet isScanRequired(byte[][] blockMaxValue, byte[][] blockMinValue) {
     BitSet bitSet = new BitSet(1);
-    boolean isScanRequired =
-        isScanRequired(blockMaxValue[dimensionBlocksIndex[0]], filterRangeValues);
+    boolean isScanRequired = false;
+    byte[] maxValue = null;
+    if (isMeasurePresentInCurrentBlock[0] || isDimensionPresentInCurrentBlock[0]) {
+      if (isMeasurePresentInCurrentBlock[0]) {
+        maxValue = blockMaxValue[measureBlocksIndex[0] + lastDimensionColOrdinal];
+        isScanRequired =
+            isScanRequired(maxValue, filterRangeValues, msrColEvalutorInfoList.get(0).getType());
+      } else {
+        maxValue = blockMaxValue[dimensionBlocksIndex[0]];
+        isScanRequired = isScanRequired(maxValue, filterRangeValues);
+      }
+    } else {
+      isScanRequired = isDefaultValuePresentInFilter;
+    }
+
     if (isScanRequired) {
       bitSet.set(0);
     }
@@ -91,67 +130,167 @@ public class RowLevelRangeGrtrThanEquaToFilterExecuterImpl extends RowLevelFilte
 
   private boolean isScanRequired(byte[] blockMaxValue, byte[][] filterValues) {
     boolean isScanRequired = false;
-    if (isDimensionPresentInCurrentBlock[0]) {
-      for (int k = 0; k < filterValues.length; k++) {
-        // filter value should be in range of max and min value i.e
-        // max>filtervalue>min
-        // so filter-max should be negative
-        int maxCompare = ByteUtil.UnsafeComparer.INSTANCE.compareTo(filterValues[k], blockMaxValue);
-        // if any filter value is in range than this block needs to be
-        // scanned less than equal to max range.
-        if (maxCompare <= 0) {
-          isScanRequired = true;
-          break;
-        }
+    for (int k = 0; k < filterValues.length; k++) {
+      // filter value should be in range of max and min value i.e
+      // max>filtervalue>min
+      // so filter-max should be negative
+      int maxCompare = ByteUtil.UnsafeComparer.INSTANCE.compareTo(filterValues[k], blockMaxValue);
+      // if any filter value is in range than this block needs to be
+      // scanned less than equal to max range.
+      if (maxCompare <= 0) {
+        isScanRequired = true;
+        break;
       }
-    } else {
-      isScanRequired = isDefaultValuePresentInFilter;
     }
     return isScanRequired;
   }
 
+  private boolean isScanRequired(byte[] maxValue, byte[][] filterValue,
+      DataType dataType) {
+    for (int i = 0; i < filterValue.length; i++) {
+      if (filterValue[i].length == 0 || maxValue.length == 0) {
+        return isScanRequired(maxValue, filterValue);
+      }
+      switch (dataType) {
+        case DOUBLE:
+          double maxValueDouble = ByteBuffer.wrap(maxValue).getDouble();
+          double filterValueDouble = ByteBuffer.wrap(filterValue[i]).getDouble();
+          if (filterValueDouble <= maxValueDouble) {
+            return true;
+          }
+          break;
+        case INT:
+        case SHORT:
+        case LONG:
+          long maxValueLong = ByteBuffer.wrap(maxValue).getLong();
+          long filterValueLong = ByteBuffer.wrap(filterValue[i]).getLong();
+          if (filterValueLong <= maxValueLong) {
+            return true;
+          }
+          break;
+        case DECIMAL:
+          BigDecimal maxDecimal = DataTypeUtil.byteToBigDecimal(maxValue);
+          BigDecimal filterDecimal = DataTypeUtil.byteToBigDecimal(filterValue[i]);
+          if (filterDecimal.compareTo(maxDecimal) <= 0) {
+            return true;
+          }
+      }
+    }
+    return false;
+  }
+
   @Override public BitSetGroup applyFilter(BlocksChunkHolder blockChunkHolder)
       throws FilterUnsupportedException, IOException {
     // select all rows if dimension does not exists in the current block
-    if (!isDimensionPresentInCurrentBlock[0]) {
+    if (!isDimensionPresentInCurrentBlock[0] && !isMeasurePresentInCurrentBlock[0]) {
       int numberOfRows = blockChunkHolder.getDataBlock().nodeSize();
       return FilterUtil
           .createBitSetGroupWithDefaultValue(blockChunkHolder.getDataBlock().numberOfPages(),
               numberOfRows, true);
     }
-    int blockIndex =
-        segmentProperties.getDimensionOrdinalToBlockMapping().get(dimensionBlocksIndex[0]);
-    if (null == blockChunkHolder.getDimensionRawDataChunk()[blockIndex]) {
-      blockChunkHolder.getDimensionRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
-          .getDimensionChunk(blockChunkHolder.getFileReader(), blockIndex);
-    }
-    DimensionRawColumnChunk rawColumnChunk =
-        blockChunkHolder.getDimensionRawDataChunk()[blockIndex];
-    BitSetGroup bitSetGroup = new BitSetGroup(rawColumnChunk.getPagesCount());
-    for (int i = 0; i < rawColumnChunk.getPagesCount(); i++) {
-      if (rawColumnChunk.getMaxValues() != null) {
-        if (isScanRequired(rawColumnChunk.getMaxValues()[i], this.filterRangeValues)) {
-          int compare = ByteUtil.UnsafeComparer.INSTANCE
-              .compareTo(filterRangeValues[0], rawColumnChunk.getMinValues()[i]);
-          if (compare <= 0) {
-            BitSet bitSet = new BitSet(rawColumnChunk.getRowCount()[i]);
-            bitSet.flip(0, rawColumnChunk.getRowCount()[i]);
-            bitSetGroup.setBitSet(bitSet, i);
-          } else {
-            BitSet bitSet = getFilteredIndexes(rawColumnChunk.convertToDimColDataChunk(i),
-                rawColumnChunk.getRowCount()[i]);
-            bitSetGroup.setBitSet(bitSet, i);
+
+    if (isDimensionPresentInCurrentBlock[0]) {
+      int blockIndex =
+          segmentProperties.getDimensionOrdinalToBlockMapping().get(dimensionBlocksIndex[0]);
+      if (null == blockChunkHolder.getDimensionRawDataChunk()[blockIndex]) {
+        blockChunkHolder.getDimensionRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
+            .getDimensionChunk(blockChunkHolder.getFileReader(), blockIndex);
+      }
+      DimensionRawColumnChunk rawColumnChunk =
+          blockChunkHolder.getDimensionRawDataChunk()[blockIndex];
+      BitSetGroup bitSetGroup = new BitSetGroup(rawColumnChunk.getPagesCount());
+      for (int i = 0; i < rawColumnChunk.getPagesCount(); i++) {
+        if (rawColumnChunk.getMaxValues() != null) {
+          if (isScanRequired(rawColumnChunk.getMaxValues()[i], this.filterRangeValues)) {
+            int compare = ByteUtil.UnsafeComparer.INSTANCE
+                .compareTo(filterRangeValues[0], rawColumnChunk.getMinValues()[i]);
+            if (compare <= 0) {
+              BitSet bitSet = new BitSet(rawColumnChunk.getRowCount()[i]);
+              bitSet.flip(0, rawColumnChunk.getRowCount()[i]);
+              bitSetGroup.setBitSet(bitSet, i);
+            } else {
+              BitSet bitSet = getFilteredIndexes(rawColumnChunk.convertToDimColDataChunk(i),
+                  rawColumnChunk.getRowCount()[i]);
+              bitSetGroup.setBitSet(bitSet, i);
+            }
           }
+        } else {
+          BitSet bitSet = getFilteredIndexes(rawColumnChunk.convertToDimColDataChunk(i),
+              rawColumnChunk.getRowCount()[i]);
+          bitSetGroup.setBitSet(bitSet, i);
+        }
+      }
+      return bitSetGroup;
+    } else if (isMeasurePresentInCurrentBlock[0]) {
+      int blockIndex =
+          segmentProperties.getMeasuresOrdinalToBlockMapping().get(measureBlocksIndex[0]);
+      if (null == blockChunkHolder.getMeasureRawDataChunk()[blockIndex]) {
+        blockChunkHolder.getMeasureRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
+            .getMeasureChunk(blockChunkHolder.getFileReader(), blockIndex);
+      }
+      MeasureRawColumnChunk rawColumnChunk = blockChunkHolder.getMeasureRawDataChunk()[blockIndex];
+      BitSetGroup bitSetGroup = new BitSetGroup(rawColumnChunk.getPagesCount());
+      for (int i = 0; i < rawColumnChunk.getPagesCount(); i++) {
+        if (rawColumnChunk.getMaxValues() != null) {
+          if (isScanRequired(rawColumnChunk.getMaxValues()[i], this.filterRangeValues,
+              msrColEvalutorInfoList.get(0).getType())) {
+            int compare = ByteUtil.UnsafeComparer.INSTANCE
+                .compareTo(filterRangeValues[0], rawColumnChunk.getMinValues()[i]);
+            if (compare <= 0) {
+              BitSet bitSet = new BitSet(rawColumnChunk.getRowCount()[i]);
+              bitSet.flip(0, rawColumnChunk.getRowCount()[i]);
+              bitSetGroup.setBitSet(bitSet, i);
+            } else {
+              BitSet bitSet =
+                  getFilteredIndexesForMeasures(rawColumnChunk.convertToMeasureColDataChunk(i),
+                      rawColumnChunk.getRowCount()[i]);
+              bitSetGroup.setBitSet(bitSet, i);
+            }
+          }
+        } else {
+          BitSet bitSet =
+              getFilteredIndexesForMeasures(rawColumnChunk.convertToMeasureColDataChunk(i),
+                  rawColumnChunk.getRowCount()[i]);
+          bitSetGroup.setBitSet(bitSet, i);
         }
-      } else {
-        BitSet bitSet = getFilteredIndexes(rawColumnChunk.convertToDimColDataChunk(i),
-            rawColumnChunk.getRowCount()[i]);
-        bitSetGroup.setBitSet(bitSet, i);
       }
+      return bitSetGroup;
     }
-    return bitSetGroup;
+    return null;
   }
 
+  private BitSet getFilteredIndexesForMeasures(MeasureColumnDataChunk measureColumnDataChunk,
+      int numerOfRows) {
+    BitSet bitSet = new BitSet(numerOfRows);
+    byte[][] filterValues = this.filterRangeValues;
+    DataType msrType = msrColEvalutorInfoList.get(0).getType();
+    Comparator comparator = FilterUtil.getComparatorByDataTypeForMeasure(msrType);
+    for (int i = 0; i < filterValues.length; i++) {
+      if (filterValues[i].length == 0) {
+        BitSet nullBitSet = measureColumnDataChunk.getNullValueIndexHolder().getBitSet();
+        for (int j = nullBitSet.nextSetBit(0); j >= 0; j = nullBitSet.nextSetBit(j + 1)) {
+          bitSet.set(j);
+        }
+        continue;
+      }
+      Object filter = DataTypeUtil.getMeasureObjectFromDataType(filterValues[i], msrType);
+      for (int startIndex = 0; startIndex < numerOfRows; startIndex++) {
+        Object msrValue = DataTypeUtil
+            .getMeasureObjectBasedOnDataType(measureColumnDataChunk, startIndex,
+                 msrColEvalutorInfoList.get(0).getMeasure());
+
+        if (comparator.compare(msrValue, filter) >= 0) {
+          // This is a match.
+          bitSet.set(startIndex);
+        }
+      }
+    }
+    CarbonUtil
+        .updateBitSetForNull(measureColumnDataChunk.getNullValueIndexHolder().getBitSet(), bitSet);
+    return bitSet;
+  }
+
+
   private BitSet getFilteredIndexes(DimensionColumnDataChunk dimensionColumnDataChunk,
       int numerOfRows) {
     BitSet bitSet = null;
@@ -283,6 +422,12 @@ public class RowLevelRangeGrtrThanEquaToFilterExecuterImpl extends RowLevelFilte
         blockChunkHolder.getDimensionRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
             .getDimensionChunk(blockChunkHolder.getFileReader(), blockIndex);
       }
+    } else if (isMeasurePresentInCurrentBlock[0]) {
+      int blockIndex = measureBlocksIndex[0];
+      if (null == blockChunkHolder.getMeasureRawDataChunk()[blockIndex]) {
+        blockChunkHolder.getMeasureRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
+            .getMeasureChunk(blockChunkHolder.getFileReader(), blockIndex);
+      }
     }
   }
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/17db292a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeLessThanEqualFilterExecuterImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeLessThanEqualFilterExecuterImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeLessThanEqualFilterExecuterImpl.java
index eaf58a4..ed14aa1 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeLessThanEqualFilterExecuterImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeLessThanEqualFilterExecuterImpl.java
@@ -17,18 +17,25 @@
 package org.apache.carbondata.core.scan.filter.executer;
 
 import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
 import java.util.BitSet;
+import java.util.Comparator;
 import java.util.List;
 
 import org.apache.carbondata.core.constants.CarbonCommonConstants;
 import org.apache.carbondata.core.datastore.block.SegmentProperties;
 import org.apache.carbondata.core.datastore.chunk.DimensionColumnDataChunk;
+import org.apache.carbondata.core.datastore.chunk.MeasureColumnDataChunk;
 import org.apache.carbondata.core.datastore.chunk.impl.DimensionRawColumnChunk;
+import org.apache.carbondata.core.datastore.chunk.impl.MeasureRawColumnChunk;
 import org.apache.carbondata.core.keygenerator.directdictionary.DirectDictionaryGenerator;
 import org.apache.carbondata.core.keygenerator.directdictionary.DirectDictionaryKeyGeneratorFactory;
 import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
+import org.apache.carbondata.core.metadata.datatype.DataType;
 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.CarbonMeasure;
 import org.apache.carbondata.core.scan.expression.Expression;
 import org.apache.carbondata.core.scan.expression.exception.FilterUnsupportedException;
 import org.apache.carbondata.core.scan.filter.FilterUtil;
@@ -38,6 +45,7 @@ import org.apache.carbondata.core.scan.processor.BlocksChunkHolder;
 import org.apache.carbondata.core.util.BitSetGroup;
 import org.apache.carbondata.core.util.ByteUtil;
 import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.DataTypeUtil;
 
 public class RowLevelRangeLessThanEqualFilterExecuterImpl extends RowLevelFilterExecuterImpl {
   protected byte[][] filterRangeValues;
@@ -46,6 +54,7 @@ public class RowLevelRangeLessThanEqualFilterExecuterImpl extends RowLevelFilter
    * flag to check whether default values is present in the filter value list
    */
   private boolean isDefaultValuePresentInFilter;
+  private int lastDimensionColOrdinal = 0;
   public RowLevelRangeLessThanEqualFilterExecuterImpl(
       List<DimColumnResolvedFilterInfo> dimColEvaluatorInfoList,
       List<MeasureColumnResolvedFilterInfo> msrColEvalutorInfoList, Expression exp,
@@ -53,17 +62,20 @@ public class RowLevelRangeLessThanEqualFilterExecuterImpl extends RowLevelFilter
       SegmentProperties segmentProperties) {
     super(dimColEvaluatorInfoList, msrColEvalutorInfoList, exp, tableIdentifier, segmentProperties,
         null);
+    lastDimensionColOrdinal = segmentProperties.getLastDimensionColOrdinal();
     this.filterRangeValues = filterRangeValues;
     ifDefaultValueMatchesFilter();
-    isNaturalSorted = dimColEvaluatorInfoList.get(0).getDimension().isUseInvertedIndex()
-        && dimColEvaluatorInfoList.get(0).getDimension().isSortColumn();
+    if (isDimensionPresentInCurrentBlock[0] == true) {
+      isNaturalSorted = dimColEvaluatorInfoList.get(0).getDimension().isUseInvertedIndex()
+          && dimColEvaluatorInfoList.get(0).getDimension().isSortColumn();
+    }
   }
 
   /**
    * This method will check whether default value is present in the given filter values
    */
   private void ifDefaultValueMatchesFilter() {
-    if (!this.isDimensionPresentInCurrentBlock[0]) {
+    if (!dimColEvaluatorInfoList.isEmpty() && !isDimensionPresentInCurrentBlock[0]) {
       CarbonDimension dimension = this.dimColEvaluatorInfoList.get(0).getDimension();
       byte[] defaultValue = dimension.getDefaultValue();
       if (null != defaultValue) {
@@ -76,13 +88,38 @@ public class RowLevelRangeLessThanEqualFilterExecuterImpl extends RowLevelFilter
           }
         }
       }
+    } else if (!msrColEvalutorInfoList.isEmpty() && !isMeasurePresentInCurrentBlock[0]) {
+      CarbonMeasure measure = this.msrColEvalutorInfoList.get(0).getMeasure();
+      byte[] defaultValue = measure.getDefaultValue();
+      if (null != defaultValue) {
+        for (int k = 0; k < filterRangeValues.length; k++) {
+          int maxCompare =
+              ByteUtil.UnsafeComparer.INSTANCE.compareTo(filterRangeValues[k], defaultValue);
+          if (maxCompare >= 0) {
+            isDefaultValuePresentInFilter = true;
+            break;
+          }
+        }
+      }
     }
   }
 
   @Override public BitSet isScanRequired(byte[][] blockMaxValue, byte[][] blockMinValue) {
     BitSet bitSet = new BitSet(1);
-    boolean isScanRequired =
-        isScanRequired(blockMinValue[dimensionBlocksIndex[0]], filterRangeValues);
+    byte[] minValue = null;
+    boolean isScanRequired = false;
+    if (isMeasurePresentInCurrentBlock[0] || isDimensionPresentInCurrentBlock[0]) {
+      if (isMeasurePresentInCurrentBlock[0]) {
+        minValue = blockMinValue[measureBlocksIndex[0] + lastDimensionColOrdinal];
+        isScanRequired =
+            isScanRequired(minValue, filterRangeValues, msrColEvalutorInfoList.get(0).getType());
+      } else {
+        minValue = blockMinValue[dimensionBlocksIndex[0]];
+        isScanRequired = isScanRequired(minValue, filterRangeValues);
+      }
+    } else {
+      isScanRequired = isDefaultValuePresentInFilter;
+    }
     if (isScanRequired) {
       bitSet.set(0);
     }
@@ -91,57 +128,147 @@ public class RowLevelRangeLessThanEqualFilterExecuterImpl extends RowLevelFilter
 
   private boolean isScanRequired(byte[] blockMinValue, byte[][] filterValues) {
     boolean isScanRequired = false;
-    if (isDimensionPresentInCurrentBlock[0]) {
-      for (int k = 0; k < filterValues.length; k++) {
-        // and filter-min should be positive
-        int minCompare = ByteUtil.UnsafeComparer.INSTANCE.compareTo(filterValues[k], blockMinValue);
+    for (int k = 0; k < filterValues.length; k++) {
+      // and filter-min should be positive
+      int minCompare = ByteUtil.UnsafeComparer.INSTANCE.compareTo(filterValues[k], blockMinValue);
 
-        // if any filter applied is not in range of min and max of block
-        // then since its a less than equal to fiter validate whether the block
-        // min range is less than equal to applied filter member
-        if (minCompare >= 0) {
-          isScanRequired = true;
-          break;
-        }
+      // if any filter applied is not in range of min and max of block
+      // then since its a less than equal to fiter validate whether the block
+      // min range is less than equal to applied filter member
+      if (minCompare >= 0) {
+        isScanRequired = true;
+        break;
       }
-    } else {
-      isScanRequired = isDefaultValuePresentInFilter;
     }
     return isScanRequired;
   }
 
+  private boolean isScanRequired(byte[] minValue, byte[][] filterValue,
+      DataType dataType) {
+    for (int i = 0; i < filterValue.length; i++) {
+      if (filterValue[i].length == 0 || minValue.length == 0) {
+        return isScanRequired(minValue, filterValue);
+      }
+      switch (dataType) {
+        case DOUBLE:
+          double minValueDouble = ByteBuffer.wrap(minValue).getDouble();
+          double filterValueDouble = ByteBuffer.wrap(filterValue[i]).getDouble();
+          if (filterValueDouble >= minValueDouble) {
+            return true;
+          }
+          break;
+        case INT:
+        case SHORT:
+        case LONG:
+          long minValueLong = ByteBuffer.wrap(minValue).getLong();
+          long filterValueLong = ByteBuffer.wrap(filterValue[i]).getLong();
+          if (filterValueLong >= minValueLong) {
+            return true;
+          }
+          break;
+        case DECIMAL:
+          BigDecimal minDecimal = DataTypeUtil.byteToBigDecimal(minValue);
+          BigDecimal filterDecimal = DataTypeUtil.byteToBigDecimal(filterValue[i]);
+          if (filterDecimal.compareTo(minDecimal) >= 0) {
+            return true;
+          }
+      }
+    }
+    return false;
+  }
+
   @Override public BitSetGroup applyFilter(BlocksChunkHolder blockChunkHolder)
       throws FilterUnsupportedException, IOException {
     // select all rows if dimension does not exists in the current block
-    if (!isDimensionPresentInCurrentBlock[0]) {
+    if (!isDimensionPresentInCurrentBlock[0] && !isMeasurePresentInCurrentBlock[0]) {
       int numberOfRows = blockChunkHolder.getDataBlock().nodeSize();
       return FilterUtil
           .createBitSetGroupWithDefaultValue(blockChunkHolder.getDataBlock().numberOfPages(),
               numberOfRows, true);
     }
-    int blockIndex =
-        segmentProperties.getDimensionOrdinalToBlockMapping().get(dimensionBlocksIndex[0]);
-    if (null == blockChunkHolder.getDimensionRawDataChunk()[blockIndex]) {
-      blockChunkHolder.getDimensionRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
-          .getDimensionChunk(blockChunkHolder.getFileReader(), blockIndex);
-    }
-    DimensionRawColumnChunk rawColumnChunk =
-        blockChunkHolder.getDimensionRawDataChunk()[blockIndex];
-    BitSetGroup bitSetGroup = new BitSetGroup(rawColumnChunk.getPagesCount());
-    for (int i = 0; i < rawColumnChunk.getPagesCount(); i++) {
-      if (rawColumnChunk.getMinValues() != null) {
-        if (isScanRequired(rawColumnChunk.getMinValues()[i], this.filterRangeValues)) {
+    if (isDimensionPresentInCurrentBlock[0]) {
+      int blockIndex =
+          segmentProperties.getDimensionOrdinalToBlockMapping().get(dimensionBlocksIndex[0]);
+      if (null == blockChunkHolder.getDimensionRawDataChunk()[blockIndex]) {
+        blockChunkHolder.getDimensionRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
+            .getDimensionChunk(blockChunkHolder.getFileReader(), blockIndex);
+      }
+      DimensionRawColumnChunk rawColumnChunk =
+          blockChunkHolder.getDimensionRawDataChunk()[blockIndex];
+      BitSetGroup bitSetGroup = new BitSetGroup(rawColumnChunk.getPagesCount());
+      for (int i = 0; i < rawColumnChunk.getPagesCount(); i++) {
+        if (rawColumnChunk.getMinValues() != null) {
+          if (isScanRequired(rawColumnChunk.getMinValues()[i], this.filterRangeValues)) {
+            BitSet bitSet = getFilteredIndexes(rawColumnChunk.convertToDimColDataChunk(i),
+                rawColumnChunk.getRowCount()[i]);
+            bitSetGroup.setBitSet(bitSet, i);
+          }
+        } else {
           BitSet bitSet = getFilteredIndexes(rawColumnChunk.convertToDimColDataChunk(i),
               rawColumnChunk.getRowCount()[i]);
           bitSetGroup.setBitSet(bitSet, i);
         }
-      } else {
-        BitSet bitSet = getFilteredIndexes(rawColumnChunk.convertToDimColDataChunk(i),
-            rawColumnChunk.getRowCount()[i]);
-        bitSetGroup.setBitSet(bitSet, i);
+      }
+      return bitSetGroup;
+    } else if (isMeasurePresentInCurrentBlock[0]) {
+      int blockIndex =
+          segmentProperties.getMeasuresOrdinalToBlockMapping().get(measureBlocksIndex[0]);
+      if (null == blockChunkHolder.getMeasureRawDataChunk()[blockIndex]) {
+        blockChunkHolder.getMeasureRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
+            .getMeasureChunk(blockChunkHolder.getFileReader(), blockIndex);
+      }
+      MeasureRawColumnChunk rawColumnChunk = blockChunkHolder.getMeasureRawDataChunk()[blockIndex];
+      BitSetGroup bitSetGroup = new BitSetGroup(rawColumnChunk.getPagesCount());
+      for (int i = 0; i < rawColumnChunk.getPagesCount(); i++) {
+        if (rawColumnChunk.getMinValues() != null) {
+          if (isScanRequired(rawColumnChunk.getMinValues()[i], this.filterRangeValues,
+              msrColEvalutorInfoList.get(0).getType())) {
+            BitSet bitSet =
+                getFilteredIndexesForMeasures(rawColumnChunk.convertToMeasureColDataChunk(i),
+                    rawColumnChunk.getRowCount()[i]);
+            bitSetGroup.setBitSet(bitSet, i);
+          }
+        } else {
+          BitSet bitSet =
+              getFilteredIndexesForMeasures(rawColumnChunk.convertToMeasureColDataChunk(i),
+                  rawColumnChunk.getRowCount()[i]);
+          bitSetGroup.setBitSet(bitSet, i);
+        }
+      }
+      return bitSetGroup;
+    }
+    return null;
+  }
+
+  private BitSet getFilteredIndexesForMeasures(MeasureColumnDataChunk measureColumnDataChunk,
+      int numerOfRows) {
+    BitSet bitSet = new BitSet(numerOfRows);
+    byte[][] filterValues = this.filterRangeValues;
+    DataType msrType = msrColEvalutorInfoList.get(0).getType();
+    Comparator comparator = FilterUtil.getComparatorByDataTypeForMeasure(msrType);
+    for (int i = 0; i < filterValues.length; i++) {
+      if (filterValues[i].length == 0) {
+        BitSet nullBitSet = measureColumnDataChunk.getNullValueIndexHolder().getBitSet();
+        for (int j = nullBitSet.nextSetBit(0); j >= 0; j = nullBitSet.nextSetBit(j + 1)) {
+          bitSet.set(j);
+        }
+        continue;
+      }
+      Object filter = DataTypeUtil.getMeasureObjectFromDataType(filterValues[i], msrType);
+      for (int startIndex = 0; startIndex < numerOfRows; startIndex++) {
+        Object msrValue = DataTypeUtil
+            .getMeasureObjectBasedOnDataType(measureColumnDataChunk, startIndex,
+                 msrColEvalutorInfoList.get(0).getMeasure());
+
+        if (comparator.compare(msrValue, filter) <= 0) {
+          // This is a match.
+          bitSet.set(startIndex);
+        }
       }
     }
-    return bitSetGroup;
+    CarbonUtil
+        .updateBitSetForNull(measureColumnDataChunk.getNullValueIndexHolder().getBitSet(), bitSet);
+    return bitSet;
   }
 
   private BitSet getFilteredIndexes(DimensionColumnDataChunk dimensionColumnDataChunk,
@@ -322,6 +449,12 @@ public class RowLevelRangeLessThanEqualFilterExecuterImpl extends RowLevelFilter
         blockChunkHolder.getDimensionRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
             .getDimensionChunk(blockChunkHolder.getFileReader(), blockIndex);
       }
+    } else if (isMeasurePresentInCurrentBlock[0]) {
+      int blockIndex = measureBlocksIndex[0];
+      if (null == blockChunkHolder.getMeasureRawDataChunk()[blockIndex]) {
+        blockChunkHolder.getMeasureRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
+            .getMeasureChunk(blockChunkHolder.getFileReader(), blockIndex);
+      }
     }
   }
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/17db292a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeLessThanFiterExecuterImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeLessThanFiterExecuterImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeLessThanFiterExecuterImpl.java
index e9b6408..a600118 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeLessThanFiterExecuterImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/executer/RowLevelRangeLessThanFiterExecuterImpl.java
@@ -17,18 +17,25 @@
 package org.apache.carbondata.core.scan.filter.executer;
 
 import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
 import java.util.BitSet;
+import java.util.Comparator;
 import java.util.List;
 
 import org.apache.carbondata.core.constants.CarbonCommonConstants;
 import org.apache.carbondata.core.datastore.block.SegmentProperties;
 import org.apache.carbondata.core.datastore.chunk.DimensionColumnDataChunk;
+import org.apache.carbondata.core.datastore.chunk.MeasureColumnDataChunk;
 import org.apache.carbondata.core.datastore.chunk.impl.DimensionRawColumnChunk;
+import org.apache.carbondata.core.datastore.chunk.impl.MeasureRawColumnChunk;
 import org.apache.carbondata.core.keygenerator.directdictionary.DirectDictionaryGenerator;
 import org.apache.carbondata.core.keygenerator.directdictionary.DirectDictionaryKeyGeneratorFactory;
 import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
+import org.apache.carbondata.core.metadata.datatype.DataType;
 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.CarbonMeasure;
 import org.apache.carbondata.core.scan.expression.Expression;
 import org.apache.carbondata.core.scan.expression.exception.FilterUnsupportedException;
 import org.apache.carbondata.core.scan.filter.FilterUtil;
@@ -38,6 +45,7 @@ import org.apache.carbondata.core.scan.processor.BlocksChunkHolder;
 import org.apache.carbondata.core.util.BitSetGroup;
 import org.apache.carbondata.core.util.ByteUtil;
 import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.carbondata.core.util.DataTypeUtil;
 
 public class RowLevelRangeLessThanFiterExecuterImpl extends RowLevelFilterExecuterImpl {
   private byte[][] filterRangeValues;
@@ -46,7 +54,7 @@ public class RowLevelRangeLessThanFiterExecuterImpl extends RowLevelFilterExecut
    * flag to check whether default values is present in the filter value list
    */
   private boolean isDefaultValuePresentInFilter;
-
+  private int lastDimensionColOrdinal = 0;
   public RowLevelRangeLessThanFiterExecuterImpl(
       List<DimColumnResolvedFilterInfo> dimColEvaluatorInfoList,
       List<MeasureColumnResolvedFilterInfo> msrColEvalutorInfoList, Expression exp,
@@ -55,16 +63,19 @@ public class RowLevelRangeLessThanFiterExecuterImpl extends RowLevelFilterExecut
     super(dimColEvaluatorInfoList, msrColEvalutorInfoList, exp, tableIdentifier, segmentProperties,
         null);
     this.filterRangeValues = filterRangeValues;
+    lastDimensionColOrdinal = segmentProperties.getLastDimensionColOrdinal();
     ifDefaultValueMatchesFilter();
-    isNaturalSorted = dimColEvaluatorInfoList.get(0).getDimension().isUseInvertedIndex()
-        && dimColEvaluatorInfoList.get(0).getDimension().isSortColumn();
+    if (isDimensionPresentInCurrentBlock[0] == true) {
+      isNaturalSorted = dimColEvaluatorInfoList.get(0).getDimension().isUseInvertedIndex()
+          && dimColEvaluatorInfoList.get(0).getDimension().isSortColumn();
+    }
   }
 
   /**
    * This method will check whether default value is present in the given filter values
    */
   private void ifDefaultValueMatchesFilter() {
-    if (!this.isDimensionPresentInCurrentBlock[0]) {
+    if (!dimColEvaluatorInfoList.isEmpty() && !isDimensionPresentInCurrentBlock[0]) {
       CarbonDimension dimension = this.dimColEvaluatorInfoList.get(0).getDimension();
       byte[] defaultValue = dimension.getDefaultValue();
       if (null != defaultValue) {
@@ -77,72 +88,188 @@ public class RowLevelRangeLessThanFiterExecuterImpl extends RowLevelFilterExecut
           }
         }
       }
+    } else if (!msrColEvalutorInfoList.isEmpty() && !isMeasurePresentInCurrentBlock[0]) {
+      CarbonMeasure measure = this.msrColEvalutorInfoList.get(0).getMeasure();
+      byte[] defaultValue = measure.getDefaultValue();
+      if (null != defaultValue) {
+        for (int k = 0; k < filterRangeValues.length; k++) {
+          int maxCompare =
+              ByteUtil.UnsafeComparer.INSTANCE.compareTo(filterRangeValues[k], defaultValue);
+          if (maxCompare > 0) {
+            isDefaultValuePresentInFilter = true;
+            break;
+          }
+        }
+      }
     }
   }
 
   @Override public BitSet isScanRequired(byte[][] blockMaxValue, byte[][] blockMinValue) {
     BitSet bitSet = new BitSet(1);
-    boolean isScanRequired =
-        isScanRequired(blockMinValue[dimensionBlocksIndex[0]], filterRangeValues);
+    byte[] minValue = null;
+    boolean isScanRequired = false;
+    if (isMeasurePresentInCurrentBlock[0] || isDimensionPresentInCurrentBlock[0]) {
+      if (isMeasurePresentInCurrentBlock[0]) {
+        minValue = blockMinValue[measureBlocksIndex[0] + lastDimensionColOrdinal];
+        isScanRequired =
+            isScanRequired(minValue, filterRangeValues, msrColEvalutorInfoList.get(0).getType());
+      } else {
+        minValue = blockMinValue[dimensionBlocksIndex[0]];
+        isScanRequired = isScanRequired(minValue, filterRangeValues);
+      }
+    } else {
+      isScanRequired = isDefaultValuePresentInFilter;
+    }
     if (isScanRequired) {
       bitSet.set(0);
     }
     return bitSet;
   }
 
+
   private boolean isScanRequired(byte[] blockMinValue, byte[][] filterValues) {
     boolean isScanRequired = false;
-    if (isDimensionPresentInCurrentBlock[0]) {
-      for (int k = 0; k < filterValues.length; k++) {
-        // and filter-min should be positive
-        int minCompare = ByteUtil.UnsafeComparer.INSTANCE.compareTo(filterValues[k], blockMinValue);
+    for (int k = 0; k < filterValues.length; k++) {
+      // and filter-min should be positive
+      int minCompare = ByteUtil.UnsafeComparer.INSTANCE.compareTo(filterValues[k], blockMinValue);
 
-        // if any filter applied is not in range of min and max of block
-        // then since its a less than fiter validate whether the block
-        // min range is less  than applied filter member
-        if (minCompare > 0) {
-          isScanRequired = true;
-          break;
-        }
+      // if any filter applied is not in range of min and max of block
+      // then since its a less than equal to fiter validate whether the block
+      // min range is less than equal to applied filter member
+      if (minCompare > 0) {
+        isScanRequired = true;
+        break;
       }
-    } else {
-      isScanRequired = isDefaultValuePresentInFilter;
     }
     return isScanRequired;
   }
 
+  private boolean isScanRequired(byte[] minValue, byte[][] filterValue,
+      DataType dataType) {
+    for (int i = 0; i < filterValue.length; i++) {
+      if (filterValue[i].length == 0 || minValue.length == 0) {
+        return isScanRequired(minValue, filterValue);
+      }
+      switch (dataType) {
+        case DOUBLE:
+          double minValueDouble = ByteBuffer.wrap(minValue).getDouble();
+          double filterValueDouble = ByteBuffer.wrap(filterValue[i]).getDouble();
+          if (filterValueDouble > minValueDouble) {
+            return true;
+          }
+          break;
+        case INT:
+        case SHORT:
+        case LONG:
+          long minValueLong = ByteBuffer.wrap(minValue).getLong();
+          long filterValueLong = ByteBuffer.wrap(filterValue[i]).getLong();
+          if (filterValueLong > minValueLong) {
+            return true;
+          }
+          break;
+        case DECIMAL:
+          BigDecimal minDecimal = DataTypeUtil.byteToBigDecimal(minValue);
+          BigDecimal filterDecimal = DataTypeUtil.byteToBigDecimal(filterValue[i]);
+          if (filterDecimal.compareTo(minDecimal) > 0) {
+            return true;
+          }
+      }
+    }
+    return false;
+  }
+
   @Override public BitSetGroup applyFilter(BlocksChunkHolder blockChunkHolder)
       throws FilterUnsupportedException, IOException {
     // select all rows if dimension does not exists in the current block
-    if (!isDimensionPresentInCurrentBlock[0]) {
+    if (!isDimensionPresentInCurrentBlock[0] && !isMeasurePresentInCurrentBlock[0]) {
       int numberOfRows = blockChunkHolder.getDataBlock().nodeSize();
       return FilterUtil
           .createBitSetGroupWithDefaultValue(blockChunkHolder.getDataBlock().numberOfPages(),
               numberOfRows, true);
     }
-    int blockIndex =
-        segmentProperties.getDimensionOrdinalToBlockMapping().get(dimensionBlocksIndex[0]);
-    if (null == blockChunkHolder.getDimensionRawDataChunk()[blockIndex]) {
-      blockChunkHolder.getDimensionRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
-          .getDimensionChunk(blockChunkHolder.getFileReader(), blockIndex);
-    }
-    DimensionRawColumnChunk rawColumnChunk =
-        blockChunkHolder.getDimensionRawDataChunk()[blockIndex];
-    BitSetGroup bitSetGroup = new BitSetGroup(rawColumnChunk.getPagesCount());
-    for (int i = 0; i < rawColumnChunk.getPagesCount(); i++) {
-      if (rawColumnChunk.getMinValues() != null) {
-        if (isScanRequired(rawColumnChunk.getMinValues()[i], this.filterRangeValues)) {
+    if (isDimensionPresentInCurrentBlock[0]) {
+      int blockIndex =
+          segmentProperties.getDimensionOrdinalToBlockMapping().get(dimensionBlocksIndex[0]);
+      if (null == blockChunkHolder.getDimensionRawDataChunk()[blockIndex]) {
+        blockChunkHolder.getDimensionRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
+            .getDimensionChunk(blockChunkHolder.getFileReader(), blockIndex);
+      }
+      DimensionRawColumnChunk rawColumnChunk =
+          blockChunkHolder.getDimensionRawDataChunk()[blockIndex];
+      BitSetGroup bitSetGroup = new BitSetGroup(rawColumnChunk.getPagesCount());
+      for (int i = 0; i < rawColumnChunk.getPagesCount(); i++) {
+        if (rawColumnChunk.getMinValues() != null) {
+          if (isScanRequired(rawColumnChunk.getMinValues()[i], this.filterRangeValues)) {
+            BitSet bitSet = getFilteredIndexes(rawColumnChunk.convertToDimColDataChunk(i),
+                rawColumnChunk.getRowCount()[i]);
+            bitSetGroup.setBitSet(bitSet, i);
+          }
+        } else {
           BitSet bitSet = getFilteredIndexes(rawColumnChunk.convertToDimColDataChunk(i),
               rawColumnChunk.getRowCount()[i]);
           bitSetGroup.setBitSet(bitSet, i);
         }
-      } else {
-        BitSet bitSet = getFilteredIndexes(rawColumnChunk.convertToDimColDataChunk(i),
-            rawColumnChunk.getRowCount()[i]);
-        bitSetGroup.setBitSet(bitSet, i);
+      }
+      return bitSetGroup;
+    } else if (isMeasurePresentInCurrentBlock[0]) {
+      int blockIndex =
+          segmentProperties.getMeasuresOrdinalToBlockMapping().get(measureBlocksIndex[0]);
+      if (null == blockChunkHolder.getMeasureRawDataChunk()[blockIndex]) {
+        blockChunkHolder.getMeasureRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
+            .getMeasureChunk(blockChunkHolder.getFileReader(), blockIndex);
+      }
+      MeasureRawColumnChunk rawColumnChunk = blockChunkHolder.getMeasureRawDataChunk()[blockIndex];
+      BitSetGroup bitSetGroup = new BitSetGroup(rawColumnChunk.getPagesCount());
+      for (int i = 0; i < rawColumnChunk.getPagesCount(); i++) {
+        if (rawColumnChunk.getMinValues() != null) {
+          if (isScanRequired(rawColumnChunk.getMinValues()[i], this.filterRangeValues,
+              msrColEvalutorInfoList.get(0).getType())) {
+            BitSet bitSet =
+                getFilteredIndexesForMeasures(rawColumnChunk.convertToMeasureColDataChunk(i),
+                    rawColumnChunk.getRowCount()[i]);
+            bitSetGroup.setBitSet(bitSet, i);
+          }
+        } else {
+          BitSet bitSet =
+              getFilteredIndexesForMeasures(rawColumnChunk.convertToMeasureColDataChunk(i),
+                  rawColumnChunk.getRowCount()[i]);
+          bitSetGroup.setBitSet(bitSet, i);
+        }
+      }
+      return bitSetGroup;
+    }
+    return null;
+  }
+
+  private BitSet getFilteredIndexesForMeasures(MeasureColumnDataChunk measureColumnDataChunk,
+      int numerOfRows) {
+    BitSet bitSet = new BitSet(numerOfRows);
+    byte[][] filterValues = this.filterRangeValues;
+    DataType msrType = msrColEvalutorInfoList.get(0).getType();
+    Comparator comparator = FilterUtil.getComparatorByDataTypeForMeasure(msrType);
+    for (int i = 0; i < filterValues.length; i++) {
+      if (filterValues[i].length == 0) {
+        BitSet nullBitSet = measureColumnDataChunk.getNullValueIndexHolder().getBitSet();
+        for (int j = nullBitSet.nextSetBit(0); j >= 0; j = nullBitSet.nextSetBit(j + 1)) {
+          bitSet.set(j);
+        }
+        continue;
+      }
+      Object filter = DataTypeUtil.getMeasureObjectFromDataType(filterValues[i], msrType);
+      for (int startIndex = 0; startIndex < numerOfRows; startIndex++) {
+        Object msrValue = DataTypeUtil
+            .getMeasureObjectBasedOnDataType(measureColumnDataChunk, startIndex,
+                 msrColEvalutorInfoList.get(0).getMeasure());
+
+        if (comparator.compare(msrValue, filter) < 0) {
+          // This is a match.
+          bitSet.set(startIndex);
+        }
       }
     }
-    return bitSetGroup;
+    CarbonUtil
+        .updateBitSetForNull(measureColumnDataChunk.getNullValueIndexHolder().getBitSet(), bitSet);
+    return bitSet;
   }
 
   private BitSet getFilteredIndexes(DimensionColumnDataChunk dimensionColumnDataChunk,
@@ -337,6 +464,12 @@ public class RowLevelRangeLessThanFiterExecuterImpl extends RowLevelFilterExecut
         blockChunkHolder.getDimensionRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
             .getDimensionChunk(blockChunkHolder.getFileReader(), blockIndex);
       }
+    } else if (isMeasurePresentInCurrentBlock[0]) {
+      int blockIndex = measureBlocksIndex[0];
+      if (null == blockChunkHolder.getMeasureRawDataChunk()[blockIndex]) {
+        blockChunkHolder.getMeasureRawDataChunk()[blockIndex] = blockChunkHolder.getDataBlock()
+            .getMeasureChunk(blockChunkHolder.getFileReader(), blockIndex);
+      }
     }
   }
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/17db292a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/ConditionalFilterResolverImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/ConditionalFilterResolverImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/ConditionalFilterResolverImpl.java
index 609a130..705cafb 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/ConditionalFilterResolverImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/ConditionalFilterResolverImpl.java
@@ -24,7 +24,6 @@ import org.apache.carbondata.core.datastore.block.SegmentProperties;
 import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
 import org.apache.carbondata.core.metadata.datatype.DataType;
 import org.apache.carbondata.core.metadata.encoder.Encoding;
-import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
 import org.apache.carbondata.core.scan.expression.ColumnExpression;
 import org.apache.carbondata.core.scan.expression.Expression;
 import org.apache.carbondata.core.scan.expression.conditional.BinaryConditionalExpression;
@@ -35,6 +34,7 @@ import org.apache.carbondata.core.scan.filter.FilterUtil;
 import org.apache.carbondata.core.scan.filter.intf.FilterExecuterType;
 import org.apache.carbondata.core.scan.filter.resolver.metadata.FilterResolverMetadata;
 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.scan.filter.resolver.resolverinfo.visitor.FilterInfoTypeVisitorFactory;
 
 public class ConditionalFilterResolverImpl implements FilterResolverIntf {
@@ -44,16 +44,20 @@ public class ConditionalFilterResolverImpl implements FilterResolverIntf {
   protected boolean isExpressionResolve;
   protected boolean isIncludeFilter;
   private DimColumnResolvedFilterInfo dimColResolvedFilterInfo;
+  private MeasureColumnResolvedFilterInfo msrColResolvedFilterInfo;
   private AbsoluteTableIdentifier tableIdentifier;
 
   public ConditionalFilterResolverImpl(Expression exp, boolean isExpressionResolve,
-      boolean isIncludeFilter, AbsoluteTableIdentifier tableIdentifier) {
+      boolean isIncludeFilter, AbsoluteTableIdentifier tableIdentifier, boolean isMeasure) {
     this.exp = exp;
     this.isExpressionResolve = isExpressionResolve;
     this.isIncludeFilter = isIncludeFilter;
     this.tableIdentifier = tableIdentifier;
-    this.dimColResolvedFilterInfo = new DimColumnResolvedFilterInfo();
-
+    if (isMeasure == false) {
+      this.dimColResolvedFilterInfo = new DimColumnResolvedFilterInfo();
+    } else {
+      this.msrColResolvedFilterInfo = new MeasureColumnResolvedFilterInfo();
+    }
   }
 
 
@@ -83,11 +87,11 @@ public class ConditionalFilterResolverImpl implements FilterResolverIntf {
         // column expression.
         // we need to check if the other expression contains column
         // expression or not in depth.
-        CarbonDimension dimension = columnExpression.getDimension();
         if (FilterUtil.checkIfExpressionContainsColumn(rightExp)
-            || FilterUtil.isExpressionNeedsToResolved(rightExp, isIncludeFilter) &&
-            dimension.hasEncoding(Encoding.DICTIONARY) && !dimension
-            .hasEncoding(Encoding.DIRECT_DICTIONARY)) {
+            || FilterUtil.isExpressionNeedsToResolved(rightExp, isIncludeFilter) && ((
+            (null != columnExpression.getDimension()) && (columnExpression.getDimension()
+                .hasEncoding(Encoding.DICTIONARY) && !columnExpression.getDimension()
+                .hasEncoding(Encoding.DIRECT_DICTIONARY))))) {
           isExpressionResolve = true;
         } else {
           //Visitor pattern is been used in this scenario inorder to populate the
@@ -96,9 +100,16 @@ public class ConditionalFilterResolverImpl implements FilterResolverIntf {
           //3 types of visitors custom,direct and no dictionary, all types of visitor populate
           //the visitable instance as per its buisness logic which is different for all the
           // visitors.
-          dimColResolvedFilterInfo.populateFilterInfoBasedOnColumnType(
-              FilterInfoTypeVisitorFactory.getResolvedFilterInfoVisitor(columnExpression, exp),
-              metadata);
+          if (columnExpression.isMeasure()) {
+            msrColResolvedFilterInfo.setMeasure(columnExpression.getMeasure());
+            msrColResolvedFilterInfo.populateFilterInfoBasedOnColumnType(
+                FilterInfoTypeVisitorFactory.getResolvedFilterInfoVisitor(columnExpression, exp),
+                metadata);
+          } else {
+            dimColResolvedFilterInfo.populateFilterInfoBasedOnColumnType(
+                FilterInfoTypeVisitorFactory.getResolvedFilterInfoVisitor(columnExpression, exp),
+                metadata);
+          }
         }
       } else if (rightExp instanceof ColumnExpression) {
         ColumnExpression columnExpression = (ColumnExpression) rightExp;
@@ -119,10 +130,15 @@ public class ConditionalFilterResolverImpl implements FilterResolverIntf {
             isExpressionResolve = true;
           } else {
 
-            dimColResolvedFilterInfo.populateFilterInfoBasedOnColumnType(
-                FilterInfoTypeVisitorFactory.getResolvedFilterInfoVisitor(columnExpression, exp),
-                metadata);
-
+            if (columnExpression.isMeasure()) {
+              msrColResolvedFilterInfo.populateFilterInfoBasedOnColumnType(
+                  FilterInfoTypeVisitorFactory.getResolvedFilterInfoVisitor(columnExpression, exp),
+                  metadata);
+            } else {
+              dimColResolvedFilterInfo.populateFilterInfoBasedOnColumnType(
+                  FilterInfoTypeVisitorFactory.getResolvedFilterInfoVisitor(columnExpression, exp),
+                  metadata);
+            }
           }
         }
       } else {
@@ -135,24 +151,34 @@ public class ConditionalFilterResolverImpl implements FilterResolverIntf {
       metadata.setColumnExpression(columnList.get(0));
       metadata.setExpression(exp);
       metadata.setIncludeFilter(isIncludeFilter);
-      if (!columnList.get(0).getDimension().hasEncoding(Encoding.DICTIONARY) || columnList.get(0)
-          .getDimension().hasEncoding(Encoding.DIRECT_DICTIONARY)
+      if ((null != columnList.get(0).getDimension()) && (
+          !columnList.get(0).getDimension().hasEncoding(Encoding.DICTIONARY) || columnList.get(0)
+              .getDimension().hasEncoding(Encoding.DIRECT_DICTIONARY))
           || (exp instanceof RangeExpression)) {
         dimColResolvedFilterInfo.populateFilterInfoBasedOnColumnType(
             FilterInfoTypeVisitorFactory.getResolvedFilterInfoVisitor(columnList.get(0), exp),
             metadata);
 
-      } else if (columnList.get(0).getDimension().hasEncoding(Encoding.DICTIONARY) && !(
-          columnList.get(0).getDimension().getDataType()
-              == org.apache.carbondata.core.metadata.datatype.DataType.STRUCT
-              || columnList.get(0).getDimension().getDataType()
-              == org.apache.carbondata.core.metadata.datatype.DataType.ARRAY)) {
+      } else if ((null != columnList.get(0).getDimension()) && (
+          columnList.get(0).getDimension().hasEncoding(Encoding.DICTIONARY) && !(
+              columnList.get(0).getDimension().getDataType()
+                  == org.apache.carbondata.core.metadata.datatype.DataType.STRUCT
+                  || columnList.get(0).getDimension().getDataType()
+                  == org.apache.carbondata.core.metadata.datatype.DataType.ARRAY))) {
         dimColResolvedFilterInfo.setFilterValues(FilterUtil
             .getFilterListForAllValues(absoluteTableIdentifier, exp, columnList.get(0),
                 isIncludeFilter));
 
         dimColResolvedFilterInfo.setColumnIndex(columnList.get(0).getDimension().getOrdinal());
         dimColResolvedFilterInfo.setDimension(columnList.get(0).getDimension());
+      } else if (columnList.get(0).isMeasure()) {
+        msrColResolvedFilterInfo.setMeasure(columnList.get(0).getMeasure());
+        msrColResolvedFilterInfo.populateFilterInfoBasedOnColumnType(
+            FilterInfoTypeVisitorFactory.getResolvedFilterInfoVisitor(columnList.get(0), exp),
+            metadata);
+        msrColResolvedFilterInfo.setCarbonColumn(columnList.get(0).getCarbonColumn());
+        msrColResolvedFilterInfo.setColumnIndex(columnList.get(0).getCarbonColumn().getOrdinal());
+        msrColResolvedFilterInfo.setType(columnList.get(0).getCarbonColumn().getDataType());
       }
     }
 
@@ -189,6 +215,17 @@ public class ConditionalFilterResolverImpl implements FilterResolverIntf {
     return dimColResolvedFilterInfo;
   }
 
+  /**
+   * Method will return the MeasureColumnResolvedFilterInfo instance which consists
+   * the mapping of the respective dimension and its surrogates involved in
+   * filter expression.
+   *
+   * @return DimColumnResolvedFilterInfo
+   */
+  public MeasureColumnResolvedFilterInfo getMsrColResolvedFilterInfo() {
+    return msrColResolvedFilterInfo;
+  }
+
   public AbsoluteTableIdentifier getTableIdentifier() {
     return tableIdentifier;
   }
@@ -198,10 +235,12 @@ public class ConditionalFilterResolverImpl implements FilterResolverIntf {
    */
   public void getStartKey(SegmentProperties segmentProperties, long[] startKey,
       SortedMap<Integer, byte[]> setOfStartKeyByteArray, List<long[]> startKeyList) {
-    FilterUtil.getStartKey(dimColResolvedFilterInfo.getDimensionResolvedFilterInstance(),
-        segmentProperties, startKey, startKeyList);
-    FilterUtil.getStartKeyForNoDictionaryDimension(dimColResolvedFilterInfo,
-        segmentProperties, setOfStartKeyByteArray);
+    if (null != dimColResolvedFilterInfo) {
+      FilterUtil.getStartKey(dimColResolvedFilterInfo.getDimensionResolvedFilterInstance(),
+          segmentProperties, startKey, startKeyList);
+      FilterUtil.getStartKeyForNoDictionaryDimension(dimColResolvedFilterInfo, segmentProperties,
+          setOfStartKeyByteArray);
+    }
   }
 
   /**
@@ -209,10 +248,12 @@ public class ConditionalFilterResolverImpl implements FilterResolverIntf {
    */
   @Override public void getEndKey(SegmentProperties segmentProperties, long[] endKeys,
       SortedMap<Integer, byte[]> setOfEndKeyByteArray, List<long[]> endKeyList) {
-    FilterUtil.getEndKey(dimColResolvedFilterInfo.getDimensionResolvedFilterInstance(), endKeys,
-        segmentProperties, endKeyList);
-    FilterUtil.getEndKeyForNoDictionaryDimension(dimColResolvedFilterInfo, segmentProperties,
-        setOfEndKeyByteArray);
+    if (null != dimColResolvedFilterInfo) {
+      FilterUtil.getEndKey(dimColResolvedFilterInfo.getDimensionResolvedFilterInstance(), endKeys,
+          segmentProperties, endKeyList);
+      FilterUtil.getEndKeyForNoDictionaryDimension(dimColResolvedFilterInfo, segmentProperties,
+          setOfEndKeyByteArray);
+    }
   }
 
   /**
@@ -258,7 +299,7 @@ public class ConditionalFilterResolverImpl implements FilterResolverIntf {
    *
    * @return
    */
-  public byte[][] getFilterRangeValues(SegmentProperties segmentProperties) {
+  public byte[][]  getFilterRangeValues(SegmentProperties segmentProperties) {
 
     if (null != dimColResolvedFilterInfo.getFilterValues() && !dimColResolvedFilterInfo
         .getDimension().hasEncoding(Encoding.DICTIONARY)) {
@@ -268,7 +309,7 @@ public class ConditionalFilterResolverImpl implements FilterResolverIntf {
     } else if (null != dimColResolvedFilterInfo.getFilterValues() && dimColResolvedFilterInfo
         .getDimension().hasEncoding(Encoding.DIRECT_DICTIONARY)) {
       return FilterUtil.getKeyArray(this.dimColResolvedFilterInfo.getFilterValues(),
-          this.dimColResolvedFilterInfo.getDimension(), segmentProperties);
+          this.dimColResolvedFilterInfo.getDimension(), null, segmentProperties);
     }
     return null;
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/17db292a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/FilterResolverIntf.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/FilterResolverIntf.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/FilterResolverIntf.java
index 746b96d..89a3890 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/FilterResolverIntf.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/FilterResolverIntf.java
@@ -27,6 +27,7 @@ import org.apache.carbondata.core.scan.expression.Expression;
 import org.apache.carbondata.core.scan.expression.exception.FilterUnsupportedException;
 import org.apache.carbondata.core.scan.filter.intf.FilterExecuterType;
 import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.DimColumnResolvedFilterInfo;
+import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.MeasureColumnResolvedFilterInfo;
 
 public interface FilterResolverIntf extends Serializable {
 
@@ -66,6 +67,14 @@ public interface FilterResolverIntf extends Serializable {
   DimColumnResolvedFilterInfo getDimColResolvedFilterInfo();
 
   /**
+   * API will return the resolved filter instance, this instance will provide
+   * the resolved surrogates based on the applied filter
+   *
+   * @return MeasureColumnResolvedFilterInfo object
+   */
+  MeasureColumnResolvedFilterInfo getMsrColResolvedFilterInfo();
+
+  /**
    * API will get the start key based on the filter applied based on the key generator
    *
    * @param segmentProperties

http://git-wip-us.apache.org/repos/asf/carbondata/blob/17db292a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/LogicalFilterResolverImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/LogicalFilterResolverImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/LogicalFilterResolverImpl.java
index db35823..368cd9a 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/LogicalFilterResolverImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/LogicalFilterResolverImpl.java
@@ -27,6 +27,7 @@ import org.apache.carbondata.core.scan.expression.Expression;
 import org.apache.carbondata.core.scan.filter.intf.ExpressionType;
 import org.apache.carbondata.core.scan.filter.intf.FilterExecuterType;
 import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.DimColumnResolvedFilterInfo;
+import org.apache.carbondata.core.scan.filter.resolver.resolverinfo.MeasureColumnResolvedFilterInfo;
 
 public class LogicalFilterResolverImpl implements FilterResolverIntf {
   /**
@@ -84,6 +85,9 @@ public class LogicalFilterResolverImpl implements FilterResolverIntf {
     return null;
   }
 
+  @Override public MeasureColumnResolvedFilterInfo getMsrColResolvedFilterInfo() {
+    return null;
+  }
   @Override public void getStartKey(SegmentProperties segmentProperties, long[] startKey,
       SortedMap<Integer, byte[]> setOfStartKeyByteArray, List<long[]> startKeyList) {
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/17db292a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/RowLevelFilterResolverImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/RowLevelFilterResolverImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/RowLevelFilterResolverImpl.java
index 919faaf..b399fd6 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/RowLevelFilterResolverImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/resolver/RowLevelFilterResolverImpl.java
@@ -39,7 +39,7 @@ public class RowLevelFilterResolverImpl extends ConditionalFilterResolverImpl {
 
   public RowLevelFilterResolverImpl(Expression exp, boolean isExpressionResolve,
       boolean isIncludeFilter, AbsoluteTableIdentifier tableIdentifier) {
-    super(exp, isExpressionResolve, isIncludeFilter, tableIdentifier);
+    super(exp, isExpressionResolve, isIncludeFilter, tableIdentifier, false);
     dimColEvaluatorInfoList =
         new ArrayList<DimColumnResolvedFilterInfo>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
     msrColEvalutorInfoList = new ArrayList<MeasureColumnResolvedFilterInfo>(
@@ -72,6 +72,7 @@ public class RowLevelFilterResolverImpl extends ConditionalFilterResolverImpl {
           msrColumnEvalutorInfo.setRowIndex(index++);
           msrColumnEvalutorInfo
               .setColumnIndex(columnExpression.getCarbonColumn().getOrdinal());
+          msrColumnEvalutorInfo.setMeasure(columnExpression.getMeasure());
           msrColumnEvalutorInfo.setType(columnExpression.getCarbonColumn().getDataType());
           msrColEvalutorInfoList.add(msrColumnEvalutorInfo);
         }


Mime
View raw message