feat: preserve bitmap-backed Parquet row selections (#25731)

## Which issue does this PR close?

- Closes #23883.
- Closes #24488.
- Follows up on #24186 using the row-group-local scan path merged in
#25608.

## Rationale for this change

An external index can supply a bitmap-backed Parquet row selection, but
splitting a file-level selection into row groups or intersecting it with
selector-backed page pruning still converts it to selectors. Keeping the
bitmap avoids materializing fragmented selector runs and lets pruning
use a bitwise intersection.

## What changes are included in this PR?

- Split bitmap-backed file selections using bitmap slices, preserving
partial-group masks and recognizing fully selected or skipped groups.
Keep the existing single-pass selector path.
- Promote incoming selectors to a bitmap when intersecting with an
existing bitmap-backed selection.
- Use Parquet's `RowSelection::total_row_count()` directly instead of
introducing the temporary `row_selection_len` helper from #24186.
- Add regression coverage for bitmap preservation and external
selections when a dynamic predicate changes between row groups.
Consolidate opener test imports at module scope.

The deprecated `into_overall_row_selection` implementation is unchanged.
Preparation and reverse scans use the local selections introduced by
#25608. Runtime pruning remains disabled for scans with selections.

## What is the testing strategy for this PR?

Unit tests cover selector/bitmap intersections, empty intersections,
file-length validation, non-byte-aligned bitmap slicing, and
preservation through preparation and reversal. Integration tests cover
bitmap selections spanning row groups and statistics pruning.

The dynamic-pruning regression covers both selector and bitmap external
selections, with an unselected control scan that verifies runtime
pruning occurs. Temporarily removing the selection guard makes the test
fail because rebuilding with `None` returns unselected rows; the guard
was restored after this check.

Passed targeted tests:

- `cargo test -p datafusion-datasource-parquet --lib access_plan::test
--offline` (26 tests)
- `cargo test -p datafusion-datasource-parquet --lib opener::test
--offline` (56 tests)
- `cargo test -p datafusion --test parquet_integration
external_access_plan --offline` (16 tests)

Pre-submission checks passed: `cargo fmt --all`, `cargo clippy
--all-targets --all-features -- -D warnings`, and the checks in `uv run
./dev/rust_lint.sh`. The final HTML documentation check was rerun
successfully with `uv run ./ci/scripts/check_docs_html.sh` after
installing its missing `cargo-depgraph` prerequisite.

## Are there any user-facing changes?

No public API or selected-row changes. Bitmap-backed external selections
retain their representation through splitting and intersection rather
than being converted to selectors.

Author: haohuaijin

datafusion/core/tests/parquet/external_access_plan.rs

@@ -23,6 +23,7 @@ use std::sync::Arc;
 use crate::parquet::utils::MetricsFinder;
 use crate::parquet::{Scenario, create_data_batch};
 
+use arrow::buffer::BooleanBuffer;
 use arrow::datatypes::SchemaRef;
 use arrow::util::pretty::pretty_format_batches;
 use datafusion::common::Result;
@@ -189,33 +190,82 @@ async fn row_selection_extension() {
 
 #[tokio::test]
 async fn row_selection_extension_spanning_row_groups() {
-    // A selection whose selectors straddle the row group boundary (row 4 is the
-    // last row of group 0, rows 5-6 are the first rows of group 1).
-    let parquet_metrics = TestFull {
-        access_plan: None,
-        row_selection: Some(ParquetRowSelection::new(RowSelection::from(vec![
+    // Row 4 is the last row of group 0; rows 5-6 begin group 1.
+    for selection in [
+        RowSelection::from(vec![
             RowSelector::skip(4),
             RowSelector::select(3),
             RowSelector::skip(3),
-        ]))),
-        expected_rows: 3,
+        ]),
+        RowSelection::from_boolean_buffer(BooleanBuffer::from(vec![
+            false, false, false, false, true, true, true, false, false, false,
+        ])),
+    ] {
+        let parquet_metrics = TestFull {
+            access_plan: None,
+            row_selection: Some(ParquetRowSelection::new(selection)),
+            expected_rows: 3,
+            expected_output: Some(&[
+                "+------+------------+",
+                "| utf8 | large_utf8 |",
+                "+------+------------+",
+                "|      |            |",
+                "| e    | e          |",
+                "| f    | f          |",
+                "+------+------------+",
+            ]),
+            predicate: None,
+        }
+        .run()
+        .await
+        .unwrap();
+
+        let bytes_scanned = metric_value(&parquet_metrics, "bytes_scanned").unwrap();
+        assert_ne!(bytes_scanned, 0, "metrics : {parquet_metrics:#?}");
+    }
+}
+
+#[tokio::test]
+async fn bitmap_row_selection_extension_with_predicate() {
+    // Pruning removes row group 0, leaving the bitmap-selected rows in group 1.
+    let parquet_metrics = TestFull {
+        access_plan: None,
+        row_selection: Some(ParquetRowSelection::new(RowSelection::from_boolean_buffer(
+            BooleanBuffer::from(vec![
+                false, false, true, false, true, true, true, false, false, false,
+            ]),
+        ))),
+        expected_rows: 2,
         expected_output: Some(&[
             "+------+------------+",
             "| utf8 | large_utf8 |",
             "+------+------------+",
-            "|      |            |",
             "| e    | e          |",
             "| f    | f          |",
             "+------+------------+",
         ]),
-        predicate: None,
+        predicate: Some(col("utf8").eq(lit("e"))),
     }
     .run()
     .await
     .unwrap();
 
-    let bytes_scanned = metric_value(&parquet_metrics, "bytes_scanned").unwrap();
-    assert_ne!(bytes_scanned, 0, "metrics : {parquet_metrics:#?}");
+    // Verify that statistics pruned row group 0.
+    let row_groups_pruned_statistics = parquet_metrics
+        .sum_by_name("row_groups_pruned_statistics")
+        .unwrap();
+    if let MetricValue::PruningMetrics {
+        pruning_metrics, ..
+    } = row_groups_pruned_statistics
+    {
+        assert_eq!(
+            pruning_metrics.pruned(),
+            1,
+            "metrics : {parquet_metrics:#?}"
+        );
+    } else {
+        unreachable!("metrics `row_groups_pruned_statistics` should exist")
+    }
 }
 
 #[tokio::test]
@@ -316,9 +366,9 @@ async fn mixed_bitmap_and_selector_selections() {
     // Each representation must keep coordinates local to its row group.
     TestFull {
         access_plan: Some(ParquetAccessPlan::new(vec![
-            RowGroupAccess::Selection(RowSelection::from(
-                arrow::buffer::BooleanBuffer::from(vec![true, false, true, false, false]),
-            )),
+            RowGroupAccess::Selection(RowSelection::from(BooleanBuffer::from(vec![
+                true, false, true, false, false,
+            ]))),
             RowGroupAccess::Selection(select_two_rows()),
         ])),
         row_selection: None,

datafusion/datasource-parquet/src/access_plan.rs

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use arrow::array::BooleanBufferBuilder;
 use arrow::datatypes::Schema;
 use datafusion_common::{Result, assert_eq_or_internal_err, exec_err};
 use datafusion_physical_expr::expressions::Column;
@@ -297,6 +298,54 @@ impl ParquetAccessPlan {
     pub fn try_new_from_overall_row_selection(
         selection: RowSelection,
         row_group_meta_data: &[RowGroupMetaData],
+    ) -> Result<Self> {
+        if selection.as_mask().is_some() {
+            Self::try_new_from_overall_row_selection_mask(selection, row_group_meta_data)
+        } else {
+            Self::try_new_from_overall_row_selection_selectors(
+                selection,
+                row_group_meta_data,
+            )
+        }
+    }
+
+    fn try_new_from_overall_row_selection_mask(
+        mut selection: RowSelection,
+        row_group_meta_data: &[RowGroupMetaData],
+    ) -> Result<Self> {
+        let selection_rows = selection.total_row_count();
+        let file_rows = row_group_meta_data
+            .iter()
+            .map(|rg| rg.num_rows() as usize)
+            .sum::<usize>();
+        if selection_rows != file_rows {
+            return exec_err!(
+                "Invalid Parquet RowSelection. File has {file_rows} rows, \
+                but selection specifies {selection_rows} rows."
+            );
+        }
+
+        // For masks, split_off uses bitmap slices without materializing
+        // selectors. Use row_count() to cache each partial group's count
+        // for later preparation.
+        let row_groups = row_group_meta_data
+            .iter()
+            .map(|rg| {
+                let row_count = rg.num_rows() as usize;
+                let group_selection = selection.split_off(row_count);
+                match group_selection.row_count() {
+                    0 => RowGroupAccess::Skip,
+                    selected if selected == row_count => RowGroupAccess::Scan,
+                    _ => RowGroupAccess::Selection(group_selection),
+                }
+            })
+            .collect();
+        Ok(Self::new(row_groups))
+    }
+
+    fn try_new_from_overall_row_selection_selectors(
+        selection: RowSelection,
+        row_group_meta_data: &[RowGroupMetaData],
     ) -> Result<Self> {
         // Keep this as a single pass over the selector stream rather than
         // repeatedly calling `RowSelection::split_off` per row group. The
@@ -392,6 +441,22 @@ impl ParquetAccessPlan {
             RowGroupAccess::Skip => RowGroupAccess::Skip,
             RowGroupAccess::Scan => RowGroupAccess::Selection(selection),
             RowGroupAccess::Selection(existing_selection) => {
+                // Parquet preserves bitmap backing only when both operands
+                // are masks. Promote selector-backed page pruning to retain
+                // an external index's bitmap and use a bitwise intersection.
+                // Revisit this conversion once Parquet optimizes mixed-backed
+                // intersections: https://github.com/apache/arrow-rs/issues/10423
+                let selection = if existing_selection.as_mask().is_some()
+                    && selection.as_mask().is_none()
+                {
+                    let mut mask = BooleanBufferBuilder::new(selection.total_row_count());
+                    for selector in selection.iter() {
+                        mask.append_n(selector.row_count, !selector.skip);
+                    }
+                    RowSelection::from(mask.finish())
+                } else {
+                    selection
+                };
                 RowGroupAccess::Selection(existing_selection.intersection(&selection))
             }
         }
@@ -838,6 +903,7 @@ impl PreparedAccessPlan {
 #[cfg(test)]
 mod test {
     use super::*;
+    use arrow::buffer::BooleanBuffer;
     use datafusion_common::assert_contains;
     use parquet::basic::LogicalType;
     use parquet::file::metadata::ColumnChunkMetaData;
@@ -862,7 +928,7 @@ mod test {
         // Check both input representations retain the same conversion behavior.
         let selectors =
             RowSelection::from(vec![RowSelector::skip(10), RowSelector::select(20)]);
-        let bitmap = RowSelection::from(arrow::buffer::BooleanBuffer::from(
+        let bitmap = RowSelection::from(BooleanBuffer::from(
             (0..30).map(|i| i >= 10).collect::<Vec<_>>(),
         ));
         for selection in [selectors, bitmap] {
@@ -986,6 +1052,110 @@ mod test {
         }
     }
 
+    #[test]
+    fn test_scan_selection_preserves_mask_backing() {
+        let mask = BooleanBuffer::from(vec![
+            true, true, false, false, true, true, false, false, true, true,
+        ]);
+        let selectors =
+            RowSelection::from(vec![RowSelector::select(5), RowSelector::skip(5)]);
+        // Both selector-backed page pruning and bitmap intersections retain
+        // the existing mask, including when no rows survive.
+        for incoming in [
+            selectors,
+            RowSelection::from(BooleanBuffer::from(vec![
+                true, true, true, true, true, false, false, false, false, false,
+            ])),
+            RowSelection::from(vec![RowSelector::skip(10)]),
+        ] {
+            let empty = incoming.row_count() == 0;
+            let mut plan = ParquetAccessPlan::new(vec![RowGroupAccess::Selection(
+                RowSelection::from(mask.clone()),
+            )]);
+            plan.scan_selection(0, incoming);
+            let RowGroupAccess::Selection(selection) = &plan.inner()[0] else {
+                panic!("expected selection");
+            };
+            let expected = if empty {
+                BooleanBuffer::new_unset(10)
+            } else {
+                BooleanBuffer::from(vec![
+                    true, true, false, false, true, false, false, false, false, false,
+                ])
+            };
+            assert_eq!(selection.as_mask(), Some(&expected));
+        }
+    }
+
+    #[test]
+    fn test_scan_selection_preserves_selector_backing() {
+        let mut plan = ParquetAccessPlan::new(vec![RowGroupAccess::Selection(
+            RowSelection::from(vec![RowSelector::select(6), RowSelector::skip(4)]),
+        )]);
+        plan.scan_selection(
+            0,
+            RowSelection::from(vec![
+                RowSelector::skip(2),
+                RowSelector::select(5),
+                RowSelector::skip(3),
+            ]),
+        );
+        let RowGroupAccess::Selection(selection) = &plan.inner()[0] else {
+            panic!("expected selection");
+        };
+        assert!(selection.as_mask().is_none());
+        assert_eq!(
+            selection,
+            &RowSelection::from(vec![
+                RowSelector::skip(2),
+                RowSelector::select(4),
+                RowSelector::skip(4),
+            ])
+        );
+    }
+
+    #[test]
+    fn test_new_from_overall_mask_preserves_bitmap_backing() {
+        // Include all-selected, all-skipped, and fragmented groups. Start at
+        // a non-byte-aligned offset to exercise slicing an existing bitmap.
+        let mut bits = vec![false; 3];
+        bits.extend(vec![true; 10]);
+        bits.extend(vec![false; 20]);
+        bits.extend((0..30).map(|i| i % 2 == 1));
+        bits.extend(vec![true; 40]);
+        let mask = BooleanBuffer::from(bits).slice(3, 100);
+        for cache_row_count in [false, true] {
+            let selection = RowSelection::from(mask.clone());
+            if cache_row_count {
+                // Exercise split_off's propagation of an already cached count.
+                assert_eq!(selection.row_count(), 65);
+            }
+            let plan = ParquetAccessPlan::try_new_from_overall_row_selection(
+                selection,
+                &ROW_GROUP_METADATA,
+            )
+            .unwrap();
+            assert_eq!(plan.inner()[0], RowGroupAccess::Scan);
+            assert_eq!(plan.inner()[1], RowGroupAccess::Skip);
+            assert_eq!(plan.inner()[3], RowGroupAccess::Scan);
+            let RowGroupAccess::Selection(selection) = &plan.inner()[2] else {
+                panic!("expected selection");
+            };
+            assert_eq!(selection.row_count(), 15);
+            let group_mask = mask.slice(30, 30);
+            assert!(selection.as_mask().unwrap().ptr_eq(&group_mask));
+
+            // The local selection must also survive preparation and reversal.
+            let prepared = plan.prepare(&ROW_GROUP_METADATA).unwrap().reverse();
+            assert_eq!(prepared.row_group_indexes(), vec![3, 2, 0]);
+            assert!(prepared.row_groups[0].selection.selection().is_none());
+            assert!(prepared.row_groups[2].selection.selection().is_none());
+            let selection = prepared.row_groups[1].selection.selection().unwrap();
+            assert_eq!(selection.row_count(), 15);
+            assert!(selection.as_mask().unwrap().ptr_eq(&group_mask));
+        }
+    }
+
     #[test]
     fn test_new_from_overall_row_selection() {
         let row_selection = RowSelection::from(vec![
@@ -1022,19 +1192,26 @@ mod test {
 
     #[test]
     fn test_new_from_overall_row_selection_invalid_row_count() {
-        let row_selection = RowSelection::from(vec![RowSelector::select(99)]);
-
-        let err = ParquetAccessPlan::try_new_from_overall_row_selection(
-            row_selection,
-            &ROW_GROUP_METADATA,
-        )
-        .unwrap_err()
-        .to_string();
-
-        assert_contains!(
-            err,
-            "Invalid Parquet RowSelection. File has 100 rows, but selection specifies 99 rows"
-        );
+        for selection_rows in [99, 101] {
+            for selection in [
+                RowSelection::from(vec![RowSelector::select(selection_rows)]),
+                RowSelection::from(BooleanBuffer::new_set(selection_rows)),
+            ] {
+                let err = ParquetAccessPlan::try_new_from_overall_row_selection(
+                    selection,
+                    &ROW_GROUP_METADATA,
+                )
+                .unwrap_err()
+                .to_string();
+                assert_contains!(
+                    err,
+                    format!(
+                        "Invalid Parquet RowSelection. File has 100 rows, \
+                         but selection specifies {selection_rows} rows"
+                    )
+                );
+            }
+        }
     }
 
     #[test]