Predicate Pushdown

Overview

Learn how Predicate Pushdown optimizes read queries by pushing filters directly down to the file storage layer.

What You Will Learn

In this lesson, you will learn:

Storage Filtering: Evaluating query filters before data is read into memory.
File Format Statistics: Using Parquet/ORC metadata to skip data blocks.
I/O Reductions: Reducing network and memory overhead.

Detailed Concept Explanation

Predicate Pushdown is a performance optimization where the filter conditions in a query (e.g. WHERE age > 21) are evaluated directly at the storage level, before the data is loaded into Spark executor memory.

How it Works with Columnar Formats

Columnar file formats like Parquet and ORC store summary statistics (min/max values) for each column block in their metadata footer.

When Spark requests data, it reads this footer first.
If a query filters for age > 50 and a block's metadata indicates max_age = 45, Spark skips reading the entire block of rows from disk.
This saves significant disk I/O, network bandwidth, and deserialization CPU time.

Code Examples

Python (PySpark) Implementation

python

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("PredicatePushdown").getOrCreate()

# Read Parquet directory (Predicate pushdown runs automatically)
df = spark.read.parquet("users.parquet")

# Filter query
result = df.filter(df.city == "Paris")

# PushedFilters list in physical plan shows optimization
result.explain()

Expected Output

text

== Physical Plan ==
*(1) ColumnarToRow
+- FileScan parquet [id#0,city#1] Batched: true, Format: Parquet, 
   PartitionFilters: [], PushedFilters: [EqualTo(city,Paris)], ReadSchema: struct<id:int,city:string>

Execution Plan Diagram (Python & Scala)

Execution Plan Diagram

SparkSession.builder

read.parquet

filter(city == Paris)

explain()

Scala Implementation

scala

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().appName("PushdownScala").getOrCreate()

val df = spark.read.parquet("users.parquet")
val result = df.filter($"city" === "Paris")

result.explain()