Skip to main content

mz_adapter/catalog/
state.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! In-memory metadata storage for the coordinator.
11
12use std::borrow::Cow;
13use std::collections::{BTreeMap, BTreeSet, VecDeque};
14use std::fmt::Debug;
15use std::sync::Arc;
16use std::sync::LazyLock;
17use std::time::Instant;
18
19use ipnet::IpNet;
20use itertools::Itertools;
21use mz_adapter_types::compaction::CompactionWindow;
22use mz_adapter_types::connection::ConnectionId;
23use mz_audit_log::{EventDetails, EventType, ObjectType, VersionedEvent};
24use mz_build_info::DUMMY_BUILD_INFO;
25use mz_catalog::SYSTEM_CONN_ID;
26use mz_catalog::builtin::{
27    BUILTINS, Builtin, BuiltinCluster, BuiltinLog, BuiltinSource, BuiltinTable, BuiltinType,
28};
29use mz_catalog::config::{AwsPrincipalContext, ClusterReplicaSizeMap};
30use mz_catalog::expr_cache::{LocalExpressions, latest_item_version};
31use mz_catalog::memory::error::{Error, ErrorKind};
32use mz_catalog::memory::objects::{
33    CatalogCollectionEntry, CatalogEntry, CatalogItem, Cluster, ClusterReplica, CommentsMap,
34    Connection, DataSourceDesc, Database, DefaultPrivileges, Index, MaterializedView, MetricSink,
35    NetworkPolicy, Role, RoleAuth, Schema, Secret, Sink, Source, SourceReferences, Table,
36    TableDataSource, Type, View,
37};
38use mz_controller::clusters::{
39    ManagedReplicaLocation, ReplicaAllocation, ReplicaLocation, UnmanagedReplicaLocation,
40};
41use mz_controller_types::{ClusterId, ReplicaId};
42use mz_expr::{CollectionPlan, MirScalarExpr, OptimizedMirRelationExpr};
43use mz_license_keys::ValidatedLicenseKey;
44use mz_ore::collections::CollectionExt;
45use mz_ore::now::NOW_ZERO;
46use mz_ore::soft_assert_no_log;
47use mz_ore::str::StrExt;
48use mz_pgrepr::oid::INVALID_OID;
49use mz_repr::adt::mz_acl_item::PrivilegeMap;
50use mz_repr::namespaces::{
51    INFORMATION_SCHEMA, MZ_CATALOG_SCHEMA, MZ_CATALOG_UNSTABLE_SCHEMA, MZ_INTERNAL_SCHEMA,
52    MZ_INTROSPECTION_SCHEMA, MZ_TEMP_SCHEMA, MZ_UNSAFE_SCHEMA, PG_CATALOG_SCHEMA, SYSTEM_SCHEMAS,
53    UNSTABLE_SCHEMAS,
54};
55use mz_repr::network_policy_id::NetworkPolicyId;
56use mz_repr::optimize::{OptimizerFeatureOverrides, OptimizerFeatures, OverrideFrom};
57use mz_repr::role_id::RoleId;
58use mz_repr::{
59    CatalogItemId, GlobalId, RelationDesc, RelationVersion, RelationVersionSelector,
60    ReprColumnType, SqlScalarType, VersionedRelationDesc,
61};
62use mz_secrets::InMemorySecretsController;
63use mz_sql::ast::Ident;
64use mz_sql::catalog::{
65    CatalogCluster, CatalogClusterReplica, CatalogDatabase, CatalogError as SqlCatalogError,
66    CatalogItem as SqlCatalogItem, CatalogItemType, CatalogRecordField, CatalogRole, CatalogSchema,
67    CatalogType, CatalogTypeDetails, IdReference, NameReference, SessionCatalog, SystemObjectType,
68    TypeReference,
69};
70use mz_sql::catalog::{CatalogConfig, EnvironmentId};
71use mz_sql::names::{
72    CommentObjectId, DatabaseId, DependencyIds, FullItemName, FullSchemaName, ObjectId,
73    PartialItemName, QualifiedItemName, QualifiedSchemaName, RawDatabaseSpecifier,
74    ResolvedDatabaseSpecifier, ResolvedIds, SchemaId, SchemaSpecifier, SystemObjectId,
75};
76use mz_sql::plan::{
77    CreateConnectionPlan, CreateIndexPlan, CreateMaterializedViewPlan, CreateMetricSinkPlan,
78    CreateSecretPlan, CreateSinkPlan, CreateSourcePlan, CreateTablePlan, CreateTypePlan,
79    CreateViewPlan, Params, Plan, PlanContext,
80};
81use mz_sql::rbac;
82use mz_sql::session::metadata::SessionMetadata;
83use mz_sql::session::user::MZ_SYSTEM_ROLE_ID;
84use mz_sql::session::vars::{DEFAULT_DATABASE_NAME, SystemVars, Var, VarInput};
85use mz_sql_parser::ast::QualifiedReplica;
86use mz_storage_client::controller::StorageMetadata;
87use mz_storage_types::connections::ConnectionContext;
88use mz_storage_types::connections::inline::{
89    ConnectionResolver, InlinedConnection, IntoInlineConnection,
90};
91use mz_transform::notice::OptimizerNotice;
92use serde::Serialize;
93use timely::progress::Antichain;
94use tokio::sync::mpsc;
95use tracing::{debug, warn};
96use uuid::Uuid;
97
98// DO NOT add any more imports from `crate` outside of `crate::catalog`.
99use crate::AdapterError;
100use crate::catalog::{Catalog, ConnCatalog};
101use crate::config::ScopedParameters;
102use crate::coord::{ConnMeta, infer_sql_type_for_catalog};
103use crate::optimize::{self, Optimize, OptimizerCatalog};
104use crate::session::Session;
105
106/// The in-memory representation of the Catalog. This struct is not directly used to persist
107/// metadata to persistent storage. For persistent metadata see
108/// [`mz_catalog::durable::DurableCatalogState`].
109///
110/// [`Serialize`] is implemented to create human readable dumps of the in-memory state, not for
111/// storing the contents of this struct on disk.
112#[derive(Debug, Clone, Serialize)]
113pub struct CatalogState {
114    // State derived from the durable catalog. These fields should only be mutated in `open.rs` or
115    // `apply.rs`. Some of these fields are not 100% derived from the durable catalog. Those
116    // include:
117    //  - Temporary items.
118    //  - Certain objects are partially derived from read-only state.
119    pub(super) database_by_name: imbl::OrdMap<String, DatabaseId>,
120    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
121    pub(super) database_by_id: imbl::OrdMap<DatabaseId, Database>,
122    #[serde(serialize_with = "skip_temp_items")]
123    pub(super) entry_by_id: imbl::OrdMap<CatalogItemId, CatalogEntry>,
124    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
125    pub(super) entry_by_global_id: imbl::OrdMap<GlobalId, CatalogItemId>,
126    pub(super) ambient_schemas_by_name: imbl::OrdMap<String, SchemaId>,
127    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
128    pub(super) ambient_schemas_by_id: imbl::OrdMap<SchemaId, Schema>,
129    pub(super) clusters_by_name: imbl::OrdMap<String, ClusterId>,
130    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
131    pub(super) clusters_by_id: imbl::OrdMap<ClusterId, Cluster>,
132    pub(super) roles_by_name: imbl::OrdMap<String, RoleId>,
133    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
134    pub(super) roles_by_id: imbl::OrdMap<RoleId, Role>,
135    pub(super) network_policies_by_name: imbl::OrdMap<String, NetworkPolicyId>,
136    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
137    pub(super) network_policies_by_id: imbl::OrdMap<NetworkPolicyId, NetworkPolicy>,
138    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
139    pub(super) role_auth_by_id: imbl::OrdMap<RoleId, RoleAuth>,
140
141    #[serde(skip)]
142    pub(super) system_configuration: Arc<SystemVars>,
143    /// In-memory mirror of the durable scoped (per-cluster and per-replica)
144    /// system-parameter cache, maintained by `apply.rs` from the durable
145    /// collections. Resolution reads from here: the optimizer's per-cluster
146    /// feature overrides (`cluster_scoped_optimizer_overrides`) and the
147    /// coordinator's per-replica dyncfg push. See the scoped feature flags
148    /// design. Skipped in the consistency-check snapshot because it is fully
149    /// derived from the durable catalog.
150    #[serde(skip)]
151    pub(super) scoped_system_parameters: ScopedParameters,
152    pub(super) default_privileges: Arc<DefaultPrivileges>,
153    pub(super) system_privileges: Arc<PrivilegeMap>,
154    pub(super) comments: Arc<CommentsMap>,
155    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
156    pub(super) source_references: imbl::OrdMap<CatalogItemId, SourceReferences>,
157    pub(super) storage_metadata: Arc<StorageMetadata>,
158    pub(super) mock_authentication_nonce: Option<String>,
159
160    // Mutable state not derived from the durable catalog. Populated
161    // during dataflow bootstrapping (`bootstrap_dataflow_plans`), which
162    // doesn't run in Testdrive's read-only consistency check, so this
163    // must be `#[serde(skip)]`.
164    #[serde(skip)]
165    pub(super) notices_by_dep_id: imbl::OrdMap<GlobalId, Vec<Arc<OptimizerNotice>>>,
166
167    // Populated by active connections creating temporary objects. The
168    // read-only catalog opened by Testdrive's consistency check has no
169    // active connections, so this must be `#[serde(skip)]`.
170    #[serde(skip)]
171    pub(super) temporary_namespaces: TemporaryNamespaces,
172
173    // Read-only state not derived from the durable catalog.
174    #[serde(skip)]
175    pub(super) config: mz_sql::catalog::CatalogConfig,
176    pub(super) cluster_replica_sizes: ClusterReplicaSizeMap,
177    #[serde(skip)]
178    pub(crate) availability_zones: Vec<String>,
179
180    // Read-only not derived from the durable catalog.
181    #[serde(skip)]
182    pub(super) egress_addresses: Vec<IpNet>,
183    pub(super) aws_principal_context: Option<AwsPrincipalContext>,
184    pub(super) aws_privatelink_availability_zones: Option<BTreeSet<String>>,
185    pub(super) http_host_name: Option<String>,
186
187    // Read-only not derived from the durable catalog.
188    #[serde(skip)]
189    pub(super) license_key: ValidatedLicenseKey,
190}
191
192/// The temporary namespaces of the sessions connected to this process: for
193/// each session that has created a temporary item, the ephemeral-owner
194/// mapping between the session's UUID (stamped on its durable temporary
195/// items) and its connection, together with the session's `mz_temp`
196/// [`Schema`] once it has materialized.
197///
198/// The coordinator registers a namespace at a session's first
199/// temporary-item creation, and unregisters it when the session terminates.
200#[derive(Debug, Clone, Default)]
201pub(super) struct TemporaryNamespaces {
202    by_conn: imbl::OrdMap<ConnectionId, TemporaryNamespace>,
203    // For resolving a durable item's owner UUID to its connection in
204    // the apply path.
205    conns_by_uuid: imbl::OrdMap<Uuid, ConnectionId>,
206}
207
208#[derive(Debug, Clone)]
209struct TemporaryNamespace {
210    uuid: Uuid,
211    // Instantiated by `ensure_schema` at the first applied temporary item
212    schema: Option<Schema>,
213}
214
215impl TemporaryNamespaces {
216    /// Registers `conn_id` as the connection of the session `uuid`.
217    ///
218    /// Callers guard on [`CatalogState::has_temporary_namespace`], so
219    /// registering an already-registered connection is a bug.
220    pub(super) fn register(&mut self, conn_id: ConnectionId, uuid: Uuid) {
221        let prev_conn = self.conns_by_uuid.insert(uuid, conn_id.clone());
222        mz_ore::soft_assert_or_log!(
223            prev_conn.is_none(),
224            "duplicate temporary namespace registration for {uuid}"
225        );
226        let prev_ns = self
227            .by_conn
228            .insert(conn_id, TemporaryNamespace { uuid, schema: None });
229        mz_ore::soft_assert_or_log!(
230            prev_ns.is_none(),
231            "duplicate temporary namespace registration for the connection of {uuid}"
232        );
233    }
234
235    /// Returns `conn_id`'s `mz_temp` schema, creating an empty one owned by
236    /// `owner_id` if one hasn't been instantiated yet.
237    pub(super) fn ensure_schema(
238        &mut self,
239        conn_id: &ConnectionId,
240        owner_id: RoleId,
241    ) -> &mut Schema {
242        let namespace = self
243            .by_conn
244            .get_mut(conn_id)
245            .expect("temporary namespace must be registered before items are applied");
246        namespace.schema.get_or_insert_with(|| {
247            // Temporary schema OIDs are never used, and it's therefore wasteful to go to the
248            // durable catalog to allocate a new OID for every temporary schema. Instead, we give
249            // them all the same invalid OID. This matches the semantics of temporary schema
250            // `GlobalId`s which are all -1.
251            let oid = INVALID_OID;
252            Schema {
253                name: QualifiedSchemaName {
254                    database: ResolvedDatabaseSpecifier::Ambient,
255                    schema: MZ_TEMP_SCHEMA.into(),
256                },
257                id: SchemaSpecifier::Temporary,
258                oid,
259                items: BTreeMap::new(),
260                functions: BTreeMap::new(),
261                types: BTreeMap::new(),
262                owner_id,
263                privileges: PrivilegeMap::from_mz_acl_items(vec![rbac::owner_privilege(
264                    mz_sql::catalog::ObjectType::Schema,
265                    owner_id,
266                )]),
267            }
268        })
269    }
270
271    /// Removes `conn_id`'s temporary namespace.
272    pub(super) fn unregister(&mut self, conn_id: &ConnectionId) {
273        let Some(namespace) = self.by_conn.get(conn_id) else {
274            return;
275        };
276        // A non-empty schema here means temporary items (and
277        // possibly their storage shards) weren't cleaned up, so we surface the
278        // invariant violation and keep the namespace registered.
279        if namespace
280            .schema
281            .as_ref()
282            .is_some_and(|schema| !schema.items.is_empty())
283        {
284            mz_ore::soft_panic_or_log!(
285                "temporary namespace for connection {conn_id} still has items at unregistration"
286            );
287            return;
288        }
289        let uuid = namespace.uuid;
290        self.by_conn.remove(conn_id);
291        self.conns_by_uuid.remove(&uuid);
292    }
293
294    pub(super) fn schema(&self, conn_id: &ConnectionId) -> Option<&Schema> {
295        self.by_conn
296            .get(conn_id)
297            .and_then(|namespace| namespace.schema.as_ref())
298    }
299
300    pub(super) fn schema_mut(&mut self, conn_id: &ConnectionId) -> Option<&mut Schema> {
301        self.by_conn
302            .get_mut(conn_id)
303            .and_then(|namespace| namespace.schema.as_mut())
304    }
305
306    pub(super) fn schemas(&self) -> impl Iterator<Item = &Schema> {
307        self.by_conn
308            .values()
309            .filter_map(|namespace| namespace.schema.as_ref())
310    }
311
312    pub(super) fn conn_for_uuid(&self, uuid: &Uuid) -> Option<&ConnectionId> {
313        self.conns_by_uuid.get(uuid)
314    }
315
316    pub(super) fn uuid_for_conn(&self, conn_id: &ConnectionId) -> Option<Uuid> {
317        self.by_conn.get(conn_id).map(|namespace| namespace.uuid)
318    }
319
320    pub(super) fn contains_conn(&self, conn_id: &ConnectionId) -> bool {
321        self.by_conn.contains_key(conn_id)
322    }
323
324    pub(super) fn contains_uuid(&self, uuid: &Uuid) -> bool {
325        self.conns_by_uuid.contains_key(uuid)
326    }
327}
328
329/// Keeps track of what expressions are cached or not during startup.
330/// It's also used during catalog transactions to avoid re-optimizing CREATE VIEW / CREATE MAT VIEW
331/// statements when going back and forth between durable catalog operations and in-memory catalog
332/// operations.
333#[derive(Debug, Clone, Serialize)]
334pub(crate) enum LocalExpressionCache {
335    /// The cache is being used.
336    Open {
337        /// The local expressions that were cached in the expression cache.
338        cached_exprs: BTreeMap<GlobalId, LocalExpressions>,
339        /// The local expressions that were NOT cached in the expression cache.
340        uncached_exprs: BTreeMap<GlobalId, LocalExpressions>,
341    },
342    /// The cache is not being used.
343    Closed,
344}
345
346impl LocalExpressionCache {
347    pub(super) fn new(cached_exprs: BTreeMap<GlobalId, LocalExpressions>) -> Self {
348        Self::Open {
349            cached_exprs,
350            uncached_exprs: BTreeMap::new(),
351        }
352    }
353
354    pub(super) fn remove_cached_expression(&mut self, id: &GlobalId) -> Option<LocalExpressions> {
355        match self {
356            LocalExpressionCache::Open { cached_exprs, .. } => cached_exprs.remove(id),
357            LocalExpressionCache::Closed => None,
358        }
359    }
360
361    /// Insert an expression that was cached, back into the cache. This is generally needed when
362    /// parsing/planning an expression fails, but we don't want to lose the cached expression.
363    pub(super) fn insert_cached_expression(
364        &mut self,
365        id: GlobalId,
366        local_expressions: LocalExpressions,
367    ) {
368        match self {
369            LocalExpressionCache::Open { cached_exprs, .. } => {
370                cached_exprs.insert(id, local_expressions);
371            }
372            LocalExpressionCache::Closed => {}
373        }
374    }
375
376    /// Inform the cache that `id` was not found in the cache and that we should add it as
377    /// `local_mir` and `optimizer_features`, recorded at `item_version`.
378    pub(super) fn insert_uncached_expression(
379        &mut self,
380        id: GlobalId,
381        local_mir: OptimizedMirRelationExpr,
382        optimizer_features: OptimizerFeatures,
383        item_version: RelationVersion,
384    ) {
385        match self {
386            LocalExpressionCache::Open { uncached_exprs, .. } => {
387                let local_expr = LocalExpressions {
388                    local_mir,
389                    optimizer_features,
390                    item_version,
391                };
392                // If we are trying to cache the same item a second time, with a different
393                // expression, then we must be migrating the object or doing something else weird.
394                // Caching the unmigrated expression may cause us to incorrectly use the unmigrated
395                // version after a restart. Caching the migrated version may cause us to incorrectly
396                // think that the object has already been migrated. To simplify things, we cache
397                // neither.
398                let prev = uncached_exprs.remove(&id);
399                match prev {
400                    Some(prev) if prev == local_expr => {
401                        uncached_exprs.insert(id, local_expr);
402                    }
403                    None => {
404                        uncached_exprs.insert(id, local_expr);
405                    }
406                    Some(_) => {}
407                }
408            }
409            LocalExpressionCache::Closed => {}
410        }
411    }
412
413    pub(super) fn into_uncached_exprs(self) -> BTreeMap<GlobalId, LocalExpressions> {
414        match self {
415            LocalExpressionCache::Open { uncached_exprs, .. } => uncached_exprs,
416            LocalExpressionCache::Closed => BTreeMap::new(),
417        }
418    }
419}
420
421fn skip_temp_items<S>(
422    entries: &imbl::OrdMap<CatalogItemId, CatalogEntry>,
423    serializer: S,
424) -> Result<S::Ok, S::Error>
425where
426    S: serde::Serializer,
427{
428    mz_ore::serde::map_key_to_string(
429        entries.iter().filter(|(_k, v)| v.conn_id().is_none()),
430        serializer,
431    )
432}
433
434impl CatalogState {
435    pub fn entries(&self) -> impl Iterator<Item = &CatalogEntry> {
436        self.entry_by_id.values()
437    }
438
439    pub fn user_connections(&self) -> impl Iterator<Item = &CatalogEntry> {
440        self.entries()
441            .filter(|entry| entry.is_connection() && entry.id().is_user())
442    }
443
444    pub fn user_tables(&self) -> impl Iterator<Item = &CatalogEntry> {
445        self.entries()
446            .filter(|entry| entry.is_table() && entry.id().is_user())
447    }
448
449    pub fn user_sources(&self) -> impl Iterator<Item = &CatalogEntry> {
450        self.entries()
451            .filter(|entry| entry.is_source() && entry.id().is_user())
452    }
453
454    pub fn user_sinks(&self) -> impl Iterator<Item = &CatalogEntry> {
455        self.entries()
456            .filter(|entry| entry.is_sink() && entry.id().is_user())
457    }
458
459    pub fn user_materialized_views(&self) -> impl Iterator<Item = &CatalogEntry> {
460        self.entries()
461            .filter(|entry| entry.is_materialized_view() && entry.id().is_user())
462    }
463
464    pub fn user_secrets(&self) -> impl Iterator<Item = &CatalogEntry> {
465        self.entries()
466            .filter(|entry| entry.is_secret() && entry.id().is_user())
467    }
468
469    pub fn clusters(&self) -> impl Iterator<Item = &Cluster> {
470        self.clusters_by_id.values()
471    }
472
473    pub fn user_clusters(&self) -> impl Iterator<Item = &Cluster> {
474        self.clusters().filter(|cluster| cluster.id.is_user())
475    }
476
477    pub fn user_cluster_replicas(&self) -> impl Iterator<Item = &ClusterReplica> {
478        self.user_clusters()
479            .flat_map(|cluster| cluster.user_replicas())
480    }
481
482    pub fn databases(&self) -> impl Iterator<Item = &Database> {
483        self.database_by_id.values()
484    }
485
486    pub fn user_roles(&self) -> impl Iterator<Item = &Role> {
487        self.roles_by_id.values().filter(|role| role.is_user())
488    }
489
490    pub fn user_network_policies(&self) -> impl Iterator<Item = &NetworkPolicy> {
491        self.network_policies_by_id
492            .iter()
493            .filter(|(id, _)| id.is_user())
494            .map(|(_, policy)| policy)
495    }
496
497    pub fn cluster_replica_sizes(&self) -> &ClusterReplicaSizeMap {
498        &self.cluster_replica_sizes
499    }
500
501    /// Returns an empty [`CatalogState`] that can be used in tests.
502    // TODO: Ideally we'd mark this as `#[cfg(test)]`, but that doesn't work with the way
503    // tests are structured in this repository.
504    pub fn empty_test() -> Self {
505        CatalogState {
506            database_by_name: Default::default(),
507            database_by_id: Default::default(),
508            entry_by_id: Default::default(),
509            entry_by_global_id: Default::default(),
510            notices_by_dep_id: Default::default(),
511            ambient_schemas_by_name: Default::default(),
512            ambient_schemas_by_id: Default::default(),
513            temporary_namespaces: Default::default(),
514            clusters_by_id: Default::default(),
515            clusters_by_name: Default::default(),
516            network_policies_by_name: Default::default(),
517            roles_by_name: Default::default(),
518            roles_by_id: Default::default(),
519            network_policies_by_id: Default::default(),
520            role_auth_by_id: Default::default(),
521            config: CatalogConfig {
522                start_time: Default::default(),
523                start_instant: Instant::now(),
524                nonce: Default::default(),
525                environment_id: EnvironmentId::for_tests(),
526                session_id: Default::default(),
527                build_info: &DUMMY_BUILD_INFO,
528                now: NOW_ZERO.clone(),
529                connection_context: ConnectionContext::for_tests(Arc::new(
530                    InMemorySecretsController::new(),
531                )),
532                aws_account_id: None,
533                helm_chart_version: None,
534            },
535            cluster_replica_sizes: ClusterReplicaSizeMap::for_tests(),
536            availability_zones: Default::default(),
537            system_configuration: Arc::new(SystemVars::default()),
538            scoped_system_parameters: Default::default(),
539            egress_addresses: Default::default(),
540            aws_principal_context: Default::default(),
541            aws_privatelink_availability_zones: Default::default(),
542            http_host_name: Default::default(),
543            default_privileges: Arc::new(DefaultPrivileges::default()),
544            system_privileges: Arc::new(PrivilegeMap::default()),
545            comments: Arc::new(CommentsMap::default()),
546            source_references: Default::default(),
547            storage_metadata: Arc::new(StorageMetadata::default()),
548            license_key: ValidatedLicenseKey::for_tests(),
549            mock_authentication_nonce: Default::default(),
550        }
551    }
552
553    pub fn for_session<'a>(&'a self, session: &'a Session) -> ConnCatalog<'a> {
554        let state = session.transaction().catalog_state().unwrap_or(self);
555        let mut catalog = state.for_session_metadata(session, session.retain_notice_transmitter());
556        catalog.prepared_statements = Some(session.prepared_statements());
557        catalog.portals = Some(session.portals());
558        catalog
559    }
560
561    /// Binds statement context to this state without substituting a transaction snapshot.
562    pub(crate) fn for_session_metadata(
563        &self,
564        session: &dyn SessionMetadata,
565        notices_tx: mpsc::UnboundedSender<crate::AdapterNotice>,
566    ) -> ConnCatalog<'_> {
567        let search_path = self.resolve_search_path(session);
568        let database = self
569            .database_by_name
570            .get(session.vars().database())
571            .map(|id| id.clone());
572        ConnCatalog {
573            state: Cow::Borrowed(self),
574            unresolvable_ids: BTreeSet::new(),
575            conn_id: session.conn_id().clone(),
576            cluster: session.vars().cluster().into(),
577            database,
578            search_path,
579            role_id: session.current_role_id().clone(),
580            prepared_statements: None,
581            portals: None,
582            notices_tx,
583            restrict_to_user_objects: session.vars().restrict_to_user_objects(),
584        }
585    }
586
587    pub fn for_sessionless_user(&self, role_id: RoleId) -> ConnCatalog<'_> {
588        let (notices_tx, _notices_rx) = mpsc::unbounded_channel();
589        let cluster = self.system_configuration.default_cluster();
590
591        ConnCatalog {
592            state: Cow::Borrowed(self),
593            unresolvable_ids: BTreeSet::new(),
594            conn_id: SYSTEM_CONN_ID.clone(),
595            cluster,
596            database: self
597                .resolve_database(DEFAULT_DATABASE_NAME)
598                .ok()
599                .map(|db| db.id()),
600            // Leaving the system's search path empty allows us to catch issues
601            // where catalog object names have not been normalized correctly.
602            search_path: Vec::new(),
603            role_id,
604            prepared_statements: None,
605            portals: None,
606            notices_tx,
607            restrict_to_user_objects: false,
608        }
609    }
610
611    pub fn for_system_session(&self) -> ConnCatalog<'_> {
612        self.for_sessionless_user(MZ_SYSTEM_ROLE_ID)
613    }
614
615    /// Returns an iterator over the deduplicated identifiers of all
616    /// objects this catalog entry transitively depends on (where
617    /// "depends on" is meant in the sense of [`CatalogItem::uses`], rather than
618    /// [`CatalogItem::references`]).
619    pub fn transitive_uses(&self, id: CatalogItemId) -> impl Iterator<Item = CatalogItemId> + '_ {
620        struct I<'a> {
621            queue: VecDeque<CatalogItemId>,
622            seen: BTreeSet<CatalogItemId>,
623            this: &'a CatalogState,
624        }
625        impl<'a> Iterator for I<'a> {
626            type Item = CatalogItemId;
627            fn next(&mut self) -> Option<Self::Item> {
628                if let Some(next) = self.queue.pop_front() {
629                    for child in self.this.get_entry(&next).item().uses() {
630                        if !self.seen.contains(&child) {
631                            self.queue.push_back(child);
632                            self.seen.insert(child);
633                        }
634                    }
635                    Some(next)
636                } else {
637                    None
638                }
639            }
640        }
641
642        I {
643            queue: [id].into_iter().collect(),
644            seen: [id].into_iter().collect(),
645            this: self,
646        }
647    }
648
649    /// Computes the IDs of any log sources this catalog entry transitively
650    /// depends on.
651    pub fn introspection_dependencies(&self, id: CatalogItemId) -> Vec<CatalogItemId> {
652        let mut out = Vec::new();
653
654        // Iterative worklist traversal rather than recursion. The dependency
655        // chain is user controlled and can be arbitrarily deep.
656        let mut queue: VecDeque<CatalogItemId> = [id].into_iter().collect();
657        let mut seen: BTreeSet<CatalogItemId> = [id].into_iter().collect();
658        while let Some(id) = queue.pop_front() {
659            match self.get_entry(&id).item() {
660                CatalogItem::Log(_) => out.push(id),
661                item @ (CatalogItem::View(_)
662                | CatalogItem::MaterializedView(_)
663                | CatalogItem::Connection(_)) => {
664                    // TODO Unclear if this table wants to include all uses or only references.
665                    for item_id in item.references().items() {
666                        if seen.insert(*item_id) {
667                            queue.push_back(*item_id);
668                        }
669                    }
670                }
671                CatalogItem::Sink(sink) => {
672                    let from_item_id = self.get_entry_by_global_id(&sink.from).id();
673                    if seen.insert(from_item_id) {
674                        queue.push_back(from_item_id);
675                    }
676                }
677                CatalogItem::MetricSink(metric_sink) => {
678                    let from_item_id = self.get_entry_by_global_id(&metric_sink.from).id();
679                    if seen.insert(from_item_id) {
680                        queue.push_back(from_item_id);
681                    }
682                }
683                CatalogItem::Index(idx) => {
684                    let on_item_id = self.get_entry_by_global_id(&idx.on).id();
685                    if seen.insert(on_item_id) {
686                        queue.push_back(on_item_id);
687                    }
688                }
689                CatalogItem::Table(_)
690                | CatalogItem::Source(_)
691                | CatalogItem::Type(_)
692                | CatalogItem::Func(_)
693                | CatalogItem::Secret(_) => (),
694            }
695        }
696
697        out
698    }
699
700    /// Returns all the IDs of all objects that depend on `ids`, including `ids` themselves.
701    ///
702    /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
703    /// earlier in the list than the roots. This is particularly useful for the order to drop
704    /// objects.
705    pub(super) fn object_dependents(
706        &self,
707        object_ids: &Vec<ObjectId>,
708        conn_id: &ConnectionId,
709        seen: &mut BTreeSet<ObjectId>,
710    ) -> Vec<ObjectId> {
711        let mut dependents = Vec::new();
712        for object_id in object_ids {
713            match object_id {
714                ObjectId::Cluster(id) => {
715                    dependents.extend_from_slice(&self.cluster_dependents(*id, seen));
716                }
717                ObjectId::ClusterReplica((cluster_id, replica_id)) => dependents.extend_from_slice(
718                    &self.cluster_replica_dependents(*cluster_id, *replica_id, seen),
719                ),
720                ObjectId::Database(id) => {
721                    dependents.extend_from_slice(&self.database_dependents(*id, conn_id, seen))
722                }
723                ObjectId::Schema((database_spec, schema_spec)) => {
724                    dependents.extend_from_slice(&self.schema_dependents(
725                        database_spec.clone(),
726                        schema_spec.clone(),
727                        conn_id,
728                        seen,
729                    ));
730                }
731                ObjectId::NetworkPolicy(id) => {
732                    dependents.extend_from_slice(&self.network_policy_dependents(*id, seen));
733                }
734                id @ ObjectId::Role(_) => {
735                    let unseen = seen.insert(id.clone());
736                    if unseen {
737                        dependents.push(id.clone());
738                    }
739                }
740                ObjectId::Item(id) => {
741                    dependents.extend_from_slice(&self.item_dependents(*id, seen))
742                }
743            }
744        }
745        dependents
746    }
747
748    /// Returns all the IDs of all objects that depend on `cluster_id`, including `cluster_id`
749    /// itself.
750    ///
751    /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
752    /// earlier in the list than the roots. This is particularly useful for the order to drop
753    /// objects.
754    fn cluster_dependents(
755        &self,
756        cluster_id: ClusterId,
757        seen: &mut BTreeSet<ObjectId>,
758    ) -> Vec<ObjectId> {
759        let mut dependents = Vec::new();
760        let object_id = ObjectId::Cluster(cluster_id);
761        if !seen.contains(&object_id) {
762            seen.insert(object_id.clone());
763            let cluster = self.get_cluster(cluster_id);
764            for item_id in cluster.bound_objects() {
765                dependents.extend_from_slice(&self.item_dependents(*item_id, seen));
766            }
767            for replica_id in cluster.replica_ids().values() {
768                dependents.extend_from_slice(&self.cluster_replica_dependents(
769                    cluster_id,
770                    *replica_id,
771                    seen,
772                ));
773            }
774            dependents.push(object_id);
775        }
776        dependents
777    }
778
779    /// Returns all the IDs of all objects that depend on `replica_id`, including `replica_id`
780    /// itself.
781    ///
782    /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
783    /// earlier in the list than the roots. This is particularly useful for the order to drop
784    /// objects.
785    pub(super) fn cluster_replica_dependents(
786        &self,
787        cluster_id: ClusterId,
788        replica_id: ReplicaId,
789        seen: &mut BTreeSet<ObjectId>,
790    ) -> Vec<ObjectId> {
791        let mut dependents = Vec::new();
792        let object_id = ObjectId::ClusterReplica((cluster_id, replica_id));
793        if !seen.contains(&object_id) {
794            seen.insert(object_id.clone());
795            // Materialized views that target this replica are implicitly
796            // dropped with it, so cascade to their dependents to avoid leaving
797            // dangling references.
798            let cluster = self.get_cluster(cluster_id);
799            for item_id in cluster.bound_objects() {
800                if let CatalogItem::MaterializedView(mv) = self.get_entry(item_id).item()
801                    && mv.target_replica == Some(replica_id)
802                {
803                    dependents.extend_from_slice(&self.item_dependents(*item_id, seen));
804                }
805            }
806            dependents.push(object_id);
807        }
808        dependents
809    }
810
811    /// Returns all the IDs of all objects that depend on `database_id`, including `database_id`
812    /// itself.
813    ///
814    /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
815    /// earlier in the list than the roots. This is particularly useful for the order to drop
816    /// objects.
817    fn database_dependents(
818        &self,
819        database_id: DatabaseId,
820        conn_id: &ConnectionId,
821        seen: &mut BTreeSet<ObjectId>,
822    ) -> Vec<ObjectId> {
823        let mut dependents = Vec::new();
824        let object_id = ObjectId::Database(database_id);
825        if !seen.contains(&object_id) {
826            seen.insert(object_id.clone());
827            let database = self.get_database(&database_id);
828            for schema_id in database.schema_ids().values() {
829                dependents.extend_from_slice(&self.schema_dependents(
830                    ResolvedDatabaseSpecifier::Id(database_id),
831                    SchemaSpecifier::Id(*schema_id),
832                    conn_id,
833                    seen,
834                ));
835            }
836            dependents.push(object_id);
837        }
838        dependents
839    }
840
841    /// Returns all the IDs of all objects that depend on `schema_id`, including `schema_id`
842    /// itself.
843    ///
844    /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
845    /// earlier in the list than the roots. This is particularly useful for the order to drop
846    /// objects.
847    fn schema_dependents(
848        &self,
849        database_spec: ResolvedDatabaseSpecifier,
850        schema_spec: SchemaSpecifier,
851        conn_id: &ConnectionId,
852        seen: &mut BTreeSet<ObjectId>,
853    ) -> Vec<ObjectId> {
854        let mut dependents = Vec::new();
855        let object_id = ObjectId::Schema((database_spec, schema_spec.clone()));
856        if !seen.contains(&object_id) {
857            seen.insert(object_id.clone());
858            let schema = self.get_schema(&database_spec, &schema_spec, conn_id);
859            for item_id in schema.item_ids() {
860                dependents.extend_from_slice(&self.item_dependents(item_id, seen));
861            }
862            dependents.push(object_id)
863        }
864        dependents
865    }
866
867    /// Returns all the IDs of all objects that depend on `item_id`, including `item_id`
868    /// itself.
869    ///
870    /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
871    /// earlier in the list than the roots. This is particularly useful for the order to drop
872    /// objects.
873    pub(super) fn item_dependents(
874        &self,
875        item_id: CatalogItemId,
876        seen: &mut BTreeSet<ObjectId>,
877    ) -> Vec<ObjectId> {
878        let mut dependents = Vec::new();
879
880        // Iterative post-order traversal rather than recursion. Dependency
881        // chains are user controlled and can be arbitrarily deep (e.g. a long
882        // chain of stacked views), so recursing risks a stack overflow.
883        enum Work {
884            Enter(CatalogItemId),
885            Emit(ObjectId),
886        }
887        let mut stack = vec![Work::Enter(item_id)];
888        while let Some(work) = stack.pop() {
889            match work {
890                Work::Enter(item_id) => {
891                    let object_id = ObjectId::Item(item_id);
892                    if !seen.insert(object_id.clone()) {
893                        continue;
894                    }
895                    let entry = self.get_entry(&item_id);
896                    // Pushed in reverse of the desired output order: dependents
897                    // first, then self, then the progress collection. We treat
898                    // the progress collection as if it depends on the source
899                    // for dropping. We have additional code in planning to
900                    // create a kind of special-case "CASCADE" for this
901                    // dependency.
902                    if let Some(progress_id) = entry.progress_id() {
903                        stack.push(Work::Enter(progress_id));
904                    }
905                    stack.push(Work::Emit(object_id));
906                    for dependent_id in entry.used_by().iter().rev() {
907                        stack.push(Work::Enter(*dependent_id));
908                    }
909                }
910                Work::Emit(object_id) => dependents.push(object_id),
911            }
912        }
913
914        dependents
915    }
916
917    /// Returns all the IDs of all objects that depend on `network_policy_id`, including `network_policy_id`
918    /// itself.
919    ///
920    /// The order is guaranteed to be in reverse dependency order, i.e. the leafs will appear
921    /// earlier in the list than the roots. This is particularly useful for the order to drop
922    /// objects.
923    pub(super) fn network_policy_dependents(
924        &self,
925        network_policy_id: NetworkPolicyId,
926        _seen: &mut BTreeSet<ObjectId>,
927    ) -> Vec<ObjectId> {
928        let object_id = ObjectId::NetworkPolicy(network_policy_id);
929        // Currently network policies have no dependents
930        // when we add the ability for users or sources/sinks to have policies
931        // this method will need to be updated.
932        vec![object_id]
933    }
934
935    /// Indicates whether the indicated item is considered stable or not.
936    ///
937    /// Only stable items can be used as dependencies of other catalog items.
938    fn is_stable(&self, id: CatalogItemId) -> bool {
939        let spec = self.get_entry(&id).name().qualifiers.schema_spec;
940        !self.is_unstable_schema_specifier(spec)
941    }
942
943    pub(super) fn check_unstable_dependencies(&self, item: &CatalogItem) -> Result<(), Error> {
944        if self.system_config().unsafe_enable_unstable_dependencies() {
945            return Ok(());
946        }
947
948        let unstable_dependencies: Vec<_> = item
949            .references()
950            .items()
951            .filter(|id| !self.is_stable(**id))
952            .map(|id| self.get_entry(id).name().item.clone())
953            .collect();
954
955        // It's okay to create a temporary object with unstable
956        // dependencies, since we will never need to reboot a catalog
957        // that contains it.
958        if unstable_dependencies.is_empty() || item.is_temporary() {
959            Ok(())
960        } else {
961            let object_type = item.typ().to_string();
962            Err(Error {
963                kind: ErrorKind::UnstableDependency {
964                    object_type,
965                    unstable_dependencies,
966                },
967            })
968        }
969    }
970
971    pub fn resolve_full_name(
972        &self,
973        name: &QualifiedItemName,
974        conn_id: Option<&ConnectionId>,
975    ) -> FullItemName {
976        let conn_id = conn_id.unwrap_or(&SYSTEM_CONN_ID);
977
978        let database = match &name.qualifiers.database_spec {
979            ResolvedDatabaseSpecifier::Ambient => RawDatabaseSpecifier::Ambient,
980            ResolvedDatabaseSpecifier::Id(id) => {
981                RawDatabaseSpecifier::Name(self.get_database(id).name().to_string())
982            }
983        };
984        // For temporary schemas, we know the name is always MZ_TEMP_SCHEMA,
985        // and the schema may not exist yet if no temporary items have been created.
986        let schema = match &name.qualifiers.schema_spec {
987            SchemaSpecifier::Temporary => MZ_TEMP_SCHEMA.to_string(),
988            SchemaSpecifier::Id(_) => self
989                .get_schema(
990                    &name.qualifiers.database_spec,
991                    &name.qualifiers.schema_spec,
992                    conn_id,
993                )
994                .name()
995                .schema
996                .clone(),
997        };
998        FullItemName {
999            database,
1000            schema,
1001            item: name.item.clone(),
1002        }
1003    }
1004
1005    pub(super) fn resolve_full_schema_name(&self, name: &QualifiedSchemaName) -> FullSchemaName {
1006        let database = match &name.database {
1007            ResolvedDatabaseSpecifier::Ambient => RawDatabaseSpecifier::Ambient,
1008            ResolvedDatabaseSpecifier::Id(id) => {
1009                RawDatabaseSpecifier::Name(self.get_database(id).name().to_string())
1010            }
1011        };
1012        FullSchemaName {
1013            database,
1014            schema: name.schema.clone(),
1015        }
1016    }
1017
1018    pub fn get_entry(&self, id: &CatalogItemId) -> &CatalogEntry {
1019        self.entry_by_id
1020            .get(id)
1021            .unwrap_or_else(|| panic!("catalog out of sync, missing id {id:?}"))
1022    }
1023
1024    pub fn get_entry_by_global_id(&self, id: &GlobalId) -> CatalogCollectionEntry {
1025        let item_id = self
1026            .entry_by_global_id
1027            .get(id)
1028            .unwrap_or_else(|| panic!("catalog out of sync, missing id {id:?}"));
1029
1030        let entry = self.get_entry(item_id).clone();
1031        let version = match entry.item() {
1032            CatalogItem::Table(table) => {
1033                let (version, _) = table
1034                    .collections
1035                    .iter()
1036                    .find(|(_verison, gid)| *gid == id)
1037                    .expect("version to exist");
1038                RelationVersionSelector::Specific(*version)
1039            }
1040            _ => RelationVersionSelector::Latest,
1041        };
1042        CatalogCollectionEntry { entry, version }
1043    }
1044
1045    pub fn get_entries(&self) -> impl Iterator<Item = (&CatalogItemId, &CatalogEntry)> + '_ {
1046        self.entry_by_id.iter()
1047    }
1048
1049    pub fn get_temp_items(&self, conn: &ConnectionId) -> impl Iterator<Item = ObjectId> + '_ {
1050        // A temporary namespace is registered at the connection's first
1051        // temporary-item creation, so it's valid for one to not exist yet.
1052        self.temporary_namespaces
1053            .schema(conn)
1054            .into_iter()
1055            .flat_map(|schema| schema.items.values().copied().map(ObjectId::from))
1056    }
1057
1058    /// Returns true if a temporary namespace is registered for the given
1059    /// connection, i.e. it has (or has had) temporary items.
1060    ///
1061    /// The namespace is registered at the connection's first temporary-item
1062    /// creation, so this returns false for connections that never created
1063    /// any temporary objects.
1064    pub fn has_temporary_namespace(&self, conn: &ConnectionId) -> bool {
1065        self.temporary_namespaces.contains_conn(conn)
1066    }
1067
1068    /// Converts an in-memory catalog entry into its durable representation.
1069    ///
1070    /// The durable owner of a temporary entry is the session whose connection
1071    /// currently holds it, resolved from the temporary namespace
1072    /// registered at the session's first temporary-item creation.
1073    pub(super) fn durable_item(
1074        &self,
1075        entry: CatalogEntry,
1076    ) -> Result<mz_catalog::durable::Item, AdapterError> {
1077        let ephemeral_owner_session = entry
1078            .conn_id()
1079            .map(|conn_id| {
1080                self.temporary_namespaces
1081                    .uuid_for_conn(conn_id)
1082                    .ok_or_else(|| {
1083                        AdapterError::Internal(format!(
1084                            "no session record for connection {conn_id} owning temporary item"
1085                        ))
1086                    })
1087            })
1088            .transpose()?;
1089        let (create_sql, global_id, extra_versions) = entry.item.into_serialized();
1090        Ok(mz_catalog::durable::Item {
1091            id: entry.id,
1092            oid: entry.oid,
1093            global_id,
1094            schema_id: entry.name.qualifiers.schema_spec.into(),
1095            name: entry.name.item,
1096            create_sql,
1097            owner_id: entry.owner_id,
1098            privileges: entry.privileges.into_all_values().collect(),
1099            extra_versions,
1100            ephemeral_owner_session,
1101        })
1102    }
1103
1104    /// Records in `tx` the planner-resolved metadata of all items in
1105    /// the current transaction.
1106    pub(super) fn refresh_item_metadata(
1107        &self,
1108        tx: &mut mz_catalog::durable::Transaction<'_>,
1109    ) -> Result<(), mz_catalog::durable::CatalogError> {
1110        let ids: Vec<_> = tx.get_items().map(|item| item.id).collect();
1111        self.record_item_metadata(tx, ids)
1112    }
1113
1114    /// Records in `tx` the planner-resolved metadata of every item in `ids`
1115    /// that this catalog has loaded.
1116    pub(super) fn record_item_metadata(
1117        &self,
1118        tx: &mut mz_catalog::durable::Transaction<'_>,
1119        ids: impl IntoIterator<Item = CatalogItemId>,
1120    ) -> Result<(), mz_catalog::durable::CatalogError> {
1121        for id in ids {
1122            // An id this catalog does not know, such as an item of a
1123            // temporary schema owned by a session that is not connected to this process,
1124            // is skipped.
1125            let Some(entry) = self.try_get_entry(&id) else {
1126                continue;
1127            };
1128            tx.set_item_metadata(id, self.durable_item_metadata(entry.item()))?;
1129        }
1130        Ok(())
1131    }
1132
1133    /// The planner-resolved metadata the durable catalog records for `item`:
1134    /// the columns of a relation, each by the identity its type presents as,
1135    /// and the keys of an index. See `mz_catalog::durable::ItemMetadata`.
1136    ///
1137    /// An index's keys are typed against the version of the indexed relation
1138    /// it was created on, which must be in this catalog.
1139    pub(super) fn durable_item_metadata(
1140        &self,
1141        item: &CatalogItem,
1142    ) -> mz_catalog::durable::ItemMetadata {
1143        let columns = item
1144            .relation_desc(RelationVersionSelector::Latest)
1145            .map(|desc| {
1146                desc.iter()
1147                    .map(|(name, typ)| {
1148                        let pg_type = mz_pgrepr::Type::from(&typ.scalar_type);
1149                        let custom_type = match &typ.scalar_type {
1150                            SqlScalarType::List {
1151                                custom_id: Some(id),
1152                                ..
1153                            }
1154                            | SqlScalarType::Map {
1155                                custom_id: Some(id),
1156                                ..
1157                            }
1158                            | SqlScalarType::Record {
1159                                custom_id: Some(id),
1160                                ..
1161                            } => Some(*id),
1162                            _ => None,
1163                        };
1164                        mz_catalog::durable::ItemColumn {
1165                            name: name.clone(),
1166                            nullable: typ.nullable,
1167                            type_oid: pg_type.oid(),
1168                            type_mod: pg_type.typmod(),
1169                            custom_type,
1170                        }
1171                    })
1172                    .collect()
1173            })
1174            .unwrap_or_default();
1175        let index_columns = match item {
1176            CatalogItem::Index(index) => {
1177                let on_entry = self.get_entry_by_global_id(&index.on);
1178                let on_desc = on_entry
1179                    .relation_desc()
1180                    .expect("indexes are built on relations");
1181                let column_types: Vec<ReprColumnType> = on_desc
1182                    .typ()
1183                    .column_types
1184                    .iter()
1185                    .map(ReprColumnType::from)
1186                    .collect();
1187                index
1188                    .keys
1189                    .iter()
1190                    .map(|key| mz_catalog::durable::IndexColumn {
1191                        column: match key {
1192                            MirScalarExpr::Column(column, _) => Some(*column),
1193                            _ => None,
1194                        },
1195                        nullable: key.typ(&column_types).nullable,
1196                    })
1197                    .collect()
1198            }
1199            _ => Vec::new(),
1200        };
1201        mz_catalog::durable::ItemMetadata {
1202            columns,
1203            index_columns,
1204        }
1205    }
1206
1207    /// Gets a type named `name` from exactly one of the system schemas.
1208    ///
1209    /// # Panics
1210    /// - If `name` is not an entry in any system schema
1211    /// - If more than one system schema has an entry named `name`.
1212    pub(super) fn get_system_type(&self, name: &str) -> &CatalogEntry {
1213        let mut res = None;
1214        for schema_id in self.system_schema_ids() {
1215            let schema = &self.ambient_schemas_by_id[&schema_id];
1216            if let Some(global_id) = schema.types.get(name) {
1217                match res {
1218                    None => res = Some(self.get_entry(global_id)),
1219                    Some(_) => panic!(
1220                        "only call get_system_type on objects uniquely identifiable in one system schema"
1221                    ),
1222                }
1223            }
1224        }
1225
1226        res.unwrap_or_else(|| panic!("cannot find type {} in system schema", name))
1227    }
1228
1229    pub fn get_item_by_name(
1230        &self,
1231        name: &QualifiedItemName,
1232        conn_id: &ConnectionId,
1233    ) -> Option<&CatalogEntry> {
1234        self.get_schema(
1235            &name.qualifiers.database_spec,
1236            &name.qualifiers.schema_spec,
1237            conn_id,
1238        )
1239        .items
1240        .get(&name.item)
1241        .and_then(|id| self.try_get_entry(id))
1242    }
1243
1244    pub fn get_type_by_name(
1245        &self,
1246        name: &QualifiedItemName,
1247        conn_id: &ConnectionId,
1248    ) -> Option<&CatalogEntry> {
1249        self.get_schema(
1250            &name.qualifiers.database_spec,
1251            &name.qualifiers.schema_spec,
1252            conn_id,
1253        )
1254        .types
1255        .get(&name.item)
1256        .and_then(|id| self.try_get_entry(id))
1257    }
1258
1259    pub(super) fn find_available_name(
1260        &self,
1261        mut name: QualifiedItemName,
1262        conn_id: &ConnectionId,
1263    ) -> QualifiedItemName {
1264        let mut i = 0;
1265        let orig_item_name = name.item.clone();
1266        while self.get_item_by_name(&name, conn_id).is_some() {
1267            i += 1;
1268            name.item = format!("{}{}", orig_item_name, i);
1269        }
1270        name
1271    }
1272
1273    pub fn try_get_entry(&self, id: &CatalogItemId) -> Option<&CatalogEntry> {
1274        self.entry_by_id.get(id)
1275    }
1276
1277    pub fn try_get_entry_by_global_id(&self, id: &GlobalId) -> Option<&CatalogEntry> {
1278        let item_id = self.entry_by_global_id.get(id)?;
1279        self.try_get_entry(item_id)
1280    }
1281
1282    /// Returns the [`RelationDesc`] for a [`GlobalId`], if the provided [`GlobalId`] refers to an
1283    /// object that returns rows.
1284    pub fn try_get_desc_by_global_id(&self, id: &GlobalId) -> Option<Cow<'_, RelationDesc>> {
1285        let entry = self.try_get_entry_by_global_id(id)?;
1286        let desc = match entry.item() {
1287            CatalogItem::Table(table) => Cow::Owned(table.desc_for(id)),
1288            // TODO(alter_table): Support schema evolution on sources.
1289            other => other.relation_desc(RelationVersionSelector::Latest)?,
1290        };
1291        Some(desc)
1292    }
1293
1294    pub(crate) fn get_cluster(&self, cluster_id: ClusterId) -> &Cluster {
1295        self.try_get_cluster(cluster_id)
1296            .unwrap_or_else(|| panic!("unknown cluster {cluster_id}"))
1297    }
1298
1299    pub(super) fn try_get_cluster(&self, cluster_id: ClusterId) -> Option<&Cluster> {
1300        self.clusters_by_id.get(&cluster_id)
1301    }
1302
1303    pub(super) fn try_get_role(&self, id: &RoleId) -> Option<&Role> {
1304        self.roles_by_id.get(id)
1305    }
1306
1307    pub fn get_role(&self, id: &RoleId) -> &Role {
1308        self.roles_by_id.get(id).expect("catalog out of sync")
1309    }
1310
1311    pub fn get_roles(&self) -> impl Iterator<Item = &RoleId> {
1312        self.roles_by_id.keys()
1313    }
1314
1315    pub(super) fn try_get_role_by_name(&self, role_name: &str) -> Option<&Role> {
1316        self.roles_by_name
1317            .get(role_name)
1318            .map(|id| &self.roles_by_id[id])
1319    }
1320
1321    pub(super) fn get_role_auth(&self, id: &RoleId) -> &RoleAuth {
1322        self.role_auth_by_id
1323            .get(id)
1324            .unwrap_or_else(|| panic!("catalog out of sync, missing role auth for {id}"))
1325    }
1326
1327    pub(super) fn try_get_role_auth_by_id(&self, id: &RoleId) -> Option<&RoleAuth> {
1328        self.role_auth_by_id.get(id)
1329    }
1330
1331    pub(super) fn try_get_network_policy_by_name(
1332        &self,
1333        policy_name: &str,
1334    ) -> Option<&NetworkPolicy> {
1335        self.network_policies_by_name
1336            .get(policy_name)
1337            .map(|id| &self.network_policies_by_id[id])
1338    }
1339
1340    pub(crate) fn collect_role_membership(&self, id: &RoleId) -> BTreeSet<RoleId> {
1341        let mut membership = BTreeSet::new();
1342        let mut queue = VecDeque::from(vec![id]);
1343        while let Some(cur_id) = queue.pop_front() {
1344            if !membership.contains(cur_id) {
1345                membership.insert(cur_id.clone());
1346                let role = self.get_role(cur_id);
1347                soft_assert_no_log!(
1348                    !role.membership().keys().contains(id),
1349                    "circular membership exists in the catalog"
1350                );
1351                queue.extend(role.membership().keys());
1352            }
1353        }
1354        membership.insert(RoleId::Public);
1355        membership
1356    }
1357
1358    pub fn get_network_policy(&self, id: &NetworkPolicyId) -> &NetworkPolicy {
1359        self.network_policies_by_id
1360            .get(id)
1361            .expect("catalog out of sync")
1362    }
1363
1364    pub fn get_network_policies(&self) -> impl Iterator<Item = &NetworkPolicyId> {
1365        self.network_policies_by_id.keys()
1366    }
1367
1368    /// Returns the URL for POST-ing data to a webhook source, if `id` corresponds to a webhook
1369    /// source.
1370    ///
1371    /// Note: Identifiers for the source, e.g. item name, are URL encoded.
1372    pub fn try_get_webhook_url(&self, id: &CatalogItemId) -> Option<url::Url> {
1373        let entry = self.try_get_entry(id)?;
1374        // Note: Webhook sources can never be created in the temporary schema, hence passing None.
1375        let name = self.resolve_full_name(entry.name(), None);
1376        let host_name = self
1377            .http_host_name
1378            .as_ref()
1379            .map(|x| x.as_str())
1380            .unwrap_or_else(|| "HOST");
1381
1382        let RawDatabaseSpecifier::Name(database) = name.database else {
1383            return None;
1384        };
1385
1386        let mut url = url::Url::parse(&format!("https://{host_name}/api/webhook")).ok()?;
1387        url.path_segments_mut()
1388            .ok()?
1389            .push(&database)
1390            .push(&name.schema)
1391            .push(&name.item);
1392
1393        Some(url)
1394    }
1395
1396    /// Parses the given SQL string into a pair of [`Plan`] and a [`ResolvedIds`].
1397    ///
1398    /// This function will temporarily enable all "enable_for_item_parsing" feature flags. See
1399    /// [`CatalogState::with_enable_for_item_parsing`] for more details.
1400    ///
1401    /// NOTE: While this method takes a `&mut self`, all mutations are temporary and restored to
1402    /// their original state before the method returns.
1403    pub(crate) fn deserialize_plan_with_enable_for_item_parsing(
1404        // DO NOT add any additional mutations to this method. It would be fairly surprising to the
1405        // caller if this method changed the state of the catalog.
1406        &mut self,
1407        create_sql: &str,
1408        force_if_exists_skip: bool,
1409    ) -> Result<(Plan, ResolvedIds), AdapterError> {
1410        self.with_enable_for_item_parsing(|state| {
1411            let pcx = PlanContext::zero().with_ignore_if_exists_errors(force_if_exists_skip);
1412            let pcx = Some(&pcx);
1413            let session_catalog = state.for_system_session();
1414
1415            let stmt = mz_sql::parse::parse(create_sql)?.into_element().ast;
1416            let (stmt, resolved_ids) = mz_sql::names::resolve(&session_catalog, stmt)?;
1417            let (plan, _sql_impl_ids) =
1418                mz_sql::plan::plan(pcx, &session_catalog, stmt, &Params::empty(), &resolved_ids)?;
1419
1420            Ok((plan, resolved_ids))
1421        })
1422    }
1423
1424    /// Parses the given SQL string into a pair of [`Plan`] and a [`ResolvedIds`].
1425    #[mz_ore::instrument]
1426    pub(crate) fn parse_plan(
1427        create_sql: &str,
1428        pcx: Option<&PlanContext>,
1429        catalog: &ConnCatalog,
1430    ) -> Result<(Plan, ResolvedIds), AdapterError> {
1431        let stmt = mz_sql::parse::parse(create_sql)?.into_element().ast;
1432        let (stmt, resolved_ids) = mz_sql::names::resolve(catalog, stmt)?;
1433        let (plan, _sql_impl_ids) =
1434            mz_sql::plan::plan(pcx, catalog, stmt, &Params::empty(), &resolved_ids)?;
1435
1436        Ok((plan, resolved_ids))
1437    }
1438
1439    /// Parses the given SQL string into a pair of [`CatalogItem`].
1440    pub(crate) fn deserialize_item(
1441        &self,
1442        global_id: GlobalId,
1443        create_sql: &str,
1444        extra_versions: &BTreeMap<RelationVersion, GlobalId>,
1445        local_expression_cache: &mut LocalExpressionCache,
1446        previous_item: Option<CatalogItem>,
1447    ) -> Result<CatalogItem, AdapterError> {
1448        self.parse_item(
1449            global_id,
1450            create_sql,
1451            extra_versions,
1452            None,
1453            false,
1454            None,
1455            local_expression_cache,
1456            previous_item,
1457        )
1458    }
1459
1460    /// Parses the given SQL string into a `CatalogItem`.
1461    #[mz_ore::instrument]
1462    pub(crate) fn parse_item(
1463        &self,
1464        global_id: GlobalId,
1465        create_sql: &str,
1466        extra_versions: &BTreeMap<RelationVersion, GlobalId>,
1467        pcx: Option<&PlanContext>,
1468        is_retained_metrics_object: bool,
1469        custom_logical_compaction_window: Option<CompactionWindow>,
1470        local_expression_cache: &mut LocalExpressionCache,
1471        previous_item: Option<CatalogItem>,
1472    ) -> Result<CatalogItem, AdapterError> {
1473        let cached_expr = local_expression_cache.remove_cached_expression(&global_id);
1474        match self.parse_item_inner(
1475            global_id,
1476            create_sql,
1477            extra_versions,
1478            pcx,
1479            is_retained_metrics_object,
1480            custom_logical_compaction_window,
1481            cached_expr,
1482            previous_item,
1483        ) {
1484            Ok((item, uncached_expr)) => {
1485                if let Some((uncached_expr, optimizer_features)) = uncached_expr {
1486                    local_expression_cache.insert_uncached_expression(
1487                        global_id,
1488                        uncached_expr,
1489                        optimizer_features,
1490                        latest_item_version(extra_versions),
1491                    );
1492                }
1493                Ok(item)
1494            }
1495            Err((err, cached_expr)) => {
1496                if let Some(local_expr) = cached_expr {
1497                    local_expression_cache.insert_cached_expression(global_id, local_expr);
1498                }
1499                Err(err)
1500            }
1501        }
1502    }
1503
1504    /// Parses the given SQL string into a `CatalogItem`, using `cached_expr` if it's Some.
1505    ///
1506    /// On success returns the `CatalogItem` and an optimized expression iff the expression was
1507    /// not cached.
1508    ///
1509    /// On failure returns an error and `cached_expr` so it can be used later.
1510    #[mz_ore::instrument]
1511    pub(crate) fn parse_item_inner(
1512        &self,
1513        global_id: GlobalId,
1514        create_sql: &str,
1515        extra_versions: &BTreeMap<RelationVersion, GlobalId>,
1516        pcx: Option<&PlanContext>,
1517        is_retained_metrics_object: bool,
1518        custom_logical_compaction_window: Option<CompactionWindow>,
1519        cached_expr: Option<LocalExpressions>,
1520        previous_item: Option<CatalogItem>,
1521    ) -> Result<
1522        (
1523            CatalogItem,
1524            Option<(OptimizedMirRelationExpr, OptimizerFeatures)>,
1525        ),
1526        (AdapterError, Option<LocalExpressions>),
1527    > {
1528        let session_catalog = self.for_system_session();
1529
1530        let (plan, resolved_ids) = match Self::parse_plan(create_sql, pcx, &session_catalog) {
1531            Ok((plan, resolved_ids)) => (plan, resolved_ids),
1532            Err(err) => return Err((err, cached_expr)),
1533        };
1534
1535        let mut uncached_expr = None;
1536
1537        // Carry over the plans (`optimized_plan`, `physical_plan`,
1538        // `dataflow_metainfo`) from the previous incarnation of this item when
1539        // re-parsing an existing item (e.g. after a RENAME). These fields live
1540        // on the `CatalogItem` since #35834, but are not reconstructable from
1541        // `create_sql` alone — they are populated by the sequencer `_finish`
1542        // paths at create time, and by the expression-cache / bootstrap
1543        // rendering path on boot. If we don't preserve them here, a RENAME
1544        // silently drops the plans and dataflow metainfo for the affected
1545        // MV/Index/CT.
1546        let previous_plans = previous_item.as_ref().map(|item| {
1547            (
1548                item.optimized_plan().cloned(),
1549                item.physical_plan().cloned(),
1550                item.dataflow_metainfo().cloned(),
1551            )
1552        });
1553
1554        let mut item = match plan {
1555            Plan::CreateTable(CreateTablePlan { table, .. }) => {
1556                let collections = extra_versions
1557                    .iter()
1558                    .map(|(version, gid)| (*version, *gid))
1559                    .chain([(RelationVersion::root(), global_id)].into_iter())
1560                    .collect();
1561
1562                CatalogItem::Table(Table {
1563                    create_sql: Some(table.create_sql),
1564                    desc: table.desc,
1565                    collections,
1566                    conn_id: None,
1567                    resolved_ids,
1568                    custom_logical_compaction_window: custom_logical_compaction_window
1569                        .or(table.compaction_window),
1570                    is_retained_metrics_object,
1571                    data_source: match table.data_source {
1572                        mz_sql::plan::TableDataSource::TableWrites { defaults } => {
1573                            TableDataSource::TableWrites { defaults }
1574                        }
1575                        mz_sql::plan::TableDataSource::DataSource {
1576                            desc: data_source_desc,
1577                            timeline,
1578                        } => match data_source_desc {
1579                            mz_sql::plan::DataSourceDesc::IngestionExport {
1580                                ingestion_id,
1581                                external_reference,
1582                                details,
1583                                data_config,
1584                            } => TableDataSource::DataSource {
1585                                desc: DataSourceDesc::IngestionExport {
1586                                    ingestion_id,
1587                                    external_reference,
1588                                    details,
1589                                    data_config,
1590                                },
1591                                timeline,
1592                            },
1593                            mz_sql::plan::DataSourceDesc::Webhook {
1594                                validate_using,
1595                                body_format,
1596                                headers,
1597                                cluster_id,
1598                            } => TableDataSource::DataSource {
1599                                desc: DataSourceDesc::Webhook {
1600                                    validate_using,
1601                                    body_format,
1602                                    headers,
1603                                    cluster_id: cluster_id
1604                                        .expect("Webhook Tables must have a cluster_id set"),
1605                                },
1606                                timeline,
1607                            },
1608                            _ => {
1609                                return Err((
1610                                    AdapterError::Unstructured(anyhow::anyhow!(
1611                                        "unsupported data source for table"
1612                                    )),
1613                                    cached_expr,
1614                                ));
1615                            }
1616                        },
1617                    },
1618                })
1619            }
1620            Plan::CreateSource(CreateSourcePlan {
1621                source,
1622                timeline,
1623                in_cluster,
1624                ..
1625            }) => CatalogItem::Source(Source {
1626                create_sql: Some(source.create_sql),
1627                data_source: match source.data_source {
1628                    mz_sql::plan::DataSourceDesc::Ingestion(desc) => DataSourceDesc::Ingestion {
1629                        desc,
1630                        cluster_id: match in_cluster {
1631                            Some(id) => id,
1632                            None => {
1633                                return Err((
1634                                    AdapterError::Unstructured(anyhow::anyhow!(
1635                                        "ingestion-based sources must have cluster specified"
1636                                    )),
1637                                    cached_expr,
1638                                ));
1639                            }
1640                        },
1641                    },
1642                    mz_sql::plan::DataSourceDesc::OldSyntaxIngestion {
1643                        desc,
1644                        progress_subsource,
1645                        data_config,
1646                        details,
1647                    } => DataSourceDesc::OldSyntaxIngestion {
1648                        desc,
1649                        progress_subsource,
1650                        data_config,
1651                        details,
1652                        cluster_id: match in_cluster {
1653                            Some(id) => id,
1654                            None => {
1655                                return Err((
1656                                    AdapterError::Unstructured(anyhow::anyhow!(
1657                                        "ingestion-based sources must have cluster specified"
1658                                    )),
1659                                    cached_expr,
1660                                ));
1661                            }
1662                        },
1663                    },
1664                    mz_sql::plan::DataSourceDesc::IngestionExport {
1665                        ingestion_id,
1666                        external_reference,
1667                        details,
1668                        data_config,
1669                    } => DataSourceDesc::IngestionExport {
1670                        ingestion_id,
1671                        external_reference,
1672                        details,
1673                        data_config,
1674                    },
1675                    mz_sql::plan::DataSourceDesc::Progress => DataSourceDesc::Progress,
1676                    mz_sql::plan::DataSourceDesc::Webhook {
1677                        validate_using,
1678                        body_format,
1679                        headers,
1680                        cluster_id,
1681                    } => {
1682                        mz_ore::soft_assert_or_log!(
1683                            cluster_id.is_none(),
1684                            "cluster_id set at Source level for Webhooks"
1685                        );
1686                        DataSourceDesc::Webhook {
1687                            validate_using,
1688                            body_format,
1689                            headers,
1690                            cluster_id: in_cluster
1691                                .expect("webhook sources must use an existing cluster"),
1692                        }
1693                    }
1694                },
1695                desc: source.desc,
1696                global_id,
1697                timeline,
1698                resolved_ids,
1699                custom_logical_compaction_window: source
1700                    .compaction_window
1701                    .or(custom_logical_compaction_window),
1702                is_retained_metrics_object,
1703            }),
1704            Plan::CreateView(CreateViewPlan { view, .. }) => {
1705                // Collect optimizer parameters.
1706                let optimizer_config =
1707                    optimize::OptimizerConfig::from(session_catalog.system_vars());
1708                let previous_exprs = previous_item.map(|item| match item {
1709                    CatalogItem::View(view) => Some((view.raw_expr, view.locally_optimized_expr)),
1710                    _ => None,
1711                });
1712
1713                let (raw_expr, optimized_expr) = match (cached_expr, previous_exprs) {
1714                    (Some(local_expr), _)
1715                        if local_expr.optimizer_features == optimizer_config.features =>
1716                    {
1717                        debug!("local expression cache hit for {global_id:?}");
1718                        (Arc::new(view.expr), Arc::new(local_expr.local_mir))
1719                    }
1720                    // If the new expr is equivalent to the old expr, then we don't need to re-optimize.
1721                    (_, Some(Some((raw_expr, optimized_expr)))) if *raw_expr == view.expr => {
1722                        (Arc::clone(&raw_expr), Arc::clone(&optimized_expr))
1723                    }
1724                    (cached_expr, _) => {
1725                        let optimizer_features = optimizer_config.features.clone();
1726                        // Build an optimizer for this VIEW.
1727                        let mut optimizer = optimize::view::Optimizer::new(optimizer_config, None);
1728
1729                        // HIR ⇒ MIR lowering and MIR ⇒ MIR optimization (local)
1730                        let raw_expr = view.expr;
1731                        let optimized_expr = match optimizer.optimize(raw_expr.clone()) {
1732                            Ok(optimzed_expr) => optimzed_expr,
1733                            Err(err) => return Err((err.into(), cached_expr)),
1734                        };
1735
1736                        uncached_expr = Some((optimized_expr.clone(), optimizer_features));
1737
1738                        (Arc::new(raw_expr), Arc::new(optimized_expr))
1739                    }
1740                };
1741
1742                // Resolve all item dependencies from the HIR expression.
1743                let dependencies: BTreeSet<_> = raw_expr
1744                    .depends_on()
1745                    .into_iter()
1746                    .map(|gid| self.get_entry_by_global_id(&gid).id())
1747                    .collect();
1748
1749                let typ = infer_sql_type_for_catalog(&raw_expr, &optimized_expr);
1750                CatalogItem::View(View {
1751                    create_sql: view.create_sql,
1752                    global_id,
1753                    raw_expr,
1754                    desc: RelationDesc::new(typ, view.column_names),
1755                    locally_optimized_expr: optimized_expr,
1756                    conn_id: None,
1757                    resolved_ids,
1758                    dependencies: DependencyIds(dependencies),
1759                })
1760            }
1761            Plan::CreateMaterializedView(CreateMaterializedViewPlan {
1762                materialized_view, ..
1763            }) => {
1764                let collections = extra_versions
1765                    .iter()
1766                    .map(|(version, gid)| (*version, *gid))
1767                    .chain([(RelationVersion::root(), global_id)].into_iter())
1768                    .collect();
1769
1770                // Collect optimizer parameters.
1771                let system_vars = session_catalog.system_vars();
1772                let overrides = self
1773                    .get_cluster(materialized_view.cluster_id)
1774                    .config
1775                    .features();
1776                let optimizer_config =
1777                    optimize::OptimizerConfig::from(system_vars).override_from(&overrides);
1778                let previous_exprs = previous_item.map(|item| match item {
1779                    CatalogItem::MaterializedView(materialized_view) => (
1780                        materialized_view.raw_expr,
1781                        materialized_view.locally_optimized_expr,
1782                    ),
1783                    item => unreachable!("expected materialized view, found: {item:#?}"),
1784                });
1785
1786                let (raw_expr, optimized_expr) = match (cached_expr, previous_exprs) {
1787                    (Some(local_expr), _)
1788                        if local_expr.optimizer_features == optimizer_config.features =>
1789                    {
1790                        debug!("local expression cache hit for {global_id:?}");
1791                        (
1792                            Arc::new(materialized_view.expr),
1793                            Arc::new(local_expr.local_mir),
1794                        )
1795                    }
1796                    // If the new expr is equivalent to the old expr, then we don't need to re-optimize.
1797                    (_, Some((raw_expr, optimized_expr)))
1798                        if *raw_expr == materialized_view.expr =>
1799                    {
1800                        (Arc::clone(&raw_expr), Arc::clone(&optimized_expr))
1801                    }
1802                    (cached_expr, _) => {
1803                        let optimizer_features = optimizer_config.features.clone();
1804                        // TODO(aalexandrov): ideally this should be a materialized_view::Optimizer.
1805                        let mut optimizer = optimize::view::Optimizer::new(optimizer_config, None);
1806
1807                        let raw_expr = materialized_view.expr;
1808                        let optimized_expr = match optimizer.optimize(raw_expr.clone()) {
1809                            Ok(optimized_expr) => optimized_expr,
1810                            Err(err) => return Err((err.into(), cached_expr)),
1811                        };
1812
1813                        uncached_expr = Some((optimized_expr.clone(), optimizer_features));
1814
1815                        (Arc::new(raw_expr), Arc::new(optimized_expr))
1816                    }
1817                };
1818                let mut typ = infer_sql_type_for_catalog(&raw_expr, &optimized_expr);
1819
1820                for &i in &materialized_view.non_null_assertions {
1821                    typ.column_types[i].nullable = false;
1822                }
1823                let desc = RelationDesc::new(typ, materialized_view.column_names);
1824                let desc = VersionedRelationDesc::new(desc);
1825
1826                let initial_as_of = materialized_view.as_of.map(Antichain::from_elem);
1827
1828                // Resolve all item dependencies from the HIR expression.
1829                let dependencies = raw_expr
1830                    .depends_on()
1831                    .into_iter()
1832                    .map(|gid| self.get_entry_by_global_id(&gid).id())
1833                    .collect();
1834
1835                CatalogItem::MaterializedView(MaterializedView {
1836                    create_sql: materialized_view.create_sql,
1837                    collections,
1838                    raw_expr,
1839                    locally_optimized_expr: optimized_expr,
1840                    desc,
1841                    resolved_ids,
1842                    dependencies,
1843                    replacement_target: materialized_view.replacement_target,
1844                    cluster_id: materialized_view.cluster_id,
1845                    target_replica: materialized_view.target_replica,
1846                    non_null_assertions: materialized_view.non_null_assertions,
1847                    custom_logical_compaction_window: materialized_view.compaction_window,
1848                    refresh_schedule: materialized_view.refresh_schedule,
1849                    initial_as_of,
1850                    optimized_plan: None,
1851                    physical_plan: None,
1852                    dataflow_metainfo: None,
1853                })
1854            }
1855            Plan::CreateIndex(CreateIndexPlan { index, .. }) => CatalogItem::Index(Index {
1856                create_sql: index.create_sql,
1857                global_id,
1858                on: index.on,
1859                keys: index.keys.into(),
1860                conn_id: None,
1861                resolved_ids,
1862                cluster_id: index.cluster_id,
1863                custom_logical_compaction_window: custom_logical_compaction_window
1864                    .or(index.compaction_window),
1865                is_retained_metrics_object,
1866                optimized_plan: None,
1867                physical_plan: None,
1868                dataflow_metainfo: None,
1869            }),
1870            Plan::CreateMetricSink(CreateMetricSinkPlan { metric_sink, .. }) => {
1871                CatalogItem::MetricSink(MetricSink {
1872                    create_sql: metric_sink.create_sql,
1873                    global_id,
1874                    from: metric_sink.from,
1875                    resolved_ids,
1876                    cluster_id: metric_sink.cluster_id,
1877                    prefix: metric_sink.prefix,
1878                    optimized_plan: None,
1879                    physical_plan: None,
1880                    dataflow_metainfo: None,
1881                })
1882            }
1883            Plan::CreateSink(CreateSinkPlan {
1884                sink,
1885                with_snapshot,
1886                in_cluster,
1887                ..
1888            }) => CatalogItem::Sink(Sink {
1889                create_sql: sink.create_sql,
1890                global_id,
1891                from: sink.from,
1892                connection: sink.connection,
1893                envelope: sink.envelope,
1894                version: sink.version,
1895                with_snapshot,
1896                resolved_ids,
1897                cluster_id: in_cluster,
1898                commit_interval: sink.commit_interval,
1899            }),
1900            Plan::CreateType(CreateTypePlan { typ, .. }) => {
1901                // Even if we don't need the `RelationDesc` here, error out
1902                // early and eagerly, as a kind of soft assertion that we _can_
1903                // build the `RelationDesc` when needed.
1904                if let Err(err) = typ.inner.desc(&session_catalog) {
1905                    return Err((err.into(), cached_expr));
1906                }
1907                CatalogItem::Type(Type {
1908                    create_sql: Some(typ.create_sql),
1909                    global_id,
1910                    details: CatalogTypeDetails {
1911                        array_id: None,
1912                        typ: typ.inner,
1913                        pg_metadata: None,
1914                    },
1915                    resolved_ids,
1916                })
1917            }
1918            Plan::CreateSecret(CreateSecretPlan { secret, .. }) => CatalogItem::Secret(Secret {
1919                create_sql: secret.create_sql,
1920                global_id,
1921            }),
1922            Plan::CreateConnection(CreateConnectionPlan {
1923                connection:
1924                    mz_sql::plan::Connection {
1925                        create_sql,
1926                        details,
1927                    },
1928                ..
1929            }) => CatalogItem::Connection(Connection {
1930                create_sql,
1931                global_id,
1932                details,
1933                resolved_ids,
1934            }),
1935            _ => {
1936                return Err((
1937                    Error::new(ErrorKind::Corruption {
1938                        detail: "catalog entry generated inappropriate plan".to_string(),
1939                    })
1940                    .into(),
1941                    cached_expr,
1942                ));
1943            }
1944        };
1945
1946        // Carry over the plans (`optimized_plan`, `physical_plan`,
1947        // `dataflow_metainfo`) from the previous incarnation of this item, if
1948        // any. See the comment on `previous_plans` above.
1949        if let Some((prev_optimized, prev_physical, prev_metainfo)) = previous_plans {
1950            if let Some((optimized_plan, physical_plan, dataflow_metainfo)) = item.plan_fields_mut()
1951            {
1952                *optimized_plan = prev_optimized;
1953                *physical_plan = prev_physical;
1954                *dataflow_metainfo = prev_metainfo;
1955            }
1956        }
1957
1958        Ok((item, uncached_expr))
1959    }
1960
1961    /// Execute function `f` on `self`, with all "enable_for_item_parsing" feature flags enabled.
1962    /// Calling this method will not permanently modify any system configuration variables.
1963    ///
1964    /// WARNING:
1965    /// Any modifications made to the system configuration variables in `f`, will be lost.
1966    pub fn with_enable_for_item_parsing<T>(&mut self, f: impl FnOnce(&mut Self) -> T) -> T {
1967        // Enable catalog features that might be required during planning existing
1968        // catalog items. Existing catalog items might have been created while
1969        // a specific feature flag was turned on, so we need to ensure that this
1970        // is also the case during catalog rehydration in order to avoid panics.
1971        //
1972        // WARNING / CONTRACT:
1973        // 1. Features used in this method that related to parsing / planning
1974        //    should be `enable_for_item_parsing` set to `true`.
1975        // 2. After this step, feature flag configuration must not be
1976        //    overridden.
1977        // 3. We don't notify `SystemVars` callbacks here, neither for the
1978        //    flags this enables nor for the `Arc` restore that undoes them
1979        //    afterwards. A callback on a `feature_flags!` var therefore won't
1980        //    observe this transient flip, only committed changes to it. See
1981        //    `SystemVars::register_callback`.
1982        let restore = Arc::clone(&self.system_configuration);
1983        Arc::make_mut(&mut self.system_configuration).enable_for_item_parsing();
1984        let res = f(self);
1985        self.system_configuration = restore;
1986        res
1987    }
1988
1989    /// Returns all indexes on the given object and cluster known in the catalog.
1990    pub fn get_indexes_on(
1991        &self,
1992        id: GlobalId,
1993        cluster: ClusterId,
1994    ) -> impl Iterator<Item = (GlobalId, &Index)> {
1995        let index_matches = move |idx: &Index| idx.on == id && idx.cluster_id == cluster;
1996
1997        self.try_get_entry_by_global_id(&id)
1998            .into_iter()
1999            .map(move |e| {
2000                e.used_by()
2001                    .iter()
2002                    .filter_map(move |uses_id| match self.get_entry(uses_id).item() {
2003                        CatalogItem::Index(index) if index_matches(index) => {
2004                            Some((index.global_id(), index))
2005                        }
2006                        _ => None,
2007                    })
2008            })
2009            .flatten()
2010    }
2011
2012    pub(super) fn get_database(&self, database_id: &DatabaseId) -> &Database {
2013        &self.database_by_id[database_id]
2014    }
2015
2016    /// Gets a reference to the specified replica of the specified cluster.
2017    ///
2018    /// Returns `None` if either the cluster or the replica does not
2019    /// exist.
2020    pub(super) fn try_get_cluster_replica(
2021        &self,
2022        id: ClusterId,
2023        replica_id: ReplicaId,
2024    ) -> Option<&ClusterReplica> {
2025        self.try_get_cluster(id)
2026            .and_then(|cluster| cluster.replica(replica_id))
2027    }
2028
2029    /// Gets a reference to the specified replica of the specified cluster.
2030    ///
2031    /// Panics if either the cluster or the replica does not exist.
2032    pub(crate) fn get_cluster_replica(
2033        &self,
2034        cluster_id: ClusterId,
2035        replica_id: ReplicaId,
2036    ) -> &ClusterReplica {
2037        self.try_get_cluster_replica(cluster_id, replica_id)
2038            .unwrap_or_else(|| panic!("unknown cluster replica: {cluster_id}.{replica_id}"))
2039    }
2040
2041    pub(super) fn resolve_replica_in_cluster(
2042        &self,
2043        cluster_id: &ClusterId,
2044        replica_name: &str,
2045    ) -> Result<&ClusterReplica, SqlCatalogError> {
2046        let cluster = self.get_cluster(*cluster_id);
2047        let replica_id = cluster
2048            .replica_id_by_name_
2049            .get(replica_name)
2050            .ok_or_else(|| SqlCatalogError::UnknownClusterReplica(replica_name.to_string()))?;
2051        Ok(&cluster.replicas_by_id_[replica_id])
2052    }
2053
2054    /// Get system configuration `name`.
2055    pub fn get_system_configuration(&self, name: &str) -> Result<&dyn Var, Error> {
2056        Ok(self.system_configuration.get(name)?)
2057    }
2058
2059    /// Parse system configuration `name` with `value` int.
2060    ///
2061    /// Returns the parsed value as a string.
2062    pub(super) fn parse_system_configuration(
2063        &self,
2064        name: &str,
2065        value: VarInput,
2066    ) -> Result<String, Error> {
2067        let value = self.system_configuration.parse(name, value)?;
2068        Ok(value.format())
2069    }
2070
2071    /// Gets the schema map for the database matching `database_spec`.
2072    pub(super) fn resolve_schema_in_database(
2073        &self,
2074        database_spec: &ResolvedDatabaseSpecifier,
2075        schema_name: &str,
2076        conn_id: &ConnectionId,
2077    ) -> Result<&Schema, SqlCatalogError> {
2078        let schema = match database_spec {
2079            ResolvedDatabaseSpecifier::Ambient if schema_name == MZ_TEMP_SCHEMA => {
2080                self.temporary_namespaces.schema(conn_id)
2081            }
2082            ResolvedDatabaseSpecifier::Ambient => self
2083                .ambient_schemas_by_name
2084                .get(schema_name)
2085                .and_then(|id| self.ambient_schemas_by_id.get(id)),
2086            ResolvedDatabaseSpecifier::Id(id) => self.database_by_id.get(id).and_then(|db| {
2087                db.schemas_by_name
2088                    .get(schema_name)
2089                    .and_then(|id| db.schemas_by_id.get(id))
2090            }),
2091        };
2092        schema.ok_or_else(|| SqlCatalogError::UnknownSchema(schema_name.into()))
2093    }
2094
2095    /// Try to get a schema, returning `None` if it doesn't exist.
2096    ///
2097    /// For temporary schemas, returns `None` if the connection's temporary
2098    /// namespace hasn't been registered yet (that happens at its first
2099    /// temporary-item creation).
2100    pub fn try_get_schema(
2101        &self,
2102        database_spec: &ResolvedDatabaseSpecifier,
2103        schema_spec: &SchemaSpecifier,
2104        conn_id: &ConnectionId,
2105    ) -> Option<&Schema> {
2106        // Keep in sync with `get_schema` and `get_schemas_mut`
2107        match (database_spec, schema_spec) {
2108            (ResolvedDatabaseSpecifier::Ambient, SchemaSpecifier::Temporary) => {
2109                self.temporary_namespaces.schema(conn_id)
2110            }
2111            (ResolvedDatabaseSpecifier::Ambient, SchemaSpecifier::Id(id)) => {
2112                self.ambient_schemas_by_id.get(id)
2113            }
2114            (ResolvedDatabaseSpecifier::Id(database_id), SchemaSpecifier::Id(schema_id)) => self
2115                .database_by_id
2116                .get(database_id)
2117                .and_then(|db| db.schemas_by_id.get(schema_id)),
2118            (ResolvedDatabaseSpecifier::Id(_), SchemaSpecifier::Temporary) => {
2119                unreachable!("temporary schemas are in the ambient database")
2120            }
2121        }
2122    }
2123
2124    pub fn get_schema(
2125        &self,
2126        database_spec: &ResolvedDatabaseSpecifier,
2127        schema_spec: &SchemaSpecifier,
2128        conn_id: &ConnectionId,
2129    ) -> &Schema {
2130        // Keep in sync with `try_get_schema` and `get_schemas_mut`
2131        self.try_get_schema(database_spec, schema_spec, conn_id)
2132            .expect("schema must exist")
2133    }
2134
2135    pub(super) fn find_non_temp_schema(&self, schema_id: &SchemaId) -> &Schema {
2136        self.database_by_id
2137            .values()
2138            .filter_map(|database| database.schemas_by_id.get(schema_id))
2139            .chain(self.ambient_schemas_by_id.values())
2140            .filter(|schema| schema.id() == &SchemaSpecifier::from(*schema_id))
2141            .into_first()
2142    }
2143
2144    pub fn get_mz_catalog_schema_id(&self) -> SchemaId {
2145        self.ambient_schemas_by_name[MZ_CATALOG_SCHEMA]
2146    }
2147
2148    pub fn get_mz_catalog_unstable_schema_id(&self) -> SchemaId {
2149        self.ambient_schemas_by_name[MZ_CATALOG_UNSTABLE_SCHEMA]
2150    }
2151
2152    pub fn get_pg_catalog_schema_id(&self) -> SchemaId {
2153        self.ambient_schemas_by_name[PG_CATALOG_SCHEMA]
2154    }
2155
2156    pub fn get_information_schema_id(&self) -> SchemaId {
2157        self.ambient_schemas_by_name[INFORMATION_SCHEMA]
2158    }
2159
2160    pub fn get_mz_internal_schema_id(&self) -> SchemaId {
2161        self.ambient_schemas_by_name[MZ_INTERNAL_SCHEMA]
2162    }
2163
2164    pub fn get_mz_introspection_schema_id(&self) -> SchemaId {
2165        self.ambient_schemas_by_name[MZ_INTROSPECTION_SCHEMA]
2166    }
2167
2168    pub fn get_mz_unsafe_schema_id(&self) -> SchemaId {
2169        self.ambient_schemas_by_name[MZ_UNSAFE_SCHEMA]
2170    }
2171
2172    pub fn system_schema_ids(&self) -> impl Iterator<Item = SchemaId> + '_ {
2173        SYSTEM_SCHEMAS
2174            .iter()
2175            .map(|name| self.ambient_schemas_by_name[*name])
2176    }
2177
2178    pub fn is_system_schema_id(&self, id: SchemaId) -> bool {
2179        self.system_schema_ids().contains(&id)
2180    }
2181
2182    pub fn is_system_schema_specifier(&self, spec: SchemaSpecifier) -> bool {
2183        match spec {
2184            SchemaSpecifier::Temporary => false,
2185            SchemaSpecifier::Id(id) => self.is_system_schema_id(id),
2186        }
2187    }
2188
2189    pub fn unstable_schema_ids(&self) -> impl Iterator<Item = SchemaId> + '_ {
2190        UNSTABLE_SCHEMAS
2191            .iter()
2192            .map(|name| self.ambient_schemas_by_name[*name])
2193    }
2194
2195    pub fn is_unstable_schema_id(&self, id: SchemaId) -> bool {
2196        self.unstable_schema_ids().contains(&id)
2197    }
2198
2199    pub fn is_unstable_schema_specifier(&self, spec: SchemaSpecifier) -> bool {
2200        match spec {
2201            SchemaSpecifier::Temporary => false,
2202            SchemaSpecifier::Id(id) => self.is_unstable_schema_id(id),
2203        }
2204    }
2205
2206    /// Return all OIDs that are allocated to temporary objects.
2207    pub(crate) fn get_temporary_oids(&self) -> impl Iterator<Item = u32> + '_ {
2208        std::iter::empty()
2209            .chain(self.ambient_schemas_by_id.values().filter_map(|schema| {
2210                if schema.id.is_temporary() {
2211                    Some(schema.oid)
2212                } else {
2213                    None
2214                }
2215            }))
2216            .chain(self.entry_by_id.values().filter_map(|entry| {
2217                if entry.item().is_temporary() {
2218                    Some(entry.oid)
2219                } else {
2220                    None
2221                }
2222            }))
2223    }
2224
2225    /// Optimized lookup for a builtin table.
2226    ///
2227    /// Panics if the builtin table doesn't exist in the catalog.
2228    pub fn resolve_builtin_table(&self, builtin: &'static BuiltinTable) -> CatalogItemId {
2229        self.resolve_builtin_object(&Builtin::<IdReference>::Table(builtin))
2230    }
2231
2232    /// Optimized lookup for a builtin log.
2233    ///
2234    /// Panics if the builtin log doesn't exist in the catalog.
2235    pub fn resolve_builtin_log(&self, builtin: &'static BuiltinLog) -> (CatalogItemId, GlobalId) {
2236        let item_id = self.resolve_builtin_object(&Builtin::<IdReference>::Log(builtin));
2237        let log = match self.get_entry(&item_id).item() {
2238            CatalogItem::Log(log) => log,
2239            other => unreachable!("programming error, expected BuiltinLog, found {other:?}"),
2240        };
2241        (item_id, log.global_id)
2242    }
2243
2244    /// Optimized lookup for a builtin storage collection.
2245    ///
2246    /// Panics if the builtin storage collection doesn't exist in the catalog.
2247    pub fn resolve_builtin_source(&self, builtin: &'static BuiltinSource) -> CatalogItemId {
2248        self.resolve_builtin_object(&Builtin::<IdReference>::Source(builtin))
2249    }
2250
2251    /// Optimized lookup for a builtin object.
2252    ///
2253    /// Panics if the builtin object doesn't exist in the catalog.
2254    pub fn resolve_builtin_object<T: TypeReference>(&self, builtin: &Builtin<T>) -> CatalogItemId {
2255        let schema_id = &self.ambient_schemas_by_name[builtin.schema()];
2256        let schema = &self.ambient_schemas_by_id[schema_id];
2257        match builtin.catalog_item_type() {
2258            CatalogItemType::Type => schema.types[builtin.name()],
2259            CatalogItemType::Func => schema.functions[builtin.name()],
2260            CatalogItemType::Table
2261            | CatalogItemType::Source
2262            | CatalogItemType::Sink
2263            | CatalogItemType::MetricSink
2264            | CatalogItemType::View
2265            | CatalogItemType::MaterializedView
2266            | CatalogItemType::Index
2267            | CatalogItemType::Secret
2268            | CatalogItemType::Connection => schema.items[builtin.name()],
2269        }
2270    }
2271
2272    /// Resolve a [`BuiltinType<NameReference>`] to a [`BuiltinType<IdReference>`].
2273    pub fn resolve_builtin_type_references(
2274        &self,
2275        builtin: &BuiltinType<NameReference>,
2276    ) -> BuiltinType<IdReference> {
2277        let typ: CatalogType<IdReference> = match &builtin.details.typ {
2278            CatalogType::AclItem => CatalogType::AclItem,
2279            CatalogType::Array { element_reference } => CatalogType::Array {
2280                element_reference: self.get_system_type(element_reference).id,
2281            },
2282            CatalogType::List {
2283                element_reference,
2284                element_modifiers,
2285            } => CatalogType::List {
2286                element_reference: self.get_system_type(element_reference).id,
2287                element_modifiers: element_modifiers.clone(),
2288            },
2289            CatalogType::Map {
2290                key_reference,
2291                value_reference,
2292                key_modifiers,
2293                value_modifiers,
2294            } => CatalogType::Map {
2295                key_reference: self.get_system_type(key_reference).id,
2296                value_reference: self.get_system_type(value_reference).id,
2297                key_modifiers: key_modifiers.clone(),
2298                value_modifiers: value_modifiers.clone(),
2299            },
2300            CatalogType::Range { element_reference } => CatalogType::Range {
2301                element_reference: self.get_system_type(element_reference).id,
2302            },
2303            CatalogType::Record { fields } => CatalogType::Record {
2304                fields: fields
2305                    .into_iter()
2306                    .map(|f| CatalogRecordField {
2307                        name: f.name.clone(),
2308                        type_reference: self.get_system_type(f.type_reference).id,
2309                        type_modifiers: f.type_modifiers.clone(),
2310                    })
2311                    .collect(),
2312            },
2313            CatalogType::Bool => CatalogType::Bool,
2314            CatalogType::Bytes => CatalogType::Bytes,
2315            CatalogType::Char => CatalogType::Char,
2316            CatalogType::Date => CatalogType::Date,
2317            CatalogType::Float32 => CatalogType::Float32,
2318            CatalogType::Float64 => CatalogType::Float64,
2319            CatalogType::Int16 => CatalogType::Int16,
2320            CatalogType::Int32 => CatalogType::Int32,
2321            CatalogType::Int64 => CatalogType::Int64,
2322            CatalogType::UInt16 => CatalogType::UInt16,
2323            CatalogType::UInt32 => CatalogType::UInt32,
2324            CatalogType::UInt64 => CatalogType::UInt64,
2325            CatalogType::MzTimestamp => CatalogType::MzTimestamp,
2326            CatalogType::Interval => CatalogType::Interval,
2327            CatalogType::Jsonb => CatalogType::Jsonb,
2328            CatalogType::Numeric => CatalogType::Numeric,
2329            CatalogType::Oid => CatalogType::Oid,
2330            CatalogType::PgLegacyChar => CatalogType::PgLegacyChar,
2331            CatalogType::PgLegacyName => CatalogType::PgLegacyName,
2332            CatalogType::Pseudo => CatalogType::Pseudo,
2333            CatalogType::RegClass => CatalogType::RegClass,
2334            CatalogType::RegProc => CatalogType::RegProc,
2335            CatalogType::RegType => CatalogType::RegType,
2336            CatalogType::String => CatalogType::String,
2337            CatalogType::Time => CatalogType::Time,
2338            CatalogType::Timestamp => CatalogType::Timestamp,
2339            CatalogType::TimestampTz => CatalogType::TimestampTz,
2340            CatalogType::Uuid => CatalogType::Uuid,
2341            CatalogType::VarChar => CatalogType::VarChar,
2342            CatalogType::Int2Vector => CatalogType::Int2Vector,
2343            CatalogType::MzAclItem => CatalogType::MzAclItem,
2344        };
2345
2346        BuiltinType {
2347            name: builtin.name,
2348            schema: builtin.schema,
2349            oid: builtin.oid,
2350            details: CatalogTypeDetails {
2351                array_id: builtin.details.array_id,
2352                typ,
2353                pg_metadata: builtin.details.pg_metadata.clone(),
2354            },
2355        }
2356    }
2357
2358    pub fn config(&self) -> &mz_sql::catalog::CatalogConfig {
2359        &self.config
2360    }
2361
2362    pub fn resolve_database(&self, database_name: &str) -> Result<&Database, SqlCatalogError> {
2363        match self.database_by_name.get(database_name) {
2364            Some(id) => Ok(&self.database_by_id[id]),
2365            None => Err(SqlCatalogError::UnknownDatabase(database_name.into())),
2366        }
2367    }
2368
2369    pub fn resolve_schema(
2370        &self,
2371        current_database: Option<&DatabaseId>,
2372        database_name: Option<&str>,
2373        schema_name: &str,
2374        conn_id: &ConnectionId,
2375    ) -> Result<&Schema, SqlCatalogError> {
2376        let database_spec = match database_name {
2377            // If a database is explicitly specified, validate it. Note that we
2378            // intentionally do not validate `current_database` to permit
2379            // querying `mz_catalog` with an invalid session database, e.g., so
2380            // that you can run `SHOW DATABASES` to *find* a valid database.
2381            Some(database) => Some(ResolvedDatabaseSpecifier::Id(
2382                self.resolve_database(database)?.id().clone(),
2383            )),
2384            None => current_database.map(|id| ResolvedDatabaseSpecifier::Id(id.clone())),
2385        };
2386
2387        // First try to find the schema in the named database.
2388        if let Some(database_spec) = database_spec {
2389            if let Ok(schema) =
2390                self.resolve_schema_in_database(&database_spec, schema_name, conn_id)
2391            {
2392                return Ok(schema);
2393            }
2394        }
2395
2396        // Then fall back to the ambient database.
2397        if let Ok(schema) = self.resolve_schema_in_database(
2398            &ResolvedDatabaseSpecifier::Ambient,
2399            schema_name,
2400            conn_id,
2401        ) {
2402            return Ok(schema);
2403        }
2404
2405        Err(SqlCatalogError::UnknownSchema(schema_name.into()))
2406    }
2407
2408    /// Optimized lookup for a system schema.
2409    ///
2410    /// Panics if the system schema doesn't exist in the catalog.
2411    pub fn resolve_system_schema(&self, name: &'static str) -> SchemaId {
2412        self.ambient_schemas_by_name[name]
2413    }
2414
2415    pub fn resolve_search_path(
2416        &self,
2417        session: &dyn SessionMetadata,
2418    ) -> Vec<(ResolvedDatabaseSpecifier, SchemaSpecifier)> {
2419        let database = self
2420            .database_by_name
2421            .get(session.database())
2422            .map(|id| id.clone());
2423
2424        // NOTE: This drops schemas that don't resolve, and consumers rely on
2425        // every returned entry existing (e.g. `allocate_full_name` and the
2426        // `current_schemas` evaluation look schemas up infallibly). In
2427        // particular, `mz_temp` is dropped until the session's temporary
2428        // namespace is registered at its first temporary-item creation.
2429        session
2430            .search_path()
2431            .iter()
2432            .map(|schema| {
2433                self.resolve_schema(database.as_ref(), None, schema.as_str(), session.conn_id())
2434            })
2435            .filter_map(|schema| schema.ok())
2436            .map(|schema| (schema.name().database.clone(), schema.id().clone()))
2437            .collect()
2438    }
2439
2440    pub fn effective_search_path(
2441        &self,
2442        search_path: &[(ResolvedDatabaseSpecifier, SchemaSpecifier)],
2443        include_temp_schema: bool,
2444    ) -> Vec<(ResolvedDatabaseSpecifier, SchemaSpecifier)> {
2445        let mut v = Vec::with_capacity(search_path.len() + 3);
2446        // Temp schema is only included for relations and data types, not for functions and operators
2447        let temp_schema = (
2448            ResolvedDatabaseSpecifier::Ambient,
2449            SchemaSpecifier::Temporary,
2450        );
2451        if include_temp_schema && !search_path.contains(&temp_schema) {
2452            v.push(temp_schema);
2453        }
2454        let default_schemas = [
2455            (
2456                ResolvedDatabaseSpecifier::Ambient,
2457                SchemaSpecifier::Id(self.get_mz_catalog_schema_id()),
2458            ),
2459            (
2460                ResolvedDatabaseSpecifier::Ambient,
2461                SchemaSpecifier::Id(self.get_pg_catalog_schema_id()),
2462            ),
2463        ];
2464        for schema in default_schemas.into_iter() {
2465            if !search_path.contains(&schema) {
2466                v.push(schema);
2467            }
2468        }
2469        v.extend_from_slice(search_path);
2470        v
2471    }
2472
2473    pub fn resolve_cluster(&self, name: &str) -> Result<&Cluster, SqlCatalogError> {
2474        let id = self
2475            .clusters_by_name
2476            .get(name)
2477            .ok_or_else(|| SqlCatalogError::UnknownCluster(name.to_string()))?;
2478        Ok(&self.clusters_by_id[id])
2479    }
2480
2481    pub fn resolve_builtin_cluster(&self, cluster: &BuiltinCluster) -> &Cluster {
2482        let id = self
2483            .clusters_by_name
2484            .get(cluster.name)
2485            .expect("failed to lookup BuiltinCluster by name");
2486        self.clusters_by_id
2487            .get(id)
2488            .expect("failed to lookup BuiltinCluster by ID")
2489    }
2490
2491    pub fn resolve_cluster_replica(
2492        &self,
2493        cluster_replica_name: &QualifiedReplica,
2494    ) -> Result<&ClusterReplica, SqlCatalogError> {
2495        let cluster = self.resolve_cluster(cluster_replica_name.cluster.as_str())?;
2496        let replica_name = cluster_replica_name.replica.as_str();
2497        let replica_id = cluster
2498            .replica_id(replica_name)
2499            .ok_or_else(|| SqlCatalogError::UnknownClusterReplica(replica_name.to_string()))?;
2500        Ok(cluster.replica(replica_id).expect("Must exist"))
2501    }
2502
2503    /// Resolves [`PartialItemName`] into a [`CatalogEntry`].
2504    ///
2505    /// If `name` does not specify a database, the `current_database` is used.
2506    /// If `name` does not specify a schema, then the schemas in `search_path`
2507    /// are searched in order.
2508    #[allow(clippy::useless_let_if_seq)]
2509    pub fn resolve(
2510        &self,
2511        get_schema_entries: fn(&Schema) -> &BTreeMap<String, CatalogItemId>,
2512        current_database: Option<&DatabaseId>,
2513        search_path: &Vec<(ResolvedDatabaseSpecifier, SchemaSpecifier)>,
2514        name: &PartialItemName,
2515        conn_id: &ConnectionId,
2516        err_gen: fn(String) -> SqlCatalogError,
2517    ) -> Result<&CatalogEntry, SqlCatalogError> {
2518        // If a schema name was specified, just try to find the item in that
2519        // schema. If no schema was specified, try to find the item in the connection's
2520        // temporary schema. If the item is not found, try to find the item in every
2521        // schema in the search path.
2522        let schemas = match &name.schema {
2523            Some(schema_name) => {
2524                match self.resolve_schema(
2525                    current_database,
2526                    name.database.as_deref(),
2527                    schema_name,
2528                    conn_id,
2529                ) {
2530                    Ok(schema) => vec![(schema.name.database.clone(), schema.id.clone())],
2531                    Err(e) => return Err(e),
2532                }
2533            }
2534            None => match self
2535                .try_get_schema(
2536                    &ResolvedDatabaseSpecifier::Ambient,
2537                    &SchemaSpecifier::Temporary,
2538                    conn_id,
2539                )
2540                .and_then(|schema| schema.items.get(&name.item))
2541            {
2542                Some(id) => return Ok(self.get_entry(id)),
2543                None => search_path.to_vec(),
2544            },
2545        };
2546
2547        for (database_spec, schema_spec) in &schemas {
2548            // Use try_get_schema because the temp schema might not exist yet
2549            // (it's created lazily when the first temp object is created).
2550            let Some(schema) = self.try_get_schema(database_spec, schema_spec, conn_id) else {
2551                continue;
2552            };
2553
2554            if let Some(id) = get_schema_entries(schema).get(&name.item) {
2555                return Ok(&self.entry_by_id[id]);
2556            }
2557        }
2558
2559        // Some relations that have previously lived in the `mz_internal` schema have been moved to
2560        // `mz_catalog_unstable` or `mz_introspection`. To simplify the transition for users, we
2561        // automatically let uses of the old schema resolve to the new ones as well.
2562        // TODO(database-issues#8173) remove this after sufficient time has passed
2563        let mz_internal_schema = SchemaSpecifier::Id(self.get_mz_internal_schema_id());
2564        if schemas.iter().any(|(_, spec)| *spec == mz_internal_schema) {
2565            for schema_id in [
2566                self.get_mz_catalog_unstable_schema_id(),
2567                self.get_mz_introspection_schema_id(),
2568            ] {
2569                let schema = self.get_schema(
2570                    &ResolvedDatabaseSpecifier::Ambient,
2571                    &SchemaSpecifier::Id(schema_id),
2572                    conn_id,
2573                );
2574
2575                if let Some(id) = get_schema_entries(schema).get(&name.item) {
2576                    debug!(
2577                        github_27831 = true,
2578                        "encountered use of outdated schema `mz_internal` for relation: {name}",
2579                    );
2580                    return Ok(&self.entry_by_id[id]);
2581                }
2582            }
2583        }
2584
2585        Err(err_gen(name.to_string()))
2586    }
2587
2588    /// Resolves `name` to a non-function [`CatalogEntry`].
2589    pub fn resolve_entry(
2590        &self,
2591        current_database: Option<&DatabaseId>,
2592        search_path: &Vec<(ResolvedDatabaseSpecifier, SchemaSpecifier)>,
2593        name: &PartialItemName,
2594        conn_id: &ConnectionId,
2595    ) -> Result<&CatalogEntry, SqlCatalogError> {
2596        self.resolve(
2597            |schema| &schema.items,
2598            current_database,
2599            search_path,
2600            name,
2601            conn_id,
2602            SqlCatalogError::UnknownItem,
2603        )
2604    }
2605
2606    /// Resolves `name` to a function [`CatalogEntry`].
2607    pub fn resolve_function(
2608        &self,
2609        current_database: Option<&DatabaseId>,
2610        search_path: &Vec<(ResolvedDatabaseSpecifier, SchemaSpecifier)>,
2611        name: &PartialItemName,
2612        conn_id: &ConnectionId,
2613    ) -> Result<&CatalogEntry, SqlCatalogError> {
2614        self.resolve(
2615            |schema| &schema.functions,
2616            current_database,
2617            search_path,
2618            name,
2619            conn_id,
2620            |name| SqlCatalogError::UnknownFunction {
2621                name,
2622                alternative: None,
2623            },
2624        )
2625    }
2626
2627    /// Resolves `name` to a type [`CatalogEntry`].
2628    pub fn resolve_type(
2629        &self,
2630        current_database: Option<&DatabaseId>,
2631        search_path: &Vec<(ResolvedDatabaseSpecifier, SchemaSpecifier)>,
2632        name: &PartialItemName,
2633        conn_id: &ConnectionId,
2634    ) -> Result<&CatalogEntry, SqlCatalogError> {
2635        static NON_PG_CATALOG_TYPES: LazyLock<
2636            BTreeMap<&'static str, &'static BuiltinType<NameReference>>,
2637        > = LazyLock::new(|| {
2638            BUILTINS::types()
2639                .filter(|typ| typ.schema != PG_CATALOG_SCHEMA)
2640                .map(|typ| (typ.name, typ))
2641                .collect()
2642        });
2643
2644        let entry = self.resolve(
2645            |schema| &schema.types,
2646            current_database,
2647            search_path,
2648            name,
2649            conn_id,
2650            |name| SqlCatalogError::UnknownType { name },
2651        )?;
2652
2653        if conn_id != &SYSTEM_CONN_ID && name.schema.as_deref() == Some(PG_CATALOG_SCHEMA) {
2654            if let Some(typ) = NON_PG_CATALOG_TYPES.get(entry.name().item.as_str()) {
2655                warn!(
2656                    "user specified an incorrect schema of {} for the type {}, which should be in \
2657                    the {} schema. This works now due to a bug but will be fixed in a later release.",
2658                    PG_CATALOG_SCHEMA.quoted(),
2659                    typ.name.quoted(),
2660                    typ.schema.quoted(),
2661                )
2662            }
2663        }
2664
2665        Ok(entry)
2666    }
2667
2668    /// For an [`ObjectId`] gets the corresponding [`CommentObjectId`].
2669    pub(super) fn get_comment_id(&self, object_id: ObjectId) -> CommentObjectId {
2670        match object_id {
2671            ObjectId::Item(item_id) => self.get_entry(&item_id).comment_object_id(),
2672            ObjectId::Role(role_id) => CommentObjectId::Role(role_id),
2673            ObjectId::Database(database_id) => CommentObjectId::Database(database_id),
2674            ObjectId::Schema((database, schema)) => CommentObjectId::Schema((database, schema)),
2675            ObjectId::Cluster(cluster_id) => CommentObjectId::Cluster(cluster_id),
2676            ObjectId::ClusterReplica(cluster_replica_id) => {
2677                CommentObjectId::ClusterReplica(cluster_replica_id)
2678            }
2679            ObjectId::NetworkPolicy(network_policy_id) => {
2680                CommentObjectId::NetworkPolicy(network_policy_id)
2681            }
2682        }
2683    }
2684
2685    /// Return current system configuration.
2686    pub fn system_config(&self) -> &SystemVars {
2687        &self.system_configuration
2688    }
2689
2690    /// Returns the cluster-coherent scoped optimizer-feature overrides for
2691    /// `cluster_id` from the in-memory scoped-parameter working copy, or empty
2692    /// if the cluster has none.
2693    pub fn cluster_scoped_optimizer_overrides(
2694        &self,
2695        cluster_id: ClusterId,
2696    ) -> OptimizerFeatureOverrides {
2697        self.scoped_system_parameters
2698            .cluster
2699            .get(&cluster_id)
2700            .cloned()
2701            .map(OptimizerFeatureOverrides::from)
2702            .unwrap_or_default()
2703    }
2704
2705    /// Returns the entire scoped system-parameter working copy, read by the
2706    /// coordinator's scoped-parameter reconcile path.
2707    pub fn scoped_system_parameters(&self) -> &ScopedParameters {
2708        &self.scoped_system_parameters
2709    }
2710
2711    /// Return a mutable reference to the current system configuration.
2712    pub fn system_config_mut(&mut self) -> &mut SystemVars {
2713        Arc::make_mut(&mut self.system_configuration)
2714    }
2715
2716    /// Serializes the catalog's in-memory state.
2717    ///
2718    /// There are no guarantees about the format of the serialized state, except
2719    /// that the serialized state for two identical catalogs will compare
2720    /// identically.
2721    ///
2722    /// Some consumers would like the ability to overwrite the `unfinalized_shards` catalog field,
2723    /// which they can accomplish by passing in a value of `Some` for the `unfinalized_shards`
2724    /// argument.
2725    pub fn dump(&self, unfinalized_shards: Option<BTreeSet<String>>) -> Result<String, Error> {
2726        // Dump the base catalog.
2727        let mut dump = serde_json::to_value(&self).map_err(|e| {
2728            Error::new(ErrorKind::Unstructured(format!(
2729                // Don't panic here because we don't have compile-time failures for maps with
2730                // non-string keys.
2731                "internal error: could not dump catalog: {}",
2732                e
2733            )))
2734        })?;
2735
2736        let dump_obj = dump.as_object_mut().expect("state must have been dumped");
2737        // Stitch in system parameter defaults.
2738        dump_obj.insert(
2739            "system_parameter_defaults".into(),
2740            serde_json::json!(self.system_config().defaults()),
2741        );
2742        // Potentially overwrite unfinalized shards.
2743        if let Some(unfinalized_shards) = unfinalized_shards {
2744            dump_obj
2745                .get_mut("storage_metadata")
2746                .expect("known to exist")
2747                .as_object_mut()
2748                .expect("storage_metadata is an object")
2749                .insert(
2750                    "unfinalized_shards".into(),
2751                    serde_json::json!(unfinalized_shards),
2752                );
2753        }
2754        // Remove GlobalIds for temporary objects from the mapping.
2755        //
2756        // Post-test consistency checks with the durable catalog don't know about temporary items
2757        // since they're kept entirely in memory.
2758        let temporary_gids: Vec<_> = self
2759            .entry_by_global_id
2760            .iter()
2761            .filter(|(_gid, item_id)| self.get_entry(item_id).conn_id().is_some())
2762            .map(|(gid, _item_id)| *gid)
2763            .collect();
2764        if !temporary_gids.is_empty() {
2765            let gids = dump_obj
2766                .get_mut("entry_by_global_id")
2767                .expect("known_to_exist")
2768                .as_object_mut()
2769                .expect("entry_by_global_id is an object");
2770            for gid in temporary_gids {
2771                gids.remove(&gid.to_string());
2772            }
2773        }
2774        // We exclude role_auth_by_id because it contains password information
2775        // which should not be included in the dump.
2776        dump_obj.remove("role_auth_by_id");
2777        // The mock authentication nonce is a server-wide secret used to derive
2778        // deterministic mock SASL challenges for absent or password-less roles.
2779        // Leaking it would re-enable the username enumeration the mock challenge
2780        // defends against, so it must not appear in the dump either.
2781        dump_obj.remove("mock_authentication_nonce");
2782
2783        // Emit as pretty-printed JSON.
2784        Ok(serde_json::to_string_pretty(&dump).expect("cannot fail on serde_json::Value"))
2785    }
2786
2787    pub fn availability_zones(&self) -> &[String] {
2788        &self.availability_zones
2789    }
2790
2791    pub fn concretize_replica_location(
2792        &self,
2793        location: mz_catalog::durable::ReplicaLocation,
2794        allowed_sizes: &Vec<String>,
2795        allowed_availability_zones: Option<&[String]>,
2796        allow_disabled: bool,
2797    ) -> Result<ReplicaLocation, Error> {
2798        let location = match location {
2799            mz_catalog::durable::ReplicaLocation::Unmanaged {
2800                storagectl_addrs,
2801                computectl_addrs,
2802            } => {
2803                if allowed_availability_zones.is_some() {
2804                    return Err(Error {
2805                        kind: ErrorKind::Internal(
2806                            "tried concretize unmanaged replica with specific availability_zones"
2807                                .to_string(),
2808                        ),
2809                    });
2810                }
2811                ReplicaLocation::Unmanaged(UnmanagedReplicaLocation {
2812                    storagectl_addrs,
2813                    computectl_addrs,
2814                })
2815            }
2816            mz_catalog::durable::ReplicaLocation::Managed {
2817                size,
2818                // The AZ list the replica was provisioned under: provisioning
2819                // paths pass the cluster's pool as `allowed_availability_zones`
2820                // to stamp it, while rebuilds from durable state pass `None` to
2821                // keep it. For an unmanaged cluster's replica it is the single
2822                // user-pinned AZ.
2823                availability_zones,
2824                billed_as,
2825                internal,
2826                pending,
2827            } => {
2828                self.ensure_valid_replica_size(allowed_sizes, &size, allow_disabled)?;
2829                let cluster_replica_sizes = &self.cluster_replica_sizes;
2830
2831                ReplicaLocation::Managed(ManagedReplicaLocation {
2832                    allocation: cluster_replica_sizes
2833                        .0
2834                        .get(&size)
2835                        .expect("catalog out of sync")
2836                        .clone(),
2837                    availability_zones: match allowed_availability_zones {
2838                        Some(azs) => azs.to_vec(),
2839                        None => availability_zones,
2840                    },
2841                    size,
2842                    billed_as,
2843                    internal,
2844                    pending,
2845                })
2846            }
2847        };
2848        Ok(location)
2849    }
2850
2851    pub(crate) fn ensure_valid_replica_size(
2852        &self,
2853        allowed_sizes: &[String],
2854        size: &String,
2855        allow_disabled: bool,
2856    ) -> Result<(), Error> {
2857        let cluster_replica_sizes = &self.cluster_replica_sizes;
2858
2859        if !cluster_replica_sizes.0.contains_key(size)
2860            || (!allowed_sizes.is_empty() && !allowed_sizes.contains(size))
2861            || (!allow_disabled && cluster_replica_sizes.0[size].disabled)
2862        {
2863            let mut entries = cluster_replica_sizes
2864                .enabled_allocations()
2865                .collect::<Vec<_>>();
2866
2867            if !allowed_sizes.is_empty() {
2868                let allowed_sizes = BTreeSet::<&String>::from_iter(allowed_sizes.iter());
2869                entries.retain(|(name, _)| allowed_sizes.contains(name));
2870            }
2871
2872            entries.sort_by_key(
2873                |(
2874                    _name,
2875                    ReplicaAllocation {
2876                        scale, cpu_limit, ..
2877                    },
2878                )| (scale, cpu_limit),
2879            );
2880
2881            Err(Error {
2882                kind: ErrorKind::InvalidClusterReplicaSize {
2883                    size: size.to_owned(),
2884                    expected: entries.into_iter().map(|(name, _)| name.clone()).collect(),
2885                },
2886            })
2887        } else {
2888            Ok(())
2889        }
2890    }
2891
2892    pub fn ensure_not_reserved_role(&self, role_id: &RoleId) -> Result<(), Error> {
2893        if role_id.is_builtin() {
2894            let role = self.get_role(role_id);
2895            Err(Error::new(ErrorKind::ReservedRoleName(
2896                role.name().to_string(),
2897            )))
2898        } else {
2899            Ok(())
2900        }
2901    }
2902
2903    pub fn ensure_not_reserved_network_policy(
2904        &self,
2905        network_policy_id: &NetworkPolicyId,
2906    ) -> Result<(), Error> {
2907        if network_policy_id.is_builtin() {
2908            let policy = self.get_network_policy(network_policy_id);
2909            Err(Error::new(ErrorKind::ReservedNetworkPolicyName(
2910                policy.name.clone(),
2911            )))
2912        } else {
2913            Ok(())
2914        }
2915    }
2916
2917    pub fn ensure_grantable_role(&self, role_id: &RoleId) -> Result<(), Error> {
2918        let is_grantable = !role_id.is_public() && !role_id.is_system();
2919        if is_grantable {
2920            Ok(())
2921        } else {
2922            let role = self.get_role(role_id);
2923            Err(Error::new(ErrorKind::UngrantableRoleName(
2924                role.name().to_string(),
2925            )))
2926        }
2927    }
2928
2929    pub fn ensure_not_system_role(&self, role_id: &RoleId) -> Result<(), Error> {
2930        if role_id.is_system() {
2931            let role = self.get_role(role_id);
2932            Err(Error::new(ErrorKind::ReservedSystemRoleName(
2933                role.name().to_string(),
2934            )))
2935        } else {
2936            Ok(())
2937        }
2938    }
2939
2940    pub fn ensure_not_predefined_role(&self, role_id: &RoleId) -> Result<(), Error> {
2941        if role_id.is_predefined() {
2942            let role = self.get_role(role_id);
2943            Err(Error::new(ErrorKind::ReservedSystemRoleName(
2944                role.name().to_string(),
2945            )))
2946        } else {
2947            Ok(())
2948        }
2949    }
2950
2951    // TODO(mjibson): Is there a way to make this a closure to avoid explicitly
2952    // passing tx, and session?
2953    pub(crate) fn add_to_audit_log(
2954        system_configuration: &SystemVars,
2955        oracle_write_ts: mz_repr::Timestamp,
2956        session: Option<&ConnMeta>,
2957        tx: &mut mz_catalog::durable::Transaction,
2958        audit_events: &mut Vec<VersionedEvent>,
2959        event_type: EventType,
2960        object_type: ObjectType,
2961        details: EventDetails,
2962    ) -> Result<(), Error> {
2963        let user = session.map(|session| session.user().name.to_string());
2964
2965        // unsafe_mock_audit_event_timestamp can only be set to Some when running in unsafe mode.
2966
2967        let occurred_at = match system_configuration.unsafe_mock_audit_event_timestamp() {
2968            Some(ts) => ts.into(),
2969            _ => oracle_write_ts.into(),
2970        };
2971        let id = tx.allocate_audit_log_id()?;
2972        let event = VersionedEvent::new(id, event_type, object_type, details, user, occurred_at);
2973        audit_events.push(event.clone());
2974        tx.insert_audit_log_event(event);
2975        Ok(())
2976    }
2977
2978    pub(super) fn get_owner_id(&self, id: &ObjectId, conn_id: &ConnectionId) -> Option<RoleId> {
2979        match id {
2980            ObjectId::Cluster(id) => Some(self.get_cluster(*id).owner_id()),
2981            ObjectId::ClusterReplica((cluster_id, replica_id)) => Some(
2982                self.get_cluster_replica(*cluster_id, *replica_id)
2983                    .owner_id(),
2984            ),
2985            ObjectId::Database(id) => Some(self.get_database(id).owner_id()),
2986            ObjectId::Schema((database_spec, schema_spec)) => Some(
2987                self.get_schema(database_spec, schema_spec, conn_id)
2988                    .owner_id(),
2989            ),
2990            ObjectId::Item(id) => Some(*self.get_entry(id).owner_id()),
2991            ObjectId::Role(_) => None,
2992            ObjectId::NetworkPolicy(id) => Some(self.get_network_policy(id).owner_id.clone()),
2993        }
2994    }
2995
2996    pub(super) fn get_object_type(&self, object_id: &ObjectId) -> mz_sql::catalog::ObjectType {
2997        match object_id {
2998            ObjectId::Cluster(_) => mz_sql::catalog::ObjectType::Cluster,
2999            ObjectId::ClusterReplica(_) => mz_sql::catalog::ObjectType::ClusterReplica,
3000            ObjectId::Database(_) => mz_sql::catalog::ObjectType::Database,
3001            ObjectId::Schema(_) => mz_sql::catalog::ObjectType::Schema,
3002            ObjectId::Role(_) => mz_sql::catalog::ObjectType::Role,
3003            ObjectId::Item(id) => self.get_entry(id).item_type().into(),
3004            ObjectId::NetworkPolicy(_) => mz_sql::catalog::ObjectType::NetworkPolicy,
3005        }
3006    }
3007
3008    pub(super) fn get_system_object_type(
3009        &self,
3010        id: &SystemObjectId,
3011    ) -> mz_sql::catalog::SystemObjectType {
3012        match id {
3013            SystemObjectId::Object(object_id) => {
3014                SystemObjectType::Object(self.get_object_type(object_id))
3015            }
3016            SystemObjectId::System => SystemObjectType::System,
3017        }
3018    }
3019
3020    /// Returns a read-only view of the current [`StorageMetadata`].
3021    ///
3022    /// To write to this struct, you must use a catalog transaction.
3023    pub fn storage_metadata(&self) -> &StorageMetadata {
3024        &self.storage_metadata
3025    }
3026
3027    /// For the Sources ids in `ids`, return their compaction windows.
3028    pub fn source_compaction_windows(
3029        &self,
3030        ids: impl IntoIterator<Item = CatalogItemId>,
3031    ) -> BTreeMap<CompactionWindow, BTreeSet<CatalogItemId>> {
3032        let mut cws: BTreeMap<CompactionWindow, BTreeSet<CatalogItemId>> = BTreeMap::new();
3033        let mut seen = BTreeSet::new();
3034        for item_id in ids {
3035            if !seen.insert(item_id) {
3036                continue;
3037            }
3038            let entry = self.get_entry(&item_id);
3039            match entry.item() {
3040                CatalogItem::Source(source) => {
3041                    let source_cw = source.custom_logical_compaction_window.unwrap_or_default();
3042                    cws.entry(source_cw).or_default().insert(item_id);
3043                }
3044                CatalogItem::Table(table) => {
3045                    let table_cw = table.custom_logical_compaction_window.unwrap_or_default();
3046                    match &table.data_source {
3047                        TableDataSource::DataSource {
3048                            desc:
3049                                DataSourceDesc::IngestionExport { .. }
3050                                // Also match webhook tables (source-to-table migration).
3051                                | DataSourceDesc::Webhook { .. },
3052                            timeline: _,
3053                        } => {
3054                            cws.entry(table_cw).or_default().insert(item_id);
3055                        }
3056                        // Regular tables handle compaction directly in
3057                        // catalog_implications, not through this function.
3058                        TableDataSource::TableWrites { .. } => {}
3059                        TableDataSource::DataSource {
3060                            desc:
3061                                DataSourceDesc::Ingestion { .. }
3062                                | DataSourceDesc::OldSyntaxIngestion { .. }
3063                                | DataSourceDesc::Introspection(_)
3064                                | DataSourceDesc::Progress
3065                                | DataSourceDesc::Catalog,
3066                            ..
3067                        } => {
3068                            unreachable!(
3069                                "unexpected DataSourceDesc for table {item_id}: {:?}",
3070                                table.data_source
3071                            )
3072                        }
3073                    }
3074                }
3075                _ => {
3076                    // Views could depend on sources, so ignore them if added by used_by above.
3077                    continue;
3078                }
3079            }
3080        }
3081        cws
3082    }
3083
3084    pub fn comment_id_to_item_id(id: &CommentObjectId) -> Option<CatalogItemId> {
3085        match id {
3086            CommentObjectId::Table(id)
3087            | CommentObjectId::View(id)
3088            | CommentObjectId::MaterializedView(id)
3089            | CommentObjectId::Source(id)
3090            | CommentObjectId::Sink(id)
3091            | CommentObjectId::MetricSink(id)
3092            | CommentObjectId::Index(id)
3093            | CommentObjectId::Func(id)
3094            | CommentObjectId::Connection(id)
3095            | CommentObjectId::Type(id)
3096            | CommentObjectId::Secret(id) => Some(*id),
3097            CommentObjectId::Role(_)
3098            | CommentObjectId::Database(_)
3099            | CommentObjectId::Schema(_)
3100            | CommentObjectId::Cluster(_)
3101            | CommentObjectId::ClusterReplica(_)
3102            | CommentObjectId::NetworkPolicy(_) => None,
3103        }
3104    }
3105
3106    pub fn get_comment_id_entry(&self, id: &CommentObjectId) -> Option<&CatalogEntry> {
3107        Self::comment_id_to_item_id(id).map(|id| self.get_entry(&id))
3108    }
3109
3110    pub fn comment_id_to_audit_log_name(
3111        &self,
3112        id: CommentObjectId,
3113        conn_id: &ConnectionId,
3114    ) -> String {
3115        match id {
3116            CommentObjectId::Table(id)
3117            | CommentObjectId::View(id)
3118            | CommentObjectId::MaterializedView(id)
3119            | CommentObjectId::Source(id)
3120            | CommentObjectId::Sink(id)
3121            | CommentObjectId::MetricSink(id)
3122            | CommentObjectId::Index(id)
3123            | CommentObjectId::Func(id)
3124            | CommentObjectId::Connection(id)
3125            | CommentObjectId::Type(id)
3126            | CommentObjectId::Secret(id) => {
3127                let item = self.get_entry(&id);
3128                let name = self.resolve_full_name(item.name(), Some(conn_id));
3129                name.to_string()
3130            }
3131            CommentObjectId::Role(id) => self.get_role(&id).name.clone(),
3132            CommentObjectId::Database(id) => self.get_database(&id).name.clone(),
3133            CommentObjectId::Schema((spec, schema_id)) => {
3134                let schema = self.get_schema(&spec, &schema_id, conn_id);
3135                self.resolve_full_schema_name(&schema.name).to_string()
3136            }
3137            CommentObjectId::Cluster(id) => self.get_cluster(id).name.clone(),
3138            CommentObjectId::ClusterReplica((cluster_id, replica_id)) => {
3139                let cluster = self.get_cluster(cluster_id);
3140                let replica = self.get_cluster_replica(cluster_id, replica_id);
3141                QualifiedReplica {
3142                    cluster: Ident::new_unchecked(cluster.name.clone()),
3143                    replica: Ident::new_unchecked(replica.name.clone()),
3144                }
3145                .to_string()
3146            }
3147            CommentObjectId::NetworkPolicy(id) => self.get_network_policy(&id).name.clone(),
3148        }
3149    }
3150
3151    pub fn mock_authentication_nonce(&self) -> String {
3152        self.mock_authentication_nonce.clone().unwrap_or_default()
3153    }
3154}
3155
3156impl ConnectionResolver for CatalogState {
3157    fn resolve_connection(
3158        &self,
3159        id: CatalogItemId,
3160    ) -> mz_storage_types::connections::Connection<InlinedConnection> {
3161        use mz_storage_types::connections::Connection::*;
3162        match self
3163            .get_entry(&id)
3164            .connection()
3165            .expect("catalog out of sync")
3166            .details
3167            .to_connection()
3168        {
3169            Kafka(conn) => Kafka(conn.into_inline_connection(self)),
3170            Postgres(conn) => Postgres(conn.into_inline_connection(self)),
3171            Csr(conn) => Csr(conn.into_inline_connection(self)),
3172            GlueSchemaRegistry(conn) => GlueSchemaRegistry(conn.into_inline_connection(self)),
3173            Ssh(conn) => Ssh(conn),
3174            Aws(conn) => Aws(conn),
3175            AwsPrivatelink(conn) => AwsPrivatelink(conn),
3176            Gcp(conn) => Gcp(conn),
3177            MySql(conn) => MySql(conn.into_inline_connection(self)),
3178            SqlServer(conn) => SqlServer(conn.into_inline_connection(self)),
3179            IcebergCatalog(conn) => IcebergCatalog(conn.into_inline_connection(self)),
3180        }
3181    }
3182}
3183
3184impl OptimizerCatalog for CatalogState {
3185    fn get_entry(&self, id: &GlobalId) -> CatalogCollectionEntry {
3186        CatalogState::get_entry_by_global_id(self, id)
3187    }
3188    fn get_entry_by_item_id(&self, id: &CatalogItemId) -> &CatalogEntry {
3189        CatalogState::get_entry(self, id)
3190    }
3191    fn resolve_full_name(
3192        &self,
3193        name: &QualifiedItemName,
3194        conn_id: Option<&ConnectionId>,
3195    ) -> FullItemName {
3196        CatalogState::resolve_full_name(self, name, conn_id)
3197    }
3198    fn get_indexes_on(
3199        &self,
3200        id: GlobalId,
3201        cluster: ClusterId,
3202    ) -> Box<dyn Iterator<Item = (GlobalId, &Index)> + '_> {
3203        Box::new(CatalogState::get_indexes_on(self, id, cluster))
3204    }
3205}
3206
3207impl OptimizerCatalog for Catalog {
3208    fn get_entry(&self, id: &GlobalId) -> CatalogCollectionEntry {
3209        self.state.get_entry_by_global_id(id)
3210    }
3211
3212    fn get_entry_by_item_id(&self, id: &CatalogItemId) -> &CatalogEntry {
3213        self.state.get_entry(id)
3214    }
3215
3216    fn resolve_full_name(
3217        &self,
3218        name: &QualifiedItemName,
3219        conn_id: Option<&ConnectionId>,
3220    ) -> FullItemName {
3221        self.state.resolve_full_name(name, conn_id)
3222    }
3223
3224    fn get_indexes_on(
3225        &self,
3226        id: GlobalId,
3227        cluster: ClusterId,
3228    ) -> Box<dyn Iterator<Item = (GlobalId, &Index)> + '_> {
3229        Box::new(self.state.get_indexes_on(id, cluster))
3230    }
3231}
3232
3233impl Catalog {
3234    pub fn as_optimizer_catalog(self: Arc<Self>) -> Arc<dyn OptimizerCatalog> {
3235        self
3236    }
3237}
3238
3239#[cfg(test)]
3240mod tests {
3241    use super::*;
3242
3243    /// A deep dependency chain (a long chain of stacked views) must not
3244    /// overflow the stack when computing its dependents, and the dependents
3245    /// must come out in reverse-dependency order (deepest dependent first,
3246    /// root last) so that `DROP ... CASCADE` drops them in a valid order.
3247    #[mz_ore::test(tokio::test)]
3248    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `TLS_client_method`
3249    async fn item_dependents_deep_chain_no_stack_overflow() {
3250        use mz_ore::cast::CastFrom;
3251
3252        Catalog::with_debug(|mut catalog| async move {
3253            // Deep enough that the previous recursive implementation overflowed
3254            // the stack.
3255            const DEPTH: usize = 100_000;
3256            // Well above any id the debug catalog assigns, so the synthetic
3257            // chain does not collide with real entries.
3258            const BASE: u64 = 1 << 40;
3259
3260            // Clone a builtin log entry as a template so we don't have to
3261            // construct a `CatalogItem` by hand. `item_dependents` only reads
3262            // `used_by`/`progress_id`, and a log's `progress_id` is `None`.
3263            let template = catalog
3264                .state
3265                .entry_by_id
3266                .values()
3267                .find(|entry| matches!(entry.item(), CatalogItem::Log(_)))
3268                .expect("debug catalog has log sources")
3269                .clone();
3270
3271            // Build a chain where entry `BASE + i` is used by `BASE + i + 1`.
3272            for i in 0..=DEPTH {
3273                let id = CatalogItemId::User(BASE + u64::cast_from(i));
3274                let mut entry = template.clone();
3275                entry.id = id;
3276                entry.referenced_by = Vec::new();
3277                entry.used_by = if i < DEPTH {
3278                    vec![CatalogItemId::User(BASE + u64::cast_from(i + 1))]
3279                } else {
3280                    Vec::new()
3281                };
3282                catalog.state.entry_by_id.insert(id, entry);
3283            }
3284
3285            let mut seen = BTreeSet::new();
3286            let dependents = catalog
3287                .state
3288                .item_dependents(CatalogItemId::User(BASE), &mut seen);
3289
3290            // Every element of the chain appears exactly once.
3291            assert_eq!(dependents.len(), DEPTH + 1);
3292            // Reverse-dependency order: the deepest dependent is first and the
3293            // root we asked about is last.
3294            for (offset, dependent) in dependents.iter().enumerate() {
3295                let expected = CatalogItemId::User(BASE + u64::cast_from(DEPTH - offset));
3296                assert_eq!(dependent, &ObjectId::Item(expected));
3297            }
3298
3299            catalog.expire().await;
3300        })
3301        .await
3302    }
3303
3304    /// Read-then-write dependency validation walks the transitive `uses()` of
3305    /// the read set. A deep chain of stacked views (user controlled, arbitrarily
3306    /// deep) must be validated without overflowing the coordinator thread's
3307    /// stack, so the traversal must not recurse.
3308    #[mz_ore::test(tokio::test)]
3309    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `TLS_client_method`
3310    async fn validate_read_then_write_deep_chain_no_stack_overflow() {
3311        use crate::coord::read_then_write::{
3312            DependencyPolicy, validate_read_then_write_dependencies,
3313        };
3314
3315        Catalog::with_debug(|mut catalog| async move {
3316            // Deep enough that the previous recursive implementation overflowed
3317            // the stack.
3318            const DEPTH: usize = 100_000;
3319            const BASE: u64 = 1 << 40;
3320            insert_synthetic_view_chain(&mut catalog, BASE, DEPTH);
3321
3322            // A generous bound so this test isolates the no-overflow property
3323            // rather than the dependency limit.
3324            validate_read_then_write_dependencies(
3325                &catalog,
3326                [CatalogItemId::User(BASE)],
3327                usize::MAX,
3328                DependencyPolicy::UserDml,
3329            )
3330            .expect("deep chain of user views is valid for read-then-write");
3331
3332            catalog.expire().await;
3333        })
3334        .await
3335    }
3336
3337    /// Read-then-write dependency validation is bounded: a read set with more
3338    /// transitive dependencies than the limit is rejected with a clean error
3339    /// rather than walking an unbounded graph.
3340    #[mz_ore::test(tokio::test)]
3341    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `TLS_client_method`
3342    async fn validate_read_then_write_dependency_limit() {
3343        use crate::coord::read_then_write::{
3344            DependencyPolicy, validate_read_then_write_dependencies,
3345        };
3346        use crate::error::AdapterError;
3347
3348        Catalog::with_debug(|mut catalog| async move {
3349            const DEPTH: usize = 100;
3350            const BASE: u64 = 1 << 40;
3351            insert_synthetic_view_chain(&mut catalog, BASE, DEPTH);
3352
3353            // The chain has DEPTH + 1 distinct objects (root plus DEPTH links).
3354            const OBJECTS: usize = DEPTH + 1;
3355
3356            // Exactly at the limit is allowed.
3357            validate_read_then_write_dependencies(
3358                &catalog,
3359                [CatalogItemId::User(BASE)],
3360                OBJECTS,
3361                DependencyPolicy::UserDml,
3362            )
3363            .expect("chain at the limit is valid");
3364
3365            // One below the limit is rejected with a clean error.
3366            let err = validate_read_then_write_dependencies(
3367                &catalog,
3368                [CatalogItemId::User(BASE)],
3369                OBJECTS - 1,
3370                DependencyPolicy::UserDml,
3371            )
3372            .expect_err("chain over the limit is rejected");
3373            assert!(matches!(
3374                err,
3375                AdapterError::ReadThenWriteDependencyLimitExceeded {
3376                    max_rw_dependencies
3377                } if max_rw_dependencies == OBJECTS - 1
3378            ));
3379
3380            catalog.expire().await;
3381        })
3382        .await
3383    }
3384
3385    /// Inserts a synthetic chain of `depth + 1` user views into `catalog` where
3386    /// view `base + i` reads from `base + i + 1`. Ids start well above any id
3387    /// the debug catalog assigns so they do not collide with real entries.
3388    ///
3389    /// Clones a builtin view as a template rather than constructing a `View` by
3390    /// hand. Read-then-write validation only reads the item type, `uses()`, and
3391    /// the optimized expression's temporal-ness, all of which a builtin view
3392    /// satisfies (user id + non-temporal).
3393    fn insert_synthetic_view_chain(catalog: &mut Catalog, base: u64, depth: usize) {
3394        use mz_ore::cast::CastFrom;
3395        use mz_sql::names::{DependencyIds, ResolvedIds};
3396
3397        let template = catalog
3398            .state
3399            .entry_by_id
3400            .values()
3401            .find(|entry| matches!(entry.item(), CatalogItem::View(_)))
3402            .expect("debug catalog has builtin views")
3403            .clone();
3404
3405        // `uses()` for a view unions `resolved_ids` and `dependencies`, so clear
3406        // both and point only at the next link.
3407        for i in 0..=depth {
3408            let id = CatalogItemId::User(base + u64::cast_from(i));
3409            let mut entry = template.clone();
3410            entry.id = id;
3411            entry.referenced_by = Vec::new();
3412            entry.used_by = Vec::new();
3413            let mut resolved_ids = ResolvedIds::empty();
3414            if i < depth {
3415                resolved_ids.add_item(CatalogItemId::User(base + u64::cast_from(i + 1)));
3416            }
3417            match &mut entry.item {
3418                CatalogItem::View(view) => {
3419                    view.resolved_ids = resolved_ids;
3420                    view.dependencies = DependencyIds(BTreeSet::new());
3421                }
3422                _ => unreachable!("template is a view"),
3423            }
3424            catalog.state.entry_by_id.insert(id, entry);
3425        }
3426    }
3427}