diff --git a/catalogs/iceberg-file-catalog/src/lib.rs b/catalogs/iceberg-file-catalog/src/lib.rs index eb0587dd..a2a24b51 100644 --- a/catalogs/iceberg-file-catalog/src/lib.rs +++ b/catalogs/iceberg-file-catalog/src/lib.rs @@ -1,5 +1,5 @@ use std::{ - collections::HashMap, + collections::{HashMap, HashSet}, convert::identity, sync::{Arc, RwLock}, }; @@ -40,6 +40,7 @@ pub struct FileCatalog { path: String, object_store: ObjectStoreBuilder, cache: Arc>>, + renamed_tables: Arc>>, } pub mod error; @@ -50,6 +51,7 @@ impl FileCatalog { path: path.to_owned(), object_store, cache: Arc::new(RwLock::new(HashMap::new())), + renamed_tables: Arc::new(RwLock::new(HashMap::new())), }) } @@ -139,7 +141,7 @@ impl Catalog for FileCatalog { let bucket = Bucket::from_path(&self.path)?; let object_store = self.object_store.build(bucket)?; - object_store + let mut identifiers: HashSet = object_store .list(Some( &strip_prefix(&self.namespace_path(&namespace[0])).into(), )) @@ -148,8 +150,32 @@ impl Catalog for FileCatalog { let path = x.location.as_ref(); self.identifier(path) }) - .try_collect() - .await + .try_collect::>() + .await?; + + let renamed_tables = self.renamed_tables.read().unwrap(); + for (destination, source) in renamed_tables.iter() { + if destination != source { + identifiers.remove(source); + } + if destination.namespace() == namespace { + identifiers.insert(destination.clone()); + } + } + drop(renamed_tables); + + identifiers.extend( + self.cache + .read() + .unwrap() + .keys() + .filter(|identifier| identifier.namespace() == namespace) + .cloned(), + ); + + let mut identifiers = identifiers.into_iter().collect::>(); + identifiers.sort_by_key(ToString::to_string); + Ok(identifiers) } async fn list_namespaces(&self, _parent: Option<&str>) -> Result, IcebergError> { let bucket = Bucket::from_path(&self.path)?; @@ -175,6 +201,61 @@ impl Catalog for FileCatalog { async fn drop_table(&self, identifier: &Identifier) -> Result<(), IcebergError> { self.drop_tabular(identifier).await } + async fn rename_table( + &self, + source: &Identifier, + destination: &Identifier, + ) -> Result<(), IcebergError> { + if !matches!(self.object_store, ObjectStoreBuilder::Memory(_)) { + return Err(IcebergError::NotSupported( + "rename table in persistent file catalogs".to_string(), + )); + } + if source == destination { + return Ok(()); + } + if self.tabular_exists(destination).await? { + return Err(IcebergError::InvalidFormat(format!( + "Destination table {destination} already exists" + ))); + } + + let metadata_location = self.metadata_location(source).await?; + let cached_metadata = { + let cache = self.cache.read().unwrap(); + cache.get(source).map(|(_, metadata)| metadata.clone()) + }; + let metadata = if let Some(metadata) = cached_metadata { + metadata + } else { + let bucket = Bucket::from_path(&self.path)?; + let object_store = self.object_store.build(bucket)?; + let bytes = object_store + .get(&strip_prefix(&metadata_location).as_str().into()) + .await + .map_err(|_| IcebergError::CatalogNotFound)? + .bytes() + .await?; + serde_json::from_slice(&bytes)? + }; + if !matches!(metadata, TabularMetadata::Table(_)) { + return Err(IcebergError::InvalidFormat(format!( + "Source {source} is not a table" + ))); + } + + let mut cache = self.cache.write().unwrap(); + cache.remove(source); + cache.insert(destination.clone(), (metadata_location, metadata)); + drop(cache); + + let mut renamed_tables = self.renamed_tables.write().unwrap(); + let physical_source = renamed_tables + .remove(source) + .unwrap_or_else(|| source.clone()); + renamed_tables.insert(destination.clone(), physical_source); + Ok(()) + } async fn drop_view(&self, identifier: &Identifier) -> Result<(), IcebergError> { self.drop_tabular(identifier).await } @@ -356,7 +437,6 @@ impl Catalog for FileCatalog { identifier.clone(), (metadata_location.clone(), metadata.clone().into()), ); - Ok(MaterializedView::new(identifier.clone(), self.clone(), metadata).await?) } @@ -575,8 +655,18 @@ impl FileCatalog { let bucket = Bucket::from_path(&self.path)?; let object_store = self.object_store.build(bucket)?; - let prefix = - strip_prefix(&self.tabular_path(&identifier.namespace()[0], identifier.name())); + let physical_identifier = self + .renamed_tables + .read() + .unwrap() + .get(identifier) + .cloned() + .unwrap_or_else(|| identifier.clone()); + let tabular_path = self.tabular_path( + &physical_identifier.namespace()[0], + physical_identifier.name(), + ); + let prefix = strip_prefix(&tabular_path); let paths: Vec<_> = object_store .list(Some(&prefix.as_str().into())) .map_ok(|x| x.location) @@ -592,14 +682,31 @@ impl FileCatalog { } self.cache.write().unwrap().remove(identifier); + self.renamed_tables.write().unwrap().remove(identifier); Ok(()) } async fn metadata_location(&self, identifier: &Identifier) -> Result { + let physical_identifier = { + let renamed_tables = self.renamed_tables.read().unwrap(); + if renamed_tables.values().any(|source| source == identifier) + && !renamed_tables.contains_key(identifier) + { + return Err(IcebergError::CatalogNotFound); + } + renamed_tables + .get(identifier) + .cloned() + .unwrap_or_else(|| identifier.clone()) + }; + let bucket = Bucket::from_path(&self.path)?; let object_store = self.object_store.build(bucket)?; - let path = self.tabular_path(&identifier.namespace()[0], identifier.name()) + "/metadata"; + let path = self.tabular_path( + &physical_identifier.namespace()[0], + physical_identifier.name(), + ) + "/metadata"; let mut files: Vec = object_store .list(Some(&strip_prefix(&path).into())) .map_ok(|x| x.location.to_string()) @@ -807,6 +914,7 @@ impl CatalogList for FileCatalogList { path: self.path.clone() + "/" + name, object_store: self.object_store.clone(), cache: Arc::new(RwLock::new(HashMap::new())), + renamed_tables: Arc::new(RwLock::new(HashMap::new())), })) } async fn list_catalogs(&self) -> Vec { @@ -843,9 +951,13 @@ pub mod tests { }; use futures::StreamExt; use iceberg_rust::{ - catalog::{namespace::Namespace, Catalog}, + catalog::{create::CreateTableBuilder, namespace::Namespace, Catalog}, object_store::{Bucket, ObjectStoreBuilder}, - spec::util::strip_prefix, + spec::{ + schema::Schema, + types::{PrimitiveType, StructField, Type}, + util::strip_prefix, + }, }; use object_store::ObjectStoreExt; use std::{sync::Arc, time::Duration}; @@ -857,6 +969,49 @@ pub mod tests { use crate::FileCatalog; + #[tokio::test] + async fn rename_table_in_memory_moves_catalog_identity() { + let catalog = Arc::new( + FileCatalog::new("/dev/embucket", ObjectStoreBuilder::memory()) + .await + .unwrap(), + ); + let namespace = Namespace::try_new(&["public".to_string()]).unwrap(); + let source = iceberg_rust::catalog::identifier::Identifier::new(&namespace, "source_table"); + let destination = + iceberg_rust::catalog::identifier::Identifier::new(&namespace, "destination_table"); + let schema = Schema::builder() + .with_struct_field(StructField { + id: 1, + name: "id".to_string(), + required: true, + field_type: Type::Primitive(PrimitiveType::Long), + doc: None, + initial_default: None, + write_default: None, + }) + .build() + .unwrap(); + let mut builder = CreateTableBuilder::default(); + builder + .with_name(source.name()) + .with_schema(schema) + .build(&namespace, catalog.clone()) + .await + .unwrap(); + + catalog.rename_table(&source, &destination).await.unwrap(); + + assert!(!catalog.tabular_exists(&source).await.unwrap()); + assert!(catalog.tabular_exists(&destination).await.unwrap()); + assert_eq!( + catalog.list_tabulars(&namespace).await.unwrap(), + vec![destination.clone()] + ); + let tabular = catalog.clone().load_tabular(&destination).await.unwrap(); + assert_eq!(tabular.identifier(), &destination); + } + #[tokio::test] async fn test_create_update_drop_table() { let localstack = LocalStack::default() diff --git a/catalogs/iceberg-rest-catalog/src/catalog.rs b/catalogs/iceberg-rest-catalog/src/catalog.rs index addfb186..072345cb 100644 --- a/catalogs/iceberg-rest-catalog/src/catalog.rs +++ b/catalogs/iceberg-rest-catalog/src/catalog.rs @@ -298,6 +298,20 @@ impl Catalog for RestCatalog { .await .map_err(Into::::into) } + /// Rename a table through the Iceberg REST catalog's atomic endpoint. + async fn rename_table( + &self, + source: &Identifier, + destination: &Identifier, + ) -> Result<(), Error> { + catalog_api_api::rename_table( + &self.configuration, + self.name.as_deref(), + models::RenameTableRequest::new(source.clone(), destination.clone()), + ) + .await + .map_err(Into::::into) + } /// Drop a table and delete all data and metadata files. async fn drop_view(&self, identifier: &Identifier) -> Result<(), Error> { let configuration = self.configuration.clone(); diff --git a/iceberg-rust/src/catalog/mod.rs b/iceberg-rust/src/catalog/mod.rs index 8d872551..d5a37a97 100644 --- a/iceberg-rust/src/catalog/mod.rs +++ b/iceberg-rust/src/catalog/mod.rs @@ -201,6 +201,17 @@ pub trait Catalog: Send + Sync + Debug { /// * The catalog fails to delete the table metadata /// * The data files cannot be deleted async fn drop_table(&self, identifier: &Identifier) -> Result<(), Error>; + /// Renames a table without changing its data or metadata location. + /// + /// Catalog implementations that cannot atomically update the identifier + /// should keep the default and report the operation as unsupported. + async fn rename_table( + &self, + _source: &Identifier, + _destination: &Identifier, + ) -> Result<(), Error> { + Err(Error::NotSupported("rename table".to_string())) + } /// Drops a view from the catalog and deletes its metadata. /// /// # Arguments