peter-toth commented on code in PR #58895:
URL: https://github.com/apache/spark/pull/58895#discussion_r4136319944
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -80,23 +85,23 @@ public class VectorizedParquetRecordReader extends
SpecificParquetRecordReaderBa
* Encapsulate writable column vectors with other Parquet related info such
as
* repetition / definition levels.
*/
- private ParquetColumnVector[] columnVectors;
+ protected ParquetColumnVector[] columnVectors;
/**
* The number of rows that have been returned.
*/
- private long rowsReturned;
+ protected long rowsReturned;
Review Comment:
Done, all four are private again. `noMoreRows()` went back inline too, since
the subclass no longer overrides it. So the base class has nine protected
methods and five protected fields, and the subclass uses every one of them.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -492,6 +945,677 @@ private void checkEndOfRowGroup() throws IOException {
totalCountLoadedSoFar += pages.getRowCount();
}
+ /**
+ * Loads the next row group using the three-phase late-materialization
pattern, all driven by the
+ * single {@link #lateMatReader} with its requested schema mutated per phase:
+ * - Phase 0 (full schema): compute {@code pushedFilterRanges} from the
pushed data filter via
+ * column index (metadata-only) using {@link
ParquetFileReader#getRowRanges}.
+ * - Phase 1 (key-only schema): read key-column pages restricted to {@code
pushedFilterRanges},
+ * evaluate the storage filter per row, build {@code finalRanges}.
+ * - Phase 2: read the non-key columns restricted to {@code finalRanges}.
A row group that gave
+ * splicing up reads the whole projection instead, still under {@code
finalRanges}, and one
+ * that gave the filter up reads it under {@code pushedFilterRanges},
which is what a plain
+ * scan reads. Skipped entirely only for an all-keys projection that is
still splicing, since
+ * emit then builds every batch from the key queues alone.
+ *
+ * Row groups for which {@code finalRanges} is empty are skipped entirely
(no phase-2 IO).
+ * Sets {@link #hitEndOfData} when all row groups have been processed.
+ */
+ private void loadNextRowGroupWithLateMaterialization() throws IOException {
+ // 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 an all-keys
row group, which reads
+ // no data pages at all, does not keep the one before it alive.
+ closeDataPages();
+ while (nextBlockIndex < totalBlockCount) {
+ int blockIdx = nextBlockIndex++;
+ long blockRowCount =
lateMatReader.getRowGroups().get(blockIdx).getRowCount();
+ if (blockRowCount == 0) {
+ // parquet-mr never writes these, but an empty block makes parquet's
own getRowRanges build
+ // Range(0, -1) and trip its `from <= to` assertion. The plain read
path skips them too.
+ continue;
+ }
+ // Splicing buffers one key value per surviving row of the whole row
group before it can emit
+ // the first batch, and that buffer is outside any MemoryConsumer, so
phase 1 counts what it
+ // holds against `maxSplicedRowGroupBytes` together with the row ranges
phase 2 will hold.
+ // Past that it gives splicing up, and past it again the filter itself,
which is what
+ // `filterGivenUp` says.
+ filterGivenUp = false;
+ // Nothing is buffered for a file phase 2 cannot read in part: it will
read the key columns
+ // again along with everything else.
+ spliceCurrentRowGroup = !fileHasNoOffsetIndex;
+
+ // Phase 0: rows allowed by the pushed data filter, at column-index
granularity. The full
+ // requestedSchema goes back on first, because phases 1 and 2 narrow it
and
+ // ParquetFileReader.getRowRanges computes ranges against the reader's
current paths.
+ lateMatReader.setRequestedSchema(requestedColumns);
+ RowRanges pushedFilterRanges = lateMatReader.getRowRanges(blockIdx);
+ // RowRanges.rowCount() walks every range, so resolve each range set's
count once.
+ long baselineRows = pushedFilterRanges.rowCount();
+ if (baselineRows == 0) {
+ // Pushed data filter rejects this block entirely via column index.
Not a storage-filter
+ // skip, so we don't increment storage-filter metrics.
+ continue;
+ }
+
+ // What this feature can avoid reading is the non-key columns of the
rows the storage filter
+ // rejects, so that is the baseline both byte metrics are measured
against: the non-key bytes
+ // a plain read of this projection would transfer for every row the
pushed filter kept. The
+ // null checks only skip work for a caller that drives this reader
without a scan's metrics;
+ // FileSourceScanLike creates all five whenever storageFilters is
non-empty.
+ // compressedBytesForRowRanges never does IO of its own.
+ StorageFilterMetrics m = storageFilter.metrics();
+ SQLMetric bytesAvoidedRg = m.bytesAvoidedByRowGroup();
+ SQLMetric bytesAvoidedPf = m.bytesAvoidedByPageFiltering();
+ boolean needBytes = bytesAvoidedRg != null || bytesAvoidedPf != null;
+ Map<ColumnPath, ColumnChunkMetaData> blockChunks =
+ needBytes ? chunksByPath(lateMatReader, blockIdx) : null;
+ long nonKeyBaselineBytes = needBytes
+ ? compressedBytesForRowRanges(blockIdx, blockChunks, nonKeyColumns,
+ pushedFilterRanges, baselineRows)
+ : 0L;
+
+ // Phase 1: switch to key-only schema, read key columns under
pushedFilterRanges, evaluate the
+ // storage filter per row. The defaults below are what a row group whose
filter is given up
+ // emits, which is every row of `pushedFilterRanges`, exactly what a
plain read would.
+ RowRanges finalRanges = pushedFilterRanges;
+ long finalRowCount = baselineRows;
+ lateMatReader.setRequestedSchema(keyOnlyColumns);
+ // Closed at the end of the phase that reads it. `readFilteredRowGroup`
hands out a store
+ // the file reader does not track, unlike `readNextRowGroup`, so nothing
else would.
+ RowRanges survivors;
+ try (PageReadStore keyPages =
Review Comment:
Added to `spark.sql.parquet.storageFilterPushdown.enabled`. It says the
answer can change under `spark.sql.files.ignoreCorruptFiles`, because a corrupt
page in a key column also drops the rows of its row group that a plain read
returns before reaching that page. The description says the same.
--
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]
