You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ch...@apache.org on 2026/09/12 03:03:35 UTC
(spark) branch branch-4.2 updated: [SPARK-59434][SQL] Non-deterministic predicates should not be pushed down to CTE definitions
This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new ac091c6d68d5 [SPARK-59434][SQL] Non-deterministic predicates should not be pushed down to CTE definitions
ac091c6d68d5 is described below
commit ac091c6d68d556cecd3e6bdfc10255dceec4b032
Author: Cheng Pan <pa...@gmail.com>
AuthorDate: Sat Sep 12 11:02:32 2026 +0800
[SPARK-59434][SQL] Non-deterministic predicates should not be pushed down to CTE definitions
### What changes were proposed in this pull request?
`PushdownPredicatesAndPruneColumnsForCTEDef` collects the predicates at each CTE reference and pushes their OR-merged combination into the shared CTE definition. Each reference keeps its own predicates, so this only works for deterministic predicates -- a non-deterministic one is then evaluated a second time in the definition.
This PR filters the collected predicates to deterministic ones before push-down. When that leaves a reference with nothing pushable, the combined predicate becomes `TRUE` and the definition gets no push-down at all, including for its other references; the scaladoc is updated to say so.
### Why are the changes needed?
A non-deterministic predicate evaluated a second time in the CTE definition can drop rows, which is a wrong-results bug:
```sql
create or replace temp view t as select * from values (0), (1), (2) as t(c1);
with v as (select c1, rand(1) r from t)
select c1 from v where rand(2) < 0.5
union all
select c1 from v where rand(3) < 0.5;
```
The definition is non-deterministic and referenced twice, so it survives `InlineCTE`. `(rand(2) < 0.5) OR (rand(3) < 0.5)` is pushed into the definition while both references keep their own filter, so the optimized plan holds four `rand` filters instead of two and rows are filtered twice.
The rule has had this behavior since it was added in SPARK-37670 (3.2.2). Its scaladoc claimed determinism was taken care of by `ScanOperation`, but that was never true -- the filter-collection guard admits the first filter whatever it is:
```scala
filters.isEmpty || (filters.forall(_.deterministic) && condition.deterministic)
```
`PhysicalOperation`, which SPARK-39764 (3.4.0) later swapped in, behaves the same way here -- it returns a single filter even when it is non-deterministic, since its `filters.length > 1` assert only binds when more than one filter was collected. The scaladoc is corrected along with the fix.
### Does this PR introduce _any_ user-facing change?
Yes, it fixes wrong results for the query shape above. Affects 3.2.2 and later.
### How was this patch tested?
Four new tests in `CTEInlineSuite`, all failing without the fix:
- non-deterministic predicates stay at the references, two `rand` filters instead of four;
- deterministic conjuncts are still pushed while the non-deterministic one stays up;
- a reference with only non-deterministic predicates blocks push-down for its deterministic sibling, which must not be pushed alone;
- a row-count assertion using `monotonically_increasing_id()`, 10 rows instead of 8.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5
Closes #58735 from pan3793/cte-nondeterministic-pushdown.
Authored-by: Cheng Pan <pa...@gmail.com>
Signed-off-by: Cheng Pan <ch...@apache.org>
(cherry picked from commit 919f080854973ad0fcaac9c42e69353b2d6229f4)
Signed-off-by: Cheng Pan <ch...@apache.org>
---
...ushdownPredicatesAndPruneColumnsForCTEDef.scala | 17 ++--
.../org/apache/spark/sql/CTEInlineSuite.scala | 105 +++++++++++++++++++++
2 files changed, 116 insertions(+), 6 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala
index 421c47f12ddb..9e7ca44d0007 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala
@@ -56,10 +56,12 @@ object PushdownPredicatesAndPruneColumnsForCTEDef extends Rule[LogicalPlan] {
/**
* Gather all the predicates and referenced attributes on different points of CTE references
- * using pattern `ScanOperation` (which takes care of determinism) and combine those predicates
- * and attributes that belong to the same CTE definition.
- * For the same CTE definition, if any of its references does not have predicates, the combined
- * predicate will be a TRUE literal, which means there will be no predicate push-down.
+ * using pattern `PhysicalOperation` and combine those predicates and attributes that belong
+ * to the same CTE definition. `PhysicalOperation` still returns a single filter even when it
+ * is non-deterministic, so such predicates are excluded below.
+ * For the same CTE definition, if any of its references does not have pushable predicates, the
+ * combined predicate will be a TRUE literal, which means there will be no predicate push-down
+ * for that definition at all, including for its other references.
*/
private def gatherPredicatesAndAttributes(plan: LogicalPlan, cteMap: CTEMap): Unit = {
plan match {
@@ -76,8 +78,11 @@ object PushdownPredicatesAndPruneColumnsForCTEDef extends Rule[LogicalPlan] {
val newPredicates = if (isTruePredicate(preds)) {
preds
} else {
- // Make sure we only push down predicates that do not contain forward CTE references.
- val filteredPredicates = restoreCTEDefAttrs(predicates.filter(_.find {
+ // Only push down deterministic predicates that do not contain forward CTE references.
+ // A reference keeps its own predicates, so a non-deterministic one pushed into the
+ // shared definition as well would be evaluated a second time.
+ val deterministicPredicates = predicates.filter(_.deterministic)
+ val filteredPredicates = restoreCTEDefAttrs(deterministicPredicates.filter(_.find {
case s: SubqueryExpression => s.plan.find {
case r: CTERelationRef =>
// If the ref's ID does not exist in the map or if ref's corresponding precedence
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala
index 3833b7f2509d..10127c29b45c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala
@@ -451,6 +451,111 @@ abstract class CTEInlineSuiteBase
}
}
+ test("SPARK-59434: non-deterministic predicates are not pushed into a CTE def") {
+ withTempView("t") {
+ Seq(0, 1, 2).toDF("c1").createOrReplaceTempView("t")
+ // The CTE def is non-deterministic and referenced twice, so it is not inlined and the
+ // references' predicates get OR-merged into the shared def. A reference keeps its own
+ // predicate, so a non-deterministic one must not be pushed down as well.
+ val df = sql(
+ """with v as (select c1, rand(1) r from t)
+ |select c1 from v where rand(2) < 0.5
+ |union all
+ |select c1 from v where rand(3) < 0.5
+ |""".stripMargin)
+ val cteRepartitions = df.queryExecution.optimizedPlan.collect {
+ case r: RepartitionOperation => r
+ }
+ assert(cteRepartitions.nonEmpty,
+ "Non-deterministic With-CTE with multiple references should not be inlined.")
+ assert(
+ cteRepartitions.forall(_.collectFirst {
+ case f: Filter if f.condition.exists(_.isInstanceOf[Rand]) => f
+ }.isEmpty),
+ "Non-deterministic predicate should not be pushed down to the CTE def 'v'.")
+ val randFilters = df.queryExecution.optimizedPlan.collect {
+ case f: Filter if f.condition.exists(_.isInstanceOf[Rand]) => f
+ }
+ assert(randFilters.length == 2,
+ "Each reference's non-deterministic predicate should be evaluated once.")
+ }
+ }
+
+ test("SPARK-59434: deterministic conjuncts are still pushed into a CTE def") {
+ withTempView("t") {
+ Seq((0, 1), (1, 2), (2, 3)).toDF("c1", "c2").createOrReplaceTempView("t")
+ val df = sql(
+ """with v as (select c1, c2, rand(1) r from t)
+ |select c1 from v where c1 > 0 and rand(2) < 0.5
+ |union all
+ |select c1 from v where c1 < 2
+ |""".stripMargin)
+ val cteRepartitions = df.queryExecution.optimizedPlan.collect {
+ case r: RepartitionOperation => r
+ }
+ assert(cteRepartitions.nonEmpty, "CTE should not be inlined after optimization.")
+ // The non-deterministic conjunct stays at the reference, the deterministic ones are
+ // still OR-merged and pushed into the definition.
+ val distinctCteRepartitions = cteRepartitions.map(_.canonicalized).distinct
+ assert(distinctCteRepartitions.length == 1)
+ assert(
+ distinctCteRepartitions.head.collectFirst {
+ case f: Filter if f.condition.semanticEquals(
+ Or(GreaterThan(f.output(0), Literal(0)), LessThan(f.output(0), Literal(2)))) => f
+ }.isDefined,
+ "Predicate 'c1 > 0 OR c1 < 2' should be pushed down to the CTE def 'v'.")
+ assert(
+ distinctCteRepartitions.head.collectFirst {
+ case f: Filter if f.condition.exists(_.isInstanceOf[Rand]) => f
+ }.isEmpty,
+ "Non-deterministic predicate should not be pushed down to the CTE def 'v'.")
+ }
+ }
+
+ test("SPARK-59434: a non-deterministic reference blocks push-down for its siblings") {
+ withTempView("t") {
+ Seq((0, 1), (1, 2), (2, 3)).toDF("c1", "c2").createOrReplaceTempView("t")
+ val df = sql(
+ """with v as (select c1, c2, rand(1) r from t)
+ |select c1 from v where rand(2) < 0.5
+ |union all
+ |select c1 from v where c1 > 0
+ |""".stripMargin)
+ val cteRepartitions = df.queryExecution.optimizedPlan.collect {
+ case r: RepartitionOperation => r
+ }
+ assert(cteRepartitions.nonEmpty, "CTE should not be inlined after optimization.")
+ // The first reference has no pushable predicate, so the combined predicate is TRUE and
+ // the definition gets no filter. The sibling's 'c1 > 0' must not be pushed on its own,
+ // which would drop rows the first reference needs.
+ assert(
+ cteRepartitions.forall(_.collectFirst { case f: Filter => f }.isEmpty),
+ "CTE def 'v' should get no pushed-down filter.")
+ }
+ }
+
+ test("SPARK-59434: a non-deterministic predicate is evaluated once per CTE reference") {
+ withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "1") {
+ withTempView("t") {
+ spark.range(0, 6, 1, 1).selectExpr("cast(id as int) c1").createOrReplaceTempView("t")
+ // `monotonically_increasing_id` counts the rows it sees within a partition, so each
+ // reference drops exactly its own first row. A second evaluation in the shared def
+ // would drop a row there as well, leaving fewer. `rand(1)` only keeps the def from
+ // being inlined; column pruning removes it, so it never reaches a filter.
+ val df = sql(
+ """with v as (select c1, rand(1) r from t)
+ |select c1 from v where monotonically_increasing_id() > 0
+ |union all
+ |select c1 from v where monotonically_increasing_id() > 0
+ |""".stripMargin)
+ assert(
+ df.queryExecution.optimizedPlan.exists(_.isInstanceOf[RepartitionOperation]),
+ "Non-deterministic With-CTE with multiple references should not be inlined.")
+ assert(df.count() === 10, "Each reference should drop only its own first row.")
+ }
+ }
+ }
+
test("Views with CTEs - 1 temp view") {
withTempView("t", "t2") {
Seq((0, 1), (1, 2)).toDF("c1", "c2").createOrReplaceTempView("t")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org