diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 4aeddb665a..0dec073bff 100644 --- a/crates/iceberg/public-api.txt +++ b/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> + 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) -> 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) -> Self +pub struct iceberg::memory::MemoryCatalogProperties +impl iceberg::memory::MemoryCatalogProperties +pub fn iceberg::memory::MemoryCatalogProperties::from_properties(properties: &std::collections::hash::map::HashMap) -> iceberg::Result +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 diff --git a/crates/iceberg/src/catalog/memory/catalog.rs b/crates/iceberg/src/catalog/memory/catalog.rs index 85fcd824ef..c3251f2dd7 100644 --- a/crates/iceberg/src/catalog/memory/catalog.rs +++ b/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>, kms_client_factory: Option>, runtime: Option, } -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,78 @@ impl CatalogBuilder for MemoryCatalogBuilder { } fn load( - mut self, + self, name: impl Into, props: HashMap, ) -> impl Future> + 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, +fn parse_warehouse(warehouse: &str) -> Result { + 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, } /// Memory catalog implementation. -#[derive(Debug)] pub struct MemoryCatalog { + name: String, + properties: MemoryCatalogProperties, root_namespace_state: Mutex, file_io: FileIO, - warehouse_location: String, runtime: Runtime, kms_client: Option>, } +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) + .finish_non_exhaustive() + } +} + impl MemoryCatalog { /// Creates a memory catalog. fn new( - config: MemoryCatalogConfig, + name: String, + properties: MemoryCatalogProperties, + props: HashMap, storage_factory: Option>, runtime: Runtime, kms_client: Option>, @@ -156,9 +150,10 @@ impl MemoryCatalog { let factory = storage_factory.unwrap_or_else(|| Arc::new(MemoryStorageFactory)); Ok(Self { + name, + properties, + file_io: FileIOBuilder::new(factory).with_props(props).build(), 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 +303,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 +463,35 @@ 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 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"); + } + + #[test] + fn test_catalog_properties_with_default_warehouse() { + let properties = MemoryCatalogProperties::from_properties(&HashMap::new()).unwrap(); + + assert_eq!(properties.warehouse, ""); + } + pub(crate) async fn new_memory_catalog() -> impl Catalog { let warehouse_location = temp_path(); MemoryCatalogBuilder::default() @@ -1405,11 +1439,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" ); }