diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index bd5736bc17..f1a3d958fb 100644 --- a/crates/iceberg/public-api.txt +++ b/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::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 +pub fn iceberg::scan::IncrementalAppendScanBuilder<'a>::select(self, column_names: impl core::iter::traits::collect::IntoIterator) -> 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) -> 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 pub async fn iceberg::table::StaticTable::from_metadata_file(metadata_location: &str, table_ident: iceberg::TableIdent, file_io: iceberg::io::FileIO) -> iceberg::Result +pub fn iceberg::table::StaticTable::incremental_append_scan(&self, from_snapshot_id: i64, to_snapshot_id: core::option::Option) -> iceberg::scan::IncrementalAppendScanBuilder<'_> +pub fn iceberg::table::StaticTable::incremental_append_scan_inclusive(&self, from_snapshot_id: i64, to_snapshot_id: core::option::Option) -> 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) -> iceberg::scan::IncrementalAppendScanBuilder<'_> +pub fn iceberg::table::Table::incremental_append_scan_inclusive(&self, from_snapshot_id: i64, to_snapshot_id: core::option::Option) -> 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 diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs index 8462eb89f4..14fb1f712e 100644 --- a/crates/iceberg/src/scan/context.rs +++ b/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 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 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>, case_sensitive: bool, + entry_filter: Option, partition_spec: Option, } @@ -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, pub manifest_evaluator_cache: Arc, pub expression_evaluator_cache: Arc, + pub manifest_file_filter: Option, + pub manifest_entry_filter: Option, +} + +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) diff --git a/crates/iceberg/src/scan/incremental.rs b/crates/iceberg/src/scan/incremental.rs new file mode 100644 index 0000000000..1db5c6a662 --- /dev/null +++ b/crates/iceberg/src/scan/incremental.rs @@ -0,0 +1,912 @@ +// 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. + +//! 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::{ManifestContentType, ManifestStatus, Operation, TableMetadataRef}; +use crate::table::Table; +use crate::util::available_parallelism; +use crate::util::snapshot::ancestors_between; +use crate::{Error, ErrorKind, Result}; + +/// Represents a validated range of snapshots for incremental scanning. +/// +/// This struct is used to track which snapshot IDs are included in an incremental +/// scan range, allowing efficient filtering of manifest entries. +#[derive(Debug, Clone)] +pub(crate) struct AppendSnapshotSet { + /// Snapshot IDs in the range + snapshot_ids: HashSet, +} + +impl AppendSnapshotSet { + /// Build a snapshot range by walking the snapshot ancestry chain. + /// + /// Validates that `from_snapshot_id` is an ancestor of `to_snapshot_id` and + /// collects all snapshot IDs in between. Also validates that all snapshots + /// in the range have APPEND operations. + /// + /// # Arguments + /// * `table_metadata` - The table metadata containing snapshot information + /// * `from_snapshot_id` - The starting snapshot ID + /// * `to_snapshot_id` - The ending snapshot ID + /// * `from_inclusive` - Whether to include the from_snapshot in the range + pub(crate) fn build( + table_metadata: &TableMetadataRef, + from_snapshot_id: i64, + to_snapshot_id: i64, + from_inclusive: bool, + ) -> Result { + // Determine the exclusive stop point for the ancestry walk. + // For inclusive mode the from-snapshot must exist so we can look up + // its parent. For exclusive mode the snapshot may have been expired + // (the parent pointer on its child still references it), so we only + // need the ID — matching Java's BaseIncrementalScan semantics. + let oldest_exclusive = if from_inclusive { + let from_snapshot = + table_metadata + .snapshot_by_id(from_snapshot_id) + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!("Snapshot {from_snapshot_id} not found"), + ) + })?; + from_snapshot.parent_snapshot_id() + } else { + Some(from_snapshot_id) + }; + + let snapshots: Vec<_> = + ancestors_between(table_metadata, to_snapshot_id, oldest_exclusive).collect(); + + // ancestors_between silently returns the full chain to root if + // oldest_exclusive isn't in the ancestry chain. Detect this: + // if we got snapshots but from_snapshot_id wasn't encountered as + // the stop point, the chain doesn't connect. + if from_snapshot_id == to_snapshot_id { + // Edge case: from == to. In exclusive mode, range is empty. + // In inclusive mode, we should have exactly one snapshot. + if !from_inclusive { + return Ok(Self { + snapshot_ids: HashSet::new(), + }); + } + } else if snapshots.is_empty() { + // to_snapshot_id doesn't exist + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "from_snapshot {from_snapshot_id} is not an ancestor of to_snapshot {to_snapshot_id}", + ), + )); + } else { + // Verify the oldest snapshot in our walk is actually connected + // to from_snapshot_id. The last snapshot's parent (for exclusive) + // or the last snapshot itself (for inclusive) should be from_snapshot_id. + let oldest_collected = snapshots.last().unwrap(); + let connects = if from_inclusive { + oldest_collected.snapshot_id() == from_snapshot_id + } else { + oldest_collected.parent_snapshot_id() == Some(from_snapshot_id) + }; + if !connects { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "from_snapshot {from_snapshot_id} is not an ancestor of to_snapshot {to_snapshot_id}", + ), + )); + } + } + + // Collect only APPEND snapshot IDs, silently skipping non-APPEND + // snapshots (e.g. replace/compaction, overwrite, delete). This matches + // the Java BaseIncrementalAppendScan behavior — only append operations + // contribute new data files to an incremental append scan. + let mut snapshot_ids = HashSet::with_capacity(snapshots.len()); + for snapshot in &snapshots { + if snapshot.summary().operation == Operation::Append { + snapshot_ids.insert(snapshot.snapshot_id()); + } + } + + Ok(Self { snapshot_ids }) + } + + /// 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) -> 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) -> 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. +/// +/// An incremental append scan returns only data files that were added in +/// snapshots between `from_snapshot_id` and the target snapshot. Only +/// snapshots with APPEND operations are supported. +/// +/// This is **not** a CDC, net-changes, or changelog scan: non-append +/// snapshots in the range (overwrite, replace/compaction, delete) are +/// ignored rather than applied as net changes. The scan reads only the rows +/// added by append snapshots in the range, so its output does not represent +/// the full table state at `to_snapshot_id`, nor does it reflect rows deleted +/// or rewritten within the range. In particular, files produced by compaction +/// (`replace`) are skipped, so appended rows are never double-counted against +/// their rewritten copies. +/// +/// Use [`Table::incremental_append_scan`] or +/// [`Table::incremental_append_scan_inclusive`] to create an instance. +pub struct IncrementalAppendScanBuilder<'a> { + table: &'a Table, + from_snapshot_id: i64, + from_inclusive: bool, + to_snapshot_id: Option, + column_names: Option>, + batch_size: Option, + case_sensitive: bool, + filter: Option, + concurrency_limit_data_files: usize, + concurrency_limit_manifest_entries: usize, + concurrency_limit_manifest_files: usize, + row_group_filtering_enabled: bool, + row_selection_enabled: bool, +} + +impl<'a> IncrementalAppendScanBuilder<'a> { + pub(crate) fn new( + table: &'a Table, + from_snapshot_id: i64, + to_snapshot_id: Option, + from_inclusive: bool, + ) -> Self { + let num_cpus = available_parallelism().get(); + + Self { + table, + from_snapshot_id, + from_inclusive, + to_snapshot_id, + column_names: None, + batch_size: None, + case_sensitive: true, + filter: None, + concurrency_limit_data_files: num_cpus, + concurrency_limit_manifest_entries: num_cpus, + concurrency_limit_manifest_files: num_cpus, + row_group_filtering_enabled: true, + row_selection_enabled: false, + } + } + + /// Sets the desired size of batches in the response + /// to something other than the default + pub fn with_batch_size(mut self, batch_size: Option) -> Self { + self.batch_size = batch_size; + self + } + + /// Sets the scan's case sensitivity + pub fn with_case_sensitive(mut self, case_sensitive: bool) -> Self { + self.case_sensitive = case_sensitive; + self + } + + /// Specifies a predicate to use as a filter + pub fn with_filter(mut self, predicate: Predicate) -> Self { + self.filter = Some(predicate.rewrite_not()); + self + } + + /// Select all columns. + pub fn select_all(mut self) -> Self { + self.column_names = None; + self + } + + /// Select empty columns. + pub fn select_empty(mut self) -> Self { + self.column_names = Some(vec![]); + self + } + + /// Select some columns of the table. + pub fn select(mut self, column_names: impl IntoIterator) -> Self { + self.column_names = Some( + column_names + .into_iter() + .map(|item| item.to_string()) + .collect(), + ); + self + } + + /// Sets the concurrency limit for manifest files, manifest entries, and + /// data files for this scan + pub fn with_concurrency_limit(mut self, limit: usize) -> Self { + self.concurrency_limit_manifest_files = limit; + self.concurrency_limit_manifest_entries = limit; + self.concurrency_limit_data_files = limit; + self + } + + /// Sets the data file concurrency limit for this scan + pub fn with_data_file_concurrency_limit(mut self, limit: usize) -> Self { + self.concurrency_limit_data_files = limit; + self + } + + /// Sets the manifest entry concurrency limit for this scan + pub fn with_manifest_entry_concurrency_limit(mut self, limit: usize) -> Self { + self.concurrency_limit_manifest_entries = limit; + self + } + + /// Determines whether to enable row group filtering. + pub fn with_row_group_filtering_enabled(mut self, row_group_filtering_enabled: bool) -> Self { + self.row_group_filtering_enabled = row_group_filtering_enabled; + self + } + + /// Determines whether to enable row selection. + pub fn with_row_selection_enabled(mut self, row_selection_enabled: bool) -> Self { + self.row_selection_enabled = row_selection_enabled; + self + } + + /// Build the incremental append scan. + pub fn build(self) -> Result { + let to_snapshot = match self.to_snapshot_id { + Some(snapshot_id) => self + .table + .metadata() + .snapshot_by_id(snapshot_id) + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!("to_snapshot with id {snapshot_id} not found"), + ) + })? + .clone(), + None => { + let Some(current_snapshot) = self.table.metadata().current_snapshot() else { + return Err(Error::new( + ErrorKind::DataInvalid, + "Cannot perform incremental scan: table has no snapshots", + )); + }; + current_snapshot.clone() + } + }; + + 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 { + table: self.table, + column_names: self.column_names, + batch_size: self.batch_size, + case_sensitive: self.case_sensitive, + filter: self.filter, + concurrency_limit_data_files: self.concurrency_limit_data_files, + concurrency_limit_manifest_entries: self.concurrency_limit_manifest_entries, + concurrency_limit_manifest_files: self.concurrency_limit_manifest_files, + row_group_filtering_enabled: self.row_group_filtering_enabled, + row_selection_enabled: self.row_selection_enabled, + // Project onto the table's current schema (not the to-snapshot's + // schema), matching the Java and PyIceberg implementations. Rows + // written under an older schema within the range are read against + // the current schema, so newer columns become `NULL`. + schema_override: Some(self.table.metadata().current_schema().clone()), + }, + to_snapshot, + Some(append_set.manifest_file_filter()), + Some(append_set.manifest_entry_filter()), + ) + } +} + +#[cfg(test)] +mod tests { + use futures::TryStreamExt; + + use super::AppendSnapshotSet; + use crate::scan::tests::TableTestFixture; + + #[test] + fn test_incremental_scan_invalid_from_snapshot_exclusive() { + let table = TableTestFixture::new().table; + + // Exclusive mode doesn't require from-snapshot to exist, but it must + // be an ancestor of the to-snapshot. 999999999 is not in the ancestry + // chain so this should fail. + let result = table.incremental_append_scan(999999999, None).build(); + + assert!(result.is_err()); + let err = result.unwrap_err(); + assert!( + err.to_string().contains("not an ancestor"), + "Expected ancestry error, got: {err}" + ); + } + + #[test] + fn test_incremental_scan_invalid_from_snapshot_inclusive() { + let table = TableTestFixture::new().table; + + // Inclusive mode requires from-snapshot to exist (we need its parent ID). + let result = table + .incremental_append_scan_inclusive(999999999, None) + .build(); + + assert!(result.is_err()); + let err = result.unwrap_err(); + assert!( + err.to_string().contains("not found"), + "Expected 'not found' error, got: {err}" + ); + } + + #[test] + fn test_incremental_scan_exclusive_from_expired_snapshot() { + // Fixture has S1 (append) -> S2 (append, current). + // Simulate S1 being expired: use S1's ID as from-snapshot in exclusive + // mode even though it wouldn't exist in metadata after expiration. + // Since exclusive mode only needs the ID (not the snapshot object), + // this should succeed — the child (S2) still has parent_snapshot_id = S1. + let table = TableTestFixture::new().table; + + let s1_id = 3051729675574597004_i64; + let s2_id = 3055729675574597004_i64; + + // Verify S2's parent is S1 (simulating the expired-parent scenario) + assert_eq!( + table + .metadata() + .snapshot_by_id(s2_id) + .unwrap() + .parent_snapshot_id(), + Some(s1_id) + ); + + let result = table.incremental_append_scan(s1_id, Some(s2_id)).build(); + + assert!( + result.is_ok(), + "Exclusive scan from an (effectively expired) parent should succeed" + ); + } + + #[test] + fn test_incremental_scan_invalid_to_snapshot() { + let table = TableTestFixture::new().table; + + let result = table + .incremental_append_scan(3051729675574597004, Some(999999999)) + .build(); + + assert!(result.is_err()); + let err = result.unwrap_err(); + assert!(err.to_string().contains("not found")); + } + + #[test] + fn test_incremental_scan_appends_after() { + // Fixture has S1 (append) -> S2 (append, current) + let table = TableTestFixture::new().table; + + let result = table + .incremental_append_scan(3051729675574597004, None) + .build(); + assert!( + result.is_ok(), + "appends_after should succeed when all snapshots are appends" + ); + + let scan = result.unwrap(); + assert!( + scan.plan_context.is_some(), + "Incremental scan should have a plan context" + ); + } + + #[test] + fn test_incremental_scan_appends_between() { + // Fixture has S1 (append) -> S2 (append, current) + let table = TableTestFixture::new().table; + + let current_snapshot_id = table.metadata().current_snapshot().unwrap().snapshot_id(); + let parent_id = table + .metadata() + .current_snapshot() + .unwrap() + .parent_snapshot_id() + .expect("Current snapshot should have a parent"); + + let result = table + .incremental_append_scan(parent_id, Some(current_snapshot_id)) + .build(); + + assert!( + result.is_ok(), + "appends_between should succeed for two append snapshots" + ); + } + + #[test] + fn test_incremental_scan_from_snapshot_inclusive() { + // Fixture has S1 (append) -> S2 (append, current) + 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, Some(current_snapshot_id)) + .build(); + assert!( + result.is_ok(), + "Inclusive scan of a single append snapshot should succeed" + ); + + // Verify AppendSnapshotSet directly + let set = AppendSnapshotSet::build( + &table.metadata_ref(), + current_snapshot_id, + current_snapshot_id, + true, + ) + .unwrap(); + assert!( + set.contains(current_snapshot_id), + "Inclusive set should contain the from_snapshot" + ); + } + + #[test] + fn test_incremental_scan_from_snapshot_exclusive() { + // Fixture has S1 (append) -> S2 (append, current) + 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, Some(current_snapshot_id)) + .build(); + assert!( + result.is_ok(), + "Exclusive scan from=to should succeed with empty range" + ); + + // Verify AppendSnapshotSet directly + let set = AppendSnapshotSet::build( + &table.metadata_ref(), + current_snapshot_id, + current_snapshot_id, + false, + ) + .unwrap(); + assert!( + !set.contains(current_snapshot_id), + "Exclusive set should not contain the from_snapshot" + ); + } + + #[test] + fn test_incremental_scan_skips_non_append_operations() { + // Deep history fixture: S1 (append) -> S2 (append) -> S3 (append) + // -> S4 (overwrite) -> S5 (append, current) + let table = TableTestFixture::new_with_deep_history().table; + + // Scanning from S1 to S5 crosses S4 (overwrite) — should succeed + // but only include APPEND snapshots (S2, S3, S5), skipping S4 + let result = table + .incremental_append_scan(3051729675574597004, Some(3059729675574597004)) + .build(); + + assert!( + result.is_ok(), + "Should succeed, skipping non-APPEND snapshots" + ); + + let set = AppendSnapshotSet::build( + &table.metadata_ref(), + 3051729675574597004, + 3059729675574597004, + false, + ) + .unwrap(); + assert!( + !set.contains(3051729675574597004), + "S1 (from) should be excluded" + ); + assert!( + set.contains(3055729675574597004), + "S2 (append) should be in set" + ); + assert!( + set.contains(3056729675574597004), + "S3 (append) should be in set" + ); + assert!( + !set.contains(3057729675574597004), + "S4 (overwrite) should be skipped" + ); + assert!( + set.contains(3059729675574597004), + "S5 (append) should be in set" + ); + } + + #[test] + fn test_incremental_scan_append_only_range() { + // Deep history fixture: S1 (append) -> S2 (append) -> S3 (append) + // -> S4 (overwrite) -> S5 (append, current) + let table = TableTestFixture::new_with_deep_history().table; + + // Scanning from S1 to S3 (all appends) + let set = AppendSnapshotSet::build( + &table.metadata_ref(), + 3051729675574597004, + 3056729675574597004, + false, + ) + .unwrap(); + assert!( + !set.contains(3051729675574597004), + "from_snapshot should be excluded" + ); + assert!(set.contains(3055729675574597004), "S2 should be in range"); + assert!(set.contains(3056729675574597004), "S3 should be in range"); + } + + #[tokio::test] + async fn test_incremental_scan_returns_only_added_files_in_range() { + // Fixture has S1 (append) -> S2 (append, current) + // Manifest contains: + // 1.parquet: status=Added, snapshot=S2 + // 2.parquet: status=Deleted, snapshot=S1 + // 3.parquet: status=Existing, snapshot=S1 + let mut fixture = TableTestFixture::new(); + fixture.setup_manifest_files().await; + + let current_snapshot = fixture.table.metadata().current_snapshot().unwrap(); + let parent_snapshot_id = current_snapshot.parent_snapshot_id().unwrap(); + + // Incremental scan from S1 (exclusive) to S2 should return only 1.parquet + let table_scan = fixture + .table + .incremental_append_scan(parent_snapshot_id, Some(current_snapshot.snapshot_id())) + .build() + .unwrap(); + + let tasks: Vec<_> = table_scan + .plan_files() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + + assert_eq!( + tasks.len(), + 1, + "Incremental scan should return exactly 1 file" + ); + assert_eq!( + tasks[0].data_file_path, + format!("{}/1.parquet", &fixture.table_location), + "Should only return the file added in S2" + ); + } + + #[tokio::test] + async fn test_incremental_scan_exclusive_same_snapshot_returns_empty() { + // Fixture has S1 (append) -> S2 (append, current) + let mut fixture = TableTestFixture::new(); + fixture.setup_manifest_files().await; + + let current_snapshot_id = fixture + .table + .metadata() + .current_snapshot() + .unwrap() + .snapshot_id(); + + // Incremental scan from S2 to S2 (exclusive) should return nothing + let table_scan = fixture + .table + .incremental_append_scan(current_snapshot_id, Some(current_snapshot_id)) + .build() + .unwrap(); + + let tasks: Vec<_> = table_scan + .plan_files() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + + assert!( + tasks.is_empty(), + "Exclusive scan from=to should return no files" + ); + } + + #[tokio::test] + async fn test_incremental_scan_compaction_not_double_counted() { + // Compaction (`rewrite_data_files`) commits a `replace` snapshot whose + // rewritten file re-adds rows that were already appended earlier. An + // incremental append scan must skip that file so the appended rows are + // read exactly once — never double-counted against the rewritten copy. + // + // Deep history fixture with S4 relabeled as a `replace` (compaction): + // S1 (append) -> S2 (append) -> S3 (append) -> S4 (replace) -> S5 (append, current) + // Scanning from S1 (exclusive) to S5 should return only files from + // APPEND snapshots: s2.parquet, s3.parquet, s5.parquet. The compacted + // s4.parquet must be skipped. + let mut fixture = TableTestFixture::new_with_deep_history_compaction(); + fixture.setup_manifest_files_deep_history().await; + + let s1_id = 3051729675574597004_i64; + let s5_id = 3059729675574597004_i64; + + let table_scan = fixture + .table + .incremental_append_scan(s1_id, Some(s5_id)) + .build() + .unwrap(); + + let mut tasks: Vec<_> = table_scan + .plan_files() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + + tasks.sort_by(|a, b| a.data_file_path.cmp(&b.data_file_path)); + + let file_names: Vec<&str> = tasks + .iter() + .map(|t| { + t.data_file_path + .rsplit('/') + .next() + .unwrap_or(&t.data_file_path) + }) + .collect(); + + assert_eq!( + file_names, + vec!["s2.parquet", "s3.parquet", "s5.parquet"], + "Compacted file (s4, from the replace snapshot) must be skipped" + ); + } + + #[tokio::test] + async fn test_incremental_scan_deep_history_skips_overwrite_files() { + // Deep history fixture: + // S1 (append) -> S2 (append) -> S3 (append) -> S4 (overwrite) -> S5 (append, current) + // Each snapshot adds one file: s1.parquet .. s5.parquet + // + // Incremental scan from S1 (exclusive) to S5 should return only files + // from APPEND snapshots: s2.parquet, s3.parquet, s5.parquet + // s4.parquet (added in overwrite S4) must be skipped. + let mut fixture = TableTestFixture::new_with_deep_history(); + fixture.setup_manifest_files_deep_history().await; + + let s1_id = 3051729675574597004_i64; + let s5_id = 3059729675574597004_i64; + + let table_scan = fixture + .table + .incremental_append_scan(s1_id, Some(s5_id)) + .build() + .unwrap(); + + let mut tasks: Vec<_> = table_scan + .plan_files() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + + // Sort by path for deterministic assertions + tasks.sort_by(|a, b| a.data_file_path.cmp(&b.data_file_path)); + + assert_eq!( + tasks.len(), + 3, + "Should return 3 files (s2, s3, s5), skipping s4 (overwrite)" + ); + + let file_names: Vec<&str> = tasks + .iter() + .map(|t| { + t.data_file_path + .rsplit('/') + .next() + .unwrap_or(&t.data_file_path) + }) + .collect(); + + assert_eq!( + file_names, + vec!["s2.parquet", "s3.parquet", "s5.parquet"], + "Only files from APPEND snapshots should be returned" + ); + } + + #[tokio::test] + async fn test_incremental_scan_deep_history_partial_range() { + // Scan from S2 (exclusive) to S3 — both appends, should return only s3.parquet + let mut fixture = TableTestFixture::new_with_deep_history(); + fixture.setup_manifest_files_deep_history().await; + + let s2_id = 3055729675574597004_i64; + let s3_id = 3056729675574597004_i64; + + let table_scan = fixture + .table + .incremental_append_scan(s2_id, Some(s3_id)) + .build() + .unwrap(); + + let tasks: Vec<_> = table_scan + .plan_files() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + + assert_eq!(tasks.len(), 1, "Should return exactly 1 file"); + assert!( + tasks[0].data_file_path.ends_with("s3.parquet"), + "Should return s3.parquet, got: {}", + tasks[0].data_file_path + ); + } + + #[test] + fn test_incremental_scan_projects_onto_current_schema() { + // The table's current schema (id 1) has three columns (x, y, z), but + // every snapshot references the older schema (id 0) with a single + // column (x). An incremental scan must project onto the *current* + // schema, matching the Java and PyIceberg implementations, so rows + // written under the older schema get NULLs for the newer columns. + let table = TableTestFixture::new_with_deep_history_stale_schema().table; + + let s1_id = 3051729675574597004_i64; + let s5_id = 3059729675574597004_i64; + + let scan = table + .incremental_append_scan(s1_id, Some(s5_id)) + .build() + .unwrap(); + + let plan_context = scan + .plan_context + .as_ref() + .expect("incremental scan should have a plan context"); + + // The scan must use the current schema (3 columns), not the + // to-snapshot's schema (1 column). + let current_schema = table.metadata().current_schema(); + assert_eq!( + plan_context.snapshot_schema.schema_id(), + current_schema.schema_id(), + "incremental scan should project onto the current schema" + ); + assert_eq!( + plan_context.snapshot_schema.as_struct().fields().len(), + 3, + "current schema has three columns (x, y, z)" + ); + + // Sanity check: the to-snapshot itself references the older schema. + let to_snapshot = table.metadata().snapshot_by_id(s5_id).unwrap(); + assert_eq!( + to_snapshot.schema(table.metadata()).unwrap().schema_id(), + 0, + "to-snapshot should reference the older single-column schema" + ); + } + + #[tokio::test] + async fn test_incremental_scan_deep_history_inclusive_with_overwrite() { + // Inclusive scan from S3 to S5: + // S3 (append) -> S4 (overwrite) -> S5 (append) + // Should return s3.parquet and s5.parquet, skipping s4.parquet + let mut fixture = TableTestFixture::new_with_deep_history(); + fixture.setup_manifest_files_deep_history().await; + + let s3_id = 3056729675574597004_i64; + let s5_id = 3059729675574597004_i64; + + let table_scan = fixture + .table + .incremental_append_scan_inclusive(s3_id, Some(s5_id)) + .build() + .unwrap(); + + let mut tasks: Vec<_> = table_scan + .plan_files() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + + tasks.sort_by(|a, b| a.data_file_path.cmp(&b.data_file_path)); + + assert_eq!( + tasks.len(), + 2, + "Should return 2 files (s3, s5), skipping s4 (overwrite)" + ); + + let file_names: Vec<&str> = tasks + .iter() + .map(|t| { + t.data_file_path + .rsplit('/') + .next() + .unwrap_or(&t.data_file_path) + }) + .collect(); + + assert_eq!( + file_names, + vec!["s3.parquet", "s5.parquet"], + "Only files from APPEND snapshots should be returned" + ); + } +} diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index b377493729..02436febfe 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -21,6 +21,7 @@ mod cache; use cache::*; mod context; use context::*; +mod incremental; mod task; use std::sync::Arc; @@ -29,6 +30,7 @@ use arrow_array::RecordBatch; use futures::channel::mpsc::{Sender, channel}; use futures::stream::BoxStream; use futures::{SinkExt, StreamExt, TryStreamExt}; +pub use incremental::IncrementalAppendScanBuilder; pub use task::*; use crate::arrow::ArrowReaderBuilder; @@ -39,7 +41,9 @@ use crate::expr::{Bind, BoundPredicate, Predicate}; use crate::io::FileIO; use crate::metadata_columns::{get_metadata_field_id, is_metadata_column_name}; use crate::runtime::Runtime; -use crate::spec::{DEFAULT_SCHEMA_NAME_MAPPING, DataContentType, NameMapping, SnapshotRef}; +use crate::spec::{ + DEFAULT_SCHEMA_NAME_MAPPING, DataContentType, NameMapping, SchemaRef, SnapshotRef, +}; use crate::table::Table; use crate::util::available_parallelism; use crate::{Error, ErrorKind, Result}; @@ -47,6 +51,150 @@ use crate::{Error, ErrorKind, Result}; /// A stream of arrow [`RecordBatch`]es. pub type ArrowRecordBatchStream = BoxStream<'static, Result>; +/// Shared configuration extracted from scan builders, used by both +/// [`TableScanBuilder`] and [`IncrementalAppendScanBuilder`]. +pub(crate) struct ScanConfig<'a> { + table: &'a Table, + column_names: Option>, + batch_size: Option, + case_sensitive: bool, + filter: Option, + concurrency_limit_data_files: usize, + concurrency_limit_manifest_entries: usize, + concurrency_limit_manifest_files: usize, + row_group_filtering_enabled: bool, + row_selection_enabled: bool, + /// Schema to project the scan onto. When `None`, the schema attached to + /// `snapshot` is used (the correct behavior for time-travel scans). An + /// incremental scan sets this to the table's current schema so that rows + /// written under an older schema in the range are projected onto the + /// current schema (newer columns become `NULL`), matching the Java and + /// PyIceberg implementations. + schema_override: Option, +} + +/// Shared build logic: validates columns, resolves field IDs, binds predicates, +/// and constructs [`PlanContext`] + [`TableScan`]. +pub(crate) fn build_table_scan( + config: ScanConfig<'_>, + snapshot: SnapshotRef, + manifest_file_filter: Option, + manifest_entry_filter: Option, +) -> Result { + let schema = match config.schema_override.clone() { + Some(schema) => schema, + None => snapshot.schema(config.table.metadata())?, + }; + + // Check that all column names exist in the schema (skip reserved columns). + if let Some(column_names) = config.column_names.as_ref() { + for column_name in column_names { + if is_metadata_column_name(column_name) { + continue; + } + if schema.field_by_name(column_name).is_none() { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("Column {column_name} not found in table. Schema: {schema}"), + )); + } + } + } + + let mut field_ids = vec![]; + let column_names = config.column_names.clone().unwrap_or_else(|| { + schema + .as_struct() + .fields() + .iter() + .map(|f| f.name.clone()) + .collect() + }); + + for column_name in column_names.iter() { + if is_metadata_column_name(column_name) { + field_ids.push(get_metadata_field_id(column_name)?); + continue; + } + + let field_id = schema.field_id_by_name(column_name).ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!("Column {column_name} not found in table. Schema: {schema}"), + ) + })?; + + schema + .as_struct() + .field_by_id(field_id) + .ok_or_else(|| { + Error::new( + ErrorKind::FeatureUnsupported, + format!( + "Column {column_name} is not a direct child of schema but a nested field, which is not supported now. Schema: {schema}" + ), + ) + })?; + + field_ids.push(field_id); + } + + let snapshot_bound_predicate = if let Some(ref predicates) = config.filter { + Some(predicates.bind(schema.clone(), true)?) + } else { + None + }; + + let name_mapping = config + .table + .metadata() + .properties() + .get(DEFAULT_SCHEMA_NAME_MAPPING) + .map(|raw| { + serde_json::from_str::(raw).map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Failed to parse table property {DEFAULT_SCHEMA_NAME_MAPPING} as a NameMapping" + ), + ) + .with_source(e) + }) + }) + .transpose()? + .map(Arc::new); + + let plan_context = PlanContext { + snapshot, + table_metadata: config.table.metadata_ref(), + snapshot_schema: schema, + case_sensitive: config.case_sensitive, + predicate: config.filter.map(Arc::new), + snapshot_bound_predicate: snapshot_bound_predicate.map(Arc::new), + object_cache: config.table.object_cache(), + field_ids: Arc::new(field_ids), + name_mapping, + partition_filter_cache: Arc::new(PartitionFilterCache::new()), + manifest_evaluator_cache: Arc::new(ManifestEvaluatorCache::new()), + expression_evaluator_cache: Arc::new(ExpressionEvaluatorCache::new()), + manifest_file_filter, + manifest_entry_filter, + }; + + Ok(TableScan { + batch_size: config.batch_size, + column_names: config.column_names, + file_io: config.table.file_io().clone(), + plan_context: Some(plan_context), + concurrency_limit_data_files: config.concurrency_limit_data_files, + concurrency_limit_manifest_entries: config.concurrency_limit_manifest_entries, + concurrency_limit_manifest_files: config.concurrency_limit_manifest_files, + row_group_filtering_enabled: config.row_group_filtering_enabled, + row_selection_enabled: config.row_selection_enabled, + runtime: config.table.runtime().clone(), + }) +} + /// Builder to create table scan. pub struct TableScanBuilder<'a> { table: &'a Table, @@ -200,7 +348,7 @@ impl<'a> TableScanBuilder<'a> { })? .clone(), None => { - let Some(current_snapshot_id) = self.table.metadata().current_snapshot() else { + let Some(current_snapshot) = self.table.metadata().current_snapshot() else { return Ok(TableScan { batch_size: self.batch_size, column_names: self.column_names, @@ -214,119 +362,28 @@ impl<'a> TableScanBuilder<'a> { runtime: self.table.runtime().clone(), }); }; - current_snapshot_id.clone() + current_snapshot.clone() } }; - let schema = snapshot.schema(self.table.metadata())?; - - // Check that all column names exist in the schema (skip reserved columns). - if let Some(column_names) = self.column_names.as_ref() { - for column_name in column_names { - // Skip reserved columns that don't exist in the schema - if is_metadata_column_name(column_name) { - continue; - } - if schema.field_by_name(column_name).is_none() { - return Err(Error::new( - ErrorKind::DataInvalid, - format!("Column {column_name} not found in table. Schema: {schema}"), - )); - } - } - } - - let mut field_ids = vec![]; - let column_names = self.column_names.clone().unwrap_or_else(|| { - schema - .as_struct() - .fields() - .iter() - .map(|f| f.name.clone()) - .collect() - }); - - for column_name in column_names.iter() { - // Handle metadata columns (like "_file") - if is_metadata_column_name(column_name) { - field_ids.push(get_metadata_field_id(column_name)?); - continue; - } - - let field_id = schema.field_id_by_name(column_name).ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - format!("Column {column_name} not found in table. Schema: {schema}"), - ) - })?; - - schema - .as_struct() - .field_by_id(field_id) - .ok_or_else(|| { - Error::new( - ErrorKind::FeatureUnsupported, - format!( - "Column {column_name} is not a direct child of schema but a nested field, which is not supported now. Schema: {schema}" - ), - ) - })?; - - field_ids.push(field_id); - } - - let snapshot_bound_predicate = if let Some(ref predicates) = self.filter { - Some(predicates.bind(schema.clone(), true)?) - } else { - None - }; - - let name_mapping = self - .table - .metadata() - .properties() - .get(DEFAULT_SCHEMA_NAME_MAPPING) - .map(|raw| { - serde_json::from_str::(raw).map_err(|e| { - Error::new( - ErrorKind::DataInvalid, - format!( - "Failed to parse table property {DEFAULT_SCHEMA_NAME_MAPPING} as a NameMapping" - ), - ) - .with_source(e) - }) - }) - .transpose()? - .map(Arc::new); - - let plan_context = PlanContext { + build_table_scan( + ScanConfig { + table: self.table, + column_names: self.column_names, + batch_size: self.batch_size, + case_sensitive: self.case_sensitive, + filter: self.filter, + concurrency_limit_data_files: self.concurrency_limit_data_files, + concurrency_limit_manifest_entries: self.concurrency_limit_manifest_entries, + concurrency_limit_manifest_files: self.concurrency_limit_manifest_files, + row_group_filtering_enabled: self.row_group_filtering_enabled, + row_selection_enabled: self.row_selection_enabled, + schema_override: None, + }, snapshot, - table_metadata: self.table.metadata_ref(), - snapshot_schema: schema, - case_sensitive: self.case_sensitive, - predicate: self.filter.map(Arc::new), - snapshot_bound_predicate: snapshot_bound_predicate.map(Arc::new), - object_cache: self.table.object_cache(), - field_ids: Arc::new(field_ids), - name_mapping, - partition_filter_cache: Arc::new(PartitionFilterCache::new()), - manifest_evaluator_cache: Arc::new(ManifestEvaluatorCache::new()), - expression_evaluator_cache: Arc::new(ExpressionEvaluatorCache::new()), - }; - - Ok(TableScan { - batch_size: self.batch_size, - column_names: self.column_names, - file_io: self.table.file_io().clone(), - plan_context: Some(plan_context), - concurrency_limit_data_files: self.concurrency_limit_data_files, - concurrency_limit_manifest_entries: self.concurrency_limit_manifest_entries, - concurrency_limit_manifest_files: self.concurrency_limit_manifest_files, - row_group_filtering_enabled: self.row_group_filtering_enabled, - row_selection_enabled: self.row_selection_enabled, - runtime: self.table.runtime().clone(), - }) + None, + None, + ) } } @@ -646,8 +703,9 @@ pub mod tests { use crate::scan::FileScanTask; use crate::spec::{ DEFAULT_SCHEMA_NAME_MAPPING, DataContentType, DataFileBuilder, DataFileFormat, Datum, - Literal, ManifestEntry, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, - NestedField, PartitionSpec, PrimitiveType, Schema, Struct, StructType, TableMetadata, Type, + Literal, ManifestEntry, ManifestFile, ManifestListWriter, ManifestStatus, + ManifestWriterBuilder, NestedField, PartitionSpec, PrimitiveType, Schema, Struct, + StructType, TableMetadata, Type, }; use crate::table::Table; use crate::test_utils::test_runtime; @@ -742,22 +800,72 @@ pub mod tests { } /// Creates a fixture with 5 snapshots chained as: - /// S1 (root) -> S2 -> S3 -> S4 -> S5 (current) - /// Useful for testing snapshot history traversal. + /// S1 (append) -> S2 (append) -> S3 (append) -> S4 (overwrite) -> S5 (append, current) + /// Useful for testing snapshot history traversal and incremental scans + /// with non-append operations in the chain. pub fn new_with_deep_history() -> Self { + Self::new_from_deep_history_metadata("example_table_metadata_v2_deep_history.json") + } + + /// Like [`Self::new_with_deep_history`] but every snapshot references + /// the older single-column schema (`schema-id` 0) while the table's + /// `current-schema-id` stays at the three-column schema (`schema-id` + /// 1). This models a table whose schema evolved *after* the snapshots + /// in an incremental range were written, so we can assert that an + /// incremental scan projects onto the current schema. + pub fn new_with_deep_history_stale_schema() -> Self { + let fixture = Self::new_from_deep_history_metadata( + "example_table_metadata_v2_deep_history_stale_schema.json", + ); + + // Sanity check: current schema (3 cols) differs from the schema the + // snapshots reference (1 col), otherwise the test would be vacuous. + assert_eq!(fixture.table.metadata().current_schema_id(), 1); + fixture + } + + /// Like [`Self::new_with_deep_history`] but the S4 snapshot is a + /// `replace` (the operation a compaction / `rewrite_data_files` + /// commits) rather than an `overwrite`. Used to prove that an + /// incremental append scan skips compaction output and never + /// double-counts the appended rows against their rewritten copies. + pub fn new_with_deep_history_compaction() -> Self { + Self::new_from_deep_history_metadata( + "example_table_metadata_v2_deep_history_compaction.json", + ) + } + + /// Builds a deep-history fixture from the named templated metadata file + /// in `testdata`. The five snapshot manifest-list paths are rendered to + /// point at this fixture's temp directory. + fn new_from_deep_history_metadata(metadata_file: &str) -> Self { let tmp_dir = TempDir::new().unwrap(); let table_location = tmp_dir.path().join("table1"); let table_metadata1_location = table_location.join("metadata/v1.json"); + let manifest_list_s1 = table_location.join("metadata/snap-3051729675574597004.avro"); + let manifest_list_s2 = table_location.join("metadata/snap-3055729675574597004.avro"); + let manifest_list_s3 = table_location.join("metadata/snap-3056729675574597004.avro"); + let manifest_list_s4 = table_location.join("metadata/snap-3057729675574597004.avro"); + let manifest_list_s5 = table_location.join("metadata/snap-3059729675574597004.avro"); + let file_io = FileIO::new_with_fs(); let table_metadata = { - let json_str = fs::read_to_string(format!( - "{}/testdata/example_table_metadata_v2_deep_history.json", + let template_json_str = fs::read_to_string(format!( + "{}/testdata/{metadata_file}", env!("CARGO_MANIFEST_DIR") )) .unwrap(); - serde_json::from_str::(&json_str).unwrap() + let metadata_json = render_template(&template_json_str, context! { + table_location => &table_location, + manifest_list_s1_location => &manifest_list_s1, + manifest_list_s2_location => &manifest_list_s2, + manifest_list_s3_location => &manifest_list_s3, + manifest_list_s4_location => &manifest_list_s4, + manifest_list_s5_location => &manifest_list_s5, + }); + serde_json::from_str::(&metadata_json).unwrap() }; let table = Table::builder() @@ -971,6 +1079,151 @@ pub mod tests { manifest_list_write.close().await.unwrap(); } + /// Sets up manifest files for the deep history fixture. + /// + /// Creates one data file per snapshot (s1.parquet through s5.parquet), + /// each with a manifest and manifest list. Manifest lists are cumulative + /// (each snapshot's list includes all prior manifests), matching real + /// Iceberg behavior. The incremental scan should skip s4.parquet + /// (added in the overwrite snapshot S4). + pub async fn setup_manifest_files_deep_history(&mut self) { + let parquet_file_size = self.write_parquet_data_files_deep_history(); + let partition_spec = self.table.metadata().default_partition_spec(); + + // Snapshot chain: S1 -> S2 -> S3 -> S4 (overwrite) -> S5 + let snapshot_ids: Vec = vec![ + 3051729675574597004, + 3055729675574597004, + 3056729675574597004, + 3057729675574597004, + 3059729675574597004, + ]; + + // Accumulate manifests across snapshots (each manifest list is cumulative) + let mut all_manifests: Vec = Vec::new(); + + for (i, &snap_id) in snapshot_ids.iter().enumerate() { + let snapshot = self + .table + .metadata() + .snapshot_by_id(snap_id) + .unwrap() + .clone(); + let schema = snapshot.schema(self.table.metadata()).unwrap(); + + let file_name = format!("s{}.parquet", i + 1); + let partition_value = (i + 1) as i64 * 100; + + let mut writer = ManifestWriterBuilder::new( + self.next_manifest_file(), + Some(snap_id), + schema, + partition_spec.as_ref().clone(), + ) + .build_v2_data(); + + writer + .add_entry( + ManifestEntry::builder() + .status(ManifestStatus::Added) + .data_file( + DataFileBuilder::default() + .partition_spec_id(0) + .content(DataContentType::Data) + .file_path(format!("{}/{}", &self.table_location, file_name)) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(parquet_file_size) + .record_count(1) + .partition(Struct::from_iter([Some(Literal::long( + partition_value, + ))])) + .key_metadata(None) + .build() + .unwrap(), + ) + .build(), + ) + .unwrap(); + + let mut data_file_manifest = writer.write_manifest_file().await.unwrap(); + // Assign sequence numbers so the manifest can be included in + // later snapshots' cumulative manifest lists without triggering + // the "unassigned sequence number" validation. + data_file_manifest.sequence_number = snapshot.sequence_number(); + data_file_manifest.min_sequence_number = snapshot.sequence_number(); + all_manifests.push(data_file_manifest); + + // Write cumulative manifest list for this snapshot + let manifest_list_writer = self + .table + .file_io() + .new_output(snapshot.manifest_list()) + .unwrap() + .writer() + .await + .unwrap(); + let mut manifest_list_write = ManifestListWriter::v2( + manifest_list_writer, + snap_id, + snapshot.parent_snapshot_id(), + snapshot.sequence_number(), + ); + manifest_list_write + .add_manifests(all_manifests.clone().into_iter()) + .unwrap(); + manifest_list_write.close().await.unwrap(); + } + } + + /// Writes parquet data files for the deep history fixture (3-column schema: x, y, z). + fn write_parquet_data_files_deep_history(&self) -> u64 { + fs::create_dir_all(&self.table_location).unwrap(); + + let schema = { + let fields = vec![ + arrow_schema::Field::new("x", arrow_schema::DataType::Int64, false) + .with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "1".to_string(), + )])), + arrow_schema::Field::new("y", arrow_schema::DataType::Int64, false) + .with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "2".to_string(), + )])), + arrow_schema::Field::new("z", arrow_schema::DataType::Int64, false) + .with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "3".to_string(), + )])), + ]; + Arc::new(arrow_schema::Schema::new(fields)) + }; + + let col1 = Arc::new(Int64Array::from_iter_values(vec![1; 10])) as ArrayRef; + let col2 = Arc::new(Int64Array::from_iter_values(vec![2; 10])) as ArrayRef; + let col3 = Arc::new(Int64Array::from_iter_values(vec![3; 10])) as ArrayRef; + + let batch = RecordBatch::try_new(schema.clone(), vec![col1, col2, col3]).unwrap(); + + let props = WriterProperties::builder() + .set_compression(Compression::SNAPPY) + .build(); + + for i in 1..=5 { + let file = + File::create(format!("{}/s{}.parquet", &self.table_location, i)).unwrap(); + let mut writer = + ArrowWriter::try_new(file, batch.schema(), Some(props.clone())).unwrap(); + writer.write(&batch).expect("Writing batch"); + writer.close().unwrap(); + } + + fs::metadata(format!("{}/s1.parquet", &self.table_location)) + .unwrap() + .len() + } + pub async fn setup_manifest_files_with_partition_evolution(&mut self) { let current_snapshot = self.table.metadata().current_snapshot().unwrap(); let parent_snapshot = current_snapshot diff --git a/crates/iceberg/src/table.rs b/crates/iceberg/src/table.rs index 31feade038..734046b4f7 100644 --- a/crates/iceberg/src/table.rs +++ b/crates/iceberg/src/table.rs @@ -26,7 +26,7 @@ use crate::inspect::MetadataTable; use crate::io::FileIO; use crate::io::object_cache::ObjectCache; use crate::runtime::Runtime; -use crate::scan::TableScanBuilder; +use crate::scan::{IncrementalAppendScanBuilder, TableScanBuilder}; use crate::spec::{ManifestListReader, SchemaRef, SnapshotRef, TableMetadata, TableMetadataRef}; use crate::{Error, ErrorKind, Result, TableIdent}; @@ -280,6 +280,30 @@ impl Table { TableScanBuilder::new(self) } + /// Creates an incremental append scan starting from the given snapshot (exclusive). + /// + /// Returns only data files added in APPEND snapshots after `from_snapshot_id`, + /// up to `to_snapshot_id` or the current snapshot if `None`. + pub fn incremental_append_scan( + &self, + from_snapshot_id: i64, + to_snapshot_id: Option, + ) -> IncrementalAppendScanBuilder<'_> { + IncrementalAppendScanBuilder::new(self, from_snapshot_id, to_snapshot_id, false) + } + + /// Creates an incremental append scan starting from the given snapshot (inclusive). + /// + /// Returns only data files added in APPEND snapshots from `from_snapshot_id` (inclusive), + /// up to `to_snapshot_id` or the current snapshot if `None`. + pub fn incremental_append_scan_inclusive( + &self, + from_snapshot_id: i64, + to_snapshot_id: Option, + ) -> IncrementalAppendScanBuilder<'_> { + IncrementalAppendScanBuilder::new(self, from_snapshot_id, to_snapshot_id, true) + } + /// Creates a metadata table which provides table-like APIs for inspecting metadata. /// See [`MetadataTable`] for more details. pub fn inspect(&self) -> MetadataTable<'_> { @@ -385,6 +409,26 @@ impl StaticTable { self.0.scan() } + /// Creates an incremental append scan starting from the given snapshot (exclusive). + pub fn incremental_append_scan( + &self, + from_snapshot_id: i64, + to_snapshot_id: Option, + ) -> IncrementalAppendScanBuilder<'_> { + self.0 + .incremental_append_scan(from_snapshot_id, to_snapshot_id) + } + + /// Creates an incremental append scan starting from the given snapshot (inclusive). + pub fn incremental_append_scan_inclusive( + &self, + from_snapshot_id: i64, + to_snapshot_id: Option, + ) -> IncrementalAppendScanBuilder<'_> { + self.0 + .incremental_append_scan_inclusive(from_snapshot_id, to_snapshot_id) + } + /// Get TableMetadataRef for the static table pub fn metadata(&self) -> TableMetadataRef { self.0.metadata_ref() diff --git a/crates/iceberg/src/util/snapshot.rs b/crates/iceberg/src/util/snapshot.rs index a96fd97831..b6fd518290 100644 --- a/crates/iceberg/src/util/snapshot.rs +++ b/crates/iceberg/src/util/snapshot.rs @@ -52,7 +52,9 @@ pub fn ancestors_of( /// Note: if `oldest_snapshot_id` is `Some(id)` but `id` is not actually an /// ancestor of `latest_snapshot_id`, the walk is never stopped and this yields /// *all* ancestors of `latest_snapshot_id` down to the root. Callers that treat -/// `oldest_snapshot_id` as a lower bound must validate the lineage themselves. +/// `oldest_snapshot_id` as a lower bound must validate the lineage themselves +/// (e.g. `AppendSnapshotSet::build` checks that the walk actually connects to +/// `from_snapshot_id`). pub fn ancestors_between( table_metadata: &TableMetadataRef, latest_snapshot_id: i64, diff --git a/crates/iceberg/testdata/example_table_metadata_v2_deep_history.json b/crates/iceberg/testdata/example_table_metadata_v2_deep_history.json index a354958697..bd192ca6e2 100644 --- a/crates/iceberg/testdata/example_table_metadata_v2_deep_history.json +++ b/crates/iceberg/testdata/example_table_metadata_v2_deep_history.json @@ -1,7 +1,7 @@ { "format-version": 2, "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1", - "location": "s3://bucket/test/location", + "location": "{{ table_location }}", "last-sequence-number": 34, "last-updated-ms": 1602638573590, "last-column-id": 3, @@ -53,7 +53,7 @@ "timestamp-ms": 1515100955770, "sequence-number": 0, "summary": {"operation": "append"}, - "manifest-list": "s3://bucket/metadata/snap-3051729675574597004.avro" + "manifest-list": "{{ manifest_list_s1_location }}" }, { "snapshot-id": 3055729675574597004, @@ -61,7 +61,7 @@ "timestamp-ms": 1555100955770, "sequence-number": 1, "summary": {"operation": "append"}, - "manifest-list": "s3://bucket/metadata/snap-3055729675574597004.avro", + "manifest-list": "{{ manifest_list_s2_location }}", "schema-id": 1 }, { @@ -70,7 +70,7 @@ "timestamp-ms": 1575100955770, "sequence-number": 2, "summary": {"operation": "append"}, - "manifest-list": "s3://bucket/metadata/snap-3056729675574597004.avro", + "manifest-list": "{{ manifest_list_s3_location }}", "schema-id": 1 }, { @@ -79,7 +79,7 @@ "timestamp-ms": 1595100955770, "sequence-number": 3, "summary": {"operation": "overwrite"}, - "manifest-list": "s3://bucket/metadata/snap-3057729675574597004.avro", + "manifest-list": "{{ manifest_list_s4_location }}", "schema-id": 1 }, { @@ -88,7 +88,7 @@ "timestamp-ms": 1602638573590, "sequence-number": 4, "summary": {"operation": "append"}, - "manifest-list": "s3://bucket/metadata/snap-3059729675574597004.avro", + "manifest-list": "{{ manifest_list_s5_location }}", "schema-id": 1 } ], diff --git a/crates/iceberg/testdata/example_table_metadata_v2_deep_history_compaction.json b/crates/iceberg/testdata/example_table_metadata_v2_deep_history_compaction.json new file mode 100644 index 0000000000..35f667caf3 --- /dev/null +++ b/crates/iceberg/testdata/example_table_metadata_v2_deep_history_compaction.json @@ -0,0 +1,104 @@ +{ + "format-version": 2, + "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1", + "location": "{{ table_location }}", + "last-sequence-number": 34, + "last-updated-ms": 1602638573590, + "last-column-id": 3, + "current-schema-id": 1, + "schemas": [ + { + "type": "struct", + "schema-id": 0, + "fields": [ + {"id": 1, "name": "x", "required": true, "type": "long"} + ] + }, + { + "type": "struct", + "schema-id": 1, + "identifier-field-ids": [1, 2], + "fields": [ + {"id": 1, "name": "x", "required": true, "type": "long"}, + {"id": 2, "name": "y", "required": true, "type": "long", "doc": "comment"}, + {"id": 3, "name": "z", "required": true, "type": "long"} + ] + } + ], + "default-spec-id": 0, + "partition-specs": [ + { + "spec-id": 0, + "fields": [ + {"name": "x", "transform": "identity", "source-id": 1, "field-id": 1000} + ] + } + ], + "last-partition-id": 1000, + "default-sort-order-id": 3, + "sort-orders": [ + { + "order-id": 3, + "fields": [ + {"transform": "identity", "source-id": 2, "direction": "asc", "null-order": "nulls-first"}, + {"transform": "bucket[4]", "source-id": 3, "direction": "desc", "null-order": "nulls-last"} + ] + } + ], + "properties": {}, + "current-snapshot-id": 3059729675574597004, + "snapshots": [ + { + "snapshot-id": 3051729675574597004, + "timestamp-ms": 1515100955770, + "sequence-number": 0, + "summary": {"operation": "append"}, + "manifest-list": "{{ manifest_list_s1_location }}" + }, + { + "snapshot-id": 3055729675574597004, + "parent-snapshot-id": 3051729675574597004, + "timestamp-ms": 1555100955770, + "sequence-number": 1, + "summary": {"operation": "append"}, + "manifest-list": "{{ manifest_list_s2_location }}", + "schema-id": 1 + }, + { + "snapshot-id": 3056729675574597004, + "parent-snapshot-id": 3055729675574597004, + "timestamp-ms": 1575100955770, + "sequence-number": 2, + "summary": {"operation": "append"}, + "manifest-list": "{{ manifest_list_s3_location }}", + "schema-id": 1 + }, + { + "snapshot-id": 3057729675574597004, + "parent-snapshot-id": 3056729675574597004, + "timestamp-ms": 1595100955770, + "sequence-number": 3, + "summary": {"operation": "replace"}, + "manifest-list": "{{ manifest_list_s4_location }}", + "schema-id": 1 + }, + { + "snapshot-id": 3059729675574597004, + "parent-snapshot-id": 3057729675574597004, + "timestamp-ms": 1602638573590, + "sequence-number": 4, + "summary": {"operation": "append"}, + "manifest-list": "{{ manifest_list_s5_location }}", + "schema-id": 1 + } + ], + "snapshot-log": [ + {"snapshot-id": 3051729675574597004, "timestamp-ms": 1515100955770}, + {"snapshot-id": 3055729675574597004, "timestamp-ms": 1555100955770}, + {"snapshot-id": 3056729675574597004, "timestamp-ms": 1575100955770}, + {"snapshot-id": 3057729675574597004, "timestamp-ms": 1595100955770}, + {"snapshot-id": 3059729675574597004, "timestamp-ms": 1602638573590} + ], + "metadata-log": [], + "refs": {"main": {"snapshot-id": 3059729675574597004, "type": "branch"}} +} diff --git a/crates/iceberg/testdata/example_table_metadata_v2_deep_history_stale_schema.json b/crates/iceberg/testdata/example_table_metadata_v2_deep_history_stale_schema.json new file mode 100644 index 0000000000..f5a08e1b9b --- /dev/null +++ b/crates/iceberg/testdata/example_table_metadata_v2_deep_history_stale_schema.json @@ -0,0 +1,105 @@ +{ + "format-version": 2, + "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1", + "location": "{{ table_location }}", + "last-sequence-number": 34, + "last-updated-ms": 1602638573590, + "last-column-id": 3, + "current-schema-id": 1, + "schemas": [ + { + "type": "struct", + "schema-id": 0, + "fields": [ + {"id": 1, "name": "x", "required": true, "type": "long"} + ] + }, + { + "type": "struct", + "schema-id": 1, + "identifier-field-ids": [1, 2], + "fields": [ + {"id": 1, "name": "x", "required": true, "type": "long"}, + {"id": 2, "name": "y", "required": true, "type": "long", "doc": "comment"}, + {"id": 3, "name": "z", "required": true, "type": "long"} + ] + } + ], + "default-spec-id": 0, + "partition-specs": [ + { + "spec-id": 0, + "fields": [ + {"name": "x", "transform": "identity", "source-id": 1, "field-id": 1000} + ] + } + ], + "last-partition-id": 1000, + "default-sort-order-id": 3, + "sort-orders": [ + { + "order-id": 3, + "fields": [ + {"transform": "identity", "source-id": 2, "direction": "asc", "null-order": "nulls-first"}, + {"transform": "bucket[4]", "source-id": 3, "direction": "desc", "null-order": "nulls-last"} + ] + } + ], + "properties": {}, + "current-snapshot-id": 3059729675574597004, + "snapshots": [ + { + "snapshot-id": 3051729675574597004, + "timestamp-ms": 1515100955770, + "sequence-number": 0, + "summary": {"operation": "append"}, + "manifest-list": "{{ manifest_list_s1_location }}", + "schema-id": 0 + }, + { + "snapshot-id": 3055729675574597004, + "parent-snapshot-id": 3051729675574597004, + "timestamp-ms": 1555100955770, + "sequence-number": 1, + "summary": {"operation": "append"}, + "manifest-list": "{{ manifest_list_s2_location }}", + "schema-id": 0 + }, + { + "snapshot-id": 3056729675574597004, + "parent-snapshot-id": 3055729675574597004, + "timestamp-ms": 1575100955770, + "sequence-number": 2, + "summary": {"operation": "append"}, + "manifest-list": "{{ manifest_list_s3_location }}", + "schema-id": 0 + }, + { + "snapshot-id": 3057729675574597004, + "parent-snapshot-id": 3056729675574597004, + "timestamp-ms": 1595100955770, + "sequence-number": 3, + "summary": {"operation": "overwrite"}, + "manifest-list": "{{ manifest_list_s4_location }}", + "schema-id": 0 + }, + { + "snapshot-id": 3059729675574597004, + "parent-snapshot-id": 3057729675574597004, + "timestamp-ms": 1602638573590, + "sequence-number": 4, + "summary": {"operation": "append"}, + "manifest-list": "{{ manifest_list_s5_location }}", + "schema-id": 0 + } + ], + "snapshot-log": [ + {"snapshot-id": 3051729675574597004, "timestamp-ms": 1515100955770}, + {"snapshot-id": 3055729675574597004, "timestamp-ms": 1555100955770}, + {"snapshot-id": 3056729675574597004, "timestamp-ms": 1575100955770}, + {"snapshot-id": 3057729675574597004, "timestamp-ms": 1595100955770}, + {"snapshot-id": 3059729675574597004, "timestamp-ms": 1602638573590} + ], + "metadata-log": [], + "refs": {"main": {"snapshot-id": 3059729675574597004, "type": "branch"}} +} diff --git a/crates/integrations/datafusion/public-api.txt b/crates/integrations/datafusion/public-api.txt index e197c2057d..35eeb411cb 100644 --- a/crates/integrations/datafusion/public-api.txt +++ b/crates/integrations/datafusion/public-api.txt @@ -17,7 +17,6 @@ impl iceberg_datafusion::physical_plan::IcebergTableScan pub fn iceberg_datafusion::physical_plan::IcebergTableScan::limit(&self) -> core::option::Option pub fn iceberg_datafusion::physical_plan::IcebergTableScan::predicates(&self) -> core::option::Option<&iceberg::expr::predicate::Predicate> pub fn iceberg_datafusion::physical_plan::IcebergTableScan::projection(&self) -> core::option::Option<&[alloc::string::String]> -pub fn iceberg_datafusion::physical_plan::IcebergTableScan::snapshot_id(&self) -> core::option::Option pub fn iceberg_datafusion::physical_plan::IcebergTableScan::table(&self) -> &iceberg::table::Table impl core::fmt::Debug for iceberg_datafusion::physical_plan::IcebergTableScan pub fn iceberg_datafusion::physical_plan::IcebergTableScan::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result @@ -57,8 +56,11 @@ impl datafusion_catalog::table::TableProviderFactory for iceberg_datafusion::tab pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::create<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _state: &'life1 dyn datafusion_session::session::Session, cmd: &'life2 datafusion_expr::logical_plan::ddl::CreateExternalTable) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait pub struct iceberg_datafusion::table::IcebergStaticTableProvider impl iceberg_datafusion::IcebergStaticTableProvider +pub async fn iceberg_datafusion::IcebergStaticTableProvider::try_new_appends_after(table: iceberg::table::Table, from_snapshot_id: i64) -> iceberg::error::Result pub async fn iceberg_datafusion::IcebergStaticTableProvider::try_new_from_table(table: iceberg::table::Table) -> iceberg::error::Result pub async fn iceberg_datafusion::IcebergStaticTableProvider::try_new_from_table_snapshot(table: iceberg::table::Table, snapshot_id: i64) -> iceberg::error::Result +pub async fn iceberg_datafusion::IcebergStaticTableProvider::try_new_incremental(table: iceberg::table::Table, from_snapshot_id: i64, to_snapshot_id: i64) -> iceberg::error::Result +pub async fn iceberg_datafusion::IcebergStaticTableProvider::try_new_incremental_inclusive(table: iceberg::table::Table, from_snapshot_id: i64, to_snapshot_id: i64) -> iceberg::error::Result impl core::clone::Clone for iceberg_datafusion::IcebergStaticTableProvider pub fn iceberg_datafusion::IcebergStaticTableProvider::clone(&self) -> iceberg_datafusion::IcebergStaticTableProvider impl core::fmt::Debug for iceberg_datafusion::IcebergStaticTableProvider @@ -101,8 +103,11 @@ pub fn iceberg_datafusion::IcebergCatalogProvider::schema(&self, name: &str) -> pub fn iceberg_datafusion::IcebergCatalogProvider::schema_names(&self) -> alloc::vec::Vec pub struct iceberg_datafusion::IcebergStaticTableProvider impl iceberg_datafusion::IcebergStaticTableProvider +pub async fn iceberg_datafusion::IcebergStaticTableProvider::try_new_appends_after(table: iceberg::table::Table, from_snapshot_id: i64) -> iceberg::error::Result pub async fn iceberg_datafusion::IcebergStaticTableProvider::try_new_from_table(table: iceberg::table::Table) -> iceberg::error::Result pub async fn iceberg_datafusion::IcebergStaticTableProvider::try_new_from_table_snapshot(table: iceberg::table::Table, snapshot_id: i64) -> iceberg::error::Result +pub async fn iceberg_datafusion::IcebergStaticTableProvider::try_new_incremental(table: iceberg::table::Table, from_snapshot_id: i64, to_snapshot_id: i64) -> iceberg::error::Result +pub async fn iceberg_datafusion::IcebergStaticTableProvider::try_new_incremental_inclusive(table: iceberg::table::Table, from_snapshot_id: i64, to_snapshot_id: i64) -> iceberg::error::Result impl core::clone::Clone for iceberg_datafusion::IcebergStaticTableProvider pub fn iceberg_datafusion::IcebergStaticTableProvider::clone(&self) -> iceberg_datafusion::IcebergStaticTableProvider impl core::fmt::Debug for iceberg_datafusion::IcebergStaticTableProvider diff --git a/crates/integrations/datafusion/src/physical_plan/scan.rs b/crates/integrations/datafusion/src/physical_plan/scan.rs index 1e53c20cec..fbfef38a28 100644 --- a/crates/integrations/datafusion/src/physical_plan/scan.rs +++ b/crates/integrations/datafusion/src/physical_plan/scan.rs @@ -35,14 +35,29 @@ use iceberg::table::Table; use super::expr_to_predicate::convert_filters_to_predicate; use crate::to_datafusion_error; +/// Describes which snapshot(s) to scan. +#[derive(Debug, Clone)] +pub(crate) enum ScanRange { + /// Scan the current (latest) snapshot. + Latest, + /// Scan a specific point-in-time snapshot. + PointInTime(i64), + /// Incremental append scan between two snapshots. + Incremental { + from: i64, + to: Option, + from_inclusive: bool, + }, +} + /// Manages the scanning process of an Iceberg [`Table`], encapsulating the /// necessary details and computed properties required for execution planning. #[derive(Debug)] pub struct IcebergTableScan { /// A table in the catalog. table: Table, - /// Snapshot of the table to scan. - snapshot_id: Option, + /// Which snapshot(s) to scan. + scan_range: ScanRange, /// Stores certain, often expensive to compute, /// plan properties used in query optimization. plan_properties: Arc, @@ -58,7 +73,7 @@ impl IcebergTableScan { /// Creates a new [`IcebergTableScan`] object. pub(crate) fn new( table: Table, - snapshot_id: Option, + scan_range: ScanRange, schema: ArrowSchemaRef, projection: Option<&Vec>, filters: &[Expr], @@ -74,7 +89,7 @@ impl IcebergTableScan { Self { table, - snapshot_id, + scan_range, plan_properties, projection, predicates, @@ -86,8 +101,9 @@ impl IcebergTableScan { &self.table } - pub fn snapshot_id(&self) -> Option { - self.snapshot_id + #[cfg(test)] + pub(crate) fn scan_range(&self) -> &ScanRange { + &self.scan_range } pub fn projection(&self) -> Option<&[String]> { @@ -143,7 +159,7 @@ impl ExecutionPlan for IcebergTableScan { ) -> DFResult { let fut = get_batch_stream( self.table.clone(), - self.snapshot_id, + self.scan_range.clone(), self.projection.clone(), self.predicates.clone(), ); @@ -204,25 +220,47 @@ impl DisplayAs for IcebergTableScan { /// /// This function initializes a [`TableScan`], builds it, /// and then converts it into a stream of Arrow [`RecordBatch`]es. +/// +/// Supports both regular point-in-time scans and incremental scans. async fn get_batch_stream( table: Table, - snapshot_id: Option, + scan_range: ScanRange, column_names: Option>, predicates: Option, ) -> DFResult> + Send>>> { - let scan_builder = match snapshot_id { - Some(snapshot_id) => table.scan().snapshot_id(snapshot_id), - None => table.scan(), - }; + // Apply column selection, predicates, and build a TableScan. + macro_rules! configure_and_build { + ($builder:expr) => {{ + let mut b = $builder; + b = match column_names { + Some(names) => b.select(names), + None => b.select_all(), + }; + if let Some(pred) = predicates { + b = b.with_filter(pred); + } + b.build().map_err(to_datafusion_error)? + }}; + } - let mut scan_builder = match column_names { - Some(column_names) => scan_builder.select(column_names), - None => scan_builder.select_all(), + let table_scan = match scan_range { + ScanRange::Incremental { + from, + to, + from_inclusive, + } => { + let scan_builder = if from_inclusive { + table.incremental_append_scan_inclusive(from, to) + } else { + table.incremental_append_scan(from, to) + }; + configure_and_build!(scan_builder) + } + ScanRange::Latest => configure_and_build!(table.scan()), + ScanRange::PointInTime(snapshot_id) => { + configure_and_build!(table.scan().snapshot_id(snapshot_id)) + } }; - if let Some(pred) = predicates { - scan_builder = scan_builder.with_filter(pred); - } - let table_scan = scan_builder.build().map_err(to_datafusion_error)?; let stream = table_scan .to_arrow() diff --git a/crates/integrations/datafusion/src/table/mod.rs b/crates/integrations/datafusion/src/table/mod.rs index 2fd958dff4..be5d018a82 100644 --- a/crates/integrations/datafusion/src/table/mod.rs +++ b/crates/integrations/datafusion/src/table/mod.rs @@ -52,7 +52,7 @@ use crate::error::to_datafusion_error; use crate::physical_plan::commit::IcebergCommitExec; use crate::physical_plan::project::project_with_partition; use crate::physical_plan::repartition::repartition; -use crate::physical_plan::scan::IcebergTableScan; +use crate::physical_plan::scan::{IcebergTableScan, ScanRange}; use crate::physical_plan::sort::sort_by_partition; use crate::physical_plan::write::IcebergWriteExec; @@ -134,7 +134,7 @@ impl TableProvider for IcebergTableProvider { // Create scan with fresh metadata (always use current snapshot) Ok(Arc::new(IcebergTableScan::new( table, - None, // Always use current snapshot for catalog-backed provider + ScanRange::Latest, self.schema.clone(), projection, filters, @@ -240,7 +240,7 @@ impl TableProvider for IcebergTableProvider { /// /// This provider holds a cached table instance and does not refresh metadata or support /// write operations. Use this for consistent analytical queries, time-travel scenarios, -/// or when you want to avoid catalog overhead. +/// incremental reads, or when you want to avoid catalog overhead. /// /// For catalog-backed tables with write support and automatic refresh, use /// [`IcebergTableProvider`] instead. @@ -248,10 +248,10 @@ impl TableProvider for IcebergTableProvider { pub struct IcebergStaticTableProvider { /// The static table instance (never refreshed) table: Table, - /// Optional snapshot ID for this static view - snapshot_id: Option, /// A reference-counted arrow `Schema` schema: ArrowSchemaRef, + /// Which snapshot(s) to scan + scan_range: ScanRange, } impl IcebergStaticTableProvider { @@ -262,8 +262,8 @@ impl IcebergStaticTableProvider { let schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?); Ok(IcebergStaticTableProvider { table, - snapshot_id: None, schema, + scan_range: ScanRange::Latest, }) } @@ -288,8 +288,134 @@ impl IcebergStaticTableProvider { let schema = Arc::new(schema_to_arrow_schema(&table_schema)?); Ok(IcebergStaticTableProvider { table, - snapshot_id: Some(snapshot_id), schema, + scan_range: ScanRange::PointInTime(snapshot_id), + }) + } + + /// Creates a provider for incremental scanning between two snapshots. + /// + /// Returns only data files that were added in snapshots between `from_snapshot_id` + /// (exclusive) and `to_snapshot_id` (inclusive). Only APPEND operations are supported. + /// + /// # Arguments + /// * `table` - The table to scan + /// * `from_snapshot_id` - Starting snapshot (exclusive - changes after this are included) + /// * `to_snapshot_id` - Ending snapshot (inclusive) + /// + /// # Example + /// ```ignore + /// let provider = IcebergStaticTableProvider::try_new_incremental(table, 100, 200).await?; + /// ctx.register_table("changes", Arc::new(provider))?; + /// let df = ctx.sql("SELECT * FROM changes").await?; + /// ``` + pub async fn try_new_incremental( + table: Table, + from_snapshot_id: i64, + to_snapshot_id: i64, + ) -> Result { + table + .metadata() + .snapshot_by_id(to_snapshot_id) + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "to_snapshot id {to_snapshot_id} not found in table {}", + table.identifier().name() + ), + ) + })?; + // Incremental scans project onto the current schema, so the reported + // schema must match it rather than the to-snapshot's schema. + let schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?); + Ok(IcebergStaticTableProvider { + table, + schema, + scan_range: ScanRange::Incremental { + from: from_snapshot_id, + to: Some(to_snapshot_id), + from_inclusive: false, + }, + }) + } + + /// Creates a provider for incremental scanning between two snapshots (inclusive). + /// + /// Returns only data files that were added in snapshots between `from_snapshot_id` + /// (inclusive) and `to_snapshot_id` (inclusive). Only APPEND operations are supported. + /// + /// # Arguments + /// * `table` - The table to scan + /// * `from_snapshot_id` - Starting snapshot (inclusive - changes from this snapshot are included) + /// * `to_snapshot_id` - Ending snapshot (inclusive) + pub async fn try_new_incremental_inclusive( + table: Table, + from_snapshot_id: i64, + to_snapshot_id: i64, + ) -> Result { + table + .metadata() + .snapshot_by_id(to_snapshot_id) + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "to_snapshot id {to_snapshot_id} not found in table {}", + table.identifier().name() + ), + ) + })?; + // Incremental scans project onto the current schema, so the reported + // schema must match it rather than the to-snapshot's schema. + let schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?); + Ok(IcebergStaticTableProvider { + table, + schema, + scan_range: ScanRange::Incremental { + from: from_snapshot_id, + to: Some(to_snapshot_id), + from_inclusive: true, + }, + }) + } + + /// Creates a provider for scanning all appends after a snapshot up to the current snapshot. + /// + /// Returns only data files that were added in snapshots after `from_snapshot_id` + /// up to and including the current snapshot. Only APPEND operations are supported. + /// + /// # Arguments + /// * `table` - The table to scan + /// * `from_snapshot_id` - Starting snapshot (exclusive - changes after this are included) + /// + /// # Example + /// ```ignore + /// let provider = IcebergStaticTableProvider::try_new_appends_after(table, 100).await?; + /// ctx.register_table("new_data", Arc::new(provider))?; + /// let df = ctx.sql("SELECT * FROM new_data").await?; + /// ``` + pub async fn try_new_appends_after(table: Table, from_snapshot_id: i64) -> Result { + table.metadata().current_snapshot().ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "table {} has no current snapshot", + table.identifier().name() + ), + ) + })?; + // Incremental scans project onto the current schema, so the reported + // schema must match it rather than the to-snapshot's schema. + let schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?); + Ok(IcebergStaticTableProvider { + table, + schema, + scan_range: ScanRange::Incremental { + from: from_snapshot_id, + to: None, + from_inclusive: false, + }, }) } } @@ -314,7 +440,7 @@ impl TableProvider for IcebergStaticTableProvider { // Use cached table (no refresh) Ok(Arc::new(IcebergTableScan::new( self.table.clone(), - self.snapshot_id, + self.scan_range.clone(), self.schema.clone(), projection, filters, @@ -894,4 +1020,122 @@ mod tests { "Limit should be None when not specified" ); } + + // Tests for incremental scan providers + + #[tokio::test] + async fn test_static_provider_incremental_creates_scan() { + use datafusion::datasource::TableProvider; + + let table = get_test_table_from_metadata_file().await; + let snapshots: Vec<_> = table.metadata().snapshots().collect(); + + assert!(snapshots.len() >= 2); + let from_id = snapshots[0].snapshot_id(); + let to_id = snapshots[snapshots.len() - 1].snapshot_id(); + + let provider = + IcebergStaticTableProvider::try_new_incremental(table.clone(), from_id, to_id) + .await + .unwrap(); + + let ctx = SessionContext::new(); + let state = ctx.state(); + + let scan_plan = provider.scan(&state, None, &[], None).await.unwrap(); + let iceberg_scan = scan_plan + .downcast_ref::() + .expect("Expected IcebergTableScan"); + + assert!(matches!( + iceberg_scan.scan_range(), + ScanRange::Incremental { + from, + to: Some(to), + from_inclusive: false + } if *from == from_id && *to == to_id + )); + } + + #[tokio::test] + async fn test_static_provider_incremental_inclusive() { + use datafusion::datasource::TableProvider; + + let table = get_test_table_from_metadata_file().await; + let snapshots: Vec<_> = table.metadata().snapshots().collect(); + + assert!(snapshots.len() >= 2); + let from_id = snapshots[0].snapshot_id(); + let to_id = snapshots[snapshots.len() - 1].snapshot_id(); + + let provider = IcebergStaticTableProvider::try_new_incremental_inclusive( + table.clone(), + from_id, + to_id, + ) + .await + .unwrap(); + + let ctx = SessionContext::new(); + let state = ctx.state(); + + let scan_plan = provider.scan(&state, None, &[], None).await.unwrap(); + let iceberg_scan = scan_plan + .downcast_ref::() + .expect("Expected IcebergTableScan"); + + assert!(matches!( + iceberg_scan.scan_range(), + ScanRange::Incremental { + from, + to: Some(to), + from_inclusive: true + } if *from == from_id && *to == to_id + )); + } + + #[tokio::test] + async fn test_static_provider_appends_after() { + use datafusion::datasource::TableProvider; + + let table = get_test_table_from_metadata_file().await; + let snapshots: Vec<_> = table.metadata().snapshots().collect(); + + assert!(!snapshots.is_empty()); + let from_id = snapshots[0].snapshot_id(); + + let provider = IcebergStaticTableProvider::try_new_appends_after(table.clone(), from_id) + .await + .unwrap(); + + let ctx = SessionContext::new(); + let state = ctx.state(); + + let scan_plan = provider.scan(&state, None, &[], None).await.unwrap(); + let iceberg_scan = scan_plan + .downcast_ref::() + .expect("Expected IcebergTableScan"); + + assert!(matches!( + iceberg_scan.scan_range(), + ScanRange::Incremental { + from, + to: None, + from_inclusive: false + } if *from == from_id + )); + } + + #[tokio::test] + async fn test_static_provider_incremental_invalid_snapshot() { + let table = get_test_table_from_metadata_file().await; + + // Test with invalid to_snapshot_id + let result = + IcebergStaticTableProvider::try_new_incremental(table.clone(), 1, 999999999).await; + + assert!(result.is_err()); + let err = result.unwrap_err(); + assert!(err.to_string().contains("not found")); + } }