One JOIN, three very different executions

The query text for a JOIN says nothing about how it will actually run. Databricks picks a physical join strategy based on size estimates and, with Adaptive Query Execution (AQE) enabled by default, can re-pick a better one mid-query once real statistics are available after a shuffle or broadcast exchange. The same logical JOIN can become a broadcast hash join (cheap), an evenly-distributed shuffle join (more expensive but predictable), or a shuffle join with a skewed partition (the worst case, where one task does most of the real work while the cluster waits on it) -- and the difference between the three isn't about how much data exists in total, it's about which side is small enough to broadcast and how evenly the join key is distributed.

See it side by side

Illustrative example built from typical AQE behavior -- not a live connection to Databricks

Same JOIN, three strategies

Click a scenario. Watch which segment grows -- and how much the total changes for the exact same join.

Advertisement

Cause 1: the broadcast threshold was missed

AQE can dynamically convert a sort-merge join into a broadcast hash join once it has accurate statistics, using spark.databricks.adaptive.autoBroadcastJoinThreshold (default 30MB) as the runtime trigger. But AQE may not make the switch until both sides have already been shuffled once -- so if you know one side is small, a broadcast hint is often cheaper than waiting for AQE to notice. A table that's only slightly over the threshold, or whose size estimate is stale, can silently fall back to a full shuffle join instead of broadcasting.

Fix: an explicit hint -- SELECT /*+ BROADCAST(small_table) */ ... -- forces the broadcast regardless of the threshold, when you know the side is small. Not every join type supports it (the outer side of a LEFT OUTER JOIN can't be broadcast), so the hint can still be ignored in some shapes.

Cause 2: a skewed key with AQE handling off or insufficient

AQE only treats a partition as skewed when its size exceeds both the median partition size times skewedPartitionFactor (default 5) and the flat skewedPartitionThresholdInBytes (default 256MB) -- below both, an unevenly-sized partition is left alone, and one task ends up carrying far more rows than the others while the rest of the cluster finishes and waits. Broadcast joins are never skew-optimized at all, because the large side was never shuffled into partitions in the first place; skew handling only matters once you're already in a shuffle join.

Fix: keep spark.sql.adaptive.skewJoin.enabled on (the default) rather than relying on the legacy skew join hint, and check whether a genuinely hot key (a single customer ID or status value dominating the table) needs a salting technique if AQE's thresholds still don't trigger a split for it.

Advertisement

Cause 3: dynamic file pruning never triggered

When a JOIN filters a large fact table by a small lookup table's values, dynamic file pruning (DFP) can push that filter down and skip whole files on the large side before they're even read -- but only when the large table is at least 10GB (deltaTableSizeThreshold) and has at least 10 files (deltaTableFilesThreshold) by default. A table just under either threshold gets scanned in full on the probe side of the join, even though the join itself looks highly selective from the query text.

Key takeaways

  • Broadcast avoids shuffling the large side entirely -- it's the cheapest strategy whenever one side is genuinely small enough.
  • AQE's broadcast conversion and skew handling both depend on documented numeric thresholds, not vague heuristics -- know them before assuming AQE "should have" fixed it.
  • Skew only gets fixed automatically above skewedPartitionFactor × median size AND skewedPartitionThresholdInBytes -- a moderately uneven key can still slip through both checks.
  • Dynamic file pruning has its own separate size and file-count thresholds on the large table -- missing them means a "selective" join still scans everything.

The controls: hints and thresholds

Join hints

/*+ BROADCAST(t) */, /*+ MERGE(t) */, /*+ SHUFFLE_HASH(t) */, /*+ SHUFFLE_REPLICATE_NL(t) */ -- explicit strategy control, with BROADCASTJOIN/MAPJOIN and SHUFFLE_MERGE/MERGEJOIN as aliases for the first two.

Broadcast threshold

spark.databricks.adaptive.autoBroadcastJoinThreshold, default 30MB -- the runtime trigger AQE uses to switch an already-planned sort-merge join into a broadcast join.

Skew thresholds

skewedPartitionFactor (default 5) and skewedPartitionThresholdInBytes (default 256MB) -- both must be exceeded before AQE treats a partition as skewed.

Diagnostic checklist

When one JOIN feels disproportionately slow, check in order: is the smaller side under the broadcast threshold, or hinted explicitly? Is any single shuffle partition's size well past both skew thresholds? Is the large side of the join at least 10GB and 10 files, so dynamic file pruning can even apply? The downloadable script below pulls table size and file counts via DESCRIBE DETAIL to check the last one directly.

Download join-strategy-diagnostics.sql
Real, ready-to-run queries for your own Databricks workspace -- table size/file-count checks against the documented broadcast and dynamic-file-pruning thresholds.