reddb_server/wire/postgres/
types.rs

1//! PostgreSQL type OID mapping (Phase 3.1 PG parity).
2//!
3//! PG clients identify column types by OID (see `pg_catalog.pg_type`).
4//! RedDB's `Value` enum is richer than PG's canonical set, so we collapse
5//! domain-specific variants (`Email`, `Phone`, `Money`, ...) onto their
6//! closest PG equivalent — TEXT, NUMERIC, etc. This keeps generic clients
7//! working; clients that need the fine-grained types call the native
8//! gRPC path.
9//!
10//! Reference: PostgreSQL source `src/include/catalog/pg_type_d.h`.
11
12use crate::runtime::query_request::ParamValue;
13use reddb_types::Value;
14
15/// A subset of PG type OIDs that cover every case we need to encode.
16#[allow(dead_code)]
17#[derive(Debug, Clone, Copy, PartialEq, Eq)]
18pub enum PgOid {
19    Bool = 16,
20    Bytea = 17,
21    Int8 = 20,
22    Int2 = 21,
23    Int4 = 23,
24    Text = 25,
25    Oid = 26,
26    Json = 114,
27    Float4 = 700,
28    Float8 = 701,
29    Unknown = 705,
30    Varchar = 1043,
31    Date = 1082,
32    Time = 1083,
33    Timestamp = 1114,
34    TimestampTz = 1184,
35    Numeric = 1700,
36    Uuid = 2950,
37    Jsonb = 3802,
38    /// RedDB-reserved synthetic vector OID. PostgreSQL extension OIDs are
39    /// cluster-local; RedDB uses a stable high value for wire clients that
40    /// want to bind vector parameters explicitly.
41    Vector = 38000,
42}
43
44impl PgOid {
45    pub fn as_u32(self) -> u32 {
46        self as u32
47    }
48
49    pub fn from_u32(oid: u32) -> Self {
50        match oid {
51            16 => PgOid::Bool,
52            17 => PgOid::Bytea,
53            20 => PgOid::Int8,
54            21 => PgOid::Int2,
55            23 => PgOid::Int4,
56            25 => PgOid::Text,
57            26 => PgOid::Oid,
58            114 => PgOid::Json,
59            700 => PgOid::Float4,
60            701 => PgOid::Float8,
61            705 => PgOid::Unknown,
62            1043 => PgOid::Varchar,
63            1082 => PgOid::Date,
64            1083 => PgOid::Time,
65            1114 => PgOid::Timestamp,
66            1184 => PgOid::TimestampTz,
67            1700 => PgOid::Numeric,
68            2950 => PgOid::Uuid,
69            3802 => PgOid::Jsonb,
70            38000 => PgOid::Vector,
71            _ => PgOid::Unknown,
72        }
73    }
74
75    /// Preferred type OID for a runtime `Value`. Used by
76    /// `RowDescription` to tell the client what each column is.
77    pub fn from_value(value: &Value) -> Self {
78        match value {
79            Value::Null => PgOid::Text,
80            Value::Boolean(_) => PgOid::Bool,
81            Value::Integer(_) => PgOid::Int8,
82            Value::UnsignedInteger(_) => PgOid::Int8,
83            Value::BigInt(_) => PgOid::Int8,
84            Value::Float(_) => PgOid::Float8,
85            Value::Text(_) => PgOid::Text,
86            Value::Blob(_) => PgOid::Bytea,
87            Value::Json(_) => PgOid::Jsonb,
88            Value::Uuid(_) => PgOid::Uuid,
89            Value::Date(_) => PgOid::Date,
90            Value::Timestamp(_) => PgOid::TimestampTz,
91            Value::TimestampMs(_) => PgOid::TimestampTz,
92            Value::Vector(_) => PgOid::Vector,
93            // Domain / richer types collapse to TEXT so psql can render them.
94            _ => PgOid::Text,
95        }
96    }
97}
98
99pub fn pg_param_to_param_value(
100    oid: u32,
101    format_code: i16,
102    bytes: Option<&[u8]>,
103) -> Result<ParamValue, String> {
104    let Some(bytes) = bytes else {
105        return Ok(ParamValue::Null);
106    };
107    match format_code {
108        0 => pg_text_param_to_value(PgOid::from_u32(oid), bytes),
109        1 => pg_binary_param_to_value(PgOid::from_u32(oid), bytes),
110        other => Err(format!("unsupported PG parameter format code {other}")),
111    }
112}
113
114fn pg_text_param_to_value(oid: PgOid, bytes: &[u8]) -> Result<ParamValue, String> {
115    let text = std::str::from_utf8(bytes).map_err(|e| format!("invalid UTF-8 parameter: {e}"))?;
116    match oid {
117        PgOid::Bool => match text.to_ascii_lowercase().as_str() {
118            "t" | "true" | "1" | "yes" | "on" => Ok(ParamValue::Bool(true)),
119            "f" | "false" | "0" | "no" | "off" => Ok(ParamValue::Bool(false)),
120            _ => Err(format!("invalid bool parameter {text:?}")),
121        },
122        PgOid::Int2 | PgOid::Int4 | PgOid::Int8 | PgOid::Oid => text
123            .parse::<i64>()
124            .map(ParamValue::Int64)
125            .map_err(|e| format!("invalid integer parameter {text:?}: {e}")),
126        PgOid::Float4 | PgOid::Float8 | PgOid::Numeric => text
127            .parse::<f64>()
128            .map(ParamValue::Float64)
129            .map_err(|e| format!("invalid float parameter {text:?}: {e}")),
130        PgOid::Bytea => parse_bytea_text(text).map(ParamValue::Bytes),
131        PgOid::Json | PgOid::Jsonb => Ok(ParamValue::Json(bytes.to_vec())),
132        PgOid::Timestamp | PgOid::TimestampTz => text
133            .parse::<i64>()
134            .map(ParamValue::Timestamp)
135            .or_else(|_| Ok(ParamValue::Text(text.to_string()))),
136        PgOid::Uuid => parse_uuid_text(text).map(ParamValue::Uuid),
137        PgOid::Vector => parse_vector_text(text).map(ParamValue::Vector),
138        PgOid::Text | PgOid::Varchar | PgOid::Unknown | PgOid::Date | PgOid::Time => {
139            Ok(ParamValue::Text(text.to_string()))
140        }
141    }
142}
143
144fn pg_binary_param_to_value(oid: PgOid, bytes: &[u8]) -> Result<ParamValue, String> {
145    match oid {
146        PgOid::Bool if bytes.len() == 1 => Ok(ParamValue::Bool(bytes[0] != 0)),
147        PgOid::Int2 if bytes.len() == 2 => {
148            Ok(ParamValue::Int64(
149                i16::from_be_bytes([bytes[0], bytes[1]]) as i64
150            ))
151        }
152        PgOid::Int4 | PgOid::Oid if bytes.len() == 4 => {
153            Ok(ParamValue::Int64(
154                i32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) as i64,
155            ))
156        }
157        PgOid::Int8 if bytes.len() == 8 => Ok(ParamValue::Int64(i64::from_be_bytes([
158            bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
159        ]))),
160        PgOid::Float4 if bytes.len() == 4 => {
161            Ok(ParamValue::Float64(
162                f32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) as f64,
163            ))
164        }
165        PgOid::Float8 if bytes.len() == 8 => Ok(ParamValue::Float64(f64::from_be_bytes([
166            bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
167        ]))),
168        PgOid::Text | PgOid::Varchar => std::str::from_utf8(bytes)
169            .map(|text| ParamValue::Text(text.to_owned()))
170            .map_err(|_| "invalid UTF-8 binary text parameter".to_string()),
171        PgOid::Bytea => Ok(ParamValue::Bytes(bytes.to_vec())),
172        PgOid::Json | PgOid::Jsonb => Ok(ParamValue::Json(bytes.to_vec())),
173        PgOid::Uuid if bytes.len() == 16 => {
174            let mut out = [0u8; 16];
175            out.copy_from_slice(bytes);
176            Ok(ParamValue::Uuid(out))
177        }
178        PgOid::Timestamp | PgOid::TimestampTz if bytes.len() == 8 => Ok(ParamValue::Timestamp(
179            pg_microseconds_to_unix_seconds(i64::from_be_bytes([
180                bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
181            ])),
182        )),
183        PgOid::Vector => parse_vector_binary(bytes).map(ParamValue::Vector),
184        _ => Err(format!(
185            "unsupported binary parameter for OID {} with {} bytes",
186            oid.as_u32(),
187            bytes.len()
188        )),
189    }
190}
191
192fn parse_bytea_text(text: &str) -> Result<Vec<u8>, String> {
193    let Some(hex) = text.strip_prefix("\\x") else {
194        return Ok(text.as_bytes().to_vec());
195    };
196    if hex.len() % 2 != 0 {
197        return Err("invalid bytea hex length".to_string());
198    }
199    (0..hex.len())
200        .step_by(2)
201        .map(|idx| {
202            u8::from_str_radix(&hex[idx..idx + 2], 16)
203                .map_err(|e| format!("invalid bytea hex: {e}"))
204        })
205        .collect()
206}
207
208fn parse_vector_text(text: &str) -> Result<Vec<f32>, String> {
209    let parsed: crate::json::Value =
210        crate::json::from_str(text).map_err(|e| format!("invalid vector parameter: {e}"))?;
211    let crate::json::Value::Array(items) = parsed else {
212        return Err("invalid vector parameter: expected JSON number array".to_string());
213    };
214    items
215        .iter()
216        .map(|item| {
217            item.as_f64().map(|value| value as f32).ok_or_else(|| {
218                "invalid vector parameter: array must contain only numbers".to_string()
219            })
220        })
221        .collect()
222}
223
224fn parse_vector_binary(bytes: &[u8]) -> Result<Vec<f32>, String> {
225    if bytes.len() < 4 {
226        return Err("invalid binary vector parameter: payload too short".to_string());
227    }
228    let dims = i16::from_be_bytes([bytes[0], bytes[1]]);
229    if dims < 0 {
230        return Err("invalid binary vector parameter: negative dimension".to_string());
231    }
232    let dims = dims as usize;
233    let expected = 4 + dims * 4;
234    if bytes.len() != expected {
235        return Err(format!(
236            "invalid binary vector parameter: expected {expected} bytes, got {}",
237            bytes.len()
238        ));
239    }
240    (0..dims)
241        .map(|idx| {
242            let off = 4 + idx * 4;
243            Ok(f32::from_be_bytes([
244                bytes[off],
245                bytes[off + 1],
246                bytes[off + 2],
247                bytes[off + 3],
248            ]))
249        })
250        .collect()
251}
252
253fn parse_uuid_text(text: &str) -> Result<[u8; 16], String> {
254    let compact = text.replace('-', "");
255    if compact.len() != 32 {
256        return Err(format!("invalid uuid parameter {text:?}"));
257    }
258    let bytes = (0..compact.len())
259        .step_by(2)
260        .map(|idx| {
261            u8::from_str_radix(&compact[idx..idx + 2], 16)
262                .map_err(|e| format!("invalid uuid parameter {text:?}: {e}"))
263        })
264        .collect::<Result<Vec<_>, _>>()?;
265    let mut out = [0u8; 16];
266    out.copy_from_slice(&bytes);
267    Ok(out)
268}
269
270fn pg_microseconds_to_unix_seconds(pg_micros: i64) -> i64 {
271    // PostgreSQL binary timestamps are microseconds since 2000-01-01.
272    const PG_UNIX_EPOCH_OFFSET_SECONDS: i64 = 946_684_800;
273    PG_UNIX_EPOCH_OFFSET_SECONDS + pg_micros / 1_000_000
274}
275
276/// Encode a `Value` as the UTF-8 text representation PG's text-mode
277/// protocol expects. Binary format is opt-in via a flag in the client's
278/// `Bind` message — we don't advertise binary support yet, so simple
279/// text encoding is sufficient for every supported client.
280///
281/// Returns `None` for `Value::Null` (the caller emits a `-1` length).
282pub fn value_to_pg_wire_bytes(value: &Value) -> Option<Vec<u8>> {
283    match value {
284        Value::Null => None,
285        Value::Boolean(b) => Some((if *b { "t" } else { "f" }).as_bytes().to_vec()),
286        Value::Integer(n) => Some(n.to_string().into_bytes()),
287        Value::UnsignedInteger(n) => Some(n.to_string().into_bytes()),
288        Value::BigInt(n) => Some(n.to_string().into_bytes()),
289        Value::Float(f) => Some(f.to_string().into_bytes()),
290        Value::Text(s) => Some(s.as_bytes().to_vec()),
291        Value::Blob(b) => {
292            // PG bytea text format: `\xHEX...`. Two chars per byte.
293            let mut out = Vec::with_capacity(2 + b.len() * 2);
294            out.extend_from_slice(b"\\x");
295            for byte in b {
296                out.extend_from_slice(format!("{byte:02x}").as_bytes());
297            }
298            Some(out)
299        }
300        Value::Json(bytes) => Some(bytes.clone()),
301        // Everything else renders via Display — catches Uuid, Date,
302        // Timestamp, Email, Phone, Money, GeoPoint, etc. PG clients see
303        // these as TEXT columns (OID 25) and can render them verbatim.
304        other => Some(other.to_string().into_bytes()),
305    }
306}
307
308#[cfg(test)]
309mod tests {
310    use super::*;
311
312    #[test]
313    fn pg_text_params_decode_common_oids() {
314        assert_eq!(
315            pg_param_to_param_value(PgOid::Bool.as_u32(), 0, Some(b"t")).unwrap(),
316            ParamValue::Bool(true)
317        );
318        assert_eq!(
319            pg_param_to_param_value(PgOid::Int4.as_u32(), 0, Some(b"42")).unwrap(),
320            ParamValue::Int64(42)
321        );
322        assert_eq!(
323            pg_param_to_param_value(PgOid::Float8.as_u32(), 0, Some(b"1.5")).unwrap(),
324            ParamValue::Float64(1.5)
325        );
326        assert_eq!(
327            pg_param_to_param_value(PgOid::Bytea.as_u32(), 0, Some(br"\xdeadbeef")).unwrap(),
328            ParamValue::Bytes(vec![0xde, 0xad, 0xbe, 0xef])
329        );
330        assert_eq!(
331            pg_param_to_param_value(PgOid::Jsonb.as_u32(), 0, Some(br#"{"a":1}"#)).unwrap(),
332            ParamValue::Json(br#"{"a":1}"#.to_vec())
333        );
334        assert_eq!(
335            pg_param_to_param_value(
336                PgOid::Uuid.as_u32(),
337                0,
338                Some(b"00112233-4455-6677-8899-aabbccddeeff")
339            )
340            .unwrap(),
341            ParamValue::Uuid([
342                0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd,
343                0xee, 0xff,
344            ])
345        );
346    }
347
348    #[test]
349    fn pg_binary_params_decode_numeric_and_uuid_oids() {
350        assert_eq!(
351            pg_param_to_param_value(PgOid::Int2.as_u32(), 1, Some(&7i16.to_be_bytes())).unwrap(),
352            ParamValue::Int64(7)
353        );
354        assert_eq!(
355            pg_param_to_param_value(PgOid::Int8.as_u32(), 1, Some(&42i64.to_be_bytes())).unwrap(),
356            ParamValue::Int64(42)
357        );
358        assert_eq!(
359            pg_param_to_param_value(PgOid::Float4.as_u32(), 1, Some(&1.5f32.to_be_bytes()))
360                .unwrap(),
361            ParamValue::Float64(1.5)
362        );
363        assert_eq!(
364            pg_param_to_param_value(PgOid::Uuid.as_u32(), 1, Some(&[0x11; 16])).unwrap(),
365            ParamValue::Uuid([0x11; 16])
366        );
367        let mut vector = Vec::new();
368        vector.extend_from_slice(&2i16.to_be_bytes());
369        vector.extend_from_slice(&0i16.to_be_bytes());
370        vector.extend_from_slice(&1.0f32.to_be_bytes());
371        vector.extend_from_slice(&(-0.5f32).to_be_bytes());
372        assert_eq!(
373            pg_param_to_param_value(PgOid::Vector.as_u32(), 1, Some(&vector)).unwrap(),
374            ParamValue::Vector(vec![1.0, -0.5])
375        );
376    }
377
378    #[test]
379    fn pg_binary_text_parameters_validate_utf8() {
380        for oid in [PgOid::Text, PgOid::Varchar] {
381            assert_eq!(
382                pg_param_to_param_value(oid.as_u32(), 1, Some("olá".as_bytes())).unwrap(),
383                ParamValue::Text("olá".to_string())
384            );
385            assert!(pg_param_to_param_value(oid.as_u32(), 1, Some(&[0xff])).is_err());
386        }
387    }
388
389    #[test]
390    fn pg_null_param_decodes_to_value_null() {
391        assert_eq!(
392            pg_param_to_param_value(PgOid::Text.as_u32(), 0, None).unwrap(),
393            ParamValue::Null
394        );
395    }
396
397    #[test]
398    fn pg_vector_text_param_decodes_json_array() {
399        assert_eq!(
400            pg_param_to_param_value(PgOid::Vector.as_u32(), 0, Some(b"[1.0, -0.5]")).unwrap(),
401            ParamValue::Vector(vec![1.0, -0.5])
402        );
403    }
404}
reddb_server/wire/postgres/types.rs

reddb_server/wire/postgres/
types.rs
