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::MetricSink(item_id)
273 | CommentObjectId::Index(item_id)
274 | CommentObjectId::Func(item_id)
275 | CommentObjectId::Connection(item_id)
276 | CommentObjectId::Type(item_id)
277 | CommentObjectId::Secret(item_id) => {
278 let entry = self.entry_by_id.get(&item_id);
279 match entry {
280 None => comment_inconsistencies
281 .push(CommentInconsistency::Dangling(comment_object_id)),
282 Some(entry) => {
283 #[allow(clippy::unnecessary_unwrap)]
285 if !entry.has_columns() && col_pos.is_some() {
286 let col_pos = col_pos.expect("checked above");
287 comment_inconsistencies.push(CommentInconsistency::NonRelation(
288 comment_object_id,
289 col_pos,
290 ));
291 }
292 }
293 }
294 }
295 CommentObjectId::NetworkPolicy(network_policy_id) => {
296 if !self.network_policies_by_id.contains_key(&network_policy_id) {
297 comment_inconsistencies
298 .push(CommentInconsistency::Dangling(comment_object_id));
299 }
300 }
301
302 CommentObjectId::Role(role_id) => {
303 if !self.roles_by_id.contains_key(&role_id) {
304 comment_inconsistencies
305 .push(CommentInconsistency::Dangling(comment_object_id));
306 }
307 }
308 CommentObjectId::Database(database_id) => {
309 if !self.database_by_id.contains_key(&database_id) {
310 comment_inconsistencies
311 .push(CommentInconsistency::Dangling(comment_object_id));
312 }
313 }
314 CommentObjectId::Schema((database, schema)) => {
315 match (database, schema) {
316 (
317 ResolvedDatabaseSpecifier::Id(database_id),
318 SchemaSpecifier::Id(schema_id),
319 ) => {
320 let schema = self
321 .database_by_id
322 .get(&database_id)
323 .and_then(|database| database.schemas_by_id.get(&schema_id));
324 if schema.is_none() {
325 comment_inconsistencies
326 .push(CommentInconsistency::Dangling(comment_object_id));
327 }
328 }
329 (ResolvedDatabaseSpecifier::Ambient, SchemaSpecifier::Id(schema_id)) => {
330 if !self.ambient_schemas_by_id.contains_key(&schema_id) {
331 comment_inconsistencies
332 .push(CommentInconsistency::Dangling(comment_object_id));
333 }
334 }
335 (ResolvedDatabaseSpecifier::Id(_id), SchemaSpecifier::Temporary) => (),
337 (ResolvedDatabaseSpecifier::Ambient, SchemaSpecifier::Temporary) => (),
339 }
340 }
341 CommentObjectId::Cluster(cluster_id) => {
342 if !self.clusters_by_id.contains_key(&cluster_id) {
343 comment_inconsistencies
344 .push(CommentInconsistency::Dangling(comment_object_id));
345 }
346 }
347 CommentObjectId::ClusterReplica((cluster_id, replica_id)) => {
348 let replica = self
349 .clusters_by_id
350 .get(&cluster_id)
351 .and_then(|cluster| cluster.replica(replica_id));
352 if replica.is_none() {
353 comment_inconsistencies
354 .push(CommentInconsistency::Dangling(comment_object_id));
355 }
356 }
357 }
358 }
359
360 if comment_inconsistencies.is_empty() {
361 Ok(())
362 } else {
363 Err(comment_inconsistencies)
364 }
365 }
366
367 fn check_object_dependencies(&self) -> Result<(), Vec<ObjectDependencyInconsistency>> {
375 let mut dependency_inconsistencies = vec![];
376
377 for (id, entry) in &self.entry_by_id {
378 for referenced_id in entry.references().items() {
379 let Some(referenced_entry) = self.entry_by_id.get(referenced_id) else {
380 dependency_inconsistencies.push(ObjectDependencyInconsistency::MissingUses {
381 object_a: *id,
382 object_b: *referenced_id,
383 });
384 continue;
385 };
386 if !referenced_entry.referenced_by().contains(id) && referenced_entry.id() != *id {
387 dependency_inconsistencies.push(
388 ObjectDependencyInconsistency::InconsistentUsedBy {
389 object_a: *id,
390 object_b: *referenced_id,
391 },
392 );
393 }
394 }
395 for used_id in entry.uses() {
396 let Some(used_entry) = self.entry_by_id.get(&used_id) else {
397 dependency_inconsistencies.push(ObjectDependencyInconsistency::MissingUses {
398 object_a: *id,
399 object_b: used_id,
400 });
401 continue;
402 };
403 if !used_entry.used_by().contains(id) && used_entry.id() != *id {
404 dependency_inconsistencies.push(
405 ObjectDependencyInconsistency::InconsistentUsedBy {
406 object_a: *id,
407 object_b: used_id,
408 },
409 );
410 }
411 }
412
413 for referenced_by in entry.referenced_by() {
414 let Some(referenced_by_entry) = self.entry_by_id.get(referenced_by) else {
415 dependency_inconsistencies.push(ObjectDependencyInconsistency::MissingUsedBy {
416 object_a: *id,
417 object_b: *referenced_by,
418 });
419 continue;
420 };
421 if !referenced_by_entry.references().contains_item(id) {
422 dependency_inconsistencies.push(
423 ObjectDependencyInconsistency::InconsistentUses {
424 object_a: *id,
425 object_b: *referenced_by,
426 },
427 );
428 }
429 }
430 for used_by in entry.used_by() {
431 let Some(used_by_entry) = self.entry_by_id.get(used_by) else {
432 dependency_inconsistencies.push(ObjectDependencyInconsistency::MissingUsedBy {
433 object_a: *id,
434 object_b: *used_by,
435 });
436 continue;
437 };
438 if !used_by_entry.uses().contains(id) {
439 dependency_inconsistencies.push(
440 ObjectDependencyInconsistency::InconsistentUses {
441 object_a: *id,
442 object_b: *used_by,
443 },
444 );
445 }
446 }
447 }
448
449 if dependency_inconsistencies.is_empty() {
450 Ok(())
451 } else {
452 Err(dependency_inconsistencies)
453 }
454 }
455
456 fn check_items(&self) -> Result<(), Vec<ItemInconsistency>> {
468 let mut item_inconsistencies = vec![];
469
470 for (db_id, db) in &self.database_by_id {
471 for (schema_name, schema_id) in &db.schemas_by_name {
472 let Some(schema) = db.schemas_by_id.get(schema_id) else {
474 item_inconsistencies.push(ItemInconsistency::MissingSchema {
475 db_id: *db_id,
476 schema_name: schema_name.clone(),
477 });
478 continue;
479 };
480 if schema_name != &schema.name.schema {
481 item_inconsistencies.push(ItemInconsistency::KeyedName {
482 db_schema_by_name: schema_name.clone(),
483 struct_name: schema.name.schema.clone(),
484 });
485 }
486
487 for (item_name, item_id) in schema
492 .items
493 .iter()
494 .chain(schema.types.iter())
495 .chain(schema.functions.iter())
496 {
497 let Some(entry) = self.entry_by_id.get(item_id) else {
498 item_inconsistencies.push(ItemInconsistency::NonExistentItem {
499 db_id: *db_id,
500 schema_id: schema.id,
501 item_id: *item_id,
502 });
503 continue;
504 };
505 if item_name != &entry.name().item {
506 item_inconsistencies.push(ItemInconsistency::ItemNameMismatch {
507 item_id: *item_id,
508 map_name: item_name.clone(),
509 entry_name: entry.name().clone(),
510 });
511 }
512 let statement = match mz_sql::parse::parse(entry.create_sql()) {
513 Ok(mut statements) if statements.len() == 1 => {
514 let statement = statements.pop().expect("checked length");
515 statement.ast
516 }
517 Ok(_) => {
518 item_inconsistencies.push(ItemInconsistency::MultiCreateStatement {
519 create_sql: entry.create_sql().to_string(),
520 });
521 continue;
522 }
523 Err(e) => {
524 item_inconsistencies.push(ItemInconsistency::StatementParseFailure {
525 create_sql: entry.create_sql().to_string(),
526 e,
527 });
528 continue;
529 }
530 };
531 match statement {
532 Statement::CreateConnection(ast::CreateConnectionStatement {
533 name,
534 ..
535 })
536 | Statement::CreateWebhookSource(ast::CreateWebhookSourceStatement {
537 name,
538 ..
539 })
540 | Statement::CreateSource(ast::CreateSourceStatement { name, .. })
541 | Statement::CreateSubsource(ast::CreateSubsourceStatement {
542 name, ..
543 })
544 | Statement::CreateSink(ast::CreateSinkStatement {
545 name: Some(name),
546 ..
547 })
548 | Statement::CreateView(ast::CreateViewStatement {
549 definition: ast::ViewDefinition { name, .. },
550 ..
551 })
552 | Statement::CreateMaterializedView(
553 ast::CreateMaterializedViewStatement { name, .. },
554 )
555 | Statement::CreateTable(ast::CreateTableStatement { name, .. })
556 | Statement::CreateType(ast::CreateTypeStatement { name, .. })
557 | Statement::CreateSecret(ast::CreateSecretStatement { name, .. }) => {
558 let [db_component, schema_component, item_component] = &name.0[..]
559 else {
560 let name =
561 name.0.into_iter().map(|ident| ident.to_string()).collect();
562 item_inconsistencies.push(
563 ItemInconsistency::NonFullyQualifiedItemName {
564 create_sql: entry.create_sql().to_string(),
565 name,
566 },
567 );
568 continue;
569 };
570 if db_component.as_str() != &db.name
571 || schema_component.as_str() != &schema.name.schema
572 || item_component.as_str() != &entry.name().item
573 {
574 item_inconsistencies.push(
575 ItemInconsistency::CreateSqlItemNameMismatch {
576 item_name: vec![
577 db.name.clone(),
578 schema.name.schema.clone(),
579 entry.name().item.clone(),
580 ],
581 create_sql: entry.create_sql().to_string(),
582 },
583 );
584 }
585 }
586 Statement::CreateSchema(ast::CreateSchemaStatement { name, .. }) => {
587 let [db_component, schema_component] = &name.0[..] else {
588 let name =
589 name.0.into_iter().map(|ident| ident.to_string()).collect();
590 item_inconsistencies.push(
591 ItemInconsistency::NonFullyQualifiedSchemaName {
592 create_sql: entry.create_sql().to_string(),
593 name,
594 },
595 );
596 continue;
597 };
598 if db_component.as_str() != &db.name
599 || schema_component.as_str() != &schema.name.schema
600 {
601 item_inconsistencies.push(
602 ItemInconsistency::CreateSqlSchemaNameMismatch {
603 schema_name: vec![
604 db.name.clone(),
605 schema.name.schema.clone(),
606 ],
607 create_sql: entry.create_sql().to_string(),
608 },
609 );
610 }
611 }
612 Statement::CreateDatabase(ast::CreateDatabaseStatement {
613 name, ..
614 }) => {
615 if db.name != name.0.as_str() {
616 item_inconsistencies.push(
617 ItemInconsistency::CreateSqlDatabaseNameMismatch {
618 database_name: db.name.clone(),
619 create_sql: entry.create_sql().to_string(),
620 },
621 );
622 }
623 }
624 _ => (),
625 }
626 }
627 }
628 }
629
630 if item_inconsistencies.is_empty() {
631 Ok(())
632 } else {
633 Err(item_inconsistencies)
634 }
635 }
636}
637
638#[derive(Debug, Serialize, Clone, PartialEq)]
639enum InternalFieldsInconsistency {
640 Database(String, DatabaseId),
641 AmbientSchema(String, SchemaId),
642 Cluster(String, ClusterId),
643 Role(String, RoleId),
644 SourceReferences(CatalogItemId),
645 EntryMissingGlobalIds(CatalogItemId, Vec<GlobalId>),
646 GlobalIdsMissingEntry(GlobalId, CatalogItemId),
647}
648
649#[derive(Debug, Serialize, Clone, PartialEq)]
650enum RoleInconsistency {
651 Database(DatabaseId, RoleId),
652 Schema(SchemaId, RoleId),
653 Entry(CatalogItemId, RoleId),
654 Cluster(ClusterId, RoleId),
655 ClusterReplica(ClusterId, ReplicaId, RoleId),
656 DefaultPrivilege(DefaultPrivilegeObject),
657 RoleAuth(RoleId),
658 DefaultPrivilegeItem {
659 grantor: RoleId,
660 grantee: RoleId,
661 },
662 SystemPrivilege {
663 grantor: Option<RoleId>,
664 grantee: Option<RoleId>,
665 },
666 Membership {
667 parent: Option<RoleId>,
668 grantor: Option<RoleId>,
669 },
670}
671
672#[derive(Debug, Serialize, Clone, PartialEq)]
673enum CommentInconsistency {
674 Dangling(CommentObjectId),
676 NonRelation(CommentObjectId, usize),
678}
679
680#[derive(Debug, Serialize, Clone, PartialEq)]
681enum ObjectDependencyInconsistency {
682 MissingUses {
684 object_a: CatalogItemId,
685 object_b: CatalogItemId,
686 },
687 MissingUsedBy {
689 object_a: CatalogItemId,
690 object_b: CatalogItemId,
691 },
692 InconsistentUsedBy {
694 object_a: CatalogItemId,
695 object_b: CatalogItemId,
696 },
697 InconsistentUses {
699 object_a: CatalogItemId,
700 object_b: CatalogItemId,
701 },
702}
703
704#[derive(Debug, Serialize, Clone, PartialEq)]
705enum ItemInconsistency {
706 KeyedName {
708 db_schema_by_name: String,
709 struct_name: String,
710 },
711 MissingSchema {
713 db_id: DatabaseId,
714 schema_name: String,
715 },
716 NonExistentItem {
718 db_id: DatabaseId,
719 schema_id: SchemaSpecifier,
720 item_id: CatalogItemId,
721 },
722 ItemNameMismatch {
724 item_id: CatalogItemId,
725 map_name: String,
727 entry_name: QualifiedItemName,
729 },
730 StatementParseFailure {
732 create_sql: String,
733 e: ParserStatementError,
734 },
735 MultiCreateStatement { create_sql: String },
737 NonFullyQualifiedItemName {
739 create_sql: String,
740 name: Vec<String>,
741 },
742 NonFullyQualifiedSchemaName {
744 create_sql: String,
745 name: Vec<String>,
746 },
747 CreateSqlItemNameMismatch {
749 item_name: Vec<String>,
750 create_sql: String,
751 },
752 CreateSqlSchemaNameMismatch {
754 schema_name: Vec<String>,
755 create_sql: String,
756 },
757 CreateSqlDatabaseNameMismatch {
759 database_name: String,
760 create_sql: String,
761 },
762}