Add History metadata table

Author: rshkv

Diff for: crates/iceberg/src/metadata_scan.rs

@@ -17,14 +17,15 @@
 
 //! Metadata table api.
 
+use std::collections::HashSet;
 use std::sync::Arc;
 
-use arrow_array::builder::{MapBuilder, PrimitiveBuilder, StringBuilder};
+use arrow_array::builder::{BooleanBuilder, MapBuilder, PrimitiveBuilder, StringBuilder};
 use arrow_array::types::{Int64Type, TimestampMillisecondType};
 use arrow_array::RecordBatch;
 use arrow_schema::{DataType, Field, Schema, TimeUnit};
 
-use crate::spec::TableMetadataRef;
+use crate::spec::{SnapshotRef, TableMetadataRef};
 use crate::table::Table;
 use crate::Result;
 
@@ -50,6 +51,11 @@ impl MetadataScan {
     pub fn snapshots(&self) -> Result<RecordBatch> {
         SnapshotsTable::scan(self)
     }
+
+    /// Return the history of the table.
+    pub fn history(&self) -> Result<RecordBatch> {
+        HistoryTable::scan(self)
+    }
 }
 
 /// Table metadata scan.
@@ -137,6 +143,93 @@ impl MetadataTable for SnapshotsTable {
     }
 }
 
+/// History table.
+///
+/// Shows how the table's current snapshot has changed over time and when each
+/// snapshot became the current snapshot.
+///
+/// Unlike the [Snapshots][SnapshotsTable], this metadata table has less detail
+/// per snapshot but includes ancestry information of the current snapshot.
+///
+/// `is_current_ancestor` indicates whether the snapshot is an ancestor of the
+/// current snapshot. If `false`, then the snapshot was rolled back.
+pub struct HistoryTable;
+
+impl MetadataTable for HistoryTable {
+    fn schema() -> Schema {
+        Schema::new(vec![
+            Field::new(
+                "made_current_at",
+                DataType::Timestamp(TimeUnit::Millisecond, Some("+00:00".into())),
+                false,
+            ),
+            Field::new("snapshot_id", DataType::Int64, false),
+            Field::new("parent_id", DataType::Int64, true),
+            Field::new("is_current_ancestor", DataType::Boolean, false),
+        ])
+    }
+
+    fn scan(scan: &MetadataScan) -> Result<RecordBatch> {
+        let mut made_current_at =
+            PrimitiveBuilder::<TimestampMillisecondType>::new().with_timezone("+00:00");
+        let mut snapshot_id = PrimitiveBuilder::<Int64Type>::new();
+        let mut parent_id = PrimitiveBuilder::<Int64Type>::new();
+        let mut is_current_ancestor = BooleanBuilder::new();
+
+        let ancestors: HashSet<i64> =
+            Ancestors::new(scan.metadata_ref.current_snapshot(), &scan.metadata_ref)
+                .map(|snapshot| snapshot.snapshot_id())
+                .collect();
+
+        for snapshot in scan.metadata_ref.snapshots() {
+            made_current_at.append_value(snapshot.timestamp_ms());
+            snapshot_id.append_value(snapshot.snapshot_id());
+            parent_id.append_option(snapshot.parent_snapshot_id());
+            is_current_ancestor.append_value(ancestors.contains(&snapshot.snapshot_id()));
+        }
+
+        Ok(RecordBatch::try_new(Arc::new(Self::schema()), vec![
+            Arc::new(made_current_at.finish()),
+            Arc::new(snapshot_id.finish()),
+            Arc::new(parent_id.finish()),
+            Arc::new(is_current_ancestor.finish()),
+        ])?)
+    }
+}
+
+struct Ancestors<'a> {
+    current_snapshot: Option<&'a SnapshotRef>,
+    table_metadata: &'a TableMetadataRef,
+}
+
+impl<'a> Ancestors<'a> {
+    fn new(
+        current_snapshot: Option<&'a SnapshotRef>,
+        table_metadata: &'a TableMetadataRef,
+    ) -> Self {
+        Ancestors {
+            current_snapshot,
+            table_metadata,
+        }
+    }
+}
+
+impl<'a> Iterator for Ancestors<'a> {
+    type Item = &'a SnapshotRef;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if let Some(snapshot) = self.current_snapshot {
+            self.current_snapshot = match snapshot.parent_snapshot_id() {
+                Some(parent_snapshot_id) => self.table_metadata.snapshot_by_id(parent_snapshot_id),
+                None => None,
+            };
+            Some(snapshot)
+        } else {
+            None
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use expect_test::{expect, Expect};
@@ -262,4 +355,41 @@ mod tests {
             Some("committed_at"),
         );
     }
+
+    #[test]
+    fn test_history_table() {
+        let table = TableTestFixture::new().table;
+        let record_batch = table.metadata_scan().history().unwrap();
+        check_record_batch(
+            record_batch,
+            expect![[r#"
+                Field { name: "made_current_at", data_type: Timestamp(Millisecond, Some("+00:00")), nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} },
+                Field { name: "snapshot_id", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} },
+                Field { name: "parent_id", data_type: Int64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} },
+                Field { name: "is_current_ancestor", data_type: Boolean, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }"#]],
+            expect![[r#"
+                made_current_at: PrimitiveArray<Timestamp(Millisecond, Some("+00:00"))>
+                [
+                  2018-01-04T21:22:35.770+00:00,
+                  2019-04-12T20:29:15.770+00:00,
+                ],
+                snapshot_id: PrimitiveArray<Int64>
+                [
+                  3051729675574597004,
+                  3055729675574597004,
+                ],
+                parent_id: PrimitiveArray<Int64>
+                [
+                  null,
+                  3051729675574597004,
+                ],
+                is_current_ancestor: BooleanArray
+                [
+                  true,
+                  true,
+                ]"#]],
+            &[],
+            Some("made_current_at"),
+        );
+    }
 }