diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md index 2607f6d0e56b9..0a85a4503a16c 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`. - 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 carries a `null` value to drop a default). ## 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 4758d063ca03c..b4e5ce783c5ca 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 @@ -7864,6 +7864,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 6685b28c7dc87..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,6 +93,19 @@ 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. 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) 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. 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( """