Definition
- Dynamic join switching – after an earlier stage executes, AQE compares actual output sizes against the broadcast threshold and can convert a planned sort-merge join into a broadcast hash join if one side is small enough.
- Why it matters – Catalyst size estimates are often wrong after filters and aggregations shrink data, so the static plan picks a shuffle join when a broadcast would be better.
- Local shuffle reader – when AQE converts a join, the large side's already-written shuffle data is read locally on each executor instead of being fetched over the network, removing the cost of the now-unnecessary repartition.
- Safety – AQE only broadcasts when the measured size is genuinely under the threshold, preventing out-of-memory errors.
Two teams at the National Bank of India plan to combine their customer lists by doing a full, slow re-sort. But before they start, AQE checks the actual lists and notices one team's list is tiny. Instead of the heavy re-sort, AQE photocopies that one page and hands it to everyone, so each clerk matches locally. The local shuffle reader is the follow-on cleverness: because the big list no longer needs re-sorting, each clerk reads the portion already on their desk instead of fetching pages from other desks.
Understanding the Concepts
What is AQE dynamic join switching?
Normally, Catalyst picks a join strategy before the job runs based on size estimates. If both sides look large, it picks sort-merge join (shuffle both sides). If one side looks small, it picks broadcast hash join (send the small side everywhere). The problem is that estimates are often wrong after filters and aggregations.
AQE defers the decision. After an earlier stage executes, Spark knows the real size of its output and can switch the strategy at runtime.
from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder \
.appName("NBI-AQE-JoinSwitch") \
.config("spark.sql.adaptive.enabled", "true") \
.config("spark.sql.autoBroadcastJoinThreshold", "10m") \
.getOrCreate()
# Large transactions table: 500 million rows
txns = spark.read.parquet("s3://nbi-data-lake/transactions/")
# Customers table: 10 million rows, but after filtering...
customers = spark.read.parquet("s3://nbi-data-lake/customers/")
vip_customers = customers.filter(F.col("tier") == "VIP") # Only 5000 rows!
# Catalyst sees customers as 10M rows (large) and plans sort-merge.
# AQE sees vip_customers is only 5000 rows (tiny) after the filter.
# AQE switches to broadcast hash join at runtime.
result = txns.join(vip_customers, "customerId")
result.explain("formatted")
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
+- BroadcastHashJoin [customerId], [customerId], Inner
:- FileScan parquet [customerId,amount,...]
+- BroadcastExchange HashedRelationBroadcastMode
+- Filter (tier = VIP)
+- FileScan parquet [customerId,tier,...]
Why do size estimates go wrong?
Catalyst estimates sizes from table statistics (row counts, column sizes) and applies heuristics for filters. But several common cases break these estimates.
- Statistics are stale after large data changes.
- A table read from files without computed statistics has unreliable size.
- A filter reduces data dramatically, but the estimator cannot propagate it accurately.
# Example: Catalyst thinks the branch dimension is 500MB (full table)
# After filtering to one city, it is only 2KB
branches = spark.read.parquet("s3://nbi-data-lake/branches/")
delhi_branches = branches.filter(F.col("city") == "Delhi")
# Without AQE: sort-merge join (unnecessary shuffle of 500M txns)
# With AQE: broadcast join (2KB broadcast, no shuffle of txns)
result = txns.join(delhi_branches, "branchId")
result.write.parquet("s3://nbi-data-lake/delhi-report/")
What is the local shuffle reader?
When AQE converts a sort-merge join to a broadcast join, the large side may have already been shuffled (its data written to local disk partitioned by the join key). A broadcast join does not need that repartitioning, so instead of fetching shuffle blocks across the network, AQE uses a local shuffle reader: each executor reads the shuffle output already on its own disk.
# What happens step by step:
# 1. Stage 1 runs: shuffle-writes both sides by join key
# 2. AQE checks: right side is only 2KB! Switch to broadcast.
# 3. Right side is broadcast to all executors.
# 4. Left side: already shuffled, but no network fetch needed.
# Each executor reads its LOCAL shuffle files.
# 5. BroadcastHashJoin runs locally on each executor.
# In the Spark UI, you see:
# - "CustomShuffleReader" or "AQEShuffleRead" with local reads
# - No shuffle read bytes on the large side (local only)
How does AQE decide when to switch?
AQE compares the actual post-stage output size against the autoBroadcastJoinThreshold (default 10MB). If the actual size of a side is under the threshold, it switches to broadcast. If both sides are under the threshold, it broadcasts the smaller one.
# AQE decision logic (simplified):
# after_filter_size = actual_bytes_after_stage_completes
# threshold = spark.sql.autoBroadcastJoinThreshold (10MB default)
#
# if after_filter_size <= threshold:
# switch to BroadcastHashJoin
# else:
# keep SortMergeJoin
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "50m")
# Now AQE will broadcast sides up to 50MB, catching more cases
What are the benefits of dynamic join switching?
The switch eliminates the most expensive operation in a join: the shuffle of the large side. Instead of writing, transferring, and reading billions of rows, the large side stays where it is.
| Aspect | Sort-Merge Join | AQE-Switched Broadcast Join |
|---|---|---|
| Large side | Shuffled across network | Stays local (no shuffle) |
| Small side | Shuffled across network | Broadcast to all executors |
| Sort step | Required | Not needed |
| Network I/O | Very high (both sides) | Low (only small side broadcast) |
| Speed | Slower | Much faster |
# Before AQE: sort-merge join on 500M txns + filtered customers
# After AQE switch: broadcast join, no shuffle of 500M rows
# Typical speedup: 3x to 10x for star-schema banking queries
When does AQE NOT switch?
AQE will not switch if the actual size of both sides exceeds the threshold. It also respects executor memory: broadcasting a side that is under the threshold but still large relative to executor memory can cause out-of-memory errors.
# Both sides are genuinely large: AQE keeps sort-merge
large_accounts = spark.read.parquet("s3://nbi-data-lake/accounts/") # 200M rows
large_txns = spark.read.parquet("s3://nbi-data-lake/transactions/") # 500M rows
# Neither side is under 10MB after execution, so AQE keeps sort-merge.
result = large_txns.join(large_accounts, "accountId")
result.explain("formatted")
How do you verify AQE switched the join?
Check the Spark UI SQL tab after the job runs. The final executed plan shows the actual strategy AQE chose, which may differ from the static explain output.
# Static explain may show SortMergeJoin (before AQE runs)
result.explain("formatted")
# But the Spark UI SQL tab shows the FINAL plan:
# BroadcastHashJoin (after AQE switched it)
# Look for:
# - BroadcastExchange on the small side
# - BroadcastHashJoin instead of SortMergeJoin
# - AQEShuffleRead with "local" reads on the large side
What should a senior engineer know?
A senior engineer knows that AQE's dynamic join switching is why you can keep the broadcast threshold at a sensible value and let AQE opportunistically broadcast whenever runtime sizes allow, rather than hand-hinting every join. They also know that the local shuffle reader reuses already-written shuffle data, so the conversion is nearly free.
# Senior practice: rely on AQE for dynamic join switching
# Keep threshold sensible, do not manually hint every join
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "10m")
# For truly critical joins where you KNOW a side is small,
# add an explicit broadcast hint as insurance:
from pyspark.sql.functions import broadcast
result = txns.join(broadcast(vip_customers), "customerId")
Visual Overview
Without AQE (static plan, bad estimate):
Transactions (500M) Customers filtered (5K rows)
│ │
▼ shuffle ▼ shuffle
┌──────────┐ ┌──────────┐
│ Sort by │ │ Sort by │
│ joinKey │ │ joinKey │
└────┬─────┘ └────┬─────┘
│ merge │
└──────────┬────────────┘
▼
SortMergeJoin (slow, unnecessary shuffle)
With AQE (runtime switch):
Transactions (500M) Customers filtered (5K rows)
│ │
│ local read ▼ broadcast (tiny)
│ ┌──────────────┐
│ │ Broadcast to │
│ │ all executors│
│ └──────┬───────┘
│ │
└────────────┬────────────┘
▼
BroadcastHashJoin (fast, no shuffle of large side)
Key Takeaway
AQE dynamically changes join strategies by observing actual data sizes at runtime instead of relying on static estimates that are often wrong. When Catalyst plans a join before execution, it uses table statistics to estimate sizes. If both sides look large, it picks sort-merge join. But after filters and aggregations in earlier stages, a side can shrink dramatically, and AQE detects this and converts the join to a broadcast.
The local shuffle reader makes this conversion even cheaper. When AQE switches from sort-merge to broadcast, the large side may have already been shuffled. Instead of fetching that data across the network, each executor reads the shuffle output already on its local disk, removing the network cost entirely.
At the National Bank of India, AQE's runtime join switches quietly eliminated many heavy shuffles in our reporting queries. A join of transactions to a regionally filtered branch dimension would have been a sort-merge join based on the static estimate, but AQE saw the filtered dimension was tiny and switched to a broadcast. The local shuffle reader meant the large transactions side was read locally with no extra network transfer.
The practical lesson is that AQE lets you keep the broadcast threshold at a sensible value and rely on it to broadcast opportunistically whenever runtime sizes allow, rather than manually hinting every join.
Remember one sentence: AQE switches a planned shuffle join to a broadcast join at runtime when it discovers a side is actually small, and the local shuffle reader reuses already-written data locally.
---