mz_adapter/catalog/

open.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 opening a [`Catalog`].

mod builtin_schema_migration;

use std::collections::{BTreeMap, BTreeSet};
use std::num::NonZeroU32;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::time::{Duration, Instant};

use futures::future::{BoxFuture, FutureExt};
use itertools::{Either, Itertools};
use mz_adapter_types::bootstrap_builtin_cluster_config::BootstrapBuiltinClusterConfig;
use mz_adapter_types::dyncfgs::ENABLE_EXPRESSION_CACHE;
use mz_audit_log::{
    CreateOrDropClusterReplicaReasonV1, EventDetails, EventType, ObjectType, VersionedEvent,
};
use mz_auth::hash::scram256_hash;
use mz_catalog::builtin::{
    BUILTIN_CLUSTERS, BUILTIN_PREFIXES, BUILTIN_ROLES, BUILTINS, Builtin, Fingerprint,
    MZ_CATALOG_RAW, RUNTIME_ALTERABLE_FINGERPRINT_SENTINEL,
};
use mz_catalog::config::StateConfig;
use mz_catalog::durable::objects::{
    SystemObjectDescription, SystemObjectMapping, SystemObjectUniqueIdentifier,
};
use mz_catalog::durable::{
    ClusterReplica, ClusterVariant, ClusterVariantManaged, ReplicaConfig, ReplicaLocation,
    Transaction, managed_cluster_replica_name,
};
use mz_catalog::expr_cache::{
    ExpressionCacheConfig, ExpressionCacheHandle, GlobalExpressions, LocalExpressions,
    latest_item_version,
};
use mz_catalog::memory::error::{Error, ErrorKind};
use mz_catalog::memory::objects::{
    CommentsMap, DefaultPrivileges, RoleAuth, StateUpdate, StateUpdateKind,
};
use mz_controller::clusters::ReplicaLogging;
use mz_controller_types::ClusterId;
use mz_ore::cast::usize_to_u64;
use mz_ore::collections::HashSet;
use mz_ore::now::{SYSTEM_TIME, to_datetime};
use mz_ore::{instrument, soft_assert_no_log};
use mz_repr::adt::mz_acl_item::PrivilegeMap;
use mz_repr::namespaces::is_unstable_schema;
use mz_repr::{CatalogItemId, Diff, GlobalId, RelationVersion, Timestamp};
use mz_sql::catalog::{CatalogError as SqlCatalogError, CatalogItemType, RoleMembership, RoleVars};
use mz_sql::func::OP_IMPLS;
use mz_sql::names::CommentObjectId;
use mz_sql::rbac;
use mz_sql::session::user::{MZ_SYSTEM_ROLE_ID, SYSTEM_USER};
use mz_sql::session::vars::{SessionVars, SystemVars, VarError, VarInput};
use mz_storage_client::controller::{StorageMetadata, StorageTxn};
use mz_storage_client::storage_collections::StorageCollections;
use semver::Version;
use tracing::{Instrument, info, warn};
use uuid::Uuid;

// DO NOT add any more imports from `crate` outside of `crate::catalog`.
use crate::AdapterError;
use crate::catalog::migrate::{self, get_migration_version, set_migration_version};
use crate::catalog::state::LocalExpressionCache;
use crate::catalog::{BuiltinTableUpdate, Catalog, CatalogState, Config, is_reserved_name};

pub struct InitializeStateResult {
    /// An initialized [`CatalogState`].
    pub state: CatalogState,
    /// Items whose builtin schema migration allocated a fresh, self-owned persist shard. Only used
    /// by 0dt migration.
    ///
    /// `Replacement`-migrated items only: derived by filtering to items with a global id in
    /// `MigrationRunResult::new_shards`, which only `migrate_replace` populates. `Evolution`
    /// migrates in place and reuses the leader's shard, so it never lands here. Read-only write
    /// paths depend on that: force-writing a shard while read-only is safe only because we
    /// exclusively own it (see `ComputeController::allow_writes_in_read_only`).
    pub migrated_storage_collections_0dt: BTreeSet<CatalogItemId>,
    /// A set of new builtin items.
    pub new_builtin_collections: BTreeSet<GlobalId>,
    /// A list of builtin table updates corresponding to the initialized state.
    pub builtin_table_updates: Vec<BuiltinTableUpdate>,
    /// The version of the binary that last committed catalog migrations, or `None` for a newly
    /// initialized catalog.
    ///
    /// While this environment is read-only during a 0dt deployment, this is the version of the
    /// leader environment: the read-only catalog transaction is a savepoint, so our own bump of
    /// the setting never lands.
    pub last_seen_version: Option<Version>,
    /// A handle to the expression cache if it's enabled.
    pub expr_cache_handle: Option<ExpressionCacheHandle>,
    /// The global expressions that were cached in `expr_cache_handle`.
    pub cached_global_exprs: BTreeMap<GlobalId, GlobalExpressions>,
    /// The local expressions that were NOT cached in `expr_cache_handle`.
    pub uncached_local_exprs: BTreeMap<GlobalId, LocalExpressions>,
}

pub struct OpenCatalogResult {
    /// An opened [`Catalog`].
    pub catalog: Catalog,
    /// See [`InitializeStateResult::last_seen_version`].
    pub last_seen_version: Option<Version>,
    /// See [`InitializeStateResult::migrated_storage_collections_0dt`]; `Replacement`-migrated
    /// items only.
    pub migrated_storage_collections_0dt: BTreeSet<CatalogItemId>,
    /// A set of new builtin items.
    pub new_builtin_collections: BTreeSet<GlobalId>,
    /// A list of builtin table updates corresponding to the initialized state.
    pub builtin_table_updates: Vec<BuiltinTableUpdate>,
    /// The global expressions that were cached in the expression cache.
    pub cached_global_exprs: BTreeMap<GlobalId, GlobalExpressions>,
    /// The local expressions that were NOT cached in the expression cache.
    pub uncached_local_exprs: BTreeMap<GlobalId, LocalExpressions>,
}

