diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 24e276db56..2afa28b673 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1328,6 +1328,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 @@ -3132,6 +3146,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 @@ -3147,6 +3163,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 67b8cde4d8..b1d62c599b 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, unified_partition_type: Option>, } @@ -84,6 +93,7 @@ impl ManifestFileContext { delete_file_index, name_mapping, case_sensitive, + entry_filter, partition_spec, unified_partition_type, } = self; @@ -91,6 +101,12 @@ impl ManifestFileContext { 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(), @@ -156,7 +172,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, @@ -172,10 +187,21 @@ 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, pub unified_partition_type: 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) + .finish_non_exhaustive() + } +} + impl PlanContext { pub(crate) async fn get_manifest_list(&self) -> Result> { self.object_cache @@ -229,6 +255,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 { @@ -299,6 +331,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..d6d77bd184 --- /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: 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 b7c7254ae2..144136a418 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; @@ -42,7 +44,9 @@ use crate::metadata_columns::{ }; use crate::partitioning::compute_unified_partition_type; 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}; @@ -50,6 +54,162 @@ 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. A standard scan passes the snapshot's + /// own schema (the correct behavior for time-travel scans). An incremental + /// scan passes the table's current schema so that rows written under an + /// older schema in the range are projected onto it (newer columns become + /// `NULL`), matching the Java and PyIceberg implementations. + schema: SchemaRef, +} + +/// 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 = config.schema.clone(); + + // 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); + + // Compute unified partition type if _partition is projected + let unified_partition_type = if field_ids.contains(&RESERVED_FIELD_ID_PARTITION) { + let partition_type = compute_unified_partition_type( + config + .table + .metadata() + .partition_specs_iter() + .map(|s| s.as_ref()), + &schema, + )?; + Some(Arc::new(partition_type)) + } else { + None + }; + + 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, + unified_partition_type, + }; + + 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, @@ -203,7 +363,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, @@ -217,134 +377,32 @@ impl<'a> TableScanBuilder<'a> { runtime: self.table.runtime().clone(), }); }; - current_snapshot_id.clone() + current_snapshot.clone() } }; + // A standard scan projects onto the snapshot's own schema, so that + // time-travel reads see the table exactly as it was at that snapshot. 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); - - // Compute unified partition type if _partition is projected - let unified_partition_type = if field_ids.contains(&RESERVED_FIELD_ID_PARTITION) { - let partition_type = compute_unified_partition_type( - self.table - .metadata() - .partition_specs_iter() - .map(|s| s.as_ref()), - &schema, - )?; - Some(Arc::new(partition_type)) - } else { - None - }; - - 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, + }, 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()), - unified_partition_type, - }; - - 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, + ) } } @@ -669,10 +727,10 @@ pub mod tests { use crate::scan::FileScanTask; use crate::spec::{ DEFAULT_SCHEMA_NAME_MAPPING, DataContentType, DataFileBuilder, DataFileFormat, Datum, - FormatVersion, Literal, MAIN_BRANCH, ManifestEntry, ManifestListWriter, ManifestStatus, - ManifestWriterBuilder, NestedField, Operation, PartitionSpec, PrimitiveType, Schema, - Snapshot, Struct, StructType, Summary, TableMetadata, TableMetadataBuilder, Type, - UnboundPartitionSpec, + FormatVersion, Literal, MAIN_BRANCH, ManifestEntry, ManifestFile, ManifestListWriter, + ManifestStatus, ManifestWriterBuilder, NestedField, Operation, PartitionSpec, + PrimitiveType, Schema, Snapshot, Struct, StructType, Summary, TableMetadata, + TableMetadataBuilder, Type, UnboundPartitionSpec, }; use crate::table::Table; use crate::test_utils::test_runtime; @@ -786,22 +844,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() @@ -1015,6 +1123,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() + } + /// Writes a v3 data manifest with a manifest-level `first_row_id` of 42, /// so live entries inherit a per-file `first_row_id` on read. Upgrades the /// table to v3 first, so the manifest list is read as v3. 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/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"}} +}