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 share the binary arms of the `PgWalstream` impls and
8//! produce byte-identical [`Value`]s.
9
10use alloc::string::{String, ToString};
11use alloc::vec::Vec;
12
13use super::decoder::{
14    BoolDecoder, DateDecoder, DecimalTextDecoder, Decoder, IntDecoder, IntervalDecoder,
15    JsonVerbatimDecoder, NullDecoder, PgByteaBinaryDecoder, RealDecoder, TextDecoder, TimeDecoder,
16    TimestampDecoder, TimestampTzDecoder, UuidBlob16Decoder,
17};
18use super::error::DecodeError;
19use super::scalar_helpers::{
20    decode_pg_bool_binary, decode_pg_int_binary, decode_pg_json_binary, decode_pg_numeric_binary,
21    decode_pg_real_binary,
22};
23use super::source::{PgBinary, PgBinaryColumn};
24use super::temporal::{Temporal, decode_temporal_binary};
25use super::type_map::{TypeMap, TypeMapDefaults};
26use super::wire_type::WireType;
27use crate::encoding::Value;
28
29impl<S, B> Decoder<PgBinary, S, B> for NullDecoder {
30    fn decode(&self, _payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
31        Ok(Value::Null)
32    }
33}
34
35// ------------------------------------------------------------------
36// BoolDecoder: single byte 0x01 -> 1, 0x00 -> 0. Null pass-through.
37// ------------------------------------------------------------------
38
39impl<S, B> Decoder<PgBinary, S, B> for BoolDecoder {
40    fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
41        match payload.raw {
42            None => Ok(Value::Null),
43            Some(bytes) => decode_pg_bool_binary(payload.column_name, bytes),
44        }
45    }
46}
47
48// ------------------------------------------------------------------
49// IntDecoder: int2/int4/int8 as 2/4/8-byte big-endian two's complement.
50// ------------------------------------------------------------------
51
52impl<S, B> Decoder<PgBinary, S, B> for IntDecoder {
53    fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
54        match payload.raw {
55            None => Ok(Value::Null),
56            Some(bytes) => decode_pg_int_binary(payload.column_name, bytes),
57        }
58    }
59}
60
61// ------------------------------------------------------------------
62// RealDecoder: float4/float8 as 4/8-byte big-endian IEEE 754. NaN
63// normalizes to Null, -0.0 to 0.0, matching `decode_value`.
64// ------------------------------------------------------------------
65
66impl<S, B> Decoder<PgBinary, S, B> for RealDecoder {
67    fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
68        match payload.raw {
69            None => Ok(Value::Null),
70            Some(bytes) => decode_pg_real_binary(payload.column_name, bytes),
71        }
72    }
73}
74
75// ------------------------------------------------------------------
76// TextDecoder: UTF-8 bytes verbatim. Invalid UTF-8 -> InvalidUtf8.
77// ------------------------------------------------------------------
78
79impl<S, B> Decoder<PgBinary, S, B> for TextDecoder
80where
81    S: From<alloc::string::String>,
82{
83    fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
84        let Some(bytes) = payload.raw else {
85            return Ok(Value::Null);
86        };
87        match core::str::from_utf8(bytes) {
88            Ok(s) => Ok(Value::Text(S::from(s.to_string()))),
89            Err(_) => Err(DecodeError::InvalidUtf8 {
90                column: payload.column_name.to_string(),
91            }),
92        }
93    }
94}
95
96// ------------------------------------------------------------------
97// PgByteaBinaryDecoder: raw bytes verbatim into Value::Blob.
98// ------------------------------------------------------------------
99
100impl<S, B> Decoder<PgBinary, S, B> for PgByteaBinaryDecoder
101where
102    B: From<Vec<u8>>,
103{
104    fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
105        let Some(bytes) = payload.raw else {
106            return Ok(Value::Null);
107        };
108        Ok(Value::Blob(B::from(bytes.to_vec())))
109    }
110}
111
112// ------------------------------------------------------------------
113// UuidBlob16Decoder: the source bytes are already the 16 raw uuid
114// bytes, taken verbatim when the length is 16. Any other length errors.
115// The output matches the CDC path's 16-byte blob, which is the point.
116// ------------------------------------------------------------------
117
118impl<S, B> Decoder<PgBinary, S, B> for UuidBlob16Decoder
119where
120    B: From<Vec<u8>>,
121{
122    fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
123        let Some(bytes) = payload.raw else {
124            return Ok(Value::Null);
125        };
126        if bytes.len() == 16 {
127            Ok(Value::Blob(B::from(bytes.to_vec())))
128        } else {
129            Err(DecodeError::InvalidUuid {
130                column: payload.column_name.to_string(),
131                source_len: bytes.len(),
132            })
133        }
134    }
135}
136
137impl<S, B> Decoder<PgBinary, S, B> for DecimalTextDecoder
138where
139    S: From<String>,
140{
141    fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
142        match payload.raw {
143            None => Ok(Value::Null),
144            Some(bytes) => decode_pg_numeric_binary(payload.column_name, bytes),
145        }
146    }
147}
148
149impl<S, B> Decoder<PgBinary, S, B> for JsonVerbatimDecoder
150where
151    S: From<String>,
152{
153    fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
154        match payload.raw {
155            None => Ok(Value::Null),
156            Some(bytes) => decode_pg_json_binary(payload.column_name, payload.wire_type, bytes),
157        }
158    }
159}
160
161macro_rules! temporal_impl {
162    ($decoder:ty, $kind:expr) => {
163        impl<S, B> Decoder<PgBinary, S, B> for $decoder
164        where
165            S: From<String>,
166        {
167            fn decode(&self, payload: PgBinaryColumn<'_>) -> Result<Value<S, B>, DecodeError> {
168                match payload.raw {
169                    None => Ok(Value::Null),
170                    Some(bytes) => decode_temporal_binary(payload.column_name, $kind, bytes),
171                }
172            }
173        }
174    };
175}
176
177temporal_impl!(DateDecoder, Temporal::Date);
178temporal_impl!(TimeDecoder, Temporal::Time);
179temporal_impl!(TimestampDecoder, Temporal::Timestamp);
180temporal_impl!(TimestampTzDecoder, Temporal::TimestampTz);
181temporal_impl!(IntervalDecoder, Temporal::Interval);
182
183impl<S, B> TypeMapDefaults<S, B> for PgBinary
184where
185    S: From<String>,
186    B: From<Vec<u8>>,
187{
188    fn defaults() -> TypeMap<Self, S, B> {
189        TypeMap::new()
190            .with(WireType::Bool, BoolDecoder)
191            .with(WireType::Int, IntDecoder)
192            .with(WireType::Real, RealDecoder)
193            .with(WireType::Text, TextDecoder)
194            .with(WireType::Bytes, PgByteaBinaryDecoder)
195            .with(WireType::Uuid, UuidBlob16Decoder)
196            .with(WireType::Decimal, DecimalTextDecoder)
197            .with(WireType::Timestamp, TimestampDecoder)
198            .with(WireType::TimestampTz, TimestampTzDecoder)
199            .with(WireType::Date, DateDecoder)
200            .with(WireType::Time, TimeDecoder)
201            .with(WireType::Interval, IntervalDecoder)
202            .with(WireType::Json, JsonVerbatimDecoder)
203            .with(WireType::Jsonb, JsonVerbatimDecoder)
204    }
205}
sqlite_diff_rs/wire/impls_pg_binary.rs

sqlite_diff_rs/wire/
impls_pg_binary.rs
