refactor: remove serde from CatalogId and CatalogNameIdent (#15068)

src/meta/api/src/schema_api_impl.rs

@@ -3952,8 +3952,8 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
             get_catalog_or_err(self, name_key, "get_catalog").await?;
 
         let catalog = CatalogInfo {
-            id: CatalogId { catalog_id },
-            name_ident: name_key.clone(),
+            id: CatalogId { catalog_id }.into(),
+            name_ident: name_key.clone().into(),
             meta: catalog_meta,
         };
 
@@ -4073,11 +4073,13 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
                 let catalog_info = CatalogInfo {
                     id: CatalogId {
                         catalog_id: catalog_ids[i],
-                    },
+                    }
+                    .into(),
                     name_ident: CatalogNameIdent {
                         tenant: name_key.tenant.clone(),
                         catalog_name: tenant_catalog_names[i].catalog_name.clone(),
-                    },
+                    }
+                    .into(),
                     meta: catalog_meta,
                 };
                 catalog_infos.push(Arc::new(catalog_info));

src/meta/app/src/schema/catalog.rs

@@ -73,13 +73,59 @@ pub struct IcebergCatalogOption {
     pub storage_params: Box<StorageParams>,
 }
 
+/// Same as `CatalogNameIdent`, but with `serde` support,
+/// and can be used a s part of a value.
+#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq, Eq)]
+pub struct CatalogName {
+    pub tenant: String,
+    pub catalog_name: String,
+}
+
+impl From<CatalogNameIdent> for CatalogName {
+    fn from(ident: CatalogNameIdent) -> Self {
+        CatalogName {
+            tenant: ident.tenant,
+            catalog_name: ident.catalog_name,
+        }
+    }
+}
+
+impl Display for CatalogName {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "'{}'/'{}'", self.tenant, self.catalog_name)
+    }
+}
+
+// serde is required by `DataSourcePlan.catalog_info`
+// serde is required by `CommitSink.catalog_info`
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Eq, PartialEq)]
 pub struct CatalogInfo {
-    pub id: CatalogId,
-    pub name_ident: CatalogNameIdent,
+    pub id: catalog_info::CatalogId,
+    pub name_ident: CatalogName,
     pub meta: CatalogMeta,
 }
 
+/// Private types for `CatalogInfo`.
+mod catalog_info {
+
+    /// Same as [`crate::schema::CatalogId`], except with serde support, and can be used in a value,
+    /// while CatalogId is only used for Key.
+    ///
+    /// This type is sealed in a private mod so that it is pub for use but can not be created directly.
+    #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Default, Eq, PartialEq)]
+    pub struct CatalogId {
+        pub catalog_id: u64,
+    }
+
+    impl From<crate::schema::CatalogId> for CatalogId {
+        fn from(value: crate::schema::CatalogId) -> Self {
+            Self {
+                catalog_id: value.catalog_id,
+            }
+        }
+    }
+}
+
 impl CatalogInfo {
     /// Get the catalog type via catalog info.
     pub fn catalog_type(&self) -> CatalogType {
@@ -94,8 +140,8 @@ impl CatalogInfo {
     /// Create a new default catalog info.
     pub fn new_default() -> CatalogInfo {
         Self {
-            id: CatalogId { catalog_id: 0 },
-            name_ident: CatalogNameIdent {
+            id: CatalogId { catalog_id: 0 }.into(),
+            name_ident: CatalogName {
                 // tenant for default catalog is not used.
                 tenant: "".to_string(),
                 catalog_name: "default".to_string(),
@@ -114,13 +160,16 @@ pub struct CatalogMeta {
     pub created_on: DateTime<Utc>,
 }
 
-#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq, Eq)]
+/// The name of a catalog,
+/// which is used as a key and does not support other codec method such as serde.
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub struct CatalogNameIdent {
     pub tenant: String,
     pub catalog_name: String,
 }
 
-#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Default, Eq, PartialEq)]
+// serde is required by `CatalogInfo.id`
+#[derive(Clone, Debug, Default, Eq, PartialEq)]
 pub struct CatalogId {
     pub catalog_id: u64,
 }
@@ -148,7 +197,7 @@ impl Display for CatalogIdToName {
     }
 }
 
-#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub struct CreateCatalogReq {
     pub if_not_exists: bool,
     pub name_ident: CatalogNameIdent,
@@ -198,7 +247,7 @@ impl Display for DropCatalogReq {
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq, Eq)]
 pub struct DropCatalogReply {}
 
-#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub struct GetCatalogReq {
     pub inner: CatalogNameIdent,
 }

src/meta/app/src/schema/mod.rs

@@ -14,7 +14,7 @@
 
 //! Schema types
 
-mod catalog;
+pub mod catalog;
 mod create_option;
 mod database;
 mod index;

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

@@ -98,11 +98,12 @@ impl CatalogManager {
                 ErrorCode::BadArguments(format!("unknown catalog type: {:?}", CatalogType::Hive))
             })?;
             let ctl = creator.try_create(&CatalogInfo {
-                id: CatalogId { catalog_id: 0 },
+                id: CatalogId { catalog_id: 0 }.into(),
                 name_ident: CatalogNameIdent {
                     tenant: tenant.to_string(),
                     catalog_name: name.clone(),
-                },
+                }
+                .into(),
                 meta: CatalogMeta {
                     catalog_option: CatalogOption::Hive(HiveCatalogOption {
                         address: hive_ctl_cfg.metastore_address.clone(),

src/query/service/src/interpreters/interpreter_catalog_create.rs

@@ -71,11 +71,12 @@ impl Interpreter for CreateCatalogInterpreter {
         // Build and check if catalog is valid.
         let ctl = catalog_manager
             .build_catalog(&CatalogInfo {
-                id: CatalogId::default(),
+                id: CatalogId::default().into(),
                 name_ident: CatalogNameIdent {
                     tenant: self.plan.tenant.clone(),
                     catalog_name: self.plan.catalog.clone(),
-                },
+                }
+                .into(),
                 meta: CatalogMeta {
                     catalog_option: self.plan.meta.catalog_option.clone(),
                     created_on: chrono::Utc::now(),

src/query/sql/src/executor/physical_plans/physical_commit_sink.rs

@@ -22,6 +22,7 @@ use databend_storages_common_table_meta::meta::TableSnapshot;
 use crate::executor::physical_plans::common::MutationKind;
 use crate::executor::PhysicalPlan;
 
+// serde is required by `PhysicalPlan`
 /// The commit sink is used to commit the data to the table.
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug)]
 pub struct CommitSink {