refactor: remove catalog prefix

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);
  }
 }