Late Materialization in DataFusion's Parquet Reader: Decode the Filter Columns First, Then the Rest

TL;DR: The expensive part of SELECT a, b FROM t WHERE c > 5 is not the comparison — it is decoding a and b for 10 million rows to throw away 9,999,990 of them. DataFusion hands the predicate to the Parquet decoder as a RowFilter: the decoder evaluates c > 5 first, produces a mask, and only decodes a and b for the rows that survive. The columns the filter needs are read before the columns the query wants, which is the opposite of the order a naive reader uses.

The shape of the problem

A Parquet file stores columns independently: a is a column chunk, b is another, c is a third. Decoding any one of them costs I/O, decompression and (for dictionary-encoded, RLE or delta encodings) a fair bit of CPU. Rows, on the other hand, only exist as an alignment between column chunks. The obvious implementation of SELECT a, b FROM t WHERE c = 42 decodes everything and then throws most of it away:

text
naive:
   decode a (10M rows) ┐
   decode b (10M rows) ┘→  apply c > 5 → 10 rows survive
                         ↑ all that decoding was wasted

The fix is to decode c first, learn which rows survive, and then decode a and b only for those rows: “late materialization” — the filter is evaluated against a subset of the columns, and the wide payload is materialized only for the rows that pass.

arrow-rs’s Parquet decoder supports this via RowFilter, and DataFusion’s row_filter.rs documents the contract in its own words:

rust
// datafusion/datasource-parquet/src/row_filter.rs:25-30
//! The `ParquetRecordBatchStream` takes an optional `RowFilter` which is itself
//! a Vec of `Box<dyn ArrowPredicate>`. During decoding, the predicates are
//! evaluated in order, to generate a mask which is used to avoid decoding rows
//! in projected columns which do not pass the filter which can significantly
//! reduce the amount of compute required for decoding and thus improve query
//! performance.

Two things follow. The predicates are evaluated in order, so order matters. And they generate a mask, so the decoder can use row ranges, not just a boolean vector — the second pass over a and b skips whole pages of rows.

Compiling a SQL predicate into a decoder filter

DataFusion’s pushdown has three stages, and each one can fail:

Split the predicate into conjuncts. a = 1 AND b = 2 AND c = 3 becomes three independent predicates. Conjunction is the only boolean operator that distributes cleanly: each conjunct must hold for the row to pass, so each conjunct’s mask can be computed independently and intersected.

rust
// datafusion/datasource-parquet/src/row_filter.rs:493-495
// Split into conjuncts:
// `a = 1 AND b = 2 AND c = 3` -> [`a = 1`, `b = 2`, `c = 3`]
let predicates = split_conjunction(expr);

Decide whether each conjunct can be evaluated during decoding. This is the PushdownChecker in projection_read_plan.rs, which walks the expression tree and classifies every column reference it finds.
Order the survivors by cost and install them as an arrow-rs RowFilter.

What “supported” means

The checker is deliberately conservative. A conjunct can be pushed down if every column it touches is a primitive column (projection_read_plan.rs:324-340), or a list column used by one of the registered list predicates:

rust
// datafusion/datasource-parquet/src/supported_predicates.rs:81-86
fn is_supported_scalar_function(expr: &dyn PhysicalExpr) -> bool {
    scalar_function_name(expr).is_some_and(|name| {
        // Registry of verified array functions
        matches!(name, "array_has" | "array_has_all" | "array_has_any")
    })
}

or a struct column accessed field-by-field through get_field where the resolved leaf is a primitive:

rust
// datafusion/datasource-parquet/src/row_filter.rs:64-66
//! For example, given a struct column `s {name: Utf8, value: Int32}`:
//! - `WHERE s['value'] > 5` — pushed down (accesses a primitive leaf)
//! - `WHERE s IS NOT NULL`  — not pushed down (references the whole struct)

and the column must actually exist in the file. A conjunct referencing a missing column sets projected_columns = true and blocks pushdown (projection_read_plan.rs:324-340).

The struct case is where the leaf-level projection pays off. A filter on s['value'] must not drag the entire struct into the read; DataFusion records the access path — root column index plus the field names forming the path — and clips the read to the leaves it touches (projection_read_plan.rs:72-78). A trie of these accesses is built so that shared prefixes collapse (projection_read_plan.rs:112-155), and the resulting leaf set becomes the ProjectionMask installed on the decoder for that predicate.