impl Catalog {
    /// Initializes a CatalogState. Separate from [`Catalog::open`] to avoid depending on state
    /// external to a [mz_catalog::durable::DurableCatalogState]
    /// (for example: no [mz_secrets::SecretsReader]).
    pub async fn initialize_state<'a>(
        config: StateConfig,
        storage: &'a mut Box<dyn mz_catalog::durable::DurableCatalogState>,
    ) -> Result<InitializeStateResult, AdapterError> {
        for builtin_role in BUILTIN_ROLES {
            assert!(
                is_reserved_name(builtin_role.name),
                "builtin role {builtin_role:?} must start with one of the following prefixes {}",
                BUILTIN_PREFIXES.join(", ")
            );
        }
        for builtin_cluster in BUILTIN_CLUSTERS {
            assert!(
                is_reserved_name(builtin_cluster.name),
                "builtin cluster {builtin_cluster:?} must start with one of the following prefixes {}",
                BUILTIN_PREFIXES.join(", ")
            );
        }

        let mut system_configuration = SystemVars::new().set_unsafe(config.unsafe_mode);
        if config.all_features {
            system_configuration.enable_all_feature_flags_by_default();
        }

        let mut state = CatalogState {
            database_by_name: imbl::OrdMap::new(),
            database_by_id: imbl::OrdMap::new(),
            entry_by_id: imbl::OrdMap::new(),
            entry_by_global_id: imbl::OrdMap::new(),
            notices_by_dep_id: imbl::OrdMap::new(),
            ambient_schemas_by_name: imbl::OrdMap::new(),
            ambient_schemas_by_id: imbl::OrdMap::new(),
            clusters_by_name: imbl::OrdMap::new(),
            clusters_by_id: imbl::OrdMap::new(),
            roles_by_name: imbl::OrdMap::new(),
            roles_by_id: imbl::OrdMap::new(),
            network_policies_by_id: imbl::OrdMap::new(),
            role_auth_by_id: imbl::OrdMap::new(),
            network_policies_by_name: imbl::OrdMap::new(),
            system_configuration: Arc::new(system_configuration),
            scoped_system_parameters: Default::default(),
            default_privileges: Arc::new(DefaultPrivileges::default()),
            system_privileges: Arc::new(PrivilegeMap::default()),
            comments: Arc::new(CommentsMap::default()),
            source_references: imbl::OrdMap::new(),
            storage_metadata: Arc::new(StorageMetadata::default()),
            temporary_namespaces: Default::default(),
            mock_authentication_nonce: Default::default(),
            config: mz_sql::catalog::CatalogConfig {
                start_time: to_datetime((config.now)()),
                start_instant: Instant::now(),
                nonce: rand::random(),
                environment_id: config.environment_id,
                session_id: Uuid::new_v4(),
                build_info: config.build_info,
                now: config.now.clone(),
                connection_context: config.connection_context,
                aws_account_id: config
                    .aws_principal_context
                    .as_ref()
                    .map(|c| c.aws_account_id.clone()),
                helm_chart_version: config.helm_chart_version,
            },
            cluster_replica_sizes: config.cluster_replica_sizes,
            availability_zones: config.availability_zones,
            egress_addresses: config.egress_addresses,
            aws_principal_context: config.aws_principal_context,
            aws_privatelink_availability_zones: config.aws_privatelink_availability_zones,
            http_host_name: config.http_host_name,
            license_key: config.license_key,
        };

        let deploy_generation = storage.get_deployment_generation().await?;

        let mut updates: Vec<_> = storage.sync_to_current_updates().await?;
        assert!(!updates.is_empty(), "initial catalog snapshot is missing");
        let mut txn = storage.transaction().await?;

        // Migrate/update durable data before we start loading the in-memory catalog.
        let new_builtin_collections = {
            migrate::durable_migrate(
                &mut txn,
                state.config.environment_id.organization_id(),
                config.boot_ts,
            )?;
            // Overwrite and persist selected parameter values in `remote_system_parameters` that
            // was pulled from a remote frontend (e.g. LaunchDarkly) if present.
            if let Some(remote_system_parameters) = config.remote_system_parameters {
                for (name, value) in remote_system_parameters {
                    txn.upsert_system_config(&name, value)?;
                }
                txn.set_system_config_synced_once()?;
            }
            // Add any new builtin objects and remove old ones.
            let new_builtin_collections = add_new_remove_old_builtin_items_migration(&mut txn)?;
            let builtin_bootstrap_cluster_config_map = BuiltinBootstrapClusterConfigMap {
                system_cluster: config.builtin_system_cluster_config,
                catalog_server_cluster: config.builtin_catalog_server_cluster_config,
                probe_cluster: config.builtin_probe_cluster_config,
                support_cluster: config.builtin_support_cluster_config,
                analytics_cluster: config.builtin_analytics_cluster_config,
            };
            add_new_remove_old_builtin_clusters_migration(
                &mut txn,
                &builtin_bootstrap_cluster_config_map,
                config.boot_ts,
            )?;
            add_new_remove_old_builtin_introspection_source_migration(&mut txn)?;
            reconcile_builtin_cluster_replicas(
                &mut txn,
                &builtin_bootstrap_cluster_config_map,
                config.boot_ts,
            )?;
            add_new_remove_old_builtin_roles_migration(&mut txn)?;
            remove_invalid_config_param_role_defaults_migration(&mut txn)?;
            remove_pending_cluster_replicas_migration(&mut txn, config.boot_ts)?;

            new_builtin_collections
        };

        let op_updates = txn.get_and_commit_op_updates();
        updates.extend(op_updates);

        let mut builtin_table_updates = Vec::new();

        // Seed the in-memory catalog with values that don't come from the durable catalog.
        {
            // Set defaults from configuration passed in the provided `system_parameter_defaults`
            // map.
            for (name, value) in config.system_parameter_defaults {
                match state.set_system_configuration_default(&name, VarInput::Flat(&value)) {
                    Ok(_) => (),
                    Err(Error {
                        kind: ErrorKind::VarError(VarError::UnknownParameter(name)),
                    }) => {
                        warn!(%name, "cannot load unknown system parameter from catalog storage to set default parameter");
                    }
                    Err(e) => return Err(e.into()),
                };
            }
        }

        // Make life easier by consolidating all updates, so that we end up with only positive
        // diffs.
        let mut updates = into_consolidatable_updates_startup(updates, config.boot_ts);
        differential_dataflow::consolidation::consolidate_updates(&mut updates);
        soft_assert_no_log!(
            updates.iter().all(|(_, _, diff)| *diff == Diff::ONE),
            "consolidated updates should be positive during startup: {updates:?}"
        );

        let mut pre_item_updates = Vec::new();
        let mut system_item_updates = Vec::new();
        let mut item_updates = Vec::new();
        let mut post_item_updates = Vec::new();
        let mut audit_log_updates = Vec::new();
        for (kind, ts, diff) in updates {
            match kind {
                StateUpdateKind::Role(_)
                | StateUpdateKind::RoleAuth(_)
                | StateUpdateKind::Database(_)
                | StateUpdateKind::Schema(_)
                | StateUpdateKind::DefaultPrivilege(_)
                | StateUpdateKind::SystemPrivilege(_)
                | StateUpdateKind::SystemConfiguration(_)
                | StateUpdateKind::ClusterSystemConfiguration(_)
                | StateUpdateKind::ReplicaSystemConfiguration(_)
                | StateUpdateKind::Cluster(_)
                | StateUpdateKind::NetworkPolicy(_)
                | StateUpdateKind::ClusterReplica(_) => pre_item_updates.push(StateUpdate {
                    kind,
                    ts,
                    diff: diff.try_into().expect("valid diff"),
                }),
                StateUpdateKind::IntrospectionSourceIndex(_)
                | StateUpdateKind::SystemObjectMapping(_) => {
                    system_item_updates.push(StateUpdate {
                        kind,
                        ts,
                        diff: diff.try_into().expect("valid diff"),
                    })
                }
                StateUpdateKind::Item(_) => item_updates.push(StateUpdate {
                    kind,
                    ts,
                    diff: diff.try_into().expect("valid diff"),
                }),
                StateUpdateKind::Comment(_)
                | StateUpdateKind::StorageCollectionMetadata(_)
                | StateUpdateKind::SourceReferences(_)
                | StateUpdateKind::UnfinalizedShard(_) => {
                    post_item_updates.push((kind, ts, diff));
                }
                StateUpdateKind::AuditLog(_) => {
                    audit_log_updates.push(StateUpdate {
                        kind,
                        ts,
                        diff: diff.try_into().expect("valid diff"),
                    });
                }
            }
        }

        let (builtin_table_update, _catalog_updates) = state
            .apply_updates(pre_item_updates, &mut LocalExpressionCache::Closed)
            .await;
        builtin_table_updates.extend(builtin_table_update);

        // Both the `system_parameter_defaults` set above and the durable
        // `SystemConfiguration` values applied via `pre_item_updates` live in
        // the `SystemVars` value map by now. Mirror their effective values into
        // the dyncfg `ConfigSet`, so that startup-only reads observe configured
        // values rather than compile-time defaults, as with
        // `ENABLE_EXPRESSION_CACHE` below. `apply_updates` only
        // syncs the `ConfigSet` when a `SystemConfiguration` update is present,
        // and a deployment configured purely via `system_parameter_default` has
        // none, so sync explicitly here.
        state.system_config().sync_dyncfgs();

        // Ensure mz_system has a password if configured to have one.
        // It's important we do this after the `pre_item_updates` so that
        // the mz_system role exists in the catalog.
        {
            if let Some(password) = config.external_login_password_mz_system {
                let role_auth = RoleAuth {
                    role_id: MZ_SYSTEM_ROLE_ID,
                    // builtin roles should always use a secure scram iteration
                    // <https://cheatsheetseries.owasp.org/cheatsheets/Password_Storage_Cheat_Sheet.html>
                    password_hash: Some(
                        scram256_hash(&password, &NonZeroU32::new(600_000).expect("known valid"))
                            .map_err(|_| {
                            AdapterError::Internal("Failed to hash mz_system password.".to_owned())
                        })?,
                    ),
                    updated_at: SYSTEM_TIME(),
                };
                state
                    .role_auth_by_id
                    .insert(MZ_SYSTEM_ROLE_ID, role_auth.clone());
                let builtin_table_update = state.generate_builtin_table_update(
                    mz_catalog::memory::objects::StateUpdateKind::RoleAuth(role_auth.into()),
                    mz_catalog::memory::objects::StateDiff::Addition,
                );
                builtin_table_updates.extend(builtin_table_update);
            }
        }

        let expr_cache_start = Instant::now();
        info!("startup: coordinator init: catalog open: expr cache open beginning");
        // We wait until after the `pre_item_updates` to open the cache so we can get accurate
        // dyncfgs because the `pre_item_updates` contains `SystemConfiguration` updates.
        let enable_expr_cache_dyncfg = ENABLE_EXPRESSION_CACHE.get(state.system_config().dyncfgs());
        let expr_cache_enabled = config
            .enable_expression_cache_override
            .unwrap_or(enable_expr_cache_dyncfg);
        let (expr_cache_handle, cached_local_exprs, cached_global_exprs) = if expr_cache_enabled {
            info!(
                ?config.enable_expression_cache_override,
                ?enable_expr_cache_dyncfg,
                "using expression cache for startup"
            );
            let current_items = txn
                .get_items()
                .flat_map(|item| {
                    let item_version = latest_item_version(&item.extra_versions);
                    std::iter::once(item.global_id)
                        .chain(item.extra_versions.into_values())
                        .map(move |gid| (gid, item_version))
                })
                .chain(
                    txn.get_system_object_mappings()
                        .map(|som| (som.unique_identifier.global_id, RelationVersion::root())),
                )
                .collect();
            let dyncfgs = config.persist_client.dyncfgs().clone();
            let build_version = if config.build_info.is_dev() {
                // A single dev version can be used for many different builds, so we need to use
                // the build version that is also enriched with build metadata.
                config
                    .build_info
                    .semver_version_build()
                    .expect("build ID is not available on your platform!")
            } else {
                config.build_info.semver_version()
            };
            let expr_cache_config = ExpressionCacheConfig {
                build_version,
                shard_id: txn
                    .get_expression_cache_shard()
                    .expect("expression cache shard should exist for opened catalogs"),
                persist: config.persist_client,
                current_items,
                remove_prior_versions: !config.read_only,
                compact_shard: config.read_only,
                dyncfgs,
            };
            let (expr_cache_handle, cached_local_exprs, cached_global_exprs) =
                ExpressionCacheHandle::spawn_expression_cache(expr_cache_config).await;
            (
                Some(expr_cache_handle),
                cached_local_exprs,
                cached_global_exprs,
            )
        } else {
            (None, BTreeMap::new(), BTreeMap::new())
        };
        let mut local_expr_cache = LocalExpressionCache::new(cached_local_exprs);
        info!(
            "startup: coordinator init: catalog open: expr cache open complete in {:?}",
            expr_cache_start.elapsed()
        );

        // When initializing/bootstrapping, we don't use the catalog updates but
        // instead load the catalog fully and then go ahead and apply commands
        // to the controller(s). Maybe we _should_ instead use the same logic
        // and return and use the updates from here. But that's at the very
        // least future work.
        let (builtin_table_update, _catalog_updates) = state
            .apply_updates(system_item_updates, &mut local_expr_cache)
            .await;
        builtin_table_updates.extend(builtin_table_update);

        let last_seen_version = get_migration_version(&txn);

        let mz_authentication_mock_nonce =
            txn.get_authentication_mock_nonce().ok_or_else(|| {
                Error::new(ErrorKind::SettingError("authentication nonce".to_string()))
            })?;

        state.mock_authentication_nonce = Some(mz_authentication_mock_nonce);

        // Migrate item ASTs.
        let (builtin_table_update, _catalog_updates) = if !config.skip_migrations {
            let migrate_result = migrate::migrate(
                &mut state,
                &mut txn,
                &mut local_expr_cache,
                item_updates,
                config.now,
                config.boot_ts,
            )
            .await
            .map_err(|e| {
                Error::new(ErrorKind::FailedCatalogMigration {
                    last_seen_version: last_seen_version
                        .as_ref()
                        .map_or_else(|| "new".to_string(), |v| v.to_string()),
                    this_version: config.build_info.version,
                    cause: e.to_string(),
                })
            })?;
            if !migrate_result.post_item_updates.is_empty() {
                // Include any post-item-updates generated by migrations, and then consolidate
                // them to ensure diffs are all positive.
                post_item_updates.extend(migrate_result.post_item_updates);
                // Push everything to the same timestamp so it consolidates cleanly.
                if let Some(max_ts) = post_item_updates.iter().map(|(_, ts, _)| ts).max().cloned() {
                    for (_, ts, _) in &mut post_item_updates {
                        *ts = max_ts;
                    }
                }
                differential_dataflow::consolidation::consolidate_updates(&mut post_item_updates);
            }

            (
                migrate_result.builtin_table_updates,
                migrate_result.catalog_updates,
            )
        } else {
            state
                .apply_updates(item_updates, &mut local_expr_cache)
                .await
        };
        builtin_table_updates.extend(builtin_table_update);

        let post_item_updates = post_item_updates
            .into_iter()
            .map(|(kind, ts, diff)| StateUpdate {
                kind,
                ts,
                diff: diff.try_into().expect("valid diff"),
            })
            .collect();
        let (builtin_table_update, _catalog_updates) = state
            .apply_updates(post_item_updates, &mut local_expr_cache)
            .await;
        builtin_table_updates.extend(builtin_table_update);

        // We don't need to apply the audit logs in memory, yet apply can be expensive when the
        // audit log grows large. Therefore, we skip the apply step and just generate the builtin
        // updates.
        for audit_log_update in audit_log_updates {
            builtin_table_updates.extend(
                state.generate_builtin_table_update(audit_log_update.kind, audit_log_update.diff),
            );
        }

        // Migrate builtin items.
        let schema_migration_result = builtin_schema_migration::run(
            config.build_info,
            deploy_generation,
            &mut txn,
            config.builtin_item_migration_config,
        )
        .await?;

        let state_updates = txn.get_and_commit_op_updates();

        // When initializing/bootstrapping, we don't use the catalog updates but
        // instead load the catalog fully and then go ahead and apply commands
        // to the controller(s). Maybe we _should_ instead use the same logic
        // and return and use the updates from here. But that's at the very
        // least future work.
        let (table_updates, _catalog_updates) = state
            .apply_updates(state_updates, &mut local_expr_cache)
            .await;
        builtin_table_updates.extend(table_updates);
        let builtin_table_updates = state.resolve_builtin_table_updates(builtin_table_updates);

        // Bump the migration version immediately before committing.
        set_migration_version(&mut txn, config.build_info.semver_version())?;

        txn.commit(config.boot_ts).await?;

        // Now that the migration is durable, run any requested deferred cleanup.
        schema_migration_result.cleanup_action.await;

        Ok(InitializeStateResult {
            state,
            migrated_storage_collections_0dt: schema_migration_result.replaced_items,
            new_builtin_collections: new_builtin_collections.into_iter().collect(),
            builtin_table_updates,
            last_seen_version,
            expr_cache_handle,
            cached_global_exprs,
            uncached_local_exprs: local_expr_cache.into_uncached_exprs(),
        })
    }

    /// Opens or creates a catalog that stores data at `path`.
    ///
    /// Returns the catalog, metadata about builtin objects that have changed
    /// schemas since last restart, a list of updates to builtin tables that
    /// describe the initial state of the catalog, and the version of the
    /// catalog before any migrations were performed.
    ///
    /// BOXED FUTURE: As of Nov 2023 the returned Future from this function was 17KB. This would
    /// get stored on the stack which is bad for runtime performance, and blow up our stack usage.
    /// Because of that we purposefully move this Future onto the heap (i.e. Box it).
    #[instrument(name = "catalog::open")]
    pub fn open(config: Config<'_>) -> BoxFuture<'static, Result<OpenCatalogResult, AdapterError>> {
        async move {
            let mut storage = config.storage;

            let InitializeStateResult {
                state,
                migrated_storage_collections_0dt,
                new_builtin_collections,
                mut builtin_table_updates,
                last_seen_version,
                expr_cache_handle,
                cached_global_exprs,
                uncached_local_exprs,
            } =
                // BOXED FUTURE: As of Nov 2023 the returned Future from this function was 7.5KB. This would
                // get stored on the stack which is bad for runtime performance, and blow up our stack usage.
                // Because of that we purposefully move this Future onto the heap (i.e. Box it).
                Self::initialize_state(config.state, &mut storage)
                    .instrument(tracing::info_span!("catalog::initialize_state"))
                    .boxed()
                    .await?;

            let catalog = Catalog {
                state,
                expr_cache_handle,
                transient_revision: 1,
                shared_transient_revision: Arc::new(AtomicU64::new(1)),
                storage: Arc::new(tokio::sync::Mutex::new(storage)),
            };

            // Operators aren't stored in the catalog, but we would like them in
            // introspection views.
            for (op, func) in OP_IMPLS.iter() {
                match func {
                    mz_sql::func::Func::Scalar(impls) => {
                        for imp in impls {
                            builtin_table_updates.push(catalog.state.resolve_builtin_table_update(
                                catalog.state.pack_op_update(op, imp.details(), Diff::ONE),
                            ));
                        }
                    }
                    _ => unreachable!("all operators must be scalar functions"),
                }
            }

            for ip in &catalog.state.egress_addresses {
                builtin_table_updates.push(
                    catalog
                        .state
                        .resolve_builtin_table_update(catalog.state.pack_egress_ip_update(ip)?),
                );
            }

            if !catalog.state.license_key.id.is_empty() {
                builtin_table_updates.push(
                    catalog.state.resolve_builtin_table_update(
                        catalog
                            .state
                            .pack_license_key_update(&catalog.state.license_key)?,
                    ),
                );
            }

            catalog.storage().await.mark_bootstrap_complete().await;

            Ok(OpenCatalogResult {
                catalog,
                last_seen_version,
                migrated_storage_collections_0dt,
                new_builtin_collections,
                builtin_table_updates,
                cached_global_exprs,
                uncached_local_exprs,
            })
        }
        .instrument(tracing::info_span!("catalog::open"))
        .boxed()
    }

    /// Initializes STORAGE to understand all shards that `self` expects to
    /// exist.
    ///
    /// Note that this must be done before creating/rendering collections
    /// because the storage controller might not be aware of new system
    /// collections created between versions.
    async fn initialize_storage_state(
        &mut self,
        storage_collections: &Arc<dyn StorageCollections + Send + Sync>,
    ) -> Result<(), mz_catalog::durable::CatalogError> {
        let collections = self
            .entries()
            .filter(|entry| entry.item().is_storage_collection())
            .flat_map(|entry| entry.global_ids())
            .collect();

        // Clone the state so that any errors that occur do not leak any
        // transformations on error.
        let mut state = self.state.clone();

        let mut storage = self.storage().await;
        let shard_id = storage.shard_id();
        let mut txn = storage.transaction().await?;

        // Ensure the storage controller knows about the catalog shard and associates it with the
        // `MZ_CATALOG_RAW` builtin source.
        let item_id = self.resolve_builtin_storage_collection(&MZ_CATALOG_RAW);
        let global_id = self.get_entry(&item_id).latest_global_id();
        match txn.get_collection_metadata().get(&global_id) {
            None => {
                txn.insert_collection_metadata([(global_id, shard_id)].into())
                    .map_err(mz_catalog::durable::DurableCatalogError::from)?;
            }
            Some(id) => assert_eq!(*id, shard_id),
        }

        storage_collections
            .initialize_state(&mut txn, collections)
            .await
            .map_err(mz_catalog::durable::DurableCatalogError::from)?;

        let updates = txn.get_and_commit_op_updates();
        let (builtin_updates, catalog_updates) = state
            .apply_updates(updates, &mut LocalExpressionCache::Closed)
            .await;
        assert!(
            builtin_updates.is_empty(),
            "storage is not allowed to generate catalog changes that would cause changes to builtin tables"
        );
        assert!(
            catalog_updates.is_empty(),
            "storage is not allowed to generate catalog changes that would change the catalog or controller state"
        );
        let commit_ts = txn.upper();
        txn.commit(commit_ts).await?;
        drop(storage);

        // Save updated state.
        self.state = state;
        Ok(())
    }

    /// [`mz_controller::Controller`] depends on durable catalog state to boot,
    /// so make it available and initialize the controller.
    pub async fn initialize_controller(
        &mut self,
        config: mz_controller::ControllerConfig,
        envd_epoch: core::num::NonZeroI64,
        read_only: bool,
    ) -> Result<mz_controller::Controller, mz_catalog::durable::CatalogError> {
        let controller_start = Instant::now();
        info!("startup: controller init: beginning");

        let controller = {
            let mut storage = self.storage().await;
            let mut tx = storage.transaction().await?;
            mz_controller::prepare_initialization(&mut tx)
                .map_err(mz_catalog::durable::DurableCatalogError::from)?;
            let updates = tx.get_and_commit_op_updates();
            assert!(
                updates.is_empty(),
                "initializing controller should not produce updates: {updates:?}"
            );
            let commit_ts = tx.upper();
            tx.commit(commit_ts).await?;

            let read_only_tx = storage.transaction().await?;

            mz_controller::Controller::new(config, envd_epoch, read_only, &read_only_tx).await
        };

        self.initialize_storage_state(&controller.storage_collections)
            .await?;

        info!(
            "startup: controller init: complete in {:?}",
            controller_start.elapsed()
        );

        Ok(controller)
    }

    /// Politely releases all external resources that can only be released in an async context.
    pub async fn expire(self) {
        // If no one else holds a reference to storage, then clean up the storage resources.
        // Otherwise, hopefully the other reference cleans up the resources when it's dropped.
        if let Some(storage) = Arc::into_inner(self.storage) {
            let storage = storage.into_inner();
            storage.expire().await;
        }
    }
}

