Inside Spark's Catalyst optimizer: how a query gets rewritten, rule by rule

• Why there is an optimizer at all: RDDs versus DataFrames
• Architecture: four trees, one pipeline
• What a rule actually is
• The engine: batches, strategies, and the fixed point
• One query, four trees
• Which rules actually fired
• A tour of rules you can see working
• Turning a rule off
• Writing your own rule
  • Testing a rule
• Physical planning: from what to how
• Rule-based versus cost-based optimization
  • What statistics Spark actually holds
  • The same query, with and without CBO
  • Join reordering, and the honest result
• Codegen, where the tree stops being a tree
• Dynamic partition pruning
• Adaptive query execution
• Debugging and inspection
• Production notes
• Frequently asked questions
• Conclusion
• References
• Trademarks

TL;DR

• Catalyst is a tree rewriter. Every phase takes a tree and returns a tree, and almost every optimization is one small function matching a pattern and returning a replacement.
• A query passes through four trees: parsed, analyzed, optimized, physical. explain() shows them, and the difference between analyzed and optimized is the entire optimizer’s work made visible.
• Rules run in batches, each either Once or to a fixed point, meaning repeatedly until the tree stops changing, which is how one rule enables another with nobody ordering them by hand. On a single join-and-aggregate query Spark invoked over two hundred rules and eleven changed the plan.
• You can switch a rule off with spark.sql.optimizer.excludedRules and add your own in about ten lines, which is the fastest way to understand what any of them actually does.
• Almost all of it is rule-based. The cost-based part is a thin layer that needs statistics you have to compute yourself, and in the join-reordering test below the rule-based ReorderJoin did the work while CostBasedJoinReorder ran and changed nothing.

Most explanations of Catalyst stop at a diagram with four boxes in it. That is a fair summary and it is not much use when a plan does something surprising, because the interesting questions are one level down. Which rule removed my filter? Why did a cast appear? Why did the optimizer not push this predicate? What runs before what?

Those questions have exact answers, and they are answerable from your own shell. Catalyst keeps its rules in named batches you can list, records which ones fired on your query, and lets you disable any of them and watch the plan change. This post works through that: the pipeline, what a rule is in the source, the engine that runs them, and then a query taken apart rule by rule.

Every plan and number below came from running the query on Spark 4.0.2 with a plain in-memory catalog session. The counts move between releases, so treat them as a snapshot of shape rather than constants.

Why there is an optimizer at all: RDDs versus DataFrames

Catalyst exists because of a change in what you hand Spark. An RDD stage is a chain of closures, and a closure is opaque: Spark knows it must call your function on every row and nothing else about it. A DataFrame operation is a description of intent, which is inspectable.

The difference is visible rather than philosophical. Both of these do the same work, filtering on fare and keeping one column:

# declarative: Catalyst sees columns and a predicate
df = (spark.read.parquet("/tmp/catalyst/trips")
        .select("trip_id", "city_id", "fare")
        .filter("fare > 10.0")
        .select("trip_id"))
print(df._jdf.queryExecution().optimizedPlan().toString())

# imperative: Catalyst sees three Python functions
rdd = (spark.read.parquet("/tmp/catalyst/trips").rdd
         .map(lambda r: (r.trip_id, r.fare))
         .filter(lambda p: p[1] > 10.0)
         .map(lambda p: p[0]))
print(hasattr(rdd, "queryExecution"))
print(rdd.toDebugString().decode().splitlines()[0])

