Adaptive Query Execution (AQE)

The last lesson ended on the line at the top of every plan we read straight past: AdaptiveSparkPlan isFinalPlan=false. We noticed it was the optimizer admitting its plan was provisional — built from estimates that lie, reserving the right to change its mind once it sees the real data. We asked what that runtime second-guesser is and what it does between stages. This lesson is it.

Catalyst builds the plan before execution, using static information (schemas, table stats). That information is often wrong — stats are stale, post-filter sizes are unknown, joins produce unpredictable amounts of data. The plan that looked good on paper isn’t always the one that runs fastest. Catalyst can’t fix this because it only reads catalog metadata; it doesn’t know how many rows actually survive a filter until a stage runs.

Adaptive Query Execution (AQE), on by default in Spark 3, fixes this. It pauses between stages, looks at what actually came out, and re-optimizes the rest of the plan.

It’s the single biggest reason Spark 3 needs less hand-tuning than Spark 2 did.

The three big AQE wins

AQE does three things at runtime:

1. Coalesce small partitions — merge tiny post-shuffle partitions into fewer, larger ones
2. Switch join strategy — turn a SortMergeJoin into a BroadcastHashJoin if the actual size turns out to be small
3. Handle skew — detect and split oversized partitions

Each one solves a real problem that used to require manual tuning.

Win 1: Coalesce shuffled partitions

Spark’s default shuffle partition count is 200 (spark.sql.shuffle.partitions). This is fine for a 100GB dataset but absurd for a 200MB result of a heavy filter — you’d get 200 partitions of 1MB each, with task scheduling overhead dwarfing the actual work.

AQE looks at the post-shuffle partition sizes and merges tiny adjacent partitions:

Without AQE:
Stage 1 output → 200 partitions × ~1MB each → 200 tasks in Stage 2

With AQE:
Stage 1 output → 200 partitions × ~1MB each
              → AQE: "these are all tiny, merge to 20 partitions of ~10MB"
              → 20 tasks in Stage 2

The result: 10x less scheduling overhead, same total work.

The config:

spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "64MB")

AQE aims for advisoryPartitionSizeInBytes per partition (default 64MB). Set it higher for very large jobs, lower if you want more parallelism.

Win 2: Switch join strategy at runtime

You wrote a join. Spark’s stats said both sides were big, so it picked SortMergeJoin (with a shuffle). But after applying filters, the right side turned out to be only 5MB. A broadcast would have been faster.

AQE notices and switches:

Plan time:    SortMergeJoin (big × big)
Stage 1 runs: actually emits 4MB
AQE:          "the right side is small now — switch to BroadcastHashJoin"
Stage 2 runs: as a broadcast join

This kills one of the most common manual-tuning chores. Pre-AQE you’d have to:

1. Run the job, watch it be slow
2. Inspect the plan, see the unnecessary SortMergeJoin
3. Manually add F.broadcast(...) or pre-compute the small side
4. Re-run

With AQE, you just run the job.

# On by default in Spark 3.x — no extra config needed for join switching
spark.conf.set("spark.sql.adaptive.enabled", "true")

Win 3: Skew join handling

If one join key dominates (say country = 'US' is 80% of your data), all those rows hash to one partition. That executor’s task runs 10-100x longer than the others, blocking the whole stage.

AQE detects this at runtime: it sees that one post-shuffle partition is much larger than the others, and splits it into sub-partitions:

Now the US data processes in parallel across 5 tasks instead of one. The stage finishes 5x faster.

The config:

spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB")

A partition is considered “skewed” if it’s both:

• More than skewedPartitionFactor × the median partition size (default 5x)
• Larger than skewedPartitionThresholdInBytes (default 256MB)

For severely skewed data, you may still need manual salting (covered in the skew lesson). But for most real-world skew, AQE handles it automatically.

How to know AQE is doing something

The Spark UI shows AQE actions. In the SQL/DataFrame tab, the DAG diagram for an AQE-affected query has a special “AdaptiveSparkPlan” node at the top with sub-plans for “isFinalPlan=false” (the original) and the final adapted plan.

In code:

result.explain(mode="formatted")
# == Physical Plan ==
# AdaptiveSparkPlan isFinalPlan=true
# +- ... the final, adapted plan

If isFinalPlan=true, you’re looking at the plan AQE actually ran (after the runtime decisions). If isFinalPlan=false, the plan hasn’t executed yet and AQE will still make decisions.

