mz_adapter/coord/

ddl.rs

// Copyright Materialize, Inc. and contributors. All rights reserved.
//
// Use of this software is governed by the Business Source License
// included in the LICENSE file.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0.

//! This module encapsulates all of the [`Coordinator`]'s logic for creating, dropping,
//! and altering objects.

use std::collections::BTreeMap;
use std::pin::Pin;
use std::sync::Arc;
use std::time::{Duration, Instant};

use fail::fail_point;
use maplit::{btreemap, btreeset};
use mz_adapter_types::compaction::SINCE_GRANULARITY;
use mz_adapter_types::connection::ConnectionId;
use mz_audit_log::VersionedEvent;
use mz_catalog::SYSTEM_CONN_ID;
use mz_catalog::memory::objects::Sink;
use mz_cluster_client::ReplicaId;
use mz_controller_types::ClusterId;
use mz_ore::cast::CastFrom;
use mz_ore::instrument;
use mz_ore::metrics::MetricsFutureExt;
use mz_ore::now::to_datetime;
use mz_ore::retry::Retry;
use mz_ore::task;
use mz_ore::tracing::OpenTelemetryContext;
use mz_repr::adt::numeric::Numeric;
use mz_repr::{CatalogItemId, GlobalId, Timestamp};
use mz_sql::catalog::CatalogClusterReplica;
use mz_sql::session::metadata::SessionMetadata;
use mz_sql::session::vars::{self, DEFAULT_TIMESTAMP_INTERVAL, SystemVars, Var};
use mz_storage_client::controller::{CollectionDescription, DataSource, ExportDescription};
use mz_storage_types::connections::inline::IntoInlineConnection;
use mz_storage_types::read_policy::ReadPolicy;
use mz_storage_types::sources::SourceConnection;
use mz_storage_types::sources::kafka::KAFKA_PROGRESS_DESC;
use serde_json::json;
use tracing::{Instrument, Level, event, info_span, warn};

use crate::active_compute_sink::{ActiveComputeSink, ActiveComputeSinkRetireReason};
use crate::catalog::{DropObjectInfo, Op, TransactionResult};
use crate::coord::appends::{BuiltinTableAppendCompletion, BuiltinTableAppendNotify};
use crate::coord::catalog_implications::parsed_state_updates::ParsedStateUpdate;
use crate::coord::{Coordinator, DeferredDdl, Message, PendingTxn};
use crate::session::{Session, TransactionOps};
use crate::telemetry::{EventDetails, SegmentClientExt};
use crate::util::ResultExt;
use crate::{AdapterError, ExecuteContext, catalog, flags};

#[derive(Debug)]
pub struct PendingSourceStatement {
    ctx: ExecuteContext,
    statement: catalog::command::source_statement::SourceStatement,
}

#[derive(Debug)]
pub struct PendingDdlCommit {
    context: DdlCommitContext,
    commands: Vec<catalog::command::CatalogCommand>,
    responses: BTreeMap<usize, mz_sql::pure::SourcePreparationResponse>,
}

#[derive(Debug)]
enum DdlCommitContext {
    Transaction(PendingTxn),
    Statement {
        ctx: ExecuteContext,
        response: crate::ExecuteResponse,
    },
}

impl PendingDdlCommit {
    pub(super) fn conn_id(&self) -> &ConnectionId {
        match &self.context {
            DdlCommitContext::Transaction(txn) => txn.ctx.session().conn_id(),
            DdlCommitContext::Statement { ctx, response: _ } => ctx.session().conn_id(),
        }
    }

    pub(super) fn transaction(
        txn: PendingTxn,
        commands: Vec<catalog::command::CatalogCommand>,
    ) -> Self {
        Self {
            context: DdlCommitContext::Transaction(txn),
            commands,
            responses: BTreeMap::new(),
        }
    }

    pub(super) fn statement(
        ctx: ExecuteContext,
        response: crate::ExecuteResponse,
        commands: Vec<catalog::command::CatalogCommand>,
    ) -> Self {
        Self {
            context: DdlCommitContext::Statement { ctx, response },
            commands,
            responses: BTreeMap::new(),
        }
    }
}

impl DdlCommitContext {
    fn context_mut(&mut self) -> &mut ExecuteContext {
        match self {
            Self::Transaction(txn) => &mut txn.ctx,
            Self::Statement { ctx, .. } => ctx,
        }
    }

    fn finish(self, result: Result<(), AdapterError>) {
        match self {
            Self::Transaction(mut txn) => {
                if let Err(error) = result {
                    txn.response = Err(error);
                    txn.action = crate::session::EndTransactionAction::Rollback;
                }
                txn.finish();
            }
            Self::Statement { ctx, response } => ctx.retire(result.map(|()| response)),
        }
    }
}

/// Waits while a test configures the named failpoint with a `return` action.
async fn catalog_preparation_test_pause(name: &'static str) {
    let paused = || {
        fail::fail_point!(name, |_| true);
        false
    };
    // Clearing the failpoint requires coordinator and runtime progress.
    while paused() {
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }
}

impl Coordinator {
    /// Sets allowed responses and checks read-only and execution-cluster restrictions.
    fn prepare_command_execution(
        &mut self,
        ctx: &mut ExecuteContext,
        plan: &mz_sql::plan::Plan,
    ) -> Result<(), AdapterError> {
        ctx.tx_mut()
            .set_allowed(crate::ExecuteResponse::generated_from(
                &mz_sql::plan::PlanKind::from(plan),
            ));
        if self.controller.read_only() && !plan.allowed_in_read_only() {
            return Err(AdapterError::ReadOnly);
        }
        self.prepare_plan_execution(ctx, plan)?;
        Ok(())
    }

    pub(super) fn sequence_catalog_result_command(
        &mut self,
        mut ctx: ExecuteContext,
        statement: Arc<mz_sql::ast::Statement<mz_sql::ast::Raw>>,
        params: mz_sql::plan::Params,
    ) {
        let prepared = {
            let catalog = self.catalog().for_session(ctx.session());
            catalog::command::CatalogResultCommand::prepare(
                catalog.state(),
                statement,
                params,
                ctx.session().meta(),
            )
        };
        let (command, plan) = match prepared {
            Ok(prepared) => prepared,
            Err(error) => return ctx.retire(Err(error)),
        };
        if let Err(error) = self.prepare_command_execution(&mut ctx, &plan) {
            return ctx.retire(Err(error));
        }
        let row = match catalog::command::CatalogResultCommand::result(&plan) {
            Ok(row) => row,
            Err(error) => return ctx.retire(Err(error)),
        };
        let command = Arc::new(command);
        self.emit_command_notices(
            ctx.session(),
            &catalog::command::CatalogCommand::CatalogResult(Arc::clone(&command)),
        );
        ctx.session_mut().retain_catalog_result(command);
        ctx.retire(Ok(Self::send_immediate_rows(row)));
    }

