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_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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
114pub enum EndTransactionAction {
115 Commit,
117 Rollback,
119}
120
121#[derive(Debug, Clone, Copy)]
127pub enum VarInput<'a> {
128 Flat(&'a str),
136 SqlSet(&'a [String]),
142}
143
144impl<'a> VarInput<'a> {
145 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#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize)]
156pub enum OwnedVarInput {
157 Flat(String),
164 SqlSet(Vec<String>),
168}
169
170impl OwnedVarInput {
171 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
180pub trait Var: Debug {
182 fn name(&self) -> &'static str;
184
185 fn value(&self) -> String;
191
192 fn description(&self) -> &'static str;
195
196 fn type_name(&self) -> Cow<'static, str>;
198
199 fn visible(&self, user: &User, system_vars: &SystemVars) -> Result<(), VarError>;
204
205 fn is_unsafe(&self) -> bool {
207 self.name().starts_with("unsafe_")
208 }
209
210 fn scope(&self) -> ParameterScope {
214 ParameterScope::Environment
215 }
216
217 fn as_var(&self) -> &dyn Var
220 where
221 Self: Sized,
222 {
223 self
224 }
225}
226
227#[derive(Debug)]
234pub struct SessionVar {
235 definition: VarDefinition,
236 default_value: Option<Box<dyn Value>>,
238 local_value: Option<Box<dyn Value>>,
240 staged_value: Option<Box<dyn Value>>,
242 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 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 pub fn set(&mut self, input: VarInput, local: bool) -> Result<(), VarError> {
280 let v = self.definition.parse(input)?;
281
282 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 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 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 pub fn reset_durable(&mut self) {
326 self.local_value = None;
327 self.staged_value = None;
328 self.session_value = None;
329 }
330
331 #[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 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 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 build_info: &'static BuildInfo,
409 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#[derive(Debug, Clone)]
427pub struct SessionVars {
428 vars: OrdMap<&'static UncasedStr, SessionVar>,
430 mz_version: MzVersion,
432 user: User,
434}
435
436impl SessionVars {
437 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 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 pub fn notify_set(&self) -> impl Iterator<Item = &dyn Var> {
504 [
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 &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 .chain(std::iter::once(self.mz_version.as_var()))
538 }
539
540 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 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 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 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 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 if !Self::allow_role_default(name) {
652 self.check_read_only(name)?;
653 }
654
655 self.vars
656 .get_mut(name)
657 .map(|v| v.set_default(input))
659 .transpose()?
660 .ok_or_else(|| VarError::UnknownParameter(name.to_string()))
661 }
662
663 fn allow_role_default(name: &UncasedStr) -> bool {
671 name == RESTRICT_TO_USER_OBJECTS.name
672 }
673
674 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 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 Err(VarError::ReadOnlyParameter(
729 RESTRICT_TO_USER_OBJECTS.name.as_str(),
730 ))
731 } else {
732 Ok(())
733 }
734 }
735
736 #[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 if before != after {
758 changed.insert(var.name(), after);
759 }
760 }
761 self.vars.extend(updates);
762 changed
763 }
764
765 pub fn application_name(&self) -> &str {
767 self.expect_value::<String>(&APPLICATION_NAME).as_str()
768 }
769
770 pub fn build_info(&self) -> &'static BuildInfo {
772 self.mz_version.build_info
773 }
774
775 pub fn client_encoding(&self) -> &ClientEncoding {
777 self.expect_value(&CLIENT_ENCODING)
778 }
779
780 pub fn client_min_messages(&self) -> &ClientSeverity {
782 self.expect_value(&CLIENT_MIN_MESSAGES)
783 }
784
785 pub fn cluster(&self) -> &str {
787 self.expect_value::<String>(&CLUSTER).as_str()
788 }
789
790 pub fn cluster_replica(&self) -> Option<&str> {
792 self.expect_value::<Option<String>>(&CLUSTER_REPLICA)
793 .as_deref()
794 }
795
796 pub fn current_object_missing_warnings(&self) -> bool {
799 *self.expect_value::<bool>(&CURRENT_OBJECT_MISSING_WARNINGS)
800 }
801
802 pub fn date_style(&self) -> &[&str] {
804 &self.expect_value::<DateStyle>(&DATE_STYLE).0
805 }
806
807 pub fn database(&self) -> &str {
809 self.expect_value::<String>(&DATABASE).as_str()
810 }
811
812 pub fn extra_float_digits(&self) -> i32 {
814 *self.expect_value(&EXTRA_FLOAT_DIGITS)
815 }
816
817 pub fn text_encode_settings(&self) -> TextEncodeSettings {
820 TextEncodeSettings {
821 extra_float_digits: self.extra_float_digits(),
822 }
823 }
824
825 pub fn integer_datetimes(&self) -> bool {
827 *self.expect_value(&INTEGER_DATETIMES)
828 }
829
830 pub fn intervalstyle(&self) -> &IntervalStyle {
832 self.expect_value(&INTERVAL_STYLE)
833 }
834
835 pub fn mz_version(&self) -> String {
837 self.mz_version.value()
838 }
839
840 pub fn search_path(&self) -> &[Ident] {
842 self.expect_value::<Vec<Ident>>(&SEARCH_PATH).as_slice()
843 }
844
845 pub fn server_version(&self) -> &str {
847 self.expect_value::<String>(&SERVER_VERSION).as_str()
848 }
849
850 pub fn server_version_num(&self) -> i32 {
852 *self.expect_value(&SERVER_VERSION_NUM)
853 }
854
855 pub fn sql_safe_updates(&self) -> bool {
857 *self.expect_value(&SQL_SAFE_UPDATES)
858 }
859
860 pub fn standard_conforming_strings(&self) -> bool {
863 *self.expect_value(&STANDARD_CONFORMING_STRINGS)
864 }
865
866 pub fn statement_timeout(&self) -> &Duration {
868 self.expect_value(&STATEMENT_TIMEOUT)
869 }
870
871 pub fn idle_in_transaction_session_timeout(&self) -> &Duration {
873 self.expect_value(&IDLE_IN_TRANSACTION_SESSION_TIMEOUT)
874 }
875
876 pub fn timezone(&self) -> &TimeZone {
878 self.expect_value(&TIMEZONE)
879 }
880
881 pub fn transaction_isolation(&self) -> &IsolationLevel {
884 self.expect_value(&TRANSACTION_ISOLATION)
885 }
886
887 pub fn real_time_recency(&self) -> bool {
889 *self.expect_value(&REAL_TIME_RECENCY)
890 }
891
892 pub fn real_time_recency_timeout(&self) -> &Duration {
894 self.expect_value(&REAL_TIME_RECENCY_TIMEOUT)
895 }
896
897 pub fn emit_plan_insights_notice(&self) -> bool {
899 *self.expect_value(&EMIT_PLAN_INSIGHTS_NOTICE)
900 }
901
902 pub fn emit_execution_time_notice(&self) -> bool {
904 *self.expect_value(&EMIT_EXECUTION_TIME_NOTICE)
905 }
906
907 pub fn emit_timestamp_notice(&self) -> bool {
909 *self.expect_value(&EMIT_TIMESTAMP_NOTICE)
910 }
911
912 pub fn emit_trace_id_notice(&self) -> bool {
914 *self.expect_value(&EMIT_TRACE_ID_NOTICE)
915 }
916
917 pub fn auto_route_catalog_queries(&self) -> bool {
919 *self.expect_value(&AUTO_ROUTE_CATALOG_QUERIES)
920 }
921
922 pub fn enable_session_rbac_checks(&self) -> bool {
924 *self.expect_value(&ENABLE_SESSION_RBAC_CHECKS)
925 }
926
927 pub fn restrict_to_user_objects(&self) -> bool {
929 *self.expect_value(&RESTRICT_TO_USER_OBJECTS)
930 }
931
932 pub fn enable_session_cardinality_estimates(&self) -> bool {
934 *self.expect_value(&ENABLE_SESSION_CARDINALITY_ESTIMATES)
935 }
936
937 pub fn is_superuser(&self) -> bool {
939 self.user.is_superuser()
940 }
941
942 pub fn user(&self) -> &User {
944 &self.user
945 }
946
947 pub fn max_query_result_size(&self) -> u64 {
949 self.expect_value::<ByteSize>(&MAX_QUERY_RESULT_SIZE)
950 .as_bytes()
951 }
952
953 pub fn set_internal_user_metadata(&mut self, metadata: InternalUserMetadata) {
955 self.user.internal_metadata = Some(metadata);
956 }
957
958 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 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 pub fn welcome_message(&self) -> bool {
996 *self.expect_value(&WELCOME_MESSAGE)
997 }
998}
999
1000pub const OLD_CATALOG_SERVER_CLUSTER: &str = "mz_introspection";
1002pub const OLD_AUTO_ROUTE_CATALOG_QUERIES: &str = "auto_route_introspection_queries";
1003
1004fn 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
1042pub 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 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#[derive(Debug)]
1071pub struct SystemVar {
1072 definition: VarDefinition,
1073 persisted_value: Option<Box<dyn Value>>,
1075 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 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#[derive(Derivative, Clone)]
1194#[derivative(Debug)]
1195pub struct SystemVars {
1196 allow_unsafe: bool,
1198 vars: BTreeMap<&'static UncasedStr, SystemVar>,
1200 #[derivative(Debug = "ignore")]
1202 callbacks: BTreeMap<String, Vec<Arc<dyn Fn(&SystemVars) + Send + Sync>>>,
1203
1204 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 var.scoped(cfg.scope())
1388 })
1389 .collect();
1390
1391 let vars: BTreeMap<_, _> = system_vars
1392 .into_iter()
1393 .chain(definitions::FEATURE_FLAGS.iter().copied())
1395 .chain(SESSION_SYSTEM_VARS.values().copied())
1397 .cloned()
1398 .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 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 pub fn iter_synced(&self) -> impl Iterator<Item = &dyn Var> {
1463 self.iter().filter(|v| v.name() != ENABLE_LAUNCHDARKLY.name)
1464 }
1465
1466 pub fn definition(&self, name: &str) -> Option<&VarDefinition> {
1468 self.vars.get(UncasedStr::new(name)).map(|v| &v.definition)
1469 }
1470
1471 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 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 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 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 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 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 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 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 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 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 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 fn notify_callbacks(&self, name: &str) {
1691 if let Some(callbacks) = self.callbacks.get(name) {
1693 for callback in callbacks {
1694 (callback)(self);
1695 }
1696 }
1697 }
1698
1699 pub fn default_cluster(&self) -> String {
1702 self.expect_value::<String>(&CLUSTER).to_owned()
1703 }
1704
1705 pub fn max_kafka_connections(&self) -> u32 {
1707 *self.expect_value(&MAX_KAFKA_CONNECTIONS)
1708 }
1709
1710 pub fn max_postgres_connections(&self) -> u32 {
1712 *self.expect_value(&MAX_POSTGRES_CONNECTIONS)
1713 }
1714
1715 pub fn max_mysql_connections(&self) -> u32 {
1717 *self.expect_value(&MAX_MYSQL_CONNECTIONS)
1718 }
1719
1720 pub fn max_sql_server_connections(&self) -> u32 {
1722 *self.expect_value(&MAX_SQL_SERVER_CONNECTIONS)
1723 }
1724
1725 pub fn max_aws_privatelink_connections(&self) -> u32 {
1727 *self.expect_value(&MAX_AWS_PRIVATELINK_CONNECTIONS)
1728 }
1729
1730 pub fn max_tables(&self) -> u32 {
1732 *self.expect_value(&MAX_TABLES)
1733 }
1734
1735 pub fn max_sources(&self) -> u32 {
1737 *self.expect_value(&MAX_SOURCES)
1738 }
1739
1740 pub fn max_sinks(&self) -> u32 {
1742 *self.expect_value(&MAX_SINKS)
1743 }
1744
1745 pub fn max_materialized_views(&self) -> u32 {
1747 *self.expect_value(&MAX_MATERIALIZED_VIEWS)
1748 }
1749
1750 pub fn max_clusters(&self) -> u32 {
1752 *self.expect_value(&MAX_CLUSTERS)
1753 }
1754
1755 pub fn max_replicas_per_cluster(&self) -> u32 {
1757 *self.expect_value(&MAX_REPLICAS_PER_CLUSTER)
1758 }
1759
1760 pub fn max_credit_consumption_rate(&self) -> Numeric {
1762 *self.expect_value(&MAX_CREDIT_CONSUMPTION_RATE)
1763 }
1764
1765 pub fn max_databases(&self) -> u32 {
1767 *self.expect_value(&MAX_DATABASES)
1768 }
1769
1770 pub fn max_schemas_per_database(&self) -> u32 {
1772 *self.expect_value(&MAX_SCHEMAS_PER_DATABASE)
1773 }
1774
1775 pub fn max_objects_per_schema(&self) -> u32 {
1777 *self.expect_value(&MAX_OBJECTS_PER_SCHEMA)
1778 }
1779
1780 pub fn max_secrets(&self) -> u32 {
1782 *self.expect_value(&MAX_SECRETS)
1783 }
1784
1785 pub fn max_roles(&self) -> u32 {
1787 *self.expect_value(&MAX_ROLES)
1788 }
1789
1790 pub fn max_network_policies(&self) -> u32 {
1792 *self.expect_value(&MAX_NETWORK_POLICIES)
1793 }
1794
1795 pub fn max_rules_per_network_policy(&self) -> u32 {
1797 *self.expect_value(&MAX_RULES_PER_NETWORK_POLICY)
1798 }
1799
1800 pub fn max_result_size(&self) -> u64 {
1802 self.expect_value::<ByteSize>(&MAX_RESULT_SIZE).as_bytes()
1803 }
1804
1805 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 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 pub fn max_concurrent_occ_writes(&self) -> u32 {
1821 *self.expect_value(&MAX_CONCURRENT_OCC_WRITES)
1822 }
1823
1824 pub fn max_occ_retries(&self) -> u32 {
1826 *self.expect_value(&MAX_OCC_RETRIES)
1827 }
1828
1829 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 pub fn pg_source_connect_timeout(&self) -> Duration {
1908 *self.expect_value(&PG_SOURCE_CONNECT_TIMEOUT)
1909 }
1910
1911 pub fn pg_source_tcp_keepalives_retries(&self) -> u32 {
1913 *self.expect_value(&PG_SOURCE_TCP_KEEPALIVES_RETRIES)
1914 }
1915
1916 pub fn pg_source_tcp_keepalives_idle(&self) -> Duration {
1918 *self.expect_value(&PG_SOURCE_TCP_KEEPALIVES_IDLE)
1919 }
1920
1921 pub fn pg_source_tcp_keepalives_interval(&self) -> Duration {
1923 *self.expect_value(&PG_SOURCE_TCP_KEEPALIVES_INTERVAL)
1924 }
1925
1926 pub fn pg_source_tcp_user_timeout(&self) -> Duration {
1928 *self.expect_value(&PG_SOURCE_TCP_USER_TIMEOUT)
1929 }
1930
1931 pub fn pg_source_tcp_configure_server(&self) -> bool {
1933 *self.expect_value(&PG_SOURCE_TCP_CONFIGURE_SERVER)
1934 }
1935
1936 pub fn pg_source_snapshot_statement_timeout(&self) -> Duration {
1938 *self.expect_value(&PG_SOURCE_SNAPSHOT_STATEMENT_TIMEOUT)
1939 }
1940
1941 pub fn pg_source_wal_sender_timeout(&self) -> Option<Duration> {
1943 *self.expect_value(&PG_SOURCE_WAL_SENDER_TIMEOUT)
1944 }
1945
1946 pub fn pg_source_snapshot_collect_strict_count(&self) -> bool {
1948 *self.expect_value(&PG_SOURCE_SNAPSHOT_COLLECT_STRICT_COUNT)
1949 }
1950
1951 pub fn mysql_source_tcp_keepalive(&self) -> Duration {
1953 *self.expect_value(&MYSQL_SOURCE_TCP_KEEPALIVE)
1954 }
1955
1956 pub fn mysql_source_snapshot_max_execution_time(&self) -> Duration {
1958 *self.expect_value(&MYSQL_SOURCE_SNAPSHOT_MAX_EXECUTION_TIME)
1959 }
1960
1961 pub fn mysql_source_snapshot_lock_wait_timeout(&self) -> Duration {
1963 *self.expect_value(&MYSQL_SOURCE_SNAPSHOT_LOCK_WAIT_TIMEOUT)
1964 }
1965
1966 pub fn mysql_source_snapshot_wait_timeout(&self) -> Duration {
1968 *self.expect_value(&MYSQL_SOURCE_SNAPSHOT_WAIT_TIMEOUT)
1969 }
1970
1971 pub fn mysql_source_connect_timeout(&self) -> Duration {
1973 *self.expect_value(&MYSQL_SOURCE_CONNECT_TIMEOUT)
1974 }
1975
1976 pub fn ssh_check_interval(&self) -> Duration {
1978 *self.expect_value(&SSH_CHECK_INTERVAL)
1979 }
1980
1981 pub fn ssh_connect_timeout(&self) -> Duration {
1983 *self.expect_value(&SSH_CONNECT_TIMEOUT)
1984 }
1985
1986 pub fn ssh_keepalives_idle(&self) -> Duration {
1988 *self.expect_value(&SSH_KEEPALIVES_IDLE)
1989 }
1990
1991 pub fn kafka_socket_keepalive(&self) -> bool {
1993 *self.expect_value(&KAFKA_SOCKET_KEEPALIVE)
1994 }
1995
1996 pub fn kafka_socket_timeout(&self) -> Option<Duration> {
1998 *self.expect_value(&KAFKA_SOCKET_TIMEOUT)
1999 }
2000
2001 pub fn kafka_transaction_timeout(&self) -> Duration {
2003 *self.expect_value(&KAFKA_TRANSACTION_TIMEOUT)
2004 }
2005
2006 pub fn kafka_socket_connection_setup_timeout(&self) -> Duration {
2008 *self.expect_value(&KAFKA_SOCKET_CONNECTION_SETUP_TIMEOUT)
2009 }
2010
2011 pub fn kafka_fetch_metadata_timeout(&self) -> Duration {
2013 *self.expect_value(&KAFKA_FETCH_METADATA_TIMEOUT)
2014 }
2015
2016 pub fn kafka_progress_record_fetch_timeout(&self) -> Option<Duration> {
2018 *self.expect_value(&KAFKA_PROGRESS_RECORD_FETCH_TIMEOUT)
2019 }
2020
2021 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 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 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 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 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 pub fn storage_dataflow_max_inflight_bytes(&self) -> Option<usize> {
2058 *self.expect_value(&STORAGE_DATAFLOW_MAX_INFLIGHT_BYTES)
2059 }
2060
2061 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 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 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 pub fn storage_statistics_interval(&self) -> Duration {
2078 *self.expect_value(&STORAGE_STATISTICS_INTERVAL)
2079 }
2080
2081 pub fn storage_statistics_collection_interval(&self) -> Duration {
2083 *self.expect_value(&STORAGE_STATISTICS_COLLECTION_INTERVAL)
2084 }
2085
2086 pub fn storage_record_source_sink_namespaced_errors(&self) -> bool {
2088 *self.expect_value(&STORAGE_RECORD_SOURCE_SINK_NAMESPACED_ERRORS)
2089 }
2090
2091 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 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 pub fn sync_dyncfgs(&self) -> ConfigUpdates {
2144 let updates = self.dyncfg_updates();
2145 updates.apply(&self.dyncfgs);
2146 updates
2147 }
2148
2149 pub fn metrics_retention(&self) -> Duration {
2151 *self.expect_value(&METRICS_RETENTION)
2152 }
2153
2154 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 pub fn unsafe_mock_audit_event_timestamp(&self) -> Option<mz_repr::Timestamp> {
2164 *self.expect_value(&UNSAFE_MOCK_AUDIT_EVENT_TIMESTAMP)
2165 }
2166
2167 pub fn enable_rbac_checks(&self) -> bool {
2169 *self.expect_value(&ENABLE_RBAC_CHECKS)
2170 }
2171
2172 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 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 pub fn enable_storage_shard_finalization(&self) -> bool {
2204 *self.expect_value(&ENABLE_STORAGE_SHARD_FINALIZATION)
2205 }
2206
2207 pub fn enable_default_connection_validation(&self) -> bool {
2209 *self.expect_value(&ENABLE_DEFAULT_CONNECTION_VALIDATION)
2210 }
2211
2212 pub fn default_timestamp_interval(&self) -> Duration {
2214 *self.expect_value(&DEFAULT_TIMESTAMP_INTERVAL)
2215 }
2216
2217 pub fn min_timestamp_interval(&self) -> Duration {
2219 *self.expect_value(&MIN_TIMESTAMP_INTERVAL)
2220 }
2221 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 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 pub fn statement_logging_max_sample_rate(&self) -> Numeric {
2338 *self.expect_value(&STATEMENT_LOGGING_MAX_SAMPLE_RATE)
2339 }
2340
2341 pub fn statement_logging_default_sample_rate(&self) -> Numeric {
2343 *self.expect_value(&STATEMENT_LOGGING_DEFAULT_SAMPLE_RATE)
2344 }
2345
2346 pub fn enable_internal_statement_logging(&self) -> bool {
2348 *self.expect_value(&ENABLE_INTERNAL_STATEMENT_LOGGING)
2349 }
2350
2351 pub fn enable_statement_arrival_logging(&self) -> bool {
2353 *self.expect_value(&ENABLE_STATEMENT_ARRIVAL_LOGGING)
2354 }
2355
2356 pub fn enable_extended_protocol_implicit_transaction(&self) -> bool {
2359 *self.expect_value(&ENABLE_EXTENDED_PROTOCOL_IMPLICIT_TRANSACTION)
2360 }
2361
2362 pub fn optimizer_stats_timeout(&self) -> Duration {
2364 *self.expect_value(&OPTIMIZER_STATS_TIMEOUT)
2365 }
2366
2367 pub fn optimizer_oneshot_stats_timeout(&self) -> Duration {
2369 *self.expect_value(&OPTIMIZER_ONESHOT_STATS_TIMEOUT)
2370 }
2371
2372 pub fn webhook_concurrent_request_limit(&self) -> usize {
2374 *self.expect_value(&WEBHOOK_CONCURRENT_REQUEST_LIMIT)
2375 }
2376
2377 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 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 pub fn pg_timestamp_oracle_connection_pool_ttl(&self) -> Duration {
2389 *self.expect_value(&PG_TIMESTAMP_ORACLE_CONNECTION_POOL_TTL)
2390 }
2391
2392 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 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 pub fn is_controller_config_var(&self, name: &str) -> bool {
2412 self.is_dyncfg_var(name)
2413 }
2414
2415 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 pub fn is_metrics_config_var(&self, name: &str) -> bool {
2424 self.is_dyncfg_var(name)
2425 }
2426
2427 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 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
2480pub 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
2500pub 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
2515pub 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
2529pub fn is_http_config_var(name: &str) -> bool {
2531 name == WEBHOOK_CONCURRENT_REQUEST_LIMIT.name()
2532}
2533
2534static 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#[derive(Debug)]
2569pub struct FeatureFlag {
2570 pub flag: &'static VarDefinition,
2571 pub feature_desc: &'static str,
2572}
2573
2574impl FeatureFlag {
2575 pub fn enabled(&'static self, system_vars: &SystemVars) -> bool {
2577 *system_vars.expect_value::<bool>(self.flag)
2578 }
2579
2580 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 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 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 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 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 #[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 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 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 #[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}