You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by "zhztheplayer (via GitHub)" <gi...@apache.org> on 2026/09/22 04:38:10 UTC
[PR] [VL] Option to insert partial-merge aggregations for eager aggregation [gluten]
zhztheplayer opened a new pull request, #13086:
URL: https://github.com/apache/gluten/pull/13086
The patch adds a new option to enable more intermediate aggregations when eager aggregation (experimental) is enabled.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Option to insert partial-merge aggregations for eager aggregation [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13086:
URL: https://github.com/apache/gluten/pull/13086#issuecomment-5771286727
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Option to insert partial-merge aggregations for eager aggregation [gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer merged PR #13086:
URL: https://github.com/apache/gluten/pull/13086
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Option to insert partial-merge aggregations for eager aggregation [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13086:
URL: https://github.com/apache/gluten/pull/13086#issuecomment-5789018399
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Option to insert partial-merge aggregations for eager aggregation [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #13086:
URL: https://github.com/apache/gluten/pull/13086#discussion_r4080951572
##########
docs/Configuration.md:
##########
@@ -138,6 +138,7 @@ nav_order: 15
| spark.gluten.sql.orc.charType.scan.fallback.enabled | 🔄 Dynamic | true | Force fallback for orc char type scan. |
| spark.gluten.sql.pushAggregateThroughJoin.enabled | 🔄 Dynamic | false | Enables the push-aggregate-through-join optimization in Gluten. When enabled, aggregate operators may be pushed below joins during logical optimization and corresponding physical plans may be rewritten to execute the aggregation earlier. |
| spark.gluten.sql.pushAggregateThroughJoin.maxDepth | 🔄 Dynamic | 2147483647 | Maximum join traversal depth when applying the push-aggregate-through-join optimization. A value of 1 allows pushing an aggregate through a single join; larger values allow the rule to traverse and push through multiple consecutive joins. |
+| spark.gluten.sql.pushAggregateThroughJoin.partialMerge.enabled | 🔄 Dynamic | false | Enables a PartialMerge aggregate above each aggregate pushed through a join. |
Review Comment:
The PR description states this option is intended “when eager aggregation (experimental) is enabled”, but the documentation entry doesn’t mention that relationship/limitation. If partial-merge is only effective (or only supported) with eager aggregation, the docstring should state the prerequisite/interaction so users don’t enable it expecting behavior changes under the non-eager path.
##########
gluten-substrait/src/main/scala/org/apache/gluten/extension/joinagg/ImplementJoinAggregate.scala:
##########
@@ -97,6 +97,8 @@ case class ImplementJoinAggregate(spark: SparkSession) extends SparkStrategy {
phase match {
case JoinAggregateFunctionWrapper.PartialPhase =>
planPartialPhase(grouping, aggExpressions, resultExpressions, childPlan)
+ case JoinAggregateFunctionWrapper.PartialMergePhase =>
+ planPartialMergePhase(grouping, aggExpressions, resultExpressions, childPlan)
case JoinAggregateFunctionWrapper.FinalPhase =>
planFinalPhase(grouping, aggExpressions, resultExpressions, childPlan)
}
Review Comment:
This PR introduces a new logical phase (`PartialMergePhase`) with a distinct physical planning path (unpack + merge + repack). The updated suites validate higher-level aggregate counts/semantics, but there’s no targeted assertion that `PartialMergePhase` is actually lowered as intended (e.g., that the unpack projection is present and that the planned aggregate mode corresponds to merge). Adding a focused test that inspects the physical plan for the partial-merge path would help prevent regressions in the unpack/pack logic and phase-to-mode mapping.
##########
gluten-substrait/src/main/scala/org/apache/gluten/extension/joinagg/JoinAggregateFunctionWrapper.scala:
##########
@@ -50,6 +55,17 @@ object JoinAggregateFunctionWrapper {
wrapperKey = wrapperKey)
}
+ def wrapperPartialMerge(
+ innerAgg: DeclarativeAggregate,
+ inputBuffer: Expression,
+ wrapperKey: String = "0"): JoinAggregateFunctionWrapper = {
+ JoinAggregateFunctionWrapper(
+ innerAgg = innerAgg,
+ targetPhase = PartialMergePhase,
+ inputBuffer = Some(inputBuffer),
+ wrapperKey = wrapperKey)
+ }
Review Comment:
`PartialMergePhase` (and `FinalPhase`) semantically require `inputBuffer` to be defined, but that invariant is only implied by helper constructors. Adding a hard validation (e.g., a `require(...)` in the case class constructor based on `targetPhase`, or stricter constructors) would prevent accidental construction of invalid wrappers that would otherwise fall back to `CreateStruct(innerAgg.inputAggBufferAttributes)` via `outputBufferExpr`, potentially masking logical bugs and producing hard-to-debug plans.
##########
gluten-substrait/src/main/scala/org/apache/gluten/extension/joinagg/ImplementJoinAggregate.scala:
##########
@@ -109,20 +111,100 @@ case class ImplementJoinAggregate(spark: SparkSession) extends SparkStrategy {
aggregateExpressions: Seq[AggregateExpression],
resultExpressions: Seq[NamedExpression],
childPlan: SparkPlan): Option[SparkPlan] = {
- // The pushed logical aggregate exposes one wrapper-typed output per pushed aggregate. Spark
- // physically computes ordinary aggregate buffers, so this phase runs a normal HashAggregateExec
- // first and then repacks those buffers into the struct-valued wrapper outputs expected by the
- // logical plan above.
- val rewrittenAggExprs = aggregateExpressions.map {
+ planPhase(
+ grouping,
+ aggregateExpressions,
+ resultExpressions,
+ childPlan,
+ unpackInputBuffers = false,
+ packOutputBuffers = true)
+ }
+
+ private def planPartialMergePhase(
+ grouping: Seq[NamedExpression],
+ aggregateExpressions: Seq[AggregateExpression],
+ resultExpressions: Seq[NamedExpression],
+ childPlan: SparkPlan): Option[SparkPlan] = {
+ planPhase(
+ grouping,
+ aggregateExpressions,
+ resultExpressions,
+ childPlan,
+ unpackInputBuffers = true,
+ packOutputBuffers = true)
+ }
+
+ private def planPhase(
+ grouping: Seq[NamedExpression],
+ aggregateExpressions: Seq[AggregateExpression],
+ resultExpressions: Seq[NamedExpression],
+ childPlan: SparkPlan,
+ unpackInputBuffers: Boolean,
+ packOutputBuffers: Boolean): Option[SparkPlan] = {
+ val rewrittenAggExprs = rewriteWrapperAggregates(aggregateExpressions)
+ if (rewrittenAggExprs.isEmpty) {
+ return None
+ }
+ val preparedChild = if (unpackInputBuffers) {
+ unpackInputBufferFields(childPlan, aggregateExpressions, rewrittenAggExprs)
+ } else {
+ childPlan
+ }
+ if (packOutputBuffers) {
+ planBufferPhase(
+ grouping,
+ aggregateExpressions,
+ resultExpressions,
+ preparedChild,
+ rewrittenAggExprs)
+ } else {
+ planFinalOutput(grouping, resultExpressions, preparedChild, rewrittenAggExprs)
+ }
+ }
+
+ private def unpackInputBufferFields(
+ childPlan: SparkPlan,
+ aggregateExpressions: Seq[AggregateExpression],
+ rewrittenAggExprs: Seq[AggregateExpression]): SparkPlan = {
+ // Recreate the wrapped aggregate's physical input-buffer attributes from the struct payload.
+ val unpackAliases = ArrayBuffer.empty[Alias]
+ val seenExprIds = scala.collection.mutable.HashSet.empty[Long]
+ rewrittenAggExprs.zip(aggregateExpressions).foreach {
+ case (rewrittenAe, AggregateExpression(wrapper: JoinAggregateFunctionWrapper, _, _, _, _)) =>
+ val bufferExpr = wrapper.children.head
+ rewrittenAe.aggregateFunction.inputAggBufferAttributes.zipWithIndex.foreach {
+ case (bufferAttr, index) if seenExprIds.add(bufferAttr.exprId.id) =>
Review Comment:
`ExprId` in Spark includes more identity than just `exprId.id`; using `Long` here bakes in an assumption about uniqueness and makes the code harder to reason about. Consider tracking `ExprId` directly (or `bufferAttr.exprId`) in `seenExprIds` to better reflect intent and avoid accidental collisions if plan fragments ever combine ExprIds from different origins.
##########
gluten-substrait/src/main/scala/org/apache/gluten/extension/joinagg/PushAggregateThroughJoin.scala:
##########
@@ -335,6 +323,50 @@ case class PushAggregateThroughJoin(spark: SparkSession)
Some(pushedJoin)
}
+ // Retain the current buffer IDs above a fresh lower Partial. The lower Partial is then the
+ // only aggregate pushed through the next join edge.
+ private def splitPartialAggregateForMerge(partialAgg: Aggregate): (Aggregate, Aggregate) = {
+ val wrapperAliases = collectPartialWrapperAliases(partialAgg.aggregateExpressions)
+ val lowerAliases = wrapperAliases.map {
+ case (alias, wrapper) =>
+ Alias(
+ JoinAggregateFunctionWrapper
+ .wrapperPartial(wrapper.innerAgg, wrapper.wrapperKey)
+ .toAggregateExpression(),
+ alias.name)()
+ }
+ val lowerGroupingOutputs =
+ partialAgg.aggregateExpressions.take(partialAgg.groupingExpressions.size)
+ val lowerAgg = Aggregate(
+ groupingExpressions = partialAgg.groupingExpressions,
+ aggregateExpressions = lowerGroupingOutputs ++ lowerAliases,
+ child = partialAgg.child)
Review Comment:
The `lowerGroupingOutputs` selection assumes the first `groupingExpressions.size` entries in `partialAgg.aggregateExpressions` are exactly the grouping outputs. That positional contract is fragile and can silently break if the aggregate output ordering changes (e.g., due to analyzer rewrites or a different construction path), causing incorrect output schemas or mis-binding in later phases. Prefer deriving the grouping output `NamedExpression`s explicitly (e.g., by matching the grouping expressions to corresponding output attributes/aliases via semantic equality or a dedicated helper that constructs grouping outputs), and/or add an assertion documenting/enforcing the ordering invariant at the point where `partialAgg` is created.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org