What is Predicate Pushdown?

How it works

Without pushdown, an engine reads every row and then filters. With pushdown, the optimizer hands the filter to the scan, which uses metadata to avoid reading data at all. There are several layers, from cheapest to finest:

1. Partition pruning: a filter on a partition column (country = 'IN') skips whole folders. Spark lists these as PartitionFilters.
2. File skipping: table formats such as Delta Lake and Iceberg keep min/max statistics per file and skip files that cannot match. Snowflake does the same per micro-partition.
3. Row-group skipping: inside each Parquet file, min/max statistics per row group let the reader skip blocks. Spark lists these as PushedFilters.
4. Column pruning (projection pushdown): only the selected columns are read, shown as ReadSchema.

Pushdown only works when the engine can translate the filter. A plain comparison on a column pushes down; a filter wrapped in a Python UDF, a cast on the column, or a function such as upper(col) = 'X' often cannot. And statistics skip only when values are clustered: if every file holds the full range of a column, min/max can never exclude anything, which is why sorting, Z-ordering and clustering matter.

Example

A table partitioned by country. The plan shows the country filter pruning partitions, the amount filter pushed into the Parquet reader, and only two columns read (PySpark 4.2.0).

import re, tempfile
from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.master("local[2]").appName("glossary-pushdown").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
path = tempfile.mkdtemp() + "/orders"

(spark.range(1_000_000)
   .withColumn("country", F.element_at(F.array(*map(F.lit, ["IN", "US", "UK", "DE"])), (F.col("id") % 4 + 1).cast("int")))
   .withColumn("amount", F.col("id") % 500)
   .write.partitionBy("country").parquet(path))

q = (spark.read.parquet(path)
       .filter((F.col("country") == "IN") & (F.col("amount") > 450))
       .select("id", "amount"))
scan = [l for l in q._jdf.queryExecution().executedPlan().toString().splitlines() if "FileScan" in l][0]
for part in ["PartitionFilters", "PushedFilters", "ReadSchema"]:
    print(re.search(part + r": (\[.*?\]|\S+)", scan).group(0))
print("rows:", q.count())
spark.stop()

PartitionFilters: [isnotnull(country#5), (country#5 = IN)]
PushedFilters: [IsNotNull(amount), GreaterThan(amount,450)]
ReadSchema: struct<id:bigint,amount:bigint>
rows: 24000

Common mistakes

• Wrapping a filter column in a function or cast (to_date(ts) = '2026-10-01'), which can stop it from being pushed down.
• Filtering after a Python UDF or a collect, when the filter could have run in the scan.
• Expecting skipping on a column whose values are spread randomly across files.
• Reading CSV or JSON and expecting the same savings; those formats have no statistics to skip with.