Skip to content

Optimize Accumulator size function performance (fix regression on clickbench) #5325

Closed

Description

@comphead

Is your feature request related to a problem or challenge? Please describe what you are trying to do.
During regression benchmarks it was found that DISTINCT queries has a performance drop. The analysis showed the size function implementation for DistinctCountAccumulator is inefficient.

Describe the solution you'd like
Need to improve size function or the way how number of bytes is collected

Describe alternatives you've considered
None

Additional context
Analysis details #5313
Original benchmarks ticket #5276

Activity

  1. comphead commented on Feb 17, 2023

    @comphead
    ContributorAuthor
  2. alamb commented on Feb 18, 2023

    @alamb
    Contributor

    Thank you @comphead

    I think this would be relatively straightforward to implement for the common case (the one in this code) of fixed size values

    Basically instead of looping over all scalar values to count their sizes (vec.iter()) we could special case by type for fixed size types

    https://github.com/apache/arrow-datafusion/blob/c5108aef4d2660cce950976af14d33444f27075e/datafusion/common/src/scalar.rs#L2367-L2374

    Instead add some code that checks "if is ScalarType::Int8, UInt8, etc then size = size[0]*vec.len()"

    This would be a good first issue I think -- a good result and straightforward implementation

  3. changed the title [-]Optimize Accumulator `size` function performance[/-] [+]Optimize Accumulator `size` function performance (fix regression on clickbench)[/+] on Feb 18, 2023
  4. comphead commented on Feb 20, 2023

    @comphead
    ContributorAuthor

    @alamb I'll take this ticket if not other volunteers, also I want to play a bit if we need super accurate size, probably we can do approx size which will serve to get the structure size with minor inaccuracy but will be faster

  5. alamb commented on Feb 20, 2023

    @alamb
    Contributor

    Thanks @comphead

  6. Dandandan commented on Feb 20, 2023

    @Dandandan
    Contributor

    @alamb I'll take this ticket if not other volunteers, also I want to play a bit if we need super accurate size, probably we can do approx size which will serve to get the structure size with minor inaccuracy but will be faster

    Maybe the size can be accumulated as well during updates.

  7. alamb commented on Feb 20, 2023

    @alamb
    Contributor

    Maybe the size can be accumulated as well during updates.

    This is an excellent idea that might work very well for variable length structures (like strings)

    Though in general the distinct accumulators are going to be fairly poor at any high cardinality usecase (as they store all values as ScalarValue). It would be great to make them better, but probably as another separate project

  8. comphead commented on Feb 22, 2023

    @comphead
    ContributorAuthor

    that was not that trivial I expected, so I have ran some experiments.

    • Update size incrementally for upcoming batch only -> doesn't seem to be a solution as we do not know in advance which hashes already counted and which are not. Expenses on calculating is higher than benefit.
    • Increase batch size from 8k to 65k improves the query speed 6 times(size() depends on batch size and gets called less often)
    • remove state constituents from size() improves 20%
    • approx size(first scalar value size * len()) improves up to 10 times, but not accurate size for variable length, like strings

    @alamb @Dandandan let me know your thoughts

  9. crepererum commented on Feb 23, 2023

    @crepererum
    Contributor

    Update size incrementally for upcoming batch only -> doesn't seem to be a solution as we do not know in advance which hashes already counted and which are not. Expenses on calculating is higher than benefit.

    Is the problem the self.values.insert(...) bit? HashSet sadly doesn't have an entry API (IIRC), but you can use a HashMap<K, ()> instead so you don't need to double-hash.

  10. alamb commented on Feb 23, 2023

    @alamb
    Contributor

    approx size(first scalar value size * len()) improves up to 10 times, but not accurate size for variable length, like strings

    Thank you for looking into this @comphead

    I think we should use this approach for fixed length (non variable length) data -- it will solve the performance regression we saw for clickbench

    In terms of handling variable length data more efficiently, I am not sure it is worth a lot of time optimizing the size() implementation because my guess is that the size() calculation will be a smaller portion of the overall runtime for high cardinality string columns (each one will be an allocated string, for example).

    I think a separate project to handle COUNT DISTINCT(high_cardinality_string_column) is probably needed

    Thus I propose:

    1. Implement the simplest thing that fixes the clickbench performance regression
    2. File a follow on ticket to track further improving the performance of COUNT DISTINCT queries (I can do this)
  11. yjshen commented on Feb 25, 2023

    @yjshen
    Member

    For the further improvement @alamb mentioned, we could rewrite an aggregate query with distinct aggregations into an expanded double aggregation. This would also eliminate the need for Vec<ScalarValue> to be stored in an accumulator. Spark does this in RewriteDistinctAggregates.

    Before

    Aggregate(
       key = ['key]
       functions = [COUNT(DISTINCT 'cat1),
                    COUNT(DISTINCT 'cat2),
                    sum('value)]
       output = ['key, 'cat1_cnt, 'cat2_cnt, 'total])
      LocalTableScan [...]
    

    After rewrite:

    Aggregate(
       key = ['key]
       functions = [count('cat1) FILTER (WHERE 'gid = 1),
                    count('cat2) FILTER (WHERE 'gid = 2),
                    first('total) ignore nulls FILTER (WHERE 'gid = 0)]
       output = ['key, 'cat1_cnt, 'cat2_cnt, 'total])
      Aggregate(
         key = ['key, 'cat1, 'cat2, 'gid]
         functions = [sum('value)]
         output = ['key, 'cat1, 'cat2, 'gid, 'total])
        Expand(
           projections = [('key, null, null, 0, cast('value as bigint)),
                          ('key, 'cat1, null, 1, null),
                          ('key, null, 'cat2, 2, null)]
           output = ['key, 'cat1, 'cat2, 'gid, 'value])
          LocalTableScan [...]
    
  12. alamb commented on Feb 25, 2023

    @alamb
    Contributor

    This would also eliminate the need for Vec to be stored in an accumulator. Spark does this in RewriteDistinctAggregates.

    Thanks -- this is a neat idea @yjshen

    One challenge I have seen with this approach in the past is it will result in a "diamond shaped plan" (where the same input stream is split into two output streams (to the different aggregates) and then brought back together. In general, this approach may required unbounded buffering if using sort based aggregation.

    But I think would definitely be worth considering

  13. Dandandan commented on Mar 3, 2023

    @Dandandan
    Contributor

    Another approach might be speeding up the accumulator.

    It might be worth looking at the dictionary and parquet interner implementations from @tustvold in arrow-rs, and use a similar approach here which should have non-trivial performance impact.

  14. comphead commented on Mar 3, 2023

    @comphead
    ContributorAuthor

    Another approach might be speeding up the accumulator.

    It might be worth looking at the dictionary and parquet interner implementations from @tustvold in arrow-rs, and use a similar approach here which should have non-trivial performance impact.

    Thanks @Dandandan if you could provide a link would be great!

  15. alamb commented on Mar 3, 2023

    @alamb
    Contributor

    If people are serious about wanting to improve the performance of the aggregators in general, I think we should consider combining our efforts as there are several people who seem interested and I think the work will collide if not done carefully.

    Here is one ticket that tracks ideas: #4973

  16. yjshen commented on Mar 4, 2023

    @yjshen
    Member

    One challenge I have seen with this approach in the past is it will result in a "diamond shaped plan" (where the same input stream is split into two output streams (to the different aggregates) and then brought back together. In general, this approach may required unbounded buffering if using sort based aggregation.

    Could you please elaborate more on 1. why two output streams are generated and 2. why it requires unbounded buffering? Thanks!

  17. alamb commented on Mar 5, 2023

    @alamb
    Contributor

    Hi @yjshen

    Here was my understanding of what you were proposing, which shows the diamond I am referring to.

    I may be misunderstanding your proposal;

    SELECT date, 
      COUNT(DISTINCT x), 
      COUNT(DISITNCT y)
    FROM
      t;
                          ┌─────────────────┐                         
                          │ Combine somehow │   Maybe Join? Could     
                          │    (on date)    │   also be some more     
                          │                 │   optimized version     
                          └─────────────────┘                         
                                   ▲                                  
                                   │                                  
                                   │                                  
                ┌──────────────────┴─────────────────────┐            
                │                                        │            
                │                                        │            
                │                                        │            
    ┌───────────────────────┐                ┌───────────────────────┐
    │     HashAggregate     │                │     HashAggregate     │
    │       gby: date       │                │       gby: date       │
    │     agg: COUNT(x)     │                │     agg: COUNT(y)     │
    └───────────────────────┘                └───────────────────────┘
                ▲                                        ▲            
                │                                        │            
                │                                        │            
    ┌───────────────────────┐                ┌───────────────────────┐
    │     HashAggregate     │                │     HashAggregate     │
    │     gby: date, x      │                │     gby: date, y      │
    │      agg: <NONE>      │                │      agg: <NONE>      │
    └───────────────────────┘                └───────────────────────┘
                ▲                                        ▲            
                │                                        │            
                │                                        │            
                └───────────────────┬────────────────────┘            
                                    │                                 
                                    │                                 
                                    │                                 
                        ┌───────────────────────┐                     
                        │         Scan          │                     
                        └───────────────────────┘                     
    
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions