vendor/iceberg-arrow59/src/catalog/mod.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.
//! Catalog API for Apache Iceberg
pub mod memory;
mod metadata_location;
mod session;
pub(crate) mod utils;
use std::collections::HashMap;
use std::fmt::{Debug, Display};
use std::future::Future;
use std::mem::take;
use std::ops::Deref;
use std::str::FromStr;
use std::sync::Arc;
use _serde::{deserialize_snapshot, serialize_snapshot};
use async_trait::async_trait;
pub use memory::MemoryCatalog;
pub use metadata_location::*;
#[cfg(test)]
use mockall::automock;
use serde_derive::{Deserialize, Serialize};
pub use session::*;
use typed_builder::TypedBuilder;
use uuid::Uuid;
use crate::encryption::kms::KmsClientFactory;
use crate::error::invalid_data;
use crate::io::StorageFactory;
use crate::runtime::Runtime;
use crate::spec::{
EncryptedKey, FormatVersion, PartitionStatisticsFile, Schema, SchemaId, Snapshot,
SnapshotReference, SortOrder, StatisticsFile, TableMetadata, TableMetadataBuilder,
UnboundPartitionSpec, ViewFormatVersion, ViewRepresentations, ViewVersion,
};
use crate::table::Table;
use crate::{Error, ErrorKind, Result};
/// The catalog API for Iceberg Rust.
#[async_trait]
#[cfg_attr(test, automock)]
pub trait Catalog: Debug + Sync + Send {
/// List namespaces inside the catalog.
async fn list_namespaces(&self, parent: Option<&NamespaceIdent>)
-> Result<Vec<NamespaceIdent>>;
/// Create a new namespace inside the catalog.
async fn create_namespace(
&self,
namespace: &NamespaceIdent,
properties: HashMap<String, String>,
) -> Result<Namespace>;
/// Get a namespace information from the catalog.
async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace>;
/// Check if namespace exists in catalog.
async fn namespace_exists(&self, namespace: &NamespaceIdent) -> Result<bool>;
/// Update a namespace inside the catalog.
///
/// # Behavior
///
/// The properties must be the full set of namespace.
async fn update_namespace(
&self,
namespace: &NamespaceIdent,
properties: HashMap<String, String>,
) -> Result<()>;
/// Drop a namespace from the catalog, or returns error if it doesn't exist.
async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()>;
/// List tables from namespace.
async fn list_tables(&self, namespace: &NamespaceIdent) -> Result<Vec<TableIdent>>;
/// Create a new table inside the namespace.
async fn create_table(
&self,
namespace: &NamespaceIdent,
creation: TableCreation,
) -> Result<Table>;
/// Load table from the catalog.
async fn load_table(&self, table: &TableIdent) -> Result<Table>;
/// Drop a table from the catalog, or returns error if it doesn't exist.
async fn drop_table(&self, table: &TableIdent) -> Result<()>;
/// Drop a table from the catalog and delete the underlying table data.
///
/// Implementations should load the table metadata, drop the table
/// from the catalog, then delete all associated data and metadata files.
/// The [`drop_table_data`](utils::drop_table_data) utility function can
/// be used for the file cleanup step.
async fn purge_table(&self, table: &TableIdent) -> Result<()>;
/// Check if a table exists in the catalog.
async fn table_exists(&self, table: &TableIdent) -> Result<bool>;
/// Rename a table in the catalog.
async fn rename_table(&self, src: &TableIdent, dest: &TableIdent) -> Result<()>;
/// Register an existing table to the catalog.
async fn register_table(&self, table: &TableIdent, metadata_location: String) -> Result<Table>;
/// Update a table to the catalog.
async fn update_table(&self, commit: TableCommit) -> Result<Table>;
}
/// Common interface for all catalog builders.
pub trait CatalogBuilder: Default + Debug + Send + Sync {
/// The catalog type that this builder creates.
type C: Catalog;
/// Set a custom StorageFactory to use for storage operations.
///
/// When a StorageFactory is provided, the catalog will use it to build FileIO
/// instances for all storage operations instead of using the default factory.
///
/// # Arguments
///
/// * `storage_factory` - The StorageFactory to use for creating storage instances
///
/// # Example
///
/// ```rust,ignore
/// use iceberg::CatalogBuilder;
/// use iceberg::io::StorageFactory;
/// use iceberg_storage_opendal::OpenDalStorageFactory;
/// use std::sync::Arc;
///
/// let catalog = MyCatalogBuilder::default()
/// .with_storage_factory(Arc::new(OpenDalStorageFactory::S3 {
/// customized_credential_load: None,
/// }))
/// .load("my_catalog", props)
/// .await?;
/// ```
fn with_storage_factory(self, storage_factory: Arc<dyn StorageFactory>) -> Self;
/// Set a [`KmsClientFactory`] to enable table encryption.
///
/// When provided, the catalog calls the factory once during
/// [`load`](Self::load) with the catalog properties to create a shared
/// [`KeyManagementClient`](crate::encryption::KeyManagementClient).
/// That client is then passed to each table's `TableBuilder` so tables
/// with `encryption.key-id` set can construct an `EncryptionManager`.
///
/// # Example
///
/// ```rust,ignore
/// use iceberg::CatalogBuilder;
/// use iceberg::encryption::kms::KmsClientFactory;
/// use std::sync::Arc;
///
/// let catalog = MyCatalogBuilder::default()
/// .with_kms_client_factory(Arc::new(MyKmsClientFactory))
/// .load("my_catalog", props)
/// .await?;
/// ```
fn with_kms_client_factory(self, kms_client_factory: Arc<dyn KmsClientFactory>) -> Self;
/// Set a custom tokio Runtime to use for spawning async tasks.
///
/// When a Runtime is provided, the catalog will propagate it to all tables
/// it creates. Tasks such as scan planning and delete file processing
/// will be spawned on this runtime.
fn with_runtime(self, runtime: Runtime) -> Self;
/// Create a new catalog instance.
fn load(
self,
name: impl Into<String>,
props: HashMap<String, String>,
) -> impl Future<Output = Result<Self::C>> + Send;
}
/// NamespaceIdent represents the identifier of a namespace in the catalog.
///
/// The namespace identifier is a list of strings, where each string is a
/// component of the namespace. It's the catalog implementer's responsibility to
/// handle the namespace identifier correctly.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct NamespaceIdent(Vec<String>);
impl NamespaceIdent {
/// Create a new namespace identifier with only one level.
pub fn new(name: String) -> Self {
Self(vec![name])
}
/// Create a multi-level namespace identifier from vector.
pub fn from_vec(names: Vec<String>) -> Result<Self> {
if names.is_empty() {
return Err(invalid_data!("Namespace identifier can't be empty!"));
}
Ok(Self(names))
}
/// Try to create namespace identifier from an iterator of string.
pub fn from_strs(iter: impl IntoIterator<Item = impl ToString>) -> Result<Self> {
Self::from_vec(iter.into_iter().map(|s| s.to_string()).collect())
}
/// Returns a string for used in url.
pub fn to_url_string(&self) -> String {
self.as_ref().join("\u{001f}")
}
/// Returns inner strings.
pub fn inner(self) -> Vec<String> {
self.0
}
/// Get the parent of this namespace.
/// Returns None if this namespace only has a single element and thus has no parent.
pub fn parent(&self) -> Option<Self> {
self.0.split_last().and_then(|(_, parent)| {
if parent.is_empty() {
None
} else {
Some(Self(parent.to_vec()))
}
})
}
}
impl AsRef<Vec<String>> for NamespaceIdent {
fn as_ref(&self) -> &Vec<String> {
&self.0
}
}
impl Deref for NamespaceIdent {
type Target = [String];
fn deref(&self) -> &Self::Target {
&self.0
}
}
/// Namespace represents a namespace in the catalog.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Namespace {
name: NamespaceIdent,
properties: HashMap<String, String>,
}
impl Namespace {
/// Create a new namespace.
pub fn new(name: NamespaceIdent) -> Self {
Self::with_properties(name, HashMap::default())
}
/// Create a new namespace with properties.
pub fn with_properties(name: NamespaceIdent, properties: HashMap<String, String>) -> Self {
Self { name, properties }
}
/// Get the name of the namespace.
pub fn name(&self) -> &NamespaceIdent {
&self.name
}
/// Get the properties of the namespace.
pub fn properties(&self) -> &HashMap<String, String> {
&self.properties
}
}
impl Display for NamespaceIdent {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.0.join("."))
}
}
/// TableIdent represents the identifier of a table in the catalog.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct TableIdent {
/// Namespace of the table.
pub namespace: NamespaceIdent,
/// Table name.
pub name: String,
}
impl TableIdent {
/// Create a new table identifier.
pub fn new(namespace: NamespaceIdent, name: String) -> Self {
Self { namespace, name }
}
/// Get the namespace of the table.
pub fn namespace(&self) -> &NamespaceIdent {
&self.namespace
}
/// Get the name of the table.
pub fn name(&self) -> &str {
&self.name
}
/// Try to create table identifier from an iterator of string.
pub fn from_strs(iter: impl IntoIterator<Item = impl ToString>) -> Result<Self> {
let mut vec: Vec<String> = iter.into_iter().map(|s| s.to_string()).collect();
let table_name = vec
.pop()
.ok_or_else(|| invalid_data!("Table identifier can't be empty!"))?;
let namespace_ident = NamespaceIdent::from_vec(vec)?;
Ok(Self {
namespace: namespace_ident,
name: table_name,
})
}
}
impl Display for TableIdent {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}.{}", self.namespace, self.name)
}
}
/// TableCreation represents the creation of a table in the catalog.
#[derive(Debug, TypedBuilder)]
pub struct TableCreation {
/// The name of the table.
pub name: String,
/// The location of the table.
#[builder(default, setter(strip_option(fallback = location_opt)))]
pub location: Option<String>,
/// The schema of the table.
pub schema: Schema,
/// The partition spec of the table, could be None.
#[builder(default, setter(strip_option(fallback = partition_spec_opt), into))]
pub partition_spec: Option<UnboundPartitionSpec>,
/// The sort order of the table.
#[builder(default, setter(strip_option(fallback = sort_order_opt)))]
pub sort_order: Option<SortOrder>,
/// The properties of the table.
#[builder(default, setter(transform = |props: impl IntoIterator<Item=(String, String)>| {
props.into_iter().collect()
}))]
pub properties: HashMap<String, String>,
/// Format version of the table. Defaults to V2.
#[builder(default = FormatVersion::V2)]
pub format_version: FormatVersion,
}
/// TableCommit represents the commit of a table in the catalog.
///
/// The builder is marked as private since it's dangerous and error-prone to construct
/// [`TableCommit`] directly.
/// Users are supposed to use [`crate::transaction::Transaction`] to update table.
#[derive(Debug, TypedBuilder)]
#[builder(build_method(vis = "pub(crate)"))]
pub struct TableCommit {
/// The table ident.
ident: TableIdent,
/// The requirements of the table.
///
/// Commit will fail if the requirements are not met.
requirements: Vec<TableRequirement>,
/// The updates of the table.
updates: Vec<TableUpdate>,
}
impl TableCommit {
/// Return the table identifier.
pub fn identifier(&self) -> &TableIdent {
&self.ident
}
/// Take all requirements.
pub fn take_requirements(&mut self) -> Vec<TableRequirement> {
take(&mut self.requirements)
}
/// Take all updates.
pub fn take_updates(&mut self) -> Vec<TableUpdate> {
take(&mut self.updates)
}
/// Applies this [`TableCommit`] to the given [`Table`] as part of a catalog update.
/// Typically used by [`Catalog::update_table`] to validate requirements and apply metadata updates.
///
/// Returns a new [`Table`] with updated metadata,
/// or an error if validation or application fails.
pub fn apply(self, table: Table) -> Result<Table> {
// check requirements
for requirement in self.requirements {
requirement.check(Some(table.metadata()))?;
}
// get current metadata location
let current_metadata_location = table.metadata_location_result()?;
// apply updates to metadata builder
let mut metadata_builder = table
.metadata()
.clone()
.into_builder(Some(current_metadata_location.to_string()));
for update in self.updates {
metadata_builder = update.apply(metadata_builder)?;
}
// Build the new metadata
let new_metadata = metadata_builder.build()?.metadata;
let new_metadata_location = MetadataLocation::from_str(current_metadata_location)?
.with_next_version()
.try_with_new_metadata(&new_metadata)?
.to_string();
Ok(table
.with_metadata(Arc::new(new_metadata))
.with_metadata_location(new_metadata_location))
}
}
/// TableRequirement represents a requirement for a table in the catalog.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
#[serde(tag = "type")]
pub enum TableRequirement {
/// The table must not already exist; used for create transactions
#[serde(rename = "assert-create")]
NotExist,
/// The table UUID must match the requirement.
#[serde(rename = "assert-table-uuid")]
UuidMatch {
/// Uuid of original table.
uuid: Uuid,
},
/// The table branch or tag identified by the requirement's `reference` must
/// reference the requirement's `snapshot-id`.
#[serde(rename = "assert-ref-snapshot-id")]
RefSnapshotIdMatch {
/// The reference of the table to assert.
r#ref: String,
/// The snapshot id of the table to assert.
/// If the id is `None`, the ref must not already exist.
#[serde(rename = "snapshot-id")]
snapshot_id: Option<i64>,
},
/// The table's last assigned column id must match the requirement.
#[serde(rename = "assert-last-assigned-field-id")]
LastAssignedFieldIdMatch {
/// The last assigned field id of the table to assert.
#[serde(rename = "last-assigned-field-id")]
last_assigned_field_id: i32,
},
/// The table's current schema id must match the requirement.
#[serde(rename = "assert-current-schema-id")]
CurrentSchemaIdMatch {
/// Current schema id of the table to assert.
#[serde(rename = "current-schema-id")]
current_schema_id: SchemaId,
},
/// The table's last assigned partition id must match the
/// requirement.
#[serde(rename = "assert-last-assigned-partition-id")]
LastAssignedPartitionIdMatch {
/// Last assigned partition id of the table to assert.
#[serde(rename = "last-assigned-partition-id")]
last_assigned_partition_id: i32,
},
/// The table's default spec id must match the requirement.
#[serde(rename = "assert-default-spec-id")]
DefaultSpecIdMatch {
/// Default spec id of the table to assert.
#[serde(rename = "default-spec-id")]
default_spec_id: i32,
},
/// The table's default sort order id must match the requirement.
#[serde(rename = "assert-default-sort-order-id")]
DefaultSortOrderIdMatch {
/// Default sort order id of the table to assert.
#[serde(rename = "default-sort-order-id")]
default_sort_order_id: i64,
},
}
/// TableUpdate represents an update to a table in the catalog.
#[derive(Debug, Serialize, Deserialize, PartialEq, Clone)]
#[serde(tag = "action", rename_all = "kebab-case")]
#[allow(clippy::large_enum_variant)]
pub enum TableUpdate {
/// Upgrade table's format version
#[serde(rename_all = "kebab-case")]
UpgradeFormatVersion {
/// Target format upgrade to.
format_version: FormatVersion,
},
/// Assign a new UUID to the table
#[serde(rename_all = "kebab-case")]
AssignUuid {
/// The new UUID to assign.
uuid: Uuid,
},
/// Add a new schema to the table
#[serde(rename_all = "kebab-case")]
AddSchema {
/// The schema to add.
schema: Schema,
},
/// Set table's current schema
#[serde(rename_all = "kebab-case")]
SetCurrentSchema {
/// Schema ID to set as current, or -1 to set last added schema
schema_id: i32,
},
/// Add a new partition spec to the table
AddSpec {
/// The partition spec to add.
spec: UnboundPartitionSpec,
},
/// Set table's default spec
#[serde(rename_all = "kebab-case")]
SetDefaultSpec {
/// Partition spec ID to set as the default, or -1 to set last added spec
spec_id: i32,
},
/// Add sort order to table.
#[serde(rename_all = "kebab-case")]
AddSortOrder {
/// Sort order to add.
sort_order: SortOrder,
},
/// Set table's default sort order
#[serde(rename_all = "kebab-case")]
SetDefaultSortOrder {
/// Sort order ID to set as the default, or -1 to set last added sort order
sort_order_id: i64,
},
/// Add snapshot to table.
#[serde(rename_all = "kebab-case")]
AddSnapshot {
/// Snapshot to add.
#[serde(
deserialize_with = "deserialize_snapshot",
serialize_with = "serialize_snapshot"
)]
snapshot: Snapshot,
},
/// Set table's snapshot ref.
#[serde(rename_all = "kebab-case")]
SetSnapshotRef {
/// Name of snapshot reference to set.
ref_name: String,
/// Snapshot reference to set.
#[serde(flatten)]
reference: SnapshotReference,
},
/// Remove table's snapshots
#[serde(rename_all = "kebab-case")]
RemoveSnapshots {
/// Snapshot ids to remove.
snapshot_ids: Vec<i64>,
},
/// Remove snapshot reference
#[serde(rename_all = "kebab-case")]
RemoveSnapshotRef {
/// Name of snapshot reference to remove.
ref_name: String,
},
/// Update table's location
SetLocation {
/// New location for table.
location: String,
},
/// Update table's properties
SetProperties {
/// Properties to update for table.
updates: HashMap<String, String>,
},
/// Remove table's properties
RemoveProperties {
/// Properties to remove
removals: Vec<String>,
},
/// Remove partition specs
#[serde(rename_all = "kebab-case")]
RemovePartitionSpecs {
/// Partition spec ids to remove.
spec_ids: Vec<i32>,
},
/// Set statistics for a snapshot
#[serde(with = "_serde_set_statistics")]
SetStatistics {
/// File containing the statistics
statistics: StatisticsFile,
},
/// Remove statistics for a snapshot
#[serde(rename_all = "kebab-case")]
RemoveStatistics {
/// Snapshot id to remove statistics for.
snapshot_id: i64,
},
/// Set partition statistics for a snapshot
#[serde(rename_all = "kebab-case")]
SetPartitionStatistics {
/// File containing the partition statistics
partition_statistics: PartitionStatisticsFile,
},
/// Remove partition statistics for a snapshot
#[serde(rename_all = "kebab-case")]
RemovePartitionStatistics {
/// Snapshot id to remove partition statistics for.
snapshot_id: i64,
},
/// Remove schemas
#[serde(rename_all = "kebab-case")]
RemoveSchemas {
/// Schema IDs to remove.
schema_ids: Vec<i32>,
},
/// Add an encryption key
#[serde(rename_all = "kebab-case")]
AddEncryptionKey {
/// The encryption key to add.
encryption_key: EncryptedKey,
},
/// Remove an encryption key
#[serde(rename_all = "kebab-case")]
RemoveEncryptionKey {
/// The id of the encryption key to remove.
key_id: String,
},
}
impl TableUpdate {
/// Applies the update to the table metadata builder.
pub fn apply(self, builder: TableMetadataBuilder) -> Result<TableMetadataBuilder> {
match self {
TableUpdate::AssignUuid { uuid } => Ok(builder.assign_uuid(uuid)),
TableUpdate::AddSchema { schema, .. } => Ok(builder.add_schema(schema)?),
TableUpdate::SetCurrentSchema { schema_id } => builder.set_current_schema(schema_id),
TableUpdate::AddSpec { spec } => builder.add_partition_spec(spec),
TableUpdate::SetDefaultSpec { spec_id } => builder.set_default_partition_spec(spec_id),
TableUpdate::AddSortOrder { sort_order } => builder.add_sort_order(sort_order),
TableUpdate::SetDefaultSortOrder { sort_order_id } => {
builder.set_default_sort_order(sort_order_id)
}
TableUpdate::AddSnapshot { snapshot } => builder.add_snapshot(snapshot),
TableUpdate::SetSnapshotRef {
ref_name,
reference,
} => builder.set_ref(&ref_name, reference),
TableUpdate::RemoveSnapshots { snapshot_ids } => {
Ok(builder.remove_snapshots(&snapshot_ids))
}
TableUpdate::RemoveSnapshotRef { ref_name } => Ok(builder.remove_ref(&ref_name)),
TableUpdate::SetLocation { location } => Ok(builder.set_location(location)),
TableUpdate::SetProperties { updates } => builder.set_properties(updates),
TableUpdate::RemoveProperties { removals } => builder.remove_properties(&removals),
TableUpdate::UpgradeFormatVersion { format_version } => {
builder.upgrade_format_version(format_version)
}
TableUpdate::RemovePartitionSpecs { spec_ids } => {
builder.remove_partition_specs(&spec_ids)
}
TableUpdate::SetStatistics { statistics } => Ok(builder.set_statistics(statistics)),
TableUpdate::RemoveStatistics { snapshot_id } => {
Ok(builder.remove_statistics(snapshot_id))
}
TableUpdate::SetPartitionStatistics {
partition_statistics,
} => Ok(builder.set_partition_statistics(partition_statistics)),
TableUpdate::RemovePartitionStatistics { snapshot_id } => {
Ok(builder.remove_partition_statistics(snapshot_id))
}
TableUpdate::RemoveSchemas { schema_ids } => builder.remove_schemas(&schema_ids),
TableUpdate::AddEncryptionKey { encryption_key } => {
Ok(builder.add_encryption_key(encryption_key))
}
TableUpdate::RemoveEncryptionKey { key_id } => {
Ok(builder.remove_encryption_key(&key_id))
}
}
}
}
impl TableRequirement {
/// Check that the requirement is met by the table metadata.
/// If the requirement is not met, an appropriate error is returned.
///
/// Provide metadata as `None` if the table does not exist.
pub fn check(&self, metadata: Option<&TableMetadata>) -> Result<()> {
if let Some(metadata) = metadata {
match self {
TableRequirement::NotExist => {
return Err(Error::new(
ErrorKind::CatalogCommitConflicts,
format!(
"Requirement failed: Table with id {} already exists",
metadata.uuid()
),
)
.with_retryable(true));
}
TableRequirement::UuidMatch { uuid } => {
if &metadata.uuid() != uuid {
return Err(Error::new(
ErrorKind::CatalogCommitConflicts,
"Requirement failed: Table UUID does not match",
)
.with_context("expected", *uuid)
.with_context("found", metadata.uuid())
.with_retryable(true));
}
}
TableRequirement::CurrentSchemaIdMatch { current_schema_id } => {
// ToDo: Harmonize the types of current_schema_id
if metadata.current_schema_id != *current_schema_id {
return Err(Error::new(
ErrorKind::CatalogCommitConflicts,
"Requirement failed: Current schema id does not match",
)
.with_context("expected", current_schema_id.to_string())
.with_context("found", metadata.current_schema_id.to_string())
.with_retryable(true));
}
}
TableRequirement::DefaultSortOrderIdMatch {
default_sort_order_id,
} => {
if metadata.default_sort_order().order_id != *default_sort_order_id {
return Err(Error::new(
ErrorKind::CatalogCommitConflicts,
"Requirement failed: Default sort order id does not match",
)
.with_context("expected", default_sort_order_id.to_string())
.with_context("found", metadata.default_sort_order().order_id.to_string())
.with_retryable(true));
}
}
TableRequirement::RefSnapshotIdMatch { r#ref, snapshot_id } => {
let snapshot_ref = metadata.snapshot_for_ref(r#ref);
if let Some(snapshot_id) = snapshot_id {
let snapshot_ref = snapshot_ref.ok_or(
Error::new(
ErrorKind::CatalogCommitConflicts,
format!("Requirement failed: Branch or tag `{ref}` not found"),
)
.with_retryable(true),
)?;
if snapshot_ref.snapshot_id() != *snapshot_id {
return Err(Error::new(
ErrorKind::CatalogCommitConflicts,
format!(
"Requirement failed: Branch or tag `{ref}`'s snapshot has changed"
),
)
.with_context("expected", snapshot_id.to_string())
.with_context("found", snapshot_ref.snapshot_id().to_string())
.with_retryable(true));
}
} else if snapshot_ref.is_some() {
// a null snapshot ID means the ref should not exist already
return Err(Error::new(
ErrorKind::CatalogCommitConflicts,
format!("Requirement failed: Branch or tag `{ref}` already exists"),
)
.with_retryable(true));
}
}
TableRequirement::DefaultSpecIdMatch { default_spec_id } => {
// ToDo: Harmonize the types of default_spec_id
if metadata.default_partition_spec_id() != *default_spec_id {
return Err(Error::new(
ErrorKind::CatalogCommitConflicts,
"Requirement failed: Default partition spec id does not match",
)
.with_context("expected", default_spec_id.to_string())
.with_context("found", metadata.default_partition_spec_id().to_string())
.with_retryable(true));
}
}
TableRequirement::LastAssignedPartitionIdMatch {
last_assigned_partition_id,
} => {
if metadata.last_partition_id != *last_assigned_partition_id {
return Err(Error::new(
ErrorKind::CatalogCommitConflicts,
"Requirement failed: Last assigned partition id does not match",
)
.with_context("expected", last_assigned_partition_id.to_string())
.with_context("found", metadata.last_partition_id.to_string())
.with_retryable(true));
}
}
TableRequirement::LastAssignedFieldIdMatch {
last_assigned_field_id,
} => {
if &metadata.last_column_id != last_assigned_field_id {
return Err(Error::new(
ErrorKind::CatalogCommitConflicts,
"Requirement failed: Last assigned field id does not match",
)
.with_context("expected", last_assigned_field_id.to_string())
.with_context("found", metadata.last_column_id.to_string())
.with_retryable(true));
}
}
};
} else {
match self {
TableRequirement::NotExist => {}
_ => {
return Err(Error::new(
ErrorKind::TableNotFound,
"Requirement failed: Table does not exist",
));
}
}
}
Ok(())
}
}
pub(super) mod _serde {
use serde::{Deserialize as _, Deserializer, Serialize as _};
use super::*;
use crate::spec::{SchemaId, Summary};
pub(super) fn deserialize_snapshot<'de, D>(
deserializer: D,
) -> std::result::Result<Snapshot, D::Error>
where D: Deserializer<'de> {
let buf = CatalogSnapshot::deserialize(deserializer)?;
Ok(buf.into())
}
pub(super) fn serialize_snapshot<S>(
snapshot: &Snapshot,
serializer: S,
) -> std::result::Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
let buf: CatalogSnapshot = snapshot.clone().into();
buf.serialize(serializer)
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "kebab-case")]
/// Defines the structure of a v2 snapshot for the catalog.
/// Main difference to SnapshotV2 is that sequence-number is optional
/// in the rest catalog spec to allow for backwards compatibility with v1.
struct CatalogSnapshot {
snapshot_id: i64,
#[serde(skip_serializing_if = "Option::is_none")]
parent_snapshot_id: Option<i64>,
#[serde(default)]
sequence_number: i64,
timestamp_ms: i64,
manifest_list: String,
summary: Summary,
#[serde(skip_serializing_if = "Option::is_none")]
schema_id: Option<SchemaId>,
#[serde(skip_serializing_if = "Option::is_none")]
first_row_id: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
added_rows: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
key_id: Option<String>,
}
impl From<CatalogSnapshot> for Snapshot {
fn from(snapshot: CatalogSnapshot) -> Self {
let CatalogSnapshot {
snapshot_id,
parent_snapshot_id,
sequence_number,
timestamp_ms,
manifest_list,
schema_id,
summary,
first_row_id,
added_rows,
key_id,
} = snapshot;
let builder = Snapshot::builder()
.with_snapshot_id(snapshot_id)
.with_parent_snapshot_id(parent_snapshot_id)
.with_sequence_number(sequence_number)
.with_timestamp_ms(timestamp_ms)
.with_manifest_list(manifest_list)
.with_summary(summary)
.with_encryption_key_id(key_id);
let row_range = first_row_id.zip(added_rows);
match (schema_id, row_range) {
(None, None) => builder.build(),
(Some(schema_id), None) => builder.with_schema_id(schema_id).build(),
(None, Some((first_row_id, last_row_id))) => {
builder.with_row_range(first_row_id, last_row_id).build()
}
(Some(schema_id), Some((first_row_id, last_row_id))) => builder
.with_schema_id(schema_id)
.with_row_range(first_row_id, last_row_id)
.build(),
}
}
}
impl From<Snapshot> for CatalogSnapshot {
fn from(snapshot: Snapshot) -> Self {
let first_row_id = snapshot.first_row_id();
let added_rows = snapshot.added_rows_count();
let Snapshot {
snapshot_id,
parent_snapshot_id,
sequence_number,
timestamp_ms,
manifest_list,
summary,
schema_id,
row_range: _,
encryption_key_id: key_id,
} = snapshot;
CatalogSnapshot {
snapshot_id,
parent_snapshot_id,
sequence_number,
timestamp_ms,
manifest_list,
summary,
schema_id,
first_row_id,
added_rows,
key_id,
}
}
}
}
/// ViewCreation represents the creation of a view in the catalog.
#[derive(Debug, TypedBuilder)]
pub struct ViewCreation {
/// The name of the view.
pub name: String,
/// The view's base location; used to create metadata file locations
pub location: String,
/// Representations for the view.
pub representations: ViewRepresentations,
/// The schema of the view.
pub schema: Schema,
/// The properties of the view.
#[builder(default)]
pub properties: HashMap<String, String>,
/// The default namespace to use when a reference in the SELECT is a single identifier
pub default_namespace: NamespaceIdent,
/// Default catalog to use when a reference in the SELECT does not contain a catalog
#[builder(default)]
pub default_catalog: Option<String>,
/// A string to string map of summary metadata about the version
/// Typical keys are "engine-name" and "engine-version"
#[builder(default)]
pub summary: HashMap<String, String>,
}
/// ViewUpdate represents an update to a view in the catalog.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "action", rename_all = "kebab-case")]
#[allow(clippy::large_enum_variant)]
pub enum ViewUpdate {
/// Assign a new UUID to the view
#[serde(rename_all = "kebab-case")]
AssignUuid {
/// The new UUID to assign.
uuid: Uuid,
},
/// Upgrade view's format version
#[serde(rename_all = "kebab-case")]
UpgradeFormatVersion {
/// Target format upgrade to.
format_version: ViewFormatVersion,
},
/// Add a new schema to the view
#[serde(rename_all = "kebab-case")]
AddSchema {
/// The schema to add.
schema: Schema,
/// The last column id of the view.
last_column_id: Option<i32>,
},
/// Set view's current schema
#[serde(rename_all = "kebab-case")]
SetLocation {
/// New location for view.
location: String,
},
/// Set view's properties
///
/// Matching keys are updated, and non-matching keys are left unchanged.
#[serde(rename_all = "kebab-case")]
SetProperties {
/// Properties to update for view.
updates: HashMap<String, String>,
},
/// Remove view's properties
#[serde(rename_all = "kebab-case")]
RemoveProperties {
/// Properties to remove
removals: Vec<String>,
},
/// Add a new version to the view
#[serde(rename_all = "kebab-case")]
AddViewVersion {
/// The view version to add.
view_version: ViewVersion,
},
/// Set view's current version
#[serde(rename_all = "kebab-case")]
SetCurrentViewVersion {
/// View version id to set as current, or -1 to set last added version
view_version_id: i32,
},
}
mod _serde_set_statistics {
// The rest spec requires an additional field `snapshot-id`
// that is redundant with the `snapshot_id` field in the statistics file.
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use super::*;
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
struct SetStatistics {
snapshot_id: Option<i64>,
statistics: StatisticsFile,
}
pub fn serialize<S>(
value: &StatisticsFile,
serializer: S,
) -> std::result::Result<S::Ok, S::Error>
where
S: Serializer,
{
SetStatistics {
snapshot_id: Some(value.snapshot_id),
statistics: value.clone(),
}
.serialize(serializer)
}
pub fn deserialize<'de, D>(deserializer: D) -> std::result::Result<StatisticsFile, D::Error>
where D: Deserializer<'de> {
let SetStatistics {
snapshot_id,
statistics,
} = SetStatistics::deserialize(deserializer)?;
if let Some(snapshot_id) = snapshot_id
&& snapshot_id != statistics.snapshot_id
{
return Err(serde::de::Error::custom(format!(
"Snapshot id to set {snapshot_id} does not match the statistics file snapshot id {}",
statistics.snapshot_id
)));
}
Ok(statistics)
}
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::fmt::Debug;
use std::fs::File;
use std::io::BufReader;
use base64::Engine as _;
use serde::Serialize;
use serde::de::DeserializeOwned;
use uuid::uuid;
use super::ViewUpdate;
use crate::io::FileIO;
use crate::spec::{
BlobMetadata, EncryptedKey, FormatVersion, MAIN_BRANCH, NestedField, NullOrder, Operation,
PartitionStatisticsFile, PrimitiveType, Schema, Snapshot, SnapshotReference,
SnapshotRetention, SortDirection, SortField, SortOrder, SqlViewRepresentation,
StatisticsFile, Summary, TableMetadata, TableMetadataBuilder, Transform, Type,
UnboundPartitionField, UnboundPartitionSpec, ViewFormatVersion, ViewRepresentation,
ViewRepresentations, ViewVersion,
};
use crate::table::Table;
use crate::test_utils::test_runtime;
use crate::{
NamespaceIdent, TableCommit, TableCreation, TableIdent, TableRequirement, TableUpdate,
};
#[test]
fn test_parent_namespace() {
let ns1 = NamespaceIdent::from_strs(vec!["ns1"]).unwrap();
let ns2 = NamespaceIdent::from_strs(vec!["ns1", "ns2"]).unwrap();
let ns3 = NamespaceIdent::from_strs(vec!["ns1", "ns2", "ns3"]).unwrap();
assert_eq!(ns1.parent(), None);
assert_eq!(ns2.parent(), Some(ns1.clone()));
assert_eq!(ns3.parent(), Some(ns2.clone()));
}
#[test]
fn test_create_table_id() {
let table_id = TableIdent {
namespace: NamespaceIdent::from_strs(vec!["ns1"]).unwrap(),
name: "t1".to_string(),
};
assert_eq!(table_id, TableIdent::from_strs(vec!["ns1", "t1"]).unwrap());
}
#[test]
fn test_table_creation_iterator_properties() {
let builder = TableCreation::builder()
.name("table".to_string())
.schema(Schema::builder().build().unwrap());
fn s(k: &str, v: &str) -> (String, String) {
(k.to_string(), v.to_string())
}
let table_creation = builder
.properties([s("key", "value"), s("foo", "bar")])
.build();
assert_eq!(
HashMap::from([s("key", "value"), s("foo", "bar")]),
table_creation.properties
);
}
fn test_serde_json<T: Serialize + DeserializeOwned + PartialEq + Debug>(
json: impl ToString,
expected: T,
) {
let json_str = json.to_string();
let actual: T = serde_json::from_str(&json_str).expect("Failed to parse from json");
assert_eq!(actual, expected, "Parsed value is not equal to expected");
let restored: T = serde_json::from_str(
&serde_json::to_string(&actual).expect("Failed to serialize to json"),
)
.expect("Failed to parse from serialized json");
assert_eq!(
restored, expected,
"Parsed restored value is not equal to expected"
);
}
fn metadata() -> TableMetadata {
let tbl_creation = TableCreation::builder()
.name("table".to_string())
.location("/path/to/table".to_string())
.schema(Schema::builder().build().unwrap())
.build();
TableMetadataBuilder::from_table_creation(tbl_creation)
.unwrap()
.assign_uuid(uuid::Uuid::nil())
.build()
.unwrap()
.metadata
}
#[test]
fn test_check_requirement_not_exist() {
let metadata = metadata();
let requirement = TableRequirement::NotExist;
assert!(requirement.check(Some(&metadata)).is_err());
assert!(requirement.check(None).is_ok());
}
#[test]
fn test_check_table_uuid() {
let metadata = metadata();
let requirement = TableRequirement::UuidMatch {
uuid: uuid::Uuid::now_v7(),
};
assert!(requirement.check(Some(&metadata)).is_err());
let requirement = TableRequirement::UuidMatch {
uuid: uuid::Uuid::nil(),
};
assert!(requirement.check(Some(&metadata)).is_ok());
}
#[test]
fn test_check_ref_snapshot_id() {
let metadata = metadata();
// Ref does not exist but should
let requirement = TableRequirement::RefSnapshotIdMatch {
r#ref: "my_branch".to_string(),
snapshot_id: Some(1),
};
assert!(requirement.check(Some(&metadata)).is_err());
// Ref does not exist and should not
let requirement = TableRequirement::RefSnapshotIdMatch {
r#ref: "my_branch".to_string(),
snapshot_id: None,
};
assert!(requirement.check(Some(&metadata)).is_ok());
// Add snapshot
let snapshot = Snapshot::builder()
.with_snapshot_id(3051729675574597004)
.with_sequence_number(10)
.with_timestamp_ms(9992191116217)
.with_manifest_list("s3://b/wh/.../s1.avro".to_string())
.with_schema_id(0)
.with_summary(Summary {
operation: Operation::Append,
additional_properties: HashMap::new(),
})
.build();
let builder = metadata.into_builder(None);
let builder = TableUpdate::AddSnapshot {
snapshot: snapshot.clone(),
}
.apply(builder)
.unwrap();
let metadata = TableUpdate::SetSnapshotRef {
ref_name: MAIN_BRANCH.to_string(),
reference: SnapshotReference {
snapshot_id: snapshot.snapshot_id(),
retention: SnapshotRetention::Branch {
min_snapshots_to_keep: Some(10),
max_snapshot_age_ms: None,
max_ref_age_ms: None,
},
},
}
.apply(builder)
.unwrap()
.build()
.unwrap()
.metadata;
// Ref exists and should match
let requirement = TableRequirement::RefSnapshotIdMatch {
r#ref: "main".to_string(),
snapshot_id: Some(3051729675574597004),
};
assert!(requirement.check(Some(&metadata)).is_ok());
// Ref exists but does not match
let requirement = TableRequirement::RefSnapshotIdMatch {
r#ref: "main".to_string(),
snapshot_id: Some(1),
};
assert!(requirement.check(Some(&metadata)).is_err());
}
#[test]
fn test_check_last_assigned_field_id() {
let metadata = metadata();
let requirement = TableRequirement::LastAssignedFieldIdMatch {
last_assigned_field_id: 1,
};
assert!(requirement.check(Some(&metadata)).is_err());
let requirement = TableRequirement::LastAssignedFieldIdMatch {
last_assigned_field_id: 0,
};
assert!(requirement.check(Some(&metadata)).is_ok());
}
#[test]
fn test_check_current_schema_id() {
let metadata = metadata();
let requirement = TableRequirement::CurrentSchemaIdMatch {
current_schema_id: 1,
};
assert!(requirement.check(Some(&metadata)).is_err());
let requirement = TableRequirement::CurrentSchemaIdMatch {
current_schema_id: 0,
};
assert!(requirement.check(Some(&metadata)).is_ok());
}
#[test]
fn test_check_last_assigned_partition_id() {
let metadata = metadata();
let requirement = TableRequirement::LastAssignedPartitionIdMatch {
last_assigned_partition_id: 0,
};
assert!(requirement.check(Some(&metadata)).is_err());
let requirement = TableRequirement::LastAssignedPartitionIdMatch {
last_assigned_partition_id: 999,
};
assert!(requirement.check(Some(&metadata)).is_ok());
}
#[test]
fn test_check_default_spec_id() {
let metadata = metadata();
let requirement = TableRequirement::DefaultSpecIdMatch { default_spec_id: 1 };
assert!(requirement.check(Some(&metadata)).is_err());
let requirement = TableRequirement::DefaultSpecIdMatch { default_spec_id: 0 };
assert!(requirement.check(Some(&metadata)).is_ok());
}
#[test]
fn test_check_default_sort_order_id() {
let metadata = metadata();
let requirement = TableRequirement::DefaultSortOrderIdMatch {
default_sort_order_id: 1,
};
assert!(requirement.check(Some(&metadata)).is_err());
let requirement = TableRequirement::DefaultSortOrderIdMatch {
default_sort_order_id: 0,
};
assert!(requirement.check(Some(&metadata)).is_ok());
}
#[test]
fn test_table_uuid() {
test_serde_json(
r#"
{
"type": "assert-table-uuid",
"uuid": "2cc52516-5e73-41f2-b139-545d41a4e151"
}
"#,
TableRequirement::UuidMatch {
uuid: uuid!("2cc52516-5e73-41f2-b139-545d41a4e151"),
},
);
}
#[test]
fn test_assert_table_not_exists() {
test_serde_json(
r#"
{
"type": "assert-create"
}
"#,
TableRequirement::NotExist,
);
}
#[test]
fn test_assert_ref_snapshot_id() {
test_serde_json(
r#"
{
"type": "assert-ref-snapshot-id",
"ref": "snapshot-name",
"snapshot-id": null
}
"#,
TableRequirement::RefSnapshotIdMatch {
r#ref: "snapshot-name".to_string(),
snapshot_id: None,
},
);
test_serde_json(
r#"
{
"type": "assert-ref-snapshot-id",
"ref": "snapshot-name",
"snapshot-id": 1
}
"#,
TableRequirement::RefSnapshotIdMatch {
r#ref: "snapshot-name".to_string(),
snapshot_id: Some(1),
},
);
}
#[test]
fn test_assert_last_assigned_field_id() {
test_serde_json(
r#"
{
"type": "assert-last-assigned-field-id",
"last-assigned-field-id": 12
}
"#,
TableRequirement::LastAssignedFieldIdMatch {
last_assigned_field_id: 12,
},
);
}
#[test]
fn test_assert_current_schema_id() {
test_serde_json(
r#"
{
"type": "assert-current-schema-id",
"current-schema-id": 4
}
"#,
TableRequirement::CurrentSchemaIdMatch {
current_schema_id: 4,
},
);
}
#[test]
fn test_assert_last_assigned_partition_id() {
test_serde_json(
r#"
{
"type": "assert-last-assigned-partition-id",
"last-assigned-partition-id": 1004
}
"#,
TableRequirement::LastAssignedPartitionIdMatch {
last_assigned_partition_id: 1004,
},
);
}
#[test]
fn test_assert_default_spec_id() {
test_serde_json(
r#"
{
"type": "assert-default-spec-id",
"default-spec-id": 5
}
"#,
TableRequirement::DefaultSpecIdMatch { default_spec_id: 5 },
);
}
#[test]
fn test_assert_default_sort_order() {
let json = r#"
{
"type": "assert-default-sort-order-id",
"default-sort-order-id": 10
}
"#;
let update = TableRequirement::DefaultSortOrderIdMatch {
default_sort_order_id: 10,
};
test_serde_json(json, update);
}
#[test]
fn test_parse_assert_invalid() {
assert!(
serde_json::from_str::<TableRequirement>(
r#"
{
"default-sort-order-id": 10
}
"#
)
.is_err(),
"Table requirements should not be parsed without type."
);
}
#[test]
fn test_assign_uuid() {
test_serde_json(
r#"
{
"action": "assign-uuid",
"uuid": "2cc52516-5e73-41f2-b139-545d41a4e151"
}
"#,
TableUpdate::AssignUuid {
uuid: uuid!("2cc52516-5e73-41f2-b139-545d41a4e151"),
},
);
}
#[test]
fn test_upgrade_format_version() {
test_serde_json(
r#"
{
"action": "upgrade-format-version",
"format-version": 2
}
"#,
TableUpdate::UpgradeFormatVersion {
format_version: FormatVersion::V2,
},
);
}
#[test]
fn test_add_schema() {
let test_schema = Schema::builder()
.with_schema_id(1)
.with_identifier_field_ids(vec![2])
.with_fields(vec![
NestedField::optional(1, "foo", Type::Primitive(PrimitiveType::String)).into(),
NestedField::required(2, "bar", Type::Primitive(PrimitiveType::Int)).into(),
NestedField::optional(3, "baz", Type::Primitive(PrimitiveType::Boolean)).into(),
])
.build()
.unwrap();
test_serde_json(
r#"
{
"action": "add-schema",
"schema": {
"type": "struct",
"schema-id": 1,
"fields": [
{
"id": 1,
"name": "foo",
"required": false,
"type": "string"
},
{
"id": 2,
"name": "bar",
"required": true,
"type": "int"
},
{
"id": 3,
"name": "baz",
"required": false,
"type": "boolean"
}
],
"identifier-field-ids": [
2
]
},
"last-column-id": 3
}
"#,
TableUpdate::AddSchema {
schema: test_schema.clone(),
},
);
test_serde_json(
r#"
{
"action": "add-schema",
"schema": {
"type": "struct",
"schema-id": 1,
"fields": [
{
"id": 1,
"name": "foo",
"required": false,
"type": "string"
},
{
"id": 2,
"name": "bar",
"required": true,
"type": "int"
},
{
"id": 3,
"name": "baz",
"required": false,
"type": "boolean"
}
],
"identifier-field-ids": [
2
]
}
}
"#,
TableUpdate::AddSchema {
schema: test_schema.clone(),
},
);
}
#[test]
fn test_set_current_schema() {
test_serde_json(
r#"
{
"action": "set-current-schema",
"schema-id": 23
}
"#,
TableUpdate::SetCurrentSchema { schema_id: 23 },
);
}
#[test]
fn test_add_spec() {
test_serde_json(
r#"
{
"action": "add-spec",
"spec": {
"fields": [
{
"source-id": 4,
"name": "ts_day",
"transform": "day"
},
{
"source-id": 1,
"name": "id_bucket",
"transform": "bucket[16]"
},
{
"source-id": 2,
"name": "id_truncate",
"transform": "truncate[4]"
}
]
}
}
"#,
TableUpdate::AddSpec {
spec: UnboundPartitionSpec::builder()
.add_partition_field(
UnboundPartitionField::builder()
.source_ids(vec![4])
.name("ts_day")
.transform(Transform::Day)
.build()
.unwrap(),
)
.unwrap()
.add_partition_field(
UnboundPartitionField::builder()
.source_ids(vec![1])
.name("id_bucket")
.transform(Transform::Bucket(16))
.build()
.unwrap(),
)
.unwrap()
.add_partition_field(
UnboundPartitionField::builder()
.source_ids(vec![2])
.name("id_truncate")
.transform(Transform::Truncate(4))
.build()
.unwrap(),
)
.unwrap()
.build(),
},
);
}
#[test]
fn test_set_default_spec() {
test_serde_json(
r#"
{
"action": "set-default-spec",
"spec-id": 1
}
"#,
TableUpdate::SetDefaultSpec { spec_id: 1 },
)
}
#[test]
fn test_add_sort_order() {
let json = r#"
{
"action": "add-sort-order",
"sort-order": {
"order-id": 1,
"fields": [
{
"transform": "identity",
"source-id": 2,
"direction": "asc",
"null-order": "nulls-first"
},
{
"transform": "bucket[4]",
"source-id": 3,
"direction": "desc",
"null-order": "nulls-last"
}
]
}
}
"#;
let update = TableUpdate::AddSortOrder {
sort_order: SortOrder::builder()
.with_order_id(1)
.with_sort_field(
SortField::builder()
.source_id(2)
.direction(SortDirection::Ascending)
.null_order(NullOrder::First)
.transform(Transform::Identity)
.build(),
)
.with_sort_field(
SortField::builder()
.source_id(3)
.direction(SortDirection::Descending)
.null_order(NullOrder::Last)
.transform(Transform::Bucket(4))
.build(),
)
.build_unbound()
.unwrap(),
};
test_serde_json(json, update);
}
#[test]
fn test_set_default_order() {
let json = r#"
{
"action": "set-default-sort-order",
"sort-order-id": 2
}
"#;
let update = TableUpdate::SetDefaultSortOrder { sort_order_id: 2 };
test_serde_json(json, update);
}
#[test]
fn test_add_snapshot() {
let json = r#"
{
"action": "add-snapshot",
"snapshot": {
"snapshot-id": 3055729675574597000,
"parent-snapshot-id": 3051729675574597000,
"timestamp-ms": 1555100955770,
"sequence-number": 1,
"summary": {
"operation": "append"
},
"manifest-list": "s3://a/b/2.avro",
"schema-id": 1
}
}
"#;
let update = TableUpdate::AddSnapshot {
snapshot: Snapshot::builder()
.with_snapshot_id(3055729675574597000)
.with_parent_snapshot_id(Some(3051729675574597000))
.with_timestamp_ms(1555100955770)
.with_sequence_number(1)
.with_manifest_list("s3://a/b/2.avro")
.with_schema_id(1)
.with_summary(Summary {
operation: Operation::Append,
additional_properties: HashMap::default(),
})
.build(),
};
test_serde_json(json, update);
}
#[test]
fn test_add_snapshot_v1() {
let json = r#"
{
"action": "add-snapshot",
"snapshot": {
"snapshot-id": 3055729675574597000,
"parent-snapshot-id": 3051729675574597000,
"timestamp-ms": 1555100955770,
"summary": {
"operation": "append"
},
"manifest-list": "s3://a/b/2.avro"
}
}
"#;
let update = TableUpdate::AddSnapshot {
snapshot: Snapshot::builder()
.with_snapshot_id(3055729675574597000)
.with_parent_snapshot_id(Some(3051729675574597000))
.with_timestamp_ms(1555100955770)
.with_sequence_number(0)
.with_manifest_list("s3://a/b/2.avro")
.with_summary(Summary {
operation: Operation::Append,
additional_properties: HashMap::default(),
})
.build(),
};
let actual: TableUpdate = serde_json::from_str(json).expect("Failed to parse from json");
assert_eq!(actual, update, "Parsed value is not equal to expected");
}
#[test]
fn test_add_snapshot_v3() {
let json = serde_json::json!(
{
"action": "add-snapshot",
"snapshot": {
"snapshot-id": 3055729675574597000i64,
"parent-snapshot-id": 3051729675574597000i64,
"timestamp-ms": 1555100955770i64,
"first-row-id":0,
"added-rows":2,
"key-id":"key123",
"summary": {
"operation": "append"
},
"manifest-list": "s3://a/b/2.avro"
}
});
let update = TableUpdate::AddSnapshot {
snapshot: Snapshot::builder()
.with_snapshot_id(3055729675574597000)
.with_parent_snapshot_id(Some(3051729675574597000))
.with_timestamp_ms(1555100955770)
.with_sequence_number(0)
.with_manifest_list("s3://a/b/2.avro")
.with_row_range(0, 2)
.with_encryption_key_id(Some("key123".to_string()))
.with_summary(Summary {
operation: Operation::Append,
additional_properties: HashMap::default(),
})
.build(),
};
let actual: TableUpdate = serde_json::from_value(json).expect("Failed to parse from json");
assert_eq!(actual, update, "Parsed value is not equal to expected");
let restored: TableUpdate = serde_json::from_str(
&serde_json::to_string(&actual).expect("Failed to serialize to json"),
)
.expect("Failed to parse from serialized json");
assert_eq!(restored, update);
}
#[test]
fn test_remove_snapshots() {
let json = r#"
{
"action": "remove-snapshots",
"snapshot-ids": [
1,
2
]
}
"#;
let update = TableUpdate::RemoveSnapshots {
snapshot_ids: vec![1, 2],
};
test_serde_json(json, update);
}
#[test]
fn test_remove_snapshot_ref() {
let json = r#"
{
"action": "remove-snapshot-ref",
"ref-name": "snapshot-ref"
}
"#;
let update = TableUpdate::RemoveSnapshotRef {
ref_name: "snapshot-ref".to_string(),
};
test_serde_json(json, update);
}
#[test]
fn test_set_snapshot_ref_tag() {
let json = r#"
{
"action": "set-snapshot-ref",
"type": "tag",
"ref-name": "hank",
"snapshot-id": 1,
"max-ref-age-ms": 1
}
"#;
let update = TableUpdate::SetSnapshotRef {
ref_name: "hank".to_string(),
reference: SnapshotReference {
snapshot_id: 1,
retention: SnapshotRetention::Tag {
max_ref_age_ms: Some(1),
},
},
};
test_serde_json(json, update);
}
#[test]
fn test_set_snapshot_ref_branch() {
let json = r#"
{
"action": "set-snapshot-ref",
"type": "branch",
"ref-name": "hank",
"snapshot-id": 1,
"min-snapshots-to-keep": 2,
"max-snapshot-age-ms": 3,
"max-ref-age-ms": 4
}
"#;
let update = TableUpdate::SetSnapshotRef {
ref_name: "hank".to_string(),
reference: SnapshotReference {
snapshot_id: 1,
retention: SnapshotRetention::Branch {
min_snapshots_to_keep: Some(2),
max_snapshot_age_ms: Some(3),
max_ref_age_ms: Some(4),
},
},
};
test_serde_json(json, update);
}
#[test]
fn test_set_properties() {
let json = r#"
{
"action": "set-properties",
"updates": {
"prop1": "v1",
"prop2": "v2"
}
}
"#;
let update = TableUpdate::SetProperties {
updates: vec![
("prop1".to_string(), "v1".to_string()),
("prop2".to_string(), "v2".to_string()),
]
.into_iter()
.collect(),
};
test_serde_json(json, update);
}
#[test]
fn test_remove_properties() {
let json = r#"
{
"action": "remove-properties",
"removals": [
"prop1",
"prop2"
]
}
"#;
let update = TableUpdate::RemoveProperties {
removals: vec!["prop1".to_string(), "prop2".to_string()],
};
test_serde_json(json, update);
}
#[test]
fn test_set_location() {
let json = r#"
{
"action": "set-location",
"location": "s3://bucket/warehouse/tbl_location"
}
"#;
let update = TableUpdate::SetLocation {
location: "s3://bucket/warehouse/tbl_location".to_string(),
};
test_serde_json(json, update);
}
#[test]
fn test_table_update_apply() {
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;
let table_metadata_builder = TableMetadataBuilder::new_from_metadata(
table_metadata,
Some("s3://db/table/metadata/metadata1.gz.json".to_string()),
);
let uuid = uuid::Uuid::new_v4();
let update = TableUpdate::AssignUuid { uuid };
let updated_metadata = update
.apply(table_metadata_builder)
.unwrap()
.build()
.unwrap()
.metadata;
assert_eq!(updated_metadata.uuid(), uuid);
}
#[test]
fn test_view_assign_uuid() {
test_serde_json(
r#"
{
"action": "assign-uuid",
"uuid": "2cc52516-5e73-41f2-b139-545d41a4e151"
}
"#,
ViewUpdate::AssignUuid {
uuid: uuid!("2cc52516-5e73-41f2-b139-545d41a4e151"),
},
);
}
#[test]
fn test_view_upgrade_format_version() {
test_serde_json(
r#"
{
"action": "upgrade-format-version",
"format-version": 1
}
"#,
ViewUpdate::UpgradeFormatVersion {
format_version: ViewFormatVersion::V1,
},
);
}
#[test]
fn test_view_add_schema() {
let test_schema = Schema::builder()
.with_schema_id(1)
.with_identifier_field_ids(vec![2])
.with_fields(vec![
NestedField::optional(1, "foo", Type::Primitive(PrimitiveType::String)).into(),
NestedField::required(2, "bar", Type::Primitive(PrimitiveType::Int)).into(),
NestedField::optional(3, "baz", Type::Primitive(PrimitiveType::Boolean)).into(),
])
.build()
.unwrap();
test_serde_json(
r#"
{
"action": "add-schema",
"schema": {
"type": "struct",
"schema-id": 1,
"fields": [
{
"id": 1,
"name": "foo",
"required": false,
"type": "string"
},
{
"id": 2,
"name": "bar",
"required": true,
"type": "int"
},
{
"id": 3,
"name": "baz",
"required": false,
"type": "boolean"
}
],
"identifier-field-ids": [
2
]
},
"last-column-id": 3
}
"#,
ViewUpdate::AddSchema {
schema: test_schema.clone(),
last_column_id: Some(3),
},
);
}
#[test]
fn test_view_set_location() {
test_serde_json(
r#"
{
"action": "set-location",
"location": "s3://db/view"
}
"#,
ViewUpdate::SetLocation {
location: "s3://db/view".to_string(),
},
);
}
#[test]
fn test_view_set_properties() {
test_serde_json(
r#"
{
"action": "set-properties",
"updates": {
"prop1": "v1",
"prop2": "v2"
}
}
"#,
ViewUpdate::SetProperties {
updates: vec![
("prop1".to_string(), "v1".to_string()),
("prop2".to_string(), "v2".to_string()),
]
.into_iter()
.collect(),
},
);
}
#[test]
fn test_view_remove_properties() {
test_serde_json(
r#"
{
"action": "remove-properties",
"removals": [
"prop1",
"prop2"
]
}
"#,
ViewUpdate::RemoveProperties {
removals: vec!["prop1".to_string(), "prop2".to_string()],
},
);
}
#[test]
fn test_view_add_view_version() {
test_serde_json(
r#"
{
"action": "add-view-version",
"view-version": {
"version-id" : 1,
"timestamp-ms" : 1573518431292,
"schema-id" : 1,
"default-catalog" : "prod",
"default-namespace" : [ "default" ],
"summary" : {
"engine-name" : "Spark"
},
"representations" : [ {
"type" : "sql",
"sql" : "SELECT\n COUNT(1), CAST(event_ts AS DATE)\nFROM events\nGROUP BY 2",
"dialect" : "spark"
} ]
}
}
"#,
ViewUpdate::AddViewVersion {
view_version: ViewVersion::builder()
.with_version_id(1)
.with_timestamp_ms(1573518431292)
.with_schema_id(1)
.with_default_catalog(Some("prod".to_string()))
.with_default_namespace(NamespaceIdent::from_strs(vec!["default"]).unwrap())
.with_summary(
vec![("engine-name".to_string(), "Spark".to_string())]
.into_iter()
.collect(),
)
.with_representations(ViewRepresentations(vec![ViewRepresentation::Sql(SqlViewRepresentation {
sql: "SELECT\n COUNT(1), CAST(event_ts AS DATE)\nFROM events\nGROUP BY 2".to_string(),
dialect: "spark".to_string(),
})]))
.build(),
},
);
}
#[test]
fn test_view_set_current_view_version() {
test_serde_json(
r#"
{
"action": "set-current-view-version",
"view-version-id": 1
}
"#,
ViewUpdate::SetCurrentViewVersion { view_version_id: 1 },
);
}
#[test]
fn test_remove_partition_specs_update() {
test_serde_json(
r#"
{
"action": "remove-partition-specs",
"spec-ids": [1, 2]
}
"#,
TableUpdate::RemovePartitionSpecs {
spec_ids: vec![1, 2],
},
);
}
#[test]
fn test_set_statistics_file() {
test_serde_json(
r#"
{
"action": "set-statistics",
"snapshot-id": 1940541653261589030,
"statistics": {
"snapshot-id": 1940541653261589030,
"statistics-path": "s3://bucket/warehouse/stats.puffin",
"file-size-in-bytes": 124,
"file-footer-size-in-bytes": 27,
"blob-metadata": [
{
"type": "boring-type",
"snapshot-id": 1940541653261589030,
"sequence-number": 2,
"fields": [
1
],
"properties": {
"prop-key": "prop-value"
}
}
]
}
}
"#,
TableUpdate::SetStatistics {
statistics: StatisticsFile {
snapshot_id: 1940541653261589030,
statistics_path: "s3://bucket/warehouse/stats.puffin".to_string(),
file_size_in_bytes: 124,
file_footer_size_in_bytes: 27,
key_metadata: None,
blob_metadata: vec![BlobMetadata {
r#type: "boring-type".to_string(),
snapshot_id: 1940541653261589030,
sequence_number: 2,
fields: vec![1],
properties: vec![("prop-key".to_string(), "prop-value".to_string())]
.into_iter()
.collect(),
}],
},
},
);
}
#[test]
fn test_remove_statistics_file() {
test_serde_json(
r#"
{
"action": "remove-statistics",
"snapshot-id": 1940541653261589030
}
"#,
TableUpdate::RemoveStatistics {
snapshot_id: 1940541653261589030,
},
);
}
#[test]
fn test_set_partition_statistics_file() {
test_serde_json(
r#"
{
"action": "set-partition-statistics",
"partition-statistics": {
"snapshot-id": 1940541653261589030,
"statistics-path": "s3://bucket/warehouse/stats1.parquet",
"file-size-in-bytes": 43
}
}
"#,
TableUpdate::SetPartitionStatistics {
partition_statistics: PartitionStatisticsFile {
snapshot_id: 1940541653261589030,
statistics_path: "s3://bucket/warehouse/stats1.parquet".to_string(),
file_size_in_bytes: 43,
},
},
)
}
#[test]
fn test_remove_partition_statistics_file() {
test_serde_json(
r#"
{
"action": "remove-partition-statistics",
"snapshot-id": 1940541653261589030
}
"#,
TableUpdate::RemovePartitionStatistics {
snapshot_id: 1940541653261589030,
},
)
}
#[test]
fn test_remove_schema_update() {
test_serde_json(
r#"
{
"action": "remove-schemas",
"schema-ids": [1, 2]
}
"#,
TableUpdate::RemoveSchemas {
schema_ids: vec![1, 2],
},
);
}
#[test]
fn test_add_encryption_key() {
let key_bytes = "key".as_bytes();
let encoded_key = base64::engine::general_purpose::STANDARD.encode(key_bytes);
test_serde_json(
format!(
r#"
{{
"action": "add-encryption-key",
"encryption-key": {{
"key-id": "a",
"encrypted-key-metadata": "{encoded_key}",
"encrypted-by-id": "b"
}}
}}
"#
),
TableUpdate::AddEncryptionKey {
encryption_key: EncryptedKey::builder()
.key_id("a")
.encrypted_key_metadata(key_bytes.to_vec())
.encrypted_by_id("b")
.build(),
},
);
}
#[test]
fn test_remove_encryption_key() {
test_serde_json(
r#"
{
"action": "remove-encryption-key",
"key-id": "a"
}
"#,
TableUpdate::RemoveEncryptionKey {
key_id: "a".to_string(),
},
);
}
#[test]
fn test_table_commit() {
let table = {
let file = File::open(format!(
"{}/testdata/table_metadata/{}",
env!("CARGO_MANIFEST_DIR"),
"TableMetadataV2Valid.json"
))
.unwrap();
let reader = BufReader::new(file);
let resp = serde_json::from_reader::<_, TableMetadata>(reader).unwrap();
Table::builder()
.metadata(resp)
.metadata_location("s3://bucket/test/location/metadata/00000-8a62c37d-4573-4021-952a-c0baef7d21d0.metadata.json")
.identifier(TableIdent::from_strs(["ns1", "test1"]).unwrap())
.file_io(FileIO::new_with_memory())
.runtime(test_runtime())
.build()
.unwrap()
};
let updates = vec![
TableUpdate::SetLocation {
location: "s3://bucket/test/new_location/data".to_string(),
},
TableUpdate::SetProperties {
updates: vec![
("prop1".to_string(), "v1".to_string()),
("prop2".to_string(), "v2".to_string()),
]
.into_iter()
.collect(),
},
];
let requirements = vec![TableRequirement::UuidMatch {
uuid: table.metadata().table_uuid,
}];
let table_commit = TableCommit::builder()
.ident(table.identifier().to_owned())
.updates(updates)
.requirements(requirements)
.build();
let updated_table = table_commit.apply(table).unwrap();
assert_eq!(
updated_table.metadata().properties.get("prop1").unwrap(),
"v1"
);
assert_eq!(
updated_table.metadata().properties.get("prop2").unwrap(),
"v2"
);
// metadata version should be bumped, and the metadata directory should be
// re-derived from the updated table location
assert!(
updated_table
.metadata_location()
.unwrap()
.starts_with("s3://bucket/test/new_location/data/metadata/00001-")
);
assert_eq!(
updated_table.metadata().location,
"s3://bucket/test/new_location/data",
);
}
}