fix: oid is u32, not i32 (#873)

Possibly fixes #847.

We incorrectly implemented `oid` as `i32`, but it's actually `u32`, so
OIDs over 2.2B would be read as `0`. On very busy databases that
constantly create/destroy tables/columns, this would overflow. OIDs are
not guaranteed to be unique and it gets pretty bad when the oid search
space gets small (i.e. very large tables), but we still want to support
large OIDs since they are valid.

Author: levkk

pgdog-postgres-types/src/lib.rs

@@ -11,6 +11,7 @@ pub mod integer;
 pub mod interface;
 pub mod interval;
 pub mod numeric;
+pub mod oid;
 pub mod text;
 pub mod timestamp;
 pub mod timestamptz;
@@ -27,5 +28,6 @@ pub use format::Format;
 pub use interface::{FromDataType, ToDataRowColumn};
 pub use interval::Interval;
 pub use numeric::Numeric;
+pub use oid::Oid;
 pub use timestamp::Timestamp;
 pub use timestamptz::TimestampTz;

pgdog-postgres-types/src/oid.rs

@@ -0,0 +1,139 @@
+//! PostgreSQL `oid` data type.
+
+use super::*;
+use bytes::{Buf, Bytes};
+use std::fmt;
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct Oid(pub u32);
+
+impl Oid {
+    pub const fn new(value: u32) -> Self {
+        Self(value)
+    }
+
+    pub const fn get(self) -> u32 {
+        self.0
+    }
+}
+
+impl fmt::Display for Oid {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+impl From<u32> for Oid {
+    fn from(value: u32) -> Self {
+        Self(value)
+    }
+}
+
+impl From<Oid> for u32 {
+    fn from(value: Oid) -> Self {
+        value.0
+    }
+}
+
+impl FromDataType for Oid {
+    fn decode(bytes: &[u8], encoding: Format) -> Result<Self, Error> {
+        match encoding {
+            Format::Binary => {
+                let bytes: [u8; 4] = bytes.try_into()?;
+                Ok(Self(bytes.as_slice().get_u32()))
+            }
+
+            Format::Text => {
+                let s = String::decode(bytes, Format::Text)?;
+                Ok(Self(s.parse()?))
+            }
+        }
+    }
+
+    fn encode(&self, encoding: Format) -> Result<Bytes, Error> {
+        match encoding {
+            Format::Text => Ok(Bytes::copy_from_slice(self.0.to_string().as_bytes())),
+            Format::Binary => Ok(Bytes::copy_from_slice(&self.0.to_be_bytes())),
+        }
+    }
+}
+
+impl ToDataRowColumn for Oid {
+    fn to_data_row_column(&self) -> Data {
+        Bytes::copy_from_slice(self.0.to_string().as_bytes()).into()
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn test_oid_text_round_trip_small() {
+        let oid = Oid(16384); // typical user-table oid
+        let encoded = oid.encode(Format::Text).unwrap();
+        let decoded = Oid::decode(&encoded, Format::Text).unwrap();
+        assert_eq!(oid, decoded);
+    }
+
+    #[test]
+    fn test_oid_text_round_trip_above_i32_max() {
+        // Regression for issue #847: pg_class.oid can exceed i32::MAX in
+        // long-lived databases. Parsing as i32 used to silently produce 0.
+        let oid = Oid(2_500_000_000);
+        let encoded = oid.encode(Format::Text).unwrap();
+        assert_eq!(&encoded[..], b"2500000000");
+        let decoded = Oid::decode(&encoded, Format::Text).unwrap();
+        assert_eq!(oid, decoded);
+    }
+
+    #[test]
+    fn test_oid_text_round_trip_max() {
+        let oid = Oid(u32::MAX);
+        let encoded = oid.encode(Format::Text).unwrap();
+        let decoded = Oid::decode(&encoded, Format::Text).unwrap();
+        assert_eq!(oid, decoded);
+    }
+
+    #[test]
+    fn test_oid_binary_round_trip() {
+        for value in [0u32, 1, 16384, 2_147_483_647, 2_147_483_648, u32::MAX] {
+            let oid = Oid(value);
+            let encoded = oid.encode(Format::Binary).unwrap();
+            assert_eq!(encoded.len(), 4);
+            let decoded = Oid::decode(&encoded, Format::Binary).unwrap();
+            assert_eq!(oid, decoded);
+        }
+    }
+
+    #[test]
+    fn test_oid_decode_text_invalid() {
+        // Non-numeric text should error, not silently produce 0.
+        let result = Oid::decode(b"not-a-number", Format::Text);
+        assert!(matches!(result, Err(Error::NotInteger(_))));
+    }
+
+    #[test]
+    fn test_oid_decode_text_negative_rejected() {
+        // Negative values are not valid OIDs.
+        let result = Oid::decode(b"-1", Format::Text);
+        assert!(matches!(result, Err(Error::NotInteger(_))));
+    }
+
+    #[test]
+    fn test_oid_decode_binary_wrong_size() {
+        let result = Oid::decode(&[0u8; 3], Format::Binary);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_oid_display() {
+        assert_eq!(Oid(0).to_string(), "0");
+        assert_eq!(Oid(2_500_000_000).to_string(), "2500000000");
+    }
+
+    #[test]
+    fn test_oid_default_is_zero() {
+        assert_eq!(Oid::default(), Oid(0));
+    }
+}

pgdog/src/backend/replication/buffer.rs

@@ -1,5 +1,6 @@
 use fnv::FnvHashMap as HashMap;
 use fnv::FnvHashSet as HashSet;
+use pgdog_postgres_types::Oid;
 use std::collections::VecDeque;
 
 use crate::backend::ShardingSchema;
@@ -23,10 +24,10 @@ pub struct Buffer {
     replication_config: ReplicationConfig,
     begin: Option<XLogData>,
     message: Option<XLogData>,
-    relations: HashMap<i32, Relation>,
-    sent_relations: HashSet<i32>,
+    relations: HashMap<Oid, Relation>,
+    sent_relations: HashSet<Oid>,
     shard: Shard,
-    oid: Option<i32>,
+    oid: Option<Oid>,
     buffer: VecDeque<Message>,
     sharding_schema: ShardingSchema,
 }
@@ -163,7 +164,7 @@ impl Buffer {
         Ok(())
     }
 
-    fn sharding_key(&self, oid: i32) -> Result<(&str, Vec<&str>), Error> {
+    fn sharding_key(&self, oid: Oid) -> Result<(&str, Vec<&str>), Error> {
         let relation = self.relations.get(&oid).ok_or(Error::NoRelationMessage)?;
         let columns = relation.columns();
         let name = relation.name();

pgdog/src/backend/replication/logical/publisher/queries.rs

@@ -5,6 +5,8 @@
 //!
 use std::fmt::Display;
 
+use pgdog_postgres_types::Oid;
+
 use crate::{
     backend::Server,
     net::{DataRow, Format},
@@ -101,7 +103,7 @@ ON (c.relnamespace = n.oid) WHERE n.nspname = $1 AND c.relname = $2";
 /// Identifies the columns part of the replica identity for a table.
 #[derive(Debug, Clone, Eq, PartialEq, Hash)]
 pub struct ReplicaIdentity {
-    pub oid: i32,
+    pub oid: Oid,
     pub identity: String,
     pub kind: String,
 }
@@ -148,9 +150,11 @@ FROM
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct PublicationTableColumn {
+    /// Column number (`pg_attribute.attnum`). Despite the name, this is not an OID.
     pub oid: i32,
     pub name: String,
-    pub type_oid: i32,
+    /// Type OID (`pg_attribute.atttypid`).
+    pub type_oid: Oid,
     pub identity: bool,
 }
 
@@ -183,6 +187,74 @@ mod test {
 
     use super::*;
 
+    #[test]
+    fn test_replica_identity_decodes_oid_above_i32_max() {
+        // Regression for issue #847: pg_class.oid is unsigned 32-bit, and in
+        // long-lived databases it routinely exceeds i32::MAX. Decoding such an
+        // OID as i32 used to silently produce 0, which caused PgDog to send
+        // queries against OID 0 and trigger Postgres' "could not open relation
+        // with OID 0" error. With Oid (u32), this round-trips correctly.
+        let mut row = DataRow::new();
+        row.add(Oid(2_500_000_000))
+            .add("d".to_string())
+            .add("r".to_string());
+        let identity = ReplicaIdentity::from(row);
+        assert_eq!(identity.oid, Oid(2_500_000_000));
+        assert_eq!(identity.identity, "d");
+        assert_eq!(identity.kind, "r");
+    }
+
+    #[test]
+    fn test_replica_identity_substitutes_high_oid_into_columns_query() {
+        // The COLUMNS query embeds identity.oid as text via Display. Verify
+        // that a high OID renders as an unsigned decimal, not a negative i32.
+        let identity = ReplicaIdentity {
+            oid: Oid(2_500_000_000),
+            identity: "d".to_string(),
+            kind: "r".to_string(),
+        };
+        let rendered = COLUMNS
+            .replace("$1", &identity.oid.to_string())
+            .replace("$2", &identity.oid.to_string());
+        assert!(rendered.contains("pg_get_replica_identity_index(2500000000)"));
+        assert!(rendered.contains("a.attrelid = 2500000000"));
+        assert!(!rendered.contains("(0)"));
+        assert!(!rendered.contains("= 0 "));
+    }
+
+    #[test]
+    fn test_publication_table_column_decodes_high_type_oid() {
+        // pg_attribute.atttypid is also of type oid; user-defined types in
+        // long-lived databases can exceed i32::MAX.
+        let mut row = DataRow::new();
+        row.add(1_i64.to_string()) // attnum
+            .add("col".to_string())
+            .add(Oid(3_000_000_000)) // atttypid
+            .add("t".to_string()); // identity bool
+        let column = PublicationTableColumn::from(row);
+        assert_eq!(column.type_oid, Oid(3_000_000_000));
+        assert_eq!(column.name, "col");
+        assert!(column.identity);
+    }
+
+    #[tokio::test]
+    async fn test_oid_above_i32_max_round_trips_from_postgres() {
+        // Regression for issue #847: ask Postgres to emit a real `oid` value
+        // above i32::MAX on the wire and verify our text-format decode handles
+        // it. With the previous i32-based parser this returned 0; with Oid(u32)
+        // the value round-trips correctly.
+        let mut server = test_server().await;
+        let rows: Vec<DataRow> = server
+            .fetch_all("SELECT 2500000000::oid, 4294967295::oid")
+            .await
+            .unwrap();
+        assert_eq!(rows.len(), 1);
+        let high: Oid = rows[0].get(0, Format::Text).unwrap();
+        let max: Oid = rows[0].get(1, Format::Text).unwrap();
+        assert_eq!(high, Oid(2_500_000_000));
+        assert_eq!(max, Oid(u32::MAX));
+    }
+
     #[tokio::test]
     async fn test_logical_publisher_queries() {
         let mut server = test_server().await;

pgdog/src/backend/replication/logical/publisher/table.rs

@@ -306,7 +306,7 @@ mod test {
                 parent_name: "".to_string(),
             },
             identity: ReplicaIdentity {
-                oid: 1,
+                oid: pgdog_postgres_types::Oid(1),
                 identity: "".to_string(),
                 kind: "".to_string(),
             },
@@ -315,7 +315,7 @@ mod test {
                 .map(|(name, identity)| PublicationTableColumn {
                     oid: 1,
                     name: name.to_string(),
-                    type_oid: 23,
+                    type_oid: pgdog_postgres_types::Oid(23),
                     identity,
                 })
                 .collect(),

pgdog/src/backend/replication/logical/subscriber/stream.rs

@@ -16,6 +16,7 @@ use pg_query::{
     NodeEnum,
 };
 use pgdog_config::QueryParserEngine;
+use pgdog_postgres_types::Oid;
 use tracing::{debug, trace};
 
 use super::super::{publisher::Table, Error};
@@ -123,19 +124,19 @@ pub struct StreamSubscriber {
 
     // Relation markers sent by the publisher.
     // Happens once per connection.
-    relations: HashMap<i32, Relation>,
+    relations: HashMap<Oid, Relation>,
 
     // Tables in the publication on the publisher.
     tables: HashMap<Key, Table>,
 
     // Statements
-    statements: HashMap<i32, Statements>,
+    statements: HashMap<Oid, Statements>,
 
     // Partitioned tables dedup.
     partitioned_dedup: HashSet<Key>,
 
     // LSNs for each table
-    table_lsns: HashMap<i32, i64>,
+    table_lsns: HashMap<Oid, i64>,
 
     // Connections to shards.
     connections: Vec<Server>,
@@ -388,7 +389,7 @@ impl StreamSubscriber {
         Ok(())
     }
 
-    pub(crate) fn lsn_applied(&self, oid: &i32) -> bool {
+    pub(crate) fn lsn_applied(&self, oid: &Oid) -> bool {
         if let Some(table_lsn) = self.table_lsns.get(oid) {
             // Don't apply change if table is ahead.
             if self.lsn < *table_lsn {