DataFusion's Parquet Push Decoder: Pruning Row Groups in the Middle of a Scan You Already Started

TL;DR: Most Parquet readers are pull-based: you ask for a batch, they decode it. DataFusion’s push decoder inverts that — you push byte ranges into it and it tells you what it needs next. Because the decoder is paused at row-group boundaries rather than in the middle of one, a filter that tightened while the scan was running can still be applied to every row group that has not been fetched yet. The scan does not need to be restarted, and the bytes that were never fetched are never paid for.

Every Parquet reader eventually has to answer the same question: given a predicate, which row groups can be skipped entirely? The standard answer is “decide once, at open time, from the footer statistics”. That works when the predicate is known at plan time.

But DataFusion now has predicates that are not known at plan time — a TopK that tightens its threshold as it fills its heap, a hash join that publishes its build-side bounds once it has read the build side. Those facts arrive mid-scan. If the reader decided everything at open time, the new fact would be useless for the file currently being read.

The push decoder exists to close that gap.

Pull vs push

The classic arrow-rs reader is pull-based:

text

reader.next()  ->  decode a batch from the current row group
                   (fetch bytes as needed, behind the scenes)

The push decoder is the other way round:

text
decoder.try_next_reader()
    -> NeedsData(ranges)     "give me these byte ranges"
    -> Data(reader)          "here is a decoded reader for this row group"
    -> Finished              "no more"

The driver in datafusion/datasource-parquet/src/push_decoder.rs is a futures::stream::unfold over a state machine:

rust
pub(crate) fn into_stream(self) -> BoxStream<'static, Result<RecordBatch>> {
    futures::stream::unfold(self, |state| async move { state.transition().await })
        .fuse()
        .boxed()
}

and the state machine itself alternates between fetching bytes and decoding:

rust
// Step 3: drive the decoder.
let decoder = self.decoder.as_mut().expect("decoder present");
match decoder.try_next_reader() {
    Ok(DecodeResult::NeedsData(ranges)) => {
        // I/O, not compute.
        timer.stop();
        let data = self.reader.get_byte_ranges(ranges.clone()).await ...;
        timer.restart();
        self.decoder.as_mut().unwrap().push_ranges(ranges, data) ...
    }
    Ok(DecodeResult::Data(reader)) => {
        // Pop the RG this reader is for ...
        if let Some(entry) = self.rg_plan.pop_front() {
            self.byte_progress.credit(entry.bytes);
        }
        self.active_reader = Some(reader);
    }
    Ok(DecodeResult::Finished) => return None,
    ...
}

Notice the timer.stop() / timer.restart() around the I/O await. The elapsed_compute metric measures compute, and the fetch is not compute. This is a small thing, but it is the kind of thing that matters: without it, every metric that claims to measure CPU time is really measuring CPU time plus object-store latency, and no one can tell the difference.

Row-group boundaries are the checkpoint

The whole design rests on one property: the decoder can be paused at a row-group boundary and its plan re-derived. The code is explicit about where this is allowed:

rust
// `into_builder` errors out mid-row-group, so we gate the prune
// pass on `is_at_row_group_boundary()`. When the decoder is
// mid-RG (e.g. byte ranges have been pushed but no reader has
// been handed back yet), step 3 drives it forward and we get
// another chance at the next boundary — the pruner is stateful
// and idempotent, so deferring loses nothing.
let at_boundary = self
    .decoder
    .as_ref()
    .expect("decoder present")
    .is_at_row_group_boundary();

So the loop is:

Drain a batch from the active reader, if there is one. Return it.
If we are at a row-group boundary, re-derive the row-group plan and rebuild the decoder if anything was dropped.
Drive the decoder: fetch bytes or hand back a reader.

Step 2 is where the dynamic filter pays off.

The row-group plan

The scan keeps a plan of the row groups it intends to read:

rust
#[derive(Debug, Clone)]
pub(crate) struct RgPlanEntry {
    pub(crate) rg_index: usize,
    /// `true` when static pruning proved every row of this RG satisfies the
    /// predicate, so the per-row `RowFilter` can be skipped as a no-op.
    pub(crate) fully_matched: bool,
    /// On-disk size of this row group, credited to the `bytes_processed` metric
    /// documented on `ParquetFileMetrics` once the scan is done with it.
    pub(crate) bytes: u64,
}

and at a boundary, the whole remaining plan is scanned and pruned at once:

rust
// Step 2: when the decoder is sitting on a row-group boundary,
// scan the entire `rg_plan` and drop every RG the pruner proves
// cannot contribute — head, interior, and tail alike. Evaluating
// per-RG stats against the cached `PruningPredicate` is cheap;
// the expensive part is the `into_builder` rebuild, so we do at
// most one rebuild per boundary regardless of how many RGs were
// dropped. Buffered bytes for already-fetched RGs carry across
// the rebuild.

The phrase “at most one rebuild per boundary” is the key efficiency insight. Rebuilding the decoder is expensive (you have to construct a new reader with new selections). Evaluating a pruning predicate against a row group’s statistics is cheap. So the algorithm prunes many row groups and then pays the rebuild cost once.

The runtime pruner

RowGroupPruner is the piece that answers “should this row group be dropped right now?”:

rust
pub(crate) struct RowGroupPruner {
    predicate: Arc<dyn PhysicalExpr>,
    arrow_schema: SchemaRef,
    parquet_metadata: Arc<ParquetMetaData>,
    /// Classifies the predicate's dynamic-filter content.
    tracking: DynamicFilterTracking,
    /// First-call sentinel: forces an initial `pruning_predicate` build
    /// even when `tracking` is `Static` / `AllComplete`.
    needs_initial_build: bool,
    /// Cached pruning predicate. `None` means we couldn't build one for the
    /// current generation (e.g. the predicate has no analyzable bounds);
    /// in that case we conservatively don't prune.
    pruning_predicate: Option<Arc<PruningPredicate>>,
    ...
}

