What is adaptive query execution (AQE)?

Summary

• Adaptive query execution (AQE) is a Spark SQL optimization that re-plans a query while it runs, using statistics collected at runtime instead of only the estimates available before execution.
• AQE has been enabled by default since Apache Spark 3.2.0 and is controlled by the umbrella setting spark.sql.adaptive.enabled.
• Its main optimizations are coalescing post-shuffle partitions, converting sort-merge joins to broadcast hash joins or shuffled hash joins, and splitting skewed partitions in shuffled joins.
• An AQE plan has an AdaptiveSparkPlan root node whose isFinalPlan flag turns true once execution completes. SQL EXPLAIN shows only the initial plan.
• AQE doesn't reorder joins, and a join it converts to a broadcast at runtime isn't as efficient as one planned as a broadcast from the start.

What is adaptive query execution (AQE)?

Adaptive query execution is the part of Spark SQL that changes a query plan while the query runs. The Spark docs define it as "an optimization technique in Spark SQL that makes use of the runtime statistics to choose the most efficient query execution plan", and it has been on by default since Apache Spark 3.2.0. Static planning has to trust table statistics that may be stale, missing, or wrong after several joins. AQE waits for real numbers from finished stages and re-optimizes the rest. Leave it on: most of its features need no tuning.

How AQE works

AQE cuts the plan at shuffle and broadcast exchanges, and each piece is a query stage. When a stage finishes, Spark knows the actual size of its output, and AQE re-plans the remaining stages with those runtime statistics. That's when it can merge tiny partitions, switch a join strategy, or split a skewed partition.
Runtime statistics show up in the plan as Statistics(..., isRuntime=true), visible in the SQL UI's "Details" section while a query runs. Databricks' AQE documentation adds that it applies to queries containing at least one exchange (usually a join, aggregate or window) or a subquery, and that not every such query ends up with a different plan.

What AQE optimizes

Optimization	What it does	Key setting (default)
Coalescing post-shuffle partitions	Merges contiguous small shuffle partitions, so you don't have to pick an exact shuffle partition count	spark.sql.adaptive.coalescePartitions.enabled (true), target spark.sql.adaptive.advisoryPartitionSizeInBytes (64 MB)
Sort-merge join to broadcast hash join	Switches when a join side's runtime size falls under the adaptive broadcast threshold, and can read shuffle files locally	spark.sql.adaptive.autoBroadcastJoinThreshold (same as spark.sql.autoBroadcastJoinThreshold)
Sort-merge join to shuffled hash join	Switches when every post-shuffle partition is under a per-partition threshold	spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold (0)
Skew join handling	Splits, and replicates if needed, skewed partitions in shuffled joins (sort-merge and shuffled hash) into roughly even tasks	spark.sql.adaptive.skewJoin.enabled (true)
Skew in rebalances	Splits skewed partitions created by RebalancePartitions	spark.sql.adaptive.optimizeSkewsInRebalancePartitions.enabled (true)

A few details matter in practice:

• Coalescing targets. By default spark.sql.adaptive.coalescePartitions.parallelismFirst is true, so Spark ignores the 64 MB target and only respects the 1MB minimum partition size, to maximize parallelism. The Spark docs advise turning it off (false) on a busy cluster.
• Skew detection. A partition counts as skewed when it's larger than skewedPartitionFactor (5.0) times the median partition size and also larger than skewedPartitionThresholdInBytes (256MB). Spark 4.2.0's release notes also list "Optimize BroadcastHashJoin skew in OptimizeSkewedJoin" (SPARK-44065).
• Shuffled hash join. The conversion only kicks in when you set the threshold at or above spark.sql.adaptive.advisoryPartitionSizeInBytes.
• Customization. spark.sql.adaptive.optimizer.excludedRules disables named AQE rules, and spark.sql.adaptive.customCostEvaluatorClass swaps in your own cost evaluator.

See AQE in a query plan

df = (spark.table("sales")
      .join(spark.table("stores"), "store_id")
      .groupBy("region").count())

df.explain()    # AdaptiveSparkPlan isFinalPlan=false: the initial plan
df.collect()
df.explain()    # AdaptiveSparkPlan isFinalPlan=true: the plan AQE actually ran

spark.conf.set("spark.sql.adaptive.coalescePartitions.parallelismFirst", "false")

Under each AdaptiveSparkPlan node you'll find the initial plan and, once execution completes, the final plan. Comparing them shows what changed, for example a SortMergeJoin that became a BroadcastHashJoin. Spark 4.2.0's Web UI also shows the initial and final plans of AQE queries side by side (SPARK-55877).

What AQE doesn't do

• It doesn't reorder joins. Databricks' AQE FAQ states that dynamic join reordering isn't part of AQE.
• A runtime broadcast is a fallback. The Spark docs rank a mid-query conversion below a broadcast hash join planned up front, though still ahead of continuing the sort-merge join. The conversion avoids sorting both sides, and with spark.sql.adaptive.localShuffleReader.enabled (true) it reads shuffle files locally. If you know a dimension is small, a broadcast hint still helps.

How Databricks approaches AQE

Databricks turns AQE on by default and documents a few settings of its own:

• spark.databricks.optimizer.adaptive.enabled (default true) turns AQE on or off.
• The runtime switch to a broadcast hash join uses spark.databricks.adaptive.autoBroadcastJoinThreshold, default 30MB.
• Set spark.sql.shuffle.partitions to auto for auto-optimized shuffle: the partition count is then worked out per query, based on its plan and input size.
• AQE on Databricks also detects and propagates empty relations (spark.databricks.adaptive.emptyRelationPropagation.enabled).

The Spark UI's plan diagram updates as stages complete, so you can watch re-optimization happen. See Databricks' AQE reference.

FAQs

Is AQE on by default?

Yes. Apache Spark has enabled it by default since 3.2.0, through spark.sql.adaptive.enabled.

Do I still need to tune spark.sql.shuffle.partitions?

Less often. With coalescing on, set a large enough initial partition count through spark.sql.adaptive.coalescePartitions.initialPartitionNum, which defaults to spark.sql.shuffle.partitions, and let Spark pick the final number at runtime.

Why doesn't SQL EXPLAIN show AQE's changes?

EXPLAIN doesn't execute the query, so its plan is always the initial plan. Run the query, then check the Spark UI or DataFrame.explain().

Does AQE work in streaming queries?

Spark 4.1.0's release notes list "Support AQE in stateless streaming workloads" (SPARK-53941).