Broadcast Joins
When one side of a join is small enough to send everywhere, the large side never moves and the shuffle disappears. The whole technique rests on a size estimate — and on what happens when that estimate is wrong.
Who needs this, what one row is, and why the obvious build breaks
Every lesson starts from the consumer, because designing from the source outward is this domain's characteristic mistake.
Why did the same join take minutes yesterday and hours today, with the same code and almost the same data?
Every model that joins a large fact table to a small dimension, which is most of a warehouse. The difference between a broadcast and a shuffle join on that pattern is the difference between a job that fits its window and one that does not (Star Schema).
The unit is one side of one join: the build side, which is replicated, and the probe side, which stays where it is. Everything here follows from which side is which and how big the build side really is once it is in memory.
Write the join and let the planner choose. It usually chooses well — a small dimension against a large fact table is the case broadcast exists for, and the planner recognises it from file statistics without being asked.
The small side grows past the planner's threshold. The plan silently switches to a shuffle join, the runtime multiplies, and nothing in the code changed (Query Optimizers).
- The small side grows past the planner's threshold. The plan silently switches to a shuffle join, the runtime multiplies, and nothing in the code changed (Query Optimizers).
- Statistics are stale after a large load, so the planner believes a large table is small, broadcasts it, and the driver runs out of memory assembling it (The Spark Execution Model).
- The build side is small on disk and large in memory. Columnar data is compressed and encoded; the hash table built from it is neither, and a modest file can become a large object graph (Why Analytical Data Compresses).
- A filter that would have made the build side tiny is not applied before the broadcast, because it sits behind a function the planner cannot see through (Predicate Pushdown).
- The join is broadcast on a cluster with many executors, so the small side is materialised once per executor — and the aggregate memory it occupies grows with the cluster rather than with the data.
What is actually happening
- In a shuffle join, both sides are repartitioned by the join key so that matching rows meet. Both sides cross the network, and the cost is dominated by the larger one (The Shuffle).
- In a broadcast join, the small side is collected, sent to every executor, and built into a hash table there. Each task then probes that table with its own partition of the large side, which never leaves the machine it was read on (Hash Table).
- The decision is made by the planner from an estimated size of the build side, compared against a configured threshold. It is an *estimate*, derived from table statistics and file metadata, and it is made before any data is read (Cost-Based Optimization).
- Cost scales with executor count rather than with the large side: the build side is transferred and materialised once per executor. A broadcast that is comfortable on ten executors can be uncomfortable on two hundred.
- Join type constrains which side may be replicated. You can broadcast the side whose unmatched rows do not need to be preserved — for a left outer join, that is the right side. Broadcasting the wrong side would lose rows, so the planner will not, and neither should you (Joins: INNER, LEFT, RIGHT, FULL, CROSS, SELF).
The same join, two physical plans
A fact table joined to a dimension is the most common shape in analytics, and the engine has two ways to execute it. In the first, both sides are redistributed by the join key so that matching rows land together. In the second, the dimension is sent to every executor and the fact table never moves at all.
The second plan removes an entire stage. There is no exchange, no barrier, no map-side spill and no reduce-side merge for the large side — the join becomes a per-partition operation and the stage is as parallel as the fact table's partition count allows.
That is a large enough win that it changes how dimension tables should be designed. A narrow, well-maintained dimension that stays broadcastable is worth more than a wide one that is more convenient to query directly, because the narrow one keeps every fact join cheap (Dimension Tables).
Repartition both the fact table and the dimension by `customer_id`. The whole fact table is serialized, spilled, fetched and merged; the stage is a barrier; and a skewed customer key concentrates rows on one reducing task.
Filter and project the dimension, send it to every executor, and probe it locally from each fact partition. No exchange for the fact table, no barrier, no skew from the join key, and the stage scales with the fact table's partition count.
The join needs matching rows to be co-located, and there are two ways to achieve that: move both sides to a common place, or put one side everywhere. When one side is small, replicating it costs far less than redistributing the other — and the cost of replication scales with executor count while the cost of redistribution scales with the large table.
When the estimate is wrong
Everything above depends on a number the planner guessed before reading any data: how large the build side will be. That estimate comes from table statistics and file metadata, and there are several ordinary ways for it to be wrong in either direction.
Too low, and the engine broadcasts something that does not fit. The failure is loud but confusing, because it lands on the driver during assembly — after the plan looked fine and before most of the work started. Too high, and the engine shuffles something that would have fitted comfortably, which costs a stage and a barrier on every run and produces no error at all.
The asymmetry matters when choosing a response. An over-estimate is a silent tax; an under-estimate is a 3 a.m. page. Raising the threshold to force more broadcasts converts taxes into pages, which is a decision worth making deliberately rather than in response to one slow job.
| Trigger | Symptom | Cause | Response |
|---|---|---|---|
| Statistics are stale after a large load. | Driver out of memory during broadcast build; the job dies before doing meaningful work. | The planner estimated the build side from statistics collected when the table was much smaller. | Refresh statistics as part of the load, and treat stale statistics as a pipeline defect rather than a maintenance chore (Cost-Based Optimization). |
| The build side is compressed columnar data. | A file of modest size becomes a much larger in-memory structure; executors hit memory pressure. | On-disk size reflects encoding and compression; the hash table reflects neither (Dictionary, Run-Length, Delta and Bit Packing). | Estimate from row count and column widths rather than file size, and project away the columns the join does not need. |
| A filter on the build side is inside a user-defined function. | The planner sees a large table where the query would have produced a small one. | The optimiser cannot see through opaque expressions, so it cannot use the filter in its estimate (Predicate Pushdown). | Express the filter in native SQL where the planner can read it, or materialise the filtered side first. |
| The dimension grew past the threshold. | Runtime multiplies; the plan now shows an exchange where it used to show a broadcast. | A physical boundary was crossed with no change to any code. | Monitor build size against the threshold and decide deliberately: narrow the dimension, or accept the shuffle and size the job for it (Pipeline SLOs). |
| A forced broadcast hint on a growing table. | A job that used to degrade gracefully now fails outright. | The hint removed the planner's ability to fall back when the size assumption stopped holding. | Treat hints as temporary, with an owner and a review date, and prefer fixing the estimate over overriding it. |
Choosing deliberately
For jobs whose timing matters, the join strategy deserves to be a decision rather than a default. That does not mean hinting everything; it means knowing which strategy each important join is getting, and what would change it.
The middle option below is the one people forget. A table that is bucketed on the join key, with a matching bucket count on both sides, can be joined without any redistribution at all and without any replication either — the buckets pair up directly. It requires a layout commitment maintained by every writer, and it is the only strategy here whose benefit does not depend on either side being small (Bucketing).
The last option is also legitimate. Sometimes both sides are large, the keys are high cardinality, and a shuffle join is simply the right execution. In that case the work is to make the shuffle as cheap as possible — filter, project, pre-aggregate — rather than to avoid it (The Shuffle).
Serialization, local disk, network and merge for every row of the fact table, plus the barrier (The Shuffle).
Real but minor. This is why the strategy is dominated by the large side, and why the small side's size is irrelevant to its cost.
The only cost that grows with cluster size rather than with data. It is the term that makes broadcasts uncomfortable on very wide clusters.
Held for the duration of the stage on every executor, and sized by the in-memory representation rather than the file (Why Analytical Data Compresses).
The cost was paid at write time by maintaining the bucketing, which is why it only pays off for a join you run repeatedly (Bucketing).
Relative weights, not measurements. The ordering is what transfers between platforms; the magnitudes do not, and this domain never publishes a price.
Relative weights within one join, to establish an ordering. The teaching is the shape: the shuffle strategy is priced by the large side, the broadcast strategy by the cluster, and the bucketed strategy by every writer that came before.
What is true about the two sides and about the layout they are stored in?
when One side is small after filtering and projection, and stays small; executor count is moderate.
cost Build-side memory on every executor, and a size assumption that must be monitored (Dimension Tables).
when Both sides are physically bucketed on the join key with compatible bucket counts, and the join is frequent enough to justify the layout.
cost A layout commitment every writer must honour, and a benefit limited to joins on exactly that key (Bucketing).
when Both sides are large and the key is high cardinality. The redistribution is genuinely required.
cost A full exchange of both sides. Make it cheap by filtering and projecting first, and size the shuffle partitions for the volume (Partitions: the Unit of Parallelism).
when Both sides are large and the join key has a dominant value.
cost Salting the hot key and replicating its matches on the other side, or relying on the engine's adaptive skew join if the version supports it (Salting a Skewed Key).
when The same join happens in dozens of downstream models and the dimension changes slowly.
cost Storage, a wider fact table, and a history problem when the dimension attribute changes (Slowly Changing Dimensions).
How to build it
Most important first.
- Filter and project the build side before the join so that what is broadcast is as small as the query allows — often dramatically smaller than the table it came from (Projection Pushdown).
- Keep statistics current on the tables involved. A cost-based decision made from stale statistics is a guess wearing a plan's clothes (Cost-Based Optimization).
- Check the physical plan for the join strategy on any job whose runtime matters, and treat a change in strategy as a change worth knowing about (Reading EXPLAIN ANALYZE).
- Prefer a small, well-maintained dimension over a wide one for the join, and add the wide attributes afterwards if they are needed (Dimension Tables).
- Where a build side is near the threshold and growing, decide deliberately: either keep it small on purpose, or plan for the shuffle join and size the job for it. Drifting across the boundary is the worst of both (Pipeline SLOs).
- Reserve explicit broadcast hints for cases where you know something the planner does not, and revisit them — a hint is a hard-coded belief about size that will eventually be false.
What this actually promises
Naming the guarantee you do not have is worth more than naming the one you do — everything downstream inherits the weakest promise in the chain.
- A broadcast join returns the same rows as a shuffle join. It is a physical strategy, not a semantic one, and the result is identical when the strategy is applicable.
- It guarantees no shuffle for the large side. That is the entire benefit and it is a strong one — the barrier disappears with the redistribution (Stages and Tasks).
- It guarantees nothing about fitting in memory. If the build side is larger than believed, the job fails — on the driver assembling it, or on the executors holding it — rather than degrading gracefully.
- The planner guarantees only that it applied its threshold to its estimate. Neither the estimate nor the threshold is a statement about your executors' actual memory headroom (The Planner: Enumerating Ways to Answer).
Can I trust it?
A green pipeline is evidence that code ran. These four fields are the evidence that the data is right.
- Assert the row count of the join result against the expected fan-out. A join that emits more rows than the probe side had is a duplicate-key problem in the build side, and it is the most common way a dimension join corrupts a fact table (Duplicate Rows).
- Add a uniqueness test on the dimension's business key. A broadcast join multiplies exactly as faithfully as a shuffle join does; the strategy neither causes nor prevents fan-out (Data Tests).
- Neither check notices a plan change. The result is identical either way — which is precisely why the failure is a duration and a bill rather than a wrong number.
- Removing a shuffle removes a barrier, which usually shortens the job more than any other single change available to it (Cost vs Freshness).
- It also makes the duration more predictable, because the stage no longer depends on the distribution of the join key — a skewed key stops mattering when nothing is redistributed (Data Skew).
- The reverse is what makes broadcast fragility a freshness problem: a plan that flips to a shuffle join turns a predictable job into an unpredictable one overnight.
- A dimension that grows crosses the threshold at some point, silently. That is an evolution of physics rather than of schema, and no schema check will see it (Semantic Changes).
- Adding columns to the dimension makes the broadcast larger without adding rows, so a table can cross the threshold from a schema change that looks harmless (Schema Evolution).
- Engine upgrades change default thresholds and estimation behaviour, which can flip a strategy on a job whose code has not been touched in a year.
- Nothing to repair — the results are the same either way. Recovery from a failed broadcast means re-running with a smaller build side, a raised threshold, or the shuffle strategy accepted.
- When a broadcast is failing because the build side genuinely grew, the durable fix is upstream: a narrower dimension, a pre-filtered view, or a different join key (Dimension Tables).
- If the job died on the driver, check whether anything was committed before deciding what to re-run — a failure during broadcast is early, so usually nothing was (Atomic Publish).
What can go wrong
- Driver out of memory while assembling the build side, taking the application down after every task had been fine.
- Executors under memory pressure holding a hash table that is larger than its on-disk footprint suggested (OOM Kills and CPU Throttling).
- A silent switch to a shuffle join as the build side grows, multiplying runtime with no code change.
- A hint that forces a broadcast of something that no longer fits, converting an automatic degradation into a hard failure.
- The mitigation failing: raising the broadcast threshold so the planner keeps choosing broadcast, which moves the failure from "slow job" to "job that dies at 03:40" (The Bottleneck Moves After Every Fix).
- "Broadcast joins are faster." They are faster when one side is genuinely small. Broadcasting something large is slower than shuffling it, and often fatal.
- "The file is only a few hundred megabytes, so it will fit." On-disk columnar size and in-memory hash-table size are different quantities, and the second is the one that has to fit (Why Analytical Data Compresses).
- "The planner will always pick the right strategy." It picks from an estimate. Stale statistics, an opaque filter, or a size just under the threshold all produce confident wrong choices (Cost-Based Optimization).
- "Broadcasting removes skew." It removes the shuffle, and with it the skew *of that join*. Any other wide operation in the job still has its own distribution (Data Skew).
Operating it
- The join strategy in the physical plan, recorded per run. A diff of plans between runs explains most "it was fine yesterday" reports (Reading EXPLAIN ANALYZE).
- Broadcast build size and time, which tells you how close to the threshold you are before you cross it (Pipeline Metrics).
- Shuffle bytes for the stage. When a broadcast is chosen this is near zero for the large side; when it flips, the number appears and is unmistakable.
- Driver memory during the build phase, which is where the assembly happens and where the failure lands (Memory Pressure, Swap and the OOM Killer).
- At 10x on the large side, a broadcast join scales beautifully — nothing about the build side changed and the probe side is embarrassingly parallel.
- At 10x on the small side, the strategy stops applying and the job changes character entirely. Which side grows is what matters, not the total (Fact Tables).
- At high executor counts the aggregate memory held by the broadcast becomes significant, and a very wide cluster can make a comfortable broadcast uncomfortable.
- Cost is the build side times the executor count, transferred and materialised. It grows with the cluster, which is the opposite direction from every other cost in this module.
- Against that, the entire shuffle of the large side is saved: serialization, disk, network, merge and barrier (The Shuffle).
- The trade is nearly always favourable when the size condition genuinely holds, and it fails catastrophically rather than gradually when it does not.
- Broadcasting trades memory on every executor for the elimination of a shuffle. That is usually a good trade and it is a trade, not a free win.
- It introduces a size assumption that will eventually be violated, so it needs monitoring rather than a comment.
- Forcing it with a hint buys determinism of plan and gives up the planner's ability to adapt when the data changes — which is the same trade every hint makes (Query Optimizers).
Where this applies
Almost nothing here is universal. These labels say what each claim is specific to, and where a different engine, format, warehouse or scale would differ.
- GENERALReplicating a small relation to avoid redistributing a large one is a strategy in every distributed query system, and in single-node engines it is simply the hash join with the small side as the build. What differs is who decides and from which statistics.
- ENGINE-SPECIFICSpark decides from a size threshold against estimated statistics and lets you hint; Trino makes a similar broadcast-versus-partitioned choice from its own cost model; a serverless warehouse chooses internally and gives you no threshold at all. The failure mode also differs: Spark assembles on the driver, so a bad estimate kills the application rather than one worker.
- SCALE-SPECIFICThe strategy applies below the point where the build side stops fitting in executor memory, and that point moves down as executor count rises because the same data is held on every machine. A build side that is comfortable on a small cluster can be a problem on a very wide one.
Where the depth lives
This domain teaches how data moves and how you know it arrived intact. It hands the rest off by name.
- — Distributed Systems owns the general trade between replicating state to every node and partitioning it across them, and why the first buys locality at a cost that scales with the cluster.