Catalyst & Tungsten¶

A teammate opens a PR: a query joining a week of SaaS events to a 40 MB customer dimension, filtered to one region and one day. In review you ask for explain("formatted"). It shows a FileScan of the full 8 TB week and a SortMergeJoin — not the 40 GB, broadcast-joined plan either of you expected.

A. The optimiser has a bug. B. The filter is on a computed column (to_date(timestamp)), not the partition column, so pruning never fires. C. The dimension is over the broadcast threshold. D. AQE is disabled in this environment.

Pick one before reading on — the answer changes what you fix. You write a DataFrame query over a week of SaaS events. Spark does not “run the API.” It rewrites the query into something that might scan 40 GB of three columns instead of 8 TB of JSON-shaped Parquet, might filter before it joins, and might generate JVM bytecode that never boxes a Row.

If you do not read explain(), you are hoping Catalyst agrees with you. Production is not a hope.

Start with the situation¶

events.join(customers, "customer_id") \
      .filter(F.col("region") == "eu-west-1") \
      .filter(F.col("date") == "2024-01-15") \
      .groupBy("service") \
      .agg(F.avg("latency_ms"), F.count("*"))

events is date-partitioned Parquet, 8 TB for the week, 1.1 TB for the 15^th. customers is 40 MB. region is a data column, not a partition.

What you want Catalyst to consider:

1. Push date = into partition pruning (only one day’s files).
2. Push region = into Parquet row-group filters if stats allow.
3. Read only customer_id, service, latency_ms, region, date.
4. Broadcast customers instead of shuffling 1.1 TB — if its measured size is under the broadcast threshold. Broadcast is not guaranteed: it depends on table/column statistics being present and accurate, AQE being enabled, and the join being broadcast-compatible. A hint (.hint("broadcast", ...)) is a request, not an order — Spark's own docs are explicit that join hints are not guarantees.
5. Partial aggregate service before any remaining exchange — the exact rewrite Catalyst produces depends on query shape and join semantics; do not assume this chain in general.

None of this is a guaranteed physical plan — it is the plan you should check for. If any of it is missing in explain("formatted") for this specific query, the job is wrong even if it “works” on a sample.

Observability: pruning on hour= is the difference between a 2 TB trace scan and a 12 GB one. CDC: MERGE plans that explode into Cartesian nested-loop joins. IoT: skipping the value column until the last projection.

Why the obvious approach breaks at scale¶

The optimiser is a pattern matcher, not an oracle:

• It cannot push a filter through a Python UDF. The UDF is a black box.
• It cannot prune timestamp if you partitioned on date and filtered on timestamp (unless Iceberg hidden partitioning).
• Statistics may be stale (Hive) or absent (a pile of Parquet). AQE exists because compile-time stats lie.
• Whole-stage codegen fails open to interpreted execution on some expressions; you will not notice except in CPU.
• AQE changes the plan mid-job. The plan you pasted in the PR may not be the plan that ran.

At 8 TB, a missed pushdown is an incident. At 80 MB, it is a lab.

Build the mental picture¶

Two layers:

• Catalyst — what operators, in which order, with which join algorithm.
• Tungsten — how a stage runs: compact rows, off-heap, generated loops.

SQL / DataFrame
    → Unresolved logical plan
    → Analysis (names, types)
    → Logical optimisation (rules)
    → Physical planning (join strategy, exchanges)
    → Codegen (Tungsten)
    → Tasks

Your job is to write queries the optimiser can see through: native functions, explicit schemas, partition columns in WHERE, small dimensions that stay small.

Under the hood¶

Catalyst pipeline¶

Predicate pushdown. Filter before join/scan.

# You write join then filter
events.join(customers, "customer_id").filter(events.region == "eu-west-1")

# Catalyst rewrites to filter events first (and maybe customers)

Column pruning. Parquet ReadSchema lists only needed fields. SELECT * then groupBy is how you disable this.

Partition pruning. PartitionFilters: [date=2024-01-15] in FileScan. If you do not see it, you are paying for the week.

Constant folding / simplify. where(true), dead casts.

Join selection (physical):

Partial vs final HashAggregate. count/sum combine map-side. countDistinct and collect_list do not shrink the same way.

Tungsten¶

• UnsafeRow: bytes, not JVM objects. Fewer allocations, worse to look at in a debugger.
• Whole-stage codegen: fuse a pipeline (FileScan → Filter → Project → PartialAgg) into one loop. Stage boundary (Exchange) breaks fusion — another reason shuffles cost CPU beyond the NIC.
• Off-heap / unified memory: execution vs storage (spark.memory.fraction, storageFraction). Cache fights shuffle buffers.

If codegen bails (WholeStageCodegen not wrapping your operators), you are back to Volcano-style virtual calls. explain shows it.

Adaptive Query Execution (Spark 3.x)¶

Compile-time \(R=200\) is a guess. AQE reoptimises after each shuffle using map-output stats.

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", str(128 * 1024 * 1024))
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.localShuffleReader.enabled", "true")
# optional: let AQE broadcast if the *measured* side is small
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", str(64 * 1024 * 1024))

What AQE actually does:

