mz_catalog/durable/
debug.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//! Functionality for manually modifying and displaying the catalog contents. This is helpful for
11//! fixing a corrupt catalog.
12
13use std::fmt::Debug;
14
15use mz_repr::Diff;
16use serde::{Deserialize, Serialize};
17use serde_plain::{derive_display_from_serialize, derive_fromstr_from_deserialize};
18
19use crate::durable::CatalogError;
20use crate::durable::objects::serialization::proto;
21use crate::durable::objects::state_update::StateUpdateKind;
22use crate::durable::persist::{Timestamp, UnopenedPersistCatalogState};
23
24/// The contents of the catalog are logically separated into separate [`Collection`]s, which
25/// describe the category of data that the content belongs to.
26pub trait Collection: Debug {
27    /// Type used to stores keys for [`Collection`].
28    type Key;
29    /// Type used to stores values for [`Collection`].
30    type Value;
31
32    /// [`CollectionType`] corresponding to [`Collection`].
33    fn collection_type() -> CollectionType;
34
35    /// Extract the [`CollectionTrace`] from a [`Trace`] that corresponds to [`Collection`].
36    fn collection_trace(trace: Trace) -> CollectionTrace<Self>;
37
38    /// Generate a `StateUpdateKind` with `key` and `value` that corresponds to [`Collection`].
39    fn update(key: Self::Key, value: Self::Value) -> StateUpdateKind;
40
41    /// The human-readable name of this collection.
42    fn name() -> String {
43        Self::collection_type().to_string()
44    }
45}
46
47/// The type of a [`Collection`].
48///
49/// See [`Collection`] for more details.
50///
51/// The names of each variant are used to determine the labels of each [`CollectionTrace`] when
52/// dumping a [`Trace`].
53#[derive(PartialEq, Eq, Hash, Clone, Debug, Serialize, Deserialize)]
54#[serde(rename_all = "snake_case")]
55pub enum CollectionType {
56    AuditLog,
57    ComputeInstance,
58    ComputeIntrospectionSourceIndex,
59    ComputeReplicas,
60    ClusterReplicaSize,
61    Comments,
62    Config,
63    Database,
64    DefaultPrivileges,
65    IdAlloc,
66    IndexColumns,
67    Item,
68    ItemColumns,
69    NetworkPolicy,
70    Role,
71    RoleAuth,
72    Schema,
73    Setting,
74    SourceReferences,
75    SystemConfiguration,
76    ClusterSystemConfiguration,
77    ReplicaSystemConfiguration,
78    SystemGidMapping,
79    SystemPrivileges,
80    StorageCollectionMetadata,
81    UnfinalizedShard,
82    TxnWalShard,
83}
84
85derive_display_from_serialize!(CollectionType);
86derive_fromstr_from_deserialize!(CollectionType);
87
88/// Macro to simplify implementing [`Collection`].
89///
90/// The arguments to `collection_impl!` are:
91/// - `$name`, which will be the name of the implementing struct.
92/// - `$key`, the type used to store keys.
93/// - `$value`, the type used to store values.
94/// - `$collection_type`, the [`CollectionType`].
95/// - `$trace_field`, the corresponding field name within a [`Trace`].
96/// - `$update`, the corresponding [`StateUpdateKind`] constructor.
97macro_rules! collection_impl {
98    ({
99    name: $name:ident,
100    key: $key:ty,
101    value: $value:ty,
102    collection_type: $collection_type:expr,
103    trace_field: $trace_field:ident,
104    update: $update:expr,
105}) => {
106        #[derive(Debug, Clone, PartialEq, Eq)]
107        pub struct $name {}
108
109        impl Collection for $name {
110            type Key = $key;
111            type Value = $value;
112
113            fn collection_type() -> CollectionType {
114                $collection_type
115            }
116
117            fn collection_trace(trace: Trace) -> CollectionTrace<Self> {
118                trace.$trace_field
119            }
120
121            fn update(key: Self::Key, value: Self::Value) -> StateUpdateKind {
122                $update(key, value)
123            }
124        }
125    };
126}
127
128collection_impl!({
129    name: AuditLogCollection,
130    key: proto::AuditLogKey,
131    value: (),
132    collection_type: CollectionType::AuditLog,
133    trace_field: audit_log,
134    update: StateUpdateKind::AuditLog,
135});
136collection_impl!({
137    name: ClusterCollection,
138    key: proto::ClusterKey,
139    value: proto::ClusterValue,
140    collection_type: CollectionType::ComputeInstance,
141    trace_field: clusters,
142    update: StateUpdateKind::Cluster,
143});
144collection_impl!({
145    name: ClusterIntrospectionSourceIndexCollection,
146    key: proto::ClusterIntrospectionSourceIndexKey,
147    value: proto::ClusterIntrospectionSourceIndexValue,
148    collection_type: CollectionType::ComputeIntrospectionSourceIndex,
149    trace_field: introspection_sources,
150    update: StateUpdateKind::IntrospectionSourceIndex,
151});
152collection_impl!({
153    name: ClusterReplicaCollection,
154    key: proto::ClusterReplicaKey,
155    value: proto::ClusterReplicaValue,
156    collection_type: CollectionType::ComputeReplicas,
157    trace_field: cluster_replicas,
158    update: StateUpdateKind::ClusterReplica,
159});
160collection_impl!({
161    name: ClusterReplicaSizeCollection,
162    key: proto::ClusterReplicaSizeKey,
163    value: proto::ClusterReplicaSizeValue,
164    collection_type: CollectionType::ClusterReplicaSize,
165    trace_field: cluster_replica_sizes,
166    update: StateUpdateKind::ClusterReplicaSize,
167});
168collection_impl!({
169    name: CommentCollection,
170    key: proto::CommentKey,
171    value: proto::CommentValue,
172    collection_type: CollectionType::Comments,
173    trace_field: comments,
174    update: StateUpdateKind::Comment,
175});
176collection_impl!({
177    name: ConfigCollection,
178    key: proto::ConfigKey,
179    value: proto::ConfigValue,
180    collection_type: CollectionType::Config,
181    trace_field: configs,
182    update: StateUpdateKind::Config,
183});
184collection_impl!({
185    name: DatabaseCollection,
186    key: proto::DatabaseKey,
187    value: proto::DatabaseValue,
188    collection_type: CollectionType::Database,
189    trace_field: databases,
190    update: StateUpdateKind::Database,
191});
192collection_impl!({
193    name: DefaultPrivilegeCollection,
194    key: proto::DefaultPrivilegesKey,
195    value: proto::DefaultPrivilegesValue,
196    collection_type: CollectionType::DefaultPrivileges,
197    trace_field: default_privileges,
198    update: StateUpdateKind::DefaultPrivilege,
199});
200collection_impl!({
201    name: IdAllocatorCollection,
202    key: proto::IdAllocKey,
203    value: proto::IdAllocValue,
204    collection_type: CollectionType::IdAlloc,
205    trace_field: id_allocator,
206    update: StateUpdateKind::IdAllocator,
207});
208collection_impl!({
209    name: ItemCollection,
210    key: proto::ItemKey,
211    value: proto::ItemValue,
212    collection_type: CollectionType::Item,
213    trace_field: items,
214    update: StateUpdateKind::Item,
215});
216collection_impl!({
217    name: ItemColumnsCollection,
218    key: proto::ItemColumnsKey,
219    value: proto::ItemColumnsValue,
220    collection_type: CollectionType::ItemColumns,
221    trace_field: item_columns,
222    update: StateUpdateKind::ItemColumns,
223});
224collection_impl!({
225    name: IndexColumnsCollection,
226    key: proto::IndexColumnsKey,
227    value: proto::IndexColumnsValue,
228    collection_type: CollectionType::IndexColumns,
229    trace_field: index_columns,
230    update: StateUpdateKind::IndexColumns,
231});
232collection_impl!({
233    name: NetworkPolicyCollection,
234    key: proto::NetworkPolicyKey,
235    value: proto::NetworkPolicyValue,
236    collection_type: CollectionType::NetworkPolicy,
237    trace_field: network_policies,
238    update: StateUpdateKind::NetworkPolicy,
239});
240collection_impl!({
241    name: RoleCollection,
242    key: proto::RoleKey,
243    value: proto::RoleValue,
244    collection_type: CollectionType::Role,
245    trace_field: roles,
246    update: StateUpdateKind::Role,
247});
248collection_impl!({
249    name: RoleAuthCollection,
250    key: proto::RoleAuthKey,
251    value: proto::RoleAuthValue,
252    collection_type: CollectionType::RoleAuth,
253    trace_field: role_auth,
254    update: StateUpdateKind::RoleAuth,
255});
256collection_impl!({
257    name: SchemaCollection,
258    key: proto::SchemaKey,
259    value: proto::SchemaValue,
260    collection_type: CollectionType::Schema,
261    trace_field: schemas,
262    update: StateUpdateKind::Schema,
263});
264collection_impl!({
265    name: SettingCollection,
266    key: proto::SettingKey,
267    value: proto::SettingValue,
268    collection_type: CollectionType::Setting,
269    trace_field: settings,
270    update: StateUpdateKind::Setting,
271});
272collection_impl!({
273    name: SourceReferencesCollection,
274    key: proto::SourceReferencesKey,
275    value: proto::SourceReferencesValue,
276    collection_type: CollectionType::SourceReferences,
277    trace_field: source_references,
278    update: StateUpdateKind::SourceReferences,
279});
280collection_impl!({
281    name: SystemConfigurationCollection,
282    key: proto::ServerConfigurationKey,
283    value: proto::ServerConfigurationValue,
284    collection_type: CollectionType::SystemConfiguration,
285    trace_field: system_configurations,
286    update: StateUpdateKind::SystemConfiguration,
287});
288collection_impl!({
289    name: ClusterSystemConfigurationCollection,
290    key: proto::ClusterSystemConfigurationKey,
291    value: proto::ClusterSystemConfigurationValue,
292    collection_type: CollectionType::ClusterSystemConfiguration,
293    trace_field: cluster_system_configurations,
294    update: StateUpdateKind::ClusterSystemConfiguration,
295});
296collection_impl!({
297    name: ReplicaSystemConfigurationCollection,
298    key: proto::ReplicaSystemConfigurationKey,
299    value: proto::ReplicaSystemConfigurationValue,
300    collection_type: CollectionType::ReplicaSystemConfiguration,
301    trace_field: replica_system_configurations,
302    update: StateUpdateKind::ReplicaSystemConfiguration,
303});
304collection_impl!({
305    name: SystemItemMappingCollection,
306    key: proto::GidMappingKey,
307    value: proto::GidMappingValue,
308    collection_type: CollectionType::SystemGidMapping,
309    trace_field: system_object_mappings,
310    update: StateUpdateKind::SystemObjectMapping,
311});
312collection_impl!({
313    name: SystemPrivilegeCollection,
314    key: proto::SystemPrivilegesKey,
315    value: proto::SystemPrivilegesValue,
316    collection_type: CollectionType::SystemPrivileges,
317    trace_field: system_privileges,
318    update: StateUpdateKind::SystemPrivilege,
319});
320
321collection_impl!({
322    name: StorageCollectionMetadataCollection,
323    key: proto::StorageCollectionMetadataKey,
324    value: proto::StorageCollectionMetadataValue,
325    collection_type: CollectionType::StorageCollectionMetadata,
326    trace_field: storage_collection_metadata,
327    update: StateUpdateKind::StorageCollectionMetadata,
328});
329collection_impl!({
330    name: UnfinalizedShardsCollection,
331    key: proto::UnfinalizedShardKey,
332    value: (),
333    collection_type: CollectionType::UnfinalizedShard,
334    trace_field: unfinalized_shards,
335    update: StateUpdateKind::UnfinalizedShard,
336});
337collection_impl!({
338    name: TxnWalShardCollection,
339    key: (),
340    value: proto::TxnWalShardValue,
341    collection_type: CollectionType::TxnWalShard,
342    trace_field: txn_wal_shard,
343    update: StateUpdateKind::TxnWalShard,
344});
345
346/// A trace of timestamped diffs for a particular [`Collection`].
347///
348/// The timestamps are represented as strings since different implementations use non-compatible
349/// timestamp types.
350#[derive(Debug, Clone, PartialEq, Eq)]
351pub struct CollectionTrace<T: Collection + ?Sized> {
352    pub values: Vec<((T::Key, T::Value), Timestamp, Diff)>,
353}
354
355impl<T: Collection> CollectionTrace<T> {
356    fn new() -> CollectionTrace<T> {
357        CollectionTrace { values: Vec::new() }
358    }
359}
360
361impl<T: Collection> CollectionTrace<T>
362where
363    T: Collection,
364    T::Key: Ord,
365    T::Value: Ord,
366{
367    fn sort(&mut self) {
368        self.values
369            .sort_by(|(x1, ts1, d1), (x2, ts2, d2)| ts1.cmp(ts2).then(d1.cmp(d2)).then(x1.cmp(x2)));
370    }
371}
372
373/// Catalog data structured as timestamped diffs.
374#[derive(Debug, Clone, PartialEq, Eq)]
375pub struct Trace {
376    pub audit_log: CollectionTrace<AuditLogCollection>,
377    pub clusters: CollectionTrace<ClusterCollection>,
378    pub introspection_sources: CollectionTrace<ClusterIntrospectionSourceIndexCollection>,
379    pub cluster_replicas: CollectionTrace<ClusterReplicaCollection>,
380    pub cluster_replica_sizes: CollectionTrace<ClusterReplicaSizeCollection>,
381    pub comments: CollectionTrace<CommentCollection>,
382    pub configs: CollectionTrace<ConfigCollection>,
383    pub databases: CollectionTrace<DatabaseCollection>,
384    pub default_privileges: CollectionTrace<DefaultPrivilegeCollection>,
385    pub id_allocator: CollectionTrace<IdAllocatorCollection>,
386    pub items: CollectionTrace<ItemCollection>,
387    pub item_columns: CollectionTrace<ItemColumnsCollection>,
388    pub index_columns: CollectionTrace<IndexColumnsCollection>,
389    pub network_policies: CollectionTrace<NetworkPolicyCollection>,
390    pub roles: CollectionTrace<RoleCollection>,
391    pub role_auth: CollectionTrace<RoleAuthCollection>,
392    pub schemas: CollectionTrace<SchemaCollection>,
393    pub settings: CollectionTrace<SettingCollection>,
394    pub source_references: CollectionTrace<SourceReferencesCollection>,
395    pub system_object_mappings: CollectionTrace<SystemItemMappingCollection>,
396    pub system_configurations: CollectionTrace<SystemConfigurationCollection>,
397    pub cluster_system_configurations: CollectionTrace<ClusterSystemConfigurationCollection>,
398    pub replica_system_configurations: CollectionTrace<ReplicaSystemConfigurationCollection>,
399    pub system_privileges: CollectionTrace<SystemPrivilegeCollection>,
400    pub storage_collection_metadata: CollectionTrace<StorageCollectionMetadataCollection>,
401    pub unfinalized_shards: CollectionTrace<UnfinalizedShardsCollection>,
402    pub txn_wal_shard: CollectionTrace<TxnWalShardCollection>,
403}
404
405impl Trace {
406    pub(crate) fn new() -> Trace {
407        Trace {
408            audit_log: CollectionTrace::new(),
409            clusters: CollectionTrace::new(),
410            introspection_sources: CollectionTrace::new(),
411            cluster_replicas: CollectionTrace::new(),
412            cluster_replica_sizes: CollectionTrace::new(),
413            comments: CollectionTrace::new(),
414            configs: CollectionTrace::new(),
415            databases: CollectionTrace::new(),
416            default_privileges: CollectionTrace::new(),
417            id_allocator: CollectionTrace::new(),
418            items: CollectionTrace::new(),
419            item_columns: CollectionTrace::new(),
420            index_columns: CollectionTrace::new(),
421            network_policies: CollectionTrace::new(),
422            roles: CollectionTrace::new(),
423            role_auth: CollectionTrace::new(),
424            schemas: CollectionTrace::new(),
425            settings: CollectionTrace::new(),
426            source_references: CollectionTrace::new(),
427            system_object_mappings: CollectionTrace::new(),
428            system_configurations: CollectionTrace::new(),
429            cluster_system_configurations: CollectionTrace::new(),
430            replica_system_configurations: CollectionTrace::new(),
431            system_privileges: CollectionTrace::new(),
432            storage_collection_metadata: CollectionTrace::new(),
433            unfinalized_shards: CollectionTrace::new(),
434            txn_wal_shard: CollectionTrace::new(),
435        }
436    }
437
438    pub fn sort(&mut self) {
439        let Trace {
440            audit_log,
441            clusters,
442            introspection_sources,
443            cluster_replicas,
444            cluster_replica_sizes,
445            comments,
446            configs,
447            databases,
448            default_privileges,
449            id_allocator,
450            items,
451            item_columns,
452            index_columns,
453            network_policies,
454            roles,
455            role_auth,
456            schemas,
457            settings,
458            source_references,
459            system_object_mappings,
460            system_configurations,
461            cluster_system_configurations,
462            replica_system_configurations,
463            system_privileges,
464            storage_collection_metadata,
465            unfinalized_shards,
466            txn_wal_shard,
467        } = self;
468        audit_log.sort();
469        clusters.sort();
470        introspection_sources.sort();
471        cluster_replicas.sort();
472        cluster_replica_sizes.sort();
473        comments.sort();
474        configs.sort();
475        databases.sort();
476        default_privileges.sort();
477        id_allocator.sort();
478        items.sort();
479        item_columns.sort();
480        index_columns.sort();
481        network_policies.sort();
482        roles.sort();
483        role_auth.sort();
484        schemas.sort();
485        settings.sort();
486        source_references.sort();
487        system_object_mappings.sort();
488        system_configurations.sort();
489        cluster_system_configurations.sort();
490        replica_system_configurations.sort();
491        system_privileges.sort();
492        storage_collection_metadata.sort();
493        unfinalized_shards.sort();
494        txn_wal_shard.sort();
495    }
496}
497
498pub struct DebugCatalogState(pub(crate) UnopenedPersistCatalogState);
499
500impl DebugCatalogState {
501    /// Manually update value of `key` in collection `T` to `value`.
502    pub async fn edit<T: Collection>(
503        &mut self,
504        key: T::Key,
505        value: T::Value,
506    ) -> Result<Option<T::Value>, CatalogError>
507    where
508        T::Key: PartialEq + Eq + Debug + Clone,
509        T::Value: Debug + Clone,
510    {
511        self.0.debug_edit::<T>(key, value).await
512    }
513
514    /// Manually delete `key` from collection `T`.
515    pub async fn delete<T: Collection>(&mut self, key: T::Key) -> Result<(), CatalogError>
516    where
517        T::Key: PartialEq + Eq + Debug + Clone,
518        T::Value: Debug,
519    {
520        self.0.debug_delete::<T>(key).await
521    }
522}
mz_catalog/durable/debug.rs

mz_catalog/durable/
debug.rs
