mz_adapter/coord/

catalog_implications.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.

//! Logic related to deriving and applying implications from [catalog
//! changes](ParsedStateUpdate).
//!
//! The flow from "raw" catalog changes to [CatalogImplication] works like this:
//!
//! StateUpdateKind -> ParsedStateUpdate -> CatalogImplication
//!
//! [ParsedStateUpdate] adds context to a "raw" catalog change
//! ([StateUpdateKind](mz_catalog::memory::objects::StateUpdateKind)). It
//! includes an in-memory representation of the updated object, which can in
//! theory be derived from the raw change but only when we have access to all
//! the other raw changes or to an in-memory Catalog, which represents a
//! "rollup" of all the raw changes.
//!
//! [CatalogImplication] is both the state machine that we use for absorbing
//! multiple state updates for the same object and the final command that has to
//! be applied to in-memory state and or the controller(s) after absorbing all
//! the state updates in a given batch of updates.

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

use fail::fail_point;
use itertools::Itertools;
use mz_adapter_types::compaction::CompactionWindow;
use mz_catalog::memory::objects::{
    CatalogItem, Cluster, ClusterReplica, Connection, DataSourceDesc, Index, MaterializedView,
    Secret, Sink, Source, StateDiff, Table, TableDataSource, View,
};
use mz_cloud_resources::VpcEndpointConfig;
use mz_compute_client::logging::LogVariant;
use mz_compute_client::protocol::response::PeekResponse;
use mz_controller::clusters::{ClusterRole, ReplicaConfig};
use mz_controller_types::{ClusterId, ReplicaId};
use mz_ore::collections::CollectionExt;
use mz_ore::error::ErrorExt;
use mz_ore::future::InTask;
use mz_ore::instrument;
use mz_ore::retry::Retry;
use mz_ore::task;
use mz_repr::{CatalogItemId, GlobalId, RelationVersion, RelationVersionSelector};
use mz_sql::plan::ConnectionDetails;
use mz_storage_client::controller::{CollectionDescription, DataSource};
use mz_storage_types::connections::PostgresConnection;
use mz_storage_types::connections::inline::{InlinedConnection, IntoInlineConnection};
use mz_storage_types::sinks::StorageSinkConnection;
use mz_storage_types::sources::{
    GenericSourceConnection, SourceDesc, SourceExport, SourceExportDataConfig,
};
use tracing::{Instrument, info_span, warn};

use crate::active_compute_sink::ActiveComputeSinkRetireReason;
use crate::coord::Coordinator;
use crate::coord::catalog_implications::parsed_state_updates::{
    ParsedStateUpdate, ParsedStateUpdateKind,
};
use crate::coord::peek::DroppedDependency;
use crate::coord::timeline::TimelineState;
use crate::statement_logging::{StatementEndedExecutionReason, StatementLoggingId};
use crate::{AdapterError, CollectionIdBundle, ExecuteContext, ResultExt};

pub mod parsed_state_updates;

impl Coordinator {
    /// Applies implications from the given bucket of [ParsedStateUpdate] to our
    /// in-memory state and our controllers. This also applies transitive
    /// implications, for example, peeks and subscribes will be cancelled when
    /// referenced objects are dropped.
    ///
    /// This _requires_ that the given updates are consolidated. There must be
    /// at most one addition and/or one retraction for a given item, as
    /// identified by that items ID type.
    #[instrument(level = "debug")]
    pub async fn apply_catalog_implications(
        &mut self,
        ctx: Option<&mut ExecuteContext>,
        catalog_updates: Vec<ParsedStateUpdate>,
    ) -> Result<(), AdapterError> {
        let start = Instant::now();

        let mut catalog_implications: BTreeMap<CatalogItemId, CatalogImplication> = BTreeMap::new();
        let mut cluster_commands: BTreeMap<ClusterId, CatalogImplication> = BTreeMap::new();
        let mut cluster_replica_commands: BTreeMap<(ClusterId, ReplicaId), CatalogImplication> =
            BTreeMap::new();
        // Introspection source index additions, collected separately and
        // merged into the AddCluster handler. Not routed through the
        // absorb machinery since they are simple additions that don't
        // need Altered support.
        let mut introspection_source_indexes: BTreeMap<ClusterId, BTreeMap<LogVariant, GlobalId>> =
            BTreeMap::new();
        // Whether any replica-scoped system-parameter override changed in this
        // batch. The push re-pushes the complete per-replica dyncfg layer, so we
        // only track that a change happened, not the individual rows.
        let mut replica_scoped_config_changed = false;

        for update in catalog_updates {
            tracing::trace!(?update, "got parsed state update");
            match &update.kind {
                ParsedStateUpdateKind::Item {
                    durable_item,
                    parsed_item: _,
                    connection: _,
                    parsed_full_name: _,
                } => {
                    let entry = catalog_implications
                        .entry(durable_item.id.clone())
                        .or_insert_with(|| CatalogImplication::None);
                    entry.absorb(update);
                }
                ParsedStateUpdateKind::TemporaryItem {
                    durable_item,
                    parsed_item: _,
                    connection: _,
                    parsed_full_name: _,
                } => {
                    let entry = catalog_implications
                        .entry(durable_item.id.clone())
                        .or_insert_with(|| CatalogImplication::None);
                    entry.absorb(update);
                }
                ParsedStateUpdateKind::Cluster {
                    durable_cluster,
                    parsed_cluster: _,
                } => {
                    let entry = cluster_commands
                        .entry(durable_cluster.id)
                        .or_insert_with(|| CatalogImplication::None);
                    entry.absorb(update.clone());
                }
                ParsedStateUpdateKind::ClusterReplica {
                    durable_cluster_replica,
                    parsed_cluster_replica: _,
                } => {
                    let entry = cluster_replica_commands
                        .entry((
                            durable_cluster_replica.cluster_id,
                            durable_cluster_replica.replica_id,
                        ))
                        .or_insert_with(|| CatalogImplication::None);
                    entry.absorb(update.clone());
                }
                ParsedStateUpdateKind::IntrospectionSourceIndex {
                    cluster_id,
                    log,
                    index_id,
                } => {
                    if update.diff == StateDiff::Addition {
                        introspection_source_indexes
                            .entry(*cluster_id)
                            .or_default()
                            .insert(log.clone(), *index_id);
                    }
                    // Retractions don't need handling: introspection
                    // source indexes are dropped with their cluster.
                }
                ParsedStateUpdateKind::ReplicaSystemConfiguration { durable: _ } => {
                    // Additions and retractions both re-derive the full
                    // per-replica layer from the working copy, so the diff sign
                    // does not matter here.
                    replica_scoped_config_changed = true;
                }
            }
        }

        self.apply_catalog_implications_inner(
            ctx,
            catalog_implications.into_iter().collect_vec(),
            cluster_commands.into_iter().collect_vec(),
            cluster_replica_commands.into_iter().collect_vec(),
            introspection_source_indexes,
            replica_scoped_config_changed,
        )
        .await?;

        self.metrics
            .apply_catalog_implications_seconds
            .observe(start.elapsed().as_secs_f64());

        Ok(())
    }

