Scan Delete Support Part 2: introduce DeleteFileManager skeleton. Use in ArrowReader

Cargo.toml

@@ -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"

crates/iceberg/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/src/arrow/delete_file_manager.rs

@@ -0,0 +1,93 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::delete_vector::DeleteVector;
+use crate::expr::BoundPredicate;
+use crate::io::FileIO;
+use crate::scan::{ArrowRecordBatchStream, FileScanTaskDeleteFile};
+use crate::spec::SchemaRef;
+use crate::{Error, ErrorKind, Result};
+
+#[allow(unused)]
+pub trait DeleteFileManager {
+    /// Read the delete file referred to in the task
+    ///
+    /// Returns the raw contents of the delete file as a RecordBatch stream
+    fn read_delete_file(task: &FileScanTaskDeleteFile) -> Result<ArrowRecordBatchStream>;
+}
+
+#[allow(unused)]
+#[derive(Clone, Debug)]
+pub(crate) struct CachingDeleteFileManager {
+    file_io: FileIO,
+    concurrency_limit_data_files: usize,
+}
+
+impl DeleteFileManager for CachingDeleteFileManager {
+    fn read_delete_file(_task: &FileScanTaskDeleteFile) -> Result<ArrowRecordBatchStream> {
+        // TODO, implementation in https://github.com/apache/iceberg-rust/pull/982
+
+        Err(Error::new(
+            ErrorKind::FeatureUnsupported,
+            "Reading delete files is not yet supported",
+        ))
+    }
+}
+
+#[allow(unused_variables)]
+impl CachingDeleteFileManager {
+    pub fn new(file_io: FileIO, concurrency_limit_data_files: usize) -> CachingDeleteFileManager {
+        Self {
+            file_io,
+            concurrency_limit_data_files,
+        }
+    }
+
+    pub(crate) async fn load_deletes(
+        &self,
+        delete_file_entries: Vec<FileScanTaskDeleteFile>,
+    ) -> Result<()> {
+        // TODO
+
+        if !delete_file_entries.is_empty() {
+            Err(Error::new(
+                ErrorKind::FeatureUnsupported,
+                "Reading delete files is not yet supported",
+            ))
+        } else {
+            Ok(())
+        }
+    }
+
+    pub(crate) fn build_delete_predicate(
+        &self,
+        snapshot_schema: SchemaRef,
+    ) -> Result<Option<BoundPredicate>> {
+        // TODO
+
+        Ok(None)
+    }
+
+    pub(crate) fn get_positional_delete_indexes_for_data_file(
+        &self,
+        data_file_path: &str,
+    ) -> Option<DeleteVector> {
+        // TODO
+
+        None
+    }
+}

crates/iceberg/src/arrow/mod.rs

@@ -19,6 +19,7 @@
 
 mod schema;
 pub use schema::*;
+pub(crate) mod delete_file_manager;
 mod reader;
 pub(crate) mod record_batch_projector;
 pub(crate) mod record_batch_transformer;