1use mz_controller_types::{ClusterId, ReplicaId};
16use mz_repr::role_id::RoleId;
17use mz_repr::{CatalogItemId, GlobalId};
18use mz_sql::catalog::{CatalogItem, DefaultPrivilegeObject};
19use mz_sql::names::{
20 CommentObjectId, DatabaseId, QualifiedItemName, ResolvedDatabaseSpecifier, SchemaId,
21 SchemaSpecifier,
22};
23use mz_sql_parser::ast::{self, Statement};
24use mz_sql_parser::parser::ParserStatementError;
25use serde::Serialize;
26
27use super::CatalogState;
29
30#[derive(Debug, Default, Clone, Serialize, PartialEq)]
31pub struct CatalogInconsistencies {
32 internal_fields: Vec<InternalFieldsInconsistency>,
34 roles: Vec<RoleInconsistency>,
36 comments: Vec<CommentInconsistency>,
38 object_dependencies: Vec<ObjectDependencyInconsistency>,
40 items: Vec<ItemInconsistency>,
42}
43
44impl CatalogInconsistencies {
45 pub fn is_empty(&self) -> bool {
46 let CatalogInconsistencies {
47 internal_fields,
48 roles,
49 comments,
50 object_dependencies,
51 items,
52 } = self;
53 internal_fields.is_empty()
54 && roles.is_empty()
55 && comments.is_empty()
56 && object_dependencies.is_empty()
57 && items.is_empty()
58 }
59}
60
61impl CatalogState {
62 pub fn check_consistency(&self) -> Result<(), Box<CatalogInconsistencies>> {
64 let mut inconsistencies = CatalogInconsistencies::default();
65
66 if let Err(internal_fields) = self.check_internal_fields() {
67 inconsistencies.internal_fields = internal_fields;
68 }
69 if let Err(roles) = self.check_roles() {
70 inconsistencies.roles = roles;
71 }
72 if let Err(comments) = self.check_comments() {
73 inconsistencies.comments = comments;
74 }
75 if let Err(dependencies) = self.check_object_dependencies() {
76 inconsistencies.object_dependencies = dependencies;
77 }
78 if let Err(items) = self.check_items() {
79 inconsistencies.items = items;
80 }
81
82 if inconsistencies.is_empty() {
83 Ok(())
84 } else {
85 Err(Box::new(inconsistencies))
86 }
87 }
88
89 fn check_internal_fields(&self) -> Result<(), Vec<InternalFieldsInconsistency>> {
96 let mut inconsistencies = Vec::new();
97 for (name, id) in &self.database_by_name {
98 if !self.database_by_id.contains_key(id) {
99 inconsistencies.push(InternalFieldsInconsistency::Database(name.clone(), *id));
100 }
101 }
102 for (name, id) in &self.ambient_schemas_by_name {
103 if !self.ambient_schemas_by_id.contains_key(id) {
104 inconsistencies.push(InternalFieldsInconsistency::AmbientSchema(
105 name.clone(),
106 *id,
107 ));
108 }
109 }
110 for (name, id) in &self.clusters_by_name {
111 if !self.clusters_by_id.contains_key(id) {
112 inconsistencies.push(InternalFieldsInconsistency::Cluster(name.clone(), *id));
113 }
114 }
115 for (name, role_id) in &self.roles_by_name {
116 if !self.roles_by_id.contains_key(role_id) {
117 inconsistencies.push(InternalFieldsInconsistency::Role(name.clone(), *role_id))
118 }
119 }
120
121 for (source_id, _references) in &self.source_references {
122 if !self.entry_by_id.contains_key(source_id) {
123 inconsistencies.push(InternalFieldsInconsistency::SourceReferences(*source_id));
124 }
125 }
126
127 for (item_id, entry) in &self.entry_by_id {
128 let missing_gids: Vec<_> = entry
129 .global_ids()
130 .filter(|gid| !self.entry_by_global_id.contains_key(gid))
131 .collect();
132 if !missing_gids.is_empty() {
133 inconsistencies.push(InternalFieldsInconsistency::EntryMissingGlobalIds(
134 *item_id,
135 missing_gids,
136 ));
137 }
138 }
139 for (gid, item_id) in &self.entry_by_global_id {
140 if !self.entry_by_id.contains_key(item_id) {
141 inconsistencies.push(InternalFieldsInconsistency::GlobalIdsMissingEntry(
142 *gid, *item_id,
143 ));
144 }
145 }
146
147 if inconsistencies.is_empty() {
148 Ok(())
149 } else {
150 Err(inconsistencies)
151 }
152 }
153
154 fn check_roles(&self) -> Result<(), Vec<RoleInconsistency>> {
159 let mut inconsistencies = Vec::new();
160 for (database_id, database) in &self.database_by_id {
161 if !self.roles_by_id.contains_key(&database.owner_id) {
162 inconsistencies.push(RoleInconsistency::Database(*database_id, database.owner_id));
163 }
164 for (schema_id, schema) in &database.schemas_by_id {
165 if !self.roles_by_id.contains_key(&schema.owner_id) {
166 inconsistencies.push(RoleInconsistency::Schema(*schema_id, schema.owner_id));
167 }
168 }
169 }
170 for (item_id, entry) in &self.entry_by_id {
171 if !self.roles_by_id.contains_key(entry.owner_id()) {
172 inconsistencies.push(RoleInconsistency::Entry(*item_id, entry.owner_id().clone()));
173 }
174 }
175 for (cluster_id, cluster) in &self.clusters_by_id {
176 if !self.roles_by_id.contains_key(&cluster.owner_id) {
177 inconsistencies.push(RoleInconsistency::Cluster(*cluster_id, cluster.owner_id));
178 }
179 for replica in cluster.replicas() {
180 if !self.roles_by_id.contains_key(&replica.owner_id) {
181 inconsistencies.push(RoleInconsistency::ClusterReplica(
182 *cluster_id,
183 replica.replica_id,
184 cluster.owner_id,
185 ));
186 }
187 }
188 }
189 for (role_id, _) in &self.role_auth_by_id {
190 if !self.roles_by_id.contains_key(role_id) {
191 inconsistencies.push(RoleInconsistency::RoleAuth(role_id.clone()));
192 }
193 }
194 for (default_priv, privileges) in self.default_privileges.iter() {
195 if !self.roles_by_id.contains_key(&default_priv.role_id) {
196 inconsistencies.push(RoleInconsistency::DefaultPrivilege(default_priv.clone()));
197 }
198 for acl_item in privileges {
199 if !self.roles_by_id.contains_key(&acl_item.grantee) {
200 inconsistencies.push(RoleInconsistency::DefaultPrivilegeItem {
201 grantor: default_priv.role_id,
202 grantee: acl_item.grantee,
203 });
204 }
205 }
206 }
207 for acl in self.system_privileges.all_values() {
208 let grantor = self.roles_by_id.get(&acl.grantor);
209 let grantee = self.roles_by_id.get(&acl.grantee);
210
211 let inconsistency = match (grantor, grantee) {
212 (None, None) => RoleInconsistency::SystemPrivilege {
213 grantor: Some(acl.grantor),
214 grantee: Some(acl.grantee),
215 },
216 (Some(_), None) => RoleInconsistency::SystemPrivilege {
217 grantor: None,
218 grantee: Some(acl.grantee),
219 },
220 (None, Some(_)) => RoleInconsistency::SystemPrivilege {
221 grantor: Some(acl.grantor),
222 grantee: None,
223 },
224 (Some(_), Some(_)) => continue,
225 };
226 inconsistencies.push(inconsistency);
227 }
228 for role in self.roles_by_id.values() {
229 for (parent_id, grantor_id) in &role.membership.map {
230 let parent = self.roles_by_id.get(parent_id);
231 let grantor = self.roles_by_id.get(grantor_id);
232 let inconsistency = match (parent, grantor) {
233 (None, None) => RoleInconsistency::Membership {
234 parent: Some(*parent_id),
235 grantor: Some(*grantor_id),
236 },
237 (Some(_), None) => RoleInconsistency::Membership {
238 parent: None,
239 grantor: Some(*grantor_id),
240 },
241 (None, Some(_)) => RoleInconsistency::Membership {
242 parent: Some(*parent_id),
243 grantor: None,
244 },
245 (Some(_), Some(_)) => continue,
246 };
247 inconsistencies.push(inconsistency);
248 }
249 }
250
251 if inconsistencies.is_empty() {
252 Ok(())
253 } else {
254 Err(inconsistencies)
255 }
256 }
257
258 fn check_comments(&self) -> Result<(), Vec<CommentInconsistency>> {
264 let mut comment_inconsistencies = Vec::new();
265 for (comment_object_id, col_pos, _comment) in self.comments.iter() {
266 match comment_object_id {
267 CommentObjectId::Table(item_id)
268 | CommentObjectId::View(item_id)
269 | CommentObjectId::MaterializedView(item_id)
270 | CommentObjectId::Source(item_id)
271 | CommentObjectId::Sink(item_id)
272 | CommentObjectId::Index(item_id)
273 | CommentObjectId::Func(item_id)
274 | CommentObjectId::Connection(item_id)
275 | CommentObjectId::Type(item_id)
276 | CommentObjectId::Secret(item_id) => {
277 let entry = self.entry_by_id.get(&item_id);
278 match entry {
279 None => comment_inconsistencies
280 .push(CommentInconsistency::Dangling(comment_object_id)),
281 Some(entry) => {
282 #[allow(clippy::unnecessary_unwrap)]
284 if !entry.has_columns() && col_pos.is_some() {
285 let col_pos = col_pos.expect("checked above");
286 comment_inconsistencies.push(CommentInconsistency::NonRelation(
287 comment_object_id,
288 col_pos,
289 ));
290 }
291 }
292 }
293 }
294 CommentObjectId::NetworkPolicy(network_policy_id) => {
295 if !self.network_policies_by_id.contains_key(&network_policy_id) {
296 comment_inconsistencies
297 .push(CommentInconsistency::Dangling(comment_object_id));
298 }
299 }
300
301 CommentObjectId::Role(role_id) => {
302 if !self.roles_by_id.contains_key(&role_id) {
303 comment_inconsistencies
304 .push(CommentInconsistency::Dangling(comment_object_id));
305 }
306 }
307 CommentObjectId::Database(database_id) => {
308 if !self.database_by_id.contains_key(&database_id) {
309 comment_inconsistencies
310 .push(CommentInconsistency::Dangling(comment_object_id));
311 }
312 }
313 CommentObjectId::Schema((database, schema)) => {
314 match (database, schema) {
315 (
316 ResolvedDatabaseSpecifier::Id(database_id),
317 SchemaSpecifier::Id(schema_id),
318 ) => {
319 let schema = self
320 .database_by_id
321 .get(&database_id)
322 .and_then(|database| database.schemas_by_id.get(&schema_id));
323 if schema.is_none() {
324 comment_inconsistencies
325 .push(CommentInconsistency::Dangling(comment_object_id));
326 }
327 }
328 (ResolvedDatabaseSpecifier::Ambient, SchemaSpecifier::Id(schema_id)) => {
329 if !self.ambient_schemas_by_id.contains_key(&schema_id) {
330 comment_inconsistencies
331 .push(CommentInconsistency::Dangling(comment_object_id));
332 }
333 }
334 (ResolvedDatabaseSpecifier::Id(_id), SchemaSpecifier::Temporary) => (),
336 (ResolvedDatabaseSpecifier::Ambient, SchemaSpecifier::Temporary) => (),
338 }
339 }
340 CommentObjectId::Cluster(cluster_id) => {
341 if !self.clusters_by_id.contains_key(&cluster_id) {
342 comment_inconsistencies
343 .push(CommentInconsistency::Dangling(comment_object_id));
344 }
345 }
346 CommentObjectId::ClusterReplica((cluster_id, replica_id)) => {
347 let replica = self
348 .clusters_by_id
349 .get(&cluster_id)
350 .and_then(|cluster| cluster.replica(replica_id));
351 if replica.is_none() {
352 comment_inconsistencies
353 .push(CommentInconsistency::Dangling(comment_object_id));
354 }
355 }
356 }
357 }
358
359 if comment_inconsistencies.is_empty() {
360 Ok(())
361 } else {
362 Err(comment_inconsistencies)
363 }
364 }
365
366 fn check_object_dependencies(&self) -> Result<(), Vec<ObjectDependencyInconsistency>> {
374 let mut dependency_inconsistencies = vec![];
375
376 for (id, entry) in &self.entry_by_id {
377 for referenced_id in entry.references().items() {
378 let Some(referenced_entry) = self.entry_by_id.get(referenced_id) else {
379 dependency_inconsistencies.push(ObjectDependencyInconsistency::MissingUses {
380 object_a: *id,
381 object_b: *referenced_id,
382 });
383 continue;
384 };
385 if !referenced_entry.referenced_by().contains(id) && referenced_entry.id() != *id {
386 dependency_inconsistencies.push(
387 ObjectDependencyInconsistency::InconsistentUsedBy {
388 object_a: *id,
389 object_b: *referenced_id,
390 },
391 );
392 }
393 }
394 for used_id in entry.uses() {
395 let Some(used_entry) = self.entry_by_id.get(&used_id) else {
396 dependency_inconsistencies.push(ObjectDependencyInconsistency::MissingUses {
397 object_a: *id,
398 object_b: used_id,
399 });
400 continue;
401 };
402 if !used_entry.used_by().contains(id) && used_entry.id() != *id {
403 dependency_inconsistencies.push(
404 ObjectDependencyInconsistency::InconsistentUsedBy {
405 object_a: *id,
406 object_b: used_id,
407 },
408 );
409 }
410 }
411
412 for referenced_by in entry.referenced_by() {
413 let Some(referenced_by_entry) = self.entry_by_id.get(referenced_by) else {
414 dependency_inconsistencies.push(ObjectDependencyInconsistency::MissingUsedBy {
415 object_a: *id,
416 object_b: *referenced_by,
417 });
418 continue;
419 };
420 if !referenced_by_entry.references().contains_item(id) {
421 dependency_inconsistencies.push(
422 ObjectDependencyInconsistency::InconsistentUses {
423 object_a: *id,
424 object_b: *referenced_by,
425 },
426 );
427 }
428 }
429 for used_by in entry.used_by() {
430 let Some(used_by_entry) = self.entry_by_id.get(used_by) else {
431 dependency_inconsistencies.push(ObjectDependencyInconsistency::MissingUsedBy {
432 object_a: *id,
433 object_b: *used_by,
434 });
435 continue;
436 };
437 if !used_by_entry.uses().contains(id) {
438 dependency_inconsistencies.push(
439 ObjectDependencyInconsistency::InconsistentUses {
440 object_a: *id,
441 object_b: *used_by,
442 },
443 );
444 }
445 }
446 }
447
448 if dependency_inconsistencies.is_empty() {
449 Ok(())
450 } else {
451 Err(dependency_inconsistencies)
452 }
453 }
454
455 fn check_items(&self) -> Result<(), Vec<ItemInconsistency>> {
467 let mut item_inconsistencies = vec![];
468
469 for (db_id, db) in &self.database_by_id {
470 for (schema_name, schema_id) in &db.schemas_by_name {
471 let Some(schema) = db.schemas_by_id.get(schema_id) else {
473 item_inconsistencies.push(ItemInconsistency::MissingSchema {
474 db_id: *db_id,
475 schema_name: schema_name.clone(),
476 });
477 continue;
478 };
479 if schema_name != &schema.name.schema {
480 item_inconsistencies.push(ItemInconsistency::KeyedName {
481 db_schema_by_name: schema_name.clone(),
482 struct_name: schema.name.schema.clone(),
483 });
484 }
485
486 for (item_name, item_id) in schema
491 .items
492 .iter()
493 .chain(schema.types.iter())
494 .chain(schema.functions.iter())
495 {
496 let Some(entry) = self.entry_by_id.get(item_id) else {
497 item_inconsistencies.push(ItemInconsistency::NonExistentItem {
498 db_id: *db_id,
499 schema_id: schema.id,
500 item_id: *item_id,
501 });
502 continue;
503 };
504 if item_name != &entry.name().item {
505 item_inconsistencies.push(ItemInconsistency::ItemNameMismatch {
506 item_id: *item_id,
507 map_name: item_name.clone(),
508 entry_name: entry.name().clone(),
509 });
510 }
511 let statement = match mz_sql::parse::parse(entry.create_sql()) {
512 Ok(mut statements) if statements.len() == 1 => {
513 let statement = statements.pop().expect("checked length");
514 statement.ast
515 }
516 Ok(_) => {
517 item_inconsistencies.push(ItemInconsistency::MultiCreateStatement {
518 create_sql: entry.create_sql().to_string(),
519 });
520 continue;
521 }
522 Err(e) => {
523 item_inconsistencies.push(ItemInconsistency::StatementParseFailure {
524 create_sql: entry.create_sql().to_string(),
525 e,
526 });
527 continue;
528 }
529 };
530 match statement {
531 Statement::CreateConnection(ast::CreateConnectionStatement {
532 name,
533 ..
534 })
535 | Statement::CreateWebhookSource(ast::CreateWebhookSourceStatement {
536 name,
537 ..
538 })
539 | Statement::CreateSource(ast::CreateSourceStatement { name, .. })
540 | Statement::CreateSubsource(ast::CreateSubsourceStatement {
541 name, ..
542 })
543 | Statement::CreateSink(ast::CreateSinkStatement {
544 name: Some(name),
545 ..
546 })
547 | Statement::CreateView(ast::CreateViewStatement {
548 definition: ast::ViewDefinition { name, .. },
549 ..
550 })
551 | Statement::CreateMaterializedView(
552 ast::CreateMaterializedViewStatement { name, .. },
553 )
554 | Statement::CreateTable(ast::CreateTableStatement { name, .. })
555 | Statement::CreateType(ast::CreateTypeStatement { name, .. })
556 | Statement::CreateSecret(ast::CreateSecretStatement { name, .. }) => {
557 let [db_component, schema_component, item_component] = &name.0[..]
558 else {
559 let name =
560 name.0.into_iter().map(|ident| ident.to_string()).collect();
561 item_inconsistencies.push(
562 ItemInconsistency::NonFullyQualifiedItemName {
563 create_sql: entry.create_sql().to_string(),
564 name,
565 },
566 );
567 continue;
568 };
569 if db_component.as_str() != &db.name
570 || schema_component.as_str() != &schema.name.schema
571 || item_component.as_str() != &entry.name().item
572 {
573 item_inconsistencies.push(
574 ItemInconsistency::CreateSqlItemNameMismatch {
575 item_name: vec![
576 db.name.clone(),
577 schema.name.schema.clone(),
578 entry.name().item.clone(),
579 ],
580 create_sql: entry.create_sql().to_string(),
581 },
582 );
583 }
584 }
585 Statement::CreateSchema(ast::CreateSchemaStatement { name, .. }) => {
586 let [db_component, schema_component] = &name.0[..] else {
587 let name =
588 name.0.into_iter().map(|ident| ident.to_string()).collect();
589 item_inconsistencies.push(
590 ItemInconsistency::NonFullyQualifiedSchemaName {
591 create_sql: entry.create_sql().to_string(),
592 name,
593 },
594 );
595 continue;
596 };
597 if db_component.as_str() != &db.name
598 || schema_component.as_str() != &schema.name.schema
599 {
600 item_inconsistencies.push(
601 ItemInconsistency::CreateSqlSchemaNameMismatch {
602 schema_name: vec![
603 db.name.clone(),
604 schema.name.schema.clone(),
605 ],
606 create_sql: entry.create_sql().to_string(),
607 },
608 );
609 }
610 }
611 Statement::CreateDatabase(ast::CreateDatabaseStatement {
612 name, ..
613 }) => {
614 if db.name != name.0.as_str() {
615 item_inconsistencies.push(
616 ItemInconsistency::CreateSqlDatabaseNameMismatch {
617 database_name: db.name.clone(),
618 create_sql: entry.create_sql().to_string(),
619 },
620 );
621 }
622 }
623 _ => (),
624 }
625 }
626 }
627 }
628
629 if item_inconsistencies.is_empty() {
630 Ok(())
631 } else {
632 Err(item_inconsistencies)
633 }
634 }
635}
636
637#[derive(Debug, Serialize, Clone, PartialEq)]
638enum InternalFieldsInconsistency {
639 Database(String, DatabaseId),
640 AmbientSchema(String, SchemaId),
641 Cluster(String, ClusterId),
642 Role(String, RoleId),
643 SourceReferences(CatalogItemId),
644 EntryMissingGlobalIds(CatalogItemId, Vec<GlobalId>),
645 GlobalIdsMissingEntry(GlobalId, CatalogItemId),
646}
647
648#[derive(Debug, Serialize, Clone, PartialEq)]
649enum RoleInconsistency {
650 Database(DatabaseId, RoleId),
651 Schema(SchemaId, RoleId),
652 Entry(CatalogItemId, RoleId),
653 Cluster(ClusterId, RoleId),
654 ClusterReplica(ClusterId, ReplicaId, RoleId),
655 DefaultPrivilege(DefaultPrivilegeObject),
656 RoleAuth(RoleId),
657 DefaultPrivilegeItem {
658 grantor: RoleId,
659 grantee: RoleId,
660 },
661 SystemPrivilege {
662 grantor: Option<RoleId>,
663 grantee: Option<RoleId>,
664 },
665 Membership {
666 parent: Option<RoleId>,
667 grantor: Option<RoleId>,
668 },
669}
670
671#[derive(Debug, Serialize, Clone, PartialEq)]
672enum CommentInconsistency {
673 Dangling(CommentObjectId),
675 NonRelation(CommentObjectId, usize),
677}
678
679#[derive(Debug, Serialize, Clone, PartialEq)]
680enum ObjectDependencyInconsistency {
681 MissingUses {
683 object_a: CatalogItemId,
684 object_b: CatalogItemId,
685 },
686 MissingUsedBy {
688 object_a: CatalogItemId,
689 object_b: CatalogItemId,
690 },
691 InconsistentUsedBy {
693 object_a: CatalogItemId,
694 object_b: CatalogItemId,
695 },
696 InconsistentUses {
698 object_a: CatalogItemId,
699 object_b: CatalogItemId,
700 },
701}
702
703#[derive(Debug, Serialize, Clone, PartialEq)]
704enum ItemInconsistency {
705 KeyedName {
707 db_schema_by_name: String,
708 struct_name: String,
709 },
710 MissingSchema {
712 db_id: DatabaseId,
713 schema_name: String,
714 },
715 NonExistentItem {
717 db_id: DatabaseId,
718 schema_id: SchemaSpecifier,
719 item_id: CatalogItemId,
720 },
721 ItemNameMismatch {
723 item_id: CatalogItemId,
724 map_name: String,
726 entry_name: QualifiedItemName,
728 },
729 StatementParseFailure {
731 create_sql: String,
732 e: ParserStatementError,
733 },
734 MultiCreateStatement { create_sql: String },
736 NonFullyQualifiedItemName {
738 create_sql: String,
739 name: Vec<String>,
740 },
741 NonFullyQualifiedSchemaName {
743 create_sql: String,
744 name: Vec<String>,
745 },
746 CreateSqlItemNameMismatch {
748 item_name: Vec<String>,
749 create_sql: String,
750 },
751 CreateSqlSchemaNameMismatch {
753 schema_name: Vec<String>,
754 create_sql: String,
755 },
756 CreateSqlDatabaseNameMismatch {
758 database_name: String,
759 create_sql: String,
760 },
761}