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