cloud-fan commented on code in PR #58895:
URL: https://github.com/apache/spark/pull/58895#discussion_r4103839051
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -492,6 +930,630 @@ 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 {
+ while (nextBlockIndex < totalBlockCount) {
+ int blockIdx = nextBlockIndex++;
+ long blockRowCount =
lateMatReader.getRowGroups().get(blockIdx).getRowCount();
+ if (blockRowCount == 0) {
+ // parquet-mr never writes these, but RowRanges.createSingle(0) would
build Range(0, -1) and
+ // trip parquet's own `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. A file already known to have no offset index
starts there.
+ filterGivenUp = fileHasNoOffsetIndex;
+ spliceCurrentRowGroup = !filterGivenUp;
+ splicedBytes = 0L;
+
+ // 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);
+ // getRowRanges checks only whether a filter is pushed, not
options.useColumnIndexFilter(),
+ // so calling it unconditionally would keep applying column-index
filtering after a user
+ // turned it off, which is the escape hatch for a file whose column
index is wrong. Every
+ // phase below reads within these ranges, so a wrong column index would
cost rows the plain
+ // path would have returned. Phase 2 is unaffected: it selects pages
through the offset index,
+ // a separate structure this conf says nothing about.
+ RowRanges pushedFilterRanges = useColumnIndexFilter
+ ? lateMatReader.getRowRanges(blockIdx)
+ : RowRanges.createSingle(blockRowCount);
+ // 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. A row group
whose filter is already
+ // given up reports nothing either way, so it does not pay for the
baseline at all.
+ StorageFilterMetrics m = storageFilter.metrics();
+ SQLMetric bytesAvoidedRg = m.bytesAvoidedByRowGroup();
+ SQLMetric bytesAvoidedPf = m.bytesAvoidedByPageFiltering();
+ boolean needBytes = (bytesAvoidedRg != null || bytesAvoidedPf != null)
&& !filterGivenUp;
+ Map<ColumnPath, ColumnChunkMetaData> blockChunks =
+ needBytes ? chunksByPath(lateMatReader, blockIdx) : null;
+ long nonKeyBaselineBytes = needBytes
+ ? compressedBytesForRowRanges(lateMatReader, blockIdx, blockChunks,
nonKeyColumns,
+ pushedFilterRanges, baselineRows)
+ : 0L;
+
+ // Phase 1: switch to key-only schema, read key columns under
pushedFilterRanges, evaluate the
+ // storage filter per row. Skipped for a row group the filter is already
given up for, which
+ // leaves every row of `pushedFilterRanges` to emit, exactly what a
plain read would.
+ RowRanges finalRanges = pushedFilterRanges;
+ long finalRowCount = baselineRows;
+ if (!filterGivenUp) {
+ lateMatReader.setRequestedSchema(keyOnlyColumns);
+ PageReadStore keyPages = lateMatReader.readFilteredRowGroup(blockIdx,
pushedFilterRanges);
Review Comment:
**Non-blocking (P2):** Please close this phase-1 `PageReadStore` after
`evaluateStorageFilter` finishes, including skip, give-up, and exception exits.
`readFilteredRowGroup` returns an `AutoCloseable` store whose page readers and
`ByteBufferReleaser` otherwise stay live; this path acquires one extra store
for every attempted row group, so long scans can accumulate direct-memory
pressure. A try-with-resources around the phase-1 acquisition and evaluation
matches the ownership boundary.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -492,6 +930,630 @@ 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 {
+ while (nextBlockIndex < totalBlockCount) {
+ int blockIdx = nextBlockIndex++;
+ long blockRowCount =
lateMatReader.getRowGroups().get(blockIdx).getRowCount();
+ if (blockRowCount == 0) {
+ // parquet-mr never writes these, but RowRanges.createSingle(0) would
build Range(0, -1) and
+ // trip parquet's own `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. A file already known to have no offset index
starts there.
+ filterGivenUp = fileHasNoOffsetIndex;
+ spliceCurrentRowGroup = !filterGivenUp;
+ splicedBytes = 0L;
+
+ // 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);
+ // getRowRanges checks only whether a filter is pushed, not
options.useColumnIndexFilter(),
+ // so calling it unconditionally would keep applying column-index
filtering after a user
+ // turned it off, which is the escape hatch for a file whose column
index is wrong. Every
+ // phase below reads within these ranges, so a wrong column index would
cost rows the plain
+ // path would have returned. Phase 2 is unaffected: it selects pages
through the offset index,
+ // a separate structure this conf says nothing about.
+ RowRanges pushedFilterRanges = useColumnIndexFilter
+ ? lateMatReader.getRowRanges(blockIdx)
+ : RowRanges.createSingle(blockRowCount);
+ // 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. A row group
whose filter is already
+ // given up reports nothing either way, so it does not pay for the
baseline at all.
+ StorageFilterMetrics m = storageFilter.metrics();
+ SQLMetric bytesAvoidedRg = m.bytesAvoidedByRowGroup();
+ SQLMetric bytesAvoidedPf = m.bytesAvoidedByPageFiltering();
+ boolean needBytes = (bytesAvoidedRg != null || bytesAvoidedPf != null)
&& !filterGivenUp;
+ Map<ColumnPath, ColumnChunkMetaData> blockChunks =
+ needBytes ? chunksByPath(lateMatReader, blockIdx) : null;
+ long nonKeyBaselineBytes = needBytes
+ ? compressedBytesForRowRanges(lateMatReader, blockIdx, blockChunks,
nonKeyColumns,
+ pushedFilterRanges, baselineRows)
+ : 0L;
+
+ // Phase 1: switch to key-only schema, read key columns under
pushedFilterRanges, evaluate the
+ // storage filter per row. Skipped for a row group the filter is already
given up for, which
+ // leaves every row of `pushedFilterRanges` to emit, exactly what a
plain read would.
+ RowRanges finalRanges = pushedFilterRanges;
+ long finalRowCount = baselineRows;
+ if (!filterGivenUp) {
+ lateMatReader.setRequestedSchema(keyOnlyColumns);
+ PageReadStore keyPages = lateMatReader.readFilteredRowGroup(blockIdx,
pushedFilterRanges);
+ if (keyPages == null) {
+ // Unreachable: readFilteredRowGroup returns null only for an empty
block, and we already
+ // 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");
+ }
+ RowRanges survivors = evaluateStorageFilter(keyPages,
pushedFilterRanges);
+ if (!filterGivenUp
+ && rowRangeStateBytes(survivorRangeCount) >
storageFilter.maxSplicedRowGroupBytes()) {
+ // Phase 1 weighs the budget once per accumulator, so a row group
whose survivors fit in
+ // a single one is only caught here, with its survivors buffered.
Those are released,
+ // since the ranges they were spliced against are about to be thrown
away.
+ giveUpFilter();
+ }
+ if (!filterGivenUp) {
+ 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;
+ }
+ }
+ }
+
+ // 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;
+ long phase2Bytes;
+ PageReadStore dataPages = null;
+ if (nonKeyColumns == null && spliceCurrentRowGroup) {
+ keptRows = finalRowCount;
+ phase2Bytes = 0L;
+ } 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.
+ //
+ // Nothing is checked up front, so a file with no page index still
reads with the filter
+ // applied wherever the filter keeps a row group whole
(`readFilteredRowGroup` degrades to a
+ // plain read when the ranges cover the block) or rejects one whole.
+ try {
+ dataPages = lateMatReader.readFilteredRowGroup(blockIdx,
finalRanges);
+ } catch (MissingOffsetIndexException e) {
+ LOG.warn("Not applying the storage filter to {}: reading part of a
row group needs a "
+ + "Parquet offset index, and this file was written without a
page index for at least "
+ + "one projected column", e, MDC.of(LogKeys.PATH,
lateMatReader.getFile()));
+ fileHasNoOffsetIndex = true;
+ giveUpFilter();
+ finalRanges = pushedFilterRanges;
+ finalRowCount = baselineRows;
+ 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 computed for a row group whose filter was given up: it
read what a plain scan
+ // reads, so the answer is a certain zero. `needBytes`, not just
`bytesAvoidedPf != null`,
+ // because that is what built `blockChunks`.
+ if (needBytes && bytesAvoidedPf != null && !filterGivenUp) {
+ phase2Bytes = compressedBytesForRowRanges(lateMatReader, 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(lateMatReader,
blockIdx, blockChunks,
+ keyOnlyColumns, finalRanges, finalRowCount);
+ }
+ } else {
+ phase2Bytes = 0L;
+ }
+ }
+ long filteredRows = baselineRows - keptRows;
+ SQLMetric rowsExcludedWithinRg = m.rowsExcludedWithinRowGroup();
+ if (rowsExcludedWithinRg != null && filteredRows > 0)
rowsExcludedWithinRg.add(filteredRows);
+ if (bytesAvoidedPf != null && !filterGivenUp) {
+ // `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);
+ }
+
+ 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. `getRowRanges` is a cache hit whenever the two can differ,
because
+ // `getFilteredRecordCount()` at initialize resolved every block's
ranges then. It has to be
+ // guarded the same way phase 0 guards it, since it consults the pushed
filter but not the
+ // conf that turns column-index filtering off.
+ long blockRowCount = blocks.get(blockIdx).getRowCount();
+ if (blockRowCount == 0) continue;
+ RowRanges blockRanges = useColumnIndexFilter
+ ? lateMatReader.getRowRanges(blockIdx)
+ : RowRanges.createSingle(blockRowCount);
+ 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(lateMatReader, 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 static long compressedBytesForRowRanges(
+ ParquetFileReader reader,
+ int blockIndex,
+ Map<ColumnPath, ColumnChunkMetaData> chunks,
+ List<ColumnDescriptor> columns,
+ RowRanges rowRanges,
+ long rowRangeCount) {
+ if (columns == null || columns.isEmpty() || rowRangeCount == 0) {
+ return 0L;
+ }
+ long blockRowCount = reader.getRowGroups().get(blockIndex).getRowCount();
+ boolean wholeBlock = rowRangeCount == blockRowCount;
+ ColumnIndexStore ciStore = wholeBlock ? null :
reader.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) {
+ 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);
+ }
+ ensureCurrentKeyAccumulatorsAllocated();
+
+ PrimitiveIterator.OfLong rowIndexIter = pushedFilterRanges.iterator();
+ RowRanges.Builder finalRangesBuilder = RowRanges.builder();
+ 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();
+ boolean accumulate = true;
+ 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();
+ if (storageFilter.test(keyScratchBatch.getRow(r))) {
+ finalRangesBuilder.addSelectedRow(blockRow);
+ if (blockRow != previousSurvivor + 1) survivorRangeCount++;
+ previousSurvivor = blockRow;
+ if (accumulate) {
+ accumulate = appendSurvivorRowToAccumulators(r);
Review Comment:
**Blocking (P1):** Once this returns false, `evaluateStorageFilter` still
adds every later match to `finalRangesBuilder`, but the per-survivor budget
check is disabled with accumulation and the remaining check happens only after
the full row group. Wide keys can therefore give up splicing early and still
build an unbounded scattered range list, violating the configured per-row-group
memory cap. Please make the survivor transition own both key-byte and range
accounting, including after splicing is abandoned, and fall back as soon as the
combined budget is crossed.
**Recommended change:** Centralize the combined spliced-value and row-range
budget check in the per-survivor loop, apply it to partial accumulators and
filter-only mode, and stop retaining survivor state immediately when the bound
is exceeded. Align the config documentation and add boundary scenarios for
large values and post-give-up scattered ranges.
**Why this works:** After each survivor changes retained value or range
state, compute the same combined accounting regardless of accumulator fullness
or spliceCurrentRowGroup. If the next retained state exceeds the cap, release
key accumulators, abandon the storage filter for that row group, and avoid
constructing the remainder of the survivor range list.
**Scope:** Make one row-group survivor-state owner enforce the advertised
memory bound in every accumulation mode.
**Compatibility:** Eligible scans may still splice within budget, while
every give-up path continues to behave as an optional optimization and
preserves query results.
**Risks:** Giving up in the middle of evaluation must not leave partial key
vectors or ranges observable by emit. Checking before versus after adding a
survivor must use a consistent inclusive cap boundary.
**Constraints:** Fallback must keep every row selected by pushedFilterRanges
because the post-scan conjunct remains authoritative. The accounting may
acknowledge column-vector capacity slack, but it must not claim a hard bound it
does not enforce.
**Success:** A partial accumulator of large variable-width keys cannot
remain retained beyond the configured combined budget. Once splicing is
abandoned, later scattered survivors cannot grow range state beyond the same
budget. Budget give-up returns the same rows as a plain read and releases
partial survivor vectors.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -492,6 +930,630 @@ 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 {
+ while (nextBlockIndex < totalBlockCount) {
+ int blockIdx = nextBlockIndex++;
+ long blockRowCount =
lateMatReader.getRowGroups().get(blockIdx).getRowCount();
+ if (blockRowCount == 0) {
+ // parquet-mr never writes these, but RowRanges.createSingle(0) would
build Range(0, -1) and
+ // trip parquet's own `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. A file already known to have no offset index
starts there.
+ filterGivenUp = fileHasNoOffsetIndex;
+ spliceCurrentRowGroup = !filterGivenUp;
+ splicedBytes = 0L;
+
+ // 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);
+ // getRowRanges checks only whether a filter is pushed, not
options.useColumnIndexFilter(),
+ // so calling it unconditionally would keep applying column-index
filtering after a user
+ // turned it off, which is the escape hatch for a file whose column
index is wrong. Every
+ // phase below reads within these ranges, so a wrong column index would
cost rows the plain
+ // path would have returned. Phase 2 is unaffected: it selects pages
through the offset index,
+ // a separate structure this conf says nothing about.
+ RowRanges pushedFilterRanges = useColumnIndexFilter
+ ? lateMatReader.getRowRanges(blockIdx)
+ : RowRanges.createSingle(blockRowCount);
+ // 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. A row group
whose filter is already
+ // given up reports nothing either way, so it does not pay for the
baseline at all.
+ StorageFilterMetrics m = storageFilter.metrics();
+ SQLMetric bytesAvoidedRg = m.bytesAvoidedByRowGroup();
+ SQLMetric bytesAvoidedPf = m.bytesAvoidedByPageFiltering();
+ boolean needBytes = (bytesAvoidedRg != null || bytesAvoidedPf != null)
&& !filterGivenUp;
+ Map<ColumnPath, ColumnChunkMetaData> blockChunks =
+ needBytes ? chunksByPath(lateMatReader, blockIdx) : null;
+ long nonKeyBaselineBytes = needBytes
+ ? compressedBytesForRowRanges(lateMatReader, blockIdx, blockChunks,
nonKeyColumns,
+ pushedFilterRanges, baselineRows)
+ : 0L;
+
+ // Phase 1: switch to key-only schema, read key columns under
pushedFilterRanges, evaluate the
+ // storage filter per row. Skipped for a row group the filter is already
given up for, which
+ // leaves every row of `pushedFilterRanges` to emit, exactly what a
plain read would.
+ RowRanges finalRanges = pushedFilterRanges;
+ long finalRowCount = baselineRows;
+ if (!filterGivenUp) {
+ lateMatReader.setRequestedSchema(keyOnlyColumns);
+ PageReadStore keyPages = lateMatReader.readFilteredRowGroup(blockIdx,
pushedFilterRanges);
Review Comment:
**Blocking (P1):** This key-only read is the first offset-index-dependent
operation, but it sits outside the `MissingOffsetIndexException` fallback used
for phase 2. A valid mixed-index file can have `pushedFilterRanges` narrowed by
one indexed column while this key column lacks an offset index; enabling this
optional optimization then fails a query the plain reader accepts. Please
extend or preflight the fallback at this phase-1 boundary so the row group is
read plainly.
--
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]
