What is predicate pushdown?

Summary

• Predicate pushdown hands a query's filter conditions to the data source during planning, so the source can skip data that can't match instead of returning every row for Spark to discard.
• In Spark's Data Source V2 API, a connector opts in by implementing SupportsPushDownV2Filters (preferred) or SupportsPushDownFilters on its ScanBuilder, keeps the predicates it can apply and returns the rest for Spark to evaluate.
• Spark's pushdown switches default to true: spark.sql.parquet.filterPushdown, spark.sql.orc.filterPushdown, spark.sql.csv.filterPushdown.enabled, spark.sql.json.filterPushdown.enabled and the JDBC option pushDownPredicate.
• Apache Iceberg and Delta Lake keep per-file statistics such as minimum and maximum values and use them to skip whole files.
• To check what was pushed, run EXPLAIN and read the scan node's PushedFilters field.

What is predicate pushdown?

Predicate pushdown is an optimization in which Spark gives the filter conditions in a query (the predicates in WHERE or filter()) to the data source while it plans the scan, so the source can avoid producing rows that can't match. Without it, the source returns everything and Spark filters afterwards. With it, a database can apply the condition in its own query, and a file or table format can skip data it knows is irrelevant. In open-source Apache Spark, pushdown is on by default for Parquet, ORC, CSV and JSON files and for the JDBC source. Your job is mostly to write filters a source can accept and to lay data out so those filters can rule something out.

How does Spark negotiate pushdown with a source?

Data Source V2 (DSV2) is Spark's connector API, and connectors opt in to filter pushdown and other optimizations by implementing mix-in interfaces. Before building a scan, Spark checks the connector's ScanBuilder for them in a fixed order: sample, filter, aggregate, limit or top-N, offset, then column pruning. For filters:

1. Spark calls pushPredicates(Predicate[]) on a source that implements SupportsPushDownV2Filters, or pushFilters(Filter[]) on one that implements SupportsPushDownFilters.
2. The source keeps what it can apply and returns the predicates "that need to be evaluated after scanning". Filters are interpreted as ANDed together.
3. pushedPredicates() (or pushedFilters()) reports what the source accepted, and Spark evaluates the returned predicates itself.

The Javadoc prefers SupportsPushDownV2Filters (since Spark 3.3.0) over SupportsPushDownFilters (since Spark 3.0.0), because the older interface uses V1 Filter objects and pays for an internal-to-external data conversion. Spark 4.1.0 added a filter pushdown API to Python data sources: DataSourceReader.pushFilters(filters) returns the filters Spark must still evaluate, and by default it returns all of them, so nothing is pushed.

Which sources push filters down?

Source	How pushdown works	Switch (default)
Parquet	Filters go to the Parquet reader	spark.sql.parquet.filterPushdown (true)
ORC	Filters go to the ORC reader	spark.sql.orc.filterPushdown (true)
CSV and JSON	Filters go to the CSV or JSON datasource	spark.sql.csv.filterPushdown.enabled, spark.sql.json.filterPushdown.enabled (true)
JDBC	Filters go to the database "as much as possible"	pushDownPredicate read option (true)
Apache Iceberg	Predicates become partition filters, then column stats eliminate files at planning	Automatic
Delta Lake	Min and max values collected on write are used to skip files at query time	Automatic, Delta Lake 1.2.0 and above

Parquet's native record-level filtering is a separate setting, spark.sql.parquet.recordLevelFilter.enabled, which defaults to false. In Delta Lake, statistics cover only columns whose schema position is below the value of the delta.dataSkippingNumIndexedCols table property, so keep frequently filtered columns inside that range. Skipping also depends on layout; the Delta Lake docs recommend Z-Ordering for best results.

How do I check whether a filter was pushed down?

Run EXPLAIN on the query and read the scan node of the physical plan:

EXPLAIN FORMATTED
SELECT id, status FROM events
WHERE status = 'active' AND region = 'US';

For a file source, the scan node prints fields such as PartitionFilters, PushedFilters and DataFilters, and PushedFilters: [] means nothing reached the source. If a condition is missing, check its expression. Spark's configuration docs note that "not all expressions can be converted into data source filters and some expressions can only be evaluated by Spark (e.g. subqueries)".

How Databricks approaches predicate pushdown

Databricks runs the same Spark planning and adds data skipping on Delta Lake and managed Apache Iceberg tables. Statistics are collected automatically on write, and Databricks uses per-file minimum and maximum values, null counts and record counts at query time to skip irrelevant files. For Unity Catalog managed tables, predictive optimization collects file-skipping statistics on the columns most used in query filters and runs ANALYZE automatically. Unity Catalog external tables collect statistics on the first 32 columns by default. For table layout, Databricks recommends liquid clustering on Databricks Runtime 13.3 and above.

FAQs

Does a pushed filter mean Spark won't evaluate it again?

Not always. The SupportsPushDownFilters Javadoc lists three kinds: pushable filters that don't need evaluating after the scan, pushable filters that still do (a Parquet row group filter is its example), and filters that can't be pushed. Both pushable kinds count as pushed and are returned by pushedFilters().

Can I turn pushdown off or force it?

You can turn it off: set spark.sql.parquet.filterPushdown or spark.sql.orc.filterPushdown to false, or set the JDBC option pushDownPredicate to false so "all filters will be handled by Spark". You can't force it, because the source decides which predicates it keeps.

Is partition pruning the same thing?

It's a close relative. Iceberg converts query predicates into predicates on partition data and applies them before checking column statistics. Spark can also add partition filters at run time: with spark.sql.optimizer.dynamicPartitionPruning.enabled (true by default), it generates a predicate for a partition column used as a join key.

Do OR conditions get pushed?

It depends on the source. In the Python data source API, a = 1 and (b = 2 or c = 3) reaches pushFilters as just a = 1, and a = 1 or b = 2 passes nothing, so Spark evaluates the rest.