x

Signed-off-by: luofucong <luofc@foxmail.com>

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(&region_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(&region_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.

src/mito2/src/worker/handle_compaction.rs

@@ -117,6 +117,24 @@ impl<S> RegionWorkerLoop<S> {
             )
             .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,
+                    &region.manifest_ctx,
+                    self.schema_metadata_manager.clone(),
+                )
+                .await;
+        } else {
+            self.compaction_scheduler
+                .remove_compaction_status(region_id);
+        }
     }
 
     pub(crate) async fn handle_compaction_cancelled(