faucet_source_bigquery/
convert.rs

1//! Type-aware conversion from BigQuery `TableRow`s to JSON objects.
2//!
3//! BigQuery returns every scalar cell as a string regardless of declared
4//! type, with non-trivial structure (`RECORD`/`STRUCT`/`REPEATED`) reflected
5//! in nested cell shapes:
6//!
7//! - A `RECORD` cell carries `{"f": [...nested cells...]}` in its `value`
8//!   field.
9//! - A `REPEATED` cell carries `[{"v": ...cell...}, ...]` in its `value`
10//!   field — one element per array entry.
11//!
12//! This module walks the schema in lockstep with each row's cells and emits
13//! a `serde_json::Value::Object` keyed by column name, with each cell typed
14//! according to its `TableFieldSchema`.
15//!
16//! ## Type mapping
17//!
18//! | BigQuery type                          | JSON output                      |
19//! |----------------------------------------|----------------------------------|
20//! | `INTEGER` / `INT64`                    | `Number` (i64 if it fits, else f64) |
21//! | `FLOAT` / `FLOAT64`                    | `Number` (f64)                   |
22//! | `NUMERIC` / `BIGNUMERIC`               | `String` (full precision preserved) |
23//! | `BOOLEAN` / `BOOL`                     | `Bool`                           |
24//! | `RECORD` / `STRUCT`                    | nested `Object`                  |
25//! | `REPEATED <T>` mode                    | `Array<T>`                       |
26//! | `JSON`                                 | parsed `Value` (falls back to `String`) |
27//! | `STRING` / `BYTES` / `TIMESTAMP` / `DATE` / `TIME` / `DATETIME` / `GEOGRAPHY` / `INTERVAL` | `String` |
28//! | `NULL` cell                            | `Null`                           |
29
30use gcp_bigquery_client::model::field_type::FieldType;
31use gcp_bigquery_client::model::table_field_schema::TableFieldSchema;
32use gcp_bigquery_client::model::table_row::TableRow;
33use serde::Deserialize;
34use serde_json::{Map, Value};
35
36/// Convert a BigQuery `TableRow` into a JSON object using its schema.
37///
38/// `fields` and the row's `columns` are expected to line up positionally —
39/// the field at index `i` describes the cell at index `i`. Missing trailing
40/// cells map to `Null`.
41pub fn row_to_json(row: &TableRow, fields: &[TableFieldSchema]) -> Value {
42    let cells = row.columns.as_deref().unwrap_or(&[]);
43    let mut obj = Map::with_capacity(fields.len());
44    for (i, field) in fields.iter().enumerate() {
45        let raw = cells.get(i).and_then(|c| c.value.as_ref());
46        obj.insert(field.name.clone(), cell_to_json(raw, field));
47    }
48    Value::Object(obj)
49}
50
51/// Convert one `TableCell.value` (or `None` for a missing cell) to JSON,
52/// honouring `REPEATED` mode and `RECORD`/`STRUCT` nesting.
53fn cell_to_json(value: Option<&Value>, field: &TableFieldSchema) -> Value {
54    let Some(value) = value else {
55        return Value::Null;
56    };
57    if value.is_null() {
58        return Value::Null;
59    }
60
61    let repeated = field
62        .mode
63        .as_deref()
64        .map(|m| m.eq_ignore_ascii_case("REPEATED"))
65        .unwrap_or(false);
66
67    if repeated {
68        // Repeated cells are arrays of `{v: ...}` wrappers; recurse into
69        // each element as if it were a single non-repeated cell of the same
70        // field type.
71        let Some(array) = value.as_array() else {
72            // Shape we didn't expect — fall back to a single-element array
73            // carrying the raw value rather than dropping data on the floor.
74            return Value::Array(vec![scalar_or_record(value, field)]);
75        };
76        let items: Vec<Value> = array
77            .iter()
78            .map(|elem| {
79                // Each element is a `TableCell`-shaped wrapper; unwrap the
80                // `{"v": ...}` envelope if present, otherwise treat the
81                // element itself as the payload.
82                match elem.get("v") {
83                    Some(inner) => scalar_or_record(inner, field),
84                    None => scalar_or_record(elem, field),
85                }
86            })
87            .collect();
88        return Value::Array(items);
89    }
90
91    scalar_or_record(value, field)
92}
93
94/// Convert one already-unwrapped cell payload (no `REPEATED` wrapper) to JSON.
95fn scalar_or_record(payload: &Value, field: &TableFieldSchema) -> Value {
96    if matches!(field.r#type, FieldType::Record | FieldType::Struct) {
97        return record_to_json(payload, field);
98    }
99
100    let Some(s) = payload.as_str() else {
101        // BigQuery may already have typed numeric/bool values for certain
102        // formats; pass through unchanged.
103        return payload.clone();
104    };
105
106    match field.r#type {
107        FieldType::Integer | FieldType::Int64 => parse_integer(s),
108        FieldType::Float | FieldType::Float64 => parse_float(s),
109        FieldType::Boolean | FieldType::Bool => parse_bool(s),
110        FieldType::Json => serde_json::from_str(s).unwrap_or_else(|_| Value::String(s.to_owned())),
111        // STRING, BYTES, TIMESTAMP, DATE, TIME, DATETIME, NUMERIC,
112        // BIGNUMERIC, GEOGRAPHY, INTERVAL — keep as string for
113        // round-trip safety.
114        _ => Value::String(s.to_owned()),
115    }
116}
117
118/// Decode a RECORD/STRUCT cell. The payload is itself a `TableRow`-shaped
119/// object `{"f": [...]}`, so we deserialise it into a `TableRow` and recurse.
120fn record_to_json(payload: &Value, field: &TableFieldSchema) -> Value {
121    let row: TableRow = match TableRow::deserialize(payload.clone()) {
122        Ok(row) => row,
123        Err(_) => return payload.clone(),
124    };
125    let nested_fields = field.fields.as_deref().unwrap_or(&[]);
126    row_to_json(&row, nested_fields)
127}
128
129fn parse_integer(s: &str) -> Value {
130    if let Ok(i) = s.parse::<i64>() {
131        return Value::Number(i.into());
132    }
133    parse_float(s)
134}
135
136fn parse_float(s: &str) -> Value {
137    match s.parse::<f64>() {
138        Ok(f) => serde_json::Number::from_f64(f)
139            .map(Value::Number)
140            .unwrap_or_else(|| Value::String(s.to_owned())),
141        Err(_) => Value::String(s.to_owned()),
142    }
143}
144
145fn parse_bool(s: &str) -> Value {
146    match s {
147        "true" | "TRUE" | "1" => Value::Bool(true),
148        "false" | "FALSE" | "0" => Value::Bool(false),
149        other => Value::String(other.to_owned()),
150    }
151}
152
153#[cfg(test)]
154mod tests {
155    use super::*;
156    use gcp_bigquery_client::model::field_type::FieldType;
157    use gcp_bigquery_client::model::table_cell::TableCell;
158    use serde_json::json;
159
160    fn field(name: &str, ty: FieldType) -> TableFieldSchema {
161        TableFieldSchema::new(name, ty)
162    }
163
164    fn repeated(mut f: TableFieldSchema) -> TableFieldSchema {
165        f.mode = Some("REPEATED".to_string());
166        f
167    }
168
169    fn cells(values: Vec<Value>) -> TableRow {
170        TableRow {
171            columns: Some(
172                values
173                    .into_iter()
174                    .map(|v| TableCell { value: Some(v) })
175                    .collect(),
176            ),
177        }
178    }
179
180    #[test]
181    fn integer_parses_as_number() {
182        let row = cells(vec![json!("42")]);
183        let fields = vec![field("id", FieldType::Integer)];
184        assert_eq!(row_to_json(&row, &fields), json!({"id": 42}));
185    }
186
187    #[test]
188    fn float_parses_as_number() {
189        let row = cells(vec![json!("2.5")]);
190        let fields = vec![field("ratio", FieldType::Float)];
191        assert_eq!(row_to_json(&row, &fields), json!({"ratio": 2.5}));
192    }
193
194    #[test]
195    fn boolean_parses_lowercase() {
196        let row = cells(vec![json!("true"), json!("false")]);
197        let fields = vec![field("a", FieldType::Boolean), field("b", FieldType::Bool)];
198        assert_eq!(row_to_json(&row, &fields), json!({"a": true, "b": false}));
199    }
200
201    #[test]
202    fn numeric_preserves_string_for_precision() {
203        let row = cells(vec![json!("1234567890123456789012345.6789")]);
204        let fields = vec![field("amount", FieldType::Numeric)];
205        assert_eq!(
206            row_to_json(&row, &fields),
207            json!({"amount": "1234567890123456789012345.6789"})
208        );
209    }
210
211    #[test]
212    fn timestamp_passes_through_as_string() {
213        let row = cells(vec![json!("1.7e9")]);
214        let fields = vec![field("ts", FieldType::Timestamp)];
215        assert_eq!(row_to_json(&row, &fields), json!({"ts": "1.7e9"}));
216    }
217
218    #[test]
219    fn null_cell_maps_to_null() {
220        let row = cells(vec![json!(null)]);
221        let fields = vec![field("x", FieldType::Integer)];
222        assert_eq!(row_to_json(&row, &fields), json!({"x": null}));
223    }
224
225    #[test]
226    fn missing_cell_maps_to_null() {
227        // Row has zero cells but the schema declares one field.
228        let row = TableRow {
229            columns: Some(vec![]),
230        };
231        let fields = vec![field("x", FieldType::String)];
232        assert_eq!(row_to_json(&row, &fields), json!({"x": null}));
233    }
234
235    #[test]
236    fn repeated_integers_become_array() {
237        let row = cells(vec![json!([{"v": "1"}, {"v": "2"}, {"v": "3"}])]);
238        let fields = vec![repeated(field("ids", FieldType::Integer))];
239        assert_eq!(row_to_json(&row, &fields), json!({"ids": [1, 2, 3]}));
240    }
241
242    #[test]
243    fn record_decodes_nested_object() {
244        let row = cells(vec![json!({"f": [{"v": "1"}, {"v": "alice"}]})]);
245        let mut record_field = field("user", FieldType::Record);
246        record_field.fields = Some(vec![
247            field("id", FieldType::Integer),
248            field("name", FieldType::String),
249        ]);
250        let fields = vec![record_field];
251        assert_eq!(
252            row_to_json(&row, &fields),
253            json!({"user": {"id": 1, "name": "alice"}})
254        );
255    }
256
257    #[test]
258    fn json_field_parses_inner_json() {
259        let row = cells(vec![json!(r#"{"k": 1}"#)]);
260        let fields = vec![field("payload", FieldType::Json)];
261        assert_eq!(row_to_json(&row, &fields), json!({"payload": {"k": 1}}));
262    }
263
264    #[test]
265    fn json_field_falls_back_to_string_on_invalid_json() {
266        let row = cells(vec![json!("not-json")]);
267        let fields = vec![field("payload", FieldType::Json)];
268        assert_eq!(row_to_json(&row, &fields), json!({"payload": "not-json"}));
269    }
270
271    #[test]
272    fn unparseable_integer_falls_back_to_string() {
273        let row = cells(vec![json!("abc")]);
274        let fields = vec![field("id", FieldType::Integer)];
275        assert_eq!(row_to_json(&row, &fields), json!({"id": "abc"}));
276    }
277}
faucet_source_bigquery/convert.rs

faucet_source_bigquery/
convert.rs
