From da00da2b6a9c1869a6e57e63c905a927903b78a5 Mon Sep 17 00:00:00 2001 From: Xander Date: Fri, 14 Aug 2026 13:47:02 +0200 Subject: [PATCH] feat(scan): incremental append scan Add IncrementalAppendScanBuilder to read files appended between two snapshots. Rows written under an older schema in the range are projected onto the table's current schema (newer columns become NULL), matching the Java and PyIceberg implementations. Refactors the shared table-scan build logic into build_table_scan / ScanConfig so it can be reused by both the standard and incremental scan builders. --- crates/iceberg/public-api.txt | 18 + crates/iceberg/src/scan/context.rs | 35 +- crates/iceberg/src/scan/incremental.rs | 912 ++++++++++++++++++ crates/iceberg/src/scan/mod.rs | 519 +++++++--- crates/iceberg/src/table.rs | 46 +- ...xample_table_metadata_v2_deep_history.json | 12 +- ...e_metadata_v2_deep_history_compaction.json | 104 ++ ...metadata_v2_deep_history_stale_schema.json | 105 ++ 8 files changed, 1610 insertions(+), 141 deletions(-) create mode 100644 crates/iceberg/src/scan/incremental.rs create mode 100644 crates/iceberg/testdata/example_table_metadata_v2_deep_history_compaction.json create mode 100644 crates/iceberg/testdata/example_table_metadata_v2_deep_history_stale_schema.json 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"}} +}