What does the Catalyst optimizer do for performance?

Summary

Catalyst is the extensible query optimizer in Spark SQL. It resolves your SQL or DataFrame query, rewrites the logical plan with rules such as predicate pushdown and constant folding, chooses physical operators such as the join algorithm, and generates code. Adaptive Query Execution (AQE) can then re-optimize the plan at runtime using runtime statistics.

What does the Catalyst optimizer do for performance?

Catalyst is the optimizer inside Spark SQL, built on functional programming constructs in Scala. You write SQL or DataFrame code, and Catalyst turns it into an optimized execution plan. It applies rule-based rewrites, pushes filters and column selections into data sources that support them, and chooses physical operators such as the join algorithm. The quality of its choices depends on statistics: the Spark docs state that missing or inaccurate statistics "will hinder Spark's ability to select an optimal plan."

How does Catalyst turn a query into a plan?

The Databricks engineering write-up on Catalyst describes four phases. You can see most of them in EXPLAIN output.

Phase	What happens	How to see it
Analysis	References in the logical plan are resolved	EXPLAIN EXTENDED prints the parsed and analyzed logical plans
Logical optimization	Rule-based rewrites run, such as constant folding, predicate pushdown, projection pruning, null propagation and Boolean expression simplification	EXPLAIN EXTENDED prints the optimized logical plan
Physical planning	Physical plans are generated with operators that match the Spark execution engine, and one is selected	Plain EXPLAIN prints the physical plan only
Code generation	Parts of the query are compiled to Java bytecode	EXPLAIN CODEGEN prints generated code, if any, and the physical plan

Which optimizations help performance?

Technique	What it does	Detail from the docs
Predicate and projection pushdown	Pushes operations from the logical plan into data sources that support predicate or projection pushdown	For Parquet, spark.sql.parquet.filterPushdown is true by default
Constant folding	Replaces an expression made of constants with its result	The tree for x+(1+2) becomes x+3
Broadcast join selection	Broadcasts a table to all worker nodes when it is small enough	spark.sql.autoBroadcastJoinThreshold defaults to 10485760 bytes (10 MB); -1 disables broadcasting
Cost-based join reordering	Reorders joins in the cost-based optimizer (CBO), which estimates plan statistics	Off by default: spark.sql.cbo.enabled and spark.sql.cbo.joinReorder.enabled are both false

What does Adaptive Query Execution add?

AQE "makes use of the runtime statistics to choose the most efficient query execution plan," and it has been enabled by default since Apache Spark 3.2.0 (spark.sql.adaptive.enabled). The Spark docs list these features:

• Coalescing post shuffle partitions: shuffle partitions are coalesced based on map output statistics, which simplifies tuning the shuffle partition number.
• Sort-merge join to broadcast hash join: AQE converts the join when the runtime statistics of any join side are smaller than the adaptive broadcast hash join threshold.
• Skew join optimization: skewed tasks in a sort-merge join are split (and replicated if needed) into roughly evenly sized tasks.
• Local shuffle reader: used when shuffle partitioning is not needed, for example after converting a sort-merge join to a broadcast hash join.

How do I see what Catalyst did?

-- Parsed, analyzed and optimized logical plans, plus the physical plan
EXPLAIN EXTENDED
SELECT o.order_id, c.name
FROM orders o JOIN customers c ON o.customer_id = c.id
WHERE o.total > 100;

-- Cost estimates in the optimized plan, when statistics are available
EXPLAIN COST
SELECT o.order_id, c.name
FROM orders o JOIN customers c ON o.customer_id = c.id
WHERE o.total > 100;

In PySpark, df.explain(mode="cost") does the same. In a physical plan, an Exchange operator requires a shuffle, which the Spark docs call likely expensive. To check and refresh the statistics behind the estimates, and to opt in to cost-based join reordering:

ANALYZE TABLE orders COMPUTE STATISTICS FOR ALL COLUMNS;
DESCRIBE EXTENDED orders;
SET spark.sql.cbo.enabled = true;
SET spark.sql.cbo.joinReorder.enabled = true;

When should I use a join hint?

A hint suggests a strategy; it does not force one. The documented join hints are BROADCAST, MERGE, SHUFFLE_HASH and SHUFFLE_REPLICATE_NL, and the docs state that "Spark is not guaranteed to use the join strategy suggested by the hint." With BROADCAST, the hinted side is broadcast regardless of autoBroadcastJoinThreshold, so hint only a side you know is small.

SELECT /*+ BROADCAST(customers) */ *
FROM orders INNER JOIN customers ON orders.customer_id = customers.id;

Fix statistics first, and add a hint only when you know something the optimizer cannot.

How Databricks approaches this

On Databricks, Catalyst still plans your query. Photon, the Databricks-native vectorized query engine, takes over at the execution layer. It processes data in batches, which lets modern CPUs use SIMD instructions. Predictive optimization automatically runs OPTIMIZE, VACUUM and ANALYZE on Unity Catalog managed tables, and ANALYZE collects statistics to improve query performance. To find where a query spent its time, open its query profile from Query History. It shows each query operator with metrics such as time spent, rows processed and memory consumption. AQE is enabled by default on Databricks too, and the AdaptiveSparkPlan node shows isFinalPlan as false until the query finishes.

FAQs

Does Catalyst work with DataFrames?

Yes. Spark SQL starts from either a SQL parser's abstract syntax tree or a DataFrame object, and the Spark SQL guide states that the same execution engine is used "independent of which API/language you are using to express the computation."

Can a bad hint make a query slower?

It can. With BROADCAST, the hinted side "will be broadcast regardless of autoBroadcastJoinThreshold," so a hint on a large table asks Spark to broadcast it anyway. Spark may still ignore a hint when the strategy does not support the join type.

What does spark.sql.shuffle.partitions control?

It is the default number of partitions to use when shuffling data for joins or aggregations. The default is 200. With AQE and partition coalescing both on (the defaults), Spark can coalesce shuffle partitions at runtime, and spark.sql.adaptive.coalescePartitions.initialPartitionNum equals this setting when it is not set.