mz_deploy/client/
models.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//! Domain models for Materialize catalog objects.
11//!
12//! These types represent objects in the Materialize system catalog and provide
13//! a type-safe interface over raw database rows.
14
15use chrono::{DateTime, Utc};
16use mz_sql::plan::AutoScalingStrategy;
17use std::fmt;
18use std::str::FromStr;
19
20use crate::project::SchemaQualifier;
21
22/// The type of deployment - either tables-only or full objects.
23#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
24#[serde(rename_all = "snake_case")]
25pub enum DeploymentKind {
26    /// Table creation deployment (apply tables command)
27    Tables,
28    /// Full object deployment (stage, apply commands)
29    Objects,
30    /// Contains sinks
31    Sinks,
32    /// Contains replacement materialized views (drop after apply, not swap)
33    Replacement,
34}
35
36impl fmt::Display for DeploymentKind {
37    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
38        match self {
39            DeploymentKind::Tables => write!(f, "tables"),
40            DeploymentKind::Objects => write!(f, "objects"),
41            DeploymentKind::Sinks => write!(f, "sinks"),
42            DeploymentKind::Replacement => write!(f, "replacement"),
43        }
44    }
45}
46
47impl FromStr for DeploymentKind {
48    type Err = String;
49
50    fn from_str(s: &str) -> Result<Self, Self::Err> {
51        match s {
52            "tables" => Ok(DeploymentKind::Tables),
53            "objects" => Ok(DeploymentKind::Objects),
54            "sinks" => Ok(DeploymentKind::Sinks),
55            "replacement" => Ok(DeploymentKind::Replacement),
56            _ => Err(format!("Invalid deployment kind: {}", s)),
57        }
58    }
59}
60
61/// Deployment mode tag stored alongside each deployment.
62///
63/// Currently only `Stage` exists; kept as an enum for schema compatibility
64/// with the `mode` column in `_mz_deploy.tables.deployments`.
65#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
66#[serde(rename_all = "snake_case")]
67pub enum DeploymentMode {
68    /// Full staging deployment that can be promoted to production.
69    Stage,
70}
71
72impl fmt::Display for DeploymentMode {
73    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
74        match self {
75            DeploymentMode::Stage => write!(f, "stage"),
76        }
77    }
78}
79
80impl FromStr for DeploymentMode {
81    type Err = String;
82
83    fn from_str(s: &str) -> Result<Self, Self::Err> {
84        match s {
85            "stage" => Ok(DeploymentMode::Stage),
86            _ => Err(format!("Invalid deployment mode: {}", s)),
87        }
88    }
89}
90
91/// A compute cluster in Materialize.
92///
93/// Clusters provide the compute resources for materialized views, indexes, and sinks.
94#[derive(Debug, Clone, PartialEq, Eq)]
95pub struct Cluster {
96    /// Materialize's unique identifier for the cluster
97    pub id: String,
98    /// Cluster name (e.g., "quickstart")
99    pub name: String,
100    /// Cluster size (e.g., "M.1-large"), None for unmanaged clusters
101    pub size: Option<String>,
102    /// Number of replicas for fault tolerance (stored as i64 to handle postgres uint4 type)
103    pub replication_factor: Option<i64>,
104    /// The configured autoscaling policy. `None` for unmanaged clusters,
105    /// clusters without a policy, and regions that predate the feature.
106    pub auto_scaling_strategy: Option<AutoScalingStrategy>,
107}
108
109/// Options for creating a new managed cluster.
110#[derive(Debug, Clone, PartialEq, Eq)]
111pub struct ClusterOptions {
112    /// Cluster size (e.g., "M.1-large", "M.1-small")
113    pub size: String,
114    /// Number of replicas (default: 1)
115    pub replication_factor: u32,
116    /// The autoscaling policy, if one is configured.
117    pub auto_scaling_strategy: Option<AutoScalingStrategy>,
118}
119
120impl ClusterOptions {
121    /// Create cluster options from a production cluster configuration.
122    pub fn from_cluster(cluster: &Cluster) -> Result<Self, String> {
123        let size = cluster.size.clone().ok_or_else(|| {
124            format!(
125                "Cluster '{}' has no size (unmanaged cluster?)",
126                cluster.name
127            )
128        })?;
129
130        let replication_factor = cluster
131            .replication_factor
132            .unwrap_or(1)
133            .try_into()
134            .map_err(|_| format!("Invalid replication_factor for cluster '{}'", cluster.name))?;
135
136        Ok(Self {
137            size,
138            replication_factor,
139            auto_scaling_strategy: cluster.auto_scaling_strategy.clone(),
140        })
141    }
142}
143
144/// Configuration for a cluster replica (used for unmanaged clusters).
145#[derive(Debug, Clone, PartialEq, Eq)]
146pub struct ClusterReplica {
147    /// Replica name (e.g., "r1", "r2")
148    pub name: String,
149    /// Replica size (e.g., "25cc")
150    pub size: String,
151    /// Optional availability zone
152    pub availability_zone: Option<String>,
153}
154
155/// A privilege grant on a cluster.
156#[derive(Debug, Clone, PartialEq, Eq)]
157pub struct ObjectGrant {
158    /// Role name that receives the grant
159    pub grantee: String,
160    /// Privilege type (e.g., "USAGE", "CREATE")
161    pub privilege_type: String,
162}
163
164/// Configuration for creating a cluster (managed or unmanaged).
165///
166/// This captures all the information needed to clone a cluster's configuration
167/// including its replicas (for unmanaged clusters) and privilege grants.
168#[derive(Debug, Clone, PartialEq, Eq)]
169pub enum ClusterConfig {
170    /// Managed cluster with a size, replication factor, and optional autoscaling policy
171    Managed {
172        /// Cluster options (size, replication factor, autoscaling policy)
173        options: ClusterOptions,
174        /// Privilege grants on the cluster
175        grants: Vec<ObjectGrant>,
176    },
177    /// Unmanaged cluster with explicit replicas
178    Unmanaged {
179        /// Replica configurations
180        replicas: Vec<ClusterReplica>,
181        /// Privilege grants on the cluster
182        grants: Vec<ObjectGrant>,
183    },
184}
185
186impl ClusterConfig {
187    /// Get the grants for this cluster configuration.
188    pub fn grants(&self) -> &[ObjectGrant] {
189        match self {
190            ClusterConfig::Managed { grants, .. } => grants,
191            ClusterConfig::Unmanaged { grants, .. } => grants,
192        }
193    }
194}
195
196/// A schema deployment record tracking when and how a schema was deployed.
197///
198/// Stored in the `deploy.deployments` table. Schemas are deployed
199/// atomically - all objects in a dirty schema are redeployed together.
200#[derive(Debug, Clone, PartialEq, Eq)]
201pub struct SchemaDeploymentRecord {
202    /// Deploy ID (e.g., `"<init>"` for direct deploy, `"staging"` for staged deploy)
203    pub deploy_id: String,
204    /// Database name (e.g., "materialize")
205    pub database: String,
206    /// Schema name (e.g., "public")
207    pub schema: String,
208    /// When this schema was deployed
209    pub deployed_at: DateTime<Utc>,
210    /// Which Materialize user/role deployed this schema
211    pub deployed_by: String,
212    /// When this schema was promoted to production (NULL for staging, set on promotion)
213    pub promoted_at: Option<DateTime<Utc>>,
214    /// Git commit hash if available
215    pub git_commit: Option<String>,
216    /// Type of deployment (tables or objects)
217    pub kind: DeploymentKind,
218    /// Whether this is a stage or preview deployment
219    pub mode: DeploymentMode,
220}
221
222/// An object deployment record tracking object-level deployment history.
223///
224/// Stored in the `deploy.objects` table (append-only).
225/// Each row records that an object with a specific hash was deployed
226/// to a deployment at a point in time.
227#[derive(Debug, Clone, PartialEq, Eq)]
228pub struct DeploymentObjectRecord {
229    /// Deploy ID (e.g., `"<init>"` for direct deploy, `"staging"` for staged deploy)
230    pub deploy_id: String,
231    /// Database name (e.g., "materialize")
232    pub database: String,
233    /// Schema name (e.g., "public")
234    pub schema: String,
235    /// Object name (e.g., "my_view")
236    pub object: String,
237    /// Hash of the HIR DatabaseObject (semantic content hash)
238    pub object_hash: String,
239    /// When this object was deployed
240    pub deployed_at: DateTime<Utc>,
241}
242
243/// Metadata about a deployment.
244///
245/// Used for validation before operations like apply or abort.
246#[derive(Debug, Clone, PartialEq, Eq)]
247pub struct DeploymentMetadata {
248    /// Deploy ID
249    pub deploy_id: String,
250    /// When this deployment was promoted (NULL if not promoted)
251    pub promoted_at: Option<DateTime<Utc>>,
252    /// Whether this is a stage or preview deployment
253    pub mode: DeploymentMode,
254    /// List of (database, schema) tuples in this deployment
255    pub schemas: Vec<SchemaQualifier>,
256}
257
258/// A conflict record indicating a schema was updated after deployment started.
259///
260/// Used for git-merge-style conflict detection when promoting deployments.
261/// Returned by conflict detection queries that check if production schemas
262/// were modified since the staging deployment began.
263#[derive(Debug, Clone, PartialEq, Eq)]
264pub struct ConflictRecord {
265    /// Database name containing the conflicting schema
266    pub database: String,
267    /// Schema name that has a conflict
268    pub schema: String,
269    /// Deploy ID that last promoted this schema
270    pub deploy_id: String,
271    /// When the schema was last promoted to production
272    pub promoted_at: DateTime<Utc>,
273}
274
275/// A cluster known to host at least one promoted deployment.
276///
277/// Returned by `list_production_clusters()` and used by `dev` to refuse
278/// overlay deployments that would land on production compute. Each record
279/// carries one representative protected deployment so the error message
280/// can explain why the cluster is considered production.
281#[derive(Debug, Clone, PartialEq, Eq)]
282pub struct ProductionClusterRecord {
283    /// Cluster name as it appears in `mz_clusters` (resolved from cluster_id).
284    pub cluster_name: String,
285    /// Database of one promoted deployment hosted on this cluster.
286    pub database: String,
287    /// Schema of one promoted deployment hosted on this cluster.
288    pub schema: String,
289    /// When that deployment was promoted.
290    pub promoted_at: DateTime<Utc>,
291}
292
293/// Details about a specific deployment.
294///
295/// Returned by `get_deployment_details()` for the describe command.
296#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
297pub struct DeploymentDetails {
298    /// When this deployment was created
299    pub deployed_at: DateTime<Utc>,
300    /// When this deployment was promoted (None if still staging)
301    pub promoted_at: Option<DateTime<Utc>>,
302    /// Which Materialize user/role deployed this
303    pub deployed_by: String,
304    /// Git commit hash if available
305    pub git_commit: Option<String>,
306    /// Type of deployment (tables or objects)
307    pub kind: DeploymentKind,
308    /// Whether this is a stage or preview deployment
309    pub mode: DeploymentMode,
310    /// List of (database, schema) tuples in this deployment
311    pub schemas: Vec<SchemaQualifier>,
312}
313
314/// Summary of a staging deployment.
315///
316/// Used by `list_staging_deployments()` for the deployments command.
317#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
318pub struct StagingDeployment {
319    /// When this deployment was created
320    pub deployed_at: DateTime<Utc>,
321    /// Which Materialize user/role deployed this
322    pub deployed_by: String,
323    /// Git commit hash if available
324    pub git_commit: Option<String>,
325    /// Type of deployment (tables or objects)
326    pub kind: DeploymentKind,
327    /// Whether this is a stage or preview deployment
328    pub mode: DeploymentMode,
329    /// List of (database, schema) tuples in this deployment
330    pub schemas: Vec<SchemaQualifier>,
331}
332
333/// A promoted deployment in history.
334///
335/// Returned by `list_deployment_history()` for the history command.
336#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
337pub struct DeploymentHistoryEntry {
338    /// Deploy ID for this deployment
339    pub deploy_id: String,
340    /// When this deployment was promoted
341    pub promoted_at: DateTime<Utc>,
342    /// Which Materialize user/role deployed this
343    pub deployed_by: String,
344    /// Git commit hash if available
345    pub git_commit: Option<String>,
346    /// Type of deployment (tables or objects)
347    pub kind: DeploymentKind,
348    /// List of (database, schema) tuples in this deployment
349    pub schemas: Vec<SchemaQualifier>,
350}
351
352/// State of an apply operation for resumable apply.
353///
354/// This is determined by checking the existence and comments of the
355/// `_mz_deploy.apply_<deploy_id>_pre` and `_mz_deploy.apply_<deploy_id>_post` schemas.
356/// Comments are set when creating the schemas; the swap transaction exchanges which
357/// schema has which comment.
358#[derive(Debug, Clone, Copy, PartialEq, Eq)]
359pub enum ApplyState {
360    /// No apply state schemas exist - fresh apply or completed.
361    NotStarted,
362    /// State schemas exist but swap hasn't happened yet.
363    /// The `_pre` schema has comment 'swapped=false'.
364    PreSwap,
365    /// Swap has completed.
366    /// After the swap, `_pre` schema has comment 'swapped=true' (it was `_post` before).
367    PostSwap,
368}
369
370/// A replacement materialized view record tracking the mapping between
371/// the replacement MV (in staging schema) and its production target.
372///
373/// The replacement MV lives in a staging schema (`target_schema` + staging suffix)
374/// within the same database, and has the same object name as the target.
375///
376/// Stored in `_mz_deploy.public.replacement_mvs` table.
377#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
378pub struct ReplacementMvRecord {
379    /// Deploy ID this replacement belongs to
380    pub deploy_id: String,
381    /// Database of the target (and replacement) MV
382    pub target_database: String,
383    /// Schema of the production target MV
384    pub target_schema: String,
385    /// Name of the target (and replacement) MV
386    pub target_name: String,
387    /// Schema of the replacement MV (staging) — typically `target_schema` + staging suffix
388    pub replacement_schema: String,
389}
390
391/// A pending statement to be executed after the swap.
392///
393/// Used for deferred execution of statements like sinks that cannot
394/// be created in staging (they write to external systems immediately).
395/// Stored in `_mz_deploy.public.pending_statements` table.
396#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
397pub struct PendingStatement {
398    /// Deploy ID this statement belongs to
399    pub deploy_id: String,
400    /// Sequence number for ordering execution
401    pub sequence_num: i32,
402    /// Database containing the object
403    pub database: String,
404    /// Schema containing the object
405    pub schema: String,
406    /// Object name
407    pub object: String,
408    /// Hash of the object definition
409    pub object_hash: String,
410    /// SQL statement to execute
411    pub statement_sql: String,
412    /// Kind of statement (e.g., "sink")
413    pub statement_kind: String,
414    /// When this statement was executed (None if not yet executed)
415    pub executed_at: Option<DateTime<Utc>>,
416}
417
418#[cfg(test)]
419mod tests {
420    use super::*;
421
422    #[mz_ore::test]
423    fn test_deployment_kind_display() {
424        assert_eq!(DeploymentKind::Tables.to_string(), "tables");
425        assert_eq!(DeploymentKind::Objects.to_string(), "objects");
426    }
427
428    #[mz_ore::test]
429    fn test_deployment_kind_from_str_valid() {
430        assert_eq!(
431            "tables".parse::<DeploymentKind>().unwrap(),
432            DeploymentKind::Tables
433        );
434        assert_eq!(
435            "objects".parse::<DeploymentKind>().unwrap(),
436            DeploymentKind::Objects
437        );
438    }
439
440    #[mz_ore::test]
441    fn test_deployment_kind_from_str_invalid() {
442        let result = "invalid".parse::<DeploymentKind>();
443        assert!(result.is_err());
444        assert_eq!(result.unwrap_err(), "Invalid deployment kind: invalid");
445    }
446
447    #[mz_ore::test]
448    fn test_deployment_kind_roundtrip() {
449        // Verify that Display and FromStr are consistent
450        for kind in [
451            DeploymentKind::Tables,
452            DeploymentKind::Objects,
453            DeploymentKind::Sinks,
454            DeploymentKind::Replacement,
455        ] {
456            let s = kind.to_string();
457            let parsed: DeploymentKind = s.parse().unwrap();
458            assert_eq!(kind, parsed);
459        }
460    }
461
462    #[mz_ore::test]
463    fn test_cluster_options_from_cluster_success() {
464        let cluster = Cluster {
465            id: "u1".to_string(),
466            name: "quickstart".to_string(),
467            size: Some("25cc".to_string()),
468            replication_factor: Some(2),
469            auto_scaling_strategy: None,
470        };
471
472        let options = ClusterOptions::from_cluster(&cluster).unwrap();
473        assert_eq!(options.size, "25cc");
474        assert_eq!(options.replication_factor, 2);
475    }
476
477    #[mz_ore::test]
478    fn test_cluster_options_from_cluster_default_replication() {
479        let cluster = Cluster {
480            id: "u1".to_string(),
481            name: "quickstart".to_string(),
482            size: Some("25cc".to_string()),
483            replication_factor: None, // Should default to 1
484            auto_scaling_strategy: None,
485        };
486
487        let options = ClusterOptions::from_cluster(&cluster).unwrap();
488        assert_eq!(options.size, "25cc");
489        assert_eq!(options.replication_factor, 1);
490    }
491
492    #[mz_ore::test]
493    fn test_cluster_options_from_cluster_no_size() {
494        let cluster = Cluster {
495            id: "u1".to_string(),
496            name: "unmanaged".to_string(),
497            size: None, // Unmanaged cluster
498            replication_factor: Some(1),
499            auto_scaling_strategy: None,
500        };
501
502        let result = ClusterOptions::from_cluster(&cluster);
503        assert!(result.is_err());
504        let err_msg = result.unwrap_err();
505        assert!(err_msg.contains("unmanaged"));
506        assert!(err_msg.contains("has no size"));
507    }
508
509    #[mz_ore::test]
510    fn test_cluster_options_from_cluster_negative_replication() {
511        let cluster = Cluster {
512            id: "u1".to_string(),
513            name: "test".to_string(),
514            size: Some("25cc".to_string()),
515            replication_factor: Some(-1), // Invalid negative value
516            auto_scaling_strategy: None,
517        };
518
519        let result = ClusterOptions::from_cluster(&cluster);
520        assert!(result.is_err());
521        assert!(result.unwrap_err().contains("Invalid replication_factor"));
522    }
523
524    #[mz_ore::test]
525    fn test_cluster_equality() {
526        let cluster1 = Cluster {
527            id: "u1".to_string(),
528            name: "test".to_string(),
529            size: Some("25cc".to_string()),
530            replication_factor: Some(1),
531            auto_scaling_strategy: None,
532        };
533
534        let cluster2 = Cluster {
535            id: "u1".to_string(),
536            name: "test".to_string(),
537            size: Some("25cc".to_string()),
538            replication_factor: Some(1),
539            auto_scaling_strategy: None,
540        };
541
542        let cluster3 = Cluster {
543            id: "u2".to_string(), // Different ID
544            name: "test".to_string(),
545            size: Some("25cc".to_string()),
546            replication_factor: Some(1),
547            auto_scaling_strategy: None,
548        };
549
550        assert_eq!(cluster1, cluster2);
551        assert_ne!(cluster1, cluster3);
552    }
553
554    #[mz_ore::test]
555    fn test_cluster_options_equality() {
556        let opts1 = ClusterOptions {
557            size: "25cc".to_string(),
558            replication_factor: 2,
559            auto_scaling_strategy: None,
560        };
561
562        let opts2 = ClusterOptions {
563            size: "25cc".to_string(),
564            replication_factor: 2,
565            auto_scaling_strategy: None,
566        };
567
568        let opts3 = ClusterOptions {
569            size: "50cc".to_string(),
570            replication_factor: 2,
571            auto_scaling_strategy: None,
572        };
573
574        assert_eq!(opts1, opts2);
575        assert_ne!(opts1, opts3);
576    }
577
578    #[mz_ore::test]
579    fn test_cluster_replica_equality() {
580        let r1 = ClusterReplica {
581            name: "r1".to_string(),
582            size: "25cc".to_string(),
583            availability_zone: Some("use1-az1".to_string()),
584        };
585
586        let r2 = ClusterReplica {
587            name: "r1".to_string(),
588            size: "25cc".to_string(),
589            availability_zone: Some("use1-az1".to_string()),
590        };
591
592        let r3 = ClusterReplica {
593            name: "r2".to_string(),
594            size: "25cc".to_string(),
595            availability_zone: None,
596        };
597
598        assert_eq!(r1, r2);
599        assert_ne!(r1, r3);
600    }
601
602    #[mz_ore::test]
603    fn test_cluster_grant_equality() {
604        let g1 = ObjectGrant {
605            grantee: "reader".to_string(),
606            privilege_type: "USAGE".to_string(),
607        };
608
609        let g2 = ObjectGrant {
610            grantee: "reader".to_string(),
611            privilege_type: "USAGE".to_string(),
612        };
613
614        let g3 = ObjectGrant {
615            grantee: "writer".to_string(),
616            privilege_type: "CREATE".to_string(),
617        };
618
619        assert_eq!(g1, g2);
620        assert_ne!(g1, g3);
621    }
622
623    #[mz_ore::test]
624    fn test_cluster_config_managed() {
625        let config = ClusterConfig::Managed {
626            options: ClusterOptions {
627                size: "25cc".to_string(),
628                replication_factor: 2,
629                auto_scaling_strategy: None,
630            },
631            grants: vec![ObjectGrant {
632                grantee: "reader".to_string(),
633                privilege_type: "USAGE".to_string(),
634            }],
635        };
636
637        assert_eq!(config.grants().len(), 1);
638        assert_eq!(config.grants()[0].grantee, "reader");
639    }
640
641    #[mz_ore::test]
642    fn test_cluster_config_unmanaged() {
643        let config = ClusterConfig::Unmanaged {
644            replicas: vec![
645                ClusterReplica {
646                    name: "r1".to_string(),
647                    size: "25cc".to_string(),
648                    availability_zone: None,
649                },
650                ClusterReplica {
651                    name: "r2".to_string(),
652                    size: "50cc".to_string(),
653                    availability_zone: Some("use1-az1".to_string()),
654                },
655            ],
656            grants: vec![],
657        };
658
659        if let ClusterConfig::Unmanaged { replicas, grants } = &config {
660            assert_eq!(replicas.len(), 2);
661            assert_eq!(replicas[0].name, "r1");
662            assert_eq!(replicas[1].availability_zone, Some("use1-az1".to_string()));
663            assert!(grants.is_empty());
664        } else {
665            panic!("Expected Unmanaged config");
666        }
667    }
668
669    #[mz_ore::test]
670    fn test_cluster_config_unmanaged_empty_replicas() {
671        // Unmanaged clusters with 0 replicas are valid
672        let config = ClusterConfig::Unmanaged {
673            replicas: vec![],
674            grants: vec![ObjectGrant {
675                grantee: "admin".to_string(),
676                privilege_type: "CREATE".to_string(),
677            }],
678        };
679
680        if let ClusterConfig::Unmanaged { replicas, grants } = &config {
681            assert!(replicas.is_empty());
682            assert_eq!(grants.len(), 1);
683        } else {
684            panic!("Expected Unmanaged config");
685        }
686    }
687}
mz_deploy/client/models.rs

mz_deploy/client/
models.rs
