Adaptive Query Execution (AQE) and how it complements Catalyst | Data Engineering

Q: What is Adaptive Query Execution?

Adaptive Query Execution is a Spark SQL feature that reoptimizes the physical plan during execution using real statistics gathered from completed stages. Instead of relying only on estimates made before the job runs, AQE reads the actual partition sizes at shuffle boundaries and rewrites the remaini

Q: How does AQE complement Catalyst rather than replace it?

Catalyst still does all the upfront work: parsing, analysis, rulebased logical optimization like predicate pushdown and column pruning, and initial physical planning. AQE layers on top, correcting only the physical plan at runtime where static estimates were wrong. Catalyst (before execution): Parse

Q: What should a senior engineer know?

AQE is enabled by default in Spark 3.2+ but was optin in Spark 3.0. The authoritative view of what actually ran is the Spark UI final plan, not the static explain() output. AQE's broadcast threshold may differ from the static autoBroadcastJoinThreshold. For extreme skew where even AQE's split subpar

Definition

Catalyst optimizes the plan before execution using estimates, and AQE re-optimizes the physical plan during execution using real runtime statistics at shuffle boundaries : the two work together.

Adaptive Query Execution is a Spark SQL runtime optimization layer.

Runtime re-optimization – AQE reads real shuffle statistics at stage boundaries and rewrites the remaining physical plan based on facts, not estimates.
Partition coalescing – It merges many small post-shuffle partitions into fewer right-sized ones, reducing task scheduling overhead.
Skew handling – It detects oversized partitions and splits them into smaller sub-partitions, eliminating straggler tasks.
Dynamic join switching – It converts a planned sort-merge join to a broadcast join when a side turns out small enough at runtime.
Complements Catalyst – Catalyst produces the initial plan from static rules and estimates; AQE refines that plan with measured facts.

AQE is enabled by default in Spark 3.x and operates transparently at shuffle boundaries.

Imagine the National Bank of India plans a monthly report based on last year's guess of how many transactions each branch processed. Catalyst is the planner who makes the best plan from those guesses before work begins. But guesses are often wrong: maybe the Mumbai branch actually did ten times more transactions this month. AQE is the supervisor standing at each checkpoint during the work, counting the actual files that just came through, and adjusting the remaining plan on the spot: merging tiny piles together, splitting one giant pile across more clerks, and even changing how two teams combine their work.

Understanding the Concepts

What is Adaptive Query Execution?

Adaptive Query Execution is a Spark SQL feature that re-optimizes the physical plan during execution using real statistics gathered from completed stages. Instead of relying only on estimates made before the job runs, AQE reads the actual partition sizes at shuffle boundaries and rewrites the remaining plan accordingly.

PYTHON

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("BankAQE") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

# AQE is enabled by default in Spark 3.2+
print(spark.conf.get("spark.sql.adaptive.enabled"))
# true

Why does AQE exist?

The Catalyst Optimizer must commit to a physical plan before the job runs, using statistics that are often stale, missing, or wrong. It may not know the true size of a table after filtering, or that one join key is wildly more common than others, or that a shuffle will produce many tiny partitions. These misestimates lead to poor choices at the National Bank of India: too many small tasks for a branch report, one giant skewed task for a corporate account, or a shuffle join where a broadcast join would have been far cheaper.

PYTHON

# Without AQE: Catalyst guesses the join strategy
# If the estimate is wrong, you get a slow sort-merge join
txns = spark.read.parquet("/data/bank/transactions")
branches = spark.read.parquet("/data/bank/branches")

# Catalyst may pick sort-merge because it estimates branches as large
result = txns.join(branches, "branch_code")
result.explain()  # Shows SortMergeJoin in the static plan

How does partition coalescing work?

After a shuffle, the output may contain hundreds or thousands of small partitions, each becoming a task. AQE reads the actual partition sizes and merges adjacent small partitions into fewer right-sized ones, so the next stage runs fewer, better-balanced tasks.

PYTHON

# Configure coalescing thresholds
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128m")

# After a shuffle, if 200 partitions average 1 MB each,
# AQE merges them into ~2 partitions of 128 MB each
account_totals = (
    spark.read.parquet("/data/bank/transactions")
    .groupBy("account_id")
    .agg({"amount": "sum"})
)
account_totals.show()
# Spark UI shows coalesced partitions in the final plan

How does AQE handle data skew?

AQE detects oversized shuffle partitions using two thresholds: the partition must be larger than skewedPartitionFactor times the median size and larger than skewedPartitionThresholdInBytes in absolute terms. It splits each skewed partition into smaller sub-partitions and replicates the other join side so results stay correct.

PYTHON

# Configure skew handling
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", "256m")

# One corporate account has 10x more transactions than others
# AQE splits that partition and parallelizes the join
corp_join = (
    spark.read.parquet("/data/bank/transactions")
    .join(
        spark.read.parquet("/data/bank/accounts"),
        "account_id"
    )
)
corp_join.show()

How does dynamic join switching work?

When an earlier stage completes, AQE reads the actual size of each join side from the shuffle statistics. If one side is now below the broadcast threshold, AQE rewrites the remaining plan to use a broadcast hash join instead of the sort-merge join that Catalyst originally planned.

PYTHON

# Catalyst planned a sort-merge join because it estimated branches as large
# But after filtering, branches is only 2 MB
active_branches = branches.filter("status = 'ACTIVE'")

result = txns.join(active_branches, "branch_code")
result.explain()  # Static plan: SortMergeJoin

# With AQE enabled, the Spark UI final plan shows:
# BroadcastHashJoin (AQE switched it at runtime)

TEXT

