The Catalyst optimizer's phases

Description

Every time you write a chain of .select(), .filter(), .join(), or .groupBy(), Spark doesn't execute it exactly as you wrote it. Before touching a single byte of data, that code goes through four distinct phases inside the Catalyst optimizer — the heart of Spark SQL, the same engine that runs both SQL and the DataFrame API: a parsed plan (not yet validated), an analyzed plan (with names and types already resolved), an optimized plan (with optimization rules applied), and finally a physical plan (with concrete execution algorithms chosen). This lesson runs .explain(mode="extended") over exactly the query you already built in module 5 — the join of orders against dim_store, aggregated by store — and reads, phase by phase, what changes between each one.

Connection to the module. This lesson answers the question module 2 left pending since its first .explain(): what exactly happens between writing a line of code and Spark deciding how to execute it? Lesson 3 is going to take this exact same query and read it in the four other .explain() modes that exist besides extended.

An analogy: the GPS, with four stops before setting off

Pick back up this module's introduction's GPS. Before showing you a single route, a real GPS goes through several internal steps, even though you never see them separately: first it interprets what you typed into the search box — an address, exactly as typed, not yet checked for whether it exists; then it validates it against its real map — confirming that street exists, in that city, with that zip code; then it simplifies the route with common-sense rules — if two consecutive stretches go the same direction, it merges them into a single instruction, and discards streets that lead nowhere useful; and finally it picks the mode of transport and the exact maneuver for each stretch — turn here, take the highway there. Catalyst makes, precisely, those same four stops: parsed (interpreting what you wrote, unvalidated), analyzed (confirming it exists and what type it is), optimized (simplifying with rules), physical (choosing the concrete execution algorithm).

Worked example: the four phases, over module 5's query

# catalyst_phases.py
import glob
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField, StringType, IntegerType, DoubleType, TimestampType,
)

spark = SparkSession.builder.appName("kiosko-spark").master("local[*]").getOrCreate()

dim_store_schema = StructType([
    StructField("store_id", StringType(), False),
    StructField("store_name", StringType(), False),
    StructField("city", StringType(), False),
])
dim_store_df = spark.read.csv("dim_store.csv", schema=dim_store_schema, header=True, enforceSchema=False)

orders_schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("store_id", StringType(), False),
    StructField("product_id", StringType(), False),
    StructField("quantity", IntegerType(), False),
    StructField("unit_price", DoubleType(), False),
    StructField("order_ts", TimestampType(), False),
])
orders_df = spark.read.csv(
    sorted(glob.glob("orders_2026-08-*.csv")), schema=orders_schema, header=True, enforceSchema=False,
)
fact_orders_df = orders_df.withColumn("revenue", F.col("quantity") * F.col("unit_price"))

# The same query from module 5: join against dim_store, aggregated by store.
q = (
    fact_orders_df
    .join(dim_store_df, "store_id")
    .groupBy("store_id")
    .agg(F.round(F.sum("revenue"), 2).alias("total_revenue"))
)

q.explain(mode="extended")

rows = {r["store_id"]: r["total_revenue"] for r in q.orderBy("store_id").collect()}
print("\nresult:", rows)
assert rows == {"S01": 38.3, "S02": 38.8, "S03": 29.05}
print("assert OK: matches the known total from module 1")

spark.stop()

What to expect. Running python3 catalyst_phases.py, the output is exactly this (executed in this run, PySpark 4.2.0):

