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

feat: clean up soft-dropped regions offline (#8458)

Browse Source 
* 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 <ratuthomm@gmail.com>

* 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 <ratuthomm@gmail.com>

* 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 <ratuthomm@gmail.com>

* 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 <ratuthomm@gmail.com>

* fix(meta): require tombstone before undrop

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(meta): reject file-engine soft drop

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* 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 <ratuthomm@gmail.com>

* 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 <ratuthomm@gmail.com>

* 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 <ratuthomm@gmail.com>

* chore: preserve soft-drop cleanup split state

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* 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 <ratuthomm@gmail.com>

* fix(meta): clean every soft-dropped region replica

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(meta): order soft-drop replica cleanup

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* Revert "fix(meta): order soft-drop replica cleanup"

This reverts commit e77162d3e5ebcf2817e2845a6a5177c328fb2c60.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* Revert "fix(meta): clean every soft-dropped region replica"

This reverts commit 2378e00cc258ca1b6a85a1aafbd68c79c666f43c.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

---------

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
This commit is contained in:
Lei, HUANG
2026-07-14 16:39:36 +08:00
committed by GitHub
parent 0beeef7fc0
commit 81d8beee3a
18 changed files with 1229 additions and 267 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/meta/src/ddl/drop_table/executor.rs
+75 -5
View File
@@ -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<RegionNumber, WalOptions>,
) -> 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::<Vec<_>>();
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::<Result<Vec<_>>>()?;
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(
src/common/meta/src/ddl/drop_table/metadata.rs
+7
View File
@@ -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;
src/common/meta/src/ddl/purge_dropped_table.rs
+7 -27
View File
@@ -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<Status> {
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<Status> {
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<Status> {
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,
}
src/common/meta/src/ddl/tests/drop_table.rs
+209 -120
View File
@@ -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::<Vec<_>>();
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::<RegionResponse, _>(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<String>) {}
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));
src/common/meta/src/ddl/undrop_table.rs
+241 -44
View File
@@ -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<TableName>,
) -> Self {
let mut data = UndropTableData::new(task);
data.table_name = table_name;
Self { context, data }
}
pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult<Self> {
@@ -108,29 +115,101 @@ impl UndropTableProcedure {
}
async fn on_restore_metadata(&mut self) -> Result<Status> {
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,
&region_routes,
false,
)
.await
} else {
Ok(())
};
self.context
.deregister_failure_detectors(convert_region_routes_to_detecting_regions(
&region_routes,
))
.await;
close_result?;
Ok(())
}
async fn on_open_regions(&mut self) -> Result<Status> {
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<RegionNumber, WalOptions>,
) -> 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<RegionNumber, WalOptions>,
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<Vec<DetectingRegion>>,
}
#[async_trait]
impl RegionFailureDetectorController for RecordingRegionFailureDetectorController {
async fn register_failure_detectors(&self, _detecting_regions: Vec<DetectingRegion>) {}
async fn reset_failure_detectors(&self, _detecting_regions: Vec<DetectingRegion>) {}
async fn deregister_failure_detectors(&self, detecting_regions: Vec<DetectingRegion>) {
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)]
);
}
}
src/common/meta/src/ddl_manager.rs
+67 -2
View File
@@ -480,7 +480,19 @@ impl DdlManager {
undrop_table_task: UndropTableTask,
) -> Result<(ProcedureId, Option<Output>)> {
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 { .. }));
}
}
src/datanode/src/region_server.rs
+135 -4
View File
@@ -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(&region_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(&region_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(&region_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 {
src/metric-engine/src/engine.rs
+57 -2
View File
@@ -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;
src/metric-engine/src/engine/open.rs
+93 -2
View File
@@ -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<AffectedRows> {
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<AffectedRows> {
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:
src/metric-engine/src/error.rs
+10
View File
@@ -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, .. }
src/mito2/src/engine/open_test.rs
+104 -2
View File
@@ -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(&region_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(&region_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(&region_dir).await.unwrap());
}
#[tokio::test]
async fn test_engine_open_existing() {
test_engine_open_existing_with_format(false).await;
src/mito2/src/region/opener.rs
+32 -25
View File
@@ -436,31 +436,7 @@ impl RegionOpener {
}
fn provider<S: LogStore>(&self, wal_options: &WalOptions) -> Result<Provider> {
match wal_options {
WalOptions::RaftEngine => {
ensure!(
TypeId::of::<RaftEngineLogStore>() == TypeId::of::<S>()
|| TypeId::of::<NoopLogStore>() == TypeId::of::<S>(),
error::IncompatibleWalProviderChangeSnafu {
global: "`kafka`",
region: "`raft_engine`",
}
);
Ok(Provider::raft_engine_provider(self.region_id.as_u64()))
}
WalOptions::Kafka(options) => {
ensure!(
TypeId::of::<KafkaLogStore>() == TypeId::of::<S>()
|| TypeId::of::<NoopLogStore>() == TypeId::of::<S>(),
error::IncompatibleWalProviderChangeSnafu {
global: "`raft_engine`",
region: "`kafka`",
}
);
Ok(Provider::kafka_provider(options.topic.clone()))
}
WalOptions::Noop => Ok(Provider::noop_provider()),
}
provider_from_wal_options::<S>(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<S: LogStore>(
region_id: RegionId,
wal_options: &WalOptions,
) -> Result<Provider> {
match wal_options {
WalOptions::RaftEngine => {
ensure!(
TypeId::of::<RaftEngineLogStore>() == TypeId::of::<S>()
|| TypeId::of::<NoopLogStore>() == TypeId::of::<S>(),
error::IncompatibleWalProviderChangeSnafu {
global: "`kafka`",
region: "`raft_engine`",
}
);
Ok(Provider::raft_engine_provider(region_id.as_u64()))
}
WalOptions::Kafka(options) => {
ensure!(
TypeId::of::<KafkaLogStore>() == TypeId::of::<S>()
|| TypeId::of::<NoopLogStore>() == TypeId::of::<S>(),
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)
src/mito2/src/request.rs
+10 -3
View File
@@ -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<WalEntryReceiver>)),
OfflineCleanup(RegionCleanUpRequest),
Close(RegionCloseRequest),
Alter(RegionAlterRequest),
Flush(RegionFlushRequest),
src/mito2/src/wal.rs
+13
View File
@@ -159,6 +159,19 @@ impl<S: LogStore> Wal<S> {
.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<()> {
src/mito2/src/worker.rs
+4
View File
@@ -1148,6 +1148,10 @@ impl<S: LogStore> RegionWorkerLoop<S> {
.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;
src/mito2/src/worker/handle_drop.rs
+53 -18
View File
@@ -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
&region.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(&region_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(&region_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.
src/mito2/src/worker/handle_open.rs
+49 -5
View File
@@ -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<S: LogStore> RegionWorkerLoop<S> {
pub(crate) async fn handle_offline_cleanup_request(
&mut self,
region_id: RegionId,
mut request: RegionCleanUpRequest,
) -> Result<AffectedRows> {
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::<S>(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(&region_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,
src/store-api/src/region_request.rs
+63 -8
View File
@@ -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<Vec<(RegionId, RegionRequest)>>
)])
}
fn make_region_clean_up(clean_up: PbCleanUpRequest) -> Result<Vec<(RegionId, RegionRequest)>> {
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<Vec<(RegionId, RegionRequest)>> {
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<String, String>,
}
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 {