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

mz_catalog/durable/
objects.rs
