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

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

@@ -16,8 +16,8 @@ use risingwave_common::constants::hummock::CompactionFilterFlag;
 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_MAX_COMPACTION_BYTES: u64 = 1 * 1024 * 1024 * 1024; // 1GB
+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 = 64;
 
 pub struct CompactionConfigBuilder {
     config: CompactionConfig,
@@ -65,6 +66,7 @@ impl CompactionConfigBuilder {
                 split_by_state_table: false,
                 level0_stop_write_threshold_sub_level_number:
                     DEFAULT_LEVEL0_STOP_WRITE_THRESHOLD_SUB_LEVEL_NUMBER,
+                level0_max_compact_file_number: DEFAULT_MAX_COMPACTION_FILE_COUNT,
             },
         }
     }
@@ -127,4 +129,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

@@ -163,9 +163,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;
         }

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;