Revert "fix: revert unfinished route table change (#3008)"

This reverts commit 8ce8a8f.

src/cmd/src/cli/bench/metadata.rs

@@ -14,6 +14,7 @@
 
 use std::time::Instant;
 
+use common_meta::key::table_route::TableRouteValue;
 use common_meta::key::TableMetadataManagerRef;
 use common_meta::table_name::TableName;
 
@@ -53,7 +54,11 @@ impl TableMetadataBencher {
                 let start = Instant::now();
 
                 self.table_metadata_manager
-                    .create_table_metadata(table_info, region_routes, region_wal_options)
+                    .create_table_metadata(
+                        table_info,
+                        TableRouteValue::physical(region_routes),
+                        region_wal_options,
+                    )
                     .await
                     .unwrap();
 

src/cmd/src/cli/upgrade.rs

@@ -27,7 +27,7 @@ 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::{TableRouteKey, TableRouteValue as NextTableRouteValue};
-use common_meta::key::{RegionDistribution, TableMetaKey};
+use common_meta::key::{RegionDistribution, TableMetaKey, TableMetaValue};
 use common_meta::kv_backend::etcd::EtcdStore;
 use common_meta::kv_backend::KvBackendRef;
 use common_meta::range_stream::PaginationStream;
@@ -153,7 +153,7 @@ impl MigrateTableMetadata {
         )
         .unwrap();
 
-        let new_table_value = NextTableRouteValue::new(table_route.region_routes);
+        let new_table_value = NextTableRouteValue::physical(table_route.region_routes);
 
         let table_id = table_route.table.id as u32;
         let new_key = TableRouteKey::new(table_id);

src/common/meta/src/ddl.rs

@@ -21,10 +21,10 @@ use store_api::storage::{RegionNumber, TableId};
 use crate::cache_invalidator::CacheInvalidatorRef;
 use crate::datanode_manager::DatanodeManagerRef;
 use crate::error::Result;
+use crate::key::table_route::TableRouteValue;
 use crate::key::TableMetadataManagerRef;
 use crate::region_keeper::MemoryRegionKeeperRef;
 use crate::rpc::ddl::{CreateTableTask, SubmitDdlTaskRequest, SubmitDdlTaskResponse};
-use crate::rpc::router::RegionRoute;
 
 pub mod alter_table;
 pub mod create_table;
@@ -58,7 +58,7 @@ pub struct TableMetadata {
     /// Table id.
     pub table_id: TableId,
     /// Route information for each region of the table.
-    pub region_routes: Vec<RegionRoute>,
+    pub table_route: TableRouteValue,
     /// The encoded wal options for regions of the table.
     // If a region does not have an associated wal options, no key for the region would be found in the map.
     pub region_wal_options: HashMap<RegionNumber, String>,

src/common/meta/src/ddl/alter_table.rs

@@ -45,7 +45,6 @@ use crate::error::{
 };
 use crate::key::table_info::TableInfoValue;
 use crate::key::table_name::TableNameKey;
-use crate::key::table_route::TableRouteValue;
 use crate::key::DeserializedValueWithBytes;
 use crate::metrics;
 use crate::rpc::ddl::AlterTableTask;
@@ -183,25 +182,23 @@ impl AlterTableProcedure {
 
     pub async fn submit_alter_region_requests(&mut self) -> Result<Status> {
         let table_id = self.data.table_id();
-        let table_ref = self.data.table_ref();
 
-        let TableRouteValue { region_routes, .. } = self
+        let table_route = self
             .context
             .table_metadata_manager
             .table_route_manager()
             .get(table_id)
             .await?
-            .with_context(|| TableRouteNotFoundSnafu {
-                table_name: table_ref.to_string(),
-            })?
+            .context(TableRouteNotFoundSnafu { table_id })?
             .into_inner();
+        let region_routes = table_route.region_routes();
 
-        let leaders = find_leaders(&region_routes);
+        let leaders = find_leaders(region_routes);
         let mut alter_region_tasks = Vec::with_capacity(leaders.len());
 
         for datanode in leaders {
             let requester = self.context.datanode_manager.datanode(&datanode).await;
-            let regions = find_leader_regions(&region_routes, &datanode);
+            let regions = find_leader_regions(region_routes, &datanode);
 
             for region in regions {
                 let region_id = RegionId::new(table_id, region);