Part XV — Adaptive Query Execution: What AQE Actually Rewrites at Runtime
Spark plans your query before it has seen your data. AQE is the engine’s admission that the planner was guessing — and the machinery that lets the runtime rewrite the plan after every shuffle, using statistics that only exist once the bytes have actually moved.
Part I | Part II | Part III | Part IV | Part V | Part VI | Part VII | Part VIII | Part IX | Part X | Part XI | Part XII | Part XIII | Part XIV | Part XV
The Job That Ran 4 Hours 47 Minutes Every Night for a Year
Friday, 1:12 AM. The clickstream sessionization job — 1.8TB of raw events, keyed by session_id, shuffled, windowed, written out — finished in 4 hours 47 minutes. It had finished in roughly that time every night for a year. The on-call runbook said "sessionization is slow, it's just big," and everyone believed it, because the Spark UI showed the same shape every night: 200 shuffle partitions, three of them glowing red, each holding ~78% of the data between them while 197 partitions finished in minutes and sat idle.
The standard playbook had been tried. More executors: no change — the three hot partitions still had to be processed serially by three tasks. spark.sql.shuffle.partitions=2000: worse, now there were 2,000 tiny files and the same three hot keys. Salting (Part VII's playbook): worked, but it meant rewriting the sessionization logic around salted keys and un-salting downstream, and nobody wanted to touch a pipeline that "worked."
Then someone noticed a config line inherited from the Spark 2.4 migration, three years stale: spark.sql.adaptive.enabled=false. AQE had shipped in Spark 3.0, been enabled by default in 3.2, and this cluster had been carrying the opt-out like a vestigial organ. We flipped it on for the next run. The job finished in 52 minutes. Same data, same cluster, same query. The planner had been flying blind for a year, and the fix was deleting one line.
So: what does AQE actually rewrite at runtime, which knobs control each rewrite, and where does the adaptive machinery run out of road?
The Mechanic Underneath: The Plan Is Re-Planned After Every Shuffle
The static Spark planner (Part III’s Catalyst) optimizes your query once, before execution, using table statistics that are frequently stale, missing, or lies. It picks join strategies, partition counts, and build sides based on estimates. AQE replaces that single planning moment with a loop:
Execute a stage → collect runtime statistics from the actual shuffle output → re-optimize the rest of the plan → repeat.
The unit of adaptivity is the query stage: a subtree of the plan that ends at a shuffle boundary. AQE wraps each one in a QueryStageExec, runs it, and harvests real numbers — actual partition sizes, actual row counts — from the shuffle files that were just written. The remaining plan is then re-planned with truth instead of estimates. In the Spark UI's SQL tab you'll see it as AdaptiveSparkPlan with collapsed stages; each collapse point is a place where the plan changed its mind.
Three rewrites happen in that loop, each controlled by its own configs, each fixing a different way the static planner guesses wrong. They are: coalescing shuffle partitions, switching join strategies, and splitting skewed partitions.
Coalescing: 200 Empty Partitions Were Never a Plan
spark.sql.shuffle.partitions (default 200) is a static guess made before the query runs. Set it for your biggest shuffle and every small shuffle pays the tax: 200 partitions for a 50MB shuffle means 200 tasks, 200 files, and scheduling overhead that dwarfs the actual work. Set it for the small shuffle and the big one gets 200 giant partitions.
With spark.sql.adaptive.coalescePartitions.enabled=true (default), AQE measures the actual shuffle output after the stage completes and merges adjacent small partitions until each is close to spark.sql.adaptive.advisoryPartitionSizeInBytes (default 64MB). Your 50MB shuffle becomes 1 partition. Your 1.8TB shuffle keeps its parallelism. One config, both problems gone.
The coalescing respects a floor: it won’t merge below the advisory size target, and spark.sql.adaptive.coalescePartitions.initialPartitionNum (defaulting to spark.sql.shuffle.partitions) caps how many it starts from. The practical upshot is that spark.sql.shuffle.partitions stops being a tuning knob you agonize over and becomes a rough upper bound. Set it generously; AQE trims.
Join Switching: Sort-Merge at Plan Time, Broadcast at Runtime
Part VI covered the static join strategy choice: broadcast hash join for small build sides, sort-merge for everything else. The static planner makes that call from estimated sizes. When the estimate is wrong — the “small” table is actually 40GB, or the “huge” table filtered down to 3MB — you’re locked into the wrong algorithm for the entire query.
AQE re-evaluates the join strategy after each stage using runtime statistics. The classic win: the planner chose sort-merge join because the build side was estimated at 2GB, but after filters and the actual shuffle the build side is 8MB. AQE converts the sort-merge join into a broadcast hash join mid-flight — no shuffle of the big table, no re-plan by you. The threshold it uses is spark.sql.adaptive.autoBroadcastJoinThreshold, which defaults to the familiar spark.sql.autoBroadcastJoinThreshold (10MB). It can also demote to a shuffled hash join where that fits.
This is the feature that quietly retires a whole class of BROADCAST hints. Part XIV showed what happens when a hint fossilizes a wrong size assumption; AQE's runtime join selection is the engine doing that decision with fresh data instead of your eighteen-month-old guess. It doesn't make broadcast free — the driver still collects (Part XIV's bottleneck is still real) — but it makes the decision correct.
Skew Handling: Splitting the Partition the Planner Couldn’t See
This is the one that fixed our sessionization job. AQE detects skew after a shuffle stage completes, by looking at actual partition sizes: a partition is skewed if it’s larger than spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (default 256MB) and larger than spark.sql.adaptive.skewJoin.skewedPartitionFactor (default 5) times the median partition size. Both conditions — absolute and relative — must hold, which is why it doesn't fire on uniformly large shuffles.
When it fires, on a sort-merge or shuffled-hash join, AQE splits the skewed partition into sub-partitions and replicates the corresponding partition of the other side to each split. The hot key’s work fans out across many tasks instead of serializing on one. In our job, the three monster partitions — 78% of 1.8TB — were split into dozens of pieces, each paired with a replicated copy of the small side. The 4h47m wall clock was three tasks; the 52-minute wall clock was hundreds.
Two details matter. First, skew splitting only applies to sort-merge and shuffled-hash joins — if AQE already converted the join to broadcast, there’s nothing to split, because every executor holds the full build side anyway. Second, the split count derives from the skewed partition’s size relative to the advisory partition size, so advisoryPartitionSizeInBytes quietly controls skew handling granularity too.
Reading an Adaptive Plan: Where the Rewrites Show Up
AQE is invisible if you don’t know where to look, which is why teams run with it for years without realizing what it’s doing — or run without it for years without realizing what they’re missing.
Get chakshu_salgotra’s stories in your inbox
Join Medium for free to get updates from this writer.
In the Spark UI’s SQL tab, an adaptive query shows AdaptiveSparkPlan at the root, with stages that render collapsed: each collapsed node is a QueryStageExec whose final plan was chosen at runtime. Expand the final stage and you'll see AQEShuffleRead nodes where the static plan had plain Exchange — those are the reads that got coalesced or skew-split. The "Details" for a stage show the runtime statistics AQE collected: actual size, actual rows.
The fastest diagnostic: EXPLAIN the same query with spark.sql.adaptive.enabled true and false, and diff them. The non-adaptive plan is what Catalyst guessed. The adaptive plan's final form is what the data required. The distance between them is the value AQE added — and on our sessionization job, that distance was 200 static partitions versus 34 coalesced ones with three skew splits.
The Fix, and What It Cost
The naive state of the sessionization pipeline: AQE disabled by a three-year-old migration config, a static partition count tuned by superstition, and a manual salting design doc nobody wanted to implement.
# Naive: static planning with a fossilized opt-out. The planner guesses 200
# partitions and sort-merge join from stale stats; skew is discovered by humans
# at 1 AM instead of by the engine at runtime.
spark.conf.set("spark.sql.adaptive.enabled", "false") # carried over from the Spark 2.4 migration
spark.conf.set("spark.sql.shuffle.partitions", "200") # tuned once, wrong for every query shapesessions = (
events
.groupBy("session_id")
.agg(...) # 3 partitions hold 78% of 1.8TB; 197 sit idle
)
# Result: 4h47m nightly, three red partitions in the UI, "just big" in the runbook.The engine-aware fix: delete the opt-out, set the advisory size to match the workload, and let the runtime see the skew.
# Optimized: let the runtime plan with real statistics. No query rewrite needed.
spark.conf.set("spark.sql.adaptive.enabled", "true") # default since 3.2; we had opted out
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", 128 * 1024 * 1024) # 128MB: our files are large
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true") # default true; explicit for the runbook
# skewJoin.skewedPartitionThresholdInBytes=256MB, skewedPartitionFactor=5: defaults were finesessions = (
events
.groupBy("session_id")
.agg(...) # identical query text
)
# AQE coalesced 200 -> 34 partitions, skew-split the 3 hot ones, runtime: 52 minutes.No salting, no query rewrite, no logic changes downstream. The query text is byte-identical; only the planning loop changed.
Before/after: 4h47m wall clock → 52 minutes. Shuffle partitions 200 static → 34 coalesced with skew splits. EMR cost per run ~$96 → ~$18. The three red partitions in the UI are gone — not because the skew disappeared, but because the engine finally saw it and split it.
The Catch: Where the Adaptive Machinery Runs Out of Road
AQE is powerful the way a good autopilot is powerful: within its envelope. Outside it, you’re hand-flying again.
First, AQE needs a shuffle to learn from. The first query stage runs on static planning, always — there are no runtime statistics yet. A query with no shuffles gets zero adaptivity. A query whose only shuffle is at the very end gets its statistics too late to matter.
Second, skew handling only covers sort-merge and shuffled-hash joins. If your skew is in a groupBy aggregation or a window function, AQE watches it happen and does nothing. That's still Part VII territory: salt it yourself.
Third, runtime statistics cost scheduling. Each query stage boundary is a synchronization point — stages can’t overlap the way a fully pipelined static plan might. For queries with many small stages, the stage-scheduling overhead can eat the coalescing gains. It’s usually worth it. Usually is doing work in that sentence.
Fourth, AQE’s broadcast conversions still hit Part XIV’s bottleneck. Converting a sort-merge join to a broadcast hash join at runtime means the driver collects the build side right then. If the runtime statistics say the build side is 9GB and your driver has 8g of heap, AQE will cheerfully schedule a driver OOM. The adaptive threshold deserves the same scrutiny as the static one.
Fifth, join reordering still wants real table statistics. AQE can change strategies mid-flight, but the initial join order comes from Catalyst, and Catalyst without CBO stats is guessing. ANALYZE TABLE ... COMPUTE STATISTICS remains the cheapest performance work nobody does.
Takeaways
- The static planner guesses; AQE measures — every rewrite is funded by runtime statistics collected at shuffle boundaries.
- Coalescing retires
spark.sql.shuffle.partitionsas a tuning knob — set it as a generous upper bound and let the 64MB advisory size trim the rest. - Runtime join switching is the anti-hint — it makes the broadcast decision with fresh data instead of your eighteen-month-old guess.
- Skew splits fire at 256MB and 5× median, on sort-merge/shuffle-hash joins only — aggregation skew is still yours to salt.
- Read the adaptive plan, not the query text — collapsed stages and
AQEShuffleReadnodes are where the rewrites show up in the UI. ANALYZE TABLEis still the cheapest optimization — AQE fixes strategy, but join order still needs real statistics to start from.
The most expensive query in your warehouse is the one the planner imagined and the data never matched.
If this saved you a 3 AM pager incident, follow me on Medium for Twice a week for the next 3 Months of Advanced Data Engineering & AI. Let’s connect on LinkedIn and X.
