refactor: remove catalog prefix (#3835)

* refactor: remove catalog prefix

* refactor: remove scope.rs

* fix: fix tests

* chore: update comments

* chore: apply suggestions from CR

src/common/meta/src/key.rs

@@ -36,16 +36,16 @@
 //!     - The value is a [TableNameValue] struct; it contains the table id.
 //!     - Used in the table name to table id lookup.
 //!
-//! 6. Flow info key: `__flow/{catalog}/info/{flow_id}`
+//! 6. Flow info key: `__flow/info/{flow_id}`
 //!     - Stores metadata of the flow.
 //!
-//! 7. Flow name key: `__flow/{catalog}/name/{flow_name}`
+//! 7. Flow name key: `__flow/name/{catalog}/{flow_name}`
 //!     - Mapping {catalog}/{flow_name} to {flow_id}
 //!
-//! 8. Flownode flow key: `__flow/{catalog}/flownode/{flownode_id}/{flow_id}/{partition_id}`
+//! 8. Flownode flow key: `__flow/flownode/{flownode_id}/{flow_id}/{partition_id}`
 //!     - Mapping {flownode_id} to {flow_id}
 //!
-//! 9. Table flow key: `__table_flow/{catalog}/source_table/{table_id}/{flownode_id}/{flow_id}/{partition_id}`
+//! 9. Table flow key: `__flow/source_table/{table_id}/{flownode_id}/{flow_id}/{partition_id}`
 //!     - Mapping source table's {table_id} to {flownode_id}
 //!     - Used in `Flownode` booting.
 //!
@@ -60,12 +60,12 @@
 //! The whole picture of flow keys will be like this:
 //!
 //! __flow/
-//!  {catalog}/
-//!    info/
-//!      {tsak_id}
+//!   info/
+//!     {flow_id}
 //!
 //!    name/
-//!      {flow_name}
+//!      {catalog_name}
+//!        {flow_name}
 //!
 //!    flownode/
 //!      {flownode_id}/
@@ -84,7 +84,6 @@ pub mod datanode_table;
 #[allow(unused)]
 pub mod flow;
 pub mod schema_name;
-pub mod scope;
 pub mod table_info;
 pub mod table_name;
 // TODO(weny): removes it.
@@ -194,6 +193,32 @@ pub trait TableMetaKey {
     fn as_raw_key(&self) -> Vec<u8>;
 }
 
+/// The key of metadata.
+pub trait MetaKey<T> {
+    fn to_bytes(&self) -> Vec<u8>;
+
+    fn from_bytes(bytes: &[u8]) -> Result<T>;
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub struct BytesAdapter(Vec<u8>);
+
+impl From<Vec<u8>> for BytesAdapter {
+    fn from(value: Vec<u8>) -> Self {
+        Self(value)
+    }
+}
+
+impl MetaKey<BytesAdapter> for BytesAdapter {
+    fn to_bytes(&self) -> Vec<u8> {
+        self.0.clone()
+    }
+
+    fn from_bytes(bytes: &[u8]) -> Result<BytesAdapter> {
+        Ok(BytesAdapter(bytes.to_vec()))
+    }
+}
+
 pub(crate) trait TableMetaKeyGetTxnOp {
     fn build_get_op(
         &self,

src/common/meta/src/key/catalog_name.rs

@@ -82,7 +82,7 @@ impl<'a> TryFrom<&'a str> for CatalogNameKey<'a> {
     }
 }
 
-/// Decoder `KeyValue` to ({catalog},())
+/// Decoder `KeyValue` to {catalog}
 pub fn catalog_decoder(kv: KeyValue) -> Result<String> {
     let str = std::str::from_utf8(&kv.key).context(error::ConvertRawKeySnafu)?;
     let catalog_name = CatalogNameKey::try_from(str)?;

src/common/meta/src/key/flow.rs

@@ -30,9 +30,8 @@ use crate::key::flow::flow_info::FlowInfoManager;
 use crate::key::flow::flow_name::FlowNameManager;
 use crate::key::flow::flownode_flow::FlownodeFlowManager;
 use crate::key::flow::table_flow::TableFlowManager;
-use crate::key::scope::MetaKey;
 use crate::key::txn_helper::TxnOpGetResponseSet;
-use crate::key::FlowId;
+use crate::key::{FlowId, MetaKey};
 use crate::kv_backend::txn::Txn;
 use crate::kv_backend::KvBackendRef;
 
@@ -139,20 +138,15 @@ impl FlowMetadataManager {
             .flow_name_manager
             .build_create_txn(&flow_value.catalog_name, &flow_value.flow_name, flow_id)?;
 
-        let (create_flow_txn, on_create_flow_failure) = self.flow_info_manager.build_create_txn(
-            &flow_value.catalog_name,
-            flow_id,
-            &flow_value,
-        )?;
+        let (create_flow_txn, on_create_flow_failure) = self
+            .flow_info_manager
+            .build_create_txn(flow_id, &flow_value)?;
 
-        let create_flownode_flow_txn = self.flownode_flow_manager.build_create_txn(
-            &flow_value.catalog_name,
-            flow_id,
-            flow_value.flownode_ids().clone(),
-        );
+        let create_flownode_flow_txn = self
+            .flownode_flow_manager
+            .build_create_txn(flow_id, flow_value.flownode_ids().clone());
 
         let create_table_flow_txn = self.table_flow_manager.build_create_txn(
-            &flow_value.catalog_name,
             flow_id,
             flow_value.flownode_ids().clone(),
             flow_value.source_table_ids(),
@@ -222,7 +216,6 @@ mod tests {
 
     use super::*;
     use crate::key::flow::table_flow::TableFlowKey;
-    use crate::key::scope::CatalogScoped;
     use crate::kv_backend::memory::MemoryKvBackend;
     use crate::table_name::TableName;
 
@@ -245,27 +238,23 @@ mod tests {
 
     #[test]
     fn test_flow_scoped_to_bytes() {
-        let key = FlowScoped::new(CatalogScoped::new(
-            "my_catalog".to_string(),
-            MockKey {
-                inner: b"hi".to_vec(),
-            },
-        ));
-        assert_eq!(b"__flow/my_catalog/hi".to_vec(), key.to_bytes());
+        let key = FlowScoped::new(MockKey {
+            inner: b"hi".to_vec(),
+        });
+        assert_eq!(b"__flow/hi".to_vec(), key.to_bytes());
     }
 
     #[test]
     fn test_flow_scoped_from_bytes() {
-        let bytes = b"__flow/my_catalog/hi";
-        let key = FlowScoped::<CatalogScoped<MockKey>>::from_bytes(bytes).unwrap();
-        assert_eq!(key.catalog(), "my_catalog");
+        let bytes = b"__flow/hi";
+        let key = FlowScoped::<MockKey>::from_bytes(bytes).unwrap();
         assert_eq!(key.inner.inner, b"hi".to_vec());
     }
 
     #[test]
     fn test_flow_scoped_from_bytes_mismatch() {
-        let bytes = b"__table/my_catalog/hi";
-        let err = FlowScoped::<CatalogScoped<MockKey>>::from_bytes(bytes).unwrap_err();
+        let bytes = b"__table/hi";
+        let err = FlowScoped::<MockKey>::from_bytes(bytes).unwrap_err();
         assert_matches!(err, error::Error::MismatchPrefix { .. });
     }
 
@@ -302,35 +291,26 @@ mod tests {
             .unwrap();
         let got = flow_metadata_manager
             .flow_info_manager()
-            .get(catalog_name, flow_id)
+            .get(flow_id)
             .await
             .unwrap()
             .unwrap();
         assert_eq!(got, flow_value);
         let flows = flow_metadata_manager
             .flownode_flow_manager()
-            .flows(catalog_name, 1)
+            .flows(1)
             .try_collect::<Vec<_>>()
             .await
             .unwrap();
         assert_eq!(flows, vec![(flow_id, 0)]);
         for table_id in [1024, 1025, 1026] {
             let nodes = flow_metadata_manager
                 .table_flow_manager()
-                .nodes(catalog_name, table_id)
+                .nodes(table_id)
                 .try_collect::<Vec<_>>()
                 .await
                 .unwrap();
-            assert_eq!(
-                nodes,
-                vec![TableFlowKey::new(
-                    catalog_name.to_string(),
-                    table_id,
-                    1,
-                    flow_id,
-                    0
-                )]
-            );
+            assert_eq!(nodes, vec![TableFlowKey::new(table_id, 1, flow_id, 0)]);
         }
     }
 

src/common/meta/src/key/flow/flow_info.rs

@@ -22,9 +22,10 @@ use table::metadata::TableId;
 
 use crate::error::{self, Result};
 use crate::key::flow::FlowScoped;
-use crate::key::scope::{CatalogScoped, MetaKey};
 use crate::key::txn_helper::TxnOpGetResponseSet;
-use crate::key::{txn_helper, DeserializedValueWithBytes, FlowId, FlowPartitionId, TableMetaValue};
+use crate::key::{
+    txn_helper, DeserializedValueWithBytes, FlowId, FlowPartitionId, MetaKey, TableMetaValue,
+};
 use crate::kv_backend::txn::Txn;
 use crate::kv_backend::KvBackendRef;
 use crate::table_name::TableName;
@@ -39,31 +40,26 @@ lazy_static! {
 
 /// The key stores the metadata of the flow.
 ///
-/// The layout: `__flow/{catalog}/info/{flow_id}`.
-pub struct FlowInfoKey(FlowScoped<CatalogScoped<FlowInfoKeyInner>>);
+/// The layout: `__flow/info/{flow_id}`.
+pub struct FlowInfoKey(FlowScoped<FlowInfoKeyInner>);
 
 impl MetaKey<FlowInfoKey> for FlowInfoKey {
     fn to_bytes(&self) -> Vec<u8> {
         self.0.to_bytes()
     }
 
     fn from_bytes(bytes: &[u8]) -> Result<FlowInfoKey> {
-        Ok(FlowInfoKey(
-            FlowScoped::<CatalogScoped<FlowInfoKeyInner>>::from_bytes(bytes)?,
-        ))
+        Ok(FlowInfoKey(FlowScoped::<FlowInfoKeyInner>::from_bytes(
+            bytes,
+        )?))
     }
 }
 
 impl FlowInfoKey {
     /// Returns the [FlowInfoKey].
-    pub fn new(catalog: String, flow_id: FlowId) -> FlowInfoKey {
+    pub fn new(flow_id: FlowId) -> FlowInfoKey {
         let inner = FlowInfoKeyInner::new(flow_id);
-        FlowInfoKey(FlowScoped::new(CatalogScoped::new(catalog, inner)))
-    }
-
-    /// Returns the catalog.
-    pub fn catalog(&self) -> &str {
-        self.0.catalog()
+        FlowInfoKey(FlowScoped::new(inner))
     }
 
     /// Returns the [FlowId].
@@ -159,8 +155,8 @@ impl FlowInfoManager {
     }
 
     /// Returns the [FlowInfoValue] of specified `flow_id`.
-    pub async fn get(&self, catalog: &str, flow_id: FlowId) -> Result<Option<FlowInfoValue>> {
-        let key = FlowInfoKey::new(catalog.to_string(), flow_id).to_bytes();
+    pub async fn get(&self, flow_id: FlowId) -> Result<Option<FlowInfoValue>> {
+        let key = FlowInfoKey::new(flow_id).to_bytes();
         self.kv_backend
             .get(&key)
             .await?
@@ -169,11 +165,10 @@ impl FlowInfoManager {
     }
 
     /// Builds a create flow transaction.
-    /// It is expected that the `__flow/{catalog}/info/{flow_id}` wasn't occupied.
+    /// It is expected that the `__flow/info/{flow_id}` wasn't occupied.
     /// Otherwise, the transaction will retrieve existing value.
     pub(crate) fn build_create_txn(
         &self,
-        catalog: &str,
         flow_id: FlowId,
         flow_value: &FlowInfoValue,
     ) -> Result<(
@@ -182,7 +177,7 @@ impl FlowInfoManager {
             &mut TxnOpGetResponseSet,
         ) -> Result<Option<DeserializedValueWithBytes<FlowInfoValue>>>,
     )> {
-        let key = FlowInfoKey::new(catalog.to_string(), flow_id).to_bytes();
+        let key = FlowInfoKey::new(flow_id).to_bytes();
         let txn = txn_helper::build_put_if_absent_txn(key.clone(), flow_value.try_as_raw_value()?);
 
         Ok((
@@ -198,15 +193,14 @@ mod tests {
 
     #[test]
     fn test_key_serialization() {
-        let flow_info = FlowInfoKey::new("my_catalog".to_string(), 2);
-        assert_eq!(b"__flow/my_catalog/info/2".to_vec(), flow_info.to_bytes());
+        let flow_info = FlowInfoKey::new(2);
+        assert_eq!(b"__flow/info/2".to_vec(), flow_info.to_bytes());
     }
 
     #[test]
     fn test_key_deserialization() {
-        let bytes = b"__flow/my_catalog/info/2".to_vec();
+        let bytes = b"__flow/info/2".to_vec();
         let key = FlowInfoKey::from_bytes(&bytes).unwrap();
-        assert_eq!(key.catalog(), "my_catalog");
         assert_eq!(key.flow_id(), 2);
     }
 }