    #[instrument(level = "debug")]
    async fn apply_catalog_implications_inner(
        &mut self,
        ctx: Option<&mut ExecuteContext>,
        implications: Vec<(CatalogItemId, CatalogImplication)>,
        cluster_commands: Vec<(ClusterId, CatalogImplication)>,
        cluster_replica_commands: Vec<((ClusterId, ReplicaId), CatalogImplication)>,
        mut introspection_source_indexes: BTreeMap<ClusterId, BTreeMap<LogVariant, GlobalId>>,
        replica_scoped_config_changed: bool,
    ) -> Result<(), AdapterError> {
        let mut tables_to_drop = BTreeSet::new();
        let mut sources_to_drop = vec![];
        let mut replication_slots_to_drop: Vec<(PostgresConnection, String)> = vec![];
        let mut storage_sink_gids_to_drop = vec![];
        let mut indexes_to_drop = vec![];
        let mut compute_sinks_to_drop = vec![];
        let mut view_gids_to_drop = vec![];
        let mut secrets_to_drop = vec![];
        let mut vpc_endpoints_to_drop = vec![];
        let mut clusters_to_drop = vec![];
        let mut cluster_replicas_to_drop = vec![];
        let mut active_compute_sinks_to_drop = BTreeMap::new();
        let mut peeks_to_drop = vec![];
        let mut copies_to_drop = vec![];

        // Maps for storing names of dropped objects for error messages.
        let mut dropped_item_names: BTreeMap<GlobalId, String> = BTreeMap::new();
        let mut dropped_cluster_names: BTreeMap<ClusterId, String> = BTreeMap::new();

        // Separate collections for tables (which need write timestamps) and
        // sources (which don't).
        let mut table_collections_to_create = BTreeMap::new();
        let mut source_collections_to_create = BTreeMap::new();
        let mut storage_policies_to_initialize = BTreeMap::new();
        let mut execution_timestamps_to_set = BTreeSet::new();
        let mut vpc_endpoints_to_create: Vec<(CatalogItemId, VpcEndpointConfig)> = vec![];

        // Sources that shouldn't be dropped, even if we saw a `Dropped` event.
        // Used for correct handling of ALTER MV.
        let mut source_gids_to_keep = BTreeSet::new();

        // Collections for batching connection-related alterations.
        let mut source_connections_to_alter: BTreeMap<
            GlobalId,
            GenericSourceConnection<InlinedConnection>,
        > = BTreeMap::new();
        let mut sink_connections_to_alter: BTreeMap<GlobalId, StorageSinkConnection> =
            BTreeMap::new();
        let mut source_export_data_configs_to_alter: BTreeMap<GlobalId, SourceExportDataConfig> =
            BTreeMap::new();
        let mut source_descs_to_alter: BTreeMap<GlobalId, SourceDesc> = BTreeMap::new();

        // We're incrementally migrating the code that manipulates the
        // controller from closures in the sequencer. For some types of catalog
        // changes we haven't done this migration yet, so there you will see
        // just a log message. Over the next couple of PRs all of these will go
        // away.

        for (catalog_id, implication) in implications {
            tracing::trace!(?implication, "have to apply catalog implication");

            match implication {
                CatalogImplication::Table(CatalogImplicationKind::Added(table)) => {
                    self.handle_create_table(
                        &ctx,
                        &mut table_collections_to_create,
                        &mut storage_policies_to_initialize,
                        &mut execution_timestamps_to_set,
                        catalog_id,
                        table.clone(),
                    )
                    .await?
                }
                CatalogImplication::Table(CatalogImplicationKind::Altered {
                    prev: prev_table,
                    new: new_table,
                }) => {
                    self.handle_alter_table(catalog_id, prev_table, new_table)
                        .await?
                }

                CatalogImplication::Table(CatalogImplicationKind::Dropped(table, full_name)) => {
                    let global_ids = table.global_ids();
                    for global_id in global_ids {
                        tables_to_drop.insert((catalog_id, global_id));
                        dropped_item_names.insert(global_id, full_name.clone());
                    }
                }
                CatalogImplication::Source(CatalogImplicationKind::Added((
                    source,
                    _connection,
                ))) => {
                    // Get the compaction windows for all sources with this
                    // catalog_id This replicates the logic from
                    // sequence_create_source where it collects all item_ids and
                    // gets their compaction windows
                    let compaction_windows = self
                        .catalog()
                        .state()
                        .source_compaction_windows(vec![catalog_id]);

                    self.handle_create_source(
                        &mut source_collections_to_create,
                        &mut storage_policies_to_initialize,
                        catalog_id,
                        source,
                        compaction_windows,
                    )
                    .await?
                }
                CatalogImplication::Source(CatalogImplicationKind::Altered {
                    prev: (prev_source, _prev_connection),
                    new: (new_source, new_connection),
                }) => {
                    if prev_source.custom_logical_compaction_window
                        != new_source.custom_logical_compaction_window
                    {
                        let new_window = new_source
                            .custom_logical_compaction_window
                            .unwrap_or(CompactionWindow::Default);
                        self.update_storage_read_policies(vec![(catalog_id, new_window.into())]);
                    }
                    match (&prev_source.data_source, &new_source.data_source) {
                        (
                            DataSourceDesc::Ingestion {
                                desc: prev_desc, ..
                            }
                            | DataSourceDesc::OldSyntaxIngestion {
                                desc: prev_desc, ..
                            },
                            DataSourceDesc::Ingestion { desc: new_desc, .. }
                            | DataSourceDesc::OldSyntaxIngestion { desc: new_desc, .. },
                        ) => {
                            if prev_desc != new_desc {
                                let inlined_connection = new_connection
                                    .expect("ingestion source should have inlined connection");
                                let inlined_desc = SourceDesc {
                                    connection: inlined_connection,
                                    timestamp_interval: new_desc.timestamp_interval,
                                };
                                source_descs_to_alter.insert(new_source.global_id, inlined_desc);
                            }
                        }
                        _ => {}
                    }
                }
                CatalogImplication::Source(CatalogImplicationKind::Dropped(
                    (source, connection),
                    full_name,
                )) => {
                    let global_id = source.global_id();
                    sources_to_drop.push((catalog_id, global_id));
                    dropped_item_names.insert(global_id, full_name);

                    if let DataSourceDesc::Ingestion { desc, .. }
                    | DataSourceDesc::OldSyntaxIngestion { desc, .. } = &source.data_source
                    {
                        match &desc.connection {
                            GenericSourceConnection::Postgres(_referenced_conn) => {
                                let inline_conn = connection.expect("missing inlined connection");

                                let pg_conn = match inline_conn {
                                    GenericSourceConnection::Postgres(pg_conn) => pg_conn,
                                    other => {
                                        panic!("expected postgres connection, got: {:?}", other)
                                    }
                                };
                                let pending_drop = (
                                    pg_conn.connection.clone(),
                                    pg_conn.publication_details.slot.clone(),
                                );
                                replication_slots_to_drop.push(pending_drop);
                            }
                            _ => {}
                        }
                    }
                }
                CatalogImplication::Sink(CatalogImplicationKind::Added(sink)) => {
                    tracing::debug!(?sink, "not handling AddSink in here yet");
                }
                CatalogImplication::Sink(CatalogImplicationKind::Altered {
                    prev: prev_sink,
                    new: new_sink,
                }) => {
                    tracing::debug!(?prev_sink, ?new_sink, "not handling AlterSink in here yet");
                }
                CatalogImplication::Sink(CatalogImplicationKind::Dropped(sink, full_name)) => {
                    storage_sink_gids_to_drop.push(sink.global_id());
                    dropped_item_names.insert(sink.global_id(), full_name);
                }
                CatalogImplication::Index(CatalogImplicationKind::Added(index)) => {
                    tracing::debug!(?index, "not handling AddIndex in here yet");
                }
                CatalogImplication::Index(CatalogImplicationKind::Altered {
                    prev: prev_index,
                    new: new_index,
                }) => {
                    if prev_index.custom_logical_compaction_window
                        != new_index.custom_logical_compaction_window
                    {
                        let new_window = new_index
                            .custom_logical_compaction_window
                            .unwrap_or(CompactionWindow::Default);
                        self.update_compute_read_policy(
                            new_index.cluster_id,
                            catalog_id,
                            new_window.into(),
                        );
                    }
                }
                CatalogImplication::Index(CatalogImplicationKind::Dropped(index, full_name)) => {
                    indexes_to_drop.push((index.cluster_id, index.global_id()));
                    dropped_item_names.insert(index.global_id(), full_name);
                }
                CatalogImplication::MaterializedView(CatalogImplicationKind::Added(mv)) => {
                    tracing::debug!(?mv, "not handling AddMaterializedView in here yet");
                }
                CatalogImplication::MaterializedView(CatalogImplicationKind::Altered {
                    prev: prev_mv,
                    new: new_mv,
                }) => {
                    // We get here for three reasons:
                    //  1. Name changes, like those caused by ALTER SCHEMA.
                    //  2. Replacement application.
                    //  3. Compaction window changes (ALTER ... SET (RETAIN HISTORY ...)).
                    //
                    // 1. Name changes: We don't have to do anything here.
                    //
                    // 2. Replacement application: This is tricky: It changes the `CatalogItemId` of
                    // the target to that of the replacement and simultaneously drops the replacement.
                    // Which means when we get here `prev_mv` is the replacement that should be
                    // dropped, and `new_mv` is the target that already exists but under a different
                    // ID (which will receive a `Dropped` event separately). We can sniff out this
                    // case by checking for version differences.
                    //
                    // 3. Compaction window changes: We handle this in an `else if`, because if there
                    // is also a replacement application, then the replacement's storage collections
                    // already have the correct read policies from when they were created, so we
                    // don't need to update them here.
                    if prev_mv.collections != new_mv.collections {
                        // Sanity check: The replacement's last (and only) version must be the same
                        // as the new target's last version.
                        assert_eq!(
                            prev_mv.global_id_writes(),
                            new_mv.global_id_writes(),
                            "unexpected MV Altered implication: prev={prev_mv:?}, new={new_mv:?}",
                        );

                        let gid = new_mv.global_id_writes();
                        self.allow_writes(new_mv.cluster_id, gid);

                        // There will be a separate `Dropped` implication for the old definition of
                        // the target MV. That will drop the old compute collection, as we desire,
                        // but we need to prevent it from dropping the old storage collection as
                        // well, since that might still be depended on.
                        source_gids_to_keep.extend(new_mv.global_ids());
                    } else if prev_mv.custom_logical_compaction_window
                        != new_mv.custom_logical_compaction_window
                    {
                        let new_window = new_mv
                            .custom_logical_compaction_window
                            .unwrap_or(CompactionWindow::Default);
                        self.update_storage_read_policies(vec![(catalog_id, new_window.into())]);
                    }
                }
                CatalogImplication::MaterializedView(CatalogImplicationKind::Dropped(
                    mv,
                    full_name,
                )) => {
                    compute_sinks_to_drop.push((mv.cluster_id, mv.global_id_writes()));
                    for gid in mv.global_ids() {
                        sources_to_drop.push((catalog_id, gid));
                        dropped_item_names.insert(gid, full_name.clone());
                    }
                }
                CatalogImplication::View(CatalogImplicationKind::Added(_view)) => {
                    // No action needed: views are catalog-only objects with no
                    // storage collections or dataflows to create.
                }
                CatalogImplication::View(CatalogImplicationKind::Altered {
                    prev: _prev_view,
                    new: _new_view,
                }) => {
                    // No action needed: view alterations (e.g. renames) are
                    // catalog-only and require no controller changes.
                }
                CatalogImplication::View(CatalogImplicationKind::Dropped(view, full_name)) => {
                    view_gids_to_drop.push(view.global_id());
                    dropped_item_names.insert(view.global_id(), full_name);
                }
                CatalogImplication::Secret(CatalogImplicationKind::Added(_secret)) => {
                    // No action needed: the secret payload is stored in
                    // secrets_controller.ensure() BEFORE the catalog transaction.
                    // By the time we see this update, the secret is already stored.
                }
                CatalogImplication::Secret(CatalogImplicationKind::Altered {
                    prev: _prev_secret,
                    new: _new_secret,
                }) => {
                    // No action needed: altering a secret updates the payload via
                    // secrets_controller.ensure() without a catalog transaction.
                }
                CatalogImplication::Secret(CatalogImplicationKind::Dropped(
                    _secret,
                    _full_name,
                )) => {
                    secrets_to_drop.push(catalog_id);
                }
                CatalogImplication::Connection(CatalogImplicationKind::Added(connection)) => {
                    match &connection.details {
                        // SSH connections: key pair is stored in secrets_controller
                        // BEFORE the catalog transaction, so no action needed here.
                        ConnectionDetails::Ssh { .. } => {}
                        // AWS PrivateLink connections: create the VPC endpoint
                        ConnectionDetails::AwsPrivatelink(privatelink) => {
                            let spec = VpcEndpointConfig {
                                aws_service_name: privatelink.service_name.to_owned(),
                                availability_zone_ids: privatelink.availability_zones.to_owned(),
                            };
                            vpc_endpoints_to_create.push((catalog_id, spec));
                        }
                        // Other connection types don't require post-transaction actions
                        _ => {}
                    }
                }
                CatalogImplication::Connection(CatalogImplicationKind::Altered {
                    prev: _prev_connection,
                    new: new_connection,
                }) => {
                    self.handle_alter_connection(
                        catalog_id,
                        new_connection,
                        &mut vpc_endpoints_to_create,
                        &mut source_connections_to_alter,
                        &mut sink_connections_to_alter,
                        &mut source_export_data_configs_to_alter,
                    );
                }
                CatalogImplication::Connection(CatalogImplicationKind::Dropped(
                    connection,
                    _full_name,
                )) => {
                    match &connection.details {
                        // SSH connections have an associated secret that should be dropped
                        ConnectionDetails::Ssh { .. } => {
                            secrets_to_drop.push(catalog_id);
                        }
                        // AWS PrivateLink connections have an associated
                        // VpcEndpoint K8S resource that should be dropped
                        ConnectionDetails::AwsPrivatelink(_) => {
                            vpc_endpoints_to_drop.push(catalog_id);
                        }
                        _ => (),
                    }
                }
                CatalogImplication::None => {
                    // Nothing to do for None commands
                }
                CatalogImplication::Cluster(_) | CatalogImplication::ClusterReplica(_) => {
                    unreachable!("clusters and cluster replicas are handled below")
                }
                CatalogImplication::Table(CatalogImplicationKind::None)
                | CatalogImplication::Source(CatalogImplicationKind::None)
                | CatalogImplication::Sink(CatalogImplicationKind::None)
                | CatalogImplication::Index(CatalogImplicationKind::None)
                | CatalogImplication::MaterializedView(CatalogImplicationKind::None)
                | CatalogImplication::View(CatalogImplicationKind::None)
                | CatalogImplication::Secret(CatalogImplicationKind::None)
                | CatalogImplication::Connection(CatalogImplicationKind::None) => {
                    unreachable!("will never leave None in place");
                }
            }
        }

        for (cluster_id, command) in cluster_commands {
            tracing::trace!(?command, "have cluster command to apply!");

            match command {
                CatalogImplication::Cluster(CatalogImplicationKind::Added(cluster)) => {
                    // The Cluster's log_indexes is empty at parse time
                    // because IntrospectionSourceIndex updates are applied
                    // after the Cluster update. Use the separately collected
                    // introspection_source_indexes instead.
                    let arranged_logs = introspection_source_indexes
                        .remove(&cluster_id)
                        .unwrap_or_default();
                    let introspection_source_ids: Vec<_> =
                        arranged_logs.values().copied().collect();

                    self.controller
                        .create_cluster(
                            cluster_id,
                            mz_controller::clusters::ClusterConfig {
                                arranged_logs,
                                workload_class: cluster.config.workload_class.clone(),
                            },
                        )
                        .expect("creating cluster must not fail");

                    if !introspection_source_ids.is_empty() {
                        self.initialize_compute_read_policies(
                            introspection_source_ids,
                            cluster_id,
                            CompactionWindow::Default,
                        )
                        .await;
                    }
                }
                CatalogImplication::Cluster(CatalogImplicationKind::Altered {
                    prev: prev_cluster,
                    new: new_cluster,
                }) => {
                    // Replica adds/drops/renames from config changes arrive as
                    // separate AddClusterReplica/DroppedClusterReplica/
                    // AlterClusterReplica events, so the only cluster-level
                    // side effect here is updating the workload class on the
                    // controller when it changes.
                    if prev_cluster.config.workload_class != new_cluster.config.workload_class {
                        self.controller.update_cluster_workload_class(
                            cluster_id,
                            new_cluster.config.workload_class.clone(),
                        );
                    }
                }
                CatalogImplication::Cluster(CatalogImplicationKind::Dropped(
                    cluster,
                    _full_name,
                )) => {
                    clusters_to_drop.push(cluster_id);
                    dropped_cluster_names.insert(cluster_id, cluster.name);
                }
                CatalogImplication::Cluster(CatalogImplicationKind::None) => {
                    unreachable!("will never leave None in place");
                }
                command => {
                    unreachable!(
                        "we only handle cluster commands in this map, got: {:?}",
                        command
                    );
                }
            }
        }

        // Apply replica-scoped overrides after clusters are created (so their
        // compute instances exist) but before replicas are created below. The
        // override layer must be set before `create_replica`, so the new
        // replica's first configuration replays with its override. The push
        // reads the catalog working copy, which already reflects this
        // transaction's scoped-config changes.
        if replica_scoped_config_changed {
            self.push_replica_dyncfg_overrides();
        }

        for ((cluster_id, replica_id), command) in cluster_replica_commands {
            tracing::trace!(?command, "have cluster replica command to apply!");

            match command {
                CatalogImplication::ClusterReplica(CatalogImplicationKind::Added(replica)) => {
                    // Read the cluster name and role from the current catalog
                    // state. This is correct as long as implications are
                    // processed right after each catalog transaction. For a
                    // more future-proof approach that tracks cluster info
                    // locally across transactions, see the last commit of
                    // https://github.com/ggevay/materialize/tree/implications-cluster-name-tracking
                    // which removes that logic.
                    let cluster = self.catalog().get_cluster(cluster_id);
                    let cluster_name = cluster.name.clone();
                    let cluster_role = cluster.role();
                    self.handle_create_cluster_replica(
                        cluster_id,
                        replica_id,
                        cluster_role,
                        cluster_name,
                        replica.name.clone(),
                        replica.config.clone(),
                    )
                    .await;
                }
                CatalogImplication::ClusterReplica(CatalogImplicationKind::Altered {
                    prev: _prev_replica,
                    new: _new_replica,
                }) => {
                    // No action needed: cluster replica alterations (e.g.
                    // renames, owner changes, pending flag changes) are
                    // catalog-only and require no controller changes.
                }
                CatalogImplication::ClusterReplica(CatalogImplicationKind::Dropped(
                    _replica,
                    _full_name,
                )) => {
                    cluster_replicas_to_drop.push((cluster_id, replica_id));
                }
                CatalogImplication::ClusterReplica(CatalogImplicationKind::None) => {
                    unreachable!("will never leave None in place");
                }
                command => {
                    unreachable!(
                        "we only handle cluster replica commands in this map, got: {:?}",
                        command
                    );
                }
            }
        }

        if !source_collections_to_create.is_empty() {
            self.create_source_collections(source_collections_to_create)
                .await?;
        }

        // Have to create sources first and then tables, because tables within
        // one transaction can depend on sources.
        if !table_collections_to_create.is_empty() {
            self.create_table_collections(table_collections_to_create, execution_timestamps_to_set)
                .await?;
        }
        // It is _very_ important that we only initialize read policies after we
        // have created all the sources/collections. Some of the sources created
        // in this collection might have dependencies on other sources, so the
        // controller must get a chance to install read holds before we set a
        // policy that might make the since advance.
        self.initialize_storage_collections(storage_policies_to_initialize)
            .await?;

        // Create VPC endpoints for AWS PrivateLink connections
        if !vpc_endpoints_to_create.is_empty() {
            if let Some(cloud_resource_controller) = self.cloud_resource_controller.as_ref() {
                for (connection_id, spec) in vpc_endpoints_to_create {
                    if let Err(err) = cloud_resource_controller
                        .ensure_vpc_endpoint(connection_id, spec)
                        .await
                    {
                        tracing::error!(?err, "failed to ensure vpc endpoint!");
                    }
                }
            } else {
                tracing::error!(
                    "AWS PrivateLink connections unsupported without cloud_resource_controller"
                );
            }
        }

        // Apply batched connection alterations to dependent sources/sinks/tables.
        if !source_connections_to_alter.is_empty() {
            self.controller
                .storage
                .alter_ingestion_connections(source_connections_to_alter)
                .await
                .unwrap_or_terminate("cannot fail to alter ingestion connections");
        }

        if !sink_connections_to_alter.is_empty() {
            self.controller
                .storage
                .alter_export_connections(sink_connections_to_alter)
                .await
                .unwrap_or_terminate("altering export connections after txn must succeed");
        }

        if !source_export_data_configs_to_alter.is_empty() {
            self.controller
                .storage
                .alter_ingestion_export_data_configs(source_export_data_configs_to_alter)
                .await
                .unwrap_or_terminate("altering source export data configs after txn must succeed");
        }

        if !source_descs_to_alter.is_empty() {
            self.controller
                .storage
                .alter_ingestion_source_desc(source_descs_to_alter)
                .await
                .unwrap_or_terminate("cannot fail to alter ingestion source desc");
        }

        // Apply source drop overwrites.
        sources_to_drop.retain(|(_, gid)| !source_gids_to_keep.contains(gid));

        let readable_collections_to_drop: BTreeSet<_> = sources_to_drop
            .iter()
            .map(|(_, gid)| *gid)
            .chain(tables_to_drop.iter().map(|(_, gid)| *gid))
            .chain(indexes_to_drop.iter().map(|(_, gid)| *gid))
            .chain(view_gids_to_drop.iter().copied())
            .collect();

        // Clean up any active compute sinks like subscribes or copy to-s that
        // rely on dropped relations or clusters.
        for (sink_id, sink) in &self.active_compute_sinks {
            let cluster_id = sink.cluster_id();
            if let Some(id) = sink
                .depends_on()
                .iter()
                .find(|id| readable_collections_to_drop.contains(id))
            {
                let name = dropped_item_names
                    .get(id)
                    .cloned()
                    .expect("missing relation name");
                active_compute_sinks_to_drop.insert(
                    *sink_id,
                    ActiveComputeSinkRetireReason::DependencyDropped(DroppedDependency::Relation {
                        name,
                    }),
                );
            } else if clusters_to_drop.contains(&cluster_id) {
                let name = dropped_cluster_names
                    .get(&cluster_id)
                    .cloned()
                    .expect("missing cluster name");
                active_compute_sinks_to_drop.insert(
                    *sink_id,
                    ActiveComputeSinkRetireReason::DependencyDropped(DroppedDependency::Cluster {
                        name,
                    }),
                );
            }
        }

        // Clean up any pending peeks that rely on dropped relations or clusters.
        for (uuid, pending_peek) in &self.pending_peeks {
            if let Some(id) = pending_peek
                .depends_on
                .iter()
                .find(|id| readable_collections_to_drop.contains(id))
            {
                let name = dropped_item_names
                    .get(id)
                    .cloned()
                    .expect("missing relation name");
                peeks_to_drop.push((DroppedDependency::Relation { name }, uuid.clone()));
            } else if clusters_to_drop.contains(&pending_peek.cluster_id) {
                let name = dropped_cluster_names
                    .get(&pending_peek.cluster_id)
                    .cloned()
                    .expect("missing cluster name");
                peeks_to_drop.push((DroppedDependency::Cluster { name }, uuid.clone()));
            }
        }

        // Clean up any pending `COPY` statements that rely on dropped relations or clusters.
        for (conn_id, pending_copy) in &self.active_copies {
            let dropping_table = tables_to_drop
                .iter()
                .any(|(item_id, _gid)| pending_copy.table_id == *item_id);
            let dropping_cluster = clusters_to_drop.contains(&pending_copy.cluster_id);

            if dropping_table || dropping_cluster {
                copies_to_drop.push(conn_id.clone());
            }
        }

        let storage_gids_to_drop: BTreeSet<_> = sources_to_drop
            .iter()
            .map(|(_id, gid)| gid)
            .chain(storage_sink_gids_to_drop.iter())
            .chain(tables_to_drop.iter().map(|(_id, gid)| gid))
            .copied()
            .collect();
        let compute_gids_to_drop: Vec<_> = indexes_to_drop
            .iter()
            .chain(compute_sinks_to_drop.iter())
            .copied()
            .collect();

        // Gather resources that we have to remove from timeline state and
        // pre-check if any Timelines become empty, when we drop the specified
        // storage and compute resources.
        //
        // Note: We only apply these changes below.
        let mut timeline_id_bundles = BTreeMap::new();

        for (timeline, TimelineState { read_holds, .. }) in &self.global_timelines {
            let mut id_bundle = CollectionIdBundle::default();

            for storage_id in read_holds.storage_ids() {
                if storage_gids_to_drop.contains(&storage_id) {
                    id_bundle.storage_ids.insert(storage_id);
                }
            }

            for (instance_id, id) in read_holds.compute_ids() {
                if compute_gids_to_drop.contains(&(instance_id, id))
                    || clusters_to_drop.contains(&instance_id)
                {
                    id_bundle
                        .compute_ids
                        .entry(instance_id)
                        .or_default()
                        .insert(id);
                }
            }

            timeline_id_bundles.insert(timeline.clone(), id_bundle);
        }

        let mut timeline_associations = BTreeMap::new();
        for (timeline, id_bundle) in timeline_id_bundles.into_iter() {
            let TimelineState { read_holds, .. } = self
                .global_timelines
                .get(&timeline)
                .expect("all timelines have a timestamp oracle");

            let empty = read_holds.id_bundle().difference(&id_bundle).is_empty();
            timeline_associations.insert(timeline, (empty, id_bundle));
        }

        // 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 !timeline_associations.is_empty() {
                for (timeline, (should_be_empty, id_bundle)) in timeline_associations {
                    let became_empty =
                        self.remove_resources_associated_with_timeline(timeline, id_bundle);
                    assert_eq!(should_be_empty, became_empty, "emptiness did not match!");
                }
            }

