This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new e8b2d0e8ea92 [SPARK-60084][SQL] Fix PlanMerger losing a reused filter
attribute before the aggregate
e8b2d0e8ea92 is described below
commit e8b2d0e8ea92b5d71920e9fcd02e4faa68a5a0b5
Author: Peter Toth <[email protected]>
AuthorDate: Thu Oct 8 21:10:45 2026 +0200
[SPARK-60084][SQL] Fix PlanMerger losing a reused filter attribute before
the aggregate
### What changes were proposed in this pull request?
`PlanMerger.mergeNamedExpressions` appended the new plan's propagated
filter attribute only when it was a fresh alias. A reused one (`isNew = false`)
was assumed to be in the merged expression list already.
This PR appends the new plan's filter attribute whenever the list doesn't
output it yet. It also drops the `isNew` flag from
`TryMergeResult.newPlanFilter`, because nothing reads it any more.
The reuse lookup in the tagged `(Filter, Filter)` branch now only matches
the filter aliases that the merged `Filter` references. Before, a computed
column with the same expression as the new plan's filter could be reused, and
the merged `Filter` would drop that plan's rows.
### Why are the changes needed?
The assumption doesn't hold. A merge round can fold a filter attribute into
another filter alias, and then only the new alias is carried up to the
aggregate. A later round that reuses the original attribute ends up with an
aggregate `FILTER` on an attribute its child doesn't output:
```sql
SET
spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled=true;
SET
spark.sql.optimizer.mergeSubplans.filterPropagation.throughJoin.enabled=true;
CREATE TABLE t USING parquet AS SELECT * FROM VALUES
(1, 1, 10), (1, 2, 3), (2, 1, 7), (2, 2, 9), (3, 1, 9), (3, 2, 6) AS v(a,
b, c);
SELECT
(SELECT sum(l.a) FROM t l LEFT JOIN t r ON l.a = r.a WHERE l.b = 1),
(SELECT max(l.a) FROM t l LEFT JOIN t r ON l.a = r.a WHERE l.b = 2 AND
coalesce(r.c, 10) > 5),
(SELECT count(*) FROM t l LEFT JOIN t r ON l.a = r.a WHERE l.b = 2);
```
```
[INTERNAL_ERROR_ATTRIBUTE_NOT_FOUND] Could not find propagatedFilter_3#179
in
[sum#141L,max#182,count#145L,a#6,propagatedFilter_5#181,propagatedFilter_4#180].
SQLSTATE: XX000
```
What happens there:
1. Merging the 2nd subquery into the 1st aliases `l.b = 2` on the left side
of the join.
2. The 2nd subquery's post-join `coalesce(r.c, 10) > 5` folds that alias
into a new `AND` alias.
3. The 3rd subquery reuses the `l.b = 2` alias, but the `Project` under the
aggregate only outputs the new one.
This was found while reviewing #59161, which runs into the same problem
through its new join `AND`.
### Does this PR introduce _any_ user-facing change?
Yes. The query above now returns `12, 3, 6` instead of failing. It needs
both configs above, which are off by default. The bug is in 4.2.0.
### How was this patch tested?
New tests in `MergeSubplansSuite` (merged plans, including the computed
column case) and `PlanMergingSuite` (end to end). They all fail without the
fix. `MergeSubplansSuite` and `PlanMergingSuite` pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
Closes #59294 from
peter-toth/SPARK-60084-planmerger-reused-filter-attribute.
Authored-by: Peter Toth <[email protected]>
Signed-off-by: Peter Toth <[email protected]>
(cherry picked from commit 90568fb163775793260ecb13f2019ea10ef4e2f3)
Signed-off-by: Peter Toth <[email protected]>
---
.../sql/execution/planmerging/PlanMerger.scala | 63 ++++++-----
.../execution/planmerging/MergeSubplansSuite.scala | 116 +++++++++++++++++++++
.../execution/planmerging/PlanMergingSuite.scala | 40 +++++++
3 files changed, 187 insertions(+), 32 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
index 2eb50ce63a4e..8c7d2f28b56e 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
@@ -311,12 +311,11 @@ class PlanMerger(
* @param newPlanFilter A boolean [[Attribute]] in the merged plan that
encodes the filter
* condition from the new plan's side, to be applied as
an aggregate
* `FILTER (WHERE ...)` clause when the propagation
reaches an enclosing
- * [[Aggregate]] node. The boolean component is `true`
if the attribute was
- * freshly aliased and must be appended to enclosing
[[Project]] nodes, or
- * `false` if it was reused from an existing alias
already present in the
- * merged plan. `None` when no differing filter was
propagated.
+ * [[Aggregate]] node. It can be an alias reused from a
previous merge
+ * round. Such an alias may be missing from an
enclosing [[Project]], which
+ * then has to append it. `None` when no differing
filter was propagated.
* @param cachedPlanFilter Like `newPlanFilter` but for the cached plan's
side. Always a freshly
- * created alias when present, so no `isNew` flag is
needed.
+ * created alias when present, never a reused one.
* @param dsv2Merged Whether an (equal-strict) DSv2 scan merge occurred
anywhere in this merged
* subtree. Unlike `dsv2DeferredScan` (consumed at the
enclosing Filter that
* builds the scan), this fact is propagated all the way
up: it lets a Filter
@@ -327,7 +326,7 @@ class PlanMerger(
case class TryMergeResult(
mergedPlan: LogicalPlan,
newPlanMapping: AttributeMap[Attribute],
- newPlanFilter: Option[(Attribute, Boolean)] = None,
+ newPlanFilter: Option[Attribute] = None,
cachedPlanFilter: Option[Attribute] = None,
dsv2DeferredScan: Option[DSv2DeferredScan] = None,
dsv2Merged: Boolean = false)
@@ -489,9 +488,8 @@ class PlanMerger(
// A None filter means the side's aggregate expressions already
carry their individual
// FILTER attributes from a previous merge round and should be
left unchanged.
// Filter propagation is consumed here and not passed further up.
- val filteredNPAggregateExpressions =
npFilterOpt.fold(np.aggregateExpressions) {
- case (f, _) =>
applyFilterToAggregateExpressions(np.aggregateExpressions, f)
- }
+ val filteredNPAggregateExpressions =
npFilterOpt.fold(np.aggregateExpressions)(
+ applyFilterToAggregateExpressions(np.aggregateExpressions, _))
val filteredCPAggregateExpressions =
cpFilterOpt.fold(cp.aggregateExpressions)(
applyFilterToAggregateExpressions(cp.aggregateExpressions, _))
val (mergedAggregateExpressions, newNPMapping) =
@@ -539,27 +537,30 @@ class PlanMerger(
// merge rather than fail, keeping the merge best-effort.
mergedChild match {
case childProject: Project =>
- val newNPCondition = npFilter.fold(mappedNPCondition) {
- case (f, _) => And(f, mappedNPCondition)
- }
+ val newNPCondition =
+ npFilter.fold(mappedNPCondition)(And(_,
mappedNPCondition))
// If newNPCondition is already aliased in the child
Project (e.g. a third
// subplan whose filter matches one from a previous
merge round), reuse the
- // existing attribute instead of creating a redundant
alias.
+ // existing attribute instead of creating a redundant
alias. Only the aliases
+ // the OR condition references are filter sides. A
computed column with the
+ // same expression is not one, because the OR doesn't
keep its rows.
val existingNPFilter =
childProject.projectList.collectFirst {
- case a: Alias if a.child.canonicalized ==
newNPCondition.canonicalized =>
+ case a: Alias if
cp.condition.references.contains(a.toAttribute) &&
+ a.child.canonicalized ==
newNPCondition.canonicalized =>
a.toAttribute
}
val (newProjectList, newCondition, newNPFilterOut) =
existingNPFilter match {
case Some(reusedFilter) =>
// np matches an existing side: no new alias, OR
condition unchanged.
- (childProject.projectList, cp.condition,
(reusedFilter, false))
+ (childProject.projectList, cp.condition,
reusedFilter)
case None =>
val newNPFilterAlias =
Alias(newNPCondition,
s"propagatedFilter_${PlanMerger.newId}")()
+ val newNPFilter = newNPFilterAlias.toAttribute
(childProject.projectList :+ newNPFilterAlias,
- Or(cp.condition, newNPFilterAlias.toAttribute):
Expression,
- (newNPFilterAlias.toAttribute, true))
+ Or(cp.condition, newNPFilter): Expression,
+ newNPFilter)
}
// Phase 2: the leaf re-merge rebuilt the scan with
strict filters only,
// dropping the OR best-effort filter established in
earlier rounds, so
@@ -589,8 +590,7 @@ class PlanMerger(
// Note: the new Project always uses mergedChild as its
child (rather than
// flattening into an existing Project below) because
mergedChild.output may
// contain previously-propagated filter attributes that
cp.condition references.
- val newNPCondition =
- npFilter.fold(mappedNPCondition) { case (f, _) => And(f,
mappedNPCondition) }
+ val newNPCondition = npFilter.fold(mappedNPCondition)(And(_,
mappedNPCondition))
val newCPCondition = cpFilter.fold(cp.condition)(And(_,
cp.condition))
// The OR-widen moves both conditions into a boolean Project
above the merged
// scan, so the scan itself would read the full table. Build
the deferred scan
@@ -612,7 +612,7 @@ class PlanMerger(
val newFilter = Filter(Or(newNPFilter, newCPFilter),
project)
newFilter.copyTagsFrom(cp)
newFilter.setTagValue(PlanMerger.MERGED_FILTER_TAG, ())
- TryMergeResult(newFilter, npMapping, Some((newNPFilter,
true)),
+ TryMergeResult(newFilter, npMapping, Some(newNPFilter),
Some(newCPFilter), dsv2Merged = dsv2Merged)
}
}
@@ -628,16 +628,14 @@ class PlanMerger(
case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter, _,
dsv2Merged)
if cpFilter.isEmpty || symmetricFilterPropagationEnabled =>
val mappedNPCondition = mapAttributes(np.condition, npMapping)
- val newNPCondition = npFilter.fold(mappedNPCondition) {
- case (f, _) => And(f, mappedNPCondition)
- }
+ val newNPCondition = npFilter.fold(mappedNPCondition)(And(_,
mappedNPCondition))
val newNPFilterAlias =
Alias(newNPCondition,
s"propagatedFilter_${PlanMerger.newId}")()
val newNPFilter = newNPFilterAlias.toAttribute
val project = Project(
mergedChild.output.toList :+ newNPFilterAlias,
mergedChild)
- TryMergeResult(project, npMapping, Some((newNPFilter, true)),
cpFilter,
+ TryMergeResult(project, npMapping, Some(newNPFilter), cpFilter,
dsv2Merged = dsv2Merged)
}
case (np, cp: Filter) if context.filterPropagationSupported =>
@@ -1048,14 +1046,16 @@ class PlanMerger(
// non-matching expression from one side evaluates to null for rows that
belong to the other side,
// which is safe for aggregate FILTER (WHERE ...) semantics and avoids
computing values for
// irrelevant rows. The filter attributes themselves are appended to the
merged expression list so
- // they remain visible to the enclosing Aggregate that will consume them. A
newPlanFilter with
- // isNew=false was reused from a previous merge round and is already present
in the merged child
- // output, so it is not appended again.
+ // they remain visible to the enclosing Aggregate that will consume them. A
newPlanFilter reused
+ // from a previous merge round can already be in the list, so it is appended
only when missing.
+ // It can be missing, because an earlier round may have folded it into
another filter alias and
+ // carried only that alias up. A cachedPlanFilter is always a fresh alias,
so it is always
+ // appended.
private def mergeNamedExpressions(
newPlanExpressions: Seq[NamedExpression],
cachedPlanExpressions: Seq[NamedExpression],
newPlanMapping: AttributeMap[Attribute],
- newPlanFilter: Option[(Attribute, Boolean)] = None,
+ newPlanFilter: Option[Attribute] = None,
cachedPlanFilter: Option[Attribute] = None) = {
val mergedExpressions =
mutable.ArrayBuffer[NamedExpression](cachedPlanExpressions: _*)
val matchedCachedIndices = mutable.HashSet.empty[Int]
@@ -1078,7 +1078,7 @@ class PlanMerger(
// is only computed for rows that belong to the new plan side. Plain
attribute references
// are not wrapped since reading a column value is free.
val wrappedExpr: NamedExpression = newPlanFilter match {
- case Some((f, _)) if !withoutAlias.isInstanceOf[Attribute] =>
+ case Some(f) if !withoutAlias.isInstanceOf[Attribute] =>
Alias(If(f, withoutAlias, Literal(null, withoutAlias.dataType)),
mapped.name)()
case _ => mapped
}
@@ -1105,9 +1105,8 @@ class PlanMerger(
}
}
- newPlanFilter.foreach {
- case (f, true) => mergedExpressions += f
- case _ =>
+ newPlanFilter.foreach { f =>
+ if (!mergedExpressions.exists(_.exprId == f.exprId)) mergedExpressions
+= f
}
cachedPlanFilter.foreach(mergedExpressions += _)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
index af50a1520e44..0f94ab43259c 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
@@ -1981,6 +1981,122 @@ class MergeSubplansSuite extends PlanTest {
}
}
+ // Merging subquery2 into subquery1 aliases `b = 2` as propagatedFilter_0
below the join, and
+ // subquery2's post-join filter folds it into propagatedFilter_2. The
Projects above the join only
+ // carry propagatedFilter_2, so when subquery3 reuses propagatedFilter_0 on
its own, the merged
+ // Project under the aggregate has to append it.
+ private def checkReusedFilterAfterPostJoinFold(selectInSubquery3: Boolean):
Unit = {
+ val subquery1 = ScalarSubquery(
+ testRelation.where($"b" === 1).join(testRelation2, LeftOuter, Some($"a"
=== $"d"))
+ .select($"a")
+ .groupBy()(sum($"a").as("sum_a")))
+ val subquery2 = ScalarSubquery(
+ testRelation.where($"b" === 2).join(testRelation2, LeftOuter, Some($"a"
=== $"d"))
+ .where(coalesce($"e", 10) > 5)
+ .select($"a")
+ .groupBy()(max($"a").as("max_a")))
+ val join3 =
+ testRelation.where($"b" === 2).join(testRelation2, LeftOuter, Some($"a"
=== $"d"))
+ val subquery3 = ScalarSubquery(
+ (if (selectInSubquery3) join3.select($"a") else join3)
+ .groupBy()(count($"a").as("cnt")))
+ val originalQuery = testRelation.select(subquery1, subquery2, subquery3)
+
+ val f0Alias = Alias($"b" === 2, "propagatedFilter_0")()
+ val f0 = f0Alias.toAttribute
+ val f1Alias = Alias($"b" === 1, "propagatedFilter_1")()
+ val f1 = f1Alias.toAttribute
+ val joined = testRelation
+ .select(testRelation.output ++ Seq(f0Alias, f1Alias): _*)
+ .where(Or(f0, f1))
+ .join(testRelation2, LeftOuter, Some($"a" === $"d"))
+ val f2Alias = Alias(And(f0, coalesce($"e", 10) > 5),
"propagatedFilter_2")()
+ val f2 = f2Alias.toAttribute
+ // Without its own Project, subquery3 merges all its join columns into the
cached Project.
+ val mergedProjectList = if (selectInSubquery3) {
+ Seq($"a", f2, f1, f0)
+ } else {
+ Seq($"a", f2, f1, $"b", $"c", $"d", $"e", f0)
+ }
+ val mergedSubquery = joined
+ .select(joined.output :+ f2Alias: _*)
+ .select(mergedProjectList: _*)
+ .groupBy()(
+ sum($"a", Some(f1)).as("sum_a"),
+ max($"a", Some(f2)).as("max_a"),
+ count($"a", Some(f0)).as("cnt"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_a"), $"sum_a",
+ Literal("max_a"), $"max_a",
+ Literal("cnt"), $"cnt"
+ )).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1),
+ extractorExpression(0, analyzedMergedSubquery.output, 2)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key ->
"true",
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_THROUGH_JOIN_ENABLED.key ->
"true") {
+ comparePlans(Optimize.execute(originalQuery.analyze),
correctAnswer.analyze)
+ }
+ }
+
+ test("SPARK-60084: Reuse a filter that an earlier merge folded into a
post-join filter") {
+ checkReusedFilterAfterPostJoinFold(selectInSubquery3 = true)
+ }
+
+ test("SPARK-60084: Reuse a folded filter when the new plan has no Project
above the join") {
+ // The `(np, cp: Project)` arm builds the merged Project from subquery3's
join output, and has
+ // to append the reused filter there.
+ checkReusedFilterAfterPostJoinFold(selectInSubquery3 = false)
+ }
+
+ test("SPARK-60084: Do not reuse a computed column as a propagated filter") {
+ // subquery3 computes `b > 2` as a column and filters on the same
expression. The column lands
+ // in the Project under the merged Filter, but the Filter's OR doesn't
keep its rows, so
+ // subquery3 needs a filter alias of its own.
+ val subquery1 = ScalarSubquery(testRelation.where($"a" <
1).groupBy()(sum($"a").as("sum_a")))
+ val subquery2 = ScalarSubquery(testRelation.where($"a" >
5).groupBy()(max($"a").as("max_a")))
+ val subquery3 = ScalarSubquery(
+ testRelation.select($"a", $"b", ($"b" > 2).as("flag")).where($"b" > 2)
+ .groupBy()(count($"a").as("cnt")))
+ val originalQuery = testRelation.select(subquery1, subquery2, subquery3)
+
+ val f0Alias = Alias($"a" > 5, "propagatedFilter_0")()
+ val f0 = f0Alias.toAttribute
+ val f1Alias = Alias($"a" < 1, "propagatedFilter_1")()
+ val f1 = f1Alias.toAttribute
+ val f2Alias = Alias($"b" > 2, "propagatedFilter_2")()
+ val f2 = f2Alias.toAttribute
+ val mergedSubquery = testRelation
+ .select(testRelation.output ++ Seq(f0Alias, f1Alias, ($"b" >
2).as("flag"), f2Alias): _*)
+ .where(Or(Or(f0, f1), f2))
+ .groupBy()(
+ sum($"a", Some(f1)).as("sum_a"),
+ max($"a", Some(f0)).as("max_a"),
+ count($"a", Some(f2)).as("cnt"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_a"), $"sum_a",
+ Literal("max_a"), $"max_a",
+ Literal("cnt"), $"cnt"
+ )).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1),
+ extractorExpression(0, analyzedMergedSubquery.output, 2)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+
withSQLConf(SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key ->
"true") {
+ comparePlans(Optimize.execute(originalQuery.analyze),
correctAnswer.analyze)
+ }
+ }
+
// ---- SPARK-40259: generic DSv2 scan merge ----
private val v2Table = new TestV2Table(StructType(Seq(
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/PlanMergingSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/PlanMergingSuite.scala
index 127bef8e2e91..b1c430404a0f 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/PlanMergingSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/PlanMergingSuite.scala
@@ -439,4 +439,44 @@ class PlanMergingSuite extends SharedSparkSession
}
}
}
+
+ test("SPARK-60084: Reuse a filter that an earlier merge folded into a
post-join filter") {
+ // Merging the 2nd subquery into the 1st aliases `l.b = 2` below the join,
and the 2nd
+ // subquery's post-join filter folds that alias into a new one, so only
the new alias is
+ // carried up to the aggregate. The 3rd subquery then reuses the `l.b = 2`
alias on its own.
+ Seq(false, true).foreach { enableAQE =>
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> enableAQE.toString,
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key ->
"true",
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_THROUGH_JOIN_ENABLED.key ->
"true",
+ // ObjectSerializerPruning produces different scan shapes depending on
whether a Filter is
+ // present. Disabling the rule makes both scans identical so
PlanMerger can merge them.
+ SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
+ "org.apache.spark.sql.catalyst.optimizer.ObjectSerializerPruning") {
+ val df = sql(
+ """
+ |SELECT
+ | (SELECT sum(l.a)
+ | FROM testData2 l LEFT JOIN testData2 r ON l.a = r.a WHERE l.b
= 1),
+ | (SELECT max(l.a)
+ | FROM testData2 l LEFT JOIN testData2 r ON l.a = r.a
+ | WHERE l.b = 2 AND coalesce(r.b, 3) > 1),
+ | (SELECT count(*)
+ | FROM testData2 l LEFT JOIN testData2 r ON l.a = r.a WHERE l.b
= 2)
+ """.stripMargin)
+
+ checkAnswer(df, Row(12, 3, 6) :: Nil)
+
+ val plan = df.queryExecution.executedPlan
+ val subqueryIds = collectWithSubqueries(plan) { case s: SubqueryExec
=> s.id }
+ val reusedSubqueryIds = collectWithSubqueries(plan) {
+ case rs: ReusedSubqueryExec => rs.child.id
+ }
+
+ assert(subqueryIds.size == 1, "Missing or unexpected SubqueryExec in
the plan")
+ assert(reusedSubqueryIds.size == 2,
+ "Missing or unexpected ReusedSubqueryExec in the plan")
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
