refactor(storage): use Arc<SStableInfo> in HummockVersion (#19535)

(cherry picked from commit 4b1cb92)

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

@@ -29,7 +29,7 @@ use risingwave_common::util::value_encoding::{BasicSerde, EitherSerde, ValueRowD
 use risingwave_frontend::TableCatalog;
 use risingwave_hummock_sdk::key::FullKey;
 use risingwave_hummock_sdk::level::Level;
-use risingwave_hummock_sdk::sstable_info::SstableInfo;
+use risingwave_hummock_sdk::sstable_info::{SstableInfo, SstableInfoInner};
 use risingwave_hummock_sdk::HummockSstableObjectId;
 use risingwave_object_store::object::{ObjectMetadata, ObjectStoreImpl};
 use risingwave_rpc_client::MetaClient;
@@ -200,12 +200,13 @@ pub async fn sst_dump_via_sstable_store(
     table_data: &TableData,
     args: &SstDumpArgs,
 ) -> anyhow::Result<()> {
-    let sstable_info = SstableInfo {
+    let sstable_info = SstableInfoInner {
         object_id,
         file_size,
         meta_offset,
         ..Default::default()
-    };
+    }
+    .into();
     let sstable_cache = sstable_store
         .sstable(&sstable_info, &mut StoreLocalStatistic::default())
         .await?;

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

@@ -110,7 +110,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.clone();
                 sstables.push(RwHummockSstable {
                     sstable_id: sst.sst_id as _,
                     object_id: sst.object_id as _,

src/meta/src/hummock/compaction/picker/manual_compaction_picker.rs

@@ -17,7 +17,7 @@ use std::sync::Arc;
 
 use itertools::Itertools;
 use risingwave_hummock_sdk::level::{InputLevel, Level, Levels, OverlappingLevel};
-use risingwave_hummock_sdk::sstable_info::SstableInfo;
+use risingwave_hummock_sdk::sstable_info::{SstableInfo, SstableInfoInner};
 use risingwave_pb::hummock::LevelType;
 
 use super::{CompactionInput, CompactionPicker, LocalPickerStatistic};
@@ -180,10 +180,11 @@ impl ManualCompactionPicker {
     fn filter_level_by_option(&self, level: &Level) -> bool {
         let mut hint_sst_ids: HashSet<u64> = HashSet::new();
         hint_sst_ids.extend(self.option.sst_ids.iter());
-        let tmp_sst_info = SstableInfo {
+        let tmp_sst_info = SstableInfoInner {
             key_range: self.option.key_range.clone(),
             ..Default::default()
-        };
+        }
+        .into();
         if self
             .overlap_strategy
             .check_overlap_with_tables(&[tmp_sst_info], &level.table_infos)
@@ -231,10 +232,10 @@ impl CompactionPicker for ManualCompactionPicker {
         }
         let mut hint_sst_ids: HashSet<u64> = HashSet::new();
         hint_sst_ids.extend(self.option.sst_ids.iter());
-        let mut tmp_sst_info = SstableInfo::default();
+        let mut tmp_sst_info = SstableInfoInner::default();
         let mut range_overlap_info = RangeOverlapInfo::default();
         tmp_sst_info.key_range = self.option.key_range.clone();
-        range_overlap_info.update(&tmp_sst_info);
+        range_overlap_info.update(&tmp_sst_info.into());
         let level = self.option.level;
         let target_level = self.target_level;
         assert!(
@@ -465,9 +466,11 @@ pub mod tests {
 
             let level_table_info = &mut levels.levels[0].table_infos;
             let table_info_1 = &mut level_table_info[1];
-            table_info_1.table_ids.resize(2, 0);
-            table_info_1.table_ids[0] = 1;
-            table_info_1.table_ids[1] = 2;
+            let mut t_inner = table_info_1.get_inner();
+            t_inner.table_ids.resize(2, 0);
+            t_inner.table_ids[0] = 1;
+            t_inner.table_ids[1] = 2;
+            *table_info_1 = t_inner.into();
 
             // test internal_table_id
             let option = ManualCompactionOption {
@@ -499,9 +502,11 @@ pub mod tests {
             // include all table_info
             let level_table_info = &mut levels.levels[0].table_infos;
             for table_info in level_table_info {
-                table_info.table_ids.resize(2, 0);
-                table_info.table_ids[0] = 1;
-                table_info.table_ids[1] = 2;
+                let mut t_inner = table_info.get_inner();
+                t_inner.table_ids.resize(2, 0);
+                t_inner.table_ids[0] = 1;
+                t_inner.table_ids[1] = 2;
+                *table_info = t_inner.into();
             }
 
             // test key range filter first
@@ -575,12 +580,14 @@ pub mod tests {
         for iter in [l0.sub_levels.iter_mut(), levels.iter_mut()] {
             for (idx, l) in iter.enumerate() {
                 for t in &mut l.table_infos {
-                    t.table_ids.clear();
+                    let mut t_inner = t.get_inner();
+                    t_inner.table_ids.clear();
                     if idx == 0 {
-                        t.table_ids.push(((t.sst_id % 2) + 1) as _);
+                        t_inner.table_ids.push(((t.sst_id % 2) + 1) as _);
                     } else {
-                        t.table_ids.push(3);
+                        t_inner.table_ids.push(3);
                     }
+                    *t = t_inner.into();
                 }
             }
         }

src/meta/src/hummock/compaction/picker/space_reclaim_compaction_picker.rs

@@ -173,7 +173,9 @@ mod test {
 
     use itertools::Itertools;
     use risingwave_common::catalog::TableId;
+    use risingwave_hummock_sdk::key_range::KeyRange;
     use risingwave_hummock_sdk::level::Level;
+    use risingwave_hummock_sdk::sstable_info::SstableInfoInner;
     use risingwave_hummock_sdk::version::HummockVersionStateTableInfo;
     use risingwave_pb::hummock::compact_task;
     pub use risingwave_pb::hummock::LevelType;
@@ -234,7 +236,14 @@ mod test {
         {
             let sst_10 = levels[3].table_infos.get_mut(8).unwrap();
             assert_eq!(10, sst_10.sst_id);
-            sst_10.key_range.right_exclusive = true;
+            *sst_10 = SstableInfoInner {
+                key_range: KeyRange {
+                    right_exclusive: true,
+                    ..sst_10.get_inner().key_range.clone()
+                },
+                ..sst_10.get_inner()
+            }
+            .into();
         }
 
         assert_eq!(levels.len(), 4);

src/meta/src/hummock/compaction/picker/tombstone_reclaim_compaction_picker.rs

@@ -135,7 +135,9 @@ pub mod tests {
     use super::*;
     use crate::hummock::compaction::compaction_config::CompactionConfigBuilder;
     use crate::hummock::compaction::create_overlap_strategy;
-    use crate::hummock::compaction::selector::tests::{generate_level, generate_table};
+    use crate::hummock::compaction::selector::tests::{
+        generate_level, generate_table, generate_table_impl,
+    };
 
     #[test]
     fn test_basic() {
@@ -165,21 +167,21 @@ pub mod tests {
         let picker = TombstoneReclaimCompactionPicker::new(strategy.clone(), 40, 20);
         let ret = picker.pick_compaction(&levels, &levels_handler, &mut state);
         assert!(ret.is_none());
-        let mut sst = generate_table(3, 1, 201, 300, 1);
+        let mut sst = generate_table_impl(3, 1, 201, 300, 1);
         sst.stale_key_count = 40;
         sst.total_key_count = 100;
-        levels.levels[1].table_infos.push(sst);
+        levels.levels[1].table_infos.push(sst.into());
 
         let ret = picker
             .pick_compaction(&levels, &levels_handler, &mut state)
             .unwrap();
         assert_eq!(2, ret.input_levels.len());
         assert_eq!(3, ret.input_levels[0].table_infos[0].sst_id);
-        let mut sst = generate_table(4, 1, 1, 100, 1);
+        let mut sst = generate_table_impl(4, 1, 1, 100, 1);
         sst.stale_key_count = 30;
         sst.range_tombstone_count = 30;
         sst.total_key_count = 100;
-        levels.levels[0].table_infos.push(sst);
+        levels.levels[0].table_infos.push(sst.into());
         let picker = TombstoneReclaimCompactionPicker::new(strategy, 50, 10);
         let mut state = TombstoneReclaimPickerState::default();
         let ret = picker

src/meta/src/hummock/compaction/picker/trivial_move_compaction_picker.rs

@@ -117,20 +117,21 @@ impl TrivialMovePicker {
 pub mod tests {
     use std::sync::Arc;
 
-    use risingwave_hummock_sdk::sstable_info::SstableInfo;
+    use risingwave_hummock_sdk::sstable_info::{SstableInfo, SstableInfoInner};
 
     use crate::hummock::compaction::compaction_config::CompactionConfigBuilder;
     use crate::hummock::compaction::create_overlap_strategy;
     use crate::hummock::level_handler::LevelHandler;
 
     #[test]
     fn test_allowed_trivial_move_min_size() {
-        let sst = SstableInfo {
+        let sst: SstableInfo = SstableInfoInner {
             sst_id: 1,
             file_size: 100,
             sst_size: 100,
             ..Default::default()
-        };
+        }
+        .into();
 
         let config = Arc::new(
             CompactionConfigBuilder::new()

src/meta/src/hummock/compaction/picker/ttl_reclaim_compaction_picker.rs

@@ -203,6 +203,7 @@ mod test {
 
     use itertools::Itertools;
     use risingwave_hummock_sdk::level::Level;
+    use risingwave_hummock_sdk::sstable_info::SstableInfoInner;
     use risingwave_hummock_sdk::version::HummockVersionStateTableInfo;
     use risingwave_pb::hummock::compact_task;
     pub use risingwave_pb::hummock::LevelType;
@@ -349,7 +350,14 @@ mod test {
         {
             let sst_10 = levels[3].table_infos.get_mut(8).unwrap();
             assert_eq!(10, sst_10.sst_id);
-            sst_10.key_range.right_exclusive = true;
+            *sst_10 = SstableInfoInner {
+                key_range: KeyRange {
+                    right_exclusive: true,
+                    ..sst_10.key_range.clone()
+                },
+                ..sst_10.get_inner()
+            }
+            .into();
         }
 
         assert_eq!(levels.len(), 4);

src/meta/src/hummock/compaction/picker/vnode_watermark_picker.rs

@@ -22,6 +22,7 @@ use risingwave_hummock_sdk::table_watermark::ReadTableWatermark;
 
 use crate::hummock::compaction::picker::CompactionInput;
 use crate::hummock::level_handler::LevelHandler;
+
 pub struct VnodeWatermarkCompactionPicker {}
 
 impl VnodeWatermarkCompactionPicker {
@@ -109,7 +110,7 @@ mod tests {
     use risingwave_common::hash::VirtualNode;
     use risingwave_hummock_sdk::key::{FullKey, TableKey};
     use risingwave_hummock_sdk::key_range::KeyRange;
-    use risingwave_hummock_sdk::sstable_info::SstableInfo;
+    use risingwave_hummock_sdk::sstable_info::SstableInfoInner;
     use risingwave_hummock_sdk::table_watermark::{ReadTableWatermark, WatermarkDirection};
 
     use crate::hummock::compaction::picker::vnode_watermark_picker::should_delete_sst_by_watermark;
@@ -132,7 +133,7 @@ mod tests {
             TableKey(builder.freeze())
         };
 
-        let sst_info = SstableInfo {
+        let sst_info = SstableInfoInner {
             object_id: 1,
             sst_id: 1,
             key_range: KeyRange {
@@ -146,13 +147,14 @@ mod tests {
             },
             table_ids: vec![2],
             ..Default::default()
-        };
+        }
+        .into();
         assert!(
             !should_delete_sst_by_watermark(&sst_info, &table_watermarks),
             "should fail because no matching watermark found"
         );
 
-        let sst_info = SstableInfo {
+        let sst_info = SstableInfoInner {
             object_id: 1,
             sst_id: 1,
             key_range: KeyRange {
@@ -166,13 +168,14 @@ mod tests {
             },
             table_ids: vec![1],
             ..Default::default()
-        };
+        }
+        .into();
         assert!(
             !should_delete_sst_by_watermark(&sst_info, &table_watermarks),
             "should fail because no matching vnode found"
         );
 
-        let sst_info = SstableInfo {
+        let sst_info = SstableInfoInner {
             object_id: 1,
             sst_id: 1,
             key_range: KeyRange {
@@ -186,13 +189,14 @@ mod tests {
             },
             table_ids: vec![1],
             ..Default::default()
-        };
+        }
+        .into();
         assert!(
             !should_delete_sst_by_watermark(&sst_info, &table_watermarks),
             "should fail because different vnodes found"
         );
 
-        let sst_info = SstableInfo {
+        let sst_info = SstableInfoInner {
             object_id: 1,
             sst_id: 1,
             key_range: KeyRange {
@@ -206,13 +210,14 @@ mod tests {
             },
             table_ids: vec![1],
             ..Default::default()
-        };
+        }
+        .into();
         assert!(
             !should_delete_sst_by_watermark(&sst_info, &table_watermarks),
             "should fail because right key is greater than watermark"
         );
 
-        let sst_info = SstableInfo {
+        let sst_info = SstableInfoInner {
             object_id: 1,
             sst_id: 1,
             key_range: KeyRange {
@@ -226,7 +231,8 @@ mod tests {
             },
             table_ids: vec![1],
             ..Default::default()
-        };
+        }
+        .into();
         assert!(should_delete_sst_by_watermark(&sst_info, &table_watermarks));
     }
 }

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

@@ -129,7 +129,7 @@ pub mod tests {
     use itertools::Itertools;
     use risingwave_hummock_sdk::key_range::KeyRange;
     use risingwave_hummock_sdk::level::{Level, OverlappingLevel};
-    use risingwave_hummock_sdk::sstable_info::SstableInfo;
+    use risingwave_hummock_sdk::sstable_info::{SstableInfo, SstableInfoInner};
     use risingwave_pb::hummock::LevelType;
 
     use super::*;
@@ -179,8 +179,18 @@ pub mod tests {
         right: usize,
         epoch: u64,
     ) -> SstableInfo {
+        generate_table_impl(id, table_prefix, left, right, epoch).into()
+    }
+
+    pub fn generate_table_impl(
+        id: u64,
+        table_prefix: u64,
+        left: usize,
+        right: usize,
+        epoch: u64,
+    ) -> SstableInfoInner {
         let object_size = (right - left + 1) as u64;
-        SstableInfo {
+        SstableInfoInner {
             object_id: id,
             sst_id: id,
             key_range: KeyRange {
@@ -209,7 +219,7 @@ pub mod tests {
         max_epoch: u64,
     ) -> SstableInfo {
         let object_size = (right - left + 1) as u64;
-        SstableInfo {
+        SstableInfoInner {
             object_id: id,
             sst_id: id,
             key_range: KeyRange {
@@ -225,6 +235,7 @@ pub mod tests {
             sst_size: object_size,
             ..Default::default()
         }
+        .into()
     }
 
     pub fn generate_tables(
@@ -237,10 +248,10 @@ pub mod tests {
         let mut start = keys.start;
         let mut tables = vec![];
         for id in ids {
-            let mut table = generate_table(id, 1, start, start + step - 1, epoch);
+            let mut table = generate_table_impl(id, 1, start, start + step - 1, epoch);
             table.file_size = file_size;
             table.sst_size = file_size;
-            tables.push(table);
+            tables.push(table.into());
             start += step;
         }
         tables