feat: Google Cloud Storage as storage backend for databend

src/common/config/src/outer_v0.rs

@@ -24,6 +24,7 @@ use common_exception::Result;
 use common_storage::StorageAzblobConfig as InnerStorageAzblobConfig;
 use common_storage::StorageConfig as InnerStorageConfig;
 use common_storage::StorageFsConfig as InnerStorageFsConfig;
+use common_storage::StorageGcsConfig as InnerStorageGcsConfig;
 use common_storage::StorageHdfsConfig as InnerStorageHdfsConfig;
 use common_storage::StorageParams;
 use common_storage::StorageS3Config as InnerStorageS3Config;
@@ -176,6 +177,10 @@ pub struct StorageConfig {
     #[clap(flatten)]
     pub fs: FsStorageConfig,
 
+    // GCS backend config
+    #[clap(flatten)]
+    pub gcs: GcsStorageConfig,
+
     // S3 storage backend config.
     #[clap(flatten)]
     pub s3: S3StorageConfig,
@@ -202,6 +207,7 @@ impl From<InnerStorageConfig> for StorageConfig {
             storage_type: "".to_string(),
             allow_insecure: inner.allow_insecure,
             fs: Default::default(),
+            gcs: Default::default(),
             s3: Default::default(),
             azblob: Default::default(),
             hdfs: Default::default(),
@@ -228,6 +234,10 @@ impl From<InnerStorageConfig> for StorageConfig {
                 cfg.storage_type = "s3".to_string();
                 cfg.s3 = v.into()
             }
+            StorageParams::Gcs(v) => {
+                cfg.storage_type = "gcs".to_string();
+                cfg.gcs = v.into()
+            }
             v => unreachable!("{v:?} should not be used as storage backend"),
         }
 
@@ -246,6 +256,7 @@ impl TryInto<InnerStorageConfig> for StorageConfig {
                 match self.storage_type.as_str() {
                     "azblob" => StorageParams::Azblob(self.azblob.try_into()?),
                     "fs" => StorageParams::Fs(self.fs.try_into()?),
+                    "gcs" => StorageParams::Gcs(self.gcs.try_into()?),
                     #[cfg(feature = "storage-hdfs")]
                     "hdfs" => StorageParams::Hdfs(self.hdfs.try_into()?),
                     "memory" => StorageParams::Memory,
@@ -289,6 +300,69 @@ impl TryInto<InnerStorageFsConfig> for FsStorageConfig {
     }
 }
 
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize, Args)]
+#[serde(default)]
+pub struct GcsStorageConfig {
+    #[clap(
+        long = "storage-gcs-endpoint-url",
+        default_value = "https://storage.googleapis.com"
+    )]
+    #[serde(rename = "endpoint_url")]
+    pub gcs_endpoint_url: String,
+
+    #[clap(long = "storage-gcs-bucket", default_value_t)]
+    #[serde(rename = "bucket")]
+    pub gcs_bucket: String,
+
+    #[clap(long = "storage-gcs-root", default_value_t)]
+    #[serde(rename = "root")]
+    pub gcs_root: String,
+
+    #[clap(long = "storage-gcs-credential", default_value_t)]
+    pub credential: String,
+}
+
+impl Default for GcsStorageConfig {
+    fn default() -> Self {
+        InnerStorageGcsConfig::default().into()
+    }
+}
+
+impl Debug for GcsStorageConfig {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        f.debug_struct("GcsStorageConfig")
+            .field("endpoint_url", &self.gcs_endpoint_url)
+            .field("root", &self.gcs_root)
+            .field("bucket", &self.gcs_bucket)
+            .field("credential", &mask_string(&self.credential, 3))
+            .finish()
+    }
+}
+
+impl From<InnerStorageGcsConfig> for GcsStorageConfig {
+    fn from(inner: InnerStorageGcsConfig) -> Self {
+        Self {
+            gcs_endpoint_url: inner.endpoint_url,
+            gcs_bucket: inner.bucket,
+            gcs_root: inner.root,
+            credential: inner.credential,
+        }
+    }
+}
+
+impl TryInto<InnerStorageGcsConfig> for GcsStorageConfig {
+    type Error = ErrorCode;
+
+    fn try_into(self) -> std::result::Result<InnerStorageGcsConfig, Self::Error> {
+        Ok(InnerStorageGcsConfig {
+            endpoint_url: self.gcs_endpoint_url,
+            bucket: self.gcs_bucket,
+            root: self.gcs_root,
+            credential: self.credential,
+        })
+    }
+}
+
 #[derive(Clone, PartialEq, Serialize, Deserialize, Eq, Args)]
 #[serde(default)]
 pub struct S3StorageConfig {

src/common/storage/src/config.rs

@@ -35,6 +35,7 @@ pub struct StorageConfig {
 pub enum StorageParams {
     Azblob(StorageAzblobConfig),
     Fs(StorageFsConfig),
+    Gcs(StorageGcsConfig),
     #[cfg(feature = "storage-hdfs")]
     Hdfs(StorageHdfsConfig),
     Http(StorageHttpConfig),
@@ -58,6 +59,11 @@ impl Display for StorageParams {
                 v.container, v.root, v.endpoint_url
             ),
             StorageParams::Fs(v) => write!(f, "fs://root={}", v.root),
+            StorageParams::Gcs(v) => write!(
+                f,
+                "gcs://bucket={},root={},endpoint={}",
+                v.bucket, v.root, v.endpoint_url
+            ),
             #[cfg(feature = "storage-hdfs")]
             StorageParams::Hdfs(v) => {
                 write!(f, "hdfs://root={},name_node={}", v.root, v.name_node)
@@ -90,6 +96,7 @@ impl StorageParams {
             StorageParams::Http(v) => v.endpoint_url.starts_with("https://"),
             StorageParams::Memory => false,
             StorageParams::S3(v) => v.endpoint_url.starts_with("https://"),
+            StorageParams::Gcs(v) => v.endpoint_url.starts_with("https://"),
         }
     }
 }
@@ -130,6 +137,39 @@ impl Default for StorageFsConfig {
     }
 }
 
+pub static STORAGE_GCS_DEFAULT_ENDPOINT: &str = "https://storage.googleapis.com";
+
+/// Config for storage backend GCS.
+#[derive(Clone, PartialEq, Eq, Deserialize, Serialize)]
+pub struct StorageGcsConfig {
+    pub endpoint_url: String,
+    pub bucket: String,
+    pub root: String,
+    pub credential: String,
+}
+
+impl Default for StorageGcsConfig {
+    fn default() -> Self {
+        Self {
+            endpoint_url: STORAGE_GCS_DEFAULT_ENDPOINT.to_string(),
+            bucket: String::new(),
+            root: String::new(),
+            credential: String::new(),
+        }
+    }
+}
+
+impl Debug for StorageGcsConfig {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("StorageGcsConfig")
+            .field("endpoint", &self.endpoint_url)
+            .field("bucket", &self.bucket)
+            .field("root", &self.root)
+            .field("credential", &mask_string(&self.credential, 3))
+            .finish()
+    }
+}
+
 /// Config for storage backend hdfs.
 ///
 /// # Notes

