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 the SST filter from serde can reduce the meta disk cache entry size
129    /// and reduce disk IO throughput at the cost of making the SST 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    /// Pull interval for iceberg compaction task requests in milliseconds.
259    #[serde(
260        default = "default::storage::iceberg_compaction_pull_interval_ms",
261        deserialize_with = "deserialize_iceberg_compaction_pull_interval_ms"
262    )]
263    pub iceberg_compaction_pull_interval_ms: u64,
264    /// Enable prefetching entire data files before compacting them.
265    ///
266    /// When enabled, each input file is downloaded with a single HTTP GET before compaction
267    /// begins, replacing the default pattern of N+1 range reads (1 footer + N column chunks)
268    /// with a single sequential read per file. This reduces object storage READ API calls
269    /// from D×(1+N) to D per compaction cycle.
270    ///
271    /// Trade-off: higher peak memory — one full file is held in memory per concurrent
272    /// compaction task. Enable only when object storage API cost is a priority and memory headroom is
273    /// sufficient. See also the memory-protection config knobs such as
274    /// `iceberg_compaction_task_parallelism_ratio`.
275    #[serde(default = "default::storage::iceberg_compaction_enable_prefetch")]
276    pub iceberg_compaction_enable_prefetch: bool,
277
278    #[serde(default = "default::storage::iceberg_compaction_target_binpack_group_size_mb")]
279    pub iceberg_compaction_target_binpack_group_size_mb: Option<u64>,
280    #[serde(default = "default::storage::iceberg_compaction_min_group_size_mb")]
281    pub iceberg_compaction_min_group_size_mb: Option<u64>,
282    #[serde(default = "default::storage::iceberg_compaction_min_group_file_count")]
283    pub iceberg_compaction_min_group_file_count: Option<usize>,
284}
285
286/// the section `[storage.cache]` in `risingwave.toml`.
287#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
288#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde, ConfigDoc)]
289pub struct CacheConfig {
290    /// Configure the capacity of the block cache in MB explicitly.
291    /// The overridden value will only be effective if:
292    /// 1. `meta_cache_capacity_mb` and `shared_buffer_capacity_mb` are also configured explicitly.
293    /// 2. `block_cache_capacity_mb` + `meta_cache_capacity_mb` + `meta_cache_capacity_mb` doesn't exceed 0.3 * non-reserved memory.
294    #[serde(default)]
295    pub block_cache_capacity_mb: Option<usize>,
296
297    /// Configure the number of shards in the block cache explicitly.
298    /// If not set, the shard number will be determined automatically based on cache capacity.
299    #[serde(default)]
300    pub block_cache_shard_num: Option<usize>,
301
302    #[serde(default)]
303    #[config_doc(omitted)]
304    pub block_cache_eviction: CacheEvictionConfig,
305
306    /// Configure the capacity of the block cache in MB explicitly.
307    /// The overridden value will only be effective if:
308    /// 1. `block_cache_capacity_mb` and `shared_buffer_capacity_mb` are also configured explicitly.
309    /// 2. `block_cache_capacity_mb` + `meta_cache_capacity_mb` + `meta_cache_capacity_mb` doesn't exceed 0.3 * non-reserved memory.
310    #[serde(default)]
311    pub meta_cache_capacity_mb: Option<usize>,
312
313    /// Configure the number of shards in the meta cache explicitly.
314    /// If not set, the shard number will be determined automatically based on cache capacity.
315    #[serde(default)]
316    pub meta_cache_shard_num: Option<usize>,
317
318    #[serde(default)]
319    #[config_doc(omitted)]
320    pub meta_cache_eviction: CacheEvictionConfig,
321
322    #[serde(default = "default::storage::vector_block_cache_capacity_mb")]
323    pub vector_block_cache_capacity_mb: usize,
324    #[serde(default = "default::storage::vector_block_cache_shard_num")]
325    pub vector_block_cache_shard_num: usize,
326    #[serde(default)]
327    #[config_doc(omitted)]
328    pub vector_block_cache_eviction_config: CacheEvictionConfig,
329    #[serde(default = "default::storage::vector_meta_cache_capacity_mb")]
330    pub vector_meta_cache_capacity_mb: usize,
331    #[serde(default = "default::storage::vector_meta_cache_shard_num")]
332    pub vector_meta_cache_shard_num: usize,
333    #[serde(default)]
334    #[config_doc(omitted)]
335    pub vector_meta_cache_eviction_config: CacheEvictionConfig,
336}
337
338/// the section `[storage.cache.eviction]` in `risingwave.toml`.
339#[derive(Clone, Debug, Serialize, Deserialize)]
340#[serde(tag = "algorithm")]
341pub enum CacheEvictionConfig {
342    Lru {
343        high_priority_ratio_in_percent: Option<usize>,
344    },
345    Lfu {
346        window_capacity_ratio_in_percent: Option<usize>,
347        protected_capacity_ratio_in_percent: Option<usize>,
348        cmsketch_eps: Option<f64>,
349        cmsketch_confidence: Option<f64>,
350    },
351    S3Fifo {
352        small_queue_capacity_ratio_in_percent: Option<usize>,
353        ghost_queue_capacity_ratio_in_percent: Option<usize>,
354        small_to_main_freq_threshold: Option<u8>,
355    },
356}
357
358impl Default for CacheEvictionConfig {
359    fn default() -> Self {
360        Self::Lru {
361            high_priority_ratio_in_percent: None,
362        }
363    }
364}
365
366#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
367#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde, ConfigDoc)]
368pub struct CacheRefillConfig {
369    /// Inflight meta cache refill tasks limit.
370    ///
371    /// 0 for unlimited.
372    #[serde(default = "default::cache_refill::meta_refill_concurrency")]
373    pub meta_refill_concurrency: usize,
374
375    /// `SSTable` levels to refill.
376    #[serde(default = "default::cache_refill::data_refill_levels")]
377    pub data_refill_levels: Vec<u32>,
378
379    /// Cache refill maximum timeout to apply version delta.
380    #[serde(default = "default::cache_refill::timeout_ms")]
381    pub timeout_ms: u64,
382
383    /// Inflight data cache refill tasks.
384    #[serde(default = "default::cache_refill::concurrency")]
385    pub concurrency: usize,
386
387    /// Block count that a data cache refill request fetches.
388    #[serde(default = "default::cache_refill::unit")]
389    pub unit: usize,
390
391    /// Data cache refill unit admission ratio.
392    ///
393    /// Only unit whose blocks are admitted above the ratio will be refilled.
394    #[serde(default = "default::cache_refill::threshold")]
395    pub threshold: f64,
396
397    /// Recent filter layer shards.
398    #[serde(default = "default::cache_refill::recent_filter_shards")]
399    pub recent_filter_shards: usize,
400
401    /// Recent filter layer count.
402    #[serde(default = "default::cache_refill::recent_filter_layers")]
403    pub recent_filter_layers: usize,
404
405    /// Recent filter layer rotate interval.
406    #[serde(default = "default::cache_refill::recent_filter_rotate_interval_ms")]
407    pub recent_filter_rotate_interval_ms: usize,
408
409    /// Skip checking recent filter on data refill.
410    ///
411    /// This option is suitable for a single compute node or debugging.
412    #[serde(default = "default::cache_refill::skip_recent_filter")]
413    pub skip_recent_filter: bool,
414
415    /// Skip checking inheritance filter on data refill.
416    ///
417    /// The inheritance filter only runs after recent-filter admission, so this
418    /// option has no effect when `skip_recent_filter` is enabled.
419    ///
420    /// This option is suitable for a single compute node or debugging.
421    #[serde(default = "default::cache_refill::skip_inheritance_filter")]
422    pub skip_inheritance_filter: bool,
423
424    #[serde(default, flatten)]
425    #[config_doc(omitted)]
426    pub unrecognized: Unrecognized<Self>,
427}
428
429/// The subsection `[storage.data_file_cache]` and `[storage.meta_file_cache]` in `risingwave.toml`.
430///
431/// It's put at [`StorageConfig::data_file_cache`] and  [`StorageConfig::meta_file_cache`].
432#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
433#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde, ConfigDoc)]
434pub struct FileCacheConfig {
435    #[serde(default = "default::file_cache::dir")]
436    pub dir: String,
437
438    #[serde(default = "default::file_cache::capacity_mb")]
439    pub capacity_mb: usize,
440
441    #[serde(default = "default::file_cache::file_capacity_mb")]
442    pub file_capacity_mb: usize,
443
444    #[serde(default = "default::file_cache::flushers")]
445    pub flushers: usize,
446
447    #[serde(default = "default::file_cache::reclaimers")]
448    pub reclaimers: usize,
449
450    #[serde(default = "default::file_cache::recover_concurrency")]
451    pub recover_concurrency: usize,
452
453    /// Deprecated soon. Please use `throttle` to do I/O throttling instead.
454    #[serde(default = "default::file_cache::insert_rate_limit_mb")]
455    pub insert_rate_limit_mb: usize,
456
457    #[serde(default = "default::file_cache::indexer_shards")]
458    pub indexer_shards: usize,
459
460    #[serde(default = "default::file_cache::compression")]
461    pub compression: Compression,
462
463    #[serde(default = "default::file_cache::flush_buffer_threshold_mb")]
464    pub flush_buffer_threshold_mb: Option<usize>,
465
466    #[serde(default = "default::file_cache::throttle")]
467    pub throttle: Throttle,
468
469    #[serde(default = "default::file_cache::fifo_probation_ratio")]
470    pub fifo_probation_ratio: f64,
471
472    /// Set the blob index size for each blob.
473    ///
474    /// A larger blob index size can hold more blob entries, but it will also increase the io size of each blob part
475    /// write.
476    ///
477    /// NOTE:
478    ///
479    /// - The size will be aligned up to a multiplier of 4K.
480    /// - Modifying this configuration will invalidate all existing file cache data.
481    ///
482    /// Default: 16 `KiB`
483    #[serde(default = "default::file_cache::blob_index_size_kb")]
484    pub blob_index_size_kb: usize,
485
486    /// Recover mode.
487    ///
488    /// Options:
489    ///
490    /// - "None": Do not recover disk cache.
491    /// - "Quiet": Recover disk cache and skip errors.
492    /// - "Strict": Recover disk cache and panic on errors.
493    ///
494    /// More details, see [`RecoverMode::None`], [`RecoverMode::Quiet`] and [`RecoverMode::Strict`],
495    #[serde(default = "default::file_cache::recover_mode")]
496    pub recover_mode: RecoverMode,
497
498    #[serde(default = "default::file_cache::runtime_config")]
499    pub runtime_config: RuntimeOptions,
500
501    #[serde(default, flatten)]
502    #[config_doc(omitted)]
503    pub unrecognized: Unrecognized<Self>,
504}
505
506/// The subsections `[storage.object_store]`.
507#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
508#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde)]
509pub struct ObjectStoreConfig {
510    // alias is for backward compatibility
511    #[serde(
512        default = "default::object_store_config::set_atomic_write_dir",
513        alias = "object_store_set_atomic_write_dir"
514    )]
515    pub set_atomic_write_dir: bool,
516
517    /// Retry and timeout configuration
518    /// Description retry strategy driven by exponential back-off
519    /// 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.
520    #[serde(default)]
521    pub retry: ObjectStoreRetryConfig,
522
523    /// Some special configuration of S3 Backend
524    #[serde(default)]
525    pub s3: S3ObjectStoreConfig,
526
527    /// Maximum number of concurrent object store requests (read, `streaming_read`, metadata, etc.).
528    /// 0 means unlimited. When set to a positive value, a semaphore will be used to limit
529    /// the number of in-flight requests to the object store, preventing HTTP connection pool
530    /// contention under high concurrency.
531    #[serde(default = "default::object_store_config::object_store_req_concurrency_limit")]
532    pub req_concurrency_limit: usize,
533
534    /// Maximum number of concurrent HTTP requests used by the `OpenDAL` GCS backend.
535    /// 0 means unlimited.
536    #[serde(default = "default::object_store_config::http_concurrent_limit")]
537    pub http_concurrent_limit: usize,
538
539    // TODO: the following field will be deprecated after opendal is stabilized
540    #[serde(default = "default::object_store_config::opendal_upload_concurrency")]
541    pub opendal_upload_concurrency: usize,
542
543    // TODO: the following field will be deprecated after opendal is stabilized
544    #[serde(default)]
545    pub opendal_writer_abort_on_err: bool,
546
547    #[serde(default = "default::object_store_config::upload_part_size")]
548    pub upload_part_size: usize,
549}
550
551fn deserialize_max_prefetch_block_number<'de, D>(deserializer: D) -> Result<usize, D::Error>
552where
553    D: serde::Deserializer<'de>,
554{
555    let value = usize::deserialize(deserializer)?;
556    if value == 0 {
557        return Err(D::Error::custom(
558            "storage.max_prefetch_block_number must be greater than 0",
559        ));
560    }
561    Ok(value)
562}
563
564fn deserialize_iceberg_compaction_pull_interval_ms<'de, D>(deserializer: D) -> Result<u64, D::Error>
565where
566    D: serde::Deserializer<'de>,
567{
568    let value = u64::deserialize(deserializer)?;
569    if value == 0 {
570        return Err(D::Error::custom(
571            "storage.iceberg_compaction_pull_interval_ms must be greater than 0",
572        ));
573    }
574    Ok(value)
575}
576
577impl ObjectStoreConfig {
578    pub fn set_atomic_write_dir(&mut self) {
579        self.set_atomic_write_dir = true;
580    }
581}
582
583/// The subsections `[storage.object_store.s3]`.
584#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
585#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde)]
586pub struct S3ObjectStoreConfig {
587    // alias is for backward compatibility
588    #[serde(
589        default = "default::object_store_config::s3::keepalive_ms",
590        alias = "object_store_keepalive_ms"
591    )]
592    pub keepalive_ms: Option<u64>,
593    #[serde(
594        default = "default::object_store_config::s3::recv_buffer_size",
595        alias = "object_store_recv_buffer_size"
596    )]
597    pub recv_buffer_size: Option<usize>,
598    #[serde(
599        default = "default::object_store_config::s3::send_buffer_size",
600        alias = "object_store_send_buffer_size"
601    )]
602    pub send_buffer_size: Option<usize>,
603    #[serde(
604        default = "default::object_store_config::s3::nodelay",
605        alias = "object_store_nodelay"
606    )]
607    pub nodelay: Option<bool>,
608    /// For backwards compatibility, users should use `S3ObjectStoreDeveloperConfig` instead.
609    #[serde(default = "default::object_store_config::s3::developer::retry_unknown_service_error")]
610    pub retry_unknown_service_error: bool,
611    #[serde(default = "default::object_store_config::s3::identity_resolution_timeout_s")]
612    pub identity_resolution_timeout_s: u64,
613    #[serde(default)]
614    pub developer: S3ObjectStoreDeveloperConfig,
615}
616
617/// The subsections `[storage.object_store.s3.developer]`.
618#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
619#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde)]
620pub struct S3ObjectStoreDeveloperConfig {
621    /// Whether to retry s3 sdk error from which no error metadata is provided.
622    #[serde(
623        default = "default::object_store_config::s3::developer::retry_unknown_service_error",
624        alias = "object_store_retry_unknown_service_error"
625    )]
626    pub retry_unknown_service_error: bool,
627    /// An array of error codes that should be retried.
628    /// e.g. `["SlowDown", "TooManyRequests"]`
629    #[serde(
630        default = "default::object_store_config::s3::developer::retryable_service_error_codes",
631        alias = "object_store_retryable_service_error_codes"
632    )]
633    pub retryable_service_error_codes: Vec<String>,
634
635    // TODO: deprecate this config when we are completely deprecate aws sdk.
636    #[serde(default = "default::object_store_config::s3::developer::use_opendal")]
637    pub use_opendal: bool,
638}
639
640#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
641#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde)]
642pub struct ObjectStoreRetryConfig {
643    // A retry strategy driven by exponential back-off.
644    // The retry strategy is used for all object store operations.
645    /// Given a base duration for retry strategy in milliseconds.
646    #[serde(default = "default::object_store_config::object_store_req_backoff_interval_ms")]
647    pub req_backoff_interval_ms: u64,
648
649    /// The max delay interval for the retry strategy. No retry delay will be longer than this `Duration`.
650    #[serde(default = "default::object_store_config::object_store_req_backoff_max_delay_ms")]
651    pub req_backoff_max_delay_ms: u64,
652
653    /// A multiplicative factor that will be applied to the exponential back-off retry delay.
654    #[serde(default = "default::object_store_config::object_store_req_backoff_factor")]
655    pub req_backoff_factor: u64,
656
657    /// Maximum timeout for `upload` operation
658    #[serde(default = "default::object_store_config::object_store_upload_attempt_timeout_ms")]
659    pub upload_attempt_timeout_ms: u64,
660
661    /// Total counts of `upload` operation retries
662    #[serde(default = "default::object_store_config::object_store_upload_retry_attempts")]
663    pub upload_retry_attempts: usize,
664
665    /// Maximum timeout for `streaming_upload_init` and `streaming_upload`
666    #[serde(
667        default = "default::object_store_config::object_store_streaming_upload_attempt_timeout_ms"
668    )]
669    pub streaming_upload_attempt_timeout_ms: u64,
670
671    /// Total counts of `streaming_upload` operation retries
672    #[serde(
673        default = "default::object_store_config::object_store_streaming_upload_retry_attempts"
674    )]
675    pub streaming_upload_retry_attempts: usize,
676
677    /// Maximum timeout for `read` operation
678    #[serde(default = "default::object_store_config::object_store_read_attempt_timeout_ms")]
679    pub read_attempt_timeout_ms: u64,
680
681    /// Total counts of `read` operation retries
682    #[serde(default = "default::object_store_config::object_store_read_retry_attempts")]
683    pub read_retry_attempts: usize,
684
685    /// Maximum timeout for `streaming_read_init` and `streaming_read` operation
686    #[serde(
687        default = "default::object_store_config::object_store_streaming_read_attempt_timeout_ms"
688    )]
689    pub streaming_read_attempt_timeout_ms: u64,
690
691    /// Total counts of `streaming_read operation` retries
692    #[serde(default = "default::object_store_config::object_store_streaming_read_retry_attempts")]
693    pub streaming_read_retry_attempts: usize,
694
695    /// Maximum timeout for `metadata` operation
696    #[serde(default = "default::object_store_config::object_store_metadata_attempt_timeout_ms")]
697    pub metadata_attempt_timeout_ms: u64,
698
699    /// Total counts of `metadata` operation retries
700    #[serde(default = "default::object_store_config::object_store_metadata_retry_attempts")]
701    pub metadata_retry_attempts: usize,
702
703    /// Maximum timeout for `delete` operation
704    #[serde(default = "default::object_store_config::object_store_delete_attempt_timeout_ms")]
705    pub delete_attempt_timeout_ms: u64,
706
707    /// Total counts of `delete` operation retries
708    #[serde(default = "default::object_store_config::object_store_delete_retry_attempts")]
709    pub delete_retry_attempts: usize,
710
711    /// Maximum timeout for `delete_object` operation
712    #[serde(
713        default = "default::object_store_config::object_store_delete_objects_attempt_timeout_ms"
714    )]
715    pub delete_objects_attempt_timeout_ms: u64,
716
717    /// Total counts of `delete_object` operation retries
718    #[serde(default = "default::object_store_config::object_store_delete_objects_retry_attempts")]
719    pub delete_objects_retry_attempts: usize,
720
721    /// Maximum timeout for `list` operation
722    #[serde(default = "default::object_store_config::object_store_list_attempt_timeout_ms")]
723    pub list_attempt_timeout_ms: u64,
724
725    /// Total counts of `list` operation retries
726    #[serde(default = "default::object_store_config::object_store_list_retry_attempts")]
727    pub list_retry_attempts: usize,
728}
729
730#[derive(Debug, Clone)]
731pub enum EvictionConfig {
732    Lru(LruConfig),
733    Lfu(LfuConfig),
734    S3Fifo(S3FifoConfig),
735}
736
737impl EvictionConfig {
738    pub fn for_test() -> Self {
739        Self::Lru(LruConfig {
740            high_priority_pool_ratio: 0.0,
741        })
742    }
743}
744
745impl From<EvictionConfig> for foyer::EvictionConfig {
746    fn from(value: EvictionConfig) -> Self {
747        match value {
748            EvictionConfig::Lru(lru) => foyer::EvictionConfig::Lru(lru),
749            EvictionConfig::Lfu(lfu) => foyer::EvictionConfig::Lfu(lfu),
750            EvictionConfig::S3Fifo(s3fifo) => foyer::EvictionConfig::S3Fifo(s3fifo),
751        }
752    }
753}
754
755pub struct StorageMemoryConfig {
756    pub block_cache_capacity_mb: usize,
757    pub block_cache_shard_num: usize,
758    pub meta_cache_capacity_mb: usize,
759    pub meta_cache_shard_num: usize,
760    pub vector_block_cache_capacity_mb: usize,
761    pub vector_block_cache_shard_num: usize,
762    pub vector_meta_cache_capacity_mb: usize,
763    pub vector_meta_cache_shard_num: usize,
764    pub shared_buffer_capacity_mb: usize,
765    pub compactor_memory_limit_mb: usize,
766    pub prefetch_buffer_capacity_mb: usize,
767    pub block_cache_eviction_config: EvictionConfig,
768    pub meta_cache_eviction_config: EvictionConfig,
769    pub vector_block_cache_eviction_config: EvictionConfig,
770    pub vector_meta_cache_eviction_config: EvictionConfig,
771    pub block_file_cache_flush_buffer_threshold_mb: usize,
772    pub meta_file_cache_flush_buffer_threshold_mb: usize,
773}
774
775pub fn extract_storage_memory_config(s: &RwConfig) -> StorageMemoryConfig {
776    let block_cache_capacity_mb = s.storage.cache.block_cache_capacity_mb.unwrap_or(
777        // adapt to old version
778        s.storage
779            .block_cache_capacity_mb
780            .unwrap_or(default::storage::block_cache_capacity_mb()),
781    );
782    let meta_cache_capacity_mb = s.storage.cache.meta_cache_capacity_mb.unwrap_or(
783        // adapt to old version
784        s.storage
785            .block_cache_capacity_mb
786            .unwrap_or(default::storage::meta_cache_capacity_mb()),
787    );
788    let shared_buffer_capacity_mb = s
789        .storage
790        .shared_buffer_capacity_mb
791        .unwrap_or(default::storage::shared_buffer_capacity_mb());
792    let meta_cache_shard_num = s.storage.cache.meta_cache_shard_num.unwrap_or_else(|| {
793        let mut shard_bits = MAX_META_CACHE_SHARD_BITS;
794        while (meta_cache_capacity_mb >> shard_bits) < MIN_BUFFER_SIZE_PER_SHARD && shard_bits > 0 {
795            shard_bits -= 1;
796        }
797        shard_bits
798    });
799    let block_cache_shard_num = s.storage.cache.block_cache_shard_num.unwrap_or_else(|| {
800        let mut shard_bits = MAX_BLOCK_CACHE_SHARD_BITS;
801        while (block_cache_capacity_mb >> shard_bits) < MIN_BUFFER_SIZE_PER_SHARD && shard_bits > 0
802        {
803            shard_bits -= 1;
804        }
805        shard_bits
806    });
807    let compactor_memory_limit_mb = s
808        .storage
809        .compactor_memory_limit_mb
810        .unwrap_or(default::storage::compactor_memory_limit_mb());
811
812    let get_eviction_config = |c: &CacheEvictionConfig| {
813        match c {
814            CacheEvictionConfig::Lru {
815                high_priority_ratio_in_percent,
816            } => EvictionConfig::Lru(LruConfig {
817                high_priority_pool_ratio: high_priority_ratio_in_percent.unwrap_or(
818                    // adapt to old version
819                    s.storage
820                        .high_priority_ratio_in_percent
821                        .unwrap_or(default::storage::high_priority_ratio_in_percent()),
822                ) as f64
823                    / 100.0,
824            }),
825            CacheEvictionConfig::Lfu {
826                window_capacity_ratio_in_percent,
827                protected_capacity_ratio_in_percent,
828                cmsketch_eps,
829                cmsketch_confidence,
830            } => EvictionConfig::Lfu(LfuConfig {
831                window_capacity_ratio: window_capacity_ratio_in_percent
832                    .unwrap_or(default::storage::window_capacity_ratio_in_percent())
833                    as f64
834                    / 100.0,
835                protected_capacity_ratio: protected_capacity_ratio_in_percent
836                    .unwrap_or(default::storage::protected_capacity_ratio_in_percent())
837                    as f64
838                    / 100.0,
839                cmsketch_eps: cmsketch_eps.unwrap_or(default::storage::cmsketch_eps()),
840                cmsketch_confidence: cmsketch_confidence
841                    .unwrap_or(default::storage::cmsketch_confidence()),
842            }),
843            CacheEvictionConfig::S3Fifo {
844                small_queue_capacity_ratio_in_percent,
845                ghost_queue_capacity_ratio_in_percent,
846                small_to_main_freq_threshold,
847            } => EvictionConfig::S3Fifo(S3FifoConfig {
848                small_queue_capacity_ratio: small_queue_capacity_ratio_in_percent
849                    .unwrap_or(default::storage::small_queue_capacity_ratio_in_percent())
850                    as f64
851                    / 100.0,
852                ghost_queue_capacity_ratio: ghost_queue_capacity_ratio_in_percent
853                    .unwrap_or(default::storage::ghost_queue_capacity_ratio_in_percent())
854                    as f64
855                    / 100.0,
856                small_to_main_freq_threshold: small_to_main_freq_threshold
857                    .unwrap_or(default::storage::small_to_main_freq_threshold()),
858            }),
859        }
860    };
861
862    let block_cache_eviction_config = get_eviction_config(&s.storage.cache.block_cache_eviction);
863    let meta_cache_eviction_config = get_eviction_config(&s.storage.cache.meta_cache_eviction);
864    let vector_block_cache_eviction_config =
865        get_eviction_config(&s.storage.cache.vector_block_cache_eviction_config);
866    let vector_meta_cache_eviction_config =
867        get_eviction_config(&s.storage.cache.vector_meta_cache_eviction_config);
868
869    let prefetch_buffer_capacity_mb =
870        s.storage
871            .shared_buffer_capacity_mb
872            .unwrap_or(match &block_cache_eviction_config {
873                EvictionConfig::Lru(lru) => {
874                    ((1.0 - lru.high_priority_pool_ratio) * block_cache_capacity_mb as f64) as usize
875                }
876                EvictionConfig::Lfu(lfu) => {
877                    ((1.0 - lfu.protected_capacity_ratio) * block_cache_capacity_mb as f64) as usize
878                }
879                EvictionConfig::S3Fifo(s3fifo) => {
880                    (s3fifo.small_queue_capacity_ratio * block_cache_capacity_mb as f64) as usize
881                }
882            });
883
884    let block_file_cache_flush_buffer_threshold_mb = s
885        .storage
886        .data_file_cache
887        .flush_buffer_threshold_mb
888        .unwrap_or(default::storage::block_file_cache_flush_buffer_threshold_mb());
889    let meta_file_cache_flush_buffer_threshold_mb = s
890        .storage
891        .meta_file_cache
892        .flush_buffer_threshold_mb
893        .unwrap_or(default::storage::block_file_cache_flush_buffer_threshold_mb());
894
895    StorageMemoryConfig {
896        block_cache_capacity_mb,
897        block_cache_shard_num,
898        meta_cache_capacity_mb,
899        meta_cache_shard_num,
900        vector_block_cache_capacity_mb: s.storage.cache.vector_block_cache_capacity_mb,
901        vector_block_cache_shard_num: s.storage.cache.vector_block_cache_shard_num,
902        vector_meta_cache_capacity_mb: s.storage.cache.vector_meta_cache_capacity_mb,
903        vector_meta_cache_shard_num: s.storage.cache.vector_meta_cache_shard_num,
904        shared_buffer_capacity_mb,
905        compactor_memory_limit_mb,
906        prefetch_buffer_capacity_mb,
907        block_cache_eviction_config,
908        meta_cache_eviction_config,
909        vector_block_cache_eviction_config,
910        vector_meta_cache_eviction_config,
911        block_file_cache_flush_buffer_threshold_mb,
912        meta_file_cache_flush_buffer_threshold_mb,
913    }
914}
915
916pub mod default {
917
918    pub mod storage {
919        pub fn share_buffers_sync_parallelism() -> u32 {
920            1
921        }
922
923        pub fn share_buffer_compaction_worker_threads_number() -> u32 {
924            4
925        }
926
927        pub fn shared_buffer_capacity_mb() -> usize {
928            1024
929        }
930
931        pub fn shared_buffer_flush_ratio() -> f32 {
932            0.8
933        }
934
935        pub fn shared_buffer_min_batch_flush_size_mb() -> usize {
936            800
937        }
938
939        pub fn imm_merge_threshold() -> usize {
940            0 // disable
941        }
942
943        pub fn write_conflict_detection_enabled() -> bool {
944            cfg!(debug_assertions)
945        }
946
947        pub fn max_cached_recent_versions_number() -> usize {
948            60
949        }
950
951        pub fn block_cache_capacity_mb() -> usize {
952            512
953        }
954
955        pub fn high_priority_ratio_in_percent() -> usize {
956            70
957        }
958
959        pub fn window_capacity_ratio_in_percent() -> usize {
960            10
961        }
962
963        pub fn protected_capacity_ratio_in_percent() -> usize {
964            80
965        }
966
967        pub fn cmsketch_eps() -> f64 {
968            0.002
969        }
970
971        pub fn cmsketch_confidence() -> f64 {
972            0.95
973        }
974
975        pub fn small_queue_capacity_ratio_in_percent() -> usize {
976            10
977        }
978
979        pub fn ghost_queue_capacity_ratio_in_percent() -> usize {
980            1000
981        }
982
983        pub fn small_to_main_freq_threshold() -> u8 {
984            1
985        }
986
987        pub fn meta_cache_capacity_mb() -> usize {
988            128
989        }
990
991        pub fn disable_remote_compactor() -> bool {
992            false
993        }
994
995        pub fn share_buffer_upload_concurrency() -> usize {
996            8
997        }
998
999        pub fn compactor_memory_limit_mb() -> usize {
1000            512
1001        }
1002
1003        pub fn compactor_max_task_multiplier() -> f32 {
1004            match std::env::var("RW_COMPACTOR_MODE")
1005                .unwrap_or_default()
1006                .as_str()
1007            {
1008                mode if mode.contains("iceberg") => 12.0000,
1009                _ => 3.0000,
1010            }
1011        }
1012
1013        pub fn compactor_memory_available_proportion() -> f64 {
1014            0.8
1015        }
1016
1017        pub fn sstable_id_remote_fetch_number() -> u32 {
1018            10
1019        }
1020
1021        pub fn min_sstable_size_mb() -> u32 {
1022            32
1023        }
1024
1025        pub fn min_sst_size_for_streaming_upload() -> u64 {
1026            // 32MB
1027            32 * 1024 * 1024
1028        }
1029
1030        pub fn max_concurrent_compaction_task_number() -> u64 {
1031            16
1032        }
1033
1034        pub fn max_preload_wait_time_mill() -> u64 {
1035            0
1036        }
1037
1038        pub fn max_version_pinning_duration_sec() -> u64 {
1039            3 * 3600
1040        }
1041
1042        pub fn compactor_max_sst_key_count() -> u64 {
1043            2 * 1024 * 1024 // 200w
1044        }
1045
1046        pub fn compact_iter_recreate_timeout_ms() -> u64 {
1047            10 * 60 * 1000
1048        }
1049
1050        pub fn compactor_iter_max_io_retry_times() -> usize {
1051            8
1052        }
1053
1054        pub fn shorten_block_meta_key_threshold() -> Option<usize> {
1055            None
1056        }
1057
1058        pub fn compactor_max_sst_size() -> u64 {
1059            512 * 1024 * 1024 // 512m
1060        }
1061
1062        pub fn enable_fast_compaction() -> bool {
1063            true
1064        }
1065
1066        pub fn check_compaction_result() -> bool {
1067            false
1068        }
1069
1070        pub fn max_preload_io_retry_times() -> usize {
1071            3
1072        }
1073
1074        pub fn mem_table_spill_threshold() -> usize {
1075            4 << 20
1076        }
1077
1078        pub fn compactor_fast_max_compact_delete_ratio() -> u32 {
1079            40
1080        }
1081
1082        pub fn compactor_fast_max_compact_task_size() -> u64 {
1083            2 * 1024 * 1024 * 1024 // 2g
1084        }
1085
1086        pub fn max_prefetch_block_number() -> usize {
1087            16
1088        }
1089
1090        pub fn compactor_concurrent_uploading_sst_count() -> Option<usize> {
1091            None
1092        }
1093
1094        pub fn compactor_max_overlap_sst_count() -> usize {
1095            64
1096        }
1097
1098        pub fn compactor_max_preload_meta_file_count() -> usize {
1099            32
1100        }
1101
1102        pub fn vector_file_block_size_kb() -> usize {
1103            1024
1104        }
1105
1106        pub fn vector_block_cache_capacity_mb() -> usize {
1107            16
1108        }
1109
1110        pub fn vector_block_cache_shard_num() -> usize {
1111            16
1112        }
1113
1114        pub fn vector_meta_cache_capacity_mb() -> usize {
1115            16
1116        }
1117
1118        pub fn vector_meta_cache_shard_num() -> usize {
1119            16
1120        }
1121
1122        // deprecated
1123        pub fn table_info_statistic_history_times() -> usize {
1124            240
1125        }
1126
1127        pub fn block_file_cache_flush_buffer_threshold_mb() -> usize {
1128            256
1129        }
1130
1131        pub fn meta_file_cache_flush_buffer_threshold_mb() -> usize {
1132            64
1133        }
1134
1135        pub fn time_travel_version_cache_capacity() -> u64 {
1136            10
1137        }
1138
1139        pub fn table_change_log_cache_capacity() -> u64 {
1140            60
1141        }
1142
1143        pub fn sst_skip_bloom_filter_in_serde() -> bool {
1144            false
1145        }
1146
1147        pub fn iceberg_compaction_enable_validate() -> bool {
1148            false
1149        }
1150
1151        pub fn iceberg_compaction_max_record_batch_rows() -> usize {
1152            1024
1153        }
1154
1155        pub fn iceberg_compaction_write_parquet_max_row_group_rows() -> usize {
1156            1024 * 100 // 100k
1157        }
1158
1159        pub fn iceberg_compaction_min_size_per_partition_mb() -> u32 {
1160            1024
1161        }
1162
1163        pub fn iceberg_compaction_max_file_count_per_partition() -> u32 {
1164            32
1165        }
1166
1167        pub fn iceberg_compaction_task_parallelism_ratio() -> f32 {
1168            4.0
1169        }
1170
1171        pub fn iceberg_compaction_enable_heuristic_output_parallelism() -> bool {
1172            false
1173        }
1174
1175        pub fn iceberg_compaction_max_concurrent_closes() -> usize {
1176            8
1177        }
1178
1179        pub fn iceberg_compaction_enable_dynamic_size_estimation() -> bool {
1180            true
1181        }
1182
1183        pub fn iceberg_compaction_size_estimation_smoothing_factor() -> f64 {
1184            0.3
1185        }
1186
1187        pub fn iceberg_compaction_pending_parallelism_budget_multiplier() -> f32 {
1188            4.0
1189        }
1190
1191        pub fn iceberg_compaction_pull_interval_ms() -> u64 {
1192            5000
1193        }
1194
1195        pub fn iceberg_compaction_enable_prefetch() -> bool {
1196            false
1197        }
1198
1199        pub fn iceberg_compaction_target_binpack_group_size_mb() -> Option<u64> {
1200            Some(100 * 1024) // 100GB
1201        }
1202
1203        pub fn iceberg_compaction_min_group_size_mb() -> Option<u64> {
1204            None
1205        }
1206
1207        pub fn iceberg_compaction_min_group_file_count() -> Option<usize> {
1208            None
1209        }
1210    }
1211
1212    pub mod file_cache {
1213        use std::num::NonZeroUsize;
1214
1215        use foyer::{Compression, RecoverMode, RuntimeOptions, Throttle, TokioRuntimeOptions};
1216
1217        pub fn dir() -> String {
1218            "".to_owned()
1219        }
1220
1221        pub fn capacity_mb() -> usize {
1222            1024
1223        }
1224
1225        pub fn file_capacity_mb() -> usize {
1226            64
1227        }
1228
1229        pub fn flushers() -> usize {
1230            4
1231        }
1232
1233        pub fn reclaimers() -> usize {
1234            4
1235        }
1236
1237        pub fn recover_concurrency() -> usize {
1238            8
1239        }
1240
1241        pub fn insert_rate_limit_mb() -> usize {
1242            0
1243        }
1244
1245        pub fn indexer_shards() -> usize {
1246            64
1247        }
1248
1249        pub fn compression() -> Compression {
1250            Compression::None
1251        }
1252
1253        pub fn flush_buffer_threshold_mb() -> Option<usize> {
1254            None
1255        }
1256
1257        pub fn fifo_probation_ratio() -> f64 {
1258            0.1
1259        }
1260
1261        pub fn blob_index_size_kb() -> usize {
1262            16
1263        }
1264
1265        pub fn recover_mode() -> RecoverMode {
1266            RecoverMode::Quiet
1267        }
1268
1269        pub fn runtime_config() -> RuntimeOptions {
1270            RuntimeOptions::Unified(TokioRuntimeOptions::default())
1271        }
1272
1273        pub fn throttle() -> Throttle {
1274            Throttle::new()
1275                .with_iops_counter(foyer::IopsCounter::PerIoSize(
1276                    NonZeroUsize::new(128 * 1024).unwrap(),
1277                ))
1278                .with_read_iops(100000)
1279                .with_write_iops(100000)
1280                .with_write_throughput(1024 * 1024 * 1024)
1281                .with_read_throughput(1024 * 1024 * 1024)
1282        }
1283    }
1284
1285    pub mod cache_refill {
1286        pub fn meta_refill_concurrency() -> usize {
1287            0
1288        }
1289
1290        pub fn data_refill_levels() -> Vec<u32> {
1291            vec![]
1292        }
1293
1294        pub fn timeout_ms() -> u64 {
1295            6000
1296        }
1297
1298        pub fn concurrency() -> usize {
1299            10
1300        }
1301
1302        pub fn unit() -> usize {
1303            64
1304        }
1305
1306        pub fn threshold() -> f64 {
1307            0.5
1308        }
1309
1310        pub fn recent_filter_shards() -> usize {
1311            16
1312        }
1313
1314        pub fn recent_filter_layers() -> usize {
1315            6
1316        }
1317
1318        pub fn recent_filter_rotate_interval_ms() -> usize {
1319            10000
1320        }
1321
1322        pub fn skip_recent_filter() -> bool {
1323            false
1324        }
1325
1326        pub fn skip_inheritance_filter() -> bool {
1327            false
1328        }
1329    }
1330
1331    pub mod object_store_config {
1332        const DEFAULT_REQ_BACKOFF_INTERVAL_MS: u64 = 1000; // 1s
1333        const DEFAULT_REQ_BACKOFF_MAX_DELAY_MS: u64 = 10 * 1000; // 10s
1334        const DEFAULT_REQ_MAX_RETRY_ATTEMPTS: usize = 3;
1335
1336        pub fn set_atomic_write_dir() -> bool {
1337            false
1338        }
1339
1340        pub fn object_store_req_concurrency_limit() -> usize {
1341            0
1342        }
1343
1344        pub fn http_concurrent_limit() -> usize {
1345            0
1346        }
1347
1348        pub fn object_store_req_backoff_interval_ms() -> u64 {
1349            DEFAULT_REQ_BACKOFF_INTERVAL_MS
1350        }
1351
1352        pub fn object_store_req_backoff_max_delay_ms() -> u64 {
1353            DEFAULT_REQ_BACKOFF_MAX_DELAY_MS // 10s
1354        }
1355
1356        pub fn object_store_req_backoff_factor() -> u64 {
1357            2
1358        }
1359
1360        pub fn object_store_upload_attempt_timeout_ms() -> u64 {
1361            8 * 1000 // 8s
1362        }
1363
1364        pub fn object_store_upload_retry_attempts() -> usize {
1365            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1366        }
1367
1368        // init + upload_part + finish
1369        pub fn object_store_streaming_upload_attempt_timeout_ms() -> u64 {
1370            5 * 1000 // 5s
1371        }
1372
1373        pub fn object_store_streaming_upload_retry_attempts() -> usize {
1374            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1375        }
1376
1377        // tips: depend on block_size
1378        pub fn object_store_read_attempt_timeout_ms() -> u64 {
1379            8 * 1000 // 8s
1380        }
1381
1382        pub fn object_store_read_retry_attempts() -> usize {
1383            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1384        }
1385
1386        pub fn object_store_streaming_read_attempt_timeout_ms() -> u64 {
1387            3 * 1000 // 3s
1388        }
1389
1390        pub fn object_store_streaming_read_retry_attempts() -> usize {
1391            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1392        }
1393
1394        pub fn object_store_metadata_attempt_timeout_ms() -> u64 {
1395            60 * 1000 // 1min
1396        }
1397
1398        pub fn object_store_metadata_retry_attempts() -> usize {
1399            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1400        }
1401
1402        pub fn object_store_delete_attempt_timeout_ms() -> u64 {
1403            5 * 1000
1404        }
1405
1406        pub fn object_store_delete_retry_attempts() -> usize {
1407            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1408        }
1409
1410        // tips: depend on batch size
1411        pub fn object_store_delete_objects_attempt_timeout_ms() -> u64 {
1412            5 * 1000
1413        }
1414
1415        pub fn object_store_delete_objects_retry_attempts() -> usize {
1416            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1417        }
1418
1419        pub fn object_store_list_attempt_timeout_ms() -> u64 {
1420            10 * 60 * 1000
1421        }
1422
1423        pub fn object_store_list_retry_attempts() -> usize {
1424            DEFAULT_REQ_MAX_RETRY_ATTEMPTS
1425        }
1426
1427        pub fn opendal_upload_concurrency() -> usize {
1428            256
1429        }
1430
1431        pub fn upload_part_size() -> usize {
1432            // 16m
1433            16 * 1024 * 1024
1434        }
1435
1436        pub mod s3 {
1437            const DEFAULT_IDENTITY_RESOLUTION_TIMEOUT_S: u64 = 5;
1438
1439            const DEFAULT_KEEPALIVE_MS: u64 = 600 * 1000; // 10min
1440
1441            pub fn keepalive_ms() -> Option<u64> {
1442                Some(DEFAULT_KEEPALIVE_MS) // 10min
1443            }
1444
1445            pub fn recv_buffer_size() -> Option<usize> {
1446                Some(1 << 21) // 2m
1447            }
1448
1449            pub fn send_buffer_size() -> Option<usize> {
1450                None
1451            }
1452
1453            pub fn nodelay() -> Option<bool> {
1454                Some(true)
1455            }
1456
1457            pub fn identity_resolution_timeout_s() -> u64 {
1458                DEFAULT_IDENTITY_RESOLUTION_TIMEOUT_S
1459            }
1460
1461            pub mod developer {
1462                pub fn retry_unknown_service_error() -> bool {
1463                    false
1464                }
1465
1466                pub fn retryable_service_error_codes() -> Vec<String> {
1467                    vec!["SlowDown".into(), "TooManyRequests".into()]
1468                }
1469
1470                pub fn use_opendal() -> bool {
1471                    true
1472                }
1473            }
1474        }
1475    }
1476}
risingwave_common/config/storage.rs

risingwave_common/config/
storage.rs
