fix

Author: zhyass

src/query/service/tests/it/storages/fuse/operations/mutation/recluster_mutator.rs

@@ -34,7 +34,6 @@ use databend_common_storages_fuse::io::TableMetaLocationGenerator;
 use databend_common_storages_fuse::operations::ReclusterMode;
 use databend_common_storages_fuse::operations::ReclusterMutator;
 use databend_common_storages_fuse::pruning::create_segment_location_vector;
-use databend_common_storages_fuse::statistics::reducers::merge_statistics_mut;
 use databend_common_storages_fuse::statistics::reducers::reduce_block_metas;
 use databend_query::sessions::TableContext;
 use databend_query::sessions::TableContextSettings;
@@ -44,7 +43,6 @@ use databend_storages_common_table_meta::meta;
 use databend_storages_common_table_meta::meta::BlockMeta;
 use databend_storages_common_table_meta::meta::ClusterStatistics;
 use databend_storages_common_table_meta::meta::SegmentInfo;
-use databend_storages_common_table_meta::meta::Statistics;
 use databend_storages_common_table_meta::meta::Versioned;
 use rand::Rng;
 use rand::thread_rng;
@@ -384,7 +382,7 @@ async fn test_recluster_mutator_limits_removed_segments_to_selected_blocks() ->
 }
 
 #[tokio::test(flavor = "multi_thread")]
-async fn test_recluster_mutator_block_select() -> anyhow::Result<()> {
+async fn test_recluster_mutator_compacts_small_overlapping_blocks() -> anyhow::Result<()> {
     let fixture = TestFixture::setup().await?;
     let ctx = fixture.new_query_ctx().await?;
     let location_generator = TableMetaLocationGenerator::new("_prefix".to_owned());
@@ -428,7 +426,8 @@ async fn test_recluster_mutator_block_select() -> anyhow::Result<()> {
 }
 
 #[tokio::test(flavor = "multi_thread")]
-async fn test_recluster_mutator_selects_multiple_segment_windows() -> anyhow::Result<()> {
+async fn test_recluster_mutator_select_segments_covers_candidates_and_merges_tail()
+-> anyhow::Result<()> {
     let fixture = TestFixture::setup().await?;
     let ctx = fixture.new_query_ctx().await?;
     ctx.get_settings().set_max_threads(2)?;
@@ -492,6 +491,7 @@ async fn test_recluster_mutator_selects_multiple_segment_windows() -> anyhow::Re
         })
         .collect::<Vec<_>>();
     // Windows are segment-disjoint and together cover every candidate segment.
+    assert_eq!(window_index_sets.len(), 2);
     let mut covered = HashSet::new();
     let mut total = 0;
     for window in &window_index_sets {
@@ -928,12 +928,7 @@ async fn test_final_recluster_groups_mature_levels_in_two_level_bands() -> anyho
     assert_eq!(block_num, 0);
     assert!(parts.tasks.is_empty());
 
-    Ok(())
-}
-
-#[tokio::test(flavor = "multi_thread")]
-async fn test_final_recluster_skips_high_level_two_block_batch() -> anyhow::Result<()> {
-    let thresholds = BlockThresholds::new(1000, 100, 100, 10);
+    // Two high-level blocks alone are not enough to produce useful FINAL work.
     let (block_num, parts) =
         target_select_segments_by_level_with_mode(&[(2, 2)], thresholds, 1, ReclusterMode::Final)
             .await?;
@@ -998,18 +993,6 @@ async fn test_safety_for_recluster() -> anyhow::Result<()> {
 
         eprintln!("data ready");
 
-        let mut summary = Statistics::default();
-        for seg in &segment_infos {
-            merge_statistics_mut(&mut summary, &seg.summary, Some(cluster_key_id));
-        }
-
-        let mut block_ids = HashSet::new();
-        for seg in &segment_infos {
-            for b in &seg.blocks {
-                block_ids.insert(b.location.clone());
-            }
-        }
-
         let ctx: Arc<dyn TableContext> = ctx.clone();
         let segment_locations = create_segment_location_vector(locations.clone(), None);
         let compact_segments = FuseTable::segment_pruning(

src/query/storages/fuse/src/operations/recluster.rs

@@ -170,7 +170,10 @@ impl FuseTable {
                 // scope stays bounded instead of growing across the whole chunk.
                 let produced_tasks = !recluster_parts.tasks.is_empty();
                 merge_recluster_parts(&mut parts, recluster_parts, mutator.cluster_key_id);
-                if !produced_tasks {
+                // `LIMIT` bounds the total segments rewritten per invocation, so
+                // stop after the first productive window instead of accumulating
+                // more across windows.
+                if !produced_tasks || limit.is_some() {
                     break;
                 }
             }