refactor(s3tables): derive catalog configuration from properties

crates/catalog/s3tables/Cargo.toml

@@ -36,6 +36,7 @@ async-trait = { workspace = true }
 aws-config = { workspace = true }
 aws-sdk-s3tables = { workspace = true }
 iceberg = { workspace = true }
+iceberg-property-macro = { workspace = true }
 iceberg-storage-opendal = { workspace = true, features = ["opendal-s3"] }
 
 [dev-dependencies]

crates/catalog/s3tables/src/catalog.rs

@@ -35,6 +35,7 @@ use iceberg::{
     Catalog, CatalogBuilder, Error, ErrorKind, MetadataLocation, Namespace, NamespaceIdent, Result,
     Runtime, TableCommit, TableCreation, TableIdent,
 };
+use iceberg_property_macro::Properties;
 use iceberg_storage_opendal::OpenDalStorageFactory;
 
 use crate::utils::create_sdk_config;
@@ -155,27 +156,22 @@ impl CatalogBuilder for S3TablesCatalogBuilder {
         let catalog_name = name.into();
         self.config.name = Some(catalog_name.clone());
 
-        if props.contains_key(S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN) {
-            self.config.table_bucket_arn = props
-                .get(S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN)
-                .cloned()
-                .unwrap_or_default();
-        }
-
-        if props.contains_key(S3TABLES_CATALOG_PROP_ENDPOINT_URL) {
-            self.config.endpoint_url = props.get(S3TABLES_CATALOG_PROP_ENDPOINT_URL).cloned();
-        }
-
-        // Collect other remaining properties
-        self.config.props = props
-            .into_iter()
-            .filter(|(k, _)| {
-                k != S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN
-                    && k != S3TABLES_CATALOG_PROP_ENDPOINT_URL
-            })
-            .collect();
-
         async move {
+            let mut catalog_properties = S3TablesCatalogProperties::from_properties(&props)?;
+            for property in [
+                S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN,
+                S3TABLES_CATALOG_PROP_ENDPOINT_URL,
+            ] {
+                catalog_properties.props.remove(property);
+            }
+            if let Some(table_bucket_arn) = catalog_properties.table_bucket_arn {
+                self.config.table_bucket_arn = table_bucket_arn;
+            }
+            if let Some(endpoint_url) = catalog_properties.endpoint_url {
+                self.config.endpoint_url = Some(endpoint_url);
+            }
+            self.config.props = catalog_properties.props;
+
             if catalog_name.trim().is_empty() {
                 Err(Error::new(
                     ErrorKind::DataInvalid,
@@ -201,6 +197,16 @@ impl CatalogBuilder for S3TablesCatalogBuilder {
     }
 }
 
+#[derive(Properties)]
+struct S3TablesCatalogProperties {
+    #[property(key = S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN, default = None)]
+    table_bucket_arn: Option<String>,
+    #[property(key = S3TABLES_CATALOG_PROP_ENDPOINT_URL, default = None)]
+    endpoint_url: Option<String>,
+    #[property(prefix = "")]
+    props: HashMap<String, String>,
+}
+
 /// S3Tables catalog implementation.
 #[derive(Debug)]
 pub struct S3TablesCatalog {
@@ -765,6 +771,29 @@ mod tests {
 
     use super::*;
 
+    #[test]
+    fn test_catalog_properties() {
+        let properties = S3TablesCatalogProperties::from_properties(&HashMap::from([
+            (
+                S3TABLES_CATALOG_PROP_TABLE_BUCKET_ARN.to_string(),
+                "arn:aws:s3tables:us-east-1:123456789012:bucket/test".to_string(),
+            ),
+            (
+                S3TABLES_CATALOG_PROP_ENDPOINT_URL.to_string(),
+                "http://localhost".to_string(),
+            ),
+            ("region_name".to_string(), "us-east-1".to_string()),
+        ]))
+        .unwrap();
+
+        assert_eq!(
+            properties.table_bucket_arn.as_deref(),
+            Some("arn:aws:s3tables:us-east-1:123456789012:bucket/test")
+        );
+        assert_eq!(properties.endpoint_url.as_deref(), Some("http://localhost"));
+        assert_eq!(properties.props["region_name"], "us-east-1");
+    }
+
     async fn load_s3tables_catalog_from_env() -> Result<Option<S3TablesCatalog>> {
         let table_bucket_arn = match std::env::var("TABLE_BUCKET_ARN").ok() {
             Some(table_bucket_arn) => table_bucket_arn,