The Catalyst Optimizer

The Catalyst Optimizer is the core of Spark SQL's query planning engine.

• Parsed Logical Plan – Spark parses your SQL text or DataFrame chain into a tree of logical operators with unresolved names.
• Analyzed Logical Plan – Catalyst resolves column and table names against the catalog and checks types.
• Optimized Logical Plan – Rule-based rewrites like predicate pushdown and column pruning reduce the data the plan must read.
• Physical Plan – Catalyst generates candidate physical plans, picks the cheapest using a cost model, and applies whole-stage code generation to fuse operators into compiled JVM bytecode.

Because queries are expressed declaratively, Catalyst sees the entire plan and optimizes it end to end, which is why DataFrames and SQL consistently outperform hand-written RDD code.

What is the Catalyst Optimizer?

The Catalyst Optimizer is Spark SQL's extensible query optimizer. It transforms every DataFrame operation or SQL query into an efficient execution plan through a series of tree transformations. Every time you write a Spark SQL query against the National Bank of India's transaction ledger, Catalyst decides how to read the least data and execute the fastest way.

# PySpark example: Catalyst optimizes this automatically
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("BankCatalyst").getOrCreate()

transactions = spark.read.parquet("/data/bank/transactions")

# Catalyst pushes the filter into the Parquet scan
# and prunes columns not needed
result = (
    transactions
    .filter("account_id = 'ACC-1001'")
    .select("account_id", "amount", "txn_date")
    .groupBy("txn_date")
    .sum("amount")
)

result.explain(True)  # Shows all four Catalyst phases

How does the Parsed Logical Plan work?

The first phase produces the parsed logical plan. Spark parses your DataFrame operations or SQL text into a tree of logical operators. At this stage column and table names are unresolved: Catalyst does not yet know whether a column or table actually exists.

-- SQL that Catalyst first parses into an unresolved tree
SELECT account_id, SUM(amount) AS total
FROM banking_prod.transactions
WHERE branch_code = 'MUMBAI-01'
GROUP BY account_id

