fix: address review comments

This commit is contained in:
Ning Sun
2026-07-20 22:23:23 +08:00
parent 01ac31f134
commit ba6c8d7db2
4 changed files with 338 additions and 63 deletions
+12 -3
View File
@@ -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<RegionId> = self
.data
.gc_report
.as_ref()
.map(|r| r.need_retry_regions.clone())
.unwrap_or_default();
let mut table_ids: HashSet<TableId> = 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);
+29 -19
View File
@@ -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<dyn RegionHook>;
+87 -34
View File
@@ -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<RemovedFile>,
/// `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<MitoRegionRef>,
tmp_ref_files: &HashSet<FileRef>,
) -> Result<Vec<RemovedFile>> {
) -> Result<RegionGcOutcome> {
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) = &region {
// 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(&region_id)
&& !self.file_ref_manifest.file_refs.contains_key(&region_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) = &region {
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(
+210 -7
View File
@@ -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(&region_id),
"a failed extension cleanup must keep the region in need_retry_regions"
);
assert!(
!report.processed_regions.contains(&region_id),
"a failed extension cleanup must not acknowledge the region as processed"
);
assert!(
!report.deleted_files.contains_key(&region_id),
"a failed extension cleanup must not report deleted_files for the region"
);
}