diff --git a/Cargo.lock b/Cargo.lock index 3802cc03ff..baae9a411c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3937,6 +3937,7 @@ version = "0.10.0" dependencies = [ "async-trait", "iceberg", + "iceberg-property-macro", "itertools 0.13.0", "regex", "sqlx", diff --git a/crates/catalog/sql/Cargo.toml b/crates/catalog/sql/Cargo.toml index 5861c7807d..e78e4e0ffc 100644 --- a/crates/catalog/sql/Cargo.toml +++ b/crates/catalog/sql/Cargo.toml @@ -32,6 +32,7 @@ repository = { workspace = true } [dependencies] async-trait = { workspace = true } iceberg = { workspace = true } +iceberg-property-macro = { workspace = true } sqlx = { version = "0.8.1", features = ["any"], default-features = false } strum = { workspace = true } tracing = { workspace = true } diff --git a/crates/catalog/sql/src/catalog.rs b/crates/catalog/sql/src/catalog.rs index 73671f5c4b..e138748e98 100644 --- a/crates/catalog/sql/src/catalog.rs +++ b/crates/catalog/sql/src/catalog.rs @@ -29,6 +29,7 @@ use iceberg::{ Catalog, CatalogBuilder, Error, ErrorKind, MetadataLocation, Namespace, NamespaceIdent, Result, Runtime, TableCommit, TableCreation, TableIdent, }; +use iceberg_property_macro::Properties; use sqlx::any::{AnyPoolOptions, AnyQueryResult, AnyRow, install_default_drivers}; use sqlx::{Any, AnyPool, Column, Executor, Row, Transaction}; @@ -204,87 +205,105 @@ impl CatalogBuilder for SqlCatalogBuilder { self.config.props.insert(k, v); } - if let Some(uri) = self.config.props.remove(SQL_CATALOG_PROP_URI) { - self.config.uri = uri; - } - if let Some(warehouse_location) = self.config.props.remove(SQL_CATALOG_PROP_WAREHOUSE) { - self.config.warehouse_location = warehouse_location; - } - let name = name.into(); - let mut valid_sql_bind_style = true; + async move { + if name.trim().is_empty() { + return Err(Error::new( + ErrorKind::DataInvalid, + "Catalog name cannot be empty", + )); + } - // Accept the preferred `sql.bind-style` key, falling back to the legacy `sql_bind_style`. - let sql_bind_style = self - .config - .props - .remove(SQL_CATALOG_PROP_BIND_STYLE) - .or_else(|| self.config.props.remove(SQL_CATALOG_PROP_BIND_STYLE_LEGACY)); + let mut catalog_properties = SqlCatalogProperties::from_properties(&self.config.props)?; - // Validate the SQL bind style - if let Some(sql_bind_style) = sql_bind_style { - if let Ok(sql_bind_style) = SqlBindStyle::from_str(&sql_bind_style) { - self.config.sql_bind_style = sql_bind_style; - } else { - valid_sql_bind_style = false; + if let Some(uri) = catalog_properties.uri { + self.config.uri = uri; } - } + catalog_properties.props.remove(SQL_CATALOG_PROP_URI); - // Parse the requested schema version up front so invalid values fail fast rather than - // silently falling back to V0. - let mut valid_schema_version = true; - if let Some(schema_version) = self.config.props.remove(SQL_CATALOG_PROP_SCHEMA_VERSION) { - match SchemaVersion::from_str(&schema_version) { - Ok(schema_version) => self.config.schema_version = Some(schema_version), - Err(_) => valid_schema_version = false, + if let Some(warehouse_location) = catalog_properties.warehouse_location { + self.config.warehouse_location = warehouse_location; + } + catalog_properties.props.remove(SQL_CATALOG_PROP_WAREHOUSE); + + // Accept the preferred `sql.bind-style` key, falling back to the legacy + // `sql_bind_style`. Preserve the existing behavior where the legacy key remains in + // the forwarded properties if both aliases are present. + let has_preferred_bind_style = catalog_properties.sql_bind_style.is_some(); + let sql_bind_style = catalog_properties + .sql_bind_style + .or(catalog_properties.legacy_sql_bind_style); + catalog_properties.props.remove(SQL_CATALOG_PROP_BIND_STYLE); + if !has_preferred_bind_style { + catalog_properties + .props + .remove(SQL_CATALOG_PROP_BIND_STYLE_LEGACY); + } + if let Some(sql_bind_style) = sql_bind_style { + self.config.sql_bind_style = + SqlBindStyle::from_str(&sql_bind_style).map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!( + "`{}` values are valid only if they're `{}` or `{}`", + SQL_CATALOG_PROP_BIND_STYLE, + SqlBindStyle::DollarNumeric, + SqlBindStyle::QMark + ), + ) + })?; } - } - - let valid_name = !name.trim().is_empty(); - async move { - if !valid_name { - Err(Error::new( - ErrorKind::DataInvalid, - "Catalog name cannot be empty", - )) - } else if !valid_sql_bind_style { - Err(Error::new( - ErrorKind::DataInvalid, - format!( - "`{}` values are valid only if they're `{}` or `{}`", - SQL_CATALOG_PROP_BIND_STYLE, - SqlBindStyle::DollarNumeric, - SqlBindStyle::QMark - ), - )) - } else if !valid_schema_version { - Err(Error::new( - ErrorKind::DataInvalid, - format!( - "`{}` values are valid only if they're `{}` or `{}`", - SQL_CATALOG_PROP_SCHEMA_VERSION, - SchemaVersion::V0, - SchemaVersion::V1 - ), - )) - } else { - self.config.name = name; - let runtime = match self.runtime { - Some(rt) => rt, - None => Runtime::try_current()?, - }; - let kms_client = match self.kms_client_factory { - Some(factory) => Some(factory.create_kms_client(&self.config.props).await?), - None => None, - }; - SqlCatalog::new(self.config, self.storage_factory, runtime, kms_client).await + if let Some(schema_version) = catalog_properties.schema_version { + self.config.schema_version = + Some(SchemaVersion::from_str(&schema_version).map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!( + "`{}` values are valid only if they're `{}` or `{}`", + SQL_CATALOG_PROP_SCHEMA_VERSION, + SchemaVersion::V0, + SchemaVersion::V1 + ), + ) + })?); } + catalog_properties + .props + .remove(SQL_CATALOG_PROP_SCHEMA_VERSION); + self.config.props = catalog_properties.props; + + self.config.name = name; + let runtime = match self.runtime { + Some(rt) => rt, + None => Runtime::try_current()?, + }; + let kms_client = match self.kms_client_factory { + Some(factory) => Some(factory.create_kms_client(&self.config.props).await?), + None => None, + }; + SqlCatalog::new(self.config, self.storage_factory, runtime, kms_client).await } } } +#[derive(Properties)] +struct SqlCatalogProperties { + #[property(key = SQL_CATALOG_PROP_URI, default = None)] + uri: Option, + #[property(key = SQL_CATALOG_PROP_WAREHOUSE, default = None)] + warehouse_location: Option, + #[property(key = SQL_CATALOG_PROP_BIND_STYLE, default = None)] + sql_bind_style: Option, + #[property(key = SQL_CATALOG_PROP_BIND_STYLE_LEGACY, default = None)] + legacy_sql_bind_style: Option, + #[property(key = SQL_CATALOG_PROP_SCHEMA_VERSION, default = None)] + schema_version: Option, + #[property(prefix = "")] + props: HashMap, +} + /// A struct representing the SQL catalog configuration. /// /// This struct contains various parameters that are used to configure a SQL catalog, @@ -1247,6 +1266,7 @@ mod tests { CATALOG_FIELD_RECORD_TYPE, CATALOG_TABLE_NAME, NAMESPACE_LOCATION_PROPERTY_KEY, NAMESPACE_TABLE_NAME, SQL_CATALOG_PROP_BIND_STYLE, SQL_CATALOG_PROP_BIND_STYLE_LEGACY, SQL_CATALOG_PROP_SCHEMA_VERSION, SQL_CATALOG_PROP_URI, SQL_CATALOG_PROP_WAREHOUSE, + SqlCatalogProperties, }; use crate::{SchemaVersion, SqlBindStyle, SqlCatalog, SqlCatalogBuilder}; @@ -1257,6 +1277,36 @@ mod tests { temp_dir.path().to_str().unwrap().to_string() } + #[test] + fn test_catalog_properties() { + let properties = SqlCatalogProperties::from_properties(&HashMap::from([ + ( + SQL_CATALOG_PROP_URI.to_string(), + "sqlite://catalog".to_string(), + ), + ( + SQL_CATALOG_PROP_WAREHOUSE.to_string(), + "/warehouse".to_string(), + ), + ( + SQL_CATALOG_PROP_BIND_STYLE_LEGACY.to_string(), + "?".to_string(), + ), + ( + SQL_CATALOG_PROP_SCHEMA_VERSION.to_string(), + "V1".to_string(), + ), + ("pool.max-connections".to_string(), "5".to_string()), + ])) + .unwrap(); + + assert_eq!(properties.uri.as_deref(), Some("sqlite://catalog")); + assert_eq!(properties.warehouse_location.as_deref(), Some("/warehouse")); + assert_eq!(properties.legacy_sql_bind_style.as_deref(), Some("?")); + assert_eq!(properties.schema_version.as_deref(), Some("V1")); + assert_eq!(properties.props["pool.max-connections"], "5"); + } + fn to_set(vec: Vec) -> HashSet { HashSet::from_iter(vec) }