impl CatalogState {
    /// Set the default value for `name`, which is the value it will be reset to.
    fn set_system_configuration_default(
        &mut self,
        name: &str,
        value: VarInput,
    ) -> Result<(), Error> {
        Ok(Arc::make_mut(&mut self.system_configuration).set_default(name, value)?)
    }
}

/// Updates the catalog with new and removed builtin items.
///
/// Returns the list of new builtin [`GlobalId`]s.
fn add_new_remove_old_builtin_items_migration(
    txn: &mut mz_catalog::durable::Transaction<'_>,
) -> Result<Vec<GlobalId>, mz_catalog::durable::CatalogError> {
    let mut new_builtin_mappings = Vec::new();
    // Used to validate unique descriptions.
    let mut builtin_descs = HashSet::new();

    // We compare the builtin items that are compiled into the binary with the builtin items that
    // are persisted in the catalog to discover new and deleted builtin items.
    let mut builtins = Vec::new();
    for builtin in BUILTINS::iter() {
        let desc = SystemObjectDescription {
            schema_name: builtin.schema().to_string(),
            object_type: builtin.catalog_item_type(),
            object_name: builtin.name().to_string(),
        };
        // Validate that the description is unique.
        if !builtin_descs.insert(desc.clone()) {
            panic!(
                "duplicate builtin description: {:?}, {:?}",
                SystemObjectDescription {
                    schema_name: builtin.schema().to_string(),
                    object_type: builtin.catalog_item_type(),
                    object_name: builtin.name().to_string(),
                },
                builtin
            );
        }
        builtins.push((desc, builtin));
    }

    let mut system_object_mappings: BTreeMap<_, _> = txn
        .get_system_object_mappings()
        .map(|system_object_mapping| {
            (
                system_object_mapping.description.clone(),
                system_object_mapping,
            )
        })
        .collect();

    let (existing_builtins, new_builtins): (Vec<_>, Vec<_>) =
        builtins.into_iter().partition_map(|(desc, builtin)| {
            let fingerprint = match builtin.runtime_alterable() {
                false => builtin.fingerprint(),
                true => RUNTIME_ALTERABLE_FINGERPRINT_SENTINEL.into(),
            };
            match system_object_mappings.remove(&desc) {
                Some(system_object_mapping) => {
                    Either::Left((builtin, system_object_mapping, fingerprint))
                }
                None => Either::Right((builtin, fingerprint)),
            }
        });
    let new_builtin_ids = txn.allocate_system_item_ids(usize_to_u64(new_builtins.len()))?;
    let new_builtins: Vec<_> = new_builtins
        .into_iter()
        .zip_eq(new_builtin_ids.clone())
        .collect();

    // Add new builtin items to catalog.
    for ((builtin, fingerprint), (catalog_id, global_id)) in new_builtins.iter().cloned() {
        new_builtin_mappings.push(SystemObjectMapping {
            description: SystemObjectDescription {
                schema_name: builtin.schema().to_string(),
                object_type: builtin.catalog_item_type(),
                object_name: builtin.name().to_string(),
            },
            unique_identifier: SystemObjectUniqueIdentifier {
                catalog_id,
                global_id,
                fingerprint,
            },
        });

        // Runtime-alterable system objects are durably recorded to the
        // usual items collection, so that they can be later altered at
        // runtime by their owner (i.e., outside of the usual builtin
        // migration framework that requires changes to the binary
        // itself).
        let handled_runtime_alterable = match builtin {
            Builtin::Connection(c) if c.runtime_alterable => {
                let mut acl_items = vec![rbac::owner_privilege(
                    mz_sql::catalog::ObjectType::Connection,
                    c.owner_id.clone(),
                )];
                acl_items.extend_from_slice(c.access);
                // Builtin Connections cannot be versioned.
                let versions = BTreeMap::new();

                txn.insert_item(
                    catalog_id,
                    c.oid,
                    global_id,
                    mz_catalog::durable::initialize::resolve_system_schema(c.schema).id,
                    c.name,
                    c.sql.into(),
                    *c.owner_id,
                    acl_items,
                    versions,
                    None,
                )?;
                true
            }
            _ => false,
        };
        assert_eq!(
            builtin.runtime_alterable(),
            handled_runtime_alterable,
            "runtime alterable object was not handled by migration",
        );
    }
    txn.set_system_object_mappings(new_builtin_mappings)?;

    // Update comments of all builtin objects
    let builtins_with_catalog_ids = existing_builtins
        .iter()
        .map(|(b, m, _)| (*b, m.unique_identifier.catalog_id))
        .chain(
            new_builtins
                .into_iter()
                .map(|((b, _), (catalog_id, _))| (b, catalog_id)),
        );

    for (builtin, id) in builtins_with_catalog_ids {
        let (comment_id, desc, comments) = match builtin {
            Builtin::Source(s) => (CommentObjectId::Source(id), &s.desc, &s.column_comments),
            Builtin::View(v) => (CommentObjectId::View(id), &v.desc, &v.column_comments),
            Builtin::Table(t) => (CommentObjectId::Table(id), &t.desc, &t.column_comments),
            Builtin::MaterializedView(mv) => (
                CommentObjectId::MaterializedView(id),
                &mv.desc,
                &mv.column_comments,
            ),
            Builtin::Log(_)
            | Builtin::Type(_)
            | Builtin::Func(_)
            | Builtin::Index(_)
            | Builtin::Connection(_) => continue,
        };
        // Drop comments under every relation-style `CommentObjectId` variant
        // for this id, not just the current one. When a builtin's type changes
        // (e.g. Table -> MaterializedView) but its catalog id is preserved, the
        // prior type's comment rows would otherwise linger forever.
        txn.drop_comments(&BTreeSet::from_iter([
            CommentObjectId::Table(id),
            CommentObjectId::View(id),
            CommentObjectId::MaterializedView(id),
            CommentObjectId::Source(id),
        ]))?;

        let mut comments = comments.clone();
        for (col_idx, name) in desc.iter_names().enumerate() {
            if let Some(comment) = comments.remove(name.as_str()) {
                // Comment column indices are 1 based
                txn.update_comment(comment_id, Some(col_idx + 1), Some(comment.to_owned()))?;
            }
        }
        assert!(
            comments.is_empty(),
            "builtin object contains dangling comments that don't correspond to columns {comments:?}"
        );
    }

    // Anything left in `system_object_mappings` must have been deleted and should be removed from
    // the catalog.
    let mut deleted_system_objects = BTreeSet::new();
    let mut deleted_runtime_alterable_system_ids = BTreeSet::new();
    let mut deleted_comments = BTreeSet::new();
    for (desc, mapping) in system_object_mappings {
        deleted_system_objects.insert(mapping.description);
        if mapping.unique_identifier.fingerprint == RUNTIME_ALTERABLE_FINGERPRINT_SENTINEL {
            deleted_runtime_alterable_system_ids.insert(mapping.unique_identifier.catalog_id);
        }

        let id = mapping.unique_identifier.catalog_id;
        let comment_id = match desc.object_type {
            CatalogItemType::Table => CommentObjectId::Table(id),
            CatalogItemType::Source => CommentObjectId::Source(id),
            CatalogItemType::View => CommentObjectId::View(id),
            CatalogItemType::MaterializedView => CommentObjectId::MaterializedView(id),
            CatalogItemType::Sink
            | CatalogItemType::MetricSink
            | CatalogItemType::Index
            | CatalogItemType::Type
            | CatalogItemType::Func
            | CatalogItemType::Secret
            | CatalogItemType::Connection => continue,
        };
        deleted_comments.insert(comment_id);
    }
    // If you are 100% positive that it is safe to delete a system object outside any of the
    // unstable schemas, then add it to this set. Make sure that no prod environments are
    // using this object and that the upgrade checker does not show any issues.
    //
    // Objects can be removed from this set after one release.
    let delete_exceptions: HashSet<SystemObjectDescription> = [].into();
    // TODO(jkosh44) Technically we could support changing the type of a builtin object outside
    // of unstable schemas (i.e. from a table to a view). However, builtin migrations don't currently
    // handle that scenario correctly.
    assert!(
        deleted_system_objects
            .iter()
            // It's okay if Indexes change because they're inherently ephemeral.
            .filter(|object| object.object_type != CatalogItemType::Index)
            .all(
                |deleted_object| is_unstable_schema(&deleted_object.schema_name)
                    || delete_exceptions.contains(deleted_object)
            ),
        "only objects in unstable schemas can be deleted, deleted objects: {:?}",
        deleted_system_objects
    );
    txn.drop_comments(&deleted_comments)?;
    txn.remove_items(&deleted_runtime_alterable_system_ids)?;
    txn.remove_system_object_mappings(deleted_system_objects)?;

    // Filter down to just the GlobalIds which are used to track the underlying collections.
    let new_builtin_collections = new_builtin_ids
        .into_iter()
        .map(|(_catalog_id, global_id)| global_id)
        .collect();

    Ok(new_builtin_collections)
}