            // Note that we drop tables before sources since there can be a weak
            // dependency on sources from tables in the storage controller that
            // will result in error logging that we'd prefer to avoid. This
            // isn't an actual dependency issue but we'd like to keep that error
            // logging around to indicate when an actual dependency error might
            // occur.
            if !tables_to_drop.is_empty() {
                let ts = self.get_local_write_ts().await;
                self.drop_tables(tables_to_drop.into_iter().collect_vec(), ts.timestamp);
            }

            if !sources_to_drop.is_empty() {
                self.drop_sources(sources_to_drop);
            }

            if !storage_sink_gids_to_drop.is_empty() {
                self.drop_storage_sinks(storage_sink_gids_to_drop);
            }

            if !active_compute_sinks_to_drop.is_empty() {
                self.retire_compute_sinks(active_compute_sinks_to_drop)
                    .await;
            }

            if !peeks_to_drop.is_empty() {
                for (dep, uuid) in peeks_to_drop {
                    if let Some(pending_peek) = self.remove_pending_peek(&uuid) {
                        let cancel_reason = PeekResponse::Error(dep.query_terminated_error());
                        self.controller
                            .compute
                            .cancel_peek(pending_peek.cluster_id, uuid, cancel_reason)
                            .unwrap_or_terminate("unable to cancel peek");
                        self.retire_execution(
                            StatementEndedExecutionReason::Canceled,
                            pending_peek.ctx_extra.defuse(),
                        );
                    }
                }
            }

