mz_catalog/durable/
objects.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! The current types used to represent catalog data stored on disk. These objects generally fall
11//! into two categories.
12//!
13//! The key-value objects are a one-to-one mapping of the protobuf objects used to save catalog
14//! data durably. They can be converted to and from protobuf via the [`mz_proto::RustType`] trait.
15//! These objects should not be exposed anywhere outside the [`crate::durable`] module.
16//!
17//! The other type of objects combine the information from keys and values into a single struct,
18//! but are still a direct representation of the data stored on disk. They can be converted to and
19//! from the key-value objects via the [`DurableType`] trait. These objects are used to pass
20//! information to other modules in this crate and other catalog related code.
21//!
22//! All non-catalog code should interact with the objects in [`crate::memory::objects`] and never
23//! directly interact with the objects in this module.
24//!
25//! As an example, [`DatabaseKey`] and [`DatabaseValue`] are key-value objects, while [`Database`]
26//! is the non-key-value counterpart.
27
28pub mod serialization;
29pub(crate) mod state_update;
30
31use std::cmp::Ordering;
32use std::collections::BTreeMap;
33use std::time::Duration;
34
35use mz_audit_log::VersionedEvent;
36use mz_controller::clusters::ReplicaLogging;
37use mz_controller_types::{ClusterId, ReplicaId};
38use mz_persist_types::ShardId;
39use mz_repr::adt::mz_acl_item::{AclMode, MzAclItem};
40use mz_repr::network_policy_id::NetworkPolicyId;
41use mz_repr::role_id::RoleId;
42use mz_repr::{CatalogItemId, GlobalId, RelationVersion};
43use mz_sql::catalog::{
44    CatalogItemType, DefaultPrivilegeAclItem, DefaultPrivilegeObject, ObjectType, RoleAttributes,
45    RoleMembership, RoleVars,
46};
47use mz_sql::names::{CommentObjectId, DatabaseId, SchemaId};
48use mz_sql::plan::{AutoScalingStrategy, ClusterSchedule, NetworkPolicyRule, OnTimeoutAction};
49#[cfg(test)]
50use proptest_derive::Arbitrary;
51
52use crate::builtin::RUNTIME_ALTERABLE_FINGERPRINT_SENTINEL;
53use crate::durable::Epoch;
54use crate::durable::objects::serialization::proto;
55
56// Structs used to pass information to outside modules.
57
58/// A trait for representing `Self` as a key-value pair of type
59/// `(Key, Value)` for the purpose of storing this value durably.
60///
61/// To encode a key-value pair, use [`DurableType::into_key_value`].
62///
63/// To decode a key-value pair, use [`DurableType::from_key_value`].
64///
65/// This trait is based on [`RustType`], however it is meant to
66/// convert the types used in [`RustType`] to a more consumable and
67/// condensed type.
68///
69/// [`RustType`]: mz_proto::RustType
70pub trait DurableType: Sized {
71    type Key;
72    type Value;
73
74    /// Consume and convert `Self` into a `(Key, Value)` key-value pair.
75    fn into_key_value(self) -> (Self::Key, Self::Value);
76
77    /// Consume and convert a `(Key, Value)` key-value pair back into a
78    /// `Self` value.
79    fn from_key_value(key: Self::Key, value: Self::Value) -> Self;
80
81    // TODO(jkosh44) Would be great to not clone, since this is always used for lookups which only
82    // needs a reference. In practice, we currently almost always use this method to clone a single
83    // 64-bit integer, so it's not a huge deal.
84    /// Produce a `Key` from self. This may involve cloning/copying required fields.
85    fn key(&self) -> Self::Key;
86}
87
88#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
89pub struct Database {
90    pub id: DatabaseId,
91    pub oid: u32,
92    pub name: String,
93    pub owner_id: RoleId,
94    pub privileges: Vec<MzAclItem>,
95}
96
97impl DurableType for Database {
98    type Key = DatabaseKey;
99    type Value = DatabaseValue;
100
101    fn into_key_value(self) -> (Self::Key, Self::Value) {
102        (
103            DatabaseKey { id: self.id },
104            DatabaseValue {
105                oid: self.oid,
106                name: self.name,
107                owner_id: self.owner_id,
108                privileges: self.privileges,
109            },
110        )
111    }
112
113    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
114        Self {
115            id: key.id,
116            oid: value.oid,
117            name: value.name,
118            owner_id: value.owner_id,
119            privileges: value.privileges,
120        }
121    }
122
123    fn key(&self) -> Self::Key {
124        DatabaseKey { id: self.id }
125    }
126}
127
128#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
129pub struct Schema {
130    pub id: SchemaId,
131    pub oid: u32,
132    pub name: String,
133    pub database_id: Option<DatabaseId>,
134    pub owner_id: RoleId,
135    pub privileges: Vec<MzAclItem>,
136}
137
138impl DurableType for Schema {
139    type Key = SchemaKey;
140    type Value = SchemaValue;
141
142    fn into_key_value(self) -> (Self::Key, Self::Value) {
143        (
144            SchemaKey { id: self.id },
145            SchemaValue {
146                oid: self.oid,
147                database_id: self.database_id,
148                name: self.name,
149                owner_id: self.owner_id,
150                privileges: self.privileges,
151            },
152        )
153    }
154
155    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
156        Self {
157            id: key.id,
158            oid: value.oid,
159            name: value.name,
160            database_id: value.database_id,
161            owner_id: value.owner_id,
162            privileges: value.privileges,
163        }
164    }
165
166    fn key(&self) -> Self::Key {
167        SchemaKey { id: self.id }
168    }
169}
170
171#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
172pub struct Role {
173    pub id: RoleId,
174    pub oid: u32,
175    pub name: String,
176    pub attributes: RoleAttributes,
177    pub membership: RoleMembership,
178    pub vars: RoleVars,
179}
180
181impl DurableType for Role {
182    type Key = RoleKey;
183    type Value = RoleValue;
184
185    fn into_key_value(self) -> (Self::Key, Self::Value) {
186        (
187            RoleKey { id: self.id },
188            RoleValue {
189                oid: self.oid,
190                name: self.name,
191                attributes: self.attributes,
192                membership: self.membership,
193                vars: self.vars,
194            },
195        )
196    }
197
198    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
199        Self {
200            id: key.id,
201            oid: value.oid,
202            name: value.name,
203            attributes: value.attributes,
204            membership: value.membership,
205            vars: value.vars,
206        }
207    }
208
209    fn key(&self) -> Self::Key {
210        RoleKey { id: self.id }
211    }
212}
213
214#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
215pub struct RoleAuth {
216    pub role_id: RoleId,
217    pub password_hash: Option<String>,
218    pub updated_at: u64,
219}
220
221impl DurableType for RoleAuth {
222    type Key = RoleAuthKey;
223    type Value = RoleAuthValue;
224
225    fn into_key_value(self) -> (Self::Key, Self::Value) {
226        (
227            RoleAuthKey {
228                role_id: self.role_id,
229            },
230            RoleAuthValue {
231                password_hash: self.password_hash,
232                updated_at: self.updated_at,
233            },
234        )
235    }
236
237    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
238        Self {
239            role_id: key.role_id,
240            password_hash: value.password_hash,
241            updated_at: value.updated_at,
242        }
243    }
244
245    fn key(&self) -> Self::Key {
246        RoleAuthKey {
247            role_id: self.role_id,
248        }
249    }
250}
251
252#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
253pub struct NetworkPolicy {
254    pub name: String,
255    pub id: NetworkPolicyId,
256    pub oid: u32,
257    pub rules: Vec<NetworkPolicyRule>,
258    pub owner_id: RoleId,
259    pub(crate) privileges: Vec<MzAclItem>,
260}
261
262impl DurableType for NetworkPolicy {
263    type Key = NetworkPolicyKey;
264    type Value = NetworkPolicyValue;
265
266    fn into_key_value(self) -> (Self::Key, Self::Value) {
267        (
268            NetworkPolicyKey { id: self.id },
269            NetworkPolicyValue {
270                oid: self.oid,
271                name: self.name,
272                rules: self.rules,
273                owner_id: self.owner_id,
274                privileges: self.privileges,
275            },
276        )
277    }
278
279    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
280        Self {
281            id: key.id,
282            oid: value.oid,
283            name: value.name,
284            rules: value.rules,
285            owner_id: value.owner_id,
286            privileges: value.privileges,
287        }
288    }
289
290    fn key(&self) -> Self::Key {
291        NetworkPolicyKey { id: self.id }
292    }
293}
294
295#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
296pub struct Cluster {
297    pub id: ClusterId,
298    pub name: String,
299    pub owner_id: RoleId,
300    pub privileges: Vec<MzAclItem>,
301    pub config: ClusterConfig,
302}
303
304impl DurableType for Cluster {
305    type Key = ClusterKey;
306    type Value = ClusterValue;
307
308    fn into_key_value(self) -> (Self::Key, Self::Value) {
309        (
310            ClusterKey { id: self.id },
311            ClusterValue {
312                name: self.name,
313                owner_id: self.owner_id,
314                privileges: self.privileges,
315                config: self.config,
316            },
317        )
318    }
319
320    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
321        Self {
322            id: key.id,
323            name: value.name,
324            owner_id: value.owner_id,
325            privileges: value.privileges,
326            config: value.config,
327        }
328    }
329
330    fn key(&self) -> Self::Key {
331        ClusterKey { id: self.id }
332    }
333}
334
335#[derive(Clone, Debug, PartialOrd, PartialEq, Eq, Ord)]
336pub struct ClusterConfig {
337    pub variant: ClusterVariant,
338    pub workload_class: Option<String>,
339}
340
341#[derive(Clone, Debug, PartialOrd, PartialEq, Eq, Ord)]
342pub enum ClusterVariant {
343    Managed(ClusterVariantManaged),
344    Unmanaged,
345}
346
347#[derive(Clone, Debug, PartialOrd, PartialEq, Eq, Ord)]
348pub struct ClusterVariantManaged {
349    pub size: String,
350    pub availability_zones: Vec<String>,
351    pub logging: ReplicaLogging,
352    /// Whether arrangements on this cluster's replicas request dictionary compression.
353    pub arrangement_compression: bool,
354    pub replication_factor: u32,
355    pub optimizer_feature_overrides: BTreeMap<String, String>,
356    pub schedule: ClusterSchedule,
357    /// User-configured autoscaling policy, distinct from the in-flight runtime
358    /// records below.
359    pub auto_scaling_strategy: Option<AutoScalingStrategy>,
360    /// Latest graceful reconfiguration record, if one has been written.
361    pub reconfiguration: Option<ReconfigurationState>,
362    /// In-flight hydration burst the controller is running.
363    pub burst: Option<BurstState>,
364}
365
366/// The canonical name of the `index`-th (zero-based) replica of a managed
367/// cluster.
368///
369/// A managed cluster's replicas are derived from its `replication_factor`: for a
370/// factor of N they are named `r1` through `rN`.
371///
372/// `ALTER CLUSTER` computes the replicas it creates and drops by this rule, and
373/// so does the catalog-open reconciler that materializes builtin replicas. The
374/// two have to share it, or `ALTER` cannot find the replicas it means to change.
375/// The cluster controller deliberately does not: its `ReplicaNameGen` picks names
376/// that avoid the observed set, which is why `ALTER` against a controller-owned
377/// cluster drops by observed id rather than by derived name.
378pub fn managed_cluster_replica_name(index: u32) -> String {
379    format!("r{}", index + 1)
380}
381
382/// The latest graceful reconfiguration: the config shape the cluster is moving
383/// to or most recently moved toward, plus its deadline and terminal state.
384///
385/// `ALTER` writes this record with [`ReconfigurationStatus::InProgress`] and
386/// returns. The realized config (`cluster.size`, ...) is advanced by the
387/// controller only at cut-over. When the reconfiguration settles, the controller
388/// retains the record with a terminal status so readers can inspect the latest
389/// outcome without reconstructing it from the audit log.
390#[derive(Clone, Debug, PartialOrd, PartialEq, Eq, Ord)]
391pub struct ReconfigurationState {
392    pub target: ReconfigurationTarget,
393    pub deadline: mz_repr::Timestamp,
394    /// The action the controller applies if `deadline` passes before the
395    /// target hydrates. Success takes precedence: a hydrated target cuts over
396    /// regardless of this field.
397    pub on_timeout: OnTimeoutAction,
398    pub status: ReconfigurationStatus,
399}
400
401/// The lifecycle status of the latest graceful reconfiguration.
402#[derive(Clone, Copy, Debug, PartialOrd, PartialEq, Eq, Ord)]
403pub enum ReconfigurationStatus {
404    /// The controller is converging the cluster onto the target shape.
405    InProgress,
406    /// The realized config reached the target shape.
407    Finalized,
408    /// The deadline fired under rollback and the realized config stayed put.
409    TimedOut,
410    /// The user retargeted the reconfiguration back to the realized shape.
411    Cancelled,
412    /// The controller could not create the target replicas within the budget.
413    ResourceExhausted,
414}
415
416impl ReconfigurationState {
417    /// Whether this record should still drive target-replica convergence.
418    pub fn is_in_progress(&self) -> bool {
419        matches!(self.status, ReconfigurationStatus::InProgress)
420    }
421}
422
423/// The full config shape a reconfiguration is moving the cluster to, so a
424/// combined size + replication-factor + availability-zone change is one record.
425#[derive(Clone, Debug, PartialOrd, PartialEq, Eq, Ord)]
426pub struct ReconfigurationTarget {
427    pub size: String,
428    pub replication_factor: u32,
429    pub availability_zones: Vec<String>,
430    pub logging: ReplicaLogging,
431    pub arrangement_compression: bool,
432}
433
434/// An active hydration burst the controller is running.
435#[derive(Clone, Debug, PartialOrd, PartialEq, Eq, Ord)]
436pub struct BurstState {
437    pub burst_size: String,
438    pub linger_duration: Duration,
439    /// When the steady-state replicas were first observed hydrated. Absent
440    /// until that observation; the linger countdown runs from this point.
441    pub steady_hydrated_at: Option<mz_repr::Timestamp>,
442}
443
444#[derive(Clone, Debug, Ord, PartialOrd, PartialEq, Eq)]
445pub struct IntrospectionSourceIndex {
446    pub cluster_id: ClusterId,
447    pub name: String,
448    pub item_id: CatalogItemId,
449    pub index_id: GlobalId,
450    pub oid: u32,
451}
452
453impl DurableType for IntrospectionSourceIndex {
454    type Key = ClusterIntrospectionSourceIndexKey;
455    type Value = ClusterIntrospectionSourceIndexValue;
456
457    fn into_key_value(self) -> (Self::Key, Self::Value) {
458        (
459            ClusterIntrospectionSourceIndexKey {
460                cluster_id: self.cluster_id,
461                name: self.name,
462            },
463            ClusterIntrospectionSourceIndexValue {
464                catalog_id: self
465                    .item_id
466                    .try_into()
467                    .expect("cluster introspection source index mapping must be an Introspection Source Index ID"),
468                global_id: self
469                    .index_id
470                    .try_into()
471                    .expect("cluster introspection source index mapping must be a Introspection Source Index ID"),
472                oid: self.oid,
473            },
474        )
475    }
476
477    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
478        Self {
479            cluster_id: key.cluster_id,
480            name: key.name,
481            item_id: value.catalog_id.into(),
482            index_id: value.global_id.into(),
483            oid: value.oid,
484        }
485    }
486
487    fn key(&self) -> Self::Key {
488        ClusterIntrospectionSourceIndexKey {
489            cluster_id: self.cluster_id,
490            name: self.name.clone(),
491        }
492    }
493}
494
495#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
496pub struct ClusterReplica {
497    pub cluster_id: ClusterId,
498    pub replica_id: ReplicaId,
499    pub name: String,
500    pub config: ReplicaConfig,
501    pub owner_id: RoleId,
502}
503
504impl DurableType for ClusterReplica {
505    type Key = ClusterReplicaKey;
506    type Value = ClusterReplicaValue;
507
508    fn into_key_value(self) -> (Self::Key, Self::Value) {
509        (
510            ClusterReplicaKey {
511                id: self.replica_id,
512            },
513            ClusterReplicaValue {
514                cluster_id: self.cluster_id,
515                name: self.name,
516                config: self.config,
517                owner_id: self.owner_id,
518            },
519        )
520    }
521
522    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
523        Self {
524            cluster_id: value.cluster_id,
525            replica_id: key.id,
526            name: value.name,
527            config: value.config,
528            owner_id: value.owner_id,
529        }
530    }
531
532    fn key(&self) -> Self::Key {
533        ClusterReplicaKey {
534            id: self.replica_id,
535        }
536    }
537}
538
539// The on-disk replica configuration does not match the in-memory replica configuration, so we need
540// separate structs. As of writing this comment, it is mainly due to the fact that we don't persist
541// the replica allocation.
542#[derive(Clone, Debug, PartialOrd, PartialEq, Eq, Ord)]
543pub struct ReplicaConfig {
544    pub location: ReplicaLocation,
545    pub logging: ReplicaLogging,
546    pub arrangement_compression: bool,
547}
548
549impl From<mz_controller::clusters::ReplicaConfig> for ReplicaConfig {
550    fn from(config: mz_controller::clusters::ReplicaConfig) -> Self {
551        Self {
552            location: config.location.into(),
553            logging: config.compute.logging,
554            arrangement_compression: config.compute.arrangement_compression,
555        }
556    }
557}
558
559#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord)]
560pub enum ReplicaLocation {
561    Unmanaged {
562        storagectl_addrs: Vec<String>,
563        computectl_addrs: Vec<String>,
564    },
565    Managed {
566        size: String,
567        /// The availability zones the replica was provisioned under.
568        ///
569        /// For a replica of a managed cluster this is the cluster's
570        /// `AVAILABILITY ZONES` pool at provision time; the cluster controller
571        /// compares it against a cluster's target `availability_zones` to tell
572        /// realized- from target-shape replicas (including an
573        /// `AVAILABILITY ZONES` divergence). For a replica of an unmanaged
574        /// cluster it is the user-pinned `AVAILABILITY ZONE`, as a zero- or
575        /// one-element list. Empty when no zones constrain placement.
576        availability_zones: Vec<String>,
577        internal: bool,
578        billed_as: Option<String>,
579        pending: bool,
580    },
581}
582
583impl From<mz_controller::clusters::ReplicaLocation> for ReplicaLocation {
584    fn from(loc: mz_controller::clusters::ReplicaLocation) -> Self {
585        match loc {
586            mz_controller::clusters::ReplicaLocation::Unmanaged(
587                mz_controller::clusters::UnmanagedReplicaLocation {
588                    storagectl_addrs,
589                    computectl_addrs,
590                },
591            ) => Self::Unmanaged {
592                storagectl_addrs,
593                computectl_addrs,
594            },
595            mz_controller::clusters::ReplicaLocation::Managed(
596                mz_controller::clusters::ManagedReplicaLocation {
597                    allocation: _,
598                    size,
599                    availability_zones,
600                    billed_as,
601                    internal,
602                    pending,
603                },
604            ) => ReplicaLocation::Managed {
605                size,
606                availability_zones,
607                internal,
608                billed_as,
609                pending,
610            },
611        }
612    }
613}
614
615#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
616pub struct Item {
617    pub id: CatalogItemId,
618    pub oid: u32,
619    pub global_id: GlobalId,
620    pub schema_id: SchemaId,
621    pub name: String,
622    pub create_sql: String,
623    pub owner_id: RoleId,
624    pub privileges: Vec<MzAclItem>,
625    pub extra_versions: BTreeMap<RelationVersion, GlobalId>,
626}
627
628impl Item {
629    pub fn item_type(&self) -> CatalogItemType {
630        item_type(&self.create_sql)
631    }
632}
633
634impl DurableType for Item {
635    type Key = ItemKey;
636    type Value = ItemValue;
637
638    fn into_key_value(self) -> (Self::Key, Self::Value) {
639        (
640            ItemKey { id: self.id },
641            ItemValue {
642                oid: self.oid,
643                global_id: self.global_id,
644                schema_id: self.schema_id,
645                name: self.name,
646                create_sql: self.create_sql,
647                owner_id: self.owner_id,
648                privileges: self.privileges,
649                extra_versions: self.extra_versions,
650            },
651        )
652    }
653
654    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
655        Self {
656            id: key.id,
657            oid: value.oid,
658            global_id: value.global_id,
659            schema_id: value.schema_id,
660            name: value.name,
661            create_sql: value.create_sql,
662            owner_id: value.owner_id,
663            privileges: value.privileges,
664            extra_versions: value.extra_versions,
665        }
666    }
667
668    fn key(&self) -> Self::Key {
669        ItemKey { id: self.id }
670    }
671}
672
673#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
674pub struct SourceReferences {
675    pub source_id: CatalogItemId,
676    pub updated_at: u64,
677    pub references: Vec<SourceReference>,
678}
679
680#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
681#[cfg_attr(test, derive(Arbitrary))]
682pub struct SourceReference {
683    pub name: String,
684    pub namespace: Option<String>,
685    pub columns: Vec<String>,
686}
687
688impl DurableType for SourceReferences {
689    type Key = SourceReferencesKey;
690    type Value = SourceReferencesValue;
691
692    fn into_key_value(self) -> (Self::Key, Self::Value) {
693        (
694            SourceReferencesKey {
695                source_id: self.source_id,
696            },
697            SourceReferencesValue {
698                updated_at: self.updated_at,
699                references: self.references,
700            },
701        )
702    }
703
704    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
705        Self {
706            source_id: key.source_id,
707            updated_at: value.updated_at,
708            references: value.references,
709        }
710    }
711
712    fn key(&self) -> Self::Key {
713        SourceReferencesKey {
714            source_id: self.source_id,
715        }
716    }
717}
718
719/// A newtype wrapper for [`CatalogItemId`] that is only for the "system" namespace.
720#[derive(Debug, Copy, Clone, Ord, PartialOrd, PartialEq, Eq)]
721pub struct SystemCatalogItemId(u64);
722
723impl TryFrom<CatalogItemId> for SystemCatalogItemId {
724    type Error = &'static str;
725
726    fn try_from(val: CatalogItemId) -> Result<Self, Self::Error> {
727        match val {
728            CatalogItemId::System(x) => Ok(SystemCatalogItemId(x)),
729            CatalogItemId::IntrospectionSourceIndex(_) => Err("introspection_source_index"),
730            CatalogItemId::User(_) => Err("user"),
731            CatalogItemId::Transient(_) => Err("transient"),
732        }
733    }
734}
735
736impl From<SystemCatalogItemId> for CatalogItemId {
737    fn from(val: SystemCatalogItemId) -> Self {
738        CatalogItemId::System(val.0)
739    }
740}
741
742/// A newtype wrapper for [`CatalogItemId`] that is only for the "introspection source index" namespace.
743#[derive(Debug, Copy, Clone, Ord, PartialOrd, PartialEq, Eq)]
744pub struct IntrospectionSourceIndexCatalogItemId(u64);
745
746impl TryFrom<CatalogItemId> for IntrospectionSourceIndexCatalogItemId {
747    type Error = &'static str;
748
749    fn try_from(val: CatalogItemId) -> Result<Self, Self::Error> {
750        match val {
751            CatalogItemId::System(_) => Err("system"),
752            CatalogItemId::IntrospectionSourceIndex(x) => {
753                Ok(IntrospectionSourceIndexCatalogItemId(x))
754            }
755            CatalogItemId::User(_) => Err("user"),
756            CatalogItemId::Transient(_) => Err("transient"),
757        }
758    }
759}
760
761impl From<IntrospectionSourceIndexCatalogItemId> for CatalogItemId {
762    fn from(val: IntrospectionSourceIndexCatalogItemId) -> Self {
763        CatalogItemId::IntrospectionSourceIndex(val.0)
764    }
765}
766
767/// A newtype wrapper for [`GlobalId`] that is only for the "system" namespace.
768#[derive(Debug, Copy, Clone, Ord, PartialOrd, PartialEq, Eq)]
769pub struct SystemGlobalId(u64);
770
771impl TryFrom<GlobalId> for SystemGlobalId {
772    type Error = &'static str;
773
774    fn try_from(val: GlobalId) -> Result<Self, Self::Error> {
775        match val {
776            GlobalId::System(x) => Ok(SystemGlobalId(x)),
777            GlobalId::IntrospectionSourceIndex(_) => Err("introspection_source_index"),
778            GlobalId::User(_) => Err("user"),
779            GlobalId::Transient(_) => Err("transient"),
780            GlobalId::Explain => Err("explain"),
781        }
782    }
783}
784
785impl From<SystemGlobalId> for GlobalId {
786    fn from(val: SystemGlobalId) -> Self {
787        GlobalId::System(val.0)
788    }
789}
790
791/// A newtype wrapper for [`GlobalId`] that is only for the "introspection source index" namespace.
792#[derive(Debug, Copy, Clone, Ord, PartialOrd, PartialEq, Eq)]
793pub struct IntrospectionSourceIndexGlobalId(u64);
794
795impl TryFrom<GlobalId> for IntrospectionSourceIndexGlobalId {
796    type Error = &'static str;
797
798    fn try_from(val: GlobalId) -> Result<Self, Self::Error> {
799        match val {
800            GlobalId::System(_) => Err("system"),
801            GlobalId::IntrospectionSourceIndex(x) => Ok(IntrospectionSourceIndexGlobalId(x)),
802            GlobalId::User(_) => Err("user"),
803            GlobalId::Transient(_) => Err("transient"),
804            GlobalId::Explain => Err("explain"),
805        }
806    }
807}
808
809impl From<IntrospectionSourceIndexGlobalId> for GlobalId {
810    fn from(val: IntrospectionSourceIndexGlobalId) -> Self {
811        GlobalId::IntrospectionSourceIndex(val.0)
812    }
813}
814
815#[derive(Debug, Clone, PartialOrd, Ord, PartialEq, Eq, Hash)]
816pub struct SystemObjectDescription {
817    pub schema_name: String,
818    pub object_type: CatalogItemType,
819    pub object_name: String,
820}
821
822#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
823pub struct SystemObjectUniqueIdentifier {
824    pub catalog_id: CatalogItemId,
825    pub global_id: GlobalId,
826    pub fingerprint: String,
827}
828
829impl SystemObjectUniqueIdentifier {
830    pub fn runtime_alterable(&self) -> bool {
831        self.fingerprint == RUNTIME_ALTERABLE_FINGERPRINT_SENTINEL
832    }
833}
834
835/// Functions can share the same name as any other catalog item type
836/// within a given schema.
837/// For example, a function can have the same name as a type, e.g.
838/// 'date'.
839/// As such, system objects are keyed in the catalog storage by the
840/// tuple (schema_name, object_type, object_name), which is guaranteed
841/// to be unique.
842#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
843pub struct SystemObjectMapping {
844    pub description: SystemObjectDescription,
845    pub unique_identifier: SystemObjectUniqueIdentifier,
846}
847
848impl DurableType for SystemObjectMapping {
849    type Key = GidMappingKey;
850    type Value = GidMappingValue;
851
852    fn into_key_value(self) -> (Self::Key, Self::Value) {
853        (
854            GidMappingKey {
855                schema_name: self.description.schema_name,
856                object_type: self.description.object_type,
857                object_name: self.description.object_name,
858            },
859            GidMappingValue {
860                catalog_id: self
861                    .unique_identifier
862                    .catalog_id
863                    .try_into()
864                    .expect("catalog_id to be in the system namespace"),
865                global_id: self
866                    .unique_identifier
867                    .global_id
868                    .try_into()
869                    .expect("collection_id to be in the system namespace"),
870                fingerprint: self.unique_identifier.fingerprint,
871            },
872        )
873    }
874
875    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
876        Self {
877            description: SystemObjectDescription {
878                schema_name: key.schema_name,
879                object_type: key.object_type,
880                object_name: key.object_name,
881            },
882            unique_identifier: SystemObjectUniqueIdentifier {
883                catalog_id: value.catalog_id.into(),
884                global_id: value.global_id.into(),
885                fingerprint: value.fingerprint,
886            },
887        }
888    }
889
890    fn key(&self) -> Self::Key {
891        GidMappingKey {
892            schema_name: self.description.schema_name.clone(),
893            object_type: self.description.object_type.clone(),
894            object_name: self.description.object_name.clone(),
895        }
896    }
897}
898
899#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
900pub struct DefaultPrivilege {
901    pub object: DefaultPrivilegeObject,
902    pub acl_item: DefaultPrivilegeAclItem,
903}
904
905impl DurableType for DefaultPrivilege {
906    type Key = DefaultPrivilegesKey;
907    type Value = DefaultPrivilegesValue;
908
909    fn into_key_value(self) -> (Self::Key, Self::Value) {
910        (
911            DefaultPrivilegesKey {
912                role_id: self.object.role_id,
913                database_id: self.object.database_id,
914                schema_id: self.object.schema_id,
915                object_type: self.object.object_type,
916                grantee: self.acl_item.grantee,
917            },
918            DefaultPrivilegesValue {
919                privileges: self.acl_item.acl_mode,
920            },
921        )
922    }
923
924    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
925        Self {
926            object: DefaultPrivilegeObject {
927                role_id: key.role_id,
928                database_id: key.database_id,
929                schema_id: key.schema_id,
930                object_type: key.object_type,
931            },
932            acl_item: DefaultPrivilegeAclItem {
933                grantee: key.grantee,
934                acl_mode: value.privileges,
935            },
936        }
937    }
938
939    fn key(&self) -> Self::Key {
940        DefaultPrivilegesKey {
941            role_id: self.object.role_id,
942            database_id: self.object.database_id,
943            schema_id: self.object.schema_id,
944            object_type: self.object.object_type,
945            grantee: self.acl_item.grantee,
946        }
947    }
948}
949
950#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
951pub struct Comment {
952    pub object_id: CommentObjectId,
953    pub sub_component: Option<usize>,
954    pub comment: String,
955}
956
957impl DurableType for Comment {
958    type Key = CommentKey;
959    type Value = CommentValue;
960
961    fn into_key_value(self) -> (Self::Key, Self::Value) {
962        (
963            CommentKey {
964                object_id: self.object_id,
965                sub_component: self.sub_component,
966            },
967            CommentValue {
968                comment: self.comment,
969            },
970        )
971    }
972
973    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
974        Self {
975            object_id: key.object_id,
976            sub_component: key.sub_component,
977            comment: value.comment,
978        }
979    }
980
981    fn key(&self) -> Self::Key {
982        CommentKey {
983            object_id: self.object_id,
984            sub_component: self.sub_component,
985        }
986    }
987}
988
989#[derive(Debug, Clone, PartialEq, Eq)]
990pub struct IdAlloc {
991    pub name: String,
992    pub next_id: u64,
993}
994
995impl DurableType for IdAlloc {
996    type Key = IdAllocKey;
997    type Value = IdAllocValue;
998
999    fn into_key_value(self) -> (Self::Key, Self::Value) {
1000        (
1001            IdAllocKey { name: self.name },
1002            IdAllocValue {
1003                next_id: self.next_id,
1004            },
1005        )
1006    }
1007
1008    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
1009        Self {
1010            name: key.name,
1011            next_id: value.next_id,
1012        }
1013    }
1014
1015    fn key(&self) -> Self::Key {
1016        IdAllocKey {
1017            name: self.name.clone(),
1018        }
1019    }
1020}
1021
1022#[derive(Debug, Clone, PartialEq, Eq)]
1023pub struct Config {
1024    pub key: String,
1025    pub value: u64,
1026}
1027
1028impl DurableType for Config {
1029    type Key = ConfigKey;
1030    type Value = ConfigValue;
1031
1032    fn into_key_value(self) -> (Self::Key, Self::Value) {
1033        (
1034            ConfigKey { key: self.key },
1035            ConfigValue { value: self.value },
1036        )
1037    }
1038
1039    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
1040        Self {
1041            key: key.key,
1042            value: value.value,
1043        }
1044    }
1045
1046    fn key(&self) -> Self::Key {
1047        ConfigKey {
1048            key: self.key.clone(),
1049        }
1050    }
1051}
1052
1053#[derive(Debug, Clone)]
1054pub struct Setting {
1055    pub name: String,
1056    pub value: String,
1057}
1058
1059impl DurableType for Setting {
1060    type Key = SettingKey;
1061    type Value = SettingValue;
1062
1063    fn into_key_value(self) -> (Self::Key, Self::Value) {
1064        (
1065            SettingKey { name: self.name },
1066            SettingValue { value: self.value },
1067        )
1068    }
1069
1070    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
1071        Self {
1072            name: key.name,
1073            value: value.value,
1074        }
1075    }
1076
1077    fn key(&self) -> Self::Key {
1078        SettingKey {
1079            name: self.name.clone(),
1080        }
1081    }
1082}
1083
1084#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
1085pub struct SystemConfiguration {
1086    pub name: String,
1087    pub value: String,
1088}
1089
1090impl DurableType for SystemConfiguration {
1091    type Key = ServerConfigurationKey;
1092    type Value = ServerConfigurationValue;
1093
1094    fn into_key_value(self) -> (Self::Key, Self::Value) {
1095        (
1096            ServerConfigurationKey { name: self.name },
1097            ServerConfigurationValue { value: self.value },
1098        )
1099    }
1100
1101    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
1102        Self {
1103            name: key.name,
1104            value: value.value,
1105        }
1106    }
1107
1108    fn key(&self) -> Self::Key {
1109        ServerConfigurationKey {
1110            name: self.name.clone(),
1111        }
1112    }
1113}
1114
1115/// A single cluster-coherent scoped system-parameter override: parameter `name`
1116/// has value `value` on the cluster `cluster_id`.
1117///
1118/// This is the in-memory shape of the durable `cluster_system_configurations`
1119/// collection that backs cluster-coherent scoped feature flags. The collection
1120/// — keyed by `(ClusterId, name)` — is the analog of `system_configurations`
1121/// (`ALTER SYSTEM`), but for per-cluster values; it is written solely by the
1122/// system-parameter sync loop, and the coordinator's in-memory working copy is
1123/// maintained from it on every catalog update.
1124#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
1125pub struct ClusterSystemConfiguration {
1126    pub cluster_id: ClusterId,
1127    pub name: String,
1128    pub value: String,
1129}
1130
1131impl DurableType for ClusterSystemConfiguration {
1132    type Key = ClusterSystemConfigurationKey;
1133    type Value = ClusterSystemConfigurationValue;
1134
1135    fn into_key_value(self) -> (Self::Key, Self::Value) {
1136        (
1137            ClusterSystemConfigurationKey {
1138                cluster_id: self.cluster_id,
1139                name: self.name,
1140            },
1141            ClusterSystemConfigurationValue { value: self.value },
1142        )
1143    }
1144
1145    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
1146        Self {
1147            cluster_id: key.cluster_id,
1148            name: key.name,
1149            value: value.value,
1150        }
1151    }
1152
1153    fn key(&self) -> Self::Key {
1154        ClusterSystemConfigurationKey {
1155            cluster_id: self.cluster_id,
1156            name: self.name.clone(),
1157        }
1158    }
1159}
1160
1161/// A single replica-local scoped system-parameter override: parameter `name`
1162/// has value `value` on the replica `replica_id`.
1163///
1164/// This is the in-memory shape of the durable `replica_system_configurations`
1165/// collection that backs replica-local scoped feature flags. The collection —
1166/// keyed by `(ReplicaId, name)` — is the analog of `system_configurations`
1167/// (`ALTER SYSTEM`), but for per-replica values; it is written solely by the
1168/// system-parameter sync loop, and the coordinator's in-memory working copy is
1169/// maintained from it on every catalog update.
1170#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
1171pub struct ReplicaSystemConfiguration {
1172    pub replica_id: ReplicaId,
1173    pub name: String,
1174    pub value: String,
1175}
1176
1177impl DurableType for ReplicaSystemConfiguration {
1178    type Key = ReplicaSystemConfigurationKey;
1179    type Value = ReplicaSystemConfigurationValue;
1180
1181    fn into_key_value(self) -> (Self::Key, Self::Value) {
1182        (
1183            ReplicaSystemConfigurationKey {
1184                replica_id: self.replica_id,
1185                name: self.name,
1186            },
1187            ReplicaSystemConfigurationValue { value: self.value },
1188        )
1189    }
1190
1191    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
1192        Self {
1193            replica_id: key.replica_id,
1194            name: key.name,
1195            value: value.value,
1196        }
1197    }
1198
1199    fn key(&self) -> Self::Key {
1200        ReplicaSystemConfigurationKey {
1201            replica_id: self.replica_id,
1202            name: self.name.clone(),
1203        }
1204    }
1205}
1206
1207impl DurableType for MzAclItem {
1208    type Key = SystemPrivilegesKey;
1209    type Value = SystemPrivilegesValue;
1210
1211    fn into_key_value(self) -> (Self::Key, Self::Value) {
1212        (
1213            SystemPrivilegesKey {
1214                grantee: self.grantee,
1215                grantor: self.grantor,
1216            },
1217            SystemPrivilegesValue {
1218                acl_mode: self.acl_mode,
1219            },
1220        )
1221    }
1222
1223    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
1224        Self {
1225            grantee: key.grantee,
1226            grantor: key.grantor,
1227            acl_mode: value.acl_mode,
1228        }
1229    }
1230
1231    fn key(&self) -> Self::Key {
1232        SystemPrivilegesKey {
1233            grantee: self.grantee,
1234            grantor: self.grantor,
1235        }
1236    }
1237}
1238
1239#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
1240pub struct AuditLog {
1241    pub event: VersionedEvent,
1242}
1243
1244impl DurableType for AuditLog {
1245    type Key = AuditLogKey;
1246    type Value = ();
1247
1248    fn into_key_value(self) -> (Self::Key, Self::Value) {
1249        (AuditLogKey { event: self.event }, ())
1250    }
1251
1252    fn from_key_value(key: Self::Key, _value: Self::Value) -> Self {
1253        Self { event: key.event }
1254    }
1255
1256    fn key(&self) -> Self::Key {
1257        AuditLogKey {
1258            event: self.event.clone(),
1259        }
1260    }
1261}
1262
1263#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
1264pub struct StorageCollectionMetadata {
1265    pub id: GlobalId,
1266    pub shard: ShardId,
1267}
1268
1269impl DurableType for StorageCollectionMetadata {
1270    type Key = StorageCollectionMetadataKey;
1271    type Value = StorageCollectionMetadataValue;
1272
1273    fn into_key_value(self) -> (Self::Key, Self::Value) {
1274        (
1275            StorageCollectionMetadataKey { id: self.id },
1276            StorageCollectionMetadataValue { shard: self.shard },
1277        )
1278    }
1279
1280    fn from_key_value(key: Self::Key, value: Self::Value) -> Self {
1281        Self {
1282            id: key.id,
1283            shard: value.shard,
1284        }
1285    }
1286
1287    fn key(&self) -> Self::Key {
1288        StorageCollectionMetadataKey { id: self.id }
1289    }
1290}
1291
1292#[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq)]
1293pub struct UnfinalizedShard {
1294    pub shard: ShardId,
1295}
1296
1297impl DurableType for UnfinalizedShard {
1298    type Key = UnfinalizedShardKey;
1299    type Value = ();
1300
1301    fn into_key_value(self) -> (Self::Key, Self::Value) {
1302        (UnfinalizedShardKey { shard: self.shard }, ())
1303    }
1304
1305    fn from_key_value(key: Self::Key, _value: Self::Value) -> Self {
1306        Self { shard: key.shard }
1307    }
1308
1309    fn key(&self) -> Self::Key {
1310        UnfinalizedShardKey {
1311            shard: self.shard.clone(),
1312        }
1313    }
1314}
1315
1316// Structs used internally to represent on-disk state.
1317
1318/// A snapshot of the current on-disk state.
1319#[derive(Debug, Clone, PartialEq, Eq, Default)]
1320pub struct Snapshot {
1321    pub databases: BTreeMap<proto::DatabaseKey, proto::DatabaseValue>,
1322    pub schemas: BTreeMap<proto::SchemaKey, proto::SchemaValue>,
1323    pub roles: BTreeMap<proto::RoleKey, proto::RoleValue>,
1324    pub role_auth: BTreeMap<proto::RoleAuthKey, proto::RoleAuthValue>,
1325    pub items: BTreeMap<proto::ItemKey, proto::ItemValue>,
1326    pub comments: BTreeMap<proto::CommentKey, proto::CommentValue>,
1327    pub clusters: BTreeMap<proto::ClusterKey, proto::ClusterValue>,
1328    pub network_policies: BTreeMap<proto::NetworkPolicyKey, proto::NetworkPolicyValue>,
1329    pub cluster_replicas: BTreeMap<proto::ClusterReplicaKey, proto::ClusterReplicaValue>,
1330    pub introspection_sources: BTreeMap<
1331        proto::ClusterIntrospectionSourceIndexKey,
1332        proto::ClusterIntrospectionSourceIndexValue,
1333    >,
1334    pub id_allocator: BTreeMap<proto::IdAllocKey, proto::IdAllocValue>,
1335    pub configs: BTreeMap<proto::ConfigKey, proto::ConfigValue>,
1336    pub settings: BTreeMap<proto::SettingKey, proto::SettingValue>,
1337    pub system_object_mappings: BTreeMap<proto::GidMappingKey, proto::GidMappingValue>,
1338    pub system_configurations:
1339        BTreeMap<proto::ServerConfigurationKey, proto::ServerConfigurationValue>,
1340    pub cluster_system_configurations:
1341        BTreeMap<proto::ClusterSystemConfigurationKey, proto::ClusterSystemConfigurationValue>,
1342    pub replica_system_configurations:
1343        BTreeMap<proto::ReplicaSystemConfigurationKey, proto::ReplicaSystemConfigurationValue>,
1344    pub default_privileges: BTreeMap<proto::DefaultPrivilegesKey, proto::DefaultPrivilegesValue>,
1345    pub source_references: BTreeMap<proto::SourceReferencesKey, proto::SourceReferencesValue>,
1346    pub system_privileges: BTreeMap<proto::SystemPrivilegesKey, proto::SystemPrivilegesValue>,
1347    pub storage_collection_metadata:
1348        BTreeMap<proto::StorageCollectionMetadataKey, proto::StorageCollectionMetadataValue>,
1349    pub unfinalized_shards: BTreeMap<proto::UnfinalizedShardKey, ()>,
1350    pub txn_wal_shard: BTreeMap<(), proto::TxnWalShardValue>,
1351}
1352
1353impl Snapshot {
1354    pub fn empty() -> Snapshot {
1355        Snapshot::default()
1356    }
1357}
1358
1359/// Token used to fence out other processes.
1360///
1361/// Every time a new process takes over, the `epoch` should be incremented.
1362/// Every time a new version is deployed, the `deploy` generation should be incremented.
1363#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1364#[cfg_attr(test, derive(Arbitrary))]
1365pub struct FenceToken {
1366    pub(crate) deploy_generation: u64,
1367    pub(crate) epoch: Epoch,
1368}
1369
1370impl PartialOrd for FenceToken {
1371    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
1372        Some(self.cmp(other))
1373    }
1374}
1375
1376impl Ord for FenceToken {
1377    fn cmp(&self, other: &Self) -> Ordering {
1378        self.deploy_generation
1379            .cmp(&other.deploy_generation)
1380            .then(self.epoch.cmp(&other.epoch))
1381    }
1382}
1383
1384#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1385pub struct SettingKey {
1386    pub(crate) name: String,
1387}
1388
1389#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord)]
1390pub struct SettingValue {
1391    pub(crate) value: String,
1392}
1393
1394#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1395pub struct IdAllocKey {
1396    pub(crate) name: String,
1397}
1398
1399#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord)]
1400pub struct IdAllocValue {
1401    pub(crate) next_id: u64,
1402}
1403
1404#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1405pub struct GidMappingKey {
1406    pub(crate) schema_name: String,
1407    pub(crate) object_type: CatalogItemType,
1408    pub(crate) object_name: String,
1409}
1410
1411#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord)]
1412pub struct GidMappingValue {
1413    pub(crate) catalog_id: SystemCatalogItemId,
1414    pub(crate) global_id: SystemGlobalId,
1415    pub(crate) fingerprint: String,
1416}
1417
1418#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1419pub struct ClusterKey {
1420    pub(crate) id: ClusterId,
1421}
1422
1423#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord)]
1424pub struct ClusterValue {
1425    pub(crate) name: String,
1426    pub(crate) owner_id: RoleId,
1427    pub(crate) privileges: Vec<MzAclItem>,
1428    pub(crate) config: ClusterConfig,
1429}
1430
1431#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1432pub struct ClusterIntrospectionSourceIndexKey {
1433    pub(crate) cluster_id: ClusterId,
1434    pub(crate) name: String,
1435}
1436
1437#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord)]
1438pub struct ClusterIntrospectionSourceIndexValue {
1439    pub(crate) catalog_id: IntrospectionSourceIndexCatalogItemId,
1440    pub(crate) global_id: IntrospectionSourceIndexGlobalId,
1441    pub(crate) oid: u32,
1442}
1443
1444#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1445pub struct ClusterReplicaKey {
1446    pub(crate) id: ReplicaId,
1447}
1448
1449#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord)]
1450pub struct ClusterReplicaValue {
1451    pub(crate) cluster_id: ClusterId,
1452    pub(crate) name: String,
1453    pub(crate) config: ReplicaConfig,
1454    pub(crate) owner_id: RoleId,
1455}
1456
1457#[derive(Clone, Copy, Debug, PartialOrd, PartialEq, Eq, Ord, Hash)]
1458#[cfg_attr(test, derive(Arbitrary))]
1459pub struct DatabaseKey {
1460    pub(crate) id: DatabaseId,
1461}
1462
1463#[derive(Clone, Debug, PartialOrd, PartialEq, Eq, Ord)]
1464#[cfg_attr(test, derive(Arbitrary))]
1465pub struct DatabaseValue {
1466    pub(crate) name: String,
1467    pub(crate) owner_id: RoleId,
1468    pub(crate) privileges: Vec<MzAclItem>,
1469    pub(crate) oid: u32,
1470}
1471
1472#[derive(Clone, Copy, Debug, PartialOrd, PartialEq, Eq, Ord, Hash)]
1473#[cfg_attr(test, derive(Arbitrary))]
1474pub struct SourceReferencesKey {
1475    pub(crate) source_id: CatalogItemId,
1476}
1477
1478#[derive(Clone, Debug, PartialOrd, PartialEq, Eq, Ord)]
1479#[cfg_attr(test, derive(Arbitrary))]
1480pub struct SourceReferencesValue {
1481    pub(crate) references: Vec<SourceReference>,
1482    pub(crate) updated_at: u64,
1483}
1484
1485#[derive(Clone, Copy, Debug, PartialOrd, PartialEq, Eq, Ord, Hash)]
1486#[cfg_attr(test, derive(Arbitrary))]
1487pub struct SchemaKey {
1488    pub(crate) id: SchemaId,
1489}
1490
1491#[derive(Clone, Debug, PartialOrd, PartialEq, Eq, Ord)]
1492#[cfg_attr(test, derive(Arbitrary))]
1493pub struct SchemaValue {
1494    pub(crate) database_id: Option<DatabaseId>,
1495    pub(crate) name: String,
1496    pub(crate) owner_id: RoleId,
1497    pub(crate) privileges: Vec<MzAclItem>,
1498    pub(crate) oid: u32,
1499}
1500
1501#[derive(Clone, PartialOrd, PartialEq, Eq, Ord, Hash, Debug)]
1502#[cfg_attr(test, derive(Arbitrary))]
1503pub struct ItemKey {
1504    pub(crate) id: CatalogItemId,
1505}
1506
1507#[derive(Clone, Debug, PartialOrd, PartialEq, Eq, Ord)]
1508#[cfg_attr(test, derive(Arbitrary))]
1509pub struct ItemValue {
1510    pub(crate) schema_id: SchemaId,
1511    pub(crate) name: String,
1512    pub(crate) create_sql: String,
1513    pub(crate) owner_id: RoleId,
1514    pub(crate) privileges: Vec<MzAclItem>,
1515    pub(crate) oid: u32,
1516    pub(crate) global_id: GlobalId,
1517    pub(crate) extra_versions: BTreeMap<RelationVersion, GlobalId>,
1518}
1519
1520impl ItemValue {
1521    pub fn item_type(&self) -> CatalogItemType {
1522        item_type(&self.create_sql)
1523    }
1524}
1525
1526pub fn item_type(create_sql: &str) -> CatalogItemType {
1527    // NOTE(benesch): the implementation of this method is hideous, but is
1528    // there a better alternative? Storing the object type alongside the
1529    // `create_sql` would introduce the possibility of skew.
1530    let mut tokens = create_sql.split_whitespace();
1531    assert_eq!(tokens.next(), Some("CREATE"));
1532
1533    // Read away item type modifiers, if any.
1534    let next_token = match tokens.next() {
1535        Some("TEMPORARY") | Some("REPLACEMENT") => tokens.next(),
1536        token => token,
1537    };
1538
1539    match next_token {
1540        Some("TABLE") => CatalogItemType::Table,
1541        Some("SOURCE") | Some("SUBSOURCE") => CatalogItemType::Source,
1542        Some("SINK") => CatalogItemType::Sink,
1543        Some("VIEW") => CatalogItemType::View,
1544        Some("MATERIALIZED") => {
1545            assert_eq!(tokens.next(), Some("VIEW"));
1546            CatalogItemType::MaterializedView
1547        }
1548        Some("INDEX") => CatalogItemType::Index,
1549        Some("TYPE") => CatalogItemType::Type,
1550        Some("FUNCTION") => CatalogItemType::Func,
1551        Some("SECRET") => CatalogItemType::Secret,
1552        Some("CONNECTION") => CatalogItemType::Connection,
1553        _ => panic!("unexpected create sql: {}", create_sql),
1554    }
1555}
1556
1557#[derive(Clone, Debug, PartialOrd, PartialEq, Eq, Ord)]
1558pub struct CommentKey {
1559    pub(crate) object_id: CommentObjectId,
1560    pub(crate) sub_component: Option<usize>,
1561}
1562
1563#[derive(Clone, Debug, PartialOrd, PartialEq, Eq, Ord)]
1564#[cfg_attr(test, derive(Arbitrary))]
1565pub struct CommentValue {
1566    pub(crate) comment: String,
1567}
1568
1569#[derive(Clone, PartialOrd, PartialEq, Eq, Ord, Hash, Debug)]
1570pub struct RoleKey {
1571    pub(crate) id: RoleId,
1572}
1573
1574#[derive(Clone, PartialOrd, PartialEq, Eq, Ord, Debug)]
1575pub struct RoleValue {
1576    pub(crate) name: String,
1577    pub(crate) attributes: RoleAttributes,
1578    pub(crate) membership: RoleMembership,
1579    pub(crate) vars: RoleVars,
1580    pub(crate) oid: u32,
1581}
1582
1583#[derive(Clone, PartialOrd, PartialEq, Eq, Ord, Hash, Debug)]
1584pub struct NetworkPolicyKey {
1585    pub(crate) id: NetworkPolicyId,
1586}
1587
1588#[derive(Clone, PartialOrd, PartialEq, Eq, Ord, Debug)]
1589pub struct NetworkPolicyValue {
1590    pub(crate) name: String,
1591    pub(crate) rules: Vec<NetworkPolicyRule>,
1592    pub(crate) owner_id: RoleId,
1593    pub(crate) privileges: Vec<MzAclItem>,
1594    pub(crate) oid: u32,
1595}
1596
1597#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord)]
1598pub struct ConfigKey {
1599    pub(crate) key: String,
1600}
1601
1602#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1603pub struct ConfigValue {
1604    pub(crate) value: u64,
1605}
1606
1607#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1608pub struct AuditLogKey {
1609    pub(crate) event: VersionedEvent,
1610}
1611
1612#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1613pub struct StorageCollectionMetadataKey {
1614    pub(crate) id: GlobalId,
1615}
1616
1617/// This value is stored transparently, however, it should only ever be
1618/// manipulated by the storage controller.
1619#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord)]
1620pub struct StorageCollectionMetadataValue {
1621    pub(crate) shard: ShardId,
1622}
1623
1624/// This value is stored transparently, however, it should only ever be
1625/// manipulated by the storage controller.
1626#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord)]
1627pub struct UnfinalizedShardKey {
1628    pub(crate) shard: ShardId,
1629}
1630
1631/// This value is stored transparently, however, it should only ever be
1632/// manipulated by the storage controller.
1633#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord)]
1634pub struct TxnWalShardValue {
1635    pub(crate) shard: ShardId,
1636}
1637
1638#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1639pub struct ServerConfigurationKey {
1640    pub(crate) name: String,
1641}
1642
1643#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord)]
1644pub struct ServerConfigurationValue {
1645    pub(crate) value: String,
1646}
1647
1648#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1649pub struct ClusterSystemConfigurationKey {
1650    pub(crate) cluster_id: ClusterId,
1651    pub(crate) name: String,
1652}
1653
1654#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord)]
1655pub struct ClusterSystemConfigurationValue {
1656    pub(crate) value: String,
1657}
1658
1659#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1660pub struct ReplicaSystemConfigurationKey {
1661    pub(crate) replica_id: ReplicaId,
1662    pub(crate) name: String,
1663}
1664
1665#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord)]
1666pub struct ReplicaSystemConfigurationValue {
1667    pub(crate) value: String,
1668}
1669
1670#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1671pub struct DefaultPrivilegesKey {
1672    pub(crate) role_id: RoleId,
1673    pub(crate) database_id: Option<DatabaseId>,
1674    pub(crate) schema_id: Option<SchemaId>,
1675    pub(crate) object_type: ObjectType,
1676    pub(crate) grantee: RoleId,
1677}
1678
1679#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1680pub struct DefaultPrivilegesValue {
1681    pub(crate) privileges: AclMode,
1682}
1683
1684#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1685pub struct SystemPrivilegesKey {
1686    pub(crate) grantee: RoleId,
1687    pub(crate) grantor: RoleId,
1688}
1689
1690#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1691pub struct SystemPrivilegesValue {
1692    pub(crate) acl_mode: AclMode,
1693}
1694
1695#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1696pub struct RoleAuthKey {
1697    // TODO(auth): Depending on what the future holds, here is where
1698    // we might also want to key by a `version` field.
1699    // That way we can store password versions or what have you.
1700    pub(crate) role_id: RoleId,
1701}
1702
1703#[derive(Debug, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
1704pub struct RoleAuthValue {
1705    pub(crate) password_hash: Option<String>,
1706    pub(crate) updated_at: u64,
1707}
1708
1709#[cfg(test)]
1710mod test {
1711    use mz_proto::{ProtoType, RustType};
1712    use proptest::prelude::*;
1713
1714    use super::{
1715        DatabaseKey, DatabaseValue, FenceToken, ItemKey, ItemValue, SchemaKey, SchemaValue,
1716    };
1717    use crate::durable::Epoch;
1718
1719    proptest! {
1720        #[mz_ore::test]
1721        #[cfg_attr(miri, ignore)] // slow
1722        fn proptest_database_key_roundtrip(key: DatabaseKey) {
1723            let proto = key.into_proto();
1724            let round = proto.into_rust().expect("to roundtrip");
1725
1726            prop_assert_eq!(key, round);
1727        }
1728
1729        #[mz_ore::test]
1730        #[cfg_attr(miri, ignore)] // slow
1731        fn proptest_database_value_roundtrip(value: DatabaseValue) {
1732            let proto = value.into_proto();
1733            let round = proto.into_rust().expect("to roundtrip");
1734
1735            prop_assert_eq!(value, round);
1736        }
1737
1738        #[mz_ore::test]
1739        #[cfg_attr(miri, ignore)] // slow
1740        fn proptest_schema_key_roundtrip(key: SchemaKey) {
1741            let proto = key.into_proto();
1742            let round = proto.into_rust().expect("to roundtrip");
1743
1744            prop_assert_eq!(key, round);
1745        }
1746
1747        #[mz_ore::test]
1748        #[cfg_attr(miri, ignore)] // slow
1749        fn proptest_schema_value_roundtrip(value: SchemaValue) {
1750            let proto = value.into_proto();
1751            let round = proto.into_rust().expect("to roundtrip");
1752
1753            prop_assert_eq!(value, round);
1754        }
1755
1756        #[mz_ore::test]
1757        #[cfg_attr(miri, ignore)] // slow
1758        fn proptest_item_key_roundtrip(key: ItemKey) {
1759            let proto = key.into_proto();
1760            let round = proto.into_rust().expect("to roundtrip");
1761
1762            prop_assert_eq!(key, round);
1763        }
1764
1765        #[mz_ore::test]
1766        #[cfg_attr(miri, ignore)] // slow
1767        fn proptest_item_value_roundtrip(value: ItemValue) {
1768            let proto = value.into_proto();
1769            let round = proto.into_rust().expect("to roundtrip");
1770
1771            prop_assert_eq!(value, round);
1772        }
1773    }
1774
1775    #[mz_ore::test]
1776    fn test_fence_token_order() {
1777        let ft1 = FenceToken {
1778            deploy_generation: 10,
1779            epoch: Epoch::new(20).expect("non-zero"),
1780        };
1781        let ft2 = FenceToken {
1782            deploy_generation: 10,
1783            epoch: Epoch::new(19).expect("non-zero"),
1784        };
1785
1786        assert!(ft1 > ft2);
1787
1788        let ft3 = FenceToken {
1789            deploy_generation: 11,
1790            epoch: Epoch::new(10).expect("non-zero"),
1791        };
1792
1793        assert!(ft3 > ft1);
1794
1795        let ft4 = FenceToken {
1796            deploy_generation: 11,
1797            epoch: Epoch::new(30).expect("non-zero"),
1798        };
1799
1800        assert!(ft4 > ft1);
1801    }
1802}
mz_catalog/durable/objects.rs

mz_catalog/durable/
objects.rs
