feat(meta): initiate static compaction groups

proto/hummock.proto

@@ -126,7 +126,7 @@ message CompactTask {
   repeated KeyRange splits = 2;
   // low watermark in 'ts-aware compaction'
   uint64 watermark = 3;
-  // compacion output, which will be added to [`target_level`] of LSM after compaction
+  // compaction output, which will be added to [`target_level`] of LSM after compaction
   repeated SstableInfo sorted_output_ssts = 4;
   // task id assigned by hummock storage service
   uint64 task_id = 5;
@@ -135,18 +135,11 @@ message CompactTask {
   bool is_target_ultimate_and_leveling = 7;
   CompactMetrics metrics = 8;
   bool task_status = 9;
-  repeated CompactionGroup.PrefixPair prefix_pairs = 10;
   // Hash mapping from virtual node to parallel unit. Since one compactor might deal with SSTs
   // with data for more than one relational state tables, here a vector is required.
   repeated common.ParallelUnitMapping vnode_mappings = 11;
-}
-
-message CompactionGroup {
-  message PrefixPair {
-    // key value with `prefix` belongs to compaction group `group_id`
-    uint64 group_id = 1;
-    bytes prefix = 2;
-  }
+  // compaction group the task belongs to
+  uint64 compaction_group_id = 12;
 }
 
 message LevelHandler {
@@ -160,7 +153,15 @@ message LevelHandler {
 }
 
 message CompactStatus {
-  repeated LevelHandler level_handlers = 1;
+  uint64 compaction_group_id = 1;
+  repeated LevelHandler level_handlers = 2;
+  CompactionConfig compaction_config = 3;
+}
+
+message CompactionGroup {
+  uint64 id = 1;
+  repeated bytes member_prefixes = 2;
+  CompactionConfig compaction_config = 3;
 }
 
 message CompactTaskAssignment {
@@ -240,3 +241,18 @@ service HummockManagerService {
 }
 
 service CompactorService {}
+
+message CompactionConfig {
+  enum CompactionMode {
+    RANGE = 0;
+    CONSISTENT_HASH = 1;
+  }
+  uint64 max_bytes_for_level_base = 1;
+  uint64 max_level = 2;
+  uint64 max_bytes_for_level_multiplier = 3;
+  uint64 max_compaction_bytes = 4;
+  uint64 min_compaction_bytes = 5;
+  uint64 level0_tigger_file_numer = 6;
+  uint64 level0_tier_compact_file_number = 7;
+  CompactionMode compaction_mode = 8;
+}

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

@@ -0,0 +1,87 @@
+// Copyright 2022 Singularity Data
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use risingwave_pb::hummock::compaction_config::CompactionMode;
+use risingwave_pb::hummock::CompactionConfig;
+
+const DEFAULT_MAX_COMPACTION_BYTES: u64 = 4 * 1024 * 1024 * 1024; // 4GB
+const DEFAULT_MIN_COMPACTION_BYTES: u64 = 128 * 1024 * 1024; // 128MB
+const DEFAULT_MAX_BYTES_FOR_LEVEL_BASE: u64 = 1024 * 1024 * 1024; // 1GB
+
+// decrease this configure when the generation of checkpoint barrier is not frequent.
+const DEFAULT_TIER_COMPACT_TRIGGER_NUMBER: u64 = 16;
+const MAX_LEVEL: u64 = 6;
+
+pub struct CompactionConfigBuilder {
+    config: CompactionConfig,
+}
+
+impl CompactionConfigBuilder {
+    pub fn new() -> Self {
+        Self {
+            config: CompactionConfig {
+                max_bytes_for_level_base: DEFAULT_MAX_BYTES_FOR_LEVEL_BASE,
+                max_bytes_for_level_multiplier: 10,
+                max_level: MAX_LEVEL,
+                max_compaction_bytes: DEFAULT_MAX_COMPACTION_BYTES,
+                min_compaction_bytes: DEFAULT_MIN_COMPACTION_BYTES,
+                level0_tigger_file_numer: DEFAULT_TIER_COMPACT_TRIGGER_NUMBER * 2,
+                level0_tier_compact_file_number: DEFAULT_TIER_COMPACT_TRIGGER_NUMBER,
+                compaction_mode: CompactionMode::ConsistentHash as i32,
+            },
+        }
+    }
+
+    pub fn new_with(config: CompactionConfig) -> Self {
+        Self { config }
+    }
+
+    pub fn build(self) -> CompactionConfig {
+        self.config
+    }
+}
+
+impl Default for CompactionConfigBuilder {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+macro_rules! builder_field {
+    ($( $name:ident: $type:ty ),* ,) => {
+        impl CompactionConfigBuilder {
+            $(
+                pub fn $name(&self, v:$type) -> Self {
+                    Self {
+                        config: CompactionConfig {
+                            $name: v,
+                            ..self.config
+                        },
+                    }
+                }
+            )*
+        }
+    }
+}
+
+builder_field! {
+    max_bytes_for_level_base: u64,
+    max_bytes_for_level_multiplier: u64,
+    max_level: u64,
+    max_compaction_bytes: u64,
+    min_compaction_bytes: u64,
+    level0_tigger_file_numer: u64,
+    level0_tier_compact_file_number: u64,
+    compaction_mode: i32,
+}

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

@@ -20,17 +20,15 @@
 use std::sync::Arc;
 
 use risingwave_hummock_sdk::HummockCompactionTaskId;
-use risingwave_pb::hummock::Level;
+use risingwave_pb::hummock::{CompactionConfig, Level};
 
+use crate::hummock::compaction::compaction_config::CompactionConfigBuilder;
 use crate::hummock::compaction::compaction_picker::{CompactionPicker, MinOverlappingPicker};
-use crate::hummock::compaction::overlap_strategy::{
-    HashStrategy, OverlapStrategy, RangeOverlapStrategy,
-};
+use crate::hummock::compaction::overlap_strategy::OverlapStrategy;
 use crate::hummock::compaction::tier_compaction_picker::{
     LevelCompactionPicker, TierCompactionPicker,
 };
-use crate::hummock::compaction::CompactionMode::{ConsistentHashMode, RangeMode};
-use crate::hummock::compaction::{CompactionConfig, SearchResult};
+use crate::hummock::compaction::{create_overlap_strategy, SearchResult};
 use crate::hummock::level_handler::LevelHandler;
 
 const SCORE_BASE: u64 = 100;
@@ -68,11 +66,8 @@ pub struct DynamicLevelSelector {
 
 impl Default for DynamicLevelSelector {
     fn default() -> Self {
-        let config = Arc::new(CompactionConfig::default());
-        let overlap_strategy = match &config.compaction_mode {
-            RangeMode => Arc::new(RangeOverlapStrategy::default()) as Arc<dyn OverlapStrategy>,
-            ConsistentHashMode => Arc::new(HashStrategy::default()),
-        };
+        let config = Arc::new(CompactionConfigBuilder::new().build());
+        let overlap_strategy = create_overlap_strategy(config.compaction_mode());
         DynamicLevelSelector::new(config, overlap_strategy)
     }
 }
@@ -137,15 +132,15 @@ impl DynamicLevelSelector {
 
         if max_level_size == 0 {
             // Use the bottommost level.
-            ctx.base_level = self.config.max_level;
+            ctx.base_level = self.config.max_level as usize;
             return ctx;
         }
 
         let base_bytes_max = std::cmp::max(self.config.max_bytes_for_level_base, l0_size);
         let base_bytes_min = base_bytes_max / self.config.max_bytes_for_level_multiplier;
 
         let mut cur_level_size = max_level_size;
-        for _ in first_non_empty_level..self.config.max_level {
+        for _ in first_non_empty_level..self.config.max_level as usize {
             cur_level_size /= self.config.max_bytes_for_level_multiplier;
         }
 
@@ -166,7 +161,7 @@ impl DynamicLevelSelector {
 
         let level_multiplier = self.config.max_bytes_for_level_multiplier as f64;
         let mut level_size = base_level_size;
-        for i in ctx.base_level..=self.config.max_level {
+        for i in ctx.base_level..=self.config.max_level as usize {
             // Don't set any level below base_bytes_max. Otherwise, the LSM can
             // assume an hourglass shape where L1+ sizes are smaller than L0. This
             // causes compaction scoring, which depends on level sizes, to favor L1+
@@ -185,7 +180,7 @@ impl DynamicLevelSelector {
         let mut ctx = self.calculate_level_base_size(levels);
 
         // The bottommost level can not be input level.
-        for level in &levels[..self.config.max_level] {
+        for level in &levels[..self.config.max_level as usize] {
             let level_idx = level.level_idx as usize;
             let idle_file_count =
                 (level.table_infos.len() - handlers[level_idx].get_pending_file_count()) as u64;
@@ -254,12 +249,13 @@ pub mod tests {
     use std::ops::Range;
 
     use itertools::Itertools;
+    use risingwave_pb::hummock::compaction_config::CompactionMode;
     use risingwave_pb::hummock::{LevelType, SstableInfo};
 
     use super::*;
+    use crate::hummock::compaction::compaction_config::CompactionConfigBuilder;
     use crate::hummock::compaction::overlap_strategy::RangeOverlapStrategy;
     use crate::hummock::compaction::tier_compaction_picker::tests::generate_table;
-    use crate::hummock::compaction::CompactionMode::RangeMode;
 
     pub fn generate_tables(
         ids: Range<u64>,
@@ -291,16 +287,16 @@ pub mod tests {
 
     #[test]
     fn test_dynamic_level() {
-        let config = CompactionConfig {
-            max_bytes_for_level_base: 100,
-            max_level: 4,
-            max_bytes_for_level_multiplier: 5,
-            max_compaction_bytes: 0,
-            min_compaction_bytes: 1,
-            level0_tigger_file_numer: 1,
-            level0_tier_compact_file_number: 2,
-            compaction_mode: RangeMode,
-        };
+        let config = CompactionConfigBuilder::new()
+            .max_bytes_for_level_base(100)
+            .max_level(4)
+            .max_bytes_for_level_multiplier(5)
+            .max_compaction_bytes(1)
+            .min_compaction_bytes(0)
+            .level0_tigger_file_numer(1)
+            .level0_tier_compact_file_number(2)
+            .compaction_mode(CompactionMode::Range as i32)
+            .build();
         let selector =
             DynamicLevelSelector::new(Arc::new(config), Arc::new(RangeOverlapStrategy::default()));
         let mut levels = vec![
@@ -373,16 +369,16 @@ pub mod tests {
 
     #[test]
     fn test_pick_compaction() {
-        let mut config = CompactionConfig {
-            max_bytes_for_level_base: 200,
-            max_level: 4,
-            max_bytes_for_level_multiplier: 5,
-            max_compaction_bytes: 10000,
-            min_compaction_bytes: 200,
-            level0_tigger_file_numer: 8,
-            level0_tier_compact_file_number: 4,
-            compaction_mode: RangeMode,
-        };
+        let config = CompactionConfigBuilder::new()
+            .max_bytes_for_level_base(200)
+            .max_level(4)
+            .max_bytes_for_level_multiplier(5)
+            .max_compaction_bytes(10000)
+            .min_compaction_bytes(200)
+            .level0_tigger_file_numer(8)
+            .level0_tier_compact_file_number(4)
+            .compaction_mode(CompactionMode::Range as i32)
+            .build();
         let mut levels = vec![
             generate_level(0, generate_tables(15..25, 0..600, 3, 10)),
             generate_level(1, vec![]),
@@ -405,9 +401,11 @@ pub mod tests {
         assert_eq!(compaction.select_level.table_infos.len(), 10);
         assert_eq!(compaction.target_level.table_infos.len(), 0);
 
-        config.min_compaction_bytes = 1;
-        config.max_bytes_for_level_base = 100;
-        config.level0_tigger_file_numer = 8;
+        let config = CompactionConfigBuilder::new_with(config)
+            .min_compaction_bytes(1)
+            .max_bytes_for_level_base(100)
+            .level0_tigger_file_numer(8)
+            .build();
         let selector =
             DynamicLevelSelector::new(Arc::new(config), Arc::new(RangeOverlapStrategy::default()));
         let mut levels_handlers = (0..5).into_iter().map(LevelHandler::new).collect_vec();