Back to Performance Tuning
Q4Question 4 of 8

How Broadcast Joins interact with AQE: runtime switch from SSMJ to BHJ

Key Concepts
  • AQE re-optimizes at stage boundaries and can convert a planned sort-merge join into a broadcast join when it measures one side is small enough.
  • The static planner may conservatively choose sort-merge because it only knows the pre-filter size estimate.
  • AQE reads actual materialized sizes from shuffle map statistics, which are accurate, not estimated.
  • The switch is governed by the adaptive broadcast threshold, which may differ from the static autoBroadcastJoinThreshold.
  • The authoritative view of what ran is the Spark UI final plan, not the static explain() output.
  • AQE pairs the switch with a local shuffle reader, reading the already-shuffled large side locally without a network fetch.

Definition

AQE watches real runtime sizes and upgrades a planned sort-merge join to a broadcast join when a side turns out small enough after filtering.

AQE's dynamic join strategy switch works as follows.

  • Static plan – Catalyst conservatively picks a sort-merge join because its compile-time estimate of the join side is too large or unavailable.
  • Runtime measurement – After an earlier stage completes (e.g., filtering a dimension), AQE reads the actual materialized size from shuffle statistics.
  • Dynamic switch – If the measured size is at or below the adaptive broadcast threshold, AQE rewrites the plan to broadcast that side.
  • Local shuffle reader – AQE reads the already-shuffled large side locally without a network fetch, further reducing cost.
  • Authoritative plan – The Spark UI final plan, not the static explain(), shows what actually ran.

This eliminates the need for manual broadcast hints in post-filter cases.

The National Bank of India plans a big matching job assuming a report is huge, so it schedules the slow sort-and-merge approach. But partway through, after filtering the report to only this month's entries, the clerk sees it is actually just two pages. AQE is the supervisor who notices this mid-job and says 'stop: that is tiny now, just photocopy it for everyone and match locally,' switching from the slow plan to the fast broadcast on the spot.

Understanding the Concepts

Why does Catalyst sometimes pick the wrong join strategy?

The static Catalyst planner decides join strategies before the job runs, using compile-time size estimates. Those estimates are often wrong for a side whose size depends on a filter or an aggregation the planner cannot evaluate ahead of time.

PYTHON
# Static plan: Catalyst estimates branches as 500 MB → picks sort-merge
branches = spark.read.parquet("/data/bank/branches")  # 500 MB on disk
txns = spark.read.parquet("/data/bank/transactions")  # 500 GB

# But after filtering, branches is only 2 MB
active_branches = branches.filter("status = 'ACTIVE'")

result = txns.join(active_branches, "branch_code")
result.explain()  # Static plan shows SortMergeJoin
# Catalyst only knows the pre-filter size (500 MB)
# It does not know the post-filter size (2 MB)

How does AQE detect the real size?

AQE re-optimizes at stage boundaries. When the filter stage completes, AQE reads the actual partition sizes from the shuffle map statistics.

DIAGRAM
Stage 1: Scan + Filter branches (completes)
  AQE reads shuffle statistics:
    Actual output size: 2 MB (not the estimated 500 MB)

Stage 2: Join
  AQE checks: 2 MB < adaptive broadcast threshold (10 MB)?
    YES → rewrite plan to BroadcastHashJoin
    Broadcast the 2 MB filtered branches
    Skip shuffle of the 500 GB transactions side
PYTHON
# With AQE enabled, the static plan differs from the executed plan
result.explain()  # Static: SortMergeJoin

# After execution, check Spark UI:
# SQL tab → query → executed plan shows BroadcastHashJoin
# AQE switched it at runtime because the filter reduced branches to 2 MB

How do you see the difference between static and executed plans?

The static explain() shows what Catalyst planned before execution. The Spark UI shows what AQE actually executed. They can differ.

PYTHON
# Static plan (pre-AQE)
result.explain()
TEXT
== Physical Plan ==
SortMergeJoin [branch_code], [branch_code]
:- Sort [branch_code ASC]
:  +- Exchange hashpartitioning(branch_code, 200)
:     +- FileScan parquet [transactions]
+- Sort [branch_code ASC]
   +- Exchange hashpartitioning(branch_code, 200)
      +- Filter (status = ACTIVE)
         +- FileScan parquet [branches]
TEXT
== Spark UI Executed Plan (after AQE) ==
BroadcastHashJoin [branch_code], [branch_code]
:- LocalShuffleReader
:  +- Exchange hashpartitioning(branch_code, 200)
:     +- FileScan parquet [transactions]
+- BroadcastExchange
   +- Filter (status = ACTIVE)
      +- FileScan parquet [branches]  (2 MB after filter)

What is the local shuffle reader optimization?

When AQE switches to a broadcast join, the large side may already have been shuffled. Instead of re-reading or re-shuffling, AQE uses a local shuffle reader that reads the already-shuffled data locally.

