sqlite_diff_rs/wire/
impls_pg_binary.rs

1//! `Decoder` implementations and `TypeMapDefaults` for the [`PgBinary`]
2//! source: decoding PostgreSQL binary result fields straight into
3//! [`Value`].
4//!
5//! The Postgres binary send format is the same whether it arrives over
6//! logical replication in binary mode or as a binary query result, so
7//! these decoders mirror the binary arms of the `PgWalstream` impls and
8//! produce byte-identical [`Value`]s.
9
10use alloc::string::ToString;
11use alloc::vec::Vec;
12
13use super::decoder::{
14    BoolDecoder, DateVerbatimDecoder, DecimalTextDecoder, Decoder, IntDecoder,
15    IntervalVerbatimDecoder, JsonVerbatimDecoder, NullDecoder, PgByteaBinaryDecoder, RealDecoder,
16    TextDecoder, TimeVerbatimDecoder, TimestampTzVerbatimDecoder, TimestampVerbatimDecoder,
17    UuidBlob16Decoder,
18};
19use super::error::DecodeError;
20use super::scalar_helpers::{decode_pg_bool_binary, decode_pg_int_binary, decode_pg_real_binary};
21use super::source::{PgBinary, PgBinaryColumn};
22use super::type_map::{TypeMap, TypeMapDefaults};
23use super::wire_type::WireType;
24use crate::encoding::Value;
25
26impl<S, B> Decoder<PgBinary, S, B> for NullDecoder {
27    fn decode(&self, _payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
28        Ok(Value::Null)
29    }
30}
31
32// ------------------------------------------------------------------
33// BoolDecoder: single byte 0x01 -> 1, 0x00 -> 0. Null pass-through.
34// ------------------------------------------------------------------
35
36impl<S, B> Decoder<PgBinary, S, B> for BoolDecoder {
37    fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
38        match payload.raw {
39            None => Ok(Value::Null),
40            Some(bytes) => decode_pg_bool_binary(payload.column_name, bytes),
41        }
42    }
43}
44
45// ------------------------------------------------------------------
46// IntDecoder: int2/int4/int8 as 2/4/8-byte big-endian two's complement.
47// ------------------------------------------------------------------
48
49impl<S, B> Decoder<PgBinary, S, B> for IntDecoder {
50    fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
51        match payload.raw {
52            None => Ok(Value::Null),
53            Some(bytes) => decode_pg_int_binary(payload.column_name, bytes),
54        }
55    }
56}
57
58// ------------------------------------------------------------------
59// RealDecoder: float4/float8 as 4/8-byte big-endian IEEE 754. NaN
60// normalizes to Null, -0.0 to 0.0, matching `decode_value`.
61// ------------------------------------------------------------------
62
63impl<S, B> Decoder<PgBinary, S, B> for RealDecoder {
64    fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
65        match payload.raw {
66            None => Ok(Value::Null),
67            Some(bytes) => decode_pg_real_binary(payload.column_name, bytes),
68        }
69    }
70}
71
72// ------------------------------------------------------------------
73// TextDecoder: UTF-8 bytes verbatim. Invalid UTF-8 -> InvalidUtf8.
74// ------------------------------------------------------------------
75
76impl<S, B> Decoder<PgBinary, S, B> for TextDecoder
77where
78    S: From<alloc::string::String>,
79{
80    fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
81        let Some(bytes) = payload.raw else {
82            return Ok(Value::Null);
83        };
84        match core::str::from_utf8(bytes) {
85            Ok(s) => Ok(Value::Text(S::from(s.to_string()))),
86            Err(_) => Err(DecodeError::InvalidUtf8 {
87                column: payload.column_name.to_string(),
88            }),
89        }
90    }
91}
92
93// ------------------------------------------------------------------
94// PgByteaBinaryDecoder: raw bytes verbatim into Value::Blob.
95// ------------------------------------------------------------------
96
97impl<S, B> Decoder<PgBinary, S, B> for PgByteaBinaryDecoder
98where
99    B: From<Vec<u8>>,
100{
101    fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
102        let Some(bytes) = payload.raw else {
103            return Ok(Value::Null);
104        };
105        Ok(Value::Blob(B::from(bytes.to_vec())))
106    }
107}
108
109// ------------------------------------------------------------------
110// UuidBlob16Decoder: the source bytes are already the 16 raw uuid
111// bytes, taken verbatim when the length is 16. Any other length errors.
112// The output matches the CDC path's 16-byte blob, which is the point.
113// ------------------------------------------------------------------
114
115impl<S, B> Decoder<PgBinary, S, B> for UuidBlob16Decoder
116where
117    B: From<Vec<u8>>,
118{
119    fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
120        let Some(bytes) = payload.raw else {
121            return Ok(Value::Null);
122        };
123        if bytes.len() == 16 {
124            Ok(Value::Blob(B::from(bytes.to_vec())))
125        } else {
126            Err(DecodeError::InvalidUuid {
127                column: payload.column_name.to_string(),
128                source_len: bytes.len(),
129            })
130        }
131    }
132}
133
134// ------------------------------------------------------------------
135// Deferred set. Their binary layouts are numeric and must be rendered
136// back to text byte-identically to the verbatim/decimal CDC decoders
137// before they can be enabled. Until then they return a clear error
138// rather than a lossy or diverging value. The `defaults()` map already
139// routes each deferred WireType to its eventual decoder so enabling one
140// later is a body change, not a wiring change.
141// ------------------------------------------------------------------
142
143macro_rules! not_yet_impl {
144    ($decoder:ty) => {
145        impl<S, B> Decoder<PgBinary, S, B> for $decoder {
146            fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
147                if payload.raw.is_none() {
148                    return Ok(Value::Null);
149                }
150                Err(DecodeError::NotYetImplemented {
151                    decoder: stringify!($decoder),
152                })
153            }
154        }
155    };
156}
157
158not_yet_impl!(DecimalTextDecoder);
159not_yet_impl!(TimestampVerbatimDecoder);
160not_yet_impl!(TimestampTzVerbatimDecoder);
161not_yet_impl!(DateVerbatimDecoder);
162not_yet_impl!(TimeVerbatimDecoder);
163not_yet_impl!(IntervalVerbatimDecoder);
164not_yet_impl!(JsonVerbatimDecoder);
165
166impl<S, B> TypeMapDefaults<S, B> for PgBinary
167where
168    S: From<alloc::string::String>,
169    B: From<Vec<u8>>,
170{
171    fn defaults() -> TypeMap<Self, S, B> {
172        TypeMap::new()
173            .with(WireType::Bool, BoolDecoder)
174            .with(WireType::Int, IntDecoder)
175            .with(WireType::Real, RealDecoder)
176            .with(WireType::Text, TextDecoder)
177            .with(WireType::Bytes, PgByteaBinaryDecoder)
178            .with(WireType::Uuid, UuidBlob16Decoder)
179            .with(WireType::Decimal, DecimalTextDecoder)
180            .with(WireType::Timestamp, TimestampVerbatimDecoder)
181            .with(WireType::TimestampTz, TimestampTzVerbatimDecoder)
182            .with(WireType::Date, DateVerbatimDecoder)
183            .with(WireType::Time, TimeVerbatimDecoder)
184            .with(WireType::Interval, IntervalVerbatimDecoder)
185            .with(WireType::Json, JsonVerbatimDecoder)
186            .with(WireType::Jsonb, JsonVerbatimDecoder)
187    }
188}
sqlite_diff_rs/wire/impls_pg_binary.rs

sqlite_diff_rs/wire/
impls_pg_binary.rs
