From 3797becbf8d7a81b315bbb1102c96147e715dd71 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Thu, 3 Sep 2026 07:03:27 +0300 Subject: [PATCH 1/2] Support catalog table rename --- catalogs/iceberg-file-catalog/src/lib.rs | 158 ++++++++++++++++++- catalogs/iceberg-rest-catalog/src/catalog.rs | 14 ++ iceberg-rust/src/catalog/mod.rs | 11 ++ 3 files changed, 175 insertions(+), 8 deletions(-) diff --git a/catalogs/iceberg-file-catalog/src/lib.rs b/catalogs/iceberg-file-catalog/src/lib.rs index eb0587dd..fcd4b597 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_sources: Arc>>, } pub mod error; @@ -50,6 +51,7 @@ impl FileCatalog { path: path.to_owned(), object_store, cache: Arc::new(RwLock::new(HashMap::new())), + renamed_sources: Arc::new(RwLock::new(HashSet::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,25 @@ impl Catalog for FileCatalog { let path = x.location.as_ref(); self.identifier(path) }) - .try_collect() - .await + .try_collect::>() + .await?; + + let renamed_sources = self.renamed_sources.read().unwrap(); + identifiers.retain(|identifier| !renamed_sources.contains(identifier)); + drop(renamed_sources); + + 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 +194,59 @@ 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_sources = self.renamed_sources.write().unwrap(); + renamed_sources.insert(source.clone()); + renamed_sources.remove(destination); + Ok(()) + } async fn drop_view(&self, identifier: &Identifier) -> Result<(), IcebergError> { self.drop_tabular(identifier).await } @@ -264,6 +336,7 @@ impl Catalog for FileCatalog { identifier.clone(), (metadata_location.clone(), metadata.clone().into()), ); + self.renamed_sources.write().unwrap().remove(&identifier); Ok(Table::new( identifier.clone(), self.clone(), @@ -306,6 +379,7 @@ impl Catalog for FileCatalog { identifier.clone(), (metadata_location.clone(), metadata.clone().into()), ); + self.renamed_sources.write().unwrap().remove(&identifier); Ok(View::new(identifier.clone(), self.clone(), metadata).await?) } @@ -356,6 +430,7 @@ impl Catalog for FileCatalog { identifier.clone(), (metadata_location.clone(), metadata.clone().into()), ); + self.renamed_sources.write().unwrap().remove(&identifier); Ok(MaterializedView::new(identifier.clone(), self.clone(), metadata).await?) } @@ -575,8 +650,20 @@ 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 cached_location = self + .cache + .read() + .unwrap() + .get(identifier) + .map(|(metadata_location, _)| metadata_location.clone()); + let tabular_path = cached_location + .as_deref() + .and_then(|location| location.rsplit_once("/metadata/").map(|(path, _)| path)) + .map_or_else( + || self.tabular_path(&identifier.namespace()[0], identifier.name()), + ToOwned::to_owned, + ); + let prefix = strip_prefix(&tabular_path); let paths: Vec<_> = object_store .list(Some(&prefix.as_str().into())) .map_ok(|x| x.location) @@ -596,6 +683,13 @@ impl FileCatalog { } async fn metadata_location(&self, identifier: &Identifier) -> Result { + if self.renamed_sources.read().unwrap().contains(identifier) { + return Err(IcebergError::CatalogNotFound); + } + if let Some((metadata_location, _)) = self.cache.read().unwrap().get(identifier) { + return Ok(metadata_location.clone()); + } + let bucket = Bucket::from_path(&self.path)?; let object_store = self.object_store.build(bucket)?; @@ -807,6 +901,7 @@ impl CatalogList for FileCatalogList { path: self.path.clone() + "/" + name, object_store: self.object_store.clone(), cache: Arc::new(RwLock::new(HashMap::new())), + renamed_sources: Arc::new(RwLock::new(HashSet::new())), })) } async fn list_catalogs(&self) -> Vec { @@ -843,9 +938,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 +956,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 From 9319e51f896bfc9f24bb1b70ce4e7a9e022331a2 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Thu, 3 Sep 2026 07:22:34 +0300 Subject: [PATCH 2/2] Keep renamed tables on fresh metadata --- catalogs/iceberg-file-catalog/src/lib.rs | 73 ++++++++++++++---------- 1 file changed, 43 insertions(+), 30 deletions(-) diff --git a/catalogs/iceberg-file-catalog/src/lib.rs b/catalogs/iceberg-file-catalog/src/lib.rs index fcd4b597..a2a24b51 100644 --- a/catalogs/iceberg-file-catalog/src/lib.rs +++ b/catalogs/iceberg-file-catalog/src/lib.rs @@ -40,7 +40,7 @@ pub struct FileCatalog { path: String, object_store: ObjectStoreBuilder, cache: Arc>>, - renamed_sources: Arc>>, + renamed_tables: Arc>>, } pub mod error; @@ -51,7 +51,7 @@ impl FileCatalog { path: path.to_owned(), object_store, cache: Arc::new(RwLock::new(HashMap::new())), - renamed_sources: Arc::new(RwLock::new(HashSet::new())), + renamed_tables: Arc::new(RwLock::new(HashMap::new())), }) } @@ -153,9 +153,16 @@ impl Catalog for FileCatalog { .try_collect::>() .await?; - let renamed_sources = self.renamed_sources.read().unwrap(); - identifiers.retain(|identifier| !renamed_sources.contains(identifier)); - drop(renamed_sources); + 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 @@ -242,9 +249,11 @@ impl Catalog for FileCatalog { cache.insert(destination.clone(), (metadata_location, metadata)); drop(cache); - let mut renamed_sources = self.renamed_sources.write().unwrap(); - renamed_sources.insert(source.clone()); - renamed_sources.remove(destination); + 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> { @@ -336,7 +345,6 @@ impl Catalog for FileCatalog { identifier.clone(), (metadata_location.clone(), metadata.clone().into()), ); - self.renamed_sources.write().unwrap().remove(&identifier); Ok(Table::new( identifier.clone(), self.clone(), @@ -379,7 +387,6 @@ impl Catalog for FileCatalog { identifier.clone(), (metadata_location.clone(), metadata.clone().into()), ); - self.renamed_sources.write().unwrap().remove(&identifier); Ok(View::new(identifier.clone(), self.clone(), metadata).await?) } @@ -430,8 +437,6 @@ impl Catalog for FileCatalog { identifier.clone(), (metadata_location.clone(), metadata.clone().into()), ); - self.renamed_sources.write().unwrap().remove(&identifier); - Ok(MaterializedView::new(identifier.clone(), self.clone(), metadata).await?) } @@ -650,19 +655,17 @@ impl FileCatalog { let bucket = Bucket::from_path(&self.path)?; let object_store = self.object_store.build(bucket)?; - let cached_location = self - .cache + let physical_identifier = self + .renamed_tables .read() .unwrap() .get(identifier) - .map(|(metadata_location, _)| metadata_location.clone()); - let tabular_path = cached_location - .as_deref() - .and_then(|location| location.rsplit_once("/metadata/").map(|(path, _)| path)) - .map_or_else( - || self.tabular_path(&identifier.namespace()[0], identifier.name()), - ToOwned::to_owned, - ); + .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())) @@ -679,21 +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 { - if self.renamed_sources.read().unwrap().contains(identifier) { - return Err(IcebergError::CatalogNotFound); - } - if let Some((metadata_location, _)) = self.cache.read().unwrap().get(identifier) { - return Ok(metadata_location.clone()); - } + 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()) @@ -901,7 +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_sources: Arc::new(RwLock::new(HashSet::new())), + renamed_tables: Arc::new(RwLock::new(HashMap::new())), })) } async fn list_catalogs(&self) -> Vec {