feat(iceberg): Add snapshot utils to scan ancestors (#2285)

## Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->

- Closes #2241 

## What changes are included in this PR?
- Add `Ancestors` to help scan past snapshots
- Moved existing util to the new utils mod
<!--
Provide a summary of the modifications in this PR. List the main changes
such as new features, bug fixes, refactoring, or any other updates.
-->

## Are these changes tested?
Yes
<!--
Specify what test covers (unit test, integration test, etc.).

If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->

Author: CTTY

crates/iceberg/src/arrow/reader.rs

@@ -58,7 +58,7 @@ use crate::io::{FileIO, FileMetadata, FileRead};
 use crate::metadata_columns::{RESERVED_FIELD_ID_FILE, is_metadata_field};
 use crate::scan::{ArrowRecordBatchStream, FileScanTask, FileScanTaskStream};
 use crate::spec::{Datum, NameMapping, NestedField, PrimitiveType, Schema, Type};
-use crate::utils::available_parallelism;
+use crate::util::available_parallelism;
 use crate::{Error, ErrorKind};
 
 /// Default gap between byte ranges below which they are coalesced into a

crates/iceberg/src/lib.rs

@@ -95,9 +95,10 @@ pub mod arrow;
 pub(crate) mod delete_file_index;
 pub mod encryption;
 pub mod test_utils;
-mod utils;
 pub mod writer;
 
 mod delete_vector;
 pub mod metadata_columns;
 pub mod puffin;
+/// Utility functions and modules.
+pub mod util;

crates/iceberg/src/scan/mod.rs

@@ -40,7 +40,7 @@ use crate::metadata_columns::{get_metadata_field_id, is_metadata_column_name};
 use crate::runtime::spawn;
 use crate::spec::{DataContentType, SnapshotRef};
 use crate::table::Table;
-use crate::utils::available_parallelism;
+use crate::util::available_parallelism;
 use crate::{Error, ErrorKind, Result};
 
 /// A stream of arrow [`RecordBatch`]es.
@@ -683,6 +683,39 @@ pub mod tests {
             }
         }
 
