From 125ec044829aa235c13c09e94d2e813e6ee65f9c Mon Sep 17 00:00:00 2001 From: Peter Toth Date: Sun, 19 Jul 2026 11:21:10 +0200 Subject: [PATCH 1/3] [SPARK-58207][SQL] Skip pushdown for nondeterministic V2 filters 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. --- .../datasources/v2/PushDownUtils.scala | 11 +++++++-- .../sql/connector/DataSourceV2Suite.scala | 23 +++++++++++++++++++ 2 files changed, 32 insertions(+), 2 deletions(-) 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 6685b28c7dc87..1fef7d0e76a45 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 must 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 c6f3c1c3886c6..cc784e6f73a07 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. From 8b03668ca6e61d2c1e8826c9ba3cdef075983d4e Mon Sep 17 00:00:00 2001 From: Peter Toth Date: Sun, 19 Jul 2026 14:44:31 +0200 Subject: [PATCH 2/3] [SPARK-58207][SQL] Gate nondeterministic V2 filter pushdown behind a 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. --- docs/sql-migration-guide.md | 1 + .../apache/spark/sql/internal/SQLConf.scala | 14 ++++++++ .../datasources/v2/PushDownUtils.scala | 10 ++++-- .../apache/spark/sql/jdbc/JDBCV2Suite.scala | 33 ++++++++++++++----- 4 files changed, 47 insertions(+), 11 deletions(-) diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md index fe19681d3357e..8061904d4e4d6 100644 --- a/docs/sql-migration-guide.md +++ b/docs/sql-migration-guide.md @@ -29,6 +29,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 `org.apache.spark.sql.execution.adaptive.DynamicJoinSelection` in `spark.sql.adaptive.optimizer.excludedRules`, that name no longer matches any rule (unknown names are silently ignored); set `spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled` to `false` instead. - 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 result these functions may return a value that differs from earlier releases in the last ULP. `percentile_disc` and `percentile_approx` are unaffected. +- 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. To restore the previous behavior, set `spark.sql.legacy.allowNonDeterministicV2FilterPushDown` to `true`. ## Upgrading from Spark SQL 4.1 to 4.2 diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index 0b3f4693853a7..3588214d0b935 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala @@ -7841,6 +7841,20 @@ object SQLConf { .createWithDefault(false) } + val LEGACY_ALLOW_NON_DETERMINISTIC_V2_FILTER_PUSHDOWN = + buildConf("spark.sql.legacy.allowNonDeterministicV2FilterPushDown") + .internal() + .doc("When set to true, restores the legacy behavior of pushing non-deterministic filters " + + "down to DataSource V2 sources that implement SupportsPushDownV2Filters. Pushing a " + + "non-deterministic predicate is unsafe because a source may evaluate it a different " + + "number of times or at a different point than Spark (e.g. use it for pruning yet also " + + "return it for post-scan re-evaluation, evaluating it twice with different results). " + + "When false, such filters are kept as post-scan filters.") + .version("4.3.0") + .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) + .booleanConf + .createWithDefault(false) + val XML_VARIANT_RESPECT_INFER_SCHEMA = buildConf("spark.sql.xml.variant.respectInferSchema") .doc( 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 1fef7d0e76a45..91208b2178061 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 @@ -98,8 +98,14 @@ object PushDownUtils extends Logging { // 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) + // SupportsPushDownCatalystFilters branch below. The legacy config restores the old behavior + // of pushing them (for sources that fully enforce such predicates, e.g. JDBC). + val (deterministicFilters, nonDeterministicFilters) = + if (SQLConf.get.getConf(SQLConf.LEGACY_ALLOW_NON_DETERMINISTIC_V2_FILTER_PUSHDOWN)) { + (filters, Seq.empty[Expression]) + } else { + 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 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 dd3d90d2df052..0af8fd34f368e 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,15 +1243,30 @@ class JDBCV2Suite extends SharedSparkSession with ExplainSuiteHelper { checkPushedInfo(df10, "PushedFilters: [ID IS NOT NULL, ID > 1]") checkAnswer(df10, Row("mary", 2)) - val df11 = sql( - """ - |SELECT * FROM h2.test.employee - |WHERE GREATEST(bonus, 1100) > 1200 AND RAND(1) < bonus - |""".stripMargin) - checkFiltersRemoved(df11) - checkPushedInfo(df11, "PushedFilters: " + - "[BONUS IS NOT NULL, (GREATEST(BONUS, 1100.0)) > 1200.0, RAND(1) < BONUS]") - checkAnswer(df11, Row(2, "david", 10000, 1300, true)) + // RAND(1) is non-deterministic. By default (SPARK-58207) it is not pushed down and stays as a + // post-scan filter; the legacy config restores pushing it down (JDBC fully enforces it in H2). + // RAND(1) < bonus is trivially true for the matching row (bonus = 1300 >= 1), so the answer is + // the same either way. + Seq(true, false).foreach { legacyPush => + withSQLConf(SQLConf.LEGACY_ALLOW_NON_DETERMINISTIC_V2_FILTER_PUSHDOWN.key -> + legacyPush.toString) { + val df11 = sql( + """ + |SELECT * FROM h2.test.employee + |WHERE GREATEST(bonus, 1100) > 1200 AND RAND(1) < bonus + |""".stripMargin) + if (legacyPush) { + checkFiltersRemoved(df11) + checkPushedInfo(df11, "PushedFilters: " + + "[BONUS IS NOT NULL, (GREATEST(BONUS, 1100.0)) > 1200.0, RAND(1) < BONUS]") + } else { + checkFiltersRemoved(df11, removed = false) + checkPushedInfo(df11, "PushedFilters: " + + "[BONUS IS NOT NULL, (GREATEST(BONUS, 1100.0)) > 1200.0]") + } + checkAnswer(df11, Row(2, "david", 10000, 1300, true)) + } + } val df12 = sql( """ From 76ebb972a31e0cc32a6bce5e3d159db6c77d1f09 Mon Sep 17 00:00:00 2001 From: Peter Toth Date: Tue, 21 Jul 2026 13:44:13 +0200 Subject: [PATCH 3/3] [SPARK-58207][SQL] Reword 'must not' to 'should not' in nondeterministic V2 filter comment Per review feedback: the legacy config can re-enable pushing non-deterministic filters, so 'must not be pushed down' was contradictory. Soften to 'should not'. --- .../spark/sql/execution/datasources/v2/PushDownUtils.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 91208b2178061..5a6c4b1b6d753 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,7 +93,7 @@ object PushDownUtils extends Logging { (postScanFilters ++ untranslatableExprs).toImmutableArraySeq) case r: SupportsPushDownV2Filters => - // Non-deterministic filters must not be pushed down: a data source may evaluate a pushed + // 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