diff --git a/Cargo.lock b/Cargo.lock index 99557339a0..4b9e7ce493 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3873,6 +3873,7 @@ version = "0.10.1" 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/public-api.txt b/crates/catalog/sql/public-api.txt index 3a7a93bdae..964a1cae5e 100644 --- a/crates/catalog/sql/public-api.txt +++ b/crates/catalog/sql/public-api.txt @@ -61,7 +61,7 @@ pub fn iceberg_catalog_sql::SqlCatalogBuilder::sql_bind_style(self, sql_bind_sty pub fn iceberg_catalog_sql::SqlCatalogBuilder::uri(self, uri: impl core::convert::Into) -> Self pub fn iceberg_catalog_sql::SqlCatalogBuilder::warehouse_location(self, location: impl core::convert::Into) -> Self impl core::default::Default for iceberg_catalog_sql::SqlCatalogBuilder -pub fn iceberg_catalog_sql::SqlCatalogBuilder::default() -> Self +pub fn iceberg_catalog_sql::SqlCatalogBuilder::default() -> iceberg_catalog_sql::SqlCatalogBuilder impl core::fmt::Debug for iceberg_catalog_sql::SqlCatalogBuilder pub fn iceberg_catalog_sql::SqlCatalogBuilder::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl iceberg::catalog::CatalogBuilder for iceberg_catalog_sql::SqlCatalogBuilder diff --git a/crates/catalog/sql/src/catalog.rs b/crates/catalog/sql/src/catalog.rs index 73671f5c4b..23588506f1 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}; @@ -77,62 +78,38 @@ static MAX_CONNECTIONS: u32 = 10; // Default the SQL pool to 10 connections if n static IDLE_TIMEOUT: u64 = 10; // Default the maximum idle timeout per connection to 10s before it is closed static TEST_BEFORE_ACQUIRE: bool = true; // Default the health-check of each connection to enabled prior to returning -fn parse_pool_property( - props: &HashMap, - property: &'static str, - default: T, -) -> Result +fn parse_pool_property(value: &str) -> Result where T: FromStr, T::Err: std::error::Error + Send + Sync + 'static, { - props.get(property).map_or(Ok(default), |value| { - value.parse().map_err(|error| { - Error::new( - ErrorKind::DataInvalid, - "Failed to parse SQL catalog pool property", - ) - .with_context("property", property) - .with_context("value", value) - .with_source(error) - }) + value.parse().map_err(|error| { + Error::new( + ErrorKind::DataInvalid, + "Failed to parse SQL catalog pool property", + ) + .with_context("value", value) + .with_source(error) }) } /// Builder for [`SqlCatalog`] -#[derive(Debug)] +#[derive(Debug, Default)] pub struct SqlCatalogBuilder { - config: SqlCatalogConfig, + props: HashMap, storage_factory: Option>, kms_client_factory: Option>, runtime: Option, } -impl Default for SqlCatalogBuilder { - fn default() -> Self { - Self { - config: SqlCatalogConfig { - uri: "".to_string(), - name: "".to_string(), - warehouse_location: "".to_string(), - sql_bind_style: SqlBindStyle::DollarNumeric, - schema_version: None, - props: HashMap::new(), - }, - storage_factory: None, - kms_client_factory: None, - runtime: None, - } - } -} - impl SqlCatalogBuilder { /// Configure the database URI /// /// If `SQL_CATALOG_PROP_URI` has a value set in `props` during `SqlCatalogBuilder::load`, /// that value takes precedence, and the value specified by this method will not be used. pub fn uri(mut self, uri: impl Into) -> Self { - self.config.uri = uri.into(); + self.props + .insert(SQL_CATALOG_PROP_URI.to_string(), uri.into()); self } @@ -141,7 +118,8 @@ impl SqlCatalogBuilder { /// If `SQL_CATALOG_PROP_WAREHOUSE` has a value set in `props` during `SqlCatalogBuilder::load`, /// that value takes precedence, and the value specified by this method will not be used. pub fn warehouse_location(mut self, location: impl Into) -> Self { - self.config.warehouse_location = location.into(); + self.props + .insert(SQL_CATALOG_PROP_WAREHOUSE.to_string(), location.into()); self } @@ -150,7 +128,10 @@ impl SqlCatalogBuilder { /// If `SQL_CATALOG_PROP_BIND_STYLE` has a value set in `props` during `SqlCatalogBuilder::load`, /// that value takes precedence, and the value specified by this method will not be used. pub fn sql_bind_style(mut self, sql_bind_style: SqlBindStyle) -> Self { - self.config.sql_bind_style = sql_bind_style; + self.props.insert( + SQL_CATALOG_PROP_BIND_STYLE.to_string(), + sql_bind_style.to_string(), + ); self } @@ -160,7 +141,7 @@ impl SqlCatalogBuilder { /// those values will take precedence. pub fn props(mut self, props: HashMap) -> Self { for (k, v) in props { - self.config.props.insert(k, v); + self.props.insert(k, v); } self } @@ -172,7 +153,7 @@ impl SqlCatalogBuilder { /// If the same key has values set in `props` during `SqlCatalogBuilder::load`, /// those values will take precedence. pub fn prop(mut self, key: impl Into, value: impl Into) -> Self { - self.config.props.insert(key.into(), value.into()); + self.props.insert(key.into(), value.into()); self } } @@ -196,61 +177,57 @@ impl CatalogBuilder for SqlCatalogBuilder { } fn load( - mut self, + self, name: impl Into, props: HashMap, ) -> impl Future> + Send { - for (k, v) in props { - 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; - - // 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)); - - // 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; + async move { + if name.trim().is_empty() { + return Err(Error::new( + ErrorKind::DataInvalid, + "Catalog name cannot be empty", + )); } - } - // 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, - } - } + let mut merged_props = self.props; + merged_props.extend(props); + let catalog_properties = SqlCatalogProperties::from_properties(&merged_props)?; - let valid_name = !name.trim().is_empty(); + 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(&merged_props).await?), + None => None, + }; + SqlCatalog::new( + name, + catalog_properties, + merged_props, + self.storage_factory, + runtime, + kms_client, + ) + .await + } + } +} - async move { - if !valid_name { - Err(Error::new( - ErrorKind::DataInvalid, - "Catalog name cannot be empty", - )) - } else if !valid_sql_bind_style { - Err(Error::new( +fn parse_sql_bind_style( + properties: &HashMap, + key: &str, + additional_keys: &[&str], + default: SqlBindStyle, +) -> Result { + properties + .get(key) + .or_else(|| additional_keys.iter().find_map(|key| properties.get(*key))) + .map_or(Ok(default), |value| { + SqlBindStyle::from_str(value).map_err(|_| { + Error::new( ErrorKind::DataInvalid, format!( "`{}` values are valid only if they're `{}` or `{}`", @@ -258,49 +235,62 @@ impl CatalogBuilder for SqlCatalogBuilder { 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 - } - } - } + ) + }) + }) } -/// A struct representing the SQL catalog configuration. -/// -/// This struct contains various parameters that are used to configure a SQL catalog, -/// such as the database URI, warehouse location, and file I/O settings. -/// You are required to provide a `SqlBindStyle`, which determines how SQL statements will be bound to values in the catalog. -/// The options available for this parameter include: -/// - `SqlBindStyle::DollarNumeric`: Binds SQL statements using `$1`, `$2`, etc., as placeholders. This is for PostgreSQL databases. -/// - `SqlBindStyle::QuestionMark`: Binds SQL statements using `?` as a placeholder. This is for MySQL and SQLite databases. -#[derive(Debug)] -struct SqlCatalogConfig { +fn parse_schema_version(value: &str) -> Result { + SchemaVersion::from_str(value).map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!( + "`{}` values are valid only if they're `{}` or `{}`", + SQL_CATALOG_PROP_SCHEMA_VERSION, + SchemaVersion::V0, + SchemaVersion::V1 + ), + ) + }) +} + +#[derive(Debug, Properties)] +pub(crate) struct SqlCatalogProperties { + #[property(key = SQL_CATALOG_PROP_URI, default = "")] uri: String, - name: String, + #[property(key = SQL_CATALOG_PROP_WAREHOUSE, default = "")] warehouse_location: String, + #[property( + key = SQL_CATALOG_PROP_BIND_STYLE, + additional_keys = [SQL_CATALOG_PROP_BIND_STYLE_LEGACY], + default = SqlBindStyle::DollarNumeric, + parse_properties_with = parse_sql_bind_style + )] sql_bind_style: SqlBindStyle, + #[property( + key = SQL_CATALOG_PROP_SCHEMA_VERSION, + default = None, + parse_with = parse_schema_version + )] schema_version: Option, - props: HashMap, + #[property( + key = "pool.max-connections", + default = MAX_CONNECTIONS, + parse_with = parse_pool_property + )] + max_connections: u32, + #[property( + key = "pool.idle-timeout", + default = IDLE_TIMEOUT, + parse_with = parse_pool_property + )] + idle_timeout: u64, + #[property( + key = "pool.test-before-acquire", + default = TEST_BEFORE_ACQUIRE, + parse_with = parse_pool_property + )] + test_before_acquire: bool, } #[derive(Debug)] @@ -310,10 +300,9 @@ struct SqlCatalogConfig { /// Catalogs can opt-in to automatic migration by configuring the `sql.schema-version` catalog property. pub struct SqlCatalog { name: String, + properties: SqlCatalogProperties, connection: AnyPool, - warehouse_location: String, fileio: FileIO, - sql_bind_style: SqlBindStyle, runtime: Runtime, kms_client: Option>, schema_version: SchemaVersion, @@ -390,7 +379,9 @@ pub enum SqlBindStyle { impl SqlCatalog { /// Create new sql catalog instance async fn new( - config: SqlCatalogConfig, + name: String, + properties: SqlCatalogProperties, + props: HashMap, storage_factory: Option>, runtime: Runtime, kms_client: Option>, @@ -401,27 +392,16 @@ impl SqlCatalog { "StorageFactory must be provided for SqlCatalog. Use `with_storage_factory` to configure it.", ) })?; - // Forward catalog props so storage-backend keys reach the FileIO. - // Unrecognized keys are ignored by backends. - let fileio = FileIOBuilder::new(factory) - .with_props(config.props.clone()) - .build(); - install_default_drivers(); - let max_connections = - parse_pool_property(&config.props, "pool.max-connections", MAX_CONNECTIONS)?; - let idle_timeout = parse_pool_property(&config.props, "pool.idle-timeout", IDLE_TIMEOUT)?; - let test_before_acquire = parse_pool_property( - &config.props, - "pool.test-before-acquire", - TEST_BEFORE_ACQUIRE, - )?; + // Forward the complete property map so storage-backend keys reach FileIO. + // Unrecognized keys are ignored by backends. + let fileio = FileIOBuilder::new(factory).with_props(props).build(); let pool = AnyPoolOptions::new() - .max_connections(max_connections) - .idle_timeout(Duration::from_secs(idle_timeout)) - .test_before_acquire(test_before_acquire) - .connect(&config.uri) + .max_connections(properties.max_connections) + .idle_timeout(Duration::from_secs(properties.idle_timeout)) + .test_before_acquire(properties.test_before_acquire) + .connect(&properties.uri) .await .map_err(from_sqlx_error)?; @@ -452,7 +432,7 @@ impl SqlCatalog { .map_err(from_sqlx_error)?; let detected_schema_version = SchemaVersion::detect(&pool).await?; - let expected_schema_version = config.schema_version; + let expected_schema_version = properties.schema_version; // Detect schema by describing columns. If expected is configured then automigrate, otherwise gracefully support older schemas. let schema_version = match (detected_schema_version, expected_schema_version) { @@ -498,11 +478,10 @@ impl SqlCatalog { }; Ok(SqlCatalog { - name: config.name.to_owned(), + name, + properties, connection: pool, - warehouse_location: config.warehouse_location, fileio, - sql_bind_style: config.sql_bind_style, runtime, kms_client, schema_version, @@ -511,7 +490,7 @@ impl SqlCatalog { /// SQLX Any does not implement PostgresSQL bindings, so we have to do this. fn replace_placeholders(&self, query: &str) -> String { - match self.sql_bind_style { + match self.properties.sql_bind_style { SqlBindStyle::DollarNumeric => { let mut count = 1; query @@ -1047,7 +1026,7 @@ impl Catalog for SqlCatalog { None => { format!( "{}/{}", - self.warehouse_location.clone(), + self.properties.warehouse_location.clone(), namespace.join("/") ) } @@ -1244,9 +1223,10 @@ mod tests { use tempfile::TempDir; use crate::catalog::{ - 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, + CATALOG_FIELD_RECORD_TYPE, CATALOG_TABLE_NAME, IDLE_TIMEOUT, MAX_CONNECTIONS, + 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, TEST_BEFORE_ACQUIRE, }; use crate::{SchemaVersion, SqlBindStyle, SqlCatalog, SqlCatalogBuilder}; @@ -1257,6 +1237,53 @@ 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(), + SqlBindStyle::QMark.to_string(), + ), + ( + SQL_CATALOG_PROP_SCHEMA_VERSION.to_string(), + "V1".to_string(), + ), + ("pool.max-connections".to_string(), "5".to_string()), + ("pool.idle-timeout".to_string(), "20".to_string()), + ("pool.test-before-acquire".to_string(), "false".to_string()), + ])) + .unwrap(); + + assert_eq!(properties.uri, "sqlite://catalog"); + assert_eq!(properties.warehouse_location, "/warehouse"); + assert_eq!(properties.sql_bind_style, SqlBindStyle::QMark); + assert_eq!(properties.schema_version, Some(SchemaVersion::V1)); + assert_eq!(properties.max_connections, 5); + assert_eq!(properties.idle_timeout, 20); + assert!(!properties.test_before_acquire); + } + + #[test] + fn test_catalog_properties_defaults() { + let properties = SqlCatalogProperties::from_properties(&HashMap::new()).unwrap(); + + assert_eq!(properties.uri, ""); + assert_eq!(properties.warehouse_location, ""); + assert_eq!(properties.sql_bind_style, SqlBindStyle::DollarNumeric); + assert_eq!(properties.schema_version, None); + assert_eq!(properties.max_connections, MAX_CONNECTIONS); + assert_eq!(properties.idle_timeout, IDLE_TIMEOUT); + assert_eq!(properties.test_before_acquire, TEST_BEFORE_ACQUIRE); + } + fn to_set(vec: Vec) -> HashSet { HashSet::from_iter(vec) } @@ -1511,8 +1538,8 @@ mod tests { assert!(catalog.is_ok()); let catalog = catalog.unwrap(); - assert!(catalog.warehouse_location == warehouse_location); - assert!(catalog.sql_bind_style == SqlBindStyle::QMark); + assert!(catalog.properties.warehouse_location == warehouse_location); + assert!(catalog.properties.sql_bind_style == SqlBindStyle::QMark); } /// Overwriting an sqlite database with a non-existent path causes @@ -1589,8 +1616,8 @@ mod tests { assert!(catalog.is_ok()); let catalog = catalog.unwrap(); - assert!(catalog.warehouse_location == warehouse_location); - assert!(catalog.sql_bind_style == SqlBindStyle::QMark); + assert!(catalog.properties.warehouse_location == warehouse_location); + assert!(catalog.properties.sql_bind_style == SqlBindStyle::QMark); } /// values assigned via props take precedence @@ -1634,8 +1661,8 @@ mod tests { assert!(catalog.is_ok()); let catalog = catalog.unwrap(); - assert!(catalog.warehouse_location == warehouse_location); - assert!(catalog.sql_bind_style == SqlBindStyle::QMark); + assert!(catalog.properties.warehouse_location == warehouse_location); + assert!(catalog.properties.sql_bind_style == SqlBindStyle::QMark); } /// values assigned via props take precedence @@ -1673,8 +1700,8 @@ mod tests { assert!(catalog.is_ok()); let catalog = catalog.unwrap(); - assert!(catalog.warehouse_location == warehouse_location); - assert!(catalog.sql_bind_style == SqlBindStyle::QMark); + assert!(catalog.properties.warehouse_location == warehouse_location); + assert!(catalog.properties.sql_bind_style == SqlBindStyle::QMark); } /// invalid value for `SqlBindStyle` causes catalog creation to fail @@ -2883,6 +2910,6 @@ mod tests { .await .expect("legacy sql_bind_style key should still be accepted"); - assert_eq!(catalog.sql_bind_style, SqlBindStyle::QMark); + assert_eq!(catalog.properties.sql_bind_style, SqlBindStyle::QMark); } }