You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/08/25 03:34:48 UTC
(datafusion) 01/01: perf(parquet): skip RowFilter on statically fully-matched row groups (#23696)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23696-c6999963455dae90f11130031ce8ee7d6add2c32
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 50f45f5878609c06aa6903de10af199a781ef032
Author: Qi Zhu <zh...@gmail.com>
AuthorDate: Tue Aug 25 03:34:22 2026 +0000
perf(parquet): skip RowFilter on statically fully-matched row groups (#23696)
## Which issue does this PR close?
- Closes #23067.
## Rationale for this change
When Parquet statistics prove that **every** row of a row group already
satisfies the pushdown predicate (`fully_matched`, tracked by
`ParquetAccessPlan` since #21637), running the per-row `RowFilter`
inside that row group is pure overhead — every row passes anyway. This
PR teaches the push-decoder stream to **toggle the `RowFilter` off
across fully-matched runs** and back on at the next straddling row
group, so fully-matched runs decode without per-row evaluation (and
two-phase reads skip fetching filter-only columns for them). Common
shape: time-partitioned tables (`WHERE ts >= X AND ts < Y` with whole
row groups inside `[X, Y)`) and the middle stretch of a TopK scan.
This PR was decomposed per [@adriangb's
proposal](https://github.com/apache/datafusion/pull/23696#issuecomment-5346992228);
the standalone pieces already merged:
- #24509 — generic `strip_empty_row_groups` in
`PreparedAccessPlan::prepare` (#24287)
- #24572 — `InitialDecoderState` extraction (#24286)
What remains here is the feature itself plus the prebuild split it
consumes (the proposal's PR 2 is **folded in** rather than standalone:
on `main`, `into_builder()` preserves the `RowFilter` across
runtime-prune rebuilds, so a prebuilt candidate list would have no
consumer and no perf story on its own — the per-RG toggle here is its
only consumer).
## What changes are included in this PR?
- `access_plan.rs` — `PreparedAccessPlan` carries per-RG `fully_matched`
flags, mapped from the access plan **after** the (generic)
empty-row-group strip; `reorder_by_statistics` / `reverse` permute them
alongside the indexes.
- `row_filter.rs` — split into `prebuild_row_filter_candidates` (once
per file: conjunct split, candidate construction,
`reassign_expr_columns`) and `row_filter_from_prebuilt` (per boundary:
metric wiring + optional `required_bytes` ordering, no tree walks). The
public `build_row_filter` is **reimplemented on top of the two**, so the
open-time path and the per-RG rebuild path share one
conjunct-split/order/metric implementation and cannot drift.
`DatafusionArrowPredicate::try_new` becomes a `#[cfg(test)]`
convenience.
- `push_decoder.rs` — `RgPlanEntry.fully_matched`; `RowFilterContext`
(the prebuilt list + settings) lets `rebuild_decoder_at_boundary` swap
between the real filter and an empty one as the stream crosses row-group
boundaries; at most one `into_builder` rebuild per boundary, shared with
the runtime pruner.
- `opener/mod.rs` — prebuilds once per file, installs the initial filter
based on the first row group's `fully_matched` state (two new fields on
`InitialDecoderState`).
- `metrics.rs` — new `row_filter_skipped_fully_matched` counter,
**registered lazily on first suppression**
(`RowFilterSkippedFullyMatchedMetric`), following the
`page_index_pages_skipped_by_fully_matched` pattern from #21637: no
public field on `ParquetFileMetrics` (no API/semver change) and no
zero-valued entries in `EXPLAIN ANALYZE` baselines — the
previously-regenerated 41 baselines across 3 slt files are untouched by
this PR. It counts *suppression events*, not row groups: a run of
consecutive fully-matched RGs shares one toggle.
Considered but deferred: carrying one `Vec<{index, fully_matched}>`
instead of the two positionally-aligned vectors. Alignment is now
maintained in exactly two adjacent permutations
(`reorder_by_statistics`, `reverse`) plus a `debug_assert`, and the type
change would churn ~40 assertion lines in `main`-side tests — the
opposite of shrinking this PR. Happy to do it as a tiny follow-up.
## Are these changes tested?
- `fully_matched_rgs_skip_row_filter` — 4 RGs, predicate `v >= 3 AND v
<= 10` makes RG 0 and RG 3 straddlers with a fully-matched run in
between, covering the full toggle lifecycle (ON → OFF → back ON);
asserts correctness, that 11/12 are filtered out by the reinstalled
filter, and `row_filter_skipped_fully_matched >= 1`.
- `dynamic_row_group_pruning.slt` — an `EXPLAIN ANALYZE` case surfacing
`row_filter_skipped_fully_matched=1`, so the optimization is observable
from SQL and a regression shows up as the metric disappearing.
- All existing `dynamic_row_group_pruning` integration tests (10), the
`datafusion-datasource-parquet` lib suite (240), and the affected slt
files pass; `clippy --all-targets --all-features -D warnings` clean.
## Are there any user-facing changes?
- `row_filter_skipped_fully_matched` appears in `EXPLAIN ANALYZE` on
parquet scans **only when the toggle actually fired**.
- No public API changes; `build_row_filter` keeps its signature and
behavior.
---
.../tests/parquet/dynamic_row_group_pruning.rs | 88 +++++++
datafusion/datasource-parquet/src/access_plan.rs | 57 +++--
datafusion/datasource-parquet/src/metrics.rs | 51 +++++
datafusion/datasource-parquet/src/opener/mod.rs | 94 ++++++--
datafusion/datasource-parquet/src/push_decoder.rs | 253 +++++++++++++++++----
datafusion/datasource-parquet/src/row_filter.rs | 198 ++++++++--------
datafusion/datasource-parquet/src/sort.rs | 7 +-
.../test_files/dynamic_row_group_pruning.slt | 42 ++++
8 files changed, 620 insertions(+), 170 deletions(-)
diff --git a/datafusion/core/tests/parquet/dynamic_row_group_pruning.rs b/datafusion/core/tests/parquet/dynamic_row_group_pruning.rs
index 5ee42b3067..678aad9f5b 100644
--- a/datafusion/core/tests/parquet/dynamic_row_group_pruning.rs
+++ b/datafusion/core/tests/parquet/dynamic_row_group_pruning.rs
@@ -641,6 +641,94 @@ fn build_q26_batches(schema: &Arc<Schema>) -> Vec<RecordBatch> {
.collect()
}
+/// Per-RG `fully_matched` `RowFilter` skip optimization.
+///
+/// Stats prove that every row of a fully-matched row group satisfies the
+/// pushdown predicate, so the parquet decoder can skip the per-row
+/// `RowFilter` for that RG entirely. The stream rebuilds the decoder at
+/// the boundary with an empty `RowFilter` and toggles back to the real
+/// one at the next non-fully-matched RG.
+///
+/// Layout: 4 RGs of 3 values each. Predicate `v >= 3 AND v <= 10` makes
+/// RG 0 a straddler (1, 2 fail the lower bound), RGs 1..=2 fully matched
+/// (every value in [3, 10] by stats), and RG 3 a straddler again (11, 12
+/// fail the upper bound). This exercises the full toggle lifecycle:
+/// filter ON (RG 0) → OFF across the fully-matched run (RGs 1..=2) → back
+/// ON (RG 3), covering both the fully-matched → non-fully-matched and the
+/// reverse transition.
+///
+/// Expected behavior:
+/// - the static prune marks RGs 1..=2 as fully_matched at file open;
+/// - the stream installs the real `RowFilter` initially (RG 0 not fm);
+/// - at the RG 0 → RG 1 boundary the toggle rebuilds with an empty filter
+/// and bumps `row_filter_skipped_fully_matched`;
+/// - at the RG 2 → RG 3 boundary the toggle reinstalls the real filter, so
+/// 11 and 12 are correctly excluded;
+/// - the query result is identical to running with the filter on.
+#[tokio::test]
+async fn fully_matched_rgs_skip_row_filter() {
+ let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)]));
+ // 4 RGs of 3 rows each. Predicate `v >= 3 AND v <= 10`:
+ // RG 0: 1, 2, 3 ← keeps {3}; min=1,max=3 → straddler, filter ON
+ // RG 1: 4, 5, 6 ← all in [3,10] → fully matched, filter OFF
+ // RG 2: 7, 8, 9 ← fully matched, filter OFF
+ // RG 3: 10,11,12 ← keeps {10}; 11,12 fail v<=10 → straddler, filter back ON
+ let groups: [[i64; 3]; 4] = [[1, 2, 3], [4, 5, 6], [7, 8, 9], [10, 11, 12]];
+ let batches: Vec<RecordBatch> = groups
+ .iter()
+ .map(|vals| {
+ let col: ArrayRef = Arc::new(Int64Array::from(vals.to_vec()));
+ RecordBatch::try_new(Arc::clone(&schema), vec![col]).unwrap()
+ })
+ .collect();
+
+ let mut ctx = ContextWithParquet::with_custom_data(
+ Scenario::Int,
+ RowGroup(3),
+ Arc::clone(&schema),
+ batches,
+ )
+ .await;
+
+ let output = ctx
+ .query("SELECT v FROM t WHERE v >= 3 AND v <= 10 ORDER BY v ASC")
+ .await;
+
+ // Correctness: every value in [3, 10], ascending.
+ let expected_rows: Vec<i64> = (3..=10).collect();
+ assert_eq!(output.result_rows, expected_rows.len());
+ let formatted = output.pretty_results();
+ for v in expected_rows {
+ assert!(
+ formatted.contains(&format!("| {v} ")),
+ "output must contain {v}; got:\n{formatted}",
+ );
+ }
+ // The RG 2 → RG 3 transition (fully-matched → non-fully-matched) must
+ // reinstall the real filter, so 11 and 12 are filtered out. If the
+ // toggle failed to restore the filter they would leak through.
+ for v in [11i64, 12] {
+ assert!(
+ !formatted.contains(&format!("| {v} ")),
+ "value {v} must be filtered out by the reinstalled RowFilter; \
+ got:\n{formatted}",
+ );
+ }
+
+ // Behavior: the per-RG `RowFilter` toggle must have fired at least
+ // once when transitioning from RG 0 (not fm) into the fully-matched
+ // run RGs 1..=2.
+ let skipped = output
+ .metric_value("row_filter_skipped_fully_matched")
+ .unwrap_or(0);
+ assert!(
+ skipped >= 1,
+ "row_filter_skipped_fully_matched must fire at least once; \
+ skipped={skipped}\n{}",
+ output.description(),
+ );
+}
+
/// Regression for #24352: with `pushdown_filters` + TopK dynamic filter, a row
/// group whose post-predicate selection is empty is silently finished by
/// arrow-rs without handing back a reader. Before `rg_plan` was synced to the
diff --git a/datafusion/datasource-parquet/src/access_plan.rs b/datafusion/datasource-parquet/src/access_plan.rs
index 6d96c68e18..b837159cf0 100644
--- a/datafusion/datasource-parquet/src/access_plan.rs
+++ b/datafusion/datasource-parquet/src/access_plan.rs
@@ -567,16 +567,28 @@ impl ParquetAccessPlan {
/// Prepare this plan and resolve to the final `PreparedAccessPlan`
pub(crate) fn prepare(
- self,
+ mut self,
row_group_meta_data: &[RowGroupMetaData],
) -> Result<PreparedAccessPlan> {
let row_group_indexes = self.row_group_indexes();
+ // `fully_matched` is indexed by absolute row-group index; take it out
+ // before `into_overall_row_selection` consumes `self`.
+ let fully_matched_by_index = std::mem::take(&mut self.fully_matched);
let row_selection = self.into_overall_row_selection(row_group_meta_data)?;
let (row_group_indexes, row_selection) =
strip_empty_row_groups(row_group_indexes, row_selection, row_group_meta_data);
- PreparedAccessPlan::new(row_group_indexes, row_selection)
+ // Carry `fully_matched` flags in the same order as the *surviving*
+ // `row_group_indexes` so downstream code (per-RG `RowFilter` skip) can
+ // look them up positionally. Mapping after the strip keeps
+ // `strip_empty_row_groups` generic (no `fully_matched` parameter).
+ let fully_matched: Vec<bool> = row_group_indexes
+ .iter()
+ .map(|&idx| fully_matched_by_index[idx])
+ .collect();
+
+ PreparedAccessPlan::new(row_group_indexes, fully_matched, row_selection)
}
}
@@ -660,6 +672,11 @@ fn strip_empty_row_groups(
pub(crate) struct PreparedAccessPlan {
/// Row group indexes to read
pub(crate) row_group_indexes: Vec<usize>,
+ /// Per-RG `fully_matched` flag, positionally aligned with
+ /// [`Self::row_group_indexes`]. A `true` entry means stats already
+ /// proved every row of this RG passes the predicate, so the per-row
+ /// `RowFilter` can be skipped for it.
+ pub(crate) fully_matched: Vec<bool>,
/// Optional row selection for filtering within row groups
pub(crate) row_selection: Option<RowSelection>,
}
@@ -668,10 +685,13 @@ impl PreparedAccessPlan {
/// Create a new prepared access plan
fn new(
row_group_indexes: Vec<usize>,
+ fully_matched: Vec<bool>,
row_selection: Option<RowSelection>,
) -> Result<Self> {
+ debug_assert_eq!(row_group_indexes.len(), fully_matched.len());
Ok(Self {
row_group_indexes,
+ fully_matched,
row_selection,
})
}
@@ -825,13 +845,18 @@ impl PreparedAccessPlan {
}
};
- // Apply the reordering
+ // Apply the reordering — `fully_matched` must be permuted alongside
+ // `row_group_indexes` so the two stay positionally aligned for the
+ // per-RG `RowFilter` skip path.
let original_indexes = self.row_group_indexes.clone();
- self.row_group_indexes = sorted_indices
+ let original_fully_matched = self.fully_matched.clone();
+ let order: Vec<usize> = sorted_indices
.values()
.iter()
- .map(|&i| original_indexes[i as usize])
+ .map(|&i| i as usize)
.collect();
+ self.row_group_indexes = order.iter().map(|&i| original_indexes[i]).collect();
+ self.fully_matched = order.iter().map(|&i| original_fully_matched[i]).collect();
Ok(self)
}
@@ -841,8 +866,9 @@ impl PreparedAccessPlan {
// Get the row group indexes before reversing
let row_groups_to_scan = self.row_group_indexes.clone();
- // Reverse the row group indexes
+ // Reverse the row group indexes (and the parallel `fully_matched`)
self.row_group_indexes = self.row_group_indexes.into_iter().rev().collect();
+ self.fully_matched = self.fully_matched.into_iter().rev().collect();
// If we have a row selection, reverse it to match the new row group order
if let Some(row_selection) = self.row_selection {
@@ -1233,7 +1259,7 @@ mod test {
#[test]
fn reorder_by_statistics_sorts_row_groups_asc_by_min() {
let metadata = parquet_metadata_with_int_mins(&[50, 10, 100]);
- let plan = PreparedAccessPlan::new(vec![0, 1, 2], None).unwrap();
+ let plan = PreparedAccessPlan::new(vec![0, 1, 2], vec![false; 3], None).unwrap();
let result = plan
.reorder_by_statistics(
@@ -1252,7 +1278,8 @@ mod test {
fn reorder_by_statistics_skips_when_row_selection_present() {
let metadata = parquet_metadata_with_int_mins(&[50, 10]);
let selection = RowSelection::from(vec![RowSelector::select(100)]);
- let plan = PreparedAccessPlan::new(vec![0, 1], Some(selection)).unwrap();
+ let plan =
+ PreparedAccessPlan::new(vec![0, 1], vec![false; 2], Some(selection)).unwrap();
let result = plan
.reorder_by_statistics(
@@ -1269,7 +1296,7 @@ mod test {
#[test]
fn reorder_by_statistics_skips_when_at_most_one_row_group() {
let metadata = parquet_metadata_with_int_mins(&[50]);
- let plan = PreparedAccessPlan::new(vec![0], None).unwrap();
+ let plan = PreparedAccessPlan::new(vec![0], vec![false; 1], None).unwrap();
let result = plan
.reorder_by_statistics(
@@ -1288,7 +1315,7 @@ mod test {
#[test]
fn reorder_by_statistics_skips_for_non_column_sort_expr() {
let metadata = parquet_metadata_with_int_mins(&[50, 10]);
- let plan = PreparedAccessPlan::new(vec![0, 1], None).unwrap();
+ let plan = PreparedAccessPlan::new(vec![0, 1], vec![false; 2], None).unwrap();
let arrow_schema = arrow_schema_a_int();
let order = LexOrdering::new(vec![PhysicalSortExpr {
expr: Arc::new(BinaryExpr::new(
@@ -1317,7 +1344,7 @@ mod test {
#[test]
fn reorder_by_statistics_skips_when_column_not_in_arrow_schema() {
let metadata = parquet_metadata_with_int_mins(&[50, 10]);
- let plan = PreparedAccessPlan::new(vec![0, 1], None).unwrap();
+ let plan = PreparedAccessPlan::new(vec![0, 1], vec![false; 2], None).unwrap();
// Arrow schema only has "a"; the sort references "b".
let arrow_schema = arrow_schema_a_int();
let order = LexOrdering::new(vec![PhysicalSortExpr {
@@ -1420,7 +1447,7 @@ mod test {
fn reorder_by_statistics_breaks_leading_ties_with_secondary_column() {
let metadata =
parquet_metadata_with_two_col_mins(&[(1, 300), (1, 100), (1, 200)]);
- let plan = PreparedAccessPlan::new(vec![0, 1, 2], None).unwrap();
+ let plan = PreparedAccessPlan::new(vec![0, 1, 2], vec![false; 3], None).unwrap();
let order =
LexOrdering::new(vec![sort_expr("a", 0, false), sort_expr("b", 1, false)])
.unwrap();
@@ -1439,7 +1466,7 @@ mod test {
fn reorder_by_statistics_honors_secondary_direction() {
let metadata =
parquet_metadata_with_two_col_mins(&[(1, 100), (1, 300), (0, 500)]);
- let plan = PreparedAccessPlan::new(vec![0, 1, 2], None).unwrap();
+ let plan = PreparedAccessPlan::new(vec![0, 1, 2], vec![false; 3], None).unwrap();
let order =
LexOrdering::new(vec![sort_expr("a", 0, false), sort_expr("b", 1, true)])
.unwrap();
@@ -1459,7 +1486,7 @@ mod test {
fn reorder_by_statistics_normalizes_desc_desc_for_reverse() {
let metadata =
parquet_metadata_with_two_col_mins(&[(1, 300), (2, 100), (1, 100)]);
- let plan = PreparedAccessPlan::new(vec![0, 1, 2], None).unwrap();
+ let plan = PreparedAccessPlan::new(vec![0, 1, 2], vec![false; 3], None).unwrap();
let order =
LexOrdering::new(vec![sort_expr("a", 0, true), sort_expr("b", 1, true)])
.unwrap();
@@ -1479,7 +1506,7 @@ mod test {
fn reorder_by_statistics_keeps_leading_prefix_on_non_column_secondary() {
let metadata =
parquet_metadata_with_two_col_mins(&[(5, 300), (3, 100), (4, 200)]);
- let plan = PreparedAccessPlan::new(vec![0, 1, 2], None).unwrap();
+ let plan = PreparedAccessPlan::new(vec![0, 1, 2], vec![false; 3], None).unwrap();
let order = LexOrdering::new(vec was suppressed
+/// because static stats proved every row of the upcoming row group(s)
+/// satisfies the predicate.
+///
+/// Like [`ParquetFileMetrics::add_page_index_pages_skipped_by_fully_matched`],
+/// the counter is only registered when it first fires, so scans that never
+/// suppress a row filter don't carry a zero-valued counter in
+/// `EXPLAIN ANALYZE` (and `ParquetFileMetrics` keeps no public field for it).
+/// Unlike that fire-once helper, the decode stream records suppressions as
+/// they happen, so this holder keeps a live [`Count`] handle after the first
+/// registration.
+///
+/// Note this counts *suppression events*, not row groups: a run of
+/// consecutive fully-matched row groups shares a single toggle (the filter
+/// stays off across the run with no further rebuilds).
+pub(crate) struct RowFilterSkippedFullyMatchedMetric {
+ metrics: ExecutionPlanMetricsSet,
+ partition: usize,
+ filename: String,
+ count: Option<Count>,
+}
+
+impl RowFilterSkippedFullyMatchedMetric {
+ pub(crate) fn new(
+ metrics: &ExecutionPlanMetricsSet,
+ partition: usize,
+ filename: &str,
+ ) -> Self {
+ Self {
+ metrics: metrics.clone(),
+ partition,
+ filename: filename.to_string(),
+ count: None,
+ }
+ }
+
+ /// Record one suppression, registering the counter on first use.
+ pub(crate) fn add_one(&mut self) {
+ let count = self.count.get_or_insert_with(|| {
+ MetricBuilder::new(&self.metrics)
+ .with_new_label("filename", self.filename.clone())
+ .with_type(MetricType::Summary)
+ .with_category(MetricCategory::Rows)
+ .counter("row_filter_skipped_fully_matched", self.partition)
+ });
+ count.add(1);
+ }
+}
diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs
index 0b3d29c27a..88241f23e7 100644
--- a/datafusion/datasource-parquet/src/opener/mod.rs
+++ b/datafusion/datasource-parquet/src/opener/mod.rs
@@ -25,13 +25,12 @@ use self::early_stop::EarlyStoppingStream;
use self::encryption::EncryptionContext;
use crate::access_plan::PreparedAccessPlan;
use crate::decoder_projection::DecoderProjection;
-use crate::metrics::ByteProgress;
+use crate::metrics::{ByteProgress, RowFilterSkippedFullyMatchedMetric};
use crate::page_filter::PagePruningAccessPlanFilter;
use crate::push_decoder::{
DecoderBuilderConfig, InitialDecoderState, PushDecoderStreamState, RgPlanEntry,
RowGroupPruner,
};
-use crate::row_filter::RowFilterGenerator;
use crate::row_group_filter::{RowGroupAccessPlanFilter, row_group_in_range};
use crate::{
BloomFilterStatistics, Int96Coercer, ParquetAccessPlan, ParquetFileMetrics,
@@ -1458,22 +1457,41 @@ impl RowGroupsPrunedParquetOpen {
prepared.virtual_state.as_deref(),
)?;
+ // Lazily-registered suppression counter shared by the open-time first-RG
+ // skip below and the stream's per-RG toggle (registered on first use so
+ // scans that never suppress don't carry a zero-valued counter).
+ let mut row_filter_skipped_fully_matched =
+ RowFilterSkippedFullyMatchedMetric::new(
+ &prepared.metrics,
+ prepared.partition_index,
+ &prepared.file_name,
+ );
let InitialDecoderState {
decoder,
rg_plan,
has_row_selection,
+ filter_installed,
+ row_filter_context,
} = {
let pushdown_predicate = prepared
.pushdown_filters
.then_some(prepared.predicate.as_ref())
.flatten();
- let mut row_filter_generator = RowFilterGenerator::new(
- pushdown_predicate,
- &prepared.physical_file_schema,
- file_metadata.as_ref(),
- prepared.reorder_predicates,
- &prepared.file_metrics,
- );
+ // Precompute the prebuilt candidate list once per file. Both the
+ // initial `RowFilter` and any per-RG rebuilds (via
+ // `RowFilterContext::build`) reuse it, so tree walks
+ // (`reassign_expr_columns`) and column resolution only run once —
+ // not once per row group.
+ let precomputed_context = pushdown_predicate.and_then(|predicate| {
+ crate::push_decoder::RowFilterContext::try_new(
+ predicate,
+ &prepared.physical_file_schema,
+ &file_metadata,
+ prepared.reorder_predicates,
+ prepared.file_metrics.clone(),
+ prepared.max_predicate_cache_size,
+ )
+ });
// Build the prepared access plan first — `prepare_access_plan` may
// call `reorder_by_statistics` (for `sort_order_for_reorder`) and
@@ -1503,24 +1521,62 @@ impl RowGroupsPrunedParquetOpen {
// https://github.com/apache/arrow-rs/issues/10624 /
// https://github.com/apache/datafusion/issues/24358.
let has_row_selection = prepared_access_plan.row_selection.is_some();
+ // Build `rg_plan` parallel to the decoder's view: the
+ // `prepared_access_plan` has already had its empty-selection
+ // row groups stripped, so 1:1 correspondence with the readers
+ // arrow-rs will hand back is restored. We zip with the
+ // `fully_matched` flag so the stream can toggle the per-row
+ // `RowFilter` per RG.
let rg_plan: VecDeque<RgPlanEntry> = prepared_access_plan
.row_group_indexes
.iter()
.copied()
- .map(|rg_index| RgPlanEntry {
+ .zip(prepared_access_plan.fully_matched.iter().copied())
+ .map(|(rg_index, fully_matched)| RgPlanEntry {
rg_index,
+ fully_matched,
bytes: row_group_bytes(&rg_metadata[rg_index]),
})
.collect();
+ // Decide the initial row filter state based on the first RG to
+ // read. If that RG is `fully_matched` the per-row predicate is
+ // a no-op for every row, so we install an empty `RowFilter`
+ // (arrow-rs's `has_predicates` check then short-circuits the
+ // per-row eval) and the stream toggles back to the real filter
+ // at the first non-fully-matched RG boundary.
+ //
+ // `RowFilterContext` carries everything `build_row_filter`
+ // needs so the stream can regenerate the filter later — the
+ // installed filter is owned by the decoder and is not
+ // recoverable once replaced.
+ let first_rg_fully_matched = rg_plan.front().is_some_and(|e| e.fully_matched);
+ let row_filter_context = precomputed_context;
+
let mut builder =
decoder_config.build(prepared_access_plan, reader_metadata.clone());
- if let Some(row_filter) = row_filter_generator.next_filter() {
- builder = builder.with_row_filter(row_filter);
- if let Some(max_predicate_cache_size) = prepared.max_predicate_cache_size
- {
- builder =
- builder.with_max_predicate_cache_size(max_predicate_cache_size);
+ let mut filter_installed = false;
+ 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, exactly as the per-RG toggle
+ // does mid-scan, so the metric is consistent whether the
+ // skip happens at open time or at a later boundary. The
+ // real filter is only built (lazily, from the prebuilt
+ // candidates) at the first non-fully-matched boundary.
+ 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());
+ filter_installed = true;
+ if let Some(max_predicate_cache_size) =
+ prepared.max_predicate_cache_size
+ {
+ builder = builder
+ .with_max_predicate_cache_size(max_predicate_cache_size);
+ }
}
}
@@ -1528,6 +1584,8 @@ impl RowGroupsPrunedParquetOpen {
decoder: builder.build()?,
rg_plan,
has_row_selection,
+ filter_installed,
+ row_filter_context,
}
};
@@ -1602,7 +1660,6 @@ impl RowGroupsPrunedParquetOpen {
.file_metrics
.row_groups_pruned_dynamic_filter
.clone();
-
let stream = PushDecoderStreamState {
decoder: Some(decoder),
active_reader: None,
@@ -1615,6 +1672,9 @@ impl RowGroupsPrunedParquetOpen {
baseline_metrics: prepared.baseline_metrics,
row_group_pruner,
row_groups_pruned_dynamic,
+ row_filter_context,
+ filter_installed,
+ row_filter_skipped_fully_matched,
byte_progress,
}
.into_stream();
diff --git a/datafusion/datasource-parquet/src/push_decoder.rs b/datafusion/datasource-parquet/src/push_decoder.rs
index 03b7d99c28..94a0e5049f 100644
--- a/datafusion/datasource-parquet/src/push_decoder.rs
+++ b/datafusion/datasource-parquet/src/push_decoder.rs
@@ -47,7 +47,7 @@ use parquet::DecodeResult;
use parquet::arrow::ProjectionMask;
use parquet::arrow::arrow_reader::metrics::ArrowReaderMetrics;
use parquet::arrow::arrow_reader::{
- ArrowReaderMetadata, ParquetRecordBatchReader, RowSelectionPolicy,
+ ArrowReaderMetadata, ParquetRecordBatchReader, RowFilter, RowSelectionPolicy,
};
use parquet::arrow::async_reader::AsyncFileReader;
use parquet::arrow::push_decoder::{ParquetPushDecoder, ParquetPushDecoderBuilder};
@@ -59,9 +59,13 @@ use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
use datafusion_physical_plan::metrics::{BaselineMetrics, Count, Gauge};
use datafusion_pruning::{PruningPredicate, PruningPredicateBuilder};
+use crate::ParquetFileMetrics;
use crate::access_plan::PreparedAccessPlan;
use crate::decoder_projection::DecoderProjection;
-use crate::metrics::ByteProgress;
+use crate::metrics::{ByteProgress, RowFilterSkippedFullyMatchedMetric};
+use crate::row_filter::{
+ PrebuiltRowFilterCandidate, prebuild_row_filter_candidates, row_filter_from_prebuilt,
+};
use crate::row_group_filter::RowGroupPruningStatistics;
/// Shared options applied to the [`ParquetPushDecoderBuilder`] for a file
@@ -82,7 +86,7 @@ impl DecoderBuilderConfig<'_> {
/// Build a [`ParquetPushDecoderBuilder`] from a prepared access plan.
///
/// The caller is expected to attach the
- /// [`RowFilter`](parquet::arrow::arrow_reader::RowFilter) and predicate
+ /// [`RowFilter`] and predicate
/// cache size on the returned builder.
pub(crate) fn build(
&self,
@@ -110,6 +114,9 @@ impl DecoderBuilderConfig<'_> {
#[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.
///
@@ -130,6 +137,13 @@ pub(crate) struct InitialDecoderState {
/// Whether a row selection is live for this scan. Runtime row-group
/// pruning is disabled when it is (see the opener for why).
pub(crate) has_row_selection: bool,
+ /// Whether the freshly built decoder carries a real (non-empty)
+ /// [`RowFilter`]. `false` when the first row group is fully matched and
+ /// the filter was suppressed at open time.
+ pub(crate) filter_installed: bool,
+ /// Cache that lets the stream rebuild the [`RowFilter`] at later
+ /// row-group boundaries. `None` when the scan has no pushdown predicate.
+ pub(crate) row_filter_context: Option<RowFilterContext>,
}
/// Runtime row-group pruner driven by a dynamic predicate (e.g. the
@@ -298,12 +312,104 @@ pub(crate) struct PushDecoderStreamState {
pub(crate) row_group_pruner: Option<RowGroupPruner>,
/// Count of row groups skipped at runtime by [`Self::row_group_pruner`].
pub(crate) row_groups_pruned_dynamic: Count,
+ /// Cache that lets the per-RG `fully_matched` toggle reinstall the
+ /// parquet [`RowFilter`] when it flips from skip → install. `None` when
+ /// the scan has no pushdown predicate (the toggle is then a no-op).
+ pub(crate) row_filter_context: Option<RowFilterContext>,
+ /// Whether the currently-installed decoder is running with a non-empty
+ /// row filter. Toggled per RG by the `fully_matched` skip path.
+ pub(crate) filter_installed: bool,
+ /// Lazily-registered counter of suppression events for the per-row
+ /// [`RowFilter`] (registered on first use so scans that never suppress
+ /// don't carry a zero-valued counter).
+ pub(crate) row_filter_skipped_fully_matched: RowFilterSkippedFullyMatchedMetric,
/// How much of this file range the scan has finished with. Credited a row
/// group at a time as they are decoded or skipped, and topped up to the
/// full range when the stream is dropped.
pub(crate) byte_progress: ByteProgress,
}
+/// A reusable, `Arc`-shared list of prebuilt row-filter candidates.
+///
+/// Wrapping the `Arc<Vec<_>>` keeps the "prebuilt candidates" concept behind
+/// a named type and makes cloning it into stream state cheap.
+#[derive(Clone)]
+pub(crate) struct PrebuiltRowFilterCandidateList {
+ inner: Arc<Vec<PrebuiltRowFilterCandidate>>,
+}
+
+impl PrebuiltRowFilterCandidateList {
+ fn new(candidates: Vec<PrebuiltRowFilterCandidate>) -> Self {
+ Self {
+ inner: Arc::new(candidates),
+ }
+ }
+
+ fn as_slice(&self) -> &[PrebuiltRowFilterCandidate] {
+ &self.inner
+ }
+}
+
+/// Cache that lets [`PushDecoderStreamState`] rebuild the parquet
+/// [`RowFilter`] mid-scan: it keeps the prebuilt candidate list alongside the
+/// stream so a non-fully-matched row group can be re-wrapped into a fresh
+/// [`RowFilter`] without redoing the tree walks and column resolution the
+/// initial build did.
+pub(crate) struct RowFilterContext {
+ /// Prebuilt candidates: expression already column-reassigned, projection
+ /// mask already resolved. Shared across the file's row groups.
+ pub(crate) prebuilt: PrebuiltRowFilterCandidateList,
+ pub(crate) reorder_predicates: bool,
+ pub(crate) file_metrics: ParquetFileMetrics,
+ pub(crate) max_predicate_cache_size: Option<usize>,
+}
+
+impl RowFilterContext {
+ /// Precompute the candidate list from the raw predicate + file schema +
+ /// metadata. Returns `None` when the predicate has no push-downable
+ /// conjuncts (mirrors the file-open path behaviour).
+ pub(crate) fn try_new(
+ predicate: &Arc<dyn PhysicalExpr>,
+ physical_file_schema: &SchemaRef,
+ file_metadata: &Arc<ParquetMetaData>,
+ reorder_predicates: bool,
+ file_metrics: ParquetFileMetrics,
+ max_predicate_cache_size: Option<usize>,
+ ) -> Option<Self> {
+ match prebuild_row_filter_candidates(
+ predicate,
+ physical_file_schema,
+ file_metadata.as_ref(),
+ ) {
+ Ok(Some(prebuilt)) => Some(Self {
+ prebuilt: PrebuiltRowFilterCandidateList::new(prebuilt),
+ reorder_predicates,
+ file_metrics,
+ max_predicate_cache_size,
+ }),
+ Ok(None) => None,
+ Err(e) => {
+ debug!("Ignoring error prebuilding row filter candidates: {e}");
+ None
+ }
+ }
+ }
+
+ /// Build a fresh [`RowFilter`] for the next non-fully-matched run using
+ /// the cached candidates. Cheap: no tree walks, only counter allocation
+ /// and (optionally) a sort by `required_bytes`.
+ ///
+ /// Infallible by construction: [`Self::try_new`] only produces a context
+ /// when the prebuilt candidate list is non-empty.
+ pub(crate) fn build_row_filter(&self) -> RowFilter {
+ row_filter_from_prebuilt(
+ self.prebuilt.as_slice(),
+ self.reorder_predicates,
+ &self.file_metrics,
+ )
+ }
+}
+
impl PushDecoderStreamState {
/// Drive the state machine to completion as a [`futures::Stream`] of record batches.
///
@@ -372,52 +478,27 @@ impl PushDecoderStreamState {
.as_ref()
.expect("decoder present")
.is_at_row_group_boundary();
- // Only the runtime pruner rebuilds the decoder from `rg_plan`, so
- // only it needs `rg_plan` kept in sync with the decoder frontier.
- // 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 re-reads an
- // already-delivered row group (#24352). Gating on the pruner also
- // avoids the O(remaining row groups) cost of `peek_next_row_group()`
- // on ordinary scans that never rebuild.
+ // Keep `rg_plan.front()` aligned with the row group the decoder will
+ // actually emit next: 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. Both the runtime pruner
+ // and the per-RG `RowFilter` toggle consume `rg_plan`, so sync when
+ // either is active; gating avoids the O(remaining row groups) cost of
+ // `peek_next_row_group()` on ordinary scans that never rebuild.
if at_boundary
- && self.row_group_pruner.is_some()
+ && (self.row_group_pruner.is_some() || self.row_filter_context.is_some())
&& let Err(e) = self.sync_rg_plan_to_decoder_frontier()
{
return Some((Err(e), self));
}
if at_boundary && !self.rg_plan.is_empty() {
- let mut pruned_count = 0usize;
- if let Some(pruner) = self.row_group_pruner.as_mut() {
- let mut kept = VecDeque::with_capacity(self.rg_plan.len());
- while let Some(entry) = self.rg_plan.pop_front() {
- if pruner.should_prune(&[entry.rg_index]) {
- pruned_count += 1;
- self.row_groups_pruned_dynamic.add(1);
- self.byte_progress.credit(entry.bytes);
- } else {
- kept.push_back(entry);
- }
- }
- self.rg_plan = kept;
- }
- if pruned_count > 0 {
- if self.rg_plan.is_empty() {
- return None;
- }
- let decoder = self.decoder.take().expect("decoder present");
- let new_indices: Vec<usize> =
- self.rg_plan.iter().map(|e| e.rg_index).collect();
- let rebuilt = match decoder.into_builder() {
- Ok(b) => b.with_row_groups(new_indices).build(),
- Err(e) => Err(e),
- };
- match rebuilt {
- Ok(d) => self.decoder = Some(d),
- Err(e) => {
- return Some((Err(DataFusionError::from(e)), self));
- }
- }
+ let pruned_count = self.prune_boundary_row_groups();
+ match self.rebuild_decoder_at_boundary(pruned_count) {
+ Ok(true) => return None,
+ Ok(false) => {}
+ Err(e) => return Some((Err(e), self)),
}
}
@@ -469,9 +550,10 @@ impl PushDecoderStreamState {
/// Keep `rg_plan.front()` aligned with the row group the decoder will emit
/// next. `try_next_reader` silently finishes row groups whose post-predicate
- /// selection is empty (no reader handed back), which would otherwise leave
- /// `rg_plan` trailing the decoder by one — a later prune/rebuild would then
- /// re-include an already-delivered row group (#24352).
+ /// selection is empty (no reader handed back) — e.g. page-index pruning
+ /// removed every page — which would otherwise leave `rg_plan` trailing the
+ /// decoder by one: a later prune/rebuild would then re-include an
+ /// already-delivered row group (#24352) or toggle the filter for the wrong RG.
fn sync_rg_plan_to_decoder_frontier(&mut self) -> Result<()> {
match self
.decoder
@@ -524,6 +606,84 @@ impl PushDecoderStreamState {
)
}
+ /// Drop every `rg_plan` entry the dynamic pruner proves cannot contribute,
+ /// returning how many were pruned. The single decoder rebuild that acts on
+ /// the survivors is left to the caller (at most one rebuild per boundary).
+ fn prune_boundary_row_groups(&mut self) -> usize {
+ let Some(pruner) = self.row_group_pruner.as_mut() else {
+ return 0;
+ };
+ let mut pruned_count = 0usize;
+ let mut kept = VecDeque::with_capacity(self.rg_plan.len());
+ while let Some(entry) = self.rg_plan.pop_front() {
+ if pruner.should_prune(&[entry.rg_index]) {
+ pruned_count += 1;
+ self.row_groups_pruned_dynamic.add(1);
+ // The scan is done with this row group's bytes.
+ self.byte_progress.credit(entry.bytes);
+ } else {
+ kept.push_back(entry);
+ }
+ }
+ self.rg_plan = kept;
+ pruned_count
+ }
+
+ /// At a row-group boundary, rebuild the decoder so it reads only the
+ /// surviving `rg_plan` and toggle the per-row `RowFilter` for the upcoming
+ /// RG. Rebuilds only when something changed (`pruned_count > 0` or the
+ /// filter status flips), doing at most one `into_builder` rebuild per
+ /// boundary. Returns `Ok(true)` when the plan is now empty (the stream
+ /// should finish).
+ fn rebuild_decoder_at_boundary(
+ &mut self,
+ pruned_count: usize,
+ ) -> Result<bool, DataFusionError> {
+ // `desired_filter` is `Some(true)` when the next RG needs a real
+ // filter, `Some(false)` when it is fully-matched (filter is a no-op, so
+ // we suppress it), and `None` when there is no pushdown predicate at
+ // all (toggling is meaningless).
+ let desired_filter: Option<bool> = self
+ .row_filter_context
+ .as_ref()
+ .and_then(|_| self.rg_plan.front().map(|e| !e.fully_matched));
+ let filter_needs_toggle =
+ desired_filter.is_some_and(|want| want != self.filter_installed);
+
+ if pruned_count == 0 && !filter_needs_toggle {
+ return Ok(false);
+ }
+ if self.rg_plan.is_empty() {
+ return Ok(true);
+ }
+
+ let decoder = self.decoder.take().expect("decoder present");
+ let new_indices: Vec<usize> = self.rg_plan.iter().map(|e| e.rg_index).collect();
+ let mut builder = decoder.into_builder().map_err(DataFusionError::from)?;
+ builder = builder.with_row_groups(new_indices);
+ 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()
+ .expect("filter_needs_toggle ⇒ context set");
+ builder = builder.with_row_filter(ctx.build_row_filter());
+ if let Some(cap) = ctx.max_predicate_cache_size {
+ builder = builder.with_max_predicate_cache_size(cap);
+ }
+ self.filter_installed = true;
+ } 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();
+ }
+ }
+ self.decoder = Some(builder.build().map_err(DataFusionError::from)?);
+ Ok(false)
+ }
+
/// Copies metrics from ArrowReaderMetrics (the metrics collected by the
/// arrow-rs parquet reader) to the parquet file metrics for DataFusion
fn copy_arrow_reader_metrics(&self) {
@@ -724,6 +884,7 @@ mod tests {
.into_iter()
.map(|rg_index| RgPlanEntry {
rg_index,
+ fully_matched: false,
bytes: 100 * (rg_index as u64 + 1),
})
.collect()
diff --git a/datafusion/datasource-parquet/src/row_filter.rs b/datafusion/datasource-parquet/src/row_filter.rs
index c1a47c896c..d5e085b637 100644
--- a/datafusion/datasource-parquet/src/row_filter.rs
+++ b/datafusion/datasource-parquet/src/row_filter.rs
@@ -120,7 +120,12 @@ pub(crate) struct DatafusionArrowPredicate {
}
impl DatafusionArrowPredicate {
- /// Create a new `DatafusionArrowPredicate` from a `FilterCandidate`
+ /// Create a new `DatafusionArrowPredicate` from a `FilterCandidate`.
+ ///
+ /// Production code goes through [`prebuild_row_filter_candidates`] +
+ /// [`row_filter_from_prebuilt`]; this constructor remains as a test
+ /// convenience for exercising a single candidate.
+ #[cfg(test)]
pub fn try_new(
candidate: FilterCandidate,
rows_pruned: metrics::Count,
@@ -401,16 +406,65 @@ pub fn build_row_filter(
reorder_predicates: bool,
file_metrics: &ParquetFileMetrics,
) -> Result<Option<RowFilter>> {
- let rows_pruned = &file_metrics.pushdown_rows_pruned;
- let rows_matched = &file_metrics.pushdown_rows_matched;
- let time = &file_metrics.row_pushdown_eval_time;
+ // Implemented on top of the prebuild split so there is a single place
+ // that splits conjuncts, orders candidates, and wires metrics — callers
+ // that build once per file go through the same code as the per-row-group
+ // rebuild path in `RowFilterContext`.
+ let Some(prebuilt) = prebuild_row_filter_candidates(expr, file_schema, metadata)?
+ else {
+ return Ok(None);
+ };
+ Ok(Some(row_filter_from_prebuilt(
+ &prebuilt,
+ reorder_predicates,
+ file_metrics,
+ )))
+}
+
+/// A precomputed [`FilterCandidate`] with its expression column-reassigned to
+/// the projected schema, ready to be wrapped into a [`DatafusionArrowPredicate`]
+/// on demand.
+///
+/// Extracting this from [`build_row_filter`] lets callers pay the tree-walk +
+/// column-resolution + `reassign_expr_columns` cost **once per file** instead
+/// of once per row group, which is the hot path for
+/// [`RowFilterContext::build_row_filter`](crate::push_decoder::RowFilterContext) rebuilds
+/// on `fully_matched → not-fully-matched` boundaries.
+#[derive(Clone, Debug)]
+pub(crate) struct PrebuiltRowFilterCandidate {
+ /// The predicate expression with all `Column` indices rewritten to point
+ /// into the projected file schema.
+ physical_expr: Arc<dyn PhysicalExpr>,
+ /// Projection mask over the parquet leaf columns needed to evaluate this
+ /// predicate.
+ projection_mask: ProjectionMask,
+ /// Precomputed sum-of-compressed-bytes for the referenced columns across
+ /// all row groups in the file. Used to sort predicates when
+ /// `reorder_predicates` is enabled. Stable across row groups within a
+ /// file, so we cache it once.
+ required_bytes: usize,
+}
+/// 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.
+///
+/// Returns `Ok(None)` when the predicate has no push-downable conjuncts, in
+/// which case callers should skip installing a `RowFilter` entirely.
+pub(crate) fn prebuild_row_filter_candidates(
+ expr: &Arc<dyn PhysicalExpr>,
+ file_schema: &SchemaRef,
+ metadata: &ParquetMetaData,
+) -> Result<Option<Vec<PrebuiltRowFilterCandidate>>> {
// Split into conjuncts:
// `a = 1 AND b = 2 AND c = 3` -> [`a = 1`, `b = 2`, `c = 3`]
let predicates = split_conjunction(expr);
-
- // Determine which conjuncts can be evaluated as ArrowPredicates, if any
- let mut candidates: Vec<FilterCandidate> = predicates
+ let candidates: Vec<FilterCandidate> = predicates
.into_iter()
.map(|expr| {
FilterCandidateBuilder::new(Arc::clone(expr), Arc::clone(file_schema))
@@ -421,106 +475,70 @@ pub fn build_row_filter(
.flatten()
.collect();
- // no candidates
if candidates.is_empty() {
return Ok(None);
}
+ let prebuilt: Vec<PrebuiltRowFilterCandidate> = candidates
+ .into_iter()
+ .map(|candidate| {
+ let physical_expr = reassign_expr_columns(
+ Arc::clone(&candidate.expr),
+ &candidate.read_plan.projected_schema,
+ )?;
+ Ok(PrebuiltRowFilterCandidate {
+ physical_expr,
+ projection_mask: candidate.read_plan.projection_mask.clone(),
+ required_bytes: candidate.required_bytes,
+ })
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ Ok(Some(prebuilt))
+}
+
+/// 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.
+pub(crate) fn row_filter_from_prebuilt(
+ prebuilt: &[PrebuiltRowFilterCandidate],
+ reorder_predicates: bool,
+ file_metrics: &ParquetFileMetrics,
+) -> RowFilter {
+ let rows_pruned = &file_metrics.pushdown_rows_pruned;
+ let rows_matched = &file_metrics.pushdown_rows_matched;
+ let time = &file_metrics.row_pushdown_eval_time;
+
+ // Clone (cheap: Arc bumps + ProjectionMask clone) into a working list we
+ // can sort without disturbing the shared cache.
+ let mut ordered: Vec<&PrebuiltRowFilterCandidate> = prebuilt.iter().collect();
if reorder_predicates {
- candidates.sort_unstable_by_key(|c| c.required_bytes);
+ ordered.sort_unstable_by_key(|c| c.required_bytes);
}
- // To avoid double-counting metrics when multiple predicates are used:
- // - All predicates should count rows_pruned (cumulative pruned rows)
- // - Only the last predicate should count rows_matched (final result)
- // This ensures: rows_matched + rows_pruned = total rows processed
- let total_candidates = candidates.len();
-
- candidates
+ let total = ordered.len();
+ let filters: Vec<Box<dyn ArrowPredicate>> = ordered
.into_iter()
.enumerate()
.map(|(idx, candidate)| {
- let is_last = idx == total_candidates - 1;
-
- // All predicates share the pruned counter (cumulative)
+ let is_last = idx == total - 1;
let predicate_rows_pruned = rows_pruned.clone();
-
- // Only the last predicate tracks matched rows (final result)
let predicate_rows_matched = if is_last {
rows_matched.clone()
} else {
metrics::Count::new()
};
-
- DatafusionArrowPredicate::try_new(
- candidate,
- predicate_rows_pruned,
- predicate_rows_matched,
- time.clone(),
- )
- .map(|pred| Box::new(pred) as _)
+ Box::new(DatafusionArrowPredicate {
+ physical_expr: Arc::clone(&candidate.physical_expr),
+ projection_mask: candidate.projection_mask.clone(),
+ rows_pruned: predicate_rows_pruned,
+ rows_matched: predicate_rows_matched,
+ time: time.clone(),
+ }) as Box<dyn ArrowPredicate>
})
- .collect::<Result<Vec<_>, _>>()
- .map(|filters| Some(RowFilter::new(filters)))
-}
-
-/// Builds row filters for a parquet decoder.
-///
-/// A [`RowFilter`] is owned by a decoder. The first filter is built eagerly
-/// during construction so the caller can attach it to the decoder via
-/// [`next_filter`](Self::next_filter) without a redundant build call.
-pub(crate) struct RowFilterGenerator<'a> {
- predicate: Option<&'a Arc<dyn PhysicalExpr>>,
- physical_file_schema: &'a SchemaRef,
- file_metadata: &'a ParquetMetaData,
- reorder_predicates: bool,
- file_metrics: &'a ParquetFileMetrics,
- first_row_filter: Option<RowFilter>,
-}
-
-impl<'a> RowFilterGenerator<'a> {
- pub(crate) fn new(
- predicate: Option<&'a Arc<dyn PhysicalExpr>>,
- physical_file_schema: &'a SchemaRef,
- file_metadata: &'a ParquetMetaData,
- reorder_predicates: bool,
- file_metrics: &'a ParquetFileMetrics,
- ) -> Self {
- let mut generator = Self {
- predicate,
- physical_file_schema,
- file_metadata,
- reorder_predicates,
- file_metrics,
- first_row_filter: None,
- };
- generator.first_row_filter = generator.build();
- generator
- }
-
- pub(crate) fn next_filter(&mut self) -> Option<RowFilter> {
- self.first_row_filter.take().or_else(|| self.build())
- }
-
- fn build(&self) -> Option<RowFilter> {
- let predicate = self.predicate?;
- match build_row_filter(
- predicate,
- self.physical_file_schema,
- self.file_metadata,
- self.reorder_predicates,
- self.file_metrics,
- ) {
- Ok(Some(filter)) => Some(filter),
- Ok(None) => None,
- Err(e) => {
- log::debug!(
- "Ignoring error building row filter for '{predicate:?}': {e}"
- );
- None
- }
- }
- }
+ .collect();
+ RowFilter::new(filters)
}
#[cfg(test)]
diff --git a/datafusion/datasource-parquet/src/sort.rs b/datafusion/datasource-parquet/src/sort.rs
index 0f73723a1d..b5bcd7eaba 100644
--- a/datafusion/datasource-parquet/src/sort.rs
+++ b/datafusion/datasource-parquet/src/sort.rs
@@ -412,8 +412,6 @@ mod tests {
let metadata = create_test_metadata(vec![100, 100, 100]);
let mut access_plan = ParquetAccessPlan::new_all(3);
-
- // Skip all rows in all row groups
for i in 0..3 {
access_plan
.scan_selection(i, RowSelection::from(vec![RowSelector::skip(100)]));
@@ -423,6 +421,11 @@ mod tests {
let prepared_plan = access_plan
.prepare(rg_metadata)
.expect("Failed to create PreparedAccessPlan");
+ assert!(
+ prepared_plan.row_group_indexes.is_empty(),
+ "all-empty selections must be stripped to an empty plan",
+ );
+ assert!(prepared_plan.row_selection.is_none());
// All row groups are empty after pruning, so they are stripped and the
// prepared plan is empty (rather than carrying a selection that skips
diff --git a/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt b/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt
index fddc14ca73..58b6ca551d 100644
--- a/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt
+++ b/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt
@@ -100,6 +100,48 @@ Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[v@0 DESC], preserve_partitioning=[false], filter=[v@0 IS NULL OR v@0 > 12], metrics=[output_rows=3, elapsed_compute=<slt:ignore>, output_bytes=<slt:ignore>]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_row_group_pruning/data.parquet]]}, projection=[v], file_type=parquet, predicate=DynamicFilter [ v@0 IS NULL OR v@0 > 12 ], sort_order_for_reorder=[v@0 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible, pruning_predicate=v_null_count@0 > 0 OR v_null_count@0 != row_count@2 AND v_max@1 > 12, required_guarantees=[], metrics=[output_rows=3, elapsed_compute=<slt:ignore>, out [...]
+# `EXPLAIN ANALYZE` with a *static* predicate must surface the
+# `row_filter_skipped_fully_matched` metric with a non-zero value — this
+# is the only way to see the fully-matched RowFilter-skip optimization
+# fire from SQL (otherwise it is only exercised by the Rust integration
+# test `fully_matched_rgs_skip_row_filter`).
+#
+# With `pushdown_filters=true` the predicate `v >= 4` is pushed into the
+# parquet decoder as a per-row `RowFilter`. Against the five row groups:
+# RG 0 (0,1,2) → max=2 < 4, dropped by row-group statistics
+# RG 1 (3,4,5) → straddles the threshold, keeps the row filter
+# RG 2 (6,7,8) → every row matches by statistics → fully matched
+# RG 3 (9,10,11) → fully matched
+# RG 4 (12,13,14)→ fully matched
+# Entering the RG 1 → RG 2 boundary the decoder rebuilds with an empty
+# row filter and leaves it off across the consecutive fully-matched run
+# RGs 2..=4. That single toggle is one *suppression event*, so
+# `row_filter_skipped_fully_matched=1` — the metric counts toggles, not
+# row groups (see `metrics.rs`), which is why a run of three
+# fully-matched RGs still reads `=1`. Time- and size-keyed fields are
+# masked with `<slt:ignore>`.
+query I
+SELECT v FROM t WHERE v >= 4 ORDER BY v;
+----
+4
+5
+6
+7
+8
+9
+10
+11
+12
+13
+14
+
+query TT
+explain analyze select v from t where v >= 4 order by v;
+----
+Plan with Metrics
+01)SortExec: expr=[v@0 ASC NULLS LAST], preserve_partitioning=[false], metrics=[output_rows=11, elapsed_compute=<slt:ignore>, output_bytes=<slt:ignore>]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_row_group_pruning/data.parquet]]}, projection=[v], file_type=parquet, predicate=v@0 >= 4, sort_order_for_reorder=[v@0 ASC NULLS LAST], pruning_predicate=v_null_count@1 != row_count@2 AND v_max@0 >= 4, required_guarantees=[], metrics=[output_rows=11, elapsed_compute=<slt:ignore>, output_bytes=<slt:ignore>, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_stat [...]
+
statement ok
drop table t;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org