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

risingwave_common/config/
storage.rs
