HashAggregate performance improvements #418

Activity

1. ravlio

   added

   bugSomething isn't workingSomething isn't working

   on May 24, 2021
2. ravlio

   changed the title [-]performance considerations of create_key_for_col (HashAggregate)[/-] [+][question] performance considerations of create_key_for_col (HashAggregate)[/+] on May 24, 2021
3. Dandandan commented on May 28, 2021

   

   Dandandan

   on May 28, 2021

   Last edited by Dandandan

   Contributor

   More actions

   @ravlio

   You are very right, that part is suboptimal.
   Also in hash aggregates there are a couple of other things:

   • Keys are created by row and indexed by Vec<u8>. Each Vec<u8> value will be hashed, but is also rehashed as the hash table grows (rusts hashmap doesn' t remember hash values by default, so it has to compute it again every time the capacity of the table is exhausted).
     We should do something similar (vectorized hashing) in the hash aggregates as we do in hash join. There the create_hashes function generates u64 hash values for each row in one simple loop without downcasting / converting per row. The challenging part of this is detecting hash collisions and making sure that that is fast too.

   • Hash aggregates with small groups generate lot of small allocations and dynamically typed scalar values and trait objects for the aggregate states. Also in some parts, intermediate Vecs are allocated which also show up in profiles.
     I've addressed some performance issues in different parts earlier, however there are still large gains possible, mainly for some more challenging queries. I think what would be bes in the long run is building a mutable typed array based for the aggregation states, and keeping only the offsets to that array in a hash table structure. This would allow for vectorized updating the states like this (pseudo code):

   // sum
   for (offset, val) in batch_offsets.chain(batch_vals) {
      states[offset] += val;
   }
   // count
   for offset in batch_offsets {
      states[offset] += 1;
   }
   // avg 
   for (offset, val) in batch_offsets.chain(batch_vals) {
      state_avg_sum[offset] += val;
      state_avg_count[offset] += 1;
   }

   After the hash aggregation - the arrays don't need any further processing as all of the offset values are already increasing for each row. Only the avg here needs to run a divide at the end, which is fast as we can use the arrow kernel directly.
   (currently the group by values are converted to arrays again, which again takes time).
4. jhorstmann commented on May 31, 2021

   

   jhorstmann

   on May 31, 2021

   Contributor

   More actions

   I think what would be bes in the long run is building a mutable typed array based for the aggregation states, and keeping only the offsets to that array in a hash table structure.

   I agree and also did some experiments around this the last weeks. The slowest part is unfortunately still that you'd need a hashmap to build this array of consecutive offsets. But this part is much easier to specialize for different key types or number of group by columns. Updating the accumulator states as in your code then trades random access into memory vs dynamic method calls and should be faster.

   I did not base those experiments on arrow or datafusion though, the approach in datafusion with collecting the row indices and using take + batch_update could be better for small cardinalities but might otherwise have bigger overhead. The two approaches I compared where

   • a simple HashMap<Vec<Scalar>, Vec<Box<dyn Accumulator>>> that gets updated row by row and afterwards iterated for each group and aggregate column
   • a HashMap<Vec<Scalar>, usize>, the value being a sequential offset, at the same time collection the keys into a Vec<Vec<Scalar>> that corresponds to those offsets. Afterwards doing basically the same as in your example code, only requiring a dispatch on the array type once.

   In a microbenchmark with random data and grouping by two u32 columns (about 1000 different groups), ignoring any nulls, the second one was about 2x faster.

   👍React with 👍2Reacted by Daniël Heres and Jorge Leitao
5. Dandandan commented on May 31, 2021

   

   Dandandan

   on May 31, 2021

   Contributor

   More actions

   Thanks for the input @jhorstmann - that adds some support for the idea! Something like ~2x speed up for more challenging queries where DF currently "struggles" (or bigger for some extreme queries like 1 value per group) sounds about what I would expect from making the change in DataFusion as well based on some profiling results, in particular for high cardinality queries. It might be an idea to do a count on the number of distinct values inside a batch (that should be doable as we already are inserting the offsets to the HashMap) to decide whether to do a take + batch_update on low cardinality or whether to combine it in one loop for high cardinality updates?

   Arrow misses some kind of mutable data currently that can mutate data at offsets (rather than append values) - what are you using for that, or did you build your own structure?

   👍React with 👍1Reacted by Jorge Leitao
6. jorgecarleitao commented on May 31, 2021

   

   jorgecarleitao

   on May 31, 2021

   Member

   More actions

   fwiw, I have used the hashing in the experimental branch and I am getting some in-memory speedups:

   	master (ms)	1b48f38 (ms)
   Q1 memory	332.9	239.6
7. Dandandan commented on May 31, 2021

   

   Dandandan

   on May 31, 2021

   Last edited by Dandandan

   Contributor

   More actions

   @jorgecarleitao

   Interesting!
   I did some earlier experiments with the vectorized hashing too (and saw similar speed ups for low-cardinality aggregates), but got a bit stuck in making the hash collision performant enough to not slow it down.

   I am not sure where the hash collision detection happens in the experimental branch yet - what if two values map to the same u64 hash? It is quite hard to trigger that without writing some unit tests - I also got all the test passing without implementing something for collisions. A trivial way to detect collisions while not changing the code rigorously (comparing GroupByValues while inserting) made it perform slower than the original implementation.

   👍React with 👍1Reacted by Jorge Leitao
8. Dandandan

   changed the title [-][question] performance considerations of create_key_for_col (HashAggregate)[/-] [+]HashAggregate performance improvements[/+] on May 31, 2021
9. alamb commented on Mar 9, 2022

   

   alamb

   on Mar 9, 2022

   Contributor

   More actions

   FWIW I think @yjshen is working on this using the #1849 and row format frameworks
10. alamb

    closed this as completedin #6904on Jul 13, 2023
11. unkloud

    added a commit that references this issue on Mar 23, 2025

    chore: Rename some columnar shuffle configs for code consistently (ap…

    ...

    5430f63