adapter: rename controller_commands to catalog_implications

Better describes what it does, because the implications are not just for
the controllers.

Author: aljoscha

src/adapter/src/catalog.rs

@@ -98,7 +98,7 @@ pub use crate::catalog::transact::{
 use crate::command::CatalogDump;
 use crate::coord::TargetCluster;
 #[cfg(test)]
-use crate::coord::controller_commands::parsed_state_updates::ParsedStateUpdate;
+use crate::coord::catalog_implications::parsed_state_updates::ParsedStateUpdate;
 use crate::session::{Portal, PreparedStatement, Session};
 use crate::util::ResultExt;
 use crate::{AdapterError, AdapterNotice, ExecuteResponse};
@@ -1519,9 +1519,8 @@ impl Catalog {
         CatalogError,
     > {
         let updates = self.storage().await.sync_to_current_updates().await?;
-        let (builtin_table_updates, controller_state_updates) =
-            self.state.apply_updates(updates)?;
-        Ok((builtin_table_updates, controller_state_updates))
+        let (builtin_table_updates, catalog_updates) = self.state.apply_updates(updates)?;
+        Ok((builtin_table_updates, catalog_updates))
     }
 }
 

src/adapter/src/catalog/apply.rs

@@ -62,7 +62,7 @@ use tracing::{Instrument, info_span, warn};
 use crate::AdapterError;
 use crate::catalog::state::LocalExpressionCache;
 use crate::catalog::{BuiltinTableUpdate, CatalogState};
-use crate::coord::controller_commands::parsed_state_updates::{self, ParsedStateUpdate};
+use crate::coord::catalog_implications::parsed_state_updates::{self, ParsedStateUpdate};
 use crate::util::index_sql;
 
 /// Maintains the state of retractions while applying catalog state updates for a single timestamp.
@@ -108,7 +108,7 @@ impl CatalogState {
         Vec<ParsedStateUpdate>,
     ) {
         let mut builtin_table_updates = Vec::with_capacity(updates.len());
-        let mut controller_state_updates = Vec::with_capacity(updates.len());
+        let mut catalog_updates = Vec::with_capacity(updates.len());
         let updates = sort_updates(updates);
 
         let mut groups: Vec<Vec<_>> = Vec::new();
@@ -121,29 +121,28 @@ impl CatalogState {
 
             for update in updates {
                 let next_apply_state = BootstrapApplyState::new(update);
-                let (next_apply_state, (builtin_table_update, controller_state_update)) =
-                    apply_state
-                        .step(
-                            next_apply_state,
-                            self,
-                            &mut retractions,
-                            local_expression_cache,
-                        )
-                        .await;
+                let (next_apply_state, (builtin_table_update, catalog_update)) = apply_state
+                    .step(
+                        next_apply_state,
+                        self,
+                        &mut retractions,
+                        local_expression_cache,
+                    )
+                    .await;
                 apply_state = next_apply_state;
                 builtin_table_updates.extend(builtin_table_update);
-                controller_state_updates.extend(controller_state_update);
+                catalog_updates.extend(catalog_update);
             }
 
             // Apply remaining state.
-            let (builtin_table_update, controller_state_update) = apply_state
+            let (builtin_table_update, catalog_update) = apply_state
                 .apply(self, &mut retractions, local_expression_cache)
                 .await;
             builtin_table_updates.extend(builtin_table_update);
-            controller_state_updates.extend(controller_state_update);
+            catalog_updates.extend(catalog_update);
         }
 
-        (builtin_table_updates, controller_state_updates)
+        (builtin_table_updates, catalog_updates)
     }
 
     /// Update in-memory catalog state from a list of updates made to the durable catalog state.
@@ -161,7 +160,7 @@ impl CatalogState {
         CatalogError,
     > {
         let mut builtin_table_updates = Vec::with_capacity(updates.len());
-        let mut controller_state_updates = Vec::with_capacity(updates.len());
+        let mut catalog_updates = Vec::with_capacity(updates.len());
 
         // First, consolidate updates. The code that applies parsed state
         // updates _requires_ that the given updates are consolidated. There
@@ -175,16 +174,16 @@ impl CatalogState {
 
         for (_, updates) in &updates.into_iter().chunk_by(|update| update.ts) {
             let mut retractions = InProgressRetractions::default();
-            let (builtin_table_update, parsed_catalog_updates_op) = self.apply_updates_inner(
+            let (builtin_table_update, catalog_updates_op) = self.apply_updates_inner(
                 updates.collect(),
                 &mut retractions,
                 &mut LocalExpressionCache::Closed,
             )?;
             builtin_table_updates.extend(builtin_table_update);
-            controller_state_updates.extend(parsed_catalog_updates_op);
+            catalog_updates.extend(catalog_updates_op);
         }
 
-        Ok((builtin_table_updates, controller_state_updates))
+        Ok((builtin_table_updates, catalog_updates))
     }
 
     /// It can happen that the sequencing logic creates "fluctuating" updates
@@ -236,7 +235,7 @@ impl CatalogState {
         let mut update_system_config = false;
 
         let mut builtin_table_updates = Vec::with_capacity(updates.len());
-        let mut controller_state_updates = Vec::new();
+        let mut catalog_updates = Vec::new();
 
         for state_update in updates {
             if matches!(state_update.kind, StateUpdateKind::SystemConfiguration(_)) {
@@ -245,14 +244,13 @@ impl CatalogState {
 
             match state_update.diff {
                 StateDiff::Retraction => {
-                    // We want the parsed controller state updates to match the
-                    // state of the catalog _before_ applying a retraction. So
-                    // that we can still have useful in-memory state to work
-                    // with.
+                    // We want the parsed catalog updates to match the state of
+                    // the catalog _before_ applying a retraction. So that we can
+                    // still have useful in-memory state to work with.
                     if let Some(update) =
                         parsed_state_updates::parse_state_update(self, state_update.clone())
                     {
-                        controller_state_updates.push(update);
+                        catalog_updates.push(update);
                     }
 
                     // We want the builtin table retraction to match the state of the catalog
@@ -283,12 +281,12 @@ impl CatalogState {
                         state_update.diff,
                     ));
 
-                    // We want the parsed controller state updates to match the
-                    // state of the catalog _after_ applying an addition.
+                    // We want the parsed catalog updates to match the state of
+                    // the catalog _after_ applying an addition.
                     if let Some(update) =
                         parsed_state_updates::parse_state_update(self, state_update.clone())
                     {
-                        controller_state_updates.push(update);
+                        catalog_updates.push(update);
                     }
                 }
             }
@@ -298,7 +296,7 @@ impl CatalogState {
             self.system_configuration.dyncfg_updates();
         }
 
-        Ok((builtin_table_updates, controller_state_updates))
+        Ok((builtin_table_updates, catalog_updates))
     }
 
     #[instrument(level = "debug")]

src/adapter/src/catalog/migrate.rs

@@ -41,7 +41,7 @@ use uuid::Uuid;
 use crate::catalog::open::into_consolidatable_updates_startup;
 use crate::catalog::state::LocalExpressionCache;
 use crate::catalog::{BuiltinTableUpdate, CatalogState, ConnCatalog};
-use crate::coord::controller_commands::parsed_state_updates::ParsedStateUpdate;
+use crate::coord::catalog_implications::parsed_state_updates::ParsedStateUpdate;
 
 /// Catalog key of the `migration_version` setting.
 ///
@@ -118,7 +118,7 @@ where
 
 pub(crate) struct MigrateResult {
     pub(crate) builtin_table_updates: Vec<BuiltinTableUpdate<&'static BuiltinTable>>,
-    pub(crate) controller_state_updates: Vec<ParsedStateUpdate>,
+    pub(crate) catalog_updates: Vec<ParsedStateUpdate>,
     pub(crate) post_item_updates: Vec<(BootstrapStateUpdateKind, Timestamp, Diff)>,
 }
 
@@ -194,7 +194,7 @@ pub(crate) async fn migrate(
             .expect("known parameter");
     }
 
-    let (mut ast_builtin_table_updates, mut ast_controller_state_updates) = state
+    let (mut ast_builtin_table_updates, mut ast_catalog_updates) = state
         .apply_updates_for_bootstrap(item_updates, local_expr_cache)
         .await;
 
@@ -240,12 +240,12 @@ pub(crate) async fn migrate(
     // input and stages arbitrary transformations to the catalog on `tx`.
 
     let op_item_updates = tx.get_and_commit_op_updates();
-    let (item_builtin_table_updates, item_controller_state_updates) = state
+    let (item_builtin_table_updates, item_catalog_updates) = state
         .apply_updates_for_bootstrap(op_item_updates, local_expr_cache)
         .await;
 
     ast_builtin_table_updates.extend(item_builtin_table_updates);
-    ast_controller_state_updates.extend(item_controller_state_updates);
+    ast_catalog_updates.extend(item_catalog_updates);
 
     info!(
         "migration from catalog version {:?} complete",
@@ -254,7 +254,7 @@ pub(crate) async fn migrate(
 
     Ok(MigrateResult {
         builtin_table_updates: ast_builtin_table_updates,
-        controller_state_updates: ast_controller_state_updates,
+        catalog_updates: ast_catalog_updates,
         post_item_updates,
     })
 }

src/adapter/src/catalog/open.rs

@@ -313,7 +313,7 @@ impl Catalog {
             }
         }
 
-        let (builtin_table_update, _controller_state_updates) = state
+        let (builtin_table_update, _catalog_updates) = state
             .apply_updates_for_bootstrap(pre_item_updates, &mut LocalExpressionCache::Closed)
             .await;
         builtin_table_updates.extend(builtin_table_update);
@@ -410,12 +410,12 @@ impl Catalog {
             expr_cache_start.elapsed()
         );
 
-        // When initializing/bootstrapping, we don't use the controller state
-        // updates but instead load the catalog fully and then go ahead and
-        // apply commands to the controller(s). Maybe we _should_ instead use
-        // the same logic and return and use the updates from here. But that's
-        // at the very least future work.
-        let (builtin_table_update, _controller_state_updates) = state
+        // When initializing/bootstrapping, we don't use the catalog updates but
+        // instead load the catalog fully and then go ahead and apply commands
+        // to the controller(s). Maybe we _should_ instead use the same logic
+        // and return and use the updates from here. But that's at the very
+        // least future work.
+        let (builtin_table_update, _catalog_updates) = state
             .apply_updates_for_bootstrap(system_item_updates, &mut local_expr_cache)
             .await;
         builtin_table_updates.extend(builtin_table_update);
@@ -431,7 +431,7 @@ impl Catalog {
         state.mock_authentication_nonce = Some(mz_authentication_mock_nonce);
 
         // Migrate item ASTs.
-        let (builtin_table_update, _controller_state_updates) = if !config.skip_migrations {
+        let (builtin_table_update, _catalog_updates) = if !config.skip_migrations {
             let migrate_result = migrate::migrate(
                 &mut state,
                 &mut txn,
@@ -463,7 +463,7 @@ impl Catalog {
 
             (
                 migrate_result.builtin_table_updates,
-                migrate_result.controller_state_updates,
+                migrate_result.catalog_updates,
             )
         } else {
             state
@@ -480,7 +480,7 @@ impl Catalog {
                 diff: diff.try_into().expect("valid diff"),
             })
             .collect();
-        let (builtin_table_update, _controller_state_updates) = state
+        let (builtin_table_update, _catalog_updates) = state
             .apply_updates_for_bootstrap(post_item_updates, &mut local_expr_cache)
             .await;
         builtin_table_updates.extend(builtin_table_update);
@@ -505,12 +505,12 @@ impl Catalog {
 
         let state_updates = txn.get_and_commit_op_updates();
 
-        // When initializing/bootstrapping, we don't use the controller state
-        // updates but instead load the catalog fully and then go ahead and
-        // apply commands to the controller(s). Maybe we _should_ instead use
-        // the same logic and return and use the updates from here. But that's
-        // at the very least future work.
-        let (table_updates, _controller_state_updates) = state
+        // When initializing/bootstrapping, we don't use the catalog updates but
+        // instead load the catalog fully and then go ahead and apply commands
+        // to the controller(s). Maybe we _should_ instead use the same logic
+        // and return and use the updates from here. But that's at the very
+        // least future work.
+        let (table_updates, _catalog_updates) = state
             .apply_updates_for_bootstrap(state_updates, &mut local_expr_cache)
             .await;
         builtin_table_updates.extend(table_updates);
@@ -656,13 +656,13 @@ impl Catalog {
             .map_err(mz_catalog::durable::DurableCatalogError::from)?;
 
         let updates = txn.get_and_commit_op_updates();
-        let (builtin_updates, controller_state_updates) = state.apply_updates(updates)?;
+        let (builtin_updates, catalog_updates) = state.apply_updates(updates)?;
         assert!(
             builtin_updates.is_empty(),
             "storage is not allowed to generate catalog changes that would cause changes to builtin tables"
         );
         assert!(
-            controller_state_updates.is_empty(),
+            catalog_updates.is_empty(),
             "storage is not allowed to generate catalog changes that would change the catalog or controller state"
         );
         let commit_ts = txn.upper();