Repository navigation
Fusing partial aggregation with repartition #12596
Description
Is your feature request related to a problem or challenge?
I impl a poc #12526, and found this idea can actually improve performance.
But for some reasons stated in #11680 (comment)
I think this improvement is not so suitable to be pushed forward currently.
Just file an issue to track it.
Describe the solution you'd like
- Introduce the partitioned hashtable in
partial aggregation, and we partition the datafusion before inserting them into hashtable. - And we push them into
final aggregationpartition by partition after, rather than split them again inrepartition, and merge them again incoalesce.
Describe alternatives you've considered
No response
Additional context
No response
Activity
Rachelint commentedon Sep 23, 2024
take
Rachelint commentedon Sep 23, 2024
I will push this forward after tasks listed in #11680 (comment) finished
Rachelint commentedon Sep 23, 2024
@waynexia may be also interested about this?
yjshen commentedon Sep 25, 2024
I'm not clear on how this proposal works. Could you please explain why it provides performance benefits compared to partial aggregation, exchange, and final aggregation? Is the proposal aimed explicitly at accelerating high cardinality aggregation, or is it intended to enhance aggregation performance?
Rachelint commentedon Sep 25, 2024
I think it enhances aggregation performance generally?
Currently we can think
GroupValuesandGroupAccumulatoruses a singleVecto manage intermediate states inpartial aggr.After finishing work in
partial aggr, we pass thebatchtoexchange, then we recompute thehashesofbatch. Actually thehasheshave been computed inGroupValues, the this recomputing isthe first avoidable cpu cost.Then we split the
batchto multiplebatches, according to thepartition nubmerscomputed fromhashes. The splitting needs to creating multiple newbatchesto hold the values from the sourcebatch, and need to copy data into them, and that isthe second avoidable cpu cost.Finally, before passing data to
final aggrof the partition, we need to copy the splitted smallbatchesof the partition to buffer incoalescefirstly, until the buffer large enough (usually the default batch size 8192), and that isthe third avoidable cpu cost.After using partitioned approach in
GroupValuesandGroupAccumulator:We can naturally reuse the computed
hashesinGroupValueswhen we calculating thepartition numbersof thebatches.We store the intermediate states in
partial aggrpartition by partition. And when we submit them tofinal aggr, we just submit them partition by partition, rather than splitting first and merging after.jayzhan211 commentedon Sep 25, 2024
I think our goal is to combine partial + repartition + final into single operator, and partial + repartition fusing is the first step of this. After that we could try doing final aggr step as well.
Rachelint commentedon Sep 25, 2024
Yes, it may be attractive if we combine them by someway, we seem to have chance to do more optimizations.
🤔
record batchfrompartialinternal states.finalfinal's internal states.It seems to be expensive for bytes and string? Maybe we can pass the internal states directly to
final, and avoid some copies?alamb commentedon Sep 16, 2025
Possibly related to
Rachelint commentedon Jun 24, 2026
Back to try this, I think it is also an important part for a promising improvement about partition-wise aggregation.
Rachelint commentedon Jun 24, 2026
Prove to be not promising through #23149
The reason is:
RepartitionExechas been highly optimized now, and usually occupy 5-% cpu time in query