diff --git a/src/meta-srv/src/gc/procedure.rs b/src/meta-srv/src/gc/procedure.rs index 05ad098749b..c1841d2c77e 100644 --- a/src/meta-srv/src/gc/procedure.rs +++ b/src/meta-srv/src/gc/procedure.rs @@ -382,6 +382,15 @@ impl BatchGcProcedure { let repart_mgr = self.table_metadata_manager.table_repart_manager(); + // Regions whose extension sidecar cleanup failed and need retry. Keep + // their tombstone so the next GC cycle can replay the cleanup. + let need_retry: HashSet = self + .data + .gc_report + .as_ref() + .map(|r| r.need_retry_regions.clone()) + .unwrap_or_default(); + let mut table_ids: HashSet = cross_refs_grouped .keys() .copied() @@ -432,9 +441,9 @@ impl BatchGcProcedure { let mut set = BTreeSet::new(); set.extend(dst_regions.iter().copied()); new_value.src_to_dst.insert(src_region, set); - } else if has_tmp_ref { - // Keep a tombstone entry with an empty set so dropped regions that still - // have tmp refs are preserved; removing it would lose the repartition trace. + } else if has_tmp_ref || need_retry.contains(&src_region) { + // Keep the tombstone: tmp refs or pending extension cleanup + // still need a future GC pass. new_value.src_to_dst.insert(src_region, BTreeSet::new()); } else { new_value.src_to_dst.remove(&src_region); diff --git a/src/mito2/src/engine/region_hook.rs b/src/mito2/src/engine/region_hook.rs index 2758d4a88f4..d3eff73c695 100644 --- a/src/mito2/src/engine/region_hook.rs +++ b/src/mito2/src/engine/region_hook.rs @@ -154,6 +154,7 @@ use store_api::metadata::RegionMetadataRef; use store_api::storage::RegionId; use crate::access_layer::AccessLayerRef; +use crate::error::Result; use crate::manifest::action::{RegionMetaActionList, RemovedFile}; use crate::sst::file::FileMeta; use crate::sst::parquet::SstInfo; @@ -388,42 +389,51 @@ pub trait RegionHook: Send + Sync + Debug { } /// 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. + /// GC pass for a region — live regions (periodic GC) or dropped/repartitioned + /// regions (the global reclamation path). Lets extensions with sidecar files + /// outside mito2's region dir clean up their own residual files. /// - /// `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). + /// Authorization is scoped to stay safe against cross-region / tmp refs mito + /// honors but an extension can't see: + /// - [`RegionGcInfo::terminal`] `false`: clean only sidecar artifacts for the + /// files in [`RegionGcInfo::removed_files`] (plus orphans on a full-listing + /// pass). [`RegionGcInfo::is_region_dropped`] is context, not authorization. + /// - [`RegionGcInfo::terminal`] `true`: all refs released; may remove the + /// whole sidecar dir. + /// + /// Returns `Err` to keep the region un-acknowledged (`need_retry_regions`) + /// so the next pass replays this callback. + /// + /// `region_metadata` is `None` for dropped regions; use `region_id` + + /// `access_layer`. Runs on the background GC task; must be idempotent. async fn on_region_gc( &self, region_id: RegionId, region_metadata: Option<&RegionMetadataRef>, access_layer: &AccessLayerRef, info: &RegionGcInfo<'_>, - ) { + ) -> Result<()> { let _ = (region_id, region_metadata, access_layer, info); + Ok(()) } } /// 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. + /// Files mito2 physically deleted this pass. When [`Self::terminal`] is + /// `false`, cleanup must be scoped to these. 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. + /// `true` when the region is dropped/absent (e.g. a repartitioned source). + /// Context only — not authorization to remove the whole sidecar dir; see + /// [`Self::terminal`]. `region_metadata` is `None` when this is `true`. pub is_region_dropped: bool, - /// Whether this pass did a full object-store listing (vs. fast mode that - /// only reaps manifest-tracked removed files). + /// Whether this pass did a full object-store listing. pub full_file_listing: bool, + /// All refs to the region's files are released (dropped, no cross-region/tmp + /// refs). Only when `true` may an extension remove its whole sidecar dir; + /// otherwise stay scoped to [`Self::removed_files`]. + pub terminal: bool, } pub type RegionHookRef = Arc; diff --git a/src/mito2/src/gc.rs b/src/mito2/src/gc.rs index bda2b0f0de3..4f59848b375 100644 --- a/src/mito2/src/gc.rs +++ b/src/mito2/src/gc.rs @@ -309,6 +309,7 @@ impl LocalGcWorker { let mut deleted_files = HashMap::new(); let mut deleted_indexes = HashMap::new(); let mut processed_regions = HashSet::new(); + let mut need_retry_regions = HashSet::new(); let tmp_ref_files = self.read_tmp_ref_files().await?; for (region_id, region) in &self.regions { let per_region_time = std::time::Instant::now(); @@ -327,23 +328,33 @@ impl LocalGcWorker { .get(region_id) .cloned() .unwrap_or_else(HashSet::new); - let files = self + let outcome = self .do_region_gc(*region_id, region.clone(), &tmp_ref_files) .await?; - let index_files = files + let RegionGcOutcome { + removed_files, + extension_cleanup_failed, + } = outcome; + let index_files = removed_files .iter() .filter_map(|f| f.index_version().map(|v| (f.file_id(), v))) .collect_vec(); - let data_files = files + let data_files = removed_files .into_iter() .filter_map(|f| match f { RemovedFile::File(file_id, _) => Some(file_id), RemovedFile::Index(_, _) => None, }) .collect(); - deleted_files.insert(*region_id, data_files); - deleted_indexes.insert(*region_id, index_files); - processed_regions.insert(*region_id); + // Don't acknowledge the region as processed until extension cleanup + // succeeds; retry it next pass instead. + if extension_cleanup_failed { + need_retry_regions.insert(*region_id); + } else { + deleted_files.insert(*region_id, data_files); + deleted_indexes.insert(*region_id, index_files); + processed_regions.insert(*region_id); + } debug!( "GC for region {} took {} secs.", region_id, @@ -357,13 +368,26 @@ impl LocalGcWorker { let report = GcReport { deleted_files, deleted_indexes, - need_retry_regions: HashSet::new(), + need_retry_regions, processed_regions, }; Ok(report) } } +/// Per-region outcome of [`LocalGcWorker::do_region_gc`]. +/// +/// `extension_cleanup_failed` records whether a registered extension's +/// [`RegionHook::on_region_gc`] could not finish; the caller must then keep the +/// region un-acknowledged (retry set) so the next GC pass replays the callback. +pub(crate) struct RegionGcOutcome { + /// Files physically deleted this pass. + pub removed_files: Vec, + /// `true` if a registered extension's `on_region_gc` returned `Err`; the + /// caller retries the region next pass. + pub extension_cleanup_failed: bool, +} + impl LocalGcWorker { /// concurrency of listing files per region. /// This is used to limit the number of concurrent listing operations and speed up listing @@ -390,7 +414,7 @@ impl LocalGcWorker { region_id: RegionId, region: Option, tmp_ref_files: &HashSet, - ) -> Result> { + ) -> Result { let mode = if self.full_file_listing { "full_listing" } else { @@ -434,7 +458,10 @@ impl LocalGcWorker { GC_ERRORS_TOTAL .with_label_values(&["manifest_mismatch"]) .inc(); - return Ok(vec![]); + return Ok(RegionGcOutcome { + removed_files: vec![], + extension_cleanup_failed: false, + }); } Some(manifest) } else { @@ -537,7 +564,53 @@ impl LocalGcWorker { "Successfully deleted {} unused files for region {}", unused_file_cnt, region_id ); - if let Some(region) = ®ion { + + // Notify extensions so they can clean up sidecar files. Fire when there + // are removed files, or on a full-listing pass (lets extensions reconcile + // orphans even when mito deleted nothing). `terminal` authorizes whole- + // sidecar-dir removal only for a dropped region with no remaining refs. + let terminal = is_region_dropped + && !self + .file_ref_manifest + .cross_region_refs + .contains_key(®ion_id) + && !self.file_ref_manifest.file_refs.contains_key(®ion_id); + + let extension_cleanup_failed = if let Some(hook) = &self.hook + && (!deletable_files.is_empty() || self.full_file_listing) + { + let region_metadata = region.as_ref().map(|r| r.metadata()); + let result = 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, + terminal, + }, + ) + .await; + if let Err(err) = result { + warn!( + err; + "Region hook on_region_gc failed for region {}, will retry on the next GC pass", + region_id + ); + true + } else { + false + } + } else { + false + }; + + // Defer clearing the manifest tracking until the extension succeeds, so + // a failed live-region cleanup is replayed next pass. (Dropped regions + // have no manifest.) + if !extension_cleanup_failed && let Some(region) = ®ion { let _update_timer = GC_DURATION_SECONDS .with_label_values(&["update_manifest"]) .start_timer(); @@ -545,30 +618,10 @@ 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. Skip the invocation when there is - // nothing to report — a live region's periodic pass that deleted no - // files — to avoid unnecessary hook overhead/I/O. A dropped region - // always notifies (the reclamation path extensions act on). - if let Some(hook) = &self.hook - && (!deletable_files.is_empty() || is_region_dropped) - { - 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) + Ok(RegionGcOutcome { + removed_files: deletable_files, + extension_cleanup_failed, + }) } #[common_telemetry::tracing::instrument( diff --git a/src/mito2/src/gc/worker_test.rs b/src/mito2/src/gc/worker_test.rs index c7fbd0aa615..939d63b5f72 100644 --- a/src/mito2/src/gc/worker_test.rs +++ b/src/mito2/src/gc/worker_test.rs @@ -32,6 +32,7 @@ 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::error::UnexpectedSnafu; use crate::gc::{GcConfig, LocalGcWorker, should_delete_file}; use crate::manifest::action::RemovedFile; use crate::region::MitoRegionRef; @@ -746,11 +747,16 @@ async fn test_file_in_tmp_ref_old_mtime_kept() { ); } -/// A region hook that only records `on_region_gc` calls, for testing the GC hook. +/// A region hook that records `on_region_gc` calls, for testing the GC hook. +/// `fail_remaining` simulates a transient extension-cleanup failure: while it +/// is greater than zero, `on_region_gc` returns `Err` (after decrementing), so +/// the GC retry path can be exercised. #[derive(Debug, Default)] struct GcCountingHook { gc_calls: AtomicUsize, last_is_region_dropped: AtomicBool, + last_terminal: AtomicBool, + fail_remaining: AtomicUsize, } #[async_trait] @@ -761,16 +767,28 @@ impl RegionHook for GcCountingHook { _region_metadata: Option<&RegionMetadataRef>, _access_layer: &AccessLayerRef, info: &RegionGcInfo<'_>, - ) { + ) -> Result<(), crate::error::Error> { self.gc_calls.fetch_add(1, Ordering::Relaxed); self.last_is_region_dropped .store(info.is_region_dropped, Ordering::Relaxed); + self.last_terminal.store(info.terminal, Ordering::Relaxed); + // Atomically decrement-and-return the previous value; fail while > 0. + let prev = self.fail_remaining.fetch_sub(1, Ordering::Relaxed); + if prev > 0 { + return UnexpectedSnafu { + reason: "simulated extension cleanup failure".to_string(), + } + .fail(); + } + Ok(()) } } -/// `on_region_gc` is **skipped** for a live region when the GC pass deleted no -/// files — the common case for a periodic pass — to avoid unnecessary hook -/// overhead/I/O. A freshly-flushed region has no deletable files. +/// `on_region_gc` is **skipped** for a live region in **fast mode** (no full +/// listing) when the GC pass deleted no files — the common case for a periodic +/// pass — to avoid unnecessary hook overhead/I/O. A freshly-flushed region has +/// no deletable files. (A full-listing pass instead always fires the hook; see +/// `test_on_region_gc_fires_on_full_listing_without_deleted_files`.) #[tokio::test] async fn test_on_region_gc_skipped_when_no_files_deleted() { init_default_ut_logging(); @@ -832,13 +850,13 @@ async fn test_on_region_gc_skipped_when_no_files_deleted() { cross_region_refs: HashMap::new(), }; - let gc_worker = create_gc_worker(&engine, regions, &file_ref_manifest, true).await; + let gc_worker = create_gc_worker(&engine, regions, &file_ref_manifest, false).await; gc_worker.run().await.unwrap(); assert_eq!( hook.gc_calls.load(Ordering::Relaxed), 0, - "on_region_gc must be skipped when a live region's GC pass deleted no files" + "on_region_gc must be skipped when a live region's fast-mode GC pass deleted no files" ); } @@ -1030,4 +1048,189 @@ async fn test_on_region_gc_fires_for_dropped_region() { hook.last_is_region_dropped.load(Ordering::Relaxed), "a dropped region's GC pass must report is_region_dropped=true" ); + assert!( + hook.last_terminal.load(Ordering::Relaxed), + "a dropped region with no cross-region/tmp refs is terminally cleaned (terminal=true)" + ); +} + +/// `on_region_gc` **fires** on a full-listing pass even when mito itself deleted +/// no files, so an extension can reconcile sidecar-only orphans (failed writes, +/// leftovers from a previous cleanup). Contrast with the fast-mode skip in +/// `test_on_region_gc_skipped_when_no_files_deleted`. +#[tokio::test] +async fn test_on_region_gc_fires_on_full_listing_without_deleted_files() { + 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, 3); + 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; + + // No truncate: there are no removed files to reclaim this pass. + 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 must fire on a full-listing pass even when no files were deleted" + ); + assert!( + !hook.last_is_region_dropped.load(Ordering::Relaxed), + "a live region's GC pass must report is_region_dropped=false" + ); + assert!( + !hook.last_terminal.load(Ordering::Relaxed), + "a live region is never terminal" + ); +} + +/// A failing `on_region_gc` must keep the region un-acknowledged: it lands in +/// `need_retry_regions` and is excluded from `processed_regions` / `deleted_files`, +/// so metasrv does not treat it as fully cleaned and the next GC pass replays +/// the callback. +#[tokio::test] +async fn test_on_region_gc_failure_routes_region_to_retry() { + init_default_ut_logging(); + + let mut env = TestEnv::new().await; + + // Fail the first (and only, in this test) extension cleanup. + let hook = Arc::new(GcCountingHook { + fail_remaining: AtomicUsize::new(1), + ..Default::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, 4); + 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; + // Truncate so the flushed SST becomes deletable. + engine + .handle_request( + region_id, + RegionRequest::Truncate(store_api::region_request::RegionTruncateRequest::All), + ) + .await + .unwrap(); + + 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; + let report = gc_worker.run().await.unwrap(); + + assert!( + hook.gc_calls.load(Ordering::Relaxed) >= 1, + "on_region_gc should fire when the GC pass deleted files" + ); + assert!( + report.need_retry_regions.contains(®ion_id), + "a failed extension cleanup must keep the region in need_retry_regions" + ); + assert!( + !report.processed_regions.contains(®ion_id), + "a failed extension cleanup must not acknowledge the region as processed" + ); + assert!( + !report.deleted_files.contains_key(®ion_id), + "a failed extension cleanup must not report deleted_files for the region" + ); }