Skip to content

Commit

Permalink
Merge b08304c into c65cc12
Browse files Browse the repository at this point in the history
  • Loading branch information
@vikramahuja1001
vikramahuja1001 committed Jul 23, 2019
2 parents c65cc12 + b08304c commit ca5db78
Showing 1 changed file with 43 additions and 35 deletions.
78 changes: 43 additions & 35 deletions core/src/main/java/org/apache/carbondata/core/scan/scanner/impl/BlockletFilterScanner.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -96,8 +96,8 @@ public BlockletFilterScanner(BlockExecutionInfo blockExecutionInfo,
* @param rawBlockletColumnChunks block chunk holder which holds the data
* @throws FilterUnsupportedException
*/
@Override
public BlockletScannedResult scanBlocklet(RawBlockletColumnChunks rawBlockletColumnChunks)
@Override public BlockletScannedResult scanBlocklet(
RawBlockletColumnChunks rawBlockletColumnChunks)
throws IOException, FilterUnsupportedException {
if (blockExecutionInfo.isDirectVectorFill()) {
return executeFilterForPages(rawBlockletColumnChunks);
Expand All @@ -106,40 +106,38 @@ public BlockletScannedResult scanBlocklet(RawBlockletColumnChunks rawBlockletCol
}
}

@Override
public boolean isScanRequired(DataRefNode dataBlock) {
@Override public boolean isScanRequired(DataRefNode dataBlock) {
// adding statistics for number of pages
QueryStatistic totalPagesScanned = queryStatisticsModel.getStatisticsTypeAndObjMap()
.get(QueryStatisticsConstants.TOTAL_PAGE_SCANNED);
totalPagesScanned.addCountStatistic(QueryStatisticsConstants.TOTAL_PAGE_SCANNED,
totalPagesScanned.getCount() + dataBlock.numberOfPages());
// apply min max
if (isMinMaxEnabled) {
if (null == dataBlock.getColumnsMaxValue()
|| null == dataBlock.getColumnsMinValue()) {
if (null == dataBlock.getColumnsMaxValue() || null == dataBlock.getColumnsMinValue()) {
return true;
}
BitSet bitSet = null;
// check for implicit include filter instance
if (filterExecuter instanceof ImplicitColumnFilterExecutor) {
String blockletId = blockExecutionInfo.getBlockIdString() +
CarbonCommonConstants.FILE_SEPARATOR + dataBlock.blockletIndex();
String blockletId =
blockExecutionInfo.getBlockIdString() + CarbonCommonConstants.FILE_SEPARATOR + dataBlock
.blockletIndex();
bitSet = ((ImplicitColumnFilterExecutor) filterExecuter)
.isFilterValuesPresentInBlockOrBlocklet(
dataBlock.getColumnsMaxValue(),
.isFilterValuesPresentInBlockOrBlocklet(dataBlock.getColumnsMaxValue(),
dataBlock.getColumnsMinValue(), blockletId, dataBlock.minMaxFlagArray());
} else {
bitSet = this.filterExecuter
.isScanRequired(dataBlock.getColumnsMaxValue(),
dataBlock.getColumnsMinValue(), dataBlock.minMaxFlagArray());
.isScanRequired(dataBlock.getColumnsMaxValue(), dataBlock.getColumnsMinValue(),
dataBlock.minMaxFlagArray());
}
return !bitSet.isEmpty();
}
return true;
}

@Override
public void readBlocklet(RawBlockletColumnChunks rawBlockletColumnChunks) throws IOException {
@Override public void readBlocklet(RawBlockletColumnChunks rawBlockletColumnChunks)
throws IOException {
long startTime = System.currentTimeMillis();
this.filterExecuter.readColumnChunks(rawBlockletColumnChunks);
// adding statistics for carbon read time
Expand Down Expand Up @@ -176,8 +174,8 @@ private BlockletScannedResult executeFilter(RawBlockletColumnChunks rawBlockletC
BitSetGroup fgBitSetGroup = rawBlockletColumnChunks.getDataBlock().getIndexedData();
rawBlockletColumnChunks.setBitSetGroup(fgBitSetGroup);
// apply filter on actual data, for each page
BitSetGroup bitSetGroup = this.filterExecuter.applyFilter(rawBlockletColumnChunks,
useBitSetPipeLine);
BitSetGroup bitSetGroup =
this.filterExecuter.applyFilter(rawBlockletColumnChunks, useBitSetPipeLine);
// if filter result is empty then return with empty result
if (bitSetGroup.isEmpty()) {
CarbonUtil.freeMemory(rawBlockletColumnChunks.getDimensionRawColumnChunks(),
Expand All @@ -198,8 +196,8 @@ private BlockletScannedResult executeFilter(RawBlockletColumnChunks rawBlockletC
BlockletScannedResult scannedResult =
new FilterQueryScannedResult(blockExecutionInfo, queryStatisticsModel);
scannedResult.setBlockletId(
blockExecutionInfo.getBlockIdString() + CarbonCommonConstants.FILE_SEPARATOR +
rawBlockletColumnChunks.getDataBlock().blockletIndex());
blockExecutionInfo.getBlockIdString() + CarbonCommonConstants.FILE_SEPARATOR
+ rawBlockletColumnChunks.getDataBlock().blockletIndex());
// valid scanned blocklet
QueryStatistic validScannedBlockletStatistic = queryStatisticsModel.getStatisticsTypeAndObjMap()
.get(QueryStatisticsConstants.VALID_SCAN_BLOCKLET_NUM);
Expand Down Expand Up @@ -232,12 +230,8 @@ private BlockletScannedResult executeFilter(RawBlockletColumnChunks rawBlockletC
}
}

long dimensionReadTime = System.currentTimeMillis();
dimensionReadTime = System.currentTimeMillis() - dimensionReadTime;

FileReader fileReader = rawBlockletColumnChunks.getFileReader();


DimensionRawColumnChunk[] dimensionRawColumnChunks =
new DimensionRawColumnChunk[blockExecutionInfo.getTotalNumberDimensionToRead()];
int numDimensionChunks = dimensionRawColumnChunks.length;
Expand All @@ -246,28 +240,34 @@ private BlockletScannedResult executeFilter(RawBlockletColumnChunks rawBlockletC
dimensionRawColumnChunks[chunkIndex] =
rawBlockletColumnChunks.getDimensionRawColumnChunks()[chunkIndex];
}
//dimensionReadTime is the time required to read the data from dimension array
long totalReadTime = System.currentTimeMillis();
int[][] allSelectedDimensionColumnIndexRange =
blockExecutionInfo.getAllSelectedDimensionColumnIndexRange();
DimensionRawColumnChunk[] projectionListDimensionChunk = rawBlockletColumnChunks.getDataBlock()
.readDimensionChunks(fileReader, allSelectedDimensionColumnIndexRange);
totalReadTime = System.currentTimeMillis() - totalReadTime;

for (int[] columnIndexRange : allSelectedDimensionColumnIndexRange) {
System.arraycopy(projectionListDimensionChunk, columnIndexRange[0],
dimensionRawColumnChunks, columnIndexRange[0],
columnIndexRange[1] + 1 - columnIndexRange[0]);
System.arraycopy(projectionListDimensionChunk, columnIndexRange[0], dimensionRawColumnChunks,
columnIndexRange[0], columnIndexRange[1] + 1 - columnIndexRange[0]);
}

/*
* in case projection if the projected dimension are not loaded in the dimensionColumnDataChunk
* then loading them
/**
* Below code is to read the dimension which is not read as part of filter or projection
* for example in case of or filter if first filter matches all the rows then it will not read
* second filter column and if it is present as part of projection, so needs to be read
*/
long filterDimensionReadTime = System.currentTimeMillis();
int[] projectionListDimensionIndexes = blockExecutionInfo.getProjectionListDimensionIndexes();
for (int projectionListDimensionIndex : projectionListDimensionIndexes) {
if (null == dimensionRawColumnChunks[projectionListDimensionIndex]) {
dimensionRawColumnChunks[projectionListDimensionIndex] =
rawBlockletColumnChunks.getDataBlock().readDimensionChunk(
fileReader, projectionListDimensionIndex);
rawBlockletColumnChunks.getDataBlock()
.readDimensionChunk(fileReader, projectionListDimensionIndex);
}
}
totalReadTime += System.currentTimeMillis() - filterDimensionReadTime;

DimensionColumnPage[][] dimensionColumnPages =
new DimensionColumnPage[numDimensionChunks][numPages];
Expand All @@ -283,25 +283,33 @@ private BlockletScannedResult executeFilter(RawBlockletColumnChunks rawBlockletC
}
}

long measureReadTime = System.currentTimeMillis();
int[][] allSelectedMeasureColumnIndexRange =
blockExecutionInfo.getAllSelectedMeasureIndexRange();
MeasureRawColumnChunk[] projectionListMeasureChunk = rawBlockletColumnChunks.getDataBlock()
.readMeasureChunks(fileReader, allSelectedMeasureColumnIndexRange);
measureReadTime = System.currentTimeMillis() - measureReadTime;

for (int[] columnIndexRange : allSelectedMeasureColumnIndexRange) {
System.arraycopy(projectionListMeasureChunk, columnIndexRange[0], measureRawColumnChunks,
columnIndexRange[0], columnIndexRange[1] + 1 - columnIndexRange[0]);
}
/*
* in case projection if the projected measure are not loaded in the ColumnPage
* then loading them
/**
* Below code is to read the measure which is not read as part of filter or projection
* for example in case of or filter if first filter matches all the rows then it will not read
* second filter column and if it is present as part of projection, so needs to be read
*/
long filterMeasureReadTime = System.currentTimeMillis();
int[] projectionListMeasureIndexes = blockExecutionInfo.getProjectionListMeasureIndexes();
for (int projectionListMeasureIndex : projectionListMeasureIndexes) {
if (null == measureRawColumnChunks[projectionListMeasureIndex]) {
measureRawColumnChunks[projectionListMeasureIndex] = rawBlockletColumnChunks.getDataBlock()
.readMeasureChunk(fileReader, projectionListMeasureIndex);
}
}
measureReadTime += System.currentTimeMillis() - filterMeasureReadTime;
totalReadTime += measureReadTime;

ColumnPage[][] measureColumnPages = new ColumnPage[numMeasureChunks][numPages];
scannedResult.setDimensionColumnPages(dimensionColumnPages);
scannedResult.setPageFilteredRowId(pageFilteredRowId);
Expand All @@ -314,11 +322,11 @@ private BlockletScannedResult executeFilter(RawBlockletColumnChunks rawBlockletC
QueryStatistic scanTime = queryStatisticsModel.getStatisticsTypeAndObjMap()
.get(QueryStatisticsConstants.SCAN_BLOCKlET_TIME);
scanTime.addCountStatistic(QueryStatisticsConstants.SCAN_BLOCKlET_TIME,
scanTime.getCount() + (System.currentTimeMillis() - startTime - dimensionReadTime));
scanTime.getCount() + (System.currentTimeMillis() - startTime - totalReadTime));
QueryStatistic readTime = queryStatisticsModel.getStatisticsTypeAndObjMap()
.get(QueryStatisticsConstants.READ_BLOCKlET_TIME);
readTime.addCountStatistic(QueryStatisticsConstants.READ_BLOCKlET_TIME,
readTime.getCount() + dimensionReadTime);
readTime.getCount() + totalReadTime);
return scannedResult;
}

Expand Down

0 comments on commit ca5db78

Please sign in to comment.