rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-27 14:16:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add invalidate cache step (#1658)

Browse Source 
* feat: add invalidate cache step

* refactor: refactor TableIdent

* chore: apply suggestions from CR
This commit is contained in:
Weny Xu
2023-05-30 11:17:59 +08:00
committed by GitHub
parent ab5dfd31ec
commit a4b884406a
16 changed files with 369 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/meta/src/instruction.rs
+31 -33
View File
@@ -22,11 +22,7 @@ use crate::{ClusterId, DatanodeId};
pub struct RegionIdent {
pub cluster_id: ClusterId,
pub datanode_id: DatanodeId,
pub catalog: String,
pub schema: String,
pub table: String,
pub table_id: u32,
pub engine: String,
pub table_ident: TableIdent,
pub region_number: u32,
}
@@ -37,20 +33,24 @@ impl Display for RegionIdent {
"RegionIdent(datanode_id='{}.{}', table_id='{}', table_name='{}.{}.{}', table_engine='{}', region_no='{}')",
self.cluster_id,
self.datanode_id,
self.table_id,
self.catalog,
self.schema,
self.table,
self.engine,
self.table_ident.table_id,
self.table_ident.catalog,
self.table_ident.schema,
self.table_ident.table,
self.table_ident.engine,
self.region_number
)
}
}
impl From<RegionIdent> for TableIdent {
fn from(region_ident: RegionIdent) -> Self {
region_ident.table_ident
}
}
#[derive(Eq, Hash, PartialEq, Clone, Debug, Serialize, Deserialize)]
pub struct TableIdent {
pub cluster_id: ClusterId,
pub datanode_id: DatanodeId,
pub catalog: String,
pub schema: String,
pub table: String,
@@ -62,14 +62,8 @@ impl Display for TableIdent {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
"TableIdent(datanode_id='{}.{}', table_id='{}', table_name='{}.{}.{}', table_engine='{}')",
self.cluster_id,
self.datanode_id,
self.table_id,
self.catalog,
self.schema,
self.table,
self.engine,
"TableIdent(table_id='{}', table_name='{}.{}.{}', table_engine='{}')",
self.table_id, self.catalog, self.schema, self.table, self.engine,
)
}
}
@@ -86,7 +80,7 @@ impl Display for SimpleReply {
}
}
#[derive(Debug, Serialize, Deserialize)]
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum Instruction {
OpenRegion(RegionIdent),
@@ -133,36 +127,40 @@ mod tests {
let open_region = Instruction::OpenRegion(RegionIdent {
cluster_id: 1,
datanode_id: 2,
catalog: "foo".to_string(),
schema: "bar".to_string(),
table: "hi".to_string(),
table_id: 1024,
engine: "mito".to_string(),
table_ident: TableIdent {
catalog: "foo".to_string(),
schema: "bar".to_string(),
table: "hi".to_string(),
table_id: 1024,
engine: "mito".to_string(),
},
region_number: 1,
});
let serialized = serde_json::to_string(&open_region).unwrap();
assert_eq!(
r#"{"type":"open_region","cluster_id":1,"datanode_id":2,"catalog":"foo","schema":"bar","table":"hi","table_id":1024,"engine":"mito","region_number":1}"#,
r#"{"type":"open_region","cluster_id":1,"datanode_id":2,"table_ident":{"catalog":"foo","schema":"bar","table":"hi","table_id":1024,"engine":"mito"},"region_number":1}"#,
serialized
);
let close_region = Instruction::CloseRegion(RegionIdent {
cluster_id: 1,
datanode_id: 2,
catalog: "foo".to_string(),
schema: "bar".to_string(),
table: "hi".to_string(),
table_id: 1024,
engine: "mito".to_string(),
table_ident: TableIdent {
catalog: "foo".to_string(),
schema: "bar".to_string(),
table: "hi".to_string(),
table_id: 1024,
engine: "mito".to_string(),
},
region_number: 1,
});
let serialized = serde_json::to_string(&close_region).unwrap();
assert_eq!(
r#"{"type":"close_region","cluster_id":1,"datanode_id":2,"catalog":"foo","schema":"bar","table":"hi","table_id":1024,"engine":"mito","region_number":1}"#,
r#"{"type":"close_region","cluster_id":1,"datanode_id":2,"table_ident":{"catalog":"foo","schema":"bar","table":"hi","table_id":1024,"engine":"mito"},"region_number":1}"#,
serialized
);
}
src/datanode/src/heartbeat/handler/close_region.rs
+11 -5
View File
@@ -20,7 +20,9 @@ use common_meta::error::Result as MetaResult;
use common_meta::heartbeat::handler::{
HandleControl, HeartbeatResponseHandler, HeartbeatResponseHandlerContext,
};
use common_meta::instruction::{Instruction, InstructionReply, RegionIdent, SimpleReply};
use common_meta::instruction::{
Instruction, InstructionReply, RegionIdent, SimpleReply, TableIdent,
};
use common_telemetry::{error, info, warn};
use snafu::ResultExt;
use store_api::storage::RegionNumber;
@@ -53,10 +55,14 @@ impl HeartbeatResponseHandler for CloseRegionHandler {
let self_ref = Arc::new(self.clone());
let RegionIdent {
engine,
catalog,
schema,
table,
table_ident:
TableIdent {
engine,
catalog,
schema,
table,
..
},
region_number,
..
} = region_ident;
src/datanode/src/heartbeat/handler/open_region.rs
+11 -6
View File
@@ -21,7 +21,9 @@ use common_meta::error::Result as MetaResult;
use common_meta::heartbeat::handler::{
HandleControl, HeartbeatResponseHandler, HeartbeatResponseHandlerContext,
};
use common_meta::instruction::{Instruction, InstructionReply, RegionIdent, SimpleReply};
use common_meta::instruction::{
Instruction, InstructionReply, RegionIdent, SimpleReply, TableIdent,
};
use common_telemetry::{error, warn};
use snafu::ResultExt;
use store_api::storage::RegionNumber;
@@ -97,12 +99,15 @@ impl OpenRegionHandler {
fn prepare_request(ident: RegionIdent) -> (String, OpenTableRequest) {
let RegionIdent {
catalog,
schema,
table,
table_id,
table_ident:
TableIdent {
catalog,
schema,
table,
table_id,
engine,
},
region_number,
engine,
..
} = ident;
src/datanode/src/tests.rs
+31 -21
View File
@@ -24,7 +24,9 @@ use common_meta::heartbeat::handler::{
HandlerGroupExecutor, HeartbeatResponseHandlerContext, HeartbeatResponseHandlerExecutor,
};
use common_meta::heartbeat::mailbox::{HeartbeatMailbox, MessageMeta};
use common_meta::instruction::{Instruction, InstructionReply, RegionIdent, SimpleReply};
use common_meta::instruction::{
Instruction, InstructionReply, RegionIdent, SimpleReply, TableIdent,
};
use common_query::Output;
use datatypes::prelude::ConcreteDataType;
use servers::query_handler::grpc::GrpcQueryHandler;
@@ -95,11 +97,13 @@ async fn test_close_region_handler() {
executor.clone(),
mailbox.clone(),
Instruction::CloseRegion(RegionIdent {
catalog: "greptime".to_string(),
schema: "public".to_string(),
table: "non-exist".to_string(),
table_id: 1025,
engine: "mito".to_string(),
table_ident: TableIdent {
catalog: "greptime".to_string(),
schema: "public".to_string(),
table: "non-exist".to_string(),
table_id: 1025,
engine: "mito".to_string(),
},
region_number: 0,
cluster_id: 1,
datanode_id: 2,
@@ -149,11 +153,13 @@ async fn test_open_region_handler() {
executor.clone(),
mailbox.clone(),
Instruction::OpenRegion(RegionIdent {
catalog: "greptime".to_string(),
schema: "public".to_string(),
table: "non-exist".to_string(),
table_id: 2024,
engine: "mito".to_string(),
table_ident: TableIdent {
catalog: "greptime".to_string(),
schema: "public".to_string(),
table: "non-exist".to_string(),
table_id: 2024,
engine: "mito".to_string(),
},
region_number: 0,
cluster_id: 1,
datanode_id: 2,
@@ -228,11 +234,13 @@ fn handle_instruction(
fn close_region_instruction() -> Instruction {
Instruction::CloseRegion(RegionIdent {
catalog: "greptime".to_string(),
schema: "public".to_string(),
table: "demo".to_string(),
table_id: 1024,
engine: "mito".to_string(),
table_ident: TableIdent {
catalog: "greptime".to_string(),
schema: "public".to_string(),
table: "demo".to_string(),
table_id: 1024,
engine: "mito".to_string(),
},
region_number: 0,
cluster_id: 1,
datanode_id: 2,
@@ -241,11 +249,13 @@ fn close_region_instruction() -> Instruction {
fn open_region_instruction() -> Instruction {
Instruction::OpenRegion(RegionIdent {
catalog: "greptime".to_string(),
schema: "public".to_string(),
table: "demo".to_string(),
table_id: 1024,
engine: "mito".to_string(),
table_ident: TableIdent {
catalog: "greptime".to_string(),
schema: "public".to_string(),
table: "demo".to_string(),
table_id: 1024,
engine: "mito".to_string(),
},
region_number: 0,
cluster_id: 1,
datanode_id: 2,
src/frontend/src/heartbeat/handler/tests.rs
-4
View File
@@ -82,8 +82,6 @@ async fn test_invalidate_table_cache_handler() {
executor.clone(),
mailbox.clone(),
Instruction::InvalidateTableCache(TableIdent {
cluster_id: 1,
datanode_id: 2,
catalog: "test".to_string(),
schema: "greptime".to_string(),
table: "foo_table".to_string(),
@@ -120,8 +118,6 @@ async fn test_invalidate_table_cache_handler() {
executor,
mailbox,
Instruction::InvalidateTableCache(TableIdent {
cluster_id: 1,
datanode_id: 2,
catalog: "test".to_string(),
schema: "greptime".to_string(),
table: "not_found".to_string(),
src/meta-srv/src/error.rs
+8 -4
View File
@@ -134,8 +134,8 @@ pub enum Error {
source: common_catalog::error::Error,
},
#[snafu(display("Unexcepted sequence value: {}", err_msg))]
UnexceptedSequenceValue { err_msg: String, location: Location },
#[snafu(display("Unexpected sequence value: {}", err_msg))]
UnexpectedSequenceValue { err_msg: String, location: Location },
#[snafu(display("Failed to decode table route, source: {}", source))]
DecodeTableRoute {
@@ -346,6 +346,9 @@ pub enum Error {
#[snafu(display("Expected to retry later, reason: {}", reason))]
RetryLater { reason: String, location: Location },
#[snafu(display("Combine error: {}", err_msg))]
Combine { err_msg: String, location: Location },
#[snafu(display("Failed to convert table route, source: {}", source))]
TableRouteConversion {
#[snafu(backtrace)]
@@ -404,7 +407,8 @@ impl ErrorExt for Error {
| Error::MailboxTimeout { .. }
| Error::MailboxReceiver { .. }
| Error::RetryLater { .. }
| Error::StartGrpc { .. } => StatusCode::Internal,
| Error::StartGrpc { .. }
| Error::Combine { .. } => StatusCode::Internal,
Error::EmptyKey { .. }
| Error::MissingRequiredParameter { .. }
| Error::MissingRequestHeader { .. }
@@ -418,7 +422,7 @@ impl ErrorExt for Error {
| Error::LeaseValueFromUtf8 { .. }
| Error::StatKeyFromUtf8 { .. }
| Error::StatValueFromUtf8 { .. }
| Error::UnexceptedSequenceValue { .. }
| Error::UnexpectedSequenceValue { .. }
| Error::TableRouteNotFound { .. }
| Error::CorruptedTableRoute { .. }
| Error::NextSequence { .. }
src/meta-srv/src/handler.rs
+2 -1
View File
@@ -346,7 +346,7 @@ impl Mailbox for HeartbeatMailbox {
self.pushers.push(&pusher_id, msg).await?;
Ok(MailboxReceiver::new(message_id, rx))
Ok(MailboxReceiver::new(message_id, rx, *ch))
}
async fn broadcast(&self, ch: &BroadcastChannel, msg: &MailboxMessage) -> Result<()> {
@@ -371,6 +371,7 @@ impl Mailbox for HeartbeatMailbox {
#[cfg(test)]
mod tests {
use std::sync::Arc;
use std::time::Duration;
src/meta-srv/src/handler/failure_handler.rs
+11 -7
View File
@@ -19,6 +19,7 @@ use std::sync::Arc;
use api::v1::meta::{HeartbeatRequest, Role};
use async_trait::async_trait;
use common_catalog::consts::MITO_ENGINE;
use common_meta::instruction::TableIdent;
use common_meta::RegionIdent;
use crate::error::Result;
@@ -83,15 +84,18 @@ impl HeartbeatHandler for RegionFailureHandler {
.region_stats
.iter()
.map(|x| RegionIdent {
catalog: x.catalog.clone(),
schema: x.schema.clone(),
table: x.table.clone(),
cluster_id: stat.cluster_id,
datanode_id: stat.id,
// TODO(#1566): Use the real table id.
table_id: 0,
// TODO(#1583): Use the actual table engine.
engine: MITO_ENGINE.to_string(),
table_ident: TableIdent {
catalog: x.catalog.clone(),
schema: x.schema.clone(),
table: x.table.clone(),
// TODO(#1566): Use the real table id.
table_id: 0,
// TODO(#1583): Use the actual table engine.
engine: MITO_ENGINE.to_string(),
},
region_number: x.id as u32,
})
.collect(),
src/meta-srv/src/handler/failure_handler/runner.rs
+22 -15
View File
@@ -243,6 +243,7 @@ impl FailureDetectorContainer {
#[cfg(test)]
mod tests {
use common_catalog::consts::MITO_ENGINE;
use common_meta::instruction::TableIdent;
use rand::Rng;
use super::*;
@@ -252,13 +253,15 @@ mod tests {
fn test_default_failure_detector_container() {
let container = FailureDetectorContainer(DashMap::new());
let ident = RegionIdent {
catalog: "a".to_string(),
schema: "b".to_string(),
table: "c".to_string(),
table_ident: TableIdent {
catalog: "a".to_string(),
schema: "b".to_string(),
table: "c".to_string(),
table_id: 1,
engine: MITO_ENGINE.to_string(),
},
cluster_id: 3,
datanode_id: 2,
table_id: 1,
engine: MITO_ENGINE.to_string(),
region_number: 1,
};
let _ = container.get_failure_detector(ident.clone());
@@ -279,13 +282,15 @@ mod tests {
let container = FailureDetectorContainer(DashMap::new());
let ident = RegionIdent {
catalog: "a".to_string(),
schema: "b".to_string(),
table: "c".to_string(),
table_ident: TableIdent {
catalog: "a".to_string(),
schema: "b".to_string(),
table: "c".to_string(),
table_id: 1,
engine: MITO_ENGINE.to_string(),
},
cluster_id: 3,
datanode_id: 2,
table_id: 1,
engine: MITO_ENGINE.to_string(),
region_number: 1,
};
container.get_failure_detector(ident.clone());
@@ -318,13 +323,15 @@ mod tests {
region_idents: region_ids
.iter()
.map(|&region_number| RegionIdent {
catalog: "a".to_string(),
schema: "b".to_string(),
table: "c".to_string(),
table_ident: TableIdent {
catalog: "a".to_string(),
schema: "b".to_string(),
table: "c".to_string(),
table_id: 0,
engine: MITO_ENGINE.to_string(),
},
cluster_id: 1,
datanode_id,
table_id: 0,
engine: MITO_ENGINE.to_string(),
region_number,
})
.collect(),
src/meta-srv/src/lock/keys.rs
+5 -1
View File
@@ -22,7 +22,11 @@ use crate::lock::Key;
pub(crate) fn table_metadata_lock_key(region: &RegionIdent) -> Key {
format!(
"table_metadata_lock_({}-{}.{}.{}-{})",
region.cluster_id, region.catalog, region.schema, region.table, region.table_id,
region.cluster_id,
region.table_ident.catalog,
region.table_ident.schema,
region.table_ident.table,
region.table_ident.table_id,
)
.into_bytes()
}
src/meta-srv/src/procedure/region_failover.rs
+30 -17
View File
@@ -16,6 +16,7 @@ mod activate_region;
mod deactivate_region;
mod failover_end;
mod failover_start;
mod invalidate_cache;
mod update_metadata;
use std::collections::HashSet;
@@ -225,6 +226,14 @@ trait State: Sync + Send + Debug {
/// │ Updates the Region
/// │ placement metadata
/// │
/// ┌───────▼───────┐
/// │InvalidateCache│
/// └───────┬───────┘
/// │
/// │ Broadcast Invalidate Table
/// │ Cache
/// │
/// │
/// ┌────────▼────────┐
/// │RegionFailoverEnd│
/// └─────────────────┘
@@ -289,9 +298,9 @@ impl Procedure for RegionFailoverProcedure {
let key = format!(
"{}/region-{}",
common_catalog::format_full_table_name(
&region_ident.catalog,
&region_ident.schema,
&region_ident.table
&region_ident.table_ident.catalog,
&region_ident.table_ident.schema,
&region_ident.table_ident.table
),
region_ident.region_number
);
@@ -307,7 +316,7 @@ mod tests {
use api::v1::meta::{HeartbeatResponse, MailboxMessage, Peer, RequestHeader};
use catalog::helper::TableGlobalKey;
use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, MITO_ENGINE};
use common_meta::instruction::{Instruction, InstructionReply, SimpleReply};
use common_meta::instruction::{Instruction, InstructionReply, SimpleReply, TableIdent};
use common_meta::DatanodeId;
use common_procedure::BoxedProcedure;
use rand::prelude::SliceRandom;
@@ -357,6 +366,7 @@ mod tests {
pub struct TestingEnv {
pub context: RegionFailoverContext,
pub heartbeat_receivers: HashMap<DatanodeId, Receiver<tonic::Result<HeartbeatResponse>>>,
pub pushers: Pushers,
}
impl TestingEnv {
@@ -386,13 +396,15 @@ mod tests {
.unwrap();
RegionIdent {
cluster_id: 0,
datanode_id: failed_datanode,
table_id: 1,
engine: MITO_ENGINE.to_string(),
region_number,
catalog: DEFAULT_CATALOG_NAME.to_string(),
schema: DEFAULT_SCHEMA_NAME.to_string(),
table: table.to_string(),
datanode_id: failed_datanode,
table_ident: TableIdent {
table_id: 1,
engine: MITO_ENGINE.to_string(),
catalog: DEFAULT_CATALOG_NAME.to_string(),
schema: DEFAULT_SCHEMA_NAME.to_string(),
table: table.to_string(),
},
}
}
}
@@ -435,7 +447,7 @@ mod tests {
let mailbox_sequence =
Sequence::new("test_heartbeat_mailbox", 0, 100, kv_store.clone());
let mailbox = HeartbeatMailbox::create(pushers, mailbox_sequence);
let mailbox = HeartbeatMailbox::create(pushers.clone(), mailbox_sequence);
let selector = self.selector.unwrap_or_else(|| {
let nodes = (1..=table_global_value.regions_id_map.len())
@@ -461,6 +473,7 @@ mod tests {
selector_ctx,
dist_lock: Arc::new(MemLock::default()),
},
pushers,
heartbeat_receivers,
}
}
@@ -569,14 +582,14 @@ mod tests {
assert_eq!(
procedure.dump().unwrap(),
r#"{"failed_region":{"cluster_id":0,"datanode_id":1,"catalog":"greptime","schema":"public","table":"my_table","table_id":1,"engine":"mito","region_number":1},"state":{"region_failover_state":"RegionFailoverEnd"}}"#
r#"{"failed_region":{"cluster_id":0,"datanode_id":1,"table_ident":{"catalog":"greptime","schema":"public","table":"my_table","table_id":1,"engine":"mito"},"region_number":1},"state":{"region_failover_state":"RegionFailoverEnd"}}"#
);
// Verifies that the failed region (region 1) is moved from failed datanode (datanode 1) to the candidate datanode.
let key = TableGlobalKey {
catalog_name: failed_region.catalog.clone(),
schema_name: failed_region.schema.clone(),
table_name: failed_region.table.clone(),
catalog_name: failed_region.table_ident.catalog.clone(),
schema_name: failed_region.table_ident.schema.clone(),
table_name: failed_region.table_ident.table.clone(),
};
let value = table_routes::get_table_global_value(&env.context.selector_ctx.kv_store, &key)
.await
@@ -614,12 +627,12 @@ mod tests {
let s = procedure.dump().unwrap();
assert_eq!(
s,
r#"{"failed_region":{"cluster_id":0,"datanode_id":1,"catalog":"greptime","schema":"public","table":"my_table","table_id":1,"engine":"mito","region_number":1},"state":{"region_failover_state":"RegionFailoverStart","failover_candidate":null}}"#
r#"{"failed_region":{"cluster_id":0,"datanode_id":1,"table_ident":{"catalog":"greptime","schema":"public","table":"my_table","table_id":1,"engine":"mito"},"region_number":1},"state":{"region_failover_state":"RegionFailoverStart","failover_candidate":null}}"#
);
let n: Node = serde_json::from_str(&s).unwrap();
assert_eq!(
format!("{n:?}"),
r#"Node { failed_region: RegionIdent { cluster_id: 0, datanode_id: 1, catalog: "greptime", schema: "public", table: "my_table", table_id: 1, engine: "mito", region_number: 1 }, state: Some(RegionFailoverStart { failover_candidate: None }) }"#
r#"Node { failed_region: RegionIdent { cluster_id: 0, datanode_id: 1, table_ident: TableIdent { catalog: "greptime", schema: "public", table: "my_table", table_id: 1, engine: "mito" }, region_number: 1 }, state: Some(RegionFailoverStart { failover_candidate: None }) }"#
);
}
}
src/meta-srv/src/procedure/region_failover/invalidate_cache.rs
+136
View File
@@ -0,0 +1,136 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use api::v1::meta::MailboxMessage;
use async_trait::async_trait;
use common_meta::instruction::{Instruction, TableIdent};
use common_meta::RegionIdent;
use serde::{Deserialize, Serialize};
use snafu::ResultExt;
use super::failover_end::RegionFailoverEnd;
use super::{RegionFailoverContext, State};
use crate::error::{self, Result};
use crate::service::mailbox::BroadcastChannel;
#[derive(Serialize, Deserialize, Debug, Default)]
pub(super) struct InvalidateCache;
impl InvalidateCache {
async fn broadcast_invalidate_table_cache_messages(
&self,
ctx: &RegionFailoverContext,
table_ident: &TableIdent,
) -> Result<()> {
let instruction = Instruction::InvalidateTableCache(table_ident.clone());
let msg = &MailboxMessage::json_message(
"Invalidate Table Cache",
&format!("Metasrv@{}", ctx.selector_ctx.server_addr),
"Frontend broadcast",
common_time::util::current_time_millis(),
&instruction,
)
.with_context(|_| error::SerializeToJsonSnafu {
input: instruction.to_string(),
})?;
ctx.mailbox
.broadcast(&BroadcastChannel::Frontend, msg)
.await
}
}
#[async_trait]
#[typetag::serde]
impl State for InvalidateCache {
async fn next(
mut self: Box<Self>,
ctx: &RegionFailoverContext,
failed_region: &RegionIdent,
) -> Result<Box<dyn State>> {
let table_ident = TableIdent::from(failed_region.clone());
self.broadcast_invalidate_table_cache_messages(ctx, &table_ident)
.await?;
Ok(Box::new(RegionFailoverEnd))
}
}
#[cfg(test)]
mod tests {
use api::v1::meta::mailbox_message::Payload;
use api::v1::meta::RequestHeader;
use super::super::tests::TestingEnvBuilder;
use super::*;
use crate::handler::Pusher;
use crate::procedure::region_failover::tests::TestingEnv;
use crate::service::mailbox::Channel;
#[tokio::test]
async fn test_invalidate_table_cache() {
common_telemetry::init_default_ut_logging();
let env = TestingEnvBuilder::new().build().await;
let failed_region = env.failed_region(1).await;
let TestingEnv {
mut heartbeat_receivers,
context,
pushers,
} = env;
for frontend_id in 4..=7 {
let (tx, rx) = tokio::sync::mpsc::channel(1);
let pusher_id = Channel::Frontend(frontend_id).pusher_id();
let pusher = Pusher::new(tx, &RequestHeader::default());
let _ = pushers.insert(pusher_id, pusher).await;
heartbeat_receivers.insert(frontend_id, rx);
}
let state = InvalidateCache;
let table_ident: TableIdent = failed_region.clone().into();
// lexicographical order
// frontend-4,5,6,7
let next_state = Box::new(state)
.next(&context, &failed_region)
.await
.unwrap();
assert_eq!(format!("{next_state:?}"), "RegionFailoverEnd");
for i in 4..=7 {
// frontend id starts from 4
let rx = heartbeat_receivers.get_mut(&i).unwrap();
let resp = rx.recv().await.unwrap().unwrap();
let received = &resp.mailbox_message.unwrap();
assert_eq!(received.id, 0);
assert_eq!(received.subject, "Invalidate Table Cache");
assert_eq!(received.from, "Metasrv@127.0.0.1:3002");
assert_eq!(received.to, "Frontend broadcast");
assert_eq!(
received.payload,
Some(Payload::Json(
serde_json::to_string(&Instruction::InvalidateTableCache(table_ident.clone()))
.unwrap(),
))
);
}
}
}
src/meta-srv/src/procedure/region_failover/update_metadata.rs
+36 -21
View File
@@ -22,7 +22,7 @@ use common_telemetry::info;
use serde::{Deserialize, Serialize};
use snafu::{OptionExt, ResultExt};
use super::failover_end::RegionFailoverEnd;
use super::invalidate_cache::InvalidateCache;
use super::{RegionFailoverContext, State};
use crate::error::{
CorruptedTableRouteSnafu, Result, RetryLaterSnafu, TableNotFoundSnafu,
@@ -75,9 +75,9 @@ impl UpdateRegionMetadata {
failed_region: &RegionIdent,
) -> Result<()> {
let key = TableGlobalKey {
catalog_name: failed_region.catalog.clone(),
schema_name: failed_region.schema.clone(),
table_name: failed_region.table.clone(),
catalog_name: failed_region.table_ident.catalog.clone(),
schema_name: failed_region.table_ident.schema.clone(),
table_name: failed_region.table_ident.table.clone(),
};
let mut value = table_routes::get_table_global_value(&ctx.selector_ctx.kv_store, &key)
.await?
@@ -120,11 +120,12 @@ impl UpdateRegionMetadata {
failed_region: &RegionIdent,
) -> Result<()> {
let table_name = TableName {
catalog_name: failed_region.catalog.clone(),
schema_name: failed_region.schema.clone(),
table_name: failed_region.table.clone(),
catalog_name: failed_region.table_ident.catalog.clone(),
schema_name: failed_region.table_ident.schema.clone(),
table_name: failed_region.table_ident.table.clone(),
};
let key = TableRouteKey::with_table_name(failed_region.table_id as _, &table_name);
let key =
TableRouteKey::with_table_name(failed_region.table_ident.table_id as _, &table_name);
let value = table_routes::get_table_route_value(&ctx.selector_ctx.kv_store, &key).await?;
let table_route = value
@@ -212,7 +213,7 @@ impl State for UpdateRegionMetadata {
}
.build()
})?;
Ok(Box::new(RegionFailoverEnd))
Ok(Box::new(InvalidateCache))
}
}
@@ -222,9 +223,23 @@ mod tests {
use catalog::helper::TableGlobalValue;
use super::super::tests::{TestingEnv, TestingEnvBuilder};
use super::*;
use super::{State, *};
use crate::table_routes::tests::new_region_route;
#[tokio::test]
async fn test_next_state() {
let env = TestingEnvBuilder::new().build().await;
let failed_region = env.failed_region(1).await;
let state = UpdateRegionMetadata::new(Peer::new(2, ""));
let next_state = Box::new(state)
.next(&env.context, &failed_region)
.await
.unwrap();
assert_eq!(format!("{next_state:?}"), "InvalidateCache");
}
#[tokio::test]
async fn test_update_table_global_value() {
common_telemetry::init_default_ut_logging();
@@ -233,9 +248,9 @@ mod tests {
let failed_region = env.failed_region(failed_region).await;
let key = TableGlobalKey {
catalog_name: failed_region.catalog.clone(),
schema_name: failed_region.schema.clone(),
table_name: failed_region.table.clone(),
catalog_name: failed_region.table_ident.catalog.clone(),
schema_name: failed_region.table_ident.schema.clone(),
table_name: failed_region.table_ident.table.clone(),
};
let original =
@@ -323,10 +338,10 @@ mod tests {
.unwrap();
let key = TableRouteKey {
table_id: failed_region.table_id as u64,
catalog_name: &failed_region.catalog,
schema_name: &failed_region.schema,
table_name: &failed_region.table,
table_id: failed_region.table_ident.table_id as u64,
catalog_name: &failed_region.table_ident.catalog,
schema_name: &failed_region.table_ident.schema,
table_name: &failed_region.table_ident.table,
};
table_routes::get_table_route_value(&env.context.selector_ctx.kv_store, &key)
.await
@@ -456,10 +471,10 @@ mod tests {
let failed_region_1 = env.failed_region(1).await;
let failed_region_2 = env.failed_region(2).await;
let catalog_name = failed_region_1.catalog.clone();
let schema_name = failed_region_1.schema.clone();
let table_name = failed_region_1.table.clone();
let table_id = failed_region_1.table_id as u64;
let catalog_name = failed_region_1.table_ident.catalog.clone();
let schema_name = failed_region_1.table_ident.schema.clone();
let table_name = failed_region_1.table_ident.table.clone();
let table_id = failed_region_1.table_ident.table_id as u64;
futures::future::join_all(vec![
tokio::spawn(async move {
src/meta-srv/src/sequence.rs
+1 -1
View File
@@ -130,7 +130,7 @@ impl Inner {
let value = kv.value;
ensure!(
value.len() == std::mem::size_of::<u64>(),
error::UnexceptedSequenceValueSnafu {
error::UnexpectedSequenceValueSnafu {
err_msg: format!("key={}, unexpected value={:?}", self.name, value)
}
);
src/meta-srv/src/service/mailbox.rs
+26 -9
View File
@@ -12,6 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::fmt::{Display, Formatter};
use std::ops::Range;
use std::pin::Pin;
use std::sync::Arc;
@@ -28,11 +29,25 @@ pub type MailboxRef = Arc<dyn Mailbox>;
pub type MessageId = u64;
#[derive(Debug, PartialEq, Clone, Copy)]
pub enum Channel {
Datanode(u64),
Frontend(u64),
}
impl Display for Channel {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
match self {
Channel::Datanode(id) => {
write!(f, "Datanode-{}", id)
}
Channel::Frontend(id) => {
write!(f, "Frontend-{}", id)
}
}
}
}
impl Channel {
pub(crate) fn pusher_id(&self) -> String {
match self {
@@ -41,7 +56,6 @@ impl Channel {
}
}
}
pub enum BroadcastChannel {
Datanode,
Frontend,
@@ -65,16 +79,25 @@ impl BroadcastChannel {
pub struct MailboxReceiver {
message_id: MessageId,
rx: oneshot::Receiver<Result<MailboxMessage>>,
ch: Channel,
}
impl MailboxReceiver {
pub fn new(message_id: MessageId, rx: oneshot::Receiver<Result<MailboxMessage>>) -> Self {
Self { message_id, rx }
pub fn new(
message_id: MessageId,
rx: oneshot::Receiver<Result<MailboxMessage>>,
ch: Channel,
) -> Self {
Self { message_id, rx, ch }
}
pub fn message_id(&self) -> MessageId {
self.message_id
}
pub fn channel(&self) -> Channel {
self.ch
}
}
impl Future for MailboxReceiver {
@@ -111,12 +134,6 @@ pub trait Mailbox: Send + Sync {
mod tests {
use super::*;
#[test]
fn test_channel_pusher_id() {
assert_eq!(Channel::Datanode(42).pusher_id(), "0-42");
assert_eq!(Channel::Frontend(42).pusher_id(), "1-42");
}
#[test]
fn test_channel_pusher_range() {
assert_eq!(
tests-integration/tests/region_failover.rs
+8 -5
View File
@@ -16,6 +16,7 @@ use std::collections::HashMap;
use catalog::helper::TableGlobalKey;
use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, MITO_ENGINE};
use common_meta::instruction::TableIdent;
use common_meta::RegionIdent;
use common_procedure::{watcher, ProcedureWithId};
use common_telemetry::info;
@@ -110,11 +111,13 @@ fn choose_failed_region(distribution: HashMap<u64, Vec<u32>>) -> RegionIdent {
RegionIdent {
cluster_id: 1000,
datanode_id: failed_datanode,
table_id: 1025,
engine: MITO_ENGINE.to_string(),
catalog: DEFAULT_CATALOG_NAME.to_string(),
schema: DEFAULT_SCHEMA_NAME.to_string(),
table: "my_table".to_string(),
table_ident: TableIdent {
table_id: 1025,
engine: MITO_ENGINE.to_string(),
catalog: DEFAULT_CATALOG_NAME.to_string(),
schema: DEFAULT_SCHEMA_NAME.to_string(),
table: "my_table".to_string(),
},
region_number: failed_region,
}
}