feat(storage): support compaction group split and merge

grafana/risingwave-dev-dashboard.dashboard.py

@@ -3254,12 +3254,17 @@ def section_hummock_manager(outer_panels):
                     ],
                 ),
                 panels.timeseries_count(
-                    "Move State Table Count",
-                    "The times of move_state_table occurs",
+                    "Compaction Group Schedule",
+                    "The times of schedule occurs",
                     [
                         panels.target(
-                            f"sum({table_metric('storage_move_state_table_count')}) by (group)",
-                            "move table cg{{group}}",
+                            f"sum({table_metric('storage_split_compaction_group_count')}) by (group)",
+                            "split_compaction_group_count cg{{group}}",
+                        ),
+
+                        panels.target(
+                            f"sum({table_metric('storage_merge_compaction_group_count')}) by (group)",
+                            "merge_compaction_group_count cg{{group}}",
                         ),
                     ],
                 ),

proto/hummock.proto

@@ -85,6 +85,13 @@ message GroupConstruct {
   uint64 group_id = 4;
   uint64 new_sst_start_id = 5;
   CompatibilityVersion version = 6;
+
+  // The split_key is the key that the group is split by.
+  // When GroupConstruct with commit_epoch, split_key will be empty
+  // When split_key is not None, GroupConstruct tells to use split_key to check each level and split the sstable_info in the level into two groups (bounded by split_key).
+  // For the left sstable_info, split_key (right_exclusive=true) will be used as key_range_right.
+  // In the current implementation split_key always contains a table_id, vnode = 0, epoch = 0
+  optional bytes split_key = 7;
 }
 
 message GroupMetaChange {
@@ -729,6 +736,7 @@ message RiseCtlUpdateCompactionConfigRequest {
       CompressionAlgorithm compression_algorithm = 17;
       uint32 max_l0_compact_level_count = 18;
       uint64 sst_allowed_trivial_move_min_size = 19;
+      uint32 split_weight_by_vnode = 20;
     }
   }
   repeated uint64 compaction_group_ids = 1;

src/common/src/config.rs

