Opensource internals

Polars Prunes Parquet Row Groups With Chunk Statistics, Not the Page Index

12 min read Jay Hubbert

How the streaming Parquet reader in Polars turns footer statistics into a DataFrame of min/max columns, evaluates the predicate on it, and drops whole row groups before any I/O.

TL;DR: Polars prunes Parquet row groups with the column chunk statistics that live in the file footer, not with the page index. Those statistics are decoded into a tiny DataFrame with <col>_min, <col>_max and <col>_nc columns, the user’s predicate is evaluated against that DataFrame, and the row groups whose bit is clear are never fetched. Polars writes the page index (column index and offset index) when it writes a file with statistics enabled, but its own reader never reads it back: page-level skipping is not implemented.

Every reader of a Parquet file faces the same question: the predicate says ts > 2026-01-01, which of the row groups in this file can be thrown away without touching the bytes? The answer has to come from somewhere, and the only metadata a reader is guaranteed to have is the footer.

Before diving in, one correction to the usual framing. Polars’ vendored Parquet implementation does write the page index — the column index and the offset index, the structures that hold per-page min/max and per-page byte offsets. If you write a Parquet file with Polars (with statistics enabled) and read it with DuckDB, Spark or DataFusion, those readers can skip pages inside a row group. Polars’ own reader does not do that. It skips whole row groups, using the statistics attached to each column chunk. The distinction matters: a row group is typically 100k+ rows, a page is typically a few thousand, so page-level skipping is a strictly finer-grained optimisation that Polars writes the data for but does not yet consume.

Which reader is the real one

The Parquet code lives in two crates. crates/polars-parquet holds the deserializers, the statistics decoding, and a vendored copy of the parquet crate itself (crates/polars-parquet/src/parquet/, with its own thrift decoder and page reader). The plan — which row groups, which columns, in what order — lives in the streaming engine, in crates/polars-stream/src/nodes/io_sources/parquet/. That split matters: the plan is where the pruning decision is made, and it is not in the crate you would guess from the name.

The plan is built in init_morsel_distributor, in the prefetch task, before a single byte of column data is read (crates/polars-stream/src/nodes/io_sources/parquet/init.rs:197):

rust 
            let has_row_index = row_index.is_some();
            let row_group_mask = calculate_row_group_pred_pushdown_skip_mask(
                row_group_slice.clone(),
                use_statistics,
                predicate.as_ref(),
                &metadata,
                projected_arrow_fields.clone(),
                row_index,
                verbose,
                task_metrics.as_deref(),
                runtime_range_skipper.as_deref(),
            )
            .await?;

            let mut row_groups = row_groups_to_read(
                &metadata,
                row_group_slice,
                row_group_mask.as_ref(),
                row_offset,
                slice_range,
            );

row_group_slice is the result of slice pushdown: LIMIT 10 over a file with 200 row groups narrows the slice to the one row group that contains row 10, so the statistics machinery never even looks at the other 199. row_group_mask is the result of statistics pushdown: a bitmap with one bit per row group, set where the row group cannot match. row_groups_to_read turns the two into a deque of row groups to fetch (crates/polars-stream/src/nodes/io_sources/parquet/row_group_data_fetch.rs:65):

rust 
/// The row groups of `row_group_slice` that `row_group_mask` keeps, in file order. The
/// first starts at `row_offset` and `slice_range` is relative to it.
pub(super) fn row_groups_to_read(
    metadata: &FileMetadata,
    row_group_slice: Range<usize>,
    row_group_mask: Option<&Bitmap>,
    mut row_offset: usize,
    mut slice_range: Option<Range<usize>>,
) -> VecDeque<RowGroupToRead> {
    let mut row_groups = VecDeque::with_capacity(row_group_slice.len());
    for (i, idx) in row_group_slice.enumerate() {
        let num_rows = metadata.row_groups[idx].num_rows();
        let slice = slice_range.as_mut().map(|slice_range| {
            let rg_row_start = slice_range.start;
            let rg_row_end = slice_range.end.min(num_rows);

            *slice_range = slice_range.start.saturating_sub(num_rows)
                ..slice_range.end.saturating_sub(num_rows);

            (rg_row_start, rg_row_end - rg_row_start)
        });
        if !row_group_mask.is_some_and(|mask| mask.get_bit(i)) {
            row_groups.push_back(RowGroupToRead {
                idx,
                row_offset,
                slice,
            });
        }
        row_offset = row_offset.saturating_add(num_rows);
    }
    row_groups
}

