ulysses-you opened a new pull request, #57742:
URL: https://github.com/apache/spark/pull/57742
### What changes were proposed in this pull request?
This PR makes hash aggregation detect at runtime that a pre-shuffle partial
aggregation is not reducing rows, and bypass it: the remaining input rows are
passed through as single-row partial buffers that the downstream `Final`
aggregation merges, so the output contract is unchanged.
Two decision tiers, both evaluated only while the regular (second-level) map
is still fully in memory (`sorter == null`):
- **no-spill tier**: from `sampleRows` regular-map rows on, bypass if
`distinctKeys / regularRows >= noSpillReductionRatioThreshold`. The sampling
window doubles after each sub-threshold check, so a low-cardinality input is
re-evaluated only rarely while a late high-cardinality tail can still be caught.
- **on-spill tier**: when the map cannot allocate for a new key (it would
otherwise spill), bypass instead if the ratio is at least
`spillReductionRatioThreshold`. This threshold is more aggressive because a
spilling partial aggregation starts paying disk I/O.
Only a pre-shuffle `Partial` hash aggregation with grouping keys is eligible
(`requiredChildDistributionExpressions.isEmpty` identifies that phase and,
importantly, keeps a group-by-only aggregate's `Final` phase out - its
`aggregateExpressions` is empty, so a mode-only check would admit it
vacuously). Both the codegen path (`HashAggregateExec`) and the interpreted
path (`TungstenAggregationIterator`) are covered.
Only the regular (second-level) map is governed, and the reduction ratio
uses the rows that entered it as the denominator. The append-only fast hash map
never spills, so a fast-map hit counts in neither the numerator nor the
denominator; using total rows would keep the ratio below the threshold even for
fully distinct input, because the fast map absorbs the first 2^16 keys.
Once pass-through is active the maps are frozen, so they are output - which
also frees their memory - before the remaining input is streamed, rather than
being held until the end.
`DISTINCT` aggregates are eligible: in the multi-phase distinct plan the
intermediate `PartialMerge` phase is not `Partial` mode (and requires a
distribution), so it always aggregates and de-duplicates, and the rows reaching
the distinct `Partial` phase therefore carry exactly one distinct value each.
New configs, all under
`spark.sql.execution.aggregate.adaptivePartialAggregation.*`:
| config | default | meaning |
| --- | --- | --- |
| `enabled` | `true` | the feature switch |
| `sampleRows` | `100000` | rows before the first no-spill evaluation |
| `noSpillReductionRatioThreshold` | `0.95` | no-spill tier threshold |
| `spillReductionRatioThreshold` | `0.8` | on-spill tier threshold |
A `numBypassingRows` SQL metric reports how many rows bypassed.
### Why are the changes needed?
For high-cardinality grouping keys the pre-shuffle partial aggregation
reduces little or nothing, but still pays for maintaining - and often spilling
- an aggregation map as large as the input. Bypassing it at runtime removes
that cost while keeping the two-phase plan intact, so the decision needs no
planner-side statistics and adapts per task.
Benchmark (`AdaptivePartialAggregationBenchmark`, run in GitHub Actions, JDK
25):
| scenario | codegen | adaptive = F | adaptive = T | relative |
| --- | --- | --- | --- | --- |
| high card, no spill | on | 4087 ms | 2443 ms | **1.7X** |
| high card, no spill | off | 4924 ms | 3220 ms | **1.5X** |
| low card, no spill | on | 267 ms | 292 ms | 0.9X |
| low card, no spill | off | 1290 ms | 1301 ms | 1.0X |
| high card, spill | on | 7986 ms | 4354 ms | **1.8X** |
| high card, spill | off | 9386 ms | 5298 ms | **1.8X** |
| low card, spill | on | 774 ms | 794 ms | 1.0X |
| low card, spill | off | 1362 ms | 1368 ms | 1.0X |
The bypassing cases win in both tiers and on both execution paths; the
low-cardinality cases, where the tiers correctly decline to bypass, show no
regression beyond the small per-row sampling overhead.
This is related to, but independent of, the static
`spark.sql.execution.bypassPartialAggregation` (SPARK-57688), which drops the
partial aggregation at planning time. The runtime version keeps the partial
aggregation when it does reduce rows and only bypasses when the observed ratio
says it does not; the two can be used together.
### Does this PR introduce _any_ user-facing change?
No, query results are unchanged. The feature is enabled by default and only
changes how the pre-shuffle partial aggregation is executed, plus the new
`numBypassingRows` metric in the SQL UI.
### How was this patch tested?
New `AdaptivePartialAggregationSuite` (34 tests), in two halves:
- **Correctness**: output identical to the feature-off reference across the
full matrix of codegen on/off, two-level map on/off, and spill/no-spill, over a
range of aggregate shapes (multi-slot buffers, imperative buffers, `FILTER
(WHERE ...)`), key types (string, decimal, date, nullable), group-by-only
aggregates with duplicate keys, empty input, and `Expand`-bearing plans (ROLLUP
/ CUBE / GROUPING SETS / multi-distinct).
- **Triggering**: the `numBypassingRows` metric proves the bypass fires when
(and only when) it should - high cardinality bypasses, low cardinality does
not, the feature switch and eligibility rules are honored, and both tiers work.
These tests also compare against the feature-off reference so a bypassing run
can never pass on metrics alone. For `count(DISTINCT v) GROUP BY k` the
bypasses of the two `Partial` phases are told apart by grouping-key count and
asserted separately.
New `AdaptivePartialAggregationBenchmark` covering the {high,
low}-cardinality x {no-spill, on-spill} grid, each across codegen on/off and
the adaptive switch; results are included.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.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: [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]
