You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "cloud-fan (via GitHub)" <gi...@apache.org> on 2026/08/07 19:52:06 UTC
[PR] [SPARK-58210][SQL][FOLLOWUP] Extend CombineAdjacentAggregation to partial merge [spark]
cloud-fan opened a new pull request, #57859:
URL: https://github.com/apache/spark/pull/57859
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g. '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
This is a follow-up to #57363. It extends `CombineAdjacentAggregation` for adjacent hash
aggregates in two ways:
- combine `PartialMerge` followed by `Final` into one `Final` aggregate;
- preserve the lower aggregate's child distribution, streaming, and shuffle-partition metadata
when removing that lower aggregate.
The rule also requires the lower aggregate output to exactly match the final aggregate inputs.
Filtered `Partial` aggregation remains supported, while a `PartialMerge` pair with filters is not
combined.
### Why are the changes needed?
`CombineAdjacentAggregation` currently handles only `Partial` followed by `Final`. A compatible
adjacent `PartialMerge` and `Final` pair can also be collapsed safely, avoiding an unnecessary hash
aggregation stage. When the lower aggregate is removed, its child-facing distribution and streaming
metadata must move to the combined node because the combined node now reads that lower aggregate's
original child.
These semantics also move the upstream rule closer to covering the safe hash-aggregate behavior of
downstream implementations, reducing the need for overlapping optimizer rules.
### Does this PR introduce _any_ user-facing change?
Yes. Spark may produce a single final hash aggregate instead of adjacent partial-merge and final
hash aggregates when their grouping, lineage, and input/output attributes are compatible. Query
results are unchanged.
### How was this patch tested?
Added coverage to `CombineAdjacentAggregationSuite` for:
- combining `PartialMerge` and `Final` hash aggregates;
- preserving distribution, streaming, shuffle, result, and buffer-offset metadata;
- refusing to combine a filtered `PartialMerge` pair.
Ran:
```
build/sbt 'sql/testOnly org.apache.spark.sql.execution.CombineAdjacentAggregationSuite'
build/sbt sql/scalastyle
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-5)
--
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-58210][SQL][FOLLOWUP] Extend CombineAdjacentAggregation to partial merge [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #57859: [SPARK-58210][SQL][FOLLOWUP] Extend CombineAdjacentAggregation to partial merge
URL: https://github.com/apache/spark/pull/57859
--
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-58210][SQL][FOLLOWUP] Extend CombineAdjacentAggregation to partial merge [spark]
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #57859:
URL: https://github.com/apache/spark/pull/57859#discussion_r3746181942
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/CombineAdjacentAggregationSuite.scala:
##########
@@ -191,6 +192,54 @@ class CombineAdjacentAggregationSuite extends QueryTest
}
}
+ test("Combine adjacent partial merge and final hash aggregates") {
Review Comment:
Combine PartialMerge - Final to one FInal is sound good to me. But is there a valid end to end test ? I can not find a real world query can reach this pattern that all agg functions are PartialMerge and the adjacent agg functions are Final.
--
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-58210][SQL][FOLLOWUP] Extend CombineAdjacentAggregation to partial merge [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #57859:
URL: https://github.com/apache/spark/pull/57859#discussion_r3746972044
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/CombineAdjacentAggregationSuite.scala:
##########
@@ -191,6 +192,54 @@ class CombineAdjacentAggregationSuite extends QueryTest
}
}
+ test("Combine adjacent partial merge and final hash aggregates") {
Review Comment:
Thanks for raising this. OSS does not currently produce a pure adjacent PartialMerge -> Final HashAggregate pair from a built-in query plan. The rule is intended to handle plans produced by physical-plan extensions that add an aggregation stage, and I updated the class documentation to say that explicitly.
The test now constructs a valid executable Partial -> PartialMerge -> Final pipeline, applies the rule to the upper pair, and compares the uncombined and combined execution results. This replaces the previous plan-shape-only coverage. I also added a separate end-to-end query test for the AQE rebalance/skew distribution regression.
--
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-58210][SQL][FOLLOWUP] Extend CombineAdjacentAggregation to partial merge [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #57859:
URL: https://github.com/apache/spark/pull/57859#issuecomment-5248359489
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/f88a554fafde5eec35c06af4c24e3e1df6cda313
- merged into branch-4.x https://github.com/apache/spark/commit/99c88b3c884dde2e32926e53bb1b0439d6a14ba7
- merged into branch-4.3 https://github.com/apache/spark/commit/0d8a1e6aaa298ac260803bd08f1acfa3e986998f
*Posted by `merge_spark_pr.py`*
--
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-58210][SQL][FOLLOWUP] Extend CombineAdjacentAggregation to partial merge [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #57859:
URL: https://github.com/apache/spark/pull/57859#issuecomment-5229628748
Thank you for extending this rule, @cloud-fan. The `PartialMerge` + `Final` -> `Final` combination itself looks semantically sound to me (merge is associative), but I have a few concerns.
**1. Weakening the required distribution of the existing `Partial` + `Final` path looks risky under AQE.**
`combineHashAggregates` now applies `requiredChildDistributionExpressions = partialAgg.requiredChildDistributionExpressions` in both modes. For a `Partial` aggregate this is `None` (`UnspecifiedDistribution`), while the previously merged code kept `finalAgg`'s `Some(groupingKeys)`. The combined `Complete`/`Final` aggregate still produces final results in a single pass, so clustering by the grouping keys remains its correctness requirement. That requirement is the safety net in AQE: `optimizeQueryStage` validates `AQEShuffleReadRule` results via `ValidateRequirements`, which checks each node's `requiredChildDistribution`.
Concretely, with `SELECT k, count(*) FROM (SELECT /*+ rebalance(k) */ * FROM t) GROUP BY k`, the rebalance shuffle satisfies the final aggregate's requirement, the two aggregates become adjacent, and the rule fires. Before this PR the combined node required `ClusteredDistribution(k)`, so `OptimizeSkewInRebalancePartitions` splitting a skewed partition would fail validation and be reverted. With this PR the combined node claims `Unspecified`, the split passes validation, the same key can land in multiple partitions, and the one-shot aggregate emits duplicate groups. The global-aggregation case (`Some(Nil)` = `AllTuples` -> `None`) has the same issue. I think the combined node should keep `finalAgg`'s `requiredChildDistributionExpressions` in both modes — removing the lower aggregate doesn't change the combined operator's own correctness requirement.
**2. The `partialAgg.outputSet != finalAgg.usedInputs` guard is always false.**
`HashAggregateExec.usedInputs` is `inputSet` (from `AggregateCodegenSupport`), i.e. `AttributeSet(child.output)`, and in this pattern `finalAgg.child` *is* `partialAgg`, so the two sides are identical by construction. The check never rejects anything. If the intent is "the lower aggregate's output exactly matches the final aggregate's inputs", that holds trivially; the layout that actually matters after combining is `partialAgg.child.output`, which this check doesn't look at.
**3. `initialInputBufferOffset` should arguably come from `partialAgg`.**
The combined `Final` aggregate reads `partialAgg.child`'s rows, so the buffer offset should be relative to that layout, i.e. `partialAgg.initialInputBufferOffset`. Today both offsets equal the grouping length by `AggUtils` construction, so they coincide, but the rule doesn't verify that. Using `partialAgg`'s offset (or guarding on equality) would be more robust.
Minor comments:
- The class doc still only describes `Partial` + `Final` -> `Complete`; the new pattern is worth documenting there, and the sort/object-hash branches don't get the new metadata handling — whatever we decide for (1) should be consistent across the three branches.
- The new test builds the `PartialMerge` plan synthetically and only checks plan shape; it never executes the combined plan. An end-to-end test (and one covering the AQE rebalance/skew scenario in (1)) would be valuable. Could you also share a concrete query where the planner naturally produces adjacent `PartialMerge` + `Final` hash aggregates? The `PartialMerge` nodes from `AggUtils` seem to be either mode-mixed (distinct) or separated by streaming state operators.
--
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-58210][SQL][FOLLOWUP] Extend CombineAdjacentAggregation to partial merge [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #57859:
URL: https://github.com/apache/spark/pull/57859#issuecomment-5243431832
Thanks, @dongjoon-hyun. Addressed these comments in 296a85085d6:
- updated the PR description to distinguish the upper aggregate's distribution requirement from
the lower aggregate's child-facing streaming and shuffle metadata, and removed the claim about
an explicit output/input equality check;
- removed the redundant `requiredChildDistributionExpressions` copy argument and added a comment
explaining why the remaining metadata comes from different nodes;
- replaced the `AggregateMode`/exception match with an explicit combined aggregate result;
- clarified that `PartialMerge` + `Final` is supported only for `HashAggregateExec`;
- made the AQE partition-spec assertion non-vacuous.
The focused suite (10 tests) and `sql/scalastyle` both pass.
--
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-58210][SQL][FOLLOWUP] Extend CombineAdjacentAggregation to partial merge [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #57859:
URL: https://github.com/apache/spark/pull/57859#issuecomment-5248352232
the failed streaming test is unrelated, thanks for review, merging to master/4.x/4.3
--
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-58210][SQL][FOLLOWUP] Extend CombineAdjacentAggregation to partial merge [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #57859:
URL: https://github.com/apache/spark/pull/57859#issuecomment-5236433239
Thanks for the detailed review. Addressed in 359d3dc5d3f:
1. The combined hash aggregate now retains `finalAgg.requiredChildDistributionExpressions`. This preserves `ClusteredDistribution`/`AllTuples` and lets AQE validation reject skewed rebalance reads that split a grouping key. I added an AQE rebalance/skew regression that checks the answer, confirms the aggregates combine, and verifies no `PartialReducerPartitionSpec` survives below the combined aggregate.
2. Removed the `partialAgg.outputSet != finalAgg.usedInputs` check, since it is true by construction and did not validate the post-rewrite input layout.
3. The `PartialMerge` -> `Final` result now takes `initialInputBufferOffset` from the lower aggregate, whose child becomes the combined aggregate child.
I also updated the class documentation. The sort/object-hash paths continue retaining the final node distribution through `copy`, consistently with the hash path.
For execution coverage, the PartialMerge test now builds a valid executable Partial -> PartialMerge -> Final pipeline and compares results before and after combining, in addition to its metadata and filter checks. OSS does not currently generate this pure adjacency from a built-in query; the documented use case is a physical-plan extension that introduces an extra aggregation stage.
I attempted `build/sbt 'sql/testOnly org.apache.spark.sql.execution.CombineAdjacentAggregationSuite'` twice, but SBT project loading was blocked before compilation by DNS failures resolving three BOMs from Maven Central (jackson-bom, jjwt-bom, and kubernetes-client-bom). Static diff/style checks pass.
--
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