mz_adapter/catalog/

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

//! In-memory metadata storage for the coordinator.

use std::borrow::Cow;
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::fmt::Debug;
use std::sync::Arc;
use std::sync::LazyLock;
use std::time::Instant;

use ipnet::IpNet;
use itertools::Itertools;
use mz_adapter_types::compaction::CompactionWindow;
use mz_adapter_types::connection::ConnectionId;
use mz_audit_log::{EventDetails, EventType, ObjectType, VersionedEvent};
use mz_build_info::DUMMY_BUILD_INFO;
use mz_catalog::SYSTEM_CONN_ID;
use mz_catalog::builtin::{
    BUILTINS, Builtin, BuiltinCluster, BuiltinLog, BuiltinSource, BuiltinTable, BuiltinType,
};
use mz_catalog::config::{AwsPrincipalContext, ClusterReplicaSizeMap};
use mz_catalog::expr_cache::LocalExpressions;
use mz_catalog::memory::error::{Error, ErrorKind};
use mz_catalog::memory::objects::{
    CatalogCollectionEntry, CatalogEntry, CatalogItem, Cluster, ClusterReplica, CommentsMap,
    Connection, DataSourceDesc, Database, DefaultPrivileges, Index, MaterializedView, MetricSink,
    NetworkPolicy, Role, RoleAuth, Schema, Secret, Sink, Source, SourceReferences, Table,
    TableDataSource, Type, View,
};
use mz_controller::clusters::{
    ManagedReplicaLocation, ReplicaAllocation, ReplicaLocation, UnmanagedReplicaLocation,
};
use mz_controller_types::{ClusterId, ReplicaId};
use mz_expr::{CollectionPlan, OptimizedMirRelationExpr};
use mz_license_keys::ValidatedLicenseKey;
use mz_ore::collections::CollectionExt;
use mz_ore::now::NOW_ZERO;
use mz_ore::soft_assert_no_log;
use mz_ore::str::StrExt;
use mz_pgrepr::oid::INVALID_OID;
use mz_repr::adt::mz_acl_item::PrivilegeMap;
use mz_repr::namespaces::{
    INFORMATION_SCHEMA, MZ_CATALOG_SCHEMA, MZ_CATALOG_UNSTABLE_SCHEMA, MZ_INTERNAL_SCHEMA,
    MZ_INTROSPECTION_SCHEMA, MZ_TEMP_SCHEMA, MZ_UNSAFE_SCHEMA, PG_CATALOG_SCHEMA, SYSTEM_SCHEMAS,
    UNSTABLE_SCHEMAS,
};
use mz_repr::network_policy_id::NetworkPolicyId;
use mz_repr::optimize::{OptimizerFeatureOverrides, OptimizerFeatures, OverrideFrom};
use mz_repr::role_id::RoleId;
use mz_repr::{
    CatalogItemId, GlobalId, RelationDesc, RelationVersion, RelationVersionSelector,
    VersionedRelationDesc,
};
use mz_secrets::InMemorySecretsController;
use mz_sql::ast::Ident;
use mz_sql::catalog::{
    CatalogCluster, CatalogClusterReplica, CatalogDatabase, CatalogError as SqlCatalogError,
    CatalogItem as SqlCatalogItem, CatalogItemType, CatalogRecordField, CatalogRole, CatalogSchema,
    CatalogType, CatalogTypeDetails, IdReference, NameReference, SessionCatalog, SystemObjectType,
    TypeReference,
};
use mz_sql::catalog::{CatalogConfig, EnvironmentId};
use mz_sql::names::{
    CommentObjectId, DatabaseId, DependencyIds, FullItemName, FullSchemaName, ObjectId,
    PartialItemName, QualifiedItemName, QualifiedSchemaName, RawDatabaseSpecifier,
    ResolvedDatabaseSpecifier, ResolvedIds, SchemaId, SchemaSpecifier, SystemObjectId,
};
use mz_sql::plan::{
    CreateConnectionPlan, CreateIndexPlan, CreateMaterializedViewPlan, CreateMetricSinkPlan,
    CreateSecretPlan, CreateSinkPlan, CreateSourcePlan, CreateTablePlan, CreateTypePlan,
    CreateViewPlan, Params, Plan, PlanContext,
};
use mz_sql::rbac;
use mz_sql::session::metadata::SessionMetadata;
use mz_sql::session::user::MZ_SYSTEM_ROLE_ID;
use mz_sql::session::vars::{DEFAULT_DATABASE_NAME, SystemVars, Var, VarInput};
use mz_sql_parser::ast::QualifiedReplica;
use mz_storage_client::controller::StorageMetadata;
use mz_storage_types::connections::ConnectionContext;
use mz_storage_types::connections::inline::{
    ConnectionResolver, InlinedConnection, IntoInlineConnection,
};
use mz_transform::notice::OptimizerNotice;
use serde::Serialize;
use timely::progress::Antichain;
use tokio::sync::mpsc;
use tracing::{debug, warn};
use uuid::Uuid;

// DO NOT add any more imports from `crate` outside of `crate::catalog`.
use crate::AdapterError;
use crate::catalog::{Catalog, ConnCatalog};
use crate::config::ScopedParameters;
use crate::coord::{ConnMeta, infer_sql_type_for_catalog};
use crate::optimize::{self, Optimize, OptimizerCatalog};
use crate::session::Session;

/// The in-memory representation of the Catalog. This struct is not directly used to persist
/// metadata to persistent storage. For persistent metadata see
/// [`mz_catalog::durable::DurableCatalogState`].
///
/// [`Serialize`] is implemented to create human readable dumps of the in-memory state, not for
/// storing the contents of this struct on disk.
#[derive(Debug, Clone, Serialize)]
pub struct CatalogState {
    // State derived from the durable catalog. These fields should only be mutated in `open.rs` or
    // `apply.rs`. Some of these fields are not 100% derived from the durable catalog. Those
    // include:
    //  - Temporary items.
    //  - Certain objects are partially derived from read-only state.
    pub(super) database_by_name: imbl::OrdMap<String, DatabaseId>,
    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
    pub(super) database_by_id: imbl::OrdMap<DatabaseId, Database>,
    #[serde(serialize_with = "skip_temp_items")]
    pub(super) entry_by_id: imbl::OrdMap<CatalogItemId, CatalogEntry>,
    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
    pub(super) entry_by_global_id: imbl::OrdMap<GlobalId, CatalogItemId>,
    pub(super) ambient_schemas_by_name: imbl::OrdMap<String, SchemaId>,
    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
    pub(super) ambient_schemas_by_id: imbl::OrdMap<SchemaId, Schema>,
    pub(super) clusters_by_name: imbl::OrdMap<String, ClusterId>,
    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
    pub(super) clusters_by_id: imbl::OrdMap<ClusterId, Cluster>,
    pub(super) roles_by_name: imbl::OrdMap<String, RoleId>,
    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
    pub(super) roles_by_id: imbl::OrdMap<RoleId, Role>,
    pub(super) network_policies_by_name: imbl::OrdMap<String, NetworkPolicyId>,
    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
    pub(super) network_policies_by_id: imbl::OrdMap<NetworkPolicyId, NetworkPolicy>,
    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
    pub(super) role_auth_by_id: imbl::OrdMap<RoleId, RoleAuth>,

    #[serde(skip)]
    pub(super) system_configuration: Arc<SystemVars>,
    /// In-memory mirror of the durable scoped (per-cluster and per-replica)
    /// system-parameter cache, maintained by `apply.rs` from the durable
    /// collections. Resolution reads from here: the optimizer's per-cluster
    /// feature overrides (`cluster_scoped_optimizer_overrides`) and the
    /// coordinator's per-replica dyncfg push. See the scoped feature flags
    /// design. Skipped in the consistency-check snapshot because it is fully
    /// derived from the durable catalog.
    #[serde(skip)]
    pub(super) scoped_system_parameters: ScopedParameters,
    pub(super) default_privileges: Arc<DefaultPrivileges>,
    pub(super) system_privileges: Arc<PrivilegeMap>,
    pub(super) comments: Arc<CommentsMap>,
    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
    pub(super) source_references: imbl::OrdMap<CatalogItemId, SourceReferences>,
    pub(super) storage_metadata: Arc<StorageMetadata>,
    pub(super) mock_authentication_nonce: Option<String>,

    // Mutable state not derived from the durable catalog. Populated
    // during dataflow bootstrapping (`bootstrap_dataflow_plans`), which
    // doesn't run in Testdrive's read-only consistency check, so this
    // must be `#[serde(skip)]`.
    #[serde(skip)]
    pub(super) notices_by_dep_id: imbl::OrdMap<GlobalId, Vec<Arc<OptimizerNotice>>>,

    // Populated by active connections creating temporary objects. The
    // read-only catalog opened by Testdrive's consistency check has no
    // active connections, so this must be `#[serde(skip)]`.
    #[serde(skip)]
    pub(super) temporary_namespaces: TemporaryNamespaces,

    // Read-only state not derived from the durable catalog.
    #[serde(skip)]
    pub(super) config: mz_sql::catalog::CatalogConfig,
    pub(super) cluster_replica_sizes: ClusterReplicaSizeMap,
    #[serde(skip)]
    pub(crate) availability_zones: Vec<String>,

    // Read-only not derived from the durable catalog.
    #[serde(skip)]
    pub(super) egress_addresses: Vec<IpNet>,
    pub(super) aws_principal_context: Option<AwsPrincipalContext>,
    pub(super) aws_privatelink_availability_zones: Option<BTreeSet<String>>,
    pub(super) http_host_name: Option<String>,

    // Read-only not derived from the durable catalog.
    #[serde(skip)]
    pub(super) license_key: ValidatedLicenseKey,
}

/// The temporary namespaces of the sessions connected to this process: for
/// each session that has created a temporary item, the ephemeral-owner
/// mapping between the session's UUID (stamped on its durable temporary
/// items) and its connection, together with the session's `mz_temp`
/// [`Schema`] once it has materialized.
///
/// The coordinator registers a namespace at a session's first
/// temporary-item creation, and unregisters it when the session terminates.
#[derive(Debug, Clone, Default)]
pub(super) struct TemporaryNamespaces {
    by_conn: imbl::OrdMap<ConnectionId, TemporaryNamespace>,
    // For resolving a durable item's owner UUID to its connection in
    // the apply path.
    conns_by_uuid: imbl::OrdMap<Uuid, ConnectionId>,
}

#[derive(Debug, Clone)]
struct TemporaryNamespace {
    uuid: Uuid,
    // Instantiated by `ensure_schema` at the first applied temporary item
    schema: Option<Schema>,
}

impl TemporaryNamespaces {
    /// Registers `conn_id` as the connection of the session `uuid`.
    ///
    /// Callers guard on [`CatalogState::has_temporary_namespace`], so
    /// registering an already-registered connection is a bug.
    pub(super) fn register(&mut self, conn_id: ConnectionId, uuid: Uuid) {
        let prev_conn = self.conns_by_uuid.insert(uuid, conn_id.clone());
        mz_ore::soft_assert_or_log!(
            prev_conn.is_none(),
            "duplicate temporary namespace registration for {uuid}"
        );
        let prev_ns = self
            .by_conn
            .insert(conn_id, TemporaryNamespace { uuid, schema: None });
        mz_ore::soft_assert_or_log!(
            prev_ns.is_none(),
            "duplicate temporary namespace registration for the connection of {uuid}"
        );
    }

    /// Returns `conn_id`'s `mz_temp` schema, creating an empty one owned by
    /// `owner_id` if one hasn't been instantiated yet.
    pub(super) fn ensure_schema(
        &mut self,
        conn_id: &ConnectionId,
        owner_id: RoleId,
    ) -> &mut Schema {
        let namespace = self
            .by_conn
            .get_mut(conn_id)
            .expect("temporary namespace must be registered before items are applied");
        namespace.schema.get_or_insert_with(|| {
            // Temporary schema OIDs are never used, and it's therefore wasteful to go to the
            // durable catalog to allocate a new OID for every temporary schema. Instead, we give
            // them all the same invalid OID. This matches the semantics of temporary schema
            // `GlobalId`s which are all -1.
            let oid = INVALID_OID;
            Schema {
                name: QualifiedSchemaName {
                    database: ResolvedDatabaseSpecifier::Ambient,
                    schema: MZ_TEMP_SCHEMA.into(),
                },
                id: SchemaSpecifier::Temporary,
                oid,
                items: BTreeMap::new(),
                functions: BTreeMap::new(),
                types: BTreeMap::new(),
                owner_id,
                privileges: PrivilegeMap::from_mz_acl_items(vec![rbac::owner_privilege(
                    mz_sql::catalog::ObjectType::Schema,
                    owner_id,
                )]),
            }
        })
    }

    /// Removes `conn_id`'s temporary namespace.
    pub(super) fn unregister(&mut self, conn_id: &ConnectionId) {
        let Some(namespace) = self.by_conn.get(conn_id) else {
            return;
        };
        // A non-empty schema here means temporary items (and
        // possibly their storage shards) weren't cleaned up, so we surface the
        // invariant violation and keep the namespace registered.
        if namespace
            .schema
            .as_ref()
            .is_some_and(|schema| !schema.items.is_empty())
        {
            mz_ore::soft_panic_or_log!(
                "temporary namespace for connection {conn_id} still has items at unregistration"
            );
            return;
        }
        let uuid = namespace.uuid;
        self.by_conn.remove(conn_id);
        self.conns_by_uuid.remove(&uuid);
    }

    pub(super) fn schema(&self, conn_id: &ConnectionId) -> Option<&Schema> {
        self.by_conn
            .get(conn_id)
            .and_then(|namespace| namespace.schema.as_ref())
    }

    pub(super) fn schema_mut(&mut self, conn_id: &ConnectionId) -> Option<&mut Schema> {
        self.by_conn
            .get_mut(conn_id)
            .and_then(|namespace| namespace.schema.as_mut())
    }

    pub(super) fn schemas(&self) -> impl Iterator<Item = &Schema> {
        self.by_conn
            .values()
            .filter_map(|namespace| namespace.schema.as_ref())
    }

    pub(super) fn conn_for_uuid(&self, uuid: &Uuid) -> Option<&ConnectionId> {
        self.conns_by_uuid.get(uuid)
    }

    pub(super) fn uuid_for_conn(&self, conn_id: &ConnectionId) -> Option<Uuid> {
        self.by_conn.get(conn_id).map(|namespace| namespace.uuid)
    }

    pub(super) fn contains_conn(&self, conn_id: &ConnectionId) -> bool {
        self.by_conn.contains_key(conn_id)
    }

    pub(super) fn contains_uuid(&self, uuid: &Uuid) -> bool {
        self.conns_by_uuid.contains_key(uuid)
    }
}

