diff --git a/src/common/meta/src/cache_invalidator.rs b/src/common/meta/src/cache_invalidator.rs
index 87bef941f9d2..7c74fa14f79b 100644
--- a/src/common/meta/src/cache_invalidator.rs
+++ b/src/common/meta/src/cache_invalidator.rs
@@ -24,7 +24,7 @@ use crate::key::table_info::TableInfoKey;
 use crate::key::table_name::TableNameKey;
 use crate::key::table_route::TableRouteKey;
 use crate::key::view_info::ViewInfoKey;
-use crate::key::MetaKey;
+use crate::key::MetadataKey;
 
 /// KvBackend cache invalidator
 #[async_trait::async_trait]
diff --git a/src/common/meta/src/error.rs b/src/common/meta/src/error.rs
index 4667a9ef8914..6de3af36dee3 100644
--- a/src/common/meta/src/error.rs
+++ b/src/common/meta/src/error.rs
@@ -392,8 +392,8 @@ pub enum Error {
         location: Location,
     },
 
-    #[snafu(display("Invalid table metadata, err: {}", err_msg))]
-    InvalidTableMetadata {
+    #[snafu(display("Invalid metadata, err: {}", err_msg))]
+    InvalidMetadata {
         err_msg: String,
         #[snafu(implicit)]
         location: Location,
@@ -702,7 +702,7 @@ impl ErrorExt for Error {
             | ParseOption { .. }
             | RouteInfoCorrupted { .. }
             | InvalidProtoMsg { .. }
-            | InvalidTableMetadata { .. }
+            | InvalidMetadata { .. }
             | MoveRegion { .. }
             | Unexpected { .. }
             | TableInfoNotFound { .. }
diff --git a/src/common/meta/src/key.rs b/src/common/meta/src/key.rs
index ad36607c1d89..df9c8161fa7e 100644
--- a/src/common/meta/src/key.rs
+++ b/src/common/meta/src/key.rs
@@ -211,7 +211,7 @@ lazy_static! {
 }
 
 /// The key of metadata.
-pub trait MetaKey<'a, T> {
+pub trait MetadataKey<'a, T> {
     fn to_bytes(&self) -> Vec<u8>;
 
     fn from_bytes(bytes: &'a [u8]) -> Result<T>;
@@ -226,7 +226,7 @@ impl From<Vec<u8>> for BytesAdapter {
     }
 }
 
-impl<'a> MetaKey<'a, BytesAdapter> for BytesAdapter {
+impl<'a> MetadataKey<'a, BytesAdapter> for BytesAdapter {
     fn to_bytes(&self) -> Vec<u8> {
         self.0.clone()
     }
@@ -236,7 +236,7 @@ impl<'a> MetaKey<'a, BytesAdapter> for BytesAdapter {
     }
 }
 
-pub(crate) trait TableMetaKeyGetTxnOp {
+pub(crate) trait MetadataKeyGetTxnOp {
     fn build_get_op(
         &self,
     ) -> (
@@ -245,7 +245,7 @@ pub(crate) trait TableMetaKeyGetTxnOp {
     );
 }
 
-pub trait TableMetaValue {
+pub trait MetadataValue {
     fn try_from_raw_value(raw_value: &[u8]) -> Result<Self>
     where
         Self: Sized;
@@ -330,7 +330,7 @@ impl<T: DeserializeOwned + Serialize> Serialize for DeserializedValueWithBytes<T
     }
 }
 
-impl<'de, T: DeserializeOwned + Serialize + TableMetaValue> Deserialize<'de>
+impl<'de, T: DeserializeOwned + Serialize + MetadataValue> Deserialize<'de>
     for DeserializedValueWithBytes<T>
 {
     /// - Deserialize behaviors:
@@ -359,7 +359,7 @@ impl<T: Serialize + DeserializeOwned + Clone> Clone for DeserializedValueWithByt
     }
 }
 
-impl<T: Serialize + DeserializeOwned + TableMetaValue> DeserializedValueWithBytes<T> {
+impl<T: Serialize + DeserializeOwned + MetadataValue> DeserializedValueWithBytes<T> {
     /// Returns a struct containing a deserialized value and an original `bytes`.
     /// It accepts original bytes of inner.
     pub fn from_inner_bytes(bytes: Bytes) -> Result<Self> {
@@ -1156,10 +1156,10 @@ impl TableMetadataManager {
 }
 
 #[macro_export]
-macro_rules! impl_table_meta_value {
+macro_rules! impl_metadata_value {
     ($($val_ty: ty), *) => {
         $(
-            impl $crate::key::TableMetaValue for $val_ty {
+            impl $crate::key::MetadataValue for $val_ty {
                 fn try_from_raw_value(raw_value: &[u8]) -> Result<Self> {
                     serde_json::from_slice(raw_value).context(SerdeJsonSnafu)
                 }
@@ -1172,10 +1172,10 @@ macro_rules! impl_table_meta_value {
     }
 }
 
-macro_rules! impl_meta_key_get_txn_op {
+macro_rules! impl_metadata_key_get_txn_op {
     ($($key: ty), *) => {
         $(
-            impl $crate::key::TableMetaKeyGetTxnOp for $key {
+            impl $crate::key::MetadataKeyGetTxnOp for $key {
                 /// Returns a [TxnOp] to retrieve the corresponding value
                 /// and a filter to retrieve the value from the [TxnOpGetResponseSet]
                 fn build_get_op(
@@ -1197,7 +1197,7 @@ macro_rules! impl_meta_key_get_txn_op {
     }
 }
 
-impl_meta_key_get_txn_op! {
+impl_metadata_key_get_txn_op! {
     TableNameKey<'_>,
     TableInfoKey,
     ViewInfoKey,
@@ -1206,7 +1206,7 @@ impl_meta_key_get_txn_op! {
 }
 
 #[macro_export]
-macro_rules! impl_optional_meta_value {
+macro_rules! impl_optional_metadata_value {
     ($($val_ty: ty), *) => {
         $(
             impl $val_ty {
@@ -1222,7 +1222,7 @@ macro_rules! impl_optional_meta_value {
     }
 }
 
-impl_table_meta_value! {
+impl_metadata_value! {
     TableNameValue,
     TableInfoValue,
     ViewInfoValue,
@@ -1233,7 +1233,7 @@ impl_table_meta_value! {
     TableFlowValue
 }
 
-impl_optional_meta_value! {
+impl_optional_metadata_value! {
     CatalogNameValue,
     SchemaNameValue
 }
diff --git a/src/common/meta/src/key/catalog_name.rs b/src/common/meta/src/key/catalog_name.rs
index fd67acf34800..a3a98416188a 100644
--- a/src/common/meta/src/key/catalog_name.rs
+++ b/src/common/meta/src/key/catalog_name.rs
@@ -20,8 +20,8 @@ use futures::stream::BoxStream;
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt};
 
-use crate::error::{self, Error, InvalidTableMetadataSnafu, Result};
-use crate::key::{MetaKey, CATALOG_NAME_KEY_PATTERN, CATALOG_NAME_KEY_PREFIX};
+use crate::error::{self, Error, InvalidMetadataSnafu, Result};
+use crate::key::{MetadataKey, CATALOG_NAME_KEY_PATTERN, CATALOG_NAME_KEY_PREFIX};
 use crate::kv_backend::KvBackendRef;
 use crate::range_stream::{PaginationStream, DEFAULT_PAGE_SIZE};
 use crate::rpc::store::RangeRequest;
@@ -56,14 +56,14 @@ impl<'a> CatalogNameKey<'a> {
     }
 }
 
-impl<'a> MetaKey<'a, CatalogNameKey<'a>> for CatalogNameKey<'_> {
+impl<'a> MetadataKey<'a, CatalogNameKey<'a>> for CatalogNameKey<'_> {
     fn to_bytes(&self) -> Vec<u8> {
         self.to_string().into_bytes()
     }
 
     fn from_bytes(bytes: &'a [u8]) -> Result<CatalogNameKey<'a>> {
         let key = std::str::from_utf8(bytes).map_err(|e| {
-            InvalidTableMetadataSnafu {
+            InvalidMetadataSnafu {
                 err_msg: format!(
                     "CatalogNameKey '{}' is not a valid UTF8 string: {e}",
                     String::from_utf8_lossy(bytes)
@@ -87,7 +87,7 @@ impl<'a> TryFrom<&'a str> for CatalogNameKey<'a> {
     fn try_from(s: &'a str) -> Result<Self> {
         let captures = CATALOG_NAME_KEY_PATTERN
             .captures(s)
-            .context(InvalidTableMetadataSnafu {
+            .context(InvalidMetadataSnafu {
                 err_msg: format!("Illegal CatalogNameKey format: '{s}'"),
             })?;
 
diff --git a/src/common/meta/src/key/datanode_table.rs b/src/common/meta/src/key/datanode_table.rs
index 57ffa6a00c29..b53060cd168c 100644
--- a/src/common/meta/src/key/datanode_table.rs
+++ b/src/common/meta/src/key/datanode_table.rs
@@ -22,10 +22,10 @@ use snafu::OptionExt;
 use store_api::storage::RegionNumber;
 use table::metadata::TableId;
 
-use super::MetaKey;
-use crate::error::{InvalidTableMetadataSnafu, Result};
+use super::MetadataKey;
+use crate::error::{InvalidMetadataSnafu, Result};
 use crate::key::{
-    RegionDistribution, TableMetaValue, DATANODE_TABLE_KEY_PATTERN, DATANODE_TABLE_KEY_PREFIX,
+    MetadataValue, RegionDistribution, DATANODE_TABLE_KEY_PATTERN, DATANODE_TABLE_KEY_PREFIX,
 };
 use crate::kv_backend::txn::{Txn, TxnOp};
 use crate::kv_backend::KvBackendRef;
@@ -77,14 +77,14 @@ impl DatanodeTableKey {
     }
 }
 
-impl<'a> MetaKey<'a, DatanodeTableKey> for DatanodeTableKey {
+impl<'a> MetadataKey<'a, DatanodeTableKey> for DatanodeTableKey {
     fn to_bytes(&self) -> Vec<u8> {
         self.to_string().into_bytes()
     }
 
     fn from_bytes(bytes: &[u8]) -> Result<DatanodeTableKey> {
         let key = std::str::from_utf8(bytes).map_err(|e| {
-            InvalidTableMetadataSnafu {
+            InvalidMetadataSnafu {
                 err_msg: format!(
                     "DatanodeTableKey '{}' is not a valid UTF8 string: {e}",
                     String::from_utf8_lossy(bytes)
@@ -92,12 +92,11 @@ impl<'a> MetaKey<'a, DatanodeTableKey> for DatanodeTableKey {
             }
             .build()
         })?;
-        let captures =
-            DATANODE_TABLE_KEY_PATTERN
-                .captures(key)
-                .context(InvalidTableMetadataSnafu {
-                    err_msg: format!("Invalid DatanodeTableKey '{key}'"),
-                })?;
+        let captures = DATANODE_TABLE_KEY_PATTERN
+            .captures(key)
+            .context(InvalidMetadataSnafu {
+                err_msg: format!("Invalid DatanodeTableKey '{key}'"),
+            })?;
         // Safety: pass the regex check above
         let datanode_id = captures[1].parse::<DatanodeId>().unwrap();
         let table_id = captures[2].parse::<TableId>().unwrap();
diff --git a/src/common/meta/src/key/flow.rs b/src/common/meta/src/key/flow.rs
index 89f68abedf1b..41a5058a331f 100644
--- a/src/common/meta/src/key/flow.rs
+++ b/src/common/meta/src/key/flow.rs
@@ -38,7 +38,7 @@ use crate::key::flow::flow_name::FlowNameManager;
 use crate::key::flow::flownode_flow::FlownodeFlowManager;
 pub use crate::key::flow::table_flow::{TableFlowManager, TableFlowManagerRef};
 use crate::key::txn_helper::TxnOpGetResponseSet;
-use crate::key::{FlowId, MetaKey};
+use crate::key::{FlowId, MetadataKey};
 use crate::kv_backend::txn::Txn;
 use crate::kv_backend::KvBackendRef;
 use crate::rpc::store::BatchDeleteRequest;
@@ -66,7 +66,7 @@ impl<T> FlowScoped<T> {
     }
 }
 
-impl<'a, T: MetaKey<'a, T>> MetaKey<'a, FlowScoped<T>> for FlowScoped<T> {
+impl<'a, T: MetadataKey<'a, T>> MetadataKey<'a, FlowScoped<T>> for FlowScoped<T> {
     fn to_bytes(&self) -> Vec<u8> {
         let prefix = FlowScoped::<T>::PREFIX.as_bytes();
         let inner = self.inner.to_bytes();
@@ -295,7 +295,7 @@ mod tests {
         inner: Vec<u8>,
     }
 
-    impl<'a> MetaKey<'a, MockKey> for MockKey {
+    impl<'a> MetadataKey<'a, MockKey> for MockKey {
         fn to_bytes(&self) -> Vec<u8> {
             self.inner.clone()
         }
diff --git a/src/common/meta/src/key/flow/flow_info.rs b/src/common/meta/src/key/flow/flow_info.rs
index 1354ec6cf49c..2057f069503e 100644
--- a/src/common/meta/src/key/flow/flow_info.rs
+++ b/src/common/meta/src/key/flow/flow_info.rs
@@ -25,7 +25,7 @@ use table::table_name::TableName;
 use crate::error::{self, Result};
 use crate::key::flow::FlowScoped;
 use crate::key::txn_helper::TxnOpGetResponseSet;
-use crate::key::{DeserializedValueWithBytes, FlowId, FlowPartitionId, MetaKey, TableMetaValue};
+use crate::key::{DeserializedValueWithBytes, FlowId, FlowPartitionId, MetadataKey, MetadataValue};
 use crate::kv_backend::txn::Txn;
 use crate::kv_backend::KvBackendRef;
 use crate::FlownodeId;
@@ -42,7 +42,7 @@ lazy_static! {
 /// The layout: `__flow/info/{flow_id}`.
 pub struct FlowInfoKey(FlowScoped<FlowInfoKeyInner>);
 
-impl<'a> MetaKey<'a, FlowInfoKey> for FlowInfoKey {
+impl<'a> MetadataKey<'a, FlowInfoKey> for FlowInfoKey {
     fn to_bytes(&self) -> Vec<u8> {
         self.0.to_bytes()
     }
@@ -80,14 +80,14 @@ impl FlowInfoKeyInner {
     }
 }
 
-impl<'a> MetaKey<'a, FlowInfoKeyInner> for FlowInfoKeyInner {
+impl<'a> MetadataKey<'a, FlowInfoKeyInner> for FlowInfoKeyInner {
     fn to_bytes(&self) -> Vec<u8> {
         format!("{FLOW_INFO_KEY_PREFIX}/{}", self.flow_id).into_bytes()
     }
 
     fn from_bytes(bytes: &'a [u8]) -> Result<FlowInfoKeyInner> {
         let key = std::str::from_utf8(bytes).map_err(|e| {
-            error::InvalidTableMetadataSnafu {
+            error::InvalidMetadataSnafu {
                 err_msg: format!(
                     "FlowInfoKeyInner '{}' is not a valid UTF8 string: {e}",
                     String::from_utf8_lossy(bytes)
@@ -98,7 +98,7 @@ impl<'a> MetaKey<'a, FlowInfoKeyInner> for FlowInfoKeyInner {
         let captures =
             FLOW_INFO_KEY_PATTERN
                 .captures(key)
-                .context(error::InvalidTableMetadataSnafu {
+                .context(error::InvalidMetadataSnafu {
                     err_msg: format!("Invalid FlowInfoKeyInner '{key}'"),
                 })?;
         // Safety: pass the regex check above
diff --git a/src/common/meta/src/key/flow/flow_name.rs b/src/common/meta/src/key/flow/flow_name.rs
index e7e763afa548..201c354f9982 100644
--- a/src/common/meta/src/key/flow/flow_name.rs
+++ b/src/common/meta/src/key/flow/flow_name.rs
@@ -24,7 +24,7 @@ use crate::error::{self, Result};
 use crate::key::flow::FlowScoped;
 use crate::key::txn_helper::TxnOpGetResponseSet;
 use crate::key::{
-    BytesAdapter, DeserializedValueWithBytes, FlowId, MetaKey, TableMetaValue, NAME_PATTERN,
+    BytesAdapter, DeserializedValueWithBytes, FlowId, MetadataKey, MetadataValue, NAME_PATTERN,
 };
 use crate::kv_backend::txn::Txn;
 use crate::kv_backend::KvBackendRef;
@@ -76,7 +76,7 @@ impl<'a> FlowNameKey<'a> {
     }
 }
 
-impl<'a> MetaKey<'a, FlowNameKey<'a>> for FlowNameKey<'a> {
+impl<'a> MetadataKey<'a, FlowNameKey<'a>> for FlowNameKey<'a> {
     fn to_bytes(&self) -> Vec<u8> {
         self.0.to_bytes()
     }
@@ -95,7 +95,7 @@ pub struct FlowNameKeyInner<'a> {
     pub flow_name: &'a str,
 }
 
-impl<'a> MetaKey<'a, FlowNameKeyInner<'a>> for FlowNameKeyInner<'_> {
+impl<'a> MetadataKey<'a, FlowNameKeyInner<'a>> for FlowNameKeyInner<'_> {
     fn to_bytes(&self) -> Vec<u8> {
         format!(
             "{FLOW_NAME_KEY_PREFIX}/{}/{}",
@@ -106,7 +106,7 @@ impl<'a> MetaKey<'a, FlowNameKeyInner<'a>> for FlowNameKeyInner<'_> {
 
     fn from_bytes(bytes: &'a [u8]) -> Result<FlowNameKeyInner> {
         let key = std::str::from_utf8(bytes).map_err(|e| {
-            error::InvalidTableMetadataSnafu {
+            error::InvalidMetadataSnafu {
                 err_msg: format!(
                     "FlowNameKeyInner '{}' is not a valid UTF8 string: {e}",
                     String::from_utf8_lossy(bytes)
@@ -117,7 +117,7 @@ impl<'a> MetaKey<'a, FlowNameKeyInner<'a>> for FlowNameKeyInner<'_> {
         let captures =
             FLOW_NAME_KEY_PATTERN
                 .captures(key)
-                .context(error::InvalidTableMetadataSnafu {
+                .context(error::InvalidMetadataSnafu {
                     err_msg: format!("Invalid FlowNameKeyInner '{key}'"),
                 })?;
         // Safety: pass the regex check above
diff --git a/src/common/meta/src/key/flow/flow_route.rs b/src/common/meta/src/key/flow/flow_route.rs
index 7de452182b50..96db01aef0a1 100644
--- a/src/common/meta/src/key/flow/flow_route.rs
+++ b/src/common/meta/src/key/flow/flow_route.rs
@@ -22,7 +22,7 @@ use snafu::OptionExt;
 
 use crate::error::{self, Result};
 use crate::key::flow::FlowScoped;
-use crate::key::{BytesAdapter, FlowId, FlowPartitionId, MetaKey, TableMetaValue};
+use crate::key::{BytesAdapter, FlowId, FlowPartitionId, MetadataKey, MetadataValue};
 use crate::kv_backend::txn::{Txn, TxnOp};
 use crate::kv_backend::KvBackendRef;
 use crate::peer::Peer;
@@ -68,7 +68,7 @@ impl FlowRouteKey {
     }
 }
 
-impl<'a> MetaKey<'a, FlowRouteKey> for FlowRouteKey {
+impl<'a> MetadataKey<'a, FlowRouteKey> for FlowRouteKey {
     fn to_bytes(&self) -> Vec<u8> {
         self.0.to_bytes()
     }
@@ -101,7 +101,7 @@ impl FlowRouteKeyInner {
     }
 }
 
-impl<'a> MetaKey<'a, FlowRouteKeyInner> for FlowRouteKeyInner {
+impl<'a> MetadataKey<'a, FlowRouteKeyInner> for FlowRouteKeyInner {
     fn to_bytes(&self) -> Vec<u8> {
         format!(
             "{FLOW_ROUTE_KEY_PREFIX}/{}/{}",
@@ -112,7 +112,7 @@ impl<'a> MetaKey<'a, FlowRouteKeyInner> for FlowRouteKeyInner {
 
     fn from_bytes(bytes: &'a [u8]) -> Result<FlowRouteKeyInner> {
         let key = std::str::from_utf8(bytes).map_err(|e| {
-            error::InvalidTableMetadataSnafu {
+            error::InvalidMetadataSnafu {
                 err_msg: format!(
                     "FlowInfoKeyInner '{}' is not a valid UTF8 string: {e}",
                     String::from_utf8_lossy(bytes)
@@ -123,7 +123,7 @@ impl<'a> MetaKey<'a, FlowRouteKeyInner> for FlowRouteKeyInner {
         let captures =
             FLOW_ROUTE_KEY_PATTERN
                 .captures(key)
-                .context(error::InvalidTableMetadataSnafu {
+                .context(error::InvalidMetadataSnafu {
                     err_msg: format!("Invalid FlowInfoKeyInner '{key}'"),
                 })?;
         // Safety: pass the regex check above
@@ -209,7 +209,7 @@ impl FlowRouteManager {
 #[cfg(test)]
 mod tests {
     use super::FlowRouteKey;
-    use crate::key::MetaKey;
+    use crate::key::MetadataKey;
 
     #[test]
     fn test_key_serialization() {
diff --git a/src/common/meta/src/key/flow/flownode_flow.rs b/src/common/meta/src/key/flow/flownode_flow.rs
index 8781cdaab275..4b4c31a7b08e 100644
--- a/src/common/meta/src/key/flow/flownode_flow.rs
+++ b/src/common/meta/src/key/flow/flownode_flow.rs
@@ -22,7 +22,7 @@ use snafu::OptionExt;
 
 use crate::error::{self, Result};
 use crate::key::flow::FlowScoped;
-use crate::key::{BytesAdapter, FlowId, FlowPartitionId, MetaKey};
+use crate::key::{BytesAdapter, FlowId, FlowPartitionId, MetadataKey};
 use crate::kv_backend::txn::{Txn, TxnOp};
 use crate::kv_backend::KvBackendRef;
 use crate::range_stream::{PaginationStream, DEFAULT_PAGE_SIZE};
@@ -44,7 +44,7 @@ const FLOWNODE_FLOW_KEY_PREFIX: &str = "flownode";
 /// The layout `__flow/flownode/{flownode_id}/{flow_id}/{partition_id}`
 pub struct FlownodeFlowKey(FlowScoped<FlownodeFlowKeyInner>);
 
-impl<'a> MetaKey<'a, FlownodeFlowKey> for FlownodeFlowKey {
+impl<'a> MetadataKey<'a, FlownodeFlowKey> for FlownodeFlowKey {
     fn to_bytes(&self) -> Vec<u8> {
         self.0.to_bytes()
     }
@@ -113,7 +113,7 @@ impl FlownodeFlowKeyInner {
     }
 }
 
-impl<'a> MetaKey<'a, FlownodeFlowKeyInner> for FlownodeFlowKeyInner {
+impl<'a> MetadataKey<'a, FlownodeFlowKeyInner> for FlownodeFlowKeyInner {
     fn to_bytes(&self) -> Vec<u8> {
         format!(
             "{FLOWNODE_FLOW_KEY_PREFIX}/{}/{}/{}",
@@ -124,7 +124,7 @@ impl<'a> MetaKey<'a, FlownodeFlowKeyInner> for FlownodeFlowKeyInner {
 
     fn from_bytes(bytes: &'a [u8]) -> Result<FlownodeFlowKeyInner> {
         let key = std::str::from_utf8(bytes).map_err(|e| {
-            error::InvalidTableMetadataSnafu {
+            error::InvalidMetadataSnafu {
                 err_msg: format!(
                     "FlownodeFlowKeyInner '{}' is not a valid UTF8 string: {e}",
                     String::from_utf8_lossy(bytes)
@@ -135,7 +135,7 @@ impl<'a> MetaKey<'a, FlownodeFlowKeyInner> for FlownodeFlowKeyInner {
         let captures =
             FLOWNODE_FLOW_KEY_PATTERN
                 .captures(key)
-                .context(error::InvalidTableMetadataSnafu {
+                .context(error::InvalidMetadataSnafu {
                     err_msg: format!("Invalid FlownodeFlowKeyInner '{key}'"),
                 })?;
         // Safety: pass the regex check above
@@ -208,7 +208,7 @@ impl FlownodeFlowManager {
 #[cfg(test)]
 mod tests {
     use crate::key::flow::flownode_flow::FlownodeFlowKey;
-    use crate::key::MetaKey;
+    use crate::key::MetadataKey;
 
     #[test]
     fn test_key_serialization() {
diff --git a/src/common/meta/src/key/flow/table_flow.rs b/src/common/meta/src/key/flow/table_flow.rs
index bc0646262517..5297734a2351 100644
--- a/src/common/meta/src/key/flow/table_flow.rs
+++ b/src/common/meta/src/key/flow/table_flow.rs
@@ -23,7 +23,7 @@ use table::metadata::TableId;
 
 use crate::error::{self, Result};
 use crate::key::flow::FlowScoped;
-use crate::key::{BytesAdapter, FlowId, FlowPartitionId, MetaKey, TableMetaValue};
+use crate::key::{BytesAdapter, FlowId, FlowPartitionId, MetadataKey, MetadataValue};
 use crate::kv_backend::txn::{Txn, TxnOp};
 use crate::kv_backend::KvBackendRef;
 use crate::peer::Peer;
@@ -56,7 +56,7 @@ struct TableFlowKeyInner {
 #[derive(Debug, PartialEq)]
 pub struct TableFlowKey(FlowScoped<TableFlowKeyInner>);
 
-impl<'a> MetaKey<'a, TableFlowKey> for TableFlowKey {
+impl<'a> MetadataKey<'a, TableFlowKey> for TableFlowKey {
     fn to_bytes(&self) -> Vec<u8> {
         self.0.to_bytes()
     }
@@ -129,7 +129,7 @@ impl TableFlowKeyInner {
     }
 }
 
-impl<'a> MetaKey<'a, TableFlowKeyInner> for TableFlowKeyInner {
+impl<'a> MetadataKey<'a, TableFlowKeyInner> for TableFlowKeyInner {
     fn to_bytes(&self) -> Vec<u8> {
         format!(
             "{TABLE_FLOW_KEY_PREFIX}/{}/{}/{}/{}",
@@ -140,7 +140,7 @@ impl<'a> MetaKey<'a, TableFlowKeyInner> for TableFlowKeyInner {
 
     fn from_bytes(bytes: &'a [u8]) -> Result<TableFlowKeyInner> {
         let key = std::str::from_utf8(bytes).map_err(|e| {
-            error::InvalidTableMetadataSnafu {
+            error::InvalidMetadataSnafu {
                 err_msg: format!(
                     "TableFlowKeyInner '{}' is not a valid UTF8 string: {e}",
                     String::from_utf8_lossy(bytes)
@@ -151,7 +151,7 @@ impl<'a> MetaKey<'a, TableFlowKeyInner> for TableFlowKeyInner {
         let captures =
             TABLE_FLOW_KEY_PATTERN
                 .captures(key)
-                .context(error::InvalidTableMetadataSnafu {
+                .context(error::InvalidMetadataSnafu {
                     err_msg: format!("Invalid TableFlowKeyInner '{key}'"),
                 })?;
         // Safety: pass the regex check above
diff --git a/src/common/meta/src/key/schema_name.rs b/src/common/meta/src/key/schema_name.rs
index 819371b910a6..7c9822e18db2 100644
--- a/src/common/meta/src/key/schema_name.rs
+++ b/src/common/meta/src/key/schema_name.rs
@@ -23,8 +23,8 @@ use humantime_serde::re::humantime;
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt};
 
-use crate::error::{self, Error, InvalidTableMetadataSnafu, ParseOptionSnafu, Result};
-use crate::key::{MetaKey, SCHEMA_NAME_KEY_PATTERN, SCHEMA_NAME_KEY_PREFIX};
+use crate::error::{self, Error, InvalidMetadataSnafu, ParseOptionSnafu, Result};
+use crate::key::{MetadataKey, SCHEMA_NAME_KEY_PATTERN, SCHEMA_NAME_KEY_PREFIX};
 use crate::kv_backend::KvBackendRef;
 use crate::range_stream::{PaginationStream, DEFAULT_PAGE_SIZE};
 use crate::rpc::store::RangeRequest;
@@ -122,14 +122,14 @@ impl Display for SchemaNameKey<'_> {
     }
 }
 
-impl<'a> MetaKey<'a, SchemaNameKey<'a>> for SchemaNameKey<'_> {
+impl<'a> MetadataKey<'a, SchemaNameKey<'a>> for SchemaNameKey<'_> {
     fn to_bytes(&self) -> Vec<u8> {
         self.to_string().into_bytes()
     }
 
     fn from_bytes(bytes: &'a [u8]) -> Result<SchemaNameKey<'a>> {
         let key = std::str::from_utf8(bytes).map_err(|e| {
-            InvalidTableMetadataSnafu {
+            InvalidMetadataSnafu {
                 err_msg: format!(
                     "SchemaNameKey '{}' is not a valid UTF8 string: {e}",
                     String::from_utf8_lossy(bytes)
@@ -155,7 +155,7 @@ impl<'a> TryFrom<&'a str> for SchemaNameKey<'a> {
     fn try_from(s: &'a str) -> Result<Self> {
         let captures = SCHEMA_NAME_KEY_PATTERN
             .captures(s)
-            .context(InvalidTableMetadataSnafu {
+            .context(InvalidMetadataSnafu {
                 err_msg: format!("Illegal SchemaNameKey format: '{s}'"),
             })?;
 
diff --git a/src/common/meta/src/key/table_info.rs b/src/common/meta/src/key/table_info.rs
index a652b7caf0fd..e5d124fcb85a 100644
--- a/src/common/meta/src/key/table_info.rs
+++ b/src/common/meta/src/key/table_info.rs
@@ -23,9 +23,9 @@ use table::table_name::TableName;
 use table::table_reference::TableReference;
 
 use super::TABLE_INFO_KEY_PATTERN;
-use crate::error::{InvalidTableMetadataSnafu, Result};
+use crate::error::{InvalidMetadataSnafu, Result};
 use crate::key::txn_helper::TxnOpGetResponseSet;
-use crate::key::{DeserializedValueWithBytes, MetaKey, TableMetaValue, TABLE_INFO_KEY_PREFIX};
+use crate::key::{DeserializedValueWithBytes, MetadataKey, MetadataValue, TABLE_INFO_KEY_PREFIX};
 use crate::kv_backend::txn::Txn;
 use crate::kv_backend::KvBackendRef;
 use crate::rpc::store::BatchGetRequest;
@@ -51,14 +51,14 @@ impl Display for TableInfoKey {
     }
 }
 
-impl<'a> MetaKey<'a, TableInfoKey> for TableInfoKey {
+impl<'a> MetadataKey<'a, TableInfoKey> for TableInfoKey {
     fn to_bytes(&self) -> Vec<u8> {
         self.to_string().into_bytes()
     }
 
     fn from_bytes(bytes: &[u8]) -> Result<TableInfoKey> {
         let key = std::str::from_utf8(bytes).map_err(|e| {
-            InvalidTableMetadataSnafu {
+            InvalidMetadataSnafu {
                 err_msg: format!(
                     "TableInfoKey '{}' is not a valid UTF8 string: {e}",
                     String::from_utf8_lossy(bytes)
@@ -68,7 +68,7 @@ impl<'a> MetaKey<'a, TableInfoKey> for TableInfoKey {
         })?;
         let captures = TABLE_INFO_KEY_PATTERN
             .captures(key)
-            .context(InvalidTableMetadataSnafu {
+            .context(InvalidMetadataSnafu {
                 err_msg: format!("Invalid TableInfoKey '{key}'"),
             })?;
         // Safety: pass the regex check above
diff --git a/src/common/meta/src/key/table_name.rs b/src/common/meta/src/key/table_name.rs
index 8a44de7cc695..7054ffc4e9aa 100644
--- a/src/common/meta/src/key/table_name.rs
+++ b/src/common/meta/src/key/table_name.rs
@@ -22,8 +22,8 @@ use snafu::OptionExt;
 use table::metadata::TableId;
 use table::table_name::TableName;
 
-use super::{MetaKey, TableMetaValue, TABLE_NAME_KEY_PATTERN, TABLE_NAME_KEY_PREFIX};
-use crate::error::{Error, InvalidTableMetadataSnafu, Result};
+use super::{MetadataKey, MetadataValue, TABLE_NAME_KEY_PATTERN, TABLE_NAME_KEY_PREFIX};
+use crate::error::{Error, InvalidMetadataSnafu, Result};
 use crate::kv_backend::memory::MemoryKvBackend;
 use crate::kv_backend::txn::{Txn, TxnOp};
 use crate::kv_backend::KvBackendRef;
@@ -63,14 +63,14 @@ impl Display for TableNameKey<'_> {
     }
 }
 
-impl<'a> MetaKey<'a, TableNameKey<'a>> for TableNameKey<'_> {
+impl<'a> MetadataKey<'a, TableNameKey<'a>> for TableNameKey<'_> {
     fn to_bytes(&self) -> Vec<u8> {
         self.to_string().into_bytes()
     }
 
     fn from_bytes(bytes: &'a [u8]) -> Result<TableNameKey<'a>> {
         let key = std::str::from_utf8(bytes).map_err(|e| {
-            InvalidTableMetadataSnafu {
+            InvalidMetadataSnafu {
                 err_msg: format!(
                     "TableNameKey '{}' is not a valid UTF8 string: {e}",
                     String::from_utf8_lossy(bytes)
@@ -80,7 +80,7 @@ impl<'a> MetaKey<'a, TableNameKey<'a>> for TableNameKey<'_> {
         })?;
         let captures = TABLE_NAME_KEY_PATTERN
             .captures(key)
-            .context(InvalidTableMetadataSnafu {
+            .context(InvalidMetadataSnafu {
                 err_msg: format!("Invalid TableNameKey '{key}'"),
             })?;
         let catalog = captures.get(1).unwrap().as_str();
@@ -128,7 +128,7 @@ impl<'a> TryFrom<&'a str> for TableNameKey<'a> {
     fn try_from(s: &'a str) -> Result<Self> {
         let captures = TABLE_NAME_KEY_PATTERN
             .captures(s)
-            .context(InvalidTableMetadataSnafu {
+            .context(InvalidMetadataSnafu {
                 err_msg: format!("Illegal TableNameKey format: '{s}'"),
             })?;
         // Safety: pass the regex check above
diff --git a/src/common/meta/src/key/table_route.rs b/src/common/meta/src/key/table_route.rs
index 979d789ab1c0..baaaeb75b95c 100644
--- a/src/common/meta/src/key/table_route.rs
+++ b/src/common/meta/src/key/table_route.rs
@@ -22,12 +22,12 @@ use store_api::storage::{RegionId, RegionNumber};
 use table::metadata::TableId;
 
 use crate::error::{
-    self, InvalidTableMetadataSnafu, MetadataCorruptionSnafu, Result, SerdeJsonSnafu,
+    self, InvalidMetadataSnafu, MetadataCorruptionSnafu, Result, SerdeJsonSnafu,
     TableRouteNotFoundSnafu, UnexpectedLogicalRouteTableSnafu,
 };
 use crate::key::txn_helper::TxnOpGetResponseSet;
 use crate::key::{
-    DeserializedValueWithBytes, MetaKey, RegionDistribution, TableMetaValue,
+    DeserializedValueWithBytes, MetadataKey, MetadataValue, RegionDistribution,
     TABLE_ROUTE_KEY_PATTERN, TABLE_ROUTE_PREFIX,
 };
 use crate::kv_backend::txn::Txn;
@@ -199,7 +199,7 @@ impl TableRouteValue {
     }
 }
 
-impl TableMetaValue for TableRouteValue {
+impl MetadataValue for TableRouteValue {
     fn try_from_raw_value(raw_value: &[u8]) -> Result<Self> {
         let r = serde_json::from_slice::<TableRouteValue>(raw_value);
         match r {
@@ -244,14 +244,14 @@ impl LogicalTableRouteValue {
     }
 }
 
-impl<'a> MetaKey<'a, TableRouteKey> for TableRouteKey {
+impl<'a> MetadataKey<'a, TableRouteKey> for TableRouteKey {
     fn to_bytes(&self) -> Vec<u8> {
         self.to_string().into_bytes()
     }
 
     fn from_bytes(bytes: &[u8]) -> Result<TableRouteKey> {
         let key = std::str::from_utf8(bytes).map_err(|e| {
-            InvalidTableMetadataSnafu {
+            InvalidMetadataSnafu {
                 err_msg: format!(
                     "TableRouteKey '{}' is not a valid UTF8 string: {e}",
                     String::from_utf8_lossy(bytes)
@@ -259,12 +259,11 @@ impl<'a> MetaKey<'a, TableRouteKey> for TableRouteKey {
             }
             .build()
         })?;
-        let captures =
-            TABLE_ROUTE_KEY_PATTERN
-                .captures(key)
-                .context(InvalidTableMetadataSnafu {
-                    err_msg: format!("Invalid TableRouteKey '{key}'"),
-                })?;
+        let captures = TABLE_ROUTE_KEY_PATTERN
+            .captures(key)
+            .context(InvalidMetadataSnafu {
+                err_msg: format!("Invalid TableRouteKey '{key}'"),
+            })?;
         // Safety: pass the regex check above
         let table_id = captures[1].parse::<TableId>().unwrap();
         Ok(TableRouteKey { table_id })
diff --git a/src/common/meta/src/key/txn_helper.rs b/src/common/meta/src/key/txn_helper.rs
index 54cd13f457b3..e816eef131c1 100644
--- a/src/common/meta/src/key/txn_helper.rs
+++ b/src/common/meta/src/key/txn_helper.rs
@@ -16,7 +16,7 @@ use serde::de::DeserializeOwned;
 use serde::Serialize;
 
 use crate::error::Result;
-use crate::key::{DeserializedValueWithBytes, TableMetaValue};
+use crate::key::{DeserializedValueWithBytes, MetadataValue};
 use crate::kv_backend::txn::TxnOpResponse;
 use crate::rpc::KeyValue;
 
@@ -41,7 +41,7 @@ impl TxnOpGetResponseSet {
     ) -> impl FnMut(&mut TxnOpGetResponseSet) -> Result<Option<DeserializedValueWithBytes<T>>>
     where
         F: FnMut(&mut TxnOpGetResponseSet) -> Option<Vec<u8>>,
-        T: Serialize + DeserializeOwned + TableMetaValue,
+        T: Serialize + DeserializeOwned + MetadataValue,
     {
         move |set| {
             f(set)
diff --git a/src/common/meta/src/key/view_info.rs b/src/common/meta/src/key/view_info.rs
index f1c4f395f2e4..e3843d54e339 100644
--- a/src/common/meta/src/key/view_info.rs
+++ b/src/common/meta/src/key/view_info.rs
@@ -24,7 +24,7 @@ use table::table_name::TableName;
 use super::VIEW_INFO_KEY_PATTERN;
 use crate::error::{InvalidViewInfoSnafu, Result};
 use crate::key::txn_helper::TxnOpGetResponseSet;
-use crate::key::{DeserializedValueWithBytes, MetaKey, TableMetaValue, VIEW_INFO_KEY_PREFIX};
+use crate::key::{DeserializedValueWithBytes, MetadataKey, MetadataValue, VIEW_INFO_KEY_PREFIX};
 use crate::kv_backend::txn::Txn;
 use crate::kv_backend::KvBackendRef;
 use crate::rpc::store::BatchGetRequest;
@@ -53,7 +53,7 @@ impl Display for ViewInfoKey {
     }
 }
 
-impl<'a> MetaKey<'a, ViewInfoKey> for ViewInfoKey {
+impl<'a> MetadataKey<'a, ViewInfoKey> for ViewInfoKey {
     fn to_bytes(&self) -> Vec<u8> {
         self.to_string().into_bytes()
     }
diff --git a/src/frontend/src/heartbeat/handler/tests.rs b/src/frontend/src/heartbeat/handler/tests.rs
index 9bbe8d903ae2..1791f6e91fcd 100644
--- a/src/frontend/src/heartbeat/handler/tests.rs
+++ b/src/frontend/src/heartbeat/handler/tests.rs
@@ -24,7 +24,7 @@ use common_meta::heartbeat::mailbox::{HeartbeatMailbox, MessageMeta};
 use common_meta::instruction::{CacheIdent, Instruction};
 use common_meta::key::schema_name::{SchemaName, SchemaNameKey};
 use common_meta::key::table_info::TableInfoKey;
-use common_meta::key::MetaKey;
+use common_meta::key::MetadataKey;
 use partition::manager::TableRouteCacheInvalidator;
 use table::metadata::TableId;
 use tokio::sync::mpsc;