MPP Execution & Shuffle

## The co-location invariant An MPP engine can only combine rows that are physically on the same machine. Two rows that must be compared, joined, or aggregated together have to arrive in the same worker's memory. That gives every relational operator a placement precondition: - A **join** on `a.k = b.k` needs all `a` rows and all `b` rows with the same `k` on one node. - A **`GROUP BY g`** needs all rows with the same `g` on one node to produce a final group total. - A **`DISTINCT`** or `COUNT(DISTINCT ...)` needs all occurrences of a value on one node to deduplicate it. - A **global `ORDER BY`** needs a single ordered stream, which ultimately means one merge point. When the current placement already satisfies the precondition — the table was distributed on exactly that key — nothing moves. When it does not, the engine has to re-place the rows at runtime. That re-placement is the exchange. ## What the exchange operator does An exchange has a sender side and a receiver side. On each node the sender applies a partitioning function to every outgoing row — typically `hash(key) mod N` — and writes it into a buffer destined for the node that owns that bucket. The receiver on each node accepts rows from all senders and feeds them to the operator above. Because every sender may talk to every receiver, a hash exchange is an all-to-all pattern; its cost grows with the number of rows moved and with how many bytes each row carries. Three variants show up in almost every engine's plan output: - **Hash repartition** — the general case above, used to co-locate join or grouping keys. - **Broadcast (replicate)** — every node sends its whole input to every other node, so each node ends up with a full copy of one side. Used when one input is small. - **Gather / merge** — all nodes send to a single point, usually the coordinator, for the final result or a global sorted `LIMIT`. Some engines also expose a *round-robin* or *rebalance* exchange, used to even out parallelism rather than to co-locate a key. ## Stages and why they behave like barriers Exchanges cut the plan into stages (fragments). Everything below an exchange runs entirely locally on the rows a node already holds; everything above it consumes rows that arrived from the network. Reading a plan therefore means reading it stage by stage: how many exchanges are there, how many rows and bytes does each one move, and which stage dominates the runtime. The exchange itself normally *streams* — rows flow as they are produced. What creates a barrier is usually the operator sitting above it: a hash join must fully build its hash table before probing, a final aggregate must see all input before emitting a group, a sort must see everything before emitting the first ordered row. So the stage boundary behaves like a synchronization point, and the slowest sender holds up the whole stage. ## Partial aggregation: the standard shuffle reduction The most important optimization around an exchange is doing work *before* it. For `SELECT g, SUM(x) FROM t GROUP BY g`, a naive plan shuffles every row of `t` on `g`. A good plan instead aggregates locally first: ```sql -- conceptually, per node: -- partial: SELECT g, SUM(x) AS s FROM local_rows GROUP BY g -- exchange on g -- final: SELECT g, SUM(s) FROM partials GROUP BY g ``` Now the network carries at most one row per distinct group per node instead of one row per input row. When the group count is small relative to the row count, this turns a terabyte shuffle into a megabyte shuffle. It works because `SUM`, `COUNT`, `MIN`, `MAX` and `AVG` (as sum plus count) are decomposable. It does not work as cleanly for `COUNT(DISTINCT ...)`, which is why exact distinct counts are notoriously the most shuffle-heavy aggregates. The same principle applies to projection: only the columns an upper stage actually needs should cross the exchange. Selecting fewer columns, filtering earlier, and pushing predicates below the exchange all reduce shuffle bytes directly. ## When no exchange is needed If both join inputs are already distributed on the join key, matching rows are already co-located and the join runs entirely locally — the plan shows no exchange between the scans and the join. The same holds for a `GROUP BY` on the distribution key: each node can produce final group totals without talking to anyone, and only the small result set is gathered. Designing physical layout so that the hottest joins and groupings are already co-located is the main lever an engineer has over shuffle cost. ## What to look for in a plan Count the exchanges. For each, note the partitioning (hash on which column, or broadcast), the estimated versus actual row count, and the bytes moved. A stage whose actual row count is orders of magnitude above the estimate is the classic cause of a bad exchange choice, and an exchange placed below rather than above a filter is a classic missed optimization.