refactor: skip RowFilter in DataFusion via per-run decoders (no arrow-rs changes)

Instead of adding a `with_fully_matched_row_groups` API to arrow-rs,
implement the optimization entirely in DataFusion by creating separate
ParquetPushDecoders for row groups that need filtering vs those that
are fully matched.

Key changes:
- Split row groups into consecutive runs of same filter requirement
  via `split_decoder_runs()`, preserving original row group ordering
  for ordered scans.
- Each filtered run gets its own RowFilter; fully-matched runs skip it.
- Use VecDeque<ParquetPushDecoder> in PushDecoderStreamState to chain
  decoders sequentially.
- Remove [patch.crates-io] arrow-rs fork dependency.

This aligns with the direction of per-row-group morsels: each decoder
run can naturally become a morsel when that infrastructure lands.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: xudong963

Cargo.lock

@@ -284,22 +284,3 @@ debug = false
 debug-assertions = false
 strip = "debuginfo"
 incremental = false
-
-[patch.crates-io]
-parquet = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow-array = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow-buffer = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow-cast = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow-ipc = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow-ord = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow-schema = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow-string = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow-select = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow-data = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow-arith = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow-csv = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow-json = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow-avro = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow-flight = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }
-arrow-row = { git = "https://github.com/xudong963/arrow-rs.git", branch = "skip-filter-fully-matched-row-groups" }

Cargo.toml

@@ -118,23 +118,40 @@ fn bench_fully_matched_filter(c: &mut Criterion) {
     });
 
     // Scenario 2: with pushdown, WITH fully_matched optimization
-    // (RowFilter skipped for all row groups)
+    // Uses two decoders: an empty filtered decoder (all RGs are fully matched)
+    // plus a second decoder without RowFilter for the fully matched RGs.
+    // This mirrors how DataFusion splits decoders in the opener.
     group.bench_function("all_fully_matched/pushdown_with_skip", |b| {
         b.iter(|| {
             let rf = rebuild_row_filter(dataset, &file_metrics);
+            // First decoder: with filter but no row groups (all are fully matched)
             let reader_metadata = ArrowReaderMetadata::try_new(
                 Arc::clone(&dataset.metadata),
                 Default::default(),
             )
             .unwrap();
-            let decoder = ParquetPushDecoderBuilder::new_with_metadata(reader_metadata)
-                .with_batch_size(8192)
-                .with_row_filter(rf)
-                .with_row_groups(all_rg_indices.clone())
-                .with_fully_matched_row_groups(all_rg_indices.clone())
-                .build()
-                .unwrap();
-            let rows = run_push_decoder(decoder, &dataset.file_bytes);
+            let filtered_decoder =
+                ParquetPushDecoderBuilder::new_with_metadata(reader_metadata)
+                    .with_batch_size(8192)
+                    .with_row_filter(rf)
+                    .with_row_groups(vec![]) // empty: no non-fully-matched RGs
+                    .build()
+                    .unwrap();
+            // Second decoder: no filter for fully matched row groups
+            let reader_metadata = ArrowReaderMetadata::try_new(
+                Arc::clone(&dataset.metadata),
+                Default::default(),
+            )
+            .unwrap();
+            let matched_decoder =
+                ParquetPushDecoderBuilder::new_with_metadata(reader_metadata)
+                    .with_batch_size(8192)
+                    .with_row_groups(all_rg_indices.clone())
+                    .build()
+                    .unwrap();
+            // Run filtered decoder first (finishes immediately), then matched
+            let rows = run_push_decoder(filtered_decoder, &dataset.file_bytes);
+            let rows = rows + run_push_decoder(matched_decoder, &dataset.file_bytes);
             assert_eq!(rows, TOTAL_ROWS);
         });
     });