refactor: simplify DatabaseFactory, just static dispatch (#15203)

* refactor: simplify `DatabaseFactory`, just static dispatch

* chore: try reduce usage of Tenant::new_nonempty()

This methods is a transitional method, to build a Tenant from a
NonEmptyString. This method will be removed in future because a Tenant
must contain a in-meta-service per-tenant config to work correctly,
e.g., determine the version of path to store a value: with tenant prefix
or without.

src/bendpy/Cargo.toml

@@ -25,7 +25,6 @@ databend-common-expression = { path = "../query/expression" }
 databend-common-license = { path = "../common/license" }
 databend-common-meta-app = { path = "../meta/app" }
 databend-common-meta-embedded = { path = "../meta/embedded" }
-databend-common-meta-types = { path = "../meta/types" }
 databend-common-users = { path = "../query/users" }
 databend-query = { path = "../query/service", features = [
     "simd",

src/bendpy/src/context.rs

@@ -20,7 +20,6 @@ use databend_common_meta_app::principal::GrantObject;
 use databend_common_meta_app::principal::UserInfo;
 use databend_common_meta_app::principal::UserPrivilegeSet;
 use databend_common_meta_app::tenant::Tenant;
-use databend_common_meta_types::NonEmptyString;
 use databend_common_users::UserApiProvider;
 use databend_query::sessions::QueryContext;
 use databend_query::sessions::Session;
@@ -57,8 +56,9 @@ impl PySessionContext {
                 uuid::Uuid::new_v4().to_string()
             };
 
-            let tenant = NonEmptyString::new(tenant).unwrap();
-            let tenant = Tenant::new_nonempty(tenant);
+            let tenant = Tenant::new_or_err(tenant, "PySessionContext::new()").map_err(|e| {
+                PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!("Error: {}", e))
+            })?;
 
             let config = GlobalConfig::instance();
             UserApiProvider::try_create_simple(config.meta.to_meta_grpc_client_conf(), &tenant)
@@ -74,8 +74,8 @@ impl PySessionContext {
             );
 
             session.set_authed_user(user, None).await.unwrap();
-            session
-        });
+            Ok::<Arc<Session>, PyErr>(session)
+        })?;
 
         let mut res = Self { session };
 

src/meta/api/src/schema_api_test_suite.rs

@@ -125,7 +125,6 @@ use databend_common_meta_kvapi::kvapi::Key;
 use databend_common_meta_kvapi::kvapi::UpsertKVReq;
 use databend_common_meta_types::MatchSeq;
 use databend_common_meta_types::MetaError;
-use databend_common_meta_types::NonEmptyString;
 use databend_common_meta_types::Operation;
 use databend_common_meta_types::UpsertKV;
 use log::debug;
