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_parser::ast::{CreateClusterStatement, Raw};
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    /// Whether the cluster is managed
101    pub managed: bool,
102    /// Cluster size (e.g., "M.1-large"), None for unmanaged clusters
103    pub size: Option<String>,
104    /// Number of replicas for fault tolerance (stored as i64 to handle postgres uint4 type)
105    pub replication_factor: Option<i64>,
106}
107
108/// Configuration for a cluster replica (used for unmanaged clusters).
109#[derive(Debug, Clone, PartialEq, Eq)]
110pub struct ClusterReplica {
111    /// Replica name (e.g., "r1", "r2")
112    pub name: String,
113    /// Replica size (e.g., "25cc")
114    pub size: String,
115    /// Optional availability zone
116    pub availability_zone: Option<String>,
117}
118
119/// A privilege grant on a cluster.
120#[derive(Debug, Clone, PartialEq, Eq)]
121pub struct ObjectGrant {
122    /// Role name that receives the grant
123    pub grantee: String,
124    /// Privilege type (e.g., "USAGE", "CREATE")
125    pub privilege_type: String,
126}
127
128/// Configuration for creating a cluster (managed or unmanaged).
129///
130/// This captures all the information needed to clone a cluster's configuration
131/// including its replicas (for unmanaged clusters) and privilege grants.
132#[derive(Debug, Clone, PartialEq, Eq)]
133pub enum ClusterConfig {
134    /// Managed cluster, captured as the canonical `CREATE CLUSTER` statement the
135    /// server renders for it. Cloning replays it under a new name.
136    Managed {
137        /// The production cluster's `CREATE CLUSTER` statement
138        create_stmt: CreateClusterStatement<Raw>,
139        /// Privilege grants on the cluster
140        grants: Vec<ObjectGrant>,
141    },
142    /// Unmanaged cluster with explicit replicas
143    Unmanaged {
144        /// Replica configurations
145        replicas: Vec<ClusterReplica>,
146        /// Privilege grants on the cluster
147        grants: Vec<ObjectGrant>,
148    },
149}
150
151impl ClusterConfig {
152    /// Get the grants for this cluster configuration.
153    pub fn grants(&self) -> &[ObjectGrant] {
154        match self {
155            ClusterConfig::Managed { grants, .. } => grants,
156            ClusterConfig::Unmanaged { grants, .. } => grants,
157        }
158    }
159}
160
161/// A schema deployment record tracking when and how a schema was deployed.
162///
163/// Stored in the `deploy.deployments` table. Schemas are deployed
164/// atomically - all objects in a dirty schema are redeployed together.
165#[derive(Debug, Clone, PartialEq, Eq)]
166pub struct SchemaDeploymentRecord {
167    /// Deploy ID (e.g., `"<init>"` for direct deploy, `"staging"` for staged deploy)
168    pub deploy_id: String,
169    /// Database name (e.g., "materialize")
170    pub database: String,
171    /// Schema name (e.g., "public")
172    pub schema: String,
173    /// When this schema was deployed
174    pub deployed_at: DateTime<Utc>,
175    /// Which Materialize user/role deployed this schema
176    pub deployed_by: String,
177    /// When this schema was promoted to production (NULL for staging, set on promotion)
178    pub promoted_at: Option<DateTime<Utc>>,
179    /// Git commit hash if available
180    pub git_commit: Option<String>,
181    /// Type of deployment (tables or objects)
182    pub kind: DeploymentKind,
183    /// Whether this is a stage or preview deployment
184    pub mode: DeploymentMode,
185}
186
187/// An object deployment record tracking object-level deployment history.
188///
189/// Stored in the `deploy.objects` table (append-only).
190/// Each row records that an object with a specific hash was deployed
191/// to a deployment at a point in time.
192#[derive(Debug, Clone, PartialEq, Eq)]
193pub struct DeploymentObjectRecord {
194    /// Deploy ID (e.g., `"<init>"` for direct deploy, `"staging"` for staged deploy)
195    pub deploy_id: String,
196    /// Database name (e.g., "materialize")
197    pub database: String,
198    /// Schema name (e.g., "public")
199    pub schema: String,
200    /// Object name (e.g., "my_view")
201    pub object: String,
202    /// Hash of the HIR DatabaseObject (semantic content hash)
203    pub object_hash: String,
204    /// When this object was deployed
205    pub deployed_at: DateTime<Utc>,
206}
207
208/// Metadata about a deployment.
209///
210/// Used for validation before operations like apply or abort.
211#[derive(Debug, Clone, PartialEq, Eq)]
212pub struct DeploymentMetadata {
213    /// Deploy ID
214    pub deploy_id: String,
215    /// When this deployment was promoted (NULL if not promoted)
216    pub promoted_at: Option<DateTime<Utc>>,
217    /// Whether this is a stage or preview deployment
218    pub mode: DeploymentMode,
219    /// List of (database, schema) tuples in this deployment
220    pub schemas: Vec<SchemaQualifier>,
221}
222
223/// A conflict record indicating a schema was updated after deployment started.
224///
225/// Used for git-merge-style conflict detection when promoting deployments.
226/// Returned by conflict detection queries that check if production schemas
227/// were modified since the staging deployment began.
228#[derive(Debug, Clone, PartialEq, Eq)]
229pub struct ConflictRecord {
230    /// Database name containing the conflicting schema
231    pub database: String,
232    /// Schema name that has a conflict
233    pub schema: String,
234    /// Deploy ID that last promoted this schema
235    pub deploy_id: String,
236    /// When the schema was last promoted to production
237    pub promoted_at: DateTime<Utc>,
238}
239
240/// A cluster known to host at least one promoted deployment.
241///
242/// Returned by `list_production_clusters()` and used by `dev` to refuse
243/// overlay deployments that would land on production compute. Each record
244/// carries one representative protected deployment so the error message
245/// can explain why the cluster is considered production.
246#[derive(Debug, Clone, PartialEq, Eq)]
247pub struct ProductionClusterRecord {
248    /// Cluster name as it appears in `mz_clusters` (resolved from cluster_id).
249    pub cluster_name: String,
250    /// Database of one promoted deployment hosted on this cluster.
251    pub database: String,
252    /// Schema of one promoted deployment hosted on this cluster.
253    pub schema: String,
254    /// When that deployment was promoted.
255    pub promoted_at: DateTime<Utc>,
256}
257
258/// Details about a specific deployment.
259///
260/// Returned by `get_deployment_details()` for the describe command.
261#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
262pub struct DeploymentDetails {
263    /// When this deployment was created
264    pub deployed_at: DateTime<Utc>,
265    /// When this deployment was promoted (None if still staging)
266    pub promoted_at: Option<DateTime<Utc>>,
267    /// Which Materialize user/role deployed this
268    pub deployed_by: String,
269    /// Git commit hash if available
270    pub git_commit: Option<String>,
271    /// Type of deployment (tables or objects)
272    pub kind: DeploymentKind,
273    /// Whether this is a stage or preview deployment
274    pub mode: DeploymentMode,
275    /// List of (database, schema) tuples in this deployment
276    pub schemas: Vec<SchemaQualifier>,
277}
278
279/// Summary of a staging deployment.
280///
281/// Used by `list_staging_deployments()` for the deployments command.
282#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
283pub struct StagingDeployment {
284    /// When this deployment was created
285    pub deployed_at: DateTime<Utc>,
286    /// Which Materialize user/role deployed this
287    pub deployed_by: String,
288    /// Git commit hash if available
289    pub git_commit: Option<String>,
290    /// Type of deployment (tables or objects)
291    pub kind: DeploymentKind,
292    /// Whether this is a stage or preview deployment
293    pub mode: DeploymentMode,
294    /// List of (database, schema) tuples in this deployment
295    pub schemas: Vec<SchemaQualifier>,
296}
297
298/// A promoted deployment in history.
299///
300/// Returned by `list_deployment_history()` for the history command.
301#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
302pub struct DeploymentHistoryEntry {
303    /// Deploy ID for this deployment
304    pub deploy_id: String,
305    /// When this deployment was promoted
306    pub promoted_at: DateTime<Utc>,
307    /// Which Materialize user/role deployed this
308    pub deployed_by: String,
309    /// Git commit hash if available
310    pub git_commit: Option<String>,
311    /// Type of deployment (tables or objects)
312    pub kind: DeploymentKind,
313    /// List of (database, schema) tuples in this deployment
314    pub schemas: Vec<SchemaQualifier>,
315}
316
317/// State of an apply operation for resumable apply.
318///
319/// This is determined by checking the existence and comments of the
320/// `_mz_deploy.apply_<deploy_id>_pre` and `_mz_deploy.apply_<deploy_id>_post` schemas.
321/// Comments are set when creating the schemas; the swap transaction exchanges which
322/// schema has which comment.
323#[derive(Debug, Clone, Copy, PartialEq, Eq)]
324pub enum ApplyState {
325    /// No apply state schemas exist - fresh apply or completed.
326    NotStarted,
327    /// State schemas exist but swap hasn't happened yet.
328    /// The `_pre` schema has comment 'swapped=false'.
329    PreSwap,
330    /// Swap has completed.
331    /// After the swap, `_pre` schema has comment 'swapped=true' (it was `_post` before).
332    PostSwap,
333}
334
335/// A replacement materialized view record tracking the mapping between
336/// the replacement MV (in staging schema) and its production target.
337///
338/// The replacement MV lives in a staging schema (`target_schema` + staging suffix)
339/// within the same database, and has the same object name as the target.
340///
341/// Stored in `_mz_deploy.public.replacement_mvs` table.
342#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
343pub struct ReplacementMvRecord {
344    /// Deploy ID this replacement belongs to
345    pub deploy_id: String,
346    /// Database of the target (and replacement) MV
347    pub target_database: String,
348    /// Schema of the production target MV
349    pub target_schema: String,
350    /// Name of the target (and replacement) MV
351    pub target_name: String,
352    /// Schema of the replacement MV (staging) — typically `target_schema` + staging suffix
353    pub replacement_schema: String,
354}
355
356/// A pending statement to be executed after the swap.
357///
358/// Used for deferred execution of statements like sinks that cannot
359/// be created in staging (they write to external systems immediately).
360/// Stored in `_mz_deploy.public.pending_statements` table.
361#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
362pub struct PendingStatement {
363    /// Deploy ID this statement belongs to
364    pub deploy_id: String,
365    /// Sequence number for ordering execution
366    pub sequence_num: i32,
367    /// Database containing the object
368    pub database: String,
369    /// Schema containing the object
370    pub schema: String,
371    /// Object name
372    pub object: String,
373    /// Hash of the object definition
374    pub object_hash: String,
375    /// SQL statement to execute
376    pub statement_sql: String,
377    /// Kind of statement (e.g., "sink")
378    pub statement_kind: String,
379    /// When this statement was executed (None if not yet executed)
380    pub executed_at: Option<DateTime<Utc>>,
381}
382
383#[cfg(test)]
384mod tests {
385    use super::*;
386    use mz_sql_parser::ast::Statement;
387    use mz_sql_parser::parser::parse_statements;
388
389    fn create_cluster(sql: &str) -> CreateClusterStatement<Raw> {
390        match parse_statements(sql).unwrap().pop().unwrap().ast {
391            Statement::CreateCluster(stmt) => stmt,
392            other => panic!("expected CREATE CLUSTER, got {:?}", other),
393        }
394    }
395
396    #[mz_ore::test]
397    fn test_deployment_kind_display() {
398        assert_eq!(DeploymentKind::Tables.to_string(), "tables");
399        assert_eq!(DeploymentKind::Objects.to_string(), "objects");
400    }
401
402    #[mz_ore::test]
403    fn test_deployment_kind_from_str_valid() {
404        assert_eq!(
405            "tables".parse::<DeploymentKind>().unwrap(),
406            DeploymentKind::Tables
407        );
408        assert_eq!(
409            "objects".parse::<DeploymentKind>().unwrap(),
410            DeploymentKind::Objects
411        );
412    }
413
414    #[mz_ore::test]
415    fn test_deployment_kind_from_str_invalid() {
416        let result = "invalid".parse::<DeploymentKind>();
417        assert!(result.is_err());
418        assert_eq!(result.unwrap_err(), "Invalid deployment kind: invalid");
419    }
420
421    #[mz_ore::test]
422    fn test_deployment_kind_roundtrip() {
423        // Verify that Display and FromStr are consistent
424        for kind in [
425            DeploymentKind::Tables,
426            DeploymentKind::Objects,
427            DeploymentKind::Sinks,
428            DeploymentKind::Replacement,
429        ] {
430            let s = kind.to_string();
431            let parsed: DeploymentKind = s.parse().unwrap();
432            assert_eq!(kind, parsed);
433        }
434    }
435
436    #[mz_ore::test]
437    fn test_cluster_equality() {
438        let cluster1 = Cluster {
439            id: "u1".to_string(),
440            name: "test".to_string(),
441            managed: true,
442            size: Some("25cc".to_string()),
443            replication_factor: Some(1),
444        };
445
446        let cluster2 = Cluster {
447            id: "u1".to_string(),
448            name: "test".to_string(),
449            managed: true,
450            size: Some("25cc".to_string()),
451            replication_factor: Some(1),
452        };
453
454        let cluster3 = Cluster {
455            id: "u2".to_string(), // Different ID
456            name: "test".to_string(),
457            managed: true,
458            size: Some("25cc".to_string()),
459            replication_factor: Some(1),
460        };
461
462        assert_eq!(cluster1, cluster2);
463        assert_ne!(cluster1, cluster3);
464    }
465
466    #[mz_ore::test]
467    fn test_cluster_replica_equality() {
468        let r1 = ClusterReplica {
469            name: "r1".to_string(),
470            size: "25cc".to_string(),
471            availability_zone: Some("use1-az1".to_string()),
472        };
473
474        let r2 = ClusterReplica {
475            name: "r1".to_string(),
476            size: "25cc".to_string(),
477            availability_zone: Some("use1-az1".to_string()),
478        };
479
480        let r3 = ClusterReplica {
481            name: "r2".to_string(),
482            size: "25cc".to_string(),
483            availability_zone: None,
484        };
485
486        assert_eq!(r1, r2);
487        assert_ne!(r1, r3);
488    }
489
490    #[mz_ore::test]
491    fn test_cluster_grant_equality() {
492        let g1 = ObjectGrant {
493            grantee: "reader".to_string(),
494            privilege_type: "USAGE".to_string(),
495        };
496
497        let g2 = ObjectGrant {
498            grantee: "reader".to_string(),
499            privilege_type: "USAGE".to_string(),
500        };
501
502        let g3 = ObjectGrant {
503            grantee: "writer".to_string(),
504            privilege_type: "CREATE".to_string(),
505        };
506
507        assert_eq!(g1, g2);
508        assert_ne!(g1, g3);
509    }
510
511    #[mz_ore::test]
512    fn test_cluster_config_managed() {
513        let config = ClusterConfig::Managed {
514            create_stmt: create_cluster("CREATE CLUSTER c (SIZE = '25cc')"),
515            grants: vec![ObjectGrant {
516                grantee: "reader".to_string(),
517                privilege_type: "USAGE".to_string(),
518            }],
519        };
520
521        assert_eq!(config.grants().len(), 1);
522        assert_eq!(config.grants()[0].grantee, "reader");
523    }
524
525    #[mz_ore::test]
526    fn test_cluster_config_unmanaged() {
527        let config = ClusterConfig::Unmanaged {
528            replicas: vec![
529                ClusterReplica {
530                    name: "r1".to_string(),
531                    size: "25cc".to_string(),
532                    availability_zone: None,
533                },
534                ClusterReplica {
535                    name: "r2".to_string(),
536                    size: "50cc".to_string(),
537                    availability_zone: Some("use1-az1".to_string()),
538                },
539            ],
540            grants: vec![],
541        };
542
543        if let ClusterConfig::Unmanaged { replicas, grants } = &config {
544            assert_eq!(replicas.len(), 2);
545            assert_eq!(replicas[0].name, "r1");
546            assert_eq!(replicas[1].availability_zone, Some("use1-az1".to_string()));
547            assert!(grants.is_empty());
548        } else {
549            panic!("Expected Unmanaged config");
550        }
551    }
552
553    #[mz_ore::test]
554    fn test_cluster_config_unmanaged_empty_replicas() {
555        // Unmanaged clusters with 0 replicas are valid
556        let config = ClusterConfig::Unmanaged {
557            replicas: vec![],
558            grants: vec![ObjectGrant {
559                grantee: "admin".to_string(),
560                privilege_type: "CREATE".to_string(),
561            }],
562        };
563
564        if let ClusterConfig::Unmanaged { replicas, grants } = &config {
565            assert!(replicas.is_empty());
566            assert_eq!(grants.len(), 1);
567        } else {
568            panic!("Expected Unmanaged config");
569        }
570    }
571}
mz_deploy/client/models.rs

mz_deploy/client/
models.rs
