Currently, GROUP BY queries run in parallel only in a few cases:

• Non-keyed GROUP BY with basic aggregate functions, e.g. select sum(value) from t.
• Single INT or SYMBOL-keyed GROUP BY with basic aggregate functions, e.g. select int_key, sum(value) from t.

This patch extends cases where we go with multi-threaded GROUP BY execution. The implementation builds on top of parallel SQL filters (a.k.a. async offload), so the same scheduling and cancellation behavior applies. The work is split into page frame tasks, aggregated by the shared workers, and accumulated in FastMap (keyed GROUP BY) or SimpleMapValue (non-keyed GROUP BY). FastMap/SimpleMapValue is reused between different query executions (and different queries).

As an optional second step in the query processing, we merge sharded maps in parallel. Shards contain non-intersecting sets of groups, so that once we have full shards, we return their rows to the caller. This behavior kicks in only in case of large enough maps (cairo.sql.parallel.groupby.sharding.threshold, defaults to 10k).

The implementation also "steals" filter from the underlying factory, so both of the following sample queries will be executed by the new parallel GROUP BY framework:

-- column keys are supported
select str_col, sum(long_col) from t;
-- function and operation keys are supported; filters are also supported
select concat(str_col1, str_col2), sum(long_col) from t where long_col > 42;
-- non-keyed GROUP BY is also supported
select vwap(price, quantity) from t where quantity > 42;

Currently supported aggregate functions: count(*) and count(col), avg, sum, min/max, vwap (all for fixed-size types).

Benchmark results aren't included, but the improvement on my 4c/8t machine varies from 2x to 10x depending on the query.

The new behavior is enabled by default, but can be switched off with cairo.sql.parallel.groupby.enabled=false.

Also includes #4078 (single count_distinct re-write to a parallel GROUP BY for all supported types except symbol).

Other limitations

• Aggregate functions that have an additional state, e.g. count_distinct, aren't yet supported.

Next steps

• Port aggregate functions with additional state to the new framework.
• Get rid of CompactMap and introduce UnorderedMap for the small fixed-size key-value case. That's to speed up key look-ups by avoiding extra access to FastMap's heap once we've determined the hash table slot.
• Start building multi-threaded SAMPLE BY factories based on the same approach.