AQE, DPP, skew join: what Spark 3 really fixes

An upgrade from Spark 2 to Spark 3 is often sold with a headline gain. The number is reachable, but it does not come from the version: it comes from three specific mechanisms, each with its own triggering conditions. Knowing them is the difference between a reproducible gain and a gain observed once.

The problem AQE solves

A query optimiser plans from statistics estimated before execution. On real data those estimates are wrong — a filter on a correlated column, a chain of joins, and the predicted cardinality is off by an order of magnitude.

Adaptive Query Execution changes when the decision is made: Spark cuts the plan at shuffle boundaries, runs a stage, observes the real statistics of what it produced, then replans the rest. It is not a better fortune teller, it is a planner allowed to change its mind.

spark.sql.adaptive.enabled = true                    # on by default from Spark 3.2
spark.sql.adaptive.coalescePartitions.enabled = true
spark.sql.adaptive.skewJoin.enabled = true
spark.sql.adaptive.advisoryPartitionSizeInBytes = 64m

The three optimisations, one by one

Coalescing shuffle partitions. The historical spark.sql.shuffle.partitions defaults to 200, whatever the data size. On a small volume you get 200 tasks handling a few kilobytes each: scheduling costs more than the computation. AQE measures the real post-shuffle size and merges partitions towards the advisory size. It is the most consistent gain, and the least noticed, because it shows up as tasks that no longer exist.

Switching join strategy. A join planned as sort-merge — hence two shuffles — can become a broadcast hash join if the reduced side turns out small enough at runtime. This is common when a selective filter precedes the join: the static planner saw a multi-gigabyte table, execution produces a few dozen megabytes.

Splitting skewed partitions. This is the most visible one symptomatically. A join whose key has a dominant value concentrates rows into a single partition: 199 tasks finish in seconds, one runs for forty minutes, and the whole job waits. AQE detects the abnormally large partition and splits it, replicating the opposite side.

spark.sql.adaptive.skewJoin.skewedPartitionFactor = 5
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes = 256m

The condition is twofold: a partition is treated as skewed if it exceeds threshold and is more than factor times the median. A moderate but widespread imbalance therefore triggers nothing — and that is often the one that costs.

Dynamic Partition Pruning and its conditions

DPP is independent of AQE and answers a different problem: reading less. On a star schema you filter the dimension and join the fact table. Without DPP, Spark reads every partition of the fact table, then discards what does not join.

DPP builds a filter from the values actually retained on the dimension side and pushes it into the fact table scan. The gain is not marginal: the files are never read.

-- DPP fires
SELECT f.*
FROM   faits f
JOIN   dim d ON f.date_valeur = d.date_valeur   -- key = partitioning column of f
WHERE  d.trimestre = '2026T1'                    -- selective filter on the dimension

Three conditions must hold, and missing one is enough to cancel the benefit:

• the fact table is partitioned on the join column;
• the filter applies to the dimension, not to the fact table;
• the dimension side is broadcastable, or spark.sql.optimizer.dynamicPartitionPruning.useStats allows estimation.

The first is the one most often missing, for a structural reason: tables are frequently partitioned by load date while queries filter by value date. That is a modelling problem, not an engine problem — and repartitioning the fact table on the column actually queried produces a gain neither AQE nor DPP can deliver.

What AQE will never fix

The boundary deserves precision, because this is where migrations disappoint.

A massively degenerate join key. If a third of the rows carry the same value — typically an « unknown » code or a default — adaptive splitting mitigates but does not remove the imbalance. The fix is upstream: isolate those rows and handle them separately.

val nonRenseigne = lit("UNKNOWN")

val principal = faits.filter(col("id_tiers") =!= nonRenseigne)
                     .join(dim, Seq("id_tiers"))

val residuel  = faits.filter(col("id_tiers") === nonRenseigne)
                     .withColumn("libelle_tiers", lit(null: String))

val resultat = principal.unionByName(residuel)

Cardinality explosion. A join producing ten times its inputs stays expensive, however you distribute the partitions. The subject is the modelling.

An opaque UDF. A user function is a black box to the optimiser: no pushdown, no column pruning. Replacing a UDF with native functions is often worth more than all adaptive tuning combined.

Too many small files. AQE acts on shuffle partitions, not on the initial read. Ten thousand 200 KB files will produce ten thousand read tasks before AQE has any say.

Read a plan before touching a parameter

The method that avoids tuning at random: read the executed plan, not the planned one. With AQE the two differ, and that is the point.

val df = requete
df.explain("formatted")     // initial plan
df.count()                  // execution
// in the Spark UI, SQL tab: the final plan carries
// AdaptiveSparkPlan isFinalPlan=true, and the statistics actually observed

What to look for: a BroadcastHashJoin where you expected a SortMergeJoin, the presence of a skew-split node, and above all the number of bytes actually read at the file scans. That last figure tells you whether DPP fired — it is the only proof that counts.

The order I follow: measure what is read, fix the partitioning if too much is read, isolate degenerate values if one task drags, and only then adjust the adaptive parameters. In that order the gains compound. In the opposite order you spend a week tuning thresholds on a modelling problem.