faucet_common_spanner/
encode.rs

1//! `serde_json::Value` → Spanner mutation value encoding.
2//!
3//! The sink writes arbitrary JSON records, so values are encoded against the
4//! *destination column type* (read once from `INFORMATION_SCHEMA`). Mutations
5//! only carry `prost_types::value::Kind`s — Spanner infers the type from the
6//! table schema — so the object-safe [`ToKind`] wrapper [`EncodedKind`] is
7//! all the mutation builders need.
8
9use crate::types::SpannerType;
10use gcloud_googleapis::spanner::v1::{Type, TypeCode};
11use gcloud_spanner::statement::{ToKind, single_type};
12use prost_types::value::Kind;
13use serde_json::Value;
14
15/// A pre-encoded mutation value. `get_type` reports STRING but is never
16/// consulted on the mutation path (mutations infer types server-side from
17/// the table schema); this wrapper must not be used for statement params.
18pub struct EncodedKind(pub Kind);
19
20impl ToKind for EncodedKind {
21    fn to_kind(&self) -> Kind {
22        self.0.clone()
23    }
24    fn get_type() -> Type
25    where
26        Self: Sized,
27    {
28        single_type(TypeCode::String)
29    }
30}
31
32/// Encode one JSON value for a destination column of type `ty`.
33///
34/// Returns a human-readable error naming the mismatch; callers wrap it with
35/// the column name and route the row to the DLQ or fail the batch.
36pub fn encode_to_kind(value: &Value, ty: &SpannerType) -> Result<Kind, String> {
37    if value.is_null() {
38        return Ok(Kind::NullValue(0));
39    }
40    match ty {
41        SpannerType::Bool => match value {
42            Value::Bool(b) => Ok(Kind::BoolValue(*b)),
43            other => Err(mismatch("BOOL", other)),
44        },
45        // INT64 travels string-encoded (f64 would corrupt > 2^53).
46        SpannerType::Int64 => match value {
47            Value::Number(n) => {
48                if let Some(i) = n.as_i64() {
49                    Ok(Kind::StringValue(i.to_string()))
50                } else if let Some(u) = n.as_u64() {
51                    // > i64::MAX cannot be stored in INT64.
52                    i64::try_from(u)
53                        .map(|i| Kind::StringValue(i.to_string()))
54                        .map_err(|_| format!("integer {u} overflows INT64"))
55                } else {
56                    Err(mismatch("INT64", value))
57                }
58            }
59            // Allow pre-stringified integers (common after transforms).
60            Value::String(s) if s.parse::<i64>().is_ok() => Ok(Kind::StringValue(s.clone())),
61            other => Err(mismatch("INT64", other)),
62        },
63        SpannerType::Float32 | SpannerType::Float64 => match value {
64            Value::Number(n) => n
65                .as_f64()
66                .map(Kind::NumberValue)
67                .ok_or_else(|| mismatch("FLOAT64", value)),
68            // NaN / Infinity travel as strings.
69            Value::String(s) if matches!(s.as_str(), "NaN" | "Infinity" | "-Infinity") => {
70                Ok(Kind::StringValue(s.clone()))
71            }
72            other => Err(mismatch("FLOAT64", other)),
73        },
74        SpannerType::String => match value {
75            Value::String(s) => Ok(Kind::StringValue(s.clone())),
76            // Scalars coerce to their canonical text; containers serialize as
77            // JSON text so a bad mapping surfaces visibly, not silently.
78            Value::Number(n) => Ok(Kind::StringValue(n.to_string())),
79            Value::Bool(b) => Ok(Kind::StringValue(b.to_string())),
80            other => Ok(Kind::StringValue(other.to_string())),
81        },
82        // Timestamps/dates must already be RFC 3339 text; bytes must already
83        // be base64 (the same form the source decodes them to).
84        SpannerType::Timestamp | SpannerType::Date | SpannerType::Bytes => match value {
85            Value::String(s) => Ok(Kind::StringValue(s.clone())),
86            other => Err(mismatch(type_name(ty), other)),
87        },
88        SpannerType::Numeric => match value {
89            Value::String(s) => Ok(Kind::StringValue(s.clone())),
90            Value::Number(n) => Ok(Kind::StringValue(n.to_string())),
91            other => Err(mismatch("NUMERIC", other)),
92        },
93        // JSON columns take the value's RFC 7159 serialization verbatim.
94        SpannerType::Json => Ok(Kind::StringValue(value.to_string())),
95        SpannerType::Array(inner) => match value {
96            Value::Array(items) => {
97                let values: Result<Vec<prost_types::Value>, String> = items
98                    .iter()
99                    .map(|item| {
100                        encode_to_kind(item, inner)
101                            .map(|kind| prost_types::Value { kind: Some(kind) })
102                    })
103                    .collect();
104                Ok(Kind::ListValue(prost_types::ListValue { values: values? }))
105            }
106            other => Err(mismatch("ARRAY", other)),
107        },
108        SpannerType::Other => Err(
109            "unsupported destination column type (STRUCT/PROTO columns are not writable)".into(),
110        ),
111    }
112}
113
114fn type_name(ty: &SpannerType) -> &'static str {
115    match ty {
116        SpannerType::Timestamp => "TIMESTAMP",
117        SpannerType::Date => "DATE",
118        SpannerType::Bytes => "BYTES",
119        _ => "value",
120    }
121}
122
123fn mismatch(expected: &str, got: &Value) -> String {
124    let got_ty = match got {
125        Value::Null => "null",
126        Value::Bool(_) => "boolean",
127        Value::Number(_) => "number",
128        Value::String(_) => "string",
129        Value::Array(_) => "array",
130        Value::Object(_) => "object",
131    };
132    format!("expected {expected}-compatible value, got {got_ty}")
133}
134
135#[cfg(test)]
136mod tests {
137    use super::*;
138    use serde_json::json;
139
140    #[test]
141    fn null_encodes_as_null_for_every_type() {
142        for ty in [
143            SpannerType::Bool,
144            SpannerType::Int64,
145            SpannerType::Json,
146            SpannerType::Array(Box::new(SpannerType::String)),
147        ] {
148            assert!(matches!(
149                encode_to_kind(&Value::Null, &ty).unwrap(),
150                Kind::NullValue(_)
151            ));
152        }
153    }
154
155    #[test]
156    fn int64_encodes_as_string_losslessly() {
157        let big = 9_007_199_254_740_993_i64;
158        assert_eq!(
159            encode_to_kind(&json!(big), &SpannerType::Int64).unwrap(),
160            Kind::StringValue(big.to_string())
161        );
162        assert_eq!(
163            encode_to_kind(&json!("42"), &SpannerType::Int64).unwrap(),
164            Kind::StringValue("42".into())
165        );
166        assert!(encode_to_kind(&json!(1.5), &SpannerType::Int64).is_err());
167        assert!(encode_to_kind(&json!("abc"), &SpannerType::Int64).is_err());
168        assert!(
169            encode_to_kind(&json!(u64::MAX), &SpannerType::Int64)
170                .unwrap_err()
171                .contains("overflows INT64")
172        );
173    }
174
175    #[test]
176    fn floats_bools_and_special_values() {
177        assert_eq!(
178            encode_to_kind(&json!(2.5), &SpannerType::Float64).unwrap(),
179            Kind::NumberValue(2.5)
180        );
181        assert_eq!(
182            encode_to_kind(&json!("NaN"), &SpannerType::Float64).unwrap(),
183            Kind::StringValue("NaN".into())
184        );
185        assert!(encode_to_kind(&json!("2.5"), &SpannerType::Float64).is_err());
186        assert_eq!(
187            encode_to_kind(&json!(true), &SpannerType::Bool).unwrap(),
188            Kind::BoolValue(true)
189        );
190        assert!(encode_to_kind(&json!(1), &SpannerType::Bool).is_err());
191    }
192
193    #[test]
194    fn strings_coerce_scalars_and_serialize_containers() {
195        assert_eq!(
196            encode_to_kind(&json!("x"), &SpannerType::String).unwrap(),
197            Kind::StringValue("x".into())
198        );
199        assert_eq!(
200            encode_to_kind(&json!(7), &SpannerType::String).unwrap(),
201            Kind::StringValue("7".into())
202        );
203        assert_eq!(
204            encode_to_kind(&json!({"a": 1}), &SpannerType::String).unwrap(),
205            Kind::StringValue("{\"a\":1}".into())
206        );
207    }
208
209    #[test]
210    fn timestamp_date_bytes_require_strings() {
211        assert_eq!(
212            encode_to_kind(&json!("2026-01-01T00:00:00Z"), &SpannerType::Timestamp).unwrap(),
213            Kind::StringValue("2026-01-01T00:00:00Z".into())
214        );
215        assert!(encode_to_kind(&json!(5), &SpannerType::Timestamp).is_err());
216        assert!(encode_to_kind(&json!(5), &SpannerType::Date).is_err());
217        assert!(encode_to_kind(&json!(5), &SpannerType::Bytes).is_err());
218    }
219
220    #[test]
221    fn numeric_accepts_strings_and_numbers() {
222        assert_eq!(
223            encode_to_kind(&json!("123.456789012345"), &SpannerType::Numeric).unwrap(),
224            Kind::StringValue("123.456789012345".into())
225        );
226        assert_eq!(
227            encode_to_kind(&json!(12), &SpannerType::Numeric).unwrap(),
228            Kind::StringValue("12".into())
229        );
230        assert!(encode_to_kind(&json!(true), &SpannerType::Numeric).is_err());
231    }
232
233    #[test]
234    fn json_columns_take_any_value() {
235        assert_eq!(
236            encode_to_kind(&json!({"a": [1]}), &SpannerType::Json).unwrap(),
237            Kind::StringValue("{\"a\":[1]}".into())
238        );
239        // A JSON string value serializes as a quoted JSON string.
240        assert_eq!(
241            encode_to_kind(&json!("s"), &SpannerType::Json).unwrap(),
242            Kind::StringValue("\"s\"".into())
243        );
244    }
245
246    #[test]
247    fn arrays_recurse_and_reject_non_arrays() {
248        let ty = SpannerType::Array(Box::new(SpannerType::Int64));
249        let Kind::ListValue(list) = encode_to_kind(&json!([1, null, 3]), &ty).unwrap() else {
250            panic!("expected list");
251        };
252        assert_eq!(list.values.len(), 3);
253        assert!(encode_to_kind(&json!(1), &ty).is_err());
254        // Element mismatch surfaces.
255        assert!(encode_to_kind(&json!(["x"]), &ty).is_err());
256    }
257
258    #[test]
259    fn unsupported_destination_types_error() {
260        assert!(encode_to_kind(&json!(1), &SpannerType::Other).is_err());
261    }
262
263    #[test]
264    fn encoded_kind_is_a_usable_tokind() {
265        let ek = EncodedKind(Kind::StringValue("v".into()));
266        assert_eq!(ek.to_kind(), Kind::StringValue("v".into()));
267        let dyn_ref: &dyn ToKind = &ek;
268        assert_eq!(dyn_ref.to_kind(), Kind::StringValue("v".into()));
269    }
270}
faucet_common_spanner/encode.rs

faucet_common_spanner/
encode.rs
