feat(storage): adjust base compaction limitation

grafana/risingwave-dev-dashboard.dashboard.py

@@ -283,7 +283,7 @@ def section_compaction(outer_panels):
                                 f"histogram_quantile({quantile}, sum(rate({metric('storage_compact_task_size_bucket')}[$__rate_interval])) by (le, group, type))",
                                 f"p{legend}" + " - cg{{group}}@{{type}}",
                             ),
-                            [90, "max"],
+                            [50, 90, "max"],
                         ),
                     ],
                 ),

proto/hummock.proto

@@ -716,6 +716,7 @@ message RiseCtlUpdateCompactionConfigRequest {
       uint32 tombstone_reclaim_ratio = 16;
       CompressionAlgorithm compression_algorithm = 17;
       uint32 max_l0_compact_level_count = 18;
+      uint64 sst_allowed_trivial_move_min_size = 19;
     }
   }
   repeated uint64 compaction_group_ids = 1;
@@ -907,6 +908,9 @@ message CompactionConfig {
 
   // The limitation of the level count of l0 compaction
   optional uint32 max_l0_compact_level_count = 21;
+
+  // The limitation of base level trivial move sst size
+  optional uint64 sst_allowed_trivial_move_min_size = 22;
 }
 
 message TableStats {

src/common/src/config.rs

@@ -1945,17 +1945,19 @@ pub mod default {
     }
 
     pub mod compaction_config {
-        const DEFAULT_MAX_COMPACTION_BYTES: u64 = 2 * 1024 * 1024 * 1024; // 2GB
-        const DEFAULT_MIN_COMPACTION_BYTES: u64 = 128 * 1024 * 1024; // 128MB
-        const DEFAULT_MAX_BYTES_FOR_LEVEL_BASE: u64 = 512 * 1024 * 1024; // 512MB
+        const MB: u64 = 1024 * 1024;
+        const GB: u64 = 1024 * 1024 * 1024;
+        const DEFAULT_MAX_COMPACTION_BYTES: u64 = 2 * GB; // 2GB
+        const DEFAULT_MIN_COMPACTION_BYTES: u64 = 128 * MB; // 128MB
+        const DEFAULT_MAX_BYTES_FOR_LEVEL_BASE: u64 = 512 * MB; // 512MB
 
         // decrease this configure when the generation of checkpoint barrier is not frequent.
         const DEFAULT_TIER_COMPACT_TRIGGER_NUMBER: u64 = 12;
-        const DEFAULT_TARGET_FILE_SIZE_BASE: u64 = 32 * 1024 * 1024;
+        const DEFAULT_TARGET_FILE_SIZE_BASE: u64 = 32 * MB;
         // 32MB
         const DEFAULT_MAX_SUB_COMPACTION: u32 = 4;
         const DEFAULT_LEVEL_MULTIPLIER: u64 = 5;
-        const DEFAULT_MAX_SPACE_RECLAIM_BYTES: u64 = 512 * 1024 * 1024; // 512MB;
+        const DEFAULT_MAX_SPACE_RECLAIM_BYTES: u64 = 512 * MB; // 512MB;
         const DEFAULT_LEVEL0_STOP_WRITE_THRESHOLD_SUB_LEVEL_NUMBER: u64 = 300;
         const DEFAULT_MAX_COMPACTION_FILE_COUNT: u64 = 100;
         const DEFAULT_MIN_SUB_LEVEL_COMPACT_LEVEL_COUNT: u32 = 3;
@@ -1964,6 +1966,7 @@ pub mod default {
         const DEFAULT_EMERGENCY_PICKER: bool = true;
         const DEFAULT_MAX_LEVEL: u32 = 6;
         const DEFAULT_MAX_L0_COMPACT_LEVEL_COUNT: u32 = 42;
+        const DEFAULT_SST_ALLOWED_TRIVIAL_MOVE_MIN_SIZE: u64 = 4 * MB;
 
         use crate::catalog::hummock::CompactionFilterFlag;
 
@@ -2034,6 +2037,10 @@ pub mod default {
         pub fn max_l0_compact_level_count() -> u32 {
             DEFAULT_MAX_L0_COMPACT_LEVEL_COUNT
         }
+
+        pub fn sst_allowed_trivial_move_min_size() -> u64 {
+            DEFAULT_SST_ALLOWED_TRIVIAL_MOVE_MIN_SIZE
+        }
     }
 
     pub mod object_store_config {

src/ctl/src/cmd_impl/hummock/compaction_group.rs

@@ -66,6 +66,7 @@ pub fn build_compaction_config_vec(
     tombstone_reclaim_ratio: Option<u32>,
     compress_algorithm: Option<CompressionAlgorithm>,
     max_l0_compact_level: Option<u32>,
+    sst_allowed_trivial_move_min_size: Option<u64>,
 ) -> Vec<MutableConfig> {
     let mut configs = vec![];
     if let Some(c) = max_bytes_for_level_base {
@@ -119,6 +120,9 @@ pub fn build_compaction_config_vec(
     if let Some(c) = max_l0_compact_level {
         configs.push(MutableConfig::MaxL0CompactLevelCount(c))
     }
+    if let Some(c) = sst_allowed_trivial_move_min_size {
+        configs.push(MutableConfig::SstAllowedTrivialMoveMinSize(c))
+    }
 
     configs
 }

src/ctl/src/lib.rs

@@ -267,6 +267,8 @@ enum HummockCommands {
         compression_algorithm: Option<String>,
         #[clap(long)]
         max_l0_compact_level: Option<u32>,
+        #[clap(long)]
+        sst_allowed_trivial_move_min_size: Option<u64>,
     },
     /// Split given compaction group into two. Moves the given tables to the new group.
     SplitCompactionGroup {
@@ -667,6 +669,7 @@ async fn start_impl(opts: CliOpts, context: &CtlContext) -> Result<()> {
             compression_level,
             compression_algorithm,
             max_l0_compact_level,
+            sst_allowed_trivial_move_min_size,
         }) => {
             cmd_impl::hummock::update_compaction_config(
                 context,
@@ -697,6 +700,7 @@ async fn start_impl(opts: CliOpts, context: &CtlContext) -> Result<()> {
                         None
                     },
                     max_l0_compact_level,
+                    sst_allowed_trivial_move_min_size,
                 ),
             )
             .await?

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

@@ -65,6 +65,9 @@ impl CompactionConfigBuilder {
                 tombstone_reclaim_ratio: compaction_config::tombstone_reclaim_ratio(),
                 enable_emergency_picker: compaction_config::enable_emergency_picker(),
                 max_l0_compact_level_count: Some(compaction_config::max_l0_compact_level_count()),
+                sst_allowed_trivial_move_min_size: Some(
+                    compaction_config::sst_allowed_trivial_move_min_size(),
+                ),
             },
         }
     }
@@ -154,4 +157,5 @@ builder_field! {
     level0_sub_level_compact_level_count: u32,
     level0_overlapping_sub_level_compact_level_count: u32,
     tombstone_reclaim_ratio: u32,
+    sst_allowed_trivial_move_min_size: Option<u64>,
 }

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

@@ -132,8 +132,17 @@ impl LevelCompactionPicker {
         }
 
         let overlap_strategy = create_overlap_strategy(self.config.compaction_mode());
-        let trivial_move_picker =
-            TrivialMovePicker::new(0, self.target_level, overlap_strategy.clone());
+        let trivial_move_picker = TrivialMovePicker::new(
+            0,
+            self.target_level,
+            overlap_strategy.clone(),
+            if self.compaction_task_validator.is_enable() {
+                // tips: Older versions of the compaction group will be upgraded without this configuration, we leave it with its default behaviour and enable it manually when needed.
+                self.config.sst_allowed_trivial_move_min_size.unwrap_or(0)
+            } else {
+                0
+            },
+        );
 
         trivial_move_picker.pick_trivial_move_task(
             &l0.sub_levels[0].table_infos,

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

@@ -78,6 +78,10 @@ impl CompactionTaskValidator {
             true
         }
     }
+
+    pub fn is_enable(&self) -> bool {
+        !self.validation_rules.is_empty()
+    }
 }
 
 pub trait CompactionTaskValidationRule {

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

@@ -267,7 +267,7 @@ impl IntraCompactionPicker {
                 continue;
             }
 
-            let trivial_move_picker = TrivialMovePicker::new(0, 0, overlap_strategy.clone());
+            let trivial_move_picker = TrivialMovePicker::new(0, 0, overlap_strategy.clone(), 0);
 
             let select_sst = trivial_move_picker.pick_trivial_move_sst(
                 &l0.sub_levels[idx + 1].table_infos,
@@ -591,6 +591,7 @@ pub mod tests {
             let config = Arc::new(
                 CompactionConfigBuilder::new()
                     .level0_sub_level_compact_level_count(1)
+                    .sub_level_max_compaction_bytes(300)
                     .build(),
             );
             let mut picker = IntraCompactionPicker::for_test(
@@ -614,7 +615,7 @@ pub mod tests {
             assert_eq!(3, ret.input_levels[1].table_infos[0].sst_id);
             assert_eq!(1, ret.input_levels[2].table_infos[0].sst_id);
 
-            // will pick sst [2, 6, 5]
+            // will pick sst [2, 6]
             let ret2 = picker
                 .pick_compaction(&levels, &levels_handler, &mut local_stats)
                 .unwrap();
@@ -624,12 +625,11 @@ pub mod tests {
                     .iter()
                     .map(|i| i.table_infos.len())
                     .sum::<usize>(),
-                3
+                2
             );
 
-            assert_eq!(5, ret2.input_levels[0].table_infos[0].sst_id);
-            assert_eq!(6, ret2.input_levels[1].table_infos[0].sst_id);
-            assert_eq!(2, ret2.input_levels[2].table_infos[0].sst_id);
+            assert_eq!(6, ret2.input_levels[0].table_infos[0].sst_id);
+            assert_eq!(2, ret2.input_levels[1].table_infos[0].sst_id);
         }
 
         {
@@ -662,6 +662,7 @@ pub mod tests {
             let config = Arc::new(
                 CompactionConfigBuilder::new()
                     .level0_sub_level_compact_level_count(1)
+                    .sub_level_max_compaction_bytes(300)
                     .build(),
             );
             let mut picker = IntraCompactionPicker::for_test(

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

@@ -171,11 +171,12 @@ impl CompactionPicker for MinOverlappingPicker {
     }
 }
 
-#[derive(Default)]
+#[derive(Default, Debug)]
 pub struct SubLevelSstables {
     pub total_file_size: u64,
     pub total_file_count: usize,
     pub sstable_infos: Vec<Vec<SstableInfo>>,
+    pub expected: bool,
 }
 
 pub struct NonOverlapSubLevelPicker {
@@ -404,13 +405,7 @@ impl NonOverlapSubLevelPicker {
                 continue;
             }
 
-            let ret = self.pick_sub_level(l0, level_handler, sst);
-            if ret.sstable_infos.len() < self.min_expected_level_count
-                && ret.total_file_size < self.min_compaction_bytes
-            {
-                continue;
-            }
-            scores.push(ret);
+            scores.push(self.pick_sub_level(l0, level_handler, sst));
         }
 
         if scores.is_empty() {
@@ -420,10 +415,15 @@ impl NonOverlapSubLevelPicker {
         let mut expected = Vec::with_capacity(scores.len());
         let mut unexpected = vec![];
 
-        for selected_task in scores {
-            if selected_task.sstable_infos.len() > self.max_expected_level_count {
+        for mut selected_task in scores {
+            if selected_task.sstable_infos.len() > self.max_expected_level_count
+                || selected_task.sstable_infos.len() < self.min_expected_level_count
+                || selected_task.total_file_size < self.min_compaction_bytes
+            {
+                selected_task.expected = false;
                 unexpected.push(selected_task);
             } else {
+                selected_task.expected = true;
                 expected.push(selected_task);
             }
         }
@@ -439,14 +439,33 @@ impl NonOverlapSubLevelPicker {
                 .then_with(|| a.total_file_size.cmp(&b.total_file_size))
         });
 
-        // For unexpected task the number of levels is as small as possible
+        // For unexpected tasks, We devised a separate algorithm to evaluate the priority of a task, based on the limit passed in,
+        // we set tasks close to the limit to be high priority, here have three attributes:
+        // 1. The number of levels selected is close to the limit
+        // 2. The number of files selected is close to the limit
+        // 3. The size of the selected file is close to the limit
         unexpected.sort_by(|a, b| {
-            a.sstable_infos
-                .len()
-                .cmp(&b.sstable_infos.len())
-                .then_with(|| a.total_file_count.cmp(&b.total_file_count))
-                .then_with(|| a.total_file_size.cmp(&b.total_file_size))
+            let a_select_count_offset =
+                (a.sstable_infos.len() as i64 - self.max_expected_level_count as i64).abs();
+            let b_select_count_offset =
+                (b.sstable_infos.len() as i64 - self.max_expected_level_count as i64).abs();
+
+            let a_file_count_offset =
+                (a.total_file_count as i64 - self.max_file_count as i64).abs();
+            let b_file_count_offset =
+                (b.total_file_count as i64 - self.max_file_count as i64).abs();
+
+            let a_file_size_offset =
+                (a.total_file_size as i64 - self.max_compaction_bytes as i64).abs();
+            let b_file_size_offset =
+                (b.total_file_size as i64 - self.max_compaction_bytes as i64).abs();
+
+            a_select_count_offset
+                .cmp(&b_select_count_offset)
+                .then_with(|| a_file_count_offset.cmp(&b_file_count_offset))
+                .then_with(|| a_file_size_offset.cmp(&b_file_size_offset))
         });
+
         expected.extend(unexpected);
 
         expected
@@ -850,7 +869,7 @@ pub mod tests {
             // limit max bytes
             let max_compaction_bytes = 100;
             let picker = NonOverlapSubLevelPicker::new(
-                0,
+                60,
                 max_compaction_bytes,
                 1,
                 10000,
@@ -860,6 +879,14 @@ pub mod tests {
             );
             let ret = picker.pick_l0_multi_non_overlap_level(&levels, &levels_handlers[0]);
             assert_eq!(6, ret.len());
+
+            for plan in &ret {
+                if plan.total_file_size >= max_compaction_bytes {
+                    assert!(plan.expected);
+                } else {
+                    assert!(!plan.expected);
+                }
+            }
         }
 
         {
@@ -890,23 +917,23 @@ pub mod tests {
             // limit min_depth
             let min_depth = 3;
             let picker = NonOverlapSubLevelPicker::new(
-                1000,
+                10,
                 10000,
                 min_depth,
-                10000,
+                100,
                 Arc::new(RangeOverlapStrategy::default()),
                 true,
                 compaction_config::max_l0_compact_level_count() as usize,
             );
             let ret = picker.pick_l0_multi_non_overlap_level(&levels, &levels_handlers[0]);
-            assert_eq!(3, ret.len());
+            assert_eq!(6, ret.len());
 
             for plan in ret {
-                let mut sst_id_set = BTreeSet::default();
-                for sst in &plan.sstable_infos {
-                    sst_id_set.insert(sst[0].sst_id);
+                if plan.sstable_infos.len() >= min_depth {
+                    assert!(plan.expected);
+                } else {
+                    assert!(!plan.expected);
                 }
-                assert!(plan.sstable_infos.len() >= min_depth);
             }
         }
     }
@@ -1091,5 +1118,29 @@ pub mod tests {
                 assert_eq!(2, plan.sstable_infos[2][0].sst_id);
             }
         }
+
+        {
+            // limit min_compacaion_bytes
+            let max_expected_level_count = 100;
+            let picker = NonOverlapSubLevelPicker::for_test(
+                1000,
+                10000,
+                1,
+                100,
+                Arc::new(RangeOverlapStrategy::default()),
+                true,
+                max_expected_level_count,
+            );
+            let ret = picker.pick_l0_multi_non_overlap_level(&levels, &levels_handlers[0]);
+            {
+                let plan = &ret[0];
+
+                assert_eq!(0, plan.sstable_infos[0][0].sst_id);
+                assert_eq!(1, plan.sstable_infos[1][0].sst_id);
+                assert_eq!(2, plan.sstable_infos[2][0].sst_id);
+                assert_eq!(3, plan.sstable_infos[3][0].sst_id);
+                assert!(!plan.expected);
+            }
+        }
     }
 }