Adaptive query execution in Apache Spark: what it re-plans, and what it leaves alone

• Architecture: where the re-planning happens
• How to inspect it, and the mistake to avoid
• Change 1: coalescing partitions
• Change 2: converting a join to a broadcast
• Change 3: reading the shuffle locally
• Change 4: splitting a skewed partition
• What AQE does not do
• Production notes
• Frequently asked questions
• Conclusion
• References
• Trademarks

TL;DR

• Everything Catalyst decides before a query starts comes from estimates. Adaptive query execution re-plans after each shuffle stage finishes, using measurements, and it is on by default.
• It does four things: coalesces small partitions, splits skewed ones, converts a sort-merge join to a broadcast join, and reads a shuffle locally when the join no longer needs it distributed.
• explain() shows the plan before any of it. isFinalPlan=false means nothing has been re-planned yet, which is why plans read from explain() and plans that actually ran often disagree.
• To inspect it you must execute the plan you are holding and re-read the same QueryExecution. df.count() builds a different query, so checking that way inspects a plan that never ran.
• It re-plans physical choices only. The logical rewrites, pushdown, pruning, constant folding, are finished before execution begins and are never revisited.

Every planning decision Spark makes before a query starts is made from estimates, and the estimates are frequently poor. A filter whose selectivity is unknown, a join whose output size nobody can predict, a shuffle whose partitions turn out to be 4 KB each. Adaptive query execution is the acknowledgement that after the first shuffle completes, Spark knows things it could only guess at before, and should be allowed to change its mind.

This post takes each thing it changes and shows the plan before and after, the threshold that gates it, and how to tell from the plan that it fired. It also covers the part most descriptions leave out: what AQE does not touch, which matters because it explains why some bad plans survive it.

Architecture: where the re-planning happens

A shuffle is a natural barrier. The stage above it cannot start until the stage below it has written its output, and at that moment the sizes of every shuffle partition are known exactly. AQE inserts itself at that barrier.

flowchart TB
  A["optimized logical plan"] --> B["physical plan, from estimates"]
  B --> C["AdaptiveSparkPlan<br/>isFinalPlan=false"]
  C --> D["run the next stage"]
  D --> M["<b>measured</b> shuffle<br/>partition sizes"]
  M --> R{"re-optimize<br/>what remains"}
  R -->|"many tiny partitions"| R1["coalesce"]
  R -->|"one huge partition"| R2["split the skew"]
  R -->|"a side is small after all"| R3["convert to broadcast"]
  R -->|"no shuffle needed now"| R4["read the shuffle locally"]
  R1 --> D
  R2 --> D
  R3 --> D
  R4 --> D
  R -->|"nothing left"| Z["isFinalPlan=true"]

The loop matters. AQE is not one re-plan at the start; it runs again at every shuffle boundary, so a query with three shuffles gets three opportunities to change the remaining plan.

The defaults, read from a session:

parallelismFirst being true is worth knowing, because it means the 64 MB advisory size is not really the target on a small cluster: Spark would rather produce at least as many partitions as you have cores than produce fewer, larger ones.

How to inspect it, and the mistake to avoid

This tripped me up, so it goes before the examples. explain() and the plan object both show the pre-execution plan. To see what AQE did, you must run the plan you are holding and then look at the same QueryExecution again:

df = spark.sql("SELECT city_id, count(*) c FROM fact GROUP BY city_id")
qe = df._jdf.queryExecution()

print(qe.executedPlan().toString().splitlines()[0])   # before
df.collect()                                          # execute THIS plan
print(qe.executedPlan().toString().splitlines()[0])   # after

AdaptiveSparkPlan isFinalPlan=false
AdaptiveSparkPlan isFinalPlan=true

Do not use df.count() for this. count() builds a different query, an aggregate over your plan, so it executes and finalises something else while the plan you are inspecting stays at isFinalPlan=false. I spent a while concluding AQE was not firing before noticing that.

isFinalPlan=true is the signal that the plan you are now reading is the one that ran. Everything below is read that way.

Change 1: coalescing partitions

The most common win, and the one that quietly fixes the spark.sql.shuffle.partitions problem everyone used to tune by hand.

With the default 200 shuffle partitions and only 400 groups to produce, the planned exchange asks for 200 partitions:

