feat(sst id): avoid SST ID gap when generating (risingwavelabs#8542)

Author: soundOfDestiny

src/meta/src/hummock/manager/compaction_group_manager.rs

@@ -12,16 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::ops::DerefMut;
 use std::sync::Arc;
 
 use function_name::named;
 use itertools::Itertools;
 use risingwave_hummock_sdk::compaction_group::hummock_version_ext::{
     build_version_delta_after_version, get_compaction_group_ids, get_compaction_group_ssts,
-    get_member_table_ids, try_get_compaction_group_id_by_table_id, HummockLevelsExt,
-    HummockVersionExt, HummockVersionUpdateExt,
+    get_member_table_ids, try_get_compaction_group_id_by_table_id, HummockVersionExt,
+    HummockVersionUpdateExt,
 };
 use risingwave_hummock_sdk::compaction_group::{StateTableId, StaticCompactionGroupId};
 use risingwave_hummock_sdk::CompactionGroupId;
@@ -485,7 +485,10 @@ impl<S: MetaStore> HummockManager<S> {
             .env
             .id_gen_manager()
             .generate_interval::<{ IdCategory::HummockSstableId }>(
-                parent_group.count_ssts() as u64 * 2,
+                versioning.current_version.count_new_ssts_in_group_split(
+                    parent_group_id,
+                    &HashSet::from_iter(table_ids.iter().cloned()),
+                ),
             )
             .await?;
         let group_deltas = &mut new_version_delta

src/meta/src/hummock/manager/versioning.rs

@@ -79,7 +79,7 @@ pub struct Versioning {
     /// - AND It either contains no SST to delete, or all these SSTs has been deleted. See
     ///   `extend_objects_to_delete_from_deltas`.
     pub deltas_to_delete: Vec<HummockVersionId>,
-    /// SST which is referenced more than once
+    /// SST whose `object_id` != `sst_id`
     pub branched_ssts: BTreeMap<
         // SST object id
         HummockSstableObjectId,

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

@@ -107,6 +107,11 @@ pub trait HummockVersionExt {
 }
 
 pub trait HummockVersionUpdateExt {
+    fn count_new_ssts_in_group_split(
+        &mut self,
+        parent_group_id: CompactionGroupId,
+        member_table_ids: &HashSet<StateTableId>,
+    ) -> u64;
     fn init_with_parent_group(
         &mut self,
         parent_group_id: CompactionGroupId,
@@ -200,6 +205,43 @@ pub type SstSplitInfo = (
 );
 
 impl HummockVersionUpdateExt for HummockVersion {
+    fn count_new_ssts_in_group_split(
+        &mut self,
+        parent_group_id: CompactionGroupId,
+        member_table_ids: &HashSet<StateTableId>,
+    ) -> u64 {
+        self.levels
+            .get(&parent_group_id)
+            .map_or(0, |parent_levels| {
+                parent_levels
+                    .l0
+                    .iter()
+                    .flat_map(|l0| l0.get_sub_levels())
+                    .chain(parent_levels.get_levels().iter())
+                    .flat_map(|level| level.get_table_infos())
+                    .map(|sst_info| {
+                        // `flag` is a bitmap
+                        let mut flag = 0;
+                        // `sst_info.table_ids` will never be empty.
+                        for table_id in sst_info.get_table_ids() {
+                            flag |= if member_table_ids.contains(table_id) {
+                                2
+                            } else {
+                                1
+                            };
+                            if flag == 3 {
+                                break;
+                            }
+                        }
+                        // We need to replace the SST id of the divided part in parent group with a
+                        // new SST id when it's not a trivial adjust. View function
+                        // `init_with_parent_group` for details.
+                        flag - 1
+                    })
+                    .sum()
+            })
+    }
+
     fn init_with_parent_group(
         &mut self,
         parent_group_id: CompactionGroupId,