feat(compaction): support trivial move multi ssts

proto/hummock.proto

@@ -661,6 +661,7 @@ message RiseCtlUpdateCompactionConfigRequest {
       uint32 split_weight_by_vnode = 20;
       bool disable_auto_group_scheduling = 21;
       uint64 max_overlapping_level_size = 22;
+      uint32 sst_allowed_trivial_move_max_count = 23;
     }
   }
   repeated uint64 compaction_group_ids = 1;
@@ -863,6 +864,9 @@ message CompactionConfig {
   // The limitation of the max size of the overlapping-level for the compaction
   // hummock will reorg the commit-sstables to the multi overlapping-level if the size of the commit-sstables is larger than `max_overlapping_level_size`
   optional uint64 max_overlapping_level_size = 24;
+
+  // The limitation of the max sst count of the trivial move task
+  optional uint32 sst_allowed_trivial_move_max_count = 25;
 }
 
 message TableStats {

src/common/src/config.rs

@@ -2215,6 +2215,7 @@ pub mod default {
         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;
+        const DEFAULT_SST_ALLOWED_TRIVIAL_MOVE_MAX_COUNT: u32 = 64;
 
         use crate::catalog::hummock::CompactionFilterFlag;
 
@@ -2297,6 +2298,10 @@ pub mod default {
         pub fn max_overlapping_level_size() -> u64 {
             256 * MB
         }
+
+        pub fn sst_allowed_trivial_move_max_count() -> u32 {
+            DEFAULT_SST_ALLOWED_TRIVIAL_MOVE_MAX_COUNT
+        }
     }
 
     pub mod object_store_config {
@@ -2676,6 +2681,10 @@ pub struct CompactionConfig {
     pub enable_emergency_picker: bool,
     #[serde(default = "default::compaction_config::max_level")]
     pub max_level: u32,
+    #[serde(default = "default::compaction_config::sst_allowed_trivial_move_min_size")]
+    pub sst_allowed_trivial_move_min_size: u64,
+    #[serde(default = "default::compaction_config::sst_allowed_trivial_move_max_count")]
+    pub sst_allowed_trivial_move_max_count: u32,
 }
 
 /// Note: only applies to meta store backends other than `SQLite`.

src/config/docs.md

@@ -88,6 +88,8 @@ This page is automatically generated by `./risedev generate-example-config`
 | max_level |  | 6 |
 | max_space_reclaim_bytes |  | 536870912 |
 | max_sub_compaction |  | 4 |
+| sst_allowed_trivial_move_max_count |  | 64 |
+| sst_allowed_trivial_move_min_size |  | 4194304 |
 | sub_level_max_compaction_bytes |  | 134217728 |
 | target_file_size_base |  | 33554432 |
 | tombstone_reclaim_ratio |  | 40 |

src/config/example.toml

@@ -81,6 +81,8 @@ level0_max_compact_file_number = 100
 tombstone_reclaim_ratio = 40
 enable_emergency_picker = true
 max_level = 6
+sst_allowed_trivial_move_min_size = 4194304
+sst_allowed_trivial_move_max_count = 64
 
 [meta.developer]
 meta_cached_traces_num = 256

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

@@ -69,6 +69,7 @@ pub fn build_compaction_config_vec(
     sst_allowed_trivial_move_min_size: Option<u64>,
     disable_auto_group_scheduling: Option<bool>,
     max_overlapping_level_size: Option<u64>,
+    sst_allowed_trivial_move_max_count: Option<u32>,
 ) -> Vec<MutableConfig> {
     let mut configs = vec![];
     if let Some(c) = max_bytes_for_level_base {
@@ -131,6 +132,9 @@ pub fn build_compaction_config_vec(
     if let Some(c) = max_overlapping_level_size {
         configs.push(MutableConfig::MaxOverlappingLevelSize(c))
     }
+    if let Some(c) = sst_allowed_trivial_move_max_count {
+        configs.push(MutableConfig::SstAllowedTrivialMoveMaxCount(c))
+    }
 
     configs
 }

src/ctl/src/lib.rs

@@ -196,6 +196,8 @@ enum HummockCommands {
         disable_auto_group_scheduling: Option<bool>,
         #[clap(long)]
         max_overlapping_level_size: Option<u64>,
+        #[clap(long)]
+        sst_allowed_trivial_move_max_count: Option<u32>,
     },
     /// Split given compaction group into two. Moves the given tables to the new group.
     SplitCompactionGroup {
@@ -605,6 +607,7 @@ async fn start_impl(opts: CliOpts, context: &CtlContext) -> Result<()> {
             sst_allowed_trivial_move_min_size,
             disable_auto_group_scheduling,
             max_overlapping_level_size,
+            sst_allowed_trivial_move_max_count,
         }) => {
             cmd_impl::hummock::update_compaction_config(
                 context,
@@ -638,6 +641,7 @@ async fn start_impl(opts: CliOpts, context: &CtlContext) -> Result<()> {
                     sst_allowed_trivial_move_min_size,
                     disable_auto_group_scheduling,
                     max_overlapping_level_size,
+                    sst_allowed_trivial_move_max_count,
                 ),
             )
             .await?

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

@@ -72,6 +72,9 @@ impl CompactionConfigBuilder {
                     compaction_config::disable_auto_group_scheduling(),
                 ),
                 max_overlapping_level_size: Some(compaction_config::max_overlapping_level_size()),
+                sst_allowed_trivial_move_max_count: Some(
+                    compaction_config::sst_allowed_trivial_move_max_count(),
+                ),
             },
         }
     }
@@ -101,6 +104,8 @@ impl CompactionConfigBuilder {
             .level0_max_compact_file_number(opt.level0_max_compact_file_number)
             .tombstone_reclaim_ratio(opt.tombstone_reclaim_ratio)
             .max_level(opt.max_level as u64)
+            .sst_allowed_trivial_move_min_size(Some(opt.sst_allowed_trivial_move_min_size))
+            .sst_allowed_trivial_move_max_count(Some(opt.sst_allowed_trivial_move_max_count))
     }
 
     pub fn build(self) -> CompactionConfig {
@@ -162,4 +167,5 @@ builder_field! {
     level0_overlapping_sub_level_compact_level_count: u32,
     tombstone_reclaim_ratio: u32,
     sst_allowed_trivial_move_min_size: Option<u64>,
+    sst_allowed_trivial_move_max_count: Option<u32>,
 }

src/meta/src/hummock/compaction/mod.rs

@@ -19,11 +19,11 @@ mod overlap_strategy;
 use risingwave_common::catalog::{TableId, TableOption};
 use risingwave_hummock_sdk::compact_task::CompactTask;
 use risingwave_hummock_sdk::level::Levels;
-use risingwave_pb::hummock::compact_task::{self, TaskType};
+use risingwave_pb::hummock::compact_task::{self};
 
 mod picker;
 pub mod selector;
-use std::collections::{BTreeSet, HashMap, HashSet};
+use std::collections::{BTreeSet, HashMap};
 use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 
@@ -32,7 +32,7 @@ use risingwave_hummock_sdk::table_watermark::TableWatermarks;
 use risingwave_hummock_sdk::version::HummockVersionStateTableInfo;
 use risingwave_hummock_sdk::{CompactionGroupId, HummockCompactionTaskId};
 use risingwave_pb::hummock::compaction_config::CompactionMode;
-use risingwave_pb::hummock::{CompactionConfig, LevelType};
+use risingwave_pb::hummock::CompactionConfig;
 pub use selector::{CompactionSelector, CompactionSelectorContext};
 
 use self::selector::{EmergencySelector, LocalSelectorStatistic};
@@ -145,47 +145,6 @@ impl CompactStatus {
         None
     }
 
-    pub fn is_trivial_move_task(task: &CompactTask) -> bool {
-        if task.task_type != TaskType::Dynamic && task.task_type != TaskType::Emergency {
-            return false;
-        }
-
-        if task.input_ssts.len() != 2 || task.input_ssts[0].level_type != LevelType::Nonoverlapping
-        {
-            return false;
-        }
-
-        // it may be a manual compaction task
-        if task.input_ssts[0].level_idx == task.input_ssts[1].level_idx
-            && task.input_ssts[0].level_idx > 0
-        {
-            return false;
-        }
-
-        if task.input_ssts[1].level_idx == task.target_level
-            && task.input_ssts[1].table_infos.is_empty()
-        {
-            return true;
-        }
-
-        false
-    }
-
-    pub fn is_trivial_reclaim(task: &CompactTask) -> bool {
-        // Currently all VnodeWatermark tasks are trivial reclaim.
-        if task.task_type == TaskType::VnodeWatermark {
-            return true;
-        }
-        let exist_table_ids = HashSet::<u32>::from_iter(task.existing_table_ids.clone());
-        task.input_ssts.iter().all(|level| {
-            level.table_infos.iter().all(|sst| {
-                sst.table_ids
-                    .iter()
-                    .all(|table_id| !exist_table_ids.contains(table_id))
-            })
-        })
-    }
-
     pub fn report_compact_task(&mut self, compact_task: &CompactTask) {
         for level in &compact_task.input_ssts {
             self.level_handlers[level.level_idx as usize].remove_task(compact_task.task_id);

src/meta/src/hummock/compaction/overlap_strategy.rs

@@ -26,6 +26,7 @@ pub trait OverlapInfo: Debug {
     fn check_multiple_overlap(&self, others: &[SstableInfo]) -> Range<usize>;
     fn check_multiple_include(&self, others: &[SstableInfo]) -> Range<usize>;
     fn update(&mut self, table: &SstableInfo);
+    fn clear(&mut self);
 }
 
 pub trait OverlapStrategy: Send + Sync {
@@ -134,6 +135,10 @@ impl OverlapInfo for RangeOverlapInfo {
         }
         self.target_range = Some(other.clone());
     }
+
+    fn clear(&mut self) {
+        self.target_range = None;
+    }
 }
 
 #[derive(Default)]

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

@@ -142,6 +142,10 @@ impl LevelCompactionPicker {
             } else {
                 0
             },
+            self.config
+                .sst_allowed_trivial_move_max_count
+                .unwrap_or(compaction_config::sst_allowed_trivial_move_max_count())
+                as usize,
         );
 
         trivial_move_picker.pick_trivial_move_task(

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

@@ -267,51 +267,53 @@ impl IntraCompactionPicker {
                 continue;
             }
 
-            let trivial_move_picker = TrivialMovePicker::new(0, 0, overlap_strategy.clone(), 0);
+            let trivial_move_picker = TrivialMovePicker::new(
+                0,
+                0,
+                overlap_strategy.clone(),
+                0,
+                self.config
+                    .sst_allowed_trivial_move_max_count
+                    .unwrap_or(compaction_config::sst_allowed_trivial_move_max_count())
+                    as usize,
+            );
 
-            let select_sst = trivial_move_picker.pick_trivial_move_sst(
+            if let Some(select_ssts) = trivial_move_picker.pick_multi_trivial_move_ssts(
                 &l0.sub_levels[idx + 1].table_infos,
                 &level.table_infos,
                 level_handlers,
                 stats,
-            );
-
-            // only pick tables for trivial move
-            if select_sst.is_none() {
-                continue;
+            ) {
+                let mut overlap = overlap_strategy.create_overlap_info();
+                select_ssts.iter().for_each(|ssts| overlap.update(ssts));
+
+                assert!(overlap
+                    .check_multiple_overlap(&l0.sub_levels[idx].table_infos)
+                    .is_empty());
+
+                let select_input_size = select_ssts.iter().map(|sst| sst.sst_size).sum();
+                let total_file_count = select_ssts.len() as u64;
+                let input_levels = vec![
+                    InputLevel {
+                        level_idx: 0,
+                        level_type: LevelType::Nonoverlapping,
+                        table_infos: select_ssts,
+                    },
+                    InputLevel {
+                        level_idx: 0,
+                        level_type: LevelType::Nonoverlapping,
+                        table_infos: vec![],
+                    },
+                ];
+                return Some(CompactionInput {
+                    input_levels,
+                    target_level: 0,
+                    target_sub_level_id: level.sub_level_id,
+                    select_input_size,
+                    total_file_count,
+                    ..Default::default()
+                });
             }
-
-            let select_sst = select_sst.unwrap();
-
-            // support trivial move cross multi sub_levels
-            let mut overlap = overlap_strategy.create_overlap_info();
-            overlap.update(&select_sst);
-
-            assert!(overlap
-                .check_multiple_overlap(&l0.sub_levels[idx].table_infos)
-                .is_empty());
-
-            let select_input_size = select_sst.sst_size;
-            let input_levels = vec![
-                InputLevel {
-                    level_idx: 0,
-                    level_type: LevelType::Nonoverlapping,
-                    table_infos: vec![select_sst],
-                },
-                InputLevel {
-                    level_idx: 0,
-                    level_type: LevelType::Nonoverlapping,
-                    table_infos: vec![],
-                },
-            ];
-            return Some(CompactionInput {
-                input_levels,
-                target_level: 0,
-                target_sub_level_id: level.sub_level_id,
-                select_input_size,
-                total_file_count: 1,
-                ..Default::default()
-            });
         }
         None
     }