Note the mask semantics: a set bit means “skip”. The function keeps the row groups whose bit is clear, and accumulates row_offset over the skipped ones too, because the row offset of a row group is its position in the file, not in the result set. That matters for with_row_index.

The columns in the plan

The other half of the plan is which columns to read, and it is a separate mechanism from the row-group pruning. Projection pushdown has already decided which Arrow fields the query needs; the reader then works out, per row group, which byte ranges those fields live in, and fetches only those (crates/polars-stream/src/nodes/io_sources/parquet/row_group_data_fetch.rs:406):

rust 
fn get_row_group_byte_ranges_for_projection<'a>(
    row_group_metadata: &'a RowGroupMetadata,
    columns: &'a mut dyn Iterator<Item = &PlSmallStr>,
) -> impl Iterator<Item = ParquetResult<std::ops::Range<usize>>> + 'a {
    columns.flat_map(|col_name| {
        row_group_metadata
            .columns_under_root_iter(col_name)
            // `Option::into_iter` so that we return an empty iterator for the
            // `allow_missing_columns` case
            .into_iter()
            .flatten()
            .map(|col| {
                let byte_range = col.byte_range()?;
                Ok(byte_range.start as usize..byte_range.end as usize)
            })
    })
}

columns_under_root_iter is what makes nested schemas work: a projected struct field is not one column but several leaf columns, and the reader has to walk down to the leaves to find their byte ranges. Empty ranges — chunks without pages — are dropped, because they need no I/O.

This is also where the “full projection” shortcut lives: if the query reads every column, the reader skips the per-column walk and just takes the row group’s whole byte range.

The statistics become a DataFrame

Here is the part I find elegant. Polars does not implement a separate “evaluate predicate against min/max bounds” interpreter. Instead it turns the per-row-group statistics into a small DataFrame — one row per row group, one column per statistic — and then runs the user’s own predicate on that DataFrame (crates/polars-stream/src/nodes/io_sources/parquet/statistics.rs:184):

rust 
        for projection in projected_arrow_fields.iter() {
            let c = projection.output_name();

            if !skip_batch_columns.contains(c) {
                continue;
            }

            let mut statistics =
                load_parquet_column_statistics(&metadata, row_group_slice.clone(), projection)?;

            // Note: Order is important here. We re-use the transform for the output column, meaning
            // that it may set the column name.
            statistics.min = projection.apply_transform(statistics.min)?;
            statistics.max = projection.apply_transform(statistics.max)?;
            // The null counts of struct fields have the file's field layout. The transform may
            // reorder, rename or insert fields (e.g. by field ID), and casts the leaves, so it
            // does not apply to them: use unknown null counts.
            if let ArrowFieldProjection::Mapped { output_dtype, .. } = projection
                && matches!(projection.arrow_field().dtype(), ArrowDataType::Struct(_))
            {
                statistics.null_count = Column::full_null(
                    PlSmallStr::EMPTY,
                    statistics.null_count.len(),
                    &null_count_dtype(output_dtype),
                );
            }

            let statistics = statistics.with_base_column_name(c);

            columns.extend([statistics.min, statistics.max, statistics.null_count]);
        }

        if let Some(row_index) = row_index {
            let statistics = build_row_index_statistics(&row_index, row_groups_slice)
                .with_base_column_name(&row_index.name);

            columns.extend([statistics.min, statistics.max, statistics.null_count]);
        }

        let statistics_df = DataFrame::new(num_row_groups, columns)?;

        sbp.evaluate_with_stat_df(&statistics_df)

