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.
@@ -178,7 +178,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

@@ -33,7 +33,10 @@ hex = "0.4.3"
 once_cell = "1.15.0"
 semver = "1.0.14"
 serde = { workspace = true }
+serde_json = { 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

@@ -29,6 +29,7 @@ use common_tracing::Config as LogConfig;
 use common_users::idm_config::IDMConfig;
 
 use super::outer_v0::Config as OuterV0Config;
+use super::outer_v0::MetaType;
 
 /// Inner config for query.
 ///
@@ -275,6 +276,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: String,
 }
 
 impl Default for MetaConfig {
@@ -289,11 +291,25 @@ 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::Embedded.to_string(),
         }
     }
 }
 
 impl MetaConfig {
+    pub fn is_embedded_meta(&self) -> Result<bool> {
+        let t = match MetaType::from_str(&self.meta_type) {
+            Err(_) => {
+                return Err(ErrorCode::InvalidConfig(format!(
+                    "Invalid MetaType: {}",
+                    self.meta_type
+                )));
+            }
+            Ok(t) => t,
+        };
+        Ok(t == 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

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

src/query/config/src/outer_v0.rs

@@ -1258,11 +1258,32 @@ impl From<InnerHiveCatalogConfig> for HiveCatalogConfig {
     }
 }
 
+#[derive(
+    serde::Serialize,
+    serde::Deserialize,
+    Debug,
+    PartialEq,
+    Eq,
+    strum_macros::EnumString,
+    strum_macros::Display,
+)]
+#[strum(serialize_all = "camelCase")]
+pub enum MetaType {
+    Remote,
+
+    Embedded,
+}
+
 /// Meta config group.
 /// TODO(xuanwo): All meta_xxx should be rename to xxx.
 #[derive(Clone, PartialEq, Eq, Serialize, Deserialize, Args)]
 #[serde(default)]
 pub struct MetaConfig {
+    /// MetaStore type: remote or embedded
+    #[clap(long = "meta-type", default_value = "embedded")]
+    #[serde(rename = "type")]
+    pub meta_type: String,
+
     /// The dir to store persisted meta state for a embedded meta store
     #[clap(
         long = "meta-embedded-dir",
@@ -1308,6 +1329,38 @@ pub struct MetaConfig {
     pub rpc_tls_meta_service_domain_name: String,
 }
 
+impl MetaConfig {
+    fn check_config(&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,
+        };
+        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()
@@ -1318,6 +1371,8 @@ impl TryInto<InnerMetaConfig> for MetaConfig {
     type Error = ErrorCode;
 
     fn try_into(self) -> Result<InnerMetaConfig> {
+        self.check_config()?;
+
         Ok(InnerMetaConfig {
             embedded_dir: self.embedded_dir,
             address: self.address,
@@ -1328,6 +1383,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: self.meta_type,
         })
     }
 }
@@ -1344,13 +1400,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,
         }
     }
 }
 
 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)