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

[peter-toth]

@aokolnychyi on "is it even safe to push a non-det condition into any connector?" — as I mentioned above, I think the answer is no, and not only because of partial-accept double-evaluation. Even if we (a) document a fully-accept-or-deny rule so nothing is evaluated twice, and (b) push the seed so the RNG sequence is identical, a full push can still change results.

Consider a filter with a translatable non-det predicate (say rand() > 0.5) next to a non-translatable one (say monotonically_increasing_id() < 100). Only the first gets pushed: it's forced down to the scan and evaluated first, over every scanned row, while the untranslatable one stays post-scan, over the survivors. Splitting and reordering two non-det predicates this way changes which rows the pushed one runs over, so the final result (and the semantics of the query) can differ. It's the same move the optimizer refuses to make elsewhere (CombineFilters, PushPredicateThroughNonJoin, etc. all gate on deterministic).

Given that, on the options:

• The capability method (pushDownNondeterministicPredicates) is the one I like least — it's a permanent public API that lets a connector keep depending on a behavior that isn't sound even for a fully-enforcing source. I'd rather not add it.
• On "this can suddenly break all connectors": that's the intended, documented breaking change (there's a migration-guide bullet in the PR). The default now does the safe thing — keep non-det filters post-scan — so nothing silently flips to a worse behavior. The legacy config only lets someone who built on the old behavior restore it, and we can deprecate and remove it later.

So I'd like to land this either as-is (safe default + the legacy config as a temporary, deprecatable escape hatch) or with no flag at all (just never push). Either works for me — I'd just avoid the capability.

[szehon-ho]

But the flag is not going to work. (If most connectors depend on new behavior, and suddenly we flip it)

One option is:

• a hidden API on SupportsPushdownV2Filters that defaults to false, and javadoc'ed to say 'do not do this unless very sure'
• and a jdbc specific flag to return true

Or not even have a flag works for me as well.

[peter-toth]

It's never the connector that depends on non-det pushdown. A connector just implements SupportsPushDownV2Filters and takes whatever Spark hands it. Getting rand() > 0.5 pushed in gains it nothing and isn't sound anyway, so no connector author would set a pushDownNondeterministicPredicates() capability to true on purpose — it'd be dead API. The thing that can accidentally rely on the old behavior is a user's workflow that happened to get the results it wanted out of Spark's weird old behavior, and the person who knows about that workflow isn't the connector author.

So the dependency lives at the query/session level, which is exactly where a spark.sql.legacy.* conf belongs: flip the default to the correct behavior now, give the few who built on the old one a temporary, deprecatable switch. And flipping the default can't break a connector — keeping non-det filters post-scan only ever makes results more correct, never crashes a source, so it's purely opt-back-in.

[peter-toth]

Sounds like we're converging on no flag. That's fine by me. If everyone's ok dropping the conf, I'll drop it tomorrow.