peter-toth commented on code in PR #58895:
URL: https://github.com/apache/spark/pull/58895#discussion_r4133524917
##########
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;
Review Comment:
Fixed, and the narrowing is gone entirely. The catch is `catch
(RuntimeException e)` now, and `isEvaluationError` is deleted.
Your `timestamp_seconds(decimal_col)` case is the one named in the comment
above it, because an `ArithmeticException` out of `longValueExact` is exactly
what a `SparkThrowable` test could not see. The all-keys-missing decision does
the same and reads the file unfiltered when the constant evaluation throws.
`supportsStorageFilter` no longer asks `canEvaluateUnconditionally` either, so
a widening `Cast` key is pushed, which is what type coercion inserts for an
int-to-bigint join, and an error from it gives the row group up.
##########
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);
Review Comment:
Documented in the conf doc, in your terms.
`spark.sql.parquet.storageFilterPushdown.enabled` now says that narrowing a row
group to part of its rows trusts the page index, that a read with no pushed
data filter never consults it, that a wrong index there can pair a row's key
with another row's values, and that `parquet.filter.columnindex.enabled=false`
is the escape hatch.
That conf also changed. It no longer turns the feature off, it leaves the
filter with the row groups it empties, so the hatch costs the page-level saving
and nothing more.
##########
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);
+ if (blockRow != previousSurvivor + 1) survivorRangeCount++;
+ previousSurvivor = blockRow;
+ if (spliceCurrentRowGroup) splicedBytes +=
appendSurvivorRowToAccumulators(r);
+ // Both halves of what this row group retains grow per survivor, and
either can cross
+ // the budget on its own, so they are weighed together here and
nowhere else. The cheaper
+ // concession comes first: release the buffer, and give the filter
up as well if the
+ // ranges alone still do not fit.
+ long rangeBytes = rowRangeStateBytes(survivorRangeCount);
+ if (splicedBytes + rangeBytes > cap) {
Review Comment:
The counting is fixed. A variable-length key is charged 17 bytes per row
rather than 9, which is the offset, the length, the null byte, and the byte
child's eager four bytes with a null byte for each of them. For your 10-byte
key at capacity 4096 the charge is now above what the vectors hold, which is
the side to be wrong on.
The eagerly allocated next accumulator is gone too. The push and the
reallocation moved to the top of the append, so a row group whose survivors are
a multiple of the capacity leaves no unused set behind. What is still uncounted
is the first set of a row group, which is capacity-sized whatever the survivors
turn out to be, so the floor is one set per key column instead of a multiple of
the cap.
The `MemoryConsumer` route is in the description's follow-up list, with
`abandonSplicing()` as the `spill()` you point at. I kept it out of this PR
because the reader would have to hold a `TaskMemoryManager`, and because this
buffer cannot spill, it can only be discarded, which is what the cap already
does.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -391,15 +574,54 @@ public void enableReturningBatches() {
* Advances to the next batch of rows. Returns false if there are no more.
*/
public boolean nextBatch() throws IOException {
+ releasePublishedKeyVectors();
for (ParquetColumnVector vector : columnVectors) {
vector.reset();
}
+ // Zeroed before the terminal checks below, so a terminal call cannot
leave a spliced batch
+ // pointing at the key vectors just released, which off heap is freed
memory.
columnarBatch.setNumRows(0);
+ if (hitEndOfData) return false;
if (rowsReturned >= totalRowCount) return false;
checkEndOfRowGroup();
+ if (hitEndOfData) return false;
int num = (int) Math.min(capacity, totalCountLoadedSoFar - rowsReturned);
- for (ParquetColumnVector cv : columnVectors) {
+ // A spliced row group takes its key slots from the survivor queues, so
phase 2 skipped them.
+ // Everything else read every projected column: a plain read with no
storage filter at all, or a
+ // row group that gave splicing up, whose batch needs its key slots put
back to the persistent
+ // vectors a previous row group rewrote. `spliceCurrentRowGroup` stays
false without a storage
+ // filter, so it answers for all three.
+ if (spliceCurrentRowGroup) {
+ publishSurvivorKeyVectors(num);
+ } else if (spliceBatchColumns != null) {
+ System.arraycopy(persistentBatchColumns, 0, spliceBatchColumns, 0,
spliceBatchColumns.length);
+ }
+ readPersistentColumns(num, /* skipKeySlots= */ spliceCurrentRowGroup);
+ // If needed, compute row indexes within a file. The row-index column is
identified by name
+ // (ROW_INDEX_TEMPORARY_COLUMN_NAME), a synthetic metadata column no
storage filter references,
+ // so its slot is always a non-key one and its ParquetColumnVector is
always the persistent one.
+ if (rowIndexGenerator != null) {
Review Comment:
Fixed by rejecting it. `ParquetStorageFilter.isSupportedStorageFilter`
answers false for a conjunct that references a column named
`_tmp_metadata_row_index`, so the planner never offers one, and `create` has a
`require` for the same thing in case anything else builds a filter.
The reason is yours: this reader writes row indexes over whatever that
column holds, so its value would depend on how each row group was read. Both
checks go through a new `ParquetRowIndexUtil.isRowIndexColumnName`, so one
place knows the name.
##########
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:
You are right, and it is inherent to evaluating the filter before the row
group's first batch. Phase 1 decodes every key page of the row group, so a
corrupt page anywhere in it costs that whole row group, where a plain read
would have returned the batches before it.
Under `ignoreCorruptFiles` the answer is a subset either way, and how big a
subset already depends on where the batch boundaries fall, so the result
shrinks rather than being wrong. I left the conf doc alone because it was
shortened in the same round. Say if you want a line there.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala:
##########
@@ -190,6 +192,53 @@ class ParquetFileFormat
filters: Seq[Filter],
options: Map[String, String],
hadoopConf: Configuration): PartitionedFile => Iterator[InternalRow] = {
+ buildParquetReader(
+ sparkSession, dataSchema, partitionSchema, requiredSchema, filters, Nil,
options, hadoopConf,
+ Map.empty)
+ }
+
+ /**
+ * The conf that turns this on is read here rather than in the planner, the
way `supportBatch`
+ * reads its own confs, so a Parquet-named conf does not decide for a format
that is not Parquet.
+ *
+ * Subclasses answer false on purpose, even though they inherit this reader:
a subclass may
+ * customize reading by overriding `buildReaderWithPartitionValues`, and a
scan with storage
+ * filters routes through `buildReaderWithStorageFilters` instead, which
would silently bypass
+ * whatever the subclass does.
+ */
+ override def supportsStorageFilterPushdown(sparkSession: SparkSession):
Boolean =
+ getSqlConf(sparkSession).parquetStorageFilterPushdownEnabled &&
+ getClass == classOf[ParquetFileFormat]
+
+ override def supportsStorageFilter(expr: Expression): Boolean =
+ ParquetStorageFilter.isSupportedStorageFilter(expr)
+
+ override def buildReaderWithStorageFilters(
Review Comment:
Done. `buildReaderWithStorageFilters` answers `None` for a subclass, for the
reason you give, that this entry point is reachable without the planner having
asked the per-scan question, and that `None` sends the caller to the builder a
subclass overrides.
The fallback is tested end to end now as well. A scan whose relation carries
a `ParquetFileFormat` subclass and a storage filter returns every row of the
file through the ordinary reader, and the test fails if the guard is removed.
##########
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();
Review Comment:
I kept the swap, and the pieces it needs are fewer than they were.
The copy would pay a value copy per surviving row at emit, on top of the one
phase 1 already pays into the accumulator, and for a variable-length key that
is a byte copy each. The swap is one array write per key column per batch. The
`arraycopy` restore is gone, because the emit path now points every key slot at
something on every batch rather than repairing the array afterwards.
`closeAllocated` is the one override that replaces rather than extends, and it
exists because the batch's key slots alias the queue heads. So what is left is
the two arrays and the `resultBatch()` contract.
On the caller that holds a column across `nextBatch()`,
`DataSourceReadBenchmark:235` does exactly that and is unaffected, because the
plain reader never rewrites a slot. Only a caller that both asks for a storage
filter and holds a column across batches would read a released vector, and
`resultBatch()`'s javadoc now says the array may be rewritten per batch. If you
would rather have the copy, it is contained enough for a follow-up.
--
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]
