feat: Incremental append scan

crates/iceberg/public-api.txt

@@ -1305,6 +1305,20 @@ impl serde_core::ser::Serialize for iceberg::scan::FileScanTaskDeleteFile
 pub fn iceberg::scan::FileScanTaskDeleteFile::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
 impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTaskDeleteFile
 pub fn iceberg::scan::FileScanTaskDeleteFile::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
+pub struct iceberg::scan::IncrementalAppendScanBuilder<'a>
+impl<'a> iceberg::scan::IncrementalAppendScanBuilder<'a>
+pub fn iceberg::scan::IncrementalAppendScanBuilder<'a>::build(self) -> iceberg::Result<iceberg::scan::TableScan>
+pub fn iceberg::scan::IncrementalAppendScanBuilder<'a>::select(self, column_names: impl core::iter::traits::collect::IntoIterator<Item = impl alloc::string::ToString>) -> Self
+pub fn iceberg::scan::IncrementalAppendScanBuilder<'a>::select_all(self) -> Self
+pub fn iceberg::scan::IncrementalAppendScanBuilder<'a>::select_empty(self) -> Self
+pub fn iceberg::scan::IncrementalAppendScanBuilder<'a>::with_batch_size(self, batch_size: core::option::Option<usize>) -> Self
+pub fn iceberg::scan::IncrementalAppendScanBuilder<'a>::with_case_sensitive(self, case_sensitive: bool) -> Self
+pub fn iceberg::scan::IncrementalAppendScanBuilder<'a>::with_concurrency_limit(self, limit: usize) -> Self
+pub fn iceberg::scan::IncrementalAppendScanBuilder<'a>::with_data_file_concurrency_limit(self, limit: usize) -> Self
+pub fn iceberg::scan::IncrementalAppendScanBuilder<'a>::with_filter(self, predicate: iceberg::expr::Predicate) -> Self
+pub fn iceberg::scan::IncrementalAppendScanBuilder<'a>::with_manifest_entry_concurrency_limit(self, limit: usize) -> Self
+pub fn iceberg::scan::IncrementalAppendScanBuilder<'a>::with_row_group_filtering_enabled(self, row_group_filtering_enabled: bool) -> Self
+pub fn iceberg::scan::IncrementalAppendScanBuilder<'a>::with_row_selection_enabled(self, row_selection_enabled: bool) -> Self
 pub struct iceberg::scan::ScanMetrics
 impl iceberg::scan::ScanMetrics
 pub fn iceberg::scan::ScanMetrics::bytes_read(&self) -> u64
@@ -3081,6 +3095,8 @@ pub struct iceberg::table::StaticTable(_)
 impl iceberg::table::StaticTable
 pub async fn iceberg::table::StaticTable::from_metadata(metadata: iceberg::spec::TableMetadata, table_ident: iceberg::TableIdent, file_io: iceberg::io::FileIO) -> iceberg::Result<Self>
 pub async fn iceberg::table::StaticTable::from_metadata_file(metadata_location: &str, table_ident: iceberg::TableIdent, file_io: iceberg::io::FileIO) -> iceberg::Result<Self>
+pub fn iceberg::table::StaticTable::incremental_append_scan(&self, from_snapshot_id: i64, to_snapshot_id: core::option::Option<i64>) -> iceberg::scan::IncrementalAppendScanBuilder<'_>
+pub fn iceberg::table::StaticTable::incremental_append_scan_inclusive(&self, from_snapshot_id: i64, to_snapshot_id: core::option::Option<i64>) -> iceberg::scan::IncrementalAppendScanBuilder<'_>
 pub fn iceberg::table::StaticTable::into_table(self) -> iceberg::table::Table
 pub fn iceberg::table::StaticTable::metadata(&self) -> iceberg::spec::TableMetadataRef
 pub fn iceberg::table::StaticTable::reader_builder(&self) -> iceberg::arrow::ArrowReaderBuilder
@@ -3096,6 +3112,8 @@ pub fn iceberg::table::Table::current_schema_ref(&self) -> iceberg::spec::Schema
 pub fn iceberg::table::Table::encryption_manager(&self) -> core::option::Option<&iceberg::encryption::EncryptionManager>
 pub fn iceberg::table::Table::file_io(&self) -> &iceberg::io::FileIO
 pub fn iceberg::table::Table::identifier(&self) -> &iceberg::TableIdent
+pub fn iceberg::table::Table::incremental_append_scan(&self, from_snapshot_id: i64, to_snapshot_id: core::option::Option<i64>) -> iceberg::scan::IncrementalAppendScanBuilder<'_>
+pub fn iceberg::table::Table::incremental_append_scan_inclusive(&self, from_snapshot_id: i64, to_snapshot_id: core::option::Option<i64>) -> iceberg::scan::IncrementalAppendScanBuilder<'_>
 pub fn iceberg::table::Table::inspect(&self) -> iceberg::inspect::MetadataTable<'_>
 pub fn iceberg::table::Table::manifest_list_reader(&self, snapshot: &iceberg::spec::SnapshotRef) -> iceberg::spec::ManifestListReader
 pub fn iceberg::table::Table::metadata(&self) -> &iceberg::spec::TableMetadata

crates/iceberg/src/scan/context.rs

@@ -33,6 +33,14 @@ use crate::spec::{
 };
 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 {
@@ -48,6 +56,7 @@ pub(crate) struct ManifestFileContext {
     delete_file_index: DeleteFileIndex,
     name_mapping: Option<Arc<NameMapping>>,
     case_sensitive: bool,
+    entry_filter: Option<ManifestEntryFilter>,
     partition_spec: Option<PartitionSpecRef>,
 }
 
@@ -82,12 +91,19 @@ impl ManifestFileContext {
             delete_file_index,
             name_mapping,
             case_sensitive,
+            entry_filter,
             partition_spec,
         } = self;
 
         let manifest = object_cache.get_manifest(&manifest_file).await?;
 
         for manifest_entry in manifest.entries() {
+            if let Some(ref filter) = entry_filter
+                && !filter(manifest_entry)
+            {
+                continue;
+            }
+
             let manifest_entry_context = ManifestEntryContext {
                 // TODO: refactor to avoid the expensive ManifestEntry clone
                 manifest_entry: manifest_entry.clone(),
@@ -149,7 +165,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,
 
@@ -165,6 +180,25 @@ pub(crate) struct PlanContext {
     pub partition_filter_cache: Arc<PartitionFilterCache>,
     pub manifest_evaluator_cache: Arc<ManifestEvaluatorCache>,
     pub expression_evaluator_cache: Arc<ExpressionEvaluatorCache>,
+    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 {
@@ -218,6 +252,12 @@ 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 {
+            if let Some(ref filter) = self.manifest_file_filter
+                && !filter(manifest_file)
+            {
+                continue;
+            }
+
             let tx = if manifest_file.content == ManifestContentType::Deletes {
                 delete_file_tx.clone()
             } else {
@@ -288,6 +328,7 @@ impl PlanContext {
             delete_file_index,
             name_mapping: self.name_mapping.clone(),
             case_sensitive: self.case_sensitive,
+            entry_filter: self.manifest_entry_filter.clone(),
             partition_spec: self
                 .table_metadata
                 .partition_spec_by_id(manifest_file.partition_spec_id)