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/storage/benches/bench_compactor.rs b/src/storage/benches/bench_compactor.rs
index 77a22c88bea2..8b02ee42f243 100644
--- a/src/storage/benches/bench_compactor.rs
+++ b/src/storage/benches/bench_compactor.rs
@@ -230,7 +230,7 @@ async fn scan_all_table(info: &SstableInfo, sstable_store: SstableStoreRef) {
     let default_read_options = Arc::new(SstableIteratorReadOptions::default());
     // warm up to make them all in memory. I do not use CachePolicy::Fill because it will fetch
     // block from meta.
-    let mut iter = SstableIterator::new(table, sstable_store.clone(), default_read_options);
+    let mut iter = SstableIterator::new(table, sstable_store.clone(), default_read_options, info);
     iter.rewind().await.unwrap();
     while iter.is_valid() {
         iter.next().await.unwrap();
diff --git a/src/storage/src/hummock/iterator/backward_merge.rs b/src/storage/src/hummock/iterator/backward_merge.rs
index 02e47fe7ec64..4896eb0bf015 100644
--- a/src/storage/src/hummock/iterator/backward_merge.rs
+++ b/src/storage/src/hummock/iterator/backward_merge.rs
@@ -24,7 +24,7 @@ mod test {
     #[tokio::test]
     async fn test_backward_merge_basic() {
         let sstable_store = mock_sstable_store().await;
-        let table0 = gen_iterator_test_sstable_base(
+        let (table0, _) = gen_iterator_test_sstable_base(
             0,
             default_builder_opt_for_test(),
             |x| x * 3 + 1,
@@ -32,7 +32,7 @@ mod test {
             TEST_KEYS_COUNT,
         )
         .await;
-        let table1 = gen_iterator_test_sstable_base(
+        let (table1, _) = gen_iterator_test_sstable_base(
             1,
             default_builder_opt_for_test(),
             |x| x * 3 + 2,
@@ -40,7 +40,7 @@ mod test {
             TEST_KEYS_COUNT,
         )
         .await;
-        let table2 = gen_iterator_test_sstable_base(
+        let (table2, _) = gen_iterator_test_sstable_base(
             2,
             default_builder_opt_for_test(),
             |x| x * 3 + 3,
@@ -77,7 +77,7 @@ mod test {
     #[tokio::test]
     async fn test_backward_merge_seek() {
         let sstable_store = mock_sstable_store().await;
-        let table0 = gen_iterator_test_sstable_base(
+        let (table0, _) = gen_iterator_test_sstable_base(
             0,
             default_builder_opt_for_test(),
             |x| x * 3 + 1,
@@ -85,7 +85,7 @@ mod test {
             TEST_KEYS_COUNT,
         )
         .await;
-        let table1 = gen_iterator_test_sstable_base(
+        let (table1, _) = gen_iterator_test_sstable_base(
             1,
             default_builder_opt_for_test(),
             |x| x * 3 + 2,
@@ -93,7 +93,7 @@ mod test {
             TEST_KEYS_COUNT,
         )
         .await;
-        let table2 = gen_iterator_test_sstable_base(
+        let (table2, _) = gen_iterator_test_sstable_base(
             2,
             default_builder_opt_for_test(),
             |x| x * 3 + 3,
@@ -152,7 +152,7 @@ mod test {
     #[tokio::test]
     async fn test_backward_merge_invalidate_reset() {
         let sstable_store = mock_sstable_store().await;
-        let table0 = gen_iterator_test_sstable_base(
+        let (table0, _) = gen_iterator_test_sstable_base(
             0,
             default_builder_opt_for_test(),
             |x| x + 1,
@@ -160,7 +160,7 @@ mod test {
             TEST_KEYS_COUNT,
         )
         .await;
-        let table1 = gen_iterator_test_sstable_base(
+        let (table1, _) = gen_iterator_test_sstable_base(
             1,
             default_builder_opt_for_test(),
             |x| TEST_KEYS_COUNT + x + 1,
diff --git a/src/storage/src/hummock/iterator/backward_user.rs b/src/storage/src/hummock/iterator/backward_user.rs
index ab348571a412..1aef1512a13f 100644
--- a/src/storage/src/hummock/iterator/backward_user.rs
+++ b/src/storage/src/hummock/iterator/backward_user.rs
@@ -330,7 +330,7 @@ mod tests {
     #[tokio::test]
     async fn test_backward_user_basic() {
         let sstable_store = mock_sstable_store().await;
-        let table0 = gen_iterator_test_sstable_base(
+        let (table0, _) = gen_iterator_test_sstable_base(
             0,
             default_builder_opt_for_test(),
             |x| x * 3 + 1,
@@ -338,7 +338,7 @@ mod tests {
             TEST_KEYS_COUNT,
         )
         .await;
-        let table1 = gen_iterator_test_sstable_base(
+        let (table1, _) = gen_iterator_test_sstable_base(
             1,
             default_builder_opt_for_test(),
             |x| x * 3 + 2,
@@ -346,7 +346,7 @@ mod tests {
             TEST_KEYS_COUNT,
         )
         .await;
-        let table2 = gen_iterator_test_sstable_base(
+        let (table2, _) = gen_iterator_test_sstable_base(
             2,
             default_builder_opt_for_test(),
             |x| x * 3 + 3,
@@ -382,7 +382,7 @@ mod tests {
     #[tokio::test]
     async fn test_backward_user_seek() {
         let sstable_store = mock_sstable_store().await;
-        let table0 = gen_iterator_test_sstable_base(
+        let (table0, _) = gen_iterator_test_sstable_base(
             0,
             default_builder_opt_for_test(),
             |x| x * 3 + 1,
@@ -390,7 +390,7 @@ mod tests {
             TEST_KEYS_COUNT,
         )
         .await;
-        let table1 = gen_iterator_test_sstable_base(
+        let (table1, _) = gen_iterator_test_sstable_base(
             1,
             default_builder_opt_for_test(),
             |x| x * 3 + 2,
@@ -398,7 +398,7 @@ mod tests {
             TEST_KEYS_COUNT,
         )
         .await;
-        let table2 = gen_iterator_test_sstable_base(
+        let (table2, _) = gen_iterator_test_sstable_base(
             2,
             default_builder_opt_for_test(),
             |x| x * 3 + 3,
@@ -461,14 +461,14 @@ mod tests {
             (1, 300, HummockValue::delete()),
             (2, 100, HummockValue::put(iterator_test_value_of(2))),
         ];
-        let table0 =
+        let (table0, _) =
             gen_iterator_test_sstable_from_kv_pair(0, kv_pairs, sstable_store.clone()).await;
 
         let kv_pairs = vec![
             (1, 400, HummockValue::put(iterator_test_value_of(1))),
             (2, 200, HummockValue::delete()),
         ];
-        let table1 =
+        let (table1, _) =
             gen_iterator_test_sstable_from_kv_pair(1, kv_pairs, sstable_store.clone()).await;
         let backward_iters = vec![
             BackwardSstableIterator::new(table0, sstable_store.clone()),
@@ -514,7 +514,7 @@ mod tests {
             (7, 100, HummockValue::put(iterator_test_value_of(7))),
             (8, 100, HummockValue::put(iterator_test_value_of(8))),
         ];
-        let sstable =
+        let (sstable, _) =
             gen_iterator_test_sstable_from_kv_pair(0, kv_pairs, sstable_store.clone()).await;
         let backward_iters = vec![BackwardSstableIterator::new(sstable, sstable_store)];
         let bmi = MergeIterator::new(backward_iters);
@@ -591,7 +591,7 @@ mod tests {
             (7, 100, HummockValue::put(iterator_test_value_of(7))),
             (8, 100, HummockValue::put(iterator_test_value_of(8))),
         ];
-        let sstable =
+        let (sstable, _) =
             gen_iterator_test_sstable_from_kv_pair(0, kv_pairs, sstable_store.clone()).await;
         let backward_iters = vec![BackwardSstableIterator::new(
             sstable.clone(),
@@ -692,7 +692,7 @@ mod tests {
             (7, 100, HummockValue::put(iterator_test_value_of(7))),
             (8, 100, HummockValue::put(iterator_test_value_of(8))),
         ];
-        let sstable =
+        let (sstable, _sstable_info) =
             gen_iterator_test_sstable_from_kv_pair(0, kv_pairs, sstable_store.clone()).await;
         let backward_iters = vec![BackwardSstableIterator::new(
             sstable.clone(),
@@ -789,7 +789,7 @@ mod tests {
             (7, 100, HummockValue::put(iterator_test_value_of(7))),
             (8, 100, HummockValue::put(iterator_test_value_of(8))),
         ];
-        let handle =
+        let (handle, _sstable_info) =
             gen_iterator_test_sstable_from_kv_pair(0, kv_pairs, sstable_store.clone()).await;
         let backward_iters = vec![BackwardSstableIterator::new(handle, sstable_store)];
         let bmi = MergeIterator::new(backward_iters);
@@ -968,7 +968,7 @@ mod tests {
             }
         }
         let sstable_store = mock_sstable_store().await;
-        let sst = gen_test_sstable(
+        let (sst, _sstable_info) = gen_test_sstable(
             default_builder_opt_for_test(),
             0,
             truth.iter().flat_map(|(key, inserts)| {
@@ -1107,7 +1107,7 @@ mod tests {
     #[tokio::test]
     async fn test_min_epoch() {
         let sstable_store = mock_sstable_store().await;
-        let table0 = gen_iterator_test_sstable_with_incr_epoch(
+        let (table0, _) = gen_iterator_test_sstable_with_incr_epoch(
             0,
             default_builder_opt_for_test(),
             |x| x * 3,
diff --git a/src/storage/src/hummock/iterator/concat_inner.rs b/src/storage/src/hummock/iterator/concat_inner.rs
index a71a9d940186..ada646d8930b 100644
--- a/src/storage/src/hummock/iterator/concat_inner.rs
+++ b/src/storage/src/hummock/iterator/concat_inner.rs
@@ -42,8 +42,8 @@ pub struct ConcatIteratorInner<TI: SstableIteratorType> {
     /// Current table index.
     cur_idx: usize,
 
-    /// All non-overlapping tables.
-    tables: Vec<SstableInfo>,
+    /// All non-overlapping `sstable_infos`.
+    sstable_infos: Vec<SstableInfo>,
 
     sstable_store: SstableStoreRef,
 
@@ -52,18 +52,18 @@ pub struct ConcatIteratorInner<TI: SstableIteratorType> {
 }
 
 impl<TI: SstableIteratorType> ConcatIteratorInner<TI> {
-    /// Caller should make sure that `tables` are non-overlapping,
+    /// Caller should make sure that `sstable_infos` are non-overlapping,
     /// arranged in ascending order when it serves as a forward iterator,
     /// and arranged in descending order when it serves as a backward iterator.
     pub fn new(
-        tables: Vec<SstableInfo>,
+        sstable_infos: Vec<SstableInfo>,
         sstable_store: SstableStoreRef,
         read_options: Arc<SstableIteratorReadOptions>,
     ) -> Self {
         Self {
             sstable_iter: None,
             cur_idx: 0,
-            tables,
+            sstable_infos,
             sstable_store,
             stats: StoreLocalStatistic::default(),
             read_options,
@@ -76,18 +76,22 @@ impl<TI: SstableIteratorType> ConcatIteratorInner<TI> {
         idx: usize,
         seek_key: Option<FullKey<&[u8]>>,
     ) -> HummockResult<()> {
-        if idx >= self.tables.len() {
+        if idx >= self.sstable_infos.len() {
             if let Some(old_iter) = self.sstable_iter.take() {
                 old_iter.collect_local_statistic(&mut self.stats);
             }
-            self.cur_idx = self.tables.len();
+            self.cur_idx = self.sstable_infos.len();
         } else {
             let table = self
                 .sstable_store
-                .sstable(&self.tables[idx], &mut self.stats)
+                .sstable(&self.sstable_infos[idx], &mut self.stats)
                 .await?;
-            let mut sstable_iter =
-                TI::create(table, self.sstable_store.clone(), self.read_options.clone());
+            let mut sstable_iter = TI::create(
+                table,
+                self.sstable_store.clone(),
+                self.read_options.clone(),
+                &self.sstable_infos[idx],
+            );
 
             if let Some(key) = seek_key {
                 sstable_iter.seek(key).await?;
@@ -118,7 +122,7 @@ impl<TI: SstableIteratorType> HummockIterator for ConcatIteratorInner<TI> {
         } else {
             // seek to next table
             let mut table_idx = self.cur_idx + 1;
-            while !self.is_valid() && table_idx < self.tables.len() {
+            while !self.is_valid() && table_idx < self.sstable_infos.len() {
                 self.seek_idx(table_idx, None).await?;
                 table_idx += 1;
             }
@@ -141,7 +145,7 @@ impl<TI: SstableIteratorType> HummockIterator for ConcatIteratorInner<TI> {
     async fn rewind(&mut self) -> HummockResult<()> {
         self.seek_idx(0, None).await?;
         let mut table_idx = 1;
-        while !self.is_valid() && table_idx < self.tables.len() {
+        while !self.is_valid() && table_idx < self.sstable_infos.len() {
             // Seek to next table
             self.seek_idx(table_idx, None).await?;
             table_idx += 1;
@@ -151,7 +155,7 @@ impl<TI: SstableIteratorType> HummockIterator for ConcatIteratorInner<TI> {
 
     async fn seek<'a>(&'a mut self, key: FullKey<&'a [u8]>) -> HummockResult<()> {
         let mut table_idx = self
-            .tables
+            .sstable_infos
             .partition_point(|table| match Self::Direction::direction() {
                 DirectionEnum::Forward => {
                     let ord = FullKey::decode(smallest_key(table)).cmp(&key);
@@ -167,7 +171,7 @@ impl<TI: SstableIteratorType> HummockIterator for ConcatIteratorInner<TI> {
 
         self.seek_idx(table_idx, Some(key)).await?;
         table_idx += 1;
-        while !self.is_valid() && table_idx < self.tables.len() {
+        while !self.is_valid() && table_idx < self.sstable_infos.len() {
             // Seek to next table
             self.seek_idx(table_idx, None).await?;
             table_idx += 1;
diff --git a/src/storage/src/hummock/iterator/forward_merge.rs b/src/storage/src/hummock/iterator/forward_merge.rs
index 60cb9a84f034..be1a748ec84d 100644
--- a/src/storage/src/hummock/iterator/forward_merge.rs
+++ b/src/storage/src/hummock/iterator/forward_merge.rs
@@ -135,11 +135,13 @@ mod test {
                 sstable_store.sstable(&table0, &mut stats).await.unwrap(),
                 sstable_store.clone(),
                 read_options.clone(),
+                &table0,
             ),
             SstableIterator::create(
                 sstable_store.sstable(&table1, &mut stats).await.unwrap(),
                 sstable_store.clone(),
                 read_options.clone(),
+                &table1,
             ),
         ]);
 
diff --git a/src/storage/src/hummock/iterator/forward_user.rs b/src/storage/src/hummock/iterator/forward_user.rs
index 768aa01af686..9e1cfe9a41c4 100644
--- a/src/storage/src/hummock/iterator/forward_user.rs
+++ b/src/storage/src/hummock/iterator/forward_user.rs
@@ -274,6 +274,7 @@ mod tests {
 
     use bytes::Bytes;
     use risingwave_common::util::epoch::test_epoch;
+    use risingwave_hummock_sdk::sstable_info::SstableInfo;
 
     use super::*;
     use crate::hummock::iterator::test_utils::{
@@ -294,7 +295,7 @@ mod tests {
     async fn test_basic() {
         let sstable_store = mock_sstable_store().await;
         let read_options = Arc::new(SstableIteratorReadOptions::default());
-        let table0 = gen_iterator_test_sstable_base(
+        let (table0, sstable_info_0) = gen_iterator_test_sstable_base(
             0,
             default_builder_opt_for_test(),
             |x| x * 3,
@@ -302,7 +303,7 @@ mod tests {
             TEST_KEYS_COUNT,
         )
         .await;
-        let table1 = gen_iterator_test_sstable_base(
+        let (table1, sstable_info_1) = gen_iterator_test_sstable_base(
             1,
             default_builder_opt_for_test(),
             |x| x * 3 + 1,
@@ -310,7 +311,7 @@ mod tests {
             TEST_KEYS_COUNT,
         )
         .await;
-        let table2 = gen_iterator_test_sstable_base(
+        let (table2, sstable_info_2) = gen_iterator_test_sstable_base(
             2,
             default_builder_opt_for_test(),
             |x| x * 3 + 2,
@@ -319,9 +320,19 @@ mod tests {
         )
         .await;
         let iters = vec![
-            SstableIterator::create(table0, sstable_store.clone(), read_options.clone()),
-            SstableIterator::create(table1, sstable_store.clone(), read_options.clone()),
-            SstableIterator::create(table2, sstable_store, read_options.clone()),
+            SstableIterator::create(
+                table0,
+                sstable_store.clone(),
+                read_options.clone(),
+                &sstable_info_0,
+            ),
+            SstableIterator::create(
+                table1,
+                sstable_store.clone(),
+                read_options.clone(),
+                &sstable_info_1,
+            ),
+            SstableIterator::create(table2, sstable_store, read_options.clone(), &sstable_info_2),
         ];
 
         let mi = MergeIterator::new(iters);
@@ -347,7 +358,7 @@ mod tests {
     #[tokio::test]
     async fn test_seek() {
         let sstable_store = mock_sstable_store().await;
-        let table0 = gen_iterator_test_sstable_base(
+        let (table0, sstable_info_0) = gen_iterator_test_sstable_base(
             0,
             default_builder_opt_for_test(),
             |x| x * 3,
@@ -355,7 +366,7 @@ mod tests {
             TEST_KEYS_COUNT,
         )
         .await;
-        let table1 = gen_iterator_test_sstable_base(
+        let (table1, sstable_info_1) = gen_iterator_test_sstable_base(
             1,
             default_builder_opt_for_test(),
             |x| x * 3 + 1,
@@ -363,7 +374,7 @@ mod tests {
             TEST_KEYS_COUNT,
         )
         .await;
-        let table2 = gen_iterator_test_sstable_base(
+        let (table2, sstable_info_2) = gen_iterator_test_sstable_base(
             2,
             default_builder_opt_for_test(),
             |x| x * 3 + 2,
@@ -373,9 +384,19 @@ mod tests {
         .await;
         let read_options = Arc::new(SstableIteratorReadOptions::default());
         let iters = vec![
-            SstableIterator::create(table0, sstable_store.clone(), read_options.clone()),
-            SstableIterator::create(table1, sstable_store.clone(), read_options.clone()),
-            SstableIterator::create(table2, sstable_store, read_options),
+            SstableIterator::create(
+                table0,
+                sstable_store.clone(),
+                read_options.clone(),
+                &sstable_info_0,
+            ),
+            SstableIterator::create(
+                table1,
+                sstable_store.clone(),
+                read_options.clone(),
+                &sstable_info_1,
+            ),
+            SstableIterator::create(table2, sstable_store, read_options, &sstable_info_2),
         ];
 
         let mi = MergeIterator::new(iters);
@@ -428,20 +449,25 @@ mod tests {
             (1, 100, HummockValue::put(iterator_test_value_of(1))),
             (2, 300, HummockValue::delete()),
         ];
-        let table0 =
+        let (table0, sstable_info_0) =
             gen_iterator_test_sstable_from_kv_pair(0, kv_pairs, sstable_store.clone()).await;
 
         let kv_pairs = vec![
             (1, 200, HummockValue::delete()),
             (2, 400, HummockValue::put(iterator_test_value_of(2))),
         ];
-        let table1 =
+        let (table1, sstable_info_1) =
             gen_iterator_test_sstable_from_kv_pair(1, kv_pairs, sstable_store.clone()).await;
 
         let read_options = Arc::new(SstableIteratorReadOptions::default());
         let iters = vec![
-            SstableIterator::create(table0, sstable_store.clone(), read_options.clone()),
-            SstableIterator::create(table1, sstable_store.clone(), read_options),
+            SstableIterator::create(
+                table0,
+                sstable_store.clone(),
+                read_options.clone(),
+                &sstable_info_0,
+            ),
+            SstableIterator::create(table1, sstable_store.clone(), read_options, &sstable_info_1),
         ];
 
         let mi = MergeIterator::new(iters);
@@ -459,7 +485,7 @@ mod tests {
         assert!(!ui.is_valid());
     }
 
-    async fn generate_test_data(sstable_store: SstableStoreRef) -> TableHolder {
+    async fn generate_test_data(sstable_store: SstableStoreRef) -> (TableHolder, SstableInfo) {
         let kv_pairs = vec![
             (0, 200, HummockValue::delete()),
             (0, 100, HummockValue::put(iterator_test_value_of(0))),
@@ -479,10 +505,13 @@ mod tests {
         let sst_info =
             gen_iterator_test_sstable_with_range_tombstones(0, kv_pairs, sstable_store.clone())
                 .await;
-        sstable_store
-            .sstable(&sst_info, &mut StoreLocalStatistic::default())
-            .await
-            .unwrap()
+        (
+            sstable_store
+                .sstable(&sst_info, &mut StoreLocalStatistic::default())
+                .await
+                .unwrap(),
+            sst_info,
+        )
     }
 
     // left..=end
@@ -490,9 +519,14 @@ mod tests {
     async fn test_range_inclusive() {
         let sstable_store = mock_sstable_store().await;
         // key=[idx, epoch], value
-        let table = generate_test_data(sstable_store.clone()).await;
+        let (table, sstable_info) = generate_test_data(sstable_store.clone()).await;
         let read_options = Arc::new(SstableIteratorReadOptions::default());
-        let iters = vec![SstableIterator::create(table, sstable_store, read_options)];
+        let iters = vec![SstableIterator::create(
+            table,
+            sstable_store,
+            read_options,
+            &sstable_info,
+        )];
         let mi = MergeIterator::new(iters);
 
         let begin_key = Included(iterator_test_bytes_user_key_of(2));
@@ -567,13 +601,14 @@ mod tests {
             (7, 100, HummockValue::put(iterator_test_value_of(7))),
             (8, 100, HummockValue::put(iterator_test_value_of(8))),
         ];
-        let table =
+        let (table, sstable_info) =
             gen_iterator_test_sstable_from_kv_pair(0, kv_pairs, sstable_store.clone()).await;
         let read_options = Arc::new(SstableIteratorReadOptions::default());
         let iters = vec![SstableIterator::create(
             table.clone(),
             sstable_store.clone(),
             read_options.clone(),
+            &sstable_info,
         )];
         let mi = MergeIterator::new(iters);
 
@@ -628,7 +663,12 @@ mod tests {
             .unwrap();
         assert!(!ui.is_valid());
 
-        let iters = vec![SstableIterator::create(table, sstable_store, read_options)];
+        let iters = vec![SstableIterator::create(
+            table,
+            sstable_store,
+            read_options,
+            &sstable_info,
+        )];
         let mi = MergeIterator::new(iters);
 
         let begin_key = Excluded(iterator_test_bytes_user_key_of(2));
@@ -664,9 +704,14 @@ mod tests {
         let sstable_store = mock_sstable_store().await;
         // key=[idx, epoch], value
 
-        let table = generate_test_data(sstable_store.clone()).await;
+        let (table, sstable_info) = generate_test_data(sstable_store.clone()).await;
         let read_options = Arc::new(SstableIteratorReadOptions::default());
-        let iters = vec![SstableIterator::create(table, sstable_store, read_options)];
+        let iters = vec![SstableIterator::create(
+            table,
+            sstable_store,
+            read_options,
+            &sstable_info,
+        )];
         let mi = MergeIterator::new(iters);
         let end_key = Included(iterator_test_bytes_user_key_of(7));
 
@@ -728,9 +773,14 @@ mod tests {
     async fn test_range_from() {
         let sstable_store = mock_sstable_store().await;
         // key=[idx, epoch], value
-        let table = generate_test_data(sstable_store.clone()).await;
+        let (table, sstable_info) = generate_test_data(sstable_store.clone()).await;
         let read_options = Arc::new(SstableIteratorReadOptions::default());
-        let iters = vec![SstableIterator::create(table, sstable_store, read_options)];
+        let iters = vec![SstableIterator::create(
+            table,
+            sstable_store,
+            read_options,
+            &sstable_info,
+        )];
         let mi = MergeIterator::new(iters);
         let begin_key = Included(iterator_test_bytes_user_key_of(2));
 
@@ -795,7 +845,7 @@ mod tests {
     async fn test_min_epoch() {
         let sstable_store = mock_sstable_store().await;
         let read_options = Arc::new(SstableIteratorReadOptions::default());
-        let table0 = gen_iterator_test_sstable_with_incr_epoch(
+        let (table0, sstable_info_0) = gen_iterator_test_sstable_with_incr_epoch(
             0,
             default_builder_opt_for_test(),
             |x| x * 3,
@@ -808,6 +858,7 @@ mod tests {
             table0,
             sstable_store.clone(),
             read_options.clone(),
+            &sstable_info_0,
         )];
 
         let min_count = (TEST_KEYS_COUNT / 5) as u64;
diff --git a/src/storage/src/hummock/iterator/mod.rs b/src/storage/src/hummock/iterator/mod.rs
index e0a1d433892d..4239be4ea81c 100644
--- a/src/storage/src/hummock/iterator/mod.rs
+++ b/src/storage/src/hummock/iterator/mod.rs
@@ -544,7 +544,7 @@ pub trait IteratorFactory {
     fn add_overlapping_sst_iter(&mut self, iter: Self::SstableIteratorType);
     fn add_concat_sst_iter(
         &mut self,
-        tables: Vec<SstableInfo>,
+        sstable_table_infos: Vec<SstableInfo>,
         sstable_store: SstableStoreRef,
         read_options: Arc<SstableIteratorReadOptions>,
     );
diff --git a/src/storage/src/hummock/mod.rs b/src/storage/src/hummock/mod.rs
index 56fcc985b28e..e97731c14ca6 100644
--- a/src/storage/src/hummock/mod.rs
+++ b/src/storage/src/hummock/mod.rs
@@ -91,19 +91,17 @@ pub async fn get_from_sstable_info(
         return Ok(None);
     }
 
-    // TODO: now SstableIterator does not use prefetch through SstableIteratorReadOptions, so we
-    // use default before refinement.
     let mut iter = SstableIterator::create(
         sstable,
         sstable_store_ref.clone(),
         Arc::new(SstableIteratorReadOptions::from_read_options(read_options)),
+        sstable_info,
     );
     iter.seek(full_key).await?;
     // Iterator has sought passed the borders.
     if !iter.is_valid() {
         return Ok(None);
     }
-
     // Iterator gets us the key, we tell if it's the key we want
     // or key next to it.
     let value = if iter.key().user_key == full_key.user_key {
@@ -118,13 +116,13 @@ pub async fn get_from_sstable_info(
 }
 
 pub fn hit_sstable_bloom_filter(
-    sstable_info_ref: &Sstable,
+    sstable_ref: &Sstable,
     user_key_range: &UserKeyRangeRef<'_>,
     prefix_hash: u64,
     local_stats: &mut StoreLocalStatistic,
 ) -> bool {
     local_stats.bloom_filter_check_counts += 1;
-    let may_exist = sstable_info_ref.may_match_hash(user_key_range, prefix_hash);
+    let may_exist = sstable_ref.may_match_hash(user_key_range, prefix_hash);
     if !may_exist {
         local_stats.bloom_filter_true_negative_counts += 1;
     }
diff --git a/src/storage/src/hummock/sstable/backward_sstable_iterator.rs b/src/storage/src/hummock/sstable/backward_sstable_iterator.rs
index b15693aacd56..c04e242c1445 100644
--- a/src/storage/src/hummock/sstable/backward_sstable_iterator.rs
+++ b/src/storage/src/hummock/sstable/backward_sstable_iterator.rs
@@ -17,6 +17,7 @@ use std::sync::Arc;
 
 use foyer::CacheHint;
 use risingwave_hummock_sdk::key::FullKey;
+use risingwave_hummock_sdk::sstable_info::SstableInfo;
 
 use crate::hummock::iterator::{Backward, HummockIterator, ValueMeta};
 use crate::hummock::sstable::SstableIteratorReadOptions;
@@ -162,6 +163,7 @@ impl SstableIteratorType for BackwardSstableIterator {
         sstable: TableHolder,
         sstable_store: SstableStoreRef,
         _: Arc<SstableIteratorReadOptions>,
+        _sstable_info_ref: &SstableInfo,
     ) -> Self {
         BackwardSstableIterator::new(sstable, sstable_store)
     }
@@ -188,7 +190,7 @@ mod tests {
     async fn test_backward_sstable_iterator() {
         // build remote sstable
         let sstable_store = mock_sstable_store().await;
-        let handle =
+        let (handle, _sstable_info) =
             gen_default_test_sstable(default_builder_opt_for_test(), 0, sstable_store.clone())
                 .await;
         // We should have at least 10 blocks, so that sstable iterator test could cover more code
@@ -213,7 +215,7 @@ mod tests {
     #[tokio::test]
     async fn test_backward_sstable_seek() {
         let sstable_store = mock_sstable_store().await;
-        let sstable =
+        let (sstable, _sstable_info) =
             gen_default_test_sstable(default_builder_opt_for_test(), 0, sstable_store.clone())
                 .await;
         // We should have at least 10 blocks, so that sstable iterator test could cover more code
diff --git a/src/storage/src/hummock/sstable/builder.rs b/src/storage/src/hummock/sstable/builder.rs
index 123e6d2f8df1..265597469190 100644
--- a/src/storage/src/hummock/sstable/builder.rs
+++ b/src/storage/src/hummock/sstable/builder.rs
@@ -786,12 +786,14 @@ pub(super) mod tests {
 
         // build remote table
         let sstable_store = mock_sstable_store().await;
+        let table_id_to_vnode = HashMap::from_iter(vec![(0, VirtualNode::COUNT_FOR_TEST)]);
         let sst_info = gen_test_sstable_impl::<Vec<u8>, F>(
             opts,
             0,
             (0..TEST_KEYS_COUNT).map(|i| (test_key_of(i), HummockValue::put(test_value_of(i)))),
             sstable_store.clone(),
             CachePolicy::NotFill,
+            table_id_to_vnode,
         )
         .await;
         let table = sstable_store
diff --git a/src/storage/src/hummock/sstable/forward_sstable_iterator.rs b/src/storage/src/hummock/sstable/forward_sstable_iterator.rs
index b9da631fc492..a8b6c6edf42c 100644
--- a/src/storage/src/hummock/sstable/forward_sstable_iterator.rs
+++ b/src/storage/src/hummock/sstable/forward_sstable_iterator.rs
@@ -12,12 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::cmp::Ordering::{Equal, Less};
 use std::ops::Bound::*;
 use std::sync::Arc;
 
 use await_tree::InstrumentAwait;
 use risingwave_hummock_sdk::key::FullKey;
+use risingwave_hummock_sdk::sstable_info::SstableInfo;
 use thiserror_ext::AsReport;
 
 use super::super::{HummockResult, HummockValue};
@@ -32,6 +32,7 @@ pub trait SstableIteratorType: HummockIterator + 'static {
         sstable: TableHolder,
         sstable_store: SstableStoreRef,
         read_options: Arc<SstableIteratorReadOptions>,
+        sstable_info_ref: &SstableInfo,
     ) -> Self;
 }
 
@@ -52,6 +53,9 @@ pub struct SstableIterator {
     sstable_store: SstableStoreRef,
     stats: StoreLocalStatistic,
     options: Arc<SstableIteratorReadOptions>,
+
+    // used for checking if the block is valid, filter out the block that is not in the table-id range
+    read_block_meta_range: (usize, usize),
 }
 
 impl SstableIterator {
@@ -59,7 +63,79 @@ impl SstableIterator {
         sstable: TableHolder,
         sstable_store: SstableStoreRef,
         options: Arc<SstableIteratorReadOptions>,
+        sstable_info_ref: &SstableInfo,
     ) -> Self {
+        let mut start_idx = 0;
+        let mut end_idx = sstable.meta.block_metas.len() - 1;
+        let read_table_id_range = (
+            *sstable_info_ref.table_ids.first().unwrap(),
+            *sstable_info_ref.table_ids.last().unwrap(),
+        );
+        assert!(
+            read_table_id_range.0 <= read_table_id_range.1,
+            "invalid table id range {} - {}",
+            read_table_id_range.0,
+            read_table_id_range.1
+        );
+        let block_meta_count = sstable.meta.block_metas.len();
+        assert!(block_meta_count > 0);
+        assert!(
+            sstable.meta.block_metas[0].table_id().table_id() <= read_table_id_range.0,
+            "table id {} not found table_ids in block_meta {:?}",
+            read_table_id_range.0,
+            sstable
+                .meta
+                .block_metas
+                .iter()
+                .map(|meta| meta.table_id())
+                .collect::<Vec<_>>()
+        );
+        assert!(
+            sstable.meta.block_metas[block_meta_count - 1]
+                .table_id()
+                .table_id()
+                >= read_table_id_range.1,
+            "table id {} not found table_ids in block_meta {:?}",
+            read_table_id_range.1,
+            sstable
+                .meta
+                .block_metas
+                .iter()
+                .map(|meta| meta.table_id())
+                .collect::<Vec<_>>()
+        );
+
+        while start_idx < block_meta_count
+            && sstable.meta.block_metas[start_idx].table_id().table_id() < read_table_id_range.0
+        {
+            start_idx += 1;
+        }
+        // We assume that the table id read must exist in the sstable, otherwise it is a fatal error.
+        assert!(
+            start_idx < block_meta_count,
+            "table id {} not found table_ids in block_meta {:?}",
+            read_table_id_range.0,
+            sstable
+                .meta
+                .block_metas
+                .iter()
+                .map(|meta| meta.table_id())
+                .collect::<Vec<_>>()
+        );
+
+        while end_idx > start_idx
+            && sstable.meta.block_metas[end_idx].table_id().table_id() > read_table_id_range.1
+        {
+            end_idx -= 1;
+        }
+        assert!(
+            end_idx >= start_idx,
+            "end_idx {} < start_idx {} block_meta_count {}",
+            end_idx,
+            start_idx,
+            block_meta_count
+        );
+
         Self {
             block_iter: None,
             cur_idx: 0,
@@ -70,6 +146,7 @@ impl SstableIterator {
             options,
             preload_end_block_idx: 0,
             preload_retry_times: 0,
+            read_block_meta_range: (start_idx, end_idx),
         }
     }
 
@@ -78,7 +155,7 @@ impl SstableIterator {
         if let Some(bound) = self.options.must_iterated_end_user_key.as_ref() {
             let block_metas = &self.sst.meta.block_metas;
             let next_to_start_idx = start_idx + 1;
-            if next_to_start_idx < block_metas.len() {
+            if next_to_start_idx <= self.read_block_meta_range.1 {
                 let end_idx = match bound {
                     Unbounded => block_metas.len(),
                     Included(dest_key) => {
@@ -120,7 +197,7 @@ impl SstableIterator {
         tokio::task::consume_budget().await;
 
         let mut hit_cache = false;
-        if idx >= self.sst.block_count() {
+        if idx > self.read_block_meta_range.1 {
             self.block_iter = None;
             return Ok(());
         }
@@ -225,8 +302,21 @@ impl SstableIterator {
         }
 
         self.cur_idx = idx;
+
         Ok(())
     }
+
+    fn calculate_block_idx_by_key(&self, key: FullKey<&[u8]>) -> usize {
+        self.read_block_meta_range.0
+            + self.sst.meta.block_metas[self.read_block_meta_range.0..=self.read_block_meta_range.1]
+                .partition_point(|block_meta| {
+                    // compare by version comparator
+                    // Note: we are comparing against the `smallest_key` of the `block`, thus the
+                    // partition point should be `prev(<=)` instead of `<`.
+                    FullKey::decode(&block_meta.smallest_key).le(&key)
+                })
+                .saturating_sub(1) // considering the boundary of 0
+    }
 }
 
 impl HummockIterator for SstableIterator {
@@ -239,6 +329,7 @@ impl HummockIterator for SstableIterator {
             // seek to next block
             self.seek_idx(self.cur_idx + 1, None).await?;
         }
+
         Ok(())
     }
 
@@ -257,24 +348,14 @@ impl HummockIterator for SstableIterator {
     }
 
     async fn rewind(&mut self) -> HummockResult<()> {
-        self.init_block_prefetch_range(0);
-        self.seek_idx(0, None).await?;
+        self.init_block_prefetch_range(self.read_block_meta_range.0);
+        // seek_idx will update the current block iter state
+        self.seek_idx(self.read_block_meta_range.0, None).await?;
         Ok(())
     }
 
     async fn seek<'a>(&'a mut self, key: FullKey<&'a [u8]>) -> HummockResult<()> {
-        let block_idx = self
-            .sst
-            .meta
-            .block_metas
-            .partition_point(|block_meta| {
-                // compare by version comparator
-                // Note: we are comparing against the `smallest_key` of the `block`, thus the
-                // partition point should be `prev(<=)` instead of `<`.
-                let ord = FullKey::decode(&block_meta.smallest_key).cmp(&key);
-                ord == Less || ord == Equal
-            })
-            .saturating_sub(1); // considering the boundary of 0
+        let block_idx = self.calculate_block_idx_by_key(key);
         self.init_block_prefetch_range(block_idx);
 
         self.seek_idx(block_idx, Some(key)).await?;
@@ -302,8 +383,9 @@ impl SstableIteratorType for SstableIterator {
         sstable: TableHolder,
         sstable_store: SstableStoreRef,
         options: Arc<SstableIteratorReadOptions>,
+        sstable_info_ref: &SstableInfo,
     ) -> Self {
-        SstableIterator::new(sstable, sstable_store, options)
+        SstableIterator::new(sstable, sstable_store, options, sstable_info_ref)
     }
 }
 
@@ -319,23 +401,30 @@ mod tests {
     use risingwave_common::hash::VirtualNode;
     use risingwave_common::util::epoch::test_epoch;
     use risingwave_hummock_sdk::key::{TableKey, UserKey};
+    use risingwave_hummock_sdk::sstable_info::{SstableInfo, SstableInfoInner};
+    use risingwave_hummock_sdk::EpochWithGap;
 
     use super::*;
     use crate::assert_bytes_eq;
     use crate::hummock::iterator::test_utils::mock_sstable_store;
     use crate::hummock::test_utils::{
-        default_builder_opt_for_test, gen_default_test_sstable, gen_test_sstable_info, test_key_of,
-        test_value_of, TEST_KEYS_COUNT,
+        default_builder_opt_for_test, gen_default_test_sstable, gen_test_sstable_info,
+        gen_test_sstable_with_table_ids, test_key_of, test_value_of, TEST_KEYS_COUNT,
     };
     use crate::hummock::CachePolicy;
 
-    async fn inner_test_forward_iterator(sstable_store: SstableStoreRef, handle: TableHolder) {
+    async fn inner_test_forward_iterator(
+        sstable_store: SstableStoreRef,
+        handle: TableHolder,
+        sstable_info: SstableInfo,
+    ) {
         // We should have at least 10 blocks, so that sstable iterator test could cover more code
         // path.
         let mut sstable_iter = SstableIterator::create(
             handle,
             sstable_store,
             Arc::new(SstableIteratorReadOptions::default()),
+            &sstable_info,
         );
         let mut cnt = 0;
         sstable_iter.rewind().await.unwrap();
@@ -356,20 +445,20 @@ mod tests {
     async fn test_table_iterator() {
         // Build remote sstable
         let sstable_store = mock_sstable_store().await;
-        let sstable =
+        let (sstable, sstable_info) =
             gen_default_test_sstable(default_builder_opt_for_test(), 0, sstable_store.clone())
                 .await;
         // We should have at least 10 blocks, so that sstable iterator test could cover more code
         // path.
         assert!(sstable.meta.block_metas.len() > 10);
 
-        inner_test_forward_iterator(sstable_store.clone(), sstable).await;
+        inner_test_forward_iterator(sstable_store.clone(), sstable, sstable_info).await;
     }
 
     #[tokio::test]
     async fn test_table_seek() {
         let sstable_store = mock_sstable_store().await;
-        let sstable =
+        let (sstable, sstable_info) =
             gen_default_test_sstable(default_builder_opt_for_test(), 0, sstable_store.clone())
                 .await;
         // We should have at least 10 blocks, so that sstable iterator test could cover more code
@@ -379,6 +468,7 @@ mod tests {
             sstable,
             sstable_store,
             Arc::new(SstableIteratorReadOptions::default()),
+            &sstable_info,
         );
         let mut all_key_to_test = (0..TEST_KEYS_COUNT).collect_vec();
         let mut rng = thread_rng();
@@ -487,6 +577,7 @@ mod tests {
             sstable_store.sstable(&sst_info, &mut stats).await.unwrap(),
             sstable_store.clone(),
             options.clone(),
+            &sst_info,
         );
         let mut cnt = 1000;
         sstable_iter.seek(test_key_of(cnt).to_ref()).await.unwrap();
@@ -509,6 +600,7 @@ mod tests {
             sstable_store.sstable(&sst_info, &mut stats).await.unwrap(),
             sstable_store,
             options.clone(),
+            &sst_info,
         );
         let mut cnt = 1000;
         sstable_iter.seek(test_key_of(cnt).to_ref()).await.unwrap();
@@ -522,4 +614,220 @@ mod tests {
         }
         assert_eq!(cnt, TEST_KEYS_COUNT);
     }
+
+    #[tokio::test]
+    async fn test_read_table_id_range() {
+        {
+            let sstable_store = mock_sstable_store().await;
+            let (sstable, sstable_info) =
+                gen_default_test_sstable(default_builder_opt_for_test(), 0, sstable_store.clone())
+                    .await;
+            let mut sstable_iter = SstableIterator::create(
+                sstable,
+                sstable_store.clone(),
+                Arc::new(SstableIteratorReadOptions::default()),
+                &sstable_info,
+            );
+            sstable_iter.rewind().await.unwrap();
+            assert!(sstable_iter.is_valid());
+            assert_eq!(sstable_iter.key(), test_key_of(0).to_ref());
+        }
+
+        {
+            let sstable_store = mock_sstable_store().await;
+            // test key_range right
+            let k1 = {
+                let mut table_key = VirtualNode::ZERO.to_be_bytes().to_vec();
+                table_key.extend_from_slice(format!("key_test_{:05}", 1).as_bytes());
+                let uk = UserKey::for_test(TableId::from(1), table_key);
+                FullKey {
+                    user_key: uk,
+                    epoch_with_gap: EpochWithGap::new_from_epoch(test_epoch(1)),
+                }
+            };
+
+            let k2 = {
+                let mut table_key = VirtualNode::ZERO.to_be_bytes().to_vec();
+                table_key.extend_from_slice(format!("key_test_{:05}", 2).as_bytes());
+                let uk = UserKey::for_test(TableId::from(2), table_key);
+                FullKey {
+                    user_key: uk,
+                    epoch_with_gap: EpochWithGap::new_from_epoch(test_epoch(1)),
+                }
+            };
+
+            let k3 = {
+                let mut table_key = VirtualNode::ZERO.to_be_bytes().to_vec();
+                table_key.extend_from_slice(format!("key_test_{:05}", 3).as_bytes());
+                let uk = UserKey::for_test(TableId::from(3), table_key);
+                FullKey {
+                    user_key: uk,
+                    epoch_with_gap: EpochWithGap::new_from_epoch(test_epoch(1)),
+                }
+            };
+
+            {
+                let kv_pairs = vec![
+                    (k1.clone(), HummockValue::put(test_value_of(1))),
+                    (k2.clone(), HummockValue::put(test_value_of(2))),
+                    (k3.clone(), HummockValue::put(test_value_of(3))),
+                ];
+
+                let (sstable, _sstable_info) = gen_test_sstable_with_table_ids(
+                    default_builder_opt_for_test(),
+                    10,
+                    kv_pairs.into_iter(),
+                    sstable_store.clone(),
+                    vec![1, 2, 3],
+                )
+                .await;
+                let mut sstable_iter = SstableIterator::create(
+                    sstable,
+                    sstable_store.clone(),
+                    Arc::new(SstableIteratorReadOptions::default()),
+                    &SstableInfo::from(SstableInfoInner {
+                        table_ids: vec![1, 2, 3],
+                        ..Default::default()
+                    }),
+                );
+                sstable_iter.rewind().await.unwrap();
+                assert!(sstable_iter.is_valid());
+                assert!(sstable_iter.key().eq(&k1.to_ref()));
+
+                let mut cnt = 0;
+                let mut last_key = k1.clone();
+                while sstable_iter.is_valid() {
+                    last_key = sstable_iter.key().to_vec();
+                    cnt += 1;
+                    sstable_iter.next().await.unwrap();
+                }
+
+                assert_eq!(3, cnt);
+                assert_eq!(last_key, k3.clone());
+            }
+
+            {
+                let kv_pairs = vec![
+                    (k1.clone(), HummockValue::put(test_value_of(1))),
+                    (k2.clone(), HummockValue::put(test_value_of(2))),
+                    (k3.clone(), HummockValue::put(test_value_of(3))),
+                ];
+
+                let (sstable, _sstable_info) = gen_test_sstable_with_table_ids(
+                    default_builder_opt_for_test(),
+                    10,
+                    kv_pairs.into_iter(),
+                    sstable_store.clone(),
+                    vec![1, 2, 3],
+                )
+                .await;
+
+                let mut sstable_iter = SstableIterator::create(
+                    sstable,
+                    sstable_store.clone(),
+                    Arc::new(SstableIteratorReadOptions::default()),
+                    &SstableInfo::from(SstableInfoInner {
+                        table_ids: vec![1, 2],
+                        ..Default::default()
+                    }),
+                );
+                sstable_iter.rewind().await.unwrap();
+                assert!(sstable_iter.is_valid());
+                assert!(sstable_iter.key().eq(&k1.to_ref()));
+
+                let mut cnt = 0;
+                let mut last_key = k1.clone();
+                while sstable_iter.is_valid() {
+                    last_key = sstable_iter.key().to_vec();
+                    cnt += 1;
+                    sstable_iter.next().await.unwrap();
+                }
+
+                assert_eq!(2, cnt);
+                assert_eq!(last_key, k2.clone());
+            }
+
+            {
+                let kv_pairs = vec![
+                    (k1.clone(), HummockValue::put(test_value_of(1))),
+                    (k2.clone(), HummockValue::put(test_value_of(2))),
+                    (k3.clone(), HummockValue::put(test_value_of(3))),
+                ];
+
+                let (sstable, _sstable_info) = gen_test_sstable_with_table_ids(
+                    default_builder_opt_for_test(),
+                    10,
+                    kv_pairs.into_iter(),
+                    sstable_store.clone(),
+                    vec![1, 2, 3],
+                )
+                .await;
+
+                let mut sstable_iter = SstableIterator::create(
+                    sstable,
+                    sstable_store.clone(),
+                    Arc::new(SstableIteratorReadOptions::default()),
+                    &SstableInfo::from(SstableInfoInner {
+                        table_ids: vec![2, 3],
+                        ..Default::default()
+                    }),
+                );
+                sstable_iter.rewind().await.unwrap();
+                assert!(sstable_iter.is_valid());
+                assert!(sstable_iter.key().eq(&k2.to_ref()));
+
+                let mut cnt = 0;
+                let mut last_key = k1.clone();
+                while sstable_iter.is_valid() {
+                    last_key = sstable_iter.key().to_vec();
+                    cnt += 1;
+                    sstable_iter.next().await.unwrap();
+                }
+
+                assert_eq!(2, cnt);
+                assert_eq!(last_key, k3.clone());
+            }
+
+            {
+                let kv_pairs = vec![
+                    (k1.clone(), HummockValue::put(test_value_of(1))),
+                    (k2.clone(), HummockValue::put(test_value_of(2))),
+                    (k3.clone(), HummockValue::put(test_value_of(3))),
+                ];
+
+                let (sstable, _sstable_info) = gen_test_sstable_with_table_ids(
+                    default_builder_opt_for_test(),
+                    10,
+                    kv_pairs.into_iter(),
+                    sstable_store.clone(),
+                    vec![1, 2, 3],
+                )
+                .await;
+
+                let mut sstable_iter = SstableIterator::create(
+                    sstable,
+                    sstable_store.clone(),
+                    Arc::new(SstableIteratorReadOptions::default()),
+                    &SstableInfo::from(SstableInfoInner {
+                        table_ids: vec![2],
+                        ..Default::default()
+                    }),
+                );
+                sstable_iter.rewind().await.unwrap();
+                assert!(sstable_iter.is_valid());
+                assert!(sstable_iter.key().eq(&k2.to_ref()));
+
+                let mut cnt = 0;
+                let mut last_key = k1.clone();
+                while sstable_iter.is_valid() {
+                    last_key = sstable_iter.key().to_vec();
+                    cnt += 1;
+                    sstable_iter.next().await.unwrap();
+                }
+
+                assert_eq!(1, cnt);
+                assert_eq!(last_key, k2.clone());
+            }
+        }
+    }
 }
diff --git a/src/storage/src/hummock/sstable_store.rs b/src/storage/src/hummock/sstable_store.rs
index 24c3acfef994..3fb0170b17f6 100644
--- a/src/storage/src/hummock/sstable_store.rs
+++ b/src/storage/src/hummock/sstable_store.rs
@@ -537,16 +537,16 @@ impl SstableStore {
     /// Returns `table_holder`
     pub fn sstable(
         &self,
-        sst: &SstableInfo,
+        sstable_info_ref: &SstableInfo,
         stats: &mut StoreLocalStatistic,
     ) -> impl Future<Output = HummockResult<TableHolder>> + Send + 'static {
-        let object_id = sst.object_id;
+        let object_id = sstable_info_ref.object_id;
 
         let entry = self.meta_cache.fetch(object_id, || {
             let store = self.store.clone();
             let meta_path = self.get_sst_data_path(object_id);
             let stats_ptr = stats.remote_io_time.clone();
-            let range = sst.meta_offset as usize..;
+            let range = sstable_info_ref.meta_offset as usize..;
             async move {
                 let now = Instant::now();
                 let buf = store.read(&meta_path, range).await?;
@@ -715,6 +715,7 @@ mod tests {
             holder,
             sstable_store,
             Arc::new(SstableIteratorReadOptions::default()),
+            info,
         );
         iter.rewind().await.unwrap();
         for i in x_range {
@@ -748,7 +749,7 @@ mod tests {
             meta.clone(),
             sstable_store.clone(),
             writer_opts,
-            vec![SST_ID as u32],
+            vec![0],
         )
         .await
         .unwrap();
@@ -779,7 +780,7 @@ mod tests {
             meta.clone(),
             sstable_store.clone(),
             writer_opts,
-            vec![SST_ID as u32],
+            vec![0],
         )
         .await
         .unwrap();
diff --git a/src/storage/src/hummock/store/version.rs b/src/storage/src/hummock/store/version.rs
index ce1173fe8262..8822cb1ebd99 100644
--- a/src/storage/src/hummock/store/version.rs
+++ b/src/storage/src/hummock/store/version.rs
@@ -877,6 +877,7 @@ impl HummockVersionReader {
                 table_holder,
                 self.sstable_store.clone(),
                 sst_read_options.clone(),
+                sstable_info,
             ));
         }
         local_stats.staging_sst_iter_count = staging_sst_iter_count;
@@ -889,22 +890,20 @@ impl HummockVersionReader {
             }
 
             if level.level_type == LevelType::Nonoverlapping {
-                let table_infos = prune_nonoverlapping_ssts(&level.table_infos, user_key_range_ref);
-                let sstables = table_infos
-                    .filter(|sstable_info| {
-                        sstable_info
-                            .table_ids
-                            .binary_search(&read_options.table_id.table_id)
-                            .is_ok()
-                    })
-                    .cloned()
-                    .collect_vec();
-                if sstables.is_empty() {
+                let mut table_infos = prune_nonoverlapping_ssts(
+                    &level.table_infos,
+                    user_key_range_ref,
+                    read_options.table_id.table_id(),
+                )
+                .peekable();
+
+                if table_infos.peek().is_none() {
                     continue;
                 }
-                if sstables.len() > 1 {
+                let sstable_infos = table_infos.cloned().collect_vec();
+                if sstable_infos.len() > 1 {
                     factory.add_concat_sst_iter(
-                        sstables,
+                        sstable_infos,
                         self.sstable_store.clone(),
                         sst_read_options.clone(),
                     );
@@ -912,7 +911,7 @@ impl HummockVersionReader {
                 } else {
                     let sstable = self
                         .sstable_store
-                        .sstable(&sstables[0], local_stats)
+                        .sstable(&sstable_infos[0], local_stats)
                         .await?;
 
                     if let Some(dist_hash) = bloom_filter_prefix_hash.as_ref() {
@@ -934,6 +933,7 @@ impl HummockVersionReader {
                         sstable,
                         self.sstable_store.clone(),
                         sst_read_options.clone(),
+                        &sstable_infos[0],
                     ));
                     local_stats.non_overlapping_iter_count += 1;
                 }
@@ -968,6 +968,7 @@ impl HummockVersionReader {
                         sstable,
                         self.sstable_store.clone(),
                         sst_read_options.clone(),
+                        sstable_info,
                     ));
                     local_stats.overlapping_iter_count += 1;
                 }
@@ -1017,19 +1018,24 @@ impl HummockVersionReader {
         });
 
         async fn make_iter(
-            ssts: impl Iterator<Item = &SstableInfo>,
+            sstable_infos: impl Iterator<Item = &SstableInfo>,
             sstable_store: &SstableStoreRef,
             read_options: Arc<SstableIteratorReadOptions>,
             local_stat: &mut StoreLocalStatistic,
         ) -> HummockResult<MergeIterator<SstableIterator>> {
-            let iters = try_join_all(ssts.map(|sst| {
+            let iters = try_join_all(sstable_infos.map(|sstable_info| {
                 let sstable_store = sstable_store.clone();
                 let read_options = read_options.clone();
                 async move {
                     let mut local_stat = StoreLocalStatistic::default();
-                    let table_holder = sstable_store.sstable(sst, &mut local_stat).await?;
+                    let table_holder = sstable_store.sstable(sstable_info, &mut local_stat).await?;
                     Ok::<_, HummockError>((
-                        SstableIterator::new(table_holder, sstable_store, read_options),
+                        SstableIterator::new(
+                            table_holder,
+                            sstable_store,
+                            read_options,
+                            sstable_info,
+                        ),
                         local_stat,
                     ))
                 }
diff --git a/src/storage/src/hummock/utils.rs b/src/storage/src/hummock/utils.rs
index f73d434cd8c5..7344e8b4e30e 100644
--- a/src/storage/src/hummock/utils.rs
+++ b/src/storage/src/hummock/utils.rs
@@ -27,6 +27,7 @@ use parking_lot::Mutex;
 use risingwave_common::catalog::{TableId, TableOption};
 use risingwave_common::config::StorageMemoryConfig;
 use risingwave_hummock_sdk::can_concat;
+use risingwave_hummock_sdk::compaction_group::StateTableId;
 use risingwave_hummock_sdk::key::{
     bound_table_key_range, EmptySliceRef, FullKey, TableKey, UserKey,
 };
@@ -123,6 +124,7 @@ where
 pub fn prune_nonoverlapping_ssts<'a>(
     ssts: &'a [SstableInfo],
     user_key_range: (Bound<UserKey<&'a [u8]>>, Bound<UserKey<&'a [u8]>>),
+    table_id: StateTableId,
 ) -> impl DoubleEndedIterator<Item = &'a SstableInfo> {
     debug_assert!(can_concat(ssts));
     let start_table_idx = match user_key_range.0 {
@@ -133,7 +135,9 @@ pub fn prune_nonoverlapping_ssts<'a>(
         Included(key) | Excluded(key) => search_sst_idx(ssts, key).saturating_sub(1),
         _ => ssts.len().saturating_sub(1),
     };
-    ssts[start_table_idx..=end_table_idx].iter()
+    ssts[start_table_idx..=end_table_idx]
+        .iter()
+        .filter(move |sst| sst.table_ids.binary_search(&table_id).is_ok())
 }
 
 type RequestQueue = VecDeque<(Sender<MemoryTracker>, u64)>;
diff --git a/src/storage/src/hummock/validator.rs b/src/storage/src/hummock/validator.rs
index 2c0efbb3ca93..f168454a0063 100644
--- a/src/storage/src/hummock/validator.rs
+++ b/src/storage/src/hummock/validator.rs
@@ -32,18 +32,30 @@ use crate::monitor::StoreLocalStatistic;
 pub async fn validate_ssts(task: ValidationTask, sstable_store: SstableStoreRef) {
     let mut visited_keys = HashMap::new();
     let mut unused = StoreLocalStatistic::default();
-    for sst in task.sst_infos {
+    for sstable_info in task.sst_infos {
         let mut key_counts = 0;
         let worker_id = *task
             .sst_id_to_worker_id
-            .get(&sst.object_id)
+            .get(&sstable_info.object_id)
             .expect("valid worker_id");
-        tracing::debug!("Validating SST {} from worker {}", sst.object_id, worker_id,);
-        let holder = match sstable_store.sstable(&sst, unused.borrow_mut()).await {
+        tracing::debug!(
+            "Validating SST sst_id {} object_id {} from worker {}",
+            sstable_info.sst_id,
+            sstable_info.object_id,
+            worker_id,
+        );
+        let holder = match sstable_store
+            .sstable(&sstable_info, unused.borrow_mut())
+            .await
+        {
             Ok(holder) => holder,
             Err(_err) => {
                 // One reasonable cause is the SST has been vacuumed.
-                tracing::info!("Skip sanity check for SST {}.", sst.object_id);
+                tracing::info!(
+                    "Skip sanity check for SST sst_id {} object_id {} .",
+                    sstable_info.sst_id,
+                    sstable_info.object_id,
+                );
                 continue;
             }
         };
@@ -59,10 +71,15 @@ pub async fn validate_ssts(task: ValidationTask, sstable_store: SstableStoreRef)
                 max_preload_retry_times: 0,
                 prefetch_for_large_query: false,
             }),
+            &sstable_info,
         );
         let mut previous_key: Option<FullKey<Vec<u8>>> = None;
         if let Err(_err) = iter.rewind().await {
-            tracing::info!("Skip sanity check for SST {}.", sst.object_id);
+            tracing::info!(
+                "Skip sanity check for SST sst_id {} object_id {}.",
+                sstable_info.sst_id,
+                sstable_info.object_id
+            );
         }
         while iter.is_valid() {
             key_counts += 1;
@@ -73,29 +90,37 @@ pub async fn validate_ssts(task: ValidationTask, sstable_store: SstableStoreRef)
             {
                 panic!("SST sanity check failed: Duplicate key {:x?} in SST object {} from worker {} and SST object {} from worker {}",
                        current_key,
-                       sst.object_id,
+                       sstable_info.object_id,
                        worker_id,
                        duplicate_sst_object_id,
                        duplicate_worker_id)
             }
-            visited_keys.insert(current_key.to_owned(), (sst.object_id, worker_id));
+            visited_keys.insert(current_key.to_owned(), (sstable_info.object_id, worker_id));
             // Ordered and Locally unique
             if let Some(previous_key) = previous_key.take() {
                 let cmp = previous_key.cmp(&current_key);
                 if cmp != cmp::Ordering::Less {
                     panic!(
-                        "SST sanity check failed: For SST {}, expect {:x?} < {:x?}, got {:#?}",
-                        sst.object_id, previous_key, current_key, cmp
+                        "SST sanity check failed: For SST sst_id {} object_id {}, expect {:x?} < {:x?}, got {:#?}",
+                        sstable_info.sst_id, sstable_info.object_id, previous_key, current_key, cmp
                     )
                 }
             }
             previous_key = Some(current_key);
             if let Err(_err) = iter.next().await {
-                tracing::info!("Skip remaining sanity check for SST {}", sst.object_id,);
+                tracing::info!(
+                    "Skip remaining sanity check for SST {}",
+                    sstable_info.object_id
+                );
                 break;
             }
         }
-        tracing::debug!("Validated {} keys for SST {}", key_counts, sst.object_id,);
+        tracing::debug!(
+            "Validated {} keys for SST sst_id {} object_id {}",
+            key_counts,
+            sstable_info.sst_id,
+            sstable_info.object_id
+        );
         iter.collect_local_statistic(&mut unused);
         unused.ignore();
     }


Test patch:
diff --git a/src/storage/src/hummock/iterator/test_utils.rs b/src/storage/src/hummock/iterator/test_utils.rs
index 5c0fdd79417a..ff5a9c9deaee 100644
--- a/src/storage/src/hummock/iterator/test_utils.rs
+++ b/src/storage/src/hummock/iterator/test_utils.rs
@@ -186,7 +186,7 @@ pub async fn gen_iterator_test_sstable_base(
     idx_mapping: impl Fn(usize) -> usize,
     sstable_store: SstableStoreRef,
     total: usize,
-) -> TableHolder {
+) -> (TableHolder, SstableInfo) {
     gen_test_sstable(
         opts,
         object_id,
@@ -206,7 +206,7 @@ pub async fn gen_iterator_test_sstable_from_kv_pair(
     object_id: HummockSstableObjectId,
     kv_pairs: Vec<(usize, u64, HummockValue<Vec<u8>>)>,
     sstable_store: SstableStoreRef,
-) -> TableHolder {
+) -> (TableHolder, SstableInfo) {
     gen_test_sstable(
         default_builder_opt_for_test(),
         object_id,
@@ -242,7 +242,7 @@ pub async fn gen_merge_iterator_interleave_test_sstable_iters(
     let sstable_store = mock_sstable_store().await;
     let mut result = vec![];
     for i in 0..count {
-        let table = gen_iterator_test_sstable_base(
+        let (table, sstable_info) = gen_iterator_test_sstable_base(
             i as HummockSstableObjectId,
             default_builder_opt_for_test(),
             |x| x * count + i,
@@ -254,6 +254,7 @@ pub async fn gen_merge_iterator_interleave_test_sstable_iters(
             table,
             sstable_store.clone(),
             Arc::new(SstableIteratorReadOptions::default()),
+            &sstable_info,
         ));
     }
     result
@@ -266,7 +267,7 @@ pub async fn gen_iterator_test_sstable_with_incr_epoch(
     sstable_store: SstableStoreRef,
     total: usize,
     epoch_base: u64,
-) -> TableHolder {
+) -> (TableHolder, SstableInfo) {
     gen_test_sstable(
         opts,
         object_id,
diff --git a/src/storage/src/hummock/test_utils.rs b/src/storage/src/hummock/test_utils.rs
index 98e081e72af1..81f6ea42eae3 100644
--- a/src/storage/src/hummock/test_utils.rs
+++ b/src/storage/src/hummock/test_utils.rs
@@ -25,6 +25,7 @@ use risingwave_common::catalog::TableId;
 use risingwave_common::config::EvictionConfig;
 use risingwave_common::hash::VirtualNode;
 use risingwave_common::util::epoch::test_epoch;
+use risingwave_hummock_sdk::compaction_group::StateTableId;
 use risingwave_hummock_sdk::key::{FullKey, TableKey, UserKey};
 use risingwave_hummock_sdk::key_range::KeyRange;
 use risingwave_hummock_sdk::sstable_info::{SstableInfo, SstableInfoInner};
@@ -231,6 +232,7 @@ pub async fn gen_test_sstable_impl<B: AsRef<[u8]> + Clone + Default + Eq, F: Fil
     kv_iter: impl IntoIterator<Item = (FullKey<B>, HummockValue<B>)>,
     sstable_store: SstableStoreRef,
     policy: CachePolicy,
+    table_id_to_vnode: HashMap<u32, usize>,
 ) -> SstableInfo {
     let writer_opts = SstableWriterOptions {
         capacity_hint: None,
@@ -241,10 +243,6 @@ pub async fn gen_test_sstable_impl<B: AsRef<[u8]> + Clone + Default + Eq, F: Fil
         .clone()
         .create_sst_writer(object_id, writer_opts);
 
-    let table_id_to_vnode = HashMap::from_iter(vec![(
-        TableId::default().table_id(),
-        VirtualNode::COUNT_FOR_TEST,
-    )]);
     let compaction_catalog_agent_ref = Arc::new(CompactionCatalogAgent::new(
         FilterKeyExtractorImpl::FullKey(FullKeyFilterKeyExtractor),
         table_id_to_vnode,
@@ -280,19 +278,59 @@ pub async fn gen_test_sstable<B: AsRef<[u8]> + Clone + Default + Eq>(
     object_id: HummockSstableObjectId,
     kv_iter: impl Iterator<Item = (FullKey<B>, HummockValue<B>)>,
     sstable_store: SstableStoreRef,
-) -> TableHolder {
+) -> (TableHolder, SstableInfo) {
+    let table_id_to_vnode = HashMap::from_iter(vec![(
+        TableId::default().table_id(),
+        VirtualNode::COUNT_FOR_TEST,
+    )]);
     let sst_info = gen_test_sstable_impl::<_, Xor16FilterBuilder>(
         opts,
         object_id,
         kv_iter,
         sstable_store.clone(),
         CachePolicy::NotFill,
+        table_id_to_vnode,
     )
     .await;
-    sstable_store
-        .sstable(&sst_info, &mut StoreLocalStatistic::default())
-        .await
-        .unwrap()
+
+    (
+        sstable_store
+            .sstable(&sst_info, &mut StoreLocalStatistic::default())
+            .await
+            .unwrap(),
+        sst_info,
+    )
+}
+
+pub async fn gen_test_sstable_with_table_ids<B: AsRef<[u8]> + Clone + Default + Eq>(
+    opts: SstableBuilderOptions,
+    object_id: HummockSstableObjectId,
+    kv_iter: impl Iterator<Item = (FullKey<B>, HummockValue<B>)>,
+    sstable_store: SstableStoreRef,
+    table_ids: Vec<StateTableId>,
+) -> (TableHolder, SstableInfo) {
+    let table_id_to_vnode = table_ids
+        .into_iter()
+        .map(|table_id| (table_id, VirtualNode::COUNT_FOR_TEST))
+        .collect();
+
+    let sst_info = gen_test_sstable_impl::<_, Xor16FilterBuilder>(
+        opts,
+        object_id,
+        kv_iter,
+        sstable_store.clone(),
+        CachePolicy::NotFill,
+        table_id_to_vnode,
+    )
+    .await;
+
+    (
+        sstable_store
+            .sstable(&sst_info, &mut StoreLocalStatistic::default())
+            .await
+            .unwrap(),
+        sst_info,
+    )
 }
 
 /// Generate a test table from the given `kv_iter` and put the kv value to `sstable_store`
@@ -302,12 +340,17 @@ pub async fn gen_test_sstable_info<B: AsRef<[u8]> + Clone + Default + Eq>(
     kv_iter: impl IntoIterator<Item = (FullKey<B>, HummockValue<B>)>,
     sstable_store: SstableStoreRef,
 ) -> SstableInfo {
+    let table_id_to_vnode = HashMap::from_iter(vec![(
+        TableId::default().table_id(),
+        VirtualNode::COUNT_FOR_TEST,
+    )]);
     gen_test_sstable_impl::<_, BlockedXor16FilterBuilder>(
         opts,
         object_id,
         kv_iter,
         sstable_store,
         CachePolicy::NotFill,
+        table_id_to_vnode,
     )
     .await
 }
@@ -319,12 +362,17 @@ pub async fn gen_test_sstable_with_range_tombstone(
     kv_iter: impl Iterator<Item = (FullKey<Vec<u8>>, HummockValue<Vec<u8>>)>,
     sstable_store: SstableStoreRef,
 ) -> SstableInfo {
+    let table_id_to_vnode = HashMap::from_iter(vec![(
+        TableId::default().table_id(),
+        VirtualNode::COUNT_FOR_TEST,
+    )]);
     gen_test_sstable_impl::<_, Xor16FilterBuilder>(
         opts,
         object_id,
         kv_iter,
         sstable_store.clone(),
         CachePolicy::Fill(CacheHint::Normal),
+        table_id_to_vnode,
     )
     .await
 }
@@ -367,7 +415,7 @@ pub async fn gen_default_test_sstable(
     opts: SstableBuilderOptions,
     object_id: HummockSstableObjectId,
     sstable_store: SstableStoreRef,
-) -> TableHolder {
+) -> (TableHolder, SstableInfo) {
     gen_test_sstable(
         opts,
         object_id,
diff --git a/src/storage/src/storage_failpoints/test_iterator.rs b/src/storage/src/storage_failpoints/test_iterator.rs
index ada1bef109b3..4800dd6d350d 100644
--- a/src/storage/src/storage_failpoints/test_iterator.rs
+++ b/src/storage/src/storage_failpoints/test_iterator.rs
@@ -158,7 +158,7 @@ async fn test_failpoints_merge_invalid_key() {
     fail::cfg("disable_block_cache", "return").unwrap();
     let mem_read_err = "mem_read_err";
     let sstable_store = mock_sstable_store().await;
-    let table0 = gen_iterator_test_sstable_base(
+    let (table0, sstable_info_0) = gen_iterator_test_sstable_base(
         0,
         default_builder_opt_for_test(),
         |x| x,
@@ -166,7 +166,7 @@ async fn test_failpoints_merge_invalid_key() {
         200,
     )
     .await;
-    let table1 = gen_iterator_test_sstable_base(
+    let (table1, sstable_info_1) = gen_iterator_test_sstable_base(
         1,
         default_builder_opt_for_test(),
         |x| 200 + x,
@@ -174,14 +174,15 @@ async fn test_failpoints_merge_invalid_key() {
         200,
     )
     .await;
-    let tables = vec![table0, table1];
+    let tables = vec![(table0, sstable_info_0), (table1, sstable_info_1)];
     let mut mi = MergeIterator::new({
         let mut iters = vec![];
-        for table in tables {
+        for (table, sstable_info) in tables {
             iters.push(SstableIterator::new(
                 table,
                 sstable_store.clone(),
                 Arc::new(SstableIteratorReadOptions::default()),
+                &sstable_info,
             ));
         }
         iters
@@ -206,7 +207,7 @@ async fn test_failpoints_backward_merge_invalid_key() {
     fail::cfg("disable_block_cache", "return").unwrap();
     let mem_read_err = "mem_read_err";
     let sstable_store = mock_sstable_store().await;
-    let table0 = gen_iterator_test_sstable_base(
+    let (table0, sstable_info_0) = gen_iterator_test_sstable_base(
         0,
         default_builder_opt_for_test(),
         |x| x,
@@ -214,7 +215,7 @@ async fn test_failpoints_backward_merge_invalid_key() {
         200,
     )
     .await;
-    let table1 = gen_iterator_test_sstable_base(
+    let (table1, sstable_info_1) = gen_iterator_test_sstable_base(
         1,
         default_builder_opt_for_test(),
         |x| 200 + x,
@@ -222,10 +223,10 @@ async fn test_failpoints_backward_merge_invalid_key() {
         200,
     )
     .await;
-    let tables = vec![table0, table1];
+    let tables = vec![(table0, sstable_info_0), (table1, sstable_info_1)];
     let mut mi = MergeIterator::new({
         let mut iters = vec![];
-        for table in tables {
+        for (table, _) in tables {
             iters.push(BackwardSstableIterator::new(table, sstable_store.clone()));
         }
         iters
@@ -250,7 +251,7 @@ async fn test_failpoints_user_read_err() {
     fail::cfg("disable_block_cache", "return").unwrap();
     let mem_read_err = "mem_read_err";
     let sstable_store = mock_sstable_store().await;
-    let table0 = gen_iterator_test_sstable_base(
+    let (table0, sstable_info_0) = gen_iterator_test_sstable_base(
         0,
         default_builder_opt_for_test(),
         |x| x,
@@ -258,7 +259,7 @@ async fn test_failpoints_user_read_err() {
         200,
     )
     .await;
-    let table1 = gen_iterator_test_sstable_base(
+    let (table1, sstable_info_1) = gen_iterator_test_sstable_base(
         1,
         default_builder_opt_for_test(),
         |x| 200 + x,
@@ -271,11 +272,13 @@ async fn test_failpoints_user_read_err() {
             table0,
             sstable_store.clone(),
             Arc::new(SstableIteratorReadOptions::default()),
+            &sstable_info_0,
         ),
         SstableIterator::new(
             table1,
             sstable_store.clone(),
             Arc::new(SstableIteratorReadOptions::default()),
+            &sstable_info_1,
         ),
     ];
 
@@ -310,7 +313,7 @@ async fn test_failpoints_backward_user_read_err() {
     fail::cfg("disable_block_cache", "return").unwrap();
     let mem_read_err = "mem_read_err";
     let sstable_store = mock_sstable_store().await;
-    let table0 = gen_iterator_test_sstable_base(
+    let (table0, _sstable_info_0) = gen_iterator_test_sstable_base(
         0,
         default_builder_opt_for_test(),
         |x| x,
@@ -318,7 +321,7 @@ async fn test_failpoints_backward_user_read_err() {
         200,
     )
     .await;
-    let table1 = gen_iterator_test_sstable_base(
+    let (table1, _sstable_info_1) = gen_iterator_test_sstable_base(
         1,
         default_builder_opt_for_test(),
         |x| 200 + x,
diff --git a/src/storage/src/storage_failpoints/test_sstable.rs b/src/storage/src/storage_failpoints/test_sstable.rs
index fa7bd970039f..e96f068cbe1c 100644
--- a/src/storage/src/storage_failpoints/test_sstable.rs
+++ b/src/storage/src/storage_failpoints/test_sstable.rs
@@ -40,7 +40,7 @@ async fn test_failpoints_table_read() {
     // We should close buffer, so that table iterator must read in object_stores
     let kv_iter =
         (0..TEST_KEYS_COUNT).map(|i| (test_key_of(i), HummockValue::put(test_value_of(i))));
-    let table = gen_test_sstable(
+    let (table, sstable_info) = gen_test_sstable(
         default_builder_opt_for_test(),
         0,
         kv_iter,
@@ -52,6 +52,7 @@ async fn test_failpoints_table_read() {
         table,
         sstable_store,
         Arc::new(SstableIteratorReadOptions::default()),
+        &sstable_info,
     );
     sstable_iter.rewind().await.unwrap();
 
@@ -130,6 +131,7 @@ async fn test_failpoints_vacuum_and_metadata() {
         sstable_store.sstable(&info, &mut stats).await.unwrap(),
         sstable_store,
         Arc::new(SstableIteratorReadOptions::default()),
+        &info,
     );
     let mut cnt = 0;
     sstable_iter.rewind().await.unwrap();