refactor: replace NonEmptyString with Tenant

- Part of databendlabs#14719

src/bendpy/src/context.rs

@@ -19,6 +19,7 @@ use databend_common_exception::Result;
 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;
@@ -57,6 +58,7 @@ impl PySessionContext {
             };
 
             let tenant = NonEmptyString::new(tenant).unwrap();
+            let tenant = Tenant::new_nonempty(tenant);
 
             let config = GlobalConfig::instance();
             UserApiProvider::try_create_simple(config.meta.to_meta_grpc_client_conf(), &tenant)

src/meta/api/src/crud/mod.rs

@@ -30,7 +30,6 @@ use databend_common_meta_kvapi::kvapi::ValueWithName;
 use databend_common_meta_types::MatchSeq;
 use databend_common_meta_types::MatchSeqExt;
 use databend_common_meta_types::MetaError;
-use databend_common_meta_types::NonEmptyString;
 use databend_common_meta_types::SeqV;
 use databend_common_meta_types::SeqValue;
 use databend_common_meta_types::With;
@@ -58,13 +57,10 @@ pub struct CrudMgr<R> {
 
 impl<R> CrudMgr<R> {
     /// Create a new `CrudMgr` instance providing CRUD access for a key space defined by `R`: [`TenantResource`].
-    pub fn create(
-        kv_api: Arc<dyn kvapi::KVApi<Error = MetaError>>,
-        tenant: &NonEmptyString,
-    ) -> Self {
+    pub fn create(kv_api: Arc<dyn kvapi::KVApi<Error = MetaError>>, tenant: &Tenant) -> Self {
         CrudMgr {
             kv_api,
-            tenant: Tenant::new_nonempty(tenant.clone()),
+            tenant: tenant.clone(),
             _p: Default::default(),
         }
     }

src/meta/api/src/schema_api_impl.rs

@@ -41,6 +41,7 @@ use databend_common_meta_app::app_error::StreamVersionMismatched;
 use databend_common_meta_app::app_error::TableAlreadyExists;
 use databend_common_meta_app::app_error::TableLockExpired;
 use databend_common_meta_app::app_error::TableVersionMismatched;
+use databend_common_meta_app::app_error::TenantIsEmpty;
 use databend_common_meta_app::app_error::UndropDbHasNoHistory;
 use databend_common_meta_app::app_error::UndropDbWithNoDropTime;
 use databend_common_meta_app::app_error::UndropTableAlreadyExists;
@@ -180,6 +181,8 @@ use databend_common_meta_app::share::ShareGrantObject;
 use databend_common_meta_app::share::ShareNameIdent;
 use databend_common_meta_app::share::ShareSpec;
 use databend_common_meta_app::share::ShareTableInfoMap;
+use databend_common_meta_app::tenant::Tenant;
+use databend_common_meta_app::KeyWithTenant;
 use databend_common_meta_kvapi::kvapi;
 use databend_common_meta_kvapi::kvapi::Key;
 use databend_common_meta_kvapi::kvapi::UpsertKVReq;
@@ -193,6 +196,7 @@ use databend_common_meta_types::MatchSeqExt;
 use databend_common_meta_types::MetaError;
 use databend_common_meta_types::MetaId;
 use databend_common_meta_types::MetaNetworkError;
+use databend_common_meta_types::NonEmptyString;
 use databend_common_meta_types::Operation;
 use databend_common_meta_types::SeqV;
 use databend_common_meta_types::TxnCondition;
@@ -3891,8 +3895,8 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
                 } else {
                     Err(KVAppError::AppError(AppError::CatalogAlreadyExists(
                         CatalogAlreadyExists::new(
-                            &name_key.catalog_name,
-                            format!("create catalog: tenant: {}", name_key.tenant),
+                            name_key.name(),
+                            format!("create catalog: tenant: {}", name_key.tenant_name()),
                         ),
                     )))
                 };
@@ -3971,8 +3975,12 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
         loop {
             trials.next().unwrap()?.await;
 
-            let res =
-                get_catalog_or_err(self, name_key, format!("drop_catalog: {}", &name_key)).await;
+            let res = get_catalog_or_err(
+                self,
+                name_key,
+                format!("drop_catalog: {}", name_key.display()),
+            )
+            .await;
 
             let (_, catalog_id, catalog_meta_seq, _) = match res {
                 Ok(x) => x,
@@ -4040,11 +4048,11 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
     ) -> Result<Vec<Arc<CatalogInfo>>, KVAppError> {
         debug!(req :? =(&req); "SchemaApi: {}", func_name!());
 
-        let name_key = CatalogNameIdent {
-            tenant: req.tenant,
-            // Using a empty catalog to to list all
-            catalog_name: "".to_string(),
-        };
+        let tenant = Tenant::new_nonempty(
+            NonEmptyString::new(req.tenant)
+                .map_err(|_e| AppError::from(TenantIsEmpty::new("SchemaApi::list_catalogs")))?,
+        );
+        let name_key = CatalogNameIdent::new(tenant, "");
 
         // Pairs of catalog-name and catalog_id with seq
         let (tenant_catalog_names, catalog_ids) = list_u64_value(self, &name_key).await?;
@@ -4075,10 +4083,10 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
                         catalog_id: catalog_ids[i],
                     }
                     .into(),
-                    name_ident: CatalogNameIdent {
-                        tenant: name_key.tenant.clone(),
-                        catalog_name: tenant_catalog_names[i].catalog_name.clone(),
-                    }
+                    name_ident: CatalogNameIdent::new(
+                        name_key.tenant().clone(),
+                        tenant_catalog_names[i].name(),
+                    )
                     .into(),
                     meta: catalog_meta,
                 };
@@ -5493,8 +5501,8 @@ pub fn catalog_has_to_exist(
 
         Err(KVAppError::AppError(AppError::UnknownCatalog(
             UnknownCatalog::new(
-                &catalog_name_ident.catalog_name,
-                format!("{}: {}", msg, catalog_name_ident),
+                catalog_name_ident.name(),
+                format!("{}: {}", msg, catalog_name_ident.display()),
             ),
         )))
     } else {

src/meta/api/src/schema_api_test_suite.rs

@@ -1419,16 +1419,17 @@ impl SchemaApiTestSuite {
 
     #[minitrace::trace]
     async fn catalog_create_get_list_drop<MT: SchemaApi>(&self, mt: &MT) -> anyhow::Result<()> {
-        let tenant = "tenant1";
+        let tenant_name = "tenant1";
+        let tenant = Tenant::new_literal(tenant_name);
+
         let catalog_name = "catalog1";
 
+        let ident = CatalogNameIdent::new(tenant.clone(), catalog_name);
+
         info!("--- create catalog1");
         let req = CreateCatalogReq {
             if_not_exists: false,
-            name_ident: CatalogNameIdent {
-                tenant: tenant.to_string(),
-                catalog_name: catalog_name.to_string(),
-            },
+            name_ident: ident.clone(),
             meta: CatalogMeta {
                 catalog_option: CatalogOption::Iceberg(IcebergCatalogOption {
                     storage_params: Box::new(StorageParams::S3(StorageS3Config {
@@ -1443,9 +1444,7 @@ impl SchemaApiTestSuite {
         let res = mt.create_catalog(req).await?;
         info!("create catalog res: {:?}", res);
 
-        let got = mt
-            .get_catalog(GetCatalogReq::new(tenant, catalog_name))
-            .await?;
+        let got = mt.get_catalog(GetCatalogReq::new(ident.clone())).await?;
         assert_eq!(got.id.catalog_id, res.catalog_id);
         assert_eq!(got.name_ident.tenant, "tenant1");
         assert_eq!(got.name_ident.catalog_name, "catalog1");
@@ -1458,10 +1457,7 @@ impl SchemaApiTestSuite {
         let _ = mt
             .drop_catalog(DropCatalogReq {
                 if_exists: false,
-                name_ident: CatalogNameIdent {
-                    tenant: tenant.to_string(),
-                    catalog_name: catalog_name.to_string(),
-                },
+                name_ident: ident.clone(),
             })
             .await?;
 

src/meta/app/src/app_error.rs

@@ -26,6 +26,20 @@ pub trait AppErrorMessage: Display {
     }
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, thiserror::Error)]
+#[error("Tenant is empty when: `{context}`")]
+pub struct TenantIsEmpty {
+    context: String,
+}
+
+impl TenantIsEmpty {
+    pub fn new(context: impl Into<String>) -> Self {
+        Self {
+            context: context.into(),
+        }
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, thiserror::Error)]
 #[error("DatabaseAlreadyExists: `{db_name}` while `{context}`")]
 pub struct DatabaseAlreadyExists {
@@ -890,6 +904,9 @@ impl VirtualColumnNotFound {
 /// The application does not get expected result but there is nothing wrong with meta-service.
 #[derive(thiserror::Error, serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq)]
 pub enum AppError {
+    #[error(transparent)]
+    TenantIsEmpty(#[from] TenantIsEmpty),
+
     #[error(transparent)]
     TableVersionMismatched(#[from] TableVersionMismatched),
 
@@ -1051,6 +1068,12 @@ pub enum AppError {
     MultiStatementTxnCommitFailed(#[from] MultiStmtTxnCommitFailed),
 }
 
+impl AppErrorMessage for TenantIsEmpty {
+    fn message(&self) -> String {
+        self.to_string()
+    }
+}
+
 impl AppErrorMessage for UnknownBackgroundJob {
     fn message(&self) -> String {
         format!("Unknown background job '{}'", self.name)
@@ -1380,6 +1403,7 @@ impl AppErrorMessage for VirtualColumnAlreadyExists {
 impl From<AppError> for ErrorCode {
     fn from(app_err: AppError) -> Self {
         match app_err {
+            AppError::TenantIsEmpty(err) => ErrorCode::TenantIsEmpty(err.message()),
             AppError::UnknownDatabase(err) => ErrorCode::UnknownDatabase(err.message()),
             AppError::UnknownDatabaseId(err) => ErrorCode::UnknownDatabaseId(err.message()),
             AppError::UnknownTableId(err) => ErrorCode::UnknownTableId(err.message()),

src/meta/app/src/principal/user_defined_function.rs

@@ -20,16 +20,18 @@ use chrono::Utc;
 use databend_common_expression::types::DataType;
 use databend_common_meta_kvapi::kvapi::Key;
 
+use crate::tenant::Tenant;
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct UdfName {
-    pub tenant: String,
+    pub tenant: Tenant,
     pub name: String,
 }
 
 impl UdfName {
-    pub fn new(tenant: impl ToString, name: impl ToString) -> Self {
+    pub fn new(tenant: &Tenant, name: impl ToString) -> Self {
         Self {
-            tenant: tenant.to_string(),
+            tenant: tenant.clone(),
             name: name.to_string(),
         }
     }
@@ -224,18 +226,20 @@ mod kv_api_impl {
 
         fn to_string_key(&self) -> String {
             kvapi::KeyBuilder::new_prefixed(Self::PREFIX)
-                .push_str(&self.tenant)
+                .push_str(self.tenant.name())
                 .push_str(&self.name)
                 .done()
         }
 
         fn from_str_key(s: &str) -> Result<Self, kvapi::KeyError> {
             let mut p = kvapi::KeyParser::new_prefixed(s, Self::PREFIX)?;
 
-            let tenant = p.next_str()?;
+            let tenant = p.next_nonempty()?;
             let name = p.next_str()?;
             p.done()?;
 
+            let tenant = Tenant::new_nonempty(tenant);
+
             Ok(UdfName { tenant, name })
         }
     }