mz_deploy/cli/commands/
apply_tables.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//! Apply tables command - create tables that don't exist in the database.
11
12use crate::cli::CliError;
13use crate::cli::commands::apply_objects;
14use crate::cli::commands::grants;
15use crate::cli::executor::{
16    ApplyPlan, ApplyResult, DeploymentExecutor, ObjectAction, ObjectResult,
17    compile_apply_project_and_connect,
18};
19use crate::client::Client;
20use crate::config::Settings;
21use crate::project;
22use crate::project::ast::Statement;
23use crate::project::ir::graph::Project;
24use std::collections::BTreeSet;
25
26const PHASE_NAME: &str = "tables";
27const GRANT_KIND: grants::GrantObjectKind = grants::GrantObjectKind::Table;
28
29fn matches(stmt: &Statement) -> bool {
30    matches!(
31        stmt,
32        Statement::CreateTable(_) | Statement::CreateTableFromSource(_)
33    )
34}
35
36/// Plan only table objects (no deployment tracking, no execution).
37pub async fn plan(
38    _settings: &Settings,
39    client: &Client,
40    executor: &DeploymentExecutor<'_>,
41    planned_project: &Project,
42    apply_plan: &mut ApplyPlan,
43) -> Result<ApplyResult, CliError> {
44    let mut target_ids = BTreeSet::new();
45    for obj in planned_project.iter_objects() {
46        if matches(&obj.typed_object.stmt) {
47            target_ids.insert(obj.id.clone());
48        }
49    }
50
51    if target_ids.is_empty() {
52        return Ok(ApplyResult {
53            phase: PHASE_NAME.to_string(),
54            results: vec![],
55        });
56    }
57
58    let target_objects = planned_project.get_sorted_objects_filtered(&target_ids)?;
59    let existing = client
60        .introspection()
61        .check_catalog_objects_exist(&target_ids, GRANT_KIND.catalog_table())
62        .await
63        .map_err(CliError::Connection)?;
64
65    let to_create: BTreeSet<_> = target_ids.difference(&existing).cloned().collect();
66    client
67        .validation()
68        .validate_source_references(planned_project, &to_create)
69        .await?;
70
71    let schemas: BTreeSet<_> = target_objects
72        .iter()
73        .filter(|(obj_id, _)| !existing.contains(obj_id))
74        .map(|(obj_id, _)| {
75            project::SchemaQualifier::new(
76                obj_id.expect_database().to_string(),
77                obj_id.schema().to_string(),
78            )
79        })
80        .collect();
81    apply_plan
82        .prepare_schemas(executor, planned_project, &schemas)
83        .await?;
84
85    let mut results = Vec::new();
86
87    for (obj_id, typed_obj) in target_objects {
88        executor.take_statements();
89
90        if existing.contains(&obj_id) {
91            apply_objects::reconcile_grants_and_comments(
92                client,
93                executor,
94                &obj_id,
95                typed_obj,
96                &GRANT_KIND,
97            )
98            .await?;
99            results.push(ObjectResult {
100                object: obj_id.to_string(),
101                action: ObjectAction::UpToDate,
102                statements: executor.take_statements(),
103                redacted_statements: vec![],
104                transaction_group: None,
105                post_statements: vec![],
106            });
107            continue;
108        }
109
110        executor.execute_sql(&typed_obj.stmt).await?;
111        let statements = executor.take_statements();
112
113        for index in &typed_obj.indexes {
114            executor.execute_sql(index).await?;
115        }
116        apply_objects::reconcile_grants_and_comments(
117            client,
118            executor,
119            &obj_id,
120            typed_obj,
121            &GRANT_KIND,
122        )
123        .await?;
124        let post_statements = executor.take_statements();
125
126        let transaction_group = match &typed_obj.stmt {
127            Statement::CreateTableFromSource(s) => Some(s.source.to_string()),
128            _ => None,
129        };
130
131        results.push(ObjectResult {
132            object: obj_id.to_string(),
133            action: ObjectAction::Created,
134            statements,
135            redacted_statements: vec![],
136            transaction_group,
137            post_statements,
138        });
139    }
140
141    // Reorder: non-grouped objects first, then grouped objects sorted by group key.
142    // stable_sort_by preserves topological order within each group.
143    results.sort_by(|a, b| a.transaction_group.cmp(&b.transaction_group));
144
145    Ok(ApplyResult {
146        phase: PHASE_NAME.to_string(),
147        results,
148    })
149}
150
151/// Run the `apply tables` command: compile, plan, optionally execute, then lock.
152pub async fn run(settings: &Settings, dry_run: bool) -> Result<ApplyPlan, CliError> {
153    let (planned_project, client) = compile_apply_project_and_connect(settings).await?;
154    let mut apply_plan = ApplyPlan::new();
155    let executor = DeploymentExecutor::new_dry_run(&client);
156    let phase = plan(
157        settings,
158        &client,
159        &executor,
160        &planned_project,
161        &mut apply_plan,
162    )
163    .await?;
164    apply_plan.add_phase(phase);
165
166    if !dry_run {
167        apply_plan.execute(&client).await?;
168        super::lock::run(settings).await?;
169    }
170
171    Ok(apply_plan)
172}
mz_deploy/cli/commands/apply_tables.rs

mz_deploy/cli/commands/
apply_tables.rs
