[SPARK-58207][SQL] Skip pushdown for nondeterministic V2 filters#57357
[SPARK-58207][SQL] Skip pushdown for nondeterministic V2 filters#57357peter-toth wants to merge 2 commits into
Conversation
|
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 The gap can be closed either way:
WDYT — which direction do you prefer for closing it? |
PushDownUtils.pushFilters does not filter out nondeterministic predicates in the SupportsPushDownV2Filters branch, and V2ExpressionBuilder translates Rand, so a predicate like rand() > 0.5 is pushed to the data source. This is unsound: the source may evaluate a pushed predicate a different number of times or at a different point than Spark, and a partial push (used for pruning yet also returned for post-scan re-check, e.g. a parquet row group filter) evaluates it twice with different results. Guard the SupportsPushDownV2Filters branch by pushing only deterministic filters and keeping nondeterministic ones as post-scan filters, mirroring the fix made for SupportsPushDownCatalystFilters in SPARK-58112. The SupportsPushDownFilters (V1) path already avoids this (nondeterministic predicates are untranslatable to sources.Filter), and the PartitionPredicate second pass already guards via isPushablePartitionFilter. Adds a regression test in DataSourceV2Suite that fails before this change (rand() is pushed to the V2 filter source) and passes after.
6a1e3ee to
125ec04
Compare
cloud-fan
left a comment
cloud-fan
left a comment
There was a problem hiding this comment.
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.
125ec04 to
00c08df
Compare
…legacy config Follow-up to the review of the previous commit. Skipping pushdown of nondeterministic filters is a behavior change: some sources fully enforce a pushed predicate (e.g. JDBC evaluates it in the database), so this could regress them. Gate the new behavior behind a legacy config spark.sql.legacy.allowNonDeterministicV2FilterPushDown (default false; set to true to restore the old behavior of pushing nondeterministic filters). - Add the legacy config in SQLConf (NOT_APPLICABLE binding policy) and honor it in the SupportsPushDownV2Filters branch of PushDownUtils.pushFilters. - Update the JDBCV2Suite RAND(1) pushdown test to run with the config both on and off. - Document the change in the SQL migration guide.
00c08df to
8b03668
Compare
|
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 It's in a separate follow-up commit on top of the one you approved. |
|
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
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. |
| (postScanFilters ++ untranslatableExprs).toImmutableArraySeq) | ||
|
|
||
| case r: SupportsPushDownV2Filters => | ||
| // Non-deterministic filters must not be pushed down: a data source may evaluate a pushed |
There was a problem hiding this comment.
this comment is a bit contradictory, maybe must => should.
as the flag is going to push them if the user chooses to
|
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.
We can just leave out the second round PartitionPredicate and say that never pushdown non-deterministic filter (unless there is a need later), wdyt? |
5 participants
What changes were proposed in this pull request?
By default, the
SupportsPushDownV2Filtersbranch ofPushDownUtils.pushFiltersnow 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 forSupportsPushDownCatalystFiltersin #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(defaultfalse), 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
SupportsPushDownV2Filtersinstead 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
SupportsPushDownV2Filtersbranch translated and pushed every conjunct, andV2ExpressionBuildertranslatesRand, so a predicate such asrand() > 0.5was 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(V1sources.Filter): nondeterministic predicates are untranslatable tosources.Filter, so they fall through to post-scan.PartitionPredicatesecond pass: guarded byPushDownUtils.isPushablePartitionFilter(f.deterministic).Only the
SupportsPushDownV2Filtersfirst-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
SupportsPushDownV2Filtersno longer receive nondeterministic filters throughpushPredicates; 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. Setspark.sql.legacy.allowNonDeterministicV2FilterPushDowntotrueto restore the previous behavior. Documented in the SQL migration guide.How was this patch tested?
DataSourceV2Suite("SPARK-58207: V2 filter pushdown skips non-deterministic filters") usingAdvancedDataSourceV2WithV2Filter: asserts thatrand() > 0.5is not pushed to the source and is retained as a post-scan filter (fails on master, passes with this change).JDBCV2SuiteRAND(1) < bonuspushdown test to run with the legacy config both on and off: on,RAND(1) < BONUSis pushed to H2 (as before); off, it stays as a post-scan filter.DataSourceV2SuiteandDataSourceV2EnhancedPartitionFilterSuite(the latter exercises the iterativePartitionPredicatesecond pass) locally, both green.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)