risingwave_common/config/
mod.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
15//! This pub module defines the structure of the configuration file `risingwave.toml`.
16//!
17//! [`RwConfig`] corresponds to the whole config file and each other config struct corresponds to a
18//! section in `risingwave.toml`.
19
20pub mod batch;
21pub use batch::BatchConfig;
22pub mod frontend;
23pub use frontend::FrontendConfig;
24pub mod hba;
25pub use hba::{AddressPattern, AuthMethod, ConnectionType, HbaConfig, HbaEntry};
26pub mod meta;
27pub use meta::{
28    CheckpointCompression, CompactionConfig, DefaultParallelism, MetaBackend, MetaConfig,
29    MetaStoreConfig,
30};
31pub mod streaming;
32pub use streaming::{AsyncStackTraceOption, StreamingConfig};
33pub mod server;
34pub use server::{HeapProfilingConfig, ServerConfig};
35
36pub use crate::session_config::SessionInitConfig;
37pub mod udf;
38pub use udf::UdfConfig;
39pub mod storage;
40pub use storage::{
41    CacheEvictionConfig, EvictionConfig, ObjectStoreConfig, StorageConfig, StorageMemoryConfig,
42    extract_storage_memory_config,
43};
44pub mod merge;
45pub mod mutate;
46pub mod none_as_empty_string;
47pub mod role;
48pub mod system;
49pub mod utils;
50use std::collections::BTreeMap;
51use std::fs;
52use std::num::NonZeroUsize;
53
54use anyhow::Context;
55use clap::ValueEnum;
56use educe::Educe;
57pub use merge::*;
58use risingwave_common_proc_macro::ConfigDoc;
59pub use risingwave_common_proc_macro::OverrideConfig;
60use risingwave_pb::meta::SystemParams;
61pub use role::*;
62use serde::{Deserialize, Serialize, Serializer};
63use serde_default::DefaultFromSerde;
64use serde_json::Value;
65pub use system::SystemConfig;
66pub use utils::*;
67
68use crate::for_all_params;
69
70/// Use the maximum value for HTTP/2 connection window size to avoid deadlock among multiplexed
71/// streams on the same connection.
72pub const MAX_CONNECTION_WINDOW_SIZE: u32 = (1 << 31) - 1;
73/// Use a large value for HTTP/2 stream window size to improve the performance of remote exchange,
74/// as we don't rely on this for back-pressure.
75pub const STREAM_WINDOW_SIZE: u32 = 32 * 1024 * 1024; // 32 MB
76
77/// [`RwConfig`] corresponds to the whole config file `risingwave.toml`. Each field corresponds to a
78/// section.
79#[derive(Educe, Clone, Serialize, Deserialize, Default, ConfigDoc)]
80#[educe(Debug)]
81pub struct RwConfig {
82    #[serde(default)]
83    #[config_doc(nested)]
84    pub server: ServerConfig,
85
86    #[serde(default)]
87    #[config_doc(nested)]
88    pub meta: MetaConfig,
89
90    #[serde(default)]
91    #[config_doc(nested)]
92    pub batch: BatchConfig,
93
94    #[serde(default)]
95    #[config_doc(nested)]
96    pub frontend: FrontendConfig,
97
98    #[serde(default)]
99    #[config_doc(nested)]
100    pub streaming: StreamingConfig,
101
102    #[serde(default)]
103    #[config_doc(nested)]
104    pub storage: StorageConfig,
105
106    #[serde(default)]
107    #[educe(Debug(ignore))]
108    #[config_doc(nested)]
109    pub system: SystemConfig,
110
111    #[serde(default)]
112    #[config_doc(nested)]
113    pub udf: UdfConfig,
114
115    #[serde(default)]
116    #[config_doc(nested)]
117    pub session_init: SessionInitConfig,
118
119    #[serde(flatten)]
120    #[config_doc(omitted)]
121    pub unrecognized: Unrecognized<Self>,
122}
123
124/// `[meta.developer.meta_compute_client_config]`
125/// `[meta.developer.meta_stream_client_config]`
126/// `[meta.developer.meta_frontend_client_config]`
127/// `[batch.developer.batch_compute_client_config]`
128/// `[batch.developer.batch_frontend_client_config]`
129/// `[streaming.developer.stream_compute_client_config]`
130#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
131#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde, ConfigDoc)]
132pub struct RpcClientConfig {
133    #[serde(default = "default::developer::rpc_client_connect_timeout_secs")]
134    pub connect_timeout_secs: u64,
135    /// Maximum concurrency when setting up an RPC client pool.
136    /// Set to 0 to keep the previous unlimited behavior.
137    #[serde(default = "default::developer::rpc_client_pool_setup_concurrency")]
138    pub pool_setup_concurrency: usize,
139}
140
141pub use risingwave_common_metrics::MetricLevel;
142
143impl RwConfig {
144    pub const fn default_connection_pool_size(&self) -> u16 {
145        self.server.connection_pool_size
146    }
147
148    /// Returns [`streaming::StreamingDeveloperConfig::exchange_connection_pool_size`] if set,
149    /// otherwise [`ServerConfig::connection_pool_size`].
150    pub fn streaming_exchange_connection_pool_size(&self) -> u16 {
151        self.streaming
152            .developer
153            .exchange_connection_pool_size
154            .unwrap_or_else(|| self.default_connection_pool_size())
155    }
156
157    /// Returns [`batch::BatchDeveloperConfig::exchange_connection_pool_size`] if set,
158    /// otherwise [`ServerConfig::connection_pool_size`].
159    pub fn batch_exchange_connection_pool_size(&self) -> u16 {
160        self.batch
161            .developer
162            .exchange_connection_pool_size
163            .unwrap_or_else(|| self.default_connection_pool_size())
164    }
165}
166
167pub mod default {
168
169    pub mod developer {
170        use crate::config::streaming::CacheRefillPolicy;
171
172        pub fn meta_cached_traces_num() -> u32 {
173            256
174        }
175
176        pub fn meta_cached_traces_memory_limit_bytes() -> usize {
177            1 << 27 // 128 MiB
178        }
179
180        pub fn batch_output_channel_size() -> usize {
181            64
182        }
183
184        pub fn batch_receiver_channel_size() -> usize {
185            1000
186        }
187
188        pub fn batch_root_stage_channel_size() -> usize {
189            100
190        }
191
192        pub fn batch_chunk_size() -> usize {
193            1024
194        }
195
196        pub fn batch_local_execute_buffer_size() -> usize {
197            64
198        }
199
200        /// Default to unset to be compatible with the behavior before this config is introduced,
201        /// that is, follow the value of `server.connection_pool_size`.
202        pub fn batch_exchange_connection_pool_size() -> Option<u16> {
203            None
204        }
205
206        pub fn stream_enable_executor_row_count() -> bool {
207            false
208        }
209
210        pub fn connector_message_buffer_size() -> usize {
211            16
212        }
213
214        pub fn unsafe_stream_extreme_cache_size() -> usize {
215            10
216        }
217
218        pub fn stream_topn_cache_min_capacity() -> usize {
219            10
220        }
221
222        pub fn stream_chunk_size() -> usize {
223            256
224        }
225
226        pub fn stream_exchange_initial_permits() -> usize {
227            2048
228        }
229
230        pub fn stream_exchange_batched_permits() -> usize {
231            256
232        }
233
234        pub fn stream_exchange_concurrent_barriers() -> usize {
235            1
236        }
237
238        pub fn stream_exchange_concurrent_dispatchers() -> usize {
239            0
240        }
241
242        pub fn stream_project_expr_concurrency() -> usize {
243            1
244        }
245
246        pub fn stream_project_expr_inflight_request_concurrency() -> usize {
247            0
248        }
249
250        pub fn stream_dml_channel_initial_permits() -> usize {
251            32768
252        }
253
254        pub fn stream_max_barrier_batch_size() -> u32 {
255            1024
256        }
257
258        pub fn stream_hash_agg_max_dirty_groups_heap_size() -> usize {
259            64 << 20 // 64MB
260        }
261
262        pub fn enable_trivial_move() -> bool {
263            true
264        }
265
266        pub fn enable_check_task_level_overlap() -> bool {
267            false
268        }
269
270        pub fn max_trivial_move_task_count_per_loop() -> usize {
271            256
272        }
273
274        pub fn max_get_task_probe_times() -> usize {
275            5
276        }
277
278        pub fn actor_cnt_per_worker_parallelism_soft_limit() -> usize {
279            100
280        }
281
282        pub fn actor_cnt_per_worker_parallelism_hard_limit() -> usize {
283            400
284        }
285
286        pub fn hummock_time_travel_sst_info_fetch_batch_size() -> usize {
287            10_000
288        }
289
290        pub fn hummock_time_travel_sst_info_insert_batch_size() -> usize {
291            100
292        }
293
294        pub fn time_travel_vacuum_interval_sec() -> u64 {
295            30
296        }
297
298        pub fn time_travel_vacuum_max_version_count() -> Option<u32> {
299            Some(10000)
300        }
301
302        pub fn hummock_time_travel_epoch_version_insert_batch_size() -> usize {
303            1000
304        }
305
306        pub fn hummock_time_travel_delta_fetch_batch_size() -> usize {
307            100
308        }
309
310        pub fn hummock_gc_history_insert_batch_size() -> usize {
311            1000
312        }
313
314        pub fn hummock_time_travel_filter_out_objects_batch_size() -> usize {
315            1000
316        }
317
318        pub fn hummock_time_travel_filter_out_objects_v1() -> bool {
319            false
320        }
321
322        pub fn hummock_time_travel_filter_out_objects_list_version_batch_size() -> usize {
323            10
324        }
325
326        pub fn hummock_time_travel_filter_out_objects_list_delta_batch_size() -> usize {
327            1000
328        }
329
330        pub fn memory_controller_threshold_aggressive() -> f64 {
331            0.9
332        }
333
334        pub fn memory_controller_threshold_graceful() -> f64 {
335            0.81
336        }
337
338        pub fn memory_controller_threshold_stable() -> f64 {
339            0.72
340        }
341
342        pub fn memory_controller_eviction_factor_aggressive() -> f64 {
343            2.0
344        }
345
346        pub fn memory_controller_eviction_factor_graceful() -> f64 {
347            1.5
348        }
349
350        pub fn memory_controller_eviction_factor_stable() -> f64 {
351            1.0
352        }
353
354        pub fn memory_controller_update_interval_ms() -> usize {
355            100
356        }
357
358        pub fn memory_controller_sequence_tls_step() -> u64 {
359            128
360        }
361
362        pub fn memory_controller_sequence_tls_lag() -> u64 {
363            32
364        }
365
366        pub fn stream_enable_arrangement_backfill() -> bool {
367            true
368        }
369
370        pub fn stream_enable_snapshot_backfill() -> bool {
371            true
372        }
373
374        pub fn enable_shared_source() -> bool {
375            true
376        }
377
378        pub fn stream_high_join_amplification_threshold() -> usize {
379            2048
380        }
381
382        pub fn stream_high_gap_fill_amplification_threshold() -> usize {
383            2048
384        }
385
386        /// Default to 1 to be compatible with the behavior before this config is introduced.
387        pub fn stream_exchange_connection_pool_size() -> Option<u16> {
388            Some(1)
389        }
390
391        pub fn enable_actor_tokio_metrics() -> bool {
392            true
393        }
394
395        pub fn stream_enable_auto_schema_change() -> bool {
396            true
397        }
398
399        pub fn switch_jdbc_pg_to_native() -> bool {
400            false
401        }
402
403        pub fn streaming_hash_join_entry_state_max_rows() -> usize {
404            // NOTE(kwannoel): This is just an arbitrary number.
405            30000
406        }
407
408        pub fn streaming_join_hash_map_evict_interval_rows() -> u32 {
409            16
410        }
411
412        pub fn streaming_now_progress_ratio() -> Option<f32> {
413            None
414        }
415
416        pub fn stream_snapshot_iter_rebuild_interval_secs() -> u64 {
417            10 * 60
418        }
419
420        pub fn enable_explain_analyze_stats() -> bool {
421            true
422        }
423
424        pub fn rpc_client_connect_timeout_secs() -> u64 {
425            5
426        }
427
428        pub fn rpc_client_pool_setup_concurrency() -> usize {
429            0
430        }
431
432        pub fn iceberg_list_interval_sec() -> u64 {
433            10
434        }
435
436        pub fn iceberg_fetch_batch_size() -> u64 {
437            1024
438        }
439
440        pub fn iceberg_sink_positional_delete_cache_size() -> usize {
441            1024
442        }
443
444        pub fn iceberg_sink_write_parquet_max_row_group_rows() -> usize {
445            100_000
446        }
447
448        pub fn iceberg_pk_index_sink_lookup_concurrency() -> usize {
449            64
450        }
451
452        pub fn iceberg_pk_index_sink_max_batch_rows() -> usize {
453            4096
454        }
455
456        pub fn materialize_force_overwrite_on_no_check() -> bool {
457            false
458        }
459
460        pub fn refresh_scheduler_interval_sec() -> u64 {
461            60
462        }
463
464        pub fn sync_log_store_pause_duration_ms() -> usize {
465            64
466        }
467
468        pub fn sync_log_store_buffer_size() -> usize {
469            2048
470        }
471
472        pub fn disable_sync_log_store_dispatcher() -> bool {
473            false
474        }
475
476        pub fn table_change_log_insert_batch_size() -> u64 {
477            1000
478        }
479
480        pub fn table_change_log_delete_batch_size() -> u64 {
481            1000
482        }
483
484        pub fn table_change_log_truncate_interval_sec() -> u64 {
485            600
486        }
487
488        pub fn enable_state_table_vnode_stats_pruning() -> bool {
489            false
490        }
491
492        pub fn enable_locality_sort_buffer() -> bool {
493            true
494        }
495
496        pub fn locality_sort_buffer_activate_threshold() -> u64 {
497            65536
498        }
499
500        pub fn cache_refill_policy() -> CacheRefillPolicy {
501            CacheRefillPolicy::Enabled
502        }
503
504        pub fn enable_vnode_key_stats_for_materialize() -> bool {
505            false
506        }
507
508        pub fn max_concurrent_kv_log_store_historical_read() -> usize {
509            0
510        }
511    }
512}
513
514pub const MAX_META_CACHE_SHARD_BITS: usize = 4;
515pub const MIN_BUFFER_SIZE_PER_SHARD: usize = 256;
516pub const MAX_BLOCK_CACHE_SHARD_BITS: usize = 6; // It means that there will be 64 shards lru-cache to avoid lock conflict.
517
518#[cfg(test)]
519pub mod tests {
520    use expect_test::expect;
521    use risingwave_license::LicenseKey;
522
523    use super::*;
524
525    fn default_config_for_docs() -> RwConfig {
526        let mut config = RwConfig::default();
527        // Set `license_key` to empty in the docs to avoid any confusion.
528        config.system.license_key = Some(LicenseKey::empty());
529        // Keep generated docs and example config aligned with the production-safe default.
530        config.frontend.unsafe_enable_local_fs_connector = false;
531        config
532    }
533
534    /// This test ensures that `config/example.toml` is up-to-date with the default values specified
535    /// in this file. Developer should run `./risedev generate-example-config` to update it if this
536    /// test fails.
537    #[test]
538    fn test_example_up_to_date() {
539        const HEADER: &str = "# This file is generated by ./risedev generate-example-config
540# Check detailed comments in src/common/src/config.rs";
541
542        let actual = expect_test::expect_file!["../../../config/example.toml"];
543        let default = toml::to_string(&default_config_for_docs()).expect("failed to serialize");
544
545        let expected = format!("{HEADER}\n\n{default}");
546        actual.assert_eq(&expected);
547
548        let expected = rw_config_to_markdown();
549        let actual = expect_test::expect_file!["../../../config/docs.md"];
550        actual.assert_eq(&expected);
551    }
552
553    #[test]
554    fn test_sql_option_redaction_allowlist_config() {
555        let config: RwConfig = toml::from_str("").unwrap();
556        assert_eq!(config.batch.redact_sql_option_allowlist, ["primary_key"]);
557
558        let config: RwConfig = toml::from_str("[batch]\nredact_sql_option_allowlist = []").unwrap();
559        assert!(config.batch.redact_sql_option_allowlist.is_empty());
560
561        let config: RwConfig =
562            toml::from_str("[batch]\nredact_sql_option_allowlist = ['partition_key']").unwrap();
563        assert_eq!(config.batch.redact_sql_option_allowlist, ["partition_key"]);
564    }
565
566    #[test]
567    fn test_session_init_entries_distinguishes_omitted_from_default() {
568        let config: RwConfig = toml::from_str(
569            r#"
570            [session_init]
571            streaming_parallelism = "bounded(8)"
572            streaming_parallelism_for_table = "default"
573            "#,
574        )
575        .unwrap();
576
577        // Omitted fields are `None`; an explicit `default` is `Some("default")`.
578        assert_eq!(
579            config.session_init.streaming_parallelism.as_deref(),
580            Some("bounded(8)")
581        );
582        assert_eq!(
583            config
584                .session_init
585                .streaming_parallelism_for_table
586                .as_deref(),
587            Some("default")
588        );
589        assert_eq!(config.session_init.streaming_parallelism_for_sink, None);
590
591        // Only explicitly-configured parameters are reported, by their session parameter name.
592        assert_eq!(
593            config.session_init.entries(),
594            vec![
595                ("streaming_parallelism", "bounded(8)"),
596                ("streaming_parallelism_for_table", "default"),
597            ]
598        );
599    }
600
601    #[test]
602    fn test_session_init_rejects_unrecognized_key() {
603        let err = toml::from_str::<RwConfig>(
604            r#"
605            [session_init]
606            streaming_parallelism = "bounded(8)"
607            not_a_real_param = "oops"
608            "#,
609        )
610        .unwrap_err();
611
612        assert!(err.to_string().contains("unknown field `not_a_real_param`"));
613    }
614
615    #[derive(Debug)]
616    struct ConfigItemDoc {
617        desc: String,
618        default: String,
619    }
620
621    fn rw_config_to_markdown() -> String {
622        let mut config_rustdocs = BTreeMap::<String, Vec<(String, String)>>::new();
623        RwConfig::config_docs("".to_owned(), &mut config_rustdocs);
624
625        // Section -> Config Name -> ConfigItemDoc
626        let mut configs: BTreeMap<String, BTreeMap<String, ConfigItemDoc>> = config_rustdocs
627            .into_iter()
628            .map(|(k, v)| {
629                let docs: BTreeMap<String, ConfigItemDoc> = v
630                    .into_iter()
631                    .map(|(name, desc)| {
632                        (
633                            name,
634                            ConfigItemDoc {
635                                desc,
636                                default: "".to_owned(), // unset
637                            },
638                        )
639                    })
640                    .collect();
641                (k, docs)
642            })
643            .collect();
644
645        let toml_doc: BTreeMap<String, toml::Value> =
646            toml::from_str(&toml::to_string(&default_config_for_docs()).unwrap()).unwrap();
647        toml_doc.into_iter().for_each(|(name, value)| {
648            set_default_values("".to_owned(), name, value, &mut configs);
649        });
650
651        let mut markdown = "# RisingWave System Configurations\n\n".to_owned()
652            + "This page is automatically generated by `./risedev generate-example-config`\n";
653        for (section, configs) in configs {
654            if configs.is_empty() {
655                continue;
656            }
657            markdown.push_str(&format!("\n## {}\n\n", section));
658            markdown.push_str("| Config | Description | Default |\n");
659            markdown.push_str("|--------|-------------|---------|\n");
660            for (config, doc) in configs {
661                markdown.push_str(&format!(
662                    "| {} | {} | {} |\n",
663                    config, doc.desc, doc.default
664                ));
665            }
666        }
667        markdown
668    }
669
670    fn set_default_values(
671        section: String,
672        name: String,
673        value: toml::Value,
674        configs: &mut BTreeMap<String, BTreeMap<String, ConfigItemDoc>>,
675    ) {
676        // Set the default value if it's a config name-value pair, otherwise it's a sub-section (Table) that should be recursively processed.
677        if let toml::Value::Table(table) = value {
678            let section_configs: BTreeMap<String, toml::Value> = table.into_iter().collect();
679            let sub_section = if section.is_empty() {
680                name
681            } else {
682                format!("{}.{}", section, name)
683            };
684            section_configs
685                .into_iter()
686                .for_each(|(k, v)| set_default_values(sub_section.clone(), k, v, configs))
687        } else if let Some(t) = configs.get_mut(&section)
688            && let Some(item_doc) = t.get_mut(&name)
689        {
690            item_doc.default = format!("{}", value);
691        }
692    }
693
694    #[test]
695    fn test_object_store_configs_backward_compatibility() {
696        // Define configs with the old name and make sure it still works
697        {
698            let config: RwConfig = toml::from_str(
699                r#"
700            [storage.object_store]
701            object_store_set_atomic_write_dir = true
702
703            [storage.object_store.s3]
704            object_store_keepalive_ms = 1
705            object_store_send_buffer_size = 1
706            object_store_recv_buffer_size = 1
707            object_store_nodelay = false
708
709            [storage.object_store.s3.developer]
710            object_store_retry_unknown_service_error = true
711            object_store_retryable_service_error_codes = ['dummy']
712
713
714            "#,
715            )
716            .unwrap();
717
718            assert!(config.storage.object_store.set_atomic_write_dir);
719            assert_eq!(config.storage.object_store.s3.keepalive_ms, Some(1));
720            assert_eq!(config.storage.object_store.s3.send_buffer_size, Some(1));
721            assert_eq!(config.storage.object_store.s3.recv_buffer_size, Some(1));
722            assert_eq!(config.storage.object_store.s3.nodelay, Some(false));
723            assert!(
724                config
725                    .storage
726                    .object_store
727                    .s3
728                    .developer
729                    .retry_unknown_service_error
730            );
731            assert_eq!(
732                config
733                    .storage
734                    .object_store
735                    .s3
736                    .developer
737                    .retryable_service_error_codes,
738                vec!["dummy".to_owned()]
739            );
740        }
741
742        // Define configs with the new name and make sure it works
743        {
744            let config: RwConfig = toml::from_str(
745                r#"
746            [storage.object_store]
747            set_atomic_write_dir = true
748
749            [storage.object_store.s3]
750            keepalive_ms = 1
751            send_buffer_size = 1
752            recv_buffer_size = 1
753            nodelay = false
754
755            [storage.object_store.s3.developer]
756            retry_unknown_service_error = true
757            retryable_service_error_codes = ['dummy']
758
759
760            "#,
761            )
762            .unwrap();
763
764            assert!(config.storage.object_store.set_atomic_write_dir);
765            assert_eq!(config.storage.object_store.s3.keepalive_ms, Some(1));
766            assert_eq!(config.storage.object_store.s3.send_buffer_size, Some(1));
767            assert_eq!(config.storage.object_store.s3.recv_buffer_size, Some(1));
768            assert_eq!(config.storage.object_store.s3.nodelay, Some(false));
769            assert!(
770                config
771                    .storage
772                    .object_store
773                    .s3
774                    .developer
775                    .retry_unknown_service_error
776            );
777            assert_eq!(
778                config
779                    .storage
780                    .object_store
781                    .s3
782                    .developer
783                    .retryable_service_error_codes,
784                vec!["dummy".to_owned()]
785            );
786        }
787    }
788
789    #[test]
790    fn test_file_cache_separated_runtime_config_backward_compatibility() {
791        let config: RwConfig = toml::from_str(
792            r#"
793            [storage.data_file_cache.runtime_config.Separated.read_runtime_options]
794            worker_threads = 2
795            max_blocking_threads = 4
796
797            [storage.data_file_cache.runtime_config.Separated.write_runtime_options]
798            worker_threads = 6
799            max_blocking_threads = 8
800            "#,
801        )
802        .unwrap();
803
804        let storage::FileCacheRuntimeConfig::Separated {
805            read_runtime_options,
806            write_runtime_options,
807        } = config.storage.data_file_cache.runtime_config
808        else {
809            panic!("expected legacy separated file-cache runtime config");
810        };
811        assert_eq!(read_runtime_options.worker_threads, 2);
812        assert_eq!(read_runtime_options.max_blocking_threads, 4);
813        assert_eq!(write_runtime_options.worker_threads, 6);
814        assert_eq!(write_runtime_options.max_blocking_threads, 8);
815    }
816
817    #[test]
818    fn test_meta_configs_backward_compatibility() {
819        // Test periodic_space_reclaim_compaction_interval_sec
820        {
821            let config: RwConfig = toml::from_str(
822                r#"
823            [meta]
824            periodic_split_compact_group_interval_sec = 1
825            table_write_throughput_threshold = 10
826            min_table_split_write_throughput = 5
827            "#,
828            )
829            .unwrap();
830
831            assert_eq!(
832                config
833                    .meta
834                    .periodic_scheduling_compaction_group_split_interval_sec,
835                1
836            );
837            assert_eq!(config.meta.table_high_write_throughput_threshold, 10);
838            assert_eq!(config.meta.table_low_write_throughput_threshold, 5);
839        }
840    }
841
842    #[test]
843    fn test_meta_max_normalize_splits_per_round_must_be_positive() {
844        let config = toml::from_str::<RwConfig>(
845            r#"
846            [meta]
847            max_normalize_splits_per_round = 0
848            "#,
849        )
850        .unwrap_err();
851
852        expect![[r#"
853            TOML parse error at line 3, column 46
854              |
855            3 |             max_normalize_splits_per_round = 0
856              |                                              ^
857            meta.max_normalize_splits_per_round must be greater than 0
858        "#]]
859        .assert_eq(&config.to_string());
860    }
861
862    // Previously, we have prefixes like `stream_` for all configs under `streaming.developer`.
863    // Later we removed the prefixes, but we still want to guarantee the backward compatibility.
864    #[test]
865    fn test_prefix_alias() {
866        let config: RwConfig = toml::from_str(
867            "
868            [streaming.developer]
869            stream_chunk_size = 114514
870
871            [streaming.developer.stream_compute_client_config]
872            connect_timeout_secs = 42
873            pool_setup_concurrency = 10
874            ",
875        )
876        .unwrap();
877
878        assert_eq!(config.streaming.developer.chunk_size, 114514);
879        assert_eq!(
880            config
881                .streaming
882                .developer
883                .compute_client_config
884                .connect_timeout_secs,
885            42
886        );
887        assert_eq!(
888            config
889                .streaming
890                .developer
891                .compute_client_config
892                .pool_setup_concurrency,
893            10
894        );
895    }
896
897    #[test]
898    fn test_prefix_alias_duplicate() {
899        let config = toml::from_str::<RwConfig>(
900            "
901            [streaming.developer]
902            stream_chunk_size = 114514
903            chunk_size = 1919810
904            ",
905        )
906        .unwrap_err();
907
908        expect![[r#"
909            TOML parse error at line 2, column 13
910              |
911            2 |             [streaming.developer]
912              |             ^^^^^^^^^^^^^^^^^^^^^
913            duplicate field `chunk_size`
914        "#]]
915        .assert_eq(&config.to_string());
916
917        let config = toml::from_str::<RwConfig>(
918            "
919            [streaming.developer.stream_compute_client_config]
920            connect_timeout_secs = 5
921
922            [streaming.developer.compute_client_config]
923            connect_timeout_secs = 10
924            ",
925        )
926        .unwrap_err();
927
928        expect![[r#"
929            TOML parse error at line 2, column 24
930              |
931            2 |             [streaming.developer.stream_compute_client_config]
932              |                        ^^^^^^^^^
933            duplicate field `compute_client_config`
934        "#]]
935        .assert_eq(&config.to_string());
936    }
937
938    #[test]
939    fn test_storage_max_prefetch_block_number_must_be_positive() {
940        let config = toml::from_str::<RwConfig>(
941            r#"
942            [storage]
943            max_prefetch_block_number = 0
944            "#,
945        )
946        .unwrap_err();
947
948        expect![[r#"
949            TOML parse error at line 3, column 41
950              |
951            3 |             max_prefetch_block_number = 0
952              |                                         ^
953            storage.max_prefetch_block_number must be greater than 0
954        "#]]
955        .assert_eq(&config.to_string());
956    }
957
958    #[test]
959    fn test_storage_iceberg_compaction_pull_interval_ms_must_be_positive() {
960        let config = toml::from_str::<RwConfig>(
961            r#"
962            [storage]
963            iceberg_compaction_pull_interval_ms = 0
964            "#,
965        )
966        .unwrap_err();
967
968        expect![[r#"
969            TOML parse error at line 3, column 51
970              |
971            3 |             iceberg_compaction_pull_interval_ms = 0
972              |                                                   ^
973            storage.iceberg_compaction_pull_interval_ms must be greater than 0
974        "#]]
975        .assert_eq(&config.to_string());
976    }
977}
risingwave_common/config/mod.rs

risingwave_common/config/
mod.rs
