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};
21use crate::{Error, ErrorKind, Result};
22
23struct Ancestors {
24    next: Option<SnapshotRef>,
25    get_snapshot: Box<dyn Fn(i64) -> Option<SnapshotRef> + Send>,
26    visited: HashSet<i64>,
27}
28
29impl Iterator for Ancestors {
30    type Item = SnapshotRef;
31
32    fn next(&mut self) -> Option<Self::Item> {
33        let snapshot = self.next.take()?;
34        // corrupt metadata with a parent cycle must not hang the traversal
35        if !self.visited.insert(snapshot.snapshot_id()) {
36            return None;
37        }
38        self.next = snapshot
39            .parent_snapshot_id()
40            .and_then(|id| (self.get_snapshot)(id));
41        Some(snapshot)
42    }
43}
44
45/// Iterate starting from `snapshot_id` (inclusive) to the root snapshot.
46///
47/// The iterator visits each snapshot at most once, so it terminates even on
48/// corrupt metadata whose parent pointers form a cycle.
49pub fn ancestors_of(
50    table_metadata: &TableMetadataRef,
51    snapshot_id: i64,
52) -> impl Iterator<Item = SnapshotRef> + Send {
53    let initial = table_metadata.snapshot_by_id(snapshot_id).cloned();
54    let table_metadata = table_metadata.clone();
55    Ancestors {
56        next: initial,
57        get_snapshot: Box::new(move |id| table_metadata.snapshot_by_id(id).cloned()),
58        visited: HashSet::new(),
59    }
60}
61
62/// Iterate starting from `latest_snapshot_id` (inclusive) to `oldest_snapshot_id` (exclusive).
63///
64/// Note: if `oldest_snapshot_id` is `Some(id)` but `id` is not actually an
65/// ancestor of `latest_snapshot_id`, the walk is never stopped and this yields
66/// *all* ancestors of `latest_snapshot_id` down to the root. Callers that treat
67/// `oldest_snapshot_id` as a lower bound must validate the lineage themselves.
68pub fn ancestors_between(
69    table_metadata: &TableMetadataRef,
70    latest_snapshot_id: i64,
71    oldest_snapshot_id: Option<i64>,
72) -> impl Iterator<Item = SnapshotRef> + Send {
73    ancestors_of(table_metadata, latest_snapshot_id).take_while(move |snapshot| {
74        oldest_snapshot_id
75            .map(|id| snapshot.snapshot_id() != id)
76            .unwrap_or(true)
77    })
78}
79
80/// Resolve the snapshot ID from the main-history entry with the greatest timestamp
81/// at or before `timestamp_ms` (milliseconds since the Unix epoch), taking the first
82/// entry on ties.
83///
84/// This matches Java's ordering, except that an entry at `i64::MIN` can match here
85/// but never matches in Java, which seeds its search with `Long.MIN_VALUE`.
86/// PyIceberg takes the last qualifying log entry, so ties or out-of-order
87/// timestamps can produce different results.
88///
89/// Returns [`ErrorKind::DataInvalid`] if no matching history exists.
90/// The returned snapshot may have expired, so
91/// [`snapshot_by_id`](crate::spec::TableMetadata::snapshot_by_id) can still return `None`.
92///
93/// ```
94/// # use iceberg::TableIdent;
95/// # use iceberg::io::FileIO;
96/// # use iceberg::table::StaticTable;
97/// use iceberg::util::snapshot::snapshot_id_as_of_time;
98/// # #[tokio::main]
99/// # async fn main() -> iceberg::Result<()> {
100/// # let location = concat!(env!("CARGO_MANIFEST_DIR"), "/testdata/example_table_metadata_v2.json");
101/// # let ident = TableIdent::from_strs(["ns", "t"])?;
102/// # let table = StaticTable::from_metadata_file(location, ident, FileIO::new_with_fs()).await?;
103/// let metadata = table.metadata();
104/// // The snapshot log has entries at 1515100955770 and 1555100955770.
105/// let snapshot_id = snapshot_id_as_of_time(&metadata, 1_555_100_955_770)?;
106/// assert_eq!(snapshot_id, 3055729675574597004);
107/// assert!(snapshot_id_as_of_time(&metadata, 1_515_100_955_769).is_err());
108/// # Ok(())
109/// # }
110/// ```
111pub fn snapshot_id_as_of_time(table_metadata: &TableMetadataRef, timestamp_ms: i64) -> Result<i64> {
112    table_metadata
113        .history()
114        .iter()
115        .enumerate()
116        .filter(|(_, entry)| entry.timestamp_ms() <= timestamp_ms)
117        .max_by(|(idx_a, a), (idx_b, b)| {
118            a.timestamp_ms()
119                .cmp(&b.timestamp_ms())
120                // Earlier entries win ties, matching Java's timestamp selection.
121                .then(idx_b.cmp(idx_a))
122        })
123        .map(|(_, entry)| entry.snapshot_id)
124        .ok_or_else(|| {
125            Error::new(
126                ErrorKind::DataInvalid,
127                format!("No snapshot history at or before timestamp {timestamp_ms} ms"),
128            )
129        })
130}
131
132#[cfg(test)]
133mod tests {
134    use super::*;
135    use crate::spec::SnapshotLog;
136    use crate::test_utils::scan::TableTestFixture;
137
138    // Five snapshots chained as: S1 (root) -> S2 -> S3 -> S4 -> S5 (current)
139    const S1: i64 = 3051729675574597004;
140    const S2: i64 = 3055729675574597004;
141    const S3: i64 = 3056729675574597004;
142    const S4: i64 = 3057729675574597004;
143    const S5: i64 = 3059729675574597004;
144
145    fn metadata() -> TableMetadataRef {
146        let fixture = TableTestFixture::new_with_deep_history();
147        std::sync::Arc::new(fixture.table.metadata().clone())
148    }
149
150    type History = [(i64, i64)];
151
152    fn metadata_with_history(history: &History) -> TableMetadataRef {
153        let mut metadata = metadata().as_ref().clone();
154        metadata.snapshot_log = history
155            .iter()
156            .map(|&(timestamp_ms, snapshot_id)| SnapshotLog {
157                timestamp_ms,
158                snapshot_id,
159            })
160            .collect();
161        metadata.into()
162    }
163
164    #[test]
165    fn test_snapshot_id_as_of_time() {
166        let cases: &[(&History, i64, i64)] = &[
167            (&[(1000, S1), (2000, S2)], 1000, S1),
168            (&[(1000, S1), (2000, S2)], 1500, S1),
169            (&[(1000, S1), (2000, S2)], 2000, S2),
170            (&[(1000, S1), (2000, S2)], 3000, S2),
171            // Rollback records when an existing snapshot becomes current again.
172            (&[(1000, S1), (2000, S2), (3000, S1)], 3500, S1),
173            // The first entry wins on ties.
174            (&[(1000, S1), (2000, S2), (2000, S3)], 2000, S2),
175            // Clock skew means the log need not be sorted by timestamp.
176            (&[(1000, S1), (3000, S2), (2500, S3)], 3500, S2),
177            (&[(1000, S1), (3000, S2), (2500, S3)], 2600, S3),
178            (&[(-2000, S1), (-1000, S2)], -1500, S1),
179            // Java returns no match here; see the docstring.
180            (&[(i64::MIN, S1), (0, S2)], i64::MIN, S1),
181            (&[(0, S1), (i64::MAX, S2)], i64::MAX, S2),
182        ];
183        for &(history, timestamp_ms, expected) in cases {
184            let metadata = metadata_with_history(history);
185            assert_eq!(
186                snapshot_id_as_of_time(&metadata, timestamp_ms).unwrap(),
187                expected,
188                "timestamp {timestamp_ms}, history {history:?}"
189            );
190        }
191    }
192
193    #[test]
194    fn test_snapshot_id_as_of_time_without_history_at_timestamp() {
195        let cases: &[(&History, i64)] = &[
196            (&[], 1000),
197            (&[(1000, S1)], 999),
198            (&[(3000, S1)], 1500), // Expired prefix; S1 is still retained.
199        ];
200        for &(history, timestamp_ms) in cases {
201            let metadata = metadata_with_history(history);
202            assert!(metadata.snapshot_by_id(S1).is_some());
203            let err = snapshot_id_as_of_time(&metadata, timestamp_ms).unwrap_err();
204            assert_eq!(err.kind(), ErrorKind::DataInvalid);
205            assert!(err.message().contains(&timestamp_ms.to_string()));
206        }
207    }
208
209    #[test]
210    fn test_snapshot_id_as_of_time_ignores_snapshots_outside_main_history() {
211        let mut metadata = metadata().as_ref().clone();
212        metadata.snapshot_log.truncate(2);
213        // Newer retained snapshots may be staged or belong only to another branch.
214        assert!(metadata.snapshot_by_id(S5).is_some());
215        assert_eq!(
216            snapshot_id_as_of_time(&metadata.into(), i64::MAX).unwrap(),
217            S2
218        );
219    }
220
221    // --- ancestors_of ---
222
223    #[test]
224    fn test_ancestors_of_nonexistent_snapshot_returns_empty() {
225        let meta = metadata();
226        let ids: Vec<i64> = ancestors_of(&meta, 999).map(|s| s.snapshot_id()).collect();
227        assert!(ids.is_empty());
228    }
229
230    #[test]
231    fn test_ancestors_of_root_returns_only_root() {
232        let meta = metadata();
233        let ids: Vec<i64> = ancestors_of(&meta, S1).map(|s| s.snapshot_id()).collect();
234        assert_eq!(ids, vec![S1]);
235    }
236
237    #[test]
238    fn test_ancestors_of_leaf_returns_full_chain() {
239        let meta = metadata();
240        let ids: Vec<i64> = ancestors_of(&meta, S5).map(|s| s.snapshot_id()).collect();
241        assert_eq!(ids, vec![S5, S4, S3, S2, S1]);
242    }
243
244    #[test]
245    fn test_ancestors_of_mid_chain_returns_partial_chain() {
246        let meta = metadata();
247        let ids: Vec<i64> = ancestors_of(&meta, S3).map(|s| s.snapshot_id()).collect();
248        assert_eq!(ids, vec![S3, S2, S1]);
249    }
250
251    #[test]
252    fn test_ancestors_of_second_snapshot() {
253        let meta = metadata();
254        let ids: Vec<i64> = ancestors_of(&meta, S2).map(|s| s.snapshot_id()).collect();
255        assert_eq!(ids, vec![S2, S1]);
256    }
257
258    // --- ancestors_between ---
259
260    #[test]
261    fn test_ancestors_between_same_id_returns_empty() {
262        let meta = metadata();
263        let ids: Vec<i64> = ancestors_between(&meta, S3, Some(S3))
264            .map(|s| s.snapshot_id())
265            .collect();
266        assert!(ids.is_empty());
267    }
268
269    #[test]
270    fn test_ancestors_between_no_oldest_returns_all_ancestors() {
271        let meta = metadata();
272        let ids: Vec<i64> = ancestors_between(&meta, S5, None)
273            .map(|s| s.snapshot_id())
274            .collect();
275        assert_eq!(ids, vec![S5, S4, S3, S2, S1]);
276    }
277
278    #[test]
279    fn test_ancestors_between_excludes_oldest_snapshot() {
280        let meta = metadata();
281        // S5 down to (but not including) S2
282        let ids: Vec<i64> = ancestors_between(&meta, S5, Some(S2))
283            .map(|s| s.snapshot_id())
284            .collect();
285        assert_eq!(ids, vec![S5, S4, S3]);
286    }
287
288    #[test]
289    fn test_ancestors_between_adjacent_snapshots() {
290        let meta = metadata();
291        // S3 down to (but not including) S2 — only S3 itself
292        let ids: Vec<i64> = ancestors_between(&meta, S3, Some(S2))
293            .map(|s| s.snapshot_id())
294            .collect();
295        assert_eq!(ids, vec![S3]);
296    }
297
298    #[test]
299    fn test_ancestors_between_leaf_and_root() {
300        let meta = metadata();
301        // S5 down to (but not including) S1
302        let ids: Vec<i64> = ancestors_between(&meta, S5, Some(S1))
303            .map(|s| s.snapshot_id())
304            .collect();
305        assert_eq!(ids, vec![S5, S4, S3, S2]);
306    }
307
308    #[test]
309    fn test_ancestors_between_nonexistent_oldest_returns_full_chain() {
310        let meta = metadata();
311        // oldest_snapshot_id doesn't exist in the chain, so take_while never stops
312        let ids: Vec<i64> = ancestors_between(&meta, S5, Some(999))
313            .map(|s| s.snapshot_id())
314            .collect();
315        assert_eq!(ids, vec![S5, S4, S3, S2, S1]);
316    }
317
318    #[test]
319    fn test_ancestors_between_nonexistent_latest_returns_empty() {
320        let meta = metadata();
321        let ids: Vec<i64> = ancestors_between(&meta, 999, Some(S1))
322            .map(|s| s.snapshot_id())
323            .collect();
324        assert!(ids.is_empty());
325    }
326}
iceberg/util/snapshot.rs

iceberg/util/
snapshot.rs
