feat(storage): Splitting table change log from HummockVersion on CN side

src/storage/hummock_sdk/src/change_log.rs

@@ -51,6 +51,10 @@ impl<T> TableChangeLogCommon<T> {
         }
         self.0.push_back(new_change_log);
     }
+
+    pub fn change_log_into_iter(self) -> impl Iterator<Item = EpochNewChangeLogCommon<T>> {
+        self.0.into_iter()
+    }
 }
 
 pub type TableChangeLog = TableChangeLogCommon<SstableInfo>;

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

@@ -36,7 +36,7 @@ use crate::level::{Level, LevelCommon, Levels, OverlappingLevel};
 use crate::sstable_info::SstableInfo;
 use crate::table_watermark::{ReadTableWatermark, TableWatermarks};
 use crate::version::{
-    GroupDelta, GroupDeltaCommon, HummockVersion, HummockVersionCommon, HummockVersionDelta,
+    GroupDelta, GroupDeltaCommon, HummockVersion, HummockVersionCommon, HummockVersionDeltaCommon,
     HummockVersionStateTableInfo, IntraLevelDelta, IntraLevelDeltaCommon, ObjectIdReader,
     SstableIdReader,
 };
@@ -50,7 +50,7 @@ pub struct SstDeltaInfo {
 
 pub type BranchedSstInfo = HashMap<CompactionGroupId, Vec<HummockSstableId>>;
 
-impl HummockVersion {
+impl<L> HummockVersionCommon<SstableInfo, L> {
     pub fn get_compaction_group_levels(&self, compaction_group_id: CompactionGroupId) -> &Levels {
         self.levels
             .get(&compaction_group_id)
@@ -187,7 +187,7 @@ pub fn safe_epoch_read_table_watermarks_impl(
         .collect()
 }
 
-impl HummockVersion {
+impl<L: Clone> HummockVersionCommon<SstableInfo, L> {
     pub fn count_new_ssts_in_group_split(
         &self,
         parent_group_id: CompactionGroupId,
@@ -356,7 +356,10 @@ impl HummockVersion {
             .all(|level| !level.table_infos.is_empty()));
     }
 
-    pub fn build_sst_delta_infos(&self, version_delta: &HummockVersionDelta) -> Vec<SstDeltaInfo> {
+    pub fn build_sst_delta_infos(
+        &self,
+        version_delta: &HummockVersionDeltaCommon<SstableInfo, L>,
+    ) -> Vec<SstDeltaInfo> {
         let mut infos = vec![];
 
         // Skip trivial move delta for refiller
@@ -459,7 +462,10 @@ impl HummockVersion {
         infos
     }
 
-    pub fn apply_version_delta(&mut self, version_delta: &HummockVersionDelta) {
+    pub fn apply_version_delta(
+        &mut self,
+        version_delta: &HummockVersionDeltaCommon<SstableInfo, L>,
+    ) {
         assert_eq!(self.id, version_delta.prev_id);
 
         let (changed_table_info, mut is_commit_epoch) = self.state_table_info.apply_delta(
@@ -930,16 +936,10 @@ impl HummockVersion {
     }
 }
 
-impl<T> HummockVersionCommon<T>
+impl<T> HummockVersionCommon<T, T>
 where
     T: SstableIdReader + ObjectIdReader,
 {
-    pub fn get_combined_levels(&self) -> impl Iterator<Item = &'_ LevelCommon<T>> + '_ {
-        self.levels
-            .values()
-            .flat_map(|level| level.l0.sub_levels.iter().rev().chain(level.levels.iter()))
-    }
-
     pub fn get_object_ids(&self) -> HashSet<HummockSstableObjectId> {
         self.get_sst_infos().map(|s| s.object_id()).collect()
     }
@@ -1094,6 +1094,14 @@ impl Levels {
     }
 }
 
+impl<T, L> HummockVersionCommon<T, L> {
+    pub fn get_combined_levels(&self) -> impl Iterator<Item = &'_ LevelCommon<T>> + '_ {
+        self.levels
+            .values()
+            .flat_map(|level| level.l0.sub_levels.iter().rev().chain(level.levels.iter()))
+    }
+}
+
 pub fn build_initial_compaction_group_levels(
     group_id: CompactionGroupId,
     compaction_config: &CompactionConfig,

src/storage/hummock_sdk/src/time_travel.rs

@@ -28,7 +28,7 @@ use crate::version::{
 };
 use crate::{CompactionGroupId, HummockSstableId, HummockSstableObjectId};
 
-pub type IncompleteHummockVersion = HummockVersionCommon<SstableIdInVersion>;
+pub type IncompleteHummockVersion = HummockVersionCommon<SstableIdInVersion, SstableIdInVersion>;
 
 /// Populates `SstableInfo` for `table_id`.
 /// `SstableInfo` not associated with `table_id` is removed.
@@ -159,7 +159,8 @@ fn rewrite_levels(mut levels: PbLevels, time_travel_table_ids: &HashSet<StateTab
 /// [`IncompleteHummockVersionDelta`] is incomplete because `SSTableInfo` only has the `sst_id` set in the following fields:
 /// - `PbGroupDeltas`
 /// - `ChangeLogDelta`
-pub type IncompleteHummockVersionDelta = HummockVersionDeltaCommon<SstableIdInVersion>;
+pub type IncompleteHummockVersionDelta =
+    HummockVersionDeltaCommon<SstableIdInVersion, SstableIdInVersion>;
 
 /// `SStableInfo` will be stripped.
 impl From<(&HummockVersionDelta, &HashSet<StateTableId>)> for IncompleteHummockVersionDelta {