Merge pull request #29673 from aljoscha/adapter-synthesize-controller-commands

adapter: derive implications from catalog changes

Author: aljoscha

src/adapter/src/catalog.rs

@@ -97,6 +97,8 @@ pub use crate::catalog::transact::{
 };
 use crate::command::CatalogDump;
 use crate::coord::TargetCluster;
+#[cfg(test)]
+use crate::coord::catalog_implications::parsed_state_updates::ParsedStateUpdate;
 use crate::session::{Portal, PreparedStatement, Session};
 use crate::util::ResultExt;
 use crate::{AdapterError, AdapterNotice, ExecuteResponse};
@@ -1509,10 +1511,16 @@ impl Catalog {
     #[cfg(test)]
     async fn sync_to_current_updates(
         &mut self,
-    ) -> Result<Vec<BuiltinTableUpdate<&'static BuiltinTable>>, CatalogError> {
+    ) -> Result<
+        (
+            Vec<BuiltinTableUpdate<&'static BuiltinTable>>,
+            Vec<ParsedStateUpdate>,
+        ),
+        CatalogError,
+    > {
         let updates = self.storage().await.sync_to_current_updates().await?;
-        let builtin_table_updates = self.state.apply_updates(updates)?;
-        Ok(builtin_table_updates)
+        let (builtin_table_updates, catalog_updates) = self.state.apply_updates(updates)?;
+        Ok((builtin_table_updates, catalog_updates))
     }
 }
 

src/adapter/src/catalog/apply.rs

@@ -41,6 +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::catalog_implications::parsed_state_updates::ParsedStateUpdate;
 
 /// Catalog key of the `migration_version` setting.
 ///
@@ -117,6 +118,7 @@ where
 
 pub(crate) struct MigrateResult {
     pub(crate) builtin_table_updates: Vec<BuiltinTableUpdate<&'static BuiltinTable>>,
+    pub(crate) catalog_updates: Vec<ParsedStateUpdate>,
     pub(crate) post_item_updates: Vec<(BootstrapStateUpdateKind, Timestamp, Diff)>,
 }
 
@@ -192,7 +194,7 @@ pub(crate) async fn migrate(
             .expect("known parameter");
     }
 
-    let mut ast_builtin_table_updates = state
+    let (mut ast_builtin_table_updates, mut ast_catalog_updates) = state
         .apply_updates_for_bootstrap(item_updates, local_expr_cache)
         .await;
 
@@ -238,18 +240,21 @@ 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 = 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_catalog_updates.extend(item_catalog_updates);
 
     info!(
         "migration from catalog version {:?} complete",
         catalog_version
     );
+
     Ok(MigrateResult {
         builtin_table_updates: ast_builtin_table_updates,
+        catalog_updates: ast_catalog_updates,
         post_item_updates,
     })
 }

src/adapter/src/catalog/migrate.rs

@@ -313,7 +313,7 @@ impl Catalog {
             }
         }
 
-        let builtin_table_update = 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,7 +410,12 @@ impl Catalog {
             expr_cache_start.elapsed()
         );
 
