From 6995e9165df4c8593094e3b097ce9cdccc0528ec Mon Sep 17 00:00:00 2001 From: Dennis Felsing Date: Mon, 6 Jul 2026 12:47:01 +0000 Subject: [PATCH 1/3] storage: keep the PostgreSQL oid source cast on the legacy i32 range Follow-up to https://github.com/MaterializeInc/materialize/pull/37440 Sorry for only realizing so late that it's an issue after all! `parse_oid` was widened to the full `u32` range (#37440), but the same helper backs `CastStringToOid` in the PostgreSQL source, whose eval behavior must stay stable across releases. Replication re-casts the old tuple on delete, so a value ingested pre-upgrade as a `CastError` would be retracted post-upgrade as a value. That leaves the error stuck and adds a phantom negative row. Keep the storage cast on a frozen `parse_oid_legacy` that accepts only the `i32` range. The SQL, COPY, and pgwire paths keep the widened `parse_oid`. Add a unit snapshot pinning the frozen behavior and a platform-check reproducing the cross-upgrade retraction. Co-Authored-By: Claude Opus 4.8 --- .../materialize/checks/all_checks/pg_cdc.py | 88 +++++++++++++++++++ src/repr/src/strconv.rs | 35 ++++++++ src/storage-types/src/sources/casts.rs | 54 +++++++++--- 3 files changed, 165 insertions(+), 12 deletions(-) diff --git a/misc/python/materialize/checks/all_checks/pg_cdc.py b/misc/python/materialize/checks/all_checks/pg_cdc.py index 079403848287f..24d104bea2688 100644 --- a/misc/python/materialize/checks/all_checks/pg_cdc.py +++ b/misc/python/materialize/checks/all_checks/pg_cdc.py @@ -375,3 +375,91 @@ def validate(self) -> Testdrive: INSERT INTO postgres_mz_now_table VALUES (NOW(), 'B3'); DELETE FROM postgres_mz_now_table WHERE f2 LIKE '%4%'; """)) + + +@externally_idempotent(False) +class PgCdcOidRetraction(Check): + """Regression guard for the text->oid cast stability across upgrades. + + An upstream `oid` value above `i32::MAX` (here `4294967295`) is inserted + while the source may still run an older release, whose storage cast rejects + it as a per-row `CastError` persisted in the `err` collection. The row is + then deleted after the upgrade. Because Postgres replication re-casts the + old tuple on delete, the delete must reproduce the *same* error so the `+1` + error is cleanly retracted. If the storage cast were widened to accept the + full `u32` range, the delete would instead emit an `ok` value at diff `-1`, + leaving the error stuck (SELECT stays poisoned) and adding a phantom `-1`. + The final SELECT below fails in that case and passes once the storage cast + is stable across versions. + """ + + def initialize(self) -> Testdrive: + return Testdrive(dedent(""" + $ postgres-execute connection=postgres://postgres:postgres@postgres + CREATE USER oid_retraction_user WITH SUPERUSER PASSWORD 'postgres'; + ALTER USER oid_retraction_user WITH replication; + DROP PUBLICATION IF EXISTS oid_retraction_publication; + DROP TABLE IF EXISTS oid_retraction_table; + + CREATE TABLE oid_retraction_table (id INT PRIMARY KEY, o OID); + # REPLICA IDENTITY FULL so the delete carries the full old tuple, + # including the oid column that gets re-cast on retraction. + ALTER TABLE oid_retraction_table REPLICA IDENTITY FULL; + + # id=1 casts on every release; id=2 holds an oid above i32::MAX, + # which older releases reject as "invalid input syntax for type oid". + INSERT INTO oid_retraction_table VALUES (1, 42); + INSERT INTO oid_retraction_table VALUES (2, 4294967295); + ANALYZE oid_retraction_table; + + CREATE PUBLICATION oid_retraction_publication FOR TABLE oid_retraction_table; + + > CREATE SECRET oid_retraction_pass AS 'postgres'; + + > CREATE CONNECTION oid_retraction_conn FOR POSTGRES + HOST 'postgres', + DATABASE postgres, + USER oid_retraction_user, + PASSWORD SECRET oid_retraction_pass + + > CREATE SOURCE oid_retraction_source + FROM POSTGRES CONNECTION oid_retraction_conn + (PUBLICATION 'oid_retraction_publication'); + + > CREATE TABLE oid_retraction_table FROM SOURCE oid_retraction_source (REFERENCE oid_retraction_table); + + # The above-i32::MAX row errors per-row on releases with the + # i32-only cast, but that is a data error in the `err` collection, + # not a source failure. Confirm the source and its table are + # healthy before the upgrade, so the retraction the delete + # produces later has a live source to apply against. + > SELECT status FROM mz_internal.mz_source_statuses WHERE name = 'oid_retraction_source'; + running + + > SELECT status FROM mz_internal.mz_source_statuses WHERE name = 'oid_retraction_table' AND type = 'table'; + running + """)) + + def manipulate(self) -> list[Testdrive]: + return [ + Testdrive(dedent(s)) + for s in [ + """ + $ postgres-execute connection=postgres://postgres:postgres@postgres + INSERT INTO oid_retraction_table VALUES (3, 7); + """, + # Runs on the upgraded binary under the upgrade scenarios. The + # delete re-casts the old tuple of the above-i32::MAX row. + """ + $ postgres-execute connection=postgres://postgres:postgres@postgres + DELETE FROM oid_retraction_table WHERE id = 2; + """, + ] + ] + + def validate(self) -> Testdrive: + return Testdrive(dedent(""" + > SELECT id, o FROM oid_retraction_table ORDER BY id; + 1 42 + 3 7 + """)) diff --git a/src/repr/src/strconv.rs b/src/repr/src/strconv.rs index 88438be2be41c..ac4ac8fa3d194 100644 --- a/src/repr/src/strconv.rs +++ b/src/repr/src/strconv.rs @@ -254,6 +254,26 @@ pub fn parse_oid(s: &str) -> Result { Ok(u32::reinterpret_cast(oid)) } +/// Parses an OID from `s`, accepting only the `i32` range. +/// +/// This is the historical [`parse_oid`] behavior: values are parsed as `i32` +/// and reinterpreted as `u32`, so text in `2147483648..=4294967295` is +/// rejected even though it denotes a valid OID. +/// +/// NOTE: This exists solely to keep the persisted PostgreSQL source cast +/// `CastStringToOid` evaluation-stable across releases (see the stability +/// contract in `mz_storage_types::sources::casts`). PostgreSQL replication +/// re-casts the old tuple on delete, so widening this cast would let a value +/// ingested pre-upgrade as an error be retracted post-upgrade as a value, +/// leaving the error stuck. Use [`parse_oid`] everywhere else. +pub fn parse_oid_legacy(s: &str) -> Result { + let oid: i32 = s + .trim() + .parse() + .map_err(|e| ParseError::invalid_input_syntax("oid", s).with_details(e))?; + Ok(u32::reinterpret_cast(oid)) +} + fn parse_float(type_name: &'static str, s: &str) -> Result where Fl: NumFloat + FromStr, @@ -2225,4 +2245,19 @@ mod tests { assert!(parse_oid("-2147483649").is_err()); assert!(parse_oid("nope").is_err()); } + + #[mz_ore::test] + fn test_parse_oid_legacy() { + // Only the i32 range is accepted, reinterpreting negatives as u32. + assert_eq!(parse_oid_legacy("0").unwrap(), 0); + assert_eq!(parse_oid_legacy("2147483647").unwrap(), 2147483647); + assert_eq!(parse_oid_legacy("-1").unwrap(), 4294967295); + assert_eq!(parse_oid_legacy("-2147483648").unwrap(), 2147483648); + + // The frozen behavior rejects the u32-only range that `parse_oid` + // accepts. This divergence must not change (storage stability). + assert!(parse_oid_legacy("2147483648").is_err()); + assert!(parse_oid_legacy("4294967295").is_err()); + assert!(parse_oid_legacy("nope").is_err()); + } } diff --git a/src/storage-types/src/sources/casts.rs b/src/storage-types/src/sources/casts.rs index 9d86a903f4cdd..21f5559ec3850 100644 --- a/src/storage-types/src/sources/casts.rs +++ b/src/storage-types/src/sources/casts.rs @@ -20,14 +20,6 @@ //! to error variants, error messages, or output types are **breaking changes** //! for storage and require a migration. //! -//! NOTE: `CastStringToOid` is a deliberate exception. `strconv::parse_oid` was -//! widened to accept the full `u32` range, so text in `2147483648..=4294967295` -//! now casts to a value instead of erroring. A source that ingested such a value -//! as an error under an older release and then retracts it after upgrading will -//! leave that error unretracted. This was accepted without a migration because -//! ingesting `oid` columns is rare, and an errored source is recreated (which -//! resnapshots with the corrected cast) rather than kept around across upgrades. -//! //! The eval implementations delegate to `mz_repr::strconv::parse_*` functions, //! which in turn depend on these external crates: //! @@ -206,7 +198,12 @@ impl CastFunc { let f: f64 = strconv::parse_float64(a).map_err(parse_err)?; Ok(Datum::Float64(f.into())) } - CastFunc::CastStringToOid => Ok(Datum::UInt32(Oid(strconv::parse_oid(a)?).0)), + // NOTE: Uses the frozen `parse_oid_legacy` (i32 range only), not + // `parse_oid`, to satisfy the stability contract above. `parse_oid` + // was widened to the full `u32` range for SQL casts, but replication + // re-casts the old tuple on delete, so changing this persisted cast + // would let a row ingested as an error later be retracted as a value. + CastFunc::CastStringToOid => Ok(Datum::UInt32(Oid(strconv::parse_oid_legacy(a)?).0)), CastFunc::CastStringToUint16 => { Ok(Datum::UInt16(strconv::parse_uint16(a).map_err(parse_err)?)) } @@ -836,6 +833,37 @@ mod tests { ); } + #[mz_ore::test] + fn error_oid_frozen_i32_range() { + // The storage oid cast must stay stable across releases: text above + // i32::MAX keeps erroring here even though the SQL cast accepts the + // full u32 range. Widening it would break retraction symmetry for + // existing PG sources (a pre-upgrade CastError could later be + // retracted as a value, leaving the error stuck). + let arena = RowArena::new(); + let expr = cast_col0(CastFunc::CastStringToOid); + assert_eq!( + expr.eval(&[Datum::String("2147483647")], &arena).unwrap(), + Datum::UInt32(2147483647), + ); + assert_eq!( + eval_cast_err(CastFunc::CastStringToOid, "2147483648"), + parse_err_with_details( + "oid", + "2147483648", + "number too large to fit in target type" + ), + ); + assert_eq!( + eval_cast_err(CastFunc::CastStringToOid, "4294967295"), + parse_err_with_details( + "oid", + "4294967295", + "number too large to fit in target type" + ), + ); + } + #[mz_ore::test] #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux` fn error_numeric() { @@ -1136,13 +1164,15 @@ mod tests { #[mz_ore::test] fn parity_oid() { use mz_expr::func::CastStringToOid; - // Cover the full u32 range that both casts accept, including values - // above i32::MAX and the negative reinterpretation. + // Inputs stay within the i32 range on purpose. The storage cast is + // frozen on `parse_oid_legacy` while the SQL cast (`mz_expr`) uses + // the widened `parse_oid`, so the two intentionally diverge for text + // in `2147483648..=4294967295`. See `error_oid_frozen_i32_range`. assert_parity( "Oid", CastFunc::CastStringToOid, UnaryFunc::CastStringToOid(CastStringToOid), - &["42", "0", "2147483648", "4294967295", "-1", "bad", ""], + &["42", "0", "bad", ""], ); } From 52acfa216001990a613fb53b5fd14ae9064c37ee Mon Sep 17 00:00:00 2001 From: Dennis Felsing Date: Mon, 6 Jul 2026 17:50:42 +0000 Subject: [PATCH 2/3] storage: version the PostgreSQL oid source cast per export Freezing CastStringToOid on the legacy i32 range fixed retraction symmetry for existing sources, but also made fresh PostgreSQL sources reject valid oid values above i32::MAX, undoing the source-side part of the widening from #37440 for all future sources. Version the cast per export instead. Newly purified exports persist cast_oid_full_range in their statement DETAILS and plan the new CastStringToOidFullRange, which accepts the full u32 range via the widened parse_oid. Exports whose details predate the flag decode as false and keep the legacy i32-range cast, so their persisted rows keep retracting cleanly. The cast choice derives from the persisted details rather than the binary version, which keeps replanning stable across restarts and upgrades. Add unit coverage for the full-range variant (error snapshot and parity with the SQL cast) and a pg-cdc testdrive file exercising snapshot, replication, and retraction of above-i32::MAX oids in scalar and array columns. Co-Authored-By: Claude Fable 5 --- .../materialize/checks/all_checks/pg_cdc.py | 7 ++ src/sql/src/plan/statement/ddl.rs | 44 +++++++------ src/sql/src/pure/postgres.rs | 32 ++++++--- src/storage-types/src/sources.rs | 13 +++- src/storage-types/src/sources/casts.rs | 61 +++++++++++++++++ src/storage-types/src/sources/postgres.proto | 4 ++ test/pg-cdc/oid.td | 65 +++++++++++++++++++ 7 files changed, 197 insertions(+), 29 deletions(-) create mode 100644 test/pg-cdc/oid.td diff --git a/misc/python/materialize/checks/all_checks/pg_cdc.py b/misc/python/materialize/checks/all_checks/pg_cdc.py index 24d104bea2688..7fb073a082fe8 100644 --- a/misc/python/materialize/checks/all_checks/pg_cdc.py +++ b/misc/python/materialize/checks/all_checks/pg_cdc.py @@ -391,6 +391,13 @@ class PgCdcOidRetraction(Check): leaving the error stuck (SELECT stays poisoned) and adding a phantom `-1`. The final SELECT below fails in that case and passes once the storage cast is stable across versions. + + Exports created on releases with `cast_oid_full_range` in their statement + details use the full-range cast from the start, so the insert ingests as a + value and the delete retracts it symmetrically. In no-upgrade scenarios + this check therefore passes trivially. The asymmetry it guards against + only threatens exports whose details predate the flag, which must stay on + the legacy `i32`-range cast. """ def initialize(self) -> Testdrive: diff --git a/src/sql/src/plan/statement/ddl.rs b/src/sql/src/plan/statement/ddl.rs index 82e9c640e122f..d8d73a0578c7d 100644 --- a/src/sql/src/plan/statement/ddl.rs +++ b/src/sql/src/plan/statement/ddl.rs @@ -1650,16 +1650,18 @@ pub fn plan_create_subsource( let details = SourceExportStatementDetails::from_proto(details).map_err(|e| sql_err!("{}", e))?; let details = match details { - SourceExportStatementDetails::Postgres { table } => { - SourceExportDetails::Postgres(PostgresSourceExportDetails { - column_casts: crate::pure::postgres::generate_column_casts( - scx, - &table, - &text_columns, - )?, - table, - }) - } + SourceExportStatementDetails::Postgres { + table, + cast_oid_full_range, + } => SourceExportDetails::Postgres(PostgresSourceExportDetails { + column_casts: crate::pure::postgres::generate_column_casts( + scx, + &table, + &text_columns, + cast_oid_full_range, + )?, + table, + }), SourceExportStatementDetails::MySql { table, initial_gtid_set, @@ -1809,16 +1811,18 @@ pub fn plan_create_table_from_source( } let details = match details { - SourceExportStatementDetails::Postgres { table } => { - SourceExportDetails::Postgres(PostgresSourceExportDetails { - column_casts: crate::pure::postgres::generate_column_casts( - scx, - &table, - &text_columns, - )?, - table, - }) - } + SourceExportStatementDetails::Postgres { + table, + cast_oid_full_range, + } => SourceExportDetails::Postgres(PostgresSourceExportDetails { + column_casts: crate::pure::postgres::generate_column_casts( + scx, + &table, + &text_columns, + cast_oid_full_range, + )?, + table, + }), SourceExportStatementDetails::MySql { table, initial_gtid_set, diff --git a/src/sql/src/pure/postgres.rs b/src/sql/src/pure/postgres.rs index 7843b3efc630d..a77d4b984d414 100644 --- a/src/sql/src/pure/postgres.rs +++ b/src/sql/src/pure/postgres.rs @@ -316,7 +316,14 @@ pub(super) fn generate_source_export_statement_values( constraints.push(constraint); } } - let details = SourceExportStatementDetails::Postgres { table }; + // Newly purified exports always take the full-range oid cast. The flag is + // persisted in the statement details so replanning keeps the choice stable + // for the lifetime of the export, while exports whose details predate the + // flag decode as `false` and stay on the legacy cast. + let details = SourceExportStatementDetails::Postgres { + table, + cast_oid_full_range: true, + }; let text_columns = text_columns.map(|mut columns| { columns.sort(); @@ -554,6 +561,7 @@ pub(crate) fn generate_column_casts( scx: &StatementContext, table: &PostgresTableDesc, text_columns: &Vec, + cast_oid_full_range: bool, ) -> Result, PlanError> { // Generate the cast expressions required to convert the text encoded columns into // the appropriate target types, creating a Vec. @@ -596,7 +604,7 @@ pub(crate) fn generate_column_casts( } }; - let cast_expr = match pg_type_to_cast_func(scx, &ty) { + let cast_expr = match pg_type_to_cast_func(scx, &ty, cast_oid_full_range) { Ok(None) => { // No cast needed (e.g. Text → String identity). StorageScalarExpr::Column(i) @@ -658,6 +666,7 @@ fn resolve_pg_type_to_scalar_type( fn pg_type_to_cast_func( scx: &StatementContext, ty: &mz_pgrepr::Type, + cast_oid_full_range: bool, ) -> Result, PlanError> { use mz_pgrepr::Type; @@ -685,7 +694,13 @@ fn pg_type_to_cast_func( _ => unreachable!("Numeric must resolve to Numeric"), } } - Type::Oid => CastFunc::CastStringToOid, + Type::Oid => { + if cast_oid_full_range { + CastFunc::CastStringToOidFullRange + } else { + CastFunc::CastStringToOid + } + } Type::Text => return Ok(None), Type::BpChar { .. } => { // Resolve through the catalog to get the repr CharLength type. @@ -740,7 +755,7 @@ fn pg_type_to_cast_func( Type::Json => CastFunc::CastStringToJsonb, Type::Array(elem) => { let return_ty = resolve_pg_type_to_scalar_type(scx, ty)?; - let elem_cast = build_element_cast_expr(scx, elem)?; + let elem_cast = build_element_cast_expr(scx, elem, cast_oid_full_range)?; CastFunc::CastStringToArray { return_ty, cast_expr: Box::new(elem_cast), @@ -748,7 +763,7 @@ fn pg_type_to_cast_func( } Type::List(elem) => { let return_ty = resolve_pg_type_to_scalar_type(scx, ty)?; - let elem_cast = build_element_cast_expr(scx, elem)?; + let elem_cast = build_element_cast_expr(scx, elem, cast_oid_full_range)?; CastFunc::CastStringToList { return_ty, cast_expr: Box::new(elem_cast), @@ -756,7 +771,7 @@ fn pg_type_to_cast_func( } Type::Map { value_type } => { let return_ty = resolve_pg_type_to_scalar_type(scx, ty)?; - let value_cast = build_element_cast_expr(scx, value_type)?; + let value_cast = build_element_cast_expr(scx, value_type, cast_oid_full_range)?; CastFunc::CastStringToMap { return_ty, cast_expr: Box::new(value_cast), @@ -764,7 +779,7 @@ fn pg_type_to_cast_func( } Type::Range { element_type } => { let return_ty = resolve_pg_type_to_scalar_type(scx, ty)?; - let elem_cast = build_element_cast_expr(scx, element_type)?; + let elem_cast = build_element_cast_expr(scx, element_type, cast_oid_full_range)?; CastFunc::CastStringToRange { return_ty, cast_expr: Box::new(elem_cast), @@ -796,8 +811,9 @@ fn pg_type_to_cast_func( fn build_element_cast_expr( scx: &StatementContext, elem_ty: &mz_pgrepr::Type, + cast_oid_full_range: bool, ) -> Result { - match pg_type_to_cast_func(scx, elem_ty)? { + match pg_type_to_cast_func(scx, elem_ty, cast_oid_full_range)? { None => Ok(StorageScalarExpr::Column(0)), Some(cast_func) => Ok(StorageScalarExpr::CallUnary( cast_func, diff --git a/src/storage-types/src/sources.rs b/src/storage-types/src/sources.rs index 10a09604259d0..0fcdda02d8783 100644 --- a/src/storage-types/src/sources.rs +++ b/src/storage-types/src/sources.rs @@ -909,6 +909,12 @@ impl crate::AlterCompatible for SourceExportDetails { pub enum SourceExportStatementDetails { Postgres { table: mz_postgres_util::desc::PostgresTableDesc, + /// Whether the text-to-oid cast for this export accepts the full `u32` + /// range. Exports created before the cast was widened decode as + /// `false` and must keep the legacy `i32`-range cast forever, because + /// replication re-casts old tuples on delete and the persisted rows + /// were ingested under the legacy semantics. + cast_oid_full_range: bool, }, MySql { table: mz_mysql_util::MySqlTableDesc, @@ -929,10 +935,14 @@ pub enum SourceExportStatementDetails { impl RustType for SourceExportStatementDetails { fn into_proto(&self) -> ProtoSourceExportStatementDetails { match self { - SourceExportStatementDetails::Postgres { table } => ProtoSourceExportStatementDetails { + SourceExportStatementDetails::Postgres { + table, + cast_oid_full_range, + } => ProtoSourceExportStatementDetails { kind: Some(proto_source_export_statement_details::Kind::Postgres( postgres::ProtoPostgresSourceExportStatementDetails { table: Some(table.into_proto()), + cast_oid_full_range: *cast_oid_full_range, }, )), }, @@ -986,6 +996,7 @@ impl RustType for SourceExportStatementDetail table: details .table .into_rust_if_some("ProtoPostgresSourceExportStatementDetails::table")?, + cast_oid_full_range: details.cast_oid_full_range, }, Some(Kind::Mysql(details)) => SourceExportStatementDetails::MySql { table: details diff --git a/src/storage-types/src/sources/casts.rs b/src/storage-types/src/sources/casts.rs index 21f5559ec3850..ec6df50cc4420 100644 --- a/src/storage-types/src/sources/casts.rs +++ b/src/storage-types/src/sources/casts.rs @@ -76,6 +76,11 @@ pub enum CastFunc { CastStringToFloat32, CastStringToFloat64, CastStringToOid, + /// The text-to-oid cast accepting the full `u32` range, used by source + /// exports whose statement details set `cast_oid_full_range`. + /// `CastStringToOid` stays on the legacy `i32` range for exports that + /// predate the widening. + CastStringToOidFullRange, CastStringToUint16, CastStringToUint32, CastStringToUint64, @@ -203,7 +208,9 @@ impl CastFunc { // was widened to the full `u32` range for SQL casts, but replication // re-casts the old tuple on delete, so changing this persisted cast // would let a row ingested as an error later be retracted as a value. + // Exports created after the widening use `CastStringToOidFullRange`. CastFunc::CastStringToOid => Ok(Datum::UInt32(Oid(strconv::parse_oid_legacy(a)?).0)), + CastFunc::CastStringToOidFullRange => Ok(Datum::UInt32(Oid(strconv::parse_oid(a)?).0)), CastFunc::CastStringToUint16 => { Ok(Datum::UInt16(strconv::parse_uint16(a).map_err(parse_err)?)) } @@ -864,6 +871,34 @@ mod tests { ); } + #[mz_ore::test] + fn oid_full_range() { + // The full-range variant accepts the whole `u32` range plus + // negative `i32` text reinterpreted as `u32`, matching the SQL + // cast. Only text outside both ranges errors. + let arena = RowArena::new(); + let expr = cast_col0(CastFunc::CastStringToOidFullRange); + for (input, expected) in [ + ("2147483647", 2147483647), + ("2147483648", 2147483648), + ("4294967295", 4294967295), + ("-1", 4294967295), + ] { + assert_eq!( + expr.eval(&[Datum::String(input)], &arena).unwrap(), + Datum::UInt32(expected), + ); + } + assert_eq!( + eval_cast_err(CastFunc::CastStringToOidFullRange, "4294967296"), + parse_err_with_details( + "oid", + "4294967296", + "number too large to fit in target type" + ), + ); + } + #[mz_ore::test] #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux` fn error_numeric() { @@ -1168,6 +1203,7 @@ mod tests { // frozen on `parse_oid_legacy` while the SQL cast (`mz_expr`) uses // the widened `parse_oid`, so the two intentionally diverge for text // in `2147483648..=4294967295`. See `error_oid_frozen_i32_range`. + // `parity_oid_full_range` covers the widened storage variant. assert_parity( "Oid", CastFunc::CastStringToOid, @@ -1176,6 +1212,31 @@ mod tests { ); } + #[mz_ore::test] + fn parity_oid_full_range() { + use mz_expr::func::CastStringToOid; + // The full-range variant matches the widened SQL cast on the whole + // `u32` range, unlike the frozen `CastFunc::CastStringToOid`. + assert_parity( + "OidFullRange", + CastFunc::CastStringToOidFullRange, + UnaryFunc::CastStringToOid(CastStringToOid), + &[ + "42", + "0", + "2147483647", + "2147483648", + "4294967295", + "-1", + "-2147483648", + "4294967296", + "-2147483649", + "bad", + "", + ], + ); + } + #[mz_ore::test] fn parity_uint16() { use mz_expr::func::CastStringToUint16; diff --git a/src/storage-types/src/sources/postgres.proto b/src/storage-types/src/sources/postgres.proto index 8d2ac8c9e0caf..718d2957434de 100644 --- a/src/storage-types/src/sources/postgres.proto +++ b/src/storage-types/src/sources/postgres.proto @@ -28,4 +28,8 @@ message ProtoPostgresSourcePublicationDetails { // compatible message ProtoPostgresSourceExportStatementDetails { mz_postgres_util.desc.ProtoPostgresTableDesc table = 1; + // Whether the text-to-oid cast for this export accepts the full u32 range. + // Absent (false) for exports created before the oid cast was widened, which + // must keep the legacy i32-range cast forever. + bool cast_oid_full_range = 2; } diff --git a/test/pg-cdc/oid.td b/test/pg-cdc/oid.td new file mode 100644 index 0000000000000..3de91282bfe40 --- /dev/null +++ b/test/pg-cdc/oid.td @@ -0,0 +1,65 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# 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. + +# +# Test that oid columns ingest the full u32 range, including values above +# i32::MAX, in both the snapshot and the replication path, and that deletes +# of such values retract cleanly. Newly created sources use the full-range +# text-to-oid cast (see cast_oid_full_range in the source export details). +# + +> CREATE SECRET pgpass AS 'postgres' +> CREATE CONNECTION pg TO POSTGRES ( + HOST postgres, + DATABASE postgres, + USER postgres, + PASSWORD SECRET pgpass + ) + +$ postgres-execute connection=postgres://postgres:postgres@postgres +ALTER USER postgres WITH replication; +DROP SCHEMA IF EXISTS public CASCADE; +DROP PUBLICATION IF EXISTS mz_source; + +CREATE SCHEMA public; + +CREATE TABLE t_oid (pk int PRIMARY KEY, o oid, oa oid[]); +ALTER TABLE t_oid REPLICA IDENTITY FULL; +INSERT INTO t_oid VALUES (1, 42, '{42}'); +INSERT INTO t_oid VALUES (2, 4294967295, '{4294967295,2147483648}'); +ANALYZE t_oid; + +CREATE PUBLICATION mz_source FOR TABLE t_oid; + +> BEGIN +> CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source'); +> CREATE TABLE t_oid FROM SOURCE mz_source (REFERENCE t_oid); +> COMMIT + +# Snapshot path: the above-i32::MAX values ingest as values, not errors. +> SELECT * FROM t_oid ORDER BY pk; +1 42 {42} +2 4294967295 {4294967295,2147483648} + +# Replication path. +$ postgres-execute connection=postgres://postgres:postgres@postgres +INSERT INTO t_oid VALUES (3, 2147483648, '{4294967294}'); + +> SELECT * FROM t_oid ORDER BY pk; +1 42 {42} +2 4294967295 {4294967295,2147483648} +3 2147483648 {4294967294} + +# Deletes re-cast the old tuple, so the above-i32::MAX rows must retract +# cleanly. +$ postgres-execute connection=postgres://postgres:postgres@postgres +DELETE FROM t_oid WHERE pk IN (2, 3); + +> SELECT * FROM t_oid ORDER BY pk; +1 42 {42} From da862b592380a91305c3e6ad550b962129ed6ef4 Mon Sep 17 00:00:00 2001 From: Dennis Felsing Date: Mon, 6 Jul 2026 20:29:14 +0000 Subject: [PATCH 3/3] checks: accept a stalled table export in PgCdcOidRetraction initialize The per-row cast error that the above-i32::MAX row produces on binaries with the legacy i32-range oid cast makes mz_source_statuses report the table export itself as stalled, while the source stays running and replication continues. The old assertion raced the running-to-stalled transition: sequential scenarios usually queried before the stall landed, but the parallel-mode shards and the earliest-release upgrade scenario saw stalled and failed in Nightly. Accept both states for the table export. Binaries whose exports carry cast_oid_full_range ingest the row and report running, legacy-cast binaries report stalled. The check's real assertion stays the validate SELECT that proves the delete cleanly retracts the error. Co-Authored-By: Claude Fable 5 --- .../python/materialize/checks/all_checks/pg_cdc.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/misc/python/materialize/checks/all_checks/pg_cdc.py b/misc/python/materialize/checks/all_checks/pg_cdc.py index 7fb073a082fe8..6ffc857f3edd3 100644 --- a/misc/python/materialize/checks/all_checks/pg_cdc.py +++ b/misc/python/materialize/checks/all_checks/pg_cdc.py @@ -436,15 +436,17 @@ def initialize(self) -> Testdrive: > CREATE TABLE oid_retraction_table FROM SOURCE oid_retraction_source (REFERENCE oid_retraction_table); # The above-i32::MAX row errors per-row on releases with the - # i32-only cast, but that is a data error in the `err` collection, - # not a source failure. Confirm the source and its table are - # healthy before the upgrade, so the retraction the delete - # produces later has a live source to apply against. + # legacy i32-only cast. That is a data error in the `err` + # collection: the source keeps running and replication + # continues, but the health operator reports the table export + # itself as stalled. Releases whose exports carry + # `cast_oid_full_range` ingest the row and report running, so + # accept both states for the table export. > SELECT status FROM mz_internal.mz_source_statuses WHERE name = 'oid_retraction_source'; running - > SELECT status FROM mz_internal.mz_source_statuses WHERE name = 'oid_retraction_table' AND type = 'table'; - running + > SELECT status IN ('running', 'stalled') FROM mz_internal.mz_source_statuses WHERE name = 'oid_retraction_table' AND type = 'table'; + true """)) def manipulate(self) -> list[Testdrive]: