refactor: rename NEXT_TABLE_ROUTE_PREFIX to TABLE_ROUTE_PREFIX (#2348)

* refactor: rename NEXT_TABLE_ROUTE_PREFIX to TABLE_ROUTE_PREFIX

* chore: apply suggestions from CR

src/cmd/src/cli/upgrade.rs

@@ -25,7 +25,7 @@ use common_meta::key::schema_name::{SchemaNameKey, SchemaNameValue};
 use common_meta::key::table_info::{TableInfoKey, TableInfoValue};
 use common_meta::key::table_name::{TableNameKey, TableNameValue};
 use common_meta::key::table_region::{TableRegionKey, TableRegionValue};
-use common_meta::key::table_route::{NextTableRouteKey, TableRouteValue as NextTableRouteValue};
+use common_meta::key::table_route::{TableRouteKey, TableRouteValue as NextTableRouteValue};
 use common_meta::key::{RegionDistribution, TableMetaKey};
 use common_meta::range_stream::PaginationStream;
 use common_meta::rpc::router::TableRoute;
@@ -154,7 +154,7 @@ impl MigrateTableMetadata {
         let new_table_value = NextTableRouteValue::new(table_route.region_routes);
 
         let table_id = table_route.table.id as u32;
-        let new_key = NextTableRouteKey::new(table_id);
+        let new_key = TableRouteKey::new(table_id);
         info!("Creating '{new_key}'");
 
         if self.dryrun {

src/common/catalog/src/error.rs

@@ -24,6 +24,12 @@ pub enum Error {
     #[snafu(display("Invalid catalog info: {}", key))]
     InvalidCatalog { key: String, location: Location },
 
+    #[snafu(display("Invalid full table name: {}", table_name))]
+    InvalidFullTableName {
+        table_name: String,
+        location: Location,
+    },
+
     #[snafu(display("Failed to deserialize catalog entry value: {}", raw))]
     DeserializeCatalogEntryValue {
         raw: String,
@@ -43,7 +49,8 @@ impl ErrorExt for Error {
         match self {
             Error::InvalidCatalog { .. }
             | Error::DeserializeCatalogEntryValue { .. }
-            | Error::SerializeCatalogEntryValue { .. } => StatusCode::Unexpected,
+            | Error::SerializeCatalogEntryValue { .. }
+            | Error::InvalidFullTableName { .. } => StatusCode::Unexpected,
         }
     }
 

src/common/catalog/src/lib.rs

@@ -13,6 +13,9 @@
 // limitations under the License.
 
 use consts::DEFAULT_CATALOG_NAME;
+use snafu::ensure;
+
+use crate::error::Result;
 
 pub mod consts;
 pub mod error;
@@ -23,6 +26,17 @@ pub fn format_full_table_name(catalog: &str, schema: &str, table: &str) -> Strin
     format!("{catalog}.{schema}.{table}")
 }
 
+pub fn parse_full_table_name(table_name: &str) -> Result<(&str, &str, &str)> {
+    let result = table_name.split('.').collect::<Vec<_>>();
+
+    ensure!(
+        result.len() == 3,
+        error::InvalidFullTableNameSnafu { table_name }
+    );
+
+    Ok((result[0], result[1], result[2]))
+}
+
 /// Build db name from catalog and schema string
 pub fn build_db_string(catalog: &str, schema: &str) -> String {
     if catalog == DEFAULT_CATALOG_NAME {

src/common/meta/src/key.rs

@@ -71,8 +71,6 @@ use self::catalog_name::{CatalogManager, CatalogNameValue};
 use self::schema_name::{SchemaManager, SchemaNameValue};
 use self::table_route::{TableRouteManager, TableRouteValue};
 use crate::error::{self, Result, SerdeJsonSnafu};
-#[allow(deprecated)]
-pub use crate::key::table_route::{TableRouteKey, TABLE_ROUTE_PREFIX};
 use crate::kv_backend::txn::Txn;
 use crate::kv_backend::KvBackendRef;
 use crate::rpc::router::{region_distribution, RegionRoute};
@@ -88,6 +86,7 @@ const TABLE_NAME_KEY_PREFIX: &str = "__table_name";
 const TABLE_REGION_KEY_PREFIX: &str = "__table_region";
 const CATALOG_NAME_KEY_PREFIX: &str = "__catalog_name";
 const SCHEMA_NAME_KEY_PREFIX: &str = "__schema_name";
+const TABLE_ROUTE_PREFIX: &str = "__table_route";
 
 pub type RegionDistribution = BTreeMap<DatanodeId, Vec<RegionNumber>>;
 

src/common/meta/src/key/table_route.rs

@@ -14,26 +14,20 @@
 
 use std::fmt::Display;
 
-use api::v1::meta::TableName;
 use serde::{Deserialize, Serialize};
 use table::metadata::TableId;
 
 use crate::error::Result;
-use crate::key::{to_removed_key, RegionDistribution, TableMetaKey};
+use crate::key::{to_removed_key, RegionDistribution, TableMetaKey, TABLE_ROUTE_PREFIX};
 use crate::kv_backend::txn::{Compare, CompareOp, Txn, TxnOp, TxnOpResponse};
 use crate::kv_backend::KvBackendRef;
 use crate::rpc::router::{region_distribution, RegionRoute};
 
-pub const TABLE_ROUTE_PREFIX: &str = "__meta_table_route";
-
-pub const NEXT_TABLE_ROUTE_PREFIX: &str = "__table_route";
-
-// TODO(weny): Renames it to TableRouteKey.
-pub struct NextTableRouteKey {
+pub struct TableRouteKey {
     pub table_id: TableId,
 }
 
-impl NextTableRouteKey {
+impl TableRouteKey {
     pub fn new(table_id: TableId) -> Self {
         Self { table_id }
     }
@@ -61,15 +55,15 @@ impl TableRouteValue {
     }
 }
 
-impl TableMetaKey for NextTableRouteKey {
+impl TableMetaKey for TableRouteKey {
     fn as_raw_key(&self) -> Vec<u8> {
         self.to_string().into_bytes()
     }
 }
 
-impl Display for NextTableRouteKey {
+impl Display for TableRouteKey {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}/{}", NEXT_TABLE_ROUTE_PREFIX, self.table_id)
+        write!(f, "{}/{}", TABLE_ROUTE_PREFIX, self.table_id)
     }
 }
 
@@ -89,7 +83,7 @@ impl TableRouteManager {
         Txn,
         impl FnOnce(&Vec<TxnOpResponse>) -> Result<Option<TableRouteValue>>,
     ) {
-        let key = NextTableRouteKey::new(table_id);
+        let key = TableRouteKey::new(table_id);
         let raw_key = key.as_raw_key();
         let txn = Txn::new().and_then(vec![TxnOp::Get(raw_key.clone())]);
 
@@ -105,7 +99,7 @@ impl TableRouteManager {
         Txn,
         impl FnOnce(&Vec<TxnOpResponse>) -> Result<Option<TableRouteValue>>,
     )> {
-        let key = NextTableRouteKey::new(table_id);
+        let key = TableRouteKey::new(table_id);
         let raw_key = key.as_raw_key();
 
         let txn = Txn::new()
@@ -133,7 +127,7 @@ impl TableRouteManager {
         Txn,
         impl FnOnce(&Vec<TxnOpResponse>) -> Result<Option<TableRouteValue>>,
     )> {
-        let key = NextTableRouteKey::new(table_id);
+        let key = TableRouteKey::new(table_id);
         let raw_key = key.as_raw_key();
         let raw_value = current_table_route_value.try_as_raw_value()?;
         let new_raw_value: Vec<u8> = new_table_route_value.try_as_raw_value()?;
@@ -156,7 +150,7 @@ impl TableRouteManager {
         table_id: TableId,
         table_route_value: &TableRouteValue,
     ) -> Result<Txn> {
-        let key = NextTableRouteKey::new(table_id);
+        let key = TableRouteKey::new(table_id);
         let raw_key = key.as_raw_key();
         let raw_value = table_route_value.try_as_raw_value()?;
         let removed_key = to_removed_key(&String::from_utf8_lossy(&raw_key));
@@ -190,7 +184,7 @@ impl TableRouteManager {
     }
 
     pub async fn get(&self, table_id: TableId) -> Result<Option<TableRouteValue>> {
-        let key = NextTableRouteKey::new(table_id);
+        let key = TableRouteKey::new(table_id);
         self.kv_backend
             .get(&key.as_raw_key())
             .await?
@@ -200,7 +194,7 @@ impl TableRouteManager {
 
     #[cfg(test)]
     pub async fn get_removed(&self, table_id: TableId) -> Result<Option<TableRouteValue>> {
-        let key = NextTableRouteKey::new(table_id).to_string();
+        let key = TableRouteKey::new(table_id).to_string();
         let removed_key = to_removed_key(&key).into_bytes();
         self.kv_backend
             .get(&removed_key)
@@ -219,86 +213,3 @@ impl TableRouteManager {
             .transpose()
     }
 }
-
-#[deprecated(since = "0.4.0", note = "Please use the NextTableRouteKey instead")]
-#[derive(Copy, Clone)]
-pub struct TableRouteKey<'a> {
-    pub table_id: TableId,
-    pub catalog_name: &'a str,
-    pub schema_name: &'a str,
-    pub table_name: &'a str,
-}
-
-impl<'a> TableRouteKey<'a> {
-    pub fn with_table_name(table_id: TableId, t: &'a TableName) -> Self {
-        Self {
-            table_id,
-            catalog_name: &t.catalog_name,
-            schema_name: &t.schema_name,
-            table_name: &t.table_name,
-        }
-    }
-
-    pub fn prefix(&self) -> String {
-        format!(
-            "{}-{}-{}-{}",
-            TABLE_ROUTE_PREFIX, self.catalog_name, self.schema_name, self.table_name
-        )
-    }
-
-    pub fn removed_key(&self) -> String {
-        to_removed_key(&self.to_string())
-    }
-}
-
-impl<'a> Display for TableRouteKey<'a> {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}-{}", self.prefix(), self.table_id)
-    }
-}
-
-#[cfg(test)]
-mod tests {
-
-    use api::v1::meta::TableName as PbTableName;
-
-    use super::TableRouteKey;
-
-    #[test]
-    fn test_table_route_key() {
-        let key = TableRouteKey {
-            table_id: 123,
-            catalog_name: "greptime",
-            schema_name: "public",
-            table_name: "demo",
-        };
-
-        let prefix = key.prefix();
-        assert_eq!("__meta_table_route-greptime-public-demo", prefix);
-
-        let key_string = key.to_string();
-        assert_eq!("__meta_table_route-greptime-public-demo-123", key_string);
-
-        let removed = key.removed_key();
-        assert_eq!(
-            "__removed-__meta_table_route-greptime-public-demo-123",
-            removed
-        );
-    }
-
-    #[test]
-    fn test_with_table_name() {
-        let table_name = PbTableName {
-            catalog_name: "greptime".to_string(),
-            schema_name: "public".to_string(),
-            table_name: "demo".to_string(),
-        };
-
-        let key = TableRouteKey::with_table_name(123, &table_name);
-
-        assert_eq!(123, key.table_id);
-        assert_eq!("greptime", key.catalog_name);
-        assert_eq!("public", key.schema_name);
-        assert_eq!("demo", key.table_name);
-    }
-}

src/frontend/src/catalog.rs

@@ -38,7 +38,7 @@ use common_meta::key::catalog_name::CatalogNameKey;
 use common_meta::key::schema_name::SchemaNameKey;
 use common_meta::key::table_info::TableInfoKey;
 use common_meta::key::table_name::TableNameKey;
-use common_meta::key::table_route::NextTableRouteKey;
+use common_meta::key::table_route::TableRouteKey;
 use common_meta::key::{TableMetaKey, TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
 use common_telemetry::debug;
@@ -85,7 +85,7 @@ impl CacheInvalidator for FrontendCatalogManager {
             String::from_utf8_lossy(&key.as_raw_key())
         );
 
-        let key = &NextTableRouteKey { table_id };
+        let key = &TableRouteKey { table_id };
         self.backend_cache_invalidator
             .invalidate_key(&key.as_raw_key())
             .await;

src/frontend/src/heartbeat/handler/invalidate_table_cache.rs

@@ -22,7 +22,7 @@ use common_meta::ident::TableIdent;
 use common_meta::instruction::{Instruction, InstructionReply, SimpleReply};
 use common_meta::key::table_info::TableInfoKey;
 use common_meta::key::table_name::TableNameKey;
-use common_meta::key::table_route::NextTableRouteKey;
+use common_meta::key::table_route::TableRouteKey;
 use common_meta::key::TableMetaKey;
 use common_telemetry::error;
 
@@ -96,7 +96,7 @@ impl InvalidateTableCacheHandler {
             .await;
 
         self.backend_cache_invalidator
-            .invalidate_key(&NextTableRouteKey { table_id }.as_raw_key())
+            .invalidate_key(&TableRouteKey { table_id }.as_raw_key())
             .await;
     }
 }

src/meta-srv/src/error.rs

@@ -228,6 +228,12 @@ pub enum Error {
         source: common_catalog::error::Error,
     },
 
+    #[snafu(display("Cannot parse full table name, source: {}", source))]
+    InvalidFullTableName {
+        location: Location,
+        source: common_catalog::error::Error,
+    },
+
     #[snafu(display("Failed to decode table route, source: {}", source))]
     DecodeTableRoute {
         source: prost::DecodeError,
@@ -580,7 +586,8 @@ impl ErrorExt for Error {
             Error::TableNotFound { .. } => StatusCode::TableNotFound,
             Error::InvalidateTableCache { source, .. } => source.status_code(),
             Error::RequestDatanode { source, .. } => source.status_code(),
-            Error::InvalidCatalogValue { source, .. } => source.status_code(),
+            Error::InvalidCatalogValue { source, .. }
+            | Error::InvalidFullTableName { source, .. } => source.status_code(),
             Error::RecoverProcedure { source, .. }
             | Error::SubmitProcedure { source, .. }
             | Error::WaitProcedure { source, .. } => source.status_code(),

src/meta-srv/src/keys.rs

@@ -14,7 +14,6 @@
 
 use std::str::FromStr;
 
-use common_meta::key::TABLE_ROUTE_PREFIX;
 use lazy_static::lazy_static;
 use regex::Regex;
 use serde::{Deserialize, Serialize};
@@ -131,15 +130,6 @@ impl TryFrom<LeaseValue> for Vec<u8> {
     }
 }
 
-pub fn build_table_route_prefix(catalog: impl AsRef<str>, schema: impl AsRef<str>) -> String {
-    format!(
-        "{}-{}-{}-",
-        TABLE_ROUTE_PREFIX,
-        catalog.as_ref(),
-        schema.as_ref()
-    )
-}
-
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
 pub struct StatKey {
     pub cluster_id: u64,
@@ -312,14 +302,6 @@ impl TryFrom<Vec<u8>> for InactiveRegionKey {
 mod tests {
     use super::*;
 
-    #[test]
-    fn test_build_prefix() {
-        assert_eq!(
-            "__meta_table_route-CATALOG-SCHEMA-",
-            build_table_route_prefix("CATALOG", "SCHEMA")
-        )
-    }
-
     #[test]
     fn test_stat_key_round_trip() {
         let key = StatKey {

src/meta-srv/src/procedure/region_failover/update_metadata.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use async_trait::async_trait;
-use common_meta::key::table_route::NextTableRouteKey;
+use common_meta::key::table_route::TableRouteKey;
 use common_meta::peer::Peer;
 use common_meta::rpc::router::RegionRoute;
 use common_meta::RegionIdent;
@@ -79,7 +79,7 @@ impl UpdateRegionMetadata {
         }
 
         pretty_log_table_route_change(
-            NextTableRouteKey::new(table_id).to_string(),
+            TableRouteKey::new(table_id).to_string(),
             &new_region_routes,
             failed_region,
         );

src/meta-srv/src/service/admin.rs

@@ -87,7 +87,7 @@ pub fn make_admin_service(meta_srv: MetaSrv) -> Admin {
     let router = router.route(
         "/route",
         route::RouteHandler {
-            kv_store: meta_srv.kv_store().clone(),
+            table_metadata_manager: meta_srv.table_metadata_manager().clone(),
         },
     );
 

src/meta-srv/src/service/admin/meta.rs

@@ -14,6 +14,7 @@
 
 use std::collections::HashMap;
 
+use common_catalog::parse_full_table_name;
 use common_error::ext::BoxedError;
 use common_meta::key::table_name::TableNameKey;
 use common_meta::key::TableMetadataManagerRef;
@@ -135,10 +136,10 @@ impl HttpHandler for TableHandler {
                     param: "full_table_name",
                 })?;
 
-        let key: TableNameKey = table_name
-            .as_str()
-            .try_into()
-            .context(TableMetadataManagerSnafu)?;
+        let (catalog, schema, table) =
+            parse_full_table_name(table_name).context(error::InvalidFullTableNameSnafu)?;
+
+        let key = TableNameKey::new(catalog, schema, table);
 
         let mut result = HashMap::with_capacity(2);
 

src/meta-srv/src/service/admin/route.rs

@@ -14,21 +14,18 @@
 
 use std::collections::HashMap;
 
-use api::v1::meta::TableRouteValue;
-use common_meta::key::TABLE_ROUTE_PREFIX;
-use common_meta::rpc::store::{RangeRequest, RangeResponse};
-use common_meta::util;
-use prost::Message;
+use common_catalog::parse_full_table_name;
+use common_meta::key::table_name::TableNameKey;
+use common_meta::key::TableMetadataManagerRef;
 use snafu::{OptionExt, ResultExt};
 use tonic::codegen::http;
 
 use super::HttpHandler;
 use crate::error;
 use crate::error::Result;
-use crate::service::store::kv::KvStoreRef;
 
 pub struct RouteHandler {
-    pub kv_store: KvStoreRef,
+    pub table_metadata_manager: TableMetadataManagerRef,
 }
 
 #[async_trait::async_trait]
@@ -38,51 +35,36 @@ impl HttpHandler for RouteHandler {
         _path: &str,
         params: &HashMap<String, String>,
     ) -> Result<http::Response<String>> {
-        let full_table_name = params
-            .get("full_table_name")
-            .map(|full_table_name| full_table_name.replace('.', "-"))
-            .context(error::MissingRequiredParameterSnafu {
-                param: "full_table_name",
-            })?;
+        let table_name =
+            params
+                .get("full_table_name")
+                .context(error::MissingRequiredParameterSnafu {
+                    param: "full_table_name",
+                })?;
 
-        let route_key = format!("{}-{}", TABLE_ROUTE_PREFIX, full_table_name).into_bytes();
+        let (catalog, schema, table) =
+            parse_full_table_name(table_name).context(error::InvalidFullTableNameSnafu)?;
 
-        let range_end = util::get_prefix_end_key(&route_key);
+        let key = TableNameKey::new(catalog, schema, table);
 
-        let req = RangeRequest {
-            key: route_key,
-            range_end,
-            keys_only: false,
-            ..Default::default()
-        };
-
-        let resp = self.kv_store.range(req).await?;
-
-        let show = pretty_fmt(resp)?;
+        let table_id = self
+            .table_metadata_manager
+            .table_name_manager()
+            .get(key)
+            .await
+            .context(error::TableMetadataManagerSnafu)?
+            .map(|x| x.table_id());
 
+        let table_route_value = self
+            .table_metadata_manager
+            .table_route_manager()
+            .get(table_id.context(error::TableNotFoundSnafu { name: table_name })?)
+            .await
+            .context(error::TableMetadataManagerSnafu)?
+            .context(error::TableRouteNotFoundSnafu { table_name })?;
         http::Response::builder()
             .status(http::StatusCode::OK)
-            .body(show)
+            .body(serde_json::to_string(&table_route_value).unwrap())
             .context(error::InvalidHttpBodySnafu)
     }
 }
-
-fn pretty_fmt(response: RangeResponse) -> Result<String> {
-    let mut show = "".to_string();
-
-    for kv in response.kvs.into_iter() {
-        let route_key = String::from_utf8(kv.key).unwrap();
-        let route_val =
-            TableRouteValue::decode(&kv.value[..]).context(error::DecodeTableRouteSnafu)?;
-
-        show.push_str("route_key:\n");
-        show.push_str(&route_key);
-        show.push('\n');
-
-        show.push_str("route_value:\n");
-        show.push_str(&format!("{:#?}", route_val));
-        show.push('\n');
-    }
-
-    Ok(show)
-}