Adaptive Query Execution

01 · What it is

Re-planning at runtime, on real numbers

Adaptive Query Execution (AQE) is Spark re-optimizing the physical plan at runtime, using the actual statistics it collects after each shuffle — instead of trusting the static, estimate-based plan Catalyst built before the job ran.

A non-adaptive plan is decided once, from table stats and file sizes that are routinely wrong: a filter shrinks a table, a groupBy makes uneven partitions, a join side turns out tiny after a WHERE. AQE waits until a shuffle stage finishes, reads the real per-partition sizes, then rewrites whatever is left of the plan to match reality.

analogy A non-adaptive plan is a route printed before you leave; AQE is live GPS that reroutes you at every junction once it sees the actual traffic. The deep dive below breaks down exactly what it reroutes; for now just hold the shape:

02 · Why it matters

The modern default answer to "my shuffle is slow"

• It fixes the two most common shuffle pathologies automatically — "too many tiny partitions" (200 default partitions → 200 wasteful tasks) and "one giant skewed partition" (a hot key → one straggler task) — with no code changes.
• Static estimates are routinely wrong. Catalyst plans from stats gathered before filters run. A 5 GB dimension a WHERE shrinks to 8 MB is still planned as a two-big-table shuffle — until AQE sees the real 8 MB and broadcasts it.
• It's what interviewers expect you to reach for first. Before manual repartition, a hand-set shuffle.partitions, or salting, AQE has usually already coalesced, split, and switched — you only hand-tune what it can't fix.
• It composes with the rest of the track. AQE's runtime join switch is the safety net behind Lesson 02's broadcast joins; its skew split is the first defence before Lesson 08's salting. The how is the next section.

03 · How it works — deep dive

The framework, then the three things it does

Everything AQE does follows one loop: run a stage up to a shuffle, measure what really came out, re-optimize the rest of the plan, submit the next stage — repeat at every boundary. So the useful way to understand it is: (1) that measure-and-replan framework, then (2) the three specific rewrites it applies. Each idea below is explained here once — the figures show it, the code verifies it, and later sections cross-reference it, but don't re-explain it.

3.1 · The framework — measure, then re-plan

• Mechanism. Spark splits the plan into query stages at shuffle (and broadcast) boundaries. It runs the leaf stages, collects exact map-output statistics (per-partition byte sizes), then re-optimizes the remaining plan with those real numbers before submitting the next stage.
• Why it needs a shuffle. The statistics AQE relies on are the shuffle's per-partition sizes. No shuffle → no runtime stats → nothing to adapt. It acts at stage boundaries, never mid-stage.
• Trade-off. A tiny planning cost per boundary in exchange for plans built on truth instead of estimates — almost always a win.

The three rewrites at a glance before the detail — each is unpacked in 3.2–3.4 below:

3.2 · Rewrite one — coalesce tiny shuffle partitions

analogy Instead of mailing 200 near-empty envelopes, AQE combines them into a few full ones — fewer trips, same contents.

• Problem. spark.sql.shuffle.partitions = 200, so a shuffle on a modest result makes 200 partitions, many a few KB — 200 tasks, most doing nothing.
• Mechanism. With coalescePartitions.enabled = true, AQE reads real post-shuffle sizes and merges adjacent partitions until each nears advisoryPartitionSizeInBytes (64 MB), never below coalescePartitions.minPartitionSize (1 MB).
• Trade-off. Slightly less parallelism (fewer, bigger tasks) — the 64 MB advisory keeps tasks reasonable even on big results.

use case A daily aggregation outputting a few hundred MB: without AQE it runs 200 tasks and writes 200 tiny files; with coalesce it runs ~5 balanced tasks and writes ~5 files.

3.3 · Rewrite two — split a skewed partition

analogy If one checkout lane has 500 shoppers and the others have 5, AQE opens extra lanes for that one queue so everyone finishes together.

