fix: skip sst cache preload for staging manifest (#8147)

Signed-off-by: WenyXu <wenymedia@gmail.com>

src/mito2/src/engine.rs

@@ -459,6 +459,7 @@ impl MitoEngine {
             region_id,
             edit,
             tx,
+            preload_sst_cache: true,
         });
         self.inner
             .workers

src/mito2/src/request.rs

@@ -1105,6 +1105,8 @@ pub(crate) struct CopyRegionFromFinished {
 pub(crate) struct RegionEditRequest {
     pub(crate) region_id: RegionId,
     pub(crate) edit: RegionEdit,
+    /// Whether to preload SST files into the write cache.
+    pub(crate) preload_sst_cache: bool,
     /// The sender to notify the result to the region engine.
     pub(crate) tx: Sender<Result<()>>,
 }

src/mito2/src/worker/handle_apply_staging.rs

@@ -139,6 +139,8 @@ impl<S: LogStore> RegionWorkerLoop<S> {
                     RegionEditRequest {
                         region_id: region.region_id,
                         edit,
+                        // we don't need to preload sst cache during repartition, as it may cause extra network overhead.
+                        preload_sst_cache: false,
                         tx,
                     },
                 )))

src/mito2/src/worker/handle_manifest.rs

@@ -263,6 +263,7 @@ impl<S: LogStore> RegionWorkerLoop<S> {
             region_id: _,
             mut edit,
             tx: sender,
+            preload_sst_cache,
         } = request;
         let file_sequence = region.version_control.committed_sequence() + 1;
         edit.committed_sequence = Some(file_sequence);
@@ -291,8 +292,15 @@ impl<S: LogStore> RegionWorkerLoop<S> {
         // Now the region is in editing state.
         // Updates manifest in background.
         common_runtime::spawn_global(async move {
-            let result =
-                edit_region(&region, edit.clone(), cache_manager, listener, is_staging).await;
+            let result = edit_region(
+                &region,
+                edit.clone(),
+                cache_manager,
+                listener,
+                is_staging,
+                preload_sst_cache,
+            )
+            .await;
             let notify = WorkerRequest::Background {
                 region_id,
                 notify: BackgroundNotify::RegionEdit(RegionEditResult {
@@ -528,9 +536,12 @@ async fn edit_region(
     cache_manager: CacheManagerRef,
     listener: WorkerListener,
     is_staging: bool,
+    preload_sst_cache: bool,
 ) -> Result<()> {
     let region_id = region.region_id;
-    if let Some(write_cache) = cache_manager.write_cache() {
+    if let Some(write_cache) = cache_manager.write_cache()
+        && preload_sst_cache
+    {
         for file_meta in &edit.files_to_add {
             let write_cache = write_cache.clone();
             let layer = region.access_layer.clone();