[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.

[szehon-ho]

Thanks for this @peter-toth ! I had some thoughts.

Currently jdbc connector uses SupportsPushDown[V2]Filters, so it should not get the non-deterministic version anyway? But agree there could be some proprietary DSV2 connector out there that depend on it getting pushed down.

So are we saying we want to support this feature fully (pushdown of non-deterministic filter)? If so, just wondering should we have a separate boolean API for each flavor

• SupportsPushDownFIlter.pushdownNonDeterministic
• SupportsPushDownV2Filter.pushdownNonDeterministic
• SupportsPushDownCatalystFilter.pushdownNonDeterministic

That makes more sense as the same Spark could be reading from multiple data source and one flag is too coarse. At the very least , we can make it an pushdownNonDeterministic() that default to false on SupportsPushDownCatalystFilter? (to prevent behavior change as is the goal here)

Or are we quite sure that not pushing down filter is always correct, then a sql conf make sense.

[szehon-ho]

this comment is a bit contradictory, maybe must => should.

as the flag is going to push them if the user chooses to

[peter-toth]

Thanks. Fixed in 76ebb97.

[szehon-ho]

Ah so the SupportsV2PushdownFilter is ugly, iiuc you are saying the first round translates non-deterministic filters, but second round PartiitonPredicate does not.

I think if we go the multi-API approach, we can be more fine grain in keeping the old behavior and having a path forward to fix the new behavior.

1. SupportsPushDownFIlter.pushdownNonDeterministic (default false)
2. SupportsPushDownV2Filter.pushdownNonDeterministic (default true)
3. SupportsPushDownCatalystFilter.pushdownNonDeterministic (default true)

We can just leave out the second round PartitionPredicate and say that never pushdown non-deterministic filter (unless there is a need later), wdyt?

[peter-toth]

Thanks @szehon-ho! We should be really careful with non-deterministic expressions — the optimizer never moves or duplicates them on purpose. So IMO allowing them to be pushed down at all was a mistake. Pushing a non-det filter into a scan is a move, and if a source prunes with it and returns it for post-scan re-check, it's also a duplicate.

As I see the three pushdown paths:

• SupportsPushDownFilters (V1): never pushed non-det — sources.Filter can only carry a pushable column vs a literal, so rand() isn't even translatable. Nothing to toggle.
• SupportsPushDownCatalystFilters: pushed non-det until SPARK-58112 fixed it. But this is an internal interface (o.a.s.sql.internal.connector), not publicly available, and only FileScanBuilder implements it — and built-in file sources default to V1 anyway (spark.sql.sources.useV1SourceList). So ~no blast radius; correct as-is, no flag needed.
• SupportsPushDownV2Filters: the only path with real, tested non-det pushdown (the H2 RAND(1) < bonus case in JDBCV2Suite), so it's the only one that needs care as a behavior change.

[cloud-fan]

0 addressed, 1 remaining, 0 new. (0 newly introduced, 0 late catches.)
1 blocking, 0 non-blocking, 0 nits.

Remaining from prior review (1)

• The legacy compatibility control is session-wide, so enabling it for one fully enforcing connector also re-enables nondeterministic pushdown for every other V2 connector in the session; the existing discussion recommends a connector-level capability instead. -- existing thread

Verification

Traced the V2 accepted, rejected, untranslatable, residual, and legacy-config paths through PushDownUtils.pushFilters; compared the default behavior with SupportsPushDownCatalystFilters; inspected the V2 test connector and both new regression-test branches. The workflow's contract and text scanners completed with no additional findings. Tests were not run as part of this review.

[peter-toth]

Thanks @cloud-fan. I don't think @szehon-ho was suggesting a connector-level enablement as the resolution — he raised it conditionally and seemed fine with a SQL conf otherwise. My own view is a bit stronger: allowing non-det pushdown at all was a mistake, so the default should just be to not push.

That said, if a per-connector control is preferred over the session-wide flag, two options both fix the "too coarse" concern:

(a) Drop the config and add a capability, e.g. default boolean pushDownNondeterministicPredicates() (default false) on SupportsPushDownV2Filters. Though I doubt any real connector wants non-det pushdown, so a new public API feels like overkill.

(b) Keep a legacy config but make it a connector allowlist like spark.sql.sources.useV1SourceList (default empty), keyed on the source class name.

[dongjoon-hyun]

+1, LGTM.

[szehon-ho]

Drop the config and add a capability, e.g. default boolean pushDownNondeterministicPredicates() (default false) on SupportsPushDownV2Filters. Though I doubt any real connector wants non-det pushdown, so a new public API feels like overkill.

Yea that's what i had in mind, though it is still a public API? I just see the session-wide conf as too wide, if the goal is to ensure no behavior change for a specific DSV2 connector. I do realize its ugly, but Im leaning towards that at the moment. The whitelist seems more confusing to configure, but maybe there is precedent. Cc @aokolnychyi , will try to talk to him today about it.

[aokolnychyi]

This does not make sense. This can suddenly break all connectors.

[aokolnychyi]

If we absolutely need this, then some method on the SupportsPushDownV2Filters API would be OKish. That said, are we sure it is always safe to do, even in case of JDBC?

[aokolnychyi]

Each connector must decide this.

[aokolnychyi]

What if the filter is used in multiple places and then it gets pushed in one connector scan and not the other? How do we ensure the pushed filter behaves exactly the same as the one we keep on the Spark side? Is it even safe to push down undeterministic conditions into any connectors?