Implement cache-efficient partial aggregation by Leis et al

[Dandandan]

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[2]

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 thread-local hashmaps (e.g. we can start of using the number of target_partitions and use hash % target_partitions instead but in future might make it more adaptive)
3. Once they are repartitioned / finalized the output batches can be sent directly to the target partitions (no RepartitionExec needed! anymore)

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.

It also avoids the double hashing in partial aggregation / hash repartition as the latter can be removed.

Care must be taken to make the code efficient, i.e. avoid materializing the batches upfront / accumulate based on indices.

I am not sure why the final aggregation doesn't take the same approach - this could be tested out as well when it is implemented for the partial aggregation.

Describe alternatives you've considered

No response

Additional context

No response

[Abhisheklearn12]

Hi @Dandandan, if you're not currently working on this, I'd be happy to take it on.

[Rachelint]

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

• 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 aggrs)

• 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 #6937 (comment) , 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.

[Rachelint]

As I see, one of the most obvious bottleneck of aggregation in datafuison is RepartitionExec.

Introduce partitioned hashtable in partial aggr maybe can help, I did an experiment about it before:
#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, and I think it really valuable to push forward.

[Dandandan]

We have tried the similar approach in datafusion before, see #6937 (comment) , 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)

[Dandandan]

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 (for all indices that match the partition in the current batch).

The problem with spilling to repartition / final aggregation (#6937 (comment)) is that it will reduce keys less, and thus cause much more cross communication / repartition overhead and cost in the upstream aggregation.

And I see, there are some mistakes in it, it use a very different way with what mentioned in Leis et al

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.

[Dandandan]

Hi @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).

[Abhisheklearn12]

@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.

[Abhisheklearn12]

take