cloud-fan opened a new pull request, #59823: URL: https://github.com/apache/spark/pull/59823
### What changes were proposed in this pull request? Adaptive partial aggregation (#57742) evaluates its compaction ratio periodically and right before its map would spill, against one threshold, `minCompaction` (default 1.05). This PR changes the evaluation before a spill in two ways, on both the codegen and the interpreted path: - It measures the recent rows, those processed since the periodic evaluation before last, against the keys they added, instead of the whole map epoch. - It demands a new, higher threshold, `spark.sql.execution.aggregate.adaptivePartialAggregation.minCompactionToSpill` (default 2.0). The evaluation uses the larger of it and `minCompaction`. The window advances at each periodic evaluation, so the change adds no per-row work. This replaces the spill threshold that an earlier revision of #59764 proposed, which judged the whole epoch, and covers the review comments on the spill check that were deferred from there. ### Why are the changes needed? Keeping the aggregation when the map is about to spill means aggregating the rest of the input through spills. That pays off only if the rest of the input keeps repeating keys: the final merge then collapses them, and the shuffle carries a fraction of the rows. Otherwise the task writes most of its rows to disk to save shuffling a few. With one threshold of 1.05, an input whose keys arrive together, at a ratio of 1.5, keeps aggregating and spills every epoch. The ratio of the whole epoch cannot tell those inputs apart. A uniformly distributed key with ten rows per key per task fills its map at a ratio of about 1.65, in the same range as an input whose epochs share no keys, yet keeping the aggregation is the right call for it. The recent rows separate the two. When keys repeat across the input, more of the recent rows find their key already in the map as it fills, so their ratio climbs above the epoch's, to between 2 and 3 in that example. When each key's rows arrive together, the two ratios match. The two evaluations weigh different costs, which is why they get separate thresholds. The periodic one weighs only maintaining the map, so a ratio barely above 1 is worth keeping, while the one before a spill weighs spilling the rest of the input. `AdaptivePartialAggregationBenchmark` gets three spill-check cases with an `Exchange` between the Partial and the Final. With codegen on my machine: - Keys that arrive together: about 20% faster than with 1.05, because the rest of the input is passed through instead of spilled every epoch. - Keys that recur: the same decision as 1.05, keeping the aggregation. - 10 hot keys over a unique tail: about 35% faster. Passing the rest of the input through also streams the later rows of keys the frozen map holds, so with a few hot keys each of them reaches the final aggregation row by row, on the one reducer that key hashes to. The docs describe this trade-off. ### Does this PR introduce _any_ user-facing change? No. Adaptive partial aggregation is new in the unreleased 4.4.0 and disabled by default. When it is enabled, a map about to spill passes the rest of its input through when its recent rows compact by less than 2.0, and the new config tunes that threshold. ### How was this patch tested? New and updated tests in `AdaptivePartialAggregationSuite`, on both paths: - An input whose keys arrive together at a ratio of 1.5 passes the rest through at the spill check by default, and keeps spilling with the threshold lowered to 1.3. It runs with one input partition and with two, which put an `Exchange` between the Partial and the Final. - An input whose epoch compacts by 1.875 but whose recent rows compact by 3.75 keeps aggregating by default, and passes through with the threshold raised to 4.0. - The exact-boundary test sets the spill threshold, and checks that the larger of the two thresholds applies. - A spill followed by a distinct-heavy epoch now flips at that epoch's first periodic check, with exact counts. - The new config rejects non-finite values, as `minCompaction` does. `build/sbt "sql/testOnly *AdaptivePartialAggregationSuite"` passes. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) This pull request and its description were written by Isaac. -- 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]
