This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new f633033db3d7 [SPARK-58207][SQL] Skip pushdown for nondeterministic V2
filters
f633033db3d7 is described below
commit f633033db3d72bd7c1d4ab547eefb3172731f8f5
Author: Peter Toth <[email protected]>
AuthorDate: Thu Jul 23 23:00:42 2026 +0800
[SPARK-58207][SQL] Skip pushdown for nondeterministic V2 filters
### What changes were proposed in this pull request?
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).
### 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.
### Does this PR introduce _any_ user-facing change?
Yes. 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. 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 assert
`RAND(1) < BONUS` is no longer pushed to H2 (only `BONUS IS NOT NULL` and
`GREATEST(BONUS, 1100.0) > 1200.0` are pushed) and is retained 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)
Closes #57357 from
peter-toth/SPARK-58207-skip-pushdown-nondeterministic-v2-filters.
Authored-by: Peter Toth <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 018ad68ccb8994425e2e79f9aa9cefca1f1b4c87)
Signed-off-by: Wenchen Fan <[email protected]>
---
docs/sql-migration-guide.md | 1 +
.../execution/datasources/v2/PushDownUtils.scala | 11 +++++++++--
.../spark/sql/connector/DataSourceV2Suite.scala | 23 ++++++++++++++++++++++
.../org/apache/spark/sql/jdbc/JDBCV2Suite.scala | 6 ++++--
4 files changed, 37 insertions(+), 4 deletions(-)
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index 9cd5b664e8eb..0db1259e1f97 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -30,6 +30,7 @@ license: |
- Since Spark 4.3, the Spark Thrift Server rejects setting JVM system
properties through the `set:system:` session configuration overlay (for
example, in a JDBC connection string). To restore the previous behavior, set
`spark.sql.legacy.hive.thriftServer.allowSettingSystemProperties` to `true`.
- Since Spark 4.3, the adaptive execution rule
`org.apache.spark.sql.execution.adaptive.DynamicJoinSelection` has been renamed
to `DemoteBroadcastHashJoin`, which now only demotes broadcast hash joins
(emitting `NO_BROADCAST_HASH`). Its selection of shuffled hash join over sort
merge join has moved to a new physical rule gated by
`spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled` (default
`true`). If you previously disabled the shuffled-hash-join preference by
listing `o [...]
- Since Spark 4.3, the exact `percentile`, `percentile_cont`, and `median`
aggregate functions (including their `WITHIN GROUP (ORDER BY ...)` forms)
compute the linear interpolation between two neighboring values as `lower +
fraction * (higher - lower)` instead of `(1 - fraction) * lower + fraction *
higher`. The two are equal in exact arithmetic, but the new form is
monotonically non-decreasing in the requested percentage and avoids a rounding
error the old form could introduce. As a re [...]
+- Since Spark 4.3, non-deterministic filters (for example predicates involving
`rand()`) are no longer pushed down to DataSource V2 sources that implement
`SupportsPushDownV2Filters`; they are evaluated by Spark after the scan
instead. This prevents a source from evaluating such a predicate a different
number of times than Spark, or using it for pruning while also returning it for
post-scan re-evaluation.
- Since Spark 4.3, the new `COMMENT ON COLUMN ... IS NULL` syntax removes a
column comment by passing a `null` comment to
`TableChange.updateColumnComment(String[], String)`, so a `UpdateColumnComment`
table change may now carry a `null` `newComment()`. Previously `newComment()`
was always non-null. DataSource V2 catalogs that handle `UpdateColumnComment`
should null-check `newComment()` and treat `null` as "remove the column
comment" (mirroring how `UpdateColumnDefaultValue` already car [...]
- Since Spark 4.3, `hash()` and `xxhash64()` include the `days` field of
`CalendarInterval` when computing the hash, so their output for interval values
differs from earlier releases. Previously the codegen path dropped `days`,
disagreeing with interpreted evaluation.
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
index 6685b28c7dc8..b74a785aafc3 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
@@ -93,6 +93,13 @@ object PushDownUtils extends Logging {
(postScanFilters ++ untranslatableExprs).toImmutableArraySeq)
case r: SupportsPushDownV2Filters =>
+ // Non-deterministic filters should not be pushed down: a data source
may evaluate a pushed
+ // predicate 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) would evaluate a non-deterministic
predicate twice with
+ // different results. Keep them as post-scan filters, matching the
+ // SupportsPushDownCatalystFilters branch below.
+ val (deterministicFilters, nonDeterministicFilters) =
filters.partition(_.deterministic)
// Divide the filters into those translatable and untranslatable to
data source filters.
// For the translated filters, we will try to push them down to the
data source,
// and the data source will return the filters that it cannot
guarantee to be true
@@ -101,7 +108,7 @@ object PushDownUtils extends Logging {
val translatedFilters = mutable.ArrayBuffer.empty[Predicate]
val untranslatableExprs = mutable.ArrayBuffer.empty[Expression]
- for (filterExpr <- filters) {
+ for (filterExpr <- deterministicFilters) {
val translated =
DataSourceV2Strategy.translateFilterV2WithMapping(
filterExpr, Some(translatedFilterToExpr))
@@ -131,7 +138,7 @@ object PushDownUtils extends Logging {
}
val orderedPostScanFilters = prioritizeFilters(finalPostScanFilters,
- ExpressionSet(untranslatableExprs))
+ ExpressionSet(untranslatableExprs)) ++ nonDeterministicFilters
(Right(r.pushedPredicates.toImmutableArraySeq), orderedPostScanFilters)
case r: SupportsPushDownCatalystFilters =>
val (deterministicFilters, nonDeterministicFilters) =
filters.partition(_.deterministic)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
index c6f3c1c3886c..cc784e6f73a0 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
@@ -1376,6 +1376,29 @@ class DataSourceV2Suite extends SharedSparkSession with
AdaptiveSparkPlanHelper
"non-deterministic filter should be retained as a post-scan Filter")
}
+ test("SPARK-58207: V2 filter pushdown skips non-deterministic filters") {
+ val df =
spark.read.format(classOf[AdvancedDataSourceV2WithV2Filter].getName).load()
+ // i > 3 is deterministic and pushable; rand() > 0.5 is non-deterministic.
rand() translates
+ // to a V2 Predicate (V2ExpressionBuilder handles Rand), so without a
guard it would be pushed
+ // to the source, which may evaluate it a different number of times than
Spark (or use it for
+ // pruning while also returning it for post-scan re-check).
+ val q = df.filter($"i" > 3 && rand() > 0.5)
+
+ // Only the deterministic predicate should reach the source's
pushPredicates.
+ val pushed = getBatchWithV2Filter(q).predicates.map(_.describe())
+ assert(pushed.exists(_.contains("i")),
+ s"the deterministic predicate on i should be pushed; got:
${pushed.mkString(", ")}")
+ assert(!pushed.exists(_.contains("RAND")),
+ s"a non-deterministic predicate must not be pushed; got:
${pushed.mkString(", ")}")
+
+ // The non-deterministic filter must be retained as a post-scan Filter.
+ val postScanConditions = q.queryExecution.optimizedPlan.collect {
+ case f: LogicalFilter => f.condition
+ }
+ assert(postScanConditions.exists(cond => cond.exists(!_.deterministic)),
+ "non-deterministic filter should be retained as a post-scan Filter")
+ }
+
test("pushedFilters drops filters referencing pruned columns") {
// Disable constraint propagation so IsNotNull(i) is not added (it would
keep
// column i in the scan output). This simulates a connector that pushes
IsNotNull.
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
index dd3d90d2df05..d7e5786ecb9c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
@@ -1243,14 +1243,16 @@ class JDBCV2Suite extends SharedSparkSession with
ExplainSuiteHelper {
checkPushedInfo(df10, "PushedFilters: [ID IS NOT NULL, ID > 1]")
checkAnswer(df10, Row("mary", 2))
+ // RAND(1) is non-deterministic, so (SPARK-58207) it is not pushed down
and stays as a
+ // post-scan filter. RAND(1) < bonus is trivially true for the matching
row (bonus = 1300 >= 1).
val df11 = sql(
"""
|SELECT * FROM h2.test.employee
|WHERE GREATEST(bonus, 1100) > 1200 AND RAND(1) < bonus
|""".stripMargin)
- checkFiltersRemoved(df11)
+ checkFiltersRemoved(df11, removed = false)
checkPushedInfo(df11, "PushedFilters: " +
- "[BONUS IS NOT NULL, (GREATEST(BONUS, 1100.0)) > 1200.0, RAND(1) <
BONUS]")
+ "[BONUS IS NOT NULL, (GREATEST(BONUS, 1100.0)) > 1200.0]")
checkAnswer(df11, Row(2, "david", 10000, 1300, true))
val df12 = sql(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
