Spark SQL optimizer architecture — Catalyst rules, cost, codegen, AQE

Why architecture matters here

Spark SQL performance is often optimizer performance. A query that runs 10x faster after an upgrade usually got new rules or better stats. A query that regressed usually lost a physical strategy or a stat.

The architecture matters because tuning happens at multiple layers: schema + stats (feed the cost model), hints (nudge the physical planner), AQE (adapt at runtime), and extensions (custom rules for specialized cases).

With the pipeline in mind, you can read explain output and understand why the plan looks as it does — the first step to improving it.

The architecture: every piece explained

The top strip is the semantic path. SQL / DataFrame lands as a logical plan tree. Analyzer resolves column references and types against the catalog. Optimizer applies rule-based rewrites — predicate pushdown, constant folding, projection pruning. Cost model uses statistics to choose join strategies and reorder joins.

The middle row is the execution path. Physical planner picks concrete operators (broadcast vs shuffle hash join, sort merge). Code gen (Tungsten) generates JVM bytecode for whole stages so tight loops run without virtual dispatch. AQE adapts at runtime — recognizes skew, coalesces small partitions, switches join strategies based on observed sizes. Executor runs the plan.

The lower rows are practice. Extension points allow custom rules and strategies for domain-specific operators. Observability centers on the explain output and query plan UI. Ops handles statistics collection, hints, and plan review for regressions.

Four phases, and what each one is allowed to change

Catalyst is not a single optimizer. It is four passes over a tree, and the value of the design is that each pass has a different, narrower contract than the one before it. Confusing the contracts is how people end up blaming the wrong phase for a bad plan.

Parsing and analysis. The parser produces an unresolved logical plan: syntactically valid and semantically meaningless. UnresolvedRelation("sales") is not a table, it is a name. The analyzer's job is binding - resolve names against a catalog, assign a data type to every expression, insert implicit casts, expand *. It is allowed to add and rewrite nodes, but it makes no cost decisions at all. Its output either is fully resolved or the query fails.

Logical optimization. Rewrites the analyzed plan into another logical plan that produces identical rows for every possible input. It knows nothing about executors, partitions, or the cluster. A Filter can move, a Project can shrink, an expression can be folded to a constant - but a Join is still an abstract Join when this phase finishes.

Physical planning. Strategies map each logical operator to one or more executable operators. Now the cluster exists: a Join becomes BroadcastHashJoinExec, SortMergeJoinExec, ShuffledHashJoinExec, or a nested-loop variant. Afterwards a rule called EnsureRequirements walks the physical tree, asks each operator what distribution and ordering it requires of its children, and inserts Exchange and Sort nodes only where the child does not already satisfy them. Every shuffle you see in a plan was inserted here; you never wrote one.

Code generation. Adjacent physical operators are fused into a single generated Java method. That machinery is its own subject - see whole-stage code generation and Tungsten.

The practical value of the boundary: a bug in phase two is a wrong-answer bug, because that phase claims to preserve semantics. A bug in phase three is only a slow-query bug. When a result is wrong, suspect a rewrite. When a result is slow, suspect a strategy or a statistic.

Trees, expressions, and why immutability is the design

Everything Catalyst manipulates is a TreeNode. Two hierarchies matter. LogicalPlan nodes are relational operators with children - Project, Filter, Join, Aggregate. Expression nodes are the scalar computations that live inside operators. A Filter holds one child plan and one condition expression, and that condition is itself a tree: GreaterThan(AttributeReference("age"), Literal(30)).

Every node is immutable. Transformation happens through transformDown and transformUp, which take a partial function from node to node. Where the function is undefined the node is returned untouched; where it fires, the node and its ancestors are copied and the untouched subtrees are shared by reference. That is why a real Catalyst rule is often ten lines:

// a rule is a partial function over an immutable tree
plan.transformAllExpressions {
  case GreaterThan(l: Literal, r: Literal) => Literal(evaluate(l, r))
}

// unchanged subtrees are shared, not copied
val optimized = analyzedPlan.transformUp {
  case Filter(c1, Filter(c2, child)) => Filter(And(c1, c2), child)
}

Immutability buys three concrete things. One rule cannot corrupt the input another rule is about to read, so batches compose and can be unit-tested one rule at a time. The earlier plans stay intact and printable, which is exactly what lets explain("extended") show you parsed, analyzed, and optimized plans side by side. And because an unchanged transform returns the identical object graph, "did anything change" is a cheap comparison - which is the signal the fixed-point loop below runs on.

