mz_catalog/durable/objects/
state_update.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//! This module contains various representations of a single catalog update and the logic necessary
11//! for converting between representations.
12//!
13//! The general lifecycle of a single update when read from persist is as follows:
14//!
15//!   1. The update is stored in persist as a [`PersistStateUpdate`].
16//!   2. After being read from persist the update is immediately converted into a
17//!      [`StateUpdate<StateUpdateKindJson>`], which models the update as a JSON.
18//!   3. The [`StateUpdateKindJson`] is converted into a protobuf message,
19//!      [`proto::StateUpdateKind`].
20//!   4. The update is then converted into a [`StateUpdate<StateUpdateKind>`], which is a strongly
21//!      typed Rust object.
22//!   5. Finally, the update is converted into an [`Option<memory::objects::StateUpdate>`], and
23//!      `Some` variants are given to the in-memory catalog. The in-memory catalog is only
24//!      interested in a subset of catalog updates which is why the [`Option`] is necessary.
25//!
26//! TLDR: [`PersistStateUpdate`] -> [`StateUpdate<StateUpdateKindJson>`] ->
27//!       [`proto::StateUpdateKind`] -> [`StateUpdate<StateUpdateKind>`] ->
28//!       [`Option<memory::objects::StateUpdate>`]
29//!
30//! The process of writing a catalog update to persist is the exact opposite.
31//!
32//! When running catalog protobuf upgrades/migrations we may need to take a detour and convert the
33//! [`StateUpdateKindJson`] to some `proto::object_v{x}::StateUpdateKind` before applying specific
34//! upgrades to get us to a valid [`proto::StateUpdateKind`].
35
36use std::fmt::Debug;
37use std::sync::LazyLock;
38
39use mz_ore::collections::HashSet;
40use mz_proto::{ProtoType, RustType, TryFromProtoError};
41use mz_repr::Diff;
42use mz_repr::adt::jsonb::Jsonb;
43use mz_storage_types::StorageDiff;
44use mz_storage_types::sources::SourceData;
45#[cfg(test)]
46use proptest_derive::Arbitrary;
47use tracing::error;
48
49use crate::durable::debug::CollectionType;
50use crate::durable::objects::serialization::proto;
51use crate::durable::objects::{DurableType, FenceToken};
52use crate::durable::persist::Timestamp;
53use crate::durable::transaction::TransactionBatch;
54use crate::durable::{DurableCatalogError, Epoch};
55use crate::memory;
56
57/// Trait for objects that can be converted to/from a [`StateUpdateKindJson`].
58pub trait IntoStateUpdateKindJson:
59    Into<StateUpdateKindJson> + PartialEq + Eq + PartialOrd + Ord + Debug + Clone
60{
61    type Error: Debug;
62
63    fn try_from(raw: StateUpdateKindJson) -> Result<Self, Self::Error>;
64}
65impl<
66    T: Into<StateUpdateKindJson>
67        + TryFrom<StateUpdateKindJson>
68        + PartialEq
69        + Eq
70        + PartialOrd
71        + Ord
72        + Debug
73        + Clone,
74> IntoStateUpdateKindJson for T
75where
76    T::Error: Debug,
77{
78    type Error = T::Error;
79
80    fn try_from(raw: StateUpdateKindJson) -> Result<Self, Self::Error> {
81        <T as TryFrom<StateUpdateKindJson>>::try_from(raw)
82    }
83}
84
85/// Trait for objects that can be converted to/from a [`StateUpdateKind`].
86pub(crate) trait TryIntoStateUpdateKind: IntoStateUpdateKindJson {
87    type Error: Debug;
88
89    fn try_into(self) -> Result<StateUpdateKind, <Self as TryIntoStateUpdateKind>::Error>;
90}
91impl<T: IntoStateUpdateKindJson + TryInto<StateUpdateKind>> TryIntoStateUpdateKind for T
92where
93    <T as TryInto<StateUpdateKind>>::Error: Debug,
94{
95    type Error = <T as TryInto<StateUpdateKind>>::Error;
96
97    fn try_into(self) -> Result<StateUpdateKind, <T as TryInto<StateUpdateKind>>::Error> {
98        <T as TryInto<StateUpdateKind>>::try_into(self)
99    }
100}
101
102/// A single update to the catalog state.
103#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
104pub struct StateUpdate<T: IntoStateUpdateKindJson = StateUpdateKind> {
105    /// They kind and contents of the state update.
106    pub kind: T,
107    /// The timestamp at which the update occurred.
108    pub ts: Timestamp,
109    /// Record count difference for the update.
110    pub diff: Diff,
111}
112
113impl StateUpdate {
114    /// Convert a [`TransactionBatch`] to a list of [`StateUpdate`]s at timestamp `ts`.
115    pub(crate) fn from_txn_batch_ts(
116        txn_batch: TransactionBatch,
117        ts: Timestamp,
118    ) -> impl Iterator<Item = StateUpdate> {
119        Self::from_txn_batch(txn_batch).map(move |(kind, diff)| StateUpdate { kind, ts, diff })
120    }
121
122    /// Convert a [`TransactionBatch`] to a list of [`StateUpdate`]s and [`Diff`]s.
123    pub(crate) fn from_txn_batch(
124        txn_batch: TransactionBatch,
125    ) -> impl Iterator<Item = (StateUpdateKind, Diff)> {
126        fn from_batch<K, V>(
127            batch: Vec<(K, V, Diff)>,
128            kind: fn(K, V) -> StateUpdateKind,
129        ) -> impl Iterator<Item = (StateUpdateKind, Diff)> {
130            batch
131                .into_iter()
132                .map(move |(k, v, diff)| (kind(k, v), diff))
133        }
134        let TransactionBatch {
135            databases,
136            schemas,
137            items,
138            comments,
139            roles,
140            role_auth,
141            clusters,
142            cluster_replicas,
143            network_policies,
144            introspection_sources,
145            id_allocator,
146            configs,
147            settings,
148            source_references,
149            system_gid_mapping,
150            system_configurations,
151            cluster_system_configurations,
152            replica_system_configurations,
153            default_privileges,
154            system_privileges,
155            storage_collection_metadata,
156            unfinalized_shards,
157            txn_wal_shard,
158            audit_log_updates,
159            upper: _,
160        } = txn_batch;
161        let databases = from_batch(databases, StateUpdateKind::Database);
162        let schemas = from_batch(schemas, StateUpdateKind::Schema);
163        let items = from_batch(items, StateUpdateKind::Item);
164        let comments = from_batch(comments, StateUpdateKind::Comment);
165        let roles = from_batch(roles, StateUpdateKind::Role);
166        let role_auth = from_batch(role_auth, StateUpdateKind::RoleAuth);
167        let clusters = from_batch(clusters, StateUpdateKind::Cluster);
168        let cluster_replicas = from_batch(cluster_replicas, StateUpdateKind::ClusterReplica);
169        let network_policies = from_batch(network_policies, StateUpdateKind::NetworkPolicy);
170        let introspection_sources = from_batch(
171            introspection_sources,
172            StateUpdateKind::IntrospectionSourceIndex,
173        );
174        let id_allocators = from_batch(id_allocator, StateUpdateKind::IdAllocator);
175        let configs = from_batch(configs, StateUpdateKind::Config);
176        let settings = from_batch(settings, StateUpdateKind::Setting);
177        let system_object_mappings =
178            from_batch(system_gid_mapping, StateUpdateKind::SystemObjectMapping);
179        let system_configurations =
180            from_batch(system_configurations, StateUpdateKind::SystemConfiguration);
181        let cluster_system_configurations = from_batch(
182            cluster_system_configurations,
183            StateUpdateKind::ClusterSystemConfiguration,
184        );
185        let replica_system_configurations = from_batch(
186            replica_system_configurations,
187            StateUpdateKind::ReplicaSystemConfiguration,
188        );
189        let default_privileges = from_batch(default_privileges, StateUpdateKind::DefaultPrivilege);
190        let source_references = from_batch(source_references, StateUpdateKind::SourceReferences);
191        let system_privileges = from_batch(system_privileges, StateUpdateKind::SystemPrivilege);
192        let storage_collection_metadata = from_batch(
193            storage_collection_metadata,
194            StateUpdateKind::StorageCollectionMetadata,
195        );
196        let unfinalized_shards = from_batch(unfinalized_shards, StateUpdateKind::UnfinalizedShard);
197        let txn_wal_shard = from_batch(txn_wal_shard, StateUpdateKind::TxnWalShard);
198        let audit_logs = from_batch(audit_log_updates, StateUpdateKind::AuditLog);
199
200        databases
201            .chain(schemas)
202            .chain(items)
203            .chain(comments)
204            .chain(roles)
205            .chain(role_auth)
206            .chain(clusters)
207            .chain(cluster_replicas)
208            .chain(network_policies)
209            .chain(introspection_sources)
210            .chain(id_allocators)
211            .chain(configs)
212            .chain(settings)
213            .chain(source_references)
214            .chain(system_object_mappings)
215            .chain(system_configurations)
216            .chain(cluster_system_configurations)
217            .chain(replica_system_configurations)
218            .chain(default_privileges)
219            .chain(system_privileges)
220            .chain(storage_collection_metadata)
221            .chain(unfinalized_shards)
222            .chain(txn_wal_shard)
223            .chain(audit_logs)
224    }
225}
226
227/// The contents of a single state update.
228///
229/// The entire catalog is serialized as bytes and saved in a single persist shard. We use this
230/// enum to determine what collection something in the catalog belongs to.
231#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
232#[cfg_attr(test, derive(Arbitrary))]
233pub enum StateUpdateKind {
234    AuditLog(proto::AuditLogKey, ()),
235    Cluster(proto::ClusterKey, proto::ClusterValue),
236    ClusterReplica(proto::ClusterReplicaKey, proto::ClusterReplicaValue),
237    Comment(proto::CommentKey, proto::CommentValue),
238    Config(proto::ConfigKey, proto::ConfigValue),
239    Database(proto::DatabaseKey, proto::DatabaseValue),
240    DefaultPrivilege(proto::DefaultPrivilegesKey, proto::DefaultPrivilegesValue),
241    FenceToken(FenceToken),
242    IdAllocator(proto::IdAllocKey, proto::IdAllocValue),
243    IntrospectionSourceIndex(
244        proto::ClusterIntrospectionSourceIndexKey,
245        proto::ClusterIntrospectionSourceIndexValue,
246    ),
247    Item(proto::ItemKey, proto::ItemValue),
248    NetworkPolicy(proto::NetworkPolicyKey, proto::NetworkPolicyValue),
249    Role(proto::RoleKey, proto::RoleValue),
250    RoleAuth(proto::RoleAuthKey, proto::RoleAuthValue),
251    Schema(proto::SchemaKey, proto::SchemaValue),
252    Setting(proto::SettingKey, proto::SettingValue),
253    SourceReferences(proto::SourceReferencesKey, proto::SourceReferencesValue),
254    SystemConfiguration(
255        proto::ServerConfigurationKey,
256        proto::ServerConfigurationValue,
257    ),
258    ClusterSystemConfiguration(
259        proto::ClusterSystemConfigurationKey,
260        proto::ClusterSystemConfigurationValue,
261    ),
262    ReplicaSystemConfiguration(
263        proto::ReplicaSystemConfigurationKey,
264        proto::ReplicaSystemConfigurationValue,
265    ),
266    SystemObjectMapping(proto::GidMappingKey, proto::GidMappingValue),
267    SystemPrivilege(proto::SystemPrivilegesKey, proto::SystemPrivilegesValue),
268    StorageCollectionMetadata(
269        proto::StorageCollectionMetadataKey,
270        proto::StorageCollectionMetadataValue,
271    ),
272    UnfinalizedShard(proto::UnfinalizedShardKey, ()),
273    TxnWalShard((), proto::TxnWalShardValue),
274}
275
276impl StateUpdateKind {
277    pub(crate) fn collection_type(&self) -> Option<CollectionType> {
278        match self {
279            StateUpdateKind::AuditLog(_, _) => Some(CollectionType::AuditLog),
280            StateUpdateKind::Cluster(_, _) => Some(CollectionType::ComputeInstance),
281            StateUpdateKind::ClusterReplica(_, _) => Some(CollectionType::ComputeReplicas),
282            StateUpdateKind::Comment(_, _) => Some(CollectionType::Comments),
283            StateUpdateKind::Config(_, _) => Some(CollectionType::Config),
284            StateUpdateKind::Database(_, _) => Some(CollectionType::Database),
285            StateUpdateKind::DefaultPrivilege(_, _) => Some(CollectionType::DefaultPrivileges),
286            StateUpdateKind::FenceToken(_) => None,
287            StateUpdateKind::IdAllocator(_, _) => Some(CollectionType::IdAlloc),
288            StateUpdateKind::IntrospectionSourceIndex(_, _) => {
289                Some(CollectionType::ComputeIntrospectionSourceIndex)
290            }
291            StateUpdateKind::Item(_, _) => Some(CollectionType::Item),
292            StateUpdateKind::NetworkPolicy(_, _) => Some(CollectionType::NetworkPolicy),
293            StateUpdateKind::Role(_, _) => Some(CollectionType::Role),
294            StateUpdateKind::RoleAuth(_, _) => Some(CollectionType::RoleAuth),
295            StateUpdateKind::Schema(_, _) => Some(CollectionType::Schema),
296            StateUpdateKind::Setting(_, _) => Some(CollectionType::Setting),
297            StateUpdateKind::SourceReferences(_, _) => Some(CollectionType::SourceReferences),
298            StateUpdateKind::SystemConfiguration(_, _) => Some(CollectionType::SystemConfiguration),
299            StateUpdateKind::ClusterSystemConfiguration(_, _) => {
300                Some(CollectionType::ClusterSystemConfiguration)
301            }
302            StateUpdateKind::ReplicaSystemConfiguration(_, _) => {
303                Some(CollectionType::ReplicaSystemConfiguration)
304            }
305            StateUpdateKind::SystemObjectMapping(_, _) => Some(CollectionType::SystemGidMapping),
306            StateUpdateKind::SystemPrivilege(_, _) => Some(CollectionType::SystemPrivileges),
307            StateUpdateKind::StorageCollectionMetadata(_, _) => {
308                Some(CollectionType::StorageCollectionMetadata)
309            }
310            StateUpdateKind::UnfinalizedShard(_, _) => Some(CollectionType::UnfinalizedShard),
311            StateUpdateKind::TxnWalShard(_, _) => Some(CollectionType::TxnWalShard),
312        }
313    }
314}
315
316/// Version of [`StateUpdateKind`] to allow reading/writing raw json from/to persist.
317#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
318pub struct StateUpdateKindJson(Jsonb);
319
320impl StateUpdateKindJson {
321    pub(crate) fn from_serde<S: serde::Serialize>(s: S) -> Self {
322        let serde_value = serde_json::to_value(s).expect("valid json");
323        let row = Jsonb::from_serde_json(serde_value).expect("valid json");
324        StateUpdateKindJson(row)
325    }
326
327    pub(crate) fn to_serde<D: serde::de::DeserializeOwned>(&self) -> D {
328        self.try_to_serde().expect("jsonb should roundtrip")
329    }
330
331    pub(crate) fn try_to_serde<D: serde::de::DeserializeOwned>(
332        &self,
333    ) -> Result<D, serde_json::error::Error> {
334        let serde_value = self.0.as_ref().to_serde_json();
335        serde_json::from_value::<D>(serde_value)
336    }
337
338    fn kind(&self) -> &str {
339        let row = self.0.row();
340        let mut iter = row.unpack_first().unwrap_map().iter();
341        let datum = iter
342            .find_map(|(field, datum)| if field == "kind" { Some(datum) } else { None })
343            .expect("kind field must exist");
344        datum.unwrap_str()
345    }
346
347    /// Returns true if this is an update kind that is always deserializable, even before migrations. Otherwise, returns false.
348    pub(crate) fn is_always_deserializable(&self) -> bool {
349        // Construct some fake update kinds so we can extract exactly what the kind field will
350        // serialize as.
351        static DESERIALIZABLE_KINDS: LazyLock<HashSet<String>> = LazyLock::new(|| {
352            [
353                StateUpdateKind::FenceToken(FenceToken {
354                    deploy_generation: 1,
355                    epoch: Epoch::new(1).expect("non-zero"),
356                }),
357                StateUpdateKind::Config(
358                    proto::ConfigKey { key: String::new() },
359                    proto::ConfigValue { value: 1 },
360                ),
361                StateUpdateKind::Setting(
362                    proto::SettingKey {
363                        name: String::new(),
364                    },
365                    proto::SettingValue {
366                        value: String::new(),
367                    },
368                ),
369                StateUpdateKind::AuditLog(
370                    proto::AuditLogKey {
371                        event: proto::AuditLogEvent::V1(proto::AuditLogEventV1 {
372                            id: 1,
373                            event_type: proto::audit_log_event_v1::EventType::Create,
374                            object_type: proto::audit_log_event_v1::ObjectType::Cluster,
375                            user: None,
376                            occurred_at: proto::EpochMillis { millis: 1 },
377                            details: proto::audit_log_event_v1::Details::ResetAllV1(
378                                proto::Empty {},
379                            ),
380                        }),
381                    },
382                    (),
383                ),
384            ]
385            .into_iter()
386            .map(|kind| {
387                let json_kind: StateUpdateKindJson = kind.into();
388                json_kind.kind().to_string()
389            })
390            .collect()
391        });
392        DESERIALIZABLE_KINDS.contains(self.kind())
393    }
394
395    /// Returns true if this is an audit log update. Otherwise, returns false.
396    pub(crate) fn is_audit_log(&self) -> bool {
397        // Construct a fake audit log so we can extract exactly what the kind field will serialize
398        // as.
399        static AUDIT_LOG_KIND: LazyLock<String> = LazyLock::new(|| {
400            let audit_log = StateUpdateKind::AuditLog(
401                proto::AuditLogKey {
402                    event: proto::AuditLogEvent::V1(proto::AuditLogEventV1 {
403                        id: 1,
404                        event_type: proto::audit_log_event_v1::EventType::Create,
405                        object_type: proto::audit_log_event_v1::ObjectType::Cluster,
406                        user: None,
407                        occurred_at: proto::EpochMillis { millis: 1 },
408                        details: proto::audit_log_event_v1::Details::ResetAllV1(proto::Empty {}),
409                    }),
410                },
411                (),
412            );
413            let json_kind: StateUpdateKindJson = audit_log.into();
414            json_kind.kind().to_string()
415        });
416        &*AUDIT_LOG_KIND == self.kind()
417    }
418}
419
420/// Version of [`StateUpdateKind`] that is stored directly in persist.
421type PersistStateUpdate = ((SourceData, ()), Timestamp, StorageDiff);
422
423impl TryFrom<&StateUpdate<StateUpdateKind>> for Option<memory::objects::StateUpdate> {
424    type Error = DurableCatalogError;
425
426    fn try_from(
427        StateUpdate { kind, ts, diff }: &StateUpdate<StateUpdateKind>,
428    ) -> Result<Self, Self::Error> {
429        let kind: Option<memory::objects::StateUpdateKind> = TryInto::try_into(kind)?;
430        let update = kind.map(|kind| memory::objects::StateUpdate {
431            kind,
432            ts: ts.clone(),
433            diff: diff.clone().try_into().expect("invalid diff"),
434        });
435        Ok(update)
436    }
437}
438
439impl TryFrom<&StateUpdateKind> for Option<memory::objects::StateUpdateKind> {
440    type Error = DurableCatalogError;
441
442    fn try_from(kind: &StateUpdateKind) -> Result<Self, Self::Error> {
443        fn into_durable<PK, PV, T>(key: &PK, value: &PV) -> Result<T, DurableCatalogError>
444        where
445            PK: ProtoType<T::Key> + Clone,
446            PV: ProtoType<T::Value> + Clone,
447            T: DurableType,
448        {
449            let key = key.clone().into_rust()?;
450            let value = value.clone().into_rust()?;
451            Ok(T::from_key_value(key, value))
452        }
453
454        Ok(match kind {
455            StateUpdateKind::AuditLog(key, value) => {
456                let audit_log = into_durable(key, value)?;
457                Some(memory::objects::StateUpdateKind::AuditLog(audit_log))
458            }
459            StateUpdateKind::Cluster(key, value) => {
460                let cluster = into_durable(key, value)?;
461                Some(memory::objects::StateUpdateKind::Cluster(cluster))
462            }
463            StateUpdateKind::ClusterReplica(key, value) => {
464                let cluster_replica = into_durable(key, value)?;
465                Some(memory::objects::StateUpdateKind::ClusterReplica(
466                    cluster_replica,
467                ))
468            }
469            StateUpdateKind::Comment(key, value) => {
470                let comment = into_durable(key, value)?;
471                Some(memory::objects::StateUpdateKind::Comment(comment))
472            }
473            StateUpdateKind::Database(key, value) => {
474                let database = into_durable(key, value)?;
475                Some(memory::objects::StateUpdateKind::Database(database))
476            }
477            StateUpdateKind::DefaultPrivilege(key, value) => {
478                let default_privilege = into_durable(key, value)?;
479                Some(memory::objects::StateUpdateKind::DefaultPrivilege(
480                    default_privilege,
481                ))
482            }
483            StateUpdateKind::Item(key, value) => {
484                let item = into_durable(key, value)?;
485                Some(memory::objects::StateUpdateKind::Item(item))
486            }
487            StateUpdateKind::IntrospectionSourceIndex(key, value) => {
488                let introspection_source_index = into_durable(key, value)?;
489                Some(memory::objects::StateUpdateKind::IntrospectionSourceIndex(
490                    introspection_source_index,
491                ))
492            }
493            StateUpdateKind::NetworkPolicy(key, value) => {
494                let policy = into_durable(key, value)?;
495                Some(memory::objects::StateUpdateKind::NetworkPolicy(policy))
496            }
497            StateUpdateKind::Role(key, value) => {
498                let role = into_durable(key, value)?;
499                Some(memory::objects::StateUpdateKind::Role(role))
500            }
501            StateUpdateKind::RoleAuth(key, value) => {
502                let role_auth = into_durable(key, value)?;
503                Some(memory::objects::StateUpdateKind::RoleAuth(role_auth))
504            }
505            StateUpdateKind::Schema(key, value) => {
506                let schema = into_durable(key, value)?;
507                Some(memory::objects::StateUpdateKind::Schema(schema))
508            }
509            StateUpdateKind::SourceReferences(key, value) => {
510                let source_references = into_durable(key, value)?;
511                Some(memory::objects::StateUpdateKind::SourceReferences(
512                    source_references,
513                ))
514            }
515            StateUpdateKind::StorageCollectionMetadata(key, value) => {
516                let storage_collection_metadata = into_durable(key, value)?;
517                Some(memory::objects::StateUpdateKind::StorageCollectionMetadata(
518                    storage_collection_metadata,
519                ))
520            }
521            StateUpdateKind::SystemConfiguration(key, value) => {
522                let system_configuration = into_durable(key, value)?;
523                Some(memory::objects::StateUpdateKind::SystemConfiguration(
524                    system_configuration,
525                ))
526            }
527            StateUpdateKind::ClusterSystemConfiguration(key, value) => {
528                let cluster_system_configuration = into_durable(key, value)?;
529                Some(
530                    memory::objects::StateUpdateKind::ClusterSystemConfiguration(
531                        cluster_system_configuration,
532                    ),
533                )
534            }
535            StateUpdateKind::ReplicaSystemConfiguration(key, value) => {
536                let replica_system_configuration = into_durable(key, value)?;
537                Some(
538                    memory::objects::StateUpdateKind::ReplicaSystemConfiguration(
539                        replica_system_configuration,
540                    ),
541                )
542            }
543            StateUpdateKind::SystemObjectMapping(key, value) => {
544                let system_object_mapping = into_durable(key, value)?;
545                Some(memory::objects::StateUpdateKind::SystemObjectMapping(
546                    system_object_mapping,
547                ))
548            }
549            StateUpdateKind::SystemPrivilege(key, value) => {
550                let system_privilege = into_durable(key, value)?;
551                Some(memory::objects::StateUpdateKind::SystemPrivilege(
552                    system_privilege,
553                ))
554            }
555            StateUpdateKind::UnfinalizedShard(key, value) => {
556                let unfinalized_shard = into_durable(key, value)?;
557                Some(memory::objects::StateUpdateKind::UnfinalizedShard(
558                    unfinalized_shard,
559                ))
560            }
561            // Not exposed to higher layers.
562            StateUpdateKind::Config(_, _)
563            | StateUpdateKind::FenceToken(_)
564            | StateUpdateKind::IdAllocator(_, _)
565            | StateUpdateKind::Setting(_, _)
566            | StateUpdateKind::TxnWalShard(_, _) => None,
567        })
568    }
569}
570
571impl TryFrom<StateUpdate<StateUpdateKindJson>> for StateUpdate<StateUpdateKind> {
572    type Error = String;
573
574    fn try_from(update: StateUpdate<StateUpdateKindJson>) -> Result<Self, Self::Error> {
575        Ok(StateUpdate {
576            kind: TryInto::try_into(update.kind)?,
577            ts: update.ts,
578            diff: update.diff,
579        })
580    }
581}
582
583impl TryFrom<StateUpdateKindJson> for StateUpdateKind {
584    type Error = String;
585
586    fn try_from(value: StateUpdateKindJson) -> Result<Self, Self::Error> {
587        let kind: proto::StateUpdateKind = value.try_to_serde().map_err(|err| err.to_string())?;
588        StateUpdateKind::from_proto(kind).map_err(|err| err.to_string())
589    }
590}
591
592impl TryFrom<&StateUpdateKindJson> for StateUpdateKind {
593    type Error = String;
594
595    fn try_from(value: &StateUpdateKindJson) -> Result<Self, Self::Error> {
596        let kind: proto::StateUpdateKind = value.try_to_serde().map_err(|err| err.to_string())?;
597        StateUpdateKind::from_proto(kind).map_err(|err| err.to_string())
598    }
599}
600
601impl From<StateUpdateKind> for StateUpdateKindJson {
602    fn from(value: StateUpdateKind) -> Self {
603        let kind = value.into_proto_owned();
604        StateUpdateKindJson::from_serde(kind)
605    }
606}
607
608// Be very careful about changing these implementations. The default impl of `into_proto_owned`
609// calls `into_proto`, and this impl of `into_proto` calls `into_proto_owned`. It would be very
610// easy to accidentally cause infinite recursion.
611impl RustType<proto::StateUpdateKind> for StateUpdateKind {
612    fn into_proto(&self) -> proto::StateUpdateKind {
613        error!("unexpected clone of catalog data");
614        self.clone().into_proto_owned()
615    }
616
617    fn into_proto_owned(self) -> proto::StateUpdateKind {
618        match self {
619            StateUpdateKind::AuditLog(key, ()) => {
620                proto::StateUpdateKind::AuditLog(proto::AuditLog { key })
621            }
622            StateUpdateKind::Cluster(key, value) => {
623                proto::StateUpdateKind::Cluster(proto::Cluster { key, value })
624            }
625            StateUpdateKind::ClusterReplica(key, value) => {
626                proto::StateUpdateKind::ClusterReplica(proto::ClusterReplica { key, value })
627            }
628            StateUpdateKind::Comment(key, value) => {
629                proto::StateUpdateKind::Comment(proto::Comment { key, value })
630            }
631            StateUpdateKind::Config(key, value) => {
632                proto::StateUpdateKind::Config(proto::Config { key, value })
633            }
634            StateUpdateKind::Database(key, value) => {
635                proto::StateUpdateKind::Database(proto::Database { key, value })
636            }
637            StateUpdateKind::DefaultPrivilege(key, value) => {
638                proto::StateUpdateKind::DefaultPrivileges(proto::DefaultPrivileges { key, value })
639            }
640            StateUpdateKind::FenceToken(fence_token) => {
641                proto::StateUpdateKind::FenceToken(proto::FenceToken {
642                    deploy_generation: fence_token.deploy_generation,
643                    epoch: fence_token.epoch.get(),
644                })
645            }
646            StateUpdateKind::IdAllocator(key, value) => {
647                proto::StateUpdateKind::IdAlloc(proto::IdAlloc { key, value })
648            }
649            StateUpdateKind::IntrospectionSourceIndex(key, value) => {
650                proto::StateUpdateKind::ClusterIntrospectionSourceIndex(
651                    proto::ClusterIntrospectionSourceIndex { key, value },
652                )
653            }
654            StateUpdateKind::Item(key, value) => {
655                proto::StateUpdateKind::Item(proto::Item { key, value })
656            }
657            StateUpdateKind::NetworkPolicy(key, value) => {
658                proto::StateUpdateKind::NetworkPolicy(proto::NetworkPolicy { key, value })
659            }
660            StateUpdateKind::Role(key, value) => {
661                proto::StateUpdateKind::Role(proto::Role { key, value })
662            }
663            StateUpdateKind::RoleAuth(key, value) => {
664                proto::StateUpdateKind::RoleAuth(proto::RoleAuth { key, value })
665            }
666            StateUpdateKind::Schema(key, value) => {
667                proto::StateUpdateKind::Schema(proto::Schema { key, value })
668            }
669            StateUpdateKind::Setting(key, value) => {
670                proto::StateUpdateKind::Setting(proto::Setting { key, value })
671            }
672            StateUpdateKind::SourceReferences(key, value) => {
673                proto::StateUpdateKind::SourceReferences(proto::SourceReferences { key, value })
674            }
675            StateUpdateKind::SystemConfiguration(key, value) => {
676                proto::StateUpdateKind::ServerConfiguration(proto::ServerConfiguration {
677                    key,
678                    value,
679                })
680            }
681            StateUpdateKind::ClusterSystemConfiguration(key, value) => {
682                proto::StateUpdateKind::ClusterSystemConfiguration(
683                    proto::ClusterSystemConfiguration { key, value },
684                )
685            }
686            StateUpdateKind::ReplicaSystemConfiguration(key, value) => {
687                proto::StateUpdateKind::ReplicaSystemConfiguration(
688                    proto::ReplicaSystemConfiguration { key, value },
689                )
690            }
691            StateUpdateKind::SystemObjectMapping(key, value) => {
692                proto::StateUpdateKind::GidMapping(proto::GidMapping { key, value })
693            }
694            StateUpdateKind::SystemPrivilege(key, value) => {
695                proto::StateUpdateKind::SystemPrivileges(proto::SystemPrivileges { key, value })
696            }
697            StateUpdateKind::StorageCollectionMetadata(key, value) => {
698                proto::StateUpdateKind::StorageCollectionMetadata(
699                    proto::StorageCollectionMetadata { key, value },
700                )
701            }
702            StateUpdateKind::UnfinalizedShard(key, ()) => {
703                proto::StateUpdateKind::UnfinalizedShard(proto::UnfinalizedShard { key })
704            }
705            StateUpdateKind::TxnWalShard((), value) => {
706                proto::StateUpdateKind::TxnWalShard(proto::TxnWalShard { value })
707            }
708        }
709    }
710
711    fn from_proto(proto: proto::StateUpdateKind) -> Result<StateUpdateKind, TryFromProtoError> {
712        Ok(match proto {
713            proto::StateUpdateKind::AuditLog(proto::AuditLog { key }) => {
714                StateUpdateKind::AuditLog(key, ())
715            }
716            proto::StateUpdateKind::Cluster(proto::Cluster { key, value }) => {
717                StateUpdateKind::Cluster(key, value)
718            }
719            proto::StateUpdateKind::ClusterReplica(proto::ClusterReplica { key, value }) => {
720                StateUpdateKind::ClusterReplica(key, value)
721            }
722            proto::StateUpdateKind::Comment(proto::Comment { key, value }) => {
723                StateUpdateKind::Comment(key, value)
724            }
725            proto::StateUpdateKind::Config(proto::Config { key, value }) => {
726                StateUpdateKind::Config(key, value)
727            }
728            proto::StateUpdateKind::Database(proto::Database { key, value }) => {
729                StateUpdateKind::Database(key, value)
730            }
731            proto::StateUpdateKind::DefaultPrivileges(proto::DefaultPrivileges { key, value }) => {
732                StateUpdateKind::DefaultPrivilege(key, value)
733            }
734            proto::StateUpdateKind::FenceToken(proto::FenceToken {
735                deploy_generation,
736                epoch,
737            }) => StateUpdateKind::FenceToken(FenceToken {
738                deploy_generation,
739                epoch: Epoch::new(epoch).ok_or_else(|| {
740                    TryFromProtoError::missing_field("state_update_kind::Epoch::epoch")
741                })?,
742            }),
743            proto::StateUpdateKind::IdAlloc(proto::IdAlloc { key, value }) => {
744                StateUpdateKind::IdAllocator(key, value)
745            }
746            proto::StateUpdateKind::ClusterIntrospectionSourceIndex(
747                proto::ClusterIntrospectionSourceIndex { key, value },
748            ) => StateUpdateKind::IntrospectionSourceIndex(key, value),
749            proto::StateUpdateKind::Item(proto::Item { key, value }) => {
750                StateUpdateKind::Item(key, value)
751            }
752            proto::StateUpdateKind::Role(proto::Role { key, value }) => {
753                StateUpdateKind::Role(key, value)
754            }
755            proto::StateUpdateKind::RoleAuth(proto::RoleAuth { key, value }) => {
756                StateUpdateKind::RoleAuth(key, value)
757            }
758            proto::StateUpdateKind::Schema(proto::Schema { key, value }) => {
759                StateUpdateKind::Schema(key, value)
760            }
761            proto::StateUpdateKind::Setting(proto::Setting { key, value }) => {
762                StateUpdateKind::Setting(key, value)
763            }
764            proto::StateUpdateKind::ServerConfiguration(proto::ServerConfiguration {
765                key,
766                value,
767            }) => StateUpdateKind::SystemConfiguration(key, value),
768            proto::StateUpdateKind::ClusterSystemConfiguration(
769                proto::ClusterSystemConfiguration { key, value },
770            ) => StateUpdateKind::ClusterSystemConfiguration(key, value),
771            proto::StateUpdateKind::ReplicaSystemConfiguration(
772                proto::ReplicaSystemConfiguration { key, value },
773            ) => StateUpdateKind::ReplicaSystemConfiguration(key, value),
774            proto::StateUpdateKind::GidMapping(proto::GidMapping { key, value }) => {
775                StateUpdateKind::SystemObjectMapping(key, value)
776            }
777            proto::StateUpdateKind::SystemPrivileges(proto::SystemPrivileges { key, value }) => {
778                StateUpdateKind::SystemPrivilege(key, value)
779            }
780            proto::StateUpdateKind::StorageCollectionMetadata(
781                proto::StorageCollectionMetadata { key, value },
782            ) => StateUpdateKind::StorageCollectionMetadata(key, value),
783            proto::StateUpdateKind::UnfinalizedShard(proto::UnfinalizedShard { key }) => {
784                StateUpdateKind::UnfinalizedShard(key, ())
785            }
786            proto::StateUpdateKind::TxnWalShard(proto::TxnWalShard { value }) => {
787                StateUpdateKind::TxnWalShard((), value)
788            }
789            proto::StateUpdateKind::SourceReferences(proto::SourceReferences { key, value }) => {
790                StateUpdateKind::SourceReferences(key, value)
791            }
792            proto::StateUpdateKind::NetworkPolicy(proto::NetworkPolicy { key, value }) => {
793                StateUpdateKind::NetworkPolicy(key, value)
794            }
795        })
796    }
797}
798
799/// Decodes a [`StateUpdate<StateUpdateKindJson>`] from the `(key, value, ts,
800/// diff)` tuple/update we store in persist.
801impl From<PersistStateUpdate> for StateUpdate<StateUpdateKindJson> {
802    fn from(kvtd: PersistStateUpdate) -> Self {
803        let ((key, ()), ts, diff) = kvtd;
804        StateUpdate {
805            kind: StateUpdateKindJson::from(key),
806            ts,
807            diff: diff.into(),
808        }
809    }
810}
811
812impl From<StateUpdateKindJson> for SourceData {
813    fn from(value: StateUpdateKindJson) -> SourceData {
814        let row = value.0.into_row();
815        SourceData(Ok(row))
816    }
817}
818
819impl From<SourceData> for StateUpdateKindJson {
820    fn from(value: SourceData) -> Self {
821        let row = value.0.expect("only Ok values stored in catalog shard");
822        StateUpdateKindJson(Jsonb::from_row(row))
823    }
824}
825
826#[cfg(test)]
827mod tests {
828    use mz_persist_types::Codec;
829    use mz_repr::{RelationDesc, SqlScalarType};
830    use mz_storage_types::sources::SourceData;
831    use proptest::prelude::*;
832
833    use crate::durable::Epoch;
834    use crate::durable::objects::FenceToken;
835    use crate::durable::objects::serialization::proto;
836    use crate::durable::objects::state_update::{StateUpdateKind, StateUpdateKindJson};
837
838    #[mz_ore::test]
839    #[cfg_attr(miri, ignore)]
840    fn kind_test() {
841        let test_cases = [
842            (
843                StateUpdateKind::FenceToken(FenceToken {
844                    deploy_generation: 1,
845                    epoch: Epoch::new(1).expect("non-zero"),
846                }),
847                "FenceToken",
848            ),
849            (
850                StateUpdateKind::Config(
851                    proto::ConfigKey { key: String::new() },
852                    proto::ConfigValue { value: 1 },
853                ),
854                "Config",
855            ),
856            (
857                StateUpdateKind::Setting(
858                    proto::SettingKey {
859                        name: String::new(),
860                    },
861                    proto::SettingValue {
862                        value: String::new(),
863                    },
864                ),
865                "Setting",
866            ),
867            (
868                StateUpdateKind::AuditLog(
869                    proto::AuditLogKey {
870                        event: proto::AuditLogEvent::V1(proto::AuditLogEventV1 {
871                            id: 1,
872                            event_type: proto::audit_log_event_v1::EventType::Create,
873                            object_type: proto::audit_log_event_v1::ObjectType::Cluster,
874                            user: None,
875                            occurred_at: proto::EpochMillis { millis: 4 },
876                            details: proto::audit_log_event_v1::Details::ResetAllV1(
877                                proto::Empty {},
878                            ),
879                        }),
880                    },
881                    (),
882                ),
883                "AuditLog",
884            ),
885        ];
886
887        for (kind, expected) in test_cases {
888            let json_kind: StateUpdateKindJson = kind.into();
889            let kind = json_kind.kind().to_string();
890            assert_eq!(expected, kind);
891        }
892    }
893
894    proptest! {
895        #[mz_ore::test]
896        #[cfg_attr(miri, ignore)] // slow
897        fn proptest_state_update_kind_roundtrip(kind: StateUpdateKind) {
898            // Verify that we can map encode into the "raw" json format. This
899            // validates things like contained integers fitting in f64.
900            let raw = StateUpdateKindJson::from(kind.clone());
901            let desc = RelationDesc::builder().with_column("a", SqlScalarType::Jsonb.nullable(false)).finish();
902
903            // Verify that the raw roundtrips through the SourceData Codec impl.
904            let source_data = SourceData::from(raw.clone());
905            let mut encoded = Vec::new();
906            source_data.encode(&mut encoded);
907            let decoded = SourceData::decode(&encoded, &desc).expect("should be valid SourceData");
908            prop_assert_eq!(&source_data, &decoded);
909            let decoded = StateUpdateKindJson::from(decoded);
910            prop_assert_eq!(&raw, &decoded);
911
912            // Verify that the enum roundtrips.
913            let decoded = StateUpdateKind::try_from(decoded).expect("should be valid StateUpdateKind");
914            prop_assert_eq!(&kind, &decoded);
915        }
916    }
917}
mz_catalog/durable/objects/state_update.rs

mz_catalog/durable/objects/
state_update.rs
