When does partial aggregation pay?

Simulate a modelNeeds Python

Problem

Partial aggregation combines rows with the same key inside each partition before the shuffle, so the network carries one row per group per partition instead of one per input row. It pays when each partition holds many rows per group; when groups are many and rows per partition few, nearly every row is its own group, the reduction is about 1, and the hash table built to find that out is pure overhead. The expected number of local groups is G·(1 − e^(−(N/P)/G)).

Simulate P = 20 partitions of N = 2,000,000 rows and measure the real reduction of a local pre-aggregation for different numbers of groups G. Compare it with the bound N/(P·G) and the exact expectation. Then time a Counter on one partition against just forwarding the rows.

The question. How much does local pre-aggregation reduce the data at each G, and what does it cost when it does not?

Predict

With 4 million groups and 100,000 rows per partition, how much does local pre-aggregation shrink a partition?

Barely (under 1.5×)1.5 to 3×About 10×Hundreds of times

Run it

1Install uv once. It fetches Python and the lab's packages for you.
Terminal
```
curl -LsSf https://astral.sh/uv/install.sh | sh
```

2Get the script. It is the whole lab: read it, then download it.

when-does-partial-aggregation-pay.py39 linesDownload

View the code

when-does-partial-aggregation-pay.py

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""When does partial aggregation pay?

N rows over P partitions, grouped by a key with G distinct values. A local
pre-aggregation on one partition reduces its rows by per-partition rows ÷
local groups. Compares the measured reduction with the bound N/(P·G) and the
expected local groups G·(1 − e^(−(N/P)/G)); then times a Counter on one
partition against simply forwarding its rows.
"""
import argparse, json, math, random, time
from collections import Counter

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--n", type=int, default=2_000_000, help="Rows.")
p.add_argument("--p", type=int, default=20, help="Partitions.")
p.add_argument("--groups", default="100,10000,200000,4000000", help="Values of G.")
p.add_argument("--seed", type=int, default=0, help="Random seed.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()
random.seed(args.seed)

per = args.n // args.p; rows = []
for G in map(int, args.groups.split(",")):
    keys = [random.randrange(G) for _ in range(per)]                   # one partition
    t = time.perf_counter(); c = Counter(keys); agg = time.perf_counter() - t
    t = time.perf_counter(); fwd = list(keys); fw = time.perf_counter() - t
    local = len(c); exp = G * (1 - math.exp(-per / G))
    rows.append({"G": G, "bound": args.n / (args.p * G), "measured": per / local, "local_groups": local, "expected_groups": exp,
                 "aggregate_ms": agg * 1000, "forward_ms": fw * 1000})
for r in rows: print(f"G={r['G']:>9,}  N/(P*G)={r['bound']:9.2f}  measured={r['measured']:8.2f}  local groups={r['local_groups']:,} (expected ~{r['expected_groups']:,.0f})  Counter {r['aggregate_ms']:.0f} ms vs forward {r['forward_ms']:.1f} ms")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"groups": rows}}, f)

3Run it in the folder you saved it to, with the defaults first. Then change a knob from the table below and run it again.
Terminal
```
uv run when-does-partial-aggregation-pay.py
uv run when-does-partial-aggregation-pay.py --help     # every knob
```

Knobs
Knob · default	What it changes
--n2,000,000	Rows.
--p20	Partitions.
--groups100, 10,000, 200,000, 4,000,000	Values of G.
--seed0	Random seed.

Our run

One partition of 100,000 rows

One partition of 100,000 rows
G	N/(P·G)	Reduction	Local groups	Expected	Counter (ms)	Forward (ms)
100	1,000	1,000×	100	100	2.19	0.104
10K	10	10×	10K	10K	3.16	0.208
200K	0.5	1.27×	78.7K	78.7K	4.19	0.625
4M	0.025	1.01×	98.8K	98.8K	4.63	0.538

Reference runApple M2 · 8 cores · 8 GB

Output · uv run when-does-partial-aggregation-pay.py

G=      100  N/(P*G)=  1000.00  measured= 1000.00  local groups=100 (expected ~100)  Counter 2 ms vs forward 0.1 ms
G=   10,000  N/(P*G)=    10.00  measured=   10.00  local groups=10,000 (expected ~10,000)  Counter 3 ms vs forward 0.2 ms
G=  200,000  N/(P*G)=     0.50  measured=    1.27  local groups=78,745 (expected ~78,694)  Counter 4 ms vs forward 0.6 ms
G=4,000,000  N/(P*G)=     0.03  measured=    1.01  local groups=98,808 (expected ~98,760)  Counter 5 ms vs forward 0.5 ms

macOS 26.6.2 · Python 3.13.3 · run 2026-10-20

What we found

With 100 groups, pre-aggregation reduced a partition 1,000×; with 10,000, 10×. With 4 million groups it reduced it only 1.01×: 98.8K local groups from 100,000 rows, as the expectation predicts (98.8K). Building that useless table took 4.63 ms against 0.538 ms to forward the rows.

Read it

This is the high-cardinality GROUP BY customer_id case: local aggregation buys nothing and costs a hash table per partition. Engines that measure the reduction on the first rows and switch it off when it is near 1 get both cases right; the arithmetic tells you which case your query is in before you run it.