Optimize Accumulator `size` function performance (fix regression on clickbench)

[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

[comphead]

@alamb @Dandandan @crepererum

[alamb]

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

[comphead]

@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

[alamb]

Thanks @comphead

[Dandandan]

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

[alamb]

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

[comphead]

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

[crepererum]

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.

[alamb]

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)

[yjshen]

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