• Problem (skew). Uneven keys mean a few post-shuffle partitions are enormous. The huge one becomes a straggler task — minutes long (maybe spilling/OOM) while every other task finished. The stage waits on it.
• Mechanism. With skewJoin.enabled = true, AQE flags a partition skewed only when it is both > skewedPartitionFactor (5.0) × the median and > skewedPartitionThresholdInBytes (256 MB). It then splits that partition into sub-partitions (replicating the matching side) so the join stays correct and balanced.
• Why both conditions. The factor catches relative skew; the 256 MB floor avoids splitting partitions merely a bit above a tiny median.
• Trade-off. Some replication of the matching side, but one multi-minute straggler becomes parallel sub-tasks. Sort-merge joins only; skew it can't tame → salting (Lesson 08).

use case transactions ⋈ accounts where one "house" account holds 40% of rows — that account_id partition is 5× the median and well over 256 MB, so AQE splits it.

3.4 · Rewrite three — switch sort-merge → broadcast at runtime

analogy You packed for a long road trip (sort-merge), then learn the destination is round the corner — so you ditch the heavy plan and walk (broadcast).

• Problem. Catalyst chose sort-merge from pre-execution estimates, but an upstream filter shrank one side below the broadcast bar — it's about to shuffle two big tables for nothing.
• Mechanism. After the shuffle feeding the join, AQE sees the actual size of each side and, if one is now small, rewrites the join to a Broadcast Hash Join — using a runtime broadcast threshold, distinct from the static autoBroadcastJoinThreshold Catalyst used up front.
• Trade-off. You swap the double shuffle for a broadcast — it fires only when the real size is genuinely small, bounded by the small side fitting in memory.

use case events ⋈ dim_country where dim_country is "big" in the catalog but a WHERE active = true reduces it to a few MB — Catalyst plans sort-merge; AQE flips it to broadcast.

3.5 · The Databricks fourth feature

Open-source Spark documents three AQE features (the three rewrites above). Azure Databricks lists a fourth: detect & propagate empty relations — when a stage produces zero rows at runtime, AQE prunes the dependent branches (e.g. an inner join with one empty side becomes empty without doing the rest of the work). Databricks also exposes a dedicated, higher runtime broadcast switch — see §4 and §5. Don't claim empty-relation propagation as an OSS feature.

Figure A · one shuffle boundary, three rewrites

Step the plan as AQE re-optimizes it

This figure isolates what happens to the partitions at a single shuffle boundary. Step through it: 200 tiny partitions coalesce toward 64 MB, then a hot partition that clears both bars (> 5× median and > 256 MB) splits, then the join flips sort-merge → broadcast once the real size is small. (Where it lands across the nodes is Figure B.) Use Prev / Play / Next.

04 · The code — and how to verify it

Each config, with its proof in the plan

Every technique below is paired with how to prove it worked — the .explain() plan node or the Spark-UI signal (AQEShuffleRead, coalesced/skewed tags). Apply, then verify; never assume.

Confirm AQE is engaged

python# Master switch — on by default since Spark 3.2.0 / DBR 7.3+.
spark.conf.get("spark.sql.adaptive.enabled")   # -> 'true'

# VERIFY — an adaptive plan roots at AdaptiveSparkPlan; after a run it flips to isFinalPlan=true.
df.explain(mode="formatted")
#   == Physical Plan ==
#   AdaptiveSparkPlan isFinalPlan=true        <- AQE is engaged
#   +- == Final Plan ==  *(...) ... AQEShuffleRead coalesced

Coalesce — tune the target and the floor

python# Defaults shown explicitly (and to tune the target).
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")               # default true
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", 64*1024*1024)   # 64 MB target
spark.conf.set("spark.sql.adaptive.coalescePartitions.minPartitionSize", 1*1024*1024)  # 1 MB floor

agg = sales.groupBy("store_id").sum("amount")   # a shuffle (wide dependency)

# VERIFY — post-shuffle partition count collapses from 200 toward a handful.
print(agg.rdd.getNumPartitions())   # AQE-coalesced count (e.g. ~5), not 200
agg.explain(mode="formatted")       # look for: AQEShuffleRead coalesced

Skew join — the lever set (needs BOTH conditions)

pythonspark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")               # default true
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5.0")  # > 5x median ...
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes",
               256*1024*1024)                                    # ... AND > 256 MB

