From 87800d3ac4c6e914f05adea302e77c2764f489e9 Mon Sep 17 00:00:00 2001 From: Peter Larsen Date: Thu, 27 Aug 2026 13:35:05 -0400 Subject: [PATCH 1/7] storage: report specific errors for incompatible Postgres schema changes PostgresTableDesc::determine_compatibility computed exactly which facet of the schema diverged but reported only "source table {name} with oid {oid} has been altered", leaving users to guess what happened and how to recover. Diff the schemas instead and report the first mismatch specifically: a dropped or altered PRIMARY KEY/UNIQUE constraint names the constraint and its columns, and each message carries recovery guidance (recreate the table in a new versioned schema and swap views, using EXCLUDE CONSTRAINTS / EXCLUDE ALL CONSTRAINTS / EXCLUDE COLUMNS / TEXT COLUMNS as the pre-drop tool where applicable). Dropped or renamed columns, type changes, position changes, DROP NOT NULL, and table renames each get their own message. The error text is written into the errs shard and becomes the permanent user-visible error for the stalled table, so it must stand on its own. Test assertions on the old error text are updated; unit tests cover the new diff logic. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_018yzDA9ywnCLXweNCzqBm12 --- src/postgres-util/Cargo.toml | 3 + src/postgres-util/src/desc.rs | 334 ++++++++++++++++-- .../alter-table-after-source.td | 24 +- .../verify-data.td | 2 +- test/pg-cdc-resumption/verify-data.td | 2 +- test/pg-cdc/alter-table-after-source-1.td | 14 +- test/pg-cdc/alter-table-after-source-2.td | 8 +- test/source-sink-errors/mzcompose.py | 2 +- 8 files changed, 329 insertions(+), 60 deletions(-) diff --git a/src/postgres-util/Cargo.toml b/src/postgres-util/Cargo.toml index 76a089362af96..c22e412c32108 100644 --- a/src/postgres-util/Cargo.toml +++ b/src/postgres-util/Cargo.toml @@ -42,6 +42,9 @@ tunnel = [ "mz-ore", ] +[dev-dependencies] +mz-ore = { path = "../ore", default-features = false, features = ["test"] } + [build-dependencies] mz-build-tools = { path = "../build-tools", default-features = false, features = ["protobuf-src"] } prost-build.workspace = true diff --git a/src/postgres-util/src/desc.rs b/src/postgres-util/src/desc.rs index 553fe207c45fb..213f68890c5af 100644 --- a/src/postgres-util/src/desc.rs +++ b/src/postgres-util/src/desc.rs @@ -62,6 +62,11 @@ impl PostgresTableDesc { /// Compatibility is defined as returning `true` for /// `PostgresColumnDesc::is_compatible`. /// - `self`'s keys are all present in `other` + /// + /// On incompatibility, the error describes the first mismatch found and, + /// where possible, how to recover from it. The error text becomes the + /// permanent, user-visible error for the stalled table, so it must stand + /// on its own. pub fn determine_compatibility( &self, other: &PostgresTableDesc, @@ -71,49 +76,148 @@ impl PostgresTableDesc { return Ok(()); } - let PostgresTableDesc { - oid: other_oid, - namespace: other_namespace, - name: other_name, - columns: other_cols, - keys: other_keys, - } = other; - - 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 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(()) - } else { + let result = self.diff_incompatibility(other, allow_type_to_change_by_col_num); + if result.is_err() { warn!( "Error validating table in publication. Expected: {:?} Actual: {:?}", &self, other ); + } + result + } + + /// Reports the first incompatibility between `self` (the schema captured + /// when the Materialize table was created) and `other` (the current + /// upstream schema), or `Ok(())` if `other` is a compatible evolution of + /// `self`. + fn diff_incompatibility( + &self, + other: &PostgresTableDesc, + allow_type_to_change_by_col_num: &BTreeSet, + ) -> Result<(), anyhow::Error> { + let table = format!("{}.{}", self.namespace, self.name); + + if self.oid != other.oid || self.namespace != other.namespace || self.name != other.name { + bail!( + "source table {} with oid {} was renamed, dropped, or recreated upstream \ + (it is now {}.{} with oid {}). Materialize binds a table to the upstream \ + table's identity and cannot follow this change. To resume ingesting, \ + recreate the Materialize table against the new upstream table in a new \ + versioned schema and swap your views to it.", + table, + self.oid, + other.namespace, + other.name, + other.oid, + ); + } + + let other_cols_by_name = BTreeMap::from_iter(other.columns.iter().map(|c| (&c.name, c))); + for info in &self.columns { + let Some(other_info) = other_cols_by_name.get(&info.name) else { + bail!( + "column {} of source table {} was dropped or renamed upstream. \ + To resume ingesting, create a replacement table in a new versioned \ + schema (its snapshot captures the current upstream schema), swap \ + your views to it, and drop this table. To make a planned column \ + drop a non-event, create the replacement table with \ + WITH (EXCLUDE COLUMNS ({})) before the upstream drop.", + quoted(&info.name), + table, + quoted(&info.name), + ); + }; + let allow_type_change = allow_type_to_change_by_col_num.contains(&info.col_num); + if info.is_compatible(other_info, allow_type_change) { + continue; + } + if info.col_num != other_info.col_num { + bail!( + "column {} of source table {} changed position upstream (the column \ + or table was likely dropped and recreated). To resume ingesting, \ + recreate the Materialize table in a new versioned schema and swap \ + your views to it.", + quoted(&info.name), + table, + ); + } + if !allow_type_change + && (info.type_oid != other_info.type_oid || info.type_mod != other_info.type_mod) + { + bail!( + "the type of column {} of source table {} changed upstream. To ingest \ + the column as text regardless of its upstream type, recreate the \ + Materialize table with WITH (TEXT COLUMNS ({})) in a new versioned \ + schema and swap your views to it.", + quoted(&info.name), + table, + quoted(&info.name), + ); + } + if !info.nullable && other_info.nullable { + bail!( + "the NOT NULL constraint on column {} of source table {} was dropped \ + upstream. Materialize relies on this constraint and cannot continue \ + ingesting the table. To resume ingesting, create a replacement table \ + in a new versioned schema (its snapshot captures the current upstream \ + schema, where the column is nullable), swap your views to it, and \ + drop this table. To make planned constraint drops a non-event, create \ + the replacement table with WITH (EXCLUDE ALL CONSTRAINTS).", + quoted(&info.name), + table, + ); + } + bail!( + "column {} of source table {} was altered upstream. To resume ingesting, \ + recreate the Materialize table in a new versioned schema and swap your \ + views to it.", + quoted(&info.name), + table, + ); + } + + if let Some(key) = self.keys.difference(&other.keys).next() { + let col_names = 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::>() + .join(", "); + let kind = if key.is_primary { + "PRIMARY KEY" + } else { + "UNIQUE" + }; bail!( - "source table {} with oid {} has been altered", - self.name, - self.oid - ) + "{} constraint {} ({}) on source table {} was dropped or altered upstream. \ + Materialize relies on this constraint and cannot continue ingesting the \ + table. To resume ingesting, create a replacement table in a new versioned \ + schema (its snapshot captures the current upstream schema, without this \ + constraint), swap your views to it, and drop this table. To make a \ + planned constraint drop a non-event, create the replacement table with \ + WITH (EXCLUDE CONSTRAINTS ('{}')) before the upstream drop.", + kind, + quoted(&key.name), + col_names, + table, + key.name, + ); } + + Ok(()) } } +/// Formats an identifier for inclusion in an error message. +fn quoted(name: &str) -> String { + format!("\"{}\"", name) +} + impl RustType for PostgresTableDesc { fn into_proto(&self) -> ProtoPostgresTableDesc { ProtoPostgresTableDesc { @@ -278,3 +382,165 @@ impl RustType for PostgresKeyDesc { }) } } + +#[cfg(test)] +mod tests { + use super::*; + + fn column(name: &str, col_num: u16, nullable: bool) -> PostgresColumnDesc { + PostgresColumnDesc { + name: name.to_string(), + col_num, + type_oid: 23, + type_mod: -1, + nullable, + } + } + + fn key(oid: u32, name: &str, cols: Vec, is_primary: bool) -> PostgresKeyDesc { + PostgresKeyDesc { + oid, + name: name.to_string(), + cols, + is_primary, + nulls_not_distinct: false, + } + } + + fn table(columns: Vec, keys: Vec) -> PostgresTableDesc { + PostgresTableDesc { + oid: 100, + namespace: "public".to_string(), + name: "users".to_string(), + columns, + keys: keys.into_iter().collect(), + } + } + + #[mz_ore::test] + fn compatible_evolutions() { + let desc = table( + vec![column("id", 1, false)], + vec![key(200, "users_pkey", vec![1], true)], + ); + + // Identical. + desc.determine_compatibility(&desc, &BTreeSet::new()) + .unwrap(); + + // Extra upstream column and extra upstream key are non-events. + let mut evolved = desc.clone(); + evolved.columns.push(column("extra", 2, true)); + evolved + .keys + .insert(key(201, "users_extra_key", vec![2], false)); + desc.determine_compatibility(&evolved, &BTreeSet::new()) + .unwrap(); + + // Upstream SET NOT NULL on a column we recorded as nullable. + let desc = table(vec![column("id", 1, true)], vec![]); + let evolved = table(vec![column("id", 1, false)], vec![]); + desc.determine_compatibility(&evolved, &BTreeSet::new()) + .unwrap(); + } + + #[mz_ore::test] + fn dropped_key_names_constraint() { + let desc = table( + vec![column("id", 1, false), column("wallet", 2, false)], + vec![ + key(200, "users_pkey", vec![1], true), + key(201, "users_wallet_id_key", vec![2], false), + ], + ); + let mut evolved = desc.clone(); + evolved + .keys + .remove(&key(201, "users_wallet_id_key", vec![2], false)); + + let err = desc + .determine_compatibility(&evolved, &BTreeSet::new()) + .unwrap_err() + .to_string(); + assert!( + err.contains("UNIQUE constraint \"users_wallet_id_key\" (wallet)"), + "{err}" + ); + assert!( + err.contains("EXCLUDE CONSTRAINTS ('users_wallet_id_key')"), + "{err}" + ); + + // Same-name key with a different constraint oid (drop + recreate) also + // reads as dropped or altered. + let mut recreated = desc.clone(); + recreated + .keys + .remove(&key(200, "users_pkey", vec![1], true)); + recreated.keys.insert(key(300, "users_pkey", vec![1], true)); + let err = desc + .determine_compatibility(&recreated, &BTreeSet::new()) + .unwrap_err() + .to_string(); + assert!( + err.contains("PRIMARY KEY constraint \"users_pkey\" (id)"), + "{err}" + ); + } + + #[mz_ore::test] + fn column_incompatibilities() { + let desc = table(vec![column("id", 1, false)], vec![]); + + // Dropped column. + let evolved = table(vec![column("other", 1, false)], vec![]); + let err = desc + .determine_compatibility(&evolved, &BTreeSet::new()) + .unwrap_err() + .to_string(); + assert!( + err.contains("column \"id\" of source table public.users was dropped or renamed"), + "{err}" + ); + + // DROP NOT NULL. + let evolved = table(vec![column("id", 1, true)], vec![]); + let err = desc + .determine_compatibility(&evolved, &BTreeSet::new()) + .unwrap_err() + .to_string(); + assert!( + err.contains("NOT NULL constraint on column \"id\""), + "{err}" + ); + assert!(err.contains("EXCLUDE ALL CONSTRAINTS"), "{err}"); + + // Type change, without and with a TEXT COLUMNS exemption. + let mut evolved = table(vec![column("id", 1, false)], vec![]); + evolved.columns[0].type_oid = 25; + let err = desc + .determine_compatibility(&evolved, &BTreeSet::new()) + .unwrap_err() + .to_string(); + assert!(err.contains("the type of column \"id\""), "{err}"); + desc.determine_compatibility(&evolved, &BTreeSet::from([1])) + .unwrap(); + + // Position change. + let evolved = table(vec![column("id", 3, false)], vec![]); + let err = desc + .determine_compatibility(&evolved, &BTreeSet::new()) + .unwrap_err() + .to_string(); + assert!(err.contains("changed position upstream"), "{err}"); + + // Table renamed or recreated. + let mut evolved = desc.clone(); + evolved.name = "users_renamed".to_string(); + let err = desc + .determine_compatibility(&evolved, &BTreeSet::new()) + .unwrap_err() + .to_string(); + assert!(err.contains("renamed, dropped, or recreated"), "{err}"); + } +} 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..1fcd9cce2ee65 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 +contains:column "f2" of source table public.remove_column 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 +contains:the type of column "f2" of source table public.alter_column changed upstream # @@ -169,7 +169,7 @@ $ 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 +contains:the NOT NULL constraint on column "f1" of source table public.alter_drop_nullability was dropped upstream # We have guaranteed that this column is not null so the optimizer eagerly # returns the empty set. @@ -201,7 +201,7 @@ $ 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 +contains:PRIMARY KEY constraint "alter_drop_pk_pkey" (f1) on source table public.alter_drop_pk was dropped or altered upstream # @@ -230,7 +230,7 @@ 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 +contains:PRIMARY KEY constraint "alter_cycle_pk_pkey" (f1) on source table public.alter_cycle_pk was dropped or altered upstream # @@ -260,7 +260,7 @@ $ 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 +contains:UNIQUE constraint "alter_drop_unique_f1_key" (f1) on source table public.alter_drop_unique was dropped or altered upstream # @@ -288,7 +288,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 +contains:the type of column "f1" of source table public.alter_extend_column changed upstream # @@ -300,7 +300,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 +contains:the type of column "f1" of source table public.alter_decimal changed upstream # @@ -313,7 +313,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 +contains:source table public.alter_table_rename with oid # # Alter table rename column @@ -327,7 +327,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 +contains:changed position upstream # @@ -342,7 +342,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 +contains:column "f2" of source table public.alter_table_change_attnum changed position upstream > SELECT * from alter_table_supported; 1 1 @@ -368,7 +368,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_table_supported DROP COLUMN f2; ! SELECT * from alter_table_supported; -contains:altered +contains:column "f2" of source table public.alter_table_supported 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..0f093a539b5e3 100644 --- a/test/pg-cdc/alter-table-after-source-1.td +++ b/test/pg-cdc/alter-table-after-source-1.td @@ -175,7 +175,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_column ALTER COLUMN f2 TYPE CHAR(2); ! SELECT * from alter_column; -contains:altered +contains:the type of column "f2" of source table public.alter_column changed upstream # @@ -188,7 +188,7 @@ $ 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 +contains:the NOT NULL constraint on column "f1" of source table public.alter_drop_nullability was dropped upstream # We have guaranteed that this column is not null so the optimizer eagerly # returns the empty set. @@ -220,7 +220,7 @@ $ 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 +contains:PRIMARY KEY constraint "alter_drop_pk_pkey" (f1) on source table public.alter_drop_pk was dropped or altered upstream # @@ -249,7 +249,7 @@ 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 +contains:PRIMARY KEY constraint "alter_cycle_pk_pkey" (f1) on source table public.alter_cycle_pk was dropped or altered upstream # @@ -279,7 +279,7 @@ $ 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 +contains:UNIQUE constraint "alter_drop_unique_f1_key" (f1) on source table public.alter_drop_unique was dropped or altered upstream # @@ -307,7 +307,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 +contains:the type of column "f1" of source table public.alter_extend_column changed upstream # @@ -319,4 +319,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 +contains:the type of column "f1" of source table public.alter_decimal 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..52631de897086 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 +contains:source table public.alter_table_rename 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 +contains:changed position upstream # # 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 +contains:column "f2" of source table public.alter_table_change_attnum 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 +contains:column "f2" of source table public.alter_table_supported was dropped or renamed upstream # diff --git a/test/source-sink-errors/mzcompose.py b/test/source-sink-errors/mzcompose.py index b3ea946aaee81..c877b4ba3516f 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='column "f1" of source table .+source1 was dropped or renamed upstream', fixage=None, ), PgDisruption( From 379eda500b8826437553b09bb956e41e31ab89af Mon Sep 17 00:00:00 2001 From: Peter Larsen Date: Tue, 8 Sep 2026 12:04:40 -0400 Subject: [PATCH 2/7] storage: carry schema-change recovery steps as error hints Replace the anyhow string from PostgresTableDesc::determine_compatibility with SchemaChangeError, whose Display is the diagnosis (incompatible schema change on : ) and whose hint holds the recovery steps, including the CREATE SCHEMA / CREATE TABLE .. FROM SOURCE statements to run. SourceError gains an optional hint (a new proto field in the errs shard encoding) so the hint survives persistence; the adapter surfaces it as the HINT of the SQL error, and per-table source statuses prefer it over the generic retraction hint. Renamed constraints are told apart from dropped and recreated ones, and the hint names the constraint as it now exists upstream. Testdrive asserts the message and hint for every schema change case. The unit tests are dropped in favor of that coverage. Co-Authored-By: Claude Fable 5.1 --- .../patterns/upstream-schema-changes.md | 7 +- src/adapter/src/error.rs | 1 + src/postgres-util/Cargo.toml | 3 - src/postgres-util/src/desc.rs | 481 +++++++----------- src/storage-types/src/errors.proto | 1 + src/storage-types/src/errors.rs | 4 + src/storage/src/source/kafka.rs | 8 +- src/storage/src/source/mysql.rs | 1 + src/storage/src/source/postgres.rs | 25 +- .../src/source/source_reader_pipeline.rs | 11 +- src/storage/src/source/sql_server.rs | 1 + .../alter-table-after-source.td | 36 +- test/pg-cdc/alter-table-after-source-1.td | 45 +- test/pg-cdc/alter-table-after-source-2.td | 12 +- test/source-sink-errors/mzcompose.py | 2 +- 15 files changed, 305 insertions(+), 333 deletions(-) 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..31f124dda486e 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,12 @@ 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: column "priority" was dropped or renamed upstream +HINT: To keep ingesting without this column, recreate the table in a new versioned schema, then swap your views to the new table: + CREATE SCHEMA v2; + CREATE TABLE v2.orders + FROM SOURCE (REFERENCE public.orders); + To make a planned column drop a non-event, create the table with WITH (EXCLUDE COLUMNS ("priority")) before the upstream drop. ``` 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 c22e412c32108..76a089362af96 100644 --- a/src/postgres-util/Cargo.toml +++ b/src/postgres-util/Cargo.toml @@ -42,9 +42,6 @@ tunnel = [ "mz-ore", ] -[dev-dependencies] -mz-ore = { path = "../ore", default-features = false, features = ["test"] } - [build-dependencies] mz-build-tools = { path = "../build-tools", default-features = false, features = ["protobuf-src"] } prost-build.workspace = true diff --git a/src/postgres-util/src/desc.rs b/src/postgres-util/src/desc.rs index 213f68890c5af..b9a9ba1eaa6c9 100644 --- a/src/postgres-util/src/desc.rs +++ b/src/postgres-util/src/desc.rs @@ -11,13 +11,12 @@ use std::collections::{BTreeMap, BTreeSet}; -use anyhow::bail; +use mz_ore::str::StrExt; 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; include!(concat!(env!("OUT_DIR"), "/mz_postgres_util.desc.rs")); @@ -52,6 +51,29 @@ pub struct PostgresTableDesc { pub keys: BTreeSet, } +/// An upstream schema change that Materialize cannot follow. +/// +/// `Display` renders the diagnosis. `hint` carries the recovery steps and is +/// surfaced separately: as the `HINT` of a SQL error and in the source status. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct SchemaChangeError { + pub table: String, + pub change: String, + pub hint: String, +} + +impl std::fmt::Display for SchemaChangeError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "incompatible schema change on {}: {}", + self.table, self.change + ) + } +} + +impl std::error::Error for SchemaChangeError {} + impl PostgresTableDesc { /// Determines if two `PostgresTableDesc` are compatible with one another in /// a way that Materialize can handle. @@ -63,159 +85,192 @@ impl PostgresTableDesc { /// `PostgresColumnDesc::is_compatible`. /// - `self`'s keys are all present in `other` /// - /// On incompatibility, the error describes the first mismatch found and, - /// where possible, how to recover from it. The error text becomes the - /// permanent, user-visible error for the stalled table, so it must stand - /// on its own. + /// 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 result = self.diff_incompatibility(other, allow_type_to_change_by_col_num); - if result.is_err() { - warn!( - "Error validating table in publication. Expected: {:?} Actual: {:?}", - &self, other - ); + if self.oid != other.oid || self.namespace != other.namespace || self.name != other.name { + let reference = format!("{}.{}", other.namespace, other.name); + return Err(self.error( + format!( + "table was renamed or moved upstream (it is now {} with oid {})", + reference, other.oid + ), + recreate_hint( + "To keep ingesting from the upstream table as it now exists", + &other.name, + &reference, + None, + ), + )); } - result + + 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); + self.check_column( + column, + other_cols_by_name.get(&column.name).copied(), + allow_type_change, + )?; + } + + if let Some(key) = self.keys.difference(&other.keys).next() { + return Err(self.key_error(key, other)); + } + + Ok(()) } - /// Reports the first incompatibility between `self` (the schema captured - /// when the Materialize table was created) and `other` (the current - /// upstream schema), or `Ok(())` if `other` is a compatible evolution of - /// `self`. - fn diff_incompatibility( + fn check_column( &self, - other: &PostgresTableDesc, - allow_type_to_change_by_col_num: &BTreeSet, - ) -> Result<(), anyhow::Error> { - let table = format!("{}.{}", self.namespace, self.name); + column: &PostgresColumnDesc, + other: Option<&PostgresColumnDesc>, + allow_type_change: bool, + ) -> Result<(), SchemaChangeError> { + let name = column.name.quoted(); + let Some(other) = other else { + return Err(self.error( + format!("column {name} was dropped or renamed upstream"), + format!( + "{}\nTo make a planned column drop a non-event, create the table with \ + WITH (EXCLUDE COLUMNS ({name})) before the upstream drop.", + self.recreate_hint("To keep ingesting without this column", None) + ), + )); + }; + if column.is_compatible(other, allow_type_change) { + return Ok(()); + } + if column.col_num != other.col_num { + return Err(self.error( + format!( + "column {name} changed position upstream (the column or table was likely \ + dropped and recreated)" + ), + self.recreate_hint("To keep ingesting", None), + )); + } + if !allow_type_change + && (column.type_oid != other.type_oid || column.type_mod != other.type_mod) + { + return Err(self.error( + format!("the type of column {name} changed upstream"), + self.recreate_hint( + "To ingest the column as text regardless of its upstream type", + Some(&format!("TEXT COLUMNS ({name})")), + ), + )); + } + if !column.nullable && other.nullable { + return Err(self.error( + format!("the NOT NULL constraint on column {name} was dropped upstream"), + self.recreate_hint( + "To keep ingesting without this constraint", + Some("EXCLUDE ALL CONSTRAINTS"), + ), + )); + } + Err(self.error( + format!("column {name} was altered upstream"), + self.recreate_hint("To keep ingesting", None), + )) + } - if self.oid != other.oid || self.namespace != other.namespace || self.name != other.name { - bail!( - "source table {} with oid {} was renamed, dropped, or recreated upstream \ - (it is now {}.{} with oid {}). Materialize binds a table to the upstream \ - table's identity and cannot follow this change. To resume ingesting, \ - recreate the Materialize table against the new upstream table in a new \ - versioned schema and swap your views to it.", - table, - self.oid, - other.namespace, - other.name, - other.oid, + fn key_error(&self, key: &PostgresKeyDesc, other: &PostgresTableDesc) -> SchemaChangeError { + let kind = if key.is_primary { + "PRIMARY KEY" + } else { + "UNIQUE" + }; + let cols = 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::>() + .join(", "); + let constraint = format!("{kind} constraint {} ({cols})", key.name.quoted()); + let exclude = |name: &str| format!("EXCLUDE CONSTRAINTS ('{}')", name.replace('\'', "''")); + + if let Some(renamed) = other.keys.iter().find(|k| k.oid == key.oid) { + return self.error( + format!( + "{constraint} was renamed upstream to {}", + renamed.name.quoted() + ), + self.recreate_hint( + "To keep ingesting without this constraint", + Some(&exclude(&renamed.name)), + ), ); } - - let other_cols_by_name = BTreeMap::from_iter(other.columns.iter().map(|c| (&c.name, c))); - for info in &self.columns { - let Some(other_info) = other_cols_by_name.get(&info.name) else { - bail!( - "column {} of source table {} was dropped or renamed upstream. \ - To resume ingesting, create a replacement table in a new versioned \ - schema (its snapshot captures the current upstream schema), swap \ - your views to it, and drop this table. To make a planned column \ - drop a non-event, create the replacement table with \ - WITH (EXCLUDE COLUMNS ({})) before the upstream drop.", - quoted(&info.name), - table, - quoted(&info.name), - ); - }; - let allow_type_change = allow_type_to_change_by_col_num.contains(&info.col_num); - if info.is_compatible(other_info, allow_type_change) { - continue; - } - if info.col_num != other_info.col_num { - bail!( - "column {} of source table {} changed position upstream (the column \ - or table was likely dropped and recreated). To resume ingesting, \ - recreate the Materialize table in a new versioned schema and swap \ - your views to it.", - quoted(&info.name), - table, - ); - } - if !allow_type_change - && (info.type_oid != other_info.type_oid || info.type_mod != other_info.type_mod) - { - bail!( - "the type of column {} of source table {} changed upstream. To ingest \ - the column as text regardless of its upstream type, recreate the \ - Materialize table with WITH (TEXT COLUMNS ({})) in a new versioned \ - schema and swap your views to it.", - quoted(&info.name), - table, - quoted(&info.name), - ); - } - if !info.nullable && other_info.nullable { - bail!( - "the NOT NULL constraint on column {} of source table {} was dropped \ - upstream. Materialize relies on this constraint and cannot continue \ - ingesting the table. To resume ingesting, create a replacement table \ - in a new versioned schema (its snapshot captures the current upstream \ - schema, where the column is nullable), swap your views to it, and \ - drop this table. To make planned constraint drops a non-event, create \ - the replacement table with WITH (EXCLUDE ALL CONSTRAINTS).", - quoted(&info.name), - table, - ); - } - bail!( - "column {} of source table {} was altered upstream. To resume ingesting, \ - recreate the Materialize table in a new versioned schema and swap your \ - views to it.", - quoted(&info.name), - table, + if other.keys.iter().any(|k| k.name == key.name) { + return self.error( + format!("{constraint} was dropped and recreated upstream"), + self.recreate_hint( + "To keep ingesting without this constraint", + Some(&exclude(&key.name)), + ), ); } + self.error( + format!("{constraint} was dropped upstream"), + format!( + "{}\nTo make a planned constraint drop a non-event, create the table with \ + WITH ({}) before the upstream drop.", + self.recreate_hint("To keep ingesting without this constraint", None), + exclude(&key.name), + ), + ) + } - if let Some(key) = self.keys.difference(&other.keys).next() { - let col_names = 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::>() - .join(", "); - let kind = if key.is_primary { - "PRIMARY KEY" - } else { - "UNIQUE" - }; - bail!( - "{} constraint {} ({}) on source table {} was dropped or altered upstream. \ - Materialize relies on this constraint and cannot continue ingesting the \ - table. To resume ingesting, create a replacement table in a new versioned \ - schema (its snapshot captures the current upstream schema, without this \ - constraint), swap your views to it, and drop this table. To make a \ - planned constraint drop a non-event, create the replacement table with \ - WITH (EXCLUDE CONSTRAINTS ('{}')) before the upstream drop.", - kind, - quoted(&key.name), - col_names, - table, - key.name, - ); + fn error(&self, change: String, hint: String) -> SchemaChangeError { + SchemaChangeError { + table: format!("{}.{}", self.namespace, self.name), + change, + hint, } + } - Ok(()) + fn recreate_hint(&self, lead: &str, with_clause: Option<&str>) -> String { + recreate_hint( + lead, + &self.name, + &format!("{}.{}", self.namespace, self.name), + with_clause, + ) } } -/// Formats an identifier for inclusion in an error message. -fn quoted(name: &str) -> String { - format!("\"{}\"", name) +fn recreate_hint( + lead: &str, + table_name: &str, + reference: &str, + with_clause: Option<&str>, +) -> String { + let mut hint = format!( + "{lead}, recreate the table in a new versioned schema, then swap your views to the \ + new table:\n CREATE SCHEMA v2;\n CREATE TABLE v2.{table_name}\n \ + FROM SOURCE (REFERENCE {reference})" + ); + if let Some(with_clause) = with_clause { + hint.push_str(&format!("\n WITH ({with_clause})")); + } + hint.push(';'); + hint } impl RustType for PostgresTableDesc { @@ -382,165 +437,3 @@ impl RustType for PostgresKeyDesc { }) } } - -#[cfg(test)] -mod tests { - use super::*; - - fn column(name: &str, col_num: u16, nullable: bool) -> PostgresColumnDesc { - PostgresColumnDesc { - name: name.to_string(), - col_num, - type_oid: 23, - type_mod: -1, - nullable, - } - } - - fn key(oid: u32, name: &str, cols: Vec, is_primary: bool) -> PostgresKeyDesc { - PostgresKeyDesc { - oid, - name: name.to_string(), - cols, - is_primary, - nulls_not_distinct: false, - } - } - - fn table(columns: Vec, keys: Vec) -> PostgresTableDesc { - PostgresTableDesc { - oid: 100, - namespace: "public".to_string(), - name: "users".to_string(), - columns, - keys: keys.into_iter().collect(), - } - } - - #[mz_ore::test] - fn compatible_evolutions() { - let desc = table( - vec![column("id", 1, false)], - vec![key(200, "users_pkey", vec![1], true)], - ); - - // Identical. - desc.determine_compatibility(&desc, &BTreeSet::new()) - .unwrap(); - - // Extra upstream column and extra upstream key are non-events. - let mut evolved = desc.clone(); - evolved.columns.push(column("extra", 2, true)); - evolved - .keys - .insert(key(201, "users_extra_key", vec![2], false)); - desc.determine_compatibility(&evolved, &BTreeSet::new()) - .unwrap(); - - // Upstream SET NOT NULL on a column we recorded as nullable. - let desc = table(vec![column("id", 1, true)], vec![]); - let evolved = table(vec![column("id", 1, false)], vec![]); - desc.determine_compatibility(&evolved, &BTreeSet::new()) - .unwrap(); - } - - #[mz_ore::test] - fn dropped_key_names_constraint() { - let desc = table( - vec![column("id", 1, false), column("wallet", 2, false)], - vec![ - key(200, "users_pkey", vec![1], true), - key(201, "users_wallet_id_key", vec![2], false), - ], - ); - let mut evolved = desc.clone(); - evolved - .keys - .remove(&key(201, "users_wallet_id_key", vec![2], false)); - - let err = desc - .determine_compatibility(&evolved, &BTreeSet::new()) - .unwrap_err() - .to_string(); - assert!( - err.contains("UNIQUE constraint \"users_wallet_id_key\" (wallet)"), - "{err}" - ); - assert!( - err.contains("EXCLUDE CONSTRAINTS ('users_wallet_id_key')"), - "{err}" - ); - - // Same-name key with a different constraint oid (drop + recreate) also - // reads as dropped or altered. - let mut recreated = desc.clone(); - recreated - .keys - .remove(&key(200, "users_pkey", vec![1], true)); - recreated.keys.insert(key(300, "users_pkey", vec![1], true)); - let err = desc - .determine_compatibility(&recreated, &BTreeSet::new()) - .unwrap_err() - .to_string(); - assert!( - err.contains("PRIMARY KEY constraint \"users_pkey\" (id)"), - "{err}" - ); - } - - #[mz_ore::test] - fn column_incompatibilities() { - let desc = table(vec![column("id", 1, false)], vec![]); - - // Dropped column. - let evolved = table(vec![column("other", 1, false)], vec![]); - let err = desc - .determine_compatibility(&evolved, &BTreeSet::new()) - .unwrap_err() - .to_string(); - assert!( - err.contains("column \"id\" of source table public.users was dropped or renamed"), - "{err}" - ); - - // DROP NOT NULL. - let evolved = table(vec![column("id", 1, true)], vec![]); - let err = desc - .determine_compatibility(&evolved, &BTreeSet::new()) - .unwrap_err() - .to_string(); - assert!( - err.contains("NOT NULL constraint on column \"id\""), - "{err}" - ); - assert!(err.contains("EXCLUDE ALL CONSTRAINTS"), "{err}"); - - // Type change, without and with a TEXT COLUMNS exemption. - let mut evolved = table(vec![column("id", 1, false)], vec![]); - evolved.columns[0].type_oid = 25; - let err = desc - .determine_compatibility(&evolved, &BTreeSet::new()) - .unwrap_err() - .to_string(); - assert!(err.contains("the type of column \"id\""), "{err}"); - desc.determine_compatibility(&evolved, &BTreeSet::from([1])) - .unwrap(); - - // Position change. - let evolved = table(vec![column("id", 3, false)], vec![]); - let err = desc - .determine_compatibility(&evolved, &BTreeSet::new()) - .unwrap_err() - .to_string(); - assert!(err.contains("changed position upstream"), "{err}"); - - // Table renamed or recreated. - let mut evolved = desc.clone(); - evolved.name = "users_renamed".to_string(); - let err = desc - .determine_compatibility(&evolved, &BTreeSet::new()) - .unwrap_err() - .to_string(); - assert!(err.contains("renamed, dropped, or recreated"), "{err}"); - } -} 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..320bfec0e949d 100644 --- a/src/storage/src/source/postgres.rs +++ b/src/storage/src/source/postgres.rs @@ -89,7 +89,7 @@ use itertools::Itertools as _; use mz_expr::EvalError; use mz_ore::cast::CastFrom; use mz_ore::error::ErrorExt; -use mz_postgres_util::desc::PostgresTableDesc; +use mz_postgres_util::desc::{PostgresTableDesc, 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 +235,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 +362,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 +372,20 @@ pub enum DefiniteError { UnexpectedBinaryData, } +impl DefiniteError { + fn hint(&self) -> Option { + match self { + DefiniteError::IncompatibleSchema(err) => Some(err.hint.clone()), + _ => 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 +499,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 1fcd9cce2ee65..d8a34ca6fb979 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,8 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE remove_column DROP COLUMN f2; ! SELECT * from remove_column; -contains:column "f2" of source table public.remove_column was dropped or renamed upstream +contains:incompatible schema change on public.remove_column: column "f2" was dropped or renamed upstream +hint:WITH (EXCLUDE COLUMNS ("f2")) before the upstream drop # @@ -156,7 +157,8 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_column ALTER COLUMN f2 TYPE CHAR(2); ! SELECT * from alter_column; -contains:the type of column "f2" of source table public.alter_column changed upstream +contains:incompatible schema change on public.alter_column: the type of column "f2" changed upstream +hint:WITH (TEXT COLUMNS ("f2")) # @@ -169,7 +171,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:the NOT NULL constraint on column "f1" of source table public.alter_drop_nullability was dropped upstream +contains:incompatible schema change on public.alter_drop_nullability: 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 +204,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:PRIMARY KEY constraint "alter_drop_pk_pkey" (f1) on source table public.alter_drop_pk was dropped or altered upstream +contains:incompatible schema change on public.alter_drop_pk: 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 +234,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:PRIMARY KEY constraint "alter_cycle_pk_pkey" (f1) on source table public.alter_cycle_pk was dropped or altered upstream +contains:incompatible schema change on public.alter_cycle_pk: PRIMARY KEY constraint "alter_cycle_pk_pkey" (f1) was dropped +hint:EXCLUDE CONSTRAINTS ('alter_cycle_pk_pkey') # @@ -260,7 +265,8 @@ $ 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:UNIQUE constraint "alter_drop_unique_f1_key" (f1) on source table public.alter_drop_unique was dropped or altered upstream +contains:incompatible schema change on public.alter_drop_unique: UNIQUE constraint "alter_drop_unique_f1_key" (f1) was dropped upstream +hint:WITH (EXCLUDE CONSTRAINTS ('alter_drop_unique_f1_key')) before the upstream drop # @@ -288,7 +294,8 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_extend_column ALTER COLUMN f1 TYPE VARCHAR(20); ! SELECT * FROM alter_extend_column; -contains:the type of column "f1" of source table public.alter_extend_column changed upstream +contains:incompatible schema change on public.alter_extend_column: the type of column "f1" changed upstream +hint:WITH (TEXT COLUMNS ("f1")) # @@ -300,7 +307,8 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_decimal ALTER COLUMN f1 TYPE DECIMAL(6,1); ! SELECT * FROM alter_decimal; -contains:the type of column "f1" of source table public.alter_decimal changed upstream +contains:incompatible schema change on public.alter_decimal: the type of column "f1" changed upstream +hint:WITH (TEXT COLUMNS ("f1")) # @@ -313,7 +321,8 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_table_rename RENAME TO alter_table_renamed; ! SELECT * FROM alter_table_rename; -contains:source table public.alter_table_rename with oid +contains:incompatible schema change on public.alter_table_rename: table was renamed or moved upstream (it is now public.alter_table_renamed with oid +hint:FROM SOURCE (REFERENCE public.alter_table_renamed) # # Alter table rename column @@ -327,7 +336,8 @@ 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:changed position upstream +contains:incompatible schema change on public.alter_table_rename_column: column +hint:CREATE SCHEMA v2; # @@ -342,7 +352,8 @@ 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:column "f2" of source table public.alter_table_change_attnum changed position upstream +contains:incompatible schema change on public.alter_table_change_attnum: column "f2" changed position upstream +hint:CREATE SCHEMA v2; > SELECT * from alter_table_supported; 1 1 @@ -368,7 +379,8 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_table_supported DROP COLUMN f2; ! SELECT * from alter_table_supported; -contains:column "f2" of source table public.alter_table_supported was dropped or renamed upstream +contains:incompatible schema change on public.alter_table_supported: column "f2" was dropped or renamed upstream +hint:WITH (EXCLUDE COLUMNS ("f2")) before the upstream drop # diff --git a/test/pg-cdc/alter-table-after-source-1.td b/test/pg-cdc/alter-table-after-source-1.td index 0f093a539b5e3..e6d1df246825f 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,8 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_column ALTER COLUMN f2 TYPE CHAR(2); ! SELECT * from alter_column; -contains:the type of column "f2" of source table public.alter_column changed upstream +contains:incompatible schema change on public.alter_column: the type of column "f2" changed upstream +hint:WITH (TEXT COLUMNS ("f2")) # @@ -188,7 +194,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:the NOT NULL constraint on column "f1" of source table public.alter_drop_nullability was dropped upstream +contains:incompatible schema change on public.alter_drop_nullability: 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 +227,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:PRIMARY KEY constraint "alter_drop_pk_pkey" (f1) on source table public.alter_drop_pk was dropped or altered upstream +contains:incompatible schema change on public.alter_drop_pk: 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 +257,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:PRIMARY KEY constraint "alter_cycle_pk_pkey" (f1) on source table public.alter_cycle_pk was dropped or altered upstream +contains:incompatible schema change on public.alter_cycle_pk: PRIMARY KEY constraint "alter_cycle_pk_pkey" (f1) was dropped +hint:EXCLUDE CONSTRAINTS ('alter_cycle_pk_pkey') # @@ -279,7 +288,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:UNIQUE constraint "alter_drop_unique_f1_key" (f1) on source table public.alter_drop_unique was dropped or altered upstream +contains:incompatible schema change on public.alter_drop_unique: 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' LIKE '%EXCLUDE CONSTRAINTS (''alter_drop_unique_f1_key'')%' + 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; +contains:incompatible schema change on public.alter_rename_unique: UNIQUE constraint "alter_rename_unique_f1_key" (f1) was renamed upstream to "alter_rename_unique_f1_key_v2" +hint:WITH (EXCLUDE CONSTRAINTS ('alter_rename_unique_f1_key_v2')); # @@ -307,7 +336,8 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_extend_column ALTER COLUMN f1 TYPE VARCHAR(20); ! SELECT * FROM alter_extend_column; -contains:the type of column "f1" of source table public.alter_extend_column changed upstream +contains:incompatible schema change on public.alter_extend_column: the type of column "f1" changed upstream +hint:WITH (TEXT COLUMNS ("f1")) # @@ -319,4 +349,5 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_decimal ALTER COLUMN f1 TYPE DECIMAL(6,1); ! SELECT * FROM alter_decimal; -contains:the type of column "f1" of source table public.alter_decimal changed upstream +contains:incompatible schema change on public.alter_decimal: the type of column "f1" changed upstream +hint:WITH (TEXT COLUMNS ("f1")) diff --git a/test/pg-cdc/alter-table-after-source-2.td b/test/pg-cdc/alter-table-after-source-2.td index 52631de897086..904c3545eb65b 100644 --- a/test/pg-cdc/alter-table-after-source-2.td +++ b/test/pg-cdc/alter-table-after-source-2.td @@ -147,7 +147,8 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_table_rename RENAME TO alter_table_renamed; ! SELECT * FROM alter_table_rename; -contains:source table public.alter_table_rename with oid +contains:incompatible schema change on public.alter_table_rename: table was renamed or moved upstream (it is now public.alter_table_renamed with oid +hint:FROM SOURCE (REFERENCE public.alter_table_renamed) # # Alter table rename colum @@ -161,7 +162,8 @@ 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:changed position upstream +contains:incompatible schema change on public.alter_table_rename_column: column +hint:CREATE SCHEMA v2; # # Change column attnum @@ -175,7 +177,8 @@ 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:column "f2" of source table public.alter_table_change_attnum changed position upstream +contains:incompatible schema change on public.alter_table_change_attnum: column "f2" changed position upstream +hint:CREATE SCHEMA v2; > SELECT * from alter_table_supported; 1 1 @@ -201,7 +204,8 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_table_supported DROP COLUMN f2; ! SELECT * from alter_table_supported; -contains:column "f2" of source table public.alter_table_supported was dropped or renamed upstream +contains:incompatible schema change on public.alter_table_supported: column "f2" was dropped or renamed upstream +hint:WITH (EXCLUDE COLUMNS ("f2")) before the upstream drop # diff --git a/test/source-sink-errors/mzcompose.py b/test/source-sink-errors/mzcompose.py index c877b4ba3516f..6570b9db0dee3 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='column "f1" of source table .+source1 was dropped or renamed upstream', + expected_error='incompatible schema change on .+source1: column "f1" was dropped or renamed upstream', fixage=None, ), PgDisruption( From a143e44caa397d6913489549e34f26a62408d652 Mon Sep 17 00:00:00 2001 From: Peter Larsen Date: Tue, 8 Sep 2026 14:05:41 -0400 Subject: [PATCH 3/7] storage: name the schema change cases as error variants Co-Authored-By: Claude Fable 5.1 --- src/postgres-util/src/desc.rs | 353 ++++++++++++++++------------- src/storage/src/source/postgres.rs | 2 +- 2 files changed, 191 insertions(+), 164 deletions(-) diff --git a/src/postgres-util/src/desc.rs b/src/postgres-util/src/desc.rs index b9a9ba1eaa6c9..445200b3ca3b4 100644 --- a/src/postgres-util/src/desc.rs +++ b/src/postgres-util/src/desc.rs @@ -53,26 +53,141 @@ pub struct PostgresTableDesc { /// An upstream schema change that Materialize cannot follow. /// -/// `Display` renders the diagnosis. `hint` carries the recovery steps and is -/// surfaced separately: as the `HINT` of a SQL error and in the source status. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +/// `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}: {change}")] pub struct SchemaChangeError { - pub table: String, - pub change: String, - pub hint: String, + pub namespace: String, + pub name: String, + pub change: SchemaChange, +} + +/// The upstream change behind a [`SchemaChangeError`]. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)] +pub enum SchemaChange { + #[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("column {} was altered upstream", .column.quoted())] + ColumnAltered { column: String }, + #[error("{key} was dropped upstream")] + KeyDropped { key: KeyRef }, + #[error("{key} was dropped and recreated upstream")] + KeyRecreated { key: KeyRef }, + #[error("{key} was renamed upstream to {}", .new_name.quoted())] + KeyRenamed { key: KeyRef, new_name: String }, } -impl std::fmt::Display for SchemaChangeError { +/// 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, - "incompatible schema change on {}: {}", - self.table, self.change + "{kind} constraint {} ({})", + self.name.quoted(), + self.columns.join(", ") ) } } -impl std::error::Error for SchemaChangeError {} +impl SchemaChangeError { + /// The recovery steps for this change, including the statements to run. + pub fn hint(&self) -> String { + let reference = format!("{}.{}", self.namespace, self.name); + let recreate = |lead: &str, with_clause: Option| { + recreate_hint(lead, &self.name, &reference, with_clause.as_deref()) + }; + let exclude = |name: &str| format!("EXCLUDE CONSTRAINTS ('{}')", name.replace('\'', "''")); + match &self.change { + SchemaChange::TableRenamed { + namespace, name, .. + } => recreate_hint( + "To keep ingesting from the upstream table as it now exists", + name, + &format!("{namespace}.{name}"), + None, + ), + SchemaChange::ColumnDropped { column } => format!( + "{}\nTo make a planned column drop a non-event, create the table with \ + WITH (EXCLUDE COLUMNS ({})) before the upstream drop.", + recreate("To keep ingesting without this column", None), + column.quoted(), + ), + SchemaChange::ColumnMoved { .. } | SchemaChange::ColumnAltered { .. } => { + recreate("To keep ingesting", None) + } + SchemaChange::ColumnTypeChanged { column } => recreate( + "To ingest the column as text regardless of its upstream type", + Some(format!("TEXT COLUMNS ({})", column.quoted())), + ), + SchemaChange::NotNullDropped { .. } => recreate( + "To keep ingesting without this constraint", + Some("EXCLUDE ALL CONSTRAINTS".into()), + ), + SchemaChange::KeyDropped { key } => format!( + "{}\nTo make a planned constraint drop a non-event, create the table with \ + WITH ({}) before the upstream drop.", + recreate("To keep ingesting without this constraint", None), + exclude(&key.name), + ), + SchemaChange::KeyRecreated { key } => recreate( + "To keep ingesting without this constraint", + Some(exclude(&key.name)), + ), + SchemaChange::KeyRenamed { new_name, .. } => recreate( + "To keep ingesting without this constraint", + Some(exclude(new_name)), + ), + } + } +} + +fn recreate_hint( + lead: &str, + table_name: &str, + reference: &str, + with_clause: Option<&str>, +) -> String { + let mut hint = format!( + "{lead}, recreate the table in a new versioned schema, then swap your views to the \ + new table:\n CREATE SCHEMA v2;\n CREATE TABLE v2.{table_name}\n \ + FROM SOURCE (REFERENCE {reference})" + ); + if let Some(with_clause) = with_clause { + hint.push_str(&format!("\n WITH ({with_clause})")); + } + hint.push(';'); + hint +} impl PostgresTableDesc { /// Determines if two `PostgresTableDesc` are compatible with one another in @@ -98,179 +213,63 @@ impl PostgresTableDesc { } if self.oid != other.oid || self.namespace != other.namespace || self.name != other.name { - let reference = format!("{}.{}", other.namespace, other.name); - return Err(self.error( - format!( - "table was renamed or moved upstream (it is now {} with oid {})", - reference, other.oid - ), - recreate_hint( - "To keep ingesting from the upstream table as it now exists", - &other.name, - &reference, - None, - ), - )); + return Err(self.schema_change(SchemaChange::TableRenamed { + namespace: other.namespace.clone(), + name: other.name.clone(), + oid: other.oid, + })); } 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); - self.check_column( - column, - other_cols_by_name.get(&column.name).copied(), - allow_type_change, - )?; + let other_column = other_cols_by_name.get(&column.name).copied(); + if let Some(change) = column.diff(other_column, allow_type_change) { + return Err(self.schema_change(change)); + } } if let Some(key) = self.keys.difference(&other.keys).next() { - return Err(self.key_error(key, other)); + return Err(self.schema_change(self.key_change(key, other))); } Ok(()) } - fn check_column( - &self, - column: &PostgresColumnDesc, - other: Option<&PostgresColumnDesc>, - allow_type_change: bool, - ) -> Result<(), SchemaChangeError> { - let name = column.name.quoted(); - let Some(other) = other else { - return Err(self.error( - format!("column {name} was dropped or renamed upstream"), - format!( - "{}\nTo make a planned column drop a non-event, create the table with \ - WITH (EXCLUDE COLUMNS ({name})) before the upstream drop.", - self.recreate_hint("To keep ingesting without this column", None) - ), - )); - }; - if column.is_compatible(other, allow_type_change) { - return Ok(()); - } - if column.col_num != other.col_num { - return Err(self.error( - format!( - "column {name} changed position upstream (the column or table was likely \ - dropped and recreated)" - ), - self.recreate_hint("To keep ingesting", None), - )); - } - if !allow_type_change - && (column.type_oid != other.type_oid || column.type_mod != other.type_mod) - { - return Err(self.error( - format!("the type of column {name} changed upstream"), - self.recreate_hint( - "To ingest the column as text regardless of its upstream type", - Some(&format!("TEXT COLUMNS ({name})")), - ), - )); - } - if !column.nullable && other.nullable { - return Err(self.error( - format!("the NOT NULL constraint on column {name} was dropped upstream"), - self.recreate_hint( - "To keep ingesting without this constraint", - Some("EXCLUDE ALL CONSTRAINTS"), - ), - )); + fn schema_change(&self, change: SchemaChange) -> SchemaChangeError { + SchemaChangeError { + namespace: self.namespace.clone(), + name: self.name.clone(), + change, } - Err(self.error( - format!("column {name} was altered upstream"), - self.recreate_hint("To keep ingesting", None), - )) } - fn key_error(&self, key: &PostgresKeyDesc, other: &PostgresTableDesc) -> SchemaChangeError { - let kind = if key.is_primary { - "PRIMARY KEY" - } else { - "UNIQUE" + 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 cols = 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::>() - .join(", "); - let constraint = format!("{kind} constraint {} ({cols})", key.name.quoted()); - let exclude = |name: &str| format!("EXCLUDE CONSTRAINTS ('{}')", name.replace('\'', "''")); - if let Some(renamed) = other.keys.iter().find(|k| k.oid == key.oid) { - return self.error( - format!( - "{constraint} was renamed upstream to {}", - renamed.name.quoted() - ), - self.recreate_hint( - "To keep ingesting without this constraint", - Some(&exclude(&renamed.name)), - ), - ); - } - if other.keys.iter().any(|k| k.name == key.name) { - return self.error( - format!("{constraint} was dropped and recreated upstream"), - self.recreate_hint( - "To keep ingesting without this constraint", - Some(&exclude(&key.name)), - ), - ); - } - self.error( - format!("{constraint} was dropped upstream"), - format!( - "{}\nTo make a planned constraint drop a non-event, create the table with \ - WITH ({}) before the upstream drop.", - self.recreate_hint("To keep ingesting without this constraint", None), - exclude(&key.name), - ), - ) - } - - fn error(&self, change: String, hint: String) -> SchemaChangeError { - SchemaChangeError { - table: format!("{}.{}", self.namespace, self.name), - change, - hint, + SchemaChange::KeyRenamed { + key: key_ref, + new_name: renamed.name.clone(), + } + } else if other.keys.iter().any(|k| k.name == key.name) { + SchemaChange::KeyRecreated { key: key_ref } + } else { + SchemaChange::KeyDropped { key: key_ref } } } - - fn recreate_hint(&self, lead: &str, with_clause: Option<&str>) -> String { - recreate_hint( - lead, - &self.name, - &format!("{}.{}", self.namespace, self.name), - with_clause, - ) - } -} - -fn recreate_hint( - lead: &str, - table_name: &str, - reference: &str, - with_clause: Option<&str>, -) -> String { - let mut hint = format!( - "{lead}, recreate the table in a new versioned schema, then swap your views to the \ - new table:\n CREATE SCHEMA v2;\n CREATE TABLE v2.{table_name}\n \ - FROM SOURCE (REFERENCE {reference})" - ); - if let Some(with_clause) = with_clause { - hint.push_str(&format!("\n WITH ({with_clause})")); - } - hint.push(';'); - hint } impl RustType for PostgresTableDesc { @@ -349,6 +348,34 @@ impl PostgresColumnDesc { } } +impl PostgresColumnDesc { + fn diff( + &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.is_compatible(other, allow_type_change) { + return None; + } + 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 }); + } + if !self.nullable && other.nullable { + return Some(SchemaChange::NotNullDropped { column }); + } + Some(SchemaChange::ColumnAltered { column }) + } +} + impl RustType for PostgresColumnDesc { fn into_proto(&self) -> ProtoPostgresColumnDesc { ProtoPostgresColumnDesc { diff --git a/src/storage/src/source/postgres.rs b/src/storage/src/source/postgres.rs index 320bfec0e949d..26400320291e4 100644 --- a/src/storage/src/source/postgres.rs +++ b/src/storage/src/source/postgres.rs @@ -375,7 +375,7 @@ pub enum DefiniteError { impl DefiniteError { fn hint(&self) -> Option { match self { - DefiniteError::IncompatibleSchema(err) => Some(err.hint.clone()), + DefiniteError::IncompatibleSchema(err) => Some(err.hint()), _ => None, } } From acda311c54a0f902d74518f091d61087e12422e2 Mon Sep 17 00:00:00 2001 From: Peter Larsen Date: Tue, 8 Sep 2026 14:40:58 -0400 Subject: [PATCH 4/7] storage: scope schema-change hints to dropped constraints Move SchemaChangeError and its variants into their own module, include the table oid in the diagnosis, and only attach recovery hints to the cases the EXCLUDE CONSTRAINTS work is about: a dropped, renamed, or recreated PRIMARY KEY/UNIQUE constraint and a dropped NOT NULL constraint. Column drops, type changes, position changes, and table renames keep their specific message but carry no hint. Testdrive asserts the exact hint text for a dropped UNIQUE constraint via the table's status details. Co-Authored-By: Claude Fable 5.1 --- .../patterns/upstream-schema-changes.md | 7 +- src/postgres-util/src/desc.rs | 155 +----------------- src/postgres-util/src/lib.rs | 2 + src/postgres-util/src/schema_change.rs | 114 +++++++++++++ src/storage/src/source/postgres.rs | 5 +- .../alter-table-after-source.td | 37 ++--- test/pg-cdc/alter-table-after-source-1.td | 23 ++- test/pg-cdc/alter-table-after-source-2.td | 12 +- test/source-sink-errors/mzcompose.py | 2 +- 9 files changed, 161 insertions(+), 196 deletions(-) create mode 100644 src/postgres-util/src/schema_change.rs 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 31f124dda486e..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,12 +455,7 @@ notice, the ingesting table stalls permanently: ``` ERROR: Source error: source must be dropped and recreated due to failure: - incompatible schema change on public.orders: column "priority" was dropped or renamed upstream -HINT: To keep ingesting without this column, recreate the table in a new versioned schema, then swap your views to the new table: - CREATE SCHEMA v2; - CREATE TABLE v2.orders - FROM SOURCE (REFERENCE public.orders); - To make a planned column drop a non-event, create the table with WITH (EXCLUDE COLUMNS ("priority")) before the upstream drop. + 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/postgres-util/src/desc.rs b/src/postgres-util/src/desc.rs index 445200b3ca3b4..73f2cacec3212 100644 --- a/src/postgres-util/src/desc.rs +++ b/src/postgres-util/src/desc.rs @@ -11,13 +11,14 @@ use std::collections::{BTreeMap, BTreeSet}; -use mz_ore::str::StrExt; use mz_proto::{IntoRustIfSome, RustType, TryFromProtoError}; use proptest::prelude::any; use proptest_derive::Arbitrary; use serde::{Deserialize, Serialize}; use tokio_postgres::types::Oid; +use crate::schema_change::{KeyRef, SchemaChange, SchemaChangeError}; + include!(concat!(env!("OUT_DIR"), "/mz_postgres_util.desc.rs")); /// Describes a schema in a PostgreSQL database. @@ -51,144 +52,6 @@ pub struct PostgresTableDesc { pub keys: BTreeSet, } -/// 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}: {change}")] -pub struct SchemaChangeError { - pub namespace: String, - pub name: String, - pub change: SchemaChange, -} - -/// The upstream change behind a [`SchemaChangeError`]. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)] -pub enum SchemaChange { - #[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("column {} was altered upstream", .column.quoted())] - ColumnAltered { column: String }, - #[error("{key} was dropped upstream")] - KeyDropped { key: KeyRef }, - #[error("{key} was dropped and recreated upstream")] - KeyRecreated { key: KeyRef }, - #[error("{key} was renamed upstream to {}", .new_name.quoted())] - KeyRenamed { key: KeyRef, new_name: String }, -} - -/// 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 this change, including the statements to run. - pub fn hint(&self) -> String { - let reference = format!("{}.{}", self.namespace, self.name); - let recreate = |lead: &str, with_clause: Option| { - recreate_hint(lead, &self.name, &reference, with_clause.as_deref()) - }; - let exclude = |name: &str| format!("EXCLUDE CONSTRAINTS ('{}')", name.replace('\'', "''")); - match &self.change { - SchemaChange::TableRenamed { - namespace, name, .. - } => recreate_hint( - "To keep ingesting from the upstream table as it now exists", - name, - &format!("{namespace}.{name}"), - None, - ), - SchemaChange::ColumnDropped { column } => format!( - "{}\nTo make a planned column drop a non-event, create the table with \ - WITH (EXCLUDE COLUMNS ({})) before the upstream drop.", - recreate("To keep ingesting without this column", None), - column.quoted(), - ), - SchemaChange::ColumnMoved { .. } | SchemaChange::ColumnAltered { .. } => { - recreate("To keep ingesting", None) - } - SchemaChange::ColumnTypeChanged { column } => recreate( - "To ingest the column as text regardless of its upstream type", - Some(format!("TEXT COLUMNS ({})", column.quoted())), - ), - SchemaChange::NotNullDropped { .. } => recreate( - "To keep ingesting without this constraint", - Some("EXCLUDE ALL CONSTRAINTS".into()), - ), - SchemaChange::KeyDropped { key } => format!( - "{}\nTo make a planned constraint drop a non-event, create the table with \ - WITH ({}) before the upstream drop.", - recreate("To keep ingesting without this constraint", None), - exclude(&key.name), - ), - SchemaChange::KeyRecreated { key } => recreate( - "To keep ingesting without this constraint", - Some(exclude(&key.name)), - ), - SchemaChange::KeyRenamed { new_name, .. } => recreate( - "To keep ingesting without this constraint", - Some(exclude(new_name)), - ), - } - } -} - -fn recreate_hint( - lead: &str, - table_name: &str, - reference: &str, - with_clause: Option<&str>, -) -> String { - let mut hint = format!( - "{lead}, recreate the table in a new versioned schema, then swap your views to the \ - new table:\n CREATE SCHEMA v2;\n CREATE TABLE v2.{table_name}\n \ - FROM SOURCE (REFERENCE {reference})" - ); - if let Some(with_clause) = with_clause { - hint.push_str(&format!("\n WITH ({with_clause})")); - } - hint.push(';'); - hint -} - impl PostgresTableDesc { /// Determines if two `PostgresTableDesc` are compatible with one another in /// a way that Materialize can handle. @@ -240,6 +103,7 @@ impl PostgresTableDesc { SchemaChangeError { namespace: self.namespace.clone(), name: self.name.clone(), + oid: self.oid, change, } } @@ -259,13 +123,12 @@ impl PostgresTableDesc { }) .collect(), }; - if let Some(renamed) = other.keys.iter().find(|k| k.oid == key.oid) { - SchemaChange::KeyRenamed { - key: key_ref, - new_name: renamed.name.clone(), - } - } else if other.keys.iter().any(|k| k.name == key.name) { - SchemaChange::KeyRecreated { key: key_ref } + 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 { SchemaChange::KeyDropped { key: key_ref } } 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..63829bf8e851e --- /dev/null +++ b/src/postgres-util/src/schema_change.rs @@ -0,0 +1,114 @@ +// 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 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 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("column {} was altered upstream", .column.quoted())] + ColumnAltered { 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 {}.{})", + self.name, self.namespace, 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), + key.name.replace('\'', "''"), + )), + SchemaChange::NotNullDropped { .. } => Some(recreate(Some("EXCLUDE ALL CONSTRAINTS"))), + SchemaChange::TableRenamed { .. } + | SchemaChange::ColumnDropped { .. } + | SchemaChange::ColumnMoved { .. } + | SchemaChange::ColumnTypeChanged { .. } + | SchemaChange::ColumnAltered { .. } => None, + } + } +} diff --git a/src/storage/src/source/postgres.rs b/src/storage/src/source/postgres.rs index 26400320291e4..0398794def390 100644 --- a/src/storage/src/source/postgres.rs +++ b/src/storage/src/source/postgres.rs @@ -89,7 +89,8 @@ use itertools::Itertools as _; use mz_expr::EvalError; use mz_ore::cast::CastFrom; use mz_ore::error::ErrorExt; -use mz_postgres_util::desc::{PostgresTableDesc, SchemaChangeError}; +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}; @@ -375,7 +376,7 @@ pub enum DefiniteError { impl DefiniteError { fn hint(&self) -> Option { match self { - DefiniteError::IncompatibleSchema(err) => Some(err.hint()), + DefiniteError::IncompatibleSchema(err) => err.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 d8a34ca6fb979..4c885f43da431 100644 --- a/test/pg-cdc-old-syntax/alter-table-after-source.td +++ b/test/pg-cdc-old-syntax/alter-table-after-source.td @@ -143,8 +143,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE remove_column DROP COLUMN f2; ! SELECT * from remove_column; -contains:incompatible schema change on public.remove_column: column "f2" was dropped or renamed upstream -hint:WITH (EXCLUDE COLUMNS ("f2")) before the upstream drop +regex:incompatible schema change on public\.remove_column \(oid \d+\): column "f2" was dropped or renamed upstream # @@ -157,8 +156,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_column ALTER COLUMN f2 TYPE CHAR(2); ! SELECT * from alter_column; -contains:incompatible schema change on public.alter_column: the type of column "f2" changed upstream -hint:WITH (TEXT COLUMNS ("f2")) +regex:incompatible schema change on public\.alter_column \(oid \d+\): the type of column "f2" changed upstream # @@ -171,7 +169,7 @@ $ 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:incompatible schema change on public.alter_drop_nullability: the NOT NULL constraint on column "f1" was dropped upstream +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 @@ -204,7 +202,7 @@ $ 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:incompatible schema change on public.alter_drop_pk: PRIMARY KEY constraint "alter_drop_pk_pkey" (f1) was dropped upstream +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 @@ -234,7 +232,7 @@ 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:incompatible schema change on public.alter_cycle_pk: PRIMARY KEY constraint "alter_cycle_pk_pkey" (f1) was dropped +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') @@ -265,9 +263,14 @@ $ 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:incompatible schema change on public.alter_drop_unique: UNIQUE constraint "alter_drop_unique_f1_key" (f1) was dropped upstream +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 + # # Add unique @@ -294,8 +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:incompatible schema change on public.alter_extend_column: the type of column "f1" changed upstream -hint:WITH (TEXT COLUMNS ("f1")) +regex:incompatible schema change on public\.alter_extend_column \(oid \d+\): the type of column "f1" changed upstream # @@ -307,8 +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:incompatible schema change on public.alter_decimal: the type of column "f1" changed upstream -hint:WITH (TEXT COLUMNS ("f1")) +regex:incompatible schema change on public\.alter_decimal \(oid \d+\): the type of column "f1" changed upstream # @@ -321,8 +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:incompatible schema change on public.alter_table_rename: table was renamed or moved upstream (it is now public.alter_table_renamed with oid -hint:FROM SOURCE (REFERENCE public.alter_table_renamed) +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 @@ -336,8 +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:incompatible schema change on public.alter_table_rename_column: column -hint:CREATE SCHEMA v2; +regex:incompatible schema change on public\.alter_table_rename_column \(oid \d+\): column # @@ -352,8 +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:incompatible schema change on public.alter_table_change_attnum: column "f2" changed position upstream -hint:CREATE SCHEMA v2; +regex:incompatible schema change on public\.alter_table_change_attnum \(oid \d+\): column "f2" changed position upstream > SELECT * from alter_table_supported; 1 1 @@ -379,8 +377,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_table_supported DROP COLUMN f2; ! SELECT * from alter_table_supported; -contains:incompatible schema change on public.alter_table_supported: column "f2" was dropped or renamed upstream -hint:WITH (EXCLUDE COLUMNS ("f2")) before the upstream drop +regex:incompatible schema change on public\.alter_table_supported \(oid \d+\): column "f2" was dropped or renamed upstream # diff --git a/test/pg-cdc/alter-table-after-source-1.td b/test/pg-cdc/alter-table-after-source-1.td index e6d1df246825f..43b475657adaf 100644 --- a/test/pg-cdc/alter-table-after-source-1.td +++ b/test/pg-cdc/alter-table-after-source-1.td @@ -180,8 +180,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_column ALTER COLUMN f2 TYPE CHAR(2); ! SELECT * from alter_column; -contains:incompatible schema change on public.alter_column: the type of column "f2" changed upstream -hint:WITH (TEXT COLUMNS ("f2")) +regex:incompatible schema change on public\.alter_column \(oid \d+\): the type of column "f2" changed upstream # @@ -194,7 +193,7 @@ $ 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:incompatible schema change on public.alter_drop_nullability: the NOT NULL constraint on column "f1" was dropped upstream +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 @@ -227,7 +226,7 @@ $ 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:incompatible schema change on public.alter_drop_pk: PRIMARY KEY constraint "alter_drop_pk_pkey" (f1) was dropped upstream +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 @@ -257,7 +256,7 @@ 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:incompatible schema change on public.alter_cycle_pk: PRIMARY KEY constraint "alter_cycle_pk_pkey" (f1) was dropped +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') @@ -288,11 +287,11 @@ $ 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:incompatible schema change on public.alter_drop_unique: UNIQUE constraint "alter_drop_unique_f1_key" (f1) was dropped upstream +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' LIKE '%EXCLUDE CONSTRAINTS (''alter_drop_unique_f1_key'')%' +> 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 @@ -307,8 +306,8 @@ $ 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; -contains:incompatible schema change on public.alter_rename_unique: UNIQUE constraint "alter_rename_unique_f1_key" (f1) was renamed upstream to "alter_rename_unique_f1_key_v2" -hint:WITH (EXCLUDE CONSTRAINTS ('alter_rename_unique_f1_key_v2')); +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 # @@ -336,8 +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:incompatible schema change on public.alter_extend_column: the type of column "f1" changed upstream -hint:WITH (TEXT COLUMNS ("f1")) +regex:incompatible schema change on public\.alter_extend_column \(oid \d+\): the type of column "f1" changed upstream # @@ -349,5 +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:incompatible schema change on public.alter_decimal: the type of column "f1" changed upstream -hint:WITH (TEXT COLUMNS ("f1")) +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 904c3545eb65b..1f7526f6d3ca5 100644 --- a/test/pg-cdc/alter-table-after-source-2.td +++ b/test/pg-cdc/alter-table-after-source-2.td @@ -147,8 +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:incompatible schema change on public.alter_table_rename: table was renamed or moved upstream (it is now public.alter_table_renamed with oid -hint:FROM SOURCE (REFERENCE public.alter_table_renamed) +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 @@ -162,8 +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:incompatible schema change on public.alter_table_rename_column: column -hint:CREATE SCHEMA v2; +regex:incompatible schema change on public\.alter_table_rename_column \(oid \d+\): column # # Change column attnum @@ -177,8 +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:incompatible schema change on public.alter_table_change_attnum: column "f2" changed position upstream -hint:CREATE SCHEMA v2; +regex:incompatible schema change on public\.alter_table_change_attnum \(oid \d+\): column "f2" changed position upstream > SELECT * from alter_table_supported; 1 1 @@ -204,8 +201,7 @@ $ postgres-execute connection=postgres://postgres:postgres@postgres ALTER TABLE alter_table_supported DROP COLUMN f2; ! SELECT * from alter_table_supported; -contains:incompatible schema change on public.alter_table_supported: column "f2" was dropped or renamed upstream -hint:WITH (EXCLUDE COLUMNS ("f2")) before the upstream drop +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 6570b9db0dee3..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='incompatible schema change on .+source1: column "f1" was dropped or renamed upstream', + expected_error=r'incompatible schema change on .+source1 \(oid \d+\): column "f1" was dropped or renamed upstream', fixage=None, ), PgDisruption( From 9d9769543acc89e6d5e1ecca024bf8c6fcd8513a Mon Sep 17 00:00:00 2001 From: Peter Larsen Date: Wed, 9 Sep 2026 10:46:25 -0400 Subject: [PATCH 5/7] claude-review Co-Authored-By: Claude Fable 5.1 --- Cargo.lock | 1 + src/postgres-util/Cargo.toml | 3 +- src/postgres-util/src/desc.rs | 34 +++++-------------- src/postgres-util/src/schema_change.rs | 14 ++++---- .../alter-table-after-source.td | 2 +- test/pg-cdc/alter-table-after-source-1.td | 2 +- 6 files changed, 20 insertions(+), 36 deletions(-) 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/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 73f2cacec3212..af6dec229012b 100644 --- a/src/postgres-util/src/desc.rs +++ b/src/postgres-util/src/desc.rs @@ -59,8 +59,7 @@ 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::is_compatible`. /// - `self`'s keys are all present in `other` /// /// On incompatibility, the error describes the first mismatch found and @@ -87,7 +86,7 @@ impl PostgresTableDesc { 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.diff(other_column, allow_type_change) { + if let Some(change) = column.is_compatible(other_column, allow_type_change) { return Err(self.schema_change(change)); } } @@ -194,25 +193,7 @@ pub struct PostgresColumnDesc { impl PostgresColumnDesc { /// Determines if data a relation with a structure of `other` can be treated /// the same as `self`. - /// - /// 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) - } -} - -impl PostgresColumnDesc { - fn diff( + fn is_compatible( &self, other: Option<&PostgresColumnDesc>, allow_type_change: bool, @@ -221,9 +202,6 @@ impl PostgresColumnDesc { let Some(other) = other else { return Some(SchemaChange::ColumnDropped { column }); }; - if self.is_compatible(other, allow_type_change) { - return None; - } if self.col_num != other.col_num { return Some(SchemaChange::ColumnMoved { column }); } @@ -232,10 +210,14 @@ impl PostgresColumnDesc { { 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 }); } - Some(SchemaChange::ColumnAltered { column }) + None } } diff --git a/src/postgres-util/src/schema_change.rs b/src/postgres-util/src/schema_change.rs index 63829bf8e851e..874864177d2e7 100644 --- a/src/postgres-util/src/schema_change.rs +++ b/src/postgres-util/src/schema_change.rs @@ -10,6 +10,7 @@ //! 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. @@ -47,8 +48,6 @@ pub enum SchemaChange { ColumnTypeChanged { column: String }, #[error("the NOT NULL constraint on column {} was dropped upstream", .column.quoted())] NotNullDropped { column: String }, - #[error("column {} was altered upstream", .column.quoted())] - ColumnAltered { column: String }, #[error("{key} was dropped upstream")] KeyDropped { key: KeyRef }, #[error("{key} was renamed or recreated upstream")] @@ -88,7 +87,9 @@ impl SchemaChangeError { "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 {}.{})", - self.name, self.namespace, self.name + 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})")); @@ -99,16 +100,15 @@ impl SchemaChangeError { 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.", + WITH (EXCLUDE CONSTRAINTS ({})) before the upstream drop.", recreate(None), - key.name.replace('\'', "''"), + escape::escape_literal(&key.name), )), SchemaChange::NotNullDropped { .. } => Some(recreate(Some("EXCLUDE ALL CONSTRAINTS"))), SchemaChange::TableRenamed { .. } | SchemaChange::ColumnDropped { .. } | SchemaChange::ColumnMoved { .. } - | SchemaChange::ColumnTypeChanged { .. } - | SchemaChange::ColumnAltered { .. } => None, + | SchemaChange::ColumnTypeChanged { .. } => 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 4c885f43da431..78c57e08f57b7 100644 --- a/test/pg-cdc-old-syntax/alter-table-after-source.td +++ b/test/pg-cdc-old-syntax/alter-table-after-source.td @@ -267,7 +267,7 @@ regex:incompatible schema change on public\.alter_drop_unique \(oid \d+\): UNIQU 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.' +> 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 diff --git a/test/pg-cdc/alter-table-after-source-1.td b/test/pg-cdc/alter-table-after-source-1.td index 43b475657adaf..54a92a34d0a7a 100644 --- a/test/pg-cdc/alter-table-after-source-1.td +++ b/test/pg-cdc/alter-table-after-source-1.td @@ -291,7 +291,7 @@ regex:incompatible schema change on public\.alter_drop_unique \(oid \d+\): UNIQU 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.' +> 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 From 66b90929d77c3ac7b2c96f6cb68b9119c7b7b7e1 Mon Sep 17 00:00:00 2001 From: Peter Larsen Date: Wed, 9 Sep 2026 10:58:51 -0400 Subject: [PATCH 6/7] claude-review: keep column name comparison Co-Authored-By: Claude Fable 5.1 --- src/postgres-util/src/desc.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/postgres-util/src/desc.rs b/src/postgres-util/src/desc.rs index af6dec229012b..d8a9cc01e32c0 100644 --- a/src/postgres-util/src/desc.rs +++ b/src/postgres-util/src/desc.rs @@ -193,6 +193,10 @@ pub struct PostgresColumnDesc { impl PostgresColumnDesc { /// Determines if data a relation with a structure of `other` can be treated /// the same as `self`. + /// + /// 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: Option<&PostgresColumnDesc>, @@ -202,6 +206,9 @@ impl PostgresColumnDesc { 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 }); } From 433fff9dc6b5afceea79a60d55b96617d15a66d5 Mon Sep 17 00:00:00 2001 From: Peter Larsen Date: Wed, 9 Sep 2026 14:04:05 -0400 Subject: [PATCH 7/7] pr feedback Co-Authored-By: Claude Fable 5.1 --- src/postgres-util/src/desc.rs | 29 +++++++++++++++++++------- src/postgres-util/src/schema_change.rs | 5 ++++- 2 files changed, 25 insertions(+), 9 deletions(-) diff --git a/src/postgres-util/src/desc.rs b/src/postgres-util/src/desc.rs index d8a9cc01e32c0..b7894288015e7 100644 --- a/src/postgres-util/src/desc.rs +++ b/src/postgres-util/src/desc.rs @@ -16,6 +16,7 @@ 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}; @@ -59,7 +60,7 @@ 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 by `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 @@ -74,8 +75,18 @@ impl PostgresTableDesc { return Ok(()); } - if self.oid != other.oid || self.namespace != other.namespace || self.name != other.name { - return Err(self.schema_change(SchemaChange::TableRenamed { + 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 }) + ); + } + + 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, @@ -86,19 +97,21 @@ impl PostgresTableDesc { 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.is_compatible(other_column, allow_type_change) { - return Err(self.schema_change(change)); + 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.schema_change(self.key_change(key, other))); + return Err(self.build_schema_change_error(self.key_change(key, other))); } Ok(()) } - fn schema_change(&self, change: SchemaChange) -> SchemaChangeError { + fn build_schema_change_error(&self, change: SchemaChange) -> SchemaChangeError { SchemaChangeError { namespace: self.namespace.clone(), name: self.name.clone(), @@ -197,7 +210,7 @@ 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( + fn get_incompatible_schema_change( &self, other: Option<&PostgresColumnDesc>, allow_type_change: bool, diff --git a/src/postgres-util/src/schema_change.rs b/src/postgres-util/src/schema_change.rs index 874864177d2e7..5b9641c898872 100644 --- a/src/postgres-util/src/schema_change.rs +++ b/src/postgres-util/src/schema_change.rs @@ -30,6 +30,8 @@ pub struct SchemaChangeError { /// 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, @@ -105,7 +107,8 @@ impl SchemaChangeError { escape::escape_literal(&key.name), )), SchemaChange::NotNullDropped { .. } => Some(recreate(Some("EXCLUDE ALL CONSTRAINTS"))), - SchemaChange::TableRenamed { .. } + SchemaChange::TableDropped { .. } + | SchemaChange::TableRenamed { .. } | SchemaChange::ColumnDropped { .. } | SchemaChange::ColumnMoved { .. } | SchemaChange::ColumnTypeChanged { .. } => None,