Catalyst — the optimizer making your code fast

The whole DataFrames chapter kept invoking one name as the invisible hand behind every speedup — predicate pushdown, column pruning, the reason native functions beat UDFs, the reason your obvious code runs fast. We said “Catalyst” a dozen times and never once opened it. The chapter was, in effect, a long lesson in writing code that keeps this optimizer happy, given without ever showing you the optimizer. This chapter shows it.

When you write a DataFrame pipeline, Spark doesn’t run it as written. A component called Catalyst takes your query, rewrites it for performance, then runs the rewritten version. The transformations are often dramatic — sometimes Catalyst eliminates 90% of the work before the cluster even starts.

Catalyst is the reason naive PySpark code is usually fast. It’s also the reason understanding the explain plan is the highest-leverage debugging skill in Spark.

The four stages

A query travels through Catalyst in four steps:

Stages 1–3 produce logical plans (what to compute, not how). Stage 4 produces a physical plan (which join algorithm, which file reader, which execution engine). You can see all of them with .explain(extended=True).

A worked example

Consider this query:

big_df = spark.read.parquet("s3://bucket/orders/")    # 100 columns, 1B rows

result = (big_df
    .filter(F.col("status") == "PAID")
    .filter(F.col("amount") > 100)
    .withColumn("net", F.col("amount") - F.col("discount"))
    .select("order_id", "user_id", "net")
    .where(F.col("net") > 50))

A naive engine would read all 100 columns and 1B rows, then apply each operation in sequence. Catalyst transforms the plan before execution. Here’s what it does:

Optimization 1: Combine filters

# Before
.filter(F.col("status") == "PAID")
.filter(F.col("amount") > 100)
# After
.filter((F.col("status") == "PAID") & (F.col("amount") > 100))

Same result, one pass through the data instead of two.

Optimization 2: Predicate pushdown

Catalyst pushes filters as close to the data source as possible. For Parquet, that means into the file scan itself:

