Merge pull request #8571 from lichuang/meta_remote_config

feat: add meta_type config

src/binaries/query/main.rs

@@ -47,7 +47,7 @@ async fn main(_global_tracker: Arc<RuntimeTracker>) -> common_exception::Result<
     init_default_metrics_recorder();
     set_panic_hook();
 
-    if conf.meta.address.is_empty() && conf.meta.endpoints.is_empty() {
+    if conf.meta.is_embedded_meta()? {
         MetaEmbedded::init_global_meta_store(conf.meta.embedded_dir.clone()).await?;
     }
     // Make sure gloabl services have been inited.
@@ -179,7 +179,7 @@ async fn main(_global_tracker: Arc<RuntimeTracker>) -> common_exception::Result<
     println!("    stderr: {}", conf.log.stderr);
     println!(
         "Meta: {}",
-        if conf.meta.address.is_empty() && conf.meta.endpoints.is_empty() {
+        if conf.meta.is_embedded_meta()? {
             format!("embedded at {}", conf.meta.embedded_dir)
         } else if !conf.meta.endpoints.is_empty() {
             format!("connected to endpoints {:#?}", conf.meta.endpoints)

src/query/config/Cargo.toml

@@ -14,7 +14,7 @@ storage-hdfs = ["common-storage/storage-hdfs"]
 hive = ["common-hive-meta-store", "thrift", "storage-hdfs"]
 
 [package.metadata.cargo-machete]
-ignored = ["thrift", "common-hive-meta-store"]
+ignored = ["thrift", "common-hive-meta-store", "strum"]
 
 [dependencies]
 common-base = { path = "../../common/base" }
@@ -34,6 +34,8 @@ once_cell = "1.15.0"
 semver = "1.0.14"
 serde = { workspace = true }
 serfig = "0.0.3"
+strum = "0.24.1"
+strum_macros = "0.24.1"
 
 [build-dependencies]
 common-building = { path = "../../common/building" }

src/query/config/src/inner.rs

@@ -262,6 +262,27 @@ impl Default for HiveCatalogConfig {
     }
 }
 
+#[derive(
+    serde::Serialize,
+    serde::Deserialize,
+    Debug,
+    PartialEq,
+    Eq,
+    Clone,
+    strum_macros::EnumString,
+    strum_macros::Display,
+)]
+#[strum(serialize_all = "camelCase")]
+pub enum MetaType {
+    Remote,
+
+    Embedded,
+
+    // Fallback is used for forward compatbility, that is:
+    // First check embedded config, then endpoints, finally address.
+    Fallback,
+}
+
 #[derive(Clone, PartialEq, Eq)]
 pub struct MetaConfig {
     /// The dir to store persisted meta state for a embedded meta store
@@ -281,6 +302,7 @@ pub struct MetaConfig {
     /// Certificate for client to identify meta rpc serve
     pub rpc_tls_meta_server_root_ca_cert: String,
     pub rpc_tls_meta_service_domain_name: String,
+    pub meta_type: MetaType,
 }
 
 impl Default for MetaConfig {
@@ -295,11 +317,16 @@ impl Default for MetaConfig {
             auto_sync_interval: 10,
             rpc_tls_meta_server_root_ca_cert: "".to_string(),
             rpc_tls_meta_service_domain_name: "localhost".to_string(),
+            meta_type: MetaType::Fallback,
         }
     }
 }
 
 impl MetaConfig {
+    pub fn is_embedded_meta(&self) -> Result<bool> {
+        Ok(self.meta_type == MetaType::Embedded)
+    }
+
     pub fn is_tls_enabled(&self) -> bool {
         !self.rpc_tls_meta_server_root_ca_cert.is_empty()
             && !self.rpc_tls_meta_service_domain_name.is_empty()

src/query/config/src/lib.rs

@@ -28,6 +28,7 @@ mod outer_v0;
 mod version;
 
 pub use inner::Config;
+pub use inner::MetaType;
 pub use inner::QueryConfig;
 pub use version::DATABEND_COMMIT_VERSION;
 pub use version::QUERY_SEMVER;

src/query/config/src/outer_v0.rs

@@ -51,6 +51,7 @@ use serfig::parsers::Toml;
 use super::inner::Config as InnerConfig;
 use super::inner::HiveCatalogConfig as InnerHiveCatalogConfig;
 use super::inner::MetaConfig as InnerMetaConfig;
+use super::inner::MetaType;
 use super::inner::QueryConfig as InnerQueryConfig;
 use crate::DATABEND_COMMIT_VERSION;
 
@@ -1362,6 +1363,11 @@ impl From<InnerHiveCatalogConfig> for HiveCatalogConfig {
 #[derive(Clone, PartialEq, Eq, Serialize, Deserialize, Args)]
 #[serde(default)]
 pub struct MetaConfig {
+    /// MetaStore type: remote, embedded, fallback
+    #[clap(long = "meta-type", default_value = "fallback")]
+    #[serde(rename = "type")]
+    pub meta_type: String,
+
     /// The dir to store persisted meta state for a embedded meta store
     #[clap(
         long = "meta-embedded-dir",
@@ -1407,6 +1413,49 @@ pub struct MetaConfig {
     pub rpc_tls_meta_service_domain_name: String,
 }
 
+impl MetaConfig {
+    fn check_config(&mut self) -> Result<()> {
+        let t = match MetaType::from_str(&self.meta_type) {
+            Err(_) => {
+                return Err(ErrorCode::InvalidConfig(format!(
+                    "Invalid MetaType: {}",
+                    self.meta_type
+                )));
+            }
+            Ok(t) => t,
+        };
+
+        // Fallback is used for forward compatbility, that is:
+        // First check embedded config, then endpoints, finally address.
+        if t == MetaType::Fallback {
+            if !self.embedded_dir.is_empty() && self.endpoints.is_empty() {
+                self.meta_type = MetaType::Embedded.to_string();
+            } else {
+                self.meta_type = MetaType::Remote.to_string();
+            }
+        }
+        match t {
+            MetaType::Embedded => {
+                if self.embedded_dir.is_empty() {
+                    return Err(ErrorCode::InvalidConfig(
+                        "Embedded Meta but embedded_dir is empty",
+                    ));
+                }
+            }
+            MetaType::Remote => {
+                if self.address.is_empty() && self.endpoints.is_empty() {
+                    return Err(ErrorCode::InvalidConfig(
+                        "Remote Meta but address and enpoints are empty",
+                    ));
+                }
+            }
+            _ => {}
+        };
+
+        Ok(())
+    }
+}
+
 impl Default for MetaConfig {
     fn default() -> Self {
         InnerMetaConfig::default().into()
@@ -1416,7 +1465,9 @@ impl Default for MetaConfig {
 impl TryInto<InnerMetaConfig> for MetaConfig {
     type Error = ErrorCode;
 
-    fn try_into(self) -> Result<InnerMetaConfig> {
+    fn try_into(mut self) -> Result<InnerMetaConfig> {
+        self.check_config()?;
+
         Ok(InnerMetaConfig {
             embedded_dir: self.embedded_dir,
             address: self.address,
@@ -1427,6 +1478,7 @@ impl TryInto<InnerMetaConfig> for MetaConfig {
             auto_sync_interval: self.auto_sync_interval,
             rpc_tls_meta_server_root_ca_cert: self.rpc_tls_meta_server_root_ca_cert,
             rpc_tls_meta_service_domain_name: self.rpc_tls_meta_service_domain_name,
+            meta_type: MetaType::from_str(&self.meta_type).unwrap(),
         })
     }
 }
@@ -1443,13 +1495,15 @@ impl From<InnerMetaConfig> for MetaConfig {
             auto_sync_interval: inner.auto_sync_interval,
             rpc_tls_meta_server_root_ca_cert: inner.rpc_tls_meta_server_root_ca_cert,
             rpc_tls_meta_service_domain_name: inner.rpc_tls_meta_service_domain_name,
+            meta_type: inner.meta_type.to_string(),
         }
     }
 }
 
 impl Debug for MetaConfig {
     fn fmt(&self, f: &mut Formatter) -> fmt::Result {
         f.debug_struct("MetaConfig")
+            .field("meta_type", &self.meta_type)
             .field("address", &self.address)
             .field("endpoints", &self.endpoints)
             .field("username", &self.username)

src/query/formats/Cargo.toml

@@ -22,8 +22,6 @@ common-settings = { path = "../settings" }
 
 # Crates.io dependencies
 once_cell = "1.15.0"
-strum = "0.24.1"
-strum_macros = "0.24.3"
 
 [dev-dependencies]
 common-arrow = { path = "../../common/arrow" }

src/query/service/tests/it/configs.rs

@@ -94,6 +94,7 @@ level = "INFO"
 format = "text"
 
 [meta]
+type = "fallback"
 embedded_dir = "./.databend/meta_embedded"
 address = ""
 endpoints = []

src/query/service/tests/it/storages/testdata/system-tables.txt

@@ -172,6 +172,7 @@ DB.Table: 'system'.'configs', Table: configs-table_id:1, ver:0, Engine: SystemCo
 | meta    | password                             |                                |             |
 | meta    | rpc_tls_meta_server_root_ca_cert     |                                |             |
 | meta    | rpc_tls_meta_service_domain_name     | localhost                      |             |
+| meta    | type                                 | fallback                       |             |
 | meta    | username                             | root                           |             |
 | query   | admin_api_address                    | 127.0.0.1:8080                 |             |
 | query   | api_tls_server_cert                  |                                |             |