mz_adapter/catalog/
command.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//! Catalog commands that can be rechecked when committing a DDL transaction.
11//!
12//! Statements run against the transaction's working catalog, but the catalog
13//! can change before commit. Commands retain the original request and any
14//! output we might have sent to the client (aka. the "promise"), so commit can
15//! "assert" on that output when re-checking catalog commands against the
16//! current catalog.
17//!
18//! Pending external work abandons the candidate. A continuation rebuilds the
19//! prefix before consuming its indexed response, so successful preparation and
20//! preparation errors are accepted only when their inputs still match.
21
22pub(crate) mod source;
23pub(crate) mod source_statement;
24
25use std::sync::Arc;
26
27use mz_sql::ast::{Raw, Statement};
28use mz_sql::catalog::ObjectType;
29use mz_sql::plan::{Params, Plan, PlanNotice};
30use mz_sql::session::metadata::SessionMetadata;
31use tokio::sync::mpsc;
32
33use crate::catalog::command::source_statement::{
34    SourceCommand, SourceCompilation, SourceResolution,
35};
36use crate::catalog::{CatalogState, CommandExecutionError, CommandWork, Op};
37use crate::session::SessionMeta;
38use crate::{AdapterError, AdapterNotice};
39
40#[derive(Clone, Debug)]
41pub enum CatalogCommand {
42    Rename(Arc<RenameCommand>),
43    CatalogResult(Arc<CatalogResultCommand>),
44    Description(Arc<DescriptionCommand>),
45    Source(Arc<source_statement::SourceCommand>),
46}
47
48impl CatalogCommand {
49    /// Resolves a command against the candidate catalog, or requests external work.
50    ///
51    /// The response must belong to this command's index in the replayed batch.
52    pub(super) fn compile(
53        &mut self,
54        state: &CatalogState,
55        index: usize,
56        response: Option<mz_sql::pure::SourcePreparationResponse>,
57    ) -> Result<CommandExecution<'_>, CommandExecutionError> {
58        match self {
59            Self::Rename(command) => Ok(CommandExecution::raw(command.execute(state)?)),
60            Self::Description(command) => {
61                command.execute(state)?;
62                Ok(CommandExecution::raw(Vec::new()))
63            }
64            Self::CatalogResult(command) => {
65                command.execute(state)?;
66                Ok(CommandExecution::raw(Vec::new()))
67            }
68            Self::Source(command) => {
69                let command = Arc::make_mut(command);
70                let resolution = match response {
71                    Some(response) => command.statement.finish(state, response)?,
72                    None => command.statement.resolve(state)?,
73                };
74                match resolution {
75                    SourceResolution::Prepare(request) => Err(CommandExecutionError::Pending {
76                        index,
77                        work: CommandWork::Prepare(request),
78                    }),
79                    SourceResolution::Allocate(keys) => Err(CommandExecutionError::Pending {
80                        index,
81                        work: CommandWork::Allocate(keys),
82                    }),
83                    SourceResolution::Ready(mut compiled) => Ok(CommandExecution {
84                        ops: std::mem::take(&mut compiled.ops),
85                        source: Some((command, compiled)),
86                    }),
87                }
88            }
89        }
90    }
91
92    /// The planning notices recorded during preparation.
93    pub(crate) fn notices(&self) -> &[PlanNotice] {
94        match self {
95            Self::Rename(command) => command.notices(),
96            Self::CatalogResult(command) => &command.notices,
97            Self::Description(_) => &[],
98            Self::Source(command) => command.notices(),
99        }
100    }
101}
102
103/// Catalog writes and the outcome validation required before publishing them.
104///
105/// Run any collision probe in an isolated transaction before taking the writes.
106/// For executions that are not skipped, apply the writes in order and call
107/// `finish` before executing the next command.
108pub(super) struct CommandExecution<'a> {
109    ops: Vec<Op>,
110    source: Option<(&'a mut SourceCommand, Box<SourceCompilation>)>,
111}
112
113impl CommandExecution<'_> {
114    pub(super) fn raw(ops: Vec<Op>) -> Self {
115        Self { ops, source: None }
116    }
117
118    pub(super) fn collision_probe_ops(&self) -> Option<&[Op]> {
119        self.source.as_ref().and_then(|(command, compiled)| {
120            command
121                .needs_collision_probe(compiled)
122                .then_some(self.ops.as_slice())
123        })
124    }
125
126    /// Validates a requested collision probe, consuming skipped executions.
127    pub(super) fn finish_probe(
128        mut self,
129        result: Result<(), AdapterError>,
130    ) -> Result<Option<Self>, AdapterError> {
131        let (command, compiled) = self.source.as_mut().expect("collision probe requested");
132        let (_, outcome) = compiled.classify_result(result)?;
133        command.check_outcome(outcome)?;
134        if command.skipped_name().is_some() {
135            Ok(None)
136        } else {
137            Ok(Some(self))
138        }
139    }
140
141    pub(super) fn take_ops(&mut self) -> Vec<Op> {
142        std::mem::take(&mut self.ops)
143    }
144
145    /// Checks the observed outcome and propagates the result unless that outcome changed.
146    pub(super) fn finish<T>(self, result: Result<T, AdapterError>) -> Result<T, AdapterError> {
147        if let Some((command, compiled)) = self.source {
148            match &result {
149                Ok(_) => command.check_outcome(compiled.created())?,
150                Err(error) => {
151                    if let Some(outcome) = compiled.skipped(error) {
152                        command.check_outcome(outcome)?;
153                    }
154                }
155            }
156        }
157        result
158    }
159}
160
161/// A rename or swap command.
162#[derive(Debug)]
163pub struct RenameCommand {
164    statement: Arc<Statement<Raw>>,
165    params: Params,
166    session: SessionMeta,
167    outcome: RenameOutcome,
168}
169
170#[derive(Debug, PartialEq)]
171struct RenameOutcome {
172    object_type: ObjectType,
173    changed: bool,
174    notices: Vec<PlanNotice>,
175}
176
177impl RenameCommand {
178    /// Prepares a rename or swap and records the outcome that replay must
179    /// preserve.
180    ///
181    /// Returns the initial plan for statement execution. The statement must
182    /// plan to a supported rename, swap, or ALTER no-op.
183    ///
184    /// # Panics
185    ///
186    /// Panics if the statement plans successfully to anything other than a
187    /// supported rename, swap, or ALTER no-op.
188    pub(crate) fn prepare(
189        state: &CatalogState,
190        statement: Arc<Statement<Raw>>,
191        params: Params,
192        session: SessionMeta,
193    ) -> Result<(Self, Plan), AdapterError> {
194        let (plan, notices) = plan_command(state, &statement, &params, &session)?;
195        let (ops, object_type) = rename_ops(&plan);
196        Ok((
197            Self {
198                statement,
199                params,
200                session,
201                outcome: RenameOutcome {
202                    object_type,
203                    changed: !ops.is_empty(),
204                    notices,
205                },
206            },
207            plan,
208        ))
209    }
210
211    /// The planning notices recorded during preparation.
212    pub(crate) fn notices(&self) -> &[PlanNotice] {
213        &self.outcome.notices
214    }
215
216    /// The object type used in the statement's completion response.
217    pub(crate) fn object_type(&self) -> ObjectType {
218        self.outcome.object_type
219    }
220
221    /// Returns writes for `state` after rechecking planning and authorization.
222    ///
223    /// Rejects changes to the object type, whether the command is a no-op, or
224    /// the planning notices. The caller must apply the returned writes.
225    pub(super) fn execute(&self, state: &CatalogState) -> Result<Vec<Op>, AdapterError> {
226        let (plan, notices) = plan_command(state, &self.statement, &self.params, &self.session)?;
227        let (ops, object_type) = rename_ops(&plan);
228        let outcome = RenameOutcome {
229            object_type,
230            changed: !ops.is_empty(),
231            notices,
232        };
233        if outcome != self.outcome {
234            return Err(AdapterError::DDLTransactionOutcomeChanged);
235        }
236        Ok(ops)
237    }
238}
239
240/// A catalog query whose result and notices must remain unchanged at DDL
241/// commit.
242///
243/// In a way, this puts an "assert" into the list of commands for the DDL
244/// transaction, that enforces that the output stays what we have already
245/// reported, when committing.
246#[derive(Debug)]
247pub struct CatalogResultCommand {
248    statement: Arc<Statement<Raw>>,
249    params: Params,
250    session: SessionMeta,
251    row: mz_repr::Row,
252    notices: Vec<PlanNotice>,
253}
254
255impl CatalogResultCommand {
256    /// Whether this statement supports retaining its catalog result for replay.
257    pub(crate) fn supports_statement(statement: &Statement<Raw>) -> bool {
258        use mz_sql::ast::ShowStatement;
259        let Statement::Show(show) = statement else {
260            return false;
261        };
262        match show {
263            ShowStatement::ShowCreateView(_)
264            | ShowStatement::ShowCreateMaterializedView(_)
265            | ShowStatement::ShowCreateSource(_)
266            | ShowStatement::ShowCreateTable(_)
267            | ShowStatement::ShowCreateSink(_)
268            | ShowStatement::ShowCreateMetricSink(_)
269            | ShowStatement::ShowCreateIndex(_)
270            | ShowStatement::ShowCreateConnection(_)
271            | ShowStatement::ShowCreateCluster(_)
272            | ShowStatement::ShowCreateType(_) => true,
273            ShowStatement::ShowObjects(_)
274            | ShowStatement::ShowColumns(_)
275            | ShowStatement::ShowVariable(_)
276            | ShowStatement::InspectShard(_) => false,
277        }
278    }
279
280    /// Prepares a catalog query and records the row and notices replay must
281    /// preserve.
282    ///
283    /// Returns the initial plan for statement execution. Fails if the plan does
284    /// not provide a supported catalog result.
285    pub(crate) fn prepare(
286        state: &CatalogState,
287        statement: Arc<Statement<Raw>>,
288        params: Params,
289        session: SessionMeta,
290    ) -> Result<(Self, Plan), AdapterError> {
291        let (plan, notices) = plan_command(state, &statement, &params, &session)?;
292        let row = Self::result(&plan)?;
293        let command = Self {
294            statement,
295            params,
296            session,
297            row,
298            notices,
299        };
300        Ok((command, plan))
301    }
302
303    /// Rechecks planning and authorization and requires the recorded row and
304    /// notices.
305    pub(in crate::catalog) fn execute(&self, state: &CatalogState) -> Result<(), AdapterError> {
306        let (plan, notices) = plan_command(state, &self.statement, &self.params, &self.session)?;
307        if Self::result(&plan)? == self.row && notices == self.notices {
308            Ok(())
309        } else {
310            Err(AdapterError::DDLTransactionOutcomeChanged)
311        }
312    }
313
314    /// Returns the row to send to the client, or an error for an unsupported
315    /// plan.
316    pub(crate) fn result(plan: &Plan) -> Result<mz_repr::Row, AdapterError> {
317        match plan {
318            Plan::ShowCreate(show) => Ok(show.row.clone()),
319            _ => Err(AdapterError::Unsupported(
320                "non-deterministic catalog result",
321            )),
322        }
323    }
324}
325
326/// A statement description that must remain unchanged at DDL commit.
327///
328/// In a way, this puts an "assert" into the list of commands for the DDL
329/// transaction, that enforces that the output stays what we have already
330/// reported, when committing.
331#[derive(Debug)]
332pub struct DescriptionCommand {
333    statement: Statement<Raw>,
334    param_types: Vec<Option<mz_repr::SqlScalarType>>,
335    session: SessionMeta,
336    expected: mz_sql::plan::StatementDesc,
337}
338
339impl DescriptionCommand {
340    /// Records a statement description for validation at DDL commit.
341    ///
342    /// `expected` must be the description obtained for this statement with the
343    /// supplied parameter types and session metadata. This does not validate
344    /// it.
345    pub(crate) fn new(
346        statement: Statement<Raw>,
347        param_types: Vec<Option<mz_repr::SqlScalarType>>,
348        session: SessionMeta,
349        expected: mz_sql::plan::StatementDesc,
350    ) -> Self {
351        Self {
352            statement,
353            param_types,
354            session,
355            expected,
356        }
357    }
358
359    /// Requires the statement to retain its recorded description in `state`.
360    pub(in crate::catalog) fn execute(&self, state: &CatalogState) -> Result<(), AdapterError> {
361        let (tx, _rx) = mpsc::unbounded_channel();
362        let catalog = state.for_session_metadata(&self.session, tx);
363        let (statement, _) = mz_sql::names::resolve(&catalog, self.statement.clone())?;
364        let current =
365            mz_sql::plan::describe(self.session.pcx(), &catalog, statement, &self.param_types)?;
366        if current == self.expected {
367            Ok(())
368        } else {
369            Err(AdapterError::DDLTransactionOutcomeChanged)
370        }
371    }
372}
373
374fn plan_command(
375    state: &CatalogState,
376    statement: &Statement<Raw>,
377    params: &Params,
378    session: &SessionMeta,
379) -> Result<(Plan, Vec<PlanNotice>), AdapterError> {
380    let (tx, mut rx) = mpsc::unbounded_channel();
381    let catalog = state.for_session_metadata(session, tx);
382    let (statement, resolved_ids) = mz_sql::names::resolve(&catalog, statement.clone())?;
383    let (plan, sql_impl_ids) = mz_sql::plan::plan(
384        Some(session.pcx()),
385        &catalog,
386        statement,
387        params,
388        &resolved_ids,
389    )?;
390    mz_sql::rbac::check_plan(
391        &catalog,
392        None,
393        session,
394        &plan,
395        None,
396        &resolved_ids,
397        &sql_impl_ids,
398    )?;
399    let mut notices = Vec::new();
400    while let Ok(notice) = rx.try_recv() {
401        match notice {
402            AdapterNotice::PlanNotice(notice) => notices.push(notice),
403            _ => unreachable!("planning only emits plan notices"),
404        }
405    }
406    Ok((plan, notices))
407}
408
409/// Returns the rename/swap ops for the given `Plan`, along with the object
410/// type.
411///
412/// # Panics
413///
414/// Panics if `plan` is not a supported rename, swap, or ALTER no-op.
415fn rename_ops(plan: &Plan) -> (Vec<Op>, ObjectType) {
416    match plan {
417        Plan::AlterNoop(plan) => (vec![], plan.object_type),
418        Plan::AlterItemRename(plan) => (
419            vec![Op::RenameItem {
420                id: plan.id,
421                current_full_name: plan.current_full_name.clone(),
422                to_name: plan.to_name.clone(),
423            }],
424            plan.object_type,
425        ),
426        Plan::AlterSchemaRename(plan) => (
427            vec![Op::RenameSchema {
428                database_spec: plan.cur_schema_spec.0,
429                schema_spec: plan.cur_schema_spec.1,
430                new_name: plan.new_schema_name.clone(),
431                check_reserved_names: true,
432            }],
433            ObjectType::Schema,
434        ),
435        Plan::AlterSchemaSwap(plan) => (
436            vec![
437                Op::RenameSchema {
438                    database_spec: plan.schema_a_spec.0,
439                    schema_spec: plan.schema_a_spec.1,
440                    new_name: plan.name_temp.clone(),
441                    check_reserved_names: false,
442                },
443                Op::RenameSchema {
444                    database_spec: plan.schema_b_spec.0,
445                    schema_spec: plan.schema_b_spec.1,
446                    new_name: plan.schema_a_name.clone(),
447                    check_reserved_names: false,
448                },
449                Op::RenameSchema {
450                    database_spec: plan.schema_a_spec.0,
451                    schema_spec: plan.schema_a_spec.1,
452                    new_name: plan.schema_b_name.clone(),
453                    check_reserved_names: false,
454                },
455            ],
456            ObjectType::Schema,
457        ),
458        Plan::AlterClusterRename(plan) => (
459            vec![Op::RenameCluster {
460                id: plan.id,
461                name: plan.name.clone(),
462                to_name: plan.to_name.clone(),
463                check_reserved_names: true,
464            }],
465            ObjectType::Cluster,
466        ),
467        Plan::AlterClusterSwap(plan) => (
468            vec![
469                Op::RenameCluster {
470                    id: plan.id_a,
471                    name: plan.name_a.clone(),
472                    to_name: plan.name_temp.clone(),
473                    check_reserved_names: false,
474                },
475                Op::RenameCluster {
476                    id: plan.id_b,
477                    name: plan.name_b.clone(),
478                    to_name: plan.name_a.clone(),
479                    check_reserved_names: false,
480                },
481                Op::RenameCluster {
482                    id: plan.id_a,
483                    name: plan.name_temp.clone(),
484                    to_name: plan.name_b.clone(),
485                    check_reserved_names: false,
486                },
487            ],
488            ObjectType::Cluster,
489        ),
490        Plan::AlterClusterReplicaRename(plan) => (
491            vec![Op::RenameClusterReplica {
492                cluster_id: plan.cluster_id,
493                replica_id: plan.replica_id,
494                name: plan.name.clone(),
495                to_name: plan.to_name.clone(),
496            }],
497            ObjectType::ClusterReplica,
498        ),
499        _ => unreachable!("rename/swap planning must produce a rename or no-op plan"),
500    }
501}
502
503pub(crate) fn plan_subsource(
504    catalog: &crate::catalog::ConnCatalog<'_>,
505    pcx: &mz_sql::plan::PlanContext,
506    params: &Params,
507    statement: mz_sql::ast::CreateSubsourceStatement<mz_sql::names::Aug>,
508    item_id: mz_repr::CatalogItemId,
509    global_id: mz_repr::GlobalId,
510) -> Result<mz_sql::plan::CreateSourcePlanBundle, AdapterError> {
511    let resolved_ids = mz_sql::names::visit_dependencies(catalog, &statement);
512    let (plan, _sql_impl_ids) = mz_sql::plan::plan(
513        Some(pcx),
514        catalog,
515        Statement::CreateSubsource(statement),
516        params,
517        &resolved_ids,
518    )?;
519    let Plan::CreateSource(plan) = plan else {
520        unreachable!("subsource planning produces a source plan");
521    };
522    Ok(mz_sql::plan::CreateSourcePlanBundle {
523        item_id,
524        global_id,
525        plan,
526        resolved_ids,
527        available_source_references: None,
528    })
529}
mz_adapter/catalog/command.rs

mz_adapter/catalog/
command.rs
