reload system config after migrations, responded to GitHub feedback

MaterializeInc · Jan 2, 2024 · 236d460 · 236d460
1 parent 53c96a0
commit 236d460
Show file tree

Hide file tree

Showing 4 changed files with 26 additions and 11 deletions.
diff --git a/src/adapter/src/catalog/open.rs b/src/adapter/src/catalog/open.rs
@@ -417,10 +417,9 @@ impl Catalog {
             Catalog::load_system_configuration(
                 &mut state,
                 &mut txn,
-                config.system_parameter_defaults,
-                config.remote_system_parameters,
-            )
-            .await?;
+                &config.system_parameter_defaults,
+                config.remote_system_parameters.as_ref(),
+            )?;
 
             // Now that LD is loaded, set the intended catalog timeout.
             // TODO: Move this into the catalog constructor.
@@ -856,6 +855,14 @@ impl Catalog {
                 txn.set_catalog_content_version(config.build_info.version.to_string())?;
             }
 
+            // Re-load the system configuration in case it changed after the migrations.
+            Catalog::load_system_configuration(
+                &mut state,
+                &mut txn,
+                &config.system_parameter_defaults,
+                config.remote_system_parameters.as_ref(),
+            )?;
+
             let mut state = Catalog::load_catalog_items(&mut txn, &state)?;
 
             let mut builtin_migration_metadata = Catalog::generate_builtin_migration_metadata(
@@ -1108,15 +1115,15 @@ impl Catalog {
     ///
     /// # Errors
     #[tracing::instrument(level = "info", skip_all)]
-    async fn load_system_configuration(
+    fn load_system_configuration(
         state: &mut CatalogState,
         txn: &mut Transaction<'_>,
-        system_parameter_defaults: BTreeMap<String, String>,
-        remote_system_parameters: Option<BTreeMap<String, OwnedVarInput>>,
+        system_parameter_defaults: &BTreeMap<String, String>,
+        remote_system_parameters: Option<&BTreeMap<String, OwnedVarInput>>,
     ) -> Result<(), AdapterError> {
         let system_config = txn.get_system_configurations();
 
-        for (name, value) in &system_parameter_defaults {
+        for (name, value) in system_parameter_defaults {
             match state.set_system_configuration_default(name, VarInput::Flat(value)) {
                 Ok(_) => (),
                 Err(Error {
@@ -1140,7 +1147,7 @@ impl Catalog {
         }
         if let Some(remote_system_parameters) = remote_system_parameters {
             for (name, value) in remote_system_parameters {
-                Catalog::update_system_configuration(state, txn, &name, value.borrow())?;
+                Catalog::update_system_configuration(state, txn, name, value.borrow())?;
             }
             txn.set_system_config_synced_once()?;
         }

diff --git a/src/adapter/src/catalog/state.rs b/src/adapter/src/catalog/state.rs
@@ -216,11 +216,13 @@ impl CatalogState {
 
     pub fn for_sessionless_user(&self, role_id: RoleId) -> ConnCatalog {
         let (notices_tx, _notices_rx) = mpsc::unbounded_channel();
+        let cluster = self.system_configuration.default_cluster();
+
         ConnCatalog {
             state: Cow::Borrowed(self),
             unresolvable_ids: BTreeSet::new(),
             conn_id: SYSTEM_CONN_ID.clone(),
-            cluster: "quickstart".into(),
+            cluster,
             database: self
                 .resolve_database(DEFAULT_DATABASE_NAME)
                 .ok()

diff --git a/src/sql/src/session/vars.rs b/src/sql/src/session/vars.rs
@@ -3253,6 +3253,12 @@ impl SystemVars {
         self.propagate_var_change(MAX_CONNECTIONS.name.as_str());
     }
 
+    /// Returns the system default for the [`CLUSTER`] session variable. To know the active cluster
+    /// for the current session, you must check the [`SessionVars`].
+    pub fn default_cluster(&self) -> String {
+        self.expect_value(&CLUSTER).to_owned()
+    }
+
     /// Returns the value of the `max_kafka_connections` configuration parameter.
     pub fn max_kafka_connections(&self) -> u32 {
         *self.expect_value(&MAX_KAFKA_CONNECTIONS)

diff --git a/test/cluster/query-without-default-cluster/query-without-default-cluster.td b/test/cluster/query-without-default-cluster/query-without-default-cluster.td
@@ -28,6 +28,6 @@ contains:unknown cluster 'quickstart'
 ! SELECT generate_series(0, 10)
 contains:unknown cluster 'quickstart'
 
-# to make it easier to run the test again, let's recreate the defaquickstartult cluster
+# to make it easier to run the test again, let's recreate the quickstart cluster
 $ postgres-execute connection=mz_system
 CREATE CLUSTER quickstart REPLICAS ()