iceberg/util/
snapshot.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::collections::HashSet;
19
20use crate::spec::{SnapshotRef, TableMetadataRef};
21
22struct Ancestors {
23    next: Option<SnapshotRef>,
24    get_snapshot: Box<dyn Fn(i64) -> Option<SnapshotRef> + Send>,
25    visited: HashSet<i64>,
26}
27
28impl Iterator for Ancestors {
29    type Item = SnapshotRef;
30
31    fn next(&mut self) -> Option<Self::Item> {
32        let snapshot = self.next.take()?;
33        // corrupt metadata with a parent cycle must not hang the traversal
34        if !self.visited.insert(snapshot.snapshot_id()) {
35            return None;
36        }
37        self.next = snapshot
38            .parent_snapshot_id()
39            .and_then(|id| (self.get_snapshot)(id));
40        Some(snapshot)
41    }
42}
43
44/// Iterate starting from `snapshot_id` (inclusive) to the root snapshot.
45///
46/// The iterator visits each snapshot at most once, so it terminates even on
47/// corrupt metadata whose parent pointers form a cycle.
48pub fn ancestors_of(
49    table_metadata: &TableMetadataRef,
50    snapshot_id: i64,
51) -> impl Iterator<Item = SnapshotRef> + Send {
52    let initial = table_metadata.snapshot_by_id(snapshot_id).cloned();
53    let table_metadata = table_metadata.clone();
54    Ancestors {
55        next: initial,
56        get_snapshot: Box::new(move |id| table_metadata.snapshot_by_id(id).cloned()),
57        visited: HashSet::new(),
58    }
59}
60
61/// Iterate starting from `latest_snapshot_id` (inclusive) to `oldest_snapshot_id` (exclusive).
62///
63/// Note: if `oldest_snapshot_id` is `Some(id)` but `id` is not actually an
64/// ancestor of `latest_snapshot_id`, the walk is never stopped and this yields
65/// *all* ancestors of `latest_snapshot_id` down to the root. Callers that treat
66/// `oldest_snapshot_id` as a lower bound must validate the lineage themselves.
67pub fn ancestors_between(
68    table_metadata: &TableMetadataRef,
69    latest_snapshot_id: i64,
70    oldest_snapshot_id: Option<i64>,
71) -> impl Iterator<Item = SnapshotRef> + Send {
72    ancestors_of(table_metadata, latest_snapshot_id).take_while(move |snapshot| {
73        oldest_snapshot_id
74            .map(|id| snapshot.snapshot_id() != id)
75            .unwrap_or(true)
76    })
77}
78
79#[cfg(test)]
80mod tests {
81    use super::*;
82    use crate::scan::tests::TableTestFixture;
83
84    // Five snapshots chained as: S1 (root) -> S2 -> S3 -> S4 -> S5 (current)
85    const S1: i64 = 3051729675574597004;
86    const S2: i64 = 3055729675574597004;
87    const S3: i64 = 3056729675574597004;
88    const S4: i64 = 3057729675574597004;
89    const S5: i64 = 3059729675574597004;
90
91    fn metadata() -> TableMetadataRef {
92        let fixture = TableTestFixture::new_with_deep_history();
93        std::sync::Arc::new(fixture.table.metadata().clone())
94    }
95
96    // --- ancestors_of ---
97
98    #[test]
99    fn test_ancestors_of_nonexistent_snapshot_returns_empty() {
100        let meta = metadata();
101        let ids: Vec<i64> = ancestors_of(&meta, 999).map(|s| s.snapshot_id()).collect();
102        assert!(ids.is_empty());
103    }
104
105    #[test]
106    fn test_ancestors_of_root_returns_only_root() {
107        let meta = metadata();
108        let ids: Vec<i64> = ancestors_of(&meta, S1).map(|s| s.snapshot_id()).collect();
109        assert_eq!(ids, vec![S1]);
110    }
111
112    #[test]
113    fn test_ancestors_of_leaf_returns_full_chain() {
114        let meta = metadata();
115        let ids: Vec<i64> = ancestors_of(&meta, S5).map(|s| s.snapshot_id()).collect();
116        assert_eq!(ids, vec![S5, S4, S3, S2, S1]);
117    }
118
119    #[test]
120    fn test_ancestors_of_mid_chain_returns_partial_chain() {
121        let meta = metadata();
122        let ids: Vec<i64> = ancestors_of(&meta, S3).map(|s| s.snapshot_id()).collect();
123        assert_eq!(ids, vec![S3, S2, S1]);
124    }
125
126    #[test]
127    fn test_ancestors_of_second_snapshot() {
128        let meta = metadata();
129        let ids: Vec<i64> = ancestors_of(&meta, S2).map(|s| s.snapshot_id()).collect();
130        assert_eq!(ids, vec![S2, S1]);
131    }
132
133    // --- ancestors_between ---
134
135    #[test]
136    fn test_ancestors_between_same_id_returns_empty() {
137        let meta = metadata();
138        let ids: Vec<i64> = ancestors_between(&meta, S3, Some(S3))
139            .map(|s| s.snapshot_id())
140            .collect();
141        assert!(ids.is_empty());
142    }
143
144    #[test]
145    fn test_ancestors_between_no_oldest_returns_all_ancestors() {
146        let meta = metadata();
147        let ids: Vec<i64> = ancestors_between(&meta, S5, None)
148            .map(|s| s.snapshot_id())
149            .collect();
150        assert_eq!(ids, vec![S5, S4, S3, S2, S1]);
151    }
152
153    #[test]
154    fn test_ancestors_between_excludes_oldest_snapshot() {
155        let meta = metadata();
156        // S5 down to (but not including) S2
157        let ids: Vec<i64> = ancestors_between(&meta, S5, Some(S2))
158            .map(|s| s.snapshot_id())
159            .collect();
160        assert_eq!(ids, vec![S5, S4, S3]);
161    }
162
163    #[test]
164    fn test_ancestors_between_adjacent_snapshots() {
165        let meta = metadata();
166        // S3 down to (but not including) S2 — only S3 itself
167        let ids: Vec<i64> = ancestors_between(&meta, S3, Some(S2))
168            .map(|s| s.snapshot_id())
169            .collect();
170        assert_eq!(ids, vec![S3]);
171    }
172
173    #[test]
174    fn test_ancestors_between_leaf_and_root() {
175        let meta = metadata();
176        // S5 down to (but not including) S1
177        let ids: Vec<i64> = ancestors_between(&meta, S5, Some(S1))
178            .map(|s| s.snapshot_id())
179            .collect();
180        assert_eq!(ids, vec![S5, S4, S3, S2]);
181    }
182
183    #[test]
184    fn test_ancestors_between_nonexistent_oldest_returns_full_chain() {
185        let meta = metadata();
186        // oldest_snapshot_id doesn't exist in the chain, so take_while never stops
187        let ids: Vec<i64> = ancestors_between(&meta, S5, Some(999))
188            .map(|s| s.snapshot_id())
189            .collect();
190        assert_eq!(ids, vec![S5, S4, S3, S2, S1]);
191    }
192
193    #[test]
194    fn test_ancestors_between_nonexistent_latest_returns_empty() {
195        let meta = metadata();
196        let ids: Vec<i64> = ancestors_between(&meta, 999, Some(S1))
197            .map(|s| s.snapshot_id())
198            .collect();
199        assert!(ids.is_empty());
200    }
201}
iceberg/util/snapshot.rs

iceberg/util/
snapshot.rs
