benday_core/
ingest.rs

1//! Data ingestion: resolve the spec's inline data and/or a piped data
2//! document into a normalized `Table`. Pure — the CLI does the I/O.
3//!
4//! Strictness boundary: the SPEC is agent-authored intent, so its data object
5//! is strict (`deny_unknown_fields`, over in `spec.rs`). The stdin document is
6//! producer-shaped payload (e.g. an MCP `structuredContent` envelope), so it
7//! is tolerant: known fields are used, unknown fields (query provenance etc.)
8//! are ignored.
9
10use std::collections::HashMap;
11
12use serde::{Deserialize, Serialize};
13use serde_json::{Map, Value};
14
15use crate::error::Error;
16use crate::spec::{Column, Data, FieldType, Spec};
17
18pub type Row = Map<String, Value>;
19
20/// A data document piped to stdin: a columnar envelope or a bare row array.
21/// Tolerant by design — no `deny_unknown_fields`.
22#[derive(Debug, Deserialize)]
23#[serde(untagged)]
24pub enum DataDoc {
25    Envelope {
26        columns: Vec<EnvColumn>,
27        rows: Vec<Vec<Value>>,
28        #[serde(default)]
29        truncated: Option<bool>,
30        #[serde(default)]
31        total_rows: Option<u64>,
32    },
33    Rows(Vec<Row>),
34}
35
36/// Envelope column: tolerant twin of `spec::Column` (producers may add keys).
37#[derive(Debug, Deserialize)]
38pub struct EnvColumn {
39    pub name: String,
40    #[serde(default, rename = "type")]
41    pub ty: Option<String>,
42}
43
44impl From<&Column> for EnvColumn {
45    fn from(c: &Column) -> Self {
46        EnvColumn {
47            name: c.name.clone(),
48            ty: c.ty.clone(),
49        }
50    }
51}
52
53#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
54#[serde(rename_all = "snake_case")]
55pub enum DataSource {
56    InlineValues,
57    InlineColumns,
58    StdinValues,
59    StdinColumns,
60}
61
62#[derive(Debug, Serialize)]
63pub struct DataProvenance {
64    pub source: DataSource,
65    pub truncated: Option<bool>,
66    pub total_rows: Option<u64>,
67}
68
69/// Normalized data ready for the compiler: row-major rows (as the compiler
70/// has always consumed), declared column types, and where it all came from.
71#[derive(Debug)]
72pub struct Table {
73    pub rows: Vec<Row>,
74    pub declared: HashMap<String, FieldType>,
75    pub provenance: DataProvenance,
76}
77
78/// Parse a stdin data document. Wraps serde's unhelpful untagged-enum error
79/// with the two accepted shapes.
80pub fn parse_data_doc(s: &str) -> Result<DataDoc, Error> {
81    serde_json::from_str(s).map_err(|e| {
82        Error::Data(format!(
83            "cannot parse stdin as a data document; expected \
84             {{\"columns\":[{{\"name\",\"type\"?}}...],\"rows\":[[...]...]}} or a JSON array \
85             of row objects: {e}"
86        ))
87    })
88}
89
90/// Map a declared column type (BigQuery + common SQL spellings, case-
91/// insensitive) to a field type. Unknown names fall back to nominal — NOT an
92/// error: producers grow types, and nominal is safe-wrong-in-the-obvious-way.
93/// DATE/DATETIME/TIMESTAMP/TIME map to temporal: benday now owns time layout
94/// (true positions on a calendar scale) when SQL is absent — see
95/// docs/plans/2026-07-05-temporal-family-design.md for why this reverses the
96/// old "no temporal scale" doctrine. An explicit `"ordinal"` on the encoding
97/// restores the evenly-spaced behavior per chart.
98pub fn declared_field_type(t: &str) -> FieldType {
99    match t.to_ascii_uppercase().as_str() {
100        "INT64" | "INTEGER" | "INT" | "SMALLINT" | "BIGINT" | "FLOAT64" | "FLOAT" | "DOUBLE"
101        | "NUMERIC" | "BIGNUMERIC" | "DECIMAL" | "REAL" => FieldType::Quantitative,
102        "DATE" | "DATETIME" | "TIMESTAMP" | "TIME" => FieldType::Temporal,
103        _ => FieldType::Nominal,
104    }
105}
106
107/// Resolve spec + optional stdin document into a Table. Owns ALL precedence
108/// and data-shape errors so the corpus can pin them.
109pub fn resolve(spec: &Spec, stdin: Option<DataDoc>) -> Result<Table, Error> {
110    match (&spec.data, stdin) {
111        (Some(_), Some(_)) => Err(Error::Spec(
112            "data provided twice: the spec has inline `data` and a data document \
113             arrived on stdin; remove one"
114                .into(),
115        )),
116        (Some(data), None) => resolve_inline(data),
117        (None, Some(doc)) => resolve_stdin(doc),
118        (None, None) => Err(Error::Spec(
119            "no data: the spec has no `data` and nothing arrived on stdin; add \
120             data.values or data.columns+rows to the spec, or pipe a data document"
121                .into(),
122        )),
123    }
124}
125
126/// Resolve the spec's inline `data` object. Exactly one form is allowed:
127/// `values` (tidy row objects) or `columns` + `rows` (columnar). Any other
128/// combination is a spec error — serde can't express either/or without
129/// mangling the error paths, so it's checked here.
130fn resolve_inline(data: &Data) -> Result<Table, Error> {
131    match (&data.values, &data.columns, &data.rows) {
132        // Inline `values`: row-major already, no declared types.
133        (Some(values), None, None) => finish(
134            values.clone(),
135            HashMap::new(),
136            DataSource::InlineValues,
137            None,
138            None,
139        ),
140        (None, Some(columns), Some(rows)) => {
141            // Reuse the envelope's zip/validation. The spec's `Column` is the
142            // strict twin of `EnvColumn`; same fields, so convert and share.
143            let env: Vec<EnvColumn> = columns.iter().map(EnvColumn::from).collect();
144            let (rows, declared) = columnar_to_rows(&env, rows)?;
145            finish(rows, declared, DataSource::InlineColumns, None, None)
146        }
147        _ => Err(Error::Spec(
148            "data must contain either `values`, or `columns` and `rows`".into(),
149        )),
150    }
151}
152
153/// Resolve a stdin data document (envelope or bare rows) into a Table.
154fn resolve_stdin(doc: DataDoc) -> Result<Table, Error> {
155    match doc {
156        DataDoc::Envelope {
157            columns,
158            rows,
159            truncated,
160            total_rows,
161        } => {
162            let (rows, declared) = columnar_to_rows(&columns, &rows)?;
163            finish(
164                rows,
165                declared,
166                DataSource::StdinColumns,
167                truncated,
168                total_rows,
169            )
170        }
171        DataDoc::Rows(rows) => finish(rows, HashMap::new(), DataSource::StdinValues, None, None),
172    }
173}
174
175/// Zip a columnar envelope into row-major objects, keyed in column order, and
176/// collect declared column types. Shared shape for inline and stdin columnar.
177fn columnar_to_rows(
178    columns: &[EnvColumn],
179    rows: &[Vec<Value>],
180) -> Result<(Vec<Row>, HashMap<String, FieldType>), Error> {
181    let mut declared = HashMap::new();
182    let mut seen = std::collections::HashSet::new();
183    for col in columns {
184        if !seen.insert(col.name.as_str()) {
185            return Err(Error::Data(format!(
186                "duplicate column name \"{}\"",
187                col.name
188            )));
189        }
190        if let Some(ty) = &col.ty {
191            declared.insert(col.name.clone(), declared_field_type(ty));
192        }
193    }
194
195    let want = columns.len();
196    let mut out = Vec::with_capacity(rows.len());
197    for (i, row) in rows.iter().enumerate() {
198        if row.len() != want {
199            return Err(Error::Data(format!(
200                "row {i} has {} values but {want} columns are declared",
201                row.len()
202            )));
203        }
204        let mut obj = Row::new();
205        for (col, val) in columns.iter().zip(row) {
206            obj.insert(col.name.clone(), val.clone());
207        }
208        out.push(obj);
209    }
210    Ok((out, declared))
211}
212
213/// Apply the empty-rows rule (any form) and assemble the Table.
214fn finish(
215    rows: Vec<Row>,
216    declared: HashMap<String, FieldType>,
217    source: DataSource,
218    truncated: Option<bool>,
219    total_rows: Option<u64>,
220) -> Result<Table, Error> {
221    if rows.is_empty() {
222        return Err(Error::Data(
223            "data has no rows; provide at least one row".into(),
224        ));
225    }
226    Ok(Table {
227        rows,
228        declared,
229        provenance: DataProvenance {
230            source,
231            truncated,
232            total_rows,
233        },
234    })
235}
236
237#[cfg(test)]
238mod tests {
239    use super::*;
240    use serde_json::json;
241
242    fn spec_with_values() -> Spec {
243        serde_json::from_str(
244            r#"{"data":{"values":[{"x":1}]},"mark":"bar",
245               "encoding":{"x":{"field":"x"},"y":{"field":"y"}}}"#,
246        )
247        .expect("fixture spec parses")
248    }
249
250    fn spec_with_empty_values() -> Spec {
251        serde_json::from_str(
252            r#"{"data":{"values":[]},"mark":"bar",
253               "encoding":{"x":{"field":"x"},"y":{"field":"y"}}}"#,
254        )
255        .expect("fixture spec parses")
256    }
257
258    #[test]
259    fn parses_bare_row_array() {
260        let doc = parse_data_doc(r#"[{"a": 1}, {"a": 2}]"#).expect("bare array parses");
261        match doc {
262            DataDoc::Rows(rows) => assert_eq!(rows.len(), 2),
263            other => panic!("expected Rows, got {other:?}"),
264        }
265    }
266
267    #[test]
268    fn envelope_tolerates_unknown_keys() {
269        // The producer emits a `query` provenance block benday ignores.
270        let doc = parse_data_doc(
271            r#"{"columns":[{"name":"day","type":"STRING"},{"name":"n","type":"INT64"}],
272                "rows":[["mon",32]],
273                "query":{"job_id":"abc","note":"ignored"}}"#,
274        )
275        .expect("envelope with extra keys parses");
276        match doc {
277            DataDoc::Envelope { columns, rows, .. } => {
278                assert_eq!(columns.len(), 2);
279                assert_eq!(rows.len(), 1);
280            }
281            other => panic!("expected Envelope, got {other:?}"),
282        }
283    }
284
285    #[test]
286    fn envelope_truncated_and_total_rows_reach_provenance() {
287        let doc = parse_data_doc(
288            r#"{"columns":[{"name":"n"}],"rows":[[1]],"truncated":true,"total_rows":123}"#,
289        )
290        .expect("envelope parses");
291        let table = resolve_stdin(doc).expect("resolves");
292        assert_eq!(table.provenance.source, DataSource::StdinColumns);
293        assert_eq!(table.provenance.truncated, Some(true));
294        assert_eq!(table.provenance.total_rows, Some(123));
295    }
296
297    #[test]
298    fn bare_rows_provenance_has_no_envelope_fields() {
299        let doc = parse_data_doc(r#"[{"a":1}]"#).expect("parses");
300        let table = resolve_stdin(doc).expect("resolves");
301        assert_eq!(table.provenance.source, DataSource::StdinValues);
302        assert_eq!(table.provenance.truncated, None);
303        assert_eq!(table.provenance.total_rows, None);
304        assert!(table.declared.is_empty());
305    }
306
307    #[test]
308    fn columnar_zips_rows_in_column_order() {
309        let doc = parse_data_doc(
310            r#"{"columns":[{"name":"day","type":"STRING"},{"name":"n","type":"INT64"}],
311                "rows":[["mon",32],["tue",78]]}"#,
312        )
313        .expect("parses");
314        let table = resolve_stdin(doc).expect("resolves");
315        assert_eq!(table.rows.len(), 2);
316        assert_eq!(table.rows[0].get("day"), Some(&json!("mon")));
317        assert_eq!(table.rows[0].get("n"), Some(&json!(32)));
318        assert_eq!(table.rows[1].get("day"), Some(&json!("tue")));
319        assert_eq!(table.rows[1].get("n"), Some(&json!(78)));
320        // keys land in declared column order
321        let keys: Vec<&String> = table.rows[0].keys().collect();
322        assert_eq!(keys, vec!["day", "n"]);
323        assert_eq!(table.declared.get("day"), Some(&FieldType::Nominal));
324        assert_eq!(table.declared.get("n"), Some(&FieldType::Quantitative));
325    }
326
327    #[test]
328    fn duplicate_column_name_errors() {
329        let doc = parse_data_doc(r#"{"columns":[{"name":"a"},{"name":"a"}],"rows":[[1,2]]}"#)
330            .expect("parses");
331        let err = resolve_stdin(doc).expect_err("duplicate must error");
332        insta::assert_snapshot!(err.to_string(), @r###"duplicate column name "a""###);
333    }
334
335    #[test]
336    fn row_length_mismatch_errors() {
337        let doc =
338            parse_data_doc(r#"{"columns":[{"name":"a"},{"name":"b"}],"rows":[[1,2],[1,2,3]]}"#)
339                .expect("parses");
340        let err = resolve_stdin(doc).expect_err("length mismatch must error");
341        insta::assert_snapshot!(
342            err.to_string(),
343            @"row 1 has 3 values but 2 columns are declared"
344        );
345    }
346
347    #[test]
348    fn empty_rows_errors_columnar() {
349        let doc = parse_data_doc(r#"{"columns":[{"name":"a"}],"rows":[]}"#).expect("parses");
350        let err = resolve_stdin(doc).expect_err("empty rows must error");
351        insta::assert_snapshot!(err.to_string(), @"data has no rows; provide at least one row");
352    }
353
354    #[test]
355    fn empty_rows_errors_bare_array() {
356        let doc = parse_data_doc(r#"[]"#).expect("parses");
357        let err = resolve_stdin(doc).expect_err("empty rows must error");
358        insta::assert_snapshot!(err.to_string(), @"data has no rows; provide at least one row");
359    }
360
361    #[test]
362    fn empty_rows_errors_inline_values() {
363        let err = resolve(&spec_with_empty_values(), None).expect_err("empty inline must error");
364        insta::assert_snapshot!(err.to_string(), @"data has no rows; provide at least one row");
365    }
366
367    #[test]
368    fn inline_values_resolve_to_inline_provenance() {
369        let table = resolve(&spec_with_values(), None).expect("resolves");
370        assert_eq!(table.provenance.source, DataSource::InlineValues);
371        assert_eq!(table.provenance.truncated, None);
372        assert_eq!(table.provenance.total_rows, None);
373        assert!(table.declared.is_empty());
374        assert_eq!(table.rows.len(), 1);
375    }
376
377    #[test]
378    fn data_provided_twice_errors() {
379        let doc = parse_data_doc(r#"[{"a":1}]"#).expect("parses");
380        let err = resolve(&spec_with_values(), Some(doc)).expect_err("data twice must error");
381        insta::assert_snapshot!(
382            err.to_string(),
383            @"data provided twice: the spec has inline `data` and a data document arrived on stdin; remove one"
384        );
385    }
386
387    #[test]
388    fn inline_columnar_resolves_with_declared_types() {
389        let spec: Spec = serde_json::from_str(
390            r#"{"data":{"columns":[{"name":"day","type":"STRING"},{"name":"n","type":"INT64"}],
391                       "rows":[["mon",32],["tue",78]]},
392                "mark":"bar","encoding":{"x":{"field":"day"},"y":{"field":"n"}}}"#,
393        )
394        .expect("inline columnar spec parses");
395        let table = resolve(&spec, None).expect("resolves");
396        assert_eq!(table.provenance.source, DataSource::InlineColumns);
397        assert_eq!(table.rows.len(), 2);
398        assert_eq!(table.rows[0].get("day"), Some(&json!("mon")));
399        assert_eq!(table.rows[0].get("n"), Some(&json!(32)));
400        assert_eq!(table.declared.get("day"), Some(&FieldType::Nominal));
401        assert_eq!(table.declared.get("n"), Some(&FieldType::Quantitative));
402    }
403
404    #[test]
405    fn inline_data_both_forms_errors() {
406        let spec: Spec = serde_json::from_str(
407            r#"{"data":{"values":[{"a":1}],"columns":[{"name":"a"}],"rows":[[1]]},
408                "mark":"bar","encoding":{"x":{"field":"a"},"y":{"field":"a"}}}"#,
409        )
410        .expect("spec parses");
411        let err = resolve(&spec, None).expect_err("both forms must error");
412        insta::assert_snapshot!(
413            err.to_string(),
414            @"data must contain either `values`, or `columns` and `rows`"
415        );
416    }
417
418    #[test]
419    fn no_data_anywhere_errors() {
420        let spec: Spec = serde_json::from_str(
421            r#"{"mark":"bar","encoding":{"x":{"field":"a"},"y":{"field":"b"}}}"#,
422        )
423        .expect("spec without data parses");
424        let err = resolve(&spec, None).expect_err("no data must error");
425        insta::assert_snapshot!(
426            err.to_string(),
427            @"no data: the spec has no `data` and nothing arrived on stdin; add data.values or data.columns+rows to the spec, or pipe a data document"
428        );
429    }
430
431    #[test]
432    fn declared_field_type_mapping() {
433        // Quantitative spellings.
434        for t in [
435            "INT64",
436            "INTEGER",
437            "INT",
438            "SMALLINT",
439            "BIGINT",
440            "FLOAT64",
441            "FLOAT",
442            "DOUBLE",
443            "NUMERIC",
444            "BIGNUMERIC",
445            "DECIMAL",
446            "REAL",
447        ] {
448            assert_eq!(declared_field_type(t), FieldType::Quantitative, "{t}");
449        }
450        // Date/time spellings map to temporal.
451        for t in ["DATE", "DATETIME", "TIMESTAMP", "TIME"] {
452            assert_eq!(declared_field_type(t), FieldType::Temporal, "{t}");
453        }
454        // Strings and unknowns fall back to nominal.
455        assert_eq!(declared_field_type("STRING"), FieldType::Nominal);
456        assert_eq!(declared_field_type("BOOL"), FieldType::Nominal);
457        assert_eq!(declared_field_type("whatever"), FieldType::Nominal);
458        // Case-insensitive.
459        assert_eq!(declared_field_type("int64"), FieldType::Quantitative);
460        assert_eq!(declared_field_type("Date"), FieldType::Temporal);
461        assert_eq!(declared_field_type("String"), FieldType::Nominal);
462    }
463}
benday_core/ingest.rs

benday_core/
ingest.rs
