mz_deploy/project/
clusters.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//! Cluster definition loading and validation.
11//!
12//! Loads cluster definitions from `<root>/clusters/` directory. Each `.sql` file
13//! defines a single cluster with a required `CREATE CLUSTER` statement and optional
14//! `GRANT` and `COMMENT` statements.
15
16use crate::project::error::{
17    LoadError, ProjectError, ValidationError, ValidationErrorKind, ValidationErrors,
18};
19use crate::project::syntax::parser::{
20    LocatedStatement, parse_statements_with_context, statement_type_name,
21};
22use crate::project::syntax::profile_files::collect_all_sql_files;
23use mz_sql_parser::ast::{
24    CommentObjectType, CommentStatement, CreateClusterStatement, GrantPrivilegesStatement,
25    GrantTargetSpecification, GrantTargetSpecificationInner, Ident, ObjectType, Raw,
26    RawClusterName, Statement, UnresolvedObjectName,
27};
28use std::collections::BTreeMap;
29use std::path::Path;
30
31/// A parsed cluster definition from a `.sql` file in the `clusters/` directory.
32pub(crate) struct ClusterDefinition {
33    /// Cluster name (derived from filename and validated against CREATE statement).
34    pub name: String,
35    /// The CREATE CLUSTER statement.
36    pub create_stmt: CreateClusterStatement<Raw>,
37    /// Optional GRANT statements targeting this cluster.
38    pub grants: Vec<GrantPrivilegesStatement<Raw>>,
39    /// Optional COMMENT statements targeting this cluster.
40    pub comments: Vec<CommentStatement<Raw>>,
41}
42
43/// Load all cluster definitions from `<root>/clusters/`.
44///
45/// Returns an empty vec if `clusters/` doesn't exist (the directory is optional).
46/// If `profile_suffix` is provided, each cluster definition is rewritten with the
47/// suffix appended to all cluster name references (CREATE, GRANT, COMMENT).
48pub(crate) fn load_clusters(
49    root: &Path,
50    profile: &str,
51    profile_suffix: Option<&str>,
52    variables: &BTreeMap<String, String>,
53) -> Result<Vec<ClusterDefinition>, ProjectError> {
54    let clusters_dir = root.join("clusters");
55
56    if !clusters_dir.exists() {
57        return Ok(vec![]);
58    }
59
60    if !clusters_dir.is_dir() {
61        return Err(LoadError::RootNotDirectory { path: clusters_dir }.into());
62    }
63
64    let all_files = collect_all_sql_files(&clusters_dir)?;
65
66    let mut definitions = Vec::new();
67    let mut errors = Vec::new();
68
69    for object_files in all_files {
70        let expected_name = &object_files.name;
71
72        // Validate all variants independently
73        let mut all_variant_paths = Vec::new();
74        if let Some(ref default_path) = object_files.default {
75            all_variant_paths.push((default_path.clone(), None));
76        }
77        for (prof, override_path) in &object_files.overrides {
78            all_variant_paths.push((override_path.clone(), Some(prof.as_str())));
79        }
80
81        // Validate each variant independently
82        for (path, _) in &all_variant_paths {
83            let sql = std::fs::read_to_string(path).map_err(|e| LoadError::FileReadFailed {
84                path: path.clone(),
85                source: e,
86            })?;
87            let located = parse_statements_with_context(&sql, path.clone(), variables, true)?;
88
89            if let Err(mut errs) = classify_cluster_statements(expected_name, path, located) {
90                errors.append(&mut errs);
91            }
92        }
93
94        // Resolve the active variant: prefer profile match, fall back to default
95        let active_path = object_files
96            .overrides
97            .get(profile)
98            .or(object_files.default.as_ref());
99
100        let active_path = match active_path {
101            Some(p) => p.clone(),
102            None => continue, // no variant for this profile
103        };
104
105        let sql = std::fs::read_to_string(&active_path).map_err(|e| LoadError::FileReadFailed {
106            path: active_path.clone(),
107            source: e,
108        })?;
109        let located = parse_statements_with_context(&sql, active_path.clone(), variables, true)?;
110
111        match classify_cluster_statements(expected_name, &active_path, located) {
112            Ok(def) => definitions.push(def),
113            Err(mut errs) => errors.append(&mut errs),
114        }
115    }
116
117    if !errors.is_empty() {
118        return Err(ValidationErrors::new(errors).into());
119    }
120
121    // Apply cluster suffix after validation (filename-vs-declared-name check uses original names)
122    if let Some(suffix) = profile_suffix {
123        for def in &mut definitions {
124            apply_cluster_suffix(def, suffix);
125        }
126    }
127
128    Ok(definitions)
129}
130
131/// Classify parsed statements into a `ClusterDefinition`, returning validation errors.
132fn classify_cluster_statements(
133    expected_name: &str,
134    path: &Path,
135    located_statements: Vec<LocatedStatement>,
136) -> Result<ClusterDefinition, Vec<ValidationError>> {
137    let mut create_stmts: Vec<(CreateClusterStatement<Raw>, usize)> = Vec::new();
138    let mut grants: Vec<GrantPrivilegesStatement<Raw>> = Vec::new();
139    let mut comments: Vec<CommentStatement<Raw>> = Vec::new();
140    let mut errors = Vec::new();
141
142    for LocatedStatement {
143        ast: stmt,
144        byte_offset,
145    } in located_statements
146    {
147        match stmt {
148            Statement::CreateCluster(s) => {
149                create_stmts.push((s, byte_offset));
150            }
151            Statement::GrantPrivileges(s) => {
152                // Validate that the grant targets a cluster
153                match &s.target {
154                    GrantTargetSpecification::Object {
155                        object_type: ObjectType::Cluster,
156                        object_spec_inner: GrantTargetSpecificationInner::Objects { names },
157                    } => {
158                        // Validate cluster name matches
159                        for name in names {
160                            let target_name = name.to_string();
161                            if target_name.to_lowercase() != expected_name.to_lowercase() {
162                                errors.push(ValidationError::with_file_sql_and_offset(
163                                    ValidationErrorKind::ClusterGrantTargetMismatch {
164                                        target: target_name,
165                                        cluster_name: expected_name.to_string(),
166                                    },
167                                    path.to_path_buf(),
168                                    s.to_string(),
169                                    byte_offset,
170                                ));
171                            }
172                        }
173                        grants.push(s);
174                    }
175                    _ => {
176                        errors.push(ValidationError::with_file_sql_and_offset(
177                            ValidationErrorKind::InvalidClusterStatement {
178                                statement_type: "GRANT (not targeting a cluster)".to_string(),
179                                cluster_name: expected_name.to_string(),
180                            },
181                            path.to_path_buf(),
182                            s.to_string(),
183                            byte_offset,
184                        ));
185                    }
186                }
187            }
188            Statement::Comment(s) => {
189                // Validate that the comment targets a cluster
190                match &s.object {
191                    CommentObjectType::Cluster { name } => {
192                        let target_name = match name {
193                            RawClusterName::Unresolved(ident) => ident.to_string(),
194                            RawClusterName::Resolved(id) => id.clone(),
195                        };
196                        if target_name.to_lowercase() != expected_name.to_lowercase() {
197                            errors.push(ValidationError::with_file_sql_and_offset(
198                                ValidationErrorKind::ClusterCommentTargetMismatch {
199                                    target: target_name,
200                                    cluster_name: expected_name.to_string(),
201                                },
202                                path.to_path_buf(),
203                                s.to_string(),
204                                byte_offset,
205                            ));
206                        }
207                        comments.push(s);
208                    }
209                    _ => {
210                        errors.push(ValidationError::with_file_sql_and_offset(
211                            ValidationErrorKind::InvalidClusterStatement {
212                                statement_type: "COMMENT (not targeting a cluster)".to_string(),
213                                cluster_name: expected_name.to_string(),
214                            },
215                            path.to_path_buf(),
216                            s.to_string(),
217                            byte_offset,
218                        ));
219                    }
220                }
221            }
222            other => {
223                errors.push(ValidationError::with_file_sql_and_offset(
224                    ValidationErrorKind::InvalidClusterStatement {
225                        statement_type: statement_type_name(&other).to_string(),
226                        cluster_name: expected_name.to_string(),
227                    },
228                    path.to_path_buf(),
229                    other.to_string(),
230                    byte_offset,
231                ));
232            }
233        }
234    }
235
236    // Validate exactly one CREATE CLUSTER (file-level errors)
237    if create_stmts.is_empty() {
238        errors.push(ValidationError::with_file(
239            ValidationErrorKind::ClusterMissingCreateStatement {
240                cluster_name: expected_name.to_string(),
241            },
242            path.to_path_buf(),
243        ));
244    } else if create_stmts.len() > 1 {
245        // Point to the second CREATE CLUSTER
246        errors.push(ValidationError::with_file_and_offset(
247            ValidationErrorKind::ClusterMultipleCreateStatements {
248                cluster_name: expected_name.to_string(),
249            },
250            path.to_path_buf(),
251            create_stmts[1].1,
252        ));
253    }
254
255    if !errors.is_empty() {
256        return Err(errors);
257    }
258
259    let (create_stmt, create_offset) = create_stmts.into_iter().next().unwrap();
260
261    // Validate cluster name matches filename
262    let declared_name = create_stmt.name.to_string();
263    if declared_name.to_lowercase() != expected_name.to_lowercase() {
264        return Err(vec![ValidationError::with_file_and_offset(
265            ValidationErrorKind::ClusterNameMismatch {
266                declared: declared_name,
267                expected: expected_name.to_string(),
268            },
269            path.to_path_buf(),
270            create_offset,
271        )]);
272    }
273
274    Ok(ClusterDefinition {
275        name: expected_name.to_string(),
276        create_stmt,
277        grants,
278        comments,
279    })
280}
281
282/// Apply a suffix to all cluster name references within a `ClusterDefinition`.
283///
284/// Rewrites the definition name, the CREATE statement name, GRANT target names,
285/// and COMMENT target names.
286fn apply_cluster_suffix(def: &mut ClusterDefinition, suffix: &str) {
287    // Rewrite definition name first, then reference it for the CREATE statement
288    def.name = format!("{}{}", def.name, suffix);
289    def.create_stmt.name = Ident::new(&def.name).expect("valid cluster identifier");
290
291    // Rewrite GRANT target cluster names
292    for grant in &mut def.grants {
293        if let GrantTargetSpecification::Object {
294            object_type: ObjectType::Cluster,
295            object_spec_inner: GrantTargetSpecificationInner::Objects { names },
296        } = &mut grant.target
297        {
298            for name in names {
299                if let UnresolvedObjectName::Cluster(ident) = name {
300                    *ident = suffixed_ident(ident, suffix);
301                }
302            }
303        }
304    }
305
306    // Rewrite COMMENT target cluster names
307    for comment in &mut def.comments {
308        if let CommentObjectType::Cluster { name } = &mut comment.object {
309            if let RawClusterName::Unresolved(ident) = name {
310                *ident = suffixed_ident(ident, suffix);
311            }
312        }
313    }
314}
315
316/// Append a suffix to an `Ident`, returning a new `Ident`.
317fn suffixed_ident(ident: &Ident, suffix: &str) -> Ident {
318    Ident::new(&format!("{}{}", ident, suffix)).expect("valid cluster identifier")
319}
320
321#[cfg(test)]
322mod tests {
323    use super::*;
324    use mz_sql_parser::ast::display::AstDisplay;
325    use std::fs;
326    use tempfile::TempDir;
327
328    fn create_test_dir() -> TempDir {
329        TempDir::new().unwrap()
330    }
331
332    /// The loaded definition's `CREATE CLUSTER`, rendered back to SQL.
333    fn create_sql(def: &ClusterDefinition) -> String {
334        def.create_stmt.to_ast_string_simple()
335    }
336
337    #[mz_ore::test]
338    fn test_load_clusters_no_directory() {
339        let dir = create_test_dir();
340        let result = load_clusters(dir.path(), "default", None, &BTreeMap::new()).unwrap();
341        assert!(
342            result.is_empty(),
343            "should return empty vec when clusters/ doesn't exist"
344        );
345    }
346
347    #[mz_ore::test]
348    fn test_load_clusters_basic() {
349        let dir = create_test_dir();
350        let clusters_dir = dir.path().join("clusters");
351        fs::create_dir(&clusters_dir).unwrap();
352
353        fs::write(
354            clusters_dir.join("analytics.sql"),
355            "CREATE CLUSTER analytics (SIZE = '100cc', REPLICATION FACTOR = 1);\n\
356             GRANT USAGE ON CLUSTER analytics TO analyst_role;\n\
357             COMMENT ON CLUSTER analytics IS 'Analytics workloads';",
358        )
359        .unwrap();
360
361        let result = load_clusters(dir.path(), "default", None, &BTreeMap::new()).unwrap();
362        assert_eq!(result.len(), 1);
363        assert_eq!(result[0].name, "analytics");
364        assert_eq!(result[0].grants.len(), 1);
365        assert_eq!(result[0].comments.len(), 1);
366
367        assert_eq!(
368            create_sql(&result[0]),
369            "CREATE CLUSTER analytics (SIZE = '100cc', REPLICATION FACTOR = 1)"
370        );
371    }
372
373    #[mz_ore::test]
374    fn test_load_clusters_create_only() {
375        let dir = create_test_dir();
376        let clusters_dir = dir.path().join("clusters");
377        fs::create_dir(&clusters_dir).unwrap();
378
379        fs::write(
380            clusters_dir.join("quickstart.sql"),
381            "CREATE CLUSTER quickstart (SIZE = '25cc');",
382        )
383        .unwrap();
384
385        let result = load_clusters(dir.path(), "default", None, &BTreeMap::new()).unwrap();
386        assert_eq!(result.len(), 1);
387        assert_eq!(result[0].name, "quickstart");
388        assert!(result[0].grants.is_empty());
389        assert!(result[0].comments.is_empty());
390    }
391
392    #[mz_ore::test]
393    fn test_load_clusters_name_mismatch() {
394        let dir = create_test_dir();
395        let clusters_dir = dir.path().join("clusters");
396        fs::create_dir(&clusters_dir).unwrap();
397
398        fs::write(
399            clusters_dir.join("analytics.sql"),
400            "CREATE CLUSTER wrong_name (SIZE = '100cc');",
401        )
402        .unwrap();
403
404        let result = load_clusters(dir.path(), "default", None, &BTreeMap::new());
405        assert!(
406            result.is_err(),
407            "should error when cluster name doesn't match filename"
408        );
409    }
410
411    #[mz_ore::test]
412    fn test_load_clusters_missing_create() {
413        let dir = create_test_dir();
414        let clusters_dir = dir.path().join("clusters");
415        fs::create_dir(&clusters_dir).unwrap();
416
417        fs::write(
418            clusters_dir.join("analytics.sql"),
419            "GRANT USAGE ON CLUSTER analytics TO analyst_role;",
420        )
421        .unwrap();
422
423        let result = load_clusters(dir.path(), "default", None, &BTreeMap::new());
424        assert!(
425            result.is_err(),
426            "should error when no CREATE CLUSTER statement"
427        );
428    }
429
430    #[mz_ore::test]
431    fn test_load_clusters_unsupported_statement() {
432        let dir = create_test_dir();
433        let clusters_dir = dir.path().join("clusters");
434        fs::create_dir(&clusters_dir).unwrap();
435
436        fs::write(
437            clusters_dir.join("analytics.sql"),
438            "CREATE CLUSTER analytics (SIZE = '100cc');\n\
439             CREATE TABLE foo (id INT);",
440        )
441        .unwrap();
442
443        let result = load_clusters(dir.path(), "default", None, &BTreeMap::new());
444        assert!(
445            result.is_err(),
446            "should error on unsupported statement type"
447        );
448    }
449
450    #[mz_ore::test]
451    fn test_load_clusters_grant_target_mismatch() {
452        let dir = create_test_dir();
453        let clusters_dir = dir.path().join("clusters");
454        fs::create_dir(&clusters_dir).unwrap();
455
456        fs::write(
457            clusters_dir.join("analytics.sql"),
458            "CREATE CLUSTER analytics (SIZE = '100cc');\n\
459             GRANT USAGE ON CLUSTER other_cluster TO analyst_role;",
460        )
461        .unwrap();
462
463        let result = load_clusters(dir.path(), "default", None, &BTreeMap::new());
464        assert!(
465            result.is_err(),
466            "should error when grant targets wrong cluster"
467        );
468    }
469
470    #[mz_ore::test]
471    fn test_load_clusters_comment_target_mismatch() {
472        let dir = create_test_dir();
473        let clusters_dir = dir.path().join("clusters");
474        fs::create_dir(&clusters_dir).unwrap();
475
476        fs::write(
477            clusters_dir.join("analytics.sql"),
478            "CREATE CLUSTER analytics (SIZE = '100cc');\n\
479             COMMENT ON CLUSTER other_cluster IS 'wrong target';",
480        )
481        .unwrap();
482
483        let result = load_clusters(dir.path(), "default", None, &BTreeMap::new());
484        assert!(
485            result.is_err(),
486            "should error when comment targets wrong cluster"
487        );
488    }
489
490    #[mz_ore::test]
491    fn test_load_clusters_multiple_files() {
492        let dir = create_test_dir();
493        let clusters_dir = dir.path().join("clusters");
494        fs::create_dir(&clusters_dir).unwrap();
495
496        fs::write(
497            clusters_dir.join("analytics.sql"),
498            "CREATE CLUSTER analytics (SIZE = '100cc');",
499        )
500        .unwrap();
501
502        fs::write(
503            clusters_dir.join("quickstart.sql"),
504            "CREATE CLUSTER quickstart (SIZE = '25cc', REPLICATION FACTOR = 2);",
505        )
506        .unwrap();
507
508        let result = load_clusters(dir.path(), "default", None, &BTreeMap::new()).unwrap();
509        assert_eq!(result.len(), 2);
510        // Sorted by filename
511        assert_eq!(result[0].name, "analytics");
512        assert_eq!(result[1].name, "quickstart");
513    }
514
515    #[mz_ore::test]
516    fn test_load_clusters_multi_variant_valid() {
517        let dir = create_test_dir();
518        let clusters_dir = dir.path().join("clusters");
519        fs::create_dir(&clusters_dir).unwrap();
520
521        fs::write(
522            clusters_dir.join("analytics.sql"),
523            "CREATE CLUSTER analytics (SIZE = '100cc');",
524        )
525        .unwrap();
526        fs::write(
527            clusters_dir.join("analytics#staging.sql"),
528            "CREATE CLUSTER analytics (SIZE = '25cc');",
529        )
530        .unwrap();
531
532        // With staging profile, should pick staging variant
533        let result = load_clusters(dir.path(), "staging", None, &BTreeMap::new()).unwrap();
534        assert_eq!(result.len(), 1);
535        assert_eq!(result[0].name, "analytics");
536        assert_eq!(
537            create_sql(&result[0]),
538            "CREATE CLUSTER analytics (SIZE = '25cc')"
539        );
540    }
541
542    #[mz_ore::test]
543    fn test_load_clusters_multi_variant_fallback_default() {
544        let dir = create_test_dir();
545        let clusters_dir = dir.path().join("clusters");
546        fs::create_dir(&clusters_dir).unwrap();
547
548        fs::write(
549            clusters_dir.join("analytics.sql"),
550            "CREATE CLUSTER analytics (SIZE = '100cc');",
551        )
552        .unwrap();
553        fs::write(
554            clusters_dir.join("analytics#staging.sql"),
555            "CREATE CLUSTER analytics (SIZE = '25cc');",
556        )
557        .unwrap();
558
559        // With prod profile (no match), should fall back to default
560        let result = load_clusters(dir.path(), "prod", None, &BTreeMap::new()).unwrap();
561        assert_eq!(result.len(), 1);
562        assert_eq!(result[0].name, "analytics");
563        assert_eq!(
564            create_sql(&result[0]),
565            "CREATE CLUSTER analytics (SIZE = '100cc')"
566        );
567    }
568
569    #[mz_ore::test]
570    fn test_load_clusters_auto_scaling_strategy() {
571        let dir = create_test_dir();
572        let clusters_dir = dir.path().join("clusters");
573        fs::create_dir(&clusters_dir).unwrap();
574
575        fs::write(
576            clusters_dir.join("analytics.sql"),
577            "CREATE CLUSTER analytics (SIZE = '100cc', AUTO SCALING STRATEGY = \
578             (ON HYDRATION (HYDRATION SIZE = '3200cc', LINGER DURATION = '600s')));",
579        )
580        .unwrap();
581
582        let result = load_clusters(dir.path(), "default", None, &BTreeMap::new()).unwrap();
583        assert_eq!(result.len(), 1);
584        assert_eq!(
585            create_sql(&result[0]),
586            "CREATE CLUSTER analytics (SIZE = '100cc', AUTO SCALING STRATEGY = \
587             (ON HYDRATION (HYDRATION SIZE = '3200cc', LINGER DURATION = '600s')))"
588        );
589    }
590
591    #[mz_ore::test]
592    fn test_load_clusters_invalid_variant_errors() {
593        let dir = create_test_dir();
594        let clusters_dir = dir.path().join("clusters");
595        fs::create_dir(&clusters_dir).unwrap();
596
597        fs::write(
598            clusters_dir.join("analytics.sql"),
599            "CREATE CLUSTER analytics (SIZE = '100cc');",
600        )
601        .unwrap();
602        // Invalid staging variant: name mismatch
603        fs::write(
604            clusters_dir.join("analytics#staging.sql"),
605            "CREATE CLUSTER wrong_name (SIZE = '25cc');",
606        )
607        .unwrap();
608
609        let result = load_clusters(dir.path(), "default", None, &BTreeMap::new());
610        assert!(
611            result.is_err(),
612            "invalid variant should error even when not active profile"
613        );
614    }
615}
mz_deploy/project/clusters.rs

mz_deploy/project/
clusters.rs