            if !copies_to_drop.is_empty() {
                for conn_id in copies_to_drop {
                    self.cancel_pending_copy(&conn_id);
                }
            }

            if !compute_gids_to_drop.is_empty() {
                self.drop_compute_collections(compute_gids_to_drop);
            }

            if !vpc_endpoints_to_drop.is_empty() {
                self.drop_vpc_endpoints_in_background(vpc_endpoints_to_drop)
            }

            if !cluster_replicas_to_drop.is_empty() {
                fail::fail_point!("after_catalog_drop_replica");

                for (cluster_id, replica_id) in cluster_replicas_to_drop {
                    self.drop_replica(cluster_id, replica_id);
                }
            }
            if !clusters_to_drop.is_empty() {
                for cluster_id in clusters_to_drop {
                    self.controller.drop_cluster(cluster_id);
                }
            }

            // We don't want to block the main coordinator thread on cleaning
            // up external resources (PostgreSQL replication slots and secrets),
            // so we perform that cleanup in a background task.
            //
            // TODO(14551): This is inherently best effort. An ill-timed crash
            // means we'll never clean these resources up. Safer cleanup for non-Materialize resources.
            // See <https://github.com/MaterializeInc/materialize/issues/14551>
            task::spawn(|| "drop_replication_slots_and_secrets", {
                let ssh_tunnel_manager = self.connection_context().ssh_tunnel_manager.clone();
                let caching_secrets_reader = self.caching_secrets_reader.clone();
                let secrets_controller = Arc::clone(&self.secrets_controller);
                let secrets_reader = Arc::clone(self.secrets_reader());
                let storage_config = self.controller.storage.config().clone();

                async move {
                    for (connection, replication_slot_name) in replication_slots_to_drop {
                        tracing::info!(?replication_slot_name, "dropping replication slot");

                        // Try to drop the replication slots, but give up after
                        // a while. The PostgreSQL server may no longer be
                        // healthy. Users often drop PostgreSQL sources
                        // *because* the PostgreSQL server has been
                        // decomissioned.
                        let result: Result<(), anyhow::Error> = Retry::default()
                            .max_duration(Duration::from_secs(60))
                            .retry_async(|_state| async {
                                let config = connection
                                    .config(&secrets_reader, &storage_config, InTask::No)
                                    .await
                                    .map_err(|e| {
                                        anyhow::anyhow!(
                                            "error creating Postgres client for \
                                            dropping acquired slots: {}",
                                            e.display_with_causes()
                                        )
                                    })?;

                                mz_postgres_util::drop_replication_slots(
                                    &ssh_tunnel_manager,
                                    config.clone(),
                                    &[(&replication_slot_name, true)],
                                )
                                .await?;

                                Ok(())
                            })
                            .await;

                        if let Err(err) = result {
                            tracing::warn!(
                                ?replication_slot_name,
                                ?err,
                                "failed to drop replication slot"
                            );
                        }
                    }

                    // Drop secrets *after* dropping the replication slots,
                    // because dropping replication slots may rely on those
                    // secrets still being present.
                    //
                    // It's okay if we crash before processing the secret drops,
                    // as we look for and remove any orphaned secrets during
                    // startup.
                    fail_point!("drop_secrets");
                    for secret in secrets_to_drop {
                        if let Err(e) = secrets_controller.delete(secret).await {
                            warn!("Dropping secrets has encountered an error: {}", e);
                        } else {
                            caching_secrets_reader.invalidate(secret);
                        }
                    }
                }
            });
        }
        .instrument(info_span!(
            "coord::apply_catalog_implications_inner::finalize"
        ))
        .await;

        Ok(())
    }

    #[instrument(level = "debug")]
    async fn create_table_collections(
        &mut self,
        table_collections_to_create: BTreeMap<GlobalId, CollectionDescription>,
        execution_timestamps_to_set: BTreeSet<StatementLoggingId>,
    ) -> Result<(), AdapterError> {
        // If we have tables, determine the initial validity for the table.
        let write_ts = self.get_local_write_ts().await;
        let register_ts = write_ts.timestamp;

        // After acquiring `register_ts` but before using it, we need to
        // be sure we're still the leader. Otherwise a new generation
        // may also be trying to use `register_ts` for a different
        // purpose. See materialize#28216.
        //
        // We also should advance the upper of the catalog shard, to ensure it
        // is readable at the oracle read ts after we bump it to the
        // `register_ts` below. Both of these needs are served by calling
        // `advance_upper`.
        self.catalog
            .advance_upper(write_ts.advance_to)
            .await
            .unwrap_or_terminate("unable to advance catalog upper");

        for id in execution_timestamps_to_set {
            self.set_statement_execution_timestamp(id, register_ts);
        }

        let storage_metadata = self.catalog.state().storage_metadata();

        self.controller
            .storage
            .create_collections(
                storage_metadata,
                Some(register_ts),
                table_collections_to_create.into_iter().collect_vec(),
            )
            .await
            .unwrap_or_terminate("cannot fail to create collections");

        self.apply_local_write(register_ts).await;

        Ok(())
    }

    #[instrument(level = "debug")]
    async fn create_source_collections(
        &mut self,
        source_collections_to_create: BTreeMap<GlobalId, CollectionDescription>,
    ) -> Result<(), AdapterError> {
        let storage_metadata = self.catalog.state().storage_metadata();

        self.controller
            .storage
            .create_collections(
                storage_metadata,
                None, // Sources don't need a write timestamp
                source_collections_to_create.into_iter().collect_vec(),
            )
            .await
            .unwrap_or_terminate("cannot fail to create collections");

        Ok(())
    }

    #[instrument(level = "debug")]
    async fn initialize_storage_collections(
        &mut self,
        storage_policies_to_initialize: BTreeMap<CompactionWindow, BTreeSet<GlobalId>>,
    ) -> Result<(), AdapterError> {
        for (compaction_window, global_ids) in storage_policies_to_initialize {
            self.initialize_read_policies(
                &CollectionIdBundle {
                    storage_ids: global_ids,
                    compute_ids: BTreeMap::new(),
                },
                compaction_window,
            )
            .await;
        }

        Ok(())
    }

    #[instrument(level = "debug")]
    async fn handle_create_table(
        &self,
        ctx: &Option<&mut ExecuteContext>,
        storage_collections_to_create: &mut BTreeMap<GlobalId, CollectionDescription>,
        storage_policies_to_initialize: &mut BTreeMap<CompactionWindow, BTreeSet<GlobalId>>,
        execution_timestamps_to_set: &mut BTreeSet<StatementLoggingId>,
        table_id: CatalogItemId,
        table: Table,
    ) -> Result<(), AdapterError> {
        // The table data_source determines whether this table will be written to
        // by environmentd (e.g. with INSERT INTO statements) or by the storage layer
        // (e.g. a source-fed table).
        match &table.data_source {
            TableDataSource::TableWrites { defaults: _ } => {
                let versions: BTreeMap<_, _> = table
                    .collection_descs()
                    .map(|(gid, version, desc)| (version, (gid, desc)))
                    .collect();
                let collection_descs = versions.iter().map(|(_version, (gid, desc))| {
                    let collection_desc = CollectionDescription::for_table(desc.clone());

                    (*gid, collection_desc)
                });

                let compaction_window = table
                    .custom_logical_compaction_window
                    .unwrap_or(CompactionWindow::Default);
                let ids_to_initialize = storage_policies_to_initialize
                    .entry(compaction_window)
                    .or_default();

                for (gid, collection_desc) in collection_descs {
                    storage_collections_to_create.insert(gid, collection_desc);
                    ids_to_initialize.insert(gid);
                }

                if let Some(id) = ctx.as_ref().and_then(|ctx| ctx.extra().contents()) {
                    execution_timestamps_to_set.insert(id);
                }
            }
            TableDataSource::DataSource {
                desc: data_source_desc,
                timeline,
            } => {
                match data_source_desc {
                    DataSourceDesc::IngestionExport {
                        ingestion_id,
                        external_reference: _,
                        details,
                        data_config,
                    } => {
                        let global_ingestion_id =
                            self.catalog().get_entry(ingestion_id).latest_global_id();

                        let collection_desc = CollectionDescription {
                            desc: table.desc.latest(),
                            data_source: DataSource::IngestionExport {
                                ingestion_id: global_ingestion_id,
                                details: details.clone(),
                                data_config: data_config
                                    .clone()
                                    .into_inline_connection(self.catalog.state()),
                            },
                            since: None,
                            timeline: Some(timeline.clone()),
                            primary: None,
                        };

                        let global_id = table
                            .global_ids()
                            .expect_element(|| "subsources cannot have multiple versions");

                        storage_collections_to_create.insert(global_id, collection_desc);

                        let read_policies = self
                            .catalog()
                            .state()
                            .source_compaction_windows(vec![table_id]);
                        for (compaction_window, catalog_ids) in read_policies {
                            let compaction_ids = storage_policies_to_initialize
                                .entry(compaction_window)
                                .or_default();

                            let gids = catalog_ids
                                .into_iter()
                                .map(|item_id| self.catalog().get_entry(&item_id).global_ids())
                                .flatten();
                            compaction_ids.extend(gids);
                        }
                    }
                    DataSourceDesc::Webhook {
                        validate_using: _,
                        body_format: _,
                        headers: _,
                        cluster_id: _,
                    } => {
                        // Create the underlying collection with the latest schema from the Table.
                        assert_eq!(
                            table.desc.latest_version(),
                            RelationVersion::root(),
                            "found webhook with more than 1 relation version, {:?}",
                            table.desc
                        );
                        let desc = table.desc.latest();

                        let collection_desc = CollectionDescription {
                            desc,
                            data_source: DataSource::Webhook,
                            since: None,
                            timeline: Some(timeline.clone()),
                            primary: None,
                        };

                        let global_id = table
                            .global_ids()
                            .expect_element(|| "webhooks cannot have multiple versions");

                        storage_collections_to_create.insert(global_id, collection_desc);

                        let read_policies = self
                            .catalog()
                            .state()
                            .source_compaction_windows(vec![table_id]);

                        for (compaction_window, catalog_ids) in read_policies {
                            let compaction_ids = storage_policies_to_initialize
                                .entry(compaction_window)
                                .or_default();

                            let gids = catalog_ids
                                .into_iter()
                                .map(|item_id| self.catalog().get_entry(&item_id).global_ids())
                                .flatten();
                            compaction_ids.extend(gids);
                        }
                    }
                    _ => unreachable!("CREATE TABLE data source got {:?}", data_source_desc),
                }
            }
        }

        Ok(())
    }

    #[instrument(level = "debug")]
    async fn handle_alter_table(
        &mut self,
        catalog_id: CatalogItemId,
        prev_table: Table,
        new_table: Table,
    ) -> Result<(), AdapterError> {
        let existing_gid = prev_table.global_id_writes();
        let new_gid = new_table.global_id_writes();

        if existing_gid == new_gid {
            // It's not an ALTER TABLE ADD COLUMN, because we still have the
            // same GlobalId. It might be a compaction window change.
            if prev_table.custom_logical_compaction_window
                != new_table.custom_logical_compaction_window
            {
                let new_window = new_table
                    .custom_logical_compaction_window
                    .unwrap_or(CompactionWindow::Default);
                self.update_storage_read_policies(vec![(catalog_id, new_window.into())]);
            }
            return Ok(());
        }

        // Acquire a read hold on the original table for the duration of
        // the alter to prevent the since of the original table from
        // getting advanced, while the ALTER is running.
        let existing_table = crate::CollectionIdBundle {
            storage_ids: BTreeSet::from([existing_gid]),
            compute_ids: BTreeMap::new(),
        };
        let existing_table_read_hold = self.acquire_read_holds(&existing_table);

        let expected_version = prev_table.desc.latest_version();
        let new_version = new_table.desc.latest_version();
        let new_desc = new_table
            .desc
            .at_version(RelationVersionSelector::Specific(new_version));

        let write_ts = self.get_local_write_ts().await;
        let register_ts = write_ts.timestamp;

        // Ensure the catalog will be immediately readable at the read ts we're
        // about to bump.
        self.catalog
            .advance_upper(write_ts.advance_to)
            .await
            .unwrap_or_terminate("unable to advance catalog upper");

        // Alter the table description, creating a "new" collection.
        self.controller
            .storage
            .alter_table_desc(
                existing_gid,
                new_gid,
                new_desc,
                expected_version,
                register_ts,
            )
            .await
            .expect("failed to alter desc of table");

        // Initialize the ReadPolicy which ensures we have the correct read holds.
        let compaction_window = new_table
            .custom_logical_compaction_window
            .unwrap_or(CompactionWindow::Default);
        self.initialize_read_policies(
            &crate::CollectionIdBundle {
                storage_ids: BTreeSet::from([new_gid]),
                compute_ids: BTreeMap::new(),
            },
            compaction_window,
        )
        .await;

        self.apply_local_write(register_ts).await;

        // Alter is complete! We can drop our read hold.
        drop(existing_table_read_hold);

        Ok(())
    }

    #[instrument(level = "debug")]
    async fn handle_create_source(
        &self,
        storage_collections_to_create: &mut BTreeMap<GlobalId, CollectionDescription>,
        storage_policies_to_initialize: &mut BTreeMap<CompactionWindow, BTreeSet<GlobalId>>,
        item_id: CatalogItemId,
        source: Source,
        compaction_windows: BTreeMap<CompactionWindow, BTreeSet<CatalogItemId>>,
    ) -> Result<(), AdapterError> {
        let data_source = match source.data_source {
            DataSourceDesc::Ingestion { desc, cluster_id } => {
                let desc = desc.into_inline_connection(self.catalog().state());
                let item_global_id = self.catalog().get_entry(&item_id).latest_global_id();

                let ingestion = mz_storage_types::sources::IngestionDescription::new(
                    desc,
                    cluster_id,
                    item_global_id,
                );

                DataSource::Ingestion(ingestion)
            }
            DataSourceDesc::OldSyntaxIngestion {
                desc,
                progress_subsource,
                data_config,
                details,
                cluster_id,
            } => {
                let desc = desc.into_inline_connection(self.catalog().state());
                let data_config = data_config.into_inline_connection(self.catalog().state());

                // TODO(parkmycar): We should probably check the type here, but I'm not
                // sure if this will always be a Source or a Table.
                let progress_subsource = self
                    .catalog()
                    .get_entry(&progress_subsource)
                    .latest_global_id();

                let mut ingestion = mz_storage_types::sources::IngestionDescription::new(
                    desc,
                    cluster_id,
                    progress_subsource,
                );

                let legacy_export = SourceExport {
                    storage_metadata: (),
                    data_config,
                    details,
                };

                ingestion
                    .source_exports
                    .insert(source.global_id, legacy_export);

                DataSource::Ingestion(ingestion)
            }
            DataSourceDesc::IngestionExport {
                ingestion_id,
                external_reference: _,
                details,
                data_config,
            } => {
                // TODO(parkmycar): We should probably check the type here, but I'm not sure if
                // this will always be a Source or a Table.
                let ingestion_id = self.catalog().get_entry(&ingestion_id).latest_global_id();

                DataSource::IngestionExport {
                    ingestion_id,
                    details,
                    data_config: data_config.into_inline_connection(self.catalog().state()),
                }
            }
            DataSourceDesc::Progress => DataSource::Progress,
            DataSourceDesc::Webhook { .. } => DataSource::Webhook,
            DataSourceDesc::Introspection(_) | DataSourceDesc::Catalog => {
                unreachable!("cannot create sources with internal data sources")
            }
        };

        storage_collections_to_create.insert(
            source.global_id,
            CollectionDescription {
                desc: source.desc.clone(),
                data_source,
                timeline: Some(source.timeline),
                since: None,
                primary: None,
            },
        );

        // Initialize read policies for the source
        for (compaction_window, catalog_ids) in compaction_windows {
            let compaction_ids = storage_policies_to_initialize
                .entry(compaction_window)
                .or_default();

            let gids = catalog_ids
                .into_iter()
                .map(|item_id| self.catalog().get_entry(&item_id).global_ids())
                .flatten();
            compaction_ids.extend(gids);
        }

        Ok(())
    }

    /// Handles altering a connection by collecting all the dependent sources,
    /// sinks, and tables that need their connection updated.
    ///
    /// This mirrors the logic from `sequence_alter_connection_stage_finish` but
    /// collects the changes into batched collections for application after all
    /// implications are processed.
    #[instrument(level = "debug")]
    fn handle_alter_connection(
        &self,
        connection_id: CatalogItemId,
        connection: Connection,
        vpc_endpoints_to_create: &mut Vec<(CatalogItemId, VpcEndpointConfig)>,
        source_connections_to_alter: &mut BTreeMap<
            GlobalId,
            GenericSourceConnection<InlinedConnection>,
        >,
        sink_connections_to_alter: &mut BTreeMap<GlobalId, StorageSinkConnection>,
        source_export_data_configs_to_alter: &mut BTreeMap<GlobalId, SourceExportDataConfig>,
    ) {
        use std::collections::VecDeque;

        // Handle AWS PrivateLink connections by queueing VPC endpoint creation.
        if let ConnectionDetails::AwsPrivatelink(ref privatelink) = connection.details {
            let spec = VpcEndpointConfig {
                aws_service_name: privatelink.service_name.to_owned(),
                availability_zone_ids: privatelink.availability_zones.to_owned(),
            };
            vpc_endpoints_to_create.push((connection_id, spec));
        }

        // Walk the dependency graph to find all sources, sinks, and tables
        // that depend on this connection (directly or transitively through
        // other connections).
        let mut connections_to_process = VecDeque::new();
        connections_to_process.push_front(connection_id.clone());

        while let Some(id) = connections_to_process.pop_front() {
            for dependent_id in self.catalog().get_entry(&id).used_by() {
                let dependent_entry = self.catalog().get_entry(dependent_id);
                match dependent_entry.item() {
                    CatalogItem::Connection(_) => {
                        // Connections can depend on other connections (e.g., a
                        // Kafka connection using an SSH tunnel connection).
                        // Process these transitively.
                        connections_to_process.push_back(*dependent_id);
                    }
                    CatalogItem::Source(source) => {
                        let desc = match &dependent_entry
                            .source()
                            .expect("known to be source")
                            .data_source
                        {
                            DataSourceDesc::Ingestion { desc, .. }
                            | DataSourceDesc::OldSyntaxIngestion { desc, .. } => {
                                desc.clone().into_inline_connection(self.catalog().state())
                            }
                            DataSourceDesc::IngestionExport { .. }
                            | DataSourceDesc::Introspection(_)
                            | DataSourceDesc::Progress
                            | DataSourceDesc::Webhook { .. }
                            | DataSourceDesc::Catalog => {
                                // Only ingestions reference connections directly.
                                continue;
                            }
                        };

                        source_connections_to_alter.insert(source.global_id, desc.connection);
                    }
                    CatalogItem::Sink(sink) => {
                        let export = dependent_entry.sink().expect("known to be sink");
                        sink_connections_to_alter.insert(
                            sink.global_id,
                            export
                                .connection
                                .clone()
                                .into_inline_connection(self.catalog().state()),
                        );
                    }
                    CatalogItem::Table(table) => {
                        // This is a source-fed table that references a schema
                        // registry connection as part of its encoding/data
                        // config.
                        if let Some((_, _, _, export_data_config)) =
                            dependent_entry.source_export_details()
                        {
                            let data_config = export_data_config.clone();
                            source_export_data_configs_to_alter.insert(
                                table.global_id_writes(),
                                data_config.into_inline_connection(self.catalog().state()),
                            );
                        }
                    }
                    CatalogItem::Log(_)
                    | CatalogItem::View(_)
                    | CatalogItem::MaterializedView(_)
                    | CatalogItem::Index(_)
                    | CatalogItem::Type(_)
                    | CatalogItem::Func(_)
                    | CatalogItem::Secret(_) => {
                        // Other item types don't have connection dependencies
                        // that need updating.
                    }
                }
            }
        }
    }

    async fn handle_create_cluster_replica(
        &mut self,
        cluster_id: ClusterId,
        replica_id: ReplicaId,
        role: ClusterRole,
        cluster_name: String,
        replica_name: String,
        replica_config: ReplicaConfig,
    ) {
        let enable_worker_core_affinity =
            self.catalog().system_config().enable_worker_core_affinity();
        let enable_storage_introspection_logs = self
            .catalog()
            .system_config()
            .enable_storage_introspection_logs();

        // This replica's scoped (replica-local) overrides were pushed into the
        // controller's per-replica layer before this loop, by the
        // replica-scoped-configuration implication, so the replica's first
        // configuration replays with them. Render-frozen flags make a later push
        // too late, which is why the push precedes `create_replica`.

        self.controller
            .create_replica(
                cluster_id,
                replica_id,
                cluster_name,
                replica_name,
                role,
                replica_config,
                enable_worker_core_affinity,
                enable_storage_introspection_logs,
            )
            .expect("creating replicas must not fail");

        self.install_introspection_subscribes(cluster_id, replica_id)
            .await;
    }
}

