Hash spill and the arithmetic of a single pass

How much data can one partitioning pass handle, and why can no number of passes handle one hot key?

Hash spill, and the arithmetic of a single pass

When the build side does not fit, a hash join switches to Grace hash partitioning: it splits both inputs by a hash of the join key into P partitions on disk, then joins them one pair at a time (Grace hash join derives the algorithm). Each build partition must fit in memory (P ≥ N/M) and each open partition needs an output buffer of size B (P ≤ M/B), so one partitioning pass handles at most

Memory enters quadratically in the one-pass capacity. With M = 1 GB and B = 1 MB buffers, one partitioning pass can handle up to 1 GB × 1 GB / 1 MB = about 1 TB. Beyond that you need recursive passes, and each pass costs another 2N of I/O. Doubling memory quadruples the data you can handle in one pass (M = 2 GB gives 4 TB). That is a far stronger argument for the last increment of memory than linear intuition suggests. It is the same arithmetic that governs the fan-in of an external sort.

The buffer size B is a granularity choice

The size B is itself a granularity trade-off. If it is too small, writes stop being sequential, which destroys the entire reason for partitioning. If it is too large, P collapses (because P ≤ M/B) and you need another pass. The fixed cost of a non-sequential write gives a natural buffer size, just as the cost of fsync gives the natural batch size of group commit.

Skew breaks it, and recursion cannot always fix it

Hash partitioning also assumes keys are spread evenly: "independence assumed, but false" again. Several keys that land together by bad luck can be re-partitioned with a different hash function. A single key whose group is bigger than memory cannot: no hash function separates a key from itself, so recursive partitioning does not converge (the dead end). This is the most common production hash-join failure there is. It is the second incident: 300 million readings under the key 0000000.

The escapes, each with its price

• Fall back to sort-merge for the oversized partition. Sort has no fitting requirement (Sort spill, and the comparison that decides operator choice), so it always finishes. Cost: a second code path, and the log factor of sorting.
• Nested loop over that partition. Compare every build row with every probe row. Quadratic in the partition size, so acceptable only if it is small relative to the whole.
• Skew-aware planning. Detect hot keys from statistics and handle them separately (for example, broadcast the hot key's rows to all workers, or process it as its own stream). Cost: a dependency on the statistics being right about exactly the tail they are worst at.
• Change the algebra. This applies to aggregation. If the aggregate is algebraic (sum, count, min, max; the result for a group can be built from small partial results that you combine in any order), then partial aggregation reduces rows before the spill, and the state per group is O(1) (a constant size), however many rows the group has. A hot key is then completely harmless. If the aggregate is holistic (exact median, exact distinct count; it needs all the values), the state grows with the group size and you are stuck.

This is a recurring idea: the right algebra makes spill unnecessary. Associativity (the result does not depend on how you group the operations) lets you reduce before you write. It is the same property that elsewhere makes parallel execution possible and makes some data structures free of conflicts. Here it is sold for a third purpose.

One-line compression: a single partitioning pass handles M²/B data, and no number of passes handles one key bigger than memory.