== Static Plan ==
SortMergeJoin [branch_code], [branch_code]

== AQE Final Plan ==
BroadcastHashJoin [branch_code], [branch_code]
  (AQE detected post-filter size = 2 MB < threshold)

How does AQE complement Catalyst rather than replace it?

Catalyst still does all the up-front work: parsing, analysis, rule-based logical optimization like predicate pushdown and column pruning, and initial physical planning. AQE layers on top, correcting only the physical plan at runtime where static estimates were wrong.

DIAGRAM

Catalyst (before execution):
  Parse → Analyze → Optimize → Physical Plan (from estimates)

AQE (during execution):
  Stage completes → Read real stats → Rewrite remaining plan

PYTHON

# You can see both plans
result.explain(True)   # Shows Catalyst's static plan
# After execution, check Spark UI for AQE's final plan
# The two may differ if AQE made runtime adjustments

What are the configuration parameters?

AQE has several key configuration parameters that control its behavior at the National Bank of India's production cluster.

PYTHON

# Core AQE settings
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128m")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256m")
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", "10m")

| Parameter | Default | Purpose |
|---|---|---|
| adaptive.enabled | true (Spark 3.2+) | Master switch for AQE |
| advisoryPartitionSizeInBytes | 64 MB | Target size for coalesced partitions |
| skewedPartitionFactor | 5 | How many times median to flag skew |
| skewedPartitionThresholdInBytes | 256 MB | Minimum absolute size to flag skew |

What are the limitations of AQE?

AQE only helps where there are shuffles and real statistics to read. It does not optimize a job with no shuffle. It is not a substitute for good data layout, partitioning, or manual salting of extreme skew.

PYTHON

# AQE cannot help here: no shuffle involved
no_shuffle = spark.read.parquet("/data/bank/transactions").filter("amount > 50000")
no_shuffle.show()  # Pure filter scan, no AQE opportunity

# AQE helps here: shuffle from join
with_shuffle = (
    spark.read.parquet("/data/bank/transactions")
    .join(spark.read.parquet("/data/bank/accounts"), "account_id")
)
with_shuffle.show()  # AQE can coalesce, handle skew, switch joins

What should a senior engineer know?

AQE is enabled by default in Spark 3.2+ but was opt-in in Spark 3.0.
The authoritative view of what actually ran is the Spark UI final plan, not the static explain() output.
AQE's broadcast threshold may differ from the static autoBroadcastJoinThreshold.
For extreme skew where even AQE's split sub-partitions are still too large, manual salting is still needed.
AQE adds no overhead for jobs without shuffles: it simply does nothing.

PYTHON

# Always check the Spark UI for the final executed plan
# Static explain may show SortMergeJoin
# But AQE may have switched to BroadcastHashJoin at runtime
result.explain()  # Static plan (may be outdated)
# Spark UI > SQL tab > query > shows the actual AQE plan

Visual Overview

DIAGRAM

  Catalyst Static Plan
       │
       ▼
  ┌──────────────────────────────────────┐
  │  Stage 1: Scan + Filter              │
  │  (executes normally)                 │
  └──────────────┬───────────────────────┘
                 │  shuffle boundary
                 ▼
  ┌──────────────────────────────────────┐
  │  AQE Checkpoint                      │
  │  Read real partition sizes            │
  │                                      │
  │  ┌──────────┐  ┌──────────┐          │
  │  │ Coalesce │  │ Skew     │          │
  │  │ small    │  │ split    │          │
  │  │ parts    │  │ large    │          │
  │  └──────────┘  └──────────┘          │
  │           │                          │
  │  ┌────────────────────┐              │
  │  │ Switch join        │              │
  │  │ strategy if needed │              │
  │  └────────────────────┘              │
  │                                      │
  │  Rewrite remaining plan              │
  └──────────────┬───────────────────────┘
                 ▼
  ┌──────────────────────────────────────┐
  │  Stage 2: Execute with new plan      │
  │  (optimized by AQE)                  │
  └──────────────────────────────────────┘

Key Takeaway

Adaptive Query Execution is a Spark SQL feature that re-optimizes the physical plan during execution using real statistics gathered from completed stages, instead of relying only on the estimates made before the job runs. The problem it solves is that the Catalyst Optimizer has to commit to a physical plan up front using statistics that are often stale, missing, or wrong: it may not know the true size of a table after filtering, or that one key is far more common than others.

AQE fixes this by making the plan adaptive at shuffle boundaries, which is exactly where accurate statistics become available. It has three headline optimizations: it coalesces many small post-shuffle partitions into fewer right-sized ones, it detects oversized skewed partitions and splits them so a hot key no longer creates a straggler, and it dynamically switches join strategies when it discovers one side is actually small enough to broadcast. The key framing is that AQE complements Catalyst rather than replacing it: Catalyst still does parsing, analysis, logical optimization, and initial physical planning, and AQE corrects that physical plan at runtime where the static estimates were wrong.

At the National Bank of India, enabling AQE smoothed out precisely the problems our static plans could not anticipate. Month-end spikes in a single branch's transactions used to create skewed partitions and long straggler tasks, and AQE's skew splitting handled that automatically. Joins whose post-filter sizes we could not predict were silently converted to broadcast joins, removing shuffles we would otherwise pay for.

The one caution is that AQE only helps where there are shuffles and real statistics to use, so it does not optimize a shuffle-free job, and it is not a substitute for good data layout, sensible partitioning, or occasionally salting extreme skew by hand.

Remember one sentence: AQE corrects the physical plan at runtime using real shuffle statistics, complementing Catalyst's static estimates with coalescing, skew splitting, and dynamic join switching.

---