Openingloading the content…

Loadingone moment…

PySpark

Spark SQL & the Catalyst Optimizer

SQL over DataFrames and temp views, and how Catalyst turns a query into a plan: parse → analyze → optimize → physical, with pushdown and pruning.

⏱ 13 min readTopics chapter readerLevel · Structured APIs

01 · Orientation

Start here

⏱ 6 min · Topic 1 of 11

You know that Spark is lazy, and that explain() prints the plan. You also know that SQL and the DataFrame API give "usually the same plan". The engine behind that is called Catalyst.

Catalyst is the part of Spark that turns your query into a plan. It rewrites the query so Spark reads and moves less data. It also chooses the join type.

Seniors are often handed a physical plan and asked to read it.

Where did Catalyst change the query? Where are the shuffles? Why is this join sort-merge?

This chapter has four parts.

The steps Catalyst takes, and the rules it applies.
The explain modes, and how to read a physical plan from the bottom.
Why Spark chose the join it did, and what statistics change.
Python UDFs (user-defined functions): your own Python code, which Catalyst cannot see into.

We use the shop again. The paid revenue by customer is Asha 65.00, Carla 15.50 and Dev 80.00.

02 · One plan

SQL and DataFrames, one plan

⏱ 4 min · Topic 2 of 11

You met temp views and spark.sql() in "DataFrames & Datasets: The Structured APIs". There, both versions printed the same rows. Here you check that they make the same plan.

Compare the plans of the SQL and the DataFrame versionworked example

SQL

Input data

orders6 rows

order_id	customer_id	status	amount	order_date
101	1	paid	40.00	2026-01-03
102	1	paid	25.00	2026-01-10
103	2	refunded	60.00	2026-01-11
104	3	paid	15.50	2026-01-15
105	3	pending	NULL	2026-01-20
106	4	paid	80.00	2026-01-21

orders.createOrReplaceTempView("orders")

sql_df = spark.sql("SELECT customer_id, SUM(amount) AS revenue FROM orders WHERE status = 'paid' GROUP BY customer_id")
api_df = orders.filter("status = 'paid'").groupBy("customer_id").agg(F.sum("amount").alias("revenue"))

sql_df.explain("extended")
api_df.explain("extended")

How the plans compare (both queries return 1 → 65.00, 3 → 15.50, 4 → 80.00)

