Skip to main content

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