risingwave_common/system_param/
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 module defines utilities to work with system parameters ([`PbSystemParams`] in
16//! `meta.proto`).
17//!
18//! To add a new system parameter:
19//! - Add a new field to [`PbSystemParams`] in `meta.proto`.
20//! - Add a new entry to `for_all_params` in this file.
21//! - Add a new method to [`reader::SystemParamsReader`].
22
23pub mod adaptive_parallelism_strategy;
24pub mod common;
25pub mod diff;
26pub mod local_manager;
27pub mod reader;
28
29use std::fmt::Debug;
30use std::ops::RangeBounds;
31use std::str::FromStr;
32
33use paste::paste;
34use risingwave_license::{LicenseKey, LicenseKeyRef};
35use risingwave_pb::meta::PbSystemParams;
36
37use self::diff::SystemParamsDiff;
38pub use crate::system_param::adaptive_parallelism_strategy::AdaptiveParallelismStrategy;
39
40pub type SystemParamsError = String;
41
42type Result<T> = core::result::Result<T, SystemParamsError>;
43
44/// The trait for the value type of a system parameter.
45pub trait ParamValue: ToString + FromStr {
46    type Borrowed<'a>;
47}
48
49macro_rules! impl_param_value {
50    ($type:ty) => {
51        impl_param_value!($type => $type);
52    };
53    ($type:ty => $borrowed:ty) => {
54        impl ParamValue for $type {
55            type Borrowed<'a> = $borrowed;
56        }
57    };
58}
59
60impl_param_value!(bool);
61impl_param_value!(u32);
62impl_param_value!(u64);
63impl_param_value!(f64);
64impl_param_value!(String => &'a str);
65impl_param_value!(LicenseKey => LicenseKeyRef<'a>);
66
67/// Define all system parameters here.
68///
69/// To match all these information, write the match arm as follows:
70/// ```text
71/// ($({ $field:ident, $type:ty, $default:expr, $is_mutable:expr, $doc:literal, $($rest:tt)* },)*) => {
72/// ```
73///
74/// Note:
75/// - Having `None` as default value means the parameter must be initialized.
76#[macro_export]
77macro_rules! for_all_params {
78    ($macro:ident) => {
79        $macro! {
80            // name                                     type                            default value                   mut?    doc
81            { barrier_interval_ms,                      u32,                            Some(1000_u32),                 true,   "The interval of periodic barrier.", },
82            { checkpoint_frequency,                     u64,                            Some(1_u64),                    true,   "There will be a checkpoint for every n barriers.", },
83            { sstable_size_mb,                          u32,                            Some(256_u32),                  false,  "Target size of the Sstable.", },
84            { parallel_compact_size_mb,                 u32,                            Some(512_u32),                  false,  "The size of parallel task for one compact/flush job.", },
85            { block_size_kb,                            u32,                            Some(64_u32),                   false,  "Size of each block in bytes in SST.", },
86            { bloom_false_positive,                     f64,                            Some(0.001_f64),                false,  "False positive probability of bloom filter.", },
87            { state_store,                              String,                         None,                           false,  "URL for the state store", },
88            { data_directory,                           String,                         None,                           false,  "Remote directory for storing data and metadata objects.", },
89            { backup_storage_url,                       String,                         None,                           true,   "Remote storage url for storing snapshots.", },
90            { backup_storage_directory,                 String,                         None,                           true,   "Remote directory for storing snapshots.", },
91            { max_concurrent_creating_streaming_jobs,   u32,                            Some(1_u32),                    true,   "Max number of concurrent creating streaming jobs.", },
92            { pause_on_next_bootstrap,                  bool,                           Some(false),                    true,   "Whether to pause all data sources on next bootstrap.", },
93            { enable_tracing,                           bool,                           Some(false),                    true,   "Whether to enable distributed tracing.", },
94            { use_new_object_prefix_strategy,           bool,                           None,                           false,  "Whether to split object prefix.", },
95            { license_key,                              risingwave_license::LicenseKey, Some(Default::default()),       true,   "The license key to activate enterprise features.", },
96            { time_travel_retention_ms,                 u64,                            Some(600000_u64),               true,   "The data retention period for time travel.", },
97            { adaptive_parallelism_strategy,            risingwave_common::system_param::AdaptiveParallelismStrategy,   Some(Default::default()),       true,   "The strategy for Adaptive Parallelism.", },
98            { per_database_isolation,                   bool,                           Some(true),                     true,   "Whether per database isolation is enabled", },
99            { enforce_secret,                  bool,                           Some(false),                    true,   "Whether to enforce secret on cloud.", },
100        }
101    };
102}
103
104// Warn user if barrier_interval_ms is set above 5mins.
105pub const NOTICE_BARRIER_INTERVAL_MS: u32 = 300000;
106// Warn user if checkpoint_frequency is set above 60.
107pub const NOTICE_CHECKPOINT_FREQUENCY: u64 = 60;
108
109/// Convert field name to string.
110#[macro_export]
111macro_rules! key_of {
112    ($field:ident) => {
113        stringify!($field)
114    };
115}
116
117/// Define key constants for fields in `PbSystemParams` for use of other modules.
118macro_rules! def_key {
119    ($({ $field:ident, $($rest:tt)* },)*) => {
120        paste! {
121            $(
122                pub const [<$field:upper _KEY>]: &str = key_of!($field);
123            )*
124        }
125    };
126}
127
128for_all_params!(def_key);
129
130/// Define default value functions returning `Option`.
131macro_rules! def_default_opt {
132    ($({ $field:ident, $type:ty, $default: expr, $($rest:tt)* },)*) => {
133        $(
134            paste::paste!(
135                pub fn [<$field _opt>]() -> Option<$type> {
136                    $default
137                }
138            );
139        )*
140    };
141}
142
143/// Define default value functions for those with `Some` default values.
144macro_rules! def_default {
145    ($({ $field:ident, $type:ty, $default: expr, $($rest:tt)* },)*) => {
146        $(
147            def_default!(@ $field, $type, $default);
148        )*
149    };
150    (@ $field:ident, $type:ty, None) => {};
151    (@ $field:ident, $type:ty, $default: expr) => {
152        pub fn $field() -> $type {
153            $default.unwrap()
154        }
155        paste::paste!(
156            pub static [<$field:upper>]: LazyLock<$type> = LazyLock::new($field);
157        );
158    };
159}
160
161/// Default values for all parameters.
162pub mod default {
163    use std::sync::LazyLock;
164
165    for_all_params!(def_default_opt);
166    for_all_params!(def_default);
167}
168
169macro_rules! impl_check_missing_fields {
170    ($({ $field:ident, $($rest:tt)* },)*) => {
171        /// Check if any undeprecated fields are missing.
172        pub fn check_missing_params(params: &PbSystemParams) -> Result<()> {
173            $(
174                if params.$field.is_none() {
175                    return Err(format!("missing system param {:?}", key_of!($field)));
176                }
177            )*
178            Ok(())
179        }
180    };
181}
182
183/// Derive serialization to kv pairs.
184macro_rules! impl_system_params_to_kv {
185    ($({ $field:ident, $($rest:tt)* },)*) => {
186        /// The returned map only contains undeprecated fields.
187        /// Return error if there are missing fields.
188        #[allow(clippy::vec_init_then_push)]
189        pub fn system_params_to_kv(params: &PbSystemParams) -> Result<Vec<(String, String)>> {
190            check_missing_params(params)?;
191            let mut ret = Vec::new();
192            $(ret.push((
193                key_of!($field).to_owned(),
194                params.$field.as_ref().unwrap().to_string(),
195            ));)*
196            Ok(ret)
197        }
198    };
199}
200
201macro_rules! impl_derive_missing_fields {
202    ($({ $field:ident, $($rest:tt)* },)*) => {
203        pub fn derive_missing_fields(params: &mut PbSystemParams) {
204            $(
205                if params.$field.is_none() && let Some(v) = OverrideFromParams::$field(params) {
206                    params.$field = Some(v.into());
207                }
208            )*
209        }
210    };
211}
212
213/// Derive deserialization from kv pairs.
214macro_rules! impl_system_params_from_kv {
215    ($({ $field:ident, $($rest:tt)* },)*) => {
216        /// Try to deserialize deprecated fields as well.
217        /// Return error if there are unrecognized fields.
218        pub fn system_params_from_kv<K, V>(mut kvs: Vec<(K, V)>) -> Result<PbSystemParams>
219        where
220            K: AsRef<[u8]> + Debug,
221            V: AsRef<[u8]> + Debug,
222        {
223            let mut ret = PbSystemParams::default();
224            kvs.retain(|(k,v)| {
225                let k = std::str::from_utf8(k.as_ref()).unwrap();
226                let v = std::str::from_utf8(v.as_ref()).unwrap();
227                match k {
228                    $(
229                        key_of!($field) => {
230                            ret.$field = Some(v.parse().unwrap());
231                            false
232                        }
233                    )*
234                    _ => {
235                        true
236                    }
237                }
238            });
239            derive_missing_fields(&mut ret);
240            if !kvs.is_empty() {
241                let unrecognized_params = kvs.into_iter().map(|(k, v)| {
242                    (
243                        std::str::from_utf8(k.as_ref()).unwrap().to_owned(),
244                        std::str::from_utf8(v.as_ref()).unwrap().to_owned(),
245                    )
246                }).collect::<Vec<_>>();
247                tracing::warn!("unrecognized system params {:?}", unrecognized_params);
248            }
249            Ok(ret)
250        }
251    };
252}
253
254/// Define check rules when a field is changed.
255/// If you want custom rules, please override the default implementation in
256/// `OverrideValidateOnSet` below.
257macro_rules! impl_default_validation {
258    ($({ $field:ident, $type:ty, $default:expr, $is_mutable:expr, $($rest:tt)* },)*) => {
259        #[allow(clippy::ptr_arg)]
260        pub trait Validate {
261            $(
262                /// Default implementation does nothing.
263                /// Specific checks are implemented in `OverrideValidate`.
264                fn $field(_v: &$type) -> Result<()> {
265                    Ok(())
266                }
267            )*
268
269            fn expect_range<T, R>(v: T, range: R) -> Result<()>
270            where
271                T: Debug + PartialOrd,
272                R: RangeBounds<T> + Debug,
273            {
274                if !range.contains::<T>(&v) {
275                    Err(format!("value {:?} out of range, expect {:?}", v, range))
276                } else {
277                    Ok(())
278                }
279            }
280        }
281    }
282}
283
284/// Define rules to derive a parameter from others. This is useful for parameter type change or
285/// semantic change, where a new parameter has to be introduced. When the cluster upgrades to a
286/// newer version, we need to ensure the effect of the new parameter is equal to its older versions.
287/// For example, if you had `interval_sec` and now you want finer granularity, you can introduce a
288/// new param `interval_ms` and try to derive it from `interval_sec` by overriding `FromParams`
289/// trait in `OverrideFromParams`:
290///
291/// ```ignore
292/// impl FromParams for OverrideFromParams {
293///     fn interval_ms(params: &PbSystemParams) -> Option<u64> {
294///         if let Some(sec) = params.interval_sec {
295///             Some(sec * 1000)
296///         } else {
297///             None
298///         }
299///     }
300/// }
301/// ```
302///
303/// Note that newer versions must be prioritized during derivation.
304macro_rules! impl_default_from_other_params {
305    ($({ $field:ident, $type:ty, $($rest:tt)* },)*) => {
306        trait FromParams {
307            $(
308                fn $field(_params: &PbSystemParams) -> Option<$type> {
309                    None
310                }
311            )*
312        }
313    };
314}
315
316macro_rules! impl_set_system_param {
317    ($({ $field:ident, $type:ty, $default:expr, $is_mutable:expr, $($rest:tt)* },)*) => {
318        /// Set a system parameter with the given value or default one.
319        ///
320        /// Returns the new value if changed, or an error if the parameter is unrecognized,
321        /// immutable, or the value is invalid.
322        pub fn set_system_param(
323            params: &mut PbSystemParams,
324            key: &str,
325            value: Option<impl AsRef<str>>,
326        ) -> Result<Option<(String, SystemParamsDiff)>> {
327            use crate::system_param::reader::{SystemParamsReader, SystemParamsRead};
328
329            match key {
330                $(
331                    key_of!($field) => {
332                        if !$is_mutable {
333                            return Err(format!("{:?} is immutable", key_of!($field)));
334                        }
335
336                        let v: $type = if let Some(v) = value {
337                            #[allow(rw::format_error)]
338                            v.as_ref().parse().map_err(|e| format!("cannot parse parameter value: {e}"))?
339                        } else {
340                            $default.ok_or_else(|| format!("{} does not have a default value", key))?
341                        };
342                        OverrideValidate::$field(&v)?;
343
344                        let changed = SystemParamsReader::new(&*params).$field() != v;
345                        if changed {
346                            let diff = SystemParamsDiff {
347                                $field: Some(v.to_owned()),
348                                ..Default::default()
349                            };
350                            params.$field = Some(v.into());                                 // do not use `to_string` to avoid writing redacted values
351                            let new_value = params.$field.as_ref().unwrap().to_string();    // can now use `to_string` on protobuf primitive types
352                            Ok(Some((new_value, diff)))
353                        } else {
354                            Ok(None)
355                        }
356                    },
357                )*
358                _ => {
359                    Err(format!(
360                        "unrecognized system parameter {:?}",
361                        key
362                    ))
363                }
364            }
365        }
366    };
367}
368
369macro_rules! impl_is_mutable {
370    ($({ $field:ident, $type:ty, $default:expr, $is_mutable:expr, $($rest:tt)* },)*) => {
371        pub fn is_mutable(field: &str) -> Result<bool> {
372            match field {
373                $(
374                    key_of!($field) => Ok($is_mutable),
375                )*
376                _ => Err(format!("{:?} is not a system parameter", field))
377            }
378        }
379    }
380}
381
382macro_rules! impl_system_params_for_test {
383    ($({ $field:ident, $type:ty, $default:expr, $($rest:tt)* },)*) => {
384        #[allow(clippy::needless_update)]
385        pub fn system_params_for_test() -> PbSystemParams {
386            let mut ret = PbSystemParams {
387                $(
388                    $field: ($default as Option<$type>).map(Into::into),
389                )*
390                ..Default::default() // `None` for deprecated params
391            };
392            ret.data_directory = Some("hummock_001".to_owned());
393            ret.state_store = Some("hummock+memory-isolated-for-test".to_owned());
394            ret.backup_storage_url = Some("memory-isolated-for-test".into());
395            ret.backup_storage_directory = Some("backup".into());
396            ret.use_new_object_prefix_strategy = Some(false);
397            ret.time_travel_retention_ms = Some(0);
398            ret
399        }
400    };
401}
402
403macro_rules! impl_validate_all_params {
404    ($({ $field:ident, $type:ty, $($rest:tt)* },)*) => {
405        /// Validates all present parameters in a `PbSystemParams`.
406        ///
407        /// This function checks the validity of values against the rules in `OverrideValidate`,
408        /// regardless of whether a parameter is mutable. It is suitable for validating
409        /// initial parameters.
410        #[allow(rw::format_error)]
411        pub fn validate_init_system_params(params: &PbSystemParams) -> Result<()> {
412            $(
413                if let Some(ref v_pb) = params.$field {
414                    // 1. Convert the protobuf value (`v_pb`) to a string. `v_pb` could be &u32, &String, etc.
415                    //    `to_string()` works for all of them.
416                    // 2. Parse the string into the target logical type (`$type`), e.g., `LicenseKey`.
417                    //    This relies on the `FromStr` bound on `ParamValue`.
418                    let logical_v: $type = v_pb.to_string().parse()
419                        .map_err(|e| format!("cannot parse value for parameter '{}': {}", key_of!($field), e))?;
420                    // 3. Pass a reference to the correctly-typed logical value to the validator.
421                    OverrideValidate::$field(&logical_v)
422                        .map_err(|e| format!("invalid value for parameter '{}': {}", key_of!($field), e))?;
423                }
424            )*
425            Ok(())
426        }
427    };
428}
429
430for_all_params!(impl_system_params_from_kv);
431for_all_params!(impl_is_mutable);
432for_all_params!(impl_derive_missing_fields);
433for_all_params!(impl_check_missing_fields);
434for_all_params!(impl_system_params_to_kv);
435for_all_params!(impl_set_system_param);
436for_all_params!(impl_default_validation);
437for_all_params!(impl_validate_all_params);
438for_all_params!(impl_system_params_for_test);
439
440pub struct OverrideValidate;
441impl Validate for OverrideValidate {
442    fn barrier_interval_ms(v: &u32) -> Result<()> {
443        Self::expect_range(*v, 50..)
444    }
445
446    fn checkpoint_frequency(v: &u64) -> Result<()> {
447        Self::expect_range(*v, 1..)
448    }
449
450    fn backup_storage_directory(v: &String) -> Result<()> {
451        if v.trim().is_empty() {
452            return Err("backup_storage_directory cannot be empty".into());
453        }
454        Ok(())
455    }
456
457    fn backup_storage_url(v: &String) -> Result<()> {
458        if v.trim().is_empty() {
459            return Err("backup_storage_url cannot be empty".into());
460        }
461        Ok(())
462    }
463
464    fn time_travel_retention_ms(v: &u64) -> Result<()> {
465        // This is intended to guarantee that non-time-travel batch query can still function even compute node's recent versions doesn't include the desired version.
466        let min_retention_ms = 600_000;
467        // 0 is used to disable time travel.
468        if *v != 0 && *v < min_retention_ms {
469            return Err(format!(
470                "time_travel_retention_ms cannot be less than {min_retention_ms}"
471            ));
472        }
473        Ok(())
474    }
475}
476
477for_all_params!(impl_default_from_other_params);
478
479struct OverrideFromParams;
480impl FromParams for OverrideFromParams {}
481
482#[cfg(test)]
483mod tests {
484    use super::*;
485
486    #[test]
487    fn test_to_from_kv() {
488        // Include all fields (deprecated also).
489        let kvs = vec![
490            (BARRIER_INTERVAL_MS_KEY, "1"),
491            (CHECKPOINT_FREQUENCY_KEY, "1"),
492            (SSTABLE_SIZE_MB_KEY, "1"),
493            (PARALLEL_COMPACT_SIZE_MB_KEY, "2"),
494            (BLOCK_SIZE_KB_KEY, "1"),
495            (BLOOM_FALSE_POSITIVE_KEY, "1"),
496            (STATE_STORE_KEY, "a"),
497            (DATA_DIRECTORY_KEY, "a"),
498            (BACKUP_STORAGE_URL_KEY, "a"),
499            (BACKUP_STORAGE_DIRECTORY_KEY, "a"),
500            (MAX_CONCURRENT_CREATING_STREAMING_JOBS_KEY, "1"),
501            (PAUSE_ON_NEXT_BOOTSTRAP_KEY, "false"),
502            (ENABLE_TRACING_KEY, "true"),
503            (USE_NEW_OBJECT_PREFIX_STRATEGY_KEY, "false"),
504            (LICENSE_KEY_KEY, "foo"),
505            (TIME_TRAVEL_RETENTION_MS_KEY, "0"),
506            (ADAPTIVE_PARALLELISM_STRATEGY_KEY, "Auto"),
507            (PER_DATABASE_ISOLATION_KEY, "true"),
508            (ENFORCE_SECRET_KEY, "false"),
509            ("a_deprecated_param", "foo"),
510        ];
511
512        // To kv - missing field.
513        let p = PbSystemParams::default();
514        assert!(system_params_to_kv(&p).is_err());
515
516        // From kv - unrecognized field should be ignored
517        assert!(system_params_from_kv(vec![("?", "?")]).is_ok());
518
519        // Deser & ser.
520        let p = system_params_from_kv(kvs).unwrap();
521        assert_eq!(
522            p,
523            system_params_from_kv(system_params_to_kv(&p).unwrap()).unwrap()
524        );
525    }
526
527    #[test]
528    fn test_set() {
529        let mut p = system_params_for_test();
530        // Unrecognized param.
531        assert!(set_system_param(&mut p, "?", Some("?".to_owned())).is_err());
532        // Value out of range.
533        assert!(set_system_param(&mut p, CHECKPOINT_FREQUENCY_KEY, Some("-1".to_owned())).is_err());
534        // Set immutable.
535        assert!(set_system_param(&mut p, STATE_STORE_KEY, Some("?".to_owned())).is_err());
536        // Parse error.
537        assert!(set_system_param(&mut p, CHECKPOINT_FREQUENCY_KEY, Some("?".to_owned())).is_err());
538        // Normal set.
539        assert!(set_system_param(&mut p, CHECKPOINT_FREQUENCY_KEY, Some("500".to_owned())).is_ok());
540        assert_eq!(p.checkpoint_frequency, Some(500));
541    }
542
543    #[test]
544    fn test_init() {
545        let mut p = system_params_for_test();
546        // Validate all params.
547        assert!(validate_init_system_params(&p).is_ok());
548        p.barrier_interval_ms = Some(10);
549        assert!(validate_init_system_params(&p).is_err());
550        p.barrier_interval_ms = Some(1000);
551        assert!(validate_init_system_params(&p).is_ok());
552    }
553
554    // Test that we always redact the value of the license key when displaying it, but when it comes to
555    // persistency, we still write and get the real value.
556    #[test]
557    fn test_redacted_type() {
558        let mut p = system_params_for_test();
559
560        let new_license_key_value = "new_license_key_value";
561        assert_ne!(p.license_key(), new_license_key_value);
562
563        let (new_string_value, diff) =
564            set_system_param(&mut p, LICENSE_KEY_KEY, Some(new_license_key_value))
565                .expect("should succeed")
566                .expect("should changed");
567
568        // New string value should be the same as what we set.
569        // This should not be redacted.
570        assert_eq!(new_string_value, new_license_key_value);
571
572        let new_value = diff.license_key.unwrap();
573        // `to_string` repr will be redacted.
574        assert_eq!(new_value.to_string(), "<redacted>");
575        // while `Into<String>` still shows the real value.
576        assert_eq!(String::from(new_value.as_ref()), new_license_key_value);
577    }
578}
risingwave_common/system_param/mod.rs

risingwave_common/system_param/
mod.rs
