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::Bytea => Ok(ParamValue::Bytes(bytes.to_vec())),
169        PgOid::Json | PgOid::Jsonb => Ok(ParamValue::Json(bytes.to_vec())),
170        PgOid::Uuid if bytes.len() == 16 => {
171            let mut out = [0u8; 16];
172            out.copy_from_slice(bytes);
173            Ok(ParamValue::Uuid(out))
174        }
175        PgOid::Timestamp | PgOid::TimestampTz if bytes.len() == 8 => Ok(ParamValue::Timestamp(
176            pg_microseconds_to_unix_seconds(i64::from_be_bytes([
177                bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
178            ])),
179        )),
180        PgOid::Vector => parse_vector_binary(bytes).map(ParamValue::Vector),
181        _ => Err(format!(
182            "unsupported binary parameter for OID {} with {} bytes",
183            oid.as_u32(),
184            bytes.len()
185        )),
186    }
187}
188
189fn parse_bytea_text(text: &str) -> Result<Vec<u8>, String> {
190    let Some(hex) = text.strip_prefix("\\x") else {
191        return Ok(text.as_bytes().to_vec());
192    };
193    if hex.len() % 2 != 0 {
194        return Err("invalid bytea hex length".to_string());
195    }
196    (0..hex.len())
197        .step_by(2)
198        .map(|idx| {
199            u8::from_str_radix(&hex[idx..idx + 2], 16)
200                .map_err(|e| format!("invalid bytea hex: {e}"))
201        })
202        .collect()
203}
204
205fn parse_vector_text(text: &str) -> Result<Vec<f32>, String> {
206    let parsed: crate::json::Value =
207        crate::json::from_str(text).map_err(|e| format!("invalid vector parameter: {e}"))?;
208    let crate::json::Value::Array(items) = parsed else {
209        return Err("invalid vector parameter: expected JSON number array".to_string());
210    };
211    items
212        .iter()
213        .map(|item| {
214            item.as_f64().map(|value| value as f32).ok_or_else(|| {
215                "invalid vector parameter: array must contain only numbers".to_string()
216            })
217        })
218        .collect()
219}
220
221fn parse_vector_binary(bytes: &[u8]) -> Result<Vec<f32>, String> {
222    if bytes.len() < 4 {
223        return Err("invalid binary vector parameter: payload too short".to_string());
224    }
225    let dims = i16::from_be_bytes([bytes[0], bytes[1]]);
226    if dims < 0 {
227        return Err("invalid binary vector parameter: negative dimension".to_string());
228    }
229    let dims = dims as usize;
230    let expected = 4 + dims * 4;
231    if bytes.len() != expected {
232        return Err(format!(
233            "invalid binary vector parameter: expected {expected} bytes, got {}",
234            bytes.len()
235        ));
236    }
237    (0..dims)
238        .map(|idx| {
239            let off = 4 + idx * 4;
240            Ok(f32::from_be_bytes([
241                bytes[off],
242                bytes[off + 1],
243                bytes[off + 2],
244                bytes[off + 3],
245            ]))
246        })
247        .collect()
248}
249
250fn parse_uuid_text(text: &str) -> Result<[u8; 16], String> {
251    let compact = text.replace('-', "");
252    if compact.len() != 32 {
253        return Err(format!("invalid uuid parameter {text:?}"));
254    }
255    let bytes = (0..compact.len())
256        .step_by(2)
257        .map(|idx| {
258            u8::from_str_radix(&compact[idx..idx + 2], 16)
259                .map_err(|e| format!("invalid uuid parameter {text:?}: {e}"))
260        })
261        .collect::<Result<Vec<_>, _>>()?;
262    let mut out = [0u8; 16];
263    out.copy_from_slice(&bytes);
264    Ok(out)
265}
266
267fn pg_microseconds_to_unix_seconds(pg_micros: i64) -> i64 {
268    // PostgreSQL binary timestamps are microseconds since 2000-01-01.
269    const PG_UNIX_EPOCH_OFFSET_SECONDS: i64 = 946_684_800;
270    PG_UNIX_EPOCH_OFFSET_SECONDS + pg_micros / 1_000_000
271}
272
273/// Encode a `Value` as the UTF-8 text representation PG's text-mode
274/// protocol expects. Binary format is opt-in via a flag in the client's
275/// `Bind` message — we don't advertise binary support yet, so simple
276/// text encoding is sufficient for every supported client.
277///
278/// Returns `None` for `Value::Null` (the caller emits a `-1` length).
279pub fn value_to_pg_wire_bytes(value: &Value) -> Option<Vec<u8>> {
280    match value {
281        Value::Null => None,
282        Value::Boolean(b) => Some((if *b { "t" } else { "f" }).as_bytes().to_vec()),
283        Value::Integer(n) => Some(n.to_string().into_bytes()),
284        Value::UnsignedInteger(n) => Some(n.to_string().into_bytes()),
285        Value::BigInt(n) => Some(n.to_string().into_bytes()),
286        Value::Float(f) => Some(f.to_string().into_bytes()),
287        Value::Text(s) => Some(s.as_bytes().to_vec()),
288        Value::Blob(b) => {
289            // PG bytea text format: `\xHEX...`. Two chars per byte.
290            let mut out = Vec::with_capacity(2 + b.len() * 2);
291            out.extend_from_slice(b"\\x");
292            for byte in b {
293                out.extend_from_slice(format!("{byte:02x}").as_bytes());
294            }
295            Some(out)
296        }
297        Value::Json(bytes) => Some(bytes.clone()),
298        // Everything else renders via Display — catches Uuid, Date,
299        // Timestamp, Email, Phone, Money, GeoPoint, etc. PG clients see
300        // these as TEXT columns (OID 25) and can render them verbatim.
301        other => Some(other.to_string().into_bytes()),
302    }
303}
304
305#[cfg(test)]
306mod tests {
307    use super::*;
308
309    #[test]
310    fn pg_text_params_decode_common_oids() {
311        assert_eq!(
312            pg_param_to_param_value(PgOid::Bool.as_u32(), 0, Some(b"t")).unwrap(),
313            ParamValue::Bool(true)
314        );
315        assert_eq!(
316            pg_param_to_param_value(PgOid::Int4.as_u32(), 0, Some(b"42")).unwrap(),
317            ParamValue::Int64(42)
318        );
319        assert_eq!(
320            pg_param_to_param_value(PgOid::Float8.as_u32(), 0, Some(b"1.5")).unwrap(),
321            ParamValue::Float64(1.5)
322        );
323        assert_eq!(
324            pg_param_to_param_value(PgOid::Bytea.as_u32(), 0, Some(br"\xdeadbeef")).unwrap(),
325            ParamValue::Bytes(vec![0xde, 0xad, 0xbe, 0xef])
326        );
327        assert_eq!(
328            pg_param_to_param_value(PgOid::Jsonb.as_u32(), 0, Some(br#"{"a":1}"#)).unwrap(),
329            ParamValue::Json(br#"{"a":1}"#.to_vec())
330        );
331        assert_eq!(
332            pg_param_to_param_value(
333                PgOid::Uuid.as_u32(),
334                0,
335                Some(b"00112233-4455-6677-8899-aabbccddeeff")
336            )
337            .unwrap(),
338            ParamValue::Uuid([
339                0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd,
340                0xee, 0xff,
341            ])
342        );
343    }
344
345    #[test]
346    fn pg_binary_params_decode_numeric_and_uuid_oids() {
347        assert_eq!(
348            pg_param_to_param_value(PgOid::Int2.as_u32(), 1, Some(&7i16.to_be_bytes())).unwrap(),
349            ParamValue::Int64(7)
350        );
351        assert_eq!(
352            pg_param_to_param_value(PgOid::Int8.as_u32(), 1, Some(&42i64.to_be_bytes())).unwrap(),
353            ParamValue::Int64(42)
354        );
355        assert_eq!(
356            pg_param_to_param_value(PgOid::Float4.as_u32(), 1, Some(&1.5f32.to_be_bytes()))
357                .unwrap(),
358            ParamValue::Float64(1.5)
359        );
360        assert_eq!(
361            pg_param_to_param_value(PgOid::Uuid.as_u32(), 1, Some(&[0x11; 16])).unwrap(),
362            ParamValue::Uuid([0x11; 16])
363        );
364        let mut vector = Vec::new();
365        vector.extend_from_slice(&2i16.to_be_bytes());
366        vector.extend_from_slice(&0i16.to_be_bytes());
367        vector.extend_from_slice(&1.0f32.to_be_bytes());
368        vector.extend_from_slice(&(-0.5f32).to_be_bytes());
369        assert_eq!(
370            pg_param_to_param_value(PgOid::Vector.as_u32(), 1, Some(&vector)).unwrap(),
371            ParamValue::Vector(vec![1.0, -0.5])
372        );
373    }
374
375    #[test]
376    fn pg_null_param_decodes_to_value_null() {
377        assert_eq!(
378            pg_param_to_param_value(PgOid::Text.as_u32(), 0, None).unwrap(),
379            ParamValue::Null
380        );
381    }
382
383    #[test]
384    fn pg_vector_text_param_decodes_json_array() {
385        assert_eq!(
386            pg_param_to_param_value(PgOid::Vector.as_u32(), 0, Some(b"[1.0, -0.5]")).unwrap(),
387            ParamValue::Vector(vec![1.0, -0.5])
388        );
389    }
390}
reddb_server/wire/postgres/types.rs

reddb_server/wire/postgres/
types.rs
