diff --git a/src/cmd/src/standalone.rs b/src/cmd/src/standalone.rs index 142068711a2..5096c9221cd 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 126c360d92d..80130be5d62 100644 --- a/src/common/meta/src/ddl.rs +++ b/src/common/meta/src/ddl.rs @@ -117,6 +117,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 { diff --git a/src/common/meta/src/ddl/drop_table.rs b/src/common/meta/src/ddl/drop_table.rs index 4f4e2d60535..55508c2f1ee 100644 --- a/src/common/meta/src/ddl/drop_table.rs +++ b/src/common/meta/src/ddl/drop_table.rs @@ -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,18 @@ 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.dropping_regions.clear(); + return Ok(Status::done()); + } + self.executor .on_drop_regions( &self.context.node_manager, @@ -182,6 +195,7 @@ impl DropTableProcedure { false, ) .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 03428aa688c..1e85a64ef60 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/tests/drop_table.rs b/src/common/meta/src/ddl/tests/drop_table.rs index c518bb36a2f..c5bdbaa5c74 100644 --- a/src/common/meta/src/ddl/tests/drop_table.rs +++ b/src/common/meta/src/ddl/tests/drop_table.rs @@ -12,6 +12,7 @@ // 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; @@ -30,6 +31,7 @@ use tokio::sync::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::test_util::create_table::test_create_table_task; use crate::ddl::test_util::datanode_handler::{DatanodeWatcher, NaiveDatanodeHandler}; @@ -37,6 +39,8 @@ 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::error::Error; +use crate::key::table_name::TableNameKey; use crate::key::table_route::TableRouteValue; use crate::kv_backend::memory::MemoryKvBackend; use crate::peer::Peer; @@ -231,6 +235,266 @@ 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 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![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()); +} + +#[tokio::test] +async fn test_hard_drop_keeps_delete_tombstone_flow() { + let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); + let ddl_context = new_ddl_context(node_manager); + 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()); +} + +#[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_on_rollback() { let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); diff --git a/src/common/meta/src/ddl_manager.rs b/src/common/meta/src/ddl_manager.rs index a92c9b679a1..bba40c1a7e3 100644 --- a/src/common/meta/src/ddl_manager.rs +++ b/src/common/meta/src/ddl_manager.rs @@ -1238,6 +1238,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 1528f4f387c..1b0d7b49f3e 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, @@ -1221,7 +1235,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/test_util.rs b/src/common/meta/src/test_util.rs index 3396870ab66..2300679b337 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 52ee3f35228..69bfd3f34bb 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 5ee473b099d..bcae558773a 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/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs index 74a15012071..122318d45f1 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,