mz_sql/session/
vars.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Run-time configuration parameters
11//!
12//! ## Overview
13//! Materialize roughly follows the PostgreSQL configuration model, which works
14//! as follows. There is a global set of named configuration parameters, like
15//! `DateStyle` and `client_encoding`. These parameters can be set in several
16//! places: in an on-disk configuration file (in Postgres, named
17//! postgresql.conf), in command line arguments when the server is started, or
18//! at runtime via the `ALTER SYSTEM` or `SET` statements. Parameters that are
19//! set in a session take precedence over database defaults, which in turn take
20//! precedence over command line arguments, which in turn take precedence over
21//! settings in the on-disk configuration. Note that changing the value of
22//! parameters obeys transaction semantics: if a transaction fails to commit,
23//! any parameters that were changed in that transaction (i.e., via `SET`) will
24//! be rolled back to their previous value.
25//!
26//! The Materialize configuration hierarchy at the moment is much simpler.
27//! Global defaults are hardcoded into the binary, and a select few parameters
28//! can be overridden per session. A select few parameters can be overridden on
29//! disk.
30//!
31//! The set of variables that can be overridden per session and the set of
32//! variables that can be overridden on disk are currently disjoint. The
33//! infrastructure has been designed with an eye towards merging these two sets
34//! and supporting additional layers to the hierarchy, however, should the need
35//! arise.
36//!
37//! The configuration parameters that exist are driven by compatibility with
38//! PostgreSQL drivers that expect them, not because they are particularly
39//! important.
40//!
41//! ## Structure
42//! The most meaningful exports from this module are:
43//!
44//! - [`SessionVars`] represent per-session parameters, which each user can
45//!   access independently of one another, and are accessed via `SET`.
46//!
47//!   The fields of [`SessionVars`] are either;
48//!     - `SessionVar`, which is preferable and simply requires full support of
49//!       the `SessionVar` impl for its embedded value type.
50//!     - `ServerVar` for types that do not currently support everything
51//!       required by `SessionVar`, e.g. they are fixed-value parameters.
52//!
53//!   In the fullness of time, all fields in [`SessionVars`] should be
54//!   `SessionVar`.
55//!
56//! - [`SystemVars`] represent system-wide configuration settings and are
57//!   accessed via `ALTER SYSTEM SET`.
58//!
59//!   All elements of [`SystemVars`] are `SystemVar`.
60//!
61//! Some [`VarDefinition`] are also marked as a [`FeatureFlag`]; this is just a
62//! wrapper to make working with a set of [`VarDefinition`] easier, primarily from
63//! within SQL planning, where we might want to check if a feature is enabled
64//! before planning it.
65
66use std::borrow::Cow;
67use std::clone::Clone;
68use std::collections::BTreeMap;
69use std::fmt::Debug;
70use std::net::IpAddr;
71use std::string::ToString;
72use std::sync::{Arc, LazyLock};
73use std::time::Duration;
74
75use chrono::{DateTime, Utc};
76use derivative::Derivative;
77use im::OrdMap;
78use mz_build_info::BuildInfo;
79use mz_dyncfg::{ConfigSet, ConfigType, ConfigUpdates, ConfigVal};
80use mz_persist_client::cfg::{CRDB_CONNECT_TIMEOUT, CRDB_TCP_USER_TIMEOUT};
81use mz_repr::adt::numeric::Numeric;
82use mz_repr::adt::timestamp::CheckedTimestamp;
83use mz_repr::bytes::ByteSize;
84use mz_repr::user::ExternalUserMetadata;
85use mz_tracing::{CloneableEnvFilter, SerializableDirective};
86use serde::Serialize;
87use thiserror::Error;
88use tracing::error;
89use uncased::UncasedStr;
90
91use crate::ast::Ident;
92use crate::session::user::User;
93
94pub(crate) mod constraints;
95pub(crate) mod definitions;
96pub(crate) mod errors;
97pub(crate) mod polyfill;
98pub(crate) mod value;
99
100pub use definitions::*;
101pub use errors::*;
102pub use value::*;
103
104/// The action to take during end_transaction.
105///
106/// This enum lives here because of convenience: it's more of an adapter
107/// concept but [`SessionVars::end_transaction`] takes it.
108#[derive(Debug, Clone, Copy, PartialEq, Eq)]
109pub enum EndTransactionAction {
110    /// Commit the transaction.
111    Commit,
112    /// Rollback the transaction.
113    Rollback,
114}
115
116/// Represents the input to a variable.
117///
118/// Each variable has different rules for how it handles each style of input.
119/// This type allows us to defer interpretation of the input until the
120/// variable-specific interpretation can be applied.
121#[derive(Debug, Clone, Copy)]
122pub enum VarInput<'a> {
123    /// The input has been flattened into a single string.
124    Flat(&'a str),
125    /// The input comes from a SQL `SET` statement and is jumbled across
126    /// multiple components.
127    SqlSet(&'a [String]),
128}
129
130impl<'a> VarInput<'a> {
131    /// Converts the variable input to an owned vector of strings.
132    pub fn to_vec(&self) -> Vec<String> {
133        match self {
134            VarInput::Flat(v) => vec![v.to_string()],
135            VarInput::SqlSet(values) => values.into_iter().map(|v| v.to_string()).collect(),
136        }
137    }
138}
139
140/// An owned version of [`VarInput`].
141#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize)]
142pub enum OwnedVarInput {
143    /// See [`VarInput::Flat`].
144    Flat(String),
145    /// See [`VarInput::SqlSet`].
146    SqlSet(Vec<String>),
147}
148
149impl OwnedVarInput {
150    /// Converts this owned variable input as a [`VarInput`].
151    pub fn borrow(&self) -> VarInput {
152        match self {
153            OwnedVarInput::Flat(v) => VarInput::Flat(v),
154            OwnedVarInput::SqlSet(v) => VarInput::SqlSet(v),
155        }
156    }
157}
158
159/// A `Var` represents a configuration parameter of an arbitrary type.
160pub trait Var: Debug {
161    /// Returns the name of the configuration parameter.
162    fn name(&self) -> &'static str;
163
164    /// Constructs a flattened string representation of the current value of the
165    /// configuration parameter.
166    ///
167    /// The resulting string is guaranteed to be parsable if provided to
168    /// `Value::parse` as a [`VarInput::Flat`].
169    fn value(&self) -> String;
170
171    /// Returns a short sentence describing the purpose of the configuration
172    /// parameter.
173    fn description(&self) -> &'static str;
174
175    /// Returns the name of the type of this variable.
176    fn type_name(&self) -> Cow<'static, str>;
177
178    /// Indicates wither the [`Var`] is visible as a function of the `user` and `system_vars`.
179    /// "Invisible" parameters return `VarErrors`.
180    ///
181    /// Variables marked as `internal` are only visible for the system user.
182    fn visible(&self, user: &User, system_vars: &SystemVars) -> Result<(), VarError>;
183
184    /// Reports whether the variable is only visible in unsafe mode.
185    fn is_unsafe(&self) -> bool {
186        self.name().starts_with("unsafe_")
187    }
188
189    /// Upcast `self` to a `dyn Var`, useful when working with multiple different implementors of
190    /// [`Var`].
191    fn as_var(&self) -> &dyn Var
192    where
193        Self: Sized,
194    {
195        self
196    }
197}
198
199/// A `SessionVar` is the session value for a configuration parameter. If unset,
200/// the server default is used instead.
201///
202/// Note: even though all of the different `*_value` fields are `Box<dyn Value>` they are enforced
203/// to be the same type because we use the `definition`s `parse(...)` method. This is guaranteed to
204/// return the same type as the compiled in default.
205#[derive(Debug)]
206pub struct SessionVar {
207    definition: VarDefinition,
208    /// System or Role default value.
209    default_value: Option<Box<dyn Value>>,
210    /// Value `LOCAL` to a transaction, will be unset at the completion of the transaction.
211    local_value: Option<Box<dyn Value>>,
212    /// Value set during a transaction, will be set if the transaction is committed.
213    staged_value: Option<Box<dyn Value>>,
214    /// Value that overrides the default.
215    session_value: Option<Box<dyn Value>>,
216}
217
218impl Clone for SessionVar {
219    fn clone(&self) -> Self {
220        SessionVar {
221            definition: self.definition.clone(),
222            default_value: self.default_value.as_ref().map(|v| v.box_clone()),
223            local_value: self.local_value.as_ref().map(|v| v.box_clone()),
224            staged_value: self.staged_value.as_ref().map(|v| v.box_clone()),
225            session_value: self.session_value.as_ref().map(|v| v.box_clone()),
226        }
227    }
228}
229
230impl SessionVar {
231    pub const fn new(var: VarDefinition) -> Self {
232        SessionVar {
233            definition: var,
234            default_value: None,
235            local_value: None,
236            staged_value: None,
237            session_value: None,
238        }
239    }
240
241    /// Checks if the provided [`VarInput`] is valid for the current session variable, returning
242    /// the formatted output if it's valid.
243    pub fn check(&self, input: VarInput) -> Result<String, VarError> {
244        let v = self.definition.parse(input)?;
245        self.validate_constraints(v.as_ref())?;
246
247        Ok(v.format())
248    }
249
250    /// Parse the input and update the stored value to match.
251    pub fn set(&mut self, input: VarInput, local: bool) -> Result<(), VarError> {
252        let v = self.definition.parse(input)?;
253
254        // Validate our parsed value.
255        self.validate_constraints(v.as_ref())?;
256
257        if local {
258            self.local_value = Some(v);
259        } else {
260            self.local_value = None;
261            self.staged_value = Some(v);
262        }
263        Ok(())
264    }
265
266    /// Sets the default value for the variable.
267    pub fn set_default(&mut self, input: VarInput) -> Result<(), VarError> {
268        let v = self.definition.parse(input)?;
269        self.validate_constraints(v.as_ref())?;
270        self.default_value = Some(v);
271        Ok(())
272    }
273
274    /// Reset the stored value to the default.
275    pub fn reset(&mut self, local: bool) {
276        let value = self
277            .default_value
278            .as_ref()
279            .map(|v| v.as_ref())
280            .unwrap_or_else(|| self.definition.value.value());
281        if local {
282            self.local_value = Some(value.box_clone());
283        } else {
284            self.local_value = None;
285            self.staged_value = Some(value.box_clone());
286        }
287    }
288
289    /// Returns a possibly new SessionVar if this needs to mutate at transaction end.
290    #[must_use]
291    pub fn end_transaction(&self, action: EndTransactionAction) -> Option<Self> {
292        if !self.is_mutating() {
293            return None;
294        }
295        let mut next: Self = self.clone();
296        next.local_value = None;
297        match action {
298            EndTransactionAction::Commit if next.staged_value.is_some() => {
299                next.session_value = next.staged_value.take()
300            }
301            _ => next.staged_value = None,
302        }
303        Some(next)
304    }
305
306    /// Whether this Var needs to mutate at the end of a transaction.
307    pub fn is_mutating(&self) -> bool {
308        self.local_value.is_some() || self.staged_value.is_some()
309    }
310
311    pub fn value_dyn(&self) -> &dyn Value {
312        self.local_value
313            .as_deref()
314            .or(self.staged_value.as_deref())
315            .or(self.session_value.as_deref())
316            .or(self.default_value.as_deref())
317            .unwrap_or_else(|| self.definition.value.value())
318    }
319
320    /// Returns the [`Value`] that is currently stored as the `session_value`.
321    ///
322    /// Note: This should __only__ be used for inspection, if you want to determine the current
323    /// value of this [`SessionVar`] you should use [`SessionVar::value`].
324    pub fn inspect_session_value(&self) -> Option<&dyn Value> {
325        self.session_value.as_deref()
326    }
327
328    fn validate_constraints(&self, val: &dyn Value) -> Result<(), VarError> {
329        if let Some(constraint) = &self.definition.constraint {
330            constraint.check_constraint(self, self.value_dyn(), val)
331        } else {
332            Ok(())
333        }
334    }
335}
336
337impl Var for SessionVar {
338    fn name(&self) -> &'static str {
339        self.definition.name.as_str()
340    }
341
342    fn value(&self) -> String {
343        self.value_dyn().format()
344    }
345
346    fn description(&self) -> &'static str {
347        self.definition.description
348    }
349
350    fn type_name(&self) -> Cow<'static, str> {
351        self.definition.type_name()
352    }
353
354    fn visible(
355        &self,
356        user: &User,
357        system_vars: &super::vars::SystemVars,
358    ) -> Result<(), super::vars::VarError> {
359        self.definition.visible(user, system_vars)
360    }
361}
362
363#[derive(Debug, Clone, PartialEq, Eq)]
364pub struct MzVersion {
365    /// Inputs to computed variables.
366    build_info: &'static BuildInfo,
367    /// Helm chart version
368    helm_chart_version: Option<String>,
369}
370
371impl MzVersion {
372    pub fn new(build_info: &'static BuildInfo, helm_chart_version: Option<String>) -> Self {
373        MzVersion {
374            build_info,
375            helm_chart_version,
376        }
377    }
378}
379
380/// Session variables.
381///
382/// See the [`crate::session::vars`] module documentation for more details on the
383/// Materialize configuration model.
384#[derive(Debug, Clone)]
385pub struct SessionVars {
386    /// The set of all session variables.
387    vars: OrdMap<&'static UncasedStr, SessionVar>,
388    /// Inputs to computed variables.
389    mz_version: MzVersion,
390    /// Information about the user associated with this Session.
391    user: User,
392}
393
394impl SessionVars {
395    /// Creates a new [`SessionVars`] without considering the System or Role defaults.
396    pub fn new_unchecked(
397        build_info: &'static BuildInfo,
398        user: User,
399        helm_chart_version: Option<String>,
400    ) -> SessionVars {
401        use definitions::*;
402
403        let vars = [
404            &FAILPOINTS,
405            &SERVER_VERSION,
406            &SERVER_VERSION_NUM,
407            &SQL_SAFE_UPDATES,
408            &REAL_TIME_RECENCY,
409            &EMIT_PLAN_INSIGHTS_NOTICE,
410            &EMIT_TIMESTAMP_NOTICE,
411            &EMIT_TRACE_ID_NOTICE,
412            &AUTO_ROUTE_CATALOG_QUERIES,
413            &ENABLE_SESSION_RBAC_CHECKS,
414            &ENABLE_SESSION_CARDINALITY_ESTIMATES,
415            &MAX_IDENTIFIER_LENGTH,
416            &STATEMENT_LOGGING_SAMPLE_RATE,
417            &EMIT_INTROSPECTION_QUERY_NOTICE,
418            &UNSAFE_NEW_TRANSACTION_WALL_TIME,
419            &WELCOME_MESSAGE,
420        ]
421        .into_iter()
422        .chain(SESSION_SYSTEM_VARS.iter().map(|(_name, var)| *var))
423        .map(|var| (var.name, SessionVar::new(var.clone())))
424        .collect();
425
426        SessionVars {
427            vars,
428            mz_version: MzVersion::new(build_info, helm_chart_version),
429            user,
430        }
431    }
432
433    fn expect_value<V: Value>(&self, var: &VarDefinition) -> &V {
434        let var = self
435            .vars
436            .get(var.name)
437            .expect("provided var should be in state");
438        let val = var.value_dyn();
439        val.as_any().downcast_ref::<V>().expect("success")
440    }
441
442    /// Returns an iterator over the configuration parameters and their current
443    /// values for this session.
444    ///
445    /// Note that this function does not check that the access variable should
446    /// be visible because of other settings or users. Before or after accessing
447    /// this method, you should call `Var::visible`.
448    pub fn iter(&self) -> impl Iterator<Item = &dyn Var> {
449        #[allow(clippy::as_conversions)]
450        self.vars
451            .values()
452            .map(|v| v.as_var())
453            .chain([&self.mz_version as &dyn Var, &self.user])
454    }
455
456    /// Returns an iterator over configuration parameters (and their current
457    /// values for this session) that are expected to be sent to the client when
458    /// a new connection is established or when their value changes.
459    pub fn notify_set(&self) -> impl Iterator<Item = &dyn Var> {
460        // WARNING: variables in this set are not checked for visibility, and
461        // are assumed to be visible for all sessions.
462        //
463        // This is fixible with some elbow grease, but at the moment it seems
464        // unlikely that we'll have a variable in the notify set that shouldn't
465        // be visible to all sessions.
466        [
467            &APPLICATION_NAME,
468            &CLIENT_ENCODING,
469            &DATE_STYLE,
470            &INTEGER_DATETIMES,
471            &SERVER_VERSION,
472            &STANDARD_CONFORMING_STRINGS,
473            &TIMEZONE,
474            &INTERVAL_STYLE,
475            // Including `cluster`, `cluster_replica`, `database`, and `search_path` in the notify
476            // set is a Materialize extension. Doing so allows users to more easily identify where
477            // their queries will be executing, which is important to know when you consider the
478            // size of a cluster, what indexes are present, etc.
479            &CLUSTER,
480            &CLUSTER_REPLICA,
481            &DEFAULT_CLUSTER_REPLICATION_FACTOR,
482            &DATABASE,
483            &SEARCH_PATH,
484        ]
485        .into_iter()
486        .map(|v| self.vars[v.name].as_var())
487        // Including `mz_version` in the notify set is a Materialize
488        // extension. Doing so allows applications to detect whether they
489        // are talking to Materialize or PostgreSQL without an additional
490        // network roundtrip. This is known to be safe because CockroachDB
491        // has an analogous extension [0].
492        // [0]: https://github.com/cockroachdb/cockroach/blob/369c4057a/pkg/sql/pgwire/conn.go#L1840
493        .chain(std::iter::once(self.mz_version.as_var()))
494    }
495
496    /// Resets all variables to their default value.
497    pub fn reset_all(&mut self) {
498        let names: Vec<_> = self.vars.keys().copied().collect();
499        for name in names {
500            self.vars[name].reset(false);
501        }
502    }
503
504    /// Returns a [`Var`] representing the configuration parameter with the
505    /// specified name.
506    ///
507    /// Configuration parameters are matched case insensitively. If no such
508    /// configuration parameter exists, `get` returns an error.
509    ///
510    /// Note that if `name` is known at compile time, you should instead use the
511    /// named accessor to access the variable with its true Rust type. For
512    /// example, `self.get("sql_safe_updates").value()` returns the string
513    /// `"true"` or `"false"`, while `self.sql_safe_updates()` returns a bool.
514    pub fn get(&self, system_vars: &SystemVars, name: &str) -> Result<&dyn Var, VarError> {
515        let name = compat_translate_name(name);
516
517        let name = UncasedStr::new(name);
518        if name == MZ_VERSION_NAME {
519            Ok(&self.mz_version)
520        } else if name == IS_SUPERUSER_NAME {
521            Ok(&self.user)
522        } else {
523            self.vars
524                .get(name)
525                .map(|v| {
526                    v.visible(&self.user, system_vars)?;
527                    Ok(v.as_var())
528                })
529                .transpose()?
530                .ok_or_else(|| VarError::UnknownParameter(name.to_string()))
531        }
532    }
533
534    /// Returns a [`SessionVar`] for inspection.
535    ///
536    /// Note: If you're trying to determine the value of the variable with `name` you should
537    /// instead use the named accessor, or [`SessionVars::get`].
538    pub fn inspect(&self, name: &str) -> Result<&SessionVar, VarError> {
539        let name = compat_translate_name(name);
540
541        self.vars
542            .get(UncasedStr::new(name))
543            .ok_or_else(|| VarError::UnknownParameter(name.to_string()))
544    }
545
546    /// Sets the configuration parameter named `name` to the value represented
547    /// by `value`.
548    ///
549    /// The new value may be either committed or rolled back by the next call to
550    /// [`SessionVars::end_transaction`]. If `local` is true, the new value is always
551    /// discarded by the next call to [`SessionVars::end_transaction`], even if the
552    /// transaction is marked to commit.
553    ///
554    /// Like with [`SessionVars::get`], configuration parameters are matched case
555    /// insensitively. If `value` is not valid, as determined by the underlying
556    /// configuration parameter, or if the named configuration parameter does
557    /// not exist, an error is returned.
558    pub fn set(
559        &mut self,
560        system_vars: &SystemVars,
561        name: &str,
562        input: VarInput,
563        local: bool,
564    ) -> Result<(), VarError> {
565        let (name, input) = compat_translate(name, input);
566
567        let name = UncasedStr::new(name);
568        self.check_read_only(name)?;
569
570        self.vars
571            .get_mut(name)
572            .map(|v| {
573                v.visible(&self.user, system_vars)?;
574                v.set(input, local)
575            })
576            .transpose()?
577            .ok_or_else(|| VarError::UnknownParameter(name.to_string()))
578    }
579
580    /// Sets the default value for the parameter named `name` to the value
581    /// represented by `value`.
582    pub fn set_default(&mut self, name: &str, input: VarInput) -> Result<(), VarError> {
583        let (name, input) = compat_translate(name, input);
584
585        let name = UncasedStr::new(name);
586        self.check_read_only(name)?;
587
588        self.vars
589            .get_mut(name)
590            // Note: visibility is checked when persisting a role default.
591            .map(|v| v.set_default(input))
592            .transpose()?
593            .ok_or_else(|| VarError::UnknownParameter(name.to_string()))
594    }
595
596    /// Sets the configuration parameter named `name` to its default value.
597    ///
598    /// The new value may be either committed or rolled back by the next call to
599    /// [`SessionVars::end_transaction`]. If `local` is true, the new value is
600    /// always discarded by the next call to [`SessionVars::end_transaction`],
601    /// even if the transaction is marked to commit.
602    ///
603    /// Like with [`SessionVars::get`], configuration parameters are matched
604    /// case insensitively. If the named configuration parameter does not exist,
605    /// an error is returned.
606    ///
607    /// If the variable does not exist or the user does not have the visibility
608    /// requires, this function returns an error.
609    pub fn reset(
610        &mut self,
611        system_vars: &SystemVars,
612        name: &str,
613        local: bool,
614    ) -> Result<(), VarError> {
615        let name = compat_translate_name(name);
616
617        let name = UncasedStr::new(name);
618        self.check_read_only(name)?;
619
620        self.vars
621            .get_mut(name)
622            .map(|v| {
623                v.visible(&self.user, system_vars)?;
624                v.reset(local);
625                Ok(())
626            })
627            .transpose()?
628            .ok_or_else(|| VarError::UnknownParameter(name.to_string()))
629    }
630
631    /// Returns an error if the variable corresponding to `name` is read only.
632    fn check_read_only(&self, name: &UncasedStr) -> Result<(), VarError> {
633        if name == MZ_VERSION_NAME {
634            Err(VarError::ReadOnlyParameter(MZ_VERSION_NAME.as_str()))
635        } else if name == IS_SUPERUSER_NAME {
636            Err(VarError::ReadOnlyParameter(IS_SUPERUSER_NAME.as_str()))
637        } else if name == MAX_IDENTIFIER_LENGTH.name {
638            Err(VarError::ReadOnlyParameter(
639                MAX_IDENTIFIER_LENGTH.name.as_str(),
640            ))
641        } else {
642            Ok(())
643        }
644    }
645
646    /// Commits or rolls back configuration parameter updates made via
647    /// [`SessionVars::set`] since the last call to `end_transaction`.
648    ///
649    /// Returns any session parameters that changed because the transaction ended.
650    #[mz_ore::instrument(level = "debug")]
651    pub fn end_transaction(
652        &mut self,
653        action: EndTransactionAction,
654    ) -> BTreeMap<&'static str, String> {
655        let mut changed = BTreeMap::new();
656        let mut updates = Vec::new();
657        for (name, var) in self.vars.iter() {
658            if !var.is_mutating() {
659                continue;
660            }
661            let before = var.value();
662            let next = var.end_transaction(action).expect("must mutate");
663            let after = next.value();
664            updates.push((*name, next));
665
666            // Report the new value of the parameter.
667            if before != after {
668                changed.insert(var.name(), after);
669            }
670        }
671        self.vars.extend(updates);
672        changed
673    }
674
675    /// Returns the value of the `application_name` configuration parameter.
676    pub fn application_name(&self) -> &str {
677        self.expect_value::<String>(&APPLICATION_NAME).as_str()
678    }
679
680    /// Returns the build info.
681    pub fn build_info(&self) -> &'static BuildInfo {
682        self.mz_version.build_info
683    }
684
685    /// Returns the value of the `client_encoding` configuration parameter.
686    pub fn client_encoding(&self) -> &ClientEncoding {
687        self.expect_value(&CLIENT_ENCODING)
688    }
689
690    /// Returns the value of the `client_min_messages` configuration parameter.
691    pub fn client_min_messages(&self) -> &ClientSeverity {
692        self.expect_value(&CLIENT_MIN_MESSAGES)
693    }
694
695    /// Returns the value of the `cluster` configuration parameter.
696    pub fn cluster(&self) -> &str {
697        self.expect_value::<String>(&CLUSTER).as_str()
698    }
699
700    /// Returns the value of the `cluster_replica` configuration parameter.
701    pub fn cluster_replica(&self) -> Option<&str> {
702        self.expect_value::<Option<String>>(&CLUSTER_REPLICA)
703            .as_deref()
704    }
705
706    /// Returns the value of the `current_object_missing_warnings` configuration
707    /// parameter.
708    pub fn current_object_missing_warnings(&self) -> bool {
709        *self.expect_value::<bool>(&CURRENT_OBJECT_MISSING_WARNINGS)
710    }
711
712    /// Returns the value of the `DateStyle` configuration parameter.
713    pub fn date_style(&self) -> &[&str] {
714        &self.expect_value::<DateStyle>(&DATE_STYLE).0
715    }
716
717    /// Returns the value of the `database` configuration parameter.
718    pub fn database(&self) -> &str {
719        self.expect_value::<String>(&DATABASE).as_str()
720    }
721
722    /// Returns the value of the `extra_float_digits` configuration parameter.
723    pub fn extra_float_digits(&self) -> i32 {
724        *self.expect_value(&EXTRA_FLOAT_DIGITS)
725    }
726
727    /// Returns the value of the `integer_datetimes` configuration parameter.
728    pub fn integer_datetimes(&self) -> bool {
729        *self.expect_value(&INTEGER_DATETIMES)
730    }
731
732    /// Returns the value of the `intervalstyle` configuration parameter.
733    pub fn intervalstyle(&self) -> &IntervalStyle {
734        self.expect_value(&INTERVAL_STYLE)
735    }
736
737    /// Returns the value of the `mz_version` configuration parameter.
738    pub fn mz_version(&self) -> String {
739        self.mz_version.value()
740    }
741
742    /// Returns the value of the `search_path` configuration parameter.
743    pub fn search_path(&self) -> &[Ident] {
744        self.expect_value::<Vec<Ident>>(&SEARCH_PATH).as_slice()
745    }
746
747    /// Returns the value of the `server_version` configuration parameter.
748    pub fn server_version(&self) -> &str {
749        self.expect_value::<String>(&SERVER_VERSION).as_str()
750    }
751
752    /// Returns the value of the `server_version_num` configuration parameter.
753    pub fn server_version_num(&self) -> i32 {
754        *self.expect_value(&SERVER_VERSION_NUM)
755    }
756
757    /// Returns the value of the `sql_safe_updates` configuration parameter.
758    pub fn sql_safe_updates(&self) -> bool {
759        *self.expect_value(&SQL_SAFE_UPDATES)
760    }
761
762    /// Returns the value of the `standard_conforming_strings` configuration
763    /// parameter.
764    pub fn standard_conforming_strings(&self) -> bool {
765        *self.expect_value(&STANDARD_CONFORMING_STRINGS)
766    }
767
768    /// Returns the value of the `statement_timeout` configuration parameter.
769    pub fn statement_timeout(&self) -> &Duration {
770        self.expect_value(&STATEMENT_TIMEOUT)
771    }
772
773    /// Returns the value of the `idle_in_transaction_session_timeout` configuration parameter.
774    pub fn idle_in_transaction_session_timeout(&self) -> &Duration {
775        self.expect_value(&IDLE_IN_TRANSACTION_SESSION_TIMEOUT)
776    }
777
778    /// Returns the value of the `timezone` configuration parameter.
779    pub fn timezone(&self) -> &TimeZone {
780        self.expect_value(&TIMEZONE)
781    }
782
783    /// Returns the value of the `transaction_isolation` configuration
784    /// parameter.
785    pub fn transaction_isolation(&self) -> &IsolationLevel {
786        self.expect_value(&TRANSACTION_ISOLATION)
787    }
788
789    /// Returns the value of `real_time_recency` configuration parameter.
790    pub fn real_time_recency(&self) -> bool {
791        *self.expect_value(&REAL_TIME_RECENCY)
792    }
793
794    /// Returns the value of the `real_time_recency_timeout` configuration parameter.
795    pub fn real_time_recency_timeout(&self) -> &Duration {
796        self.expect_value(&REAL_TIME_RECENCY_TIMEOUT)
797    }
798
799    /// Returns the value of `emit_plan_insights_notice` configuration parameter.
800    pub fn emit_plan_insights_notice(&self) -> bool {
801        *self.expect_value(&EMIT_PLAN_INSIGHTS_NOTICE)
802    }
803
804    /// Returns the value of `emit_timestamp_notice` configuration parameter.
805    pub fn emit_timestamp_notice(&self) -> bool {
806        *self.expect_value(&EMIT_TIMESTAMP_NOTICE)
807    }
808
809    /// Returns the value of `emit_trace_id_notice` configuration parameter.
810    pub fn emit_trace_id_notice(&self) -> bool {
811        *self.expect_value(&EMIT_TRACE_ID_NOTICE)
812    }
813
814    /// Returns the value of `auto_route_catalog_queries` configuration parameter.
815    pub fn auto_route_catalog_queries(&self) -> bool {
816        *self.expect_value(&AUTO_ROUTE_CATALOG_QUERIES)
817    }
818
819    /// Returns the value of `enable_session_rbac_checks` configuration parameter.
820    pub fn enable_session_rbac_checks(&self) -> bool {
821        *self.expect_value(&ENABLE_SESSION_RBAC_CHECKS)
822    }
823
824    /// Returns the value of `enable_session_cardinality_estimates` configuration parameter.
825    pub fn enable_session_cardinality_estimates(&self) -> bool {
826        *self.expect_value(&ENABLE_SESSION_CARDINALITY_ESTIMATES)
827    }
828
829    /// Returns the value of `is_superuser` configuration parameter.
830    pub fn is_superuser(&self) -> bool {
831        self.user.is_superuser()
832    }
833
834    /// Returns the user associated with this `SessionVars` instance.
835    pub fn user(&self) -> &User {
836        &self.user
837    }
838
839    /// Returns the value of the `max_query_result_size` configuration parameter.
840    pub fn max_query_result_size(&self) -> u64 {
841        self.expect_value::<ByteSize>(&MAX_QUERY_RESULT_SIZE)
842            .as_bytes()
843    }
844
845    /// Sets the external metadata associated with the user.
846    pub fn set_external_user_metadata(&mut self, metadata: ExternalUserMetadata) {
847        self.user.external_metadata = Some(metadata);
848    }
849
850    pub fn set_cluster(&mut self, cluster: String) {
851        let var = self
852            .vars
853            .get_mut(UncasedStr::new(CLUSTER.name()))
854            .expect("cluster variable must exist");
855        var.set(VarInput::Flat(&cluster), false)
856            .expect("setting cluster must succeed");
857    }
858
859    pub fn set_local_transaction_isolation(&mut self, transaction_isolation: IsolationLevel) {
860        let var = self
861            .vars
862            .get_mut(UncasedStr::new(TRANSACTION_ISOLATION.name()))
863            .expect("transaction_isolation variable must exist");
864        var.set(VarInput::Flat(transaction_isolation.as_str()), true)
865            .expect("setting transaction isolation must succeed");
866    }
867
868    pub fn get_statement_logging_sample_rate(&self) -> Numeric {
869        *self.expect_value(&STATEMENT_LOGGING_SAMPLE_RATE)
870    }
871
872    /// Returns the value of the `emit_introspection_query_notice` configuration parameter.
873    pub fn emit_introspection_query_notice(&self) -> bool {
874        *self.expect_value(&EMIT_INTROSPECTION_QUERY_NOTICE)
875    }
876
877    pub fn unsafe_new_transaction_wall_time(&self) -> Option<CheckedTimestamp<DateTime<Utc>>> {
878        *self.expect_value(&UNSAFE_NEW_TRANSACTION_WALL_TIME)
879    }
880
881    /// Returns the value of the `welcome_message` configuration parameter.
882    pub fn welcome_message(&self) -> bool {
883        *self.expect_value(&WELCOME_MESSAGE)
884    }
885}
886
887// TODO(database-issues#8069) remove together with `compat_translate`
888pub const OLD_CATALOG_SERVER_CLUSTER: &str = "mz_introspection";
889pub const OLD_AUTO_ROUTE_CATALOG_QUERIES: &str = "auto_route_introspection_queries";
890
891/// If the given variable name and/or input is deprecated, return a corresponding updated value,
892/// otherwise return the original.
893///
894/// This method was introduced to gracefully handle the rename of the `mz_introspection` cluster to
895/// `mz_cluster_server`. The plan is to remove it once all users have migrated to the new name. The
896/// debug logs will be helpful for checking this in production.
897// TODO(database-issues#8069) remove this after sufficient time has passed
898fn compat_translate<'a, 'b>(name: &'a str, input: VarInput<'b>) -> (&'a str, VarInput<'b>) {
899    if name == CLUSTER.name() {
900        if let Ok(value) = CLUSTER.parse(input) {
901            if value.format() == OLD_CATALOG_SERVER_CLUSTER {
902                tracing::debug!(
903                    github_27285 = true,
904                    "encountered deprecated `cluster` variable value: {}",
905                    OLD_CATALOG_SERVER_CLUSTER,
906                );
907                return (name, VarInput::Flat("mz_catalog_server"));
908            }
909        }
910    }
911
912    if name == OLD_AUTO_ROUTE_CATALOG_QUERIES {
913        tracing::debug!(
914            github_27285 = true,
915            "encountered deprecated `{}` variable name",
916            OLD_AUTO_ROUTE_CATALOG_QUERIES,
917        );
918        return (AUTO_ROUTE_CATALOG_QUERIES.name(), input);
919    }
920
921    (name, input)
922}
923
924fn compat_translate_name(name: &str) -> &str {
925    let (name, _) = compat_translate(name, VarInput::Flat(""));
926    name
927}
928
929/// A `SystemVar` is persisted on disk value for a configuration parameter. If unset,
930/// the server default is used instead.
931#[derive(Debug)]
932pub struct SystemVar {
933    definition: VarDefinition,
934    /// Value currently persisted to disk.
935    persisted_value: Option<Box<dyn Value>>,
936    /// Current default, not persisted to disk.
937    dynamic_default: Option<Box<dyn Value>>,
938}
939
940impl Clone for SystemVar {
941    fn clone(&self) -> Self {
942        SystemVar {
943            definition: self.definition.clone(),
944            persisted_value: self.persisted_value.as_ref().map(|v| v.box_clone()),
945            dynamic_default: self.dynamic_default.as_ref().map(|v| v.box_clone()),
946        }
947    }
948}
949
950impl SystemVar {
951    pub fn new(definition: VarDefinition) -> Self {
952        SystemVar {
953            definition,
954            persisted_value: None,
955            dynamic_default: None,
956        }
957    }
958
959    fn is_default(&self, input: VarInput) -> Result<bool, VarError> {
960        let v = self.definition.parse(input)?;
961        Ok(self.definition.default_value() == v.as_ref())
962    }
963
964    pub fn value_dyn(&self) -> &dyn Value {
965        self.persisted_value
966            .as_deref()
967            .or(self.dynamic_default.as_deref())
968            .unwrap_or_else(|| self.definition.default_value())
969    }
970
971    pub fn value<V: 'static>(&self) -> &V {
972        let val = self.value_dyn();
973        val.as_any().downcast_ref::<V>().expect("success")
974    }
975
976    fn parse(&self, input: VarInput) -> Result<Box<dyn Value>, VarError> {
977        let v = self.definition.parse(input)?;
978        // Validate our parsed value.
979        self.validate_constraints(v.as_ref())?;
980        Ok(v)
981    }
982
983    fn set(&mut self, input: VarInput) -> Result<bool, VarError> {
984        let v = self.parse(input)?;
985
986        if self.persisted_value.as_ref() != Some(&v) {
987            self.persisted_value = Some(v);
988            Ok(true)
989        } else {
990            Ok(false)
991        }
992    }
993
994    fn reset(&mut self) -> bool {
995        if self.persisted_value.is_some() {
996            self.persisted_value = None;
997            true
998        } else {
999            false
1000        }
1001    }
1002
1003    fn set_default(&mut self, input: VarInput) -> Result<(), VarError> {
1004        let v = self.definition.parse(input)?;
1005        self.dynamic_default = Some(v);
1006        Ok(())
1007    }
1008
1009    fn validate_constraints(&self, val: &dyn Value) -> Result<(), VarError> {
1010        if let Some(constraint) = &self.definition.constraint {
1011            constraint.check_constraint(self, self.value_dyn(), val)
1012        } else {
1013            Ok(())
1014        }
1015    }
1016}
1017
1018impl Var for SystemVar {
1019    fn name(&self) -> &'static str {
1020        self.definition.name.as_str()
1021    }
1022
1023    fn value(&self) -> String {
1024        self.value_dyn().format()
1025    }
1026
1027    fn description(&self) -> &'static str {
1028        self.definition.description
1029    }
1030
1031    fn type_name(&self) -> Cow<'static, str> {
1032        self.definition.type_name()
1033    }
1034
1035    fn visible(&self, user: &User, system_vars: &SystemVars) -> Result<(), VarError> {
1036        self.definition.visible(user, system_vars)
1037    }
1038}
1039
1040#[derive(Debug, Error)]
1041pub enum NetworkPolicyError {
1042    #[error("Access denied for address {0}")]
1043    AddressDenied(IpAddr),
1044}
1045
1046/// On disk variables.
1047///
1048/// See the [`crate::session::vars`] module documentation for more details on the
1049/// Materialize configuration model.
1050#[derive(Derivative, Clone)]
1051#[derivative(Debug)]
1052pub struct SystemVars {
1053    /// Allows "unsafe" parameters to be set.
1054    allow_unsafe: bool,
1055    /// Set of all [`SystemVar`]s.
1056    vars: BTreeMap<&'static UncasedStr, SystemVar>,
1057    /// External components interested in when a [`SystemVar`] gets updated.
1058    #[derivative(Debug = "ignore")]
1059    callbacks: BTreeMap<String, Vec<Arc<dyn Fn(&SystemVars) + Send + Sync>>>,
1060
1061    /// NB: This is intentionally disconnected from the one that is plumbed around to persist and
1062    /// the controllers. This is so we can explicitly control and reason about when changes to config
1063    /// values are propagated to the rest of the system.
1064    dyncfgs: ConfigSet,
1065}
1066
1067impl Default for SystemVars {
1068    fn default() -> Self {
1069        Self::new()
1070    }
1071}
1072
1073impl SystemVars {
1074    pub fn new() -> Self {
1075        let system_vars = vec![
1076            &MAX_KAFKA_CONNECTIONS,
1077            &MAX_POSTGRES_CONNECTIONS,
1078            &MAX_MYSQL_CONNECTIONS,
1079            &MAX_SQL_SERVER_CONNECTIONS,
1080            &MAX_AWS_PRIVATELINK_CONNECTIONS,
1081            &MAX_TABLES,
1082            &MAX_SOURCES,
1083            &MAX_SINKS,
1084            &MAX_MATERIALIZED_VIEWS,
1085            &MAX_CLUSTERS,
1086            &MAX_REPLICAS_PER_CLUSTER,
1087            &MAX_CREDIT_CONSUMPTION_RATE,
1088            &MAX_DATABASES,
1089            &MAX_SCHEMAS_PER_DATABASE,
1090            &MAX_OBJECTS_PER_SCHEMA,
1091            &MAX_SECRETS,
1092            &MAX_ROLES,
1093            &MAX_CONTINUAL_TASKS,
1094            &MAX_NETWORK_POLICIES,
1095            &MAX_RULES_PER_NETWORK_POLICY,
1096            &MAX_RESULT_SIZE,
1097            &MAX_COPY_FROM_SIZE,
1098            &ALLOWED_CLUSTER_REPLICA_SIZES,
1099            &DISK_CLUSTER_REPLICAS_DEFAULT,
1100            &upsert_rocksdb::UPSERT_ROCKSDB_AUTO_SPILL_TO_DISK,
1101            &upsert_rocksdb::UPSERT_ROCKSDB_AUTO_SPILL_THRESHOLD_BYTES,
1102            &upsert_rocksdb::UPSERT_ROCKSDB_COMPACTION_STYLE,
1103            &upsert_rocksdb::UPSERT_ROCKSDB_OPTIMIZE_COMPACTION_MEMTABLE_BUDGET,
1104            &upsert_rocksdb::UPSERT_ROCKSDB_LEVEL_COMPACTION_DYNAMIC_LEVEL_BYTES,
1105            &upsert_rocksdb::UPSERT_ROCKSDB_UNIVERSAL_COMPACTION_RATIO,
1106            &upsert_rocksdb::UPSERT_ROCKSDB_PARALLELISM,
1107            &upsert_rocksdb::UPSERT_ROCKSDB_COMPRESSION_TYPE,
1108            &upsert_rocksdb::UPSERT_ROCKSDB_BOTTOMMOST_COMPRESSION_TYPE,
1109            &upsert_rocksdb::UPSERT_ROCKSDB_BATCH_SIZE,
1110            &upsert_rocksdb::UPSERT_ROCKSDB_RETRY_DURATION,
1111            &upsert_rocksdb::UPSERT_ROCKSDB_STATS_LOG_INTERVAL_SECONDS,
1112            &upsert_rocksdb::UPSERT_ROCKSDB_STATS_PERSIST_INTERVAL_SECONDS,
1113            &upsert_rocksdb::UPSERT_ROCKSDB_POINT_LOOKUP_BLOCK_CACHE_SIZE_MB,
1114            &upsert_rocksdb::UPSERT_ROCKSDB_SHRINK_ALLOCATED_BUFFERS_BY_RATIO,
1115            &upsert_rocksdb::UPSERT_ROCKSDB_WRITE_BUFFER_MANAGER_CLUSTER_MEMORY_FRACTION,
1116            &upsert_rocksdb::UPSERT_ROCKSDB_WRITE_BUFFER_MANAGER_MEMORY_BYTES,
1117            &upsert_rocksdb::UPSERT_ROCKSDB_WRITE_BUFFER_MANAGER_ALLOW_STALL,
1118            &STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES,
1119            &STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES_TO_CLUSTER_SIZE_FRACTION,
1120            &STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES_DISK_ONLY,
1121            &STORAGE_STATISTICS_INTERVAL,
1122            &STORAGE_STATISTICS_COLLECTION_INTERVAL,
1123            &STORAGE_SHRINK_UPSERT_UNUSED_BUFFERS_BY_RATIO,
1124            &STORAGE_RECORD_SOURCE_SINK_NAMESPACED_ERRORS,
1125            &PERSIST_FAST_PATH_LIMIT,
1126            &METRICS_RETENTION,
1127            &UNSAFE_MOCK_AUDIT_EVENT_TIMESTAMP,
1128            &ENABLE_RBAC_CHECKS,
1129            &PG_SOURCE_CONNECT_TIMEOUT,
1130            &PG_SOURCE_TCP_KEEPALIVES_IDLE,
1131            &PG_SOURCE_TCP_KEEPALIVES_INTERVAL,
1132            &PG_SOURCE_TCP_KEEPALIVES_RETRIES,
1133            &PG_SOURCE_TCP_USER_TIMEOUT,
1134            &PG_SOURCE_TCP_CONFIGURE_SERVER,
1135            &PG_SOURCE_SNAPSHOT_STATEMENT_TIMEOUT,
1136            &PG_SOURCE_WAL_SENDER_TIMEOUT,
1137            &PG_SOURCE_SNAPSHOT_COLLECT_STRICT_COUNT,
1138            &MYSQL_SOURCE_TCP_KEEPALIVE,
1139            &MYSQL_SOURCE_SNAPSHOT_MAX_EXECUTION_TIME,
1140            &MYSQL_SOURCE_SNAPSHOT_LOCK_WAIT_TIMEOUT,
1141            &MYSQL_SOURCE_CONNECT_TIMEOUT,
1142            &SSH_CHECK_INTERVAL,
1143            &SSH_CONNECT_TIMEOUT,
1144            &SSH_KEEPALIVES_IDLE,
1145            &KAFKA_SOCKET_KEEPALIVE,
1146            &KAFKA_SOCKET_TIMEOUT,
1147            &KAFKA_TRANSACTION_TIMEOUT,
1148            &KAFKA_SOCKET_CONNECTION_SETUP_TIMEOUT,
1149            &KAFKA_FETCH_METADATA_TIMEOUT,
1150            &KAFKA_PROGRESS_RECORD_FETCH_TIMEOUT,
1151            &ENABLE_LAUNCHDARKLY,
1152            &MAX_CONNECTIONS,
1153            &NETWORK_POLICY,
1154            &SUPERUSER_RESERVED_CONNECTIONS,
1155            &KEEP_N_SOURCE_STATUS_HISTORY_ENTRIES,
1156            &KEEP_N_SINK_STATUS_HISTORY_ENTRIES,
1157            &KEEP_N_PRIVATELINK_STATUS_HISTORY_ENTRIES,
1158            &REPLICA_STATUS_HISTORY_RETENTION_WINDOW,
1159            &ENABLE_STORAGE_SHARD_FINALIZATION,
1160            &ENABLE_CONSOLIDATE_AFTER_UNION_NEGATE,
1161            &ENABLE_DEFAULT_CONNECTION_VALIDATION,
1162            &ENABLE_REDUCE_REDUCTION,
1163            &MIN_TIMESTAMP_INTERVAL,
1164            &MAX_TIMESTAMP_INTERVAL,
1165            &LOGGING_FILTER,
1166            &OPENTELEMETRY_FILTER,
1167            &LOGGING_FILTER_DEFAULTS,
1168            &OPENTELEMETRY_FILTER_DEFAULTS,
1169            &SENTRY_FILTERS,
1170            &WEBHOOKS_SECRETS_CACHING_TTL_SECS,
1171            &COORD_SLOW_MESSAGE_WARN_THRESHOLD,
1172            &grpc_client::CONNECT_TIMEOUT,
1173            &grpc_client::HTTP2_KEEP_ALIVE_INTERVAL,
1174            &grpc_client::HTTP2_KEEP_ALIVE_TIMEOUT,
1175            &cluster_scheduling::CLUSTER_MULTI_PROCESS_REPLICA_AZ_AFFINITY_WEIGHT,
1176            &cluster_scheduling::CLUSTER_SOFTEN_REPLICATION_ANTI_AFFINITY,
1177            &cluster_scheduling::CLUSTER_SOFTEN_REPLICATION_ANTI_AFFINITY_WEIGHT,
1178            &cluster_scheduling::CLUSTER_ENABLE_TOPOLOGY_SPREAD,
1179            &cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_IGNORE_NON_SINGULAR_SCALE,
1180            &cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_MAX_SKEW,
1181            &cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_MIN_DOMAINS,
1182            &cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_SOFT,
1183            &cluster_scheduling::CLUSTER_SOFTEN_AZ_AFFINITY,
1184            &cluster_scheduling::CLUSTER_SOFTEN_AZ_AFFINITY_WEIGHT,
1185            &cluster_scheduling::CLUSTER_ALWAYS_USE_DISK,
1186            &cluster_scheduling::CLUSTER_ALTER_CHECK_READY_INTERVAL,
1187            &cluster_scheduling::CLUSTER_CHECK_SCHEDULING_POLICIES_INTERVAL,
1188            &cluster_scheduling::CLUSTER_SECURITY_CONTEXT_ENABLED,
1189            &cluster_scheduling::CLUSTER_REFRESH_MV_COMPACTION_ESTIMATE,
1190            &grpc_client::HTTP2_KEEP_ALIVE_TIMEOUT,
1191            &STATEMENT_LOGGING_MAX_SAMPLE_RATE,
1192            &STATEMENT_LOGGING_DEFAULT_SAMPLE_RATE,
1193            &STATEMENT_LOGGING_TARGET_DATA_RATE,
1194            &STATEMENT_LOGGING_MAX_DATA_CREDIT,
1195            &ENABLE_INTERNAL_STATEMENT_LOGGING,
1196            &OPTIMIZER_STATS_TIMEOUT,
1197            &OPTIMIZER_ONESHOT_STATS_TIMEOUT,
1198            &PRIVATELINK_STATUS_UPDATE_QUOTA_PER_MINUTE,
1199            &WEBHOOK_CONCURRENT_REQUEST_LIMIT,
1200            &PG_TIMESTAMP_ORACLE_CONNECTION_POOL_MAX_SIZE,
1201            &PG_TIMESTAMP_ORACLE_CONNECTION_POOL_MAX_WAIT,
1202            &PG_TIMESTAMP_ORACLE_CONNECTION_POOL_TTL,
1203            &PG_TIMESTAMP_ORACLE_CONNECTION_POOL_TTL_STAGGER,
1204            &USER_STORAGE_MANAGED_COLLECTIONS_BATCH_DURATION,
1205            &FORCE_SOURCE_TABLE_SYNTAX,
1206            &OPTIMIZER_E2E_LATENCY_WARNING_THRESHOLD,
1207        ];
1208
1209        let dyncfgs = mz_dyncfgs::all_dyncfgs();
1210        let dyncfg_vars: Vec<_> = dyncfgs
1211            .entries()
1212            .map(|cfg| match cfg.default() {
1213                ConfigVal::Bool(default) => {
1214                    VarDefinition::new_runtime(cfg.name(), *default, cfg.desc(), false)
1215                }
1216                ConfigVal::U32(default) => {
1217                    VarDefinition::new_runtime(cfg.name(), *default, cfg.desc(), false)
1218                }
1219                ConfigVal::Usize(default) => {
1220                    VarDefinition::new_runtime(cfg.name(), *default, cfg.desc(), false)
1221                }
1222                ConfigVal::OptUsize(default) => {
1223                    VarDefinition::new_runtime(cfg.name(), *default, cfg.desc(), false)
1224                }
1225                ConfigVal::F64(default) => {
1226                    VarDefinition::new_runtime(cfg.name(), *default, cfg.desc(), false)
1227                }
1228                ConfigVal::String(default) => {
1229                    VarDefinition::new_runtime(cfg.name(), default.clone(), cfg.desc(), false)
1230                }
1231                ConfigVal::Duration(default) => {
1232                    VarDefinition::new_runtime(cfg.name(), default.clone(), cfg.desc(), false)
1233                }
1234                ConfigVal::Json(default) => {
1235                    VarDefinition::new_runtime(cfg.name(), default.clone(), cfg.desc(), false)
1236                }
1237            })
1238            .collect();
1239
1240        let vars: BTreeMap<_, _> = system_vars
1241            .into_iter()
1242            // Include all of our feature flags.
1243            .chain(definitions::FEATURE_FLAGS.iter().copied())
1244            // Include the subset of Session variables we allow system defaults for.
1245            .chain(SESSION_SYSTEM_VARS.values().copied())
1246            .cloned()
1247            // Include Persist configs.
1248            .chain(dyncfg_vars)
1249            .map(|var| (var.name, SystemVar::new(var)))
1250            .collect();
1251
1252        let vars = SystemVars {
1253            vars,
1254            callbacks: BTreeMap::new(),
1255            allow_unsafe: false,
1256            dyncfgs,
1257        };
1258
1259        vars
1260    }
1261
1262    pub fn dyncfgs(&self) -> &ConfigSet {
1263        &self.dyncfgs
1264    }
1265
1266    pub fn set_unsafe(mut self, allow_unsafe: bool) -> Self {
1267        self.allow_unsafe = allow_unsafe;
1268        self
1269    }
1270
1271    pub fn allow_unsafe(&self) -> bool {
1272        self.allow_unsafe
1273    }
1274
1275    fn expect_value<V: 'static>(&self, var: &VarDefinition) -> &V {
1276        let val = self
1277            .vars
1278            .get(var.name)
1279            .expect("provided var should be in state");
1280
1281        val.value_dyn()
1282            .as_any()
1283            .downcast_ref::<V>()
1284            .expect("provided var type should matched stored var")
1285    }
1286
1287    fn expect_config_value<V: ConfigType + 'static>(&self, name: &UncasedStr) -> &V {
1288        let val = self
1289            .vars
1290            .get(name)
1291            .unwrap_or_else(|| panic!("provided var {name} should be in state"));
1292
1293        val.value_dyn()
1294            .as_any()
1295            .downcast_ref()
1296            .expect("provided var type should matched stored var")
1297    }
1298
1299    /// Reset all the values to their defaults (preserving
1300    /// defaults from `VarMut::set_default).
1301    pub fn reset_all(&mut self) {
1302        for (_, var) in &mut self.vars {
1303            var.reset();
1304        }
1305    }
1306
1307    /// Returns an iterator over the configuration parameters and their current
1308    /// values on disk.
1309    pub fn iter(&self) -> impl Iterator<Item = &dyn Var> {
1310        self.vars
1311            .values()
1312            .map(|v| v.as_var())
1313            .filter(|v| !SESSION_SYSTEM_VARS.contains_key(UncasedStr::new(v.name())))
1314    }
1315
1316    /// Returns an iterator over the configuration parameters and their current
1317    /// values on disk. Compared to [`SystemVars::iter`], this should omit vars
1318    /// that shouldn't be synced by SystemParameterFrontend.
1319    pub fn iter_synced(&self) -> impl Iterator<Item = &dyn Var> {
1320        self.iter().filter(|v| v.name() != ENABLE_LAUNCHDARKLY.name)
1321    }
1322
1323    /// Returns an iterator over the configuration parameters that can be overriden per-Session.
1324    pub fn iter_session(&self) -> impl Iterator<Item = &dyn Var> {
1325        self.vars
1326            .values()
1327            .map(|v| v.as_var())
1328            .filter(|v| SESSION_SYSTEM_VARS.contains_key(UncasedStr::new(v.name())))
1329    }
1330
1331    /// Returns whether or not this parameter can be modified by a superuser.
1332    pub fn user_modifiable(&self, name: &str) -> bool {
1333        SESSION_SYSTEM_VARS.contains_key(UncasedStr::new(name))
1334            || name == ENABLE_RBAC_CHECKS.name()
1335            || name == NETWORK_POLICY.name()
1336    }
1337
1338    /// Returns a [`Var`] representing the configuration parameter with the
1339    /// specified name.
1340    ///
1341    /// Configuration parameters are matched case insensitively. If no such
1342    /// configuration parameter exists, `get` returns an error.
1343    ///
1344    /// Note that:
1345    /// - If `name` is known at compile time, you should instead use the named
1346    /// accessor to access the variable with its true Rust type. For example,
1347    /// `self.get("max_tables").value()` returns the string `"25"` or the
1348    /// current value, while `self.max_tables()` returns an i32.
1349    ///
1350    /// - This function does not check that the access variable should be
1351    /// visible because of other settings or users. Before or after accessing
1352    /// this method, you should call `Var::visible`.
1353    ///
1354    /// # Errors
1355    ///
1356    /// The call will return an error:
1357    /// 1. If `name` does not refer to a valid [`SystemVars`] field.
1358    pub fn get(&self, name: &str) -> Result<&dyn Var, VarError> {
1359        self.vars
1360            .get(UncasedStr::new(name))
1361            .map(|v| v.as_var())
1362            .ok_or_else(|| VarError::UnknownParameter(name.into()))
1363    }
1364
1365    /// Check if the given `values` is the default value for the [`Var`]
1366    /// identified by `name`.
1367    ///
1368    /// Note that this function does not check that the access variable should
1369    /// be visible because of other settings or users. Before or after accessing
1370    /// this method, you should call `Var::visible`.
1371    ///
1372    /// # Errors
1373    ///
1374    /// The call will return an error:
1375    /// 1. If `name` does not refer to a valid [`SystemVars`] field.
1376    /// 2. If `values` does not represent a valid [`SystemVars`] value for
1377    ///    `name`.
1378    pub fn is_default(&self, name: &str, input: VarInput) -> Result<bool, VarError> {
1379        self.vars
1380            .get(UncasedStr::new(name))
1381            .ok_or_else(|| VarError::UnknownParameter(name.into()))
1382            .and_then(|v| v.is_default(input))
1383    }
1384
1385    /// Sets the configuration parameter named `name` to the value represented
1386    /// by `input`.
1387    ///
1388    /// Like with [`SystemVars::get`], configuration parameters are matched case
1389    /// insensitively. If `input` is not valid, as determined by the underlying
1390    /// configuration parameter, or if the named configuration parameter does
1391    /// not exist, an error is returned.
1392    ///
1393    /// Return a `bool` value indicating whether the [`Var`] identified by
1394    /// `name` was modified by this call (it won't be if it already had the
1395    /// given `input`).
1396    ///
1397    /// Note that this function does not check that the access variable should
1398    /// be visible because of other settings or users. Before or after accessing
1399    /// this method, you should call `Var::visible`.
1400    ///
1401    /// # Errors
1402    ///
1403    /// The call will return an error:
1404    /// 1. If `name` does not refer to a valid [`SystemVars`] field.
1405    /// 2. If `input` does not represent a valid [`SystemVars`] value for
1406    ///    `name`.
1407    pub fn set(&mut self, name: &str, input: VarInput) -> Result<bool, VarError> {
1408        let result = self
1409            .vars
1410            .get_mut(UncasedStr::new(name))
1411            .ok_or_else(|| VarError::UnknownParameter(name.into()))
1412            .and_then(|v| v.set(input))?;
1413        self.notify_callbacks(name);
1414        Ok(result)
1415    }
1416
1417    /// Parses the configuration parameter value represented by `input` named
1418    /// `name`.
1419    ///
1420    /// Like with [`SystemVars::get`], configuration parameters are matched case
1421    /// insensitively. If `input` is not valid, as determined by the underlying
1422    /// configuration parameter, or if the named configuration parameter does
1423    /// not exist, an error is returned.
1424    ///
1425    /// Return a `Box<dyn Value>` that is the result of parsing `input`.
1426    ///
1427    /// Note that this function does not check that the access variable should
1428    /// be visible because of other settings or users. Before or after accessing
1429    /// this method, you should call `Var::visible`.
1430    ///
1431    /// # Errors
1432    ///
1433    /// The call will return an error:
1434    /// 1. If `name` does not refer to a valid [`SystemVars`] field.
1435    /// 2. If `input` does not represent a valid [`SystemVars`] value for
1436    ///    `name`.
1437    pub fn parse(&self, name: &str, input: VarInput) -> Result<Box<dyn Value>, VarError> {
1438        self.vars
1439            .get(UncasedStr::new(name))
1440            .ok_or_else(|| VarError::UnknownParameter(name.into()))
1441            .and_then(|v| v.parse(input))
1442    }
1443
1444    /// Set the default for this variable. This is the value this
1445    /// variable will be be `reset` to. If no default is set, the static default in the
1446    /// variable definition is used instead.
1447    ///
1448    /// Note that this function does not check that the access variable should
1449    /// be visible because of other settings or users. Before or after accessing
1450    /// this method, you should call `Var::visible`.
1451    pub fn set_default(&mut self, name: &str, input: VarInput) -> Result<(), VarError> {
1452        let result = self
1453            .vars
1454            .get_mut(UncasedStr::new(name))
1455            .ok_or_else(|| VarError::UnknownParameter(name.into()))
1456            .and_then(|v| v.set_default(input))?;
1457        self.notify_callbacks(name);
1458        Ok(result)
1459    }
1460
1461    /// Sets the configuration parameter named `name` to its default value.
1462    ///
1463    /// Like with [`SystemVars::get`], configuration parameters are matched case
1464    /// insensitively. If the named configuration parameter does not exist, an
1465    /// error is returned.
1466    ///
1467    /// Return a `bool` value indicating whether the [`Var`] identified by
1468    /// `name` was modified by this call (it won't be if was already reset).
1469    ///
1470    /// Note that this function does not check that the access variable should
1471    /// be visible because of other settings or users. Before or after accessing
1472    /// this method, you should call `Var::visible`.
1473    ///
1474    /// # Errors
1475    ///
1476    /// The call will return an error:
1477    /// 1. If `name` does not refer to a valid [`SystemVars`] field.
1478    pub fn reset(&mut self, name: &str) -> Result<bool, VarError> {
1479        let result = self
1480            .vars
1481            .get_mut(UncasedStr::new(name))
1482            .ok_or_else(|| VarError::UnknownParameter(name.into()))
1483            .map(|v| v.reset())?;
1484        self.notify_callbacks(name);
1485        Ok(result)
1486    }
1487
1488    /// Returns a map from each system parameter's name to its default value.
1489    pub fn defaults(&self) -> BTreeMap<String, String> {
1490        self.vars
1491            .iter()
1492            .map(|(name, var)| {
1493                let default = var
1494                    .dynamic_default
1495                    .as_deref()
1496                    .unwrap_or_else(|| var.definition.default_value());
1497                (name.as_str().to_owned(), default.format())
1498            })
1499            .collect()
1500    }
1501
1502    /// Registers a closure that will get called when the value for the
1503    /// specified [`VarDefinition`] changes.
1504    ///
1505    /// The callback is guaranteed to be called at least once.
1506    pub fn register_callback(
1507        &mut self,
1508        var: &VarDefinition,
1509        callback: Arc<dyn Fn(&SystemVars) + Send + Sync>,
1510    ) {
1511        self.callbacks
1512            .entry(var.name().to_string())
1513            .or_default()
1514            .push(callback);
1515        self.notify_callbacks(var.name());
1516    }
1517
1518    /// Notify any external components interested in this variable.
1519    fn notify_callbacks(&self, name: &str) {
1520        // Get the callbacks interested in this variable.
1521        if let Some(callbacks) = self.callbacks.get(name) {
1522            for callback in callbacks {
1523                (callback)(self);
1524            }
1525        }
1526    }
1527
1528    /// Returns the system default for the [`CLUSTER`] session variable. To know the active cluster
1529    /// for the current session, you must check the [`SessionVars`].
1530    pub fn default_cluster(&self) -> String {
1531        self.expect_value::<String>(&CLUSTER).to_owned()
1532    }
1533
1534    /// Returns the value of the `max_kafka_connections` configuration parameter.
1535    pub fn max_kafka_connections(&self) -> u32 {
1536        *self.expect_value(&MAX_KAFKA_CONNECTIONS)
1537    }
1538
1539    /// Returns the value of the `max_postgres_connections` configuration parameter.
1540    pub fn max_postgres_connections(&self) -> u32 {
1541        *self.expect_value(&MAX_POSTGRES_CONNECTIONS)
1542    }
1543
1544    /// Returns the value of the `max_mysql_connections` configuration parameter.
1545    pub fn max_mysql_connections(&self) -> u32 {
1546        *self.expect_value(&MAX_MYSQL_CONNECTIONS)
1547    }
1548
1549    /// Returns the value of the `max_sql_server_connections` configuration parameter.
1550    pub fn max_sql_server_connections(&self) -> u32 {
1551        *self.expect_value(&MAX_SQL_SERVER_CONNECTIONS)
1552    }
1553
1554    /// Returns the value of the `max_aws_privatelink_connections` configuration parameter.
1555    pub fn max_aws_privatelink_connections(&self) -> u32 {
1556        *self.expect_value(&MAX_AWS_PRIVATELINK_CONNECTIONS)
1557    }
1558
1559    /// Returns the value of the `max_tables` configuration parameter.
1560    pub fn max_tables(&self) -> u32 {
1561        *self.expect_value(&MAX_TABLES)
1562    }
1563
1564    /// Returns the value of the `max_sources` configuration parameter.
1565    pub fn max_sources(&self) -> u32 {
1566        *self.expect_value(&MAX_SOURCES)
1567    }
1568
1569    /// Returns the value of the `max_sinks` configuration parameter.
1570    pub fn max_sinks(&self) -> u32 {
1571        *self.expect_value(&MAX_SINKS)
1572    }
1573
1574    /// Returns the value of the `max_materialized_views` configuration parameter.
1575    pub fn max_materialized_views(&self) -> u32 {
1576        *self.expect_value(&MAX_MATERIALIZED_VIEWS)
1577    }
1578
1579    /// Returns the value of the `max_clusters` configuration parameter.
1580    pub fn max_clusters(&self) -> u32 {
1581        *self.expect_value(&MAX_CLUSTERS)
1582    }
1583
1584    /// Returns the value of the `max_replicas_per_cluster` configuration parameter.
1585    pub fn max_replicas_per_cluster(&self) -> u32 {
1586        *self.expect_value(&MAX_REPLICAS_PER_CLUSTER)
1587    }
1588
1589    /// Returns the value of the `max_credit_consumption_rate` configuration parameter.
1590    pub fn max_credit_consumption_rate(&self) -> Numeric {
1591        *self.expect_value(&MAX_CREDIT_CONSUMPTION_RATE)
1592    }
1593
1594    /// Returns the value of the `max_databases` configuration parameter.
1595    pub fn max_databases(&self) -> u32 {
1596        *self.expect_value(&MAX_DATABASES)
1597    }
1598
1599    /// Returns the value of the `max_schemas_per_database` configuration parameter.
1600    pub fn max_schemas_per_database(&self) -> u32 {
1601        *self.expect_value(&MAX_SCHEMAS_PER_DATABASE)
1602    }
1603
1604    /// Returns the value of the `max_objects_per_schema` configuration parameter.
1605    pub fn max_objects_per_schema(&self) -> u32 {
1606        *self.expect_value(&MAX_OBJECTS_PER_SCHEMA)
1607    }
1608
1609    /// Returns the value of the `max_secrets` configuration parameter.
1610    pub fn max_secrets(&self) -> u32 {
1611        *self.expect_value(&MAX_SECRETS)
1612    }
1613
1614    /// Returns the value of the `max_roles` configuration parameter.
1615    pub fn max_roles(&self) -> u32 {
1616        *self.expect_value(&MAX_ROLES)
1617    }
1618
1619    /// Returns the value of the `max_continual_tasks` configuration parameter.
1620    pub fn max_continual_tasks(&self) -> u32 {
1621        *self.expect_value(&MAX_CONTINUAL_TASKS)
1622    }
1623
1624    /// Returns the value of the `max_network_policies` configuration parameter.
1625    pub fn max_network_policies(&self) -> u32 {
1626        *self.expect_value(&MAX_NETWORK_POLICIES)
1627    }
1628
1629    /// Returns the value of the `max_network_policies` configuration parameter.
1630    pub fn max_rules_per_network_policy(&self) -> u32 {
1631        *self.expect_value(&MAX_RULES_PER_NETWORK_POLICY)
1632    }
1633
1634    /// Returns the value of the `max_result_size` configuration parameter.
1635    pub fn max_result_size(&self) -> u64 {
1636        self.expect_value::<ByteSize>(&MAX_RESULT_SIZE).as_bytes()
1637    }
1638
1639    /// Returns the value of the `max_copy_from_size` configuration parameter.
1640    pub fn max_copy_from_size(&self) -> u32 {
1641        *self.expect_value(&MAX_COPY_FROM_SIZE)
1642    }
1643
1644    /// Returns the value of the `allowed_cluster_replica_sizes` configuration parameter.
1645    pub fn allowed_cluster_replica_sizes(&self) -> Vec<String> {
1646        self.expect_value::<Vec<Ident>>(&ALLOWED_CLUSTER_REPLICA_SIZES)
1647            .into_iter()
1648            .map(|s| s.as_str().into())
1649            .collect()
1650    }
1651
1652    /// Returns the value of the `default_cluster_replication_factor` configuration parameter.
1653    pub fn default_cluster_replication_factor(&self) -> u32 {
1654        *self.expect_value::<u32>(&DEFAULT_CLUSTER_REPLICATION_FACTOR)
1655    }
1656
1657    /// Returns the `disk_cluster_replicas_default` configuration parameter.
1658    pub fn disk_cluster_replicas_default(&self) -> bool {
1659        *self.expect_value(&DISK_CLUSTER_REPLICAS_DEFAULT)
1660    }
1661
1662    pub fn upsert_rocksdb_auto_spill_to_disk(&self) -> bool {
1663        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_AUTO_SPILL_TO_DISK)
1664    }
1665
1666    pub fn upsert_rocksdb_auto_spill_threshold_bytes(&self) -> usize {
1667        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_AUTO_SPILL_THRESHOLD_BYTES)
1668    }
1669
1670    pub fn upsert_rocksdb_compaction_style(&self) -> mz_rocksdb_types::config::CompactionStyle {
1671        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_COMPACTION_STYLE)
1672    }
1673
1674    pub fn upsert_rocksdb_optimize_compaction_memtable_budget(&self) -> usize {
1675        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_OPTIMIZE_COMPACTION_MEMTABLE_BUDGET)
1676    }
1677
1678    pub fn upsert_rocksdb_level_compaction_dynamic_level_bytes(&self) -> bool {
1679        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_LEVEL_COMPACTION_DYNAMIC_LEVEL_BYTES)
1680    }
1681
1682    pub fn upsert_rocksdb_universal_compaction_ratio(&self) -> i32 {
1683        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_UNIVERSAL_COMPACTION_RATIO)
1684    }
1685
1686    pub fn upsert_rocksdb_parallelism(&self) -> Option<i32> {
1687        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_PARALLELISM)
1688    }
1689
1690    pub fn upsert_rocksdb_compression_type(&self) -> mz_rocksdb_types::config::CompressionType {
1691        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_COMPRESSION_TYPE)
1692    }
1693
1694    pub fn upsert_rocksdb_bottommost_compression_type(
1695        &self,
1696    ) -> mz_rocksdb_types::config::CompressionType {
1697        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_BOTTOMMOST_COMPRESSION_TYPE)
1698    }
1699
1700    pub fn upsert_rocksdb_batch_size(&self) -> usize {
1701        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_BATCH_SIZE)
1702    }
1703
1704    pub fn upsert_rocksdb_retry_duration(&self) -> Duration {
1705        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_RETRY_DURATION)
1706    }
1707
1708    pub fn upsert_rocksdb_stats_log_interval_seconds(&self) -> u32 {
1709        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_STATS_LOG_INTERVAL_SECONDS)
1710    }
1711
1712    pub fn upsert_rocksdb_stats_persist_interval_seconds(&self) -> u32 {
1713        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_STATS_PERSIST_INTERVAL_SECONDS)
1714    }
1715
1716    pub fn upsert_rocksdb_point_lookup_block_cache_size_mb(&self) -> Option<u32> {
1717        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_POINT_LOOKUP_BLOCK_CACHE_SIZE_MB)
1718    }
1719
1720    pub fn upsert_rocksdb_shrink_allocated_buffers_by_ratio(&self) -> usize {
1721        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_SHRINK_ALLOCATED_BUFFERS_BY_RATIO)
1722    }
1723
1724    pub fn upsert_rocksdb_write_buffer_manager_cluster_memory_fraction(&self) -> Option<Numeric> {
1725        *self.expect_value(
1726            &upsert_rocksdb::UPSERT_ROCKSDB_WRITE_BUFFER_MANAGER_CLUSTER_MEMORY_FRACTION,
1727        )
1728    }
1729
1730    pub fn upsert_rocksdb_write_buffer_manager_memory_bytes(&self) -> Option<usize> {
1731        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_WRITE_BUFFER_MANAGER_MEMORY_BYTES)
1732    }
1733
1734    pub fn upsert_rocksdb_write_buffer_manager_allow_stall(&self) -> bool {
1735        *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_WRITE_BUFFER_MANAGER_ALLOW_STALL)
1736    }
1737
1738    pub fn persist_fast_path_limit(&self) -> usize {
1739        *self.expect_value(&PERSIST_FAST_PATH_LIMIT)
1740    }
1741
1742    /// Returns the `pg_source_connect_timeout` configuration parameter.
1743    pub fn pg_source_connect_timeout(&self) -> Duration {
1744        *self.expect_value(&PG_SOURCE_CONNECT_TIMEOUT)
1745    }
1746
1747    /// Returns the `pg_source_tcp_keepalives_retries` configuration parameter.
1748    pub fn pg_source_tcp_keepalives_retries(&self) -> u32 {
1749        *self.expect_value(&PG_SOURCE_TCP_KEEPALIVES_RETRIES)
1750    }
1751
1752    /// Returns the `pg_source_tcp_keepalives_idle` configuration parameter.
1753    pub fn pg_source_tcp_keepalives_idle(&self) -> Duration {
1754        *self.expect_value(&PG_SOURCE_TCP_KEEPALIVES_IDLE)
1755    }
1756
1757    /// Returns the `pg_source_tcp_keepalives_interval` configuration parameter.
1758    pub fn pg_source_tcp_keepalives_interval(&self) -> Duration {
1759        *self.expect_value(&PG_SOURCE_TCP_KEEPALIVES_INTERVAL)
1760    }
1761
1762    /// Returns the `pg_source_tcp_user_timeout` configuration parameter.
1763    pub fn pg_source_tcp_user_timeout(&self) -> Duration {
1764        *self.expect_value(&PG_SOURCE_TCP_USER_TIMEOUT)
1765    }
1766
1767    /// Returns the `pg_source_tcp_configure_server` configuration parameter.
1768    pub fn pg_source_tcp_configure_server(&self) -> bool {
1769        *self.expect_value(&PG_SOURCE_TCP_CONFIGURE_SERVER)
1770    }
1771
1772    /// Returns the `pg_source_snapshot_statement_timeout` configuration parameter.
1773    pub fn pg_source_snapshot_statement_timeout(&self) -> Duration {
1774        *self.expect_value(&PG_SOURCE_SNAPSHOT_STATEMENT_TIMEOUT)
1775    }
1776
1777    /// Returns the `pg_source_wal_sender_timeout` configuration parameter.
1778    pub fn pg_source_wal_sender_timeout(&self) -> Option<Duration> {
1779        *self.expect_value(&PG_SOURCE_WAL_SENDER_TIMEOUT)
1780    }
1781
1782    /// Returns the `pg_source_snapshot_collect_strict_count` configuration parameter.
1783    pub fn pg_source_snapshot_collect_strict_count(&self) -> bool {
1784        *self.expect_value(&PG_SOURCE_SNAPSHOT_COLLECT_STRICT_COUNT)
1785    }
1786
1787    /// Returns the `mysql_source_tcp_keepalive` configuration parameter.
1788    pub fn mysql_source_tcp_keepalive(&self) -> Duration {
1789        *self.expect_value(&MYSQL_SOURCE_TCP_KEEPALIVE)
1790    }
1791
1792    /// Returns the `mysql_source_snapshot_max_execution_time` configuration parameter.
1793    pub fn mysql_source_snapshot_max_execution_time(&self) -> Duration {
1794        *self.expect_value(&MYSQL_SOURCE_SNAPSHOT_MAX_EXECUTION_TIME)
1795    }
1796
1797    /// Returns the `mysql_source_snapshot_lock_wait_timeout` configuration parameter.
1798    pub fn mysql_source_snapshot_lock_wait_timeout(&self) -> Duration {
1799        *self.expect_value(&MYSQL_SOURCE_SNAPSHOT_LOCK_WAIT_TIMEOUT)
1800    }
1801
1802    /// Returns the `mysql_source_connect_timeout` configuration parameter.
1803    pub fn mysql_source_connect_timeout(&self) -> Duration {
1804        *self.expect_value(&MYSQL_SOURCE_CONNECT_TIMEOUT)
1805    }
1806
1807    /// Returns the `ssh_check_interval` configuration parameter.
1808    pub fn ssh_check_interval(&self) -> Duration {
1809        *self.expect_value(&SSH_CHECK_INTERVAL)
1810    }
1811
1812    /// Returns the `ssh_connect_timeout` configuration parameter.
1813    pub fn ssh_connect_timeout(&self) -> Duration {
1814        *self.expect_value(&SSH_CONNECT_TIMEOUT)
1815    }
1816
1817    /// Returns the `ssh_keepalives_idle` configuration parameter.
1818    pub fn ssh_keepalives_idle(&self) -> Duration {
1819        *self.expect_value(&SSH_KEEPALIVES_IDLE)
1820    }
1821
1822    /// Returns the `kafka_socket_keepalive` configuration parameter.
1823    pub fn kafka_socket_keepalive(&self) -> bool {
1824        *self.expect_value(&KAFKA_SOCKET_KEEPALIVE)
1825    }
1826
1827    /// Returns the `kafka_socket_timeout` configuration parameter.
1828    pub fn kafka_socket_timeout(&self) -> Option<Duration> {
1829        *self.expect_value(&KAFKA_SOCKET_TIMEOUT)
1830    }
1831
1832    /// Returns the `kafka_transaction_timeout` configuration parameter.
1833    pub fn kafka_transaction_timeout(&self) -> Duration {
1834        *self.expect_value(&KAFKA_TRANSACTION_TIMEOUT)
1835    }
1836
1837    /// Returns the `kafka_socket_connection_setup_timeout` configuration parameter.
1838    pub fn kafka_socket_connection_setup_timeout(&self) -> Duration {
1839        *self.expect_value(&KAFKA_SOCKET_CONNECTION_SETUP_TIMEOUT)
1840    }
1841
1842    /// Returns the `kafka_fetch_metadata_timeout` configuration parameter.
1843    pub fn kafka_fetch_metadata_timeout(&self) -> Duration {
1844        *self.expect_value(&KAFKA_FETCH_METADATA_TIMEOUT)
1845    }
1846
1847    /// Returns the `kafka_progress_record_fetch_timeout` configuration parameter.
1848    pub fn kafka_progress_record_fetch_timeout(&self) -> Option<Duration> {
1849        *self.expect_value(&KAFKA_PROGRESS_RECORD_FETCH_TIMEOUT)
1850    }
1851
1852    /// Returns the `crdb_connect_timeout` configuration parameter.
1853    pub fn crdb_connect_timeout(&self) -> Duration {
1854        *self.expect_config_value(UncasedStr::new(
1855            mz_persist_client::cfg::CRDB_CONNECT_TIMEOUT.name(),
1856        ))
1857    }
1858
1859    /// Returns the `crdb_tcp_user_timeout` configuration parameter.
1860    pub fn crdb_tcp_user_timeout(&self) -> Duration {
1861        *self.expect_config_value(UncasedStr::new(
1862            mz_persist_client::cfg::CRDB_TCP_USER_TIMEOUT.name(),
1863        ))
1864    }
1865
1866    /// Returns the `storage_dataflow_max_inflight_bytes` configuration parameter.
1867    pub fn storage_dataflow_max_inflight_bytes(&self) -> Option<usize> {
1868        *self.expect_value(&STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES)
1869    }
1870
1871    /// Returns the `storage_dataflow_max_inflight_bytes_to_cluster_size_fraction` configuration parameter.
1872    pub fn storage_dataflow_max_inflight_bytes_to_cluster_size_fraction(&self) -> Option<Numeric> {
1873        *self.expect_value(&STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES_TO_CLUSTER_SIZE_FRACTION)
1874    }
1875
1876    /// Returns the `storage_shrink_upsert_unused_buffers_by_ratio` configuration parameter.
1877    pub fn storage_shrink_upsert_unused_buffers_by_ratio(&self) -> usize {
1878        *self.expect_value(&STORAGE_SHRINK_UPSERT_UNUSED_BUFFERS_BY_RATIO)
1879    }
1880
1881    /// Returns the `storage_dataflow_max_inflight_bytes_disk_only` configuration parameter.
1882    pub fn storage_dataflow_max_inflight_bytes_disk_only(&self) -> bool {
1883        *self.expect_value(&STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES_DISK_ONLY)
1884    }
1885
1886    /// Returns the `storage_statistics_interval` configuration parameter.
1887    pub fn storage_statistics_interval(&self) -> Duration {
1888        *self.expect_value(&STORAGE_STATISTICS_INTERVAL)
1889    }
1890
1891    /// Returns the `storage_statistics_collection_interval` configuration parameter.
1892    pub fn storage_statistics_collection_interval(&self) -> Duration {
1893        *self.expect_value(&STORAGE_STATISTICS_COLLECTION_INTERVAL)
1894    }
1895
1896    /// Returns the `storage_record_source_sink_namespaced_errors` configuration parameter.
1897    pub fn storage_record_source_sink_namespaced_errors(&self) -> bool {
1898        *self.expect_value(&STORAGE_RECORD_SOURCE_SINK_NAMESPACED_ERRORS)
1899    }
1900
1901    /// Returns the `persist_stats_filter_enabled` configuration parameter.
1902    pub fn persist_stats_filter_enabled(&self) -> bool {
1903        *self.expect_config_value(UncasedStr::new(
1904            mz_persist_client::stats::STATS_FILTER_ENABLED.name(),
1905        ))
1906    }
1907
1908    pub fn dyncfg_updates(&self) -> ConfigUpdates {
1909        let mut updates = ConfigUpdates::default();
1910        for entry in self.dyncfgs.entries() {
1911            let name = UncasedStr::new(entry.name());
1912            let val = match entry.val() {
1913                ConfigVal::Bool(_) => ConfigVal::from(*self.expect_config_value::<bool>(name)),
1914                ConfigVal::U32(_) => ConfigVal::from(*self.expect_config_value::<u32>(name)),
1915                ConfigVal::Usize(_) => ConfigVal::from(*self.expect_config_value::<usize>(name)),
1916                ConfigVal::OptUsize(_) => {
1917                    ConfigVal::from(*self.expect_config_value::<Option<usize>>(name))
1918                }
1919                ConfigVal::F64(_) => ConfigVal::from(*self.expect_config_value::<f64>(name)),
1920                ConfigVal::String(_) => {
1921                    ConfigVal::from(self.expect_config_value::<String>(name).clone())
1922                }
1923                ConfigVal::Duration(_) => {
1924                    ConfigVal::from(*self.expect_config_value::<Duration>(name))
1925                }
1926                ConfigVal::Json(_) => {
1927                    ConfigVal::from(self.expect_config_value::<serde_json::Value>(name).clone())
1928                }
1929            };
1930            updates.add_dynamic(entry.name(), val);
1931        }
1932        updates.apply(&self.dyncfgs);
1933        updates
1934    }
1935
1936    /// Returns the `metrics_retention` configuration parameter.
1937    pub fn metrics_retention(&self) -> Duration {
1938        *self.expect_value(&METRICS_RETENTION)
1939    }
1940
1941    /// Returns the `unsafe_mock_audit_event_timestamp` configuration parameter.
1942    pub fn unsafe_mock_audit_event_timestamp(&self) -> Option<mz_repr::Timestamp> {
1943        *self.expect_value(&UNSAFE_MOCK_AUDIT_EVENT_TIMESTAMP)
1944    }
1945
1946    /// Returns the `enable_rbac_checks` configuration parameter.
1947    pub fn enable_rbac_checks(&self) -> bool {
1948        *self.expect_value(&ENABLE_RBAC_CHECKS)
1949    }
1950
1951    /// Returns the `max_connections` configuration parameter.
1952    pub fn max_connections(&self) -> u32 {
1953        *self.expect_value(&MAX_CONNECTIONS)
1954    }
1955
1956    pub fn default_network_policy_name(&self) -> String {
1957        self.expect_value::<String>(&NETWORK_POLICY).clone()
1958    }
1959
1960    /// Returns the `superuser_reserved_connections` configuration parameter.
1961    pub fn superuser_reserved_connections(&self) -> u32 {
1962        *self.expect_value(&SUPERUSER_RESERVED_CONNECTIONS)
1963    }
1964
1965    pub fn keep_n_source_status_history_entries(&self) -> usize {
1966        *self.expect_value(&KEEP_N_SOURCE_STATUS_HISTORY_ENTRIES)
1967    }
1968
1969    pub fn keep_n_sink_status_history_entries(&self) -> usize {
1970        *self.expect_value(&KEEP_N_SINK_STATUS_HISTORY_ENTRIES)
1971    }
1972
1973    pub fn keep_n_privatelink_status_history_entries(&self) -> usize {
1974        *self.expect_value(&KEEP_N_PRIVATELINK_STATUS_HISTORY_ENTRIES)
1975    }
1976
1977    pub fn replica_status_history_retention_window(&self) -> Duration {
1978        *self.expect_value(&REPLICA_STATUS_HISTORY_RETENTION_WINDOW)
1979    }
1980
1981    /// Returns the `enable_storage_shard_finalization` configuration parameter.
1982    pub fn enable_storage_shard_finalization(&self) -> bool {
1983        *self.expect_value(&ENABLE_STORAGE_SHARD_FINALIZATION)
1984    }
1985
1986    pub fn enable_consolidate_after_union_negate(&self) -> bool {
1987        *self.expect_value(&ENABLE_CONSOLIDATE_AFTER_UNION_NEGATE)
1988    }
1989
1990    pub fn enable_reduce_reduction(&self) -> bool {
1991        *self.expect_value(&ENABLE_REDUCE_REDUCTION)
1992    }
1993
1994    /// Returns the `enable_default_connection_validation` configuration parameter.
1995    pub fn enable_default_connection_validation(&self) -> bool {
1996        *self.expect_value(&ENABLE_DEFAULT_CONNECTION_VALIDATION)
1997    }
1998
1999    /// Returns the `min_timestamp_interval` configuration parameter.
2000    pub fn min_timestamp_interval(&self) -> Duration {
2001        *self.expect_value(&MIN_TIMESTAMP_INTERVAL)
2002    }
2003    /// Returns the `max_timestamp_interval` configuration parameter.
2004    pub fn max_timestamp_interval(&self) -> Duration {
2005        *self.expect_value(&MAX_TIMESTAMP_INTERVAL)
2006    }
2007
2008    pub fn logging_filter(&self) -> CloneableEnvFilter {
2009        self.expect_value::<CloneableEnvFilter>(&LOGGING_FILTER)
2010            .clone()
2011    }
2012
2013    pub fn opentelemetry_filter(&self) -> CloneableEnvFilter {
2014        self.expect_value::<CloneableEnvFilter>(&OPENTELEMETRY_FILTER)
2015            .clone()
2016    }
2017
2018    pub fn logging_filter_defaults(&self) -> Vec<SerializableDirective> {
2019        self.expect_value::<Vec<SerializableDirective>>(&LOGGING_FILTER_DEFAULTS)
2020            .clone()
2021    }
2022
2023    pub fn opentelemetry_filter_defaults(&self) -> Vec<SerializableDirective> {
2024        self.expect_value::<Vec<SerializableDirective>>(&OPENTELEMETRY_FILTER_DEFAULTS)
2025            .clone()
2026    }
2027
2028    pub fn sentry_filters(&self) -> Vec<SerializableDirective> {
2029        self.expect_value::<Vec<SerializableDirective>>(&SENTRY_FILTERS)
2030            .clone()
2031    }
2032
2033    pub fn webhooks_secrets_caching_ttl_secs(&self) -> usize {
2034        *self.expect_value(&WEBHOOKS_SECRETS_CACHING_TTL_SECS)
2035    }
2036
2037    pub fn coord_slow_message_warn_threshold(&self) -> Duration {
2038        *self.expect_value(&COORD_SLOW_MESSAGE_WARN_THRESHOLD)
2039    }
2040
2041    pub fn grpc_client_http2_keep_alive_interval(&self) -> Duration {
2042        *self.expect_value(&grpc_client::HTTP2_KEEP_ALIVE_INTERVAL)
2043    }
2044
2045    pub fn grpc_client_http2_keep_alive_timeout(&self) -> Duration {
2046        *self.expect_value(&grpc_client::HTTP2_KEEP_ALIVE_TIMEOUT)
2047    }
2048
2049    pub fn grpc_connect_timeout(&self) -> Duration {
2050        *self.expect_value(&grpc_client::CONNECT_TIMEOUT)
2051    }
2052
2053    pub fn cluster_multi_process_replica_az_affinity_weight(&self) -> Option<i32> {
2054        *self.expect_value(&cluster_scheduling::CLUSTER_MULTI_PROCESS_REPLICA_AZ_AFFINITY_WEIGHT)
2055    }
2056
2057    pub fn cluster_soften_replication_anti_affinity(&self) -> bool {
2058        *self.expect_value(&cluster_scheduling::CLUSTER_SOFTEN_REPLICATION_ANTI_AFFINITY)
2059    }
2060
2061    pub fn cluster_soften_replication_anti_affinity_weight(&self) -> i32 {
2062        *self.expect_value(&cluster_scheduling::CLUSTER_SOFTEN_REPLICATION_ANTI_AFFINITY_WEIGHT)
2063    }
2064
2065    pub fn cluster_enable_topology_spread(&self) -> bool {
2066        *self.expect_value(&cluster_scheduling::CLUSTER_ENABLE_TOPOLOGY_SPREAD)
2067    }
2068
2069    pub fn cluster_topology_spread_ignore_non_singular_scale(&self) -> bool {
2070        *self.expect_value(&cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_IGNORE_NON_SINGULAR_SCALE)
2071    }
2072
2073    pub fn cluster_topology_spread_max_skew(&self) -> i32 {
2074        *self.expect_value(&cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_MAX_SKEW)
2075    }
2076
2077    pub fn cluster_topology_spread_set_min_domains(&self) -> Option<i32> {
2078        *self.expect_value(&cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_MIN_DOMAINS)
2079    }
2080
2081    pub fn cluster_topology_spread_soft(&self) -> bool {
2082        *self.expect_value(&cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_SOFT)
2083    }
2084
2085    pub fn cluster_soften_az_affinity(&self) -> bool {
2086        *self.expect_value(&cluster_scheduling::CLUSTER_SOFTEN_AZ_AFFINITY)
2087    }
2088
2089    pub fn cluster_soften_az_affinity_weight(&self) -> i32 {
2090        *self.expect_value(&cluster_scheduling::CLUSTER_SOFTEN_AZ_AFFINITY_WEIGHT)
2091    }
2092
2093    pub fn cluster_always_use_disk(&self) -> bool {
2094        *self.expect_value(&cluster_scheduling::CLUSTER_ALWAYS_USE_DISK)
2095    }
2096
2097    pub fn cluster_alter_check_ready_interval(&self) -> Duration {
2098        *self.expect_value(&cluster_scheduling::CLUSTER_ALTER_CHECK_READY_INTERVAL)
2099    }
2100
2101    pub fn cluster_check_scheduling_policies_interval(&self) -> Duration {
2102        *self.expect_value(&cluster_scheduling::CLUSTER_CHECK_SCHEDULING_POLICIES_INTERVAL)
2103    }
2104
2105    pub fn cluster_security_context_enabled(&self) -> bool {
2106        *self.expect_value(&cluster_scheduling::CLUSTER_SECURITY_CONTEXT_ENABLED)
2107    }
2108
2109    pub fn cluster_refresh_mv_compaction_estimate(&self) -> Duration {
2110        *self.expect_value(&cluster_scheduling::CLUSTER_REFRESH_MV_COMPACTION_ESTIMATE)
2111    }
2112
2113    /// Returns the `privatelink_status_update_quota_per_minute` configuration parameter.
2114    pub fn privatelink_status_update_quota_per_minute(&self) -> u32 {
2115        *self.expect_value(&PRIVATELINK_STATUS_UPDATE_QUOTA_PER_MINUTE)
2116    }
2117
2118    pub fn statement_logging_target_data_rate(&self) -> Option<usize> {
2119        *self.expect_value(&STATEMENT_LOGGING_TARGET_DATA_RATE)
2120    }
2121
2122    pub fn statement_logging_max_data_credit(&self) -> Option<usize> {
2123        *self.expect_value(&STATEMENT_LOGGING_MAX_DATA_CREDIT)
2124    }
2125
2126    /// Returns the `statement_logging_max_sample_rate` configuration parameter.
2127    pub fn statement_logging_max_sample_rate(&self) -> Numeric {
2128        *self.expect_value(&STATEMENT_LOGGING_MAX_SAMPLE_RATE)
2129    }
2130
2131    /// Returns the `statement_logging_default_sample_rate` configuration parameter.
2132    pub fn statement_logging_default_sample_rate(&self) -> Numeric {
2133        *self.expect_value(&STATEMENT_LOGGING_DEFAULT_SAMPLE_RATE)
2134    }
2135
2136    /// Returns the `enable_internal_statement_logging` configuration parameter.
2137    pub fn enable_internal_statement_logging(&self) -> bool {
2138        *self.expect_value(&ENABLE_INTERNAL_STATEMENT_LOGGING)
2139    }
2140
2141    /// Returns the `optimizer_stats_timeout` configuration parameter.
2142    pub fn optimizer_stats_timeout(&self) -> Duration {
2143        *self.expect_value(&OPTIMIZER_STATS_TIMEOUT)
2144    }
2145
2146    /// Returns the `optimizer_oneshot_stats_timeout` configuration parameter.
2147    pub fn optimizer_oneshot_stats_timeout(&self) -> Duration {
2148        *self.expect_value(&OPTIMIZER_ONESHOT_STATS_TIMEOUT)
2149    }
2150
2151    /// Returns the `webhook_concurrent_request_limit` configuration parameter.
2152    pub fn webhook_concurrent_request_limit(&self) -> usize {
2153        *self.expect_value(&WEBHOOK_CONCURRENT_REQUEST_LIMIT)
2154    }
2155
2156    /// Returns the `pg_timestamp_oracle_connection_pool_max_size` configuration parameter.
2157    pub fn pg_timestamp_oracle_connection_pool_max_size(&self) -> usize {
2158        *self.expect_value(&PG_TIMESTAMP_ORACLE_CONNECTION_POOL_MAX_SIZE)
2159    }
2160
2161    /// Returns the `pg_timestamp_oracle_connection_pool_max_wait` configuration parameter.
2162    pub fn pg_timestamp_oracle_connection_pool_max_wait(&self) -> Option<Duration> {
2163        *self.expect_value(&PG_TIMESTAMP_ORACLE_CONNECTION_POOL_MAX_WAIT)
2164    }
2165
2166    /// Returns the `pg_timestamp_oracle_connection_pool_ttl` configuration parameter.
2167    pub fn pg_timestamp_oracle_connection_pool_ttl(&self) -> Duration {
2168        *self.expect_value(&PG_TIMESTAMP_ORACLE_CONNECTION_POOL_TTL)
2169    }
2170
2171    /// Returns the `pg_timestamp_oracle_connection_pool_ttl_stagger` configuration parameter.
2172    pub fn pg_timestamp_oracle_connection_pool_ttl_stagger(&self) -> Duration {
2173        *self.expect_value(&PG_TIMESTAMP_ORACLE_CONNECTION_POOL_TTL_STAGGER)
2174    }
2175
2176    /// Returns the `user_storage_managed_collections_batch_duration` configuration parameter.
2177    pub fn user_storage_managed_collections_batch_duration(&self) -> Duration {
2178        *self.expect_value(&USER_STORAGE_MANAGED_COLLECTIONS_BATCH_DURATION)
2179    }
2180
2181    pub fn force_source_table_syntax(&self) -> bool {
2182        *self.expect_value(&FORCE_SOURCE_TABLE_SYNTAX)
2183    }
2184
2185    pub fn optimizer_e2e_latency_warning_threshold(&self) -> Duration {
2186        *self.expect_value(&OPTIMIZER_E2E_LATENCY_WARNING_THRESHOLD)
2187    }
2188
2189    /// Returns whether the named variable is a compute configuration parameter
2190    /// (things that go in `ComputeParameters` and are sent to replicas via `UpdateConfiguration`
2191    /// commands).
2192    pub fn is_compute_config_var(&self, name: &str) -> bool {
2193        name == MAX_RESULT_SIZE.name() || self.is_dyncfg_var(name) || is_tracing_var(name)
2194    }
2195
2196    /// Returns whether the named variable is a metrics configuration parameter
2197    pub fn is_metrics_config_var(&self, name: &str) -> bool {
2198        self.is_dyncfg_var(name)
2199    }
2200
2201    /// Returns whether the named variable is a storage configuration parameter.
2202    pub fn is_storage_config_var(&self, name: &str) -> bool {
2203        name == PG_SOURCE_CONNECT_TIMEOUT.name()
2204            || name == PG_SOURCE_TCP_KEEPALIVES_IDLE.name()
2205            || name == PG_SOURCE_TCP_KEEPALIVES_INTERVAL.name()
2206            || name == PG_SOURCE_TCP_KEEPALIVES_RETRIES.name()
2207            || name == PG_SOURCE_TCP_USER_TIMEOUT.name()
2208            || name == PG_SOURCE_TCP_CONFIGURE_SERVER.name()
2209            || name == PG_SOURCE_SNAPSHOT_STATEMENT_TIMEOUT.name()
2210            || name == PG_SOURCE_WAL_SENDER_TIMEOUT.name()
2211            || name == PG_SOURCE_SNAPSHOT_COLLECT_STRICT_COUNT.name()
2212            || name == MYSQL_SOURCE_TCP_KEEPALIVE.name()
2213            || name == MYSQL_SOURCE_SNAPSHOT_MAX_EXECUTION_TIME.name()
2214            || name == MYSQL_SOURCE_SNAPSHOT_LOCK_WAIT_TIMEOUT.name()
2215            || name == MYSQL_SOURCE_CONNECT_TIMEOUT.name()
2216            || name == ENABLE_STORAGE_SHARD_FINALIZATION.name()
2217            || name == SSH_CHECK_INTERVAL.name()
2218            || name == SSH_CONNECT_TIMEOUT.name()
2219            || name == SSH_KEEPALIVES_IDLE.name()
2220            || name == KAFKA_SOCKET_KEEPALIVE.name()
2221            || name == KAFKA_SOCKET_TIMEOUT.name()
2222            || name == KAFKA_TRANSACTION_TIMEOUT.name()
2223            || name == KAFKA_SOCKET_CONNECTION_SETUP_TIMEOUT.name()
2224            || name == KAFKA_FETCH_METADATA_TIMEOUT.name()
2225            || name == KAFKA_PROGRESS_RECORD_FETCH_TIMEOUT.name()
2226            || name == STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES.name()
2227            || name == STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES_TO_CLUSTER_SIZE_FRACTION.name()
2228            || name == STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES_DISK_ONLY.name()
2229            || name == STORAGE_SHRINK_UPSERT_UNUSED_BUFFERS_BY_RATIO.name()
2230            || name == STORAGE_RECORD_SOURCE_SINK_NAMESPACED_ERRORS.name()
2231            || name == STORAGE_STATISTICS_INTERVAL.name()
2232            || name == STORAGE_STATISTICS_COLLECTION_INTERVAL.name()
2233            || name == USER_STORAGE_MANAGED_COLLECTIONS_BATCH_DURATION.name()
2234            || is_upsert_rocksdb_config_var(name)
2235            || self.is_dyncfg_var(name)
2236            || is_tracing_var(name)
2237    }
2238
2239    /// Returns whether the named variable is a dyncfg configuration parameter.
2240    fn is_dyncfg_var(&self, name: &str) -> bool {
2241        self.dyncfgs.entries().any(|e| name == e.name())
2242    }
2243}
2244
2245pub fn is_tracing_var(name: &str) -> bool {
2246    name == LOGGING_FILTER.name()
2247        || name == LOGGING_FILTER_DEFAULTS.name()
2248        || name == OPENTELEMETRY_FILTER.name()
2249        || name == OPENTELEMETRY_FILTER_DEFAULTS.name()
2250        || name == SENTRY_FILTERS.name()
2251}
2252
2253/// Returns whether the named variable is a caching configuration parameter.
2254pub fn is_secrets_caching_var(name: &str) -> bool {
2255    name == WEBHOOKS_SECRETS_CACHING_TTL_SECS.name()
2256}
2257
2258fn is_upsert_rocksdb_config_var(name: &str) -> bool {
2259    name == upsert_rocksdb::UPSERT_ROCKSDB_COMPACTION_STYLE.name()
2260        || name == upsert_rocksdb::UPSERT_ROCKSDB_OPTIMIZE_COMPACTION_MEMTABLE_BUDGET.name()
2261        || name == upsert_rocksdb::UPSERT_ROCKSDB_LEVEL_COMPACTION_DYNAMIC_LEVEL_BYTES.name()
2262        || name == upsert_rocksdb::UPSERT_ROCKSDB_UNIVERSAL_COMPACTION_RATIO.name()
2263        || name == upsert_rocksdb::UPSERT_ROCKSDB_PARALLELISM.name()
2264        || name == upsert_rocksdb::UPSERT_ROCKSDB_COMPRESSION_TYPE.name()
2265        || name == upsert_rocksdb::UPSERT_ROCKSDB_BOTTOMMOST_COMPRESSION_TYPE.name()
2266        || name == upsert_rocksdb::UPSERT_ROCKSDB_BATCH_SIZE.name()
2267        || name == upsert_rocksdb::UPSERT_ROCKSDB_STATS_LOG_INTERVAL_SECONDS.name()
2268        || name == upsert_rocksdb::UPSERT_ROCKSDB_STATS_PERSIST_INTERVAL_SECONDS.name()
2269        || name == upsert_rocksdb::UPSERT_ROCKSDB_POINT_LOOKUP_BLOCK_CACHE_SIZE_MB.name()
2270        || name == upsert_rocksdb::UPSERT_ROCKSDB_SHRINK_ALLOCATED_BUFFERS_BY_RATIO.name()
2271}
2272
2273/// Returns whether the named variable is a Postgres/CRDB timestamp oracle
2274/// configuration parameter.
2275pub fn is_pg_timestamp_oracle_config_var(name: &str) -> bool {
2276    name == PG_TIMESTAMP_ORACLE_CONNECTION_POOL_MAX_SIZE.name()
2277        || name == PG_TIMESTAMP_ORACLE_CONNECTION_POOL_MAX_WAIT.name()
2278        || name == PG_TIMESTAMP_ORACLE_CONNECTION_POOL_TTL.name()
2279        || name == PG_TIMESTAMP_ORACLE_CONNECTION_POOL_TTL_STAGGER.name()
2280        || name == CRDB_CONNECT_TIMEOUT.name()
2281        || name == CRDB_TCP_USER_TIMEOUT.name()
2282}
2283
2284/// Returns whether the named variable is a cluster scheduling config
2285pub fn is_cluster_scheduling_var(name: &str) -> bool {
2286    name == cluster_scheduling::CLUSTER_MULTI_PROCESS_REPLICA_AZ_AFFINITY_WEIGHT.name()
2287        || name == cluster_scheduling::CLUSTER_SOFTEN_REPLICATION_ANTI_AFFINITY.name()
2288        || name == cluster_scheduling::CLUSTER_SOFTEN_REPLICATION_ANTI_AFFINITY_WEIGHT.name()
2289        || name == cluster_scheduling::CLUSTER_ENABLE_TOPOLOGY_SPREAD.name()
2290        || name == cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_IGNORE_NON_SINGULAR_SCALE.name()
2291        || name == cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_MAX_SKEW.name()
2292        || name == cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_SOFT.name()
2293        || name == cluster_scheduling::CLUSTER_SOFTEN_AZ_AFFINITY.name()
2294        || name == cluster_scheduling::CLUSTER_SOFTEN_AZ_AFFINITY_WEIGHT.name()
2295        || name == cluster_scheduling::CLUSTER_ALWAYS_USE_DISK.name()
2296}
2297
2298/// Returns whether the named variable is an HTTP server related config var.
2299pub fn is_http_config_var(name: &str) -> bool {
2300    name == WEBHOOK_CONCURRENT_REQUEST_LIMIT.name()
2301}
2302
2303/// Set of [`SystemVar`]s that can also get set at a per-Session level.
2304///
2305/// TODO(parkmycar): Instead of a separate list, make this a field on VarDefinition.
2306static SESSION_SYSTEM_VARS: LazyLock<BTreeMap<&'static UncasedStr, &'static VarDefinition>> =
2307    LazyLock::new(|| {
2308        [
2309            &APPLICATION_NAME,
2310            &CLIENT_ENCODING,
2311            &CLIENT_MIN_MESSAGES,
2312            &CLUSTER,
2313            &CLUSTER_REPLICA,
2314            &DEFAULT_CLUSTER_REPLICATION_FACTOR,
2315            &CURRENT_OBJECT_MISSING_WARNINGS,
2316            &DATABASE,
2317            &DATE_STYLE,
2318            &EXTRA_FLOAT_DIGITS,
2319            &INTEGER_DATETIMES,
2320            &INTERVAL_STYLE,
2321            &REAL_TIME_RECENCY_TIMEOUT,
2322            &SEARCH_PATH,
2323            &STANDARD_CONFORMING_STRINGS,
2324            &STATEMENT_TIMEOUT,
2325            &IDLE_IN_TRANSACTION_SESSION_TIMEOUT,
2326            &TIMEZONE,
2327            &TRANSACTION_ISOLATION,
2328            &MAX_QUERY_RESULT_SIZE,
2329        ]
2330        .into_iter()
2331        .map(|var| (UncasedStr::new(var.name()), var))
2332        .collect()
2333    });
2334
2335// Provides a wrapper to express that a particular `ServerVar` is meant to be used as a feature
2336/// flag.
2337#[derive(Debug)]
2338pub struct FeatureFlag {
2339    pub flag: &'static VarDefinition,
2340    pub feature_desc: &'static str,
2341}
2342
2343impl FeatureFlag {
2344    /// Returns an error unless the feature flag is enabled in the provided
2345    /// `system_vars`.
2346    pub fn require(&'static self, system_vars: &SystemVars) -> Result<(), VarError> {
2347        match *system_vars.expect_value::<bool>(self.flag) {
2348            true => Ok(()),
2349            false => Err(VarError::RequiresFeatureFlag { feature_flag: self }),
2350        }
2351    }
2352}
2353
2354impl PartialEq for FeatureFlag {
2355    fn eq(&self, other: &FeatureFlag) -> bool {
2356        self.flag.name() == other.flag.name()
2357    }
2358}
2359
2360impl Eq for FeatureFlag {}
2361
2362impl Var for MzVersion {
2363    fn name(&self) -> &'static str {
2364        MZ_VERSION_NAME.as_str()
2365    }
2366
2367    fn value(&self) -> String {
2368        self.build_info
2369            .human_version(self.helm_chart_version.clone())
2370    }
2371
2372    fn description(&self) -> &'static str {
2373        "Shows the Materialize server version (Materialize)."
2374    }
2375
2376    fn type_name(&self) -> Cow<'static, str> {
2377        String::type_name()
2378    }
2379
2380    fn visible(&self, _: &User, _: &SystemVars) -> Result<(), VarError> {
2381        Ok(())
2382    }
2383}
2384
2385impl Var for User {
2386    fn name(&self) -> &'static str {
2387        IS_SUPERUSER_NAME.as_str()
2388    }
2389
2390    fn value(&self) -> String {
2391        self.is_superuser().format()
2392    }
2393
2394    fn description(&self) -> &'static str {
2395        "Reports whether the current session is a superuser (PostgreSQL)."
2396    }
2397
2398    fn type_name(&self) -> Cow<'static, str> {
2399        bool::type_name()
2400    }
2401
2402    fn visible(&self, _: &User, _: &SystemVars) -> Result<(), VarError> {
2403        Ok(())
2404    }
2405}