Dhruv-meesho opened a new pull request, #12960:
URL: https://github.com/apache/gluten/pull/12960
Fixes #12959.
## What changes are proposed in this pull request?
`FlushableHashAggregateRule` decides an aggregate is intermediate via:
```scala
agg.aggregateExpressions.forall(p => p.mode == Partial || p.mode ==
PartialMerge)
```
For a grouping-only aggregate (`SELECT DISTINCT`, or `GROUP BY` with no
aggregate functions), `aggregateExpressions` is empty (`Seq.empty`), so
`forall` is vacuously `true`. The check cannot distinguish the partial stage
from the final (or complete) stage. The final stage is therefore converted into
`FlushableHashAggregateExecTransformer`, which sets `allowFlush = true` and
maps to `AggregationNode::Step::kPartial` in Velox.
When Velox abandons aggregation (based on memory/cardinality heuristics), it
emits duplicate grouping keys. Because this is the final aggregation stage, no
subsequent aggregate collapses the duplicates, resulting in silent duplicate
rows.
The rule only visits aggregates below a shuffle exchange, so a plain `SELECT
DISTINCT` whose result is consumed directly is not affected. However, when the
distinct output is repartitioned again (e.g., when feeding a join on a strict
subset of the distinct columns), the final aggregate enters the rule's scope
and gets converted.
### History & Context
- This case was previously covered by
`isAggInputAlreadyDistributedWithAggKeys`, added in #4443 (for #4421).
- It regressed in #12098 when the rule was narrowed to only protect against
`AggUtils.planAggregateWithOneDistinct`.
### Fix
Rather than restoring the broader distribution check and giving up the
performance gains of #12098, this PR introduces a targeted guard
`isGroupingOnlyFinalAgg`:
```scala
private def isGroupingOnlyFinalAgg(agg: HashAggregateExecTransformer):
Boolean = {
agg.aggregateExpressions.isEmpty &&
agg.requiredChildDistributionExpressions.isDefined
}
```
This skips conversion only when `aggregateExpressions` is empty and
`requiredChildDistributionExpressions.isDefined` (`None` for partial
aggregates, `Some(groupingAttributes)` for final aggregates in
`AggUtils.planAggregateWithoutDistinct`). This matches the condition Spark
checks in `HashAggregateExec.adaptivePartialAggEnabled` (SPARK-58511).
Aggregates with aggregate functions are unaffected and preserve the behavior
from #12098.
## How was this patch tested?
- Added regression test `flushable aggregate rule - distinct feeding a join
keeps final agg regular` to `VeloxAggregateFunctionsFlushSuite`.
- Verified that joining on a strict subset of distinct keys leaves the final
distinct aggregate as `RegularHashAggregateExecTransformer` and produces output
identical to vanilla Spark.
- Verified style check passed via `./dev/format-scala-code.sh`.
## Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude claude-3-7-sonnet
--
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]
