diff --git a/datafusion/core/tests/parquet/dynamic_row_group_pruning.rs b/datafusion/core/tests/parquet/dynamic_row_group_pruning.rs
index 5ee42b30674bf..678aad9f5b1fd 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 6d96c68e18cb0..b837159cf0c9b 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![
             sort_expr("a", 0, false),
             PhysicalSortExpr {
diff --git a/datafusion/datasource-parquet/src/metrics.rs b/datafusion/datasource-parquet/src/metrics.rs
index ad605ccaf6b9a..c9a908a989924 100644
--- a/datafusion/datasource-parquet/src/metrics.rs
+++ b/datafusion/datasource-parquet/src/metrics.rs
@@ -396,3 +396,54 @@ impl ParquetFileMetrics {
         count.add(n);
     }
 }
+
+/// Lazily-registered counter for `row_filter_skipped_fully_matched`: the
+/// number of times the per-row
+/// [`RowFilter`](parquet::arrow::arrow_reader::RowFilter) 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 5e004cf017484..796cabf5775a7 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,
@@ -1459,22 +1458,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
@@ -1504,24 +1522,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);
+                    }
                 }
             }
 
@@ -1529,6 +1585,8 @@ impl RowGroupsPrunedParquetOpen {
                 decoder: builder.build()?,
                 rg_plan,
                 has_row_selection,
+                filter_installed,
+                row_filter_context,
             }
         };
 
@@ -1603,7 +1661,6 @@ impl RowGroupsPrunedParquetOpen {
             .file_metrics
             .row_groups_pruned_dynamic_filter
             .clone();
-
         let stream = PushDecoderStreamState {
             decoder: Some(decoder),
             active_reader: None,
@@ -1616,6 +1673,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 8238f13b006e9..40944dce10786 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
@@ -293,12 +307,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.
     ///
@@ -367,52 +473,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)),
                 }
             }
 
@@ -464,9 +545,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
@@ -519,6 +601,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) {
@@ -719,6 +879,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 c1a47c896c170..d5e085b63737f 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 0f73723a1de91..b5bcd7eaba017 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 fddc14ca73d84..58b6ca551d3fa 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>, output_bytes=<slt:ignore>, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=5 total → 5 matched, row_groups_pruned_bloom_filter=5 total → 5 matched, page_index_pages_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=<slt:ignore>, bytes_scanned=<slt:ignore>, row_groups_pruned_dynamic_filter=4, metadata_load_time=<slt:ignore>, scan_efficiency_ratio=<slt:ignore>]
 
+# `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_statistics=5 total → 4 matched -> 3 fully matched, row_groups_pruned_bloom_filter=4 total → 4 matched, page_index_pages_pruned=1 total → 1 matched, page_index_pages_skipped_by_fully_matched=3, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=<slt:ignore>, bytes_scanned=<slt:ignore>, row_filter_skipped_fully_matched=1, row_groups_pruned_dynamic_filter=0, metadata_load_time=<slt:ignore>, scan_efficiency_ratio=<slt:ignore>]
+
 statement ok
 drop table t;