fn add_new_remove_old_builtin_clusters_migration(
    txn: &mut mz_catalog::durable::Transaction<'_>,
    builtin_cluster_config_map: &BuiltinBootstrapClusterConfigMap,
    boot_ts: Timestamp,
) -> Result<(), mz_catalog::durable::CatalogError> {
    let mut durable_clusters: BTreeMap<_, _> = txn
        .get_clusters()
        .filter(|cluster| cluster.id.is_system())
        .map(|cluster| (cluster.name.to_string(), cluster))
        .collect();

    // Add new clusters.
    for builtin_cluster in BUILTIN_CLUSTERS {
        if durable_clusters.remove(builtin_cluster.name).is_none() {
            let cluster_config = builtin_cluster_config_map.get_config(builtin_cluster.name)?;

            let cluster_id = txn.insert_system_cluster(
                builtin_cluster.name,
                vec![],
                builtin_cluster.privileges.to_vec(),
                builtin_cluster.owner_id.to_owned(),
                mz_catalog::durable::ClusterConfig {
                    variant: mz_catalog::durable::ClusterVariant::Managed(ClusterVariantManaged {
                        size: cluster_config.size,
                        availability_zones: vec![],
                        replication_factor: cluster_config.replication_factor,
                        logging: default_logging_config(),
                        arrangement_compression: false,
                        optimizer_feature_overrides: Default::default(),
                        schedule: Default::default(),
                        auto_scaling_strategy: None,
                        reconfiguration: None,
                        burst: None,
                    }),
                    workload_class: None,
                },
                &HashSet::new(),
            )?;

            let audit_id = txn.allocate_audit_log_id()?;
            txn.insert_audit_log_event(VersionedEvent::new(
                audit_id,
                EventType::Create,
                ObjectType::Cluster,
                EventDetails::IdNameV1(mz_audit_log::IdNameV1 {
                    id: cluster_id.to_string(),
                    name: builtin_cluster.name.to_string(),
                }),
                None,
                boot_ts.into(),
            ));
        }
    }

    // Remove old clusters.
    let old_clusters = durable_clusters
        .values()
        .map(|cluster| cluster.id)
        .collect();
    txn.remove_clusters(&old_clusters)?;

    for (_name, cluster) in &durable_clusters {
        let audit_id = txn.allocate_audit_log_id()?;
        txn.insert_audit_log_event(VersionedEvent::new(
            audit_id,
            EventType::Drop,
            ObjectType::Cluster,
            EventDetails::IdNameV1(mz_audit_log::IdNameV1 {
                id: cluster.id.to_string(),
                name: cluster.name.clone(),
            }),
            None,
            boot_ts.into(),
        ));
    }

    Ok(())
}