/// Keeps track of what expressions are cached or not during startup.
/// It's also used during catalog transactions to avoid re-optimizing CREATE VIEW / CREATE MAT VIEW
/// statements when going back and forth between durable catalog operations and in-memory catalog
/// operations.
#[derive(Debug, Clone, Serialize)]
pub(crate) enum LocalExpressionCache {
    /// The cache is being used.
    Open {
        /// The local expressions that were cached in the expression cache.
        cached_exprs: BTreeMap<GlobalId, LocalExpressions>,
        /// The local expressions that were NOT cached in the expression cache.
        uncached_exprs: BTreeMap<GlobalId, LocalExpressions>,
    },
    /// The cache is not being used.
    Closed,
}

impl LocalExpressionCache {
    pub(super) fn new(cached_exprs: BTreeMap<GlobalId, LocalExpressions>) -> Self {
        Self::Open {
            cached_exprs,
            uncached_exprs: BTreeMap::new(),
        }
    }

    pub(super) fn remove_cached_expression(&mut self, id: &GlobalId) -> Option<LocalExpressions> {
        match self {
            LocalExpressionCache::Open { cached_exprs, .. } => cached_exprs.remove(id),
            LocalExpressionCache::Closed => None,
        }
    }

    /// Insert an expression that was cached, back into the cache. This is generally needed when
    /// parsing/planning an expression fails, but we don't want to lose the cached expression.
    pub(super) fn insert_cached_expression(
        &mut self,
        id: GlobalId,
        local_expressions: LocalExpressions,
    ) {
        match self {
            LocalExpressionCache::Open { cached_exprs, .. } => {
                cached_exprs.insert(id, local_expressions);
            }
            LocalExpressionCache::Closed => {}
        }
    }

    /// Inform the cache that `id` was not found in the cache and that we should add it as
    /// `local_mir` and `optimizer_features`.
    pub(super) fn insert_uncached_expression(
        &mut self,
        id: GlobalId,
        local_mir: OptimizedMirRelationExpr,
        optimizer_features: OptimizerFeatures,
    ) {
        match self {
            LocalExpressionCache::Open { uncached_exprs, .. } => {
                let local_expr = LocalExpressions {
                    local_mir,
                    optimizer_features,
                };
                // If we are trying to cache the same item a second time, with a different
                // expression, then we must be migrating the object or doing something else weird.
                // Caching the unmigrated expression may cause us to incorrectly use the unmigrated
                // version after a restart. Caching the migrated version may cause us to incorrectly
                // think that the object has already been migrated. To simplify things, we cache
                // neither.
                let prev = uncached_exprs.remove(&id);
                match prev {
                    Some(prev) if prev == local_expr => {
                        uncached_exprs.insert(id, local_expr);
                    }
                    None => {
                        uncached_exprs.insert(id, local_expr);
                    }
                    Some(_) => {}
                }
            }
            LocalExpressionCache::Closed => {}
        }
    }

    pub(super) fn into_uncached_exprs(self) -> BTreeMap<GlobalId, LocalExpressions> {
        match self {
            LocalExpressionCache::Open { uncached_exprs, .. } => uncached_exprs,
            LocalExpressionCache::Closed => BTreeMap::new(),
        }
    }
}

fn skip_temp_items<S>(
    entries: &imbl::OrdMap<CatalogItemId, CatalogEntry>,
    serializer: S,
) -> Result<S::Ok, S::Error>
where
    S: serde::Serializer,
{
    mz_ore::serde::map_key_to_string(
        entries.iter().filter(|(_k, v)| v.conn_id().is_none()),
        serializer,
    )
}

impl CatalogState {
    /// Returns an empty [`CatalogState`] that can be used in tests.
    // TODO: Ideally we'd mark this as `#[cfg(test)]`, but that doesn't work with the way
    // tests are structured in this repository.
    pub fn empty_test() -> Self {
        CatalogState {
            database_by_name: Default::default(),
            database_by_id: Default::default(),
            entry_by_id: Default::default(),
            entry_by_global_id: Default::default(),
            notices_by_dep_id: Default::default(),
            ambient_schemas_by_name: Default::default(),
            ambient_schemas_by_id: Default::default(),
            temporary_namespaces: Default::default(),
            clusters_by_id: Default::default(),
            clusters_by_name: Default::default(),
            network_policies_by_name: Default::default(),
            roles_by_name: Default::default(),
            roles_by_id: Default::default(),
            network_policies_by_id: Default::default(),
            role_auth_by_id: Default::default(),
            config: CatalogConfig {
                start_time: Default::default(),
                start_instant: Instant::now(),
                nonce: Default::default(),
                environment_id: EnvironmentId::for_tests(),
                session_id: Default::default(),
                build_info: &DUMMY_BUILD_INFO,
                now: NOW_ZERO.clone(),
                connection_context: ConnectionContext::for_tests(Arc::new(
                    InMemorySecretsController::new(),
                )),
                aws_account_id: None,
                helm_chart_version: None,
            },
            cluster_replica_sizes: ClusterReplicaSizeMap::for_tests(),
            availability_zones: Default::default(),
            system_configuration: Arc::new(SystemVars::default()),
            scoped_system_parameters: Default::default(),
            egress_addresses: Default::default(),
            aws_principal_context: Default::default(),
            aws_privatelink_availability_zones: Default::default(),
            http_host_name: Default::default(),
            default_privileges: Arc::new(DefaultPrivileges::default()),
            system_privileges: Arc::new(PrivilegeMap::default()),
            comments: Arc::new(CommentsMap::default()),
            source_references: Default::default(),
            storage_metadata: Arc::new(StorageMetadata::default()),
            license_key: ValidatedLicenseKey::for_tests(),
            mock_authentication_nonce: Default::default(),
        }
    }

    pub fn for_session<'a>(&'a self, session: &'a Session) -> ConnCatalog<'a> {
        let search_path = self.resolve_search_path(session);
        let database = self
            .database_by_name
            .get(session.vars().database())
            .map(|id| id.clone());
        let state = match session.transaction().catalog_state() {
            Some(txn_catalog_state) => Cow::Borrowed(txn_catalog_state),
            None => Cow::Borrowed(self),
        };
        ConnCatalog {
            state,
            unresolvable_ids: BTreeSet::new(),
            conn_id: session.conn_id().clone(),
            cluster: session.vars().cluster().into(),
            database,
            search_path,
            role_id: session.current_role_id().clone(),
            prepared_statements: Some(session.prepared_statements()),
            portals: Some(session.portals()),
            notices_tx: session.retain_notice_transmitter(),
            restrict_to_user_objects: session.vars().restrict_to_user_objects(),
        }
    }

    pub fn for_sessionless_user(&self, role_id: RoleId) -> ConnCatalog<'_> {
        let (notices_tx, _notices_rx) = mpsc::unbounded_channel();
        let cluster = self.system_configuration.default_cluster();

