From a4b884406a18533d0ef23bccb3962fe83fd391b3 Mon Sep 17 00:00:00 2001 From: Weny Xu Date: Tue, 30 May 2023 12:17:59 +0900 Subject: [PATCH] feat: add invalidate cache step (#1658) * feat: add invalidate cache step * refactor: refactor TableIdent * chore: apply suggestions from CR --- src/common/meta/src/instruction.rs | 64 ++++----- .../src/heartbeat/handler/close_region.rs | 16 ++- .../src/heartbeat/handler/open_region.rs | 17 ++- src/datanode/src/tests.rs | 52 ++++--- src/frontend/src/heartbeat/handler/tests.rs | 4 - src/meta-srv/src/error.rs | 12 +- src/meta-srv/src/handler.rs | 3 +- src/meta-srv/src/handler/failure_handler.rs | 18 ++- .../src/handler/failure_handler/runner.rs | 37 +++-- src/meta-srv/src/lock/keys.rs | 6 +- src/meta-srv/src/procedure/region_failover.rs | 47 +++--- .../region_failover/invalidate_cache.rs | 136 ++++++++++++++++++ .../region_failover/update_metadata.rs | 57 +++++--- src/meta-srv/src/sequence.rs | 2 +- src/meta-srv/src/service/mailbox.rs | 35 +++-- tests-integration/tests/region_failover.rs | 13 +- 16 files changed, 369 insertions(+), 150 deletions(-) create mode 100644 src/meta-srv/src/procedure/region_failover/invalidate_cache.rs diff --git a/src/common/meta/src/instruction.rs b/src/common/meta/src/instruction.rs index efb1992234d..0b9bce4bdd6 100644 --- a/src/common/meta/src/instruction.rs +++ b/src/common/meta/src/instruction.rs @@ -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 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 ); } diff --git a/src/datanode/src/heartbeat/handler/close_region.rs b/src/datanode/src/heartbeat/handler/close_region.rs index 08a6b4a0470..355a04a4c69 100644 --- a/src/datanode/src/heartbeat/handler/close_region.rs +++ b/src/datanode/src/heartbeat/handler/close_region.rs @@ -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; diff --git a/src/datanode/src/heartbeat/handler/open_region.rs b/src/datanode/src/heartbeat/handler/open_region.rs index e8369bf0389..361b50e279a 100644 --- a/src/datanode/src/heartbeat/handler/open_region.rs +++ b/src/datanode/src/heartbeat/handler/open_region.rs @@ -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; diff --git a/src/datanode/src/tests.rs b/src/datanode/src/tests.rs index 3720610b4d1..e17c1891422 100644 --- a/src/datanode/src/tests.rs +++ b/src/datanode/src/tests.rs @@ -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, diff --git a/src/frontend/src/heartbeat/handler/tests.rs b/src/frontend/src/heartbeat/handler/tests.rs index 7e495154f01..b3fa1f2810c 100644 --- a/src/frontend/src/heartbeat/handler/tests.rs +++ b/src/frontend/src/heartbeat/handler/tests.rs @@ -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(), diff --git a/src/meta-srv/src/error.rs b/src/meta-srv/src/error.rs index 8babeb4892d..cc9d19a16da 100644 --- a/src/meta-srv/src/error.rs +++ b/src/meta-srv/src/error.rs @@ -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 { .. } diff --git a/src/meta-srv/src/handler.rs b/src/meta-srv/src/handler.rs index 39d417d21d6..7b1205dfb9f 100644 --- a/src/meta-srv/src/handler.rs +++ b/src/meta-srv/src/handler.rs @@ -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; diff --git a/src/meta-srv/src/handler/failure_handler.rs b/src/meta-srv/src/handler/failure_handler.rs index f1637145344..c5b31b24f7b 100644 --- a/src/meta-srv/src/handler/failure_handler.rs +++ b/src/meta-srv/src/handler/failure_handler.rs @@ -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(), diff --git a/src/meta-srv/src/handler/failure_handler/runner.rs b/src/meta-srv/src/handler/failure_handler/runner.rs index 34c764465cf..065ae61574f 100644 --- a/src/meta-srv/src/handler/failure_handler/runner.rs +++ b/src/meta-srv/src/handler/failure_handler/runner.rs @@ -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(|®ion_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(), diff --git a/src/meta-srv/src/lock/keys.rs b/src/meta-srv/src/lock/keys.rs index b1b87244bff..513b308ac08 100644 --- a/src/meta-srv/src/lock/keys.rs +++ b/src/meta-srv/src/lock/keys.rs @@ -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() } diff --git a/src/meta-srv/src/procedure/region_failover.rs b/src/meta-srv/src/procedure/region_failover.rs index 167ba1c9b3d..07de7e7c954 100644 --- a/src/meta-srv/src/procedure/region_failover.rs +++ b/src/meta-srv/src/procedure/region_failover.rs @@ -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( - ®ion_ident.catalog, - ®ion_ident.schema, - ®ion_ident.table + ®ion_ident.table_ident.catalog, + ®ion_ident.table_ident.schema, + ®ion_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>>, + 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 }) }"# ); } } diff --git a/src/meta-srv/src/procedure/region_failover/invalidate_cache.rs b/src/meta-srv/src/procedure/region_failover/invalidate_cache.rs new file mode 100644 index 00000000000..7aff12804aa --- /dev/null +++ b/src/meta-srv/src/procedure/region_failover/invalidate_cache.rs @@ -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, + ctx: &RegionFailoverContext, + failed_region: &RegionIdent, + ) -> Result> { + 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(), + )) + ); + } + } +} diff --git a/src/meta-srv/src/procedure/region_failover/update_metadata.rs b/src/meta-srv/src/procedure/region_failover/update_metadata.rs index 731f7df18dc..e92f72a1689 100644 --- a/src/meta-srv/src/procedure/region_failover/update_metadata.rs +++ b/src/meta-srv/src/procedure/region_failover/update_metadata.rs @@ -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 { diff --git a/src/meta-srv/src/sequence.rs b/src/meta-srv/src/sequence.rs index 0ebed971ef3..a4e763944f9 100644 --- a/src/meta-srv/src/sequence.rs +++ b/src/meta-srv/src/sequence.rs @@ -130,7 +130,7 @@ impl Inner { let value = kv.value; ensure!( value.len() == std::mem::size_of::(), - error::UnexceptedSequenceValueSnafu { + error::UnexpectedSequenceValueSnafu { err_msg: format!("key={}, unexpected value={:?}", self.name, value) } ); diff --git a/src/meta-srv/src/service/mailbox.rs b/src/meta-srv/src/service/mailbox.rs index dd0b266d1c2..97aad7427a2 100644 --- a/src/meta-srv/src/service/mailbox.rs +++ b/src/meta-srv/src/service/mailbox.rs @@ -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; 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>, + ch: Channel, } impl MailboxReceiver { - pub fn new(message_id: MessageId, rx: oneshot::Receiver>) -> Self { - Self { message_id, rx } + pub fn new( + message_id: MessageId, + rx: oneshot::Receiver>, + 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!( diff --git a/tests-integration/tests/region_failover.rs b/tests-integration/tests/region_failover.rs index 83af81f2c22..5166d2a7fd1 100644 --- a/tests-integration/tests/region_failover.rs +++ b/tests-integration/tests/region_failover.rs @@ -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>) -> 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, } }