storage: version the PostgreSQL oid source cast per export

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
+                """))

src/repr/src/strconv.rs

@@ -254,6 +254,26 @@ pub fn parse_oid(s: &str) -> Result<u32, ParseError> {
     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<u32, ParseError> {
+    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<Fl>(type_name: &'static str, s: &str) -> Result<Fl, ParseError>
 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());
+    }
 }

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,

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<Ident>,
+    cast_oid_full_range: bool,
 ) -> Result<Vec<(CastType, StorageScalarExpr)>, PlanError> {
     // Generate the cast expressions required to convert the text encoded columns into
     // the appropriate target types, creating a Vec<StorageScalarExpr>.
@@ -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<Option<CastFunc>, 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,31 +755,31 @@ 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),
             }
         }
         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),
             }
         }
         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),
             }
         }
         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<StorageScalarExpr, PlanError> {
-    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,

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<ProtoSourceExportStatementDetails> 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<ProtoSourceExportStatementDetails> 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