mz_catalog/durable/upgrade/
v92_to_v93.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_v92 as v92;
13use crate::durable::upgrade::objects_v93 as v93;
14
15crate::json_compatible!(v92::ClusterKey with v93::ClusterKey);
16crate::json_compatible!(v92::RoleId with v93::RoleId);
17crate::json_compatible!(v92::MzAclItem with v93::MzAclItem);
18crate::json_compatible!(v92::ReplicaLogging with v93::ReplicaLogging);
19crate::json_compatible!(v92::OptimizerFeatureOverride with v93::OptimizerFeatureOverride);
20crate::json_compatible!(v92::ClusterSchedule with v93::ClusterSchedule);
21crate::json_compatible!(v92::AutoScalingStrategy with v93::AutoScalingStrategy);
22crate::json_compatible!(v92::ReconfigurationState with v93::ReconfigurationState);
23crate::json_compatible!(v92::Duration with v93::Duration);
24
25/// Rewrites active bursts with an unset caught-up readiness timestamp.
26pub fn upgrade(
27    snapshot: Vec<v92::StateUpdateKind>,
28) -> Vec<MigrationAction<v92::StateUpdateKind, v93::StateUpdateKind>> {
29    // Unstamped bursts also need rewriting so future retractions match the stored JSON.
30    snapshot
31        .into_iter()
32        .filter_map(|update| match update {
33            v92::StateUpdateKind::Cluster(cluster)
34                if matches!(
35                    &cluster.value.config.variant,
36                    v92::ClusterVariant::Managed(managed) if managed.burst.is_some()
37                ) =>
38            {
39                let new_cluster = migrate_cluster(cluster.clone());
40                Some(MigrationAction::Update(
41                    v92::StateUpdateKind::Cluster(cluster),
42                    v93::StateUpdateKind::Cluster(new_cluster),
43                ))
44            }
45            _ => None,
46        })
47        .collect()
48}
49
50fn migrate_cluster(old: v92::Cluster) -> v93::Cluster {
51    let v92::Cluster { key, value } = old;
52    let v92::ClusterVariant::Managed(m) = value.config.variant else {
53        unreachable!("caller guards on an active managed burst");
54    };
55    v93::Cluster {
56        key: JsonCompatible::convert(&key),
57        value: v93::ClusterValue {
58            name: value.name,
59            owner_id: JsonCompatible::convert(&value.owner_id),
60            privileges: value
61                .privileges
62                .iter()
63                .map(JsonCompatible::convert)
64                .collect(),
65            config: v93::ClusterConfig {
66                workload_class: value.config.workload_class,
67                variant: v93::ClusterVariant::Managed(v93::ManagedCluster {
68                    size: m.size,
69                    replication_factor: m.replication_factor,
70                    availability_zones: m.availability_zones,
71                    logging: JsonCompatible::convert(&m.logging),
72                    arrangement_compression: m.arrangement_compression,
73                    optimizer_feature_overrides: m
74                        .optimizer_feature_overrides
75                        .iter()
76                        .map(JsonCompatible::convert)
77                        .collect(),
78                    schedule: JsonCompatible::convert(&m.schedule),
79                    auto_scaling_strategy: m
80                        .auto_scaling_strategy
81                        .as_ref()
82                        .map(JsonCompatible::convert),
83                    reconfiguration: m.reconfiguration.as_ref().map(JsonCompatible::convert),
84                    burst: m.burst.map(|burst| v93::BurstState {
85                        burst_size: burst.burst_size,
86                        linger_duration: JsonCompatible::convert(&burst.linger_duration),
87                        // Hydration alone does not establish caught-up readiness.
88                        steady_ready_at: None,
89                    }),
90                }),
91            },
92        },
93    }
94}
95
96#[cfg(test)]
97mod tests {
98    use proptest::prelude::*;
99
100    use super::*;
101
102    proptest! {
103        #[mz_ore::test]
104        fn resets_burst_stamp_and_preserves_record(
105            mut cluster in any::<v92::Cluster>(),
106            mut managed in any::<v92::ManagedCluster>(),
107            mut burst in any::<v92::BurstState>(),
108            hydrated_at in any::<u64>(),
109        ) {
110            for stamp in [Some(hydrated_at), None] {
111                burst.steady_hydrated_at = stamp;
112                managed.burst = Some(burst.clone());
113                cluster.value.config.variant = v92::ClusterVariant::Managed(managed.clone());
114                let old = v92::StateUpdateKind::Cluster(cluster.clone());
115                let actions = upgrade(vec![old.clone()]);
116                prop_assert_eq!(actions.len(), 1);
117                let MigrationAction::Update(retracted, inserted) = &actions[0] else {
118                    panic!("expected a cluster update");
119                };
120                prop_assert_eq!(retracted, &old);
121                let mut expected = serde_json::to_value(&old).expect("serializable");
122                let burst_json = expected["value"]["config"]["variant"]["Managed"]["burst"]
123                    .as_object_mut().expect("burst object");
124                burst_json.remove("steady_hydrated_at");
125                burst_json.insert("steady_ready_at".into(), serde_json::Value::Null);
126                prop_assert_eq!(serde_json::to_value(inserted).expect("serializable"), expected);
127            }
128        }
129
130        #[mz_ore::test]
131        fn leaves_records_without_bursts_unchanged(
132            mut cluster in any::<v92::Cluster>(),
133            mut managed in any::<v92::ManagedCluster>(),
134            replica in any::<v92::ClusterReplica>(),
135        ) {
136            managed.burst = None;
137            cluster.value.config.variant = v92::ClusterVariant::Managed(managed);
138            let managed = v92::StateUpdateKind::Cluster(cluster.clone());
139            cluster.value.config.variant = v92::ClusterVariant::Unmanaged;
140            let unmanaged = v92::StateUpdateKind::Cluster(cluster);
141            let replica = v92::StateUpdateKind::ClusterReplica(replica);
142            for old in [&managed, &unmanaged, &replica] {
143                let encoded = serde_json::to_vec(old).expect("serializable");
144                let new: v93::StateUpdateKind = serde_json::from_slice(&encoded).expect("compatible");
145                prop_assert_eq!(serde_json::to_vec(&new).expect("serializable"), encoded);
146            }
147            prop_assert!(upgrade(vec![managed, unmanaged, replica]).is_empty());
148        }
149    }
150}
mz_catalog/durable/upgrade/v92_to_v93.rs

mz_catalog/durable/upgrade/
v92_to_v93.rs
