fix: preserve imperfect block statistics for replace and merge (#20187)

* fix: preserve imperfect block statistics for replace and merge

* fix test

Author: zhyass

src/query/storages/fuse/src/operations/common/processors/transform_mutation_aggregator.rs

@@ -22,6 +22,7 @@ use databend_common_base::runtime::execute_futures_in_parallel;
 use databend_common_catalog::plan::BlockMetaWithHLL;
 use databend_common_catalog::table::Table;
 use databend_common_catalog::table_context::TableContext;
+use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::BlockMetaInfoPtr;
 use databend_common_expression::BlockThresholds;
@@ -387,10 +388,19 @@ impl TableMutationAggregator {
                     .collect::<Vec<_>>();
                 Some(SegmentStatistics::new(hlls, Vec::new()).to_bytes()?)
             };
-            let all_perfect = new_blocks.len() > 1;
+            // Only compaction/reclustering output may be force-marked perfect to keep
+            // those operations at a fixed point. REPLACE/MERGE append after-images
+            // must retain the physical perfect-block count from reduce_block_metas.
+            let force_all_blocks_perfect = matches!(
+                self.write_segment_ctx.kind,
+                MutationKind::Compact | MutationKind::Recluster
+            ) && new_blocks.len() > 1;
 
             let ctx = self.write_segment_ctx.clone();
-            tasks.push(async move { ctx.write_segment(new_blocks, new_hlls, all_perfect).await });
+            tasks.push(async move {
+                ctx.write_segment(new_blocks, new_hlls, force_all_blocks_perfect)
+                    .await
+            });
         }
 
         let threads_nums = self.ctx.get_settings().get_max_threads()? as usize;
@@ -507,7 +517,7 @@ impl TableMutationAggregator {
             let write_segment_ctx = self.write_segment_ctx.clone();
 
             tasks.push(async move {
-                let mut all_perfect = false;
+                let mut force_all_blocks_perfect = false;
                 let (new_blocks, new_hlls, origin_summary) = if let Some(loc) = location {
                     // read the old segment
                     let compact_segment_info = SegmentsIO::read_compact_segment(
@@ -559,11 +569,20 @@ impl TableMutationAggregator {
                     let stats = generate_segment_stats(new_hlls)?;
                     (new_blocks, stats, Some(segment_info.summary))
                 } else {
-                    // use by compact.
+                    // Only compact builds replacement segments without corresponding
+                    // entries in base_segments. Treating a missing base segment from
+                    // any other mutation as compact output could silently corrupt its
+                    // segment statistics.
+                    if !matches!(write_segment_ctx.kind, MutationKind::Compact) {
+                        return Err(ErrorCode::Internal(format!(
+                            "{} mutation references missing base segment index {}",
+                            write_segment_ctx.kind, index
+                        )));
+                    }
                     assert!(segment_mutation.deleted_blocks.is_empty());
                     // There are more than 1 blocks, means that the blocks can no longer be compacted.
                     // They can be marked as perfect blocks.
-                    all_perfect = segment_mutation.replaced_blocks.len() > 1;
+                    force_all_blocks_perfect = segment_mutation.replaced_blocks.len() > 1;
                     let (new_blocks, new_hlls) = segment_mutation
                         .replaced_blocks
                         .into_iter()
@@ -575,7 +594,7 @@ impl TableMutationAggregator {
                 };
 
                 let new_segment_info = write_segment_ctx
-                    .write_segment(new_blocks, new_hlls, all_perfect)
+                    .write_segment(new_blocks, new_hlls, force_all_blocks_perfect)
                     .await?;
 
                 Ok(SegmentLite {
@@ -830,14 +849,14 @@ impl WriteSegmentCtx {
         &self,
         blocks: Vec<Arc<BlockMeta>>,
         stats: Option<Vec<u8>>,
-        all_perfect: bool,
+        force_all_blocks_perfect: bool,
     ) -> Result<(String, Statistics)> {
         let location = self
             .location_gen
             .gen_segment_info_location(self.table_meta_timestamps, false);
         let mut new_summary =
             reduce_block_metas(&blocks, self.thresholds, self.default_cluster_key);
-        if all_perfect {
+        if force_all_blocks_perfect {
             // To fix issue #13217.
             if new_summary.block_count > new_summary.perfect_block_count {
                 warn!(

tests/sqllogictests/suites/base/09_fuse_engine/09_0054_issue_20181.test

@@ -0,0 +1,82 @@
+statement ok
+DROP DATABASE IF EXISTS issue_20181
+
+statement ok
+CREATE DATABASE issue_20181
+
+statement ok
+USE issue_20181
+
+statement ok
+SET max_block_size = 6
+
+# Keep block generation deterministic: each statement writes one perfect block
+# and one imperfect tail block, so compaction is triggered by the second write.
+statement ok
+SET max_threads = 1
+
+# REPLACE append after-images must retain their physical imperfect-block count so
+# the second statement reaches the table-level auto-compaction threshold.
+statement ok
+CREATE TABLE replace_target(id UInt64)
+ROW_PER_BLOCK = 10
+AUTO_COMPACTION_IMPERFECT_BLOCKS_THRESHOLD = 2
+
+statement ok
+REPLACE INTO replace_target ON(id)
+SELECT number FROM numbers(10)
+UNION ALL
+SELECT number + 10 FROM numbers(2)
+
+statement ok
+REPLACE INTO replace_target ON(id)
+SELECT number + 12 FROM numbers(10)
+UNION ALL
+SELECT number + 22 FROM numbers(2)
+
+# Each REPLACE writes one perfect block and one imperfect tail block. The second
+# tail reaches the threshold, and auto compaction merges both input segments.
+query IIB
+SELECT count(), sum(row_count), sum(block_count) < 4
+FROM fuse_segment('issue_20181', 'replace_target')
+----
+1 24 1
+
+# MERGE unmatched inserts use the same AppendBlock segment-generation path.
+statement ok
+CREATE TABLE merge_target(id UInt64)
+ROW_PER_BLOCK = 10
+AUTO_COMPACTION_IMPERFECT_BLOCKS_THRESHOLD = 2
+
+query I
+MERGE INTO merge_target AS t
+USING (
+    SELECT number AS id FROM numbers(10)
+    UNION ALL
+    SELECT number + 10 AS id FROM numbers(2)
+) AS s
+ON t.id = s.id
+WHEN NOT MATCHED THEN INSERT (id) VALUES (s.id)
+----
+12
+
+query I
+MERGE INTO merge_target AS t
+USING (
+    SELECT number + 12 AS id FROM numbers(10)
+    UNION ALL
+    SELECT number + 22 AS id FROM numbers(2)
+) AS s
+ON t.id = s.id
+WHEN NOT MATCHED THEN INSERT (id) VALUES (s.id)
+----
+12
+
+query IIB
+SELECT count(), sum(row_count), sum(block_count) < 4
+FROM fuse_segment('issue_20181', 'merge_target')
+----
+1 24 1
+
+statement ok
+DROP DATABASE issue_20181