feat(catalog): catalog registry introspection and clearer error messages

- Add supported_types() to list registry entries
- Include supported types in unsupported-type error

Author: fvaleye

crates/catalog/loader/src/lib.rs

@@ -22,6 +22,23 @@ use async_trait::async_trait;
 use iceberg::{Catalog, CatalogBuilder, Error, ErrorKind, Result};
 use iceberg_catalog_rest::RestCatalogBuilder;
 
+/// A CatalogBuilderFactory creating a new catalog builder.
+type CatalogBuilderFactory = fn() -> Box<dyn BoxedCatalogBuilder>;
+
+// A registry of catalog builders.
+fn rest_factory() -> Box<dyn BoxedCatalogBuilder> {
+    Box::new(RestCatalogBuilder::default())
+}
+
+/// A registry of catalog builders.
+static CATALOG_REGISTRY: &[(&str, CatalogBuilderFactory)] =
+    &[("rest", rest_factory as CatalogBuilderFactory)];
+
+/// 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 {
     async fn load(
@@ -43,13 +60,45 @@ 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>),
-        _ => 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
     }
 }
 
@@ -59,7 +108,7 @@ mod tests {
 
     use iceberg_catalog_rest::REST_CATALOG_PROP_URI;
 
-    use crate::load;
+    use crate::{CatalogLoader, load};
 
     #[tokio::test]
     async fn test_load_rest_catalog() {
@@ -79,4 +128,40 @@ mod tests {
 
         assert!(catalog.is_ok());
     }
+
+    #[tokio::test]
+    async fn test_load_unsupported_catalog() {
+        let result = load("unsupported");
+        assert!(result.is_err());
+    }
+
+    #[tokio::test]
+    async fn test_catalog_loader_pattern() {
+        let catalog = CatalogLoader::from("rest")
+            .load(
+                "rest".to_string(),
+                HashMap::from([
+                    (
+                        REST_CATALOG_PROP_URI.to_string(),
+                        "http://localhost:8080".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"));
+    }
 }