mz_adapter/catalog/
cluster_state.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//! Projects a managed cluster's durable catalog config into the
11//! [`ExpectedClusterState`] compare-and-append witness, and checks a witness
12//! against the current config.
13//!
14//! [`project_expected`] is the one projection from catalog config to the
15//! witness. Building the witness the same way wherever a write is conditioned
16//! and wherever it is checked keeps the compared fields from drifting apart.
17
18use mz_adapter_types::cluster_state::{
19    AutoScalingPolicy, AvailabilityZones, BurstRecord, ClusterSchedule, ExpectedClusterState,
20    OnHydrationPolicy, OnTimeout, ReconfigurationRecord, ReconfigurationStatus,
21    ReconfigurationTarget,
22};
23use mz_catalog::memory::objects::{
24    BurstState, ClusterVariant, ClusterVariantManaged, ReconfigurationState,
25};
26use mz_controller_types::ClusterId;
27use mz_sql::plan::OnTimeoutAction;
28
29use crate::catalog::CatalogState;
30
31/// Project a managed cluster's durable config into the compare-and-append
32/// witness: the fields a conditional write is conditioned on.
33pub(crate) fn project_expected(managed: &ClusterVariantManaged) -> ExpectedClusterState {
34    // Exhaustive destructure (no `..`): a field added to the managed config is a
35    // compile error here until we decide whether the witness must cover it.
36    let ClusterVariantManaged {
37        size,
38        availability_zones,
39        logging,
40        arrangement_compression,
41        replication_factor,
42        optimizer_feature_overrides: _,
43        schedule,
44        auto_scaling_strategy,
45        reconfiguration,
46        burst,
47    } = managed;
48    ExpectedClusterState {
49        size: size.clone(),
50        replication_factor: *replication_factor,
51        availability_zones: AvailabilityZones(availability_zones.clone()),
52        logging: logging.clone(),
53        arrangement_compression: *arrangement_compression,
54        schedule: cluster_schedule(schedule),
55        auto_scaling_policy: auto_scaling_strategy.as_ref().map(auto_scaling_policy),
56        reconfiguration: reconfiguration.as_ref().map(reconfiguration_record),
57        burst: burst.as_ref().map(burst_record),
58    }
59}
60
61/// Whether `cluster_id`'s current managed state still equals `expected`. A
62/// missing or unmanaged cluster never matches. This is the compare half of the
63/// compare-and-append, evaluated inside the catalog transaction so the check and
64/// the commit cannot be separated.
65pub(crate) fn cluster_matches_expected(
66    state: &CatalogState,
67    cluster_id: ClusterId,
68    expected: &ExpectedClusterState,
69) -> bool {
70    let Some(cluster) = state.try_get_cluster(cluster_id) else {
71        return false;
72    };
73    let ClusterVariant::Managed(managed) = &cluster.config.variant else {
74        return false;
75    };
76    project_expected(managed) == *expected
77}
78
79fn reconfiguration_record(record: &ReconfigurationState) -> ReconfigurationRecord {
80    // Exhaustive destructure (no `..`), like `project_expected`: a field added
81    // to either catalog type is a compile error here until we decide whether the
82    // witness must carry it.
83    let ReconfigurationState {
84        target,
85        deadline,
86        on_timeout: on_timeout_action,
87        status,
88    } = record;
89    let mz_catalog::memory::objects::ReconfigurationTarget {
90        size,
91        replication_factor,
92        availability_zones,
93        logging,
94        arrangement_compression,
95    } = target;
96    ReconfigurationRecord {
97        target: ReconfigurationTarget {
98            size: size.clone(),
99            replication_factor: *replication_factor,
100            availability_zones: AvailabilityZones(availability_zones.clone()),
101            logging: logging.clone(),
102            arrangement_compression: *arrangement_compression,
103        },
104        deadline: *deadline,
105        on_timeout: on_timeout(*on_timeout_action),
106        status: reconfiguration_status(*status),
107    }
108}
109
110fn reconfiguration_status(
111    status: mz_catalog::memory::objects::ReconfigurationStatus,
112) -> ReconfigurationStatus {
113    match status {
114        mz_catalog::memory::objects::ReconfigurationStatus::InProgress => {
115            ReconfigurationStatus::InProgress
116        }
117        mz_catalog::memory::objects::ReconfigurationStatus::Finalized => {
118            ReconfigurationStatus::Finalized
119        }
120        mz_catalog::memory::objects::ReconfigurationStatus::TimedOut => {
121            ReconfigurationStatus::TimedOut
122        }
123        mz_catalog::memory::objects::ReconfigurationStatus::Cancelled => {
124            ReconfigurationStatus::Cancelled
125        }
126        mz_catalog::memory::objects::ReconfigurationStatus::ResourceExhausted => {
127            ReconfigurationStatus::ResourceExhausted
128        }
129    }
130}
131
132fn on_timeout(action: OnTimeoutAction) -> OnTimeout {
133    match action {
134        OnTimeoutAction::Commit => OnTimeout::Commit,
135        OnTimeoutAction::Rollback => OnTimeout::Rollback,
136    }
137}
138
139fn cluster_schedule(schedule: &mz_sql::plan::ClusterSchedule) -> ClusterSchedule {
140    match schedule {
141        mz_sql::plan::ClusterSchedule::Manual => ClusterSchedule::Manual,
142        mz_sql::plan::ClusterSchedule::Refresh {
143            hydration_time_estimate,
144        } => ClusterSchedule::Refresh {
145            hydration_time_estimate: *hydration_time_estimate,
146        },
147    }
148}
149
150fn burst_record(record: &BurstState) -> BurstRecord {
151    // Exhaustive destructure (no `..`), like `project_expected`: a field added
152    // to the catalog type is a compile error here until the witness accounts for
153    // it.
154    let BurstState {
155        burst_size,
156        linger_duration,
157        steady_hydrated_at,
158    } = record;
159    BurstRecord {
160        burst_size: burst_size.clone(),
161        linger_duration: *linger_duration,
162        steady_hydrated_at: *steady_hydrated_at,
163    }
164}
165
166fn auto_scaling_policy(strategy: &mz_sql::plan::AutoScalingStrategy) -> AutoScalingPolicy {
167    let mz_sql::plan::AutoScalingStrategy { on_hydration } = strategy;
168    AutoScalingPolicy {
169        on_hydration: on_hydration.as_ref().map(|on_hydration| {
170            let mz_sql::plan::OnHydration {
171                hydration_size,
172                linger_duration,
173            } = on_hydration;
174            OnHydrationPolicy {
175                hydration_size: hydration_size.clone(),
176                linger_duration: *linger_duration,
177            }
178        }),
179    }
180}
mz_adapter/catalog/cluster_state.rs

mz_adapter/catalog/
cluster_state.rs