        ConnCatalog {
            state: Cow::Borrowed(self),
            unresolvable_ids: BTreeSet::new(),
            conn_id: SYSTEM_CONN_ID.clone(),
            cluster,
            database: self
                .resolve_database(DEFAULT_DATABASE_NAME)
                .ok()
                .map(|db| db.id()),
            // Leaving the system's search path empty allows us to catch issues
            // where catalog object names have not been normalized correctly.
            search_path: Vec::new(),
            role_id,
            prepared_statements: None,
            portals: None,
            notices_tx,
            restrict_to_user_objects: false,
        }
    }

    pub fn for_system_session(&self) -> ConnCatalog<'_> {
        self.for_sessionless_user(MZ_SYSTEM_ROLE_ID)
    }

    /// Returns an iterator over the deduplicated identifiers of all
    /// objects this catalog entry transitively depends on (where
    /// "depends on" is meant in the sense of [`CatalogItem::uses`], rather than
    /// [`CatalogItem::references`]).
    pub fn transitive_uses(&self, id: CatalogItemId) -> impl Iterator<Item = CatalogItemId> + '_ {
        struct I<'a> {
            queue: VecDeque<CatalogItemId>,
            seen: BTreeSet<CatalogItemId>,
            this: &'a CatalogState,
        }
        impl<'a> Iterator for I<'a> {
            type Item = CatalogItemId;
            fn next(&mut self) -> Option<Self::Item> {
                if let Some(next) = self.queue.pop_front() {
                    for child in self.this.get_entry(&next).item().uses() {
                        if !self.seen.contains(&child) {
                            self.queue.push_back(child);
                            self.seen.insert(child);
                        }
                    }
                    Some(next)
                } else {
                    None
                }
            }
        }

        I {
            queue: [id].into_iter().collect(),
            seen: [id].into_iter().collect(),
            this: self,
        }
    }

    /// Computes the IDs of any log sources this catalog entry transitively
    /// depends on.
    pub fn introspection_dependencies(&self, id: CatalogItemId) -> Vec<CatalogItemId> {
        let mut out = Vec::new();

        // Iterative worklist traversal rather than recursion. The dependency
        // chain is user controlled and can be arbitrarily deep.
        let mut queue: VecDeque<CatalogItemId> = [id].into_iter().collect();
        let mut seen: BTreeSet<CatalogItemId> = [id].into_iter().collect();
        while let Some(id) = queue.pop_front() {
            match self.get_entry(&id).item() {
                CatalogItem::Log(_) => out.push(id),
                item @ (CatalogItem::View(_)
                | CatalogItem::MaterializedView(_)
                | CatalogItem::Connection(_)) => {
                    // TODO Unclear if this table wants to include all uses or only references.
                    for item_id in item.references().items() {
                        if seen.insert(*item_id) {
                            queue.push_back(*item_id);
                        }
                    }
                }
                CatalogItem::Sink(sink) => {
                    let from_item_id = self.get_entry_by_global_id(&sink.from).id();
                    if seen.insert(from_item_id) {
                        queue.push_back(from_item_id);
                    }
                }
                CatalogItem::MetricSink(metric_sink) => {
                    let from_item_id = self.get_entry_by_global_id(&metric_sink.from).id();
                    if seen.insert(from_item_id) {
                        queue.push_back(from_item_id);
                    }
                }
                CatalogItem::Index(idx) => {
                    let on_item_id = self.get_entry_by_global_id(&idx.on).id();
                    if seen.insert(on_item_id) {
                        queue.push_back(on_item_id);
                    }
                }
                CatalogItem::Table(_)
                | CatalogItem::Source(_)
                | CatalogItem::Type(_)
                | CatalogItem::Func(_)
                | CatalogItem::Secret(_) => (),
            }
        }

        out
    }

    /// Returns all the IDs of all objects that depend on `ids`, including `ids` themselves.
    ///
    /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
    /// earlier in the list than the roots. This is particularly useful for the order to drop
    /// objects.
    pub(super) fn object_dependents(
        &self,
        object_ids: &Vec<ObjectId>,
        conn_id: &ConnectionId,
        seen: &mut BTreeSet<ObjectId>,
    ) -> Vec<ObjectId> {
        let mut dependents = Vec::new();
        for object_id in object_ids {
            match object_id {
                ObjectId::Cluster(id) => {
                    dependents.extend_from_slice(&self.cluster_dependents(*id, seen));
                }
                ObjectId::ClusterReplica((cluster_id, replica_id)) => dependents.extend_from_slice(
                    &self.cluster_replica_dependents(*cluster_id, *replica_id, seen),
                ),
                ObjectId::Database(id) => {
                    dependents.extend_from_slice(&self.database_dependents(*id, conn_id, seen))
                }
                ObjectId::Schema((database_spec, schema_spec)) => {
                    dependents.extend_from_slice(&self.schema_dependents(
                        database_spec.clone(),
                        schema_spec.clone(),
                        conn_id,
                        seen,
                    ));
                }
                ObjectId::NetworkPolicy(id) => {
                    dependents.extend_from_slice(&self.network_policy_dependents(*id, seen));
                }
                id @ ObjectId::Role(_) => {
                    let unseen = seen.insert(id.clone());
                    if unseen {
                        dependents.push(id.clone());
                    }
                }
                ObjectId::Item(id) => {
                    dependents.extend_from_slice(&self.item_dependents(*id, seen))
                }
            }
        }
        dependents
    }

    /// Returns all the IDs of all objects that depend on `cluster_id`, including `cluster_id`
    /// itself.
    ///
    /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
    /// earlier in the list than the roots. This is particularly useful for the order to drop
    /// objects.
    fn cluster_dependents(
        &self,
        cluster_id: ClusterId,
        seen: &mut BTreeSet<ObjectId>,
    ) -> Vec<ObjectId> {
        let mut dependents = Vec::new();
        let object_id = ObjectId::Cluster(cluster_id);
        if !seen.contains(&object_id) {
            seen.insert(object_id.clone());
            let cluster = self.get_cluster(cluster_id);
            for item_id in cluster.bound_objects() {
                dependents.extend_from_slice(&self.item_dependents(*item_id, seen));
            }
            for replica_id in cluster.replica_ids().values() {
                dependents.extend_from_slice(&self.cluster_replica_dependents(
                    cluster_id,
                    *replica_id,
                    seen,
                ));
            }
            dependents.push(object_id);
        }
        dependents
    }

    /// Returns all the IDs of all objects that depend on `replica_id`, including `replica_id`
    /// itself.
    ///
    /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
    /// earlier in the list than the roots. This is particularly useful for the order to drop
    /// objects.
    pub(super) fn cluster_replica_dependents(
        &self,
        cluster_id: ClusterId,
        replica_id: ReplicaId,
        seen: &mut BTreeSet<ObjectId>,
    ) -> Vec<ObjectId> {
        let mut dependents = Vec::new();
        let object_id = ObjectId::ClusterReplica((cluster_id, replica_id));
        if !seen.contains(&object_id) {
            seen.insert(object_id.clone());
            // Materialized views that target this replica are implicitly
            // dropped with it, so cascade to their dependents to avoid leaving
            // dangling references.
            let cluster = self.get_cluster(cluster_id);
            for item_id in cluster.bound_objects() {
                if let CatalogItem::MaterializedView(mv) = self.get_entry(item_id).item()
                    && mv.target_replica == Some(replica_id)
                {
                    dependents.extend_from_slice(&self.item_dependents(*item_id, seen));
                }
            }
            dependents.push(object_id);
        }
        dependents
    }

    /// Returns all the IDs of all objects that depend on `database_id`, including `database_id`
    /// itself.
    ///
    /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
    /// earlier in the list than the roots. This is particularly useful for the order to drop
    /// objects.
    fn database_dependents(
        &self,
        database_id: DatabaseId,
        conn_id: &ConnectionId,
        seen: &mut BTreeSet<ObjectId>,
    ) -> Vec<ObjectId> {
        let mut dependents = Vec::new();
        let object_id = ObjectId::Database(database_id);
        if !seen.contains(&object_id) {
            seen.insert(object_id.clone());
            let database = self.get_database(&database_id);
            for schema_id in database.schema_ids().values() {
                dependents.extend_from_slice(&self.schema_dependents(
                    ResolvedDatabaseSpecifier::Id(database_id),
                    SchemaSpecifier::Id(*schema_id),
                    conn_id,
                    seen,
                ));
            }
            dependents.push(object_id);
        }
        dependents
    }

    /// Returns all the IDs of all objects that depend on `schema_id`, including `schema_id`
    /// itself.
    ///
    /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
    /// earlier in the list than the roots. This is particularly useful for the order to drop
    /// objects.
    fn schema_dependents(
        &self,
        database_spec: ResolvedDatabaseSpecifier,
        schema_spec: SchemaSpecifier,
        conn_id: &ConnectionId,
        seen: &mut BTreeSet<ObjectId>,
    ) -> Vec<ObjectId> {
        let mut dependents = Vec::new();
        let object_id = ObjectId::Schema((database_spec, schema_spec.clone()));
        if !seen.contains(&object_id) {
            seen.insert(object_id.clone());
            let schema = self.get_schema(&database_spec, &schema_spec, conn_id);
            for item_id in schema.item_ids() {
                dependents.extend_from_slice(&self.item_dependents(item_id, seen));
            }
            dependents.push(object_id)
        }
        dependents
    }

    /// Returns all the IDs of all objects that depend on `item_id`, including `item_id`
    /// itself.
    ///
    /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
    /// earlier in the list than the roots. This is particularly useful for the order to drop
    /// objects.
    pub(super) fn item_dependents(
        &self,
        item_id: CatalogItemId,
        seen: &mut BTreeSet<ObjectId>,
    ) -> Vec<ObjectId> {
        let mut dependents = Vec::new();

        // Iterative post-order traversal rather than recursion. Dependency
        // chains are user controlled and can be arbitrarily deep (e.g. a long
        // chain of stacked views), so recursing risks a stack overflow.
        enum Work {
            Enter(CatalogItemId),
            Emit(ObjectId),
        }
        let mut stack = vec![Work::Enter(item_id)];
        while let Some(work) = stack.pop() {
            match work {
                Work::Enter(item_id) => {
                    let object_id = ObjectId::Item(item_id);
                    if !seen.insert(object_id.clone()) {
                        continue;
                    }
                    let entry = self.get_entry(&item_id);
                    // Pushed in reverse of the desired output order: dependents
                    // first, then self, then the progress collection. We treat
                    // the progress collection as if it depends on the source
                    // for dropping. We have additional code in planning to
                    // create a kind of special-case "CASCADE" for this
                    // dependency.
                    if let Some(progress_id) = entry.progress_id() {
                        stack.push(Work::Enter(progress_id));
                    }
                    stack.push(Work::Emit(object_id));
                    for dependent_id in entry.used_by().iter().rev() {
                        stack.push(Work::Enter(*dependent_id));
                    }
                }
                Work::Emit(object_id) => dependents.push(object_id),
            }
        }

        dependents
    }

    /// Returns all the IDs of all objects that depend on `network_policy_id`, including `network_policy_id`
    /// itself.
    ///
    /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
    /// earlier in the list than the roots. This is particularly useful for the order to drop
    /// objects.
    pub(super) fn network_policy_dependents(
        &self,
        network_policy_id: NetworkPolicyId,
        _seen: &mut BTreeSet<ObjectId>,
    ) -> Vec<ObjectId> {
        let object_id = ObjectId::NetworkPolicy(network_policy_id);
        // Currently network policies have no dependents
        // when we add the ability for users or sources/sinks to have policies
        // this method will need to be updated.
        vec![object_id]
    }

    /// Indicates whether the indicated item is considered stable or not.
    ///
    /// Only stable items can be used as dependencies of other catalog items.
    fn is_stable(&self, id: CatalogItemId) -> bool {
        let spec = self.get_entry(&id).name().qualifiers.schema_spec;
        !self.is_unstable_schema_specifier(spec)
    }

    pub(super) fn check_unstable_dependencies(&self, item: &CatalogItem) -> Result<(), Error> {
        if self.system_config().unsafe_enable_unstable_dependencies() {
            return Ok(());
        }

        let unstable_dependencies: Vec<_> = item
            .references()
            .items()
            .filter(|id| !self.is_stable(**id))
            .map(|id| self.get_entry(id).name().item.clone())
            .collect();

        // It's okay to create a temporary object with unstable
        // dependencies, since we will never need to reboot a catalog
        // that contains it.
        if unstable_dependencies.is_empty() || item.is_temporary() {
            Ok(())
        } else {
            let object_type = item.typ().to_string();
            Err(Error {
                kind: ErrorKind::UnstableDependency {
                    object_type,
                    unstable_dependencies,
                },
            })
        }
    }

    pub fn resolve_full_name(
        &self,
        name: &QualifiedItemName,
        conn_id: Option<&ConnectionId>,
    ) -> FullItemName {
        let conn_id = conn_id.unwrap_or(&SYSTEM_CONN_ID);

        let database = match &name.qualifiers.database_spec {
            ResolvedDatabaseSpecifier::Ambient => RawDatabaseSpecifier::Ambient,
            ResolvedDatabaseSpecifier::Id(id) => {
                RawDatabaseSpecifier::Name(self.get_database(id).name().to_string())
            }
        };
        // For temporary schemas, we know the name is always MZ_TEMP_SCHEMA,
        // and the schema may not exist yet if no temporary items have been created.
        let schema = match &name.qualifiers.schema_spec {
            SchemaSpecifier::Temporary => MZ_TEMP_SCHEMA.to_string(),
            SchemaSpecifier::Id(_) => self
                .get_schema(
                    &name.qualifiers.database_spec,
                    &name.qualifiers.schema_spec,
                    conn_id,
                )
                .name()
                .schema
                .clone(),
        };
        FullItemName {
            database,
            schema,
            item: name.item.clone(),
        }
    }

    pub(super) fn resolve_full_schema_name(&self, name: &QualifiedSchemaName) -> FullSchemaName {
        let database = match &name.database {
            ResolvedDatabaseSpecifier::Ambient => RawDatabaseSpecifier::Ambient,
            ResolvedDatabaseSpecifier::Id(id) => {
                RawDatabaseSpecifier::Name(self.get_database(id).name().to_string())
            }
        };
        FullSchemaName {
            database,
            schema: name.schema.clone(),
        }
    }

    pub fn get_entry(&self, id: &CatalogItemId) -> &CatalogEntry {
        self.entry_by_id
            .get(id)
            .unwrap_or_else(|| panic!("catalog out of sync, missing id {id:?}"))
    }

    pub fn get_entry_by_global_id(&self, id: &GlobalId) -> CatalogCollectionEntry {
        let item_id = self
            .entry_by_global_id
            .get(id)
            .unwrap_or_else(|| panic!("catalog out of sync, missing id {id:?}"));

        let entry = self.get_entry(item_id).clone();
        let version = match entry.item() {
            CatalogItem::Table(table) => {
                let (version, _) = table
                    .collections
                    .iter()
                    .find(|(_verison, gid)| *gid == id)
                    .expect("version to exist");
                RelationVersionSelector::Specific(*version)
            }
            _ => RelationVersionSelector::Latest,
        };
        CatalogCollectionEntry { entry, version }
    }

    pub fn get_entries(&self) -> impl Iterator<Item = (&CatalogItemId, &CatalogEntry)> + '_ {
        self.entry_by_id.iter()
    }

    pub fn get_temp_items(&self, conn: &ConnectionId) -> impl Iterator<Item = ObjectId> + '_ {
        // A temporary namespace is registered at the connection's first
        // temporary-item creation, so it's valid for one to not exist yet.
        self.temporary_namespaces
            .schema(conn)
            .into_iter()
            .flat_map(|schema| schema.items.values().copied().map(ObjectId::from))
    }

    /// Returns true if a temporary namespace is registered for the given
    /// connection, i.e. it has (or has had) temporary items.
    ///
    /// The namespace is registered at the connection's first temporary-item
    /// creation, so this returns false for connections that never created
    /// any temporary objects.
    pub fn has_temporary_namespace(&self, conn: &ConnectionId) -> bool {
        self.temporary_namespaces.contains_conn(conn)
    }

    /// Converts an in-memory catalog entry into its durable representation.
    ///
    /// The durable owner of a temporary entry is the session whose connection
    /// currently holds it, resolved from the temporary namespace
    /// registered at the session's first temporary-item creation.
    pub(super) fn durable_item(
        &self,
        entry: CatalogEntry,
    ) -> Result<mz_catalog::durable::Item, AdapterError> {
        let ephemeral_owner_session = entry
            .conn_id()
            .map(|conn_id| {
                self.temporary_namespaces
                    .uuid_for_conn(conn_id)
                    .ok_or_else(|| {
                        AdapterError::Internal(format!(
                            "no session record for connection {conn_id} owning temporary item"
                        ))
                    })
            })
            .transpose()?;
        let (create_sql, global_id, extra_versions) = entry.item.into_serialized();
        Ok(mz_catalog::durable::Item {
            id: entry.id,
            oid: entry.oid,
            global_id,
            schema_id: entry.name.qualifiers.schema_spec.into(),
            name: entry.name.item,
            create_sql,
            owner_id: entry.owner_id,
            privileges: entry.privileges.into_all_values().collect(),
            extra_versions,
            ephemeral_owner_session,
        })
    }

    /// Gets a type named `name` from exactly one of the system schemas.
    ///
    /// # Panics
    /// - If `name` is not an entry in any system schema
    /// - If more than one system schema has an entry named `name`.
    pub(super) fn get_system_type(&self, name: &str) -> &CatalogEntry {
        let mut res = None;
        for schema_id in self.system_schema_ids() {
            let schema = &self.ambient_schemas_by_id[&schema_id];
            if let Some(global_id) = schema.types.get(name) {
                match res {
                    None => res = Some(self.get_entry(global_id)),
                    Some(_) => panic!(
                        "only call get_system_type on objects uniquely identifiable in one system schema"
                    ),
                }
            }
        }

        res.unwrap_or_else(|| panic!("cannot find type {} in system schema", name))
    }

    pub fn get_item_by_name(
        &self,
        name: &QualifiedItemName,
        conn_id: &ConnectionId,
    ) -> Option<&CatalogEntry> {
        self.get_schema(
            &name.qualifiers.database_spec,
            &name.qualifiers.schema_spec,
            conn_id,
        )
        .items
        .get(&name.item)
        .and_then(|id| self.try_get_entry(id))
    }

    pub fn get_type_by_name(
        &self,
        name: &QualifiedItemName,
        conn_id: &ConnectionId,
    ) -> Option<&CatalogEntry> {
        self.get_schema(
            &name.qualifiers.database_spec,
            &name.qualifiers.schema_spec,
            conn_id,
        )
        .types
        .get(&name.item)
        .and_then(|id| self.try_get_entry(id))
    }

    pub(super) fn find_available_name(
        &self,
        mut name: QualifiedItemName,
        conn_id: &ConnectionId,
    ) -> QualifiedItemName {
        let mut i = 0;
        let orig_item_name = name.item.clone();
        while self.get_item_by_name(&name, conn_id).is_some() {
            i += 1;
            name.item = format!("{}{}", orig_item_name, i);
        }
        name
    }

    pub fn try_get_entry(&self, id: &CatalogItemId) -> Option<&CatalogEntry> {
        self.entry_by_id.get(id)
    }

    pub fn try_get_entry_by_global_id(&self, id: &GlobalId) -> Option<&CatalogEntry> {
        let item_id = self.entry_by_global_id.get(id)?;
        self.try_get_entry(item_id)
    }

    /// Returns the [`RelationDesc`] for a [`GlobalId`], if the provided [`GlobalId`] refers to an
    /// object that returns rows.
    pub fn try_get_desc_by_global_id(&self, id: &GlobalId) -> Option<Cow<'_, RelationDesc>> {
        let entry = self.try_get_entry_by_global_id(id)?;
        let desc = match entry.item() {
            CatalogItem::Table(table) => Cow::Owned(table.desc_for(id)),
            // TODO(alter_table): Support schema evolution on sources.
            other => other.relation_desc(RelationVersionSelector::Latest)?,
        };
        Some(desc)
    }

    pub(crate) fn get_cluster(&self, cluster_id: ClusterId) -> &Cluster {
        self.try_get_cluster(cluster_id)
            .unwrap_or_else(|| panic!("unknown cluster {cluster_id}"))
    }

    pub(super) fn try_get_cluster(&self, cluster_id: ClusterId) -> Option<&Cluster> {
        self.clusters_by_id.get(&cluster_id)
    }

    pub(super) fn try_get_role(&self, id: &RoleId) -> Option<&Role> {
        self.roles_by_id.get(id)
    }

    pub fn get_role(&self, id: &RoleId) -> &Role {
        self.roles_by_id.get(id).expect("catalog out of sync")
    }

    pub fn get_roles(&self) -> impl Iterator<Item = &RoleId> {
        self.roles_by_id.keys()
    }

    pub(super) fn try_get_role_by_name(&self, role_name: &str) -> Option<&Role> {
        self.roles_by_name
            .get(role_name)
            .map(|id| &self.roles_by_id[id])
    }

    pub(super) fn get_role_auth(&self, id: &RoleId) -> &RoleAuth {
        self.role_auth_by_id
            .get(id)
            .unwrap_or_else(|| panic!("catalog out of sync, missing role auth for {id}"))
    }

    pub(super) fn try_get_role_auth_by_id(&self, id: &RoleId) -> Option<&RoleAuth> {
        self.role_auth_by_id.get(id)
    }

    pub(super) fn try_get_network_policy_by_name(
        &self,
        policy_name: &str,
    ) -> Option<&NetworkPolicy> {
        self.network_policies_by_name
            .get(policy_name)
            .map(|id| &self.network_policies_by_id[id])
    }

    pub(crate) fn collect_role_membership(&self, id: &RoleId) -> BTreeSet<RoleId> {
        let mut membership = BTreeSet::new();
        let mut queue = VecDeque::from(vec![id]);
        while let Some(cur_id) = queue.pop_front() {
            if !membership.contains(cur_id) {
                membership.insert(cur_id.clone());
                let role = self.get_role(cur_id);
                soft_assert_no_log!(
                    !role.membership().keys().contains(id),
                    "circular membership exists in the catalog"
                );
                queue.extend(role.membership().keys());
            }
        }
        membership.insert(RoleId::Public);
        membership
    }

    pub fn get_network_policy(&self, id: &NetworkPolicyId) -> &NetworkPolicy {
        self.network_policies_by_id
            .get(id)
            .expect("catalog out of sync")
    }

    pub fn get_network_policies(&self) -> impl Iterator<Item = &NetworkPolicyId> {
        self.network_policies_by_id.keys()
    }

    /// Returns the URL for POST-ing data to a webhook source, if `id` corresponds to a webhook
    /// source.
    ///
    /// Note: Identifiers for the source, e.g. item name, are URL encoded.
    pub fn try_get_webhook_url(&self, id: &CatalogItemId) -> Option<url::Url> {
        let entry = self.try_get_entry(id)?;
        // Note: Webhook sources can never be created in the temporary schema, hence passing None.
        let name = self.resolve_full_name(entry.name(), None);
        let host_name = self
            .http_host_name
            .as_ref()
            .map(|x| x.as_str())
            .unwrap_or_else(|| "HOST");

        let RawDatabaseSpecifier::Name(database) = name.database else {
            return None;
        };

        let mut url = url::Url::parse(&format!("https://{host_name}/api/webhook")).ok()?;
        url.path_segments_mut()
            .ok()?
            .push(&database)
            .push(&name.schema)
            .push(&name.item);

        Some(url)
    }

    /// Parses the given SQL string into a pair of [`Plan`] and a [`ResolvedIds`].
    ///
    /// This function will temporarily enable all "enable_for_item_parsing" feature flags. See
    /// [`CatalogState::with_enable_for_item_parsing`] for more details.
    ///
    /// NOTE: While this method takes a `&mut self`, all mutations are temporary and restored to
    /// their original state before the method returns.
    pub(crate) fn deserialize_plan_with_enable_for_item_parsing(
        // DO NOT add any additional mutations to this method. It would be fairly surprising to the
        // caller if this method changed the state of the catalog.
        &mut self,
        create_sql: &str,
        force_if_exists_skip: bool,
    ) -> Result<(Plan, ResolvedIds), AdapterError> {
        self.with_enable_for_item_parsing(|state| {
            let pcx = PlanContext::zero().with_ignore_if_exists_errors(force_if_exists_skip);
            let pcx = Some(&pcx);
            let session_catalog = state.for_system_session();

            let stmt = mz_sql::parse::parse(create_sql)?.into_element().ast;
            let (stmt, resolved_ids) = mz_sql::names::resolve(&session_catalog, stmt)?;
            let (plan, _sql_impl_ids) =
                mz_sql::plan::plan(pcx, &session_catalog, stmt, &Params::empty(), &resolved_ids)?;

            Ok((plan, resolved_ids))
        })
    }

    /// Parses the given SQL string into a pair of [`Plan`] and a [`ResolvedIds`].
    #[mz_ore::instrument]
    pub(crate) fn parse_plan(
        create_sql: &str,
        pcx: Option<&PlanContext>,
        catalog: &ConnCatalog,
    ) -> Result<(Plan, ResolvedIds), AdapterError> {
        let stmt = mz_sql::parse::parse(create_sql)?.into_element().ast;
        let (stmt, resolved_ids) = mz_sql::names::resolve(catalog, stmt)?;
        let (plan, _sql_impl_ids) =
            mz_sql::plan::plan(pcx, catalog, stmt, &Params::empty(), &resolved_ids)?;

        Ok((plan, resolved_ids))
    }

    /// Parses the given SQL string into a pair of [`CatalogItem`].
    pub(crate) fn deserialize_item(
        &self,
        global_id: GlobalId,
        create_sql: &str,
        extra_versions: &BTreeMap<RelationVersion, GlobalId>,
        local_expression_cache: &mut LocalExpressionCache,
        previous_item: Option<CatalogItem>,
    ) -> Result<CatalogItem, AdapterError> {
        self.parse_item(
            global_id,
            create_sql,
            extra_versions,
            None,
            false,
            None,
            local_expression_cache,
            previous_item,
        )
    }

    /// Parses the given SQL string into a `CatalogItem`.
    #[mz_ore::instrument]
    pub(crate) fn parse_item(
        &self,
        global_id: GlobalId,
        create_sql: &str,
        extra_versions: &BTreeMap<RelationVersion, GlobalId>,
        pcx: Option<&PlanContext>,
        is_retained_metrics_object: bool,
        custom_logical_compaction_window: Option<CompactionWindow>,
        local_expression_cache: &mut LocalExpressionCache,
        previous_item: Option<CatalogItem>,
    ) -> Result<CatalogItem, AdapterError> {
        let cached_expr = local_expression_cache.remove_cached_expression(&global_id);
        match self.parse_item_inner(
            global_id,
            create_sql,
            extra_versions,
            pcx,
            is_retained_metrics_object,
            custom_logical_compaction_window,
            cached_expr,
            previous_item,
        ) {
            Ok((item, uncached_expr)) => {
                if let Some((uncached_expr, optimizer_features)) = uncached_expr {
                    local_expression_cache.insert_uncached_expression(
                        global_id,
                        uncached_expr,
                        optimizer_features,
                    );
                }
                Ok(item)
            }
            Err((err, cached_expr)) => {
                if let Some(local_expr) = cached_expr {
                    local_expression_cache.insert_cached_expression(global_id, local_expr);
                }
                Err(err)
            }
        }
    }

    /// Parses the given SQL string into a `CatalogItem`, using `cached_expr` if it's Some.
    ///
    /// On success returns the `CatalogItem` and an optimized expression iff the expression was
    /// not cached.
    ///
    /// On failure returns an error and `cached_expr` so it can be used later.
    #[mz_ore::instrument]
    pub(crate) fn parse_item_inner(
        &self,
        global_id: GlobalId,
        create_sql: &str,
        extra_versions: &BTreeMap<RelationVersion, GlobalId>,
        pcx: Option<&PlanContext>,
        is_retained_metrics_object: bool,
        custom_logical_compaction_window: Option<CompactionWindow>,
        cached_expr: Option<LocalExpressions>,
        previous_item: Option<CatalogItem>,
    ) -> Result<
        (
            CatalogItem,
            Option<(OptimizedMirRelationExpr, OptimizerFeatures)>,
        ),
        (AdapterError, Option<LocalExpressions>),
    > {
        let session_catalog = self.for_system_session();

        let (plan, resolved_ids) = match Self::parse_plan(create_sql, pcx, &session_catalog) {
            Ok((plan, resolved_ids)) => (plan, resolved_ids),
            Err(err) => return Err((err, cached_expr)),
        };

        let mut uncached_expr = None;

        // Carry over the plans (`optimized_plan`, `physical_plan`,
        // `dataflow_metainfo`) from the previous incarnation of this item when
        // re-parsing an existing item (e.g. after a RENAME). These fields live
        // on the `CatalogItem` since #35834, but are not reconstructable from
        // `create_sql` alone — they are populated by the sequencer `_finish`
        // paths at create time, and by the expression-cache / bootstrap
        // rendering path on boot. If we don't preserve them here, a RENAME
        // silently drops the plans and dataflow metainfo for the affected
        // MV/Index/CT.
        let previous_plans = previous_item.as_ref().map(|item| {
            (
                item.optimized_plan().cloned(),
                item.physical_plan().cloned(),
                item.dataflow_metainfo().cloned(),
            )
        });

        let mut item = match plan {
            Plan::CreateTable(CreateTablePlan { table, .. }) => {
                let collections = extra_versions
                    .iter()
                    .map(|(version, gid)| (*version, *gid))
                    .chain([(RelationVersion::root(), global_id)].into_iter())
                    .collect();

                CatalogItem::Table(Table {
                    create_sql: Some(table.create_sql),
                    desc: table.desc,
                    collections,
                    conn_id: None,
                    resolved_ids,
                    custom_logical_compaction_window: custom_logical_compaction_window
                        .or(table.compaction_window),
                    is_retained_metrics_object,
                    data_source: match table.data_source {
                        mz_sql::plan::TableDataSource::TableWrites { defaults } => {
                            TableDataSource::TableWrites { defaults }
                        }
                        mz_sql::plan::TableDataSource::DataSource {
                            desc: data_source_desc,
                            timeline,
                        } => match data_source_desc {
                            mz_sql::plan::DataSourceDesc::IngestionExport {
                                ingestion_id,
                                external_reference,
                                details,
                                data_config,
                            } => TableDataSource::DataSource {
                                desc: DataSourceDesc::IngestionExport {
                                    ingestion_id,
                                    external_reference,
                                    details,
                                    data_config,
                                },
                                timeline,
                            },
                            mz_sql::plan::DataSourceDesc::Webhook {
                                validate_using,
                                body_format,
                                headers,
                                cluster_id,
                            } => TableDataSource::DataSource {
                                desc: DataSourceDesc::Webhook {
                                    validate_using,
                                    body_format,
                                    headers,
                                    cluster_id: cluster_id
                                        .expect("Webhook Tables must have a cluster_id set"),
                                },
                                timeline,
                            },
                            _ => {
                                return Err((
                                    AdapterError::Unstructured(anyhow::anyhow!(
                                        "unsupported data source for table"
                                    )),
                                    cached_expr,
                                ));
                            }
                        },
                    },
                })
            }
            Plan::CreateSource(CreateSourcePlan {
                source,
                timeline,
                in_cluster,
                ..
            }) => CatalogItem::Source(Source {
                create_sql: Some(source.create_sql),
                data_source: match source.data_source {
                    mz_sql::plan::DataSourceDesc::Ingestion(desc) => DataSourceDesc::Ingestion {
                        desc,
                        cluster_id: match in_cluster {
                            Some(id) => id,
                            None => {
                                return Err((
                                    AdapterError::Unstructured(anyhow::anyhow!(
                                        "ingestion-based sources must have cluster specified"
                                    )),
                                    cached_expr,
                                ));
                            }
                        },
                    },
                    mz_sql::plan::DataSourceDesc::OldSyntaxIngestion {
                        desc,
                        progress_subsource,
                        data_config,
                        details,
                    } => DataSourceDesc::OldSyntaxIngestion {
                        desc,
                        progress_subsource,
                        data_config,
                        details,
                        cluster_id: match in_cluster {
                            Some(id) => id,
                            None => {
                                return Err((
                                    AdapterError::Unstructured(anyhow::anyhow!(
                                        "ingestion-based sources must have cluster specified"
                                    )),
                                    cached_expr,
                                ));
                            }
                        },
                    },
                    mz_sql::plan::DataSourceDesc::IngestionExport {
                        ingestion_id,
                        external_reference,
                        details,
                        data_config,
                    } => DataSourceDesc::IngestionExport {
                        ingestion_id,
                        external_reference,
                        details,
                        data_config,
                    },
                    mz_sql::plan::DataSourceDesc::Progress => DataSourceDesc::Progress,
                    mz_sql::plan::DataSourceDesc::Webhook {
                        validate_using,
                        body_format,
                        headers,
                        cluster_id,
                    } => {
                        mz_ore::soft_assert_or_log!(
                            cluster_id.is_none(),
                            "cluster_id set at Source level for Webhooks"
                        );
                        DataSourceDesc::Webhook {
                            validate_using,
                            body_format,
                            headers,
                            cluster_id: in_cluster
                                .expect("webhook sources must use an existing cluster"),
                        }
                    }
                },
                desc: source.desc,
                global_id,
                timeline,
                resolved_ids,
                custom_logical_compaction_window: source
                    .compaction_window
                    .or(custom_logical_compaction_window),
                is_retained_metrics_object,
            }),
            Plan::CreateView(CreateViewPlan { view, .. }) => {
                // Collect optimizer parameters.
                let optimizer_config =
                    optimize::OptimizerConfig::from(session_catalog.system_vars());
                let previous_exprs = previous_item.map(|item| match item {
                    CatalogItem::View(view) => Some((view.raw_expr, view.locally_optimized_expr)),
                    _ => None,
                });

                let (raw_expr, optimized_expr) = match (cached_expr, previous_exprs) {
                    (Some(local_expr), _)
                        if local_expr.optimizer_features == optimizer_config.features =>
                    {
                        debug!("local expression cache hit for {global_id:?}");
                        (Arc::new(view.expr), Arc::new(local_expr.local_mir))
                    }
                    // If the new expr is equivalent to the old expr, then we don't need to re-optimize.
                    (_, Some(Some((raw_expr, optimized_expr)))) if *raw_expr == view.expr => {
                        (Arc::clone(&raw_expr), Arc::clone(&optimized_expr))
                    }
                    (cached_expr, _) => {
                        let optimizer_features = optimizer_config.features.clone();
                        // Build an optimizer for this VIEW.
                        let mut optimizer = optimize::view::Optimizer::new(optimizer_config, None);

                        // HIR ⇒ MIR lowering and MIR ⇒ MIR optimization (local)
                        let raw_expr = view.expr;
                        let optimized_expr = match optimizer.optimize(raw_expr.clone()) {
                            Ok(optimzed_expr) => optimzed_expr,
                            Err(err) => return Err((err.into(), cached_expr)),
                        };

                        uncached_expr = Some((optimized_expr.clone(), optimizer_features));

                        (Arc::new(raw_expr), Arc::new(optimized_expr))
                    }
                };

                // Resolve all item dependencies from the HIR expression.
                let dependencies: BTreeSet<_> = raw_expr
                    .depends_on()
                    .into_iter()
                    .map(|gid| self.get_entry_by_global_id(&gid).id())
                    .collect();

                let typ = infer_sql_type_for_catalog(&raw_expr, &optimized_expr);
                CatalogItem::View(View {
                    create_sql: view.create_sql,
                    global_id,
                    raw_expr,
                    desc: RelationDesc::new(typ, view.column_names),
                    locally_optimized_expr: optimized_expr,
                    conn_id: None,
                    resolved_ids,
                    dependencies: DependencyIds(dependencies),
                })
            }
            Plan::CreateMaterializedView(CreateMaterializedViewPlan {
                materialized_view, ..
            }) => {
                let collections = extra_versions
                    .iter()
                    .map(|(version, gid)| (*version, *gid))
                    .chain([(RelationVersion::root(), global_id)].into_iter())
                    .collect();

                // Collect optimizer parameters.
                let system_vars = session_catalog.system_vars();
                let overrides = self
                    .get_cluster(materialized_view.cluster_id)
                    .config
                    .features();
                let optimizer_config =
                    optimize::OptimizerConfig::from(system_vars).override_from(&overrides);
                let previous_exprs = previous_item.map(|item| match item {
                    CatalogItem::MaterializedView(materialized_view) => (
                        materialized_view.raw_expr,
                        materialized_view.locally_optimized_expr,
                    ),
                    item => unreachable!("expected materialized view, found: {item:#?}"),
                });

                let (raw_expr, optimized_expr) = match (cached_expr, previous_exprs) {
                    (Some(local_expr), _)
                        if local_expr.optimizer_features == optimizer_config.features =>
                    {
                        debug!("local expression cache hit for {global_id:?}");
                        (
                            Arc::new(materialized_view.expr),
                            Arc::new(local_expr.local_mir),
                        )
                    }
                    // If the new expr is equivalent to the old expr, then we don't need to re-optimize.
                    (_, Some((raw_expr, optimized_expr)))
                        if *raw_expr == materialized_view.expr =>
                    {
                        (Arc::clone(&raw_expr), Arc::clone(&optimized_expr))
                    }
                    (cached_expr, _) => {
                        let optimizer_features = optimizer_config.features.clone();
                        // TODO(aalexandrov): ideally this should be a materialized_view::Optimizer.
                        let mut optimizer = optimize::view::Optimizer::new(optimizer_config, None);

                        let raw_expr = materialized_view.expr;
                        let optimized_expr = match optimizer.optimize(raw_expr.clone()) {
                            Ok(optimized_expr) => optimized_expr,
                            Err(err) => return Err((err.into(), cached_expr)),
                        };

                        uncached_expr = Some((optimized_expr.clone(), optimizer_features));

                        (Arc::new(raw_expr), Arc::new(optimized_expr))
                    }
                };
                let mut typ = infer_sql_type_for_catalog(&raw_expr, &optimized_expr);

                for &i in &materialized_view.non_null_assertions {
                    typ.column_types[i].nullable = false;
                }
                let desc = RelationDesc::new(typ, materialized_view.column_names);
                let desc = VersionedRelationDesc::new(desc);

                let initial_as_of = materialized_view.as_of.map(Antichain::from_elem);

                // Resolve all item dependencies from the HIR expression.
                let dependencies = raw_expr
                    .depends_on()
                    .into_iter()
                    .map(|gid| self.get_entry_by_global_id(&gid).id())
                    .collect();

                CatalogItem::MaterializedView(MaterializedView {
                    create_sql: materialized_view.create_sql,
                    collections,
                    raw_expr,
                    locally_optimized_expr: optimized_expr,
                    desc,
                    resolved_ids,
                    dependencies,
                    replacement_target: materialized_view.replacement_target,
                    cluster_id: materialized_view.cluster_id,
                    target_replica: materialized_view.target_replica,
                    non_null_assertions: materialized_view.non_null_assertions,
                    custom_logical_compaction_window: materialized_view.compaction_window,
                    refresh_schedule: materialized_view.refresh_schedule,
                    initial_as_of,
                    optimized_plan: None,
                    physical_plan: None,
                    dataflow_metainfo: None,
                })
            }
            Plan::CreateIndex(CreateIndexPlan { index, .. }) => CatalogItem::Index(Index {
                create_sql: index.create_sql,
                global_id,
                on: index.on,
                keys: index.keys.into(),
                conn_id: None,
                resolved_ids,
                cluster_id: index.cluster_id,
                custom_logical_compaction_window: custom_logical_compaction_window
                    .or(index.compaction_window),
                is_retained_metrics_object,
                optimized_plan: None,
                physical_plan: None,
                dataflow_metainfo: None,
            }),
            Plan::CreateMetricSink(CreateMetricSinkPlan { metric_sink, .. }) => {
                CatalogItem::MetricSink(MetricSink {
                    create_sql: metric_sink.create_sql,
                    global_id,
                    from: metric_sink.from,
                    resolved_ids,
                    cluster_id: metric_sink.cluster_id,
                    prefix: metric_sink.prefix,
                    optimized_plan: None,
                    physical_plan: None,
                    dataflow_metainfo: None,
                })
            }
            Plan::CreateSink(CreateSinkPlan {
                sink,
                with_snapshot,
                in_cluster,
                ..
            }) => CatalogItem::Sink(Sink {
                create_sql: sink.create_sql,
                global_id,
                from: sink.from,
                connection: sink.connection,
                envelope: sink.envelope,
                version: sink.version,
                with_snapshot,
                resolved_ids,
                cluster_id: in_cluster,
                commit_interval: sink.commit_interval,
            }),
            Plan::CreateType(CreateTypePlan { typ, .. }) => {
                // Even if we don't need the `RelationDesc` here, error out
                // early and eagerly, as a kind of soft assertion that we _can_
                // build the `RelationDesc` when needed.
                if let Err(err) = typ.inner.desc(&session_catalog) {
                    return Err((err.into(), cached_expr));
                }
                CatalogItem::Type(Type {
                    create_sql: Some(typ.create_sql),
                    global_id,
                    details: CatalogTypeDetails {
                        array_id: None,
                        typ: typ.inner,
                        pg_metadata: None,
                    },
                    resolved_ids,
                })
            }
            Plan::CreateSecret(CreateSecretPlan { secret, .. }) => CatalogItem::Secret(Secret {
                create_sql: secret.create_sql,
                global_id,
            }),
            Plan::CreateConnection(CreateConnectionPlan {
                connection:
                    mz_sql::plan::Connection {
                        create_sql,
                        details,
                    },
                ..
            }) => CatalogItem::Connection(Connection {
                create_sql,
                global_id,
                details,
                resolved_ids,
            }),
            _ => {
                return Err((
                    Error::new(ErrorKind::Corruption {
                        detail: "catalog entry generated inappropriate plan".to_string(),
                    })
                    .into(),
                    cached_expr,
                ));
            }
        };

        // Carry over the plans (`optimized_plan`, `physical_plan`,
        // `dataflow_metainfo`) from the previous incarnation of this item, if
        // any. See the comment on `previous_plans` above.
        if let Some((prev_optimized, prev_physical, prev_metainfo)) = previous_plans {
            if let Some((optimized_plan, physical_plan, dataflow_metainfo)) = item.plan_fields_mut()
            {
                *optimized_plan = prev_optimized;
                *physical_plan = prev_physical;
                *dataflow_metainfo = prev_metainfo;
            }
        }

        Ok((item, uncached_expr))
    }

    /// Execute function `f` on `self`, with all "enable_for_item_parsing" feature flags enabled.
    /// Calling this method will not permanently modify any system configuration variables.
    ///
    /// WARNING:
    /// Any modifications made to the system configuration variables in `f`, will be lost.
    pub fn with_enable_for_item_parsing<T>(&mut self, f: impl FnOnce(&mut Self) -> T) -> T {
        // Enable catalog features that might be required during planning existing
        // catalog items. Existing catalog items might have been created while
        // a specific feature flag was turned on, so we need to ensure that this
        // is also the case during catalog rehydration in order to avoid panics.
        //
        // WARNING / CONTRACT:
        // 1. Features used in this method that related to parsing / planning
        //    should be `enable_for_item_parsing` set to `true`.
        // 2. After this step, feature flag configuration must not be
        //    overridden.
        // 3. We don't notify `SystemVars` callbacks here, neither for the
        //    flags this enables nor for the `Arc` restore that undoes them
        //    afterwards. A callback on a `feature_flags!` var therefore won't
        //    observe this transient flip, only committed changes to it. See
        //    `SystemVars::register_callback`.
        let restore = Arc::clone(&self.system_configuration);
        Arc::make_mut(&mut self.system_configuration).enable_for_item_parsing();
        let res = f(self);
        self.system_configuration = restore;
        res
    }

    /// Returns all indexes on the given object and cluster known in the catalog.
    pub fn get_indexes_on(
        &self,
        id: GlobalId,
        cluster: ClusterId,
    ) -> impl Iterator<Item = (GlobalId, &Index)> {
        let index_matches = move |idx: &Index| idx.on == id && idx.cluster_id == cluster;

        self.try_get_entry_by_global_id(&id)
            .into_iter()
            .map(move |e| {
                e.used_by()
                    .iter()
                    .filter_map(move |uses_id| match self.get_entry(uses_id).item() {
                        CatalogItem::Index(index) if index_matches(index) => {
                            Some((index.global_id(), index))
                        }
                        _ => None,
                    })
            })
            .flatten()
    }

    pub(super) fn get_database(&self, database_id: &DatabaseId) -> &Database {
        &self.database_by_id[database_id]
    }

    /// Gets a reference to the specified replica of the specified cluster.
    ///
    /// Returns `None` if either the cluster or the replica does not
    /// exist.
    pub(super) fn try_get_cluster_replica(
        &self,
        id: ClusterId,
        replica_id: ReplicaId,
    ) -> Option<&ClusterReplica> {
        self.try_get_cluster(id)
            .and_then(|cluster| cluster.replica(replica_id))
    }

    /// Gets a reference to the specified replica of the specified cluster.
    ///
    /// Panics if either the cluster or the replica does not exist.
    pub(crate) fn get_cluster_replica(
        &self,
        cluster_id: ClusterId,
        replica_id: ReplicaId,
    ) -> &ClusterReplica {
        self.try_get_cluster_replica(cluster_id, replica_id)
            .unwrap_or_else(|| panic!("unknown cluster replica: {cluster_id}.{replica_id}"))
    }

    pub(super) fn resolve_replica_in_cluster(
        &self,
        cluster_id: &ClusterId,
        replica_name: &str,
    ) -> Result<&ClusterReplica, SqlCatalogError> {
        let cluster = self.get_cluster(*cluster_id);
        let replica_id = cluster
            .replica_id_by_name_
            .get(replica_name)
            .ok_or_else(|| SqlCatalogError::UnknownClusterReplica(replica_name.to_string()))?;
        Ok(&cluster.replicas_by_id_[replica_id])
    }

    /// Get system configuration `name`.
    pub fn get_system_configuration(&self, name: &str) -> Result<&dyn Var, Error> {
        Ok(self.system_configuration.get(name)?)
    }

    /// Parse system configuration `name` with `value` int.
    ///
    /// Returns the parsed value as a string.
    pub(super) fn parse_system_configuration(
        &self,
        name: &str,
        value: VarInput,
    ) -> Result<String, Error> {
        let value = self.system_configuration.parse(name, value)?;
        Ok(value.format())
    }

    /// Gets the schema map for the database matching `database_spec`.
    pub(super) fn resolve_schema_in_database(
        &self,
        database_spec: &ResolvedDatabaseSpecifier,
        schema_name: &str,
        conn_id: &ConnectionId,
    ) -> Result<&Schema, SqlCatalogError> {
        let schema = match database_spec {
            ResolvedDatabaseSpecifier::Ambient if schema_name == MZ_TEMP_SCHEMA => {
                self.temporary_namespaces.schema(conn_id)
            }
            ResolvedDatabaseSpecifier::Ambient => self
                .ambient_schemas_by_name
                .get(schema_name)
                .and_then(|id| self.ambient_schemas_by_id.get(id)),
            ResolvedDatabaseSpecifier::Id(id) => self.database_by_id.get(id).and_then(|db| {
                db.schemas_by_name
                    .get(schema_name)
                    .and_then(|id| db.schemas_by_id.get(id))
            }),
        };
        schema.ok_or_else(|| SqlCatalogError::UnknownSchema(schema_name.into()))
    }

    /// Try to get a schema, returning `None` if it doesn't exist.
    ///
    /// For temporary schemas, returns `None` if the connection's temporary
    /// namespace hasn't been registered yet (that happens at its first
    /// temporary-item creation).
    pub fn try_get_schema(
        &self,
        database_spec: &ResolvedDatabaseSpecifier,
        schema_spec: &SchemaSpecifier,
        conn_id: &ConnectionId,
    ) -> Option<&Schema> {
        // Keep in sync with `get_schema` and `get_schemas_mut`
        match (database_spec, schema_spec) {
            (ResolvedDatabaseSpecifier::Ambient, SchemaSpecifier::Temporary) => {
                self.temporary_namespaces.schema(conn_id)
            }
            (ResolvedDatabaseSpecifier::Ambient, SchemaSpecifier::Id(id)) => {
                self.ambient_schemas_by_id.get(id)
            }
            (ResolvedDatabaseSpecifier::Id(database_id), SchemaSpecifier::Id(schema_id)) => self
                .database_by_id
                .get(database_id)
                .and_then(|db| db.schemas_by_id.get(schema_id)),
            (ResolvedDatabaseSpecifier::Id(_), SchemaSpecifier::Temporary) => {
                unreachable!("temporary schemas are in the ambient database")
            }
        }
    }

    pub fn get_schema(
        &self,
        database_spec: &ResolvedDatabaseSpecifier,
        schema_spec: &SchemaSpecifier,
        conn_id: &ConnectionId,
    ) -> &Schema {
        // Keep in sync with `try_get_schema` and `get_schemas_mut`
        self.try_get_schema(database_spec, schema_spec, conn_id)
            .expect("schema must exist")
    }

    pub(super) fn find_non_temp_schema(&self, schema_id: &SchemaId) -> &Schema {
        self.database_by_id
            .values()
            .filter_map(|database| database.schemas_by_id.get(schema_id))
            .chain(self.ambient_schemas_by_id.values())
            .filter(|schema| schema.id() == &SchemaSpecifier::from(*schema_id))
            .into_first()
    }

    pub fn get_mz_catalog_schema_id(&self) -> SchemaId {
        self.ambient_schemas_by_name[MZ_CATALOG_SCHEMA]
    }

    pub fn get_mz_catalog_unstable_schema_id(&self) -> SchemaId {
        self.ambient_schemas_by_name[MZ_CATALOG_UNSTABLE_SCHEMA]
    }

    pub fn get_pg_catalog_schema_id(&self) -> SchemaId {
        self.ambient_schemas_by_name[PG_CATALOG_SCHEMA]
    }

    pub fn get_information_schema_id(&self) -> SchemaId {
        self.ambient_schemas_by_name[INFORMATION_SCHEMA]
    }

    pub fn get_mz_internal_schema_id(&self) -> SchemaId {
        self.ambient_schemas_by_name[MZ_INTERNAL_SCHEMA]
    }

    pub fn get_mz_introspection_schema_id(&self) -> SchemaId {
        self.ambient_schemas_by_name[MZ_INTROSPECTION_SCHEMA]
    }

    pub fn get_mz_unsafe_schema_id(&self) -> SchemaId {
        self.ambient_schemas_by_name[MZ_UNSAFE_SCHEMA]
    }

    pub fn system_schema_ids(&self) -> impl Iterator<Item = SchemaId> + '_ {
        SYSTEM_SCHEMAS
            .iter()
            .map(|name| self.ambient_schemas_by_name[*name])
    }

    pub fn is_system_schema_id(&self, id: SchemaId) -> bool {
        self.system_schema_ids().contains(&id)
    }

    pub fn is_system_schema_specifier(&self, spec: SchemaSpecifier) -> bool {
        match spec {
            SchemaSpecifier::Temporary => false,
            SchemaSpecifier::Id(id) => self.is_system_schema_id(id),
        }
    }

    pub fn unstable_schema_ids(&self) -> impl Iterator<Item = SchemaId> + '_ {
        UNSTABLE_SCHEMAS
            .iter()
            .map(|name| self.ambient_schemas_by_name[*name])
    }

    pub fn is_unstable_schema_id(&self, id: SchemaId) -> bool {
        self.unstable_schema_ids().contains(&id)
    }

    pub fn is_unstable_schema_specifier(&self, spec: SchemaSpecifier) -> bool {
        match spec {
            SchemaSpecifier::Temporary => false,
            SchemaSpecifier::Id(id) => self.is_unstable_schema_id(id),
        }
    }

    /// Return all OIDs that are allocated to temporary objects.
    pub(crate) fn get_temporary_oids(&self) -> impl Iterator<Item = u32> + '_ {
        std::iter::empty()
            .chain(self.ambient_schemas_by_id.values().filter_map(|schema| {
                if schema.id.is_temporary() {
                    Some(schema.oid)
                } else {
                    None
                }
            }))
            .chain(self.entry_by_id.values().filter_map(|entry| {
                if entry.item().is_temporary() {
                    Some(entry.oid)
                } else {
                    None
                }
            }))
    }

    /// Optimized lookup for a builtin table.
    ///
    /// Panics if the builtin table doesn't exist in the catalog.
    pub fn resolve_builtin_table(&self, builtin: &'static BuiltinTable) -> CatalogItemId {
        self.resolve_builtin_object(&Builtin::<IdReference>::Table(builtin))
    }

    /// Optimized lookup for a builtin log.
    ///
    /// Panics if the builtin log doesn't exist in the catalog.
    pub fn resolve_builtin_log(&self, builtin: &'static BuiltinLog) -> (CatalogItemId, GlobalId) {
        let item_id = self.resolve_builtin_object(&Builtin::<IdReference>::Log(builtin));
        let log = match self.get_entry(&item_id).item() {
            CatalogItem::Log(log) => log,
            other => unreachable!("programming error, expected BuiltinLog, found {other:?}"),
        };
        (item_id, log.global_id)
    }

    /// Optimized lookup for a builtin storage collection.
    ///
    /// Panics if the builtin storage collection doesn't exist in the catalog.
    pub fn resolve_builtin_source(&self, builtin: &'static BuiltinSource) -> CatalogItemId {
        self.resolve_builtin_object(&Builtin::<IdReference>::Source(builtin))
    }

    /// Optimized lookup for a builtin object.
    ///
    /// Panics if the builtin object doesn't exist in the catalog.
    pub fn resolve_builtin_object<T: TypeReference>(&self, builtin: &Builtin<T>) -> CatalogItemId {
        let schema_id = &self.ambient_schemas_by_name[builtin.schema()];
        let schema = &self.ambient_schemas_by_id[schema_id];
        match builtin.catalog_item_type() {
            CatalogItemType::Type => schema.types[builtin.name()],
            CatalogItemType::Func => schema.functions[builtin.name()],
            CatalogItemType::Table
            | CatalogItemType::Source
            | CatalogItemType::Sink
            | CatalogItemType::MetricSink
            | CatalogItemType::View
            | CatalogItemType::MaterializedView
            | CatalogItemType::Index
            | CatalogItemType::Secret
            | CatalogItemType::Connection => schema.items[builtin.name()],
        }
    }

    /// Resolve a [`BuiltinType<NameReference>`] to a [`BuiltinType<IdReference>`].
    pub fn resolve_builtin_type_references(
        &self,
        builtin: &BuiltinType<NameReference>,
    ) -> BuiltinType<IdReference> {
        let typ: CatalogType<IdReference> = match &builtin.details.typ {
            CatalogType::AclItem => CatalogType::AclItem,
            CatalogType::Array { element_reference } => CatalogType::Array {
                element_reference: self.get_system_type(element_reference).id,
            },
            CatalogType::List {
                element_reference,
                element_modifiers,
            } => CatalogType::List {
                element_reference: self.get_system_type(element_reference).id,
                element_modifiers: element_modifiers.clone(),
            },
            CatalogType::Map {
                key_reference,
                value_reference,
                key_modifiers,
                value_modifiers,
            } => CatalogType::Map {
                key_reference: self.get_system_type(key_reference).id,
                value_reference: self.get_system_type(value_reference).id,
                key_modifiers: key_modifiers.clone(),
                value_modifiers: value_modifiers.clone(),
            },
            CatalogType::Range { element_reference } => CatalogType::Range {
                element_reference: self.get_system_type(element_reference).id,
            },
            CatalogType::Record { fields } => CatalogType::Record {
                fields: fields
                    .into_iter()
                    .map(|f| CatalogRecordField {
                        name: f.name.clone(),
                        type_reference: self.get_system_type(f.type_reference).id,
                        type_modifiers: f.type_modifiers.clone(),
                    })
                    .collect(),
            },
            CatalogType::Bool => CatalogType::Bool,
            CatalogType::Bytes => CatalogType::Bytes,
            CatalogType::Char => CatalogType::Char,
            CatalogType::Date => CatalogType::Date,
            CatalogType::Float32 => CatalogType::Float32,
            CatalogType::Float64 => CatalogType::Float64,
            CatalogType::Int16 => CatalogType::Int16,
            CatalogType::Int32 => CatalogType::Int32,
            CatalogType::Int64 => CatalogType::Int64,
            CatalogType::UInt16 => CatalogType::UInt16,
            CatalogType::UInt32 => CatalogType::UInt32,
            CatalogType::UInt64 => CatalogType::UInt64,
            CatalogType::MzTimestamp => CatalogType::MzTimestamp,
            CatalogType::Interval => CatalogType::Interval,
            CatalogType::Jsonb => CatalogType::Jsonb,
            CatalogType::Numeric => CatalogType::Numeric,
            CatalogType::Oid => CatalogType::Oid,
            CatalogType::PgLegacyChar => CatalogType::PgLegacyChar,
            CatalogType::PgLegacyName => CatalogType::PgLegacyName,
            CatalogType::Pseudo => CatalogType::Pseudo,
            CatalogType::RegClass => CatalogType::RegClass,
            CatalogType::RegProc => CatalogType::RegProc,
            CatalogType::RegType => CatalogType::RegType,
            CatalogType::String => CatalogType::String,
            CatalogType::Time => CatalogType::Time,
            CatalogType::Timestamp => CatalogType::Timestamp,
            CatalogType::TimestampTz => CatalogType::TimestampTz,
            CatalogType::Uuid => CatalogType::Uuid,
            CatalogType::VarChar => CatalogType::VarChar,
            CatalogType::Int2Vector => CatalogType::Int2Vector,
            CatalogType::MzAclItem => CatalogType::MzAclItem,
        };

        BuiltinType {
            name: builtin.name,
            schema: builtin.schema,
            oid: builtin.oid,
            details: CatalogTypeDetails {
                array_id: builtin.details.array_id,
                typ,
                pg_metadata: builtin.details.pg_metadata.clone(),
            },
        }
    }

    pub fn config(&self) -> &mz_sql::catalog::CatalogConfig {
        &self.config
    }

    pub fn resolve_database(&self, database_name: &str) -> Result<&Database, SqlCatalogError> {
        match self.database_by_name.get(database_name) {
            Some(id) => Ok(&self.database_by_id[id]),
            None => Err(SqlCatalogError::UnknownDatabase(database_name.into())),
        }
    }

    pub fn resolve_schema(
        &self,
        current_database: Option<&DatabaseId>,
        database_name: Option<&str>,
        schema_name: &str,
        conn_id: &ConnectionId,
    ) -> Result<&Schema, SqlCatalogError> {
        let database_spec = match database_name {
            // If a database is explicitly specified, validate it. Note that we
            // intentionally do not validate `current_database` to permit
            // querying `mz_catalog` with an invalid session database, e.g., so
            // that you can run `SHOW DATABASES` to *find* a valid database.
            Some(database) => Some(ResolvedDatabaseSpecifier::Id(
                self.resolve_database(database)?.id().clone(),
            )),
            None => current_database.map(|id| ResolvedDatabaseSpecifier::Id(id.clone())),
        };

        // First try to find the schema in the named database.
        if let Some(database_spec) = database_spec {
            if let Ok(schema) =
                self.resolve_schema_in_database(&database_spec, schema_name, conn_id)
            {
                return Ok(schema);
            }
        }

        // Then fall back to the ambient database.
        if let Ok(schema) = self.resolve_schema_in_database(
            &ResolvedDatabaseSpecifier::Ambient,
            schema_name,
            conn_id,
        ) {
            return Ok(schema);
        }

        Err(SqlCatalogError::UnknownSchema(schema_name.into()))
    }

    /// Optimized lookup for a system schema.
    ///
    /// Panics if the system schema doesn't exist in the catalog.
    pub fn resolve_system_schema(&self, name: &'static str) -> SchemaId {
        self.ambient_schemas_by_name[name]
    }

    pub fn resolve_search_path(
        &self,
        session: &dyn SessionMetadata,
    ) -> Vec<(ResolvedDatabaseSpecifier, SchemaSpecifier)> {
        let database = self
            .database_by_name
            .get(session.database())
            .map(|id| id.clone());

        // NOTE: This drops schemas that don't resolve, and consumers rely on
        // every returned entry existing (e.g. `allocate_full_name` and the
        // `current_schemas` evaluation look schemas up infallibly). In
        // particular, `mz_temp` is dropped until the session's temporary
        // namespace is registered at its first temporary-item creation.
        session
            .search_path()
            .iter()
            .map(|schema| {
                self.resolve_schema(database.as_ref(), None, schema.as_str(), session.conn_id())
            })
            .filter_map(|schema| schema.ok())
            .map(|schema| (schema.name().database.clone(), schema.id().clone()))
            .collect()
    }

    pub fn effective_search_path(
        &self,
        search_path: &[(ResolvedDatabaseSpecifier, SchemaSpecifier)],
        include_temp_schema: bool,
    ) -> Vec<(ResolvedDatabaseSpecifier, SchemaSpecifier)> {
        let mut v = Vec::with_capacity(search_path.len() + 3);
        // Temp schema is only included for relations and data types, not for functions and operators
        let temp_schema = (
            ResolvedDatabaseSpecifier::Ambient,
            SchemaSpecifier::Temporary,
        );
        if include_temp_schema && !search_path.contains(&temp_schema) {
            v.push(temp_schema);
        }
        let default_schemas = [
            (
                ResolvedDatabaseSpecifier::Ambient,
                SchemaSpecifier::Id(self.get_mz_catalog_schema_id()),
            ),
            (
                ResolvedDatabaseSpecifier::Ambient,
                SchemaSpecifier::Id(self.get_pg_catalog_schema_id()),
            ),
        ];
        for schema in default_schemas.into_iter() {
            if !search_path.contains(&schema) {
                v.push(schema);
            }
        }
        v.extend_from_slice(search_path);
        v
    }

    pub fn resolve_cluster(&self, name: &str) -> Result<&Cluster, SqlCatalogError> {
        let id = self
            .clusters_by_name
            .get(name)
            .ok_or_else(|| SqlCatalogError::UnknownCluster(name.to_string()))?;
        Ok(&self.clusters_by_id[id])
    }

    pub fn resolve_builtin_cluster(&self, cluster: &BuiltinCluster) -> &Cluster {
        let id = self
            .clusters_by_name
            .get(cluster.name)
            .expect("failed to lookup BuiltinCluster by name");
        self.clusters_by_id
            .get(id)
            .expect("failed to lookup BuiltinCluster by ID")
    }

    pub fn resolve_cluster_replica(
        &self,
        cluster_replica_name: &QualifiedReplica,
    ) -> Result<&ClusterReplica, SqlCatalogError> {
        let cluster = self.resolve_cluster(cluster_replica_name.cluster.as_str())?;
        let replica_name = cluster_replica_name.replica.as_str();
        let replica_id = cluster
            .replica_id(replica_name)
            .ok_or_else(|| SqlCatalogError::UnknownClusterReplica(replica_name.to_string()))?;
        Ok(cluster.replica(replica_id).expect("Must exist"))
    }

    /// Resolves [`PartialItemName`] into a [`CatalogEntry`].
    ///
    /// If `name` does not specify a database, the `current_database` is used.
    /// If `name` does not specify a schema, then the schemas in `search_path`
    /// are searched in order.
    #[allow(clippy::useless_let_if_seq)]
    pub fn resolve(
        &self,
        get_schema_entries: fn(&Schema) -> &BTreeMap<String, CatalogItemId>,
        current_database: Option<&DatabaseId>,
        search_path: &Vec<(ResolvedDatabaseSpecifier, SchemaSpecifier)>,
        name: &PartialItemName,
        conn_id: &ConnectionId,
        err_gen: fn(String) -> SqlCatalogError,
    ) -> Result<&CatalogEntry, SqlCatalogError> {
        // If a schema name was specified, just try to find the item in that
        // schema. If no schema was specified, try to find the item in the connection's
        // temporary schema. If the item is not found, try to find the item in every
        // schema in the search path.
        let schemas = match &name.schema {
            Some(schema_name) => {
                match self.resolve_schema(
                    current_database,
                    name.database.as_deref(),
                    schema_name,
                    conn_id,
                ) {
                    Ok(schema) => vec![(schema.name.database.clone(), schema.id.clone())],
                    Err(e) => return Err(e),
                }
            }
            None => match self
                .try_get_schema(
                    &ResolvedDatabaseSpecifier::Ambient,
                    &SchemaSpecifier::Temporary,
                    conn_id,
                )
                .and_then(|schema| schema.items.get(&name.item))
            {
                Some(id) => return Ok(self.get_entry(id)),
                None => search_path.to_vec(),
            },
        };

        for (database_spec, schema_spec) in &schemas {
            // Use try_get_schema because the temp schema might not exist yet
            // (it's created lazily when the first temp object is created).
            let Some(schema) = self.try_get_schema(database_spec, schema_spec, conn_id) else {
                continue;
            };

            if let Some(id) = get_schema_entries(schema).get(&name.item) {
                return Ok(&self.entry_by_id[id]);
            }
        }

        // Some relations that have previously lived in the `mz_internal` schema have been moved to
        // `mz_catalog_unstable` or `mz_introspection`. To simplify the transition for users, we
        // automatically let uses of the old schema resolve to the new ones as well.
        // TODO(database-issues#8173) remove this after sufficient time has passed
        let mz_internal_schema = SchemaSpecifier::Id(self.get_mz_internal_schema_id());
        if schemas.iter().any(|(_, spec)| *spec == mz_internal_schema) {
            for schema_id in [
                self.get_mz_catalog_unstable_schema_id(),
                self.get_mz_introspection_schema_id(),
            ] {
                let schema = self.get_schema(
                    &ResolvedDatabaseSpecifier::Ambient,
                    &SchemaSpecifier::Id(schema_id),
                    conn_id,
                );

                if let Some(id) = get_schema_entries(schema).get(&name.item) {
                    debug!(
                        github_27831 = true,
                        "encountered use of outdated schema `mz_internal` for relation: {name}",
                    );
                    return Ok(&self.entry_by_id[id]);
                }
            }
        }

        Err(err_gen(name.to_string()))
    }

    /// Resolves `name` to a non-function [`CatalogEntry`].
    pub fn resolve_entry(
        &self,
        current_database: Option<&DatabaseId>,
        search_path: &Vec<(ResolvedDatabaseSpecifier, SchemaSpecifier)>,
        name: &PartialItemName,
        conn_id: &ConnectionId,
    ) -> Result<&CatalogEntry, SqlCatalogError> {
        self.resolve(
            |schema| &schema.items,
            current_database,
            search_path,
            name,
            conn_id,
            SqlCatalogError::UnknownItem,
        )
    }

    /// Resolves `name` to a function [`CatalogEntry`].
    pub fn resolve_function(
        &self,
        current_database: Option<&DatabaseId>,
        search_path: &Vec<(ResolvedDatabaseSpecifier, SchemaSpecifier)>,
        name: &PartialItemName,
        conn_id: &ConnectionId,
    ) -> Result<&CatalogEntry, SqlCatalogError> {
        self.resolve(
            |schema| &schema.functions,
            current_database,
            search_path,
            name,
            conn_id,
            |name| SqlCatalogError::UnknownFunction {
                name,
                alternative: None,
            },
        )
    }

    /// Resolves `name` to a type [`CatalogEntry`].
    pub fn resolve_type(
        &self,
        current_database: Option<&DatabaseId>,
        search_path: &Vec<(ResolvedDatabaseSpecifier, SchemaSpecifier)>,
        name: &PartialItemName,
        conn_id: &ConnectionId,
    ) -> Result<&CatalogEntry, SqlCatalogError> {
        static NON_PG_CATALOG_TYPES: LazyLock<
            BTreeMap<&'static str, &'static BuiltinType<NameReference>>,
        > = LazyLock::new(|| {
            BUILTINS::types()
                .filter(|typ| typ.schema != PG_CATALOG_SCHEMA)
                .map(|typ| (typ.name, typ))
                .collect()
        });

        let entry = self.resolve(
            |schema| &schema.types,
            current_database,
            search_path,
            name,
            conn_id,
            |name| SqlCatalogError::UnknownType { name },
        )?;

        if conn_id != &SYSTEM_CONN_ID && name.schema.as_deref() == Some(PG_CATALOG_SCHEMA) {
            if let Some(typ) = NON_PG_CATALOG_TYPES.get(entry.name().item.as_str()) {
                warn!(
                    "user specified an incorrect schema of {} for the type {}, which should be in \
                    the {} schema. This works now due to a bug but will be fixed in a later release.",
                    PG_CATALOG_SCHEMA.quoted(),
                    typ.name.quoted(),
                    typ.schema.quoted(),
                )
            }
        }

        Ok(entry)
    }

    /// For an [`ObjectId`] gets the corresponding [`CommentObjectId`].
    pub(super) fn get_comment_id(&self, object_id: ObjectId) -> CommentObjectId {
        match object_id {
            ObjectId::Item(item_id) => self.get_entry(&item_id).comment_object_id(),
            ObjectId::Role(role_id) => CommentObjectId::Role(role_id),
            ObjectId::Database(database_id) => CommentObjectId::Database(database_id),
            ObjectId::Schema((database, schema)) => CommentObjectId::Schema((database, schema)),
            ObjectId::Cluster(cluster_id) => CommentObjectId::Cluster(cluster_id),
            ObjectId::ClusterReplica(cluster_replica_id) => {
                CommentObjectId::ClusterReplica(cluster_replica_id)
            }
            ObjectId::NetworkPolicy(network_policy_id) => {
                CommentObjectId::NetworkPolicy(network_policy_id)
            }
        }
    }

    /// Return current system configuration.
    pub fn system_config(&self) -> &SystemVars {
        &self.system_configuration
    }

    /// Returns the cluster-coherent scoped optimizer-feature overrides for
    /// `cluster_id` from the in-memory scoped-parameter working copy, or empty
    /// if the cluster has none.
    pub fn cluster_scoped_optimizer_overrides(
        &self,
        cluster_id: ClusterId,
    ) -> OptimizerFeatureOverrides {
        self.scoped_system_parameters
            .cluster
            .get(&cluster_id)
            .cloned()
            .map(OptimizerFeatureOverrides::from)
            .unwrap_or_default()
    }

    /// Returns the entire scoped system-parameter working copy, read by the
    /// coordinator's scoped-parameter reconcile path.
    pub fn scoped_system_parameters(&self) -> &ScopedParameters {
        &self.scoped_system_parameters
    }

    /// Return a mutable reference to the current system configuration.
    pub fn system_config_mut(&mut self) -> &mut SystemVars {
        Arc::make_mut(&mut self.system_configuration)
    }

    /// Serializes the catalog's in-memory state.
    ///
    /// There are no guarantees about the format of the serialized state, except
    /// that the serialized state for two identical catalogs will compare
    /// identically.
    ///
    /// Some consumers would like the ability to overwrite the `unfinalized_shards` catalog field,
    /// which they can accomplish by passing in a value of `Some` for the `unfinalized_shards`
    /// argument.
    pub fn dump(&self, unfinalized_shards: Option<BTreeSet<String>>) -> Result<String, Error> {
        // Dump the base catalog.
        let mut dump = serde_json::to_value(&self).map_err(|e| {
            Error::new(ErrorKind::Unstructured(format!(
                // Don't panic here because we don't have compile-time failures for maps with
                // non-string keys.
                "internal error: could not dump catalog: {}",
                e
            )))
        })?;

        let dump_obj = dump.as_object_mut().expect("state must have been dumped");
        // Stitch in system parameter defaults.
        dump_obj.insert(
            "system_parameter_defaults".into(),
            serde_json::json!(self.system_config().defaults()),
        );
        // Potentially overwrite unfinalized shards.
        if let Some(unfinalized_shards) = unfinalized_shards {
            dump_obj
                .get_mut("storage_metadata")
                .expect("known to exist")
                .as_object_mut()
                .expect("storage_metadata is an object")
                .insert(
                    "unfinalized_shards".into(),
                    serde_json::json!(unfinalized_shards),
                );
        }
        // Remove GlobalIds for temporary objects from the mapping.
        //
        // Post-test consistency checks with the durable catalog don't know about temporary items
        // since they're kept entirely in memory.
        let temporary_gids: Vec<_> = self
            .entry_by_global_id
            .iter()
            .filter(|(_gid, item_id)| self.get_entry(item_id).conn_id().is_some())
            .map(|(gid, _item_id)| *gid)
            .collect();
        if !temporary_gids.is_empty() {
            let gids = dump_obj
                .get_mut("entry_by_global_id")
                .expect("known_to_exist")
                .as_object_mut()
                .expect("entry_by_global_id is an object");
            for gid in temporary_gids {
                gids.remove(&gid.to_string());
            }
        }
        // We exclude role_auth_by_id because it contains password information
        // which should not be included in the dump.
        dump_obj.remove("role_auth_by_id");
        // The mock authentication nonce is a server-wide secret used to derive
        // deterministic mock SASL challenges for absent or password-less roles.
        // Leaking it would re-enable the username enumeration the mock challenge
        // defends against, so it must not appear in the dump either.
        dump_obj.remove("mock_authentication_nonce");

        // Emit as pretty-printed JSON.
        Ok(serde_json::to_string_pretty(&dump).expect("cannot fail on serde_json::Value"))
    }

    pub fn availability_zones(&self) -> &[String] {
        &self.availability_zones
    }

    pub fn concretize_replica_location(
        &self,
        location: mz_catalog::durable::ReplicaLocation,
        allowed_sizes: &Vec<String>,
        allowed_availability_zones: Option<&[String]>,
        allow_disabled: bool,
    ) -> Result<ReplicaLocation, Error> {
        let location = match location {
            mz_catalog::durable::ReplicaLocation::Unmanaged {
                storagectl_addrs,
                computectl_addrs,
            } => {
                if allowed_availability_zones.is_some() {
                    return Err(Error {
                        kind: ErrorKind::Internal(
                            "tried concretize unmanaged replica with specific availability_zones"
                                .to_string(),
                        ),
                    });
                }
                ReplicaLocation::Unmanaged(UnmanagedReplicaLocation {
                    storagectl_addrs,
                    computectl_addrs,
                })
            }
            mz_catalog::durable::ReplicaLocation::Managed {
                size,
                // The AZ list the replica was provisioned under: provisioning
                // paths pass the cluster's pool as `allowed_availability_zones`
                // to stamp it, while rebuilds from durable state pass `None` to
                // keep it. For an unmanaged cluster's replica it is the single
                // user-pinned AZ.
                availability_zones,
                billed_as,
                internal,
                pending,
            } => {
                self.ensure_valid_replica_size(allowed_sizes, &size, allow_disabled)?;
                let cluster_replica_sizes = &self.cluster_replica_sizes;

                ReplicaLocation::Managed(ManagedReplicaLocation {
                    allocation: cluster_replica_sizes
                        .0
                        .get(&size)
                        .expect("catalog out of sync")
                        .clone(),
                    availability_zones: match allowed_availability_zones {
                        Some(azs) => azs.to_vec(),
                        None => availability_zones,
                    },
                    size,
                    billed_as,
                    internal,
                    pending,
                })
            }
        };
        Ok(location)
    }

    pub(crate) fn ensure_valid_replica_size(
        &self,
        allowed_sizes: &[String],
        size: &String,
        allow_disabled: bool,
    ) -> Result<(), Error> {
        let cluster_replica_sizes = &self.cluster_replica_sizes;

        if !cluster_replica_sizes.0.contains_key(size)
            || (!allowed_sizes.is_empty() && !allowed_sizes.contains(size))
            || (!allow_disabled && cluster_replica_sizes.0[size].disabled)
        {
            let mut entries = cluster_replica_sizes
                .enabled_allocations()
                .collect::<Vec<_>>();

            if !allowed_sizes.is_empty() {
                let allowed_sizes = BTreeSet::<&String>::from_iter(allowed_sizes.iter());
                entries.retain(|(name, _)| allowed_sizes.contains(name));
            }

            entries.sort_by_key(
                |(
                    _name,
                    ReplicaAllocation {
                        scale, cpu_limit, ..
                    },
                )| (scale, cpu_limit),
            );

            Err(Error {
                kind: ErrorKind::InvalidClusterReplicaSize {
                    size: size.to_owned(),
                    expected: entries.into_iter().map(|(name, _)| name.clone()).collect(),
                },
            })
        } else {
            Ok(())
        }
    }

    pub fn ensure_not_reserved_role(&self, role_id: &RoleId) -> Result<(), Error> {
        if role_id.is_builtin() {
            let role = self.get_role(role_id);
            Err(Error::new(ErrorKind::ReservedRoleName(
                role.name().to_string(),
            )))
        } else {
            Ok(())
        }
    }

    pub fn ensure_not_reserved_network_policy(
        &self,
        network_policy_id: &NetworkPolicyId,
    ) -> Result<(), Error> {
        if network_policy_id.is_builtin() {
            let policy = self.get_network_policy(network_policy_id);
            Err(Error::new(ErrorKind::ReservedNetworkPolicyName(
                policy.name.clone(),
            )))
        } else {
            Ok(())
        }
    }

    pub fn ensure_grantable_role(&self, role_id: &RoleId) -> Result<(), Error> {
        let is_grantable = !role_id.is_public() && !role_id.is_system();
        if is_grantable {
            Ok(())
        } else {
            let role = self.get_role(role_id);
            Err(Error::new(ErrorKind::UngrantableRoleName(
                role.name().to_string(),
            )))
        }
    }

    pub fn ensure_not_system_role(&self, role_id: &RoleId) -> Result<(), Error> {
        if role_id.is_system() {
            let role = self.get_role(role_id);
            Err(Error::new(ErrorKind::ReservedSystemRoleName(
                role.name().to_string(),
            )))
        } else {
            Ok(())
        }
    }

    pub fn ensure_not_predefined_role(&self, role_id: &RoleId) -> Result<(), Error> {
        if role_id.is_predefined() {
            let role = self.get_role(role_id);
            Err(Error::new(ErrorKind::ReservedSystemRoleName(
                role.name().to_string(),
            )))
        } else {
            Ok(())
        }
    }

    // TODO(mjibson): Is there a way to make this a closure to avoid explicitly
    // passing tx, and session?
    pub(crate) fn add_to_audit_log(
        system_configuration: &SystemVars,
        oracle_write_ts: mz_repr::Timestamp,
        session: Option<&ConnMeta>,
        tx: &mut mz_catalog::durable::Transaction,
        audit_events: &mut Vec<VersionedEvent>,
        event_type: EventType,
        object_type: ObjectType,
        details: EventDetails,
    ) -> Result<(), Error> {
        let user = session.map(|session| session.user().name.to_string());

        // unsafe_mock_audit_event_timestamp can only be set to Some when running in unsafe mode.

        let occurred_at = match system_configuration.unsafe_mock_audit_event_timestamp() {
            Some(ts) => ts.into(),
            _ => oracle_write_ts.into(),
        };
        let id = tx.allocate_audit_log_id()?;
        let event = VersionedEvent::new(id, event_type, object_type, details, user, occurred_at);
        audit_events.push(event.clone());
        tx.insert_audit_log_event(event);
        Ok(())
    }

    pub(super) fn get_owner_id(&self, id: &ObjectId, conn_id: &ConnectionId) -> Option<RoleId> {
        match id {
            ObjectId::Cluster(id) => Some(self.get_cluster(*id).owner_id()),
            ObjectId::ClusterReplica((cluster_id, replica_id)) => Some(
                self.get_cluster_replica(*cluster_id, *replica_id)
                    .owner_id(),
            ),
            ObjectId::Database(id) => Some(self.get_database(id).owner_id()),
            ObjectId::Schema((database_spec, schema_spec)) => Some(
                self.get_schema(database_spec, schema_spec, conn_id)
                    .owner_id(),
            ),
            ObjectId::Item(id) => Some(*self.get_entry(id).owner_id()),
            ObjectId::Role(_) => None,
            ObjectId::NetworkPolicy(id) => Some(self.get_network_policy(id).owner_id.clone()),
        }
    }

    pub(super) fn get_object_type(&self, object_id: &ObjectId) -> mz_sql::catalog::ObjectType {
        match object_id {
            ObjectId::Cluster(_) => mz_sql::catalog::ObjectType::Cluster,
            ObjectId::ClusterReplica(_) => mz_sql::catalog::ObjectType::ClusterReplica,
            ObjectId::Database(_) => mz_sql::catalog::ObjectType::Database,
            ObjectId::Schema(_) => mz_sql::catalog::ObjectType::Schema,
            ObjectId::Role(_) => mz_sql::catalog::ObjectType::Role,
            ObjectId::Item(id) => self.get_entry(id).item_type().into(),
            ObjectId::NetworkPolicy(_) => mz_sql::catalog::ObjectType::NetworkPolicy,
        }
    }

    pub(super) fn get_system_object_type(
        &self,
        id: &SystemObjectId,
    ) -> mz_sql::catalog::SystemObjectType {
        match id {
            SystemObjectId::Object(object_id) => {
                SystemObjectType::Object(self.get_object_type(object_id))
            }
            SystemObjectId::System => SystemObjectType::System,
        }
    }

    /// Returns a read-only view of the current [`StorageMetadata`].
    ///
    /// To write to this struct, you must use a catalog transaction.
    pub fn storage_metadata(&self) -> &StorageMetadata {
        &self.storage_metadata
    }

    /// For the Sources ids in `ids`, return their compaction windows.
    pub fn source_compaction_windows(
        &self,
        ids: impl IntoIterator<Item = CatalogItemId>,
    ) -> BTreeMap<CompactionWindow, BTreeSet<CatalogItemId>> {
        let mut cws: BTreeMap<CompactionWindow, BTreeSet<CatalogItemId>> = BTreeMap::new();
        let mut seen = BTreeSet::new();
        for item_id in ids {
            if !seen.insert(item_id) {
                continue;
            }
            let entry = self.get_entry(&item_id);
            match entry.item() {
                CatalogItem::Source(source) => {
                    let source_cw = source.custom_logical_compaction_window.unwrap_or_default();
                    cws.entry(source_cw).or_default().insert(item_id);
                }
                CatalogItem::Table(table) => {
                    let table_cw = table.custom_logical_compaction_window.unwrap_or_default();
                    match &table.data_source {
                        TableDataSource::DataSource {
                            desc:
                                DataSourceDesc::IngestionExport { .. }
                                // Also match webhook tables (source-to-table migration).
                                | DataSourceDesc::Webhook { .. },
                            timeline: _,
                        } => {
                            cws.entry(table_cw).or_default().insert(item_id);
                        }
                        // Regular tables handle compaction directly in
                        // catalog_implications, not through this function.
                        TableDataSource::TableWrites { .. } => {}
                        TableDataSource::DataSource {
                            desc:
                                DataSourceDesc::Ingestion { .. }
                                | DataSourceDesc::OldSyntaxIngestion { .. }
                                | DataSourceDesc::Introspection(_)
                                | DataSourceDesc::Progress
                                | DataSourceDesc::Catalog,
                            ..
                        } => {
                            unreachable!(
                                "unexpected DataSourceDesc for table {item_id}: {:?}",
                                table.data_source
                            )
                        }
                    }
                }
                _ => {
                    // Views could depend on sources, so ignore them if added by used_by above.
                    continue;
                }
            }
        }
        cws
    }

    pub fn comment_id_to_item_id(id: &CommentObjectId) -> Option<CatalogItemId> {
        match id {
            CommentObjectId::Table(id)
            | CommentObjectId::View(id)
            | CommentObjectId::MaterializedView(id)
            | CommentObjectId::Source(id)
            | CommentObjectId::Sink(id)
            | CommentObjectId::MetricSink(id)
            | CommentObjectId::Index(id)
            | CommentObjectId::Func(id)
            | CommentObjectId::Connection(id)
            | CommentObjectId::Type(id)
            | CommentObjectId::Secret(id) => Some(*id),
            CommentObjectId::Role(_)
            | CommentObjectId::Database(_)
            | CommentObjectId::Schema(_)
            | CommentObjectId::Cluster(_)
            | CommentObjectId::ClusterReplica(_)
            | CommentObjectId::NetworkPolicy(_) => None,
        }
    }

    pub fn get_comment_id_entry(&self, id: &CommentObjectId) -> Option<&CatalogEntry> {
        Self::comment_id_to_item_id(id).map(|id| self.get_entry(&id))
    }

    pub fn comment_id_to_audit_log_name(
        &self,
        id: CommentObjectId,
        conn_id: &ConnectionId,
    ) -> String {
        match id {
            CommentObjectId::Table(id)
            | CommentObjectId::View(id)
            | CommentObjectId::MaterializedView(id)
            | CommentObjectId::Source(id)
            | CommentObjectId::Sink(id)
            | CommentObjectId::MetricSink(id)
            | CommentObjectId::Index(id)
            | CommentObjectId::Func(id)
            | CommentObjectId::Connection(id)
            | CommentObjectId::Type(id)
            | CommentObjectId::Secret(id) => {
                let item = self.get_entry(&id);
                let name = self.resolve_full_name(item.name(), Some(conn_id));
                name.to_string()
            }
            CommentObjectId::Role(id) => self.get_role(&id).name.clone(),
            CommentObjectId::Database(id) => self.get_database(&id).name.clone(),
            CommentObjectId::Schema((spec, schema_id)) => {
                let schema = self.get_schema(&spec, &schema_id, conn_id);
                self.resolve_full_schema_name(&schema.name).to_string()
            }
            CommentObjectId::Cluster(id) => self.get_cluster(id).name.clone(),
            CommentObjectId::ClusterReplica((cluster_id, replica_id)) => {
                let cluster = self.get_cluster(cluster_id);
                let replica = self.get_cluster_replica(cluster_id, replica_id);
                QualifiedReplica {
                    cluster: Ident::new_unchecked(cluster.name.clone()),
                    replica: Ident::new_unchecked(replica.name.clone()),
                }
                .to_string()
            }
            CommentObjectId::NetworkPolicy(id) => self.get_network_policy(&id).name.clone(),
        }
    }

    pub fn mock_authentication_nonce(&self) -> String {
        self.mock_authentication_nonce.clone().unwrap_or_default()
    }
}

impl ConnectionResolver for CatalogState {
    fn resolve_connection(
        &self,
        id: CatalogItemId,
    ) -> mz_storage_types::connections::Connection<InlinedConnection> {
        use mz_storage_types::connections::Connection::*;
        match self
            .get_entry(&id)
            .connection()
            .expect("catalog out of sync")
            .details
            .to_connection()
        {
            Kafka(conn) => Kafka(conn.into_inline_connection(self)),
            Postgres(conn) => Postgres(conn.into_inline_connection(self)),
            Csr(conn) => Csr(conn.into_inline_connection(self)),
            GlueSchemaRegistry(conn) => GlueSchemaRegistry(conn.into_inline_connection(self)),
            Ssh(conn) => Ssh(conn),
            Aws(conn) => Aws(conn),
            AwsPrivatelink(conn) => AwsPrivatelink(conn),
            Gcp(conn) => Gcp(conn),
            MySql(conn) => MySql(conn.into_inline_connection(self)),
            SqlServer(conn) => SqlServer(conn.into_inline_connection(self)),
            IcebergCatalog(conn) => IcebergCatalog(conn.into_inline_connection(self)),
        }
    }
}

impl OptimizerCatalog for CatalogState {
    fn get_entry(&self, id: &GlobalId) -> CatalogCollectionEntry {
        CatalogState::get_entry_by_global_id(self, id)
    }
    fn get_entry_by_item_id(&self, id: &CatalogItemId) -> &CatalogEntry {
        CatalogState::get_entry(self, id)
    }
    fn resolve_full_name(
        &self,
        name: &QualifiedItemName,
        conn_id: Option<&ConnectionId>,
    ) -> FullItemName {
        CatalogState::resolve_full_name(self, name, conn_id)
    }
    fn get_indexes_on(
        &self,
        id: GlobalId,
        cluster: ClusterId,
    ) -> Box<dyn Iterator<Item = (GlobalId, &Index)> + '_> {
        Box::new(CatalogState::get_indexes_on(self, id, cluster))
    }
}

impl OptimizerCatalog for Catalog {
    fn get_entry(&self, id: &GlobalId) -> CatalogCollectionEntry {
        self.state.get_entry_by_global_id(id)
    }

    fn get_entry_by_item_id(&self, id: &CatalogItemId) -> &CatalogEntry {
        self.state.get_entry(id)
    }

    fn resolve_full_name(
        &self,
        name: &QualifiedItemName,
        conn_id: Option<&ConnectionId>,
    ) -> FullItemName {
        self.state.resolve_full_name(name, conn_id)
    }

    fn get_indexes_on(
        &self,
        id: GlobalId,
        cluster: ClusterId,
    ) -> Box<dyn Iterator<Item = (GlobalId, &Index)> + '_> {
        Box::new(self.state.get_indexes_on(id, cluster))
    }
}

impl Catalog {
    pub fn as_optimizer_catalog(self: Arc<Self>) -> Arc<dyn OptimizerCatalog> {
        self
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A deep dependency chain (a long chain of stacked views) must not
    /// overflow the stack when computing its dependents, and the dependents
    /// must come out in reverse-dependency order (deepest dependent first,
    /// root last) so that `DROP ... CASCADE` drops them in a valid order.
    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `TLS_client_method`
    async fn item_dependents_deep_chain_no_stack_overflow() {
        use mz_ore::cast::CastFrom;

        Catalog::with_debug(|mut catalog| async move {
            // Deep enough that the previous recursive implementation overflowed
            // the stack.
            const DEPTH: usize = 100_000;
            // Well above any id the debug catalog assigns, so the synthetic
            // chain does not collide with real entries.
            const BASE: u64 = 1 << 40;

            // Clone a builtin log entry as a template so we don't have to
            // construct a `CatalogItem` by hand. `item_dependents` only reads
            // `used_by`/`progress_id`, and a log's `progress_id` is `None`.
            let template = catalog
                .state
                .entry_by_id
                .values()
                .find(|entry| matches!(entry.item(), CatalogItem::Log(_)))
                .expect("debug catalog has log sources")
                .clone();

            // Build a chain where entry `BASE + i` is used by `BASE + i + 1`.
            for i in 0..=DEPTH {
                let id = CatalogItemId::User(BASE + u64::cast_from(i));
                let mut entry = template.clone();
                entry.id = id;
                entry.referenced_by = Vec::new();
                entry.used_by = if i < DEPTH {
                    vec![CatalogItemId::User(BASE + u64::cast_from(i + 1))]
                } else {
                    Vec::new()
                };
                catalog.state.entry_by_id.insert(id, entry);
            }

            let mut seen = BTreeSet::new();
            let dependents = catalog
                .state
                .item_dependents(CatalogItemId::User(BASE), &mut seen);

            // Every element of the chain appears exactly once.
            assert_eq!(dependents.len(), DEPTH + 1);
            // Reverse-dependency order: the deepest dependent is first and the
            // root we asked about is last.
            for (offset, dependent) in dependents.iter().enumerate() {
                let expected = CatalogItemId::User(BASE + u64::cast_from(DEPTH - offset));
                assert_eq!(dependent, &ObjectId::Item(expected));
            }

            catalog.expire().await;
        })
        .await
    }

    /// Read-then-write dependency validation walks the transitive `uses()` of
    /// the read set. A deep chain of stacked views (user controlled, arbitrarily
    /// deep) must be validated without overflowing the coordinator thread's
    /// stack, so the traversal must not recurse.
    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `TLS_client_method`
    async fn validate_read_then_write_deep_chain_no_stack_overflow() {
        use crate::coord::read_then_write::validate_read_then_write_dependencies;

        Catalog::with_debug(|mut catalog| async move {
            // Deep enough that the previous recursive implementation overflowed
            // the stack.
            const DEPTH: usize = 100_000;
            const BASE: u64 = 1 << 40;
            insert_synthetic_view_chain(&mut catalog, BASE, DEPTH);

            // A generous bound so this test isolates the no-overflow property
            // rather than the dependency limit.
            validate_read_then_write_dependencies(
                &catalog,
                [CatalogItemId::User(BASE)],
                usize::MAX,
            )
            .expect("deep chain of user views is valid for read-then-write");

            catalog.expire().await;
        })
        .await
    }

    /// Read-then-write dependency validation is bounded: a read set with more
    /// transitive dependencies than the limit is rejected with a clean error
    /// rather than walking an unbounded graph.
    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `TLS_client_method`
    async fn validate_read_then_write_dependency_limit() {
        use crate::coord::read_then_write::validate_read_then_write_dependencies;
        use crate::error::AdapterError;

        Catalog::with_debug(|mut catalog| async move {
            const DEPTH: usize = 100;
            const BASE: u64 = 1 << 40;
            insert_synthetic_view_chain(&mut catalog, BASE, DEPTH);

            // The chain has DEPTH + 1 distinct objects (root plus DEPTH links).
            const OBJECTS: usize = DEPTH + 1;

            // Exactly at the limit is allowed.
            validate_read_then_write_dependencies(&catalog, [CatalogItemId::User(BASE)], OBJECTS)
                .expect("chain at the limit is valid");

            // One below the limit is rejected with a clean error.
            let err = validate_read_then_write_dependencies(
                &catalog,
                [CatalogItemId::User(BASE)],
                OBJECTS - 1,
            )
            .expect_err("chain over the limit is rejected");
            assert!(matches!(
                err,
                AdapterError::ReadThenWriteDependencyLimitExceeded {
                    max_rw_dependencies
                } if max_rw_dependencies == OBJECTS - 1
            ));

            catalog.expire().await;
        })
        .await
    }

    /// Inserts a synthetic chain of `depth + 1` user views into `catalog` where
    /// view `base + i` reads from `base + i + 1`. Ids start well above any id
    /// the debug catalog assigns so they do not collide with real entries.
    ///
    /// Clones a builtin view as a template rather than constructing a `View` by
    /// hand. Read-then-write validation only reads the item type, `uses()`, and
    /// the optimized expression's temporal-ness, all of which a builtin view
    /// satisfies (user id + non-temporal).
    fn insert_synthetic_view_chain(catalog: &mut Catalog, base: u64, depth: usize) {
        use mz_ore::cast::CastFrom;
        use mz_sql::names::{DependencyIds, ResolvedIds};

        let template = catalog
            .state
            .entry_by_id
            .values()
            .find(|entry| matches!(entry.item(), CatalogItem::View(_)))
            .expect("debug catalog has builtin views")
            .clone();

        // `uses()` for a view unions `resolved_ids` and `dependencies`, so clear
        // both and point only at the next link.
        for i in 0..=depth {
            let id = CatalogItemId::User(base + u64::cast_from(i));
            let mut entry = template.clone();
            entry.id = id;
            entry.referenced_by = Vec::new();
            entry.used_by = Vec::new();
            let mut resolved_ids = ResolvedIds::empty();
            if i < depth {
                resolved_ids.add_item(CatalogItemId::User(base + u64::cast_from(i + 1)));
            }
            match &mut entry.item {
                CatalogItem::View(view) => {
                    view.resolved_ids = resolved_ids;
                    view.dependencies = DependencyIds(BTreeSet::new());
                }
                _ => unreachable!("template is a view"),
            }
            catalog.state.entry_by_id.insert(id, entry);
        }
    }
}