From 65ce111f52fec70eefbc8757d365a82371a8fc89 Mon Sep 17 00:00:00 2001
From: Huaijin <haohuaijin@gmail.com>
Date: Sun, 19 Apr 2026 15:12:46 +0800
Subject: [PATCH 1/6] perf: add limit to last predicate

---
 parquet/benches/arrow_reader_row_filter.rs    | 117 ++++++-
 parquet/src/arrow/array_reader/mod.rs         |   2 +-
 parquet/src/arrow/arrow_reader/read_plan.rs   | 204 +++++++++++-
 parquet/src/arrow/push_decoder/mod.rs         | 292 ++++++++++++++++++
 .../push_decoder/reader_builder/filter.rs     |   7 +
 .../arrow/push_decoder/reader_builder/mod.rs  |  48 ++-
 6 files changed, 658 insertions(+), 12 deletions(-)

diff --git a/parquet/benches/arrow_reader_row_filter.rs b/parquet/benches/arrow_reader_row_filter.rs
index 331f5617ca8a..2b5a09eebcb3 100644
--- a/parquet/benches/arrow_reader_row_filter.rs
+++ b/parquet/benches/arrow_reader_row_filter.rs
@@ -180,10 +180,16 @@ fn create_record_batch(size: usize) -> RecordBatch {
     RecordBatch::try_new(schema, arrays).unwrap()
 }
 
