Copilot commented on code in PR #13328:
URL: https://github.com/apache/gluten/pull/13328#discussion_r4294737802
##########
gluten-substrait/src/main/scala/org/apache/gluten/extension/joinagg/ImplementJoinAggregate.scala:
##########
@@ -150,16 +179,74 @@ case class ImplementJoinAggregate(spark: SparkSession)
extends SparkStrategy {
} else {
childPlan
}
+ val stageChild = if (stage == AggregationStage.TWO_STAGE) {
+ buildMapperAggregate(grouping, preparedChild, rewrittenAggExprs)
+ } else {
+ preparedChild
+ }
+ val stageExpressions = if (stage == AggregationStage.TWO_STAGE) {
+ rewrittenAggExprs.map {
+ ae =>
+ val mode = ae.mode match {
+ case Partial => PartialMerge
+ case Complete => Final
+ case other => other
+ }
+ ae.copy(mode = mode)
Review Comment:
With `twoStage.mode=TOP`, a final wrapper can coexist with an ordinary
filtered aggregate, such as `SUM(f.x)` alongside `MAX(d.v) FILTER (WHERE
d.flag)`. This copy changes the ordinary aggregate from `Complete` to `Final`
but retains its filter. The mapper has already applied that filter and no
longer outputs `d.flag`. Velox still translates reducer filters in
`HashAggregateExecTransformer.getAggRelInternal`, so the reference cannot bind
and native aggregation falls back. Clear filters on the reducer expressions,
retaining them on the mapper.
##########
gluten-substrait/src/main/scala/org/apache/gluten/extension/joinagg/ImplementJoinAggregate.scala:
##########
@@ -150,16 +179,74 @@ case class ImplementJoinAggregate(spark: SparkSession)
extends SparkStrategy {
} else {
childPlan
}
+ val stageChild = if (stage == AggregationStage.TWO_STAGE) {
+ buildMapperAggregate(grouping, preparedChild, rewrittenAggExprs)
+ } else {
+ preparedChild
+ }
+ val stageExpressions = if (stage == AggregationStage.TWO_STAGE) {
+ rewrittenAggExprs.map {
+ ae =>
+ val mode = ae.mode match {
+ case Partial => PartialMerge
+ case Complete => Final
+ case other => other
+ }
+ ae.copy(mode = mode)
+ }
+ } else {
+ rewrittenAggExprs
+ }
+ val distribution = stage match {
+ case AggregationStage.LOCAL => None
+ case AggregationStage.DISTRIBUTED | AggregationStage.TWO_STAGE =>
+ Some(grouping.map(_.toAttribute))
+ }
if (packOutputBuffers) {
- planBufferPhase(
+ buildBufferAggregate(
grouping,
aggregateExpressions,
resultExpressions,
- preparedChild,
- rewrittenAggExprs)
+ stageChild,
+ stageExpressions,
+ distribution)
} else {
- planFinalOutput(grouping, resultExpressions, preparedChild,
rewrittenAggExprs)
+ buildFinalAggregate(grouping, resultExpressions, stageChild,
stageExpressions, distribution)
Review Comment:
With `twoStage.mode=TOP`, the mapper outputs grouping attributes and
aggregate buffers, but the reducer still receives the original grouping
expressions. For a computed key such as `substr(i_item_desc, 3)`, the reducer
tries to bind `i_item_desc`, which the mapper no longer outputs. Use the
grouping attributes for the reducer while retaining the original expressions on
the mapper.
##########
gluten-substrait/src/main/scala/org/apache/gluten/extension/joinagg/ImplementJoinAggregate.scala:
##########
@@ -150,16 +179,74 @@ case class ImplementJoinAggregate(spark: SparkSession)
extends SparkStrategy {
} else {
childPlan
}
+ val stageChild = if (stage == AggregationStage.TWO_STAGE) {
+ buildMapperAggregate(grouping, preparedChild, rewrittenAggExprs)
Review Comment:
With aggregate pushdown enabled, `TOP` also adds this mapper when no join
was crossed. `PushAggregateThroughJoin.splitAggregate` wraps ordinary
aggregations regardless of their child, so `SELECT k, SUM(v) FROM t GROUP BY k`
gets `Partial → PartialMerge → Final`. The extra local merge receives
already-unique per-partition keys and builds another hash table without
reducing rows. Restrict mapper insertion to actual pushdown, or skip it when
the child already performs the same-key local aggregation.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