1. Coalesce tiny shuffle partitions (fixes “200 tasks on 80 MB”).
2. SMJ → BHJ if runtime size says so.
3. Skew join split on fat join partitions.
4. Local shuffle reader when reducers are on the same executor as the map output (saves NIC).

It will not: invent a partition column you forgot; push through UDFs; make SELECT * cheap; split a skewed aggregation key (that is salting — Shuffle).

Put it to work¶

Always explain expensive jobs¶

from pyspark.sql import functions as F

result = (
    events
    .filter((F.col("date") == "2024-01-15") & (F.col("status_code") >= 500))
    .select("customer_id", "service", "latency_ms", "date", "status_code")
    .groupBy("service")
    .agg(F.count("*").alias("n"), F.avg("latency_ms").alias("avg_ms"))
)
result.explain(mode="formatted")
# mode="extended" for logical + physical
# mode="cost" when stats exist

Checklist in the formatted plan:

Native expressions, not UDFs¶

# Catalyst can push / codegen
df.filter(F.col("endpoint").startswith("/api/internal"))

# Catalyst cannot see through this
@F.udf("boolean")
def is_internal(ep):
    return ep.startswith("/api/internal") if ep else False
df.filter(is_internal(F.col("endpoint")))

The UDF version scans more, serialises to Python, and disables some fusion. See Gotchas.

Join hints when you know sizes¶

df.hint("broadcast", "customers").join(customers, "customer_id")
# or
events.join(customers.hint("broadcast"), "customer_id")

Hints are how you stop AQE oscillating in a tight SLA. They are also how you OOM if you are wrong. Prefer measuring.

Iceberg / Delta¶

Table formats give Catalyst manifest-level pruning (min/max, partition specs, sometimes clustering). A well-sorted Iceberg table can skip files for customer_id = even without identity partitioning — if you maintained the sort. That is optimiser input, not magic.

Production gotchas¶

How it fails¶

Type mismatch joins are infamous: customer_id int vs string → cast → sometimes a BroadcastNestedLoopJoin. explain catches it in review; production catches it in the bill.

How to investigate¶

spark.conf.set("spark.sql.adaptive.enabled", "true")
# After the job, SQL tab shows the *final* AQE plan. explain() before run is compile-time.

Compare bytes scanned to bytes on disk for that date. If they match the whole table, pruning failed.

Scale¶

Cost-based optimiser (CBO) helps when stats exist. On a raw lake they often do not — AQE is the CBO for people without stats.

Dynamic partition pruning (DPP)¶

A common SaaS pattern: join a tiny tenant table (WHERE customer_id IN (whales we care about)) to a date-partitioned fact.

whales = spark.read.parquet("s3://dims/watchlist/")  # few hundred ids
fact = spark.read.parquet("s3://analytics/events/")  # 90 days
fact.join(F.broadcast(whales), "customer_id") \
    .groupBy("date", "customer_id").count()

If the fact is partitioned by date and you also filter dates from the dim, Spark can inject a dynamic pruning subquery so unused day directories are never listed. In explain, look for dynamicpruningexpression. If you SMJ two huge sides with no broadcast and no stats, DPP never fires and you scan 90 days to answer “yesterday for three tenants.”

Statistics you can actually collect¶

spark.sql("ANALYZE TABLE events COMPUTE STATISTICS FOR ALL COLUMNS")
# Iceberg: snapshot summaries + manifests already carry min/max per file

Hive-style ANALYZE goes stale the next ingest. Prefer file-level min/max (Parquet footers, Iceberg manifests) plus AQE. Do not build a religion around table-level row counts from last March.

Trade-offs¶

Alternatives¶

• Trino / ClickHouse optimiser + storage layout may beat Spark for interactive SQL with the same Parquet. Spark wins at heavy transforms, UDF-ish ETL (still prefer native), and writes (Iceberg jobs).
• Materialised marts so Catalyst never sees the 8 TB join again.
• Flink for incremental plans; its optimiser is different (operator chaining, not Catalyst).

How to apply at work¶

PR template:

1. Paste explain("formatted") on production-sized partitions (or a sampled path that still has the partition spec).
2. Assert PartitionFilters / PushedFilters.
3. Assert join type.
4. List UDFs and why they are not native.
5. AQE flags in the SparkConf of the submit, not in a notebook cell that never shipped.

If scan bytes in the last prod run were 8 TB for a “yesterday” job, Catalyst is not your problem — you never gave it a date filter.

Check your understanding¶

q = (
    spark.read.parquet("s3://analytics/events/")   # partitioned by date
    .filter(F.to_date("timestamp") == F.lit("2024-01-15").cast("date"))
    .join(customers, F.col("customer_id") == F.col("cust_id"))
    .withColumn("bucket", udf_bucket("endpoint"))
    .groupBy("bucket", "region")
    .count()
)

customers is 25 MB with column cust_id integer; events customer_id is string. udf_bucket is a Python UDF. AQE on.

1. Will partition pruning fire? Why?
2. What join strategy do you fear, and why?
3. Where does the UDF sit relative to pushdown?
4. Rewrite the query so Catalyst can do the right thing. Name each change.
5. After the rewrite, which AQE feature still matters at 100× files?