What does adaptive query execution change in a Spark query, and how do you see it in the plan?

A query plan is a guess made before any data is read. Spark estimates how big each table and each intermediate result will be, and picks joins and partition counts from those estimates. Adaptive query execution, AQE, lets Spark revise the guess. The query runs in pieces called query stages, split at each shuffleSpark ShuffleThe step that moves rows between executors so rows with the same key end up together, for a join, group by or repartition; it splits a job into stages.. When a stage finishes, Spark knows the real size of every partition it wrote, and re-plans the rest of the query with those numbers. It has been on by default since Spark 3.2.0 (spark.sql.adaptive.enabled).

What it changes

1. Merges small partitions. A shuffle into 200 partitions of a few megabytes leaves 200 tiny tasks. AQE combines neighbouring partitions. By default it favours parallelism: it ignores the 64 MB target and only keeps each merged partition over 1 MB. Spark's docs suggest setting spark.sql.adaptive.coalescePartitions.parallelismFirst to false on a busy cluster, so partitions grow toward 64 MB and fewer tasks run.
2. Splits skewed partitions. In a sort-merge joinSort-Merge JoinSpark's default join for 2 large tables: it shuffles both by the join key, sorts each partition, then walks the 2 sorted sides together to find matching rows., a partition over 5 times the median and over 256 MB counts as skewed. AQE splits it into pieces near 64 MB and copies the matching rows of the other side to each piece. This is the main defence against data skewShuffle and Data SkewA shuffle moves rows between machines so rows with the same key meet, for a join or group by; skew is when one key has far more rows, so one task does most of the work..
3. Switches the join. If one side of a planned sort-merge join turns out smaller than the broadcast threshold (10 MB by default), AQE turns it into a broadcast hash join. The shuffle already ran, so it is not as good as planning a broadcast from the start, but it skips both sorts and reads the shuffle files locally.

How to see it in a plan

explain() before the query runs prints AdaptiveSparkPlan isFinalPlan=false: the plan Spark would start with. The real plan exists only after a run. In the SQL / DataFrame tab of the Spark UI, or in explain() on a DataFrame that has already run, it shows isFinalPlan=true with a Final Plan and an Initial Plan.

We ran this on Spark 4.2. A join of 2 filtered copies of an orders table was planned as a SortMergeJoin, because nothing told Spark the filter would leave 103 rows. After the run, the final plan read:

BroadcastHashJoin [order_id], [order_id], Inner, BuildLeft
:- BroadcastExchange
:  +- AQEShuffleRead local
+- AQEShuffleRead local

Every AQEShuffleRead box is AQE's work. Its label says which kind: coalesced for merged partitions, local for a switched join, skewed for a split. A join that was split shows as SortMergeJoin(skew=true). Click the box in the SQL tab to see the number of partitions before and after, and the count of skewed partitions and splits.

When it is not enough

AQE works from what a shuffle wrote, so it cannot fix the first stage that reads the files, and it only splits skew inside sort-merge joins (and rebalance hints), not inside a groupBy. It also needs enough partitions to merge: with spark.sql.shuffle.partitions far too low, every partition is already too big and tasks spillSpark SpillWhen a task's sort, join or aggregation no longer fits in execution memory, Spark writes part of it to local disk and reads it back later, so the task finishes, only slower.. Our Spark UI guide shows the same skewed join before and after AQE's skew handling, and the shuffle partition calculator works out how AQE will merge and split a given shuffle.