refactor: add stop methods for LocalFilePurger and CompactionRegion (#6848)

* refactor: add `LocalFilePurger::stop(&self)` and `stop_file_purger()` of `CompactionRegion`

Signed-off-by: zyy17 <zyylsxm@gmail.com>

* chore: rename methods

Signed-off-by: zyy17 <zyylsxm@gmail.com>

---------

Signed-off-by: zyy17 <zyylsxm@gmail.com>

src/mito2/src/compaction/compactor.rs

@@ -242,9 +242,19 @@ pub async fn open_compaction_region(
 }
 
 impl CompactionRegion {
+    /// Get the file purger of the compaction region.
     pub fn file_purger(&self) -> Option<Arc<LocalFilePurger>> {
         self.file_purger.clone()
     }
+
+    /// Stop the file purger scheduler of the compaction region.
+    pub async fn stop_purger_scheduler(&self) -> Result<()> {
+        if let Some(file_purger) = &self.file_purger {
+            file_purger.stop_scheduler().await
+        } else {
+            Ok(())
+        }
+    }
 }
 
 /// `[MergeOutput]` represents the output of merging SST files.

src/mito2/src/sst/file_purger.rs

@@ -20,6 +20,7 @@ use common_telemetry::{error, info};
 use crate::access_layer::AccessLayerRef;
 use crate::cache::file_cache::{FileType, IndexKey};
 use crate::cache::CacheManagerRef;
+use crate::error::Result;
 use crate::schedule::scheduler::SchedulerRef;
 use crate::sst::file::FileMeta;
 
@@ -76,6 +77,11 @@ impl LocalFilePurger {
             cache_manager,
         }
     }
+
+    /// Stop the scheduler of the file purger.
+    pub async fn stop_scheduler(&self) -> Result<()> {
+        self.scheduler.stop(true).await
+    }
 }
 
 impl FilePurger for LocalFilePurger {