risingwave_common/config/
mod.rs1pub 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
70pub const MAX_CONNECTION_WINDOW_SIZE: u32 = (1 << 31) - 1;
73pub const STREAM_WINDOW_SIZE: u32 = 32 * 1024 * 1024; #[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#[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 #[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 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 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 }
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 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 }
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 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 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 materialize_force_overwrite_on_no_check() -> bool {
449 false
450 }
451
452 pub fn refresh_scheduler_interval_sec() -> u64 {
453 60
454 }
455
456 pub fn sync_log_store_pause_duration_ms() -> usize {
457 64
458 }
459
460 pub fn sync_log_store_buffer_size() -> usize {
461 2048
462 }
463
464 pub fn disable_sync_log_store_dispatcher() -> bool {
465 false
466 }
467
468 pub fn table_change_log_insert_batch_size() -> u64 {
469 1000
470 }
471
472 pub fn table_change_log_delete_batch_size() -> u64 {
473 1000
474 }
475
476 pub fn enable_state_table_vnode_stats_pruning() -> bool {
477 false
478 }
479
480 pub fn cache_refill_policy() -> CacheRefillPolicy {
481 CacheRefillPolicy::Enabled
482 }
483
484 pub fn enable_vnode_key_stats_for_materialize() -> bool {
485 false
486 }
487
488 pub fn max_concurrent_kv_log_store_historical_read() -> usize {
489 0
490 }
491 }
492}
493
494pub const MAX_META_CACHE_SHARD_BITS: usize = 4;
495pub const MIN_BUFFER_SIZE_PER_SHARD: usize = 256;
496pub const MAX_BLOCK_CACHE_SHARD_BITS: usize = 6; #[cfg(test)]
499pub mod tests {
500 use expect_test::expect;
501 use risingwave_license::LicenseKey;
502
503 use super::*;
504
505 fn default_config_for_docs() -> RwConfig {
506 let mut config = RwConfig::default();
507 config.system.license_key = Some(LicenseKey::empty());
509 config.frontend.unsafe_enable_local_fs_connector = false;
511 config
512 }
513
514 #[test]
518 fn test_example_up_to_date() {
519 const HEADER: &str = "# This file is generated by ./risedev generate-example-config
520# Check detailed comments in src/common/src/config.rs";
521
522 let actual = expect_test::expect_file!["../../../config/example.toml"];
523 let default = toml::to_string(&default_config_for_docs()).expect("failed to serialize");
524
525 let expected = format!("{HEADER}\n\n{default}");
526 actual.assert_eq(&expected);
527
528 let expected = rw_config_to_markdown();
529 let actual = expect_test::expect_file!["../../../config/docs.md"];
530 actual.assert_eq(&expected);
531 }
532
533 #[test]
534 fn test_session_init_entries_distinguishes_omitted_from_default() {
535 let config: RwConfig = toml::from_str(
536 r#"
537 [session_init]
538 streaming_parallelism = "bounded(8)"
539 streaming_parallelism_for_table = "default"
540 "#,
541 )
542 .unwrap();
543
544 assert_eq!(
546 config.session_init.streaming_parallelism.as_deref(),
547 Some("bounded(8)")
548 );
549 assert_eq!(
550 config
551 .session_init
552 .streaming_parallelism_for_table
553 .as_deref(),
554 Some("default")
555 );
556 assert_eq!(config.session_init.streaming_parallelism_for_sink, None);
557
558 assert_eq!(
560 config.session_init.entries(),
561 vec![
562 ("streaming_parallelism", "bounded(8)"),
563 ("streaming_parallelism_for_table", "default"),
564 ]
565 );
566 }
567
568 #[test]
569 fn test_session_init_rejects_unrecognized_key() {
570 let err = toml::from_str::<RwConfig>(
571 r#"
572 [session_init]
573 streaming_parallelism = "bounded(8)"
574 not_a_real_param = "oops"
575 "#,
576 )
577 .unwrap_err();
578
579 assert!(err.to_string().contains("unknown field `not_a_real_param`"));
580 }
581
582 #[derive(Debug)]
583 struct ConfigItemDoc {
584 desc: String,
585 default: String,
586 }
587
588 fn rw_config_to_markdown() -> String {
589 let mut config_rustdocs = BTreeMap::<String, Vec<(String, String)>>::new();
590 RwConfig::config_docs("".to_owned(), &mut config_rustdocs);
591
592 let mut configs: BTreeMap<String, BTreeMap<String, ConfigItemDoc>> = config_rustdocs
594 .into_iter()
595 .map(|(k, v)| {
596 let docs: BTreeMap<String, ConfigItemDoc> = v
597 .into_iter()
598 .map(|(name, desc)| {
599 (
600 name,
601 ConfigItemDoc {
602 desc,
603 default: "".to_owned(), },
605 )
606 })
607 .collect();
608 (k, docs)
609 })
610 .collect();
611
612 let toml_doc: BTreeMap<String, toml::Value> =
613 toml::from_str(&toml::to_string(&default_config_for_docs()).unwrap()).unwrap();
614 toml_doc.into_iter().for_each(|(name, value)| {
615 set_default_values("".to_owned(), name, value, &mut configs);
616 });
617
618 let mut markdown = "# RisingWave System Configurations\n\n".to_owned()
619 + "This page is automatically generated by `./risedev generate-example-config`\n";
620 for (section, configs) in configs {
621 if configs.is_empty() {
622 continue;
623 }
624 markdown.push_str(&format!("\n## {}\n\n", section));
625 markdown.push_str("| Config | Description | Default |\n");
626 markdown.push_str("|--------|-------------|---------|\n");
627 for (config, doc) in configs {
628 markdown.push_str(&format!(
629 "| {} | {} | {} |\n",
630 config, doc.desc, doc.default
631 ));
632 }
633 }
634 markdown
635 }
636
637 fn set_default_values(
638 section: String,
639 name: String,
640 value: toml::Value,
641 configs: &mut BTreeMap<String, BTreeMap<String, ConfigItemDoc>>,
642 ) {
643 if let toml::Value::Table(table) = value {
645 let section_configs: BTreeMap<String, toml::Value> = table.into_iter().collect();
646 let sub_section = if section.is_empty() {
647 name
648 } else {
649 format!("{}.{}", section, name)
650 };
651 section_configs
652 .into_iter()
653 .for_each(|(k, v)| set_default_values(sub_section.clone(), k, v, configs))
654 } else if let Some(t) = configs.get_mut(§ion)
655 && let Some(item_doc) = t.get_mut(&name)
656 {
657 item_doc.default = format!("{}", value);
658 }
659 }
660
661 #[test]
662 fn test_object_store_configs_backward_compatibility() {
663 {
665 let config: RwConfig = toml::from_str(
666 r#"
667 [storage.object_store]
668 object_store_set_atomic_write_dir = true
669
670 [storage.object_store.s3]
671 object_store_keepalive_ms = 1
672 object_store_send_buffer_size = 1
673 object_store_recv_buffer_size = 1
674 object_store_nodelay = false
675
676 [storage.object_store.s3.developer]
677 object_store_retry_unknown_service_error = true
678 object_store_retryable_service_error_codes = ['dummy']
679
680
681 "#,
682 )
683 .unwrap();
684
685 assert!(config.storage.object_store.set_atomic_write_dir);
686 assert_eq!(config.storage.object_store.s3.keepalive_ms, Some(1));
687 assert_eq!(config.storage.object_store.s3.send_buffer_size, Some(1));
688 assert_eq!(config.storage.object_store.s3.recv_buffer_size, Some(1));
689 assert_eq!(config.storage.object_store.s3.nodelay, Some(false));
690 assert!(
691 config
692 .storage
693 .object_store
694 .s3
695 .developer
696 .retry_unknown_service_error
697 );
698 assert_eq!(
699 config
700 .storage
701 .object_store
702 .s3
703 .developer
704 .retryable_service_error_codes,
705 vec!["dummy".to_owned()]
706 );
707 }
708
709 {
711 let config: RwConfig = toml::from_str(
712 r#"
713 [storage.object_store]
714 set_atomic_write_dir = true
715
716 [storage.object_store.s3]
717 keepalive_ms = 1
718 send_buffer_size = 1
719 recv_buffer_size = 1
720 nodelay = false
721
722 [storage.object_store.s3.developer]
723 retry_unknown_service_error = true
724 retryable_service_error_codes = ['dummy']
725
726
727 "#,
728 )
729 .unwrap();
730
731 assert!(config.storage.object_store.set_atomic_write_dir);
732 assert_eq!(config.storage.object_store.s3.keepalive_ms, Some(1));
733 assert_eq!(config.storage.object_store.s3.send_buffer_size, Some(1));
734 assert_eq!(config.storage.object_store.s3.recv_buffer_size, Some(1));
735 assert_eq!(config.storage.object_store.s3.nodelay, Some(false));
736 assert!(
737 config
738 .storage
739 .object_store
740 .s3
741 .developer
742 .retry_unknown_service_error
743 );
744 assert_eq!(
745 config
746 .storage
747 .object_store
748 .s3
749 .developer
750 .retryable_service_error_codes,
751 vec!["dummy".to_owned()]
752 );
753 }
754 }
755
756 #[test]
757 fn test_meta_configs_backward_compatibility() {
758 {
760 let config: RwConfig = toml::from_str(
761 r#"
762 [meta]
763 periodic_split_compact_group_interval_sec = 1
764 table_write_throughput_threshold = 10
765 min_table_split_write_throughput = 5
766 "#,
767 )
768 .unwrap();
769
770 assert_eq!(
771 config
772 .meta
773 .periodic_scheduling_compaction_group_split_interval_sec,
774 1
775 );
776 assert_eq!(config.meta.table_high_write_throughput_threshold, 10);
777 assert_eq!(config.meta.table_low_write_throughput_threshold, 5);
778 }
779 }
780
781 #[test]
782 fn test_meta_max_normalize_splits_per_round_must_be_positive() {
783 let config = toml::from_str::<RwConfig>(
784 r#"
785 [meta]
786 max_normalize_splits_per_round = 0
787 "#,
788 )
789 .unwrap_err();
790
791 expect![[r#"
792 TOML parse error at line 3, column 46
793 |
794 3 | max_normalize_splits_per_round = 0
795 | ^
796 meta.max_normalize_splits_per_round must be greater than 0
797 "#]]
798 .assert_eq(&config.to_string());
799 }
800
801 #[test]
804 fn test_prefix_alias() {
805 let config: RwConfig = toml::from_str(
806 "
807 [streaming.developer]
808 stream_chunk_size = 114514
809
810 [streaming.developer.stream_compute_client_config]
811 connect_timeout_secs = 42
812 pool_setup_concurrency = 10
813 ",
814 )
815 .unwrap();
816
817 assert_eq!(config.streaming.developer.chunk_size, 114514);
818 assert_eq!(
819 config
820 .streaming
821 .developer
822 .compute_client_config
823 .connect_timeout_secs,
824 42
825 );
826 assert_eq!(
827 config
828 .streaming
829 .developer
830 .compute_client_config
831 .pool_setup_concurrency,
832 10
833 );
834 }
835
836 #[test]
837 fn test_prefix_alias_duplicate() {
838 let config = toml::from_str::<RwConfig>(
839 "
840 [streaming.developer]
841 stream_chunk_size = 114514
842 chunk_size = 1919810
843 ",
844 )
845 .unwrap_err();
846
847 expect![[r#"
848 TOML parse error at line 2, column 13
849 |
850 2 | [streaming.developer]
851 | ^^^^^^^^^^^^^^^^^^^^^
852 duplicate field `chunk_size`
853 "#]]
854 .assert_eq(&config.to_string());
855
856 let config = toml::from_str::<RwConfig>(
857 "
858 [streaming.developer.stream_compute_client_config]
859 connect_timeout_secs = 5
860
861 [streaming.developer.compute_client_config]
862 connect_timeout_secs = 10
863 ",
864 )
865 .unwrap_err();
866
867 expect![[r#"
868 TOML parse error at line 2, column 24
869 |
870 2 | [streaming.developer.stream_compute_client_config]
871 | ^^^^^^^^^
872 duplicate field `compute_client_config`
873 "#]]
874 .assert_eq(&config.to_string());
875 }
876
877 #[test]
878 fn test_storage_max_prefetch_block_number_must_be_positive() {
879 let config = toml::from_str::<RwConfig>(
880 r#"
881 [storage]
882 max_prefetch_block_number = 0
883 "#,
884 )
885 .unwrap_err();
886
887 expect![[r#"
888 TOML parse error at line 3, column 41
889 |
890 3 | max_prefetch_block_number = 0
891 | ^
892 storage.max_prefetch_block_number must be greater than 0
893 "#]]
894 .assert_eq(&config.to_string());
895 }
896
897 #[test]
898 fn test_storage_iceberg_compaction_pull_interval_ms_must_be_positive() {
899 let config = toml::from_str::<RwConfig>(
900 r#"
901 [storage]
902 iceberg_compaction_pull_interval_ms = 0
903 "#,
904 )
905 .unwrap_err();
906
907 expect![[r#"
908 TOML parse error at line 3, column 51
909 |
910 3 | iceberg_compaction_pull_interval_ms = 0
911 | ^
912 storage.iceberg_compaction_pull_interval_ms must be greater than 0
913 "#]]
914 .assert_eq(&config.to_string());
915 }
916}