So a filter like pl.col("ts") > cutoff becomes, internally, a filter over a DataFrame with columns ts_min, ts_max and ts_nc. The predicate compiler already knows how to rewrite ts > cutoff into ts_min > cutoff; the skip batch predicate is that rewrite, evaluated over statistics. One code path for “filter a frame” and “filter a set of row groups”, and no second expression language to maintain.

Two details are worth pulling out. The first is the len column: the number of rows in each row group is pushed into the statistics frame as a plain column, so a predicate like pl.len() > 1000 prunes row groups too, without any special case (crates/polars-stream/src/nodes/io_sources/parquet/statistics.rs:167):

rust 
        let mut columns = Vec::with_capacity(1 + skip_batch_columns.len() * 3);

        let lengths: Vec<IdxSize> = row_groups_slice
            .iter()
            .map(|rg| rg.num_rows() as IdxSize)
            .collect();

        columns.push(Column::new("len".into(), lengths));

The second is that the statistics are pushed through the projection’s transform before the predicate sees them. A projection can rename a column, reorder struct fields, or cast a leaf (projection.apply_transform), and the predicate was compiled against the transformed column. Applying the transform to the statistics as well is what keeps the two in agreement.

The columns are named <col>_min, <col>_max and <col>_nc (crates/polars-io/src/predicates.rs:363):

rust 
            columns.extend([
                Column::new_scalar(format_pl_smallstr!("{col}_min"), min, 1),
                Column::new_scalar(format_pl_smallstr!("{col}_max"), max, 1),
                Column::new_scalar(format_pl_smallstr!("{col}_nc"), nc, 1),
            ]);

The result is a bitmap, one bit per row group, and the verbose log prints the savings (crates/polars-stream/src/nodes/io_sources/parquet/statistics.rs:130):

rust 
    if verbose && (mask.is_some() || runtime_range_skipper.is_some()) {
        let num_row_groups = row_group_slice.len();
        eprintln!(
            "[ParquetFileReader]: Predicate pushdown: \
            reading {} / {} row groups",
            mask.as_ref().map_or(num_row_groups, |m| m.unset_bits()),
            num_row_groups,
        );
    }

Where the numbers come from

The footer is kept in memory as a buffer, and the statistics are stored as byte ranges into it. Resolving a bound is therefore a slice, not a copy (crates/polars-parquet/src/parquet/metadata/column_chunk_metadata.rs:111):

rust 
    /// The plain-encoded min and max of the chunk, borrowed from `footer_buf`;
    /// `None` when the chunk has no statistics.
    pub fn raw_bounds<'a>(&self, footer_buf: &'a [u8]) -> Option<RawBounds<'a>> {
        let stats = self.column_chunk.meta_data.as_ref()?.statistics.as_ref()?;
        Some(RawBounds {
            min: stats.min_value.map(|range| range.resolve(footer_buf)),
            max: stats.max_value.map(|range| range.resolve(footer_buf)),
            min_is_exact: stats.is_min_value_exact != Some(false),
            max_is_exact: stats.is_max_value_exact != Some(false),
        })
    }

The is_exact flags are a nice touch: a writer may record a truncated or approximate bound, and an inexact bound is worthless for pruning (it may exclude rows that do match). Polars validates the bounds and then throws away the inexact ones rather than pruning on them.

Before a bound can be compared, the file’s declared sort order has to match the order Polars compares with — an unsigned column whose statistics were collected as signed cannot be used (crates/polars-parquet/src/arrow/read/statistics/bounds.rs:151):

