diff --git a/Cargo.lock b/Cargo.lock index 3802cc03ff..3e0dc95996 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3923,6 +3923,7 @@ dependencies = [ "aws-sdk-s3tables", "futures", "iceberg", + "iceberg-property-macro", "iceberg-storage-opendal", "iceberg_test_utils", "itertools 0.13.0", diff --git a/crates/catalog/s3tables/Cargo.toml b/crates/catalog/s3tables/Cargo.toml index f7bfa06114..12dabc9b74 100644 --- a/crates/catalog/s3tables/Cargo.toml +++ b/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] diff --git a/crates/catalog/s3tables/src/catalog.rs b/crates/catalog/s3tables/src/catalog.rs index 73127bd903..e0a340b1c8 100644 --- a/crates/catalog/s3tables/src/catalog.rs +++ b/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, + #[property(key = S3TABLES_CATALOG_PROP_ENDPOINT_URL, default = None)] + endpoint_url: Option, + #[property(prefix = "")] + props: HashMap, +} + /// 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> { let table_bucket_arn = match std::env::var("TABLE_BUCKET_ARN").ok() { Some(table_bucket_arn) => table_bucket_arn,