refactor(memory): derive catalog configuration from properties

crates/iceberg/public-api.txt

@@ -1112,7 +1112,7 @@ pub fn iceberg::memory::MemoryCatalog::update_namespace<'life0, 'life1, 'async_t
 pub fn iceberg::memory::MemoryCatalog::update_table<'life0, 'async_trait>(&'life0 self, commit: iceberg::TableCommit) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<iceberg::table::Table>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
 pub struct iceberg::memory::MemoryCatalogBuilder
 impl core::default::Default for iceberg::memory::MemoryCatalogBuilder
-pub fn iceberg::memory::MemoryCatalogBuilder::default() -> Self
+pub fn iceberg::memory::MemoryCatalogBuilder::default() -> iceberg::memory::MemoryCatalogBuilder
 impl core::fmt::Debug for iceberg::memory::MemoryCatalogBuilder
 pub fn iceberg::memory::MemoryCatalogBuilder::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 impl iceberg::CatalogBuilder for iceberg::memory::MemoryCatalogBuilder
@@ -1121,6 +1121,11 @@ pub fn iceberg::memory::MemoryCatalogBuilder::load(self, name: impl core::conver
 pub fn iceberg::memory::MemoryCatalogBuilder::with_kms_client_factory(self, kms_client_factory: alloc::sync::Arc<dyn iceberg::encryption::kms::KmsClientFactory>) -> Self
 pub fn iceberg::memory::MemoryCatalogBuilder::with_runtime(self, runtime: iceberg::Runtime) -> Self
 pub fn iceberg::memory::MemoryCatalogBuilder::with_storage_factory(self, storage_factory: alloc::sync::Arc<dyn iceberg::io::StorageFactory>) -> Self
+pub struct iceberg::memory::MemoryCatalogProperties
+impl iceberg::memory::MemoryCatalogProperties
+pub fn iceberg::memory::MemoryCatalogProperties::from_properties(properties: &std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> iceberg::Result<Self>
+impl core::fmt::Debug for iceberg::memory::MemoryCatalogProperties
+pub fn iceberg::memory::MemoryCatalogProperties::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 pub const iceberg::memory::MEMORY_CATALOG_WAREHOUSE: &str
 pub mod iceberg::metadata_columns
 pub const iceberg::metadata_columns::RESERVED_COL_NAME_CHANGE_ORDINAL: &str

crates/iceberg/src/catalog/memory/catalog.rs

@@ -23,6 +23,7 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use futures::lock::{Mutex, MutexGuard};
+use iceberg_property_macro::Properties;
 use itertools::Itertools;
 
 use super::namespace_state::NamespaceState;
@@ -43,29 +44,13 @@ pub const MEMORY_CATALOG_WAREHOUSE: &str = "warehouse";
 const LOCATION: &str = "location";
 
 /// Builder for [`MemoryCatalog`].
-#[derive(Debug)]
+#[derive(Debug, Default)]
 pub struct MemoryCatalogBuilder {
-    config: MemoryCatalogConfig,
     storage_factory: Option<Arc<dyn StorageFactory>>,
     kms_client_factory: Option<Arc<dyn KmsClientFactory>>,
     runtime: Option<Runtime>,
 }
 
-impl Default for MemoryCatalogBuilder {
-    fn default() -> Self {
-        Self {
-            config: MemoryCatalogConfig {
-                name: None,
-                warehouse: "".to_string(),
-                props: HashMap::new(),
-            },
-            storage_factory: None,
-            kms_client_factory: None,
-            runtime: None,
-        }
-    }
-}
-
 impl CatalogBuilder for MemoryCatalogBuilder {
     type C = MemoryCatalog;
 
@@ -85,69 +70,80 @@ impl CatalogBuilder for MemoryCatalogBuilder {
     }
 
     fn load(
-        mut self,
+        self,
         name: impl Into<String>,
         props: HashMap<String, String>,
     ) -> impl Future<Output = Result<Self::C>> + Send {
-        self.config.name = Some(name.into());
-
-        if props.contains_key(MEMORY_CATALOG_WAREHOUSE) {
-            self.config.warehouse = props
-                .get(MEMORY_CATALOG_WAREHOUSE)
-                .cloned()
-                .unwrap_or_default()
-        }
-
-        // Collect other remaining properties
-        self.config.props = props
-            .into_iter()
-            .filter(|(k, _)| k != MEMORY_CATALOG_WAREHOUSE)
-            .collect();
+        let name = name.into();
 
         async move {
-            if self.config.name.is_none() {
-                Err(Error::new(
-                    ErrorKind::DataInvalid,
-                    "Catalog name is required",
-                ))
-            } else if self.config.warehouse.is_empty() {
-                Err(Error::new(
-                    ErrorKind::DataInvalid,
-                    "Catalog warehouse is required",
-                ))
-            } else {
-                let runtime = self.runtime.unwrap_or_else(Runtime::current);
-                let kms_client = match self.kms_client_factory {
-                    Some(factory) => Some(factory.create_kms_client(&self.config.props).await?),
-                    None => None,
-                };
-                MemoryCatalog::new(self.config, self.storage_factory, runtime, kms_client)
-            }
+            let catalog_properties = MemoryCatalogProperties::from_properties(&props)?;
+            let runtime = self.runtime.unwrap_or_else(Runtime::current);
+            let kms_client = match self.kms_client_factory {
+                Some(factory) => Some(factory.create_kms_client(&props).await?),
+                None => None,
+            };
+            MemoryCatalog::new(
+                name,
+                catalog_properties,
+                props,
+                self.storage_factory,
+                runtime,
+                kms_client,
+            )
         }
     }
 }
 
-#[derive(Clone, Debug)]
-pub(crate) struct MemoryCatalogConfig {
-    name: Option<String>,
+fn parse_warehouse(warehouse: &str) -> Result<String> {
+    if warehouse.is_empty() {
+        Err(Error::new(
+            ErrorKind::DataInvalid,
+            "Catalog warehouse is required",
+        ))
+    } else {
+        Ok(warehouse.to_string())
+    }
+}
+
+/// Memory catalog properties parsed from a catalog property map.
+#[derive(Debug, Properties)]
+pub struct MemoryCatalogProperties {
+    #[property(
+        key = MEMORY_CATALOG_WAREHOUSE,
+        default = "",
+        parse_with = parse_warehouse
+    )]
     warehouse: String,
-    props: HashMap<String, String>,
 }
 
 /// Memory catalog implementation.
-#[derive(Debug)]
 pub struct MemoryCatalog {
+    name: String,
+    properties: MemoryCatalogProperties,
+    props: HashMap<String, String>,
     root_namespace_state: Mutex<NamespaceState>,
     file_io: FileIO,
-    warehouse_location: String,
     runtime: Runtime,
     kms_client: Option<Arc<dyn KeyManagementClient>>,
 }
 
+impl std::fmt::Debug for MemoryCatalog {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("MemoryCatalog")
+            .field("name", &self.name)
+            .field("properties", &self.properties)
+            .field("props", &self.props)
+            .finish_non_exhaustive()
+    }
+}
+
 impl MemoryCatalog {
     /// Creates a memory catalog.
     fn new(
-        config: MemoryCatalogConfig,
+        name: String,
+        properties: MemoryCatalogProperties,
+        props: HashMap<String, String>,
         storage_factory: Option<Arc<dyn StorageFactory>>,
         runtime: Runtime,
         kms_client: Option<Arc<dyn KeyManagementClient>>,
@@ -156,9 +152,13 @@ impl MemoryCatalog {
         let factory = storage_factory.unwrap_or_else(|| Arc::new(MemoryStorageFactory));
 
         Ok(Self {
+            name,
+            properties,
+            file_io: FileIOBuilder::new(factory)
+                .with_props(props.clone())
+                .build(),
+            props,
             root_namespace_state: Mutex::new(NamespaceState::default()),
-            file_io: FileIOBuilder::new(factory).with_props(config.props).build(),
-            warehouse_location: config.warehouse,
             runtime,
             kms_client,
         })
@@ -308,7 +308,17 @@ impl Catalog for MemoryCatalog {
             let namespace_properties = root_namespace_state.get_properties(namespace_ident)?;
             let location_prefix = match namespace_properties.get(LOCATION) {
                 Some(namespace_location) => namespace_location.clone(),
-                None => format!("{}/{}", self.warehouse_location, namespace_ident.join("/")),
+                None if self.properties.warehouse.is_empty() => {
+                    return Err(Error::new(
+                        ErrorKind::DataInvalid,
+                        "Catalog warehouse is required",
+                    ));
+                }
+                None => format!(
+                    "{}/{}",
+                    self.properties.warehouse,
+                    namespace_ident.join("/")
+                ),
             };
 
             let location = format!("{}/{}", location_prefix, table_ident.name());
@@ -458,6 +468,31 @@ pub(crate) mod tests {
         temp_dir.path().to_str().unwrap().to_string()
     }
 
+    #[test]
+    fn test_catalog_properties() {
+        let properties = MemoryCatalogProperties::from_properties(&HashMap::from([
+            (
+                MEMORY_CATALOG_WAREHOUSE.to_string(),
+                "memory:///warehouse".to_string(),
+            ),
+            ("custom.property".to_string(), "value".to_string()),
+        ]))
+        .unwrap();
+
+        assert_eq!(properties.warehouse, "memory:///warehouse");
+
+        let properties = MemoryCatalogProperties::from_properties(&HashMap::new()).unwrap();
+        assert!(properties.warehouse.is_empty());
+
+        let error = MemoryCatalogProperties::from_properties(&HashMap::from([(
+            MEMORY_CATALOG_WAREHOUSE.to_string(),
+            String::new(),
+        )]))
+        .unwrap_err();
+        assert_eq!(error.kind(), ErrorKind::DataInvalid);
+        assert_eq!(error.message(), "Catalog warehouse is required");
+    }
+
     pub(crate) async fn new_memory_catalog() -> impl Catalog {
         let warehouse_location = temp_path();
         MemoryCatalogBuilder::default()
@@ -1405,11 +1440,26 @@ pub(crate) mod tests {
      {
         let catalog = MemoryCatalogBuilder::default()
             .load("memory", HashMap::from([]))
-            .await;
+            .await
+            .unwrap();
+        let namespace_ident = NamespaceIdent::new("namespace".into());
+        catalog
+            .create_namespace(&namespace_ident, HashMap::new())
+            .await
+            .unwrap();
+        let error = catalog
+            .create_table(
+                &namespace_ident,
+                TableCreation::builder()
+                    .name("table".into())
+                    .schema(simple_table_schema())
+                    .build(),
+            )
+            .await
+            .unwrap_err();
 
-        assert!(catalog.is_err());
         assert_eq!(
-            catalog.unwrap_err().to_string(),
+            error.to_string(),
             "DataInvalid => Catalog warehouse is required"
         );
     }