mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-28 14:46:22 +00:00
feat(TableRouteValue): add panic notes and type checks (#3031)
* refactor(TableRouteValue): add panic notes and type checks * chore: add deprecate develop branch warning Signed-off-by: Ruihang Xia <waynestxia@gmail.com> * add error defines and checks * Update README.md * update code format and fix tests * update name of error * delete unused note * fix unsafe .expect() for region_route() * update error name * update unwrap * update code format --------- Signed-off-by: Ruihang Xia <waynestxia@gmail.com> Co-authored-by: Ruihang Xia <waynestxia@gmail.com>
This commit is contained in:
co-authored by
Ruihang Xia
parent
69a53130c2
commit
4460af800f
@@ -191,7 +191,7 @@ impl AlterTableProcedure {
|
||||
.await?
|
||||
.context(TableRouteNotFoundSnafu { table_id })?
|
||||
.into_inner();
|
||||
let region_routes = table_route.region_routes();
|
||||
let region_routes = table_route.region_routes()?;
|
||||
|
||||
let leaders = find_leaders(region_routes);
|
||||
let mut alter_region_tasks = Vec::with_capacity(leaders.len());
|
||||
|
||||
@@ -217,7 +217,7 @@ impl CreateTableProcedure {
|
||||
.context(TableRouteNotFoundSnafu {
|
||||
table_id: physical_table_id,
|
||||
})?;
|
||||
let region_routes = physical_table_route.region_routes();
|
||||
let region_routes = physical_table_route.region_routes()?;
|
||||
|
||||
let request_builder = self.new_region_request_builder(Some(physical_table_id))?;
|
||||
|
||||
|
||||
@@ -116,7 +116,7 @@ impl DropTableProcedure {
|
||||
|
||||
/// Register dropping regions if doesn't exist.
|
||||
fn register_dropping_regions(&mut self) -> Result<()> {
|
||||
let region_routes = self.data.region_routes();
|
||||
let region_routes = self.data.region_routes()?;
|
||||
|
||||
let dropping_regions = operating_leader_regions(region_routes);
|
||||
|
||||
@@ -190,7 +190,7 @@ impl DropTableProcedure {
|
||||
pub async fn on_datanode_drop_regions(&self) -> Result<Status> {
|
||||
let table_id = self.data.table_id();
|
||||
|
||||
let region_routes = &self.data.region_routes();
|
||||
let region_routes = &self.data.region_routes()?;
|
||||
let leaders = find_leaders(region_routes);
|
||||
let mut drop_region_tasks = Vec::with_capacity(leaders.len());
|
||||
|
||||
@@ -306,7 +306,7 @@ impl DropTableData {
|
||||
self.task.table_ref()
|
||||
}
|
||||
|
||||
fn region_routes(&self) -> &Vec<RegionRoute> {
|
||||
fn region_routes(&self) -> Result<&Vec<RegionRoute>> {
|
||||
self.table_route_value.region_routes()
|
||||
}
|
||||
|
||||
|
||||
@@ -278,7 +278,7 @@ async fn handle_truncate_table_task(
|
||||
let table_route_value =
|
||||
table_route_value.context(error::TableRouteNotFoundSnafu { table_id })?;
|
||||
|
||||
let table_route = table_route_value.into_inner().region_routes().clone();
|
||||
let table_route = table_route_value.into_inner().region_routes()?.clone();
|
||||
|
||||
let id = ddl_manager
|
||||
.submit_truncate_table_task(
|
||||
|
||||
@@ -351,6 +351,9 @@ pub enum Error {
|
||||
|
||||
#[snafu(display("The topic pool is empty"))]
|
||||
EmptyTopicPool { location: Location },
|
||||
|
||||
#[snafu(display("Unexpected table route type: {}", err_msg))]
|
||||
UnexpectedLogicalRouteTable { location: Location, err_msg: String },
|
||||
}
|
||||
|
||||
pub type Result<T> = std::result::Result<T, Error>;
|
||||
@@ -392,7 +395,8 @@ impl ErrorExt for Error {
|
||||
| BuildKafkaPartitionClient { .. }
|
||||
| ProduceRecord { .. }
|
||||
| CreateKafkaWalTopic { .. }
|
||||
| EmptyTopicPool { .. } => StatusCode::Unexpected,
|
||||
| EmptyTopicPool { .. }
|
||||
| UnexpectedLogicalRouteTable { .. } => StatusCode::Unexpected,
|
||||
|
||||
SendMessage { .. }
|
||||
| GetKvCache { .. }
|
||||
|
||||
+25
-17
@@ -483,7 +483,7 @@ impl TableMetadataManager {
|
||||
.build_delete_txn(table_id, table_info_value)?;
|
||||
|
||||
// Deletes datanode table key value pairs.
|
||||
let distribution = region_distribution(table_route_value.region_routes())?;
|
||||
let distribution = region_distribution(table_route_value.region_routes()?)?;
|
||||
let delete_datanode_txn = self
|
||||
.datanode_table_manager()
|
||||
.build_delete_txn(table_id, distribution)?;
|
||||
@@ -608,7 +608,7 @@ impl TableMetadataManager {
|
||||
) -> Result<()> {
|
||||
// Updates the datanode table key value pairs.
|
||||
let current_region_distribution =
|
||||
region_distribution(current_table_route_value.region_routes())?;
|
||||
region_distribution(current_table_route_value.region_routes()?)?;
|
||||
let new_region_distribution = region_distribution(&new_region_routes)?;
|
||||
|
||||
let update_datanode_table_txn = self.datanode_table_manager().build_update_txn(
|
||||
@@ -621,7 +621,7 @@ impl TableMetadataManager {
|
||||
)?;
|
||||
|
||||
// Updates the table_route.
|
||||
let new_table_route_value = current_table_route_value.update(new_region_routes);
|
||||
let new_table_route_value = current_table_route_value.update(new_region_routes)?;
|
||||
|
||||
let (update_table_route_txn, on_update_table_route_failure) = self
|
||||
.table_route_manager()
|
||||
@@ -656,7 +656,7 @@ impl TableMetadataManager {
|
||||
where
|
||||
F: Fn(&RegionRoute) -> Option<Option<RegionStatus>>,
|
||||
{
|
||||
let mut new_region_routes = current_table_route_value.region_routes().clone();
|
||||
let mut new_region_routes = current_table_route_value.region_routes()?.clone();
|
||||
|
||||
let mut updated = 0;
|
||||
for route in &mut new_region_routes {
|
||||
@@ -673,7 +673,7 @@ impl TableMetadataManager {
|
||||
}
|
||||
|
||||
// Updates the table_route.
|
||||
let new_table_route_value = current_table_route_value.update(new_region_routes);
|
||||
let new_table_route_value = current_table_route_value.update(new_region_routes)?;
|
||||
|
||||
let (update_table_route_txn, on_update_table_route_failure) = self
|
||||
.table_route_manager()
|
||||
@@ -897,7 +897,11 @@ mod tests {
|
||||
table_info
|
||||
);
|
||||
assert_eq!(
|
||||
remote_table_route.unwrap().into_inner().region_routes(),
|
||||
remote_table_route
|
||||
.unwrap()
|
||||
.into_inner()
|
||||
.region_routes()
|
||||
.unwrap(),
|
||||
region_routes
|
||||
);
|
||||
}
|
||||
@@ -978,7 +982,7 @@ mod tests {
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.into_inner();
|
||||
assert_eq!(removed_table_route.region_routes(), region_routes);
|
||||
assert_eq!(removed_table_route.region_routes().unwrap(), region_routes);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1173,11 +1177,11 @@ mod tests {
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
updated_route_value.region_routes()[0].leader_status,
|
||||
updated_route_value.region_routes().unwrap()[0].leader_status,
|
||||
Some(RegionStatus::Downgraded)
|
||||
);
|
||||
assert_eq!(
|
||||
updated_route_value.region_routes()[1].leader_status,
|
||||
updated_route_value.region_routes().unwrap()[1].leader_status,
|
||||
Some(RegionStatus::Downgraded)
|
||||
);
|
||||
}
|
||||
@@ -1271,7 +1275,8 @@ mod tests {
|
||||
let current_table_route_value = DeserializedValueWithBytes::from_inner(
|
||||
current_table_route_value
|
||||
.inner
|
||||
.update(new_region_routes.clone()),
|
||||
.update(new_region_routes.clone())
|
||||
.unwrap(),
|
||||
);
|
||||
let new_region_routes = vec![new_region_route(2, 4), new_region_route(5, 5)];
|
||||
// it should be ok.
|
||||
@@ -1295,13 +1300,16 @@ mod tests {
|
||||
|
||||
// if the current_table_route_value is wrong, it should return an error.
|
||||
// The ABA problem.
|
||||
let wrong_table_route_value =
|
||||
DeserializedValueWithBytes::from_inner(current_table_route_value.update(vec![
|
||||
new_region_route(1, 1),
|
||||
new_region_route(2, 2),
|
||||
new_region_route(3, 3),
|
||||
new_region_route(4, 4),
|
||||
]));
|
||||
let wrong_table_route_value = DeserializedValueWithBytes::from_inner(
|
||||
current_table_route_value
|
||||
.update(vec![
|
||||
new_region_route(1, 1),
|
||||
new_region_route(2, 2),
|
||||
new_region_route(3, 3),
|
||||
new_region_route(4, 4),
|
||||
])
|
||||
.unwrap(),
|
||||
);
|
||||
assert!(table_metadata_manager
|
||||
.update_table_route(
|
||||
table_id,
|
||||
|
||||
@@ -16,12 +16,12 @@ use std::collections::HashMap;
|
||||
use std::fmt::Display;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use snafu::ResultExt;
|
||||
use snafu::{ensure, ResultExt};
|
||||
use store_api::storage::{RegionId, RegionNumber};
|
||||
use table::metadata::TableId;
|
||||
|
||||
use super::{DeserializedValueWithBytes, TableMetaValue};
|
||||
use crate::error::{Result, SerdeJsonSnafu};
|
||||
use crate::error::{Result, SerdeJsonSnafu, UnexpectedLogicalRouteTableSnafu};
|
||||
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;
|
||||
@@ -62,29 +62,48 @@ impl TableRouteValue {
|
||||
}
|
||||
|
||||
/// Returns a new version [TableRouteValue] with `region_routes`.
|
||||
pub fn update(&self, region_routes: Vec<RegionRoute>) -> Self {
|
||||
pub fn update(&self, region_routes: Vec<RegionRoute>) -> Result<Self> {
|
||||
ensure!(
|
||||
self.is_physical(),
|
||||
UnexpectedLogicalRouteTableSnafu {
|
||||
err_msg: "{self:?} is a non-physical TableRouteValue.",
|
||||
}
|
||||
);
|
||||
let version = self.physical_table_route().version;
|
||||
Self::Physical(PhysicalTableRouteValue {
|
||||
Ok(Self::Physical(PhysicalTableRouteValue {
|
||||
region_routes,
|
||||
version: version + 1,
|
||||
})
|
||||
}))
|
||||
}
|
||||
|
||||
/// Returns the version.
|
||||
///
|
||||
/// For test purpose.
|
||||
#[cfg(any(test, feature = "testing"))]
|
||||
pub fn version(&self) -> u64 {
|
||||
self.physical_table_route().version
|
||||
pub fn version(&self) -> Result<u64> {
|
||||
ensure!(
|
||||
self.is_physical(),
|
||||
UnexpectedLogicalRouteTableSnafu {
|
||||
err_msg: "{self:?} is a non-physical TableRouteValue.",
|
||||
}
|
||||
);
|
||||
Ok(self.physical_table_route().version)
|
||||
}
|
||||
|
||||
/// Returns the corresponding [RegionRoute].
|
||||
pub fn region_route(&self, region_id: RegionId) -> Option<RegionRoute> {
|
||||
self.physical_table_route()
|
||||
pub fn region_route(&self, region_id: RegionId) -> Result<Option<RegionRoute>> {
|
||||
ensure!(
|
||||
self.is_physical(),
|
||||
UnexpectedLogicalRouteTableSnafu {
|
||||
err_msg: "{self:?} is a non-physical TableRouteValue.",
|
||||
}
|
||||
);
|
||||
Ok(self
|
||||
.physical_table_route()
|
||||
.region_routes
|
||||
.iter()
|
||||
.find(|route| route.region.id == region_id)
|
||||
.cloned()
|
||||
.cloned())
|
||||
}
|
||||
|
||||
/// Returns true if it's [TableRouteValue::Physical].
|
||||
@@ -93,11 +112,14 @@ impl TableRouteValue {
|
||||
}
|
||||
|
||||
/// Gets the [RegionRoute]s of this [TableRouteValue::Physical].
|
||||
///
|
||||
/// # Panics
|
||||
/// The route type is not the [TableRouteValue::Physical].
|
||||
pub fn region_routes(&self) -> &Vec<RegionRoute> {
|
||||
&self.physical_table_route().region_routes
|
||||
pub fn region_routes(&self) -> Result<&Vec<RegionRoute>> {
|
||||
ensure!(
|
||||
self.is_physical(),
|
||||
UnexpectedLogicalRouteTableSnafu {
|
||||
err_msg: "{self:?} is a non-physical TableRouteValue.",
|
||||
}
|
||||
);
|
||||
Ok(&self.physical_table_route().region_routes)
|
||||
}
|
||||
|
||||
fn physical_table_route(&self) -> &PhysicalTableRouteValue {
|
||||
@@ -354,7 +376,7 @@ impl TableRouteManager {
|
||||
) -> Result<Option<RegionDistribution>> {
|
||||
self.get(table_id)
|
||||
.await?
|
||||
.map(|table_route| region_distribution(table_route.region_routes()))
|
||||
.map(|table_route| region_distribution(table_route.region_routes()?))
|
||||
.transpose()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -602,6 +602,13 @@ pub enum Error {
|
||||
|
||||
#[snafu(display("Weight array is not set"))]
|
||||
NotSetWeightArray { location: Location },
|
||||
|
||||
#[snafu(display("Unexpected table route type: {}", err_msg))]
|
||||
UnexpectedLogicalRouteTable {
|
||||
location: Location,
|
||||
err_msg: String,
|
||||
source: common_meta::error::Error,
|
||||
},
|
||||
}
|
||||
|
||||
impl Error {
|
||||
@@ -717,7 +724,8 @@ impl ErrorExt for Error {
|
||||
| Error::TableMetadataManager { source, .. }
|
||||
| Error::KvBackend { source, .. }
|
||||
| Error::UpdateTableRoute { source, .. }
|
||||
| Error::GetFullTableInfo { source, .. } => source.status_code(),
|
||||
| Error::GetFullTableInfo { source, .. }
|
||||
| Error::UnexpectedLogicalRouteTable { source, .. } => source.status_code(),
|
||||
|
||||
Error::InitMetadata { source, .. } | Error::InitDdlManager { source, .. } => {
|
||||
source.status_code()
|
||||
|
||||
@@ -208,6 +208,7 @@ mod tests {
|
||||
|
||||
let should_downgraded = table_route_value
|
||||
.region_routes()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.find(|route| route.region.id.region_number() == failed_region.region_number)
|
||||
.unwrap();
|
||||
|
||||
@@ -85,7 +85,12 @@ impl UpdateRegionMetadata {
|
||||
.context(error::TableMetadataManagerSnafu)?
|
||||
.context(TableRouteNotFoundSnafu { table_id })?;
|
||||
|
||||
let mut new_region_routes = table_route_value.region_routes().clone();
|
||||
let mut new_region_routes = table_route_value
|
||||
.region_routes()
|
||||
.context(error::UnexpectedLogicalRouteTableSnafu {
|
||||
err_msg: "{self:?} is a non-physical TableRouteValue.",
|
||||
})?
|
||||
.clone();
|
||||
|
||||
for region_route in new_region_routes.iter_mut() {
|
||||
if region_route.region.id.region_number() == failed_region.region_number {
|
||||
@@ -234,6 +239,7 @@ mod tests {
|
||||
.unwrap()
|
||||
.into_inner()
|
||||
.region_routes()
|
||||
.unwrap()
|
||||
.clone()
|
||||
}
|
||||
|
||||
@@ -396,8 +402,8 @@ mod tests {
|
||||
.unwrap()
|
||||
.into_inner();
|
||||
|
||||
let peers = &extract_all_peers(table_route_value.region_routes());
|
||||
let actual = table_route_value.region_routes();
|
||||
let peers = &extract_all_peers(table_route_value.region_routes().unwrap());
|
||||
let actual = table_route_value.region_routes().unwrap();
|
||||
let expected = &vec![
|
||||
new_region_route(1, peers, 2),
|
||||
new_region_route(2, peers, 3),
|
||||
@@ -416,7 +422,7 @@ mod tests {
|
||||
.unwrap()
|
||||
.into_inner();
|
||||
|
||||
let map = region_distribution(table_route_value.region_routes()).unwrap();
|
||||
let map = region_distribution(table_route_value.region_routes().unwrap()).unwrap();
|
||||
assert_eq!(map.len(), 2);
|
||||
assert_eq!(map.get(&2), Some(&vec![1, 3]));
|
||||
assert_eq!(map.get(&3), Some(&vec![2, 4]));
|
||||
|
||||
@@ -753,7 +753,7 @@ mod tests {
|
||||
.unwrap()
|
||||
.version();
|
||||
// Should be unchanged.
|
||||
assert_eq!(table_routes_version, 0);
|
||||
assert_eq!(table_routes_version.unwrap(), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -244,6 +244,9 @@ impl RegionMigrationManager {
|
||||
// Safety: checked before.
|
||||
let region_route = table_route
|
||||
.region_route(region_id)
|
||||
.context(error::UnexpectedLogicalRouteTableSnafu {
|
||||
err_msg: "{self:?} is a non-physical TableRouteValue.",
|
||||
})?
|
||||
.context(error::RegionRouteNotFoundSnafu { region_id })?;
|
||||
|
||||
if self.has_migrated(®ion_route, &task)? {
|
||||
|
||||
@@ -18,7 +18,7 @@ use common_meta::peer::Peer;
|
||||
use common_meta::rpc::router::RegionRoute;
|
||||
use common_procedure::Status;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use snafu::OptionExt;
|
||||
use snafu::{OptionExt, ResultExt};
|
||||
use store_api::storage::RegionId;
|
||||
|
||||
use super::migration_end::RegionMigrationEnd;
|
||||
@@ -85,6 +85,9 @@ impl RegionMigrationStart {
|
||||
|
||||
let region_route = table_route
|
||||
.region_routes()
|
||||
.context(error::UnexpectedLogicalRouteTableSnafu {
|
||||
err_msg: "{self:?} is a non-physical TableRouteValue.",
|
||||
})?
|
||||
.iter()
|
||||
.find(|route| route.region.id == region_id)
|
||||
.cloned()
|
||||
|
||||
@@ -419,7 +419,7 @@ impl ProcedureMigrationTestSuite {
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.into_inner();
|
||||
let region_routes = table_route.region_routes();
|
||||
let region_routes = table_route.region_routes().unwrap();
|
||||
|
||||
let expected_leader_id = self.context.persistent_ctx.to_peer.id;
|
||||
let removed_follower_id = self.context.persistent_ctx.from_peer.id;
|
||||
|
||||
+3
-3
@@ -208,8 +208,8 @@ mod tests {
|
||||
.unwrap();
|
||||
|
||||
// It should remain unchanged.
|
||||
assert_eq!(latest_table_route.version(), 0);
|
||||
assert!(!latest_table_route.region_routes()[0].is_leader_downgraded());
|
||||
assert_eq!(latest_table_route.version().unwrap(), 0);
|
||||
assert!(!latest_table_route.region_routes().unwrap()[0].is_leader_downgraded());
|
||||
assert!(ctx.volatile_ctx.table_route.is_none());
|
||||
}
|
||||
|
||||
@@ -249,7 +249,7 @@ mod tests {
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
|
||||
assert!(latest_table_route.region_routes()[0].is_leader_downgraded());
|
||||
assert!(latest_table_route.region_routes().unwrap()[0].is_leader_downgraded());
|
||||
assert!(ctx.volatile_ctx.table_route.is_none());
|
||||
}
|
||||
}
|
||||
|
||||
+8
-2
@@ -170,7 +170,10 @@ mod tests {
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.into_inner();
|
||||
assert_eq!(&expected_region_routes, table_route.region_routes());
|
||||
assert_eq!(
|
||||
&expected_region_routes,
|
||||
table_route.region_routes().unwrap()
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -231,6 +234,9 @@ mod tests {
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.into_inner();
|
||||
assert_eq!(&expected_region_routes, table_route.region_routes());
|
||||
assert_eq!(
|
||||
&expected_region_routes,
|
||||
table_route.region_routes().unwrap()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
+13
-3
@@ -33,7 +33,12 @@ impl UpdateMetadata {
|
||||
let region_id = ctx.region_id();
|
||||
let table_route_value = ctx.get_table_route_value().await?.clone();
|
||||
|
||||
let mut region_routes = table_route_value.region_routes().clone();
|
||||
let mut region_routes = table_route_value
|
||||
.region_routes()
|
||||
.context(error::UnexpectedLogicalRouteTableSnafu {
|
||||
err_msg: "{self:?} is a non-physical TableRouteValue.",
|
||||
})?
|
||||
.clone();
|
||||
let region_route = region_routes
|
||||
.iter_mut()
|
||||
.find(|route| route.region.id == region_id)
|
||||
@@ -81,7 +86,12 @@ impl UpdateMetadata {
|
||||
let region_id = ctx.region_id();
|
||||
let table_route_value = ctx.get_table_route_value().await?.clone();
|
||||
|
||||
let region_routes = table_route_value.region_routes().clone();
|
||||
let region_routes = table_route_value
|
||||
.region_routes()
|
||||
.context(error::UnexpectedLogicalRouteTableSnafu {
|
||||
err_msg: "{self:?} is a non-physical TableRouteValue.",
|
||||
})?
|
||||
.clone();
|
||||
let region_route = region_routes
|
||||
.into_iter()
|
||||
.find(|route| route.region.id == region_id)
|
||||
@@ -465,7 +475,7 @@ mod tests {
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.into_inner();
|
||||
let region_routes = table_route.region_routes();
|
||||
let region_routes = table_route.region_routes().unwrap();
|
||||
|
||||
assert!(ctx.volatile_ctx.table_route.is_none());
|
||||
assert!(ctx.volatile_ctx.opening_region_guard.is_none());
|
||||
|
||||
@@ -127,7 +127,7 @@ impl RegionLeaseKeeper {
|
||||
}
|
||||
|
||||
if let Some(table_route) = table_metadata.get(®ion_id.table_id()) {
|
||||
if let Some(region_route) = table_route.region_route(region_id) {
|
||||
if let Ok(Some(region_route)) = table_route.region_route(region_id) {
|
||||
return renew_region_lease_via_region_route(®ion_route, datanode_id, region_id);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -142,13 +142,19 @@ async fn get_leader_peer_ids(
|
||||
.await
|
||||
.context(error::TableMetadataManagerSnafu)
|
||||
.map(|route| {
|
||||
route.map_or_else(Vec::new, |route| {
|
||||
find_leaders(route.region_routes())
|
||||
.into_iter()
|
||||
.map(|peer| peer.id)
|
||||
.collect()
|
||||
})
|
||||
})
|
||||
route.map_or_else(
|
||||
|| Ok(Vec::new()),
|
||||
|route| {
|
||||
let region_routes = route
|
||||
.region_routes()
|
||||
.context(error::UnexpectedLogicalRouteTableSnafu { err_msg: "" })?;
|
||||
Ok(find_leaders(region_routes)
|
||||
.into_iter()
|
||||
.map(|peer| peer.id)
|
||||
.collect())
|
||||
},
|
||||
)
|
||||
})?
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -119,6 +119,13 @@ pub enum Error {
|
||||
region_id: RegionId,
|
||||
location: Location,
|
||||
},
|
||||
|
||||
#[snafu(display("Unexpected table route type: {}", err_msg))]
|
||||
UnexpectedLogicalRouteTable {
|
||||
location: Location,
|
||||
err_msg: String,
|
||||
source: common_meta::error::Error,
|
||||
},
|
||||
}
|
||||
|
||||
impl ErrorExt for Error {
|
||||
@@ -138,6 +145,7 @@ impl ErrorExt for Error {
|
||||
Error::FindDatanode { .. } => StatusCode::InvalidArguments,
|
||||
Error::TableRouteManager { source, .. } => source.status_code(),
|
||||
Error::MissingDefaultValue { .. } => StatusCode::Internal,
|
||||
Error::UnexpectedLogicalRouteTable { source, .. } => source.status_code(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -75,8 +75,13 @@ impl PartitionRuleManager {
|
||||
.context(error::TableRouteManagerSnafu)?
|
||||
.context(error::FindTableRoutesSnafu { table_id })?
|
||||
.into_inner();
|
||||
|
||||
Ok(RegionRoutes(route.region_routes().clone()))
|
||||
let region_routes =
|
||||
route
|
||||
.region_routes()
|
||||
.context(error::UnexpectedLogicalRouteTableSnafu {
|
||||
err_msg: "{self:?} is a non-physical TableRouteValue.",
|
||||
})?;
|
||||
Ok(RegionRoutes(region_routes.clone()))
|
||||
}
|
||||
|
||||
pub async fn find_table_partitions(&self, table_id: TableId) -> Result<Vec<PartitionInfo>> {
|
||||
@@ -87,7 +92,12 @@ impl PartitionRuleManager {
|
||||
.context(error::TableRouteManagerSnafu)?
|
||||
.context(error::FindTableRoutesSnafu { table_id })?
|
||||
.into_inner();
|
||||
let region_routes = route.region_routes();
|
||||
let region_routes =
|
||||
route
|
||||
.region_routes()
|
||||
.context(error::UnexpectedLogicalRouteTableSnafu {
|
||||
err_msg: "{self:?} is a non-physical TableRouteValue.",
|
||||
})?;
|
||||
|
||||
ensure!(
|
||||
!region_routes.is_empty(),
|
||||
|
||||
@@ -521,11 +521,15 @@ CREATE TABLE {table_name} (
|
||||
.unwrap()
|
||||
.into_inner();
|
||||
|
||||
let region_to_dn_map = region_distribution(table_route_value.region_routes())
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|(k, v)| (v[0], *k))
|
||||
.collect::<HashMap<u32, u64>>();
|
||||
let region_to_dn_map = region_distribution(
|
||||
table_route_value
|
||||
.region_routes()
|
||||
.expect("physical table route"),
|
||||
)
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|(k, v)| (v[0], *k))
|
||||
.collect::<HashMap<u32, u64>>();
|
||||
assert!(region_to_dn_map.len() <= instance.datanodes().len());
|
||||
|
||||
let stmt = QueryLanguageParser::parse_sql(&format!(
|
||||
|
||||
@@ -216,11 +216,15 @@ mod tests {
|
||||
.unwrap()
|
||||
.into_inner();
|
||||
|
||||
let region_to_dn_map = region_distribution(table_route_value.region_routes())
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|(k, v)| (v[0], *k))
|
||||
.collect::<HashMap<u32, u64>>();
|
||||
let region_to_dn_map = region_distribution(
|
||||
table_route_value
|
||||
.region_routes()
|
||||
.expect("region routes should be physical"),
|
||||
)
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|(k, v)| (v[0], *k))
|
||||
.collect::<HashMap<u32, u64>>();
|
||||
assert!(region_to_dn_map.len() <= instance.datanodes().len());
|
||||
|
||||
let stmt = QueryLanguageParser::parse_sql("SELECT ts, host FROM demo ORDER BY ts").unwrap();
|
||||
|
||||
Reference in New Issue
Block a user