Exchange hashpartitioning(city_id#3L, 200), ENSURE_REQUIREMENTS, [plan_id=25]

and the final plan reads them back merged:

AdaptiveSparkPlan isFinalPlan=true
+- ...
   +- AQEShuffleRead coalesced
      +- ShuffleQueryStage 0

AQEShuffleRead coalesced is the marker. The number of reduce tasks is now derived from the measured data rather than from a config you guessed at, which is why the old advice to tune shuffle.partitions per job has largely evaporated: set it high enough and let AQE bring it down.

What it cannot do is go the other way on a single partition. Coalescing merges neighbours; it never splits one, except through the skew path below.

Change 2: converting a join to a broadcast

The most dramatic change, because the operator itself is different after execution. Plan-time broadcasting is disabled here and runtime broadcasting allowed, which is the arrangement that isolates AQE’s contribution:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")        # plan time: never
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", "10MB")  # run time: allowed

q = spark.sql("SELECT count(*) FROM fact f JOIN dim_city c ON f.city_id = c.city_id")
qe = q._jdf.queryExecution()
planned = qe.executedPlan().toString()
q.collect()
final = qe.executedPlan().toString()

planned  : SortMergeJoin [city_id#3L], [city_id#10L], Inner
executed : *(3) BroadcastHashJoin [city_id#3L], [city_id#10L], Inner, BuildRight, false

A sort-merge join became a broadcast hash join once the build side’s real size was known. Note that these are two separate thresholds: the plan-time autoBroadcastJoinThreshold and the runtime adaptive.autoBroadcastJoinThreshold. When the latter is unset it falls back to the former, so the usual trick of setting the plan-time threshold to -1 to stop broadcasting also stops the runtime conversion unless you set the adaptive one too.

Change 3: reading the shuffle locally

This one follows from the last and is easy to miss. Once a join has become a broadcast join, its inputs no longer need to be partitioned by the join key, so fetching shuffle blocks across the network is pointless. AQE switches those reads to local:

+- *(3) BroadcastHashJoin [city_id#3L], [city_id#9L], Inner, BuildRight, false
   :- AQEShuffleRead local
   +- AQEShuffleRead local

AQEShuffleRead local means each task reads the shuffle blocks already on its own executor rather than fetching from peers. It is controlled by spark.sql.adaptive.localShuffleReader.enabled, default true, and it is a consequence of change 2 rather than something you trigger directly.

Change 4: splitting a skewed partition

When one shuffle partition is far larger than the others, AQE divides it into several and replicates the matching rows from the other side so each piece can be joined independently.

flowchart TB
  subgraph B["before: one task does most of the work"]
    P1["partition 3<br/>329,000 rows"] --> T1["one long task"]
  end
  subgraph A["after: the partition is split"]
    S1["3a"] --> U1["task"]
    S2["3b"] --> U2["task"]
    S3["3c"] --> U3["task"]
  end
  B --> A

# skewJoin.enabled = true
*(5) SortMergeJoin(skew=true) [city_id#1L], [city_id#6L], Inner
  +- AQEShuffleRead coalesced and skewed

# skewJoin.enabled = false
*(5) SortMergeJoin [city_id#1L], [city_id#6L], Inner
  +- AQEShuffleRead coalesced

(skew=true) on the join and and skewed on the shuffle read are the markers.

The threshold is the thing to understand, because it is the usual reason nothing happens. The rule takes the larger of two numbers as its skew threshold:

  def getSkewThreshold(medianSize: Long): Long = {
    conf.getConf(SQLConf.SKEW_JOIN_SKEWED_PARTITION_THRESHOLD).max(
      (medianSize * conf.getConf(SQLConf.SKEW_JOIN_SKEWED_PARTITION_FACTOR)).toLong)
  }

so a partition must beat both 5.0 times the median and 256MB. Taking the maximum makes the byte floor absolute: on data where no partition reaches 256 MB this never fires, however lopsided the key distribution is. That limitation, and the fixes that do not depend on it, are the subject of data skew in Apache Spark.

What AQE does not do

This is the part that matters for expectations, and it is short.

It does not revisit the logical plan. Predicate pushdown, column pruning, constant folding and the rest run once, before execution, and their results are fixed. If a predicate failed to push into your scan, AQE will not rescue it after the first stage. The logical phase is the subject of inside Spark’s Catalyst optimizer.

It does not re-plan a stage that is already running. Re-planning happens at shuffle boundaries, so a query with no shuffle gets no adaptive treatment at all, and the first stage is always executed as planned.

It does not fix a bad scan. If the plan reads five columns because pruning did not apply, or scans every partition because a partition filter was not recognised, that work happens in the first stage on the estimated plan.

It does not choose a different join order. Join reordering is a logical rewrite. AQE changes the join algorithm, not which tables are joined first.

It does not make statistics unnecessary. It removes the need for accurate estimates after the first shuffle. Everything before that, including the first join’s strategy and every scan decision, still runs on whatever the catalog knows.

Production notes

• Leave it on. It is on by default, and the failure mode of turning it off is a shuffle.partitions value that is wrong for every query but one.
• Set spark.sql.shuffle.partitions high rather than precisely. Coalescing brings it down from too many; nothing brings it up from too few.
• Set both broadcast thresholds deliberately. adaptive.autoBroadcastJoinThreshold falls back to the plan-time one, so -1 to disable plan-time broadcasting also disables the runtime conversion.
• Trust the SQL tab over explain(). For anything adaptive, explain() is a statement of intent. The UI shows the final plan with real row counts per operator.
• Check the markers before believing a claim. isFinalPlan=true for “this is what ran”, AQEShuffleRead coalesced for coalescing, AQEShuffleRead local for local reads, (skew=true) for a skew split.
• Do not expect it to cover skew. The 256 MB floor means it often does not, and it never helps a skewed aggregation, only a skewed join.
• Fix the logical plan separately. Pushdown and pruning are decided before AQE exists, so ReadSchema and PushedFilters on the scan are still yours to get right.

Frequently asked questions

Why does my explain() output not match what the UI shows? Because explain() prints the plan before execution, where AdaptiveSparkPlan isFinalPlan=false. The UI shows the plan after AQE has finished re-planning. They are both correct about different moments.

I set spark.sql.shuffle.partitions to 2000 and nothing got slower. Why? Coalescing. The exchange still asks for 2000 partitions, and AQE merges them back into a number derived from the measured data, so the config now sets an upper bound rather than the actual parallelism.

Why did my join not convert to a broadcast at runtime? Most often because autoBroadcastJoinThreshold is -1 and adaptive.autoBroadcastJoinThreshold was never set, so the runtime threshold inherited the disabled value. Otherwise, the join type may forbid broadcasting that side, which no amount of measurement changes.

Does AQE help a query with no joins and no aggregations? No. With no shuffle there is no boundary at which to re-plan, so a scan-and-filter query executes exactly as planned.

Is isFinalPlan=true a guarantee that AQE changed something? No. It only means re-planning has finished. The plan may be identical to the planned one, which is common and fine. Look for the specific markers to see whether anything was altered.

Should I still compute statistics if AQE is on? Yes. AQE uses measurements only after a shuffle. The first stage’s scan decisions, the first join’s strategy and any logical rewrite that depends on size all happen beforehand, on catalog statistics.

Conclusion

Adaptive query execution is best understood as a correction layer with a narrow and well-defined scope. It fixes the decisions that depend on sizes Spark could not have known, at the only moment it could know them, which is when a shuffle has just finished writing.

That scope is the useful thing to hold onto. Four changes, all physical: partition counts, skew splits, join algorithm, fetch locality. Everything logical is settled before the query starts and stays settled. So AQE removes an entire category of tuning, the one that used to be about guessing a good shuffle.partitions, and removes none of the work of getting the logical plan right.

The practical habit that follows is small. Read plans after execution rather than before, from the same QueryExecution or from the SQL tab, and look for the four markers. A plan printed from explain() is a plan of what Spark intended, and on any query with a shuffle that is not the same document as what it did.

References

• SQL performance tuning for the adaptive-execution configuration reference and defaults
• AdaptiveSparkPlanExec.scala, the operator that runs the re-planning loop at each shuffle boundary
• OptimizeSkewedJoin.scala for the skew threshold quoted above
• Data skew in Apache Spark for the fixes that work below AQE’s skew threshold
• Inside Spark’s Catalyst optimizer for the logical phase that AQE never revisits

Trademarks

Apache Spark, Apache Hudi, Apache Iceberg, Apache Parquet and Apache are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries.