You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "peter-toth (via GitHub)" <gi...@apache.org> on 2026/04/22 18:00:36 UTC
[PR] [SPARK-56570][SQL] `PlanMerger` code cleanup [spark]
peter-toth opened a new pull request, #55482:
URL: https://github.com/apache/spark/pull/55482
### What changes were proposed in this pull request?
Remove `cachedPlanMapping` / `cpMapping` from `PlanMerger` by preserving ExprIds when wrapping cached-plan expressions in `mergeNamedExpressions`.
Previously, wrapping a cached expression in `Alias(If(filter, expr, null))` generated a fresh `ExprId`, making parent nodes that referenced the original attribute by ExprId stale. A `cachedPlanMapping` field was threaded through `TryMergeResult` and `mergeNamedExpressions` to remap those references at every level of `tryMergePlans`.
This PR eliminates the need for that mapping by passing `exprId = ce.toAttribute.exprId` to the wrapping `Alias`, preserving the original `ExprId`. Since the mapping is now always identity, `TryMergeResult.cachedPlanMapping` and the `cachedPlanMapping` parameter of `mergeNamedExpressions` are removed entirely, along with all `cpMapping` threading throughout `tryMergePlans`.
Additional cleanups:
- `mappedCPCondition` locals replaced with direct `cp.condition` references (no remapping needed after cpMapping removal).
- `mappedCPGroupingExpression` assigned directly from `cp.groupingExpressions`.
- Cached-expression wrapping loop limited to `cachedPlanExpressions.size` entries to avoid accidentally wrapping newly-appended new-plan expressions.
- Cp-expression wrapping simplified to a single `case Alias(child, _) if !child.isInstanceOf[Attribute]` match.
### Why are the changes needed?
`cachedPlanMapping` was purely mechanical bookkeeping to compensate for `Alias(...)()` generating fresh ExprIds. Preserving the original ExprId makes the mapping a no-op everywhere, so it can be removed. This simplifies `TryMergeResult` from 5 fields to 4, changes `mergeNamedExpressions` to return a pair instead of a triple, and removes cpMapping threading from all branches of `tryMergePlans`.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing `MergeSubplansSuite` unit tests.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Sonnet 4.6
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56570][SQL] `PlanMerger` code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #55482:
URL: https://github.com/apache/spark/pull/55482#issuecomment-4304425617
@cloud-fan, I've cherry picked the test from https://github.com/apache/spark/pull/55500 in https://github.com/apache/spark/pull/55482/commits/31621fea1043a0a10f66d2d7a53dd0e05c99fc50.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56570][SQL] `PlanMerger` code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #55482:
URL: https://github.com/apache/spark/pull/55482#issuecomment-4303005762
cc @dongjoon-hyun , this is a minor code simplification after https://github.com/apache/spark/pull/55298.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56570][SQL] `PlanMerger` correctness fix and code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #55482:
URL: https://github.com/apache/spark/pull/55482#issuecomment-4311569515
Thanks for the review again!
Merged to `master` (4.2.0).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56570][SQL] `PlanMerger` code cleanup [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #55482:
URL: https://github.com/apache/spark/pull/55482#discussion_r3132260608
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PlanMerger.scala:
##########
@@ -244,57 +238,52 @@ class PlanMerger(
(newPlan, cachedPlan) match {
case (np: Project, cp: Project) =>
tryMergePlans(np.child, cp.child, filterPropagationSupported).map {
- case TryMergeResult(mergedChild, npMapping, cpMapping, npFilter, cpFilter) =>
- val (mergedProjectList, newNPMapping, newCPMapping) =
- mergeNamedExpressions(np.projectList, cp.projectList, npMapping, cpMapping,
- npFilter, cpFilter)
- TryMergeResult(Project(mergedProjectList, mergedChild), newNPMapping, newCPMapping,
- npFilter, cpFilter)
+ case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter) =>
+ val (mergedProjectList, newNPMapping) =
+ mergeNamedExpressions(np.projectList, cp.projectList, npMapping, npFilter, cpFilter)
+ TryMergeResult(Project(mergedProjectList, mergedChild), newNPMapping, npFilter,
+ cpFilter)
}
case (np, cp: Project) =>
tryMergePlans(np, cp.child, filterPropagationSupported).map {
- case TryMergeResult(mergedChild, npMapping, cpMapping, npFilter, cpFilter) =>
- val (mergedProjectList, newNPMapping, newCPMapping) =
- mergeNamedExpressions(np.output, cp.projectList, npMapping, cpMapping, npFilter,
- cpFilter)
- TryMergeResult(Project(mergedProjectList, mergedChild), newNPMapping, newCPMapping,
- npFilter, cpFilter)
+ case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter) =>
+ val (mergedProjectList, newNPMapping) =
+ mergeNamedExpressions(np.output, cp.projectList, npMapping, npFilter, cpFilter)
+ TryMergeResult(Project(mergedProjectList, mergedChild), newNPMapping, npFilter,
+ cpFilter)
}
case (np: Project, cp) =>
tryMergePlans(np.child, cp, filterPropagationSupported).map {
- case TryMergeResult(mergedChild, npMapping, cpMapping, npFilter, cpFilter) =>
- val (mergedProjectList, newNPMapping, newCPMapping) =
- mergeNamedExpressions(np.projectList, cp.output, npMapping, cpMapping, npFilter,
- cpFilter)
- TryMergeResult(Project(mergedProjectList, mergedChild), newNPMapping, newCPMapping,
- npFilter, cpFilter)
+ case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter) =>
+ val (mergedProjectList, newNPMapping) =
+ mergeNamedExpressions(np.projectList, cp.output, npMapping, npFilter, cpFilter)
+ TryMergeResult(Project(mergedProjectList, mergedChild), newNPMapping, npFilter,
+ cpFilter)
}
case (np: Aggregate, cp: Aggregate) if supportedAggregateMerge(np, cp) =>
// Filter propagation into the aggregate is only safe when there is no grouping.
val childFilterPropagationSupported = filterPropagationEnabled &&
np.groupingExpressions.isEmpty && cp.groupingExpressions.isEmpty
tryMergePlans(np.child, cp.child, childFilterPropagationSupported).flatMap {
- case TryMergeResult(mergedChild, npMapping, cpMapping, None, None) =>
+ case TryMergeResult(mergedChild, npMapping, None, None) =>
val mappedNPGroupingExpression =
np.groupingExpressions.map(mapAttributes(_, npMapping))
- val mappedCPGroupingExpression =
- cp.groupingExpressions.map(mapAttributes(_, cpMapping))
+ val mappedCPGroupingExpression = cp.groupingExpressions
// Order of grouping expression does matter as merging different grouping orders can
// introduce "extra" shuffles/sorts that might not present in all of the original
// subqueries.
if (mappedNPGroupingExpression.map(_.canonicalized) ==
mappedCPGroupingExpression.map(_.canonicalized)) {
- val (mergedAggregateExpressions, newNPMapping, newCPMapping) =
- mergeNamedExpressions(np.aggregateExpressions, cp.aggregateExpressions, npMapping,
- cpMapping)
+ val (mergedAggregateExpressions, newNPMapping) =
+ mergeNamedExpressions(np.aggregateExpressions, cp.aggregateExpressions, npMapping)
val mergedPlan =
Aggregate(mappedCPGroupingExpression, mergedAggregateExpressions, mergedChild)
Review Comment:
After dropping `cpMapping`, the `mapped` prefix no longer reflects what this local holds — it's just `cp.groupingExpressions`. Same for the reference used to build the merged `Aggregate`.
```suggestion
val cpGroupingExpressions = cp.groupingExpressions
// Order of grouping expression does matter as merging different grouping orders can
// introduce "extra" shuffles/sorts that might not present in all of the original
// subqueries.
if (mappedNPGroupingExpression.map(_.canonicalized) ==
cpGroupingExpressions.map(_.canonicalized)) {
val (mergedAggregateExpressions, newNPMapping) =
mergeNamedExpressions(np.aggregateExpressions, cp.aggregateExpressions, npMapping)
val mergedPlan =
Aggregate(cpGroupingExpressions, mergedAggregateExpressions, mergedChild)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56570][SQL] `PlanMerger` correctness fix and code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55482:
URL: https://github.com/apache/spark/pull/55482#discussion_r3132395324
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PlanMerger.scala:
##########
@@ -244,57 +238,52 @@ class PlanMerger(
(newPlan, cachedPlan) match {
case (np: Project, cp: Project) =>
tryMergePlans(np.child, cp.child, filterPropagationSupported).map {
- case TryMergeResult(mergedChild, npMapping, cpMapping, npFilter, cpFilter) =>
- val (mergedProjectList, newNPMapping, newCPMapping) =
- mergeNamedExpressions(np.projectList, cp.projectList, npMapping, cpMapping,
- npFilter, cpFilter)
- TryMergeResult(Project(mergedProjectList, mergedChild), newNPMapping, newCPMapping,
- npFilter, cpFilter)
+ case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter) =>
+ val (mergedProjectList, newNPMapping) =
+ mergeNamedExpressions(np.projectList, cp.projectList, npMapping, npFilter, cpFilter)
+ TryMergeResult(Project(mergedProjectList, mergedChild), newNPMapping, npFilter,
+ cpFilter)
}
case (np, cp: Project) =>
tryMergePlans(np, cp.child, filterPropagationSupported).map {
- case TryMergeResult(mergedChild, npMapping, cpMapping, npFilter, cpFilter) =>
- val (mergedProjectList, newNPMapping, newCPMapping) =
- mergeNamedExpressions(np.output, cp.projectList, npMapping, cpMapping, npFilter,
- cpFilter)
- TryMergeResult(Project(mergedProjectList, mergedChild), newNPMapping, newCPMapping,
- npFilter, cpFilter)
+ case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter) =>
+ val (mergedProjectList, newNPMapping) =
+ mergeNamedExpressions(np.output, cp.projectList, npMapping, npFilter, cpFilter)
+ TryMergeResult(Project(mergedProjectList, mergedChild), newNPMapping, npFilter,
+ cpFilter)
}
case (np: Project, cp) =>
tryMergePlans(np.child, cp, filterPropagationSupported).map {
- case TryMergeResult(mergedChild, npMapping, cpMapping, npFilter, cpFilter) =>
- val (mergedProjectList, newNPMapping, newCPMapping) =
- mergeNamedExpressions(np.projectList, cp.output, npMapping, cpMapping, npFilter,
- cpFilter)
- TryMergeResult(Project(mergedProjectList, mergedChild), newNPMapping, newCPMapping,
- npFilter, cpFilter)
+ case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter) =>
+ val (mergedProjectList, newNPMapping) =
+ mergeNamedExpressions(np.projectList, cp.output, npMapping, npFilter, cpFilter)
+ TryMergeResult(Project(mergedProjectList, mergedChild), newNPMapping, npFilter,
+ cpFilter)
}
case (np: Aggregate, cp: Aggregate) if supportedAggregateMerge(np, cp) =>
// Filter propagation into the aggregate is only safe when there is no grouping.
val childFilterPropagationSupported = filterPropagationEnabled &&
np.groupingExpressions.isEmpty && cp.groupingExpressions.isEmpty
tryMergePlans(np.child, cp.child, childFilterPropagationSupported).flatMap {
- case TryMergeResult(mergedChild, npMapping, cpMapping, None, None) =>
+ case TryMergeResult(mergedChild, npMapping, None, None) =>
val mappedNPGroupingExpression =
np.groupingExpressions.map(mapAttributes(_, npMapping))
- val mappedCPGroupingExpression =
- cp.groupingExpressions.map(mapAttributes(_, cpMapping))
+ val mappedCPGroupingExpression = cp.groupingExpressions
// Order of grouping expression does matter as merging different grouping orders can
// introduce "extra" shuffles/sorts that might not present in all of the original
// subqueries.
if (mappedNPGroupingExpression.map(_.canonicalized) ==
mappedCPGroupingExpression.map(_.canonicalized)) {
- val (mergedAggregateExpressions, newNPMapping, newCPMapping) =
- mergeNamedExpressions(np.aggregateExpressions, cp.aggregateExpressions, npMapping,
- cpMapping)
+ val (mergedAggregateExpressions, newNPMapping) =
+ mergeNamedExpressions(np.aggregateExpressions, cp.aggregateExpressions, npMapping)
val mergedPlan =
Aggregate(mappedCPGroupingExpression, mergedAggregateExpressions, mergedChild)
Review Comment:
Thanks, fixed in https://github.com/apache/spark/pull/55482/commits/79184550e43383a8507206249bca5738c5103b8c.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PlanMerger.scala:
##########
@@ -523,35 +500,27 @@ class PlanMerger(
ne.toAttribute -> resultAttr
})
- // Wrap unmatched cached expressions with the cached plan's filter so they are only computed
- // for rows that belong to the cached plan side. Plain attribute references are not wrapped.
- // Record each attr rewrite in the cached plan map so ancestor nodes can remap their stale
- // references.
- val newCPMapping = AttributeMap(cachedPlanFilter.toSeq.flatMap { f =>
- mergedExpressions.zipWithIndex.flatMap {
- case (ce, i) if !matchedCachedIndices.contains(i) =>
- val withoutAlias = ce match {
- case Alias(child, _) => child
- case e => e
- }
- // Plain attribute references are not wrapped: no remapping entry needed.
- Option.when(!withoutAlias.isInstanceOf[Attribute]) {
- val newAlias =
- Alias(If(f, withoutAlias, Literal(null, withoutAlias.dataType)), ce.name)()
- mergedExpressions(i) = newAlias
- ce.toAttribute -> newAlias.toAttribute
- }
- case _ => None
+ // Wrap unmatched cached expressions with the cached plan's filter so they are only computed for
+ // rows that belong to the cached plan side. Plain attribute references are not wrapped.
+ cachedPlanFilter.foreach { f =>
+ for (i <- 0 until cachedPlanExpressions.size if !matchedCachedIndices.contains(i)) {
+ mergedExpressions(i) match {
+ case ce @ Alias(child, _) if !child.isInstanceOf[Attribute] =>
+ mergedExpressions(i) =
+ Alias(If(f, child, Literal(null, child.dataType)), ce.name)(
+ exprId = ce.toAttribute.exprId)
Review Comment:
Added in https://github.com/apache/spark/pull/55482/commits/79184550e43383a8507206249bca5738c5103b8c.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56570][SQL] `PlanMerger` correctness fix and code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #55482:
URL: https://github.com/apache/spark/pull/55482#issuecomment-4306252858
Thank you @dongjoon-hyun and @cloud-fan. Initially I thought this is just a cosmetic issue (double `If` wrapping), but adjusted the PR title and description to reflect the latest findings.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56570][SQL] `PlanMerger` correctness fix and code cleanup [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth closed pull request #55482: [SPARK-56570][SQL] `PlanMerger` correctness fix and code cleanup
URL: https://github.com/apache/spark/pull/55482
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org