== Parsed Logical Plan ==
'Aggregate ['store_id], ['store_id, 'round('sum('revenue), 2) AS total_revenue#12]
+- Project [store_id#4, order_id#3, product_id#5, quantity#6, unit_price#7, order_ts#8, revenue#10, store_name#1, city#2]
   +- Join Inner, (store_id#4 = store_id#0)
      :- Project [order_id#3, store_id#4, product_id#5, quantity#6, unit_price#7, order_ts#8, (cast(quantity#6 as double) * unit_price#7) AS revenue#10]
      :  +- Relation [order_id#3,store_id#4,product_id#5,quantity#6,unit_price#7,order_ts#8] csv
      +- Relation [store_id#0,store_name#1,city#2] csv

== Analyzed Logical Plan ==
store_id: string, total_revenue: double
Aggregate [store_id#4], [store_id#4, round(sum(revenue#10), 2) AS total_revenue#12]
+- Project [store_id#4, order_id#3, product_id#5, quantity#6, unit_price#7, order_ts#8, revenue#10, store_name#1, city#2]
   +- Join Inner, (store_id#4 = store_id#0)
      :- Project [order_id#3, store_id#4, product_id#5, quantity#6, unit_price#7, order_ts#8, (cast(quantity#6 as double) * unit_price#7) AS revenue#10]
      :  +- Relation [order_id#3,store_id#4,product_id#5,quantity#6,unit_price#7,order_ts#8] csv
      +- Relation [store_id#0,store_name#1,city#2] csv

== Optimized Logical Plan ==
Aggregate [store_id#4], [store_id#4, round(sum(revenue#10), 2) AS total_revenue#12]
+- Project [store_id#4, revenue#10]
   +- Join Inner, (store_id#4 = store_id#0)
      :- Project [store_id#4, (cast(quantity#6 as double) * unit_price#7) AS revenue#10]
      :  +- Filter isnotnull(store_id#4)
      :     +- Relation [order_id#3,store_id#4,product_id#5,quantity#6,unit_price#7,order_ts#8] csv
      +- Project [store_id#0]
         +- Filter isnotnull(store_id#0)
            +- Relation [store_id#0,store_name#1,city#2] csv

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[store_id#4], functions=[sum(revenue#10)], output=[store_id#4, total_revenue#12])
   +- Exchange hashpartitioning(store_id#4, 200), ENSURE_REQUIREMENTS, [plan_id=40]
      +- HashAggregate(keys=[store_id#4], functions=[partial_sum(revenue#10)], output=[store_id#4, sum#24])
         +- Project [store_id#4, revenue#10]
            +- BroadcastHashJoin [store_id#4], [store_id#0], Inner, BuildRight, false, false
               :- Project [store_id#4, (cast(quantity#6 as double) * unit_price#7) AS revenue#10]
               :  +- Filter isnotnull(store_id#4)
               :     +- FileScan csv [store_id#4,quantity#6,unit_price#7] Batched: false, DataFilters: [isnotnull(store_id#4)], Format: CSV, ...
               +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, false]),false), [plan_id=35]
                  +- Filter isnotnull(store_id#0)
                     +- FileScan csv [store_id#0] Batched: false, DataFilters: [isnotnull(store_id#0)], Format: CSV, ...

result: {'S01': 38.3, 'S02': 38.8, 'S03': 29.05}
assert OK: matches the known total from module 1

Four blocks, over exactly the same query. Read each one slowly before continuing — this lesson's "going deeper" section explains what changed between each consecutive pair, and why.

Diagram: the four phases, in order

flowchart TD
    A["Python code\n.join().groupBy().agg()"] --> B["Parsed Logical Plan\nunvalidated -- names with a quote mark\n('store_id, 'sum('revenue))"]
    B --> C["Analyzed Logical Plan\nnames and types resolved --\nstore_id: string, total_revenue: double"]
    C --> D["Optimized Logical Plan\nrules applied --\ncolumn pruning, predicate pushdown"]
    D --> E["Physical Plan\nconcrete algorithms --\nHashAggregate, BroadcastHashJoin, Exchange"]
    E --> F["AdaptiveSparkPlan\nre-optimizes in real time\n-- this module's lesson 4"]

    style B fill:#f96,stroke:#333
    style C fill:#fc6,stroke:#333
    style D fill:#9c6,stroke:#333
    style E fill:#69c,stroke:#333

Going deeper: what changes, exactly, between each pair of phases

From Parsed to Analyzed: resolving names and types. Look at the Parsed Logical Plan's first line: 'Aggregate ['store_id], ['store_id, 'round('sum('revenue), 2) AS total_revenue#12]. Every symbol with a quote mark in front ('store_id, 'sum, 'revenue) is an unresolved reference — Catalyst hasn't yet confirmed a column named store_id exists, nor that sum is a valid function over revenue's type. The Parsed Logical Plan is, literally, a direct translation of your Python code into a tree structure, with no verification yet. The Analyzed Logical Plan is the result of Catalyst walking that tree against the real catalog of columns and types — the StructTypes you declared when reading each CSV — and resolving every reference. The most visible proof this already happened: the line store_id: string, total_revenue: double, which shows up only in the analyzed plan, never in the parsed one — it's the output schema, already confirmed, something Catalyst couldn't know without first resolving each column.

From Analyzed to Optimized: pruning and pushing down filters. This is where the plan changes shape, not just content. Compare the two Projects right after the Join: in the analyzed plan, that Project keeps all nine complete columns — store_id, order_id, product_id, quantity, unit_price, order_ts, revenue, store_name, city — even though the query only needs two (store_id to group by, revenue to sum). In the optimized plan, that same Project shrank down to [store_id#4, revenue#10] — column pruning: Catalyst discarded, as soon as it could, any column no later step needs. Also notice the two new Filter isnotnull(...) nodes, one before each side of the Join — predicate pushdown: since an INNER JOIN discards any row with a null key anyway, Catalyst pushes that filter as close to the read as possible, so it doesn't load or process rows that were going to get discarded later regardless.

From Optimized to Physical: choosing the concrete algorithm. The optimized plan still speaks in the abstract: it says Aggregate and Join, without specifying how to run them. The physical plan replaces each of those logical nodes with a concrete algorithm an executor can actually run: Aggregate turns into two HashAggregates (one partial, per partition, and one final, after the Exchange — the same two-step aggregation pattern you already saw in module 4), and Join turns into BroadcastHashJoin with its BroadcastExchange — the same decision module 5 explained in depth, applied here once more because dim_store is still tiny. Also showing up, named for the first time in this guide, is the node wrapping the entire physical plan: AdaptiveSparkPlan isFinalPlan=false. That isFinalPlan=false is the mark that this physical plan can still change — this module's lesson 4 explains exactly when and why.

The fifth phase, not yet executed: codegen. After the physical plan gets fixed, Spark takes one more step that doesn't show up in mode="extended": it compiles several consecutive physical operators — a Filter followed by a Project, for example — into a single class of JVM bytecode, generated at runtime, instead of interpreting each operator separately, row by row. This technique is called whole-stage code generation, and it's why you saw the * symbol before some operators in earlier plans in this guide (*(1) Filter, for example) — that asterisk mark shows which operators got fused into a single piece of compiled code. This module's lesson 3 shows it with real evidence, using .explain(mode="codegen").

Common mistakes

Reading the Parsed Logical Plan's quote marks as a syntax error. What happens: someone sees 'store_id, with a quote mark in front, and assumes something went wrong in the code — as if Spark didn't recognize the column name. Why it happens: in Python, a stray quote mark before an identifier means nothing on its own, so it's easy to read it as noise or a formatting error. How to spot it: if your reading of the Parsed Logical Plan includes the word "error," check whether the rest of the pipeline — the Analyzed Logical Plan onward — resolved those same names with no problem. How to fix it: the quote mark is Catalyst's standard notation (inherited from Spark internally) for marking a reference that's not yet resolved — it's exactly what you'd expect to see in the first phase, before Catalyst confirms the column exists. If the quote mark still shows up in the Analyzed Logical Plan, that's a real problem: it means Spark couldn't resolve that reference.

Assuming the Optimized Logical Plan is what Spark actually executes. What happens: someone reads the optimized plan, sees Filter and Project in the right places, and concludes they already understand how the query is going to run — without checking the Physical Plan. Why it happens: "optimized" sounds like the final version, the one that counts, and it's easy to stop there. How to spot it: if your explanation of this query doesn't mention any concrete algorithm name — HashAggregate, BroadcastHashJoin, Exchange — you stayed at a logical level, not the real execution level. How to fix it: the Optimized Logical Plan is still a logical plan — it describes what needs to happen (group, join, filter), not how. Only the Physical Plan describes the concrete algorithms the executors run; it's the only one of the four phases that determines the real work.

Not distinguishing "column pruning" from simply "writing fewer columns in the code." What happens: someone concludes that, to achieve the same effect as the optimized plan's Project [store_id#4, revenue#10], they'd need to rewrite their code to manually select only those two columns before the .join(). Why it happens: seeing the reduced column list in the plan gives the impression you have to replicate that reduction by hand in the Python code. How to spot it: if your code has a manual .select() right before every .join() "to help Spark," you're probably doing work Catalyst already does on its own. How to fix it: column pruning is automatic — Catalyst analyzes the query's complete tree and discards any column no later step needs, with no need for you to anticipate it in your code. You can (and generally should) write your .join() with the complete DataFrame, as is, and trust the optimized plan is going to reduce it — verifying this with .explain(), as this lesson did, is the right way to confirm it, not rewriting the code by hand.

Exercises

Exercise 1 — Count how many columns survive in each Project after the Join, across the four phases. Without running anything yet, count by hand how many columns show up in the Project immediately after the Join/BroadcastHashJoin node in each of the four phases in this lesson's worked example. Verify your count against the text already shown.

# Programmatic verification of the count, without relying on counting by hand
plans = {
    "Parsed/Analyzed": ["store_id", "order_id", "product_id", "quantity", "unit_price", "order_ts", "revenue", "store_name", "city"],
    "Optimized/Physical": ["store_id", "revenue"],
}
for name, cols in plans.items():
    print(f"{name}: {len(cols)} columns")

Parsed/Analyzed: 9 columns
Optimized/Physical: 2 columns

Exercise 2 — Predict what would happen to the optimized plan's Filter isnotnull(...) if the JOIN were LEFT OUTER instead of INNER. Without running anything, reason it out: a LEFT OUTER JOIN keeps the left side's rows even without a match on the right side — including rows with a null key. Would it still make sense for Catalyst to add the same Filter isnotnull(store_id) before the join's left side?

Exercise 3 — Explain, without code, why AdaptiveSparkPlan isFinalPlan=false shows up wrapping the ENTIRE physical plan, not just part of it. In 2-3 sentences, explain why this node wraps the Physical Plan's complete tree, instead of showing up only around the Exchange or the Join.

Summary and next step

This lesson showed, with .explain(mode="extended") over module 5's same query, the four phases Catalyst goes through before Spark runs any real work: Parsed (unresolved references, marked with a quote), Analyzed (names and types already confirmed against the real schema), Optimized (rules applied — column pruning, predicate pushdown, visible as concrete changes in the tree's shape), and Physical (concrete execution algorithms, wrapped in AdaptiveSparkPlan). You also named, without running it yet, the fifth phase — whole-stage code generation — which lesson 3 is going to show with real evidence.

Before moving on you should be able to: explain what a quote mark in front of a column name means in the Parsed Logical Plan; identify, in a real plan, at least one example of column pruning and one of predicate pushdown; and explain why the Optimized Logical Plan still isn't what the executors run.

Lesson 3 takes this exact same query and reads it in the four other .explain() modes that exist — simple, cost, formatted, codegen — showing what new information each one adds that extended doesn't show.

Resources

• Apache Spark — SQL Performance Tuning (the central reference for Catalyst, .explain(mode=...), and Adaptive Query Execution, the foundation for this entire module). spark.apache.org/docs/latest/sql-performance-tuning.html.
• Apache Spark — SQL Programming Guide (DataFrames/Datasets as the interface Catalyst optimizes, with "richer optimizations" than the RDD API — already cited in module 2). spark.apache.org/docs/latest/sql-programming-guide.html.
• This guide's DESIGN doc (spark-and-distributed-processing-guide/DISENO.md) — this lesson's exact specification: Catalyst's phases read with explain(mode="extended") over module 5's query.