From 29a1ef9636c4dd0ca89ef4ff177a2f983fcf79cd Mon Sep 17 00:00:00 2001
From: Erik Bogado <erikbogado@gmail.com>
Date: Tue, 13 Oct 2026 17:48:57 -0300
Subject: [PATCH 1/2] perf: preserve native DV bitmap compression

Transfer decoded sub-bitmaps instead of reinserting each deleted row
to retain compressed runs and implicit high-partition keys.

Closes #5656
---
 native/core/src/execution/delta_dv.rs | 30 +++++++++++++++++++++------
 1 file changed, 24 insertions(+), 6 deletions(-)

diff --git a/native/core/src/execution/delta_dv.rs b/native/core/src/execution/delta_dv.rs
index 982c4434207..2a2a224fb3d 100644
--- a/native/core/src/execution/delta_dv.rs
+++ b/native/core/src/execution/delta_dv.rs
@@ -129,8 +129,8 @@ pub fn deserialize_dv_bitmap(data: &[u8]) -> Result<RoaringTreemap, ExecutionErr
                 )));
             }
             let mut pos = 4usize;
-            let mut treemap = RoaringTreemap::new();
-            for key in 0..count as u64 {
+            let mut bitmaps = Vec::new();
+            for key in 0..count as u32 {
                 if rest.len() < pos + 4 {
                     return Err(GeneralError(
                         "Native deletion vector bitmap truncated".to_string(),
@@ -148,11 +148,11 @@ pub fn deserialize_dv_bitmap(data: &[u8]) -> Result<RoaringTreemap, ExecutionErr
                         GeneralError(format!("Invalid deletion vector sub-bitmap: {e}"))
                     })?;
                 pos += size as usize;
-                for value in bitmap {
-                    treemap.insert((key << 32) | value as u64);
+                if !bitmap.is_empty() {
+                    bitmaps.push((key, bitmap));
                 }
             }
-            Ok(treemap)
+            Ok(RoaringTreemap::from_bitmaps(bitmaps))
         }
         other => Err(GeneralError(format!(
             "Unexpected RoaringBitmapArray magic number {other}"
@@ -1120,12 +1120,30 @@ mod tests {
 
     #[test]
     fn native_format_decodes() {
-        let values = [0u64, 2, 3, 100, (1u64 << 32) + 7];
+        let values = [0u64, 2, 3, 100, (2u64 << 32) + 7];
         let decoded = deserialize_dv_bitmap(&native_bytes(&values)).unwrap();
         let expected: RoaringTreemap = values.into_iter().collect();
         assert_eq!(decoded, expected);
     }
 
+    #[test]
+    fn native_decode_preserves_dense_bitmap_compression() {
+        let mut bitmap = RoaringBitmap::new();
+        bitmap.insert_range(0..1_000_000);
+        let compressed_size = bitmap.serialized_size();
+        let mut bitmap_bytes = Vec::new();
+        bitmap.serialize_into(&mut bitmap_bytes).unwrap();
+
+        let mut data = NATIVE_MAGIC.to_le_bytes().to_vec();
+        data.extend(1i32.to_le_bytes());
+        data.extend((bitmap_bytes.len() as i32).to_le_bytes());
+        data.extend(bitmap_bytes);
+
+        let decoded = deserialize_dv_bitmap(&data).unwrap();
+        let (_, decoded_bitmap) = decoded.bitmaps().next().unwrap();
+        assert_eq!(decoded_bitmap.serialized_size(), compressed_size);
+    }
+
     #[test]
     fn framing_rejects_bad_size_and_crc() {
         let deleted: RoaringTreemap = [1u64, 2].into_iter().collect();

From ae0df6e9b22689e4d040276b24ab598656188c96 Mon Sep 17 00:00:00 2001
From: Erik Bogado <erikbogado@gmail.com>
Date: Tue, 13 Oct 2026 23:12:19 -0300
Subject: [PATCH 2/2] test: assert native DV plan selection and empty-partition
 keys

Review on #6945 asked for the empty intermediate partition to carry its
intent where the value guards it, and for the dense native DV to be driven
through build_access_plan, which nothing in this file did before.
---
 native/core/src/execution/delta_dv.rs | 22 ++++++++++++++++++++++
 1 file changed, 22 insertions(+)

diff --git a/native/core/src/execution/delta_dv.rs b/native/core/src/execution/delta_dv.rs
index 2a2a224fb3d..2faf00c6c68 100644
--- a/native/core/src/execution/delta_dv.rs
+++ b/native/core/src/execution/delta_dv.rs
@@ -1124,6 +1124,10 @@ mod tests {
         let decoded = deserialize_dv_bitmap(&native_bytes(&values)).unwrap();
         let expected: RoaringTreemap = values.into_iter().collect();
         assert_eq!(decoded, expected);
+        // `native_bytes` emits an entry for every key up to the maximum, so key 1 is an empty
+        // partition. The decoder must omit it without shifting key 2's partition index.
+        let keys: Vec<u32> = decoded.bitmaps().map(|(key, _)| key).collect();
+        assert_eq!(keys, vec![0, 2]);
     }
 
     #[test]
@@ -1142,6 +1146,24 @@ mod tests {
         let decoded = deserialize_dv_bitmap(&data).unwrap();
         let (_, decoded_bitmap) = decoded.bitmaps().next().unwrap();
         assert_eq!(decoded_bitmap.serialized_size(), compressed_size);
+
+        // Dense native DVs now reach `build_access_plan`, so the same container must survive
+        // plan construction. Group 0 (rows 0..600_000) is fully deleted and must be skipped;
+        // group 1 (rows 600_000..1_200_000) has its first 400_000 rows deleted, leaving a
+        // trailing run of 200_000 live rows to select.
+        assert_eq!(decoded.len(), 1_000_000);
+        let plan = full_file_access_plan(&[600_000, 600_000], &decoded).unwrap();
+        assert_eq!(&plan.inner()[0], &RowGroupAccess::Skip);
+        match &plan.inner()[1] {
+            RowGroupAccess::Selection(sel) => {
+                let selectors: Vec<RowSelector> = sel.clone().into();
+                assert_eq!(
+                    selectors,
+                    vec![RowSelector::skip(400_000), RowSelector::select(200_000)]
+                );
+            }
+            other => panic!("expected selection, got {other:?}"),
+        }
     }
 
     #[test]