fn add_new_remove_old_builtin_introspection_source_migration(
    txn: &mut mz_catalog::durable::Transaction<'_>,
) -> Result<(), AdapterError> {
    let mut new_indexes = Vec::new();
    let mut removed_indexes = BTreeSet::new();
    for cluster in txn.get_clusters() {
        let mut introspection_source_index_ids = txn.get_introspection_source_indexes(cluster.id);

        let mut new_logs = Vec::new();

        for log in BUILTINS::logs() {
            if introspection_source_index_ids.remove(log.name).is_none() {
                new_logs.push(log);
            }
        }

        for log in new_logs {
            let (item_id, gid) =
                Transaction::allocate_introspection_source_index_id(&cluster.id, log.variant);
            new_indexes.push((cluster.id, log.name.to_string(), item_id, gid));
        }

        // Anything left in `introspection_source_index_ids` must have been deleted and should be
        // removed from the catalog.
        removed_indexes.extend(
            introspection_source_index_ids
                .into_keys()
                .map(|name| (cluster.id, name.to_string())),
        );
    }
    txn.insert_introspection_source_indexes(new_indexes, &HashSet::new())?;
    txn.remove_introspection_source_indexes(removed_indexes)?;
    Ok(())
}

fn add_new_remove_old_builtin_roles_migration(
    txn: &mut mz_catalog::durable::Transaction<'_>,
) -> Result<(), mz_catalog::durable::CatalogError> {
    let mut durable_roles: BTreeMap<_, _> = txn
        .get_roles()
        .filter(|role| role.id.is_system() || role.id.is_predefined())
        .map(|role| (role.name.to_string(), role))
        .collect();

    // Add new roles.
    for builtin_role in BUILTIN_ROLES {
        if durable_roles.remove(builtin_role.name).is_none() {
            txn.insert_builtin_role(
                builtin_role.id,
                builtin_role.name.to_string(),
                builtin_role.attributes.clone(),
                RoleMembership::new(),
                RoleVars::default(),
                builtin_role.oid,
            )?;
        }
    }

    // Remove old roles.
    let old_roles = durable_roles.values().map(|role| role.id).collect();
    txn.remove_roles(&old_roles)?;

    Ok(())
}

