[SPARK-58207][SQL] Skip pushdown for nondeterministic V2 filters

[peter-toth]

What changes were proposed in this pull request?

By default, the SupportsPushDownV2Filters branch of PushDownUtils.pushFilters now pushes only deterministic filters to the data source; nondeterministic filters are kept as post-scan filters and evaluated by Spark after the scan. This mirrors the fix made for SupportsPushDownCatalystFilters in #57235 (SPARK-58112).

Because some sources fully enforce a pushed predicate (e.g. JDBC evaluates it in the database, so pushing it is sound), this is a behavior change. A legacy config, spark.sql.legacy.allowNonDeterministicV2FilterPushDown (default false), restores the previous behavior of pushing nondeterministic filters down.

Note there is an alternative, documentation-only approach discussed in the comment below (documenting a "fully accept or fully decline" contract on SupportsPushDownV2Filters instead of changing the framework). This PR takes the code approach with a legacy escape hatch; feedback on the direction is welcome.

Why are the changes needed?

Before this change the SupportsPushDownV2Filters branch translated and pushed every conjunct, and V2ExpressionBuilder translates Rand, so a predicate such as rand() > 0.5 was pushed to a V2 source.

Pushing a nondeterministic predicate is unsound in general: the data source may evaluate it a different number of times or at a different point than Spark, and a partial push — a predicate used for pruning yet also returned for post-scan re-evaluation (e.g. a parquet row group filter, an explicitly documented mode of SupportsPushDownV2Filters#pushedPredicates) — evaluates the predicate twice with different results, which can change query results.

This is the same class of problem that #57235 (SPARK-58112) fixed for SupportsPushDownCatalystFilters. The other pushdown paths already avoid it:

• SupportsPushDownFilters (V1 sources.Filter): nondeterministic predicates are untranslatable to sources.Filter, so they fall through to post-scan.
• the iterative PartitionPredicate second pass: guarded by PushDownUtils.isPushablePartitionFilter (f.deterministic).

Only the SupportsPushDownV2Filters first-pass push was left unguarded; this closes that gap so all pushdown paths handle nondeterministic filters consistently by default.

Does this PR introduce any user-facing change?

Yes. By default, data sources implementing SupportsPushDownV2Filters no longer receive nondeterministic filters through pushPredicates; those filters remain in Spark as post-scan filters. For a source that fully enforced such a predicate (e.g. JDBC), the predicate is now evaluated by Spark instead of the source. Set spark.sql.legacy.allowNonDeterministicV2FilterPushDown to true to restore the previous behavior. Documented in the SQL migration guide.

How was this patch tested?

• Added a regression test in DataSourceV2Suite ("SPARK-58207: V2 filter pushdown skips non-deterministic filters") using AdvancedDataSourceV2WithV2Filter: asserts that rand() > 0.5 is not pushed to the source and is retained as a post-scan filter (fails on master, passes with this change).
• Updated the JDBCV2Suite RAND(1) < bonus pushdown test to run with the legacy config both on and off: on, RAND(1) < BONUS is pushed to H2 (as before); off, it stays as a post-scan filter.
• Ran DataSourceV2Suite and DataSourceV2EnhancedPartitionFilterSuite (the latter exercises the iterative PartitionPredicate second pass) locally, both green.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Opus 4.8)

[peter-toth]

cc @szehon-ho @cloud-fan @dongjoon-hyun — as the author/reviewers of #57235 (SPARK-58112).

Non-deterministic predicate handling looks like a gap in the SupportsPushDownV2Filters path as well. #57235 guarded SupportsPushDownCatalystFilters, but the V2 Predicate branch of PushDownUtils.pushFilters still translates and pushes non-deterministic predicates — V2ExpressionBuilder translates Rand, so rand() > 0.5 reaches the source. The V1 sources.Filter path avoids this (nondeterministic predicates are untranslatable), and the iterative PartitionPredicate second pass already guards via isPushablePartitionFilter, so this branch is the only one left unguarded.

The gap can be closed either way:

• by this PR — guard the branch and keep nondeterministic filters post-scan, the same approach as [SPARK-58112][SQL] Skip pushdown for nondeterministic Catalyst filters #57235; or
• by documenting a "fully accept or fully decline" rule on SupportsPushDownV2Filters: a source must never push a nondeterministic predicate partially (prune with it and also return it for post-scan re-check, e.g. as a row group filter), since that evaluates it twice with different results.

WDYT — which direction do you prefer for closing it?

[cloud-fan]

0 blocking, 0 non-blocking, 0 nits.
No findings; the change closes the V2 pushdown gap while preserving existing deterministic and residual-filter paths.

Verification

Traced accepted, connector-rejected, untranslatable, nondeterministic, and iterative-partition paths through PushDownUtils.pushFilters; compared the nondeterministic behavior with SupportsPushDownCatalystFilters; confirmed the test observes both connector input and residual plan filtering. Tests were not run as part of this review.

[peter-toth]

Thanks for the review @cloud-fan!

Since not pushing non-deterministic filters is a behavior change — some sources were allowed to evaluate some non-deterministic predicates — I added a legacy flag spark.sql.legacy.allowNonDeterministicV2FilterPushDown (default off) to restore the old behavior if needed. I also use that flag in the H2 RAND(1) < bonus pushdown test so it runs both ways and stays green.

It's in a separate follow-up commit on top of the one you approved.