Spark SQL and the Catalyst Optimizer

Spark SQL is not a separate engine bolted onto Spark. It is the engine — the DataFrame API and SQL are two front ends over the same optimiser.

Registering views

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("sql").master("local[*]").getOrCreate()

orders = spark.createDataFrame(
    [(1, "alice", "UK", 250.00, "2026-01-15"),
     (2, "bob",   "US", 180.50, "2026-01-16"),
     (3, "carol", "UK", 320.75, "2026-01-16"),
     (4, "dave",  "DE",  95.00, "2026-01-17"),
     (5, "erin",  "US", 410.20, "2026-01-18")],
    ["order_id", "customer", "country", "amount", "order_date"],
)
customers = spark.createDataFrame(
    [("alice", "gold"), ("bob", "silver"), ("carol", "gold"), ("dave", "bronze")],
    ["customer", "tier"],
)

orders.createOrReplaceTempView("orders")
customers.createOrReplaceTempView("customers")

spark.sql("SHOW TABLES").show()

+---------+---------+-----------+
|namespace|tableName|isTemporary|
+---------+---------+-----------+
|         |customers|       true|
|         |   orders|       true|
+---------+---------+-----------+

Querying

spark.sql("""
    SELECT c.tier,
           COUNT(*)              AS orders,
           ROUND(SUM(o.amount),2) AS revenue,
           ROUND(AVG(o.amount),2) AS avg_order
    FROM orders o
    JOIN customers c ON o.customer = c.customer
    WHERE o.amount > 100
    GROUP BY c.tier
    ORDER BY revenue DESC
""").show()

+------+------+-------+---------+
|  tier|orders|revenue|avg_order|
+------+------+-------+---------+
|  gold|     2| 570.75|   285.38|
|silver|     1|  180.5|    180.5|
+------+------+-------+---------+

SQL and DataFrames compile to the same thing

sql_plan = spark.sql("""
    SELECT country, SUM(amount) AS total
    FROM orders WHERE amount > 100 GROUP BY country
""")

df_plan = (
    orders.filter(F.col("amount") > 100)
          .groupBy("country")
          .agg(F.sum("amount").alias("total"))
)

print(sql_plan._jdf.queryExecution().optimizedPlan().toString())
print("---")
print(df_plan._jdf.queryExecution().optimizedPlan().toString())
print("identical:", sql_plan._jdf.queryExecution().optimizedPlan().toString()
                 == df_plan._jdf.queryExecution().optimizedPlan().toString())