@@ -3604,7 +3603,7 @@ impl SchemaApiTestSuite {
         mt: &MT,
     ) -> anyhow::Result<()> {
         let tenant_name = "db_table_gc_out_of_retention_time";
-        let tenant = Tenant::new_nonempty(NonEmptyString::new(tenant_name).unwrap());
+        let tenant = Tenant::new_literal(tenant_name);
         let db1_name = "db1";
         let tb1_name = "tb1";
         let idx1_name = "idx1";
@@ -5940,7 +5939,7 @@ impl SchemaApiTestSuite {
     async fn index_create_list_drop<MT>(&self, mt: &MT) -> anyhow::Result<()>
     where MT: SchemaApi + kvapi::AsKVApi<Error = MetaError> {
         let tenant_name = "tenant1";
-        let tenant = Tenant::new_nonempty(NonEmptyString::new(tenant_name).unwrap());
+        let tenant = Tenant::new_literal(tenant_name);
 
         let mut util = Util::new(mt, tenant_name, "db1", "tb1", "eng1");
         let table_id;

src/meta/app/src/principal/user_identity.rs

@@ -139,14 +139,13 @@ mod kvapi_key_impl {
 #[cfg(test)]
 mod tests {
     use databend_common_meta_kvapi::kvapi::Key;
-    use databend_common_meta_types::NonEmptyString;
 
     use crate::principal::user_identity::TenantUserIdent;
     use crate::principal::UserIdentity;
     use crate::tenant::Tenant;
 
     fn test_format_parse(user: &str, host: &str, expect: &str) {
-        let tenant = Tenant::new_nonempty(NonEmptyString::new("test_tenant").unwrap());
+        let tenant = Tenant::new_literal("test_tenant");
         let user_ident = UserIdentity::new(user, host);
         let tenant_user_ident = TenantUserIdent {
             tenant,

src/meta/app/src/principal/user_stage_file_ident.rs

@@ -79,15 +79,14 @@ mod kvapi_key_impl {
 #[cfg(test)]
 mod tests {
     use databend_common_meta_kvapi::kvapi::Key;
-    use databend_common_meta_types::NonEmptyString;
 
     use crate::principal::user_stage_file_ident::StageFileIdent;
     use crate::principal::StageIdent;
     use crate::tenant::Tenant;
 
     #[test]
     fn test_kvapi_key_for_stage_file_ident() {
-        let tenant = Tenant::new_nonempty(NonEmptyString::new("tenant1").unwrap());
+        let tenant = Tenant::new_literal("tenant1");
         let stage = StageIdent::new(tenant, "stage1");
         let sfi = StageFileIdent::new(stage, "file1");
 

src/meta/app/src/principal/user_stage_ident.rs

@@ -42,14 +42,13 @@ mod kvapi_impl {
 #[cfg(test)]
 mod tests {
     use databend_common_meta_kvapi::kvapi::Key;
-    use databend_common_meta_types::NonEmptyString;
 
     use crate::principal::user_stage_ident::StageIdent;
     use crate::tenant::Tenant;
 
     #[test]
     fn test_kvapi_key_for_stage_ident() {
-        let tenant = Tenant::new_nonempty(NonEmptyString::new("test").unwrap());
+        let tenant = Tenant::new_literal("test");
         let stage = StageIdent::new(tenant, "stage");
 
         let key = stage.to_string_key();

src/meta/app/src/tenant/tenant_quota_ident.rs

@@ -69,13 +69,12 @@ mod kvapi_key_impl {
 #[cfg(test)]
 mod tests {
     use databend_common_meta_kvapi::kvapi::Key;
-    use databend_common_meta_types::NonEmptyString;
 
     use super::*;
 
     #[test]
     fn test_quota_ident() {
-        let tenant = Tenant::new_nonempty(NonEmptyString::new("test").unwrap());
+        let tenant = Tenant::new_literal("test");
         let ident = TenantQuotaIdent::new(tenant.clone());
 
         let key = ident.to_string_key();

src/meta/app/src/tenant_key/ident.rs

@@ -135,7 +135,6 @@ mod tests {
     use std::convert::Infallible;
 
     use databend_common_meta_kvapi::kvapi::Key;
-    use databend_common_meta_types::NonEmptyString;
 
     use crate::tenant::Tenant;
     use crate::tenant_key::ident::TIdent;
@@ -150,7 +149,7 @@ mod tests {
             type ValueType = Infallible;
         }
 
-        let tenant = Tenant::new_nonempty(NonEmptyString::new("test").unwrap());
+        let tenant = Tenant::new_literal("test");
         let ident = TIdent::<Foo>::new(tenant, "test1");
 
         let key = ident.to_string_key();

src/query/catalog/src/catalog/manager.rs

@@ -36,7 +36,7 @@ use databend_common_meta_app::schema::ListCatalogReq;
 use databend_common_meta_app::tenant::Tenant;
 use databend_common_meta_store::MetaStore;
 use databend_common_meta_store::MetaStoreProvider;
-use databend_common_meta_types::NonEmptyString;
+use databend_common_meta_types::anyerror::func_name;
 use databend_storages_common_txn::TxnManagerRef;
 
 use super::Catalog;
@@ -176,10 +176,7 @@ impl CatalogManager {
             return Ok(ctl.clone());
         }
 
-        let non_empty = NonEmptyString::new(tenant)
-            .map_err(|_e| ErrorCode::TenantIsEmpty("tenant is empty when get_catalog"))?;
-
-        let tenant = Tenant::new_nonempty(non_empty);
+        let tenant = Tenant::new_or_err(tenant, func_name!())?;
         let ident = CatalogNameIdent::new(tenant, catalog_name);
 
         // Get catalog from metasrv.

src/query/service/src/catalogs/default/mutable_catalog.rs

@@ -187,7 +187,9 @@ impl MutableCatalog {
             storage_factory: self.ctx.storage_factory.clone(),
             tenant: self.tenant.clone(),
         };
-        self.ctx.database_factory.get_database(ctx, db_info)
+        self.ctx
+            .database_factory
+            .build_database_by_engine(ctx, db_info)
     }
 }
 

src/query/service/src/databases/database_factory.rs

@@ -14,7 +14,6 @@
 
 use std::sync::Arc;
 
-use dashmap::DashMap;
 use databend_common_config::InnerConfig;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
@@ -25,57 +24,36 @@ use crate::databases::share::ShareDatabase;
 use crate::databases::Database;
 use crate::databases::DatabaseContext;
 
-pub trait DatabaseCreator: Send + Sync {
-    fn try_create(&self, ctx: DatabaseContext, db_info: DatabaseInfo) -> Result<Box<dyn Database>>;
-}
-
-impl<T> DatabaseCreator for T
-where
-    T: Fn(DatabaseContext, DatabaseInfo) -> Result<Box<dyn Database>>,
-    T: Send + Sync,
-{
-    fn try_create(&self, ctx: DatabaseContext, db_info: DatabaseInfo) -> Result<Box<dyn Database>> {
-        self(ctx, db_info)
-    }
-}
-
-#[derive(Default)]
-pub struct DatabaseFactory {
-    creators: DashMap<String, Arc<dyn DatabaseCreator>>,
-}
+pub struct DatabaseFactory {}
 
 impl DatabaseFactory {
     pub fn create(_: InnerConfig) -> Self {
-        let creators: DashMap<String, Arc<dyn DatabaseCreator>> = DashMap::new();
-        creators.insert(
-            DefaultDatabase::NAME.to_string(),
-            Arc::new(DefaultDatabase::try_create),
-        );
-        creators.insert(
-            ShareDatabase::NAME.to_string(),
-            Arc::new(ShareDatabase::try_create),
-        );
-
-        DatabaseFactory { creators }
+        DatabaseFactory {}
     }
 
-    pub fn get_database(
+    pub fn build_database_by_engine(
         &self,
         ctx: DatabaseContext,
         db_info: &DatabaseInfo,
     ) -> Result<Arc<dyn Database>> {
-        let db_engine = &db_info.engine();
+        let db_engine = db_info.engine();
+
         let engine = if db_engine.is_empty() {
             "DEFAULT".to_string()
         } else {
             db_engine.to_uppercase()
         };
 
-        let factory = self.creators.get(&engine).ok_or_else(|| {
-            ErrorCode::UnknownDatabaseEngine(format!("Unknown database engine {}", engine))
-        })?;
+        let db = match engine.as_str() {
+            DefaultDatabase::NAME => DefaultDatabase::try_create(ctx, db_info.clone())?,
+            ShareDatabase::NAME => ShareDatabase::try_create(ctx, db_info.clone())?,
 
-        let db: Arc<dyn Database> = factory.try_create(ctx, db_info.clone())?.into();
-        Ok(db)
+            _ => {
+                let err =
+                    ErrorCode::UnknownDatabaseEngine(format!("Unknown database engine {}", engine));
+                return Err(err);
+            }
+        };
+        Ok(db.into())
     }
 }

src/query/sql/src/planner/binder/ddl/index.rs

@@ -36,8 +36,6 @@ use databend_common_license::license_manager::get_license_manager;
 use databend_common_meta_app::schema::GetIndexReq;
 use databend_common_meta_app::schema::IndexMeta;
 use databend_common_meta_app::schema::IndexNameIdent;
-use databend_common_meta_app::tenant::Tenant;
-use databend_common_meta_types::NonEmptyString;
 use databend_storages_common_table_meta::meta::Location;
 use derive_visitor::Drive;
 use derive_visitor::DriveMut;
@@ -263,15 +261,7 @@ impl Binder {
             .get_catalog(&self.ctx.get_current_catalog())
             .await?;
 
-        let tenant_name = self.ctx.get_tenant();
-
-        let non_empty = NonEmptyString::new(tenant_name.name().to_string()).map_err(|_| {
-            ErrorCode::TenantIsEmpty(
-                "Tenant is empty(when Binder::build_refresh_index()".to_string(),
-            )
-        })?;
-
-        let tenant = Tenant::new_nonempty(non_empty);
+        let tenant = self.ctx.get_tenant();
 
         let get_index_req = GetIndexReq {
             name_ident: IndexNameIdent::new(tenant, &index_name),