feat(catalog): implement catalog loader for s3tables (apache#1598)

## Which issue does this PR close?
- Closes apache#1256

## What changes are included in this PR?
### Catalog Registry Refactor

* Introduced a static `CATALOG_REGISTRY` mapping catalog type strings
(e.g., `"rest"`, `"s3tables"`) to their corresponding builder factory
closures.
* Eliminates hard-coded catalog type handling in `load()` and
centralizes catalog registration logic.
* Makes it easier to add new catalog implementations in the future
without touching core loader logic.

### New S3 Tables Catalog Support

* Implemented an **S3 Tables Catalog Builder**.
* Added the `"s3tables"` catalog type to the registry in
`crates/catalog/loader`.
* Updated workspace configuration to include the new
`iceberg-catalog-s3tables` crate.
* Added comprehensive example usage in the crate documentation.

### Hive metastore
* The `hive_metastore` was upgraded from `0.1.0` to fix after rebasing
your S3Tables feature branch because the HMS catalog and S3Tables
catalog share common Rust dependencies (pilota, volo-thrift, etc.)
* These shared dependencies created version conflicts that manifested as
"derivative attribute compilation errors."
Upgrading to hive_metastore `0.2.0` resolved these transitive dependency
conflicts.

## Are these changes tested?
Yes.

* Added new unit tests for the S3 Tables catalog.
* Expanded loader tests to:

  * Verify loading of supported catalogs via the registry.
  * Validate improved error messages listing supported catalog types.
* Test the new ergonomic `CatalogLoader` API for both `"rest"` and
`"s3tables"`.

Author: fvaleye

Cargo.lock

@@ -55,6 +55,7 @@ async-std = "1.12"
 async-trait = "0.1.88"
 aws-config = "1.6.1"
 aws-sdk-glue = "1.39"
+aws-sdk-s3tables = "1.28.0"
 backon = "1.5.1"
 base64 = "0.22.1"
 bimap = "0.6"
@@ -73,11 +74,12 @@ faststr = "0.2.31"
 fnv = "1.0.7"
 fs-err = "3.1.0"
 futures = "0.3"
-hive_metastore = "0.1"
+hive_metastore = "0.2.0"
 http = "1.2"
 iceberg = { version = "0.6.0", path = "./crates/iceberg" }
 iceberg-catalog-rest = { version = "0.6.0", path = "./crates/catalog/rest" }
 iceberg-catalog-glue = { version = "0.6.0", path = "./crates/catalog/glue" }
+iceberg-catalog-s3tables = { version = "0.6.0", path = "./crates/catalog/s3tables" }
 iceberg-datafusion = { version = "0.6.0", path = "./crates/integrations/datafusion" }
 indicatif = "0.17"
 itertools = "0.13"
@@ -93,7 +95,7 @@ once_cell = "1.20"
 opendal = "0.54.0"
 ordered-float = "4"
 parquet = "55"
-pilota = "0.11.2"
+pilota = "0.11.10"
 port_scanner = "0.1.5"
 pretty_assertions = "1.4"
 rand = "0.8.5"
@@ -121,5 +123,5 @@ typed-builder = "0.20"
 url = "2.5.4"
 uuid = { version = "1.16", features = ["v7"] }
 volo = "0.10.6"
-volo-thrift = "0.10.6"
+volo-thrift = "0.10.8"
 zstd = "0.13.2"

Cargo.toml

@@ -32,5 +32,6 @@ repository = { workspace = true }
 iceberg = { workspace = true }
 iceberg-catalog-rest = { workspace = true }
 iceberg-catalog-glue = { workspace = true }
+iceberg-catalog-s3tables = { workspace = true }
 tokio = { workspace = true }
 async-trait = { workspace = true }

crates/catalog/loader/Cargo.toml

@@ -22,6 +22,22 @@ use async_trait::async_trait;
 use iceberg::{Catalog, CatalogBuilder, Error, ErrorKind, Result};
 use iceberg_catalog_glue::GlueCatalogBuilder;
 use iceberg_catalog_rest::RestCatalogBuilder;
+use iceberg_catalog_s3tables::S3TablesCatalogBuilder;
+
+/// A CatalogBuilderFactory creating a new catalog builder.
+type CatalogBuilderFactory = fn() -> Box<dyn BoxedCatalogBuilder>;
+
+/// A registry of catalog builders.
+static CATALOG_REGISTRY: &[(&str, CatalogBuilderFactory)] = &[
+    ("rest", || Box::new(RestCatalogBuilder::default())),
+    ("glue", || Box::new(GlueCatalogBuilder::default())),
+    ("s3tables", || Box::new(S3TablesCatalogBuilder::default())),
+];
+
+/// Return the list of supported catalog types.
+pub fn supported_types() -> Vec<&'static str> {
+    CATALOG_REGISTRY.iter().map(|(k, _)| *k).collect()
+}
 
 #[async_trait]
 pub trait BoxedCatalogBuilder {
@@ -44,22 +60,74 @@ impl<T: CatalogBuilder + 'static> BoxedCatalogBuilder for T {
     }
 }
 
