Distributed Query Planning
Concept. Partition the data, run local operators, and shuffle rows when matching keys must meet.
Scaling Queries Across Multiple Machines
When your data outgrows a single machine, the query has to be split across nodes, run in parallel, and its partial results reassembled.
What is "Shuffle"?
Shuffle is data relocation across nodes. It gets all rows with the same key onto the same machine before joining or grouping. It is hash partitioning over the network.
Figure: Send matching keys to the same machineDistributed shuffle: rows starting on multiple nodes are hashed on the join key and sent over the network so all rows with the same key land on the same target node, ready for local join. This is data movement across machines, not a disk or RAM story. Color key: mint follows one key (user_id 42) from every source node, over the network, onto its single colocated node; grey is the other keys and the node structure.
Figure 1. Rows begin scattered across multiple nodes. Hashing each row on the join key picks its target machine, so every row sharing a key travels over the network to the same node. Once the keys sit together, each node runs an ordinary local join on the rows it now owns. This is exactly hash partitioning with the network as the destination, so the cost is the same 2N IO plus the network transfer.
| Local hash partitioning | Distributed shuffle |
|---|---|
hash(key) % B picks a partition file on the same disk |
hash(key) % nodes picks a target machine on the network |
| Output: B files on local disk | Output: each node receives the keys it owns |
| Cost: 2N IO | Cost: 2N IO + network transfer |
Same algorithm, different destination. See Hash Partitioning for the local primitive.
Reaggregation Patterns: The Two-Phase Execution
Problem: Global grouping across 1,000 nodes would flood the network with billions of tiny rows.
Phase 1: Local Pre-Aggregation
Each node groups and aggregates its local data.
-
Input: 1,000,000 "listen" rows per node.
-
Output: 1,000 "artist totals" per node.
-
Impact: Shrinks data by 1,000x before network transfer.
Phase 2: Global Merge
Nodes shuffle their partial results by the grouping key (e.g., artist_id) to a coordinator node.
-
Action: Coordinator aggregates partial counts to derive the true total.
-
Why: Produces correct global totals while cutting network traffic.