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

mz_postgres_util/
schema_change.rs