== Parsed Logical Plan ==
'Aggregate ['account_id], ['account_id, 'SUM('amount) AS total]
+- 'Filter ('branch_code = MUMBAI-01)
   +- 'UnresolvedRelation [banking_prod, transactions]

The single quotes around names indicate they are unresolved. The plan is a tree, not executable code.

How does the Analysis phase resolve names?

The second phase is analysis. Catalyst resolves references against the catalog, binding column names to actual columns and table names to actual tables, and checking types. If you referenced a column that does not exist, this is where the query fails with an AnalysisException.

# This fails at analysis because 'nonexistent_col' is not in the table
try:
    transactions.select("nonexistent_col").show()
except Exception as e:
    print(f"Analysis error: {e}")
# AnalysisException: Column 'nonexistent_col' does not exist

After analysis, every column reference is bound to a concrete type and position, producing the analyzed logical plan.

What are the key logical optimizations?

The third phase is rule-based logical optimization, where the biggest wins happen. Catalyst applies a library of transformation rules repeatedly until the plan stops changing.

• Predicate pushdown moves filters as close to the data source as possible, so a filter on account_id is pushed into the Parquet or JDBC scan and far less data is read.
• Column pruning removes columns the query never uses, so reading only the amount column from a wide Parquet file skips all the others.
• Constant folding evaluates constant expressions at plan time rather than per row.
• Boolean simplification reduces complex filter expressions.
• Combine filters merges adjacent filter nodes into a single predicate.

# Catalyst pushes the filter below the join
accounts = spark.read.parquet("/data/bank/accounts")
txns = spark.read.parquet("/data/bank/transactions")

# The filter on branch_code is pushed into the accounts scan
result = (
    txns.join(accounts, "account_id")
    .filter("accounts.branch_code = 'MUMBAI-01'")
    .select("txns.amount", "accounts.branch_code")
)

result.explain(True)
# Optimized plan shows the filter pushed below the join

What is whole-stage code generation?

The fourth phase is physical planning. Catalyst generates one or more physical plans, uses a cost model to pick the cheapest, and then applies whole-stage code generation (WholeStageCodegen). This fuses a chain of operators within a stage into a single compiled Java method, so data is processed in tight loops over the JVM rather than through many virtual method calls.

# Check codegen in the physical plan
result.explain("codegen")

== Physical Plan ==
*(1) HashAggregate(keys=[txn_date], functions=[sum(amount)])
+- *(1) HashAggregate(keys=[txn_date], functions=[partial_sum(amount)])
   +- *(1) FileScan parquet [amount,txn_date]
      PushedFilters: [EqualTo(account_id, ACC-1001)]
      ReadSchema: struct<amount:decimal(18,2),txn_date:date>

The *(1) prefix indicates whole-stage codegen is active. Without it, each operator would process rows through virtual dispatch, adding significant per-row overhead.

How does predicate pushdown work with different data sources?

Predicate pushdown behaves differently depending on the source. With Parquet, filters are pushed into the file reader using min/max statistics to skip entire row groups. With JDBC, filters become part of the SQL WHERE clause sent to the database.

# JDBC predicate pushdown: filter sent to the database
jdbc_df = (
    spark.read
    .format("jdbc")
    .option("url", "jdbc:postgresql://bank-db:5432/banking")
    .option("dbtable", "transactions")
    .option("user", "spark_reader")
    .option("password", "*****")
    .load()
)

# Catalyst pushes this filter into the JDBC query
filtered = jdbc_df.filter("account_id = 'ACC-1001' AND amount > 10000")
filtered.explain(True)
# PushedFilters: [EqualTo(account_id,ACC-1001), GreaterThan(amount,10000)]

Why do DataFrames outperform hand-written RDD code?

With RDDs, Spark cannot see inside your lambdas, so it cannot push filters, prune columns, or choose join strategies. With DataFrames and SQL, you express what you want declaratively, and Catalyst sees the whole query to optimize it end to end.

# RDD approach: Spark cannot optimize this
rdd = spark.sparkContext.textFile("/data/bank/transactions.csv")
result_rdd = (
    rdd.map(lambda line: line.split(","))
    .filter(lambda cols: cols[1] == "ACC-1001")
    .map(lambda cols: (cols[3], float(cols[2])))  # (date, amount)
)
# No pushdown, no pruning, no codegen

# DataFrame approach: Catalyst optimizes everything
result_df = (
    spark.read.parquet("/data/bank/transactions")
    .filter("account_id = 'ACC-1001'")
    .select("txn_date", "amount")
    .groupBy("txn_date").sum("amount")
)
# Pushdown, pruning, codegen all applied automatically

How is Catalyst extended with custom rules?

Catalyst is built on a tree-rewriting framework. Each optimization is a rule that pattern-matches on the plan tree and produces a rewritten tree. You can add custom rules by extending Rule[LogicalPlan].

// Scala example: a custom Catalyst rule
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.expressions._

object BankAmountValidation extends Rule[LogicalPlan] {
  def apply(plan: LogicalPlan): LogicalPlan = plan transform {
    case filter @ Filter(condition, child) =>
      // Add validation: amount must be positive for bank transactions
      filter
  }
}

This extensibility is what makes Catalyst "catalyst" : it is a framework for building optimizers, not just a fixed set of rules.

What should a senior engineer know about Catalyst?

• Catalyst is complemented at runtime by Adaptive Query Execution (AQE), which adjusts the physical plan using real statistics gathered while the job runs.
• The cost model for physical plan selection is basic compared to traditional databases: it primarily uses data size estimates and often falls back to heuristics.
• Statistics collection via ANALYZE TABLE improves the cost model for join strategy selection.
• When the optimizer makes a wrong choice, you can use hints to override it, but hints are a last resort.
• Catalyst cannot optimize across stage boundaries: each stage is optimized independently.

-- Collect statistics to help Catalyst's cost model
ANALYZE TABLE banking_prod.transactions COMPUTE STATISTICS FOR ALL COLUMNS;

-- Use a hint to force a broadcast join (last resort)
SELECT /*+ BROADCAST(branches) */
    t.account_id, t.amount, b.branch_name
FROM transactions t
JOIN branches b ON t.branch_code = b.branch_code

What are common interview follow-up questions?

• How does Catalyst decide between a broadcast join and a sort-merge join? It uses the autoBroadcastJoinThreshold and table size estimates from the catalog.
• Can Catalyst optimize across multiple queries? No, each query is optimized independently.
• What is the relationship between Catalyst and Tungsten? Tungsten is the execution engine that provides memory management and code generation; Catalyst is the optimizer that produces the plan Tungsten executes.
• How does AQE complement Catalyst? AQE re-optimizes the physical plan at runtime using real shuffle statistics, correcting mistakes the static optimizer made.