simplify sink

risingwavelabs · Sep 19, 2024 · ff19cec · ff19cec
1 parent 28b88c6
commit ff19cec
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 21 deletions.
diff --git a/src/connector/src/sink/iceberg/mod.rs b/src/connector/src/sink/iceberg/mod.rs
@@ -76,7 +76,7 @@ pub struct IcebergConfig {
     pub force_append_only: bool,
 
     #[serde(flatten)]
-    pub common: IcebergCommon,
+    common: IcebergCommon,
 
     #[serde(
         rename = "s3.path.style.access",
@@ -106,6 +106,13 @@ pub struct IcebergConfig {
 }
 
 impl IcebergConfig {
+    pub async fn load_table(&self) -> Result<Table> {
+        self.common
+            .load_table(&self.path_style_access, &self.java_catalog_props)
+            .await
+            .map_err(|e| SinkError::Iceberg(anyhow!(e)))
+    }
+
     pub fn from_btreemap(values: BTreeMap<String, String>) -> Result<Self> {
         let mut config =
             serde_json::from_value::<IcebergConfig>(serde_json::to_value(&values).unwrap())
@@ -198,11 +205,7 @@ impl IcebergSink {
 
         let table = self
             .config
-            .common
-            .load_table(
-                &self.config.path_style_access,
-                &self.config.java_catalog_props,
-            )
+            .load_table()
             .await
             .map_err(|err| SinkError::Iceberg(anyhow!(err)))?;
 
@@ -970,14 +973,7 @@ mod test {
     async fn test_create_catalog(configs: BTreeMap<String, String>) {
         let iceberg_config = IcebergConfig::from_btreemap(configs).unwrap();
 
-        let table = iceberg_config
-            .common
-            .load_table(
-                &iceberg_config.path_style_access,
-                &iceberg_config.java_catalog_props,
-            )
-            .await
-            .unwrap();
+        let table = iceberg_config.load_table().await.unwrap();
 
         println!("{:?}", table.table_name());
     }

diff --git a/src/frontend/src/handler/create_sink.rs b/src/frontend/src/handler/create_sink.rs
@@ -345,13 +345,7 @@ async fn get_partition_compute_info_for_iceberg(
     if iceberg_config.create_table_if_not_exists {
         return Ok(None);
     }
-    let table = iceberg_config
-        .common
-        .load_table(
-            &iceberg_config.path_style_access,
-            &iceberg_config.java_catalog_props,
-        )
-        .await?;
+    let table = iceberg_config.load_table().await?;
     let Some(partition_spec) = table.current_table_metadata().current_partition_spec().ok() else {
         return Ok(None);
     };