src/common/storage/src/lib.rs

@@ -20,15 +20,18 @@ mod config;
 pub use config::StorageAzblobConfig;
 pub use config::StorageConfig;
 pub use config::StorageFsConfig;
+pub use config::StorageGcsConfig;
 pub use config::StorageHdfsConfig;
 pub use config::StorageHttpConfig;
 pub use config::StorageParams;
 pub use config::StorageS3Config;
+pub use config::STORAGE_GCS_DEFAULT_ENDPOINT;
 pub use config::STORAGE_S3_DEFAULT_ENDPOINT;
 
 mod operator;
 pub use operator::init_azblob_operator;
 pub use operator::init_fs_operator;
+pub use operator::init_gcs_operator;
 #[cfg(feature = "storage-hdfs")]
 pub use operator::init_hdfs_operator;
 pub use operator::init_http_operator;

src/common/storage/src/operator.rs

@@ -17,6 +17,7 @@ use std::io::Result;
 
 use opendal::services::azblob;
 use opendal::services::fs;
+use opendal::services::gcs;
 use opendal::services::http;
 use opendal::services::memory;
 use opendal::services::s3;
@@ -26,13 +27,15 @@ use super::StorageAzblobConfig;
 use super::StorageFsConfig;
 use super::StorageParams;
 use super::StorageS3Config;
+use crate::config::StorageGcsConfig;
 use crate::config::StorageHttpConfig;
 
 /// init_operator will init an opendal operator based on storage config.
 pub fn init_operator(cfg: &StorageParams) -> Result<Operator> {
     Ok(match &cfg {
         StorageParams::Azblob(cfg) => init_azblob_operator(cfg)?,
         StorageParams::Fs(cfg) => init_fs_operator(cfg)?,
+        StorageParams::Gcs(cfg) => init_gcs_operator(cfg)?,
         #[cfg(feature = "storage-hdfs")]
         StorageParams::Hdfs(cfg) => init_hdfs_operator(cfg)?,
         StorageParams::Http(cfg) => init_http_operator(cfg)?,
@@ -74,6 +77,20 @@ pub fn init_fs_operator(cfg: &StorageFsConfig) -> Result<Operator> {
     Ok(Operator::new(builder.build()?))
 }
 
+/// init_gcs_operator will init a opendal gcs operator.
+pub fn init_gcs_operator(cfg: &StorageGcsConfig) -> Result<Operator> {
+    let mut builder = gcs::Builder::default();
+
+    let accessor = builder
+        .endpoint(&cfg.endpoint_url)
+        .bucket(&cfg.bucket)
+        .root(&cfg.root)
+        .credential(&cfg.credential)
+        .build()?;
+
+    Ok(Operator::new(accessor))
+}
+
 /// init_hdfs_operator will init an opendal hdfs operator.
 #[cfg(feature = "storage-hdfs")]
 pub fn init_hdfs_operator(cfg: &super::StorageHdfsConfig) -> Result<Operator> {

src/common/storages/preludes/src/system/configs_table.rs

@@ -88,6 +88,7 @@ impl SyncSystemTable for ConfigsTable {
         let mut storage_config = config.storage;
         storage_config.s3.access_key_id = mask_string(&storage_config.s3.access_key_id, 3);
         storage_config.s3.secret_access_key = mask_string(&storage_config.s3.secret_access_key, 3);
+        storage_config.gcs.credential = mask_string(&storage_config.gcs.credential, 3);
         storage_config.azblob.account_name = mask_string(&storage_config.azblob.account_name, 3);
         storage_config.azblob.account_key = mask_string(&storage_config.azblob.account_key, 3);
         let storage_config_value = serde_json::to_value(storage_config)?;