Spark Adaptive Query Execution: Dynamic Coalescing, Pruning, and Skew Handling

Scenario context

Before AQE, Spark compiled your entire query into a static physical plan using size estimates that were frequently wrong — and a wrong estimate at planning time meant a skewed join, 800 small tasks, or a missed broadcast opportunity that no amount of executor tuning could fix. Adaptive Query Execution (introduced in Spark 3.0, production-hardened in 3.2) rewrites that plan at every shuffle boundary using real partit…

This walkthrough focuses on Three Walkthroughs: Watching AQE Change the Query Plan in Real Scenarios.

Before making a design choice, clarify:

• These three walkthroughs trace the exact before-and-after execution path for each major AQE feature.
• The intent is to make the re-planning rules concrete by connecting the abstract mechanism to observable Spark UI and EXPLAIN plan changes.
• Scenario 1: Dynamic Coalescing Converts 800 Tiny Tasks into 12 Right-Sized Ones The query: a daily revenue aggregation that reads 30 GB of transaction events, filters to the last 7 days (retaining ~3 GB), groups by merc…

Concepts in play: spark-data-engineering, jvm-internals-and-garbage-collection, tree-data-structures, database-sharding

Stage 1 — The 199 Idle Tasks: When One Skewed Partition Holds an Entire Cluster Hostage

• A data engineering team at a logistics company had promoted their weekly customer-segment enrichment pipeline to production.
• The job joined a 600 GB behavioral event table against a 12 GB customer dimension table, partitioned by region.
• On staging — a 15 GB sample — the job completed in four minutes.
• In production, it hit the nine-hour timeout and produced no output.

Observable outcome: In production, it hit the nine-hour timeout and produced no output.

Stage 2 — How AQE Breaks the Static Plan Barrier: The Re-Optimization Loop Between Shuffle Stages

• To understand what AQE does differently, you need to understand what Catalyst does without it.
• When you submit a Spark SQL query or DataFrame chain, Catalyst transforms it through four phases: analysis (bind column names to the catalog), logical optimization (push predicates, fold constants, eliminate redundant p…
• broadcast hash join, pick sort orders), and code generation (compile the physical plan to JVM bytecode).
• The physical plan produced by this process is fixed before a single row of data is read .

flowchart TD
    A["Query Submitted to Spark"] --> B["Catalyst Produces Static Physical Plan"]
    B --> C["AdaptiveSparkPlanExec Wraps the Plan"]
    C --> D["Leaf QueryStageExec Stages Execute"]
    D --> E["ShuffleQueryStageExec Stage Completes"]
    E --> F["Driver Collects MapOutputStatistics"]
    F --> G{"Rewrite Rules Fire?"}
    G -->|"Coalesce or Skew Split or Broadcast Upgrade"| H["Plan Tree Rewritten with Real Stats"]
    G -->|"No Rule Matches"| I["Plan Continues Unchanged"]
    H --> J["Next Stages Submit with Rewritten Plan"]
    I --> J
    J --> K{"More Shuffle Boundaries Remain?"}
    K -->|"Yes"| E
    K -->|"No"| L["Job Completes"]

Observable outcome: The physical plan produced by this process is fixed before a single row of data is read .

Stage 3 — Deep Dive: QueryStageExec, ShuffleQueryStageExec, and How Spark Rebuilds the Plan Mid-Flight

• The Internals: From Catalyst Physical Plan to Adaptive Re-Planning The entry point for AQE is the AdaptiveSparkPlanExec node.
• When spark.sql.adaptive.enabled = true , Catalyst's QueryExecution wraps the final physical plan in this node before code generation.
• AdaptiveSparkPlanExec holds a mutable reference to the current physical plan and maintains the state machine that controls re-planning.
• The plan tree is decomposed into QueryStageExec subtrees at shuffle and broadcast exchange boundaries.

targetPartitions = max(ceil(B / T), minPartitionNum)

s_i > skewedPartitionFactor × M
s_i > skewedPartitionThresholdInBytes

Observable outcome: The plan tree is decomposed into QueryStageExec subtrees at shuffle and broadcast exchange boundaries.

Stage 4 — AQE in Multi-Stage Production DAGs: Where Runtime Re-Planning Fits in the Catalyst Pipeline

• AQE does not replace Catalyst — it extends it.
• Understanding the boundary between static optimization and adaptive re-planning is essential to using both correctly.
• What Catalyst still owns: All logical optimization (predicate pushdown, constant folding, subquery decorrelation, CBO statistics-based join reordering) happens before the physical plan is produced and before AQE wraps i…
• AQE does not re-run logical optimization rules at shuffle boundaries.