+        /// Creates a fixture with 5 snapshots chained as:
+        ///   S1 (root) -> S2 -> S3 -> S4 -> S5 (current)
+        /// Useful for testing snapshot history traversal.
+        pub fn new_with_deep_history() -> 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 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",
+                    env!("CARGO_MANIFEST_DIR")
+                ))
+                .unwrap();
+                serde_json::from_str::<TableMetadata>(&json_str).unwrap()
+            };
+
+            let table = Table::builder()
+                .metadata(table_metadata)
+                .identifier(TableIdent::from_strs(["db", "table1"]).unwrap())
+                .file_io(file_io.clone())
+                .metadata_location(table_metadata1_location.as_os_str().to_str().unwrap())
+                .build()
+                .unwrap();
+
+            Self {
+                table_location: table_location.to_str().unwrap().to_string(),
+                table,
+            }
+        }
+
         pub fn new_unpartitioned() -> Self {
             let tmp_dir = TempDir::new().unwrap();
             let table_location = tmp_dir.path().join("table1");

crates/iceberg/src/utils.rs crates/iceberg/src/util/mod.rscrates/iceberg/src/utils.rs renamed to crates/iceberg/src/util/mod.rs

@@ -17,6 +17,9 @@
 
 use std::num::NonZeroUsize;
 
+/// Utilities for working with snapshots.
+pub mod snapshot;
+
 // Use a default value of 1 as the safest option.
 // See https://doc.rust-lang.org/std/thread/fn.available_parallelism.html#limitations
 // for more details.

crates/iceberg/src/util/snapshot.rs

@@ -0,0 +1,185 @@
+// 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.
+
+use crate::spec::{SnapshotRef, TableMetadataRef};
+
+struct Ancestors {
+    next: Option<SnapshotRef>,
+    get_snapshot: Box<dyn Fn(i64) -> Option<SnapshotRef> + Send>,
+}
+
+impl Iterator for Ancestors {
+    type Item = SnapshotRef;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let snapshot = self.next.take()?;
+        self.next = snapshot
+            .parent_snapshot_id()
+            .and_then(|id| (self.get_snapshot)(id));
+        Some(snapshot)
+    }
+}
+
+/// Iterate starting from `snapshot_id` (inclusive) to the root snapshot.
+pub fn ancestors_of(
+    table_metadata: &TableMetadataRef,
+    snapshot_id: i64,
+) -> impl Iterator<Item = SnapshotRef> + Send {
+    let initial = table_metadata.snapshot_by_id(snapshot_id).cloned();
+    let table_metadata = table_metadata.clone();
+    Ancestors {
+        next: initial,
+        get_snapshot: Box::new(move |id| table_metadata.snapshot_by_id(id).cloned()),
+    }
+}
+
+/// Iterate starting from `latest_snapshot_id` (inclusive) to `oldest_snapshot_id` (exclusive).
+pub fn ancestors_between(
+    table_metadata: &TableMetadataRef,
+    latest_snapshot_id: i64,
+    oldest_snapshot_id: Option<i64>,
+) -> impl Iterator<Item = SnapshotRef> + Send {
+    ancestors_of(table_metadata, latest_snapshot_id).take_while(move |snapshot| {
+        oldest_snapshot_id
+            .map(|id| snapshot.snapshot_id() != id)
+            .unwrap_or(true)
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::scan::tests::TableTestFixture;
+
+    // Five snapshots chained as: S1 (root) -> S2 -> S3 -> S4 -> S5 (current)
+    const S1: i64 = 3051729675574597004;
+    const S2: i64 = 3055729675574597004;
+    const S3: i64 = 3056729675574597004;
+    const S4: i64 = 3057729675574597004;
+    const S5: i64 = 3059729675574597004;
+
+    fn metadata() -> TableMetadataRef {
+        let fixture = TableTestFixture::new_with_deep_history();
+        std::sync::Arc::new(fixture.table.metadata().clone())
+    }
+
+    // --- ancestors_of ---
+
+    #[test]
+    fn test_ancestors_of_nonexistent_snapshot_returns_empty() {
+        let meta = metadata();
+        let ids: Vec<i64> = ancestors_of(&meta, 999).map(|s| s.snapshot_id()).collect();
+        assert!(ids.is_empty());
+    }
+
+    #[test]
+    fn test_ancestors_of_root_returns_only_root() {
+        let meta = metadata();
+        let ids: Vec<i64> = ancestors_of(&meta, S1).map(|s| s.snapshot_id()).collect();
+        assert_eq!(ids, vec![S1]);
+    }
+
+    #[test]
+    fn test_ancestors_of_leaf_returns_full_chain() {
+        let meta = metadata();
+        let ids: Vec<i64> = ancestors_of(&meta, S5).map(|s| s.snapshot_id()).collect();
+        assert_eq!(ids, vec![S5, S4, S3, S2, S1]);
+    }
+
+    #[test]
+    fn test_ancestors_of_mid_chain_returns_partial_chain() {
+        let meta = metadata();
+        let ids: Vec<i64> = ancestors_of(&meta, S3).map(|s| s.snapshot_id()).collect();
+        assert_eq!(ids, vec![S3, S2, S1]);
+    }
+
+    #[test]
+    fn test_ancestors_of_second_snapshot() {
+        let meta = metadata();
+        let ids: Vec<i64> = ancestors_of(&meta, S2).map(|s| s.snapshot_id()).collect();
+        assert_eq!(ids, vec![S2, S1]);
+    }
+
+    // --- ancestors_between ---
+
+    #[test]
+    fn test_ancestors_between_same_id_returns_empty() {
+        let meta = metadata();
+        let ids: Vec<i64> = ancestors_between(&meta, S3, Some(S3))
+            .map(|s| s.snapshot_id())
+            .collect();
+        assert!(ids.is_empty());
+    }
+
+    #[test]
+    fn test_ancestors_between_no_oldest_returns_all_ancestors() {
+        let meta = metadata();
+        let ids: Vec<i64> = ancestors_between(&meta, S5, None)
+            .map(|s| s.snapshot_id())
+            .collect();
+        assert_eq!(ids, vec![S5, S4, S3, S2, S1]);
+    }
+
+    #[test]
+    fn test_ancestors_between_excludes_oldest_snapshot() {
+        let meta = metadata();
+        // S5 down to (but not including) S2
+        let ids: Vec<i64> = ancestors_between(&meta, S5, Some(S2))
+            .map(|s| s.snapshot_id())
+            .collect();
+        assert_eq!(ids, vec![S5, S4, S3]);
+    }
+
+    #[test]
+    fn test_ancestors_between_adjacent_snapshots() {
+        let meta = metadata();
+        // S3 down to (but not including) S2 — only S3 itself
+        let ids: Vec<i64> = ancestors_between(&meta, S3, Some(S2))
+            .map(|s| s.snapshot_id())
+            .collect();
+        assert_eq!(ids, vec![S3]);
+    }
+
+    #[test]
+    fn test_ancestors_between_leaf_and_root() {
+        let meta = metadata();
+        // S5 down to (but not including) S1
+        let ids: Vec<i64> = ancestors_between(&meta, S5, Some(S1))
+            .map(|s| s.snapshot_id())
+            .collect();
+        assert_eq!(ids, vec![S5, S4, S3, S2]);
+    }
+
+    #[test]
+    fn test_ancestors_between_nonexistent_oldest_returns_full_chain() {
+        let meta = metadata();
+        // oldest_snapshot_id doesn't exist in the chain, so take_while never stops
+        let ids: Vec<i64> = ancestors_between(&meta, S5, Some(999))
+            .map(|s| s.snapshot_id())
+            .collect();
+        assert_eq!(ids, vec![S5, S4, S3, S2, S1]);
+    }
+
+    #[test]
+    fn test_ancestors_between_nonexistent_latest_returns_empty() {
+        let meta = metadata();
+        let ids: Vec<i64> = ancestors_between(&meta, 999, Some(S1))
+            .map(|s| s.snapshot_id())
+            .collect();
+        assert!(ids.is_empty());
+    }
+}

crates/iceberg/testdata/example_table_metadata_v2_deep_history.json

@@ -0,0 +1,104 @@
+{
+  "format-version": 2,
+  "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
+  "location": "s3://bucket/test/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": "s3://bucket/metadata/snap-3051729675574597004.avro"
+    },
+    {
+      "snapshot-id": 3055729675574597004,
+      "parent-snapshot-id": 3051729675574597004,
+      "timestamp-ms": 1555100955770,
+      "sequence-number": 1,
+      "summary": {"operation": "append"},
+      "manifest-list": "s3://bucket/metadata/snap-3055729675574597004.avro",
+      "schema-id": 1
+    },
+    {
+      "snapshot-id": 3056729675574597004,
+      "parent-snapshot-id": 3055729675574597004,
+      "timestamp-ms": 1575100955770,
+      "sequence-number": 2,
+      "summary": {"operation": "append"},
+      "manifest-list": "s3://bucket/metadata/snap-3056729675574597004.avro",
+      "schema-id": 1
+    },
+    {
+      "snapshot-id": 3057729675574597004,
+      "parent-snapshot-id": 3056729675574597004,
+      "timestamp-ms": 1595100955770,
+      "sequence-number": 3,
+      "summary": {"operation": "overwrite"},
+      "manifest-list": "s3://bucket/metadata/snap-3057729675574597004.avro",
+      "schema-id": 1
+    },
+    {
+      "snapshot-id": 3059729675574597004,
+      "parent-snapshot-id": 3057729675574597004,
+      "timestamp-ms": 1602638573590,
+      "sequence-number": 4,
+      "summary": {"operation": "append"},
+      "manifest-list": "s3://bucket/metadata/snap-3059729675574597004.avro",
+      "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"}}
+}