From dd4698002a3539ed5632e81af5d12ac0bc721531 Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Fri, 30 Jan 2026 18:36:59 +0800 Subject: [PATCH] fix: send get file ref to all regions (#7640) * fix: send get file ref to all regions Signed-off-by: discord9 * refactor: return err on fail to get table route Signed-off-by: discord9 * refactor: batch get Signed-off-by: discord9 * chore: add loggin in all places Signed-off-by: discord9 --------- Signed-off-by: discord9 --- src/common/meta/src/key/table_repart.rs | 9 ++- src/meta-srv/src/gc/procedure.rs | 76 ++++++++++++++++++------- src/mito2/src/gc.rs | 6 +- 3 files changed, 67 insertions(+), 24 deletions(-) diff --git a/src/common/meta/src/key/table_repart.rs b/src/common/meta/src/key/table_repart.rs index 30295f0d616..ce74396dc9b 100644 --- a/src/common/meta/src/key/table_repart.rs +++ b/src/common/meta/src/key/table_repart.rs @@ -15,6 +15,7 @@ use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::fmt::Display; +use common_telemetry::debug; use futures::TryStreamExt; use serde::{Deserialize, Serialize}; use snafu::{OptionExt as _, ResultExt, ensure}; @@ -356,8 +357,8 @@ impl TableRepartManager { return Ok(()); } - if let Some(current) = current { - let (txn, _) = self.build_update_txn(table_id, ¤t, new_value)?; + if let Some(current) = ¤t { + let (txn, _) = self.build_update_txn(table_id, current, new_value)?; let result = self.kv_backend.txn(txn).await?; ensure!( @@ -384,6 +385,10 @@ impl TableRepartManager { ); } + debug!( + "Upserted repartition value for table {}: current: {:?}, new: {:?}", + table_id, current, new_value + ); Ok(()) } diff --git a/src/meta-srv/src/gc/procedure.rs b/src/meta-srv/src/gc/procedure.rs index 0cc9fc5dc11..b612e0fe8f2 100644 --- a/src/meta-srv/src/gc/procedure.rs +++ b/src/meta-srv/src/gc/procedure.rs @@ -29,7 +29,7 @@ use common_procedure::{ Context as ProcedureContext, Error as ProcedureError, LockKey, Procedure, Result as ProcedureResult, Status, }; -use common_telemetry::{error, info, warn}; +use common_telemetry::{debug, error, info, warn}; use itertools::Itertools as _; use serde::{Deserialize, Serialize}; use snafu::ResultExt as _; @@ -277,28 +277,61 @@ impl BatchGcProcedure { } /// Return related regions for the given regions. - /// The returned map uses the source regions (where those files originally came from) as the key, - /// and the destination regions (where files are currently stored) as the value. - /// If a region is not found in the repartition manager, the returned map still have this region as key, - /// just empty value + /// The returned map uses the input region as key, and all other regions + /// from the same table as values (excluding the input region itself). async fn find_related_regions( &self, regions: &[RegionId], ) -> Result>> { - let repart_mgr = self.table_metadata_manager.table_repart_manager(); - let mut related_regions: HashMap> = HashMap::new(); - for src_region in regions { - // TODO(discord9): batch get - if let Some(dst_regions) = repart_mgr - .get_dst_regions(*src_region) - .await - .context(KvBackendSnafu)? - { - related_regions.insert(*src_region, dst_regions.into_iter().collect()); - } else { - related_regions.insert(*src_region, Default::default()); + let table_ids: HashSet = regions.iter().map(|r| r.table_id()).collect(); + let table_ids = table_ids.into_iter().collect::>(); + + let table_routes = self + .table_metadata_manager + .table_route_manager() + .batch_get_physical_table_routes(&table_ids) + .await + .context(TableMetadataManagerSnafu)?; + + if table_routes.len() != table_ids.len() { + // batch_get_physical_table_routes returns a subset on misses; treat that as error + for table_id in &table_ids { + if !table_routes.contains_key(table_id) { + // indicate is a logical table id + return error::InvalidArgumentsSnafu { + err_msg: format!( + "Unexpected logical table route: table {} resolved to physical table regions", + table_id + ), + } + .fail(); + } } } + + let mut table_all_regions: HashMap> = HashMap::new(); + for (table_id, table_route) in table_routes { + let all_regions: HashSet = table_route + .region_routes + .iter() + .map(|r| r.region.id) + .collect(); + + table_all_regions.insert(table_id, all_regions); + } + + let mut related_regions: HashMap> = HashMap::new(); + for region_id in regions { + let table_id = region_id.table_id(); + if let Some(all_regions) = table_all_regions.get(&table_id) { + let mut related: HashSet = all_regions.clone(); + related.remove(region_id); + related_regions.insert(*region_id, related); + } else { + related_regions.insert(*region_id, Default::default()); + } + } + Ok(related_regions) } @@ -554,8 +587,8 @@ impl BatchGcProcedure { } let instruction = GetFileRefs { - query_regions: regions, - related_regions: related_regions_for_peer, + query_regions: regions.clone(), + related_regions: related_regions_for_peer.clone(), }; let reply = send_get_file_refs( @@ -566,6 +599,10 @@ impl BatchGcProcedure { timeout, ) .await?; + debug!( + "Got file references from datanode: {:?}, query_regions: {:?}, related_regions: {:?}, reply: {:?}", + peer, regions, related_regions_for_peer, reply + ); if !reply.success { return error::UnexpectedSnafu { @@ -738,6 +775,7 @@ impl Procedure for BatchGcProcedure { } .fail(); }; + info!("GC report: {:?}", report); Ok(Status::done_with_output(report)) } Err(e) => { diff --git a/src/mito2/src/gc.rs b/src/mito2/src/gc.rs index af77ce20363..6dffb710f6c 100644 --- a/src/mito2/src/gc.rs +++ b/src/mito2/src/gc.rs @@ -441,8 +441,8 @@ impl LocalGcWorker { let unused_file_cnt = deletable_files.len(); - debug!( - "gc: for region{}{region_id}: In manifest files: {}, Tmp ref file cnt: {}, recently removed files: {}, Unused files to delete: {}", + info!( + "gc: for region{}{region_id}: In manifest file cnt: {}, Tmp ref file cnt: {}, recently removed files: {}, Unused files to delete: {:?}", if region.is_none() { "(region dropped)" } else { @@ -451,7 +451,7 @@ impl LocalGcWorker { current_files.map(|c| c.len()).unwrap_or(0), tmp_ref_files.len(), removed_file_cnt, - deletable_files.len() + &deletable_files, ); debug!(