/// A state machine for building catalog implications from catalog updates.
///
/// Once all [ParsedStateUpdate] of a timestamp are ingested this is a command
/// that has to potentially be applied to in-memory state and/or the
/// controller(s).
#[derive(Debug, Clone)]
enum CatalogImplication {
    None,
    Table(CatalogImplicationKind<Table>),
    Source(CatalogImplicationKind<(Source, Option<GenericSourceConnection>)>),
    Sink(CatalogImplicationKind<Sink>),
    Index(CatalogImplicationKind<Index>),
    MaterializedView(CatalogImplicationKind<MaterializedView>),
    View(CatalogImplicationKind<View>),
    Secret(CatalogImplicationKind<Secret>),
    Connection(CatalogImplicationKind<Connection>),
    Cluster(CatalogImplicationKind<Cluster>),
    ClusterReplica(CatalogImplicationKind<ClusterReplica>),
}

#[derive(Debug, Clone)]
enum CatalogImplicationKind<T> {
    /// No operations seen yet.
    None,
    /// Item was added.
    Added(T),
    /// Item was dropped (with its name retained for error messages).
    Dropped(T, String),
    /// Item is being altered from one state to another.
    Altered { prev: T, new: T },
}

impl<T: Clone> CatalogImplicationKind<T> {
    /// Apply a state transition based on a diff. Returns an error message if
    /// the transition is invalid.
    fn transition(&mut self, item: T, name: Option<String>, diff: StateDiff) -> Result<(), String> {
        use CatalogImplicationKind::*;
        use StateDiff::*;

        let new_state = match (&*self, diff) {
            // Initial state transitions
            (None, Addition) => Added(item),
            (None, Retraction) => Dropped(item, name.unwrap_or_else(|| "<unknown>".to_string())),

            // From Added state
            (Added(existing), Retraction) => {
                // Add -> Drop means the item is being altered
                Altered {
                    prev: item,
                    new: existing.clone(),
                }
            }
            (Added(_), Addition) => {
                return Err("Cannot add an already added object".to_string());
            }

            // From Dropped state
            (Dropped(existing, _), Addition) => {
                // Drop -> Add means the item is being altered
                Altered {
                    prev: existing.clone(),
                    new: item,
                }
            }
            (Dropped(_, _), Retraction) => {
                return Err("Cannot drop an already dropped object".to_string());
            }

            // From Altered state
            (Altered { .. }, _) => {
                return Err(format!(
                    "Cannot apply {:?} to an object in Altered state",
                    diff
                ));
            }
        };

        *self = new_state;
        Ok(())
    }
}

