This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new 2d82460a20d8 [SPARK-58207][SQL][FOLLOWUP] Skip runtime filter pushdown
for nondeterministic filters
2d82460a20d8 is described below
commit 2d82460a20d8df40074f8e7e980c5b704fa336dc
Author: Peter Toth <[email protected]>
AuthorDate: Wed Aug 5 11:08:40 2026 +0200
[SPARK-58207][SQL][FOLLOWUP] Skip runtime filter pushdown for
nondeterministic filters
### What changes were proposed in this pull request?
Follow-up to #57357. `DataSourceV2Strategy` no longer routes a
non-deterministic post-scan filter into `BatchScanExec.runtimeFilters`, so such
a filter is not pushed to a `SupportsRuntimeV2Filtering` scan at runtime. The
migration guide entry added by #57357 is widened to cover the runtime path.
#57357 guarded the `SupportsPushDownV2Filters` branch of
`PushDownUtils.pushFilters`, i.e. pushdown at query compilation. It did not
touch the runtime filter path: `DataSourceV2Strategy` routes scalar subquery
filters on runtime-filterable columns into `runtimeFilters` (SPARK-56467), and
`PushDownUtils.pushRuntimeFilters` translates them with
`DataSourceV2Strategy.translateScalarSubqueryFilterV2`, which has no
determinism guard.
Since #57357 shipped in 4.3 and this completes it, **this should be merged
to `branch-4.3` as well as `master` and `branch-4.x`** so the widened
migration-guide bullet ("Since Spark 4.3") is accurate and 4.3 carries the
whole fix. This is a correctness fix rather than a new feature, and the
cherry-pick is clean on both branches (verified locally).
### Why are the changes needed?
`V2ExpressionBuilder` translates `Rand`, so a non-deterministic runtime
filter reaches the data source. On master, against a
`SupportsRuntimeV2Filtering` in-memory table:
SELECT * FROM tbl WHERE part = (SELECT max(val) FROM dim) OR rand() <
0.5
pushedPredicates: (part = 3) OR (RAND() < 0.5)
plan: Filter ((part#38 = Subquery subquery#36) OR
(rand(-6129082941936456980) < 0.5))
+- BatchScan ...
This is the problem #57357 describes, and the runtime path is more exposed
to it than the compilation path. A scalar subquery runtime filter is
deliberately kept in `postScanFilters` as well ("These filters stay in
postScanFilters for correctness"), so Spark always re-evaluates it above the
scan while the source is also free to prune on it. When the source does, the
two evaluations of `rand()` disagree, and a partition the source dropped is
gone -- rows that Spark's evaluation would h [...]
Gating the routing rather than the translation keeps the non-deterministic
filter out of `runtimeFilters` entirely, so it also stays out of
`BatchScanExec`'s `equals`/`doCanonicalize` and out of `EXPLAIN`, and no
`filter()` + `planInputPartitions()` round is wasted. Dynamic partition pruning
filters need no such check: a `DynamicPruningSubquery` over a non-deterministic
filtering plan is itself non-deterministic (`PlanExpression.deterministic`
folds in `plan.deterministic`), so the `F [...]
### Does this PR introduce _any_ user-facing change?
Yes, in unreleased 4.3, extending what #57357 already documented. Data
sources implementing `SupportsRuntimeV2Filtering` no longer receive
non-deterministic runtime filters through `filter`. The filter is still
evaluated by Spark after the scan, as it already was, so query results do not
change for a source that ignored the pushed predicate; a source that pruned on
it returns more rows than before, which is the point of the fix. The migration
guide entry is updated.
### How was this patch tested?
Added `SPARK-58207: non-deterministic scalar subquery filters are not
pushed into runtimeFilters` to `DataSourceV2SQLSuiteV2Filter`, next to the
SPARK-56467 test it mirrors. It fails on master and passes with this change;
the assertion that discriminates is `runtimeFilters.isEmpty` (on master
`runtimeFilters` holds `(part = subquery) OR (rand(...) < 0.5)`). The other two
assertions -- no partition pruned, filter still evaluated above the scan --
hold on master too and are there as san [...]
Ran `DataSourceV2SQLSuiteV2Filter`,
`DataSourceV2EnhancedRuntimePartitionFilterSuite` (iterative
`PartitionPredicate` second pass) and `DataSourceV2Suite` locally, 243 tests
green, plus `dev/lint-scala`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
Closes #57760 from
peter-toth/SPARK-58207-followup-runtime-filter-nondeterministic.
Authored-by: Peter Toth <[email protected]>
Signed-off-by: Peter Toth <[email protected]>
(cherry picked from commit 64c896aafd56a62e0ca958f27171d071d9ed3270)
Signed-off-by: Peter Toth <[email protected]>
---
docs/sql-migration-guide.md | 2 +-
.../datasources/v2/DataSourceV2Strategy.scala | 10 +++++-
.../execution/datasources/v2/PushDownUtils.scala | 5 +++
.../spark/sql/connector/DataSourceV2SQLSuite.scala | 41 ++++++++++++++++++++++
4 files changed, 56 insertions(+), 2 deletions(-)
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index c04d20f44d5e..40fd895d9983 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -34,7 +34,7 @@ license: |
- Since Spark 4.3, `spark.sql.execution.replaceHashWithSortAgg` defaults to
`true`. Spark now replaces a hash-based aggregate with a sort aggregate when
the aggregate's child is already sorted on the grouping keys. To restore the
previous behavior, set `spark.sql.execution.replaceHashWithSortAgg` to `false`.
- Since Spark 4.3, `spark.sql.execution.combineAdjacentAggregation` defaults
to `true`. Spark now merges an adjacent partial/final aggregate pair (with no
shuffle between them) into a single complete-mode aggregate. This setting is
independent of `spark.sql.execution.replaceHashWithSortAgg`, so disabling only
`replaceHashWithSortAgg` still leaves adjacent aggregation combined; to fully
restore the previous partial/final staging, set both
`spark.sql.execution.replaceHashWithSortAgg` and ` [...]
- 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, non-deterministic filters (for example predicates involving
`rand()`) are no longer pushed down to DataSource V2 sources: neither at query
compilation to sources that implement `SupportsPushDownV2Filters`, nor as
runtime filters to sources that implement `SupportsRuntimeV2Filtering`; 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 retur [...]
- 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, `unix_seconds`, `unix_millis`, and `unix_micros` accept
`TIMESTAMP_NTZ` and the nanosecond-precision timestamp types directly, reading
them with no time-zone shift. Previously these functions accepted only
`TIMESTAMP_LTZ`; a `TIMESTAMP_NTZ` or nanosecond-timestamp argument was
rejected with a `DATATYPE_MISMATCH` error. This is a new capability and does
not change the result of any query that previously succeeded.
- 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/DataSourceV2Strategy.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
index 72807a5242c7..ce9092ea810b 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
@@ -172,15 +172,23 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
// These filters stay in postScanFilters for correctness (FilterExec
above scan),
// but are also routed into runtimeFilters so BatchScanExec can use them
for
// partition pruning via SupportsRuntimeV2Filtering.filter().
+ // Non-deterministic filters are not routed: they would be pushed to the
source for
+ // pruning while the FilterExec above the scan re-evaluates them, so the
two evaluations
+ // may disagree and rows the source pruned away could not be recovered.
This is the
+ // runtime counterpart of the pushFilters guard in PushDownUtils
(SPARK-58207).
val scalarSubqueryFilters = if (relation.runtimeFilterAttrs.nonEmpty) {
postScanFilters.filter { f =>
- f.containsPattern(SCALAR_SUBQUERY) &&
+ f.deterministic &&
+ f.containsPattern(SCALAR_SUBQUERY) &&
f.references.nonEmpty &&
f.references.subsetOf(relation.runtimeFilterAttrs)
}
} else {
Seq.empty
}
+ // dynamicFilters need no such check: a DynamicPruningSubquery over a
non-deterministic
+ // filtering plan is itself non-deterministic, so
CleanupDynamicPruningFilters has already
+ // rewritten it to TrueLiteral by the time we get here.
val runtimeFilters = dynamicFilters ++ scalarSubqueryFilters
val batchExec = BatchScanExec(relation.output, relation.scan,
runtimeFilters,
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 46c4e9859582..5e1e22a0c66f 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
@@ -170,6 +170,11 @@ object PushDownUtils extends Logging {
* Note: Do not call multiple times for the same `scan` instance;
* [[SupportsRuntimeV2Filtering.filter]] is mutating.
*
+ * Note: `runtimeFilters` must not contain non-deterministic filters. A
runtime filter is also
+ * evaluated by the `FilterExec` above the scan, so pushing a
non-deterministic one would
+ * evaluate it twice with different results. `DataSourceV2Strategy` enforces
this where
+ * `runtimeFilters` is built (SPARK-58207).
+ *
* @return true if any filters were pushed to the data source
*/
def pushRuntimeFilters(
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
index 79c46bdb4ccf..76c7902188ed 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
@@ -5478,6 +5478,47 @@ class DataSourceV2SQLSuiteV2Filter extends
DataSourceV2SQLSuite {
s"Expected 1 partition after scalar subquery pruning, got
$numPartitions")
}
}
+
+ test("SPARK-58207: non-deterministic scalar subquery filters are not pushed
into " +
+ "runtimeFilters") {
+ val tbl = s"${catalogAndNamespace}tbl"
+ val dim = s"${catalogAndNamespace}dim"
+ withTable(tbl, dim) {
+ sql(s"CREATE TABLE $tbl (id INT, part INT) USING $v2Format PARTITIONED
BY (part)")
+ for (i <- 0 until 10) {
+ sql(s"INSERT INTO $tbl VALUES ($i, $i)")
+ }
+
+ sql(s"CREATE TABLE $dim (val INT) USING $v2Format")
+ sql(s"INSERT INTO $dim VALUES (3)")
+
+ // `part = (subquery) OR rand() < 0.5` references only the partition
column and holds a
+ // scalar subquery, so it is a candidate for runtime pushdown, but it is
non-deterministic.
+ // Routing it would push it to the source for pruning while the
FilterExec above the scan
+ // re-evaluates it, and a partition the source dropped on its own
evaluation could not be
+ // recovered.
+ val df = sql(
+ s"SELECT * FROM $tbl WHERE part = (SELECT max(val) FROM $dim) OR
rand() < 0.5")
+ df.collect()
+
+ val batchScan = collect(df.queryExecution.executedPlan) {
+ case b: BatchScanExec => b
+ }.head
+ assert(batchScan.runtimeFilters.isEmpty,
+ s"Expected no runtime filters for a non-deterministic filter, " +
+ s"got ${batchScan.runtimeFilters}")
+
+ // No pruning at the source, and the filter is still evaluated by Spark
after the scan.
+ val numPartitions = batchScan.filteredPartitions.count(_.isDefined)
+ assert(numPartitions == 10,
+ s"Expected all 10 partitions to be retained, got $numPartitions")
+ val postScanConditions = collect(df.queryExecution.executedPlan) {
+ case f: FilterExec => f.condition
+ }
+ assert(postScanConditions.exists(!_.deterministic),
+ s"Expected the non-deterministic filter above the scan, got
$postScanConditions")
+ }
+ }
}
class ReserveSchemaNullabilityCatalog extends InMemoryCatalog {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
