diff --git a/Cargo.lock b/Cargo.lock index ed5c19ab2a..abaf53df47 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5939,7 +5939,7 @@ dependencies = [ [[package]] name = "greptime-proto" version = "0.1.0" -source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=b6b665d8b6fce4f1f0da49c8c79fb150636df4ff#b6b665d8b6fce4f1f0da49c8c79fb150636df4ff" +source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=4a69c9e75240a76afeb0796798085813669288f8#4a69c9e75240a76afeb0796798085813669288f8" dependencies = [ "prost 0.14.1", "prost-types 0.14.1", @@ -11115,7 +11115,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "22505a5c94da8e3b7c2996394d1c933236c4d743e81a410bcca4e6989fc066a4" dependencies = [ "bytes", - "heck 0.4.1", + "heck 0.5.0", "itertools 0.12.1", "log", "multimap", @@ -11135,8 +11135,8 @@ version = "0.14.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ac6c3320f9abac597dcbc668774ef006702672474aad53c6d596b62e487b40b1" dependencies = [ - "heck 0.4.1", - "itertools 0.14.0", + "heck 0.5.0", + "itertools 0.11.0", "log", "multimap", "once_cell", @@ -11184,7 +11184,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.11.0", "proc-macro2", "quote", "syn 2.0.117", @@ -11197,7 +11197,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9120690fafc389a67ba3803df527d0ec9cbbc9cc45e4cc20b332996dfb672425" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.11.0", "proc-macro2", "quote", "syn 2.0.117", @@ -13484,7 +13484,7 @@ version = "0.8.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1961e2ef424c1424204d3a5d6975f934f56b6d50ff5732382d84ebf460e147f7" dependencies = [ - "heck 0.4.1", + "heck 0.5.0", "proc-macro2", "quote", "syn 2.0.117", diff --git a/Cargo.toml b/Cargo.toml index c7412cde0f..10111156ce 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -158,7 +158,7 @@ fs2 = "0.4" fst = "0.4.7" futures = "0.3" futures-util = "0.3" -greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "b6b665d8b6fce4f1f0da49c8c79fb150636df4ff" } +greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "4a69c9e75240a76afeb0796798085813669288f8" } hex = "0.4" http = "1" humantime = "2.1" diff --git a/src/cmd/src/standalone.rs b/src/cmd/src/standalone.rs index 142068711a..5096c9221c 100644 --- a/src/cmd/src/standalone.rs +++ b/src/cmd/src/standalone.rs @@ -551,6 +551,7 @@ impl StartCommand { flow_metadata_manager: flow_metadata_manager.clone(), flow_metadata_allocator: flow_metadata_allocator.clone(), region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl), + soft_drop_enabled: false, }; let ddl_manager = DdlManager::try_new( diff --git a/src/common/meta/src/ddl.rs b/src/common/meta/src/ddl.rs index 126c360d92..cc578b7552 100644 --- a/src/common/meta/src/ddl.rs +++ b/src/common/meta/src/ddl.rs @@ -44,12 +44,14 @@ pub mod drop_flow; pub mod drop_table; pub mod drop_view; pub mod flow_meta; +pub mod purge_dropped_table; pub mod table_meta; #[cfg(any(test, feature = "testing"))] pub mod test_util; #[cfg(test)] pub(crate) mod tests; pub mod truncate_table; +pub mod undrop_table; pub mod utils; /// Metadata allocated to a table. @@ -117,6 +119,8 @@ pub struct DdlContext { pub flow_metadata_allocator: FlowMetadataAllocatorRef, /// controller of region failure detector. pub region_failure_detector_controller: RegionFailureDetectorControllerRef, + /// Whether table drops should stop after tombstoning metadata. + pub soft_drop_enabled: bool, } impl DdlContext { @@ -134,7 +138,10 @@ impl DdlContext { /// /// Once the regions were dropped, subsequent heartbeats no longer include these regions. /// Therefore, we should remove the failure detectors for these dropped regions. - async fn deregister_failure_detectors(&self, detecting_regions: Vec) { + pub(crate) async fn deregister_failure_detectors( + &self, + detecting_regions: Vec, + ) { self.region_failure_detector_controller .deregister_failure_detectors(detecting_regions) .await; diff --git a/src/common/meta/src/ddl/drop_table.rs b/src/common/meta/src/ddl/drop_table.rs index 4f4e2d6053..f5b5bc7a22 100644 --- a/src/common/meta/src/ddl/drop_table.rs +++ b/src/common/meta/src/ddl/drop_table.rs @@ -36,7 +36,7 @@ use table::table_reference::TableReference; use self::executor::DropTableExecutor; use crate::ddl::DdlContext; -use crate::ddl::utils::map_to_procedure_error; +use crate::ddl::utils::{convert_region_routes_to_detecting_regions, map_to_procedure_error}; use crate::error::{self, Result}; use crate::key::table_route::TableRouteValue; use crate::lock_key::{CatalogLock, SchemaLock, TableLock}; @@ -149,6 +149,7 @@ impl DropTableProcedure { /// Broadcasts invalidate table cache instruction. async fn on_broadcast(&mut self) -> Result { self.executor.invalidate_table_cache(&self.context).await?; + self.data.state = DropTableState::DatanodeDropRegions; Ok(Status::executing(true)) @@ -172,6 +173,23 @@ impl DropTableProcedure { .await?; } + if self.context.soft_drop_enabled { + self.executor + .on_close_regions( + &self.context.node_manager, + &self.context.leader_region_registry, + &self.data.physical_region_routes, + ) + .await?; + self.context + .deregister_failure_detectors(convert_region_routes_to_detecting_regions( + &self.data.physical_region_routes, + )) + .await; + self.dropping_regions.clear(); + return Ok(Status::done()); + } + self.executor .on_drop_regions( &self.context.node_manager, @@ -182,6 +200,12 @@ impl DropTableProcedure { false, ) .await?; + self.context + .deregister_failure_detectors(convert_region_routes_to_detecting_regions( + &self.data.physical_region_routes, + )) + .await; + self.data.state = DropTableState::DeleteTombstone; Ok(Status::executing(true)) } diff --git a/src/common/meta/src/ddl/drop_table/executor.rs b/src/common/meta/src/ddl/drop_table/executor.rs index 03428aa688..1e85a64ef6 100644 --- a/src/common/meta/src/ddl/drop_table/executor.rs +++ b/src/common/meta/src/ddl/drop_table/executor.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use api::v1::region::{ CloseRequest as PbCloseRegionRequest, DropRequest as PbDropRegionRequest, RegionRequest, @@ -82,6 +82,8 @@ impl DropTableExecutor { /// Checks whether table exists. /// - Early returns if table not exists and `drop_if_exists` is `true`. /// - Throws an error if table not exists and `drop_if_exists` is `false`. + /// - Rejects dropping a recreated live table while an older tombstone still owns the same + /// fully qualified name. pub async fn on_prepare(&self, ctx: &DdlContext) -> Result> { let table_ref = self.table.table_ref(); @@ -106,6 +108,21 @@ impl DropTableExecutor { } ); + if ctx.soft_drop_enabled + && let Some(dropped_table) = ctx + .table_metadata_manager + .get_dropped_table(&self.table) + .await? + && dropped_table.table_id != self.table_id + { + return error::TableNameTombstoneConflictSnafu { + table_name: table_ref.to_string(), + existing_table_id: dropped_table.table_id, + dropping_table_id: self.table_id, + } + .fail(); + } + Ok(Control::Continue(())) } @@ -211,7 +228,17 @@ impl DropTableExecutor { Ok(()) } - /// Drops region on datanode. + /// Drops regions on datanodes. + /// + /// Arguments: + /// - `node_manager`: resolves datanode clients from peers in `region_routes`. + /// - `leader_region_registry`: tracks in-flight leader region operations. + /// - `region_routes`: table region placement; leaders receive drop requests and followers + /// receive close requests. + /// - `fast_path`: forwards to datanode drop requests to skip extra cleanup when safe. + /// - `force`: forwards to datanode drop requests to allow forced region removal. + /// - `partial_drop`: forwards to datanode drop requests for partial table/region drops. + #[allow(clippy::too_many_arguments)] pub async fn on_drop_regions( &self, node_manager: &NodeManagerRef, @@ -319,6 +346,64 @@ impl DropTableExecutor { Ok(()) } + + /// Closes all table regions on datanodes without deleting region files or metadata tombstones. + pub async fn on_close_regions( + &self, + node_manager: &NodeManagerRef, + leader_region_registry: &LeaderRegionRegistryRef, + region_routes: &[RegionRoute], + ) -> Result<()> { + let table_id = self.table_id; + let mut seen_peer_ids = HashSet::new(); + let peers = find_leaders(region_routes) + .into_iter() + .chain(find_followers(region_routes)) + .filter(|peer| seen_peer_ids.insert(peer.id)); + let mut close_region_tasks = Vec::new(); + + for datanode in peers { + let requester = node_manager.datanode(&datanode).await; + let region_ids = find_leader_regions(region_routes, &datanode) + .into_iter() + .chain(find_follower_regions(region_routes, &datanode)) + .map(|region_number| RegionId::new(table_id, region_number)); + + for region_id in region_ids { + debug!("Closing region {region_id} on Datanode {datanode:?}"); + let request = RegionRequest { + header: Some(RegionRequestHeader { + tracing_context: TracingContext::from_current_span().to_w3c(), + ..Default::default() + }), + body: Some(region_request::Body::Close(PbCloseRegionRequest { + region_id: region_id.as_u64(), + })), + }; + + let datanode = datanode.clone(); + let requester = requester.clone(); + close_region_tasks.push(async move { + if let Err(err) = requester.handle(request).await + && err.status_code() != StatusCode::RegionNotFound + { + return Err(add_peer_context_if_needed(datanode)(err)); + } + Ok(()) + }); + } + } + + join_all(close_region_tasks) + .await + .into_iter() + .collect::>>()?; + + let region_ids = operating_leader_regions(region_routes); + leader_region_registry.batch_delete(region_ids.into_iter().map(|(region_id, _)| region_id)); + + Ok(()) + } } #[cfg(test)] diff --git a/src/common/meta/src/ddl/drop_table/metadata.rs b/src/common/meta/src/ddl/drop_table/metadata.rs index c1a8a90d4e..6289b6c61a 100644 --- a/src/common/meta/src/ddl/drop_table/metadata.rs +++ b/src/common/meta/src/ddl/drop_table/metadata.rs @@ -13,12 +13,13 @@ // limitations under the License. use common_catalog::format_full_table_name; -use snafu::OptionExt; +use snafu::{OptionExt, ensure}; use store_api::metric_engine_consts::METRIC_ENGINE_NAME; use crate::ddl::drop_table::DropTableProcedure; -use crate::ddl::utils::extract_region_wal_options; +use crate::ddl::utils::{extract_region_wal_options, is_metric_engine_logical_table}; use crate::error::{self, Result}; +use crate::key::table_route::TableRouteValue; impl DropTableProcedure { /// Fetches the table info and physical table route. @@ -31,18 +32,34 @@ impl DropTableProcedure { .get_physical_table_route(task.table_id) .await?; - if physical_table_id == self.data.table_id() { - let table_info_value = self - .context - .table_metadata_manager - .table_info_manager() - .get(task.table_id) - .await? - .with_context(|| error::TableInfoNotFoundSnafu { - table: format_full_table_name(&task.catalog, &task.schema, &task.table), - })? - .into_inner(); + let table_info_value = self + .context + .table_metadata_manager + .table_info_manager() + .get(task.table_id) + .await? + .with_context(|| error::TableInfoNotFoundSnafu { + table: format_full_table_name(&task.catalog, &task.schema, &task.table), + })? + .into_inner(); + let table_route_value = TableRouteValue::new( + self.data.table_id(), + physical_table_id, + physical_table_route_value.region_routes.clone(), + ); + // TODO(hl): support soft-dropping logical tables. + ensure!( + !(self.context.soft_drop_enabled + && is_metric_engine_logical_table( + &table_info_value.table_info, + &table_route_value + )), + error::UnsupportedSnafu { + operation: "soft-dropping metric logical tables".to_string() + } + ); + if physical_table_id == self.data.table_id() { let engine = table_info_value.table_info.meta.engine; // rollback only if dropping the metric physical table fails self.data.allow_rollback = engine.as_str() == METRIC_ENGINE_NAME; diff --git a/src/common/meta/src/ddl/purge_dropped_table.rs b/src/common/meta/src/ddl/purge_dropped_table.rs new file mode 100644 index 0000000000..b95e5db976 --- /dev/null +++ b/src/common/meta/src/ddl/purge_dropped_table.rs @@ -0,0 +1,231 @@ +// 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 std::collections::HashMap; + +use async_trait::async_trait; +use common_procedure::error::{FromJsonSnafu, ToJsonSnafu}; +use common_procedure::{ + Context as ProcedureContext, LockKey, Procedure, Result as ProcedureResult, Status, +}; +use common_wal::options::WalOptions; +use serde::{Deserialize, Serialize}; +use snafu::{ResultExt, ensure}; +use store_api::storage::{RegionNumber, TableId}; +use strum::AsRefStr; +use table::metadata::TableInfo; +use table::table_name::TableName; + +use crate::ddl::DdlContext; +use crate::ddl::drop_table::executor::DropTableExecutor; +use crate::ddl::undrop_table::open_regions_ignore_region_not_found; +use crate::ddl::utils::{ + convert_region_routes_to_detecting_regions, is_metric_engine_logical_table, + map_to_procedure_error, +}; +use crate::error::{self, Result}; +use crate::key::table_route::TableRouteValue; +use crate::lock_key::TableLock; +use crate::rpc::ddl::PurgeDroppedTableTask; +use crate::rpc::router::RegionRoute; + +pub struct PurgeDroppedTableProcedure { + context: DdlContext, + data: PurgeDroppedTableData, +} + +impl PurgeDroppedTableProcedure { + pub const TYPE_NAME: &'static str = "metasrv-procedure::PurgeDroppedTable"; + + pub fn new(task: PurgeDroppedTableTask, context: DdlContext) -> Self { + Self { + context, + data: PurgeDroppedTableData::new(task), + } + } + + pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult { + let data: PurgeDroppedTableData = serde_json::from_str(json).context(FromJsonSnafu)?; + Ok(Self { context, data }) + } + + async fn on_prepare(&mut self) -> Result { + let dropped_table = self + .context + .table_metadata_manager + .get_dropped_table_by_id(self.data.task.table_id) + .await?; + + let Some(dropped_table) = dropped_table else { + return Ok(Status::done()); + }; + ensure!( + !is_metric_engine_logical_table( + &dropped_table.table_info_value.table_info, + &dropped_table.table_route_value + ), + error::UnsupportedSnafu { + operation: "purging metric logical tables".to_string() + } + ); + self.data.table_id = Some(dropped_table.table_id); + self.data.table_name = Some(dropped_table.table_name); + self.data.table_info = Some(dropped_table.table_info_value.table_info); + self.data.table_route_value = Some(dropped_table.table_route_value); + self.data.region_wal_options = dropped_table.region_wal_options; + self.data.state = PurgeDroppedTableState::OpenRegions; + Ok(Status::executing(true)) + } + + async fn on_open_regions(&mut self) -> Result { + if let Some(region_routes) = self.data.physical_region_routes() { + open_regions_ignore_region_not_found( + &self.context, + self.data.table_id(), + self.data.table_name(), + self.data.table_info(), + region_routes, + &self.data.region_wal_options, + ) + .await?; + } + self.data.state = PurgeDroppedTableState::DropRegions; + Ok(Status::executing(true)) + } + + async fn on_drop_regions(&mut self) -> Result { + if let Some(region_routes) = self.data.physical_region_routes() { + self.executor() + .on_drop_regions( + &self.context.node_manager, + &self.context.leader_region_registry, + region_routes, + false, + false, + false, + ) + .await?; + self.context + .deregister_failure_detectors(convert_region_routes_to_detecting_regions( + region_routes, + )) + .await; + } + self.data.state = PurgeDroppedTableState::DeleteTombstone; + Ok(Status::executing(true)) + } + + async fn on_delete_tombstone(&mut self) -> Result { + self.context + .table_metadata_manager + .delete_table_metadata_tombstone( + self.data.table_id(), + self.data.table_name(), + self.data.table_route_value(), + &self.data.region_wal_options, + ) + .await?; + Ok(Status::done()) + } + + fn executor(&self) -> DropTableExecutor { + DropTableExecutor::new(self.data.table_name().clone(), self.data.table_id(), false) + } +} + +#[async_trait] +impl Procedure for PurgeDroppedTableProcedure { + fn type_name(&self) -> &str { + Self::TYPE_NAME + } + + fn recover(&mut self) -> ProcedureResult<()> { + Ok(()) + } + + async fn execute(&mut self, _: &ProcedureContext) -> ProcedureResult { + match self.data.state { + PurgeDroppedTableState::Prepare => self.on_prepare().await, + PurgeDroppedTableState::OpenRegions => self.on_open_regions().await, + PurgeDroppedTableState::DropRegions => self.on_drop_regions().await, + PurgeDroppedTableState::DeleteTombstone => self.on_delete_tombstone().await, + } + .map_err(map_to_procedure_error) + } + + fn dump(&self) -> ProcedureResult { + serde_json::to_string(&self.data).context(ToJsonSnafu) + } + + fn lock_key(&self) -> LockKey { + LockKey::new(vec![TableLock::Write(self.data.task.table_id).into()]) + } +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct PurgeDroppedTableData { + state: PurgeDroppedTableState, + task: PurgeDroppedTableTask, + table_id: Option, + table_name: Option, + table_info: Option, + table_route_value: Option, + #[serde(default)] + region_wal_options: HashMap, +} + +impl PurgeDroppedTableData { + fn new(task: PurgeDroppedTableTask) -> Self { + Self { + state: PurgeDroppedTableState::Prepare, + task, + table_id: None, + table_name: None, + table_info: None, + table_route_value: None, + region_wal_options: HashMap::new(), + } + } + + fn table_id(&self) -> TableId { + self.table_id.unwrap() + } + + fn table_name(&self) -> &TableName { + self.table_name.as_ref().unwrap() + } + + fn table_info(&self) -> &TableInfo { + self.table_info.as_ref().unwrap() + } + + fn table_route_value(&self) -> &TableRouteValue { + self.table_route_value.as_ref().unwrap() + } + + fn physical_region_routes(&self) -> Option<&[RegionRoute]> { + match self.table_route_value() { + TableRouteValue::Physical(route) => Some(&route.region_routes), + TableRouteValue::Logical(_) => None, + } + } +} + +#[derive(Debug, Serialize, Deserialize, AsRefStr, PartialEq)] +enum PurgeDroppedTableState { + Prepare, + OpenRegions, + DropRegions, + DeleteTombstone, +} diff --git a/src/common/meta/src/ddl/tests/drop_table.rs b/src/common/meta/src/ddl/tests/drop_table.rs index c518bb36a2..fd41dd5126 100644 --- a/src/common/meta/src/ddl/tests/drop_table.rs +++ b/src/common/meta/src/ddl/tests/drop_table.rs @@ -12,35 +12,44 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::assert_matches; use std::collections::{HashMap, HashSet}; -use std::sync::Arc; +use std::sync::{Arc, Mutex as StdMutex}; +use api::region::RegionResponse; use api::v1::region::{RegionRequest, region_request}; +use async_trait::async_trait; use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME}; -use common_error::ext::ErrorExt; +use common_error::ext::{BoxedError, ErrorExt, StackError}; use common_error::status_code::StatusCode; use common_procedure::Procedure; use common_procedure_test::{ execute_procedure_until, execute_procedure_until_done, new_test_procedure_context, }; +use snafu::ResultExt; use store_api::region_engine::RegionRole; use store_api::storage::RegionId; use table::metadata::TableId; -use tokio::sync::mpsc; +use tokio::sync::{Mutex, mpsc}; -use crate::ddl::TableMetadata; use crate::ddl::create_logical_tables::CreateLogicalTablesProcedure; +use crate::ddl::create_table::CreateTableProcedure; use crate::ddl::drop_table::{DropTableProcedure, DropTableState}; +use crate::ddl::purge_dropped_table::PurgeDroppedTableProcedure; use crate::ddl::test_util::create_table::test_create_table_task; use crate::ddl::test_util::datanode_handler::{DatanodeWatcher, NaiveDatanodeHandler}; use crate::ddl::test_util::{ create_logical_table, create_physical_table, create_physical_table_metadata, put_datanode_address, test_create_logical_table_task, test_create_physical_table_task, }; +use crate::ddl::undrop_table::UndropTableProcedure; +use crate::ddl::{DetectingRegion, RegionFailureDetectorController, TableMetadata}; +use crate::error::{self, Error}; +use crate::key::table_name::TableNameKey; use crate::key::table_route::TableRouteValue; use crate::kv_backend::memory::MemoryKvBackend; use crate::peer::Peer; -use crate::rpc::ddl::DropTableTask; +use crate::rpc::ddl::{DropTableTask, PurgeDroppedTableTask, UndropTableTask}; use crate::rpc::router::{Region, RegionRoute}; use crate::test_util::{MockDatanodeManager, new_ddl_context, new_ddl_context_with_kv_backend}; @@ -231,6 +240,929 @@ async fn test_on_datanode_drop_regions_remaps_addresses_when_retrying() { assert_eq!(peers[1].addr, "new-follower"); } +#[tokio::test] +async fn test_soft_drop_closes_regions_and_keeps_tombstone() { + let (tx, mut rx) = mpsc::channel(8); + let datanode_handler = DatanodeWatcher::new(tx); + let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler)); + let detector_controller = Arc::new(RecordingRegionFailureDetectorController::default()); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + ddl_context.region_failure_detector_controller = detector_controller.clone(); + let table_id = 1024; + let table_name = "foo"; + let task = test_create_table_task(table_name, table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::physical(vec![ + RegionRoute { + region: Region::new_test(RegionId::new(table_id, 1)), + leader_peer: Some(Peer::empty(1)), + follower_peers: vec![Peer::empty(2)], + leader_state: None, + leader_down_since: None, + write_route_policy: None, + }, + RegionRoute { + region: Region::new_test(RegionId::new(table_id, 2)), + leader_peer: Some(Peer::empty(2)), + follower_peers: vec![Peer::empty(1)], + leader_state: None, + leader_down_since: None, + write_route_policy: None, + }, + ]), + HashMap::new(), + ) + .await + .unwrap(); + + let task = new_drop_table_task(table_name, table_id, false); + let mut procedure = DropTableProcedure::new(task, ddl_context.clone()); + + execute_procedure_until_done(&mut procedure).await; + + assert!(procedure.dropping_regions.is_empty()); + assert_eq!(ddl_context.memory_region_keeper.len(), 0); + + let mut requests = Vec::new(); + for _ in 0..4 { + let (peer, request) = rx.try_recv().unwrap(); + let Some(region_request::Body::Close(req)) = request.body else { + unreachable!(); + }; + requests.push((peer.id, req.region_id)); + } + requests.sort_unstable(); + assert_eq!( + requests, + vec![ + (1, RegionId::new(table_id, 1).as_u64()), + (1, RegionId::new(table_id, 2).as_u64()), + (2, RegionId::new(table_id, 1).as_u64()), + (2, RegionId::new(table_id, 2).as_u64()), + ] + ); + assert!(rx.try_recv().is_err()); + + let table_name = procedure.data.task.table_name(); + let live_table = ddl_context + .table_metadata_manager + .table_name_manager() + .get(TableNameKey::from(&table_name)) + .await + .unwrap(); + assert!(live_table.is_none()); + + let dropped_table = ddl_context + .table_metadata_manager + .get_dropped_table(&table_name) + .await + .unwrap(); + assert!(dropped_table.is_some()); + + assert_eq!( + detector_controller.deregistered().await, + vec![ + (1, RegionId::new(table_id, 1)), + (2, RegionId::new(table_id, 2)) + ] + ); +} + +#[tokio::test] +async fn test_hard_drop_keeps_delete_tombstone_flow() { + let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); + let detector_controller = Arc::new(RecordingRegionFailureDetectorController::default()); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.region_failure_detector_controller = detector_controller.clone(); + let table_id = 1024; + let table_name = "foo"; + let task = test_create_table_task(table_name, table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::physical(vec![RegionRoute { + region: Region::new_test(RegionId::new(table_id, 1)), + leader_peer: Some(Peer::empty(1)), + follower_peers: vec![], + leader_state: None, + leader_down_since: None, + write_route_policy: None, + }]), + HashMap::new(), + ) + .await + .unwrap(); + + let task = new_drop_table_task(table_name, table_id, false); + let mut procedure = DropTableProcedure::new(task, ddl_context.clone()); + + execute_procedure_until(&mut procedure, |p| { + p.data.state == DropTableState::DeleteTombstone + }) + .await; + + assert_eq!(procedure.data.state, DropTableState::DeleteTombstone); + + execute_procedure_until_done(&mut procedure).await; + + let dropped_table = ddl_context + .table_metadata_manager + .get_dropped_table(&procedure.data.task.table_name()) + .await + .unwrap(); + assert!(dropped_table.is_none()); + assert_eq!( + detector_controller.deregistered().await, + vec![(1, RegionId::new(table_id, 1))] + ); +} + +#[tokio::test] +async fn test_create_table_succeeds_while_tombstone_exists() { + let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + let dropped_table_id = 1024; + let table_name = "foo"; + let task = test_create_table_task(table_name, dropped_table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::physical(vec![]), + HashMap::new(), + ) + .await + .unwrap(); + + let drop_task = new_drop_table_task(table_name, dropped_table_id, false); + let mut drop_procedure = DropTableProcedure::new(drop_task, ddl_context.clone()); + execute_procedure_until_done(&mut drop_procedure).await; + + let mut create_task = test_create_table_task(table_name, 1025); + create_task.create_table.table_id = None; + create_task.table_info.ident.table_id = 0; + let mut create_procedure = CreateTableProcedure::new(create_task, ddl_context.clone()).unwrap(); + execute_procedure_until_done(&mut create_procedure).await; + + let live_table = ddl_context + .table_metadata_manager + .table_name_manager() + .get(TableNameKey::new( + DEFAULT_CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + table_name, + )) + .await + .unwrap() + .unwrap(); + assert_eq!(live_table.table_id(), create_procedure.table_id()); + + let dropped_table = ddl_context + .table_metadata_manager + .get_dropped_table(&create_procedure.data.task.table_name()) + .await + .unwrap(); + assert_eq!(dropped_table.unwrap().table_id, dropped_table_id); +} + +#[tokio::test] +async fn test_drop_recreated_table_fails_when_previous_tombstone_exists() { + let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + let original_table_id = 1024; + let recreated_table_id = 1025; + let table_name = "foo"; + let task = test_create_table_task(table_name, original_table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::physical(vec![]), + HashMap::new(), + ) + .await + .unwrap(); + + let drop_task = new_drop_table_task(table_name, original_table_id, false); + let mut drop_procedure = DropTableProcedure::new(drop_task, ddl_context.clone()); + execute_procedure_until_done(&mut drop_procedure).await; + + ddl_context + .table_metadata_manager + .create_table_metadata( + test_create_table_task(table_name, recreated_table_id).table_info, + TableRouteValue::physical(vec![]), + HashMap::new(), + ) + .await + .unwrap(); + + let mut procedure = DropTableProcedure::new( + new_drop_table_task(table_name, recreated_table_id, false), + ddl_context, + ); + let err = procedure.on_prepare().await.unwrap_err(); + + assert_matches!(err, Error::TableNameTombstoneConflict { .. }); +} + +#[tokio::test] +async fn test_hard_drop_recreated_table_ignores_previous_orphan_tombstone() { + let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); + let ddl_context = new_ddl_context(node_manager); + let original_table_id = 1024; + let recreated_table_id = 1025; + let table_name = "foo"; + let task = test_create_table_task(table_name, original_table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::physical(vec![]), + HashMap::new(), + ) + .await + .unwrap(); + + let drop_task = new_drop_table_task(table_name, original_table_id, false); + let mut drop_procedure = DropTableProcedure::new(drop_task, ddl_context.clone()); + execute_procedure_until(&mut drop_procedure, |p| { + p.data.state == DropTableState::DeleteTombstone + }) + .await; + + ddl_context + .table_metadata_manager + .create_table_metadata( + test_create_table_task(table_name, recreated_table_id).table_info, + TableRouteValue::physical(vec![]), + HashMap::new(), + ) + .await + .unwrap(); + + let mut procedure = DropTableProcedure::new( + new_drop_table_task(table_name, recreated_table_id, false), + ddl_context, + ); + + procedure.on_prepare().await.unwrap(); +} + +#[tokio::test] +async fn test_undrop_table_restores_metadata_and_reopens_regions() { + let (tx, mut rx) = mpsc::channel(8); + let datanode_handler = DatanodeWatcher::new(tx); + let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler)); + let detector_controller = Arc::new(RecordingRegionFailureDetectorController::default()); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + ddl_context.region_failure_detector_controller = detector_controller.clone(); + let table_id = 1024; + let table_name = "foo"; + let task = test_create_table_task(table_name, table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::physical(vec![RegionRoute { + region: Region::new_test(RegionId::new(table_id, 1)), + leader_peer: Some(Peer::empty(1)), + follower_peers: vec![Peer::empty(2)], + leader_state: None, + leader_down_since: None, + write_route_policy: None, + }]), + HashMap::new(), + ) + .await + .unwrap(); + + let mut drop_procedure = DropTableProcedure::new( + new_drop_table_task(table_name, table_id, false), + ddl_context.clone(), + ); + execute_procedure_until_done(&mut drop_procedure).await; + detector_controller.clear().await; + + while rx.try_recv().is_ok() {} + + let mut procedure = + UndropTableProcedure::new(new_undrop_table_task(table_id), ddl_context.clone()); + execute_procedure_until_done(&mut procedure).await; + + let live_table = ddl_context + .table_metadata_manager + .table_name_manager() + .get(TableNameKey::new( + DEFAULT_CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + table_name, + )) + .await + .unwrap() + .unwrap(); + assert_eq!(live_table.table_id(), table_id); + assert!( + ddl_context + .table_metadata_manager + .get_dropped_table(&drop_procedure.data.task.table_name()) + .await + .unwrap() + .is_none() + ); + + let mut opened_regions = HashSet::new(); + for _ in 0..2 { + let (peer, request) = rx.try_recv().unwrap(); + let Some(region_request::Body::Open(req)) = request.body else { + unreachable!(); + }; + opened_regions.insert((peer.id, req.region_id)); + } + assert_eq!( + opened_regions, + HashSet::from([ + (1, RegionId::new(table_id, 1).as_u64()), + (2, RegionId::new(table_id, 1).as_u64()), + ]) + ); + assert!(rx.try_recv().is_err()); + + assert_eq!( + detector_controller.registered().await, + vec![(1, RegionId::new(table_id, 1))] + ); +} + +#[tokio::test] +async fn test_undrop_table_opens_regions_before_restoring_live_metadata() { + let (tx, mut rx) = mpsc::channel(8); + let datanode_handler = DatanodeWatcher::new(tx); + let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler)); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + let table_id = 1024; + let table_name = "foo"; + let task = test_create_table_task(table_name, table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::physical(vec![RegionRoute { + region: Region::new_test(RegionId::new(table_id, 1)), + leader_peer: Some(Peer::empty(1)), + follower_peers: vec![], + leader_state: None, + leader_down_since: None, + write_route_policy: None, + }]), + HashMap::new(), + ) + .await + .unwrap(); + let mut drop_procedure = DropTableProcedure::new( + new_drop_table_task(table_name, table_id, false), + ddl_context.clone(), + ); + execute_procedure_until_done(&mut drop_procedure).await; + while rx.try_recv().is_ok() {} + + let mut procedure = + UndropTableProcedure::new(new_undrop_table_task(table_id), ddl_context.clone()); + let ctx = new_test_procedure_context(); + procedure.execute(&ctx).await.unwrap(); + procedure.execute(&ctx).await.unwrap(); + + let (_, request) = rx.try_recv().unwrap(); + assert_matches!(request.body, Some(region_request::Body::Open(_))); + assert!(rx.try_recv().is_err()); + assert!( + ddl_context + .table_metadata_manager + .table_name_manager() + .get(TableNameKey::new( + DEFAULT_CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + table_name, + )) + .await + .unwrap() + .is_none() + ); +} + +#[tokio::test] +async fn test_undrop_logical_table_skips_datanode_open() { + let (tx, mut rx) = mpsc::channel(8); + let datanode_handler = DatanodeWatcher::new(tx); + let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler)); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + let physical_table_id = 1024; + let logical_table_id = 1025; + let table_name = "foo"; + ddl_context + .table_metadata_manager + .create_table_metadata( + test_create_table_task("phy", physical_table_id).table_info, + TableRouteValue::physical(vec![RegionRoute { + region: Region::new_test(RegionId::new(physical_table_id, 1)), + leader_peer: Some(Peer::empty(1)), + follower_peers: vec![], + leader_state: None, + leader_down_since: None, + write_route_policy: None, + }]), + HashMap::new(), + ) + .await + .unwrap(); + let task = test_create_table_task(table_name, logical_table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::logical(physical_table_id), + HashMap::new(), + ) + .await + .unwrap(); + + let mut drop_procedure = DropTableProcedure::new( + new_drop_table_task(table_name, logical_table_id, false), + ddl_context.clone(), + ); + execute_procedure_until_done(&mut drop_procedure).await; + + while rx.try_recv().is_ok() {} + + let mut procedure = + UndropTableProcedure::new(new_undrop_table_task(logical_table_id), ddl_context.clone()); + execute_procedure_until_done(&mut procedure).await; + + let live_table = ddl_context + .table_metadata_manager + .table_name_manager() + .get(TableNameKey::new( + DEFAULT_CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + table_name, + )) + .await + .unwrap() + .unwrap(); + assert_eq!(live_table.table_id(), logical_table_id); + assert!(rx.try_recv().is_err()); +} + +#[tokio::test] +async fn test_soft_drop_metric_logical_table_fails() { + let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + let physical_table_id = create_physical_table(&ddl_context, "phy").await; + let logical_table_id = + create_logical_table(ddl_context.clone(), physical_table_id, "foo").await; + + let mut procedure = DropTableProcedure::new( + new_drop_table_task("foo", logical_table_id, false), + ddl_context, + ); + let err = procedure.on_prepare().await.unwrap_err(); + + assert_eq!(err.status_code(), StatusCode::Unsupported); +} + +#[tokio::test] +async fn test_undrop_metric_logical_table_fails() { + let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); + let ddl_context = new_ddl_context(node_manager); + let physical_table_id = create_physical_table(&ddl_context, "phy").await; + let logical_table_id = + create_metric_logical_table_tombstone(&ddl_context, physical_table_id, "foo").await; + + let mut procedure = + UndropTableProcedure::new(new_undrop_table_task(logical_table_id), ddl_context); + let err = procedure.on_prepare().await.unwrap_err(); + + assert_eq!(err.status_code(), StatusCode::Unsupported); +} + +#[tokio::test] +async fn test_purge_metric_logical_table_fails() { + let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); + let ddl_context = new_ddl_context(node_manager); + let physical_table_id = create_physical_table(&ddl_context, "phy").await; + let logical_table_id = + create_metric_logical_table_tombstone(&ddl_context, physical_table_id, "foo").await; + + let mut procedure = PurgeDroppedTableProcedure::new( + new_purge_dropped_table_task(logical_table_id), + ddl_context, + ); + let err = procedure + .execute(&new_test_procedure_context()) + .await + .unwrap_err(); + + assert_eq!(err.status_code(), StatusCode::Unsupported); +} + +#[tokio::test] +async fn test_undrop_table_fails_when_live_name_exists() { + let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + let dropped_table_id = 1024; + let live_table_id = 1025; + let table_name = "foo"; + let task = test_create_table_task(table_name, dropped_table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::physical(vec![]), + HashMap::new(), + ) + .await + .unwrap(); + let mut drop_procedure = DropTableProcedure::new( + new_drop_table_task(table_name, dropped_table_id, false), + ddl_context.clone(), + ); + execute_procedure_until_done(&mut drop_procedure).await; + ddl_context + .table_metadata_manager + .create_table_metadata( + test_create_table_task(table_name, live_table_id).table_info, + TableRouteValue::physical(vec![]), + HashMap::new(), + ) + .await + .unwrap(); + + let mut procedure = + UndropTableProcedure::new(new_undrop_table_task(dropped_table_id), ddl_context); + let err = procedure.on_prepare().await.unwrap_err(); + + assert_matches!(err, Error::TableAlreadyExists { .. }); +} + +#[tokio::test] +async fn test_undrop_table_fails_when_live_name_is_created_after_prepare() { + let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + let dropped_table_id = 1024; + let live_table_id = 1025; + let table_name = "foo"; + let task = test_create_table_task(table_name, dropped_table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::physical(vec![]), + HashMap::new(), + ) + .await + .unwrap(); + let mut drop_procedure = DropTableProcedure::new( + new_drop_table_task(table_name, dropped_table_id, false), + ddl_context.clone(), + ); + execute_procedure_until_done(&mut drop_procedure).await; + + let mut procedure = + UndropTableProcedure::new(new_undrop_table_task(dropped_table_id), ddl_context.clone()); + procedure.on_prepare().await.unwrap(); + ddl_context + .table_metadata_manager + .create_table_metadata( + test_create_table_task(table_name, live_table_id).table_info, + TableRouteValue::physical(vec![]), + HashMap::new(), + ) + .await + .unwrap(); + + let err = procedure + .execute(&new_test_procedure_context()) + .await + .unwrap_err(); + assert_eq!(err.status_code(), StatusCode::TableAlreadyExists); + let live_table = ddl_context + .table_metadata_manager + .table_name_manager() + .get(TableNameKey::new( + DEFAULT_CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + table_name, + )) + .await + .unwrap() + .unwrap(); + assert_eq!(live_table.table_id(), live_table_id); +} + +#[tokio::test] +async fn test_undrop_table_replayed_restore_metadata_is_idempotent() { + let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + let table_id = 1024; + let table_name = "foo"; + let task = test_create_table_task(table_name, table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::physical(vec![]), + HashMap::new(), + ) + .await + .unwrap(); + let mut drop_procedure = DropTableProcedure::new( + new_drop_table_task(table_name, table_id, false), + ddl_context.clone(), + ); + execute_procedure_until_done(&mut drop_procedure).await; + + let mut procedure = + UndropTableProcedure::new(new_undrop_table_task(table_id), ddl_context.clone()); + let ctx = new_test_procedure_context(); + procedure.execute(&ctx).await.unwrap(); + let restore_metadata_data = procedure.dump().unwrap(); + procedure.execute(&ctx).await.unwrap(); + + let mut replayed = + UndropTableProcedure::from_json(&restore_metadata_data, ddl_context.clone()).unwrap(); + execute_procedure_until_done(&mut replayed).await; + + let live_table = ddl_context + .table_metadata_manager + .table_name_manager() + .get(TableNameKey::new( + DEFAULT_CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + table_name, + )) + .await + .unwrap() + .unwrap(); + assert_eq!(live_table.table_id(), table_id); +} + +#[tokio::test] +async fn test_purge_dropped_table_drops_regions_and_deletes_tombstone() { + let (tx, mut rx) = mpsc::channel(8); + let datanode_handler = DatanodeWatcher::new(tx); + let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler)); + let detector_controller = Arc::new(RecordingRegionFailureDetectorController::default()); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + ddl_context.region_failure_detector_controller = detector_controller.clone(); + let table_id = 1024; + let table_name = "foo"; + let task = test_create_table_task(table_name, table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::physical(vec![RegionRoute { + region: Region::new_test(RegionId::new(table_id, 1)), + leader_peer: Some(Peer::empty(1)), + follower_peers: vec![Peer::empty(2)], + leader_state: None, + leader_down_since: None, + write_route_policy: None, + }]), + HashMap::new(), + ) + .await + .unwrap(); + let mut drop_procedure = DropTableProcedure::new( + new_drop_table_task(table_name, table_id, false), + ddl_context.clone(), + ); + execute_procedure_until_done(&mut drop_procedure).await; + while rx.try_recv().is_ok() {} + detector_controller.clear().await; + + let mut procedure = PurgeDroppedTableProcedure::new( + new_purge_dropped_table_task(table_id), + ddl_context.clone(), + ); + execute_procedure_until_done(&mut procedure).await; + + let mut requests = Vec::new(); + for _ in 0..4 { + let (peer, request) = rx.try_recv().unwrap(); + requests.push((peer.id, request.body.unwrap())); + } + requests.sort_unstable_by_key(|(peer_id, _)| *peer_id); + assert_matches!(requests[0].1, region_request::Body::Open(_)); + assert_matches!(requests[1].1, region_request::Body::Drop(_)); + assert_matches!(requests[2].1, region_request::Body::Open(_)); + assert_matches!(requests[3].1, region_request::Body::Close(_)); + assert!(rx.try_recv().is_err()); + assert!( + ddl_context + .table_metadata_manager + .get_dropped_table(&drop_procedure.data.task.table_name()) + .await + .unwrap() + .is_none() + ); + + assert_eq!( + detector_controller.deregistered().await, + vec![(1, RegionId::new(table_id, 1))] + ); +} + +#[tokio::test] +async fn test_purge_dropped_table_by_id_selects_tombstone_when_live_table_exists() { + let (tx, mut rx) = mpsc::channel(8); + let datanode_handler = DatanodeWatcher::new(tx); + let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler)); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + let dropped_table_id = 1024; + let live_table_id = 1025; + let table_name = "foo"; + ddl_context + .table_metadata_manager + .create_table_metadata( + test_create_table_task(table_name, dropped_table_id).table_info, + TableRouteValue::physical(vec![RegionRoute { + region: Region::new_test(RegionId::new(dropped_table_id, 1)), + leader_peer: Some(Peer::empty(1)), + follower_peers: vec![], + leader_state: None, + leader_down_since: None, + write_route_policy: None, + }]), + HashMap::new(), + ) + .await + .unwrap(); + let mut drop_procedure = DropTableProcedure::new( + new_drop_table_task(table_name, dropped_table_id, false), + ddl_context.clone(), + ); + execute_procedure_until_done(&mut drop_procedure).await; + ddl_context + .table_metadata_manager + .create_table_metadata( + test_create_table_task(table_name, live_table_id).table_info, + TableRouteValue::physical(vec![]), + HashMap::new(), + ) + .await + .unwrap(); + while rx.try_recv().is_ok() {} + + let mut procedure = PurgeDroppedTableProcedure::new( + new_purge_dropped_table_task(dropped_table_id), + ddl_context.clone(), + ); + execute_procedure_until_done(&mut procedure).await; + + let live_table = ddl_context + .table_metadata_manager + .table_name_manager() + .get(TableNameKey::new( + DEFAULT_CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + table_name, + )) + .await + .unwrap() + .unwrap(); + assert_eq!(live_table.table_id(), live_table_id); + + let (_, request) = rx.try_recv().unwrap(); + let Some(region_request::Body::Open(req)) = request.body else { + unreachable!(); + }; + assert_eq!(req.region_id, RegionId::new(dropped_table_id, 1).as_u64()); + + let (_, request) = rx.try_recv().unwrap(); + let Some(region_request::Body::Drop(req)) = request.body else { + unreachable!(); + }; + assert_eq!(req.region_id, RegionId::new(dropped_table_id, 1).as_u64()); + assert!(rx.try_recv().is_err()); +} + +#[tokio::test] +async fn test_purge_dropped_table_replayed_open_regions_ignores_dropped_regions() { + let (tx, mut rx) = mpsc::channel(8); + let dropped_regions = Arc::new(StdMutex::new(HashSet::new())); + let datanode_handler = DatanodeWatcher::new(tx).with_handler({ + let dropped_regions = dropped_regions.clone(); + move |_peer, request| { + let Some(body) = request.body.as_ref() else { + return Ok(RegionResponse::new(0)); + }; + match body { + region_request::Body::Open(req) + if dropped_regions.lock().unwrap().contains(&req.region_id) => + { + Err::(BoxedError::new(MockRegionNotFoundError)) + .context(error::ExternalSnafu) + } + region_request::Body::Drop(req) => { + dropped_regions.lock().unwrap().insert(req.region_id); + Ok(RegionResponse::new(0)) + } + _ => Ok(RegionResponse::new(0)), + } + } + }); + let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler)); + let mut ddl_context = new_ddl_context(node_manager); + ddl_context.soft_drop_enabled = true; + let table_id = 1024; + let table_name = "foo"; + let task = test_create_table_task(table_name, table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::physical(vec![RegionRoute { + region: Region::new_test(RegionId::new(table_id, 1)), + leader_peer: Some(Peer::empty(1)), + follower_peers: vec![], + leader_state: None, + leader_down_since: None, + write_route_policy: None, + }]), + HashMap::new(), + ) + .await + .unwrap(); + let mut drop_procedure = DropTableProcedure::new( + new_drop_table_task(table_name, table_id, false), + ddl_context.clone(), + ); + execute_procedure_until_done(&mut drop_procedure).await; + while rx.try_recv().is_ok() {} + + let mut procedure = PurgeDroppedTableProcedure::new( + new_purge_dropped_table_task(table_id), + ddl_context.clone(), + ); + let ctx = new_test_procedure_context(); + procedure.execute(&ctx).await.unwrap(); + let open_regions_data = procedure.dump().unwrap(); + procedure.execute(&ctx).await.unwrap(); + procedure.execute(&ctx).await.unwrap(); + + let mut replayed = + PurgeDroppedTableProcedure::from_json(&open_regions_data, ddl_context.clone()).unwrap(); + execute_procedure_until_done(&mut replayed).await; + + assert!( + ddl_context + .table_metadata_manager + .get_dropped_table(&drop_procedure.data.task.table_name()) + .await + .unwrap() + .is_none() + ); +} + +#[derive(Debug, snafu::Snafu)] +#[snafu(display("mock region not found"))] +struct MockRegionNotFoundError; + +impl StackError for MockRegionNotFoundError { + fn debug_fmt(&self, _: usize, _: &mut Vec) {} + + fn next(&self) -> Option<&dyn StackError> { + None + } +} + +impl ErrorExt for MockRegionNotFoundError { + fn as_any(&self) -> &dyn std::any::Any { + self + } + + fn status_code(&self) -> StatusCode { + StatusCode::RegionNotFound + } +} + #[tokio::test] async fn test_on_rollback() { let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); @@ -302,6 +1234,72 @@ fn new_drop_table_task(table_name: &str, table_id: TableId, drop_if_exists: bool } } +fn new_undrop_table_task(table_id: TableId) -> UndropTableTask { + UndropTableTask { table_id } +} + +fn new_purge_dropped_table_task(table_id: TableId) -> PurgeDroppedTableTask { + PurgeDroppedTableTask { table_id } +} + +async fn create_metric_logical_table_tombstone( + ddl_context: &crate::ddl::DdlContext, + physical_table_id: TableId, + table_name: &str, +) -> TableId { + let logical_table_id = + create_logical_table(ddl_context.clone(), physical_table_id, table_name).await; + let mut task = test_create_logical_table_task(table_name); + task.set_table_id(logical_table_id); + ddl_context + .table_metadata_manager + .delete_table_metadata( + logical_table_id, + &task.table_name(), + &TableRouteValue::logical(physical_table_id), + &HashMap::new(), + ) + .await + .unwrap(); + logical_table_id +} + +#[derive(Default)] +struct RecordingRegionFailureDetectorController { + registered: Mutex>, + deregistered: Mutex>, +} + +impl RecordingRegionFailureDetectorController { + async fn registered(&self) -> Vec { + self.registered.lock().await.clone() + } + + async fn deregistered(&self) -> Vec { + self.deregistered.lock().await.clone() + } + + async fn clear(&self) { + self.registered.lock().await.clear(); + self.deregistered.lock().await.clear(); + } +} + +#[async_trait] +impl RegionFailureDetectorController for RecordingRegionFailureDetectorController { + async fn register_failure_detectors(&self, detecting_regions: Vec) { + self.registered.lock().await.extend(detecting_regions); + } + + async fn reset_failure_detectors(&self, detecting_regions: Vec) { + self.registered.lock().await.extend(detecting_regions); + } + + async fn deregister_failure_detectors(&self, detecting_regions: Vec) { + self.deregistered.lock().await.extend(detecting_regions); + } +} + #[tokio::test] async fn test_memory_region_keeper_guard_dropped_on_procedure_done() { let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); diff --git a/src/common/meta/src/ddl/undrop_table.rs b/src/common/meta/src/ddl/undrop_table.rs new file mode 100644 index 0000000000..f08d606a82 --- /dev/null +++ b/src/common/meta/src/ddl/undrop_table.rs @@ -0,0 +1,368 @@ +// 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 std::collections::{HashMap, HashSet}; + +use api::v1::region::{ + OpenRequest as PbOpenRegionRequest, RegionRequest, RegionRequestHeader, region_request, +}; +use async_trait::async_trait; +use common_error::ext::ErrorExt; +use common_error::status_code::StatusCode; +use common_procedure::error::{FromJsonSnafu, ToJsonSnafu}; +use common_procedure::{ + Context as ProcedureContext, LockKey, Procedure, Result as ProcedureResult, Status, +}; +use common_telemetry::tracing_context::TracingContext; +use common_wal::options::WalOptions; +use futures::future::join_all; +use serde::{Deserialize, Serialize}; +use snafu::{OptionExt, ResultExt, ensure}; +use store_api::storage::{RegionId, RegionNumber}; +use strum::AsRefStr; +use table::metadata::TableId; +use table::table_name::TableName; + +use crate::ddl::utils::{ + add_peer_context_if_needed, convert_region_routes_to_detecting_regions, + is_metric_engine_logical_table, map_to_procedure_error, region_storage_path, +}; +use crate::ddl::{CreateRequestBuilder, DdlContext, build_template_from_raw_table_info}; +use crate::error::{self, Result}; +use crate::instruction::CacheIdent; +use crate::key::table_name::TableNameKey; +use crate::key::table_route::TableRouteValue; +use crate::lock_key::TableLock; +use crate::rpc::ddl::UndropTableTask; +use crate::rpc::router::{ + RegionRoute, find_follower_regions, find_followers, find_leader_regions, find_leaders, +}; + +pub struct UndropTableProcedure { + context: DdlContext, + data: UndropTableData, +} + +impl UndropTableProcedure { + pub const TYPE_NAME: &'static str = "metasrv-procedure::UndropTable"; + + pub fn new(task: UndropTableTask, context: DdlContext) -> Self { + Self { + context, + data: UndropTableData::new(task), + } + } + + pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult { + let data: UndropTableData = serde_json::from_str(json).context(FromJsonSnafu)?; + Ok(Self { context, data }) + } + + pub(crate) async fn on_prepare(&mut self) -> Result { + let dropped_table = self + .context + .table_metadata_manager + .get_dropped_table_by_id(self.data.task.table_id) + .await? + .with_context(|| error::TableNotFoundSnafu { + table_name: self.data.task.table_id.to_string(), + })?; + let table_name = &dropped_table.table_name; + ensure!( + !self + .context + .table_metadata_manager + .table_name_manager() + .exists(TableNameKey::from(table_name)) + .await?, + error::TableAlreadyExistsSnafu { + table_name: table_name.to_string() + } + ); + self.data.table_name = Some(dropped_table.table_name.clone()); + self.data.table_route_value = Some(dropped_table.table_route_value.clone()); + self.data.region_wal_options = dropped_table.region_wal_options; + ensure!( + !is_metric_engine_logical_table( + &dropped_table.table_info_value.table_info, + self.data.table_route_value() + ), + error::UnsupportedSnafu { + operation: "undropping metric logical tables".to_string() + } + ); + self.data.table_info = Some(dropped_table.table_info_value.table_info); + self.data.state = UndropTableState::OpenRegions; + Ok(Status::executing(true)) + } + + async fn on_restore_metadata(&mut self) -> Result { + let table_route_value = self.data.table_route_value(); + self.context + .table_metadata_manager + .restore_table_metadata( + self.data.task.table_id, + self.data.table_name(), + table_route_value, + &self.data.region_wal_options, + ) + .await + .map_err(|err| match err { + error::Error::TombstoneTargetAlreadyExists { .. } => { + error::TableAlreadyExistsSnafu { + table_name: self.data.table_name().to_string(), + } + .build() + } + err => err, + })?; + self.data.state = UndropTableState::InvalidateTableCache; + Ok(Status::executing(true)) + } + + async fn on_open_regions(&mut self) -> Result { + self.ensure_live_table_not_exists().await?; + let TableRouteValue::Physical(route) = self.data.table_route_value() else { + self.data.state = UndropTableState::RestoreMetadata; + return Ok(Status::executing(true)); + }; + + open_regions( + &self.context, + self.data.task.table_id, + self.data.table_name(), + self.data.table_info(), + &route.region_routes, + &self.data.region_wal_options, + ) + .await?; + self.data.state = UndropTableState::RestoreMetadata; + Ok(Status::executing(true)) + } + + async fn ensure_live_table_not_exists(&self) -> Result<()> { + ensure!( + !self + .context + .table_metadata_manager + .table_name_manager() + .exists(TableNameKey::from(self.data.table_name())) + .await?, + error::TableAlreadyExistsSnafu { + table_name: self.data.table_name().to_string() + } + ); + Ok(()) + } + + async fn on_broadcast(&mut self) -> Result { + let ctx = crate::cache_invalidator::Context { + subject: Some(format!( + "Invalidate table cache by undropping table {}, table_id: {}", + self.data.table_name().table_ref(), + self.data.task.table_id, + )), + }; + self.context + .cache_invalidator + .invalidate( + &ctx, + &[ + CacheIdent::TableName(self.data.table_name().table_ref().into()), + CacheIdent::TableId(self.data.task.table_id), + ], + ) + .await?; + Ok(Status::done()) + } +} + +#[async_trait] +impl Procedure for UndropTableProcedure { + fn type_name(&self) -> &str { + Self::TYPE_NAME + } + + fn recover(&mut self) -> ProcedureResult<()> { + Ok(()) + } + + async fn execute(&mut self, _: &ProcedureContext) -> ProcedureResult { + match self.data.state { + UndropTableState::Prepare => self.on_prepare().await, + UndropTableState::RestoreMetadata => self.on_restore_metadata().await, + UndropTableState::OpenRegions => self.on_open_regions().await, + UndropTableState::InvalidateTableCache => self.on_broadcast().await, + } + .map_err(map_to_procedure_error) + } + + fn dump(&self) -> ProcedureResult { + serde_json::to_string(&self.data).context(ToJsonSnafu) + } + + fn lock_key(&self) -> LockKey { + LockKey::new(vec![TableLock::Write(self.data.task.table_id).into()]) + } +} + +pub(crate) async fn open_regions( + context: &DdlContext, + table_id: TableId, + table_name: &TableName, + table_info: &table::metadata::TableInfo, + region_routes: &[RegionRoute], + region_wal_options: &HashMap, +) -> Result<()> { + open_regions_inner( + context, + table_id, + table_name, + table_info, + region_routes, + region_wal_options, + false, + ) + .await +} + +pub(crate) async fn open_regions_ignore_region_not_found( + context: &DdlContext, + table_id: TableId, + table_name: &TableName, + table_info: &table::metadata::TableInfo, + region_routes: &[RegionRoute], + region_wal_options: &HashMap, +) -> Result<()> { + open_regions_inner( + context, + table_id, + table_name, + table_info, + region_routes, + region_wal_options, + true, + ) + .await +} + +async fn open_regions_inner( + context: &DdlContext, + table_id: TableId, + table_name: &TableName, + table_info: &table::metadata::TableInfo, + region_routes: &[RegionRoute], + region_wal_options: &HashMap, + ignore_region_not_found: bool, +) -> Result<()> { + let template = build_template_from_raw_table_info(table_info)?; + let builder = CreateRequestBuilder::new(template, None); + let storage_path = region_storage_path(&table_name.catalog_name, &table_name.schema_name); + let mut seen_peer_ids = HashSet::new(); + let peers = find_leaders(region_routes) + .into_iter() + .chain(find_followers(region_routes)) + .filter(|peer| seen_peer_ids.insert(peer.id)); + let mut tasks = Vec::new(); + for datanode in peers { + let requester = context.node_manager.datanode(&datanode).await; + let region_numbers = find_leader_regions(region_routes, &datanode) + .into_iter() + .chain(find_follower_regions(region_routes, &datanode)); + for region_number in region_numbers { + let region_id = RegionId::new(table_id, region_number); + let create_request = builder.build_one( + region_id, + storage_path.clone(), + region_wal_options, + &HashMap::new(), + )?; + let request = RegionRequest { + header: Some(RegionRequestHeader { + tracing_context: TracingContext::from_current_span().to_w3c(), + ..Default::default() + }), + body: Some(region_request::Body::Open(PbOpenRegionRequest { + region_id: create_request.region_id, + engine: create_request.engine, + path: create_request.path, + options: create_request.options, + })), + }; + let datanode = datanode.clone(); + let requester = requester.clone(); + tasks.push(async move { + if let Err(err) = requester.handle(request).await + && !(ignore_region_not_found && err.status_code() == StatusCode::RegionNotFound) + { + return Err(add_peer_context_if_needed(datanode)(err)); + } + Ok(()) + }); + } + } + + join_all(tasks) + .await + .into_iter() + .collect::>>()?; + context + .register_failure_detectors(convert_region_routes_to_detecting_regions(region_routes)) + .await; + Ok(()) +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct UndropTableData { + state: UndropTableState, + task: UndropTableTask, + table_name: Option, + table_info: Option, + table_route_value: Option, + #[serde(default)] + region_wal_options: HashMap, +} + +impl UndropTableData { + fn new(task: UndropTableTask) -> Self { + Self { + state: UndropTableState::Prepare, + task, + table_name: None, + table_info: None, + table_route_value: None, + region_wal_options: HashMap::new(), + } + } + + fn table_name(&self) -> &TableName { + self.table_name.as_ref().unwrap() + } + + fn table_info(&self) -> &table::metadata::TableInfo { + self.table_info.as_ref().unwrap() + } + + fn table_route_value(&self) -> &TableRouteValue { + self.table_route_value.as_ref().unwrap() + } +} + +#[derive(Debug, Serialize, Deserialize, AsRefStr, PartialEq)] +enum UndropTableState { + Prepare, + RestoreMetadata, + OpenRegions, + InvalidateTableCache, +} diff --git a/src/common/meta/src/ddl/utils.rs b/src/common/meta/src/ddl/utils.rs index e90f224c99..e953ee064d 100644 --- a/src/common/meta/src/ddl/utils.rs +++ b/src/common/meta/src/ddl/utils.rs @@ -38,7 +38,7 @@ use store_api::metadata::ColumnMetadata; use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, MANIFEST_INFO_EXTENSION_KEY}; use store_api::region_engine::RegionManifestInfo; use store_api::storage::RegionId; -use table::metadata::TableId; +use table::metadata::{TableId, TableInfo}; use table::table_reference::TableReference; use crate::ddl::{DdlContext, DetectingRegion}; @@ -68,6 +68,14 @@ pub fn add_peer_context_if_needed(datanode: Peer) -> impl FnOnce(Error) -> Error } } +pub(crate) fn is_metric_engine_logical_table( + table_info: &TableInfo, + table_route_value: &TableRouteValue, +) -> bool { + table_info.meta.engine == METRIC_ENGINE + && matches!(table_route_value, TableRouteValue::Logical(_)) +} + /// Maps the error to the corresponding procedure error. /// /// This function determines whether the error should be retried and if poison cleanup is needed, diff --git a/src/common/meta/src/ddl_manager.rs b/src/common/meta/src/ddl_manager.rs index a92c9b679a..bfd858a196 100644 --- a/src/common/meta/src/ddl_manager.rs +++ b/src/common/meta/src/ddl_manager.rs @@ -43,7 +43,9 @@ use crate::ddl::drop_database::DropDatabaseProcedure; use crate::ddl::drop_flow::DropFlowProcedure; use crate::ddl::drop_table::DropTableProcedure; use crate::ddl::drop_view::DropViewProcedure; +use crate::ddl::purge_dropped_table::PurgeDroppedTableProcedure; use crate::ddl::truncate_table::TruncateTableProcedure; +use crate::ddl::undrop_table::UndropTableProcedure; use crate::ddl::{DdlContext, utils}; use crate::error::{ self, CreateRepartitionProcedureSnafu, EmptyDdlTasksSnafu, ProcedureOutputSnafu, @@ -62,7 +64,7 @@ use crate::rpc::ddl::DdlTask::DropTrigger; use crate::rpc::ddl::DdlTask::{ AlterDatabase, AlterLogicalTables, AlterTable, CommentOn, CreateDatabase, CreateFlow, CreateLogicalTables, CreateTable, CreateView, DropDatabase, DropFlow, DropLogicalTables, - DropTable, DropView, TruncateTable, + DropTable, DropView, PurgeDroppedTable, TruncateTable, UndropTable, }; #[cfg(feature = "enterprise")] use crate::rpc::ddl::trigger::CreateTriggerTask; @@ -71,7 +73,8 @@ use crate::rpc::ddl::trigger::DropTriggerTask; use crate::rpc::ddl::{ AlterDatabaseTask, AlterTableTask, CommentOnTask, CreateDatabaseTask, CreateFlowTask, CreateTableTask, CreateViewTask, DropDatabaseTask, DropFlowTask, DropTableTask, DropViewTask, - QueryContext, SubmitDdlTaskRequest, SubmitDdlTaskResponse, TruncateTableTask, + PurgeDroppedTableTask, QueryContext, SubmitDdlTaskRequest, SubmitDdlTaskResponse, + TruncateTableTask, UndropTableTask, }; /// A configurator that customizes or enhances a [`DdlManager`]. @@ -252,6 +255,8 @@ impl DdlManager { AlterLogicalTablesProcedure, AlterDatabaseProcedure, DropTableProcedure, + UndropTableProcedure, + PurgeDroppedTableProcedure, DropFlowProcedure, TruncateTableProcedure, CreateDatabaseProcedure, @@ -468,6 +473,32 @@ impl DdlManager { self.execute_procedure_and_wait(procedure_with_id).await } + /// Submits and executes an undrop table task. + #[tracing::instrument(skip_all)] + pub async fn submit_undrop_table_task( + &self, + undrop_table_task: UndropTableTask, + ) -> Result<(ProcedureId, Option)> { + let context = self.create_context(); + let procedure = UndropTableProcedure::new(undrop_table_task, context); + let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure)); + + self.execute_procedure_and_wait(procedure_with_id).await + } + + /// Submits and executes a purge dropped table task. + #[tracing::instrument(skip_all)] + pub async fn submit_purge_dropped_table_task( + &self, + purge_dropped_table_task: PurgeDroppedTableTask, + ) -> Result<(ProcedureId, Option)> { + let context = self.create_context(); + let procedure = PurgeDroppedTableProcedure::new(purge_dropped_table_task, context); + let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure)); + + self.execute_procedure_and_wait(procedure_with_id).await + } + /// Submits and executes a create database task. #[tracing::instrument(skip_all)] pub async fn submit_create_database( @@ -643,6 +674,12 @@ impl DdlManager { handle_create_table_task(self, create_table_task, request.query_context).await } DropTable(drop_table_task) => handle_drop_table_task(self, drop_table_task).await, + UndropTable(undrop_table_task) => { + handle_undrop_table_task(self, undrop_table_task).await + } + PurgeDroppedTable(purge_dropped_table_task) => { + handle_purge_dropped_table_task(self, purge_dropped_table_task).await + } AlterTable(alter_table_task) => { handle_alter_table_task(self, alter_table_task, ddl_options).await } @@ -790,6 +827,39 @@ async fn handle_drop_table_task( }) } +async fn handle_undrop_table_task( + ddl_manager: &DdlManager, + undrop_table_task: UndropTableTask, +) -> Result { + let table_id = undrop_table_task.table_id; + let (id, _) = ddl_manager + .submit_undrop_table_task(undrop_table_task) + .await?; + + info!("Table: {table_id} is undropped via procedure_id {id:?}"); + + Ok(SubmitDdlTaskResponse { + key: id.to_string().into(), + ..Default::default() + }) +} + +async fn handle_purge_dropped_table_task( + ddl_manager: &DdlManager, + purge_dropped_table_task: PurgeDroppedTableTask, +) -> Result { + let (id, _) = ddl_manager + .submit_purge_dropped_table_task(purge_dropped_table_task) + .await?; + + info!("Dropped table is purged via procedure_id {id:?}"); + + Ok(SubmitDdlTaskResponse { + key: id.to_string().into(), + ..Default::default() + }) +} + async fn handle_create_table_task( ddl_manager: &DdlManager, create_table_task: CreateTableTask, @@ -1238,6 +1308,7 @@ mod tests { memory_region_keeper: Arc::new(MemoryRegionKeeper::default()), leader_region_registry: Arc::new(LeaderRegionRegistry::default()), region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl), + soft_drop_enabled: false, }, procedure_manager.clone(), Arc::new(DummyRepartitionProcedureFactory), diff --git a/src/common/meta/src/error.rs b/src/common/meta/src/error.rs index 1528f4f387..6832940d11 100644 --- a/src/common/meta/src/error.rs +++ b/src/common/meta/src/error.rs @@ -379,6 +379,20 @@ pub enum Error { location: Location, }, + #[snafu(display( + "Cannot drop table '{}': tombstoned table id {} already uses the same full name", + table_name, + existing_table_id + ))] + /// Raised when a live table is recreated with a name still reserved by an older tombstone. + TableNameTombstoneConflict { + table_name: String, + existing_table_id: TableId, + dropping_table_id: TableId, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("View already exists, view: {}", view_name))] ViewAlreadyExists { view_name: String, @@ -727,6 +741,13 @@ pub enum Error { location: Location, }, + #[snafu(display("Failed to restore tombstone, target key already exists: {key}"))] + TombstoneTargetAlreadyExists { + key: String, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Failed to parse {} from utf8", name))] FromUtf8 { name: String, @@ -1130,6 +1151,7 @@ impl ErrorExt for Error { | EtcdTxnFailed { .. } | ConnectEtcd { .. } | MoveValues { .. } + | TombstoneTargetAlreadyExists { .. } | GetCache { .. } | GetLatestCacheRetryExceeded { .. } | SerializeToJson { .. } @@ -1221,7 +1243,9 @@ impl ErrorExt for Error { ViewNotFound { .. } | TableNotFound { .. } | RegionNotFound { .. } => { StatusCode::TableNotFound } - ViewAlreadyExists { .. } | TableAlreadyExists { .. } => StatusCode::TableAlreadyExists, + ViewAlreadyExists { .. } + | TableAlreadyExists { .. } + | TableNameTombstoneConflict { .. } => StatusCode::TableAlreadyExists, SubmitProcedure { source, .. } | QueryProcedure { source, .. } diff --git a/src/common/meta/src/key.rs b/src/common/meta/src/key.rs index 0be3ee1a45..56fa35b943 100644 --- a/src/common/meta/src/key.rs +++ b/src/common/meta/src/key.rs @@ -1155,7 +1155,10 @@ impl TableMetadataManager { table_route_value: &TableRouteValue, ) -> Result> { let mut region_wal_options = HashMap::new(); - let datanode_table_keys = region_distribution(table_route_value.region_routes()?) + let Some(region_routes) = table_route_value.region_routes().ok() else { + return Ok(region_wal_options); + }; + let datanode_table_keys = region_distribution(region_routes) .into_keys() .map(|datanode_id| DatanodeTableKey::new(datanode_id, table_id)) .collect::>(); @@ -1637,7 +1640,7 @@ mod tests { use common_wal::options::{KafkaWalOptions, WalOptions}; use futures::TryStreamExt; use store_api::storage::{RegionId, RegionNumber}; - use table::metadata::TableInfo; + use table::metadata::{TableId, TableInfo}; use table::table_name::TableName; use super::datanode_table::DatanodeTableKey; @@ -1645,7 +1648,7 @@ mod tests { use crate::ddl::allocator::wal_options::WalOptionsAllocator; use crate::ddl::test_util::create_table::test_create_table_task; use crate::ddl::utils::region_storage_path; - use crate::error::Result; + use crate::error::{Error, Result}; use crate::key::datanode_table::RegionInfo; use crate::key::node_address::{NodeAddressKey, NodeAddressValue}; use crate::key::table_info::TableInfoValue; @@ -1653,8 +1656,8 @@ mod tests { use crate::key::table_route::TableRouteValue; use crate::key::topic_region::TopicRegionKey; use crate::key::{ - DeserializedValueWithBytes, MetadataValue, RegionDistribution, RegionRoleSet, - TOPIC_REGION_PREFIX, TableMetadataManager, ViewInfoValue, + DeserializedValueWithBytes, DroppedTableMetadata, MetadataValue, RegionDistribution, + RegionRoleSet, TOPIC_REGION_PREFIX, TableMetadataManager, ViewInfoValue, }; use crate::kv_backend::KvBackend; use crate::kv_backend::memory::MemoryKvBackend; @@ -1776,6 +1779,96 @@ mod tests { ]) } + fn test_physical_region_route( + table_id: TableId, + region_number: RegionNumber, + leader_peer: u64, + follower_peers: Vec, + ) -> RegionRoute { + RegionRoute { + region: Region::new_test(RegionId::new(table_id, region_number)), + leader_peer: Some(Peer::empty(leader_peer)), + follower_peers: follower_peers.into_iter().map(Peer::empty).collect(), + leader_state: None, + leader_down_since: None, + write_route_policy: None, + } + } + + async fn create_dropped_physical_table_metadata( + table_id: TableId, + table_name: &str, + region_routes: Vec, + region_wal_options: HashMap, + ) -> ( + Arc>, + TableMetadataManager, + TableName, + TableInfo, + Vec, + HashMap, + ) { + let mem_kv = Arc::new(MemoryKvBackend::default()); + let table_metadata_manager = TableMetadataManager::new(mem_kv.clone()); + let task = test_create_table_task(table_name, table_id); + let table_info = task.table_info.clone(); + table_metadata_manager + .create_table_metadata( + table_info.clone(), + TableRouteValue::physical(region_routes), + region_wal_options.clone(), + ) + .await + .unwrap(); + + let table_route_value = table_metadata_manager + .table_route_manager + .table_route_storage() + .get_with_raw_bytes(table_id) + .await + .unwrap() + .unwrap(); + let region_routes = table_route_value.region_routes().unwrap().clone(); + let table_name = TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, table_name); + let table_route_value = TableRouteValue::physical(region_routes.clone()); + table_metadata_manager + .delete_table_metadata( + table_id, + &table_name, + &table_route_value, + ®ion_wal_options, + ) + .await + .unwrap(); + + ( + mem_kv, + table_metadata_manager, + table_name, + table_info, + region_routes, + region_wal_options, + ) + } + + fn assert_dropped_table_metadata( + dropped_table: &DroppedTableMetadata, + table_id: TableId, + table_name: &TableName, + table_info: &TableInfo, + region_routes: &[RegionRoute], + region_wal_options: &HashMap, + ) { + assert_eq!(dropped_table.table_id, table_id); + assert_eq!(&dropped_table.table_name, table_name); + assert_eq!(&dropped_table.table_info_value.table_info, table_info); + assert_eq!( + dropped_table.table_route_value.region_routes().unwrap(), + region_routes + ); + assert_eq!(&dropped_table.region_wal_options, region_wal_options); + } + #[tokio::test] async fn test_raft_engine_topic_region_map() { let mem_kv = Arc::new(MemoryKvBackend::default()); @@ -2854,62 +2947,20 @@ mod tests { #[tokio::test] async fn test_dropped_table_metadata_enumeration_and_lookup() { - let mem_kv = Arc::new(MemoryKvBackend::default()); - let table_metadata_manager = TableMetadataManager::new(mem_kv.clone()); let table_id = 1025; let table_name = "foo"; - let task = test_create_table_task(table_name, table_id); - let table_info = task.table_info.clone(); - let options = create_mixed_region_wal_options(); - let serialized_options = options.clone(); - table_metadata_manager - .create_table_metadata( - table_info.clone(), - TableRouteValue::physical(vec![ - RegionRoute { - region: Region::new_test(RegionId::new(table_id, 1)), - leader_peer: Some(Peer::empty(1)), - follower_peers: vec![Peer::empty(5)], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }, - RegionRoute { - region: Region::new_test(RegionId::new(table_id, 2)), - leader_peer: Some(Peer::empty(2)), - follower_peers: vec![Peer::empty(4)], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }, - RegionRoute { - region: Region::new_test(RegionId::new(table_id, 3)), - leader_peer: Some(Peer::empty(3)), - follower_peers: vec![], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }, - ]), - serialized_options, + let (_, table_metadata_manager, table_name, table_info, region_routes, options) = + create_dropped_physical_table_metadata( + table_id, + table_name, + vec![ + test_physical_region_route(table_id, 1, 1, vec![5]), + test_physical_region_route(table_id, 2, 2, vec![4]), + test_physical_region_route(table_id, 3, 3, vec![]), + ], + create_mixed_region_wal_options(), ) - .await - .unwrap(); - let table_route_value = table_metadata_manager - .table_route_manager - .table_route_storage() - .get_with_raw_bytes(table_id) - .await - .unwrap() - .unwrap(); - let region_routes = table_route_value.region_routes().unwrap(); - let table_name = TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, table_name); - let table_route_value = TableRouteValue::physical(region_routes.clone()); - - table_metadata_manager - .delete_table_metadata(table_id, &table_name, &table_route_value, &options) - .await - .unwrap(); + .await; let dropped_tables = table_metadata_manager.list_dropped_tables().await.unwrap(); assert_eq!(dropped_tables.len(), 1); @@ -2921,104 +2972,64 @@ mod tests { .await .unwrap() .unwrap(); - assert_eq!(dropped_table.table_id, table_id); - assert_eq!(dropped_table.table_name, table_name); - assert_eq!(dropped_table.table_info_value.table_info, table_info); - assert_eq!( - dropped_table.table_route_value.region_routes().unwrap(), - region_routes + assert_dropped_table_metadata( + &dropped_table, + table_id, + &table_name, + &table_info, + ®ion_routes, + &options, ); - assert_eq!(dropped_table.region_wal_options, options); let dropped_table_by_id = table_metadata_manager .get_dropped_table_by_id(table_id) .await .unwrap() .unwrap(); - assert_eq!(dropped_table_by_id.table_id, table_id); - assert_eq!(dropped_table_by_id.table_name, table_name); - assert_eq!(dropped_table_by_id.table_info_value.table_info, table_info); - assert_eq!( - dropped_table_by_id - .table_route_value - .region_routes() - .unwrap(), - region_routes + assert_dropped_table_metadata( + &dropped_table_by_id, + table_id, + &table_name, + &table_info, + ®ion_routes, + &options, ); - assert_eq!(dropped_table_by_id.region_wal_options, options); } #[tokio::test] async fn test_dropped_table_lookup_survives_live_name_recreation() { - let mem_kv = Arc::new(MemoryKvBackend::default()); - let table_metadata_manager = TableMetadataManager::new(mem_kv.clone()); let dropped_table_id = 1025; let recreated_table_id = 1026; let table_name = "foo"; - let dropped_task = test_create_table_task(table_name, dropped_table_id); - let dropped_table_info = dropped_task.table_info.clone(); - let options = create_mock_region_wal_options(); - let serialized_options = options.clone(); - table_metadata_manager - .create_table_metadata( - dropped_table_info.clone(), - TableRouteValue::physical(vec![ - RegionRoute { - region: Region::new_test(RegionId::new(dropped_table_id, 1)), - leader_peer: Some(Peer::empty(1)), - follower_peers: vec![Peer::empty(5)], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }, - RegionRoute { - region: Region::new_test(RegionId::new(dropped_table_id, 2)), - leader_peer: Some(Peer::empty(2)), - follower_peers: vec![Peer::empty(4)], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }, - ]), - serialized_options.clone(), - ) - .await - .unwrap(); - - let dropped_table_name = - TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, table_name); - let dropped_table_route = table_metadata_manager - .table_route_manager - .table_route_storage() - .get_with_raw_bytes(dropped_table_id) - .await - .unwrap() - .unwrap(); - let dropped_table_route = - TableRouteValue::physical(dropped_table_route.region_routes().unwrap().clone()); - table_metadata_manager - .delete_table_metadata( - dropped_table_id, - &dropped_table_name, - &dropped_table_route, - &options, - ) - .await - .unwrap(); + let ( + _, + table_metadata_manager, + dropped_table_name, + dropped_table_info, + region_routes, + options, + ) = create_dropped_physical_table_metadata( + dropped_table_id, + table_name, + vec![ + test_physical_region_route(dropped_table_id, 1, 1, vec![5]), + test_physical_region_route(dropped_table_id, 2, 2, vec![4]), + ], + create_mock_region_wal_options(), + ) + .await; let recreated_task = test_create_table_task(table_name, recreated_table_id); table_metadata_manager .create_table_metadata( recreated_task.table_info, - TableRouteValue::physical(vec![RegionRoute { - region: Region::new_test(RegionId::new(recreated_table_id, 1)), - leader_peer: Some(Peer::empty(4)), - follower_peers: vec![], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }]), - serialized_options, + TableRouteValue::physical(vec![test_physical_region_route( + recreated_table_id, + 1, + 4, + vec![], + )]), + HashMap::new(), ) .await .unwrap(); @@ -3039,11 +3050,13 @@ mod tests { .await .unwrap() .unwrap(); - assert_eq!(dropped_table.table_id, dropped_table_id); - assert_eq!(dropped_table.table_name, dropped_table_name); - assert_eq!( - dropped_table.table_info_value.table_info, - dropped_table_info + assert_dropped_table_metadata( + &dropped_table, + dropped_table_id, + &dropped_table_name, + &dropped_table_info, + ®ion_routes, + &options, ); let dropped_tables = table_metadata_manager.list_dropped_tables().await.unwrap(); @@ -3054,55 +3067,19 @@ mod tests { #[tokio::test] async fn test_dropped_table_lookup_ignores_unrelated_malformed_datanode_tombstones() { - let mem_kv = Arc::new(MemoryKvBackend::default()); - let table_metadata_manager = TableMetadataManager::new(mem_kv.clone()); let table_id = 1025; let table_name = "foo"; - let task = test_create_table_task(table_name, table_id); - let table_info = task.table_info.clone(); - let options = create_mixed_region_wal_options(); - let serialized_options = options.clone(); - table_metadata_manager - .create_table_metadata( - table_info.clone(), - TableRouteValue::physical(vec![ - RegionRoute { - region: Region::new_test(RegionId::new(table_id, 1)), - leader_peer: Some(Peer::empty(1)), - follower_peers: vec![Peer::empty(5)], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }, - RegionRoute { - region: Region::new_test(RegionId::new(table_id, 2)), - leader_peer: Some(Peer::empty(2)), - follower_peers: vec![Peer::empty(4)], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }, - ]), - serialized_options, + let (mem_kv, table_metadata_manager, table_name, table_info, region_routes, options) = + create_dropped_physical_table_metadata( + table_id, + table_name, + vec![ + test_physical_region_route(table_id, 1, 1, vec![5]), + test_physical_region_route(table_id, 2, 2, vec![4]), + ], + create_mixed_region_wal_options(), ) - .await - .unwrap(); - - let table_route_value = table_metadata_manager - .table_route_manager - .table_route_storage() - .get_with_raw_bytes(table_id) - .await - .unwrap() - .unwrap(); - let region_routes = table_route_value.region_routes().unwrap(); - let table_name = TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, table_name); - let table_route_value = TableRouteValue::physical(region_routes.clone()); - - table_metadata_manager - .delete_table_metadata(table_id, &table_name, &table_route_value, &options) - .await - .unwrap(); + .await; mem_kv .put( @@ -3118,10 +3095,14 @@ mod tests { .await .unwrap() .unwrap(); - assert_eq!(dropped_table.table_id, table_id); - assert_eq!(dropped_table.table_name, table_name); - assert_eq!(dropped_table.table_info_value.table_info, table_info); - assert_eq!(dropped_table.region_wal_options, options); + assert_dropped_table_metadata( + &dropped_table, + table_id, + &table_name, + &table_info, + ®ion_routes, + &options, + ); } #[tokio::test] diff --git a/src/common/meta/src/key/tombstone.rs b/src/common/meta/src/key/tombstone.rs index a6cb6efa89..01fa440577 100644 --- a/src/common/meta/src/key/tombstone.rs +++ b/src/common/meta/src/key/tombstone.rs @@ -59,6 +59,8 @@ pub struct TombstoneManager { } const TOMBSTONE_PREFIX: &str = "__tombstone/"; +const MOVE_VALUE_TXN_OPS_PER_KEY: usize = 4; +const RESTORE_VALUE_TXN_OPS_PER_KEY: usize = 6; impl TombstoneManager { /// Returns [TombstoneManager]. @@ -155,23 +157,40 @@ impl TombstoneManager { src_key: Vec, value: Vec, dest_key: Vec, - ) -> (Txn, impl FnMut(&mut TxnOpGetResponseSet) -> Option>) { - let txn = Txn::new() - .when(vec![Compare::with_value( - src_key.clone(), + require_dest_not_exists: bool, + ) -> Txn { + let mut compares = vec![Compare::with_value( + src_key.clone(), + CompareOp::Equal, + value.clone(), + )]; + if require_dest_not_exists { + compares.push(Compare::with_value_not_exists( + dest_key.clone(), CompareOp::Equal, - value.clone(), - )]) + )); + } + + let mut failure = vec![TxnOp::Get(src_key.clone())]; + if require_dest_not_exists { + failure.push(TxnOp::Get(dest_key.clone())); + } + + Txn::new() + .when(compares) .and_then(vec![ TxnOp::Put(dest_key.clone(), value.clone()), TxnOp::Delete(src_key.clone()), ]) - .or_else(vec![TxnOp::Get(src_key.clone())]); - - (txn, TxnOpGetResponseSet::filter(src_key)) + .or_else(failure) } - async fn move_values_inner(&self, keys: &[Vec], dest_keys: &[Vec]) -> Result { + async fn move_values_inner( + &self, + keys: &[Vec], + dest_keys: &[Vec], + require_dest_not_exists: bool, + ) -> Result { ensure!( keys.len() == dest_keys.len(), error::UnexpectedSnafu { @@ -197,15 +216,16 @@ impl TombstoneManager { const MAX_RETRIES: usize = 8; for _ in 0..MAX_RETRIES { - let (txns, (keys, filters)): (Vec<_>, (Vec<_>, Vec<_>)) = results + let (txns, keys): (Vec<_>, Vec<_>) = results .iter() .map(|(key, value)| { - let (txn, filter) = self.build_move_value_txn( + let txn = self.build_move_value_txn( key.clone(), value.clone(), lookup_table[&key].clone(), + require_dest_not_exists, ); - (txn, (key.clone(), filter)) + (txn, key.clone()) }) .unzip(); let mut resp = self.kv_backend.txn(Txn::merge_all(txns)).await?; @@ -214,11 +234,23 @@ impl TombstoneManager { } let mut set = TxnOpGetResponseSet::from(&mut resp.responses); // Updates results. - for (idx, mut filter) in filters.into_iter().enumerate() { + for key in &keys { + let dest_key = lookup_table[&key].clone(); + if require_dest_not_exists { + let mut filter = TxnOpGetResponseSet::filter(dest_key.clone()); + if filter(&mut set).is_some() { + return error::TombstoneTargetAlreadyExistsSnafu { + key: String::from_utf8_lossy(&dest_key).to_string(), + } + .fail(); + } + } + + let mut filter = TxnOpGetResponseSet::filter(key.clone()); if let Some(value) = filter(&mut set) { - results.insert(keys[idx].clone(), value); + results.insert(key.clone(), value); } else { - results.remove(&keys[idx]); + results.remove(key); } } } @@ -243,7 +275,12 @@ impl TombstoneManager { /// Moves values to `dest_key`. /// /// Returns the number of keys that were moved. - async fn move_values(&self, keys: Vec>, dest_keys: Vec>) -> Result { + async fn move_values( + &self, + keys: Vec>, + dest_keys: Vec>, + require_dest_not_exists: bool, + ) -> Result { ensure!( keys.len() == dest_keys.len(), error::UnexpectedSnafu { @@ -257,7 +294,21 @@ impl TombstoneManager { if keys.is_empty() { return Ok(0); } - let chunk_size = self.max_txn_ops() / 2; + let txn_ops_per_key = if require_dest_not_exists { + RESTORE_VALUE_TXN_OPS_PER_KEY + } else { + MOVE_VALUE_TXN_OPS_PER_KEY + }; + let max_txn_ops = self.max_txn_ops(); + ensure!( + max_txn_ops >= txn_ops_per_key, + error::UnexpectedSnafu { + err_msg: format!( + "max_txn_ops {max_txn_ops} is smaller than required txn ops per key {txn_ops_per_key}" + ) + } + ); + let chunk_size = max_txn_ops / txn_ops_per_key; if keys.len() > chunk_size { debug!( "Moving values with multiple chunks, keys len: {}, chunk_size: {}", @@ -268,11 +319,14 @@ impl TombstoneManager { let keys_chunks = keys.chunks(chunk_size).collect::>(); let dest_keys_chunks = dest_keys.chunks(chunk_size).collect::>(); for (keys, dest_keys) in keys_chunks.into_iter().zip(dest_keys_chunks) { - moved_keys += self.move_values_inner(keys, dest_keys).await?; + moved_keys += self + .move_values_inner(keys, dest_keys, require_dest_not_exists) + .await?; } Ok(moved_keys) } else { - self.move_values_inner(&keys, &dest_keys).await + self.move_values_inner(&keys, &dest_keys, require_dest_not_exists) + .await } } @@ -292,7 +346,7 @@ impl TombstoneManager { }) .unzip(); - self.move_values(keys, dest_keys).await + self.move_values(keys, dest_keys, false).await } /// Restores tombstones for keys. @@ -311,7 +365,7 @@ impl TombstoneManager { }) .unzip(); - self.move_values(keys, dest_keys).await + self.move_values(keys, dest_keys, true).await } /// Deletes tombstones values for the specified `keys`. @@ -336,14 +390,84 @@ impl TombstoneManager { #[cfg(test)] mod tests { + use std::any::Any; use std::collections::HashMap; use std::sync::Arc; - use crate::error::Error; + use crate::error::{Error, Result}; use crate::key::tombstone::TombstoneManager; - use crate::kv_backend::KvBackend; use crate::kv_backend::memory::MemoryKvBackend; - use crate::rpc::store::PutRequest; + use crate::kv_backend::txn::{Txn, TxnRequest, TxnResponse}; + use crate::kv_backend::{KvBackend, TxnService}; + use crate::rpc::store::{ + BatchDeleteRequest, BatchDeleteResponse, BatchGetRequest, BatchGetResponse, + BatchPutRequest, BatchPutResponse, DeleteRangeRequest, DeleteRangeResponse, PutRequest, + PutResponse, RangeRequest, RangeResponse, + }; + + struct TxnOpLimitKvBackend { + inner: Arc>, + max_txn_ops: usize, + } + + #[async_trait::async_trait] + impl TxnService for TxnOpLimitKvBackend { + type Error = Error; + + async fn txn(&self, txn: Txn) -> Result { + let TxnRequest { + compare, + success, + failure, + } = txn.req(); + let txn_ops = compare.len() + success.len() + failure.len(); + assert!( + txn_ops <= self.max_txn_ops, + "txn ops {txn_ops} exceeds limit {}", + self.max_txn_ops + ); + self.inner.txn(txn).await + } + + fn max_txn_ops(&self) -> usize { + self.max_txn_ops + } + } + + #[async_trait::async_trait] + impl KvBackend for TxnOpLimitKvBackend { + fn name(&self) -> &str { + "txn_op_limit" + } + + fn as_any(&self) -> &dyn Any { + self + } + + async fn range(&self, req: RangeRequest) -> Result { + self.inner.range(req).await + } + + async fn put(&self, req: PutRequest) -> Result { + self.inner.put(req).await + } + + async fn batch_put(&self, req: BatchPutRequest) -> Result { + self.inner.batch_put(req).await + } + + async fn batch_get(&self, req: BatchGetRequest) -> Result { + self.inner.batch_get(req).await + } + + async fn delete_range(&self, req: DeleteRangeRequest) -> Result { + self.inner.delete_range(req).await + } + + async fn batch_delete(&self, req: BatchDeleteRequest) -> Result { + self.inner.batch_delete(req).await + } + } #[derive(Debug, Clone)] struct MoveValue { @@ -547,14 +671,14 @@ mod tests { .map(|kv| (kv.key, kv.dest_key)) .unzip(); let moved_keys = tombstone_manager - .move_values(keys.clone(), dest_keys.clone()) + .move_values(keys.clone(), dest_keys.clone(), false) .await .unwrap(); assert_eq!(kvs.len(), moved_keys); check_moved_values(kv_backend.clone(), &move_values).await; // Moves again let moved_keys = tombstone_manager - .move_values(keys.clone(), dest_keys.clone()) + .move_values(keys.clone(), dest_keys.clone(), false) .await .unwrap(); assert_eq!(0, moved_keys); @@ -602,20 +726,111 @@ mod tests { .map(|kv| (kv.key, kv.dest_key)) .unzip(); let moved_keys = tombstone_manager - .move_values(keys.clone(), dest_keys.clone()) + .move_values(keys.clone(), dest_keys.clone(), false) .await .unwrap(); assert_eq!(kvs.len(), moved_keys); check_moved_values(kv_backend.clone(), &move_values).await; // Moves again let moved_keys = tombstone_manager - .move_values(keys.clone(), dest_keys.clone()) + .move_values(keys.clone(), dest_keys.clone(), false) .await .unwrap(); assert_eq!(0, moved_keys); check_moved_values(kv_backend.clone(), &move_values).await; } + #[tokio::test] + async fn test_restore_chunks_by_total_txn_ops_limit() { + let inner = Arc::new(MemoryKvBackend::default()); + let kv_backend = Arc::new(TxnOpLimitKvBackend { + inner: inner.clone(), + max_txn_ops: 6, + }); + let tombstone_manager = TombstoneManager::new(kv_backend); + let kvs = HashMap::from([ + (b"bar".to_vec(), b"baz".to_vec()), + (b"foo".to_vec(), b"hi".to_vec()), + (b"baz".to_vec(), b"hello".to_vec()), + ]); + for (key, value) in &kvs { + inner + .put( + PutRequest::new() + .with_key(tombstone_manager.to_tombstone(key)) + .with_value(value.clone()), + ) + .await + .unwrap(); + } + + let restored = tombstone_manager + .restore(kvs.keys().cloned().collect()) + .await + .unwrap(); + + assert_eq!(kvs.len(), restored); + } + + #[tokio::test] + async fn test_restore_fails_fast_when_txn_op_limit_too_small() { + let inner = Arc::new(MemoryKvBackend::default()); + let kv_backend = Arc::new(TxnOpLimitKvBackend { + inner: inner.clone(), + max_txn_ops: 5, + }); + let tombstone_manager = TombstoneManager::new(kv_backend); + let key = b"foo".to_vec(); + inner + .put( + PutRequest::new() + .with_key(tombstone_manager.to_tombstone(&key)) + .with_value(b"hi".to_vec()), + ) + .await + .unwrap(); + + let err = tombstone_manager.restore(vec![key]).await.unwrap_err(); + + assert!(matches!(err, Error::Unexpected { .. })); + assert!( + err.to_string() + .contains("max_txn_ops 5 is smaller than required txn ops per key 6") + ); + } + + #[tokio::test] + async fn test_create_chunks_by_total_txn_ops_limit() { + let inner = Arc::new(MemoryKvBackend::default()); + let kv_backend = Arc::new(TxnOpLimitKvBackend { + inner: inner.clone(), + max_txn_ops: 4, + }); + let tombstone_manager = TombstoneManager::new(kv_backend); + let kvs = HashMap::from([ + (b"bar".to_vec(), b"baz".to_vec()), + (b"foo".to_vec(), b"hi".to_vec()), + (b"baz".to_vec(), b"hello".to_vec()), + ]); + for (key, value) in &kvs { + inner + .put( + PutRequest::new() + .with_key(key.clone()) + .with_value(value.clone()), + ) + .await + .unwrap(); + } + + let moved = tombstone_manager + .create(kvs.keys().cloned().collect()) + .await + .unwrap(); + + assert_eq!(kvs.len(), moved); + } + #[tokio::test] async fn test_move_values_with_non_exists_values() { let kv_backend = Arc::new(MemoryKvBackend::default()); @@ -651,14 +866,14 @@ mod tests { keys.push(b"non-exists".to_vec()); dest_keys.push(b"hi/non-exists".to_vec()); let moved_keys = tombstone_manager - .move_values(keys.clone(), dest_keys.clone()) + .move_values(keys.clone(), dest_keys.clone(), false) .await .unwrap(); check_moved_values(kv_backend.clone(), &move_values).await; assert_eq!(3, moved_keys); // Moves again let moved_keys = tombstone_manager - .move_values(keys.clone(), dest_keys.clone()) + .move_values(keys.clone(), dest_keys.clone(), false) .await .unwrap(); check_moved_values(kv_backend.clone(), &move_values).await; @@ -704,7 +919,7 @@ mod tests { .map(|kv| (kv.key, kv.dest_key)) .unzip(); let moved_keys = tombstone_manager - .move_values(keys, dest_keys) + .move_values(keys, dest_keys, false) .await .unwrap(); assert_eq!(kvs.len(), moved_keys); @@ -745,7 +960,7 @@ mod tests { .map(|kv| (kv.key, kv.dest_key)) .unzip(); tombstone_manager - .move_values(keys, dest_keys) + .move_values(keys, dest_keys, false) .await .unwrap(); check_moved_values(kv_backend.clone(), &move_values).await; @@ -780,7 +995,7 @@ mod tests { .map(|kv| (kv.key, kv.dest_key)) .unzip(); tombstone_manager - .move_values(keys, dest_keys) + .move_values(keys, dest_keys, false) .await .unwrap(); check_moved_values(kv_backend.clone(), &move_values).await; @@ -795,7 +1010,7 @@ mod tests { let dest_keys = vec![b"bar".to_vec(), b"foo".to_vec(), b"baz".to_vec()]; let err = tombstone_manager - .move_values(keys, dest_keys) + .move_values(keys, dest_keys, false) .await .unwrap_err(); assert!( @@ -803,7 +1018,10 @@ mod tests { .contains("The length of keys(2) does not match the length of dest_keys(3)."), ); - let moved_keys = tombstone_manager.move_values(vec![], vec![]).await.unwrap(); + let moved_keys = tombstone_manager + .move_values(vec![], vec![], false) + .await + .unwrap(); assert_eq!(0, moved_keys); } } diff --git a/src/common/meta/src/rpc/ddl.rs b/src/common/meta/src/rpc/ddl.rs index fd11e4c310..9ddd1ed7b5 100644 --- a/src/common/meta/src/rpc/ddl.rs +++ b/src/common/meta/src/rpc/ddl.rs @@ -31,7 +31,8 @@ use api::v1::meta::{ DdlTaskResponse as PbDdlTaskResponse, DropDatabaseTask as PbDropDatabaseTask, DropFlowTask as PbDropFlowTask, DropTableTask as PbDropTableTask, DropTableTasks as PbDropTableTasks, DropViewTask as PbDropViewTask, Partition, ProcedureId, - TruncateTableTask as PbTruncateTableTask, + PurgeDroppedTableTask as PbPurgeDroppedTableTask, TruncateTableTask as PbTruncateTableTask, + UndropTableTask as PbUndropTableTask, }; use api::v1::{ AlterDatabaseExpr, AlterTableExpr, CommentObjectType as PbCommentObjectType, CommentOnExpr, @@ -71,6 +72,8 @@ pub const ORIGIN_FRONTEND_ADDR_EXTENSION_KEY: &str = "__greptime_origin_frontend pub enum DdlTask { CreateTable(CreateTableTask), DropTable(DropTableTask), + UndropTable(UndropTableTask), + PurgeDroppedTable(PurgeDroppedTableTask), AlterTable(AlterTableTask), TruncateTable(TruncateTableTask), CreateLogicalTables(Vec), @@ -152,6 +155,16 @@ impl DdlTask { }) } + /// Creates a [`DdlTask`] to undrop a table. + pub fn new_undrop_table(table_id: TableId) -> Self { + DdlTask::UndropTable(UndropTableTask { table_id }) + } + + /// Creates a [`DdlTask`] to purge a dropped table. + pub fn new_purge_dropped_table(table_id: TableId) -> Self { + DdlTask::PurgeDroppedTable(PurgeDroppedTableTask { table_id }) + } + /// Creates a [`DdlTask`] to create a database. pub fn new_create_database( catalog: String, @@ -225,6 +238,12 @@ impl TryFrom for DdlTask { Ok(DdlTask::CreateTable(create_table.try_into()?)) } Task::DropTableTask(drop_table) => Ok(DdlTask::DropTable(drop_table.try_into()?)), + Task::UndropTableTask(undrop_table) => { + Ok(DdlTask::UndropTable(undrop_table.try_into()?)) + } + Task::PurgeDroppedTableTask(purge_dropped_table) => { + Ok(DdlTask::PurgeDroppedTable(purge_dropped_table.try_into()?)) + } Task::AlterTableTask(alter_table) => Ok(DdlTask::AlterTable(alter_table.try_into()?)), Task::TruncateTableTask(truncate_table) => { Ok(DdlTask::TruncateTable(truncate_table.try_into()?)) @@ -335,6 +354,8 @@ impl TryFrom for PbDdlTaskRequest { let task = match request.task { DdlTask::CreateTable(task) => Task::CreateTableTask(task.try_into()?), DdlTask::DropTable(task) => Task::DropTableTask(task.into()), + DdlTask::UndropTable(task) => Task::UndropTableTask(task.into()), + DdlTask::PurgeDroppedTable(task) => Task::PurgeDroppedTableTask(task.into()), DdlTask::AlterTable(task) => Task::AlterTableTask(task.try_into()?), DdlTask::TruncateTable(task) => Task::TruncateTableTask(task.try_into()?), DdlTask::CreateLogicalTables(tasks) => { @@ -596,6 +617,62 @@ pub struct DropTableTask { pub drop_if_exists: bool, } +#[derive(Debug, PartialEq, Serialize, Deserialize, Clone)] +pub struct UndropTableTask { + pub table_id: TableId, +} + +#[derive(Debug, PartialEq, Serialize, Deserialize, Clone)] +pub struct PurgeDroppedTableTask { + pub table_id: TableId, +} + +impl TryFrom for UndropTableTask { + type Error = error::Error; + + fn try_from(pb: PbUndropTableTask) -> Result { + Ok(Self { + table_id: pb + .table_id + .context(error::InvalidProtoMsgSnafu { + err_msg: "expected table_id", + })? + .id, + }) + } +} + +impl From for PbUndropTableTask { + fn from(task: UndropTableTask) -> Self { + Self { + table_id: Some(api::v1::TableId { id: task.table_id }), + } + } +} + +impl TryFrom for PurgeDroppedTableTask { + type Error = error::Error; + + fn try_from(pb: PbPurgeDroppedTableTask) -> Result { + Ok(Self { + table_id: pb + .table_id + .context(error::InvalidProtoMsgSnafu { + err_msg: "expected table_id", + })? + .id, + }) + } +} + +impl From for PbPurgeDroppedTableTask { + fn from(task: PurgeDroppedTableTask) -> Self { + Self { + table_id: Some(api::v1::TableId { id: task.table_id }), + } + } +} + impl DropTableTask { pub fn table_ref(&self) -> TableReference<'_> { TableReference { @@ -1764,6 +1841,56 @@ mod tests { assert_eq!(task, de); } + #[test] + fn test_undrop_table_task_pb_roundtrip() { + let expected = UndropTableTask { table_id: 1024 }; + let request = SubmitDdlTaskRequest::new( + QueryContext::default(), + DdlTask::UndropTable(expected.clone()), + ); + + let pb = PbDdlTaskRequest::try_from(request).unwrap(); + let pb_task = pb.task.unwrap(); + let de = DdlTask::try_from(pb_task).unwrap(); + + assert!(matches!(de, DdlTask::UndropTable(task) if task == expected)); + } + + #[test] + fn test_purge_dropped_table_task_pb_roundtrip() { + let expected = PurgeDroppedTableTask { table_id: 1024 }; + let request = SubmitDdlTaskRequest::new( + QueryContext::default(), + DdlTask::PurgeDroppedTable(expected.clone()), + ); + + let pb = PbDdlTaskRequest::try_from(request).unwrap(); + let pb_task = pb.task.unwrap(); + let de = DdlTask::try_from(pb_task).unwrap(); + + assert!(matches!(de, DdlTask::PurgeDroppedTable(task) if task == expected)); + } + + #[test] + fn test_undrop_table_task_json_roundtrip() { + let task = UndropTableTask { table_id: 1024 }; + + let output = serde_json::to_vec(&task).unwrap(); + + let de = serde_json::from_slice(&output).unwrap(); + assert_eq!(task, de); + } + + #[test] + fn test_purge_dropped_table_task_json_roundtrip() { + let task = PurgeDroppedTableTask { table_id: 1024 }; + + let output = serde_json::to_vec(&task).unwrap(); + + let de = serde_json::from_slice(&output).unwrap(); + assert_eq!(task, de); + } + #[test] fn test_sort_columns() { // construct RawSchema diff --git a/src/common/meta/src/test_util.rs b/src/common/meta/src/test_util.rs index 3396870ab6..2300679b33 100644 --- a/src/common/meta/src/test_util.rs +++ b/src/common/meta/src/test_util.rs @@ -206,6 +206,7 @@ pub fn new_ddl_context_with_kv_backend( flow_metadata_allocator, flow_metadata_manager, region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl), + soft_drop_enabled: false, } } diff --git a/src/meta-srv/src/metasrv/builder.rs b/src/meta-srv/src/metasrv/builder.rs index 52ee3f3522..69bfd3f34b 100644 --- a/src/meta-srv/src/metasrv/builder.rs +++ b/src/meta-srv/src/metasrv/builder.rs @@ -425,6 +425,7 @@ impl MetasrvBuilder { flow_metadata_manager: flow_metadata_manager.clone(), flow_metadata_allocator: flow_metadata_allocator.clone(), region_failure_detector_controller, + soft_drop_enabled: ddl_soft_drop_enabled(&options), }; let procedure_manager_c = procedure_manager.clone(); let repartition_procedure_factory = Arc::new(DefaultRepartitionProcedureFactory::new( @@ -673,6 +674,13 @@ fn build_procedure_manager( )) } +/// Resolves if soft-drop is enabled from metasrv options. +fn ddl_soft_drop_enabled(_options: &MetasrvOptions) -> bool { + // TODO(hl): add a dedicated soft-drop cluster config + // when wiring the user-facing option. + false +} + impl Default for MetasrvBuilder { fn default() -> Self { Self::new() diff --git a/src/meta-srv/src/procedure/utils.rs b/src/meta-srv/src/procedure/utils.rs index 5ee473b099..bcae558773 100644 --- a/src/meta-srv/src/procedure/utils.rs +++ b/src/meta-srv/src/procedure/utils.rs @@ -546,6 +546,7 @@ pub mod test_data { memory_region_keeper: Arc::new(MemoryRegionKeeper::new()), leader_region_registry: Arc::new(LeaderRegionRegistry::default()), region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl), + soft_drop_enabled: false, } } } diff --git a/src/mito2/src/engine/open_test.rs b/src/mito2/src/engine/open_test.rs index b22e17dd01..cf4bfb1ff1 100644 --- a/src/mito2/src/engine/open_test.rs +++ b/src/mito2/src/engine/open_test.rs @@ -137,13 +137,31 @@ async fn test_engine_reopen_region_with_format(flat_format: bool) { let region_id = RegionId::new(1, 1); let request = CreateRequestBuilder::new().build(); let table_dir = request.table_dir.clone(); + let column_schemas = rows_schema(&request); engine .handle_request(region_id, RegionRequest::Create(request)) .await .unwrap(); + put_rows( + &engine, + region_id, + Rows { + schema: column_schemas, + rows: build_rows(0, 2), + }, + ) + .await; reopen_region(&engine, region_id, table_dir, false, Default::default()).await; assert!(engine.is_region_exists(region_id)); + + let scanner = engine + .scanner(region_id, ScanRequest::default()) + .await + .unwrap(); + let stream = scanner.scan().await.unwrap(); + let batches = RecordBatches::try_collect(stream).await.unwrap(); + assert_eq!(2, batches.iter().map(|b| b.num_rows()).sum::()); } #[tokio::test] diff --git a/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs index 74a1501207..122318d45f 100644 --- a/tests-integration/src/standalone.rs +++ b/tests-integration/src/standalone.rs @@ -250,6 +250,7 @@ impl GreptimeDbStandaloneBuilder { flow_metadata_manager, flow_metadata_allocator, region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl), + soft_drop_enabled: false, }, procedure_manager.clone(), repartition_procedure_factory,