risingwave_common/config/
storage.rs

1// Copyright 2025 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use foyer::{
16    Compression, LfuConfig, LruConfig, RecoverMode, RuntimeOptions, S3FifoConfig, Throttle,
17};
18use serde::de::Error as _;
19
20use super::*;
21
22/// The section `[storage]` in `risingwave.toml`.
23#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
24#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde, ConfigDoc)]
25pub struct StorageConfig {
26    /// parallelism while syncing share buffers into L0 SST. Should NOT be 0.
27    #[serde(default = "default::storage::share_buffers_sync_parallelism")]
28    pub share_buffers_sync_parallelism: u32,
29
30    /// Worker threads number of dedicated tokio runtime for share buffer compaction. 0 means use
31    /// tokio's default value (number of CPU core).
32    #[serde(default = "default::storage::share_buffer_compaction_worker_threads_number")]
33    pub share_buffer_compaction_worker_threads_number: u32,
34
35    /// Configure the maximum shared buffer size in MB explicitly. Writes attempting to exceed the capacity
36    /// will stall until there is enough space. The overridden value will only be effective if:
37    /// 1. `block_cache_capacity_mb` and `meta_cache_capacity_mb` are also configured explicitly.
38    /// 2. `block_cache_capacity_mb` + `meta_cache_capacity_mb` + `meta_cache_capacity_mb` doesn't exceed 0.3 * non-reserved memory.
39    #[serde(default)]
40    pub shared_buffer_capacity_mb: Option<usize>,
41
42    /// The shared buffer will start flushing data to object when the ratio of memory usage to the
43    /// shared buffer capacity exceed such ratio.
44    #[serde(default = "default::storage::shared_buffer_flush_ratio")]
45    pub shared_buffer_flush_ratio: f32,
46
47    /// The minimum total flush size of shared buffer spill. When a shared buffer spilled is trigger,
48    /// the total flush size across multiple epochs should be at least higher than this size.
49    #[serde(default = "default::storage::shared_buffer_min_batch_flush_size_mb")]
50    pub shared_buffer_min_batch_flush_size_mb: usize,
51
52    /// The threshold for the number of immutable memtables to merge to a new imm.
53    #[serde(default = "default::storage::imm_merge_threshold")]
54    #[deprecated]
55    pub imm_merge_threshold: usize,
56
57    /// Whether to enable write conflict detection
58    #[serde(default = "default::storage::write_conflict_detection_enabled")]
59    pub write_conflict_detection_enabled: bool,
60
61    #[serde(default)]
62    #[config_doc(nested)]
63    pub cache: CacheConfig,
64
65    /// DEPRECATED: This config will be deprecated in the future version, use `storage.cache.block_cache_capacity_mb` instead.
66    #[serde(default)]
67    pub block_cache_capacity_mb: Option<usize>,
68
69    /// DEPRECATED: This config will be deprecated in the future version, use `storage.cache.meta_cache_capacity_mb` instead.
70    #[serde(default)]
71    pub meta_cache_capacity_mb: Option<usize>,
72
73    /// DEPRECATED: This config will be deprecated in the future version, use `storage.cache.block_cache_eviction.high_priority_ratio_in_percent` with `storage.cache.block_cache_eviction.algorithm = "Lru"` instead.
74    #[serde(default)]
75    pub high_priority_ratio_in_percent: Option<usize>,
76
77    /// max memory usage for large query
78    #[serde(default)]
79    pub prefetch_buffer_capacity_mb: Option<usize>,
80
81    #[serde(default = "default::storage::max_cached_recent_versions_number")]
82    pub max_cached_recent_versions_number: usize,
83
84    /// max prefetch block number
85    #[serde(
86        default = "default::storage::max_prefetch_block_number",
87        deserialize_with = "deserialize_max_prefetch_block_number"
88    )]
89    pub max_prefetch_block_number: usize,
90
91    #[serde(default = "default::storage::disable_remote_compactor")]
92    pub disable_remote_compactor: bool,
93
94    /// Number of tasks shared buffer can upload in parallel.
95    #[serde(default = "default::storage::share_buffer_upload_concurrency")]
96    pub share_buffer_upload_concurrency: usize,
97
98    #[serde(default)]
99    pub compactor_memory_limit_mb: Option<usize>,
100
101    /// Compactor calculates the maximum number of tasks that can be executed on the node based on
102    /// `worker_num` and `compactor_max_task_multiplier`.
103    /// `max_pull_task_count` = `worker_num` * `compactor_max_task_multiplier`
104    #[serde(default = "default::storage::compactor_max_task_multiplier")]
105    pub compactor_max_task_multiplier: f32,
106
107    /// The percentage of memory available when compactor is deployed separately.
108    /// `non_reserved_memory_bytes` = `system_memory_available_bytes` * `compactor_memory_available_proportion`
109    #[serde(default = "default::storage::compactor_memory_available_proportion")]
110    pub compactor_memory_available_proportion: f64,
111
112    /// Number of SST ids fetched from meta per RPC
113    #[serde(default = "default::storage::sstable_id_remote_fetch_number")]
114    pub sstable_id_remote_fetch_number: u32,
115
116    #[serde(default = "default::storage::min_sstable_size_mb")]
117    pub min_sstable_size_mb: u32,
118
119    #[serde(default)]
120    #[config_doc(nested)]
121    pub data_file_cache: FileCacheConfig,
122
123    #[serde(default)]
124    #[config_doc(nested)]
125    pub meta_file_cache: FileCacheConfig,
126
127    /// sst serde happens when a sst meta is written to meta disk cache.
128    /// excluding bloom filter from serde can reduce the meta disk cache entry size
129    /// and reduce the disk io throughput at the cost of making the bloom filter useless
130    #[serde(default = "default::storage::sst_skip_bloom_filter_in_serde")]
131    pub sst_skip_bloom_filter_in_serde: bool,
132
133    #[serde(default)]
134    #[config_doc(nested)]
135    pub cache_refill: CacheRefillConfig,
136
137    /// Whether to enable streaming upload for sstable.
138    #[serde(default = "default::storage::min_sst_size_for_streaming_upload")]
139    pub min_sst_size_for_streaming_upload: u64,
140
141    #[serde(default = "default::storage::max_concurrent_compaction_task_number")]
142    pub max_concurrent_compaction_task_number: u64,
143
144    #[serde(default = "default::storage::max_preload_wait_time_mill")]
145    pub max_preload_wait_time_mill: u64,
146
147    #[serde(default = "default::storage::max_version_pinning_duration_sec")]
148    pub max_version_pinning_duration_sec: u64,
149
150    #[serde(default = "default::storage::compactor_max_sst_key_count")]
151    pub compactor_max_sst_key_count: u64,
152    // DEPRECATED: This config will be deprecated in the future version, use `storage.compactor_iter_max_io_retry_times` instead.
153    #[serde(default = "default::storage::compact_iter_recreate_timeout_ms")]
154    pub compact_iter_recreate_timeout_ms: u64,
155    #[serde(default = "default::storage::compactor_max_sst_size")]
156    pub compactor_max_sst_size: u64,
157    #[serde(default = "default::storage::enable_fast_compaction")]
158    pub enable_fast_compaction: bool,
159    #[serde(default = "default::storage::check_compaction_result")]
160    pub check_compaction_result: bool,
161    #[serde(default = "default::storage::max_preload_io_retry_times")]
162    pub max_preload_io_retry_times: usize,
163    #[serde(default = "default::storage::compactor_fast_max_compact_delete_ratio")]
164    pub compactor_fast_max_compact_delete_ratio: u32,
165    #[serde(default = "default::storage::compactor_fast_max_compact_task_size")]
166    pub compactor_fast_max_compact_task_size: u64,
167    #[serde(default = "default::storage::compactor_iter_max_io_retry_times")]
168    pub compactor_iter_max_io_retry_times: usize,
169
170    /// If set, block metadata keys will be shortened when their length exceeds this threshold.
171    /// This reduces `SSTable` metadata size by storing only the minimal distinguishing prefix.
172    /// - `None`: Disabled (default)
173    /// - `Some(n)`: Only shorten keys with length >= n bytes
174    #[serde(default = "default::storage::shorten_block_meta_key_threshold")]
175    pub shorten_block_meta_key_threshold: Option<usize>,
176
177    /// Deprecated: The window size of table info statistic history.
178    #[serde(default = "default::storage::table_info_statistic_history_times")]
179    #[deprecated]
180    pub table_info_statistic_history_times: usize,
181
182    #[serde(default, flatten)]
183    #[config_doc(omitted)]
184    pub unrecognized: Unrecognized<Self>,
185
186    /// The spill threshold for mem table.
187    #[serde(default = "default::storage::mem_table_spill_threshold")]
188    pub mem_table_spill_threshold: usize,
189
190    /// The concurrent uploading number of `SSTables` of builder
191    #[serde(default = "default::storage::compactor_concurrent_uploading_sst_count")]
192    pub compactor_concurrent_uploading_sst_count: Option<usize>,
193
194    #[serde(default = "default::storage::compactor_max_overlap_sst_count")]
195    pub compactor_max_overlap_sst_count: usize,
196
197    /// The maximum number of meta files that can be preloaded.
198    /// If the number of meta files exceeds this value, the compactor will try to compute parallelism only through `SstableInfo`, no longer preloading `SstableMeta`.
199    /// This is to prevent the compactor from consuming too much memory, but it may cause the compactor to be less efficient.
200    #[serde(default = "default::storage::compactor_max_preload_meta_file_count")]
201    pub compactor_max_preload_meta_file_count: usize,
202
203    #[serde(default = "default::storage::vector_file_block_size_kb")]
204    pub vector_file_block_size_kb: usize,
205
206    /// Object storage configuration
207    /// 1. General configuration
208    /// 2. Some special configuration of Backend
209    /// 3. Retry and timeout configuration
210    #[serde(default)]
211    pub object_store: ObjectStoreConfig,
212
213    #[serde(default = "default::storage::time_travel_version_cache_capacity")]
214    pub time_travel_version_cache_capacity: u64,
215
216    #[serde(default = "default::storage::table_change_log_cache_capacity")]
217    pub table_change_log_cache_capacity: u64,
218
219    // iceberg compaction
220    #[serde(default = "default::storage::iceberg_compaction_enable_validate")]
221    pub iceberg_compaction_enable_validate: bool,
222    #[serde(default = "default::storage::iceberg_compaction_max_record_batch_rows")]
223    pub iceberg_compaction_max_record_batch_rows: usize,
224    #[serde(default = "default::storage::iceberg_compaction_min_size_per_partition_mb")]
225    pub iceberg_compaction_min_size_per_partition_mb: u32,
226    #[serde(default = "default::storage::iceberg_compaction_max_file_count_per_partition")]
227    pub iceberg_compaction_max_file_count_per_partition: u32,
228    /// DEPRECATED: This config will be deprecated in the future version.
229    /// Use sink config `compaction.write_parquet_max_row_group_rows` instead.
230    #[serde(default = "default::storage::iceberg_compaction_write_parquet_max_row_group_rows")]
231    #[deprecated(
232        note = "This config is deprecated. Use sink config `compaction.write_parquet_max_row_group_rows` instead."
233    )]
234    pub iceberg_compaction_write_parquet_max_row_group_rows: usize,
235
236    /// The ratio of iceberg compaction max parallelism to the number of CPU cores
237    #[serde(default = "default::storage::iceberg_compaction_task_parallelism_ratio")]
238    pub iceberg_compaction_task_parallelism_ratio: f32,
239    /// Whether to enable heuristic output parallelism in iceberg compaction.
240    #[serde(default = "default::storage::iceberg_compaction_enable_heuristic_output_parallelism")]
241    pub iceberg_compaction_enable_heuristic_output_parallelism: bool,
242    /// Maximum number of concurrent file close operations
243    #[serde(default = "default::storage::iceberg_compaction_max_concurrent_closes")]
244    pub iceberg_compaction_max_concurrent_closes: usize,
245    /// Whether to enable dynamic size estimation for iceberg compaction.
246    #[serde(default = "default::storage::iceberg_compaction_enable_dynamic_size_estimation")]
247    pub iceberg_compaction_enable_dynamic_size_estimation: bool,
248    /// The smoothing factor for size estimation in iceberg compaction.(default: 0.3)
249    #[serde(default = "default::storage::iceberg_compaction_size_estimation_smoothing_factor")]
250    pub iceberg_compaction_size_estimation_smoothing_factor: f64,
251    /// Multiplier for pending waiting parallelism budget for iceberg compaction task queue.
252    /// Effective pending budget = `ceil(max_task_parallelism * multiplier)`. Default 4.0.
253    /// Set < 1.0 to reduce buffering (may increase `PullTask` RPC frequency); set higher to batch more tasks.
254    #[serde(
255        default = "default::storage::iceberg_compaction_pending_parallelism_budget_multiplier"
256    )]
257    pub iceberg_compaction_pending_parallelism_budget_multiplier: f32,
258
259    #[serde(default = "default::storage::iceberg_compaction_target_binpack_group_size_mb")]
260    pub iceberg_compaction_target_binpack_group_size_mb: Option<u64>,
261    #[serde(default = "default::storage::iceberg_compaction_min_group_size_mb")]
262    pub iceberg_compaction_min_group_size_mb: Option<u64>,
263    #[serde(default = "default::storage::iceberg_compaction_min_group_file_count")]
264    pub iceberg_compaction_min_group_file_count: Option<usize>,
265}
266
267/// the section `[storage.cache]` in `risingwave.toml`.
268#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
269#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde, ConfigDoc)]
270pub struct CacheConfig {
271    /// Configure the capacity of the block cache in MB explicitly.
272    /// The overridden value will only be effective if:
273    /// 1. `meta_cache_capacity_mb` and `shared_buffer_capacity_mb` are also configured explicitly.
274    /// 2. `block_cache_capacity_mb` + `meta_cache_capacity_mb` + `meta_cache_capacity_mb` doesn't exceed 0.3 * non-reserved memory.
275    #[serde(default)]
276    pub block_cache_capacity_mb: Option<usize>,
277
278    /// Configure the number of shards in the block cache explicitly.
279    /// If not set, the shard number will be determined automatically based on cache capacity.
280    #[serde(default)]
281    pub block_cache_shard_num: Option<usize>,
282
283    #[serde(default)]
284    #[config_doc(omitted)]
285    pub block_cache_eviction: CacheEvictionConfig,
286
287    /// Configure the capacity of the block cache in MB explicitly.
288    /// The overridden value will only be effective if:
289    /// 1. `block_cache_capacity_mb` and `shared_buffer_capacity_mb` are also configured explicitly.
290    /// 2. `block_cache_capacity_mb` + `meta_cache_capacity_mb` + `meta_cache_capacity_mb` doesn't exceed 0.3 * non-reserved memory.
291    #[serde(default)]
292    pub meta_cache_capacity_mb: Option<usize>,
293
294    /// Configure the number of shards in the meta cache explicitly.
295    /// If not set, the shard number will be determined automatically based on cache capacity.
296    #[serde(default)]
297    pub meta_cache_shard_num: Option<usize>,
298
299    #[serde(default)]
300    #[config_doc(omitted)]
301    pub meta_cache_eviction: CacheEvictionConfig,
302
303    #[serde(default = "default::storage::vector_block_cache_capacity_mb")]
304    pub vector_block_cache_capacity_mb: usize,
305    #[serde(default = "default::storage::vector_block_cache_shard_num")]
306    pub vector_block_cache_shard_num: usize,
307    #[serde(default)]
308    #[config_doc(omitted)]
309    pub vector_block_cache_eviction_config: CacheEvictionConfig,
310    #[serde(default = "default::storage::vector_meta_cache_capacity_mb")]
311    pub vector_meta_cache_capacity_mb: usize,
312    #[serde(default = "default::storage::vector_meta_cache_shard_num")]
313    pub vector_meta_cache_shard_num: usize,
314    #[serde(default)]
315    #[config_doc(omitted)]
316    pub vector_meta_cache_eviction_config: CacheEvictionConfig,
317}
318
319/// the section `[storage.cache.eviction]` in `risingwave.toml`.
320#[derive(Clone, Debug, Serialize, Deserialize)]
321#[serde(tag = "algorithm")]
322pub enum CacheEvictionConfig {
323    Lru {
324        high_priority_ratio_in_percent: Option<usize>,
325    },
326    Lfu {
327        window_capacity_ratio_in_percent: Option<usize>,
328        protected_capacity_ratio_in_percent: Option<usize>,
329        cmsketch_eps: Option<f64>,
330        cmsketch_confidence: Option<f64>,
331    },
332    S3Fifo {
333        small_queue_capacity_ratio_in_percent: Option<usize>,
334        ghost_queue_capacity_ratio_in_percent: Option<usize>,
335        small_to_main_freq_threshold: Option<u8>,
336    },
337}
338
339impl Default for CacheEvictionConfig {
340    fn default() -> Self {
341        Self::Lru {
342            high_priority_ratio_in_percent: None,
343        }
344    }
345}
346
347#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
348#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde, ConfigDoc)]
349pub struct CacheRefillConfig {
350    /// Inflight meta cache refill tasks limit.
351    ///
352    /// 0 for unlimited.
353    #[serde(default = "default::cache_refill::meta_refill_concurrency")]
354    pub meta_refill_concurrency: usize,
355
356    /// `SSTable` levels to refill.
357    #[serde(default = "default::cache_refill::data_refill_levels")]
358    pub data_refill_levels: Vec<u32>,
359
360    /// Cache refill maximum timeout to apply version delta.
361    #[serde(default = "default::cache_refill::timeout_ms")]
362    pub timeout_ms: u64,
363
364    /// Inflight data cache refill tasks.
365    #[serde(default = "default::cache_refill::concurrency")]
366    pub concurrency: usize,
367
368    /// Block count that a data cache refill request fetches.
369    #[serde(default = "default::cache_refill::unit")]
370    pub unit: usize,
371
372    /// Data cache refill unit admission ratio.
373    ///
374    /// Only unit whose blocks are admitted above the ratio will be refilled.
375    #[serde(default = "default::cache_refill::threshold")]
376    pub threshold: f64,
377
378    /// Recent filter layer shards.
379    #[serde(default = "default::cache_refill::recent_filter_shards")]
380    pub recent_filter_shards: usize,
381
382    /// Recent filter layer count.
383    #[serde(default = "default::cache_refill::recent_filter_layers")]
384    pub recent_filter_layers: usize,
385
386    /// Recent filter layer rotate interval.
387    #[serde(default = "default::cache_refill::recent_filter_rotate_interval_ms")]
388    pub recent_filter_rotate_interval_ms: usize,
389
390    /// Skip check recent filter on data refill.
391    ///
392    /// This option is suitable for a single compute node or debugging.
393    #[serde(default = "default::cache_refill::skip_recent_filter")]
394    pub skip_recent_filter: bool,
395
396    #[serde(default, flatten)]
397    #[config_doc(omitted)]
398    pub unrecognized: Unrecognized<Self>,
399}
400
401/// The subsection `[storage.data_file_cache]` and `[storage.meta_file_cache]` in `risingwave.toml`.
402///
403/// It's put at [`StorageConfig::data_file_cache`] and  [`StorageConfig::meta_file_cache`].
404#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
405#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde, ConfigDoc)]
406pub struct FileCacheConfig {
407    #[serde(default = "default::file_cache::dir")]
408    pub dir: String,
409
410    #[serde(default = "default::file_cache::capacity_mb")]
411    pub capacity_mb: usize,
412
413    #[serde(default = "default::file_cache::file_capacity_mb")]
414    pub file_capacity_mb: usize,
415
416    #[serde(default = "default::file_cache::flushers")]
417    pub flushers: usize,
418
419    #[serde(default = "default::file_cache::reclaimers")]
420    pub reclaimers: usize,
421
422    #[serde(default = "default::file_cache::recover_concurrency")]
423    pub recover_concurrency: usize,
424
425    /// Deprecated soon. Please use `throttle` to do I/O throttling instead.
426    #[serde(default = "default::file_cache::insert_rate_limit_mb")]
427    pub insert_rate_limit_mb: usize,
428
429    #[serde(default = "default::file_cache::indexer_shards")]
430    pub indexer_shards: usize,
431
432    #[serde(default = "default::file_cache::compression")]
433    pub compression: Compression,
434
435    #[serde(default = "default::file_cache::flush_buffer_threshold_mb")]
436    pub flush_buffer_threshold_mb: Option<usize>,
437
438    #[serde(default = "default::file_cache::throttle")]
439    pub throttle: Throttle,
440
441    #[serde(default = "default::file_cache::fifo_probation_ratio")]
442    pub fifo_probation_ratio: f64,
443
444    /// Set the blob index size for each blob.
445    ///
446    /// A larger blob index size can hold more blob entries, but it will also increase the io size of each blob part
447    /// write.
448    ///
449    /// NOTE:
450    ///
451    /// - The size will be aligned up to a multiplier of 4K.
452    /// - Modifying this configuration will invalidate all existing file cache data.
453    ///
454    /// Default: 16 `KiB`
455    #[serde(default = "default::file_cache::blob_index_size_kb")]
456    pub blob_index_size_kb: usize,
457
458    /// Recover mode.
459    ///
460    /// Options:
461    ///
462    /// - "None": Do not recover disk cache.
463    /// - "Quiet": Recover disk cache and skip errors.
464    /// - "Strict": Recover disk cache and panic on errors.
465    ///
466    /// More details, see [`RecoverMode::None`], [`RecoverMode::Quiet`] and [`RecoverMode::Strict`],
467    #[serde(default = "default::file_cache::recover_mode")]
468    pub recover_mode: RecoverMode,
469
470    #[serde(default = "default::file_cache::runtime_config")]
471    pub runtime_config: RuntimeOptions,
472
473    #[serde(default, flatten)]
474    #[config_doc(omitted)]
475    pub unrecognized: Unrecognized<Self>,
476}
477
478/// The subsections `[storage.object_store]`.
479#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
480#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde)]
481pub struct ObjectStoreConfig {
482    // alias is for backward compatibility
483    #[serde(
484        default = "default::object_store_config::set_atomic_write_dir",
485        alias = "object_store_set_atomic_write_dir"
486    )]
487    pub set_atomic_write_dir: bool,
488
489    /// Retry and timeout configuration
490    /// Description retry strategy driven by exponential back-off
491    /// Exposes the timeout and retries of each Object store interface. Therefore, the total timeout for each interface is determined based on the interface's timeout/retry configuration and the exponential back-off policy.
492    #[serde(default)]
493    pub retry: ObjectStoreRetryConfig,
494
495    /// Some special configuration of S3 Backend
496    #[serde(default)]
497    pub s3: S3ObjectStoreConfig,
498
499    /// Maximum number of concurrent object store requests (read, `streaming_read`, metadata, etc.).
500    /// 0 means unlimited. When set to a positive value, a semaphore will be used to limit
501    /// the number of in-flight requests to the object store, preventing HTTP connection pool
502    /// contention under high concurrency.
503    #[serde(default = "default::object_store_config::object_store_req_concurrency_limit")]
504    pub req_concurrency_limit: usize,
505
506    /// Maximum number of concurrent HTTP requests used by the `OpenDAL` GCS backend.
507    /// 0 means unlimited.
508    #[serde(default = "default::object_store_config::http_concurrent_limit")]
509    pub http_concurrent_limit: usize,
510
511    // TODO: the following field will be deprecated after opendal is stabilized
512    #[serde(default = "default::object_store_config::opendal_upload_concurrency")]
513    pub opendal_upload_concurrency: usize,
514
515    // TODO: the following field will be deprecated after opendal is stabilized
516    #[serde(default)]
517    pub opendal_writer_abort_on_err: bool,
518
519    #[serde(default = "default::object_store_config::upload_part_size")]
520    pub upload_part_size: usize,
521}
522
523fn deserialize_max_prefetch_block_number<'de, D>(deserializer: D) -> Result<usize, D::Error>
524where
525    D: serde::Deserializer<'de>,
526{
527    let value = usize::deserialize(deserializer)?;
528    if value == 0 {
529        return Err(D::Error::custom(
530            "storage.max_prefetch_block_number must be greater than 0",
531        ));
532    }
533    Ok(value)
534}
535
536impl ObjectStoreConfig {
537    pub fn set_atomic_write_dir(&mut self) {
538        self.set_atomic_write_dir = true;
539    }
540}
541
542/// The subsections `[storage.object_store.s3]`.
543#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
544#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde)]
545pub struct S3ObjectStoreConfig {
546    // alias is for backward compatibility
547    #[serde(
548        default = "default::object_store_config::s3::keepalive_ms",
549        alias = "object_store_keepalive_ms"
550    )]
551    pub keepalive_ms: Option<u64>,
552    #[serde(
553        default = "default::object_store_config::s3::recv_buffer_size",
554        alias = "object_store_recv_buffer_size"
555    )]
556    pub recv_buffer_size: Option<usize>,
557    #[serde(
558        default = "default::object_store_config::s3::send_buffer_size",
559        alias = "object_store_send_buffer_size"
560    )]
561    pub send_buffer_size: Option<usize>,
562    #[serde(
563        default = "default::object_store_config::s3::nodelay",
564        alias = "object_store_nodelay"
565    )]
566    pub nodelay: Option<bool>,
567    /// For backwards compatibility, users should use `S3ObjectStoreDeveloperConfig` instead.
568    #[serde(default = "default::object_store_config::s3::developer::retry_unknown_service_error")]
569    pub retry_unknown_service_error: bool,
570    #[serde(default = "default::object_store_config::s3::identity_resolution_timeout_s")]
571    pub identity_resolution_timeout_s: u64,
572    #[serde(default)]
573    pub developer: S3ObjectStoreDeveloperConfig,
574}
575
576/// The subsections `[storage.object_store.s3.developer]`.
577#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
578#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde)]
579pub struct S3ObjectStoreDeveloperConfig {
580    /// Whether to retry s3 sdk error from which no error metadata is provided.
581    #[serde(
582        default = "default::object_store_config::s3::developer::retry_unknown_service_error",
583        alias = "object_store_retry_unknown_service_error"
584    )]
585    pub retry_unknown_service_error: bool,
586    /// An array of error codes that should be retried.
587    /// e.g. `["SlowDown", "TooManyRequests"]`
588    #[serde(
589        default = "default::object_store_config::s3::developer::retryable_service_error_codes",
590        alias = "object_store_retryable_service_error_codes"
591    )]
592    pub retryable_service_error_codes: Vec<String>,
593
594    // TODO: deprecate this config when we are completely deprecate aws sdk.
595    #[serde(default = "default::object_store_config::s3::developer::use_opendal")]
596    pub use_opendal: bool,
597}
598
599#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
600#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde)]
601pub struct ObjectStoreRetryConfig {
602    // A retry strategy driven by exponential back-off.
603    // The retry strategy is used for all object store operations.
604    /// Given a base duration for retry strategy in milliseconds.
605    #[serde(default = "default::object_store_config::object_store_req_backoff_interval_ms")]
606    pub req_backoff_interval_ms: u64,
607
608    /// The max delay interval for the retry strategy. No retry delay will be longer than this `Duration`.
609    #[serde(default = "default::object_store_config::object_store_req_backoff_max_delay_ms")]
610    pub req_backoff_max_delay_ms: u64,
611
612    /// A multiplicative factor that will be applied to the exponential back-off retry delay.
613    #[serde(default = "default::object_store_config::object_store_req_backoff_factor")]
614    pub req_backoff_factor: u64,
615
616    /// Maximum timeout for `upload` operation
617    #[serde(default = "default::object_store_config::object_store_upload_attempt_timeout_ms")]
618    pub upload_attempt_timeout_ms: u64,
619
620    /// Total counts of `upload` operation retries
621    #[serde(default = "default::object_store_config::object_store_upload_retry_attempts")]
622    pub upload_retry_attempts: usize,
623
624    /// Maximum timeout for `streaming_upload_init` and `streaming_upload`
625    #[serde(
626        default = "default::object_store_config::object_store_streaming_upload_attempt_timeout_ms"
627    )]
628    pub streaming_upload_attempt_timeout_ms: u64,
629
630    /// Total counts of `streaming_upload` operation retries
631    #[serde(
632        default = "default::object_store_config::object_store_streaming_upload_retry_attempts"
633    )]
634    pub streaming_upload_retry_attempts: usize,
635
636    /// Maximum timeout for `read` operation
637    #[serde(default = "default::object_store_config::object_store_read_attempt_timeout_ms")]
638    pub read_attempt_timeout_ms: u64,
639
640    /// Total counts of `read` operation retries
641    #[serde(default = "default::object_store_config::object_store_read_retry_attempts")]
642    pub read_retry_attempts: usize,
643
644    /// Maximum timeout for `streaming_read_init` and `streaming_read` operation
645    #[serde(
646        default = "default::object_store_config::object_store_streaming_read_attempt_timeout_ms"
647    )]
648    pub streaming_read_attempt_timeout_ms: u64,
649
650    /// Total counts of `streaming_read operation` retries
651    #[serde(default = "default::object_store_config::object_store_streaming_read_retry_attempts")]
652    pub streaming_read_retry_attempts: usize,
653
654    /// Maximum timeout for `metadata` operation
655    #[serde(default = "default::object_store_config::object_store_metadata_attempt_timeout_ms")]
656    pub metadata_attempt_timeout_ms: u64,
657
658    /// Total counts of `metadata` operation retries
659    #[serde(default = "default::object_store_config::object_store_metadata_retry_attempts")]
660    pub metadata_retry_attempts: usize,
661
662    /// Maximum timeout for `delete` operation
663    #[serde(default = "default::object_store_config::object_store_delete_attempt_timeout_ms")]
664    pub delete_attempt_timeout_ms: u64,
665
666    /// Total counts of `delete` operation retries
667    #[serde(default = "default::object_store_config::object_store_delete_retry_attempts")]
668    pub delete_retry_attempts: usize,
669
670    /// Maximum timeout for `delete_object` operation
671    #[serde(
672        default = "default::object_store_config::object_store_delete_objects_attempt_timeout_ms"
673    )]
674    pub delete_objects_attempt_timeout_ms: u64,
675
676    /// Total counts of `delete_object` operation retries
677    #[serde(default = "default::object_store_config::object_store_delete_objects_retry_attempts")]
678    pub delete_objects_retry_attempts: usize,
679
680    /// Maximum timeout for `list` operation
681    #[serde(default = "default::object_store_config::object_store_list_attempt_timeout_ms")]
682    pub list_attempt_timeout_ms: u64,
683
684    /// Total counts of `list` operation retries
685    #[serde(default = "default::object_store_config::object_store_list_retry_attempts")]
686    pub list_retry_attempts: usize,
687}
688
689#[derive(Debug, Clone)]
690pub enum EvictionConfig {
691    Lru(LruConfig),
692    Lfu(LfuConfig),
693    S3Fifo(S3FifoConfig),
694}
695
696impl EvictionConfig {
697    pub fn for_test() -> Self {
698        Self::Lru(LruConfig {
699            high_priority_pool_ratio: 0.0,
700        })
701    }
702}
703
704impl From<EvictionConfig> for foyer::EvictionConfig {
705    fn from(value: EvictionConfig) -> Self {
706        match value {
707            EvictionConfig::Lru(lru) => foyer::EvictionConfig::Lru(lru),
708            EvictionConfig::Lfu(lfu) => foyer::EvictionConfig::Lfu(lfu),
709            EvictionConfig::S3Fifo(s3fifo) => foyer::EvictionConfig::S3Fifo(s3fifo),
710        }
711    }
712}
713
714pub struct StorageMemoryConfig {
715    pub block_cache_capacity_mb: usize,
716    pub block_cache_shard_num: usize,
717    pub meta_cache_capacity_mb: usize,
718    pub meta_cache_shard_num: usize,
719    pub vector_block_cache_capacity_mb: usize,
720    pub vector_block_cache_shard_num: usize,
721    pub vector_meta_cache_capacity_mb: usize,
722    pub vector_meta_cache_shard_num: usize,
723    pub shared_buffer_capacity_mb: usize,
724    pub compactor_memory_limit_mb: usize,
725    pub prefetch_buffer_capacity_mb: usize,
726    pub block_cache_eviction_config: EvictionConfig,
727    pub meta_cache_eviction_config: EvictionConfig,
728    pub vector_block_cache_eviction_config: EvictionConfig,
729    pub vector_meta_cache_eviction_config: EvictionConfig,
730    pub block_file_cache_flush_buffer_threshold_mb: usize,
731    pub meta_file_cache_flush_buffer_threshold_mb: usize,
732}
733
734pub fn extract_storage_memory_config(s: &RwConfig) -> StorageMemoryConfig {
735    let block_cache_capacity_mb = s.storage.cache.block_cache_capacity_mb.unwrap_or(
736        // adapt to old version
737        s.storage
738            .block_cache_capacity_mb
739            .unwrap_or(default::storage::block_cache_capacity_mb()),
740    );
741    let meta_cache_capacity_mb = s.storage.cache.meta_cache_capacity_mb.unwrap_or(
742        // adapt to old version
743        s.storage
744            .block_cache_capacity_mb
745            .unwrap_or(default::storage::meta_cache_capacity_mb()),
746    );
747    let shared_buffer_capacity_mb = s
748        .storage
749        .shared_buffer_capacity_mb
750        .unwrap_or(default::storage::shared_buffer_capacity_mb());
751    let meta_cache_shard_num = s.storage.cache.meta_cache_shard_num.unwrap_or_else(|| {
752        let mut shard_bits = MAX_META_CACHE_SHARD_BITS;
753        while (meta_cache_capacity_mb >> shard_bits) < MIN_BUFFER_SIZE_PER_SHARD && shard_bits > 0 {
754            shard_bits -= 1;
755        }
756        shard_bits
757    });
758    let block_cache_shard_num = s.storage.cache.block_cache_shard_num.unwrap_or_else(|| {
759        let mut shard_bits = MAX_BLOCK_CACHE_SHARD_BITS;
760        while (block_cache_capacity_mb >> shard_bits) < MIN_BUFFER_SIZE_PER_SHARD && shard_bits > 0
761        {
762            shard_bits -= 1;
763        }
764        shard_bits
765    });
766    let compactor_memory_limit_mb = s
767        .storage
768        .compactor_memory_limit_mb
769        .unwrap_or(default::storage::compactor_memory_limit_mb());
770
771    let get_eviction_config = |c: &CacheEvictionConfig| {
772        match c {
773            CacheEvictionConfig::Lru {
774                high_priority_ratio_in_percent,
775            } => EvictionConfig::Lru(LruConfig {
776                high_priority_pool_ratio: high_priority_ratio_in_percent.unwrap_or(
777                    // adapt to old version
778                    s.storage
779                        .high_priority_ratio_in_percent
780                        .unwrap_or(default::storage::high_priority_ratio_in_percent()),
781                ) as f64
782                    / 100.0,
783            }),
784            CacheEvictionConfig::Lfu {
785                window_capacity_ratio_in_percent,
786                protected_capacity_ratio_in_percent,
787                cmsketch_eps,
788                cmsketch_confidence,
789            } => EvictionConfig::Lfu(LfuConfig {
790                window_capacity_ratio: window_capacity_ratio_in_percent
791                    .unwrap_or(default::storage::window_capacity_ratio_in_percent())
792                    as f64
793                    / 100.0,
794                protected_capacity_ratio: protected_capacity_ratio_in_percent
795                    .unwrap_or(default::storage::protected_capacity_ratio_in_percent())
796                    as f64
797                    / 100.0,
798                cmsketch_eps: cmsketch_eps.unwrap_or(default::storage::cmsketch_eps()),
799                cmsketch_confidence: cmsketch_confidence
800                    .unwrap_or(default::storage::cmsketch_confidence()),
801            }),
802            CacheEvictionConfig::S3Fifo {
803                small_queue_capacity_ratio_in_percent,
804                ghost_queue_capacity_ratio_in_percent,
805                small_to_main_freq_threshold,
806            } => EvictionConfig::S3Fifo(S3FifoConfig {
807                small_queue_capacity_ratio: small_queue_capacity_ratio_in_percent
808                    .unwrap_or(default::storage::small_queue_capacity_ratio_in_percent())
809                    as f64
810                    / 100.0,
811                ghost_queue_capacity_ratio: ghost_queue_capacity_ratio_in_percent
812                    .unwrap_or(default::storage::ghost_queue_capacity_ratio_in_percent())
813                    as f64
814                    / 100.0,
815                small_to_main_freq_threshold: small_to_main_freq_threshold
816                    .unwrap_or(default::storage::small_to_main_freq_threshold()),
817            }),
818        }
819    };
820
821    let block_cache_eviction_config = get_eviction_config(&s.storage.cache.block_cache_eviction);
822    let meta_cache_eviction_config = get_eviction_config(&s.storage.cache.meta_cache_eviction);
823    let vector_block_cache_eviction_config =
824        get_eviction_config(&s.storage.cache.vector_block_cache_eviction_config);
825    let vector_meta_cache_eviction_config =
826        get_eviction_config(&s.storage.cache.vector_meta_cache_eviction_config);
827
828    let prefetch_buffer_capacity_mb =
829        s.storage
830            .shared_buffer_capacity_mb
831            .unwrap_or(match &block_cache_eviction_config {
832                EvictionConfig::Lru(lru) => {
833                    ((1.0 - lru.high_priority_pool_ratio) * block_cache_capacity_mb as f64) as usize
834                }
835                EvictionConfig::Lfu(lfu) => {
836                    ((1.0 - lfu.protected_capacity_ratio) * block_cache_capacity_mb as f64) as usize
837                }
838                EvictionConfig::S3Fifo(s3fifo) => {
839                    (s3fifo.small_queue_capacity_ratio * block_cache_capacity_mb as f64) as usize
840                }
841            });
842
843    let block_file_cache_flush_buffer_threshold_mb = s
844        .storage
845        .data_file_cache
846        .flush_buffer_threshold_mb
847        .unwrap_or(default::storage::block_file_cache_flush_buffer_threshold_mb());
848    let meta_file_cache_flush_buffer_threshold_mb = s
849        .storage
850        .meta_file_cache
851        .flush_buffer_threshold_mb
852        .unwrap_or(default::storage::block_file_cache_flush_buffer_threshold_mb());
853
854    StorageMemoryConfig {
855        block_cache_capacity_mb,
856        block_cache_shard_num,
857        meta_cache_capacity_mb,
858        meta_cache_shard_num,
859        vector_block_cache_capacity_mb: s.storage.cache.vector_block_cache_capacity_mb,
860        vector_block_cache_shard_num: s.storage.cache.vector_block_cache_shard_num,
861        vector_meta_cache_capacity_mb: s.storage.cache.vector_meta_cache_capacity_mb,
862        vector_meta_cache_shard_num: s.storage.cache.vector_meta_cache_shard_num,
863        shared_buffer_capacity_mb,
864        compactor_memory_limit_mb,
865        prefetch_buffer_capacity_mb,
866        block_cache_eviction_config,
867        meta_cache_eviction_config,
868        vector_block_cache_eviction_config,
869        vector_meta_cache_eviction_config,
870        block_file_cache_flush_buffer_threshold_mb,
871        meta_file_cache_flush_buffer_threshold_mb,
872    }
873}
874
875pub mod default {
876
877    pub mod storage {
878        pub fn share_buffers_sync_parallelism() -> u32 {
879            1
880        }
881
882        pub fn share_buffer_compaction_worker_threads_number() -> u32 {
883            4
884        }
885
886        pub fn shared_buffer_capacity_mb() -> usize {
887            1024
888        }
889
890        pub fn shared_buffer_flush_ratio() -> f32 {
891            0.8
892        }
893
894        pub fn shared_buffer_min_batch_flush_size_mb() -> usize {
895            800
896        }
897
898        pub fn imm_merge_threshold() -> usize {
899            0 // disable
900        }
901
902        pub fn write_conflict_detection_enabled() -> bool {
903            cfg!(debug_assertions)
904        }
905
906        pub fn max_cached_recent_versions_number() -> usize {
907            60
908        }
909
910        pub fn block_cache_capacity_mb() -> usize {
911            512
912        }
913
914        pub fn high_priority_ratio_in_percent() -> usize {
915            70
916        }
917
918        pub fn window_capacity_ratio_in_percent() -> usize {
919            10
920        }
921
922        pub fn protected_capacity_ratio_in_percent() -> usize {
923            80
924        }
925
926        pub fn cmsketch_eps() -> f64 {
927            0.002
928        }
929
930        pub fn cmsketch_confidence() -> f64 {
931            0.95
932        }
933
934        pub fn small_queue_capacity_ratio_in_percent() -> usize {
935            10
936        }
937
938        pub fn ghost_queue_capacity_ratio_in_percent() -> usize {
939            1000
940        }
941
942        pub fn small_to_main_freq_threshold() -> u8 {
943            1
944        }
945
946        pub fn meta_cache_capacity_mb() -> usize {
947            128
948        }
949
950        pub fn disable_remote_compactor() -> bool {
951            false
952        }
953
954        pub fn share_buffer_upload_concurrency() -> usize {
955            8
956        }
957
958        pub fn compactor_memory_limit_mb() -> usize {
959            512
960        }
961
962        pub fn compactor_max_task_multiplier() -> f32 {
963            match std::env::var("RW_COMPACTOR_MODE")
964                .unwrap_or_default()
965                .as_str()
966            {
967                mode if mode.contains("iceberg") => 12.0000,
968                _ => 3.0000,
969            }
970        }
971
972        pub fn compactor_memory_available_proportion() -> f64 {
973            0.8
974        }
975
976        pub fn sstable_id_remote_fetch_number() -> u32 {
977            10
978        }
979
980        pub fn min_sstable_size_mb() -> u32 {
981            32
982        }
983
984        pub fn min_sst_size_for_streaming_upload() -> u64 {
985            // 32MB
986            32 * 1024 * 1024
987        }
988
989        pub fn max_concurrent_compaction_task_number() -> u64 {
990            16
991        }
992
993        pub fn max_preload_wait_time_mill() -> u64 {
994            0
995        }
996
997        pub fn max_version_pinning_duration_sec() -> u64 {
998            3 * 3600
999        }
1000
1001        pub fn compactor_max_sst_key_count() -> u64 {
1002            2 * 1024 * 1024 // 200w
1003        }
1004
1005        pub fn compact_iter_recreate_timeout_ms() -> u64 {
1006            10 * 60 * 1000
1007        }
1008
1009        pub fn compactor_iter_max_io_retry_times() -> usize {
1010            8
1011        }
1012
1013        pub fn shorten_block_meta_key_threshold() -> Option<usize> {
1014            None
1015        }
1016
1017        pub fn compactor_max_sst_size() -> u64 {
1018            512 * 1024 * 1024 // 512m
1019        }
1020
1021        pub fn enable_fast_compaction() -> bool {
1022            true
1023        }
1024
1025        pub fn check_compaction_result() -> bool {
1026            false
1027        }
1028
1029        pub fn max_preload_io_retry_times() -> usize {
1030            3
1031        }
1032
1033        pub fn mem_table_spill_threshold() -> usize {
1034            4 << 20
1035        }
1036
1037        pub fn compactor_fast_max_compact_delete_ratio() -> u32 {
1038            40
1039        }
1040
1041        pub fn compactor_fast_max_compact_task_size() -> u64 {
1042            2 * 1024 * 1024 * 1024 // 2g
1043        }
1044
1045        pub fn max_prefetch_block_number() -> usize {
1046            16
1047        }
1048
1049        pub fn compactor_concurrent_uploading_sst_count() -> Option<usize> {
1050            None
1051        }
1052
1053        pub fn compactor_max_overlap_sst_count() -> usize {
1054            64
1055        }
1056
1057        pub fn compactor_max_preload_meta_file_count() -> usize {
1058            32
1059        }
1060
1061        pub fn vector_file_block_size_kb() -> usize {
1062            1024
1063        }
1064
1065        pub fn vector_block_cache_capacity_mb() -> usize {
1066            16
1067        }
1068
1069        pub fn vector_block_cache_shard_num() -> usize {
1070            16
1071        }
1072
1073        pub fn vector_meta_cache_capacity_mb() -> usize {
1074            16
1075        }
1076
1077        pub fn vector_meta_cache_shard_num() -> usize {
1078            16
1079        }
1080
1081        // deprecated
1082        pub fn table_info_statistic_history_times() -> usize {
1083            240
1084        }
1085
1086        pub fn block_file_cache_flush_buffer_threshold_mb() -> usize {
1087            256
1088        }
1089
1090        pub fn meta_file_cache_flush_buffer_threshold_mb() -> usize {
1091            64
1092        }
1093
1094        pub fn time_travel_version_cache_capacity() -> u64 {
1095            10
1096        }
1097
1098        pub fn table_change_log_cache_capacity() -> u64 {
1099            60
1100        }
1101
1102        pub fn sst_skip_bloom_filter_in_serde() -> bool {
1103            false
1104        }
1105
1106        pub fn iceberg_compaction_enable_validate() -> bool {
1107            false
1108        }
1109
1110        pub fn iceberg_compaction_max_record_batch_rows() -> usize {
1111            1024
1112        }
1113
1114        pub fn iceberg_compaction_write_parquet_max_row_group_rows() -> usize {
1115            1024 * 100 // 100k
1116        }
1117
1118        pub fn iceberg_compaction_min_size_per_partition_mb() -> u32 {
1119            1024
1120        }
1121
1122        pub fn iceberg_compaction_max_file_count_per_partition() -> u32 {
1123            32
1124        }
1125
1126        pub fn iceberg_compaction_task_parallelism_ratio() -> f32 {
1127            4.0
1128        }
1129
1130        pub fn iceberg_compaction_enable_heuristic_output_parallelism() -> bool {
1131            false
1132        }
1133
1134        pub fn iceberg_compaction_max_concurrent_closes() -> usize {
1135            8
1136        }
1137
1138        pub fn iceberg_compaction_enable_dynamic_size_estimation() -> bool {
1139            true
1140        }
1141
1142        pub fn iceberg_compaction_size_estimation_smoothing_factor() -> f64 {
1143            0.3
1144        }
1145
1146        pub fn iceberg_compaction_pending_parallelism_budget_multiplier() -> f32 {
1147            4.0
1148        }
1149
1150        pub fn iceberg_compaction_target_binpack_group_size_mb() -> Option<u64> {
1151            Some(100 * 1024) // 100GB
1152        }
1153
1154        pub fn iceberg_compaction_min_group_size_mb() -> Option<u64> {
1155            None
1156        }
1157
1158        pub fn iceberg_compaction_min_group_file_count() -> Option<usize> {
1159            None
1160        }
1161    }
1162
1163    pub mod file_cache {
1164        use std::num::NonZeroUsize;
1165
1166        use foyer::{Compression, RecoverMode, RuntimeOptions, Throttle, TokioRuntimeOptions};
1167
1168        pub fn dir() -> String {
1169            "".to_owned()
1170        }
1171
1172        pub fn capacity_mb() -> usize {
1173            1024
1174        }
1175
1176        pub fn file_capacity_mb() -> usize {
1177            64
1178        }
1179
1180        pub fn flushers() -> usize {
1181            4
1182        }
1183
1184        pub fn reclaimers() -> usize {
1185            4
1186        }
1187
1188        pub fn recover_concurrency() -> usize {
1189            8
1190        }
1191
1192        pub fn insert_rate_limit_mb() -> usize {
1193            0
1194        }
1195
1196        pub fn indexer_shards() -> usize {
1197            64
1198        }
1199
1200        pub fn compression() -> Compression {
1201            Compression::None
1202        }
1203
1204        pub fn flush_buffer_threshold_mb() -> Option<usize> {
1205            None
1206        }
1207
1208        pub fn fifo_probation_ratio() -> f64 {
1209            0.1
1210        }
1211
1212        pub fn blob_index_size_kb() -> usize {
1213            16
1214        }
1215
1216        pub fn recover_mode() -> RecoverMode {
1217            RecoverMode::Quiet
1218        }
1219
1220        pub fn runtime_config() -> RuntimeOptions {
1221            RuntimeOptions::Unified(TokioRuntimeOptions::default())
1222        }
1223
1224        pub fn throttle() -> Throttle {
1225            Throttle::new()
1226                .with_iops_counter(foyer::IopsCounter::PerIoSize(
1227                    NonZeroUsize::new(128 * 1024).unwrap(),
1228                ))
1229                .with_read_iops(100000)
1230                .with_write_iops(100000)
1231                .with_write_throughput(1024 * 1024 * 1024)
1232                .with_read_throughput(1024 * 1024 * 1024)
1233        }
1234    }
1235
1236    pub mod cache_refill {
1237        pub fn meta_refill_concurrency() -> usize {
1238            0
1239        }
1240
1241        pub fn data_refill_levels() -> Vec<u32> {
1242            vec![]
1243        }
1244
1245        pub fn timeout_ms() -> u64 {
1246            6000
1247        }
1248
1249        pub fn concurrency() -> usize {
1250            10
1251        }
1252
1253        pub fn unit() -> usize {
1254            64
1255        }
1256
1257        pub fn threshold() -> f64 {
1258            0.5
1259        }
1260
1261        pub fn recent_filter_shards() -> usize {
1262            16
1263        }
1264
1265        pub fn recent_filter_layers() -> usize {
1266            6
1267        }
1268
1269        pub fn recent_filter_rotate_interval_ms() -> usize {
1270            10000
1271        }
1272
1273        pub fn skip_recent_filter() -> bool {
1274            false
1275        }
1276    }
1277
1278    pub mod object_store_config {
1279        const DEFAULT_REQ_BACKOFF_INTERVAL_MS: u64 = 1000; // 1s
1280        const DEFAULT_REQ_BACKOFF_MAX_DELAY_MS: u64 = 10 * 1000; // 10s
1281        const DEFAULT_REQ_MAX_RETRY_ATTEMPTS: usize = 3;
1282
1283        pub fn set_atomic_write_dir() -> bool {
1284            false
1285        }
1286
1287        pub fn object_store_req_concurrency_limit() -> usize {
1288            0
1289        }
1290
1291        pub fn http_concurrent_limit() -> usize {
1292            0
1293        }
1294
1295        pub fn object_store_req_backoff_interval_ms() -> u64 {
1296            DEFAULT_REQ_BACKOFF_INTERVAL_MS
1297        }
1298
1299        pub fn object_store_req_backoff_max_delay_ms() -> u64 {
1300            DEFAULT_REQ_BACKOFF_MAX_DELAY_MS // 10s
1301        }
1302
1303        pub fn object_store_req_backoff_factor() -> u64 {
1304            2
1305        }
1306
1307        pub fn object_store_upload_attempt_timeout_ms() -> u64 {
1308            8 * 1000 // 8s
1309        }
1310
1311        pub fn object_store_upload_retry_attempts() -> usize {
1312            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1313        }
1314
1315        // init + upload_part + finish
1316        pub fn object_store_streaming_upload_attempt_timeout_ms() -> u64 {
1317            5 * 1000 // 5s
1318        }
1319
1320        pub fn object_store_streaming_upload_retry_attempts() -> usize {
1321            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1322        }
1323
1324        // tips: depend on block_size
1325        pub fn object_store_read_attempt_timeout_ms() -> u64 {
1326            8 * 1000 // 8s
1327        }
1328
1329        pub fn object_store_read_retry_attempts() -> usize {
1330            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1331        }
1332
1333        pub fn object_store_streaming_read_attempt_timeout_ms() -> u64 {
1334            3 * 1000 // 3s
1335        }
1336
1337        pub fn object_store_streaming_read_retry_attempts() -> usize {
1338            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1339        }
1340
1341        pub fn object_store_metadata_attempt_timeout_ms() -> u64 {
1342            60 * 1000 // 1min
1343        }
1344
1345        pub fn object_store_metadata_retry_attempts() -> usize {
1346            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1347        }
1348
1349        pub fn object_store_delete_attempt_timeout_ms() -> u64 {
1350            5 * 1000
1351        }
1352
1353        pub fn object_store_delete_retry_attempts() -> usize {
1354            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1355        }
1356
1357        // tips: depend on batch size
1358        pub fn object_store_delete_objects_attempt_timeout_ms() -> u64 {
1359            5 * 1000
1360        }
1361
1362        pub fn object_store_delete_objects_retry_attempts() -> usize {
1363            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1364        }
1365
1366        pub fn object_store_list_attempt_timeout_ms() -> u64 {
1367            10 * 60 * 1000
1368        }
1369
1370        pub fn object_store_list_retry_attempts() -> usize {
1371            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1372        }
1373
1374        pub fn opendal_upload_concurrency() -> usize {
1375            256
1376        }
1377
1378        pub fn upload_part_size() -> usize {
1379            // 16m
1380            16 * 1024 * 1024
1381        }
1382
1383        pub mod s3 {
1384            const DEFAULT_IDENTITY_RESOLUTION_TIMEOUT_S: u64 = 5;
1385
1386            const DEFAULT_KEEPALIVE_MS: u64 = 600 * 1000; // 10min
1387
1388            pub fn keepalive_ms() -> Option<u64> {
1389                Some(DEFAULT_KEEPALIVE_MS) // 10min
1390            }
1391
1392            pub fn recv_buffer_size() -> Option<usize> {
1393                Some(1 << 21) // 2m
1394            }
1395
1396            pub fn send_buffer_size() -> Option<usize> {
1397                None
1398            }
1399
1400            pub fn nodelay() -> Option<bool> {
1401                Some(true)
1402            }
1403
1404            pub fn identity_resolution_timeout_s() -> u64 {
1405                DEFAULT_IDENTITY_RESOLUTION_TIMEOUT_S
1406            }
1407
1408            pub mod developer {
1409                pub fn retry_unknown_service_error() -> bool {
1410                    false
1411                }
1412
1413                pub fn retryable_service_error_codes() -> Vec<String> {
1414                    vec!["SlowDown".into(), "TooManyRequests".into()]
1415                }
1416
1417                pub fn use_opendal() -> bool {
1418                    true
1419                }
1420            }
1421        }
1422    }
1423}
risingwave_common/config/storage.rs

risingwave_common/config/
storage.rs
