Add History metadata table

Author: rshkv

crates/iceberg/src/metadata_scan.rs

@@ -17,6 +17,7 @@
 
 //! Metadata table api.
 
+use std::collections::HashSet;
 use std::sync::Arc;
 
 use arrow_array::builder::{
@@ -26,6 +27,7 @@ use arrow_array::types::{Int32Type, Int64Type, Int8Type, TimestampMillisecondTyp
 use arrow_array::RecordBatch;
 use arrow_schema::{DataType, Field, Fields, Schema, TimeUnit};
 
+use crate::spec::{Snapshot, TableMetadata};
 use crate::table::Table;
 use crate::Result;
 
@@ -44,6 +46,11 @@ impl MetadataTable {
         Self(table)
     }
 
+    /// Get the history table.
+    pub fn history(&self) -> HistoryTable {
+        HistoryTable { table: &self.0 }
+    }
+
     /// Get the snapshots table.
     pub fn snapshots(&self) -> SnapshotsTable {
         SnapshotsTable { table: &self.0 }
@@ -128,6 +135,102 @@ impl<'a> SnapshotsTable<'a> {
     }
 }
 
+/// 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<'a> {
+    table: &'a Table,
+}
+
+impl<'a> HistoryTable<'a> {
+    /// Return the schema of the history table.
+    pub fn schema(&self) -> 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),
+        ])
+    }
+
+    /// Scan the history table.
+    pub fn scan(&self) -> Result<RecordBatch> {
+        let table_metadata = self.table.metadata();
+        let ancestors_by_snapshot_id: HashSet<i64> =
+            SnapshotAncestors::from_current_snapshot(table_metadata)
+                .map(|snapshot| snapshot.snapshot_id())
+                .collect();
+
+        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();
+
+        for snapshot in table_metadata.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_by_snapshot_id.contains(&snapshot.snapshot_id()));
+        }
+
+        Ok(RecordBatch::try_new(Arc::new(Self::schema(self)), vec![
+            Arc::new(made_current_at.finish()),
+            Arc::new(snapshot_id.finish()),
+            Arc::new(parent_id.finish()),
+            Arc::new(is_current_ancestor.finish()),
+        ])?)
+    }
+}
+
+/// Utility to iterate parent-by-parent over the ancestors of a snapshot.
+struct SnapshotAncestors<'a> {
+    table_metadata: &'a TableMetadata,
+    snapshot: Option<&'a Snapshot>,
+}
+
+impl<'a> SnapshotAncestors<'a> {
+    fn from_current_snapshot(table_metadata: &'a TableMetadata) -> Self {
+        SnapshotAncestors {
+            table_metadata,
+            snapshot: table_metadata.current_snapshot().map(|s| s.as_ref()),
+        }
+    }
+}
+
+impl<'a> Iterator for SnapshotAncestors<'a> {
+    type Item = &'a Snapshot;
+
+    /// Return the current `snapshot` and move this iterator to the parent snapshot.
+    fn next(&mut self) -> Option<Self::Item> {
+        if let Some(snapshot) = self.snapshot {
+            let parent = match snapshot.parent_snapshot_id() {
+                Some(parent_snapshot_id) => self
+                    .table_metadata
+                    .snapshot_by_id(parent_snapshot_id)
+                    .map(|s| s.as_ref()),
+                None => None,
+            };
+            self.snapshot = parent;
+            Some(snapshot)
+        } else {
+            None
+        }
+    }
+}
+
 /// Manifests table.
 pub struct ManifestsTable<'a> {
     table: &'a Table,
@@ -381,6 +484,43 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_history_table() {
+        let table = TableTestFixture::new().table;
+        let record_batch = table.metadata_table().history().scan().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"),
+        );
+    }
+
     #[tokio::test]
     async fn test_manifests_table() {
         let mut fixture = TableTestFixture::new();
@@ -451,7 +591,7 @@ mod tests {
                 partition_summaries: ListArray
                 [
                   StructArray
-                -- validity: 
+                -- validity:
                 [
                   valid,
                 ]