peter-toth opened a new pull request, #59715:
URL: https://github.com/apache/spark/pull/59715
### What changes were proposed in this pull request?
A follow-up to SPARK-59620, which lets the V1 Parquet reader apply a join's
runtime bloom filter while it reads. This PR gives data source V2 scans the
same. A connector such as Iceberg can then read the bloom's key columns first,
and skip the other columns of the rows the bloom rejects.
The API, all `@Evolving`:
- `SupportsStorageFilters`, a `Scan` mix-in. `storageFilterAttributes()`
names the columns the scan's readers can evaluate a storage filter on.
- `StorageFilter`, an opaque predicate that Spark implements. It has
`references()`, `eval(InternalRow)` and `int select(ColumnarBatch, int[]
selected)`, which writes the positions of the rows it keeps. The default
`select` loops over `eval`.
- `StorageFilterNotAppliedException`, error condition
`STORAGE_FILTER_NOT_APPLIED`. `eval` and `select` throw it when the filter
cannot be applied to a row.
- `StorageFilterRequest`, with `filter()` and `appendsCheckedColumn()`.
- `PartitionReaderFactory.createReader(partition, request)` and
`createColumnarReader(partition, request)`. The factory of a scan that declares
storage-filter columns must implement the one for its read mode.
- `Scan.toBatch` now says it may be called more than once for a scan offered
a storage filter, as for runtime filtering.
Planning and execution:
- `DataSourceV2Strategy` offers a top-level runtime bloom over declared
columns, under the new `spark.sql.sources.v2.storageFilterPushdown.enabled`.
The per-conjunct rules are shared with `FileSourceStrategy` through the new
`StorageFilterPlanning`.
- The bloom stays in the post-scan `Filter`. Where whole-stage codegen is on
and the scan's output still fits it with the column, the scan gets a boolean
checked column last, and the `Filter` evaluates the bloom as `IF(checked, true,
bloom)`. A `Project` drops the column.
- `BatchScanExec` ANDs the storage filters into one `StorageFilter`, and
`DataSourceRDD` hands each task's readers a `StorageFilterRequest`. Each task
deserializes its own filter.
The design decisions:
- **Applying the filter is optional per unit,** such as a file or a row
group. Spark keeps the predicate above the scan. A reader that cannot use it on
a unit reads that unit as a plain read does, and marks its rows unchecked. When
the filter throws `StorageFilterNotAppliedException`, the reader reads the rest
of the unit that way. Spark's filter throws it for an error evaluating it, but
not for a task kill or a fatal error.
- **The checked column says which rows the reader filtered.** It is
requested in the task, through the request, since the scan's read schema is
fixed before `InjectRuntimeFilter` adds the bloom. It is requested only where
dropping it copies no extra row. A columnar scan out of whole-stage codegen
would copy every row in the `Project` that drops it. A row-based V2 scan has
that `Project` anyway, for the UnsafeRow conversion.
- **The filter is delivered in the task, not at planning.** The reader
factory does not carry it. The same API can later hand each partition its own
bloom, for storage-partitioned joins.
- **One combined filter per scan, opaque to the connector.** Spark
implements it, so a connector does not reproduce Spark's bloom format or hash.
Only runtime blooms are offered for now. Offering more is a planner change.
- **`select` returns positions, not a bitmap.** Iceberg's Spark reader
already wraps a filtered batch as `ColumnVectorWithFilter(delegate, int[]
rowIdMapping)`.
`BatchScanExec` gets a `storageFilters` field, so the positional patterns
over it in eight test files gain one `_`.
### Why are the changes needed?
SPARK-59620 saves IO only in Spark's own Parquet reader. A table read
through a V2 connector, such as Iceberg, still reads every projected column of
the rows a runtime bloom then drops. With this API a connector can do what the
V1 reader does. The Iceberg side is a follow-up in that project.
### Does this PR introduce _any_ user-facing change?
Yes:
- a new public connector API, listed above;
- with `spark.sql.sources.v2.storageFilterPushdown.enabled` on, which is off
by default:
- a V2 scan that implements it lists the offered filters under
`StorageFilters` in its description and in `EXPLAIN FORMATTED`;
- its output gains the checked column, which a `Project` drops;
- the post-scan `Filter` skips the bloom on the rows the reader marks
checked.
Nothing changes for a scan that does not implement `SupportsStorageFilters`.
### How was this patch tested?
New `StorageFilterPushDownSuite`, over a V2 test source whose reader applies
the filter row by row or through `select`:
- a row-based and a columnar scan, with AQE on and off, each get the bloom
and the checked column;
- the `Filter` skips the bloom on the rows a reader marks checked, even rows
it did not filter;
- a reader that gives the filter up returns its rows unchecked, and the
`Filter` applies the bloom to them;
- a filter that throws `StorageFilterNotAppliedException` is given up, and
its rows come back unchecked;
- a scan too wide for whole-stage codegen with the column, or without
whole-stage codegen, gets the bloom without it;
- nothing is offered with the conf off, to a scan that declares no column or
only another one, or to a scan with a column named like the checked column;
- a non-deterministic conjunct is not offered;
- the filters show in the scan description and in `EXPLAIN FORMATTED`, and
keep a plain scan's canonicalized plan apart, while a twin with fresh exprIds
keeps the same result.
New `StorageFilterImplSuite`: the positions `select` writes, the not-applied
answer, that a task kill propagates, and the internal error for a column
outside the scan output. `ParquetStorageFilterSuite` passes over the shared
planning rules.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code with Claude Opus 5.5
--
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]