Plan	SQL version	DataFrame version
Parsed	Every name has a quote mark	Fewer quote marks: each step was checked when you wrote it
Analyzed	Has extra lines for the view name	No view lines
Optimized	Same tree	Same tree (only the # numbers differ)
Physical	Same tree	Same tree

The four paid orders are 101, 102, 104 and 106. Customer 1 has 40.00 + 25.00 = 65.00. A # number is an id that Spark gives one column.

The temporary view lives only as long as the Spark session. The next sections read these plans line by line.

The parsed and analyzed plans differ a little. From the optimized plan on, the trees match. So for this query, the speed is the same.

Remember

You can mix SQL and DataFrame steps in one job. Use SQL for long joins, and DataFrame code for steps you build in a loop.

03 · The optimizer

The Catalyst pipeline

⏱ 6 min · Topic 3 of 11

Catalyst works in five steps. Each step takes the result of the step before it.

Parse: your text becomes an unresolved logical plan. The names are not checked yet.
Analysis: Spark looks the names up in the catalog, its list of tables and columns. It adds the types.
Optimization: rules rewrite the logical plan into an optimized logical plan.
Physical planning: Spark chooses real operators, such as the join type, and picks one plan.
Code generation: Spark combines simple steps into one piece of Java code. This is called whole-stage code generation.

A logical plan says what to do. A physical plan says how to do it. A plan is a tree: each line is one operation, and the lines below it feed it.

Now look at all four plans for one query.

explain("extended") for two filters (simplified)worked example

SQL

Input data

orders6 rows

order_id	customer_id	status	amount	order_date
101	1	paid	40.00	2026-01-03
102	1	paid	25.00	2026-01-10
103	2	refunded	60.00	2026-01-11
104	3	paid	15.50	2026-01-15
105	3	pending	NULL	2026-01-20
106	4	paid	80.00	2026-01-21

orders.filter("status = 'paid'") \
      .filter("order_id > 100 + 3") \
      .select("customer_id", "amount") \
      .explain("extended")

== Parsed Logical Plan ==
'Project ['customer_id, 'amount]
+- Filter (order_id#0 > (100 + 3))
   +- Filter (status#2 = paid)
      +- Relation [order_id#0,customer_id#1,status#2,amount#3,order_date#4] parquet

== Analyzed Logical Plan ==
Project [customer_id#1, amount#3]
+- Filter (order_id#0 > (100 + 3))
   +- Filter (status#2 = paid)
      +- Relation [order_id#0,customer_id#1,status#2,amount#3,order_date#4] parquet

== Optimized Logical Plan ==
Project [customer_id#1, amount#3]
+- Filter (isnotnull(status#2) AND isnotnull(order_id#0) AND (status#2 = paid) AND (order_id#0 > 103))
   +- Relation [order_id#0,customer_id#1,status#2,amount#3,order_date#4] parquet

== Physical Plan ==
*(1) Project [customer_id#1, amount#3]
+- *(1) Filter (isnotnull(status#2) AND isnotnull(order_id#0) AND (status#2 = paid) AND (order_id#0 > 103))
   +- *(1) ColumnarToRow
      +- FileScan parquet [order_id#0,customer_id#1,status#2,amount#3]
           PushedFilters: [IsNotNull(status), IsNotNull(order_id), EqualTo(status,paid), GreaterThan(order_id,103)]

The query returns 2 rows (orders 104 and 106)

customer_id	amount
3	15.50
4	80.00

Order 103 is refunded, and 101 and 102 have order_id 103 or less. Order 105 is pending.

This text is simplified: real output has more detail and different numbers after the # signs. The number after # is an id for one column.

In the parsed plan, a quote mark before a name means "not checked yet". In DataFrame code, Spark checks each step when you write it. So only the last step, the select, still has quotes.

A typo in a column name fails at that line, before any action.

The optimized plan has one Filter, not two. It also shows 103, not 100 + 3. The next section names each change.

04 · The optimizer

Rules that rewrite the plan

⏱ 5 min · Topic 4 of 11

Catalyst has many rules. Each rule looks for one shape in the plan and rewrites it. A predicate is a test that keeps or drops a row.

The plan above shows four rules at work.

Rule	What it does	In the plan above
Constant folding	Computes a fixed expression once, before the job runs	100 + 3 became 103
Filter combining	Merges two Filter lines into one	Two Filters became one with AND
Column pruning	Keeps only the columns the query needs	order_date is gone from FileScan
Predicate pushdown	Moves the filter below other steps, then into the file reader	PushedFilters on FileScan

Pushdown lets the Parquet reader skip data. The skipped data is never loaded.

Look at the Filter line in the optimized plan. Catalyst also added isnotnull(status) and isnotnull(order_id). Catalyst adds them because a NULL never passes status = paid or order_id > 103.

Spark keeps the Filter line even after it pushes the filter to the scan. The scan is allowed to skip data, but it may still return rows that do not match.

05 · Reading plans

The explain modes

⏱ 5 min · Topic 5 of 11

The function explain() takes a mode. Each mode answers a different question. The call explain(True) is the same as explain("extended").

Mode	What it prints	Use it to
explain() or explain("simple")	The physical plan only	Take a quick look
explain("extended")	The four plans: parsed, analyzed, optimized, physical	See what Catalyst changed
explain("formatted")	An outline, then the details of each step	Read a long plan
explain("cost")	The optimized plan with a size on each step. Row counts need statistics and the cost-based optimizer (CBO).	Check the statistics
explain("codegen")	The Java code made by whole-stage code generation	Debug code generation

The formatted mode also numbers each step. Look at it for the revenue query.

explain("formatted") for paid revenue (simplified)worked example

SQL

== Physical Plan ==
AdaptiveSparkPlan (7)
+- HashAggregate (6)
   +- Exchange (5)
      +- HashAggregate (4)
         +- Project (3)
            +- Filter (2)
               +- Scan parquet orders (1)

(1) Scan parquet orders
Output [3]: [customer_id#1, status#2, amount#3]
PushedFilters: [IsNotNull(status), EqualTo(status,paid)]
ReadSchema: struct<customer_id:int,status:string,amount:decimal(10,2)>

(2) Filter
Condition : (isnotnull(status#2) AND (status#2 = paid))

Only steps (1) and (2) are shown. Step (5) is the Exchange. After the job runs, helper lines such as ColumnarToRow appear, so the numbers change.

The cost mode adds Statistics(...) to the end of each line of the optimized plan.

The section "Statistics and the cost model" shows the cost mode in use.

06 · Reading plans

Read a physical plan

⏱ 7 min · Topic 6 of 11

You read the paid revenue plan from the bottom in "Transformations, Actions & Lazy Evaluation". Here you follow the shop rows through it. Then you read the parts that chapter skipped.

Here is the plan for the paid revenue query. It is simplified: real output has more detail.

Physical plan for paid revenue by customer (simplified)worked example

SQL

Input data

orders6 rows

order_id	customer_id	status	amount	order_date
101	1	paid	40.00	2026-01-03
102	1	paid	25.00	2026-01-10
103	2	refunded	60.00	2026-01-11
104	3	paid	15.50	2026-01-15
105	3	pending	NULL	2026-01-20
106	4	paid	80.00	2026-01-21

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[customer_id#1], functions=[sum(amount#3)])
   +- Exchange hashpartitioning(customer_id#1, 200), ENSURE_REQUIREMENTS, [plan_id=27]
      +- HashAggregate(keys=[customer_id#1], functions=[partial_sum(amount#3)])
         +- Project [customer_id#1, amount#3]
            +- Filter (isnotnull(status#2) AND (status#2 = paid))
               +- FileScan parquet [customer_id#1,status#2,amount#3]
                    PushedFilters: [IsNotNull(status), EqualTo(status,paid)]
                    ReadSchema: struct<customer_id:int,status:string,amount:decimal(10,2)>

What each line does, from the bottom

Line	What it does on the shop data
FileScan	Reads 3 of 5 columns. The columns order_id and order_date are skipped.
Filter	Keeps orders 101, 102, 104 and 106.
Project	Keeps customer_id and amount.
HashAggregate (partial)	Each task adds up its own rows, per customer.
Exchange	Moves the partial sums so each customer is in one place.
HashAggregate (final)	Adds the partial sums: 65.00, 15.50 and 80.00.

ReadSchema has status because the Filter needs it. The partial step means the Exchange moves sums, not orders.

The chapter "The Shuffle: Spark's Performance Killer" explains the Exchange. The number 200 is the default value of spark.sql.shuffle.partitions.

Some lines of this plan need a short explanation.

In the plan	What it means
AdaptiveSparkPlan isFinalPlan=false	Adaptive Query Execution (AQE) is on. Spark adds this line when the plan has an Exchange, a subquery, a join or a groupBy. AQE may still change the plan while the job runs.
#1, #3	Ids for columns. The same id in two lines is the same column.
hashpartitioning(customer_id#1, 200)	Each row goes to one of 200 partitions, chosen by a hash of customer_id.
ENSURE_REQUIREMENTS	Spark added this Exchange because the final aggregate needs each customer in one place.
ReadSchema	The columns and types the scan reads from the file.

Lines with the same *(n) mark run as one generated function. When a plan starts with AdaptiveSparkPlan, explain() shows no *(n) marks before the job runs. After you run df.collect(), call df.explain() again: the plan says isFinalPlan=true and shows the marks.

The Exchange never has a *(n) mark. It is the border between two generated functions.

Count the Exchange lines. Each one is a shuffle. This plan has one.

Interview question

The question: "Explain this plan to me, step by step." A good answer reads from the bottom. Say what the scan reads, which filters were pushed, where the Exchange is, and what each aggregate does.

07 · Join strategy

Why this join is sort-merge

⏱ 6 min · Topic 7 of 11

Choosing the join is part of physical planning. Catalyst picks one strategy for each join. You read the choice from the join line.

A sort-merge join shuffles both tables by the join key and sorts each side. Then it reads the two sorted sides together and matches the keys. A broadcast join copies a small table to every executor, so the big table does not move.

Take paid orders joined to customers. The shop tables are tiny, so Spark would broadcast the customers table.

The first line below turns broadcast off. Then the plan shows a sort-merge join. It is simplified.

Paid orders joined to customers (simplified)worked example

SQL

Input data

orders6 rows

order_id	customer_id	status	amount	order_date
101	1	paid	40.00	2026-01-03
102	1	paid	25.00	2026-01-10
103	2	refunded	60.00	2026-01-11
104	3	paid	15.50	2026-01-15
105	3	pending	NULL	2026-01-20
106	4	paid	80.00	2026-01-21

customers5 rows

customer_id	name	country
1	Asha	IN
2	Ben	GB
3	Carla	US
4	Dev	NULL
5	Omar	IN

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)   # turn off broadcast joins

orders.filter("status = 'paid'") \
      .join(customers, "customer_id") \
      .select("name", "amount") \
      .explain()

AdaptiveSparkPlan isFinalPlan=false
+- Project [name, amount]
   +- SortMergeJoin [customer_id], [customer_id], Inner
      :- Sort [customer_id ASC NULLS FIRST], false, 0
      :  +- Exchange hashpartitioning(customer_id, 200)
      :     +- Filter ((isnotnull(status) AND (status = paid)) AND isnotnull(customer_id))
      :        +- FileScan parquet orders
      +- Sort [customer_id ASC NULLS FIRST], false, 0
         +- Exchange hashpartitioning(customer_id, 200)
            +- Filter isnotnull(customer_id)
               +- FileScan parquet customers

The query returns 4 rows

name	amount
Asha	40.00
Asha	25.00
Carla	15.50
Dev	80.00

Omar has no orders, so Omar is not in the result of the inner join. Ben has only a refunded order, which the Filter removed.

Two Exchange lines mean two shuffles. A broadcast join would show BroadcastExchange on the customers side and no shuffle of orders.

Catalyst chooses sort-merge when it cannot broadcast either side. There are a few common reasons.

Both tables are bigger than spark.sql.autoBroadcastJoinThreshold (10 MB by default), or you set it to -1.
Spark cannot estimate the small side’s size, so it treats that side as very big.
The join keeps every row of the small side, so Spark cannot broadcast it. An example is a left join with the small table on the left.

The chapter "Joins in Spark" explains each strategy. In a plan, find the join line first. Then look at the two branches below it to see which Exchange lines feed it.

08 · Statistics

Statistics and the cost model

⏱ 5 min · Topic 8 of 11

Most Catalyst rules do not need numbers. They run on every query. Some choices need size estimates, such as the join order and the join type.

The cost-based optimizer, or CBO, uses statistics to compare plans. Statistics are facts about a table: row count, size, and facts about each column. You collect them with the ANALYZE TABLE command.

Collect statistics for orders, then ask for the costworked example

SQL

Input data

orders6 rows

order_id	customer_id	status	amount	order_date
101	1	paid	40.00	2026-01-03
102	1	paid	25.00	2026-01-10
103	2	refunded	60.00	2026-01-11
104	3	paid	15.50	2026-01-15
105	3	pending	NULL	2026-01-20
106	4	paid	80.00	2026-01-21

# orders must be a saved table, for example from saveAsTable("orders")
spark.catalog.dropTempView("orders")   # the temp view from earlier would hide the table
spark.sql("ANALYZE TABLE orders COMPUTE STATISTICS FOR COLUMNS customer_id, status, amount")
spark.conf.set("spark.sql.cbo.enabled", "true")
spark.table("orders").filter("status = 'paid'").explain("cost")

# each line of the optimized plan now ends like this
Filter (...), Statistics(sizeInBytes=<bytes>, rowCount=<rows>)

Facts that the command stores for orders

Fact	Value on the shop data
Row count	6
Distinct status values	3 (paid, refunded, pending)
Distinct customer_id values	4 (1, 2, 3, 4)
NULL amounts	1 (order 105)

The command stores the row count, the size, and facts for each column you list. The distinct counts are estimates.

With the counts, Spark can estimate how many rows status = paid keeps. It uses the estimate to compare join orders.

The setting spark.sql.cbo.enabled is false by default in Spark 3.x and 4.x. Without it, the cost mode shows only sizeInBytes. Join reordering also needs spark.sql.cbo.joinReorder.enabled.

Statistics get out of date when the table changes. Run the command again after big loads.

AQE uses real sizes while the job runs. See the chapter "Adaptive Query Execution & Cost-Based Optimization".

09 · Limits

When Catalyst cannot help

⏱ 7 min · Topic 9 of 11

Catalyst can only rewrite what it can read. A Python UDF, or user-defined function, is your own Python code.

Catalyst cannot look inside a UDF. It sees only what goes in and what comes out. We call this a black box.

So it cannot push a UDF filter to the scan. Spark also has to send each row to a Python process and back. Look at the same filter written two ways.

A Python UDF against a built-in filter (simplified)worked example

SQL

Input data

orders6 rows

order_id	customer_id	status	amount	order_date
101	1	paid	40.00	2026-01-03
102	1	paid	25.00	2026-01-10
103	2	refunded	60.00	2026-01-11
104	3	paid	15.50	2026-01-15
105	3	pending	NULL	2026-01-20
106	4	paid	80.00	2026-01-21

@F.udf("boolean")
def is_paid(status):
    return status == "paid"

orders.filter(is_paid("status"))          # Python UDF
orders.filter(F.col("status") == "paid")   # built-in

-- UDF plan
*(2) Filter pythonUDF0#9
+- BatchEvalPython [is_paid(status#2)], [pythonUDF0#9]
   +- *(1) ColumnarToRow
      +- FileScan parquet [order_id,customer_id,status,amount,order_date]
           PushedFilters: []

-- built-in plan
*(1) Filter (isnotnull(status#2) AND (status#2 = paid))
+- *(1) ColumnarToRow
   +- FileScan parquet [order_id,customer_id,status,amount,order_date]
        PushedFilters: [IsNotNull(status), EqualTo(status,paid)]

Both versions return 4 rows

order_id	customer_id	status	amount	order_date
101	1	paid	40.00	2026-01-03
102	1	paid	25.00	2026-01-10
104	3	paid	15.50	2026-01-15
106	4	paid	80.00	2026-01-21

The result is the same. The plan is not. The UDF plan has BatchEvalPython and an empty PushedFilters.

Both plans read all five columns, because the query keeps every column. Catalyst knows is_paid needs only status. What it cannot read is the test inside is_paid.

Use a built-in function when one exists. The module pyspark.sql.functions has many. A built-in function is visible to Catalyst, so rules and pushdown still work.

A pandas UDF uses pandas, a Python library for tables. Spark sends batches of rows to it through Apache Arrow, a fast column format.

The plan shows ArrowEvalPython. A pandas UDF is often faster than a Python UDF, but it has the same limit: no pushdown.

The chapter "UDFs & the Pandas API on Spark" covers both kinds in detail.

Watch out

A UDF inside a filter stops that filter from reaching the scan. On a big Parquet table, you read much more data. Look for BatchEvalPython or ArrowEvalPython in the plan.

Openingloading the content…

PySpark

Spark SQL & the Catalyst Optimizer

SQL over DataFrames and temp views, and how Catalyst turns a query into a plan: parse → analyze → optimize → physical, with pushdown and pruning.

⏱ 13 min readTopics chapter readerLevel · Structured APIs

01 · Orientation

Start here

⏱ 6 min · Topic 1 of 11

You know that Spark is lazy, and that explain() prints the plan. You also know that SQL and the DataFrame API give "usually the same plan". The engine behind that is called Catalyst.

Catalyst is the part of Spark that turns your query into a plan. It rewrites the query so Spark reads and moves less data. It also chooses the join type.

Seniors are often handed a physical plan and asked to read it.

Where did Catalyst change the query? Where are the shuffles? Why is this join sort-merge?

This chapter has four parts.

The steps Catalyst takes, and the rules it applies.
The explain modes, and how to read a physical plan from the bottom.
Why Spark chose the join it did, and what statistics change.
Python UDFs (user-defined functions): your own Python code, which Catalyst cannot see into.

We use the shop again. The paid revenue by customer is Asha 65.00, Carla 15.50 and Dev 80.00.

02 · One plan

SQL and DataFrames, one plan

⏱ 4 min · Topic 2 of 11

You met temp views and spark.sql() in "DataFrames & Datasets: The Structured APIs". There, both versions printed the same rows. Here you check that they make the same plan.

Compare the plans of the SQL and the DataFrame versionworked example

SQL

Input data

orders6 rows

order_id	customer_id	status	amount	order_date
101	1	paid	40.00	2026-01-03
102	1	paid	25.00	2026-01-10
103	2	refunded	60.00	2026-01-11
104	3	paid	15.50	2026-01-15
105	3	pending	NULL	2026-01-20
106	4	paid	80.00	2026-01-21

orders.createOrReplaceTempView("orders")

sql_df = spark.sql("SELECT customer_id, SUM(amount) AS revenue FROM orders WHERE status = 'paid' GROUP BY customer_id")
api_df = orders.filter("status = 'paid'").groupBy("customer_id").agg(F.sum("amount").alias("revenue"))

sql_df.explain("extended")
api_df.explain("extended")

How the plans compare (both queries return 1 → 65.00, 3 → 15.50, 4 → 80.00)

Plan	SQL version	DataFrame version
Parsed	Every name has a quote mark	Fewer quote marks: each step was checked when you wrote it
Analyzed	Has extra lines for the view name	No view lines
Optimized	Same tree	Same tree (only the # numbers differ)
Physical	Same tree	Same tree

The four paid orders are 101, 102, 104 and 106. Customer 1 has 40.00 + 25.00 = 65.00. A # number is an id that Spark gives one column.

The temporary view lives only as long as the Spark session. The next sections read these plans line by line.

The parsed and analyzed plans differ a little. From the optimized plan on, the trees match. So for this query, the speed is the same.

Remember

You can mix SQL and DataFrame steps in one job. Use SQL for long joins, and DataFrame code for steps you build in a loop.

03 · The optimizer

The Catalyst pipeline

⏱ 6 min · Topic 3 of 11

Catalyst works in five steps. Each step takes the result of the step before it.

Parse: your text becomes an unresolved logical plan. The names are not checked yet.
Analysis: Spark looks the names up in the catalog, its list of tables and columns. It adds the types.
Optimization: rules rewrite the logical plan into an optimized logical plan.
Physical planning: Spark chooses real operators, such as the join type, and picks one plan.
Code generation: Spark combines simple steps into one piece of Java code. This is called whole-stage code generation.

A logical plan says what to do. A physical plan says how to do it. A plan is a tree: each line is one operation, and the lines below it feed it.

Now look at all four plans for one query.

explain("extended") for two filters (simplified)worked example

SQL

Input data

orders6 rows

order_id	customer_id	status	amount	order_date
101	1	paid	40.00	2026-01-03
102	1	paid	25.00	2026-01-10
103	2	refunded	60.00	2026-01-11
104	3	paid	15.50	2026-01-15
105	3	pending	NULL	2026-01-20
106	4	paid	80.00	2026-01-21

orders.filter("status = 'paid'") \
      .filter("order_id > 100 + 3") \
      .select("customer_id", "amount") \
      .explain("extended")

== Parsed Logical Plan ==
'Project ['customer_id, 'amount]
+- Filter (order_id#0 > (100 + 3))
   +- Filter (status#2 = paid)
      +- Relation [order_id#0,customer_id#1,status#2,amount#3,order_date#4] parquet

== Analyzed Logical Plan ==
Project [customer_id#1, amount#3]
+- Filter (order_id#0 > (100 + 3))
   +- Filter (status#2 = paid)
      +- Relation [order_id#0,customer_id#1,status#2,amount#3,order_date#4] parquet

== Optimized Logical Plan ==
Project [customer_id#1, amount#3]
+- Filter (isnotnull(status#2) AND isnotnull(order_id#0) AND (status#2 = paid) AND (order_id#0 > 103))
   +- Relation [order_id#0,customer_id#1,status#2,amount#3,order_date#4] parquet

== Physical Plan ==
*(1) Project [customer_id#1, amount#3]
+- *(1) Filter (isnotnull(status#2) AND isnotnull(order_id#0) AND (status#2 = paid) AND (order_id#0 > 103))
   +- *(1) ColumnarToRow
      +- FileScan parquet [order_id#0,customer_id#1,status#2,amount#3]
           PushedFilters: [IsNotNull(status), IsNotNull(order_id), EqualTo(status,paid), GreaterThan(order_id,103)]

The query returns 2 rows (orders 104 and 106)

customer_id	amount
3	15.50
4	80.00

Order 103 is refunded, and 101 and 102 have order_id 103 or less. Order 105 is pending.

This text is simplified: real output has more detail and different numbers after the # signs. The number after # is an id for one column.

In the parsed plan, a quote mark before a name means "not checked yet". In DataFrame code, Spark checks each step when you write it. So only the last step, the select, still has quotes.

A typo in a column name fails at that line, before any action.

The optimized plan has one Filter, not two. It also shows 103, not 100 + 3. The next section names each change.

04 · The optimizer

Rules that rewrite the plan

⏱ 5 min · Topic 4 of 11

Catalyst has many rules. Each rule looks for one shape in the plan and rewrites it. A predicate is a test that keeps or drops a row.

The plan above shows four rules at work.

Rule	What it does	In the plan above
Constant folding	Computes a fixed expression once, before the job runs	100 + 3 became 103
Filter combining	Merges two Filter lines into one	Two Filters became one with AND
Column pruning	Keeps only the columns the query needs	order_date is gone from FileScan
Predicate pushdown	Moves the filter below other steps, then into the file reader	PushedFilters on FileScan

Pushdown lets the Parquet reader skip data. The skipped data is never loaded.

Look at the Filter line in the optimized plan. Catalyst also added isnotnull(status) and isnotnull(order_id). Catalyst adds them because a NULL never passes status = paid or order_id > 103.

Spark keeps the Filter line even after it pushes the filter to the scan. The scan is allowed to skip data, but it may still return rows that do not match.

05 · Reading plans

The explain modes

⏱ 5 min · Topic 5 of 11

The function explain() takes a mode. Each mode answers a different question. The call explain(True) is the same as explain("extended").

Mode	What it prints	Use it to
explain() or explain("simple")	The physical plan only	Take a quick look
explain("extended")	The four plans: parsed, analyzed, optimized, physical	See what Catalyst changed
explain("formatted")	An outline, then the details of each step	Read a long plan
explain("cost")	The optimized plan with a size on each step. Row counts need statistics and the cost-based optimizer (CBO).	Check the statistics
explain("codegen")	The Java code made by whole-stage code generation	Debug code generation

The formatted mode also numbers each step. Look at it for the revenue query.

explain("formatted") for paid revenue (simplified)worked example

SQL

== Physical Plan ==
AdaptiveSparkPlan (7)
+- HashAggregate (6)
   +- Exchange (5)
      +- HashAggregate (4)
         +- Project (3)
            +- Filter (2)
               +- Scan parquet orders (1)

(1) Scan parquet orders
Output [3]: [customer_id#1, status#2, amount#3]
PushedFilters: [IsNotNull(status), EqualTo(status,paid)]
ReadSchema: struct<customer_id:int,status:string,amount:decimal(10,2)>

(2) Filter
Condition : (isnotnull(status#2) AND (status#2 = paid))

Only steps (1) and (2) are shown. Step (5) is the Exchange. After the job runs, helper lines such as ColumnarToRow appear, so the numbers change.

The cost mode adds Statistics(...) to the end of each line of the optimized plan.

The section "Statistics and the cost model" shows the cost mode in use.

06 · Reading plans

Read a physical plan

⏱ 7 min · Topic 6 of 11

You read the paid revenue plan from the bottom in "Transformations, Actions & Lazy Evaluation". Here you follow the shop rows through it. Then you read the parts that chapter skipped.

Here is the plan for the paid revenue query. It is simplified: real output has more detail.

Physical plan for paid revenue by customer (simplified)worked example

SQL

Input data

orders6 rows

order_id	customer_id	status	amount	order_date
101	1	paid	40.00	2026-01-03
102	1	paid	25.00	2026-01-10
103	2	refunded	60.00	2026-01-11
104	3	paid	15.50	2026-01-15
105	3	pending	NULL	2026-01-20
106	4	paid	80.00	2026-01-21

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[customer_id#1], functions=[sum(amount#3)])
   +- Exchange hashpartitioning(customer_id#1, 200), ENSURE_REQUIREMENTS, [plan_id=27]
      +- HashAggregate(keys=[customer_id#1], functions=[partial_sum(amount#3)])
         +- Project [customer_id#1, amount#3]
            +- Filter (isnotnull(status#2) AND (status#2 = paid))
               +- FileScan parquet [customer_id#1,status#2,amount#3]
                    PushedFilters: [IsNotNull(status), EqualTo(status,paid)]
                    ReadSchema: struct<customer_id:int,status:string,amount:decimal(10,2)>

What each line does, from the bottom

Line	What it does on the shop data
FileScan	Reads 3 of 5 columns. The columns order_id and order_date are skipped.
Filter	Keeps orders 101, 102, 104 and 106.
Project	Keeps customer_id and amount.
HashAggregate (partial)	Each task adds up its own rows, per customer.
Exchange	Moves the partial sums so each customer is in one place.
HashAggregate (final)	Adds the partial sums: 65.00, 15.50 and 80.00.

ReadSchema has status because the Filter needs it. The partial step means the Exchange moves sums, not orders.

The chapter "The Shuffle: Spark's Performance Killer" explains the Exchange. The number 200 is the default value of spark.sql.shuffle.partitions.

Some lines of this plan need a short explanation.

In the plan	What it means
AdaptiveSparkPlan isFinalPlan=false	Adaptive Query Execution (AQE) is on. Spark adds this line when the plan has an Exchange, a subquery, a join or a groupBy. AQE may still change the plan while the job runs.
#1, #3	Ids for columns. The same id in two lines is the same column.
hashpartitioning(customer_id#1, 200)	Each row goes to one of 200 partitions, chosen by a hash of customer_id.
ENSURE_REQUIREMENTS	Spark added this Exchange because the final aggregate needs each customer in one place.
ReadSchema	The columns and types the scan reads from the file.

The Exchange never has a *(n) mark. It is the border between two generated functions.

Count the Exchange lines. Each one is a shuffle. This plan has one.

Interview question

The question: "Explain this plan to me, step by step." A good answer reads from the bottom. Say what the scan reads, which filters were pushed, where the Exchange is, and what each aggregate does.

07 · Join strategy

Why this join is sort-merge

⏱ 6 min · Topic 7 of 11

Choosing the join is part of physical planning. Catalyst picks one strategy for each join. You read the choice from the join line.

Take paid orders joined to customers. The shop tables are tiny, so Spark would broadcast the customers table.

The first line below turns broadcast off. Then the plan shows a sort-merge join. It is simplified.

Paid orders joined to customers (simplified)worked example

SQL

Input data

orders6 rows

order_id	customer_id	status	amount	order_date
101	1	paid	40.00	2026-01-03
102	1	paid	25.00	2026-01-10
103	2	refunded	60.00	2026-01-11
104	3	paid	15.50	2026-01-15
105	3	pending	NULL	2026-01-20
106	4	paid	80.00	2026-01-21

customers5 rows

customer_id	name	country
1	Asha	IN
2	Ben	GB
3	Carla	US
4	Dev	NULL
5	Omar	IN

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)   # turn off broadcast joins

orders.filter("status = 'paid'") \
      .join(customers, "customer_id") \
      .select("name", "amount") \
      .explain()

AdaptiveSparkPlan isFinalPlan=false
+- Project [name, amount]
   +- SortMergeJoin [customer_id], [customer_id], Inner
      :- Sort [customer_id ASC NULLS FIRST], false, 0
      :  +- Exchange hashpartitioning(customer_id, 200)
      :     +- Filter ((isnotnull(status) AND (status = paid)) AND isnotnull(customer_id))
      :        +- FileScan parquet orders
      +- Sort [customer_id ASC NULLS FIRST], false, 0
         +- Exchange hashpartitioning(customer_id, 200)
            +- Filter isnotnull(customer_id)
               +- FileScan parquet customers

The query returns 4 rows

name	amount
Asha	40.00
Asha	25.00
Carla	15.50
Dev	80.00

Omar has no orders, so Omar is not in the result of the inner join. Ben has only a refunded order, which the Filter removed.

Two Exchange lines mean two shuffles. A broadcast join would show BroadcastExchange on the customers side and no shuffle of orders.

Catalyst chooses sort-merge when it cannot broadcast either side. There are a few common reasons.

Both tables are bigger than spark.sql.autoBroadcastJoinThreshold (10 MB by default), or you set it to -1.
Spark cannot estimate the small side’s size, so it treats that side as very big.
The join keeps every row of the small side, so Spark cannot broadcast it. An example is a left join with the small table on the left.

The chapter "Joins in Spark" explains each strategy. In a plan, find the join line first. Then look at the two branches below it to see which Exchange lines feed it.

08 · Statistics

Statistics and the cost model

⏱ 5 min · Topic 8 of 11

Most Catalyst rules do not need numbers. They run on every query. Some choices need size estimates, such as the join order and the join type.

The cost-based optimizer, or CBO, uses statistics to compare plans. Statistics are facts about a table: row count, size, and facts about each column. You collect them with the ANALYZE TABLE command.

Collect statistics for orders, then ask for the costworked example

SQL

Input data

orders6 rows

order_id	customer_id	status	amount	order_date
101	1	paid	40.00	2026-01-03
102	1	paid	25.00	2026-01-10
103	2	refunded	60.00	2026-01-11
104	3	paid	15.50	2026-01-15
105	3	pending	NULL	2026-01-20
106	4	paid	80.00	2026-01-21

# orders must be a saved table, for example from saveAsTable("orders")
spark.catalog.dropTempView("orders")   # the temp view from earlier would hide the table
spark.sql("ANALYZE TABLE orders COMPUTE STATISTICS FOR COLUMNS customer_id, status, amount")
spark.conf.set("spark.sql.cbo.enabled", "true")
spark.table("orders").filter("status = 'paid'").explain("cost")

# each line of the optimized plan now ends like this
Filter (...), Statistics(sizeInBytes=<bytes>, rowCount=<rows>)

Facts that the command stores for orders

Fact	Value on the shop data
Row count	6
Distinct status values	3 (paid, refunded, pending)
Distinct customer_id values	4 (1, 2, 3, 4)
NULL amounts	1 (order 105)

The command stores the row count, the size, and facts for each column you list. The distinct counts are estimates.

With the counts, Spark can estimate how many rows status = paid keeps. It uses the estimate to compare join orders.

The setting spark.sql.cbo.enabled is false by default in Spark 3.x and 4.x. Without it, the cost mode shows only sizeInBytes. Join reordering also needs spark.sql.cbo.joinReorder.enabled.

Statistics get out of date when the table changes. Run the command again after big loads.

AQE uses real sizes while the job runs. See the chapter "Adaptive Query Execution & Cost-Based Optimization".

09 · Limits

When Catalyst cannot help

⏱ 7 min · Topic 9 of 11

Catalyst can only rewrite what it can read. A Python UDF, or user-defined function, is your own Python code.

Catalyst cannot look inside a UDF. It sees only what goes in and what comes out. We call this a black box.

So it cannot push a UDF filter to the scan. Spark also has to send each row to a Python process and back. Look at the same filter written two ways.

A Python UDF against a built-in filter (simplified)worked example

SQL

Input data

orders6 rows

order_id	customer_id	status	amount	order_date
101	1	paid	40.00	2026-01-03
102	1	paid	25.00	2026-01-10
103	2	refunded	60.00	2026-01-11
104	3	paid	15.50	2026-01-15
105	3	pending	NULL	2026-01-20
106	4	paid	80.00	2026-01-21

@F.udf("boolean")
def is_paid(status):
    return status == "paid"

orders.filter(is_paid("status"))          # Python UDF
orders.filter(F.col("status") == "paid")   # built-in

-- UDF plan
*(2) Filter pythonUDF0#9
+- BatchEvalPython [is_paid(status#2)], [pythonUDF0#9]
   +- *(1) ColumnarToRow
      +- FileScan parquet [order_id,customer_id,status,amount,order_date]
           PushedFilters: []

-- built-in plan
*(1) Filter (isnotnull(status#2) AND (status#2 = paid))
+- *(1) ColumnarToRow
   +- FileScan parquet [order_id,customer_id,status,amount,order_date]
        PushedFilters: [IsNotNull(status), EqualTo(status,paid)]

Both versions return 4 rows

order_id	customer_id	status	amount	order_date
101	1	paid	40.00	2026-01-03
102	1	paid	25.00	2026-01-10
104	3	paid	15.50	2026-01-15
106	4	paid	80.00	2026-01-21

The result is the same. The plan is not. The UDF plan has BatchEvalPython and an empty PushedFilters.

Both plans read all five columns, because the query keeps every column. Catalyst knows is_paid needs only status. What it cannot read is the test inside is_paid.

Use a built-in function when one exists. The module pyspark.sql.functions has many. A built-in function is visible to Catalyst, so rules and pushdown still work.

A pandas UDF uses pandas, a Python library for tables. Spark sends batches of rows to it through Apache Arrow, a fast column format.

The plan shows ArrowEvalPython. A pandas UDF is often faster than a Python UDF, but it has the same limit: no pushdown.

The chapter "UDFs & the Pandas API on Spark" covers both kinds in detail.

Watch out

A UDF inside a filter stops that filter from reaching the scan. On a big Parquet table, you read much more data. Look for BatchEvalPython or ArrowEvalPython in the plan.

Spark SQL & the Catalyst Optimizer

Start here

SQL and DataFrames, one plan

The Catalyst pipeline

Rules that rewrite the plan

The explain modes

Read a physical plan

Why this join is sort-merge

Statistics and the cost model

When Catalyst cannot help

Check yourself

Next Chapter

Spark SQL & the Catalyst Optimizer

Start here

SQL and DataFrames, one plan

The Catalyst pipeline

Rules that rewrite the plan

The explain modes

Read a physical plan

Why this join is sort-merge

Statistics and the cost model

When Catalyst cannot help

Check yourself

Next Chapter