catalog: Add tests for all durable state methods (#22372)

This commit adds integration tests to cover all methods of the
DurableCatalogState trait.

src/adapter/src/catalog.rs

@@ -27,7 +27,8 @@ use mz_catalog::builtin::{
     BUILTIN_PREFIXES, MZ_INTROSPECTION_CLUSTER,
 };
 use mz_catalog::{
-    BootstrapArgs, DurableCatalogState, OpenableDurableCatalogState, StashConfig, Transaction,
+    debug_bootstrap_args, DurableCatalogState, OpenableDurableCatalogState, StashConfig,
+    Transaction,
 };
 use mz_compute_types::dataflows::DataflowDescription;
 use mz_controller::clusters::{
@@ -456,7 +457,7 @@ impl Catalog {
         let mut openable_storage =
             mz_catalog::debug_stash_backed_catalog_state(debug_stash_factory);
         let storage = openable_storage
-            .open(now.clone(), &Self::debug_bootstrap_args(), None)
+            .open(now.clone(), &debug_bootstrap_args(), None)
             .await?;
         Self::open_debug_stash_catalog(storage, now).await
     }
@@ -480,7 +481,7 @@ impl Catalog {
         };
         let mut openable_storage = mz_catalog::stash_backed_catalog_state(stash_config);
         let storage = openable_storage
-            .open(now.clone(), &Self::debug_bootstrap_args(), None)
+            .open(now.clone(), &debug_bootstrap_args(), None)
             .await?;
         Self::open_debug_stash_catalog(storage, now).await
     }
@@ -494,7 +495,7 @@ impl Catalog {
     ) -> Result<Catalog, anyhow::Error> {
         let mut openable_storage = mz_catalog::stash_backed_catalog_state(stash_config);
         let storage = openable_storage
-            .open_read_only(now.clone(), &Self::debug_bootstrap_args())
+            .open_read_only(now.clone(), &debug_bootstrap_args())
             .await?;
         Self::open_debug_stash_catalog(storage, now).await
     }
@@ -546,14 +547,6 @@ impl Catalog {
         Ok(catalog)
     }
 
-    fn debug_bootstrap_args() -> BootstrapArgs {
-        BootstrapArgs {
-            default_cluster_replica_size: "1".into(),
-            builtin_cluster_replica_size: "1".into(),
-            bootstrap_role: None,
-        }
-    }
-
     pub fn for_session<'a>(&'a self, session: &'a Session) -> ConnCatalog<'a> {
         Self::for_session_state(&self.state, session)
     }

src/catalog/src/lib.rs

@@ -122,13 +122,15 @@ mod error;
 pub mod initialize;
 pub mod objects;
 
-const DATABASE_ID_ALLOC_KEY: &str = "database";
-const SCHEMA_ID_ALLOC_KEY: &str = "schema";
-const USER_ROLE_ID_ALLOC_KEY: &str = "user_role";
-const USER_CLUSTER_ID_ALLOC_KEY: &str = "user_compute";
-const SYSTEM_CLUSTER_ID_ALLOC_KEY: &str = "system_compute";
-const USER_REPLICA_ID_ALLOC_KEY: &str = "replica";
-const SYSTEM_REPLICA_ID_ALLOC_KEY: &str = "system_replica";
+pub const DATABASE_ID_ALLOC_KEY: &str = "database";
+pub const SCHEMA_ID_ALLOC_KEY: &str = "schema";
+pub const USER_ITEM_ALLOC_KEY: &str = "user";
+pub const SYSTEM_ITEM_ALLOC_KEY: &str = "system";
+pub const USER_ROLE_ID_ALLOC_KEY: &str = "user_role";
+pub const USER_CLUSTER_ID_ALLOC_KEY: &str = "user_compute";
+pub const SYSTEM_CLUSTER_ID_ALLOC_KEY: &str = "system_compute";
+pub const USER_REPLICA_ID_ALLOC_KEY: &str = "replica";
+pub const SYSTEM_REPLICA_ID_ALLOC_KEY: &str = "system_replica";
 pub const AUDIT_LOG_ID_ALLOC_KEY: &str = "auditlog";
 pub const STORAGE_USAGE_ID_ALLOC_KEY: &str = "storage_usage";
 
@@ -353,14 +355,14 @@ pub trait DurableCatalogState: ReadOnlyDurableCatalogState {
 
     /// Allocates and returns `amount` system [`GlobalId`]s.
     async fn allocate_system_ids(&mut self, amount: u64) -> Result<Vec<GlobalId>, CatalogError> {
-        let id = self.allocate_id("system", amount).await?;
+        let id = self.allocate_id(SYSTEM_ITEM_ALLOC_KEY, amount).await?;
 
         Ok(id.into_iter().map(GlobalId::System).collect())
     }
 
     /// Allocates and returns a user [`GlobalId`].
     async fn allocate_user_id(&mut self) -> Result<GlobalId, CatalogError> {
-        let id = self.allocate_id("user", 1).await?;
+        let id = self.allocate_id(USER_ITEM_ALLOC_KEY, 1).await?;
         let id = id.into_element();
         Ok(GlobalId::User(id))
     }
@@ -402,3 +404,11 @@ pub fn debug_stash_backed_catalog_state(
 ) -> impl OpenableDurableCatalogState<Connection> + '_ {
     DebugOpenableConnection::new(debug_stash_factory)
 }
+
+pub fn debug_bootstrap_args() -> BootstrapArgs {
+    BootstrapArgs {
+        default_cluster_replica_size: "1".into(),
+        builtin_cluster_replica_size: "1".into(),
+        bootstrap_role: None,
+    }
+}

src/catalog/src/objects.rs

@@ -24,7 +24,6 @@ use mz_stash_types::objects::{proto, RustType, TryFromProtoError};
 use mz_storage_types::sources::Timeline;
 use proptest_derive::Arbitrary;
 use std::collections::BTreeMap;
-use std::time::Duration;
 
 // Structs used to pass information to outside modules.
 
@@ -38,7 +37,7 @@ use std::time::Duration;
 /// This trait is based on [`RustType`], however it is meant to
 /// convert the types used in [`RustType`] to a more consumable and
 /// condensed type.
-pub(crate) trait DurableType<K, V>: Sized {
+pub trait DurableType<K, V>: Sized {
     /// Consume and convert `Self` into a `(K, V)` key-value pair.
     fn into_key_value(self) -> (K, V);
 
@@ -47,7 +46,7 @@ pub(crate) trait DurableType<K, V>: Sized {
     fn from_key_value(key: K, value: V) -> Self;
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Database {
     pub id: DatabaseId,
     pub name: String,
@@ -77,7 +76,7 @@ impl DurableType<DatabaseKey, DatabaseValue> for Database {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Schema {
     pub id: SchemaId,
     pub name: String,
@@ -110,7 +109,7 @@ impl DurableType<SchemaKey, SchemaValue> for Schema {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Role {
     pub id: RoleId,
     pub name: String,
@@ -143,7 +142,7 @@ impl DurableType<RoleKey, RoleValue> for Role {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Cluster {
     pub id: ClusterId,
     pub name: String,
@@ -258,7 +257,7 @@ pub struct ClusterVariantManaged {
     pub disk: bool,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub struct IntrospectionSourceIndex {
     pub cluster_id: ClusterId,
     pub name: String,
@@ -307,7 +306,7 @@ impl DurableType<ClusterIntrospectionSourceIndexKey, ClusterIntrospectionSourceI
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ClusterReplica {
     pub cluster_id: ClusterId,
     pub replica_id: ReplicaId,
@@ -505,13 +504,7 @@ impl RustType<proto::replica_config::Location> for ReplicaLocation {
     }
 }
 
-#[derive(Clone, Debug)]
-pub struct ComputeReplicaLogging {
-    pub log_logging: bool,
-    pub interval: Option<Duration>,
-}
-
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Item {
     pub id: GlobalId,
     pub schema_id: SchemaId,
@@ -554,7 +547,7 @@ pub struct SystemObjectDescription {
     pub object_name: String,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct SystemObjectUniqueIdentifier {
     pub id: GlobalId,
     pub fingerprint: String,
@@ -567,7 +560,7 @@ pub struct SystemObjectUniqueIdentifier {
 /// As such, system objects are keyed in the catalog storage by the
 /// tuple (schema_name, object_type, object_name), which is guaranteed
 /// to be unique.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct SystemObjectMapping {
     pub description: SystemObjectDescription,
     pub unique_identifier: SystemObjectUniqueIdentifier,
@@ -608,7 +601,7 @@ impl DurableType<GidMappingKey, GidMappingValue> for SystemObjectMapping {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct DefaultPrivilege {
     pub object: DefaultPrivilegeObject,
     pub acl_item: DefaultPrivilegeAclItem,
@@ -646,7 +639,7 @@ impl DurableType<DefaultPrivilegesKey, DefaultPrivilegesValue> for DefaultPrivil
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Comment {
     pub object_id: CommentObjectId,
     pub sub_component: Option<usize>,
@@ -675,7 +668,7 @@ impl DurableType<CommentKey, CommentValue> for Comment {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct IdAlloc {
     pub name: String,
     pub next_id: u64,
@@ -699,7 +692,7 @@ impl DurableType<IdAllocKey, IdAllocValue> for IdAlloc {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Config {
     pub key: String,
     pub value: u64,
@@ -743,7 +736,7 @@ impl DurableType<SettingKey, SettingValue> for Setting {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct TimelineTimestamp {
     pub timeline: Timeline,
     pub ts: mz_repr::Timestamp,
@@ -767,7 +760,7 @@ impl DurableType<TimestampKey, TimestampValue> for TimelineTimestamp {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct SystemConfiguration {
     pub name: String,
     pub value: String,