    pub(super) async fn sequence_source_statement(
        &mut self,
        ctx: ExecuteContext,
        statement: Arc<mz_sql::ast::Statement<mz_sql::ast::Raw>>,
        params: mz_sql::plan::Params,
    ) {
        let statement = catalog::command::source_statement::SourceStatement::new(
            statement,
            params,
            ctx.session().meta(),
            (self.catalog().config().now)(),
        );
        self.sequence_source_statement_ready(PendingSourceStatement { ctx, statement }, None)
            .await;
    }

    pub(super) async fn sequence_source_statement_ready(
        &mut self,
        mut pending: PendingSourceStatement,
        mut response: Option<mz_sql::pure::SourcePreparationResponse>,
    ) {
        use catalog::command::source_statement::{SourceCommand, SourceResolution};
        if self.controller.read_only() {
            return pending.ctx.retire(Err(AdapterError::ReadOnly));
        }
        let compiled = loop {
            let state = pending
                .ctx
                .session()
                .transaction()
                .catalog_state()
                .unwrap_or_else(|| self.catalog().state());
            let result = match response.take() {
                Some(response) => pending.statement.finish(state, response),
                None => pending.statement.resolve(state),
            };
            match result {
                Err(error) => return pending.ctx.retire(Err(error)),
                Ok(SourceResolution::Ready(compiled)) => break compiled,
                Ok(SourceResolution::Allocate(keys)) => {
                    let ids = match self.allocate_user_ids(u64::cast_from(keys.len())).await {
                        Ok(ids) => ids,
                        Err(error) => return pending.ctx.retire(Err(error)),
                    };
                    pending.statement.record_ids(keys, ids);
                }
                Ok(SourceResolution::Prepare(request)) => {
                    let conn_id = pending.ctx.session().conn_id().clone();
                    let config = self.controller.storage.config().clone();
                    let tx = self.internal_cmd_tx.clone();
                    let otel_ctx = OpenTelemetryContext::obtain();
                    task::spawn(|| "ddl_statement_preparation", async move {
                        let response = request.execute(&config).await;
                        tracing::debug!(
                            target: "mz_adapter::catalog_preparation",
                            %conn_id, kind = "purification", "catalog preparation complete"
                        );
                        catalog_preparation_test_pause("catalog_purification_ready").await;
                        if let Err(error) = tx.send(Message::SourceStatementReady {
                            pending, response, otel_ctx,
                        }) {
                            warn!("internal_cmd_rx dropped before source preparation completed: {error:?}");
                        }
                    }.instrument(tracing::Span::current()));
                    return;
                }
            }
        };
        let mut ctx = pending.ctx;
        if let Err(error) = self.prepare_command_execution(&mut ctx, &compiled.plan) {
            return ctx.retire(Err(error));
        }
        let response = match &compiled.plan {
            mz_sql::plan::Plan::CreateSources(_) => crate::ExecuteResponse::CreatedSource,
            mz_sql::plan::Plan::CreateTable(_) => crate::ExecuteResponse::CreatedTable,
            mz_sql::plan::Plan::AlterSource(_) => {
                crate::ExecuteResponse::AlteredObject(mz_sql::catalog::ObjectType::Source)
            }
            mz_sql::plan::Plan::AlterNoop(plan) => {
                crate::ExecuteResponse::AlteredObject(plan.object_type)
            }
            _ => unreachable!("source command plan"),
        };
        if ctx.session().transaction().is_ddl() {
            let ts = self.get_local_write_ts().await.timestamp;
            let TransactionOps::DDL {
                commands: _,
                state,
                snapshot,
            } = &ctx
                .session()
                .transaction()
                .inner()
                .expect("DDL transaction")
                .ops
            else {
                unreachable!("DDL transaction")
            };
            let staged = match self
                .catalog()
                .stage_source_statement(
                    pending.statement,
                    *compiled,
                    (state, snapshot),
                    self.active_conns.get(ctx.session().conn_id()),
                    ts,
                )
                .await
            {
                Ok(staged) => staged,
                Err(error) => return ctx.retire(Err(error)),
            };
            let command = catalog::command::CatalogCommand::Source(Arc::new(staged.command));
            if let Err(error) = ctx
                .session_mut()
                .transaction_mut()
                .add_ops(TransactionOps::DDL {
                    commands: vec![command.clone()],
                    state: staged.state,
                    snapshot: staged.snapshot,
                })
            {
                return ctx.retire(Err(error));
            }
            self.emit_command_notices(ctx.session(), &command);
            ctx.retire(Ok(response));
        } else {
            let command = catalog::command::CatalogCommand::Source(Arc::new(
                SourceCommand::unobserved(pending.statement, *compiled),
            ));
            self.sequence_ddl_commit(PendingDdlCommit::statement(ctx, response, vec![command]))
                .await;
        }
    }

    fn emit_command_notices(&self, session: &Session, command: &catalog::command::CatalogCommand) {
        session.add_notices(command.notices().iter().cloned().map(Into::into));
        let catalog::command::CatalogCommand::Source(command) = command else {
            return;
        };
        if let Some(notice) = command.skipped_notice() {
            session.add_notice(notice);
            return;
        }
        let state = session
            .transaction()
            .catalog_state()
            .unwrap_or_else(|| self.catalog().state());
        for id in command.item_ids() {
            let Some(entry) = state.try_get_entry(&id) else {
                continue;
            };
            let Some(source) = entry.source() else {
                continue;
            };
            use mz_catalog::memory::objects::DataSourceDesc;
            let cluster_id = match &source.data_source {
                DataSourceDesc::Ingestion { desc, cluster_id }
                | DataSourceDesc::OldSyntaxIngestion {
                    desc, cluster_id, ..
                } if desc.connection.prefers_single_replica() => cluster_id,
                _ => continue,
            };
            let Some(cluster) = self.catalog().try_get_cluster(*cluster_id) else {
                continue;
            };
            let name = state.resolve_full_name(entry.name(), None);
            self.notify_single_replica_sources(session, cluster, Some(&name));
        }
    }

    pub(super) async fn sequence_rename_command(
        &mut self,
        mut ctx: ExecuteContext,
        statement: Arc<mz_sql::ast::Statement<mz_sql::ast::Raw>>,
        params: mz_sql::plan::Params,
    ) {
        let prepared = {
            let catalog = self.catalog().for_session(ctx.session());
            catalog::command::RenameCommand::prepare(
                catalog.state(),
                statement,
                params,
                ctx.session().meta(),
            )
        };
        let (command, plan) = match prepared {
            Ok(prepared) => prepared,
            Err(error) => return ctx.retire(Err(error)),
        };
        if let Err(error) = self.prepare_command_execution(&mut ctx, &plan) {
            return ctx.retire(Err(error));
        }
        let command = Arc::new(command);
        if !ctx.session().transaction().is_ddl() {
            let response = crate::ExecuteResponse::AlteredObject(command.object_type());
            self.sequence_ddl_commit(PendingDdlCommit::statement(
                ctx,
                response,
                vec![catalog::command::CatalogCommand::Rename(command)],
            ))
            .await;
            return;
        }
        let TransactionOps::DDL {
            state, snapshot, ..
        } = &ctx
            .session()
            .transaction()
            .inner()
            .expect("DDL transaction")
            .ops
        else {
            unreachable!("standalone command handled above")
        };
        let state = state.clone();
        let snapshot = snapshot.clone();
        let ts = self.get_local_write_ts().await.timestamp;
        let result = Box::pin(self.catalog().transact_incremental_dry_run(
            &state,
            vec![Op::RenameCommand(Arc::clone(&command))],
            self.active_conns.get(ctx.session().conn_id()),
            Some(snapshot),
            ts,
        ))
        .await
        .and_then(|(state, snapshot)| {
            ctx.session_mut()
                .transaction_mut()
                .add_ops(TransactionOps::DDL {
                    commands: vec![catalog::command::CatalogCommand::Rename(Arc::clone(
                        &command,
                    ))],
                    state,
                    snapshot,
                })
        });
        match result {
            Ok(()) => {
                ctx.session()
                    .add_notices(command.notices().iter().cloned().map(Into::into));
                ctx.retire(Ok(crate::ExecuteResponse::AlteredObject(
                    command.object_type(),
                )));
            }
            Err(error) => ctx.retire(Err(error)),
        }
    }

