chore: after rebase fix

Signed-off-by: discord9 <[email protected]>

Author: discord9

src/meta-srv/src/gc/ctx.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-
-
 use std::collections::{HashMap, HashSet};
 use std::time::Duration;
 
@@ -34,11 +32,6 @@ use table::metadata::TableId;
 use crate::cluster::MetaPeerClientRef;
 use crate::error::{self, Result, TableMetadataManagerSnafu, UnexpectedSnafu};
 use crate::gc::Region2Peers;
-
-use crate::cluster::MetaPeerClientRef;
-use crate::error;
-use crate::error::{Result, TableMetadataManagerSnafu};
-use crate::gc::handler::Region2Peers;
 use crate::gc::procedure::GcRegionProcedure;
 use crate::handler::HeartbeatMailbox;
 use crate::service::mailbox::{Channel, MailboxRef};

src/meta-srv/src/gc/handler.rs

@@ -19,6 +19,7 @@ use common_meta::key::table_route::PhysicalTableRouteValue;
 use common_meta::peer::Peer;
 use common_telemetry::{debug, error, info, warn};
 use futures::StreamExt;
+use itertools::Itertools;
 use store_api::storage::{FileRefsManifest, GcReport, RegionId};
 use table::metadata::TableId;
 use tokio::time::sleep;
@@ -88,8 +89,8 @@ impl GcScheduler {
     pub(crate) async fn find_related_regions(
         &self,
         candidate_region_ids: &[RegionId],
-    ) -> Result<Vec<RegionId>> {
-        Ok(candidate_region_ids.to_vec())
+    ) -> Result<HashMap<RegionId, Vec<RegionId>>> {
+        Ok(candidate_region_ids.iter().map(|&r| (r, vec![r])).collect())
     }
 
     /// Aggregate GC candidates by their corresponding datanode peer.
@@ -235,15 +236,18 @@ impl GcScheduler {
 
         let all_region_ids: Vec<RegionId> = candidates.iter().map(|(_, c)| c.region_id).collect();
 
+        let all_related_regions = self.find_related_regions(&all_region_ids).await?;
+
         let (region_to_peer, _) = self
-            .discover_datanodes_for_regions(&all_region_ids, 0)
+            .discover_datanodes_for_regions(&all_related_regions.keys().cloned().collect_vec(), 0)
             .await?;
 
         // Step 1: Get file references for all regions on this datanode
         let file_refs_manifest = self
             .ctx
             .get_file_references(
                 &all_region_ids,
+                all_related_regions,
                 &region_to_peer,
                 self.config.mailbox_timeout,
             )
@@ -407,7 +411,7 @@ impl GcScheduler {
             // Step 2: Process retry regions by calling gc_regions on rediscovered datanodes (batched)
             let retry_result = self
                 .process_retry_regions_by_peers(region_to_peer, peer_to_regions, retry_round)
-                .await;
+                .await?;
 
             // Merge the retry results into the final report
             final_report.merge(retry_result);
@@ -429,19 +433,29 @@ impl GcScheduler {
         Ok(final_report)
     }
 
-    /// Discover datanodes for the given regions by fetching table routes in batches.
+    /// Discover datanodes for the given regions(and it's related regions) by fetching table routes in batches.
     /// Returns mappings from region to peer(leader, Vec<followers>) and peer to regions.
     async fn discover_datanodes_for_regions(
         &self,
         regions: &[RegionId],
         retry_round: usize,
     ) -> Result<(Region2Peers, Peer2Regions)> {
+        let all_related_regions = self
+            .find_related_regions(&regions)
+            .await?
+            .into_iter()
+            .map(|(k, mut v)| {
+                v.push(k);
+                v
+            })
+            .flatten()
+            .collect_vec();
         let mut region_to_peer = HashMap::new();
         let mut peer_to_regions = HashMap::new();
 
         // Group regions by table ID for batch processing
         let mut table_to_regions: HashMap<TableId, Vec<RegionId>> = HashMap::new();
-        for &region_id in regions {
+        for region_id in all_related_regions {
             let table_id = region_id.table_id();
             table_to_regions
                 .entry(table_id)
@@ -520,7 +534,7 @@ impl GcScheduler {
         region_to_peer: Region2Peers,
         peer_to_regions: Peer2Regions,
         retry_round: usize,
-    ) -> GcJobReport {
+    ) -> Result<GcJobReport> {
         let mut round_report = GcJobReport::default();
 
         for (peer, regions) in peer_to_regions {
@@ -531,17 +545,17 @@ impl GcScheduler {
                 retry_round
             );
 
-            let all_related_regions = self
-                .find_related_regions(&regions.iter().cloned().collect::<Vec<_>>())
-                .await
-                .unwrap_or_default();
+            let query_regions = regions.clone().into_iter().collect::<Vec<_>>();
+
+            let all_related_regions = self.find_related_regions(&query_regions).await?;
 
             // Get fresh file references for these regions
             let file_refs_manifest = match self
                 .ctx
                 .get_file_references(
-                    &all_related_regions,
-                    &region_to_peer,
+                    &query_regions,
+                    all_related_regions,
+                    &region_to_peer, // FIXME: update region_to_peer?
                     self.config.mailbox_timeout,
                 )
                 .await
@@ -576,7 +590,7 @@ impl GcScheduler {
             }
         }
 
-        round_report
+        Ok(round_report)
     }
 
     /// Process a batch of regions on a single peer.

src/meta-srv/src/gc/mock.rs

@@ -212,7 +212,8 @@ impl SchedulerCtx for MockSchedulerCtx {
 
     async fn get_file_references(
         &self,
-        region_ids: &[RegionId],
+        query_regions: &[RegionId],
+        related_regions: HashMap<RegionId, Vec<RegionId>>,
         region_to_peer: &Region2Peers,
         _timeout: Duration,
     ) -> Result<FileRefsManifest> {
@@ -222,14 +223,14 @@ impl SchedulerCtx for MockSchedulerCtx {
         if let Some(error) = self.get_file_references_error.lock().unwrap().take() {
             return Err(error);
         }
-        if region_ids
+        if query_regions
             .iter()
             .any(|region_id| !region_to_peer.contains_key(region_id))
         {
             UnexpectedSnafu {
                 violated: format!(
                     "region_to_peer{region_to_peer:?} does not contain all region_ids requested: {:?}",
-                    region_ids
+                    query_regions
                 ),
             }.fail()?;
         }

src/mito2/src/gc.rs

@@ -322,8 +322,6 @@ impl LocalGcWorker {
         let region_id = manifest.metadata.region_id;
         let current_files = &manifest.files;
 
-        let in_manifest_file_cnt = current_files.len();
-
         let recently_removed_files = self.get_removed_files_expel_times(&manifest).await?;
 
         if recently_removed_files.is_empty() {
@@ -346,8 +344,6 @@ impl LocalGcWorker {
             .chain(tmp_ref_files.clone().into_iter())
             .collect();
 
-        let in_used_file_cnt = in_used.len();
-
         let unused_files = self
             .list_to_be_deleted_files(region_id, &in_used, recently_removed_files, concurrency)
             .await?;