Scan Delete Support Part 2: introduce DeleteFileManager skeleton. Use in ArrowReader (#950)

Second part of delete file read support. See
#630.

This PR provides the basis for delete file support within `ArrowReader`.

`DeleteFileManager` is introduced, in skeleton form. Full implementation
of its behaviour will be submitted in follow-up PRs.

`DeleteFileManager` is responsible for loading and parsing positional
and equality delete files from `FileIO`. Once delete files for a task
have been loaded and parsed, `ArrowReader::process_file_scan_task` uses
the resulting `DeleteFileManager` in two places:

* `DeleteFileManager::get_delete_vector_for_task` is passed a data file
path and will return an ~`Option<Vec<usize>>`~ `Option<RoaringTreeMap>`
containing the indices of all rows that are positionally deleted in that
data file (or `None` if there are none)
* `DeleteFileManager::build_delete_predicate` is invoked with the schema
from the file scan task. It will return an `Option<BoundPredicate>`
representing the filter predicate derived from all of the applicable
equality deletes being transformed into predicates, logically joined
into a single predicate and then bound to the schema (or `None` if there
are no applicable equality deletes)


This PR integrates the skeleton of the `DeleteFileManager` into
`ArrowReader::process_file_scan_task`, extending the `RowFilter` and
`RowSelection` logic to take into account any `RowFilter` that results
from equality deletes and any `RowSelection` that results from
positional deletes.

## Updates:
* refactored `DeleteFileManager` so that
`get_positional_delete_indexes_for_data_file` returns a `RoaringTreemap`
rather than a `Vec<usize>`. This was based on @liurenjie1024's
recommendation in a comment on the v1 PR, and makes a lot of sense from
a performance perspective and made it easier to implement
`ArrowReader::build_deletes_row_selection` in the follow-up PR to this
one, #951
* `DeleteFileManager` is instantiated in the `ArrowReader` constructor
rather than per-scan-task, so that delete files that apply to more than
one task don't end up getting loaded and parsed twice

## Potential further enhancements:
* Go one step further and move loading of delete files, and parsing of
positional delete files, into `ObjectCache` to ensure that loading and
parsing of the same files persists across scans

Author: sdd

Diff for: 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"

Diff for: 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 }

Diff for: crates/iceberg/Cargo.toml

@@ -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
+    }
+}

Diff for: crates/iceberg/src/arrow/delete_file_manager.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;