Repository: risingwavelabs/risingwave (language: python)

Issue:
Sqlsmith: Sql feature generation
As we add more features, these should be tested via sqlsmith as well.

SQL Features:
- [ ] Generate other table functions: `generate_series`, `unnest`. See: https://github.com/singularity-data/risingwave/issues/4540
- [ ] Generate array_agg, jsonb_agg, jsonb_object_agg
- [ ] Generate select from System Tables
- [ ] Generate `NATURAL JOIN`
- [ ] Generate `JOIN` with `USING`
- [ ] Generate query w/o `GROUP BY`
- [ ] Randomly toggle session variables
- [ ] Generate `pow`/`exp`.
- [ ] Generate TopN by group: https://www.risingwave.dev/docs/current/sql-pattern-topn/
- [ ] Generate jsonb exprs, see: https://github.com/risingwavelabs/risingwave/issues/7714
- [ ] #7132
- [x] Generate order-by-clauses with `NULLS { FIRST | LAST }` (will be done in https://github.com/risingwavelabs/risingwave/pull/10864/commits/000cbc11cbc04f20abfd127b1bf4b3fd24e75c44)
- [ ] Generate `max`,`min`,`count` for `timestamptz`
- [ ] Generate `bit_and`, `bit_or` and `big_xor` in batch queries, and `bit_xor` in streaming queries.
- [ ] Generate array range access.
- [ ] https://github.com/risingwavelabs/risingwave/issues/10241
- [ ] Generate window functions other than `row_number`, `rank`.
- [ ] Generate `ALTER` DDL statements.\
- [ ] https://github.com/risingwavelabs/risingwave/issues/10516
- [ ] Generate temporal filter in sqlsmith


Discussion / hints:

Fix patch:
diff --git a/src/ctl/src/cmd_impl/hummock/sst_dump.rs b/src/ctl/src/cmd_impl/hummock/sst_dump.rs
index 6a36b8bd86833..3594858ac2111 100644
--- a/src/ctl/src/cmd_impl/hummock/sst_dump.rs
+++ b/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?;
diff --git a/src/frontend/src/catalog/system_catalog/rw_catalog/rw_hummock_version.rs b/src/frontend/src/catalog/system_catalog/rw_catalog/rw_hummock_version.rs
index 6fb22ebf65090..77101b88fbeaf 100644
--- a/src/frontend/src/catalog/system_catalog/rw_catalog/rw_hummock_version.rs
+++ b/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 _,
diff --git a/src/meta/src/hummock/compaction/picker/manual_compaction_picker.rs b/src/meta/src/hummock/compaction/picker/manual_compaction_picker.rs
index 597d18cce4fd0..5fc4d19242ae5 100644
--- a/src/meta/src/hummock/compaction/picker/manual_compaction_picker.rs
+++ b/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();
                 }
             }
         }
diff --git a/src/meta/src/hummock/compaction/picker/space_reclaim_compaction_picker.rs b/src/meta/src/hummock/compaction/picker/space_reclaim_compaction_picker.rs
index b3e9eb75321a0..8d7f08ca7f5c1 100644
--- a/src/meta/src/hummock/compaction/picker/space_reclaim_compaction_picker.rs
+++ b/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);
diff --git a/src/meta/src/hummock/compaction/picker/tombstone_reclaim_compaction_picker.rs b/src/meta/src/hummock/compaction/picker/tombstone_reclaim_compaction_picker.rs
index b4ab933931bd7..a1dc751606f6f 100644
--- a/src/meta/src/hummock/compaction/picker/tombstone_reclaim_compaction_picker.rs
+++ b/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
diff --git a/src/meta/src/hummock/compaction/picker/trivial_move_compaction_picker.rs b/src/meta/src/hummock/compaction/picker/trivial_move_compaction_picker.rs
index 458919cd8b717..5d4b8ad9eb8d7 100644
--- a/src/meta/src/hummock/compaction/picker/trivial_move_compaction_picker.rs
+++ b/src/meta/src/hummock/compaction/picker/trivial_move_compaction_picker.rs
@@ -117,7 +117,7 @@ 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;
@@ -125,12 +125,13 @@ pub mod tests {
 
     #[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()
diff --git a/src/meta/src/hummock/compaction/picker/ttl_reclaim_compaction_picker.rs b/src/meta/src/hummock/compaction/picker/ttl_reclaim_compaction_picker.rs
index 36ee06a3bb795..97215574b8761 100644
--- a/src/meta/src/hummock/compaction/picker/ttl_reclaim_compaction_picker.rs
+++ b/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);
diff --git a/src/meta/src/hummock/compaction/picker/vnode_watermark_picker.rs b/src/meta/src/hummock/compaction/picker/vnode_watermark_picker.rs
index 5171c48ad9c34..26626e48a2423 100644
--- a/src/meta/src/hummock/compaction/picker/vnode_watermark_picker.rs
+++ b/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));
     }
 }
