feat: Add a config to disbale storage iter prefetch

src/common/src/config.rs

@@ -582,6 +582,8 @@ pub struct StorageConfig {
     pub enable_fast_compaction: bool,
     #[serde(default, flatten)]
     pub unrecognized: Unrecognized<Self>,
+    #[serde(default = "default::storage::disable_iter_prefetch")]
+    pub disable_iter_prefetch: bool,
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde)]
@@ -1062,6 +1064,10 @@ pub mod default {
         pub fn enable_fast_compaction() -> bool {
             true
         }
+
+        pub fn disable_iter_prefetch() -> bool {
+            false
+        }
     }
 
     pub mod streaming {

src/jni_core/src/hummock_iterator.rs

@@ -86,8 +86,12 @@ impl HummockJavaBindingIterator {
             FileCache::none(),
             FileCache::none(),
         ));
-        let reader =
-            HummockVersionReader::new(sstable_store, Arc::new(HummockStateStoreMetrics::unused()));
+        let disable_iter_prefetch = false;
+        let reader = HummockVersionReader::new(
+            sstable_store,
+            Arc::new(HummockStateStoreMetrics::unused()),
+            disable_iter_prefetch,
+        );
 
         let mut streams = Vec::with_capacity(read_plan.vnode_ids.len());
         let key_range = read_plan.key_range.unwrap();

src/storage/src/hummock/store/hummock_storage.rs

@@ -198,6 +198,7 @@ impl HummockStorage {
             hummock_version_reader: HummockVersionReader::new(
                 sstable_store,
                 state_store_metrics.clone(),
+                options.disable_iter_prefetch,
             ),
             _shutdown_guard: Arc::new(HummockStorageShutdownGuard {
                 shutdown_sender: event_tx,

src/storage/src/hummock/store/version.rs

@@ -53,6 +53,7 @@ use crate::mem_table::{ImmId, ImmutableMemtable};
 use crate::monitor::{
     GetLocalMetricsGuard, HummockStateStoreMetrics, MayExistLocalMetricsGuard, StoreLocalStatistic,
 };
+use crate::opts::StorageOpts;
 use crate::store::{gen_min_epoch, ReadOptions, StateStoreIterExt, StreamTypeOfIter};
 
 // TODO: use a custom data structure to allow in-place update instead of proto
@@ -514,6 +515,8 @@ pub struct HummockVersionReader {
 
     /// Statistics
     state_store_metrics: Arc<HummockStateStoreMetrics>,
+
+    disable_iter_prefetch: bool,
 }
 
 /// use `HummockVersionReader` to reuse `get` and `iter` implement for both `batch_query` and
@@ -522,10 +525,12 @@ impl HummockVersionReader {
     pub fn new(
         sstable_store: SstableStoreRef,
         state_store_metrics: Arc<HummockStateStoreMetrics>,
+        disable_iter_prefetch: bool,
     ) -> Self {
         Self {
             sstable_store,
             state_store_metrics,
+            disable_iter_prefetch,
         }
     }
 
@@ -684,9 +689,12 @@ impl HummockVersionReader {
         &self,
         table_key_range: TableKeyRange,
         epoch: u64,
-        read_options: ReadOptions,
+        mut read_options: ReadOptions,
         read_version_tuple: (Vec<ImmutableMemtable>, Vec<SstableInfo>, CommittedVersion),
     ) -> StorageResult<StreamTypeOfIter<HummockStorageIterator>> {
+        if self.disable_iter_prefetch {
+            read_options.disable_prefetch();
+        }
         let table_id_string = read_options.table_id.to_string();
         let table_id_label = table_id_string.as_str();
         let (imms, uncommitted_ssts, committed) = read_version_tuple;

src/storage/src/opts.rs

@@ -119,6 +119,9 @@ pub struct StorageOpts {
     pub compactor_max_sst_size: u64,
     /// enable FastCompactorRunner.
     pub enable_fast_compaction: bool,
+    /// false: enable block-level prefetch when `PrefetchOptions` is set
+    /// true: disable block-level prefetch
+    pub disable_iter_prefetch: bool,
 }
 
 impl Default for StorageOpts {
@@ -216,6 +219,7 @@ impl From<(&RwConfig, &SystemParamsReader, &StorageMemoryConfig)> for StorageOpt
             compactor_max_task_multiplier: c.storage.compactor_max_task_multiplier,
             compactor_max_sst_size: c.storage.compactor_max_sst_size,
             enable_fast_compaction: c.storage.enable_fast_compaction,
+            disable_iter_prefetch: c.storage.disable_iter_prefetch,
         }
     }
 }

src/storage/src/store.rs

@@ -347,6 +347,12 @@ impl From<ReadOptions> for TracedReadOptions {
     }
 }
 
+impl ReadOptions {
+    pub fn disable_prefetch(&mut self) {
+        self.prefetch_options.exhaust_iter = false;
+    }
+}
+
 pub fn gen_min_epoch(base_epoch: u64, retention_seconds: Option<&u32>) -> u64 {
     let base_epoch = Epoch(base_epoch);
     match retention_seconds {