chore(storage): make max_prefetch_block_number configurable (#14514)

src/common/src/config.rs

@@ -556,6 +556,10 @@ pub struct StorageConfig {
     #[serde(default)]
     pub prefetch_buffer_capacity_mb: Option<usize>,
 
+    /// max prefetch block number
+    #[serde(default = "default::storage::max_prefetch_block_number")]
+    pub max_prefetch_block_number: usize,
+
     #[serde(default = "default::storage::disable_remote_compactor")]
     pub disable_remote_compactor: bool,
 
@@ -1206,6 +1210,10 @@ pub mod default {
         pub fn compactor_fast_max_compact_task_size() -> u64 {
             2 * 1024 * 1024 * 1024 // 2g
         }
+
+        pub fn max_prefetch_block_number() -> usize {
+            16
+        }
     }
 
     pub mod streaming {

src/config/example.toml

@@ -101,6 +101,7 @@ share_buffer_compaction_worker_threads_number = 4
 shared_buffer_flush_ratio = 0.800000011920929
 imm_merge_threshold = 4
 write_conflict_detection_enabled = true
+max_prefetch_block_number = 16
 disable_remote_compactor = false
 share_buffer_upload_concurrency = 8
 compactor_max_task_multiplier = 2.5

src/ctl/src/common/hummock_service.rs

@@ -169,6 +169,7 @@ impl HummockServiceOpts {
             opts.meta_cache_capacity_mb * (1 << 20),
             0,
             opts.block_cache_capacity_mb * (1 << 20),
+            opts.max_prefetch_block_number,
             FileCache::none(),
             FileCache::none(),
             None,

src/jni_core/src/hummock_iterator.rs

@@ -73,6 +73,7 @@ impl HummockJavaBindingIterator {
             1 << 10,
             0,
             1 << 10,
+            16,
             FileCache::none(),
             FileCache::none(),
             None,

src/storage/benches/bench_compactor.rs

@@ -57,6 +57,7 @@ pub fn mock_sstable_store() -> SstableStoreRef {
         128 << 20,
         0,
         64 << 20,
+        16,
         FileCache::none(),
         FileCache::none(),
         None,

src/storage/benches/bench_multi_builder.rs

@@ -148,6 +148,7 @@ fn bench_builder(
         128 << 20,
         0,
         64 << 20,
+        16,
         FileCache::none(),
         FileCache::none(),
         None,

src/storage/hummock_test/src/bin/replay/main.rs

@@ -115,6 +115,7 @@ async fn create_replay_hummock(r: Record, args: &Args) -> Result<impl GlobalRepl
             storage_opts.meta_cache_capacity_mb * (1 << 20),
             storage_opts.high_priority_ratio,
             storage_opts.prefetch_buffer_capacity_mb * (1 << 20),
+            storage_opts.max_prefetch_block_number,
             FileCache::none(),
             FileCache::none(),
             None,

src/storage/src/hummock/iterator/test_utils.rs

@@ -66,6 +66,7 @@ pub fn mock_sstable_store_with_object_store(store: ObjectStoreRef) -> SstableSto
         64 << 20,
         0,
         64 << 20,
+        16,
         FileCache::none(),
         FileCache::none(),
         None,

src/storage/src/hummock/sstable_store.rs

@@ -172,6 +172,7 @@ pub struct SstableStore {
     recent_filter: Option<Arc<RecentFilter<(HummockSstableObjectId, usize)>>>,
     prefetch_buffer_usage: Arc<AtomicUsize>,
     prefetch_buffer_capacity: usize,
+    max_prefetch_block_number: usize,
 }
 
 impl SstableStore {
@@ -182,6 +183,7 @@ impl SstableStore {
         meta_cache_capacity: usize,
         high_priority_ratio: usize,
         prefetch_buffer_capacity: usize,
+        max_prefetch_block_number: usize,
         data_file_cache: FileCache<SstableBlockIndex, CachedBlock>,
         meta_file_cache: FileCache<HummockSstableObjectId, CachedSstable>,
         recent_filter: Option<Arc<RecentFilter<(HummockSstableObjectId, usize)>>>,
@@ -218,6 +220,7 @@ impl SstableStore {
             recent_filter,
             prefetch_buffer_usage: Arc::new(AtomicUsize::new(0)),
             prefetch_buffer_capacity,
+            max_prefetch_block_number,
         }
     }
 
@@ -239,6 +242,7 @@ impl SstableStore {
             meta_file_cache: FileCache::none(),
             prefetch_buffer_usage: Arc::new(AtomicUsize::new(0)),
             prefetch_buffer_capacity: block_cache_capacity,
+            max_prefetch_block_number: 16, /* compactor won't use this parameter, so just assign a default value. */
             recent_filter: None,
         }
     }
@@ -306,7 +310,6 @@ impl SstableStore {
         policy: CachePolicy,
         stats: &mut StoreLocalStatistic,
     ) -> HummockResult<Box<dyn BlockStream>> {
-        const MAX_PREFETCH_BLOCK: usize = 16;
         let object_id = sst.id;
         if self.prefetch_buffer_usage.load(Ordering::Acquire) > self.prefetch_buffer_capacity {
             let block = self.get(sst, block_index, policy, stats).await?;
@@ -324,7 +327,7 @@ impl SstableStore {
                 None,
             )));
         }
-        let end_index = std::cmp::min(end_index, block_index + MAX_PREFETCH_BLOCK);
+        let end_index = std::cmp::min(end_index, block_index + self.max_prefetch_block_number);
         let mut end_index = std::cmp::min(end_index, sst.meta.block_metas.len());
         let start_offset = sst.meta.block_metas[block_index].offset as usize;
         let mut min_hit_index = end_index;

src/storage/src/opts.rs

@@ -53,6 +53,9 @@ pub struct StorageOpts {
     pub high_priority_ratio: usize,
     /// max memory usage for large query.
     pub prefetch_buffer_capacity_mb: usize,
+
+    pub max_prefetch_block_number: usize,
+
     pub disable_remote_compactor: bool,
     /// Number of tasks shared buffer can upload in parallel.
     pub share_buffer_upload_concurrency: usize,
@@ -165,6 +168,7 @@ impl From<(&RwConfig, &SystemParamsReader, &StorageMemoryConfig)> for StorageOpt
             high_priority_ratio: s.high_priority_ratio_in_percent,
             block_cache_capacity_mb: s.block_cache_capacity_mb,
             prefetch_buffer_capacity_mb: s.prefetch_buffer_capacity_mb,
+            max_prefetch_block_number: c.storage.max_prefetch_block_number,
             meta_cache_capacity_mb: s.meta_cache_capacity_mb,
             disable_remote_compactor: c.storage.disable_remote_compactor,
             share_buffer_upload_concurrency: c.storage.share_buffer_upload_concurrency,

src/storage/src/store_impl.rs

@@ -617,6 +617,7 @@ impl StateStoreImpl {
                     opts.meta_cache_capacity_mb * (1 << 20),
                     opts.high_priority_ratio,
                     opts.prefetch_buffer_capacity_mb * (1 << 20),
+                    opts.max_prefetch_block_number,
                     data_file_cache,
                     meta_file_cache,
                     recent_filter,

src/tests/compaction_test/src/delete_range_runner.rs

@@ -218,6 +218,7 @@ async fn compaction_test(
         storage_memory_config.meta_cache_capacity_mb * (1 << 20),
         0,
         storage_memory_config.prefetch_buffer_capacity_mb * (1 << 20),
+        storage_opts.max_prefetch_block_number,
         FileCache::none(),
         FileCache::none(),
         None,