In Spark AQE, how can a sort-merge join become a broadcast hash join mid-query?

## Query stages and re-planning Without adaptive execution, Spark chooses every physical operator before the first task runs, using estimated row counts and sizes derived from data-source metadata, catalog statistics and the optimizer's selectivity guesses. Those estimates decay fast: after a filter on an un-analyzed column, after a join whose output cardinality nobody knows, after a UDF, the number the planner is holding may be off by orders of magnitude. AQE turns the plan into a sequence of *query stages* separated by shuffles. It executes a stage, reads its real map-output statistics, re-optimises the remainder of the plan with those exact numbers, and continues. The join-strategy switch is the most visible consequence. ## The conversion When the runtime statistics of any side of a sort-merge join come out below the adaptive broadcast threshold, AQE rewrites that join into a broadcast hash join for the stages that have not yet run. `spark.sql.adaptive.autoBroadcastJoinThreshold` controls it and defaults to the same value as `spark.sql.autoBroadcastJoinThreshold` (10 MB); setting it to `-1` disables adaptive broadcasting specifically, without touching the plan-time threshold. The documentation is explicit that this is not as good as planning a broadcast hash join in the first place — both sides have already paid the shuffle write — but better than continuing with the sort-merge join, because the sort of both sides and the merge are avoided. ## The local shuffle reader Once the join no longer needs data partitioned by key, the reducer tasks do not have to fetch a specific bucket from every executor. With `spark.sql.adaptive.localShuffleReader.enabled` true (the default), Spark reads the shuffle output that already sits on the local executor, cutting the network fetch out of the picture. This is a small optimisation that only makes sense in exactly this situation: the shuffle exists, but its partitioning has become irrelevant. ## A second conversion AQE can also convert a sort-merge join into a **shuffled hash join** — building a hash map per partition instead of sorting both sides — when all post-shuffle partitions are below `spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold`. That config defaults to `0`, so this conversion is effectively off until you set a real per-partition size. It is worth knowing about mostly as evidence that AQE's join re-planning is not limited to broadcasting. ## Why this matters in practice The conversion is the reason a Spark 3 job can be dramatically faster than the same code on Spark 2 with no changes: the pathological case of a sort-merge join between a huge table and a tiny filtered one, which the estimator failed to spot, now fixes itself after the shuffle. It is also the reason "my plan changed halfway through the query" is a normal observation rather than a bug report. The practical consequence for tuning: if a broadcast is what you want, getting it at *plan* time is still better. Give the planner real statistics with `ANALYZE TABLE`, or state your intent with a `broadcast()` hint, and you skip the shuffle entirely instead of paying for it and then discarding its partitioning. ## Seeing it happen An adaptive plan is rooted at an `AdaptiveSparkPlan` node carrying an `isFinalPlan` flag. Before execution completes it reads `isFinalPlan=false` and shows the initial plan; the SQL tab in the Spark UI updates the plan as stages complete, and the final plan — the one that actually ran — is what you should compare against your expectations. Reading `explain()` output on an AQE query and concluding that a sort-merge join ran, when in fact the final plan broadcast, is a routine mistake.