sqlite_diff_rs/wire/
source.rs

1//! [`WireSource`]: sealed marker trait for CDC wire formats.
2
3use super::sealed::Sealed;
4use super::wire_type::WireType;
5
6/// Per-format marker naming a CDC wire source.
7///
8/// Implementors are unit structs owned by each format module
9/// (`PgWalstream`, `Wal2Json`, `Maxwell`). The associated payload type
10/// describes the format's per-column wire data.
11///
12/// Type identity is no longer source-native. Every payload carries a
13/// source-independent [`WireType`] that selects the decoder, so one
14/// semantic catalog drives every source without a per-source
15/// translation table.
16pub trait WireSource: Sealed {
17    /// Per-column payload the format hands to a decoder.
18    ///
19    /// Every payload struct carries a `column_name: &'a str` so decoder
20    /// errors are self-describing without an outer wrapping layer.
21    type Payload<'a>;
22
23    /// Semantic type of the column carried by the payload, used for
24    /// decoder dispatch.
25    fn wire_type(payload: &Self::Payload<'_>) -> WireType;
26
27    /// Extract the column name from a payload for diagnostic messages.
28    fn column_name<'a>(payload: &'a Self::Payload<'_>) -> &'a str;
29}
30
31/// Schema-side semantic type for one column of one table.
32pub trait WireColumnTypes {
33    /// Semantic [`WireType`] for the column at `column_index`.
34    fn column_type(&self, column_index: usize) -> WireType;
35}
36
37/// Table-name lookup for the [`DiffSetBuilder::digest`](crate::DiffSetBuilder::digest) entry point.
38pub trait WireSchema {
39    /// Concrete schema type for one table.
40    type Table: crate::schema::NamedColumns + WireColumnTypes;
41
42    /// Resolve a table name to its schema entry.
43    fn get(&self, table_name: &str) -> Option<&Self::Table>;
44}
45
46/// One CDC wire event digested via [`DiffSetBuilder::digest`](crate::DiffSetBuilder::digest).
47///
48/// Implemented in-crate for `pg_walstream::EventType`, `wal2json::MessageV2`,
49/// `wal2json::ChangeV1`, and `maxwell::Message` (each times both formats).
50pub trait Digestable<F, T, S, B>
51where
52    F: crate::builders::Format<S, B>,
53    T: crate::schema::NamedColumns + WireColumnTypes,
54{
55    /// Wire source this event came from.
56    type Src: WireSource;
57
58    /// Failure mode raised on schema lookup or decode failure.
59    type Error;
60
61    /// Fold this event into `builder`, resolving affected tables via `schema`
62    /// and decoding column payloads via `adapter`.
63    ///
64    /// # Errors
65    ///
66    /// Any per-source `ConversionError`.
67    fn digest_into<Sch, A>(
68        &self,
69        builder: crate::builders::DiffSetBuilder<F, T, S, B>,
70        schema: &Sch,
71        adapter: &A,
72    ) -> Result<crate::builders::DiffSetBuilder<F, T, S, B>, Self::Error>
73    where
74        Sch: WireSchema<Table = T>,
75        A: super::WireAdapter<Self::Src, S, B>;
76}
77
78/// A value read back from an executed Postgres query in binary result
79/// format, tagged with the catalog's semantic type for the column.
80///
81/// Unlike the CDC sources, the semantic type is not recoverable from the
82/// raw bytes (binary `int4` and `float4` are both four opaque bytes), so
83/// the payload carries the [`WireType`] explicitly, supplied by the
84/// caller from its catalog. This makes decoder dispatch deterministic and
85/// makes the caller choose the type the same way CDC does, which is what
86/// guarantees representation parity with the CDC paths.
87#[derive(Debug, Clone, Copy, Default)]
88pub struct PgBinary;
89
90impl Sealed for PgBinary {}
91
92impl WireSource for PgBinary {
93    type Payload<'a> = PgBinaryColumn<'a>;
94
95    fn wire_type(payload: &Self::Payload<'_>) -> WireType {
96        payload.wire_type
97    }
98
99    fn column_name<'a>(payload: &'a Self::Payload<'_>) -> &'a str {
100        payload.column_name
101    }
102}
103
104/// One binary result field for the [`PgBinary`] source.
105///
106/// `raw` is `None` for a SQL NULL. `wire_type` is the caller's catalog
107/// type for the column, not inferred from the bytes.
108#[derive(Debug, Clone, Copy)]
109pub struct PgBinaryColumn<'a> {
110    /// Column name, carried for self-describing decoder errors.
111    pub column_name: &'a str,
112    /// Semantic column type driving decoder dispatch.
113    pub wire_type: WireType,
114    /// Raw binary result bytes, or `None` for SQL NULL.
115    pub raw: Option<&'a [u8]>,
116}
117
118impl PgBinaryColumn<'_> {
119    /// Ergonomic helper for calling a specific [`Decoder`](super::Decoder)
120    /// on this payload without fully-qualified syntax. Fixes the `Src`
121    /// generic to [`PgBinary`] so the compiler can pick the impl.
122    ///
123    /// # Errors
124    ///
125    /// Propagates the decoder's [`DecodeError`](super::DecodeError).
126    pub fn decoded_by<D, S, B>(
127        self,
128        decoder: &D,
129    ) -> Result<crate::encoding::Value<S, B>, super::error::DecodeError>
130    where
131        D: super::decoder::Decoder<PgBinary, S, B>,
132    {
133        decoder.decode(self)
134    }
135}
sqlite_diff_rs/wire/source.rs

sqlite_diff_rs/wire/
source.rs
