fix(meta): limit file count and l0 compact score (risingwavelabs#8563)

Signed-off-by: Little-Wallace <bupt2013211450@gmail.com>

proto/hummock.proto

@@ -605,6 +605,7 @@ message CompactionConfig {
   bool split_by_state_table = 14;
   // soft limit for max number of sub level number
   uint64 level0_stop_write_threshold_sub_level_number = 15;
+  uint64 level0_max_compact_file_number = 16;
 }
 
 message TableStats {

src/meta/src/hummock/compaction/compaction_config.rs

@@ -17,7 +17,7 @@ use risingwave_pb::hummock::compaction_config::CompactionMode;
 use risingwave_pb::hummock::CompactionConfig;
 
 const DEFAULT_MAX_COMPACTION_BYTES: u64 = 2 * 1024 * 1024 * 1024; // 2GB
-const DEFAULT_MIN_COMPACTION_BYTES: u64 = 256 * 1024 * 1024; // 256MB
+const DEFAULT_MIN_COMPACTION_BYTES: u64 = 128 * 1024 * 1024; // 128MB
 const DEFAULT_MAX_BYTES_FOR_LEVEL_BASE: u64 = 512 * 1024 * 1024; // 512MB
 
 // decrease this configure when the generation of checkpoint barrier is not frequent.
@@ -28,6 +28,7 @@ const MAX_LEVEL: u64 = 6;
 const DEFAULT_LEVEL_MULTIPLIER: u64 = 5;
 const DEFAULT_MAX_SPACE_RECLAIM_BYTES: u64 = 512 * 1024 * 1024; // 512MB;
 const DEFAULT_LEVEL0_STOP_WRITE_THRESHOLD_SUB_LEVEL_NUMBER: u64 = u32::MAX as u64;
+const DEFAULT_MAX_COMPACTION_FILE_COUNT: u64 = 96;
 
 pub struct CompactionConfigBuilder {
     config: CompactionConfig,
@@ -65,6 +66,11 @@ impl CompactionConfigBuilder {
                 split_by_state_table: false,
                 level0_stop_write_threshold_sub_level_number:
                     DEFAULT_LEVEL0_STOP_WRITE_THRESHOLD_SUB_LEVEL_NUMBER,
+                // This configure variable shall be larger than level0_tier_compact_file_number, and
+                // it shall meet the following condition:
+                //    level0_max_compact_file_number * target_file_size_base >
+                // max_bytes_for_level_base
+                level0_max_compact_file_number: DEFAULT_MAX_COMPACTION_FILE_COUNT,
             },
         }
     }
@@ -127,4 +133,5 @@ builder_field! {
     max_sub_compaction: u32,
     max_space_reclaim_bytes: u64,
     level0_stop_write_threshold_sub_level_number: u64,
+    level0_max_compact_file_number: u64,
 }

src/meta/src/hummock/compaction/level_selector.rs

@@ -20,6 +20,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use risingwave_common::catalog::TableOption;
+use risingwave_hummock_sdk::compaction_group::hummock_version_ext::HummockLevelsExt;
 use risingwave_hummock_sdk::HummockCompactionTaskId;
 use risingwave_pb::hummock::hummock_version::Levels;
 use risingwave_pb::hummock::{compact_task, CompactionConfig};
@@ -200,13 +201,15 @@ impl DynamicLevelSelectorCore {
 
         let total_size = levels.l0.as_ref().unwrap().total_file_size
             - handlers[0].get_pending_output_file_size(ctx.base_level as u32);
+        let base_level_size = levels.get_level(ctx.base_level).total_file_size;
         if idle_file_count > 0 {
             // trigger intra-l0 compaction at first when the number of files is too large.
             let l0_score =
                 idle_file_count as u64 * SCORE_BASE / self.config.level0_tier_compact_file_number;
             ctx.score_levels
                 .push((std::cmp::min(l0_score, max_l0_score), 0, 0));
-            let score = total_size * SCORE_BASE / self.config.max_bytes_for_level_base;
+            let score = total_size * SCORE_BASE
+                / std::cmp::max(self.config.max_bytes_for_level_base, base_level_size);
             ctx.score_levels.push((score, 0, ctx.base_level));
         }
 

src/meta/src/hummock/compaction/picker/base_level_compaction_picker.rs

@@ -109,6 +109,7 @@ impl LevelCompactionPicker {
     ) -> Option<CompactionInput> {
         let mut input_levels = vec![];
         let mut l0_total_file_size = 0;
+        let mut l0_total_file_count = 0;
         for level in &l0.sub_levels {
             // This break is optional. We can include overlapping sub-level actually.
             if level.level_type() != LevelType::Nonoverlapping {
@@ -118,6 +119,10 @@ impl LevelCompactionPicker {
                 break;
             }
 
+            if l0_total_file_count > self.config.level0_max_compact_file_number {
+                break;
+            }
+
             let mut pending_compact = false;
             let mut cur_level_size = 0;
             let mut select_level = InputLevel {
@@ -144,6 +149,7 @@ impl LevelCompactionPicker {
                 continue;
             }
 
+            l0_total_file_count += select_level.table_infos.len() as u64;
             l0_total_file_size += cur_level_size;
             input_levels.push(select_level);
         }
@@ -163,9 +169,7 @@ impl LevelCompactionPicker {
             target_level_size += sst.file_size;
         }
 
-        if target_level_size > l0_total_file_size
-            && l0_total_file_size < self.config.max_compaction_bytes
-        {
+        if target_level_size > l0_total_file_size {
             stats.skip_by_write_amp_limit += 1;
             return None;
         }
@@ -440,7 +444,7 @@ pub mod tests {
             .is_none());
     }
 
-    // compact the whole level and upper sub-level when the write-amplification is more than 1.5.
+    // compact the whole level and upper sub-level when the write-amplification is more than 1.0.
     #[test]
     fn test_compact_whole_level_write_amplification_limit() {
         let config = CompactionConfigBuilder::new()
@@ -497,10 +501,8 @@ pub mod tests {
         sub_level.table_infos[0].file_size += 1000 - sub_level.total_file_size;
         sub_level.total_file_size = 1000;
         levels.l0.as_mut().unwrap().sub_levels[1].total_file_size = 1000;
-        let ret = picker
-            .pick_compaction(&levels, &levels_handler, &mut local_stats)
-            .unwrap();
-        assert_eq!(ret.input_levels.len(), 2);
+        let ret = picker.pick_compaction(&levels, &levels_handler, &mut local_stats);
+        assert!(ret.is_none());
     }
 
     #[test]

src/meta/src/hummock/compaction/picker/tier_compaction_picker.rs

@@ -69,12 +69,15 @@ impl TierCompactionPicker {
             );
             let mut compaction_bytes = level.total_file_size;
             let mut max_level_size = level.total_file_size;
+            let mut compact_file_count = level.table_infos.len() as u64;
 
             for other in &l0.sub_levels[idx + 1..] {
                 if compaction_bytes > max_compaction_bytes {
                     break;
                 }
-
+                if compact_file_count > self.config.level0_max_compact_file_number {
+                    break;
+                }
                 if other.level_type != non_overlapping_type
                     || other.total_file_size > self.config.sub_level_max_compaction_bytes
                 {
@@ -85,6 +88,7 @@ impl TierCompactionPicker {
                 }
 
                 compaction_bytes += other.total_file_size;
+                compact_file_count += other.table_infos.len() as u64;
                 max_level_size = std::cmp::max(max_level_size, other.total_file_size);
                 select_level_inputs.push(InputLevel {
                     level_idx: 0,
@@ -107,6 +111,7 @@ impl TierCompactionPicker {
             if level.level_type == non_overlapping_type
                 && is_write_amp_large
                 && select_level_inputs.len() < self.config.level0_tier_compact_file_number as usize
+                && compact_file_count < self.config.level0_max_compact_file_number
             {
                 stats.skip_by_write_amp_limit += 1;
                 continue;
@@ -303,7 +308,7 @@ impl TierCompactionPicker {
             );
 
             let mut compaction_bytes = level.total_file_size;
-            let mut compact_file_count = level.table_infos.len();
+            let mut compact_file_count = level.table_infos.len() as u64;
             let mut waiting_enough_files = true;
 
             for other in &l0.sub_levels[idx + 1..] {
@@ -312,6 +317,10 @@ impl TierCompactionPicker {
                     break;
                 }
 
+                if compact_file_count > self.config.level0_max_compact_file_number {
+                    break;
+                }
+
                 if other.level_type != overlapping_type {
                     waiting_enough_files = false;
                     break;
@@ -322,15 +331,15 @@ impl TierCompactionPicker {
                 }
 
                 compaction_bytes += other.total_file_size;
-                compact_file_count += other.table_infos.len();
+                compact_file_count += other.table_infos.len() as u64;
                 select_level_inputs.push(InputLevel {
                     level_idx: 0,
                     level_type: other.level_type,
                     table_infos: other.table_infos.clone(),
                 });
             }
 
-            if compact_file_count < self.config.level0_tier_compact_file_number as usize
+            if compact_file_count < self.config.level0_tier_compact_file_number
                 && waiting_enough_files
             {
                 stats.skip_by_count_limit += 1;

src/meta/src/hummock/test_utils.rs

@@ -338,6 +338,7 @@ pub async fn setup_compute_env(
 ) {
     let config = CompactionConfigBuilder::new()
         .level0_tier_compact_file_number(1)
+        .level0_max_compact_file_number(130)
         .build();
     setup_compute_env_with_config(port, config).await
 }