[SPARK-58207][SQL][FOLLOWUP] Skip runtime filter pushdown for nondeterministic filters

[peter-toth]

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.

Since #57357 shipped in 4.3 and this completes it, this should be merged to branch-4.3 as well as master and branch-4.x so the widened migration-guide bullet ("Since Spark 4.3") is accurate and 4.3 carries the whole fix. This is a correctness fix rather than a new feature, and the cherry-pick is clean on both branches (verified locally).

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 the runtime path is more exposed to it than the compilation path. A scalar subquery runtime filter is deliberately kept in postScanFilters as well ("These filters stay in postScanFilters for correctness"), so Spark always re-evaluates it above the scan while the source is also free to prune on it. When the source does, the two evaluations of rand() disagree, and a partition the source dropped 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 and out of EXPLAIN, and no filter() + planInputPartitions() round is wasted. Dynamic partition pruning filters need no such check: a DynamicPruningSubquery over a non-deterministic filtering plan is itself non-deterministic (PlanExpression.deterministic folds in plan.deterministic), so the Filter above the scan fails NodeWithOnlyDeterministicProjectAndFilter and CleanupDynamicPruningFilters has already rewritten it to TrueLiteral before planning.

Does this PR introduce any user-facing change?

Yes, in unreleased 4.3, extending what #57357 already documented. 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, so query results do not change for a source that ignored the pushed predicate; a source that pruned on it returns more rows than before, which is the point of the fix. The migration guide entry is updated.

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. It fails on master and passes with this change; the assertion that discriminates is runtimeFilters.isEmpty (on master runtimeFilters holds (part = subquery) OR (rand(...) < 0.5)). The other two assertions -- no partition pruned, filter still evaluated above the scan -- hold on master too and are there as sanity checks, not as evidence: InMemoryV2FilterBatchScan.filter acts only on a top-level =/IN predicate and ignores the OR, so no in-tree fixture prunes on a non-deterministic predicate today.

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)

[uros-b]

Thank you @peter-toth!