/// Converges each builtin cluster's replica set on the cluster's own managed
/// config.
///
/// A builtin cluster is normally managed, and a managed cluster's replicas are
/// derived state: exactly `replication_factor` replicas, named by
/// [`managed_cluster_replica_name`]. One can be altered to unmanaged, which leaves
/// no factor to derive from, and then its replica set is the operator's.
///
/// Replicas this creates are shaped from the cluster's config. An existing replica
/// is matched by name alone and left untouched, so this converges cardinality and
/// names rather than shape. An internal replica is never derived state and is left
/// alone entirely.
///
/// The bootstrap flags seed `replication_factor` and `size` when a cluster is
/// first created (see [`add_new_remove_old_builtin_clusters_migration`]) and are
/// deliberately not consulted here, so an `ALTER CLUSTER` against a builtin
/// cluster survives a restart.
///
/// This runs at catalog open so the replicas a cluster's config calls for exist as
/// early as possible. The coordinator's bootstrap brings up only replicas already
/// recorded durably and runs before the cluster controller is spawned, and the
/// controller does not run at all while a deployment is read-only.
///
/// The cluster controller owns these replica sets at runtime and derives its
/// target from the same cluster config, so the two converge on the same set
/// rather than competing for it. The controller matches replicas by shape and
/// count, never by name, so the `r1..rN` this creates satisfy it. This converges
/// by name, so a boot after the controller reshaped a cluster renames or
/// re-creates replicas it had materialized under generator names. That is
/// harmless: every replica is a cold process at boot anyway, so the cost is
/// replica-id and audit-log noise.
fn reconcile_builtin_cluster_replicas(
    txn: &mut Transaction<'_>,
    builtin_cluster_config_map: &BuiltinBootstrapClusterConfigMap,
    boot_ts: Timestamp,
) -> Result<(), AdapterError> {
    let builtin_cluster_names: BTreeSet<&str> = BUILTIN_CLUSTERS
        .iter()
        .map(|cluster| cluster.name)
        .collect();

    // Replicas of a cluster that is no longer a builtin need no handling here:
    // `Transaction::remove_clusters` cascades to a cluster's replicas, and the
    // clusters migration has already run against this transaction.
    let clusters: Vec<_> = txn
        .get_clusters()
        .filter(|cluster| {
            cluster.id.is_system() && builtin_cluster_names.contains(cluster.name.as_str())
        })
        .collect();

    let builtin_cluster_ids: BTreeSet<ClusterId> =
        clusters.iter().map(|cluster| cluster.id).collect();

    // Only the builtin clusters' replicas. An environment can have thousands of
    // user replicas and none of them are in scope here.
    //
    // An internal replica is deliberately out of scope. `CREATE CLUSTER REPLICA ...
    // INTERNAL` is allowed on a managed cluster, and its name is barred from
    // matching the derived `r1..rN` pattern so it cannot collide with one. It is a
    // break-glass replica an operator added rather than derived state, so reaping it
    // would undo that. The `ALTER CLUSTER` path skips it for the same reason.
    let mut replicas_by_cluster: BTreeMap<ClusterId, BTreeMap<String, ClusterReplica>> =
        BTreeMap::new();
    for replica in txn.get_cluster_replicas().filter(|replica| {
        builtin_cluster_ids.contains(&replica.cluster_id)
            && !matches!(
                replica.config.location,
                ReplicaLocation::Managed { internal: true, .. }
            )
    }) {
        replicas_by_cluster
            .entry(replica.cluster_id)
            .or_default()
            .insert(replica.name.clone(), replica);
    }

    let mut to_drop: Vec<(String, ClusterReplica)> = Vec::new();

    for cluster in clusters {
        // An unmanaged cluster has no replication factor, so there is no target to
        // converge on. Its replica set is whatever an operator made it, and it is
        // not ours to reshape.
        let ClusterVariant::Managed(managed) = &cluster.config.variant else {
            continue;
        };

        // The bootstrap flags seed a cluster at creation and are never re-applied,
        // which is what keeps an `ALTER CLUSTER` from being reverted on every
        // restart. That leaves an operator who changed a flag on an existing
        // deployment with no feedback, so say so.
        let bootstrap_config = builtin_cluster_config_map.get_config(&cluster.name)?;
        if bootstrap_config.replication_factor != managed.replication_factor {
            warn!(
                cluster = %cluster.name,
                configured_replication_factor = managed.replication_factor,
                bootstrap_replication_factor = bootstrap_config.replication_factor,
                "bootstrap replication factor is not applied to an already-existing \
                 builtin cluster. Use ALTER CLUSTER ... SET (REPLICATION FACTOR ...) \
                 to change it",
            );
        }

        // Reading the cluster's factor is what makes this compose with the other
        // writers of a replica set. The controller's on-refresh strategy parks a
        // scheduled cluster by writing its factor to 0, so converging on the
        // factor honors that instead of resurrecting a replica it just dropped.
        let mut surplus = replicas_by_cluster.remove(&cluster.id).unwrap_or_default();
        for index in 0..managed.replication_factor {
            let replica_name = managed_cluster_replica_name(index);
            if surplus.remove(&replica_name).is_some() {
                continue;
            }

            // Builtin replicas live on system clusters. This runs inside the
            // catalog-open transaction with no coordinator, so allocating from the
            // same transaction is single-source and safe.
            let replica_id = txn.allocate_system_replica_id()?;
            txn.insert_cluster_replica_with_id(
                cluster.id,
                replica_id,
                &replica_name,
                managed_replica_config(managed),
                // The cluster's owner, not `mz_system`. `mz_support` and
                // `mz_analytics` are owned by their own roles, and replica
                // ownership is checked against the replica's own `owner_id`, so
                // stamping `mz_system` here would stop those roles from altering
                // a replica of a cluster they own.
                cluster.owner_id,
            )?;
            info!(
                cluster = %cluster.name, replica = %replica_name, %replica_id,
                "creating builtin cluster replica to match the cluster's replication factor"
            );

            let audit_id = txn.allocate_audit_log_id()?;
            txn.insert_audit_log_event(VersionedEvent::new(
                audit_id,
                EventType::Create,
                ObjectType::ClusterReplica,
                EventDetails::CreateClusterReplicaV4(mz_audit_log::CreateClusterReplicaV4 {
                    cluster_id: cluster.id.to_string(),
                    cluster_name: cluster.name.clone(),
                    replica_id: Some(replica_id.to_string()),
                    replica_name,
                    logical_size: managed.size.clone(),
                    billed_as: None,
                    internal: false,
                    reason: CreateOrDropClusterReplicaReasonV1::System,
                    scheduling_policies: None,
                }),
                None,
                boot_ts.into(),
            ));
        }

        // Whatever the config does not call for is surplus: a replica left over from
        // a higher replication factor, or a `-pending` replica belonging to a
        // graceful reconfiguration that a restart interrupted. Dropping the latter
        // here means `remove_pending_cluster_replicas_migration` never sees it,
        // which matches the abort semantics it would have applied anyway.
        to_drop.extend(
            surplus
                .into_values()
                .map(|replica| (cluster.name.clone(), replica)),
        );
    }

    // Batched: `Transaction::remove_cluster_replica` is linear in the total replica
    // count, so it must not be called in a loop.
    let drop_ids = to_drop
        .iter()
        .map(|(_cluster_name, replica)| replica.replica_id)
        .collect();
    txn.remove_cluster_replicas(&drop_ids)?;

    for (cluster_name, replica) in to_drop {
        info!(
            cluster = %cluster_name, replica = %replica.name, replica_id = %replica.replica_id,
            "dropping builtin cluster replica not called for by the cluster's replication factor"
        );

        let audit_id = txn.allocate_audit_log_id()?;
        txn.insert_audit_log_event(VersionedEvent::new(
            audit_id,
            EventType::Drop,
            ObjectType::ClusterReplica,
            EventDetails::DropClusterReplicaV3(mz_audit_log::DropClusterReplicaV3 {
                cluster_id: replica.cluster_id.to_string(),
                cluster_name,
                replica_id: Some(replica.replica_id.to_string()),
                replica_name: replica.name,
                reason: CreateOrDropClusterReplicaReasonV1::System,
                scheduling_policies: None,
            }),
            None,
            boot_ts.into(),
        ));
    }

    Ok(())
}