+/// Load a catalog from a string.
 pub fn load(r#type: &str) -> Result<Box<dyn BoxedCatalogBuilder>> {
-    match r#type {
-        "rest" => Ok(Box::new(RestCatalogBuilder::default()) as Box<dyn BoxedCatalogBuilder>),
-        "glue" => Ok(Box::new(GlueCatalogBuilder::default()) as Box<dyn BoxedCatalogBuilder>),
-        _ => Err(Error::new(
+    let key = r#type.trim();
+    if let Some((_, factory)) = CATALOG_REGISTRY
+        .iter()
+        .find(|(k, _)| k.eq_ignore_ascii_case(key))
+    {
+        Ok(factory())
+    } else {
+        Err(Error::new(
             ErrorKind::FeatureUnsupported,
-            format!("Unsupported catalog type: {}", r#type),
-        )),
+            format!(
+                "Unsupported catalog type: {}. Supported types: {}",
+                r#type,
+                supported_types().join(", ")
+            ),
+        ))
+    }
+}
+
+/// Ergonomic catalog loader builder pattern.
+pub struct CatalogLoader<'a> {
+    catalog_type: &'a str,
+}
+
+impl<'a> From<&'a str> for CatalogLoader<'a> {
+    fn from(s: &'a str) -> Self {
+        Self { catalog_type: s }
+    }
+}
+
+impl CatalogLoader<'_> {
+    pub async fn load(
+        self,
+        name: String,
+        props: HashMap<String, String>,
+    ) -> Result<Arc<dyn Catalog>> {
+        let builder = load(self.catalog_type)?;
+        builder.load(name, props).await
     }
 }
 
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;
 
-    use crate::load;
+    use crate::{CatalogLoader, load};
+
+    #[tokio::test]
+    async fn test_load_glue_catalog() {
+        use iceberg_catalog_glue::GLUE_CATALOG_PROP_WAREHOUSE;
+
+        let catalog_loader = load("glue").unwrap();
+        let catalog = catalog_loader
+            .load(
+                "glue".to_string(),
+                HashMap::from([
+                    (
+                        GLUE_CATALOG_PROP_WAREHOUSE.to_string(),
+                        "s3://test".to_string(),
+                    ),
+                    ("key".to_string(), "value".to_string()),
+                ]),
+            )
+            .await;
+
+        assert!(catalog.is_ok());
+    }
 
     #[tokio::test]
     async fn test_load_rest_catalog() {
@@ -83,17 +151,22 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_load_glue_catalog() {
-        use iceberg_catalog_glue::GLUE_CATALOG_PROP_WAREHOUSE;
+    async fn test_load_unsupported_catalog() {
+        let result = load("unsupported");
+        assert!(result.is_err());
+    }
 
-        let catalog_loader = load("glue").unwrap();
-        let catalog = catalog_loader
+    #[tokio::test]
+    async fn test_catalog_loader_pattern() {
+        use iceberg_catalog_rest::REST_CATALOG_PROP_URI;
+
+        let catalog = CatalogLoader::from("rest")
             .load(
-                "glue".to_string(),
+                "rest".to_string(),
                 HashMap::from([
                     (
-                        GLUE_CATALOG_PROP_WAREHOUSE.to_string(),
-                        "s3://test".to_string(),
+                        REST_CATALOG_PROP_URI.to_string(),
+                        "http://localhost:8080".to_string(),
                     ),
                     ("key".to_string(), "value".to_string()),
                 ]),
@@ -102,4 +175,36 @@ mod tests {
 
         assert!(catalog.is_ok());
     }
+
+    #[tokio::test]
+    async fn test_catalog_loader_pattern_s3tables() {
+        use iceberg_catalog_s3tables::S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN;
+
+        let catalog = CatalogLoader::from("s3tables")
+            .load(
+                "s3tables".to_string(),
+                HashMap::from([
+                    (
+                        S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN.to_string(),
+                        "arn:aws:s3tables:us-east-1:123456789012:bucket/test".to_string(),
+                    ),
+                    ("key".to_string(), "value".to_string()),
+                ]),
+            )
+            .await;
+
+        assert!(catalog.is_ok());
+    }
+
+    #[tokio::test]
+    async fn test_error_message_includes_supported_types() {
+        let err = match load("does-not-exist") {
+            Ok(_) => panic!("expected error for unsupported type"),
+            Err(e) => e,
+        };
+        let msg = err.message().to_string();
+        assert!(msg.contains("Supported types:"));
+        // Should include at least the built-in type
+        assert!(msg.contains("rest"));
+    }
 }

crates/catalog/loader/src/lib.rs

@@ -699,7 +699,7 @@ impl Catalog for RestCatalog {
             .config
             .unwrap_or_default()
             .into_iter()
-            .chain(self.user_config.props.clone().into_iter())
+            .chain(self.user_config.props.clone())
             .collect();
 
         let file_io = self
@@ -750,7 +750,7 @@ impl Catalog for RestCatalog {
             .config
             .unwrap_or_default()
             .into_iter()
-            .chain(self.user_config.props.clone().into_iter())
+            .chain(self.user_config.props.clone())
             .collect();
 
         let file_io = self

crates/catalog/rest/src/catalog.rs

@@ -29,12 +29,11 @@ license = { workspace = true }
 repository = { workspace = true }
 
 [dependencies]
-anyhow = { workspace = true }
 async-trait = { workspace = true }
 aws-config = { workspace = true }
-aws-sdk-s3tables = "1.10.0"
+aws-sdk-s3tables = { workspace = true }
 iceberg = { workspace = true }
-typed-builder = { workspace = true }
+
 
 [dev-dependencies]
 iceberg_test_utils = { path = "../../test_utils", features = ["tests"] }