Ordering the predicates: bytes, not “selectivity”

The module documentation is explicit about the heuristic:

rust
// datafusion/datasource-parquet/src/row_filter.rs:32-41
//! Since the predicates are applied serially in the order defined in the
//! `RowFilter`, the optimal ordering depends on the exact filters. The best
//! filters to execute first have two properties:
//!
//! 1. They are relatively inexpensive to evaluate (e.g. they read
//!    column chunks which are relatively small)
//!
//! 2. They filter many (contiguous) rows, reducing the amount of decoding
//!    required for subsequent filters and projected columns

Property 1 is what DataFusion implements, and it implements it literally: the cost of a predicate is the compressed size of the columns it reads, summed over all row groups in the file.

rust
// datafusion/datasource-parquet/src/row_filter.rs:387-402
fn size_of_columns(columns: &[usize], metadata: &ParquetMetaData) -> Result<usize> {
    let mut total_size = 0;
    let row_groups = metadata.row_groups();
    for idx in columns {
        for rg in row_groups.iter() {
            total_size += rg.column(*idx).compressed_size() as usize;
        }
    }
    Ok(total_size)
}

The metric is bytes, not row count, and that matters: a filter on a narrow, highly compressible string column is cheap to evaluate even if it matches nothing, while a filter on a wide, poorly compressible column is expensive per row. Sorting candidates by required_bytes puts the cheap-but-selective filter first:

rust
// datafusion/datasource-parquet/src/row_filter.rs:544-547
let mut ordered: Vec<&PrebuiltRowFilterCandidate> = prebuilt.iter().collect();
if reorder_predicates {
    ordered.sort_unstable_by_key(|c| c.required_bytes);
}

Property 2 (selectivity) is deliberately not used, because DataFusion cannot know it: it has no histogram, only min/max. Bytes is knowable from the footer.

The expensive part: don’t rebuild the filter per row group

The filter has to be rebuilt during a scan, because the scan may skip row groups at runtime and the decoder is rebuilt at each boundary. Doing the full construction (tree walks, column resolution, reassign_expr_columns) per row group showed up as a top hot spot in TPC-H profiles. DataFusion split the work in two:

rust
// datafusion/datasource-parquet/src/row_filter.rs:477-486
/// Precompute the list of [`PrebuiltRowFilterCandidate`]s for a predicate.
///
/// This is the expensive part of [`build_row_filter`]: split into conjuncts,
/// resolve columns for each conjunct against the file schema, reassign
/// `Column` indices, and compute the sort-order metadata. Doing it once per
/// file (and reusing across row groups) avoids repeated `TreeNode::transform`
/// walks and `Arc<PhysicalExpr>` allocations that showed up as top hot spots
/// in TPCH profiles.

The prebuilt candidate holds an expression whose Column indices have already been rewritten to point into the projected schema, plus the resolved ProjectionMask. Reinstalling the filter is then just “clone some Arcs and allocate counters”:

rust
// datafusion/datasource-parquet/src/row_filter.rs:529-537
/// Wrap a list of prebuilt candidates into a fresh [`RowFilter`], assigning
/// per-predicate metric counters and (optionally) reordering by
/// `required_bytes`. This is the cheap per-row-group rebuild path — no tree
/// walks, no column resolution, only counter allocation.

RowFilterContext (push_decoder.rs:351-412) is the cache that makes this possible: it owns the shared Arc<Vec<PrebuiltRowFilterCandidate>> and can mint a fresh RowFilter at any row-group boundary.

Late materialization, taken one step further

There is a second, cheaper form of late materialization available when statistics are good enough: prove that every row in a row group matches, and skip the filter entirely for that row group.

At open time, if the first row group to be read is fully matched, DataFusion installs an empty RowFilter and lets arrow-rs’s “does this decoder have predicates?” check short-circuit the per-row evaluation:

