add 'glue.id' to iceberg sink config

risingwavelabs · Jan 27, 2025 · 0c3fa31 · 0c3fa31
1 parent 0f5bae0
commit 0c3fa31
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 0 deletions.
diff --git a/src/connector/src/connector_common/connection.rs b/src/connector/src/connector_common/connection.rs
@@ -135,6 +135,10 @@ pub struct IcebergConnection {
     /// Path of iceberg warehouse, only applicable in storage catalog.
     #[serde(rename = "warehouse.path")]
     pub warehouse_path: Option<String>,
+    /// Catalog id, can be omitted for storage catalog or when
+    /// caller's AWS account ID matches catalog id
+    #[serde(rename = "catalog.id")]
+    pub catalog_id: Option<String>,
     /// Catalog name, can be omitted for storage catalog, but
     /// must be set for other catalogs.
     #[serde(rename = "catalog.name")]
@@ -255,6 +259,7 @@ impl Connection for IcebergConnection {
             secret_key: self.secret_key.clone(),
             gcs_credential: self.gcs_credential.clone(),
             warehouse_path: self.warehouse_path.clone(),
+            catalog_id: self.catalog_id.clone(),
             catalog_name: self.catalog_name.clone(),
             catalog_uri: self.catalog_uri.clone(),
             credential: self.credential.clone(),

diff --git a/src/connector/src/connector_common/iceberg/mod.rs b/src/connector/src/connector_common/iceberg/mod.rs
@@ -57,6 +57,10 @@ pub struct IcebergCommon {
     /// Path of iceberg warehouse, only applicable in storage catalog.
     #[serde(rename = "warehouse.path")]
     pub warehouse_path: Option<String>,
+    /// Catalog id, can be omitted for storage catalog or when
+    /// caller's AWS account ID matches catalog id
+    #[serde(rename = "catalog.id")]
+    pub catalog_id: Option<String>,
     /// Catalog name, can be omitted for storage catalog, but
     /// must be set for other catalogs.
     #[serde(rename = "catalog.name")]
@@ -274,6 +278,10 @@ impl IcebergCommon {
                             format!("https://glue.{}.amazonaws.com", region),
                         );
                     }
+
+                    if let Some(catalog_id) = self.catalog_id.as_deref() {
+                        java_catalog_configs.insert("glue.id".to_owned(), catalog_id.to_owned());
+                    }
                 }
                 _ => {}
             }

diff --git a/src/connector/src/sink/iceberg/mod.rs b/src/connector/src/sink/iceberg/mod.rs
@@ -1432,6 +1432,7 @@ mod test {
                 secret_key: Some("hummockadmin".to_owned()),
                 gcs_credential: None,
                 catalog_type: Some("jdbc".to_owned()),
+                catalog_id: None,
                 catalog_name: Some("demo".to_owned()),
                 database_name: Some("demo_db".to_owned()),
                 table_name: "demo_table".to_owned(),