DIAGRAM
Without local shuffle reader:
  Large side: shuffled → read from remote executors → join

With local shuffle reader:
  Large side: already shuffled → read locally → join
  (saves network I/O)

What are the configuration parameters?

The switch is governed by AQE-specific settings, not just the static autoBroadcastJoinThreshold.

PYTHON
# AQE settings that govern dynamic join switching
spark.conf.set("spark.sql.adaptive.enabled", "true")

# Static threshold (used before execution)
print(spark.conf.get("spark.sql.autoBroadcastJoinThreshold"))  # 10485760 (10 MB)

# AQE may use a different effective threshold based on runtime stats
# The switch happens only at stage boundaries where real stats exist

| Parameter | Role |
|---|---|
| adaptive.enabled | Master switch for AQE |
| autoBroadcastJoinThreshold | Static threshold for pre-run planning |
| AQE runtime check | Uses measured sizes at stage boundaries |

When does AQE NOT switch?

The switch only happens at a stage boundary where AQE has real statistics. A side whose size is only known mid-stage will not trigger it.

DIAGRAM
AQE switches:
  ✓ Filter in a separate stage → AQE measures output → switch
  ✓ Aggregation in a separate stage → AQE measures output → switch

AQE does NOT switch:
  ✗ Filter fused into the same stage as the join → no boundary → no measurement
  ✗ No shuffle between filter and join → no statistics

How does this reduce the need for broadcast hints?

Before AQE, you had to manually add broadcast hints for post-filter cases. AQE makes this automatic.

PYTHON
# Before AQE: manual hint required
from pyspark.sql.functions import broadcast
result = txns.join(broadcast(active_branches), "branch_code")
# Risk: if the filter does not reduce enough, OOM

# With AQE: no hint needed
result = txns.join(active_branches, "branch_code")
# AQE measures the real size and switches only if safe
# If the filter does not reduce enough, AQE keeps sort-merge

What should a senior engineer know?

  • Always check the Spark UI final plan rather than static explain() to see what AQE actually did.
  • AQE's dynamic switch is safer than manual broadcast hints because it is based on measured sizes, not estimates.
  • The switch adds no overhead: it simply rewrites the plan before the next stage starts.
  • AQE must be enabled (default in Spark 3.2+) for the switch to work.
  • For the National Bank of India, this saved us from maintaining broadcast hints on every post-filter join.
PYTHON
# Best practice: rely on AQE instead of broadcast hints
# Let AQE measure and switch automatically
spark.conf.set("spark.sql.adaptive.enabled", "true")

# Only use explicit broadcast() for tables you KNOW are tiny
result = txns.join(broadcast(currency_rates), "currency_code")  # 100 KB: always safe

Visual Overview

DIAGRAM
  AQE Dynamic Join Switch:

  Static Plan (Catalyst):
  ┌────────────────────────────────┐
  │ SortMergeJoin                   │
  │ ├── Shuffle transactions (500GB)│
  │ └── Shuffle branches (est 500MB)│
  └────────────────────────────────┘

  Runtime (AQE):
  Stage 1: Filter branches → completes
  ┌────────────────────────────────┐
  │ AQE reads real size: 2 MB      │
  │ 2 MB < threshold (10 MB)?  YES │
  │ → REWRITE to BroadcastHashJoin │
  └────────────────────────────────┘

  Executed Plan (Spark UI):
  ┌────────────────────────────────┐
  │ BroadcastHashJoin               │
  │ ├── LocalShuffleReader (txns)   │
  │ └── BroadcastExchange (2 MB)    │
  └────────────────────────────────┘

  Static explain() ≠ Spark UI final plan

Key Takeaway

AQE interacts with broadcast joins by converting a planned sort-merge join into a broadcast hash join at runtime once it sees a side is actually small. Catalyst conservatively picks sort-merge because its compile-time estimates are often wrong for tables whose size depends on filters or aggregations. AQE re-optimizes at stage boundaries, reading the real materialized sizes from shuffle statistics, and if a side is below the threshold, it rewrites the remaining plan to broadcast that side.

The key difference from static auto-broadcast is the timing: AQE decides from measured sizes after a stage runs, which are accurate, rather than from pre-run estimates. This is why a query can show SortMergeJoin in the static explain() but BroadcastHashJoin in the Spark UI final plan. AQE also pairs the switch with a local shuffle reader so the already-shuffled large side is read locally.

At the National Bank of India, a join to a product dimension that was 500 MB on disk but only a few megabytes after filtering was planned as sort-merge, but AQE switched it to broadcast at runtime. This saved us from maintaining broadcast hints on every post-filter join.

The practical rule is: rely on AQE for post-filter cases, check the Spark UI for the final executed plan, and only use explicit broadcast hints for tables you know are unconditionally tiny.

Remember one sentence: AQE dynamically switches from sort-merge to broadcast join at runtime when it measures that a post-filter side is small enough, using real statistics instead of estimates.

---

Share
Use ← → arrow keys to navigate