storage: report specific errors for incompatible Postgres schema changes

doc/user/content/ingest-data/patterns/upstream-schema-changes.md

@@ -455,7 +455,7 @@ notice, the ingesting table stalls permanently:
 
 ```
 ERROR:  Source error: source must be dropped and recreated due to failure:
-        incompatible schema change: source table orders with oid 16385 has been altered
+        incompatible schema change on public.orders (oid 16385): column "priority" was dropped or renamed upstream
 ```
 
 While the table is stalled, reads against the public interface return this

src/adapter/src/error.rs

@@ -800,6 +800,7 @@ impl AdapterError {
             ),
             AdapterError::Dataflow(e) => match &**e {
                 DataflowError::EvalError(e) => e.hint(),
+                DataflowError::SourceError(e) => e.hint.as_ref().map(|hint| hint.to_string()),
                 _ => None,
             },
             AdapterError::AlterClusterUnmanagedWhileReconfiguring => Some(

src/postgres-util/Cargo.toml

@@ -20,6 +20,7 @@ mz-tls-util = { path = "../tls-util", default-features = false }
 openssl.workspace = true
 openssh = { workspace = true, optional = true }
 postgres_array = { workspace = true, optional = true }
+postgres-protocol = { workspace = true, optional = true }
 proptest = { workspace = true, optional = true }
 proptest-derive.workspace = true
 prost = { workspace = true, optional = true }
@@ -32,7 +33,7 @@ tracing.workspace = true
 [features]
 default = ["mz-build-tools/default", "replication", "schemas", "tunnel"]
 replication = ["tunnel"]
-schemas = ["prost", "serde", "proptest", "mz-proto", "tunnel"]
+schemas = ["postgres-protocol", "prost", "serde", "proptest", "mz-proto", "tunnel"]
 tunnel = [
     "mz-cloud-resources",
     "serde",

src/postgres-util/src/desc.rs

@@ -11,14 +11,15 @@
 
 use std::collections::{BTreeMap, BTreeSet};
 
-use anyhow::bail;
 use mz_proto::{IntoRustIfSome, RustType, TryFromProtoError};
 use proptest::prelude::any;
 use proptest_derive::Arbitrary;
 use serde::{Deserialize, Serialize};
 use tokio_postgres::types::Oid;
 use tracing::warn;
 
+use crate::schema_change::{KeyRef, SchemaChange, SchemaChangeError};
+
 include!(concat!(env!("OUT_DIR"), "/mz_postgres_util.desc.rs"));
 
 /// Describes a schema in a PostgreSQL database.
@@ -59,57 +60,89 @@ impl PostgresTableDesc {
     /// Currently this means that the values are equal except for the following
     /// exceptions:
     /// - `self`'s columns are a compatible prefix of `other`'s columns.
-    ///   Compatibility is defined as returning `true` for
-    ///   `PostgresColumnDesc::is_compatible`.
+    ///   Compatibility is defined by `PostgresColumnDesc::get_incompatible_schema_change`.
     /// - `self`'s keys are all present in `other`
+    ///
+    /// On incompatibility, the error describes the first mismatch found and
+    /// how to recover from it. The error becomes the permanent, user-visible
+    /// error for the stalled table, so it must stand on its own.
     pub fn determine_compatibility(
         &self,
         other: &PostgresTableDesc,
         allow_type_to_change_by_col_num: &BTreeSet<u16>,
-    ) -> Result<(), anyhow::Error> {
+    ) -> Result<(), SchemaChangeError> {
         if self == other {
             return Ok(());
         }
 
-        let PostgresTableDesc {
-            oid: other_oid,
-            namespace: other_namespace,
-            name: other_name,
-            columns: other_cols,
-            keys: other_keys,
-        } = other;
+        if self.oid != other.oid {
+            warn!(
+                "table {}.{} changed oid from {} to {} during schema verification",
+                self.namespace, self.name, self.oid, other.oid
+            );
+            return Err(
+                self.build_schema_change_error(SchemaChange::TableDropped { oid: other.oid })
+            );
+        }
 
-        let other_cols_by_name = BTreeMap::from_iter(other_cols.iter().map(|c| (&c.name, c)));
-        let columns_compatible =
-            self.columns
-                .iter()
-                .all(|info| match other_cols_by_name.get(&info.name) {
-                    Some(other_info) => {
-                        let allow_type_change =
-                            allow_type_to_change_by_col_num.contains(&info.col_num);
-                        info.is_compatible(other_info, allow_type_change)
-                    }
-                    None => false,
-                });
+        if self.namespace != other.namespace || self.name != other.name {
+            return Err(self.build_schema_change_error(SchemaChange::TableRenamed {
+                namespace: other.namespace.clone(),
+                name: other.name.clone(),
+                oid: other.oid,
+            }));
+        }
 
-        if columns_compatible
-            && &self.name == other_name
-            && &self.oid == other_oid
-            && &self.namespace == other_namespace
-            // Our keys are all still present in exactly the same shape.
-            && self.keys.difference(other_keys).next().is_none()
-        {
-            Ok(())
+        let other_cols_by_name = BTreeMap::from_iter(other.columns.iter().map(|c| (&c.name, c)));
+        for column in &self.columns {
+            let allow_type_change = allow_type_to_change_by_col_num.contains(&column.col_num);
+            let other_column = other_cols_by_name.get(&column.name).copied();
+            if let Some(change) =
+                column.get_incompatible_schema_change(other_column, allow_type_change)
+            {
+                return Err(self.build_schema_change_error(change));
+            }
+        }
+
+        if let Some(key) = self.keys.difference(&other.keys).next() {
+            return Err(self.build_schema_change_error(self.key_change(key, other)));
+        }
+
+        Ok(())
+    }
+
+    fn build_schema_change_error(&self, change: SchemaChange) -> SchemaChangeError {
+        SchemaChangeError {
+            namespace: self.namespace.clone(),
+            name: self.name.clone(),
+            oid: self.oid,
+            change,
+        }
+    }
+
+    fn key_change(&self, key: &PostgresKeyDesc, other: &PostgresTableDesc) -> SchemaChange {
+        let key_ref = KeyRef {
+            name: key.name.clone(),
+            is_primary: key.is_primary,
+            columns: key
+                .cols
+                .iter()
+                .map(|attnum| {
+                    self.columns
+                        .iter()
+                        .find(|c| c.col_num == *attnum)
+                        .map_or_else(|| format!("attnum {}", attnum), |c| c.name.clone())
+                })
+                .collect(),
+        };
+        let still_exists = other
+            .keys
+            .iter()
+            .any(|k| k.oid == key.oid || k.name == key.name);
+        if still_exists {
+            SchemaChange::KeyAltered { key: key_ref }
         } else {
-            warn!(
-                "Error validating table in publication. Expected: {:?} Actual: {:?}",
-                &self, other
-            );
-            bail!(
-                "source table {} with oid {} has been altered",
-                self.name,
-                self.oid
-            )
+            SchemaChange::KeyDropped { key: key_ref }
         }
     }
 }
@@ -177,16 +210,34 @@ impl PostgresColumnDesc {
     /// Note that this function somewhat unnecessarily errors if the names
     /// differ; this is negotiable but we want users to understand the fixedness
     /// of names in our schemas.
-    fn is_compatible(&self, other: &PostgresColumnDesc, allow_type_change: bool) -> bool {
-        self.name == other.name
-            && self.col_num == other.col_num
-            && (self.type_oid == other.type_oid || allow_type_change)
-            && (self.type_mod == other.type_mod || allow_type_change)
-            // Columns are compatible if:
-            // - self is nullable; introducing a not null constraint doesn't
-            //   change this column's behavior.
-            // - self and other are both not nullable
-            && (self.nullable || self.nullable == other.nullable)
+    fn get_incompatible_schema_change(
+        &self,
+        other: Option<&PostgresColumnDesc>,
+        allow_type_change: bool,
+    ) -> Option<SchemaChange> {
+        let column = self.name.clone();
+        let Some(other) = other else {
+            return Some(SchemaChange::ColumnDropped { column });
+        };
+        if self.name != other.name {
+            return Some(SchemaChange::ColumnDropped { column });
+        }
+        if self.col_num != other.col_num {
+            return Some(SchemaChange::ColumnMoved { column });
+        }
+        if !allow_type_change
+            && (self.type_oid != other.type_oid || self.type_mod != other.type_mod)
+        {
+            return Some(SchemaChange::ColumnTypeChanged { column });
+        }
+        // Columns are compatible if:
+        // - self is nullable; introducing a not null constraint doesn't
+        //   change this column's behavior.
+        // - self and other are both not nullable
+        if !self.nullable && other.nullable {
+            return Some(SchemaChange::NotNullDropped { column });
+        }
+        None
     }
 }
 

src/postgres-util/src/lib.rs

@@ -19,6 +19,8 @@ pub use replication::{
 #[cfg(feature = "schemas")]
 pub mod desc;
 #[cfg(feature = "schemas")]
+pub mod schema_change;
+#[cfg(feature = "schemas")]
 pub mod schemas;
 #[cfg(feature = "schemas")]
 pub use schemas::{get_schemas, publication_info};

src/postgres-util/src/schema_change.rs

@@ -0,0 +1,117 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+//! Upstream schema changes that Materialize cannot follow.
+
+use mz_ore::str::StrExt;
+use postgres_protocol::escape;
+use serde::{Deserialize, Serialize};
+
+/// An upstream schema change that Materialize cannot follow.
+///
+/// `Display` renders the diagnosis. [`SchemaChangeError::hint`] renders the
+/// recovery steps, which are surfaced separately: as the `HINT` of a SQL error
+/// and in the source status.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
+#[error("incompatible schema change on {namespace}.{name} (oid {oid}): {change}")]
+pub struct SchemaChangeError {
+    pub namespace: String,
+    pub name: String,
+    pub oid: u32,
+    pub change: SchemaChange,
+}
+
+/// The upstream change behind a [`SchemaChangeError`].
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
+pub enum SchemaChange {
+    #[error("table was dropped and recreated upstream (it now has oid {oid})")]
+    TableDropped { oid: u32 },
+    #[error("table was renamed or moved upstream (it is now {namespace}.{name} with oid {oid})")]
+    TableRenamed {
+        namespace: String,
+        name: String,
+        oid: u32,
+    },
+    #[error("column {} was dropped or renamed upstream", .column.quoted())]
+    ColumnDropped { column: String },
+    #[error(
+        "column {} changed position upstream (the column or table was likely dropped and \
+         recreated)",
+        .column.quoted()
+    )]
+    ColumnMoved { column: String },
+    #[error("the type of column {} changed upstream", .column.quoted())]
+    ColumnTypeChanged { column: String },
+    #[error("the NOT NULL constraint on column {} was dropped upstream", .column.quoted())]
+    NotNullDropped { column: String },
+    #[error("{key} was dropped upstream")]
+    KeyDropped { key: KeyRef },
+    #[error("{key} was renamed or recreated upstream")]
+    KeyAltered { key: KeyRef },
+}
+
+/// A PRIMARY KEY or UNIQUE constraint as recorded when the table was created.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct KeyRef {
+    pub name: String,
+    pub is_primary: bool,
+    pub columns: Vec<String>,
+}
+
+impl std::fmt::Display for KeyRef {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let kind = if self.is_primary {
+            "PRIMARY KEY"
+        } else {
+            "UNIQUE"
+        };
+        write!(
+            f,
+            "{kind} constraint {} ({})",
+            self.name.quoted(),
+            self.columns.join(", ")
+        )
+    }
+}
+
+impl SchemaChangeError {
+    /// The recovery steps for a dropped constraint, including the statements
+    /// to run. Other changes carry no hint.
+    pub fn hint(&self) -> Option<String> {
+        let recreate = |with_clause: Option<&str>| {
+            let mut hint = format!(
+                "To keep ingesting without this constraint, recreate the table in a new \
+                 versioned schema, then swap your views to the new table:\n  CREATE SCHEMA v2;\n  \
+                 CREATE TABLE v2.{}\n  FROM SOURCE <source> (REFERENCE {}.{})",
+                escape::escape_identifier(&self.name),
+                escape::escape_identifier(&self.namespace),
+                escape::escape_identifier(&self.name),
+            );
+            if let Some(with_clause) = with_clause {
+                hint.push_str(&format!("\n  WITH ({with_clause})"));
+            }
+            hint.push(';');
+            hint
+        };
+        match &self.change {
+            SchemaChange::KeyDropped { key } | SchemaChange::KeyAltered { key } => Some(format!(
+                "{}\nTo make a planned constraint drop a non-event, create the table with \
+                 WITH (EXCLUDE CONSTRAINTS ({})) before the upstream drop.",
+                recreate(None),
+                escape::escape_literal(&key.name),
+            )),
+            SchemaChange::NotNullDropped { .. } => Some(recreate(Some("EXCLUDE ALL CONSTRAINTS"))),
+            SchemaChange::TableDropped { .. }
+            | SchemaChange::TableRenamed { .. }
+            | SchemaChange::ColumnDropped { .. }
+            | SchemaChange::ColumnMoved { .. }
+            | SchemaChange::ColumnTypeChanged { .. } => None,
+        }
+    }
+}

src/storage-types/src/errors.proto

@@ -38,6 +38,7 @@ message ProtoSourceErrorDetails {
 message ProtoSourceError {
   reserved 1;
   ProtoSourceErrorDetails error = 2;
+  optional string hint = 3;
 }
 
 message ProtoUpsertValueError {

src/storage-types/src/errors.rs

@@ -376,18 +376,21 @@ impl Display for UpsertError {
 #[cfg_attr(any(test, feature = "proptest"), derive(Arbitrary))]
 pub struct SourceError {
     pub error: SourceErrorDetails,
+    pub hint: Option<Box<str>>,
 }
 
 impl RustType<ProtoSourceError> for SourceError {
     fn into_proto(&self) -> ProtoSourceError {
         ProtoSourceError {
             error: Some(self.error.into_proto()),
+            hint: self.hint.as_ref().map(|hint| hint.into_proto()),
         }
     }
 
     fn from_proto(proto: ProtoSourceError) -> Result<Self, TryFromProtoError> {
         Ok(SourceError {
             error: proto.error.into_rust_if_some("ProtoSourceError::error")?,
+            hint: proto.hint.map(Into::into),
         })
     }
 }
@@ -895,6 +898,7 @@ mod columnation {
                                 SourceErrorDetails::Other(self.string_region.copy(string))
                             }
                         },
+                        hint: err.hint.as_ref().map(|hint| self.string_region.copy(hint)),
                     };
                     let reference = self.source_error_region.copy_iter(once(err));
                     let boxed = unsafe { Box::from_raw(reference.as_mut_ptr()) };