diff --git a/misc/python/materialize/checks/all_checks/pg_cdc.py b/misc/python/materialize/checks/all_checks/pg_cdc.py index 079403848287f..6ffc857f3edd3 100644 --- a/misc/python/materialize/checks/all_checks/pg_cdc.py +++ b/misc/python/materialize/checks/all_checks/pg_cdc.py @@ -375,3 +375,100 @@ 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. + + 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: + 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 + # 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 IN ('running', 'stalled') FROM mz_internal.mz_source_statuses WHERE name = 'oid_retraction_table' AND type = 'table'; + true + """)) + + 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/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 9d86a903f4cdd..ec6df50cc4420 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: //! @@ -84,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, @@ -206,7 +203,14 @@ 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. + // 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)?)) } @@ -836,6 +840,65 @@ 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] + 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() { @@ -1136,13 +1199,41 @@ 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`. + // `parity_oid_full_range` covers the widened storage variant. assert_parity( "Oid", CastFunc::CastStringToOid, UnaryFunc::CastStringToOid(CastStringToOid), - &["42", "0", "2147483648", "4294967295", "-1", "bad", ""], + &["42", "0", "bad", ""], + ); + } + + #[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", + "", + ], ); } 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}