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

risingwave_common/config/
storage.rs