-        let builtin_table_update = 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);
@@ -426,7 +431,7 @@ impl Catalog {
         state.mock_authentication_nonce = Some(mz_authentication_mock_nonce);
 
         // Migrate item ASTs.
-        let builtin_table_update = if !config.skip_migrations {
+        let (builtin_table_update, _catalog_updates) = if !config.skip_migrations {
             let migrate_result = migrate::migrate(
                 &mut state,
                 &mut txn,
@@ -456,7 +461,10 @@ impl Catalog {
                 differential_dataflow::consolidation::consolidate_updates(&mut post_item_updates);
             }
 
-            migrate_result.builtin_table_updates
+            (
+                migrate_result.builtin_table_updates,
+                migrate_result.catalog_updates,
+            )
         } else {
             state
                 .apply_updates_for_bootstrap(item_updates, &mut local_expr_cache)
@@ -472,7 +480,7 @@ impl Catalog {
                 diff: diff.try_into().expect("valid diff"),
             })
             .collect();
-        let builtin_table_update = 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);
@@ -496,7 +504,13 @@ impl Catalog {
         .await?;
 
         let state_updates = txn.get_and_commit_op_updates();
-        let table_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);
@@ -642,8 +656,15 @@ impl Catalog {
             .map_err(mz_catalog::durable::DurableCatalogError::from)?;
 
         let updates = txn.get_and_commit_op_updates();
-        let builtin_updates = state.apply_updates(updates)?;
-        assert!(builtin_updates.is_empty());
+        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!(
+            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();
         txn.commit(commit_ts).await?;
         drop(storage);

src/adapter/src/catalog/open.rs

@@ -1533,7 +1533,7 @@ impl CatalogState {
     /// Gets a reference to the specified replica of the specified cluster.
     ///
     /// Panics if either the cluster or the replica does not exist.
-    pub(super) fn get_cluster_replica(
+    pub(crate) fn get_cluster_replica(
         &self,
         cluster_id: ClusterId,
         replica_id: ReplicaId,
@@ -1629,6 +1629,13 @@ impl CatalogState {
             .into_first()
     }
 
+    pub(super) fn find_temp_schema(&self, schema_id: &SchemaId) -> &Schema {
+        self.temporary_schemas
+            .values()
+            .filter(|schema| schema.id() == &SchemaSpecifier::from(*schema_id))
+            .into_first()
+    }
+
     pub fn get_mz_catalog_schema_id(&self) -> SchemaId {
         self.ambient_schemas_by_name[MZ_CATALOG_SCHEMA]
     }

src/adapter/src/catalog/state.rs

@@ -70,6 +70,7 @@ use crate::catalog::{
     system_object_type_to_audit_object_type,
 };
 use crate::coord::ConnMeta;
+use crate::coord::catalog_implications::parsed_state_updates::ParsedStateUpdate;
 use crate::coord::cluster_scheduling::SchedulingDecision;
 use crate::util::ResultExt;
 
@@ -327,6 +328,8 @@ impl ReplicaCreateDropReason {
 
 pub struct TransactionResult {
     pub builtin_table_updates: Vec<BuiltinTableUpdate>,
+    /// Parsed catalog updates from which we will derive catalog implications.
+    pub catalog_updates: Vec<ParsedStateUpdate>,
     pub audit_events: Vec<VersionedEvent>,
 }
 
@@ -421,6 +424,7 @@ impl Catalog {
 
         let temporary_ids = self.temporary_ids(&ops, temporary_drops)?;
         let mut builtin_table_updates = vec![];
+        let mut catalog_updates = vec![];
         let mut audit_events = vec![];
         let mut storage = self.storage().await;
         let mut tx = storage
@@ -435,6 +439,7 @@ impl Catalog {
             ops,
             temporary_ids,
             &mut builtin_table_updates,
+            &mut catalog_updates,
             &mut audit_events,
             &mut tx,
             &self.state,
@@ -470,6 +475,7 @@ impl Catalog {
 
         Ok(TransactionResult {
             builtin_table_updates,
+            catalog_updates,
             audit_events,
         })
     }
@@ -491,10 +497,40 @@ impl Catalog {
         mut ops: Vec<Op>,
         temporary_ids: BTreeSet<CatalogItemId>,
         builtin_table_updates: &mut Vec<BuiltinTableUpdate>,
+        parsed_catalog_updates: &mut Vec<ParsedStateUpdate>,
         audit_events: &mut Vec<VersionedEvent>,
         tx: &mut Transaction<'_>,
         state: &CatalogState,
     ) -> Result<Option<CatalogState>, AdapterError> {
+        // We come up with new catalog state, builtin state updates, and parsed
+        // catalog updates (for deriving catalog implications) in two phases:
+        //
+        // 1. We (cow)-clone catalog state as `preliminary_state` and apply ops
+        //    one-by-one. This will give us the full list of updates to apply to
+        //    the catalog, which will allow us to apply it in one batch, which
+        //    in turn will allow the apply machinery to consolidate the updates.
+        // 2. We do one final apply call with all updates, which gives us the
+        //    final builtin table updates and parsed catalog updates.
+        //
+        // The reason is that the loop that is working off ops first does a
+        // transact_op to derive the state updates for that op, and then calls
+        // apply_updates on the catalog state. And successive ops might expect
+        // the catalog state to reflect the modified state _after_ applying
+        // previous ops.
+        //
+        // We want to, however, have one final apply_state that takes all the
+        // accumulated updates to derive the required controller updates and the
+        // builtin table updates.
+        //
+        // We won't win any DDL throughput benchmarks, but so far that's not
+        // what we're optimizing for and there would probably be other
+        // bottlenecks before we hit this one as a bottleneck.
+        //
+        // We could work around this by refactoring how the interplay of
+        // transact_op and apply_updates works, but that's a larger undertaking.
+        let mut preliminary_state = Cow::Borrowed(state);
+
+        // The final state that we will return, if modified.
         let mut state = Cow::Borrowed(state);
 
         let dry_run_ops = match ops.last() {
@@ -512,6 +548,8 @@ impl Catalog {
         let mut storage_collections_to_drop = BTreeSet::new();
         let mut storage_collections_to_register = BTreeMap::new();
 
+        let mut updates = Vec::new();
+
         for op in ops {
             let (weird_builtin_table_update, temporary_item_updates) = Self::transact_op(
                 oracle_write_ts,
@@ -520,7 +558,7 @@ impl Catalog {
                 &temporary_ids,
                 audit_events,
                 tx,
-                &*state,
+                &*preliminary_state,
                 &mut storage_collections_to_create,
                 &mut storage_collections_to_drop,
                 &mut storage_collections_to_register,
@@ -543,25 +581,35 @@ impl Catalog {
             // separately for updating state and builtin tables.
             // TODO(jkosh44) Some more thought needs to be given as to how temporary tables work
             // in a multi-subscriber catalog world.
-            let op_id = tx.op_id().into();
+            let upper = tx.upper();
             let temporary_item_updates =
                 temporary_item_updates
                     .into_iter()
                     .map(|(item, diff)| StateUpdate {
                         kind: StateUpdateKind::TemporaryItem(item),
-                        ts: op_id,
+                        ts: upper,
                         diff,
                     });
 
-            let mut updates: Vec<_> = tx.get_and_commit_op_updates();
-            updates.extend(temporary_item_updates);
-            if !updates.is_empty() {
-                let op_builtin_table_updates = state.to_mut().apply_updates(updates)?;
-                let op_builtin_table_updates = state
-                    .to_mut()
-                    .resolve_builtin_table_updates(op_builtin_table_updates);
-                builtin_table_updates.extend(op_builtin_table_updates);
+            let mut op_updates: Vec<_> = tx.get_and_commit_op_updates();
+            op_updates.extend(temporary_item_updates);
+            if !op_updates.is_empty() {
+                let (_op_builtin_table_updates, _op_catalog_updates) =
+                    preliminary_state
+                        .to_mut()
+                        .apply_updates(op_updates.clone())?;
             }
+            updates.append(&mut op_updates);
+        }
+
+        if !updates.is_empty() {
+            let (op_builtin_table_updates, op_catalog_updates) =
+                state.to_mut().apply_updates(updates.clone())?;
+            let op_builtin_table_updates = state
+                .to_mut()
+                .resolve_builtin_table_updates(op_builtin_table_updates);
+            builtin_table_updates.extend(op_builtin_table_updates);
+            parsed_catalog_updates.extend(op_catalog_updates);
         }
 
         if dry_run_ops.is_empty() {
@@ -578,11 +626,13 @@ impl Catalog {
 
             let updates = tx.get_and_commit_op_updates();
             if !updates.is_empty() {
-                let op_builtin_table_updates = state.to_mut().apply_updates(updates)?;
+                let (op_builtin_table_updates, op_catalog_updates) =
+                    state.to_mut().apply_updates(updates.clone())?;
                 let op_builtin_table_updates = state
                     .to_mut()
                     .resolve_builtin_table_updates(op_builtin_table_updates);
                 builtin_table_updates.extend(op_builtin_table_updates);
+                parsed_catalog_updates.extend(op_catalog_updates);
             }
 
             match state {
@@ -1146,15 +1196,31 @@ impl Catalog {
                         )));
                     }
                     let oid = tx.allocate_oid(&temporary_oids)?;
+
+                    let schema_id = name.qualifiers.schema_spec.clone().into();
+                    let item_type = item.typ();
+                    let (create_sql, global_id, versions) = item.to_serialized();
+
                     let item = TemporaryItem {
                         id,
                         oid,
-                        name: name.clone(),
-                        item: item.clone(),
+                        global_id,
+                        schema_id,
+                        name: name.item.clone(),
+                        create_sql,
+                        conn_id: item.conn_id().cloned(),
                         owner_id,
-                        privileges: PrivilegeMap::from_mz_acl_items(privileges),
+                        privileges: privileges.clone(),
+                        extra_versions: versions,
                     };
                     temporary_item_updates.push((item, StateDiff::Addition));
+
+                    info!(
+                        "create temporary {} {} ({})",
+                        item_type,
+                        state.resolve_full_name(&name, None),
+                        id
+                    );
                 } else {
                     if let Some(temp_id) =
                         item.uses()

src/adapter/src/catalog/transact.rs

@@ -221,6 +221,7 @@ pub(crate) mod statement_logging;
 pub(crate) mod timeline;
 pub(crate) mod timestamp_selection;
 
+pub mod catalog_implications;
 mod caught_up;
 mod command_handler;
 mod ddl;