generify

xanderbailey · xanderbailey · commit e88d44470300 · 2026-04-15T23:45:13.000+01:00
diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs
@@ -24,15 +24,25 @@ use crate::delete_file_index::DeleteFileIndex;
 use crate::expr::{Bind, BoundPredicate, Predicate};
 use crate::io::object_cache::ObjectCache;
 use crate::scan::{
-    AppendSnapshotSet, BoundPredicates, ExpressionEvaluatorCache, FileScanTask,
-    ManifestEvaluatorCache, PartitionFilterCache,
+    BoundPredicates, ExpressionEvaluatorCache, FileScanTask, ManifestEvaluatorCache,
+    PartitionFilterCache,
 };
 use crate::spec::{
-    ManifestContentType, ManifestEntryRef, ManifestFile, ManifestList, ManifestStatus, SchemaRef,
-    SnapshotRef, TableMetadataRef,
+    ManifestContentType, ManifestEntryRef, ManifestFile, ManifestList, SchemaRef, SnapshotRef,
+    TableMetadataRef,
 };
 use crate::{Error, ErrorKind, Result};
 
+/// Filter applied to each [`ManifestFile`] before fetching it.
+/// Returns `true` to include the manifest, `false` to skip it.
+pub(crate) type ManifestFileFilter =
+    Arc<dyn Fn(&ManifestFile) -> bool + Send + Sync>;
+
+/// Filter applied to each manifest entry after loading a manifest.
+/// Returns `true` to include the entry, `false` to skip it.
+pub(crate) type ManifestEntryFilter =
+    Arc<dyn Fn(&ManifestEntryRef) -> bool + Send + Sync>;
+
 /// Wraps a [`ManifestFile`] alongside the objects that are needed
 /// to process it in a thread-safe manner
 pub(crate) struct ManifestFileContext {
@@ -47,7 +57,7 @@ pub(crate) struct ManifestFileContext {
     expression_evaluator_cache: Arc<ExpressionEvaluatorCache>,
     delete_file_index: DeleteFileIndex,
     case_sensitive: bool,
-    snapshot_range: Option<Arc<AppendSnapshotSet>>,
+    entry_filter: Option<ManifestEntryFilter>,
 }
 
 /// Wraps a [`ManifestEntryRef`] alongside the objects that are needed
@@ -78,33 +88,16 @@ impl ManifestFileContext {
             expression_evaluator_cache,
             delete_file_index,
             case_sensitive,
-            snapshot_range,
+            entry_filter,
         } = self;
 
         let manifest = object_cache.get_manifest(&manifest_file).await?;
 
         for manifest_entry in manifest.entries() {
-            // For incremental scans, filter entries to only include those:
-            // 1. With status ADDED (not EXISTING or DELETED)
-            // 2. With a snapshot_id that falls within the range
-            if let Some(ref range) = snapshot_range {
-                // Only include entries with status ADDED
-                if manifest_entry.status() != ManifestStatus::Added {
+            if let Some(ref filter) = entry_filter {
+                if !filter(manifest_entry) {
                     continue;
                 }
-
-                // Only include entries from snapshots in the range
-                match manifest_entry.snapshot_id() {
-                    Some(entry_snapshot_id) => {
-                        if !range.contains(entry_snapshot_id) {
-                            continue;
-                        }
-                    }
-                    None => {
-                        // Skip entries without a snapshot_id in incremental mode
-                        continue;
-                    }
-                }
             }
 
             let manifest_entry_context = ManifestEntryContext {
@@ -171,7 +164,6 @@ impl ManifestEntryContext {
 
 /// PlanContext wraps a [`SnapshotRef`] alongside all the other
 /// objects that are required to perform a scan file plan.
-#[derive(Debug)]
 pub(crate) struct PlanContext {
     pub snapshot: SnapshotRef,
 
@@ -186,7 +178,25 @@ pub(crate) struct PlanContext {
     pub partition_filter_cache: Arc<PartitionFilterCache>,
     pub manifest_evaluator_cache: Arc<ManifestEvaluatorCache>,
     pub expression_evaluator_cache: Arc<ExpressionEvaluatorCache>,
-    pub snapshot_range: Option<Arc<AppendSnapshotSet>>,
+    pub manifest_file_filter: Option<ManifestFileFilter>,
+    pub manifest_entry_filter: Option<ManifestEntryFilter>,
+}
+
+impl std::fmt::Debug for PlanContext {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("PlanContext")
+            .field("snapshot", &self.snapshot)
+            .field("case_sensitive", &self.case_sensitive)
+            .field(
+                "manifest_file_filter",
+                &self.manifest_file_filter.as_ref().map(|_| "..."),
+            )
+            .field(
+                "manifest_entry_filter",
+                &self.manifest_entry_filter.as_ref().map(|_| "..."),
+            )
+            .finish_non_exhaustive()
+    }
 }
 
 impl PlanContext {
@@ -240,17 +250,8 @@ impl PlanContext {
         // TODO: Ideally we could ditch this intermediate Vec as we return an iterator.
         let mut filtered_mfcs = vec![];
         for manifest_file in manifest_files {
-            // For incremental scans, skip manifests that can't contain relevant entries:
-            // 1. Delete manifests — we only care about newly added data files.
-            // 2. Data manifests whose added_snapshot_id is outside the scan range —
-            //    they can't contain entries added in the snapshots we care about.
-            //    (We still keep the entry-level filter because a manifest can contain
-            //    entries from multiple snapshots via manifest reuse.)
-            if let Some(ref range) = self.snapshot_range {
-                if manifest_file.content == ManifestContentType::Deletes {
-                    continue;
-                }
-                if !range.contains(manifest_file.added_snapshot_id) {
+            if let Some(ref filter) = self.manifest_file_filter {
+                if !filter(manifest_file) {
                     continue;
                 }
             }
@@ -324,7 +325,7 @@ impl PlanContext {
             expression_evaluator_cache: self.expression_evaluator_cache.clone(),
             delete_file_index,
             case_sensitive: self.case_sensitive,
-            snapshot_range: self.snapshot_range.clone(),
+            entry_filter: self.manifest_entry_filter.clone(),
         }
     }
 }
diff --git a/crates/iceberg/src/scan/incremental.rs b/crates/iceberg/src/scan/incremental.rs
@@ -18,10 +18,12 @@
 //! Incremental append scan for reading only newly added data between snapshots.
 
 use std::collections::HashSet;
+use std::sync::Arc;
 
 use crate::expr::Predicate;
+use crate::scan::context::{ManifestEntryFilter, ManifestFileFilter};
 use crate::scan::{ScanConfig, TableScan, build_table_scan};
-use crate::spec::{Operation, TableMetadataRef};
+use crate::spec::{ManifestContentType, ManifestStatus, Operation, TableMetadataRef};
 use crate::table::Table;
 use crate::util::available_parallelism;
 use crate::util::snapshot::ancestors_between;
@@ -136,10 +138,32 @@ impl AppendSnapshotSet {
         Ok(Self { snapshot_ids })
     }
 
-    /// Check if a snapshot_id is within this range
+    /// Check if a snapshot_id is within this set
     pub(crate) fn contains(&self, snapshot_id: i64) -> bool {
         self.snapshot_ids.contains(&snapshot_id)
     }
+
+    /// Create a manifest file filter that skips delete manifests and data
+    /// manifests whose `added_snapshot_id` is outside this set.
+    pub(crate) fn manifest_file_filter(self: &Arc<Self>) -> ManifestFileFilter {
+        let set = self.clone();
+        Arc::new(move |manifest_file| {
+            manifest_file.content != ManifestContentType::Deletes
+                && set.contains(manifest_file.added_snapshot_id)
+        })
+    }
+
+    /// Create a manifest entry filter that includes only entries with
+    /// status ADDED and a snapshot_id within this set.
+    pub(crate) fn manifest_entry_filter(self: &Arc<Self>) -> ManifestEntryFilter {
+        let set = self.clone();
+        Arc::new(move |entry| {
+            entry.status() == ManifestStatus::Added
+                && entry
+                    .snapshot_id()
+                    .is_some_and(|id| set.contains(id))
+        })
+    }
 }
 
 /// Builder to create an incremental append scan between two snapshots.
@@ -294,12 +318,12 @@ impl<'a> IncrementalAppendScanBuilder<'a> {
             }
         };
 
-        let snapshot_range = AppendSnapshotSet::build(
+        let append_set = Arc::new(AppendSnapshotSet::build(
             &self.table.metadata_ref(),
             self.from_snapshot_id,
             to_snapshot.snapshot_id(),
             self.from_inclusive,
-        )?;
+        )?);
 
         build_table_scan(
             ScanConfig {
@@ -315,7 +339,8 @@ impl<'a> IncrementalAppendScanBuilder<'a> {
                 row_selection_enabled: self.row_selection_enabled,
             },
             to_snapshot,
-            Some(snapshot_range),
+            Some(append_set.manifest_file_filter()),
+            Some(append_set.manifest_entry_filter()),
         )
     }
 }
@@ -402,22 +427,27 @@ mod tests {
         let table = TableTestFixture::new().table;
         let current_snapshot_id = table.metadata().current_snapshot().unwrap().snapshot_id();
 
+        // Verify the scan builds successfully
         let result = table
             .incremental_append_scan_inclusive(current_snapshot_id)
             .to_snapshot(current_snapshot_id)
             .build();
-
         assert!(
             result.is_ok(),
             "Inclusive scan of a single append snapshot should succeed"
         );
 
-        let scan = result.unwrap();
-        let plan_context = scan.plan_context.as_ref().unwrap();
-        let range = plan_context.snapshot_range.as_ref().unwrap();
+        // Verify AppendSnapshotSet directly
+        let set = AppendSnapshotSet::build(
+            &table.metadata_ref(),
+            current_snapshot_id,
+            current_snapshot_id,
+            true,
+        )
+        .unwrap();
         assert!(
-            range.contains(current_snapshot_id),
-            "Inclusive range should contain the from_snapshot"
+            set.contains(current_snapshot_id),
+            "Inclusive set should contain the from_snapshot"
         );
     }
 
@@ -427,22 +457,27 @@ mod tests {
         let table = TableTestFixture::new().table;
         let current_snapshot_id = table.metadata().current_snapshot().unwrap().snapshot_id();
 
+        // Verify the scan builds successfully
         let result = table
             .incremental_append_scan(current_snapshot_id)
             .to_snapshot(current_snapshot_id)
             .build();
-
         assert!(
             result.is_ok(),
             "Exclusive scan from=to should succeed with empty range"
         );
 
-        let scan = result.unwrap();
-        let plan_context = scan.plan_context.as_ref().unwrap();
-        let range = plan_context.snapshot_range.as_ref().unwrap();
+        // Verify AppendSnapshotSet directly
+        let set = AppendSnapshotSet::build(
+            &table.metadata_ref(),
+            current_snapshot_id,
+            current_snapshot_id,
+            false,
+        )
+        .unwrap();
         assert!(
-            !range.contains(current_snapshot_id),
-            "Exclusive range should not contain the from_snapshot"
+            !set.contains(current_snapshot_id),
+            "Exclusive set should not contain the from_snapshot"
         );
     }
 
@@ -484,20 +519,20 @@ mod tests {
             "Range of only append snapshots should succeed"
         );
 
-        let scan = result.unwrap();
-        let range = scan
-            .plan_context
-            .as_ref()
-            .unwrap()
-            .snapshot_range
-            .as_ref()
-            .unwrap();
+        // Verify AppendSnapshotSet directly
+        let set = AppendSnapshotSet::build(
+            &table.metadata_ref(),
+            3051729675574597004,
+            3056729675574597004,
+            false,
+        )
+        .unwrap();
         assert!(
-            !range.contains(3051729675574597004),
+            !set.contains(3051729675574597004),
             "from_snapshot should be excluded"
         );
-        assert!(range.contains(3055729675574597004), "S2 should be in range");
-        assert!(range.contains(3056729675574597004), "S3 should be in range");
+        assert!(set.contains(3055729675574597004), "S2 should be in range");
+        assert!(set.contains(3056729675574597004), "S3 should be in range");
     }
 
     #[tokio::test]
diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs
@@ -30,7 +30,6 @@ use arrow_array::RecordBatch;
 use futures::channel::mpsc::{Sender, channel};
 use futures::stream::BoxStream;
 use futures::{SinkExt, StreamExt, TryStreamExt};
-pub(crate) use incremental::AppendSnapshotSet;
 pub use incremental::IncrementalAppendScanBuilder;
 pub use task::*;
 
@@ -69,7 +68,8 @@ pub(crate) struct ScanConfig<'a> {
 pub(crate) fn build_table_scan(
     config: ScanConfig<'_>,
     snapshot: SnapshotRef,
-    snapshot_range: Option<AppendSnapshotSet>,
+    manifest_file_filter: Option<ManifestFileFilter>,
+    manifest_entry_filter: Option<ManifestEntryFilter>,
 ) -> Result<TableScan> {
     let schema = snapshot.schema(config.table.metadata())?;
 
@@ -144,7 +144,8 @@ pub(crate) fn build_table_scan(
         partition_filter_cache: Arc::new(PartitionFilterCache::new()),
         manifest_evaluator_cache: Arc::new(ManifestEvaluatorCache::new()),
         expression_evaluator_cache: Arc::new(ExpressionEvaluatorCache::new()),
-        snapshot_range: snapshot_range.map(Arc::new),
+        manifest_file_filter,
+        manifest_entry_filter,
     };
 
     Ok(TableScan {
@@ -345,6 +346,7 @@ impl<'a> TableScanBuilder<'a> {
             },
             snapshot,
             None,
+            None,
         )
     }
 }
