rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-06-01 21:00:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: gc worker on dropped region (#7537)

Browse Source 
* feat: allow clean up for dropped region

Signed-off-by: discord9 <discord9@163.com>

* clippy

Signed-off-by: discord9 <discord9@163.com>

* pcr

Signed-off-by: discord9 <discord9@163.com>

* fix: get access layer correct

Signed-off-by: discord9 <discord9@163.com>

* chore: invalid gc args

Signed-off-by: discord9 <discord9@163.com>

* chore: fix test

Signed-off-by: discord9 <discord9@163.com>

* feat: more defend check

Signed-off-by: discord9 <discord9@163.com>

* per review

Signed-off-by: discord9 <discord9@163.com>

* feat: messy impl of drop region

Signed-off-by: discord9 <discord9@163.com>

* feat: add dropped region GC handling module and integrate with GcScheduler

Signed-off-by: discord9 <discord9@163.com>

* refactor: simplify access layer creation

Signed-off-by: discord9 <discord9@163.com>

* c

Signed-off-by: discord9 <discord9@163.com>

* fix: path type

Signed-off-by: discord9 <discord9@163.com>

* feat: gc handle drop

Signed-off-by: discord9 <discord9@163.com>

* chore: use proper const

Signed-off-by: discord9 <discord9@163.com>

* fix: recursive list when check empty dir

Signed-off-by: discord9 <discord9@163.com>

* per review

Signed-off-by: discord9 <discord9@163.com>

* refactor: with gc only delete if metadata region

Signed-off-by: discord9 <discord9@163.com>

* feat: add batch_get_table_route method to SchedulerCtx and MockSchedulerCtx

Signed-off-by: discord9 <discord9@163.com>

* chore: comment

Signed-off-by: discord9 <discord9@163.com>

* refactor: retry delete method

Signed-off-by: discord9 <discord9@163.com>

---------

Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
discord9
2026-01-20 19:45:37 +08:00
committed by GitHub
parent 25687bb282
commit 67e51b4573
36 changed files with 1277 additions and 245 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/datanode/Cargo.toml
View File

@@ -18,6 +18,7 @@ async-trait.workspace = true
bytes.workspace = true
client.workspace = true
common-base.workspace = true
common-catalog.workspace = true
common-config.workspace = true
common-error.workspace = true
common-function.workspace = true

15
src/datanode/src/datanode.rs
View File

@@ -244,7 +244,19 @@ impl DatanodeBuilder {
table_id_schema_cache,
schema_cache,
));
let file_ref_manager = Arc::new(FileReferenceManager::new(Some(node_id)));
let gc_enabled = self.opts.region_engine.iter().any(|engine| {
if let RegionEngineConfig::Mito(config) = engine {
config.gc.enable
} else {
false
}
});
let file_ref_manager = Arc::new(FileReferenceManager::with_gc_enabled(
Some(node_id),
gc_enabled,
));
let region_server = self
.new_region_server(
schema_metadata_manager,
@@ -331,6 +343,7 @@ impl DatanodeBuilder {
&self.opts,
region_server.clone(),
meta_client,
self.kv_backend.clone(),
cache_invalidator,
self.plugins.clone(),
stat,

15
src/datanode/src/error.rs
View File

@@ -332,13 +332,6 @@ pub enum Error {
location: Location,
},
#[snafu(display("Invalid arguments for GC: {}", msg))]
InvalidGcArgs {
msg: String,
#[snafu(implicit)]
location: Location,
},
#[snafu(display("Failed to list SST entries from storage"))]
ListStorageSsts {
#[snafu(implicit)]
@@ -455,11 +448,9 @@ impl ErrorExt for Error {
AsyncTaskExecute { source, .. } => source.status_code(),
CreateDir { .. }
| RemoveDir { .. }
| ShutdownInstance { .. }
| DataFusion { .. }
| InvalidGcArgs { .. } => StatusCode::Internal,
CreateDir { .. } | RemoveDir { .. } | ShutdownInstance { .. } | DataFusion { .. } => {
StatusCode::Internal
}
RegionNotFound { .. } => StatusCode::RegionNotFound,
RegionNotReady { .. } => StatusCode::RegionNotReady,

4
src/datanode/src/heartbeat.rs
View File

@@ -31,6 +31,7 @@ use common_meta::heartbeat::handler::{
};
use common_meta::heartbeat::mailbox::{HeartbeatMailbox, MailboxRef};
use common_meta::heartbeat::utils::outgoing_message_to_mailbox_message;
use common_meta::kv_backend::KvBackendRef;
use common_stat::ResourceStatRef;
use common_telemetry::{debug, error, info, trace, warn};
use common_workload::DatanodeWorkloadType;
@@ -79,6 +80,7 @@ impl HeartbeatTask {
opts: &DatanodeOptions,
region_server: RegionServer,
meta_client: MetaClientRef,
kv_backend: KvBackendRef,
cache_invalidator: CacheInvalidatorRef,
plugins: Plugins,
resource_stat: ResourceStatRef,
@@ -94,7 +96,7 @@ impl HeartbeatTask {
Arc::new(ParseMailboxMessageHandler),
Arc::new(SuspendHandler::new(region_server.suspend_state())),
Arc::new(
RegionHeartbeatResponseHandler::new(region_server.clone())
RegionHeartbeatResponseHandler::new(region_server.clone(), kv_backend)
.with_open_region_parallelism(opts.init_regions_parallelism),
),
Arc::new(InvalidateCacheHandler::new(cache_invalidator)),

39
src/datanode/src/heartbeat/handler.rs
View File

@@ -18,6 +18,7 @@ use common_meta::heartbeat::handler::{
HandleControl, HeartbeatResponseHandler, HeartbeatResponseHandlerContext,
};
use common_meta::instruction::{Instruction, InstructionReply};
use common_meta::kv_backend::KvBackendRef;
use common_telemetry::error;
use snafu::OptionExt;
use store_api::storage::GcReport;
@@ -56,6 +57,7 @@ pub struct RegionHeartbeatResponseHandler {
flush_tasks: TaskTracker<()>,
open_region_parallelism: usize,
gc_tasks: TaskTracker<GcReport>,
kv_backend: KvBackendRef,
}
#[async_trait::async_trait]
@@ -70,27 +72,29 @@ pub trait InstructionHandler: Send + Sync {
#[derive(Clone)]
pub struct HandlerContext {
region_server: RegionServer,
downgrade_tasks: TaskTracker<()>,
flush_tasks: TaskTracker<()>,
gc_tasks: TaskTracker<GcReport>,
pub region_server: RegionServer,
pub downgrade_tasks: TaskTracker<()>,
pub flush_tasks: TaskTracker<()>,
pub gc_tasks: TaskTracker<GcReport>,
pub kv_backend: KvBackendRef,
}
impl HandlerContext {
#[cfg(test)]
pub fn new_for_test(region_server: RegionServer) -> Self {
pub fn new_for_test(region_server: RegionServer, kv_backend: KvBackendRef) -> Self {
Self {
region_server,
downgrade_tasks: TaskTracker::new(),
flush_tasks: TaskTracker::new(),
gc_tasks: TaskTracker::new(),
kv_backend,
}
}
}
impl RegionHeartbeatResponseHandler {
/// Returns the [RegionHeartbeatResponseHandler].
pub fn new(region_server: RegionServer) -> Self {
pub fn new(region_server: RegionServer, kv_backend: KvBackendRef) -> Self {
Self {
region_server,
downgrade_tasks: TaskTracker::new(),
@@ -98,6 +102,7 @@ impl RegionHeartbeatResponseHandler {
// Default to half of the number of CPUs.
open_region_parallelism: (num_cpus::get() / 2).max(1),
gc_tasks: TaskTracker::new(),
kv_backend,
}
}
@@ -254,6 +259,7 @@ impl HeartbeatResponseHandler for RegionHeartbeatResponseHandler {
downgrade_tasks: self.downgrade_tasks.clone(),
flush_tasks: self.flush_tasks.clone(),
gc_tasks: self.gc_tasks.clone(),
kv_backend: self.kv_backend.clone(),
};
let _handle = common_runtime::spawn_global(async move {
let reply = handler.handle(&context, instruction).await;
@@ -285,6 +291,7 @@ mod tests {
use common_meta::instruction::{
DowngradeRegion, EnterStagingRegion, OpenRegion, UpgradeRegion,
};
use common_meta::kv_backend::memory::MemoryKvBackend;
use mito2::config::MitoConfig;
use mito2::engine::MITO_ENGINE_NAME;
use mito2::test_util::{CreateRequestBuilder, TestEnv};
@@ -330,7 +337,9 @@ mod tests {
fn test_is_acceptable() {
common_telemetry::init_default_ut_logging();
let region_server = mock_region_server();
let heartbeat_handler = RegionHeartbeatResponseHandler::new(region_server.clone());
let kv_backend = Arc::new(MemoryKvBackend::new());
let heartbeat_handler =
RegionHeartbeatResponseHandler::new(region_server.clone(), kv_backend);
let heartbeat_env = HeartbeatResponseTestEnv::new();
let meta = MessageMeta::new_test(1, "test", "dn-1", "me-0");
@@ -409,7 +418,9 @@ mod tests {
common_telemetry::init_default_ut_logging();
let mut region_server = mock_region_server();
let heartbeat_handler = RegionHeartbeatResponseHandler::new(region_server.clone());
let kv_backend = Arc::new(MemoryKvBackend::new());
let heartbeat_handler =
RegionHeartbeatResponseHandler::new(region_server.clone(), kv_backend);
let mut engine_env = TestEnv::with_prefix("close-region").await;
let engine = engine_env.create_engine(MitoConfig::default()).await;
@@ -457,7 +468,9 @@ mod tests {
common_telemetry::init_default_ut_logging();
let mut region_server = mock_region_server();
let heartbeat_handler = RegionHeartbeatResponseHandler::new(region_server.clone());
let kv_backend = Arc::new(MemoryKvBackend::new());
let heartbeat_handler =
RegionHeartbeatResponseHandler::new(region_server.clone(), kv_backend);
let mut engine_env = TestEnv::with_prefix("open-region").await;
let engine = engine_env.create_engine(MitoConfig::default()).await;
@@ -505,7 +518,9 @@ mod tests {
common_telemetry::init_default_ut_logging();
let mut region_server = mock_region_server();
let heartbeat_handler = RegionHeartbeatResponseHandler::new(region_server.clone());
let kv_backend = Arc::new(MemoryKvBackend::new());
let heartbeat_handler =
RegionHeartbeatResponseHandler::new(region_server.clone(), kv_backend);
let mut engine_env = TestEnv::with_prefix("open-not-exists-region").await;
let engine = engine_env.create_engine(MitoConfig::default()).await;
@@ -537,7 +552,9 @@ mod tests {
common_telemetry::init_default_ut_logging();
let mut region_server = mock_region_server();
let heartbeat_handler = RegionHeartbeatResponseHandler::new(region_server.clone());
let kv_backend = Arc::new(MemoryKvBackend::new());
let heartbeat_handler =
RegionHeartbeatResponseHandler::new(region_server.clone(), kv_backend);
let mut engine_env = TestEnv::with_prefix("downgrade-region").await;
let engine = engine_env.create_engine(MitoConfig::default()).await;

10
src/datanode/src/heartbeat/handler/apply_staging_manifest.rs
View File

@@ -117,6 +117,7 @@ mod tests {
use std::sync::Arc;
use common_meta::instruction::RemapManifest;
use common_meta::kv_backend::memory::MemoryKvBackend;
use datatypes::value::Value;
use mito2::config::MitoConfig;
use mito2::engine::MITO_ENGINE_NAME;
@@ -137,7 +138,8 @@ mod tests {
let mut mock_region_server = mock_region_server();
let (mock_engine, _) = MockRegionEngine::new(MITO_ENGINE_NAME);
mock_region_server.register_engine(mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let region_id = RegionId::new(1024, 1);
let reply = ApplyStagingManifestsHandler
.handle(
@@ -168,7 +170,8 @@ mod tests {
region_engine.handle_request_mock_fn = Some(Box::new(|_, _| Ok(0)));
});
mock_region_server.register_test_region(region_id, mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let region_id = RegionId::new(1024, 1);
let reply = ApplyStagingManifestsHandler
.handle(
@@ -231,7 +234,8 @@ mod tests {
region_server.register_engine(Arc::new(engine.clone()));
prepare_region(&region_server).await;
let handler_context = HandlerContext::new_for_test(region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(region_server, kv_backend);
let region_id2 = RegionId::new(1024, 2);
let reply = RemapManifestHandler
.handle(

5
src/datanode/src/heartbeat/handler/close_region.rs
View File

@@ -84,6 +84,7 @@ mod tests {
use common_meta::heartbeat::handler::{HandleControl, HeartbeatResponseHandler};
use common_meta::heartbeat::mailbox::MessageMeta;
use common_meta::instruction::Instruction;
use common_meta::kv_backend::memory::MemoryKvBackend;
use mito2::config::MitoConfig;
use mito2::engine::MITO_ENGINE_NAME;
use mito2::test_util::{CreateRequestBuilder, TestEnv};
@@ -113,7 +114,9 @@ mod tests {
common_telemetry::init_default_ut_logging();
let mut region_server = mock_region_server();
let heartbeat_handler = RegionHeartbeatResponseHandler::new(region_server.clone());
let kv_backend = Arc::new(MemoryKvBackend::new());
let heartbeat_handler =
RegionHeartbeatResponseHandler::new(region_server.clone(), kv_backend);
let mut engine_env = TestEnv::with_prefix("close-regions").await;
let engine = engine_env.create_engine(MitoConfig::default()).await;
region_server.register_engine(Arc::new(engine.clone()));

26
src/datanode/src/heartbeat/handler/downgrade_region.rs
View File

@@ -232,6 +232,7 @@ mod tests {
use common_meta::heartbeat::handler::{HandleControl, HeartbeatResponseHandler};
use common_meta::heartbeat::mailbox::MessageMeta;
use common_meta::instruction::{DowngradeRegion, Instruction};
use common_meta::kv_backend::memory::MemoryKvBackend;
use mito2::config::MitoConfig;
use mito2::engine::MITO_ENGINE_NAME;
use mito2::test_util::{CreateRequestBuilder, TestEnv};
@@ -255,7 +256,8 @@ mod tests {
let mut mock_region_server = mock_region_server();
let (mock_engine, _) = MockRegionEngine::new(MITO_ENGINE_NAME);
mock_region_server.register_engine(mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let region_id = RegionId::new(1024, 1);
let waits = vec![None, Some(Duration::from_millis(100u64))];
@@ -299,7 +301,8 @@ mod tests {
}))
});
mock_region_server.register_test_region(region_id, mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let waits = vec![None, Some(Duration::from_millis(100u64))];
for flush_timeout in waits {
@@ -335,7 +338,8 @@ mod tests {
}))
});
mock_region_server.register_test_region(region_id, mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let flush_timeout = Duration::from_millis(100);
let reply = DowngradeRegionsHandler
@@ -369,7 +373,8 @@ mod tests {
}))
});
mock_region_server.register_test_region(region_id, mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let waits = vec![
Some(Duration::from_millis(100u64)),
@@ -432,7 +437,8 @@ mod tests {
}))
});
mock_region_server.register_test_region(region_id, mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let waits = vec![
Some(Duration::from_millis(100u64)),
@@ -483,7 +489,8 @@ mod tests {
Some(Box::new(|_| Ok(SetRegionRoleStateResponse::NotFound)));
});
mock_region_server.register_test_region(region_id, mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let reply = DowngradeRegionsHandler
.handle(
&handler_context,
@@ -514,7 +521,8 @@ mod tests {
}));
});
mock_region_server.register_test_region(region_id, mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let reply = DowngradeRegionsHandler
.handle(
&handler_context,
@@ -541,7 +549,9 @@ mod tests {
common_telemetry::init_default_ut_logging();
let mut region_server = mock_region_server();
let heartbeat_handler = RegionHeartbeatResponseHandler::new(region_server.clone());
let kv_backend = Arc::new(MemoryKvBackend::new());
let heartbeat_handler =
RegionHeartbeatResponseHandler::new(region_server.clone(), kv_backend);
let mut engine_env = TestEnv::with_prefix("downgrade-regions").await;
let engine = engine_env.create_engine(MitoConfig::default()).await;
region_server.register_engine(Arc::new(engine.clone()));

10
src/datanode/src/heartbeat/handler/enter_staging.rs
View File

@@ -107,6 +107,7 @@ mod tests {
use std::sync::Arc;
use common_meta::instruction::EnterStagingRegion;
use common_meta::kv_backend::memory::MemoryKvBackend;
use mito2::config::MitoConfig;
use mito2::engine::MITO_ENGINE_NAME;
use mito2::test_util::{CreateRequestBuilder, TestEnv};
@@ -127,7 +128,8 @@ mod tests {
let mut mock_region_server = mock_region_server();
let (mock_engine, _) = MockRegionEngine::new(MITO_ENGINE_NAME);
mock_region_server.register_engine(mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let region_id = RegionId::new(1024, 1);
let replies = EnterStagingRegionsHandler
.handle(
@@ -156,7 +158,8 @@ mod tests {
region_engine.handle_request_mock_fn = Some(Box::new(|_, _| Ok(0)));
});
mock_region_server.register_test_region(region_id, mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let replies = EnterStagingRegionsHandler
.handle(
&handler_context,
@@ -194,7 +197,8 @@ mod tests {
region_server.register_engine(Arc::new(engine.clone()));
prepare_region(&region_server).await;
let handler_context = HandlerContext::new_for_test(region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(region_server, kv_backend);
let replies = EnterStagingRegionsHandler
.handle(
&handler_context,

13
src/datanode/src/heartbeat/handler/flush_region.rs
View File

@@ -177,6 +177,7 @@ mod tests {
use std::sync::{Arc, RwLock};
use common_meta::instruction::{FlushErrorStrategy, FlushRegions};
use common_meta::kv_backend::memory::MemoryKvBackend;
use mito2::engine::MITO_ENGINE_NAME;
use store_api::storage::RegionId;
@@ -201,7 +202,8 @@ mod tests {
});
mock_region_server.register_test_region(*region_id, mock_engine);
}
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
// Async hint mode
let flush_instruction = FlushRegions::async_batch(region_ids.clone());
@@ -238,7 +240,8 @@ mod tests {
}))
});
mock_region_server.register_test_region(region_id, mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let flush_instruction = FlushRegions::sync_single(region_id);
let reply = FlushRegionsHandler
@@ -273,7 +276,8 @@ mod tests {
}))
});
mock_region_server.register_test_region(region_ids[0], mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
// Sync batch with fail-fast strategy
let flush_instruction =
@@ -304,7 +308,8 @@ mod tests {
}))
});
mock_region_server.register_test_region(region_ids[0], mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
// Sync batch with try-all strategy
let flush_instruction =

360
src/datanode/src/heartbeat/handler/gc_worker.rs
View File

@@ -12,13 +12,24 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use common_meta::instruction::{GcRegions, GcRegionsReply, InstructionReply};
use common_telemetry::{debug, warn};
use mito2::gc::LocalGcWorker;
use snafu::{OptionExt, ResultExt, ensure};
use store_api::storage::{FileRefsManifest, RegionId};
use std::collections::{BTreeMap, HashMap};
use std::sync::Arc;
use crate::error::{GcMitoEngineSnafu, InvalidGcArgsSnafu, Result, UnexpectedSnafu};
use common_meta::instruction::{GcRegions, GcRegionsReply, InstructionReply};
use common_meta::key::table_info::TableInfoManager;
use common_meta::key::table_route::TableRouteManager;
use common_telemetry::{debug, warn};
use mito2::access_layer::{AccessLayer, AccessLayerRef};
use mito2::engine::MitoEngine;
use mito2::gc::LocalGcWorker;
use mito2::region::MitoRegionRef;
use snafu::{OptionExt, ResultExt};
use store_api::path_utils::table_dir;
use store_api::region_request::PathType;
use store_api::storage::{FileRefsManifest, GcReport, RegionId};
use table::requests::STORAGE_KEY;
use crate::error::{GcMitoEngineSnafu, GetMetadataSnafu, Result, UnexpectedSnafu};
use crate::heartbeat::handler::{HandlerContext, InstructionHandler};
pub struct GcRegionsHandler;
@@ -35,43 +46,80 @@ impl InstructionHandler for GcRegionsHandler {
let region_ids = gc_regions.regions.clone();
debug!("Received gc regions instruction: {:?}", region_ids);
let (region_id, gc_worker) = match self
.create_gc_worker(
ctx,
region_ids,
&gc_regions.file_refs_manifest,
gc_regions.full_file_listing,
)
.await
{
Ok(worker) => worker,
Err(e) => {
return Some(InstructionReply::GcRegions(GcRegionsReply {
result: Err(format!("Failed to create GC worker: {}", e)),
}));
}
};
if region_ids.is_empty() {
return Some(InstructionReply::GcRegions(GcRegionsReply {
result: Ok(GcReport::default()),
}));
}
// Always use the smallest region id on datanode as the target region id for task tracker
let mut sorted_region_ids = gc_regions.regions.clone();
sorted_region_ids.sort_by_key(|r| r.region_number());
let target_region_id = sorted_region_ids[0];
// Group regions by table_id
let mut table_to_regions: HashMap<u32, Vec<RegionId>> = HashMap::new();
for rid in region_ids {
table_to_regions
.entry(rid.table_id())
.or_default()
.push(rid);
}
let file_refs_manifest = gc_regions.file_refs_manifest.clone();
let full_file_listing = gc_regions.full_file_listing;
let ctx_clone = ctx.clone();
let register_result = ctx
.gc_tasks
.try_register(
region_id,
target_region_id,
Box::pin(async move {
debug!("Starting gc worker for region {}", region_id);
let report = gc_worker
.run()
.await
.context(GcMitoEngineSnafu { region_id })?;
debug!("Gc worker for region {} finished", region_id);
Ok(report)
let mut reports = Vec::with_capacity(table_to_regions.len());
for (table_id, regions) in table_to_regions {
debug!(
"Starting gc worker for table {}, regions: {:?}",
table_id, regions
);
let gc_worker = GcRegionsHandler::create_gc_worker(
&ctx_clone,
table_id,
regions,
&file_refs_manifest,
full_file_listing,
)
.await?;
let report = gc_worker.run().await.context(GcMitoEngineSnafu {
region_id: target_region_id,
})?;
debug!(
"Gc worker for table {} finished, report: {:?}",
table_id, report
);
reports.push(report);
}
// Merge reports
let mut merged_report = GcReport::default();
for report in reports {
merged_report
.deleted_files
.extend(report.deleted_files.into_iter());
merged_report
.deleted_indexes
.extend(report.deleted_indexes.into_iter());
}
Ok(merged_report)
}),
)
.await;
if register_result.is_busy() {
warn!("Another gc task is running for the region: {region_id}");
warn!("Another gc task is running for the region: {target_region_id}");
return Some(InstructionReply::GcRegions(GcRegionsReply {
result: Err(format!(
"Another gc task is running for the region: {region_id}"
"Another gc task is running for the region: {target_region_id}"
)),
}));
}
@@ -89,17 +137,15 @@ impl InstructionHandler for GcRegionsHandler {
}
impl GcRegionsHandler {
/// Create a GC worker for the given region IDs.
/// Return the first region ID(after sort by given region id) and the GC worker.
/// Create a GC worker for the given table and region IDs.
async fn create_gc_worker(
&self,
ctx: &HandlerContext,
mut region_ids: Vec<RegionId>,
table_id: u32,
region_ids: Vec<RegionId>,
file_ref_manifest: &FileRefsManifest,
full_file_listing: bool,
) -> Result<(RegionId, LocalGcWorker)> {
// always use the smallest region id on datanode as the target region id
region_ids.sort_by_key(|r| r.region_number());
) -> Result<LocalGcWorker> {
debug_assert!(!region_ids.is_empty(), "region_ids should not be empty");
let mito_engine = ctx
.region_server
@@ -108,50 +154,13 @@ impl GcRegionsHandler {
violated: "MitoEngine not found".to_string(),
})?;
let region_id = *region_ids.first().with_context(|| InvalidGcArgsSnafu {
msg: "No region ids provided".to_string(),
})?;
// also need to ensure all regions are on this datanode
ensure!(
region_ids
.iter()
.all(|rid| mito_engine.find_region(*rid).is_some()),
InvalidGcArgsSnafu {
msg: format!(
"Some regions are not on current datanode:{:?}",
region_ids
.iter()
.filter(|rid| mito_engine.find_region(**rid).is_none())
.collect::<Vec<_>>()
),
}
);
// Find the access layer from one of the regions that exists on this datanode
let access_layer = mito_engine
.find_region(region_id)
.with_context(|| InvalidGcArgsSnafu {
msg: format!(
"None of the regions is on current datanode:{:?}",
region_ids
),
})?
.access_layer();
// if region happen to be dropped before this but after gc scheduler send gc instr,
// need to deal with it properly(it is ok for region to be dropped after GC worker started)
// region not found here can only be drop table/database case, since region migration is prevented by lock in gc procedure
// TODO(discord9): add integration test for this drop case
let mito_regions = region_ids
.iter()
.filter_map(|rid| mito_engine.find_region(*rid).map(|r| (*rid, r)))
.collect();
let (access_layer, mito_regions) =
Self::get_access_layer(ctx, &mito_engine, table_id, &region_ids).await?;
let cache_manager = mito_engine.cache_manager();
let gc_worker = LocalGcWorker::try_new(
access_layer.clone(),
access_layer,
Some(cache_manager),
mito_regions,
mito_engine.mito_config().gc.clone(),
@@ -160,8 +169,197 @@ impl GcRegionsHandler {
full_file_listing,
)
.await
.context(GcMitoEngineSnafu { region_id })?;
.context(GcMitoEngineSnafu {
region_id: region_ids[0],
})?;
Ok((region_id, gc_worker))
Ok(gc_worker)
}
/// Get the access layer for the given table and region IDs.
/// It also returns the mito regions if they are found in the engine.
///
/// This method validates:
/// 1. Any found region must be a Leader (not Follower)
/// 2. Any missing region must not be routed to another datanode
///
/// The AccessLayer is always constructed from table metadata for consistency.
async fn get_access_layer(
ctx: &HandlerContext,
mito_engine: &MitoEngine,
table_id: u32,
region_ids: &[RegionId],
) -> Result<(AccessLayerRef, BTreeMap<RegionId, Option<MitoRegionRef>>)> {
// 1. Collect mito regions and validate Leader status
let mut mito_regions = BTreeMap::new();
for rid in region_ids {
let region = mito_engine.find_region(*rid);
if let Some(ref r) = region {
// Validation: Check if region is a leader
if r.is_follower() {
return Err(UnexpectedSnafu {
violated: format!(
"Region {} is a follower, cannot perform GC on follower regions",
rid
),
}
.build());
}
}
mito_regions.insert(*rid, region);
}
// 2. Validate that missing regions are not routed to other datanodes
let missing_regions: Vec<_> = mito_regions
.iter()
.filter(|(_, r)| r.is_none())
.map(|(rid, _)| *rid)
.collect();
if !missing_regions.is_empty() {
Self::validate_regions_not_routed_elsewhere(ctx, table_id, &missing_regions).await?;
}
// 3. Construct AccessLayer directly from table metadata
let access_layer = Self::construct_access_layer(ctx, mito_engine, table_id).await?;
Ok((access_layer, mito_regions))
}
/// Manually construct an access layer from table metadata.
async fn construct_access_layer(
ctx: &HandlerContext,
mito_engine: &MitoEngine,
table_id: u32,
) -> Result<AccessLayerRef> {
let table_info_manager = TableInfoManager::new(ctx.kv_backend.clone());
let table_info_value = table_info_manager
.get(table_id)
.await
.context(GetMetadataSnafu)?
.with_context(|| UnexpectedSnafu {
violated: format!("Table metadata not found for table {}", table_id),
})?;
let table_dir = table_dir(&table_info_value.region_storage_path(), table_id);
let storage_name = table_info_value
.table_info
.meta
.options
.extra_options
.get(STORAGE_KEY);
let engine = &table_info_value.table_info.meta.engine;
let path_type = match engine.as_str() {
common_catalog::consts::MITO2_ENGINE => PathType::Bare,
common_catalog::consts::MITO_ENGINE => PathType::Bare,
common_catalog::consts::METRIC_ENGINE => PathType::Data,
_ => PathType::Bare,
};
let object_store = if let Some(name) = storage_name {
mito_engine
.object_store_manager()
.find(name)
.cloned()
.with_context(|| UnexpectedSnafu {
violated: format!("Object store {} not found", name),
})?
} else {
mito_engine
.object_store_manager()
.default_object_store()
.clone()
};
Ok(Arc::new(AccessLayer::new(
table_dir,
path_type,
object_store,
mito_engine.puffin_manager_factory().clone(),
mito_engine.intermediate_manager().clone(),
)))
}
/// Validate that the given regions are not routed to other datanodes.
///
/// If any region is still active on another datanode (has a leader_peer in route table),
/// this function returns an error to prevent accidental deletion of files
/// that are still in use.
async fn validate_regions_not_routed_elsewhere(
ctx: &HandlerContext,
table_id: u32,
missing_region_ids: &[RegionId],
) -> Result<()> {
if missing_region_ids.is_empty() {
return Ok(());
}
let table_route_manager = TableRouteManager::new(ctx.kv_backend.clone());
// Get table route
let table_route = match table_route_manager
.table_route_storage()
.get(table_id)
.await
.context(GetMetadataSnafu)?
{
Some(route) => route,
None => {
// Table route not found, all regions are likely deleted
debug!(
"Table route not found for table {}, regions {:?} are considered deleted",
table_id, missing_region_ids
);
return Ok(());
}
};
// Get region routes for physical table
let region_routes = match table_route.region_routes() {
Ok(routes) => routes,
Err(_) => {
// Logical table, skip validation
debug!(
"Table {} is a logical table, skipping region route validation",
table_id
);
return Ok(());
}
};
let region_routes_map: HashMap<RegionId, _> = region_routes
.iter()
.map(|route| (route.region.id, route))
.collect();
// Check each missing region
for region_id in missing_region_ids {
if let Some(route) = region_routes_map.get(region_id) {
if let Some(leader_peer) = &route.leader_peer {
// Region still has a leader on some datanode.
return Err(UnexpectedSnafu {
violated: format!(
"Region {} is not on this datanode but is routed to datanode {}. \
GC request may have been sent to wrong datanode.",
region_id, leader_peer.id
),
}
.build());
}
return Err(UnexpectedSnafu {
violated: format!(
"Region {} has no leader in route table; refusing GC without explicit tombstone/deleted state.",
region_id
),
}
.build());
}
// Region not in route table: treat as deleted and allow GC.
}
Ok(())
}
}

5
src/datanode/src/heartbeat/handler/open_region.rs
View File

@@ -80,6 +80,7 @@ mod tests {
use common_meta::heartbeat::handler::{HandleControl, HeartbeatResponseHandler};
use common_meta::heartbeat::mailbox::MessageMeta;
use common_meta::instruction::{Instruction, OpenRegion};
use common_meta::kv_backend::memory::MemoryKvBackend;
use mito2::config::MitoConfig;
use mito2::engine::MITO_ENGINE_NAME;
use mito2::test_util::{CreateRequestBuilder, TestEnv};
@@ -119,7 +120,9 @@ mod tests {
common_telemetry::init_default_ut_logging();
let mut region_server = mock_region_server();
let heartbeat_handler = RegionHeartbeatResponseHandler::new(region_server.clone());
let kv_backend = Arc::new(MemoryKvBackend::new());
let heartbeat_handler =
RegionHeartbeatResponseHandler::new(region_server.clone(), kv_backend);
let mut engine_env = TestEnv::with_prefix("open-regions").await;
let engine = engine_env.create_engine(MitoConfig::default()).await;
region_server.register_engine(Arc::new(engine.clone()));

10
src/datanode/src/heartbeat/handler/remap_manifest.rs
View File

@@ -97,6 +97,7 @@ mod tests {
use std::sync::Arc;
use common_meta::instruction::RemapManifest;
use common_meta::kv_backend::memory::MemoryKvBackend;
use datatypes::value::Value;
use mito2::config::MitoConfig;
use mito2::engine::MITO_ENGINE_NAME;
@@ -117,7 +118,8 @@ mod tests {
let mut mock_region_server = mock_region_server();
let (mock_engine, _) = MockRegionEngine::new(MITO_ENGINE_NAME);
mock_region_server.register_engine(mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let region_id = RegionId::new(1024, 1);
let reply = RemapManifestHandler
.handle(
@@ -147,7 +149,8 @@ mod tests {
region_engine.handle_request_mock_fn = Some(Box::new(|_, _| Ok(0)));
});
mock_region_server.register_test_region(region_id, mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let reply = RemapManifestHandler
.handle(
&handler_context,
@@ -207,7 +210,8 @@ mod tests {
region_server.register_engine(Arc::new(engine.clone()));
prepare_region(&region_server).await;
let handler_context = HandlerContext::new_for_test(region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(region_server, kv_backend);
let region_id2 = RegionId::new(1024, 2);
let reply = RemapManifestHandler
.handle(

12
src/datanode/src/heartbeat/handler/sync_region.rs
View File

@@ -97,6 +97,9 @@ impl SyncRegionHandler {
#[cfg(test)]
mod tests {
use std::sync::Arc;
use common_meta::kv_backend::memory::MemoryKvBackend;
use store_api::metric_engine_consts::METRIC_ENGINE_NAME;
use store_api::region_engine::{RegionRole, SyncRegionFromRequest};
use store_api::storage::RegionId;
@@ -111,7 +114,8 @@ mod tests {
let (mock_engine, _) = MockRegionEngine::new(METRIC_ENGINE_NAME);
mock_region_server.register_engine(mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let handler = SyncRegionHandler;
let region_id = RegionId::new(1024, 1);
@@ -141,7 +145,8 @@ mod tests {
});
mock_region_server.register_test_region(region_id, mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let handler = SyncRegionHandler;
let sync_region = common_meta::instruction::SyncRegion {
@@ -171,7 +176,8 @@ mod tests {
});
mock_region_server.register_test_region(region_id, mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let handler = SyncRegionHandler;
let sync_region = common_meta::instruction::SyncRegion {

20
src/datanode/src/heartbeat/handler/upgrade_region.rs
View File

@@ -220,9 +220,11 @@ impl InstructionHandler for UpgradeRegionsHandler {
#[cfg(test)]
mod tests {
use std::sync::Arc;
use std::time::Duration;
use common_meta::instruction::UpgradeRegion;
use common_meta::kv_backend::memory::MemoryKvBackend;
use mito2::engine::MITO_ENGINE_NAME;
use store_api::region_engine::RegionRole;
use store_api::storage::RegionId;
@@ -237,8 +239,8 @@ mod tests {
let mut mock_region_server = mock_region_server();
let (mock_engine, _) = MockRegionEngine::new(MITO_ENGINE_NAME);
mock_region_server.register_engine(mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let region_id = RegionId::new(1024, 1);
let region_id2 = RegionId::new(1024, 2);
@@ -286,7 +288,8 @@ mod tests {
});
mock_region_server.register_test_region(region_id, mock_engine.clone());
mock_region_server.register_test_region(region_id2, mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let replay_timeout = Duration::from_millis(100u64);
let reply = UpgradeRegionsHandler::new_test()
.handle(
@@ -330,8 +333,8 @@ mod tests {
region_engine.handle_request_delay = Some(Duration::from_secs(100));
});
mock_region_server.register_test_region(region_id, mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let replay_timeout = Duration::from_millis(100u64);
let reply = UpgradeRegionsHandler::new_test()
.handle(
@@ -365,7 +368,8 @@ mod tests {
});
mock_region_server.register_test_region(region_id, mock_engine);
let waits = vec![Duration::from_millis(100u64), Duration::from_millis(100u64)];
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
for replay_timeout in waits {
let reply = UpgradeRegionsHandler::new_test()
.handle(
@@ -420,8 +424,8 @@ mod tests {
region_engine.handle_request_delay = Some(Duration::from_millis(100));
});
mock_region_server.register_test_region(region_id, mock_engine);
let handler_context = HandlerContext::new_for_test(mock_region_server);
let kv_backend = Arc::new(MemoryKvBackend::new());
let handler_context = HandlerContext::new_for_test(mock_region_server, kv_backend);
let reply = UpgradeRegionsHandler::new_test()
.handle(
&handler_context,