From 81d8beee3a4a323f73be00a93821bd3316d9e4b2 Mon Sep 17 00:00:00 2001 From: "Lei, HUANG" <6406592+v0y4g3r@users.noreply.github.com> Date: Tue, 14 Jul 2026 16:39:36 +0800 Subject: [PATCH] feat: clean up soft-dropped regions offline (#8458) * fix(meta): skip reopening dropped tables during purge Purge soft-dropped tables by dropping stored routes directly instead of reopening tombstoned regions first. Treat legacy \`PurgeDroppedTableState::OpenRegions\` snapshots as a compatibility-only transition to \`DropRegions\`. Files: - \`src/common/meta/src/ddl/purge_dropped_table.rs\` - \`src/common/meta/src/ddl/undrop_table.rs\` - \`src/common/meta/src/ddl/tests/drop_table.rs\` Signed-off-by: Lei, HUANG * fix(meta): guard undrop restore race Serialize `UNDROP TABLE` with same-name creates by seeding the original table name before procedure submission, and clean up reopened regions when metadata restore fails. Files: - `src/common/meta/src/ddl/undrop_table.rs` - `src/common/meta/src/ddl_manager.rs` - `src/common/meta/src/ddl/tests/drop_table.rs` Signed-off-by: Lei, HUANG * feat: clean up soft-dropped regions offline Use an explicit RegionCleanUp request for purge-table cleanup so tombstoned regions can be removed without reopening them. Route cleanup through datanode, Mito, and metric-engine offline paths, including WAL obsoletion and region directory removal. Signed-off-by: Lei, HUANG * fix(datanode): reject cleanup for registered regions Return `RegionBusy` when `RegionCleanUp` targets a region already tracked by the datanode, so offline cleanup only runs for regions without a local mapping. Add coverage for `OfflineCleanup` engine selection and registered-region rejection. Files: - `src/datanode/src/region_server.rs` Signed-off-by: Lei, HUANG * fix(meta): require tombstone before undrop Signed-off-by: Lei, HUANG * fix(meta): reject file-engine soft drop Signed-off-by: Lei, HUANG * fix: harden soft-drop cleanup paths Reject `RegionCleanUp` for already-open Mito regions instead of turning cleanup into a drop. Make `UndropTableProcedure` tolerate missing persisted table names and always deregister failure detectors after restore-failure cleanup. Files: - `src/common/meta/src/ddl/undrop_table.rs` - `src/mito2/src/engine/open_test.rs` - `src/mito2/src/worker/handle_open.rs` Signed-off-by: Lei, HUANG * chore: bump cleanup proto dependency Bump \`greptime-proto\` to the reviewed cleanup RPC revision and align cleanup request parsing and dispatch with the renamed \`CleanUpRequest\` payload. Files: - \`Cargo.toml\` - \`Cargo.lock\` - \`src/store-api/src/region_request.rs\` - \`src/common/meta/src/ddl/drop_table/executor.rs\` Signed-off-by: Lei, HUANG * refactor: share region cleanup helpers Share common region cleanup helpers between normal drop and offline cleanup while keeping their preconditions separate. - Extract shared dropped-region runtime cleanup for `handle_drop_request` and `handle_offline_cleanup_request`. - Share runtime artifact and manifest cache cleanup after full deletion paths. - Make full-drop directory removal policy explicit: full drop and purge cleanup force physical deletion, while partial drop may defer to global GC. Files: - `src/mito2/src/worker/handle_drop.rs` - `src/mito2/src/worker/handle_open.rs` Signed-off-by: Lei, HUANG * chore: preserve soft-drop cleanup split state Signed-off-by: Lei, HUANG * refactor: remove obsolete CleanUp match arm from RegionRequest The `CleanUp` variant in the `region_request::Body` match is now handled exclusively by `RegionServer` via a separate path. This arm would have returned an unexpected error, so removing it eliminates dead code. Signed-off-by: Lei, HUANG * fix(meta): clean every soft-dropped region replica Signed-off-by: Lei, HUANG * fix(meta): order soft-drop replica cleanup Signed-off-by: Lei, HUANG * Revert "fix(meta): order soft-drop replica cleanup" This reverts commit e77162d3e5ebcf2817e2845a6a5177c328fb2c60. Signed-off-by: Lei, HUANG * Revert "fix(meta): clean every soft-dropped region replica" This reverts commit 2378e00cc258ca1b6a85a1aafbd68c79c666f43c. Signed-off-by: Lei, HUANG --------- Signed-off-by: Lei, HUANG --- .../meta/src/ddl/drop_table/executor.rs | 80 ++++- .../meta/src/ddl/drop_table/metadata.rs | 7 + .../meta/src/ddl/purge_dropped_table.rs | 34 +- src/common/meta/src/ddl/tests/drop_table.rs | 329 +++++++++++------- src/common/meta/src/ddl/undrop_table.rs | 285 ++++++++++++--- src/common/meta/src/ddl_manager.rs | 69 +++- src/datanode/src/region_server.rs | 139 +++++++- src/metric-engine/src/engine.rs | 59 +++- src/metric-engine/src/engine/open.rs | 95 ++++- src/metric-engine/src/error.rs | 10 + src/mito2/src/engine/open_test.rs | 106 +++++- src/mito2/src/region/opener.rs | 57 +-- src/mito2/src/request.rs | 13 +- src/mito2/src/wal.rs | 13 + src/mito2/src/worker.rs | 4 + src/mito2/src/worker/handle_drop.rs | 71 +++- src/mito2/src/worker/handle_open.rs | 54 ++- src/store-api/src/region_request.rs | 71 +++- 18 files changed, 1229 insertions(+), 267 deletions(-) diff --git a/src/common/meta/src/ddl/drop_table/executor.rs b/src/common/meta/src/ddl/drop_table/executor.rs index 74b2b2482b..1b0993170e 100644 --- a/src/common/meta/src/ddl/drop_table/executor.rs +++ b/src/common/meta/src/ddl/drop_table/executor.rs @@ -15,8 +15,8 @@ use std::collections::{HashMap, HashSet}; use api::v1::region::{ - CloseRequest as PbCloseRegionRequest, DropRequest as PbDropRegionRequest, RegionRequest, - RegionRequestHeader, region_request, + CleanUpRequest as PbCleanUpRequest, CloseRequest as PbCloseRegionRequest, + DropRequest as PbDropRegionRequest, RegionRequest, RegionRequestHeader, region_request, }; use common_error::ext::ErrorExt; use common_error::status_code::StatusCode; @@ -26,12 +26,14 @@ use common_wal::options::WalOptions; use futures::future::join_all; use snafu::ensure; use store_api::storage::{RegionId, RegionNumber}; -use table::metadata::TableId; +use table::metadata::{TableId, TableInfo}; use table::table_name::TableName; use crate::cache_invalidator::Context; -use crate::ddl::DdlContext; -use crate::ddl::utils::{add_peer_context_if_needed, convert_region_routes_to_detecting_regions}; +use crate::ddl::utils::{ + add_peer_context_if_needed, convert_region_routes_to_detecting_regions, 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; @@ -348,6 +350,74 @@ impl DropTableExecutor { Ok(()) } + /// Cleans leader regions on datanodes without reopening them as live regions. + pub async fn on_cleanup_regions_offline( + &self, + node_manager: &NodeManagerRef, + leader_region_registry: &LeaderRegionRegistryRef, + table_info: &TableInfo, + region_routes: &[RegionRoute], + region_wal_options: &HashMap, + ) -> Result<()> { + let template = build_template_from_raw_table_info(table_info)?; + let builder = CreateRequestBuilder::new(template, None); + let storage_path = region_storage_path(&self.table.catalog_name, &self.table.schema_name); + + let leaders = find_leaders(region_routes); + let mut cleanup_region_tasks = Vec::with_capacity(leaders.len()); + let table_id = self.table_id; + for datanode in leaders { + let requester = node_manager.datanode(&datanode).await; + let regions = find_leader_regions(region_routes, &datanode); + let region_ids = regions + .iter() + .map(|region_number| RegionId::new(table_id, *region_number)) + .collect::>(); + + for region_id in region_ids { + debug!("Cleaning region {region_id} offline on Datanode {datanode:?}"); + 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::CleanUp(PbCleanUpRequest { + 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(); + cleanup_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(cleanup_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(()) + } + /// Closes all table regions on datanodes without deleting region files or metadata tombstones. /// When `flush_leaders_on_close` is set, only leader regions are flushed before close. pub async fn on_close_regions( diff --git a/src/common/meta/src/ddl/drop_table/metadata.rs b/src/common/meta/src/ddl/drop_table/metadata.rs index 6289b6c61a..982182ffbf 100644 --- a/src/common/meta/src/ddl/drop_table/metadata.rs +++ b/src/common/meta/src/ddl/drop_table/metadata.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use common_catalog::consts::FILE_ENGINE; use common_catalog::format_full_table_name; use snafu::{OptionExt, ensure}; use store_api::metric_engine_consts::METRIC_ENGINE_NAME; @@ -61,6 +62,12 @@ impl DropTableProcedure { if physical_table_id == self.data.table_id() { let engine = table_info_value.table_info.meta.engine; + ensure!( + !(self.context.soft_drop_enabled && engine == FILE_ENGINE), + error::UnsupportedSnafu { + operation: "soft-dropping file-engine tables".to_string() + } + ); // 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 index b95e5db976..759e908c8b 100644 --- a/src/common/meta/src/ddl/purge_dropped_table.rs +++ b/src/common/meta/src/ddl/purge_dropped_table.rs @@ -29,7 +29,6 @@ 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, @@ -84,22 +83,6 @@ impl PurgeDroppedTableProcedure { 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)) } @@ -107,13 +90,12 @@ impl PurgeDroppedTableProcedure { async fn on_drop_regions(&mut self) -> Result { if let Some(region_routes) = self.data.physical_region_routes() { self.executor() - .on_drop_regions( + .on_cleanup_regions_offline( &self.context.node_manager, &self.context.leader_region_registry, + self.data.table_info(), region_routes, - false, - false, - false, + &self.data.region_wal_options, ) .await?; self.context @@ -157,7 +139,6 @@ impl Procedure for PurgeDroppedTableProcedure { 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, } @@ -206,14 +187,14 @@ impl PurgeDroppedTableData { 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 table_info(&self) -> &TableInfo { + self.table_info.as_ref().unwrap() + } + fn physical_region_routes(&self) -> Option<&[RegionRoute]> { match self.table_route_value() { TableRouteValue::Physical(route) => Some(&route.region_routes), @@ -225,7 +206,6 @@ impl PurgeDroppedTableData { #[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 5c90dcbd79..b14fe77cc5 100644 --- a/src/common/meta/src/ddl/tests/drop_table.rs +++ b/src/common/meta/src/ddl/tests/drop_table.rs @@ -14,19 +14,17 @@ use std::assert_matches; use std::collections::{HashMap, HashSet}; -use std::sync::{Arc, Mutex as StdMutex}; +use std::sync::Arc; -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::{BoxedError, ErrorExt, StackError}; +use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, FILE_ENGINE}; +use common_error::ext::ErrorExt; use common_error::status_code::StatusCode; -use common_procedure::Procedure; +use common_procedure::{Procedure, StringKey}; 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; @@ -44,7 +42,7 @@ use crate::ddl::test_util::{ }; use crate::ddl::undrop_table::UndropTableProcedure; use crate::ddl::{DetectingRegion, RegionFailureDetectorController, TableMetadata}; -use crate::error::{self, Error}; +use crate::error::Error; use crate::key::table_name::TableNameKey; use crate::key::table_route::TableRouteValue; use crate::kv_backend::memory::MemoryKvBackend; @@ -741,6 +739,34 @@ async fn test_soft_drop_metric_logical_table_fails() { assert_eq!(err.status_code(), StatusCode::Unsupported); } +#[tokio::test] +async fn test_soft_drop_file_engine_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 table_id = 1024; + let table_name = "foo"; + let mut task = test_create_table_task(table_name, table_id); + task.table_info.meta.engine = FILE_ENGINE.to_string(); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::physical(vec![]), + HashMap::new(), + ) + .await + .unwrap(); + + let mut procedure = DropTableProcedure::new( + new_drop_table_task(table_name, 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)); @@ -872,6 +898,174 @@ async fn test_undrop_table_fails_when_live_name_is_created_after_prepare() { assert_eq!(live_table.table_id(), live_table_id); } +#[tokio::test] +async fn test_undrop_table_closes_opened_regions_when_restore_metadata_races_with_create() { + 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 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![RegionRoute { + region: Region::new_test(RegionId::new(dropped_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, dropped_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 = + UndropTableProcedure::new(new_undrop_table_task(dropped_table_id), ddl_context.clone()); + let ctx = new_test_procedure_context(); + procedure.execute(&ctx).await.unwrap(); + procedure.execute(&ctx).await.unwrap(); + + 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(dropped_table_id, 1).as_u64()), + (2, RegionId::new(dropped_table_id, 1).as_u64()), + ]) + ); + assert_eq!( + detector_controller.registered().await, + vec![(1, RegionId::new(dropped_table_id, 1))] + ); + + 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(&ctx).await.unwrap_err(); + assert_eq!(err.status_code(), StatusCode::TableAlreadyExists); + + let mut closed_regions = HashSet::new(); + for _ in 0..2 { + let (peer, request) = rx.try_recv().unwrap(); + let Some(region_request::Body::Close(req)) = request.body else { + unreachable!(); + }; + closed_regions.insert((peer.id, req.region_id)); + } + assert_eq!( + closed_regions, + HashSet::from([ + (1, RegionId::new(dropped_table_id, 1).as_u64()), + (2, RegionId::new(dropped_table_id, 1).as_u64()), + ]) + ); + assert!(rx.try_recv().is_err()); + assert_eq!( + detector_controller.deregistered().await, + vec![(1, RegionId::new(dropped_table_id, 1))] + ); + + 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_lock_key_includes_original_table_name_before_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 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 original_table_name = ddl_context + .table_metadata_manager + .get_dropped_table_by_id(table_id) + .await + .unwrap() + .unwrap() + .table_name; + let procedure = UndropTableProcedure::new_with_original_table_name( + new_undrop_table_task(table_id), + ddl_context, + Some(original_table_name), + ); + + let keys = procedure + .lock_key() + .keys_to_lock() + .cloned() + .collect::>(); + assert!( + keys.iter().any(|key| matches!( + key, + StringKey::Exclusive(key) if key == "__table_name_lock/greptime.public.foo" + )), + "undrop lock keys should include the original table name: {keys:?}" + ); + assert!( + keys.iter().any(|key| matches!( + key, + StringKey::Exclusive(key) if key == "__table_lock/1024" + )), + "undrop lock keys should include the dropped table id: {keys:?}" + ); +} + #[tokio::test] async fn test_undrop_table_replayed_restore_metadata_is_idempotent() { let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); @@ -921,7 +1115,7 @@ async fn test_undrop_table_replayed_restore_metadata_is_idempotent() { } #[tokio::test] -async fn test_purge_dropped_table_drops_regions_and_deletes_tombstone() { +async fn test_purge_dropped_table_cleans_regions_offline_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)); @@ -963,15 +1157,16 @@ async fn test_purge_dropped_table_drops_regions_and_deletes_tombstone() { execute_procedure_until_done(&mut procedure).await; let mut requests = Vec::new(); - for _ in 0..4 { + for _ in 0..1 { 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(_)); + let region_request::Body::CleanUp(req) = &requests[0].1 else { + unreachable!(); + }; + assert_eq!(requests[0].0, 1); + assert_eq!(req.region_id, RegionId::new(table_id, 1).as_u64()); assert!(rx.try_recv().is_err()); assert!( ddl_context @@ -1050,119 +1245,13 @@ async fn test_purge_dropped_table_by_id_selects_tombstone_when_live_table_exists 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 { + let Some(region_request::Body::CleanUp(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)); diff --git a/src/common/meta/src/ddl/undrop_table.rs b/src/common/meta/src/ddl/undrop_table.rs index f08d606a82..9bd3b24822 100644 --- a/src/common/meta/src/ddl/undrop_table.rs +++ b/src/common/meta/src/ddl/undrop_table.rs @@ -18,13 +18,12 @@ 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_telemetry::warn; use common_wal::options::WalOptions; use futures::future::join_all; use serde::{Deserialize, Serialize}; @@ -34,6 +33,7 @@ use strum::AsRefStr; use table::metadata::TableId; use table::table_name::TableName; +use crate::ddl::drop_table::executor::DropTableExecutor; 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, @@ -43,7 +43,7 @@ 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::lock_key::{CatalogLock, SchemaLock, TableLock, TableNameLock}; use crate::rpc::ddl::UndropTableTask; use crate::rpc::router::{ RegionRoute, find_follower_regions, find_followers, find_leader_regions, find_leaders, @@ -58,10 +58,17 @@ 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), - } + Self::new_with_original_table_name(task, context, None) + } + + pub(crate) fn new_with_original_table_name( + task: UndropTableTask, + context: DdlContext, + table_name: Option, + ) -> Self { + let mut data = UndropTableData::new(task); + data.table_name = table_name; + Self { context, data } } pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult { @@ -108,29 +115,101 @@ impl UndropTableProcedure { } async fn on_restore_metadata(&mut self) -> Result { + self.ensure_table_name_loaded().await?; + let table_name = self.data.table_name().clone(); let table_route_value = self.data.table_route_value(); - self.context + if let Err(err) = self + .context .table_metadata_manager .restore_table_metadata( self.data.task.table_id, - self.data.table_name(), + &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, - })?; + { + let should_cleanup_opened_regions = !err.is_retry_later(); + let err = self.map_restore_metadata_error(err); + if should_cleanup_opened_regions + && let Err(cleanup_err) = self.cleanup_opened_regions_after_restore_failure().await + { + warn!( + cleanup_err; + "Failed to close opened regions after undrop metadata restore failure, table_id: {}", + self.data.task.table_id + ); + } + return Err(err); + } self.data.state = UndropTableState::InvalidateTableCache; Ok(Status::executing(true)) } + async fn ensure_table_name_loaded(&mut self) -> Result<()> { + if self.data.table_name.is_some() { + return Ok(()); + } + + 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(), + })?; + self.data.table_name = Some(dropped_table.table_name); + Ok(()) + } + + fn map_restore_metadata_error(&self, err: error::Error) -> error::Error { + match err { + error::Error::TombstoneTargetAlreadyExists { .. } => { + let table_name = self + .data + .table_name + .as_ref() + .map(ToString::to_string) + .unwrap_or_else(|| self.data.task.table_id.to_string()); + error::TableAlreadyExistsSnafu { table_name }.build() + } + err => err, + } + } + + async fn cleanup_opened_regions_after_restore_failure(&self) -> Result<()> { + let Some(table_route_value) = self.data.table_route_value.as_ref() else { + return Ok(()); + }; + let TableRouteValue::Physical(route) = table_route_value else { + return Ok(()); + }; + let region_routes = route.region_routes.clone(); + let close_result = if let Some(table_name) = self.data.table_name.as_ref() { + let executor = + DropTableExecutor::new(table_name.clone(), self.data.task.table_id, false); + + executor + .on_close_regions( + &self.context.node_manager, + &self.context.leader_region_registry, + ®ion_routes, + false, + ) + .await + } else { + Ok(()) + }; + self.context + .deregister_failure_detectors(convert_region_routes_to_detecting_regions( + ®ion_routes, + )) + .await; + close_result?; + Ok(()) + } + 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 { @@ -213,7 +292,22 @@ impl Procedure for UndropTableProcedure { } fn lock_key(&self) -> LockKey { - LockKey::new(vec![TableLock::Write(self.data.task.table_id).into()]) + let mut lock_key = Vec::new(); + if let Some(table_name) = &self.data.table_name { + lock_key.push(CatalogLock::Read(&table_name.catalog_name).into()); + lock_key + .push(SchemaLock::read(&table_name.catalog_name, &table_name.schema_name).into()); + lock_key.push( + TableNameLock::new( + &table_name.catalog_name, + &table_name.schema_name, + &table_name.table_name, + ) + .into(), + ); + } + lock_key.push(TableLock::Write(self.data.task.table_id).into()); + LockKey::new(lock_key) } } @@ -232,27 +326,6 @@ pub(crate) async fn open_regions( 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 } @@ -264,7 +337,6 @@ async fn open_regions_inner( 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); @@ -303,9 +375,7 @@ async fn open_regions_inner( 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) - { + if let Err(err) = requester.handle(request).await { return Err(add_peer_context_if_needed(datanode)(err)); } Ok(()) @@ -366,3 +436,130 @@ enum UndropTableState { OpenRegions, InvalidateTableCache, } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use api::region::RegionResponse; + use api::v1::region::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::status_code::StatusCode; + use store_api::storage::RegionId; + use table::table_name::TableName; + use tokio::sync::{Mutex, mpsc}; + + use super::*; + use crate::ddl::test_util::datanode_handler::DatanodeWatcher; + use crate::ddl::{DetectingRegion, RegionFailureDetectorController}; + use crate::peer::Peer; + use crate::rpc::router::{Region, RegionRoute}; + use crate::test_util::{MockDatanodeManager, new_ddl_context}; + + #[derive(Default)] + struct RecordingRegionFailureDetectorController { + deregistered: Mutex>, + } + + #[async_trait] + impl RegionFailureDetectorController for RecordingRegionFailureDetectorController { + async fn register_failure_detectors(&self, _detecting_regions: Vec) {} + + async fn reset_failure_detectors(&self, _detecting_regions: Vec) {} + + async fn deregister_failure_detectors(&self, detecting_regions: Vec) { + self.deregistered.lock().await.extend(detecting_regions); + } + } + + #[test] + fn test_map_restore_metadata_error_without_table_name() { + let context = new_ddl_context(Arc::new(MockDatanodeManager::new(()))); + let procedure = UndropTableProcedure::new(UndropTableTask { table_id: 42 }, context); + + let err = procedure.map_restore_metadata_error( + error::TombstoneTargetAlreadyExistsSnafu { + key: "table-name-key".to_string(), + } + .build(), + ); + + assert_eq!(StatusCode::TableAlreadyExists, err.status_code()); + } + + #[tokio::test] + async fn test_cleanup_opened_regions_without_table_name_deregisters_detectors() { + let detector_controller = Arc::new(RecordingRegionFailureDetectorController::default()); + let mut context = new_ddl_context(Arc::new(MockDatanodeManager::new(()))); + context.region_failure_detector_controller = detector_controller.clone(); + + let table_id = 1024; + let region_id = RegionId::new(table_id, 1); + let mut procedure = UndropTableProcedure::new(UndropTableTask { table_id }, context); + procedure.data.table_route_value = Some(TableRouteValue::physical(vec![RegionRoute { + region: Region::new_test(region_id), + leader_peer: Some(Peer::empty(1)), + follower_peers: vec![], + leader_state: None, + leader_down_since: None, + write_route_policy: None, + }])); + + procedure + .cleanup_opened_regions_after_restore_failure() + .await + .unwrap(); + + assert_eq!( + detector_controller.deregistered.lock().await.as_slice(), + &[(1, region_id)] + ); + } + + #[tokio::test] + async fn test_cleanup_opened_regions_deregisters_detectors_when_close_fails() { + let (tx, _rx) = mpsc::channel(8); + let datanode_handler = DatanodeWatcher::new(tx).with_handler(|_, request| { + if matches!(request.body, Some(region_request::Body::Close(_))) { + return error::UnexpectedSnafu { + err_msg: "mock close error".to_string(), + } + .fail(); + } + Ok(RegionResponse::new(0)) + }); + let detector_controller = Arc::new(RecordingRegionFailureDetectorController::default()); + let mut context = new_ddl_context(Arc::new(MockDatanodeManager::new(datanode_handler))); + context.region_failure_detector_controller = detector_controller.clone(); + + let table_id = 1024; + let region_id = RegionId::new(table_id, 1); + let mut procedure = UndropTableProcedure::new(UndropTableTask { table_id }, context); + procedure.data.table_name = Some(TableName::new( + DEFAULT_CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + "foo", + )); + procedure.data.table_route_value = Some(TableRouteValue::physical(vec![RegionRoute { + region: Region::new_test(region_id), + leader_peer: Some(Peer::empty(1)), + follower_peers: vec![], + leader_state: None, + leader_down_since: None, + write_route_policy: None, + }])); + + let err = procedure + .cleanup_opened_regions_after_restore_failure() + .await + .unwrap_err(); + + assert_eq!(StatusCode::Unexpected, err.status_code()); + assert_eq!( + detector_controller.deregistered.lock().await.as_slice(), + &[(1, region_id)] + ); + } +} diff --git a/src/common/meta/src/ddl_manager.rs b/src/common/meta/src/ddl_manager.rs index bfd858a196..3f421d3b02 100644 --- a/src/common/meta/src/ddl_manager.rs +++ b/src/common/meta/src/ddl_manager.rs @@ -480,7 +480,19 @@ impl DdlManager { undrop_table_task: UndropTableTask, ) -> Result<(ProcedureId, Option)> { let context = self.create_context(); - let procedure = UndropTableProcedure::new(undrop_table_task, context); + let original_table_name = context + .table_metadata_manager + .get_dropped_table_by_id(undrop_table_task.table_id) + .await? + .with_context(|| TableNotFoundSnafu { + table_name: undrop_table_task.table_id.to_string(), + })? + .table_name; + let procedure = UndropTableProcedure::new_with_original_table_name( + undrop_table_task, + context, + Some(original_table_name), + ); let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure)); self.execute_procedure_and_wait(procedure_with_id).await @@ -1204,7 +1216,8 @@ mod tests { use std::sync::Arc; use std::time::Duration; - use common_error::ext::BoxedError; + use common_error::ext::{BoxedError, ErrorExt}; + use common_error::status_code::StatusCode; use common_procedure::local::LocalManager; use common_procedure::test_util::InMemoryPoisonStore; use common_procedure::{BoxedProcedure, ProcedureManagerRef}; @@ -1228,6 +1241,7 @@ mod tests { use crate::peer::Peer; use crate::region_keeper::MemoryRegionKeeper; use crate::region_registry::LeaderRegionRegistry; + use crate::rpc::ddl::UndropTableTask; use crate::sequence::SequenceBuilder; use crate::state_store::KvStateStore; use crate::wal_provider::WalProvider; @@ -1326,4 +1340,55 @@ mod tests { assert!(procedure_manager.contains_loader(loader)); } } + + #[tokio::test] + async fn test_submit_undrop_missing_tombstone_returns_table_not_found_directly() { + let kv_backend = Arc::new(MemoryKvBackend::new()); + let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone())); + let table_metadata_allocator = Arc::new(TableMetadataAllocator::new( + Arc::new(SequenceBuilder::new("test", kv_backend.clone()).build()), + Arc::new(WalProvider::default()), + )); + let flow_metadata_manager = Arc::new(FlowMetadataManager::new(kv_backend.clone())); + let flow_metadata_allocator = Arc::new(FlowMetadataAllocator::with_noop_peer_allocator( + Arc::new(SequenceBuilder::new("flow-test", kv_backend.clone()).build()), + )); + + let state_store = Arc::new(KvStateStore::new(kv_backend.clone())); + let poison_manager = Arc::new(InMemoryPoisonStore::default()); + let procedure_manager = Arc::new(LocalManager::new( + Default::default(), + state_store, + poison_manager, + None, + None, + )); + + let ddl_manager = DdlManager::try_new( + DdlContext { + node_manager: Arc::new(DummyDatanodeManager), + cache_invalidator: Arc::new(DummyCacheInvalidator), + table_metadata_manager, + table_metadata_allocator, + flow_metadata_manager, + flow_metadata_allocator, + memory_region_keeper: Arc::new(MemoryRegionKeeper::default()), + leader_region_registry: Arc::new(LeaderRegionRegistry::default()), + region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl), + soft_drop_enabled: true, + }, + procedure_manager, + Arc::new(DummyRepartitionProcedureFactory), + true, + ) + .unwrap(); + + let err = ddl_manager + .submit_undrop_table_task(UndropTableTask { table_id: 1024 }) + .await + .unwrap_err(); + + assert_eq!(err.status_code(), StatusCode::TableNotFound); + assert!(matches!(err, crate::error::Error::TableNotFound { .. })); + } } diff --git a/src/datanode/src/region_server.rs b/src/datanode/src/region_server.rs index 3932e30748..5a6ab70085 100644 --- a/src/datanode/src/region_server.rs +++ b/src/datanode/src/region_server.rs @@ -1214,6 +1214,24 @@ impl RegionServerInner { })? .clone(), }, + RegionChange::OfflineCleanup(attribute) => match current_region_status { + Some(status) => match status.clone() { + RegionEngineWithStatus::Registering(_) + | RegionEngineWithStatus::Deregistering(_) + | RegionEngineWithStatus::Ready(_) => { + return error::RegionBusySnafu { region_id }.fail(); + } + }, + None => self + .engines + .read() + .unwrap() + .get(attribute.engine()) + .with_context(|| RegionEngineNotFoundSnafu { + name: attribute.engine(), + })? + .clone(), + }, RegionChange::Deregisters => match current_region_status { Some(status) => match status.clone() { RegionEngineWithStatus::Registering(_) => { @@ -1559,6 +1577,10 @@ impl RegionServerInner { let attribute = parse_region_attribute(&open.engine, &open.options)?; RegionChange::Register(attribute) } + RegionRequest::CleanUp(clean_up) => { + let attribute = parse_region_attribute(&clean_up.engine, &clean_up.options)?; + RegionChange::OfflineCleanup(attribute) + } RegionRequest::Close(_) | RegionRequest::Drop(_) => RegionChange::Deregisters, RegionRequest::Put(_) | RegionRequest::Delete(_) | RegionRequest::BulkInserts(_) => { RegionChange::Ingest @@ -1678,7 +1700,7 @@ impl RegionServerInner { region_change: RegionChange, ) { match region_change { - RegionChange::None | RegionChange::Ingest => {} + RegionChange::None | RegionChange::Ingest | RegionChange::OfflineCleanup(_) => {} RegionChange::Register(_) => { self.region_map.remove(®ion_id); } @@ -1698,7 +1720,7 @@ impl RegionServerInner { ) -> Result<()> { let engine_type = engine.name(); match region_change { - RegionChange::None | RegionChange::Ingest => {} + RegionChange::None | RegionChange::Ingest | RegionChange::OfflineCleanup(_) => {} RegionChange::Register(attribute) => { info!( "Region {region_id} is registered to engine {}", @@ -1883,6 +1905,7 @@ impl RegionServerInner { enum RegionChange { None, Register(RegionAttribute), + OfflineCleanup(RegionAttribute), Deregisters, Catchup, Ingest, @@ -1947,8 +1970,8 @@ mod tests { use store_api::metadata::{ColumnMetadata, RegionMetadata, RegionMetadataBuilder}; use store_api::region_engine::RegionEngine; use store_api::region_request::{ - PathType, RegionCompactRequest, RegionDeleteRequest, RegionDropRequest, RegionOpenRequest, - RegionPutRequest, RegionTruncateRequest, + PathType, RegionCleanUpRequest, RegionCompactRequest, RegionDeleteRequest, + RegionDropRequest, RegionOpenRequest, RegionPutRequest, RegionTruncateRequest, }; use store_api::storage::RegionId; @@ -2119,6 +2142,77 @@ mod tests { assert!(pinned.as_ref().get_ref().metrics().is_none()); } + #[tokio::test] + async fn test_offline_cleanup_does_not_register_region() { + let mut mock_region_server = mock_region_server(); + let (engine, mut receiver) = MockRegionEngine::new(MITO_ENGINE_NAME); + mock_region_server.register_engine(engine); + + let region_id = RegionId::new(1, 1); + let response = mock_region_server + .handle_request( + region_id, + RegionRequest::CleanUp(RegionCleanUpRequest { + engine: MITO_ENGINE_NAME.to_string(), + table_dir: String::new(), + path_type: PathType::Bare, + options: HashMap::new(), + }), + ) + .await + .unwrap(); + + assert_eq!(response.affected_rows, 0); + let (handled_region_id, handled_request) = receiver.try_recv().unwrap(); + assert_eq!(handled_region_id, region_id); + assert_matches!(handled_request, RegionRequest::CleanUp(_)); + assert!( + mock_region_server + .inner + .region_map + .get(®ion_id) + .is_none() + ); + } + + #[tokio::test] + async fn test_offline_cleanup_rejects_registered_region() { + let mut mock_region_server = mock_region_server(); + let (engine, mut receiver) = MockRegionEngine::new(MITO_ENGINE_NAME); + mock_region_server.register_engine(engine.clone()); + + let region_id = RegionId::new(1, 1); + mock_region_server + .inner + .region_map + .insert(region_id, RegionEngineWithStatus::Ready(engine)); + + let err = mock_region_server + .handle_request( + region_id, + RegionRequest::CleanUp(RegionCleanUpRequest { + engine: MITO_ENGINE_NAME.to_string(), + table_dir: String::new(), + path_type: PathType::Bare, + options: HashMap::new(), + }), + ) + .await + .unwrap_err(); + + assert_eq!(err.status_code(), StatusCode::RegionBusy); + assert!(receiver.try_recv().is_err()); + assert!(matches!( + mock_region_server + .inner + .region_map + .get(®ion_id) + .unwrap() + .clone(), + RegionEngineWithStatus::Ready(_) + )); + } + #[tokio::test] async fn test_region_registering() { common_telemetry::init_default_ut_logging(); @@ -2530,6 +2624,43 @@ mod tests { assert_matches!(current_engine, CurrentEngine::Engine(_)); }), }, + // RegionChange::OfflineCleanup + CurrentEngineTest { + region_id, + current_region_status: None, + region_change: RegionChange::OfflineCleanup(RegionAttribute::Mito), + assert: Box::new(|result| { + let current_engine = result.unwrap(); + assert_matches!(current_engine, CurrentEngine::Engine(_)); + }), + }, + CurrentEngineTest { + region_id, + current_region_status: Some(RegionEngineWithStatus::Registering(engine.clone())), + region_change: RegionChange::OfflineCleanup(RegionAttribute::Mito), + assert: Box::new(|result| { + let err = result.unwrap_err(); + assert_eq!(err.status_code(), StatusCode::RegionBusy); + }), + }, + CurrentEngineTest { + region_id, + current_region_status: Some(RegionEngineWithStatus::Deregistering(engine.clone())), + region_change: RegionChange::OfflineCleanup(RegionAttribute::Mito), + assert: Box::new(|result| { + let err = result.unwrap_err(); + assert_eq!(err.status_code(), StatusCode::RegionBusy); + }), + }, + CurrentEngineTest { + region_id, + current_region_status: Some(RegionEngineWithStatus::Ready(engine.clone())), + region_change: RegionChange::OfflineCleanup(RegionAttribute::Mito), + assert: Box::new(|result| { + let err = result.unwrap_err(); + assert_eq!(err.status_code(), StatusCode::RegionBusy); + }), + }, ]; for test in tests { diff --git a/src/metric-engine/src/engine.rs b/src/metric-engine/src/engine.rs index f8eceda9e8..f5c7863e82 100644 --- a/src/metric-engine/src/engine.rs +++ b/src/metric-engine/src/engine.rs @@ -237,6 +237,9 @@ impl RegionEngine for MetricEngine { } RegionRequest::Drop(drop) => self.inner.drop_region(region_id, drop).await, RegionRequest::Open(open) => self.inner.open_region(region_id, open).await, + RegionRequest::CleanUp(clean_up) => { + self.inner.clean_up_region(region_id, clean_up).await + } RegionRequest::Close(close) => self.inner.close_region(region_id, close).await, RegionRequest::Alter(alter) => { self.inner @@ -585,8 +588,8 @@ mod test { use store_api::metric_engine_consts::PHYSICAL_TABLE_METADATA_KEY; use store_api::mito_engine_options::WAL_OPTIONS_KEY; use store_api::region_request::{ - PathType, RegionCloseRequest, RegionDropRequest, RegionFlushRequest, RegionOpenRequest, - RegionRequest, + PathType, RegionCleanUpRequest, RegionCloseRequest, RegionDropRequest, RegionFlushRequest, + RegionOpenRequest, RegionRequest, }; use super::*; @@ -656,6 +659,58 @@ mod test { .unwrap(); } + #[tokio::test] + async fn test_offline_cleanup_physical_region() { + let env = TestEnv::new().await; + env.init_metric_region().await; + let engine = env.metric(); + let mito = env.mito(); + let physical_region_id = env.default_physical_region_id(); + let metadata_region_id = crate::utils::to_metadata_region_id(physical_region_id); + let data_region_id = crate::utils::to_data_region_id(physical_region_id); + + engine + .handle_request( + physical_region_id, + RegionRequest::Close(RegionCloseRequest::default()), + ) + .await + .unwrap(); + + let object_store = env.get_object_store().unwrap(); + let metadata_region_dir = region_dir_from_table_dir( + &TestEnv::default_table_dir(), + metadata_region_id, + PathType::Metadata, + ); + let data_region_dir = region_dir_from_table_dir( + &TestEnv::default_table_dir(), + data_region_id, + PathType::Data, + ); + assert!(object_store.exists(&metadata_region_dir).await.unwrap()); + assert!(object_store.exists(&data_region_dir).await.unwrap()); + + let physical_region_option = [(PHYSICAL_TABLE_METADATA_KEY.to_string(), String::new())] + .into_iter() + .collect(); + let clean_up_request = RegionCleanUpRequest { + engine: METRIC_ENGINE_NAME.to_string(), + table_dir: TestEnv::default_table_dir(), + path_type: PathType::Bare, + options: physical_region_option, + }; + engine + .handle_request(physical_region_id, RegionRequest::CleanUp(clean_up_request)) + .await + .unwrap(); + + assert!(!mito.is_region_exists(metadata_region_id)); + assert!(!mito.is_region_exists(data_region_id)); + assert!(!object_store.exists(&metadata_region_dir).await.unwrap()); + assert!(!object_store.exists(&data_region_dir).await.unwrap()); + } + #[tokio::test] async fn test_role() { let env = TestEnv::new().await; diff --git a/src/metric-engine/src/engine/open.rs b/src/metric-engine/src/engine/open.rs index 65ffcd77eb..54bbdfd703 100644 --- a/src/metric-engine/src/engine/open.rs +++ b/src/metric-engine/src/engine/open.rs @@ -22,14 +22,17 @@ use datafusion::common::HashMap; use mito2::engine::MITO_ENGINE_NAME; use snafu::{OptionExt, ResultExt}; use store_api::region_engine::{BatchResponses, RegionEngine}; -use store_api::region_request::{AffectedRows, PathType, RegionOpenRequest, ReplayCheckpoint}; +use store_api::region_request::{ + AffectedRows, PathType, RegionCleanUpRequest, RegionOpenRequest, RegionRequest, + ReplayCheckpoint, +}; use store_api::storage::RegionId; use crate::engine::MetricEngineInner; use crate::engine::create::region_options_for_metadata_region; use crate::engine::options::{PhysicalRegionOptions, set_data_region_options}; use crate::error::{ - BatchOpenMitoRegionSnafu, NoOpenRegionResultSnafu, OpenMitoRegionSnafu, + BatchOpenMitoRegionSnafu, CleanUpMitoRegionSnafu, NoOpenRegionResultSnafu, OpenMitoRegionSnafu, PhysicalRegionNotFoundSnafu, Result, }; use crate::metrics::{LOGICAL_REGION_COUNT, PHYSICAL_REGION_COUNT}; @@ -200,6 +203,94 @@ impl MetricEngineInner { } } + pub async fn clean_up_region( + &self, + region_id: RegionId, + request: RegionCleanUpRequest, + ) -> Result { + if request.is_physical_table() { + self.cleanup_physical_region_offline(region_id, request) + .await + } else { + Ok(0) + } + } + + async fn cleanup_physical_region_offline( + &self, + region_id: RegionId, + request: RegionCleanUpRequest, + ) -> Result { + let metadata_region_id = utils::to_metadata_region_id(region_id); + let data_region_id = utils::to_data_region_id(region_id); + let (clean_up_metadata_region_request, clean_up_data_region_request) = + self.transform_clean_up_physical_region_request(request); + let _ = self + .mito + .handle_request( + metadata_region_id, + RegionRequest::CleanUp(clean_up_metadata_region_request), + ) + .await + .context(CleanUpMitoRegionSnafu { + region_type: "metadata", + })?; + let data_region_response = self + .mito + .handle_request( + data_region_id, + RegionRequest::CleanUp(clean_up_data_region_request), + ) + .await + .context(CleanUpMitoRegionSnafu { + region_type: "data", + })?; + + if self.state.read().unwrap().exist_physical_region(region_id) { + self.state + .write() + .unwrap() + .remove_physical_region(region_id)?; + PHYSICAL_REGION_COUNT.dec(); + } + + Ok(data_region_response.affected_rows) + } + + /// Transform the cleanup request to metadata region and data region cleanup requests. + /// + /// Returns: + /// - The cleanup request for metadata region. + /// - The cleanup request for data region. + fn transform_clean_up_physical_region_request( + &self, + request: RegionCleanUpRequest, + ) -> (RegionCleanUpRequest, RegionCleanUpRequest) { + let clean_up_metadata_region_request = RegionCleanUpRequest { + table_dir: request.table_dir.clone(), + path_type: PathType::Metadata, + options: region_options_for_metadata_region(&request.options), + engine: MITO_ENGINE_NAME.to_string(), + }; + + let mut data_region_options = request.options; + set_data_region_options( + &mut data_region_options, + self.config.sparse_primary_key_encoding, + ); + let clean_up_data_region_request = RegionCleanUpRequest { + table_dir: request.table_dir, + path_type: PathType::Data, + options: data_region_options, + engine: MITO_ENGINE_NAME.to_string(), + }; + + ( + clean_up_metadata_region_request, + clean_up_data_region_request, + ) + } + /// Transform the open request to open metadata region and data region. /// /// Returns: diff --git a/src/metric-engine/src/error.rs b/src/metric-engine/src/error.rs index 59aee77b67..316588e6b3 100644 --- a/src/metric-engine/src/error.rs +++ b/src/metric-engine/src/error.rs @@ -43,6 +43,14 @@ pub enum Error { location: Location, }, + #[snafu(display("Failed to clean up mito region, region type: {}", region_type))] + CleanUpMitoRegion { + region_type: String, + source: BoxedError, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Failed to batch open mito region"))] BatchOpenMitoRegion { source: BoxedError, @@ -438,6 +446,7 @@ impl ErrorExt for Error { CreateMitoRegion { source, .. } | OpenMitoRegion { source, .. } + | CleanUpMitoRegion { source, .. } | CloseMitoRegion { source, .. } | MitoReadOperation { source, .. } | MitoWriteOperation { source, .. } @@ -471,6 +480,7 @@ impl ErrorExt for Error { match self { CreateMitoRegion { source, .. } | OpenMitoRegion { source, .. } + | CleanUpMitoRegion { source, .. } | BatchOpenMitoRegion { source, .. } | BatchCatchupMitoRegion { source, .. } | CloseMitoRegion { source, .. } diff --git a/src/mito2/src/engine/open_test.rs b/src/mito2/src/engine/open_test.rs index 2b709a7f08..f5ee4dd19f 100644 --- a/src/mito2/src/engine/open_test.rs +++ b/src/mito2/src/engine/open_test.rs @@ -23,9 +23,12 @@ use common_error::ext::ErrorExt; use common_error::status_code::StatusCode; use common_recordbatch::RecordBatches; use either::Either; +use store_api::logstore::LogStore; +use store_api::logstore::provider::Provider; use store_api::region_engine::{RegionEngine, RegionRole, SettableRegionRoleState}; use store_api::region_request::{ - PathType, RegionCloseRequest, RegionOpenRequest, RegionPutRequest, RegionRequest, + PathType, RegionCleanUpRequest, RegionCloseRequest, RegionOpenRequest, RegionPutRequest, + RegionRequest, }; use store_api::storage::{RegionId, ScanRequest}; use tokio::sync::oneshot; @@ -37,8 +40,10 @@ use crate::engine::region_hook::RegionHookRef; use crate::error; use crate::region::opener::{PartitionExprFetcher, PartitionExprFetcherRef}; use crate::region::options::RegionOptions; +use crate::sst::location::region_dir_from_table_dir; use crate::test_util::{ - CreateRequestBuilder, TestEnv, build_rows, flush_region, put_rows, reopen_region, rows_schema, + CreateRequestBuilder, LogStoreImpl, TestEnv, build_rows, flush_region, put_rows, reopen_region, + rows_schema, }; #[tokio::test] @@ -81,6 +86,103 @@ async fn test_engine_open_empty_with_format(flat_format: bool) { assert_eq!(role, None); } +#[tokio::test] +async fn test_engine_offline_cleanup_closed_region() { + let mut env = TestEnv::with_prefix("offline-cleanup").await; + let engine = env.create_engine(MitoConfig::default()).await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new().build(); + let table_dir = request.table_dir.clone(); + let path_type = request.path_type; + let options = request.options.clone(); + let region_dir = region_dir_from_table_dir(&table_dir, region_id, path_type); + let object_store = env.get_object_store().unwrap(); + + engine + .handle_request(region_id, RegionRequest::Create(request.clone())) + .await + .unwrap(); + assert!(object_store.exists(®ion_dir).await.unwrap()); + + put_rows( + &engine, + region_id, + Rows { + schema: rows_schema(&request), + rows: build_rows(0, 3), + }, + ) + .await; + let Some(LogStoreImpl::RaftEngine(log_store)) = env.get_log_store() else { + unreachable!() + }; + let provider = Provider::raft_engine_provider(region_id.as_u64()); + assert!(log_store.latest_entry_id(&provider).unwrap() > 0); + + engine + .handle_request( + region_id, + RegionRequest::Close(RegionCloseRequest::default()), + ) + .await + .unwrap(); + assert!(!engine.is_region_exists(region_id)); + + let cleanup_request = RegionCleanUpRequest { + engine: String::new(), + table_dir, + path_type, + options, + }; + for _ in 0..2 { + engine + .handle_request(region_id, RegionRequest::CleanUp(cleanup_request.clone())) + .await + .unwrap(); + } + + assert!(!engine.is_region_exists(region_id)); + assert!(!object_store.exists(®ion_dir).await.unwrap()); + assert_eq!(0, log_store.latest_entry_id(&provider).unwrap()); +} + +#[tokio::test] +async fn test_engine_offline_cleanup_rejects_opened_region() { + let mut env = TestEnv::with_prefix("offline-cleanup-opened").await; + let engine = env.create_engine(MitoConfig::default()).await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new().build(); + let table_dir = request.table_dir.clone(); + let path_type = request.path_type; + let options = request.options.clone(); + let region_dir = region_dir_from_table_dir(&table_dir, region_id, path_type); + let object_store = env.get_object_store().unwrap(); + + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + let err = engine + .handle_request( + region_id, + RegionRequest::CleanUp(RegionCleanUpRequest { + engine: String::new(), + table_dir, + path_type, + options, + }), + ) + .await + .unwrap_err(); + + assert_eq!(StatusCode::RegionBusy, err.status_code()); + assert!(engine.is_region_exists(region_id)); + assert!(object_store.exists(®ion_dir).await.unwrap()); +} + #[tokio::test] async fn test_engine_open_existing() { test_engine_open_existing_with_format(false).await; diff --git a/src/mito2/src/region/opener.rs b/src/mito2/src/region/opener.rs index 2d5cdcb284..ad59e4fb94 100644 --- a/src/mito2/src/region/opener.rs +++ b/src/mito2/src/region/opener.rs @@ -436,31 +436,7 @@ impl RegionOpener { } fn provider(&self, wal_options: &WalOptions) -> Result { - match wal_options { - WalOptions::RaftEngine => { - ensure!( - TypeId::of::() == TypeId::of::() - || TypeId::of::() == TypeId::of::(), - error::IncompatibleWalProviderChangeSnafu { - global: "`kafka`", - region: "`raft_engine`", - } - ); - Ok(Provider::raft_engine_provider(self.region_id.as_u64())) - } - WalOptions::Kafka(options) => { - ensure!( - TypeId::of::() == TypeId::of::() - || TypeId::of::() == TypeId::of::(), - error::IncompatibleWalProviderChangeSnafu { - global: "`raft_engine`", - region: "`kafka`", - } - ); - Ok(Provider::kafka_provider(options.topic.clone())) - } - WalOptions::Noop => Ok(Provider::noop_provider()), - } + provider_from_wal_options::(self.region_id, wal_options) } /// Tries to open the region and returns `None` if the region directory is empty. @@ -656,6 +632,37 @@ impl RegionOpener { } } +pub(crate) fn provider_from_wal_options( + region_id: RegionId, + wal_options: &WalOptions, +) -> Result { + match wal_options { + WalOptions::RaftEngine => { + ensure!( + TypeId::of::() == TypeId::of::() + || TypeId::of::() == TypeId::of::(), + error::IncompatibleWalProviderChangeSnafu { + global: "`kafka`", + region: "`raft_engine`", + } + ); + Ok(Provider::raft_engine_provider(region_id.as_u64())) + } + WalOptions::Kafka(options) => { + ensure!( + TypeId::of::() == TypeId::of::() + || TypeId::of::() == TypeId::of::(), + error::IncompatibleWalProviderChangeSnafu { + global: "`raft_engine`", + region: "`kafka`", + } + ); + Ok(Provider::kafka_provider(options.topic.clone())) + } + WalOptions::Noop => Ok(Provider::noop_provider()), + } +} + #[cfg(not(feature = "test-shared-fs-region-migration"))] fn supports_open_region_object_storage_requirement(object_store: &ObjectStore) -> bool { is_object_storage(object_store) diff --git a/src/mito2/src/request.rs b/src/mito2/src/request.rs index e9df654f83..3fdd15db4f 100644 --- a/src/mito2/src/request.rs +++ b/src/mito2/src/request.rs @@ -38,9 +38,10 @@ use store_api::region_engine::{ }; use store_api::region_request::{ AffectedRows, ApplyStagingManifestRequest, EnterStagingRequest, RegionAlterRequest, - RegionBuildIndexRequest, RegionBulkInsertsRequest, RegionCatchupRequest, RegionCloseRequest, - RegionCompactRequest, RegionCreateRequest, RegionDropRequest, RegionFlushRequest, - RegionOpenRequest, RegionRequest, RegionTruncateRequest, StagingPartitionDirective, + RegionBuildIndexRequest, RegionBulkInsertsRequest, RegionCatchupRequest, RegionCleanUpRequest, + RegionCloseRequest, RegionCompactRequest, RegionCreateRequest, RegionDropRequest, + RegionFlushRequest, RegionOpenRequest, RegionRequest, RegionTruncateRequest, + StagingPartitionDirective, }; use store_api::storage::{FileId, RegionId}; use tokio::sync::oneshot::{self, Receiver, Sender}; @@ -711,6 +712,11 @@ impl WorkerRequest { sender: sender.into(), request: DdlRequest::Open((v, None)), }), + RegionRequest::CleanUp(v) => WorkerRequest::Ddl(SenderDdlRequest { + region_id, + sender: sender.into(), + request: DdlRequest::OfflineCleanup(v), + }), RegionRequest::Close(v) => WorkerRequest::Ddl(SenderDdlRequest { region_id, sender: sender.into(), @@ -863,6 +869,7 @@ pub(crate) enum DdlRequest { Create(RegionCreateRequest), Drop(RegionDropRequest), Open((RegionOpenRequest, Option)), + OfflineCleanup(RegionCleanUpRequest), Close(RegionCloseRequest), Alter(RegionAlterRequest), Flush(RegionFlushRequest), diff --git a/src/mito2/src/wal.rs b/src/mito2/src/wal.rs index eb9d4a251b..316cbbd612 100644 --- a/src/mito2/src/wal.rs +++ b/src/mito2/src/wal.rs @@ -159,6 +159,19 @@ impl Wal { .map_err(BoxedError::new) .context(DeleteWalSnafu { region_id }) } + + /// Deletes all WAL entries in the namespace represented by `provider`. + pub async fn delete_namespace(&self, region_id: RegionId, provider: &Provider) -> Result<()> { + if let Provider::Noop = provider { + return Ok(()); + } + self.store + .delete_namespace(provider) + .await + .map_err(BoxedError::new) + .context(DeleteWalSnafu { region_id }) + } + /// Marks all WAL entries of a region as obsolete and removes its dedicated namespace when /// supported by the backend. pub async fn obsolete_all(&self, region_id: RegionId, provider: &Provider) -> Result<()> { diff --git a/src/mito2/src/worker.rs b/src/mito2/src/worker.rs index 3cd4a402c8..23b6dd4d7b 100644 --- a/src/mito2/src/worker.rs +++ b/src/mito2/src/worker.rs @@ -1148,6 +1148,10 @@ impl RegionWorkerLoop { .await; continue; } + DdlRequest::OfflineCleanup(req) => { + self.handle_offline_cleanup_request(ddl.region_id, req) + .await + } DdlRequest::Close(req) => { self.handle_close_request(ddl.region_id, req, ddl.sender) .await; diff --git a/src/mito2/src/worker/handle_drop.rs b/src/mito2/src/worker/handle_drop.rs index 3d111776b8..1efa7cb99e 100644 --- a/src/mito2/src/worker/handle_drop.rs +++ b/src/mito2/src/worker/handle_drop.rs @@ -28,9 +28,11 @@ use store_api::region_request::{AffectedRows, PathType}; use store_api::storage::RegionId; use tokio::time::sleep; +use crate::cache::CacheManagerRef; use crate::engine::region_hook::RegionHookRef; use crate::error::{OpenDalSnafu, Result}; use crate::region::{RegionLeaderState, RegionMapRef}; +use crate::sst::index::intermediate::IntermediateManager; use crate::worker::{DROPPING_MARKER_FILE, RegionWorkerLoop}; const GC_TASK_INTERVAL_SEC: u64 = 5 * 60; // 5 minutes @@ -90,14 +92,7 @@ where ®ion.provider, ) .await?; - // Notifies flush scheduler. - self.flush_scheduler.on_region_dropped(region_id); - // Notifies compaction scheduler. - self.compaction_scheduler.on_region_dropped(region_id); - // notifies index build scheduler. - self.index_build_scheduler - .on_region_dropped(region_id) - .await; + self.cleanup_dropped_region_runtime_state(region_id).await; // Marks region version as dropped region.version_control.mark_dropped(); @@ -159,21 +154,57 @@ where .await }; - if let Err(err) = intm_manager.prune_region_dir(®ion_id).await { - warn!(err; "Failed to prune intermediate region directory, region_id: {}", region_id); - } - - if let Some(write_cache) = cache_manager.write_cache() - && let Some(manifest_cache) = write_cache.manifest_cache() - { - manifest_cache.clean_manifests(&table_dir).await; - } + cleanup_region_file_artifacts(region_id, &table_dir, &intm_manager, &cache_manager) + .await; listener.on_later_drop_end(region_id, removed); }); Ok(0) } + + /// Cleans runtime state for a region that is no longer available to serve requests. + pub(crate) async fn cleanup_dropped_region_runtime_state(&mut self, region_id: RegionId) { + // Notifies flush scheduler. + self.flush_scheduler.on_region_dropped(region_id); + // Notifies compaction scheduler. + self.compaction_scheduler.on_region_dropped(region_id); + // Notifies index build scheduler. + self.index_build_scheduler + .on_region_dropped(region_id) + .await; + } +} + +/// Cleans files and caches that are produced at runtime but are not part of the +/// primary region directory deletion. +pub(crate) async fn cleanup_region_file_artifacts( + region_id: RegionId, + table_dir: &str, + intermediate_manager: &IntermediateManager, + cache_manager: &CacheManagerRef, +) { + if let Err(err) = intermediate_manager.prune_region_dir(®ion_id).await { + warn!(err; "Failed to prune intermediate region directory, region_id: {}", region_id); + } + + if let Some(write_cache) = cache_manager.write_cache() + && let Some(manifest_cache) = write_cache.manifest_cache() + { + manifest_cache.clean_manifests(table_dir).await; + } +} + +/// Removes a region directory for full-drop style cleanup. +/// +/// Full drop and purge/offline cleanup force physical deletion. Only partial +/// drop may leave data files for global GC. +pub(crate) async fn remove_region_dir_for_full_drop( + region_path: &str, + object_store: &ObjectStore, +) -> Result<()> { + remove_region_dir_once(region_path, object_store, true).await?; + Ok(()) } /// Carries the region hook and region metadata into the background GC task so @@ -281,7 +312,7 @@ async fn later_drop_task_with_global_gc( // the region directory is forcefully removed immediately. // // TODO(discord9): Evaluate removing files instantly rather than waiting for the GC period. - if path_type == PathType::Metadata || !partial_drop { + if should_force_remove_region_dir(path_type, partial_drop) { remove_region_with_retry( region_id, region_path, @@ -299,6 +330,10 @@ async fn later_drop_task_with_global_gc( } } +fn should_force_remove_region_dir(path_type: PathType, partial_drop: bool) -> bool { + path_type == PathType::Metadata || !partial_drop +} + // TODO(ruihang): place the marker in a separate dir /// Removes region dir if there is no parquet files, returns whether the directory is removed. /// If `force = true`, always removes the dir. diff --git a/src/mito2/src/worker/handle_open.rs b/src/mito2/src/worker/handle_open.rs index 48fef5ddd6..802ab9c1a8 100644 --- a/src/mito2/src/worker/handle_open.rs +++ b/src/mito2/src/worker/handle_open.rs @@ -18,24 +18,68 @@ use std::sync::Arc; use std::time::Instant; use common_telemetry::info; -use object_store::util::join_path; +use object_store::util::{join_path, normalize_dir}; use snafu::{OptionExt, ResultExt}; use store_api::logstore::LogStore; -use store_api::region_request::RegionOpenRequest; +use store_api::region_request::{AffectedRows, RegionCleanUpRequest, RegionOpenRequest}; use store_api::storage::RegionId; use table::requests::STORAGE_KEY; use crate::error::{ - ObjectStoreNotFoundSnafu, OpenDalSnafu, OpenRegionSnafu, RegionNotFoundSnafu, Result, + ObjectStoreNotFoundSnafu, OpenDalSnafu, OpenRegionSnafu, RegionBusySnafu, RegionNotFoundSnafu, + Result, }; -use crate::region::opener::{RegionOpener, sanitize_open_request_options}; +use crate::region::opener::{ + RegionOpener, get_object_store, provider_from_wal_options, sanitize_open_request_options, +}; +use crate::region::options::RegionOptions; use crate::request::OptionOutputTx; use crate::sst::location::region_dir_from_table_dir; use crate::wal::entry_distributor::WalEntryReceiver; -use crate::worker::handle_drop::remove_region_dir_once; +use crate::worker::handle_drop::{ + cleanup_region_file_artifacts, remove_region_dir_for_full_drop, remove_region_dir_once, +}; use crate::worker::{DROPPING_MARKER_FILE, RegionWorkerLoop}; impl RegionWorkerLoop { + pub(crate) async fn handle_offline_cleanup_request( + &mut self, + region_id: RegionId, + mut request: RegionCleanUpRequest, + ) -> Result { + info!( + "Try to clean region {} offline, worker: {}", + region_id, self.id + ); + + if self.regions.is_region_exists(region_id) { + return RegionBusySnafu { region_id }.fail(); + } + + sanitize_open_request_options(&mut request.options); + + let options = RegionOptions::try_from_options(region_id, &request.options)?; + let object_store = get_object_store(&options.storage, &self.object_store_manager)?; + let provider = provider_from_wal_options::(region_id, &options.wal_options)?; + self.wal.obsolete_all(region_id, &provider).await?; + + let table_dir = normalize_dir(&request.table_dir); + let region_dir = region_dir_from_table_dir(&table_dir, region_id, request.path_type); + remove_region_dir_for_full_drop(®ion_dir, &object_store).await?; + + self.cleanup_dropped_region_runtime_state(region_id).await; + self.dropping_regions.remove_region(region_id); + cleanup_region_file_artifacts( + region_id, + &table_dir, + &self.intermediate_manager, + &self.cache_manager, + ) + .await; + + Ok(0) + } + async fn check_and_cleanup_region( &self, region_id: RegionId, diff --git a/src/store-api/src/region_request.rs b/src/store-api/src/region_request.rs index 61b353956b..6aa3e3f49d 100644 --- a/src/store-api/src/region_request.rs +++ b/src/store-api/src/region_request.rs @@ -23,10 +23,10 @@ use api::v1::column_def::{ }; use api::v1::region::bulk_insert_request::Body; use api::v1::region::{ - AlterRequest, AlterRequests, BuildIndexRequest, BulkInsertRequest, CloseRequest, - CompactRequest, CreateRequest, CreateRequests, DeleteRequests, DropRequest, DropRequests, - FlushRequest, InsertRequests, OpenRequest, TruncateRequest, alter_request, compact_request, - region_request, truncate_request, + AlterRequest, AlterRequests, BuildIndexRequest, BulkInsertRequest, + CleanUpRequest as PbCleanUpRequest, CloseRequest, CompactRequest, CreateRequest, + CreateRequests, DeleteRequests, DropRequest, DropRequests, FlushRequest, InsertRequests, + OpenRequest, TruncateRequest, alter_request, compact_request, region_request, truncate_request, }; use api::v1::{ self, Analyzer, ArrowIpc, FulltextBackend as PbFulltextBackend, Option as PbOption, Rows, @@ -145,6 +145,7 @@ pub enum RegionRequest { Create(RegionCreateRequest), Drop(RegionDropRequest), Open(RegionOpenRequest), + CleanUp(RegionCleanUpRequest), Close(RegionCloseRequest), Alter(RegionAlterRequest), Flush(RegionFlushRequest), @@ -167,6 +168,7 @@ impl RegionRequest { region_request::Body::Create(create) => make_region_create(create), region_request::Body::Drop(drop) => make_region_drop(drop), region_request::Body::Open(open) => make_region_open(open), + region_request::Body::CleanUp(clean_up) => make_region_clean_up(clean_up), region_request::Body::Close(close) => make_region_close(close), region_request::Body::Alter(alter) => make_region_alter(alter), region_request::Body::Flush(flush) => make_region_flush(flush), @@ -189,10 +191,6 @@ impl RegionRequest { reason: "RemoteDynFilter request should be handled separately by RegionServer", } .fail(), - region_request::Body::CleanUp(_) => UnexpectedSnafu { - reason: "CleanUp request should be handled separately by RegionServer", - } - .fail(), region_request::Body::ApplyStagingManifest(apply) => { make_region_apply_staging_manifest(apply) } @@ -329,6 +327,20 @@ fn make_region_open(open: OpenRequest) -> Result> )]) } +fn make_region_clean_up(clean_up: PbCleanUpRequest) -> Result> { + let region_id = RegionId::from(clean_up.region_id); + let table_dir = table_dir(&clean_up.path, region_id.table_id()); + Ok(vec![( + region_id, + RegionRequest::CleanUp(RegionCleanUpRequest { + engine: clean_up.engine, + table_dir, + path_type: PathType::Bare, + options: clean_up.options, + }), + )]) +} + fn make_region_close(close: CloseRequest) -> Result> { let region_id = close.region_id.into(); Ok(vec![( @@ -652,6 +664,26 @@ impl RegionOpenRequest { } } +/// Offline region cleanup request. +#[derive(Debug, Clone)] +pub struct RegionCleanUpRequest { + /// Region engine name + pub engine: String, + /// Directory for table's data home. Usually is composed by catalog and table id + pub table_dir: String, + /// Path type for generating paths + pub path_type: PathType, + /// Options of the cleaned region. + pub options: HashMap, +} + +impl RegionCleanUpRequest { + /// Returns true when the region belongs to the metric engine's physical table. + pub fn is_physical_table(&self) -> bool { + self.options.contains_key(PHYSICAL_TABLE_METADATA_KEY) + } +} + /// Close region request. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct RegionCloseRequest { @@ -1642,6 +1674,7 @@ impl fmt::Display for RegionRequest { RegionRequest::Create(_) => write!(f, "Create"), RegionRequest::Drop(_) => write!(f, "Drop"), RegionRequest::Open(_) => write!(f, "Open"), + RegionRequest::CleanUp(_) => write!(f, "CleanUp"), RegionRequest::Close(_) => write!(f, "Close"), RegionRequest::Alter(_) => write!(f, "Alter"), RegionRequest::Flush(_) => write!(f, "Flush"), @@ -2203,6 +2236,28 @@ mod tests { assert_eq!(request.requirements, RegionRequirements::object_storage()); } + #[test] + fn test_parse_region_cleanup_from_proto() { + let clean_up = api::v1::region::CleanUpRequest { + region_id: RegionId::new(42, 3).as_u64(), + engine: "mito".to_string(), + path: "test".to_string(), + options: HashMap::from([("k".to_string(), "v".to_string())]), + }; + + let requests = + RegionRequest::try_from_request_body(region_request::Body::CleanUp(clean_up)).unwrap(); + let RegionRequest::CleanUp(request) = &requests[0].1 else { + unreachable!() + }; + + assert_eq!(requests[0].0, RegionId::new(42, 3)); + assert_eq!(request.engine, "mito"); + assert_eq!(request.table_dir, "data/test/42/"); + assert_eq!(request.path_type, PathType::Bare); + assert_eq!(request.options.get("k"), Some(&"v".to_string())); + } + #[test] fn test_validate_modify_column_fulltext_options() { let kind = AlterKind::SetIndexes {