Predicate and column pushdown

Description

Lesson 2 organized the warehouse: fact_orders_at_scale.parquet, three aisles labeled by store_id. This lesson measures, with .explain() — never a stopwatch — exactly how much that organization pays off when someone searches for just one store. You're going to see, with real evidence, two distinct mechanisms Spark applies when reading Parquet: partition pruning (PartitionFilters, skipping whole folders without opening them) and predicate/column pushdown (PushedFilters, skipping data inside the files that do get opened). They sound similar, but they operate at different layers of the filesystem, and this lesson teaches you to tell them apart in the same plan text.

Connection to the module. This lesson completes the write-read cycle lesson 2 opened: writing with a criterion (partitionBy) is only worth it if you later know how to read the evidence that criterion worked. Without this lesson, partitionBy("store_id") would be an act of faith; with it, it's a verifiable decision.

An analogy: the aisle door, and the labels inside the boxes

Pick back up lesson 2's warehouse: three labeled aisles, S01, S02, S03. A forklift needing only S01's boxes does two things, at two distinct moments. First, it doesn't even open the door to aisles S02 and S03 — it ignores them entirely, without spending a second checking them — that's partition pruning, and it happens before touching a single file. Second, once inside aisle S01, if it's also looking for just one specific product's boxes (P004), it checks each box's label without necessarily unpacking it fully — a label is enough to decide whether that box is useful or not — that's predicate pushdown, and it happens inside the files that did get opened. And if, on top of that, it only needs each box's weight (not its whole contents), it can read just that specific label, without loading the rest — that's column pushdown. All three mechanisms work together, but at different layers — folder, file, column — and this lesson measures them separately.

Worked example: three reads, three different plans

Step 1 — The starting point: read everything, with no filter

# read_partitioned_with_pushdown.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col

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

fact_at_scale_df = spark.read.parquet("fact_orders_at_scale.parquet")
print(f"fact_at_scale_df.rdd.getNumPartitions() (no filter) = {fact_at_scale_df.rdd.getNumPartitions()}")

What to expect (executed in this run):

fact_at_scale_df.rdd.getNumPartitions() (no filter) = 12

Step 2 — Filter by the partition column: partition pruning

s01_only_df = fact_at_scale_df.filter(col("store_id") == "S01")

print("--- .explain() over store_id == 'S01' (partition column) ---")
s01_only_df.explain()

print(f"s01_only_df.count() = {s01_only_df.count()}")
assert s01_only_df.count() == 4_000_000

What to expect (executed in this run):

