Why architecture matters here
The architectural insight behind AQE is that Spark's execution model already contains perfect measurement points. A shuffle is a full materialization barrier: every map task writes its output, indexed by reduce partition, before any reducer starts. At that instant, the sizes are not estimates — they are files with lengths. Classic Spark threw that information away; AQE feeds it back. The elegance is that no new synchronization is added: the barrier was always there; the optimizer just starts listening to it. This is also why AQE's scope is precisely 'everything after the next shuffle' — work already executed is sunk cost, but join strategies, partition counts, and physical operators downstream are all still negotiable.
What it buys operationally is the death of three chronic tuning battles. spark.sql.shuffle.partitions — the config every team fought over (200 default: too many for small stages, too few for giant ones) — becomes a ceiling rather than a decision: set it high, let coalescing shrink each stage to fit its actual data. Broadcast threshold guesswork — under-estimates that OOM'd executors, over-estimates that forfeited the fastest join — becomes self-correcting: the decision re-fires with exact sizes. Skew firefighting — the salting ceremonies and hand-split hot keys — becomes a runtime service: partitions exceeding the skew threshold are subdivided automatically, with the matching side's data duplicated to preserve join semantics.
The honest costs: adaptivity trades a little scheduling latency (stages submit as their inputs complete, and re-planning takes milliseconds) and some plan-stability (the same query may execute differently as data drifts — a feature for performance, an occasional surprise for debugging). And AQE cannot fix the plan before the first shuffle — a catastrophically wrong scan-side decision still executes; fresh statistics and DPP remain the compile-time defenses. Understanding where adaptation can and cannot reach is exactly what separates informed tuning from cargo-culted configs.
The architecture: every piece explained
Top row: the loop. The initial plan comes from Catalyst's CBO as always; AQE then cuts it into query stages at every exchange (shuffle or broadcast) boundary. Stages whose inputs are ready execute; as each materializes, runtime statistics — MapOutputStatistics: bytes and rows per reduce partition — flow to the AQE framework, which runs re-optimization: the remaining logical plan is revisited with true sizes, physical strategies are re-chosen, and the next stages are created from the updated plan. In the UI this appears as AdaptiveSparkPlan with isFinalPlan=false evolving to the final plan — reading the difference between initial and final plans is the diagnostic skill AQE asks of you.
Middle row: the adaptations. Coalescing: post-shuffle partitions are merged up to the advisory size (default 64MB) — 200 configured partitions carrying 900MB total become ~14 real tasks; small-stage overhead, task-launch storms, and tiny-file outputs all shrink together. Join switching: when a materialized side lands under the broadcast threshold (10MB default for the adaptive check), sort-merge converts to broadcast-hash — and with localShuffleReader, the already-shuffled other side is read map-locally, avoiding the network round it no longer needs. Skew handling: a partition larger than skew-factor × median (and above the absolute threshold) is split into sub-partitions; the join's other side duplicates its matching partition per split — correctness preserved, the straggler dissolved into parallel tasks. Empty-relation pruning: stages that materialize zero rows collapse entire subtrees (joins become empty, unions simplify) — common with selective filters and over-provisioned unions.
Bottom rows: the context. Shuffle materialization is the enabling substrate — which is why AQE's benefits concentrate in multi-stage queries; a single-stage scan-aggregate adapts little. DPP interplay: dynamic partition pruning (broadcast a dimension's filter results into the fact scan) happens at compile/execution start and composes with AQE's later adaptations; both features want accurate broadcast decisions, so thresholds interact. The ops strip holds the knobs that matter: advisoryPartitionSizeInBytes (coalescing target — tune per workload memory profile), autoBroadcastJoinThreshold and the adaptive variant, skewJoin.skewedPartitionFactor/ThresholdInBytes, and coalescePartitions.parallelismFirst (default favors parallelism over the advisory size — the setting most teams should flip to false for stable batch workloads).
End-to-end flow
Run a retail join through AQE's decisions. The query: transactions (1.2TB) joined to a filtered merchant dimension, grouped by category. The CBO, working from week-old stats, estimates the filtered merchants at 400MB — too big to broadcast — and plans sort-merge: shuffle both sides into 1,000 partitions (the team's configured ceiling).
Stage 1 scans and shuffles the merchant side. Materialized truth: the filter was vicious — 22MB total. Re-optimization fires: 22MB < threshold → the sort-merge becomes a broadcast-hash join; the planned shuffle of 1.2TB of transactions is replaced by a local shuffle reader over the map outputs stage 1's sibling already wrote — no network redistribution of the big side at all. The transaction-side aggregation still shuffles for the group-by: its map output totals 6.1GB across 1,000 configured partitions, and coalescing merges them into 96 tasks of ~64MB. Within that shuffle, reality adds a twist: one merchant category ('GROCERY') dominates — partition 41 holds 1.9GB against a 12MB median. Skew handling splits it into 30 sub-tasks. The stage runs 96+29 tasks of comparable duration; the UI's final adaptive plan shows skewed=true on the reader and the join node annotated with the strategy switch. Wall clock: 4m 10s. The same query with AQE off, benchmarked for the migration review: 19m — a 1,000-task shuffle of 1.2TB it didn't need, followed by one straggler task grinding 1.9GB for eleven minutes.
The team's tuning session afterward is the realistic coda. First finding: their ETL's output stage was writing 96 files where 24 would do — advisory size raised to 256MB for the write-heavy pipeline, with parallelismFirst=false so the advisory actually governs. Second: one dashboard query flip-flopped between broadcast and sort-merge day to day (its dimension hovers near the threshold), producing confusing latency variance; they pinned it with a broadcast hint — AQE is a safety net, and hints still outrank it where stability matters more than adaptivity. Third: a lingering straggler turned out to be skew below the absolute threshold (a 240MB partition against 8MB medians); lowering skewedPartitionThresholdInBytes for that job let the splitter engage. The meta-lesson they wrote down: AQE moved tuning from guessing-before to reading the adaptive plan after — the plan diff is the new profiler.