feat!: support table ddl for custom storage (#2733)

* feat: support table ddl for custom_storage

* refactor: rename extract_variant_name to name

* chore: add blank

* chore: keep compatible

* feat: rename custom_stores to providers

* chore: rename

* chore: config

* refactor: add should_retry in client Error

* fix: test fail

* chore: remove unused options

* chore: remove unused import

* chore: remove the blanks.

* chore: revert

---------

Co-authored-by: dennis zhuang <killme2008@gmail.com>

src/datanode/src/config.rs

@@ -48,6 +48,18 @@ pub enum ObjectStoreConfig {
     Gcs(GcsConfig),
 }
 
+impl ObjectStoreConfig {
+    pub fn name(&self) -> &'static str {
+        match self {
+            Self::File(_) => "File",
+            Self::S3(_) => "S3",
+            Self::Oss(_) => "Oss",
+            Self::Azblob(_) => "Azblob",
+            Self::Gcs(_) => "Gcs",
+        }
+    }
+}
+
 /// Storage engine config
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(default)]
@@ -63,6 +75,7 @@ pub struct StorageConfig {
     pub data_home: String,
     #[serde(flatten)]
     pub store: ObjectStoreConfig,
+    pub providers: Vec<ObjectStoreConfig>,
 }
 
 impl Default for StorageConfig {
@@ -71,6 +84,7 @@ impl Default for StorageConfig {
             global_ttl: None,
             data_home: DEFAULT_DATA_HOME.to_string(),
             store: ObjectStoreConfig::default(),
+            providers: vec![],
         }
     }
 }
@@ -295,7 +309,7 @@ mod tests {
             secret_access_key = "secret_access_key"
         "#;
         let opts: DatanodeOptions = toml::from_str(toml_str).unwrap();
-        match opts.storage.store {
+        match &opts.storage.store {
             ObjectStoreConfig::S3(cfg) => {
                 assert_eq!(
                     "Secret([REDACTED alloc::string::String])".to_string(),

src/datanode/src/datanode.rs

@@ -354,7 +354,6 @@ impl DatanodeBuilder {
                 ));
             }
         }
-
         info!("going to open {} regions", regions.len());
         let semaphore = Arc::new(tokio::sync::Semaphore::new(OPEN_REGION_PARALLELISM));
         let mut tasks = vec![];
@@ -417,7 +416,6 @@ impl DatanodeBuilder {
         );
 
         let table_provider_factory = Arc::new(DummyTableProviderFactory);
-
         let mut region_server = RegionServer::with_table_provider(
             query_engine,
             runtime,
@@ -425,13 +423,8 @@ impl DatanodeBuilder {
             table_provider_factory,
         );
 
-        let object_store = store::new_object_store(opts).await?;
-        let object_store_manager = ObjectStoreManager::new(
-            "default", // TODO: use a name which is set in the configuration when #919 is done.
-            object_store,
-        );
-        let engines =
-            Self::build_store_engines(opts, log_store, Arc::new(object_store_manager)).await?;
+        let object_store_manager = Self::build_object_store_manager(opts).await?;
+        let engines = Self::build_store_engines(opts, log_store, object_store_manager).await?;
         for engine in engines {
             region_server.register_engine(engine);
         }
@@ -496,6 +489,21 @@ impl DatanodeBuilder {
         }
         Ok(engines)
     }
+
+    /// Builds [ObjectStoreManager]
+    async fn build_object_store_manager(opts: &DatanodeOptions) -> Result<ObjectStoreManagerRef> {
+        let object_store =
+            store::new_object_store(opts.storage.store.clone(), &opts.storage.data_home).await?;
+        let default_name = opts.storage.store.name();
+        let mut object_store_manager = ObjectStoreManager::new(default_name, object_store);
+        for store in &opts.storage.providers {
+            object_store_manager.add(
+                store.name(),
+                store::new_object_store(store.clone(), &opts.storage.data_home).await?,
+            );
+        }
+        Ok(Arc::new(object_store_manager))
+    }
 }
 
 #[cfg(test)]

src/datanode/src/store.rs

@@ -32,12 +32,15 @@ use object_store::util::normalize_dir;
 use object_store::{util, HttpClient, ObjectStore, ObjectStoreBuilder};
 use snafu::prelude::*;
 
-use crate::config::{DatanodeOptions, ObjectStoreConfig, DEFAULT_OBJECT_STORE_CACHE_SIZE};
+use crate::config::{ObjectStoreConfig, DEFAULT_OBJECT_STORE_CACHE_SIZE};
 use crate::error::{self, Result};
 
-pub(crate) async fn new_object_store(opts: &DatanodeOptions) -> Result<ObjectStore> {
-    let data_home = normalize_dir(&opts.storage.data_home);
-    let object_store = match &opts.storage.store {
+pub(crate) async fn new_object_store(
+    store: ObjectStoreConfig,
+    data_home: &str,
+) -> Result<ObjectStore> {
+    let data_home = normalize_dir(data_home);
+    let object_store = match &store {
         ObjectStoreConfig::File(file_config) => {
             fs::new_fs_object_store(&data_home, file_config).await
         }
@@ -50,9 +53,8 @@ pub(crate) async fn new_object_store(opts: &DatanodeOptions) -> Result<ObjectSto
     }?;
 
     // Enable retry layer and cache layer for non-fs object storages
-    let object_store = if !matches!(opts.storage.store, ObjectStoreConfig::File(..)) {
-        let object_store =
-            create_object_store_with_cache(object_store, &opts.storage.store).await?;
+    let object_store = if !matches!(store, ObjectStoreConfig::File(..)) {
+        let object_store = create_object_store_with_cache(object_store, &store).await?;
         object_store.layer(RetryLayer::new().with_jitter())
     } else {
         object_store