mz_catalog/durable/
objects.rs

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

mz_catalog/durable/
objects.rs
