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

risingwave_common/config/
storage.rs
