peter-toth commented on code in PR #58895:
URL: https://github.com/apache/spark/pull/58895#discussion_r4133526563
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -409,23 +631,250 @@ public boolean nextBatch() throws IOException {
}
cv.assemble();
}
- // If needed, compute row indexes within a file.
- if (rowIndexGenerator != null) {
- rowIndexGenerator.populateRowIndex(columnVectors, num);
- }
+ }
- rowsReturned += num;
+ /** Publishes {@code num} rows as the current batch. */
+ private void finishBatch(int num) {
columnarBatch.setNumRows(num);
+ rowsReturned += num;
numBatched = num;
batchIdx = 0;
- return true;
+ }
+
+ /**
+ * Points the batch's key slots at one survivor key vector each, which is
what publishes them: the
+ * same {@link ColumnarBatch} is handed out every time, over an array it
holds by reference. The
+ * queues keep owning those vectors until the next batch releases them, so a
phase-2 read that
+ * throws afterwards leaves them reachable for {@link #close()}.
+ */
+ private void publishSurvivorKeyVectors(int num) {
+ for (int i = 0; i < keyVectorQueues.length; i++) {
+ if (keyVectorQueues[i].isEmpty()) {
+ // Unreachable: the queues hold exactly the survivors phase 1
accumulated, and the emit loop
+ // is driven by that same count. Named rather than left to
NoSuchElementException.
+ throw new IllegalStateException(String.format(
+ "Storage-filter survivor queue %d of row group %d in %s ran out
with %d rows still to "
+ + "emit", i, nextBlockIndex - 1, lateMatReader.getFile(),
num));
+ }
+ }
+ keyVectorsPublished = true;
+ // Queue k holds the survivors of key column k, and `keyColumnIndices[k]`
is the batch slot that
+ // key column sits in, so the pairing is read off rather than re-derived
from slot order. Only
+ // key slots are touched, since a non-key slot never holds anything but
its persistent vector.
+ int[] keyIndices = storageFilter.keyColumnIndices();
+ for (int k = 0; k < keyIndices.length; k++) {
+ spliceBatchColumns[keyIndices[k]] = keyVectorQueues[k].peekFirst();
+ }
+ }
+
+ /**
+ * Closes the key vectors the previous batch was built on. The queues own
them until here, so
+ * survivor memory drains as a row group is emitted rather than all at its
end.
+ */
+ private void releasePublishedKeyVectors() {
+ if (!keyVectorsPublished) return;
+ keyVectorsPublished = false;
+ for (java.util.ArrayDeque<WritableColumnVector> queue : keyVectorQueues) {
+ queue.removeFirst().close();
+ }
}
private void initializeInternal() throws IOException,
UnsupportedOperationException {
missingColumns = new HashSet<>();
for (ParquetColumn column :
CollectionConverters.asJava(parquetColumn.children())) {
checkColumn(column);
}
+ if (storageFilter != null) {
+ initializeLateMaterialization();
+ }
+ }
+
+ /**
+ * Sets the storage filter for late materialization. Must be called before
{@link #initialize};
+ * {@link #initializeLateMaterialization()} then inspects the per-file
schema and decides whether
+ * splicing engages.
+ *
+ * <p>It does not engage in one real case: every key column is missing from
this physical file
+ * under schema evolution. The predicate is then rewritten with each missing
key replaced by the
+ * constant the reader materializes for it (its existence DEFAULT, else
null) and evaluated once.
+ * True keeps the file unfiltered, false or null skips it.
+ *
+ * <p>Everything else the filter needs is guaranteed by
+ * {@code FileSourceStrategy.storageFiltersFor} and {@code
ParquetStorageFilter.create}, so a
+ * violation of it here is a planner bug and is asserted rather than
handled. A file the reader
+ * simply cannot prune is a different matter: it reads it the way a plain
scan would.
+ */
+ public void setStorageFilter(ParquetStorageFilter storageFilter) {
+ this.storageFilter = storageFilter;
+ }
+
+ private void initializeLateMaterialization() throws IOException {
+ if
(!configuration.getBoolean(ParquetInputFormat.COLUMN_INDEX_FILTERING_ENABLED,
true)) {
Review Comment:
Fixed the way you describe. The conf no longer turns the feature off. It
sets the same flag a missing index sets, and `pushedFilterRangesFor` returns
`RowRanges.createSingle(blockRowCount)` for phase 0 and for the whole-file skip
path, which is where `getRowRanges` ignoring `useColumnIndexFilter()` would
otherwise have bitten. So the filter keeps the row groups it empties, which
need no index at all. A test runs both arms and asserts the metrics, so it
cannot pass by the feature being off in both.
One thing came out of that flag afterwards. It carried two meanings, the
conf and a partial read that had failed, and for the second one dropping
`getRowRanges` for the rest of the file dropped the pushed data filter's own
page pruning too, which a plain read keeps. Measured at 215 rows against a
plain read's 160, on a three-row-group file with one damaged index. There are
two flags now, and the latch takes the narrowing alone.
##########
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 =
+ lateMatReader.readFilteredRowGroup(blockIdx,
pushedFilterRanges)) {
+ if (keyPages == null) {
+ // Unreachable: readFilteredRowGroup returns null only for an empty
block, and we know
+ // pushedFilterRanges selects at least one row. Skipping the block
here would drop its
+ // surviving rows from the output, so assert rather than `continue`.
+ throw new IllegalStateException(
+ "No key pages for row group " + blockIdx + " despite " +
baselineRows
+ + " rows selected by the pushed filter");
+ }
+ survivors = evaluateStorageFilter(keyPages, pushedFilterRanges);
+ }
+ if (survivors != null) {
+ finalRanges = survivors;
+ finalRowCount = survivors.rowCount();
+ if (finalRowCount == 0) {
+ // Every surviving row was rejected by the storage filter; skip the
block entirely,
+ // which avoids the whole non-key baseline. Phase 1 still paid to
read the key columns,
+ // and that cost is not part of the baseline, so nothing is
subtracted from it here.
+ recordRowGroupSkipped(m, baselineRows, nonKeyBaselineBytes);
+ continue;
+ }
+ // Reading part of a row group needs the offset index, so on a file
without one phase 2
+ // reads the whole row group instead. The rows the filter rejected are
then emitted and
+ // the post-scan Filter drops them. What the filter still saves on
such a file is the row
+ // groups it empties, which is decided above and needs no index at all.
+ if (fileHasNoOffsetIndex && finalRowCount < baselineRows) {
Review Comment:
Both done. There is a footer check per row group now, and phase 1 stops at
the first survivor when the row group cannot be narrowed, since whether
anything survives at all is the only thing left to learn there. Parquet's
exception stays as the fallback for an index the footer claims and parquet
cannot then produce.
One correction to the check as you describe it. Asking about the phase-2
columns alone is not enough, because parquet builds a block's
`ColumnIndexStore` over the requested schema at the first lookup, and with a
data filter pushed that is phase 0's, over the whole projection. So a key
column with no offset index empties the store as well. The check asks about the
whole projection for that reason, and a test on a hand-built file whose key
column is the only unindexed chunk fails if it asks about the non-key columns
alone.
##########
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 =
+ lateMatReader.readFilteredRowGroup(blockIdx,
pushedFilterRanges)) {
+ if (keyPages == null) {
+ // Unreachable: readFilteredRowGroup returns null only for an empty
block, and we know
+ // pushedFilterRanges selects at least one row. Skipping the block
here would drop its
+ // surviving rows from the output, so assert rather than `continue`.
+ throw new IllegalStateException(
+ "No key pages for row group " + blockIdx + " despite " +
baselineRows
+ + " rows selected by the pushed filter");
+ }
+ survivors = evaluateStorageFilter(keyPages, pushedFilterRanges);
+ }
+ if (survivors != null) {
+ finalRanges = survivors;
+ finalRowCount = survivors.rowCount();
+ if (finalRowCount == 0) {
+ // Every surviving row was rejected by the storage filter; skip the
block entirely,
+ // which avoids the whole non-key baseline. Phase 1 still paid to
read the key columns,
+ // and that cost is not part of the baseline, so nothing is
subtracted from it here.
+ recordRowGroupSkipped(m, baselineRows, nonKeyBaselineBytes);
+ continue;
+ }
+ // Reading part of a row group needs the offset index, so on a file
without one phase 2
+ // reads the whole row group instead. The rows the filter rejected are
then emitted and
+ // the post-scan Filter drops them. What the filter still saves on
such a file is the row
+ // groups it empties, which is decided above and needs no index at all.
+ if (fileHasNoOffsetIndex && finalRowCount < baselineRows) {
+ giveUpFilter();
+ finalRanges = pushedFilterRanges;
+ finalRowCount = baselineRows;
+ }
+ }
+
+ // Phase 2 reads the non-key columns under the surviving rows, or the
whole projection under
+ // `pushedFilterRanges` for a row group whose filter was given up. It is
skipped only when the
+ // projection is all keys and their values were buffered, since emit
then builds every batch
+ // from the key queues alone.
+ long keptRows;
+ if (nonKeyColumns == null && spliceCurrentRowGroup) {
+ keptRows = finalRowCount;
+ } else {
+ lateMatReader.setRequestedSchema(
+ spliceCurrentRowGroup ? nonKeyColumns : requestedColumns);
+ // Reading a strict subset of a block's rows needs a Parquet offset
index, and parquet
+ // enforces that itself: it resolves every requested column's offset
index before reading
+ // anything, and a column without one makes its column index store
throw
+ // MissingOffsetIndexException. Files written before parquet-mr 1.11,
or by a writer that
+ // omits the page index (pyarrow's `write_table` defaults to
`write_page_index=False`), have
+ // none. The filter is then given up for this row group and the read
retried over
+ // `pushedFilterRanges`, which is what a plain scan reads. That retry
cannot hit the same
+ // wall: a store missing one column's offset index reports no column
index either, so
+ // `getRowRanges` could not have narrowed anything and the ranges
cover the whole block.
+ //
+ // Asked this way rather than up front, from the footer. Parquet
resolves the index over the
+ // paths current at its first lookup for the block, which without a
pushed data filter is
+ // the non-key columns alone, so a footer walk over the projection is
both stricter than the
+ // read and blind to an index that is claimed but unreadable. It is
also where the throw
+ // costs least: it lands before any data page is read.
+ try {
+ dataPages = lateMatReader.readFilteredRowGroup(blockIdx,
finalRanges);
+ } catch (MissingOffsetIndexException e) {
+ LOG.warn("Reading {} without page-level storage filtering: reading
part of a row group "
+ + "needs a Parquet offset index, and this file has none for at
least one column the "
+ + "read needs. Row groups the filter empties are still skipped
whole", e,
+ MDC.of(LogKeys.PATH, lateMatReader.getFile()));
+ fileHasNoOffsetIndex = true;
+ giveUpFilter();
+ finalRanges = pushedFilterRanges;
+ finalRowCount = baselineRows;
+ // The retry only avoids the same wall because a store missing one
column's offset index
+ // reports no column index either, so these ranges cover the whole
block and parquet reads
+ // it without consulting an index. That is three parquet internals
deep, so it is checked:
+ // a release that changes any of them should fail here rather than
throw from the read.
+ if (baselineRows != blockRowCount) {
+ throw new IllegalStateException(String.format(
+ "Cannot read row group %d of %s without an offset index: the
pushed filter selects "
+ + "%d of %d rows, so a plain read of the block is not what
it asks for",
+ blockIdx, lateMatReader.getFile(), baselineRows,
blockRowCount));
+ }
+ lateMatReader.setRequestedSchema(requestedColumns);
+ dataPages = lateMatReader.readFilteredRowGroup(blockIdx,
finalRanges);
+ }
+ if (dataPages == null) {
+ // Unreachable: readFilteredRowGroup returns null only for an empty
block or empty ranges,
+ // both excluded above. Match phase 1 and fail with a message rather
than an NPE.
+ throw new IllegalStateException(
+ "No data pages for row group " + blockIdx + " despite " +
finalRowCount
+ + " rows to read");
+ }
+ keptRows = dataPages.getRowCount();
+ // Nothing is reported for a row group whose filter was given up: it
read what a plain scan
+ // reads, so the saving is a certain zero.
+ if (bytesAvoidedPf != null && !filterGivenUp) {
+ long phase2Bytes = compressedBytesForRowRanges(blockIdx, blockChunks,
+ nonKeyColumns, finalRanges, finalRowCount);
+ if (!spliceCurrentRowGroup) {
+ // This row group gave splicing up, so phase 2 read the key
columns a second time. The
+ // baseline counts them once, in phase 1, so the extra read is a
cost against it.
+ phase2Bytes += compressedBytesForRowRanges(blockIdx, blockChunks,
+ keyOnlyColumns, finalRanges, finalRowCount);
+ }
+ // `SQLMetric.add` ignores a negative value, so a row group that
read more than the
+ // baseline after giving splicing up contributes nothing rather than
subtracting.
+ bytesAvoidedPf.add(nonKeyBaselineBytes - phase2Bytes);
+ }
+ }
+ long filteredRows = baselineRows - keptRows;
+ SQLMetric rowsExcludedWithinRg = m.rowsExcludedWithinRowGroup();
+ if (rowsExcludedWithinRg != null && filteredRows > 0)
rowsExcludedWithinRg.add(filteredRows);
+
+ if (dataPages != null) {
+ if (rowIndexGenerator != null) {
+ rowIndexGenerator.initFromPageReadStore(dataPages);
+ }
+ for (int i = 0; i < columnVectors.length; i++) {
+ if (spliceCurrentRowGroup && isKeyTopLevel[i]) {
+ // Key columns are sourced from the queues during emit; skip
phase-2 reader init.
+ continue;
+ }
+ initColumnReader(dataPages, columnVectors[i]);
+ }
+ }
+ totalCountLoadedSoFar += keptRows;
+ return;
+ }
+ hitEndOfData = true;
+ }
+
+
+ /** Counts a row group whose data columns the filter kept the reader from
touching at all. */
+ private static void recordRowGroupSkipped(
+ StorageFilterMetrics m, long excludedRows, long avoidedBytes) {
+ SQLMetric rgSkipped = m.rowGroupsSkipped();
+ if (rgSkipped != null) rgSkipped.add(1L);
+ SQLMetric rowsExcluded = m.rowsExcludedByRowGroup();
+ if (rowsExcluded != null) rowsExcluded.add(excludedRows);
+ SQLMetric bytesAvoided = m.bytesAvoidedByRowGroup();
+ if (bytesAvoided != null) bytesAvoided.add(avoidedBytes);
+ }
+
+ /**
+ * Counts a file the filter rejects whole, which happens when every key
column is missing from it
+ * and the predicate is constant-false for the value the reader would have
materialized. Every row
+ * group counts as skipped and every projected byte as avoided, which is
what the counters mean
+ * for a row group the filter empties.
+ */
+ private void recordFileSkipped() {
+ StorageFilterMetrics m = storageFilter.metrics();
+ boolean needBytes = m.bytesAvoidedByRowGroup() != null;
+ if (m.rowGroupsSkipped() == null && m.rowsExcludedByRowGroup() == null &&
!needBytes) return;
+ List<ColumnDescriptor> projected = requestedSchema.getColumns();
+ List<BlockMetaData> blocks = lateMatReader.getRowGroups();
+ for (int blockIdx = 0; blockIdx < blocks.size(); blockIdx++) {
+ // Measured against the rows the pushed data filter kept, which is the
baseline every other
+ // skip path uses: the rows its column index already excluded were never
this filter's to
+ // save. Resolving them again is a cache hit whenever the two can
differ, because
+ // `getFilteredRecordCount()` at initialize resolved every block's
ranges then.
+ long blockRowCount = blocks.get(blockIdx).getRowCount();
+ if (blockRowCount == 0) continue;
+ RowRanges blockRanges = lateMatReader.getRowRanges(blockIdx);
+ long survivingRows = blockRanges.rowCount();
+ if (survivingRows == 0) continue;
+ // The key columns are missing from this file, so they contribute
nothing to the walk, and the
+ // whole projection is what a plain read would have transferred.
+ long avoidedBytes = needBytes
+ ? compressedBytesForRowRanges(blockIdx,
+ chunksByPath(lateMatReader, blockIdx), projected, blockRanges,
survivingRows)
+ : 0L;
+ recordRowGroupSkipped(m, survivingRows, avoidedBytes);
+ }
+ }
+
+ /**
+ * The block's column chunks by path, built once per row group and shared by
the byte-metric calls
+ * that consume it, since {@link BlockMetaData} offers no lookup of its own.
+ */
+ private static Map<ColumnPath, ColumnChunkMetaData> chunksByPath(
+ ParquetFileReader reader, int blockIndex) {
+ Map<ColumnPath, ColumnChunkMetaData> chunks = new HashMap<>();
+ for (ColumnChunkMetaData chunk :
reader.getRowGroups().get(blockIndex).getColumns()) {
+ chunks.put(chunk.getPath(), chunk);
+ }
+ return chunks;
+ }
+
+ /**
+ * Compressed bytes the reader transfers for the given leaf {@code columns}
when it reads exactly
+ * {@code rowRanges} of the given block. Page headers and the dictionary
page are included, since
+ * both are read whenever any page of a chunk is read. {@code rowRangeCount}
is
+ * {@code rowRanges.rowCount()}, passed in because that walks every range
and the caller has it.
+ *
+ * <p>Two sources, chosen so this never causes IO of its own:
+ * <ul>
+ * <li>{@code rowRanges} covers the whole block: the answer is the sum of
the chunks'
+ * {@code getTotalSize()}, which is already in the footer. This is the
case that matters:
+ * whenever nothing else has built the block's {@link
ColumnIndexStore}, {@code rowRanges}
+ * is necessarily the whole block, because a narrower range can only
come from column-index
+ * filtering, which builds the store as a side effect.
+ * <li>{@code rowRanges} is a strict subset: walk the offset index, as
parquet's own read path
+ * does, and add the dictionary page the way {@code
calculateOffsetRanges} does. The store
+ * is guaranteed to exist here, so the walk is pure metadata
arithmetic. For the ranges the
+ * storage filter narrowed, which column-index filtering had no hand
in, that guarantee is
+ * an ordering one: phase 2's own read of those ranges built the store
first.
+ * </ul>
+ *
+ * <p>Columns absent from this physical file (schema evolution) contribute
nothing, which is
+ * correct: the reader transfers nothing for them.
+ */
+ private long compressedBytesForRowRanges(
+ int blockIndex,
+ Map<ColumnPath, ColumnChunkMetaData> chunks,
+ List<ColumnDescriptor> columns,
+ RowRanges rowRanges,
+ long rowRangeCount) {
+ if (columns == null || columns.isEmpty() || rowRangeCount == 0) {
+ return 0L;
+ }
+ long blockRowCount =
lateMatReader.getRowGroups().get(blockIndex).getRowCount();
+ boolean wholeBlock = rowRangeCount == blockRowCount;
+ ColumnIndexStore ciStore =
+ wholeBlock ? null : lateMatReader.getColumnIndexStore(blockIndex);
+ long total = 0L;
+ for (ColumnDescriptor column : columns) {
+ ColumnPath path = ColumnPath.get(column.getPath());
+ ColumnChunkMetaData chunk = chunks.get(path);
+ if (chunk == null) {
+ // Column is in the (clipped) requested schema but not in this file.
+ continue;
+ }
+ if (wholeBlock) {
+ total += chunk.getTotalSize();
+ continue;
+ }
+ OffsetIndex offsetIndex;
+ try {
+ offsetIndex = ciStore.getOffsetIndex(path);
+ } catch (MissingOffsetIndexException e) {
+ continue;
+ }
+ if (offsetIndex == null) {
+ // The store answers null, rather than throwing, for a path it was not
built with. That can
+ // only mean the block's store was built while a narrower schema was
requested than this
+ // walk asks about, an ordering bug rather than a property of the
file, and the footer tells
+ // the two apart. It is reported rather than thrown: this walk only
produces a counter, and
+ // `ignoreCorruptFiles` turns any exception from a reader into a
silently truncated file, so
+ // a byte metric must not be able to change the answer.
+ if (chunk.getOffsetIndexReference() != null &&
!loggedMissingStoreEntry) {
+ loggedMissingStoreEntry = true;
+ LOG.warn("Undercounting the storage filter's avoided bytes for {}:
column "
+ + path.toDotString() + " of row group " + blockIndex + " has an
offset index the "
+ + "block's column index store was not built with",
+ MDC.of(LogKeys.PATH, lateMatReader.getFile()));
+ }
+ continue;
+ }
+ // The dictionary page is read whenever any data page of the chunk is,
so count it here the
+ // same way parquet's ColumnIndexFilterUtils.calculateOffsetRanges does.
+ total += dictionaryPageSize(chunk);
+ int pageCount = offsetIndex.getPageCount();
+ for (int i = 0; i < pageCount; i++) {
+ long from = offsetIndex.getFirstRowIndex(i);
+ long to = offsetIndex.getLastRowIndex(i, blockRowCount);
+ if (rowRanges.isOverlapping(from, to)) {
+ total += offsetIndex.getCompressedPageSize(i);
+ }
+ }
+ }
+ return total;
+ }
+
+ /**
+ * Compressed size of a chunk's dictionary page, or 0 if it has none.
+ * {@link ColumnChunkMetaData#getStartingPos()} already resolves to the
dictionary page offset
+ * when there is a valid one, so the gap up to the first data page is
exactly the dictionary page.
+ */
+ private static long dictionaryPageSize(ColumnChunkMetaData chunk) {
+ long startingPos = chunk.getStartingPos();
+ long firstDataPageOffset = chunk.getFirstDataPageOffset();
+ return startingPos < firstDataPageOffset ? firstDataPageOffset -
startingPos : 0L;
+ }
+
+ /**
+ * Evaluates the storage filter over every row of a key-only {@link
PageReadStore}, in
+ * capacity-sized chunks, and returns the surviving rows as {@link
RowRanges} in block-row
+ * coordinates. The result is a subset of {@code pushedFilterRanges}: rows
outside it were never
+ * read.
+ *
+ * <p>Each survivor's key values are appended to {@link
#currentKeyAccumulators} for the emit path
+ * to splice, until the buffer passes its cap. From there the row group is
evaluated without
+ * buffering and {@link #spliceCurrentRowGroup} is false, so its phase 2
reads the key columns
+ * again along with everything else.
+ *
+ * <p>Returns null once the budget makes the reader give the filter up for
this row group: the
+ * ranges built so far are then incomplete, and the caller reads the row
group the plain way.
+ */
+ private RowRanges evaluateStorageFilter(
+ PageReadStore keyPages,
+ RowRanges pushedFilterRanges) throws IOException {
+ ensureKeyScratchAllocated();
+ VectorizedColumnReader[] readers = new
VectorizedColumnReader[keyDescriptors.length];
+ for (int i = 0; i < readers.length; i++) {
+ readers[i] = new VectorizedColumnReader(
+ keyDescriptors[i], keyRequired[i], keyPages, convertTz,
datetimeRebaseMode,
+ datetimeRebaseTz, int96RebaseMode, int96RebaseTz, writerVersion);
+ }
+ if (spliceCurrentRowGroup) ensureCurrentKeyAccumulatorsAllocated();
+
+ PrimitiveIterator.OfLong rowIndexIter = pushedFilterRanges.iterator();
+ RowRanges.Builder finalRangesBuilder = RowRanges.builder();
+ // What this row group retains, weighed against the budget below: the
bytes buffered for
+ // splicing, and the ranges the surviving rows fall into.
+ long splicedBytes = 0L;
+ long survivorRangeCount = 0L;
+ long previousSurvivor = -2L;
+ // Recomputed rather than taken from the caller: a count that disagreed
with this iterator would
+ // silently drop surviving rows, and no post-scan Filter is left to catch
that.
+ long remaining = pushedFilterRanges.rowCount();
+ long cap = storageFilter.maxSplicedRowGroupBytes();
+ while (remaining > 0) {
+ int num = (int) Math.min((long) capacity, remaining);
+ for (int i = 0; i < keyScratchVectors.length; i++) {
+ keyScratchVectors[i].reset();
+ readers[i].readBatch(num, keyScratchVectors[i], null, null);
+ }
+ keyScratchBatch.setNumRows(num);
+ for (int r = 0; r < num; r++) {
+ long blockRow = rowIndexIter.nextLong();
+ boolean survives;
+ try {
+ survives = storageFilter.test(keyScratchBatch.getRow(r));
+ } catch (RuntimeException e) {
+ if (!storageFilter.isEvaluationError(e)) throw e;
+ // Fail open. The predicate ran on a row that, in the plan, an
earlier conjunct would
+ // have rejected before it, so a plain scan never evaluates it
there. Giving the filter up
+ // for this row group puts every row the pushed filter kept back in
the output, and the
+ // post-scan Filter then evaluates the conjuncts in their own order.
Either an earlier one
+ // drops the row before this expression runs, or it does not and the
query fails the way
+ // it would have without this feature.
+ logFilterGivenUpOnError(e);
+ giveUpFilter();
+ return null;
+ }
+ if (survives) {
+ finalRangesBuilder.addSelectedRow(blockRow);
Review Comment:
Measured, and at the extreme it is worse than the guess. On a synthetic
table of one long key and twenty long payload columns written in 256-row pages,
with one survivor every other row, phase 2 skips no page at all and the read
takes 5.6x the plain one. It falls to 1.8x at runs of sixteen rows. So the
shape you name is real, and it is the one a gate needs most. The numbers are in
the description.
I did not add the gate, because it cannot be as simple as a ranges-per-row
ratio. Giving up also re-reads the key columns and hands the post-scan `Filter`
every row the filter would have dropped, and those costs grow with what the
filter excluded rather than with what it kept, so the direction flips as the
filter gets more selective. That wants a cost model, which is the second
follow-up in the description and one of the reasons the conf is off by default.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -409,23 +631,250 @@ public boolean nextBatch() throws IOException {
}
cv.assemble();
}
- // If needed, compute row indexes within a file.
- if (rowIndexGenerator != null) {
- rowIndexGenerator.populateRowIndex(columnVectors, num);
- }
+ }
- rowsReturned += num;
+ /** Publishes {@code num} rows as the current batch. */
+ private void finishBatch(int num) {
columnarBatch.setNumRows(num);
+ rowsReturned += num;
numBatched = num;
batchIdx = 0;
- return true;
+ }
+
+ /**
+ * Points the batch's key slots at one survivor key vector each, which is
what publishes them: the
+ * same {@link ColumnarBatch} is handed out every time, over an array it
holds by reference. The
+ * queues keep owning those vectors until the next batch releases them, so a
phase-2 read that
+ * throws afterwards leaves them reachable for {@link #close()}.
+ */
+ private void publishSurvivorKeyVectors(int num) {
+ for (int i = 0; i < keyVectorQueues.length; i++) {
+ if (keyVectorQueues[i].isEmpty()) {
+ // Unreachable: the queues hold exactly the survivors phase 1
accumulated, and the emit loop
+ // is driven by that same count. Named rather than left to
NoSuchElementException.
+ throw new IllegalStateException(String.format(
+ "Storage-filter survivor queue %d of row group %d in %s ran out
with %d rows still to "
+ + "emit", i, nextBlockIndex - 1, lateMatReader.getFile(),
num));
+ }
+ }
+ keyVectorsPublished = true;
+ // Queue k holds the survivors of key column k, and `keyColumnIndices[k]`
is the batch slot that
+ // key column sits in, so the pairing is read off rather than re-derived
from slot order. Only
+ // key slots are touched, since a non-key slot never holds anything but
its persistent vector.
+ int[] keyIndices = storageFilter.keyColumnIndices();
+ for (int k = 0; k < keyIndices.length; k++) {
+ spliceBatchColumns[keyIndices[k]] = keyVectorQueues[k].peekFirst();
+ }
+ }
+
+ /**
+ * Closes the key vectors the previous batch was built on. The queues own
them until here, so
+ * survivor memory drains as a row group is emitted rather than all at its
end.
+ */
+ private void releasePublishedKeyVectors() {
+ if (!keyVectorsPublished) return;
+ keyVectorsPublished = false;
+ for (java.util.ArrayDeque<WritableColumnVector> queue : keyVectorQueues) {
+ queue.removeFirst().close();
+ }
}
private void initializeInternal() throws IOException,
UnsupportedOperationException {
missingColumns = new HashSet<>();
for (ParquetColumn column :
CollectionConverters.asJava(parquetColumn.children())) {
checkColumn(column);
}
+ if (storageFilter != null) {
+ initializeLateMaterialization();
+ }
+ }
+
+ /**
+ * Sets the storage filter for late materialization. Must be called before
{@link #initialize};
+ * {@link #initializeLateMaterialization()} then inspects the per-file
schema and decides whether
+ * splicing engages.
+ *
+ * <p>It does not engage in one real case: every key column is missing from
this physical file
+ * under schema evolution. The predicate is then rewritten with each missing
key replaced by the
+ * constant the reader materializes for it (its existence DEFAULT, else
null) and evaluated once.
+ * True keeps the file unfiltered, false or null skips it.
+ *
+ * <p>Everything else the filter needs is guaranteed by
+ * {@code FileSourceStrategy.storageFiltersFor} and {@code
ParquetStorageFilter.create}, so a
+ * violation of it here is a planner bug and is asserted rather than
handled. A file the reader
+ * simply cannot prune is a different matter: it reads it the way a plain
scan would.
+ */
+ public void setStorageFilter(ParquetStorageFilter storageFilter) {
+ this.storageFilter = storageFilter;
+ }
+
+ private void initializeLateMaterialization() throws IOException {
+ if
(!configuration.getBoolean(ParquetInputFormat.COLUMN_INDEX_FILTERING_ENABLED,
true)) {
+ // That conf is the escape hatch for a file whose page index is wrong,
so it has to turn this
+ // feature off whole rather than only its filtering. Phase 2 reads part
of a row group through
+ // the offset index, which parquet consults whatever the conf says, and
a wrong one there
+ // pairs a row's key with another row's values. The post-scan filter
cannot catch that: the
+ // key it sees is the right one.
+ storageFilter = null;
+ return;
+ }
+ lateMatReader = reader.getUnderlyingReader();
+ if (lateMatReader == null) {
+ // Late materialization drives a ParquetFileReader directly, so without
one the filter is
+ // simply not applied and the plain read path runs. The conjunct is in
the post-scan filter as
+ // well, so the rows it would have dropped are dropped above the scan.
+ storageFilter = null;
+ return;
+ }
+ // Resolve each key column's top-level ParquetColumn. Partition into
present and missing (the
+ // latter can happen under schema evolution: a column is in the requested
schema but not in this
+ // physical parquet file). For any non-primitive key we still bail;
phase-1 reads only primitive
+ // leaves.
+ int[] keyIndices = storageFilter.keyColumnIndices();
+ List<ParquetColumn> presentKeyColumns = new ArrayList<>(keyIndices.length);
+ List<Integer> missingKeyLocalPositions = new ArrayList<>();
+ for (int i = 0; i < keyIndices.length; i++) {
+ int idx = keyIndices[i];
+ if (idx < 0 || idx >= parquetColumn.children().size()) {
+ // Unreachable: ParquetStorageFilter.create rejects out-of-range
ordinals.
+ throw new IllegalStateException(String.format(
+ "Storage-filter key ordinal %d is out of range for a %d-column
requested schema",
+ idx, parquetColumn.children().size()));
+ }
+ ParquetColumn column = parquetColumn.children().apply(idx);
+ if (!column.isPrimitive()) {
+ // Unreachable: ParquetStorageFilter.isSupportedKeyType admits only
types with a primitive
+ // Parquet leaf, and it gates both planning and
ParquetStorageFilter.create.
+ throw new IllegalStateException(
+ "Storage-filter key column is not a primitive Parquet column: " +
column.path());
+ }
+ if (missingColumns.contains(column)) {
+ missingKeyLocalPositions.add(i);
+ } else {
+ presentKeyColumns.add(column);
+ }
+ }
+
+ // If any key column is missing from this file, rewrite the predicate to
substitute the constant
+ // the reader will actually materialize for that column. That is the
column's existence DEFAULT
+ // when it has one (ParquetColumnVector writes it into the output vector
and marks the vector
+ // constant), otherwise null. Substituting null for a column that reads
back as its default
+ // would filter on a value the scan never returns. The predicate must be
evaluated against the
+ // substituted constant rather than skipped: null does not always mean
false in a filter.
+ if (!missingKeyLocalPositions.isEmpty()) {
+ int[] missing = new int[missingKeyLocalPositions.size()];
+ Object[] missingValues = new Object[missingKeyLocalPositions.size()];
+ Object[] existenceDefaults =
+ ResolveDefaultColumns.existenceDefaultValues(sparkRequestedSchema);
+ for (int i = 0; i < missing.length; i++) {
+ missing[i] = missingKeyLocalPositions.get(i);
+ missingValues[i] = existenceDefaults[keyIndices[missing[i]]];
+ }
+ storageFilter = storageFilter.rewriteForMissingKeys(missing,
missingValues);
Review Comment:
Adopted, all three parts. The predicate and `keyColumnIndices` are
unchanged, the row phase 1 evaluates takes the vector `initBatch` built for a
key column this file does not have, and splicing covers only the keys the file
does have. The rewrite, its per-task cache and the invariant it needed are gone.
The all-keys-missing decision moved into `initBatch`, after
`super.initBatch`, which is where those vectors exist. So there is one rule for
what a missing column reads as, and it is `ParquetColumnVector`'s.
##########
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 =
+ lateMatReader.readFilteredRowGroup(blockIdx,
pushedFilterRanges)) {
+ if (keyPages == null) {
+ // Unreachable: readFilteredRowGroup returns null only for an empty
block, and we know
+ // pushedFilterRanges selects at least one row. Skipping the block
here would drop its
+ // surviving rows from the output, so assert rather than `continue`.
+ throw new IllegalStateException(
+ "No key pages for row group " + blockIdx + " despite " +
baselineRows
+ + " rows selected by the pushed filter");
+ }
+ survivors = evaluateStorageFilter(keyPages, pushedFilterRanges);
+ }
+ if (survivors != null) {
+ finalRanges = survivors;
+ finalRowCount = survivors.rowCount();
+ if (finalRowCount == 0) {
+ // Every surviving row was rejected by the storage filter; skip the
block entirely,
+ // which avoids the whole non-key baseline. Phase 1 still paid to
read the key columns,
+ // and that cost is not part of the baseline, so nothing is
subtracted from it here.
+ recordRowGroupSkipped(m, baselineRows, nonKeyBaselineBytes);
+ continue;
+ }
+ // Reading part of a row group needs the offset index, so on a file
without one phase 2
+ // reads the whole row group instead. The rows the filter rejected are
then emitted and
+ // the post-scan Filter drops them. What the filter still saves on
such a file is the row
+ // groups it empties, which is decided above and needs no index at all.
+ if (fileHasNoOffsetIndex && finalRowCount < baselineRows) {
+ giveUpFilter();
+ finalRanges = pushedFilterRanges;
+ finalRowCount = baselineRows;
+ }
+ }
+
+ // Phase 2 reads the non-key columns under the surviving rows, or the
whole projection under
+ // `pushedFilterRanges` for a row group whose filter was given up. It is
skipped only when the
+ // projection is all keys and their values were buffered, since emit
then builds every batch
+ // from the key queues alone.
+ long keptRows;
+ if (nonKeyColumns == null && spliceCurrentRowGroup) {
+ keptRows = finalRowCount;
+ } else {
+ lateMatReader.setRequestedSchema(
+ spliceCurrentRowGroup ? nonKeyColumns : requestedColumns);
+ // Reading a strict subset of a block's rows needs a Parquet offset
index, and parquet
+ // enforces that itself: it resolves every requested column's offset
index before reading
+ // anything, and a column without one makes its column index store
throw
+ // MissingOffsetIndexException. Files written before parquet-mr 1.11,
or by a writer that
+ // omits the page index (pyarrow's `write_table` defaults to
`write_page_index=False`), have
+ // none. The filter is then given up for this row group and the read
retried over
+ // `pushedFilterRanges`, which is what a plain scan reads. That retry
cannot hit the same
+ // wall: a store missing one column's offset index reports no column
index either, so
+ // `getRowRanges` could not have narrowed anything and the ranges
cover the whole block.
+ //
+ // Asked this way rather than up front, from the footer. Parquet
resolves the index over the
+ // paths current at its first lookup for the block, which without a
pushed data filter is
+ // the non-key columns alone, so a footer walk over the projection is
both stricter than the
+ // read and blind to an index that is claimed but unreadable. It is
also where the throw
+ // costs least: it lands before any data page is read.
+ try {
+ dataPages = lateMatReader.readFilteredRowGroup(blockIdx,
finalRanges);
+ } catch (MissingOffsetIndexException e) {
+ LOG.warn("Reading {} without page-level storage filtering: reading
part of a row group "
Review Comment:
Fixed for that one. It logs the exception's message rather than the
exception, since a file written without a page index is a shape this feature
expects and degrades on. It stays `WARN` and stays once per file, because which
files degrade is what someone tuning this needs, and the line is one line.
`logFilterGivenUpOnError` keeps the exception on purpose. That is the error
the query would have raised without this feature, and it is the user's own
expression that raised it, so the stack trace is the useful part of it. It is
also once per file.
##########
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 =
+ lateMatReader.readFilteredRowGroup(blockIdx,
pushedFilterRanges)) {
+ if (keyPages == null) {
+ // Unreachable: readFilteredRowGroup returns null only for an empty
block, and we know
+ // pushedFilterRanges selects at least one row. Skipping the block
here would drop its
+ // surviving rows from the output, so assert rather than `continue`.
+ throw new IllegalStateException(
+ "No key pages for row group " + blockIdx + " despite " +
baselineRows
+ + " rows selected by the pushed filter");
+ }
+ survivors = evaluateStorageFilter(keyPages, pushedFilterRanges);
+ }
+ if (survivors != null) {
+ finalRanges = survivors;
+ finalRowCount = survivors.rowCount();
+ if (finalRowCount == 0) {
+ // Every surviving row was rejected by the storage filter; skip the
block entirely,
+ // which avoids the whole non-key baseline. Phase 1 still paid to
read the key columns,
+ // and that cost is not part of the baseline, so nothing is
subtracted from it here.
+ recordRowGroupSkipped(m, baselineRows, nonKeyBaselineBytes);
+ continue;
+ }
+ // Reading part of a row group needs the offset index, so on a file
without one phase 2
+ // reads the whole row group instead. The rows the filter rejected are
then emitted and
+ // the post-scan Filter drops them. What the filter still saves on
such a file is the row
+ // groups it empties, which is decided above and needs no index at all.
+ if (fileHasNoOffsetIndex && finalRowCount < baselineRows) {
+ giveUpFilter();
+ finalRanges = pushedFilterRanges;
+ finalRowCount = baselineRows;
+ }
+ }
+
+ // Phase 2 reads the non-key columns under the surviving rows, or the
whole projection under
+ // `pushedFilterRanges` for a row group whose filter was given up. It is
skipped only when the
+ // projection is all keys and their values were buffered, since emit
then builds every batch
+ // from the key queues alone.
+ long keptRows;
+ if (nonKeyColumns == null && spliceCurrentRowGroup) {
Review Comment:
Done. The reader answers an all-keys projection with `storageFilter = null`
and reads plainly, so that branch, the `dataPages` guard and the `columns ==
null` arm of the byte walk are all gone.
Your prediction about the tests was right. Every key-type fixture carries a
non-key column now, and the helper that writes them says why, so a read that
keeps every row still goes through the survivor queues.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetStorageFilter.scala:
##########
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.parquet
+
+import java.util.concurrent.ConcurrentHashMap
+
+import org.apache.spark.SparkThrowable
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{And, BasePredicate,
BloomFilterMightContain, BoundReference, Expression, Literal, Predicate,
XxHash64}
+import org.apache.spark.sql.catalyst.trees.TreePattern.PLAN_EXPRESSION
+import org.apache.spark.sql.execution.metric.SQLMetric
+import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType,
DataType, DateType, DayTimeIntervalType, DecimalType, DoubleType, FloatType,
IntegerType, LongType, ShortType, StringType, StructType, TimestampNTZType,
TimestampType, TimeType, YearMonthIntervalType}
+
+/**
+ * Optional SQL metrics the reader updates while applying a
[[ParquetStorageFilter]]. Every
+ * counter is scoped to what the storage filter added on top of a read of the
same projection
+ * without one. All fields are nullable; a null field disables that metric.
+ *
+ * - [[rowGroupsSkipped]] counts row groups whose data columns were never
read.
+ * - [[rowsExcludedByRowGroup]] sums the rows those skips excluded, per
skipped block the rows that
+ * survived the pushed data filter.
+ * - [[rowsExcludedWithinRowGroup]] sums rows excluded inside row groups that
were kept.
+ * - [[bytesAvoidedByRowGroup]] sums, per skipped row group, the non-key
bytes a plain read of this
+ * projection would have transferred for the rows that survived the pushed
data filter. Phase 1
+ * reads the key columns of every block, so key bytes are never part of it,
and it is zero on an
+ * all-keys projection, which can avoid nothing.
+ * - [[bytesAvoidedByPageFiltering]] sums, per kept row group, that same
non-key baseline minus the
+ * bytes phase 2 read, which is what `finalRanges` page selection pruned.
+ *
+ * The row counters' suffix says where a row was excluded, not what would have
saved it: a row
+ * inside a kept row group is read as part of its page and dropped during
decode, so page
+ * filtering did not save it. An all-keys projection has no page filtering at
all, and
+ * [[rowsExcludedWithinRowGroup]] still counts every row the filter dropped.
+ */
+case class StorageFilterMetrics(
Review Comment:
Done. `StorageFilterMetrics` takes all five as required fields, the null
checks and the `needBytes` mode are gone, and the `storageFilterMetrics`
parameter has no default any more.
The keys and the labels moved onto its companion, so no format code names
one, and the tests build real `SQLMetric`s through the same `create` the
production path uses.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:
##########
@@ -642,7 +656,26 @@ trait FileSourceScanLike extends DataSourceScanExec with
SessionStateHelper {
} else {
None
}
- } ++ driverMetrics
+ } ++ storageFilterMetrics ++ driverMetrics
+
+ protected lazy val storageFilterMetrics: Map[String, SQLMetric] = if
(storageFilters.nonEmpty) {
Review Comment:
Done, with the hook you propose.
`FileFormat.storageFilterMetrics(sparkContext)` answers with the format's own
metric map, and `FileSourceScanLike` adds whatever it gets to `metrics`. So the
scan names nothing, and a format that counts something else declares it. What
there is to count is the format's own vocabulary, since row groups and page
filtering are Parquet's words.
--
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]
