faucet_common_spanner/
decode.rs

1//! Generic Spanner row → `serde_json::Value` decoding.
2//!
3//! faucet sources run arbitrary user SQL, so rows must decode without a
4//! compile-time schema. [`SpannerJson`] implements the client's
5//! `TryFromValue` hook, which hands us the raw protobuf value *plus* the
6//! column's full [`Type`] metadata — enough to decode every Spanner type
7//! into faithful JSON:
8//!
9//! | Spanner | JSON |
10//! |---|---|
11//! | INT64 / ENUM | integer (lossless — arrives string-encoded) |
12//! | FLOAT64 / FLOAT32 | number (`NaN`/`Infinity` → string) |
13//! | BOOL | boolean |
14//! | STRING / TIMESTAMP / DATE / UUID / INTERVAL | string |
15//! | BYTES / PROTO | base64 string (as transmitted) |
16//! | NUMERIC | string (precision preserved) |
17//! | JSON | parsed object/array/scalar |
18//! | ARRAY | array (element type recursed) |
19//! | STRUCT | object keyed by field name |
20
21use gcloud_googleapis::spanner::v1::struct_type::Field;
22use gcloud_googleapis::spanner::v1::{Type, TypeCode};
23use gcloud_spanner::row::{Error as RowError, Row, TryFromValue};
24use prost_types::value::Kind;
25use serde_json::Value;
26
27/// Newtype decoding one Spanner column into a [`serde_json::Value`].
28pub struct SpannerJson(pub Value);
29
30impl TryFromValue for SpannerJson {
31    fn try_from(value: &prost_types::Value, field: &Field) -> Result<Self, RowError> {
32        decode_value(value, field.r#type.as_ref(), &field.name).map(SpannerJson)
33    }
34}
35
36/// Decode a whole [`Row`] into a JSON object keyed by column name, using the
37/// stream's column metadata (`RowIterator::columns_metadata`).
38pub fn row_to_json(row: &Row, fields: &[Field]) -> Result<Value, RowError> {
39    let mut obj = serde_json::Map::with_capacity(fields.len());
40    for (idx, field) in fields.iter().enumerate() {
41        let SpannerJson(v) = row.column::<SpannerJson>(idx)?;
42        obj.insert(field.name.clone(), v);
43    }
44    Ok(Value::Object(obj))
45}
46
47fn type_code(ty: Option<&Type>) -> TypeCode {
48    ty.and_then(|t| TypeCode::try_from(t.code).ok())
49        .unwrap_or(TypeCode::Unspecified)
50}
51
52/// Whether the named column in a result's metadata is Spanner `NUMERIC`.
53///
54/// The source uses this to reject an incremental-replication cursor on a NUMERIC
55/// column: NUMERIC decodes to a JSON *string* (to preserve precision), so the
56/// generic replication comparison orders it lexicographically (`"9" > "10"`),
57/// which would advance the bookmark incorrectly and skip or re-read rows. INT64
58/// (JSON number) and TIMESTAMP/DATE (RFC-3339 string, lexicographic == chrono)
59/// cursors are unaffected. Checked once the first page's metadata is known.
60pub fn column_is_numeric(fields: &[Field], name: &str) -> bool {
61    fields
62        .iter()
63        .find(|f| f.name == name)
64        .map(|f| type_code(f.r#type.as_ref()) == TypeCode::Numeric)
65        .unwrap_or(false)
66}
67
68/// Decode one protobuf value with its (possibly absent) Spanner type.
69pub fn decode_value(
70    value: &prost_types::Value,
71    ty: Option<&Type>,
72    field_name: &str,
73) -> Result<Value, RowError> {
74    let kind = match &value.kind {
75        None | Some(Kind::NullValue(_)) => return Ok(Value::Null),
76        Some(kind) => kind,
77    };
78    let code = type_code(ty);
79    match (code, kind) {
80        // INT64 (and ENUM numbers) arrive string-encoded to survive f64
81        // precision loss; keep them lossless as JSON integers.
82        (TypeCode::Int64 | TypeCode::Enum, Kind::StringValue(s)) => {
83            let n: i64 = s.parse().map_err(|_| {
84                RowError::CustomParseError(format!("{field_name}: non-integer INT64 `{s}`"))
85            })?;
86            Ok(Value::Number(n.into()))
87        }
88        (TypeCode::Float64 | TypeCode::Float32, Kind::NumberValue(f)) => {
89            Ok(serde_json::Number::from_f64(*f).map_or(Value::Null, Value::Number))
90        }
91        // FLOAT64 NaN / Infinity / -Infinity arrive as strings; JSON has no
92        // representation for them, so keep the string form.
93        (TypeCode::Float64 | TypeCode::Float32, Kind::StringValue(s)) => {
94            Ok(Value::String(s.clone()))
95        }
96        (TypeCode::Bool, Kind::BoolValue(b)) => Ok(Value::Bool(*b)),
97        // JSON columns carry their RFC 7159 text in a string value.
98        (TypeCode::Json, Kind::StringValue(s)) => serde_json::from_str(s).map_err(|e| {
99            RowError::CustomParseError(format!("{field_name}: invalid JSON column value: {e}"))
100        }),
101        // Strings and every string-encoded scalar (timestamps, dates,
102        // NUMERIC, base64 bytes, UUID, INTERVAL, PROTO) pass through.
103        (_, Kind::StringValue(s)) => Ok(Value::String(s.clone())),
104        (TypeCode::Array, Kind::ListValue(list)) => {
105            let elem_ty = ty.and_then(|t| t.array_element_type.as_deref());
106            let items: Result<Vec<Value>, RowError> = list
107                .values
108                .iter()
109                .map(|v| decode_value(v, elem_ty, field_name))
110                .collect();
111            Ok(Value::Array(items?))
112        }
113        // STRUCT rows are encoded as a list whose i-th element matches
114        // `struct_type.fields[i]`.
115        (TypeCode::Struct, Kind::ListValue(list)) => {
116            let fields = ty
117                .and_then(|t| t.struct_type.as_ref())
118                .map(|st| st.fields.as_slice())
119                .unwrap_or(&[]);
120            let mut obj = serde_json::Map::with_capacity(list.values.len());
121            for (idx, v) in list.values.iter().enumerate() {
122                let (name, elem_ty) = fields
123                    .get(idx)
124                    .map(|f| (f.name.clone(), f.r#type.as_ref()))
125                    .unwrap_or_else(|| (format!("_{idx}"), None));
126                obj.insert(name, decode_value(v, elem_ty, field_name)?);
127            }
128            Ok(Value::Object(obj))
129        }
130        // Untyped / unexpected pairings: decode the raw protobuf value
131        // structurally so nothing is silently dropped.
132        (_, kind) => Ok(kind_to_json(kind)),
133    }
134}
135
136/// Structural protobuf → JSON fallback for values whose Spanner type is
137/// absent or does not match the wire kind. Infallible: every protobuf value
138/// kind has a structural JSON form.
139fn kind_to_json(kind: &Kind) -> Value {
140    match kind {
141        Kind::NullValue(_) => Value::Null,
142        Kind::BoolValue(b) => Value::Bool(*b),
143        Kind::NumberValue(f) => serde_json::Number::from_f64(*f).map_or(Value::Null, Value::Number),
144        Kind::StringValue(s) => Value::String(s.clone()),
145        Kind::ListValue(list) => Value::Array(
146            list.values
147                .iter()
148                .map(|v| match &v.kind {
149                    None => Value::Null,
150                    Some(k) => kind_to_json(k),
151                })
152                .collect(),
153        ),
154        Kind::StructValue(st) => {
155            let mut obj = serde_json::Map::with_capacity(st.fields.len());
156            for (k, v) in &st.fields {
157                let decoded = match &v.kind {
158                    None => Value::Null,
159                    Some(kind) => kind_to_json(kind),
160                };
161                obj.insert(k.clone(), decoded);
162            }
163            Value::Object(obj)
164        }
165    }
166}
167
168#[cfg(test)]
169mod tests {
170    use super::*;
171    use gcloud_googleapis::spanner::v1::StructType;
172    use serde_json::json;
173
174    fn pv(kind: Kind) -> prost_types::Value {
175        prost_types::Value { kind: Some(kind) }
176    }
177
178    fn ty(code: TypeCode) -> Type {
179        Type {
180            code: code.into(),
181            array_element_type: None,
182            struct_type: None,
183            ..Default::default()
184        }
185    }
186
187    #[test]
188    fn column_is_numeric_only_flags_numeric_columns() {
189        let field = |name: &str, code: TypeCode| Field {
190            name: name.to_string(),
191            r#type: Some(ty(code)),
192        };
193        let fields = vec![
194            field("id", TypeCode::Int64),
195            field("amount", TypeCode::Numeric),
196            field("updated_at", TypeCode::Timestamp),
197        ];
198        // Only the NUMERIC column is flagged (the #466 L3 cursor guard).
199        assert!(column_is_numeric(&fields, "amount"));
200        assert!(!column_is_numeric(&fields, "id"));
201        assert!(!column_is_numeric(&fields, "updated_at"));
202        // An unknown column is not numeric (and must not panic).
203        assert!(!column_is_numeric(&fields, "nope"));
204    }
205
206    #[test]
207    fn int64_decodes_losslessly_from_string() {
208        let big = 9_007_199_254_740_993_i64; // 2^53 + 1 — breaks under f64
209        let v = decode_value(
210            &pv(Kind::StringValue(big.to_string())),
211            Some(&ty(TypeCode::Int64)),
212            "n",
213        )
214        .unwrap();
215        assert_eq!(v, json!(big));
216    }
217
218    #[test]
219    fn malformed_int64_is_a_typed_error() {
220        let err = decode_value(
221            &pv(Kind::StringValue("abc".into())),
222            Some(&ty(TypeCode::Int64)),
223            "n",
224        )
225        .unwrap_err();
226        assert!(err.to_string().contains("non-integer INT64"));
227    }
228
229    #[test]
230    fn floats_bools_strings_and_nulls() {
231        assert_eq!(
232            decode_value(
233                &pv(Kind::NumberValue(1.5)),
234                Some(&ty(TypeCode::Float64)),
235                "f"
236            )
237            .unwrap(),
238            json!(1.5)
239        );
240        assert_eq!(
241            decode_value(
242                &pv(Kind::StringValue("NaN".into())),
243                Some(&ty(TypeCode::Float64)),
244                "f"
245            )
246            .unwrap(),
247            json!("NaN")
248        );
249        assert_eq!(
250            decode_value(&pv(Kind::BoolValue(true)), Some(&ty(TypeCode::Bool)), "b").unwrap(),
251            json!(true)
252        );
253        assert_eq!(
254            decode_value(
255                &pv(Kind::StringValue("hi".into())),
256                Some(&ty(TypeCode::String)),
257                "s"
258            )
259            .unwrap(),
260            json!("hi")
261        );
262        assert_eq!(
263            decode_value(&pv(Kind::NullValue(0)), Some(&ty(TypeCode::String)), "s").unwrap(),
264            Value::Null
265        );
266        assert_eq!(
267            decode_value(&prost_types::Value { kind: None }, None, "x").unwrap(),
268            Value::Null
269        );
270    }
271
272    #[test]
273    fn numeric_timestamp_date_bytes_stay_strings() {
274        for code in [
275            TypeCode::Numeric,
276            TypeCode::Timestamp,
277            TypeCode::Date,
278            TypeCode::Bytes,
279        ] {
280            let v =
281                decode_value(&pv(Kind::StringValue("raw".into())), Some(&ty(code)), "c").unwrap();
282            assert_eq!(v, json!("raw"));
283        }
284    }
285
286    #[test]
287    fn json_columns_parse_to_structured_values() {
288        let v = decode_value(
289            &pv(Kind::StringValue(r#"{"a": [1, 2]}"#.into())),
290            Some(&ty(TypeCode::Json)),
291            "j",
292        )
293        .unwrap();
294        assert_eq!(v, json!({"a": [1, 2]}));
295        let err = decode_value(
296            &pv(Kind::StringValue("{not json".into())),
297            Some(&ty(TypeCode::Json)),
298            "j",
299        )
300        .unwrap_err();
301        assert!(err.to_string().contains("invalid JSON column value"));
302    }
303
304    #[test]
305    fn arrays_recurse_with_element_type() {
306        let mut arr_ty = ty(TypeCode::Array);
307        arr_ty.array_element_type = Some(Box::new(ty(TypeCode::Int64)));
308        let v = decode_value(
309            &pv(Kind::ListValue(prost_types::ListValue {
310                values: vec![pv(Kind::StringValue("1".into())), pv(Kind::NullValue(0))],
311            })),
312            Some(&arr_ty),
313            "a",
314        )
315        .unwrap();
316        assert_eq!(v, json!([1, null]));
317    }
318
319    #[test]
320    fn structs_become_objects_keyed_by_field_name() {
321        let mut st_ty = ty(TypeCode::Struct);
322        st_ty.struct_type = Some(StructType {
323            fields: vec![
324                Field {
325                    name: "id".into(),
326                    r#type: Some(ty(TypeCode::Int64)),
327                },
328                Field {
329                    name: "name".into(),
330                    r#type: Some(ty(TypeCode::String)),
331                },
332            ],
333        });
334        let v = decode_value(
335            &pv(Kind::ListValue(prost_types::ListValue {
336                values: vec![
337                    pv(Kind::StringValue("7".into())),
338                    pv(Kind::StringValue("x".into())),
339                ],
340            })),
341            Some(&st_ty),
342            "s",
343        )
344        .unwrap();
345        assert_eq!(v, json!({"id": 7, "name": "x"}));
346    }
347
348    #[test]
349    fn untyped_values_fall_back_to_structural_decode() {
350        let v = decode_value(&pv(Kind::NumberValue(2.0)), None, "u").unwrap();
351        assert_eq!(v, json!(2.0));
352        let v = decode_value(
353            &pv(Kind::ListValue(prost_types::ListValue {
354                values: vec![pv(Kind::BoolValue(false))],
355            })),
356            None,
357            "u",
358        )
359        .unwrap();
360        assert_eq!(v, json!([false]));
361    }
362}
faucet_common_spanner/decode.rs

faucet_common_spanner/
decode.rs