rust 
/// Whether the `min_value` and `max_value` of a leaf bound the values polars decodes
/// as `dtype`: the file declares an order for the leaf, and it is the one polars
/// compares the decoded values with.
fn bounds_are_usable(
    column_order: ColumnOrder,
    physical_type: &PhysicalType,
    dtype: &ArrowDataType,
) -> bool {
    let Some(required) = required_sort_order(dtype, physical_type) else {
        return false;
    };
    match column_order {
        ColumnOrder::TypeDefinedOrder(order) => order == required,
        // Total order agrees with the float comparison on every non-NaN value.
        ColumnOrder::IEEE754TotalOrder => {
            matches!(physical_type, PhysicalType::Float | PhysicalType::Double)
        },
        ColumnOrder::Unsupported | ColumnOrder::Undefined => false,
    }
}

The same care shows up for timestamps: if the reader multiplies a microsecond timestamp into nanoseconds and the multiplication overflows, the values have no order, so both bounds are required and both must be in range, otherwise the chunk bounds nothing (crates/polars-parquet/src/arrow/read/statistics/bounds.rs:264).

Nested columns need one more step. A struct column is not one Parquet leaf but several, and a predicate on a single struct field should be able to prune on that field’s statistics. load_struct_column_statistics therefore assembles a per-field min/max/null-count structure from the leaves, and gives up — falls back to all-null statistics, which prune nothing — if the leaf count and the struct field count do not line up (crates/polars-stream/src/nodes/io_sources/parquet/statistics.rs:732).

The “already applied” flag

There is a second level of pruning that happens before the reader ever sees the file: table statistics. If a Parquet dataset has per-file min/max statistics recorded out of band, a predicate can be evaluated against them and whole files skipped. When that happens, the reader must not evaluate the predicate again — or rather, the plan has to know whether there is anything left to evaluate. That is what PredicateFileSkip is for (crates/polars-descriptions/src/lib.rs:21):

rust 
pub struct PredicateFileSkip {
    /// If `true` the predicate can be skipped at runtime.
    pub no_residual_predicate: bool,
    /// Number of files before skipping
    pub original_len: usize,
}

It is set when the file-level statistics are applied (crates/polars-mem-engine/src/scan_predicate/functions.rs:563):

