Spark Broadcast Hash Join

A broadcast hash join is the join with no shuffle, and it is usually the right plan. It is also the only join plan whose cost is paid by a machine that is not doing the work: the small side goes to the driver first, is rebuilt there as a hash table, and only then reaches the cluster.

That detour is the whole lesson. Everything people find surprising about spark.sql.autoBroadcastJoinThreshold — that raising it causes driver failures, that lowering it does not stop broadcasts, that a 40 MB Parquet file needs gigabytes of heap — follows from where the bytes go and what shape they are in when they get there.

Below is a single join: a 900 GB fact table against one small dimension. Nothing about the query changes as you move these controls. Only the size and shape of the small side, the size of the cluster, and the number Spark is allowed to test against.

Start at the defaults: an 8 MB dimension, under the 10 MB threshold, so Spark broadcasts. It wins by a wide margin, and the reason is in the network readout rather than the clock — sort-merge moves about a terabyte, broadcast moves about half a gigabyte.

Now push small side to 600 MB and raise the threshold to match, so the planner still says yes. The driver heap meter crosses the line and the job dies with an OOM inside the broadcast exchange. Nothing about the cluster changed and the query is still legal SQL; the small side simply stopped fitting in a machine you were not thinking about. Raise spark.driver.memory to 8 GB and it runs. Then take the driver to 16 GB and drag in-memory expansion to 16x: the driver now has room and the job still dies, because the built relation crossed a ceiling that no configuration raises.

Two different numbers, and only one of them is measured

The planner's test is one predicate. In Spark 3.5.1, JoinSelectionHelper.canBroadcastBySize is:

def canBroadcastBySize(plan: LogicalPlan, conf: SQLConf): Boolean = {
  val autoBroadcastJoinThreshold = if (plan.stats.isRuntime) {
    conf.getConf(SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD)
      .getOrElse(conf.autoBroadcastJoinThreshold)
  } else {
    conf.autoBroadcastJoinThreshold
  }
  plan.stats.sizeInBytes >= 0 && plan.stats.sizeInBytes <= autoBroadcastJoinThreshold
}

Hold the isRuntime branch for the AQE section below. At planning time it is false, and the whole decision is the last line.

plan.stats.sizeInBytes is an estimate produced by walking the logical plan. For a plain Parquet scan it is the sum of the file lengths, multiplied by spark.sql.sources.fileCompressionFactor — an internal setting whose default is 1.0, so by default the multiplication does nothing. Spark compares a compressed on-disk figure against the threshold and calls it the table's size.

What the driver then pays is a different quantity entirely. BroadcastExchangeExec calls child.executeCollectIterator(), which pulls every row of the build side back to the driver through one network interface and one heap. It then calls mode.transform(...) to build a HashedRelation — decompressed, deserialised, with key slots and pointers — and measures that with relation.estimatedSize. Expansion of 3x to 8x is ordinary and string-heavy dimensions go further, so a 40 MB Parquet file can arrive as a few hundred megabytes of driver heap. Do not take that ratio on faith: BroadcastExchange's dataSize metric reports it for your own table, and the threshold never saw that number.

Three hard stops sit at the end of that path. Two of them are constants in BroadcastExchangeExec that no configuration raises; the first is a config, but not the one you were tuning:

• the collect goes through the normal action result path, so the scheduler's spark.driver.maxResultSize check (default 1g) can abort the job before the hash table is even built — the message comes from TaskSetManager, not from the join;
• MAX_BROADCAST_TABLE_BYTES is 8L << 30 — a hard 8 GB ceiling, checked against the dataSize of the relation after it has been built;
• a row cap, checked before the build. It is 512000000 for a single LongType key (which becomes a LongHashedRelation) and for non-hash broadcast modes; otherwise the build side becomes a BytesToBytesMap-backed UnsafeHashedRelation and the cap is BytesToBytesMap.MAX_CAPACITY / 1.5, which with MAX_CAPACITY = 1 << 29 is about 358 million rows.

And wrapping all of it, spark.sql.broadcastTimeout, default 300 seconds. That clock covers the collect and the build and the distribution. Could not execute broadcast in 300 secs is therefore not a network diagnosis. Read it against collectTime and buildTime in the SQL tab before you blame anything else; that is usually where the clock went.

Why the estimate is wrong in the direction that hurts

If the build side is a plain table scan, sizeInBytes is at least in the right neighbourhood. The moment anything happens above the scan, it stops being. With spark.sql.cbo.enabled at its default of false, estimation is done by SizeInBytesOnlyStatsPlanVisitor, and that visitor is much blunter than people assume.

