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

risingwave_common/config/
mod.rs
