refactor(storage): maintain sub_level id order

proto/hummock.proto

@@ -189,6 +189,7 @@ message HummockVersion {
   map<uint32, TableWatermarks> table_watermarks = 5;
   map<uint32, TableChangeLog> table_change_logs = 6;
   map<uint32, StateTableInfo> state_table_info = 7;
+  optional uint64 max_sub_level_id = 8;
 }
 
 message HummockVersionDelta {

src/meta/src/hummock/manager/transaction.rs

@@ -25,6 +25,7 @@ use risingwave_hummock_sdk::version::{
 };
 use risingwave_hummock_sdk::{
     CompactionGroupId, FrontendHummockVersionDelta, HummockEpoch, HummockVersionId,
+    FIRST_SUB_LEVEL_ID,
 };
 use risingwave_pb::hummock::{
     CompactionConfig, CompatibilityVersion, GroupConstruct, HummockVersionDeltas,
@@ -149,14 +150,32 @@ impl<'a> HummockVersionTransaction<'a> {
 
         // Append SSTs to a new version.
         for (compaction_group_id, inserted_table_infos) in commit_sstables {
+            let l0_sub_level_id = new_version_delta
+                .latest_version()
+                .levels
+                .get(&compaction_group_id)
+                .and_then(|levels| {
+                    levels
+                        .l0
+                        .sub_levels
+                        .last()
+                        .map(|level| level.sub_level_id + 1)
+                })
+                .unwrap_or(
+                    new_version_delta
+                        .latest_version()
+                        .max_sub_level_id
+                        .map(|id| id + 1)
+                        .unwrap_or(FIRST_SUB_LEVEL_ID),
+                );
             let group_deltas = &mut new_version_delta
                 .group_deltas
                 .entry(compaction_group_id)
                 .or_default()
                 .group_deltas;
             let group_delta = GroupDelta::IntraLevel(IntraLevelDelta::new(
                 0,
-                0,      // l0_sub_level_id will be generated during apply_version_delta
+                l0_sub_level_id,
                 vec![], // default
                 inserted_table_infos,
                 0, // default

src/storage/hummock_sdk/src/compaction_group/hummock_version_ext.rs

@@ -18,6 +18,7 @@ use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 use std::sync::Arc;
 
 use bytes::Bytes;
+use itertools::Either::{Left, Right};
 use itertools::Itertools;
 use risingwave_common::catalog::TableId;
 use risingwave_common::hash::VnodeBitmapExt;
@@ -434,10 +435,10 @@ impl HummockVersion {
                     continue;
                 }
                 match group_split::get_sub_level_insert_hint(&target_l0.sub_levels, sub_level) {
-                    Ok(idx) => {
+                    Left(idx) => {
                         add_ssts_to_sub_level(target_l0, idx, insert_table_infos);
                     }
-                    Err(idx) => {
+                    Right(idx) => {
                         insert_new_sub_level(
                             target_l0,
                             sub_level.sub_level_id,
@@ -660,15 +661,10 @@ impl HummockVersion {
                         || group_destroy.is_some(),
                     "no sst should be deleted when committing an epoch"
                 );
-                let mut next_l0_sub_level_id = levels
-                    .l0
-                    .sub_levels
-                    .last()
-                    .map(|level| level.sub_level_id + 1)
-                    .unwrap_or(1);
                 for group_delta in &group_deltas.group_deltas {
                     if let GroupDelta::IntraLevel(IntraLevelDelta {
                         level_idx,
+                        l0_sub_level_id,
                         inserted_table_infos,
                         ..
                     }) = group_delta
@@ -680,12 +676,11 @@ impl HummockVersion {
                         if !inserted_table_infos.is_empty() {
                             insert_new_sub_level(
                                 &mut levels.l0,
-                                next_l0_sub_level_id,
+                                *l0_sub_level_id,
                                 PbLevelType::Overlapping,
                                 inserted_table_infos.clone(),
                                 None,
                             );
-                            next_l0_sub_level_id += 1;
                         }
                     }
                 }
@@ -764,6 +759,14 @@ impl HummockVersion {
             &version_delta.state_table_info_delta,
             &changed_table_info,
         );
+
+        // The max_sub_level_id may have been increased, e.g. during commit_epoch or merge_group.
+        // Recalculate it.
+        self.max_sub_level_id = self
+            .levels
+            .values()
+            .filter_map(|levels| levels.l0.sub_levels.iter().map(|s| s.sub_level_id).max())
+            .max();
     }
 
     pub fn apply_change_log_delta<T: Clone>(
@@ -883,7 +886,7 @@ impl HummockVersion {
             )
         });
 
-        group_split::merge_levels(left_levels, right_levels);
+        group_split::merge_levels(left_levels, right_levels, self.max_sub_level_id);
     }
 }
 
@@ -952,10 +955,10 @@ impl HummockVersionCommon<SstableInfo> {
                         l0.uncompressed_file_size -= sst_info.uncompressed_file_size;
                     });
                 match group_split::get_sub_level_insert_hint(&target_l0.sub_levels, sub_level) {
-                    Ok(idx) => {
+                    Left(idx) => {
                         add_ssts_to_sub_level(target_l0, idx, insert_table_infos);
                     }
-                    Err(idx) => {
+                    Right(idx) => {
                         insert_new_sub_level(
                             target_l0,
                             sub_level.sub_level_id,
@@ -2032,7 +2035,7 @@ mod tests {
             let mut left_levels = Levels::default();
             let right_levels = Levels::default();
 
-            group_split::merge_levels(&mut left_levels, right_levels);
+            group_split::merge_levels(&mut left_levels, right_levels, None);
         }
 
         {
@@ -2046,7 +2049,7 @@ mod tests {
             );
             let right_levels = right_levels.clone();
 
-            group_split::merge_levels(&mut left_levels, right_levels);
+            group_split::merge_levels(&mut left_levels, right_levels, None);
 
             assert!(left_levels.l0.sub_levels.len() == 3);
             assert!(left_levels.l0.sub_levels[0].sub_level_id == 101);
@@ -2071,7 +2074,7 @@ mod tests {
                 },
             );
 
-            group_split::merge_levels(&mut left_levels, right_levels);
+            group_split::merge_levels(&mut left_levels, right_levels, None);
 
             assert!(left_levels.l0.sub_levels.len() == 3);
             assert!(left_levels.l0.sub_levels[0].sub_level_id == 101);
@@ -2089,7 +2092,7 @@ mod tests {
             let mut left_levels = left_levels.clone();
             let right_levels = right_levels.clone();
 
-            group_split::merge_levels(&mut left_levels, right_levels);
+            group_split::merge_levels(&mut left_levels, right_levels, None);
 
             assert!(left_levels.l0.sub_levels.len() == 6);
             assert!(left_levels.l0.sub_levels[0].sub_level_id == 101);
@@ -2459,7 +2462,7 @@ mod tests {
                 ..Default::default()
             };
 
-            merge_levels(cg1, right_levels);
+            merge_levels(cg1, right_levels, None);
         }
 
         {

src/storage/hummock_sdk/src/compaction_group/mod.rs

@@ -49,6 +49,8 @@ pub mod group_split {
     use std::collections::BTreeSet;
 
     use bytes::Bytes;
+    use itertools::Either;
+    use itertools::Either::{Left, Right};
     use risingwave_common::catalog::TableId;
     use risingwave_common::hash::VirtualNode;
     use risingwave_pb::hummock::PbLevelType;
@@ -59,7 +61,7 @@ pub mod group_split {
     use crate::key_range::KeyRange;
     use crate::level::{Level, Levels};
     use crate::sstable_info::SstableInfo;
-    use crate::{can_concat, HummockEpoch, KeyComparator};
+    use crate::{can_concat, HummockEpoch, KeyComparator, FIRST_SUB_LEVEL_ID};
 
     /// The split will follow the following rules:
     /// 1. Ssts with `split_key` will be split into two separate ssts and their `key_range` will be changed `sst_1`: [`sst.key_range.right`, `split_key`) `sst_2`: [`split_key`, `sst.key_range.right`].
@@ -272,24 +274,32 @@ pub mod group_split {
     }
 
     /// Merge the right levels into the left levels.
-    pub fn merge_levels(left_levels: &mut Levels, right_levels: Levels) {
+    pub fn merge_levels(
+        left_levels: &mut Levels,
+        right_levels: Levels,
+        max_sub_level_id: Option<u64>,
+    ) {
         let right_l0 = right_levels.l0;
 
-        let mut max_left_sub_level_id = left_levels
+        let mut next_right_sub_level_id = left_levels
             .l0
             .sub_levels
             .iter()
             .map(|sub_level| sub_level.sub_level_id + 1)
             .max()
-            .unwrap_or(0); // If there are no sub levels, the max sub level id is 0.
-        let need_rewrite_right_sub_level_id = max_left_sub_level_id != 0;
+            .unwrap_or(
+                max_sub_level_id
+                    .map(|v| v + 1)
+                    .unwrap_or(FIRST_SUB_LEVEL_ID),
+            );
+        let need_rewrite_right_sub_level_id = !left_levels.l0.sub_levels.is_empty();
 
         for mut right_sub_level in right_l0.sub_levels {
             // Rewrtie the sub level id of right sub level to avoid conflict with left sub levels. (conflict level type)
             // e.g. left sub levels: [0, 1, 2], right sub levels: [0, 1, 2], after rewrite, right sub levels: [3, 4, 5]
             if need_rewrite_right_sub_level_id {
-                right_sub_level.sub_level_id = max_left_sub_level_id;
-                max_left_sub_level_id += 1;
+                right_sub_level.sub_level_id = next_right_sub_level_id;
+                next_right_sub_level_id += 1;
             }
 
             insert_new_sub_level(
@@ -354,26 +364,26 @@ pub mod group_split {
         }
     }
 
-    // When `insert_hint` is `Ok(idx)`, it means that the sub level `idx` in `target_l0`
-    // will extend these SSTs. When `insert_hint` is `Err(idx)`, it
+    // When `insert_hint` is `Left(idx)`, it means that the sub level `idx` in `target_l0`
+    // will extend these SSTs. When `insert_hint` is `Right(idx)`, it
     // means that we will add a new sub level `idx` into `target_l0`.
     pub fn get_sub_level_insert_hint(
         target_levels: &Vec<Level>,
         sub_level: &Level,
-    ) -> Result<usize, usize> {
+    ) -> Either<usize, usize> {
         for (idx, other) in target_levels.iter().enumerate() {
             match other.sub_level_id.cmp(&sub_level.sub_level_id) {
                 Ordering::Less => {}
                 Ordering::Equal => {
-                    return Ok(idx);
+                    return Left(idx);
                 }
                 Ordering::Greater => {
-                    return Err(idx);
+                    return Right(idx);
                 }
             }
         }
 
-        Err(target_levels.len())
+        Right(target_levels.len())
     }
 
     /// Split the SSTs in the level according to the split key.

src/storage/hummock_sdk/src/frontend_version.rs

@@ -89,6 +89,8 @@ impl FrontendHummockVersion {
                 })
                 .collect(),
             state_table_info: self.state_table_info.to_protobuf(),
+            // max_sub_level_id is not expected to be used
+            max_sub_level_id: None,
         }
     }
 

src/storage/hummock_sdk/src/lib.rs

@@ -130,6 +130,7 @@ impl Sub for HummockVersionId {
 
 pub const INVALID_VERSION_ID: HummockVersionId = HummockVersionId(0);
 pub const FIRST_VERSION_ID: HummockVersionId = HummockVersionId(1);
+pub const FIRST_SUB_LEVEL_ID: u64 = 1;
 pub const SPLIT_TABLE_COMPACTION_GROUP_ID_HEAD: u64 = 1u64 << 56;
 pub const SINGLE_TABLE_COMPACTION_GROUP_ID_HEAD: u64 = 2u64 << 56;
 pub const OBJECT_SUFFIX: &str = "data";

src/storage/hummock_sdk/src/time_travel.rs

@@ -124,6 +124,7 @@ impl From<(&HummockVersion, &HashSet<CompactionGroupId>)> for IncompleteHummockV
                 })
                 .collect(),
             state_table_info: version.state_table_info.clone(),
+            max_sub_level_id: version.max_sub_level_id,
         }
     }
 }

src/storage/hummock_sdk/src/version.rs

@@ -224,6 +224,8 @@ pub struct HummockVersionCommon<T> {
     pub table_watermarks: HashMap<TableId, Arc<TableWatermarks>>,
     pub table_change_log: HashMap<TableId, TableChangeLogCommon<T>>,
     pub state_table_info: HummockVersionStateTableInfo,
+    /// The maximum `sub_level_id` that has been recorded thus far, regardless of whether the `sub_level_id` currently exists.
+    pub max_sub_level_id: Option<u64>,
 }
 
 pub type HummockVersion = HummockVersionCommon<SstableInfo>;
@@ -312,6 +314,21 @@ where
             state_table_info: HummockVersionStateTableInfo::from_protobuf(
                 &pb_version.state_table_info,
             ),
+            // The next_sub_level_id is expected to be None in Pb from previous kernel version.
+            // For backward compatibility, try to initialize next_sub_level_id with max(existing sub_level ids).
+            max_sub_level_id: pb_version.max_sub_level_id.or_else(|| {
+                pb_version
+                    .levels
+                    .values()
+                    .filter_map(|levels| {
+                        levels
+                            .l0
+                            .as_ref()
+                            .map(|l0| l0.sub_levels.iter().map(|s| s.sub_level_id).max())
+                    })
+                    .flatten()
+                    .max()
+            }),
         }
     }
 }
@@ -341,6 +358,7 @@ where
                 .map(|(table_id, change_log)| (table_id.table_id, change_log.to_protobuf()))
                 .collect(),
             state_table_info: version.state_table_info.to_protobuf(),
+            max_sub_level_id: version.max_sub_level_id,
         }
     }
 }
@@ -371,6 +389,7 @@ where
                 .map(|(table_id, change_log)| (table_id.table_id, change_log.to_protobuf()))
                 .collect(),
             state_table_info: version.state_table_info.to_protobuf(),
+            max_sub_level_id: version.max_sub_level_id,
         }
     }
 }
@@ -433,6 +452,7 @@ impl HummockVersion {
             table_watermarks: HashMap::new(),
             table_change_log: HashMap::new(),
             state_table_info: HummockVersionStateTableInfo::empty(),
+            max_sub_level_id: None,
         };
         for group_id in [
             StaticCompactionGroupId::StateDefault as CompactionGroupId,