The cost is allocation churn, and it is paid on the driver before a single task is scheduled. Plans with thousands of columns, deep stacks of views, or long chains of withColumn calls can spend seconds in analysis and optimization. If a small query has a large gap between submission and the first task, look at planning time before you look at the cluster.

Rule batches and the fixed point

Rules are grouped into ordered batches, and each batch carries a strategy. Once means apply the rules in order, one pass, done. FixedPoint(n) means loop the whole batch until the plan stops changing, or until n iterations have run.

The loop exists because rules enable each other and hand-encoding that dependency graph would be miserable. Collapsing two adjacent projections makes two filters adjacent; combining those filters produces a single predicate; that predicate is now a candidate for pushdown through a join; pushing it exposes a constant on one side that folding can eliminate. None of those rules knows about the others. Iterating to a fixed point makes the ordering emergent instead of hand-maintained.

The price is a hard requirement: a rule must be idempotent. Applying it to its own output has to produce that same output. A rule that wraps an expression in a new alias on every pass never reaches a fixed point; the batch spins until it hits the iteration cap and logs that maximum iterations were reached for that batch. The plan is still correct - every rewrite was semantics-preserving - but the driver burned a hundred passes over the tree to get there. In practice this is almost always a custom rule, not a built-in one, which is the main reason the extension section below leads with idempotence.

The cap is spark.sql.optimizer.maxIterations, default 100. Raising it is the wrong fix nearly every time; the message is a bug report about a non-converging rule.

The complementary knob is spark.sql.optimizer.excludedRules, a comma-separated list of fully qualified rule class names to skip. It is the bisection instrument when a rewrite is suspected of causing a regression or a wrong answer. Rules that are required for correctness cannot be excluded, and Spark logs when it ignores an exclusion you asked for.

From unresolved to resolved - the analyzer and the catalog

The parser emits placeholders: UnresolvedRelation, UnresolvedAttribute, UnresolvedFunction. Every node exposes a resolved flag, and the analyzer runs its own fixed-point batches until the whole tree reports resolved.

Relation resolution goes through the catalog. In modern Spark the catalog is plural: a catalog manager routes the first part of a multi-part identifier to a registered catalog implementation, with spark_catalog as the built-in session catalog backed by a metastore or an in-memory store. That indirection is what lets a table format plug itself in as a first-class catalog rather than a path you agree to interpret consistently.

Attribute resolution matches each unresolved name against the output attributes of the node's children, producing an AttributeReference carrying a name, a data type, a nullability flag, and an ExprId. The ExprId is the load-bearing part: names are for humans, and everything downstream compares identifiers. This is why a self-join shows region#42 on one side and region#87 on the other - one side was given fresh ids precisely so that the join condition is unambiguous. When ambiguity cannot be broken, you get an AnalysisException instead of a guess.

Type coercion inserts casts so that comparing an int column to a bigint literal has defined semantics, and structural check rules reject things like an aggregate expression in a WHERE clause. The phase is all-or-nothing, and that is the property the rest of Catalyst is built on: the optimizer is entitled to assume every expression has a type. It is also why a misspelled column name fails immediately, before any job is submitted.

The rewrite families that do most of the work

Predicate pushdown. Move filters as close to the leaves as legality allows. Fewer rows enter joins and shuffles, and for file sources the surviving predicates are handed to the scan, where they can eliminate partitions and skip row groups using footer statistics. See partition pruning for what happens once a predicate reaches the storage layer.

Column pruning. Push projections down and drop columns nothing references. On a columnar format this is the difference between reading four columns and reading three hundred. It is also why SELECT * in an intermediate view is harmless if the final query only touches a few columns - the pruning rule sees through it - and expensive the moment something downstream genuinely references everything.

Constant folding and expression simplification. 2 + 3 becomes 5 and upper('a') becomes 'A' at plan time rather than once per row. Over a billion rows, removing a single per-row addition removes a billion operations. A related rule rewrites a long IN list from a chain of Or comparisons into a hash-set lookup, which changes an O(n) per-row scan into a constant-time one.

Null propagation. Null semantics are mechanical enough to exploit. Expressions that are provably null collapse to a null literal, coalesce chains shrink when an earlier argument is non-nullable, and a col IS NOT NULL check on a column the analyzer already marked non-nullable folds to true and disappears. This is one of the places where declared schema nullability pays for itself.

