nordisk/skade
vendor/iceberg-arrow59/src/spec/table_metadata.rs
// 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.
//! Defines the [table metadata](https://iceberg.apache.org/spec/#table-metadata).
//! The main struct here is [TableMetadataV2] which defines the data for a table.
use std::cmp::Ordering;
use std::collections::HashMap;
use std::fmt::{Display, Formatter};
use std::hash::Hash;
use std::sync::Arc;
use _serde::TableMetadataEnum;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_repr::{Deserialize_repr, Serialize_repr};
use uuid::Uuid;
use super::snapshot::SnapshotReference;
pub use super::table_metadata_builder::{TableMetadataBuildResult, TableMetadataBuilder};
use super::{
DEFAULT_PARTITION_SPEC_ID, PartitionSpecRef, PartitionStatisticsFile, Schema, SchemaId,
SchemaRef, SnapshotRef, SnapshotRetention, SortOrder, SortOrderRef, StatisticsFile, StructType,
TableProperties, Transform,
};
use crate::catalog::{METADATA_FOLDER_NAME, MetadataLocation};
use crate::compression::CompressionCodec;
use crate::error::{Result, invalid_data, timestamp_ms_to_utc};
use crate::io::FileIO;
use crate::partitioning::compute_unified_partition_type;
use crate::spec::EncryptedKey;
use crate::{Error, ErrorKind};
static MAIN_BRANCH: &str = "main";
pub(crate) static ONE_MINUTE_MS: i64 = 60_000;
/// Sentinel value used by the Java implementation and older metadata files
/// to represent a missing/empty current snapshot ID. During deserialization,
/// this value is normalized to `None`.
pub(crate) static EMPTY_SNAPSHOT_ID: i64 = -1;
pub(crate) static INITIAL_SEQUENCE_NUMBER: i64 = 0;
/// Initial row id for row lineage for new v3 tables and older tables upgrading to v3.
pub const INITIAL_ROW_ID: u64 = 0;
/// Minimum format version that supports row lineage (v3).
pub const MIN_FORMAT_VERSION_ROW_LINEAGE: FormatVersion = FormatVersion::V3;
/// Reference to [`TableMetadata`].
pub type TableMetadataRef = Arc<TableMetadata>;
#[derive(Debug, PartialEq, Deserialize, Eq, Clone)]
#[serde(try_from = "TableMetadataEnum")]
/// Fields for the version 2 of the table metadata.
///
/// We assume that this data structure is always valid, so we will panic when invalid error happens.
/// We check the validity of this data structure when constructing.
pub struct TableMetadata {
/// Integer Version for the format.
pub(crate) format_version: FormatVersion,
/// A UUID that identifies the table
pub(crate) table_uuid: Uuid,
/// Location tables base location
pub(crate) location: String,
/// The tables highest sequence number
pub(crate) last_sequence_number: i64,
/// Timestamp in milliseconds from the unix epoch when the table was last updated.
pub(crate) last_updated_ms: i64,
/// An integer; the highest assigned column ID for the table.
pub(crate) last_column_id: i32,
/// A list of schemas, stored as objects with schema-id.
pub(crate) schemas: HashMap<i32, SchemaRef>,
/// ID of the table’s current schema.
pub(crate) current_schema_id: i32,
/// A list of partition specs, stored as full partition spec objects.
pub(crate) partition_specs: HashMap<i32, PartitionSpecRef>,
/// ID of the “current” spec that writers should use by default.
pub(crate) default_spec: PartitionSpecRef,
/// Partition type of the default partition spec.
pub(crate) default_partition_type: StructType,
/// An integer; the highest assigned partition field ID across all partition specs for the table.
pub(crate) last_partition_id: i32,
///A string to string map of table properties. This is used to control settings that
/// affect reading and writing and is not intended to be used for arbitrary metadata.
/// For example, commit.retry.num-retries is used to control the number of commit retries.
pub(crate) properties: HashMap<String, String>,
/// long ID of the current table snapshot; must be the same as the current
/// ID of the main branch in refs.
pub(crate) current_snapshot_id: Option<i64>,
///A list of valid snapshots. Valid snapshots are snapshots for which all
/// data files exist in the file system. A data file must not be deleted
/// from the file system until the last snapshot in which it was listed is
/// garbage collected.
pub(crate) snapshots: HashMap<i64, SnapshotRef>,
/// A list (optional) of timestamp and snapshot ID pairs that encodes changes
/// to the current snapshot for the table. Each time the current-snapshot-id
/// is changed, a new entry should be added with the last-updated-ms
/// and the new current-snapshot-id. When snapshots are expired from
/// the list of valid snapshots, all entries before a snapshot that has
/// expired should be removed.
pub(crate) snapshot_log: Vec<SnapshotLog>,
/// A list (optional) of timestamp and metadata file location pairs
/// that encodes changes to the previous metadata files for the table.
/// Each time a new metadata file is created, a new entry of the
/// previous metadata file location should be added to the list.
/// Tables can be configured to remove the oldest metadata log entries and
/// keep a fixed-size log of the most recent entries after a commit.
pub(crate) metadata_log: Vec<MetadataLog>,
/// A list of sort orders, stored as full sort order objects.
pub(crate) sort_orders: HashMap<i64, SortOrderRef>,
/// Default sort order id of the table. Note that this could be used by
/// writers, but is not used when reading because reads use the specs
/// stored in manifest files.
pub(crate) default_sort_order_id: i64,
/// A map of snapshot references. The map keys are the unique snapshot reference
/// names in the table, and the map values are snapshot reference objects.
/// There is always a main branch reference pointing to the current-snapshot-id
/// even if the refs map is null.
pub(crate) refs: HashMap<String, SnapshotReference>,
/// Mapping of snapshot ids to statistics files.
pub(crate) statistics: HashMap<i64, StatisticsFile>,
/// Mapping of snapshot ids to partition statistics files.
pub(crate) partition_statistics: HashMap<i64, PartitionStatisticsFile>,
/// Encryption Keys - map of key id to the actual key
pub(crate) encryption_keys: HashMap<String, EncryptedKey>,
/// Next row id to be assigned for Row Lineage (v3)
pub(crate) next_row_id: u64,
}
impl TableMetadata {
/// Convert this Table Metadata into a builder for modification.
///
/// `current_file_location` is the location where the current version
/// of the metadata file is stored. This is used to update the metadata log.
/// If `current_file_location` is `None`, the metadata log will not be updated.
/// This should only be used to stage-create tables.
#[must_use]
pub fn into_builder(self, current_file_location: Option<String>) -> TableMetadataBuilder {
TableMetadataBuilder::new_from_metadata(self, current_file_location)
}
/// Check if a partition field name exists in any partition spec.
#[inline]
pub(crate) fn partition_name_exists(&self, name: &str) -> bool {
self.partition_specs
.values()
.any(|spec| spec.fields().iter().any(|pf| pf.name == name))
}
/// Check if a field name exists in any schema.
#[inline]
pub(crate) fn name_exists_in_any_schema(&self, name: &str) -> bool {
self.schemas
.values()
.any(|schema| schema.field_by_name(name).is_some())
}
/// Returns format version of this metadata.
#[inline]
pub fn format_version(&self) -> FormatVersion {
self.format_version
}
/// Returns uuid of current table.
#[inline]
pub fn uuid(&self) -> Uuid {
self.table_uuid
}
/// Returns table location.
#[inline]
pub fn location(&self) -> &str {
self.location.as_str()
}
/// Returns last sequence number.
#[inline]
pub fn last_sequence_number(&self) -> i64 {
self.last_sequence_number
}
/// Returns the next sequence number for the table.
///
/// For format version 1, it always returns the initial sequence number.
/// For other versions, it returns the last sequence number incremented by 1.
#[inline]
pub fn next_sequence_number(&self) -> i64 {
match self.format_version {
FormatVersion::V1 => INITIAL_SEQUENCE_NUMBER,
_ => self.last_sequence_number + 1,
}
}
/// Returns the last column id.
#[inline]
pub fn last_column_id(&self) -> i32 {
self.last_column_id
}
/// Returns the last partition_id
#[inline]
pub fn last_partition_id(&self) -> i32 {
self.last_partition_id
}
/// Returns last updated time.
#[inline]
pub fn last_updated_timestamp(&self) -> Result<DateTime<Utc>> {
timestamp_ms_to_utc(self.last_updated_ms)
}
/// Returns last updated time in milliseconds.
#[inline]
pub fn last_updated_ms(&self) -> i64 {
self.last_updated_ms
}
/// Returns schemas
#[inline]
pub fn schemas_iter(&self) -> impl ExactSizeIterator<Item = &SchemaRef> {
self.schemas.values()
}
/// Lookup schema by id.
#[inline]
pub fn schema_by_id(&self, schema_id: SchemaId) -> Option<&SchemaRef> {
self.schemas.get(&schema_id)
}
/// Get current schema
#[inline]
pub fn current_schema(&self) -> &SchemaRef {
self.schema_by_id(self.current_schema_id)
.expect("Current schema id set, but not found in table metadata")
}
/// Get the id of the current schema
#[inline]
pub fn current_schema_id(&self) -> SchemaId {
self.current_schema_id
}
/// Returns all partition specs.
#[inline]
pub fn partition_specs_iter(&self) -> impl ExactSizeIterator<Item = &PartitionSpecRef> {
self.partition_specs.values()
}
/// Lookup partition spec by id.
#[inline]
pub fn partition_spec_by_id(&self, spec_id: i32) -> Option<&PartitionSpecRef> {
self.partition_specs.get(&spec_id)
}
/// Get default partition spec
#[inline]
pub fn default_partition_spec(&self) -> &PartitionSpecRef {
&self.default_spec
}
/// Return the partition type of the default partition spec.
#[inline]
pub fn default_partition_type(&self) -> &StructType {
&self.default_partition_type
}
/// Returns the unified partition type across every partition spec in the table, resolved
/// against `schema`.
///
/// Unlike [`Self::default_partition_type`], the result contains all partition fields ever
/// used by the table, so partition values stay readable across partition spec evolution.
/// See [`compute_unified_partition_type`] for the exact merge rules.
pub fn unified_partition_type(&self, schema: &Schema) -> Result<StructType> {
compute_unified_partition_type(
self.partition_specs_iter().map(|spec| spec.as_ref()),
schema,
)
}
#[inline]
/// Returns spec id of the "current" partition spec.
pub fn default_partition_spec_id(&self) -> i32 {
self.default_spec.spec_id()
}
/// Returns all snapshots
#[inline]
pub fn snapshots(&self) -> impl ExactSizeIterator<Item = &SnapshotRef> {
self.snapshots.values()
}
/// Lookup snapshot by id.
#[inline]
pub fn snapshot_by_id(&self, snapshot_id: i64) -> Option<&SnapshotRef> {
self.snapshots.get(&snapshot_id)
}
/// Returns snapshot history.
#[inline]
pub fn history(&self) -> &[SnapshotLog] {
&self.snapshot_log
}
/// Returns the metadata log.
#[inline]
pub fn metadata_log(&self) -> &[MetadataLog] {
&self.metadata_log
}
/// Get current snapshot
#[inline]
pub fn current_snapshot(&self) -> Option<&SnapshotRef> {
self.current_snapshot_id.map(|s| {
self.snapshot_by_id(s)
.expect("Current snapshot id has been set, but doesn't exist in metadata")
})
}
/// Get the current snapshot id
#[inline]
pub fn current_snapshot_id(&self) -> Option<i64> {
self.current_snapshot_id
}
/// Returns the table's snapshot references (branches and tags) keyed by ref
/// name. The `main` branch, when present, is included. This is the whole
/// `refs` map — the git-like surface (branch/tag enumeration) reads it.
#[inline]
pub fn refs(&self) -> &HashMap<String, SnapshotReference> {
&self.refs
}
/// Get the snapshot for a reference
/// Returns an option if the `ref_name` is not found
#[inline]
pub fn snapshot_for_ref(&self, ref_name: &str) -> Option<&SnapshotRef> {
self.refs.get(ref_name).map(|r| {
self.snapshot_by_id(r.snapshot_id)
.unwrap_or_else(|| panic!("Snapshot id of ref {ref_name} doesn't exist"))
})
}
/// The ancestor snapshot ids of `snapshot_id`, **newest → oldest**,
/// following `parent_snapshot_id` to the root — inclusive of `snapshot_id`
/// itself. Mirrors Java `SnapshotUtil.ancestorIds`.
///
/// Walks the parent chain; stops at the root (`parent_snapshot_id == None`)
/// or a **dangling** parent (an id no longer in metadata, e.g. after
/// expiry). Cycle-guarded: a malformed parent chain that revisits an id
/// terminates rather than looping. An unknown `snapshot_id` yields an empty
/// vec.
pub fn snapshot_ancestor_ids(&self, snapshot_id: i64) -> Vec<i64> {
let mut out = Vec::new();
let mut seen = std::collections::HashSet::new();
let mut cur = self.snapshot_by_id(snapshot_id);
while let Some(snap) = cur {
if !seen.insert(snap.snapshot_id()) {
break; // cycle guard — never loop on a malformed chain
}
out.push(snap.snapshot_id());
cur = match snap.parent_snapshot_id() {
Some(pid) => self.snapshot_by_id(pid),
None => None,
};
}
out
}
/// The snapshots added in the half-open lineage range
/// `(from_snapshot_id, to_snapshot_id]` — the ancestors of `to` that are
/// **not** `from` and not an ancestor of `from` — **oldest → newest** (the
/// order an incremental append scan applies them). Mirrors Java
/// `SnapshotUtil.ancestorsBetween` — the snapshot-range resolution that
/// [`iceberg-rust` still lacks](https://github.com/apache/iceberg-rust/issues/1469)
/// and that a Spark/Flink-style incremental streaming read iterates.
///
/// `from_snapshot_id = None` means "from the very beginning" — every
/// ancestor of `to`. Errors if `to` is unknown, or if `from` is set but is
/// **not** an ancestor of `to` (the two are not on one lineage, so an
/// incremental read between them is undefined — the same guard Java raises).
pub fn snapshots_added_between(
&self,
from_snapshot_id: Option<i64>,
to_snapshot_id: i64,
) -> Result<Vec<i64>> {
if self.snapshot_by_id(to_snapshot_id).is_none() {
return Err(Error::new(
ErrorKind::DataInvalid,
format!("Ending snapshot {to_snapshot_id} is not in the table"),
));
}
// `to`'s ancestors, newest → oldest, inclusive of `to`.
let to_ancestors = self.snapshot_ancestor_ids(to_snapshot_id);
let mut added: Vec<i64> = match from_snapshot_id {
None => to_ancestors,
Some(from) => {
// `from` must lie on `to`'s lineage; collect everything newer
// than it (i.e. up to but excluding `from`).
let mut acc = Vec::new();
let mut found = false;
for id in to_ancestors {
if id == from {
found = true;
break;
}
acc.push(id);
}
// An EXPIRED `from` (snapshot object gone, e.g. a compaction
// expired everything below its replace) is still PROVABLY on
// the lineage when the oldest resolvable ancestor's parent
// pointer names it: the pointer survives the parent's expiry.
// The range is then exactly the resolvable ancestors. A
// dangling `from` deeper than that pointer stays an error —
// the lineage cannot be proven. skade patch, 2026-10-20.
if !found
&& self.snapshot_by_id(from).is_none()
&& acc.last().is_some_and(|oldest| {
self.snapshot_by_id(*oldest)
.and_then(|s| s.parent_snapshot_id())
== Some(from)
})
{
found = true;
}
if !found {
return Err(Error::new(
ErrorKind::DataInvalid,
format!(
"Starting snapshot {from} is not an ancestor of ending snapshot \
{to_snapshot_id} — incremental read between them is undefined"
),
));
}
acc
}
};
// Return oldest → newest (the application order).
added.reverse();
Ok(added)
}
/// `true` iff **every** snapshot added in `(from, to]` is an append-only
/// operation ([`Operation::Append`](super::snapshot::Operation::Append)) —
/// the precondition for a correct *incremental append* read.
///
/// An `Overwrite` / `Delete` / `Replace` in the range means rows were
/// changed or removed, not only added, so scanning "the files added since
/// `from`" would over-report — Java's `BaseIncrementalAppendScan` enforces
/// the same invariant. Callers that need change semantics across such a
/// range must use a CDC/changelog read instead. Errors under the same
/// conditions as [`snapshots_added_between`](Self::snapshots_added_between).
pub fn is_append_only_between(
&self,
from_snapshot_id: Option<i64>,
to_snapshot_id: i64,
) -> Result<bool> {
use super::snapshot::Operation;
let ids = self.snapshots_added_between(from_snapshot_id, to_snapshot_id)?;
Ok(ids.iter().all(|id| {
self.snapshot_by_id(*id)
.map(|s| s.summary().operation == Operation::Append)
.unwrap_or(false)
}))
}
/// Return all sort orders.
#[inline]
pub fn sort_orders_iter(&self) -> impl ExactSizeIterator<Item = &SortOrderRef> {
self.sort_orders.values()
}
/// Lookup sort order by id.
#[inline]
pub fn sort_order_by_id(&self, sort_order_id: i64) -> Option<&SortOrderRef> {
self.sort_orders.get(&sort_order_id)
}
/// Returns default sort order id.
#[inline]
pub fn default_sort_order(&self) -> &SortOrderRef {
self.sort_orders
.get(&self.default_sort_order_id)
.expect("Default order id has been set, but not found in table metadata!")
}
/// Returns default sort order id.
#[inline]
pub fn default_sort_order_id(&self) -> i64 {
self.default_sort_order_id
}
/// Returns properties of table.
#[inline]
pub fn properties(&self) -> &HashMap<String, String> {
&self.properties
}
/// Returns the base location for metadata files (manifests, manifest lists).
///
/// Honors the `write.metadata.path` table property when set, otherwise defaults
/// to the `metadata` subdirectory under the table location.
pub fn metadata_location(&self) -> Result<String> {
Ok(self
.table_properties()
.write_metadata_path()?
.unwrap_or_else(|| format!("{}/{}", self.location(), METADATA_FOLDER_NAME)))
}
/// Returns the metadata compression codec from table properties.
///
/// Returns `CompressionCodec::None` if compression is disabled or not configured.
/// Returns `CompressionCodec::Gzip` if gzip compression is enabled.
///
/// # Errors
///
/// Returns an error if the compression codec property has an invalid value.
pub fn metadata_compression_codec(&self) -> Result<CompressionCodec> {
self.table_properties().metadata_compression_codec()
}
/// Returns a typed view that parses each table property when its getter is called.
#[inline]
pub fn table_properties(&self) -> TableProperties<'_> {
TableProperties::new(&self.properties)
}
/// Return location of statistics files.
#[inline]
pub fn statistics_iter(&self) -> impl ExactSizeIterator<Item = &StatisticsFile> {
self.statistics.values()
}
/// Return location of partition statistics files.
#[inline]
pub fn partition_statistics_iter(
&self,
) -> impl ExactSizeIterator<Item = &PartitionStatisticsFile> {
self.partition_statistics.values()
}
/// Get a statistics file for a snapshot id.
#[inline]
pub fn statistics_for_snapshot(&self, snapshot_id: i64) -> Option<&StatisticsFile> {
self.statistics.get(&snapshot_id)
}
/// Get a partition statistics file for a snapshot id.
#[inline]
pub fn partition_statistics_for_snapshot(
&self,
snapshot_id: i64,
) -> Option<&PartitionStatisticsFile> {
self.partition_statistics.get(&snapshot_id)
}
fn construct_refs(&mut self) {
if let Some(current_snapshot_id) = self.current_snapshot_id
&& !self.refs.contains_key(MAIN_BRANCH)
{
self.refs
.insert(MAIN_BRANCH.to_string(), SnapshotReference {
snapshot_id: current_snapshot_id,
retention: SnapshotRetention::Branch {
min_snapshots_to_keep: None,
max_snapshot_age_ms: None,
max_ref_age_ms: None,
},
});
}
}
/// Iterate over all encryption keys
#[inline]
pub fn encryption_keys_iter(&self) -> impl ExactSizeIterator<Item = &EncryptedKey> {
self.encryption_keys.values()
}
/// Get the encryption key for a given key id
#[inline]
pub fn encryption_key(&self, key_id: &str) -> Option<&EncryptedKey> {
self.encryption_keys.get(key_id)
}
/// Get the next row id to be assigned
#[inline]
pub fn next_row_id(&self) -> u64 {
self.next_row_id
}
/// Read table metadata from the given location.
pub async fn read_from(
file_io: &FileIO,
metadata_location: impl AsRef<str>,
) -> Result<TableMetadata> {
let metadata_location = metadata_location.as_ref();
let input_file = file_io.new_input(metadata_location)?;
let metadata_content = input_file.read().await?;
// Check if the file is compressed by looking for the gzip "magic number".
let metadata = if metadata_content.len() > 2
&& metadata_content[0] == 0x1F
&& metadata_content[1] == 0x8B
{
let decompressed_data = CompressionCodec::gzip_default()
.decompress(metadata_content.to_vec())
.map_err(|e| {
invalid_data!("Trying to read compressed metadata file")
.with_context("file_path", metadata_location)
.with_source(e)
})?;
serde_json::from_slice(&decompressed_data)?
} else {
serde_json::from_slice(&metadata_content)?
};
Ok(metadata)
}
/// Write table metadata to the given location.
pub async fn write_to(
&self,
file_io: &FileIO,
metadata_location: &MetadataLocation,
) -> Result<()> {
let json_data = serde_json::to_vec(self)?;
// Check if compression codec from properties matches the one in metadata_location
let codec = self.table_properties().metadata_compression_codec()?;
if codec != metadata_location.compression_codec() {
return Err(invalid_data!(
"Compression codec mismatch: metadata_location has {:?}, but table properties specify {:?}",
metadata_location.compression_codec(),
codec
));
}
// Apply compression based on codec
let data_to_write = match codec {
CompressionCodec::Gzip(_) => codec.compress(json_data)?,
CompressionCodec::None => json_data,
_ => {
return Err(invalid_data!(
"Unsupported metadata compression codec: {codec:?}"
));
}
};
file_io
.new_output(metadata_location.to_string())?
.write(data_to_write.into())
.await
}
/// Normalize this partition spec.
///
/// This is an internal method
/// meant to be called after constructing table metadata from untrusted sources.
/// We run this method after json deserialization.
/// All constructors for `TableMetadata` which are part of `iceberg-rust`
/// should return normalized `TableMetadata`.
pub(super) fn try_normalize(&mut self) -> Result<&mut Self> {
self.validate_current_schema()?;
self.normalize_current_snapshot()?;
self.construct_refs();
self.validate_refs()?;
self.validate_chronological_snapshot_logs()?;
self.validate_chronological_metadata_logs()?;
// Normalize location (remove trailing slash)
self.location = self.location.trim_end_matches('/').to_string();
self.validate_snapshot_sequence_number()?;
self.validate_schema_format_compatibility()?;
self.try_normalize_partition_spec()?;
self.try_normalize_sort_order()?;
Ok(self)
}
/// Validate active default-spec sources and add the spec if it is not present.
fn try_normalize_partition_spec(&mut self) -> Result<()> {
for field in self.default_spec.fields() {
// Historical specs may reference dropped columns, but active default-spec
// fields need a source for new writes. Void fields do not read their source.
if field.transform != Transform::Void
&& self.current_schema().field_by_id(field.source_id).is_none()
{
return Err(invalid_data!(
"Default partition spec {} references missing source field {} in current schema {}",
self.default_spec.spec_id(),
field.source_id,
self.current_schema_id
));
}
}
if self
.partition_spec_by_id(self.default_spec.spec_id())
.is_none()
{
self.partition_specs.insert(
self.default_spec.spec_id(),
Arc::new(Arc::unwrap_or_clone(self.default_spec.clone())),
);
}
Ok(())
}
/// If the default sort order is unsorted but the sort order is not present, add it
fn try_normalize_sort_order(&mut self) -> Result<()> {
// Validate that sort order ID 0 (reserved for unsorted) has no fields
if let Some(sort_order) = self.sort_order_by_id(SortOrder::UNSORTED_ORDER_ID)
&& !sort_order.fields.is_empty()
{
return Err(Error::new(
ErrorKind::Unexpected,
format!(
"Sort order ID {} is reserved for unsorted order",
SortOrder::UNSORTED_ORDER_ID
),
));
}
if self.sort_order_by_id(self.default_sort_order_id).is_some() {
return Ok(());
}
if self.default_sort_order_id != SortOrder::UNSORTED_ORDER_ID {
return Err(invalid_data!(
"No sort order exists with the default sort order id {}.",
self.default_sort_order_id
));
}
let sort_order = SortOrder::unsorted_order();
self.sort_orders
.insert(SortOrder::UNSORTED_ORDER_ID, Arc::new(sort_order));
Ok(())
}
/// Validate the current schema is set and exists.
fn validate_current_schema(&self) -> Result<()> {
if self.schema_by_id(self.current_schema_id).is_none() {
return Err(invalid_data!(
"No schema exists with the current schema id {}.",
self.current_schema_id
));
}
Ok(())
}
/// If current snapshot is Some(-1) then set it to None.
fn normalize_current_snapshot(&mut self) -> Result<()> {
if let Some(current_snapshot_id) = self.current_snapshot_id {
if current_snapshot_id == EMPTY_SNAPSHOT_ID {
self.current_snapshot_id = None;
} else if self.snapshot_by_id(current_snapshot_id).is_none() {
return Err(invalid_data!(
"Snapshot for current snapshot id {current_snapshot_id} does not exist in the existing snapshots list"
));
}
}
Ok(())
}
/// Validate that all refs are valid (snapshot exists)
fn validate_refs(&self) -> Result<()> {
for (name, snapshot_ref) in self.refs.iter() {
if self.snapshot_by_id(snapshot_ref.snapshot_id).is_none() {
return Err(invalid_data!(
"Snapshot for reference {name} does not exist in the existing snapshots list"
));
}
}
let main_ref = self.refs.get(MAIN_BRANCH);
if self.current_snapshot_id.is_some() {
if let Some(main_ref) = main_ref
&& main_ref.snapshot_id != self.current_snapshot_id.unwrap_or_default()
{
return Err(invalid_data!(
"Current snapshot id does not match main branch ({:?} != {:?})",
self.current_snapshot_id.unwrap_or_default(),
main_ref.snapshot_id
));
}
} else if main_ref.is_some() {
return Err(invalid_data!(
"Current snapshot is not set, but main branch exists"
));
}
Ok(())
}
/// Validate that for V1 Metadata the last_sequence_number is 0
fn validate_snapshot_sequence_number(&self) -> Result<()> {
if self.format_version < FormatVersion::V2 && self.last_sequence_number != 0 {
return Err(invalid_data!(
"Last sequence number must be 0 in v1. Found {}",
self.last_sequence_number
));
}
if self.format_version >= FormatVersion::V2
&& let Some(snapshot) = self
.snapshots
.values()
.find(|snapshot| snapshot.sequence_number() > self.last_sequence_number)
{
return Err(invalid_data!(
"Invalid snapshot with id {} and sequence number {} greater than last sequence number {}",
snapshot.snapshot_id(),
snapshot.sequence_number(),
self.last_sequence_number
));
}
Ok(())
}
/// Validate snapshots logs are chronological and last updated is after the last snapshot log.
fn validate_chronological_snapshot_logs(&self) -> Result<()> {
for window in self.snapshot_log.windows(2) {
let (prev, curr) = (&window[0], &window[1]);
// commits can happen concurrently from different machines.
// A tolerance helps us avoid failure for small clock skew
if curr.timestamp_ms - prev.timestamp_ms < -ONE_MINUTE_MS {
return Err(invalid_data!("Expected sorted snapshot log entries"));
}
}
if let Some(last) = self.snapshot_log.last() {
// commits can happen concurrently from different machines.
// A tolerance helps us avoid failure for small clock skew
if self.last_updated_ms - last.timestamp_ms < -ONE_MINUTE_MS {
return Err(invalid_data!(
"Invalid update timestamp {}: before last snapshot log entry at {}",
self.last_updated_ms,
last.timestamp_ms
));
}
}
Ok(())
}
fn validate_chronological_metadata_logs(&self) -> Result<()> {
for window in self.metadata_log.windows(2) {
let (prev, curr) = (&window[0], &window[1]);
// commits can happen concurrently from different machines.
// A tolerance helps us avoid failure for small clock skew
if curr.timestamp_ms - prev.timestamp_ms < -ONE_MINUTE_MS {
return Err(invalid_data!("Expected sorted metadata log entries"));
}
}
if let Some(last) = self.metadata_log.last() {
// commits can happen concurrently from different machines.
// A tolerance helps us avoid failure for small clock skew
if self.last_updated_ms - last.timestamp_ms < -ONE_MINUTE_MS {
return Err(invalid_data!(
"Invalid update timestamp {}: before last metadata log entry at {}",
self.last_updated_ms,
last.timestamp_ms
));
}
}
Ok(())
}
/// Validates that every type used in the current schema is supported by the
/// table's format version. Delegates to [`Schema::check_format_compatibility`].
fn validate_schema_format_compatibility(&self) -> Result<()> {
self.current_schema()
.check_format_compatibility(self.format_version)
}
}
pub(super) mod _serde {
use std::borrow::BorrowMut;
/// This is a helper module that defines types to help with serialization/deserialization.
/// For deserialization the input first gets read into either the [TableMetadataV1] or [TableMetadataV2] struct
/// and then converted into the [TableMetadata] struct. Serialization works the other way around.
/// [TableMetadataV1] and [TableMetadataV2] are internal struct that are only used for serialization and deserialization.
use std::collections::HashMap;
/// This is a helper module that defines types to help with serialization/deserialization.
/// For deserialization the input first gets read into either the [TableMetadataV1] or [TableMetadataV2] struct
/// and then converted into the [TableMetadata] struct. Serialization works the other way around.
/// [TableMetadataV1] and [TableMetadataV2] are internal struct that are only used for serialization and deserialization.
use std::sync::Arc;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use super::{
DEFAULT_PARTITION_SPEC_ID, EMPTY_SNAPSHOT_ID, FormatVersion, MAIN_BRANCH, MetadataLog,
SnapshotLog, TableMetadata,
};
use crate::error::invalid_data;
use crate::spec::schema::_serde::{SchemaV1, SchemaV2};
use crate::spec::snapshot::_serde::{SnapshotV1, SnapshotV2, SnapshotV3};
use crate::spec::{
EncryptedKey, INITIAL_ROW_ID, PartitionField, PartitionSpec, PartitionSpecRef,
PartitionStatisticsFile, Schema, SchemaRef, Snapshot, SnapshotReference, SnapshotRetention,
SortOrder, StatisticsFile,
};
use crate::{Error, ErrorKind};
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(untagged)]
pub(super) enum TableMetadataEnum {
V3(TableMetadataV3),
V2(TableMetadataV2),
V1(TableMetadataV1),
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "kebab-case")]
/// Defines the structure of a v2 table metadata for serialization/deserialization
pub(super) struct TableMetadataV3 {
pub format_version: VersionNumber<3>,
#[serde(flatten)]
pub shared: TableMetadataV2V3Shared,
pub next_row_id: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub encryption_keys: Option<Vec<EncryptedKey>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub snapshots: Option<Vec<SnapshotV3>>,
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "kebab-case")]
/// Defines the structure of a v2 table metadata for serialization/deserialization
pub(super) struct TableMetadataV2V3Shared {
pub table_uuid: Uuid,
pub location: String,
pub last_sequence_number: i64,
pub last_updated_ms: i64,
pub last_column_id: i32,
pub schemas: Vec<SchemaV2>,
pub current_schema_id: i32,
pub partition_specs: Vec<PartitionSpec>,
pub default_spec_id: i32,
pub last_partition_id: i32,
#[serde(skip_serializing_if = "Option::is_none")]
pub properties: Option<HashMap<String, String>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub current_snapshot_id: Option<i64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub snapshot_log: Option<Vec<SnapshotLog>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub metadata_log: Option<Vec<MetadataLog>>,
pub sort_orders: Vec<SortOrder>,
pub default_sort_order_id: i64,
#[serde(skip_serializing_if = "Option::is_none")]
pub refs: Option<HashMap<String, SnapshotReference>>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub statistics: Vec<StatisticsFile>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub partition_statistics: Vec<PartitionStatisticsFile>,
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "kebab-case")]
/// Defines the structure of a v2 table metadata for serialization/deserialization
pub(super) struct TableMetadataV2 {
pub format_version: VersionNumber<2>,
#[serde(flatten)]
pub shared: TableMetadataV2V3Shared,
#[serde(skip_serializing_if = "Option::is_none")]
pub snapshots: Option<Vec<SnapshotV2>>,
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "kebab-case")]
/// Defines the structure of a v1 table metadata for serialization/deserialization
pub(super) struct TableMetadataV1 {
pub format_version: VersionNumber<1>,
#[serde(skip_serializing_if = "Option::is_none")]
pub table_uuid: Option<Uuid>,
pub location: String,
pub last_updated_ms: i64,
pub last_column_id: i32,
/// `schema` is optional to prioritize `schemas` and `current-schema-id`, allowing liberal reading of V1 metadata.
pub schema: Option<SchemaV1>,
#[serde(skip_serializing_if = "Option::is_none")]
pub schemas: Option<Vec<SchemaV1>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub current_schema_id: Option<i32>,
/// `partition_spec` is optional to prioritize `partition_specs`, aligning with liberal reading of potentially invalid V1 metadata.
pub partition_spec: Option<Vec<PartitionField>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub partition_specs: Option<Vec<PartitionSpec>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub default_spec_id: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub last_partition_id: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub properties: Option<HashMap<String, String>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub current_snapshot_id: Option<i64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub snapshots: Option<Vec<SnapshotV1>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub snapshot_log: Option<Vec<SnapshotLog>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub metadata_log: Option<Vec<MetadataLog>>,
pub sort_orders: Option<Vec<SortOrder>>,
pub default_sort_order_id: Option<i64>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub statistics: Vec<StatisticsFile>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub partition_statistics: Vec<PartitionStatisticsFile>,
}
/// Helper to serialize and deserialize the format version.
#[derive(Debug, PartialEq, Eq)]
pub(crate) struct VersionNumber<const V: u8>;
impl Serialize for TableMetadata {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where S: serde::Serializer {
// we must do a clone here
let table_metadata_enum: TableMetadataEnum =
self.clone().try_into().map_err(serde::ser::Error::custom)?;
table_metadata_enum.serialize(serializer)
}
}
impl<const V: u8> Serialize for VersionNumber<V> {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where S: serde::Serializer {
serializer.serialize_u8(V)
}
}
impl<'de, const V: u8> Deserialize<'de> for VersionNumber<V> {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where D: serde::Deserializer<'de> {
let value = u8::deserialize(deserializer)?;
if value == V {
Ok(VersionNumber::<V>)
} else {
Err(serde::de::Error::custom("Invalid Version"))
}
}
}
impl TryFrom<TableMetadataEnum> for TableMetadata {
type Error = Error;
fn try_from(value: TableMetadataEnum) -> Result<Self, Error> {
match value {
TableMetadataEnum::V3(value) => value.try_into(),
TableMetadataEnum::V2(value) => value.try_into(),
TableMetadataEnum::V1(value) => value.try_into(),
}
}
}
impl TryFrom<TableMetadata> for TableMetadataEnum {
type Error = Error;
fn try_from(value: TableMetadata) -> Result<Self, Error> {
Ok(match value.format_version {
FormatVersion::V3 => TableMetadataEnum::V3(value.try_into()?),
FormatVersion::V2 => TableMetadataEnum::V2(value.into()),
FormatVersion::V1 => TableMetadataEnum::V1(value.try_into()?),
})
}
}
impl TryFrom<TableMetadataV3> for TableMetadata {
type Error = Error;
fn try_from(value: TableMetadataV3) -> Result<Self, Error> {
let TableMetadataV3 {
format_version: _,
shared: value,
next_row_id,
encryption_keys,
snapshots,
} = value;
let current_snapshot_id = if value.current_snapshot_id == Some(EMPTY_SNAPSHOT_ID) {
None
} else {
value.current_snapshot_id
};
let schemas = HashMap::from_iter(
value
.schemas
.into_iter()
.map(|schema| Ok((schema.schema_id, Arc::new(schema.try_into()?))))
.collect::<Result<Vec<_>, Error>>()?,
);
let current_schema: &SchemaRef =
schemas.get(&value.current_schema_id).ok_or_else(|| {
invalid_data!(
"No schema exists with the current schema id {}.",
value.current_schema_id
)
})?;
let partition_specs = HashMap::from_iter(
value
.partition_specs
.into_iter()
.map(|x| (x.spec_id(), Arc::new(x))),
);
let default_spec_id = value.default_spec_id;
let default_spec: PartitionSpecRef = partition_specs
.get(&value.default_spec_id)
.map(|spec| (**spec).clone())
.or_else(|| {
(DEFAULT_PARTITION_SPEC_ID == default_spec_id)
.then(PartitionSpec::unpartition_spec)
})
.ok_or_else(|| invalid_data!("Default partition spec {default_spec_id} not found"))?
.into();
let default_partition_type = default_spec.partition_type(current_schema)?;
let mut metadata = TableMetadata {
format_version: FormatVersion::V3,
table_uuid: value.table_uuid,
location: value.location,
last_sequence_number: value.last_sequence_number,
last_updated_ms: value.last_updated_ms,
last_column_id: value.last_column_id,
current_schema_id: value.current_schema_id,
schemas,
partition_specs,
default_partition_type,
default_spec,
last_partition_id: value.last_partition_id,
properties: value.properties.unwrap_or_default(),
current_snapshot_id,
snapshots: snapshots
.map(|snapshots| {
HashMap::from_iter(
snapshots
.into_iter()
.map(|x| (x.snapshot_id, Arc::new(x.into()))),
)
})
.unwrap_or_default(),
snapshot_log: value.snapshot_log.unwrap_or_default(),
metadata_log: value.metadata_log.unwrap_or_default(),
sort_orders: HashMap::from_iter(
value
.sort_orders
.into_iter()
.map(|x| (x.order_id, Arc::new(x))),
),
default_sort_order_id: value.default_sort_order_id,
refs: value.refs.unwrap_or_else(|| {
if let Some(snapshot_id) = current_snapshot_id {
HashMap::from_iter(vec![(MAIN_BRANCH.to_string(), SnapshotReference {
snapshot_id,
retention: SnapshotRetention::Branch {
min_snapshots_to_keep: None,
max_snapshot_age_ms: None,
max_ref_age_ms: None,
},
})])
} else {
HashMap::new()
}
}),
statistics: index_statistics(value.statistics),
partition_statistics: index_partition_statistics(value.partition_statistics),
encryption_keys: encryption_keys
.map(|keys| {
HashMap::from_iter(keys.into_iter().map(|key| (key.key_id.clone(), key)))
})
.unwrap_or_default(),
next_row_id,
};
metadata.borrow_mut().try_normalize()?;
Ok(metadata)
}
}
impl TryFrom<TableMetadataV2> for TableMetadata {
type Error = Error;
fn try_from(value: TableMetadataV2) -> Result<Self, Error> {
let snapshots = value.snapshots;
let value = value.shared;
let current_snapshot_id = if value.current_snapshot_id == Some(EMPTY_SNAPSHOT_ID) {
None
} else {
value.current_snapshot_id
};
let schemas = HashMap::from_iter(
value
.schemas
.into_iter()
.map(|schema| Ok((schema.schema_id, Arc::new(schema.try_into()?))))
.collect::<Result<Vec<_>, Error>>()?,
);
let current_schema: &SchemaRef =
schemas.get(&value.current_schema_id).ok_or_else(|| {
invalid_data!(
"No schema exists with the current schema id {}.",
value.current_schema_id
)
})?;
let partition_specs = HashMap::from_iter(
value
.partition_specs
.into_iter()
.map(|x| (x.spec_id(), Arc::new(x))),
);
let default_spec_id = value.default_spec_id;
let default_spec: PartitionSpecRef = partition_specs
.get(&value.default_spec_id)
.map(|spec| (**spec).clone())
.or_else(|| {
(DEFAULT_PARTITION_SPEC_ID == default_spec_id)
.then(PartitionSpec::unpartition_spec)
})
.ok_or_else(|| invalid_data!("Default partition spec {default_spec_id} not found"))?
.into();
let default_partition_type = default_spec.partition_type(current_schema)?;
let mut metadata = TableMetadata {
format_version: FormatVersion::V2,
table_uuid: value.table_uuid,
location: value.location,
last_sequence_number: value.last_sequence_number,
last_updated_ms: value.last_updated_ms,
last_column_id: value.last_column_id,
current_schema_id: value.current_schema_id,
schemas,
partition_specs,
default_partition_type,
default_spec,
last_partition_id: value.last_partition_id,
properties: value.properties.unwrap_or_default(),
current_snapshot_id,
snapshots: snapshots
.map(|snapshots| {
HashMap::from_iter(
snapshots
.into_iter()
.map(|x| (x.snapshot_id, Arc::new(x.into()))),
)
})
.unwrap_or_default(),
snapshot_log: value.snapshot_log.unwrap_or_default(),
metadata_log: value.metadata_log.unwrap_or_default(),
sort_orders: HashMap::from_iter(
value
.sort_orders
.into_iter()
.map(|x| (x.order_id, Arc::new(x))),
),
default_sort_order_id: value.default_sort_order_id,
refs: value.refs.unwrap_or_else(|| {
if let Some(snapshot_id) = current_snapshot_id {
HashMap::from_iter(vec![(MAIN_BRANCH.to_string(), SnapshotReference {
snapshot_id,
retention: SnapshotRetention::Branch {
min_snapshots_to_keep: None,
max_snapshot_age_ms: None,
max_ref_age_ms: None,
},
})])
} else {
HashMap::new()
}
}),
statistics: index_statistics(value.statistics),
partition_statistics: index_partition_statistics(value.partition_statistics),
encryption_keys: HashMap::new(),
next_row_id: INITIAL_ROW_ID,
};
metadata.borrow_mut().try_normalize()?;
Ok(metadata)
}
}
impl TryFrom<TableMetadataV1> for TableMetadata {
type Error = Error;
fn try_from(value: TableMetadataV1) -> Result<Self, Error> {
let current_snapshot_id = if value.current_snapshot_id == Some(EMPTY_SNAPSHOT_ID) {
None
} else {
value.current_snapshot_id
};
let (schemas, current_schema_id, current_schema) =
if let (Some(schemas_vec), Some(schema_id)) =
(&value.schemas, value.current_schema_id)
{
// Option 1: Use 'schemas' + 'current_schema_id'
let schema_map = HashMap::from_iter(
schemas_vec
.clone()
.into_iter()
.map(|schema| {
let schema: Schema = schema.try_into()?;
Ok((schema.schema_id(), Arc::new(schema)))
})
.collect::<Result<Vec<_>, Error>>()?,
);
let schema = schema_map
.get(&schema_id)
.ok_or_else(|| {
invalid_data!(
"No schema exists with the current schema id {schema_id}."
)
})?
.clone();
(schema_map, schema_id, schema)
} else if let Some(schema) = value.schema {
// Option 2: Fall back to `schema`
let schema: Schema = schema.try_into()?;
let schema_id = schema.schema_id();
let schema_arc = Arc::new(schema);
let schema_map = HashMap::from_iter(vec![(schema_id, schema_arc.clone())]);
(schema_map, schema_id, schema_arc)
} else {
// Option 3: No valid schema configuration found
return Err(invalid_data!(
"No valid schema configuration found in table metadata"
));
};
// Prioritize 'partition_specs' over 'partition_spec'
let partition_specs = if let Some(specs_vec) = value.partition_specs {
// Option 1: Use 'partition_specs'
specs_vec
.into_iter()
.map(|x| (x.spec_id(), Arc::new(x)))
.collect::<HashMap<_, _>>()
} else if let Some(partition_spec) = value.partition_spec {
// Option 2: Fall back to 'partition_spec'
let spec = PartitionSpec::builder(current_schema.clone())
.with_spec_id(DEFAULT_PARTITION_SPEC_ID)
.add_unbound_fields(partition_spec.into_iter().map(|f| f.into_unbound()))?
.build()?;
HashMap::from_iter(vec![(DEFAULT_PARTITION_SPEC_ID, Arc::new(spec))])
} else {
// Option 3: Create empty partition spec
let spec = PartitionSpec::builder(current_schema.clone())
.with_spec_id(DEFAULT_PARTITION_SPEC_ID)
.build()?;
HashMap::from_iter(vec![(DEFAULT_PARTITION_SPEC_ID, Arc::new(spec))])
};
// Get the default_spec_id, prioritizing the explicit value if provided
let default_spec_id = value
.default_spec_id
.unwrap_or_else(|| partition_specs.keys().copied().max().unwrap_or_default());
// Get the default spec
let default_spec: PartitionSpecRef = partition_specs
.get(&default_spec_id)
.map(|x| Arc::unwrap_or_clone(x.clone()))
.ok_or_else(|| invalid_data!("Default partition spec {default_spec_id} not found"))?
.into();
let default_partition_type = default_spec.partition_type(¤t_schema)?;
let mut metadata = TableMetadata {
format_version: FormatVersion::V1,
table_uuid: value.table_uuid.unwrap_or_default(),
location: value.location,
last_sequence_number: 0,
last_updated_ms: value.last_updated_ms,
last_column_id: value.last_column_id,
current_schema_id,
default_spec,
default_partition_type,
last_partition_id: value
.last_partition_id
.unwrap_or_else(|| partition_specs.keys().copied().max().unwrap_or_default()),
partition_specs,
schemas,
properties: value.properties.unwrap_or_default(),
current_snapshot_id,
snapshots: value
.snapshots
.map(|snapshots| {
Ok::<_, Error>(HashMap::from_iter(
snapshots
.into_iter()
.map(|x| Ok((x.snapshot_id, Arc::new(x.try_into()?))))
.collect::<Result<Vec<_>, Error>>()?,
))
})
.transpose()?
.unwrap_or_default(),
snapshot_log: value.snapshot_log.unwrap_or_default(),
metadata_log: value.metadata_log.unwrap_or_default(),
sort_orders: match value.sort_orders {
Some(sort_orders) => HashMap::from_iter(
sort_orders.into_iter().map(|x| (x.order_id, Arc::new(x))),
),
None => HashMap::new(),
},
default_sort_order_id: value
.default_sort_order_id
.unwrap_or(SortOrder::UNSORTED_ORDER_ID),
refs: if let Some(snapshot_id) = current_snapshot_id {
HashMap::from_iter(vec![(MAIN_BRANCH.to_string(), SnapshotReference {
snapshot_id,
retention: SnapshotRetention::Branch {
min_snapshots_to_keep: None,
max_snapshot_age_ms: None,
max_ref_age_ms: None,
},
})])
} else {
HashMap::new()
},
statistics: index_statistics(value.statistics),
partition_statistics: index_partition_statistics(value.partition_statistics),
encryption_keys: HashMap::new(),
next_row_id: INITIAL_ROW_ID, // v1 has no row lineage
};
metadata.borrow_mut().try_normalize()?;
Ok(metadata)
}
}
impl TryFrom<TableMetadata> for TableMetadataV3 {
type Error = Error;
fn try_from(mut v: TableMetadata) -> Result<Self, Self::Error> {
let next_row_id = v.next_row_id;
let encryption_keys = std::mem::take(&mut v.encryption_keys);
let snapshots = std::mem::take(&mut v.snapshots);
let shared = v.into();
Ok(TableMetadataV3 {
format_version: VersionNumber::<3>,
shared,
next_row_id,
encryption_keys: if encryption_keys.is_empty() {
None
} else {
Some(encryption_keys.into_values().collect())
},
snapshots: if snapshots.is_empty() {
None
} else {
Some(
snapshots
.into_values()
.map(|s| SnapshotV3::try_from(Arc::unwrap_or_clone(s)))
.collect::<Result<_, _>>()?,
)
},
})
}
}
impl From<TableMetadata> for TableMetadataV2 {
fn from(mut v: TableMetadata) -> Self {
let snapshots = std::mem::take(&mut v.snapshots);
let shared = v.into();
TableMetadataV2 {
format_version: VersionNumber::<2>,
shared,
snapshots: if snapshots.is_empty() {
None
} else {
Some(
snapshots
.into_values()
.map(|s| SnapshotV2::from(Arc::unwrap_or_clone(s)))
.collect(),
)
},
}
}
}
impl From<TableMetadata> for TableMetadataV2V3Shared {
fn from(v: TableMetadata) -> Self {
TableMetadataV2V3Shared {
table_uuid: v.table_uuid,
location: v.location,
last_sequence_number: v.last_sequence_number,
last_updated_ms: v.last_updated_ms,
last_column_id: v.last_column_id,
schemas: v
.schemas
.into_values()
.map(|x| {
Arc::try_unwrap(x)
.unwrap_or_else(|schema| schema.as_ref().clone())
.into()
})
.collect(),
current_schema_id: v.current_schema_id,
partition_specs: v
.partition_specs
.into_values()
.map(|x| Arc::try_unwrap(x).unwrap_or_else(|s| s.as_ref().clone()))
.collect(),
default_spec_id: v.default_spec.spec_id(),
last_partition_id: v.last_partition_id,
properties: if v.properties.is_empty() {
None
} else {
Some(v.properties)
},
current_snapshot_id: v.current_snapshot_id,
snapshot_log: if v.snapshot_log.is_empty() {
None
} else {
Some(v.snapshot_log)
},
metadata_log: if v.metadata_log.is_empty() {
None
} else {
Some(v.metadata_log)
},
sort_orders: v
.sort_orders
.into_values()
.map(|x| Arc::try_unwrap(x).unwrap_or_else(|s| s.as_ref().clone()))
.collect(),
default_sort_order_id: v.default_sort_order_id,
refs: Some(v.refs),
statistics: v.statistics.into_values().collect(),
partition_statistics: v.partition_statistics.into_values().collect(),
}
}
}
impl TryFrom<TableMetadata> for TableMetadataV1 {
type Error = Error;
fn try_from(v: TableMetadata) -> Result<Self, Error> {
Ok(TableMetadataV1 {
format_version: VersionNumber::<1>,
table_uuid: Some(v.table_uuid),
location: v.location,
last_updated_ms: v.last_updated_ms,
last_column_id: v.last_column_id,
schema: Some(
v.schemas
.get(&v.current_schema_id)
.ok_or(Error::new(
ErrorKind::Unexpected,
"current_schema_id not found in schemas",
))?
.as_ref()
.clone()
.into(),
),
schemas: Some(
v.schemas
.into_values()
.map(|x| {
Arc::try_unwrap(x)
.unwrap_or_else(|schema| schema.as_ref().clone())
.into()
})
.collect(),
),
current_schema_id: Some(v.current_schema_id),
partition_spec: Some(v.default_spec.fields().to_vec()),
partition_specs: Some(
v.partition_specs
.into_values()
.map(|x| Arc::try_unwrap(x).unwrap_or_else(|s| s.as_ref().clone()))
.collect(),
),
default_spec_id: Some(v.default_spec.spec_id()),
last_partition_id: Some(v.last_partition_id),
properties: if v.properties.is_empty() {
None
} else {
Some(v.properties)
},
current_snapshot_id: v.current_snapshot_id,
snapshots: if v.snapshots.is_empty() {
None
} else {
Some(
v.snapshots
.into_values()
.map(|x| Snapshot::clone(&x).into())
.collect(),
)
},
snapshot_log: if v.snapshot_log.is_empty() {
None
} else {
Some(v.snapshot_log)
},
metadata_log: if v.metadata_log.is_empty() {
None
} else {
Some(v.metadata_log)
},
sort_orders: Some(
v.sort_orders
.into_values()
.map(|s| Arc::try_unwrap(s).unwrap_or_else(|s| s.as_ref().clone()))
.collect(),
),
default_sort_order_id: Some(v.default_sort_order_id),
statistics: v.statistics.into_values().collect(),
partition_statistics: v.partition_statistics.into_values().collect(),
})
}
}
fn index_statistics(statistics: Vec<StatisticsFile>) -> HashMap<i64, StatisticsFile> {
statistics
.into_iter()
.rev()
.map(|s| (s.snapshot_id, s))
.collect()
}
fn index_partition_statistics(
statistics: Vec<PartitionStatisticsFile>,
) -> HashMap<i64, PartitionStatisticsFile> {
statistics
.into_iter()
.rev()
.map(|s| (s.snapshot_id, s))
.collect()
}
}
#[derive(Debug, Serialize_repr, Deserialize_repr, PartialEq, Eq, Clone, Copy, Hash)]
#[repr(u8)]
/// Iceberg format version
pub enum FormatVersion {
/// Iceberg spec version 1
V1 = 1u8,
/// Iceberg spec version 2
V2 = 2u8,
/// Iceberg spec version 3
V3 = 3u8,
}
impl PartialOrd for FormatVersion {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
Some(self.cmp(other))
}
}
impl Ord for FormatVersion {
fn cmp(&self, other: &Self) -> Ordering {
(*self as u8).cmp(&(*other as u8))
}
}
impl Display for FormatVersion {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
match self {
FormatVersion::V1 => write!(f, "v1"),
FormatVersion::V2 => write!(f, "v2"),
FormatVersion::V3 => write!(f, "v3"),
}
}
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)]
#[serde(rename_all = "kebab-case")]
/// Encodes changes to the previous metadata files for the table
pub struct MetadataLog {
/// The file for the log.
pub metadata_file: String,
/// Time new metadata was created
pub timestamp_ms: i64,
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)]
#[serde(rename_all = "kebab-case")]
/// A log of when each snapshot was made.
pub struct SnapshotLog {
/// Id of the snapshot.
pub snapshot_id: i64,
/// Last updated timestamp
pub timestamp_ms: i64,
}
impl SnapshotLog {
/// Returns the last updated timestamp as a [`DateTime<Utc>`] with millisecond precision
pub fn timestamp(self) -> Result<DateTime<Utc>> {
timestamp_ms_to_utc(self.timestamp_ms)
}
/// Returns the timestamp in milliseconds
#[inline]
pub fn timestamp_ms(&self) -> i64 {
self.timestamp_ms
}
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::fs;
use std::sync::Arc;
use anyhow::Result;
use base64::Engine as _;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
use uuid::Uuid;
use super::{FormatVersion, MetadataLog, SnapshotLog, TableMetadataBuilder};
use crate::catalog::MetadataLocation;
use crate::compression::CompressionCodec;
use crate::io::FileIO;
use crate::spec::table_metadata::TableMetadata;
use crate::spec::{
BlobMetadata, EncryptedKey, INITIAL_ROW_ID, Literal, NestedField, NullOrder, Operation,
PartitionSpec, PartitionStatisticsFile, PrimitiveLiteral, PrimitiveType, Schema, Snapshot,
SnapshotReference, SnapshotRetention, SortDirection, SortField, SortOrder, StatisticsFile,
Summary, TableProperties, Transform, Type, UnboundPartitionField, UnboundPartitionSpec,
};
use crate::{ErrorKind, TableCreation};
fn check_table_metadata_serde(json: &str, expected_type: TableMetadata) {
let desered_type: TableMetadata = serde_json::from_str(json).unwrap();
assert_eq!(desered_type, expected_type);
let sered_json = serde_json::to_string(&expected_type).unwrap();
let parsed_json_value = serde_json::from_str::<TableMetadata>(&sered_json).unwrap();
assert_eq!(parsed_json_value, desered_type);
}
fn get_test_table_metadata(file_name: &str) -> TableMetadata {
let path = format!("testdata/table_metadata/{file_name}");
let metadata: String = fs::read_to_string(path).unwrap();
serde_json::from_str(&metadata).unwrap()
}
/// Loads a test table metadata and relocates it to `location`, so that derived
/// metadata paths point at a writable (e.g. temp) directory.
fn get_test_table_metadata_at(file_name: &str, location: &str) -> TableMetadata {
TableMetadataBuilder::new_from_metadata(get_test_table_metadata(file_name), None)
.set_location(location.to_string())
.build()
.unwrap()
.metadata
}
// ── incremental-scan snapshot-range primitives (SnapshotUtil parity) ─────
/// `TableMetadataV2Valid` carries a two-snapshot append lineage
/// A(3051729675574597004, root) → B(3055729675574597004): the ancestor
/// walk, the `(from, to]` range, and the append-only check all resolve it.
#[test]
fn snapshot_range_resolves_append_lineage() {
let md = get_test_table_metadata("TableMetadataV2Valid.json");
let a = 3051729675574597004_i64;
let b = 3055729675574597004_i64;
// Ancestors newest→oldest, inclusive of the tip.
assert_eq!(md.snapshot_ancestor_ids(b), vec![b, a]);
assert_eq!(md.snapshot_ancestor_ids(a), vec![a]);
assert!(
md.snapshot_ancestor_ids(999).is_empty(),
"unknown id → empty"
);
// Added in (None, b] == the whole lineage, oldest→newest.
assert_eq!(md.snapshots_added_between(None, b).unwrap(), vec![a, b]);
// Added in (a, b] == just b.
assert_eq!(md.snapshots_added_between(Some(a), b).unwrap(), vec![b]);
// Empty range (b, b].
assert!(md.snapshots_added_between(Some(b), b).unwrap().is_empty());
// Both snapshots are appends → the range is append-only.
assert!(md.is_append_only_between(Some(a), b).unwrap());
assert!(md.is_append_only_between(None, b).unwrap());
// Guards: unknown `to`, and a `from` that isn't an ancestor of `to`.
assert!(md.snapshots_added_between(None, 999).is_err());
assert!(
md.snapshots_added_between(Some(999), b).is_err(),
"a `from` off the lineage is rejected"
);
}
/// An `overwrite` snapshot spliced onto the lineage flips the append-only
/// check to `false` for a range that spans it — the precondition an
/// incremental *append* scan enforces (rows changed, not only added).
#[test]
fn overwrite_in_range_is_not_append_only() {
// Take the valid V2 metadata as JSON, append a 3rd snapshot C that is an
// `overwrite` child of B, and make C current.
let path = "testdata/table_metadata/TableMetadataV2Valid.json";
let mut v: serde_json::Value =
serde_json::from_str(&fs::read_to_string(path).unwrap()).unwrap();
let b = 3055729675574597004_i64;
let c = 3055729675574597099_i64;
let snaps = v["snapshots"].as_array_mut().unwrap();
let template = snaps[1].clone();
let mut overwrite = template;
overwrite["snapshot-id"] = serde_json::json!(c);
overwrite["parent-snapshot-id"] = serde_json::json!(b);
overwrite["sequence-number"] = serde_json::json!(2);
overwrite["summary"]["operation"] = serde_json::json!("overwrite");
snaps.push(overwrite);
v["current-snapshot-id"] = serde_json::json!(c);
let md: TableMetadata = serde_json::from_value(v).unwrap();
// The range (b, c] is a single overwrite → NOT append-only.
assert_eq!(md.snapshots_added_between(Some(b), c).unwrap(), vec![c]);
assert!(
!md.is_append_only_between(Some(b), c).unwrap(),
"an overwrite in the range fails the append-only precondition"
);
// But the earlier all-append range still passes.
let a = 3051729675574597004_i64;
assert!(md.is_append_only_between(Some(a), b).unwrap());
}
#[test]
fn test_table_data_v2() {
let data = r#"
{
"format-version" : 2,
"table-uuid": "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94",
"location": "s3://b/wh/data.db/table",
"last-sequence-number" : 1,
"last-updated-ms": 1515100955770,
"last-column-id": 1,
"schemas": [
{
"schema-id" : 1,
"type" : "struct",
"fields" :[
{
"id": 1,
"name": "struct_name",
"required": true,
"type": "fixed[1]"
},
{
"id": 4,
"name": "ts",
"required": true,
"type": "timestamp"
}
]
}
],
"current-schema-id" : 1,
"partition-specs": [
{
"spec-id": 0,
"fields": [
{
"source-id": 4,
"field-id": 1000,
"name": "ts_day",
"transform": "day"
}
]
}
],
"default-spec-id": 0,
"last-partition-id": 1000,
"properties": {
"commit.retry.num-retries": "1"
},
"metadata-log": [
{
"metadata-file": "s3://bucket/.../v1.json",
"timestamp-ms": 1515100
}
],
"refs": {},
"sort-orders": [
{
"order-id": 0,
"fields": []
}
],
"default-sort-order-id": 0
}
"#;
let schema = Schema::builder()
.with_schema_id(1)
.with_fields(vec![
Arc::new(NestedField::required(
1,
"struct_name",
Type::Primitive(PrimitiveType::Fixed(1)),
)),
Arc::new(NestedField::required(
4,
"ts",
Type::Primitive(PrimitiveType::Timestamp),
)),
])
.build()
.unwrap();
let partition_spec = PartitionSpec::builder(schema.clone())
.with_spec_id(0)
.add_unbound_field(
UnboundPartitionField::builder()
.source_ids(vec![4])
.field_id(1000)
.name("ts_day".to_string())
.transform(Transform::Day)
.build()
.unwrap(),
)
.unwrap()
.build()
.unwrap();
let default_partition_type = partition_spec.partition_type(&schema).unwrap();
let expected = TableMetadata {
format_version: FormatVersion::V2,
table_uuid: Uuid::parse_str("fb072c92-a02b-11e9-ae9c-1bb7bc9eca94").unwrap(),
location: "s3://b/wh/data.db/table".to_string(),
last_updated_ms: 1515100955770,
last_column_id: 1,
schemas: HashMap::from_iter(vec![(1, Arc::new(schema))]),
current_schema_id: 1,
partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
default_partition_type,
default_spec: partition_spec.into(),
last_partition_id: 1000,
default_sort_order_id: 0,
sort_orders: HashMap::from_iter(vec![(0, SortOrder::unsorted_order().into())]),
snapshots: HashMap::default(),
current_snapshot_id: None,
last_sequence_number: 1,
properties: HashMap::from_iter(vec![(
"commit.retry.num-retries".to_string(),
"1".to_string(),
)]),
snapshot_log: Vec::new(),
metadata_log: vec![MetadataLog {
metadata_file: "s3://bucket/.../v1.json".to_string(),
timestamp_ms: 1515100,
}],
refs: HashMap::new(),
statistics: HashMap::new(),
partition_statistics: HashMap::new(),
encryption_keys: HashMap::new(),
next_row_id: INITIAL_ROW_ID,
};
let expected_json_value = serde_json::to_value(&expected).unwrap();
check_table_metadata_serde(data, expected);
let json_value = serde_json::from_str::<serde_json::Value>(data).unwrap();
assert_eq!(json_value, expected_json_value);
}
#[test]
fn test_table_data_v3() {
let data = r#"
{
"format-version" : 3,
"table-uuid": "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94",
"location": "s3://b/wh/data.db/table",
"last-sequence-number" : 1,
"last-updated-ms": 1515100955770,
"last-column-id": 1,
"next-row-id": 5,
"schemas": [
{
"schema-id" : 1,
"type" : "struct",
"fields" :[
{
"id": 4,
"name": "ts",
"required": true,
"type": "timestamp"
}
]
}
],
"current-schema-id" : 1,
"partition-specs": [
{
"spec-id": 0,
"fields": [
{
"source-id": 4,
"field-id": 1000,
"name": "ts_day",
"transform": "day"
}
]
}
],
"default-spec-id": 0,
"last-partition-id": 1000,
"properties": {
"commit.retry.num-retries": "1"
},
"metadata-log": [
{
"metadata-file": "s3://bucket/.../v1.json",
"timestamp-ms": 1515100
}
],
"refs": {},
"snapshots" : [ {
"snapshot-id" : 1,
"timestamp-ms" : 1662532818843,
"sequence-number" : 0,
"first-row-id" : 0,
"added-rows" : 4,
"key-id" : "key1",
"summary" : {
"operation" : "append"
},
"manifest-list" : "/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro",
"schema-id" : 0
}
],
"encryption-keys": [
{
"key-id": "key1",
"encrypted-by-id": "KMS",
"encrypted-key-metadata": "c29tZS1lbmNyeXB0aW9uLWtleQ==",
"properties": {
"p1": "v1"
}
}
],
"sort-orders": [
{
"order-id": 0,
"fields": []
}
],
"default-sort-order-id": 0
}
"#;
let schema = Schema::builder()
.with_schema_id(1)
.with_fields(vec![Arc::new(NestedField::required(
4,
"ts",
Type::Primitive(PrimitiveType::Timestamp),
))])
.build()
.unwrap();
let partition_spec = PartitionSpec::builder(schema.clone())
.with_spec_id(0)
.add_unbound_field(
UnboundPartitionField::builder()
.source_ids(vec![4])
.field_id(1000)
.name("ts_day".to_string())
.transform(Transform::Day)
.build()
.unwrap(),
)
.unwrap()
.build()
.unwrap();
let snapshot = Snapshot::builder()
.with_snapshot_id(1)
.with_timestamp_ms(1662532818843)
.with_sequence_number(0)
.with_row_range(0, 4)
.with_encryption_key_id(Some("key1".to_string()))
.with_summary(Summary {
operation: Operation::Append,
additional_properties: HashMap::new(),
})
.with_manifest_list("/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro".to_string())
.with_schema_id(0)
.build();
let encryption_key = EncryptedKey::builder()
.key_id("key1".to_string())
.encrypted_by_id("KMS".to_string())
.encrypted_key_metadata(
base64::prelude::BASE64_STANDARD
.decode("c29tZS1lbmNyeXB0aW9uLWtleQ==")
.unwrap(),
)
.properties(HashMap::from_iter(vec![(
"p1".to_string(),
"v1".to_string(),
)]))
.build();
let default_partition_type = partition_spec.partition_type(&schema).unwrap();
let expected = TableMetadata {
format_version: FormatVersion::V3,
table_uuid: Uuid::parse_str("fb072c92-a02b-11e9-ae9c-1bb7bc9eca94").unwrap(),
location: "s3://b/wh/data.db/table".to_string(),
last_updated_ms: 1515100955770,
last_column_id: 1,
schemas: HashMap::from_iter(vec![(1, Arc::new(schema))]),
current_schema_id: 1,
partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
default_partition_type,
default_spec: partition_spec.into(),
last_partition_id: 1000,
default_sort_order_id: 0,
sort_orders: HashMap::from_iter(vec![(0, SortOrder::unsorted_order().into())]),
snapshots: HashMap::from_iter(vec![(1, snapshot.into())]),
current_snapshot_id: None,
last_sequence_number: 1,
properties: HashMap::from_iter(vec![(
"commit.retry.num-retries".to_string(),
"1".to_string(),
)]),
snapshot_log: Vec::new(),
metadata_log: vec![MetadataLog {
metadata_file: "s3://bucket/.../v1.json".to_string(),
timestamp_ms: 1515100,
}],
refs: HashMap::new(),
statistics: HashMap::new(),
partition_statistics: HashMap::new(),
encryption_keys: HashMap::from_iter(vec![("key1".to_string(), encryption_key)]),
next_row_id: 5,
};
let expected_json_value = serde_json::to_value(&expected).unwrap();
check_table_metadata_serde(data, expected);
let json_value = serde_json::from_str::<serde_json::Value>(data).unwrap();
assert_eq!(json_value, expected_json_value);
}
#[test]
fn test_table_data_v1() {
let data = r#"
{
"format-version" : 1,
"table-uuid" : "df838b92-0b32-465d-a44e-d39936e538b7",
"location" : "/home/iceberg/warehouse/nyc/taxis",
"last-updated-ms" : 1662532818843,
"last-column-id" : 5,
"schema" : {
"type" : "struct",
"schema-id" : 0,
"fields" : [ {
"id" : 1,
"name" : "vendor_id",
"required" : false,
"type" : "long"
}, {
"id" : 2,
"name" : "trip_id",
"required" : false,
"type" : "long"
}, {
"id" : 3,
"name" : "trip_distance",
"required" : false,
"type" : "float"
}, {
"id" : 4,
"name" : "fare_amount",
"required" : false,
"type" : "double"
}, {
"id" : 5,
"name" : "store_and_fwd_flag",
"required" : false,
"type" : "string"
} ]
},
"partition-spec" : [ {
"name" : "vendor_id",
"transform" : "identity",
"source-id" : 1,
"field-id" : 1000
} ],
"last-partition-id" : 1000,
"default-sort-order-id" : 0,
"sort-orders" : [ {
"order-id" : 0,
"fields" : [ ]
} ],
"properties" : {
"owner" : "root"
},
"current-snapshot-id" : 638933773299822130,
"refs" : {
"main" : {
"snapshot-id" : 638933773299822130,
"type" : "branch"
}
},
"snapshots" : [ {
"snapshot-id" : 638933773299822130,
"timestamp-ms" : 1662532818843,
"sequence-number" : 0,
"summary" : {
"operation" : "append",
"spark.app.id" : "local-1662532784305",
"added-data-files" : "4",
"added-records" : "4",
"added-files-size" : "6001"
},
"manifest-list" : "/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro",
"schema-id" : 0
} ],
"snapshot-log" : [ {
"timestamp-ms" : 1662532818843,
"snapshot-id" : 638933773299822130
} ],
"metadata-log" : [ {
"timestamp-ms" : 1662532805245,
"metadata-file" : "/home/iceberg/warehouse/nyc/taxis/metadata/00000-8a62c37d-4573-4021-952a-c0baef7d21d0.metadata.json"
} ]
}
"#;
let schema = Schema::builder()
.with_fields(vec![
Arc::new(NestedField::optional(
1,
"vendor_id",
Type::Primitive(PrimitiveType::Long),
)),
Arc::new(NestedField::optional(
2,
"trip_id",
Type::Primitive(PrimitiveType::Long),
)),
Arc::new(NestedField::optional(
3,
"trip_distance",
Type::Primitive(PrimitiveType::Float),
)),
Arc::new(NestedField::optional(
4,
"fare_amount",
Type::Primitive(PrimitiveType::Double),
)),
Arc::new(NestedField::optional(
5,
"store_and_fwd_flag",
Type::Primitive(PrimitiveType::String),
)),
])
.build()
.unwrap();
let schema = Arc::new(schema);
let partition_spec = PartitionSpec::builder(schema.clone())
.with_spec_id(0)
.add_partition_field("vendor_id", "vendor_id", Transform::Identity)
.unwrap()
.build()
.unwrap();
let sort_order = SortOrder::builder()
.with_order_id(0)
.build_unbound()
.unwrap();
let snapshot = Snapshot::builder()
.with_snapshot_id(638933773299822130)
.with_timestamp_ms(1662532818843)
.with_sequence_number(0)
.with_schema_id(0)
.with_manifest_list("/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro")
.with_summary(Summary { operation: Operation::Append, additional_properties: HashMap::from_iter(vec![("spark.app.id".to_string(), "local-1662532784305".to_string()), ("added-data-files".to_string(), "4".to_string()), ("added-records".to_string(), "4".to_string()), ("added-files-size".to_string(), "6001".to_string())]) })
.build();
let default_partition_type = partition_spec.partition_type(&schema).unwrap();
let expected = TableMetadata {
format_version: FormatVersion::V1,
table_uuid: Uuid::parse_str("df838b92-0b32-465d-a44e-d39936e538b7").unwrap(),
location: "/home/iceberg/warehouse/nyc/taxis".to_string(),
last_updated_ms: 1662532818843,
last_column_id: 5,
schemas: HashMap::from_iter(vec![(0, schema)]),
current_schema_id: 0,
partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
default_partition_type,
default_spec: Arc::new(partition_spec),
last_partition_id: 1000,
default_sort_order_id: 0,
sort_orders: HashMap::from_iter(vec![(0, sort_order.into())]),
snapshots: HashMap::from_iter(vec![(638933773299822130, Arc::new(snapshot))]),
current_snapshot_id: Some(638933773299822130),
last_sequence_number: 0,
properties: HashMap::from_iter(vec![("owner".to_string(), "root".to_string())]),
snapshot_log: vec![SnapshotLog {
snapshot_id: 638933773299822130,
timestamp_ms: 1662532818843,
}],
metadata_log: vec![MetadataLog { metadata_file: "/home/iceberg/warehouse/nyc/taxis/metadata/00000-8a62c37d-4573-4021-952a-c0baef7d21d0.metadata.json".to_string(), timestamp_ms: 1662532805245 }],
refs: HashMap::from_iter(vec![("main".to_string(), SnapshotReference { snapshot_id: 638933773299822130, retention: SnapshotRetention::Branch { min_snapshots_to_keep: None, max_snapshot_age_ms: None, max_ref_age_ms: None } })]),
statistics: HashMap::new(),
partition_statistics: HashMap::new(),
encryption_keys: HashMap::new(),
next_row_id: INITIAL_ROW_ID,
};
check_table_metadata_serde(data, expected);
}
#[test]
fn test_table_data_v2_no_snapshots() {
let data = r#"
{
"format-version" : 2,
"table-uuid": "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94",
"location": "s3://b/wh/data.db/table",
"last-sequence-number" : 1,
"last-updated-ms": 1515100955770,
"last-column-id": 1,
"schemas": [
{
"schema-id" : 1,
"type" : "struct",
"fields" :[
{
"id": 1,
"name": "struct_name",
"required": true,
"type": "fixed[1]"
}
]
}
],
"current-schema-id" : 1,
"partition-specs": [
{
"spec-id": 0,
"fields": []
}
],
"refs": {},
"default-spec-id": 0,
"last-partition-id": 1000,
"metadata-log": [
{
"metadata-file": "s3://bucket/.../v1.json",
"timestamp-ms": 1515100
}
],
"sort-orders": [
{
"order-id": 0,
"fields": []
}
],
"default-sort-order-id": 0
}
"#;
let schema = Schema::builder()
.with_schema_id(1)
.with_fields(vec![Arc::new(NestedField::required(
1,
"struct_name",
Type::Primitive(PrimitiveType::Fixed(1)),
))])
.build()
.unwrap();
let partition_spec = PartitionSpec::builder(schema.clone())
.with_spec_id(0)
.build()
.unwrap();
let default_partition_type = partition_spec.partition_type(&schema).unwrap();
let expected = TableMetadata {
format_version: FormatVersion::V2,
table_uuid: Uuid::parse_str("fb072c92-a02b-11e9-ae9c-1bb7bc9eca94").unwrap(),
location: "s3://b/wh/data.db/table".to_string(),
last_updated_ms: 1515100955770,
last_column_id: 1,
schemas: HashMap::from_iter(vec![(1, Arc::new(schema))]),
current_schema_id: 1,
partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
default_partition_type,
default_spec: partition_spec.into(),
last_partition_id: 1000,
default_sort_order_id: 0,
sort_orders: HashMap::from_iter(vec![(0, SortOrder::unsorted_order().into())]),
snapshots: HashMap::default(),
current_snapshot_id: None,
last_sequence_number: 1,
properties: HashMap::new(),
snapshot_log: Vec::new(),
metadata_log: vec![MetadataLog {
metadata_file: "s3://bucket/.../v1.json".to_string(),
timestamp_ms: 1515100,
}],
refs: HashMap::new(),
statistics: HashMap::new(),
partition_statistics: HashMap::new(),
encryption_keys: HashMap::new(),
next_row_id: INITIAL_ROW_ID,
};
let expected_json_value = serde_json::to_value(&expected).unwrap();
check_table_metadata_serde(data, expected);
let json_value = serde_json::from_str::<serde_json::Value>(data).unwrap();
assert_eq!(json_value, expected_json_value);
}
#[test]
fn test_current_snapshot_id_must_match_main_branch() {
let data = r#"
{
"format-version" : 2,
"table-uuid": "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94",
"location": "s3://b/wh/data.db/table",
"last-sequence-number" : 1,
"last-updated-ms": 1515100955770,
"last-column-id": 1,
"schemas": [
{
"schema-id" : 1,
"type" : "struct",
"fields" :[
{
"id": 1,
"name": "struct_name",
"required": true,
"type": "fixed[1]"
},
{
"id": 4,
"name": "ts",
"required": true,
"type": "timestamp"
}
]
}
],
"current-schema-id" : 1,
"partition-specs": [
{
"spec-id": 0,
"fields": [
{
"source-id": 4,
"field-id": 1000,
"name": "ts_day",
"transform": "day"
}
]
}
],
"default-spec-id": 0,
"last-partition-id": 1000,
"properties": {
"commit.retry.num-retries": "1"
},
"metadata-log": [
{
"metadata-file": "s3://bucket/.../v1.json",
"timestamp-ms": 1515100
}
],
"sort-orders": [
{
"order-id": 0,
"fields": []
}
],
"default-sort-order-id": 0,
"current-snapshot-id" : 1,
"refs" : {
"main" : {
"snapshot-id" : 2,
"type" : "branch"
}
},
"snapshots" : [ {
"snapshot-id" : 1,
"timestamp-ms" : 1662532818843,
"sequence-number" : 0,
"summary" : {
"operation" : "append",
"spark.app.id" : "local-1662532784305",
"added-data-files" : "4",
"added-records" : "4",
"added-files-size" : "6001"
},
"manifest-list" : "/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro",
"schema-id" : 0
},
{
"snapshot-id" : 2,
"timestamp-ms" : 1662532818844,
"sequence-number" : 0,
"summary" : {
"operation" : "append",
"spark.app.id" : "local-1662532784305",
"added-data-files" : "4",
"added-records" : "4",
"added-files-size" : "6001"
},
"manifest-list" : "/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro",
"schema-id" : 0
} ]
}
"#;
let err = serde_json::from_str::<TableMetadata>(data).unwrap_err();
assert!(
err.to_string()
.contains("Current snapshot id does not match main branch")
);
}
#[test]
fn test_main_without_current() {
let data = r#"
{
"format-version" : 2,
"table-uuid": "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94",
"location": "s3://b/wh/data.db/table",
"last-sequence-number" : 1,
"last-updated-ms": 1515100955770,
"last-column-id": 1,
"schemas": [
{
"schema-id" : 1,
"type" : "struct",
"fields" :[
{
"id": 1,
"name": "struct_name",
"required": true,
"type": "fixed[1]"
},
{
"id": 4,
"name": "ts",
"required": true,
"type": "timestamp"
}
]
}
],
"current-schema-id" : 1,
"partition-specs": [
{
"spec-id": 0,
"fields": [
{
"source-id": 4,
"field-id": 1000,
"name": "ts_day",
"transform": "day"
}
]
}
],
"default-spec-id": 0,
"last-partition-id": 1000,
"properties": {
"commit.retry.num-retries": "1"
},
"metadata-log": [
{
"metadata-file": "s3://bucket/.../v1.json",
"timestamp-ms": 1515100
}
],
"sort-orders": [
{
"order-id": 0,
"fields": []
}
],
"default-sort-order-id": 0,
"refs" : {
"main" : {
"snapshot-id" : 1,
"type" : "branch"
}
},
"snapshots" : [ {
"snapshot-id" : 1,
"timestamp-ms" : 1662532818843,
"sequence-number" : 0,
"summary" : {
"operation" : "append",
"spark.app.id" : "local-1662532784305",
"added-data-files" : "4",
"added-records" : "4",
"added-files-size" : "6001"
},
"manifest-list" : "/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro",
"schema-id" : 0
} ]
}
"#;
let err = serde_json::from_str::<TableMetadata>(data).unwrap_err();
assert!(
err.to_string()
.contains("Current snapshot is not set, but main branch exists")
);
}
#[test]
fn test_branch_snapshot_missing() {
let data = r#"
{
"format-version" : 2,
"table-uuid": "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94",
"location": "s3://b/wh/data.db/table",
"last-sequence-number" : 1,
"last-updated-ms": 1515100955770,
"last-column-id": 1,
"schemas": [
{
"schema-id" : 1,
"type" : "struct",
"fields" :[
{
"id": 1,
"name": "struct_name",
"required": true,
"type": "fixed[1]"
},
{
"id": 4,
"name": "ts",
"required": true,
"type": "timestamp"
}
]
}
],
"current-schema-id" : 1,
"partition-specs": [
{
"spec-id": 0,
"fields": [
{
"source-id": 4,
"field-id": 1000,
"name": "ts_day",
"transform": "day"
}
]
}
],
"default-spec-id": 0,
"last-partition-id": 1000,
"properties": {
"commit.retry.num-retries": "1"
},
"metadata-log": [
{
"metadata-file": "s3://bucket/.../v1.json",
"timestamp-ms": 1515100
}
],
"sort-orders": [
{
"order-id": 0,
"fields": []
}
],
"default-sort-order-id": 0,
"refs" : {
"main" : {
"snapshot-id" : 1,
"type" : "branch"
},
"foo" : {
"snapshot-id" : 2,
"type" : "branch"
}
},
"snapshots" : [ {
"snapshot-id" : 1,
"timestamp-ms" : 1662532818843,
"sequence-number" : 0,
"summary" : {
"operation" : "append",
"spark.app.id" : "local-1662532784305",
"added-data-files" : "4",
"added-records" : "4",
"added-files-size" : "6001"
},
"manifest-list" : "/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro",
"schema-id" : 0
} ]
}
"#;
let err = serde_json::from_str::<TableMetadata>(data).unwrap_err();
assert!(
err.to_string().contains(
"Snapshot for reference foo does not exist in the existing snapshots list"
)
);
}
#[test]
fn test_v2_wrong_max_snapshot_sequence_number() {
let data = r#"
{
"format-version": 2,
"table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
"location": "s3://bucket/test/location",
"last-sequence-number": 1,
"last-updated-ms": 1602638573590,
"last-column-id": 3,
"current-schema-id": 0,
"schemas": [
{
"type": "struct",
"schema-id": 0,
"fields": [
{
"id": 1,
"name": "x",
"required": true,
"type": "long"
}
]
}
],
"default-spec-id": 0,
"partition-specs": [
{
"spec-id": 0,
"fields": []
}
],
"last-partition-id": 1000,
"default-sort-order-id": 0,
"sort-orders": [
{
"order-id": 0,
"fields": []
}
],
"properties": {},
"current-snapshot-id": 3055729675574597004,
"snapshots": [
{
"snapshot-id": 3055729675574597004,
"timestamp-ms": 1555100955770,
"sequence-number": 4,
"summary": {
"operation": "append"
},
"manifest-list": "s3://a/b/2.avro",
"schema-id": 0
}
],
"statistics": [],
"snapshot-log": [],
"metadata-log": []
}
"#;
let err = serde_json::from_str::<TableMetadata>(data).unwrap_err();
assert!(err.to_string().contains(
"Invalid snapshot with id 3055729675574597004 and sequence number 4 greater than last sequence number 1"
));
// Change max sequence number to 4 - should work
let data = data.replace(
r#""last-sequence-number": 1,"#,
r#""last-sequence-number": 4,"#,
);
let metadata = serde_json::from_str::<TableMetadata>(data.as_str()).unwrap();
assert_eq!(metadata.last_sequence_number, 4);
// Change max sequence number to 5 - should work
let data = data.replace(
r#""last-sequence-number": 4,"#,
r#""last-sequence-number": 5,"#,
);
let metadata = serde_json::from_str::<TableMetadata>(data.as_str()).unwrap();
assert_eq!(metadata.last_sequence_number, 5);
}
#[test]
fn test_statistic_files() {
let data = r#"
{
"format-version": 2,
"table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
"location": "s3://bucket/test/location",
"last-sequence-number": 34,
"last-updated-ms": 1602638573590,
"last-column-id": 3,
"current-schema-id": 0,
"schemas": [
{
"type": "struct",
"schema-id": 0,
"fields": [
{
"id": 1,
"name": "x",
"required": true,
"type": "long"
}
]
}
],
"default-spec-id": 0,
"partition-specs": [
{
"spec-id": 0,
"fields": []
}
],
"last-partition-id": 1000,
"default-sort-order-id": 0,
"sort-orders": [
{
"order-id": 0,
"fields": []
}
],
"properties": {},
"current-snapshot-id": 3055729675574597004,
"snapshots": [
{
"snapshot-id": 3055729675574597004,
"timestamp-ms": 1555100955770,
"sequence-number": 1,
"summary": {
"operation": "append"
},
"manifest-list": "s3://a/b/2.avro",
"schema-id": 0
}
],
"statistics": [
{
"snapshot-id": 3055729675574597004,
"statistics-path": "s3://a/b/stats.puffin",
"file-size-in-bytes": 413,
"file-footer-size-in-bytes": 42,
"blob-metadata": [
{
"type": "ndv",
"snapshot-id": 3055729675574597004,
"sequence-number": 1,
"fields": [
1
]
}
]
}
],
"snapshot-log": [],
"metadata-log": []
}
"#;
let schema = Schema::builder()
.with_schema_id(0)
.with_fields(vec![Arc::new(NestedField::required(
1,
"x",
Type::Primitive(PrimitiveType::Long),
))])
.build()
.unwrap();
let partition_spec = PartitionSpec::builder(schema.clone())
.with_spec_id(0)
.build()
.unwrap();
let snapshot = Snapshot::builder()
.with_snapshot_id(3055729675574597004)
.with_timestamp_ms(1555100955770)
.with_sequence_number(1)
.with_manifest_list("s3://a/b/2.avro")
.with_schema_id(0)
.with_summary(Summary {
operation: Operation::Append,
additional_properties: HashMap::new(),
})
.build();
let default_partition_type = partition_spec.partition_type(&schema).unwrap();
let expected = TableMetadata {
format_version: FormatVersion::V2,
table_uuid: Uuid::parse_str("9c12d441-03fe-4693-9a96-a0705ddf69c1").unwrap(),
location: "s3://bucket/test/location".to_string(),
last_updated_ms: 1602638573590,
last_column_id: 3,
schemas: HashMap::from_iter(vec![(0, Arc::new(schema))]),
current_schema_id: 0,
partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
default_partition_type,
default_spec: Arc::new(partition_spec),
last_partition_id: 1000,
default_sort_order_id: 0,
sort_orders: HashMap::from_iter(vec![(0, SortOrder::unsorted_order().into())]),
snapshots: HashMap::from_iter(vec![(3055729675574597004, Arc::new(snapshot))]),
current_snapshot_id: Some(3055729675574597004),
last_sequence_number: 34,
properties: HashMap::new(),
snapshot_log: Vec::new(),
metadata_log: Vec::new(),
statistics: HashMap::from_iter(vec![(3055729675574597004, StatisticsFile {
snapshot_id: 3055729675574597004,
statistics_path: "s3://a/b/stats.puffin".to_string(),
file_size_in_bytes: 413,
file_footer_size_in_bytes: 42,
key_metadata: None,
blob_metadata: vec![BlobMetadata {
snapshot_id: 3055729675574597004,
sequence_number: 1,
fields: vec![1],
r#type: "ndv".to_string(),
properties: HashMap::new(),
}],
})]),
partition_statistics: HashMap::new(),
refs: HashMap::from_iter(vec![("main".to_string(), SnapshotReference {
snapshot_id: 3055729675574597004,
retention: SnapshotRetention::Branch {
min_snapshots_to_keep: None,
max_snapshot_age_ms: None,
max_ref_age_ms: None,
},
})]),
encryption_keys: HashMap::new(),
next_row_id: INITIAL_ROW_ID,
};
check_table_metadata_serde(data, expected);
}
#[test]
fn test_partition_statistics_file() {
let data = r#"
{
"format-version": 2,
"table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
"location": "s3://bucket/test/location",
"last-sequence-number": 34,
"last-updated-ms": 1602638573590,
"last-column-id": 3,
"current-schema-id": 0,
"schemas": [
{
"type": "struct",
"schema-id": 0,
"fields": [
{
"id": 1,
"name": "x",
"required": true,
"type": "long"
}
]
}
],
"default-spec-id": 0,
"partition-specs": [
{
"spec-id": 0,
"fields": []
}
],
"last-partition-id": 1000,
"default-sort-order-id": 0,
"sort-orders": [
{
"order-id": 0,
"fields": []
}
],
"properties": {},
"current-snapshot-id": 3055729675574597004,
"snapshots": [
{
"snapshot-id": 3055729675574597004,
"timestamp-ms": 1555100955770,
"sequence-number": 1,
"summary": {
"operation": "append"
},
"manifest-list": "s3://a/b/2.avro",
"schema-id": 0
}
],
"partition-statistics": [
{
"snapshot-id": 3055729675574597004,
"statistics-path": "s3://a/b/partition-stats.parquet",
"file-size-in-bytes": 43
}
],
"snapshot-log": [],
"metadata-log": []
}
"#;
let schema = Schema::builder()
.with_schema_id(0)
.with_fields(vec![Arc::new(NestedField::required(
1,
"x",
Type::Primitive(PrimitiveType::Long),
))])
.build()
.unwrap();
let partition_spec = PartitionSpec::builder(schema.clone())
.with_spec_id(0)
.build()
.unwrap();
let snapshot = Snapshot::builder()
.with_snapshot_id(3055729675574597004)
.with_timestamp_ms(1555100955770)
.with_sequence_number(1)
.with_manifest_list("s3://a/b/2.avro")
.with_schema_id(0)
.with_summary(Summary {
operation: Operation::Append,
additional_properties: HashMap::new(),
})
.build();
let default_partition_type = partition_spec.partition_type(&schema).unwrap();
let expected = TableMetadata {
format_version: FormatVersion::V2,
table_uuid: Uuid::parse_str("9c12d441-03fe-4693-9a96-a0705ddf69c1").unwrap(),
location: "s3://bucket/test/location".to_string(),
last_updated_ms: 1602638573590,
last_column_id: 3,
schemas: HashMap::from_iter(vec![(0, Arc::new(schema))]),
current_schema_id: 0,
partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
default_spec: Arc::new(partition_spec),
default_partition_type,
last_partition_id: 1000,
default_sort_order_id: 0,
sort_orders: HashMap::from_iter(vec![(0, SortOrder::unsorted_order().into())]),
snapshots: HashMap::from_iter(vec![(3055729675574597004, Arc::new(snapshot))]),
current_snapshot_id: Some(3055729675574597004),
last_sequence_number: 34,
properties: HashMap::new(),
snapshot_log: Vec::new(),
metadata_log: Vec::new(),
statistics: HashMap::new(),
partition_statistics: HashMap::from_iter(vec![(
3055729675574597004,
PartitionStatisticsFile {
snapshot_id: 3055729675574597004,
statistics_path: "s3://a/b/partition-stats.parquet".to_string(),
file_size_in_bytes: 43,
},
)]),
refs: HashMap::from_iter(vec![("main".to_string(), SnapshotReference {
snapshot_id: 3055729675574597004,
retention: SnapshotRetention::Branch {
min_snapshots_to_keep: None,
max_snapshot_age_ms: None,
max_ref_age_ms: None,
},
})]),
encryption_keys: HashMap::new(),
next_row_id: INITIAL_ROW_ID,
};
check_table_metadata_serde(data, expected);
}
#[test]
fn test_invalid_table_uuid() -> Result<()> {
let data = r#"
{
"format-version" : 2,
"table-uuid": "xxxx"
}
"#;
assert!(serde_json::from_str::<TableMetadata>(data).is_err());
Ok(())
}
#[test]
fn test_deserialize_table_data_v2_invalid_format_version() -> Result<()> {
let data = r#"
{
"format-version" : 1
}
"#;
assert!(serde_json::from_str::<TableMetadata>(data).is_err());
Ok(())
}
#[test]
fn test_table_metadata_v3_valid_minimal() {
let metadata_str =
fs::read_to_string("testdata/table_metadata/TableMetadataV3ValidMinimal.json").unwrap();
let table_metadata = serde_json::from_str::<TableMetadata>(&metadata_str).unwrap();
assert_eq!(table_metadata.format_version, FormatVersion::V3);
let schema = Schema::builder()
.with_schema_id(0)
.with_fields(vec![
Arc::new(
NestedField::required(1, "x", Type::Primitive(PrimitiveType::Long))
.with_initial_default(Literal::Primitive(PrimitiveLiteral::Long(1)))
.with_write_default(Literal::Primitive(PrimitiveLiteral::Long(1))),
),
Arc::new(
NestedField::required(2, "y", Type::Primitive(PrimitiveType::Long))
.with_doc("comment"),
),
Arc::new(NestedField::required(
3,
"z",
Type::Primitive(PrimitiveType::Long),
)),
])
.build()
.unwrap();
let partition_spec = PartitionSpec::builder(schema.clone())
.with_spec_id(0)
.add_unbound_field(
UnboundPartitionField::builder()
.source_ids(vec![1])
.field_id(1000)
.name("x".to_string())
.transform(Transform::Identity)
.build()
.unwrap(),
)
.unwrap()
.build()
.unwrap();
let sort_order = SortOrder::builder()
.with_order_id(3)
.with_sort_field(SortField {
source_id: 2,
transform: Transform::Identity,
direction: SortDirection::Ascending,
null_order: NullOrder::First,
})
.with_sort_field(SortField {
source_id: 3,
transform: Transform::Bucket(4),
direction: SortDirection::Descending,
null_order: NullOrder::Last,
})
.build_unbound()
.unwrap();
let default_partition_type = partition_spec.partition_type(&schema).unwrap();
let expected = TableMetadata {
format_version: FormatVersion::V3,
table_uuid: Uuid::parse_str("9c12d441-03fe-4693-9a96-a0705ddf69c1").unwrap(),
location: "s3://bucket/test/location".to_string(),
last_updated_ms: 1602638573590,
last_column_id: 3,
schemas: HashMap::from_iter(vec![(0, Arc::new(schema))]),
current_schema_id: 0,
partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
default_spec: Arc::new(partition_spec),
default_partition_type,
last_partition_id: 1000,
default_sort_order_id: 3,
sort_orders: HashMap::from_iter(vec![(3, sort_order.into())]),
snapshots: HashMap::default(),
current_snapshot_id: None,
last_sequence_number: 34,
properties: HashMap::new(),
snapshot_log: Vec::new(),
metadata_log: Vec::new(),
refs: HashMap::new(),
statistics: HashMap::new(),
partition_statistics: HashMap::new(),
encryption_keys: HashMap::new(),
next_row_id: 0, // V3 specific field from the JSON
};
check_table_metadata_serde(&metadata_str, expected);
}
#[test]
fn test_table_metadata_v2_file_valid() {
let metadata =
fs::read_to_string("testdata/table_metadata/TableMetadataV2Valid.json").unwrap();
let schema1 = Schema::builder()
.with_schema_id(0)
.with_fields(vec![Arc::new(NestedField::required(
1,
"x",
Type::Primitive(PrimitiveType::Long),
))])
.build()
.unwrap();
let schema2 = Schema::builder()
.with_schema_id(1)
.with_fields(vec![
Arc::new(NestedField::required(
1,
"x",
Type::Primitive(PrimitiveType::Long),
)),
Arc::new(
NestedField::required(2, "y", Type::Primitive(PrimitiveType::Long))
.with_doc("comment"),
),
Arc::new(NestedField::required(
3,
"z",
Type::Primitive(PrimitiveType::Long),
)),
])
.with_identifier_field_ids(vec![1, 2])
.build()
.unwrap();
let partition_spec = PartitionSpec::builder(schema2.clone())
.with_spec_id(0)
.add_unbound_field(
UnboundPartitionField::builder()
.source_ids(vec![1])
.field_id(1000)
.name("x".to_string())
.transform(Transform::Identity)
.build()
.unwrap(),
)
.unwrap()
.build()
.unwrap();
let sort_order = SortOrder::builder()
.with_order_id(3)
.with_sort_field(SortField {
source_id: 2,
transform: Transform::Identity,
direction: SortDirection::Ascending,
null_order: NullOrder::First,
})
.with_sort_field(SortField {
source_id: 3,
transform: Transform::Bucket(4),
direction: SortDirection::Descending,
null_order: NullOrder::Last,
})
.build_unbound()
.unwrap();
let snapshot1 = Snapshot::builder()
.with_snapshot_id(3051729675574597004)
.with_timestamp_ms(1515100955770)
.with_sequence_number(0)
.with_manifest_list("s3://a/b/1.avro")
.with_summary(Summary {
operation: Operation::Append,
additional_properties: HashMap::new(),
})
.build();
let snapshot2 = Snapshot::builder()
.with_snapshot_id(3055729675574597004)
.with_parent_snapshot_id(Some(3051729675574597004))
.with_timestamp_ms(1555100955770)
.with_sequence_number(1)
.with_schema_id(1)
.with_manifest_list("s3://a/b/2.avro")
.with_summary(Summary {
operation: Operation::Append,
additional_properties: HashMap::new(),
})
.build();
let default_partition_type = partition_spec.partition_type(&schema2).unwrap();
let expected = TableMetadata {
format_version: FormatVersion::V2,
table_uuid: Uuid::parse_str("9c12d441-03fe-4693-9a96-a0705ddf69c1").unwrap(),
location: "s3://bucket/test/location".to_string(),
last_updated_ms: 1602638573590,
last_column_id: 3,
schemas: HashMap::from_iter(vec![(0, Arc::new(schema1)), (1, Arc::new(schema2))]),
current_schema_id: 1,
partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
default_spec: Arc::new(partition_spec),
default_partition_type,
last_partition_id: 1000,
default_sort_order_id: 3,
sort_orders: HashMap::from_iter(vec![(3, sort_order.into())]),
snapshots: HashMap::from_iter(vec![
(3051729675574597004, Arc::new(snapshot1)),
(3055729675574597004, Arc::new(snapshot2)),
]),
current_snapshot_id: Some(3055729675574597004),
last_sequence_number: 34,
properties: HashMap::new(),
snapshot_log: vec![
SnapshotLog {
snapshot_id: 3051729675574597004,
timestamp_ms: 1515100955770,
},
SnapshotLog {
snapshot_id: 3055729675574597004,
timestamp_ms: 1555100955770,
},
],
metadata_log: Vec::new(),
refs: HashMap::from_iter(vec![("main".to_string(), SnapshotReference {
snapshot_id: 3055729675574597004,
retention: SnapshotRetention::Branch {
min_snapshots_to_keep: None,
max_snapshot_age_ms: None,
max_ref_age_ms: None,
},
})]),
statistics: HashMap::new(),
partition_statistics: HashMap::new(),
encryption_keys: HashMap::new(),
next_row_id: INITIAL_ROW_ID,
};
check_table_metadata_serde(&metadata, expected);
}
#[test]
fn test_table_metadata_v2_file_valid_minimal() {
let metadata =
fs::read_to_string("testdata/table_metadata/TableMetadataV2ValidMinimal.json").unwrap();
let schema = Schema::builder()
.with_schema_id(0)
.with_fields(vec![
Arc::new(NestedField::required(
1,
"x",
Type::Primitive(PrimitiveType::Long),
)),
Arc::new(
NestedField::required(2, "y", Type::Primitive(PrimitiveType::Long))
.with_doc("comment"),
),
Arc::new(NestedField::required(
3,
"z",
Type::Primitive(PrimitiveType::Long),
)),
])
.build()
.unwrap();
let partition_spec = PartitionSpec::builder(schema.clone())
.with_spec_id(0)
.add_unbound_field(
UnboundPartitionField::builder()
.source_ids(vec![1])
.field_id(1000)
.name("x".to_string())
.transform(Transform::Identity)
.build()
.unwrap(),
)
.unwrap()
.build()
.unwrap();
let sort_order = SortOrder::builder()
.with_order_id(3)
.with_sort_field(SortField {
source_id: 2,
transform: Transform::Identity,
direction: SortDirection::Ascending,
null_order: NullOrder::First,
})
.with_sort_field(SortField {
source_id: 3,
transform: Transform::Bucket(4),
direction: SortDirection::Descending,
null_order: NullOrder::Last,
})
.build_unbound()
.unwrap();
let default_partition_type = partition_spec.partition_type(&schema).unwrap();
let expected = TableMetadata {
format_version: FormatVersion::V2,
table_uuid: Uuid::parse_str("9c12d441-03fe-4693-9a96-a0705ddf69c1").unwrap(),
location: "s3://bucket/test/location".to_string(),
last_updated_ms: 1602638573590,
last_column_id: 3,
schemas: HashMap::from_iter(vec![(0, Arc::new(schema))]),
current_schema_id: 0,
partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
default_partition_type,
default_spec: Arc::new(partition_spec),
last_partition_id: 1000,
default_sort_order_id: 3,
sort_orders: HashMap::from_iter(vec![(3, sort_order.into())]),
snapshots: HashMap::default(),
current_snapshot_id: None,
last_sequence_number: 34,
properties: HashMap::new(),
snapshot_log: vec![],
metadata_log: Vec::new(),
refs: HashMap::new(),
statistics: HashMap::new(),
partition_statistics: HashMap::new(),
encryption_keys: HashMap::new(),
next_row_id: INITIAL_ROW_ID,
};
check_table_metadata_serde(&metadata, expected);
}
#[test]
fn test_table_metadata_v1_file_valid() {
let metadata =
fs::read_to_string("testdata/table_metadata/TableMetadataV1Valid.json").unwrap();
let schema = Schema::builder()
.with_schema_id(0)
.with_fields(vec![
Arc::new(NestedField::required(
1,
"x",
Type::Primitive(PrimitiveType::Long),
)),
Arc::new(
NestedField::required(2, "y", Type::Primitive(PrimitiveType::Long))
.with_doc("comment"),
),
Arc::new(NestedField::required(
3,
"z",
Type::Primitive(PrimitiveType::Long),
)),
])
.build()
.unwrap();
let partition_spec = PartitionSpec::builder(schema.clone())
.with_spec_id(0)
.add_unbound_field(
UnboundPartitionField::builder()
.source_ids(vec![1])
.field_id(1000)
.name("x".to_string())
.transform(Transform::Identity)
.build()
.unwrap(),
)
.unwrap()
.build()
.unwrap();
let default_partition_type = partition_spec.partition_type(&schema).unwrap();
let expected = TableMetadata {
format_version: FormatVersion::V1,
table_uuid: Uuid::parse_str("d20125c8-7284-442c-9aea-15fee620737c").unwrap(),
location: "s3://bucket/test/location".to_string(),
last_updated_ms: 1602638573874,
last_column_id: 3,
schemas: HashMap::from_iter(vec![(0, Arc::new(schema))]),
current_schema_id: 0,
partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
default_spec: Arc::new(partition_spec),
default_partition_type,
last_partition_id: 0,
default_sort_order_id: 0,
// Sort order is added during deserialization for V2 compatibility
sort_orders: HashMap::from_iter(vec![(0, SortOrder::unsorted_order().into())]),
snapshots: HashMap::new(),
current_snapshot_id: None,
last_sequence_number: 0,
properties: HashMap::new(),
snapshot_log: vec![],
metadata_log: Vec::new(),
refs: HashMap::new(),
statistics: HashMap::new(),
partition_statistics: HashMap::new(),
encryption_keys: HashMap::new(),
next_row_id: INITIAL_ROW_ID,
};
check_table_metadata_serde(&metadata, expected);
}
#[test]
fn test_empty_snapshot_id_is_normalized_to_none() {
let metadata =
fs::read_to_string("testdata/table_metadata/TableMetadataV1Valid.json").unwrap();
let deserialized: TableMetadata = serde_json::from_str(&metadata).unwrap();
assert_eq!(
deserialized.current_snapshot_id(),
None,
"current_snapshot_id of -1 should be deserialized as None"
);
}
#[test]
fn test_table_metadata_v1_compat() {
let metadata =
fs::read_to_string("testdata/table_metadata/TableMetadataV1Compat.json").unwrap();
// Deserialize the JSON to verify it works
let desered_type: TableMetadata = serde_json::from_str(&metadata)
.expect("Failed to deserialize TableMetadataV1Compat.json");
// Verify some key fields match
assert_eq!(desered_type.format_version(), FormatVersion::V1);
assert_eq!(
desered_type.uuid(),
Uuid::parse_str("3276010d-7b1d-488c-98d8-9025fc4fde6b").unwrap()
);
assert_eq!(
desered_type.location(),
"s3://bucket/warehouse/iceberg/glue.db/table_name"
);
assert_eq!(desered_type.last_updated_ms(), 1727773114005);
assert_eq!(desered_type.current_schema_id(), 0);
}
#[test]
fn test_table_metadata_v1_schemas_without_current_id() {
let metadata = fs::read_to_string(
"testdata/table_metadata/TableMetadataV1SchemasWithoutCurrentId.json",
)
.unwrap();
// Deserialize the JSON - this should succeed by using the 'schema' field instead of 'schemas'
let desered_type: TableMetadata = serde_json::from_str(&metadata)
.expect("Failed to deserialize TableMetadataV1SchemasWithoutCurrentId.json");
// Verify it used the 'schema' field
assert_eq!(desered_type.format_version(), FormatVersion::V1);
assert_eq!(
desered_type.uuid(),
Uuid::parse_str("d20125c8-7284-442c-9aea-15fee620737c").unwrap()
);
// Get the schema and verify it has the expected fields
let schema = desered_type.current_schema();
assert_eq!(schema.as_struct().fields().len(), 3);
assert_eq!(schema.as_struct().fields()[0].name, "x");
assert_eq!(schema.as_struct().fields()[1].name, "y");
assert_eq!(schema.as_struct().fields()[2].name, "z");
}
#[test]
fn test_table_metadata_v1_no_valid_schema() {
let metadata =
fs::read_to_string("testdata/table_metadata/TableMetadataV1NoValidSchema.json")
.unwrap();
// Deserialize the JSON - this should fail because neither schemas + current_schema_id nor schema is valid
let desered: Result<TableMetadata, serde_json::Error> = serde_json::from_str(&metadata);
assert!(desered.is_err());
let error_message = desered.unwrap_err().to_string();
assert!(
error_message.contains("No valid schema configuration found"),
"Expected error about no valid schema configuration, got: {error_message}"
);
}
#[test]
fn test_table_metadata_v1_partition_specs_without_default_id() {
let metadata = fs::read_to_string(
"testdata/table_metadata/TableMetadataV1PartitionSpecsWithoutDefaultId.json",
)
.unwrap();
// Deserialize the JSON - this should succeed by inferring default_spec_id as the max spec ID
let desered_type: TableMetadata = serde_json::from_str(&metadata)
.expect("Failed to deserialize TableMetadataV1PartitionSpecsWithoutDefaultId.json");
// Verify basic metadata
assert_eq!(desered_type.format_version(), FormatVersion::V1);
assert_eq!(
desered_type.uuid(),
Uuid::parse_str("d20125c8-7284-442c-9aea-15fee620737c").unwrap()
);
// Verify partition specs
assert_eq!(desered_type.default_partition_spec_id(), 2); // Should pick the largest spec ID (2)
assert_eq!(desered_type.partition_specs.len(), 2);
// Verify the default spec has the expected fields
let default_spec = &desered_type.default_spec;
assert_eq!(default_spec.spec_id(), 2);
assert_eq!(default_spec.fields().len(), 1);
assert_eq!(default_spec.fields()[0].name, "y");
assert_eq!(default_spec.fields()[0].transform, Transform::Identity);
assert_eq!(default_spec.fields()[0].source_id, 2);
}
#[test]
fn test_table_metadata_v2_schema_not_found() {
let metadata =
fs::read_to_string("testdata/table_metadata/TableMetadataV2CurrentSchemaNotFound.json")
.unwrap();
let desered: Result<TableMetadata, serde_json::Error> = serde_json::from_str(&metadata);
assert_eq!(
desered.unwrap_err().to_string(),
"DataInvalid => No schema exists with the current schema id 2."
)
}
#[test]
fn test_table_metadata_v2_missing_sort_order() {
let metadata =
fs::read_to_string("testdata/table_metadata/TableMetadataV2MissingSortOrder.json")
.unwrap();
let desered: Result<TableMetadata, serde_json::Error> = serde_json::from_str(&metadata);
assert_eq!(
desered.unwrap_err().to_string(),
"data did not match any variant of untagged enum TableMetadataEnum"
)
}
#[test]
fn test_table_metadata_v2_missing_partition_specs() {
let metadata =
fs::read_to_string("testdata/table_metadata/TableMetadataV2MissingPartitionSpecs.json")
.unwrap();
let desered: Result<TableMetadata, serde_json::Error> = serde_json::from_str(&metadata);
assert_eq!(
desered.unwrap_err().to_string(),
"data did not match any variant of untagged enum TableMetadataEnum"
)
}
#[test]
fn test_table_metadata_v2_missing_last_partition_id() {
let metadata = fs::read_to_string(
"testdata/table_metadata/TableMetadataV2MissingLastPartitionId.json",
)
.unwrap();
let desered: Result<TableMetadata, serde_json::Error> = serde_json::from_str(&metadata);
assert_eq!(
desered.unwrap_err().to_string(),
"data did not match any variant of untagged enum TableMetadataEnum"
)
}
#[test]
fn test_table_metadata_v2_missing_schemas() {
let metadata =
fs::read_to_string("testdata/table_metadata/TableMetadataV2MissingSchemas.json")
.unwrap();
let desered: Result<TableMetadata, serde_json::Error> = serde_json::from_str(&metadata);
assert_eq!(
desered.unwrap_err().to_string(),
"data did not match any variant of untagged enum TableMetadataEnum"
)
}
#[test]
fn test_table_metadata_v2_unsupported_version() {
let metadata =
fs::read_to_string("testdata/table_metadata/TableMetadataUnsupportedVersion.json")
.unwrap();
let desered: Result<TableMetadata, serde_json::Error> = serde_json::from_str(&metadata);
assert_eq!(
desered.unwrap_err().to_string(),
"data did not match any variant of untagged enum TableMetadataEnum"
)
}
#[test]
fn test_order_of_format_version() {
assert!(FormatVersion::V1 < FormatVersion::V2);
assert_eq!(FormatVersion::V1, FormatVersion::V1);
assert_eq!(FormatVersion::V2, FormatVersion::V2);
}
#[test]
fn test_deserialize_default_partition_spec_with_dropped_source() {
for version in [1, 2, 3] {
for transform in ["identity", "truncate[4]", "bucket[16]"] {
let mut metadata = serde_json::json!({
"format-version": version,
"table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
"location": "s3://bucket/table",
"last-sequence-number": 0,
"last-updated-ms": 1602638573590_i64,
"last-column-id": 2,
"current-schema-id": 1,
"schemas": [
{
"schema-id": 0,
"type": "struct",
"fields": [
{"id": 1, "name": "x", "required": false, "type": "long"},
{"id": 2, "name": "y", "required": false, "type": "long"}
]
},
{
"schema-id": 1,
"type": "struct",
"fields": [
{"id": 2, "name": "y", "required": false, "type": "long"}
]
}
],
"default-spec-id": 0,
"partition-specs": [{"spec-id": 0, "fields": [{
"source-id": 1, "field-id": 1000, "name": "x_partition",
"transform": transform
}]}],
"last-partition-id": 1000,
"default-sort-order-id": 0,
"sort-orders": [{"order-id": 0, "fields": []}],
"next-row-id": 0
});
// Retaining the source in history does not make it writable using
// the current schema and default spec.
let err = serde_json::from_value::<TableMetadata>(metadata.clone()).unwrap_err();
assert!(err.to_string().contains(
"Default partition spec 0 references missing source field 1 in current schema 1"
));
// The same spec is allowed once it becomes historical.
metadata["partition-specs"]
.as_array_mut()
.unwrap()
.push(serde_json::json!({"spec-id": 1, "fields": []}));
metadata["default-spec-id"] = serde_json::json!(1);
serde_json::from_value::<TableMetadata>(metadata.clone()).unwrap();
// A voided field does not require its dropped source, even in the default spec.
metadata["default-spec-id"] = serde_json::json!(0);
metadata["partition-specs"][0]["fields"][0]["transform"] =
serde_json::json!("void");
serde_json::from_value::<TableMetadata>(metadata).unwrap();
}
}
}
#[test]
fn test_default_partition_spec() {
let default_spec_id = 1234;
let mut table_meta_data = get_test_table_metadata("TableMetadataV2Valid.json");
let partition_spec = PartitionSpec::unpartition_spec();
table_meta_data.default_spec = partition_spec.clone().into();
table_meta_data
.partition_specs
.insert(default_spec_id, Arc::new(partition_spec));
assert_eq!(
(*table_meta_data.default_partition_spec().clone()).clone(),
(*table_meta_data
.partition_spec_by_id(default_spec_id)
.unwrap()
.clone())
.clone()
);
}
#[test]
fn test_default_sort_order() {
let default_sort_order_id = 1234;
let mut table_meta_data = get_test_table_metadata("TableMetadataV2Valid.json");
table_meta_data.default_sort_order_id = default_sort_order_id;
table_meta_data
.sort_orders
.insert(default_sort_order_id, Arc::new(SortOrder::default()));
assert_eq!(
table_meta_data.default_sort_order(),
table_meta_data
.sort_orders
.get(&default_sort_order_id)
.unwrap()
)
}
#[test]
fn test_table_metadata_builder_from_table_creation() {
let table_creation = TableCreation::builder()
.location("s3://db/table".to_string())
.name("table".to_string())
.properties(HashMap::new())
.schema(Schema::builder().build().unwrap())
.build();
let table_metadata = TableMetadataBuilder::from_table_creation(table_creation)
.unwrap()
.build()
.unwrap()
.metadata;
assert_eq!(table_metadata.location, "s3://db/table");
assert_eq!(table_metadata.schemas.len(), 1);
assert_eq!(
table_metadata
.schemas
.get(&0)
.unwrap()
.as_struct()
.fields()
.len(),
0
);
assert_eq!(table_metadata.properties.len(), 0);
assert_eq!(
table_metadata.partition_specs,
HashMap::from([(
0,
Arc::new(
PartitionSpec::builder(table_metadata.schemas.get(&0).unwrap().clone())
.with_spec_id(0)
.build()
.unwrap()
)
)])
);
assert_eq!(
table_metadata.sort_orders,
HashMap::from([(
0,
Arc::new(SortOrder {
order_id: 0,
fields: vec![]
})
)])
);
}
#[tokio::test]
async fn test_table_metadata_read_write() {
// Create a temporary directory for our test
let temp_dir = TempDir::new().unwrap();
let temp_path = temp_dir.path().to_str().unwrap();
// Create a FileIO instance
let file_io = FileIO::new_with_fs();
// Use an existing test metadata from the test files
let original_metadata: TableMetadata =
get_test_table_metadata_at("TableMetadataV2Valid.json", temp_path);
// Define the metadata location
let metadata_location =
MetadataLocation::try_new_with_metadata(&original_metadata).unwrap();
let metadata_location_str = metadata_location.to_string();
// Write the metadata
original_metadata
.write_to(&file_io, &metadata_location)
.await
.unwrap();
// Verify the file exists
assert!(fs::metadata(&metadata_location_str).is_ok());
// Read the metadata back
let read_metadata = TableMetadata::read_from(&file_io, &metadata_location_str)
.await
.unwrap();
// Verify the metadata matches
assert_eq!(read_metadata, original_metadata);
}
#[tokio::test]
async fn test_table_metadata_read_compressed() {
let temp_dir = TempDir::new().unwrap();
let metadata_location = temp_dir.path().join("v1.gz.metadata.json");
let original_metadata: TableMetadata = get_test_table_metadata("TableMetadataV2Valid.json");
let json = serde_json::to_string(&original_metadata).unwrap();
let compressed = CompressionCodec::gzip_default()
.compress(json.into_bytes())
.expect("failed to compress metadata");
fs::write(&metadata_location, &compressed).expect("failed to write metadata");
// Read the metadata back
let file_io = FileIO::new_with_fs();
let metadata_location = metadata_location.to_str().unwrap();
let read_metadata = TableMetadata::read_from(&file_io, metadata_location)
.await
.unwrap();
// Verify the metadata matches
assert_eq!(read_metadata, original_metadata);
}
#[tokio::test]
async fn test_table_metadata_read_nonexistent_file() {
// Create a FileIO instance
let file_io = FileIO::new_with_fs();
// Try to read a non-existent file
let result = TableMetadata::read_from(&file_io, "/nonexistent/path/metadata.json").await;
// Verify it returns an error
assert!(result.is_err());
}
#[tokio::test]
async fn test_table_metadata_write_with_gzip_compression() {
let temp_dir = TempDir::new().unwrap();
let temp_path = temp_dir.path().to_str().unwrap();
let file_io = FileIO::new_with_fs();
// Get a test metadata and add gzip compression property
let original_metadata: TableMetadata =
get_test_table_metadata_at("TableMetadataV2Valid.json", temp_path);
// Modify properties to enable gzip compression (using mixed case to test case-insensitive matching)
let mut props = original_metadata.properties.clone();
props.insert(
TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(),
"GziP".to_string(),
);
// Use builder to create new metadata with updated properties
let compressed_metadata =
TableMetadataBuilder::new_from_metadata(original_metadata.clone(), None)
.assign_uuid(original_metadata.table_uuid)
.set_properties(props.clone())
.unwrap()
.build()
.unwrap()
.metadata;
// Create MetadataLocation with compression codec from metadata
let metadata_location =
MetadataLocation::try_new_with_metadata(&compressed_metadata).unwrap();
let metadata_location_str = metadata_location.to_string();
// Verify the location has the .gz extension
assert!(metadata_location_str.contains(".gz.metadata.json"));
// Write the metadata with compression
compressed_metadata
.write_to(&file_io, &metadata_location)
.await
.unwrap();
// Verify the compressed file exists
assert!(std::path::Path::new(&metadata_location_str).exists());
// Read the raw file and check it's gzip compressed
let raw_content = fs::read(&metadata_location_str).unwrap();
assert!(raw_content.len() > 2);
assert_eq!(raw_content[0], 0x1F); // gzip magic number
assert_eq!(raw_content[1], 0x8B); // gzip magic number
// Read the metadata back using the compressed location
let read_metadata = TableMetadata::read_from(&file_io, &metadata_location_str)
.await
.unwrap();
The file goes on; the console shows all of it.