feat(scan): incremental append scan - #2997
Conversation
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.
smaheshwar-pltr
left a comment
smaheshwar-pltr
left a comment
There was a problem hiding this comment.
Thanks for kicking this off! Left some initial comments
| // the current schema, so newer columns become `NULL`. | ||
| schema: self.table.metadata().current_schema().clone(), | ||
| }, | ||
| to_snapshot, |
There was a problem hiding this comment.
It looks like we're just reading the manifest list of this snapshot, but I think we need to be dedup-ing manifests from ADDED snapshots in the range. https://github.com/apache/iceberg/blob/2f6606a247e2b16be46ca6c02fc4cfc2e17691e6/core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java#L68-L98 / https://github.com/apache/iceberg-python/blob/b4e696c5331e9ac9b4e09fe82c47d4b1b62732a2/pyiceberg/table/__init__.py#L2443-L2476.
There was a problem hiding this comment.
Operations often rewrite ADDED manifests as EXISTING, so this would cause dropped rows. If that's the case, it's worth having test coverage of this scenario
There was a problem hiding this comment.
This is a very good catch and would have been a correctness issue. 2b4b71b
| ) -> Result<TableScan> { | ||
| let schema = config.schema.clone(); | ||
|
|
||
| // Check that all column names exist in the schema (skip reserved columns). |
There was a problem hiding this comment.
Should this respect config.case_sensitive ?
| } | ||
|
|
||
| let snapshot_bound_predicate = if let Some(ref predicates) = config.filter { | ||
| Some(predicates.bind(schema.clone(), true)?) |
There was a problem hiding this comment.
Should this respect config.case_sensitive ?
There was a problem hiding this comment.
Another good catch, this is actually pre-existing so I've put up a PR here #3059 to fix on main.
There was a problem hiding this comment.
Merged and resolved on this branch also.
| /// | ||
| /// 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( |
There was a problem hiding this comment.
Pointing out that the API we decided on for the PyIceberg V0 is https://github.com/apache/iceberg-python/blob/b4e696c5331e9ac9b4e09fe82c47d4b1b62732a2/pyiceberg/table/__init__.py#L1272-L1325 - a missing from_snapshot_id is permitted, mirroring Java's behaviour here https://github.com/apache/iceberg/blob/2f6606a247e2b16be46ca6c02fc4cfc2e17691e6/api/src/main/java/org/apache/iceberg/IncrementalScan.java#L27-L28.
There was a problem hiding this comment.
Happy to align with java and python api here.
| // 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(), | ||
| }); | ||
| } |
There was a problem hiding this comment.
Is this consistent with the other implementations? I suspect this diverges, but I've not looked closely at Java though I do remember considering this for PyIceberg and disallowing it so I imagine Java also does.
I think https://github.com/apache/iceberg/blob/2f6606a247e2b16be46ca6c02fc4cfc2e17691e6/core/src/main/java/org/apache/iceberg/BaseIncrementalScan.java#L177-L184 is what fails in Java and that's why we mirrored that for PyIceberg https://github.com/apache/iceberg-python/blob/b4e696c5331e9ac9b4e09fe82c47d4b1b62732a2/tests/table/test_init.py#L468-L487.
| } | ||
|
|
||
| #[test] | ||
| fn test_incremental_scan_projects_onto_current_schema() { |
There was a problem hiding this comment.
This test feels a bit odd to me. I think when verifying this behaviour a test like https://github.com/apache/iceberg-python/blob/b4e696c5331e9ac9b4e09fe82c47d4b1b62732a2/tests/integration/test_reads.py#L1350-L1373 would be nice, if that's feasible.
| impl iceberg::scan::TableScan | ||
| pub fn iceberg::scan::TableScan::column_names(&self) -> core::option::Option<&[alloc::string::String]> | ||
| pub async fn iceberg::scan::TableScan::plan_files(&self) -> iceberg::Result<iceberg::scan::FileScanTaskStream> | ||
| pub fn iceberg::scan::TableScan::snapshot(&self) -> core::option::Option<&iceberg::spec::SnapshotRef> |
There was a problem hiding this comment.
This was only used in tests and is now ambiguous in the case of incremental scans so I think this is fine to remove.
Resolves the conflict with apache#3059 (case-sensitive column resolution). That fix landed in TableScanBuilder::build, which this branch extracts into the shared build_table_scan, so it is carried into the shared path via config.case_sensitive rather than reverted. The incremental scan now honours case sensitivity as a result.
2 participants
Which issue does this PR close?
What changes are included in this PR?
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.
Are these changes tested?
AI Disclosure