The needs_initial_build flag exists because DynamicFilterTracking::classify distinguishes “no dynamic filters” from “dynamic filters, all complete” — and in the latter case the current value of the filter may already be different from what planning assumed, so a one-shot prune is still worth doing, but there is nothing to watch afterwards.

The check itself is where the “cheap atomic load” trick from the previous article in this series pays off:

rust
pub(crate) fn should_prune(&mut self, row_group_indices: &[usize]) -> bool {
    if row_group_indices.is_empty() {
        return false;
    }

    // Refresh the cached `PruningPredicate` on the first call and
    // whenever a watched dynamic filter has advanced since we last
    // looked. `changed()` is a single atomic load per still-incomplete
    // filter — no tree walk on every check.
    let dynamic_changed = self
        .tracking
        .watcher()
        .is_some_and(|tracker| tracker.changed());
    if self.needs_initial_build || dynamic_changed {
        self.pruning_predicate = PruningPredicateBuilder::new()
            .with_file_schema(Arc::clone(&self.arrow_schema))
            .with_error_counter(&self.predicate_creation_errors)
            .with_max_in_list_size(self.max_in_list_size)
            .build(Arc::clone(&self.predicate));
        self.needs_initial_build = false;
    }
    ...

Building a PruningPredicate involves real expression analysis. Doing it once per row group would dominate the cost of pruning. Doing it once per filter generation is nearly free.

And the failure policy is conservative in the right direction:

rust
match pp.prune(&stats) {
    // `prune` returns `false` per container that the predicate proves
    // cannot contain matching rows. We can skip the run only when
    // every requested row group is in that state.
    Ok(values) => values.iter().all(|&keep| !keep),
    Err(e) => {
        debug!("Ignoring error evaluating runtime row-group pruning predicate: {e}");
        self.predicate_evaluation_errors.add(1);
        false
    }
}

Two safety properties: a row group is dropped only when every container in it is prunable, and any error means “do not prune”. A flaky pruning path never silently drops data.

The other half: stopping the whole file

The row-group pruner handles the tail of a file. The file-level pruner handles the case where the tightened filter proves that no row group can match — at which point the whole file should be abandoned.

EarlyStoppingStream (datafusion/datasource-parquet/src/opener/early_stop.rs) is a RecordBatchStream adapter that re-checks the file pruner on every batch:

rust
fn check_prune(&mut self, input: Result<RecordBatch>) -> Result<Option<RecordBatch>> {
    let batch = input?;

    // Since dynamic filters may have been updated, see if we can stop
    // reading this stream entirely.
    if self.file_pruner.should_prune()? {
        self.files_ranges_pruned_statistics.add_pruned(1);
        // Previously this file range has been counted as matched
        self.files_ranges_pruned_statistics.subtract_matched(1);
        self.finish();
        Ok(None)
    } else {
        // Return the adapted batch
        Ok(Some(batch))
    }
}

The metric adjustment is a nice detail: the file was counted as matched when it was opened, and now it is being reclassified as pruned, so the counters have to be corrected or the metrics will lie.

The struct holds the inner stream as Option<S> specifically so that finishing releases the decoder immediately:

rust
/// The inner stream, dropped as soon as this stream is done with it.
///
/// Held as an `Option` so finishing releases the decoder — and the buffers
/// and per-file metric state it owns — at the moment we stop reading, not
/// whenever the caller gets around to dropping this wrapper. Notably the
/// scan's byte-progress accounting is completed by that drop, so deferring
/// it would leave the file reading as partially scanned after the scan had
/// demonstrably finished with it.
inner: Option<S>,

That is a genuinely subtle piece of engineering: the drop of the parquet reader is what finalises the “bytes processed” accounting. If the adapter merely stopped polling and let the drop happen later — or never — the metrics would report a partially-scanned file after the scan had finished with it.

The catch

The decoder must be rebuildable. Pausing at a row-group boundary requires the decoder to support being torn down and reconstructed with a new set of row-group selections. That is a real constraint on arrow-rs’s decoder API, and it is why the push decoder is a separate code path rather than a flag on the ordinary reader.
Statistics must be good enough. Runtime pruning is only as good as the Parquet footer. A file written without statistics, or with statistics that do not reflect the actual data, cannot be pruned at runtime — and, worse, wrong statistics would cause rows to be dropped.
The rebuild is not free. One rebuild per boundary is fine. If a dynamic filter updated on every batch, you would rebuild the decoder on every batch. The design assumes updates are rare (a TopK threshold refresh, a hash join publishing once).
Alignment bugs are the main hazard. The comment on sync_rg_plan_to_decoder_frontier is a war story:
arrow-rs silently finishes row groups whose post-predicate selection is empty without handing back a reader, so without this sync rg_plan trails the decoder by one and a rebuild either re-reads an already delivered row group (#24352) or toggles the per-RG filter for the wrong row group.
A one-row-group skew between “the plan we think we are executing” and “the row group the decoder is actually on” produces either duplicated rows or a filter applied to the wrong data. Both are correctness bugs. The reference to issue #24352 suggests this bit at least once.

Why this matters

Consider a SELECT ... ORDER BY x LIMIT 10 over a Parquet table. The TopK operator learns, after the first few batches, that only values greater than 417 are interesting. If the scan can be told that, the remaining row groups of every file can be skipped at runtime — not because the planner knew the threshold, but because the consumer told the producer.

That is the shape of the whole dynamic-filter idea, and it is why DataFusion treats DynamicFilterPhysicalExpr as a first-class physical expression rather than a special case bolted onto one operator.