Skew: the assumption of uniformity, priced

What does a hot key do to a distributed join, which fixes split it, and when is there no fix at all?

A distributed join's run time is its maximum partition, not the mean, so a hot key is the serial fraction of Amdahl's law. Under the usual Zipf distribution of keys the hottest one holds a predictable share, about 7% for a million distinct keys, which caps the speedup near 14 however many executors you add (the data-parallel ceiling derives it). On Omar's 200 executors, 93% of the cluster sits idle at the end, waiting for one task, because the uniformity that hash partitioning assumes was never there. How such a hot key is routed is data movement's subject; what a join can do about it is this page's.

One special case is free to fix and often forgotten: NULL keys. They all hash to the same partition and match nothing in an inner join. They are pure waste, and by the three-valued logic result in Relational operators they can be filtered from both inputs before the shuffle at no semantic cost.

The general fixes all make the same move: turn one oversized partition into s smaller ones by replicating the other side s times.

Salting. Append a random value 0 … s−1 to the hot key on the large side, and replicate the matching rows of the small side once per salt value. The maximum partition drops by a factor s. The cost is s× the small side's hot-key rows. It requires knowing which keys are hot, and it writes that knowledge into the query text. That is why it ages badly: hot keys move, and last quarter's salt list is this quarter's skew.

-- Salting key 0 with s = 16 (illustrative)
-- Large side: random salt for the hot key only
SELECT o.*, CASE WHEN customer_id = 0 THEN CAST(RANDOM()*16 AS INT) ELSE 0 END AS salt
FROM orders o;
-- Small side: replicate the hot key's rows once per salt value
SELECT c.*, s.salt FROM customers c
JOIN (SELECT * FROM generate_series(0, 15) AS salt) s
  ON c.customer_id = 0 OR s.salt = 0;     -- non-hot keys get salt 0 only
-- Then join on (customer_id, salt)

Skew splitting. The same move, done at run time. After the shuffle's map stage, the engine detects the oversized partitions, splits them, and broadcasts the matching rows of the other side to the pieces. The interesting part is conceptual: the right join algorithm differs per key. Broadcast for the hot keys, shuffle for the tail, and a union of the results. No static plan can say that. It belongs to adaptive execution (re-planning while running); the optimizer owns the decision and the algorithm is here.

Sampled range partitioning. Replace hash partitioning with boundaries chosen from a random sample, which equalizes partition sizes by construction at the price of a sampling pass. This is randomization replaces coordination: a sample stands in for the exact global computation.

The case with no fix: two-sided skew. When the hot key is frequent in both inputs, the output for that key is f_R(v)·f_S(v) rows, and it is genuinely quadratic. No partitioning scheme reduces work that the answer really contains. The only remedies change the question: pre-aggregate one side before the join, or accept an approximate result. It is worth saying plainly, because writing about salting often implies that skew is always a structural problem.

Case, resolvedConstructed case · Omar and Tarini at Shopsy

199 executors idle, one running for two hours

A join on customer_id left 199 of 200 executors idle while one task held customer 0 and every NULL key, and a 'price at order time' range join had run for a day. Where it started: Join algorithms →

Tarini fixes both reports with ideas from this article.

199 idle executors. The shuffle sends every row of key 0 (140 million rows) and every NULL key (40 million rows) to one partition each, and the job runs at the speed of its largest partition. She does three things:

She filters the NULL keys from orders before the shuffle. They match nothing in an inner join, so the answer does not change, and 40 million rows disappear.
She splits key 0 (skew splitting). Guest orders are joined to the single customers row with id 0, which is one row. A broadcast of that one row to the pieces of the hot partition is essentially free. The remaining keys use the normal shuffle.
No two-sided skew exists here (key 0 appears once in customers), so the fix is complete. If it had appeared millions of times on both sides, no partitioning could help.

The day-long price lookup. The range join plus ROW_NUMBER() built every (order, earlier price) pair. She replaces it with an as-of join, a merge that keeps one price row per product while walking the sorted data. The query that ran for a day now runs in minutes.