1use 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
98use 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#[derive(Debug, Clone, Serialize)]
113pub struct CatalogState {
114 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 #[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 #[serde(skip)]
165 pub(super) notices_by_dep_id: imbl::OrdMap<GlobalId, Vec<Arc<OptimizerNotice>>>,
166
167 #[serde(skip)]
171 pub(super) temporary_namespaces: TemporaryNamespaces,
172
173 #[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 #[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 #[serde(skip)]
189 pub(super) license_key: ValidatedLicenseKey,
190}
191
192#[derive(Debug, Clone, Default)]
201pub(super) struct TemporaryNamespaces {
202 by_conn: imbl::OrdMap<ConnectionId, TemporaryNamespace>,
203 conns_by_uuid: imbl::OrdMap<Uuid, ConnectionId>,
206}
207
208#[derive(Debug, Clone)]
209struct TemporaryNamespace {
210 uuid: Uuid,
211 schema: Option<Schema>,
213}
214
215impl TemporaryNamespaces {
216 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 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 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 pub(super) fn unregister(&mut self, conn_id: &ConnectionId) {
273 let Some(namespace) = self.by_conn.get(conn_id) else {
274 return;
275 };
276 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#[derive(Debug, Clone, Serialize)]
334pub(crate) enum LocalExpressionCache {
335 Open {
337 cached_exprs: BTreeMap<GlobalId, LocalExpressions>,
339 uncached_exprs: BTreeMap<GlobalId, LocalExpressions>,
341 },
342 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 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 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 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 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 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 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 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 pub fn introspection_dependencies(&self, id: CatalogItemId) -> Vec<CatalogItemId> {
652 let mut out = Vec::new();
653
654 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 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 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 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 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 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 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 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 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 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 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 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 vec![object_id]
933 }
934
935 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 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 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 self.temporary_namespaces
1053 .schema(conn)
1054 .into_iter()
1055 .flat_map(|schema| schema.items.values().copied().map(ObjectId::from))
1056 }
1057
1058 pub fn has_temporary_namespace(&self, conn: &ConnectionId) -> bool {
1065 self.temporary_namespaces.contains_conn(conn)
1066 }
1067
1068 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 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 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 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 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 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 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 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 pub fn try_get_webhook_url(&self, id: &CatalogItemId) -> Option<url::Url> {
1373 let entry = self.try_get_entry(id)?;
1374 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 pub(crate) fn deserialize_plan_with_enable_for_item_parsing(
1404 &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 #[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 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 #[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 #[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 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 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 (_, 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 let mut optimizer = optimize::view::Optimizer::new(optimizer_config, None);
1728
1729 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 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 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 (_, 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 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 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 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 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 pub fn with_enable_for_item_parsing<T>(&mut self, f: impl FnOnce(&mut Self) -> T) -> T {
1967 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 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 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 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 pub fn get_system_configuration(&self, name: &str) -> Result<&dyn Var, Error> {
2056 Ok(self.system_configuration.get(name)?)
2057 }
2058
2059 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 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 pub fn try_get_schema(
2101 &self,
2102 database_spec: &ResolvedDatabaseSpecifier,
2103 schema_spec: &SchemaSpecifier,
2104 conn_id: &ConnectionId,
2105 ) -> Option<&Schema> {
2106 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 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 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 pub fn resolve_builtin_table(&self, builtin: &'static BuiltinTable) -> CatalogItemId {
2229 self.resolve_builtin_object(&Builtin::<IdReference>::Table(builtin))
2230 }
2231
2232 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 pub fn resolve_builtin_source(&self, builtin: &'static BuiltinSource) -> CatalogItemId {
2248 self.resolve_builtin_object(&Builtin::<IdReference>::Source(builtin))
2249 }
2250
2251 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 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 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 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 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 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 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 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 #[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 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 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 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 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 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 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 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 pub fn system_config(&self) -> &SystemVars {
2687 &self.system_configuration
2688 }
2689
2690 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 pub fn scoped_system_parameters(&self) -> &ScopedParameters {
2708 &self.scoped_system_parameters
2709 }
2710
2711 pub fn system_config_mut(&mut self) -> &mut SystemVars {
2713 Arc::make_mut(&mut self.system_configuration)
2714 }
2715
2716 pub fn dump(&self, unfinalized_shards: Option<BTreeSet<String>>) -> Result<String, Error> {
2726 let mut dump = serde_json::to_value(&self).map_err(|e| {
2728 Error::new(ErrorKind::Unstructured(format!(
2729 "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 dump_obj.insert(
2739 "system_parameter_defaults".into(),
2740 serde_json::json!(self.system_config().defaults()),
2741 );
2742 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 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 dump_obj.remove("role_auth_by_id");
2777 dump_obj.remove("mock_authentication_nonce");
2782
2783 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 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 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 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 pub fn storage_metadata(&self) -> &StorageMetadata {
3024 &self.storage_metadata
3025 }
3026
3027 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 | DataSourceDesc::Webhook { .. },
3052 timeline: _,
3053 } => {
3054 cws.entry(table_cw).or_default().insert(item_id);
3055 }
3056 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 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 #[mz_ore::test(tokio::test)]
3248 #[cfg_attr(miri, ignore)] 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 const DEPTH: usize = 100_000;
3256 const BASE: u64 = 1 << 40;
3259
3260 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 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 assert_eq!(dependents.len(), DEPTH + 1);
3292 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 #[mz_ore::test(tokio::test)]
3309 #[cfg_attr(miri, ignore)] 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 const DEPTH: usize = 100_000;
3319 const BASE: u64 = 1 << 40;
3320 insert_synthetic_view_chain(&mut catalog, BASE, DEPTH);
3321
3322 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 #[mz_ore::test(tokio::test)]
3341 #[cfg_attr(miri, ignore)] 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 const OBJECTS: usize = DEPTH + 1;
3355
3356 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 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 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 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}