Scan parquet(s3://bucket/orders/) 
  PushedFilters: [
    EqualTo(status, PAID),
    GreaterThan(amount, 100),
  ]

Parquet stores per-row-group min/max statistics. If a row group has status values only FAILED and REFUNDED, Spark skips the entire row group without reading the data. On a typical filtered query you might read 5% of the file instead of 100%.

Optimization 3: Column pruning

Spark sees that the final .select() keeps only order_id, user_id, and net. net is computed from amount and discount. So Catalyst reads only 5 columns out of the 100 stored in Parquet.

ReadSchema: struct<order_id:bigint, user_id:bigint, amount:double, discount:double, status:string>

Notice status is also read (the filter needs it) but the other ~95 columns are skipped. Column pruning often saves 80-95% of I/O.

Optimization 4: Constant folding

df.withColumn("vat_rate", F.lit(1.18))
  .withColumn("net", F.col("amount") * F.lit(1.18) * F.lit(1.05))

Catalyst pre-computes 1.18 * 1.05 = 1.239 once, instead of computing it per row.

Optimization 5: Filter reordering

Catalyst tries to apply cheap filters before expensive ones. A status = 'PAID' (one column compare) runs before a UDF that calls a regex (slow).

The same query, two plans

To see Catalyst at work, here are simplified before/after plans:

Before optimization (what you wrote):

Filter (net > 50)
+- Project [order_id, user_id, net]
   +- Project [..., (amount - discount) as net, ...]
      +- Filter (amount > 100)
         +- Filter (status = 'PAID')
            +- Scan parquet(..., all 100 cols, no pushdown)

After optimization (what runs):

Project [order_id, user_id, (amount - discount) as net]
+- Filter (status = 'PAID' AND amount > 100 AND (amount - discount) > 50)
   +- Scan parquet(..., 5 cols, PushedFilters: [status=PAID, amount>100])

The optimized plan reads ~5% of the file by row-group skipping, ~5% of the columns by pruning, and applies all filters in a single pass. The naive plan read everything and filtered in three passes.

Reading Catalyst’s decisions with explain()

You’ll learn to read the formatted plan in the next lesson, but here’s a quick taste:

result.explain(mode="formatted")

Output (abbreviated):

== Physical Plan ==
* Project (5)
+- * Filter (4)
   +- * ColumnarToRow (3)
      +- BatchScan parquet (2)
         +- Filter [status = 'PAID', amount > 100] (1)

(1) PushedFilters: status, amount
(2) ReadSchema: order_id, user_id, amount, discount, status

The * markers indicate whole-stage codegen — Catalyst compiles the chain of operations into a single Java function with no virtual calls. That’s another 2-5x speedup on top of the planning work.

When Catalyst can’t help

Catalyst optimizes what it can see. Things it can’t see through:

• Python UDFs — opaque Python functions, no pushdown possible
• Complex when().otherwise() chains — Catalyst can optimize some but not arbitrary ones
• RDDs — fully opaque; Catalyst doesn’t optimize the RDD API at all

The practical advice: prefer built-in functions over UDFs, and prefer DataFrames over RDDs. Catalyst rewards you generously for both.

Adaptive Query Execution — runtime optimization

Catalyst optimizes the plan before execution, based on static information (schema, table stats). Spark 3.x added Adaptive Query Execution (AQE) which can re-optimize during execution based on actual data sizes observed at runtime. You’ll learn about AQE in its own lesson.

The combination — Catalyst plus AQE — handles most of what a senior engineer used to do by hand-tuning. You write clean code; the engine does the work.

A toy optimizer pass

Predicate pushdown isn’t magic. Here’s the kernel of it in 20 lines:

# Toy: push filters down through projections.
# Plan represented as a list of (op_name, args) tuples, leaf first.

plan = [
    ("scan", {"cols": ["a", "b", "c"]}),
    ("project", {"keep": ["a", "b"]}),
    ("filter", {"pred": "a > 5"}),         # depends only on column 'a'
]

def pushdown(plan):
    new_plan = list(plan)
    for i in range(len(new_plan) - 1, 0, -1):
        op, args = new_plan[i]
        below_op, below_args = new_plan[i - 1]
        if op == "filter" and below_op == "project":
            cols_needed = args["pred"].split()[0]
            if cols_needed in below_args["keep"]:
                # Swap filter and project — apply filter before project
                new_plan[i], new_plan[i - 1] = new_plan[i - 1], new_plan[i]
    return new_plan

print("Original plan:")
for op, args in plan:
    print(f"  {op}({args})")

print("\nAfter pushdown:")
for op, args in pushdown(plan):
    print(f"  {op}({args})")

Original plan:
  scan({'cols': ['a', 'b', 'c']})
  project({'keep': ['a', 'b']})
  filter({'pred': 'a > 5'})

After pushdown:
  scan({'cols': ['a', 'b', 'c']})
  filter({'pred': 'a > 5'})
  project({'keep': ['a', 'b']})

Watch the filter move. In the original plan it sat above the project, meaning Spark would have built the projected rows first and only then thrown most of them away. The rule noticed that a > 5 needs only column a, which project keeps, so it is safe to swap them — and after the pass the filter sits directly on the scan, discarding rows before the projection ever touches them. That is predicate pushdown in one rule: less data flows up the tree. Real Catalyst has hundreds of these rules (written in Scala), but the shape is exactly this — pattern- match a sub-tree, replace it with a cheaper equivalent that computes the same answer.

In one breath

Catalyst is the optimizer that rewrites your DataFrame query before it runs, in four stages (parse → analyze references → optimize the logical plan → pick a physical plan), applying rules that routinely cut most of the work: predicate pushdown (push filters into the Parquet scan so row groups whose min/max stats can’t match are skipped — often reading 5% of the file), column pruning (read only the columns the query needs — 80–95% less I/O on wide tables), constant folding, filter/join reordering, and whole-stage codegen (compile the operator chain into one tight Java loop, the * in explain); it can only optimize what it can see, which is precisely why opaque Python UDFs and RDDs are slow and native F.* functions are fast — and AQE later re-optimizes at runtime.

Practice

Before the quiz, trace the worked query’s two plans. The naive version read all 100 columns and 1B rows and filtered in three passes; the optimized version read 5 columns, ~5% of the rows, in one pass. Name the two specific Catalyst optimizations responsible for the “5 columns” and the “5% of rows,” and where in the Parquet file each one acts. Then connect it to the chapter’s refrain: explain, in terms of “what Catalyst can see,” exactly why a Python @udf defeats both of those optimizations.

A question to carry forward

We just watched Catalyst do remarkable things to a query — push filters into the scan, prune 95 of 100 columns, fold constants, reorder for cost. But notice how we watched: through a hand-simplified before/after plan we wrote out by hand, a toy that swaps two list elements, and a one-line “quick taste” of .explain() that we waved past with “you’ll learn to read this next.” Every claim in this lesson — “Catalyst pushed the filter down,” “it pruned to 5 columns” — was asserted. You took it on faith.

In production you cannot take it on faith. You force a broadcast and need to confirm it broadcast; you expect a filter pushed into Parquet and need to confirm the scan skipped row groups; a job is mysteriously slow and the answer is sitting in the plan if you can read it. The single most valuable debugging skill in Spark is reading what Catalyst actually decided for your specific query — and that lives in one output we keep glancing at and never deciphering. So the question to carry forward is: how do you read a Spark physical plan — the operators, the PushedFilters, the ReadSchema, the join type, the * codegen markers — and verify with your own eyes that the optimizer did what you hoped? That is reading execution plans, and it is the next lesson.