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

@@ -12,7 +12,7 @@ pyo3-build-config = "0.18.3"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 [lib]
 name = "databend"
-crate-type = ["cdylib"]
+crate-ype = ["cdylib"]
 
 [dependencies]
 # Workspace dependencies
@@ -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/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),