In Spark AQE, when does spark.sql.adaptive.skewJoin.enabled split a shuffle partition?

## Where the rule sits Adaptive Query Execution cuts the physical plan at shuffle boundaries. When a query stage completes, its map-output statistics give the exact byte size of every shuffle partition, and AQE re-optimises the rest of the plan with those numbers in hand. The skew-join rule is one of those re-optimisations. It runs after the join's inputs have been shuffled and before the join's reducer tasks are scheduled. Two switches gate it: `spark.sql.adaptive.enabled` (the umbrella, `true` by default since Spark 3.2) and `spark.sql.adaptive.skewJoin.enabled` (`true` by default since 3.0). Turning off the umbrella turns off everything below it — a common cause of "AQE didn't help" on clusters where someone disabled adaptive execution years ago and never revisited it. ## The two thresholds, and why there are two A partition is considered skewed only if **both** are true: 1. its size is greater than `spark.sql.adaptive.skewJoin.skewedPartitionFactor` × the median partition size — default factor `5.0`; 2. its size is greater than `spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes` — default `256MB`. The factor alone would fire constantly on healthy stages: if the median partition is 2 MB, a 12 MB partition is six times the median and completely harmless. The absolute floor stops Spark from splitting partitions that were never going to be slow. The trade-off is the mirror image: a stage whose partitions are all under 256 MB gets no skew handling even when one is twenty times its neighbours, which is a real reason a skewed job sometimes goes untouched. The documentation also advises keeping the threshold above `spark.sql.adaptive.advisoryPartitionSizeInBytes`, so the rule does not split into pieces at the same scale as the thing it is judging. ## Splitting and replicating The rule does **not** re-hash the key. A shuffle partition is physically a set of blocks, one per map task. Splitting means handing subsets of those blocks to different reducer tasks, targeting roughly the advisory partition size per piece. Since each piece now contains only some of the hot key's rows, the matching partition on the other side of the join must be given to *every* piece — that is the replication. For a hot key joined against a handful of dimension rows this is cheap; the replicated side is by definition the non-skewed one. The result is that a partition that would have been one 9 GB task becomes, say, thirty tasks of a few hundred megabytes, each joining its slice of the fact rows against the same replicated dimension rows. Output is identical, because every original pair of matching rows still meets exactly once: each fact row lives in exactly one piece, and the dimension rows are present in all of them. ## Where it applies, and where it does not - **Sort-merge joins** are the documented target. The rule needs post-shuffle partitions on both sides to measure and replicate. - **Broadcast hash joins** are untouched — there is no shuffle on the large side, and no concentration to fix. - **Aggregations are not covered.** A skewed `groupBy` sends every row of the hot group to one reducer task, and no splitting is possible without changing the aggregation itself, since the group's final value depends on all of its rows. That is what two-phase salted aggregation is for. - **Map-side skew is not covered.** If one input *file* or one Kafka partition is enormous, the imbalance exists before any shuffle statistic is produced. - The rule only fires when the plan shape lets it. `spark.sql.adaptive.forceOptimizeSkewedJoin` (default `false`) forces the optimisation even when it introduces an extra shuffle; leave it off unless you have measured that the extra exchange costs less than the straggler. ## Related shuffle-side setting AQE's decision depends on the map-output statistics being accurate. To keep `MapStatus` small, Spark compresses per-block sizes above a certain number of blocks, recording exact sizes only for blocks above `spark.shuffle.accurateBlockThreshold`. `spark.shuffle.accurateBlockSkewedFactor` (default `-1.0`, i.e. off) additionally records a block accurately when it exceeds that factor times the median block size; the documentation suggests setting it to the same value as `spark.sql.adaptive.skewJoin.skewedPartitionFactor`. On a job where you know skew exists and AQE seems blind to it, this is worth checking. ## What good tuning looks like Mostly, leave the defaults alone and verify from the Spark UI that the straggler disappeared. When it did not: check that adaptive execution is on at all; check whether the operator is a join or an aggregation; check whether the skewed partition actually clears 256 MB, and lower `skewedPartitionThresholdInBytes` if the whole stage runs at a smaller scale; and if the hot key is enormous — a large fraction of a multi-terabyte fact table — accept that splitting after the shuffle still means the shuffle wrote one giant partition, and change the data distribution with salting or a broadcast instead.