The output does not grow with the input
Why is aggregation the one operator that can be done in pieces, and how do hash and sort aggregation spend memory?
Every operator so far produced output in proportion to its input. A filter shrinks the input by its selectivity. A join multiplies by how often keys repeat. A sort keeps the row count exactly. Aggregation is the first operator whose output size is almost independent of its input size. N rows go in, G groups come out. G is set by the domain of the data, not by its volume. A billion rows grouped by country produce about two hundred rows.
That single fact generates the whole article, because it allows something a join can never do. Since the output is a summary, the summarizing can happen before the data is brought together. Each machine collapses the rows it holds, and only the collapsed form travels over the network. A join cannot do this: you cannot "pre-join" one partition, because a matching row may live on any other machine. Aggregation is the only operator here that can be legally and profitably done in pieces before the pieces meet.
So the main question is not "hash or sort?" It is: can the reduction happen early, and if not, why not? The answer is almost entirely a property of the aggregate function, not of the engine. That is why the classification from the algebra of aggregate functions (distributive, algebraic, holistic) is used so heavily here.
The core tension, as one ratio
Let N be the number of rows, G the number of distinct groups, and P the number of partitions (here, executors).
- Without partial aggregation, the shuffle (the step that moves rows between executors so that equal keys meet) moves all N rows.
- With partial aggregation, each partition first reduces its local share to at most min(N/P, G) rows. The shuffle then moves at most min(N, P·G) rows.
The reduction factor is
- rows to aggregate
- partitions (executors)
- distinct groups
Data table
| N / (P·G) | Reduction from partial aggregation |
|---|---|
| 0.01 | 1.01× |
| 0.012 | 1.01× |
| 0.013 | 1.01× |
| 0.015 | 1.01× |
| 0.018 | 1.01× |
| 0.021 | 1.01× |
| 0.024 | 1.01× |
| 0.027 | 1.01× |
| 0.032 | 1.02× |
| 0.037 | 1.02× |
| 0.042 | 1.02× |
| 0.049 | 1.02× |
| 0.056 | 1.03× |
| 0.065 | 1.03× |
| 0.075 | 1.04× |
| 0.087 | 1.04× |
| 0.1 | 1.05× |
| 0.115 | 1.06× |
| 0.133 | 1.07× |
| 0.154 | 1.08× |
| 0.178 | 1.09× |
| 0.205 | 1.11× |
| 0.237 | 1.12× |
| 0.274 | 1.14× |
| 0.316 | 1.17× |
| 0.365 | 1.19× |
| 0.422 | 1.23× |
| 0.487 | 1.26× |
| 0.562 | 1.31× |
| 0.649 | 1.36× |
| 0.75 | 1.42× |
| 0.866 | 1.49× |
| 1 | 1.58× |
| 1.15 | 1.69× |
| 1.33 | 1.81× |
| 1.54 | 1.96× |
| 1.78 | 2.14× |
| 2.05 | 2.36× |
| 2.37 | 2.62× |
| 2.74 | 2.93× |
| 3.16 | 3.3× |
| 3.65 | 3.75× |
| 4.22 | 4.28× |
| 4.87 | 4.91× |
| 5.62 | 5.64× |
| 6.49 | 6.5× |
| 7.5 | 7.5× |
| 8.66 | 8.66× |
| 10 | 10× |
| 11.5 | 11.5× |
| 13.3 | 13.3× |
| 15.4 | 15.4× |
| 17.8 | 17.8× |
| 20.5 | 20.5× |
| 23.7 | 23.7× |
| 27.4 | 27.4× |
| 31.6 | 31.6× |
| 36.5 | 36.5× |
| 42.2 | 42.2× |
| 48.7 | 48.7× |
| 56.2 | 56.2× |
| 64.9 | 64.9× |
| 75 | 75× |
| 86.6 | 86.6× |
| 100 | 100× |
| 115 | 115× |
| 133 | 133× |
| 154 | 154× |
| 178 | 178× |
| 205 | 205× |
| 237 | 237× |
| 274 | 274× |
| 316 | 316× |
| 365 | 365× |
| 422 | 422× |
| 487 | 487× |
| 562 | 562× |
| 649 | 649× |
| 750 | 750× |
| 866 | 866× |
| 1,000 | 1,000× |
which is the average number of rows per group per partition.
Production engines therefore do not decide this at plan time. They sample the first tens of thousands of rows, measure the hash table's hit rate (how often a row finds its group already in the table), and abandon partial aggregation if the reduction is not happening. This exploits an information gap. The planner had only an estimate of NDV (the number of distinct values), and NDV estimates are the least reliable statistic in any catalogue. The executor sees the real keys.
Hash aggregation, and the spill choice the algebra makes for you
A hash aggregation keeps one table entry per group: the key plus a fixed-size accumulator (the running value, such as a sum). For each input row it finds the group's entry and updates it in place.
Two properties distinguish it from the hash join you studied earlier. Both are easy to carry over wrongly.
1. State is O(G), not O(N). A hash join's build side holds every row of one input. A hash aggregation holds only each distinct key once. The table is also simpler: no chains for multiple matches, and fixed-width accumulators updated in place. So it is much more cache-friendly than a join's build table with the same number of entries. It also uses the other equality (grouping equality, where all NULLs fall into one group), so it is not even the same comparison function as a join's.
When memory fills, there is a choice a hash join does not have
| Spill strategy | What is written to disk | Requirement | Cost |
|---|---|---|---|
| Spill the input rows, partitioned Grace-style by group key (split into disk partitions by hash, then aggregate each partition alone) | raw rows | none; always correct | high: N rows of I/O |
| Spill the hash table's contents as partial aggregates, clear the table, keep going, merge the spilled partials at the end | G-sized batches of partial results | the function must be distributive or algebraic (partials must be mergeable) | often about three orders of magnitude less I/O |
| Keep the hottest groups resident and spill only the cold ones | rows of cold groups only | none | depends on skew |
A holistic aggregate cannot use the second strategy, because it has no mergeable partial result.
The third strategy produces a surprise. Under a skewed distribution of groups, a small resident set absorbs most rows. Suppose the top 1% of groups carry 60% of the input.
So skew, which was the enemy in every earlier article, is an asset here. The same Zipf distribution (a very uneven distribution where a few keys are huge and most are tiny) that caps a distributed join's speedup at 14x makes a hash aggregation's working set small.
This sharpens the "maximum, not mean" principle from the start of this mechanism. "Variance costs more than volume" is true when the work is proportional to the rows. It reverses when the work is proportional to the distinct values. A join's cost follows rows, so concentration hurts. An aggregation's cost follows groups, so concentration helps. The principle needs this boundary condition: concentration is expensive for operators that process rows and cheap for operators that reduce them.
Sort-based aggregation: O(1) memory, and the only cheap exact holistic
Sort-based aggregation sorts the input on the group key, then makes one pass holding a single accumulator and emits each group when it closes (when the key changes). Memory is O(1). The cost is the sort.
Two situations make this a good trade, not merely a safe one.
- The query needs its output ordered by the grouping key. This is extremely common:
GROUP BY x ORDER BY xis most of analytics. The sort was owed anyway, so the aggregation is free. - The input already arrives grouped. This happens with a clustered table layout, a bucketed table (a table stored pre-split by a key), or a previous merge join on the same key. Then the sort disappears and the aggregation costs only the pass.
A third situation is the one people miss. Sort by (group key, value), not by group key alone. Now every holistic aggregate becomes a position lookup:
- The median of a group is the row at the midpoint of the group.
- p99 (the value below which 99% of the values fall) is the row at 99% of the group's offset.
- The whole job is a counted scan.
Exact quantiles are the aggregate that hash aggregation handles worst, because it must keep every value of every group in memory. In the sorted plan they become the cheapest thing it does.
Sort-based aggregation is the only strategy that computes exact holistic aggregates without holding a group's full contents in memory. This is why it survived the hash era. It is not a legacy fallback. It owns two cases outright: exact holistic aggregates, and queries that need ordered output.