One MPP node runs at 100% while the rest idle during a large GROUP BY — what is happening and how do you fix it?

## Why one node ends up with everything A hash exchange assigns a row to a node by `hash(key) mod N`. This is deterministic and, for a uniformly distributed key, near-perfectly even. But it distributes *values*, not rows: all rows sharing a value are inseparable and must land together, because that is the entire point of the shuffle. So the per-node load is the sum of the frequencies of the values that hash to that node. When the key's value distribution is heavily lopsided, the shuffle faithfully reproduces the lopsidedness. A query's stage completes when its slowest worker completes, so a node holding 40% of the rows makes the stage take roughly 40% of the single-threaded time no matter how many nodes are idle beside it. The cluster's aggregate throughput is irrelevant; the critical path runs through one worker. ## The usual culprits - **NULLs.** In many engines all NULLs hash to a single bucket. A left join or grouping on a mostly-NULL column concentrates that entire population on one node. - **Sentinel values.** `-1`, `0`, `'unknown'`, `'N/A'`, the epoch date, or an `unassigned` foreign key inserted by an upstream ETL default. These are semantically "no value" but syntactically one very frequent value. - **Whale entities.** One tenant, one customer, one bot account, one popular product that legitimately generates orders of magnitude more rows than the median. - **Low-cardinality keys.** Grouping or joining on a column with fewer distinct values than the cluster has workers guarantees that most workers get nothing at all. ## Diagnosing it Two measurements settle it. First, from the query profile or per-node execution statistics, look at rows or bytes *received* per worker for the stage that is slow — a healthy hash exchange shows near-identical counts, a skewed one shows one worker one or two orders of magnitude above the median. Second, in SQL, measure the key directly: ```sql SELECT user_id, COUNT(*) AS n FROM events GROUP BY user_id ORDER BY n DESC LIMIT 20; SELECT COUNT(*) FILTER (WHERE user_id IS NULL) AS nulls, COUNT(*) AS total FROM events; ``` If the top value's count is a meaningful fraction of the total, the shuffle cannot be balanced by any placement of that key. Distinguish this from a **straggler**: a straggler is one node slow for an environmental reason — a degraded disk, a noisy co-tenant, a cold cache, an uneven file assignment — and it moves around between runs. Skew is deterministic: the same node (relative to the key's hash) is slow every single run, and the profile shows it *received* more data, not that it processed the same data slower. ## Fixes, roughly in order **Eliminate junk values.** If the hot key is a sentinel that carries no meaning, filter it out before the shuffle or rewrite the join so those rows do not participate. This is the cheapest fix and often the correct data-quality fix as well. **Make sure pre-aggregation is happening.** For a `GROUP BY`, a two-phase plan aggregates on each node before the exchange, so the hot group crosses the network as one partial row per node rather than as billions of raw rows. Skew then costs almost nothing for decomposable aggregates. If the plan is not doing this — often because the aggregate is not decomposable, such as an exact `COUNT(DISTINCT ...)` — that is the real problem to attack. **Salt the key.** Add a synthetic bucket component to the shuffle key so the hot value is split across many workers, then combine the partial results in a second, much smaller aggregation: ```sql SELECT customer_id, SUM(part) AS total FROM ( SELECT customer_id, MOD(order_id, 32) AS salt, SUM(amount) AS part FROM orders GROUP BY customer_id, MOD(order_id, 32) ) s GROUP BY customer_id; ``` Salting turns one hot bucket into 32 warm ones. It costs an extra aggregation stage, so apply it only to the skewed step, and only when pre-aggregation alone is insufficient. **Split hot from cold.** Run the query twice: once for the handful of hot keys, using a strategy that avoids redistributing them (for a join, broadcasting the other side works well because the hot rows then never move), and once for everything else with the normal plan. `UNION ALL` the results. Some engines do a version of this automatically as skew-aware or adaptive join handling. **Change the layout.** If the same key skews every query, the underlying model may be wrong — a composite key, a different grain, or a denormalized table that removes the join entirely may be the durable answer. ## What does not work Adding nodes does not help: the hot bucket remains indivisible and still lands on exactly one worker, so the critical path is unchanged while you pay for more idle machines. Raising the memory limit only converts a spill into a slow in-memory run. Reordering the query or adding hints without addressing the key distribution moves the symptom rather than the cause.