feat: add a region hook for gc cleanup

This commit is contained in:
Ning Sun
2026-07-17 10:34:22 +08:00
parent 2123108db0
commit b573089485
5 changed files with 191 additions and 3 deletions
@@ -169,6 +169,7 @@ impl GcRegionsHandler {
file_ref_manifest.clone(),
&mito_engine.gc_limiter(),
full_file_listing,
mito_engine.region_hook(),
)
.await
.context(GcMitoEngineSnafu {
+15
View File
@@ -105,6 +105,7 @@ use common_wal::options::WalOptions;
use futures::future::{join_all, try_join_all};
use futures::stream::{self, Stream, StreamExt};
use object_store::manager::ObjectStoreManagerRef;
use region_hook::RegionHookRef;
use snafu::{OptionExt, ResultExt, ensure};
use store_api::ManifestVersion;
use store_api::codec::PrimaryKeyEncoding;
@@ -219,6 +220,9 @@ impl<'a, S: LogStore> MitoEngineBuilder<'a, S> {
self.config.sanitize(self.data_home)?;
let config = Arc::new(self.config);
// Extract the region hook before `plugins` is moved into the WorkerGroup,
// so the engine (and thus the GC worker) can fire `on_region_gc`.
let region_hook = self.plugins.get::<RegionHookRef>();
let workers = WorkerGroup::start(
config.clone(),
self.log_store.clone(),
@@ -250,6 +254,7 @@ impl<'a, S: LogStore> MitoEngineBuilder<'a, S> {
config,
wal_raw_entry_reader,
scan_memory_tracker,
region_hook,
#[cfg(feature = "enterprise")]
extension_range_provider_factory: None,
};
@@ -328,6 +333,12 @@ impl MitoEngine {
self.inner.workers.schema_metadata_manager()
}
/// Returns the registered region hook (if any), for the GC worker to fire
/// [`RegionHook::on_region_gc`].
pub fn region_hook(&self) -> Option<RegionHookRef> {
self.inner.region_hook.clone()
}
/// Get all tmp ref files for given region ids, excluding files that's already in manifest.
pub async fn get_snapshot_of_file_refs(
&self,
@@ -728,6 +739,9 @@ struct EngineInner {
wal_raw_entry_reader: Arc<dyn RawEntryReader>,
/// Memory tracker for table scans.
scan_memory_tracker: QueryMemoryTracker,
/// The region hook (if any) registered via plugins; exposed for the GC worker
/// to fire [`RegionHook::on_region_gc`].
region_hook: Option<RegionHookRef>,
#[cfg(feature = "enterprise")]
extension_range_provider_factory: Option<BoxedExtensionRangeProviderFactory>,
}
@@ -1458,6 +1472,7 @@ impl MitoEngine {
config,
wal_raw_entry_reader,
scan_memory_tracker,
region_hook: None,
#[cfg(feature = "enterprise")]
extension_range_provider_factory: None,
}),
+45 -3
View File
@@ -89,6 +89,7 @@
//! | Close | [`on_region_closed`] | A close request (or a close-after-flush) removes the region from the active set. Data files, manifest and WAL state are **preserved**; the region may be reopened. |
//! | Logical drop | [`on_region_dropped`] | A drop request has been handled: the region leaves the active set and its WAL entries are marked obsolete. Data files are **not yet deleted**. |
//! | Physical file removal | [`on_region_files_removed`] | The drop GC worker has deleted the region directory. Terminal file-lifecycle event. |
//! | Global GC pass | [`on_region_gc`] | The datanode's global GC worker finished a GC pass for a region — both periodic GC for live regions and the global reclamation of dropped/repartitioned regions (`is_region_dropped`). |
//!
//! Notes:
//! - `on_region_closed` / `on_region_dropped` run **inline in the region worker loop**,
@@ -128,8 +129,9 @@
//!
//! `on_region_files_removed` currently covers only the **drop** GC worker's physical
//! directory removal. A broader per-file `on_files_removed` hook covering compaction
//! removal, truncate, and the global GC reclamation path is not yet implemented
//! (though logical file removal is already observable via `on_manifest_updated`).
//! removal and truncate is not yet implemented
//! (though logical file removal is already observable via `on_manifest_updated`,
//! and the global GC reclamation path is covered by `on_region_gc`).
//! Role/leadership transitions (`on_region_role_changed`) are also not hooked.
//!
//! [`on_sst_files_written`]: RegionHook::on_sst_files_written
@@ -138,6 +140,7 @@
//! [`on_region_closed`]: RegionHook::on_region_closed
//! [`on_region_dropped`]: RegionHook::on_region_dropped
//! [`on_region_files_removed`]: RegionHook::on_region_files_removed
//! [`on_region_gc`]: RegionHook::on_region_gc
//! [`RegionManifestManager::update`]: crate::manifest::manager::RegionManifestManager::update
//! [`ManifestContext::update_locked`]: crate::region::ManifestContext::update_locked
//! [`ManifestContext::update_manifest`]: crate::region::ManifestContext::update_manifest
@@ -150,7 +153,8 @@ use store_api::ManifestVersion;
use store_api::metadata::RegionMetadataRef;
use store_api::storage::RegionId;
use crate::manifest::action::RegionMetaActionList;
use crate::access_layer::AccessLayerRef;
use crate::manifest::action::{RegionMetaActionList, RemovedFile};
use crate::sst::file::FileMeta;
use crate::sst::parquet::SstInfo;
@@ -382,6 +386,44 @@ pub trait RegionHook: Send + Sync + Debug {
) {
let _ = (region_id, region_metadata);
}
/// Called after the datanode's global GC worker (`LocalGcWorker`) finishes a
/// GC pass for a region — for both live regions (periodic GC) and dropped /
/// repartitioned regions (the global reclamation path, where
/// [`RegionGcInfo::is_region_dropped`] is `true`). This is the hook point for
/// extensions that keep sidecar files outside mito2's region dir — e.g. an
/// Iceberg manifest tree under a separate warehouse root — to clean up their
/// own residual files: the whole sidecar dir for a dropped region, or stale
/// files for a live one.
///
/// `region_metadata` is `None` for dropped regions; locate files from
/// `region_id` + `access_layer` instead. Runs on the background GC task, so
/// it may do non-trivial I/O, and must be idempotent (global GC re-runs on a
/// cooldown).
async fn on_region_gc(
&self,
region_id: RegionId,
region_metadata: Option<&RegionMetadataRef>,
access_layer: &AccessLayerRef,
info: &RegionGcInfo<'_>,
) {
let _ = (region_id, region_metadata, access_layer, info);
}
}
/// What mito2's GC pass deleted for a region, handed to
/// [`RegionHook::on_region_gc`].
pub struct RegionGcInfo<'a> {
/// Files mito2 physically deleted this pass.
pub removed_files: &'a [RemovedFile],
/// `true` when the region is dropped/absent (the global-GC reclamation path,
/// e.g. a repartitioned-away source region). When `true`, `on_region_gc`
/// receives no `region_metadata` and should locate any sidecar files from
/// `region_id` + [`AccessLayer::table_dir`] alone.
pub is_region_dropped: bool,
/// Whether this pass did a full object-store listing (vs. fast mode that
/// only reaps manifest-tracked removed files).
pub full_file_listing: bool,
}
pub type RegionHookRef = Arc<dyn RegionHook>;
+25
View File
@@ -41,6 +41,7 @@ use crate::access_layer::AccessLayerRef;
use crate::cache::CacheManagerRef;
use crate::cache::file_cache::FileType;
use crate::config::MitoConfig;
use crate::engine::region_hook::{RegionGcInfo, RegionHookRef};
use crate::error::{
DurationOutOfRangeSnafu, InvalidRequestSnafu, JoinSnafu, OpenDalSnafu, Result,
TooManyGcJobsSnafu, UnexpectedSnafu,
@@ -205,6 +206,9 @@ pub struct LocalGcWorker {
/// Set to false for regular GC operations to optimize performance.
/// Set to true periodically or when you need to clean up orphan files.
pub full_file_listing: bool,
/// The region hook (if any), fired via `on_region_gc` after each GC pass so
/// extensions with sidecar files outside the mito2 region dir can clean up.
pub(crate) hook: Option<RegionHookRef>,
}
pub struct ManifestOpenConfig {
@@ -240,6 +244,7 @@ impl LocalGcWorker {
file_ref_manifest: FileRefsManifest,
limiter: &GcLimiterRef,
full_file_listing: bool,
hook: Option<RegionHookRef>,
) -> Result<Self> {
if let Some(first_region_id) = regions_to_gc.keys().next() {
let table_id = first_region_id.table_id();
@@ -267,6 +272,7 @@ impl LocalGcWorker {
file_ref_manifest,
_permit: permit,
full_file_listing,
hook,
})
}
@@ -539,6 +545,25 @@ impl LocalGcWorker {
.await?;
}
// Notify extensions (e.g. an Iceberg manifest tree under a separate
// warehouse root) that GC reclaimed these files, so they can clean up
// their own residual sidecar files. Fires for both live regions and
// dropped/repartitioned regions (`is_region_dropped`).
if let Some(hook) = &self.hook {
let region_metadata = region.as_ref().map(|r| r.metadata());
hook.on_region_gc(
region_id,
region_metadata.as_ref(),
&self.access_layer,
&RegionGcInfo {
removed_files: &deletable_files,
is_region_dropped,
full_file_listing: self.full_file_listing,
},
)
.await;
}
Ok(deletable_files)
}
+105
View File
@@ -14,18 +14,24 @@
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use api::v1::Rows;
use async_trait::async_trait;
use common_base::Plugins;
use common_telemetry::init_default_ut_logging;
use futures::TryStreamExt;
use object_store::{Entry, ObjectStore, services};
use store_api::metadata::RegionMetadataRef;
use store_api::region_engine::RegionEngine as _;
use store_api::region_request::{RegionCompactRequest, RegionRequest};
use store_api::storage::{FileRef, FileRefsManifest, RegionId};
use crate::access_layer::AccessLayerRef;
use crate::config::MitoConfig;
use crate::engine::MitoEngine;
use crate::engine::compaction_test::{delete_and_flush, put_and_flush};
use crate::engine::region_hook::{RegionGcInfo, RegionHook, RegionHookRef};
use crate::gc::{GcConfig, LocalGcWorker, should_delete_file};
use crate::manifest::action::RemovedFile;
use crate::region::MitoRegionRef;
@@ -58,6 +64,7 @@ async fn create_gc_worker(
file_ref_manifest.clone(),
&mito_engine.gc_limiter(),
full_file_listing,
mito_engine.region_hook(),
)
.await
.unwrap()
@@ -738,3 +745,101 @@ async fn test_file_in_tmp_ref_old_mtime_kept() {
"File in tmp_ref should NOT be deleted even with old last-modified time"
);
}
/// A region hook that only records `on_region_gc` calls, for testing the GC hook.
#[derive(Debug, Default)]
struct GcCountingHook {
gc_calls: AtomicUsize,
last_is_region_dropped: AtomicBool,
}
#[async_trait]
impl RegionHook for GcCountingHook {
async fn on_region_gc(
&self,
_region_id: RegionId,
_region_metadata: Option<&RegionMetadataRef>,
_access_layer: &AccessLayerRef,
info: &RegionGcInfo<'_>,
) {
self.gc_calls.fetch_add(1, Ordering::Relaxed);
self.last_is_region_dropped
.store(info.is_region_dropped, Ordering::Relaxed);
}
}
/// `on_region_gc` fires after a GC pass for a live region.
#[tokio::test]
async fn test_on_region_gc_fires_on_gc_pass() {
init_default_ut_logging();
let mut env = TestEnv::new().await;
let hook = Arc::new(GcCountingHook::default());
let plugins = Plugins::new();
plugins.insert(hook.clone() as RegionHookRef);
let engine = env
.create_engine_with_plugins(
MitoConfig {
gc: GcConfig {
enable: true,
lingering_time: None,
..Default::default()
},
..Default::default()
},
plugins,
)
.await;
let region_id = RegionId::new(1, 1);
env.get_schema_metadata_manager()
.register_region_table_info(
region_id.table_id(),
"test_table",
"test_catalog",
"test_schema",
None,
env.get_kv_backend(),
)
.await;
let request = CreateRequestBuilder::new().build();
engine
.handle_request(region_id, RegionRequest::Create(request.clone()))
.await
.unwrap();
put_rows(
&engine,
region_id,
Rows {
schema: rows_schema(&request),
rows: build_rows(0, 3),
},
)
.await;
flush_region(&engine, region_id, None).await;
let region = engine.get_region(region_id).unwrap();
let version = region.manifest_ctx.manifest().await.manifest_version;
let regions = BTreeMap::from([(region_id, Some(region.clone()))]);
let file_ref_manifest = FileRefsManifest {
file_refs: Default::default(),
manifest_version: [(region_id, version)].into(),
cross_region_refs: HashMap::new(),
};
let gc_worker = create_gc_worker(&engine, regions, &file_ref_manifest, true).await;
gc_worker.run().await.unwrap();
assert!(
hook.gc_calls.load(Ordering::Relaxed) >= 1,
"on_region_gc should fire after a GC pass"
);
assert_eq!(
hook.last_is_region_dropped.load(Ordering::Relaxed),
false,
"a live region's GC pass must report is_region_dropped=false"
);
}