    /// Same as [`Self::catalog_transact_with_context`] but takes a [`Session`].
    #[instrument(name = "coord::catalog_transact")]
    pub(crate) async fn catalog_transact(
        &mut self,
        session: Option<&Session>,
        ops: Vec<catalog::Op>,
    ) -> Result<(), AdapterError> {
        if session.is_some_and(|session| session.transaction().is_ddl()) {
            return Err(AdapterError::Unsupported(
                "raw operations in a DDL command transaction",
            ));
        }
        let start = Instant::now();
        let result = self
            .catalog_transact_with_context(session.map(|session| session.conn_id()), None, ops)
            .await;
        self.metrics
            .catalog_transact_seconds
            .with_label_values(&["catalog_transact"])
            .observe(start.elapsed().as_secs_f64());
        result
    }

    /// Same as [`Self::catalog_transact_with_context`] but takes a [`Session`]
    /// and runs builtin table updates concurrently with any side effects (e.g.
    /// creating collections).
    // TODO(aljoscha): Remove this method once all call-sites have been migrated
    // to the newer catalog_transact_with_context. The latter is what allows us
    // to apply catalog implications that we derive from catalog chanages either
    // when initially applying the ops to the catalog _or_ when following
    // catalog changes from another process.
    #[instrument(name = "coord::catalog_transact_with_side_effects")]
    pub(crate) async fn catalog_transact_with_side_effects<F>(
        &mut self,
        mut ctx: Option<&mut ExecuteContext>,
        ops: Vec<catalog::Op>,
        side_effect: F,
    ) -> Result<(), AdapterError>
    where
        F: for<'a> FnOnce(
                &'a mut Coordinator,
                Option<&'a mut ExecuteContext>,
            ) -> Pin<Box<dyn Future<Output = ()> + 'a>>
            + 'static,
    {
        if ctx
            .as_ref()
            .is_some_and(|ctx| ctx.session().transaction().is_ddl())
        {
            return Err(AdapterError::Unsupported(
                "raw operations in a DDL command transaction",
            ));
        }
        let start = Instant::now();

        // Keep catalog work out of the inline state of callers such as ordinary
        // transaction completion and the coordinator's command dispatcher.
        let (table_updates, catalog_updates) = Box::pin(
            self.catalog_transact_inner(ctx.as_ref().map(|ctx| ctx.session().conn_id()), ops),
        )
        .await?;

        // We can't run this concurrently with the explicit side effects,
        // because both want to borrow self mutably.
        let apply_implications_res =
            Box::pin(self.apply_catalog_implications(ctx.as_deref_mut(), catalog_updates)).await;

        // We would get into an inconsistent state if we updated the catalog but
        // then failed to apply commands/updates to the controller. Easiest
        // thing to do is panic and let restart/bootstrap handle it.
        apply_implications_res.expect("cannot fail to apply catalog update implications");

        // NOTE: `check_consistency` only runs with soft assertions enabled, so
        // this phase reads about zero in production. We time it because a local
        // rig debugging a transact stall commonly has them on, where the check is
        // O(catalog size) and would otherwise appear as an unexplained remainder
        // against the wrapper metric. The observation stays outside the macro,
        // anything inside it compiles out exactly where soft assertions are off.
        let consistency_start = Instant::now();

        // Note: It's important that we keep the function call inside macro, this way we only run
        // the consistency checks if soft assertions are enabled.
        mz_ore::soft_assert_eq_no_log!(
            self.check_consistency(),
            Ok(()),
            "coordinator inconsistency detected"
        );

        self.metrics
            .catalog_transact_phase_seconds
            .with_label_values(&["consistency_check"])
            .observe(consistency_start.elapsed().as_secs_f64());

        let side_effects_seconds = self
            .metrics
            .catalog_transact_phase_seconds
            .with_label_values(&["side_effects"]);
        // Distinct from `table_updates_wait` in `catalog_transact_with_context`.
        // Here the group commit has already been running concurrently with
        // `apply_catalog_implications` above, so this wrapper is first polled
        // late and only records the residual wait.
        let table_updates_wait = self
            .metrics
            .catalog_transact_phase_seconds
            .with_label_values(&["table_updates_residual_wait"]);
        let side_effects_fut = side_effect(self, ctx);

        // Run our side effects concurrently with the table updates.
        let ((), ()) = futures::future::join(
            side_effects_fut
                .wall_time()
                .observe(side_effects_seconds)
                .instrument(info_span!(
                    "coord::catalog_transact_with_side_effects::side_effects_fut"
                )),
            table_updates
                .wall_time()
                .observe(table_updates_wait)
                .instrument(info_span!(
                    "coord::catalog_transact_with_side_effects::table_updates"
                )),
        )
        .await;

        self.metrics
            .catalog_transact_seconds
            .with_label_values(&["catalog_transact_with_side_effects"])
            .observe(start.elapsed().as_secs_f64());

        Ok(())
    }

    /// Same as [`Self::catalog_transact_inner`] but takes an execution context
    /// or connection ID and runs builtin table updates concurrently with any
    /// catalog implications that are generated as part of applying the given
    /// `ops` (e.g. creating collections).
    ///
    /// This will use a connection ID if provided and otherwise fall back to
    /// getting a connection ID from the execution context.
    #[instrument(name = "coord::catalog_transact_with_context")]
    pub(crate) async fn catalog_transact_with_context(
        &mut self,
        conn_id: Option<&ConnectionId>,
        ctx: Option<&mut ExecuteContext>,
        ops: Vec<catalog::Op>,
    ) -> Result<(), AdapterError> {
        if ctx
            .as_ref()
            .is_some_and(|ctx| ctx.session().transaction().is_ddl())
        {
            return Err(AdapterError::Unsupported(
                "raw operations in a DDL command transaction",
            ));
        }
        let start = Instant::now();

        let conn_id = conn_id.or_else(|| ctx.as_ref().map(|ctx| ctx.session().conn_id()));

        let (table_updates, catalog_updates) =
            Box::pin(self.catalog_transact_inner(conn_id, ops)).await?;

        self.finish_catalog_transaction(ctx, table_updates, catalog_updates)
            .await;

        self.metrics
            .catalog_transact_seconds
            .with_label_values(&["catalog_transact_with_context"])
            .observe(start.elapsed().as_secs_f64());

        Ok(())
    }

    pub(super) fn release_ddl_lock(&self, guard: tokio::sync::OwnedMutexGuard<()>) {
        drop(guard);
        if !self.serialized_ddl.is_empty() {
            let _ = self.internal_cmd_tx.send(Message::DeferredStatementReady);
        }
    }

    pub(super) async fn sequence_ddl_commit(&mut self, mut pending: PendingDdlCommit) {
        // Command replay must respect the plans other DDLs hold across off-thread work.
        // Never await this lock on the coordinator, which its owner needs to finish.
        let guard = self
            .active_conns
            .get_mut(pending.conn_id())
            .expect("connection must exist")
            .deferred_lock
            .take();
        let guard = match guard {
            Some(guard) => guard,
            None => match self.serialized_ddl.try_lock_owned() {
                Ok(guard) => guard,
                Err(_) => {
                    tracing::debug!(
                        conn_id = %pending.conn_id(),
                        "deferring catalog commit for DDL lock"
                    );
                    self.serialized_ddl.push_back(DeferredDdl::Commit(pending));
                    return;
                }
            },
        };
        let result = loop {
            match self
                .catalog_transact_commands(
                    pending.context.context_mut(),
                    &mut pending.commands,
                    &mut pending.responses,
                )
                .await
            {
                Ok(()) => break Ok(()),
                Err(catalog::CommandExecutionError::Error(error)) => break Err(error),
                Err(catalog::CommandExecutionError::Pending { index, work }) => match work {
                    catalog::CommandWork::Allocate(keys) => {
                        let ids = match self.allocate_user_ids(u64::cast_from(keys.len())).await {
                            Ok(ids) => ids,
                            Err(error) => break Err(error),
                        };
                        match &mut pending.commands[index] {
                            catalog::command::CatalogCommand::Source(command) => {
                                Arc::make_mut(command).record_ids(keys, ids);
                            }
                            catalog::command::CatalogCommand::Rename(_)
                            | catalog::command::CatalogCommand::CatalogResult(_)
                            | catalog::command::CatalogCommand::Description(_) => {
                                unreachable!("renames do not allocate IDs");
                            }
                        }
                    }
                    catalog::CommandWork::Prepare(request) => {
                        self.release_ddl_lock(guard);
                        let conn_id = pending.context.context_mut().session().conn_id().clone();
                        let config = self.controller.storage.config().clone();
                        let tx = self.internal_cmd_tx.clone();
                        let otel_ctx = OpenTelemetryContext::obtain();
                        task::spawn(|| "ddl_command_preparation", async move {
                            let response = request.execute(&config).await;
                            tracing::debug!(
                                target: "mz_adapter::catalog_preparation",
                                %conn_id, kind = "purification", "catalog preparation complete"
                            );
                            catalog_preparation_test_pause("catalog_purification_ready").await;
                            pending.responses.insert(index, response);
                            if let Err(error) =
                                tx.send(Message::DdlCommandsReady { pending, otel_ctx })
                            {
                                warn!(
                                    "internal_cmd_rx dropped before DDL preparation completed: {error:?}"
                                );
                            }
                        }.instrument(tracing::Span::current()));
                        return;
                    }
                },
            }
        };
        self.release_ddl_lock(guard);
        self.finish_ddl_commit(pending, result);
    }

    pub(super) fn finish_ddl_commit(
        &self,
        mut pending: PendingDdlCommit,
        result: Result<(), AdapterError>,
    ) {
        if result.is_ok() && matches!(pending.context, DdlCommitContext::Statement { .. }) {
            for command in &pending.commands {
                self.emit_command_notices(pending.context.context_mut().session(), command);
            }
        }
        pending.context.finish(result);
    }

    pub(crate) async fn catalog_transact_commands(
        &mut self,
        ctx: &mut ExecuteContext,
        commands: &mut [catalog::command::CatalogCommand],
        responses: &mut BTreeMap<usize, mz_sql::pure::SourcePreparationResponse>,
    ) -> Result<(), catalog::CommandExecutionError> {
        if self.controller.read_only() {
            return Err(AdapterError::ReadOnly.into());
        }
        let start = Instant::now();
        let phase_seconds = self.metrics.catalog_transact_phase_seconds.clone();
        let oracle_write_ts = self
            .get_catalog_write_ts()
            .wall_time()
            .observe(phase_seconds.with_label_values(&["write_ts"]))
            .await;
        let conn_id = ctx.session().conn_id();
        let Coordinator {
            catalog,
            active_conns,
            controller,
            ..
        } = self;
        let conn = active_conns.get(conn_id).expect("connection must exist");
        let TransactionResult {
            builtin_table_updates,
            catalog_updates,
            audit_events,
        } = Box::pin(Arc::make_mut(catalog).transact_commands(
            Some(&mut controller.storage_collections),
            oracle_write_ts,
            Some(conn),
            commands,
            responses,
        ))
        .wall_time()
        .observe(phase_seconds.with_label_values(&["transact"]))
        .await?;

        self.track_catalog_audit_events(Some(conn_id), audit_events);
        let stage_start = Instant::now();
        let table_updates = self.builtin_table_update().execute(builtin_table_updates);
        phase_seconds
            .with_label_values(&["stage_builtin"])
            .observe(stage_start.elapsed().as_secs_f64());
        self.finish_catalog_transaction(Some(ctx), table_updates, catalog_updates)
            .await;
        self.metrics
            .catalog_transact_seconds
            .with_label_values(&["catalog_transact_with_context"])
            .observe(start.elapsed().as_secs_f64());
        Ok(())
    }

    async fn finish_catalog_transaction(
        &mut self,
        ctx: Option<&mut ExecuteContext>,
        table_updates: BuiltinTableAppendNotify,
        catalog_updates: Vec<ParsedStateUpdate>,
    ) {
        let table_updates_wait = self
            .metrics
            .catalog_transact_phase_seconds
            .with_label_values(&["table_updates_wait"]);
        let apply_catalog_implications_fut =
            Box::pin(self.apply_catalog_implications(ctx, catalog_updates));

        // Apply catalog implications concurrently with the table updates.
        let (combined_apply_res, ()) = futures::future::join(
            apply_catalog_implications_fut.instrument(info_span!(
                "coord::catalog_transact_with_context::side_effects_fut"
            )),
            table_updates
                .wall_time()
                .observe(table_updates_wait)
                .instrument(info_span!(
                    "coord::catalog_transact_with_context::table_updates"
                )),
        )
        .await;

        // We would get into an inconsistent state if we updated the catalog but
        // then failed to apply implications. Easiest thing to do is panic and
        // let restart/bootstrap handle it.
        combined_apply_res.expect("cannot fail to apply catalog implications");

        // See the note in `catalog_transact_with_side_effects` on why this is
        // timed outside the macro and reads about zero in production.
        let consistency_start = Instant::now();

        // Note: It's important that we keep the function call inside macro, this way we only run
        // the consistency checks if soft assertions are enabled.
        mz_ore::soft_assert_eq_no_log!(
            self.check_consistency(),
            Ok(()),
            "coordinator inconsistency detected"
        );

        self.metrics
            .catalog_transact_phase_seconds
            .with_label_values(&["consistency_check"])
            .observe(consistency_start.elapsed().as_secs_f64());
    }

    /// Perform a catalog transaction. [`Coordinator::ship_dataflow`] must be
    /// called after this function successfully returns on any built
    /// [`MirDataflowDescription`](mz_compute_types::dataflows::MirDataflowDescription).
    #[instrument(name = "coord::catalog_transact_inner")]
    pub(crate) async fn catalog_transact_inner(
        &mut self,
        conn_id: Option<&ConnectionId>,
        mut ops: Vec<catalog::Op>,
    ) -> Result<(BuiltinTableAppendNotify, Vec<ParsedStateUpdate>), AdapterError> {
        if self.controller.read_only() {
            return Err(AdapterError::ReadOnly);
        }

        if let Some(scoped_op) = self.scoped_overrides_create_op(&ops) {
            ops.push(scoped_op);
        }

        event!(Level::TRACE, ops = format!("{:?}", ops));

        let phase_seconds = self.metrics.catalog_transact_phase_seconds.clone();
        let phase_start = Instant::now();

        let mut clusters_to_drop = vec![];
        let mut cluster_replicas_to_drop = vec![];
        let mut clusters_to_create = vec![];
        let mut cluster_replicas_to_create = vec![];
        let mut update_metrics_config = false;
        let mut update_tracing_config = false;
        let mut update_controller_config = false;
        let mut update_compute_config = false;
        let mut update_storage_config = false;
        let mut update_timestamp_oracle_config = false;
        let mut update_metrics_retention = false;
        let mut update_secrets_caching_config = false;
        let mut update_cluster_scheduling_config = false;
        let mut update_http_config = false;
        let mut update_advance_timelines_interval = false;
        let mut update_optimizer_e2e_latency_warning_threshold = false;
        let mut reconcile_metric_sinks = false;

        for op in &ops {
            match op {
                catalog::Op::DropObjects(drop_object_infos) => {
                    for drop_object_info in drop_object_infos {
                        match &drop_object_info {
                            catalog::DropObjectInfo::Item(_) => {
                                // Nothing to do, these will be handled by
                                // applying the side effects that we return.
                            }
                            catalog::DropObjectInfo::Cluster(id) => {
                                clusters_to_drop.push(*id);
                            }
                            catalog::DropObjectInfo::ClusterReplica((
                                cluster_id,
                                replica_id,
                                _reason,
                            )) => {
                                // Drop the cluster replica itself.
                                cluster_replicas_to_drop.push((*cluster_id, *replica_id));
                            }
                            _ => (),
                        }
                    }
                }
                catalog::Op::ResetSystemConfiguration { name }
                | catalog::Op::UpdateSystemConfiguration { name, .. } => {
                    update_metrics_config |= self
                        .catalog
                        .state()
                        .system_config()
                        .is_metrics_config_var(name);
                    update_tracing_config |= vars::is_tracing_var(name);
                    update_controller_config |= self
                        .catalog
                        .state()
                        .system_config()
                        .is_controller_config_var(name);
                    update_compute_config |= self
                        .catalog
                        .state()
                        .system_config()
                        .is_compute_config_var(name);
                    update_storage_config |= self
                        .catalog
                        .state()
                        .system_config()
                        .is_storage_config_var(name);
                    update_timestamp_oracle_config |= vars::is_timestamp_oracle_config_var(name);
                    update_metrics_retention |= name == vars::METRICS_RETENTION.name();
                    update_secrets_caching_config |= vars::is_secrets_caching_var(name);
                    update_cluster_scheduling_config |= vars::is_cluster_scheduling_var(name);
                    update_http_config |= vars::is_http_config_var(name);
                    update_advance_timelines_interval |= name == DEFAULT_TIMESTAMP_INTERVAL.name();
                    update_optimizer_e2e_latency_warning_threshold |=
                        name == vars::OPTIMIZER_E2E_LATENCY_WARNING_THRESHOLD.name();
                    reconcile_metric_sinks |= name == vars::DISABLED_METRIC_SINKS.name();
                }
                catalog::Op::ResetAllSystemConfiguration => {
                    // Assume they all need to be updated.
                    // We could see if the config's have actually changed, but
                    // this is simpler.
                    update_tracing_config = true;
                    update_controller_config = true;
                    update_compute_config = true;
                    update_storage_config = true;
                    update_timestamp_oracle_config = true;
                    update_metrics_retention = true;
                    update_secrets_caching_config = true;
                    update_cluster_scheduling_config = true;
                    update_metrics_config = true;
                    update_http_config = true;
                    update_advance_timelines_interval = true;
                    update_optimizer_e2e_latency_warning_threshold = true;
                    reconcile_metric_sinks = true;
                }
                catalog::Op::CreateCluster { id, .. } => {
                    clusters_to_create.push(*id);
                }
                catalog::Op::CreateClusterReplica {
                    cluster_id,
                    name,
                    config,
                    ..
                } => {
                    cluster_replicas_to_create.push((
                        *cluster_id,
                        name.clone(),
                        config.location.num_processes(),
                    ));
                }
                _ => (),
            }
        }

        // Observe before propagating, so a transaction rejected on resource
        // limits still accounts for the op scan it burned on the loop.
        let validate_res = self
            .catalog()
            .state()
            .validate_resource_limits(&ops, conn_id.unwrap_or(&SYSTEM_CONN_ID));
        phase_seconds
            .with_label_values(&["prep"])
            .observe(phase_start.elapsed().as_secs_f64());
        validate_res?;

        // This will produce timestamps that are guaranteed to increase on each
        // call, and also never be behind the system clock. If the system clock
        // hasn't advanced (or has gone backward), it will increment by 1. For
        // the audit log, we need to balance "close (within 10s or so) to the
        // system clock" and "always goes up". We've chosen here to prioritize
        // always going up, and believe we will always be close to the system
        // clock because it is well configured (chrony) and so may only rarely
        // regress or pause for 10s.
        let oracle_write_ts = self
            .get_catalog_write_ts()
            .wall_time()
            .observe(phase_seconds.with_label_values(&["write_ts"]))
            .await;

        let Coordinator {
            catalog,
            active_conns,
            controller,
            cluster_replica_statuses,
            ..
        } = self;
        let catalog = Arc::make_mut(catalog);
        let conn = conn_id.map(|id| active_conns.get(id).expect("connection must exist"));

        // Register the session as an ephemeral owner (its uuid <-> connection
        // mapping) at its first temporary-item creation.
        if let Some(conn) = conn {
            let creates_temp_item = ops.iter().any(
                |op| matches!(op, catalog::Op::CreateItem { item, .. } if item.is_temporary()),
            );
            if creates_temp_item && !catalog.state().has_temporary_namespace(conn.conn_id()) {
                catalog.register_temporary_namespace(conn.conn_id(), conn.uuid());
            }
        }

        // NOTE: This phase contains every durable `sync` and `commit` a catalog
        // transaction performs, which is what makes `transact` minus those two
        // histograms an estimate of the in-memory work. Two caveats. More than
        // one sync happens per transaction, so the subtraction is only valid on
        // rates of `_sum`, never on per-observation means. And durable
        // `allocate_id` (user ID pool refills, storage usage batch IDs) observes
        // into the same histograms from outside any catalog transaction, so the
        // estimate is biased low while allocation is active.
        let TransactionResult {
            builtin_table_updates,
            catalog_updates,
            audit_events,
        } = catalog
            .transact(
                Some(&mut controller.storage_collections),
                oracle_write_ts,
                conn,
                ops,
            )
            .wall_time()
            .observe(phase_seconds.with_label_values(&["transact"]))
            .await?;

        for (cluster_id, replica_id) in &cluster_replicas_to_drop {
            cluster_replica_statuses.remove_cluster_replica_statuses(cluster_id, replica_id);
        }
        for cluster_id in &clusters_to_drop {
            cluster_replica_statuses.remove_cluster_statuses(cluster_id);
        }
        for cluster_id in clusters_to_create {
            cluster_replica_statuses.initialize_cluster_statuses(cluster_id);
        }
        let now = to_datetime((catalog.config().now)());
        for (cluster_id, replica_name, num_processes) in cluster_replicas_to_create {
            let replica_id = catalog
                .resolve_replica_in_cluster(&cluster_id, &replica_name)
                .expect("just created")
                .replica_id();
            cluster_replica_statuses.initialize_cluster_replica_statuses(
                cluster_id,
                replica_id,
                num_processes,
                now,
            );
        }

        // Append our builtin table updates, then return the notify so we can run other tasks in
        // parallel.
        let stage_start = Instant::now();
        let builtin_update_notify = self.builtin_table_update().execute(builtin_table_updates);
        phase_seconds
            .with_label_values(&["stage_builtin"])
            .observe(stage_start.elapsed().as_secs_f64());

        let finalize_start = Instant::now();

        // No error returns are allowed after this point. Enforce this at compile time
        // by using this odd structure so we don't accidentally add a stray `?`.
        let _: () = async {
            if update_metrics_config {
                mz_metrics::update_dyncfg(&self.catalog().system_config().dyncfg_updates());
            }
            if update_controller_config {
                self.update_controller_config();
            }
            if update_compute_config {
                self.update_compute_config();
            }
            if update_storage_config {
                self.update_storage_config();
            }
            if update_timestamp_oracle_config {
                self.update_timestamp_oracle_config();
            }
            if update_metrics_retention {
                self.update_metrics_retention();
            }
            if update_tracing_config {
                self.update_tracing_config();
            }
            if update_secrets_caching_config {
                self.update_secrets_caching_config();
            }
            if update_cluster_scheduling_config {
                self.update_cluster_scheduling_config();
            }
            if update_http_config {
                self.update_http_config();
            }
            if update_advance_timelines_interval {
                let new_interval = self.catalog().system_config().default_timestamp_interval();
                if new_interval != self.advance_timelines_interval.period() {
                    self.advance_timelines_interval = tokio::time::interval(new_interval);
                }
            }
            if update_optimizer_e2e_latency_warning_threshold {
                let threshold = self
                    .catalog()
                    .system_config()
                    .optimizer_e2e_latency_warning_threshold();
                self.optimizer_metrics
                    .set_e2e_optimization_time_log_threshold(threshold);
            }
            if reconcile_metric_sinks {
                self.reconcile_metric_sinks().await;
            }
        }
        .instrument(info_span!("coord::catalog_transact_with::finalize"))
        .await;

        self.track_catalog_audit_events(conn_id, audit_events);

        phase_seconds
            .with_label_values(&["finalize"])
            .observe(finalize_start.elapsed().as_secs_f64());

        Ok((builtin_update_notify, catalog_updates))
    }

    fn track_catalog_audit_events(
        &self,
        conn_id: Option<&ConnectionId>,
        audit_events: Vec<VersionedEvent>,
    ) {
        let conn = conn_id.and_then(|id| self.active_conns.get(id));
        if let Some(segment_client) = &self.segment_client {
            for VersionedEvent::V1(event) in audit_events {
                let event_type = format!(
                    "{} {}",
                    event.object_type.as_title_case(),
                    event.event_type.as_title_case()
                );
                segment_client.environment_track(
                    &self.catalog().config().environment_id,
                    event_type,
                    json!({ "details": event.details.as_json() }),
                    EventDetails {
                        user_id: conn
                            .and_then(|c| c.user().external_metadata.as_ref())
                            .map(|m| m.user_id),
                        application_name: conn.map(|c| c.application_name()),
                        ..Default::default()
                    },
                );
            }
        }
    }

    pub(crate) fn drop_replica(&mut self, cluster_id: ClusterId, replica_id: ReplicaId) {
        self.drop_introspection_subscribes(replica_id);
        self.drop_metric_sinks(replica_id);

        self.controller
            .drop_replica(cluster_id, replica_id)
            .expect("dropping replica must not fail");
    }

    /// A convenience method for dropping sources.
    pub(crate) fn drop_sources(&mut self, sources: Vec<(CatalogItemId, GlobalId)>) {
        for (item_id, _gid) in &sources {
            self.active_webhooks.remove(item_id);
        }
        let storage_metadata = self.catalog.state().storage_metadata();
        let source_gids = sources.into_iter().map(|(_id, gid)| gid).collect();
        self.controller
            .storage
            .drop_sources(storage_metadata, source_gids)
            .unwrap_or_terminate("cannot fail to drop sources");
    }

    /// A convenience method for dropping tables.
    pub(crate) async fn drop_tables(&mut self, tables: Vec<(CatalogItemId, GlobalId)>) {
        for (item_id, _gid) in &tables {
            self.active_webhooks.remove(item_id);
        }

        let table_gids: Vec<_> = tables.into_iter().map(|(_id, gid)| gid).collect();

        // FIFO ordering places the forget after every staged append.
        let forget_ids = self
            .controller
            .storage
            .txns_table_ids(table_gids.clone())
            .unwrap_or_terminate("cannot fail to look up txns-registered tables");
        if !forget_ids.is_empty() {
            self.forget_tables_via_committer(forget_ids).await;
        }

        let storage_metadata = self.catalog.state().storage_metadata();
        self.controller
            .storage
            .drop_tables(storage_metadata, table_gids)
            .unwrap_or_terminate("cannot fail to drop tables");
    }

    /// Like `drop_compute_sinks`, but for a single compute sink.
    ///
    /// Returns the controller's state for the compute sink if the identified
    /// sink was known to the controller. It is the caller's responsibility to
    /// retire the returned sink. Consider using `retire_compute_sinks` instead.
    #[must_use]
    pub async fn drop_compute_sink(
        &mut self,
        sink_id: GlobalId,
    ) -> Option<(ActiveComputeSink, BuiltinTableAppendNotify)> {
        self.drop_compute_sinks([sink_id]).await.remove(&sink_id)
    }

    /// Drops a batch of compute sinks.
    ///
    /// For each sink that exists, the coordinator and controller's state
    /// associated with the sink is removed.
    ///
    /// Returns a map from sink id to the controller's state for the sink and a notify that
    /// resolves once the sink's `mz_subscriptions` retraction is durable (see
    /// `remove_active_compute_sink`). It is the caller's responsibility to await the notify
    /// off the coordinator loop and then retire the returned sinks. Consider using
    /// `retire_compute_sinks` instead.
    #[must_use]
    pub async fn drop_compute_sinks(
        &mut self,
        sink_ids: impl IntoIterator<Item = GlobalId>,
    ) -> BTreeMap<GlobalId, (ActiveComputeSink, BuiltinTableAppendNotify)> {
        let mut by_id = BTreeMap::new();
        let mut by_cluster: BTreeMap<_, Vec<_>> = BTreeMap::new();
        for sink_id in sink_ids {
            let (sink, write_notify) = match self.remove_active_compute_sink(sink_id).await {
                None => {
                    // This can happen due to a race condition: an internal
                    // subscribe may be cleaned up via its own message while
                    // session disconnect cleanup is in progress. This is
                    // benign.
                    tracing::debug!(%sink_id, "drop_compute_sinks: sink already removed");
                    continue;
                }
                Some(entry) => entry,
            };

            by_cluster
                .entry(sink.cluster_id())
                .or_default()
                .push(sink_id);
            by_id.insert(sink_id, (sink, write_notify));
        }
        for (cluster_id, ids) in by_cluster {
            let compute = &mut self.controller.compute;
            // A cluster could have been dropped, so verify it exists.
            if compute.instance_exists(cluster_id) {
                compute
                    .drop_collections(cluster_id, ids)
                    .unwrap_or_terminate("cannot fail to drop collections");
            }
        }
        by_id
    }

    /// Retires a batch of sinks with disparate reasons for retirement.
    ///
    /// Each sink identified in `reasons` is dropped (see `drop_compute_sinks`),
    /// then retired with its corresponding reason. Returns a notify that resolves
    /// once all `mz_subscriptions` retractions are durable and the sinks are retired.
    pub async fn retire_compute_sinks(
        &mut self,
        mut reasons: BTreeMap<GlobalId, ActiveComputeSinkRetireReason>,
    ) -> BuiltinTableAppendCompletion {
        let sink_ids = reasons.keys().cloned();
        let to_retire: Vec<_> = self
            .drop_compute_sinks(sink_ids)
            .await
            .into_iter()
            .map(|(id, (sink, write_notify))| {
                let reason = reasons
                    .remove(&id)
                    .expect("all returned IDs are in `reasons`");
                (sink, write_notify, reason)
            })
            .collect();

        if to_retire.is_empty() {
            return BuiltinTableAppendCompletion::completed();
        }

        // Retire off the coordinator loop. We wait for each `mz_subscriptions` retraction
        // before telling the subscribing client that the sink is gone. The returned notify
        // lets statements that caused the retirement also wait before sending their response.
        // The wait must not happen on the coordinator loop, since that would block every
        // other session on the group-commit oracle round trip.
        let (done_tx, done_rx) = tokio::sync::oneshot::channel();
        task::spawn(|| "retire_compute_sinks", async move {
            for (sink, write_notify, reason) in to_retire {
                write_notify.await;
                sink.retire(reason);
            }
            let _ = done_tx.send(());
        });
        BuiltinTableAppendCompletion::new(Box::pin(async move {
            let _ = done_rx.await;
        }))
    }

    /// Cancels all active compute sinks for the identified connection.
    #[mz_ore::instrument(level = "debug")]
    pub(crate) async fn cancel_compute_sinks_for_conn(
        &mut self,
        conn_id: &ConnectionId,
    ) -> BuiltinTableAppendCompletion {
        self.retire_compute_sinks_for_conn(conn_id, ActiveComputeSinkRetireReason::Canceled)
            .await
    }

    /// Retires all active compute sinks for the identified connection with the
    /// specified reason.
    #[mz_ore::instrument(level = "debug")]
    pub(crate) async fn retire_compute_sinks_for_conn(
        &mut self,
        conn_id: &ConnectionId,
        reason: ActiveComputeSinkRetireReason,
    ) -> BuiltinTableAppendCompletion {
        let drop_sinks = self
            .active_conns
            .get_mut(conn_id)
            .expect("must exist for active session")
            .drop_sinks
            .iter()
            .map(|sink_id| (*sink_id, reason.clone()))
            .collect();
        self.retire_compute_sinks(drop_sinks).await
    }

    pub(crate) fn drop_storage_sinks(&mut self, sink_gids: Vec<GlobalId>) {
        let storage_metadata = self.catalog.state().storage_metadata();
        self.controller
            .storage
            .drop_sinks(storage_metadata, sink_gids)
            .unwrap_or_terminate("cannot fail to drop sinks");
    }

    pub(crate) fn drop_compute_collections(&mut self, collections: Vec<(ClusterId, GlobalId)>) {
        let mut by_cluster: BTreeMap<_, Vec<_>> = BTreeMap::new();
        for (cluster_id, gid) in collections {
            by_cluster.entry(cluster_id).or_default().push(gid);
        }
        for (cluster_id, gids) in by_cluster {
            let compute = &mut self.controller.compute;
            // A cluster could have been dropped, so verify it exists.
            if compute.instance_exists(cluster_id) {
                compute
                    .drop_collections(cluster_id, gids)
                    .unwrap_or_terminate("cannot fail to drop collections");
            }
        }
    }

    pub(crate) fn drop_vpc_endpoints_in_background(&self, vpc_endpoints: Vec<CatalogItemId>) {
        // Match the create path (catalog_implications.rs) which gracefully
        // logs an error when cloud_resource_controller is None, rather than
        // panicking.
        let Some(cloud_resource_controller) = self.cloud_resource_controller.as_ref() else {
            warn!("dropping VPC endpoints without cloud_resource_controller; skipping cleanup");
            return;
        };
        let cloud_resource_controller = Arc::clone(cloud_resource_controller);
        // We don't want to block the coordinator on an external delete api
        // calls, so move the drop vpc_endpoint to a separate task. This does
        // mean that a failed drop won't bubble up to the user as an error
        // message. However, even if it did (and how the code previously
        // worked), mz has already dropped it from our catalog, and so we
        // wouldn't be able to retry anyway. Any orphaned vpc_endpoints will
        // eventually be cleaned during restart via coord bootstrap.
        task::spawn(
            || "drop_vpc_endpoints",
            async move {
                for vpc_endpoint in vpc_endpoints {
                    let _ = Retry::default()
                        .max_duration(Duration::from_secs(60))
                        .retry_async(|_state| async {
                            fail_point!("drop_vpc_endpoint", |r| {
                                Err(anyhow::anyhow!("Fail point error {:?}", r))
                            });
                            match cloud_resource_controller
                                .delete_vpc_endpoint(vpc_endpoint)
                                .await
                            {
                                Ok(_) => Ok(()),
                                Err(e) => {
                                    warn!("Dropping VPC Endpoints has encountered an error: {}", e);
                                    Err(e)
                                }
                            }
                        })
                        .await;
                }
            }
            .instrument(info_span!(
                "coord::catalog_transact_inner::drop_vpc_endpoints"
            )),
        );
    }

    /// Removes all temporary items created by the specified connection, though
    /// not the temporary schema itself.
    pub(crate) async fn drop_temp_items(&mut self, conn_id: &ConnectionId) {
        let temp_items = self.catalog().state().get_temp_items(conn_id).collect();
        let all_items = self.catalog().object_dependents(&temp_items, conn_id);

        if all_items.is_empty() {
            return;
        }
        let op = Op::DropObjects(
            all_items
                .into_iter()
                .map(DropObjectInfo::manual_drop_from_object_id)
                .collect(),
        );

        self.catalog_transact_with_context(Some(conn_id), None, vec![op])
            .await
            .expect("unable to drop temporary items for conn_id");
    }

    fn update_cluster_scheduling_config(&self) {
        let config = flags::orchestrator_scheduling_config(self.catalog.system_config());
        self.controller
            .update_orchestrator_scheduling_config(config);
    }

    fn update_secrets_caching_config(&self) {
        let config = flags::caching_config(self.catalog.system_config());
        self.caching_secrets_reader.set_policy(config);
    }

    fn update_tracing_config(&self) {
        let tracing = flags::tracing_config(self.catalog().system_config());
        tracing.apply(&self.tracing_handle);
    }

    fn update_compute_config(&mut self) {
        let config_params = flags::compute_config(self.catalog().system_config());
        self.controller.compute.update_configuration(config_params);
    }

    fn update_storage_config(&mut self) {
        let config_params = flags::storage_config(self.catalog().system_config());
        self.controller.storage.update_parameters(config_params);
    }

    fn update_timestamp_oracle_config(&self) {
        let config_params = flags::timestamp_oracle_config(self.catalog().system_config());
        if let Some(config) = self.timestamp_oracle_config.as_ref() {
            config.apply_parameters(config_params)
        }
    }

    fn update_metrics_retention(&self) {
        let duration = self.catalog().system_config().metrics_retention();
        let policy = ReadPolicy::lag_writes_by(
            Timestamp::new(u64::try_from(duration.as_millis()).unwrap_or_else(|_e| {
                tracing::error!("Absurd metrics retention duration: {duration:?}.");
                u64::MAX
            })),
            SINCE_GRANULARITY,
        );
        let storage_policies = self
            .catalog()
            .entries()
            .filter(|entry| {
                entry.item().is_retained_metrics_object()
                    && entry.item().is_compute_object_on_cluster().is_none()
            })
            .map(|entry| (entry.id(), policy.clone()))
            .collect::<Vec<_>>();
        let compute_policies = self
            .catalog()
            .entries()
            .filter_map(|entry| {
                if let (true, Some(cluster_id)) = (
                    entry.item().is_retained_metrics_object(),
                    entry.item().is_compute_object_on_cluster(),
                ) {
                    Some((cluster_id, entry.id(), policy.clone()))
                } else {
                    None
                }
            })
            .collect::<Vec<_>>();
        self.update_storage_read_policies(storage_policies);
        self.update_compute_read_policies(compute_policies);
    }

    fn update_controller_config(&mut self) {
        let sys_config = self.catalog().system_config();
        self.controller
            .update_configuration(sys_config.dyncfg_updates());
    }

    fn update_http_config(&mut self) {
        let webhook_request_limit = self
            .catalog()
            .system_config()
            .webhook_concurrent_request_limit();
        self.webhook_concurrency_limit
            .set_limit(webhook_request_limit);
    }

    pub(crate) async fn create_storage_export(
        &mut self,
        id: GlobalId,
        sink: &Sink,
    ) -> Result<(), AdapterError> {
        // Validate `sink.from` is in fact a storage collection
        self.controller.storage.check_exists(sink.from)?;

        // The AsOf is used to determine at what time to snapshot reading from
        // the persist collection.  This is primarily relevant when we do _not_
        // want to include the snapshot in the sink.
        //
        // We choose the smallest as_of that is legal, according to the sinked
        // collection's since.
        let id_bundle = crate::CollectionIdBundle {
            storage_ids: btreeset! {sink.from},
            compute_ids: btreemap! {},
        };

        // We're putting in place read holds, such that create_exports, below,
        // which calls update_read_capabilities, can successfully do so.
        // Otherwise, the since of dependencies might move along concurrently,
        // pulling the rug from under us!
        //
        // TODO: Maybe in the future, pass those holds on to storage, to hold on
        // to them and downgrade when possible?
        let read_holds = self.acquire_read_holds(&id_bundle);
        let as_of = read_holds.least_valid_read();

        let storage_sink_from_entry = self.catalog().get_entry_by_global_id(&sink.from);
        let storage_sink_desc = mz_storage_types::sinks::StorageSinkDesc {
            from: sink.from,
            from_desc: storage_sink_from_entry
                .relation_desc()
                .expect("sinks can only be built on items with descs")
                .into_owned(),
            connection: sink
                .connection
                .clone()
                .into_inline_connection(self.catalog().state()),
            envelope: sink.envelope,
            as_of,
            with_snapshot: sink.with_snapshot,
            version: sink.version,
            from_storage_metadata: (),
            to_storage_metadata: (),
            commit_interval: sink.commit_interval,
        };

        let collection_desc = CollectionDescription {
            // TODO(sinks): make generic once we have more than one sink type.
            desc: KAFKA_PROGRESS_DESC.clone(),
            data_source: DataSource::Sink {
                desc: ExportDescription {
                    sink: storage_sink_desc,
                    instance_id: sink.cluster_id,
                },
            },
            since: None,
            timeline: None,
            primary: None,
        };
        let collections = vec![(id, collection_desc)];

        // Create the collections.
        let storage_metadata = self.catalog.state().storage_metadata();
        let res = self
            .controller
            .storage
            .create_collections(storage_metadata, None, collections)
            .await;

        // Drop read holds after the export has been created, at which point
        // storage will have put in its own read holds.
        drop(read_holds);

        Ok(res?)
    }

    pub(crate) fn validate_resource_limit<F>(
        &self,
        current_amount: usize,
        new_instances: i64,
        resource_limit: F,
        resource_type: &str,
        limit_name: &str,
    ) -> Result<(), AdapterError>
    where
        F: Fn(&SystemVars) -> u32,
    {
        self.catalog().state().validate_resource_limit(
            current_amount,
            new_instances,
            resource_limit,
            resource_type,
            limit_name,
        )
    }

    pub(crate) fn validate_resource_limit_numeric<F>(
        &self,
        current_amount: Numeric,
        new_amount: Numeric,
        resource_limit: F,
        resource_type: &str,
        limit_name: &str,
    ) -> Result<(), AdapterError>
    where
        F: Fn(&SystemVars) -> Numeric,
    {
        self.catalog().state().validate_resource_limit_numeric(
            current_amount,
            new_amount,
            resource_limit,
            resource_type,
            limit_name,
        )
    }
}