From 67deecf7de448306023e9050545a22b967b42f80 Mon Sep 17 00:00:00 2001 From: luofucong Date: Wed, 27 May 2026 11:47:44 +0800 Subject: [PATCH] x Signed-off-by: luofucong --- src/mito2/src/compaction.rs | 18 ++++++++++++++++-- src/mito2/src/worker/handle_compaction.rs | 18 ++++++++++++++++++ 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/src/mito2/src/compaction.rs b/src/mito2/src/compaction.rs index 5cf3c444a8..d34454322a 100644 --- a/src/mito2/src/compaction.rs +++ b/src/mito2/src/compaction.rs @@ -334,6 +334,22 @@ impl CompactionScheduler { // And skip try to schedule next compaction task. return pending_ddl_requests; } + Vec::new() + } + + pub(crate) fn remove_compaction_status(&mut self, region_id: RegionId) { + self.region_status.remove(®ion_id); + } + + pub(crate) async fn schedule_next_compaction( + &mut self, + region_id: RegionId, + manifest_ctx: &ManifestContextRef, + schema_metadata_manager: SchemaMetadataManagerRef, + ) { + let Some(status) = self.region_status.get_mut(®ion_id) else { + return; + }; // We should always try to compact the region until picker returns None. let request = status.new_compaction_request( @@ -376,8 +392,6 @@ impl CompactionScheduler { self.remove_region_on_failure(region_id, Arc::new(e)); } } - - Vec::new() } /// Notifies the scheduler that the compaction job is cancelled cooperatively. diff --git a/src/mito2/src/worker/handle_compaction.rs b/src/mito2/src/worker/handle_compaction.rs index 05623ae7df..de37d7bf18 100644 --- a/src/mito2/src/worker/handle_compaction.rs +++ b/src/mito2/src/worker/handle_compaction.rs @@ -117,6 +117,24 @@ impl RegionWorkerLoop { ) .await; self.handle_ddl_requests(&mut pending_ddls).await; + + let now = self.time_provider.current_time_millis(); + if now - region.last_schedule_compaction_millis() + >= self.config.min_compaction_interval.as_millis() as i64 + { + region.update_schedule_compaction_millis(); + + self.compaction_scheduler + .schedule_next_compaction( + region_id, + ®ion.manifest_ctx, + self.schema_metadata_manager.clone(), + ) + .await; + } else { + self.compaction_scheduler + .remove_compaction_status(region_id); + } } pub(crate) async fn handle_compaction_cancelled(