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    AvailabilityZones, BurstRecord, ExpectedClusterState, OnTimeout, ReconfigurationRecord,
20    ReconfigurationStatus, ReconfigurationTarget,
21};
22use mz_catalog::memory::objects::{
23    BurstState, ClusterVariant, ClusterVariantManaged, ReconfigurationState,
24};
25use mz_controller_types::ClusterId;
26use mz_sql::plan::OnTimeoutAction;
27
28use crate::catalog::CatalogState;
29
30/// Project a managed cluster's durable config into the compare-and-append
31/// witness: the fields a conditional write is conditioned on.
32pub(crate) fn project_expected(managed: &ClusterVariantManaged) -> ExpectedClusterState {
33    // Exhaustive destructure (no `..`): a field added to the managed config is a
34    // compile error here until we decide whether the witness must cover it.
35    let ClusterVariantManaged {
36        size,
37        availability_zones,
38        logging,
39        replication_factor,
40        optimizer_feature_overrides: _,
41        schedule: _,
42        auto_scaling_strategy: _,
43        reconfiguration,
44        burst,
45    } = managed;
46    ExpectedClusterState {
47        size: size.clone(),
48        replication_factor: *replication_factor,
49        availability_zones: AvailabilityZones(availability_zones.clone()),
50        logging: logging.clone(),
51        reconfiguration: reconfiguration.as_ref().map(reconfiguration_record),
52        burst: burst.as_ref().map(burst_record),
53    }
54}
55
56/// Whether `cluster_id`'s current managed state still equals `expected`. A
57/// missing or unmanaged cluster never matches. This is the compare half of the
58/// compare-and-append, evaluated inside the catalog transaction so the check and
59/// the commit cannot be separated.
60pub(crate) fn cluster_matches_expected(
61    state: &CatalogState,
62    cluster_id: ClusterId,
63    expected: &ExpectedClusterState,
64) -> bool {
65    let Some(cluster) = state.try_get_cluster(cluster_id) else {
66        return false;
67    };
68    let ClusterVariant::Managed(managed) = &cluster.config.variant else {
69        return false;
70    };
71    project_expected(managed) == *expected
72}
73
74fn reconfiguration_record(record: &ReconfigurationState) -> ReconfigurationRecord {
75    // Exhaustive destructure (no `..`), like `project_expected`: a field added
76    // to either catalog type is a compile error here until we decide whether the
77    // witness must carry it.
78    let ReconfigurationState {
79        target,
80        deadline,
81        on_timeout: on_timeout_action,
82        status,
83    } = record;
84    let mz_catalog::memory::objects::ReconfigurationTarget {
85        size,
86        replication_factor,
87        availability_zones,
88        logging,
89    } = target;
90    ReconfigurationRecord {
91        target: ReconfigurationTarget {
92            size: size.clone(),
93            replication_factor: *replication_factor,
94            availability_zones: AvailabilityZones(availability_zones.clone()),
95            logging: logging.clone(),
96        },
97        deadline: *deadline,
98        on_timeout: on_timeout(*on_timeout_action),
99        status: reconfiguration_status(*status),
100    }
101}
102
103fn reconfiguration_status(
104    status: mz_catalog::memory::objects::ReconfigurationStatus,
105) -> ReconfigurationStatus {
106    match status {
107        mz_catalog::memory::objects::ReconfigurationStatus::InProgress => {
108            ReconfigurationStatus::InProgress
109        }
110        mz_catalog::memory::objects::ReconfigurationStatus::Finalized => {
111            ReconfigurationStatus::Finalized
112        }
113        mz_catalog::memory::objects::ReconfigurationStatus::TimedOut => {
114            ReconfigurationStatus::TimedOut
115        }
116        mz_catalog::memory::objects::ReconfigurationStatus::Cancelled => {
117            ReconfigurationStatus::Cancelled
118        }
119        mz_catalog::memory::objects::ReconfigurationStatus::ResourceExhausted => {
120            ReconfigurationStatus::ResourceExhausted
121        }
122    }
123}
124
125fn on_timeout(action: OnTimeoutAction) -> OnTimeout {
126    match action {
127        OnTimeoutAction::Commit => OnTimeout::Commit,
128        OnTimeoutAction::Rollback => OnTimeout::Rollback,
129    }
130}
131
132fn burst_record(record: &BurstState) -> BurstRecord {
133    // Exhaustive destructure (no `..`), like `project_expected`: a field added
134    // to the catalog type is a compile error here until the witness accounts for
135    // it.
136    let BurstState {
137        burst_size,
138        linger_duration,
139        steady_hydrated_at,
140    } = record;
141    BurstRecord {
142        burst_size: burst_size.clone(),
143        linger_duration: *linger_duration,
144        steady_hydrated_at: *steady_hydrated_at,
145    }
146}
mz_adapter/catalog/cluster_state.rs

mz_adapter/catalog/
cluster_state.rs