Move only the rows the filter keeps slider. The estimate does not change. That is not a rounding error, it is the implementation: visitFilter(p) = visitUnaryNode(p), and visitUnaryNode assumes the output row count equals the input row count. A predicate that keeps one row in ten thousand is estimated to keep all of them. This is the single most common reason a genuinely tiny DataFrame refuses to broadcast, and no amount of raising the threshold fixes it, because the number being tested is the size of the table you filtered, not the size of the result.

Column pruning, by contrast, is estimated — visitProject scales sizeInBytes by the ratio of output row width to input row width, which is why it is the one rewrite that reliably changes a join strategy. Aggregation is not: a groupBy with a non-empty grouping list also falls through to visitUnaryNode, so collapsing millions of rows to 2,400 groups leaves the estimate exactly where it was. Only a global aggregate with no grouping keys is estimated as a single row.

Then tick the join. The estimate does not grow, it detonates. The default visitor for a node it has no rule for is p.children.map(_.stats.sizeInBytes).filter(_ > 0L).product — the product of the children's sizes, in bytes. Two modest inputs produce a petabyte-scale estimate, which is why the output of a join is essentially never auto-broadcast without real table statistics. Spark takes the sum instead only in a narrow case, added in 3.4: an inner or outer equi-join where one side's known distinctKeys are a subset of that side's join keys, which in practice means you put a distinct() or a groupBy on the key first. (Semi and anti joins skip all of this and simply inherit the left side's statistics, which is correct: they cannot produce more rows than the left side has.)

Both failure directions are live, and this simulation only shows one of them. Over-estimates are what you see here, and they cost you a shuffle you did not need. Under-estimates cost more: tick Statistics are stale in the first simulation, where a table that grew since the last ANALYZE gets waved through the threshold on a recorded number that no longer describes it. A fileCompressionFactor that does not match your codec, or a broadcast() hint that skips the test altogether, land in the same place: the driver is handed something it cannot hold.

What actually scales with executor count

The common mental model is that the driver sends one copy to each executor, so 400 executors means 400 sends from one machine. That is not what happens, and getting it wrong leads people to blame the wrong resource. Spark broadcasts through TorrentBroadcast: the relation is chopped into blocks of spark.broadcast.blockSize (default 4m), the driver stores them in its block manager, and executors fetch blocks — from the driver, and from each other, because every executor stores each fetched piece with tellMaster = true, which registers it and makes that executor a source for the next fetcher. So the driver does not pay one full send per executor. The simulation models the spread as logarithmic in the executor count, which is the shape you get from peer-to-peer dissemination; the real curve depends on block count and link speeds, but its defining property — sublinear, not linear — is in the mechanism. Drag executors from 4 to 400 in the first simulation and watch the seed phase on the driver lane grow slowly.

What is linear in executor count is the resident copy. Each executor JVM deserialises the relation once and keeps it (MEMORY_AND_DISK) for the life of the broadcast — shared across that executor's cores, which is the one piece of good news, but paid once per executor. At 400 executors and a 2 GB relation that is 800 GB of cluster memory doing nothing but holding duplicates, and it comes out of the same unified pool the probe side wanted. The simulation is charitable here: it counts only the deserialised copy, while a real executor also keeps the serialised pieces it fetched. Watch the cluster RAM held readout and the spill on the executor lanes.

Put those together and the crossover appears. The driver's collect-and-build phase is serial and does not get faster when you add executors, and the seed phase gets slowly worse. The sort-merge alternative is almost entirely parallel and gets better in proportion. So on a small cluster broadcast wins enormously, and on a large enough cluster the same join flips: set the small side to 600 MB with an 8 GB driver, then drag executors from 40 to 400 and watch the hero number cross the one beside it. The broadcast plan still moves fewer bytes and still finishes second. Fewer bytes is not the same as less time when one of the machines moving them is the only one of its kind.

This is also why the shuffle that broadcast avoids is not automatically the expensive option. A sort-merge join spreads its cost over every core in the cluster; a broadcast concentrates part of its cost on one JVM. Which is cheaper depends on how many cores you are buying.

Where the threshold has no say

Setting spark.sql.autoBroadcastJoinThreshold to -1 is widely recommended as the way to stop broadcasts. It stops size-based broadcasts. Two paths route around it.

The first is the explicit hint. The documentation is unambiguous: a BROADCAST hint means that side "will be prioritized by Spark even if the size of table suggested by the statistics is above the configuration spark.sql.autoBroadcastJoinThreshold". In getBroadcastBuildSide the hinted branch calls hintToBroadcastLeft/Right and never reaches canBroadcastBySize at all. Tick the hint box in the first simulation with the threshold at 0 and the plan still comes out as a broadcast, all the way to a driver OOM. A hint is not a suggestion with a safety net.

