Parallel-Grouped-Aggregation-in-DuckDB-DuckDB.pdf

Parallel Grouped Aggregation in

DuckDB - DuckDB

https://duckdb.org/2022/03/07/aggregate-hashtable.html

TL;DR: DuckDB has a fully parallelized aggregate hash table that can efficiently

aggregate over millions of groups.

Grouped aggregations are a core data analysis command. It is particularly important for

rows of the input (like a normal SELECT ), GROUP BY returns as many rows as there are

groups in the data. Consider this (weirdly familiar) example query:

SELECT

l_returnflag,

l_linestatus,

sum(l_extendedprice),

avg(l_quantity)

FROM

lineitem

GROUP BY

therein as “groups”. The SELECT clause contains four (not five) expressions: References

to the grouping columns, and two aggregates: the sum over l_extendedprice and the avg

over l_quantity . We refer to those as the “aggregates”. If executed, the result of this

query looks something like this:

l_returnflag l_linestatus sum(l_extendedprice) avg(l_quantity)

N O 114935210409.19 25.5

R F 56568041380.9 25.51

A F 56586554400.73 25.52

N F 1487504710.38 25.52

In general, SQL allows only columns that are mentioned in the GROUP BY clause to be

the groups can occur in the input table in any order. Were the input already sorted on

the grouping columns, computing the aggregation would be trivial, as we could just

compare the current values for the grouping columns with the previous ones. If a

change occurs, the next group begins and a new aggregation result needs to be

computed. Since the sorted case is easy, one straightforward way of computing

grouped aggregates is to sort the input table on the grouping columns first, and then

use the trivial approach. But sorting the input is unfortunately still a computationally

expensive operation despite our best efforts. In general, sorting has a computational

complexity of O(nlogn) with n being the number of rows sorted.

Hash Tables for Aggregation

To add n rows to a hash table we are looking at a complexity of O(n) , much, much

better than O(nlogn) for sorting, especially when n goes into the billions. The figure

above illustrates how the complexity develops as the table size increases. Another big

advantage is that we do not have to make a sorted copy of the input first, which is going

to be just as large as the input. Instead, the hash table will have at most as many

entries as there are groups, which can be (and usually are) dramatically fewer than

input rows. The overall process is thus this: Scan the input table, and for each row,

update the hash table accordingly. Once the input is exhausted, we scan the hash table

to provide rows to upstream operators or the query result directly.

So, hash table it is then! We build a hash table on the input with the groups as keys and

the aggregates as the entries. Then, for every input row, we compute a hash of the

group values, find the entry in the hash table, and either create or update the aggregate

states with the values from the row? Its unfortunately not that simple: Two rows with

different values for the grouping columns may result in a hash that points to the same

hash table entry, which would lead to incorrect results.