Implement hash-partitioned hash aggregate

[Dandandan]

This is (up to 4x in my earlier tests) faster than the current implementation that collects all parts to one "full" for cases with very high cardinality in the aggregate (think deduplication code). However, not hash partitioning is faster for very "simple" aggregates as less work needs to be done.

We probably need some fast way to have a rough estimate on the number of distinct values in the aggregate keys, maybe dynamically based on the first batch(es).

Also this work creates a building block for ballista to distribute data across workers, parallelizing it, avoiding collecting it to one worker, and making it scale to bigger datasets.

[returnString]

We probably need some fast way to have a rough estimate on the number of distinct values in the aggregate keys, maybe dynamically based on the first batch(es).

Does your TableProvider column stats work provide any useful base for this in situations where we're running aggregations on original table columns (as opposed to computed exprs) or is that too coarse?

[Dandandan]

We probably need some fast way to have a rough estimate on the number of distinct values in the aggregate keys, maybe dynamically based on the first batch(es).

Does your TableProvider column stats work provide any useful base for this in situations where we're running aggregations on original table columns (as opposed to computed exprs) or is that too coarse?

I think yes, for cases where the table is queried directly and we have statistics for distinct values available, those could be using heuristics for group by expressions based on cardinality statistics (say if you use column a and b maybe distinct(a)*distinct(b) would be an OK heuristic).
We also need the same support of distinct values per column for generalizing the join order optimization rule to more complicated joins and with more expressions than those directly on tables.
Support for collecting those statistics (i.e. ANALYZE TABLE) would need to be added too.

To support the more general case we probably need a way to estimate the cardinality of the intermediate results, based on sampling one or a couple of batches with the particular group by expression.

I added this requirement in this design doc:
https://docs.google.com/document/d/17DCBe_HygkjsoMzC4Znw-t8i1okLGlBkf0kp8DfLBhk/edit?usp=drivesdk

[Dandandan]

I also realize I may have implemented this in an inefficient way. Looking at Spark it does first a partial aggregate followed by a hash repartition followed by another aggregation. I think this will not slow down the low cardinality query that much (as the result set will be small already after first aggregate) while being fast - maybe even faster - for high cardinality aggregates.