Why does an MPP query spill to disk after a shuffle even when total cluster memory looks sufficient?

## Cluster memory is an illusion "The cluster has 2 TB of RAM" is almost never the number that matters. That total is split three ways before an operator sees any of it: 1. **Per node.** Only the memory on the node where a worker runs is usable by that worker; a free gigabyte on another node cannot be borrowed. 2. **Per worker.** Each node runs several parallel workers, so one node's RAM is divided again. 3. **Per concurrent query.** Warehouses admit multiple queries at once, and each is given a share (or competes for a shared pool with a per-query cap). So an operator's real budget can be a small fraction of a percent of the headline number. The relevant question is never "does the dataset fit in the cluster" but "does *this worker's partition* of the intermediate result fit in *this operator's* budget right now." ## Why the shuffle is where it bites Below an exchange, work is local and streaming: a scan reads a block, filters it, and passes it on, holding almost nothing. Above an exchange sits an operator that accumulates — a hash join building its table, a final aggregate holding one entry per group, a sort holding a run. Those are the memory-hungry operators, and they are fed by whatever the exchange delivered. That delivery is where the even-split assumption fails. If the shuffle key is skewed, one worker receives many times the average partition and blows through a budget that was perfectly adequate for its peers. This is the signature MPP memory failure: the profile shows spill on one worker and none on the other 63. ## Row width is half the problem Shuffle and hash-build costs are measured in bytes, not rows. A plan that carries thirty columns through an exchange when the operators above need four is moving and buffering many times the necessary volume. Wide `VARCHAR` columns and semi-structured blobs dominate quickly. Projecting only what is needed — and pushing predicates below the exchange so fewer rows travel at all — often removes the spill entirely without touching a single setting. ## What spilling actually costs When an accumulating operator exceeds its budget, it writes some of its partitions out and continues, then reads them back for a later pass. The costs compound: - **Extra I/O**, both write and read, for data that was already in memory once. - **Extra passes.** If the spilled partitions are themselves too large when re-read, the operator recurses, multiplying the I/O. - **Contention.** Local scratch space is shared with other queries on that node, so one spilling query slows its neighbours. - **Remote spill.** Some cloud engines, having little local disk, spill to object storage. That is orders of magnitude slower than local SSD and turns a degradation into a stall; it is worth knowing whether your engine reports local and remote spill separately, because remote spill is a much louder alarm. Spilling is a graceful degradation, not an error — it is what lets a query bigger than memory complete at all. The failure to avoid is treating any spill as fatal, or treating a large one as normal. ## Diagnosing it Read the profile for the stage that regressed: bytes spilled, per worker. Then ask which of three shapes it is. - **Uniform spill across all workers** means the intermediate is genuinely bigger than the cluster's per-query memory. Options: more memory per query (fewer concurrent queries, or a larger compute cluster), or a rewrite that produces less intermediate data. - **Spill on one worker only** means skew. Fix the key distribution; raising memory only buys time. - **Spill that appeared without a data-volume change** usually means a plan change — an estimate flipped a broadcast into place, or a filter stopped being pushed down after a rewrite. Compare plans, not settings. ## Reducing what crosses the exchange In rough order of leverage: 1. **Project.** Name only the columns needed above the join or aggregate. 2. **Filter early.** Every predicate that can be evaluated before the exchange removes rows from the network and from the hash table. 3. **Pre-aggregate.** A partial aggregation below the shuffle can shrink billions of rows to thousands of partial groups. 4. **Fix skew.** Salt, split, or eliminate hot key values so partitions are even. 5. **Reduce concurrency or resize.** Only after the above — a bigger cluster hides an inefficient plan at recurring cost, and lower concurrency trades one workload's latency for another's. The order matters because the first three shrink the work permanently and for every future run, while the last one rents headroom by the hour.