public PageReadStore readNextFilteredRowGroup ( ) throws IOException { if ( currentBlock == blocks . size ( ) ) { return null ; } if ( ! options . useColumnIndexFilter ( ) ) { return readNextRowGroup ( ) ; } BlockMetaData block = blocks . get ( currentBlock ) ; if ( block . getRowCount ( ) == 0 ) { throw new RuntimeException ( "Illegal row group of 0 rows" ) ; } ColumnIndexStore ciStore = getColumnIndexStore ( currentBlock ) ; RowRanges rowRanges = getRowRanges ( currentBlock ) ; long rowCount = rowRanges . rowCount ( ) ; if ( rowCount == 0 ) { // There are no matching rows -> skipping this row-group advanceToNextBlock ( ) ; return readNextFilteredRowGroup ( ) ; } if ( rowCount == block . getRowCount ( ) ) { // All rows are matching -> fall back to the non-filtering path return readNextRowGroup ( ) ; } this . currentRowGroup = new ColumnChunkPageReadStore ( rowRanges ) ; // prepare the list of consecutive parts to read them in one scan ChunkListBuilder builder = new ChunkListBuilder ( ) ; List < ConsecutivePartList > allParts = new ArrayList < ConsecutivePartList > ( ) ; ConsecutivePartList currentParts = null ; for ( ColumnChunkMetaData mc : block . getColumns ( ) ) { ColumnPath pathKey = mc . getPath ( ) ; ColumnDescriptor columnDescriptor = paths . get ( pathKey ) ; if ( columnDescriptor != null ) { OffsetIndex offsetIndex = ciStore . getOffsetIndex ( mc . getPath ( ) ) ; OffsetIndex filteredOffsetIndex = filterOffsetIndex ( offsetIndex , rowRanges , block . getRowCount ( ) ) ; for ( OffsetRange range : calculateOffsetRanges ( filteredOffsetIndex , mc , offsetIndex . getOffset ( 0 ) ) ) { BenchmarkCounter . incrementTotalBytes ( range . getLength ( ) ) ; long startingPos = range . getOffset ( ) ; // first part or not consecutive => new list if ( currentParts == null || currentParts . endPos ( ) != startingPos ) { currentParts = new ConsecutivePartList ( startingPos ) ; allParts . add ( currentParts ) ; } ChunkDescriptor chunkDescriptor = new ChunkDescriptor ( columnDescriptor , mc , startingPos , ( int ) range . getLength ( ) ) ; currentParts . addChunk ( chunkDescriptor ) ; builder . setOffsetIndex ( chunkDescriptor , filteredOffsetIndex ) ; } } } // actually read all the chunks for ( ConsecutivePartList consecutiveChunks : allParts ) { consecutiveChunks . readAll ( f , builder ) ; } for ( Chunk chunk : builder . build ( ) ) { currentRowGroup . addColumn ( chunk . descriptor . col , chunk . readAllPages ( ) ) ; } // avoid re-reading bytes the dictionary reader is used after this call if ( nextDictionaryReader != null ) { nextDictionaryReader . setRowGroup ( currentRowGroup ) ; } advanceToNextBlock ( ) ; return currentRowGroup ; }
Reads all the columns requested from the row group at the current file position . It may skip specific pages based on the column indexes according to the actual filter . As the rows are not aligned among the pages of the different columns row synchronization might be required . See the documentation of the class SynchronizingColumnReader for details .