You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by pt...@apache.org on 2026/05/04 18:47:13 UTC
(spark) branch branch-4.2 updated: [SPARK-56570][SQL][FOLLOWUP] `PlanMerger` correctness fix and code cleanup followup
This is an automated email from the ASF dual-hosted git repository.
peter-toth 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 359e444e2b63 [SPARK-56570][SQL][FOLLOWUP] `PlanMerger` correctness fix and code cleanup followup
359e444e2b63 is described below
commit 359e444e2b63a8cb5392c8205171bd76fb928533
Author: Peter Toth <pe...@gmail.com>
AuthorDate: Mon May 4 20:44:27 2026 +0200
[SPARK-56570][SQL][FOLLOWUP] `PlanMerger` correctness fix and code cleanup followup
### What changes were proposed in this pull request?
This is a follow-up to https://github.com/apache/spark/pull/55482 and contains four bug fixes and two small cleanups in `PlanMerger`:
Bug fixes in `PlanMerger`:
1. Tagged `(Filter, Filter)` reuse preserves `mergedChild`'s appended columns:
When the reuse check finds an existing `propagatedFilter` alias, the branch now rebuilds the Filter over `mergedChild` (via `cp.withNewChildren(Seq(mergedChild))`) instead of returning `cp` unchanged. If the recursion extended `cp.child`'s output with new columns (e.g. a computed `d = a + b` from a user Project below the Filter), returning `cp` would drop those columns while `npMapping` still pointed into them, leaving the enclosing `Aggregate` with unresolved references.
2. `(np: Filter, cp)` does not duplicate a `cpFilter` already present in `mergedChild`:
`cpFilter`, when set, was produced by a deeper `(np, cp: Filter)` (or `(Join, Join)` pass-through) and is already part of `mergedChild`'s output. Appending it a second time via `++ cpFilter.toSeq` duplicated the attribute in the outer Project's projectList.
3. `(np, cp: Filter)` does not duplicate an `npFilter` already present in `mergedChild`:
Symmetric to 2. on the `np` side.
4. `(np, cp: Filter)` with a `MERGED_FILTER_TAG`-tagged `cp` drops the tagged Filter:
`cp`'s condition is `OR(pf_0, pf_1, ...)` and `cp`'s aggregate expressions already carry individual `FILTER (WHERE pf_i)` clauses. Synthesising a new `propagatedFilter_X = OR(pf_0, pf_1, ...)` would just add `FILTER AND(OR(...), pf_i)` wrapping upstream (simplifying to `FILTER pf_i`) plus a redundant alias in the Project. The branch now drops `cp`'s Filter and returns `cpFilter = None` so `cp`'s aggregates are left untouched.
Cleanups in `PlanMerger.merge()`:
- Unify the local variable name to `newMergedPlan` across all three branches (was `newMergedPlan` in one and `newMergePlan` in the other two) -- matches the `MergedPlan` case class name.
- Replace `cache(i).merged` with `mp.merged`; `mp` and `cache(i)` are the same object inside the `collectFirst` pattern.
### Why are the changes needed?
Fix 1. is a correctness bug. Fixes 2-4. are plan-shape bugs that produce duplicated attributes or redundant `OR`-of-propagated-filter aliases in the merged plan. The cleanups are minor readability improvements.
### Does this PR introduce _any_ user-facing change?
No. All changes are internal to the optimizer; they produce cleaner merged plans for queries that `MergeSubplans` already handled.
### How was this patch tested?
Four new tests in `MergeSubplansSuite`, one per fix:
- `(np: Filter, cp)` does not duplicate a `cpFilter` already present in mergedChild -- exercises 2. via a `Join` with a `Filter` on the right child, routing a `cpFilter` up through `(Join, Join)` so that `mergedChild.output` already contains the attribute the branch used to re-append.
- `(np, cp: Filter)` does not duplicate an `npFilter` already present in mergedChild -- exercises 3., mirror shape on the `np` side.
- tagged `(Filter, Filter)` reuse must keep mergedChild's appended columns -- exercises 1. with three subqueries (sq1/sq2 create the tagged structure; sq3's Filter sits above a user Project introducing `d = a + b`, so the `(Filter, Filter)` tagged recursion extends `mergedChild` with `d`).
- `(np, cp: Filter)` drops a tagged `cp` Filter without synthesising a redundant alias -- exercises 4. with three subqueries (sq1/sq2 create the tagged structure; sq3 has no filter).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.7
Closes #55659 from peter-toth/SPARK-56570-planmerger-code-cleanup-followup.
Authored-by: Peter Toth <pe...@gmail.com>
Signed-off-by: Peter Toth <pe...@gmail.com>
(cherry picked from commit 3ae7da7d1fd3af21824c213d3c1986f58b0a0f77)
Signed-off-by: Peter Toth <pe...@gmail.com>
---
.../spark/sql/catalyst/optimizer/PlanMerger.scala | 46 +++--
.../catalyst/optimizer/MergeSubplansSuite.scala | 197 +++++++++++++++++++++
2 files changed, 226 insertions(+), 17 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PlanMerger.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PlanMerger.scala
index a85fad2f758c..1c43f91cee9d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PlanMerger.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PlanMerger.scala
@@ -161,27 +161,27 @@ class PlanMerger(
// `ReusedSubqueryExec` rule can handle them without extracting the plans to CTEs.
// But, when a non-subquery subplan is identical to a cached plan we need to mark the plan
// `merged` and so extract it to a CTE later.
- val newMergedPlan = MergedPlan(mp.plan, cache(i).merged || !subqueryPlan)
+ val newMergedPlan = MergedPlan(mp.plan, mp.merged || !subqueryPlan)
cache(i) = newMergedPlan
val outputMap = AttributeMap(plan.output.zipWithIndex)
MergeResult(newMergedPlan, i, outputMap)
}.orElse {
tryMergePlans(plan, mp.plan, false).collect {
case TryMergeResult(mergedPlan, npMapping, None, None) =>
- val newMergePlan = MergedPlan(mergedPlan, true)
- cache(i) = newMergePlan
+ val newMergedPlan = MergedPlan(mergedPlan, true)
+ cache(i) = newMergedPlan
val outputMap = AttributeMap(npMapping.iterator.map { case (origAttr, mergedAttr) =>
origAttr -> mergedPlan.output.indexWhere(_.exprId == mergedAttr.exprId)
}.toSeq)
- MergeResult(newMergePlan, i, outputMap)
+ MergeResult(newMergedPlan, i, outputMap)
}
}
case _ => None
}).getOrElse {
- val newMergePlan = MergedPlan(plan, false)
- cache += newMergePlan
+ val newMergedPlan = MergedPlan(plan, false)
+ cache += newMergedPlan
val outputMap = AttributeMap(plan.output.zipWithIndex)
- MergeResult(newMergePlan, cache.length - 1, outputMap)
+ MergeResult(newMergedPlan, cache.length - 1, outputMap)
}
}
@@ -355,7 +355,8 @@ class PlanMerger(
}
existingNPFilter match {
case Some(reusedFilter) =>
- Some(TryMergeResult(cp, npMapping, Some((reusedFilter, false)), None))
+ val newFilter = cp.withNewChildren(Seq(mergedChild))
+ Some(TryMergeResult(newFilter, npMapping, Some((reusedFilter, false)), None))
case None =>
val newNPFilterAlias =
Alias(newNPCondition, s"propagatedFilter_${PlanMerger.newId}")()
@@ -410,7 +411,7 @@ class PlanMerger(
Alias(newNPCondition, s"propagatedFilter_${PlanMerger.newId}")()
val newNPFilter = newNPFilterAlias.toAttribute
val project = Project(
- mergedChild.output.toList ++ Seq(newNPFilterAlias) ++ cpFilter.toSeq,
+ mergedChild.output.toList :+ newNPFilterAlias,
mergedChild)
TryMergeResult(project, npMapping, Some((newNPFilter, true)), cpFilter)
}
@@ -421,14 +422,25 @@ class PlanMerger(
// symmetricFilterPropagationEnabled.
case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter)
if npFilter.isEmpty || symmetricFilterPropagationEnabled =>
- val newCPCondition = cpFilter.fold(cp.condition)(And(_, cp.condition))
- val newCPFilterAlias =
- Alias(newCPCondition, s"propagatedFilter_${PlanMerger.newId}")()
- val newCPFilter = newCPFilterAlias.toAttribute
- val project = Project(
- mergedChild.output.toList ++ npFilter.map(_._1).toSeq ++ Seq(newCPFilterAlias),
- mergedChild)
- TryMergeResult(project, npMapping, npFilter, Some(newCPFilter))
+ if (cp.getTagValue(PlanMerger.MERGED_FILTER_TAG).isDefined) {
+ // cp is a previously-merged Filter: its condition is `OR(pf_0, pf_1, ...)` and cp's
+ // aggregate expressions already carry individual `FILTER (WHERE pf_i)` clauses that
+ // restrict each aggregation to its originating side. Synthesising a new cpFilter
+ // alias for cp.condition would just produce `FILTER AND(OR(pf_0, pf_1, ...), pf_i)`
+ // upstream, which simplifies to `FILTER pf_i` -- wasted work and plan bloat.
+ // Drop cp's Filter and let the recursion's result flow up with cpFilter = None so
+ // cp's aggregates are left untouched.
+ TryMergeResult(mergedChild, npMapping, npFilter, None)
+ } else {
+ val newCPCondition = cpFilter.fold(cp.condition)(And(_, cp.condition))
+ val newCPFilterAlias =
+ Alias(newCPCondition, s"propagatedFilter_${PlanMerger.newId}")()
+ val newCPFilter = newCPFilterAlias.toAttribute
+ val project = Project(
+ mergedChild.output.toList :+ newCPFilterAlias,
+ mergedChild)
+ TryMergeResult(project, npMapping, npFilter, Some(newCPFilter))
+ }
}
case (np: Join, cp: Join) if np.joinType == cp.joinType && np.hint == cp.hint =>
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/MergeSubplansSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/MergeSubplansSuite.scala
index c9e5db581059..dfa17f926c5a 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/MergeSubplansSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/MergeSubplansSuite.scala
@@ -1516,6 +1516,203 @@ class MergeSubplansSuite extends PlanTest {
}
}
+ test("SPARK-56570: `(np: Filter, cp)` does not duplicate a cpFilter already present in " +
+ "mergedChild") {
+ // The `(np: Filter, cp)` create-new branch is only reached with a non-None recursion
+ // `cpFilter` when cp has a shape that lets filter propagation bubble a cpFilter up through
+ // the recursion without being consumed by an `Aggregate`. A Join with a Filter on one side
+ // does this:
+ // - sq1 (cp): Aggregate -> Join(testRelation, Filter(e < 5, testRelation2), a = d).
+ // - sq2 (np): Aggregate -> Filter(a > 1) -> Join(testRelation, testRelation2, a = d).
+ // At `(Agg, Agg)` children, the pair is `(Filter, Join)`. `(Filter, cp)` fires, peels np's
+ // Filter and recurses `(Join, Join)`. The right-child recursion hits `(np, cp: Filter)`,
+ // creates `propagatedFilter_0` for `e < 5`, and `(Join, Join)` propagates that as cpFilter
+ // all the way back to the outer `(Filter, cp)` case. `mergedChild` at that point is a Join
+ // whose output already contains the `propagatedFilter_0` attribute.
+ val subquery1 = ScalarSubquery(
+ testRelation.join(testRelation2.where($"e" < 5), Inner, Some($"a" === $"d"))
+ .groupBy()(max($"a").as("max_a")))
+ val subquery2 = ScalarSubquery(
+ testRelation.join(testRelation2, Inner, Some($"a" === $"d")).where($"a" > 1)
+ .groupBy()(sum($"a").as("sum_a")))
+ val originalQuery = testRelation.select(subquery1, subquery2)
+
+ val f0Alias = Alias($"e" < 5, "propagatedFilter_0")()
+ val f0 = f0Alias.toAttribute
+ val f1Alias = Alias($"a" > 1, "propagatedFilter_1")()
+ val f1 = f1Alias.toAttribute
+ val innerProject = testRelation2.select(testRelation2.output ++ Seq(f0Alias): _*)
+ val joinNode = testRelation.join(innerProject, Inner, Some($"a" === $"d"))
+ val mergedSubquery = joinNode
+ .select(joinNode.output ++ Seq(f1Alias): _*)
+ .groupBy()(
+ max($"a", Some(f0)).as("max_a"),
+ sum($"a", Some(f1)).as("sum_a"))
+ .select(CreateNamedStruct(Seq(
+ Literal("max_a"), $"max_a",
+ Literal("sum_a"), $"sum_a"
+ )).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1)),
+ 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-56570: `(np, cp: Filter)` does not duplicate an npFilter already present in " +
+ "mergedChild") {
+ // Mirror of the previous test: the `(np, cp: Filter)` create-new branch is only reached with
+ // a non-None recursion `npFilter` when np has a shape that lets filter propagation bubble up
+ // through the recursion. A Join with a Filter on one side does this:
+ // - sq1 (cp): Aggregate -> Filter(a > 1) -> Join(testRelation, testRelation2, a = d).
+ // - sq2 (np): Aggregate -> Join(testRelation, Filter(e < 5, testRelation2), a = d).
+ // At `(Agg, Agg)` children, the pair is `(Join, Filter)`. `(np, cp: Filter)` fires, peels
+ // cp's Filter and recurses `(Join, Join)`. The right-child recursion hits `(np: Filter, cp)`,
+ // creates `propagatedFilter_0` for `e < 5`, and `(Join, Join)` propagates that as npFilter
+ // all the way back to the outer `(np, cp: Filter)` case. `mergedChild` at that point is a
+ // Join whose output already contains the `propagatedFilter_0` attribute.
+ val subquery1 = ScalarSubquery(
+ testRelation.join(testRelation2, Inner, Some($"a" === $"d")).where($"a" > 1)
+ .groupBy()(max($"a").as("max_a")))
+ val subquery2 = ScalarSubquery(
+ testRelation.join(testRelation2.where($"e" < 5), Inner, Some($"a" === $"d"))
+ .groupBy()(sum($"a").as("sum_a")))
+ val originalQuery = testRelation.select(subquery1, subquery2)
+
+ val f0Alias = Alias($"e" < 5, "propagatedFilter_0")()
+ val f0 = f0Alias.toAttribute
+ val f1Alias = Alias($"a" > 1, "propagatedFilter_1")()
+ val f1 = f1Alias.toAttribute
+ val innerProject = testRelation2.select(testRelation2.output ++ Seq(f0Alias): _*)
+ val joinNode = testRelation.join(innerProject, Inner, Some($"a" === $"d"))
+ val mergedSubquery = joinNode
+ .select(joinNode.output ++ Seq(f1Alias): _*)
+ .groupBy()(
+ max($"a", Some(f1)).as("max_a"),
+ sum($"a", Some(f0)).as("sum_a"))
+ .select(CreateNamedStruct(Seq(
+ Literal("max_a"), $"max_a",
+ Literal("sum_a"), $"sum_a"
+ )).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1)),
+ 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-56570: tagged `(Filter, Filter)` reuse must keep mergedChild's appended columns") {
+ // Round 1-2 build a tagged Filter (condition `OR(pf_0=(b<5), pf_1=(a>1))`) over a tagged
+ // Project carrying both `propagatedFilter_*` aliases.
+ // Round 3 merges a subplan whose Filter sits *above* a user Project introducing
+ // `d = a + b`:
+ // sq3 = Aggregate(sum(d)) -> Filter(a>1) -> Project([d=(a+b), a, b, c]) -> testRelation.
+ // At `(Agg, Agg)` children the pair is `(Filter(a>1) -> Project, Filter[tagged] -> Project)`
+ // -- neither side is a Project at this level, so `(Filter, Filter)` tagged fires directly
+ // and recurses on the children `(Project[d,a,b,c], Project[tagged][a,b,c,pf_0,pf_1])`. That
+ // recursion's `(Project, Project)` case builds `Project([a,b,c,pf_0_alias,pf_1_alias,d], t)`
+ // -- `mergedChild` now carries a column (`d`) that `cp.child` doesn't. The reuse check
+ // finds `pf_1` already matches sq3's `(a > 1)`, so the tagged-reuse branch fires and must
+ // rebuild the Filter over `mergedChild` so that `d` stays visible to the enclosing
+ // Aggregate's `sum(d)`.
+ val subquery1 = ScalarSubquery(testRelation.where($"a" > 1).groupBy()(max($"a").as("max_a")))
+ val subquery2 = ScalarSubquery(testRelation.where($"b" < 5).groupBy()(min($"b").as("min_b")))
+ val subquery3 = ScalarSubquery(
+ testRelation
+ .select(($"a" + $"b").as("d"), $"a", $"b", $"c")
+ .where($"a" > 1)
+ .groupBy()(sum($"d").as("sum_d")))
+ val originalQuery = testRelation.select(subquery1, subquery2, subquery3)
+
+ val f0Alias = Alias($"b" < 5, "propagatedFilter_0")()
+ val f0 = f0Alias.toAttribute
+ val f1Alias = Alias($"a" > 1, "propagatedFilter_1")()
+ val f1 = f1Alias.toAttribute
+ val dAlias = Alias($"a" + $"b", "d")()
+ val d = dAlias.toAttribute
+ val innerProject = testRelation.select(testRelation.output ++ Seq(f0Alias, f1Alias, dAlias): _*)
+ val mergedSubquery = innerProject
+ .where(Or(f0, f1))
+ .groupBy()(
+ max($"a", Some(f1)).as("max_a"),
+ min($"b", Some(f0)).as("min_b"),
+ sum(d, Some(f1)).as("sum_d"))
+ .select(CreateNamedStruct(Seq(
+ Literal("max_a"), $"max_a",
+ Literal("min_b"), $"min_b",
+ Literal("sum_d"), $"sum_d"
+ )).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)
+ }
+ }
+
+ test("SPARK-56570: `(np, cp: Filter)` drops a tagged cp Filter without synthesising a " +
+ "redundant alias") {
+ // Round 1+2: sq1 and sq2 merge via `(Filter, Filter)` first-time, creating a tagged Filter
+ // (condition `OR(pf_0=(b<5), pf_1=(a>1))`) over a tagged Project carrying both aliases.
+ // cp's aggregates are `[max(a) FILTER pf_1, min(b) FILTER pf_0]`.
+ // Round 3: sq3 has no Filter, so `(np, cp: Filter)` with cp tagged fires. Synthesising a new
+ // `propagatedFilter_2 = OR(pf_0, pf_1)` would leave the enclosing Aggregate wrapping cp's
+ // already-filtered aggregates with `FILTER AND(OR(pf_0, pf_1), pf_i)` (which simplifies to
+ // `FILTER pf_i`) -- wasted work and plan bloat. Dropping cp's Filter returns the recursion's
+ // Project unchanged, leaves cp's per-side FILTER clauses untouched, and leaves the base
+ // unrestricted for np's unfiltered aggregate.
+ val subquery1 = ScalarSubquery(testRelation.where($"a" > 1).groupBy()(max($"a").as("max_a")))
+ val subquery2 = ScalarSubquery(testRelation.where($"b" < 5).groupBy()(min($"b").as("min_b")))
+ val subquery3 = ScalarSubquery(testRelation.groupBy()(sum($"a").as("sum_a")))
+ val originalQuery = testRelation.select(subquery1, subquery2, subquery3)
+
+ val f0Alias = Alias($"b" < 5, "propagatedFilter_0")()
+ val f0 = f0Alias.toAttribute
+ val f1Alias = Alias($"a" > 1, "propagatedFilter_1")()
+ val f1 = f1Alias.toAttribute
+ val mergedSubquery = testRelation
+ .select(testRelation.output ++ Seq(f0Alias, f1Alias): _*)
+ .groupBy()(
+ max($"a", Some(f1)).as("max_a"),
+ min($"b", Some(f0)).as("min_b"),
+ sum($"a").as("sum_a"))
+ .select(CreateNamedStruct(Seq(
+ Literal("max_a"), $"max_a",
+ Literal("min_b"), $"min_b",
+ Literal("sum_a"), $"sum_a"
+ )).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)
+ }
+ }
+
test("SPARK-40193: Merge non-grouping subqueries where one aggregate already carries a " +
"FILTER clause") {
val subquery1 = ScalarSubquery(testRelation.groupBy()(max($"a").as("max_a")))
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org