drasi_postgres_common/
decode.rs

1// Copyright 2025 The Drasi Authors.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Text-format PostgreSQL value decoding (pgoutput / bootstrap cast-to-text).
16
17use crate::oid::{self, array_element_oid, is_array_oid};
18use crate::value::{parse_bytea_text, PostgresValue};
19use anyhow::{anyhow, Result};
20use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Utc};
21use drasi_core::models::ElementValue;
22use rust_decimal::Decimal;
23use serde_json::Value as JsonValue;
24use std::sync::Arc;
25use uuid::Uuid;
26
27/// Decode a pgoutput/text column into a [`PostgresValue`].
28///
29/// Unknown or unparseable values fall back to `Text` rather than hard-failing,
30/// except where a clear type-specific parse is expected and fails in a way that
31/// callers may want to handle (still returns Ok with Text fallback for safety).
32pub fn decode_text_to_postgres_value(text: &str, type_oid: u32) -> Result<PostgresValue> {
33    let trimmed = text.trim();
34
35    if is_array_oid(type_oid) {
36        let elem_oid = array_element_oid(type_oid).unwrap_or(oid::TEXT);
37        return parse_array_text(trimmed, elem_oid);
38    }
39
40    match type_oid {
41        oid::BOOL => {
42            // Strict pgoutput forms: t/f/true/false
43            let value = match trimmed {
44                "t" | "true" => true,
45                "f" | "false" => false,
46                _ => return Err(anyhow!("Invalid boolean value for OID {type_oid}")),
47            };
48            Ok(PostgresValue::Bool(value))
49        }
50        oid::INT2 => {
51            Ok(PostgresValue::Int2(trimmed.parse::<i16>().map_err(
52                |e| anyhow!("Failed to parse int2 (OID {type_oid}): {e}"),
53            )?))
54        }
55        oid::INT4 => {
56            Ok(PostgresValue::Int4(trimmed.parse::<i32>().map_err(
57                |e| anyhow!("Failed to parse int4 (OID {type_oid}): {e}"),
58            )?))
59        }
60        oid::INT8 => {
61            Ok(PostgresValue::Int8(trimmed.parse::<i64>().map_err(
62                |e| anyhow!("Failed to parse int8 (OID {type_oid}): {e}"),
63            )?))
64        }
65        oid::FLOAT4 => {
66            Ok(PostgresValue::Float4(trimmed.parse::<f32>().map_err(
67                |e| anyhow!("Failed to parse float4 (OID {type_oid}): {e}"),
68            )?))
69        }
70        oid::FLOAT8 => {
71            Ok(PostgresValue::Float8(trimmed.parse::<f64>().map_err(
72                |e| anyhow!("Failed to parse float8 (OID {type_oid}): {e}"),
73            )?))
74        }
75        oid::NUMERIC => {
76            let value = Decimal::from_str_exact(trimmed)
77                .or_else(|_| trimmed.parse::<Decimal>())
78                .map_err(|e| anyhow!("Failed to parse numeric (OID {type_oid}): {e}"))?;
79            Ok(PostgresValue::Numeric(value))
80        }
81        oid::TEXT | oid::NAME => Ok(PostgresValue::Text(text.to_string())),
82        oid::VARCHAR => Ok(PostgresValue::Varchar(text.to_string())),
83        oid::CHAR => Ok(PostgresValue::Char(text.trim_end().to_string())),
84        oid::UUID => {
85            let uuid = Uuid::parse_str(trimmed)
86                .map_err(|e| anyhow!("Failed to parse uuid (OID {type_oid}): {e}"))?;
87            Ok(PostgresValue::Uuid(uuid))
88        }
89        oid::TIMESTAMP => {
90            if let Ok(dt) = NaiveDateTime::parse_from_str(trimmed, "%Y-%m-%d %H:%M:%S%.f") {
91                Ok(PostgresValue::Timestamp(dt))
92            } else if let Ok(dt) = NaiveDateTime::parse_from_str(trimmed, "%Y-%m-%d %H:%M:%S") {
93                Ok(PostgresValue::Timestamp(dt))
94            } else {
95                // Malformed for this OID (e.g. includes timezone) — keep as text
96                Ok(PostgresValue::Text(text.to_string()))
97            }
98        }
99        oid::TIMESTAMPTZ => decode_timestamptz_text(trimmed, text),
100        oid::DATE => {
101            if let Ok(d) = NaiveDate::parse_from_str(trimmed, "%Y-%m-%d") {
102                Ok(PostgresValue::Date(d))
103            } else {
104                Ok(PostgresValue::Text(text.to_string()))
105            }
106        }
107        oid::TIME => {
108            if let Ok(t) = NaiveTime::parse_from_str(trimmed, "%H:%M:%S%.f") {
109                Ok(PostgresValue::Time(t))
110            } else if let Ok(t) = NaiveTime::parse_from_str(trimmed, "%H:%M:%S") {
111                Ok(PostgresValue::Time(t))
112            } else {
113                Ok(PostgresValue::Text(text.to_string()))
114            }
115        }
116        oid::JSON => {
117            let value: JsonValue = serde_json::from_str(trimmed)
118                .map_err(|e| anyhow!("Failed to parse json (OID {type_oid}): {e}"))?;
119            Ok(PostgresValue::Json(value))
120        }
121        oid::JSONB => {
122            // Text mode has no version byte
123            let value: JsonValue = serde_json::from_str(trimmed)
124                .map_err(|e| anyhow!("Failed to parse jsonb (OID {type_oid}): {e}"))?;
125            Ok(PostgresValue::Jsonb(value))
126        }
127        oid::BYTEA => {
128            let bytes = parse_bytea_text(trimmed)?;
129            Ok(PostgresValue::Bytea(bytes))
130        }
131        _ => Ok(PostgresValue::Text(text.to_string())),
132    }
133}
134
135fn decode_timestamptz_text(trimmed: &str, original: &str) -> Result<PostgresValue> {
136    // RFC3339 first
137    if let Ok(dt) = DateTime::parse_from_rfc3339(trimmed) {
138        return Ok(PostgresValue::TimestampTz(dt.with_timezone(&Utc)));
139    }
140    // Full offset with optional fractional seconds: +00:00 / +0000 / short +00 / Z
141    if let Ok(dt) = DateTime::parse_from_str(trimmed, "%Y-%m-%d %H:%M:%S%.f%#z") {
142        return Ok(PostgresValue::TimestampTz(dt.with_timezone(&Utc)));
143    }
144    if let Ok(dt) = DateTime::parse_from_str(trimmed, "%Y-%m-%d %H:%M:%S%#z") {
145        return Ok(PostgresValue::TimestampTz(dt.with_timezone(&Utc)));
146    }
147    if let Ok(dt) = DateTime::parse_from_str(trimmed, "%Y-%m-%dT%H:%M:%S%.f%#z") {
148        return Ok(PostgresValue::TimestampTz(dt.with_timezone(&Utc)));
149    }
150    // No offset — assume UTC (matches existing bootstrap text helper)
151    if let Ok(dt) = NaiveDateTime::parse_from_str(trimmed, "%Y-%m-%d %H:%M:%S%.f") {
152        return Ok(PostgresValue::TimestampTz(dt.and_utc()));
153    }
154    if let Ok(dt) = NaiveDateTime::parse_from_str(trimmed, "%Y-%m-%d %H:%M:%S") {
155        return Ok(PostgresValue::TimestampTz(dt.and_utc()));
156    }
157    Ok(PostgresValue::Text(original.to_string()))
158}
159
160/// Decode text directly to [`ElementValue`] using the canonical mapping.
161pub fn decode_text_to_element_value(text: &str, type_oid: i32) -> Result<ElementValue> {
162    Ok(decode_text_to_postgres_value(text, type_oid as u32)?.to_element_value())
163}
164
165/// Back-compat alias used by existing tests in the source crate.
166pub fn decode_column_value_text(text: &str, type_oid: i32) -> Result<ElementValue> {
167    decode_text_to_element_value(text, type_oid)
168}
169
170/// Parse a PostgreSQL 1-D array text literal like `{1,2,NULL,"a"}`.
171///
172/// Quote markers are tracked separately from element content so a quoted
173/// `"NULL"` string is preserved while an unquoted `NULL` becomes
174/// [`PostgresValue::Null`].
175fn parse_array_text(text: &str, element_oid: u32) -> Result<PostgresValue> {
176    let s = text.trim();
177    let inner = if let Some(body) = s.strip_prefix('{').and_then(|x| x.strip_suffix('}')) {
178        body
179    } else {
180        // Not an array literal — keep as text
181        return Ok(PostgresValue::Text(text.to_string()));
182    };
183
184    if inner.is_empty() {
185        return Ok(PostgresValue::Array(vec![]));
186    }
187
188    let mut elements = Vec::new();
189    let mut cur = String::new();
190    let mut in_quotes = false;
191    let mut escape = false;
192    // True if this element used PG array double-quote delimiters.
193    let mut was_quoted = false;
194    for c in inner.chars() {
195        if escape {
196            cur.push(c);
197            escape = false;
198            continue;
199        }
200        match c {
201            '\\' if in_quotes => {
202                escape = true;
203            }
204            '"' => {
205                in_quotes = !in_quotes;
206                was_quoted = true;
207            }
208            ',' if !in_quotes => {
209                elements.push(parse_array_element(&cur, element_oid, was_quoted)?);
210                cur.clear();
211                was_quoted = false;
212            }
213            _ => cur.push(c),
214        }
215    }
216    elements.push(parse_array_element(&cur, element_oid, was_quoted)?);
217
218    Ok(PostgresValue::Array(elements))
219}
220
221fn parse_array_element(raw: &str, element_oid: u32, was_quoted: bool) -> Result<PostgresValue> {
222    let t = raw.trim();
223    // Unquoted NULL is SQL null. Quoted "NULL" is the literal string (or typed value).
224    if !was_quoted && t.eq_ignore_ascii_case("NULL") {
225        return Ok(PostgresValue::Null);
226    }
227    decode_text_to_postgres_value(t, element_oid)
228}
229
230/// Format helper used when a column cannot be converted — preserve as string ElementValue.
231pub fn string_element(s: impl AsRef<str>) -> ElementValue {
232    ElementValue::String(Arc::from(s.as_ref()))
233}
234
235#[cfg(test)]
236mod tests {
237    use super::*;
238    use chrono::{DateTime, NaiveDate, Utc};
239
240    #[test]
241    fn decode_uuid_text() {
242        let pv = decode_text_to_postgres_value("550e8400-e29b-41d4-a716-446655440000", oid::UUID)
243            .unwrap();
244        match pv {
245            PostgresValue::Uuid(u) => {
246                assert_eq!(u.to_string(), "550e8400-e29b-41d4-a716-446655440000");
247            }
248            other => panic!("expected Uuid, got {other:?}"),
249        }
250    }
251
252    #[test]
253    fn decode_date_text() {
254        let pv = decode_text_to_postgres_value("2024-06-15", oid::DATE).unwrap();
255        assert!(matches!(
256            pv,
257            PostgresValue::Date(d) if d == NaiveDate::from_ymd_opt(2024, 6, 15).unwrap()
258        ));
259    }
260
261    #[test]
262    fn decode_time_text() {
263        let pv = decode_text_to_postgres_value("10:30:45.123456", oid::TIME).unwrap();
264        match pv {
265            PostgresValue::Time(t) => assert_eq!(t.to_string(), "10:30:45.123456"),
266            other => panic!("expected Time, got {other:?}"),
267        }
268    }
269
270    #[test]
271    fn decode_jsonb_text_no_version_byte() {
272        let pv = decode_text_to_postgres_value(r#"{"k":1}"#, oid::JSONB).unwrap();
273        match pv {
274            PostgresValue::Jsonb(v) => assert_eq!(v["k"], 1),
275            other => panic!("expected Jsonb, got {other:?}"),
276        }
277    }
278
279    #[test]
280    fn decode_bytea_hex() {
281        let pv = decode_text_to_postgres_value(r"\xdeadbeef", oid::BYTEA).unwrap();
282        match pv {
283            PostgresValue::Bytea(b) => assert_eq!(b, vec![0xde, 0xad, 0xbe, 0xef]),
284            other => panic!("expected Bytea, got {other:?}"),
285        }
286    }
287
288    #[test]
289    fn decode_int_array() {
290        let pv = decode_text_to_postgres_value("{1,2,3}", oid::INT4_ARRAY).unwrap();
291        match pv {
292            PostgresValue::Array(items) => {
293                assert_eq!(items.len(), 3);
294                assert!(matches!(items[0], PostgresValue::Int4(1)));
295                assert!(matches!(items[2], PostgresValue::Int4(3)));
296            }
297            other => panic!("expected Array, got {other:?}"),
298        }
299    }
300
301    #[test]
302    fn decode_char_trims_padding() {
303        let pv = decode_text_to_postgres_value("abc     ", oid::CHAR).unwrap();
304        match pv {
305            PostgresValue::Char(s) => assert_eq!(s, "abc"),
306            other => panic!("expected Char, got {other:?}"),
307        }
308    }
309
310    #[test]
311    fn decode_timestamp_fractional() {
312        let ev = decode_column_value_text("2024-06-15 10:30:45.123456", 1114).unwrap();
313        let expected = NaiveDate::from_ymd_opt(2024, 6, 15)
314            .unwrap()
315            .and_hms_micro_opt(10, 30, 45, 123456)
316            .unwrap();
317        assert_eq!(ev, ElementValue::LocalDateTime(expected));
318    }
319
320    #[test]
321    fn decode_bool_t_f() {
322        assert_eq!(
323            decode_column_value_text("t", 16).unwrap(),
324            ElementValue::Bool(true)
325        );
326        assert_eq!(
327            decode_column_value_text("f", 16).unwrap(),
328            ElementValue::Bool(false)
329        );
330    }
331
332    #[test]
333    fn parity_numeric_whole() {
334        let pv = decode_text_to_postgres_value("4200", oid::NUMERIC).unwrap();
335        match pv.to_element_value() {
336            ElementValue::Float(f) => assert_eq!(f.into_inner(), 4200.0),
337            other => panic!("expected Float, got {other:?}"),
338        }
339    }
340
341    #[test]
342    fn decode_text_array_distinguishes_null_and_quoted_null() {
343        let pv = decode_text_to_postgres_value(r#"{NULL,"NULL","a"}"#, oid::TEXT_ARRAY).unwrap();
344        match pv {
345            PostgresValue::Array(items) => {
346                assert_eq!(items.len(), 3);
347                assert!(matches!(items[0], PostgresValue::Null));
348                assert!(matches!(items[1], PostgresValue::Text(ref s) if s == "NULL"));
349                assert!(matches!(items[2], PostgresValue::Text(ref s) if s == "a"));
350            }
351            other => panic!("expected Array, got {other:?}"),
352        }
353    }
354
355    #[test]
356    fn decode_int_array_null_element() {
357        let pv = decode_text_to_postgres_value("{1,NULL,3}", oid::INT4_ARRAY).unwrap();
358        match pv {
359            PostgresValue::Array(items) => {
360                assert!(matches!(items[0], PostgresValue::Int4(1)));
361                assert!(matches!(items[1], PostgresValue::Null));
362                assert!(matches!(items[2], PostgresValue::Int4(3)));
363            }
364            other => panic!("expected Array, got {other:?}"),
365        }
366    }
367
368    #[test]
369    fn decode_timestamptz_rfc3339() {
370        let pv =
371            decode_text_to_postgres_value("2024-06-15T10:30:45+02:00", oid::TIMESTAMPTZ).unwrap();
372        match pv {
373            PostgresValue::TimestampTz(ts) => {
374                assert_eq!(ts.to_rfc3339(), "2024-06-15T08:30:45+00:00");
375            }
376            other => panic!("expected TimestampTz, got {other:?}"),
377        }
378    }
379
380    #[test]
381    fn decode_timestamptz_postgres_offset() {
382        let pv =
383            decode_text_to_postgres_value("2024-06-15 10:30:45.123456+02:00", oid::TIMESTAMPTZ)
384                .unwrap();
385        let expected = DateTime::parse_from_rfc3339("2024-06-15T10:30:45.123456+02:00")
386            .unwrap()
387            .with_timezone(&Utc);
388        match pv {
389            PostgresValue::TimestampTz(ts) => {
390                assert_eq!(ts, expected);
391            }
392            other => panic!("expected TimestampTz, got {other:?}"),
393        }
394    }
395
396    #[test]
397    fn decode_timestamptz_without_offset_assumes_utc() {
398        let pv = decode_text_to_postgres_value("2024-06-15 10:30:45", oid::TIMESTAMPTZ).unwrap();
399        match pv {
400            PostgresValue::TimestampTz(ts) => {
401                assert_eq!(ts.to_rfc3339(), "2024-06-15T10:30:45+00:00");
402            }
403            other => panic!("expected TimestampTz, got {other:?}"),
404        }
405    }
406
407    #[test]
408    fn decode_timestamptz_offset_matches_rfc3339_instant() {
409        let a =
410            decode_text_to_postgres_value("2024-06-15T10:30:45+02:00", oid::TIMESTAMPTZ).unwrap();
411        let b =
412            decode_text_to_postgres_value("2024-06-15 10:30:45+02:00", oid::TIMESTAMPTZ).unwrap();
413        match (a, b) {
414            (PostgresValue::TimestampTz(ta), PostgresValue::TimestampTz(tb)) => {
415                assert_eq!(ta.timestamp_micros(), tb.timestamp_micros());
416            }
417            other => panic!("expected TimestampTz pair, got {other:?}"),
418        }
419    }
420}
drasi_postgres_common/decode.rs

drasi_postgres_common/
decode.rs