The relevant configs

AQE has a lot of knobs. The five you’ll touch in real work:

# Master switch — on by default in Spark 3
# Also enables runtime join strategy switching (SMJ → BHJ) automatically
spark.conf.set("spark.sql.adaptive.enabled", "true")

# Coalesce small post-shuffle partitions
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "64MB")

# Skew join handling
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

You can leave all of these at defaults for most jobs. Tune advisoryPartitionSizeInBytes if you find AQE is producing too many small partitions (raise it) or single huge partitions (lower it).

A toy adaptive plan

The idea behind AQE — observe a stage’s output, then re-plan — can be sketched in pure Python:

# Toy: pick a join strategy based on observed sizes.

def initial_plan(left_estimated_size, right_estimated_size):
    # At plan time, estimates say both are big → SortMergeJoin
    if left_estimated_size > 100 and right_estimated_size > 100:
        return "SortMergeJoin"
    elif left_estimated_size < 10 or right_estimated_size < 10:
        return "BroadcastHashJoin"
    return "SortMergeJoin"

def aqe_replan(initial, actual_left_size, actual_right_size):
    # After running stage 0, we see the actual sizes.
    if initial == "SortMergeJoin":
        if actual_right_size < 5:
            return "BroadcastHashJoin (right side actually small)"
        if actual_left_size < 5:
            return "BroadcastHashJoin (left side actually small)"
    return initial

# Scenario: estimates said both big, but a filter made one tiny
estimated_left, estimated_right = 500, 200
plan_now = initial_plan(estimated_left, estimated_right)
print(f"Plan at compile time: {plan_now}")

actual_left, actual_right = 480, 3   # actual size after filters
plan_aqe = aqe_replan(plan_now, actual_left, actual_right)
print(f"Plan after AQE:       {plan_aqe}")

Plan at compile time: SortMergeJoin
Plan after AQE:       BroadcastHashJoin (right side actually small)

There it is in two lines: the estimates said both sides were big (500 and 200), so the compile-time plan chose a shuffle-heavy SortMergeJoin. But once stage 0 actually ran, the right side came out at 3 — a filter had demolished it — and the re-plan step swapped in a BroadcastHashJoin, killing the shuffle on the big side. That is the entire idea of AQE compressed into an if: don’t trust the estimate, run a stage, measure, then decide. Real Spark weighs cost, shuffle bytes, and a dozen other factors, but the shape is exactly this — observe, then adapt.

In one breath

Adaptive Query Execution (on by default in Spark 3) closes the gap Catalyst can’t: it pauses between stages, looks at what actually came out, and re-optimizes the rest — doing three things that used to be manual tuning chores: coalescing tiny post-shuffle partitions into fewer ~64 MB ones (cutting scheduling overhead from the default 200), switching a SortMergeJoin to a BroadcastHashJoin when a side turns out small after filters, and splitting skewed partitions so one hot key doesn’t gate a stage for hours; you confirm it ran via isFinalPlan=true in the explain output, the defaults are good for ~80% of jobs, and severe skew may still need manual salting.

Practice

Before the quiz, map each AQE feature to the manual chore it retired. Pre-Spark-3, what four-step dance did an engineer do when they spotted a SortMergeJoin that should’ve been a broadcast — and which single AQE feature collapses all four into “just run the job”? Then the limits: the lesson says AQE handles most skew but severe skew still needs salting — why can’t advisoryPartitionSizeInBytes and skew-splitting fully solve a key that is 80% of the data?

A question to carry forward

Notice the quiet boundary on everything AQE does. Coalesce, switch the join, split the skew — every one of those is a runtime fix to a query. AQE re-tunes the plan after the data is already moving. It made “optimize this slow Spark job” mostly disappear from the ticket queue, which is exactly why the industry note in this lesson said modern Spark performance work has shifted to something AQE never touches: data layout — how the bytes are physically arranged on disk before any query runs.

And that lever is entirely yours. AQE can split a skewed partition once it has formed, but it can’t change the fact that your table was written as 50,000 tiny files, or that it’s sorted so that a date filter still has to scan everything. Those are decisions baked in at write time, invisible to any runtime optimizer, and they often dominate job cost more than any query rewrite. So the question to carry forward, now that the engine optimizes the query for you: what is the one performance lever the optimizer can’t pull — how you physically partition data on disk so that reads skip what they don’t need — and how do you wield it? That is partitioning, and it is the next lesson.