This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3102-ba849a6ed3030c237e46be3e1d39e67cebf035a7 in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit b3470c8463c281c6e00ea9eee402881bb6c6d1fd Author: blackmwk <[email protected]> AuthorDate: Thu Oct 15 18:12:58 2026 +0000 refactor(glue): derive catalog configuration from properties (#3102) * refactor(glue): derive catalog configuration from properties Use the shared property derive for Glue catalog settings while preserving warehouse validation and forwarding AWS and storage properties. Tracks [apache/iceberg-rust#3096](https://github.com/apache/iceberg-rust/issues/3096). Generated-by: Codex * refactor(glue): simplify catalog builder and tests Derive Default for GlueCatalogBuilder, use the conventional tests module name, and remove the FileIO forwarding test requested in review. Generated-by: Codex --- Cargo.lock | 1 + crates/catalog/glue/Cargo.toml | 1 + crates/catalog/glue/public-api.txt | 2 +- crates/catalog/glue/src/catalog.rs | 187 ++++++++++++++++++++----------------- 4 files changed, 106 insertions(+), 85 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b98bd2da4..3bb94e339 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2779,6 +2779,7 @@ dependencies = [ "aws-config", "aws-sdk-glue", "iceberg", + "iceberg-property-macro", "iceberg-storage-opendal", "iceberg_test_utils", "serde_json", diff --git a/crates/catalog/glue/Cargo.toml b/crates/catalog/glue/Cargo.toml index decb78134..ac912bd40 100644 --- a/crates/catalog/glue/Cargo.toml +++ b/crates/catalog/glue/Cargo.toml @@ -35,6 +35,7 @@ async-trait = { workspace = true } aws-config = { workspace = true } aws-sdk-glue = { workspace = true } iceberg = { workspace = true } +iceberg-property-macro = { workspace = true } iceberg-storage-opendal = { workspace = true, features = ["opendal-s3"] } serde_json = { workspace = true } tokio = { workspace = true } diff --git a/crates/catalog/glue/public-api.txt b/crates/catalog/glue/public-api.txt index 902daf635..eafce2054 100644 --- a/crates/catalog/glue/public-api.txt +++ b/crates/catalog/glue/public-api.txt @@ -23,7 +23,7 @@ pub fn iceberg_catalog_glue::GlueCatalog::update_namespace<'life0, 'life1, 'asyn pub fn iceberg_catalog_glue::GlueCatalog::update_table<'life0, 'async_trait>(&'life0 self, commit: iceberg::catalog::TableCommit) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<iceberg::table::Table>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub struct iceberg_catalog_glue::GlueCatalogBuilder impl core::default::Default for iceberg_catalog_glue::GlueCatalogBuilder -pub fn iceberg_catalog_glue::GlueCatalogBuilder::default() -> Self +pub fn iceberg_catalog_glue::GlueCatalogBuilder::default() -> iceberg_catalog_glue::GlueCatalogBuilder impl core::fmt::Debug for iceberg_catalog_glue::GlueCatalogBuilder pub fn iceberg_catalog_glue::GlueCatalogBuilder::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl iceberg::catalog::CatalogBuilder for iceberg_catalog_glue::GlueCatalogBuilder diff --git a/crates/catalog/glue/src/catalog.rs b/crates/catalog/glue/src/catalog.rs index 721fbb548..06e8b8400 100644 --- a/crates/catalog/glue/src/catalog.rs +++ b/crates/catalog/glue/src/catalog.rs @@ -36,6 +36,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::error::{from_aws_build_error, from_aws_sdk_error}; @@ -55,31 +56,13 @@ pub const GLUE_CATALOG_PROP_CATALOG_ID: &str = "catalog_id"; pub const GLUE_CATALOG_PROP_WAREHOUSE: &str = "warehouse"; /// Builder for [`GlueCatalog`]. -#[derive(Debug)] +#[derive(Debug, Default)] pub struct GlueCatalogBuilder { - config: GlueCatalogConfig, storage_factory: Option<Arc<dyn StorageFactory>>, kms_client_factory: Option<Arc<dyn KmsClientFactory>>, runtime: Option<Runtime>, } -impl Default for GlueCatalogBuilder { - fn default() -> Self { - Self { - config: GlueCatalogConfig { - name: None, - uri: None, - catalog_id: None, - warehouse: "".to_string(), - props: HashMap::new(), - }, - storage_factory: None, - kms_client_factory: None, - runtime: None, - } - } -} - impl CatalogBuilder for GlueCatalogBuilder { type C = GlueCatalog; @@ -99,45 +82,15 @@ impl CatalogBuilder for GlueCatalogBuilder { } fn load( - mut self, + self, name: impl Into<String>, props: HashMap<String, String>, ) -> impl Future<Output = Result<Self::C>> + Send { - self.config.name = Some(name.into()); - - if props.contains_key(GLUE_CATALOG_PROP_URI) { - self.config.uri = props.get(GLUE_CATALOG_PROP_URI).cloned() - } - - if props.contains_key(GLUE_CATALOG_PROP_CATALOG_ID) { - self.config.catalog_id = props.get(GLUE_CATALOG_PROP_CATALOG_ID).cloned() - } - - if props.contains_key(GLUE_CATALOG_PROP_WAREHOUSE) { - self.config.warehouse = props - .get(GLUE_CATALOG_PROP_WAREHOUSE) - .cloned() - .unwrap_or_default(); - } - - // Collect other remaining properties - self.config.props = props - .into_iter() - .filter(|(k, _)| { - k != GLUE_CATALOG_PROP_URI - && k != GLUE_CATALOG_PROP_CATALOG_ID - && k != GLUE_CATALOG_PROP_WAREHOUSE - }) - .collect(); + let name = name.into(); async move { - if self.config.name.is_none() { - return Err(Error::new( - ErrorKind::DataInvalid, - "Catalog name is required", - )); - } - if self.config.warehouse.is_empty() { + let catalog_properties = GlueCatalogProperties::from_properties(&props)?; + if catalog_properties.warehouse.is_empty() { return Err(Error::new( ErrorKind::DataInvalid, "Catalog warehouse is required", @@ -149,29 +102,38 @@ impl CatalogBuilder for GlueCatalogBuilder { None => Runtime::try_current()?, }; let kms_client = match self.kms_client_factory { - Some(factory) => Some(factory.create_kms_client(&self.config.props).await?), + Some(factory) => Some(factory.create_kms_client(&props).await?), None => None, }; - GlueCatalog::new(self.config, self.storage_factory, runtime, kms_client).await + GlueCatalog::new( + name, + catalog_properties, + props, + self.storage_factory, + runtime, + kms_client, + ) + .await } } } -#[derive(Debug)] -/// Glue Catalog configuration -pub(crate) struct GlueCatalogConfig { - name: Option<String>, +#[derive(Debug, Properties)] +pub(crate) struct GlueCatalogProperties { + #[property(key = GLUE_CATALOG_PROP_URI, default = None)] uri: Option<String>, + #[property(key = GLUE_CATALOG_PROP_CATALOG_ID, default = None)] catalog_id: Option<String>, + #[property(key = GLUE_CATALOG_PROP_WAREHOUSE, default = "")] warehouse: String, - props: HashMap<String, String>, } struct GlueClient(aws_sdk_glue::Client); /// Glue Catalog pub struct GlueCatalog { - config: GlueCatalogConfig, + name: String, + properties: GlueCatalogProperties, client: GlueClient, file_io: FileIO, runtime: Runtime, @@ -181,7 +143,8 @@ pub struct GlueCatalog { impl Debug for GlueCatalog { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("GlueCatalog") - .field("config", &self.config) + .field("name", &self.name) + .field("properties", &self.properties) .finish_non_exhaustive() } } @@ -189,13 +152,15 @@ impl Debug for GlueCatalog { impl GlueCatalog { /// Create a new glue catalog async fn new( - config: GlueCatalogConfig, + name: String, + properties: GlueCatalogProperties, + props: HashMap<String, String>, storage_factory: Option<Arc<dyn StorageFactory>>, runtime: Runtime, kms_client: Option<Arc<dyn KeyManagementClient>>, ) -> Result<Self> { - let sdk_config = create_sdk_config(&config.props, config.uri.as_ref()).await; - let mut file_io_props = config.props.clone(); + let sdk_config = create_sdk_config(&props, properties.uri.as_ref()).await; + let mut file_io_props = props; if !file_io_props.contains_key(S3_ACCESS_KEY_ID) && let Some(access_key_id) = file_io_props.get(AWS_ACCESS_KEY_ID) { @@ -220,7 +185,7 @@ impl GlueCatalog { file_io_props.insert(S3_SESSION_TOKEN.to_string(), session_token.to_string()); } if !file_io_props.contains_key(S3_ENDPOINT) - && let Some(aws_endpoint) = config.uri.as_ref() + && let Some(aws_endpoint) = properties.uri.as_ref() { file_io_props.insert(S3_ENDPOINT.to_string(), aws_endpoint.to_string()); } @@ -238,7 +203,8 @@ impl GlueCatalog { .build(); Ok(GlueCatalog { - config, + name, + properties, client: GlueClient(client), file_io, runtime, @@ -275,7 +241,7 @@ impl GlueCatalog { .get_table() .database_name(&db_name) .name(table_name); - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); let glue_table_output = builder.send().await.map_err(from_aws_sdk_error)?; @@ -333,7 +299,7 @@ impl Catalog for GlueCatalog { Some(token) => self.client.0.get_databases().next_token(token), None => self.client.0.get_databases(), }; - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); let resp = builder.send().await.map_err(from_aws_sdk_error)?; let dbs: Vec<NamespaceIdent> = resp @@ -381,7 +347,7 @@ impl Catalog for GlueCatalog { let db_input = convert_to_database(namespace, &properties)?; let builder = self.client.0.create_database().database_input(db_input); - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); builder.send().await.map_err(from_aws_sdk_error)?; @@ -402,7 +368,7 @@ impl Catalog for GlueCatalog { let db_name = validate_namespace(namespace)?; let builder = self.client.0.get_database().name(&db_name); - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); let resp = builder.send().await.map_err(|err| { if err @@ -446,7 +412,7 @@ impl Catalog for GlueCatalog { let db_name = validate_namespace(namespace)?; let builder = self.client.0.get_database().name(&db_name); - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); let resp = builder.send().await; @@ -496,7 +462,7 @@ impl Catalog for GlueCatalog { .update_database() .name(&db_name) .database_input(db_input); - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); builder.send().await.map_err(from_aws_sdk_error)?; @@ -533,7 +499,7 @@ impl Catalog for GlueCatalog { .get_tables() .database_name(&db_name) .max_results(1); - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); let resp = builder.send().await.map_err(from_aws_sdk_error)?; if !resp.table_list().is_empty() { @@ -544,7 +510,7 @@ impl Catalog for GlueCatalog { } let builder = self.client.0.delete_database().name(db_name); - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); builder.send().await.map_err(from_aws_sdk_error)?; @@ -579,7 +545,7 @@ impl Catalog for GlueCatalog { .next_token(token), None => self.client.0.get_tables().database_name(&db_name), }; - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); let resp = builder.send().await.map_err(from_aws_sdk_error)?; let tables: Vec<_> = resp @@ -623,7 +589,7 @@ impl Catalog for GlueCatalog { if creation.location.is_none() { let ns = self.get_namespace(namespace).await?; let location = - get_default_table_location(&ns, &db_name, &table_name, &self.config.warehouse); + get_default_table_location(&ns, &db_name, &table_name, &self.properties.warehouse); creation.location = Some(location); } let metadata = TableMetadataBuilder::from_table_creation(creation)? @@ -648,7 +614,7 @@ impl Catalog for GlueCatalog { .create_table() .database_name(&db_name) .table_input(glue_table); - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); builder.send().await.map_err(from_aws_sdk_error)?; @@ -701,7 +667,7 @@ impl Catalog for GlueCatalog { .delete_table() .database_name(&db_name) .name(table_name); - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); builder.send().await.map_err(from_aws_sdk_error)?; @@ -731,7 +697,7 @@ impl Catalog for GlueCatalog { .get_table() .database_name(&db_name) .name(table_name); - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); let resp = builder.send().await; @@ -769,7 +735,7 @@ impl Catalog for GlueCatalog { .get_table() .database_name(&src_db_name) .name(src_table_name); - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); let glue_table_output = builder.send().await.map_err(from_aws_sdk_error)?; @@ -796,7 +762,7 @@ impl Catalog for GlueCatalog { .create_table() .database_name(&dest_db_name) .table_input(rename_table_input); - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); builder.send().await.map_err(from_aws_sdk_error)?; @@ -864,7 +830,7 @@ impl Catalog for GlueCatalog { .create_table() .database_name(&db_name) .table_input(table_input); - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); builder.send().await.map_err(|e| { let error = e.into_service_error(); @@ -935,7 +901,7 @@ impl Catalog for GlueCatalog { builder = builder.version_id(version_id); } - let builder = with_catalog_id!(builder, self.config); + let builder = with_catalog_id!(builder, self.properties); let _ = builder.send().await.map_err(|e| { let error = e.into_service_error(); match error { @@ -959,3 +925,56 @@ impl Catalog for GlueCatalog { Ok(staged_table) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_catalog_properties() { + let properties = GlueCatalogProperties::from_properties(&HashMap::from([ + ( + GLUE_CATALOG_PROP_URI.to_string(), + "http://localhost".to_string(), + ), + ( + GLUE_CATALOG_PROP_CATALOG_ID.to_string(), + "catalog".to_string(), + ), + ( + GLUE_CATALOG_PROP_WAREHOUSE.to_string(), + "s3://warehouse".to_string(), + ), + ])) + .unwrap(); + + assert_eq!(properties.uri.as_deref(), Some("http://localhost";)); + assert_eq!(properties.catalog_id.as_deref(), Some("catalog")); + assert_eq!(properties.warehouse, "s3://warehouse"); + } + + #[test] + fn test_catalog_properties_defaults() { + let properties = GlueCatalogProperties::from_properties(&HashMap::new()).unwrap(); + + assert_eq!(properties.uri, None); + assert_eq!(properties.catalog_id, None); + assert_eq!(properties.warehouse, ""); + } + + #[tokio::test] + async fn test_load_rejects_missing_or_empty_warehouse() { + for props in [ + HashMap::new(), + HashMap::from([(GLUE_CATALOG_PROP_WAREHOUSE.to_string(), String::new())]), + ] { + let error = GlueCatalogBuilder::default() + .load("glue", props) + .await + .unwrap_err(); + + assert_eq!(error.kind(), ErrorKind::DataInvalid); + assert_eq!(error.message(), "Catalog warehouse is required"); + } + } +}