/// The replica config implied by a managed cluster's config.
///
/// Every field a replica shares with its cluster is taken from the cluster.
/// `availability_zones` is the cluster's pool, which is what the provisioning
/// paths stamp onto a managed replica.
///
/// Two things the `ALTER CLUSTER` path does that this does not. It normalizes
/// logging through the plan, which discards `log_logging` when no interval is
/// set, so for the odd cluster config that has debugging on and logging off the
/// two produce different records. And it validates the size against the replica
/// size map and the role's allowed sizes, which cannot be done from here because
/// the size map is not part of the durable catalog. A builtin cluster's size only
/// ever comes from a bootstrap flag or an `ALTER` that already validated it.
fn managed_replica_config(managed: &ClusterVariantManaged) -> ReplicaConfig {
    // Exhaustive destructure (no `..`): a field added to the managed config is a
    // compile error here until we decide whether a replica has to carry it.
    let ClusterVariantManaged {
        size,
        availability_zones,
        logging,
        arrangement_compression,
        replication_factor: _,
        optimizer_feature_overrides: _,
        schedule: _,
        auto_scaling_strategy: _,
        reconfiguration: _,
        burst: _,
    } = managed;
    ReplicaConfig {
        location: ReplicaLocation::Managed {
            size: size.clone(),
            availability_zones: availability_zones.clone(),
            billed_as: None,
            internal: false,
            pending: false,
        },
        logging: logging.clone(),
        arrangement_compression: *arrangement_compression,
    }
}