/// Macro to generate absorb methods for each item type.
macro_rules! impl_absorb_method {
    (
        $method_name:ident,
        $variant:ident,
        $item_type:ty
    ) => {
        fn $method_name(
            &mut self,
            item: $item_type,
            parsed_full_name: Option<String>,
            diff: StateDiff,
        ) {
            let state = match self {
                CatalogImplication::$variant(state) => state,
                CatalogImplication::None => {
                    *self = CatalogImplication::$variant(CatalogImplicationKind::None);
                    match self {
                        CatalogImplication::$variant(state) => state,
                        _ => unreachable!(),
                    }
                }
                _ => {
                    panic!(
                        "Unexpected command type for {:?}: {} {:?}",
                        self,
                        stringify!($variant),
                        diff,
                    );
                }
            };

            if let Err(e) = state.transition(item, parsed_full_name, diff) {
                panic!(
                    "Invalid state transition for {}: {}",
                    stringify!($variant),
                    e
                );
            }
        }
    };
}

impl CatalogImplication {
    /// Absorbs the given catalog update into this [CatalogImplication], causing
    /// a state transition or error.
    fn absorb(&mut self, catalog_update: ParsedStateUpdate) {
        match catalog_update.kind {
            ParsedStateUpdateKind::Item {
                durable_item: _,
                parsed_item,
                connection,
                parsed_full_name,
            } => match parsed_item {
                CatalogItem::Table(table) => {
                    self.absorb_table(table, Some(parsed_full_name), catalog_update.diff)
                }
                CatalogItem::Source(source) => {
                    self.absorb_source(
                        (source, connection),
                        Some(parsed_full_name),
                        catalog_update.diff,
                    );
                }
                CatalogItem::Sink(sink) => {
                    self.absorb_sink(sink, Some(parsed_full_name), catalog_update.diff);
                }
                CatalogItem::Index(index) => {
                    self.absorb_index(index, Some(parsed_full_name), catalog_update.diff);
                }
                CatalogItem::MaterializedView(mv) => {
                    self.absorb_materialized_view(mv, Some(parsed_full_name), catalog_update.diff);
                }
                CatalogItem::View(view) => {
                    self.absorb_view(view, Some(parsed_full_name), catalog_update.diff);
                }

                CatalogItem::Secret(secret) => {
                    self.absorb_secret(secret, None, catalog_update.diff);
                }
                CatalogItem::Connection(connection) => {
                    self.absorb_connection(connection, None, catalog_update.diff);
                }
                CatalogItem::Log(_) => {}
                CatalogItem::Type(_) => {}
                CatalogItem::Func(_) => {}
            },
            ParsedStateUpdateKind::TemporaryItem {
                durable_item: _,
                parsed_item,
                connection,
                parsed_full_name,
            } => match parsed_item {
                CatalogItem::Table(table) => {
                    self.absorb_table(table, Some(parsed_full_name), catalog_update.diff)
                }
                CatalogItem::Source(source) => {
                    self.absorb_source(
                        (source, connection),
                        Some(parsed_full_name),
                        catalog_update.diff,
                    );
                }
                CatalogItem::Sink(sink) => {
                    self.absorb_sink(sink, Some(parsed_full_name), catalog_update.diff);
                }
                CatalogItem::Index(index) => {
                    self.absorb_index(index, Some(parsed_full_name), catalog_update.diff);
                }
                CatalogItem::MaterializedView(mv) => {
                    self.absorb_materialized_view(mv, Some(parsed_full_name), catalog_update.diff);
                }
                CatalogItem::View(view) => {
                    self.absorb_view(view, Some(parsed_full_name), catalog_update.diff);
                }

                CatalogItem::Secret(secret) => {
                    self.absorb_secret(secret, None, catalog_update.diff);
                }
                CatalogItem::Connection(connection) => {
                    self.absorb_connection(connection, None, catalog_update.diff);
                }
                CatalogItem::Log(_) => {}
                CatalogItem::Type(_) => {}
                CatalogItem::Func(_) => {}
            },
            ParsedStateUpdateKind::Cluster {
                durable_cluster: _,
                parsed_cluster,
            } => {
                let name = parsed_cluster.name.clone();
                self.absorb_cluster(parsed_cluster, Some(name), catalog_update.diff);
            }
            ParsedStateUpdateKind::ClusterReplica {
                durable_cluster_replica: _,
                parsed_cluster_replica,
            } => {
                let name = parsed_cluster_replica.name.clone();
                self.absorb_cluster_replica(
                    parsed_cluster_replica,
                    Some(name),
                    catalog_update.diff,
                );
            }
            ParsedStateUpdateKind::IntrospectionSourceIndex { .. } => {
                // IntrospectionSourceIndex updates are collected
                // separately in apply_catalog_implications and not
                // routed through absorb.
                unreachable!("IntrospectionSourceIndex should not be passed to absorb");
            }
            ParsedStateUpdateKind::ReplicaSystemConfiguration { .. } => {
                // ReplicaSystemConfiguration updates are collected separately in
                // apply_catalog_implications and not routed through absorb.
                unreachable!("ReplicaSystemConfiguration should not be passed to absorb");
            }
        }
    }

