risingwave_common/system_param/
mod.rs

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

risingwave_common/system_param/
mod.rs
