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

mz_catalog/durable/
objects.rs
