sql: Emit notice when Role or System default changes for a variable (M…

…aterializeInc#24075)

### Motivation

This PR adds a feature that has not yet been specified.

Currently when running `ALTER ROLE ... SET ...` or `ALTER SYSTEM SET
...` you won't observe the new value for the variable until you start a
new session. This is what Postgres does, but it could be a bit confusing
to users. This PR adds a new notice that gets emitted whenever a user
changes a variable, to let them know it won't take effect until they
start a new session. For example:

```
materialize=> ALTER ROLE parker SET cluster TO foo;
NOTICE:  variable "cluster" was updated for "parker", please reconnect to take effect
ALTER ROLE
```

### Checklist

- [ ] This PR has adequate test coverage / QA involvement has been duly
considered.
- [ ] This PR has an associated up-to-date [design
doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md),
is a design doc
([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)),
or is sufficiently small to not require a design.
  <!-- Reference the design in the description. -->
- [ ] If this PR evolves [an existing `$T ⇔ Proto$T`
mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md)
(possibly in a backwards-incompatible way), then it is tagged with a
`T-proto` label.
- [ ] If this PR will require changes to cloud orchestration or tests,
there is a companion cloud PR to account for those changes that is
tagged with the release-blocker label
([example](MaterializeInc/cloud#5021)).
<!-- Ask in #team-cloud on Slack if you need help preparing the cloud
PR. -->
- [x] This PR includes the following [user-facing behavior
changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note):
  - Adds a database notice for whenever users update a variable default

src/adapter/src/coord/sequencer.rs

@@ -166,8 +166,8 @@ impl Coordinator {
                     let result = self
                         .sequence_create_role(Some(ctx.session().conn_id()), plan)
                         .await;
-                    if result.is_ok() {
-                        self.maybe_send_rbac_notice(ctx.session());
+                    if let Some(notice) = self.should_emit_rbac_notice(ctx.session()) {
+                        ctx.session().add_notice(notice);
                     }
                     ctx.retire(result);
                 }
@@ -398,9 +398,6 @@ impl Coordinator {
                 }
                 Plan::AlterRole(plan) => {
                     let result = self.sequence_alter_role(ctx.session_mut(), plan).await;
-                    if result.is_ok() {
-                        self.maybe_send_rbac_notice(ctx.session());
-                    }
                     ctx.retire(result);
                 }
                 Plan::AlterSecret(plan) => {
@@ -706,9 +703,11 @@ impl Coordinator {
         Ok(GlobalId::Transient(id))
     }
 
-    fn maybe_send_rbac_notice(&self, session: &Session) {
+    fn should_emit_rbac_notice(&self, session: &Session) -> Option<AdapterNotice> {
         if !rbac::is_rbac_enabled_for_session(self.catalog.system_config(), session.vars()) {
-            session.add_notice(AdapterNotice::RbacUserDisabled);
+            Some(AdapterNotice::RbacUserDisabled)
+        } else {
+            None
         }
     }
 

src/adapter/src/coord/sequencer/inner.rs

@@ -4083,6 +4083,9 @@ impl Coordinator {
         let catalog = self.catalog().for_session(session);
         let role = catalog.get_role(&id);
 
+        // We'll send these notices to the user, if the operation is successful.
+        let mut notices = vec![];
+
         // Get the attributes and variables from the role, as they currently are.
         let mut attributes = role.attributes().clone();
         let mut vars = role.vars().clone();
@@ -4093,14 +4096,18 @@ impl Coordinator {
                 if let Some(inherit) = attrs.inherit {
                     attributes.inherit = inherit;
                 }
+
+                if let Some(notice) = self.should_emit_rbac_notice(session) {
+                    notices.push(notice);
+                }
             }
             PlannedAlterRoleOption::Variable(variable) => {
                 // Get the variable to make sure it's valid and visible.
                 session
                     .vars()
                     .get(Some(catalog.system_vars()), variable.name())?;
 
-                match variable {
+                let var_name = match variable {
                     PlannedRoleVariable::Set { name, value } => {
                         // Update our persisted set.
                         match &value {
@@ -4115,12 +4122,20 @@ impl Coordinator {
                                 vars.insert(name.clone(), var);
                             }
                         };
+                        name
                     }
                     PlannedRoleVariable::Reset { name } => {
                         // Remove it from our persisted values.
                         vars.remove(&name);
+                        name
                     }
-                }
+                };
+
+                // Emit a notice that they need to reconnect to see the change take effect.
+                notices.push(AdapterNotice::VarDefaultUpdated {
+                    role: Some(name.clone()),
+                    var_name: Some(var_name),
+                })
             }
         }
 
@@ -4130,9 +4145,15 @@ impl Coordinator {
             attributes,
             vars: RoleVars { map: vars },
         };
-        self.catalog_transact(Some(session), vec![op])
+        let response = self
+            .catalog_transact(Some(session), vec![op])
             .await
-            .map(|_| ExecuteResponse::AlteredRole)
+            .map(|_| ExecuteResponse::AlteredRole)?;
+
+        // Send all of our queued notices.
+        session.add_notices(notices);
+
+        Ok(response)
     }
 
     pub(super) async fn sequence_alter_secret(
@@ -5047,12 +5068,19 @@ impl Coordinator {
         self.is_user_allowed_to_alter_system(session, Some(&name))?;
         let op = match value {
             plan::VariableValue::Values(values) => catalog::Op::UpdateSystemConfiguration {
-                name,
+                name: name.clone(),
                 value: OwnedVarInput::SqlSet(values),
             },
-            plan::VariableValue::Default => catalog::Op::ResetSystemConfiguration { name },
+            plan::VariableValue::Default => {
+                catalog::Op::ResetSystemConfiguration { name: name.clone() }
+            }
         };
         self.catalog_transact(Some(session), vec![op]).await?;
+
+        session.add_notice(AdapterNotice::VarDefaultUpdated {
+            role: None,
+            var_name: Some(name),
+        });
         Ok(ExecuteResponse::AlteredSystemConfiguration)
     }
 
@@ -5062,8 +5090,12 @@ impl Coordinator {
         plan::AlterSystemResetPlan { name }: plan::AlterSystemResetPlan,
     ) -> Result<ExecuteResponse, AdapterError> {
         self.is_user_allowed_to_alter_system(session, Some(&name))?;
-        let op = catalog::Op::ResetSystemConfiguration { name };
+        let op = catalog::Op::ResetSystemConfiguration { name: name.clone() };
         self.catalog_transact(Some(session), vec![op]).await?;
+        session.add_notice(AdapterNotice::VarDefaultUpdated {
+            role: None,
+            var_name: Some(name),
+        });
         Ok(ExecuteResponse::AlteredSystemConfiguration)
     }
 
@@ -5075,6 +5107,10 @@ impl Coordinator {
         self.is_user_allowed_to_alter_system(session, None)?;
         let op = catalog::Op::ResetAllSystemConfiguration;
         self.catalog_transact(Some(session), vec![op]).await?;
+        session.add_notice(AdapterNotice::VarDefaultUpdated {
+            role: None,
+            var_name: None,
+        });
         Ok(ExecuteResponse::AlteredSystemConfiguration)
     }
 

src/adapter/src/notice.rs

@@ -125,6 +125,10 @@ pub enum AdapterNotice {
     PerReplicaLogRead {
         log_names: Vec<String>,
     },
+    VarDefaultUpdated {
+        role: Option<String>,
+        var_name: Option<String>,
+    },
     Welcome(String),
 }
 
@@ -185,6 +189,7 @@ impl AdapterNotice {
             AdapterNotice::WebhookSourceCreated { .. } => Severity::Notice,
             AdapterNotice::DroppedInUseIndex { .. } => Severity::Notice,
             AdapterNotice::PerReplicaLogRead { .. } => Severity::Notice,
+            AdapterNotice::VarDefaultUpdated { .. } => Severity::Notice,
             AdapterNotice::Welcome(_) => Severity::Notice,
         }
     }
@@ -274,6 +279,7 @@ impl AdapterNotice {
             AdapterNotice::DroppedInUseIndex { .. } => SqlState::WARNING,
             AdapterNotice::WebhookSourceCreated { .. } => SqlState::WARNING,
             AdapterNotice::PerReplicaLogRead { .. } => SqlState::WARNING,
+            AdapterNotice::VarDefaultUpdated { .. } => SqlState::SUCCESSFUL_COMPLETION,
             AdapterNotice::Welcome(_) => SqlState::SUCCESSFUL_COMPLETION,
         }
     }
@@ -425,6 +431,20 @@ impl fmt::Display for AdapterNotice {
             AdapterNotice::PerReplicaLogRead { log_names } => {
                 write!(f, "Queried introspection relations: {}. Unlike other objects in Materialize, results from querying these objects depend on the current values of the `cluster` and `cluster_replica` session variables.", log_names.join(", "))
             }
+            AdapterNotice::VarDefaultUpdated { role, var_name } => {
+                let vars = match var_name {
+                    Some(name) => format!("variable {} was", name.quoted()),
+                    None => "variables were".to_string(),
+                };
+                let target = match role {
+                    Some(role_name) => role_name.quoted().to_string(),
+                    None => "the system".to_string(),
+                };
+                write!(
+                    f,
+                    "{vars} updated for {target}, this will have no effect on the current session"
+                )
+            }
             AdapterNotice::Welcome(message) => message.fmt(f),
         }
     }