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

mz_catalog/durable/
objects.rs
