polydat_nodes/
emit.rs

1// Copyright 2024-2026 Jonathan Shook
2// SPDX-License-Identifier: Apache-2.0
3
4//! Row emission as a graph node.
5//!
6//! The `polydat` binary's `--emit` option is a source transform: it
7//! appends one binding, `__emit := emit_row(format, names, a, b, ...)`,
8//! to the program. The node sees every wire it names through ordinary
9//! wiring, so emission is part of the kernel rather than a decorator
10//! around it. When the `for` construct lands, the same binding is
11//! inserted inside the traversed block and observes that block's
12//! scope.
13//!
14//! Rows accumulate in a thread-local buffer so concurrent fibers never
15//! contend. The harness drains each fiber's buffer with [`take_rows`]
16//! at chunk boundaries and decides on ordering.
17
18use std::cell::RefCell;
19
20use polydat::ast::Value;
21
22thread_local! {
23    static ROWS: RefCell<Vec<String>> = const { RefCell::new(Vec::new()) };
24}
25
26/// Formats supported by `emit_row`. Parsed from the node's `format`
27/// constant.
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
29pub enum EmitFormat {
30    /// `name=value` pairs separated by single spaces.
31    Map,
32    /// Comma-separated values; strings are quoted when they contain a
33    /// comma, quote, or newline.
34    Csv,
35    /// One JSON object per row.
36    Jsonl,
37    /// The values' text as it is, one per line. This is how a tile is
38    /// emitted: `--emit tile:<name>` selects the tile and this format.
39    Text,
40}
41
42impl EmitFormat {
43    /// The format named by `s`, case-insensitively, if any.
44    pub fn parse(s: &str) -> Option<Self> {
45        match s.trim().to_ascii_lowercase().as_str() {
46            "map" => Some(Self::Map),
47            "csv" => Some(Self::Csv),
48            "json" | "jsonl" => Some(Self::Jsonl),
49            "text" => Some(Self::Text),
50            _ => None,
51        }
52    }
53}
54
55/// The header line for a format, if the format has one.
56pub fn header(format: EmitFormat, names: &[&str]) -> Option<String> {
57    match format {
58        EmitFormat::Csv => Some(names.join(",")),
59        EmitFormat::Map | EmitFormat::Jsonl | EmitFormat::Text => None,
60    }
61}
62
63/// Render one row without touching the buffer.
64pub fn render_row(format: EmitFormat, names: &[&str], values: &[Value]) -> String {
65    match format {
66        EmitFormat::Text => values
67            .iter()
68            .map(|v| v.to_display_string())
69            .collect::<Vec<_>>()
70            .join("\n"),
71        EmitFormat::Map => names
72            .iter()
73            .zip(values)
74            .map(|(n, v)| format!("{n}={}", v.to_display_string()))
75            .collect::<Vec<_>>()
76            .join(" "),
77        EmitFormat::Csv => values
78            .iter()
79            .map(|v| csv_cell(&v.to_display_string()))
80            .collect::<Vec<_>>()
81            .join(","),
82        EmitFormat::Jsonl => {
83            let mut map = serde_json::Map::with_capacity(names.len());
84            for (n, v) in names.iter().zip(values) {
85                map.insert((*n).to_string(), v.to_json_value());
86            }
87            serde_json::Value::Object(map).to_string()
88        }
89    }
90}
91
92fn csv_cell(s: &str) -> String {
93    if s.contains([',', '"', '\n']) {
94        format!("\"{}\"", s.replace('"', "\"\""))
95    } else {
96        s.to_string()
97    }
98}
99
100/// Drain the calling thread's emitted rows.
101pub fn take_rows() -> Vec<String> {
102    ROWS.with(|r| std::mem::take(&mut *r.borrow_mut()))
103}
104
105/// Append one formatted row to the calling thread's buffer and return
106/// the number of values emitted. `format` is `map`, `csv`, or `jsonl`;
107/// `names` is the comma-separated list of wire names in the same order
108/// as the variadic values.
109#[polydat::polydat_node(category = Diagnostic, purity = SideChannel(Other), variadic_min = 0)]
110fn emit_row(format: Const<&str>, names: Const<&str>, values: &[Value]) -> u64 {
111    let fmt = EmitFormat::parse(&format).unwrap_or(EmitFormat::Map);
112    let names: Vec<&str> = names
113        .split(',')
114        .map(str::trim)
115        .filter(|s| !s.is_empty())
116        .collect();
117    let row = render_row(fmt, &names, values);
118    ROWS.with(|r| r.borrow_mut().push(row));
119    values.len() as u64
120}
121
122#[cfg(test)]
123mod tests {
124    use super::*;
125    use polydat::ast::PolydatNode;
126
127    #[test]
128    fn csv_quotes_only_when_needed() {
129        let vals = [
130            Value::U64(1),
131            Value::Str("a,b".into()),
132            Value::Str("plain".into()),
133        ];
134        let row = render_row(EmitFormat::Csv, &["x", "y", "z"], &vals);
135        assert_eq!(row, "1,\"a,b\",plain");
136    }
137
138    #[test]
139    fn map_and_jsonl_shapes() {
140        let vals = [Value::U64(7), Value::F64(1.5)];
141        assert_eq!(render_row(EmitFormat::Map, &["a", "b"], &vals), "a=7 b=1.5");
142        assert_eq!(
143            render_row(EmitFormat::Jsonl, &["a", "b"], &vals),
144            r#"{"a":7,"b":1.5}"#
145        );
146    }
147
148    #[test]
149    fn rows_accumulate_per_thread_and_drain() {
150        let node = EmitRow::new("csv".to_string(), "a".to_string(), 1);
151        let mut out = [Value::None];
152        node.eval(&[Value::U64(3)], &mut out);
153        node.eval(&[Value::U64(4)], &mut out);
154        assert_eq!(out[0].as_u64(), 1);
155        assert_eq!(take_rows(), vec!["3".to_string(), "4".to_string()]);
156        assert!(take_rows().is_empty());
157    }
158}
polydat_nodes/emit.rs

polydat_nodes/
emit.rs