joined = transactions.join(accounts, "account_id")   # sort-merge join on a skewed key

# VERIFY — the plan shows the skew was detected and split.
joined.explain(mode="formatted")    # look for: AQEShuffleRead ... skewed=true
# Spark UI (Stages tab): max task time collapses toward the median.

Join switch — OSS static vs Databricks runtime threshold

python# OSS static threshold — used by Catalyst BEFORE runtime:
spark.conf.get("spark.sql.autoBroadcastJoinThreshold")             # -> '10485760' (10 MB)
# Databricks AQE RUNTIME switch — a higher, dedicated bar:
spark.conf.get("spark.databricks.adaptive.autoBroadcastJoinThreshold")  # -> 30 MB (Databricks)

joined = events.join(dim_country.where("active = true"), "country_code")

# VERIFY — the FINAL (post-run) plan flips SortMergeJoin -> BroadcastHashJoin.
joined.explain(mode="formatted")
#   AdaptiveSparkPlan isFinalPlan=true
#   +- == Final Plan ==   BroadcastHashJoin ...   <- AQE switched it
#   +- == Initial Plan == SortMergeJoin ...        <- stale-estimate plan

Same configs from Spark SQL

sqlSET spark.sql.adaptive.enabled = true;
SET spark.sql.adaptive.coalescePartitions.enabled = true;
SET spark.sql.adaptive.skewJoin.enabled = true;

-- VERIFY — read the adaptive plan (AdaptiveSparkPlan root):
EXPLAIN FORMATTED
SELECT account_id, count(*) FROM transactions GROUP BY account_id;

Figure B · where it runs

See it re-plan across the cluster

This figure isolates a different thing from Figure A: not what happens to the partitions, but where the driver and executors do the work. Step through the same job on 1 driver + 3 executors — the driver re-plans at each boundary from the stats the executors report, coalescing the tiny partitions, splitting the hot one across nodes, and flipping the join to a broadcast. Use Prev / Play / Next.

05 · The three rewrites at a glance

Which rewrite, and what proves it fired

Read this by the problem you're seeing — it answers "which AQE rewrite is this, and what's the plan/UI signal?" at a glance. (The why is §3; this is the which.)

06 · Edge cases & limitations

Where it gets subtle

New nuance an interviewer probes — not a re-run of the three rewrites:

• No shuffle, no adaptation. A map-only job (pure select/filter) has no shuffle stats — don't expect coalescing on a shuffle-free plan.
• Skew that misses a threshold. A partition 5× the median but only 100 MB (< 256 MB) is not split — both conditions must hold. Residual skew is a salting job (Lesson 08).
• Skew split is sort-merge only. It doesn't split skew inside a broadcast or shuffle-hash join.
• Forcing broadcast() can pre-empt AQE. A manual hint fixes the strategy up front and removes AQE's runtime choice — prefer letting AQE decide unless you have a known, capped small side.

Hard limits worth stating plainly:

• Planning-layer only. AQE rewrites the plan; it does not change execution (that's Tungsten/Photon). It won't add partition pruning (Lesson 07) or bucketing (Lesson 11) for you.
• It adapts the remaining plan, not the past. Work in a completed stage isn't redone; AQE only optimizes downstream of the boundary it just observed.
• Read the plan post-run. explain() before execution may show isFinalPlan=false / the initial plan. The final adaptive plan is only certain after the run — or in the Spark UI's final DAG.
• OSS-vs-Databricks caveat. The three rewrites are OSS-documented; the fourth (empty-relation propagation) and the 30 MB runtime switch are Databricks-specific, stated as such.

07 · Common mistakes

Gotchas — the mistake and its one-line fix

08 · References

Sources

• Apache Spark — SQL Performance Tuning (AQE: coalesce, skew join, join switch)
• Apache Spark — Configuration (spark.sql.adaptive.*, shuffle.partitions, autoBroadcastJoinThreshold)
• Apache Spark — SQL join/partitioning hints (REBALANCE & AQE interplay)
• Azure Databricks — Adaptive Query Execution (DBR default, 4th feature, 30 MB runtime switch)
• Azure Databricks — Photon (native execution engine, complementary to AQE)