Stale Statistics and Bad Joins: How Spark's Adaptive Query Execution Fixes Plans Mid-Flight

Your Spark job is slow. The query plan shows a sort-merge join between a large fact table and a dimension table that you know is heavily filtered. The optimizer did not know the filter would shrink the table, so it planned a full shuffle. Meanwhile, the next stage spawns hundreds of tasks that each process a few kilobytes. The cluster looks busy, but most executors are idle.

Adaptive Query Execution (AQE), introduced in Spark 3.0, tackles this by re-optimizing the physical plan between stages using actual runtime statistics instead of compile-time guesses. The result is often fewer shuffle stages, smaller joins, and better cluster utilization without rewriting your SQL.

Why Static Planning Misses the Mark

Spark's Catalyst optimizer builds a physical plan before execution begins. For joins, it estimates the size of each side using table statistics stored in the metastore. When those statistics are missing, stale, or simply absent—common after a partial load or a complex view—Spark falls back to conservative defaults. A dimension table that will shrink to a few megabytes after a WHERE clause may look enormous to the planner, forcing an expensive sort-merge join and a full shuffle of both sides.

The same problem produces tiny shuffle partitions. If the planner guesses a large output but the actual data is small, Spark creates the default 200 partitions and leaves most tasks with almost nothing to do.

What AQE Changes at Runtime

AQE inserts an AdaptiveSparkPlan wrapper around the physical plan. After each shuffle stage completes, it can rewrite the remaining plan based on real metrics: actual row counts, shuffle bytes, and partition sizes. Three optimizations matter most in practice:

• Coalescing small shuffle partitions. If many partitions fall below the target size, AQE merges them into fewer tasks, reducing scheduling overhead.
• Switching sort-merge joins to broadcast hash joins. When one side proves small enough at runtime, AQE replaces the sort-merge join with a broadcast join, eliminating the large-side shuffle.
• Splitting skewed partitions. If one partition holds far more data than the others, AQE can split it so the work spreads across more executors.

Worked Example: Catching a Small Build Side

Consider a nightly report joining a fact table with a filtered dimension. You can reproduce the setup in spark-shell, a Databricks notebook, or any Spark session where you have permission to change session configuration.

// spark-shell or notebook session
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "64m")

val result = spark.table("sales_fact")
  .join(
    spark.table("product_dim").filter($"category" === "electronics"),
    Seq("product_id")
  )
  .groupBy("region")
  .count()

result.explain(mode = "formatted")

With AQE disabled, the formatted plan may show a SortMergeJoin because the planner cannot predict the filter's selectivity. After enabling AQE, the plan is wrapped in an AdaptiveSparkPlan. Once the filter stage finishes, AQE reads the actual output size of product_dim. If that size is below spark.sql.autoBroadcastJoinThreshold, the optimizer rewrites the join to a BroadcastHashJoin before the fact table is shuffled.

To confirm the switch happened, open the Spark UI SQL tab and select the completed query. In the AQE section, look for evidence of partition coalescing or a changed join strategy. If the plan still shows a sort-merge join, check whether the runtime size exceeded the broadcast threshold or whether a connector failed to report accurate statistics.

Trade-offs and Limitations

AQE is not free. Re-optimizing between stages adds planning latency and CPU overhead. For sub-second queries, that overhead can outweigh the savings. Very short, interactive workloads may perform better with AQE disabled.

Configuration names and defaults vary across Spark versions. Before relying on a specific knob, confirm it exists in your deployment with spark.conf.get("spark.sql.adaptive.enabled") and review the documentation for your Spark release. A setting that worked in one minor version may be renamed or deprecated in another.

AQE also does not fix bad source layout. Severely skewed input files or inadequate bucketing can still create imbalanced stages that runtime re-planning only partially mitigates. In those cases, upstream repartitioning or bucketing remains necessary. Similarly, custom UDFs or third-party connectors that do not expose accurate size estimates can limit AQE's effectiveness.

Practical Next Steps

1. Enable AQE at the session or cluster level: spark.sql.adaptive.enabled=true.
2. Set a sensible advisory partition size, for example spark.sql.adaptive.advisoryPartitionSizeInBytes=64m, and adjust based on your executor memory.
3. Run EXPLAIN FORMATTED on a representative query with AQE off and on. Look for AdaptiveSparkPlan and changed operators.
4. Inspect the Spark UI SQL tab for coalescing counts and join strategy changes.
5. Benchmark a skewed join workload with the setting toggled, comparing total task time and shuffle spill metrics.

If AQE degrades a specific workload, revert by setting spark.sql.adaptive.enabled=false in the same session configuration. Because this is a session-level change, it does not alter stored data, but it will apply to all subsequent queries in that session until reset.