flowchart LR
    Mapper["Mapper
(Shuffle Write)"]
    Network["Network
(Shuffle Read)"]
    Reducer["Reducer
(Aggregate)"]

    Mapper -->|"Partitioned Data"| Network
    Network -->|"Shuffled
Partitions"| Reducer

Observable outcome: AQE does not re-run logical optimization rules at shuffle boundaries.

Stage 5 — Partition Counts, Skew Detection Thresholds, and the Runtime Metrics That Drive Every AQE Decision

• AQE's three core features are each governed by a specific set of statistics derived from MapOutputStatistics .
• Understanding what those statistics look like and where to find them in the Spark UI is essential for diagnosing whether AQE is helping and tuning it when it is not.
• Dynamic Coalescing Metrics: The coalesce rule works on bytesByPartitionId — the post-serialization byte size of each partition after a shuffle stage completes.
• Before coalescing, a query with spark.sql.shuffle.partitions = 200 on a 2 GB post-aggregation result produces 200 partitions averaging 10 MB each — acceptable, but potentially worth coalescing to reduce task scheduling…

flowchart TD
    Input["Input Stream"]
    Batch1["Micro-Batch [t, t+Δt)"]
    Process["Processing"]
    Output["Output"]

    Input -->|"Split by interval"| Batch1
    Batch1 --> Process
    Process --> Output
    Output -->|"Repeat"| Input

Observable outcome: Before coalescing, a query with spark.sql.shuffle.partitions = 200 on a 2 GB post-aggregation result produces 200 partitions averaging 10 MB each — acceptable, but potentially worth coalescing to reduce task scheduling…

Stage 6 — Visualizing the AQE Re-Planning Loop and Skew Join Splitting

• The following diagrams show the two most important AQE flows: the full re-planning loop that runs at every shuffle boundary, and the skew join splitting mechanism that converts one blocking task into many parallel ones.
• The re-planning loop below captures the lifecycle of an AdaptiveSparkPlanExec node.
• The key insight is that the plan tree is mutable — each completed shuffle stage produces statistics that feed directly into rule evaluation, and any triggered rule rewrites the plan before the next set of stages is subm…
• Stages that pass through without triggering any rule are scheduled with their original structure unchanged.

flowchart TD
    A[Catalyst produces static physical plan] --> B[AdaptiveSparkPlanExec wraps the plan tree]
    B --> C[Identify leaf-most QueryStageExec nodes]
    C --> D[Submit stages with no pending dependencies]
    D --> E[ShuffleQueryStageExec tasks execute and write shuffle files]
    E --> F[MapOutputStatistics collected per partition]
    F --> G{Evaluate AQE rewrite rules}
    G -->|Coalesce rule fires| H[Merge small consecutive partitions into fewer tasks]
    G -->|Skew rule fires| I[Split oversized partitions into sub-tasks and replicate join side]
    G -->|Broadcast upgrade fires| J[Rewrite sort-merge join as broadcast hash join]
    G -->|No rule fires| K[Retain current stage plan unchanged]
    H --> L[Submit next stages with rewritten plan]
    I --> L
    J --> L
    K --> L
    L --> M{More stages remain?}
    M -->|Yes - more shuffle boundaries| C
    M -->|No - final stage| N[Job completes with fully adaptive plan]

flowchart TD
    A[Shuffle stage completes - partition stats available] --> B[Compute median partition byte size]
    B --> C{Check each partition against skew thresholds}
    C -->|Partition within normal range| D[Schedule as single join task]
    C -->|Partition exceeds skewedPartitionFactor x median and 256MB minimum| E[Flag partition as skewed]
    E --> F[Split into N sub-partitions by byte range]
    F --> G[Replicate right-side matching rows for each sub-partition]
    G --> H[Schedule N parallel join tasks - each handles one sub-partition]
    H --> I[Union all N sub-task outputs]
    D --> J[Collect all join results]
    I --> J
    J --> K[Final joined dataset - uniform task durations]

Observable outcome: Stages that pass through without triggering any rule are scheduled with their original structure unchanged.

End-to-end outcome

After working through the stages, you should be able to explain:

• AQE does not replace data quality work — it rescues you from estimation failures.
• The most common misconception about AQE is that enabling it means you no longer need to think about partition sizing, skew, or join strategy.
• • AQE wraps the static Catalyst physical plan in AdaptiveSparkPlanExec and re-evaluates the plan after each shuffle stage completes, using real MapOutputStatistics instead of pre-execution catalog estimates.
• • Dynamic coalescing groups consecutive small post-shuffle partitions into fewer, right-sized tasks — eliminating the "800 tasks on 400 MB" scheduling overhead without changing partition count settings manually.

Alternatives considered

Configuring AQE for Your Workload: A Decision Guide for Production Engineers

The table below maps workload characteristics to specific AQE configuration decisions.

Failure and degraded modes

When AQE Cannot Save You: Failure Modes, Blind Spots, and Workloads Where Static Plans Win

• Detection: AQE is powerful but not universal.
• User or operator impact: AQE is powerful but not universal.