diff --git a/src/meta/src/hummock/compaction/selector/mod.rs b/src/meta/src/hummock/compaction/selector/mod.rs
index 02b50ff2ff671..07309ff72ee8f 100644
--- a/src/meta/src/hummock/compaction/selector/mod.rs
+++ b/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
diff --git a/src/meta/src/hummock/manager/commit_epoch.rs b/src/meta/src/hummock/manager/commit_epoch.rs
index 37ff58b25a4d4..4dbb1c71b35d8 100644
--- a/src/meta/src/hummock/manager/commit_epoch.rs
+++ b/src/meta/src/hummock/manager/commit_epoch.rs
@@ -396,13 +396,14 @@ impl HummockManager {
                     .sum();
 
                 // TODO(li0k): replace with `split_sst`
-                let branch_sst = split_sst_with_table_ids(
-                    &mut sst.sst_info,
+                let (modified_sst_info, branch_sst) = split_sst_with_table_ids(
+                    &sst.sst_info,
                     &mut new_sst_id,
                     origin_sst_size - new_sst_size,
                     new_sst_size,
                     match_ids,
                 );
+                sst.sst_info = modified_sst_info;
 
                 commit_sstables
                     .entry(group_id)
diff --git a/src/meta/src/hummock/manager/versioning.rs b/src/meta/src/hummock/manager/versioning.rs
index fd516a2ba3e1e..78d84b829d93d 100644
--- a/src/meta/src/hummock/manager/versioning.rs
+++ b/src/meta/src/hummock/manager/versioning.rs
@@ -333,7 +333,7 @@ mod tests {
 
     use risingwave_hummock_sdk::key_range::KeyRange;
     use risingwave_hummock_sdk::level::{Level, Levels};
-    use risingwave_hummock_sdk::sstable_info::SstableInfo;
+    use risingwave_hummock_sdk::sstable_info::SstableInfoInner;
     use risingwave_hummock_sdk::version::HummockVersion;
     use risingwave_hummock_sdk::{CompactionGroupId, HummockVersionId};
     use risingwave_pb::hummock::write_limits::WriteLimit;
@@ -458,7 +458,7 @@ mod tests {
 
     #[test]
     fn test_estimate_table_stats() {
-        let sst = SstableInfo {
+        let sst = SstableInfoInner {
             key_range: KeyRange {
                 left: vec![1; 10].into(),
                 right: vec![1; 20].into(),
@@ -468,7 +468,8 @@ mod tests {
             total_key_count: 6000,
             uncompressed_file_size: 6_000_000,
             ..Default::default()
-        };
+        }
+        .into();
         let changes = estimate_table_stats(&sst);
         assert_eq!(changes.len(), 3);
         for stats in changes.values() {
@@ -519,7 +520,7 @@ mod tests {
 
     #[test]
     fn test_estimate_table_stats_large_key_range() {
-        let sst = SstableInfo {
+        let sst = SstableInfoInner {
             key_range: KeyRange {
                 left: vec![1; 1000].into(),
                 right: vec![1; 2000].into(),
@@ -529,7 +530,8 @@ mod tests {
             total_key_count: 6000,
             uncompressed_file_size: 60_000,
             ..Default::default()
-        };
+        }
+        .into();
         let changes = estimate_table_stats(&sst);
         assert_eq!(changes.len(), 3);
         for t in &sst.table_ids {
diff --git a/src/storage/hummock_sdk/src/compaction_group/hummock_version_ext.rs b/src/storage/hummock_sdk/src/compaction_group/hummock_version_ext.rs
index ac9521b0f5272..bcf99e18bfe7a 100644
--- a/src/storage/hummock_sdk/src/compaction_group/hummock_version_ext.rs
+++ b/src/storage/hummock_sdk/src/compaction_group/hummock_version_ext.rs
@@ -1175,13 +1175,14 @@ fn split_sst_info_for_level(
             .cloned()
             .collect_vec();
         if !removed_table_ids.is_empty() {
-            let branch_sst = split_sst_with_table_ids(
+            let (modified_sst, branch_sst) = split_sst_with_table_ids(
                 sst_info,
                 new_sst_id,
                 sst_info.sst_size / 2,
                 sst_info.sst_size / 2,
                 member_table_ids.iter().cloned().collect_vec(),
             );
+            *sst_info = modified_sst;
             insert_table_infos.push(branch_sst);
         }
     }
@@ -1511,13 +1512,17 @@ mod tests {
     use crate::key::{gen_key_from_str, FullKey};
     use crate::key_range::KeyRange;
     use crate::level::{Level, Levels, OverlappingLevel};
-    use crate::sstable_info::SstableInfo;
+    use crate::sstable_info::{SstableInfo, SstableInfoInner};
     use crate::version::{
         GroupDelta, GroupDeltas, HummockVersion, HummockVersionDelta, IntraLevelDelta,
     };
     use crate::HummockVersionId;
 
     fn gen_sstable_info(sst_id: u64, table_ids: Vec<u32>, epoch: u64) -> SstableInfo {
+        gen_sstable_info_impl(sst_id, table_ids, epoch).into()
+    }
+
+    fn gen_sstable_info_impl(sst_id: u64, table_ids: Vec<u32>, epoch: u64) -> SstableInfoInner {
         let table_key_l = gen_key_from_str(VirtualNode::ZERO, "1");
         let table_key_r = gen_key_from_str(VirtualNode::MAX_FOR_TEST, "1");
         let full_key_l = FullKey::for_test(
@@ -1530,7 +1535,7 @@ mod tests {
             FullKey::for_test(TableId::new(*table_ids.last().unwrap()), table_key_r, epoch)
                 .encode();
 
-        SstableInfo {
+        SstableInfoInner {
             sst_id,
             key_range: KeyRange {
                 left: full_key_l.into(),
@@ -1575,22 +1580,24 @@ mod tests {
             .l0
             .sub_levels
             .push(Level {
-                table_infos: vec![SstableInfo {
+                table_infos: vec![SstableInfoInner {
                     object_id: 11,
                     sst_id: 11,
                     ..Default::default()
-                }],
+                }
+                .into()],
                 ..Default::default()
             });
         assert_eq!(version.get_object_ids().len(), 1);
 
         // Add to non sub level
         version.levels.get_mut(&0).unwrap().levels.push(Level {
-            table_infos: vec![SstableInfo {
+            table_infos: vec![SstableInfoInner {
                 object_id: 22,
                 sst_id: 22,
                 ..Default::default()
-            }],
+            }
+            .into()],
             ..Default::default()
         });
         assert_eq!(version.get_object_ids().len(), 2);
@@ -1652,11 +1659,12 @@ mod tests {
                             1,
                             0,
                             HashSet::new(),
-                            vec![SstableInfo {
+                            vec![SstableInfoInner {
                                 object_id: 1,
                                 sst_id: 1,
                                 ..Default::default()
-                            }],
+                            }
+                            .into()],
                             0,
                         ))],
                     },
@@ -1677,11 +1685,12 @@ mod tests {
         cg1.levels[0] = Level {
             level_idx: 1,
             level_type: LevelType::Nonoverlapping,
-            table_infos: vec![SstableInfo {
+            table_infos: vec![SstableInfoInner {
                 object_id: 1,
                 sst_id: 1,
                 ..Default::default()
-            }],
+            }
+            .into()],
             ..Default::default()
         };
         assert_eq!(
@@ -1707,7 +1716,16 @@ mod tests {
     }
 
     fn gen_sst_info(object_id: u64, table_ids: Vec<u32>, left: Bytes, right: Bytes) -> SstableInfo {
-        SstableInfo {
+        gen_sst_info_impl(object_id, table_ids, left, right).into()
+    }
+
+    fn gen_sst_info_impl(
+        object_id: u64,
+        table_ids: Vec<u32>,
+        left: Bytes,
+        right: Bytes,
+    ) -> SstableInfoInner {
+        SstableInfoInner {
             object_id,
             sst_id: object_id,
             key_range: KeyRange {
@@ -2214,8 +2232,9 @@ mod tests {
 
         {
             // test key_range left = right
-            let mut sst = gen_sstable_info(1, vec![1], epoch);
+            let mut sst = gen_sstable_info_impl(1, vec![1], epoch);
             sst.key_range.right = sst.key_range.left.clone();
+            let sst: SstableInfo = sst.into();
             let split_key = group_split::build_split_key(1, VirtualNode::ZERO);
             let origin_sst = sst.clone();
             let sst_size = origin_sst.sst_size;
diff --git a/src/storage/hummock_sdk/src/compaction_group/mod.rs b/src/storage/hummock_sdk/src/compaction_group/mod.rs
index dbe58018f15e3..62ced96501723 100644
--- a/src/storage/hummock_sdk/src/compaction_group/mod.rs
+++ b/src/storage/hummock_sdk/src/compaction_group/mod.rs
@@ -135,12 +135,13 @@ pub mod group_split {
     ///    `sst_size`: `right_size`,
     /// }
     pub fn split_sst(
-        mut origin_sst_info: SstableInfo,
+        origin_sst_info: SstableInfo,
         new_sst_id: &mut u64,
         split_key: Bytes,
         left_size: u64,
         right_size: u64,
     ) -> (Option<SstableInfo>, Option<SstableInfo>) {
+        let mut origin_sst_info = origin_sst_info.get_inner();
         let mut branch_table_info = origin_sst_info.clone();
         branch_table_info.sst_id = *new_sst_id;
         *new_sst_id += 1;
@@ -183,9 +184,9 @@ pub mod group_split {
 
         // This function does not make any assumptions about the incoming sst, so add some judgement to ensure that the generated sst meets the restrictions.
         if origin_sst_info.table_ids.is_empty() {
-            (None, Some(branch_table_info))
+            (None, Some(branch_table_info.into()))
         } else if branch_table_info.table_ids.is_empty() {
-            (Some(origin_sst_info), None)
+            (Some(origin_sst_info.into()), None)
         } else if KeyComparator::compare_encoded_full_key(
             &origin_sst_info.key_range.left,
             &origin_sst_info.key_range.right,
@@ -193,9 +194,9 @@ pub mod group_split {
         .is_eq()
         {
             // avoid empty key_range of origin_sst
-            (None, Some(branch_table_info))
+            (None, Some(branch_table_info.into()))
         } else {
-            (Some(origin_sst_info), Some(branch_table_info))
+            (Some(origin_sst_info.into()), Some(branch_table_info.into()))
         }
     }
 
@@ -203,12 +204,13 @@ pub mod group_split {
     /// This function is used to split the sst into two parts based on the `table_ids`.
     /// In contrast to `split_sst`, this function does not modify the `key_range` and does not guarantee that the split ssts can be merged, which needs to be guaranteed by the caller.
     pub fn split_sst_with_table_ids(
-        sst_info: &mut SstableInfo,
+        origin_sst_info: &SstableInfo,
         new_sst_id: &mut u64,
         old_sst_size: u64,
         new_sst_size: u64,
         new_table_ids: Vec<u32>,
-    ) -> SstableInfo {
+    ) -> (SstableInfo, SstableInfo) {
+        let mut sst_info = origin_sst_info.get_inner();
         let mut branch_table_info = sst_info.clone();
         branch_table_info.sst_id = *new_sst_id;
         branch_table_info.sst_size = new_sst_size;
@@ -235,7 +237,7 @@ pub mod group_split {
                 .retain(|table_id| !branch_table_info.table_ids.contains(table_id));
         }
 
-        branch_table_info
+        (sst_info.into(), branch_table_info.into())
     }
 
     // Should avoid split same table_id into two groups
diff --git a/src/storage/hummock_sdk/src/sstable_info.rs b/src/storage/hummock_sdk/src/sstable_info.rs
index 10afe52ab9ad9..eb43b1c61b750 100644
--- a/src/storage/hummock_sdk/src/sstable_info.rs
+++ b/src/storage/hummock_sdk/src/sstable_info.rs
@@ -13,6 +13,8 @@
 // limitations under the License.
 
 use std::mem::size_of;
+use std::ops::Deref;
+use std::sync::Arc;
 
 use risingwave_pb::hummock::{PbBloomFilterType, PbKeyRange, PbSstableInfo};
 
@@ -21,7 +23,7 @@ use crate::version::{ObjectIdReader, SstableIdReader};
 use crate::{HummockSstableId, HummockSstableObjectId};
 
 #[derive(Debug, PartialEq, Clone, Default)]
-pub struct SstableInfo {
+pub struct SstableInfoInner {
     pub object_id: u64,
     pub sst_id: u64,
     pub key_range: KeyRange,
@@ -38,7 +40,7 @@ pub struct SstableInfo {
     pub sst_size: u64,
 }
 
-impl SstableInfo {
+impl SstableInfoInner {
     pub fn estimated_encode_len(&self) -> usize {
         let mut basic = size_of::<u64>() // object_id
             + size_of::<u64>() // sstable_id
@@ -63,7 +65,7 @@ impl SstableInfo {
     }
 }
 
-impl From<PbSstableInfo> for SstableInfo {
+impl From<PbSstableInfo> for SstableInfoInner {
     fn from(pb_sstable_info: PbSstableInfo) -> Self {
         assert!(pb_sstable_info.table_ids.is_sorted());
         Self {
@@ -101,7 +103,7 @@ impl From<PbSstableInfo> for SstableInfo {
     }
 }
 
-impl From<&PbSstableInfo> for SstableInfo {
+impl From<&PbSstableInfo> for SstableInfoInner {
     fn from(pb_sstable_info: &PbSstableInfo) -> Self {
         assert!(pb_sstable_info.table_ids.is_sorted());
         Self {
@@ -138,8 +140,8 @@ impl From<&PbSstableInfo> for SstableInfo {
     }
 }
 
-impl From<SstableInfo> for PbSstableInfo {
-    fn from(sstable_info: SstableInfo) -> Self {
+impl From<SstableInfoInner> for PbSstableInfo {
+    fn from(sstable_info: SstableInfoInner) -> Self {
         assert!(sstable_info.sst_size > 0);
         assert!(sstable_info.table_ids.is_sorted());
         PbSstableInfo {
@@ -177,8 +179,8 @@ impl From<SstableInfo> for PbSstableInfo {
     }
 }
 
-impl From<&SstableInfo> for PbSstableInfo {
-    fn from(sstable_info: &SstableInfo) -> Self {
+impl From<&SstableInfoInner> for PbSstableInfo {
+    fn from(sstable_info: &SstableInfoInner) -> Self {
         assert!(sstable_info.sst_size > 0);
         assert!(sstable_info.table_ids.is_sorted());
         PbSstableInfo {
@@ -215,7 +217,68 @@ impl From<&SstableInfo> for PbSstableInfo {
 
 impl SstableInfo {
     pub fn remove_key_range(&mut self) {
-        self.key_range = KeyRange::default();
+        let mut sst = self.get_inner();
+        sst.key_range = KeyRange::default();
+        *self = sst.into()
+    }
+}
+
+impl SstableIdReader for SstableInfoInner {
+    fn sst_id(&self) -> HummockSstableId {
+        self.sst_id
+    }
+}
+
+impl ObjectIdReader for SstableInfoInner {
+    fn object_id(&self) -> HummockSstableObjectId {
+        self.object_id
+    }
+}
+
+#[derive(Debug, PartialEq, Clone, Default)]
+pub struct SstableInfo(Arc<SstableInfoInner>);
+
+impl From<&PbSstableInfo> for SstableInfo {
+    fn from(s: &PbSstableInfo) -> Self {
+        SstableInfo(SstableInfoInner::from(s).into())
+    }
+}
+
+impl From<PbSstableInfo> for SstableInfo {
+    fn from(s: PbSstableInfo) -> Self {
+        SstableInfo(SstableInfoInner::from(s).into())
+    }
+}
+
+impl From<SstableInfo> for PbSstableInfo {
+    fn from(s: SstableInfo) -> Self {
+        (&s).into()
+    }
+}
+
+impl From<SstableInfoInner> for SstableInfo {
+    fn from(s: SstableInfoInner) -> Self {
+        Self(s.into())
+    }
+}
+
+impl From<&SstableInfo> for PbSstableInfo {
+    fn from(s: &SstableInfo) -> Self {
+        s.0.as_ref().into()
+    }
+}
+
+impl Deref for SstableInfo {
+    type Target = SstableInfoInner;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl SstableInfo {
+    pub fn get_inner(&self) -> SstableInfoInner {
+        (*self.0).clone()
     }
 }
 
diff --git a/src/storage/src/hummock/compactor/iterator.rs b/src/storage/src/hummock/compactor/iterator.rs
index c53f945af2c50..b5137ebf4d63f 100644
--- a/src/storage/src/hummock/compactor/iterator.rs
+++ b/src/storage/src/hummock/compactor/iterator.rs
@@ -726,6 +726,7 @@ mod tests {
     use risingwave_common::catalog::TableId;
     use risingwave_hummock_sdk::key::{next_full_key, prev_full_key, FullKey, FullKeyTracker};
     use risingwave_hummock_sdk::key_range::KeyRange;
+    use risingwave_hummock_sdk::sstable_info::{SstableInfo, SstableInfoInner};
 
     use crate::hummock::compactor::ConcatSstableIterator;
     use crate::hummock::iterator::test_utils::mock_sstable_store;
@@ -1198,15 +1199,27 @@ mod tests {
         .await;
 
         let split_key = test_key_of(5000).encode();
-
-        let mut sst_1 = table_info.clone();
-        sst_1.key_range.right = split_key.clone().into();
-        sst_1.key_range.right_exclusive = true;
+        let sst_1: SstableInfo = SstableInfoInner {
+            key_range: KeyRange {
+                left: table_info.key_range.left.clone(),
+                right: split_key.clone().into(),
+                right_exclusive: true,
+            },
+            ..table_info.get_inner()
+        }
+        .into();
 
         let total_key_count = sst_1.total_key_count;
-        let mut sst_2 = table_info.clone();
-        sst_2.sst_id = sst_1.sst_id + 1;
-        sst_2.key_range.left = split_key.clone().into();
+        let sst_2: SstableInfo = SstableInfoInner {
+            sst_id: sst_1.sst_id + 1,
+            key_range: KeyRange {
+                left: split_key.clone().into(),
+                right: table_info.key_range.right.clone(),
+                right_exclusive: table_info.key_range.right_exclusive,
+            },
+            ..table_info.get_inner()
+        }
+        .into();
 
         {
             // test concate
diff --git a/src/storage/src/hummock/sstable/builder.rs b/src/storage/src/hummock/sstable/builder.rs
index 0ad46e5e000a2..123e6d2f8df1f 100644
--- a/src/storage/src/hummock/sstable/builder.rs
+++ b/src/storage/src/hummock/sstable/builder.rs
@@ -20,7 +20,7 @@ use bytes::{Bytes, BytesMut};
 use risingwave_hummock_sdk::compaction_group::StateTableId;
 use risingwave_hummock_sdk::key::{user_key, FullKey, MAX_KEY_LEN};
 use risingwave_hummock_sdk::key_range::KeyRange;
-use risingwave_hummock_sdk::sstable_info::SstableInfo;
+use risingwave_hummock_sdk::sstable_info::{SstableInfo, SstableInfoInner};
 use risingwave_hummock_sdk::table_stats::{TableStats, TableStatsMap};
 use risingwave_hummock_sdk::{HummockEpoch, LocalSstableInfo};
 use risingwave_pb::hummock::BloomFilterType;
@@ -507,7 +507,7 @@ impl<W: SstableWriter, F: FilterBuilder> SstableBuilder<W, F> {
             }
         };
 
-        let sst_info = SstableInfo {
+        let sst_info: SstableInfo = SstableInfoInner {
             object_id: self.sstable_id,
             sst_id: self.sstable_id,
             bloom_filter_kind,
@@ -526,7 +526,8 @@ impl<W: SstableWriter, F: FilterBuilder> SstableBuilder<W, F> {
             max_epoch,
             range_tombstone_count: 0,
             sst_size: meta.estimated_size as u64,
-        };
+        }
+        .into();
         tracing::trace!(
             "meta_size {} bloom_filter_size {}  add_key_counts {} stale_key_count {} min_epoch {} max_epoch {} epoch_count {}",
             meta.encoded_size(),


Test patch:
diff --git a/src/meta/src/hummock/manager/tests.rs b/src/meta/src/hummock/manager/tests.rs
index c96911aea163d..2380a3ade966f 100644
--- a/src/meta/src/hummock/manager/tests.rs
+++ b/src/meta/src/hummock/manager/tests.rs
@@ -29,7 +29,7 @@ use risingwave_hummock_sdk::compaction_group::hummock_version_ext::get_compactio
 use risingwave_hummock_sdk::compaction_group::StaticCompactionGroupId;
 use risingwave_hummock_sdk::key::{gen_key_from_str, FullKey};
 use risingwave_hummock_sdk::key_range::KeyRange;
-use risingwave_hummock_sdk::sstable_info::SstableInfo;
+use risingwave_hummock_sdk::sstable_info::{SstableInfo, SstableInfoInner};
 use risingwave_hummock_sdk::table_stats::{to_prost_table_stats_map, TableStats, TableStatsMap};
 use risingwave_hummock_sdk::version::HummockVersion;
 use risingwave_hummock_sdk::{
@@ -76,6 +76,10 @@ fn pin_versions_sum(pin_versions: &[HummockPinnedVersion]) -> usize {
 }
 
 fn gen_sstable_info(sst_id: u64, table_ids: Vec<u32>, epoch: u64) -> SstableInfo {
+    gen_sstable_info_impl(sst_id, table_ids, epoch).into()
+}
+
+fn gen_sstable_info_impl(sst_id: u64, table_ids: Vec<u32>, epoch: u64) -> SstableInfoInner {
     let table_key_l = gen_key_from_str(VirtualNode::ZERO, "1");
     let table_key_r = gen_key_from_str(VirtualNode::MAX_FOR_TEST, "1");
     let full_key_l = FullKey::for_test(
@@ -87,7 +91,7 @@ fn gen_sstable_info(sst_id: u64, table_ids: Vec<u32>, epoch: u64) -> SstableInfo
     let full_key_r =
         FullKey::for_test(TableId::new(*table_ids.last().unwrap()), table_key_r, epoch).encode();
 
-    SstableInfo {
+    SstableInfoInner {
         sst_id,
         key_range: KeyRange {
             left: full_key_l.into(),
@@ -1287,7 +1291,7 @@ async fn test_version_stats() {
         .into_iter()
         .enumerate()
         .map(|(idx, table_ids)| LocalSstableInfo {
-            sst_info: SstableInfo {
+            sst_info: SstableInfoInner {
                 object_id: sst_ids[idx],
                 sst_id: sst_ids[idx],
                 key_range: KeyRange {
@@ -1299,7 +1303,8 @@ async fn test_version_stats() {
                 table_ids: table_ids.clone(),
                 sst_size: 1024 * 1024 * 1024,
                 ..Default::default()
-            },
+            }
+            .into(),
             table_stats: table_ids
                 .iter()
                 .map(|table_id| (*table_id, table_stats_change.clone()))
@@ -1673,12 +1678,24 @@ async fn test_move_state_tables_to_dedicated_compaction_group_trivial_expired()
         table_stats: Default::default(),
         created_at: u64::MAX,
     };
-    let mut sst_3 = sst_2.clone();
-    let mut sst_4 = sst_1.clone();
-    sst_3.sst_info.sst_id = 8;
-    sst_3.sst_info.object_id = 8;
-    sst_4.sst_info.sst_id = 9;
-    sst_4.sst_info.object_id = 9;
+    let sst_3 = LocalSstableInfo {
+        sst_info: SstableInfoInner {
+            sst_id: 8,
+            object_id: 8,
+            ..sst_2.sst_info.get_inner()
+        }
+        .into(),
+        ..sst_2.clone()
+    };
+    let sst_4 = LocalSstableInfo {
+        sst_info: SstableInfoInner {
+            sst_id: 9,
+            object_id: 9,
+            ..sst_1.sst_info.get_inner()
+        }
+        .into(),
+        ..sst_1.clone()
+    };
     hummock_meta_client
         .commit_epoch(
             30,
@@ -1745,7 +1762,7 @@ async fn test_move_state_tables_to_dedicated_compaction_group_trivial_expired()
         .report_compact_task(
             task2.task_id,
             TaskStatus::Success,
-            vec![SstableInfo {
+            vec![SstableInfoInner {
                 object_id: 12,
                 sst_id: 12,
                 key_range: KeyRange::default(),
@@ -1755,7 +1772,8 @@ async fn test_move_state_tables_to_dedicated_compaction_group_trivial_expired()
                 file_size: 100,
                 sst_size: 100,
                 ..Default::default()
-            }],
+            }
+            .into()],
             None,
             HashMap::default(),
         )
@@ -2222,10 +2240,15 @@ async fn test_partition_level() {
     const MB: u64 = 1024 * 1024;
     let mut selector = default_compaction_selector();
     for epoch in 31..100 {
-        let mut sst = gen_local_sstable_info(global_sst_id, vec![100], test_epoch(epoch));
-        sst.sst_info.file_size = 10 * MB;
-        sst.sst_info.sst_size = 10 * MB;
-        sst.sst_info.uncompressed_file_size = 10 * MB;
+        let mut sst = gen_sstable_info_impl(global_sst_id, vec![100], test_epoch(epoch));
+        sst.file_size = 10 * MB;
+        sst.sst_size = 10 * MB;
+        sst.uncompressed_file_size = 10 * MB;
+        let sst = LocalSstableInfo {
+            sst_info: sst.into(),
+            table_stats: Default::default(),
+            created_at: u64::MAX,
+        };
 
         hummock_meta_client
             .commit_epoch(
@@ -2245,7 +2268,7 @@ async fn test_partition_level() {
             .await
             .unwrap()
         {
-            let mut sst = gen_sstable_info(global_sst_id, vec![100], test_epoch(epoch));
+            let mut sst = gen_sstable_info_impl(global_sst_id, vec![100], test_epoch(epoch));
             sst.file_size = task
                 .input_ssts
                 .iter()
@@ -2263,7 +2286,7 @@ async fn test_partition_level() {
                 .report_compact_task(
                     task.task_id,
                     TaskStatus::Success,
-                    vec![sst],
+                    vec![sst.into()],
                     None,
                     HashMap::default(),
                 )
@@ -2426,12 +2449,24 @@ async fn test_merge_compaction_group_task_expired() {
         table_stats: Default::default(),
         created_at: u64::MAX,
     };
-    let mut sst_3 = sst_2.clone();
-    let mut sst_4 = sst_1.clone();
-    sst_3.sst_info.sst_id = 3;
-    sst_3.sst_info.object_id = 3;
-    sst_4.sst_info.sst_id = 4;
-    sst_4.sst_info.object_id = 4;
+    let sst_3 = LocalSstableInfo {
+        sst_info: SstableInfoInner {
+            sst_id: 3,
+            object_id: 3,
+            ..sst_2.sst_info.get_inner()
+        }
+        .into(),
+        ..sst_2.clone()
+    };
+    let sst_4 = LocalSstableInfo {
+        sst_info: SstableInfoInner {
+            sst_id: 4,
+            object_id: 4,
+            ..sst_1.sst_info.get_inner()
+        }
+        .into(),
+        ..sst_1.clone()
+    };
     hummock_meta_client
         .commit_epoch(
             30,
@@ -2521,7 +2556,7 @@ async fn test_merge_compaction_group_task_expired() {
         .report_compact_task(
             task2.task_id,
             TaskStatus::Success,
-            vec![SstableInfo {
+            vec![SstableInfoInner {
                 object_id: report_sst_id,
                 sst_id: report_sst_id,
                 key_range: KeyRange::default(),
@@ -2531,7 +2566,8 @@ async fn test_merge_compaction_group_task_expired() {
                 file_size: 100,
                 sst_size: 100,
                 ..Default::default()
-            }],
+            }
+            .into()],
             None,
             HashMap::default(),
         )
diff --git a/src/meta/src/hummock/test_utils.rs b/src/meta/src/hummock/test_utils.rs
index 2e750a06aa222..5c5e20f94126e 100644
--- a/src/meta/src/hummock/test_utils.rs
+++ b/src/meta/src/hummock/test_utils.rs
@@ -25,7 +25,7 @@ use risingwave_common::util::epoch::test_epoch;
 use risingwave_hummock_sdk::key::key_with_epoch;
 use risingwave_hummock_sdk::key_range::KeyRange;
 use risingwave_hummock_sdk::level::Levels;
-use risingwave_hummock_sdk::sstable_info::SstableInfo;
+use risingwave_hummock_sdk::sstable_info::{SstableInfo, SstableInfoInner};
 use risingwave_hummock_sdk::table_watermark::TableWatermarks;
 use risingwave_hummock_sdk::version::{HummockVersion, HummockVersionStateTableInfo};
 use risingwave_hummock_sdk::{
@@ -153,31 +153,34 @@ pub fn generate_test_sstables_with_table_id(
     let mut sst_info = vec![];
     for (i, sst_id) in sst_ids.into_iter().enumerate() {
         let object_size = 2;
-        sst_info.push(SstableInfo {
-            object_id: sst_id,
-            sst_id,
-            key_range: KeyRange {
-                left: Bytes::from(key_with_epoch(
-                    format!("{:03}\0\0_key_test_{:05}", table_id, i + 1)
-                        .as_bytes()
-                        .to_vec(),
-                    epoch,
-                )),
-                right: Bytes::from(key_with_epoch(
-                    format!("{:03}\0\0_key_test_{:05}", table_id, (i + 1) * 10)
-                        .as_bytes()
-                        .to_vec(),
-                    epoch,
-                )),
-                right_exclusive: false,
-            },
-            file_size: object_size,
-            table_ids: vec![table_id],
-            uncompressed_file_size: object_size,
-            max_epoch: epoch,
-            sst_size: object_size,
-            ..Default::default()
-        });
+        sst_info.push(
+            SstableInfoInner {
+                object_id: sst_id,
+                sst_id,
+                key_range: KeyRange {
+                    left: Bytes::from(key_with_epoch(
+                        format!("{:03}\0\0_key_test_{:05}", table_id, i + 1)
+                            .as_bytes()
+                            .to_vec(),
+                        epoch,
+                    )),
+                    right: Bytes::from(key_with_epoch(
+                        format!("{:03}\0\0_key_test_{:05}", table_id, (i + 1) * 10)
+                            .as_bytes()
+                            .to_vec(),
+                        epoch,
+                    )),
+                    right_exclusive: false,
+                },
+                file_size: object_size,
+                table_ids: vec![table_id],
+                uncompressed_file_size: object_size,
+                max_epoch: epoch,
+                sst_size: object_size,
+                ..Default::default()
+            }
+            .into(),
+        );
     }
     sst_info
 }
@@ -186,21 +189,24 @@ pub fn generate_test_tables(epoch: u64, sst_ids: Vec<HummockSstableObjectId>) ->
     let mut sst_info = vec![];
     for (i, sst_id) in sst_ids.into_iter().enumerate() {
         let object_size = 2;
-        sst_info.push(SstableInfo {
-            object_id: sst_id,
-            sst_id,
-            key_range: KeyRange {
-                left: Bytes::from(iterator_test_key_of_epoch(sst_id, i + 1, epoch)),
-                right: Bytes::from(iterator_test_key_of_epoch(sst_id, (i + 1) * 10, epoch)),
-                right_exclusive: false,
-            },
-            file_size: object_size,
-            table_ids: vec![sst_id as u32, sst_id as u32 * 10000],
-            uncompressed_file_size: object_size,
-            max_epoch: epoch,
-            sst_size: object_size,
-            ..Default::default()
-        });
+        sst_info.push(
+            SstableInfoInner {
+                object_id: sst_id,
+                sst_id,
+                key_range: KeyRange {
+                    left: Bytes::from(iterator_test_key_of_epoch(sst_id, i + 1, epoch)),
+                    right: Bytes::from(iterator_test_key_of_epoch(sst_id, (i + 1) * 10, epoch)),
+                    right_exclusive: false,
+                },
+                file_size: object_size,
+                table_ids: vec![sst_id as u32, sst_id as u32 * 10000],
+                uncompressed_file_size: object_size,
+                max_epoch: epoch,
+                sst_size: object_size,
+                ..Default::default()
+            }
+            .into(),
+        );
     }
     sst_info
 }
diff --git a/src/storage/hummock_test/src/hummock_read_version_tests.rs b/src/storage/hummock_test/src/hummock_read_version_tests.rs
index e0fdf9011aca6..2e763dfd3954b 100644
--- a/src/storage/hummock_test/src/hummock_read_version_tests.rs
+++ b/src/storage/hummock_test/src/hummock_read_version_tests.rs
@@ -25,7 +25,7 @@ use risingwave_common::hash::VirtualNode;
 use risingwave_common::util::epoch::{test_epoch, EpochExt};
 use risingwave_hummock_sdk::key::{key_with_epoch, map_table_key_range};
 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::LocalSstableInfo;
 use risingwave_meta::hummock::test_utils::setup_compute_env;
 use risingwave_storage::hummock::event_handler::TEST_LOCAL_INSTANCE_ID;
@@ -150,44 +150,62 @@ async fn test_read_version_basic() {
 
         let dummy_sst = Arc::new(StagingSstableInfo::new(
             vec![
-                LocalSstableInfo::for_test(SstableInfo {
-                    object_id: 1,
-                    sst_id: 1,
-                    key_range: KeyRange {
-                        left: key_with_epoch(iterator_test_user_key_of(1).encode(), test_epoch(1))
+                LocalSstableInfo::for_test(
+                    SstableInfoInner {
+                        object_id: 1,
+                        sst_id: 1,
+                        key_range: KeyRange {
+                            left: key_with_epoch(
+                                iterator_test_user_key_of(1).encode(),
+                                test_epoch(1),
+                            )
                             .into(),
-                        right: key_with_epoch(iterator_test_user_key_of(2).encode(), test_epoch(2))
+                            right: key_with_epoch(
+                                iterator_test_user_key_of(2).encode(),
+                                test_epoch(2),
+                            )
                             .into(),
-                        right_exclusive: false,
-                    },
-                    file_size: 1,
-                    table_ids: vec![0],
-                    meta_offset: 1,
-                    stale_key_count: 1,
-                    total_key_count: 1,
-                    uncompressed_file_size: 1,
-                    sst_size: 1,
-                    ..Default::default()
-                }),
-                LocalSstableInfo::for_test(SstableInfo {
-                    object_id: 2,
-                    sst_id: 2,
-                    key_range: KeyRange {
-                        left: key_with_epoch(iterator_test_user_key_of(3).encode(), test_epoch(3))
+                            right_exclusive: false,
+                        },
+                        file_size: 1,
+                        table_ids: vec![0],
+                        meta_offset: 1,
+                        stale_key_count: 1,
+                        total_key_count: 1,
+                        uncompressed_file_size: 1,
+                        sst_size: 1,
+                        ..Default::default()
+                    }
+                    .into(),
+                ),
+                LocalSstableInfo::for_test(
+                    SstableInfoInner {
+                        object_id: 2,
+                        sst_id: 2,
+                        key_range: KeyRange {
+                            left: key_with_epoch(
+                                iterator_test_user_key_of(3).encode(),
+                                test_epoch(3),
+                            )
                             .into(),
-                        right: key_with_epoch(iterator_test_user_key_of(3).encode(), test_epoch(3))
+                            right: key_with_epoch(
+                                iterator_test_user_key_of(3).encode(),
+                                test_epoch(3),
+                            )
                             .into(),
-                        right_exclusive: false,
-                    },
-                    file_size: 1,
-                    table_ids: vec![0],
-                    meta_offset: 1,
-                    stale_key_count: 1,
-                    total_key_count: 1,
-                    uncompressed_file_size: 1,
-                    sst_size: 1,
-                    ..Default::default()
-                }),
+                            right_exclusive: false,
+                        },
+                        file_size: 1,
+                        table_ids: vec![0],
+                        meta_offset: 1,
+                        stale_key_count: 1,
+                        total_key_count: 1,
+                        uncompressed_file_size: 1,
+                        sst_size: 1,
+                        ..Default::default()
+                    }
+                    .into(),
+                ),
             ],
             vec![],
             epoch_id_vec_for_clear,
diff --git a/src/storage/hummock_test/src/hummock_storage_tests.rs b/src/storage/hummock_test/src/hummock_storage_tests.rs
index d85354636a850..fa95238e49d20 100644
--- a/src/storage/hummock_test/src/hummock_storage_tests.rs
+++ b/src/storage/hummock_test/src/hummock_storage_tests.rs
@@ -29,7 +29,7 @@ use risingwave_common::util::epoch::{test_epoch, EpochExt, INVALID_EPOCH};
 use risingwave_hummock_sdk::key::{
     gen_key_from_bytes, prefixed_range_with_vnode, FullKey, TableKey, UserKey, TABLE_PREFIX_LEN,
 };
-use risingwave_hummock_sdk::sstable_info::SstableInfo;
+use risingwave_hummock_sdk::sstable_info::{SstableInfo, SstableInfoInner};
 use risingwave_hummock_sdk::table_stats::TableStats;
 use risingwave_hummock_sdk::table_watermark::{
     TableWatermarksIndex, VnodeWatermark, WatermarkDirection,
@@ -2663,14 +2663,15 @@ async fn test_commit_multi_epoch() {
         };
 
     let epoch1 = initial_epoch.next_epoch();
-    let sst1_epoch1 = SstableInfo {
+    let sst1_epoch1: SstableInfo = SstableInfoInner {
         sst_id: 11,
         object_id: 1,
         table_ids: vec![existing_table_id.table_id],
         file_size: 100,
         sst_size: 100,
         ..Default::default()
-    };
+    }
+    .into();
 
     commit_epoch(
         epoch1,
@@ -2704,14 +2705,15 @@ async fn test_commit_multi_epoch() {
         assert_eq!(cg_id, info.compaction_group_id);
     }
 
-    let sst1_epoch2 = SstableInfo {
+    let sst1_epoch2: SstableInfo = SstableInfoInner {
         sst_id: 22,
         object_id: 2,
         table_ids: vec![existing_table_id.table_id],
         file_size: 100,
         sst_size: 100,
         ..Default::default()
-    };
+    }
+    .into();
 
     let epoch2 = epoch1.next_epoch();
 
@@ -2740,14 +2742,15 @@ async fn test_commit_multi_epoch() {
 
     let new_table_id = TableId::new(2);
 
-    let sst2_epoch1 = SstableInfo {
+    let sst2_epoch1: SstableInfo = SstableInfoInner {
         sst_id: 33,
         object_id: 3,
         table_ids: vec![new_table_id.table_id],
         file_size: 100,
         sst_size: 100,
         ..Default::default()
-    };
+    }
+    .into();
 
     commit_epoch(
         epoch1,
@@ -2779,14 +2782,15 @@ async fn test_commit_multi_epoch() {
         new_cg_id
     };
 
-    let sst2_epoch2 = SstableInfo {
+    let sst2_epoch2: SstableInfo = SstableInfoInner {
         sst_id: 44,
         object_id: 4,
         table_ids: vec![new_table_id.table_id],
         file_size: 100,
         sst_size: 100,
         ..Default::default()
-    };
+    }
+    .into();
 
     commit_epoch(epoch2, sst2_epoch2.clone(), vec![], &[new_table_id]).await;
 
@@ -2810,14 +2814,15 @@ async fn test_commit_multi_epoch() {
 
     let epoch3 = epoch2.next_epoch();
 
-    let sst_epoch3 = SstableInfo {
+    let sst_epoch3: SstableInfo = SstableInfoInner {
         sst_id: 55,
         object_id: 5,
         table_ids: vec![existing_table_id.table_id, new_table_id.table_id],
         file_size: 100,
         sst_size: 100,
         ..Default::default()
-    };
+    }
+    .into();
 
     commit_epoch(
         epoch3,
@@ -2920,32 +2925,35 @@ async fn test_commit_with_large_size() {
         };
 
     let epoch1 = initial_epoch.next_epoch();
-    let sst1_epoch1 = SstableInfo {
+    let sst1_epoch1: SstableInfo = SstableInfoInner {
         sst_id: 11,
         object_id: 1,
         table_ids: vec![existing_table_id.table_id],
         file_size: 512 << 20,
         sst_size: 512 << 20,
         ..Default::default()
-    };
+    }
+    .into();
 
-    let sst1_epoch2 = SstableInfo {
+    let sst1_epoch2: SstableInfo = SstableInfoInner {
         sst_id: 12,
         object_id: 2,
         table_ids: vec![existing_table_id.table_id],
         file_size: 512 << 20,
         sst_size: 512 << 20,
         ..Default::default()
-    };
+    }
+    .into();
 
-    let sst1_epoch3 = SstableInfo {
+    let sst1_epoch3: SstableInfo = SstableInfoInner {
         sst_id: 13,
         object_id: 3,
         table_ids: vec![existing_table_id.table_id],
         file_size: 512 << 20,
         sst_size: 512 << 20,
         ..Default::default()
-    };
+    }
+    .into();
 
     commit_epoch(
         epoch1,
diff --git a/src/storage/src/hummock/event_handler/uploader/test_utils.rs b/src/storage/src/hummock/event_handler/uploader/test_utils.rs
index 3e7b92624109a..26fe3dc19220a 100644
--- a/src/storage/src/hummock/event_handler/uploader/test_utils.rs
+++ b/src/storage/src/hummock/event_handler/uploader/test_utils.rs
@@ -32,7 +32,7 @@ use risingwave_common::util::epoch::{test_epoch, EpochExt};
 use risingwave_hummock_sdk::compaction_group::StaticCompactionGroupId;
 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::version::HummockVersion;
 use risingwave_hummock_sdk::{HummockEpoch, LocalSstableInfo};
 use risingwave_pb::hummock::{PbHummockVersion, StateTableInfoDelta};
@@ -159,17 +159,20 @@ pub(super) fn gen_sstable_info(
     let start_full_key = FullKey::new(TEST_TABLE_ID, TableKey(dummy_table_key()), start_epoch);
     let end_full_key = FullKey::new(TEST_TABLE_ID, TableKey(dummy_table_key()), end_epoch);
     let gen_sst_object_id = (start_epoch << 8) + end_epoch;
-    vec![LocalSstableInfo::for_test(SstableInfo {
-        object_id: gen_sst_object_id,
-        sst_id: gen_sst_object_id,
-        key_range: KeyRange {
-            left: start_full_key.encode().into(),
-            right: end_full_key.encode().into(),
-            right_exclusive: true,
-        },
-        table_ids: vec![TEST_TABLE_ID.table_id],
-        ..Default::default()
-    })]
+    vec![LocalSstableInfo::for_test(
+        SstableInfoInner {
+            object_id: gen_sst_object_id,
+            sst_id: gen_sst_object_id,
+            key_range: KeyRange {
+                left: start_full_key.encode().into(),
+                right: end_full_key.encode().into(),
+                right_exclusive: true,
+            },
+            table_ids: vec![TEST_TABLE_ID.table_id],
+            ..Default::default()
+        }
+        .into(),
+    )]
 }
 
 pub(super) fn test_uploader_context<F, Fut>(upload_fn: F) -> UploaderContext
diff --git a/src/storage/src/hummock/test_utils.rs b/src/storage/src/hummock/test_utils.rs
index 148810f9b016d..98e081e72af15 100644
--- a/src/storage/src/hummock/test_utils.rs
+++ b/src/storage/src/hummock/test_utils.rs
@@ -27,7 +27,7 @@ use risingwave_common::hash::VirtualNode;
 use risingwave_common::util::epoch::test_epoch;
 use risingwave_hummock_sdk::key::{FullKey, TableKey, UserKey};
 use risingwave_hummock_sdk::key_range::KeyRange;
-use risingwave_hummock_sdk::sstable_info::SstableInfo;
+use risingwave_hummock_sdk::sstable_info::{SstableInfo, SstableInfoInner};
 use risingwave_hummock_sdk::{EpochWithGap, HummockEpoch, HummockSstableObjectId};
 
 use super::iterator::test_utils::iterator_test_table_key_of;
@@ -108,7 +108,7 @@ pub fn gen_dummy_sst_info(
         }
         file_size += batch.size() as u64;
     }
-    SstableInfo {
+    SstableInfoInner {
         object_id: id,
         sst_id: id,
         key_range: KeyRange {
@@ -124,6 +124,7 @@ pub fn gen_dummy_sst_info(
         sst_size: file_size,
         ..Default::default()
     }
+    .into()
 }
 
 /// Number of keys in table generated in `generate_table`.
@@ -203,7 +204,7 @@ pub async fn put_sst(
 
     meta.meta_offset = writer.data_len() as u64;
     meta.bloom_filter = bloom_filter;
-    let sst = SstableInfo {
+    let sst = SstableInfoInner {
         object_id: sst_object_id,
         sst_id: sst_object_id,
         key_range: KeyRange {
@@ -216,7 +217,8 @@ pub async fn put_sst(
         uncompressed_file_size: meta.estimated_size as u64,
         table_ids,
         ..Default::default()
-    };
+    }
+    .into();
     let writer_output = writer.finish(meta).await?;
     writer_output.await.unwrap()?;
     Ok(sst)