[SPARK-58207][SQL] Skip pushdown for nondeterministic V2 filters

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
 

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(

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala

@@ -93,6 +93,19 @@ 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. 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
@@ -101,7 +114,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 +144,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)

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.

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(
       """