--- .explain() over store_id == 'S01' (partition column) ---
== Physical Plan ==
*(1) ColumnarToRow
+- FileScan parquet [product_id#0,order_id#1,franchise_id#2,quantity#3,unit_price#4,order_ts#5,store_name#6,city#7,product_name#8,category#9,unit_cost#10,revenue#11,store_id#12] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:.../fact_orders_at_scale.parquet], PartitionFilters: [isnotnull(store_id#12), (store_id#12 = S01)], PushedFilters: [], ReadSchema: struct<product_id:string,order_id:string,franchise_id:int,quantity:int,unit_price:double,order_ts...
s01_only_df.count() = 4000000

Notice two fields in the plan, right next to each other. PartitionFilters: [isnotnull(store_id#12), (store_id#12 = S01)] is full — Spark knows, before opening a single file, that it only needs to enter the store_id=S01/ folder, because store_id's value lives in the directory name, not inside the data. PushedFilters: [], by contrast, is empty — there's no need to push any predicate down to the Parquet reader, because the complete filter already got resolved at the folder level, before the file reader even entered the picture. s01_only_df.count() gives 4,000,000 — exactly 16 orders per franchise (the S01 proportion you already know from module 1) × 250,000 franchises.

Step 3 — Filter by a column that is NOT the partition column: real predicate pushdown

print("--- .explain() over product_id == 'P004' (a column that is NOT the partition column) ---")
p004_only_df = fact_at_scale_df.filter(col("product_id") == "P004")
p004_only_df.explain()

What to expect (executed in this run):

--- .explain() over product_id == 'P004' (a column that is NOT the partition column) ---
== Physical Plan ==
*(1) Filter (isnotnull(product_id#0) AND (product_id#0 = P004))
+- *(1) ColumnarToRow
   +- FileScan parquet [product_id#0,order_id#1,franchise_id#2,quantity#3,unit_price#4,order_ts#5,store_name#6,city#7,product_name#8,category#9,unit_cost#10,revenue#11,store_id#12] Batched: true, DataFilters: [isnotnull(product_id#0), (product_id#0 = P004)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:.../fact_orders_at_scale.parquet], PartitionFilters: [], PushedFilters: [IsNotNull(product_id), EqualTo(product_id,P004)], ReadSchema: struct<product_id:string,order_id:string,franchise_id:int,quantity:int,unit_price:double,order_ts...

Now the two fields swap. PartitionFilters: [] is empty — product_id isn't the partition column, so Spark has no way to skip any of the three folders with just this filter; it has to check store_id=S01/, store_id=S02/, and store_id=S03/ completely. PushedFilters: [IsNotNull(product_id), EqualTo(product_id,P004)] is full — this is real predicate pushdown: the predicate (product_id = P004) gets pushed down to the Parquet reader, which can use the per-column statistics stored in each file's metadata (min/max per block) to discard entire blocks that can't contain P004, without decompressing their content. Filter (isnotnull(product_id#0) AND (product_id#0 = P004)) also shows up as an explicit node above the FileScan — unlike partition pruning, which resolves the complete filter before reading, predicate pushdown reduces the work but still needs a final row-by-row check over what did get read.

Step 4 — Column pushdown: asking for fewer columns, reading fewer bytes

print("--- Column pruning: request only store_id and revenue ---")
narrow_df = fact_at_scale_df.filter(col("store_id") == "S01").select("store_id", "revenue")
narrow_df.explain()

spark.stop()

What to expect (executed in this run):

--- Column pruning: request only store_id and revenue ---
== Physical Plan ==
*(1) Project [store_id#12, revenue#11]
+- *(1) ColumnarToRow
   +- FileScan parquet [revenue#11,store_id#12] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:.../fact_orders_at_scale.parquet], PartitionFilters: [isnotnull(store_id#12), (store_id#12 = S01)], PushedFilters: [], ReadSchema: struct<revenue:double>

Compare this FileScan against step 2's: there, ReadSchema listed all thirteen complete columns; here, with .select("store_id", "revenue") before .explain(), ReadSchema: struct<revenue:double> — a single column. store_id doesn't even show up in ReadSchema, because there's no need to read it from any file: it's already known from the folder name (store_id=S01/), the same partition pruning from step 2. This is column pushdown in its clearest form: Parquet stores each column separately inside the file, so Spark can literally read only revenue's bytes, without touching order_id, product_name, category, or any of the other ten columns you didn't ask for. A CSV, being a row-based format, can't do this — to read any row's revenue value, a CSV reader has to decode the complete row, column by column, in order, even if you only care about one.

Diagram: three layers of pruning, from outside in

flowchart TD
    A["fact_orders_at_scale.parquet/\n3 folders: store_id=S01, S02, S03"] --> B{"filtering by store_id?"}
    B -->|"yes -- partition column"| C["PartitionFilters full\nSpark doesn't even open the other 2 folders"]
    B -->|"no"| D["PartitionFilters empty\nSpark checks all 3 folders"]
    C --> E{"filtering by another column\n(e.g. product_id)?"}
    D --> E
    E -->|"yes"| F["PushedFilters full\nSpark uses per-block statistics\nto skip whole row-groups"]
    E -->|"no"| G["PushedFilters empty\nSpark reads every row-group\nin the files it did open"]
    F --> H{".select() with\nspecific columns?"}
    G --> H
    H -->|"yes"| I["Narrow ReadSchema\nOnly the requested columns get read"]
    H -->|"no"| J["Complete ReadSchema\nAll 13 columns get read"]

Going deeper: why CSV never shows up with this same pruning power

An honest comparison is worth making against everything you've already read in .explain() since module 1: when you filter a CSV by a column (as you did, for example, with orders_df.filter(col("product_id") == "P004") in earlier lessons), the plan can also show a PushedFilters field with content — Spark tries to push the predicate down to any data source, not just Parquet. But there's a real difference, not just a naming one: Spark's official documentation explicitly documents the spark.sql.parquet.filterPushdown setting (active by default) as the one that "enables Parquet filter push-down optimization" — an optimization depending on Parquet storing per-column and per-block statistics (min, max, null count) in its own metadata, allowing it to skip complete blocks without decompressing them. A CSV has no metadata of that kind at all — it's plain text, row by row — so even though the filter gets "pushed" in the sense of being evaluated as early as possible in the read cycle, every row still has to be fully decoded before the filter can be applied. Parquet's real payoff isn't just "the filter gets applied earlier" — it's that Parquet can, literally, not read certain data blocks from disk at all, something a plain text format can't offer.

Common mistakes

Expecting PartitionFilters to show up when filtering by any column, not just the partition one. What happens: someone filters fact_at_scale_df.filter(col("product_id") == "P004"), checks the plan, and is surprised to see PartitionFilters: [] empty. Why it happens: the two fields — PartitionFilters and PushedFilters — have similar names and show up in the same block of the plan, and it's easy not to notice they respond to different mechanisms. How to spot it: check which column you used in partitionBy(...) when writing — in this dataset, only store_id; any other filter, no matter how selective, is never going to trigger PartitionFilters. How to fix it: PartitionFilters only gets filled when the filter is exactly on the column (or columns) you used in partitionBy(...) when writing the Parquet — for any other column, the most you can expect is PushedFilters, a more modest but real mechanism.

Using df.inputFiles() to confirm partition pruning worked, and drawing the wrong conclusion. What happens: someone calls s01_only_df.inputFiles() expecting the list to include only the files under store_id=S01/, and is surprised to see all 36 files from the three complete folders. Why it happens: inputFiles() reflects the files the data source's index discovered when building the DataFrame, not the ones partition pruning actually avoids reading at runtime — that decision happens later, inside the physical plan's FileScan. How to spot it: if you compare len(df.inputFiles()) before and after a filter on the partition column and the number doesn't change, that isn't a sign pruning failed — it's that inputFiles() simply isn't the right tool to measure it. How to fix it: the right evidence for partition pruning is the PartitionFilters field inside .explain(), as this lesson shows — not inputFiles(), which lists the universe of candidate files, not the subset that genuinely gets read.

Concluding that empty PushedFilters means the filter "didn't work." What happens: someone sees PushedFilters: [] in this lesson's step 2 (filter on store_id) and assumes the filter didn't get applied, or that there's a bug. Why it happens: it's intuitive to expect any valid filter to leave a trace in PushedFilters. How to spot it: check the result — s01_only_df.count() == 4_000_000, exactly as expected — before assuming something failed just because a field is empty. How to fix it: when the filter is on the partition column, PartitionFilters already did the work at the folder level — there's no need, and no point, pushing the same predicate down to the individual file reader again. An empty PushedFilters next to a full PartitionFilters is a sign the filter got resolved the cheapest way possible, not that it failed.

Exercises

Exercise 1 — Repeat the filter on store_id, but for S03, and confirm the exact count. Filter fact_at_scale_df by store_id == "S03", run .explain(), and confirm with assert the expected count using module 1's proportion (S03 = 11 orders per franchise).

s03_only_df = fact_at_scale_df.filter(col("store_id") == "S03")
s03_only_df.explain()
print(f"s03_only_df.count() = {s03_only_df.count()}")
assert s03_only_df.count() == 11 * 250_000 == 2_750_000
print("Verification: S03 has 2,750,000 rows -> OK")

PartitionFilters: [isnotnull(store_id#12), (store_id#12 = S03)], PushedFilters: []
s03_only_df.count() = 2750000
Verification: S03 has 2,750,000 rows -> OK

Exercise 2 — Combine a partition filter with one that isn't, and predict which plan fields fill up. Before running anything, predict: filtering fact_at_scale_df.filter((col("store_id") == "S01") & (col("product_id") == "P004")), what do you expect to see in PartitionFilters and in PushedFilters? Then, verify with code.

combined_df = fact_at_scale_df.filter((col("store_id") == "S01") & (col("product_id") == "P004"))
combined_df.explain()
print(f"combined_df.count() = {combined_df.count()}")

PartitionFilters: [isnotnull(store_id#12), (store_id#12 = S01)], PushedFilters: [IsNotNull(product_id), EqualTo(product_id,P004)]
combined_df.count() = 700000

Exercise 3 — Explain, without code, why step 4's ReadSchema: struct<revenue:double> doesn't include store_id, even though the filter uses it. In 2-3 sentences, explain why a column can be part of a query's filter without showing up in ReadSchema.

Summary and next step

In this lesson you read fact_orders_at_scale.parquet three different ways, and learned to tell apart, in the same .explain() text, three pruning mechanisms working at different layers: PartitionFilters (skipping whole folders, the cheapest of the three), PushedFilters (skipping blocks inside a file, using Parquet statistics), and column pruning visible in ReadSchema (reading only the bytes of the columns you actually asked for). You confirmed, with real counts, that each mechanism produces the correct result — 4,000,000 rows for S01, 2,750,000 for S03, 700,000 for combining both filters — and understood why a flat CSV can never offer the same level of pruning as Parquet.

Before moving on you should be able to: tell PartitionFilters apart from PushedFilters in any .explain() plan; explain why inputFiles() isn't reliable evidence of partition pruning; and explain, in your own words, why columnar Parquet lets you read fewer bytes than a CSV for the same query.

With partitioned Parquet read and measured, this module changes topic entirely: lesson 4 opens the Python UDF box, and explains — with the same discipline of reading the plan, never the clock — why a plain @udf is the slowest black box Catalyst can run.

Resources

• Apache Spark — SQL Data Sources: Parquet (spark.sql.parquet.filterPushdown, active by default; partition discovery). spark.apache.org/docs/latest/sql-data-sources-parquet.html.
• Apache Spark — SQL Data Sources: CSV (reference for CSV read options; with no mention at all of filter pushdown or per-column statistics, the foundation for this lesson's comparison). spark.apache.org/docs/latest/sql-data-sources-csv.html.
• This guide's DESIGN doc (spark-and-distributed-processing-guide/DISENO.md) — this lesson's exact specification: filtered reads with .explain() showing PushedFilters and partition pruning.