In this lesson, you will learn how to minimize cross-node latency by selecting the optimal join strategy—Broadcast or Shuffle—based on data distribution.
When performing a join across a distributed database, moving massive amounts of data over the network is often the most expensive part of the query. Distributed join planning involves choosing the execution strategy that minimizes this "network tax."
Broadcast Joins
A Broadcast Join is ideal when one table is relatively small (often a dimension table) while the other is large (a fact table). Instead of moving both tables, the query planner copies the entire small table to every node containing a shard of the large table.
- Why choose this? It eliminates the need to reshuffle the large table. Since the large table stays put, you avoid massive data movement.
- The constraint: It is only viable if the smaller table fits into the memory of the compute nodes.
/* Query Planner decides: Broadcast small 'users' table to all partitions of 'orders' */
SELECT * FROM orders
JOIN users ON orders.user_id = users.id;
Shuffle Joins
A Shuffle Join (or Repartition Join) is used when both tables are too large to fit in memory. The database re-partitions both tables across the cluster based on the join key (e.g., user_id). By hashing the join key, the system ensures that rows with the same user_id from both tables land on the same physical node.
- Why choose this? It provides linear scalability for massive joins.
- The cost: You incur significant "all-to-all" network traffic as both tables are effectively reorganized across the cluster.
Planning for Performance
The query planner makes this decision based on metadata stored in the Scalable Metadata Store. If the planner identifies a table as "tiny" (based on Bloom filters or statistical histograms), it will prefer a Broadcast Join to bypass the network shuffle. If both tables are identified as "large," it forces a Shuffle Join to avoid memory overflow.
Effective distributed join planning is a constant balancing act between memory availability and network throughput. When designing your schema, ensuring high-cardinality join keys are indexed appropriately assists the optimizer in making these routing decisions efficiently.
Key Takeaway: Choose a Broadcast Join to keep large data stationary when joining with small tables, and use a Shuffle Join to parallelize the processing of two large, co-partitioned datasets.