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

risingwave_common/config/
storage.rs
