peter-toth opened a new pull request, #58895: URL: https://github.com/apache/spark/pull/58895
### What changes were proposed in this pull request? This adds a late-materialization read path to the vectorized Parquet reader, so the optimizer can push a runtime filter into the scan and have it prune value-column IO instead of running as a post-scan `FilterExec`. The filter it pushes today is `BloomFilterMightContain`, the runtime bloom `InjectRuntimeFilter` builds for a join. The reader reads the filter's key columns first, evaluates the predicate per row, and then reads the remaining columns restricted to the surviving row ranges. Each output batch is spliced together from the key vectors it kept and the value vectors it read for those rows. **Planning.** - New SQL conf `spark.sql.parquet.storageFilterPushdown.enabled`, default `false`, session-bound. It is a planning-time decision only. With it off, no storage filter is attached to a scan in the first place and the bloom stays where it is today. - `FileSourceStrategy.extractStorageFilters` lifts eligible top-level `BloomFilterMightContain` conjuncts out of `afterScanFilters` into a new `storageFilters` slot on `FileSourceScanExec`. Eligibility is checked in full at planning time, because extraction removes the conjunct from the post-scan `Filter` and nothing else would apply it afterwards. A conjunct qualifies when the file format is exactly `ParquetFileFormat`, the vectorized reader is feasible for `partitionSchema ++ outputDataSchema`, the conjunct is deterministic, and every reference on the bloom's value side is a projected data column of a type the reader can copy. - `ParquetStorageFilter.isSupportedKeyType` is the single authority on key-column types. Both `extractStorageFilters` and `ParquetStorageFilter.create` consult it, and it lists exactly the types `VectorizedParquetRecordReader.copierFor` handles. It is deliberately narrower than `AtomicType`, since `VariantType`, `GeometryType` and `GeographyType` are atomic but have no primitive Parquet leaf for phase 1 to read into. - `FileSourceScanLike` gains `storageFilters: Seq[Expression]` and five SQL metrics, described below. The `StorageFilters` entry in the scan description is only emitted when the scan has storage filters, so explain output is unchanged for everyone else. - `FileSourceScanExec.preparedStorageFilters` materializes scalar subqueries and binds attributes to `BoundReference`s indexing `requiredSchema`. The conf is deliberately not rechecked at execution time. Once extraction has dropped a bloom from the post-scan `Filter`, the runtime has to honour that decision or fail, and it does the second. **`FileFormat` API.** - A new `buildReaderWithStorageFilters` overload takes `storageFilters: Seq[Expression]` and the metric map. Its default body requires `storageFilters` to be empty and otherwise delegates to `buildReaderWithPartitionValues`, so a format that does not implement the feature rejects a filter it cannot honour rather than dropping it. - `FileSourceScanExec.inputRDD` only routes through the new entry point when there is something to push, so a `ParquetFileFormat` subclass that customizes reading by overriding `buildReaderWithPartitionValues` keeps working unchanged. **Parquet reader.** - A new `ParquetStorageFilter` value object holds the bound expressions, the key-column indices into the requested schema, and the optional metrics. `rewriteForMissingKeys` and `evalAllMissing` handle schema evolution, where a key column is in the requested schema but absent from the physical file. The rewritten predicate is evaluated against the value the reader will actually materialize for that column, which is the column's existence `DEFAULT` when it has one and null otherwise. Evaluating against null instead would filter on a value the scan never returns, and `XxHash64` is `nullable = false` so a null input hashes to the seed rather than producing null. - `SpecificParquetRecordReaderBase` exposes the underlying `ParquetFileReader`, the input file and the footer, so the late-materialization driver can switch the requested schema per phase and call `readFilteredRowGroup(blockIdx, rowRanges)`. - `VectorizedParquetRecordReader` gains a three-phase per-row-group loop driven by that one reader. Phase 0 computes `pushedFilterRanges` from the pushed data filter through the column index, which is metadata only. Phase 1 switches to the key-only schema, reads the key columns under those ranges, evaluates the storage filter per row, and accumulates the survivors into per-key-column queues of capacity-sized `WritableColumnVector`s. Phase 2 switches to the non-key schema and reads those columns under the surviving ranges. Emit splices the dequeued key vectors with the freshly read non-key vectors in the projection's own order. - A projection that is all key columns skips phase 2 entirely and reconstructs every batch from the key queues, which is the shape with the largest saving. - Phase 0 checks `parquet.filter.columnindex.enabled` itself, because `ParquetFileReader.getRowRanges` only asks whether a filter is pushed. That conf is the documented escape hatch for a file whose column index is wrong, and trusting a wrong column index here would drop rows for good. - `requireOffsetIndexesForPhase2` fails at reader init if any projected column of any row group has no offset index. Phase 2 reads a strict subset of a row group's rows, which parquet can only do through the offset index, and neither widening the read nor skipping the filter is correct. The check reads only footer fields, and it covers every projected column rather than the non-key ones alone, because parquet builds one column index store per row group and returns an empty one as soon as any path in it lacks an offset index. - Per-key-column value copying uses a `ValueCopier` chosen once at init, so the survivor loop has no per-value type dispatch. - `initBatch` threads a `skipDataSlots` set through `allocateColumns`, so the persistent output vectors for key slots are never allocated. Under splicing those slots come from the queues. - Preconditions that planning already guarantees throw instead of falling back, for the same reason the conf is not rechecked. That covers a key ordinal out of range, a non-primitive key column, missing key pages, and a vectorized-reader conf flipped between planning and execution. The two remaining silent fallbacks cannot change the result, namely a reader with no underlying `ParquetFileReader`, which only test mocks produce, and a file where every key column is missing, which is answered by evaluating the rewritten constant predicate. **Metrics.** Five, all created only when the scan has storage filters, and all scoped to what the storage filter added on top of a no-storage-filter read of the same projection. Each counter names its own quantity, so the three verbs are deliberate. A row group is skipped, meaning its data columns were never read while phase 1 did read its key columns. A row is excluded, meaning it never reached the output. A byte is avoided, meaning it was never transferred. - `storageFilterRowGroupsSkipped`, "row groups skipped by storage filter". - `storageFilterRowsExcludedByRowGroup`, "rows excluded by storage filter (whole row group)". - `storageFilterRowsExcludedWithinRowGroup`, "rows excluded by storage filter (within row group)". The suffix says where the row was excluded rather than by which mechanism, because a row sharing a page with a survivor is read and dropped during decode. - `storageFilterBytesAvoidedByRowGroup`, "bytes avoided by storage filter (whole row group)". - `storageFilterBytesAvoidedByPageFiltering`, "bytes avoided by storage filter (page filtering)". The byte counters must not cost IO to report, so `compressedBytesForRowRanges` answers from the footer's `ColumnChunkMetaData.getTotalSize()` whenever the row range covers the whole block, and walks the offset index only for a strict subset. That is complete rather than a mitigation. A range narrower than the block can only come from column-index filtering, which builds and memoizes the store as a side effect, and phase 2's own read builds it before the walk in the other case. The walk counts the dictionary page too, since parquet reads it whenever it reads any data page of a chunk. ### Why are the changes needed? A runtime bloom filter from join runtime filtering is applied as a post-scan `FilterExec` today. The scan still reads every value page of every row group, even where the bloom drops almost every row immediately. On a selective join over a wide table that read is the dominant cost. Late materialization turns that around. The scan reads the bloom's key column first, decides which rows survive, and never reads the value pages no surviving row touches. A row group where nothing survives costs one key-column read and no value IO at all. ### Does this PR introduce _any_ user-facing change? No change by default, since the conf is off and nothing is attached to a scan in that case. With the conf on: - Queries return the same rows. The late-materialization path is exact, and a filter that fails any eligibility check keeps its existing post-scan `FilterExec`. - `FileSourceScanExec` reports the five metrics above in the SQL UI, and its description gains a `StorageFilters` entry. - A scan reading a Parquet file written without offset indexes fails with an error naming the conf to turn off. Files written by parquet-mr 1.11 and later always have them. - A consumer that illegally retains a `ColumnarBatch` across `next()` sees a sharper edge than before. On the plain path the previous batch's key vectors are reused, and under splicing they are freed at the next `nextBatch()`, so with off-heap vectors the retained reference points at released memory rather than stale values. The contract already forbids retaining a batch. - Phase 1 buffers a whole row group's surviving key values before it produces that row group's first batch, so a task holds up to one extra copy of the key columns for one row group. The conf's documentation says so. ### How was this patch tested? New `ParquetStorageFilterSuite`, 89 tests. Four blind review rounds ran over the change, each by an agent with no knowledge of the earlier ones, and every finding is fixed in this commit. - Reader-level tests over hand-built filters: whole row group rejected, nothing rejected, mixed, multi-batch emit, key-only projection, a survivor count that is an exact multiple of the batch capacity, and a projection whose key column is not in the leading slot. - One case per `ValueCopier` branch across 17 types and both encodings, each comparing the splicing path against the plain reader over the same file, so values are checked and not only row counts. `isSupportedKeyType` is asserted to cover exactly the copier's types. - Nullable keys, two key columns, partition columns alongside spliced keys, off-heap vectors, the row-at-a-time path, `_metadata.row_index`, and a complex non-key column. - Schema evolution: `rewriteForMissingKeys` and `evalAllMissing` as units, plus end-to-end reads where a key column is missing from the older file with a `DEFAULT`, with a `DEFAULT` that fails the filter, and with no `DEFAULT`. - Page-level ranges, which need a multi-page row group: phase 1 stays aligned with the row indexes, and the byte metrics stay non-negative. - Metric arithmetic: emitted plus excluded accounts for every row of the file, and an all-key projection reports zero avoided bytes. - Planner gates: the bloom stays in the post-scan `Filter` when the vectorized reader is unavailable, when the conf is off, and when the conjunct is non-deterministic. A scan whose vectorized reader is disabled after planning fails loudly. Extraction preserves results with AQE on and off. Canonicalization keeps a storage-filter scan distinct from a plain one, so exchange and subquery reuse cannot cross them. - Whole-stage codegen off, where the planner's gate is weaker than the runtime's, so the bloom is extracted and the spliced batch is served one row at a time. - Column-index filtering off, the branch that decides where phase 0's ranges come from. The row accounting tells the two arms apart, since with the column index off every row of the block reaches phase 1. - A limit under whole-stage codegen, which is the only shape that closes the spliced batch from outside while the reader is still open. The test asserts the generated source contains that close, so it cannot pass for the wrong reason. - The byte metrics cost no extra IO, measured rather than argued. The same read runs twice over a filesystem that counts every byte handed back, once with all five metrics wired and once with none, and the counts match. Both range shapes are covered, the offset-index one and the footer one. Knowingly untested, both for the same reason. The offset-index check's throw needs a file with no offset index, which parquet-mr cannot write. The `ParquetFileFormat`-subclass bypass needs a third-party subclass. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code with Claude Opus 4.7 and Claude Opus 5 Co-authored-by: Matt Butrovich <[email protected]> -- 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]
