peter-toth opened a new pull request, #57760:
URL: https://github.com/apache/spark/pull/57760
### What changes were proposed in this pull request?
Follow-up to #57357. `DataSourceV2Strategy` no longer routes a
non-deterministic post-scan filter into `BatchScanExec.runtimeFilters`, so such
a filter is not pushed to a `SupportsRuntimeV2Filtering` scan at runtime. The
migration guide entry added by #57357 is widened to cover the runtime path.
#57357 guarded the `SupportsPushDownV2Filters` branch of
`PushDownUtils.pushFilters`, i.e. pushdown at query compilation. It did not
touch the runtime filter path: `DataSourceV2Strategy` routes scalar subquery
filters on runtime-filterable columns into `runtimeFilters` (SPARK-56467), and
`PushDownUtils.pushRuntimeFilters` translates them with
`DataSourceV2Strategy.translateScalarSubqueryFilterV2`, which has no
determinism guard.
### Why are the changes needed?
`V2ExpressionBuilder` translates `Rand`, so a non-deterministic runtime
filter reaches the data source. On master, against a
`SupportsRuntimeV2Filtering` in-memory table:
SELECT * FROM tbl WHERE part = (SELECT max(val) FROM dim) OR rand() < 0.5
pushedPredicates: (part = 3) OR (RAND() < 0.5)
plan: Filter ((part#38 = Subquery subquery#36) OR
(rand(-6129082941936456980) < 0.5))
+- BatchScan ...
This is the problem #57357 describes, and here it is unconditional rather
than possible. A scalar subquery runtime filter is deliberately kept in
`postScanFilters` as well ("These filters stay in postScanFilters for
correctness"), so the predicate is *always* evaluated twice: once by the source
to prune input partitions, once by the `FilterExec` above the scan. The two
evaluations of `rand()` disagree, and a partition the source dropped on its own
evaluation is gone -- rows that Spark's evaluation would have kept cannot be
recovered.
Gating the routing rather than the translation keeps the non-deterministic
filter out of `runtimeFilters` entirely, so it also stays out of
`BatchScanExec`'s `equals`/`doCanonicalize`. Dynamic partition pruning filters
are unaffected: they are not routed through this branch, and they are removed
from `postScanFilters`, so they are never evaluated twice.
### Does this PR introduce _any_ user-facing change?
No, beyond what #57357 already documented for unreleased 4.3. Data sources
implementing `SupportsRuntimeV2Filtering` no longer receive non-deterministic
runtime filters through `filter`; the filter is still evaluated by Spark after
the scan, as it already was.
### How was this patch tested?
Added `SPARK-58207: non-deterministic scalar subquery filters are not pushed
into runtimeFilters` to `DataSourceV2SQLSuiteV2Filter`, next to the SPARK-56467
test it mirrors: asserts `runtimeFilters` is empty, that no partition is
pruned, and that the non-deterministic filter is still evaluated above the
scan. It fails on master (`runtimeFilters` holds `(part = subquery) OR
(rand(...) < 0.5)`) and passes with this change.
Ran `DataSourceV2SQLSuiteV2Filter`,
`DataSourceV2EnhancedRuntimePartitionFilterSuite` (iterative
`PartitionPredicate` second pass) and `DataSourceV2Suite` locally, 243 tests
green, plus `dev/lint-scala`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 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]