    impl_absorb_method!(absorb_table, Table, Table);
    impl_absorb_method!(
        absorb_source,
        Source,
        (Source, Option<GenericSourceConnection>)
    );
    impl_absorb_method!(absorb_sink, Sink, Sink);
    impl_absorb_method!(absorb_index, Index, Index);
    impl_absorb_method!(absorb_materialized_view, MaterializedView, MaterializedView);
    impl_absorb_method!(absorb_view, View, View);

    impl_absorb_method!(absorb_secret, Secret, Secret);
    impl_absorb_method!(absorb_connection, Connection, Connection);

    impl_absorb_method!(absorb_cluster, Cluster, Cluster);
    impl_absorb_method!(absorb_cluster_replica, ClusterReplica, ClusterReplica);
}

#[cfg(test)]
mod tests {
    use super::*;
    use mz_repr::{GlobalId, RelationDesc, RelationVersion, VersionedRelationDesc};
    use mz_sql::names::ResolvedIds;
    use std::collections::BTreeMap;

    fn create_test_table(name: &str) -> Table {
        Table {
            desc: VersionedRelationDesc::new(
                RelationDesc::builder()
                    .with_column(name, mz_repr::SqlScalarType::String.nullable(false))
                    .finish(),
            ),
            create_sql: None,
            collections: BTreeMap::from([(RelationVersion::root(), GlobalId::System(1))]),
            conn_id: None,
            resolved_ids: ResolvedIds::empty(),
            custom_logical_compaction_window: None,
            is_retained_metrics_object: false,
            data_source: TableDataSource::TableWrites { defaults: vec![] },
        }
    }

