mz_source_schema_change/
lib.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//! Upstream schema changes that Materialize sources cannot follow.
11
12use mz_ore::str::StrExt;
13use postgres_protocol::escape;
14use serde::{Deserialize, Serialize};
15
16/// An upstream schema change that Materialize cannot follow.
17///
18/// Each source wraps this in an error naming the affected table. `Display`
19/// renders the diagnosis. [`SchemaChange::hint`] renders the recovery steps,
20/// which are surfaced separately: as the `HINT` of a SQL error and in the
21/// source status.
22#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
23pub enum SchemaChange {
24    #[error("table was dropped and recreated upstream")]
25    TableDropped,
26    #[error("table was renamed or moved upstream")]
27    TableRenamed,
28    #[error("column {} was dropped or renamed upstream", .column.quoted())]
29    ColumnDropped { column: String },
30    #[error(
31        "column {} changed position upstream (the column or table was likely dropped and \
32         recreated)",
33        .column.quoted()
34    )]
35    ColumnMoved { column: String },
36    #[error("the type of column {} changed upstream", .column.quoted())]
37    ColumnTypeChanged { column: String },
38    #[error("the NOT NULL constraint on column {} was dropped upstream", .column.quoted())]
39    NotNullDropped { column: String },
40    #[error("{key} was dropped upstream")]
41    KeyDropped { key: KeyRef },
42    #[error("{key} was renamed or recreated upstream")]
43    KeyAltered { key: KeyRef },
44    #[error("{0}")]
45    DescriptionFailed(String),
46}
47
48/// A PRIMARY KEY or UNIQUE constraint as recorded when the table was created.
49#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
50pub struct KeyRef {
51    pub name: String,
52    pub is_primary: bool,
53    pub columns: Vec<String>,
54}
55
56impl std::fmt::Display for KeyRef {
57    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
58        let kind = if self.is_primary {
59            "PRIMARY KEY"
60        } else {
61            "UNIQUE"
62        };
63        write!(
64            f,
65            "{kind} constraint {} ({})",
66            self.name.quoted(),
67            self.columns.join(", ")
68        )
69    }
70}
71
72impl SchemaChange {
73    /// The recovery steps for a dropped constraint on `schema.name`, including
74    /// the statements to run. Other changes carry no hint.
75    pub fn hint(&self, schema: &str, name: &str) -> Option<String> {
76        let recreate = |with_clause: Option<&str>| {
77            let mut hint = format!(
78                "To keep ingesting without this constraint, recreate the table in a new \
79                 versioned schema, then swap your views to the new table:\n  CREATE SCHEMA v2;\n  \
80                 CREATE TABLE v2.{}\n  FROM SOURCE <source> (REFERENCE {}.{})",
81                escape::escape_identifier(name),
82                escape::escape_identifier(schema),
83                escape::escape_identifier(name),
84            );
85            if let Some(with_clause) = with_clause {
86                hint.push_str(&format!("\n  WITH ({with_clause})"));
87            }
88            hint.push(';');
89            hint
90        };
91        match self {
92            SchemaChange::KeyDropped { key } | SchemaChange::KeyAltered { key } => Some(format!(
93                "{}\nTo make a planned constraint drop a non-event, create the table with \
94                 WITH (EXCLUDE CONSTRAINTS ({})) before the upstream drop.",
95                recreate(None),
96                escape::escape_literal(&key.name),
97            )),
98            SchemaChange::NotNullDropped { .. } => Some(recreate(Some("EXCLUDE ALL CONSTRAINTS"))),
99            SchemaChange::TableDropped
100            | SchemaChange::TableRenamed
101            | SchemaChange::ColumnDropped { .. }
102            | SchemaChange::ColumnMoved { .. }
103            | SchemaChange::ColumnTypeChanged { .. }
104            | SchemaChange::DescriptionFailed(_) => None,
105        }
106    }
107}
mz_source_schema_change/lib.rs

mz_source_schema_change/
lib.rs
