refactor: pos deletes for task represented as RoaringBitmap.

* refactor: only pass row groups metadata rather than entire
  parquet metadata to . This
  makes it easier to test  as
  we don't need to mock up a full

Author: sdd

Cargo.lock

@@ -86,6 +86,7 @@ port_scanner = "0.1.5"
 rand = "0.8.5"
 regex = "1.10.5"
 reqwest = { version = "0.12.2", default-features = false, features = ["json"] }
+roaring = "0.10"
 rust_decimal = "1.31"
 serde = { version = "1.0.204", features = ["rc"] }
 serde_bytes = "0.11.15"

Cargo.toml

@@ -72,6 +72,7 @@ parquet = { workspace = true, features = ["async"] }
 paste = { workspace = true }
 rand = { workspace = true }
 reqwest = { workspace = true }
+roaring = { workspace = true }
 rust_decimal = { workspace = true }
 serde = { workspace = true }
 serde_bytes = { workspace = true }

crates/iceberg/Cargo.toml

@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use roaring::RoaringTreemap;
+
 use crate::expr::BoundPredicate;
 use crate::io::FileIO;
 use crate::scan::FileScanTaskDeleteFile;
@@ -54,7 +56,7 @@ impl DeleteFileManager {
     pub(crate) fn get_positional_delete_indexes_for_data_file(
         &self,
         data_file_path: &str,
-    ) -> Option<Vec<usize>> {
+    ) -> Option<RoaringTreemap> {
         // TODO
 
         None

crates/iceberg/src/arrow/delete_file_manager.rs

@@ -36,8 +36,9 @@ use futures::{try_join, FutureExt, StreamExt, TryFutureExt, TryStreamExt};
 use parquet::arrow::arrow_reader::{ArrowPredicateFn, ArrowReaderOptions, RowFilter, RowSelection};
 use parquet::arrow::async_reader::AsyncFileReader;
 use parquet::arrow::{ParquetRecordBatchStreamBuilder, ProjectionMask, PARQUET_FIELD_ID_META_KEY};
-use parquet::file::metadata::{ParquetMetaData, ParquetMetaDataReader};
+use parquet::file::metadata::{ParquetMetaData, ParquetMetaDataReader, RowGroupMetaData};
 use parquet::schema::types::{SchemaDescriptor, Type as ParquetType};
+use roaring::RoaringTreemap;
 
 use crate::arrow::delete_file_manager::DeleteFileManager;
 use crate::arrow::record_batch_transformer::RecordBatchTransformer;
@@ -276,9 +277,9 @@ impl ArrowReader {
 
         if let Some(positional_delete_indexes) = positional_delete_indexes {
             let delete_row_selection = Self::build_deletes_row_selection(
-                record_batch_stream_builder.metadata(),
+                record_batch_stream_builder.metadata().row_groups(),
                 &selected_row_group_indices,
-                &positional_delete_indexes,
+                positional_delete_indexes,
             )?;
 
             // merge the row selection from the delete files with the row selection
@@ -342,9 +343,9 @@ impl ArrowReader {
     /// been skipped entirely by the filter predicate
     #[allow(unused)]
     fn build_deletes_row_selection(
-        parquet_metadata: &Arc<ParquetMetaData>,
+        row_group_metadata: &[RowGroupMetaData],
         selected_row_groups: &Option<Vec<usize>>,
-        positional_deletes: &[usize],
+        mut positional_deletes: RoaringTreemap,
     ) -> Result<RowSelection> {
         // TODO