1use std::borrow::Cow;
67use std::clone::Clone;
68use std::collections::BTreeMap;
69use std::fmt::Debug;
70use std::net::IpAddr;
71use std::num::NonZeroU32;
72use std::string::ToString;
73use std::sync::{Arc, LazyLock};
74use std::time::Duration;
75
76use chrono::{DateTime, Utc};
77use derivative::Derivative;
78use imbl::OrdMap;
79use mz_build_info::BuildInfo;
80use mz_dyncfg::{ConfigSet, ConfigType, ConfigUpdates, ConfigVal, ParameterScope};
81use mz_persist_client::cfg::{
82 CRDB_CONNECT_TIMEOUT, CRDB_KEEPALIVES_IDLE, CRDB_KEEPALIVES_INTERVAL, CRDB_KEEPALIVES_RETRIES,
83 CRDB_TCP_USER_TIMEOUT,
84};
85use mz_pgrepr::TextEncodeSettings;
86use mz_repr::adt::numeric::Numeric;
87use mz_repr::adt::timestamp::CheckedTimestamp;
88use mz_repr::bytes::ByteSize;
89use mz_repr::user::{ExternalUserMetadata, InternalUserMetadata};
90use mz_tracing::{CloneableEnvFilter, SerializableDirective};
91use serde::Serialize;
92use thiserror::Error;
93use uncased::UncasedStr;
94
95use crate::ast::Ident;
96use crate::session::user::User;
97
98pub(crate) mod constraints;
99pub(crate) mod definitions;
100pub(crate) mod errors;
101pub(crate) mod polyfill;
102pub(crate) mod value;
103
104pub use definitions::*;
105pub use errors::*;
106pub use value::*;
107
108#[derive(Debug, Clone, Copy, PartialEq, Eq)]
113pub enum EndTransactionAction {
114 Commit,
116 Rollback,
118}
119
120#[derive(Debug, Clone, Copy)]
126pub enum VarInput<'a> {
127 Flat(&'a str),
135 SqlSet(&'a [String]),
141}
142
143impl<'a> VarInput<'a> {
144 pub fn to_vec(&self) -> Vec<String> {
146 match self {
147 VarInput::Flat(v) => vec![v.to_string()],
148 VarInput::SqlSet(values) => values.into_iter().map(|v| v.to_string()).collect(),
149 }
150 }
151}
152
153#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize)]
155pub enum OwnedVarInput {
156 Flat(String),
163 SqlSet(Vec<String>),
167}
168
169impl OwnedVarInput {
170 pub fn borrow(&self) -> VarInput<'_> {
172 match self {
173 OwnedVarInput::Flat(v) => VarInput::Flat(v),
174 OwnedVarInput::SqlSet(v) => VarInput::SqlSet(v),
175 }
176 }
177}
178
179pub trait Var: Debug {
181 fn name(&self) -> &'static str;
183
184 fn value(&self) -> String;
190
191 fn description(&self) -> &'static str;
194
195 fn type_name(&self) -> Cow<'static, str>;
197
198 fn visible(&self, user: &User, system_vars: &SystemVars) -> Result<(), VarError>;
203
204 fn is_unsafe(&self) -> bool {
206 self.name().starts_with("unsafe_")
207 }
208
209 fn scope(&self) -> ParameterScope {
213 ParameterScope::Environment
214 }
215
216 fn as_var(&self) -> &dyn Var
219 where
220 Self: Sized,
221 {
222 self
223 }
224}
225
226#[derive(Debug)]
233pub struct SessionVar {
234 definition: VarDefinition,
235 default_value: Option<Box<dyn Value>>,
237 local_value: Option<Box<dyn Value>>,
239 staged_value: Option<Box<dyn Value>>,
241 session_value: Option<Box<dyn Value>>,
243}
244
245impl Clone for SessionVar {
246 fn clone(&self) -> Self {
247 SessionVar {
248 definition: self.definition.clone(),
249 default_value: self.default_value.as_ref().map(|v| v.box_clone()),
250 local_value: self.local_value.as_ref().map(|v| v.box_clone()),
251 staged_value: self.staged_value.as_ref().map(|v| v.box_clone()),
252 session_value: self.session_value.as_ref().map(|v| v.box_clone()),
253 }
254 }
255}
256
257impl SessionVar {
258 pub const fn new(var: VarDefinition) -> Self {
259 SessionVar {
260 definition: var,
261 default_value: None,
262 local_value: None,
263 staged_value: None,
264 session_value: None,
265 }
266 }
267
268 pub fn check(&self, input: VarInput) -> Result<String, VarError> {
271 let v = self.definition.parse(input)?;
272 self.validate_constraints(v.as_ref())?;
273
274 Ok(v.format())
275 }
276
277 pub fn set(&mut self, input: VarInput, local: bool) -> Result<(), VarError> {
279 let v = self.definition.parse(input)?;
280
281 self.validate_constraints(v.as_ref())?;
283
284 if local {
285 self.local_value = Some(v);
286 } else {
287 self.local_value = None;
288 self.staged_value = Some(v);
289 }
290 Ok(())
291 }
292
293 pub fn set_default(&mut self, input: VarInput) -> Result<(), VarError> {
295 let v = self.definition.parse(input)?;
296 self.validate_constraints(v.as_ref())?;
297 self.default_value = Some(v);
298 Ok(())
299 }
300
301 pub fn reset(&mut self, local: bool) {
303 let value = self
304 .default_value
305 .as_ref()
306 .map(|v| v.as_ref())
307 .unwrap_or_else(|| self.definition.value.value());
308 if local {
309 self.local_value = Some(value.box_clone());
310 } else {
311 self.local_value = None;
312 self.staged_value = Some(value.box_clone());
313 }
314 }
315
316 pub fn reset_durable(&mut self) {
325 self.local_value = None;
326 self.staged_value = None;
327 self.session_value = None;
328 }
329
330 #[must_use]
332 pub fn end_transaction(&self, action: EndTransactionAction) -> Option<Self> {
333 if !self.is_mutating() {
334 return None;
335 }
336 let mut next: Self = self.clone();
337 next.local_value = None;
338 match action {
339 EndTransactionAction::Commit if next.staged_value.is_some() => {
340 next.session_value = next.staged_value.take()
341 }
342 _ => next.staged_value = None,
343 }
344 Some(next)
345 }
346
347 pub fn is_mutating(&self) -> bool {
349 self.local_value.is_some() || self.staged_value.is_some()
350 }
351
352 pub fn value_dyn(&self) -> &dyn Value {
353 self.local_value
354 .as_deref()
355 .or(self.staged_value.as_deref())
356 .or(self.session_value.as_deref())
357 .or(self.default_value.as_deref())
358 .unwrap_or_else(|| self.definition.value.value())
359 }
360
361 pub fn inspect_session_value(&self) -> Option<&dyn Value> {
366 self.session_value.as_deref()
367 }
368
369 fn validate_constraints(&self, val: &dyn Value) -> Result<(), VarError> {
370 if let Some(constraint) = &self.definition.constraint {
371 constraint.check_constraint(self, self.value_dyn(), val)
372 } else {
373 Ok(())
374 }
375 }
376}
377
378impl Var for SessionVar {
379 fn name(&self) -> &'static str {
380 self.definition.name.as_str()
381 }
382
383 fn value(&self) -> String {
384 self.value_dyn().format()
385 }
386
387 fn description(&self) -> &'static str {
388 self.definition.description
389 }
390
391 fn type_name(&self) -> Cow<'static, str> {
392 self.definition.type_name()
393 }
394
395 fn visible(
396 &self,
397 user: &User,
398 system_vars: &super::vars::SystemVars,
399 ) -> Result<(), super::vars::VarError> {
400 self.definition.visible(user, system_vars)
401 }
402}
403
404#[derive(Debug, Clone, PartialEq, Eq)]
405pub struct MzVersion {
406 build_info: &'static BuildInfo,
408 helm_chart_version: Option<String>,
410}
411
412impl MzVersion {
413 pub fn new(build_info: &'static BuildInfo, helm_chart_version: Option<String>) -> Self {
414 MzVersion {
415 build_info,
416 helm_chart_version,
417 }
418 }
419}
420
421#[derive(Debug, Clone)]
426pub struct SessionVars {
427 vars: OrdMap<&'static UncasedStr, SessionVar>,
429 mz_version: MzVersion,
431 user: User,
433}
434
435impl SessionVars {
436 pub fn new_unchecked(
438 build_info: &'static BuildInfo,
439 user: User,
440 helm_chart_version: Option<String>,
441 ) -> SessionVars {
442 use definitions::*;
443
444 let vars = [
445 &FAILPOINTS,
446 &SERVER_VERSION,
447 &SERVER_VERSION_NUM,
448 &SQL_SAFE_UPDATES,
449 &REAL_TIME_RECENCY,
450 &EMIT_PLAN_INSIGHTS_NOTICE,
451 &EMIT_TIMESTAMP_NOTICE,
452 &EMIT_TRACE_ID_NOTICE,
453 &AUTO_ROUTE_CATALOG_QUERIES,
454 &ENABLE_SESSION_RBAC_CHECKS,
455 &RESTRICT_TO_USER_OBJECTS,
456 &ENABLE_SESSION_CARDINALITY_ESTIMATES,
457 &MAX_IDENTIFIER_LENGTH,
458 &STATEMENT_LOGGING_SAMPLE_RATE,
459 &EMIT_INTROSPECTION_QUERY_NOTICE,
460 &UNSAFE_NEW_TRANSACTION_WALL_TIME,
461 &WELCOME_MESSAGE,
462 ]
463 .into_iter()
464 .chain(SESSION_SYSTEM_VARS.iter().map(|(_name, var)| *var))
465 .map(|var| (var.name, SessionVar::new(var.clone())))
466 .collect();
467
468 SessionVars {
469 vars,
470 mz_version: MzVersion::new(build_info, helm_chart_version),
471 user,
472 }
473 }
474
475 fn expect_value<V: Value>(&self, var: &VarDefinition) -> &V {
476 let var = self
477 .vars
478 .get(var.name)
479 .expect("provided var should be in state");
480 let val = var.value_dyn();
481 val.as_any().downcast_ref::<V>().expect("success")
482 }
483
484 pub fn iter(&self) -> impl Iterator<Item = &dyn Var> {
491 #[allow(clippy::as_conversions)]
492 self.vars
493 .values()
494 .map(|v| v.as_var())
495 .chain([&self.mz_version as &dyn Var, &self.user])
496 }
497
498 pub fn notify_set(&self) -> impl Iterator<Item = &dyn Var> {
502 [
509 &APPLICATION_NAME,
510 &CLIENT_ENCODING,
511 &DATE_STYLE,
512 &INTEGER_DATETIMES,
513 &SERVER_VERSION,
514 &STANDARD_CONFORMING_STRINGS,
515 &TIMEZONE,
516 &INTERVAL_STYLE,
517 &CLUSTER,
522 &CLUSTER_REPLICA,
523 &DEFAULT_CLUSTER_REPLICATION_FACTOR,
524 &DATABASE,
525 &SEARCH_PATH,
526 ]
527 .into_iter()
528 .map(|v| self.vars[v.name].as_var())
529 .chain(std::iter::once(self.mz_version.as_var()))
536 }
537
538 pub fn reset_all(&mut self) {
545 let names: Vec<_> = self.vars.keys().copied().collect();
546 for name in names {
547 self.vars[name].reset_durable();
548 }
549 }
550
551 pub fn get(&self, system_vars: &SystemVars, name: &str) -> Result<&dyn Var, VarError> {
562 let name = compat_translate_name(name);
563
564 let name = UncasedStr::new(name);
565 if name == MZ_VERSION_NAME {
566 Ok(&self.mz_version)
567 } else if name == IS_SUPERUSER_NAME {
568 Ok(&self.user)
569 } else {
570 self.vars
571 .get(name)
572 .map(|v| {
573 v.visible(&self.user, system_vars)?;
574 Ok(v.as_var())
575 })
576 .transpose()?
577 .ok_or_else(|| VarError::UnknownParameter(name.to_string()))
578 }
579 }
580
581 pub fn inspect(&self, name: &str) -> Result<&SessionVar, VarError> {
586 let name = compat_translate_name(name);
587
588 self.vars
589 .get(UncasedStr::new(name))
590 .ok_or_else(|| VarError::UnknownParameter(name.to_string()))
591 }
592
593 pub fn set(
606 &mut self,
607 system_vars: &SystemVars,
608 name: &str,
609 input: VarInput,
610 local: bool,
611 ) -> Result<(), VarError> {
612 let (name, input) = compat_translate(name, input);
613
614 check_transaction_isolation_feature_flag(name, input, system_vars)?;
615
616 let name = UncasedStr::new(name);
617 self.check_read_only(name)?;
618
619 self.vars
620 .get_mut(name)
621 .map(|v| {
622 v.visible(&self.user, system_vars)?;
623 v.set(input, local)
624 })
625 .transpose()?
626 .ok_or_else(|| VarError::UnknownParameter(name.to_string()))
627 }
628
629 pub fn set_default(&mut self, name: &str, input: VarInput) -> Result<(), VarError> {
632 let (name, input) = compat_translate(name, input);
633
634 let name = UncasedStr::new(name);
635
636 if !Self::allow_role_default(name) {
640 self.check_read_only(name)?;
641 }
642
643 self.vars
644 .get_mut(name)
645 .map(|v| v.set_default(input))
647 .transpose()?
648 .ok_or_else(|| VarError::UnknownParameter(name.to_string()))
649 }
650
651 fn allow_role_default(name: &UncasedStr) -> bool {
659 name == RESTRICT_TO_USER_OBJECTS.name
660 }
661
662 pub fn reset(
676 &mut self,
677 system_vars: &SystemVars,
678 name: &str,
679 local: bool,
680 ) -> Result<(), VarError> {
681 let name = compat_translate_name(name);
682
683 let name = UncasedStr::new(name);
684 self.check_read_only(name)?;
685
686 self.vars
687 .get_mut(name)
688 .map(|v| {
689 v.visible(&self.user, system_vars)?;
690 v.reset(local);
691 Ok(())
692 })
693 .transpose()?
694 .ok_or_else(|| VarError::UnknownParameter(name.to_string()))
695 }
696
697 fn check_read_only(&self, name: &UncasedStr) -> Result<(), VarError> {
704 if name == MZ_VERSION_NAME {
705 Err(VarError::ReadOnlyParameter(MZ_VERSION_NAME.as_str()))
706 } else if name == IS_SUPERUSER_NAME {
707 Err(VarError::ReadOnlyParameter(IS_SUPERUSER_NAME.as_str()))
708 } else if name == MAX_IDENTIFIER_LENGTH.name {
709 Err(VarError::ReadOnlyParameter(
710 MAX_IDENTIFIER_LENGTH.name.as_str(),
711 ))
712 } else if name == RESTRICT_TO_USER_OBJECTS.name {
713 Err(VarError::ReadOnlyParameter(
717 RESTRICT_TO_USER_OBJECTS.name.as_str(),
718 ))
719 } else {
720 Ok(())
721 }
722 }
723
724 #[mz_ore::instrument(level = "debug")]
729 pub fn end_transaction(
730 &mut self,
731 action: EndTransactionAction,
732 ) -> BTreeMap<&'static str, String> {
733 let mut changed = BTreeMap::new();
734 let mut updates = Vec::new();
735 for (name, var) in self.vars.iter() {
736 if !var.is_mutating() {
737 continue;
738 }
739 let before = var.value();
740 let next = var.end_transaction(action).expect("must mutate");
741 let after = next.value();
742 updates.push((*name, next));
743
744 if before != after {
746 changed.insert(var.name(), after);
747 }
748 }
749 self.vars.extend(updates);
750 changed
751 }
752
753 pub fn application_name(&self) -> &str {
755 self.expect_value::<String>(&APPLICATION_NAME).as_str()
756 }
757
758 pub fn build_info(&self) -> &'static BuildInfo {
760 self.mz_version.build_info
761 }
762
763 pub fn client_encoding(&self) -> &ClientEncoding {
765 self.expect_value(&CLIENT_ENCODING)
766 }
767
768 pub fn client_min_messages(&self) -> &ClientSeverity {
770 self.expect_value(&CLIENT_MIN_MESSAGES)
771 }
772
773 pub fn cluster(&self) -> &str {
775 self.expect_value::<String>(&CLUSTER).as_str()
776 }
777
778 pub fn cluster_replica(&self) -> Option<&str> {
780 self.expect_value::<Option<String>>(&CLUSTER_REPLICA)
781 .as_deref()
782 }
783
784 pub fn current_object_missing_warnings(&self) -> bool {
787 *self.expect_value::<bool>(&CURRENT_OBJECT_MISSING_WARNINGS)
788 }
789
790 pub fn date_style(&self) -> &[&str] {
792 &self.expect_value::<DateStyle>(&DATE_STYLE).0
793 }
794
795 pub fn database(&self) -> &str {
797 self.expect_value::<String>(&DATABASE).as_str()
798 }
799
800 pub fn extra_float_digits(&self) -> i32 {
802 *self.expect_value(&EXTRA_FLOAT_DIGITS)
803 }
804
805 pub fn text_encode_settings(&self) -> TextEncodeSettings {
808 TextEncodeSettings {
809 extra_float_digits: self.extra_float_digits(),
810 }
811 }
812
813 pub fn integer_datetimes(&self) -> bool {
815 *self.expect_value(&INTEGER_DATETIMES)
816 }
817
818 pub fn intervalstyle(&self) -> &IntervalStyle {
820 self.expect_value(&INTERVAL_STYLE)
821 }
822
823 pub fn mz_version(&self) -> String {
825 self.mz_version.value()
826 }
827
828 pub fn search_path(&self) -> &[Ident] {
830 self.expect_value::<Vec<Ident>>(&SEARCH_PATH).as_slice()
831 }
832
833 pub fn server_version(&self) -> &str {
835 self.expect_value::<String>(&SERVER_VERSION).as_str()
836 }
837
838 pub fn server_version_num(&self) -> i32 {
840 *self.expect_value(&SERVER_VERSION_NUM)
841 }
842
843 pub fn sql_safe_updates(&self) -> bool {
845 *self.expect_value(&SQL_SAFE_UPDATES)
846 }
847
848 pub fn standard_conforming_strings(&self) -> bool {
851 *self.expect_value(&STANDARD_CONFORMING_STRINGS)
852 }
853
854 pub fn statement_timeout(&self) -> &Duration {
856 self.expect_value(&STATEMENT_TIMEOUT)
857 }
858
859 pub fn idle_in_transaction_session_timeout(&self) -> &Duration {
861 self.expect_value(&IDLE_IN_TRANSACTION_SESSION_TIMEOUT)
862 }
863
864 pub fn timezone(&self) -> &TimeZone {
866 self.expect_value(&TIMEZONE)
867 }
868
869 pub fn transaction_isolation(&self) -> &IsolationLevel {
872 self.expect_value(&TRANSACTION_ISOLATION)
873 }
874
875 pub fn real_time_recency(&self) -> bool {
877 *self.expect_value(&REAL_TIME_RECENCY)
878 }
879
880 pub fn real_time_recency_timeout(&self) -> &Duration {
882 self.expect_value(&REAL_TIME_RECENCY_TIMEOUT)
883 }
884
885 pub fn emit_plan_insights_notice(&self) -> bool {
887 *self.expect_value(&EMIT_PLAN_INSIGHTS_NOTICE)
888 }
889
890 pub fn emit_timestamp_notice(&self) -> bool {
892 *self.expect_value(&EMIT_TIMESTAMP_NOTICE)
893 }
894
895 pub fn emit_trace_id_notice(&self) -> bool {
897 *self.expect_value(&EMIT_TRACE_ID_NOTICE)
898 }
899
900 pub fn auto_route_catalog_queries(&self) -> bool {
902 *self.expect_value(&AUTO_ROUTE_CATALOG_QUERIES)
903 }
904
905 pub fn enable_session_rbac_checks(&self) -> bool {
907 *self.expect_value(&ENABLE_SESSION_RBAC_CHECKS)
908 }
909
910 pub fn restrict_to_user_objects(&self) -> bool {
912 *self.expect_value(&RESTRICT_TO_USER_OBJECTS)
913 }
914
915 pub fn enable_session_cardinality_estimates(&self) -> bool {
917 *self.expect_value(&ENABLE_SESSION_CARDINALITY_ESTIMATES)
918 }
919
920 pub fn is_superuser(&self) -> bool {
922 self.user.is_superuser()
923 }
924
925 pub fn user(&self) -> &User {
927 &self.user
928 }
929
930 pub fn max_query_result_size(&self) -> u64 {
932 self.expect_value::<ByteSize>(&MAX_QUERY_RESULT_SIZE)
933 .as_bytes()
934 }
935
936 pub fn set_internal_user_metadata(&mut self, metadata: InternalUserMetadata) {
938 self.user.internal_metadata = Some(metadata);
939 }
940
941 pub fn set_external_user_metadata(&mut self, metadata: ExternalUserMetadata) {
943 self.user.external_metadata = Some(metadata);
944 }
945
946 pub fn set_cluster(&mut self, cluster: String) {
947 let var = self
948 .vars
949 .get_mut(UncasedStr::new(CLUSTER.name()))
950 .expect("cluster variable must exist");
951 var.set(VarInput::Flat(&cluster), false)
952 .expect("setting cluster must succeed");
953 }
954
955 pub fn set_local_transaction_isolation(&mut self, transaction_isolation: IsolationLevel) {
956 let var = self
957 .vars
958 .get_mut(UncasedStr::new(TRANSACTION_ISOLATION.name()))
959 .expect("transaction_isolation variable must exist");
960 var.set(VarInput::Flat(&transaction_isolation.to_string()), true)
961 .expect("setting transaction isolation must succeed");
962 }
963
964 pub fn get_statement_logging_sample_rate(&self) -> Numeric {
965 *self.expect_value(&STATEMENT_LOGGING_SAMPLE_RATE)
966 }
967
968 pub fn emit_introspection_query_notice(&self) -> bool {
970 *self.expect_value(&EMIT_INTROSPECTION_QUERY_NOTICE)
971 }
972
973 pub fn unsafe_new_transaction_wall_time(&self) -> Option<CheckedTimestamp<DateTime<Utc>>> {
974 *self.expect_value(&UNSAFE_NEW_TRANSACTION_WALL_TIME)
975 }
976
977 pub fn welcome_message(&self) -> bool {
979 *self.expect_value(&WELCOME_MESSAGE)
980 }
981}
982
983pub const OLD_CATALOG_SERVER_CLUSTER: &str = "mz_introspection";
985pub const OLD_AUTO_ROUTE_CATALOG_QUERIES: &str = "auto_route_introspection_queries";
986
987fn compat_translate<'a, 'b>(name: &'a str, input: VarInput<'b>) -> (&'a str, VarInput<'b>) {
995 if name == CLUSTER.name() {
996 if let Ok(value) = CLUSTER.parse(input) {
997 if value.format() == OLD_CATALOG_SERVER_CLUSTER {
998 tracing::debug!(
999 github_27285 = true,
1000 "encountered deprecated `cluster` variable value: {}",
1001 OLD_CATALOG_SERVER_CLUSTER,
1002 );
1003 return (name, VarInput::Flat("mz_catalog_server"));
1004 }
1005 }
1006 }
1007
1008 if name == OLD_AUTO_ROUTE_CATALOG_QUERIES {
1009 tracing::debug!(
1010 github_27285 = true,
1011 "encountered deprecated `{}` variable name",
1012 OLD_AUTO_ROUTE_CATALOG_QUERIES,
1013 );
1014 return (AUTO_ROUTE_CATALOG_QUERIES.name(), input);
1015 }
1016
1017 (name, input)
1018}
1019
1020fn compat_translate_name(name: &str) -> &str {
1021 let (name, _) = compat_translate(name, VarInput::Flat(""));
1022 name
1023}
1024
1025pub fn check_transaction_isolation_feature_flag(
1035 name: &str,
1036 input: VarInput,
1037 system_vars: &SystemVars,
1038) -> Result<(), VarError> {
1039 if UncasedStr::new(name) != UncasedStr::new(TRANSACTION_ISOLATION_VAR_NAME) {
1040 return Ok(());
1041 }
1042 let Ok(level) = IsolationLevel::parse(input) else {
1044 return Ok(());
1045 };
1046 match level {
1047 IsolationLevel::StrongSessionSerializable => ENABLE_SESSION_TIMELINES.require(system_vars),
1048 IsolationLevel::BoundedStaleness(_) => {
1049 ENABLE_BOUNDED_STALENESS_ISOLATION.require(system_vars)
1050 }
1051 _ => Ok(()),
1052 }
1053}
1054
1055#[derive(Debug)]
1058pub struct SystemVar {
1059 definition: VarDefinition,
1060 persisted_value: Option<Box<dyn Value>>,
1062 dynamic_default: Option<Box<dyn Value>>,
1064}
1065
1066impl Clone for SystemVar {
1067 fn clone(&self) -> Self {
1068 SystemVar {
1069 definition: self.definition.clone(),
1070 persisted_value: self.persisted_value.as_ref().map(|v| v.box_clone()),
1071 dynamic_default: self.dynamic_default.as_ref().map(|v| v.box_clone()),
1072 }
1073 }
1074}
1075
1076impl SystemVar {
1077 pub fn new(definition: VarDefinition) -> Self {
1078 SystemVar {
1079 definition,
1080 persisted_value: None,
1081 dynamic_default: None,
1082 }
1083 }
1084
1085 fn is_default(&self, input: VarInput) -> Result<bool, VarError> {
1086 let v = self.definition.parse(input)?;
1087 Ok(self.definition.default_value() == v.as_ref())
1088 }
1089
1090 pub fn value_dyn(&self) -> &dyn Value {
1091 self.persisted_value
1092 .as_deref()
1093 .or(self.dynamic_default.as_deref())
1094 .unwrap_or_else(|| self.definition.default_value())
1095 }
1096
1097 pub fn value<V: 'static>(&self) -> &V {
1098 let val = self.value_dyn();
1099 val.as_any().downcast_ref::<V>().expect("success")
1100 }
1101
1102 fn parse(&self, input: VarInput) -> Result<Box<dyn Value>, VarError> {
1103 let v = self.definition.parse(input)?;
1104 self.validate_constraints(v.as_ref())?;
1106 Ok(v)
1107 }
1108
1109 fn set(&mut self, input: VarInput) -> Result<bool, VarError> {
1110 let v = self.parse(input)?;
1111
1112 if self.persisted_value.as_ref() != Some(&v) {
1113 self.persisted_value = Some(v);
1114 Ok(true)
1115 } else {
1116 Ok(false)
1117 }
1118 }
1119
1120 fn reset(&mut self) -> bool {
1121 if self.persisted_value.is_some() {
1122 self.persisted_value = None;
1123 true
1124 } else {
1125 false
1126 }
1127 }
1128
1129 fn set_default(&mut self, input: VarInput) -> Result<(), VarError> {
1130 let v = self.parse(input)?;
1131 self.dynamic_default = Some(v);
1132 Ok(())
1133 }
1134
1135 fn validate_constraints(&self, val: &dyn Value) -> Result<(), VarError> {
1136 if let Some(constraint) = &self.definition.constraint {
1137 constraint.check_constraint(self, self.value_dyn(), val)
1138 } else {
1139 Ok(())
1140 }
1141 }
1142}
1143
1144impl Var for SystemVar {
1145 fn name(&self) -> &'static str {
1146 self.definition.name.as_str()
1147 }
1148
1149 fn value(&self) -> String {
1150 self.value_dyn().format()
1151 }
1152
1153 fn description(&self) -> &'static str {
1154 self.definition.description
1155 }
1156
1157 fn type_name(&self) -> Cow<'static, str> {
1158 self.definition.type_name()
1159 }
1160
1161 fn scope(&self) -> ParameterScope {
1162 self.definition.scope()
1163 }
1164
1165 fn visible(&self, user: &User, system_vars: &SystemVars) -> Result<(), VarError> {
1166 self.definition.visible(user, system_vars)
1167 }
1168}
1169
1170#[derive(Debug, Error)]
1171pub enum NetworkPolicyError {
1172 #[error("Access denied for address {0}")]
1173 AddressDenied(IpAddr),
1174}
1175
1176#[derive(Derivative, Clone)]
1181#[derivative(Debug)]
1182pub struct SystemVars {
1183 allow_unsafe: bool,
1185 vars: BTreeMap<&'static UncasedStr, SystemVar>,
1187 #[derivative(Debug = "ignore")]
1189 callbacks: BTreeMap<String, Vec<Arc<dyn Fn(&SystemVars) + Send + Sync>>>,
1190
1191 dyncfgs: ConfigSet,
1195}
1196
1197impl Default for SystemVars {
1198 fn default() -> Self {
1199 Self::new()
1200 }
1201}
1202
1203impl SystemVars {
1204 pub fn new() -> Self {
1205 let system_vars = vec![
1206 &MAX_KAFKA_CONNECTIONS,
1207 &MAX_POSTGRES_CONNECTIONS,
1208 &MAX_MYSQL_CONNECTIONS,
1209 &MAX_SQL_SERVER_CONNECTIONS,
1210 &MAX_AWS_PRIVATELINK_CONNECTIONS,
1211 &MAX_TABLES,
1212 &MAX_SOURCES,
1213 &MAX_SINKS,
1214 &MAX_MATERIALIZED_VIEWS,
1215 &MAX_CLUSTERS,
1216 &MAX_REPLICAS_PER_CLUSTER,
1217 &MAX_CREDIT_CONSUMPTION_RATE,
1218 &MAX_DATABASES,
1219 &MAX_SCHEMAS_PER_DATABASE,
1220 &MAX_OBJECTS_PER_SCHEMA,
1221 &MAX_SECRETS,
1222 &MAX_ROLES,
1223 &MAX_NETWORK_POLICIES,
1224 &MAX_RULES_PER_NETWORK_POLICY,
1225 &MAX_RESULT_SIZE,
1226 &MAX_COPY_FROM_ROW_SIZE,
1227 &ALLOWED_CLUSTER_REPLICA_SIZES,
1228 &MAX_CONCURRENT_OCC_WRITES,
1229 &MAX_OCC_RETRIES,
1230 &upsert_rocksdb::UPSERT_ROCKSDB_COMPACTION_STYLE,
1231 &upsert_rocksdb::UPSERT_ROCKSDB_OPTIMIZE_COMPACTION_MEMTABLE_BUDGET,
1232 &upsert_rocksdb::UPSERT_ROCKSDB_LEVEL_COMPACTION_DYNAMIC_LEVEL_BYTES,
1233 &upsert_rocksdb::UPSERT_ROCKSDB_UNIVERSAL_COMPACTION_RATIO,
1234 &upsert_rocksdb::UPSERT_ROCKSDB_PARALLELISM,
1235 &upsert_rocksdb::UPSERT_ROCKSDB_COMPRESSION_TYPE,
1236 &upsert_rocksdb::UPSERT_ROCKSDB_BOTTOMMOST_COMPRESSION_TYPE,
1237 &upsert_rocksdb::UPSERT_ROCKSDB_BATCH_SIZE,
1238 &upsert_rocksdb::UPSERT_ROCKSDB_RETRY_DURATION,
1239 &upsert_rocksdb::UPSERT_ROCKSDB_STATS_LOG_INTERVAL_SECONDS,
1240 &upsert_rocksdb::UPSERT_ROCKSDB_STATS_PERSIST_INTERVAL_SECONDS,
1241 &upsert_rocksdb::UPSERT_ROCKSDB_POINT_LOOKUP_BLOCK_CACHE_SIZE_MB,
1242 &upsert_rocksdb::UPSERT_ROCKSDB_SHRINK_ALLOCATED_BUFFERS_BY_RATIO,
1243 &upsert_rocksdb::UPSERT_ROCKSDB_WRITE_BUFFER_MANAGER_CLUSTER_MEMORY_FRACTION,
1244 &upsert_rocksdb::UPSERT_ROCKSDB_WRITE_BUFFER_MANAGER_MEMORY_BYTES,
1245 &upsert_rocksdb::UPSERT_ROCKSDB_WRITE_BUFFER_MANAGER_ALLOW_STALL,
1246 &STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES,
1247 &STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES_TO_CLUSTER_SIZE_FRACTION,
1248 &STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES_DISK_ONLY,
1249 &STORAGE_STATISTICS_INTERVAL,
1250 &STORAGE_STATISTICS_COLLECTION_INTERVAL,
1251 &STORAGE_SHRINK_UPSERT_UNUSED_BUFFERS_BY_RATIO,
1252 &STORAGE_RECORD_SOURCE_SINK_NAMESPACED_ERRORS,
1253 &PERSIST_FAST_PATH_LIMIT,
1254 &METRICS_RETENTION,
1255 &UNSAFE_MOCK_AUDIT_EVENT_TIMESTAMP,
1256 &ENABLE_RBAC_CHECKS,
1257 &PG_SOURCE_CONNECT_TIMEOUT,
1258 &PG_SOURCE_TCP_KEEPALIVES_IDLE,
1259 &PG_SOURCE_TCP_KEEPALIVES_INTERVAL,
1260 &PG_SOURCE_TCP_KEEPALIVES_RETRIES,
1261 &PG_SOURCE_TCP_USER_TIMEOUT,
1262 &PG_SOURCE_TCP_CONFIGURE_SERVER,
1263 &PG_SOURCE_SNAPSHOT_STATEMENT_TIMEOUT,
1264 &PG_SOURCE_WAL_SENDER_TIMEOUT,
1265 &PG_SOURCE_SNAPSHOT_COLLECT_STRICT_COUNT,
1266 &MYSQL_SOURCE_TCP_KEEPALIVE,
1267 &MYSQL_SOURCE_SNAPSHOT_MAX_EXECUTION_TIME,
1268 &MYSQL_SOURCE_SNAPSHOT_LOCK_WAIT_TIMEOUT,
1269 &MYSQL_SOURCE_SNAPSHOT_WAIT_TIMEOUT,
1270 &MYSQL_SOURCE_CONNECT_TIMEOUT,
1271 &SSH_CHECK_INTERVAL,
1272 &SSH_CONNECT_TIMEOUT,
1273 &SSH_KEEPALIVES_IDLE,
1274 &KAFKA_SOCKET_KEEPALIVE,
1275 &KAFKA_SOCKET_TIMEOUT,
1276 &KAFKA_TRANSACTION_TIMEOUT,
1277 &KAFKA_SOCKET_CONNECTION_SETUP_TIMEOUT,
1278 &KAFKA_FETCH_METADATA_TIMEOUT,
1279 &KAFKA_PROGRESS_RECORD_FETCH_TIMEOUT,
1280 &ENABLE_LAUNCHDARKLY,
1281 &MAX_CONNECTIONS,
1282 &NETWORK_POLICY,
1283 &SUPERUSER_RESERVED_CONNECTIONS,
1284 &KEEP_N_SOURCE_STATUS_HISTORY_ENTRIES,
1285 &KEEP_N_SINK_STATUS_HISTORY_ENTRIES,
1286 &KEEP_N_PRIVATELINK_STATUS_HISTORY_ENTRIES,
1287 &REPLICA_STATUS_HISTORY_RETENTION_WINDOW,
1288 &ENABLE_STORAGE_SHARD_FINALIZATION,
1289 &ENABLE_DEFAULT_CONNECTION_VALIDATION,
1290 &DEFAULT_TIMESTAMP_INTERVAL,
1291 &MIN_TIMESTAMP_INTERVAL,
1292 &MAX_TIMESTAMP_INTERVAL,
1293 &LOGGING_FILTER,
1294 &OPENTELEMETRY_FILTER,
1295 &LOGGING_FILTER_DEFAULTS,
1296 &OPENTELEMETRY_FILTER_DEFAULTS,
1297 &SENTRY_FILTERS,
1298 &WEBHOOKS_SECRETS_CACHING_TTL_SECS,
1299 &COORD_SLOW_MESSAGE_WARN_THRESHOLD,
1300 &grpc_client::CONNECT_TIMEOUT,
1301 &grpc_client::HTTP2_KEEP_ALIVE_INTERVAL,
1302 &grpc_client::HTTP2_KEEP_ALIVE_TIMEOUT,
1303 &cluster_scheduling::CLUSTER_MULTI_PROCESS_REPLICA_AZ_AFFINITY_WEIGHT,
1304 &cluster_scheduling::CLUSTER_SOFTEN_REPLICATION_ANTI_AFFINITY,
1305 &cluster_scheduling::CLUSTER_SOFTEN_REPLICATION_ANTI_AFFINITY_WEIGHT,
1306 &cluster_scheduling::CLUSTER_ENABLE_TOPOLOGY_SPREAD,
1307 &cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_IGNORE_NON_SINGULAR_SCALE,
1308 &cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_MAX_SKEW,
1309 &cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_MIN_DOMAINS,
1310 &cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_SOFT,
1311 &cluster_scheduling::CLUSTER_SOFTEN_AZ_AFFINITY,
1312 &cluster_scheduling::CLUSTER_SOFTEN_AZ_AFFINITY_WEIGHT,
1313 &cluster_scheduling::CLUSTER_ALTER_CHECK_READY_INTERVAL,
1314 &cluster_scheduling::CLUSTER_SECURITY_CONTEXT_ENABLED,
1315 &cluster_scheduling::CLUSTER_REFRESH_MV_COMPACTION_ESTIMATE,
1316 &grpc_client::HTTP2_KEEP_ALIVE_TIMEOUT,
1317 &STATEMENT_LOGGING_MAX_SAMPLE_RATE,
1318 &STATEMENT_LOGGING_DEFAULT_SAMPLE_RATE,
1319 &STATEMENT_LOGGING_TARGET_DATA_RATE,
1320 &STATEMENT_LOGGING_MAX_DATA_CREDIT,
1321 &ENABLE_INTERNAL_STATEMENT_LOGGING,
1322 &ENABLE_STATEMENT_ARRIVAL_LOGGING,
1323 &ENABLE_EXTENDED_PROTOCOL_IMPLICIT_TRANSACTION,
1324 &OPTIMIZER_STATS_TIMEOUT,
1325 &OPTIMIZER_ONESHOT_STATS_TIMEOUT,
1326 &PRIVATELINK_STATUS_UPDATE_QUOTA_PER_MINUTE,
1327 &WEBHOOK_CONCURRENT_REQUEST_LIMIT,
1328 &PG_TIMESTAMP_ORACLE_CONNECTION_POOL_MAX_SIZE,
1329 &PG_TIMESTAMP_ORACLE_CONNECTION_POOL_MAX_WAIT,
1330 &PG_TIMESTAMP_ORACLE_CONNECTION_POOL_TTL,
1331 &PG_TIMESTAMP_ORACLE_CONNECTION_POOL_TTL_STAGGER,
1332 &USER_STORAGE_MANAGED_COLLECTIONS_BATCH_DURATION,
1333 &FORCE_SOURCE_TABLE_SYNTAX,
1334 &OPTIMIZER_E2E_LATENCY_WARNING_THRESHOLD,
1335 &SCRAM_ITERATIONS,
1336 ];
1337
1338 let dyncfgs = mz_dyncfgs::all_dyncfgs();
1339 let dyncfg_vars: Vec<_> = dyncfgs
1340 .entries()
1341 .map(|cfg| {
1342 let var = match cfg.default() {
1343 ConfigVal::Bool(default) => {
1344 VarDefinition::new_runtime(cfg.name(), *default, cfg.desc(), false)
1345 }
1346 ConfigVal::U32(default) => {
1347 VarDefinition::new_runtime(cfg.name(), *default, cfg.desc(), false)
1348 }
1349 ConfigVal::Usize(default) => {
1350 VarDefinition::new_runtime(cfg.name(), *default, cfg.desc(), false)
1351 }
1352 ConfigVal::OptUsize(default) => {
1353 VarDefinition::new_runtime(cfg.name(), *default, cfg.desc(), false)
1354 }
1355 ConfigVal::F64(default) => {
1356 VarDefinition::new_runtime(cfg.name(), *default, cfg.desc(), false)
1357 }
1358 ConfigVal::String(default) => {
1359 VarDefinition::new_runtime(cfg.name(), default.clone(), cfg.desc(), false)
1360 }
1361 ConfigVal::OptString(default) => {
1362 VarDefinition::new_runtime(cfg.name(), default.clone(), cfg.desc(), false)
1363 }
1364 ConfigVal::Duration(default) => {
1365 VarDefinition::new_runtime(cfg.name(), default.clone(), cfg.desc(), false)
1366 }
1367 ConfigVal::Json(default) => {
1368 VarDefinition::new_runtime(cfg.name(), default.clone(), cfg.desc(), false)
1369 }
1370 };
1371 var.scoped(cfg.scope())
1374 })
1375 .collect();
1376
1377 let vars: BTreeMap<_, _> = system_vars
1378 .into_iter()
1379 .chain(definitions::FEATURE_FLAGS.iter().copied())
1381 .chain(SESSION_SYSTEM_VARS.values().copied())
1383 .cloned()
1384 .chain(dyncfg_vars)
1386 .map(|var| (var.name, SystemVar::new(var)))
1387 .collect();
1388
1389 let vars = SystemVars {
1390 vars,
1391 callbacks: BTreeMap::new(),
1392 allow_unsafe: false,
1393 dyncfgs,
1394 };
1395
1396 vars
1397 }
1398
1399 pub fn dyncfgs(&self) -> &ConfigSet {
1400 &self.dyncfgs
1401 }
1402
1403 pub fn set_unsafe(mut self, allow_unsafe: bool) -> Self {
1404 self.allow_unsafe = allow_unsafe;
1405 self
1406 }
1407
1408 pub fn allow_unsafe(&self) -> bool {
1409 self.allow_unsafe
1410 }
1411
1412 fn expect_value<V: 'static>(&self, var: &VarDefinition) -> &V {
1413 let val = self
1414 .vars
1415 .get(var.name)
1416 .expect("provided var should be in state");
1417
1418 val.value_dyn()
1419 .as_any()
1420 .downcast_ref::<V>()
1421 .expect("provided var type should matched stored var")
1422 }
1423
1424 fn expect_config_value<V: ConfigType + 'static>(&self, name: &UncasedStr) -> &V {
1425 let val = self
1426 .vars
1427 .get(name)
1428 .unwrap_or_else(|| panic!("provided var {name} should be in state"));
1429
1430 val.value_dyn()
1431 .as_any()
1432 .downcast_ref()
1433 .expect("provided var type should matched stored var")
1434 }
1435
1436 pub fn iter(&self) -> impl Iterator<Item = &dyn Var> {
1439 self.vars
1440 .values()
1441 .map(|v| v.as_var())
1442 .filter(|v| !SESSION_SYSTEM_VARS.contains_key(UncasedStr::new(v.name())))
1443 }
1444
1445 pub fn iter_synced(&self) -> impl Iterator<Item = &dyn Var> {
1449 self.iter().filter(|v| v.name() != ENABLE_LAUNCHDARKLY.name)
1450 }
1451
1452 pub fn iter_session(&self) -> impl Iterator<Item = &dyn Var> {
1454 self.vars
1455 .values()
1456 .map(|v| v.as_var())
1457 .filter(|v| SESSION_SYSTEM_VARS.contains_key(UncasedStr::new(v.name())))
1458 }
1459
1460 pub fn user_modifiable(&self, name: &str) -> bool {
1462 SESSION_SYSTEM_VARS.contains_key(UncasedStr::new(name))
1463 || name == ENABLE_RBAC_CHECKS.name()
1464 || name == NETWORK_POLICY.name()
1465 }
1466
1467 pub fn get(&self, name: &str) -> Result<&dyn Var, VarError> {
1488 self.vars
1489 .get(UncasedStr::new(name))
1490 .map(|v| v.as_var())
1491 .ok_or_else(|| VarError::UnknownParameter(name.into()))
1492 }
1493
1494 pub fn is_default(&self, name: &str, input: VarInput) -> Result<bool, VarError> {
1508 self.vars
1509 .get(UncasedStr::new(name))
1510 .ok_or_else(|| VarError::UnknownParameter(name.into()))
1511 .and_then(|v| v.is_default(input))
1512 }
1513
1514 pub fn set(&mut self, name: &str, input: VarInput) -> Result<bool, VarError> {
1537 let result = self
1538 .vars
1539 .get_mut(UncasedStr::new(name))
1540 .ok_or_else(|| VarError::UnknownParameter(name.into()))
1541 .and_then(|v| v.set(input))?;
1542 Ok(result)
1543 }
1544
1545 pub fn parse(&self, name: &str, input: VarInput) -> Result<Box<dyn Value>, VarError> {
1566 self.vars
1567 .get(UncasedStr::new(name))
1568 .ok_or_else(|| VarError::UnknownParameter(name.into()))
1569 .and_then(|v| v.parse(input))
1570 }
1571
1572 pub fn set_default(&mut self, name: &str, input: VarInput) -> Result<(), VarError> {
1580 self.vars
1581 .get_mut(UncasedStr::new(name))
1582 .ok_or_else(|| VarError::UnknownParameter(name.into()))
1583 .and_then(|v| v.set_default(input))?;
1584 Ok(())
1585 }
1586
1587 pub fn reset(&mut self, name: &str) -> Result<bool, VarError> {
1605 let result = self
1606 .vars
1607 .get_mut(UncasedStr::new(name))
1608 .ok_or_else(|| VarError::UnknownParameter(name.into()))
1609 .map(|v| v.reset())?;
1610 Ok(result)
1611 }
1612
1613 pub fn defaults(&self) -> BTreeMap<String, String> {
1615 self.vars
1616 .iter()
1617 .map(|(name, var)| {
1618 let default = var
1619 .dynamic_default
1620 .as_deref()
1621 .unwrap_or_else(|| var.definition.default_value());
1622 (name.as_str().to_owned(), default.format())
1623 })
1624 .collect()
1625 }
1626
1627 pub fn register_callback(
1646 &mut self,
1647 var: &VarDefinition,
1648 callback: Arc<dyn Fn(&SystemVars) + Send + Sync>,
1649 ) {
1650 self.callbacks
1651 .entry(var.name().to_string())
1652 .or_default()
1653 .push(callback);
1654 self.notify_callbacks(var.name());
1655 }
1656
1657 pub fn notify_all_callbacks(&self) {
1663 for callbacks in self.callbacks.values() {
1664 for callback in callbacks {
1665 (callback)(self);
1666 }
1667 }
1668 }
1669
1670 fn notify_callbacks(&self, name: &str) {
1672 if let Some(callbacks) = self.callbacks.get(name) {
1674 for callback in callbacks {
1675 (callback)(self);
1676 }
1677 }
1678 }
1679
1680 pub fn default_cluster(&self) -> String {
1683 self.expect_value::<String>(&CLUSTER).to_owned()
1684 }
1685
1686 pub fn max_kafka_connections(&self) -> u32 {
1688 *self.expect_value(&MAX_KAFKA_CONNECTIONS)
1689 }
1690
1691 pub fn max_postgres_connections(&self) -> u32 {
1693 *self.expect_value(&MAX_POSTGRES_CONNECTIONS)
1694 }
1695
1696 pub fn max_mysql_connections(&self) -> u32 {
1698 *self.expect_value(&MAX_MYSQL_CONNECTIONS)
1699 }
1700
1701 pub fn max_sql_server_connections(&self) -> u32 {
1703 *self.expect_value(&MAX_SQL_SERVER_CONNECTIONS)
1704 }
1705
1706 pub fn max_aws_privatelink_connections(&self) -> u32 {
1708 *self.expect_value(&MAX_AWS_PRIVATELINK_CONNECTIONS)
1709 }
1710
1711 pub fn max_tables(&self) -> u32 {
1713 *self.expect_value(&MAX_TABLES)
1714 }
1715
1716 pub fn max_sources(&self) -> u32 {
1718 *self.expect_value(&MAX_SOURCES)
1719 }
1720
1721 pub fn max_sinks(&self) -> u32 {
1723 *self.expect_value(&MAX_SINKS)
1724 }
1725
1726 pub fn max_materialized_views(&self) -> u32 {
1728 *self.expect_value(&MAX_MATERIALIZED_VIEWS)
1729 }
1730
1731 pub fn max_clusters(&self) -> u32 {
1733 *self.expect_value(&MAX_CLUSTERS)
1734 }
1735
1736 pub fn max_replicas_per_cluster(&self) -> u32 {
1738 *self.expect_value(&MAX_REPLICAS_PER_CLUSTER)
1739 }
1740
1741 pub fn max_credit_consumption_rate(&self) -> Numeric {
1743 *self.expect_value(&MAX_CREDIT_CONSUMPTION_RATE)
1744 }
1745
1746 pub fn max_databases(&self) -> u32 {
1748 *self.expect_value(&MAX_DATABASES)
1749 }
1750
1751 pub fn max_schemas_per_database(&self) -> u32 {
1753 *self.expect_value(&MAX_SCHEMAS_PER_DATABASE)
1754 }
1755
1756 pub fn max_objects_per_schema(&self) -> u32 {
1758 *self.expect_value(&MAX_OBJECTS_PER_SCHEMA)
1759 }
1760
1761 pub fn max_secrets(&self) -> u32 {
1763 *self.expect_value(&MAX_SECRETS)
1764 }
1765
1766 pub fn max_roles(&self) -> u32 {
1768 *self.expect_value(&MAX_ROLES)
1769 }
1770
1771 pub fn max_network_policies(&self) -> u32 {
1773 *self.expect_value(&MAX_NETWORK_POLICIES)
1774 }
1775
1776 pub fn max_rules_per_network_policy(&self) -> u32 {
1778 *self.expect_value(&MAX_RULES_PER_NETWORK_POLICY)
1779 }
1780
1781 pub fn max_result_size(&self) -> u64 {
1783 self.expect_value::<ByteSize>(&MAX_RESULT_SIZE).as_bytes()
1784 }
1785
1786 pub fn max_copy_from_row_size(&self) -> u64 {
1788 self.expect_value::<ByteSize>(&MAX_COPY_FROM_ROW_SIZE)
1789 .as_bytes()
1790 }
1791
1792 pub fn allowed_cluster_replica_sizes(&self) -> Vec<String> {
1794 self.expect_value::<Vec<Ident>>(&ALLOWED_CLUSTER_REPLICA_SIZES)
1795 .into_iter()
1796 .map(|s| s.as_str().into())
1797 .collect()
1798 }
1799
1800 pub fn max_concurrent_occ_writes(&self) -> u32 {
1802 *self.expect_value(&MAX_CONCURRENT_OCC_WRITES)
1803 }
1804
1805 pub fn max_occ_retries(&self) -> u32 {
1807 *self.expect_value(&MAX_OCC_RETRIES)
1808 }
1809
1810 pub fn default_cluster_replication_factor(&self) -> u32 {
1812 *self.expect_value::<u32>(&DEFAULT_CLUSTER_REPLICATION_FACTOR)
1813 }
1814
1815 pub fn upsert_rocksdb_compaction_style(&self) -> mz_rocksdb_types::config::CompactionStyle {
1816 *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_COMPACTION_STYLE)
1817 }
1818
1819 pub fn upsert_rocksdb_optimize_compaction_memtable_budget(&self) -> usize {
1820 *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_OPTIMIZE_COMPACTION_MEMTABLE_BUDGET)
1821 }
1822
1823 pub fn upsert_rocksdb_level_compaction_dynamic_level_bytes(&self) -> bool {
1824 *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_LEVEL_COMPACTION_DYNAMIC_LEVEL_BYTES)
1825 }
1826
1827 pub fn upsert_rocksdb_universal_compaction_ratio(&self) -> i32 {
1828 *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_UNIVERSAL_COMPACTION_RATIO)
1829 }
1830
1831 pub fn upsert_rocksdb_parallelism(&self) -> Option<i32> {
1832 *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_PARALLELISM)
1833 }
1834
1835 pub fn upsert_rocksdb_compression_type(&self) -> mz_rocksdb_types::config::CompressionType {
1836 *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_COMPRESSION_TYPE)
1837 }
1838
1839 pub fn upsert_rocksdb_bottommost_compression_type(
1840 &self,
1841 ) -> mz_rocksdb_types::config::CompressionType {
1842 *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_BOTTOMMOST_COMPRESSION_TYPE)
1843 }
1844
1845 pub fn upsert_rocksdb_batch_size(&self) -> usize {
1846 *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_BATCH_SIZE)
1847 }
1848
1849 pub fn upsert_rocksdb_retry_duration(&self) -> Duration {
1850 *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_RETRY_DURATION)
1851 }
1852
1853 pub fn upsert_rocksdb_stats_log_interval_seconds(&self) -> u32 {
1854 *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_STATS_LOG_INTERVAL_SECONDS)
1855 }
1856
1857 pub fn upsert_rocksdb_stats_persist_interval_seconds(&self) -> u32 {
1858 *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_STATS_PERSIST_INTERVAL_SECONDS)
1859 }
1860
1861 pub fn upsert_rocksdb_point_lookup_block_cache_size_mb(&self) -> Option<u32> {
1862 *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_POINT_LOOKUP_BLOCK_CACHE_SIZE_MB)
1863 }
1864
1865 pub fn upsert_rocksdb_shrink_allocated_buffers_by_ratio(&self) -> usize {
1866 *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_SHRINK_ALLOCATED_BUFFERS_BY_RATIO)
1867 }
1868
1869 pub fn upsert_rocksdb_write_buffer_manager_cluster_memory_fraction(&self) -> Option<Numeric> {
1870 *self.expect_value(
1871 &upsert_rocksdb::UPSERT_ROCKSDB_WRITE_BUFFER_MANAGER_CLUSTER_MEMORY_FRACTION,
1872 )
1873 }
1874
1875 pub fn upsert_rocksdb_write_buffer_manager_memory_bytes(&self) -> Option<usize> {
1876 *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_WRITE_BUFFER_MANAGER_MEMORY_BYTES)
1877 }
1878
1879 pub fn upsert_rocksdb_write_buffer_manager_allow_stall(&self) -> bool {
1880 *self.expect_value(&upsert_rocksdb::UPSERT_ROCKSDB_WRITE_BUFFER_MANAGER_ALLOW_STALL)
1881 }
1882
1883 pub fn persist_fast_path_limit(&self) -> usize {
1884 *self.expect_value(&PERSIST_FAST_PATH_LIMIT)
1885 }
1886
1887 pub fn pg_source_connect_timeout(&self) -> Duration {
1889 *self.expect_value(&PG_SOURCE_CONNECT_TIMEOUT)
1890 }
1891
1892 pub fn pg_source_tcp_keepalives_retries(&self) -> u32 {
1894 *self.expect_value(&PG_SOURCE_TCP_KEEPALIVES_RETRIES)
1895 }
1896
1897 pub fn pg_source_tcp_keepalives_idle(&self) -> Duration {
1899 *self.expect_value(&PG_SOURCE_TCP_KEEPALIVES_IDLE)
1900 }
1901
1902 pub fn pg_source_tcp_keepalives_interval(&self) -> Duration {
1904 *self.expect_value(&PG_SOURCE_TCP_KEEPALIVES_INTERVAL)
1905 }
1906
1907 pub fn pg_source_tcp_user_timeout(&self) -> Duration {
1909 *self.expect_value(&PG_SOURCE_TCP_USER_TIMEOUT)
1910 }
1911
1912 pub fn pg_source_tcp_configure_server(&self) -> bool {
1914 *self.expect_value(&PG_SOURCE_TCP_CONFIGURE_SERVER)
1915 }
1916
1917 pub fn pg_source_snapshot_statement_timeout(&self) -> Duration {
1919 *self.expect_value(&PG_SOURCE_SNAPSHOT_STATEMENT_TIMEOUT)
1920 }
1921
1922 pub fn pg_source_wal_sender_timeout(&self) -> Option<Duration> {
1924 *self.expect_value(&PG_SOURCE_WAL_SENDER_TIMEOUT)
1925 }
1926
1927 pub fn pg_source_snapshot_collect_strict_count(&self) -> bool {
1929 *self.expect_value(&PG_SOURCE_SNAPSHOT_COLLECT_STRICT_COUNT)
1930 }
1931
1932 pub fn mysql_source_tcp_keepalive(&self) -> Duration {
1934 *self.expect_value(&MYSQL_SOURCE_TCP_KEEPALIVE)
1935 }
1936
1937 pub fn mysql_source_snapshot_max_execution_time(&self) -> Duration {
1939 *self.expect_value(&MYSQL_SOURCE_SNAPSHOT_MAX_EXECUTION_TIME)
1940 }
1941
1942 pub fn mysql_source_snapshot_lock_wait_timeout(&self) -> Duration {
1944 *self.expect_value(&MYSQL_SOURCE_SNAPSHOT_LOCK_WAIT_TIMEOUT)
1945 }
1946
1947 pub fn mysql_source_snapshot_wait_timeout(&self) -> Duration {
1949 *self.expect_value(&MYSQL_SOURCE_SNAPSHOT_WAIT_TIMEOUT)
1950 }
1951
1952 pub fn mysql_source_connect_timeout(&self) -> Duration {
1954 *self.expect_value(&MYSQL_SOURCE_CONNECT_TIMEOUT)
1955 }
1956
1957 pub fn ssh_check_interval(&self) -> Duration {
1959 *self.expect_value(&SSH_CHECK_INTERVAL)
1960 }
1961
1962 pub fn ssh_connect_timeout(&self) -> Duration {
1964 *self.expect_value(&SSH_CONNECT_TIMEOUT)
1965 }
1966
1967 pub fn ssh_keepalives_idle(&self) -> Duration {
1969 *self.expect_value(&SSH_KEEPALIVES_IDLE)
1970 }
1971
1972 pub fn kafka_socket_keepalive(&self) -> bool {
1974 *self.expect_value(&KAFKA_SOCKET_KEEPALIVE)
1975 }
1976
1977 pub fn kafka_socket_timeout(&self) -> Option<Duration> {
1979 *self.expect_value(&KAFKA_SOCKET_TIMEOUT)
1980 }
1981
1982 pub fn kafka_transaction_timeout(&self) -> Duration {
1984 *self.expect_value(&KAFKA_TRANSACTION_TIMEOUT)
1985 }
1986
1987 pub fn kafka_socket_connection_setup_timeout(&self) -> Duration {
1989 *self.expect_value(&KAFKA_SOCKET_CONNECTION_SETUP_TIMEOUT)
1990 }
1991
1992 pub fn kafka_fetch_metadata_timeout(&self) -> Duration {
1994 *self.expect_value(&KAFKA_FETCH_METADATA_TIMEOUT)
1995 }
1996
1997 pub fn kafka_progress_record_fetch_timeout(&self) -> Option<Duration> {
1999 *self.expect_value(&KAFKA_PROGRESS_RECORD_FETCH_TIMEOUT)
2000 }
2001
2002 pub fn crdb_connect_timeout(&self) -> Duration {
2004 *self.expect_config_value(UncasedStr::new(
2005 mz_persist_client::cfg::CRDB_CONNECT_TIMEOUT.name(),
2006 ))
2007 }
2008
2009 pub fn crdb_tcp_user_timeout(&self) -> Duration {
2011 *self.expect_config_value(UncasedStr::new(
2012 mz_persist_client::cfg::CRDB_TCP_USER_TIMEOUT.name(),
2013 ))
2014 }
2015
2016 pub fn crdb_keepalives_idle(&self) -> Duration {
2018 *self.expect_config_value(UncasedStr::new(
2019 mz_persist_client::cfg::CRDB_KEEPALIVES_IDLE.name(),
2020 ))
2021 }
2022
2023 pub fn crdb_keepalives_interval(&self) -> Duration {
2025 *self.expect_config_value(UncasedStr::new(
2026 mz_persist_client::cfg::CRDB_KEEPALIVES_INTERVAL.name(),
2027 ))
2028 }
2029
2030 pub fn crdb_keepalives_retries(&self) -> u32 {
2032 *self.expect_config_value(UncasedStr::new(
2033 mz_persist_client::cfg::CRDB_KEEPALIVES_RETRIES.name(),
2034 ))
2035 }
2036
2037 pub fn storage_dataflow_max_inflight_bytes(&self) -> Option<usize> {
2039 *self.expect_value(&STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES)
2040 }
2041
2042 pub fn storage_dataflow_max_inflight_bytes_to_cluster_size_fraction(&self) -> Option<Numeric> {
2044 *self.expect_value(&STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES_TO_CLUSTER_SIZE_FRACTION)
2045 }
2046
2047 pub fn storage_shrink_upsert_unused_buffers_by_ratio(&self) -> usize {
2049 *self.expect_value(&STORAGE_SHRINK_UPSERT_UNUSED_BUFFERS_BY_RATIO)
2050 }
2051
2052 pub fn storage_dataflow_max_inflight_bytes_disk_only(&self) -> bool {
2054 *self.expect_value(&STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES_DISK_ONLY)
2055 }
2056
2057 pub fn storage_statistics_interval(&self) -> Duration {
2059 *self.expect_value(&STORAGE_STATISTICS_INTERVAL)
2060 }
2061
2062 pub fn storage_statistics_collection_interval(&self) -> Duration {
2064 *self.expect_value(&STORAGE_STATISTICS_COLLECTION_INTERVAL)
2065 }
2066
2067 pub fn storage_record_source_sink_namespaced_errors(&self) -> bool {
2069 *self.expect_value(&STORAGE_RECORD_SOURCE_SINK_NAMESPACED_ERRORS)
2070 }
2071
2072 pub fn persist_stats_filter_enabled(&self) -> bool {
2074 *self.expect_config_value(UncasedStr::new(
2075 mz_persist_client::stats::STATS_FILTER_ENABLED.name(),
2076 ))
2077 }
2078
2079 pub fn scram_iterations(&self) -> NonZeroU32 {
2080 *self.expect_value(&SCRAM_ITERATIONS)
2081 }
2082
2083 pub fn dyncfg_updates(&self) -> ConfigUpdates {
2088 let mut updates = ConfigUpdates::default();
2089 for entry in self.dyncfgs.entries() {
2090 let name = UncasedStr::new(entry.name());
2091 let val = match entry.val() {
2092 ConfigVal::Bool(_) => ConfigVal::from(*self.expect_config_value::<bool>(name)),
2093 ConfigVal::U32(_) => ConfigVal::from(*self.expect_config_value::<u32>(name)),
2094 ConfigVal::Usize(_) => ConfigVal::from(*self.expect_config_value::<usize>(name)),
2095 ConfigVal::OptUsize(_) => {
2096 ConfigVal::from(*self.expect_config_value::<Option<usize>>(name))
2097 }
2098 ConfigVal::F64(_) => ConfigVal::from(*self.expect_config_value::<f64>(name)),
2099 ConfigVal::String(_) => {
2100 ConfigVal::from(self.expect_config_value::<String>(name).clone())
2101 }
2102 ConfigVal::OptString(_) => {
2103 ConfigVal::from(self.expect_config_value::<Option<String>>(name).clone())
2104 }
2105 ConfigVal::Duration(_) => {
2106 ConfigVal::from(*self.expect_config_value::<Duration>(name))
2107 }
2108 ConfigVal::Json(_) => {
2109 ConfigVal::from(self.expect_config_value::<serde_json::Value>(name).clone())
2110 }
2111 };
2112 updates.add_dynamic(entry.name(), val);
2113 }
2114 updates
2115 }
2116
2117 pub fn sync_dyncfgs(&self) -> ConfigUpdates {
2125 let updates = self.dyncfg_updates();
2126 updates.apply(&self.dyncfgs);
2127 updates
2128 }
2129
2130 pub fn metrics_retention(&self) -> Duration {
2132 *self.expect_value(&METRICS_RETENTION)
2133 }
2134
2135 pub fn unsafe_mock_audit_event_timestamp(&self) -> Option<mz_repr::Timestamp> {
2137 *self.expect_value(&UNSAFE_MOCK_AUDIT_EVENT_TIMESTAMP)
2138 }
2139
2140 pub fn enable_rbac_checks(&self) -> bool {
2142 *self.expect_value(&ENABLE_RBAC_CHECKS)
2143 }
2144
2145 pub fn max_connections(&self) -> u32 {
2147 *self.expect_value(&MAX_CONNECTIONS)
2148 }
2149
2150 pub fn default_network_policy_name(&self) -> String {
2151 self.expect_value::<String>(&NETWORK_POLICY).clone()
2152 }
2153
2154 pub fn superuser_reserved_connections(&self) -> u32 {
2156 *self.expect_value(&SUPERUSER_RESERVED_CONNECTIONS)
2157 }
2158
2159 pub fn keep_n_source_status_history_entries(&self) -> usize {
2160 *self.expect_value(&KEEP_N_SOURCE_STATUS_HISTORY_ENTRIES)
2161 }
2162
2163 pub fn keep_n_sink_status_history_entries(&self) -> usize {
2164 *self.expect_value(&KEEP_N_SINK_STATUS_HISTORY_ENTRIES)
2165 }
2166
2167 pub fn keep_n_privatelink_status_history_entries(&self) -> usize {
2168 *self.expect_value(&KEEP_N_PRIVATELINK_STATUS_HISTORY_ENTRIES)
2169 }
2170
2171 pub fn replica_status_history_retention_window(&self) -> Duration {
2172 *self.expect_value(&REPLICA_STATUS_HISTORY_RETENTION_WINDOW)
2173 }
2174
2175 pub fn enable_storage_shard_finalization(&self) -> bool {
2177 *self.expect_value(&ENABLE_STORAGE_SHARD_FINALIZATION)
2178 }
2179
2180 pub fn enable_default_connection_validation(&self) -> bool {
2182 *self.expect_value(&ENABLE_DEFAULT_CONNECTION_VALIDATION)
2183 }
2184
2185 pub fn default_timestamp_interval(&self) -> Duration {
2187 *self.expect_value(&DEFAULT_TIMESTAMP_INTERVAL)
2188 }
2189
2190 pub fn min_timestamp_interval(&self) -> Duration {
2192 *self.expect_value(&MIN_TIMESTAMP_INTERVAL)
2193 }
2194 pub fn max_timestamp_interval(&self) -> Duration {
2196 *self.expect_value(&MAX_TIMESTAMP_INTERVAL)
2197 }
2198
2199 pub fn logging_filter(&self) -> CloneableEnvFilter {
2200 self.expect_value::<CloneableEnvFilter>(&LOGGING_FILTER)
2201 .clone()
2202 }
2203
2204 pub fn opentelemetry_filter(&self) -> CloneableEnvFilter {
2205 self.expect_value::<CloneableEnvFilter>(&OPENTELEMETRY_FILTER)
2206 .clone()
2207 }
2208
2209 pub fn logging_filter_defaults(&self) -> Vec<SerializableDirective> {
2210 self.expect_value::<Vec<SerializableDirective>>(&LOGGING_FILTER_DEFAULTS)
2211 .clone()
2212 }
2213
2214 pub fn opentelemetry_filter_defaults(&self) -> Vec<SerializableDirective> {
2215 self.expect_value::<Vec<SerializableDirective>>(&OPENTELEMETRY_FILTER_DEFAULTS)
2216 .clone()
2217 }
2218
2219 pub fn sentry_filters(&self) -> Vec<SerializableDirective> {
2220 self.expect_value::<Vec<SerializableDirective>>(&SENTRY_FILTERS)
2221 .clone()
2222 }
2223
2224 pub fn webhooks_secrets_caching_ttl_secs(&self) -> usize {
2225 *self.expect_value(&WEBHOOKS_SECRETS_CACHING_TTL_SECS)
2226 }
2227
2228 pub fn coord_slow_message_warn_threshold(&self) -> Duration {
2229 *self.expect_value(&COORD_SLOW_MESSAGE_WARN_THRESHOLD)
2230 }
2231
2232 pub fn grpc_client_http2_keep_alive_interval(&self) -> Duration {
2233 *self.expect_value(&grpc_client::HTTP2_KEEP_ALIVE_INTERVAL)
2234 }
2235
2236 pub fn grpc_client_http2_keep_alive_timeout(&self) -> Duration {
2237 *self.expect_value(&grpc_client::HTTP2_KEEP_ALIVE_TIMEOUT)
2238 }
2239
2240 pub fn grpc_connect_timeout(&self) -> Duration {
2241 *self.expect_value(&grpc_client::CONNECT_TIMEOUT)
2242 }
2243
2244 pub fn cluster_multi_process_replica_az_affinity_weight(&self) -> Option<i32> {
2245 *self.expect_value(&cluster_scheduling::CLUSTER_MULTI_PROCESS_REPLICA_AZ_AFFINITY_WEIGHT)
2246 }
2247
2248 pub fn cluster_soften_replication_anti_affinity(&self) -> bool {
2249 *self.expect_value(&cluster_scheduling::CLUSTER_SOFTEN_REPLICATION_ANTI_AFFINITY)
2250 }
2251
2252 pub fn cluster_soften_replication_anti_affinity_weight(&self) -> i32 {
2253 *self.expect_value(&cluster_scheduling::CLUSTER_SOFTEN_REPLICATION_ANTI_AFFINITY_WEIGHT)
2254 }
2255
2256 pub fn cluster_enable_topology_spread(&self) -> bool {
2257 *self.expect_value(&cluster_scheduling::CLUSTER_ENABLE_TOPOLOGY_SPREAD)
2258 }
2259
2260 pub fn cluster_topology_spread_ignore_non_singular_scale(&self) -> bool {
2261 *self.expect_value(&cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_IGNORE_NON_SINGULAR_SCALE)
2262 }
2263
2264 pub fn cluster_topology_spread_max_skew(&self) -> i32 {
2265 *self.expect_value(&cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_MAX_SKEW)
2266 }
2267
2268 pub fn cluster_topology_spread_set_min_domains(&self) -> Option<i32> {
2269 *self.expect_value(&cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_MIN_DOMAINS)
2270 }
2271
2272 pub fn cluster_topology_spread_soft(&self) -> bool {
2273 *self.expect_value(&cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_SOFT)
2274 }
2275
2276 pub fn cluster_soften_az_affinity(&self) -> bool {
2277 *self.expect_value(&cluster_scheduling::CLUSTER_SOFTEN_AZ_AFFINITY)
2278 }
2279
2280 pub fn cluster_soften_az_affinity_weight(&self) -> i32 {
2281 *self.expect_value(&cluster_scheduling::CLUSTER_SOFTEN_AZ_AFFINITY_WEIGHT)
2282 }
2283
2284 pub fn cluster_alter_check_ready_interval(&self) -> Duration {
2285 *self.expect_value(&cluster_scheduling::CLUSTER_ALTER_CHECK_READY_INTERVAL)
2286 }
2287
2288 pub fn cluster_security_context_enabled(&self) -> bool {
2289 *self.expect_value(&cluster_scheduling::CLUSTER_SECURITY_CONTEXT_ENABLED)
2290 }
2291
2292 pub fn cluster_refresh_mv_compaction_estimate(&self) -> Duration {
2293 *self.expect_value(&cluster_scheduling::CLUSTER_REFRESH_MV_COMPACTION_ESTIMATE)
2294 }
2295
2296 pub fn privatelink_status_update_quota_per_minute(&self) -> u32 {
2298 *self.expect_value(&PRIVATELINK_STATUS_UPDATE_QUOTA_PER_MINUTE)
2299 }
2300
2301 pub fn statement_logging_target_data_rate(&self) -> Option<usize> {
2302 *self.expect_value(&STATEMENT_LOGGING_TARGET_DATA_RATE)
2303 }
2304
2305 pub fn statement_logging_max_data_credit(&self) -> Option<usize> {
2306 *self.expect_value(&STATEMENT_LOGGING_MAX_DATA_CREDIT)
2307 }
2308
2309 pub fn statement_logging_max_sample_rate(&self) -> Numeric {
2311 *self.expect_value(&STATEMENT_LOGGING_MAX_SAMPLE_RATE)
2312 }
2313
2314 pub fn statement_logging_default_sample_rate(&self) -> Numeric {
2316 *self.expect_value(&STATEMENT_LOGGING_DEFAULT_SAMPLE_RATE)
2317 }
2318
2319 pub fn enable_internal_statement_logging(&self) -> bool {
2321 *self.expect_value(&ENABLE_INTERNAL_STATEMENT_LOGGING)
2322 }
2323
2324 pub fn enable_statement_arrival_logging(&self) -> bool {
2326 *self.expect_value(&ENABLE_STATEMENT_ARRIVAL_LOGGING)
2327 }
2328
2329 pub fn enable_extended_protocol_implicit_transaction(&self) -> bool {
2332 *self.expect_value(&ENABLE_EXTENDED_PROTOCOL_IMPLICIT_TRANSACTION)
2333 }
2334
2335 pub fn optimizer_stats_timeout(&self) -> Duration {
2337 *self.expect_value(&OPTIMIZER_STATS_TIMEOUT)
2338 }
2339
2340 pub fn optimizer_oneshot_stats_timeout(&self) -> Duration {
2342 *self.expect_value(&OPTIMIZER_ONESHOT_STATS_TIMEOUT)
2343 }
2344
2345 pub fn webhook_concurrent_request_limit(&self) -> usize {
2347 *self.expect_value(&WEBHOOK_CONCURRENT_REQUEST_LIMIT)
2348 }
2349
2350 pub fn pg_timestamp_oracle_connection_pool_max_size(&self) -> usize {
2352 *self.expect_value(&PG_TIMESTAMP_ORACLE_CONNECTION_POOL_MAX_SIZE)
2353 }
2354
2355 pub fn pg_timestamp_oracle_connection_pool_max_wait(&self) -> Option<Duration> {
2357 *self.expect_value(&PG_TIMESTAMP_ORACLE_CONNECTION_POOL_MAX_WAIT)
2358 }
2359
2360 pub fn pg_timestamp_oracle_connection_pool_ttl(&self) -> Duration {
2362 *self.expect_value(&PG_TIMESTAMP_ORACLE_CONNECTION_POOL_TTL)
2363 }
2364
2365 pub fn pg_timestamp_oracle_connection_pool_ttl_stagger(&self) -> Duration {
2367 *self.expect_value(&PG_TIMESTAMP_ORACLE_CONNECTION_POOL_TTL_STAGGER)
2368 }
2369
2370 pub fn user_storage_managed_collections_batch_duration(&self) -> Duration {
2372 *self.expect_value(&USER_STORAGE_MANAGED_COLLECTIONS_BATCH_DURATION)
2373 }
2374
2375 pub fn force_source_table_syntax(&self) -> bool {
2376 *self.expect_value(&FORCE_SOURCE_TABLE_SYNTAX)
2377 }
2378
2379 pub fn optimizer_e2e_latency_warning_threshold(&self) -> Duration {
2380 *self.expect_value(&OPTIMIZER_E2E_LATENCY_WARNING_THRESHOLD)
2381 }
2382
2383 pub fn is_controller_config_var(&self, name: &str) -> bool {
2385 self.is_dyncfg_var(name)
2386 }
2387
2388 pub fn is_compute_config_var(&self, name: &str) -> bool {
2392 name == MAX_RESULT_SIZE.name() || self.is_dyncfg_var(name) || is_tracing_var(name)
2393 }
2394
2395 pub fn is_metrics_config_var(&self, name: &str) -> bool {
2397 self.is_dyncfg_var(name)
2398 }
2399
2400 pub fn is_storage_config_var(&self, name: &str) -> bool {
2402 name == PG_SOURCE_CONNECT_TIMEOUT.name()
2403 || name == PG_SOURCE_TCP_KEEPALIVES_IDLE.name()
2404 || name == PG_SOURCE_TCP_KEEPALIVES_INTERVAL.name()
2405 || name == PG_SOURCE_TCP_KEEPALIVES_RETRIES.name()
2406 || name == PG_SOURCE_TCP_USER_TIMEOUT.name()
2407 || name == PG_SOURCE_TCP_CONFIGURE_SERVER.name()
2408 || name == PG_SOURCE_SNAPSHOT_STATEMENT_TIMEOUT.name()
2409 || name == PG_SOURCE_WAL_SENDER_TIMEOUT.name()
2410 || name == PG_SOURCE_SNAPSHOT_COLLECT_STRICT_COUNT.name()
2411 || name == MYSQL_SOURCE_TCP_KEEPALIVE.name()
2412 || name == MYSQL_SOURCE_SNAPSHOT_MAX_EXECUTION_TIME.name()
2413 || name == MYSQL_SOURCE_SNAPSHOT_LOCK_WAIT_TIMEOUT.name()
2414 || name == MYSQL_SOURCE_SNAPSHOT_WAIT_TIMEOUT.name()
2415 || name == MYSQL_SOURCE_CONNECT_TIMEOUT.name()
2416 || name == ENABLE_STORAGE_SHARD_FINALIZATION.name()
2417 || name == SSH_CHECK_INTERVAL.name()
2418 || name == SSH_CONNECT_TIMEOUT.name()
2419 || name == SSH_KEEPALIVES_IDLE.name()
2420 || name == KAFKA_SOCKET_KEEPALIVE.name()
2421 || name == KAFKA_SOCKET_TIMEOUT.name()
2422 || name == KAFKA_TRANSACTION_TIMEOUT.name()
2423 || name == KAFKA_SOCKET_CONNECTION_SETUP_TIMEOUT.name()
2424 || name == KAFKA_FETCH_METADATA_TIMEOUT.name()
2425 || name == KAFKA_PROGRESS_RECORD_FETCH_TIMEOUT.name()
2426 || name == STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES.name()
2427 || name == STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES_TO_CLUSTER_SIZE_FRACTION.name()
2428 || name == STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES_DISK_ONLY.name()
2429 || name == STORAGE_SHRINK_UPSERT_UNUSED_BUFFERS_BY_RATIO.name()
2430 || name == STORAGE_RECORD_SOURCE_SINK_NAMESPACED_ERRORS.name()
2431 || name == STORAGE_STATISTICS_INTERVAL.name()
2432 || name == STORAGE_STATISTICS_COLLECTION_INTERVAL.name()
2433 || name == USER_STORAGE_MANAGED_COLLECTIONS_BATCH_DURATION.name()
2434 || is_upsert_rocksdb_config_var(name)
2435 || self.is_dyncfg_var(name)
2436 || is_tracing_var(name)
2437 }
2438
2439 fn is_dyncfg_var(&self, name: &str) -> bool {
2441 self.dyncfgs.entries().any(|e| name == e.name())
2442 }
2443}
2444
2445pub fn is_tracing_var(name: &str) -> bool {
2446 name == LOGGING_FILTER.name()
2447 || name == LOGGING_FILTER_DEFAULTS.name()
2448 || name == OPENTELEMETRY_FILTER.name()
2449 || name == OPENTELEMETRY_FILTER_DEFAULTS.name()
2450 || name == SENTRY_FILTERS.name()
2451}
2452
2453pub fn is_secrets_caching_var(name: &str) -> bool {
2455 name == WEBHOOKS_SECRETS_CACHING_TTL_SECS.name()
2456}
2457
2458fn is_upsert_rocksdb_config_var(name: &str) -> bool {
2459 name == upsert_rocksdb::UPSERT_ROCKSDB_COMPACTION_STYLE.name()
2460 || name == upsert_rocksdb::UPSERT_ROCKSDB_OPTIMIZE_COMPACTION_MEMTABLE_BUDGET.name()
2461 || name == upsert_rocksdb::UPSERT_ROCKSDB_LEVEL_COMPACTION_DYNAMIC_LEVEL_BYTES.name()
2462 || name == upsert_rocksdb::UPSERT_ROCKSDB_UNIVERSAL_COMPACTION_RATIO.name()
2463 || name == upsert_rocksdb::UPSERT_ROCKSDB_PARALLELISM.name()
2464 || name == upsert_rocksdb::UPSERT_ROCKSDB_COMPRESSION_TYPE.name()
2465 || name == upsert_rocksdb::UPSERT_ROCKSDB_BOTTOMMOST_COMPRESSION_TYPE.name()
2466 || name == upsert_rocksdb::UPSERT_ROCKSDB_BATCH_SIZE.name()
2467 || name == upsert_rocksdb::UPSERT_ROCKSDB_STATS_LOG_INTERVAL_SECONDS.name()
2468 || name == upsert_rocksdb::UPSERT_ROCKSDB_STATS_PERSIST_INTERVAL_SECONDS.name()
2469 || name == upsert_rocksdb::UPSERT_ROCKSDB_POINT_LOOKUP_BLOCK_CACHE_SIZE_MB.name()
2470 || name == upsert_rocksdb::UPSERT_ROCKSDB_SHRINK_ALLOCATED_BUFFERS_BY_RATIO.name()
2471}
2472
2473pub fn is_timestamp_oracle_config_var(name: &str) -> bool {
2476 name == PG_TIMESTAMP_ORACLE_CONNECTION_POOL_MAX_SIZE.name()
2477 || name == PG_TIMESTAMP_ORACLE_CONNECTION_POOL_MAX_WAIT.name()
2478 || name == PG_TIMESTAMP_ORACLE_CONNECTION_POOL_TTL.name()
2479 || name == PG_TIMESTAMP_ORACLE_CONNECTION_POOL_TTL_STAGGER.name()
2480 || name == CRDB_CONNECT_TIMEOUT.name()
2481 || name == CRDB_TCP_USER_TIMEOUT.name()
2482 || name == CRDB_KEEPALIVES_IDLE.name()
2483 || name == CRDB_KEEPALIVES_INTERVAL.name()
2484 || name == CRDB_KEEPALIVES_RETRIES.name()
2485 || name == mz_adapter_types::dyncfgs::PG_TIMESTAMP_ORACLE_STATEMENT_TIMEOUT.name()
2486}
2487
2488pub fn is_cluster_scheduling_var(name: &str) -> bool {
2490 name == cluster_scheduling::CLUSTER_MULTI_PROCESS_REPLICA_AZ_AFFINITY_WEIGHT.name()
2491 || name == cluster_scheduling::CLUSTER_SOFTEN_REPLICATION_ANTI_AFFINITY.name()
2492 || name == cluster_scheduling::CLUSTER_SOFTEN_REPLICATION_ANTI_AFFINITY_WEIGHT.name()
2493 || name == cluster_scheduling::CLUSTER_ENABLE_TOPOLOGY_SPREAD.name()
2494 || name == cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_IGNORE_NON_SINGULAR_SCALE.name()
2495 || name == cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_MAX_SKEW.name()
2496 || name == cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_MIN_DOMAINS.name()
2497 || name == cluster_scheduling::CLUSTER_TOPOLOGY_SPREAD_SOFT.name()
2498 || name == cluster_scheduling::CLUSTER_SOFTEN_AZ_AFFINITY.name()
2499 || name == cluster_scheduling::CLUSTER_SOFTEN_AZ_AFFINITY_WEIGHT.name()
2500}
2501
2502pub fn is_http_config_var(name: &str) -> bool {
2504 name == WEBHOOK_CONCURRENT_REQUEST_LIMIT.name()
2505}
2506
2507static SESSION_SYSTEM_VARS: LazyLock<BTreeMap<&'static UncasedStr, &'static VarDefinition>> =
2511 LazyLock::new(|| {
2512 [
2513 &APPLICATION_NAME,
2514 &CLIENT_ENCODING,
2515 &CLIENT_MIN_MESSAGES,
2516 &CLUSTER,
2517 &CLUSTER_REPLICA,
2518 &DEFAULT_CLUSTER_REPLICATION_FACTOR,
2519 &CURRENT_OBJECT_MISSING_WARNINGS,
2520 &DATABASE,
2521 &DATE_STYLE,
2522 &EXTRA_FLOAT_DIGITS,
2523 &INTEGER_DATETIMES,
2524 &INTERVAL_STYLE,
2525 &REAL_TIME_RECENCY_TIMEOUT,
2526 &SEARCH_PATH,
2527 &STANDARD_CONFORMING_STRINGS,
2528 &STATEMENT_TIMEOUT,
2529 &IDLE_IN_TRANSACTION_SESSION_TIMEOUT,
2530 &TIMEZONE,
2531 &TRANSACTION_ISOLATION,
2532 &MAX_QUERY_RESULT_SIZE,
2533 ]
2534 .into_iter()
2535 .map(|var| (UncasedStr::new(var.name()), var))
2536 .collect()
2537 });
2538
2539#[derive(Debug)]
2542pub struct FeatureFlag {
2543 pub flag: &'static VarDefinition,
2544 pub feature_desc: &'static str,
2545}
2546
2547impl FeatureFlag {
2548 pub fn require(&'static self, system_vars: &SystemVars) -> Result<(), VarError> {
2551 match *system_vars.expect_value::<bool>(self.flag) {
2552 true => Ok(()),
2553 false => Err(VarError::RequiresFeatureFlag { feature_flag: self }),
2554 }
2555 }
2556}
2557
2558impl PartialEq for FeatureFlag {
2559 fn eq(&self, other: &FeatureFlag) -> bool {
2560 self.flag.name() == other.flag.name()
2561 }
2562}
2563
2564impl Eq for FeatureFlag {}
2565
2566impl Var for MzVersion {
2567 fn name(&self) -> &'static str {
2568 MZ_VERSION_NAME.as_str()
2569 }
2570
2571 fn value(&self) -> String {
2572 self.build_info
2573 .human_version(self.helm_chart_version.clone())
2574 }
2575
2576 fn description(&self) -> &'static str {
2577 "Shows the Materialize server version (Materialize)."
2578 }
2579
2580 fn type_name(&self) -> Cow<'static, str> {
2581 String::type_name()
2582 }
2583
2584 fn visible(&self, _: &User, _: &SystemVars) -> Result<(), VarError> {
2585 Ok(())
2586 }
2587}
2588
2589impl Var for User {
2590 fn name(&self) -> &'static str {
2591 IS_SUPERUSER_NAME.as_str()
2592 }
2593
2594 fn value(&self) -> String {
2595 self.is_superuser().format()
2596 }
2597
2598 fn description(&self) -> &'static str {
2599 "Reports whether the current session is a superuser (PostgreSQL)."
2600 }
2601
2602 fn type_name(&self) -> Cow<'static, str> {
2603 bool::type_name()
2604 }
2605
2606 fn visible(&self, _: &User, _: &SystemVars) -> Result<(), VarError> {
2607 Ok(())
2608 }
2609}
2610
2611#[cfg(test)]
2612mod isolation_feature_flag_tests {
2613 use super::*;
2614
2615 #[mz_ore::test]
2616 fn gates_bounded_staleness_value() {
2617 let mut system_vars = SystemVars::new();
2618
2619 check_transaction_isolation_feature_flag(
2621 TRANSACTION_ISOLATION_VAR_NAME,
2622 VarInput::Flat("bounded staleness 5s"),
2623 &system_vars,
2624 )
2625 .expect("flag on by default");
2626
2627 system_vars
2632 .set("enable_bounded_staleness_isolation", VarInput::Flat("off"))
2633 .expect("set flag");
2634 for name in ["transaction_isolation", "TRANSACTION_ISOLATION"] {
2635 let err = check_transaction_isolation_feature_flag(
2636 name,
2637 VarInput::Flat("bounded staleness 5s"),
2638 &system_vars,
2639 )
2640 .expect_err("flag off rejects bounded staleness");
2641 assert!(matches!(err, VarError::RequiresFeatureFlag { .. }));
2642 }
2643
2644 check_transaction_isolation_feature_flag(
2646 TRANSACTION_ISOLATION_VAR_NAME,
2647 VarInput::Flat("serializable"),
2648 &system_vars,
2649 )
2650 .expect("serializable always allowed");
2651
2652 check_transaction_isolation_feature_flag(
2654 CLUSTER.name(),
2655 VarInput::Flat("bounded staleness 5s"),
2656 &system_vars,
2657 )
2658 .expect("unrelated var ignored");
2659 }
2660}
2661
2662#[cfg(test)]
2663mod reset_all_tests {
2664 use super::*;
2665 use crate::session::user::SYSTEM_USER;
2666
2667 fn test_vars() -> SessionVars {
2668 SessionVars::new_unchecked(&mz_build_info::DUMMY_BUILD_INFO, SYSTEM_USER.clone(), None)
2669 }
2670
2671 #[mz_ore::test]
2675 fn reset_all_clears_committed_session_value() {
2676 let system_vars = SystemVars::new();
2677 let mut vars = test_vars();
2678 let default = vars.application_name().to_string();
2679
2680 vars.set(
2682 &system_vars,
2683 "application_name",
2684 VarInput::Flat("custom"),
2685 false,
2686 )
2687 .expect("set");
2688 vars.end_transaction(EndTransactionAction::Commit);
2689 assert_eq!(vars.application_name(), "custom");
2690 assert_eq!(
2691 vars.inspect("application_name")
2692 .unwrap()
2693 .inspect_session_value()
2694 .map(|v| v.format()),
2695 Some("custom".to_string())
2696 );
2697
2698 vars.reset_all();
2699
2700 let var = vars.inspect("application_name").unwrap();
2703 assert_eq!(vars.application_name(), default);
2704 assert_eq!(var.inspect_session_value(), None);
2705 assert!(!var.is_mutating());
2706 }
2707
2708 #[mz_ore::test]
2712 fn reset_all_preserves_installed_default() {
2713 let system_vars = SystemVars::new();
2714 let mut vars = test_vars();
2715
2716 vars.set_default("application_name", VarInput::Flat("startup_default"))
2717 .expect("set_default");
2718 vars.set(
2719 &system_vars,
2720 "application_name",
2721 VarInput::Flat("custom"),
2722 false,
2723 )
2724 .expect("set");
2725 vars.end_transaction(EndTransactionAction::Commit);
2726 assert_eq!(vars.application_name(), "custom");
2727
2728 vars.reset_all();
2729
2730 assert_eq!(vars.application_name(), "startup_default");
2731 assert_eq!(
2732 vars.inspect("application_name")
2733 .unwrap()
2734 .inspect_session_value(),
2735 None
2736 );
2737 }
2738}