@@ -1405,7 +1405,7 @@ pub mod default {
         }
 
         pub fn move_table_size_limit() -> u64 {
-            10 * 1024 * 1024 * 1024 // 10GB
+            4 * 1024 * 1024 * 1024 // 4GB
         }
 
         pub fn split_group_size_limit() -> u64 {

src/config/docs.md

@@ -47,7 +47,7 @@ This page is automatically generated by `./risedev generate-example-config`
 | min_delta_log_num_for_hummock_version_checkpoint | The minimum delta log number a new checkpoint should compact, otherwise the checkpoint attempt is rejected. | 10 |
 | min_sst_retention_time_sec | Objects within `min_sst_retention_time_sec` won't be deleted by hummock full GC, even they are dangling. | 86400 |
 | min_table_split_write_throughput | If the size of one table is smaller than `min_table_split_write_throughput`, we would not split it to an single group. | 4194304 |
-| move_table_size_limit |  | 10737418240 |
+| move_table_size_limit |  | 4294967296 |
 | node_num_monitor_interval_sec |  | 10 |
 | parallelism_control_batch_size | The number of streaming jobs per scaling operation. | 10 |
 | parallelism_control_trigger_first_delay_sec | The first delay of parallelism control. | 30 |

src/config/example.toml

@@ -40,7 +40,7 @@ periodic_space_reclaim_compaction_interval_sec = 3600
 periodic_ttl_reclaim_compaction_interval_sec = 1800
 periodic_tombstone_reclaim_compaction_interval_sec = 600
 periodic_split_compact_group_interval_sec = 10
-move_table_size_limit = 10737418240
+move_table_size_limit = 4294967296
 split_group_size_limit = 68719476736
 cut_table_size_limit = 1073741824
 do_not_config_object_storage_lifecycle = false

src/frontend/src/catalog/system_catalog/rw_catalog/rw_hummock_version.rs

@@ -114,7 +114,7 @@ fn version_to_sstable_rows(version: HummockVersion) -> Vec<RwHummockSstable> {
     for cg in version.levels.into_values() {
         for level in cg.levels.into_iter().chain(cg.l0.sub_levels) {
             for sst in level.table_infos {
-                let key_range = sst.key_range;
+                let key_range = &sst.key_range;
                 sstables.push(RwHummockSstable {
                     sstable_id: sst.sst_id as _,
                     object_id: sst.object_id as _,

src/meta/src/hummock/manager/commit_epoch.rs

@@ -13,11 +13,12 @@
 // limitations under the License.
 
 use std::collections::{BTreeMap, HashMap, HashSet};
+use std::ops::Deref;
+use std::sync::Arc;
 
 use risingwave_common::catalog::TableId;
 use risingwave_hummock_sdk::change_log::ChangeLogDelta;
-use risingwave_hummock_sdk::compaction_group::hummock_version_ext::split_sst;
-use risingwave_hummock_sdk::compaction_group::StaticCompactionGroupId;
+use risingwave_hummock_sdk::compaction_group::group_split::split_sst_for_commit_epoch;
 use risingwave_hummock_sdk::sstable_info::SstableInfo;
 use risingwave_hummock_sdk::table_stats::{
     add_prost_table_stats_map, purge_prost_table_stats, to_prost_table_stats_map, PbTableStatsMap,
@@ -160,65 +161,78 @@ impl HummockManager {
         let state_table_info = &version.latest_version().state_table_info;
         let mut table_compaction_group_mapping = state_table_info.build_table_compaction_group_id();
 
-        // Add new table
-        let (new_table_ids, new_compaction_group, compaction_group_manager_txn) =
+        let mut new_tables = None;
+        let mut compaction_group_manager_txn = None;
+        let mut compaction_group_config = None;
+
+        if !matches!(new_table_fragment_info, NewTableFragmentInfo::None) {
+            let mut new_table_ids = Vec::default();
+            if compaction_group_manager_txn.is_none() {
+                let compaction_group_manager_guard = self.compaction_group_manager.write().await;
+                let config = compaction_group_manager_guard
+                    .default_compaction_config()
+                    .deref()
+                    .clone();
+                compaction_group_config = Some(config);
+                compaction_group_manager_txn =
+                    Some(CompactionGroupManager::start_owned_compaction_groups_txn(
+                        compaction_group_manager_guard,
+                    ));
+            }
+
+            let new_compaction_group_id = next_compaction_group_id(&self.env).await?;
+            compaction_group_manager_txn.as_mut().unwrap().insert(
+                new_compaction_group_id,
+                CompactionGroup {
+                    group_id: new_compaction_group_id,
+                    compaction_config: Arc::new(compaction_group_config.as_ref().unwrap().clone()),
+                },
+            );
+
             match new_table_fragment_info {
                 NewTableFragmentInfo::Normal {
                     mv_table_id,
                     internal_table_ids,
                 } => {
-                    let mut new_table_ids = HashMap::new();
-                    on_handle_add_new_table(
-                        state_table_info,
-                        &internal_table_ids,
-                        StaticCompactionGroupId::StateDefault as u64,
-                        &mut table_compaction_group_mapping,
-                        &mut new_table_ids,
-                    )?;
+                    if !internal_table_ids.is_empty() {
+                        on_handle_add_new_table(
+                            state_table_info,
+                            &internal_table_ids,
+                            new_compaction_group_id,
+                            &mut table_compaction_group_mapping,
+                        )?;
+
+                        new_table_ids.extend(internal_table_ids);
+                    }
 
-                    on_handle_add_new_table(
-                        state_table_info,
-                        &mv_table_id,
-                        StaticCompactionGroupId::MaterializedView as u64,
-                        &mut table_compaction_group_mapping,
-                        &mut new_table_ids,
-                    )?;
-                    (new_table_ids, None, None)
+                    if let Some(mv_table_id) = mv_table_id {
+                        on_handle_add_new_table(
+                            state_table_info,
+                            &[mv_table_id],
+                            new_compaction_group_id,
+                            &mut table_compaction_group_mapping,
+                        )?;
+                        new_table_ids.push(mv_table_id);
+                    }
                 }
                 NewTableFragmentInfo::NewCompactionGroup { table_ids } => {
-                    let compaction_group_manager_guard =
-                        self.compaction_group_manager.write().await;
-                    let compaction_group_config =
-                        compaction_group_manager_guard.default_compaction_config();
-                    let mut compaction_group_manager =
-                        CompactionGroupManager::start_owned_compaction_groups_txn(
-                            compaction_group_manager_guard,
-                        );
-                    let mut new_table_ids = HashMap::new();
-                    let new_compaction_group_id = next_compaction_group_id(&self.env).await?;
-                    compaction_group_manager.insert(
-                        new_compaction_group_id,
-                        CompactionGroup {
-                            group_id: new_compaction_group_id,
-                            compaction_config: compaction_group_config.clone(),
-                        },
-                    );
-
                     on_handle_add_new_table(
                         state_table_info,
                         &table_ids,
                         new_compaction_group_id,
                         &mut table_compaction_group_mapping,
-                        &mut new_table_ids,
                     )?;
-                    (
-                        new_table_ids,
-                        Some((new_compaction_group_id, (*compaction_group_config).clone())),
-                        Some(compaction_group_manager),
-                    )
+                    new_table_ids.extend(table_ids.iter().cloned());
                 }
-                NewTableFragmentInfo::None => (HashMap::new(), None, None),
-            };
+                NewTableFragmentInfo::None => unreachable!(),
+            }
+
+            new_tables = Some((
+                new_table_ids,
+                new_compaction_group_id,
+                compaction_group_config.as_ref().unwrap().clone(),
+            ));
+        }
 
         let commit_sstables = self
             .correct_commit_ssts(sstables, &table_compaction_group_mapping)
@@ -230,9 +244,8 @@ impl HummockManager {
             committed_epoch,
             &tables_to_commit,
             is_visible_table_committed_epoch,
-            new_compaction_group,
             commit_sstables,
-            &new_table_ids,
+            new_tables,
             new_table_watermarks,
             change_log_delta,
         );
@@ -375,7 +388,7 @@ impl HummockManager {
         }
     }
 
-    async fn correct_commit_ssts(
+    pub(crate) async fn correct_commit_ssts(
         &self,
         sstables: Vec<LocalSstableInfo>,
         table_compaction_group_mapping: &HashMap<TableId, CompactionGroupId>,
@@ -424,7 +437,6 @@ impl HummockManager {
                         .push(sst.sst_info);
                     break;
                 }
-
                 let origin_sst_size = sst.sst_info.sst_size;
                 let new_sst_size = match_ids
                     .iter()
@@ -434,7 +446,8 @@ impl HummockManager {
                     })
                     .sum();
 
-                let branch_sst = split_sst(
+                // let split_key = build_split_key_with_table_id(match_ids.last().unwrap() + 1);
+                let branch_sst = split_sst_for_commit_epoch(
                     &mut sst.sst_info,
                     &mut new_sst_id,
                     origin_sst_size - new_sst_size,
@@ -458,7 +471,6 @@ fn on_handle_add_new_table(
     table_ids: impl IntoIterator<Item = &TableId>,
     compaction_group_id: CompactionGroupId,
     table_compaction_group_mapping: &mut HashMap<TableId, CompactionGroupId>,
-    new_table_ids: &mut HashMap<TableId, CompactionGroupId>,
 ) -> Result<()> {
     for table_id in table_ids {
         if let Some(info) = state_table_info.info().get(table_id) {
@@ -468,7 +480,6 @@ fn on_handle_add_new_table(
             )));
         }
         table_compaction_group_mapping.insert(*table_id, compaction_group_id);
-        new_table_ids.insert(*table_id, compaction_group_id);
     }
 
     Ok(())