feat(sink): support delta sink with gcs

Cargo.toml

@@ -148,6 +148,7 @@ arrow-schema-deltalake = { package = "arrow-schema", version = "48.0.1" }
 # compile time and binary size.
 deltalake = { git = "https://github.com/risingwavelabs/delta-rs", rev = "5c2dccd4640490202ffe98adbd13b09cef8e007b", features = [
   "s3-no-concurrent-write",
+  "gcs",
 ] }
 lru = { git = "https://github.com/risingwavelabs/lru-rs.git", rev = "2682b85" }
 parquet = "50"

src/connector/src/sink/deltalake.rs

@@ -48,6 +48,7 @@ use crate::sink::writer::SinkWriterExt;
 
 pub const DELTALAKE_SINK: &str = "deltalake";
 pub const DEFAULT_REGION: &str = "us-east-1";
+pub const GCS_SERVICE_ACCOUNT: &str = "service_account_key";
 
 #[derive(Deserialize, Serialize, Debug, Clone, WithOptions)]
 pub struct DeltaLakeCommon {
@@ -61,6 +62,8 @@ pub struct DeltaLakeCommon {
     pub s3_region: Option<String>,
     #[serde(rename = "s3.endpoint")]
     pub s3_endpoint: Option<String>,
+    #[serde(rename = "gcs.service.account")]
+    pub gcs_service_account: Option<String>,
 }
 impl DeltaLakeCommon {
     pub async fn create_deltalake_client(&self) -> Result<DeltaTable> {
@@ -98,18 +101,29 @@ impl DeltaLakeCommon {
                 deltalake::open_table_with_storage_options(s3_path.clone(), storage_options).await?
             }
             DeltaTableUrl::Local(local_path) => deltalake::open_table(local_path).await?,
+            DeltaTableUrl::Gcs(gcs_path) => {
+                let mut storage_options = HashMap::new();
+                storage_options.insert(
+                    GCS_SERVICE_ACCOUNT.to_string(),
+                    self.gcs_service_account.clone().unwrap(),
+                );
+                deltalake::open_table_with_storage_options(gcs_path.clone(), storage_options)
+                    .await?
+            }
         };
         Ok(table)
     }
 
     fn get_table_url(path: &str) -> Result<DeltaTableUrl> {
         if path.starts_with("s3://") || path.starts_with("s3a://") {
             Ok(DeltaTableUrl::S3(path.to_string()))
+        } else if path.starts_with("gs://") {
+            Ok(DeltaTableUrl::Gcs(path.to_string()))
         } else if let Some(path) = path.strip_prefix("file://") {
             Ok(DeltaTableUrl::Local(path.to_string()))
         } else {
             Err(SinkError::DeltaLake(anyhow!(
-                "path need to start with 's3://','s3a://'(s3) or file://(local)"
+                "path need to start with 's3://','s3a://'(s3) ,gs://(gcs) or file://(local)"
             )))
         }
     }
@@ -118,6 +132,7 @@ impl DeltaLakeCommon {
 enum DeltaTableUrl {
     S3(String),
     Local(String),
+    Gcs(String),
 }
 
 #[serde_as]

src/connector/with_options_sink.yaml

@@ -87,6 +87,9 @@ DeltaLakeConfig:
   - name: s3.endpoint
     field_type: String
     required: false
+  - name: gcs.service.account
+    field_type: String
+    required: false
   - name: r#type
     field_type: String
     required: true