diff --git a/Cargo.lock b/Cargo.lock
index 4e4d7ca038632..345399bc15ab4 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1216,7 +1216,6 @@ dependencies = [
  "databend-common-license",
  "databend-common-meta-app",
  "databend-common-meta-embedded",
- "databend-common-meta-types",
  "databend-common-users",
  "databend-query",
  "pyo3",
diff --git a/src/bendpy/Cargo.toml b/src/bendpy/Cargo.toml
index ea8ff2948663c..fa20c229aceb8 100644
--- a/src/bendpy/Cargo.toml
+++ b/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",
diff --git a/src/bendpy/src/context.rs b/src/bendpy/src/context.rs
index 6d2ce2b8e305c..357fc4229c4ee 100644
--- a/src/bendpy/src/context.rs
+++ b/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 };
 
diff --git a/src/meta/api/src/schema_api_test_suite.rs b/src/meta/api/src/schema_api_test_suite.rs
index 4e228d5f1f062..19009d12806f2 100644
--- a/src/meta/api/src/schema_api_test_suite.rs
+++ b/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;
diff --git a/src/meta/app/src/principal/user_identity.rs b/src/meta/app/src/principal/user_identity.rs
index 6c2821fa70cef..962acae993d8c 100644
--- a/src/meta/app/src/principal/user_identity.rs
+++ b/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,
diff --git a/src/meta/app/src/principal/user_stage_file_ident.rs b/src/meta/app/src/principal/user_stage_file_ident.rs
index 5f8cb788eceda..389b2b9a73909 100644
--- a/src/meta/app/src/principal/user_stage_file_ident.rs
+++ b/src/meta/app/src/principal/user_stage_file_ident.rs
@@ -79,7 +79,6 @@ 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;
@@ -87,7 +86,7 @@ mod tests {
 
     #[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");
 
diff --git a/src/meta/app/src/principal/user_stage_ident.rs b/src/meta/app/src/principal/user_stage_ident.rs
index 60e96d0acd785..5e5abdf79de65 100644
--- a/src/meta/app/src/principal/user_stage_ident.rs
+++ b/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();
diff --git a/src/meta/app/src/tenant/tenant_quota_ident.rs b/src/meta/app/src/tenant/tenant_quota_ident.rs
index f1f3c594dbb26..a5f40734dec56 100644
--- a/src/meta/app/src/tenant/tenant_quota_ident.rs
+++ b/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();
diff --git a/src/meta/app/src/tenant_key/ident.rs b/src/meta/app/src/tenant_key/ident.rs
index d68c584b08ec4..468a56ab8aec2 100644
--- a/src/meta/app/src/tenant_key/ident.rs
+++ b/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();
diff --git a/src/query/catalog/src/catalog/manager.rs b/src/query/catalog/src/catalog/manager.rs
index 598169cd7269c..3c8e68a348214 100644
--- a/src/query/catalog/src/catalog/manager.rs
+++ b/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.
diff --git a/src/query/service/src/catalogs/default/mutable_catalog.rs b/src/query/service/src/catalogs/default/mutable_catalog.rs
index 8ac2a0af718f4..9ccc035854d1f 100644
--- a/src/query/service/src/catalogs/default/mutable_catalog.rs
+++ b/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)
     }
 }
 
diff --git a/src/query/service/src/databases/database_factory.rs b/src/query/service/src/databases/database_factory.rs
index 3e169740ee003..f69dc4f56f1ae 100644
--- a/src/query/service/src/databases/database_factory.rs
+++ b/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())
     }
 }
diff --git a/src/query/sql/src/planner/binder/ddl/index.rs b/src/query/sql/src/planner/binder/ddl/index.rs
index 13026018387d0..4ef6585030212 100644
--- a/src/query/sql/src/planner/binder/ddl/index.rs
+++ b/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),