rust 
    let predicate_to_readers = (!can_skip_scan_predicate).then_some(&scan_predicate);

    if let Some(skip_files_mask) = skip_files_mask {
        assert_eq!(skip_files_mask.len(), sources.len());

        let predicate_file_skip = PredicateFileSkip {
            no_residual_predicate: predicate_to_readers.is_none(),
            original_len: sources.len(),
        };

and honoured when the pipeline is initialised: with no_residual_predicate set, the predicate is simply not handed to the reader (crates/polars-stream/src/nodes/io_sources/multi_scan/pipeline/initialization.rs:124):

rust 
        Some(PredicateFileSkip {
            no_residual_predicate: false,
            original_len: _,
        }) => (None, config.predicate.as_ref()),
        Some(PredicateFileSkip {
            no_residual_predicate: true,
            original_len: _,
        }) => (None, None),

The one subtlety is runtime ranges: a predicate with a runtime range is not consumed by the file-level statistics, because the range can change while the scan runs. That is why can_skip_scan_predicate is predicate.runtime_ranges.is_empty() in the hive-partition branch (crates/polars-mem-engine/src/scan_predicate/functions.rs:407).

Ranges that arrive late

The runtime range is the case where a join discovers a filter after the scan has started. The plan cannot prune on it, but the reader can still check each row group against it just before fetching it. RuntimeRangeSkipper resolves the range against the file’s statistics once (crates/polars-stream/src/nodes/io_sources/parquet/statistics.rs:540):

rust 
    fn skips_with(&self, checks: &[RangeCheck], row_group_idx: usize) -> bool {
        checks.iter().any(|check| match check {
            RangeCheck::KeepAll => false,
            RangeCheck::SkipAll => true,
            RangeCheck::Bounds {
                leaf,
                lo,
                hi,
                nulls_match,
            } => {
                let leaf = self.hints[*leaf].leaf.as_ref().unwrap();
                let Some((min, max)) = leaf.row_group_bounds(row_group_idx, &self.metadata) else {
                    return false;
                };
                let below = lo
                    .as_ref()
                    .is_some_and(|lo| max.as_ref().is_some_and(|max| max < lo));
                let above = hi
                    .as_ref()
                    .is_some_and(|hi| min.as_ref().is_some_and(|min| min > hi));
                (below || above)
                    && (!nulls_match
                        || has_no_nulls(&self.metadata.row_groups[row_group_idx], leaf.leaf_idx))
            },
        })
    }

and the fetcher re-checks the front of the queue before each fetch, so a range that tightened since the plan was made still skips work (crates/polars-stream/src/nodes/io_sources/parquet/row_group_data_fetch.rs:139):

rust 
    /// Moves past the row groups at the front that the runtime ranges skip.
    fn skip_front(&mut self) {
        while let Some(row_group) = self.row_groups.front()
            && self
                .runtime_range_skipper
                .as_ref()
                .is_some_and(|s| s.skips(row_group.idx))
        {
            self.row_groups.pop_front();
        }
    }

The page index: written, not read

The write side is where the page index shows up. When the footer is written, Polars serialises a column index and an offset index for every column chunk — the column index only if the writer was configured to collect statistics, the offset index unconditionally (crates/polars-parquet/src/parquet/write/file.rs:205):

rust 
        if self.options.write_statistics {
            // write column indexes (require page statistics)
            self.row_groups
                .iter_mut()
                .zip(self.page_specs.iter())
                .try_for_each(|(group, pages)| {
                    group.columns.iter_mut().zip(pages.iter()).try_for_each(
                        |(column, pages)| {
                            let offset = self.offset;
                            if let Some(length) = write_column_index(&mut self.writer, pages)? {
                                self.offset += length;
                                column.column_index_offset = Some(offset as i64);
                                column.column_index_length = Some(length as i32);
                            }
                            ParquetResult::Ok(())
                        },
                    )?;
                    ParquetResult::Ok(())
                })?;
        };

and the offset index right after it, unconditionally. So a file written by Polars carries per-page min/max and per-page offsets, and any reader that implements the page index can use them.

Polars’ own reader does not. The vendored page reader still carries the comment from the crate it was forked from (crates/polars-parquet/src/parquet/read/page/reader.rs:68):

rust 
/// A fallible [`Iterator`] of [`CompressedDataPage`]. This iterator reads pages back
/// to back until all pages have been consumed.
///
/// The pages from this iterator always have [`None`] [`crate::parquet::page::CompressedDataPage::selected_rows()`] since
/// filter pushdown is not supported without a
/// pre-computed [page index](https://github.com/apache/parquet-format/blob/master/PageIndex.md).
pub struct PageReader {

and the read path never opens the column index or the offset index: the only references to them in the reader are the metadata accessors that tell you where the index is, not what is in it.

The bottom line

  • Pruning is per row group, using the per-column-chunk min/max/null-count in the footer. The bounds are borrowed from the footer buffer, not copied, and are dropped if the file’s sort order or an exactness flag makes them unusable.
  • The predicate is not re-implemented for statistics. It is evaluated on a DataFrame of <col>_min/<col>_max/<col>_nc columns, which means the filter you wrote and the filter that prunes row groups are the same expression.
  • PredicateFileSkip::no_residual_predicate is the “already applied” flag: it tells the reader that table statistics already consumed the predicate, so the reader must not evaluate it again.
  • A range published at run time is checked again per row group, immediately before the fetch.
  • Polars writes the page index (with statistics enabled) so that other engines can prune at page granularity. If you want that from Polars itself, it is not there yet, and the win would be largest on files with many small row groups and highly selective predicates.

PolarsInternalsParquetPerformanceStreaming