feat: introduce DeleteVector struct to decouple consumers from implementation details

Author: sdd

crates/iceberg/src/arrow/delete_file_manager.rs

@@ -15,8 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use roaring::RoaringTreemap;
-
+use crate::delete_vector::DeleteVector;
 use crate::expr::BoundPredicate;
 use crate::io::FileIO;
 use crate::scan::{ArrowRecordBatchStream, FileScanTaskDeleteFile};
@@ -42,7 +41,10 @@ impl DeleteFileManager for CachingDeleteFileManager {
     fn read_delete_file(_task: &FileScanTaskDeleteFile) -> Result<ArrowRecordBatchStream> {
         // TODO, implementation in https://github.com/apache/iceberg-rust/pull/982
 
-        unimplemented!()
+        Err(Error::new(
+            ErrorKind::FeatureUnsupported,
+            "Reading delete files is not yet supported",
+        ))
     }
 }
 
@@ -83,7 +85,7 @@ impl CachingDeleteFileManager {
     pub(crate) fn get_positional_delete_indexes_for_data_file(
         &self,
         data_file_path: &str,
-    ) -> Option<RoaringTreemap> {
+    ) -> Option<DeleteVector> {
         // TODO
 
         None

crates/iceberg/src/arrow/reader.rs

@@ -38,11 +38,11 @@ use parquet::arrow::async_reader::AsyncFileReader;
 use parquet::arrow::{ParquetRecordBatchStreamBuilder, ProjectionMask, PARQUET_FIELD_ID_META_KEY};
 use parquet::file::metadata::{ParquetMetaData, ParquetMetaDataReader, RowGroupMetaData};
 use parquet::schema::types::{SchemaDescriptor, Type as ParquetType};
-use roaring::RoaringTreemap;
 
 use crate::arrow::delete_file_manager::CachingDeleteFileManager;
 use crate::arrow::record_batch_transformer::RecordBatchTransformer;
 use crate::arrow::{arrow_schema_to_schema, get_arrow_datum};
+use crate::delete_vector::DeleteVector;
 use crate::error::Result;
 use crate::expr::visitors::bound_predicate_visitor::{visit, BoundPredicateVisitor};
 use crate::expr::visitors::page_index_evaluator::PageIndexEvaluator;
@@ -346,7 +346,7 @@ impl ArrowReader {
     fn build_deletes_row_selection(
         row_group_metadata: &[RowGroupMetaData],
         selected_row_groups: &Option<Vec<usize>>,
-        mut positional_deletes: RoaringTreemap,
+        mut positional_deletes: DeleteVector,
     ) -> Result<RowSelection> {
         // TODO
 

crates/iceberg/src/delete_vector.rs

@@ -0,0 +1,73 @@
+use roaring::RoaringTreemap;
+
+#[allow(unused)]
+pub struct DeleteVector {
+    inner: RoaringTreemap,
+}
+
+impl DeleteVector {
+    pub fn iter(&self) -> DeleteVectorIterator {
+        let mut iter = self.inner.bitmaps();
+        match iter.next() {
+            Some((high_bits, bitmap)) => {
+                DeleteVectorIterator {
+                    inner: Some(DeleteVectorIteratorInner {
+                        // iter,
+                        high_bits: (high_bits as u64) << 32,
+                        bitmap_iter: bitmap.iter(),
+                    }),
+                }
+            }
+            _ => DeleteVectorIterator { inner: None },
+        }
+    }
+}
+
+pub struct DeleteVectorIterator<'a> {
+    inner: Option<DeleteVectorIteratorInner<'a>>,
+}
+
+struct DeleteVectorIteratorInner<'a> {
+    // TODO: roaring::treemap::iter::BitmapIter is currently private.
+    // See https://github.com/RoaringBitmap/roaring-rs/issues/312
+    // iter: roaring::treemap::iter::BitmapIter<'a>,
+    high_bits: u64,
+    bitmap_iter: roaring::bitmap::Iter<'a>,
+}
+
+impl Iterator for DeleteVectorIterator<'_> {
+    type Item = u64;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let Some(ref mut inner) = &mut self.inner else {
+            return None;
+        };
+
+        if let Some(lower) = inner.bitmap_iter.next() {
+            return Some(inner.high_bits & lower as u64);
+        };
+
+        // TODO: roaring::treemap::iter::BitmapIter is currently private.
+        // See https://github.com/RoaringBitmap/roaring-rs/issues/312
+
+        // replace with commented-out code below once BitmapIter is pub,
+        // or use RoaringTreemap::iter if `advance_to` gets implemented natively
+        None
+
+        // let Some((high_bits, bitmap)) = inner.iter.next() else {
+        //     self.inner = None;
+        //     return None;
+        // };
+        //
+        // inner.high_bits = (high_bits as u64) << 32;
+        // inner.bitmap_iter = bitmap.iter();
+        //
+        // self.next()
+    }
+}
+
+impl<'a> DeleteVectorIterator<'a> {
+    pub fn advance_to(&'a mut self, _pos: u64) {
+        // TODO
+    }
+}

crates/iceberg/src/lib.rs

@@ -87,4 +87,5 @@ pub(crate) mod delete_file_index;
 mod utils;
 pub mod writer;
 
+mod delete_vector;
 mod puffin;