/// Roles can have default values for configuration parameters, e.g. you can set a Role default for
/// the 'cluster' parameter.
///
/// This migration exists to remove the Role default for a configuration parameter, if the persisted
/// input is no longer valid. For example if we remove a configuration parameter or change the
/// accepted set of values.
fn remove_invalid_config_param_role_defaults_migration(
    txn: &mut Transaction<'_>,
) -> Result<(), AdapterError> {
    static BUILD_INFO: mz_build_info::BuildInfo = mz_build_info::build_info!();

    let roles_to_migrate: BTreeMap<_, _> = txn
        .get_roles()
        .filter_map(|mut role| {
            // Create an empty SessionVars just so we can check if a var is valid.
            //
            // TODO(parkmycar): This is a bit hacky, instead we should have a static list of all
            // session variables.
            let session_vars = SessionVars::new_unchecked(&BUILD_INFO, SYSTEM_USER.clone(), None);

            // Iterate over all of the variable defaults for this role.
            let mut invalid_roles_vars = BTreeMap::new();
            for (name, value) in &role.vars.map {
                // If one does not exist or its value is invalid, then mark it for removal.
                let Ok(session_var) = session_vars.inspect(name) else {
                    invalid_roles_vars.insert(name.clone(), value.clone());
                    continue;
                };
                if session_var.check(value.borrow()).is_err() {
                    invalid_roles_vars.insert(name.clone(), value.clone());
                }
            }

            // If the role has no invalid values, nothing to do!
            if invalid_roles_vars.is_empty() {
                return None;
            }

            tracing::warn!(?role, ?invalid_roles_vars, "removing invalid role vars");

            // Otherwise, remove the variables from the role and return it to be updated.
            for (name, _value) in invalid_roles_vars {
                role.vars.map.remove(&name);
            }
            Some(role)
        })
        .map(|role| (role.id, role))
        .collect();

    txn.update_roles_without_auth(roles_to_migrate)?;

    Ok(())
}

/// Drops replicas left durably marked `pending`.
///
/// No runtime path creates one anymore. An upgrade can still come from a version
/// whose staged reconfiguration machine crashed between the pending-create commit
/// and the finalize, and those replicas are excluded from the cluster
/// controller's ownership test, so this catalog-open sweep is their only
/// remaining cleaner. It goes away together with the durable `pending` field,
/// once no supported upgrade source can still write one.
fn remove_pending_cluster_replicas_migration(
    tx: &mut Transaction,
    boot_ts: mz_repr::Timestamp,
) -> Result<(), anyhow::Error> {
    // Build a map of cluster_id -> cluster_name for audit events.
    let cluster_names: BTreeMap<_, _> = tx.get_clusters().map(|c| (c.id, c.name)).collect();

    let occurred_at = boot_ts.into();

    for replica in tx.get_cluster_replicas().collect::<Vec<_>>() {
        if let mz_catalog::durable::ReplicaLocation::Managed { pending: true, .. } =
            replica.config.location
        {
            let cluster_name = cluster_names
                .get(&replica.cluster_id)
                .cloned()
                .unwrap_or_else(|| "<unknown>".to_string());

            info!(
                "removing pending cluster replica '{}' from cluster '{}'",
                replica.name, cluster_name,
            );

            tx.remove_cluster_replica(replica.replica_id)?;

            // Emit an audit log event so that the drop is visible in
            // mz_audit_events, matching the create event that was
            // recorded when the pending replica was first created.
            let audit_id = tx.allocate_audit_log_id()?;
            tx.insert_audit_log_event(VersionedEvent::new(
                audit_id,
                EventType::Drop,
                ObjectType::ClusterReplica,
                EventDetails::DropClusterReplicaV3(mz_audit_log::DropClusterReplicaV3 {
                    cluster_id: replica.cluster_id.to_string(),
                    cluster_name,
                    replica_id: Some(replica.replica_id.to_string()),
                    replica_name: replica.name,
                    reason: CreateOrDropClusterReplicaReasonV1::System,
                    scheduling_policies: None,
                }),
                None,
                occurred_at,
            ));
        }
    }
    Ok(())
}

fn default_logging_config() -> ReplicaLogging {
    ReplicaLogging {
        log_logging: false,
        interval: Some(Duration::from_secs(1)),
    }
}

#[derive(Debug)]
pub struct BuiltinBootstrapClusterConfigMap {
    /// Size and replication factor to default system_cluster on bootstrap
    pub system_cluster: BootstrapBuiltinClusterConfig,
    /// Size and replication factor to default catalog_server_cluster on bootstrap
    pub catalog_server_cluster: BootstrapBuiltinClusterConfig,
    /// Size and replication factor to default probe_cluster on bootstrap
    pub probe_cluster: BootstrapBuiltinClusterConfig,
    /// Size and replication factor to default support_cluster on bootstrap
    pub support_cluster: BootstrapBuiltinClusterConfig,
    /// Size to default analytics_cluster on bootstrap
    pub analytics_cluster: BootstrapBuiltinClusterConfig,
}

impl BuiltinBootstrapClusterConfigMap {
    /// Gets the size of the builtin cluster based on the provided name
    fn get_config(
        &self,
        cluster_name: &str,
    ) -> Result<BootstrapBuiltinClusterConfig, mz_catalog::durable::CatalogError> {
        let cluster_config = if cluster_name == mz_catalog::builtin::MZ_SYSTEM_CLUSTER.name {
            &self.system_cluster
        } else if cluster_name == mz_catalog::builtin::MZ_CATALOG_SERVER_CLUSTER.name {
            &self.catalog_server_cluster
        } else if cluster_name == mz_catalog::builtin::MZ_PROBE_CLUSTER.name {
            &self.probe_cluster
        } else if cluster_name == mz_catalog::builtin::MZ_SUPPORT_CLUSTER.name {
            &self.support_cluster
        } else if cluster_name == mz_catalog::builtin::MZ_ANALYTICS_CLUSTER.name {
            &self.analytics_cluster
        } else {
            return Err(mz_catalog::durable::CatalogError::Catalog(
                SqlCatalogError::UnexpectedBuiltinCluster(cluster_name.to_owned()),
            ));
        };
        Ok(cluster_config.clone())
    }
}

/// Convert `updates` into a `Vec` that can be consolidated by doing the following:
///
///   - Update the timestamp of each update to the same value.
///   - Convert the diff of each update to a type that implements
///     [`differential_dataflow::difference::Semigroup`].
///
/// WARNING: Do not call outside of startup.
pub(crate) fn into_consolidatable_updates_startup(
    updates: Vec<StateUpdate>,
    ts: Timestamp,
) -> Vec<(StateUpdateKind, Timestamp, Diff)> {
    updates
        .into_iter()
        .map(|StateUpdate { kind, ts: _, diff }| (kind, ts, Diff::from(diff)))
        .collect()
}

#[cfg(test)]
mod tests {
    use mz_catalog::durable::ClusterVariantManaged;

    use super::*;

    /// A replica the reconciler creates takes every shared field from its
    /// cluster. Nothing here may fall back to a default, or a builtin cluster
    /// ends up running a replica that disagrees with its own config.
    #[mz_ore::test]
    fn test_managed_replica_config_derives_every_shared_field() {
        let managed = ClusterVariantManaged {
            size: "somesize".into(),
            availability_zones: vec!["az1".into(), "az2".into()],
            logging: ReplicaLogging {
                log_logging: true,
                interval: Some(Duration::from_millis(10)),
            },
            arrangement_compression: true,
            replication_factor: 3,
            optimizer_feature_overrides: Default::default(),
            schedule: Default::default(),
            auto_scaling_strategy: None,
            reconfiguration: None,
            burst: None,
        };

        let config = managed_replica_config(&managed);

        // Exhaustive destructures throughout, so a field added to either type is a
        // compile error here rather than a silently unasserted one.
        let ReplicaConfig {
            location,
            logging,
            arrangement_compression,
        } = config;

        assert_eq!(logging, managed.logging);
        assert_eq!(arrangement_compression, managed.arrangement_compression);
        match location {
            ReplicaLocation::Managed {
                size,
                availability_zones,
                billed_as,
                internal,
                pending,
            } => {
                assert_eq!(size, managed.size);
                assert_eq!(availability_zones, managed.availability_zones);
                // A builtin replica is neither manually managed nor part of an
                // in-flight reconfiguration, and those three traits are what
                // exclude a replica from controller ownership.
                assert_eq!(billed_as, None);
                assert!(!internal);
                assert!(!pending);
            }
            ReplicaLocation::Unmanaged { .. } => panic!("expected a managed location"),
        }
    }
}