HyukjinKwon opened a new pull request, #59929:
URL: https://github.com/apache/spark/pull/59929
### What changes were proposed in this pull request?
This PR adds filter pushdown to native data sources (SPARK-60101), as a
follow-up of #59424, which left predicate pushdown out. Instead of passing
filters to a library as expression trees, Spark passes the *domains* of the
columns that the filters constrain: for each column, the values that it can
have in the rows that match. A library compares them with the statistics of its
data to skip what cannot match, such as files, row groups or blocks.
**C interface.** `spark_datasource.h` gets a new struct, `SparkDsDomain`,
and a new optional function, `push_domains`. The function is appended at the
end of `SparkDsLibrary`, after `close_data_writer`. A library built against the
4.4 header has a smaller table, so Spark never calls it there, and
`SPARK_DS_ABI_VERSION` stays 1. A domain has:
- `path`: the index of the column in the schema to read, then the index of
the field in each struct, for a field of a struct;
- `null_allowed` and `nan_allowed`;
- `ranges`: a struct array of `low`, `high`, `low_inclusive` and
`high_inclusive`, with `NULL` for any other value. The bounds have the Arrow
type of the column in the schema to read, and a null bound means no bound. The
ranges are sorted and do not overlap, and a single value is a range whose
bounds are equal and inclusive.
The library sets `used` for the domains that it uses, and the plan shows the
filters of those domains as `PushedFilters`, and the domains as `PushedDomains`.
**Deriving domains.** `ColumnDomains` derives the domains from the V2
predicates that Data Source V2 pushes:
- comparisons of a column with a literal (`=`, `<>`, `<=>`, `<`, `<=`, `>`,
`>=`), `IN`, `IS NULL`, `IS NOT NULL` and `STARTS_WITH`;
- combined with `AND`, `OR` and `NOT`. `NOT` is pushed to the leaves, and an
`OR` keeps only the columns that both sides constrain.
The semantics are Spark's:
- comparisons with null are null, so `NOT (x IN (1, NULL))` matches no row;
- NaN is equal to NaN and greater than any other value, so it is a flag
(`nan_allowed`) rather than a bound;
- -0.0 is normalized to 0.0;
- strings compare by UTF-8 bytes, and only the binary collation is pushed;
- binary values compare as unsigned bytes.
A domain is a superset of the values of the matching rows. Spark keeps
evaluating all the filters after the scan, which is the second kind of
`SupportsPushDownV2Filters.pushedPredicates()`, as for Parquet row-group
filters. So a library can only read less, never return wrong results by
over-filtering. The existing `push_limit` and `prune_columns` semantics don't
change: Spark still doesn't push a limit with a filter, and still reads the
filtered columns.
**Other changes:**
- `ColumnarScanBuilder` implements `SupportsPushDownV2Filters`.
- `spark.sql.dataSource.native.filterPushdown.enabled` (default true) turns
the pushdown off, for example to check whether a library skips data that
matches.
### Why are the changes needed?
Without filter pushdown, a native scan reads all of its data even for
selective queries. The predicate design dropped from #59424 passed JSON
expression trees. That made every library implement Spark's expression
semantics: three-valued logic, partial use only under AND/OR, NaN order,
literal encodings. Each of these was easy to get wrong silently, as the review
threads on #59424 showed.
Domains move all of that into Spark, where it is implemented once and
tested. A library only checks ranges against its statistics, as Presto/Trino
(`TupleDomain`), DuckDB (`TableFilterSet`) and Velox (per-column filters) do
for their scans. Expression-tree pushdown, for example with Substrait, can be
added later as another function, for sources that evaluate whole expressions.
### Does this PR introduce _any_ user-facing change?
Yes, for native data sources, which are unreleased (4.4.0):
- a new optional function `push_domains` and struct `SparkDsDomain` in
`spark_datasource.h`;
- `PushedFilters` and `PushedDomains` in the plan of a native scan;
- a new configuration, `spark.sql.dataSource.native.filterPushdown.enabled`.
### How was this patch tested?
- `ColumnDomainsSuite` (new):
- The domains of comparisons, `IN`, null checks, `AND`/`OR`/`NOT`, NaN and
-0.0, strings, prefixes (including U+10FFFF and the surrogate gap), collations,
binary values, nested fields and other types.
- A check against Spark's own evaluation. For about 450 filters, a fixed
list of edge cases plus random combinations of comparisons over rows with NULL,
NaN, infinities, -0.0, extreme integers, non-BMP strings, U+FFFD and high-byte
binary values, every row for which Spark finds the filter true must be in the
domains.
- `NativeDataSourceSuite`, end to end with the C++ test library:
- The test library uses the domain of `id` to skip blocks of rows, and
records the domains that it gets.
- Checks the domains of ranges, of other columns, unused domains, NaN,
`LIKE 'n1%'`, a nested field, timestamps and an empty domain.
- Checks that filters on functions are not pushed, a collated column, a
read-only library, and the configuration turned off.
- A library built against a header without `push_domains`: Spark must not
call it, although it is set in the table past its end.
- The existing layout test covers the offsets of `SparkDsDomain`.
- `ColumnarDataSourceSuite`: the same plumbing with a JVM data source.
### Was this patch authored or co-authored using generative AI tooling?
Yes.
This pull request and its description were written by Isaac.
--
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]
