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/// Decode one protobuf value with its (possibly absent) Spanner type.
53pub fn decode_value(
54    value: &prost_types::Value,
55    ty: Option<&Type>,
56    field_name: &str,
57) -> Result<Value, RowError> {
58    let kind = match &value.kind {
59        None | Some(Kind::NullValue(_)) => return Ok(Value::Null),
60        Some(kind) => kind,
61    };
62    let code = type_code(ty);
63    match (code, kind) {
64        // INT64 (and ENUM numbers) arrive string-encoded to survive f64
65        // precision loss; keep them lossless as JSON integers.
66        (TypeCode::Int64 | TypeCode::Enum, Kind::StringValue(s)) => {
67            let n: i64 = s.parse().map_err(|_| {
68                RowError::CustomParseError(format!("{field_name}: non-integer INT64 `{s}`"))
69            })?;
70            Ok(Value::Number(n.into()))
71        }
72        (TypeCode::Float64 | TypeCode::Float32, Kind::NumberValue(f)) => {
73            Ok(serde_json::Number::from_f64(*f).map_or(Value::Null, Value::Number))
74        }
75        // FLOAT64 NaN / Infinity / -Infinity arrive as strings; JSON has no
76        // representation for them, so keep the string form.
77        (TypeCode::Float64 | TypeCode::Float32, Kind::StringValue(s)) => {
78            Ok(Value::String(s.clone()))
79        }
80        (TypeCode::Bool, Kind::BoolValue(b)) => Ok(Value::Bool(*b)),
81        // JSON columns carry their RFC 7159 text in a string value.
82        (TypeCode::Json, Kind::StringValue(s)) => serde_json::from_str(s).map_err(|e| {
83            RowError::CustomParseError(format!("{field_name}: invalid JSON column value: {e}"))
84        }),
85        // Strings and every string-encoded scalar (timestamps, dates,
86        // NUMERIC, base64 bytes, UUID, INTERVAL, PROTO) pass through.
87        (_, Kind::StringValue(s)) => Ok(Value::String(s.clone())),
88        (TypeCode::Array, Kind::ListValue(list)) => {
89            let elem_ty = ty.and_then(|t| t.array_element_type.as_deref());
90            let items: Result<Vec<Value>, RowError> = list
91                .values
92                .iter()
93                .map(|v| decode_value(v, elem_ty, field_name))
94                .collect();
95            Ok(Value::Array(items?))
96        }
97        // STRUCT rows are encoded as a list whose i-th element matches
98        // `struct_type.fields[i]`.
99        (TypeCode::Struct, Kind::ListValue(list)) => {
100            let fields = ty
101                .and_then(|t| t.struct_type.as_ref())
102                .map(|st| st.fields.as_slice())
103                .unwrap_or(&[]);
104            let mut obj = serde_json::Map::with_capacity(list.values.len());
105            for (idx, v) in list.values.iter().enumerate() {
106                let (name, elem_ty) = fields
107                    .get(idx)
108                    .map(|f| (f.name.clone(), f.r#type.as_ref()))
109                    .unwrap_or_else(|| (format!("_{idx}"), None));
110                obj.insert(name, decode_value(v, elem_ty, field_name)?);
111            }
112            Ok(Value::Object(obj))
113        }
114        // Untyped / unexpected pairings: decode the raw protobuf value
115        // structurally so nothing is silently dropped.
116        (_, kind) => Ok(kind_to_json(kind)),
117    }
118}
119
120/// Structural protobuf → JSON fallback for values whose Spanner type is
121/// absent or does not match the wire kind. Infallible: every protobuf value
122/// kind has a structural JSON form.
123fn kind_to_json(kind: &Kind) -> Value {
124    match kind {
125        Kind::NullValue(_) => Value::Null,
126        Kind::BoolValue(b) => Value::Bool(*b),
127        Kind::NumberValue(f) => serde_json::Number::from_f64(*f).map_or(Value::Null, Value::Number),
128        Kind::StringValue(s) => Value::String(s.clone()),
129        Kind::ListValue(list) => Value::Array(
130            list.values
131                .iter()
132                .map(|v| match &v.kind {
133                    None => Value::Null,
134                    Some(k) => kind_to_json(k),
135                })
136                .collect(),
137        ),
138        Kind::StructValue(st) => {
139            let mut obj = serde_json::Map::with_capacity(st.fields.len());
140            for (k, v) in &st.fields {
141                let decoded = match &v.kind {
142                    None => Value::Null,
143                    Some(kind) => kind_to_json(kind),
144                };
145                obj.insert(k.clone(), decoded);
146            }
147            Value::Object(obj)
148        }
149    }
150}
151
152#[cfg(test)]
153mod tests {
154    use super::*;
155    use gcloud_googleapis::spanner::v1::StructType;
156    use serde_json::json;
157
158    fn pv(kind: Kind) -> prost_types::Value {
159        prost_types::Value { kind: Some(kind) }
160    }
161
162    fn ty(code: TypeCode) -> Type {
163        Type {
164            code: code.into(),
165            array_element_type: None,
166            struct_type: None,
167            ..Default::default()
168        }
169    }
170
171    #[test]
172    fn int64_decodes_losslessly_from_string() {
173        let big = 9_007_199_254_740_993_i64; // 2^53 + 1 — breaks under f64
174        let v = decode_value(
175            &pv(Kind::StringValue(big.to_string())),
176            Some(&ty(TypeCode::Int64)),
177            "n",
178        )
179        .unwrap();
180        assert_eq!(v, json!(big));
181    }
182
183    #[test]
184    fn malformed_int64_is_a_typed_error() {
185        let err = decode_value(
186            &pv(Kind::StringValue("abc".into())),
187            Some(&ty(TypeCode::Int64)),
188            "n",
189        )
190        .unwrap_err();
191        assert!(err.to_string().contains("non-integer INT64"));
192    }
193
194    #[test]
195    fn floats_bools_strings_and_nulls() {
196        assert_eq!(
197            decode_value(
198                &pv(Kind::NumberValue(1.5)),
199                Some(&ty(TypeCode::Float64)),
200                "f"
201            )
202            .unwrap(),
203            json!(1.5)
204        );
205        assert_eq!(
206            decode_value(
207                &pv(Kind::StringValue("NaN".into())),
208                Some(&ty(TypeCode::Float64)),
209                "f"
210            )
211            .unwrap(),
212            json!("NaN")
213        );
214        assert_eq!(
215            decode_value(&pv(Kind::BoolValue(true)), Some(&ty(TypeCode::Bool)), "b").unwrap(),
216            json!(true)
217        );
218        assert_eq!(
219            decode_value(
220                &pv(Kind::StringValue("hi".into())),
221                Some(&ty(TypeCode::String)),
222                "s"
223            )
224            .unwrap(),
225            json!("hi")
226        );
227        assert_eq!(
228            decode_value(&pv(Kind::NullValue(0)), Some(&ty(TypeCode::String)), "s").unwrap(),
229            Value::Null
230        );
231        assert_eq!(
232            decode_value(&prost_types::Value { kind: None }, None, "x").unwrap(),
233            Value::Null
234        );
235    }
236
237    #[test]
238    fn numeric_timestamp_date_bytes_stay_strings() {
239        for code in [
240            TypeCode::Numeric,
241            TypeCode::Timestamp,
242            TypeCode::Date,
243            TypeCode::Bytes,
244        ] {
245            let v =
246                decode_value(&pv(Kind::StringValue("raw".into())), Some(&ty(code)), "c").unwrap();
247            assert_eq!(v, json!("raw"));
248        }
249    }
250
251    #[test]
252    fn json_columns_parse_to_structured_values() {
253        let v = decode_value(
254            &pv(Kind::StringValue(r#"{"a": [1, 2]}"#.into())),
255            Some(&ty(TypeCode::Json)),
256            "j",
257        )
258        .unwrap();
259        assert_eq!(v, json!({"a": [1, 2]}));
260        let err = decode_value(
261            &pv(Kind::StringValue("{not json".into())),
262            Some(&ty(TypeCode::Json)),
263            "j",
264        )
265        .unwrap_err();
266        assert!(err.to_string().contains("invalid JSON column value"));
267    }
268
269    #[test]
270    fn arrays_recurse_with_element_type() {
271        let mut arr_ty = ty(TypeCode::Array);
272        arr_ty.array_element_type = Some(Box::new(ty(TypeCode::Int64)));
273        let v = decode_value(
274            &pv(Kind::ListValue(prost_types::ListValue {
275                values: vec![pv(Kind::StringValue("1".into())), pv(Kind::NullValue(0))],
276            })),
277            Some(&arr_ty),
278            "a",
279        )
280        .unwrap();
281        assert_eq!(v, json!([1, null]));
282    }
283
284    #[test]
285    fn structs_become_objects_keyed_by_field_name() {
286        let mut st_ty = ty(TypeCode::Struct);
287        st_ty.struct_type = Some(StructType {
288            fields: vec![
289                Field {
290                    name: "id".into(),
291                    r#type: Some(ty(TypeCode::Int64)),
292                },
293                Field {
294                    name: "name".into(),
295                    r#type: Some(ty(TypeCode::String)),
296                },
297            ],
298        });
299        let v = decode_value(
300            &pv(Kind::ListValue(prost_types::ListValue {
301                values: vec![
302                    pv(Kind::StringValue("7".into())),
303                    pv(Kind::StringValue("x".into())),
304                ],
305            })),
306            Some(&st_ty),
307            "s",
308        )
309        .unwrap();
310        assert_eq!(v, json!({"id": 7, "name": "x"}));
311    }
312
313    #[test]
314    fn untyped_values_fall_back_to_structural_decode() {
315        let v = decode_value(&pv(Kind::NumberValue(2.0)), None, "u").unwrap();
316        assert_eq!(v, json!(2.0));
317        let v = decode_value(
318            &pv(Kind::ListValue(prost_types::ListValue {
319                values: vec![pv(Kind::BoolValue(false))],
320            })),
321            None,
322            "u",
323        )
324        .unwrap();
325        assert_eq!(v, json!([false]));
326    }
327}
faucet_common_spanner/decode.rs

faucet_common_spanner/
decode.rs
