The Catalyst Optimizer & Project Tungsten

The 4 Phases of Catalyst Optimization

flowchart TD
    A[Unresolved Logical Plan] -->|1. Analysis| B[Analyzed Logical Plan]
    B -->|2. Logical Optimization| C[Optimized Logical Plan]
    C -->|3. Physical Planning| D[Selected Physical Plan]
    D -->|4. Code Generation| E[Java Bytecode Execution]

    style A fill:#f1f5f9,stroke:#94a3b8,stroke-width:2px,color:#000
    style B fill:#dbeafe,stroke:#3b82f6,stroke-width:2px,color:#000
    style C fill:#dcfce7,stroke:#22c55e,stroke-width:2px,color:#000
    style D fill:#fef08a,stroke:#eab308,stroke-width:2px,color:#000
    style E fill:#ffedd5,stroke:#f97316,stroke-width:2px,color:#000

1. Analysis: Uses the Session Catalog to resolve table/column names and data types.
2. Logical Optimization: Applies rule-based optimizations — Constant Folding, Predicate Pushdown (filtering at storage level), and Projection Pruning (dropping unused columns).
3. Physical Planning: Evaluates join strategies (Broadcast Hash Join vs. Sort-Merge Join) via the Cost-Based Optimizer (CBO).
4. Code Generation: Compiles the physical plan into flat, single-loop Java Bytecode (Whole-Stage Code Gen).

Project Tungsten Hardware Optimizations

graph LR
    subgraph Tungsten["Project Tungsten Core Pillars"]
        P1["1. Off-Heap Memory (Unsafe binary array storage - Bypasses GC)"]
        P2["2. Cache-Aware Memory Layout (Maximizes CPU L1/L2/L3 cache hits)"]
        P3["3. Whole-Stage Code Gen (Flattens query loops into single functions)"]
    end
    style Tungsten fill:#fff7ed,stroke:#ea580c,stroke-width:2px;

Inspecting Plans with `.explain(True)`

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

spark = SparkSession.builder.appName("Catalyst Explain").getOrCreate()

data = [("Laptop", 1000, "US"), ("Mouse", 50, "US"), ("Keyboard", 80, "CA")]
df = spark.createDataFrame(data, ["product", "price", "country"])

result_df = df.filter(col("country") == "US") \
              .select("product", (col("price") * 1.10).alias("taxed_price"))

# Inspect Catalyst execution plans
result_df.explain(True)