Aggregate [country#2], [country#2, sum(amount#3) AS total#41]
+- Project [country#2, amount#3]
   +- Filter (isnotnull(amount#3) AND (amount#3 > 100.0))
      +- LogicalRDD [order_id#0L, customer#1, country#2, amount#3, order_date#4], false
---
Aggregate [country#2], [country#2, sum(amount#3) AS total#48]
+- Project [country#2, amount#3]
   +- Filter (isnotnull(amount#3) AND (amount#3 > 100.0))
      +- LogicalRDD [order_id#0L, customer#1, country#2, amount#3, order_date#4], false
identical: False

Structurally identical — only the auto-generated expression IDs differ, which is why the string comparison says False. Use whichever front end reads better; there is no performance argument either way.

The four stages

spark.sql("""
    SELECT o.customer, o.amount
    FROM orders o JOIN customers c ON o.customer = c.customer
    WHERE o.country = 'UK' AND o.amount > 100
""").explain(True)

== Parsed Logical Plan ==
'Project ['o.customer, 'o.amount]
+- 'Filter (('o.country = UK) AND ('o.amount > 100))
   +- 'Join Inner, ('o.customer = 'c.customer)
      :- 'SubqueryAlias o
      :  +- 'UnresolvedRelation [orders]
      +- 'SubqueryAlias c
         +- 'UnresolvedRelation [customers]

== Analyzed Logical Plan ==
customer: string, amount: double
Project [customer#1, amount#3]
+- Filter ((country#2 = UK) AND (amount#3 > 100.0))
   +- Join Inner, (customer#1 = customer#5)
      ...

== Optimized Logical Plan ==
Project [customer#1, amount#3]
+- Join Inner, (customer#1 = customer#5)
   :- Project [customer#1, amount#3]
   :  +- Filter (((isnotnull(country#2) AND isnotnull(amount#3)) AND (country#2 = UK))
   :             AND (amount#3 > 100.0)) AND isnotnull(customer#1))
   :     +- LogicalRDD [order_id#0L, customer#1, country#2, amount#3, order_date#4]
   +- Project [customer#5]
      +- Filter isnotnull(customer#5)
         +- LogicalRDD [customer#5, tier#6]

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- BroadcastHashJoin [customer#1], [customer#5], Inner, BuildRight
   :- Project [customer#1, amount#3]
   :  +- Filter (...)
   :     +- Scan ExistingRDD[...]
   +- BroadcastExchange HashedRelationBroadcastMode(...)
      +- Project [customer#5]
         +- Filter isnotnull(customer#5)
            +- Scan ExistingRDD[customer#5, tier#6]

Four things happened between the parsed and optimised plans, none of which you wrote:

Filter pushdown. The WHERE clause moved below the join, so filtering happens before rows are joined rather than after. Fewer rows enter the join.

Column pruning. The right side became Project [customer#5] — tier was dropped because the query never selects it.

Null inference. isnotnull(customer) was added on both sides. An inner join can never match a null key, so Catalyst makes that explicit and uses a faster code path.

Join strategy. The physical plan chose BroadcastHashJoin because the customers table is small, avoiding a shuffle entirely.

Watching pushdown reach the file

orders.write.mode("overwrite").parquet("data/orders_parquet")
spark.read.parquet("data/orders_parquet").createOrReplaceTempView("orders_pq")

spark.sql("SELECT customer FROM orders_pq WHERE amount > 300").explain()

== Physical Plan ==
*(1) Project [customer#71]
+- *(1) Filter (isnotnull(amount#73) AND (amount#73 > 300.0))
   +- FileScan parquet [customer#71,amount#73]
        DataFilters: [isnotnull(amount#73), (amount#73 > 300.0)],
        PushedFilters: [IsNotNull(amount), GreaterThan(amount,300.0)],
        ReadSchema: struct<customer:string,amount:double>

Two lines matter. PushedFilters means the predicate went into the Parquet reader, which uses per-row-group min/max statistics to skip chunks without decoding them. ReadSchema lists two columns of five — the other three are never read off disk.

CSV cannot do either:

orders.write.mode("overwrite").option("header", True).csv("data/orders_csv")
spark.read.option("header", True).csv("data/orders_csv").createOrReplaceTempView("orders_csv")
spark.sql("SELECT customer FROM orders_csv WHERE amount > 300").explain()

+- FileScan csv [customer#88,amount#90]
     DataFilters: [isnotnull(amount#90), (cast(amount#90 as double) > 300.0)],
     PushedFilters: [],
     ReadSchema: struct<customer:string,amount:double>

PushedFilters: [] — empty. Every row is read and parsed, then filtered in Spark. This is the concrete reason Parquet beats CSV, expressed in the plan rather than as folklore.

Adaptive execution changes the plan mid-flight

spark.conf.set("spark.sql.adaptive.enabled", True)

q = spark.sql("SELECT country, COUNT(*) FROM orders GROUP BY country")
q.explain()
q.collect()
print("--- after execution ---")
q.explain()

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[country#2], functions=[count(1)])
   +- Exchange hashpartitioning(country#2, 200), ENSURE_REQUIREMENTS
      ...
--- after execution ---
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
   *(2) HashAggregate(keys=[country#2], functions=[count(1)])
   +- AQEShuffleRead coalesced
      +- ShuffleQueryStage 0

isFinalPlan flipped from false to true, and AQEShuffleRead coalesced appeared. Spark measured the actual shuffle output — three groups, tiny — and collapsed 200 partitions into one. The plan you see before running is a plan, not a promise.

Catalog access

print([t.name for t in spark.catalog.listTables()])
print([c.name for c in spark.catalog.listColumns("orders")])
print(spark.catalog.currentDatabase())

['customers', 'orders', 'orders_csv', 'orders_pq']
['order_id', 'customer', 'country', 'amount', 'order_date']
default

Useful when generating SQL dynamically or validating that an expected table exists before a job runs.

Practice

parquet: PushedFilters: [IsNotNull(amount), GreaterThan(amount,300.0)]
csv:     PushedFilters: []

AnalysisException: Table or view not found: my_view

Next: reading the Spark UI to work out why a job was slow.