peter-toth commented on code in PR #59324:
URL: https://github.com/apache/spark/pull/59324#discussion_r4246401641
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/LateMaterializationParquetRecordReader.java:
##########
@@ -523,75 +446,61 @@ private void initializeLateMaterialization() throws
IOException {
}
/**
- * Loads the next row group with rows left to emit, through the three phases
of the class javadoc.
- * Phase 2 reads the non-key columns over the survivors while splicing, the
whole projection over
- * the survivors once splicing is given up, and the whole projection over
- * {@code pushedFilterRanges} once the filter is. Returns false when the row
groups run out, which
- * ends a read the filter narrowed.
+ * Loads the next row group with rows left to emit, through the phases of
the class javadoc.
+ * Returns false when the row groups run out.
*/
@Override
protected boolean loadNextRowGroup() throws IOException {
if (keyColumns == null) return super.loadNextRowGroup();
- // The previous row group is fully emitted by the time this is called, so
its pages are done
- // with. Released here rather than at the next assignment, so a row group
the filter empties,
- // or the end of the file, does not keep the previous one's pages alive.
Its column readers go
- // too. They hold the row ranges it was read over, which the budget no
longer counts, and the
- // decoders of their last page.
+ // The previous row group is fully emitted, so its pages and column
readers go now rather than
+ // at the next assignment, which a row group the filter empties would
delay.
closeDataPages();
releaseRowGroupReaders();
StorageFilterMetrics m = storageFilter.metrics();
List<BlockMetaData> blocks = fileReader.getRowGroups();
while (nextBlockIndex < blocks.size()) {
- // A plain read goes back to `FileScanRDD`, which checks for a task
kill, after every batch.
- // This loop can pass any number of row groups the filter empties
without emitting one.
+ // A plain read checks for a task kill after every batch, and this loop
can pass any number
+ // of row groups without emitting one.
ParquetStorageFilter.throwIfKilled();
int blockIdx = nextBlockIndex++;
long blockRowCount = blocks.get(blockIdx).getRowCount();
- // Phase 0: rows the pushed data filter allows, by column index. The
full projection goes back
- // on first, because `getRowRanges` builds an unseen block's store over
the current paths.
+ // Phase 0. The full projection goes back on first, because
`getRowRanges` builds an unseen
+ // block's store over the current paths.
fileReader.setRequestedSchema(requestedColumns);
RowRanges pushedFilterRanges = pushedFilterRangesFor(blockIdx,
blockRowCount);
- // RowRanges.rowCount() walks every range, so resolve each range set's
count once.
long baselineRows = pushedFilterRanges.rowCount();
if (baselineRows == 0) {
- // An empty block, or one the pushed data filter rejects whole. Not
this filter's skip, so
- // no storage-filter metric counts it.
+ // Not this filter's skip, so no storage-filter metric counts it.
continue;
}
- // Whether phase 2 can read part of this row group. That needs an offset
index for every
- // column of the block's column index store, and parquet empties the
store if one column
- // lacks it. With a data filter pushed, the store is built at initialize
over the whole
- // projection, so every projected column is checked. Files written
before parquet-mr 1.11,
- // or by pyarrow's `write_table` by default, have no offset index at
all. With no data filter
- // pushed the check is stricter than needed, since the store would cover
only the non-key
- // columns. False means nothing is buffered and the filter can only skip
the row group whole.
- // Phase 2 then reads the whole projection over `pushedFilterRanges`.
Parquet narrows those
- // only by an index it could read, so that read cannot throw.
+ // Phase 2 can read part of the row group only with an offset index for
every column of the
+ // block's store, since parquet empties the store when one column lacks
it. Files written by
+ // parquet-mr before 1.11, or by pyarrow's `write_table` by default,
have none. Otherwise the
Review Comment:
Thanks, fixed in
[0eb44e78ba1](https://github.com/apache/spark/pull/59324/commits/0eb44e78ba139cfaf181e82d29fda059f381af44):
"Without one, the filter can only skip the row group whole."
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
