mz_catalog/durable/upgrade/
v88_to_v89.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_v88 as v88;
13use crate::durable::upgrade::objects_v89 as v89;
14
15crate::json_compatible!(v88::ClusterKey with v89::ClusterKey);
16crate::json_compatible!(v88::RoleId with v89::RoleId);
17crate::json_compatible!(v88::MzAclItem with v89::MzAclItem);
18crate::json_compatible!(v88::ReplicaLogging with v89::ReplicaLogging);
19crate::json_compatible!(v88::OptimizerFeatureOverride with v89::OptimizerFeatureOverride);
20crate::json_compatible!(v88::ClusterSchedule with v89::ClusterSchedule);
21crate::json_compatible!(v88::AutoScalingStrategy with v89::AutoScalingStrategy);
22crate::json_compatible!(v88::BurstState with v89::BurstState);
23crate::json_compatible!(v88::ReconfigurationTarget with v89::ReconfigurationTarget);
24crate::json_compatible!(v88::OnTimeoutAction with v89::OnTimeoutAction);
25
26/// Adds `ReconfigurationState::status`, backfilling any in-flight
27/// reconfiguration as `InProgress`. The remaining v88->v89 changes are
28/// additive and confined to the append-only audit log (new reconfiguration
29/// lifecycle transitions and hydration-burst lifecycle details), so nothing
30/// else is rewritten.
31pub fn upgrade(
32    snapshot: Vec<v88::StateUpdateKind>,
33) -> Vec<MigrationAction<v88::StateUpdateKind, v89::StateUpdateKind>> {
34    let mut migrations = Vec::new();
35    for update in snapshot {
36        match update {
37            v88::StateUpdateKind::Cluster(old_cluster)
38                if matches!(
39                    old_cluster.value.config.variant,
40                    v88::ClusterVariant::Managed(v88::ManagedCluster {
41                        reconfiguration: Some(_),
42                        ..
43                    })
44                ) =>
45            {
46                let new_cluster = migrate_cluster(old_cluster.clone());
47                migrations.push(MigrationAction::Update(
48                    v88::StateUpdateKind::Cluster(old_cluster),
49                    v89::StateUpdateKind::Cluster(new_cluster),
50                ));
51            }
52            _ => {}
53        }
54    }
55    migrations
56}
57
58fn migrate_cluster(old: v88::Cluster) -> v89::Cluster {
59    let v88::Cluster { key, value } = old;
60    let v88::ClusterVariant::Managed(m) = value.config.variant else {
61        unreachable!("caller guards on the managed variant");
62    };
63    let logging = JsonCompatible::convert(&m.logging);
64    let optimizer_feature_overrides = m
65        .optimizer_feature_overrides
66        .iter()
67        .map(JsonCompatible::convert)
68        .collect();
69    let schedule = JsonCompatible::convert(&m.schedule);
70    let auto_scaling_strategy = m
71        .auto_scaling_strategy
72        .as_ref()
73        .map(JsonCompatible::convert);
74    let reconfiguration = m.reconfiguration.map(migrate_reconfiguration);
75    let burst = m.burst.as_ref().map(JsonCompatible::convert);
76    v89::Cluster {
77        key: JsonCompatible::convert(&key),
78        value: v89::ClusterValue {
79            name: value.name,
80            owner_id: JsonCompatible::convert(&value.owner_id),
81            privileges: value
82                .privileges
83                .iter()
84                .map(JsonCompatible::convert)
85                .collect(),
86            config: v89::ClusterConfig {
87                workload_class: value.config.workload_class,
88                variant: v89::ClusterVariant::Managed(v89::ManagedCluster {
89                    size: m.size,
90                    replication_factor: m.replication_factor,
91                    availability_zones: m.availability_zones,
92                    logging,
93                    optimizer_feature_overrides,
94                    schedule,
95                    auto_scaling_strategy,
96                    reconfiguration,
97                    burst,
98                }),
99            },
100        },
101    }
102}
103
104fn migrate_reconfiguration(old: v88::ReconfigurationState) -> v89::ReconfigurationState {
105    v89::ReconfigurationState {
106        target: JsonCompatible::convert(&old.target),
107        deadline: old.deadline,
108        on_timeout: JsonCompatible::convert(&old.on_timeout),
109        status: v89::ReconfigurationStatus::InProgress,
110    }
111}
112
113#[cfg(test)]
114mod tests {
115    use crate::durable::upgrade::MigrationAction;
116    use crate::durable::upgrade::v88_to_v89::upgrade;
117    use crate::durable::upgrade::{objects_v88 as v88, objects_v89 as v89};
118
119    fn managed_cluster(
120        id: u64,
121        reconfiguration: Option<v88::ReconfigurationState>,
122    ) -> v88::Cluster {
123        v88::Cluster {
124            key: v88::ClusterKey {
125                id: v88::ClusterId::User(id),
126            },
127            value: v88::ClusterValue {
128                name: format!("cluster{id}"),
129                owner_id: v88::RoleId::User(1),
130                privileges: Vec::new(),
131                config: v88::ClusterConfig {
132                    workload_class: Some("production".to_string()),
133                    variant: v88::ClusterVariant::Managed(v88::ManagedCluster {
134                        size: "small".to_string(),
135                        replication_factor: 1,
136                        availability_zones: vec!["az1".to_string()],
137                        logging: v88::ReplicaLogging {
138                            log_logging: false,
139                            interval: None,
140                        },
141                        optimizer_feature_overrides: Vec::new(),
142                        schedule: v88::ClusterSchedule::Manual,
143                        auto_scaling_strategy: None,
144                        reconfiguration,
145                        burst: None,
146                    }),
147                },
148            },
149        }
150    }
151
152    fn reconfiguration() -> v88::ReconfigurationState {
153        v88::ReconfigurationState {
154            target: v88::ReconfigurationTarget {
155                size: "large".to_string(),
156                replication_factor: 2,
157                availability_zones: vec!["az2".to_string()],
158                logging: v88::ReplicaLogging {
159                    log_logging: true,
160                    interval: None,
161                },
162            },
163            deadline: 42,
164            on_timeout: v88::OnTimeoutAction::Rollback,
165        }
166    }
167
168    #[mz_ore::test]
169    fn backfills_existing_reconfiguration_as_in_progress() {
170        let migrations = upgrade(vec![v88::StateUpdateKind::Cluster(managed_cluster(
171            1,
172            Some(reconfiguration()),
173        ))]);
174        assert_eq!(migrations.len(), 1);
175
176        let MigrationAction::Update(_, v89::StateUpdateKind::Cluster(cluster)) = &migrations[0]
177        else {
178            panic!("expected a cluster update");
179        };
180        let v89::ClusterVariant::Managed(managed) = &cluster.value.config.variant else {
181            panic!("expected a managed cluster");
182        };
183        let record = managed
184            .reconfiguration
185            .as_ref()
186            .expect("reconfiguration should be retained");
187        assert_eq!(record.status, v89::ReconfigurationStatus::InProgress);
188        assert_eq!(record.target.size, "large");
189        assert_eq!(record.deadline, 42);
190        assert_eq!(
191            cluster.value.config.workload_class.as_deref(),
192            Some("production")
193        );
194    }
195
196    #[mz_ore::test]
197    fn skips_clusters_without_reconfiguration() {
198        let migrations = upgrade(vec![v88::StateUpdateKind::Cluster(managed_cluster(
199            1, None,
200        ))]);
201        assert!(migrations.is_empty());
202    }
203}
mz_catalog/durable/upgrade/v88_to_v89.rs

mz_catalog/durable/upgrade/
v88_to_v89.rs