rust
// datafusion/datasource-parquet/src/opener/mod.rs:1784-1795
if let Some(ctx) = row_filter_context.as_ref() {
    if first_rg_fully_matched {
        // The first RG is fully matched: install an empty filter
        // and count the suppression ...
        builder = builder.with_row_filter(
            parquet::arrow::arrow_reader::RowFilter::new(vec![]),
        );
        row_filter_skipped_fully_matched.add_one();
    } else {
        builder = builder.with_row_filter(ctx.build_row_filter());

Mid-scan, the same trick is applied per row group: when the decoder reaches the boundary of a fully-matched row group, the stream rebuilds the decoder with the filter removed; when it reaches a row group that does need filtering, the filter is reinstalled from the cached candidates.

rust
// datafusion/datasource-parquet/src/push_decoder.rs:685-703
if filter_needs_toggle {
    let want_filter = desired_filter.expect("filter_needs_toggle ⇒ desired Some");
    if want_filter {
        let ctx = self.row_filter_context.as_ref()...;
        builder = builder.with_row_filter(ctx.build_row_filter());
        ...
    } else {
        // Skip per-row filtering for the upcoming fully-matched RG.
        builder = builder.with_row_filter(RowFilter::new(vec![]));
        self.filter_installed = false;
        self.row_filter_skipped_fully_matched.add_one();
    }
}

The rebuild is the expensive part, so DataFusion does at most one into_builder rebuild per row-group boundary, no matter how many row groups were dropped (push_decoder.rs:467-509).

Finally, the decoded batch is projected onto the query’s output schema by a per-batch transform (decoder_projection.rs:141-160), which rebuilds the batch only when the projector’s schema differs from the scan’s output schema in metadata or nullability.

The catch

The residual filter is not a cop-out; it is a correctness requirement. DataFusion only claims a filter is “exact” for the conjuncts it can actually evaluate during decoding:

rust
// datafusion/datasource-parquet/src/source.rs:682-694
fn exact_filter(&self) -> Option<Arc<dyn PhysicalExpr>> {
    if !self.pushdown_filters() { return None; }
    let predicate = self.predicate.as_ref()?;
    let pushable_schema = self.table_schema.schema_without_virtual_columns();
    let exact = split_conjunction(predicate)
        .into_iter()
        .filter(|expr| can_expr_be_pushed_down_with_schemas(expr, pushable_schema))
        .cloned()
        .collect::<Vec<_>>();
    (!exact.is_empty()).then(|| conjunction(exact))
}

and the pushdown protocol marks the unsupported conjuncts as not pushed down, so the parent FilterExec keeps them:

rust
// datafusion/datasource-parquet/src/source.rs:853-872
let filters: Vec<PushedDownPredicate> = filters.into_iter()
    .map(|filter| {
        if can_expr_be_pushed_down_with_schemas(&filter, pushable_schema) {
            PushedDownPredicate::supported(filter)
        } else {
            PushedDownPredicate::unsupported(filter)
        }
    })
    .collect();

This matters twice: a conjunct that was not evaluated during decoding must still be evaluated somewhere, or the query returns wrong rows; and exact_filter is what FileScanConfig uses to derive equivalence properties — and a filter that is only used for pruning does not remove non-matching rows, so it must not be advertised as exact:

rust
// datafusion/datasource/src/file.rs:113-128
/// Returns the part of [`Self::filter`] that every output row is
/// guaranteed to satisfy.
/// ...
/// A filter that the source uses only to skip data (for example, row group
/// or page pruning with statistics) does not remove every non-matching row,
/// so it must not be returned here.

Other caveats:

Pushdown is off by default. Both datafusion.execution.parquet.pushdown_filters and the per-table option default to false. Pushdown changes the shape of the scan, and for some workloads (a predicate that matches almost everything, on a very wide table) the extra pass over the filter columns is a net loss. reorder_filters is a separate knob for the ordering heuristic.
Late materialization can cost a second pass. If the predicate is unselective, you have paid for an extra scan of the filter columns and decoded everything anyway.
Nested types are a minefield. The rules for lists and structs are per-expression, not per-column: array_has_all(col, ...) is pushable, col[1] = 5 may not be, and col IS NOT NULL on a struct is not pushable at all. Each of those distinctions is a place where a bug would produce wrong answers rather than slow queries.
Virtual columns cannot participate. Predicates referencing the reader-produced row_number column are excluded from pushdown (source.rs:830-845): the reader cannot evaluate them against a mask it has not produced yet.