mz_deploy/client/
provisioning.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//! DDL provisioning operations.
11//!
12//! Methods on [`ProvisioningClient`] issue `CREATE … IF NOT EXISTS` and
13//! `ALTER` statements to ensure that the target region's databases, schemas,
14//! and clusters match the project definition. These are idempotent and run
15//! before any object-level deployment.
16//!
17//! ## Ordering
18//!
19//! Provisioning must follow referential order: **databases → schemas → clusters**.
20//! A schema cannot be created until its parent database exists, so callers
21//! (e.g., [`super::super::cli::executor::DeploymentExecutor`]) are responsible
22//! for invoking provisioning methods in the correct order.
23//!
24//! ## Idempotency
25//!
26//! All `create_*` methods use `IF NOT EXISTS` (or catch "already exists" errors)
27//! so that re-running provisioning on an already-provisioned environment is a
28//! no-op.
29
30use crate::client::auto_scaling::strategy_to_cluster_option;
31use crate::client::connection::ProvisioningClient;
32use crate::client::errors::ConnectionError;
33use crate::client::models::{ClusterConfig, ClusterOptions};
34use crate::client::quote_identifier;
35use mz_sql_parser::ast::display::AstDisplay;
36
37impl ProvisioningClient<'_> {
38    /// Create a database if it does not already exist.
39    pub async fn create_database(&self, database: &str) -> Result<(), ConnectionError> {
40        let sql = format!(
41            "CREATE DATABASE IF NOT EXISTS {}",
42            quote_identifier(database)
43        );
44
45        self.client.execute(&sql, &[]).await.map_err(|e| {
46            ConnectionError::DatabaseCreationFailed {
47                database: database.to_string(),
48                source: Box::new(e),
49            }
50        })?;
51
52        Ok(())
53    }
54
55    /// Create a schema in the specified database if it does not already exist.
56    pub async fn create_schema(&self, database: &str, schema: &str) -> Result<(), ConnectionError> {
57        let sql = format!(
58            "CREATE SCHEMA IF NOT EXISTS {}.{}",
59            quote_identifier(database),
60            quote_identifier(schema)
61        );
62
63        self.client.execute(&sql, &[]).await.map_err(|e| {
64            ConnectionError::SchemaCreationFailed {
65                database: database.to_string(),
66                schema: schema.to_string(),
67                source: Box::new(e),
68            }
69        })?;
70
71        Ok(())
72    }
73
74    /// Create a managed cluster with the requested size, replication factor,
75    /// and autoscaling policy.
76    pub async fn create_cluster(
77        &self,
78        name: &str,
79        options: &ClusterOptions,
80    ) -> Result<(), ConnectionError> {
81        let mut sql = format!(
82            "CREATE CLUSTER {} (SIZE = '{}', REPLICATION FACTOR = {}",
83            quote_identifier(name),
84            options.size,
85            options.replication_factor
86        );
87        if let Some(strategy) = &options.auto_scaling_strategy {
88            sql.push_str(", ");
89            sql.push_str(&strategy_to_cluster_option(strategy).to_ast_string_simple());
90        }
91        sql.push(')');
92
93        self.client.execute(&sql, &[]).await.map_err(|e| {
94            if e.to_string().contains("already exists") {
95                ConnectionError::ClusterAlreadyExists {
96                    name: name.to_string(),
97                }
98            } else {
99                ConnectionError::ClusterCreationFailed {
100                    name: name.to_string(),
101                    source: Box::new(e),
102                }
103            }
104        })?;
105
106        Ok(())
107    }
108
109    /// Create a cluster from a captured cluster configuration.
110    pub async fn create_cluster_with_config(
111        &self,
112        name: &str,
113        config: &ClusterConfig,
114    ) -> Result<(), ConnectionError> {
115        let grants = match config {
116            ClusterConfig::Managed { options, grants } => {
117                self.create_cluster(name, options).await?;
118                grants
119            }
120            ClusterConfig::Unmanaged { replicas, grants } => {
121                let create_cluster_sql =
122                    format!("CREATE CLUSTER {} REPLICAS ()", quote_identifier(name));
123
124                self.client
125                    .execute(&create_cluster_sql, &[])
126                    .await
127                    .map_err(|e| {
128                        if e.to_string().contains("already exists") {
129                            ConnectionError::ClusterAlreadyExists {
130                                name: name.to_string(),
131                            }
132                        } else {
133                            ConnectionError::ClusterCreationFailed {
134                                name: name.to_string(),
135                                source: Box::new(e),
136                            }
137                        }
138                    })?;
139
140                for replica in replicas {
141                    let mut options_parts = vec![format!("SIZE = '{}'", replica.size)];
142
143                    if let Some(ref az) = replica.availability_zone {
144                        options_parts.push(format!("AVAILABILITY ZONE '{}'", az));
145                    }
146
147                    let create_replica_sql = format!(
148                        "CREATE CLUSTER REPLICA {}.{} ({})",
149                        quote_identifier(name),
150                        quote_identifier(&replica.name),
151                        options_parts.join(", ")
152                    );
153
154                    self.client
155                        .execute(&create_replica_sql, &[])
156                        .await
157                        .map_err(|e| ConnectionError::ClusterCreationFailed {
158                            name: format!("{}.{}", name, replica.name),
159                            source: Box::new(e),
160                        })?;
161                }
162
163                grants
164            }
165        };
166
167        for grant in grants {
168            let sql = format!(
169                "GRANT {} ON CLUSTER {} TO {}",
170                grant.privilege_type,
171                quote_identifier(name),
172                quote_identifier(&grant.grantee)
173            );
174            self.client.execute(&sql, &[]).await.map_err(|e| {
175                ConnectionError::Message(format!(
176                    "Failed to grant {} to {} on cluster '{}': {}",
177                    grant.privilege_type, grant.grantee, name, e
178                ))
179            })?;
180        }
181
182        Ok(())
183    }
184}
mz_deploy/client/provisioning.rs

mz_deploy/client/
provisioning.rs
