refactor: use last compaction scheduled time instead of finished time

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

src/mito2/src/compaction.rs

@@ -150,6 +150,7 @@ impl CompactionScheduler {
     }
 
     /// Schedules a compaction for the region.
+    /// Returns whether a compaction is scheduled.
     #[allow(clippy::too_many_arguments)]
     pub(crate) async fn schedule_compaction(
         &mut self,
@@ -161,7 +162,7 @@ impl CompactionScheduler {
         manifest_ctx: &ManifestContextRef,
         schema_metadata_manager: SchemaMetadataManagerRef,
         max_parallelism: usize,
-    ) -> Result<()> {
+    ) -> Result<bool> {
         // skip compaction if region is in staging state
         let current_state = manifest_ctx.current_state();
         if current_state == RegionRoleState::Leader(RegionLeaderState::Staging) {
@@ -170,7 +171,7 @@ impl CompactionScheduler {
                 region_id, compact_options
             );
             waiter.send(Ok(0));
-            return Ok(());
+            return Ok(false);
         }
 
         if let Some(status) = self.region_status.get_mut(&region_id) {
@@ -192,7 +193,7 @@ impl CompactionScheduler {
                     );
                 }
             }
-            return Ok(());
+            return Ok(false);
         }
 
         // The region can compact directly.
@@ -220,9 +221,9 @@ impl CompactionScheduler {
                 status.active_compaction = Some(active_compaction);
                 self.region_status.insert(region_id, status);
 
-                Ok(())
+                Ok(true)
             }
-            Ok(None) => Ok(()),
+            Ok(None) => Ok(false),
             Err(e) => Err(e),
         };
 
@@ -334,6 +335,24 @@ 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);
+    }
+
+    /// Schedules next compaction upon a finished compaction.
+    /// Returns whether the compaction is scheduled.
+    pub(crate) async fn schedule_next_compaction(
+        &mut self,
+        region_id: RegionId,
+        manifest_ctx: &ManifestContextRef,
+        schema_metadata_manager: SchemaMetadataManagerRef,
+    ) -> bool {
+        let Some(status) = self.region_status.get_mut(&region_id) else {
+            return false;
+        };
 
         // We should always try to compact the region until picker returns None.
         let request = status.new_compaction_request(
@@ -364,20 +383,21 @@ impl CompactionScheduler {
                     "Successfully scheduled next compaction for region id: {}",
                     region_id
                 );
+                true
             }
             Ok(None) => {
                 // No further compaction tasks can be scheduled; cleanup the `CompactionStatus` for this region.
                 // All DDL requests and pending compaction requests have already been processed.
                 // Safe to remove the region from status tracking.
                 self.region_status.remove(&region_id);
+                false
             }
             Err(e) => {
                 error!(e; "Failed to schedule next compaction for region {}", region_id);
                 self.remove_region_on_failure(region_id, Arc::new(e));
+                false
             }
         }
-
-        Vec::new()
     }
 
     /// Notifies the scheduler that the compaction job is cancelled cooperatively.