    #[mz_ore::test]
    fn test_item_state_transitions() {
        // Test None -> Added
        let mut state = CatalogImplicationKind::None;
        assert!(
            state
                .transition("item1".to_string(), None, StateDiff::Addition)
                .is_ok()
        );
        assert!(matches!(state, CatalogImplicationKind::Added(_)));

        // Test Added -> Altered (via retraction)
        let mut state = CatalogImplicationKind::Added("new_item".to_string());
        assert!(
            state
                .transition("old_item".to_string(), None, StateDiff::Retraction)
                .is_ok()
        );
        match &state {
            CatalogImplicationKind::Altered { prev, new } => {
                // The retracted item is the OLD state
                assert_eq!(prev, "old_item");
                // The existing Added item is the NEW state
                assert_eq!(new, "new_item");
            }
            _ => panic!("Expected Altered state"),
        }

        // Test None -> Dropped
        let mut state = CatalogImplicationKind::None;
        assert!(
            state
                .transition(
                    "item1".to_string(),
                    Some("test_name".to_string()),
                    StateDiff::Retraction
                )
                .is_ok()
        );
        assert!(matches!(state, CatalogImplicationKind::Dropped(_, _)));

        // Test Dropped -> Altered (via addition)
        let mut state = CatalogImplicationKind::Dropped("old_item".to_string(), "name".to_string());
        assert!(
            state
                .transition("new_item".to_string(), None, StateDiff::Addition)
                .is_ok()
        );
        match &state {
            CatalogImplicationKind::Altered { prev, new } => {
                // The existing Dropped item is the OLD state
                assert_eq!(prev, "old_item");
                // The added item is the NEW state
                assert_eq!(new, "new_item");
            }
            _ => panic!("Expected Altered state"),
        }

        // Test invalid transitions
        let mut state = CatalogImplicationKind::Added("item".to_string());
        assert!(
            state
                .transition("item2".to_string(), None, StateDiff::Addition)
                .is_err()
        );

        let mut state = CatalogImplicationKind::Dropped("item".to_string(), "name".to_string());
        assert!(
            state
                .transition("item2".to_string(), None, StateDiff::Retraction)
                .is_err()
        );
    }

    #[mz_ore::test]
    fn test_table_absorb_state_machine() {
        let table1 = create_test_table("table1");
        let table2 = create_test_table("table2");

        // Test None -> AddTable
        let mut cmd = CatalogImplication::None;
        cmd.absorb_table(
            table1.clone(),
            Some("schema.table1".to_string()),
            StateDiff::Addition,
        );
        // Check that we have an Added state
        match &cmd {
            CatalogImplication::Table(state) => match state {
                CatalogImplicationKind::Added(t) => {
                    assert_eq!(t.desc.latest().arity(), table1.desc.latest().arity())
                }
                _ => panic!("Expected Added state"),
            },
            _ => panic!("Expected Table command"),
        }

        // Test AddTable -> AlterTable (via retraction)
        // This tests the bug fix: when we have AddTable(table1) and receive Retraction(table2),
        // table2 is the old state being removed, table1 is the new state
        cmd.absorb_table(
            table2.clone(),
            Some("schema.table2".to_string()),
            StateDiff::Retraction,
        );
        match &cmd {
            CatalogImplication::Table(state) => match state {
                CatalogImplicationKind::Altered { prev, new } => {
                    // Verify the fix: prev should be the retracted table, new should be the added table
                    assert_eq!(prev.desc.latest().arity(), table2.desc.latest().arity());
                    assert_eq!(new.desc.latest().arity(), table1.desc.latest().arity());
                }
                _ => panic!("Expected Altered state"),
            },
            _ => panic!("Expected Table command"),
        }

        // Test None -> DropTable
        let mut cmd = CatalogImplication::None;
        cmd.absorb_table(
            table1.clone(),
            Some("schema.table1".to_string()),
            StateDiff::Retraction,
        );
        match &cmd {
            CatalogImplication::Table(state) => match state {
                CatalogImplicationKind::Dropped(t, name) => {
                    assert_eq!(t.desc.latest().arity(), table1.desc.latest().arity());
                    assert_eq!(name, "schema.table1");
                }
                _ => panic!("Expected Dropped state"),
            },
            _ => panic!("Expected Table command"),
        }

        // Test DropTable -> AlterTable (via addition)
        cmd.absorb_table(
            table2.clone(),
            Some("schema.table2".to_string()),
            StateDiff::Addition,
        );
        match &cmd {
            CatalogImplication::Table(state) => match state {
                CatalogImplicationKind::Altered { prev, new } => {
                    // prev should be the dropped table, new should be the added table
                    assert_eq!(prev.desc.latest().arity(), table1.desc.latest().arity());
                    assert_eq!(new.desc.latest().arity(), table2.desc.latest().arity());
                }
                _ => panic!("Expected Altered state"),
            },
            _ => panic!("Expected Table command"),
        }
    }

    #[mz_ore::test]
    #[should_panic(expected = "Cannot add an already added object")]
    fn test_invalid_double_add() {
        let table = create_test_table("table");
        let mut cmd = CatalogImplication::None;

        // First addition
        cmd.absorb_table(
            table.clone(),
            Some("schema.table".to_string()),
            StateDiff::Addition,
        );

        // Second addition should panic
        cmd.absorb_table(
            table.clone(),
            Some("schema.table".to_string()),
            StateDiff::Addition,
        );
    }

    #[mz_ore::test]
    #[should_panic(expected = "Cannot drop an already dropped object")]
    fn test_invalid_double_drop() {
        let table = create_test_table("table");
        let mut cmd = CatalogImplication::None;

        // First drop
        cmd.absorb_table(
            table.clone(),
            Some("schema.table".to_string()),
            StateDiff::Retraction,
        );

        // Second drop should panic
        cmd.absorb_table(
            table.clone(),
            Some("schema.table".to_string()),
            StateDiff::Retraction,
        );
    }
}