You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "Dandandan (via GitHub)" <gi...@apache.org> on 2026/03/07 08:09:04 UTC
[I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Dandandan opened a new issue, #20773:
URL: https://github.com/apache/datafusion/issues/20773
### Is your feature request related to a problem or challenge?
The paper "Morsel-Driven Parallelism"[1] by Leis et al introduces a cache efficient aggregation algorithm.
It is also implemented by DuckDB.
We can implement it and see if it improves performance.
[1] https://db.in.tum.de/~leis/papers/morsels.pdf
[2] https://duckdb.org/2022/03/07/aggregate-hashtable#parallel-aggregation
### Describe the solution you'd like
The steps are as follows
1. Start with a single hashtable (current approach) without pre-repartitioning
2. Once the table exceeds a threshold (i.e. roughly CPU cache size), (radix) repartition the maps into a number of hashmaps (e.g. we can start of using the number of `target_partitions`)
3. Once they are repartitioned the output batches can be sent directly to the target partitions (no Repartition needed!)
<img width="351" height="274" alt="Image" src="https://github.com/user-attachments/assets/d41845bb-1c14-4715-9f44-dcbb6b516014" />
The benefit of it I think is mostly that by local grouping, we process the rows (smaller) hashmaps partition-by-partition, so while doing it they more likely fit in CPU cache.
Care must be taken to make the code efficient, i.e. avoid materializing the batches upfront / accumulate based on indices.
### Describe alternatives you've considered
_No response_
### Additional context
_No response_
--
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: github-unsubscribe@datafusion.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on issue #20773:
URL: https://github.com/apache/datafusion/issues/20773#issuecomment-4017058606
> We have tried the similar approach in datafusion before, see https://github.com/apache/datafusion/issues/6937#issuecomment-1681310199 , but found no obvious improvement.
Note that this is significantly different:
The algorithm used by DuckDB/in the paper:
* inserts into the hashmap *partition by partition* / *hashmap by hashmap* (I think this is the most important point) - this will make the partial aggregation much more cache efficient, even for aggregations that are not *that* big as the hashmap that is "in progress" will more likely fit in cache / lookups are likely to be in cache.
* Avoids the extra partitioning step (both the double hashing as the copying)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on issue #20773:
URL: https://github.com/apache/datafusion/issues/20773#issuecomment-4017061730
The core difference: The spilling in this paper doesn't just spill to the final partition, it partitions to a number of local tables and will insert to them one-by-one.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on issue #20773:
URL: https://github.com/apache/datafusion/issues/20773#issuecomment-4017342353
Thanks for your thoughts @Rachelint and sharing the differences in the DuckDB blogs / versions.
I agree we should probably take the ideas from the paper and see if (and perhaps mainly _how_) we can apply it here.
Also there are cases where DF is currently already faster than DuckDB (even without morsels and filter pushdown) - so we're probably currently do better in certain areas.
Probably is gonna take quite some experimentation until we can find out what works.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Posted by "Rachelint (via GitHub)" <gi...@apache.org>.
Rachelint commented on issue #20773:
URL: https://github.com/apache/datafusion/issues/20773#issuecomment-4017011592
As I see, these two articles actually talk about how `duckdb` improve performance in high cardinality groups aggregation.
But `datafusion` and `duckdb` are too different in aggregation today,
and methods in `duckdb` seems to help few about continuing to improve `high cardinality groups aggregation` of datafusion?
At first let's see the difference in `high cardinality groups aggregation`.
# Duckdb 0.7
https://duckdb.org/2022/03/07/aggregate-hashtable#parallel-aggregation
This article describes the total method to improve performance of high cardinality groups aggregation in duckdb 0.7.
And I see, there are some mistakes in it, it use a very different way with what mentioned in [Leis et al](https://15721.courses.cs.cmu.edu/spring2016/papers/p743-leis.pdf)
- In low cardinality case (the hashtable len in partial aggr <= 10K)
- use single hashtable in `partial aggr`
- use single thread to perform `final aggr` (merge hashtables from `partial aggr`s)
- In high cardinality case (the hashtable len in partial aggr > 10K)
- use partitioned hashtable in `partial aggr`
- use multiple threads to perform parallel `final aggr`(like `final aggr` in datafusion)
Actually, it give up cache-efficient of hashtable in `partial aggr`, and use the parallel `final aggr` to improve performance.
# Duckdb now
https://db.in.tum.de/~leis/papers/morsels.pdf
Duckdb use the similar method in the paper currently.
- In `partial aggr`, it always keep a very small and fixed hashtable
- and when hashtable in `partial aggr` become too big, it convert the data to partitions, and reset the hashtable.
- In `final aggr`, like datafusion, always partitioning and merge in parallel
We have tried the similar approach in datafusion before, see https://github.com/apache/datafusion/issues/6937#issuecomment-1681310199 , but found no obvious improvement.
I guess, it is due to the execution model of datafusion (pull-based and async, hard to be cache-efficient).
# Datafusion
datafusion use `skip partial aggregations` to improve `high cardinality groups aggregation` like `doris`.
- In `partial aggr`, we will perform the groups aggregation logic when not exceed the threshold at first
- Then we will totally skip the `partial aggr` if excceed the threshold.
- So, I guess few obvious improvement can got about optimizing `partial aggr` in `datafusion`, because we will totally skip it in `high cardinality groups 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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on issue #20773:
URL: https://github.com/apache/datafusion/issues/20773#issuecomment-4026078215
Some idea that came up today during discussing morsel-driven changes (https://github.com/apache/datafusion/pull/20481) is that we can as well sort on (file / row group / page) statistics (i.e. cluster min/maxes together for each worker/partition). This should make the the cardinality of the partial aggregate smaller (and thus much faster / less repartitioning going on).
Not an algorithm change, but could potentially help a lot of queries.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Posted by "Abhisheklearn12 (via GitHub)" <gi...@apache.org>.
Abhisheklearn12 commented on issue #20773:
URL: https://github.com/apache/datafusion/issues/20773#issuecomment-4016227238
Hi @Dandandan, if you're not currently working on this, I'd be happy to take it on.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Posted by "Rachelint (via GitHub)" <gi...@apache.org>.
Rachelint commented on issue #20773:
URL: https://github.com/apache/datafusion/issues/20773#issuecomment-6045514684
> My concerns are that:
>
> * The paper describe about how to keep the hashtable small and cache-efficient in high cardinality case in `partial aggr`
> * And situation seems in datafuson: ....
>
It is my mistake, after further researching, I found cache efficient partial aggr actually very promising (https://github.com/apache/datafusion/pull/26116) .
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Posted by "Rachelint (via GitHub)" <gi...@apache.org>.
Rachelint commented on issue #20773:
URL: https://github.com/apache/datafusion/issues/20773#issuecomment-4017282421
My concerns are that:
- The paper describe about how to keep the hashtable small and cache-efficient in high cardinality case
- And situation seems in datafuson:
- In `high cardinality` case, `datafusion` use `skip aggregations approach` to even totally skip the `partial aggr`, and skipping possible to be a better approach(`doris` use the similar approach).
- In `low cardinality` case, the hashtable usually small enough without extra processing?
> The description in DuckDB blog sounds to me very similar/the same as the paper, "spills when ht becomes full" means there is some threshold on the hashtable, after which it will repartition/spill to the local hash tables.
Method in [Leis et al](https://15721.courses.cs.cmu.edu/spring2016/papers/p743-leis.pdf) may be more similar as current duckdb(the approach is mentioned together with their external aggregtion in https://duckdb.org/2024/03/29/external-aggregation), it is different with duckdb 0.7:
- Only one small enough fixed hashtable in `partial aggr` thread, and only partition the group values
- When hashtable full, flush the partitioned group values
- Clear and reuse the hashtable
- And in `final aggr`, what `current duckdb` do is similar as `datafusion`, always perform partition-wise parallel merge (`duckdb 0.7` will perform single thread merge when groups are few)
As I see, `making partial aggr cache-friendly` mentioned in paper(and current duckdb) is actually `making the hashtable in partial aggr cache-friendly`.
And the paper and current duckdb use reset + rebuild method to reach it even in `high cardinality` case.
Duckdb 0.7 ([method in the blog](https://duckdb.org/2022/03/07/aggregate-hashtable#parallel-aggregation) seems to only describe how they use partitioned hashtable to support parallel merge in `final aggr`. And actually partitioned hashtable help few with`cache-friendly` due to some experiment I did before.
> The problem with spilling to repartition / final aggregation (https://github.com/apache/datafusion/issues/6937#issuecomment-1681310199) is that it will reduce keys less, and thus cause much more cross communication / repartition overhead and cost in the upstream aggregation.
Agree with "flush" is too expansive here, that possible lead to no obvious improvement.
It may be not the good case to measure if the method in [Leis et al](https://15721.courses.cs.cmu.edu/spring2016/papers/p743-leis.pdf) can also work well in `datafusion`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Posted by "Rachelint (via GitHub)" <gi...@apache.org>.
Rachelint commented on issue #20773:
URL: https://github.com/apache/datafusion/issues/20773#issuecomment-4017032783
As I see, one of the most obviously bottleneck of aggregation in datafuison is `RepartitionExec`.
The partitioned hashtable approach maybe can help, I did an experiment about it before:
https://github.com/apache/datafusion/pull/12526
The performance improvement got from `removing RepartitionExec`.
But taking consider with `skip partial aggregations`, I think it not the good and general way to solve the performance problem of `RepartitionExec`.
I think #15383 may be the better approach about it.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on issue #20773:
URL: https://github.com/apache/datafusion/issues/20773#issuecomment-5762104712
I prototyped the second half of this idea (draft PR #25567) — a Final aggregation that, once its table outgrows the
cache, moves its state and all further input into 64 hash buckets and aggregates them one at a
time — and measured where the time goes. Two findings seem relevant to the design here.
**1. Re-partitioning a table that was already built is what loses.** Emitting the first table,
scattering it and interning its groups again costs about as much as building it did. With a
threshold of `T` groups, every query that ends at 1-4x `T` groups per partition pays that without
earning it back, and raising `T` only moves the loss to other queries:
| threshold | slower than main |
|---|---|
| 256k | Q13 1.18x, Q14 1.11x |
| 512k | Q13 1.60x, Q14 1.26x |
| 1M | Q33 1.23x, Q31 1.15x, Q35 1.12x |
| 2M | Q16 1.09x |
while the large aggregations are best with an early switch (Q18 0.39x at 256k, 0.51x at 2M).
DuckDB's ICDE 2024 paper makes the same point against the HyPer / BLU design: "Rather than
partitioning tuples when the hash table is reset ... our implementation directly materializes
tuples into partitions ... we avoid copying tuples more than once", and a full table only has its
pointer array reset "while the tuples stay in place". I would aim for that shape — rows go to their
partition once, from the first batch — rather than "one table, radix-repartition it when it grows".
**2. The second scatter is 35-55% of the bucketed Final's compute.** Per Final input row:
| query | one table | bucketed = routing + aggregation |
|---|---|---|
| Q31 (2 int keys) | 85 ns | 44 = 24.5 + 19.5 |
| Q18 (int + string keys) | 229 | 71 = 27 + 44 |
| Q14 (string key) | 79-85 | 91-99 = 42-46 + 49-54 |
| Q12 (string key) | 91 | 97 = 39 + 58 |
`RepartitionExec` has already hashed, gathered and copied each of these rows once; the Final does
it again. If the exchange produced `partitions x K` sub-buckets from the same hash, the routing
column would disappear: the mid-size string-key queries would turn from 1.03-1.07x into roughly
0.92x, and the queries that already win would gain about as much again. A standalone harness
(no Arrow; one copy into 64 buckets, hash kept with the row) confirms that partitioning pays for
string keys from 250k groups per thread: 24-byte keys 0.63x / 0.56x / 0.48x / 0.34x of a single
table at 250k / 500k / 1M / 4M groups, 12 threads. Re-hashing in the bucket instead of carrying the
hash costs little for short keys and a lot for long ones (60 bytes at 1M groups: 0.54x -> 0.78x),
which ties in with #11680.
Numbers are from an Apple M4 Pro, 12 partitions, ClickBench `hits_partitioned`, interleaved runs. I am prototyping the "repartition produces the buckets" variant next and will report what it measures here.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on issue #20773:
URL: https://github.com/apache/datafusion/issues/20773#issuecomment-4017170204
> Hi [@Dandandan](https://github.com/Dandandan), if you're not currently working on this, I'd be happy to take it on.
Feel feel to have a go at it, though it will not be easy as it touches on multiple parts within the aggregation code and planning (and making sure not to introduce any slow parts).
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Posted by "Abhisheklearn12 (via GitHub)" <gi...@apache.org>.
Abhisheklearn12 commented on issue #20773:
URL: https://github.com/apache/datafusion/issues/20773#issuecomment-4017188394
@Dandandan , thanks for the heads up! I’ll start digging into the aggregation and planning parts and will be careful to avoid introducing any performance regressions.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Posted by "Abhisheklearn12 (via GitHub)" <gi...@apache.org>.
Abhisheklearn12 commented on issue #20773:
URL: https://github.com/apache/datafusion/issues/20773#issuecomment-4017188603
take
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Posted by "Rachelint (via GitHub)" <gi...@apache.org>.
Rachelint commented on issue #20773:
URL: https://github.com/apache/datafusion/issues/20773#issuecomment-4021933734
> I wonder if we can make things also more cache aware with some sorting / partitioning based on hash as well with more minimal changes with higher cardinality, perhaps getting a similar win in cache efficiency for high cardinality (without the large changes):
>
> * sort/partition hashes so we make the access to the table more linear (and also detect/skip duplicate keys avoiding probes
> * sort/partition group indices to make accumulate state access more linear (rather than scattered over entire group state)
Yes, after reading duckdb codes, my current thoughts:
- For `partial aggr`, skipping is alread a good method to handle high cardinality groups in `partial aggr`, what we should contintue to do for `partial aggr` I think is improving performance of `RepartExec`
- For `final aggr`, we should use the partition-wise method to keep the hashmap always small like what is done in `duckdb`. And luckily, I found the code changes to reach it is not actually large, I am doing experiment about this.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Implement cache-efficient partial aggregation by Leis et al [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on issue #20773:
URL: https://github.com/apache/datafusion/issues/20773#issuecomment-4018583798
I wonder if we can make things also more cache aware with some sorting / partitioning based on hash as well with more minimal changes with higher cardinality, perhaps getting a similar win in cache efficiency:
* sort/partition hashes so we make the access to the table more linear (and also detect/skip duplicate keys)
* sort/partition group indices to make accumulate access more linear
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org