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

risingwave_common/config/
storage.rs
