Catalyst, code generation & AQE

From DataFrame to executed plan

A DataFrame is a description of work. Its QueryExecution turns that description into something that runs, in a fixed sequence, and explain("extended") prints each stage:

1. Parsed logical plan — the tree as written, still unresolved. For the example below it begins 'Aggregate ['category], ['category, sum('gross) AS total#12]; the leading quote marks names not yet resolved.
2. Analyzed logical plan — names resolved against the schema and catalog, types checked, and every attribute given an ID such as amount#2. Analysis runs as soon as you build the DataFrame: selecting a misspelled column raised AnalysisException before any action.
3. Optimized logical plan — batches of rewrite rules. The filter amount > 10 became isnotnull(amount#2) AND (amount#2 > 10.0): a null check was inferred and the analyzer's cast(10 as double) folded into a literal.
4. Physical plan — operators that execute: Scan, Filter, Project, a partial and a final HashAggregate, and the Exchange between them. What explain prints here is the prepared plan, after exchanges and code generation stages were added.

Read physical plans bottom up: data flows from the scan toward the root. The other modes answer other questions. "formatted" prints a numbered outline followed by per-operator details — Input, Condition, Keys, Functions, and for file scans PartitionFilters and PushedFilters — which is easier to scan than one long tree. "cost" prints the optimized logical plan with its statistics when they are available, and "codegen" prints generated code when there is any.

Check the sources: EXPLAIN modes · DataFrame.explain · QueryExecution source

Physical planning: rule order, estimates, and where cost enters

SparkPlanner holds a list of strategies — join selection, aggregation, file-source scans, basic operators — and applies them to the optimized plan. The planner interface can yield several candidates, but in Spark 3.5.7 QueryExecution.createSparkPlan takes the first one: its source comment says it uses next() "for now, but we will implement to choose the best plan." No search costs alternative physical plans against each other.

Choices such as the join strategy are rules evaluated in order. For an equi-join, JoinSelection checks hints first. Without an applicable hint it picks a broadcast hash join if one side's estimated size fits spark.sql.autoBroadcastJoinThreshold (10 MB) and the join type allows it; then a shuffled hash join only if one side is small enough for a local hash map, much smaller than the other, and spark.sql.join.preferSortMergeJoin is false; then a sort-merge join for sortable keys; and finally a cartesian product or broadcast nested loop join.

Cost enters through statistics on the logical plan. By default Spark estimates size only: a file relation reports its file size, and a filter keeps its child's estimate. A 1,000-row table filtered to amount > 900 still estimated 5.5 KiB. ANALYZE TABLE … COMPUTE STATISTICS records size and row count for a catalog table, and FOR COLUMNS adds column statistics such as min, max, null count, and distinct count. Those numbers drive estimates only when spark.sql.cbo.enabled is true, and it is false by default. With column statistics and CBO on, the same filter estimated rowCount=100 and 2000 B; with CBO off, the estimate stayed 5.5 KiB. Cost-based join reordering also requires spark.sql.cbo.joinReorder.enabled, false by default.

The accurate summary: rule-based logical optimization and rule-ordered physical planning, fed by estimates that become cost-based only when you collect statistics and turn CBO on. Better estimates change the physical plan by changing what the rules see.

Check the sources: QueryExecution source · QueryPlanner source · JoinSelection · Size-only estimates · ANALYZE TABLE · CostBasedJoinReorder · CBO settings

Preparations: exchanges and whole-stage codegen

The chosen physical plan still needs preparation rules. In 3.5.7 they include PlanSubqueries, RemoveRedundantProjects, EnsureRequirements, RemoveRedundantSorts, ApplyColumnarRulesAndInsertTransitions, CollapseCodegenStages, and exchange reuse. With AQE enabled, InsertAdaptiveSparkPlan runs first; when it wraps the query in AdaptiveSparkPlan, the remaining rules are applied later, stage by stage, as the adaptive plan executes.

EnsureRequirements compares what each operator needs from its children — a hash distribution on the grouping keys, a sort order for a sort-merge join — with what they produce, and inserts a shuffle Exchange or a Sort where they differ. That is the origin of Exchange hashpartitioning(category#1, 200), ENSURE_REQUIREMENTS between the partial and final aggregates.

CollapseCodegenStages fuses each run of adjacent operators that support code generation into one generated Java function. A *(n) prefix marks membership in codegen stage n. The run breaks at an operator without codegen support, and at one whose expressions need the interpreted fallback. Observed boundaries:

• Exchange separated *(1) (scan, filter, project, partial aggregate) from *(2) (final aggregate).
• collect_list planned as ObjectHashAggregate, printed without a star.
• A Python UDF became BatchEvalPython, splitting *(1) below from *(2) above — inside one scheduler stage, since no shuffle was involved. A codegen stage ID is not a scheduler stage ID.
• A Project computing transform(xs, x -> x * 2) lost its star: that higher-order function evaluates through the interpreted fallback, although it never leaves the JVM.

Codegen can also switch itself off: if a generated method's bytecode exceeds the internal spark.sql.codegen.hugeMethodLimit (65535), whole-stage codegen is deactivated for that subtree.

With AQE on, look after execution. Before an action, AdaptiveSparkPlan isFinalPlan=false showed no *(n) markers, and explain("codegen") reported "Found 0 WholeStageCodegen subtrees." After collect(), the final plan showed *(1) and *(2). With AQE disabled, the same query showed both stages up front and found 2 subtrees.

Check the sources: Preparation rules · EnsureRequirements · WholeStageCodegenExec · AdaptiveSparkPlanExec

What blocks predicate pushdown

A file scan in the physical plan lists three kinds of filter. PartitionFilters are evaluated by Spark against partition values from the directory listing, so pruned directories are never opened. PushedFilters are predicates translated into the data source filter API and handed to the reader; for Parquet, that lets the reader skip data using its statistics. DataFilters are the data-column predicates. Spark keeps a Filter above the scan for every non-partition predicate, so pushdown saves reading, not correctness checks.

Translation accepts specific shapes: a column compared with a literal, IN with literals, null checks, string prefix, suffix, and contains tests, and AND/OR/NOT combinations of those. Everything else stays in Spark. On a 3,000-row Parquet dataset partitioned by day (3 directories, 6 files):

Pushdown depends on the shape of an expression, not on how simple it looks. Write predicates as built-in expressions, compare columns to literals rather than wrapping the column, and keep partition predicates separate from data predicates. Then confirm in the plan and in the scan's files-read metric. Storage formats explains what a Parquet reader does with a pushed filter; Lazy evaluation shows why the plan exists before anything is read.

Check the sources: FileSourceStrategy · Filter translation · ExtractPythonUDFs · Parquet pushdown

Adaptive query execution: re-planning between stages

AQE is enabled by default since Spark 3.2.0 (spark.sql.adaptive.enabled). AdaptiveSparkPlanExec splits the plan into query stages at exchanges, runs the stages whose inputs are ready, and after each stage materializes it re-optimizes and re-plans the rest of the query using the measured output. It wraps queries that contain an exchange or a subquery, or whose operators require a distribution; a plain filtered scan is not wrapped.

Before an action, explain prints AdaptiveSparkPlan isFinalPlan=false. After an action, explain on the same DataFrame prints isFinalPlan=true with a == Final Plan == and the == Initial Plan ==. The final plan names completed stages (ShuffleQueryStage 0, BroadcastQueryStage 2) and how their output is read: AQEShuffleRead coalesced, local, or skewed. The SQL tab in the Spark UI shows the same final plan.

• Coalescing post-shuffle partitions. A groupBy over 10 keys with the default 200 shuffle partitions ran its reduce stage as 1 task behind AQEShuffleRead coalesced; with coalescing disabled, the same stage ran 200 tasks. Under the default spark.sql.adaptive.coalescePartitions.parallelismFirst=true, Spark ignores the 64 MB advisory size and respects only the 1 MB minimum partition size; the setting's own description recommends false.
• Switching a join to broadcast. A Parquet side filtered to 50 rows kept its 13.4 MiB file-size estimate, so under a 1 MiB broadcast threshold the initial plan was SortMergeJoin. After both shuffle map stages ran, the final plan was BroadcastHashJoin, reading each shuffle through AQEShuffleRead local. The shuffle writes had already happened; the switch saved the sorts and reduce-side fetches. (The optimizer had also inferred fk < 50 for the other side from the join key, so that side was built.) Converting to a shuffled hash join instead requires spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold above its default of 0.
• Splitting skewed partitions. A join partition counts as skewed when it exceeds both 5 × the median partition size and 256 MB (defaults). With both thresholds lowered for a laptop-sized dataset, the plan showed SortMergeJoin(skew=true) and AQEShuffleRead skewed, and the join stage ran 9 tasks instead of 8. The documentation describes sort-merge joins; the 3.5.7 rule also matches shuffled hash joins.

Check the sources: AQE guide · AdaptiveSparkPlanExec · InsertAdaptiveSparkPlan · OptimizeSkewedJoin · AQE settings

What AQE cannot fix

AQE changes decisions that come after a completed stage, using that stage's output sizes. That leaves a lot out of reach:

• Work already done. A runtime broadcast arrives after the shuffle map stages were written. If you know a side is small, let the static plan broadcast it: collect statistics or add a hint you have checked.
• The first read. Scan parallelism comes from file splitting (spark.sql.files.maxPartitionBytes), not from AQE. Thousands of small files, an unpushable predicate, or unsorted data that defeats row-group statistics are untouched.
• Queries without an exchange. They are not wrapped in an adaptive plan at all.
• Hot keys outside joins. Skew splitting covers joins and rebalance partitions. One huge key in a groupBy or a window still lands in one task, and coalescing only merges partitions — it never splits one.
• Opaque code. A Python UDF still blocks pushdown of predicates on its result and splits codegen.
• Memory inside one task or the driver. A partition too large for its task, or a large collect(), is not a planning decision.

Treat the final plan as evidence of what AQE decided, then fix the remaining cause where it lives: file layout, the query's expressions, or configuration. Joins & data skew covers manual fixes when AQE's split is not enough.

Check the sources: AQE guide · When AQE applies · OptimizeSkewedJoin

Try it on a small dataset

Run this in a PySpark notebook with an existing spark session. Create a local session if needed. The example was run for review; this website does not execute Spark.

from pyspark.sql import functions as F

orders = spark.createDataFrame(
    [(1, "books", 12.0), (2, "games", 30.0), (3, "books", 8.0), (4, "music", 15.0)],
    "order_id INT, category STRING, amount DOUBLE",
)

totals = (
    orders.filter(F.col("amount") > 10)
    .select("category", (F.col("amount") * 1.1).alias("gross"))
    .groupBy("category")
    .agg(F.sum("gross").alias("total"))
)

totals.explain("extended")  # parsed, analyzed, optimized, physical
totals.collect()            # run it: AQE plans each stage with real sizes
totals.explain()            # same DataFrame: now the final and initial plans

Ran on local PySpark 3.5.7 (local[2], JDK 17) with default SQL settings on 2026-09-15; output above is observed.

An explain before an action shows a sort-merge join and 200 shuffle partitions, but the SQL tab shows a broadcast hash join and far fewer tasks. Which plan is wrong?

Say your answer out loud before opening the notes.