Boolean simplification. a AND true becomes a, a OR true becomes true, and a common conjunct is factored out of a disjunction so it can be pushed independently. This matters more than it looks, because machine-generated SQL and long hand-written OR chains routinely hide a factorable term that unlocks a pushdown.

Constraint inference. Given a.id = b.id and a.id > 10, the equality is a constraint, so b.id > 10 is inferable and pushable to b - potentially eliminating most of the other table before the join. The same machinery infers IS NOT NULL on equi-join keys, which is where the isnotnull(id#12) predicates you never wrote come from. It is governed by spark.sql.constraintPropagation.enabled; on pathologically wide plans the inference itself can become the planning bottleneck, which is the only reason to turn it off.

Empty and local relation collapse. A predicate provably matching nothing turns a subtree into an empty relation, and the operators above it fold away. Separately, a small enough local relation with only foldable operations above it is evaluated on the driver. That is why some trivial queries return without ever launching a job.

When a predicate cannot move

The refusals are the interesting part, because every one of them is a correctness constraint rather than a missing feature. "Why did my filter not get pushed" almost always has an answer on this list.

Non-deterministic expressions

Every expression carries a deterministic flag, and pushdown rules check it. rand(), monotonically_increasing_id(), and input_file_name() are not deterministic, and moving them changes how many times and on which rows they are evaluated - which changes the answer. So WHERE rand() < 0.01 pins that filter high in the plan and forces everything below it to be materialized. A UDF you deliberately registered as non-deterministic becomes a wall the optimizer will not cross, which is the correct behaviour and worth knowing before you set that flag.

UDFs are opaque

A Scala or Java UDF appears in the tree as a node wrapping a function object. Catalyst knows its input types, its output type, and that it is deterministic by default. It knows nothing else: it cannot fold it, cannot translate it into a filter a data source would understand, and cannot infer constraints from it. So WHERE my_udf(col) = 1 will never become a pushed filter - the source returns everything and Spark evaluates row by row. The rewrite that pays is expressing the predicate in built-in expressions so it is legible to the optimizer, even when the UDF version reads better.

Python UDFs are a stronger boundary still. A dedicated analyzer rule extracts them out of the surrounding expression into a separate evaluation operator, because they run in an external worker process. The plan makes this visible, and everything on the far side of that operator is evaluated after data has been serialized out of the JVM. A filter that lands there is the most expensive place a filter can be.

Outer join null semantics

In a LEFT OUTER JOIN, the right side's columns may be null because the join itself padded an unmatched left row. A predicate on right-side columns therefore cannot be pushed into the right input: pushing it removes rows before the padding happens, and left rows that should have appeared with nulls vanish from the result. Predicates on the preserved side are still pushable to that side. For a FULL OUTER JOIN neither side is safe. This is the single most common answer to "the filter stayed above the join".

The mirror image is a rewrite worth recognizing. If the WHERE clause contains a predicate on the null-supplying side that cannot be satisfied when that side is null - b.status = 'active', say - then no null-padded row can survive the filter, so the outer join is equivalent to an inner join. Catalyst performs that rewrite, and once it does, pushdown becomes legal on both sides. This is why LEFT JOIN ... WHERE b.col = x behaves like an inner join: usually a bug in the query, occasionally a large speedup.

Aggregates, windows, and row-count-sensitive operators

A predicate on a grouping column can be pushed below an Aggregate, because grouping preserves the meaning of that column. A predicate on an aggregate result cannot - HAVING sum(x) > 10 refers to a value that does not exist until the aggregate has run.

Below a window operator, only predicates on the partitioning columns are safe. Filtering on a window result obviously has to wait, but so does filtering on any non-partitioning column, because removing rows changes the frame that the window function computes over. The same logic restricts pushing anything below a Limit or below an explode: those operators' output depends on how many rows arrived, so changing the input changes the output.

End-to-end flow

End-to-end: a join query lands. Analyzer resolves tables + columns. Optimizer pushes filters below the join, prunes projections, reorders joins using stats. Cost model picks broadcast join for a 4 GB dimension. Physical planner generates a SortMergeJoin for the fact-fact join. Tungsten generates bytecode for the tight filter loop. Execution begins; AQE observes skew on the fact table; splits skewed partition. Coalesces small post-shuffle partitions. Total runtime 12 seconds vs 40 without AQE and codegen. Explain output shows all rules applied.

Cost-based optimization and the statistics you probably do not have

Everything above is rule-based: the rewrite fires whenever the pattern matches, regardless of the data. A handful of decisions cannot be made that way, because they depend on how big things actually are.

Every logical node carries a statistics object with an estimated size in bytes and, when available, a row count. In the default configuration that estimate is derived from file sizes at the leaves and propagated upward with coarse rules. Crucially, in the default path a Filter does not meaningfully shrink the estimate. That estimated size is what gets compared against spark.sql.autoBroadcastJoinThreshold, which is why a table you filtered down to a few thousand rows still is not broadcast: the planner is still looking at the size of what is on disk.

Real statistics are opt-in and are collected explicitly:

-- table-level: row count and total size, stored in the catalog
ANALYZE TABLE sales COMPUTE STATISTICS;

-- column-level: distinct count, min, max, null count, avg/max length
ANALYZE TABLE sales COMPUTE STATISTICS FOR COLUMNS region, customer_id, order_ts;

-- partitioned tables: analyze the partitions you actually touched
ANALYZE TABLE sales PARTITION (dt = '2026-08-01') COMPUTE STATISTICS;

Column statistics are what make selectivity estimation possible at all. With min, max, and a distinct count, WHERE age > 30 can be estimated as a fraction of the rows instead of being treated as free. With spark.sql.statistics.histogram.enabled the collection also builds equi-height histograms, which handle skewed distributions that min/max alone describes badly.

Then the optimizer has to be told to use them. spark.sql.cbo.enabled is false by default. Join reordering has a second flag, spark.sql.cbo.joinReorder.enabled, also off, with the dynamic-programming search bounded by spark.sql.cbo.joinReorder.dp.threshold - around a dozen relations - because the search space grows exponentially in the number of joined tables.

The dangerous failure mode is not missing statistics, it is stale ones. Statistics are a snapshot in the catalog. Appending to or overwriting a table does not recompute them; spark.sql.statistics.size.autoUpdate.enabled maintains the table size on write but not the column statistics. A table that has grown fiftyfold since the last ANALYZE still reports its old row count, the planner confidently chooses a broadcast, and the driver collects a multi-gigabyte relation and dies with an out-of-memory error or a broadcast timeout that mentions nothing about statistics. Absent statistics produce conservative plans; wrong statistics produce confidently bad ones.

Treat column statistics as something you maintain deliberately for join keys and frequently filtered columns. Computing distinct counts across three hundred columns of a wide table is its own expensive job, and most of that work will never influence a plan.

Runtime statistics are the other half of the answer, and they belong to a different mechanism: adaptive query execution re-plans using the actual sizes observed at shuffle boundaries, which is precisely the information the static cost model does not have.

Join order, join strategy, and where hints sit

Two different things get called join optimization. Reordering is logical: it decides which tables are joined first, which determines the size of intermediate results. Without the cost-based path this is heuristic - conditions are pushed so that inner joins become possible and a star-shaped query is arranged around its fact table. With the cost-based path enabled and column statistics present, a dynamic-programming search enumerates orders and scores them by estimated cardinality.

Strategy selection is physical, and happens in the join strategy rule during physical planning. It considers the join type, whether the keys are equi-join keys, the estimated sizes, and any hints, then emits a broadcast hash join, a shuffled hash join, a sort-merge join, or a nested-loop variant. Hints - BROADCAST, MERGE, SHUFFLE_HASH, SHUFFLE_REPLICATE_NL - override the cost decision. That is what makes them useful and what makes them a liability: a hint outlives the data distribution that made it correct, and a broadcast hint on a dimension table that quietly grew is a common cause of driver out-of-memory failures. The mechanics of the broadcast itself are covered in broadcast joins, and the shuffle path in the shuffle article.

One physical-planning detail is worth internalizing because it is directly actionable: EnsureRequirements inserts an exchange only where a child does not already satisfy the required distribution. If an upstream operator already partitioned the data the right way, the shuffle is not added. That "already satisfied" test is the entire reason a well-chosen repartition, or a bucketed table, can remove a shuffle rather than add one.

Where Catalyst stops - the connector boundary

Predicate and column pushdown reach the storage layer through a negotiated interface rather than an assumption. In DataSource V2 the optimizer obtains a scan builder and interrogates it for capabilities: it offers filters, and the source returns the filters it could not handle. That return value is the whole contract. If the source returns nothing unhandled, Catalyst deletes the filter node from the plan and trusts the source completely; if the source hands filters back, Catalyst keeps the node and evaluates them itself. Column pruning works the same way, with an ordering constraint - the requested columns must still include anything an unhandled filter needs.

A source may also report statistics after pushdown has been applied, and those feed straight into the cost model described above. That is the cleanest way to make a broadcast decision correct for an external system, and also the easiest way to make it catastrophically wrong. The connector-side detail is covered in the DataSource V2 article; from Catalyst's side, the thing to remember is that pushdown is a negotiation whose outcome is recorded in the physical plan and is therefore auditable.

Extending Catalyst with your own rules

Catalyst is extensible by design, which is unusual for a query optimizer, and the extension surface is SparkSessionExtensions. You supply a function that registers injections, and point Spark at it with spark.sql.extensions or register it programmatically at session build time.

class MyExtensions extends (SparkSessionExtensions => Unit) {
  def apply(ext: SparkSessionExtensions): Unit = {
    ext.injectResolutionRule(spark => MyResolutionRule(spark))  // analyzer
    ext.injectCheckRule(spark => RejectUnpartitionedScan)       // reject bad plans
    ext.injectOptimizerRule(spark => MyRewrite)                 // logical optimizer
    ext.injectPlannerStrategy(spark => MyStrategy)              // physical planning
  }
}
// spark.sql.extensions = com.example.MyExtensions

The injection points mirror the phases: resolution and post-hoc resolution rules run inside the analyzer, check rules can fail a query with a message of your choosing, optimizer rules join the logical rewrite batches, and planner strategies get a chance to produce physical operators before the built-in strategies do. There is also a hook for rules that must run before the cost-based phase, which matters if your rewrite changes cardinality.

Writing one is easy; writing a safe one has rules of its own. It must preserve semantics - nothing verifies that for you. It must be idempotent, because it runs inside a fixed-point batch. It should match narrowly: a rule that keys only on an operator type will fire on plans you never anticipated, including the internal plans behind views and cached queries. It must preserve expression ids when rewriting an expression in place, or downstream references break in ways that surface as confusing analysis errors. And a resolution rule sees partially resolved trees, so it cannot assume types are present.

Test the rule by asserting on the optimized plan, not on the results. Correct results prove nothing about whether your rule fired - that is exactly the failure mode a plan assertion catches and an output assertion does not.

The uses that justify the machinery are narrow and real: mapping a domain-specific function onto a native expression so it stops being opaque, injecting row-level security predicates that users cannot omit, and check rules that reject a query scanning a large partitioned table without a partition predicate - a cheap guardrail that has saved more cluster-hours than most tuning work.

Telling whether a rule actually fired

Reading physical plans generally is covered in EXPLAIN plans. The narrower question here - did a specific rewrite happen - has three precise instruments.

Diff the phases. The difference between the analyzed plan and the optimized plan is the optimizer's work. explain("extended") prints both, and the plans are also reachable programmatically, which makes them assertable in a test.

df.queryExecution.analyzed       // before any optimization
df.queryExecution.optimizedPlan  // after the logical rewrite batches
df.queryExecution.sparkPlan      // after strategy selection
df.queryExecution.executedPlan   // after exchange insertion + codegen fusion

df.explain("extended")   // parsed / analyzed / optimized / physical
df.explain("cost")       // logical plan annotated with the stats the planner used

If a filter sits above a join in the analyzed plan and below it in the optimized plan, pushdown fired. If it is in the same position in both, it did not, and one of the refusals above applies - work down that list rather than guessing.

Turn on the plan change log. Setting spark.sql.planChangeLog.level to INFO or WARN makes the rule executor log the before and after plan for every rule that changed anything, plus a per-batch summary of which rules were effective. It is decisive and extremely noisy, so scope it with spark.sql.planChangeLog.rules, which takes a comma-separated list of rule class names. On older builds these settings live under a spark.sql.optimizer. prefix.

Bisect by exclusion. Put the suspected rule in spark.sql.optimizer.excludedRules and re-run. If the plan or the timing changes, you have identified the rule. This is the fastest way to attribute a regression after an upgrade, and it doubles as a temporary mitigation while the real fix is found.

One trap worth naming: explain("cost") is the fastest way to answer "why was this not broadcast", because it shows whether the planner believes your table is eight megabytes or eight gigabytes. That single number explains more surprising plans than any other piece of output.