The one carve-out is in the next paragraph of the same doc: "there is no guarantee that Spark will choose the join strategy specified in the hint since a specific strategy may not support all join types". A BROADCAST hint on the left side of a left outer join asks for a build side that plan cannot use, and Spark quietly ignores it. When a hint fails to broadcast it is because of the join type, or because the hint never resolved to a relation at all. Size does not enter into it.

The second is BroadcastNestedLoopJoinExec, and it is the one that catches people. Spark's join strategy list for a join with no equi-join key — a range predicate, a LIKE, an inequality — ends with a comment in SparkStrategies.scala that is worth quoting exactly:

//   3. Pick broadcast nested loop join as the final solution. It may OOM but we don't have
//      other choice. It broadcasts the smaller side for inner and full joins, broadcasts the
//      left side for right join, and broadcasts right side for left join.

There is no size test on that branch. If your join condition has no equality to hash on, and it is not an inner join that can become a Cartesian product, Spark broadcasts one side regardless of how large it is, because nothing else in the strategy list can produce a correct answer. Setting the threshold to -1 does not disable it. This is the real explanation behind "broadcast join error even though autoBroadcastJoinThreshold=-1": the fix is to give the join an equality to work with, not to turn a dial.

The same decision, made later, with real numbers

Adaptive query execution converts a sort-merge join to a broadcast hash join at runtime. It is the identical predicate — canBroadcastBySize — with two differences that matter.

First, plan.stats.isRuntime is true, so the size being tested is the measured byte count a completed shuffle stage wrote, rather than an estimate walked up from file lengths. The estimator failures in the second simulation stop mattering for that side: there is nothing left to estimate. Two caveats keep this from being magic. AQE only re-plans at a query-stage boundary, so a side that is never shuffled is never measured; and the measured number is shuffle bytes on disk, which is not the same quantity as the hash table the driver will build. Tick AQE re-plans on runtime statistics and watch the ratio go to 1.0. Second, the threshold consulted is spark.sql.adaptive.autoBroadcastJoinThreshold (added in 3.2.0), which defaults to the same value as the static one but can be set independently — which is exactly what you want, because the runtime number is trustworthy and the planning-time number is not.

The catch is in the Spark docs' own wording: this "is not as efficient as planning a broadcast hash join in the first place". By the time AQE decides, the shuffle map stages on both sides have already run: serialised, compressed, written to local disk, spilling if the map side needed to. What has not happened yet is the read side, which is why AQE can still save you something. It saves the sort and, with spark.sql.adaptive.localShuffleReader.enabled, most of the fetch. It cannot refund the write. A join that AQE converts is a join whose partitioning work you paid for and threw away.

So the honest ordering is: get the statistics right so the planner broadcasts at planning time; let AQE catch what the estimator could not see; and treat a hint as the thing you reach for when you have measured the build side yourself and know what the driver is about to be handed.

Checking it on a real query

Four places, in the order you should look:

• df.explain("cost") prints Statistics(sizeInBytes=..., rowCount=...) on every node of the optimized plan. This is the number canBroadcastBySize tests. If it is wildly larger than the data, you have found your answer without running anything.
• DESCRIBE EXTENDED <table> shows the catalog statistics and when ANALYZE TABLE ... COMPUTE STATISTICS last ran. Absent or stale entries are why the estimator fell back to file lengths.
• The SQL tab's BroadcastExchange node carries the four numbers that settle this — dataSize, collectTime, buildTime and broadcastTime — alongside numOutputRows. Compare dataSize to the file size you expected — that ratio is your real expansion factor, and it is the number to plug into a threshold, not the file size. If collectTime plus buildTime is most of your stage, the cluster was idle while the driver worked.
• In an AQE plan, the SQL tab's top node is AdaptiveSparkPlan (isFinalPlan=true) and its details pane shows the initial plan beside the final one — a SortMergeJoin in the first and a BroadcastHashJoin in the second is a runtime conversion. The driver log records the same thing as Plan changed: with a side-by-side diff, though it is written at the level named by spark.sql.adaptive.logLevel — an internal setting whose default is debug, so either turn DEBUG logging on or set that setting to info. Do not go looking for isRuntime in a plan string: it is a field on Statistics, but simpleString prints only sizeInBytes and rowCount, so it never appears.

And the error strings, which are diagnostic if you read them as pointers to a phase: Could not execute broadcast in 300 secs is the collect or the build; Not enough memory to build and broadcast the table to all worker nodes is the driver heap during mode.transform; Cannot broadcast the table that is larger than 8.0 GiB is the hard ceiling (Spark formats the constant with bytesToString, so it reads as GiB, not 8GB); Total size of serialized results ... is bigger than spark.driver.maxResultSize is the collect being stopped before it finishes.

Next, the plan this one replaces and the machine that pays for it: sort-merge join and driver memory.