peter-toth commented on code in PR #58895:
URL: https://github.com/apache/spark/pull/58895#discussion_r4097170674
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala:
##########
@@ -151,6 +151,60 @@ object FileSourceStrategy extends Strategy with
PredicateHelper with Logging {
}
}
+ /**
+ * Splits `afterScanFilters` into conjuncts the file format can evaluate at
the storage layer for
+ * late materialization (returned as the first element) and the remaining
filters that stay as a
+ * post-scan FilterExec (returned as the second element).
+ *
+ * Everything is checked here, statically, so the runtime never silently
loses a filter. Two of
+ * the conditions are per scan, and failing either leaves every filter in
the plan:
+ * - The storage-filter pushdown SQL conf is on.
+ * - [[FileFormat.supportBatch]] holds for the schema the reader will see,
+ * `partitionSchema ++ outputDataSchema`, which is the schema a format's
reader builder derives
+ * its own vectorized-read decision from. Late materialization needs a
batch read, so this asks
+ * about batch support rather than naming a format.
+ *
+ * The rest are per conjunct, and one that fails any of them stays in the
post-scan Filter:
+ * - It is deterministic. A reader evaluates the predicate without
+ * `BasePredicate.initialize(partitionIndex)`, which `GeneratePredicate`
emits for a
+ * `Nondeterministic` expression, so a non-deterministic conjunct would
fail at task time.
+ * - It references at least one column, and every column it references is a
projected data
+ * column. A reference to something the scan does not read cannot be
evaluated by the reader.
+ * - [[FileFormat.supportsStorageFilter]] accepts it. That is where the
expression shapes and
+ * column types a reader can evaluate live, so this method names neither
a format nor a type.
+ *
+ * One last condition is on the set that survives: at least one projected
data column must be left
+ * for the reader to prune. A scan that projects nothing but the filter's
own key columns reads
+ * the same columns for the same rows either way, since the reader has to
read a key column to
+ * evaluate the filter on it, so pushing can only add the cost of evaluating
the predicate outside
+ * the generated code. Extraction is dropped entirely in that case.
+ */
+ private def extractStorageFilters(
+ afterScanFilters: ExpressionSet,
+ fsRelation: HadoopFsRelation,
+ readDataColumns: Seq[Attribute],
+ outputDataSchema: StructType): (Seq[Expression], ExpressionSet) = {
+ val sparkSession = fsRelation.sparkSession
+ val sqlConf = sparkSession.sessionState.conf
+ if (!sqlConf.parquetStorageFilterPushdownEnabled) return (Nil,
afterScanFilters)
+ val resultSchema = StructType(fsRelation.partitionSchema.fields ++
outputDataSchema.fields)
+ if (!fsRelation.fileFormat.supportBatch(sparkSession, resultSchema)) {
+ return (Nil, afterScanFilters)
+ }
+
+ val dataAttrs = AttributeSet(readDataColumns)
+ val (eligible, rest) = afterScanFilters.partition { expr =>
Review Comment:
Real, and reproduced rather than reasoned about: with the gate ablated, a
`CAST(s AS BIGINT)` key plus a `WHERE kind = 'num'` guard over a non-numeric
row in the same row group fails with `CAST_INVALID_INPUT`.
`ParquetStorageFilter.isSupportedStorageFilter` now requires the hash's
children to pass `ExprUtils.canEvaluateUnconditionally`, which is the whitelist
master already uses for this hazard class. That covers `evalAllMissing` too,
since it evaluates the same expression. You were right about `throwable`: it is
opt-in and `Cast` does not set it, which `ExprUtils`' own scaladoc calls out.
Tested on the gate and end to end, with the offending row in the same row
group as matching ones so no statistic prunes it away. c255f88
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -492,6 +938,535 @@ 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 (non-key schema): read non-key columns restricted to {@code
finalRanges}.
+ * Skipped entirely when {@link #nonKeyColumns} is null (all-keys
projection).
+ *
+ * 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 and gives up past the cap. This row group is then read the
plain way: phase 2 takes
+ // every projected column under finalRanges, key columns included, so
nothing is buffered and
+ // the cost is one extra read of the key columns.
+ spliceCurrentRowGroup = true;
+ 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 -- the documented escape hatch for files whose column
index is wrong.
+ // Trusting a wrong column index here drops rows for good, since
finalRanges is a subset of
+ // these ranges and the post-scan Filter no longer holds the predicate.
Phase 2 is
+ // unaffected: it selects pages through the offset index, which 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.
+ 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(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.
+ 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 UnsupportedFileReadException(
+ "No key pages for row group " + blockIdx + " despite " +
baselineRows
+ + " rows selected by the pushed filter");
+ }
+ RowRanges finalRanges = evaluateStorageFilter(keyPages,
pushedFilterRanges);
+ long finalRowCount = finalRanges.rowCount();
+
+ if (finalRowCount == 0) {
+ // Every surviving row was rejected by the storage filter; skip 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 has to be subtracted from it
here.
+ SQLMetric rgSkipped = m.rowGroupsSkipped();
+ if (rgSkipped != null) rgSkipped.add(1L);
+ SQLMetric rowsExcludedRg = m.rowsExcludedByRowGroup();
+ if (rowsExcludedRg != null) rowsExcludedRg.add(baselineRows);
+ if (bytesAvoidedRg != null) bytesAvoidedRg.add(nonKeyBaselineBytes);
+ continue;
+ }
+
+ // Phase 2: switch to non-key schema, read non-key columns under
finalRanges. Skipped entirely
+ // when the projection is all keys (nonKeyColumns is null); emit
reconstructs each batch from
+ // the key queues alone.
+ long keptRows;
+ long phase2Bytes;
+ PageReadStore dataPages = null;
+ // An all-keys projection has nothing for phase 2 to read -- but only if
the key values were
+ // buffered. A row group past the cap has to read them here like any
other column.
+ 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. Degrading to a whole-block read is not an option,
because the key vectors
+ // hold only the survivors and the batch would misalign, and neither
is dropping the
+ // predicate, which extraction has removed from the post-scan Filter.
So the read fails,
+ // and all this adds is what the user can do about it.
+ //
+ // Nothing is checked up front: a row group the filter keeps whole
never needs the index
+ // (`readFilteredRowGroup` falls back to a plain read when the ranges
cover the block), and
+ // one it rejects whole is never read at all, so a file with no page
index still scans as
+ // long as the filter never has to prune inside a row group.
+ try {
+ dataPages = lateMatReader.readFilteredRowGroup(blockIdx,
finalRanges);
+ } catch (MissingOffsetIndexException e) {
+ throw new UnsupportedFileReadException(String.format(
+ "Storage-filter pushdown needs a Parquet offset index to read
the %d of %d rows its "
+ + "filter kept in row group %d of %s, but the file was
written without a page "
+ + "index. Set %s=false to read this file.",
+ finalRowCount, blockRowCount, blockIdx, lateMatReader.getFile(),
+
SQLConf$.MODULE$.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED().key()), e);
+ }
+ 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 UnsupportedFileReadException(
+ "No data pages for row group " + blockIdx + " despite " +
finalRowCount
+ + " surviving rows");
+ }
+ keptRows = dataPages.getRowCount();
+ // `needBytes`, not just `bytesAvoidedPf != null`: it is what built
`blockChunks`.
+ if (needBytes && bytesAvoidedPf != null) {
+ 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 --
+ // which can make the row group's contribution negative, and that
is the truth about 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) {
+ 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;
+ }
+
+
+ /**
+ * 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.
+ * </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.
+ */
+ 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();
+ // 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))) {
Review Comment:
Gone with the design change in the `DataSourceUtils` thread: there is no
`UnsupportedFileReadException` and no case for it in the shared classifier any
more.
The cast error you describe cannot reach the reader either, since such an
expression is no longer offered. What is left is that an internal error in this
reader is swallowed under `ignoreCorruptFiles` the way any other reader's is,
which is now the same behaviour as the rest of the read path rather than a
special case. c255f88
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -492,6 +938,535 @@ 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 (non-key schema): read non-key columns restricted to {@code
finalRanges}.
+ * Skipped entirely when {@link #nonKeyColumns} is null (all-keys
projection).
+ *
+ * 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 and gives up past the cap. This row group is then read the
plain way: phase 2 takes
+ // every projected column under finalRanges, key columns included, so
nothing is buffered and
+ // the cost is one extra read of the key columns.
+ spliceCurrentRowGroup = true;
+ 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 -- the documented escape hatch for files whose column
index is wrong.
+ // Trusting a wrong column index here drops rows for good, since
finalRanges is a subset of
+ // these ranges and the post-scan Filter no longer holds the predicate.
Phase 2 is
+ // unaffected: it selects pages through the offset index, which 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.
+ 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(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.
+ 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 UnsupportedFileReadException(
+ "No key pages for row group " + blockIdx + " despite " +
baselineRows
+ + " rows selected by the pushed filter");
+ }
+ RowRanges finalRanges = evaluateStorageFilter(keyPages,
pushedFilterRanges);
+ long finalRowCount = finalRanges.rowCount();
+
+ if (finalRowCount == 0) {
+ // Every surviving row was rejected by the storage filter; skip 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 has to be subtracted from it
here.
+ SQLMetric rgSkipped = m.rowGroupsSkipped();
+ if (rgSkipped != null) rgSkipped.add(1L);
+ SQLMetric rowsExcludedRg = m.rowsExcludedByRowGroup();
+ if (rowsExcludedRg != null) rowsExcludedRg.add(baselineRows);
+ if (bytesAvoidedRg != null) bytesAvoidedRg.add(nonKeyBaselineBytes);
+ continue;
+ }
+
+ // Phase 2: switch to non-key schema, read non-key columns under
finalRanges. Skipped entirely
+ // when the projection is all keys (nonKeyColumns is null); emit
reconstructs each batch from
+ // the key queues alone.
+ long keptRows;
+ long phase2Bytes;
+ PageReadStore dataPages = null;
+ // An all-keys projection has nothing for phase 2 to read -- but only if
the key values were
+ // buffered. A row group past the cap has to read them here like any
other column.
+ 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. Degrading to a whole-block read is not an option,
because the key vectors
+ // hold only the survivors and the batch would misalign, and neither
is dropping the
+ // predicate, which extraction has removed from the post-scan Filter.
So the read fails,
+ // and all this adds is what the user can do about it.
+ //
+ // Nothing is checked up front: a row group the filter keeps whole
never needs the index
+ // (`readFilteredRowGroup` falls back to a plain read when the ranges
cover the block), and
+ // one it rejects whole is never read at all, so a file with no page
index still scans as
+ // long as the filter never has to prune inside a row group.
+ try {
+ dataPages = lateMatReader.readFilteredRowGroup(blockIdx,
finalRanges);
+ } catch (MissingOffsetIndexException e) {
+ throw new UnsupportedFileReadException(String.format(
+ "Storage-filter pushdown needs a Parquet offset index to read
the %d of %d rows its "
+ + "filter kept in row group %d of %s, but the file was
written without a page "
+ + "index. Set %s=false to read this file.",
+ finalRowCount, blockRowCount, blockIdx, lateMatReader.getFile(),
+
SQLConf$.MODULE$.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED().key()), e);
+ }
+ 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 UnsupportedFileReadException(
+ "No data pages for row group " + blockIdx + " despite " +
finalRowCount
+ + " surviving rows");
+ }
+ keptRows = dataPages.getRowCount();
+ // `needBytes`, not just `bytesAvoidedPf != null`: it is what built
`blockChunks`.
+ if (needBytes && bytesAvoidedPf != null) {
+ 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 --
+ // which can make the row group's contribution negative, and that
is the truth about 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) {
+ 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;
+ }
+
+
+ /**
+ * 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.
+ * </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.
+ */
+ 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();
+ // 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);
Review Comment:
Confirmed in the code: `ParquetReadState.constructRanges` materializes a
`List<RowRange>` per leaf per row group, so scattered survivors cost about 40
bytes a range times the leaf count, and phase 2 drives one reader per leaf.
It is counted now and weighed against the same cap as the survivor buffer,
in two rungs: drop the buffer first, and give the filter up for that row group
if the ranges alone still do not fit. There is a second check right after phase
1, for a row group whose survivors fit in a single accumulator and are
therefore never weighed inside the loop.
The better answer is a follow-up in the description: coarsen the ranges
instead of giving the filter up. Closing a gap shorter than the smallest page
row count of the columns phase 2 reads costs no extra bytes, since no page fits
wholly inside such a gap. c255f88
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -492,6 +938,535 @@ 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 (non-key schema): read non-key columns restricted to {@code
finalRanges}.
+ * Skipped entirely when {@link #nonKeyColumns} is null (all-keys
projection).
+ *
+ * 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 and gives up past the cap. This row group is then read the
plain way: phase 2 takes
+ // every projected column under finalRanges, key columns included, so
nothing is buffered and
+ // the cost is one extra read of the key columns.
+ spliceCurrentRowGroup = true;
+ 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 -- the documented escape hatch for files whose column
index is wrong.
+ // Trusting a wrong column index here drops rows for good, since
finalRanges is a subset of
+ // these ranges and the post-scan Filter no longer holds the predicate.
Phase 2 is
+ // unaffected: it selects pages through the offset index, which 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.
+ 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(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.
+ 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 UnsupportedFileReadException(
+ "No key pages for row group " + blockIdx + " despite " +
baselineRows
+ + " rows selected by the pushed filter");
+ }
+ RowRanges finalRanges = evaluateStorageFilter(keyPages,
pushedFilterRanges);
+ long finalRowCount = finalRanges.rowCount();
+
+ if (finalRowCount == 0) {
+ // Every surviving row was rejected by the storage filter; skip 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 has to be subtracted from it
here.
+ SQLMetric rgSkipped = m.rowGroupsSkipped();
+ if (rgSkipped != null) rgSkipped.add(1L);
+ SQLMetric rowsExcludedRg = m.rowsExcludedByRowGroup();
+ if (rowsExcludedRg != null) rowsExcludedRg.add(baselineRows);
+ if (bytesAvoidedRg != null) bytesAvoidedRg.add(nonKeyBaselineBytes);
+ continue;
+ }
+
+ // Phase 2: switch to non-key schema, read non-key columns under
finalRanges. Skipped entirely
+ // when the projection is all keys (nonKeyColumns is null); emit
reconstructs each batch from
+ // the key queues alone.
+ long keptRows;
+ long phase2Bytes;
+ PageReadStore dataPages = null;
+ // An all-keys projection has nothing for phase 2 to read -- but only if
the key values were
+ // buffered. A row group past the cap has to read them here like any
other column.
+ 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. Degrading to a whole-block read is not an option,
because the key vectors
+ // hold only the survivors and the batch would misalign, and neither
is dropping the
+ // predicate, which extraction has removed from the post-scan Filter.
So the read fails,
+ // and all this adds is what the user can do about it.
+ //
+ // Nothing is checked up front: a row group the filter keeps whole
never needs the index
+ // (`readFilteredRowGroup` falls back to a plain read when the ranges
cover the block), and
+ // one it rejects whole is never read at all, so a file with no page
index still scans as
+ // long as the filter never has to prune inside a row group.
+ try {
+ dataPages = lateMatReader.readFilteredRowGroup(blockIdx,
finalRanges);
Review Comment:
Fixed by not letting it escape. The phase-2 read is wrapped, and on
`MissingOffsetIndexException` the filter is given up for that row group and the
read retried over `pushedFilterRanges`, which is what a plain scan reads. The
file is latched, so later row groups skip phase 1 rather than evaluate a filter
they cannot use.
So there is no exception left to be misleading, and nothing for
`ignoreCorruptFiles` to swallow. The conf doc says what it costs: the first row
group the reader tries it on also pays a key-column read, because a missing
index is only reported by attempting the read. c255f88
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -492,6 +938,535 @@ 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 (non-key schema): read non-key columns restricted to {@code
finalRanges}.
+ * Skipped entirely when {@link #nonKeyColumns} is null (all-keys
projection).
+ *
+ * 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 and gives up past the cap. This row group is then read the
plain way: phase 2 takes
+ // every projected column under finalRanges, key columns included, so
nothing is buffered and
+ // the cost is one extra read of the key columns.
+ spliceCurrentRowGroup = true;
+ 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 -- the documented escape hatch for files whose column
index is wrong.
+ // Trusting a wrong column index here drops rows for good, since
finalRanges is a subset of
+ // these ranges and the post-scan Filter no longer holds the predicate.
Phase 2 is
+ // unaffected: it selects pages through the offset index, which 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.
+ 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(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.
+ 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 UnsupportedFileReadException(
+ "No key pages for row group " + blockIdx + " despite " +
baselineRows
+ + " rows selected by the pushed filter");
+ }
+ RowRanges finalRanges = evaluateStorageFilter(keyPages,
pushedFilterRanges);
+ long finalRowCount = finalRanges.rowCount();
+
+ if (finalRowCount == 0) {
+ // Every surviving row was rejected by the storage filter; skip 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 has to be subtracted from it
here.
+ SQLMetric rgSkipped = m.rowGroupsSkipped();
+ if (rgSkipped != null) rgSkipped.add(1L);
+ SQLMetric rowsExcludedRg = m.rowsExcludedByRowGroup();
+ if (rowsExcludedRg != null) rowsExcludedRg.add(baselineRows);
+ if (bytesAvoidedRg != null) bytesAvoidedRg.add(nonKeyBaselineBytes);
+ continue;
+ }
+
+ // Phase 2: switch to non-key schema, read non-key columns under
finalRanges. Skipped entirely
+ // when the projection is all keys (nonKeyColumns is null); emit
reconstructs each batch from
+ // the key queues alone.
+ long keptRows;
+ long phase2Bytes;
+ PageReadStore dataPages = null;
+ // An all-keys projection has nothing for phase 2 to read -- but only if
the key values were
+ // buffered. A row group past the cap has to read them here like any
other column.
+ 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. Degrading to a whole-block read is not an option,
because the key vectors
+ // hold only the survivors and the batch would misalign, and neither
is dropping the
+ // predicate, which extraction has removed from the post-scan Filter.
So the read fails,
+ // and all this adds is what the user can do about it.
+ //
+ // Nothing is checked up front: a row group the filter keeps whole
never needs the index
+ // (`readFilteredRowGroup` falls back to a plain read when the ranges
cover the block), and
+ // one it rejects whole is never read at all, so a file with no page
index still scans as
+ // long as the filter never has to prune inside a row group.
+ try {
+ dataPages = lateMatReader.readFilteredRowGroup(blockIdx,
finalRanges);
+ } catch (MissingOffsetIndexException e) {
+ throw new UnsupportedFileReadException(String.format(
+ "Storage-filter pushdown needs a Parquet offset index to read
the %d of %d rows its "
+ + "filter kept in row group %d of %s, but the file was
written without a page "
+ + "index. Set %s=false to read this file.",
+ finalRowCount, blockRowCount, blockIdx, lateMatReader.getFile(),
+
SQLConf$.MODULE$.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED().key()), e);
+ }
+ 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 UnsupportedFileReadException(
+ "No data pages for row group " + blockIdx + " despite " +
finalRowCount
+ + " surviving rows");
+ }
+ keptRows = dataPages.getRowCount();
+ // `needBytes`, not just `bytesAvoidedPf != null`: it is what built
`blockChunks`.
+ if (needBytes && bytesAvoidedPf != null) {
+ 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 --
+ // which can make the row group's contribution negative, and that
is the truth about 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) {
+ bytesAvoidedPf.add(nonKeyBaselineBytes - phase2Bytes);
Review Comment:
Right, `SQLMetrics.scala` has `if (v >= 0)`. The second key read is now
charged into that row group's own phase-2 bytes, so the per-row-group number is
honest, and the comment says what the clamp then does: a row group that read
more than the baseline contributes nothing rather than subtracting.
A test asserts the charge, by running the same file and filter with the cap
high and low and requiring the low arm to report less avoided. c255f88
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala:
##########
@@ -230,6 +274,40 @@ class ParquetFileFormat
val int96RebaseModeInRead = parquetOptions.int96RebaseModeInRead
val archiveFormatEnabled = parquetOptions.archiveFormatEnabled
+ // Extraction has already removed these conjuncts from the post-scan
Filter, so nothing else in
+ // the plan would apply them: anything that stops the reader from honoring
them fails loudly
+ // rather than dropping them. `enableVectorizedReader` is one such thing,
and it is recomputed
+ // from the live session conf when the RDD is built, so a flip of
+ // spark.sql.parquet.enableVectorizedReader (or the nested-column variant)
after planning lands
+ // here.
+ val storageFilterOpt: Option[ParquetStorageFilter] = if
(storageFilters.isEmpty) {
+ None
+ } else if (!enableVectorizedReader) {
+ throw new UnsupportedFileReadException(
Review Comment:
Fixed: the messages name the expression by its node rather than printing it,
so a prepared bloom cannot render as megabytes of hex. The `require` in
`FileFormat.buildReaderWithStorageFilters` is gone with the model change, since
that default is now a working implementation rather than a guard. c255f88
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala:
##########
@@ -269,7 +269,15 @@ object DataSourceUtils extends PredicateHelper {
QueryExecutionErrors.sparkUpgradeInWritingDatesError(format, config)
}
+ /**
+ * Whether `ignoreCorruptFiles` may swallow this failure and skip the rest
of its file.
+ *
+ * [[UnsupportedFileReadException]] is excluded because it does not report a
corrupt file: it says
+ * the file cannot support a read the plan depends on. Skipping the rest of
such a file would drop
+ * rows that are perfectly readable, and would do it silently.
+ */
def shouldIgnoreCorruptFileException(e: Throwable): Boolean = e match {
+ case _: UnsupportedFileReadException => false
Review Comment:
Taken, and it is the biggest change in this round. Thank you for pushing on
it.
The conjunct is not moved out of the plan at all now, it is copied: it stays
in the post-scan `Filter` as well, so the reader is free to give a file or a
row group up with no correctness argument attached.
`UnsupportedFileReadException` and its case in the shared classifier are gone
with it, and so is the hard failure on a file written without a page index.
Two things settled it beyond the fallback you asked for. It is v1's own
taxonomy already, where only partition filters leave the plan because partition
pruning is exact, while the data filters handed to Parquet stay, and
`PushDownUtils` draws the same line from the other side in DSv2. And the bloom
is never built for nothing: the `Filter` uses it whatever the scan managed to
do, and both sides read the same bytes because the subquery is executed once.
The price is evaluating the conjunct a second time for the rows the scan
emits. That falls on what the filter kept rather than on what it discarded, and
anyone turning this conf on is doing it because the filter is selective, so it
is a small price. c255f88
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -492,6 +938,535 @@ 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 (non-key schema): read non-key columns restricted to {@code
finalRanges}.
+ * Skipped entirely when {@link #nonKeyColumns} is null (all-keys
projection).
+ *
+ * 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 and gives up past the cap. This row group is then read the
plain way: phase 2 takes
+ // every projected column under finalRanges, key columns included, so
nothing is buffered and
+ // the cost is one extra read of the key columns.
+ spliceCurrentRowGroup = true;
+ 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 -- the documented escape hatch for files whose column
index is wrong.
+ // Trusting a wrong column index here drops rows for good, since
finalRanges is a subset of
+ // these ranges and the post-scan Filter no longer holds the predicate.
Phase 2 is
+ // unaffected: it selects pages through the offset index, which 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.
+ 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(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.
+ 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 UnsupportedFileReadException(
+ "No key pages for row group " + blockIdx + " despite " +
baselineRows
+ + " rows selected by the pushed filter");
+ }
+ RowRanges finalRanges = evaluateStorageFilter(keyPages,
pushedFilterRanges);
+ long finalRowCount = finalRanges.rowCount();
+
+ if (finalRowCount == 0) {
+ // Every surviving row was rejected by the storage filter; skip 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 has to be subtracted from it
here.
+ SQLMetric rgSkipped = m.rowGroupsSkipped();
+ if (rgSkipped != null) rgSkipped.add(1L);
+ SQLMetric rowsExcludedRg = m.rowsExcludedByRowGroup();
+ if (rowsExcludedRg != null) rowsExcludedRg.add(baselineRows);
+ if (bytesAvoidedRg != null) bytesAvoidedRg.add(nonKeyBaselineBytes);
+ continue;
+ }
+
+ // Phase 2: switch to non-key schema, read non-key columns under
finalRanges. Skipped entirely
+ // when the projection is all keys (nonKeyColumns is null); emit
reconstructs each batch from
+ // the key queues alone.
+ long keptRows;
+ long phase2Bytes;
+ PageReadStore dataPages = null;
+ // An all-keys projection has nothing for phase 2 to read -- but only if
the key values were
+ // buffered. A row group past the cap has to read them here like any
other column.
+ if (nonKeyColumns == null && spliceCurrentRowGroup) {
Review Comment:
You are right, and the description was wrong. `clipParquetGroupFields` keeps
a field that is missing from the physical file, so `nonKeyFieldCount` is always
greater than zero once the planner rejects all-keys projections. Both the
description and the class comment are corrected.
The branch stays, since the reader should not depend on a planner rule to be
correct, and the comment now says what is true: nothing in production reaches
it, only a test that drives the reader directly. c255f88
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java:
##########
@@ -492,6 +938,535 @@ 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 (non-key schema): read non-key columns restricted to {@code
finalRanges}.
+ * Skipped entirely when {@link #nonKeyColumns} is null (all-keys
projection).
+ *
+ * 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 and gives up past the cap. This row group is then read the
plain way: phase 2 takes
+ // every projected column under finalRanges, key columns included, so
nothing is buffered and
+ // the cost is one extra read of the key columns.
+ spliceCurrentRowGroup = true;
+ 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 -- the documented escape hatch for files whose column
index is wrong.
+ // Trusting a wrong column index here drops rows for good, since
finalRanges is a subset of
+ // these ranges and the post-scan Filter no longer holds the predicate.
Phase 2 is
+ // unaffected: it selects pages through the offset index, which 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.
+ 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(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.
+ 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 UnsupportedFileReadException(
+ "No key pages for row group " + blockIdx + " despite " +
baselineRows
+ + " rows selected by the pushed filter");
+ }
+ RowRanges finalRanges = evaluateStorageFilter(keyPages,
pushedFilterRanges);
+ long finalRowCount = finalRanges.rowCount();
+
+ if (finalRowCount == 0) {
+ // Every surviving row was rejected by the storage filter; skip 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 has to be subtracted from it
here.
+ SQLMetric rgSkipped = m.rowGroupsSkipped();
+ if (rgSkipped != null) rgSkipped.add(1L);
+ SQLMetric rowsExcludedRg = m.rowsExcludedByRowGroup();
+ if (rowsExcludedRg != null) rowsExcludedRg.add(baselineRows);
+ if (bytesAvoidedRg != null) bytesAvoidedRg.add(nonKeyBaselineBytes);
+ continue;
+ }
+
+ // Phase 2: switch to non-key schema, read non-key columns under
finalRanges. Skipped entirely
+ // when the projection is all keys (nonKeyColumns is null); emit
reconstructs each batch from
+ // the key queues alone.
+ long keptRows;
+ long phase2Bytes;
+ PageReadStore dataPages = null;
+ // An all-keys projection has nothing for phase 2 to read -- but only if
the key values were
+ // buffered. A row group past the cap has to read them here like any
other column.
+ 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. Degrading to a whole-block read is not an option,
because the key vectors
+ // hold only the survivors and the batch would misalign, and neither
is dropping the
+ // predicate, which extraction has removed from the post-scan Filter.
So the read fails,
+ // and all this adds is what the user can do about it.
+ //
+ // Nothing is checked up front: a row group the filter keeps whole
never needs the index
+ // (`readFilteredRowGroup` falls back to a plain read when the ranges
cover the block), and
+ // one it rejects whole is never read at all, so a file with no page
index still scans as
+ // long as the filter never has to prune inside a row group.
+ try {
+ dataPages = lateMatReader.readFilteredRowGroup(blockIdx,
finalRanges);
+ } catch (MissingOffsetIndexException e) {
+ throw new UnsupportedFileReadException(String.format(
+ "Storage-filter pushdown needs a Parquet offset index to read
the %d of %d rows its "
+ + "filter kept in row group %d of %s, but the file was
written without a page "
+ + "index. Set %s=false to read this file.",
+ finalRowCount, blockRowCount, blockIdx, lateMatReader.getFile(),
+
SQLConf$.MODULE$.PARQUET_STORAGE_FILTER_PUSHDOWN_ENABLED().key()), e);
+ }
+ 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 UnsupportedFileReadException(
+ "No data pages for row group " + blockIdx + " despite " +
finalRowCount
+ + " surviving rows");
+ }
+ keptRows = dataPages.getRowCount();
+ // `needBytes`, not just `bytesAvoidedPf != null`: it is what built
`blockChunks`.
+ if (needBytes && bytesAvoidedPf != null) {
+ 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 --
+ // which can make the row group's contribution negative, and that
is the truth about 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) {
+ 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;
+ }
+
+
+ /**
+ * 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.
+ * </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.
+ */
+ 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();
+ // 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 (accumulate) {
+ accumulate = appendSurvivorRowToAccumulators(r);
+ }
+ }
+ }
+ remaining -= num;
+ }
+
+ if (accumulate) {
+ finalizePartialAccumulators();
+ }
+
+ return finalRangesBuilder.build();
+ }
+
+ private void ensureKeyScratchAllocated() {
+ if (keyScratchVectors != null) return;
+ // Assigned before the loop on purpose: an allocation failure part way
through then leaves the
+ // vectors allocated so far reachable for `close()`, which walks this
array element-wise.
+ keyScratchVectors = new WritableColumnVector[keyDescriptors.length];
+ boolean useOffHeap = MEMORY_MODE == MemoryMode.OFF_HEAP;
+ int[] keyIndices = storageFilter.keyColumnIndices();
+ for (int i = 0; i < keyDescriptors.length; i++) {
+ DataType dt = sparkRequestedSchema.fields()[keyIndices[i]].dataType();
+ keyScratchVectors[i] = useOffHeap
+ ? new OffHeapColumnVector(capacity, dt)
+ : new OnHeapColumnVector(capacity, dt);
+ }
+ keyScratchBatch = new ColumnarBatch(keyScratchVectors);
+ }
+
+ /**
+ * Allocates any accumulator slot left null by the last push to the queues,
{@link #capacity} rows
+ * each.
+ */
+ private void ensureCurrentKeyAccumulatorsAllocated() {
+ boolean useOffHeap = MEMORY_MODE == MemoryMode.OFF_HEAP;
+ int[] keyIndices = storageFilter.keyColumnIndices();
+ for (int i = 0; i < currentKeyAccumulators.length; i++) {
+ if (currentKeyAccumulators[i] == null) {
+ DataType dt = sparkRequestedSchema.fields()[keyIndices[i]].dataType();
+ currentKeyAccumulators[i] = useOffHeap
+ ? new OffHeapColumnVector(capacity, dt)
+ : new OnHeapColumnVector(capacity, dt);
+ }
+ }
+ currentKeyAccumulatorRowCount = 0;
+ }
+
+ /**
+ * Appends row {@code srcRow} of every key column to the accumulators,
pushing them onto their
+ * queues once full. All key columns advance in lockstep, which is what
keeps the queues aligned.
+ *
+ * <p>Returns false when the buffered survivors have passed
+ * {@code spark.sql.parquet.storageFilterPushdown.maxSplicedRowGroupBytes}
and this row group has
+ * given splicing up, in which case it has already released what it held and
the caller must stop
+ * calling this.
+ */
+ private boolean appendSurvivorRowToAccumulators(int srcRow) {
+ final int dstRow = currentKeyAccumulatorRowCount;
+ final WritableColumnVector[] accs = currentKeyAccumulators;
+ final WritableColumnVector[] srcs = keyScratchVectors;
+ final ValueCopier[] copiers = keyCopiers;
+ long valueBytes = 0L;
+ for (int i = 0, n = accs.length; i < n; i++) {
+ WritableColumnVector src = srcs[i];
+ WritableColumnVector dst = accs[i];
+ if (src.isNullAt(srcRow)) {
+ dst.putNull(dstRow);
+ } else {
+ copiers[i].copy(dst, dstRow, src, srcRow);
+ // Measured on the destination: a dictionary-encoded source has no
length of its own, since
+ // its values are read through the dictionary.
+ if (keyVariableLength[i]) valueBytes += dst.getArrayLength(dstRow);
+ }
+ }
+ splicedBytes += keyFixedBytesPerRow + valueBytes;
Review Comment:
Follow-up, now listed in the description as accounting the buffer through a
`MemoryConsumer` whose `spill()` gives splicing up.
The 2x you describe is acknowledged rather than hidden: the conf doc says
what is counted, the buffered values and their per-row overhead, not the
backing arrays a column vector may grow beyond that.
--
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]
