mz_catalog/durable/upgrade/
v89_to_v90.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
10use crate::durable::upgrade::MigrationAction;
11use crate::durable::upgrade::json_compatible::JsonCompatible;
12use crate::durable::upgrade::objects_v89 as v89;
13use crate::durable::upgrade::objects_v90 as v90;
14
15crate::json_compatible!(v89::ClusterKey with v90::ClusterKey);
16crate::json_compatible!(v89::ClusterReplicaKey with v90::ClusterReplicaKey);
17crate::json_compatible!(v89::ClusterId with v90::ClusterId);
18crate::json_compatible!(v89::RoleId with v90::RoleId);
19crate::json_compatible!(v89::MzAclItem with v90::MzAclItem);
20crate::json_compatible!(v89::ReplicaLogging with v90::ReplicaLogging);
21crate::json_compatible!(v89::ReplicaLocation with v90::ReplicaLocation);
22crate::json_compatible!(v89::OptimizerFeatureOverride with v90::OptimizerFeatureOverride);
23crate::json_compatible!(v89::ClusterSchedule with v90::ClusterSchedule);
24crate::json_compatible!(v89::AutoScalingStrategy with v90::AutoScalingStrategy);
25crate::json_compatible!(v89::BurstState with v90::BurstState);
26crate::json_compatible!(v89::ReconfigurationStatus with v90::ReconfigurationStatus);
27crate::json_compatible!(v89::OnTimeoutAction with v90::OnTimeoutAction);
28
29/// Adds the `arrangement_compression` flag to managed clusters and cluster
30/// replicas, backfilling it as `false`. The flag also appears on a managed
31/// cluster's in-flight `reconfiguration` target, backfilled the same way.
32///
33/// Managed `Cluster` and `ClusterReplica` records gained a new field, so their
34/// stored JSON is no longer readable as the v90 type. Every such record is
35/// rewritten. Unmanaged clusters and all other records are unchanged and pass
36/// through untouched.
37pub fn upgrade(
38    snapshot: Vec<v89::StateUpdateKind>,
39) -> Vec<MigrationAction<v89::StateUpdateKind, v90::StateUpdateKind>> {
40    let mut migrations = Vec::new();
41    for update in snapshot {
42        match update {
43            v89::StateUpdateKind::Cluster(old_cluster)
44                if matches!(
45                    old_cluster.value.config.variant,
46                    v89::ClusterVariant::Managed(_)
47                ) =>
48            {
49                let new_cluster = migrate_cluster(old_cluster.clone());
50                migrations.push(MigrationAction::Update(
51                    v89::StateUpdateKind::Cluster(old_cluster),
52                    v90::StateUpdateKind::Cluster(new_cluster),
53                ));
54            }
55            v89::StateUpdateKind::ClusterReplica(old_replica) => {
56                let new_replica = migrate_replica(old_replica.clone());
57                migrations.push(MigrationAction::Update(
58                    v89::StateUpdateKind::ClusterReplica(old_replica),
59                    v90::StateUpdateKind::ClusterReplica(new_replica),
60                ));
61            }
62            _ => {}
63        }
64    }
65    migrations
66}
67
68fn migrate_cluster(old: v89::Cluster) -> v90::Cluster {
69    let v89::Cluster { key, value } = old;
70    let v89::ClusterVariant::Managed(m) = value.config.variant else {
71        unreachable!("caller guards on the managed variant");
72    };
73    v90::Cluster {
74        key: JsonCompatible::convert(&key),
75        value: v90::ClusterValue {
76            name: value.name,
77            owner_id: JsonCompatible::convert(&value.owner_id),
78            privileges: value
79                .privileges
80                .iter()
81                .map(JsonCompatible::convert)
82                .collect(),
83            config: v90::ClusterConfig {
84                workload_class: value.config.workload_class,
85                variant: v90::ClusterVariant::Managed(migrate_managed(m)),
86            },
87        },
88    }
89}
90
91fn migrate_managed(m: v89::ManagedCluster) -> v90::ManagedCluster {
92    v90::ManagedCluster {
93        size: m.size,
94        replication_factor: m.replication_factor,
95        availability_zones: m.availability_zones,
96        logging: JsonCompatible::convert(&m.logging),
97        arrangement_compression: false,
98        optimizer_feature_overrides: m
99            .optimizer_feature_overrides
100            .iter()
101            .map(JsonCompatible::convert)
102            .collect(),
103        schedule: JsonCompatible::convert(&m.schedule),
104        auto_scaling_strategy: m
105            .auto_scaling_strategy
106            .as_ref()
107            .map(JsonCompatible::convert),
108        reconfiguration: m.reconfiguration.map(migrate_reconfiguration),
109        burst: m.burst.as_ref().map(JsonCompatible::convert),
110    }
111}
112
113fn migrate_reconfiguration(old: v89::ReconfigurationState) -> v90::ReconfigurationState {
114    let v89::ReconfigurationTarget {
115        size,
116        replication_factor,
117        availability_zones,
118        logging,
119    } = old.target;
120    v90::ReconfigurationState {
121        target: v90::ReconfigurationTarget {
122            size,
123            replication_factor,
124            availability_zones,
125            logging: JsonCompatible::convert(&logging),
126            arrangement_compression: false,
127        },
128        deadline: old.deadline,
129        on_timeout: JsonCompatible::convert(&old.on_timeout),
130        status: JsonCompatible::convert(&old.status),
131    }
132}
133
134fn migrate_replica(old: v89::ClusterReplica) -> v90::ClusterReplica {
135    let v89::ClusterReplica { key, value } = old;
136    let v89::ReplicaConfig { logging, location } = value.config;
137    v90::ClusterReplica {
138        key: JsonCompatible::convert(&key),
139        value: v90::ClusterReplicaValue {
140            cluster_id: JsonCompatible::convert(&value.cluster_id),
141            name: value.name,
142            config: v90::ReplicaConfig {
143                logging: JsonCompatible::convert(&logging),
144                location: JsonCompatible::convert(&location),
145                arrangement_compression: false,
146            },
147            owner_id: JsonCompatible::convert(&value.owner_id),
148        },
149    }
150}
151
152#[cfg(test)]
153mod tests {
154    use crate::durable::upgrade::MigrationAction;
155    use crate::durable::upgrade::v89_to_v90::upgrade;
156    use crate::durable::upgrade::{objects_v89 as v89, objects_v90 as v90};
157
158    fn managed_cluster(id: u64) -> v89::Cluster {
159        v89::Cluster {
160            key: v89::ClusterKey {
161                id: v89::ClusterId::User(id),
162            },
163            value: v89::ClusterValue {
164                name: format!("cluster{id}"),
165                owner_id: v89::RoleId::User(1),
166                privileges: Vec::new(),
167                config: v89::ClusterConfig {
168                    workload_class: None,
169                    variant: v89::ClusterVariant::Managed(v89::ManagedCluster {
170                        size: "small".to_string(),
171                        replication_factor: 1,
172                        availability_zones: vec!["az1".to_string()],
173                        logging: v89::ReplicaLogging {
174                            log_logging: false,
175                            interval: None,
176                        },
177                        optimizer_feature_overrides: Vec::new(),
178                        schedule: v89::ClusterSchedule::Manual,
179                        auto_scaling_strategy: None,
180                        reconfiguration: None,
181                        burst: None,
182                    }),
183                },
184            },
185        }
186    }
187
188    fn unmanaged_cluster(id: u64) -> v89::Cluster {
189        v89::Cluster {
190            key: v89::ClusterKey {
191                id: v89::ClusterId::User(id),
192            },
193            value: v89::ClusterValue {
194                name: format!("cluster{id}"),
195                owner_id: v89::RoleId::User(1),
196                privileges: Vec::new(),
197                config: v89::ClusterConfig {
198                    workload_class: None,
199                    variant: v89::ClusterVariant::Unmanaged,
200                },
201            },
202        }
203    }
204
205    fn replica(id: u64) -> v89::ClusterReplica {
206        v89::ClusterReplica {
207            key: v89::ClusterReplicaKey {
208                id: v89::ReplicaId::User(id),
209            },
210            value: v89::ClusterReplicaValue {
211                cluster_id: v89::ClusterId::User(1),
212                name: format!("r{id}"),
213                config: v89::ReplicaConfig {
214                    logging: v89::ReplicaLogging {
215                        log_logging: false,
216                        interval: None,
217                    },
218                    location: v89::ReplicaLocation::Managed(v89::ManagedLocation {
219                        size: "small".to_string(),
220                        availability_zones: vec!["az1".to_string()],
221                        billed_as: None,
222                        internal: false,
223                        pending: false,
224                    }),
225                },
226                owner_id: v89::RoleId::User(1),
227            },
228        }
229    }
230
231    #[mz_ore::test]
232    fn backfills_managed_cluster_and_replica_as_false() {
233        let migrations = upgrade(vec![
234            v89::StateUpdateKind::Cluster(managed_cluster(1)),
235            v89::StateUpdateKind::Cluster(unmanaged_cluster(2)),
236            v89::StateUpdateKind::ClusterReplica(replica(1)),
237        ]);
238        // Managed cluster and replica migrate; the unmanaged cluster passes through.
239        assert_eq!(migrations.len(), 2);
240
241        let MigrationAction::Update(_, v90::StateUpdateKind::Cluster(cluster)) = &migrations[0]
242        else {
243            panic!("expected a cluster update");
244        };
245        let v90::ClusterVariant::Managed(managed) = &cluster.value.config.variant else {
246            panic!("expected a managed cluster");
247        };
248        assert!(!managed.arrangement_compression);
249
250        let MigrationAction::Update(_, v90::StateUpdateKind::ClusterReplica(replica)) =
251            &migrations[1]
252        else {
253            panic!("expected a replica update");
254        };
255        assert!(!replica.value.config.arrangement_compression);
256    }
257}
mz_catalog/durable/upgrade/v89_to_v90.rs

mz_catalog/durable/upgrade/
v89_to_v90.rs
