Partial, two-phase and incremental aggregation

Why is partial aggregation the only state that may be thrown away, and why is COUNT(DISTINCT) the aggregate that breaks the plan?

Partial aggregation: the only operator whose state is a cache

A combiner (the partial aggregation step) reduces rows locally before they move. Its value is the ratio above. Its constraints are worth stating precisely.

The requirement is a fixed-size accumulator. This is the working definition of "algebraic".

A holistic function has no bounded accumulator by definition. This is a theorem about the function, not a limitation of any runtime.

A historical correction. The older MapReduce system required its combiner to be the same function as its reducer. That limited combining to distributive functions: you could combine SUM but not AVG. Modern engines instead split an aggregate into a partial function and a final function with a different intermediate type. For AVG, the partial function outputs (sum, count) and the final function divides. This is exactly what admits the algebraic class. The ability people credit to "better engines" is really a change in the type system: letting the intermediate value differ from the output.

The property that makes partial aggregation unlike anything else here: its state is a cache, not a requirement. If the hash table fills, the engine can flush it, send the partial results on, and start again with an empty table. The answer stays exactly correct, because a final merge follows. Compare:

• A hash join that forgets a build-side row produces a wrong answer.
• Partial aggregation cannot be made wrong by forgetting.

This is why it never runs out of memory, why it can be on by default, and why it can be switched off in the middle of a query when sampling says it is not paying.

Two-phase aggregation, and what happens when the function will not cooperate

For a distributive or algebraic function the plan is settled:

1. Partial aggregation on each executor.
2. Shuffle the partials by group key.
3. Final (merge) aggregation.

There is one shuffle, with volume min(N, P·G). And it is immune to group skew: the hot group was already collapsed to one row per partition, so the final stage receives P rows for the hottest group and P rows for the rarest.

For a holistic function none of this is available. That gives the most useful prediction in this article:

Take a Zipf distribution of pages. SUM(amount) GROUP BY page is fine. COUNT(DISTINCT user_id) GROUP BY page over the same data sends every raw row of the biggest page to one reducer. Worked example: if the home page has 10% of the 50 billion events, that is 5 billion rows. At 8 bytes for the user_id alone, that is 5×10⁹ × 8 B = 40 GB arriving at one executor that has 16 GB of operator memory. The query stalls or fails. Same data, same grouping key, same cluster. The only difference is the algebra of the function. This is why one of these queries is routine and the other appears in incident reviews. It is Omar's first symptom.

The parallel plan for exact COUNT DISTINCT

Exact COUNT(DISTINCT) has one real parallel strategy: shuffle by the distinct column instead of the group key, so identical values meet and can be removed locally. Then shuffle again by the group key and count. Two shuffles instead of one, but now the work is spread out. In SQL form (the usual version hashes the pair, so a hot group is spread over many executors):

-- Step 1: deduplicate (page, user_id) pairs. The engine shuffles by the pair.
-- Step 2: count the surviving pairs per page (a tiny, distributive COUNT).
SELECT page, COUNT(*) AS distinct_users
FROM (SELECT DISTINCT page, user_id FROM events) t
GROUP BY page;

Note what this is: a repartitioning chosen to make a property (distinctness) locally checkable. That is the idea of an enforcer (an operator inserted to supply a required property), applied to a property other than order.

Salting does not work here in its usual form. Salting means adding a random number to a hot key to split it over several partitions. For COUNT(DISTINCT), random salts scatter identical values across partitions, and each partition counts them separately, so the sum counts the same user several times. The salt must be a function of the value being counted. At that point it is the same idea as the two-shuffle plan.

Several distinct aggregates

Several distinct aggregates over different columns cannot share one pass. The input is replicated once per distinct column, with a tag saying which column the copy belongs to. That multiplies the shuffle volume by k. Three COUNT(DISTINCT ...) in one query means three copies of the data on the wire.

Streaming and incremental aggregation: the same invertibility question again

Two different things are called by this name. They rest on different algebraic properties.

Streaming (grouped-input) aggregation. Blocking means "I do not yet know that a group is complete". If the input is sorted on the group key, each group can be emitted the moment it closes, with O(1) state and no pipeline break. Nothing is approximated and nothing is lost. The cost was paid earlier by whoever created the order.

Incremental aggregation. The new state is combine(old state, delta). That needs an associative operation with a neutral element, and once deltas include deletions it also needs an inverse: SUM can subtract a retracted row, MIN and MAX cannot. Which aggregates can be maintained this way, and at what cost, is Incremental view maintenance. It is the same property that decided the cost of sliding window frames earlier (a sliding SUM updates in O(1); a sliding MAX cannot undo): one algebraic fact, two apparently unrelated consequences.