iceberg/inspect/
history.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
18//! History table: the chronological log of every snapshot that was ever the
19//! table's current snapshot.
20//!
21//! Each row is one entry of the metadata snapshot-log and answers two questions
22//! about the table's lineage: *when* a snapshot became current, and *whether*
23//! it is still an ancestor of the current state. Snapshots that were rolled back
24//! remain visible in the log yet are flagged `is_current_ancestor = false`, so
25//! the live lineage stays distinguishable from abandoned history.
26//!
27//! References:
28//! - <https://github.com/apache/iceberg/blob/ac865e334e143dfd9e33011d8cf710b46d91f1e5/core/src/main/java/org/apache/iceberg/HistoryTable.java#L50-L54>
29
30use std::collections::HashSet;
31use std::sync::Arc;
32
33use arrow_array::RecordBatch;
34use arrow_array::builder::{BooleanBuilder, PrimitiveBuilder};
35use arrow_array::types::{Int64Type, TimestampMicrosecondType};
36use futures::{StreamExt, stream};
37
38use crate::Result;
39use crate::arrow::{UTC_TIME_ZONE, schema_to_arrow_schema};
40use crate::scan::ArrowRecordBatchStream;
41use crate::spec::{NestedField, PrimitiveType, Type};
42use crate::table::Table;
43use crate::util::snapshot::ancestors_of;
44
45/// History table.
46pub struct HistoryTable<'a> {
47    table: &'a Table,
48}
49
50impl<'a> HistoryTable<'a> {
51    /// Create a new History table instance.
52    pub fn new(table: &'a Table) -> Self {
53        Self { table }
54    }
55
56    /// Returns the iceberg schema of the history table.
57    pub fn schema(&self) -> crate::spec::Schema {
58        let fields = vec![
59            NestedField::required(
60                1,
61                "made_current_at",
62                Type::Primitive(PrimitiveType::Timestamptz),
63            ),
64            NestedField::required(2, "snapshot_id", Type::Primitive(PrimitiveType::Long)),
65            NestedField::optional(3, "parent_id", Type::Primitive(PrimitiveType::Long)),
66            NestedField::required(
67                4,
68                "is_current_ancestor",
69                Type::Primitive(PrimitiveType::Boolean),
70            ),
71        ];
72        crate::spec::Schema::builder()
73            .with_fields(fields.into_iter().map(|f| f.into()))
74            .build()
75            .unwrap()
76    }
77
78    /// Scans the history table.
79    pub async fn scan(&self) -> Result<ArrowRecordBatchStream> {
80        let schema = schema_to_arrow_schema(&self.schema())?;
81        let metadata = self.table.metadata();
82
83        // Walk the current snapshot's parent chain once: membership in this set
84        // is what tells the live lineage apart from rolled-back history entries.
85        let ancestors: HashSet<i64> = metadata
86            .current_snapshot_id()
87            .map(|id| {
88                ancestors_of(&self.table.metadata_ref(), id)
89                    .map(|snapshot| snapshot.snapshot_id())
90                    .collect()
91            })
92            .unwrap_or_default();
93
94        let mut made_current_at =
95            PrimitiveBuilder::<TimestampMicrosecondType>::new().with_timezone(UTC_TIME_ZONE);
96        let mut snapshot_id = PrimitiveBuilder::<Int64Type>::new();
97        let mut parent_id = PrimitiveBuilder::<Int64Type>::new();
98        let mut is_current_ancestor = BooleanBuilder::new();
99
100        for entry in metadata.history() {
101            made_current_at.append_value(entry.timestamp_ms.saturating_mul(1000)); // ms -> µs
102            snapshot_id.append_value(entry.snapshot_id);
103            parent_id.append_option(
104                metadata
105                    .snapshot_by_id(entry.snapshot_id)
106                    .and_then(|snapshot| snapshot.parent_snapshot_id()),
107            );
108            is_current_ancestor.append_value(ancestors.contains(&entry.snapshot_id));
109        }
110
111        let batch = RecordBatch::try_new(Arc::new(schema), vec![
112            Arc::new(made_current_at.finish()),
113            Arc::new(snapshot_id.finish()),
114            Arc::new(parent_id.finish()),
115            Arc::new(is_current_ancestor.finish()),
116        ])?;
117
118        Ok(stream::iter(vec![Ok(batch)]).boxed())
119    }
120}
121
122#[cfg(test)]
123mod tests {
124    use expect_test::expect;
125    use futures::TryStreamExt;
126
127    use crate::TableIdent;
128    use crate::io::FileIO;
129    use crate::scan::tests::TableTestFixture;
130    use crate::spec::TableMetadata;
131    use crate::table::Table;
132    use crate::test_utils::{check_record_batches, test_runtime};
133
134    fn table_from_metadata_json(metadata_json: &str) -> Table {
135        let metadata = serde_json::from_str::<TableMetadata>(metadata_json).unwrap();
136        Table::builder()
137            .metadata(metadata)
138            .identifier(TableIdent::from_strs(["db", "history"]).unwrap())
139            .file_io(FileIO::new_with_fs())
140            .metadata_location("s3://bucket/test/location/metadata/v3.json")
141            .runtime(test_runtime())
142            .build()
143            .unwrap()
144    }
145
146    #[tokio::test]
147    async fn test_history_table() {
148        let table = TableTestFixture::new().table;
149
150        let batch_stream = table.inspect().history().scan().await.unwrap();
151
152        check_record_batches(
153            batch_stream.try_collect::<Vec<_>>().await.unwrap(),
154            expect![[r#"
155                Field { "made_current_at": Timestamp(µs, "+00:00"), metadata: {"PARQUET:field_id": "1"} },
156                Field { "snapshot_id": Int64, metadata: {"PARQUET:field_id": "2"} },
157                Field { "parent_id": nullable Int64, metadata: {"PARQUET:field_id": "3"} },
158                Field { "is_current_ancestor": Boolean, metadata: {"PARQUET:field_id": "4"} }"#]],
159            expect![[r#"
160                made_current_at: PrimitiveArray<Timestamp(µs, "+00:00")>
161                [
162                  2018-01-04T21:22:35.770+00:00,
163                  2019-04-12T20:29:15.770+00:00,
164                ],
165                snapshot_id: PrimitiveArray<Int64>
166                [
167                  3051729675574597004,
168                  3055729675574597004,
169                ],
170                parent_id: PrimitiveArray<Int64>
171                [
172                  null,
173                  3051729675574597004,
174                ],
175                is_current_ancestor: BooleanArray
176                [
177                  true,
178                  true,
179                ]"#]],
180            &[],
181            Some("made_current_at"),
182        );
183    }
184
185    /// A rolled-back snapshot (S2) stays in the log but drops out of the current
186    /// lineage: the current snapshot (S3) descends from S1, so only S1 and S3 are
187    /// current ancestors while S2 is flagged `false`. The log also keeps an entry
188    /// for an expired snapshot (99, absent from `snapshots`) which renders with a
189    /// null parent and `is_current_ancestor = false`, and a second entry for S1
190    /// (a roll-forward makes a snapshot current again) which yields one row per
191    /// log entry.
192    #[tokio::test]
193    async fn test_history_table_with_rolled_back_snapshot() {
194        let metadata_json = r#"{
195            "format-version": 2,
196            "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
197            "location": "s3://bucket/test/location",
198            "last-sequence-number": 3,
199            "last-updated-ms": 1600000000000,
200            "last-column-id": 1,
201            "current-schema-id": 0,
202            "schemas": [
203                {"type": "struct", "schema-id": 0, "fields": [{"id": 1, "name": "x", "required": true, "type": "long"}]}
204            ],
205            "default-spec-id": 0,
206            "partition-specs": [{"spec-id": 0, "fields": []}],
207            "last-partition-id": 999,
208            "default-sort-order-id": 0,
209            "sort-orders": [{"order-id": 0, "fields": []}],
210            "properties": {},
211            "current-snapshot-id": 3,
212            "snapshots": [
213                {
214                    "snapshot-id": 1,
215                    "timestamp-ms": 1515100955770,
216                    "sequence-number": 1,
217                    "summary": {"operation": "append"},
218                    "manifest-list": "s3://bucket/metadata/snap-1.avro",
219                    "schema-id": 0
220                },
221                {
222                    "snapshot-id": 2,
223                    "parent-snapshot-id": 1,
224                    "timestamp-ms": 1555100955770,
225                    "sequence-number": 2,
226                    "summary": {"operation": "append"},
227                    "manifest-list": "s3://bucket/metadata/snap-2.avro",
228                    "schema-id": 0
229                },
230                {
231                    "snapshot-id": 3,
232                    "parent-snapshot-id": 1,
233                    "timestamp-ms": 1600000000000,
234                    "sequence-number": 3,
235                    "summary": {"operation": "append"},
236                    "manifest-list": "s3://bucket/metadata/snap-3.avro",
237                    "schema-id": 0
238                }
239            ],
240            "snapshot-log": [
241                {"snapshot-id": 99, "timestamp-ms": 1500000000000},
242                {"snapshot-id": 1, "timestamp-ms": 1515100955770},
243                {"snapshot-id": 2, "timestamp-ms": 1555100955770},
244                {"snapshot-id": 1, "timestamp-ms": 1580000000000},
245                {"snapshot-id": 3, "timestamp-ms": 1600000000000}
246            ],
247            "metadata-log": [],
248            "refs": {"main": {"snapshot-id": 3, "type": "branch"}}
249        }"#;
250
251        let table = table_from_metadata_json(metadata_json);
252
253        let batch_stream = table.inspect().history().scan().await.unwrap();
254
255        check_record_batches(
256            batch_stream.try_collect::<Vec<_>>().await.unwrap(),
257            expect![[r#"
258                Field { "made_current_at": Timestamp(µs, "+00:00"), metadata: {"PARQUET:field_id": "1"} },
259                Field { "snapshot_id": Int64, metadata: {"PARQUET:field_id": "2"} },
260                Field { "parent_id": nullable Int64, metadata: {"PARQUET:field_id": "3"} },
261                Field { "is_current_ancestor": Boolean, metadata: {"PARQUET:field_id": "4"} }"#]],
262            expect![[r#"
263                made_current_at: PrimitiveArray<Timestamp(µs, "+00:00")>
264                [
265                  2017-07-14T02:40:00+00:00,
266                  2018-01-04T21:22:35.770+00:00,
267                  2019-04-12T20:29:15.770+00:00,
268                  2020-01-26T00:53:20+00:00,
269                  2020-09-13T12:26:40+00:00,
270                ],
271                snapshot_id: PrimitiveArray<Int64>
272                [
273                  99,
274                  1,
275                  2,
276                  1,
277                  3,
278                ],
279                parent_id: PrimitiveArray<Int64>
280                [
281                  null,
282                  null,
283                  1,
284                  null,
285                  1,
286                ],
287                is_current_ancestor: BooleanArray
288                [
289                  false,
290                  true,
291                  false,
292                  true,
293                  true,
294                ]"#]],
295            &[],
296            Some("made_current_at"),
297        );
298    }
299
300    /// Corrupt metadata whose parent pointers form a cycle (S1 and S2 claim each
301    /// other as parent) must not hang the scan: the ancestor traversal visits each
302    /// snapshot once, so both log entries still render, flagged as ancestors.
303    #[tokio::test]
304    async fn test_history_table_with_parent_cycle() {
305        let metadata_json = r#"{
306            "format-version": 2,
307            "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
308            "location": "s3://bucket/test/location",
309            "last-sequence-number": 2,
310            "last-updated-ms": 1555100955770,
311            "last-column-id": 1,
312            "current-schema-id": 0,
313            "schemas": [
314                {"type": "struct", "schema-id": 0, "fields": [{"id": 1, "name": "x", "required": true, "type": "long"}]}
315            ],
316            "default-spec-id": 0,
317            "partition-specs": [{"spec-id": 0, "fields": []}],
318            "last-partition-id": 999,
319            "default-sort-order-id": 0,
320            "sort-orders": [{"order-id": 0, "fields": []}],
321            "properties": {},
322            "current-snapshot-id": 2,
323            "snapshots": [
324                {
325                    "snapshot-id": 1,
326                    "parent-snapshot-id": 2,
327                    "timestamp-ms": 1515100955770,
328                    "sequence-number": 1,
329                    "summary": {"operation": "append"},
330                    "manifest-list": "s3://bucket/metadata/snap-1.avro",
331                    "schema-id": 0
332                },
333                {
334                    "snapshot-id": 2,
335                    "parent-snapshot-id": 1,
336                    "timestamp-ms": 1555100955770,
337                    "sequence-number": 2,
338                    "summary": {"operation": "append"},
339                    "manifest-list": "s3://bucket/metadata/snap-2.avro",
340                    "schema-id": 0
341                }
342            ],
343            "snapshot-log": [
344                {"snapshot-id": 1, "timestamp-ms": 1515100955770},
345                {"snapshot-id": 2, "timestamp-ms": 1555100955770}
346            ],
347            "metadata-log": [],
348            "refs": {"main": {"snapshot-id": 2, "type": "branch"}}
349        }"#;
350
351        let table = table_from_metadata_json(metadata_json);
352
353        let batch_stream = table.inspect().history().scan().await.unwrap();
354
355        check_record_batches(
356            batch_stream.try_collect::<Vec<_>>().await.unwrap(),
357            expect![[r#"
358                Field { "made_current_at": Timestamp(µs, "+00:00"), metadata: {"PARQUET:field_id": "1"} },
359                Field { "snapshot_id": Int64, metadata: {"PARQUET:field_id": "2"} },
360                Field { "parent_id": nullable Int64, metadata: {"PARQUET:field_id": "3"} },
361                Field { "is_current_ancestor": Boolean, metadata: {"PARQUET:field_id": "4"} }"#]],
362            expect![[r#"
363                made_current_at: PrimitiveArray<Timestamp(µs, "+00:00")>
364                [
365                  2018-01-04T21:22:35.770+00:00,
366                  2019-04-12T20:29:15.770+00:00,
367                ],
368                snapshot_id: PrimitiveArray<Int64>
369                [
370                  1,
371                  2,
372                ],
373                parent_id: PrimitiveArray<Int64>
374                [
375                  2,
376                  1,
377                ],
378                is_current_ancestor: BooleanArray
379                [
380                  true,
381                  true,
382                ]"#]],
383            &[],
384            Some("made_current_at"),
385        );
386    }
387
388    /// An empty snapshot-log must yield a zero-row batch that still carries the
389    /// full history schema.
390    #[tokio::test]
391    async fn test_history_table_with_empty_snapshot_log() {
392        let metadata_json = r#"{
393            "format-version": 2,
394            "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
395            "location": "s3://bucket/test/location",
396            "last-sequence-number": 1,
397            "last-updated-ms": 1515100955770,
398            "last-column-id": 1,
399            "current-schema-id": 0,
400            "schemas": [
401                {"type": "struct", "schema-id": 0, "fields": [{"id": 1, "name": "x", "required": true, "type": "long"}]}
402            ],
403            "default-spec-id": 0,
404            "partition-specs": [{"spec-id": 0, "fields": []}],
405            "last-partition-id": 999,
406            "default-sort-order-id": 0,
407            "sort-orders": [{"order-id": 0, "fields": []}],
408            "properties": {},
409            "current-snapshot-id": 1,
410            "snapshots": [
411                {
412                    "snapshot-id": 1,
413                    "timestamp-ms": 1515100955770,
414                    "sequence-number": 1,
415                    "summary": {"operation": "append"},
416                    "manifest-list": "s3://bucket/metadata/snap-1.avro",
417                    "schema-id": 0
418                }
419            ],
420            "snapshot-log": [],
421            "metadata-log": [],
422            "refs": {"main": {"snapshot-id": 1, "type": "branch"}}
423        }"#;
424
425        let table = table_from_metadata_json(metadata_json);
426
427        let batch_stream = table.inspect().history().scan().await.unwrap();
428
429        check_record_batches(
430            batch_stream.try_collect::<Vec<_>>().await.unwrap(),
431            expect![[r#"
432                Field { "made_current_at": Timestamp(µs, "+00:00"), metadata: {"PARQUET:field_id": "1"} },
433                Field { "snapshot_id": Int64, metadata: {"PARQUET:field_id": "2"} },
434                Field { "parent_id": nullable Int64, metadata: {"PARQUET:field_id": "3"} },
435                Field { "is_current_ancestor": Boolean, metadata: {"PARQUET:field_id": "4"} }"#]],
436            expect![[r#"
437                made_current_at: PrimitiveArray<Timestamp(µs, "+00:00")>
438                [
439                ],
440                snapshot_id: PrimitiveArray<Int64>
441                [
442                ],
443                parent_id: PrimitiveArray<Int64>
444                [
445                ],
446                is_current_ancestor: BooleanArray
447                [
448                ]"#]],
449            &[],
450            Some("made_current_at"),
451        );
452    }
453
454    /// Without a current snapshot (`current-snapshot-id: -1`) there is no live
455    /// lineage, so every log entry renders with `is_current_ancestor = false`.
456    #[tokio::test]
457    async fn test_history_table_without_current_snapshot() {
458        let metadata_json = r#"{
459            "format-version": 2,
460            "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
461            "location": "s3://bucket/test/location",
462            "last-sequence-number": 2,
463            "last-updated-ms": 1555100955770,
464            "last-column-id": 1,
465            "current-schema-id": 0,
466            "schemas": [
467                {"type": "struct", "schema-id": 0, "fields": [{"id": 1, "name": "x", "required": true, "type": "long"}]}
468            ],
469            "default-spec-id": 0,
470            "partition-specs": [{"spec-id": 0, "fields": []}],
471            "last-partition-id": 999,
472            "default-sort-order-id": 0,
473            "sort-orders": [{"order-id": 0, "fields": []}],
474            "properties": {},
475            "current-snapshot-id": -1,
476            "snapshots": [
477                {
478                    "snapshot-id": 1,
479                    "timestamp-ms": 1515100955770,
480                    "sequence-number": 1,
481                    "summary": {"operation": "append"},
482                    "manifest-list": "s3://bucket/metadata/snap-1.avro",
483                    "schema-id": 0
484                },
485                {
486                    "snapshot-id": 2,
487                    "parent-snapshot-id": 1,
488                    "timestamp-ms": 1555100955770,
489                    "sequence-number": 2,
490                    "summary": {"operation": "append"},
491                    "manifest-list": "s3://bucket/metadata/snap-2.avro",
492                    "schema-id": 0
493                }
494            ],
495            "snapshot-log": [
496                {"snapshot-id": 1, "timestamp-ms": 1515100955770},
497                {"snapshot-id": 2, "timestamp-ms": 1555100955770}
498            ],
499            "metadata-log": [],
500            "refs": {}
501        }"#;
502
503        let table = table_from_metadata_json(metadata_json);
504
505        let batch_stream = table.inspect().history().scan().await.unwrap();
506
507        check_record_batches(
508            batch_stream.try_collect::<Vec<_>>().await.unwrap(),
509            expect![[r#"
510                Field { "made_current_at": Timestamp(µs, "+00:00"), metadata: {"PARQUET:field_id": "1"} },
511                Field { "snapshot_id": Int64, metadata: {"PARQUET:field_id": "2"} },
512                Field { "parent_id": nullable Int64, metadata: {"PARQUET:field_id": "3"} },
513                Field { "is_current_ancestor": Boolean, metadata: {"PARQUET:field_id": "4"} }"#]],
514            expect![[r#"
515                made_current_at: PrimitiveArray<Timestamp(µs, "+00:00")>
516                [
517                  2018-01-04T21:22:35.770+00:00,
518                  2019-04-12T20:29:15.770+00:00,
519                ],
520                snapshot_id: PrimitiveArray<Int64>
521                [
522                  1,
523                  2,
524                ],
525                parent_id: PrimitiveArray<Int64>
526                [
527                  null,
528                  1,
529                ],
530                is_current_ancestor: BooleanArray
531                [
532                  false,
533                  false,
534                ]"#]],
535            &[],
536            Some("made_current_at"),
537        );
538    }
539}
iceberg/inspect/history.rs

iceberg/inspect/
history.rs
