refactor(sql): derive catalog configuration from properties

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 }

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<String>,
+    #[property(key = SQL_CATALOG_PROP_WAREHOUSE, default = None)]
+    warehouse_location: Option<String>,
+    #[property(key = SQL_CATALOG_PROP_BIND_STYLE, default = None)]
+    sql_bind_style: Option<String>,
+    #[property(key = SQL_CATALOG_PROP_BIND_STYLE_LEGACY, default = None)]
+    legacy_sql_bind_style: Option<String>,
+    #[property(key = SQL_CATALOG_PROP_SCHEMA_VERSION, default = None)]
+    schema_version: Option<String>,
+    #[property(prefix = "")]
+    props: HashMap<String, String>,
+}
+
 /// 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<T: Eq + Hash>(vec: Vec<T>) -> HashSet<T> {
         HashSet::from_iter(vec)
     }