diff --git a/Cargo.lock b/Cargo.lock index 7a180dee6b4ed..c759a521a6142 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8378,6 +8378,7 @@ dependencies = [ "mz-tls-util", "openssh", "openssl", + "postgres-protocol", "postgres_array", "proptest", "proptest-derive", diff --git a/doc/user/content/ingest-data/patterns/upstream-schema-changes.md b/doc/user/content/ingest-data/patterns/upstream-schema-changes.md index f5a4f6bdd47b5..aa97b5aa51a16 100644 --- a/doc/user/content/ingest-data/patterns/upstream-schema-changes.md +++ b/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 diff --git a/src/adapter/src/error.rs b/src/adapter/src/error.rs index 6f5dc48798faa..39d930ed70dad 100644 --- a/src/adapter/src/error.rs +++ b/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( diff --git a/src/postgres-util/Cargo.toml b/src/postgres-util/Cargo.toml index 76a089362af96..09801ff3e4216 100644 --- a/src/postgres-util/Cargo.toml +++ b/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", diff --git a/src/postgres-util/src/desc.rs b/src/postgres-util/src/desc.rs index 553fe207c45fb..b7894288015e7 100644 --- a/src/postgres-util/src/desc.rs +++ b/src/postgres-util/src/desc.rs @@ -11,7 +11,6 @@ use std::collections::{BTreeMap, BTreeSet}; -use anyhow::bail; use mz_proto::{IntoRustIfSome, RustType, TryFromProtoError}; use proptest::prelude::any; use proptest_derive::Arbitrary; @@ -19,6 +18,8 @@ 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, - ) -> 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 { + 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 } } diff --git a/src/postgres-util/src/lib.rs b/src/postgres-util/src/lib.rs index 20fc5f887ed2c..ea665fc69e0d1 100644 --- a/src/postgres-util/src/lib.rs +++ b/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}; diff --git a/src/postgres-util/src/schema_change.rs b/src/postgres-util/src/schema_change.rs new file mode 100644 index 0000000000000..5b9641c898872 --- /dev/null +++ b/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, +} + +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 { + 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 (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, + } + } +} diff --git a/src/storage-types/src/errors.proto b/src/storage-types/src/errors.proto index 2855df889d27d..17c4bf86981e0 100644 --- a/src/storage-types/src/errors.proto +++ b/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 { diff --git a/src/storage-types/src/errors.rs b/src/storage-types/src/errors.rs index a7f9da24bda0e..e3f0c4e4494ba 100644 --- a/src/storage-types/src/errors.rs +++ b/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>, } impl RustType 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 { 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()) }; diff --git a/src/storage/src/source/kafka.rs b/src/storage/src/source/kafka.rs index e6895207d652d..1a9e51ce48d0c 100644 --- a/src/storage/src/source/kafka.rs +++ b/src/storage/src/source/kafka.rs @@ -536,6 +536,7 @@ fn render_reader<'scope>( // If the topic doesn't exist, that is a definite error let error = Err(SourceError { error: SourceErrorDetails::Initialization(e.to_string().into()), + hint: None, } .into()); let time = data_cap.time().clone(); @@ -594,7 +595,8 @@ fn render_reader<'scope>( ); let error = Err( SourceError{ - error:SourceErrorDetails::Initialization(err_str.into()) + error:SourceErrorDetails::Initialization(err_str.into()), + hint: None, }.into() ); let time = data_cap.time().clone(); @@ -951,6 +953,7 @@ fn render_reader<'scope>( let msg = msg.map_err(|e| { DataflowError::SourceError(Box::new(SourceError { error: SourceErrorDetails::Other(e.to_string().into()), + hint: None, })) }); let update = ((output_index, msg), time, diff); @@ -999,6 +1002,7 @@ fn render_reader<'scope>( let msg = msg.map_err(|e| { DataflowError::SourceError(Box::new(SourceError { error: SourceErrorDetails::Other(e.to_string().into()), + hint: None, })) }); let update = @@ -1831,6 +1835,7 @@ fn render_metadata_fetcher<'scope>( if !PartialOrder::less_equal(&prev_upstream_frontier, &upstream_frontier) { let error = SourceError { error: SourceErrorDetails::Other("topic was recreated".into()), + hint: None, }; update = MetadataUpdate::DefiniteError(error); } @@ -1906,6 +1911,7 @@ fn spawn_metadata_thread( Err(GetPartitionsError::TopicDoesNotExist) => { let error = SourceError { error: SourceErrorDetails::Other("topic was deleted".into()), + hint: None, }; MetadataUpdate::DefiniteError(error) } diff --git a/src/storage/src/source/mysql.rs b/src/storage/src/source/mysql.rs index 40e9af10d7999..f5a399d3e5d96 100644 --- a/src/storage/src/source/mysql.rs +++ b/src/storage/src/source/mysql.rs @@ -315,6 +315,7 @@ impl From for DataflowError { fn from(err: DefiniteError) -> Self { let m = err.to_string().into(); DataflowError::SourceError(Box::new(SourceError { + hint: None, error: match &err { DefiniteError::ValueDecodeError(_) => SourceErrorDetails::Other(m), DefiniteError::TableTruncated(_) => SourceErrorDetails::Other(m), diff --git a/src/storage/src/source/postgres.rs b/src/storage/src/source/postgres.rs index 11b23b36e5a63..0398794def390 100644 --- a/src/storage/src/source/postgres.rs +++ b/src/storage/src/source/postgres.rs @@ -90,6 +90,7 @@ use mz_expr::EvalError; use mz_ore::cast::CastFrom; use mz_ore::error::ErrorExt; use mz_postgres_util::desc::PostgresTableDesc; +use mz_postgres_util::schema_change::SchemaChangeError; use mz_postgres_util::{Client, PostgresError, Sql, query_opt, simple_query_opt, sql}; use mz_repr::{Datum, Diff, GlobalId, Row}; use mz_storage_types::errors::{DataflowError, SourceError, SourceErrorDetails}; @@ -235,7 +236,11 @@ impl SourceRender for PostgresSourceConnection { let errs = snapshot_err.concat(repl_err).map(move |err| { // This update will cause the dataflow to restart let err_string = err.display_with_causes().to_string(); - let update = HealthStatusUpdate::halting(err_string.clone(), None); + let hint = match &err { + ReplicationError::Definite(err) => err.hint(), + ReplicationError::Transient(_) => None, + }; + let update = HealthStatusUpdate::halting(err_string.clone(), hint); let namespace = match err { ReplicationError::Transient(err) @@ -358,9 +363,8 @@ pub enum DefiniteError { "old row missing from replication stream. Did you forget to set REPLICA IDENTITY to FULL for your table?" )] DefaultReplicaIdentity, - #[error("incompatible schema change: {0}")] - // TODO: proper error variants for all the expected schema violations - IncompatibleSchema(String), + #[error("{0}")] + IncompatibleSchema(SchemaChangeError), #[error("invalid UTF8 string: {0:?}")] InvalidUTF8(Vec), #[error("failed to cast raw column: {0}")] @@ -369,10 +373,20 @@ pub enum DefiniteError { UnexpectedBinaryData, } +impl DefiniteError { + fn hint(&self) -> Option { + match self { + DefiniteError::IncompatibleSchema(err) => err.hint(), + _ => None, + } + } +} + impl From for DataflowError { fn from(err: DefiniteError) -> Self { let m = err.to_string().into(); DataflowError::SourceError(Box::new(SourceError { + hint: err.hint().map(Into::into), error: match &err { DefiniteError::SlotCompactedPastResumePoint(_, _) => SourceErrorDetails::Other(m), DefiniteError::TableTruncated => SourceErrorDetails::Other(m), @@ -486,7 +500,7 @@ fn verify_schema( .determine_compatibility(current_desc, &allow_oids_to_change_by_col_num) { Ok(()) => Ok(()), - Err(err) => Err(DefiniteError::IncompatibleSchema(err.to_string())), + Err(err) => Err(DefiniteError::IncompatibleSchema(err)), } } diff --git a/src/storage/src/source/source_reader_pipeline.rs b/src/storage/src/source/source_reader_pipeline.rs index 76fafe5670b3b..003d4b3a28b0d 100644 --- a/src/storage/src/source/source_reader_pipeline.rs +++ b/src/storage/src/source/source_reader_pipeline.rs @@ -352,13 +352,16 @@ where // All errors coming into the data stream are definite. // Downstream consumers of this data will preserve this // status. - let update = HealthStatusUpdate::stalled( - error.to_string(), - Some( + let hint = match error { + DataflowError::SourceError(e) if e.hint.is_some() => { + e.hint.as_deref().map(str::to_string) + } + _ => Some( "retracting the errored value may resume the source" .to_string(), ), - ); + }; + let update = HealthStatusUpdate::stalled(error.to_string(), hint); let status = HealthStatusMessage { id: Some(id), namespace: C::STATUS_NAMESPACE.clone(), diff --git a/src/storage/src/source/sql_server.rs b/src/storage/src/source/sql_server.rs index a02bc7378cbb7..bc8594b48a9eb 100644 --- a/src/storage/src/source/sql_server.rs +++ b/src/storage/src/source/sql_server.rs @@ -109,6 +109,7 @@ impl From for DataflowError { let msg = val.to_string().into(); DataflowError::SourceError(Box::new(SourceError { error: SourceErrorDetails::Other(msg), + hint: None, })) } } diff --git a/test/pg-cdc-old-syntax/alter-table-after-source.td b/test/pg-cdc-old-syntax/alter-table-after-source.td index fe8e3c19fc972..78c57e08f57b7 100644 --- a/test/pg-cdc-old-syntax/alter-table-after-source.td +++ b/test/pg-cdc-old-syntax/alter-table-after-source.td @@ -143,7 +143,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE remove_column DROP COLUMN f2; ! SELECT * from remove_column; -contains:altered +regex:incompatible schema change on public\.remove_column \(oid \d+\): column "f2" was dropped or renamed upstream # @@ -156,7 +156,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_column ALTER COLUMN f2 TYPE CHAR(2); ! SELECT * from alter_column; -contains:altered +regex:incompatible schema change on public\.alter_column \(oid \d+\): the type of column "f2" changed upstream # @@ -169,7 +169,8 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_drop_nullability ALTER COLUMN f1 DROP NOT NULL; ! SELECT * FROM alter_drop_nullability WHERE f1 IS NOT NULL; -contains:altered +regex:incompatible schema change on public\.alter_drop_nullability \(oid \d+\): the NOT NULL constraint on column "f1" was dropped upstream +hint:WITH (EXCLUDE ALL CONSTRAINTS) # We have guaranteed that this column is not null so the optimizer eagerly # returns the empty set. @@ -201,7 +202,8 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_drop_pk DROP CONSTRAINT alter_drop_pk_pkey; ! SELECT f1 FROM alter_drop_pk; -contains:altered +regex:incompatible schema change on public\.alter_drop_pk \(oid \d+\): PRIMARY KEY constraint "alter_drop_pk_pkey" \(f1\) was dropped upstream +hint:WITH (EXCLUDE CONSTRAINTS ('alter_drop_pk_pkey')) before the upstream drop # @@ -230,7 +232,8 @@ ALTER TABLE alter_cycle_pk DROP CONSTRAINT alter_cycle_pk_pkey; ALTER TABLE alter_cycle_pk ADD PRIMARY KEY(f1); ! SELECT * FROM alter_cycle_pk; -contains:altered +regex:incompatible schema change on public\.alter_cycle_pk \(oid \d+\): PRIMARY KEY constraint "alter_cycle_pk_pkey" \(f1\) was renamed or recreated upstream +hint:EXCLUDE CONSTRAINTS ('alter_cycle_pk_pkey') # @@ -260,7 +263,13 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_drop_unique DROP CONSTRAINT alter_drop_unique_f1_key; ! SELECT f1 FROM alter_drop_unique; -contains:altered +regex:incompatible schema change on public\.alter_drop_unique \(oid \d+\): UNIQUE constraint "alter_drop_unique_f1_key" \(f1\) was dropped upstream +hint:WITH (EXCLUDE CONSTRAINTS ('alter_drop_unique_f1_key')) before the upstream drop + +# The same hint is surfaced in the table's status. +> SELECT details->'hints'->>0 = E'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."alter_drop_unique"\n FROM SOURCE (REFERENCE "public"."alter_drop_unique");\nTo make a planned constraint drop a non-event, create the table with WITH (EXCLUDE CONSTRAINTS (''alter_drop_unique_f1_key'')) before the upstream drop.' + FROM mz_internal.mz_source_statuses WHERE name = 'alter_drop_unique' +true # @@ -288,7 +297,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_extend_column ALTER COLUMN f1 TYPE VARCHAR(20); ! SELECT * FROM alter_extend_column; -contains:altered +regex:incompatible schema change on public\.alter_extend_column \(oid \d+\): the type of column "f1" changed upstream # @@ -300,7 +309,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_decimal ALTER COLUMN f1 TYPE DECIMAL(6,1); ! SELECT * FROM alter_decimal; -contains:altered +regex:incompatible schema change on public\.alter_decimal \(oid \d+\): the type of column "f1" changed upstream # @@ -313,7 +322,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_table_rename RENAME TO alter_table_renamed; ! SELECT * FROM alter_table_rename; -contains:altered +regex:incompatible schema change on public\.alter_table_rename \(oid \d+\): table was renamed or moved upstream \(it is now public\.alter_table_renamed with oid # # Alter table rename column @@ -327,7 +336,7 @@ ALTER TABLE alter_table_rename_column RENAME COLUMN f2 TO f1; ALTER TABLE alter_table_rename_column RENAME COLUMN f3 TO f2; ! SELECT * FROM alter_table_rename_column; -contains:altered +regex:incompatible schema change on public\.alter_table_rename_column \(oid \d+\): column # @@ -342,7 +351,7 @@ ALTER TABLE alter_table_change_attnum DROP COLUMN f2; ALTER TABLE alter_table_change_attnum ADD COLUMN f2 VARCHAR(10); ! SELECT * FROM alter_table_change_attnum; -contains:altered +regex:incompatible schema change on public\.alter_table_change_attnum \(oid \d+\): column "f2" changed position upstream > SELECT * from alter_table_supported; 1 1 @@ -368,7 +377,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_table_supported DROP COLUMN f2; ! SELECT * from alter_table_supported; -contains:altered +regex:incompatible schema change on public\.alter_table_supported \(oid \d+\): column "f2" was dropped or renamed upstream # diff --git a/test/pg-cdc-resumption-old-syntax/verify-data.td b/test/pg-cdc-resumption-old-syntax/verify-data.td index 141b062a25d4b..e9ce140f64f70 100644 --- a/test/pg-cdc-resumption-old-syntax/verify-data.td +++ b/test/pg-cdc-resumption-old-syntax/verify-data.td @@ -20,7 +20,7 @@ contains:Source error ! SELECT * FROM alter_fail_drop_col; -contains:has been altered +contains:was dropped or renamed upstream # Ensure non-definite errors are cleared. > SELECT COUNT(*) = 0 FROM mz_internal.mz_source_statuses WHERE error LIKE '%Connection refused%'; diff --git a/test/pg-cdc-resumption/verify-data.td b/test/pg-cdc-resumption/verify-data.td index 141b062a25d4b..e9ce140f64f70 100644 --- a/test/pg-cdc-resumption/verify-data.td +++ b/test/pg-cdc-resumption/verify-data.td @@ -20,7 +20,7 @@ contains:Source error ! SELECT * FROM alter_fail_drop_col; -contains:has been altered +contains:was dropped or renamed upstream # Ensure non-definite errors are cleared. > SELECT COUNT(*) = 0 FROM mz_internal.mz_source_statuses WHERE error LIKE '%Connection refused%'; diff --git a/test/pg-cdc/alter-table-after-source-1.td b/test/pg-cdc/alter-table-after-source-1.td index 2baf3d4a0e516..54a92a34d0a7a 100644 --- a/test/pg-cdc/alter-table-after-source-1.td +++ b/test/pg-cdc/alter-table-after-source-1.td @@ -75,6 +75,10 @@ CREATE TABLE alter_add_unique (f1 INTEGER); ALTER TABLE alter_add_unique REPLICA IDENTITY FULL; INSERT INTO alter_add_unique VALUES (1); +CREATE TABLE alter_rename_unique (f1 INTEGER UNIQUE); +ALTER TABLE alter_rename_unique REPLICA IDENTITY FULL; +INSERT INTO alter_rename_unique VALUES (1); + CREATE TABLE alter_extend_column (f1 VARCHAR(2)); ALTER TABLE alter_extend_column REPLICA IDENTITY FULL; INSERT INTO alter_extend_column VALUES ('ab'); @@ -126,6 +130,7 @@ CREATE PUBLICATION mz_source FOR ALL TABLES; > CREATE TABLE alter_cycle_pk_off FROM SOURCE mz_source (REFERENCE alter_cycle_pk_off); > CREATE TABLE alter_drop_unique FROM SOURCE mz_source (REFERENCE alter_drop_unique); > CREATE TABLE alter_add_unique FROM SOURCE mz_source (REFERENCE alter_add_unique); +> CREATE TABLE alter_rename_unique FROM SOURCE mz_source (REFERENCE alter_rename_unique); > CREATE TABLE alter_extend_column FROM SOURCE mz_source (REFERENCE alter_extend_column); > CREATE TABLE alter_decimal FROM SOURCE mz_source (REFERENCE alter_decimal); > CREATE TABLE alter_table_rename FROM SOURCE mz_source (REFERENCE alter_table_rename); @@ -175,7 +180,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_column ALTER COLUMN f2 TYPE CHAR(2); ! SELECT * from alter_column; -contains:altered +regex:incompatible schema change on public\.alter_column \(oid \d+\): the type of column "f2" changed upstream # @@ -188,7 +193,8 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_drop_nullability ALTER COLUMN f1 DROP NOT NULL; ! SELECT * FROM alter_drop_nullability WHERE f1 IS NOT NULL; -contains:altered +regex:incompatible schema change on public\.alter_drop_nullability \(oid \d+\): the NOT NULL constraint on column "f1" was dropped upstream +hint:WITH (EXCLUDE ALL CONSTRAINTS) # We have guaranteed that this column is not null so the optimizer eagerly # returns the empty set. @@ -220,7 +226,8 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_drop_pk DROP CONSTRAINT alter_drop_pk_pkey; ! SELECT f1 FROM alter_drop_pk; -contains:altered +regex:incompatible schema change on public\.alter_drop_pk \(oid \d+\): PRIMARY KEY constraint "alter_drop_pk_pkey" \(f1\) was dropped upstream +hint:WITH (EXCLUDE CONSTRAINTS ('alter_drop_pk_pkey')) before the upstream drop # @@ -249,7 +256,8 @@ ALTER TABLE alter_cycle_pk DROP CONSTRAINT alter_cycle_pk_pkey; ALTER TABLE alter_cycle_pk ADD PRIMARY KEY(f1); ! SELECT * FROM alter_cycle_pk; -contains:altered +regex:incompatible schema change on public\.alter_cycle_pk \(oid \d+\): PRIMARY KEY constraint "alter_cycle_pk_pkey" \(f1\) was renamed or recreated upstream +hint:EXCLUDE CONSTRAINTS ('alter_cycle_pk_pkey') # @@ -279,7 +287,27 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_drop_unique DROP CONSTRAINT alter_drop_unique_f1_key; ! SELECT f1 FROM alter_drop_unique; -contains:altered +regex:incompatible schema change on public\.alter_drop_unique \(oid \d+\): UNIQUE constraint "alter_drop_unique_f1_key" \(f1\) was dropped upstream +hint:WITH (EXCLUDE CONSTRAINTS ('alter_drop_unique_f1_key')) before the upstream drop + +# The same hint is surfaced in the table's status. +> SELECT details->'hints'->>0 = E'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."alter_drop_unique"\n FROM SOURCE (REFERENCE "public"."alter_drop_unique");\nTo make a planned constraint drop a non-event, create the table with WITH (EXCLUDE CONSTRAINTS (''alter_drop_unique_f1_key'')) before the upstream drop.' + FROM mz_internal.mz_source_statuses WHERE name = 'alter_drop_unique' +true + + +# +# Rename unique + +> SELECT * from alter_rename_unique +1 + +$ postgres-execute connection=postgres://postgres:postgres@postgres +ALTER TABLE alter_rename_unique RENAME CONSTRAINT alter_rename_unique_f1_key TO alter_rename_unique_f1_key_v2; + +! SELECT f1 FROM alter_rename_unique; +regex:incompatible schema change on public\.alter_rename_unique \(oid \d+\): UNIQUE constraint "alter_rename_unique_f1_key" \(f1\) was renamed or recreated upstream +hint:WITH (EXCLUDE CONSTRAINTS ('alter_rename_unique_f1_key')) before the upstream drop # @@ -307,7 +335,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_extend_column ALTER COLUMN f1 TYPE VARCHAR(20); ! SELECT * FROM alter_extend_column; -contains:altered +regex:incompatible schema change on public\.alter_extend_column \(oid \d+\): the type of column "f1" changed upstream # @@ -319,4 +347,4 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_decimal ALTER COLUMN f1 TYPE DECIMAL(6,1); ! SELECT * FROM alter_decimal; -contains:altered +regex:incompatible schema change on public\.alter_decimal \(oid \d+\): the type of column "f1" changed upstream diff --git a/test/pg-cdc/alter-table-after-source-2.td b/test/pg-cdc/alter-table-after-source-2.td index f1e77239b105f..1f7526f6d3ca5 100644 --- a/test/pg-cdc/alter-table-after-source-2.td +++ b/test/pg-cdc/alter-table-after-source-2.td @@ -147,7 +147,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_table_rename RENAME TO alter_table_renamed; ! SELECT * FROM alter_table_rename; -contains:altered +regex:incompatible schema change on public\.alter_table_rename \(oid \d+\): table was renamed or moved upstream \(it is now public\.alter_table_renamed with oid # # Alter table rename colum @@ -161,7 +161,7 @@ ALTER TABLE alter_table_rename_column RENAME COLUMN f2 TO f1; ALTER TABLE alter_table_rename_column RENAME COLUMN f3 TO f2; ! SELECT * FROM alter_table_rename_column; -contains:altered +regex:incompatible schema change on public\.alter_table_rename_column \(oid \d+\): column # # Change column attnum @@ -175,7 +175,7 @@ ALTER TABLE alter_table_change_attnum DROP COLUMN f2; ALTER TABLE alter_table_change_attnum ADD COLUMN f2 VARCHAR(10); ! SELECT * FROM alter_table_change_attnum; -contains:altered +regex:incompatible schema change on public\.alter_table_change_attnum \(oid \d+\): column "f2" changed position upstream > SELECT * from alter_table_supported; 1 1 @@ -201,7 +201,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_table_supported DROP COLUMN f2; ! SELECT * from alter_table_supported; -contains:altered +regex:incompatible schema change on public\.alter_table_supported \(oid \d+\): column "f2" was dropped or renamed upstream # diff --git a/test/source-sink-errors/mzcompose.py b/test/source-sink-errors/mzcompose.py index b3ea946aaee81..bdd93346c87e6 100644 --- a/test/source-sink-errors/mzcompose.py +++ b/test/source-sink-errors/mzcompose.py @@ -497,7 +497,7 @@ def assert_recovery(self, c: Composition) -> None: PgDisruption( name="alter-postgres", breakage=lambda c, _: alter_pg_table(c), - expected_error="source table source1 with oid .+ has been altered", + expected_error=r'incompatible schema change on .+source1 \(oid \d+\): column "f1" was dropped or renamed upstream', fixage=None, ), PgDisruption(