From 0cf4c2a311336479489c5b6973c6a98a31114397 Mon Sep 17 00:00:00 2001 From: Gustavo Schneiter Date: Sat, 11 Jul 2026 19:28:48 -0300 Subject: [PATCH] feat: add refs metadata table --- crates/iceberg/public-api.txt | 7 + crates/iceberg/src/inspect/metadata_table.rs | 11 +- crates/iceberg/src/inspect/mod.rs | 2 + crates/iceberg/src/inspect/refs.rs | 351 ++++++++++++++++++ .../datafusion/src/table/metadata_table.rs | 2 + .../tests/integration_datafusion_test.rs | 1 + .../testdata/slts/df_test/insert_into.slt | 4 + .../testdata/slts/df_test/show_tables.slt | 2 + 8 files changed, 379 insertions(+), 1 deletion(-) create mode 100644 crates/iceberg/src/inspect/refs.rs diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index c2013294e3..6074546121 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -622,6 +622,7 @@ pub type iceberg::expr::Term = iceberg::expr::Reference pub mod iceberg::inspect pub enum iceberg::inspect::MetadataTableType pub iceberg::inspect::MetadataTableType::Manifests +pub iceberg::inspect::MetadataTableType::Refs pub iceberg::inspect::MetadataTableType::Snapshots impl iceberg::inspect::MetadataTableType pub fn iceberg::inspect::MetadataTableType::all_types() -> impl core::iter::traits::iterator::Iterator @@ -645,6 +646,7 @@ pub struct iceberg::inspect::MetadataTable<'a>(_) impl<'a> iceberg::inspect::MetadataTable<'a> pub fn iceberg::inspect::MetadataTable<'a>::manifests(&self) -> iceberg::inspect::ManifestsTable<'_> pub fn iceberg::inspect::MetadataTable<'a>::new(table: &'a iceberg::table::Table) -> Self +pub fn iceberg::inspect::MetadataTable<'a>::refs(&self) -> iceberg::inspect::RefsTable<'_> pub fn iceberg::inspect::MetadataTable<'a>::snapshots(&self) -> iceberg::inspect::SnapshotsTable<'_> impl<'a> core::fmt::Debug for iceberg::inspect::MetadataTable<'a> pub fn iceberg::inspect::MetadataTable<'a>::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result @@ -663,6 +665,11 @@ pub fn iceberg::inspect::MetadataTableTypeIter::next(&mut self) -> core::option: pub fn iceberg::inspect::MetadataTableTypeIter::nth(&mut self, n: usize) -> core::option::Option<::Item> pub fn iceberg::inspect::MetadataTableTypeIter::size_hint(&self) -> (usize, core::option::Option) impl core::iter::traits::marker::FusedIterator for iceberg::inspect::MetadataTableTypeIter +pub struct iceberg::inspect::RefsTable<'a> +impl<'a> iceberg::inspect::RefsTable<'a> +pub fn iceberg::inspect::RefsTable<'a>::new(table: &'a iceberg::table::Table) -> Self +pub async fn iceberg::inspect::RefsTable<'a>::scan(&self) -> iceberg::Result +pub fn iceberg::inspect::RefsTable<'a>::schema(&self) -> iceberg::spec::Schema pub struct iceberg::inspect::SnapshotsTable<'a> impl<'a> iceberg::inspect::SnapshotsTable<'a> pub fn iceberg::inspect::SnapshotsTable<'a>::new(table: &'a iceberg::table::Table) -> Self diff --git a/crates/iceberg/src/inspect/metadata_table.rs b/crates/iceberg/src/inspect/metadata_table.rs index d5e9d60869..c66ad2a984 100644 --- a/crates/iceberg/src/inspect/metadata_table.rs +++ b/crates/iceberg/src/inspect/metadata_table.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use super::{ManifestsTable, SnapshotsTable}; +use super::{ManifestsTable, RefsTable, SnapshotsTable}; use crate::table::Table; /// Metadata table is used to inspect a table's history, snapshots, and other metadata as a table. @@ -34,6 +34,8 @@ pub enum MetadataTableType { Snapshots, /// [`ManifestsTable`] Manifests, + /// [`RefsTable`] + Refs, } impl MetadataTableType { @@ -42,6 +44,7 @@ impl MetadataTableType { match self { MetadataTableType::Snapshots => "snapshots", MetadataTableType::Manifests => "manifests", + MetadataTableType::Refs => "refs", } } @@ -59,6 +62,7 @@ impl TryFrom<&str> for MetadataTableType { match value { "snapshots" => Ok(Self::Snapshots), "manifests" => Ok(Self::Manifests), + "refs" => Ok(Self::Refs), _ => Err(format!("invalid metadata table type: {value}")), } } @@ -79,4 +83,9 @@ impl<'a> MetadataTable<'a> { pub fn manifests(&self) -> ManifestsTable<'_> { ManifestsTable::new(self.0) } + + /// Get the refs table. + pub fn refs(&self) -> RefsTable<'_> { + RefsTable::new(self.0) + } } diff --git a/crates/iceberg/src/inspect/mod.rs b/crates/iceberg/src/inspect/mod.rs index b64420ea11..f531c5f0af 100644 --- a/crates/iceberg/src/inspect/mod.rs +++ b/crates/iceberg/src/inspect/mod.rs @@ -19,8 +19,10 @@ mod manifests; mod metadata_table; +mod refs; mod snapshots; pub use manifests::ManifestsTable; pub use metadata_table::*; +pub use refs::RefsTable; pub use snapshots::SnapshotsTable; diff --git a/crates/iceberg/src/inspect/refs.rs b/crates/iceberg/src/inspect/refs.rs new file mode 100644 index 0000000000..d2e9fbbd62 --- /dev/null +++ b/crates/iceberg/src/inspect/refs.rs @@ -0,0 +1,351 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::BTreeMap; +use std::sync::Arc; + +use arrow_array::RecordBatch; +use arrow_array::builder::{PrimitiveBuilder, StringBuilder}; +use arrow_array::types::{Int32Type, Int64Type}; +use futures::{StreamExt, stream}; + +use crate::Result; +use crate::arrow::schema_to_arrow_schema; +use crate::scan::ArrowRecordBatchStream; +use crate::spec::{NestedField, PrimitiveType, SnapshotRetention, Type}; +use crate::table::Table; + +/// Refs table. +/// +/// Contains one row for every named reference in the table metadata's `refs` map — branches +/// and tags — sorted by reference name. Each row carries the referenced snapshot id, the +/// reference type (`"BRANCH"` or `"TAG"`, uppercase, matching the Java implementation rather +/// than the lowercase spec serialization), and the retention fields: `max_reference_age_in_ms` +/// for both kinds, plus `min_snapshots_to_keep` and `max_snapshot_age_in_ms` for branches only +/// (null for tags). +/// +/// Reference: +pub struct RefsTable<'a> { + table: &'a Table, +} + +impl<'a> RefsTable<'a> { + /// Create a new Refs table instance. + pub fn new(table: &'a Table) -> Self { + Self { table } + } + + /// Returns the iceberg schema of the refs table. + pub fn schema(&self) -> crate::spec::Schema { + let fields = vec![ + NestedField::required(1, "name", Type::Primitive(PrimitiveType::String)), + NestedField::required(2, "type", Type::Primitive(PrimitiveType::String)), + NestedField::required(3, "snapshot_id", Type::Primitive(PrimitiveType::Long)), + NestedField::optional( + 4, + "max_reference_age_in_ms", + Type::Primitive(PrimitiveType::Long), + ), + NestedField::optional( + 5, + "min_snapshots_to_keep", + Type::Primitive(PrimitiveType::Int), + ), + NestedField::optional( + 6, + "max_snapshot_age_in_ms", + Type::Primitive(PrimitiveType::Long), + ), + ]; + crate::spec::Schema::builder() + .with_fields(fields.into_iter().map(|f| f.into())) + .build() + .unwrap() + } + + /// Scans the refs table. + pub async fn scan(&self) -> Result { + let schema = schema_to_arrow_schema(&self.schema())?; + + let mut name = StringBuilder::new(); + let mut ref_type = StringBuilder::new(); + let mut snapshot_id = PrimitiveBuilder::::new(); + let mut max_reference_age_in_ms = PrimitiveBuilder::::new(); + let mut min_snapshots_to_keep = PrimitiveBuilder::::new(); + let mut max_snapshot_age_in_ms = PrimitiveBuilder::::new(); + + // Collect into a `BTreeMap` first so the resulting rows are ordered + // deterministically by ref name, since `TableMetadata::refs` is a `HashMap`. + let refs: BTreeMap<_, _> = self.table.metadata().refs.iter().collect(); + for (ref_name, snapshot_ref) in refs { + name.append_value(ref_name); + snapshot_id.append_value(snapshot_ref.snapshot_id); + match &snapshot_ref.retention { + SnapshotRetention::Branch { + min_snapshots_to_keep: min_snapshots, + max_snapshot_age_ms, + max_ref_age_ms, + } => { + ref_type.append_value("BRANCH"); + max_reference_age_in_ms.append_option(*max_ref_age_ms); + min_snapshots_to_keep.append_option(*min_snapshots); + max_snapshot_age_in_ms.append_option(*max_snapshot_age_ms); + } + SnapshotRetention::Tag { max_ref_age_ms } => { + ref_type.append_value("TAG"); + max_reference_age_in_ms.append_option(*max_ref_age_ms); + min_snapshots_to_keep.append_null(); + max_snapshot_age_in_ms.append_null(); + } + } + } + + let batch = RecordBatch::try_new(Arc::new(schema), vec![ + Arc::new(name.finish()), + Arc::new(ref_type.finish()), + Arc::new(snapshot_id.finish()), + Arc::new(max_reference_age_in_ms.finish()), + Arc::new(min_snapshots_to_keep.finish()), + Arc::new(max_snapshot_age_in_ms.finish()), + ])?; + + Ok(stream::iter(vec![Ok(batch)]).boxed()) + } +} + +#[cfg(test)] +mod tests { + use expect_test::expect; + use futures::TryStreamExt; + use tempfile::TempDir; + + use crate::TableIdent; + use crate::io::FileIO; + use crate::scan::tests::TableTestFixture; + use crate::spec::TableMetadata; + use crate::table::Table; + use crate::test_utils::{check_record_batches, test_runtime}; + + #[tokio::test] + async fn test_refs_table() { + let table = TableTestFixture::new().table; + + let batch_stream = table.inspect().refs().scan().await.unwrap(); + + check_record_batches( + batch_stream.try_collect::>().await.unwrap(), + expect![[r#" + Field { "name": Utf8, metadata: {"PARQUET:field_id": "1"} }, + Field { "type": Utf8, metadata: {"PARQUET:field_id": "2"} }, + Field { "snapshot_id": Int64, metadata: {"PARQUET:field_id": "3"} }, + Field { "max_reference_age_in_ms": nullable Int64, metadata: {"PARQUET:field_id": "4"} }, + Field { "min_snapshots_to_keep": nullable Int32, metadata: {"PARQUET:field_id": "5"} }, + Field { "max_snapshot_age_in_ms": nullable Int64, metadata: {"PARQUET:field_id": "6"} }"#]], + expect![[r#" + name: StringArray + [ + "main", + "test", + ], + type: StringArray + [ + "BRANCH", + "TAG", + ], + snapshot_id: PrimitiveArray + [ + 3055729675574597004, + 3051729675574597004, + ], + max_reference_age_in_ms: PrimitiveArray + [ + null, + 10000000, + ], + min_snapshots_to_keep: PrimitiveArray + [ + null, + null, + ], + max_snapshot_age_in_ms: PrimitiveArray + [ + null, + null, + ]"#]], + &[], + Some("name"), + ); + } + + #[tokio::test] + async fn test_refs_table_retention_fields() { + // Same shape as `testdata/example_table_metadata_v2.json`, but with the + // "main" branch carrying explicit retention settings instead of the + // implicit ref that `TableMetadata::try_normalize` would construct. + let tmp_dir = TempDir::new().unwrap(); + let table_location = tmp_dir.path().join("table1"); + let table_metadata1_location = table_location.join("metadata/v1.json"); + + let metadata_json = format!( + r#"{{ + "format-version": 2, + "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1", + "location": "{table_location}", + "last-sequence-number": 34, + "last-updated-ms": 1602638573590, + "last-column-id": 3, + "current-schema-id": 1, + "schemas": [ + {{ + "type": "struct", + "schema-id": 0, + "fields": [ + {{"id": 1, "name": "x", "required": true, "type": "long"}} + ]}}, + {{ + "type": "struct", + "schema-id": 1, + "identifier-field-ids": [1, 2], + "fields": [ + {{"id": 1, "name": "x", "required": true, "type": "long"}}, + {{"id": 2, "name": "y", "required": true, "type": "long", "doc": "comment"}}, + {{"id": 3, "name": "z", "required": true, "type": "long"}}, + {{"id": 4, "name": "a", "required": true, "type": "string"}}, + {{"id": 5, "name": "dbl", "required": true, "type": "double"}}, + {{"id": 6, "name": "i32", "required": true, "type": "int"}}, + {{"id": 7, "name": "i64", "required": true, "type": "long"}}, + {{"id": 8, "name": "bool", "required": true, "type": "boolean"}} + ] + }} + ], + "default-spec-id": 0, + "partition-specs": [ + {{ + "spec-id": 0, + "fields": [ + {{"name": "x", "transform": "identity", "source-id": 1, "field-id": 1000}} + ] + }} + ], + "last-partition-id": 1000, + "default-sort-order-id": 3, + "sort-orders": [ + {{ + "order-id": 3, + "fields": [ + {{"transform": "identity", "source-id": 2, "direction": "asc", "null-order": "nulls-first"}}, + {{"transform": "bucket[4]", "source-id": 3, "direction": "desc", "null-order": "nulls-last"}} + ] + }} + ], + "properties": {{"read.split.target.size": "134217728"}}, + "current-snapshot-id": 3055729675574597004, + "snapshots": [ + {{ + "snapshot-id": 3051729675574597004, + "timestamp-ms": 1515100955770, + "sequence-number": 0, + "summary": {{"operation": "append"}}, + "manifest-list": "{table_location}/metadata/manifests_list_1.avro" + }}, + {{ + "snapshot-id": 3055729675574597004, + "parent-snapshot-id": 3051729675574597004, + "timestamp-ms": 1555100955770, + "sequence-number": 1, + "summary": {{"operation": "append"}}, + "manifest-list": "{table_location}/metadata/manifests_list_2.avro", + "schema-id": 1 + }} + ], + "snapshot-log": [ + {{"snapshot-id": 3051729675574597004, "timestamp-ms": 1515100955770}}, + {{"snapshot-id": 3055729675574597004, "timestamp-ms": 1555100955770}} + ], + "metadata-log": [{{"metadata-file": "{table_location}/metadata/v1.json", "timestamp-ms": 1515100}}], + "refs": {{ + "main": {{ + "snapshot-id": 3055729675574597004, + "type": "branch", + "min-snapshots-to-keep": 5, + "max-snapshot-age-ms": 86400000, + "max-ref-age-ms": 259200000 + }}, + "test": {{"snapshot-id": 3051729675574597004, "type": "tag", "max-ref-age-ms": 10000000}} + }} + }}"#, + table_location = table_location.to_str().unwrap(), + ); + + let table_metadata = serde_json::from_str::(&metadata_json).unwrap(); + + let table = Table::builder() + .metadata(table_metadata) + .identifier(TableIdent::from_strs(["db", "table1"]).unwrap()) + .file_io(FileIO::new_with_fs()) + .metadata_location(table_metadata1_location.as_os_str().to_str().unwrap()) + .runtime(test_runtime()) + .build() + .unwrap(); + + let batch_stream = table.inspect().refs().scan().await.unwrap(); + + check_record_batches( + batch_stream.try_collect::>().await.unwrap(), + expect![[r#" + Field { "name": Utf8, metadata: {"PARQUET:field_id": "1"} }, + Field { "type": Utf8, metadata: {"PARQUET:field_id": "2"} }, + Field { "snapshot_id": Int64, metadata: {"PARQUET:field_id": "3"} }, + Field { "max_reference_age_in_ms": nullable Int64, metadata: {"PARQUET:field_id": "4"} }, + Field { "min_snapshots_to_keep": nullable Int32, metadata: {"PARQUET:field_id": "5"} }, + Field { "max_snapshot_age_in_ms": nullable Int64, metadata: {"PARQUET:field_id": "6"} }"#]], + expect![[r#" + name: StringArray + [ + "main", + "test", + ], + type: StringArray + [ + "BRANCH", + "TAG", + ], + snapshot_id: PrimitiveArray + [ + 3055729675574597004, + 3051729675574597004, + ], + max_reference_age_in_ms: PrimitiveArray + [ + 259200000, + 10000000, + ], + min_snapshots_to_keep: PrimitiveArray + [ + 5, + null, + ], + max_snapshot_age_in_ms: PrimitiveArray + [ + 86400000, + null, + ]"#]], + &[], + Some("name"), + ); + } +} diff --git a/crates/integrations/datafusion/src/table/metadata_table.rs b/crates/integrations/datafusion/src/table/metadata_table.rs index 43e069cf5a..9f964eeea4 100644 --- a/crates/integrations/datafusion/src/table/metadata_table.rs +++ b/crates/integrations/datafusion/src/table/metadata_table.rs @@ -49,6 +49,7 @@ impl TableProvider for IcebergMetadataTableProvider { let schema = match self.r#type { MetadataTableType::Snapshots => metadata_table.snapshots().schema(), MetadataTableType::Manifests => metadata_table.manifests().schema(), + MetadataTableType::Refs => metadata_table.refs().schema(), }; schema_to_arrow_schema(&schema).unwrap().into() } @@ -74,6 +75,7 @@ impl IcebergMetadataTableProvider { let stream = match self.r#type { MetadataTableType::Snapshots => metadata_table.snapshots().scan().await, MetadataTableType::Manifests => metadata_table.manifests().scan().await, + MetadataTableType::Refs => metadata_table.refs().scan().await, } .map_err(to_datafusion_error)?; let stream = stream.map_err(to_datafusion_error); diff --git a/crates/integrations/datafusion/tests/integration_datafusion_test.rs b/crates/integrations/datafusion/tests/integration_datafusion_test.rs index cebac75dd9..a7e3ef18cb 100644 --- a/crates/integrations/datafusion/tests/integration_datafusion_test.rs +++ b/crates/integrations/datafusion/tests/integration_datafusion_test.rs @@ -173,6 +173,7 @@ async fn test_provider_list_table_names() -> Result<()> { "my_table", "my_table$snapshots", "my_table$manifests", + "my_table$refs", ] "#]] .assert_debug_eq(&result); diff --git a/crates/sqllogictest/testdata/slts/df_test/insert_into.slt b/crates/sqllogictest/testdata/slts/df_test/insert_into.slt index 56fc1fefb1..0eb762751c 100644 --- a/crates/sqllogictest/testdata/slts/df_test/insert_into.slt +++ b/crates/sqllogictest/testdata/slts/df_test/insert_into.slt @@ -39,6 +39,10 @@ query ?????? SELECT * FROM default.default.test_unpartitioned_table$snapshots ---- +query ?????? +SELECT * FROM default.default.test_unpartitioned_table$refs +---- + # Insert a single row and verify the count query I INSERT INTO default.default.test_unpartitioned_table VALUES (1, 'Alice') diff --git a/crates/sqllogictest/testdata/slts/df_test/show_tables.slt b/crates/sqllogictest/testdata/slts/df_test/show_tables.slt index a0f0e55b5b..5822f93782 100644 --- a/crates/sqllogictest/testdata/slts/df_test/show_tables.slt +++ b/crates/sqllogictest/testdata/slts/df_test/show_tables.slt @@ -27,9 +27,11 @@ datafusion information_schema tables VIEW datafusion information_schema views VIEW default default test_binary_table BASE TABLE default default test_binary_table$manifests BASE TABLE +default default test_binary_table$refs BASE TABLE default default test_binary_table$snapshots BASE TABLE default default test_partitioned_table BASE TABLE default default test_partitioned_table$manifests BASE TABLE +default default test_partitioned_table$refs BASE TABLE default default test_partitioned_table$snapshots BASE TABLE default information_schema columns VIEW default information_schema df_settings VIEW