refactor(storage): distinguish SST id and object id (close #8434)

proto/hummock.proto

@@ -10,15 +10,14 @@ option java_package = "com.risingwave.proto";
 option optimize_for = SPEED;
 
 message SstableInfo {
-  uint64 id = 1;
-  KeyRange key_range = 2;
-  uint64 file_size = 3;
-  repeated uint32 table_ids = 4;
-  uint64 meta_offset = 5;
-  uint64 stale_key_count = 6;
-  uint64 total_key_count = 7;
-  // When a SST is divided, its divide_version will increase one.
-  uint64 divide_version = 8;
+  uint64 object_id = 1;
+  uint64 sst_id = 2;
+  KeyRange key_range = 3;
+  uint64 file_size = 4;
+  repeated uint32 table_ids = 5;
+  uint64 meta_offset = 6;
+  uint64 stale_key_count = 7;
+  uint64 total_key_count = 8;
   uint64 min_epoch = 9;
   uint64 max_epoch = 10;
   uint64 uncompressed_file_size = 11;
@@ -64,6 +63,7 @@ message GroupConstruct {
   uint64 parent_group_id = 2;
   repeated uint32 table_ids = 3;
   uint64 group_id = 4;
+  uint64 new_sst_start_id = 5;
 }
 
 message GroupMetaChange {

src/ctl/src/cmd_impl/hummock/sst_dump.rs

@@ -26,6 +26,7 @@ use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_frontend::TableCatalog;
 use risingwave_hummock_sdk::compaction_group::hummock_version_ext::HummockVersionExt;
 use risingwave_hummock_sdk::key::FullKey;
+use risingwave_hummock_sdk::HummockSstableId;
 use risingwave_object_store::object::BlockLocation;
 use risingwave_pb::hummock::{Level, SstableInfo};
 use risingwave_rpc_client::MetaClient;
@@ -73,14 +74,14 @@ pub async fn sst_dump(context: &CtlContext, args: SstDumpArgs) -> anyhow::Result
     for level in version.get_combined_levels() {
         for sstable_info in &level.table_infos {
             if let Some(sst_id) = &args.sst_id {
-                if *sst_id == sstable_info.id {
+                if *sst_id == sstable_info.get_object_id() {
                     if args.print_level {
                         print_level(level);
                     }
 
                     sst_dump_via_sstable_store(
                         sstable_store,
-                        sstable_info.id,
+                        sstable_info.get_object_id(),
                         sstable_info.meta_offset,
                         sstable_info.file_size,
                         &table_data,
@@ -96,7 +97,7 @@ pub async fn sst_dump(context: &CtlContext, args: SstDumpArgs) -> anyhow::Result
 
                 sst_dump_via_sstable_store(
                     sstable_store,
-                    sstable_info.id,
+                    sstable_info.get_object_id(),
                     sstable_info.meta_offset,
                     sstable_info.file_size,
                     &table_data,
@@ -111,16 +112,16 @@ pub async fn sst_dump(context: &CtlContext, args: SstDumpArgs) -> anyhow::Result
 
 pub async fn sst_dump_via_sstable_store(
     sstable_store: &SstableStore,
-    sst_id: u64,
+    object_id: HummockSstableId,
     meta_offset: u64,
     file_size: u64,
     table_data: &TableData,
     args: &SstDumpArgs,
 ) -> anyhow::Result<()> {
     let sstable_info = SstableInfo {
-        id: sst_id,
-        meta_offset,
+        object_id,
         file_size,
+        meta_offset,
         ..Default::default()
     };
     let sstable_cache = sstable_store
@@ -129,7 +130,7 @@ pub async fn sst_dump_via_sstable_store(
     let sstable = sstable_cache.value().as_ref();
     let sstable_meta = &sstable.meta;
 
-    println!("SST id: {}", sst_id);
+    println!("SST object id: {}", object_id);
     println!("-------------------------------------");
     println!("File Size: {}", sstable.estimate_size());
 

src/meta/src/barrier/mod.rs

@@ -1022,7 +1022,7 @@ fn collect_synced_ssts(
             .iter_mut()
             .map(|grouped| {
                 let sst_info = std::mem::take(&mut grouped.sst).expect("field not None");
-                sst_to_worker.insert(sst_info.id, resp.worker_id);
+                sst_to_worker.insert(sst_info.get_object_id(), resp.worker_id);
                 ExtendedSstableInfo::new(
                     grouped.compaction_group_id,
                     sst_info,

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

@@ -540,7 +540,7 @@ pub mod tests {
             level_type: LevelType::Overlapping as i32,
             total_file_size: sst.file_size,
             uncompressed_file_size: sst.uncompressed_file_size,
-            sub_level_id: sst.id,
+            sub_level_id: sst.get_sst_id(),
             table_infos: vec![sst],
         });
     }
@@ -563,7 +563,7 @@ pub mod tests {
             .iter()
             .map(|table| table.uncompressed_file_size)
             .sum();
-        let sub_level_id = table_infos[0].id;
+        let sub_level_id = table_infos[0].get_sst_id();
         levels.l0.as_mut().unwrap().total_file_size += total_file_size;
         levels.l0.as_mut().unwrap().sub_levels.push(Level {
             level_idx: 0,
@@ -583,7 +583,8 @@ pub mod tests {
         epoch: u64,
     ) -> SstableInfo {
         SstableInfo {
-            id,
+            object_id: id,
+            sst_id: id,
             key_range: Some(KeyRange {
                 left: iterator_test_key_of_epoch(table_prefix, left, epoch),
                 right: iterator_test_key_of_epoch(table_prefix, right, epoch),
@@ -594,7 +595,6 @@ pub mod tests {
             meta_offset: 0,
             stale_key_count: 0,
             total_key_count: 0,
-            divide_version: 0,
             uncompressed_file_size: (right - left + 1) as u64,
             min_epoch: 0,
             max_epoch: 0,
@@ -613,7 +613,8 @@ pub mod tests {
         max_epoch: u64,
     ) -> SstableInfo {
         SstableInfo {
-            id,
+            object_id: id,
+            sst_id: id,
             key_range: Some(KeyRange {
                 left: iterator_test_key_of_epoch(table_prefix, left, epoch),
                 right: iterator_test_key_of_epoch(table_prefix, right, epoch),
@@ -624,7 +625,6 @@ pub mod tests {
             meta_offset: 0,
             stale_key_count: 0,
             total_key_count: 0,
-            divide_version: 0,
             uncompressed_file_size: (right - left + 1) as u64,
             min_epoch,
             max_epoch,
@@ -730,7 +730,7 @@ pub mod tests {
     ) {
         for i in &compact_task.input.input_levels {
             for t in &i.table_infos {
-                assert!(level_handlers[i.level_idx as usize].is_pending_compact(&t.id));
+                assert!(level_handlers[i.level_idx as usize].is_pending_compact(&t.sst_id));
             }
         }
     }

src/meta/src/hummock/compaction/mod.rs

@@ -16,6 +16,7 @@ pub mod compaction_config;
 mod level_selector;
 mod overlap_strategy;
 use risingwave_common::catalog::TableOption;
+use risingwave_hummock_sdk::compaction_group::StateTableId;
 use risingwave_hummock_sdk::prost_key_range::KeyRangeExt;
 use risingwave_pb::hummock::compact_task::{self, TaskStatus};
 
@@ -27,7 +28,9 @@ use std::sync::Arc;
 use picker::{
     LevelCompactionPicker, ManualCompactionPicker, MinOverlappingPicker, TierCompactionPicker,
 };
-use risingwave_hummock_sdk::{CompactionGroupId, HummockCompactionTaskId, HummockEpoch};
+use risingwave_hummock_sdk::{
+    CompactionGroupId, HummockCompactionTaskId, HummockEpoch, HummockSstableId,
+};
 use risingwave_pb::hummock::compaction_config::CompactionMode;
 use risingwave_pb::hummock::hummock_version::Levels;
 use risingwave_pb::hummock::{CompactTask, CompactionConfig, InputLevel, KeyRange, LevelType};
@@ -218,11 +221,11 @@ impl CompactStatus {
 #[derive(Clone, Debug, PartialEq)]
 pub struct ManualCompactionOption {
     /// Filters out SSTs to pick. Has no effect if empty.
-    pub sst_ids: Vec<u64>,
+    pub sst_ids: Vec<HummockSstableId>,
     /// Filters out SSTs to pick.
     pub key_range: KeyRange,
     /// Filters out SSTs to pick. Has no effect if empty.
-    pub internal_table_id: HashSet<u32>,
+    pub internal_table_id: HashSet<StateTableId>,
     /// Input level.
     pub level: usize,
 }