Project [trip_id#0L]
+- Filter (isnotnull(fare#3) AND (fare#3 > 10.0))
   +- Relation [trip_id#0L,city_id#1L,rider_id#2L,fare#3,trip_date#4] parquet

False
(2) PythonRDD[8] at RDD at PythonRDD.scala:56 []

The DataFrame has a plan, so it can be rewritten: the executed plan reads struct<trip_id:bigint,fare:double>, two columns out of five, and hands the predicate to Parquet. The RDD has no plan at all, only a lineage of MapPartitionsRDD stages, so there is nothing to prune and nothing to push. It reads every column and evaluates your comparison in Python, row by row.

That is the whole trade. You give up control over how the work happens, and in return a rewriting engine gets to see what the work is.

Architecture: four trees, one pipeline

Catalyst never mutates anything. Each phase consumes a tree and produces a new one, which is why you can print any intermediate stage without disturbing the rest.

flowchart TB
  A["SQL string or<br/>DataFrame calls"] --> B["<b>Unresolved</b><br/>logical plan"]
  B -->|"<b>Analysis</b>"| C["<b>Analyzed</b><br/>logical plan"]
  CAT[("Catalog /<br/>metastore")] -.->|"relations, columns, types"| C
  C -->|"<b>Logical optimization</b><br/>rule batches, to a fixed point"| D["<b>Optimized</b><br/>logical plan"]
  D -->|"<b>Physical planning</b><br/>strategies"| P1["physical plan 1"]
  D --> P2["physical plan 2"]
  D --> P3["physical plan n"]
  P1 --> CM{"<b>Cost model</b><br/>statistics"}
  P2 --> CM
  P3 --> CM
  CM -->|"selected"| SP["<b>Selected</b><br/>physical plan"]
  SP -->|"<b>Code generation</b><br/>Tungsten"| RDD["RDDs of<br/>InternalRow"]
  STATS[("Table and column<br/>statistics")] -.-> CM

That fan-out in the middle is the part worth being precise about, because the familiar version of this diagram promises more than Spark delivers. Catalyst does generate alternative physical plans and does consult a cost model, but for most queries the alternatives are per-operator choices, above all which join algorithm to use, and the “cost” consulted is a size estimate rather than a search over whole plan shapes. The section on rule-based versus cost-based optimization below measures exactly how thin that layer is.

The names are worth keeping straight, because the phases fail in different ways and the error messages differ:

Analysis is not optimization. Analysis makes the query meaningful, so it adds things: catalog relations, SubqueryAlias wrappers, casts. Optimization makes it cheap, so it mostly removes things. Reading the two trees side by side is the single most useful habit in this whole area.

What a rule actually is

A rule is an object with one method, from tree to tree. The mechanism is easiest to see in ConstantFolding, whose core is a pattern match over expressions. These are the two clauses that do the actual folding, quoted from the source:

    // Skip redundant folding of literals. This rule is technically not necessary. Placing this
    // here avoids running the next rule for Literal values, which would create a new Literal
    // object and running eval unnecessarily.
    case l: Literal => l

    // Fold expressions that are foldable.
    case e if e.foldable => tryFold(e, isConditionalBranch)

and tryFold is where the expression is evaluated at planning time:

  private def tryFold(expr: Expression, isConditionalBranch: Boolean): Expression = {
    try {
      Literal.create(expr.freshCopyIfContainsStatefulExpression().eval(EmptyRow), expr.dataType)
    } catch {
      case NonFatal(_) if isConditionalBranch =>
        // When doing constant folding inside conditional expressions, we should not fail
        // during expression evaluation, as the branch we are evaluating may not be reached at
        // runtime, and we shouldn't fail the query, to match the original behavior.
        expr.setTagValue(FAILED_TO_EVALUATE, ())
        expr
    }
  }

That is the whole idea: e.foldable is the test for “this expression depends on no column”, eval(EmptyRow) computes it with no input row, and the result replaces the subtree as a Literal. The surrounding match has other clauses for cases where folding would be wrong, and the catch exists because an expression in an untaken CASE branch must not fail the query at planning time. That pattern repeats across the rule set: a small core, plus the exceptions found the hard way.

Two tree-walking helpers appear constantly, and the difference matters when you write your own:

The transformWithPruning variant is a detail worth noticing, because it explains a line you see in nearly every modern rule. Each tree node carries a set of TreePattern flags, so a rule that only cares about literals can skip whole subtrees that contain none, rather than visiting every node to discover it has nothing to do.

The engine: batches, strategies, and the fixed point

Rules do not run in a flat list. RuleExecutor runs batches, and each batch has a strategy that decides how many times its rules run.

flowchart TB
  S["a batch: name, strategy, rules"] --> L{"for each rule<br/>in order"}
  L --> R["apply rule to the tree"]
  R --> M["record invocation,<br/>and whether it changed anything"]
  M --> L
  L -->|"end of the rule list"| C{"strategy"}
  C -->|"Once"| D["done after one pass"]
  C -->|"FixedPoint(n)"| E{"did the tree change?"}
  E -->|"yes, and passes < n"| L
  E -->|"no, tree is stable"| D
  E -->|"passes = n"| W["log a max-iterations warning,<br/>stop anyway"]

The termination test is plan equality, not a change counter:

if (curPlan.fastEquals(lastPlan)) {
  logTrace(s"Fixed point reached for batch ${batch.name} after ${iteration - 1} iterations.")
  continue = false
}

That design is what makes the rule set composable. ConstantFolding can create a literal that lets PruneFilters delete an entire subtree on the next pass, and nobody had to encode that dependency: the batch simply runs again because the tree changed.

Reading the real structure out of a session is more informative than any description of it:

from pyspark.sql import SparkSession

spark = (SparkSession.builder.appName("catalyst").master("local[2]")
         .config("spark.sql.catalogImplementation", "in-memory")
         .getOrCreate())

batches = spark._jsparkSession.sessionState().optimizer().batches()
print(batches.size(), batches.apply(0).name(), batches.apply(0).rules().size())

39 Finish Analysis 1

Scala is friendlier for walking the whole list, and spark-shell has it to hand:

val bs = spark.sessionState.optimizer.batches
println(s"batches=${bs.size}")
println(s"distinct rules=${bs.flatMap(_.rules.map(_.ruleName)).distinct.size}")
println(s"once batches=${bs.count(_.strategy.toString.startsWith("Once"))}")
bs.take(12).foreach(b => println(f"${b.name}%-46s ${b.strategy}%-46s ${b.rules.size}"))

On this build:

batches=39   distinct rules=117   once batches=21   fixed-point batches=18

Finish Analysis                                FixedPoint(1,false,null)         1
Rewrite With expression                        FixedPoint(100, ...maxIterations) 1
Eliminate Distinct                             Once                              1
Inline CTE                                     Once                              1
Union                                          FixedPoint(100, ...maxIterations) 3
LocalRelation early                            FixedPoint(100, ...maxIterations) 3
Pullup Correlated Expressions                  Once                              3
Subquery                                       FixedPoint(1,false,null)          2
Replace Operators                              FixedPoint(100, ...maxIterations) 7
Aggregate                                      FixedPoint(100, ...maxIterations) 2
Operator Optimization before Inferring Filters FixedPoint(100, ...maxIterations) 53
Infer Filters                                  Once                              2

Three things in that listing are worth pausing on.

The big batch is the one people mean by “the optimizer”. Operator Optimization before Inferring Filters holds 53 rules and runs to a fixed point. It contains the whole familiar cast: PushDownPredicates, ColumnPruning, ConstantFolding, BooleanSimplification, CollapseProject, PruneFilters and the rest.

Infer Filters sits between two copies of it. The same 53-rule set runs again afterwards, as Operator Optimization after Inferring Filters, because inferring a new predicate creates new work for the push-down rules. This is the fixed-point idea applied at the batch level.

FixedPoint(1, ...) is not the same as Once. Both run the rules a single time, but Once batches are additionally checked for idempotence in Spark’s own test builds, so a Once rule that changes the tree twice is a bug that gets caught. The iteration cap comes from spark.sql.optimizer.maxIterations, default 100; exceeding it logs a warning in production and throws in testing.

One query, four trees

Here is the whole pipeline on a query with a join, an aggregate, a real predicate and a deliberately silly one.

from pyspark.sql import SparkSession

spark = (SparkSession.builder.appName("catalyst").master("local[2]")
         .config("spark.sql.shuffle.partitions", "4")
         .getOrCreate())

trips = spark.createDataFrame(
    [(i, i % 500, i % 90000, float(i % 4000) / 100.0, "2026-09-0%d" % (i % 9 + 1))
     for i in range(20000)],
    "trip_id bigint, city_id bigint, rider_id bigint, fare double, trip_date string")
cities = spark.createDataFrame(
    [(i, "city-%d" % i, ["IN", "US", "DE"][i % 3]) for i in range(500)],
    "city_id bigint, city_name string, country_code string")

trips.write.mode("overwrite").parquet("/tmp/catalyst/trips")
cities.write.mode("overwrite").parquet("/tmp/catalyst/cities")
spark.read.parquet("/tmp/catalyst/trips").createOrReplaceTempView("trips")
spark.read.parquet("/tmp/catalyst/cities").createOrReplaceTempView("cities")

df = spark.sql("""
  SELECT c.city_name, count(*) AS n
  FROM trips t JOIN cities c ON t.city_id = c.city_id
  WHERE t.fare > 10.0 AND 1 = 1
  GROUP BY c.city_name""")

qe = df._jdf.queryExecution()
print(qe.logical().toString())        # parsed
print(qe.analyzed().toString())
print(qe.optimizedPlan().toString())
print(qe.executedPlan().toString())

Parsed. Nothing is resolved. The leading ticks mark unresolved nodes and attributes, and the tables are UnresolvedRelation:

'Aggregate ['c.city_name], ['c.city_name, 'count(1) AS n#16]
+- 'Filter (('t.fare > 10.0) AND (1 = 1))
   +- 'Join Inner, ('t.city_id = 'c.city_id)
      :- 'SubqueryAlias t
      :  +- 'UnresolvedRelation [trips], [], false
      +- 'SubqueryAlias c
         +- 'UnresolvedRelation [cities], [], false

Analyzed. The ticks are gone. Every attribute now has an expression id (city_id#9L), the views and aliases are explicit, and note what analysis added: cast(10.0 as double) around the literal, because the comparison needs matching types.

Aggregate [city_name#14], [city_name#14, count(1) AS n#16L]
+- Filter ((fare#11 > cast(10.0 as double)) AND (1 = 1))
   +- Join Inner, (city_id#9L = city_id#13L)
      :- SubqueryAlias t
      :  +- SubqueryAlias trips
      :     +- View (`trips`, [trip_id#8L, city_id#9L, rider_id#10L, fare#11, trip_date#12])
      :        +- Relation [trip_id#8L,city_id#9L,rider_id#10L,fare#11,trip_date#12] parquet
      +- SubqueryAlias c
         +- SubqueryAlias cities
            +- View (`cities`, [city_id#13L, city_name#14, country_code#15])
               +- Relation [city_id#13L,city_name#14,country_code#15] parquet

Optimized. This is the interesting one, and five separate rewrites are visible in it at once:

Aggregate [city_name#14], [city_name#14, count(1) AS n#16L]
+- Project [city_name#14]
   +- Join Inner, (city_id#9L = city_id#13L)
      :- Project [city_id#9L]
      :  +- Filter ((isnotnull(fare#11) AND (fare#11 > 10.0)) AND isnotnull(city_id#9L))
      :     +- Relation [trip_id#8L,city_id#9L,rider_id#10L,fare#11,trip_date#12] parquet
      +- Project [city_id#13L, city_name#14]
         +- Filter isnotnull(city_id#13L)
            +- Relation [city_id#13L,city_name#14,country_code#15] parquet

• 1 = 1 has disappeared, folded to true and then dropped from the AND.
• The cast(10.0 as double) is gone, folded into the literal 10.0.
• SubqueryAlias and View wrappers are removed: they carried naming information analysis needed and execution does not.
• A Filter now sits under the join on each side, pushed down from above it.
• isnotnull(...) predicates were added on both join keys, which is the optimizer inserting work because an inner join already discards nulls, and a cheap null check at the scan is worth more than the rows it removes later.

Physical. The scan is where the logical rewrites turn into less IO:

AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[city_name#14], functions=[count(1)], ...)
   +- Exchange hashpartitioning(city_name#14, 4), ENSURE_REQUIREMENTS, [plan_id=76]
      +- HashAggregate(keys=[city_name#14], functions=[partial_count(1)], ...)
         +- Project [city_name#14]
            +- BroadcastHashJoin [city_id#9L], [city_id#13L], Inner, BuildRight, false
               :- Project [city_id#9L]
               :  +- Filter ((isnotnull(fare#11) AND (fare#11 > 10.0)) AND isnotnull(city_id#9L))
               :     +- FileScan parquet [city_id#9L,fare#11] ...
                        PushedFilters: [IsNotNull(fare), GreaterThan(fare,10.0), IsNotNull(city_id)],
                        ReadSchema: struct<city_id:bigint,fare:double>

ReadSchema: struct<city_id:bigint,fare:double> is the payoff. The table has five columns and the scan reads two, because ColumnPruning worked out that nothing else is needed. PushedFilters is the predicate handed to Parquet itself, so row groups whose statistics cannot satisfy fare > 10.0 are never decompressed. And the aggregate is split into partial_count before the exchange and count after it, which is map-side pre-aggregation.

Which rules actually fired

You do not have to infer the responsible rules. Every QueryExecution carries a QueryPlanningTracker that records each rule’s invocations and how many of those changed the tree:

qe = df._jdf.queryExecution()
qe.optimizedPlan()                     # force analysis and optimization
rules = qe.tracker().rules()

it, rows = rules.keysIterator(), []
while it.hasNext():
    k = it.next()
    v = rules.apply(k)
    rows.append((k.split(".")[-1], v.numInvocations(), v.numEffectiveInvocations()))

for name, n, eff in sorted(rows, key=lambda r: -r[2]):
    if eff:
        print(f"{name:<46} invocations={n:<4} effective={eff}")
print("invoked but never effective:", sum(1 for _, _, e in rows if e == 0))

For the query above:

PushDownPredicates                             invocations=6    effective=2
Analyzer$ResolveReferences                     invocations=3    effective=2
ColumnPruning                                  invocations=6    effective=1
ConstantFolding                                invocations=5    effective=1
BooleanSimplification                          invocations=5    effective=1
ResolveTimeZone                                invocations=3    effective=1
Analyzer$ResolveRelations                      invocations=3    effective=1
AnsiTypeCoercion$AnsiCombinedTypeCoercionRule  invocations=3    effective=1
Analyzer$ResolveFunctions                      invocations=3    effective=1
InferFiltersFromConstraints                    invocations=1    effective=1
Optimizer$FinishAnalysis                       invocations=1    effective=1

total distinct rules invoked: 217
invoked but never effective: 206

That table answers the questions from the introduction directly. 1 = 1 went to BooleanSimplification, the casts to ConstantFolding, the two-column read to ColumnPruning, and the isnotnull predicates to InferFiltersFromConstraints. Six invocations of PushDownPredicates with two effective is the fixed point at work: it kept being retried until it had nothing left to move.

The headline number is the last one. 217 rules ran, 11 did something, 206 found nothing to do. Catalyst’s cost is dominated by rules cheaply declining, which is exactly why the TreePattern pruning mentioned earlier exists.

A tour of rules you can see working

Each of these is one small query and its optimized plan, so you can watch a single rule in isolation. spark.sql(q)._jdf.queryExecution().optimizedPlan() is the only tool needed.

Two are worth showing in full because the rewrite is drastic.

PruneFilters deletes the query. An always-false predicate does not become a filter that matches nothing; it becomes an empty relation, and the scan is never planned:

spark.sql("SELECT trip_id FROM trips WHERE 1 = 2") \
     ._jdf.queryExecution().optimizedPlan().toString()

LocalRelation <empty>, [trip_id#0L]

NullPropagation answers without reading anything. count(NULL) is zero by definition, so the aggregate is replaced by a literal and the projection below it is emptied:

Aggregate [0 AS count(NULL)#10L]
+- Project
   +- Relation [trip_id#0L,city_id#1L,rider_id#2L,fare#3,trip_date#4] parquet

LikeSimplification is the one with the most practical value, because StartsWith is a predicate Parquet can evaluate against column statistics and a general LIKE is not. A leading wildcard (LIKE '%city-1') cannot be rewritten that way, which is the real reason the advice about leading wildcards exists.

Turning a rule off

spark.sql.optimizer.excludedRules takes fully qualified rule names. This is the fastest way to find out what a rule does, and it is genuinely useful when a rule misbehaves on your workload.

Q = "SELECT trip_id, fare * (2 + 3) AS f5 FROM trips"

spark.conf.unset("spark.sql.optimizer.excludedRules")
print(spark.sql(Q)._jdf.queryExecution().optimizedPlan().toString())

spark.conf.set("spark.sql.optimizer.excludedRules",
               "org.apache.spark.sql.catalyst.optimizer.ConstantFolding")
print(spark.sql(Q)._jdf.queryExecution().optimizedPlan().toString())

# default
Project [trip_id#0L, (fare#3 * 5.0) AS f5#8]

# ConstantFolding excluded
Project [trip_id#0L, (fare#3 * cast((2 + 3) as double)) AS f5#9]

The unfolded version computes 2 + 3 and a cast for every row. The same trick on PushDownPredicates leaves the filter stranded above the scan in a nested shape rather than combined at the leaf.

Two limits to know. Some rules cannot be excluded: the optimizer keeps a nonExcludableRules list, 25 entries on this build, covering rewrites that produce an invalid plan if skipped rather than merely a slow one. Naming one of those in the config is silently ignored rather than rejected, so an exclusion that appears to do nothing may be one of them.

Writing your own rule

A rule is an object with one method, and spark.experimental.extraOptimizations injects it without building a session extension. This runs in spark-shell:

import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule

// Rewrite x * 1 to x, and x + 0 to x.
object RemoveIdentityArithmetic extends Rule[LogicalPlan] {
  def apply(plan: LogicalPlan): LogicalPlan = plan.transformAllExpressions {
    case Multiply(left, Literal(1, _), _)   => left
    case Multiply(left, Literal(1.0, _), _) => left
    case Add(left, Literal(0, _), _)        => left
    case Add(left, Literal(0.0, _), _)      => left
  }
}

spark.read.parquet("/tmp/catalyst/trips").createOrReplaceTempView("trips")
val q = "SELECT trip_id, fare * 1 AS f, rider_id + 0 AS r FROM trips"

println(spark.sql(q).queryExecution.optimizedPlan)
spark.experimental.extraOptimizations = Seq(RemoveIdentityArithmetic)
println(spark.sql(q).queryExecution.optimizedPlan)

# before
Project [trip_id#0L, (fare#3 * 1.0) AS f#5, rider_id#2L AS r#6L]

# after
Project [trip_id#0L, fare#3 AS f#7, rider_id#2L AS r#8L]

Look closely at the “before” line, because it contains a free lesson. `rider_id

• 0 was **already** simplified to rider_id by Spark's existing rules, while fare * 1 survived as (fare#3 * 1.0). The gap is not an oversight to be smug about: x * 1 is only safe to remove for some type and nullability combinations, and the multiplication here is on a double` where the literal had to be cast. The example is useful precisely because half of it was redundant.

An injected rule lands in its own batch at the end of the pipeline:

Batch(User Provided Optimizers, FixedPoint(100, ...maxIterations),
      List(RemoveIdentityArithmetic))

FixedPoint means your rule is retried until the tree stops changing, so it must be idempotent. A rule that keeps rewriting its own output will spin to maxIterations and log a warning, and in a test build it throws.

For production, spark.experimental.extraOptimizations is a session-local hook. The supported path is SparkSessionExtensions with injectOptimizerRule, set through spark.sql.extensions, which is also how Iceberg and Hudi install their rules.

Testing a rule

A rule is a pure function on a tree, so it needs no cluster to test. Apply it to a plan and compare:

def optimized(sql):
    return spark.sql(sql)._jdf.queryExecution().optimizedPlan().toString()

def assert_rule_effective(sql, rule_name):
    """Fail unless the named rule changed the plan for this query."""
    qe = spark.sql(sql)._jdf.queryExecution()
    qe.optimizedPlan()
    rules = qe.tracker().rules()
    it = rules.keysIterator()
    while it.hasNext():
        key = it.next()
        if key.endswith(rule_name):
            eff = rules.apply(key).numEffectiveInvocations()
            assert eff > 0, f"{rule_name} ran but changed nothing"
            return eff
    raise AssertionError(f"{rule_name} was never invoked")

assert "fare#" in optimized("SELECT fare * 1 AS f FROM trips")   # before injection
print(assert_rule_effective("SELECT trip_id FROM trips WHERE 1 = 2", "PruneFilters"))

Three properties are worth asserting on any rule you write. That it fires at all, which the tracker answers. That it is idempotent, since it runs in a fixed-point batch: applying it to its own output must change nothing. And that it preserves results, which means running the query with and without the rule and comparing the rows, not just the plans. Spark’s own suites do exactly this, with RuleExecutor’s idempotence check on Once batches as the automated version of the second one.

Physical planning: from what to how

The optimized logical plan still says nothing about execution. Join Inner is a requirement, not an algorithm. Turning each logical operator into a physical one is the job of the SparkPlanner, which applies a list of strategies rather than rules:

val st = spark.sessionState.planner.strategies
println(s"strategies=${st.size}")
st.foreach(s => println("  " + s.getClass.getSimpleName.replace("$", "")))

strategies=16
  HiveTableScans          LogicalQueryStageStrategy   PythonEvals
  HiveScripts             DataSourceV2Strategy        V2CommandStrategy
  FileSourceStrategy      DataSourceStrategy          SpecialLimits
  Aggregation             Window                      WindowGroupLimit
  JoinSelection           InMemoryScans               SparkScripts
  BasicOperators

A strategy differs from a rule in what it returns. A rule returns one tree; a strategy returns a list of candidate physical plans for a logical operator, and the planner takes the first that applies. JoinSelection is where BroadcastHashJoinExec and SortMergeJoinExec are chosen between, and FileSourceStrategy is where a scan acquires its PushedFilters and its pruned ReadSchema.

This is also where the first genuinely cost-sensitive decision happens, because JoinSelection compares a size estimate against spark.sql.autoBroadcastJoinThreshold. The strategy ordering, the build-side rules per join type and the full decision procedure are their own subject, taken apart in Apache Spark joins in depth.

Rule-based versus cost-based optimization

This is the distinction that most descriptions of Catalyst blur, and it is measurable.

Rule-based optimization (RBO) applies rewrites that are always improvements. ConstantFolding never makes a plan worse, so no cost model is consulted and no statistics are needed. Every rule in the earlier tour is RBO.

Cost-based optimization (CBO) chooses between alternatives using estimated cardinalities, which means it needs statistics about your data, which means somebody has to compute them.

What statistics Spark actually holds

A fresh table has none, which is worth seeing before anything else:

DESCRIBE EXTENDED fact;
-- Statistics row: absent

ANALYZE TABLE fills in two different things, and the distinction matters:

-- table level: a row count and a size
ANALYZE TABLE fact COMPUTE STATISTICS;

-- column level: per-column cardinality, nulls, bounds, and a histogram
ANALYZE TABLE fact COMPUTE STATISTICS FOR COLUMNS city_id, rider_id, fare;

fact       1779205 bytes, 200000 rows
dim_city   4818 bytes, 400 rows
dim_rider  449573 bytes, 50000 rows

Column statistics are the interesting ones, because selectivity estimation is impossible without them:

DESCRIBE EXTENDED fact city_id;

min                    0
max                    399
num_nulls              0
distinct_count         397
avg_col_len            8
max_col_len            8
histogram              height: 787.4015748031496, num_of_bins: 254
bin_0                  lower_bound: 0.0, upper_bound: 1.0, distinct_count: 2
bin_1                  lower_bound: 1.0, upper_bound: 3.0, distinct_count: 2
bin_2                  lower_bound: 3.0, upper_bound: 4.0, distinct_count: 1

That is an equi-height histogram: 254 bins each holding about 787 rows, so the bin widths vary and dense ranges get more bins. It is built only when spark.sql.statistics.histogram.enabled is true, which is not the default, and it is what lets Catalyst estimate a range predicate rather than guessing.

The same query, with and without CBO

Now the measurement. One filter, explain("cost") twice:

spark.conf.set("spark.sql.cbo.enabled", "true")
spark.sql("SELECT city_id FROM fact WHERE fare > 10.0").explain("cost")

spark.conf.set("spark.sql.cbo.enabled", "false")
spark.sql("SELECT city_id FROM fact WHERE fare > 10.0").explain("cost")

# cbo = true, with column statistics
Project [city_id#1L], Statistics(sizeInBytes=2.3 MiB, rowCount=1.50E+5)
+- Filter (isnotnull(fare#3) AND (fare#3 > 10.0)), Statistics(sizeInBytes=5.7 MiB, rowCount=1.50E+5)
   +- Relation ... fact[...] parquet, Statistics(sizeInBytes=7.6 MiB, rowCount=2.00E+5)

# cbo = false
Project [city_id#1L], Statistics(sizeInBytes=695.0 KiB)
+- Filter (isnotnull(fare#3) AND (fare#3 > 10.0)), Statistics(sizeInBytes=1737.5 KiB)
   +- Relation ... fact[...] parquet, Statistics(sizeInBytes=1737.5 KiB)

Two differences, both important:

• With CBO there is a rowCount, and the filter reduces it, from 200,000 to 150,000. That is a selectivity estimate computed from the histogram.
• Without CBO there is no rowCount at all, and the filter does not change sizeInBytes: 1737.5 KiB in, 1737.5 KiB out.

The second line is the one to remember. Off the default path, a WHERE clause does not make Spark expect fewer rows, so any later decision that depends on size is made as though the filter were not there. That is the mechanism behind a great many “why did it not broadcast” questions.

Join reordering, and the honest result

CostBasedJoinReorder is the rule people mean by “Spark has a CBO”. It sits in a Join Reorder batch with FixedPoint(1) and needs both spark.sql.cbo.enabled and spark.sql.cbo.joinReorder.enabled.

So I wrote the joins in a deliberately bad order, big table first, computed table and column statistics for all three tables, and compared:

Q = """SELECT count(*) FROM dim_rider r
       JOIN fact f     ON f.rider_id = r.rider_id
       JOIN dim_city c ON f.city_id  = c.city_id
       WHERE c.city_name = 'city-7'"""

The resulting join order was identical with CBO on and off, and the tracker says why:

CostBasedJoinReorder   invocations=1  effective=0
ReorderJoin            invocations=5  effective=1

ReorderJoin, which is rule-based and always on, had already moved the joins into a sensible order. By the time the cost-based rule ran there was nothing left for it to improve, so it declined.

That is not a claim that CostBasedJoinReorder never helps. It is a claim about where the value usually comes from: the rule-based rewrites do the heavy lifting, and CBO is a narrow addition that needs statistics you must maintain. Treat “enable CBO and joins get reordered” as something to verify on your own workload with the tracker, rather than as a given.

Codegen, where the tree stops being a tree

The physical plan is the last tree. Tungsten then collapses runs of operators into one generated Java class, so there is no per-row virtual call between a filter and a projection.

spark.sql("SELECT trip_id + 1 AS x FROM trips WHERE fare > 10.0").explain("codegen")

Found 1 WholeStageCodegen subtrees.
== Subtree 1 / 1 (maxMethodCodeSize:359; maxConstantPoolSize:173(0.26% used); numInnerClasses:0) ==
*(1) Project [(trip_id#0L + 1) AS x#9L]
+- *(1) Filter (isnotnull(fare#3) AND (fare#3 > 10.0))
   +- *(1) ColumnarToRow
      +- FileScan parquet [trip_id#0L,fare#3] ...

Generated code:
/* 001 */ public Object generate(Object[] references) {
/* 002 */   return new GeneratedIteratorForCodegenStage1(references);
/* 003 */ }
/* 005 */ // codegenStageId=1
/* 006 */ final class GeneratedIteratorForCodegenStage1 extends
/* 006 */     org.apache.spark.sql.execution.BufferedRowIterator {

The *(1) prefix marks operators fused into codegen stage 1. An operator without the star is not generating code, which is the fastest way to spot a fallback. The maxMethodCodeSize figure matters because the JVM refuses to JIT-compile very large methods, so Spark gives up on codegen for a stage past spark.sql.codegen.hugeMethodLimit and runs it interpreted instead.

Dynamic partition pruning

Dynamic partition pruning (DPP) is the runtime trick for the classic star-schema shape: a partitioned fact table joined to a filtered dimension. The partition values that survive the dimension filter are not known at planning time, so Catalyst plants a subquery that supplies them once the dimension side has been broadcast, and the fact scan then lists only those partitions.

It is on by default:

spark.conf.get("spark.sql.optimizer.dynamicPartitionPruning.enabled")   # true

The shape it looks for:

SELECT sum(s.amount)
FROM   sales s JOIN dates d ON s.day = d.day     -- s.day is a partition column
WHERE  d.is_promo = true                          -- a filter on the dimension

Worth reporting honestly: I could not get it to fire on this local setup. With a 12-partition, 400,000-row partitioned sales table, a 12-row dimension that does get broadcast, statistics computed on both, and adaptive execution disabled, the plan came back with a plain partition filter every time:

join operator    : BroadcastHashJoin [day#2], [day#3], Inner, BuildRight
PartitionFilters : [isnotnull(day#2)]
dynamicpruning in optimized plan: False

The tracker shows the rule ran and declined rather than being absent:

PartitionPruning                inv=1 eff=0
CleanupDynamicPruningFilters    inv=1 eff=0

I tried dynamicPartitionPruning.useStats=false and fallbackFilterRatio=0.01, which are the two knobs on its cost check, and the result did not change. The most likely explanation is that on a table this small the estimated saving never clears the cost of the extra subquery, but I did not establish which condition failed, so I am reporting the observation rather than a cause.

The useful takeaway is the diagnostic, not the disappointment. DPP is a rule like any other, so PartitionPruning with eff=0 in the tracker tells you it considered your query and said no, which is a different problem from it being switched off. The marker to look for when it does apply is dynamicpruning#... inside PartitionFilters, or a SubqueryBroadcast node in the physical plan.

Adaptive query execution

Everything up to here happens before any data moves, from estimates. Adaptive query execution re-plans after each shuffle stage completes, using measured sizes, and it is on by default.

flowchart TB
  A["optimized logical plan"] --> B["physical plan"]
  B --> C["AdaptiveSparkPlan<br/>isFinalPlan=false"]
  C --> D["run the next stage"]
  D --> E["real statistics from<br/>the completed stage"]
  E --> F{"re-optimize the rest"}
  F -->|"coalesce small partitions"| D
  F -->|"split a skewed partition"| D
  F -->|"switch the join strategy"| D
  F -->|"nothing left to run"| G["isFinalPlan=true"]

The thing to know before testing any of this: you have to execute the plan you are holding and then read the same QueryExecution again. Calling df.count() builds a different query, so the plan you inspect afterwards was never the one that ran. Use collect(), or keep a handle on the QueryExecution:

df = spark.sql("SELECT city_id, count(*) c FROM fact GROUP BY city_id")
qe = df._jdf.queryExecution()

print(qe.executedPlan().toString().splitlines()[0])   # before
df.collect()                                          # run THIS plan
print(qe.executedPlan().toString().splitlines()[0])   # after

AdaptiveSparkPlan isFinalPlan=false
AdaptiveSparkPlan isFinalPlan=true

Coalescing partitions. With spark.sql.shuffle.partitions left at 200 and 400 groups to produce, the final plan shows the shuffle being read back as fewer, larger partitions:

+- AQEShuffleRead coalesced
   +- ShuffleQueryStage 0

Switching the join strategy. This is the most dramatic one, because the operator changes after planning. Broadcasting is disabled at plan time and allowed at runtime:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", "10MB")
q = spark.sql("SELECT count(*) FROM fact f JOIN dim_city c ON f.city_id = c.city_id")
qe = q._jdf.queryExecution()
planned = qe.executedPlan().toString()
q.collect()
final = qe.executedPlan().toString()

planned  : SortMergeJoin [city_id#3L], [city_id#10L], Inner
executed : *(3) BroadcastHashJoin [city_id#3L], [city_id#10L], Inner, BuildRight, false

A sort-merge join at plan time became a broadcast hash join at run time, once the measured size of the build side was known. Note that the two thresholds are separate configs, so plan-time and runtime broadcasting can be governed by different numbers.

Splitting a skewed partition. With a key holding most of the rows, AQE divides the oversized partition into several and the join operator says so:

# skewJoin.enabled = true
*(5) SortMergeJoin(skew=true) [city_id#1L], [city_id#6L], Inner
  +- AQEShuffleRead coalesced and skewed

# skewJoin.enabled = false
*(5) SortMergeJoin [city_id#1L], [city_id#6L], Inner
  +- AQEShuffleRead coalesced

(skew=true) on the operator and and skewed on the shuffle read are the two markers. The thresholds are strict and are the usual reason nothing happens: a partition counts as skewed only when it is larger than skewJoin.skewedPartitionFactor (default 5.0) times the median and larger than skewJoin.skewedPartitionThresholdInBytes (default 256MB). On data where no partition reaches 256 MB the rule can never fire, however lopsided the distribution is. Skew, and the six fixes that do not depend on AQE, are taken apart in data skew in Apache Spark, and adaptive execution itself in adaptive query execution in Apache Spark.

Debugging and inspection

Five tools, in the order they are usually worth reaching for.

explain(mode). The modes are not interchangeable:

explain("extended") is the one to start with, because seeing analyzed and optimized next to each other is what makes a missing rewrite obvious.

The rule tracker, shown earlier, which names the rule that changed the plan.

The scan line. ReadSchema and PushedFilters on a FileScan are the only proof that pruning and pushdown reached the file format rather than stopping at a Spark-side Filter.

Rule-level logs. Setting the logger for org.apache.spark.sql.catalyst.rules.RuleExecutor to TRACE makes Spark print the plan change per rule and the fixed point being reached per batch. It is verbose and definitive, and it is the tool for a rule that is firing when you did not expect it.

The SQL tab in the UI. For anything adaptive this beats explain(), because the tab shows the final plan with the actual row counts and spill per operator, where explain() shows the intent.

Production notes

• Read the analyzed and optimized plans together. The diff between them is the optimizer’s entire contribution, and it is where a missing pushdown or a surprise cast becomes obvious.
• Use the tracker rather than guessing. queryExecution.tracker().rules() names the rule that changed your plan, which turns an argument into a lookup.
• Check ReadSchema and PushedFilters on every scan you care about. They are the only proof that pruning and pushdown actually reached the file format.
• Do not expect a Filter to reduce the size estimate. Without column statistics there is no selectivity, so the estimate passes through unchanged and broadcast decisions are made on the unfiltered size.
• Reach for excludedRules to diagnose, not to fix. It is excellent for understanding a rule and a poor permanent setting, and some rules cannot be excluded at all.
• Keep a custom rule idempotent. It runs in a fixed-point batch, so a rule that rewrites its own output spins until the iteration cap.
• Prefer SparkSessionExtensions over extraOptimizations in production. The experimental hook is session-local and Scala-only; the extension point is the one other projects use.
• Compute column statistics, not just table statistics. COMPUTE STATISTICS alone gives a row count; only FOR COLUMNS gives the cardinality and histogram that let a filter reduce an estimate.
• Verify that CBO earned its keep before leaving it on. Enable it, run your query, and check whether CostBasedJoinReorder reports a non-zero effective count. In the test here the rule-based ReorderJoin had already done the work.
• Execute the plan you are holding before reading it back. df.count() builds a different query, so an AQE check done that way inspects a plan that never ran. Use collect() and re-read the same QueryExecution.

Frequently asked questions

Why did a cast appear in my plan that I did not write? Analysis added it, before the optimizer ran. Type coercion is part of making the query meaningful, so comparing a double column to a decimal literal inserts a cast. ConstantFolding then usually removes it again by folding the literal, but only if the cast is on the literal side.

Which rule removed my filter? Ask the tracker. If the filter vanished entirely rather than moving, the usual answers are PruneFilters for an always-true or always-false predicate, and BooleanSimplification for a redundant clause inside an AND or OR.

Is Catalyst cost-based? Barely, and the post measures it. The large majority of rules are unconditional rewrites that are always wins, which is rule-based optimization. Statistics are used for broadcast-join selection and, if you explicitly enable CBO and compute column statistics, for CostBasedJoinReorder. In the join-reordering test above that cost-based rule ran and changed nothing, because the rule-based ReorderJoin had already produced the same order. Treat “Catalyst is a cost-based optimizer” as the wrong mental model.

Why does WHERE not reduce the size estimate? Because without column statistics there is no selectivity to apply, so the estimate passes through the Filter unchanged. Compute ANALYZE TABLE ... COMPUTE STATISTICS FOR COLUMNS and turn on spark.sql.cbo.enabled, and explain("cost") starts showing a rowCount that the filter actually reduces.

Why is dynamic partition pruning not happening? Check the tracker before the config. If PartitionPruning shows inv=1 eff=0 the rule considered your query and declined, which is a cost decision rather than a switched-off feature. It also needs the join key to be an actual partition column of the fact table and, by default, a reusable broadcast of the dimension side.

Why is my predicate not pushed into the scan? Push-down happens in stages: the optimizer moves the Filter down the logical tree, and then the data source decides what it can accept. A predicate on a non-deterministic expression, a UDF, or a column the format cannot filter on stops at the scan boundary and appears in DataFilters but not PushedFilters.

Does the optimizer ever make a plan worse? It can, and the usual mechanism is a bad size estimate rather than a bad rule. The rewrites themselves are equivalence-preserving; the choices that depend on statistics are the ones that go wrong.

How do I see rules running as they fire? Set the log level for org.apache.spark.sql.catalyst.rules.RuleExecutor to TRACE. RuleExecutor logs the fixed point being reached per batch and the plan changes per rule, which is verbose but definitive.

Conclusion

Catalyst is easier to reason about once you stop thinking of it as an optimizer and start thinking of it as a tree rewriter with an execution schedule. There are four trees, each produced from the last. There are batches of small rules, each either run once or run until the tree stops changing. Nearly every rule is a pattern match and a replacement, and nearly every rule declines to apply.

That framing makes the debugging loop concrete. A surprising plan is not a mystery about a black box; it is a question about which rule fired, and the tracker answers it by name. A rule you distrust can be switched off and the difference read straight out of the plan. A rewrite you want that Spark does not do is about ten lines in a Rule[LogicalPlan].

The number worth remembering from all of this is 11 of 217. On an ordinary join and aggregate, eleven rules changed the plan and two hundred and six had nothing to say. The optimizer’s power is not in any single clever rewrite but in having enough small, safe, composable ones that the useful few always get their turn.

References

• Optimizer.scala, whose defaultBatches is the batch list and order quoted above
• RuleExecutor.scala for the fixed-point loop, the iteration cap and the idempotence check
• QueryExecution.scala, where the four phases are the lazy members you printed
• SQL performance tuning for the CBO and adaptive-execution configs named here
• Apache Spark architecture for the scheduler and executor machinery the physical plan is handed to

Trademarks

Apache Spark, Apache Hudi, Apache Iceberg, Apache Parquet and Apache are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries.