mz_adapter/

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

// TODO(jkosh44) Move to mz_catalog crate.

//! Persistent metadata storage for the coordinator.

use std::borrow::Cow;
use std::collections::{BTreeMap, BTreeSet};
use std::convert;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;

use futures::future::BoxFuture;
use futures::{Future, FutureExt};
use itertools::Itertools;
use mz_adapter_types::bootstrap_builtin_cluster_config::{
    ANALYTICS_CLUSTER_DEFAULT_REPLICATION_FACTOR, BootstrapBuiltinClusterConfig,
    CATALOG_SERVER_CLUSTER_DEFAULT_REPLICATION_FACTOR, PROBE_CLUSTER_DEFAULT_REPLICATION_FACTOR,
    SUPPORT_CLUSTER_DEFAULT_REPLICATION_FACTOR, SYSTEM_CLUSTER_DEFAULT_REPLICATION_FACTOR,
};
use mz_adapter_types::connection::ConnectionId;
use mz_audit_log::{EventType, FullNameV1, ObjectType, VersionedStorageUsage};
use mz_build_info::{BuildInfo, DUMMY_BUILD_INFO};
use mz_catalog::builtin::{
    BUILTIN_PREFIXES, BuiltinCluster, BuiltinLog, BuiltinSource, BuiltinTable,
    MZ_CATALOG_SERVER_CLUSTER,
};
use mz_catalog::config::{
    AwsPrincipalContext, BuiltinItemMigrationConfig, ClusterReplicaSizeMap, Config, StateConfig,
};
#[cfg(test)]
use mz_catalog::durable::CatalogError;
use mz_catalog::durable::{
    BootstrapArgs, DurableCatalogState, STORAGE_USAGE_ID_ALLOC_KEY, TestCatalogStateBuilder,
    test_bootstrap_args,
};
use mz_catalog::expr_cache::{ExpressionCacheHandle, GlobalExpressions, LocalExpressions};
use mz_catalog::memory::error::{Error, ErrorKind};
use mz_catalog::memory::objects::{
    CatalogCollectionEntry, CatalogEntry, CatalogItem, Cluster, ClusterReplica, Database,
    NetworkPolicy, Role, RoleAuth, Schema,
};
use mz_cloud_resources::AwsExternalIdPrefix;
use mz_compute_types::dataflows::DataflowDescription;
use mz_controller::clusters::ReplicaLocation;
use mz_controller_types::{ClusterId, ReplicaId};
use mz_expr::OptimizedMirRelationExpr;
use mz_license_keys::ValidatedLicenseKey;
use mz_ore::metrics::MetricsRegistry;
use mz_ore::now::{EpochMillis, NowFn, SYSTEM_TIME};
use mz_ore::result::ResultExt as _;
use mz_persist_client::PersistClient;
use mz_repr::adt::mz_acl_item::{AclMode, PrivilegeMap};
use mz_repr::explain::ExprHumanizer;
use mz_repr::network_policy_id::NetworkPolicyId;
use mz_repr::optimize::OptimizerFeatures;
use mz_repr::role_id::RoleId;
use mz_repr::{
    CatalogItemId, Diff, GlobalId, RelationVersion, RelationVersionSelector, SqlScalarType,
};
use mz_secrets::InMemorySecretsController;
use mz_sql::catalog::{
    CatalogCluster, CatalogClusterReplica, CatalogDatabase, CatalogError as SqlCatalogError,
    CatalogItem as SqlCatalogItem, CatalogItemType as SqlCatalogItemType, CatalogNetworkPolicy,
    CatalogRole, CatalogSchema, DefaultPrivilegeAclItem, DefaultPrivilegeObject, EnvironmentId,
    SessionCatalog, SystemObjectType,
};
use mz_sql::names::{
    CommentObjectId, DatabaseId, FullItemName, FullSchemaName, ItemQualifiers, ObjectId,
    PUBLIC_ROLE_NAME, PartialItemName, QualifiedItemName, QualifiedSchemaName,
    ResolvedDatabaseSpecifier, ResolvedIds, SchemaId, SchemaSpecifier, SystemObjectId,
};
use mz_sql::plan::{Plan, PlanNotice, StatementDesc};
use mz_sql::rbac;
use mz_sql::session::metadata::SessionMetadata;
use mz_sql::session::user::{MZ_SYSTEM_ROLE_ID, SUPPORT_USER, SYSTEM_USER};
use mz_sql::session::vars::SystemVars;
use mz_sql_parser::ast::QualifiedReplica;
use mz_storage_types::connections::ConnectionContext;
use mz_storage_types::connections::inline::{ConnectionResolver, InlinedConnection};
use mz_transform::dataflow::DataflowMetainfo;
use mz_transform::notice::OptimizerNotice;
use tokio::sync::MutexGuard;
use tokio::sync::mpsc::UnboundedSender;
use uuid::Uuid;

// DO NOT add any more imports from `crate` outside of `crate::catalog`.
pub use crate::catalog::builtin_table_updates::BuiltinTableUpdate;
pub use crate::catalog::open::{InitializeStateResult, OpenCatalogResult};
pub use crate::catalog::state::CatalogState;
pub(crate) use crate::catalog::transact::{CommandExecutionError, CommandWork};
pub use crate::catalog::transact::{
    DropObjectInfo, InjectedAuditEvent, Op, ReplicaCreateDropReason, TransactionResult,
};
use crate::command::CatalogDump;
use crate::coord::TargetCluster;
#[cfg(test)]
use crate::coord::catalog_implications::parsed_state_updates::ParsedStateUpdate;
use crate::session::{Portal, PreparedStatement, Session};
use crate::util::ResultExt;
use crate::{AdapterError, AdapterNotice, ExecuteResponse};

mod builtin_table_updates;
pub(crate) mod command;
pub(crate) mod consistency;
mod limits;
mod migrate;

mod apply;
pub(crate) mod cluster_state;
mod open;
mod state;
mod timeline;
mod transact;

/// A `Catalog` keeps track of the SQL objects known to the planner.
///
/// For each object, it keeps track of both forward and reverse dependencies:
/// i.e., which objects are depended upon by the object, and which objects
/// depend upon the object. It enforces the SQL rules around dropping: an object
/// cannot be dropped until all of the objects that depend upon it are dropped.
/// It also enforces uniqueness of names.
///
/// SQL mandates a hierarchy of exactly three layers. A catalog contains
/// databases, databases contain schemas, and schemas contain catalog items,
/// like sources, sinks, view, and indexes.
///
/// To the outside world, databases, schemas, and items are all identified by
/// name. Items can be referred to by their [`FullItemName`], which fully and
/// unambiguously specifies the item, or a [`PartialItemName`], which can omit the
/// database name and/or the schema name. Partial names can be converted into
/// full names via a complicated resolution process documented by the
/// [`CatalogState::resolve`] method.
///
/// The catalog also maintains special "ambient schemas": virtual schemas,
/// implicitly present in all databases, that house various system views.
/// The big examples of ambient schemas are `pg_catalog` and `mz_catalog`.
#[derive(Debug)]
pub struct Catalog {
    state: CatalogState,
    expr_cache_handle: Option<ExpressionCacheHandle>,
    storage: Arc<tokio::sync::Mutex<Box<dyn mz_catalog::durable::DurableCatalogState>>>,
    transient_revision: u64,
    /// The latest `transient_revision`, shared by all clones of this catalog.
    /// While `transient_revision` is this clone's own revision, frozen when
    /// the snapshot was taken, this field always tracks the latest revision
    /// across all clones. Comparing the two lets a snapshot holder detect
    /// from off-thread whether its snapshot is still current, via
    /// [`Catalog::transient_revision_is_current`], without a Coordinator
    /// round-trip (see `PeekClient::catalog_snapshot`).
    ///
    /// The store happens in `transact`, before the transaction's effects can
    /// be observed anywhere (responses, notices, builtin table writes), so a
    /// session that has observed any evidence of a catalog change is
    /// guaranteed to see the corresponding bump.
    shared_transient_revision: Arc<AtomicU64>,
}

/// A handle for advancing the durable catalog upper off the coordinator loop.
#[derive(Debug, Clone)]
pub struct CatalogUpperHandle {
    storage: Arc<tokio::sync::Mutex<Box<dyn mz_catalog::durable::DurableCatalogState>>>,
}

impl CatalogUpperHandle {
    /// Advances the durable catalog upper to at least `new_upper`.
    pub async fn advance_upper(
        &self,
        new_upper: mz_repr::Timestamp,
    ) -> Result<(), mz_catalog::durable::CatalogError> {
        self.storage.lock().await.advance_upper(new_upper).await
    }
}

// Implement our own Clone because derive can't unless S is Clone, which it's
// not (hence the Arc).
impl Clone for Catalog {
    fn clone(&self) -> Self {
        Self {
            state: self.state.clone(),
            expr_cache_handle: self.expr_cache_handle.clone(),
            storage: Arc::clone(&self.storage),
            transient_revision: self.transient_revision,
            shared_transient_revision: Arc::clone(&self.shared_transient_revision),
        }
    }
}

impl Catalog {
    /// Set the optimized plan for the item identified by `id`.
    ///
    /// # Panics
    /// If the item is not an `Index`, `MaterializedView`, or
    /// `ContinualTask`.
    #[mz_ore::instrument(level = "trace")]
    pub fn set_optimized_plan(
        &mut self,
        id: GlobalId,
        plan: DataflowDescription<OptimizedMirRelationExpr>,
    ) {
        self.state.set_optimized_plan(id, plan);
    }

    /// Set the physical plan for the item identified by `id`.
    ///
    /// # Panics
    /// If the item is not an `Index`, `MaterializedView`, or
    /// `ContinualTask`.
    #[mz_ore::instrument(level = "trace")]
    pub fn set_physical_plan(
        &mut self,
        id: GlobalId,
        plan: mz_compute_types::dataflows::LirDataflowDescription,
    ) {
        self.state.set_physical_plan(id, plan);
    }

    /// Try to get the optimized plan for the item identified by `id`.
    #[mz_ore::instrument(level = "trace")]
    pub fn try_get_optimized_plan(
        &self,
        id: &GlobalId,
    ) -> Option<&DataflowDescription<OptimizedMirRelationExpr>> {
        let entry = self.state.try_get_entry_by_global_id(id)?;
        entry.item().optimized_plan().map(AsRef::as_ref)
    }

    /// Try to get the physical plan for the item identified by `id`.
    #[mz_ore::instrument(level = "trace")]
    pub fn try_get_physical_plan(
        &self,
        id: &GlobalId,
    ) -> Option<&mz_compute_types::dataflows::LirDataflowDescription> {
        let entry = self.state.try_get_entry_by_global_id(id)?;
        entry.item().physical_plan().map(AsRef::as_ref)
    }

    /// Set the `DataflowMetainfo` for the item identified by `id`.
    ///
    /// # Panics
    /// If the item is not an `Index`, `MaterializedView`, or
    /// `ContinualTask`.
    #[mz_ore::instrument(level = "trace")]
    pub fn set_dataflow_metainfo(
        &mut self,
        id: GlobalId,
        metainfo: DataflowMetainfo<Arc<OptimizerNotice>>,
    ) {
        self.state.set_dataflow_metainfo(id, metainfo);
    }

    /// Try to get the `DataflowMetainfo` for the item identified by `id`.
    #[mz_ore::instrument(level = "trace")]
    pub fn try_get_dataflow_metainfo(
        &self,
        id: &GlobalId,
    ) -> Option<&DataflowMetainfo<Arc<OptimizerNotice>>> {
        let entry = self.state.try_get_entry_by_global_id(id)?;
        entry.item().dataflow_metainfo()
    }
}

#[derive(Debug)]
pub struct ConnCatalog<'a> {
    state: Cow<'a, CatalogState>,
    /// Because we don't have any way of removing items from the catalog
    /// temporarily, we allow the ConnCatalog to pretend that a set of items
    /// don't exist during resolution.
    ///
    /// This feature is necessary to allow re-planning of statements, which is
    /// either incredibly useful or required when altering item definitions.
    ///
    /// Note that uses of this field should be used by short-lived
    /// catalogs.
    unresolvable_ids: BTreeSet<CatalogItemId>,
    conn_id: ConnectionId,
    cluster: String,
    database: Option<DatabaseId>,
    search_path: Vec<(ResolvedDatabaseSpecifier, SchemaSpecifier)>,
    role_id: RoleId,
    prepared_statements: Option<&'a BTreeMap<String, PreparedStatement>>,
    portals: Option<&'a BTreeMap<String, Portal>>,
    notices_tx: UnboundedSender<AdapterNotice>,
    restrict_to_user_objects: bool,
}

impl ConnCatalog<'_> {
    pub fn conn_id(&self) -> &ConnectionId {
        &self.conn_id
    }

    pub fn state(&self) -> &CatalogState {
        &*self.state
    }

    /// Returns a system-role view of this catalog snapshot.
    pub fn for_system_session(&self) -> ConnCatalog<'_> {
        self.state.for_system_session()
    }

    /// Prevent planning from resolving item with the provided ID. Instead,
    /// return an error as if the item did not exist.
    ///
    /// This feature is meant exclusively to permit re-planning statements
    /// during update operations and should not be used otherwise given its
    /// extremely "powerful" semantics.
    ///
    /// # Panics
    /// If the catalog's role ID is not [`MZ_SYSTEM_ROLE_ID`].
    pub fn mark_id_unresolvable_for_replanning(&mut self, id: CatalogItemId) {
        assert_eq!(
            self.role_id, MZ_SYSTEM_ROLE_ID,
            "only the system role can mark IDs unresolvable",
        );
        self.unresolvable_ids.insert(id);
    }

    /// Returns the schemas:
    /// - mz_catalog
    /// - pg_catalog
    /// - temp (if requested)
    /// - all schemas from the session's search_path var that exist
    pub fn effective_search_path(
        &self,
        include_temp_schema: bool,
    ) -> Vec<(ResolvedDatabaseSpecifier, SchemaSpecifier)> {
        self.state
            .effective_search_path(&self.search_path, include_temp_schema)
    }
}

impl ConnectionResolver for ConnCatalog<'_> {
    fn resolve_connection(
        &self,
        id: CatalogItemId,
    ) -> mz_storage_types::connections::Connection<InlinedConnection> {
        self.state().resolve_connection(id)
    }
}

/// AWS environment context for a debug catalog.
///
/// The `mz_aws_connections` and `mz_aws_privatelink_connections` builtin
/// materialized views fold `mz_aws_account_id()`, `mz_aws_external_id_prefix()`,
/// and `mz_aws_connection_role_arn()` into their optimized expressions. A debug
/// catalog opened to be compared against a live environment (the testdrive
/// consistency check) must fold those functions to the same values the live
/// environment used, otherwise the optimized expressions diverge. This mirrors
/// how `environment_id` is threaded into the catalog copy. Fields left `None`
/// fold to SQL NULL, matching an environment without that context.
#[derive(Debug, Clone, Default)]
pub struct DebugAwsContext {
    pub aws_account_id: Option<String>,
    pub aws_external_id_prefix: Option<String>,
    pub aws_connection_role_arn: Option<String>,
}

impl Catalog {
    /// Returns the catalog's transient revision, which starts at 1 and is
    /// incremented on every change. This is not persisted to disk, and will
    /// restart on every load.
    pub fn transient_revision(&self) -> u64 {
        self.transient_revision
    }

    /// Reports whether this catalog's transient revision is still the latest,
    /// i.e., whether no catalog transaction has committed since this snapshot
    /// was taken. Can be called on a snapshot from off-thread, without a
    /// Coordinator round-trip. See the field documentation on
    /// `shared_transient_revision`.
    pub fn transient_revision_is_current(&self) -> bool {
        self.transient_revision
            == self
                .shared_transient_revision
                .load(std::sync::atomic::Ordering::SeqCst)
    }

    /// Creates a debug catalog from the current
    /// `METADATA_BACKEND_URL` with parameters set appropriately for debug contexts,
    /// like in tests.
    ///
    /// WARNING! This function can arbitrarily fail because it does not make any
    /// effort to adjust the catalog's contents' structure or semantics to the
    /// currently running version, i.e. it does not apply any migrations.
    ///
    /// This function must not be called in production contexts. Use
    /// [`Catalog::open`] with appropriately set configuration parameters
    /// instead.
    pub async fn with_debug<F, Fut, T>(f: F) -> T
    where
        F: FnOnce(Catalog) -> Fut,
        Fut: Future<Output = T>,
    {
        let persist_client = PersistClient::new_for_tests().await;
        let organization_id = Uuid::new_v4();
        let bootstrap_args = test_bootstrap_args();
        let catalog = Self::open_debug_catalog(persist_client, organization_id, &bootstrap_args)
            .await
            .expect("can open debug catalog");
        f(catalog).await
    }

    /// Like [`Catalog::with_debug`], but folds the given AWS context into the
    /// catalog. Used to exercise builtin materialized views that reproduce AWS
    /// environment context in SQL (see [`DebugAwsContext`]).
    pub async fn with_debug_aws_context<F, Fut, T>(aws_context: DebugAwsContext, f: F) -> T
    where
        F: FnOnce(Catalog) -> Fut,
        Fut: Future<Output = T>,
    {
        let persist_client = PersistClient::new_for_tests().await;
        let organization_id = Uuid::new_v4();
        let bootstrap_args = test_bootstrap_args();
        let catalog = Self::open_debug_catalog_with_aws_context(
            persist_client,
            organization_id,
            &bootstrap_args,
            Some(aws_context),
        )
        .await
        .expect("can open debug catalog");
        f(catalog).await
    }

    /// Like [`Catalog::with_debug`], but the catalog created believes that bootstrap is still
    /// in progress.
    pub async fn with_debug_in_bootstrap<F, Fut, T>(f: F) -> T
    where
        F: FnOnce(Catalog) -> Fut,
        Fut: Future<Output = T>,
    {
        let persist_client = PersistClient::new_for_tests().await;
        let organization_id = Uuid::new_v4();
        let bootstrap_args = test_bootstrap_args();
        let mut catalog =
            Self::open_debug_catalog(persist_client.clone(), organization_id, &bootstrap_args)
                .await
                .expect("can open debug catalog");

        // Replace `storage` in `catalog` with one that doesn't think bootstrap is over.
        let now = SYSTEM_TIME.clone();
        let openable_storage = TestCatalogStateBuilder::new(persist_client)
            .with_organization_id(organization_id)
            .with_default_deploy_generation()
            .build()
            .await
            .expect("can create durable catalog");
        let mut storage = openable_storage
            .open(now().into(), &bootstrap_args)
            .await
            .expect("can open durable catalog");
        // Drain updates.
        let _ = storage
            .sync_to_current_updates()
            .await
            .expect("can sync to current updates");
        catalog.storage = Arc::new(tokio::sync::Mutex::new(storage));

        f(catalog).await
    }

    /// Opens a debug catalog.
    ///
    /// See [`Catalog::with_debug`].
    pub async fn open_debug_catalog(
        persist_client: PersistClient,
        organization_id: Uuid,
        bootstrap_args: &BootstrapArgs,
    ) -> Result<Catalog, anyhow::Error> {
        let now = SYSTEM_TIME.clone();
        let environment_id = None;
        let openable_storage = TestCatalogStateBuilder::new(persist_client.clone())
            .with_organization_id(organization_id)
            .with_default_deploy_generation()
            .build()
            .await?;
        let storage = openable_storage.open(now().into(), bootstrap_args).await?;
        let system_parameter_defaults = BTreeMap::default();
        Self::open_debug_catalog_inner(
            persist_client,
            storage,
            now,
            environment_id,
            &DUMMY_BUILD_INFO,
            system_parameter_defaults,
            bootstrap_args,
            None,
            None,
        )
        .await
    }

    /// Like [`Catalog::open_debug_catalog`], but folds `aws_context` into the
    /// catalog (see [`DebugAwsContext`]).
    ///
    /// Calls `open_debug_catalog_inner` directly rather than delegating through
    /// [`Catalog::open_debug_catalog`], to avoid adding an `async fn` layer to
    /// the returned future. The catalog-open future is close to the crate
    /// `recursion_limit`, and the extra nesting overflows it when computing the
    /// layout of callers such as the `catalog` benchmark.
    pub async fn open_debug_catalog_with_aws_context(
        persist_client: PersistClient,
        organization_id: Uuid,
        bootstrap_args: &BootstrapArgs,
        aws_context: Option<DebugAwsContext>,
    ) -> Result<Catalog, anyhow::Error> {
        let now = SYSTEM_TIME.clone();
        let environment_id = None;
        let openable_storage = TestCatalogStateBuilder::new(persist_client.clone())
            .with_organization_id(organization_id)
            .with_default_deploy_generation()
            .build()
            .await?;
        let storage = openable_storage.open(now().into(), bootstrap_args).await?;
        let system_parameter_defaults = BTreeMap::default();
        Self::open_debug_catalog_inner(
            persist_client,
            storage,
            now,
            environment_id,
            &DUMMY_BUILD_INFO,
            system_parameter_defaults,
            bootstrap_args,
            None,
            aws_context,
        )
        .await
    }

    /// Opens a read only debug persist backed catalog defined by `persist_client` and
    /// `organization_id`.
    ///
    /// See [`Catalog::with_debug`].
    pub async fn open_debug_read_only_catalog(
        persist_client: PersistClient,
        organization_id: Uuid,
        bootstrap_args: &BootstrapArgs,
    ) -> Result<Catalog, anyhow::Error> {
        let now = SYSTEM_TIME.clone();
        let environment_id = None;
        let openable_storage = TestCatalogStateBuilder::new(persist_client.clone())
            .with_organization_id(organization_id)
            .build()
            .await?;
        let storage = openable_storage
            .open_read_only(&test_bootstrap_args())
            .await?;
        let system_parameter_defaults = BTreeMap::default();
        Self::open_debug_catalog_inner(
            persist_client,
            storage,
            now,
            environment_id,
            &DUMMY_BUILD_INFO,
            system_parameter_defaults,
            bootstrap_args,
            None,
            None,
        )
        .await
    }

    /// Opens a read only debug persist backed catalog defined by `persist_client` and
    /// `organization_id`.
    ///
    /// See [`Catalog::with_debug`].
    pub async fn open_debug_read_only_persist_catalog_config(
        persist_client: PersistClient,
        now: NowFn,
        environment_id: EnvironmentId,
        system_parameter_defaults: BTreeMap<String, String>,
        build_info: &'static BuildInfo,
        bootstrap_args: &BootstrapArgs,
        enable_expression_cache_override: Option<bool>,
        aws_context: Option<DebugAwsContext>,
    ) -> Result<Catalog, anyhow::Error> {
        let openable_storage = TestCatalogStateBuilder::new(persist_client.clone())
            .with_organization_id(environment_id.organization_id())
            .with_version(
                build_info
                    .version
                    .parse()
                    .expect("build version is parseable"),
            )
            .build()
            .await?;
        let storage = openable_storage.open_read_only(bootstrap_args).await?;
        Self::open_debug_catalog_inner(
            persist_client,
            storage,
            now,
            Some(environment_id),
            build_info,
            system_parameter_defaults,
            bootstrap_args,
            enable_expression_cache_override,
            aws_context,
        )
        .await
    }

    async fn open_debug_catalog_inner(
        persist_client: PersistClient,
        storage: Box<dyn DurableCatalogState>,
        now: NowFn,
        environment_id: Option<EnvironmentId>,
        build_info: &'static BuildInfo,
        system_parameter_defaults: BTreeMap<String, String>,
        bootstrap_args: &BootstrapArgs,
        enable_expression_cache_override: Option<bool>,
        aws_context: Option<DebugAwsContext>,
    ) -> Result<Catalog, anyhow::Error> {
        let metrics_registry = &MetricsRegistry::new();
        let secrets_reader = Arc::new(InMemorySecretsController::new());
        // Used as a lower boundary of the boot_ts, but it's ok to use now() for
        // debugging/testing.
        let previous_ts = now().into();
        let replica_size = &bootstrap_args.default_cluster_replica_size;
        let read_only = false;

        // Fold the requested AWS context into the connection context and
        // principal context. The builtin AWS connection views reproduce these
        // values in SQL, so a catalog compared against a live environment must
        // resolve them identically. When `aws_context` is `None` the default
        // `for_tests` context is used unchanged.
        let mut connection_context = ConnectionContext::for_tests(secrets_reader);
        let aws_principal_context = match aws_context {
            None => None,
            Some(aws_context) => {
                connection_context.aws_external_id_prefix =
                    aws_context.aws_external_id_prefix.as_deref().map(|prefix| {
                        AwsExternalIdPrefix::new_from_cli_argument_or_environment_variable(prefix)
                            .expect("infallible")
                    });
                connection_context.aws_connection_role_arn = aws_context.aws_connection_role_arn;
                aws_context.aws_account_id.map(|aws_account_id| {
                    AwsPrincipalContext {
                        aws_account_id,
                        // Only `aws_account_id` reaches the `mz_aws_account_id()`
                        // fold. The external id prefix is read from the connection
                        // context above, so any placeholder works here.
                        aws_external_id_prefix:
                            AwsExternalIdPrefix::new_from_cli_argument_or_environment_variable(
                                "debug",
                            )
                            .expect("infallible"),
                    }
                })
            }
        };

        let OpenCatalogResult {
            catalog,
            last_seen_version: _,
            migrated_storage_collections_0dt: _,
            new_builtin_collections: _,
            builtin_table_updates: _,
            cached_global_exprs: _,
            uncached_local_exprs: _,
        } = Catalog::open(Config {
            storage,
            metrics_registry,
            state: StateConfig {
                unsafe_mode: true,
                all_features: false,
                build_info,
                environment_id: environment_id.unwrap_or_else(EnvironmentId::for_tests),
                read_only,
                now,
                boot_ts: previous_ts,
                skip_migrations: true,
                cluster_replica_sizes: bootstrap_args.cluster_replica_size_map.clone(),
                builtin_system_cluster_config: BootstrapBuiltinClusterConfig {
                    size: replica_size.clone(),
                    replication_factor: SYSTEM_CLUSTER_DEFAULT_REPLICATION_FACTOR,
                },
                builtin_catalog_server_cluster_config: BootstrapBuiltinClusterConfig {
                    size: replica_size.clone(),
                    replication_factor: CATALOG_SERVER_CLUSTER_DEFAULT_REPLICATION_FACTOR,
                },
                builtin_probe_cluster_config: BootstrapBuiltinClusterConfig {
                    size: replica_size.clone(),
                    replication_factor: PROBE_CLUSTER_DEFAULT_REPLICATION_FACTOR,
                },
                builtin_support_cluster_config: BootstrapBuiltinClusterConfig {
                    size: replica_size.clone(),
                    replication_factor: SUPPORT_CLUSTER_DEFAULT_REPLICATION_FACTOR,
                },
                builtin_analytics_cluster_config: BootstrapBuiltinClusterConfig {
                    size: replica_size.clone(),
                    replication_factor: ANALYTICS_CLUSTER_DEFAULT_REPLICATION_FACTOR,
                },
                system_parameter_defaults,
                remote_system_parameters: None,
                availability_zones: vec![],
                egress_addresses: vec![],
                aws_principal_context,
                aws_privatelink_availability_zones: None,
                http_host_name: None,
                connection_context,
                builtin_item_migration_config: BuiltinItemMigrationConfig {
                    persist_client: persist_client.clone(),
                    read_only,
                    force_migration: None,
                },
                persist_client,
                enable_expression_cache_override,
                helm_chart_version: None,
                external_login_password_mz_system: None,
                license_key: ValidatedLicenseKey::for_tests(),
            },
        })
        .await?;
        Ok(catalog)
    }

    pub fn for_session<'a>(&'a self, session: &'a Session) -> ConnCatalog<'a> {
        self.state.for_session(session)
    }

    pub fn for_sessionless_user(&self, role_id: RoleId) -> ConnCatalog<'_> {
        self.state.for_sessionless_user(role_id)
    }

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

    async fn storage<'a>(
        &'a self,
    ) -> MutexGuard<'a, Box<dyn mz_catalog::durable::DurableCatalogState>> {
        self.storage.lock().await
    }

    pub async fn current_upper(&self) -> mz_repr::Timestamp {
        self.storage().await.current_upper().await
    }

    /// Allocates and returns both a user [`CatalogItemId`] and [`GlobalId`], delegating to
    /// [`DurableCatalogState::allocate_user_id`].
    pub async fn allocate_user_id(
        &self,
        commit_ts: mz_repr::Timestamp,
    ) -> Result<(CatalogItemId, GlobalId), Error> {
        Ok(self
            .storage()
            .await
            .allocate_user_id(commit_ts)
            .await
            .maybe_terminate("allocating user ids")?)
    }

    /// Allocate `amount` many user IDs. See [`DurableCatalogState::allocate_user_ids`].
    pub async fn allocate_user_ids(
        &self,
        amount: u64,
        commit_ts: mz_repr::Timestamp,
    ) -> Result<Vec<(CatalogItemId, GlobalId)>, Error> {
        Ok(self
            .storage()
            .await
            .allocate_user_ids(amount, commit_ts)
            .await
            .maybe_terminate("allocating user ids")?)
    }

    pub async fn allocate_user_id_for_test(&self) -> Result<(CatalogItemId, GlobalId), Error> {
        let commit_ts = self.storage().await.current_upper().await;
        self.allocate_user_id(commit_ts).await
    }

    /// Allocates a single durable id for a storage usage collection batch.
    ///
    /// Bumps the durable `STORAGE_USAGE_ID_ALLOC_KEY` allocator by one and
    /// returns the previous value. The bump is committed at `commit_ts`.
    /// One id is shared by every row produced by a collection cycle (see
    /// `Coordinator::storage_usage_update`), so the durable cost is one
    /// allocator round-trip per cycle, not per shard.
    pub async fn allocate_storage_usage_id(
        &self,
        commit_ts: mz_repr::Timestamp,
    ) -> Result<u64, Error> {
        use mz_ore::collections::CollectionExt;

        self.storage()
            .await
            .allocate_id(STORAGE_USAGE_ID_ALLOC_KEY, 1, commit_ts)
            .await
            .maybe_terminate("allocating storage usage id")
            .map(|ids| ids.into_element())
            .err_into()
    }

    /// Get the next user item ID without allocating it.
    pub async fn get_next_user_item_id(&self) -> Result<u64, Error> {
        self.storage()
            .await
            .get_next_user_item_id()
            .await
            .err_into()
    }

    #[cfg(test)]
    pub async fn allocate_system_id(
        &self,
        commit_ts: mz_repr::Timestamp,
    ) -> Result<(CatalogItemId, GlobalId), Error> {
        use mz_ore::collections::CollectionExt;

        let mut storage = self.storage().await;
        let mut txn = storage.transaction().await?;
        let id = txn
            .allocate_system_item_ids(1)
            .maybe_terminate("allocating system ids")?
            .into_element();
        // Drain transaction.
        let _ = txn.get_and_commit_op_updates();
        txn.commit(commit_ts).await?;
        Ok(id)
    }

    /// Get the next system item ID without allocating it.
    pub async fn get_next_system_item_id(&self) -> Result<u64, Error> {
        self.storage()
            .await
            .get_next_system_item_id()
            .await
            .err_into()
    }

    /// Allocates and returns a user [`ClusterId`], delegating to
    /// [`DurableCatalogState::allocate_user_cluster_id`].
    pub async fn allocate_user_cluster_id(
        &self,
        commit_ts: mz_repr::Timestamp,
    ) -> Result<ClusterId, Error> {
        Ok(self
            .storage()
            .await
            .allocate_user_cluster_id(commit_ts)
            .await
            .maybe_terminate("allocating user cluster ids")?)
    }

    /// Allocate `amount` many user replica IDs. See
    /// [`DurableCatalogState::allocate_user_replica_ids`].
    pub async fn allocate_user_replica_ids(
        &self,
        amount: u64,
        commit_ts: mz_repr::Timestamp,
    ) -> Result<Vec<ReplicaId>, Error> {
        Ok(self
            .storage()
            .await
            .allocate_user_replica_ids(amount, commit_ts)
            .await
            .maybe_terminate("allocating user replica ids")?)
    }

    /// Allocate `amount` many system replica IDs. See
    /// [`DurableCatalogState::allocate_system_replica_ids`].
    pub async fn allocate_system_replica_ids(
        &self,
        amount: u64,
        commit_ts: mz_repr::Timestamp,
    ) -> Result<Vec<ReplicaId>, Error> {
        Ok(self
            .storage()
            .await
            .allocate_system_replica_ids(amount, commit_ts)
            .await
            .maybe_terminate("allocating system replica ids")?)
    }

    /// Allocate `amount` many replica IDs for `cluster_id`, picking user or
    /// system IDs based on the cluster's ID type.
    pub async fn allocate_replica_ids(
        &self,
        cluster_id: ClusterId,
        amount: u64,
        commit_ts: mz_repr::Timestamp,
    ) -> Result<Vec<ReplicaId>, Error> {
        if cluster_id.is_system() {
            self.allocate_system_replica_ids(amount, commit_ts).await
        } else {
            self.allocate_user_replica_ids(amount, commit_ts).await
        }
    }

    /// Get the next system replica id without allocating it.
    pub async fn get_next_system_replica_id(&self) -> Result<u64, Error> {
        self.storage()
            .await
            .get_next_system_replica_id()
            .await
            .err_into()
    }

    /// Get the next user replica id without allocating it.
    pub async fn get_next_user_replica_id(&self) -> Result<u64, Error> {
        self.storage()
            .await
            .get_next_user_replica_id()
            .await
            .err_into()
    }

    pub fn resolve_database(&self, database_name: &str) -> Result<&Database, SqlCatalogError> {
        self.state.resolve_database(database_name)
    }

    pub fn resolve_schema(
        &self,
        current_database: Option<&DatabaseId>,
        database_name: Option<&str>,
        schema_name: &str,
        conn_id: &ConnectionId,
    ) -> Result<&Schema, SqlCatalogError> {
        self.state
            .resolve_schema(current_database, database_name, schema_name, conn_id)
    }

    pub fn resolve_schema_in_database(
        &self,
        database_spec: &ResolvedDatabaseSpecifier,
        schema_name: &str,
        conn_id: &ConnectionId,
    ) -> Result<&Schema, SqlCatalogError> {
        self.state
            .resolve_schema_in_database(database_spec, schema_name, conn_id)
    }

    pub fn resolve_replica_in_cluster(
        &self,
        cluster_id: &ClusterId,
        replica_name: &str,
    ) -> Result<&ClusterReplica, SqlCatalogError> {
        self.state
            .resolve_replica_in_cluster(cluster_id, replica_name)
    }

    pub fn resolve_system_schema(&self, name: &'static str) -> SchemaId {
        self.state.resolve_system_schema(name)
    }

    pub fn resolve_search_path(
        &self,
        session: &Session,
    ) -> Vec<(ResolvedDatabaseSpecifier, SchemaSpecifier)> {
        self.state.resolve_search_path(session)
    }

    /// 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.state
            .resolve_entry(current_database, search_path, name, conn_id)
    }

    /// Resolves a `BuiltinTable`.
    pub fn resolve_builtin_table(&self, builtin: &'static BuiltinTable) -> CatalogItemId {
        self.state.resolve_builtin_table(builtin)
    }

    /// Resolves a `BuiltinLog`.
    pub fn resolve_builtin_log(&self, builtin: &'static BuiltinLog) -> CatalogItemId {
        self.state.resolve_builtin_log(builtin).0
    }

    /// Resolves a `BuiltinSource`.
    pub fn resolve_builtin_storage_collection(
        &self,
        builtin: &'static BuiltinSource,
    ) -> CatalogItemId {
        self.state.resolve_builtin_source(builtin)
    }

    /// 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.state
            .resolve_function(current_database, search_path, name, conn_id)
    }

    /// 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> {
        self.state
            .resolve_type(current_database, search_path, name, conn_id)
    }

    pub fn resolve_cluster(&self, name: &str) -> Result<&Cluster, SqlCatalogError> {
        self.state.resolve_cluster(name)
    }

    /// Resolves a [`Cluster`] for a [`BuiltinCluster`].
    ///
    /// # Panics
    /// * If the [`BuiltinCluster`] doesn't exist.
    ///
    pub fn resolve_builtin_cluster(&self, cluster: &BuiltinCluster) -> &Cluster {
        self.state.resolve_builtin_cluster(cluster)
    }

    pub fn get_mz_catalog_server_cluster_id(&self) -> &ClusterId {
        &self.resolve_builtin_cluster(&MZ_CATALOG_SERVER_CLUSTER).id
    }

    /// Resolves a [`Cluster`] for a TargetCluster.
    pub fn resolve_target_cluster(
        &self,
        target_cluster: TargetCluster,
        session: &Session,
    ) -> Result<&Cluster, AdapterError> {
        match target_cluster {
            TargetCluster::CatalogServer => {
                Ok(self.resolve_builtin_cluster(&MZ_CATALOG_SERVER_CLUSTER))
            }
            TargetCluster::Active => self.active_cluster(session),
            TargetCluster::Transaction(cluster_id) => self
                .try_get_cluster(cluster_id)
                .ok_or(AdapterError::ConcurrentClusterDrop),
        }
    }

    pub fn active_cluster(&self, session: &Session) -> Result<&Cluster, AdapterError> {
        // TODO(benesch): this check here is not sufficiently protective. It'd
        // be very easy for a code path to accidentally avoid this check by
        // calling `resolve_cluster(session.vars().cluster())`.
        if session.user().name != SYSTEM_USER.name
            && session.user().name != SUPPORT_USER.name
            && session.vars().cluster() == SYSTEM_USER.name
        {
            coord_bail!(
                "system cluster '{}' cannot execute user queries",
                SYSTEM_USER.name
            );
        }
        let cluster = self.resolve_cluster(session.vars().cluster())?;
        Ok(cluster)
    }

    pub fn state(&self) -> &CatalogState {
        &self.state
    }

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

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

    pub fn try_get_entry_by_global_id(&self, id: &GlobalId) -> Option<&CatalogEntry> {
        self.state.try_get_entry_by_global_id(id)
    }

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

    pub fn get_entry_by_global_id(&self, id: &GlobalId) -> CatalogCollectionEntry {
        self.state.get_entry_by_global_id(id)
    }

    pub fn get_global_ids<'a>(
        &'a self,
        id: &CatalogItemId,
    ) -> impl Iterator<Item = GlobalId> + use<'a> {
        self.get_entry(id).global_ids()
    }

    pub fn resolve_item_id(&self, id: &GlobalId) -> CatalogItemId {
        self.get_entry_by_global_id(id).id()
    }

    pub fn try_resolve_item_id(&self, id: &GlobalId) -> Option<CatalogItemId> {
        let item = self.try_get_entry_by_global_id(id)?;
        Some(item.id())
    }

    pub fn get_schema(
        &self,
        database_spec: &ResolvedDatabaseSpecifier,
        schema_spec: &SchemaSpecifier,
        conn_id: &ConnectionId,
    ) -> &Schema {
        self.state.get_schema(database_spec, schema_spec, conn_id)
    }

    pub fn try_get_schema(
        &self,
        database_spec: &ResolvedDatabaseSpecifier,
        schema_spec: &SchemaSpecifier,
        conn_id: &ConnectionId,
    ) -> Option<&Schema> {
        self.state
            .try_get_schema(database_spec, schema_spec, conn_id)
    }

    pub fn get_mz_catalog_schema_id(&self) -> SchemaId {
        self.state.get_mz_catalog_schema_id()
    }

    pub fn get_pg_catalog_schema_id(&self) -> SchemaId {
        self.state.get_pg_catalog_schema_id()
    }

    pub fn get_information_schema_id(&self) -> SchemaId {
        self.state.get_information_schema_id()
    }

    pub fn get_mz_internal_schema_id(&self) -> SchemaId {
        self.state.get_mz_internal_schema_id()
    }

    pub fn get_mz_introspection_schema_id(&self) -> SchemaId {
        self.state.get_mz_introspection_schema_id()
    }

    pub fn get_mz_unsafe_schema_id(&self) -> SchemaId {
        self.state.get_mz_unsafe_schema_id()
    }

    pub fn system_schema_ids(&self) -> impl Iterator<Item = SchemaId> + '_ {
        self.state.system_schema_ids()
    }

    pub fn get_database(&self, id: &DatabaseId) -> &Database {
        self.state.get_database(id)
    }

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

    pub fn get_role(&self, id: &RoleId) -> &Role {
        self.state.get_role(id)
    }

    pub fn try_get_role_by_name(&self, role_name: &str) -> Option<&Role> {
        self.state.try_get_role_by_name(role_name)
    }

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

    /// Registers the connection's temporary namespace: the `uuid` <->
    /// `conn_id` mapping used to stamp and apply durable temporary items
    /// owned by the session. The `mz_temp` schema itself is created when
    /// the first temporary item is applied.
    ///
    /// The coordinator calls this at a session's first temporary-item
    /// creation, strictly before the transaction that persists the item, and
    /// guards on [`CatalogState::has_temporary_namespace`], so registering
    /// an already-registered namespace is a bug.
    pub fn register_temporary_namespace(&mut self, conn_id: &ConnectionId, uuid: Uuid) {
        self.state
            .temporary_namespaces
            .register(conn_id.clone(), uuid);
    }

    fn item_exists_in_temp_schemas(&self, conn_id: &ConnectionId, item_name: &str) -> bool {
        // A temporary namespace is registered at the connection's first
        // temporary-item creation, so it's valid for one to not exist yet.
        self.state
            .temporary_namespaces
            .schema(conn_id)
            .map(|schema| schema.items.contains_key(item_name))
            .unwrap_or(false)
    }

    /// Removes the connection's temporary namespace, if it has one.
    pub fn drop_temporary_namespace(&mut self, conn_id: &ConnectionId) {
        self.state.temporary_namespaces.unregister(conn_id)
    }

    pub(crate) fn object_dependents(
        &self,
        object_ids: &Vec<ObjectId>,
        conn_id: &ConnectionId,
    ) -> Vec<ObjectId> {
        let mut seen = BTreeSet::new();
        self.state.object_dependents(object_ids, conn_id, &mut seen)
    }

    fn full_name_detail(name: &FullItemName) -> FullNameV1 {
        FullNameV1 {
            database: name.database.to_string(),
            schema: name.schema.clone(),
            item: name.item.clone(),
        }
    }

    pub fn find_available_cluster_name(&self, name: &str) -> String {
        let mut i = 0;
        let mut candidate = name.to_string();
        while self.state.clusters_by_name.contains_key(&candidate) {
            i += 1;
            candidate = format!("{}{}", name, i);
        }
        candidate
    }

    pub fn get_role_allowed_cluster_sizes(&self, role_id: &Option<RoleId>) -> Vec<String> {
        // Disabled sizes are included; `ensure_valid_replica_size` decides
        // whether those are allowed.
        if role_id == &Some(MZ_SYSTEM_ROLE_ID) {
            self.cluster_replica_sizes().0.keys().cloned().collect()
        } else {
            self.system_config().allowed_cluster_replica_sizes()
        }
    }

    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> {
        self.state.concretize_replica_location(
            location,
            allowed_sizes,
            allowed_availability_zones,
            allow_disabled,
        )
    }

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

    pub fn cluster_replica_sizes(&self) -> &ClusterReplicaSizeMap {
        self.state.cluster_replica_sizes()
    }

    /// Returns the privileges of an object by its ID.
    pub fn get_privileges(
        &self,
        id: &SystemObjectId,
        conn_id: &ConnectionId,
    ) -> Option<&PrivilegeMap> {
        match id {
            SystemObjectId::Object(id) => match id {
                ObjectId::Cluster(id) => Some(self.get_cluster(*id).privileges()),
                ObjectId::Database(id) => Some(self.get_database(id).privileges()),
                ObjectId::Schema((database_spec, schema_spec)) => Some(
                    self.get_schema(database_spec, schema_spec, conn_id)
                        .privileges(),
                ),
                ObjectId::Item(id) => Some(self.get_entry(id).privileges()),
                ObjectId::ClusterReplica(_) | ObjectId::Role(_) => None,
                ObjectId::NetworkPolicy(id) => Some(self.get_network_policy(*id).privileges()),
            },
            SystemObjectId::System => Some(&self.state.system_privileges),
        }
    }

    /// Advances the catalog upper to at least `new_upper`.
    ///
    /// Empty progress can overtake `new_upper`. A durable upper mismatch with content returns
    /// `CatalogOutOfSync`. See [`mz_catalog::durable::DurableCatalogState::advance_upper`] for
    /// fencing semantics.
    #[mz_ore::instrument(level = "debug")]
    pub async fn advance_upper(&self, new_upper: mz_repr::Timestamp) -> Result<(), AdapterError> {
        Ok(self.storage().await.advance_upper(new_upper).await?)
    }

    /// Returns a durable-upper handle that shares the catalog storage mutex.
    pub fn upper_handle(&self) -> CatalogUpperHandle {
        CatalogUpperHandle {
            storage: Arc::clone(&self.storage),
        }
    }

    /// Return the ids of all log sources the given object depends on.
    pub fn introspection_dependencies(&self, id: CatalogItemId) -> Vec<CatalogItemId> {
        self.state.introspection_dependencies(id)
    }

    /// 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.
    pub fn dump(&self) -> Result<CatalogDump, Error> {
        Ok(CatalogDump::new(self.state.dump(None)?))
    }

    /// Checks the [`Catalog`]s internal consistency.
    ///
    /// Returns a JSON object describing the inconsistencies, if there are any.
    pub fn check_consistency(&self) -> Result<(), serde_json::Value> {
        self.state.check_consistency().map_err(|inconsistencies| {
            serde_json::to_value(inconsistencies).unwrap_or_else(|_| {
                serde_json::Value::String("failed to serialize inconsistencies".to_string())
            })
        })
    }

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

    pub fn entries(&self) -> impl Iterator<Item = &CatalogEntry> {
        self.state.entries()
    }

    pub fn user_connections(&self) -> impl Iterator<Item = &CatalogEntry> {
        self.state.user_connections()
    }

    pub fn user_tables(&self) -> impl Iterator<Item = &CatalogEntry> {
        self.state.user_tables()
    }

    pub fn user_sources(&self) -> impl Iterator<Item = &CatalogEntry> {
        self.state.user_sources()
    }

    pub fn user_sinks(&self) -> impl Iterator<Item = &CatalogEntry> {
        self.state.user_sinks()
    }

    pub fn user_materialized_views(&self) -> impl Iterator<Item = &CatalogEntry> {
        self.state.user_materialized_views()
    }

    pub fn user_secrets(&self) -> impl Iterator<Item = &CatalogEntry> {
        self.state.user_secrets()
    }

    pub fn get_network_policy(&self, network_policy_id: NetworkPolicyId) -> &NetworkPolicy {
        self.state.get_network_policy(&network_policy_id)
    }

    pub fn get_network_policy_by_name(&self, name: &str) -> Option<&NetworkPolicy> {
        self.state.try_get_network_policy_by_name(name)
    }

    pub fn clusters(&self) -> impl Iterator<Item = &Cluster> {
        self.state.clusters()
    }

    pub fn get_cluster(&self, cluster_id: ClusterId) -> &Cluster {
        self.state.get_cluster(cluster_id)
    }

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

    pub fn user_clusters(&self) -> impl Iterator<Item = &Cluster> {
        self.state.user_clusters()
    }

    pub fn get_cluster_replica(
        &self,
        cluster_id: ClusterId,
        replica_id: ReplicaId,
    ) -> &ClusterReplica {
        self.state.get_cluster_replica(cluster_id, replica_id)
    }

    pub fn try_get_cluster_replica(
        &self,
        cluster_id: ClusterId,
        replica_id: ReplicaId,
    ) -> Option<&ClusterReplica> {
        self.state.try_get_cluster_replica(cluster_id, replica_id)
    }

    pub fn user_cluster_replicas(&self) -> impl Iterator<Item = &ClusterReplica> {
        self.state.user_cluster_replicas()
    }

    pub fn databases(&self) -> impl Iterator<Item = &Database> {
        self.state.databases()
    }

    pub fn user_roles(&self) -> impl Iterator<Item = &Role> {
        self.state.user_roles()
    }

    pub fn user_network_policies(&self) -> impl Iterator<Item = &NetworkPolicy> {
        self.state.user_network_policies()
    }

    pub fn system_privileges(&self) -> &PrivilegeMap {
        &self.state.system_privileges
    }

    pub fn default_privileges(
        &self,
    ) -> impl Iterator<
        Item = (
            &DefaultPrivilegeObject,
            impl Iterator<Item = &DefaultPrivilegeAclItem>,
        ),
    > {
        self.state.default_privileges.iter()
    }

    pub fn pack_item_update(&self, id: CatalogItemId, diff: Diff) -> Vec<BuiltinTableUpdate> {
        self.state
            .resolve_builtin_table_updates(self.state.pack_item_update(id, diff))
    }

    pub fn pack_storage_usage_update(
        &self,
        event: VersionedStorageUsage,
        diff: Diff,
    ) -> BuiltinTableUpdate {
        self.state
            .resolve_builtin_table_update(self.state.pack_storage_usage_update(event, diff))
    }

    pub fn system_config(&self) -> &SystemVars {
        self.state.system_config()
    }

    pub fn system_config_mut(&mut self) -> &mut SystemVars {
        self.state.system_config_mut()
    }

    pub fn ensure_not_reserved_role(&self, role_id: &RoleId) -> Result<(), Error> {
        self.state.ensure_not_reserved_role(role_id)
    }

    pub fn ensure_grantable_role(&self, role_id: &RoleId) -> Result<(), Error> {
        self.state.ensure_grantable_role(role_id)
    }

    pub fn ensure_not_system_role(&self, role_id: &RoleId) -> Result<(), Error> {
        self.state.ensure_not_system_role(role_id)
    }

    pub fn ensure_not_predefined_role(&self, role_id: &RoleId) -> Result<(), Error> {
        self.state.ensure_not_predefined_role(role_id)
    }

    pub fn ensure_not_reserved_network_policy(
        &self,
        network_policy_id: &NetworkPolicyId,
    ) -> Result<(), Error> {
        self.state
            .ensure_not_reserved_network_policy(network_policy_id)
    }

    pub fn ensure_not_reserved_object(
        &self,
        object_id: &ObjectId,
        conn_id: &ConnectionId,
    ) -> Result<(), Error> {
        match object_id {
            ObjectId::Cluster(cluster_id) => {
                if cluster_id.is_system() {
                    let cluster = self.get_cluster(*cluster_id);
                    Err(Error::new(ErrorKind::ReadOnlyCluster(
                        cluster.name().to_string(),
                    )))
                } else {
                    Ok(())
                }
            }
            ObjectId::ClusterReplica((cluster_id, replica_id)) => {
                if replica_id.is_system() {
                    let replica = self.get_cluster_replica(*cluster_id, *replica_id);
                    Err(Error::new(ErrorKind::ReadOnlyClusterReplica(
                        replica.name().to_string(),
                    )))
                } else {
                    Ok(())
                }
            }
            ObjectId::Database(database_id) => {
                if database_id.is_system() {
                    let database = self.get_database(database_id);
                    Err(Error::new(ErrorKind::ReadOnlyDatabase(
                        database.name().to_string(),
                    )))
                } else {
                    Ok(())
                }
            }
            ObjectId::Schema((database_spec, schema_spec)) => {
                if schema_spec.is_system() {
                    let schema = self.get_schema(database_spec, schema_spec, conn_id);
                    Err(Error::new(ErrorKind::ReadOnlySystemSchema(
                        schema.name().schema.clone(),
                    )))
                } else {
                    Ok(())
                }
            }
            ObjectId::Role(role_id) => self.ensure_not_reserved_role(role_id),
            ObjectId::Item(item_id) => {
                if item_id.is_system() {
                    let item = self.get_entry(item_id);
                    let name = self.resolve_full_name(item.name(), Some(conn_id));
                    Err(Error::new(ErrorKind::ReadOnlyItem(name.to_string())))
                } else {
                    Ok(())
                }
            }
            ObjectId::NetworkPolicy(network_policy_id) => {
                self.ensure_not_reserved_network_policy(network_policy_id)
            }
        }
    }

    /// See [`CatalogState::deserialize_plan_with_enable_for_item_parsing`].
    pub(crate) fn deserialize_plan_with_enable_for_item_parsing(
        &mut self,
        create_sql: &str,
        force_if_exists_skip: bool,
    ) -> Result<(Plan, ResolvedIds), AdapterError> {
        self.state
            .deserialize_plan_with_enable_for_item_parsing(create_sql, force_if_exists_skip)
    }

    /// Cache global and, optionally, local expressions for the given
    /// `GlobalId` of an item being created, whose only version is therefore
    /// the root [`RelationVersion`].
    ///
    /// Takes the plans and metainfo directly as parameters (rather than
    /// fishing them out of catalog state), so this can be called **before**
    /// the catalog transaction that creates the item. Returns the future
    /// returned by [`Catalog::update_expression_cache`]; callers should
    /// `.await` it before the catalog transaction commits, so the durable
    /// expression cache is observed to contain the entries by the time any
    /// other process (or a subsequent bootstrap on this process) reads them.
    pub(crate) fn cache_expressions(
        &self,
        id: GlobalId,
        local_mir: Option<OptimizedMirRelationExpr>,
        mut global_mir: DataflowDescription<OptimizedMirRelationExpr>,
        mut physical_plan: mz_compute_types::dataflows::LirDataflowDescription,
        dataflow_metainfos: DataflowMetainfo<Arc<OptimizerNotice>>,
        optimizer_features: OptimizerFeatures,
    ) -> BoxFuture<'static, ()> {
        // Make sure we're not caching the result of timestamp selection, as
        // it will almost certainly be wrong if we re-install the dataflow at
        // a later time.
        global_mir.as_of = None;
        global_mir.until = Default::default();
        physical_plan.as_of = None;
        physical_plan.until = Default::default();

        let mut local_exprs = Vec::new();
        if let Some(local_mir) = local_mir {
            local_exprs.push((
                id,
                LocalExpressions {
                    local_mir,
                    optimizer_features: optimizer_features.clone(),
                    item_version: RelationVersion::root(),
                },
            ));
        }
        let global_exprs = vec![(
            id,
            GlobalExpressions {
                global_mir,
                physical_plan,
                dataflow_metainfos,
                optimizer_features,
                item_version: RelationVersion::root(),
            },
        )];
        self.update_expression_cache(local_exprs, global_exprs, Default::default())
    }

    /// Cache the local expression for the given `GlobalId` of an item being
    /// created, which has no dataflow, such as a view. Otherwise the same as
    /// [`Catalog::cache_expressions`].
    pub(crate) fn cache_local_expression(
        &self,
        id: GlobalId,
        local_mir: OptimizedMirRelationExpr,
        optimizer_features: OptimizerFeatures,
    ) -> BoxFuture<'static, ()> {
        let local_exprs = vec![(
            id,
            LocalExpressions {
                local_mir,
                optimizer_features,
                item_version: RelationVersion::root(),
            },
        )];
        self.update_expression_cache(local_exprs, Vec::new(), Default::default())
    }

    pub(crate) fn update_expression_cache<'a, 'b>(
        &'a self,
        new_local_expressions: Vec<(GlobalId, LocalExpressions)>,
        new_global_expressions: Vec<(GlobalId, GlobalExpressions)>,
        invalidate_ids: BTreeSet<GlobalId>,
    ) -> BoxFuture<'b, ()> {
        if let Some(expr_cache) = &self.expr_cache_handle {
            expr_cache
                .update(
                    new_local_expressions,
                    new_global_expressions,
                    invalidate_ids,
                )
                .boxed()
        } else {
            async {}.boxed()
        }
    }

    /// Listen for and apply all unconsumed updates to the durable catalog state.
    // TODO(jkosh44) When this method is actually used outside of a test we can remove the
    // `#[cfg(test)]` annotation.
    #[cfg(test)]
    async fn sync_to_current_updates(
        &mut self,
    ) -> Result<
        (
            Vec<BuiltinTableUpdate<&'static BuiltinTable>>,
            Vec<ParsedStateUpdate>,
        ),
        CatalogError,
    > {
        let updates = self.storage().await.sync_to_current_updates().await?;
        let (builtin_table_updates, catalog_updates) = self
            .state
            .apply_updates(updates, &mut state::LocalExpressionCache::Closed)
            .await;
        Ok((builtin_table_updates, catalog_updates))
    }
}

pub fn is_reserved_name(name: &str) -> bool {
    BUILTIN_PREFIXES
        .iter()
        .any(|prefix| name.starts_with(prefix))
}

/// Role names that PostgreSQL reserves for role specifications in statements
/// like `GRANT ... TO CURRENT_USER` and `SET ROLE NONE`. A role with such a
/// name would be ambiguous there, so creating one is not allowed.
///
/// PostgreSQL rejects most of these at parse time, only when they appear as
/// unquoted keywords. Our parser does not track whether an identifier was
/// quoted, so we instead reject the names themselves. Only the lowercase
/// spellings are reserved, which is what the unquoted forms normalize to, so
/// quoted names like `"CURRENT_USER"` remain valid, as in PostgreSQL.
const RESERVED_ROLE_SPECIFICATION_NAMES: [&str; 5] = [
    "current_user",
    "current_role",
    "session_user",
    "user",
    "none",
];

pub fn is_reserved_role_name(name: &str) -> bool {
    is_reserved_name(name)
        || is_public_role(name)
        || RESERVED_ROLE_SPECIFICATION_NAMES.contains(&name)
}

pub fn is_public_role(name: &str) -> bool {
    name == &*PUBLIC_ROLE_NAME
}

pub(crate) fn catalog_type_to_audit_object_type(sql_type: SqlCatalogItemType) -> ObjectType {
    object_type_to_audit_object_type(sql_type.into())
}

pub(crate) fn comment_id_to_audit_object_type(id: CommentObjectId) -> ObjectType {
    match id {
        CommentObjectId::Table(_) => ObjectType::Table,
        CommentObjectId::View(_) => ObjectType::View,
        CommentObjectId::MaterializedView(_) => ObjectType::MaterializedView,
        CommentObjectId::Source(_) => ObjectType::Source,
        CommentObjectId::Sink(_) => ObjectType::Sink,
        // Unreachable by construction: `COMMENT ON METRIC SINK` is rejected at parse, so no
        // metric-sink comment id is ever built. The arm exists only for exhaustiveness.
        CommentObjectId::MetricSink(_) => ObjectType::MetricSink,
        CommentObjectId::Index(_) => ObjectType::Index,
        CommentObjectId::Func(_) => ObjectType::Func,
        CommentObjectId::Connection(_) => ObjectType::Connection,
        CommentObjectId::Type(_) => ObjectType::Type,
        CommentObjectId::Secret(_) => ObjectType::Secret,
        CommentObjectId::Role(_) => ObjectType::Role,
        CommentObjectId::Database(_) => ObjectType::Database,
        CommentObjectId::Schema(_) => ObjectType::Schema,
        CommentObjectId::Cluster(_) => ObjectType::Cluster,
        CommentObjectId::ClusterReplica(_) => ObjectType::ClusterReplica,
        CommentObjectId::NetworkPolicy(_) => ObjectType::NetworkPolicy,
    }
}

pub(crate) fn object_type_to_audit_object_type(
    object_type: mz_sql::catalog::ObjectType,
) -> ObjectType {
    system_object_type_to_audit_object_type(&SystemObjectType::Object(object_type))
}

pub(crate) fn system_object_type_to_audit_object_type(
    system_type: &SystemObjectType,
) -> ObjectType {
    match system_type {
        SystemObjectType::Object(object_type) => match object_type {
            mz_sql::catalog::ObjectType::Table => ObjectType::Table,
            mz_sql::catalog::ObjectType::View => ObjectType::View,
            mz_sql::catalog::ObjectType::MaterializedView => ObjectType::MaterializedView,
            mz_sql::catalog::ObjectType::Source => ObjectType::Source,
            mz_sql::catalog::ObjectType::Sink => ObjectType::Sink,
            mz_sql::catalog::ObjectType::MetricSink => ObjectType::MetricSink,
            mz_sql::catalog::ObjectType::Index => ObjectType::Index,
            mz_sql::catalog::ObjectType::Type => ObjectType::Type,
            mz_sql::catalog::ObjectType::Role => ObjectType::Role,
            mz_sql::catalog::ObjectType::Cluster => ObjectType::Cluster,
            mz_sql::catalog::ObjectType::ClusterReplica => ObjectType::ClusterReplica,
            mz_sql::catalog::ObjectType::Secret => ObjectType::Secret,
            mz_sql::catalog::ObjectType::Connection => ObjectType::Connection,
            mz_sql::catalog::ObjectType::Database => ObjectType::Database,
            mz_sql::catalog::ObjectType::Schema => ObjectType::Schema,
            mz_sql::catalog::ObjectType::Func => ObjectType::Func,
            mz_sql::catalog::ObjectType::NetworkPolicy => ObjectType::NetworkPolicy,
        },
        SystemObjectType::System => ObjectType::System,
    }
}

#[derive(Debug, Copy, Clone)]
pub enum UpdatePrivilegeVariant {
    Grant,
    Revoke,
}

impl From<UpdatePrivilegeVariant> for ExecuteResponse {
    fn from(variant: UpdatePrivilegeVariant) -> Self {
        match variant {
            UpdatePrivilegeVariant::Grant => ExecuteResponse::GrantedPrivilege,
            UpdatePrivilegeVariant::Revoke => ExecuteResponse::RevokedPrivilege,
        }
    }
}

impl From<UpdatePrivilegeVariant> for EventType {
    fn from(variant: UpdatePrivilegeVariant) -> Self {
        match variant {
            UpdatePrivilegeVariant::Grant => EventType::Grant,
            UpdatePrivilegeVariant::Revoke => EventType::Revoke,
        }
    }
}

impl ConnCatalog<'_> {
    fn resolve_item_name(
        &self,
        name: &PartialItemName,
    ) -> Result<&QualifiedItemName, SqlCatalogError> {
        self.resolve_item(name).map(|entry| entry.name())
    }

    fn resolve_function_name(
        &self,
        name: &PartialItemName,
    ) -> Result<&QualifiedItemName, SqlCatalogError> {
        self.resolve_function(name).map(|entry| entry.name())
    }

    fn resolve_type_name(
        &self,
        name: &PartialItemName,
    ) -> Result<&QualifiedItemName, SqlCatalogError> {
        self.resolve_type(name).map(|entry| entry.name())
    }
}

impl ExprHumanizer for ConnCatalog<'_> {
    fn humanize_id(&self, id: GlobalId) -> Option<String> {
        let entry = self.state.try_get_entry_by_global_id(&id)?;
        Some(self.resolve_full_name(entry.name()).to_string())
    }

    fn humanize_id_unqualified(&self, id: GlobalId) -> Option<String> {
        let entry = self.state.try_get_entry_by_global_id(&id)?;
        Some(entry.name().item.clone())
    }

    fn humanize_id_parts(&self, id: GlobalId) -> Option<Vec<String>> {
        let entry = self.state.try_get_entry_by_global_id(&id)?;
        Some(self.resolve_full_name(entry.name()).into_parts())
    }

    fn humanize_sql_scalar_type(&self, typ: &SqlScalarType, postgres_compat: bool) -> String {
        use SqlScalarType::*;

        match typ {
            Array(t) => format!("{}[]", self.humanize_sql_scalar_type(t, postgres_compat)),
            List {
                custom_id: Some(item_id),
                ..
            }
            | Map {
                custom_id: Some(item_id),
                ..
            } => {
                let item = self.get_item(item_id);
                self.minimal_qualification(item.name()).to_string()
            }
            List { element_type, .. } => {
                format!(
                    "{} list",
                    self.humanize_sql_scalar_type(element_type, postgres_compat)
                )
            }
            Map { value_type, .. } => format!(
                "map[{}=>{}]",
                self.humanize_sql_scalar_type(&SqlScalarType::String, postgres_compat),
                self.humanize_sql_scalar_type(value_type, postgres_compat)
            ),
            Record {
                custom_id: Some(item_id),
                ..
            } => {
                let item = self.get_item(item_id);
                self.minimal_qualification(item.name()).to_string()
            }
            Record { fields, .. } => format!(
                "record({})",
                fields
                    .iter()
                    .map(|f| format!(
                        "{}: {}",
                        f.0,
                        self.humanize_sql_column_type(&f.1, postgres_compat)
                    ))
                    .join(",")
            ),
            PgLegacyChar => "\"char\"".into(),
            Char { length } if !postgres_compat => match length {
                None => "char".into(),
                Some(length) => format!("char({})", length.into_u32()),
            },
            VarChar { max_length } if !postgres_compat => match max_length {
                None => "varchar".into(),
                Some(length) => format!("varchar({})", length.into_u32()),
            },
            UInt16 => "uint2".into(),
            UInt32 => "uint4".into(),
            UInt64 => "uint8".into(),
            ty => {
                let pgrepr_type = mz_pgrepr::Type::from(ty);
                let pg_catalog_schema = SchemaSpecifier::Id(self.state.get_pg_catalog_schema_id());

                let res = if self
                    .effective_search_path(true)
                    .iter()
                    .any(|(_, schema)| schema == &pg_catalog_schema)
                {
                    pgrepr_type.name().to_string()
                } else {
                    // If PG_CATALOG_SCHEMA is not in search path, you need
                    // qualified object name to refer to type.
                    let name = QualifiedItemName {
                        qualifiers: ItemQualifiers {
                            database_spec: ResolvedDatabaseSpecifier::Ambient,
                            schema_spec: pg_catalog_schema,
                        },
                        item: pgrepr_type.name().to_string(),
                    };
                    self.resolve_full_name(&name).to_string()
                };
                res
            }
        }
    }

    fn column_names_for_id(&self, id: GlobalId) -> Option<Vec<String>> {
        let entry = self.state.try_get_entry_by_global_id(&id)?;

        match entry.index() {
            Some(index) => {
                let on_desc = self.state.try_get_desc_by_global_id(&index.on)?;
                let mut on_names = on_desc
                    .iter_names()
                    .map(|col_name| col_name.to_string())
                    .collect::<Vec<_>>();

                let (p, _) = mz_expr::permutation_for_arrangement(&index.keys, on_desc.arity());

                // Init ix_names with unknown column names. Unknown columns are
                // represented as an empty String and rendered as `#c` by the
                // Display::fmt implementation for HumanizedExpr<'a, usize, M>.
                let ix_arity = p.iter().map(|x| *x + 1).max().unwrap_or(0);
                let mut ix_names = vec![String::new(); ix_arity];

                // Apply the permutation by swapping on_names with ix_names.
                for (on_pos, ix_pos) in p.into_iter().enumerate() {
                    let on_name = on_names.get_mut(on_pos).expect("on_name");
                    let ix_name = ix_names.get_mut(ix_pos).expect("ix_name");
                    std::mem::swap(on_name, ix_name);
                }

                Some(ix_names) // Return the updated ix_names vector.
            }
            None => {
                let desc = self.state.try_get_desc_by_global_id(&id)?;
                let column_names = desc
                    .iter_names()
                    .map(|col_name| col_name.to_string())
                    .collect();

                Some(column_names)
            }
        }
    }

    fn humanize_column(&self, id: GlobalId, column: usize) -> Option<String> {
        let desc = self.state.try_get_desc_by_global_id(&id)?;
        Some(desc.get_name(column).to_string())
    }

    fn id_exists(&self, id: GlobalId) -> bool {
        self.state.entry_by_global_id.contains_key(&id)
    }
}

impl SessionCatalog for ConnCatalog<'_> {
    fn active_role_id(&self) -> &RoleId {
        &self.role_id
    }

    fn restrict_to_user_objects(&self) -> bool {
        self.restrict_to_user_objects
    }

    fn get_prepared_statement_desc(&self, name: &str) -> Option<&StatementDesc> {
        self.prepared_statements
            .as_ref()
            .map(|ps| ps.get(name).map(|ps| ps.desc()))
            .flatten()
    }

    fn get_portal_desc_unverified(&self, portal_name: &str) -> Option<&StatementDesc> {
        self.portals
            .and_then(|portals| portals.get(portal_name).map(|portal| &portal.desc))
    }

    fn active_database(&self) -> Option<&DatabaseId> {
        self.database.as_ref()
    }

    fn active_cluster(&self) -> &str {
        &self.cluster
    }

    fn search_path(&self) -> &[(ResolvedDatabaseSpecifier, SchemaSpecifier)] {
        &self.search_path
    }

    fn resolve_database(
        &self,
        database_name: &str,
    ) -> Result<&dyn mz_sql::catalog::CatalogDatabase, SqlCatalogError> {
        Ok(self.state.resolve_database(database_name)?)
    }

    fn get_database(&self, id: &DatabaseId) -> &dyn mz_sql::catalog::CatalogDatabase {
        self.state
            .database_by_id
            .get(id)
            .expect("database doesn't exist")
    }

    // `as` is ok to use to cast to a trait object.
    #[allow(clippy::as_conversions)]
    fn get_databases(&self) -> Vec<&dyn CatalogDatabase> {
        self.state
            .database_by_id
            .values()
            .map(|database| database as &dyn CatalogDatabase)
            .collect()
    }

    fn resolve_schema(
        &self,
        database_name: Option<&str>,
        schema_name: &str,
    ) -> Result<&dyn mz_sql::catalog::CatalogSchema, SqlCatalogError> {
        Ok(self.state.resolve_schema(
            self.database.as_ref(),
            database_name,
            schema_name,
            &self.conn_id,
        )?)
    }

    fn resolve_schema_in_database(
        &self,
        database_spec: &ResolvedDatabaseSpecifier,
        schema_name: &str,
    ) -> Result<&dyn mz_sql::catalog::CatalogSchema, SqlCatalogError> {
        Ok(self
            .state
            .resolve_schema_in_database(database_spec, schema_name, &self.conn_id)?)
    }

    fn get_schema(
        &self,
        database_spec: &ResolvedDatabaseSpecifier,
        schema_spec: &SchemaSpecifier,
    ) -> &dyn CatalogSchema {
        self.state
            .get_schema(database_spec, schema_spec, &self.conn_id)
    }

    // `as` is ok to use to cast to a trait object.
    #[allow(clippy::as_conversions)]
    fn get_schemas(&self) -> Vec<&dyn CatalogSchema> {
        self.get_databases()
            .into_iter()
            .flat_map(|database| database.schemas().into_iter())
            .chain(
                self.state
                    .ambient_schemas_by_id
                    .values()
                    .chain(self.state.temporary_namespaces.schemas())
                    .map(|schema| schema as &dyn CatalogSchema),
            )
            .collect()
    }

    fn get_mz_internal_schema_id(&self) -> SchemaId {
        self.state().get_mz_internal_schema_id()
    }

    fn get_mz_unsafe_schema_id(&self) -> SchemaId {
        self.state().get_mz_unsafe_schema_id()
    }

    fn is_system_schema_specifier(&self, schema: SchemaSpecifier) -> bool {
        self.state.is_system_schema_specifier(schema)
    }

    fn resolve_role(
        &self,
        role_name: &str,
    ) -> Result<&dyn mz_sql::catalog::CatalogRole, SqlCatalogError> {
        match self.state.try_get_role_by_name(role_name) {
            Some(role) => Ok(role),
            None => Err(SqlCatalogError::UnknownRole(role_name.into())),
        }
    }

    fn resolve_network_policy(
        &self,
        policy_name: &str,
    ) -> Result<&dyn mz_sql::catalog::CatalogNetworkPolicy, SqlCatalogError> {
        match self.state.try_get_network_policy_by_name(policy_name) {
            Some(policy) => Ok(policy),
            None => Err(SqlCatalogError::UnknownNetworkPolicy(policy_name.into())),
        }
    }

    fn try_get_role(&self, id: &RoleId) -> Option<&dyn CatalogRole> {
        Some(self.state.roles_by_id.get(id)?)
    }

    fn get_role(&self, id: &RoleId) -> &dyn mz_sql::catalog::CatalogRole {
        self.state.get_role(id)
    }

    fn get_roles(&self) -> Vec<&dyn CatalogRole> {
        // `as` is ok to use to cast to a trait object.
        #[allow(clippy::as_conversions)]
        self.state
            .roles_by_id
            .values()
            .map(|role| role as &dyn CatalogRole)
            .collect()
    }

    fn mz_system_role_id(&self) -> RoleId {
        MZ_SYSTEM_ROLE_ID
    }

    fn collect_role_membership(&self, id: &RoleId) -> BTreeSet<RoleId> {
        self.state.collect_role_membership(id)
    }

    fn get_network_policy(
        &self,
        id: &NetworkPolicyId,
    ) -> &dyn mz_sql::catalog::CatalogNetworkPolicy {
        self.state.get_network_policy(id)
    }

    fn get_network_policies(&self) -> Vec<&dyn mz_sql::catalog::CatalogNetworkPolicy> {
        // `as` is ok to use to cast to a trait object.
        #[allow(clippy::as_conversions)]
        self.state
            .network_policies_by_id
            .values()
            .map(|policy| policy as &dyn CatalogNetworkPolicy)
            .collect()
    }

    fn resolve_cluster(
        &self,
        cluster_name: Option<&str>,
    ) -> Result<&dyn mz_sql::catalog::CatalogCluster<'_>, SqlCatalogError> {
        Ok(self
            .state
            .resolve_cluster(cluster_name.unwrap_or_else(|| self.active_cluster()))?)
    }

    fn resolve_cluster_replica(
        &self,
        cluster_replica_name: &QualifiedReplica,
    ) -> Result<&dyn CatalogClusterReplica<'_>, SqlCatalogError> {
        Ok(self.state.resolve_cluster_replica(cluster_replica_name)?)
    }

    fn resolve_item(
        &self,
        name: &PartialItemName,
    ) -> Result<&dyn mz_sql::catalog::CatalogItem, SqlCatalogError> {
        let r = self.state.resolve_entry(
            self.database.as_ref(),
            &self.effective_search_path(true),
            name,
            &self.conn_id,
        )?;
        if self.unresolvable_ids.contains(&r.id()) {
            Err(SqlCatalogError::UnknownItem(name.to_string()))
        } else {
            Ok(r)
        }
    }

    fn resolve_function(
        &self,
        name: &PartialItemName,
    ) -> Result<&dyn mz_sql::catalog::CatalogItem, SqlCatalogError> {
        let r = self.state.resolve_function(
            self.database.as_ref(),
            &self.effective_search_path(false),
            name,
            &self.conn_id,
        )?;

        if self.unresolvable_ids.contains(&r.id()) {
            Err(SqlCatalogError::UnknownFunction {
                name: name.to_string(),
                alternative: None,
            })
        } else {
            Ok(r)
        }
    }

    fn resolve_type(
        &self,
        name: &PartialItemName,
    ) -> Result<&dyn mz_sql::catalog::CatalogItem, SqlCatalogError> {
        let r = self.state.resolve_type(
            self.database.as_ref(),
            &self.effective_search_path(false),
            name,
            &self.conn_id,
        )?;

        if self.unresolvable_ids.contains(&r.id()) {
            Err(SqlCatalogError::UnknownType {
                name: name.to_string(),
            })
        } else {
            Ok(r)
        }
    }

    fn get_system_type(&self, name: &str) -> &dyn mz_sql::catalog::CatalogItem {
        self.state.get_system_type(name)
    }

    fn try_get_item(&self, id: &CatalogItemId) -> Option<&dyn mz_sql::catalog::CatalogItem> {
        Some(self.state.try_get_entry(id)?)
    }

    fn try_get_item_by_global_id(
        &self,
        id: &GlobalId,
    ) -> Option<Box<dyn mz_sql::catalog::CatalogCollectionItem>> {
        let entry = self.state.try_get_entry_by_global_id(id)?;
        let entry = match &entry.item {
            CatalogItem::Table(table) => {
                let (version, _gid) = table
                    .collections
                    .iter()
                    .find(|(_version, gid)| *gid == id)
                    .expect("catalog out of sync, mismatched GlobalId");
                entry.at_version(RelationVersionSelector::Specific(*version))
            }
            _ => entry.at_version(RelationVersionSelector::Latest),
        };
        Some(entry)
    }

    fn get_item(&self, id: &CatalogItemId) -> &dyn mz_sql::catalog::CatalogItem {
        self.state.get_entry(id)
    }

    fn get_item_by_global_id(
        &self,
        id: &GlobalId,
    ) -> Box<dyn mz_sql::catalog::CatalogCollectionItem> {
        let entry = self.state.get_entry_by_global_id(id);
        let entry = match &entry.item {
            CatalogItem::Table(table) => {
                let (version, _gid) = table
                    .collections
                    .iter()
                    .find(|(_version, gid)| *gid == id)
                    .expect("catalog out of sync, mismatched GlobalId");
                entry.at_version(RelationVersionSelector::Specific(*version))
            }
            _ => entry.at_version(RelationVersionSelector::Latest),
        };
        entry
    }

    fn get_items(&self) -> Vec<&dyn mz_sql::catalog::CatalogItem> {
        self.get_schemas()
            .into_iter()
            .flat_map(|schema| schema.item_ids())
            .map(|id| self.get_item(&id))
            .collect()
    }

    fn get_item_by_name(&self, name: &QualifiedItemName) -> Option<&dyn SqlCatalogItem> {
        self.state
            .get_item_by_name(name, &self.conn_id)
            .map(|item| convert::identity::<&dyn SqlCatalogItem>(item))
    }

    fn get_type_by_name(&self, name: &QualifiedItemName) -> Option<&dyn SqlCatalogItem> {
        self.state
            .get_type_by_name(name, &self.conn_id)
            .map(|item| convert::identity::<&dyn SqlCatalogItem>(item))
    }

    fn get_cluster(&self, id: ClusterId) -> &dyn mz_sql::catalog::CatalogCluster<'_> {
        &self.state.clusters_by_id[&id]
    }

    fn get_clusters(&self) -> Vec<&dyn mz_sql::catalog::CatalogCluster<'_>> {
        self.state
            .clusters_by_id
            .values()
            .map(|cluster| convert::identity::<&dyn mz_sql::catalog::CatalogCluster>(cluster))
            .collect()
    }

    fn get_cluster_replica(
        &self,
        cluster_id: ClusterId,
        replica_id: ReplicaId,
    ) -> &dyn mz_sql::catalog::CatalogClusterReplica<'_> {
        let cluster = self.get_cluster(cluster_id);
        cluster.replica(replica_id)
    }

    fn get_cluster_replicas(&self) -> Vec<&dyn mz_sql::catalog::CatalogClusterReplica<'_>> {
        self.get_clusters()
            .into_iter()
            .flat_map(|cluster| cluster.replicas().into_iter())
            .collect()
    }

    fn get_system_privileges(&self) -> &PrivilegeMap {
        &self.state.system_privileges
    }

    fn get_default_privileges(
        &self,
    ) -> Vec<(&DefaultPrivilegeObject, Vec<&DefaultPrivilegeAclItem>)> {
        self.state
            .default_privileges
            .iter()
            .map(|(object, acl_items)| (object, acl_items.collect()))
            .collect()
    }

    fn find_available_name(&self, name: QualifiedItemName) -> QualifiedItemName {
        self.state.find_available_name(name, &self.conn_id)
    }

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

    fn resolve_full_schema_name(&self, name: &QualifiedSchemaName) -> FullSchemaName {
        self.state.resolve_full_schema_name(name)
    }

    fn resolve_item_id(&self, global_id: &GlobalId) -> CatalogItemId {
        self.state.get_entry_by_global_id(global_id).id()
    }

    fn resolve_global_id(
        &self,
        item_id: &CatalogItemId,
        version: RelationVersionSelector,
    ) -> GlobalId {
        self.state
            .get_entry(item_id)
            .at_version(version)
            .global_id()
    }

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

    fn now(&self) -> EpochMillis {
        (self.state.config().now)()
    }

    fn aws_privatelink_availability_zones(&self) -> Option<BTreeSet<String>> {
        self.state.aws_privatelink_availability_zones.clone()
    }

    fn system_vars(&self) -> &SystemVars {
        &self.state.system_configuration
    }

    fn system_vars_mut(&mut self) -> &mut SystemVars {
        Arc::make_mut(&mut self.state.to_mut().system_configuration)
    }

    fn get_owner_id(&self, id: &ObjectId) -> Option<RoleId> {
        self.state().get_owner_id(id, self.conn_id())
    }

    fn get_privileges(&self, id: &SystemObjectId) -> Option<&PrivilegeMap> {
        match id {
            SystemObjectId::System => Some(&self.state.system_privileges),
            SystemObjectId::Object(ObjectId::Cluster(id)) => {
                Some(self.get_cluster(*id).privileges())
            }
            SystemObjectId::Object(ObjectId::Database(id)) => {
                Some(self.get_database(id).privileges())
            }
            SystemObjectId::Object(ObjectId::Schema((database_spec, schema_spec))) => {
                // For temporary schemas that haven't been created yet (lazy creation),
                // we return None - the RBAC check will need to handle this case.
                self.state
                    .try_get_schema(database_spec, schema_spec, &self.conn_id)
                    .map(|schema| schema.privileges())
            }
            SystemObjectId::Object(ObjectId::Item(id)) => Some(self.get_item(id).privileges()),
            SystemObjectId::Object(ObjectId::NetworkPolicy(id)) => {
                Some(self.get_network_policy(id).privileges())
            }
            SystemObjectId::Object(ObjectId::ClusterReplica(_))
            | SystemObjectId::Object(ObjectId::Role(_)) => None,
        }
    }

    fn object_dependents(&self, ids: &Vec<ObjectId>) -> Vec<ObjectId> {
        let mut seen = BTreeSet::new();
        self.state.object_dependents(ids, &self.conn_id, &mut seen)
    }

    fn item_dependents(&self, id: CatalogItemId) -> Vec<ObjectId> {
        let mut seen = BTreeSet::new();
        self.state.item_dependents(id, &mut seen)
    }

    fn all_object_privileges(&self, object_type: mz_sql::catalog::SystemObjectType) -> AclMode {
        rbac::all_object_privileges(object_type)
    }

    fn get_object_type(&self, object_id: &ObjectId) -> mz_sql::catalog::ObjectType {
        self.state.get_object_type(object_id)
    }

    fn get_system_object_type(&self, id: &SystemObjectId) -> mz_sql::catalog::SystemObjectType {
        self.state.get_system_object_type(id)
    }

    /// Returns a [`PartialItemName`] with the minimum amount of qualifiers to unambiguously resolve
    /// the object.
    ///
    /// Warning: This is broken for temporary objects. Don't use this function for serious stuff,
    /// i.e., don't expect that what you get back is a thing you can resolve. Current usages are
    /// only for error msgs and other humanizations.
    fn minimal_qualification(&self, qualified_name: &QualifiedItemName) -> PartialItemName {
        if qualified_name.qualifiers.schema_spec.is_temporary() {
            // All bets are off. Just give up and return the qualified name as is.
            // TODO: Figure out what's going on with temporary objects.

            // See e.g. `temporary_objects.slt` fail if you comment this out, which has the repro
            // from https://github.com/MaterializeInc/database-issues/issues/9973#issuecomment-3646382143
            // There is also https://github.com/MaterializeInc/database-issues/issues/9974, for
            // which we don't have a simple repro.
            return qualified_name.item.clone().into();
        }

        let database_id = match &qualified_name.qualifiers.database_spec {
            ResolvedDatabaseSpecifier::Ambient => None,
            ResolvedDatabaseSpecifier::Id(id)
                if self.database.is_some() && self.database == Some(*id) =>
            {
                None
            }
            ResolvedDatabaseSpecifier::Id(id) => Some(id.clone()),
        };

        let schema_spec = if database_id.is_none()
            && self.resolve_item_name(&PartialItemName {
                database: None,
                schema: None,
                item: qualified_name.item.clone(),
            }) == Ok(qualified_name)
            || self.resolve_function_name(&PartialItemName {
                database: None,
                schema: None,
                item: qualified_name.item.clone(),
            }) == Ok(qualified_name)
            || self.resolve_type_name(&PartialItemName {
                database: None,
                schema: None,
                item: qualified_name.item.clone(),
            }) == Ok(qualified_name)
        {
            None
        } else {
            // If `search_path` does not contain `full_name.schema`, the
            // `PartialName` must contain it.
            Some(qualified_name.qualifiers.schema_spec.clone())
        };

        let res = PartialItemName {
            database: database_id.map(|id| self.get_database(&id).name().to_string()),
            schema: schema_spec.map(|spec| {
                self.get_schema(&qualified_name.qualifiers.database_spec, &spec)
                    .name()
                    .schema
                    .clone()
            }),
            item: qualified_name.item.clone(),
        };
        assert!(
            self.resolve_item_name(&res) == Ok(qualified_name)
                || self.resolve_function_name(&res) == Ok(qualified_name)
                || self.resolve_type_name(&res) == Ok(qualified_name)
        );
        res
    }

    fn add_notice(&self, notice: PlanNotice) {
        let _ = self.notices_tx.send(notice.into());
    }

    fn get_item_comments(&self, id: &CatalogItemId) -> Option<&BTreeMap<Option<usize>, String>> {
        let comment_id = self.state.get_comment_id(ObjectId::Item(*id));
        self.state.comments.get_object_comments(comment_id)
    }

    fn is_cluster_size_cc(&self, size: &str) -> bool {
        self.state
            .cluster_replica_sizes
            .0
            .get(size)
            .map_or(false, |a| a.is_cc)
    }
}

#[cfg(test)]
mod source_preparation_tests;

#[cfg(test)]
mod try_cast_tests;

#[cfg(test)]
mod tests {
    use std::collections::{BTreeMap, BTreeSet};
    use std::sync::Arc;
    use std::{env, iter};

    use itertools::Itertools;
    use mz_catalog::memory::objects::{
        CatalogItem, ClusterConfig, ClusterVariant, ClusterVariantManaged, ReconfigurationState,
        ReconfigurationStatus, ReconfigurationTarget,
    };
    use mz_controller::clusters::ReplicaLogging;
    use mz_postgres_util::{query, sql};
    use tokio_postgres::NoTls;
    use tokio_postgres::types::Type;
    use uuid::Uuid;

    use mz_catalog::SYSTEM_CONN_ID;
    use mz_catalog::builtin::{BUILTINS, Builtin, BuiltinType};
    use mz_catalog::durable::{CatalogError, DurableCatalogError, FenceError, test_bootstrap_args};
    use mz_controller_types::{ClusterId, ReplicaId};
    use mz_expr::{Eval, MirScalarExpr};
    use mz_ore::now::to_datetime;
    use mz_ore::{assert_err, assert_ok, soft_assert_eq_or_log, task};
    use mz_persist_client::PersistClient;
    use mz_pgrepr::oid::{FIRST_MATERIALIZE_OID, FIRST_UNPINNED_OID, FIRST_USER_OID};
    use mz_repr::namespaces::{INFORMATION_SCHEMA, PG_CATALOG_SCHEMA};
    use mz_repr::role_id::RoleId;
    use mz_repr::{
        CatalogItemId, Datum, GlobalId, RelationVersionSelector, Row, RowArena, SqlRelationType,
        SqlScalarType, Timestamp,
    };
    use mz_sql::catalog::{CatalogSchema, CatalogType, SessionCatalog};
    use mz_sql::func::{Func, FuncImpl, OP_IMPLS, Operation};
    use mz_sql::names::{
        self, DatabaseId, ItemQualifiers, ObjectId, PartialItemName, QualifiedItemName,
        ResolvedDatabaseSpecifier, SchemaId, SchemaSpecifier, SystemObjectId,
    };
    use mz_sql::plan::{
        AutoScalingStrategy, CoercibleScalarExpr, ExprContext, HirScalarExpr, HirToMirConfig,
        OnHydration, OnTimeoutAction, PlanContext, QueryContext, QueryLifetime, Scope,
        StatementContext,
    };
    use mz_sql::session::user::MZ_SYSTEM_ROLE_ID;
    use mz_sql::session::vars::{SystemVars, VarInput};

    use crate::catalog::state::LocalExpressionCache;
    use crate::catalog::{Catalog, DebugAwsContext, Op};
    use crate::optimize::dataflows::{EvalTime, ExprPrep, ExprPrepOneShot};
    use crate::session::Session;

    /// System sessions have an empty `search_path` so it's necessary to
    /// schema-qualify all referenced items.
    ///
    /// Dummy (and ostensibly client) sessions contain system schemas in their
    /// search paths, so do not require schema qualification on system objects such
    /// as types.
    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] //  unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
    async fn test_minimal_qualification() {
        Catalog::with_debug(|catalog| async move {
            struct TestCase {
                input: QualifiedItemName,
                system_output: PartialItemName,
                normal_output: PartialItemName,
            }

            let test_cases = vec![
                TestCase {
                    input: QualifiedItemName {
                        qualifiers: ItemQualifiers {
                            database_spec: ResolvedDatabaseSpecifier::Ambient,
                            schema_spec: SchemaSpecifier::Id(catalog.get_pg_catalog_schema_id()),
                        },
                        item: "numeric".to_string(),
                    },
                    system_output: PartialItemName {
                        database: None,
                        schema: None,
                        item: "numeric".to_string(),
                    },
                    normal_output: PartialItemName {
                        database: None,
                        schema: None,
                        item: "numeric".to_string(),
                    },
                },
                TestCase {
                    input: QualifiedItemName {
                        qualifiers: ItemQualifiers {
                            database_spec: ResolvedDatabaseSpecifier::Ambient,
                            schema_spec: SchemaSpecifier::Id(catalog.get_mz_catalog_schema_id()),
                        },
                        item: "mz_array_types".to_string(),
                    },
                    system_output: PartialItemName {
                        database: None,
                        schema: None,
                        item: "mz_array_types".to_string(),
                    },
                    normal_output: PartialItemName {
                        database: None,
                        schema: None,
                        item: "mz_array_types".to_string(),
                    },
                },
            ];

            for tc in test_cases {
                assert_eq!(
                    catalog
                        .for_system_session()
                        .minimal_qualification(&tc.input),
                    tc.system_output
                );
                assert_eq!(
                    catalog
                        .for_session(&Session::dummy())
                        .minimal_qualification(&tc.input),
                    tc.normal_output
                );
            }
            catalog.expire().await;
        })
        .await
    }

    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] //  unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
    async fn test_catalog_revision() {
        let persist_client = PersistClient::new_for_tests().await;
        let organization_id = Uuid::new_v4();
        let bootstrap_args = test_bootstrap_args();
        {
            let mut catalog = Catalog::open_debug_catalog(
                persist_client.clone(),
                organization_id.clone(),
                &bootstrap_args,
            )
            .await
            .expect("unable to open debug catalog");
            assert_eq!(catalog.transient_revision(), 1);
            assert!(catalog.transient_revision_is_current());
            let snapshot = catalog.clone();
            let commit_ts = catalog.current_upper().await;
            catalog
                .transact(
                    None,
                    commit_ts,
                    None,
                    vec![Op::CreateDatabase {
                        name: "test".to_string(),
                        owner_id: MZ_SYSTEM_ROLE_ID,
                    }],
                )
                .await
                .expect("failed to transact");
            assert_eq!(catalog.transient_revision(), 2);
            assert!(catalog.transient_revision_is_current());
            // The pre-transaction snapshot detects its own staleness through
            // the shared latest revision.
            assert!(!snapshot.transient_revision_is_current());
            catalog.expire().await;
        }
        {
            let catalog =
                Catalog::open_debug_catalog(persist_client, organization_id, &bootstrap_args)
                    .await
                    .expect("unable to open debug catalog");
            // Re-opening the same catalog resets the transient_revision to 1.
            assert_eq!(catalog.transient_revision(), 1);
            catalog.expire().await;
        }
    }

    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] //  unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
    async fn test_builtin_cluster_replica_size_sync() {
        let persist_client = PersistClient::new_for_tests().await;
        let organization_id = Uuid::new_v4();
        let mut bootstrap_args = test_bootstrap_args();
        // Builtin cluster replicas use the default size.
        let in_use = bootstrap_args.default_cluster_replica_size.clone();
        let unused = "scale=1,workers=2".to_string();
        let changed = "scale=1,workers=4".to_string();
        let cluster_only = "scale=1,workers=8".to_string();
        let hydration = "scale=1,workers=16".to_string();
        let timed_out_target = "scale=1,workers=32".to_string();

        let mut catalog =
            Catalog::open_debug_catalog(persist_client.clone(), organization_id, &bootstrap_args)
                .await
                .expect("unable to open debug catalog");
        assert_eq!(
            catalog.cluster_replica_sizes().0,
            bootstrap_args.cluster_replica_size_map.0
        );
        // A managed cluster with no replicas still depends on its size and
        // hydration size, but not on a settled reconfiguration's target.
        let commit_ts = catalog.current_upper().await;
        let cluster_id = catalog
            .allocate_user_cluster_id(commit_ts)
            .await
            .expect("allocate cluster id");
        let config = ClusterConfig {
            variant: ClusterVariant::Managed(ClusterVariantManaged {
                size: cluster_only.clone(),
                availability_zones: Vec::new(),
                logging: ReplicaLogging {
                    log_logging: false,
                    interval: None,
                },
                arrangement_compression: false,
                replication_factor: 0,
                optimizer_feature_overrides: Default::default(),
                schedule: Default::default(),
                auto_scaling_strategy: Some(AutoScalingStrategy {
                    on_hydration: Some(OnHydration {
                        hydration_size: hydration.clone(),
                        linger_duration: None,
                    }),
                }),
                reconfiguration: Some(ReconfigurationState {
                    target: ReconfigurationTarget {
                        size: timed_out_target.clone(),
                        replication_factor: 1,
                        availability_zones: Vec::new(),
                        logging: ReplicaLogging {
                            log_logging: false,
                            interval: None,
                        },
                        arrangement_compression: false,
                    },
                    deadline: Timestamp::MIN,
                    on_timeout: OnTimeoutAction::Rollback,
                    status: ReconfigurationStatus::TimedOut,
                }),
                burst: None,
            }),
            workload_class: None,
        };
        let commit_ts = catalog.current_upper().await;
        catalog
            .transact(
                None,
                commit_ts,
                None,
                vec![Op::CreateCluster {
                    id: cluster_id,
                    name: "no_replicas".into(),
                    introspection_sources: Vec::new(),
                    owner_id: MZ_SYSTEM_ROLE_ID,
                    config,
                }],
            )
            .await
            .expect("create cluster");
        catalog.expire().await;

        let config_sizes = &mut bootstrap_args.cluster_replica_size_map.0;
        config_sizes.remove(&in_use);
        config_sizes.remove(&unused);
        config_sizes.remove(&cluster_only);
        config_sizes.remove(&hydration);
        config_sizes.remove(&timed_out_target);
        config_sizes
            .get_mut(&changed)
            .expect("test size")
            .cpu_exclusive = true;
        let catalog =
            Catalog::open_debug_catalog(persist_client.clone(), organization_id, &bootstrap_args)
                .await
                .expect("unable to open debug catalog");
        let sizes = &catalog.cluster_replica_sizes().0;
        assert!(sizes[&in_use].disabled);
        assert!(sizes[&cluster_only].disabled);
        assert!(sizes[&hydration].disabled);
        assert!(!sizes.contains_key(&unused));
        assert!(!sizes.contains_key(&timed_out_target));
        assert_eq!(
            sizes[&changed],
            bootstrap_args.cluster_replica_size_map.0[&changed]
        );
        let durable_sizes = catalog.cluster_replica_sizes().clone();
        catalog.expire().await;

        // A read-only open with a different size config uses the durable sizes.
        bootstrap_args
            .cluster_replica_size_map
            .0
            .insert(unused.clone(), durable_sizes.0[&changed].clone());
        let catalog =
            Catalog::open_debug_read_only_catalog(persist_client, organization_id, &bootstrap_args)
                .await
                .expect("unable to open read-only debug catalog");
        assert_eq!(catalog.cluster_replica_sizes().0, durable_sizes.0);
        catalog.expire().await;
    }

    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
    async fn test_effective_search_path() {
        Catalog::with_debug(|catalog| async move {
            let mz_catalog_schema = (
                ResolvedDatabaseSpecifier::Ambient,
                SchemaSpecifier::Id(catalog.state().get_mz_catalog_schema_id()),
            );
            let pg_catalog_schema = (
                ResolvedDatabaseSpecifier::Ambient,
                SchemaSpecifier::Id(catalog.state().get_pg_catalog_schema_id()),
            );
            let mz_temp_schema = (
                ResolvedDatabaseSpecifier::Ambient,
                SchemaSpecifier::Temporary,
            );

            // Behavior with the default search_schema (public)
            let session = Session::dummy();
            let conn_catalog = catalog.for_session(&session);
            assert_ne!(
                conn_catalog.effective_search_path(false),
                conn_catalog.search_path
            );
            assert_ne!(
                conn_catalog.effective_search_path(true),
                conn_catalog.search_path
            );
            assert_eq!(
                conn_catalog.effective_search_path(false),
                vec![
                    mz_catalog_schema.clone(),
                    pg_catalog_schema.clone(),
                    conn_catalog.search_path[0].clone()
                ]
            );
            assert_eq!(
                conn_catalog.effective_search_path(true),
                vec![
                    mz_temp_schema.clone(),
                    mz_catalog_schema.clone(),
                    pg_catalog_schema.clone(),
                    conn_catalog.search_path[0].clone()
                ]
            );

            // missing schemas are added when missing
            let mut session = Session::dummy();
            session
                .vars_mut()
                .set(
                    &SystemVars::new(),
                    "search_path",
                    VarInput::Flat(mz_repr::namespaces::PG_CATALOG_SCHEMA),
                    false,
                )
                .expect("failed to set search_path");
            let conn_catalog = catalog.for_session(&session);
            assert_ne!(
                conn_catalog.effective_search_path(false),
                conn_catalog.search_path
            );
            assert_ne!(
                conn_catalog.effective_search_path(true),
                conn_catalog.search_path
            );
            assert_eq!(
                conn_catalog.effective_search_path(false),
                vec![mz_catalog_schema.clone(), pg_catalog_schema.clone()]
            );
            assert_eq!(
                conn_catalog.effective_search_path(true),
                vec![
                    mz_temp_schema.clone(),
                    mz_catalog_schema.clone(),
                    pg_catalog_schema.clone()
                ]
            );

            let mut session = Session::dummy();
            session
                .vars_mut()
                .set(
                    &SystemVars::new(),
                    "search_path",
                    VarInput::Flat(mz_repr::namespaces::MZ_CATALOG_SCHEMA),
                    false,
                )
                .expect("failed to set search_path");
            let conn_catalog = catalog.for_session(&session);
            assert_ne!(
                conn_catalog.effective_search_path(false),
                conn_catalog.search_path
            );
            assert_ne!(
                conn_catalog.effective_search_path(true),
                conn_catalog.search_path
            );
            assert_eq!(
                conn_catalog.effective_search_path(false),
                vec![pg_catalog_schema.clone(), mz_catalog_schema.clone()]
            );
            assert_eq!(
                conn_catalog.effective_search_path(true),
                vec![
                    mz_temp_schema.clone(),
                    pg_catalog_schema.clone(),
                    mz_catalog_schema.clone()
                ]
            );

            let mut session = Session::dummy();
            session
                .vars_mut()
                .set(
                    &SystemVars::new(),
                    "search_path",
                    VarInput::Flat(mz_repr::namespaces::MZ_TEMP_SCHEMA),
                    false,
                )
                .expect("failed to set search_path");
            let conn_catalog = catalog.for_session(&session);
            assert_ne!(
                conn_catalog.effective_search_path(false),
                conn_catalog.search_path
            );
            assert_ne!(
                conn_catalog.effective_search_path(true),
                conn_catalog.search_path
            );
            // Because we lazily initialize the `mz_temp` schema,
            // an explicit `mz_temp` search path before the first
            // temporary item creation gets filtered out in
            // `effective_search_path`.
            assert_eq!(
                conn_catalog.effective_search_path(false),
                vec![mz_catalog_schema.clone(), pg_catalog_schema.clone(),]
            );
            assert_eq!(
                conn_catalog.effective_search_path(true),
                vec![mz_temp_schema, mz_catalog_schema, pg_catalog_schema]
            );
            catalog.expire().await;
        })
        .await
    }

    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] //  unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
    async fn test_normalized_create() {
        use mz_ore::collections::CollectionExt;
        Catalog::with_debug(|catalog| async move {
            let conn_catalog = catalog.for_system_session();
            let scx = &mut StatementContext::new(None, &conn_catalog);

            let parsed = mz_sql_parser::parser::parse_statements(
                "create view public.foo as select 1 as bar",
            )
            .expect("")
            .into_element()
            .ast;

            let (stmt, _) = names::resolve(scx.catalog, parsed).expect("");

            // Ensure that all identifiers are quoted.
            assert_eq!(
                r#"CREATE VIEW "materialize"."public"."foo" AS SELECT 1 AS "bar""#,
                mz_sql::normalize::create_statement(scx, stmt).expect(""),
            );
            catalog.expire().await;
        })
        .await;
    }

    /// Resolving a statement and resolving its normalized `create_sql` must
    /// yield the same `ResolvedIds`.
    ///
    /// The in-memory catalog records the ids from the first resolution, while
    /// a catalog reload re-derives them from the stored `create_sql`. Any
    /// disagreement makes the reloaded catalog differ from the in-memory one.
    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] // slow
    async fn test_resolved_ids_survive_create_sql_round_trip() {
        use mz_ore::collections::CollectionExt;
        Catalog::with_debug(|catalog| async move {
            let conn_catalog = catalog.for_system_session();
            let scx = &mut StatementContext::new(None, &conn_catalog);

            let resolve_and_normalize = |scx: &mut StatementContext, sql: &str| {
                let parsed = mz_sql_parser::parser::parse_statements(sql)
                    .expect("parses")
                    .into_element()
                    .ast;
                let (stmt, ids) = names::resolve(scx.catalog, parsed).expect("resolves");
                let normalized =
                    mz_sql::normalize::create_statement(scx, stmt).expect("normalizes");
                (ids, normalized)
            };

            const ARRAY_SUFFIX: &str = "create table public.t (a pg_catalog.int4[])";
            const ARRAY_TYPE: &str = "create table public.t (a pg_catalog._int4)";
            const ELEMENT_TYPE: &str = "create table public.t (a pg_catalog.int4)";

            let mut ids_by_spelling: BTreeMap<&str, Vec<CatalogItemId>> = BTreeMap::new();
            for sql in [
                ARRAY_SUFFIX,
                ARRAY_TYPE,
                ELEMENT_TYPE,
                "create table public.t (a pg_catalog.int4 list)",
                "create view public.v as select null::pg_catalog.text[]",
            ] {
                let (ids, normalized) = resolve_and_normalize(scx, sql);
                let (round_tripped_ids, _) = resolve_and_normalize(scx, &normalized);
                assert_eq!(
                    ids.items().collect::<Vec<_>>(),
                    round_tripped_ids.items().collect::<Vec<_>>(),
                    "resolving {normalized:?} produced different ids than {sql:?}",
                );
                ids_by_spelling.insert(sql, ids.items().copied().collect());
            }

            // `int4[]` and `_int4` name the same type, so both spellings must
            // record the same ids. `int4[]` is only ever stored as `_int4`,
            // so this is what keeps the reloaded catalog identical to the
            // in-memory one.
            assert_eq!(
                ids_by_spelling[ARRAY_SUFFIX], ids_by_spelling[ARRAY_TYPE],
                "{ARRAY_SUFFIX:?} and {ARRAY_TYPE:?} must resolve to the same ids",
            );

            // Neither spelling records the element type: an array reference
            // names the array type alone.
            let element_ids = &ids_by_spelling[ELEMENT_TYPE];
            assert!(
                ids_by_spelling[ARRAY_SUFFIX]
                    .iter()
                    .all(|id| !element_ids.contains(id)),
                "{ARRAY_SUFFIX:?} recorded an element type id from {ELEMENT_TYPE:?}",
            );

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

    // Test that if a large catalog item is somehow committed, then we can still load the catalog.
    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] // slow
    async fn test_large_catalog_item() {
        let view_def = "CREATE VIEW \"materialize\".\"public\".\"v\" AS SELECT 1 FROM (SELECT 1";
        let column = ", 1";
        let view_def_size = view_def.bytes().count();
        let column_size = column.bytes().count();
        let column_count =
            (mz_sql_parser::parser::MAX_STATEMENT_BATCH_SIZE - view_def_size) / column_size + 1;
        let columns = iter::repeat(column).take(column_count).join("");
        let create_sql = format!("{view_def}{columns})");
        let create_sql_check = create_sql.clone();
        assert_ok!(mz_sql_parser::parser::parse_statements(&create_sql));
        assert_err!(mz_sql_parser::parser::parse_statements_with_limit(
            &create_sql
        ));

        let persist_client = PersistClient::new_for_tests().await;
        let organization_id = Uuid::new_v4();
        let id = CatalogItemId::User(1);
        let gid = GlobalId::User(1);
        let bootstrap_args = test_bootstrap_args();
        {
            let mut catalog = Catalog::open_debug_catalog(
                persist_client.clone(),
                organization_id.clone(),
                &bootstrap_args,
            )
            .await
            .expect("unable to open debug catalog");
            let item = catalog
                .state()
                .deserialize_item(
                    gid,
                    &create_sql,
                    &BTreeMap::new(),
                    &mut LocalExpressionCache::Closed,
                    None,
                )
                .expect("unable to parse view");
            let commit_ts = catalog.current_upper().await;
            catalog
                .transact(
                    None,
                    commit_ts,
                    None,
                    vec![Op::CreateItem {
                        item,
                        name: QualifiedItemName {
                            qualifiers: ItemQualifiers {
                                database_spec: ResolvedDatabaseSpecifier::Id(DatabaseId::User(1)),
                                schema_spec: SchemaSpecifier::Id(SchemaId::User(3)),
                            },
                            item: "v".to_string(),
                        },
                        id,
                        owner_id: MZ_SYSTEM_ROLE_ID,
                    }],
                )
                .await
                .expect("failed to transact");
            catalog.expire().await;
        }
        {
            let catalog =
                Catalog::open_debug_catalog(persist_client, organization_id, &bootstrap_args)
                    .await
                    .expect("unable to open debug catalog");
            let view = catalog.get_entry(&id);
            assert_eq!("v", view.name.item);
            match &view.item {
                CatalogItem::View(view) => assert_eq!(create_sql_check, view.create_sql),
                item => panic!("expected view, got {}", item.typ()),
            }
            catalog.expire().await;
        }
    }

    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] //  unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
    async fn test_object_type() {
        Catalog::with_debug(|catalog| async move {
            let conn_catalog = catalog.for_system_session();

            assert_eq!(
                mz_sql::catalog::ObjectType::ClusterReplica,
                conn_catalog.get_object_type(&ObjectId::ClusterReplica((
                    ClusterId::user(1).expect("1 is a valid ID"),
                    ReplicaId::User(1)
                )))
            );
            assert_eq!(
                mz_sql::catalog::ObjectType::Role,
                conn_catalog.get_object_type(&ObjectId::Role(RoleId::User(1)))
            );
            catalog.expire().await;
        })
        .await;
    }

    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] //  unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
    async fn test_get_privileges() {
        Catalog::with_debug(|catalog| async move {
            let conn_catalog = catalog.for_system_session();

            assert_eq!(
                None,
                conn_catalog.get_privileges(&SystemObjectId::Object(ObjectId::ClusterReplica((
                    ClusterId::user(1).expect("1 is a valid ID"),
                    ReplicaId::User(1),
                ))))
            );
            assert_eq!(
                None,
                conn_catalog
                    .get_privileges(&SystemObjectId::Object(ObjectId::Role(RoleId::User(1))))
            );
            catalog.expire().await;
        })
        .await;
    }

    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] //  unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
    async fn verify_builtin_descs() {
        // Provide a full AWS context so the builtin AWS connection views resolve
        // as they do in a cloud environment. Without an account id,
        // `mz_aws_privatelink_connections.principal` folds to NULL, its
        // `WHERE principal IS NOT NULL` filter is statically unsatisfiable, and
        // the optimizer infers an extra empty key that the declared desc, which
        // describes the cloud shape, does not carry.
        let aws_context = DebugAwsContext {
            aws_account_id: Some("123456789000".to_string()),
            aws_external_id_prefix: Some("eb5cb59b-e2fe-41f3-87ca-d2176a495345".to_string()),
            aws_connection_role_arn: Some(
                "arn:aws:iam::123456789000:role/MaterializeConnection".to_string(),
            ),
        };
        Catalog::with_debug_aws_context(aws_context, |catalog| async move {
            let conn_catalog = catalog.for_system_session();

            for builtin in BUILTINS::iter() {
                let (schema, name, expected_desc, shard_backed) = match builtin {
                    Builtin::Table(t) => (&t.schema, &t.name, &t.desc, true),
                    Builtin::View(v) => (&v.schema, &v.name, &v.desc, false),
                    Builtin::MaterializedView(mv) => (&mv.schema, &mv.name, &mv.desc, true),
                    Builtin::Source(s) => (&s.schema, &s.name, &s.desc, true),
                    Builtin::Log(_)
                    | Builtin::Type(_)
                    | Builtin::Func(_)
                    | Builtin::Index(_)
                    | Builtin::Connection(_) => continue,
                };
                // A shard-backed builtin registers `desc` as its persist shard's
                // schema, so a type change in `desc` needs a `MigrationStep`. Tables
                // and sources carry `desc` into the catalog verbatim and cannot fail
                // here. A materialized view is planned, and its fingerprint is its
                // SQL, so nothing else detects a `desc` change at upgrade. See the
                // `Fingerprint` impl for `BuiltinMaterializedView`.
                let migration_hint = if shard_backed {
                    format!(
                        "; {schema}.{name} is backed by a persist shard, so after updating its \
                         `desc` also add a `MigrationStep` for it to `MIGRATIONS` in \
                         `src/adapter/src/catalog/open/builtin_schema_migration.rs`"
                    )
                } else {
                    String::new()
                };
                let item = conn_catalog
                    .resolve_item(&PartialItemName {
                        database: None,
                        schema: Some(schema.to_string()),
                        item: name.to_string(),
                    })
                    .expect("unable to resolve item")
                    .at_version(RelationVersionSelector::Latest);

                let actual_desc = item.relation_desc().expect("invalid item type");
                assert_eq!(
                    actual_desc.arity(),
                    expected_desc.arity(),
                    "item {schema}.{name} column count did not match its expected column count{migration_hint}"
                );
                for (index, ((actual_name, actual_typ), (expected_name, expected_typ))) in
                    actual_desc.iter().zip_eq(expected_desc.iter()).enumerate()
                {
                    assert_eq!(
                        actual_name, expected_name,
                        "item {schema}.{name} column {index} name did not match its expected name{migration_hint}"
                    );
                    assert_eq!(
                        actual_typ, expected_typ,
                        "item {schema}.{name} column {index} ('{actual_name}') type did not match its expected type{migration_hint}"
                    );
                }
                // Keys never reach the shard schema, so a key mismatch needs no
                // migration, only a corrected `desc`.
                assert_eq!(
                    actual_desc.typ().keys,
                    expected_desc.typ().keys,
                    "item {schema}.{name} keys did not match its expected keys"
                );
                assert_eq!(
                    &*actual_desc, expected_desc,
                    "item {schema}.{name} did not match its expected RelationDesc"
                );
            }
            catalog.expire().await;
        })
        .await
    }

    // Connect to a running Postgres server and verify that our builtin
    // types and functions match it, in addition to some other things.
    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] //  unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
    async fn test_compare_builtins_postgres() {
        async fn inner(catalog: Catalog) {
            // Verify that all builtin functions:
            // - have a unique OID
            // - if they have a postgres counterpart (same oid) then they have matching name
            let (client, connection) = tokio_postgres::connect(
                &env::var("POSTGRES_URL").unwrap_or_else(|_| "host=localhost user=postgres".into()),
                NoTls,
            )
            .await
            .expect("failed to connect to Postgres");

            task::spawn(|| "compare_builtin_postgres", async move {
                if let Err(e) = connection.await {
                    panic!("connection error: {}", e);
                }
            });

            struct PgProc {
                name: String,
                arg_oids: Vec<u32>,
                ret_oid: Option<u32>,
                ret_set: bool,
            }

            struct PgType {
                name: String,
                ty: String,
                elem: u32,
                array: u32,
                input: u32,
                receive: u32,
                send: u32,
                len: i16,
            }

            struct PgOper {
                oprresult: u32,
                name: String,
            }

            let pg_proc: BTreeMap<_, _> = query(
                &client,
                sql!(
                    "SELECT
                    p.oid,
                    proname,
                    proargtypes,
                    prorettype,
                    proretset
                FROM pg_proc p
                JOIN pg_namespace n ON p.pronamespace = n.oid"
                ),
                &[],
            )
            .await
            .expect("pg query failed")
            .into_iter()
            .map(|row| {
                let oid: u32 = row.get("oid");
                let pg_proc = PgProc {
                    name: row.get("proname"),
                    arg_oids: row.get("proargtypes"),
                    ret_oid: row.get("prorettype"),
                    ret_set: row.get("proretset"),
                };
                (oid, pg_proc)
            })
            .collect();

            let pg_type: BTreeMap<_, _> = query(
                &client,
                sql!(
                    "SELECT oid, typname, typtype::text, typelem, typarray, typinput::oid, typreceive::oid as typreceive, typsend::oid as typsend, typlen FROM pg_type"
                ),
                &[],
            )
            .await
                .expect("pg query failed")
                .into_iter()
                .map(|row| {
                    let oid: u32 = row.get("oid");
                    let pg_type = PgType {
                        name: row.get("typname"),
                        ty: row.get("typtype"),
                        elem: row.get("typelem"),
                        array: row.get("typarray"),
                        input: row.get("typinput"),
                        receive: row.get("typreceive"),
                        send: row.get("typsend"),
                        len: row.get("typlen"),
                    };
                    (oid, pg_type)
                })
                .collect();

            let pg_oper: BTreeMap<_, _> = query(
                &client,
                sql!("SELECT oid, oprname, oprresult FROM pg_operator"),
                &[],
            )
            .await
            .expect("pg query failed")
            .into_iter()
            .map(|row| {
                let oid: u32 = row.get("oid");
                let pg_oper = PgOper {
                    name: row.get("oprname"),
                    oprresult: row.get("oprresult"),
                };
                (oid, pg_oper)
            })
            .collect();

            let conn_catalog = catalog.for_system_session();
            let resolve_type_oid = |item: &str| {
                conn_catalog
                    .resolve_type(&PartialItemName {
                        database: None,
                        // All functions we check exist in PG, so the types must, as
                        // well
                        schema: Some(PG_CATALOG_SCHEMA.into()),
                        item: item.to_string(),
                    })
                    .expect("unable to resolve type")
                    .oid()
            };

            let func_oids: BTreeSet<_> = BUILTINS::funcs()
                .flat_map(|f| f.inner.func_impls().into_iter().map(|f| f.oid))
                .collect();

            let mut all_oids = BTreeSet::new();

            // A function to determine if two oids are equivalent enough for these tests. We don't
            // support some types, so map exceptions here.
            let equivalent_types: BTreeSet<(Option<u32>, Option<u32>)> = BTreeSet::from_iter(
                [
                    // We don't support NAME.
                    (Type::NAME, Type::TEXT),
                    (Type::NAME_ARRAY, Type::TEXT_ARRAY),
                    // We don't support time with time zone.
                    (Type::TIME, Type::TIMETZ),
                    (Type::TIME_ARRAY, Type::TIMETZ_ARRAY),
                ]
                .map(|(a, b)| (Some(a.oid()), Some(b.oid()))),
            );
            let ignore_return_types: BTreeSet<u32> = BTreeSet::from([
                1619, // pg_typeof: TODO: We now have regtype and can correctly implement this.
            ]);
            let is_same_type = |fn_oid: u32, a: Option<u32>, b: Option<u32>| -> bool {
                if ignore_return_types.contains(&fn_oid) {
                    return true;
                }
                if equivalent_types.contains(&(a, b)) || equivalent_types.contains(&(b, a)) {
                    return true;
                }
                a == b
            };

            for builtin in BUILTINS::iter() {
                match builtin {
                    Builtin::Type(ty) => {
                        assert!(all_oids.insert(ty.oid), "{} reused oid {}", ty.name, ty.oid);

                        if ty.oid >= FIRST_MATERIALIZE_OID {
                            // High OIDs are reserved in Materialize and don't have
                            // PostgreSQL counterparts.
                            continue;
                        }

                        // For types that have a PostgreSQL counterpart, verify that
                        // the name and oid match.
                        let pg_ty = pg_type.get(&ty.oid).unwrap_or_else(|| {
                            panic!("pg_proc missing type {}: oid {}", ty.name, ty.oid)
                        });
                        assert_eq!(
                            ty.name, pg_ty.name,
                            "oid {} has name {} in postgres; expected {}",
                            ty.oid, pg_ty.name, ty.name,
                        );

                        let (typinput_oid, typreceive_oid, typsend_oid) =
                            match &ty.details.pg_metadata {
                                None => (0, 0, 0),
                                Some(pgmeta) => (
                                    pgmeta.typinput_oid,
                                    pgmeta.typreceive_oid,
                                    pgmeta.typsend_oid,
                                ),
                            };
                        assert_eq!(
                            typinput_oid, pg_ty.input,
                            "type {} has typinput OID {:?} in mz but {:?} in pg",
                            ty.name, typinput_oid, pg_ty.input,
                        );
                        assert_eq!(
                            typreceive_oid, pg_ty.receive,
                            "type {} has typreceive OID {:?} in mz but {:?} in pg",
                            ty.name, typreceive_oid, pg_ty.receive,
                        );
                        // Unlike typinput and typreceive below, typsend is not also
                        // checked against `func_oids`. Nothing resolves a typsend OID
                        // to a name, so the corresponding `*send` functions are
                        // deliberately not registered as builtins.
                        assert_eq!(
                            typsend_oid, pg_ty.send,
                            "type {} has typsend OID {:?} in mz but {:?} in pg",
                            ty.name, typsend_oid, pg_ty.send,
                        );
                        if typinput_oid != 0 {
                            assert!(
                                func_oids.contains(&typinput_oid),
                                "type {} has typinput OID {} that does not exist in pg_proc",
                                ty.name,
                                typinput_oid,
                            );
                        }
                        if typreceive_oid != 0 {
                            assert!(
                                func_oids.contains(&typreceive_oid),
                                "type {} has typreceive OID {} that does not exist in pg_proc",
                                ty.name,
                                typreceive_oid,
                            );
                        }
                        let id = conn_catalog
                            .resolve_type(&PartialItemName {
                                database: None,
                                schema: Some(ty.schema.into()),
                                item: ty.name.to_string(),
                            })
                            .expect("unable to resolve type")
                            .id();
                        // Types with no scalar type report an unknown typlen.
                        if let Ok(scalar_type) =
                            mz_sql::plan::scalar_type_from_catalog(&conn_catalog, id, &[])
                        {
                            let typlen = mz_pgrepr::Type::from(&scalar_type).typlen();
                            assert_eq!(
                                typlen, pg_ty.len,
                                "type {} has typlen {} in mz but {} in pg",
                                ty.name, typlen, pg_ty.len,
                            );
                        }

                        // Ensure the type matches.
                        match &ty.details.typ {
                            CatalogType::Array { element_reference } => {
                                let elem_ty = BUILTINS::iter()
                                    .filter_map(|builtin| match builtin {
                                        Builtin::Type(ty @ BuiltinType { name, .. })
                                            if element_reference == name =>
                                        {
                                            Some(ty)
                                        }
                                        _ => None,
                                    })
                                    .next();
                                let elem_ty = match elem_ty {
                                    Some(ty) => ty,
                                    None => {
                                        panic!("{} is unexpectedly not a type", element_reference)
                                    }
                                };
                                assert_eq!(
                                    pg_ty.elem, elem_ty.oid,
                                    "type {} has mismatched element OIDs",
                                    ty.name
                                )
                            }
                            CatalogType::Pseudo => {
                                assert_eq!(
                                    pg_ty.ty, "p",
                                    "type {} is not a pseudo type as expected",
                                    ty.name
                                )
                            }
                            CatalogType::Range { .. } => {
                                assert_eq!(
                                    pg_ty.ty, "r",
                                    "type {} is not a range type as expected",
                                    ty.name
                                );
                            }
                            _ => {
                                assert_eq!(
                                    pg_ty.ty, "b",
                                    "type {} is not a base type as expected",
                                    ty.name
                                )
                            }
                        }

                        // Ensure the array type reference is correct.
                        let schema = catalog
                            .resolve_schema_in_database(
                                &ResolvedDatabaseSpecifier::Ambient,
                                ty.schema,
                                &SYSTEM_CONN_ID,
                            )
                            .expect("unable to resolve schema");
                        let allocated_type = catalog
                            .resolve_type(
                                None,
                                &vec![(ResolvedDatabaseSpecifier::Ambient, schema.id().clone())],
                                &PartialItemName {
                                    database: None,
                                    schema: Some(schema.name().schema.clone()),
                                    item: ty.name.to_string(),
                                },
                                &SYSTEM_CONN_ID,
                            )
                            .expect("unable to resolve type");
                        let ty = if let CatalogItem::Type(ty) = &allocated_type.item {
                            ty
                        } else {
                            panic!("unexpectedly not a type")
                        };
                        match ty.details.array_id {
                            Some(array_id) => {
                                let array_ty = catalog.get_entry(&array_id);
                                assert_eq!(
                                    pg_ty.array, array_ty.oid,
                                    "type {} has mismatched array OIDs",
                                    allocated_type.name.item,
                                );
                            }
                            None => assert_eq!(
                                pg_ty.array, 0,
                                "type {} does not have an array type in mz but does in pg",
                                allocated_type.name.item,
                            ),
                        }
                    }
                    Builtin::Func(func) => {
                        for imp in func.inner.func_impls() {
                            assert!(
                                all_oids.insert(imp.oid),
                                "{} reused oid {}",
                                func.name,
                                imp.oid
                            );

                            assert!(
                                imp.oid < FIRST_USER_OID,
                                "built-in function {} erroneously has OID in user space ({})",
                                func.name,
                                imp.oid,
                            );

                            // For functions that have a postgres counterpart, verify that the name and
                            // oid match.
                            let pg_fn = if imp.oid >= FIRST_UNPINNED_OID {
                                continue;
                            } else {
                                pg_proc.get(&imp.oid).unwrap_or_else(|| {
                                    panic!(
                                        "pg_proc missing function {}: oid {}",
                                        func.name, imp.oid
                                    )
                                })
                            };
                            assert_eq!(
                                func.name, pg_fn.name,
                                "funcs with oid {} don't match names: {} in mz, {} in pg",
                                imp.oid, func.name, pg_fn.name
                            );

                            // Complain, but don't fail, if argument oids don't match.
                            // TODO: make these match.
                            let imp_arg_oids = imp
                                .arg_typs
                                .iter()
                                .map(|item| resolve_type_oid(item))
                                .collect::<Vec<_>>();

                            if imp_arg_oids != pg_fn.arg_oids {
                                println!(
                                    "funcs with oid {} ({}) don't match arguments: {:?} in mz, {:?} in pg",
                                    imp.oid, func.name, imp_arg_oids, pg_fn.arg_oids
                                );
                            }

                            let imp_return_oid = imp.return_typ.map(resolve_type_oid);

                            assert!(
                                is_same_type(imp.oid, imp_return_oid, pg_fn.ret_oid),
                                "funcs with oid {} ({}) don't match return types: {:?} in mz, {:?} in pg",
                                imp.oid,
                                func.name,
                                imp_return_oid,
                                pg_fn.ret_oid
                            );

                            assert_eq!(
                                imp.return_is_set, pg_fn.ret_set,
                                "funcs with oid {} ({}) don't match set-returning value: {:?} in mz, {:?} in pg",
                                imp.oid, func.name, imp.return_is_set, pg_fn.ret_set
                            );
                        }
                    }
                    _ => (),
                }
            }

            for (op, func) in OP_IMPLS.iter() {
                for imp in func.func_impls() {
                    assert!(all_oids.insert(imp.oid), "{} reused oid {}", op, imp.oid);

                    // For operators that have a postgres counterpart, verify that the name and oid match.
                    let pg_op = if imp.oid >= FIRST_UNPINNED_OID {
                        continue;
                    } else {
                        pg_oper.get(&imp.oid).unwrap_or_else(|| {
                            panic!("pg_operator missing operator {}: oid {}", op, imp.oid)
                        })
                    };

                    assert_eq!(*op, pg_op.name);

                    let imp_return_oid =
                        imp.return_typ.map(resolve_type_oid).expect("must have oid");
                    if imp_return_oid != pg_op.oprresult {
                        panic!(
                            "operators with oid {} ({}) don't match return typs: {} in mz, {} in pg",
                            imp.oid, op, imp_return_oid, pg_op.oprresult
                        );
                    }
                }
            }
            catalog.expire().await;
        }

        Catalog::with_debug(inner).await
    }

    // Execute all builtin functions with all combinations of arguments from interesting datums.
    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] //  unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
    async fn test_smoketest_all_builtins() {
        fn inner(catalog: Catalog) -> Vec<mz_ore::task::JoinHandle<()>> {
            let catalog = Arc::new(catalog);
            let conn_catalog = catalog.for_system_session();

            let resolve_type_oid = |item: &str| conn_catalog.state().get_system_type(item).oid();
            let mut handles = Vec::new();

            // Extracted during planning; always panics when executed.
            let ignore_names = BTreeSet::from([
                "avg",
                "avg_internal_v1",
                "bool_and",
                "bool_or",
                "has_table_privilege", // > 3 s each
                "has_type_privilege",  // > 3 s each
                "mod",
                "mz_panic",
                "mz_sleep",
                "pow",
                "stddev_pop",
                "stddev_samp",
                "stddev",
                "var_pop",
                "var_samp",
                "variance",
            ]);

            let fns = BUILTINS::funcs()
                .map(|func| (&func.name, func.inner))
                .chain(OP_IMPLS.iter());

            for (name, func) in fns {
                if ignore_names.contains(name) {
                    continue;
                }
                let Func::Scalar(impls) = func else {
                    continue;
                };

                'outer: for imp in impls {
                    let details = imp.details();
                    let mut styps = Vec::new();
                    for item in details.arg_typs.iter() {
                        let oid = resolve_type_oid(item);
                        let Ok(pgtyp) = mz_pgrepr::Type::from_oid(oid) else {
                            continue 'outer;
                        };
                        styps.push(SqlScalarType::try_from(&pgtyp).expect("must exist"));
                    }
                    let datums = styps
                        .iter()
                        .map(|styp| {
                            let mut datums = vec![Datum::Null];
                            datums.extend(styp.interesting_datums());
                            datums
                        })
                        .collect::<Vec<_>>();
                    // Skip nullary fns.
                    if datums.is_empty() {
                        continue;
                    }

                    let return_oid = details
                        .return_typ
                        .map(resolve_type_oid)
                        .expect("must exist");
                    let return_styp = mz_pgrepr::Type::from_oid(return_oid)
                        .ok()
                        .map(|typ| SqlScalarType::try_from(&typ).expect("must exist"));

                    let mut idxs = vec![0; datums.len()];
                    while idxs[0] < datums[0].len() {
                        let mut args = Vec::with_capacity(idxs.len());
                        for i in 0..(datums.len()) {
                            args.push(datums[i][idxs[i]]);
                        }

                        let op = &imp.op;
                        let scalars = args
                            .iter()
                            .enumerate()
                            .map(|(i, datum)| {
                                CoercibleScalarExpr::Coerced(HirScalarExpr::literal(
                                    datum.clone(),
                                    styps[i].clone(),
                                ))
                            })
                            .collect();

                        let call_name = format!(
                            "{name}({}) (oid: {})",
                            args.iter()
                                .map(|d| d.to_string())
                                .collect::<Vec<_>>()
                                .join(", "),
                            imp.oid
                        );
                        let catalog = Arc::clone(&catalog);
                        let call_name_fn = call_name.clone();
                        let return_styp = return_styp.clone();
                        let handle = task::spawn_blocking(
                            || call_name,
                            move || {
                                smoketest_fn(
                                    name,
                                    call_name_fn,
                                    op,
                                    imp,
                                    args,
                                    catalog,
                                    scalars,
                                    return_styp,
                                )
                            },
                        );
                        handles.push(handle);

                        // Advance to the next datum combination.
                        for i in (0..datums.len()).rev() {
                            idxs[i] += 1;
                            if idxs[i] >= datums[i].len() {
                                if i == 0 {
                                    break;
                                }
                                idxs[i] = 0;
                                continue;
                            } else {
                                break;
                            }
                        }
                    }
                }
            }
            handles
        }

        let handles = Catalog::with_debug(|catalog| async { inner(catalog) }).await;
        for handle in handles {
            handle.await;
        }
    }

    fn smoketest_fn(
        name: &&str,
        call_name: String,
        op: &Operation<HirScalarExpr>,
        imp: &FuncImpl<HirScalarExpr>,
        args: Vec<Datum<'_>>,
        catalog: Arc<Catalog>,
        scalars: Vec<CoercibleScalarExpr>,
        return_styp: Option<SqlScalarType>,
    ) {
        let conn_catalog = catalog.for_system_session();
        let pcx = PlanContext::zero();
        let scx = StatementContext::new(Some(&pcx), &conn_catalog);
        let qcx = QueryContext::root(&scx, QueryLifetime::OneShot);
        let ecx = ExprContext {
            qcx: &qcx,
            name: "smoketest",
            scope: &Scope::empty(),
            relation_type: &SqlRelationType::empty(),
            allow_aggregates: false,
            allow_subqueries: false,
            allow_parameters: false,
            allow_windows: false,
        };
        let arena = RowArena::new();
        let mut session = Session::dummy();
        session
            .start_transaction(to_datetime(0), None, None)
            .expect("must succeed");
        let prep_style = ExprPrepOneShot {
            logical_time: EvalTime::Time(Timestamp::MIN),
            session: &session,
            catalog_state: &catalog.state,
        };

        // Execute the function as much as possible, ensuring no panics occur, but
        // otherwise ignoring eval errors. We also do various other checks.
        let res = (op.0)(&ecx, scalars, &imp.params, vec![]);
        if let Ok(hir) = res {
            let uneliminated_result_row = {
                if let HirScalarExpr::CallUnary { func, .. } = &hir
                    && func.is_eliminable_cast()
                {
                    let mut uneliminated_mir = hir
                        .clone()
                        .lower_uncorrelated(HirToMirConfig {
                            enable_cast_elimination: false,
                            ..catalog.system_config().into()
                        })
                        .expect("lowering eliminable cast should always succeed");
                    prep_style
                        .prep_scalar_expr(&mut uneliminated_mir)
                        .expect("must succeed");

                    // Pack the row, to avoid lifetime issues with the MIR we lowered here
                    uneliminated_mir
                        .eval(&[], &arena)
                        .ok()
                        .map(|datum| Row::pack([datum]))
                } else {
                    None
                }
            };

            if let Ok(mut mir) = hir.lower_uncorrelated(catalog.system_config()) {
                // Populate unmaterialized functions.
                prep_style.prep_scalar_expr(&mut mir).expect("must succeed");

                if let Ok(eval_result_datum) = mir.eval(&[], &arena) {
                    if let Some(return_styp) = return_styp {
                        let mir_typ = mir.typ(&[]);
                        // MIR type inference should be consistent with the type
                        // we get from the catalog.
                        soft_assert_eq_or_log!(
                            mir_typ.scalar_type,
                            (&return_styp).into(),
                            "MIR type did not match the catalog type (cast elimination/repr type error)"
                        );
                        // The following will check not just that the scalar type
                        // is ok, but also catches if the function returned a null
                        // but the MIR type inference said "non-nullable".
                        if !eval_result_datum.is_instance_of(&mir_typ) {
                            panic!(
                                "{call_name}: expected return type of {return_styp:?}, got {eval_result_datum}"
                            );
                        }
                        // Check the consistency of `is_eliminable_cast`---we should get the same datum either way.
                        if let Some(row) = uneliminated_result_row {
                            let uneliminated_result_datum = row.unpack_first();
                            assert_eq!(
                                uneliminated_result_datum, eval_result_datum,
                                "datums should not change if cast is eliminable"
                            );
                        }
                        // Check the consistency of `introduces_nulls` and
                        // `propagates_nulls` with `MirScalarExpr::typ`.
                        if let Some((introduces_nulls, propagates_nulls)) =
                            call_introduces_propagates_nulls(&mir)
                        {
                            if introduces_nulls {
                                // If the function introduces_nulls, then the return
                                // type should always be nullable, regardless of
                                // the nullability of the input types.
                                assert!(
                                    mir_typ.nullable,
                                    "fn named `{}` called on args `{:?}` (lowered to `{}`) yielded mir_typ.nullable: {}",
                                    name, args, mir, mir_typ.nullable
                                );
                            } else {
                                let any_input_null = args.iter().any(|arg| arg.is_null());
                                if !any_input_null {
                                    assert!(
                                        !mir_typ.nullable,
                                        "fn named `{}` called on args `{:?}` (lowered to `{}`) yielded mir_typ.nullable: {}",
                                        name, args, mir, mir_typ.nullable
                                    );
                                } else if propagates_nulls {
                                    // propagates_nulls means the optimizer short-circuits
                                    // all-null inputs, so the output must be nullable.
                                    assert!(
                                        mir_typ.nullable,
                                        "fn named `{}` called on args `{:?}` (lowered to `{}`) yielded mir_typ.nullable: {}",
                                        name, args, mir, mir_typ.nullable
                                    );
                                }
                                // When propagates_nulls is false, the output may still
                                // be nullable if a non-nullable parameter received a null
                                // input (per-position null rejection). The is_instance_of
                                // check above ensures type consistency.
                            }
                        }
                        // Check that `MirScalarExpr::reduce` yields the same result
                        // as the real evaluation.
                        let mut reduced = mir.clone();
                        reduced.reduce(&[]);
                        match reduced {
                            MirScalarExpr::Literal(reduce_result, ctyp) => {
                                match reduce_result {
                                    Ok(reduce_result_row) => {
                                        let reduce_result_datum = reduce_result_row.unpack_first();
                                        assert_eq!(
                                            reduce_result_datum,
                                            eval_result_datum,
                                            "eval/reduce datum mismatch: fn named `{}` called on args `{:?}` (lowered to `{}`) evaluated to `{}` with typ `{:?}`, but reduced to `{}` with typ `{:?}`",
                                            name,
                                            args,
                                            mir,
                                            eval_result_datum,
                                            mir_typ.scalar_type,
                                            reduce_result_datum,
                                            ctyp.scalar_type
                                        );
                                        // Let's check that the types also match.
                                        // (We are not checking nullability here,
                                        // because it's ok when we know a more
                                        // precise nullability after actually
                                        // evaluating a function than before.)
                                        assert_eq!(
                                            ctyp.scalar_type,
                                            mir_typ.scalar_type,
                                            "eval/reduce type mismatch: fn named `{}` called on args `{:?}` (lowered to `{}`) evaluated to `{}` with typ `{:?}`, but reduced to `{}` with typ `{:?}`",
                                            name,
                                            args,
                                            mir,
                                            eval_result_datum,
                                            mir_typ.scalar_type,
                                            reduce_result_datum,
                                            ctyp.scalar_type
                                        );
                                    }
                                    Err(..) => {} // It's ok, we might have given invalid args to the function
                                }
                            }
                            _ => unreachable!(
                                "all args are literals, so should have reduced to a literal"
                            ),
                        }
                    }
                }
            }
        }
    }

    /// If the given MirScalarExpr
    ///  - is a function call, and
    ///  - all arguments are literals
    /// then it returns whether the called function (introduces_nulls, propagates_nulls).
    fn call_introduces_propagates_nulls(mir_func_call: &MirScalarExpr) -> Option<(bool, bool)> {
        match mir_func_call {
            MirScalarExpr::CallUnary { func, expr } => {
                if expr.is_literal() {
                    Some((func.introduces_nulls(), func.propagates_nulls()))
                } else {
                    None
                }
            }
            MirScalarExpr::CallBinary { func, expr1, expr2 } => {
                if expr1.is_literal() && expr2.is_literal() {
                    Some((func.introduces_nulls(), func.propagates_nulls()))
                } else {
                    None
                }
            }
            MirScalarExpr::CallVariadic { func, exprs } => {
                if exprs.iter().all(|arg| arg.is_literal()) {
                    Some((func.introduces_nulls(), func.propagates_nulls()))
                } else {
                    None
                }
            }
            _ => None,
        }
    }

    // Make sure pg views don't use types that only exist in Materialize.
    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] //  unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
    async fn test_pg_views_forbidden_types() {
        Catalog::with_debug(|catalog| async move {
            let conn_catalog = catalog.for_system_session();

            for view in BUILTINS::views().filter(|view| {
                view.schema == PG_CATALOG_SCHEMA || view.schema == INFORMATION_SCHEMA
            }) {
                let item = conn_catalog
                    .resolve_item(&PartialItemName {
                        database: None,
                        schema: Some(view.schema.to_string()),
                        item: view.name.to_string(),
                    })
                    .expect("unable to resolve view")
                    // TODO(alter_table)
                    .at_version(RelationVersionSelector::Latest);
                let full_name = conn_catalog.resolve_full_name(item.name());
                let desc = item.relation_desc().expect("invalid item type");
                for col_type in desc.iter_types() {
                    match &col_type.scalar_type {
                        typ @ SqlScalarType::UInt16
                        | typ @ SqlScalarType::UInt32
                        | typ @ SqlScalarType::UInt64
                        | typ @ SqlScalarType::MzTimestamp
                        | typ @ SqlScalarType::List { .. }
                        | typ @ SqlScalarType::Map { .. }
                        | typ @ SqlScalarType::MzAclItem => {
                            panic!("{typ:?} type found in {full_name}");
                        }
                        SqlScalarType::AclItem
                        | SqlScalarType::Bool
                        | SqlScalarType::Int16
                        | SqlScalarType::Int32
                        | SqlScalarType::Int64
                        | SqlScalarType::Float32
                        | SqlScalarType::Float64
                        | SqlScalarType::Numeric { .. }
                        | SqlScalarType::Date
                        | SqlScalarType::Time
                        | SqlScalarType::Timestamp { .. }
                        | SqlScalarType::TimestampTz { .. }
                        | SqlScalarType::Interval
                        | SqlScalarType::PgLegacyChar
                        | SqlScalarType::Bytes
                        | SqlScalarType::String
                        | SqlScalarType::Char { .. }
                        | SqlScalarType::VarChar { .. }
                        | SqlScalarType::Jsonb
                        | SqlScalarType::Uuid
                        | SqlScalarType::Array(_)
                        | SqlScalarType::Record { .. }
                        | SqlScalarType::Oid
                        | SqlScalarType::RegProc
                        | SqlScalarType::RegType
                        | SqlScalarType::RegClass
                        | SqlScalarType::Int2Vector
                        | SqlScalarType::Range { .. }
                        | SqlScalarType::PgLegacyName => {}
                    }
                }
            }
            catalog.expire().await;
        })
        .await
    }

    // Make sure objects reside in the `mz_introspection` schema iff they depend on per-replica
    // introspection relations.
    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] // unsupported operation: returning ready events from epoll_wait is not yet implemented
    async fn test_mz_introspection_builtins() {
        Catalog::with_debug(|catalog| async move {
            let conn_catalog = catalog.for_system_session();

            let introspection_schema_id = catalog.get_mz_introspection_schema_id();
            let introspection_schema_spec = SchemaSpecifier::Id(introspection_schema_id);

            for entry in catalog.entries() {
                let schema_spec = entry.name().qualifiers.schema_spec;
                let introspection_deps = catalog.introspection_dependencies(entry.id);
                if introspection_deps.is_empty() {
                    assert!(
                        schema_spec != introspection_schema_spec,
                        "entry does not depend on introspection sources but is in \
                         `mz_introspection`: {}",
                        conn_catalog.resolve_full_name(entry.name()),
                    );
                } else {
                    assert!(
                        schema_spec == introspection_schema_spec,
                        "entry depends on introspection sources but is not in \
                         `mz_introspection`: {}",
                        conn_catalog.resolve_full_name(entry.name()),
                    );
                }
            }
        })
        .await
    }

    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] //  unsupported operation: can't call foreign function `TLS_client_method` on OS `linux`
    async fn test_multi_subscriber_catalog() {
        let persist_client = PersistClient::new_for_tests().await;
        let bootstrap_args = test_bootstrap_args();
        let organization_id = Uuid::new_v4();
        let db_name = "DB";

        let mut writer_catalog = Catalog::open_debug_catalog(
            persist_client.clone(),
            organization_id.clone(),
            &bootstrap_args,
        )
        .await
        .expect("open_debug_catalog");
        let mut read_only_catalog = Catalog::open_debug_read_only_catalog(
            persist_client.clone(),
            organization_id.clone(),
            &bootstrap_args,
        )
        .await
        .expect("open_debug_read_only_catalog");
        assert_err!(writer_catalog.resolve_database(db_name));
        assert_err!(read_only_catalog.resolve_database(db_name));

        let commit_ts = writer_catalog.current_upper().await;
        writer_catalog
            .transact(
                None,
                commit_ts,
                None,
                vec![Op::CreateDatabase {
                    name: db_name.to_string(),
                    owner_id: MZ_SYSTEM_ROLE_ID,
                }],
            )
            .await
            .expect("failed to transact");

        let write_db = writer_catalog
            .resolve_database(db_name)
            .expect("resolve_database");
        read_only_catalog
            .sync_to_current_updates()
            .await
            .expect("sync_to_current_updates");
        let read_db = read_only_catalog
            .resolve_database(db_name)
            .expect("resolve_database");

        assert_eq!(write_db, read_db);

        let writer_catalog_fencer =
            Catalog::open_debug_catalog(persist_client, organization_id, &bootstrap_args)
                .await
                .expect("open_debug_catalog for fencer");
        let fencer_db = writer_catalog_fencer
            .resolve_database(db_name)
            .expect("resolve_database for fencer");
        assert_eq!(fencer_db, read_db);

        let write_fence_err = writer_catalog
            .sync_to_current_updates()
            .await
            .expect_err("sync_to_current_updates for fencer");
        assert!(matches!(
            write_fence_err,
            CatalogError::Durable(DurableCatalogError::Fence(FenceError::Epoch { .. }))
        ));
        let read_fence_err = read_only_catalog
            .sync_to_current_updates()
            .await
            .expect_err("sync_to_current_updates after fencer");
        assert!(matches!(
            read_fence_err,
            CatalogError::Durable(DurableCatalogError::Fence(FenceError::Epoch { .. }))
        ));

        writer_catalog.expire().await;
        read_only_catalog.expire().await;
        writer_catalog_fencer.expire().await;
    }
}