feat(compaction): support trivial move multi ssts (#20284)

proto/hummock.proto

@@ -669,6 +669,8 @@ message RiseCtlUpdateCompactionConfigRequest {
       uint32 level0_stop_write_threshold_max_sst_count = 27;
       // The limitation of the max sst size of the level0 to trigger the write stop
       uint64 level0_stop_write_threshold_max_size = 28;
+      // The limitation of the max sst count of the trivial move task
+      uint32 sst_allowed_trivial_move_max_count = 29;
     }
   }
   repeated uint64 compaction_group_ids = 1;
@@ -880,8 +882,12 @@ message CompactionConfig {
 
   // The limitation of the max sst count of the level0 to trigger the write stop
   optional uint32 level0_stop_write_threshold_max_sst_count = 27;
+
   // The limitation of the max sst size of the level0 to trigger the write stop
   optional uint64 level0_stop_write_threshold_max_size = 28;
+
+  // The limitation of the max sst count of the trivial move task
+  optional uint32 sst_allowed_trivial_move_max_count = 29;
 }
 
 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;
         const DEFAULT_EMERGENCY_LEVEL0_SST_FILE_COUNT: u32 = 2000; // > 50G / 32M = 1600
         const DEFAULT_EMERGENCY_LEVEL0_SUB_LEVEL_PARTITION: u32 = 256;
         const DEFAULT_LEVEL0_STOP_WRITE_THRESHOLD_MAX_SST_COUNT: u32 = 10000; // 10000 * 32M = 320G
@@ -2302,6 +2303,10 @@ pub mod default {
             256 * MB
         }
 
+        pub fn sst_allowed_trivial_move_max_count() -> u32 {
+            DEFAULT_SST_ALLOWED_TRIVIAL_MOVE_MAX_COUNT
+        }
+
         pub fn emergency_level0_sst_file_count() -> u32 {
             DEFAULT_EMERGENCY_LEVEL0_SST_FILE_COUNT
         }
@@ -2696,10 +2701,12 @@ pub struct CompactionConfig {
     pub enable_emergency_picker: bool,
     #[serde(default = "default::compaction_config::max_level")]
     pub max_level: u32,
-    #[serde(default = "default::compaction_config::max_l0_compact_level_count")]
-    pub max_l0_compact_level_count: 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,
+    #[serde(default = "default::compaction_config::max_l0_compact_level_count")]
+    pub max_l0_compact_level_count: u32,
     #[serde(default = "default::compaction_config::disable_auto_group_scheduling")]
     pub disable_auto_group_scheduling: bool,
     #[serde(default = "default::compaction_config::max_overlapping_level_size")]

src/config/docs.md

@@ -95,6 +95,7 @@ This page is automatically generated by `./risedev generate-example-config`
 | max_overlapping_level_size |  | 268435456 |
 | 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 |

src/config/example.toml

@@ -81,8 +81,9 @@ level0_max_compact_file_number = 100
 tombstone_reclaim_ratio = 40
 enable_emergency_picker = true
 max_level = 6
-max_l0_compact_level_count = 42
 sst_allowed_trivial_move_min_size = 4194304
+sst_allowed_trivial_move_max_count = 64
+max_l0_compact_level_count = 42
 disable_auto_group_scheduling = false
 max_overlapping_level_size = 268435456
 emergency_level0_sst_file_count = 2000

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>,
     emergency_level0_sst_file_count: Option<u32>,
     emergency_level0_sub_level_partition: Option<u32>,
     level0_stop_write_threshold_max_sst_count: Option<u32>,
@@ -135,6 +136,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))
+    }
     if let Some(c) = emergency_level0_sst_file_count {
         configs.push(MutableConfig::EmergencyLevel0SstFileCount(c))
     }

src/ctl/src/lib.rs

@@ -197,6 +197,8 @@ enum HummockCommands {
         #[clap(long)]
         max_overlapping_level_size: Option<u64>,
         #[clap(long)]
+        sst_allowed_trivial_move_max_count: Option<u32>,
+        #[clap(long)]
         emergency_level0_sst_file_count: Option<u32>,
         #[clap(long)]
         emergency_level0_sub_level_partition: Option<u32>,
@@ -613,6 +615,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,
             emergency_level0_sst_file_count,
             emergency_level0_sub_level_partition,
             level0_stop_write_threshold_max_sst_count,
@@ -650,6 +653,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,
                     emergency_level0_sst_file_count,
                     emergency_level0_sub_level_partition,
                     level0_stop_write_threshold_max_sst_count,

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(),
+                ),
                 emergency_level0_sst_file_count: Some(
                     compaction_config::emergency_level0_sst_file_count(),
                 ),
@@ -114,6 +117,7 @@ impl CompactionConfigBuilder {
             .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))
             .max_overlapping_level_size(Some(opt.max_overlapping_level_size))
             .emergency_level0_sst_file_count(Some(opt.emergency_level0_sst_file_count))
             .emergency_level0_sub_level_partition(Some(opt.emergency_level0_sub_level_partition))
@@ -182,6 +186,7 @@ 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>,
     disable_auto_group_scheduling: Option<bool>,
     max_overlapping_level_size: Option<u64>,
     emergency_level0_sst_file_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};
@@ -146,47 +146,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/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
     }