@@ -1434,7 +1454,7 @@ mod tests {
         let manifest_ctx = env
             .mock_manifest_context(version_control.current().version.metadata.clone())
             .await;
-        scheduler
+        let scheduled = scheduler
             .schedule_compaction(
                 builder.region_id(),
                 compact_request::Options::Regular(Default::default()),
@@ -1447,6 +1467,7 @@ mod tests {
             )
             .await
             .unwrap();
+        assert!(!scheduled);
         let output = output_rx.await.unwrap().unwrap();
         assert_eq!(output, 0);
         assert!(scheduler.region_status.is_empty());
@@ -1455,7 +1476,7 @@ mod tests {
         let version_control = Arc::new(builder.push_l0_file(0, 1000).build());
         let (output_tx, output_rx) = oneshot::channel();
         let waiter = OptionOutputTx::from(output_tx);
-        scheduler
+        let scheduled = scheduler
             .schedule_compaction(
                 builder.region_id(),
                 compact_request::Options::Regular(Default::default()),
@@ -1468,11 +1489,67 @@ mod tests {
             )
             .await
             .unwrap();
+        assert!(!scheduled);
         let output = output_rx.await.unwrap().unwrap();
         assert_eq!(output, 0);
         assert!(scheduler.region_status.is_empty());
     }
 
+    #[tokio::test]
+    async fn test_schedule_compaction_returns_true_when_task_scheduled() {
+        let job_scheduler = Arc::new(VecScheduler::default());
+        let env = SchedulerEnv::new().await.scheduler(job_scheduler.clone());
+        let (tx, _rx) = mpsc::channel(4);
+        let mut scheduler = env.mock_compaction_scheduler(tx);
+        let mut builder = VersionControlBuilder::new();
+        let region_id = builder.region_id();
+        let end = 1000 * 1000;
+        // Five overlapping L0 files are enough for the regular picker to create a task.
+        let version_control = Arc::new(
+            builder
+                .push_l0_file(0, end)
+                .push_l0_file(10, end)
+                .push_l0_file(50, end)
+                .push_l0_file(80, end)
+                .push_l0_file(90, end)
+                .build(),
+        );
+        let manifest_ctx = env
+            .mock_manifest_context(version_control.current().version.metadata.clone())
+            .await;
+        let (schema_metadata_manager, kv_backend) = mock_schema_metadata_manager();
+        schema_metadata_manager
+            .register_region_table_info(
+                region_id.table_id(),
+                "test_table",
+                "test_catalog",
+                "test_schema",
+                None,
+                kv_backend,
+            )
+            .await;
+
+        let scheduled = scheduler
+            .schedule_compaction(
+                region_id,
+                Options::Regular(Default::default()),
+                &version_control,
+                &env.access_layer,
+                OptionOutputTx::none(),
+                &manifest_ctx,
+                schema_metadata_manager,
+                1,
+            )
+            .await
+            .unwrap();
+
+        // The boolean result is what the worker uses to decide whether to update
+        // last_schedule_compaction_millis.
+        assert!(scheduled);
+        assert_eq!(1, job_scheduler.num_jobs());
+        assert!(scheduler.region_status.contains_key(&region_id));
+    }
+
     #[tokio::test]
     async fn test_schedule_on_finished() {
         common_telemetry::init_default_ut_logging();
@@ -1510,7 +1587,7 @@ mod tests {
         let manifest_ctx = env
             .mock_manifest_context(version_control.current().version.metadata.clone())
             .await;
-        scheduler
+        let scheduled = scheduler
             .schedule_compaction(
                 region_id,
                 compact_request::Options::Regular(Default::default()),
@@ -1524,6 +1601,7 @@ mod tests {
             .await
             .unwrap();
         // Should schedule 1 compaction.
+        assert!(scheduled);
         assert_eq!(1, scheduler.region_status.len());
         assert_eq!(1, job_scheduler.num_jobs());
         let data = version_control.current();
@@ -1542,7 +1620,7 @@ mod tests {
         );
         // The task is pending.
         let (tx, _rx) = oneshot::channel();
-        scheduler
+        let scheduled = scheduler
             .schedule_compaction(
                 region_id,
                 compact_request::Options::Regular(Default::default()),
@@ -1555,6 +1633,7 @@ mod tests {
             )
             .await
             .unwrap();
+        assert!(!scheduled);
         assert_eq!(1, scheduler.region_status.len());
         assert_eq!(1, job_scheduler.num_jobs());
         assert!(
@@ -1570,6 +1649,10 @@ mod tests {
         scheduler
             .on_compaction_finished(region_id, &manifest_ctx, schema_metadata_manager.clone())
             .await;
+        let scheduled = scheduler
+            .schedule_next_compaction(region_id, &manifest_ctx, schema_metadata_manager.clone())
+            .await;
+        assert!(scheduled);
         assert_eq!(1, scheduler.region_status.len());
         assert_eq!(2, job_scheduler.num_jobs());
 
@@ -1582,7 +1665,7 @@ mod tests {
         );
         let (tx, _rx) = oneshot::channel();
         // The task is pending.
-        scheduler
+        let scheduled = scheduler
             .schedule_compaction(
                 region_id,
                 compact_request::Options::Regular(Default::default()),
@@ -1595,6 +1678,7 @@ mod tests {
             )
             .await
             .unwrap();
+        assert!(!scheduled);
         assert_eq!(2, job_scheduler.num_jobs());
         assert!(
             !scheduler
@@ -2328,6 +2412,15 @@ mod tests {
             .await;
 
         assert!(pending_ddls.is_empty());
+        assert!(scheduler.region_status.contains_key(&region_id));
+
+        let (schema_metadata_manager, _kv_backend) = mock_schema_metadata_manager();
+        // With no compactable files, next scheduling returns false and removes
+        // the status without creating a background task.
+        let scheduled = scheduler
+            .schedule_next_compaction(region_id, &manifest_ctx, schema_metadata_manager)
+            .await;
+        assert!(!scheduled);
         assert!(!scheduler.region_status.contains_key(&region_id));
     }
 
@@ -2370,6 +2463,14 @@ mod tests {
             .await;
 
         assert!(pending_ddls.is_empty());
+        assert!(scheduler.region_status.contains_key(&region_id));
+
+        let (schema_metadata_manager, _kv_backend) = mock_schema_metadata_manager();
+        // The failing scheduler simulates a submit error; callers must see false.
+        let scheduled = scheduler
+            .schedule_next_compaction(region_id, &manifest_ctx, schema_metadata_manager)
+            .await;
+        assert!(!scheduled);
         assert!(!scheduler.region_status.contains_key(&region_id));
     }
 

src/mito2/src/engine/edit_region_test.rs

@@ -67,7 +67,8 @@ async fn test_edit_region_schedule_compaction_with_format(flat_format: bool) {
         default_flat_format: flat_format,
         ..Default::default()
     };
-    let time_provider = Arc::new(MockTimeProvider::new(current_time_millis()));
+    let initial_time = current_time_millis();
+    let time_provider = Arc::new(MockTimeProvider::new(initial_time));
     let engine = env
         .create_engine_with_time(
             config.clone(),
@@ -99,6 +100,8 @@ async fn test_edit_region_schedule_compaction_with_format(flat_format: bool) {
         .await
         .unwrap();
     let region = engine.get_region(region_id).unwrap();
+    let initial_schedule_time = region.last_schedule_compaction_millis();
+    assert_eq!(initial_time, initial_schedule_time);
 
     let new_edit = || RegionEdit {
         files_to_add: vec![FileMeta {
@@ -121,10 +124,14 @@ async fn test_edit_region_schedule_compaction_with_format(flat_format: bool) {
     // Asserts that the compaction of the region is not scheduled,
     // because the minimum time interval between two compactions is not passed.
     assert_eq!(rx.try_recv(), Err(oneshot::error::TryRecvError::Empty));
+    assert_eq!(
+        initial_schedule_time,
+        region.last_schedule_compaction_millis()
+    );
 
     // Simulates the time has passed the min compaction interval,
-    time_provider
-        .set_now(current_time_millis() + config.min_compaction_interval.as_millis() as i64);
+    let next_schedule_time = initial_time + config.min_compaction_interval.as_millis() as i64;
+    time_provider.set_now(next_schedule_time);
     // ... then edits the region again,
     engine
         .edit_region(region.region_id, new_edit())
@@ -136,6 +143,7 @@ async fn test_edit_region_schedule_compaction_with_format(flat_format: bool) {
         .unwrap()
         .unwrap();
     assert_eq!(region_id, actual);
+    assert_eq!(next_schedule_time, region.last_schedule_compaction_millis());
 }
 
 #[tokio::test]

src/mito2/src/region.rs

@@ -140,8 +140,8 @@ pub struct MitoRegion {
     pub(crate) provider: Provider,
     /// Last flush time in millis.
     last_flush_millis: AtomicI64,
-    /// Last compaction time in millis.
-    last_compaction_millis: AtomicI64,
+    /// Last schedule compaction time in millis.
+    last_schedule_compaction_millis: AtomicI64,
     /// Provider to get current time.
     time_provider: TimeProviderRef,
     /// The topic's latest entry id since the region's last flushing.
@@ -234,15 +234,16 @@ impl MitoRegion {
         self.last_flush_millis.store(now, Ordering::Relaxed);
     }
 
-    /// Returns last compaction timestamp in millis.
-    pub(crate) fn last_compaction_millis(&self) -> i64 {
-        self.last_compaction_millis.load(Ordering::Relaxed)
+    /// Returns last schedule compaction timestamp in millis.
+    pub(crate) fn last_schedule_compaction_millis(&self) -> i64 {
+        self.last_schedule_compaction_millis.load(Ordering::Relaxed)
     }
 
-    /// Update compaction time to current time.
-    pub(crate) fn update_compaction_millis(&self) {
+    /// Update schedule compaction time to current time.
+    pub(crate) fn update_schedule_compaction_millis(&self) {
         let now = self.time_provider.current_time_millis();
-        self.last_compaction_millis.store(now, Ordering::Relaxed);
+        self.last_schedule_compaction_millis
+            .store(now, Ordering::Relaxed);
     }
 
     /// Returns the table dir.
@@ -1658,7 +1659,7 @@ mod tests {
             file_purger: crate::test_util::new_noop_file_purger(),
             provider: Provider::noop_provider(),
             last_flush_millis: Default::default(),
-            last_compaction_millis: Default::default(),
+            last_schedule_compaction_millis: Default::default(),
             time_provider: Arc::new(StdTimeProvider),
             topic_latest_entry_id: Default::default(),
             written_bytes: Arc::new(AtomicU64::new(0)),
@@ -2015,7 +2016,7 @@ mod tests {
             file_purger: crate::test_util::new_noop_file_purger(),
             provider: Provider::noop_provider(),
             last_flush_millis: Default::default(),
-            last_compaction_millis: Default::default(),
+            last_schedule_compaction_millis: Default::default(),
             time_provider: Arc::new(StdTimeProvider),
             topic_latest_entry_id: Default::default(),
             written_bytes: Arc::new(AtomicU64::new(0)),

src/mito2/src/region/opener.rs

@@ -345,7 +345,7 @@ impl RegionOpener {
             ),
             provider,
             last_flush_millis: AtomicI64::new(now),
-            last_compaction_millis: AtomicI64::new(now),
+            last_schedule_compaction_millis: AtomicI64::new(now),
             time_provider: self.time_provider.clone(),
             topic_latest_entry_id: AtomicU64::new(0),
             written_bytes: Arc::new(AtomicU64::new(0)),
@@ -581,7 +581,7 @@ impl RegionOpener {
             file_purger,
             provider: provider.clone(),
             last_flush_millis: AtomicI64::new(now),
-            last_compaction_millis: AtomicI64::new(now),
+            last_schedule_compaction_millis: AtomicI64::new(now),
             time_provider: self.time_provider.clone(),
             topic_latest_entry_id: AtomicU64::new(topic_latest_entry_id),
             written_bytes: Arc::new(AtomicU64::new(0)),

src/mito2/src/worker/handle_compaction.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use api::v1::region::compact_request;
-use common_telemetry::{error, info, warn};
+use common_telemetry::{error, info};
 use store_api::logstore::LogStore;
 use store_api::region_request::RegionCompactRequest;
 use store_api::storage::RegionId;
@@ -80,7 +80,6 @@ impl<S> RegionWorkerLoop<S> {
                 return;
             }
         };
-        region.update_compaction_millis();
 
         region.version_control.apply_edit(
             Some(request.edit.clone()),
@@ -118,6 +117,26 @@ 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
+        {
+            if self
+                .compaction_scheduler
+                .schedule_next_compaction(
+                    region_id,
+                    &region.manifest_ctx,
+                    self.schema_metadata_manager.clone(),
+                )
+                .await
+            {
+                region.update_schedule_compaction_millis();
+            }
+        } else {
+            self.compaction_scheduler
+                .remove_compaction_status(region_id);
+        }
     }
 
     pub(crate) async fn handle_compaction_cancelled(
@@ -160,9 +179,10 @@ impl<S> RegionWorkerLoop<S> {
             return;
         }
         let now = self.time_provider.current_time_millis();
-        if now - region.last_compaction_millis()
+        if now - region.last_schedule_compaction_millis()
             >= self.config.min_compaction_interval.as_millis() as i64
-            && let Err(e) = self
+        {
+            match self
                 .compaction_scheduler
                 .schedule_compaction(
                     region.region_id,
@@ -175,11 +195,13 @@ impl<S> RegionWorkerLoop<S> {
                     1, // Default for automatic compaction
                 )
                 .await
-        {
-            warn!(
-                "Failed to schedule compaction for region: {}, err: {}",
-                region.region_id, e
-            );
+            {
+                Ok(true) => region.update_schedule_compaction_millis(),
+                Ok(false) => {}
+                Err(e) => {
+                    error!(e; "Failed to schedule compaction for region: {}", region.region_id)
+                }
+            }
         }
     }
 }