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

mz_catalog/durable/objects/
state_update.rs
