diff --git a/src/datanode/src/heartbeat/handler/gc_worker.rs b/src/datanode/src/heartbeat/handler/gc_worker.rs index 9489f1461d9..b32a1c29ad4 100644 --- a/src/datanode/src/heartbeat/handler/gc_worker.rs +++ b/src/datanode/src/heartbeat/handler/gc_worker.rs @@ -169,6 +169,7 @@ impl GcRegionsHandler { file_ref_manifest.clone(), &mito_engine.gc_limiter(), full_file_listing, + mito_engine.region_hook(), ) .await .context(GcMitoEngineSnafu { diff --git a/src/mito2/src/engine.rs b/src/mito2/src/engine.rs index 94911814b56..a128531b535 100644 --- a/src/mito2/src/engine.rs +++ b/src/mito2/src/engine.rs @@ -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::(); 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 { + 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, /// 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, #[cfg(feature = "enterprise")] extension_range_provider_factory: Option, } @@ -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, }), diff --git a/src/mito2/src/engine/region_hook.rs b/src/mito2/src/engine/region_hook.rs index 9c5a7dfc95d..2758d4a88f4 100644 --- a/src/mito2/src/engine/region_hook.rs +++ b/src/mito2/src/engine/region_hook.rs @@ -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; diff --git a/src/mito2/src/gc.rs b/src/mito2/src/gc.rs index 61ba362a7bd..a8cbea911a7 100644 --- a/src/mito2/src/gc.rs +++ b/src/mito2/src/gc.rs @@ -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, } pub struct ManifestOpenConfig { @@ -240,6 +244,7 @@ impl LocalGcWorker { file_ref_manifest: FileRefsManifest, limiter: &GcLimiterRef, full_file_listing: bool, + hook: Option, ) -> Result { 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) } diff --git a/src/mito2/src/gc/worker_test.rs b/src/mito2/src/gc/worker_test.rs index 794b5d61c0c..a15855b7cb7 100644 --- a/src/mito2/src/gc/worker_test.rs +++ b/src/mito2/src/gc/worker_test.rs @@ -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" + ); +}