You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by li...@apache.org on 2026/03/12 07:33:54 UTC
(iceberg-rust) branch main updated: Add catalog test suite to unify catalog's behavior. (#2131)
This is an automated email from the ASF dual-hosted git repository.
liurenjie1024 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 335961ae0 Add catalog test suite to unify catalog's behavior. (#2131)
335961ae0 is described below
commit 335961ae013b491667d8c815b26aa6a2c43d4de2
Author: blackmwk <li...@outlook.com>
AuthorDate: Thu Mar 12 15:33:49 2026 +0800
Add catalog test suite to unify catalog's behavior. (#2131)
## Which issue does this PR close?
- Closes #2086 .
## What changes are included in this PR?
In this pr we introduced catalog test suite in catalog-loader, which
could unify the behavior of catalogs.
## Are these changes tested?
Yes.
---------
Co-authored-by: Ray Liu <li...@gmail.com>
---
Cargo.lock | 47 +-
Cargo.toml | 1 +
crates/catalog/glue/Cargo.toml | 1 -
crates/catalog/glue/src/catalog.rs | 37 +-
crates/catalog/glue/tests/glue_catalog_test.rs | 499 ---------------------
crates/catalog/hms/src/catalog.rs | 65 ++-
crates/catalog/hms/tests/hms_catalog_test.rs | 296 +-----------
crates/catalog/loader/Cargo.toml | 4 +
crates/catalog/loader/tests/common/mod.rs | 342 ++++++++++++++
crates/catalog/loader/tests/namespace_suite.rs | 355 +++++++++++++++
.../catalog/loader/tests/table_register_suite.rs | 167 +++++++
crates/catalog/loader/tests/table_rename_suite.rs | 163 +++++++
crates/catalog/loader/tests/table_suite.rs | 276 ++++++++++++
crates/catalog/rest/Cargo.toml | 1 -
crates/catalog/rest/src/catalog.rs | 22 +-
crates/catalog/rest/tests/rest_catalog_test.rs | 497 --------------------
crates/catalog/s3tables/src/catalog.rs | 21 +
crates/catalog/sql/src/catalog.rs | 449 +-----------------
crates/catalog/sql/src/error.rs | 6 +-
crates/iceberg/src/catalog/memory/catalog.rs | 4 +-
20 files changed, 1496 insertions(+), 1757 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 43b46822c..8419709a4 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2878,6 +2878,12 @@ version = "0.3.32"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393"
+[[package]]
+name = "futures-timer"
+version = "3.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f288b0a4f20f9a56b5d1da57e2227c661b7b16168e2f72365f57b63326e29b24"
+
[[package]]
name = "futures-util"
version = "0.3.32"
@@ -3386,7 +3392,6 @@ dependencies = [
"iceberg_test_utils",
"serde_json",
"tokio",
- "tracing",
]
[[package]]
@@ -3423,6 +3428,10 @@ dependencies = [
"iceberg-catalog-rest",
"iceberg-catalog-s3tables",
"iceberg-catalog-sql",
+ "iceberg-storage-opendal",
+ "iceberg_test_utils",
+ "reqwest",
+ "rstest",
"sqlx",
"tempfile",
"tokio",
@@ -3444,7 +3453,6 @@ dependencies = [
"serde_derive",
"serde_json",
"tokio",
- "tracing",
"typed-builder",
"uuid",
]
@@ -5355,6 +5363,12 @@ dependencies = [
"memchr",
]
+[[package]]
+name = "relative-path"
+version = "1.9.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba39f3699c378cd8970968dcbff9c43159ea4cfbd88d43c00b22f2ef10a435d2"
+
[[package]]
name = "rend"
version = "0.5.3"
@@ -5510,6 +5524,35 @@ dependencies = [
"zeroize",
]
+[[package]]
+name = "rstest"
+version = "0.26.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f5a3193c063baaa2a95a33f03035c8a72b83d97a54916055ba22d35ed3839d49"
+dependencies = [
+ "futures-timer",
+ "futures-util",
+ "rstest_macros",
+]
+
+[[package]]
+name = "rstest_macros"
+version = "0.26.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9c845311f0ff7951c5506121a9ad75aec44d083c31583b2ea5a30bcb0b0abba0"
+dependencies = [
+ "cfg-if",
+ "glob",
+ "proc-macro-crate",
+ "proc-macro2",
+ "quote",
+ "regex",
+ "relative-path",
+ "rustc_version",
+ "syn",
+ "unicode-ident",
+]
+
[[package]]
name = "rust-ini"
version = "0.21.3"
diff --git a/Cargo.toml b/Cargo.toml
index b3290d564..6a361ecbd 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -110,6 +110,7 @@ rand = "0.8.5"
regex = "1.11.3"
reqwest = { version = "0.12.12", default-features = false, features = ["json"] }
roaring = { version = "0.11" }
+rstest = "0.26"
fastnum = { version = "0.7", default-features = false, features = ["std", "serde"] }
serde = { version = "1.0.219", features = ["rc"] }
serde_bytes = "0.11.17"
diff --git a/crates/catalog/glue/Cargo.toml b/crates/catalog/glue/Cargo.toml
index e41253de3..d8d0927a9 100644
--- a/crates/catalog/glue/Cargo.toml
+++ b/crates/catalog/glue/Cargo.toml
@@ -37,7 +37,6 @@ iceberg = { workspace = true }
iceberg-storage-opendal = { workspace = true, features = ["opendal-s3"] }
serde_json = { workspace = true }
tokio = { workspace = true }
-tracing = { workspace = true }
[dev-dependencies]
iceberg_test_utils = { path = "../../test_utils", features = ["tests"] }
diff --git a/crates/catalog/glue/src/catalog.rs b/crates/catalog/glue/src/catalog.rs
index 968b19716..9e9d4580c 100644
--- a/crates/catalog/glue/src/catalog.rs
+++ b/crates/catalog/glue/src/catalog.rs
@@ -339,6 +339,13 @@ impl Catalog for GlueCatalog {
namespace: &NamespaceIdent,
properties: HashMap<String, String>,
) -> Result<Namespace> {
+ if self.namespace_exists(namespace).await? {
+ return Err(Error::new(
+ ErrorKind::NamespaceAlreadyExists,
+ format!("Namespace {namespace:?} already exists"),
+ ));
+ }
+
let db_input = convert_to_database(namespace, &properties)?;
let builder = self.client.0.create_database().database_input(db_input);
@@ -365,7 +372,19 @@ impl Catalog for GlueCatalog {
let builder = self.client.0.get_database().name(&db_name);
let builder = with_catalog_id!(builder, self.config);
- let resp = builder.send().await.map_err(from_aws_sdk_error)?;
+ let resp = builder.send().await.map_err(|err| {
+ if err
+ .as_service_error()
+ .map(|e| e.is_entity_not_found_exception())
+ == Some(true)
+ {
+ return Error::new(
+ ErrorKind::NamespaceNotFound,
+ format!("Namespace {namespace:?} does not exist"),
+ );
+ }
+ from_aws_sdk_error(err)
+ })?;
match resp.database() {
Some(db) => {
@@ -373,7 +392,7 @@ impl Catalog for GlueCatalog {
Ok(namespace)
}
None => Err(Error::new(
- ErrorKind::DataInvalid,
+ ErrorKind::NamespaceNotFound,
format!("Database with name: {db_name} does not exist"),
)),
}
@@ -429,6 +448,13 @@ impl Catalog for GlueCatalog {
namespace: &NamespaceIdent,
properties: HashMap<String, String>,
) -> Result<()> {
+ if !self.namespace_exists(namespace).await? {
+ return Err(Error::new(
+ ErrorKind::NamespaceNotFound,
+ format!("Namespace {namespace:?} does not exist"),
+ ));
+ }
+
let db_name = validate_namespace(namespace)?;
let db_input = convert_to_database(namespace, &properties)?;
@@ -456,6 +482,13 @@ impl Catalog for GlueCatalog {
/// - `Err(...)` signifies failure to drop the namespace due to validation
/// errors, connectivity issues, or Glue Catalog constraints.
async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()> {
+ if !self.namespace_exists(namespace).await? {
+ return Err(Error::new(
+ ErrorKind::NamespaceNotFound,
+ format!("Namespace {namespace:?} does not exist"),
+ ));
+ }
+
let db_name = validate_namespace(namespace)?;
let table_list = self.list_tables(namespace).await?;
diff --git a/crates/catalog/glue/tests/glue_catalog_test.rs b/crates/catalog/glue/tests/glue_catalog_test.rs
deleted file mode 100644
index 0b7dbe9f2..000000000
--- a/crates/catalog/glue/tests/glue_catalog_test.rs
+++ /dev/null
@@ -1,499 +0,0 @@
-// 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.
-
-//! Integration tests for glue catalog.
-//!
-//! These tests assume Docker containers are started externally via `make docker-up`.
-//! Each test uses unique namespaces based on module path to avoid conflicts.
-
-use std::collections::HashMap;
-use std::sync::Arc;
-
-use iceberg::io::{FileIOBuilder, S3_ACCESS_KEY_ID, S3_ENDPOINT, S3_REGION, S3_SECRET_ACCESS_KEY};
-use iceberg::spec::{NestedField, PrimitiveType, Schema, Type};
-use iceberg::transaction::{ApplyTransactionAction, Transaction};
-use iceberg::{
- Catalog, CatalogBuilder, Namespace, NamespaceIdent, Result, TableCreation, TableIdent,
-};
-use iceberg_catalog_glue::{
- AWS_ACCESS_KEY_ID, AWS_REGION_NAME, AWS_SECRET_ACCESS_KEY, GLUE_CATALOG_PROP_URI,
- GLUE_CATALOG_PROP_WAREHOUSE, GlueCatalog, GlueCatalogBuilder,
-};
-use iceberg_storage_opendal::OpenDalStorageFactory;
-use iceberg_test_utils::{
- cleanup_namespace, get_glue_endpoint, get_minio_endpoint, normalize_test_name_with_parts,
- set_up,
-};
-use tokio::time::sleep;
-use tracing::info;
-
-async fn get_catalog() -> GlueCatalog {
- set_up();
-
- let glue_endpoint = get_glue_endpoint();
- let minio_endpoint = get_minio_endpoint();
-
- let props = HashMap::from([
- (AWS_ACCESS_KEY_ID.to_string(), "my_access_id".to_string()),
- (
- AWS_SECRET_ACCESS_KEY.to_string(),
- "my_secret_key".to_string(),
- ),
- (AWS_REGION_NAME.to_string(), "us-east-1".to_string()),
- (S3_ENDPOINT.to_string(), minio_endpoint),
- (S3_ACCESS_KEY_ID.to_string(), "admin".to_string()),
- (S3_SECRET_ACCESS_KEY.to_string(), "password".to_string()),
- (S3_REGION.to_string(), "us-east-1".to_string()),
- ]);
-
- // Wait for bucket to actually exist
- let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 {
- configured_scheme: "s3a".to_string(),
- customized_credential_load: None,
- }))
- .with_props(props.clone())
- .build();
-
- let mut retries = 0;
- while retries < 30 {
- if file_io.exists("s3a://warehouse/").await.unwrap_or(false) {
- info!("S3 bucket 'warehouse' is ready");
- break;
- }
- info!("Waiting for bucket creation... (attempt {})", retries + 1);
- sleep(std::time::Duration::from_millis(1000)).await;
- retries += 1;
- }
-
- let mut glue_props = HashMap::from([
- (GLUE_CATALOG_PROP_URI.to_string(), glue_endpoint),
- (
- GLUE_CATALOG_PROP_WAREHOUSE.to_string(),
- "s3a://warehouse/hive".to_string(),
- ),
- ]);
- glue_props.extend(props.clone());
-
- GlueCatalogBuilder::default()
- .load("glue", glue_props)
- .await
- .unwrap()
-}
-
-async fn set_test_namespace(catalog: &GlueCatalog, namespace: &NamespaceIdent) -> Result<()> {
- let properties = HashMap::new();
- catalog.create_namespace(namespace, properties).await?;
-
- Ok(())
-}
-
-fn set_table_creation(location: Option<String>, name: impl ToString) -> Result<TableCreation> {
- let schema = Schema::builder()
- .with_schema_id(0)
- .with_fields(vec![
- NestedField::required(1, "foo", Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "bar", Type::Primitive(PrimitiveType::String)).into(),
- ])
- .build()?;
-
- let builder = TableCreation::builder()
- .name(name.to_string())
- .properties(HashMap::new())
- .location_opt(location)
- .schema(schema);
-
- Ok(builder.build())
-}
-
-#[tokio::test]
-async fn test_rename_table() -> Result<()> {
- let catalog = get_catalog().await;
- let creation = set_table_creation(None, "my_table")?;
- // Use unique namespace to avoid conflicts
- let namespace = Namespace::new(NamespaceIdent::new(normalize_test_name_with_parts!(
- "test_rename_table"
- )));
- cleanup_namespace(&catalog, namespace.name()).await;
-
- catalog
- .create_namespace(namespace.name(), HashMap::new())
- .await?;
-
- let table = catalog.create_table(namespace.name(), creation).await?;
-
- let dest = TableIdent::new(namespace.name().clone(), "my_table_rename".to_string());
-
- catalog.rename_table(table.identifier(), &dest).await?;
-
- let table = catalog.load_table(&dest).await?;
- assert_eq!(table.identifier(), &dest);
-
- let src = TableIdent::new(namespace.name().clone(), "my_table".to_string());
-
- let src_table_exists = catalog.table_exists(&src).await?;
- assert!(!src_table_exists);
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_table_exists() -> Result<()> {
- let catalog = get_catalog().await;
- let creation = set_table_creation(None, "my_table")?;
- // Use unique namespace to avoid conflicts
- let namespace = Namespace::new(NamespaceIdent::new(normalize_test_name_with_parts!(
- "test_table_exists"
- )));
- cleanup_namespace(&catalog, namespace.name()).await;
-
- catalog
- .create_namespace(namespace.name(), HashMap::new())
- .await?;
-
- let ident = TableIdent::new(namespace.name().clone(), "my_table".to_string());
-
- let exists = catalog.table_exists(&ident).await?;
- assert!(!exists);
-
- let table = catalog.create_table(namespace.name(), creation).await?;
-
- let exists = catalog.table_exists(table.identifier()).await?;
-
- assert!(exists);
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_drop_table() -> Result<()> {
- let catalog = get_catalog().await;
- let creation = set_table_creation(None, "my_table")?;
- // Use unique namespace to avoid conflicts
- let namespace = Namespace::new(NamespaceIdent::new(normalize_test_name_with_parts!(
- "test_drop_table"
- )));
- cleanup_namespace(&catalog, namespace.name()).await;
-
- catalog
- .create_namespace(namespace.name(), HashMap::new())
- .await?;
-
- let table = catalog.create_table(namespace.name(), creation).await?;
-
- catalog.drop_table(table.identifier()).await?;
-
- let result = catalog.table_exists(table.identifier()).await?;
-
- assert!(!result);
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_load_table() -> Result<()> {
- let catalog = get_catalog().await;
- let creation = set_table_creation(None, "my_table")?;
- // Use unique namespace to avoid conflicts
- let namespace = Namespace::new(NamespaceIdent::new(normalize_test_name_with_parts!(
- "test_load_table"
- )));
- cleanup_namespace(&catalog, namespace.name()).await;
-
- catalog
- .create_namespace(namespace.name(), HashMap::new())
- .await?;
-
- let expected = catalog.create_table(namespace.name(), creation).await?;
-
- let result = catalog
- .load_table(&TableIdent::new(
- namespace.name().clone(),
- "my_table".to_string(),
- ))
- .await?;
-
- assert_eq!(result.identifier(), expected.identifier());
- assert_eq!(result.metadata_location(), expected.metadata_location());
- assert_eq!(result.metadata(), expected.metadata());
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_create_table() -> Result<()> {
- let catalog = get_catalog().await;
- // Use unique namespace to avoid conflicts
- let namespace = NamespaceIdent::new(normalize_test_name_with_parts!("test_create_table"));
- cleanup_namespace(&catalog, &namespace).await;
- set_test_namespace(&catalog, &namespace).await?;
- // inject custom location, ignore the namespace prefix
- let creation = set_table_creation(Some("s3a://warehouse/hive".into()), "my_table")?;
- let result = catalog.create_table(&namespace, creation).await?;
-
- assert_eq!(result.identifier().name(), "my_table");
- assert!(
- result
- .metadata_location()
- .is_some_and(|location| location.starts_with("s3a://warehouse/hive/metadata/00000-"))
- );
- assert!(
- catalog
- .file_io()
- .exists("s3a://warehouse/hive/metadata/")
- .await?
- );
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_list_tables() -> Result<()> {
- let catalog = get_catalog().await;
- // Use unique namespace to avoid conflicts
- let namespace = NamespaceIdent::new(normalize_test_name_with_parts!("test_list_tables"));
- cleanup_namespace(&catalog, &namespace).await;
- set_test_namespace(&catalog, &namespace).await?;
-
- let expected = vec![];
- let result = catalog.list_tables(&namespace).await?;
-
- assert_eq!(result, expected);
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_drop_namespace() -> Result<()> {
- let catalog = get_catalog().await;
- // Use unique namespace to avoid conflicts
- let namespace = NamespaceIdent::new(normalize_test_name_with_parts!("test_drop_namespace"));
- cleanup_namespace(&catalog, &namespace).await;
- set_test_namespace(&catalog, &namespace).await?;
-
- let exists = catalog.namespace_exists(&namespace).await?;
- assert!(exists);
-
- catalog.drop_namespace(&namespace).await?;
-
- let exists = catalog.namespace_exists(&namespace).await?;
- assert!(!exists);
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_update_namespace() -> Result<()> {
- let catalog = get_catalog().await;
- // Use unique namespace to avoid conflicts
- let namespace = NamespaceIdent::new(normalize_test_name_with_parts!("test_update_namespace"));
- cleanup_namespace(&catalog, &namespace).await;
- set_test_namespace(&catalog, &namespace).await?;
-
- let before_update = catalog.get_namespace(&namespace).await?;
- let before_update = before_update.properties().get("description");
-
- assert_eq!(before_update, None);
-
- let properties = HashMap::from([("description".to_string(), "my_update".to_string())]);
-
- catalog.update_namespace(&namespace, properties).await?;
-
- let after_update = catalog.get_namespace(&namespace).await?;
- let after_update = after_update.properties().get("description");
-
- assert_eq!(after_update, Some("my_update".to_string()).as_ref());
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_namespace_exists() -> Result<()> {
- let catalog = get_catalog().await;
-
- // Use unique namespace to avoid conflicts
- let namespace = NamespaceIdent::new(normalize_test_name_with_parts!("test_namespace_exists"));
- cleanup_namespace(&catalog, &namespace).await;
-
- let exists = catalog.namespace_exists(&namespace).await?;
- assert!(!exists);
-
- set_test_namespace(&catalog, &namespace).await?;
-
- let exists = catalog.namespace_exists(&namespace).await?;
- assert!(exists);
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_get_namespace() -> Result<()> {
- let catalog = get_catalog().await;
-
- // Use unique namespace to avoid conflicts
- let namespace = NamespaceIdent::new(normalize_test_name_with_parts!("test_get_namespace"));
- cleanup_namespace(&catalog, &namespace).await;
-
- let does_not_exist = catalog.get_namespace(&namespace).await;
- assert!(does_not_exist.is_err());
-
- set_test_namespace(&catalog, &namespace).await?;
-
- let result = catalog.get_namespace(&namespace).await?;
- let expected = Namespace::new(namespace);
-
- assert_eq!(result, expected);
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_create_namespace() -> Result<()> {
- let catalog = get_catalog().await;
-
- let properties = HashMap::new();
- // Use unique namespace to avoid conflicts
- let namespace = NamespaceIdent::new(normalize_test_name_with_parts!("test_create_namespace"));
- cleanup_namespace(&catalog, &namespace).await;
-
- let expected = Namespace::new(namespace.clone());
-
- let result = catalog.create_namespace(&namespace, properties).await?;
-
- assert_eq!(result, expected);
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_list_namespace() -> Result<()> {
- let catalog = get_catalog().await;
-
- // Use unique namespace to avoid conflicts
- let namespace = NamespaceIdent::new(normalize_test_name_with_parts!("test_list_namespace"));
- cleanup_namespace(&catalog, &namespace).await;
- set_test_namespace(&catalog, &namespace).await?;
-
- let result = catalog.list_namespaces(None).await?;
- assert!(result.contains(&namespace));
-
- let empty_result = catalog.list_namespaces(Some(&namespace)).await?;
- assert!(empty_result.is_empty());
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_update_table() -> Result<()> {
- let catalog = get_catalog().await;
- let creation = set_table_creation(None, "my_table")?;
- // Use unique namespace to avoid conflicts
- let namespace = Namespace::new(NamespaceIdent::new(normalize_test_name_with_parts!(
- "test_update_table"
- )));
- cleanup_namespace(&catalog, namespace.name()).await;
-
- catalog
- .create_namespace(namespace.name(), HashMap::new())
- .await?;
-
- let expected = catalog.create_table(namespace.name(), creation).await?;
-
- let table = catalog
- .load_table(&TableIdent::new(
- namespace.name().clone(),
- "my_table".to_string(),
- ))
- .await?;
-
- assert_eq!(table.identifier(), expected.identifier());
- assert_eq!(table.metadata_location(), expected.metadata_location());
- assert_eq!(table.metadata(), expected.metadata());
-
- // Store the original metadata location for comparison
- let original_metadata_location = table.metadata_location();
-
- // Update table properties using the transaction
- let tx = Transaction::new(&table);
- let tx = tx
- .update_table_properties()
- .set("test_property".to_string(), "test_value".to_string())
- .apply(tx)?;
-
- // Commit the transaction to the catalog
- let updated_table = tx.commit(&catalog).await?;
-
- // Verify the update was successful
- assert_eq!(
- updated_table.metadata().properties().get("test_property"),
- Some(&"test_value".to_string())
- );
-
- // Verify the metadata location has been updated
- assert_ne!(
- updated_table.metadata_location(),
- original_metadata_location,
- "Metadata location should be updated after commit"
- );
-
- // Load the table again from the catalog to verify changes were persisted
- let reloaded_table = catalog.load_table(table.identifier()).await?;
-
- // Verify the reloaded table matches the updated table
- assert_eq!(
- reloaded_table.metadata().properties().get("test_property"),
- Some(&"test_value".to_string())
- );
- assert_eq!(
- reloaded_table.metadata_location(),
- updated_table.metadata_location(),
- "Reloaded table should have the same metadata location as the updated table"
- );
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_register_table() -> Result<()> {
- let catalog = get_catalog().await;
- // Use unique namespace to avoid conflicts
- let namespace = NamespaceIdent::new(normalize_test_name_with_parts!("test_register_table"));
- cleanup_namespace(&catalog, &namespace).await;
- set_test_namespace(&catalog, &namespace).await?;
-
- let location = format!("s3a://warehouse/hive/{namespace}");
- let creation = set_table_creation(Some(location), "my_table")?;
- let table = catalog.create_table(&namespace, creation).await?;
- let metadata_location = table
- .metadata_location()
- .expect("Expected metadata location to be set")
- .to_string();
-
- catalog.drop_table(table.identifier()).await?;
- let ident = TableIdent::new(namespace.clone(), "my_table".to_string());
-
- let registered = catalog
- .register_table(&ident, metadata_location.clone())
- .await?;
-
- assert_eq!(registered.identifier(), &ident);
- assert_eq!(
- registered.metadata_location(),
- Some(metadata_location.as_str())
- );
-
- Ok(())
-}
diff --git a/crates/catalog/hms/src/catalog.rs b/crates/catalog/hms/src/catalog.rs
index 69c01847a..bd7819373 100644
--- a/crates/catalog/hms/src/catalog.rs
+++ b/crates/catalog/hms/src/catalog.rs
@@ -279,6 +279,12 @@ impl Catalog for HmsCatalog {
namespace: &NamespaceIdent,
properties: HashMap<String, String>,
) -> Result<Namespace> {
+ if self.namespace_exists(namespace).await? {
+ return Err(Error::new(
+ ErrorKind::NamespaceAlreadyExists,
+ format!("Namespace {namespace:?} already exists"),
+ ));
+ }
let database = convert_to_database(namespace, &properties)?;
self.client
@@ -303,13 +309,29 @@ impl Catalog for HmsCatalog {
async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
let name = validate_namespace(namespace)?;
- let db = self
+ let resp = self
.client
.0
.get_database(name.into())
.await
- .map(from_thrift_exception)
- .map_err(from_thrift_error)??;
+ .map_err(from_thrift_error)?;
+
+ let db = match resp {
+ MaybeException::Ok(db) => db,
+ MaybeException::Exception(ThriftHiveMetastoreGetDatabaseException::O1(_)) => {
+ return Err(Error::new(
+ ErrorKind::NamespaceNotFound,
+ format!("Namespace {namespace:?} not found"),
+ ));
+ }
+ MaybeException::Exception(exception) => {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "Operation failed for hitting thrift error".to_string(),
+ )
+ .with_source(anyhow!("thrift error: {exception:?}")));
+ }
+ };
let ns = convert_to_namespace(&db)?;
@@ -362,6 +384,12 @@ impl Catalog for HmsCatalog {
namespace: &NamespaceIdent,
properties: HashMap<String, String>,
) -> Result<()> {
+ if !self.namespace_exists(namespace).await? {
+ return Err(Error::new(
+ ErrorKind::NamespaceNotFound,
+ format!("Namespace {namespace:?} does not exist"),
+ ));
+ }
let db = convert_to_database(namespace, &properties)?;
let name = match &db.name {
@@ -393,6 +421,13 @@ impl Catalog for HmsCatalog {
async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()> {
let name = validate_namespace(namespace)?;
+ if !self.namespace_exists(namespace).await? {
+ return Err(Error::new(
+ ErrorKind::NamespaceNotFound,
+ format!("Namespace {namespace:?} does not exist"),
+ ));
+ }
+
self.client
.0
.drop_database(name.into(), false, false)
@@ -413,6 +448,12 @@ impl Catalog for HmsCatalog {
/// querying the database.
async fn list_tables(&self, namespace: &NamespaceIdent) -> Result<Vec<TableIdent>> {
let name = validate_namespace(namespace)?;
+ if !self.namespace_exists(namespace).await? {
+ return Err(Error::new(
+ ErrorKind::NamespaceNotFound,
+ format!("Namespace {namespace:?} does not exist"),
+ ));
+ }
let tables = self
.client
@@ -541,6 +582,18 @@ impl Catalog for HmsCatalog {
/// - Any network or communication error occurs with the database backend.
async fn drop_table(&self, table: &TableIdent) -> Result<()> {
let db_name = validate_namespace(table.namespace())?;
+ if !self.namespace_exists(table.namespace()).await? {
+ return Err(Error::new(
+ ErrorKind::NamespaceNotFound,
+ format!("Namespace {:?} does not exist", table.namespace()),
+ ));
+ }
+ if !self.table_exists(table).await? {
+ return Err(Error::new(
+ ErrorKind::TableNotFound,
+ format!("Table {table:?} does not exist"),
+ ));
+ }
self.client
.0
@@ -589,6 +642,12 @@ impl Catalog for HmsCatalog {
async fn rename_table(&self, src: &TableIdent, dest: &TableIdent) -> Result<()> {
let src_dbname = validate_namespace(src.namespace())?;
let dest_dbname = validate_namespace(dest.namespace())?;
+ if self.table_exists(dest).await? {
+ return Err(Error::new(
+ ErrorKind::TableAlreadyExists,
+ format!("Destination table {dest:?} already exists"),
+ ));
+ }
let src_tbl_name = src.name.clone();
let dest_tbl_name = dest.name.clone();
diff --git a/crates/catalog/hms/tests/hms_catalog_test.rs b/crates/catalog/hms/tests/hms_catalog_test.rs
index 74c9e52e9..f19cf7bff 100644
--- a/crates/catalog/hms/tests/hms_catalog_test.rs
+++ b/crates/catalog/hms/tests/hms_catalog_test.rs
@@ -24,16 +24,13 @@ use std::collections::HashMap;
use std::sync::Arc;
use iceberg::io::{FileIOBuilder, S3_ACCESS_KEY_ID, S3_ENDPOINT, S3_REGION, S3_SECRET_ACCESS_KEY};
-use iceberg::spec::{NestedField, PrimitiveType, Schema, Type};
-use iceberg::{Catalog, CatalogBuilder, Namespace, NamespaceIdent, TableCreation, TableIdent};
+use iceberg::{Catalog, CatalogBuilder, Namespace, NamespaceIdent};
use iceberg_catalog_hms::{
HMS_CATALOG_PROP_THRIFT_TRANSPORT, HMS_CATALOG_PROP_URI, HMS_CATALOG_PROP_WAREHOUSE,
HmsCatalog, HmsCatalogBuilder, THRIFT_TRANSPORT_BUFFERED,
};
use iceberg_storage_opendal::OpenDalStorageFactory;
-use iceberg_test_utils::{
- cleanup_namespace, get_hms_endpoint, get_minio_endpoint, normalize_test_name_with_parts, set_up,
-};
+use iceberg_test_utils::{get_hms_endpoint, get_minio_endpoint, set_up};
use tokio::time::sleep;
use tracing::info;
@@ -90,230 +87,6 @@ async fn get_catalog() -> HmsCatalog {
.unwrap()
}
-async fn set_test_namespace(catalog: &HmsCatalog, namespace: &NamespaceIdent) -> Result<()> {
- let properties = HashMap::new();
-
- catalog.create_namespace(namespace, properties).await?;
-
- Ok(())
-}
-
-fn set_table_creation(location: Option<String>, name: impl ToString) -> Result<TableCreation> {
- let schema = Schema::builder()
- .with_schema_id(0)
- .with_fields(vec![
- NestedField::required(1, "foo", Type::Primitive(PrimitiveType::Int)).into(),
- NestedField::required(2, "bar", Type::Primitive(PrimitiveType::String)).into(),
- ])
- .build()?;
-
- let builder = TableCreation::builder()
- .name(name.to_string())
- .properties(HashMap::new())
- .location_opt(location)
- .schema(schema);
-
- Ok(builder.build())
-}
-
-#[tokio::test]
-async fn test_rename_table() -> Result<()> {
- let catalog = get_catalog().await;
- let creation: TableCreation = set_table_creation(None, "my_table")?;
- // Use unique namespace to avoid conflicts
- let namespace = Namespace::new(NamespaceIdent::new(normalize_test_name_with_parts!(
- "test_rename_table"
- )));
- // Clean up from any previous test runs
- cleanup_namespace(&catalog, namespace.name()).await;
- set_test_namespace(&catalog, namespace.name()).await?;
-
- let table: iceberg::table::Table = catalog.create_table(namespace.name(), creation).await?;
-
- let dest = TableIdent::new(namespace.name().clone(), "my_table_rename".to_string());
-
- catalog.rename_table(table.identifier(), &dest).await?;
-
- let result = catalog.table_exists(&dest).await?;
-
- assert!(result);
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_table_exists() -> Result<()> {
- let catalog = get_catalog().await;
- let creation = set_table_creation(None, "my_table")?;
- // Use unique namespace to avoid conflicts
- let namespace = Namespace::new(NamespaceIdent::new(normalize_test_name_with_parts!(
- "test_table_exists"
- )));
- cleanup_namespace(&catalog, namespace.name()).await;
- set_test_namespace(&catalog, namespace.name()).await?;
-
- let table = catalog.create_table(namespace.name(), creation).await?;
-
- let result = catalog.table_exists(table.identifier()).await?;
-
- assert!(result);
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_drop_table() -> Result<()> {
- let catalog = get_catalog().await;
- let creation = set_table_creation(None, "my_table")?;
- // Use unique namespace to avoid conflicts
- let namespace = Namespace::new(NamespaceIdent::new(normalize_test_name_with_parts!(
- "test_drop_table"
- )));
- cleanup_namespace(&catalog, namespace.name()).await;
- set_test_namespace(&catalog, namespace.name()).await?;
-
- let table = catalog.create_table(namespace.name(), creation).await?;
-
- catalog.drop_table(table.identifier()).await?;
-
- let result = catalog.table_exists(table.identifier()).await?;
-
- assert!(!result);
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_load_table() -> Result<()> {
- let catalog = get_catalog().await;
- let creation = set_table_creation(None, "my_table")?;
- // Use unique namespace to avoid conflicts
- let namespace = Namespace::new(NamespaceIdent::new(normalize_test_name_with_parts!(
- "test_load_table"
- )));
- cleanup_namespace(&catalog, namespace.name()).await;
- set_test_namespace(&catalog, namespace.name()).await?;
-
- let expected = catalog.create_table(namespace.name(), creation).await?;
-
- let result = catalog
- .load_table(&TableIdent::new(
- namespace.name().clone(),
- "my_table".to_string(),
- ))
- .await?;
-
- assert_eq!(result.identifier(), expected.identifier());
- assert_eq!(result.metadata_location(), expected.metadata_location());
- assert_eq!(result.metadata(), expected.metadata());
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_create_table() -> Result<()> {
- let catalog = get_catalog().await;
- // inject custom location, ignore the namespace prefix
- let creation = set_table_creation(Some("s3a://warehouse/hive".into()), "my_table")?;
- // Use unique namespace to avoid conflicts
- let namespace = Namespace::new(NamespaceIdent::new(normalize_test_name_with_parts!(
- "test_create_table"
- )));
- cleanup_namespace(&catalog, namespace.name()).await;
- set_test_namespace(&catalog, namespace.name()).await?;
-
- let result = catalog.create_table(namespace.name(), creation).await?;
-
- assert_eq!(result.identifier().name(), "my_table");
- assert!(
- result
- .metadata_location()
- .is_some_and(|location| location.starts_with("s3a://warehouse/hive/metadata/00000-"))
- );
- assert!(
- catalog
- .file_io()
- .exists("s3a://warehouse/hive/metadata/")
- .await?
- );
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_list_tables() -> Result<()> {
- let catalog = get_catalog().await;
- // Use unique namespace to avoid conflicts
- let ns = Namespace::new(NamespaceIdent::new(normalize_test_name_with_parts!(
- "test_list_tables"
- )));
- // Clean up and create namespace, then verify it's empty
- cleanup_namespace(&catalog, ns.name()).await;
- set_test_namespace(&catalog, ns.name()).await?;
- let result = catalog.list_tables(ns.name()).await?;
-
- assert_eq!(result, vec![]);
-
- let creation = set_table_creation(None, "my_table")?;
- catalog.create_table(ns.name(), creation).await?;
- let result = catalog.list_tables(ns.name()).await?;
-
- assert_eq!(result, vec![TableIdent::new(
- ns.name().clone(),
- "my_table".to_string()
- )]);
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_list_namespace() -> Result<()> {
- let catalog = get_catalog().await;
-
- let result_no_parent = catalog.list_namespaces(None).await?;
-
- let result_with_parent = catalog
- .list_namespaces(Some(&NamespaceIdent::new("parent".into())))
- .await?;
-
- assert!(result_no_parent.contains(&NamespaceIdent::new("default".into())));
- assert!(result_with_parent.is_empty());
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_create_namespace() -> Result<()> {
- let catalog = get_catalog().await;
-
- let properties = HashMap::from([
- ("comment".to_string(), "my_description".to_string()),
- ("location".to_string(), "my_location".to_string()),
- (
- "hive.metastore.database.owner".to_string(),
- "apache".to_string(),
- ),
- (
- "hive.metastore.database.owner-type".to_string(),
- "user".to_string(),
- ),
- ("key1".to_string(), "value1".to_string()),
- ]);
-
- // Use unique namespace to avoid conflicts
- let ns = Namespace::with_properties(
- NamespaceIdent::new(normalize_test_name_with_parts!("test_create_namespace")),
- properties.clone(),
- );
- cleanup_namespace(&catalog, ns.name()).await;
-
- let result = catalog.create_namespace(ns.name(), properties).await?;
-
- assert_eq!(result, ns);
-
- Ok(())
-}
-
#[tokio::test]
async fn test_get_default_namespace() -> Result<()> {
let catalog = get_catalog().await;
@@ -340,68 +113,3 @@ async fn test_get_default_namespace() -> Result<()> {
Ok(())
}
-
-#[tokio::test]
-async fn test_namespace_exists() -> Result<()> {
- let catalog = get_catalog().await;
-
- let ns_exists = Namespace::new(NamespaceIdent::new("default".into()));
- // Use unique namespace to ensure it doesn't exist
- let ns_not_exists = Namespace::new(NamespaceIdent::new(normalize_test_name_with_parts!(
- "test_namespace_exists"
- )));
- cleanup_namespace(&catalog, ns_not_exists.name()).await;
-
- let result_exists = catalog.namespace_exists(ns_exists.name()).await?;
- let result_not_exists = catalog.namespace_exists(ns_not_exists.name()).await?;
-
- assert!(result_exists);
- assert!(!result_not_exists);
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_update_namespace() -> Result<()> {
- let catalog = get_catalog().await;
-
- // Use unique namespace to avoid conflicts
- let ns = NamespaceIdent::new(normalize_test_name_with_parts!("test_update_namespace"));
- cleanup_namespace(&catalog, &ns).await;
- set_test_namespace(&catalog, &ns).await?;
- let properties = HashMap::from([("comment".to_string(), "my_update".to_string())]);
-
- catalog.update_namespace(&ns, properties).await?;
-
- let db = catalog.get_namespace(&ns).await?;
-
- assert_eq!(
- db.properties().get("comment"),
- Some(&"my_update".to_string())
- );
-
- Ok(())
-}
-
-#[tokio::test]
-async fn test_drop_namespace() -> Result<()> {
- let catalog = get_catalog().await;
-
- // Use unique namespace to avoid conflicts
- let ns = Namespace::new(NamespaceIdent::new(normalize_test_name_with_parts!(
- "test_drop_namespace"
- )));
- cleanup_namespace(&catalog, ns.name()).await;
-
- catalog.create_namespace(ns.name(), HashMap::new()).await?;
-
- let result = catalog.namespace_exists(ns.name()).await?;
- assert!(result);
-
- catalog.drop_namespace(ns.name()).await?;
-
- let result = catalog.namespace_exists(ns.name()).await?;
- assert!(!result);
-
- Ok(())
-}
diff --git a/crates/catalog/loader/Cargo.toml b/crates/catalog/loader/Cargo.toml
index d4b925fb9..20daf84f5 100644
--- a/crates/catalog/loader/Cargo.toml
+++ b/crates/catalog/loader/Cargo.toml
@@ -39,5 +39,9 @@ tokio = { workspace = true }
async-trait = { workspace = true }
[dev-dependencies]
+iceberg_test_utils = { path = "../../test_utils", features = ["tests"] }
+iceberg-storage-opendal = { workspace = true }
+reqwest = { workspace = true }
+rstest = { workspace = true }
sqlx = { workspace = true, features = ["runtime-tokio", "sqlite", "migrate"] }
tempfile = { workspace = true }
diff --git a/crates/catalog/loader/tests/common/mod.rs b/crates/catalog/loader/tests/common/mod.rs
new file mode 100644
index 000000000..90b72df8a
--- /dev/null
+++ b/crates/catalog/loader/tests/common/mod.rs
@@ -0,0 +1,342 @@
+// 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.
+
+//! Shared helpers for catalog integration suites.
+
+#![allow(dead_code)]
+
+use std::collections::HashMap;
+use std::fmt;
+use std::sync::Arc;
+
+use iceberg::io::{
+ FileIOBuilder, LocalFsStorageFactory, S3_ACCESS_KEY_ID, S3_ENDPOINT, S3_REGION,
+ S3_SECRET_ACCESS_KEY,
+};
+use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
+use iceberg::spec::{NestedField, PrimitiveType, Schema, Type};
+use iceberg::{Catalog, CatalogBuilder, NamespaceIdent, TableCreation};
+use iceberg_catalog_glue::{
+ AWS_ACCESS_KEY_ID, AWS_REGION_NAME, AWS_SECRET_ACCESS_KEY, GLUE_CATALOG_PROP_URI,
+ GLUE_CATALOG_PROP_WAREHOUSE, GlueCatalog, GlueCatalogBuilder,
+};
+use iceberg_catalog_hms::{
+ HMS_CATALOG_PROP_THRIFT_TRANSPORT, HMS_CATALOG_PROP_URI, HMS_CATALOG_PROP_WAREHOUSE,
+ HmsCatalog, HmsCatalogBuilder, THRIFT_TRANSPORT_BUFFERED,
+};
+use iceberg_catalog_rest::{REST_CATALOG_PROP_URI, RestCatalog, RestCatalogBuilder};
+use iceberg_catalog_s3tables::{
+ S3TABLES_CATALOG_PROP_ENDPOINT_URL, S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN,
+ S3TablesCatalogBuilder,
+};
+use iceberg_catalog_sql::{
+ SQL_CATALOG_PROP_BIND_STYLE, SQL_CATALOG_PROP_URI, SQL_CATALOG_PROP_WAREHOUSE, SqlBindStyle,
+ SqlCatalogBuilder,
+};
+use iceberg_storage_opendal::OpenDalStorageFactory;
+use iceberg_test_utils::{
+ get_glue_endpoint, get_hms_endpoint, get_minio_endpoint, get_rest_catalog_endpoint, set_up,
+};
+use sqlx::migrate::MigrateDatabase;
+use tempfile::TempDir;
+use tokio::time::sleep;
+
+#[derive(Debug, Clone, Copy)]
+pub enum CatalogKind {
+ Rest,
+ Glue,
+ Hms,
+ Sql,
+ S3Tables,
+ Memory,
+}
+
+impl fmt::Display for CatalogKind {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ let name = match self {
+ CatalogKind::Rest => "rest_catalog",
+ CatalogKind::Glue => "glue_catalog",
+ CatalogKind::Hms => "hms_catalog",
+ CatalogKind::Sql => "sql_catalog",
+ CatalogKind::S3Tables => "s3tables_catalog",
+ CatalogKind::Memory => "memory_catalog",
+ };
+ f.write_str(name)
+ }
+}
+
+pub struct CatalogHarness {
+ pub catalog: Arc<dyn Catalog>,
+ pub label: &'static str,
+ _tempdirs: Vec<TempDir>,
+}
+
+// Shared setup for each catalog implementation so the suites exercise
+// the same behavior against all backends.
+pub async fn load_catalog(kind: CatalogKind) -> Option<CatalogHarness> {
+ set_up();
+ match kind {
+ CatalogKind::Rest => Some(CatalogHarness {
+ catalog: Arc::new(rest_catalog().await) as Arc<dyn Catalog>,
+ label: "rest",
+ _tempdirs: Vec::new(),
+ }),
+ CatalogKind::Glue => Some(CatalogHarness {
+ catalog: Arc::new(glue_catalog().await) as Arc<dyn Catalog>,
+ label: "glue",
+ _tempdirs: Vec::new(),
+ }),
+ CatalogKind::Hms => Some(CatalogHarness {
+ catalog: Arc::new(hms_catalog().await) as Arc<dyn Catalog>,
+ label: "hms",
+ _tempdirs: Vec::new(),
+ }),
+ CatalogKind::Sql => {
+ let warehouse_dir = TempDir::new().unwrap();
+ let db_dir = TempDir::new().unwrap();
+ let db_path = db_dir.path().join("catalog.db");
+ let db_uri = format!("sqlite:{}", db_path.to_str().unwrap());
+ sqlx::Sqlite::create_database(&db_uri).await.unwrap();
+
+ let catalog = SqlCatalogBuilder::default()
+ .with_storage_factory(Arc::new(LocalFsStorageFactory))
+ .load(
+ "sql",
+ HashMap::from([
+ (SQL_CATALOG_PROP_URI.to_string(), db_uri),
+ (
+ SQL_CATALOG_PROP_WAREHOUSE.to_string(),
+ warehouse_dir.path().to_str().unwrap().to_string(),
+ ),
+ (
+ SQL_CATALOG_PROP_BIND_STYLE.to_string(),
+ SqlBindStyle::QMark.to_string(),
+ ),
+ ]),
+ )
+ .await
+ .unwrap();
+
+ Some(CatalogHarness {
+ catalog: Arc::new(catalog) as Arc<dyn Catalog>,
+ label: "sql",
+ _tempdirs: vec![warehouse_dir, db_dir],
+ })
+ }
+ CatalogKind::S3Tables => {
+ let table_bucket_arn = match std::env::var("TABLE_BUCKET_ARN").ok() {
+ Some(value) => value,
+ None => return None,
+ };
+
+ let mut props = HashMap::from([(
+ S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN.to_string(),
+ table_bucket_arn,
+ )]);
+
+ if let Ok(endpoint_url) = std::env::var("S3TABLES_ENDPOINT_URL") {
+ props.insert(S3TABLES_CATALOG_PROP_ENDPOINT_URL.to_string(), endpoint_url);
+ }
+
+ let catalog = S3TablesCatalogBuilder::default()
+ .load("s3tables", props)
+ .await
+ .unwrap();
+
+ Some(CatalogHarness {
+ catalog: Arc::new(catalog) as Arc<dyn Catalog>,
+ label: "s3tables",
+ _tempdirs: Vec::new(),
+ })
+ }
+ CatalogKind::Memory => {
+ let warehouse_dir = TempDir::new().unwrap();
+ let props = HashMap::from([(
+ MEMORY_CATALOG_WAREHOUSE.to_string(),
+ warehouse_dir.path().to_str().unwrap().to_string(),
+ )]);
+ let catalog = MemoryCatalogBuilder::default()
+ .load("memory", props)
+ .await
+ .unwrap();
+
+ Some(CatalogHarness {
+ catalog: Arc::new(catalog) as Arc<dyn Catalog>,
+ label: "memory",
+ _tempdirs: vec![warehouse_dir],
+ })
+ }
+ }
+}
+
+// Catalog-specific setup is intentionally isolated here so the suites
+// remain implementation-agnostic.
+async fn rest_catalog() -> RestCatalog {
+ let rest_endpoint = get_rest_catalog_endpoint();
+
+ let client = reqwest::Client::new();
+ let mut retries = 0;
+ while retries < 30 {
+ if client
+ .get(format!("{rest_endpoint}/v1/config"))
+ .send()
+ .await
+ .map(|resp| resp.status().is_success())
+ .unwrap_or(false)
+ {
+ break;
+ }
+ sleep(std::time::Duration::from_millis(1000)).await;
+ retries += 1;
+ }
+
+ RestCatalogBuilder::default()
+ .with_storage_factory(Arc::new(LocalFsStorageFactory))
+ .load(
+ "rest",
+ HashMap::from([(REST_CATALOG_PROP_URI.to_string(), rest_endpoint)]),
+ )
+ .await
+ .unwrap()
+}
+
+async fn glue_catalog() -> GlueCatalog {
+ let glue_endpoint = get_glue_endpoint();
+ let minio_endpoint = get_minio_endpoint();
+
+ let props = HashMap::from([
+ (AWS_ACCESS_KEY_ID.to_string(), "my_access_id".to_string()),
+ (
+ AWS_SECRET_ACCESS_KEY.to_string(),
+ "my_secret_key".to_string(),
+ ),
+ (AWS_REGION_NAME.to_string(), "us-east-1".to_string()),
+ (S3_ENDPOINT.to_string(), minio_endpoint),
+ (S3_ACCESS_KEY_ID.to_string(), "admin".to_string()),
+ (S3_SECRET_ACCESS_KEY.to_string(), "password".to_string()),
+ (S3_REGION.to_string(), "us-east-1".to_string()),
+ ]);
+
+ let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 {
+ configured_scheme: "s3a".to_string(),
+ customized_credential_load: None,
+ }))
+ .with_props(props.clone())
+ .build();
+
+ let mut retries = 0;
+ while retries < 30 {
+ if file_io.exists("s3a://warehouse/").await.unwrap_or(false) {
+ break;
+ }
+ sleep(std::time::Duration::from_millis(1000)).await;
+ retries += 1;
+ }
+
+ let mut glue_props = HashMap::from([
+ (GLUE_CATALOG_PROP_URI.to_string(), glue_endpoint),
+ (
+ GLUE_CATALOG_PROP_WAREHOUSE.to_string(),
+ "s3a://warehouse/hive".to_string(),
+ ),
+ ]);
+ glue_props.extend(props);
+
+ GlueCatalogBuilder::default()
+ .load("glue", glue_props)
+ .await
+ .unwrap()
+}
+
+async fn hms_catalog() -> HmsCatalog {
+ let hms_endpoint = get_hms_endpoint();
+ let minio_endpoint = get_minio_endpoint();
+
+ let props = HashMap::from([
+ (HMS_CATALOG_PROP_URI.to_string(), hms_endpoint),
+ (
+ HMS_CATALOG_PROP_THRIFT_TRANSPORT.to_string(),
+ THRIFT_TRANSPORT_BUFFERED.to_string(),
+ ),
+ (
+ HMS_CATALOG_PROP_WAREHOUSE.to_string(),
+ "s3a://warehouse/hive".to_string(),
+ ),
+ (S3_ENDPOINT.to_string(), minio_endpoint),
+ (S3_ACCESS_KEY_ID.to_string(), "admin".to_string()),
+ (S3_SECRET_ACCESS_KEY.to_string(), "password".to_string()),
+ (S3_REGION.to_string(), "us-east-1".to_string()),
+ ]);
+
+ let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 {
+ configured_scheme: "s3a".to_string(),
+ customized_credential_load: None,
+ }))
+ .with_props(props.clone())
+ .build();
+
+ let mut retries = 0;
+ while retries < 30 {
+ if file_io.exists("s3a://warehouse/").await.unwrap_or(false) {
+ break;
+ }
+ sleep(std::time::Duration::from_millis(1000)).await;
+ retries += 1;
+ }
+
+ HmsCatalogBuilder::default()
+ .with_storage_factory(Arc::new(OpenDalStorageFactory::S3 {
+ configured_scheme: "s3a".to_string(),
+ customized_credential_load: None,
+ }))
+ .load("hms", props)
+ .await
+ .unwrap()
+}
+
+// Common table schema used across suites to validate shared behavior.
+pub fn table_creation(name: impl ToString) -> TableCreation {
+ let schema = Schema::builder()
+ .with_schema_id(0)
+ .with_fields(vec![
+ NestedField::required(1, "foo", Type::Primitive(PrimitiveType::Int)).into(),
+ NestedField::required(2, "bar", Type::Primitive(PrimitiveType::String)).into(),
+ ])
+ .build()
+ .unwrap();
+
+ TableCreation::builder()
+ .name(name.to_string())
+ .properties(HashMap::new())
+ .schema(schema)
+ .build()
+}
+
+pub fn assert_map_contains(expected: &HashMap<String, String>, actual: &HashMap<String, String>) {
+ for (key, value) in expected {
+ assert_eq!(actual.get(key), Some(value));
+ }
+}
+
+pub async fn cleanup_namespace_dyn(catalog: &dyn Catalog, namespace: &NamespaceIdent) {
+ if let Ok(tables) = catalog.list_tables(namespace).await {
+ for table in tables {
+ let _ = catalog.drop_table(&table).await;
+ }
+ }
+ let _ = catalog.drop_namespace(namespace).await;
+}
diff --git a/crates/catalog/loader/tests/namespace_suite.rs b/crates/catalog/loader/tests/namespace_suite.rs
new file mode 100644
index 000000000..024e8a62b
--- /dev/null
+++ b/crates/catalog/loader/tests/namespace_suite.rs
@@ -0,0 +1,355 @@
+// 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.
+
+//! Common namespace behavior across catalogs.
+//!
+//! These tests assume Docker containers are started externally via `make docker-up`.
+
+mod common;
+
+use std::collections::HashMap;
+
+use common::{CatalogKind, assert_map_contains, cleanup_namespace_dyn, load_catalog};
+use iceberg::{ErrorKind, NamespaceIdent, Result};
+use iceberg_test_utils::normalize_test_name_with_parts;
+use rstest::rstest;
+
+// Common behavior: querying a missing namespace should error.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_namespace_missing_returns_error(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_namespace_missing_returns_error",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+
+ let err = catalog.get_namespace(&namespace).await.unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::NamespaceNotFound);
+
+ Ok(())
+}
+
+// Common behavior: namespace lifecycle CRUD.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_namespace_lifecycle(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_namespace_lifecycle",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+
+ assert!(!catalog.namespace_exists(&namespace).await?);
+
+ let props = HashMap::from([
+ ("owner".to_string(), "rust".to_string()),
+ ("purpose".to_string(), "catalog_suite".to_string()),
+ ]);
+ let created = catalog.create_namespace(&namespace, props.clone()).await?;
+ assert_eq!(created.name(), &namespace);
+ assert_map_contains(&props, created.properties());
+
+ let fetched = catalog.get_namespace(&namespace).await?;
+ assert_eq!(fetched.name(), &namespace);
+ assert_map_contains(&props, fetched.properties());
+
+ let namespaces = catalog.list_namespaces(None).await?;
+ assert!(namespaces.contains(&namespace));
+
+ catalog.drop_namespace(&namespace).await?;
+ assert!(!catalog.namespace_exists(&namespace).await?);
+
+ Ok(())
+}
+
+// Common behavior: update_namespace persists changes when supported.
+#[rstest]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_update_namespace_supported(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_update_namespace_supported",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+ catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+ let updated_props = HashMap::from([("owner".to_string(), "updated".to_string())]);
+ catalog
+ .update_namespace(&namespace, updated_props.clone())
+ .await?;
+
+ let updated = catalog.get_namespace(&namespace).await?;
+ assert_map_contains(&updated_props, updated.properties());
+
+ Ok(())
+}
+
+// Common behavior: update_namespace returns FeatureUnsupported when not implemented.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[tokio::test]
+async fn test_catalog_update_namespace_unsupported(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_update_namespace_unsupported",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+ catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+ let err = catalog
+ .update_namespace(
+ &namespace,
+ HashMap::from([("key".to_string(), "value".to_string())]),
+ )
+ .await
+ .unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::FeatureUnsupported);
+
+ Ok(())
+}
+
+// Common behavior: listing namespaces under a parent returns its children.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_namespace_listing_with_parent(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let parent_name =
+ normalize_test_name_with_parts!("catalog_namespace_listing_with_parent", harness.label);
+ let parent = NamespaceIdent::new(parent_name.clone());
+ let child1 = NamespaceIdent::from_strs([&parent_name, "child1"]).unwrap();
+ let child2 = NamespaceIdent::from_strs([&parent_name, "child2"]).unwrap();
+
+ cleanup_namespace_dyn(catalog.as_ref(), &child1).await;
+ cleanup_namespace_dyn(catalog.as_ref(), &child2).await;
+ cleanup_namespace_dyn(catalog.as_ref(), &parent).await;
+
+ catalog.create_namespace(&parent, HashMap::new()).await?;
+
+ catalog.create_namespace(&child1, HashMap::new()).await?;
+ catalog.create_namespace(&child2, HashMap::new()).await?;
+
+ let top_level = catalog.list_namespaces(None).await?;
+ assert!(top_level.contains(&parent));
+
+ let children = catalog.list_namespaces(Some(&parent)).await?;
+ assert!(children.contains(&child1));
+ assert!(children.contains(&child2));
+
+ Ok(())
+}
+
+// Common behavior: hierarchical namespaces are rejected when unsupported.
+#[rstest]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[tokio::test]
+async fn test_catalog_namespace_listing_with_parent_unsupported(
+ #[case] kind: CatalogKind,
+) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let parent_name = normalize_test_name_with_parts!(
+ "catalog_namespace_listing_with_parent_unsupported",
+ harness.label
+ );
+ let parent = NamespaceIdent::new(parent_name.clone());
+ let child = NamespaceIdent::from_strs([&parent_name, "child"]).unwrap();
+
+ cleanup_namespace_dyn(catalog.as_ref(), &child).await;
+ cleanup_namespace_dyn(catalog.as_ref(), &parent).await;
+
+ catalog.create_namespace(&parent, HashMap::new()).await?;
+
+ let err = catalog
+ .create_namespace(&child, HashMap::new())
+ .await
+ .unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+
+ Ok(())
+}
+
+// Common behavior: listing top-level namespaces includes created namespaces.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_list_namespaces_contains_created(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let ns_one = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_list_namespaces_contains_created",
+ harness.label,
+ "one"
+ ));
+ let ns_two = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_list_namespaces_contains_created",
+ harness.label,
+ "two"
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &ns_one).await;
+ cleanup_namespace_dyn(catalog.as_ref(), &ns_two).await;
+
+ catalog.create_namespace(&ns_one, HashMap::new()).await?;
+ catalog.create_namespace(&ns_two, HashMap::new()).await?;
+
+ let namespaces = catalog.list_namespaces(None).await?;
+ assert!(namespaces.contains(&ns_one));
+ assert!(namespaces.contains(&ns_two));
+
+ Ok(())
+}
+
+// Common behavior: creating an existing namespace should error.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_create_namespace_duplicate_fails(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_create_namespace_duplicate_fails",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+ catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+ let err = catalog
+ .create_namespace(&namespace, HashMap::new())
+ .await
+ .unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::NamespaceAlreadyExists);
+ Ok(())
+}
+
+// Common behavior: update on a missing namespace should return NamespaceNotFound.
+#[rstest]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_update_namespace_missing_errors(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_update_namespace_missing_errors",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+
+ let err = catalog
+ .update_namespace(
+ &namespace,
+ HashMap::from([("key".to_string(), "value".to_string())]),
+ )
+ .await
+ .unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::NamespaceNotFound);
+
+ Ok(())
+}
+
+// Common behavior: dropping a missing namespace should error.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_drop_namespace_missing_errors(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_drop_namespace_missing_errors",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+
+ let err = catalog.drop_namespace(&namespace).await.unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::NamespaceNotFound);
+ Ok(())
+}
diff --git a/crates/catalog/loader/tests/table_register_suite.rs b/crates/catalog/loader/tests/table_register_suite.rs
new file mode 100644
index 000000000..b43054b83
--- /dev/null
+++ b/crates/catalog/loader/tests/table_register_suite.rs
@@ -0,0 +1,167 @@
+// 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.
+
+//! Common register-table behavior across catalogs.
+//!
+//! These tests assume Docker containers are started externally via `make docker-up`.
+
+mod common;
+
+use std::collections::HashMap;
+
+use common::{CatalogKind, cleanup_namespace_dyn, load_catalog, table_creation};
+use iceberg::{ErrorKind, NamespaceIdent, Result, TableIdent};
+use iceberg_test_utils::normalize_test_name_with_parts;
+use rstest::rstest;
+
+// Common behavior: register_table rehydrates a dropped table.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_register_table_roundtrip(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_register_table_roundtrip",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+ catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+ let table = catalog
+ .create_table(
+ &namespace,
+ table_creation(normalize_test_name_with_parts!(
+ "catalog_register_table_roundtrip",
+ harness.label,
+ "table"
+ )),
+ )
+ .await?;
+ let table_ident = table.identifier().clone();
+ let metadata_location = table
+ .metadata_location()
+ .ok_or_else(|| iceberg::Error::new(ErrorKind::Unexpected, "Missing metadata location"))?
+ .to_string();
+
+ catalog.drop_table(&table_ident).await?;
+
+ let registered = catalog
+ .register_table(&table_ident, metadata_location.clone())
+ .await?;
+ assert_eq!(registered.identifier(), &table_ident);
+ assert_eq!(
+ registered.metadata_location(),
+ Some(metadata_location.as_str())
+ );
+
+ Ok(())
+}
+
+// HMS and S3Tables do not support register_table yet.
+#[rstest]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[tokio::test]
+async fn test_catalog_register_table_unsupported(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_register_table_unsupported",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+ catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+ let table = catalog
+ .create_table(
+ &namespace,
+ table_creation(normalize_test_name_with_parts!(
+ "catalog_register_table_unsupported",
+ harness.label,
+ "table"
+ )),
+ )
+ .await?;
+ let table_ident = table.identifier().clone();
+ let metadata_location = table
+ .metadata_location()
+ .ok_or_else(|| iceberg::Error::new(ErrorKind::Unexpected, "Missing metadata location"))?
+ .to_string();
+
+ let err = catalog
+ .register_table(&table_ident, metadata_location)
+ .await
+ .unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::FeatureUnsupported);
+
+ Ok(())
+}
+
+// Common behavior: registering a table with an existing name should error.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_register_table_conflict_errors(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_register_table_conflict_errors",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+ catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+ let table_ident = TableIdent::new(
+ namespace.clone(),
+ normalize_test_name_with_parts!(
+ "catalog_register_table_conflict_errors",
+ harness.label,
+ "table"
+ ),
+ );
+ let table = catalog
+ .create_table(&namespace, table_creation(table_ident.name.clone()))
+ .await?;
+ let metadata_location = table
+ .metadata_location()
+ .ok_or_else(|| iceberg::Error::new(ErrorKind::Unexpected, "Missing metadata location"))?
+ .to_string();
+
+ assert!(
+ catalog
+ .register_table(&table_ident, metadata_location)
+ .await
+ .is_err()
+ );
+ Ok(())
+}
diff --git a/crates/catalog/loader/tests/table_rename_suite.rs b/crates/catalog/loader/tests/table_rename_suite.rs
new file mode 100644
index 000000000..757d1cd04
--- /dev/null
+++ b/crates/catalog/loader/tests/table_rename_suite.rs
@@ -0,0 +1,163 @@
+// 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.
+
+//! Common rename behavior across catalogs.
+//!
+//! These tests assume Docker containers are started externally via `make docker-up`.
+
+mod common;
+
+use std::collections::HashMap;
+
+use common::{CatalogKind, cleanup_namespace_dyn, load_catalog, table_creation};
+use iceberg::{NamespaceIdent, Result, TableIdent};
+use iceberg_test_utils::normalize_test_name_with_parts;
+use rstest::rstest;
+
+// Common behavior: renaming across namespaces moves the table.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_rename_table_across_namespaces(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let src_namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_rename_table_across_namespaces",
+ harness.label,
+ "src"
+ ));
+ let dst_namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_rename_table_across_namespaces",
+ harness.label,
+ "dst"
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &src_namespace).await;
+ cleanup_namespace_dyn(catalog.as_ref(), &dst_namespace).await;
+ catalog
+ .create_namespace(&src_namespace, HashMap::new())
+ .await?;
+ catalog
+ .create_namespace(&dst_namespace, HashMap::new())
+ .await?;
+
+ let table = catalog
+ .create_table(
+ &src_namespace,
+ table_creation(normalize_test_name_with_parts!(
+ "catalog_rename_table_across_namespaces",
+ harness.label,
+ "table"
+ )),
+ )
+ .await?;
+ let src_ident = table.identifier().clone();
+ let dst_ident = TableIdent::new(dst_namespace.clone(), src_ident.name.clone());
+
+ catalog.rename_table(&src_ident, &dst_ident).await?;
+ assert!(catalog.table_exists(&dst_ident).await?);
+ assert!(!catalog.table_exists(&src_ident).await?);
+
+ Ok(())
+}
+
+// Common behavior: renaming a missing table should error.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_rename_table_missing_source_errors(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_rename_table_missing_source_errors",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+ catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+ let src_ident = TableIdent::new(namespace.clone(), "missing".to_string());
+ let dst_ident = TableIdent::new(namespace.clone(), "dest".to_string());
+
+ assert!(catalog.rename_table(&src_ident, &dst_ident).await.is_err());
+ Ok(())
+}
+
+// Common behavior: renaming to an existing destination should error.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_rename_table_dest_exists_errors(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_rename_table_dest_exists_errors",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+ catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+ let src = catalog
+ .create_table(
+ &namespace,
+ table_creation(normalize_test_name_with_parts!(
+ "catalog_rename_table_dest_exists_errors",
+ harness.label,
+ "src"
+ )),
+ )
+ .await?
+ .identifier()
+ .clone();
+ let dst = catalog
+ .create_table(
+ &namespace,
+ table_creation(normalize_test_name_with_parts!(
+ "catalog_rename_table_dest_exists_errors",
+ harness.label,
+ "dst"
+ )),
+ )
+ .await?
+ .identifier()
+ .clone();
+
+ assert!(catalog.rename_table(&src, &dst).await.is_err());
+ Ok(())
+}
diff --git a/crates/catalog/loader/tests/table_suite.rs b/crates/catalog/loader/tests/table_suite.rs
new file mode 100644
index 000000000..6b7a3a822
--- /dev/null
+++ b/crates/catalog/loader/tests/table_suite.rs
@@ -0,0 +1,276 @@
+// 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.
+
+//! Common table behavior across catalogs.
+//!
+//! These tests assume Docker containers are started externally via `make docker-up`.
+
+mod common;
+
+use std::collections::HashMap;
+
+use common::{CatalogKind, cleanup_namespace_dyn, load_catalog, table_creation};
+use iceberg::transaction::{ApplyTransactionAction, Transaction};
+use iceberg::{ErrorKind, NamespaceIdent, Result, TableIdent};
+use iceberg_test_utils::normalize_test_name_with_parts;
+use rstest::rstest;
+
+// Common behavior: table lifecycle CRUD.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_table_lifecycle(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_table_lifecycle",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+ catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+ let table_name =
+ normalize_test_name_with_parts!("catalog_table_lifecycle", harness.label, "table");
+ let table = catalog
+ .create_table(&namespace, table_creation(table_name))
+ .await?;
+ let ident = table.identifier().clone();
+
+ assert!(catalog.table_exists(&ident).await?);
+ let loaded = catalog.load_table(&ident).await?;
+ assert_eq!(loaded.identifier(), &ident);
+
+ let tables = catalog.list_tables(&namespace).await?;
+ assert!(tables.contains(&ident));
+
+ let dest = TableIdent::new(ident.namespace.clone(), format!("{}_renamed", ident.name));
+ catalog.rename_table(&ident, &dest).await?;
+ assert!(catalog.table_exists(&dest).await?);
+ assert!(!catalog.table_exists(&ident).await?);
+
+ catalog.drop_table(&dest).await?;
+ assert!(!catalog.table_exists(&dest).await?);
+
+ catalog.drop_namespace(&namespace).await?;
+
+ Ok(())
+}
+
+// Common behavior: listing tables for a missing namespace should error.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_list_tables_missing_namespace_errors(
+ #[case] kind: CatalogKind,
+) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_list_tables_missing_namespace_errors",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+
+ assert!(catalog.list_tables(&namespace).await.is_err());
+ Ok(())
+}
+
+// Common behavior: listing tables in an empty namespace returns empty.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_list_tables_empty_namespace(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_list_tables_empty_namespace",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+ catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+ let tables = catalog.list_tables(&namespace).await?;
+ assert!(tables.is_empty());
+
+ Ok(())
+}
+
+// Common behavior: created tables expose the requested schema.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_create_table_schema(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_create_table_schema",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+ catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+ let table_name =
+ normalize_test_name_with_parts!("catalog_create_table_schema", harness.label, "table");
+ let creation = table_creation(table_name);
+ let expected_schema = creation.schema.clone();
+
+ let table = catalog.create_table(&namespace, creation).await?;
+ assert_eq!(table.identifier().namespace, namespace);
+ assert_eq!(table.metadata().current_schema().as_ref(), &expected_schema);
+
+ Ok(())
+}
+
+// Common behavior: updating table properties persists through the catalog.
+// HMS is excluded because update_table is not supported yet.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_update_table_properties(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_update_table_properties",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+ catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+ let table_name =
+ normalize_test_name_with_parts!("catalog_update_table_properties", harness.label, "table");
+ let table = catalog
+ .create_table(&namespace, table_creation(table_name))
+ .await?;
+
+ let tx = Transaction::new(&table);
+ let tx = tx
+ .update_table_properties()
+ .set("test_property".to_string(), "test_value".to_string())
+ .apply(tx)?;
+ let updated = tx.commit(catalog.as_ref()).await?;
+
+ assert_eq!(
+ updated.metadata().properties().get("test_property"),
+ Some(&"test_value".to_string())
+ );
+
+ Ok(())
+}
+
+// Common behavior: update_table_properties is rejected when unsupported.
+#[rstest]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[tokio::test]
+async fn test_catalog_update_table_properties_unsupported(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_update_table_properties_unsupported",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+ catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+ let table_name = normalize_test_name_with_parts!(
+ "catalog_update_table_properties_unsupported",
+ harness.label,
+ "table"
+ );
+ let table = catalog
+ .create_table(&namespace, table_creation(table_name))
+ .await?;
+
+ let tx = Transaction::new(&table);
+ let tx = tx
+ .update_table_properties()
+ .set("test_property".to_string(), "test_value".to_string())
+ .apply(tx)?;
+
+ let err = tx.commit(catalog.as_ref()).await.unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::FeatureUnsupported);
+
+ Ok(())
+}
+
+// Common behavior: dropping a missing table should error.
+#[rstest]
+#[case::rest_catalog(CatalogKind::Rest)]
+#[case::glue_catalog(CatalogKind::Glue)]
+#[case::hms_catalog(CatalogKind::Hms)]
+#[case::sql_catalog(CatalogKind::Sql)]
+#[case::s3tables_catalog(CatalogKind::S3Tables)]
+#[case::memory_catalog(CatalogKind::Memory)]
+#[tokio::test]
+async fn test_catalog_drop_table_missing_errors(#[case] kind: CatalogKind) -> Result<()> {
+ let Some(harness) = load_catalog(kind).await else {
+ return Ok(());
+ };
+ let catalog = harness.catalog;
+ let namespace = NamespaceIdent::new(normalize_test_name_with_parts!(
+ "catalog_drop_table_missing_errors",
+ harness.label
+ ));
+
+ cleanup_namespace_dyn(catalog.as_ref(), &namespace).await;
+ catalog.create_namespace(&namespace, HashMap::new()).await?;
+
+ let table_ident = TableIdent::new(namespace.clone(), "missing".to_string());
+ assert!(catalog.drop_table(&table_ident).await.is_err());
+ Ok(())
+}
diff --git a/crates/catalog/rest/Cargo.toml b/crates/catalog/rest/Cargo.toml
index de72b6c61..40dd70a95 100644
--- a/crates/catalog/rest/Cargo.toml
+++ b/crates/catalog/rest/Cargo.toml
@@ -39,7 +39,6 @@ serde = { workspace = true }
serde_derive = { workspace = true }
serde_json = { workspace = true }
tokio = { workspace = true, features = ["sync"] }
-tracing = { workspace = true }
typed-builder = { workspace = true }
uuid = { workspace = true, features = ["v4"] }
diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs
index 871643b36..3551b0516 100644
--- a/crates/catalog/rest/src/catalog.rs
+++ b/crates/catalog/rest/src/catalog.rs
@@ -504,7 +504,7 @@ impl Catalog for RestCatalog {
}
StatusCode::NOT_FOUND => {
return Err(Error::new(
- ErrorKind::Unexpected,
+ ErrorKind::NamespaceNotFound,
"The parent parameter of the namespace provided does not exist",
));
}
@@ -546,7 +546,7 @@ impl Catalog for RestCatalog {
Ok(Namespace::from(response))
}
StatusCode::CONFLICT => Err(Error::new(
- ErrorKind::Unexpected,
+ ErrorKind::NamespaceAlreadyExists,
"Tried to create a namespace that already exists",
)),
_ => Err(deserialize_unexpected_catalog_error(
@@ -574,7 +574,7 @@ impl Catalog for RestCatalog {
Ok(Namespace::from(response))
}
StatusCode::NOT_FOUND => Err(Error::new(
- ErrorKind::Unexpected,
+ ErrorKind::NamespaceNotFound,
"Tried to get a namespace that does not exist",
)),
_ => Err(deserialize_unexpected_catalog_error(
@@ -630,7 +630,7 @@ impl Catalog for RestCatalog {
match http_response.status() {
StatusCode::NO_CONTENT | StatusCode::OK => Ok(()),
StatusCode::NOT_FOUND => Err(Error::new(
- ErrorKind::Unexpected,
+ ErrorKind::NamespaceNotFound,
"Tried to drop a namespace that does not exist",
)),
_ => Err(deserialize_unexpected_catalog_error(
@@ -670,7 +670,7 @@ impl Catalog for RestCatalog {
}
StatusCode::NOT_FOUND => {
return Err(Error::new(
- ErrorKind::Unexpected,
+ ErrorKind::NamespaceNotFound,
"Tried to list tables of a namespace that does not exist",
));
}
@@ -724,13 +724,13 @@ impl Catalog for RestCatalog {
}
StatusCode::NOT_FOUND => {
return Err(Error::new(
- ErrorKind::Unexpected,
+ ErrorKind::NamespaceNotFound,
"Tried to create a table under a namespace that does not exist",
));
}
StatusCode::CONFLICT => {
return Err(Error::new(
- ErrorKind::Unexpected,
+ ErrorKind::TableAlreadyExists,
"The table already exists",
));
}
@@ -791,7 +791,7 @@ impl Catalog for RestCatalog {
}
StatusCode::NOT_FOUND => {
return Err(Error::new(
- ErrorKind::Unexpected,
+ ErrorKind::TableNotFound,
"Tried to load a table that does not exist",
));
}
@@ -840,7 +840,7 @@ impl Catalog for RestCatalog {
match http_response.status() {
StatusCode::NO_CONTENT | StatusCode::OK => Ok(()),
StatusCode::NOT_FOUND => Err(Error::new(
- ErrorKind::Unexpected,
+ ErrorKind::TableNotFound,
"Tried to drop a table that does not exist",
)),
_ => Err(deserialize_unexpected_catalog_error(
@@ -891,11 +891,11 @@ impl Catalog for RestCatalog {
match http_response.status() {
StatusCode::NO_CONTENT | StatusCode::OK => Ok(()),
StatusCode::NOT_FOUND => Err(Error::new(
- ErrorKind::Unexpected,
+ ErrorKind::TableNotFound,
"Tried to rename a table that does not exist (is the namespace correct?)",
)),
StatusCode::CONFLICT => Err(Error::new(
- ErrorKind::Unexpected,
+ ErrorKind::TableAlreadyExists,
"Tried to rename a table to a name that already exists",
)),
_ => Err(deserialize_unexpected_catalog_error(
diff --git a/crates/catalog/rest/tests/rest_catalog_test.rs b/crates/catalog/rest/tests/rest_catalog_test.rs
deleted file mode 100644
index 98d8ee9b7..000000000
--- a/crates/catalog/rest/tests/rest_catalog_test.rs
+++ /dev/null
@@ -1,497 +0,0 @@
-// 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.
-
-//! Integration tests for rest catalog.
-//!
-//! These tests assume Docker containers are started externally via `make docker-up`.
-//! Each test uses unique namespaces based on module path to avoid conflicts.
-
-use std::collections::HashMap;
-use std::sync::Arc;
-
-use iceberg::io::LocalFsStorageFactory;
-use iceberg::spec::{FormatVersion, NestedField, PrimitiveType, Schema, Type};
-use iceberg::transaction::{ApplyTransactionAction, Transaction};
-use iceberg::{Catalog, CatalogBuilder, Namespace, NamespaceIdent, TableCreation, TableIdent};
-use iceberg_catalog_rest::{REST_CATALOG_PROP_URI, RestCatalog, RestCatalogBuilder};
-use iceberg_test_utils::{
- cleanup_namespace, get_rest_catalog_endpoint, normalize_test_name_with_parts, set_up,
-};
-use tokio::time::sleep;
-use tracing::info;
-
-async fn get_catalog() -> RestCatalog {
- set_up();
-
- let rest_endpoint = get_rest_catalog_endpoint();
-
- // Wait for catalog to be ready
- let client = reqwest::Client::new();
- let mut retries = 0;
- while retries < 30 {
- match client
- .get(format!("{rest_endpoint}/v1/config"))
- .send()
- .await
- {
- Ok(resp) if resp.status().is_success() => {
- info!("REST catalog is ready at {}", rest_endpoint);
- break;
- }
- _ => {
- info!(
- "Waiting for REST catalog to be ready... (attempt {})",
- retries + 1
- );
- sleep(std::time::Duration::from_millis(1000)).await;
- retries += 1;
- }
- }
- }
-
- RestCatalogBuilder::default()
- .with_storage_factory(Arc::new(LocalFsStorageFactory))
- .load(
- "rest",
- HashMap::from([(REST_CATALOG_PROP_URI.to_string(), rest_endpoint)]),
- )
- .await
- .unwrap()
-}
-
-#[tokio::test]
-async fn test_get_non_exist_namespace() {
- let catalog = get_catalog().await;
-
- // Use unique namespace name to ensure it doesn't exist
- let ns_ident = NamespaceIdent::new(normalize_test_name_with_parts!(
- "test_get_non_exist_namespace"
- ));
- // Clean up from any previous test runs
- cleanup_namespace(&catalog, &ns_ident).await;
-
- let result = catalog.get_namespace(&ns_ident).await;
-
- assert!(result.is_err());
- assert!(result.unwrap_err().to_string().contains("does not exist"));
-}
-
-#[tokio::test]
-async fn test_get_namespace() {
- let catalog = get_catalog().await;
-
- // Use unique namespace to avoid conflicts with other tests
- let ns = Namespace::with_properties(
- NamespaceIdent::from_strs([
- "apple",
- "ios",
- &normalize_test_name_with_parts!("test_get_namespace"),
- ])
- .unwrap(),
- HashMap::from([
- ("owner".to_string(), "ray".to_string()),
- ("community".to_string(), "apache".to_string()),
- ]),
- );
-
- // Clean up from any previous test runs
- cleanup_namespace(&catalog, ns.name()).await;
-
- // Verify that namespace doesn't exist
- assert!(catalog.get_namespace(ns.name()).await.is_err());
-
- // Create this namespace
- let created_ns = catalog
- .create_namespace(ns.name(), ns.properties().clone())
- .await
- .unwrap();
-
- assert_eq!(ns.name(), created_ns.name());
- assert_map_contains(ns.properties(), created_ns.properties());
-
- // Check that this namespace already exists
- let get_ns = catalog.get_namespace(ns.name()).await.unwrap();
- assert_eq!(ns.name(), get_ns.name());
- assert_map_contains(ns.properties(), created_ns.properties());
-}
-
-#[tokio::test]
-async fn test_list_namespace() {
- let catalog = get_catalog().await;
-
- // Use unique parent namespace to avoid conflicts
- let parent_ns_name = normalize_test_name_with_parts!("test_list_namespace");
- let parent_ident = NamespaceIdent::from_strs([&parent_ns_name]).unwrap();
-
- let ns1 = Namespace::with_properties(
- NamespaceIdent::from_strs([&parent_ns_name, "ios"]).unwrap(),
- HashMap::from([
- ("owner".to_string(), "ray".to_string()),
- ("community".to_string(), "apache".to_string()),
- ]),
- );
-
- let ns2 = Namespace::with_properties(
- NamespaceIdent::from_strs([&parent_ns_name, "macos"]).unwrap(),
- HashMap::from([
- ("owner".to_string(), "xuanwo".to_string()),
- ("community".to_string(), "apache".to_string()),
- ]),
- );
-
- // Clean up from any previous test runs
- cleanup_namespace(&catalog, ns1.name()).await;
- cleanup_namespace(&catalog, ns2.name()).await;
- cleanup_namespace(&catalog, &parent_ident).await;
-
- // Currently this namespace doesn't exist, so it should return error.
- assert!(catalog.list_namespaces(Some(&parent_ident)).await.is_err());
-
- // Create namespaces
- catalog
- .create_namespace(ns1.name(), ns1.properties().clone())
- .await
- .unwrap();
- catalog
- .create_namespace(ns2.name(), ns1.properties().clone())
- .await
- .unwrap();
-
- // List namespace
- let nss = catalog.list_namespaces(Some(&parent_ident)).await.unwrap();
-
- assert!(nss.contains(ns1.name()));
- assert!(nss.contains(ns2.name()));
-}
-
-#[tokio::test]
-async fn test_list_empty_namespace() {
- let catalog = get_catalog().await;
-
- // Use unique namespace to avoid conflicts
- let ns_apple = Namespace::with_properties(
- NamespaceIdent::from_strs([
- "list_empty",
- "apple",
- &normalize_test_name_with_parts!("test_list_empty_namespace"),
- ])
- .unwrap(),
- HashMap::from([
- ("owner".to_string(), "ray".to_string()),
- ("community".to_string(), "apache".to_string()),
- ]),
- );
-
- // Clean up from any previous test runs
- cleanup_namespace(&catalog, ns_apple.name()).await;
-
- // Currently this namespace doesn't exist, so it should return error.
- assert!(
- catalog
- .list_namespaces(Some(ns_apple.name()))
- .await
- .is_err()
- );
-
- // Create namespaces
- catalog
- .create_namespace(ns_apple.name(), ns_apple.properties().clone())
- .await
- .unwrap();
-
- // List namespace
- let nss = catalog
- .list_namespaces(Some(ns_apple.name()))
- .await
- .unwrap();
- assert!(nss.is_empty());
-}
-
-#[tokio::test]
-async fn test_list_root_namespace() {
- let catalog = get_catalog().await;
-
- // Use unique root namespace to avoid conflicts
- let root_ns_name = normalize_test_name_with_parts!("test_list_root_namespace");
- let root_ident = NamespaceIdent::from_strs([&root_ns_name]).unwrap();
-
- let ns1 = Namespace::with_properties(
- NamespaceIdent::from_strs([&root_ns_name, "apple", "ios"]).unwrap(),
- HashMap::from([
- ("owner".to_string(), "ray".to_string()),
- ("community".to_string(), "apache".to_string()),
- ]),
- );
-
- let ns2 = Namespace::with_properties(
- NamespaceIdent::from_strs([&root_ns_name, "google", "android"]).unwrap(),
- HashMap::from([
- ("owner".to_string(), "xuanwo".to_string()),
- ("community".to_string(), "apache".to_string()),
- ]),
- );
-
- // Clean up from any previous test runs
- cleanup_namespace(&catalog, ns1.name()).await;
- cleanup_namespace(&catalog, ns2.name()).await;
- cleanup_namespace(&catalog, &root_ident).await;
-
- // Currently this namespace doesn't exist, so it should return error.
- assert!(catalog.list_namespaces(Some(&root_ident)).await.is_err());
-
- // Create namespaces
- catalog
- .create_namespace(ns1.name(), ns1.properties().clone())
- .await
- .unwrap();
- catalog
- .create_namespace(ns2.name(), ns1.properties().clone())
- .await
- .unwrap();
-
- // List namespace
- let nss = catalog.list_namespaces(None).await.unwrap();
- assert!(nss.contains(&root_ident));
-}
-
-#[tokio::test]
-async fn test_create_table() {
- let catalog = get_catalog().await;
-
- // Use unique namespace to avoid conflicts
- let ns = Namespace::with_properties(
- NamespaceIdent::from_strs([
- "create_table",
- "apple",
- "ios",
- &normalize_test_name_with_parts!("test_create_table"),
- ])
- .unwrap(),
- HashMap::from([
- ("owner".to_string(), "ray".to_string()),
- ("community".to_string(), "apache".to_string()),
- ]),
- );
-
- // Clean up from any previous test runs
- cleanup_namespace(&catalog, ns.name()).await;
-
- // Create namespaces
- catalog
- .create_namespace(ns.name(), ns.properties().clone())
- .await
- .unwrap();
-
- let 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();
-
- let table_creation = TableCreation::builder()
- .name("t1".to_string())
- .schema(schema.clone())
- .build();
-
- let table = catalog
- .create_table(ns.name(), table_creation)
- .await
- .unwrap();
-
- assert_eq!(
- table.identifier(),
- &TableIdent::new(ns.name().clone(), "t1".to_string())
- );
-
- assert_eq!(
- table.metadata().current_schema().as_struct(),
- schema.as_struct()
- );
- assert_eq!(table.metadata().format_version(), FormatVersion::V2);
- assert!(table.metadata().current_snapshot().is_none());
- assert!(table.metadata().history().is_empty());
- assert!(table.metadata().default_sort_order().is_unsorted());
- assert!(table.metadata().default_partition_spec().is_unpartitioned());
-}
-
-#[tokio::test]
-async fn test_update_table() {
- let catalog = get_catalog().await;
-
- // Use unique namespace to avoid conflicts
- let ns = Namespace::with_properties(
- NamespaceIdent::from_strs([
- "update_table",
- "apple",
- "ios",
- &normalize_test_name_with_parts!("test_update_table"),
- ])
- .unwrap(),
- HashMap::from([
- ("owner".to_string(), "ray".to_string()),
- ("community".to_string(), "apache".to_string()),
- ]),
- );
-
- // Clean up from any previous test runs
- cleanup_namespace(&catalog, ns.name()).await;
-
- // Create namespaces
- catalog
- .create_namespace(ns.name(), ns.properties().clone())
- .await
- .unwrap();
-
- let 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();
-
- // Now we create a table
- let table_creation = TableCreation::builder()
- .name("t1".to_string())
- .schema(schema.clone())
- .build();
-
- let table = catalog
- .create_table(ns.name(), table_creation)
- .await
- .unwrap();
-
- assert_eq!(
- table.identifier(),
- &TableIdent::new(ns.name().clone(), "t1".to_string())
- );
-
- let tx = Transaction::new(&table);
- // Update table by committing transaction
- let table2 = tx
- .update_table_properties()
- .set("prop1".to_string(), "v1".to_string())
- .apply(tx)
- .unwrap()
- .commit(&catalog)
- .await
- .unwrap();
-
- assert_map_contains(
- &HashMap::from([("prop1".to_string(), "v1".to_string())]),
- table2.metadata().properties(),
- );
-}
-
-fn assert_map_contains(map1: &HashMap<String, String>, map2: &HashMap<String, String>) {
- for (k, v) in map1 {
- assert!(map2.contains_key(k));
- assert_eq!(map2.get(k).unwrap(), v);
- }
-}
-
-#[tokio::test]
-async fn test_list_empty_multi_level_namespace() {
- let catalog = get_catalog().await;
-
- // Use unique namespace to avoid conflicts
- let ns_apple = Namespace::with_properties(
- NamespaceIdent::from_strs([
- "multi_level",
- "a_a",
- "apple",
- &normalize_test_name_with_parts!("test_list_empty_multi_level_namespace"),
- ])
- .unwrap(),
- HashMap::from([
- ("owner".to_string(), "ray".to_string()),
- ("community".to_string(), "apache".to_string()),
- ]),
- );
-
- // Clean up from any previous test runs
- cleanup_namespace(&catalog, ns_apple.name()).await;
-
- // Currently this namespace doesn't exist, so it should return error.
- assert!(
- catalog
- .list_namespaces(Some(ns_apple.name()))
- .await
- .is_err()
- );
-
- // Create namespaces
- catalog
- .create_namespace(ns_apple.name(), ns_apple.properties().clone())
- .await
- .unwrap();
-
- // List namespace
- let nss = catalog
- .list_namespaces(Some(ns_apple.name()))
- .await
- .unwrap();
- assert!(nss.is_empty());
-}
-
-#[tokio::test]
-async fn test_register_table() {
- let catalog = get_catalog().await;
-
- // Create unique namespace to avoid conflicts
- let ns = NamespaceIdent::new(normalize_test_name_with_parts!("test_register_table"));
-
- // Clean up from any previous test runs
- cleanup_namespace(&catalog, &ns).await;
-
- catalog.create_namespace(&ns, HashMap::new()).await.unwrap();
-
- // Create the table, store the metadata location, drop the table
- let empty_schema = Schema::builder().build().unwrap();
- let table_creation = TableCreation::builder()
- .name("t1".to_string())
- .schema(empty_schema)
- .build();
-
- let table = catalog.create_table(&ns, table_creation).await.unwrap();
-
- let metadata_location = table.metadata_location().unwrap();
- catalog.drop_table(table.identifier()).await.unwrap();
-
- let new_table_identifier = TableIdent::new(ns.clone(), "t2".to_string());
- let table_registered = catalog
- .register_table(&new_table_identifier, metadata_location.to_string())
- .await
- .unwrap();
-
- assert_eq!(
- table.metadata_location(),
- table_registered.metadata_location()
- );
- assert_ne!(
- table.identifier().to_string(),
- table_registered.identifier().to_string()
- );
-}
diff --git a/crates/catalog/s3tables/src/catalog.rs b/crates/catalog/s3tables/src/catalog.rs
index e956937ce..a416c38f2 100644
--- a/crates/catalog/s3tables/src/catalog.rs
+++ b/crates/catalog/s3tables/src/catalog.rs
@@ -308,6 +308,13 @@ impl Catalog for S3TablesCatalog {
namespace: &NamespaceIdent,
_properties: HashMap<String, String>,
) -> Result<Namespace> {
+ if self.namespace_exists(namespace).await? {
+ return Err(Error::new(
+ ErrorKind::NamespaceAlreadyExists,
+ format!("Namespace {namespace:?} already exists"),
+ ));
+ }
+
let req = self
.s3tables_client
.create_namespace()
@@ -330,6 +337,13 @@ impl Catalog for S3TablesCatalog {
/// - If there is an error querying the database, returned by
/// `from_aws_sdk_error`.
async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
+ if !self.namespace_exists(namespace).await? {
+ return Err(Error::new(
+ ErrorKind::NamespaceNotFound,
+ format!("Namespace {namespace:?} does not exist"),
+ ));
+ }
+
let req = self
.s3tables_client
.get_namespace()
@@ -397,6 +411,13 @@ impl Catalog for S3TablesCatalog {
/// - Errors from the underlying database deletion process, converted using
/// `from_aws_sdk_error`.
async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()> {
+ if !self.namespace_exists(namespace).await? {
+ return Err(Error::new(
+ ErrorKind::NamespaceNotFound,
+ format!("Namespace {namespace:?} does not exist"),
+ ));
+ }
+
let req = self
.s3tables_client
.delete_namespace()
diff --git a/crates/catalog/sql/src/catalog.rs b/crates/catalog/sql/src/catalog.rs
index f2242dc40..195f6c9de 100644
--- a/crates/catalog/sql/src/catalog.rs
+++ b/crates/catalog/sql/src/catalog.rs
@@ -434,7 +434,7 @@ impl Catalog for SqlCatalog {
if exists {
return Err(Error::new(
- iceberg::ErrorKind::Unexpected,
+ iceberg::ErrorKind::NamespaceAlreadyExists,
format!("Namespace {namespace:?} already exists"),
));
}
@@ -1006,7 +1006,6 @@ mod tests {
use iceberg::io::LocalFsStorageFactory;
use iceberg::spec::{NestedField, PartitionSpec, PrimitiveType, Schema, SortOrder, Type};
use iceberg::table::Table;
- use iceberg::transaction::{ApplyTransactionAction, Transaction};
use iceberg::{Catalog, CatalogBuilder, Namespace, NamespaceIdent, TableCreation, TableIdent};
use itertools::Itertools;
use regex::Regex;
@@ -1386,20 +1385,6 @@ mod tests {
assert_eq!(catalog2.list_namespaces(None).await.unwrap(), vec![]);
}
- #[tokio::test]
- async fn test_list_namespaces_returns_multiple_namespaces() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
- let namespace_ident_1 = NamespaceIdent::new("a".into());
- let namespace_ident_2 = NamespaceIdent::new("b".into());
- create_namespaces(&catalog, &vec![&namespace_ident_1, &namespace_ident_2]).await;
-
- assert_eq!(
- to_set(catalog.list_namespaces(None).await.unwrap()),
- to_set(vec![namespace_ident_1, namespace_ident_2])
- );
- }
-
#[tokio::test]
async fn test_list_namespaces_returns_only_top_level_namespaces() {
let warehouse_loc = temp_path();
@@ -1546,31 +1531,6 @@ mod tests {
);
}
- #[tokio::test]
- async fn test_create_namespace_throws_error_if_namespace_already_exists() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
- let namespace_ident = NamespaceIdent::new("a".into());
- create_namespace(&catalog, &namespace_ident).await;
-
- assert_eq!(
- catalog
- .create_namespace(&namespace_ident, HashMap::new())
- .await
- .unwrap_err()
- .to_string(),
- format!(
- "Unexpected => Namespace {:?} already exists",
- &namespace_ident
- )
- );
-
- assert_eq!(
- catalog.get_namespace(&namespace_ident).await.unwrap(),
- Namespace::with_properties(namespace_ident, default_properties())
- );
- }
-
#[tokio::test]
async fn test_create_nested_namespace() {
let warehouse_loc = temp_path();
@@ -1640,35 +1600,6 @@ mod tests {
)
}
- #[tokio::test]
- async fn test_update_namespace() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
- let namespace_ident = NamespaceIdent::new("a".into());
- create_namespace(&catalog, &namespace_ident).await;
-
- let mut props = HashMap::from_iter([
- ("prop1".to_string(), "val1".to_string()),
- ("prop2".into(), "val2".into()),
- ]);
-
- catalog
- .update_namespace(&namespace_ident, props.clone())
- .await
- .unwrap();
-
- props.insert("exists".into(), "true".into());
-
- assert_eq!(
- *catalog
- .get_namespace(&namespace_ident)
- .await
- .unwrap()
- .properties(),
- props
- )
- }
-
#[tokio::test]
async fn test_update_nested_namespace() {
let warehouse_loc = temp_path();
@@ -1698,28 +1629,6 @@ mod tests {
)
}
- #[tokio::test]
- async fn test_update_namespace_errors_if_namespace_doesnt_exist() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
- let namespace_ident = NamespaceIdent::new("a".into());
-
- let props = HashMap::from_iter([
- ("prop1".to_string(), "val1".to_string()),
- ("prop2".into(), "val2".into()),
- ]);
-
- let err = catalog
- .update_namespace(&namespace_ident, props)
- .await
- .unwrap_err();
-
- assert_eq!(
- err.message(),
- format!("No such namespace: {namespace_ident:?}")
- );
- }
-
#[tokio::test]
async fn test_update_namespace_errors_if_nested_namespace_doesnt_exist() {
let warehouse_loc = temp_path();
@@ -1742,18 +1651,6 @@ mod tests {
);
}
- #[tokio::test]
- async fn test_drop_namespace() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
- let namespace_ident = NamespaceIdent::new("abc".into());
- create_namespace(&catalog, &namespace_ident).await;
-
- catalog.drop_namespace(&namespace_ident).await.unwrap();
-
- assert!(!catalog.namespace_exists(&namespace_ident).await.unwrap())
- }
-
#[tokio::test]
async fn test_drop_nested_namespace() {
let warehouse_loc = temp_path();
@@ -1810,22 +1707,6 @@ mod tests {
assert!(catalog.namespace_exists(&namespace_ident_a).await.unwrap());
}
- #[tokio::test]
- async fn test_drop_namespace_throws_error_if_namespace_doesnt_exist() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
-
- let non_existent_namespace_ident = NamespaceIdent::new("abc".into());
- assert_eq!(
- catalog
- .drop_namespace(&non_existent_namespace_ident)
- .await
- .unwrap_err()
- .to_string(),
- format!("Unexpected => No such namespace: {non_existent_namespace_ident:?}")
- )
- }
-
#[tokio::test]
async fn test_drop_namespace_throws_error_if_nested_namespace_doesnt_exist() {
let warehouse_loc = temp_path();
@@ -1840,7 +1721,7 @@ mod tests {
.await
.unwrap_err()
.to_string(),
- format!("Unexpected => No such namespace: {non_existent_namespace_ident:?}")
+ format!("NamespaceNotFound => No such namespace: {non_existent_namespace_ident:?}")
)
}
@@ -1864,72 +1745,6 @@ mod tests {
);
}
- #[tokio::test]
- async fn test_list_tables_returns_empty_vector() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
- let namespace_ident = NamespaceIdent::new("a".into());
- create_namespace(&catalog, &namespace_ident).await;
-
- assert_eq!(catalog.list_tables(&namespace_ident).await.unwrap(), vec![]);
- }
-
- #[tokio::test]
- async fn test_list_tables_throws_error_if_namespace_doesnt_exist() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
-
- let non_existent_namespace_ident = NamespaceIdent::new("n1".into());
-
- assert_eq!(
- catalog
- .list_tables(&non_existent_namespace_ident)
- .await
- .unwrap_err()
- .to_string(),
- format!("Unexpected => No such namespace: {non_existent_namespace_ident:?}"),
- );
- }
-
- #[tokio::test]
- async fn test_create_table_with_location() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await;
- let namespace_ident = NamespaceIdent::new("a".into());
- create_namespace(&catalog, &namespace_ident).await;
-
- let table_name = "abc";
- let location = warehouse_loc.clone();
- let table_creation = TableCreation::builder()
- .name(table_name.into())
- .location(location.clone())
- .schema(simple_table_schema())
- .build();
-
- let expected_table_ident = TableIdent::new(namespace_ident.clone(), table_name.into());
-
- assert_table_eq(
- &catalog
- .create_table(&namespace_ident, table_creation)
- .await
- .unwrap(),
- &expected_table_ident,
- &simple_table_schema(),
- );
-
- let table = catalog.load_table(&expected_table_ident).await.unwrap();
-
- assert_table_eq(&table, &expected_table_ident, &simple_table_schema());
-
- assert!(
- table
- .metadata_location()
- .unwrap()
- .to_string()
- .starts_with(&location)
- )
- }
-
#[tokio::test]
async fn test_create_table_falls_back_to_namespace_location_if_table_location_is_missing() {
let warehouse_loc = temp_path();
@@ -2128,54 +1943,10 @@ mod tests {
.await
.unwrap_err()
.to_string(),
- format!("Unexpected => Table {:?} already exists.", &table_ident)
- );
- }
-
- #[tokio::test]
- async fn test_rename_table_in_same_namespace() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
- let namespace_ident = NamespaceIdent::new("n1".into());
- create_namespace(&catalog, &namespace_ident).await;
- let src_table_ident = TableIdent::new(namespace_ident.clone(), "tbl1".into());
- let dst_table_ident = TableIdent::new(namespace_ident.clone(), "tbl2".into());
- create_table(&catalog, &src_table_ident).await;
-
- catalog
- .rename_table(&src_table_ident, &dst_table_ident)
- .await
- .unwrap();
-
- assert_eq!(catalog.list_tables(&namespace_ident).await.unwrap(), vec![
- dst_table_ident
- ],);
- }
-
- #[tokio::test]
- async fn test_rename_table_across_namespaces() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
- let src_namespace_ident = NamespaceIdent::new("a".into());
- let dst_namespace_ident = NamespaceIdent::new("b".into());
- create_namespaces(&catalog, &vec![&src_namespace_ident, &dst_namespace_ident]).await;
- let src_table_ident = TableIdent::new(src_namespace_ident.clone(), "tbl1".into());
- let dst_table_ident = TableIdent::new(dst_namespace_ident.clone(), "tbl2".into());
- create_table(&catalog, &src_table_ident).await;
-
- catalog
- .rename_table(&src_table_ident, &dst_table_ident)
- .await
- .unwrap();
-
- assert_eq!(
- catalog.list_tables(&src_namespace_ident).await.unwrap(),
- vec![],
- );
-
- assert_eq!(
- catalog.list_tables(&dst_namespace_ident).await.unwrap(),
- vec![dst_table_ident],
+ format!(
+ "TableAlreadyExists => Table {:?} already exists.",
+ &table_ident
+ )
);
}
@@ -2244,213 +2015,7 @@ mod tests {
.await
.unwrap_err()
.to_string(),
- format!("Unexpected => No such namespace: {non_existent_dst_namespace_ident:?}"),
- );
- }
-
- #[tokio::test]
- async fn test_rename_table_throws_error_if_src_table_doesnt_exist() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
- let namespace_ident = NamespaceIdent::new("n1".into());
- create_namespace(&catalog, &namespace_ident).await;
- let src_table_ident = TableIdent::new(namespace_ident.clone(), "tbl1".into());
- let dst_table_ident = TableIdent::new(namespace_ident.clone(), "tbl2".into());
-
- assert_eq!(
- catalog
- .rename_table(&src_table_ident, &dst_table_ident)
- .await
- .unwrap_err()
- .to_string(),
- format!("Unexpected => No such table: {src_table_ident:?}"),
- );
- }
-
- #[tokio::test]
- async fn test_rename_table_throws_error_if_dst_table_already_exists() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
- let namespace_ident = NamespaceIdent::new("n1".into());
- create_namespace(&catalog, &namespace_ident).await;
- let src_table_ident = TableIdent::new(namespace_ident.clone(), "tbl1".into());
- let dst_table_ident = TableIdent::new(namespace_ident.clone(), "tbl2".into());
- create_tables(&catalog, vec![&src_table_ident, &dst_table_ident]).await;
-
- assert_eq!(
- catalog
- .rename_table(&src_table_ident, &dst_table_ident)
- .await
- .unwrap_err()
- .to_string(),
- format!("Unexpected => Table {:?} already exists.", &dst_table_ident),
- );
- }
-
- #[tokio::test]
- async fn test_drop_table_throws_error_if_table_not_exist() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await;
- let namespace_ident = NamespaceIdent::new("a".into());
- let table_name = "tbl1";
- let table_ident = TableIdent::new(namespace_ident.clone(), table_name.into());
- create_namespace(&catalog, &namespace_ident).await;
-
- let err = catalog
- .drop_table(&table_ident)
- .await
- .unwrap_err()
- .to_string();
- assert_eq!(
- err,
- "Unexpected => No such table: TableIdent { namespace: NamespaceIdent([\"a\"]), name: \"tbl1\" }"
- );
- }
-
- #[tokio::test]
- async fn test_drop_table() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await;
- let namespace_ident = NamespaceIdent::new("a".into());
- let table_name = "tbl1";
- let table_ident = TableIdent::new(namespace_ident.clone(), table_name.into());
- create_namespace(&catalog, &namespace_ident).await;
-
- let location = warehouse_loc.clone();
- let table_creation = TableCreation::builder()
- .name(table_name.into())
- .location(location.clone())
- .schema(simple_table_schema())
- .build();
-
- catalog
- .create_table(&namespace_ident, table_creation)
- .await
- .unwrap();
-
- let table = catalog.load_table(&table_ident).await.unwrap();
- assert_table_eq(&table, &table_ident, &simple_table_schema());
-
- catalog.drop_table(&table_ident).await.unwrap();
- let err = catalog
- .load_table(&table_ident)
- .await
- .unwrap_err()
- .to_string();
- assert_eq!(
- err,
- "Unexpected => No such table: TableIdent { namespace: NamespaceIdent([\"a\"]), name: \"tbl1\" }"
- );
- }
-
- #[tokio::test]
- async fn test_register_table_throws_error_if_table_with_same_name_already_exists() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await;
- let namespace_ident = NamespaceIdent::new("a".into());
- create_namespace(&catalog, &namespace_ident).await;
- let table_name = "tbl1";
- let table_ident = TableIdent::new(namespace_ident.clone(), table_name.into());
- create_table(&catalog, &table_ident).await;
-
- assert_eq!(
- catalog
- .register_table(&table_ident, warehouse_loc)
- .await
- .unwrap_err()
- .to_string(),
- format!("Unexpected => Table {:?} already exists.", &table_ident)
- );
- }
-
- #[tokio::test]
- async fn test_register_table() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await;
- let namespace_ident = NamespaceIdent::new("a".into());
- create_namespace(&catalog, &namespace_ident).await;
-
- let table_name = "abc";
- let location = warehouse_loc.clone();
- let table_creation = TableCreation::builder()
- .name(table_name.into())
- .location(location.clone())
- .schema(simple_table_schema())
- .build();
-
- let table_ident = TableIdent::new(namespace_ident.clone(), table_name.into());
- let expected_table = catalog
- .create_table(&namespace_ident, table_creation)
- .await
- .unwrap();
-
- let metadata_location = expected_table
- .metadata_location()
- .expect("Expected metadata location to be set")
- .to_string();
-
- assert_table_eq(&expected_table, &table_ident, &simple_table_schema());
-
- let _ = catalog.drop_table(&table_ident).await;
-
- let table = catalog
- .register_table(&table_ident, metadata_location.clone())
- .await
- .unwrap();
-
- assert_eq!(table.identifier(), expected_table.identifier());
- assert_eq!(table.metadata_location(), Some(metadata_location.as_str()));
- }
-
- #[tokio::test]
- async fn test_update_table() {
- let warehouse_loc = temp_path();
- let catalog = new_sql_catalog(warehouse_loc, Some("iceberg")).await;
-
- // Create a test namespace and table
- let namespace_ident = NamespaceIdent::new("ns1".into());
- create_namespace(&catalog, &namespace_ident).await;
- let table_ident = TableIdent::new(namespace_ident.clone(), "tbl1".into());
- create_table(&catalog, &table_ident).await;
-
- let table = catalog.load_table(&table_ident).await.unwrap();
-
- // Store the original metadata location for comparison
- let original_metadata_location = table.metadata_location().unwrap().to_string();
-
- // Create a transaction to update the table
- let tx = Transaction::new(&table);
- let tx = tx
- .update_table_properties()
- .set("test_property".to_string(), "test_value".to_string())
- .apply(tx)
- .unwrap();
-
- // Commit the transaction to the catalog
- let updated_table = tx.commit(&catalog).await.unwrap();
-
- // Verify the update was successful
- assert_eq!(
- updated_table.metadata().properties().get("test_property"),
- Some(&"test_value".to_string())
- );
- // Verify the metadata location has been updated
- assert_ne!(
- updated_table.metadata_location().unwrap(),
- original_metadata_location.as_str()
- );
-
- // Load the table again from the catalog to verify changes were persisted
- let reloaded = catalog.load_table(&table_ident).await.unwrap();
-
- // Verify the reloaded table matches the updated table
- assert_eq!(
- reloaded.metadata().properties().get("test_property"),
- Some(&"test_value".to_string())
- );
- assert_eq!(
- reloaded.metadata_location(),
- updated_table.metadata_location()
+ format!("NamespaceNotFound => No such namespace: {non_existent_dst_namespace_ident:?}"),
);
}
}
diff --git a/crates/catalog/sql/src/error.rs b/crates/catalog/sql/src/error.rs
index a08f75559..55e0e0a36 100644
--- a/crates/catalog/sql/src/error.rs
+++ b/crates/catalog/sql/src/error.rs
@@ -28,21 +28,21 @@ pub fn from_sqlx_error(error: sqlx::Error) -> Error {
pub fn no_such_namespace_err<T>(namespace: &NamespaceIdent) -> Result<T> {
Err(Error::new(
- ErrorKind::Unexpected,
+ ErrorKind::NamespaceNotFound,
format!("No such namespace: {namespace:?}"),
))
}
pub fn no_such_table_err<T>(table_ident: &TableIdent) -> Result<T> {
Err(Error::new(
- ErrorKind::Unexpected,
+ ErrorKind::TableNotFound,
format!("No such table: {table_ident:?}"),
))
}
pub fn table_already_exists_err<T>(table_ident: &TableIdent) -> Result<T> {
Err(Error::new(
- ErrorKind::Unexpected,
+ ErrorKind::TableAlreadyExists,
format!("Table {table_ident:?} already exists."),
))
}
diff --git a/crates/iceberg/src/catalog/memory/catalog.rs b/crates/iceberg/src/catalog/memory/catalog.rs
index 992a0ed61..25ae00441 100644
--- a/crates/iceberg/src/catalog/memory/catalog.rs
+++ b/crates/iceberg/src/catalog/memory/catalog.rs
@@ -322,8 +322,8 @@ impl Catalog for MemoryCatalog {
async fn drop_table(&self, table_ident: &TableIdent) -> Result<()> {
let mut root_namespace_state = self.root_namespace_state.lock().await;
- let metadata_location = root_namespace_state.remove_existing_table(table_ident)?;
- self.file_io.delete(&metadata_location).await
+ root_namespace_state.remove_existing_table(table_ident)?;
+ Ok(())
}
/// Check if a table exists in the catalog.