Catalyst Optimizer

The Four Phases

Key Numbers

TreeNode and Rule

// Catalyst's core abstractions (simplified)
abstract class TreeNode[BaseType <: TreeNode[BaseType]] {
  def children: Seq[BaseType]
  def transform(rule: PartialFunction[BaseType,BaseType]): BaseType =
    rule.applyOrElse(this, identity).mapChildren(_.transform(rule))
}

abstract class LogicalPlan extends TreeNode[LogicalPlan]
abstract class Expression  extends TreeNode[Expression]

// A rule is a function LogicalPlan => LogicalPlan
abstract class Rule[T <: TreeNode[_]] {
  def apply(plan: T): T
}

// Example rule: collapse Project(Project(x, p1), p2) into Project(x, p2 o p1)
object CollapseProject extends Rule[LogicalPlan] {
  def apply(plan: LogicalPlan): LogicalPlan = plan.transform {
    case Project(p2, Project(p1, child)) =>
      Project(buildCleanedProjectList(p2, p1), child)
  }
}

Common Optimization Rules

Cost-Based Optimization (CBO)

# CBO is opt-in (off by default)
spark.sql.cbo.enabled = true
spark.sql.cbo.joinReorder.enabled = true
spark.sql.cbo.joinReorder.dp.threshold = 12

# Compute statistics first
ANALYZE TABLE orders COMPUTE STATISTICS;
ANALYZE TABLE orders COMPUTE STATISTICS FOR COLUMNS cust_id, total, status;

# What CBO uses:
# - Table size (rowCount, sizeInBytes)
# - Per-column: min, max, num distinct, num nulls, avg/max length
# - Histograms for non-uniform distributions

# Without CBO, Catalyst uses heuristic estimates
# (often "size = sum of input sizes after filter selectivity 0.5")

In practice, CBO alone has limited adoption because computing column stats is expensive and the statistics drift. AQE largely supplanted it for runtime decisions, while CBO remains useful for large multi-way join reordering at compile time when stats are reliable.

Physical Planning Strategies

# A Strategy turns a LogicalPlan into one or more SparkPlans.
# SparkPlanner tries each strategy and picks the best.

abstract class Strategy {
  def apply(plan: LogicalPlan): Seq[SparkPlan]
}

# Built-in strategies for joins:
JoinSelection extends Strategy {
  // 1. Broadcast hash join if one side <= autoBroadcastJoinThreshold
  // 2. Shuffle hash join if one side small enough to fit in executor
  //    (controlled by spark.sql.join.preferSortMergeJoin = false)
  // 3. Sort-merge join (default for big-vs-big)
  // 4. Broadcast nested loop / cartesian for non-equi
}

# Other strategies:
# - Aggregation: hash-based vs sort-based
# - Window: per-partition window or global
# - DataSourceV2: file scan with partition pruning

Inspecting the Plan

# Get the plan at each stage
val df = spark.sql("SELECT c.country, SUM(o.total) FROM orders o " +
                   "JOIN customers c ON o.cust_id=c.id WHERE c.region='EU' " +
                   "GROUP BY c.country")

df.queryExecution.parsed         // Unresolved
df.queryExecution.analyzed       // Resolved (column types known)
df.queryExecution.optimizedPlan  // After Catalyst rules
df.queryExecution.executedPlan   // Physical plan with strategies
df.queryExecution.debug.codegen  // Generated Java source

# EXPLAIN modes
df.explain(false)         // simple
df.explain(true)          // all four phases
df.explain("formatted")   // Spark 3+ — most readable
df.explain("cost")        // includes CBO row counts (if computed)

# Sample formatted output
== Physical Plan ==
* HashAggregate (5)
+- Exchange (4)              <-- shuffle
   +- * HashAggregate (3)
      +- * SortMergeJoin Inner (2)
         :- * Filter (1)
            +- * Scan parquet [orders]
         +- * Filter
            +- * Scan parquet [customers]

Extending Catalyst

# Add a custom rule via SparkSessionExtensions
class MyOptimizationRule extends Rule[LogicalPlan] {
  def apply(plan: LogicalPlan): LogicalPlan = plan.transform {
    case Filter(EqualTo(AttributeReference("status",_,_,_), Literal("X",_)), child) =>
      // Skip the filter entirely if we know our table never has status="X"
      child
  }
}

class MyExtensions extends (SparkSessionExtensions => Unit) {
  def apply(ext: SparkSessionExtensions): Unit = {
    ext.injectOptimizerRule(_ => new MyOptimizationRule)
  }
}

# Register in spark-defaults.conf
spark.sql.extensions = com.example.MyExtensions

# Catalysts used by big projects:
# - Iceberg adds rules for incremental scans and metadata pruning
# - Delta Lake adds time-travel and merge-into rewriting
# - Photon (Databricks) replaces parts of physical planning

Tradeoffs and Limits

Frequently Asked Questions