+/// Total number of rows.
+const TOTAL_ROWS: usize = 500_000;
+
+/// Maximum rows per row group.
+const ROW_GROUP_SIZE: usize = 100_000;
+
 /// Writes the RecordBatch to an in memory buffer, returning the buffer
 fn write_parquet_file() -> Vec<u8> {
-    let batch = create_record_batch(100_000);
-    println!("Batch created with {} rows", 100_000);
+    let batch = create_record_batch(TOTAL_ROWS);
+    println!("Batch created with {TOTAL_ROWS} rows, row group size = {ROW_GROUP_SIZE}");
     println!(
         "First 100 rows:\n{}",
         pretty_format_batches(&[batch.clone().slice(0, 100)]).unwrap()
@@ -191,6 +197,7 @@ fn write_parquet_file() -> Vec<u8> {
     let schema = batch.schema();
     let props = WriterProperties::builder()
         .set_compression(Compression::SNAPPY)
+        .set_max_row_group_row_count(Some(ROW_GROUP_SIZE))
         .build();
     let mut buffer = vec![];
     {
@@ -522,6 +529,28 @@ async fn benchmark_async_reader(
     }
 }
 
+/// Like [`benchmark_async_reader`] but also threads `with_limit(limit)` into
+/// the stream builder. Used by the `LIMIT` benchmark below.
+async fn benchmark_async_reader_with_limit(
+    reader: InMemoryReader,
+    projection_mask: ProjectionMask,
+    row_filter: RowFilter,
+    limit: usize,
+) {
+    let mut stream = ParquetRecordBatchStreamBuilder::new(reader)
+        .await
+        .unwrap()
+        .with_batch_size(8192)
+        .with_projection(projection_mask)
+        .with_row_filter(row_filter)
+        .with_limit(limit)
+        .build()
+        .unwrap();
+    while let Some(b) = stream.next().await {
+        b.unwrap(); // consume the batches, no buffering
+    }
+}
+
 /// Use sync API
 fn benchmark_sync_reader(
     reader: InMemoryReader,
@@ -586,5 +615,87 @@ impl AsyncFileReader for InMemoryReader {
     }
 }
 
-criterion_group!(benches, benchmark_filters_and_projections,);
+/// Benchmark filters with `LIMIT` short-circuit (`with_limit(N)`)
+///
+/// `PointLookup` is excluded because the filter has only 1 match in the
+/// whole file; `LIMIT 10` is not binding.
+fn benchmark_filters_with_limit(c: &mut Criterion) {
+    const LIMIT: usize = 10;
+
+    let parquet_file = Bytes::from(write_parquet_file());
+    let filter_types = vec![
+        FilterType::SelectiveUnclustered,
+        FilterType::ModeratelySelectiveClustered,
+        FilterType::ModeratelySelectiveUnclustered,
+        FilterType::UnselectiveUnclustered,
+        FilterType::UnselectiveClustered,
+        FilterType::Utf8ViewNonEmpty,
+        FilterType::Composite,
+    ];
+    let projection_cases = vec![
+        ProjectionCase::AllColumns,
+        ProjectionCase::ExcludeFilterColumn,
+    ];
+    let all_indices = vec![0, 1, 2, 3];
+
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+
+    let mut group = c.benchmark_group("arrow_reader_row_filter_limit");
+
+    for filter_type in filter_types {
+        for proj_case in &projection_cases {
+            let filter_col = filter_type.filter_projection().to_vec();
+            let output_projection: Vec<usize> = match proj_case {
+                ProjectionCase::AllColumns => all_indices.clone(),
+                ProjectionCase::ExcludeFilterColumn => all_indices
+                    .iter()
+                    .copied()
+                    .filter(|i| !filter_col.contains(i))
+                    .collect(),
+            };
+
+            let reader = InMemoryReader::try_new(&parquet_file).unwrap();
+            let metadata = Arc::clone(reader.metadata());
+            let schema_descr = metadata.file_metadata().schema_descr();
+            let projection_mask = ProjectionMask::roots(schema_descr, output_projection);
+            let pred_mask = ProjectionMask::roots(schema_descr, filter_col);
+
+            let benchmark_name = format!("{filter_type}/{proj_case}/limit{LIMIT}");
+
+            // async variant
+            let bench_id = BenchmarkId::new(benchmark_name.clone(), "async");
+            let rt_handle = rt.handle().clone();
+            let pred_mask_async = pred_mask.clone();
+            let projection_mask_async = projection_mask.clone();
+            let reader_async = reader.clone();
+            group.bench_function(bench_id, |b| {
+                b.iter(|| {
+                    let reader = reader_async.clone();
+                    let pred_mask = pred_mask_async.clone();
+                    let projection_mask = projection_mask_async.clone();
+                    // RowFilter and ArrowPredicateFn are not Clone — fresh each iter.
+                    let predicate = ArrowPredicateFn::new(pred_mask, move |batch: RecordBatch| {
+                        Ok(filter_type.filter_batch(&batch).unwrap())
+                    });
+                    let row_filter = RowFilter::new(vec![Box::new(predicate)]);
+                    rt_handle.block_on(benchmark_async_reader_with_limit(
+                        reader,
+                        projection_mask,
+                        row_filter,
+                        LIMIT,
+                    ));
+                });
+            });
+        }
+    }
+}
+
+criterion_group!(
+    benches,
+    benchmark_filters_and_projections,
+    benchmark_filters_with_limit,
+);
 criterion_main!(benches);
diff --git a/parquet/src/arrow/array_reader/mod.rs b/parquet/src/arrow/array_reader/mod.rs
index 726eae1f51c5..989c93480993 100644
--- a/parquet/src/arrow/array_reader/mod.rs
+++ b/parquet/src/arrow/array_reader/mod.rs
@@ -49,7 +49,7 @@ mod row_number;
 mod struct_array;
 
 #[cfg(test)]
-mod test_util;
+pub(crate) mod test_util;
 
 // Note that this crate is public under the `experimental` feature flag.
 use crate::file::metadata::RowGroupMetaData;
diff --git a/parquet/src/arrow/arrow_reader/read_plan.rs b/parquet/src/arrow/arrow_reader/read_plan.rs
index 99ffe0febc95..670615577284 100644
--- a/parquet/src/arrow/arrow_reader/read_plan.rs
+++ b/parquet/src/arrow/arrow_reader/read_plan.rs
@@ -25,7 +25,8 @@ use crate::arrow::arrow_reader::{
     ArrowPredicate, ParquetRecordBatchReader, RowSelection, RowSelectionCursor, RowSelector,
 };
 use crate::errors::{ParquetError, Result};
-use arrow_array::Array;
+use arrow_array::{Array, BooleanArray};
+use arrow_buffer::{BooleanBuffer, BooleanBufferBuilder};
 use arrow_select::filter::prep_null_mask_filter;
 use std::collections::VecDeque;
 
@@ -144,12 +145,64 @@ impl ReadPlanBuilder {
     /// or if the [`ParquetRecordBatchReader`] specified an explicit
     /// [`RowSelection`] in addition to one or more predicates.
     pub fn with_predicate(
+        self,
+        array_reader: Box<dyn ArrayReader>,
+        predicate: &mut dyn ArrowPredicate,
+    ) -> Result<Self> {
+        // No limit ⇒ loop always exhausts, so `total_rows` is not consulted.
+        self.with_predicate_limited(array_reader, predicate, None, 0)
+    }
+
+    /// Like [`Self::with_predicate`] but additionally caps the number of
+    /// matches retained from `predicate` at `limit`, stopping the scan of
+    /// `array_reader` once that many matches have been accumulated.
+    ///
+    /// This is a performance optimization for TopK / `LIMIT` queries: once
+    /// the cumulative `true_count` across evaluated filter batches reaches
+    /// `limit`, the current batch's filter is truncated at the `limit`-th
+    /// match and iteration of `array_reader` stops. Remaining batches of
+    /// the predicate's column chunks are never decoded and the predicate is
+    /// never re-invoked — avoiding both decode and predicate work.
+    ///
+    /// `limit` counts predicate matches, not final output rows. Callers that
+    /// also apply an offset must pass `offset + limit` here so enough matches
+    /// survive the later offset step.
+    ///
+    /// `total_rows` is the number of rows the predicate's `array_reader`
+    /// would yield if iterated to completion (i.e. the row-group size when
+    /// no prior selection exists). It is only consulted on early break, to
+    /// pad the trailing un-evaluated rows as "not selected" so the resulting
+    /// [`RowSelection`] covers the full row group. When `limit` is `None`
+    /// this parameter is ignored.
+    ///
+    /// This must only be called when `predicate` is the *last* predicate in a
+    /// filter chain: match counts of intermediate predicates do not correspond
+    /// 1:1 to output rows, so truncating them early would drop rows that
+    /// should have been passed to subsequent predicates.
+    ///
+    /// If `limit` is `None` this behaves exactly like [`Self::with_predicate`].
+    pub fn with_predicate_limited(
         mut self,
         array_reader: Box<dyn ArrayReader>,
         predicate: &mut dyn ArrowPredicate,
+        limit: Option<usize>,
+        total_rows: usize,
     ) -> Result<Self> {
+        // Target length for the concatenated filter output:
+        // - Prior selection ⇒ the reader yields that many rows; `and_then`
+        //   below requires the filter output to match.
+        // - No prior selection ⇒ the reader yields `total_rows`. We only
+        //   need to pad when `limit` may short-circuit the loop; otherwise
+        //   iteration naturally exhausts.
+        let expected_rows = match self.selection.as_ref() {
+            Some(s) => Some(s.row_count()),
+            None => limit.map(|_| total_rows),
+        };
+
         let reader = ParquetRecordBatchReader::new(array_reader, self.clone().build());
         let mut filters = vec![];
+        let mut processed_rows: usize = 0;
+        let mut cumulative_matches: usize = 0;
         for maybe_batch in reader {
             let maybe_batch = maybe_batch?;
             let input_rows = maybe_batch.num_rows();
@@ -161,10 +214,37 @@ impl ReadPlanBuilder {
                     filter.len()
                 ));
             }
-            match filter.null_count() {
-                0 => filters.push(filter),
-                _ => filters.push(prep_null_mask_filter(&filter)),
+            let filter = match filter.null_count() {
+                0 => filter,
+                _ => prep_null_mask_filter(&filter),
             };
+
+            processed_rows += input_rows;
+
+            match limit {
+                Some(limit) if cumulative_matches + filter.true_count() >= limit => {
+                    let needed = limit - cumulative_matches;
+                    let truncated = truncate_filter_after_n_trues(&filter, needed);
+                    filters.push(truncated);
+                    break;
+                }
+                _ => {
+                    cumulative_matches += filter.true_count();
+                    filters.push(filter);
+                }
+            }
+        }
+
+        // Pad the tail so the filters cover `expected_rows` total. This keeps
+        // the invariant that the resulting `RowSelection` spans every row the
+        // reader would have produced — rows past the early break are marked
+        // "not selected". When no limit is set the loop always exhausts and
+        // no padding is needed.
+        if let Some(expected) = expected_rows
+            && processed_rows < expected
+        {
+            let pad_len = expected - processed_rows;
+            filters.push(BooleanArray::new(BooleanBuffer::new_unset(pad_len), None));
         }
 
         // If the predicate selected all rows and there is no prior selection,
@@ -305,6 +385,35 @@ impl LimitedReadPlanBuilder {
     }
 }
 
+/// Produce a new `BooleanArray` of the same length as `filter` in which only
+/// the first `n` `true` positions from `filter` remain `true`; any `true`
+/// positions beyond the first `n` are replaced with `false`.
+///
+/// `filter` must not contain nulls (callers apply [`prep_null_mask_filter`]
+/// first). If `filter` has at most `n` `true` values, a clone is returned.
+fn truncate_filter_after_n_trues(filter: &BooleanArray, n: usize) -> BooleanArray {
+    if filter.true_count() <= n {
+        return filter.clone();
+    }
+    let len = filter.len();
+    if n == 0 {
+        return BooleanArray::new(BooleanBuffer::new_unset(len), None);
+    }
+    // `set_indices` scans 64 bits at a time via `trailing_zeros`, so locating
+    // the `n`-th set bit is cheaper than visiting every bit. Everything up to
+    // and including that position is copied verbatim; the rest is zeroed.
+    let values = filter.values();
+    let last_kept = values
+        .set_indices()
+        .nth(n - 1)
+        .expect("n - 1 < true_count, checked above");
+
+    let mut builder = BooleanBufferBuilder::new(len);
+    builder.append_buffer(&values.slice(0, last_kept + 1));
+    builder.append_n(len - last_kept - 1, false);
+    BooleanArray::new(builder.finish(), None)
+}
+
 /// A plan reading specific rows from a Parquet Row Group.
 ///
 /// See [`ReadPlanBuilder`] to create `ReadPlan`s
@@ -367,4 +476,91 @@ mod tests {
             RowSelectionStrategy::Selectors
         );
     }
+
+    #[test]
+    fn truncate_filter_after_n_trues_keeps_first_n_matches() {
+        let f = BooleanArray::from(vec![true, false, true, true, false, true, true]);
+        // true positions: 0, 2, 3, 5, 6
+        let t = truncate_filter_after_n_trues(&f, 3);
+        assert_eq!(t.len(), f.len());
+        assert_eq!(t.true_count(), 3);
+        let out: Vec<bool> = (0..t.len()).map(|i| t.value(i)).collect();
+        assert_eq!(
+            out,
+            vec![true, false, true, true, false, false, false],
+            "first three trues should survive, the rest become false"
+        );
+    }
+
+    #[test]
+    fn truncate_filter_after_n_trues_passes_through_when_already_small_enough() {
+        let f = BooleanArray::from(vec![true, false, true, false]);
+        let t = truncate_filter_after_n_trues(&f, 5);
+        assert_eq!(t.len(), f.len());
+        assert_eq!(t.true_count(), 2);
+    }
+
+    #[test]
+    fn truncate_filter_after_n_trues_zero_returns_all_false() {
+        let f = BooleanArray::from(vec![true, true, true]);
+        let t = truncate_filter_after_n_trues(&f, 0);
+        assert_eq!(t.len(), 3);
+        assert_eq!(t.true_count(), 0);
+    }
+
+    #[test]
+    fn with_predicate_limited_pads_tail_when_no_prior_selection() {
+        use crate::arrow::ProjectionMask;
+        use crate::arrow::array_reader::StructArrayReader;
+        use crate::arrow::array_reader::test_util::InMemoryArrayReader;
+        use crate::arrow::arrow_reader::ArrowPredicateFn;
+        use arrow_array::Int32Array;
+        use arrow_schema::{DataType as ArrowType, Field, Fields};
+        use std::sync::Arc;
+
+        // 100 rows, all match the predicate. Limit stops the loop after 10
+        // matches — but the resulting RowSelection must still describe the
+        // full 100-row row group (90 trailing rows as "not selected"), not
+        // only the 10 rows we happened to evaluate before breaking.
+        const TOTAL_ROWS: usize = 100;
+        const LIMIT: usize = 10;
+
+        let data: Vec<i32> = (0..TOTAL_ROWS as i32).collect();
+        let array = Arc::new(Int32Array::from(data));
+        let leaf = InMemoryArrayReader::new(ArrowType::Int32, array.clone(), None, None);
+        let struct_type = ArrowType::Struct(Fields::from(vec![Field::new(
+            "c0",
+            ArrowType::Int32,
+            false,
+        )]));
+        let struct_reader = StructArrayReader::new(struct_type, vec![Box::new(leaf)], 0, 0, false);
+
+        let mut predicate = ArrowPredicateFn::new(ProjectionMask::all(), |batch| {
+            Ok(BooleanArray::from(vec![true; batch.num_rows()]))
+        });
+
+        let builder = ReadPlanBuilder::new(16)
+            .with_predicate_limited(
+                Box::new(struct_reader),
+                &mut predicate,
+                Some(LIMIT),
+                TOTAL_ROWS,
+            )
+            .unwrap();
+
+        let selection = builder
+            .selection()
+            .expect("limit-driven early break must produce a selection");
+
+        // `row_count` counts selected rows — must equal the limit.
+        assert_eq!(selection.row_count(), LIMIT);
+
+        // Total rows covered (selects + skips) must equal the full row group
+        // so downstream offset/limit math stays in absolute-row space.
+        let total: usize = selection.iter().map(|s| s.row_count).sum();
+        assert_eq!(
+            total, TOTAL_ROWS,
+            "selection must span the full row group, not only the prefix evaluated before the limit"
+        );
+    }
 }
diff --git a/parquet/src/arrow/push_decoder/mod.rs b/parquet/src/arrow/push_decoder/mod.rs
index cdb0715edb55..6e5299180906 100644
--- a/parquet/src/arrow/push_decoder/mod.rs
+++ b/parquet/src/arrow/push_decoder/mod.rs
@@ -1021,6 +1021,298 @@ mod test {
         expect_finished(decoder.try_decode());
     }
 
+    /// When filter pushdown is combined with a `LIMIT`, the predicate must
+    /// not be evaluated for rows beyond the `limit`-th match.
+    ///
+    /// Filter `a > 175` produces 24 matches in row group 0 (rows 176..199).
+    /// With `limit = 10`, only the first 10 matches (rows 176..185) should be
+    /// emitted, AND the predicate counter should observe that evaluation was
+    /// short-circuited.
+    #[test]
+    fn test_decoder_filter_with_limit_short_circuits_within_row_group() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+
+        let builder =
+            ParquetPushDecoderBuilder::try_new_decoder(test_file_parquet_metadata()).unwrap();
+        let schema_descr = builder.metadata().file_metadata().schema_descr_ptr();
+
+        let rows_filtered = Arc::new(AtomicUsize::new(0));
+        let rows_filtered_for_predicate = Arc::clone(&rows_filtered);
+
+        let row_filter_a = ArrowPredicateFn::new(
+            ProjectionMask::columns(&schema_descr, ["a"]),
+            move |batch: RecordBatch| {
+                rows_filtered_for_predicate.fetch_add(batch.num_rows(), Ordering::Relaxed);
+                let scalar_175 = Int64Array::new_scalar(175);
+                let column = batch.column(0).as_primitive::<Int64Type>();
+                gt(column, &scalar_175)
+            },
+        );
+
+        // Use a small batch size so the row group is evaluated across
+        // multiple predicate batches; that is the regime where Layer 2's
+        // short-circuit saves predicate evaluation work. Matching rows are
+        // 176..199 (24 rows); with batch_size = 10 those span batches 17, 18,
+        // and 19 (rows 170..199). A limit of 10 should stop filter evaluation
+        // in the middle of batch 18.
+        let mut decoder = builder
+            .with_projection(ProjectionMask::columns(&schema_descr, ["a"]))
+            .with_row_filter(RowFilter::new(vec![Box::new(row_filter_a)]))
+            .with_batch_size(10)
+            .with_limit(10)
+            .build()
+            .unwrap();
+
+        // First row group: filter columns fetch (predicate is evaluated here)
+        let ranges = expect_needs_data(decoder.try_decode());
+        push_ranges_to_decoder(&mut decoder, ranges);
+
+        // The first 10 matching rows come out: 176..185, column "a"
+        let batch = expect_data(decoder.try_decode());
+        let expected = TEST_BATCH.slice(176, 10).project(&[0]).unwrap();
+        assert_eq!(batch, expected);
+
+        // no data for row group 1 should be requested — the limit
+        // was satisfied by row group 0 and the `Start` state for row group 1
+        // short-circuits to `Finished`.
+        expect_finished(decoder.try_decode());
+
+        // within row group 0, the predicate should have seen
+        // strictly fewer than 200 rows — the short-circuit stops filter
+        // evaluation as soon as `limit` matches have been produced. Without
+        // the fix, all 200 rows of row group 0 would be evaluated.
+        let evaluated = rows_filtered.load(Ordering::Relaxed);
+        assert!(
+            evaluated < 200,
+            "predicate evaluated {evaluated} rows; expected < 200 with limit=10 short-circuit"
+        );
+        // Row 186 is the 11th match; the scan should stop no later than the
+        // batch containing it (batch 18 of 10 rows = rows 180..189), so at
+        // most 190 rows are evaluated.
+        assert!(
+            evaluated <= 190,
+            "predicate evaluated {evaluated} rows; expected ≤ 190 (stop within batch containing 11th match)"
+        );
+    }
+
+    /// Once the limit has been satisfied by a prior row group, subsequent
+    /// row groups should be skipped entirely — no data request for their
+    /// filter columns.
+    #[test]
+    fn test_decoder_filter_with_limit_skips_later_row_groups() {
+        let builder =
+            ParquetPushDecoderBuilder::try_new_decoder(test_file_parquet_metadata()).unwrap();
+        let schema_descr = builder.metadata().file_metadata().schema_descr_ptr();
+
+        // `a > 175` matches rows 176..199 in row group 0 (24 matches) and
+        // 200..399 in row group 1 (200 matches). With limit = 5, all matches
+        // should come from row group 0.
+        let row_filter_a = ArrowPredicateFn::new(
+            ProjectionMask::columns(&schema_descr, ["a"]),
+            |batch: RecordBatch| {
+                let scalar_175 = Int64Array::new_scalar(175);
+                let column = batch.column(0).as_primitive::<Int64Type>();
+                gt(column, &scalar_175)
+            },
+        );
+
+        let mut decoder = builder
+            .with_projection(ProjectionMask::columns(&schema_descr, ["a"]))
+            .with_row_filter(RowFilter::new(vec![Box::new(row_filter_a)]))
+            .with_limit(5)
+            .build()
+            .unwrap();
+
+        // Row group 0: fetch filter pages
+        let ranges = expect_needs_data(decoder.try_decode());
+        push_ranges_to_decoder(&mut decoder, ranges);
+
+        // First 5 matches: 176..180
+        let batch = expect_data(decoder.try_decode());
+        let expected = TEST_BATCH.slice(176, 5).project(&[0]).unwrap();
+        assert_eq!(batch, expected);
+
+        // Row group 1 must NOT request data — the limit is already satisfied
+        // so `Start` in row group 1 short-circuits to `Finished`.
+        expect_finished(decoder.try_decode());
+    }
+
+    /// The predicate short-circuit must account for `self.offset` as well as
+    /// `self.limit`. The post-predicate `with_offset` step skips that many
+    /// already-selected rows before `with_limit` counts output rows — so the
+    /// predicate must retain at least `offset + limit` matches. Without the
+    /// fix, Layer 2 caps at just `limit` and the later `with_offset` consumes
+    /// all of them, producing 0 rows instead of `limit`.
+    ///
+    /// `a > 175` matches rows 176..199 in row group 0 (24 matches). With
+    /// `offset = 10, limit = 5`, the expected output is rows 186..190 (the
+    /// 11th through 15th matches).
+    #[test]
+    fn test_decoder_filter_with_offset_and_limit() {
+        let builder =
+            ParquetPushDecoderBuilder::try_new_decoder(test_file_parquet_metadata()).unwrap();
+        let schema_descr = builder.metadata().file_metadata().schema_descr_ptr();
+
+        let row_filter_a = ArrowPredicateFn::new(
+            ProjectionMask::columns(&schema_descr, ["a"]),
+            |batch: RecordBatch| {
+                let scalar_175 = Int64Array::new_scalar(175);
+                let column = batch.column(0).as_primitive::<Int64Type>();
+                gt(column, &scalar_175)
+            },
+        );
+
+        let mut decoder = builder
+            .with_projection(ProjectionMask::columns(&schema_descr, ["a"]))
+            .with_row_filter(RowFilter::new(vec![Box::new(row_filter_a)]))
+            .with_offset(10)
+            .with_limit(5)
+            .build()
+            .unwrap();
+
+        let ranges = expect_needs_data(decoder.try_decode());
+        push_ranges_to_decoder(&mut decoder, ranges);
+
+        let batch = expect_data(decoder.try_decode());
+        let expected = TEST_BATCH.slice(186, 5).project(&[0]).unwrap();
+        assert_eq!(batch, expected);
+
+        expect_finished(decoder.try_decode());
+    }
+
+    /// The limit short-circuit must also be correct when the limited predicate
+    /// is the last predicate in a multi-predicate chain.
+    ///
+    /// `a > 175` first narrows row group 0 to rows 176..199. The final
+    /// predicate `b < 625` is then evaluated only over those 24 rows, all of
+    /// which match. With `limit = 10`, the final output should still be rows
+    /// 176..185, and the second predicate should stop before consuming all 24
+    /// selected rows.
+    #[test]
+    fn test_decoder_multi_filters_with_limit() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+
+        let builder =
+            ParquetPushDecoderBuilder::try_new_decoder(test_file_parquet_metadata()).unwrap();
+        let schema_descr = builder.metadata().file_metadata().schema_descr_ptr();
+
+        let first_predicate_rows = Arc::new(AtomicUsize::new(0));
+        let second_predicate_rows = Arc::new(AtomicUsize::new(0));
+
+        let first_predicate_rows_for_filter = Arc::clone(&first_predicate_rows);
+        let row_filter_a = ArrowPredicateFn::new(
+            ProjectionMask::columns(&schema_descr, ["a"]),
+            move |batch: RecordBatch| {
+                first_predicate_rows_for_filter.fetch_add(batch.num_rows(), Ordering::Relaxed);
+                let scalar_175 = Int64Array::new_scalar(175);
+                let column = batch.column(0).as_primitive::<Int64Type>();
+                gt(column, &scalar_175)
+            },
+        );
+
+        let second_predicate_rows_for_filter = Arc::clone(&second_predicate_rows);
+        let row_filter_b = ArrowPredicateFn::new(
+            ProjectionMask::columns(&schema_descr, ["b"]),
+            move |batch: RecordBatch| {
+                second_predicate_rows_for_filter.fetch_add(batch.num_rows(), Ordering::Relaxed);
+                let scalar_625 = Int64Array::new_scalar(625);
+                let column = batch.column(0).as_primitive::<Int64Type>();
+                lt(column, &scalar_625)
+            },
+        );
+
+        let mut decoder = builder
+            .with_projection(ProjectionMask::columns(&schema_descr, ["c"]))
+            .with_row_filter(RowFilter::new(vec![
+                Box::new(row_filter_a),
+                Box::new(row_filter_b),
+            ]))
+            .with_batch_size(10)
+            .with_limit(10)
+            .build()
+            .unwrap();
+
+        // Row group 0, first predicate
+        let ranges = expect_needs_data(decoder.try_decode());
+        push_ranges_to_decoder(&mut decoder, ranges);
+
+        // Row group 0, second predicate
+        let ranges = expect_needs_data(decoder.try_decode());
+        push_ranges_to_decoder(&mut decoder, ranges);
+
+        // Final projected data
+        let ranges = expect_needs_data(decoder.try_decode());
+        push_ranges_to_decoder(&mut decoder, ranges);
+
+        let batch = expect_data(decoder.try_decode());
+        let expected = TEST_BATCH.slice(176, 10).project(&[2]).unwrap();
+        assert_eq!(batch, expected);
+
+        // The overall limit was satisfied by row group 0.
+        expect_finished(decoder.try_decode());
+
+        assert_eq!(first_predicate_rows.load(Ordering::Relaxed), 200);
+        assert!(
+            second_predicate_rows.load(Ordering::Relaxed) < 24,
+            "final predicate should short-circuit before consuming all 24 rows selected by the first predicate"
+        );
+    }
+
+    /// When a row selection already exists, limiting the predicate must still
+    /// preserve alignment with that prior selection.
+    ///
+    /// The explicit selection narrows row group 0 to rows 150..199. Applying
+    /// `a > 175` over that selection yields rows 176..199. With `limit = 10`,
+    /// the decoder should emit rows 176..185 and stop without evaluating the
+    /// remaining selected rows.
+    #[test]
+    fn test_decoder_filter_with_row_selection_and_limit() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+
+        let builder =
+            ParquetPushDecoderBuilder::try_new_decoder(test_file_parquet_metadata()).unwrap();
+        let schema_descr = builder.metadata().file_metadata().schema_descr_ptr();
+
+        let rows_filtered = Arc::new(AtomicUsize::new(0));
+        let rows_filtered_for_predicate = Arc::clone(&rows_filtered);
+
+        let row_filter_a = ArrowPredicateFn::new(
+            ProjectionMask::columns(&schema_descr, ["a"]),
+            move |batch: RecordBatch| {
+                rows_filtered_for_predicate.fetch_add(batch.num_rows(), Ordering::Relaxed);
+                let scalar_175 = Int64Array::new_scalar(175);
+                let column = batch.column(0).as_primitive::<Int64Type>();
+                gt(column, &scalar_175)
+            },
+        );
+
+        let mut decoder = builder
+            .with_projection(ProjectionMask::columns(&schema_descr, ["a"]))
+            .with_row_selection(RowSelection::from(vec![
+                RowSelector::skip(150),
+                RowSelector::select(50),
+            ]))
+            .with_row_filter(RowFilter::new(vec![Box::new(row_filter_a)]))
+            .with_batch_size(10)
+            .with_limit(10)
+            .build()
+            .unwrap();
+
+        let ranges = expect_needs_data(decoder.try_decode());
+        push_ranges_to_decoder(&mut decoder, ranges);
+
+        let batch = expect_data(decoder.try_decode());
+        let expected = TEST_BATCH.slice(176, 10).project(&[0]).unwrap();
+        assert_eq!(batch, expected);
+
+        expect_finished(decoder.try_decode());
+
+        assert!(
+            rows_filtered.load(Ordering::Relaxed) < 50,
+            "predicate should short-circuit before consuming all 50 rows from the explicit row selection"
+        );
+    }
+
     #[test]
     fn test_decoder_offset_limit() {
         let mut decoder = ParquetPushDecoderBuilder::try_new_decoder(test_file_parquet_metadata())
diff --git a/parquet/src/arrow/push_decoder/reader_builder/filter.rs b/parquet/src/arrow/push_decoder/reader_builder/filter.rs
index 380211cca66e..219ab06da9b3 100644
--- a/parquet/src/arrow/push_decoder/reader_builder/filter.rs
+++ b/parquet/src/arrow/push_decoder/reader_builder/filter.rs
@@ -126,6 +126,13 @@ impl FilterInfo {
             .as_ref()
     }
 
+    /// Returns `true` if the current predicate is the last one in the chain
+    /// (i.e. the next call to [`Self::advance`] will return
+    /// [`AdvanceResult::Done`]).
+    pub(super) fn is_last(&self) -> bool {
+        self.next_predicate.get() == self.filter.predicates.len()
+    }
+
     /// Return a reference to the cache projection
     pub(super) fn cache_projection(&self) -> &ProjectionMask {
         &self.cache_info.cache_projection
diff --git a/parquet/src/arrow/push_decoder/reader_builder/mod.rs b/parquet/src/arrow/push_decoder/reader_builder/mod.rs
index d3d78ca7c263..aca0fc006c7b 100644
--- a/parquet/src/arrow/push_decoder/reader_builder/mod.rs
+++ b/parquet/src/arrow/push_decoder/reader_builder/mod.rs
@@ -304,6 +304,26 @@ impl RowGroupReaderBuilder {
     ) -> Result<NextState, ParquetError> {
         let result = match current_state {
             RowGroupDecoderState::Start { row_group_info } => {
+                // Short-circuit once the overall output limit is exhausted.
+                //
+                // `self.limit` tracks how many more rows the reader is still
+                // allowed to emit and is decremented as each row group is
+                // planned in `StartData`, so `Some(0)` means earlier row
+                // groups have already produced the full requested output.
+                //
+                // Without this guard we would still fall through to
+                // `Filters`, set up the filter plan, fetch at least one
+                // batch of this row group's filter columns, and fully
+                // evaluate any intermediate predicates before
+                // `with_limit(Some(0))` discarded the selection in
+                // `StartData` and finally returned `Finished`.
+                if matches!(self.limit, Some(0)) {
+                    return Ok(NextState::result(
+                        RowGroupDecoderState::Finished,
+                        DecodeResult::Finished,
+                    ));
+                }
+
                 let column_chunks = None; // no prior column chunks
 
                 let Some(filter) = self.filter.take() else {
@@ -452,10 +472,30 @@ impl RowGroupReaderBuilder {
                     predicate.projection(),
                     self.row_group_offset_index(row_group_idx),
                 );
-                // `with_predicate` actually evaluates the filter
-
-                plan_builder =
-                    plan_builder.with_predicate(array_reader, filter_info.current_mut())?;
+                // `with_predicate` actually evaluates the filter.
+                //
+                // When this is the final predicate in the chain and an output
+                // limit is set, tell the filter evaluation to stop once enough
+                // matching rows have been accumulated. Only safe for the last
+                // predicate: match counts of intermediate predicates do not
+                // correspond 1:1 to output rows.
+                //
+                // The cap must include `self.offset`: the later
+                // `with_offset(self.offset)` step skips that many selected
+                // rows before `with_limit` counts output, so the predicate
+                // must retain at least `offset + limit` matches or offset
+                // would drop matches the output still needs.
+                let predicate_limit = self
+                    .limit
+                    .filter(|_| filter_info.is_last())
+                    .map(|l| l.saturating_add(self.offset.unwrap_or(0)));
+
+                plan_builder = plan_builder.with_predicate_limited(
+                    array_reader,
+                    filter_info.current_mut(),
+                    predicate_limit,
+                    row_count,
+                )?;
 
                 let row_group_info = RowGroupInfo {
                     row_group_idx,

From 011e068210ad51d272982d48c5ea8795419ee605 Mon Sep 17 00:00:00 2001
From: Huaijin <haohuaijin@gmail.com>
Date: Sun, 19 Apr 2026 15:56:48 +0800
Subject: [PATCH 2/6] fix msrv

---
 parquet/src/arrow/arrow_reader/read_plan.rs | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/parquet/src/arrow/arrow_reader/read_plan.rs b/parquet/src/arrow/arrow_reader/read_plan.rs
index 670615577284..85241908b086 100644
--- a/parquet/src/arrow/arrow_reader/read_plan.rs
+++ b/parquet/src/arrow/arrow_reader/read_plan.rs
@@ -240,11 +240,11 @@ impl ReadPlanBuilder {
         // reader would have produced — rows past the early break are marked
         // "not selected". When no limit is set the loop always exhausts and
         // no padding is needed.
-        if let Some(expected) = expected_rows
-            && processed_rows < expected
-        {
-            let pad_len = expected - processed_rows;
-            filters.push(BooleanArray::new(BooleanBuffer::new_unset(pad_len), None));
+        if let Some(expected) = expected_rows {
+            if processed_rows < expected {
+                let pad_len = expected - processed_rows;
+                filters.push(BooleanArray::new(BooleanBuffer::new_unset(pad_len), None));
+            }
         }
 
         // If the predicate selected all rows and there is no prior selection,

From 4418425ea01ef9d6b20ab6fa60e8e08f7ae2cbee Mon Sep 17 00:00:00 2001
From: Huaijin <haohuaijin@gmail.com>
Date: Tue, 21 Apr 2026 21:16:02 +0800
Subject: [PATCH 3/6] update comment

---
 parquet/src/arrow/push_decoder/mod.rs                | 10 +---------
 parquet/src/arrow/push_decoder/reader_builder/mod.rs |  7 -------
 2 files changed, 1 insertion(+), 16 deletions(-)

diff --git a/parquet/src/arrow/push_decoder/mod.rs b/parquet/src/arrow/push_decoder/mod.rs
index 26f57ce54c9d..4c667e534366 100644
--- a/parquet/src/arrow/push_decoder/mod.rs
+++ b/parquet/src/arrow/push_decoder/mod.rs
@@ -1149,18 +1149,10 @@ mod test {
         // short-circuits to `Finished`.
         expect_finished(decoder.try_decode());
 
-        // within row group 0, the predicate should have seen
-        // strictly fewer than 200 rows — the short-circuit stops filter
-        // evaluation as soon as `limit` matches have been produced. Without
-        // the fix, all 200 rows of row group 0 would be evaluated.
-        let evaluated = rows_filtered.load(Ordering::Relaxed);
-        assert!(
-            evaluated < 200,
-            "predicate evaluated {evaluated} rows; expected < 200 with limit=10 short-circuit"
-        );
         // Row 186 is the 11th match; the scan should stop no later than the
         // batch containing it (batch 18 of 10 rows = rows 180..189), so at
         // most 190 rows are evaluated.
+        let evaluated = rows_filtered.load(Ordering::Relaxed);
         assert!(
             evaluated <= 190,
             "predicate evaluated {evaluated} rows; expected ≤ 190 (stop within batch containing 11th match)"
diff --git a/parquet/src/arrow/push_decoder/reader_builder/mod.rs b/parquet/src/arrow/push_decoder/reader_builder/mod.rs
index c620389dc624..01b53c13a921 100644
--- a/parquet/src/arrow/push_decoder/reader_builder/mod.rs
+++ b/parquet/src/arrow/push_decoder/reader_builder/mod.rs
@@ -315,13 +315,6 @@ impl RowGroupReaderBuilder {
                 // allowed to emit and is decremented as each row group is
                 // planned in `StartData`, so `Some(0)` means earlier row
                 // groups have already produced the full requested output.
-                //
-                // Without this guard we would still fall through to
-                // `Filters`, set up the filter plan, fetch at least one
-                // batch of this row group's filter columns, and fully
-                // evaluate any intermediate predicates before
-                // `with_limit(Some(0))` discarded the selection in
-                // `StartData` and finally returned `Finished`.
                 if matches!(self.limit, Some(0)) {
                     return Ok(NextState::result(
                         RowGroupDecoderState::Finished,

From 0e189eebf77b5782a92d875241b5f50b9b901cb0 Mon Sep 17 00:00:00 2001
From: Huaijin <haohuaijin@gmail.com>
Date: Tue, 21 Apr 2026 21:40:24 +0800
Subject: [PATCH 4/6] update comment

---
 parquet/src/arrow/push_decoder/reader_builder/mod.rs | 12 ++----------
 1 file changed, 2 insertions(+), 10 deletions(-)

diff --git a/parquet/src/arrow/push_decoder/reader_builder/mod.rs b/parquet/src/arrow/push_decoder/reader_builder/mod.rs
index 01b53c13a921..4a5046d29728 100644
--- a/parquet/src/arrow/push_decoder/reader_builder/mod.rs
+++ b/parquet/src/arrow/push_decoder/reader_builder/mod.rs
@@ -471,19 +471,11 @@ impl RowGroupReaderBuilder {
                     predicate.projection(),
                     self.row_group_offset_index(row_group_idx),
                 );
-                // `with_predicate` actually evaluates the filter.
+                // `with_predicate_limited` actually evaluates the filter.
                 //
                 // When this is the final predicate in the chain and an output
                 // limit is set, tell the filter evaluation to stop once enough
-                // matching rows have been accumulated. Only safe for the last
-                // predicate: match counts of intermediate predicates do not
-                // correspond 1:1 to output rows.
-                //
-                // The cap must include `self.offset`: the later
-                // `with_offset(self.offset)` step skips that many selected
-                // rows before `with_limit` counts output, so the predicate
-                // must retain at least `offset + limit` matches or offset
-                // would drop matches the output still needs.
+                // matching rows have been accumulated.
                 let predicate_limit = self
                     .limit
                     .filter(|_| filter_info.is_last())

From afe169e00c00d556b59585add0316a5685c8e278 Mon Sep 17 00:00:00 2001
From: Huaijin <haohuaijin@gmail.com>
Date: Tue, 21 Apr 2026 21:42:18 +0800
Subject: [PATCH 5/6] update

---
 parquet/src/arrow/push_decoder/reader_builder/mod.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/parquet/src/arrow/push_decoder/reader_builder/mod.rs b/parquet/src/arrow/push_decoder/reader_builder/mod.rs
index 4a5046d29728..4cbc68cac621 100644
--- a/parquet/src/arrow/push_decoder/reader_builder/mod.rs
+++ b/parquet/src/arrow/push_decoder/reader_builder/mod.rs
@@ -471,8 +471,7 @@ impl RowGroupReaderBuilder {
                     predicate.projection(),
                     self.row_group_offset_index(row_group_idx),
                 );
-                // `with_predicate_limited` actually evaluates the filter.
-                //
+
                 // When this is the final predicate in the chain and an output
                 // limit is set, tell the filter evaluation to stop once enough
                 // matching rows have been accumulated.
@@ -481,6 +480,7 @@ impl RowGroupReaderBuilder {
                     .filter(|_| filter_info.is_last())
                     .map(|l| l.saturating_add(self.offset.unwrap_or(0)));
 
+                // `with_predicate_limited` actually evaluates the filter.
                 plan_builder = plan_builder.with_predicate_limited(
                     array_reader,
                     filter_info.current_mut(),

From 8c3b313bcd0c5ab28b2e343fa77442b68128d856 Mon Sep 17 00:00:00 2001
From: Huaijin <haohuaijin@gmail.com>
Date: Fri, 24 Apr 2026 12:31:38 +0800
Subject: [PATCH 6/6] apply suggestion

---
 parquet/src/arrow/arrow_reader/mod.rs         |   2 +-
 parquet/src/arrow/arrow_reader/read_plan.rs   | 126 ++++++++++--------
 .../arrow/push_decoder/reader_builder/mod.rs  |  19 +--
 3 files changed, 86 insertions(+), 61 deletions(-)

diff --git a/parquet/src/arrow/arrow_reader/mod.rs b/parquet/src/arrow/arrow_reader/mod.rs
index 8a7e602618c2..f9ef4eac651f 100644
--- a/parquet/src/arrow/arrow_reader/mod.rs
+++ b/parquet/src/arrow/arrow_reader/mod.rs
@@ -49,7 +49,7 @@ use crate::schema::types::SchemaDescriptor;
 
 use crate::arrow::arrow_reader::metrics::ArrowReaderMetrics;
 // Exposed so integration tests and benchmarks can temporarily override the threshold.
-pub use read_plan::{ReadPlan, ReadPlanBuilder};
+pub use read_plan::{PredicateOptions, ReadPlan, ReadPlanBuilder};
 
 mod filter;
 pub mod metrics;
diff --git a/parquet/src/arrow/arrow_reader/read_plan.rs b/parquet/src/arrow/arrow_reader/read_plan.rs
index 85241908b086..ac2e105ecf4f 100644
--- a/parquet/src/arrow/arrow_reader/read_plan.rs
+++ b/parquet/src/arrow/arrow_reader/read_plan.rs
@@ -30,6 +30,52 @@ use arrow_buffer::{BooleanBuffer, BooleanBufferBuilder};
 use arrow_select::filter::prep_null_mask_filter;
 use std::collections::VecDeque;
 
+/// Options for [`ReadPlanBuilder::with_predicate_options`].
+pub struct PredicateOptions<'a> {
+    array_reader: Box<dyn ArrayReader>,
+    predicate: &'a mut dyn ArrowPredicate,
+    limit: Option<usize>,
+    total_rows: usize,
+}
+
+impl<'a> PredicateOptions<'a> {
+    /// Create options for evaluating `predicate` against rows produced by
+    /// `array_reader`.
+    ///
+    /// By default there is no match-count limit; the predicate is evaluated
+    /// over every row the reader yields. Use [`Self::with_limit`] to enable
+    /// early termination.
+    pub fn new(array_reader: Box<dyn ArrayReader>, predicate: &'a mut dyn ArrowPredicate) -> Self {
+        Self {
+            array_reader,
+            predicate,
+            limit: None,
+            total_rows: 0,
+        }
+    }
+
+    /// Stop scanning `array_reader` once `limit` matches have accumulated.
+    ///
+    /// Performance optimization for `LIMIT` / TopK: when the cumulative
+    /// `true_count` reaches `limit`, the current filter batch is truncated
+    /// at the `limit`-th match and remaining batches are never decoded.
+    ///
+    /// `limit` counts predicate matches, not output rows — callers applying
+    /// an offset must pass `offset + limit`.
+    ///
+    /// `total_rows` is the row count `array_reader` would yield if iterated
+    /// to completion. It is used to pad un-evaluated trailing rows as "not
+    /// selected" so the returned [`RowSelection`] covers the full row group.
+    ///
+    /// Only valid for the *last* predicate in a filter chain: intermediate
+    /// predicates' match counts do not map 1:1 to output rows.
+    pub fn with_limit(mut self, limit: usize, total_rows: usize) -> Self {
+        self.limit = Some(limit);
+        self.total_rows = total_rows;
+        self
+    }
+}
+
 /// A builder for [`ReadPlan`]
 #[derive(Clone, Debug)]
 pub struct ReadPlanBuilder {
@@ -149,45 +195,23 @@ impl ReadPlanBuilder {
         array_reader: Box<dyn ArrayReader>,
         predicate: &mut dyn ArrowPredicate,
     ) -> Result<Self> {
-        // No limit ⇒ loop always exhausts, so `total_rows` is not consulted.
-        self.with_predicate_limited(array_reader, predicate, None, 0)
+        self.with_predicate_options(PredicateOptions::new(array_reader, predicate))
     }
 
-    /// Like [`Self::with_predicate`] but additionally caps the number of
-    /// matches retained from `predicate` at `limit`, stopping the scan of
-    /// `array_reader` once that many matches have been accumulated.
-    ///
-    /// This is a performance optimization for TopK / `LIMIT` queries: once
-    /// the cumulative `true_count` across evaluated filter batches reaches
-    /// `limit`, the current batch's filter is truncated at the `limit`-th
-    /// match and iteration of `array_reader` stops. Remaining batches of
-    /// the predicate's column chunks are never decoded and the predicate is
-    /// never re-invoked — avoiding both decode and predicate work.
-    ///
-    /// `limit` counts predicate matches, not final output rows. Callers that
-    /// also apply an offset must pass `offset + limit` here so enough matches
-    /// survive the later offset step.
-    ///
-    /// `total_rows` is the number of rows the predicate's `array_reader`
-    /// would yield if iterated to completion (i.e. the row-group size when
-    /// no prior selection exists). It is only consulted on early break, to
-    /// pad the trailing un-evaluated rows as "not selected" so the resulting
-    /// [`RowSelection`] covers the full row group. When `limit` is `None`
-    /// this parameter is ignored.
+    /// Evaluates an [`ArrowPredicate`] with the given [`PredicateOptions`],
+    /// updating this plan's `selection`.
     ///
-    /// This must only be called when `predicate` is the *last* predicate in a
-    /// filter chain: match counts of intermediate predicates do not correspond
-    /// 1:1 to output rows, so truncating them early would drop rows that
-    /// should have been passed to subsequent predicates.
-    ///
-    /// If `limit` is `None` this behaves exactly like [`Self::with_predicate`].
-    pub fn with_predicate_limited(
-        mut self,
-        array_reader: Box<dyn ArrayReader>,
-        predicate: &mut dyn ArrowPredicate,
-        limit: Option<usize>,
-        total_rows: usize,
-    ) -> Result<Self> {
+    /// Like [`Self::with_predicate`], but allows additional options such as a
+    /// match-count limit for early termination (see
+    /// [`PredicateOptions::with_limit`]).
+    pub fn with_predicate_options(mut self, options: PredicateOptions<'_>) -> Result<Self> {
+        let PredicateOptions {
+            array_reader,
+            predicate,
+            limit,
+            total_rows,
+        } = options;
+
         // Target length for the concatenated filter output:
         // - Prior selection ⇒ the reader yields that many rows; `and_then`
         //   below requires the filter output to match.
@@ -202,7 +226,7 @@ impl ReadPlanBuilder {
         let reader = ParquetRecordBatchReader::new(array_reader, self.clone().build());
         let mut filters = vec![];
         let mut processed_rows: usize = 0;
-        let mut cumulative_matches: usize = 0;
+        let mut matched_rows: usize = 0;
         for maybe_batch in reader {
             let maybe_batch = maybe_batch?;
             let input_rows = maybe_batch.num_rows();
@@ -222,14 +246,14 @@ impl ReadPlanBuilder {
             processed_rows += input_rows;
 
             match limit {
-                Some(limit) if cumulative_matches + filter.true_count() >= limit => {
-                    let needed = limit - cumulative_matches;
-                    let truncated = truncate_filter_after_n_trues(&filter, needed);
+                Some(limit) if matched_rows + filter.true_count() >= limit => {
+                    let needed = limit - matched_rows;
+                    let truncated = truncate_filter_after_n_trues(filter, needed);
                     filters.push(truncated);
                     break;
                 }
                 _ => {
-                    cumulative_matches += filter.true_count();
+                    matched_rows += filter.true_count();
                     filters.push(filter);
                 }
             }
@@ -391,9 +415,9 @@ impl LimitedReadPlanBuilder {
 ///
 /// `filter` must not contain nulls (callers apply [`prep_null_mask_filter`]
 /// first). If `filter` has at most `n` `true` values, a clone is returned.
-fn truncate_filter_after_n_trues(filter: &BooleanArray, n: usize) -> BooleanArray {
+fn truncate_filter_after_n_trues(filter: BooleanArray, n: usize) -> BooleanArray {
     if filter.true_count() <= n {
-        return filter.clone();
+        return filter;
     }
     let len = filter.len();
     if n == 0 {
@@ -481,7 +505,7 @@ mod tests {
     fn truncate_filter_after_n_trues_keeps_first_n_matches() {
         let f = BooleanArray::from(vec![true, false, true, true, false, true, true]);
         // true positions: 0, 2, 3, 5, 6
-        let t = truncate_filter_after_n_trues(&f, 3);
+        let t = truncate_filter_after_n_trues(f.clone(), 3);
         assert_eq!(t.len(), f.len());
         assert_eq!(t.true_count(), 3);
         let out: Vec<bool> = (0..t.len()).map(|i| t.value(i)).collect();
@@ -495,7 +519,7 @@ mod tests {
     #[test]
     fn truncate_filter_after_n_trues_passes_through_when_already_small_enough() {
         let f = BooleanArray::from(vec![true, false, true, false]);
-        let t = truncate_filter_after_n_trues(&f, 5);
+        let t = truncate_filter_after_n_trues(f.clone(), 5);
         assert_eq!(t.len(), f.len());
         assert_eq!(t.true_count(), 2);
     }
@@ -503,13 +527,13 @@ mod tests {
     #[test]
     fn truncate_filter_after_n_trues_zero_returns_all_false() {
         let f = BooleanArray::from(vec![true, true, true]);
-        let t = truncate_filter_after_n_trues(&f, 0);
+        let t = truncate_filter_after_n_trues(f, 0);
         assert_eq!(t.len(), 3);
         assert_eq!(t.true_count(), 0);
     }
 
     #[test]
-    fn with_predicate_limited_pads_tail_when_no_prior_selection() {
+    fn with_predicate_options_limit_pads_tail_when_no_prior_selection() {
         use crate::arrow::ProjectionMask;
         use crate::arrow::array_reader::StructArrayReader;
         use crate::arrow::array_reader::test_util::InMemoryArrayReader;
@@ -540,11 +564,9 @@ mod tests {
         });
 
         let builder = ReadPlanBuilder::new(16)
-            .with_predicate_limited(
-                Box::new(struct_reader),
-                &mut predicate,
-                Some(LIMIT),
-                TOTAL_ROWS,
+            .with_predicate_options(
+                PredicateOptions::new(Box::new(struct_reader), &mut predicate)
+                    .with_limit(LIMIT, TOTAL_ROWS),
             )
             .unwrap();
 
diff --git a/parquet/src/arrow/push_decoder/reader_builder/mod.rs b/parquet/src/arrow/push_decoder/reader_builder/mod.rs
index 4cbc68cac621..60e50d29524e 100644
--- a/parquet/src/arrow/push_decoder/reader_builder/mod.rs
+++ b/parquet/src/arrow/push_decoder/reader_builder/mod.rs
@@ -24,7 +24,8 @@ use crate::arrow::array_reader::{ArrayReaderBuilder, CacheOptions, RowGroupCache
 use crate::arrow::arrow_reader::metrics::ArrowReaderMetrics;
 use crate::arrow::arrow_reader::selection::RowSelectionStrategy;
 use crate::arrow::arrow_reader::{
-    ParquetRecordBatchReader, ReadPlanBuilder, RowFilter, RowSelection, RowSelectionPolicy,
+    ParquetRecordBatchReader, PredicateOptions, ReadPlanBuilder, RowFilter, RowSelection,
+    RowSelectionPolicy,
 };
 use crate::arrow::in_memory_row_group::ColumnChunkData;
 use crate::arrow::push_decoder::reader_builder::data::DataRequestBuilder;
@@ -480,13 +481,15 @@ impl RowGroupReaderBuilder {
                     .filter(|_| filter_info.is_last())
                     .map(|l| l.saturating_add(self.offset.unwrap_or(0)));
 
-                // `with_predicate_limited` actually evaluates the filter.
-                plan_builder = plan_builder.with_predicate_limited(
-                    array_reader,
-                    filter_info.current_mut(),
-                    predicate_limit,
-                    row_count,
-                )?;
+                // Evaluate the filter via `with_predicate_options`, opting into
+                // early termination when this is the final predicate and an
+                // output limit was set.
+                let mut predicate_options =
+                    PredicateOptions::new(array_reader, filter_info.current_mut());
+                if let Some(limit) = predicate_limit {
+                    predicate_options = predicate_options.with_limit(limit, row_count);
+                }
+                plan_builder = plan_builder.with_predicate_options(predicate_options)?;
 
                 let row_group_info = RowGroupInfo {
                     row_group_idx,