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

risingwave_common/config/
storage.rs
