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::connection::ProvisioningClient;
31use crate::client::errors::ConnectionError;
32use crate::client::models::ClusterConfig;
33use crate::client::quote_identifier;
34use mz_sql_parser::ast::Ident;
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 cluster from a captured cluster configuration.
75    pub async fn create_cluster_with_config(
76        &self,
77        name: &str,
78        config: &ClusterConfig,
79    ) -> Result<(), ConnectionError> {
80        let grants = match config {
81            ClusterConfig::Managed {
82                create_stmt,
83                grants,
84            } => {
85                let mut create_stmt = create_stmt.clone();
86                create_stmt.name =
87                    Ident::new(name).map_err(|e| ConnectionError::ClusterCreationFailed {
88                        name: name.to_string(),
89                        source: Box::new(e),
90                    })?;
91
92                self.client
93                    .execute(&create_stmt.to_ast_string_simple(), &[])
94                    .await
95                    .map_err(|e| {
96                        if e.to_string().contains("already exists") {
97                            ConnectionError::ClusterAlreadyExists {
98                                name: name.to_string(),
99                            }
100                        } else {
101                            ConnectionError::ClusterCreationFailed {
102                                name: name.to_string(),
103                                source: Box::new(e),
104                            }
105                        }
106                    })?;
107
108                grants
109            }
110            ClusterConfig::Unmanaged { replicas, grants } => {
111                let create_cluster_sql =
112                    format!("CREATE CLUSTER {} REPLICAS ()", quote_identifier(name));
113
114                self.client
115                    .execute(&create_cluster_sql, &[])
116                    .await
117                    .map_err(|e| {
118                        if e.to_string().contains("already exists") {
119                            ConnectionError::ClusterAlreadyExists {
120                                name: name.to_string(),
121                            }
122                        } else {
123                            ConnectionError::ClusterCreationFailed {
124                                name: name.to_string(),
125                                source: Box::new(e),
126                            }
127                        }
128                    })?;
129
130                for replica in replicas {
131                    let mut options_parts = vec![format!("SIZE = '{}'", replica.size)];
132
133                    if let Some(ref az) = replica.availability_zone {
134                        options_parts.push(format!("AVAILABILITY ZONE '{}'", az));
135                    }
136
137                    let create_replica_sql = format!(
138                        "CREATE CLUSTER REPLICA {}.{} ({})",
139                        quote_identifier(name),
140                        quote_identifier(&replica.name),
141                        options_parts.join(", ")
142                    );
143
144                    self.client
145                        .execute(&create_replica_sql, &[])
146                        .await
147                        .map_err(|e| ConnectionError::ClusterCreationFailed {
148                            name: format!("{}.{}", name, replica.name),
149                            source: Box::new(e),
150                        })?;
151                }
152
153                grants
154            }
155        };
156
157        for grant in grants {
158            let sql = format!(
159                "GRANT {} ON CLUSTER {} TO {}",
160                grant.privilege_type,
161                quote_identifier(name),
162                quote_identifier(&grant.grantee)
163            );
164            self.client.execute(&sql, &[]).await.map_err(|e| {
165                ConnectionError::Message(format!(
166                    "Failed to grant {} to {} on cluster '{}': {}",
167                    grant.privilege_type, grant.grantee, name, e
168                ))
169            })?;
170        }
171
172        Ok(())
173    }
174}
mz_deploy/client/provisioning.rs

mz_deploy/client/
provisioning.rs
