Frame the Static-Plan Limit

The opening sentence of a strong recital names the limit before the interviewer has to dig for it: Catalyst produces a single physical plan from what it can estimate at compile time, and adaptive execution can revise that plan only at shuffle boundaries, using sizes measured after a stage finishes. Every later claim in the answer hangs off that frame, so it pays to rehearse it on a concrete query.

The query on the table

SELECT
  u.country,
  count(*) AS purchases
FROM events e
JOIN users u
  ON e.user_id = u.user_id
WHERE e.event_type = 'purchase'
GROUP BY u.country

Walk the estimate the way the planner does. With the cost-based optimizer off, which is the default (spark.sql.cbo.enabled is false), a filter's output is estimated at the size of its input. The planner sees 2GB of events joined to 400MB of users, neither below the 10MB broadcast threshold, and picks a sort-merge join. That plan shuffles both sides by user_id and then shuffles again by country for the aggregation: 3 exchanges and 4 stages.

Statistics move the guess, and the guess stays a guess

spark.sql("ANALYZE TABLE events COMPUTE STATISTICS FOR COLUMNS event_type, user_id")
spark.conf.set("spark.sql.cbo.enabled", "true")

With column statistics and no histogram, the cost model estimates an equality filter as 1 divided by the column's distinct count. Events has 25 event types, so the estimate is 4% of 2GB, roughly 80MB. That shrinks the error from about 250 times to 10 times, and 80MB is still 8 times over the 10MB broadcast threshold. Purchases are 0.4% of events because event types are far from uniform: page views make up most rows and purchases are rare. An estimate that assumes every value is equally common has no way to see that. Histograms (spark.sql.statistics.histogram.enabled, off by default) would, at the price of a heavier ANALYZE that must be rerun as the data shifts.

What adaptive execution changes, and when

Adaptive query execution, on by default since Spark 3.2, re-plans at each shuffle boundary from the sizes the finished map tasks report. Here, once the events map stage completes, AQE sees around 8MB of shuffle output, under spark.sql.adaptive.autoBroadcastJoinThreshold (which takes the 10MB static value when unset), and rewrites the sort-merge join into a broadcast hash join. Notice what it can't take back. The users map stage was submitted alongside the events stage and has usually already written its 400MB of shuffle files. AQE saves the sort and reads those files through a local shuffle reader instead of fetching them across the network. The exchange by country stays.

The frame, said out loud

A frame with no limit

• Spark builds a logical plan, optimizes it and generates a physical plan.
• Catalyst applies rules like predicate pushdown.
• AQE then optimizes the plan at runtime.

A frame that names the limit

• The driver commits to a physical plan from estimates before reading data.
• Without column statistics a filter is assumed to keep its whole input, so a selective filter loses its broadcast.
• AQE revises the plan only between stages, from shuffle sizes, never inside a running stage.

The 0.4% share and the 25 event types are this lesson's example. On your own tables, explain(mode="cost") prints the planner's estimates for the same comparison.

About this lesson

This section is part of the The Architecture Recital lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.