Skip to main content

stryke/
native_data.rs

1//! Native CSV (`csv` crate), SQLite (`rusqlite`), and HTTP JSON (`ureq` + `serde_json`) helpers.
2
3use std::io::Read;
4use std::sync::Arc;
5use std::time::Duration;
6
7use indexmap::IndexMap;
8use jaq_core::data::JustLut;
9use num_traits::cast::ToPrimitive;
10use parking_lot::{Mutex, RwLock};
11use rayon::prelude::*;
12use rusqlite::{types::Value, Connection};
13use serde_json::Value as JsonValue;
14
15use crate::ast::StructDef;
16use crate::error::{PerlError, PerlResult};
17use crate::value::{HeapObject, PerlDataFrame, PerlValue, StructInstance};
18
19/// Parallel row→hashref conversion after a sequential CSV parse (good CPU parallelism on wide files).
20pub(crate) fn par_csv_read(path: &str) -> PerlResult<PerlValue> {
21    let mut rdr = csv::Reader::from_path(path)
22        .map_err(|e| PerlError::runtime(format!("par_csv_read: {}: {}", path, e), 0))?;
23    let headers: Vec<String> = rdr
24        .headers()
25        .map_err(|e| PerlError::runtime(format!("par_csv_read: {}: {}", path, e), 0))?
26        .iter()
27        .map(|s| s.to_string())
28        .collect();
29    let mut raw_rows: Vec<csv::StringRecord> = Vec::new();
30    for rec in rdr.records() {
31        raw_rows.push(rec.map_err(|e| PerlError::runtime(format!("par_csv_read: {}", e), 0))?);
32    }
33    let rows: Vec<PerlValue> = raw_rows
34        .into_par_iter()
35        .map(|record| {
36            let mut map = IndexMap::new();
37            for (i, h) in headers.iter().enumerate() {
38                let cell = record.get(i).unwrap_or("");
39                map.insert(h.clone(), PerlValue::string(cell.to_string()));
40            }
41            PerlValue::hash_ref(Arc::new(RwLock::new(map)))
42        })
43        .collect();
44    Ok(PerlValue::array(rows))
45}
46
47/// Columnar dataframe from a CSV path (header row + string cells; use `sum` etc. with numeric strings).
48pub(crate) fn dataframe_from_elements(val: &PerlValue) -> PerlResult<PerlValue> {
49    let rows = val.map_flatten_outputs(true);
50    if rows.is_empty() {
51        return Ok(PerlValue::dataframe(Arc::new(Mutex::new(PerlDataFrame {
52            columns: vec![],
53            cols: vec![],
54            group_by: None,
55        }))));
56    }
57
58    // Detect format: list of hashrefs or list of arrayrefs
59    let first_row = &rows[0];
60    if let Some(first_row_map) = first_row.as_hash_ref() {
61        // List of hashrefs: use keys of the first row as columns
62        let columns: Vec<String> = first_row_map.read().keys().cloned().collect();
63        let mut cols: Vec<Vec<PerlValue>> = (0..columns.len()).map(|_| Vec::new()).collect();
64        for row_val in rows {
65            if let Some(row_lock) = row_val.as_hash_ref() {
66                let row_map = row_lock.read();
67                for (i, col_name) in columns.iter().enumerate() {
68                    cols[i].push(row_map.get(col_name).cloned().unwrap_or(PerlValue::UNDEF));
69                }
70            }
71        }
72        return Ok(PerlValue::dataframe(Arc::new(Mutex::new(PerlDataFrame {
73            columns,
74            cols,
75            group_by: None,
76        }))));
77    } else if let Some(first_row_lock) = first_row.as_array_ref() {
78        // List of arrayrefs: first row is headers
79        let first_row_arr = first_row_lock.read();
80        let columns: Vec<String> = first_row_arr.iter().map(|v| v.to_string()).collect();
81        let mut cols: Vec<Vec<PerlValue>> = (0..columns.len()).map(|_| Vec::new()).collect();
82        for row_val in rows.iter().skip(1) {
83            if let Some(row_lock) = row_val.as_array_ref() {
84                let row_arr = row_lock.read();
85                for (i, col) in cols.iter_mut().enumerate().take(columns.len()) {
86                    col.push(row_arr.get(i).cloned().unwrap_or(PerlValue::UNDEF));
87                }
88            }
89        }
90        return Ok(PerlValue::dataframe(Arc::new(Mutex::new(PerlDataFrame {
91            columns,
92            cols,
93            group_by: None,
94        }))));
95    }
96
97    Err(PerlError::runtime(
98        "dataframe expects a file path or a list of hashrefs/arrayrefs",
99        0,
100    ))
101}
102
103pub(crate) fn dataframe_from_path(path: &str) -> PerlResult<PerlValue> {
104    let mut rdr = csv::Reader::from_path(path)
105        .map_err(|e| PerlError::runtime(format!("dataframe: {}: {}", path, e), 0))?;
106    let headers: Vec<String> = rdr
107        .headers()
108        .map_err(|e| PerlError::runtime(format!("dataframe: {}: {}", path, e), 0))?
109        .iter()
110        .map(|s| s.to_string())
111        .collect();
112    let ncols = headers.len();
113    let mut cols: Vec<Vec<PerlValue>> = (0..ncols).map(|_| Vec::new()).collect();
114    for rec in rdr.records() {
115        let record = rec.map_err(|e| PerlError::runtime(format!("dataframe: {}", e), 0))?;
116        for (i, col) in cols.iter_mut().enumerate().take(ncols) {
117            let cell = record.get(i).unwrap_or("");
118            col.push(PerlValue::string(cell.to_string()));
119        }
120    }
121    let df = PerlDataFrame {
122        columns: headers,
123        cols,
124        group_by: None,
125    };
126    Ok(PerlValue::dataframe(Arc::new(Mutex::new(df))))
127}
128
129pub(crate) fn csv_read(path: &str) -> PerlResult<PerlValue> {
130    let mut rdr = csv::Reader::from_path(path)
131        .map_err(|e| PerlError::runtime(format!("csv_read: {}: {}", path, e), 0))?;
132    let headers: Vec<String> = rdr
133        .headers()
134        .map_err(|e| PerlError::runtime(format!("csv_read: {}: {}", path, e), 0))?
135        .iter()
136        .map(|s| s.to_string())
137        .collect();
138    let mut rows = Vec::new();
139    for rec in rdr.records() {
140        let record = rec.map_err(|e| PerlError::runtime(format!("csv_read: {}", e), 0))?;
141        let mut map = IndexMap::new();
142        for (i, h) in headers.iter().enumerate() {
143            let cell = record.get(i).unwrap_or("");
144            map.insert(h.clone(), PerlValue::string(cell.to_string()));
145        }
146        rows.push(PerlValue::hash_ref(Arc::new(RwLock::new(map))));
147    }
148    Ok(PerlValue::array(rows))
149}
150
151/// Writes rows as CSV. Each row is a hash or hashref; header row is the union of keys
152/// (first-seen order, then keys from later rows in order).
153pub(crate) fn csv_write(path: &str, rows: &[PerlValue]) -> PerlResult<PerlValue> {
154    let mut header: Vec<String> = Vec::new();
155    let mut seen = std::collections::HashSet::<String>::new();
156    let mut normalized: Vec<IndexMap<String, PerlValue>> = Vec::new();
157
158    for row in rows {
159        let map = hash_like(row)?;
160        for k in map.keys() {
161            if seen.insert(k.clone()) {
162                header.push(k.clone());
163            }
164        }
165        normalized.push(map);
166    }
167
168    let mut wtr = csv::Writer::from_path(path)
169        .map_err(|e| PerlError::runtime(format!("csv_write: {}: {}", path, e), 0))?;
170    wtr.write_record(&header)
171        .map_err(|e| PerlError::runtime(format!("csv_write: {}", e), 0))?;
172    for map in &normalized {
173        let record: Vec<String> = header
174            .iter()
175            .map(|k| map.get(k).map(|v| v.to_string()).unwrap_or_default())
176            .collect();
177        wtr.write_record(&record)
178            .map_err(|e| PerlError::runtime(format!("csv_write: {}", e), 0))?;
179    }
180    wtr.flush()
181        .map_err(|e| PerlError::runtime(format!("csv_write: {}", e), 0))?;
182    Ok(PerlValue::integer(normalized.len() as i64))
183}
184
185fn hash_like(v: &PerlValue) -> PerlResult<IndexMap<String, PerlValue>> {
186    if let Some(h) = v.as_hash_map() {
187        return Ok(h);
188    }
189    if let Some(r) = v.as_hash_ref() {
190        return Ok(r.read().clone());
191    }
192    if let Some(b) = v.as_blessed_ref() {
193        let d = b.data.read();
194        if let Some(h) = d.as_hash_map() {
195            return Ok(h);
196        }
197    }
198    Err(PerlError::runtime(
199        "csv_write: row must be hash or hashref",
200        0,
201    ))
202}
203
204pub(crate) fn sqlite_open(path: &str) -> PerlResult<PerlValue> {
205    let conn = Connection::open(path)
206        .map_err(|e| PerlError::runtime(format!("sqlite: {}: {}", path, e), 0))?;
207    Ok(PerlValue::sqlite_conn(Arc::new(Mutex::new(conn))))
208}
209
210pub(crate) fn sqlite_dispatch(
211    conn: &Arc<Mutex<Connection>>,
212    method: &str,
213    args: &[PerlValue],
214    line: usize,
215) -> PerlResult<PerlValue> {
216    let c = conn.lock();
217    match method {
218        "exec" => {
219            if args.is_empty() {
220                return Err(PerlError::runtime("sqlite->exec needs SQL string", line));
221            }
222            let sql = args[0].to_string();
223            let params: Vec<Value> = args[1..].iter().map(perl_to_sql_value).collect();
224            let n = exec_sql(&c, &sql, &params)?;
225            Ok(PerlValue::integer(n as i64))
226        }
227        "query" => {
228            if args.is_empty() {
229                return Err(PerlError::runtime("sqlite->query needs SQL string", line));
230            }
231            let sql = args[0].to_string();
232            let params: Vec<Value> = args[1..].iter().map(perl_to_sql_value).collect();
233            query_sql(&c, &sql, &params, line)
234        }
235        "last_insert_rowid" => {
236            if !args.is_empty() {
237                return Err(PerlError::runtime(
238                    "sqlite->last_insert_rowid takes no arguments",
239                    line,
240                ));
241            }
242            Ok(PerlValue::integer(c.last_insert_rowid()))
243        }
244        _ => Err(PerlError::runtime(
245            format!("unknown sqlite method: {}", method),
246            line,
247        )),
248    }
249}
250
251fn exec_sql(conn: &Connection, sql: &str, params: &[Value]) -> PerlResult<usize> {
252    conn.execute(sql, rusqlite::params_from_iter(params.iter()))
253        .map_err(|e| PerlError::runtime(format!("sqlite exec: {}", e), 0))
254}
255
256fn query_sql(conn: &Connection, sql: &str, params: &[Value], line: usize) -> PerlResult<PerlValue> {
257    let mut stmt = conn
258        .prepare(sql)
259        .map_err(|e| PerlError::runtime(format!("sqlite query: {}", e), line))?;
260    let col_count = stmt.column_count();
261    let mut col_names = Vec::with_capacity(col_count);
262    for i in 0..col_count {
263        col_names.push(
264            stmt.column_name(i)
265                .map(|s| s.to_string())
266                .unwrap_or_else(|_| format!("col{}", i)),
267        );
268    }
269    let mut rows = stmt
270        .query(rusqlite::params_from_iter(params.iter()))
271        .map_err(|e| PerlError::runtime(format!("sqlite query: {}", e), line))?;
272    let mut rows_out = Vec::new();
273    while let Some(row) = rows
274        .next()
275        .map_err(|e| PerlError::runtime(format!("sqlite query: {}", e), line))?
276    {
277        let mut map = IndexMap::new();
278        for (i, col_name) in col_names.iter().enumerate().take(col_count) {
279            let v = row
280                .get::<_, Value>(i)
281                .map_err(|e| PerlError::runtime(format!("sqlite query: {}", e), line))?;
282            map.insert(col_name.clone(), sqlite_value_to_perl(v));
283        }
284        rows_out.push(PerlValue::hash_ref(Arc::new(RwLock::new(map))));
285    }
286    Ok(PerlValue::array(rows_out))
287}
288
289fn perl_to_sql_value(v: &PerlValue) -> Value {
290    if v.is_undef() {
291        return Value::Null;
292    }
293    if let Some(i) = v.as_integer() {
294        return Value::Integer(i);
295    }
296    if let Some(f) = v.as_float() {
297        return Value::Real(f);
298    }
299    if let Some(s) = v.as_str() {
300        return Value::Text(s);
301    }
302    if let Some(b) = v.as_bytes_arc() {
303        return Value::Blob((*b).clone());
304    }
305    Value::Text(v.to_string())
306}
307
308fn sqlite_value_to_perl(v: Value) -> PerlValue {
309    match v {
310        Value::Null => PerlValue::UNDEF,
311        Value::Integer(i) => PerlValue::integer(i),
312        Value::Real(r) => PerlValue::float(r),
313        Value::Text(s) => PerlValue::string(s),
314        Value::Blob(b) => PerlValue::bytes(Arc::new(b)),
315    }
316}
317
318/// Build a struct instance with defaults evaluated by the interpreter.
319/// Called from interpreter when constructing structs so default expressions can be evaluated.
320pub(crate) fn struct_new_with_defaults(
321    def: &Arc<StructDef>,
322    provided: &[(String, PerlValue)],
323    defaults: &[Option<PerlValue>],
324    line: usize,
325) -> PerlResult<PerlValue> {
326    let mut values = vec![PerlValue::UNDEF; def.fields.len()];
327    for (k, v) in provided {
328        let idx = def.field_index(k).ok_or_else(|| {
329            PerlError::runtime(format!("struct {}: unknown field `{}`", def.name, k), line)
330        })?;
331        let field = &def.fields[idx];
332        field.ty.check_value(v).map_err(|msg| {
333            PerlError::type_error(format!("struct {} field `{}`: {}", def.name, k, msg), line)
334        })?;
335        values[idx] = v.clone();
336    }
337    for (idx, field) in def.fields.iter().enumerate() {
338        if values[idx].is_undef() {
339            if let Some(dv) = defaults.get(idx).and_then(|o| o.as_ref()) {
340                // Skip type check if default is undef (nullable field pattern)
341                if !dv.is_undef() {
342                    field.ty.check_value(dv).map_err(|msg| {
343                        PerlError::type_error(
344                            format!(
345                                "struct {} field `{}` default: {}",
346                                def.name, field.name, msg
347                            ),
348                            line,
349                        )
350                    })?;
351                }
352                values[idx] = dv.clone();
353            } else if field.default.is_none() && !matches!(field.ty, crate::ast::PerlTypeName::Any)
354            {
355                return Err(PerlError::runtime(
356                    format!(
357                        "struct {}: missing field `{}` ({})",
358                        def.name,
359                        field.name,
360                        field.ty.display_name()
361                    ),
362                    line,
363                ));
364            }
365        }
366    }
367    Ok(PerlValue::struct_inst(Arc::new(StructInstance::new(
368        Arc::clone(def),
369        values,
370    ))))
371}
372
373/// GET `url` and return the response body as a UTF-8 string (invalid UTF-8 is lossy).
374pub(crate) fn fetch(url: &str) -> PerlResult<PerlValue> {
375    let s = http_get_body(url)?;
376    Ok(PerlValue::string(s))
377}
378
379/// GET `url`, parse JSON, map to [`PerlValue`] (objects → `HashRef`, arrays → `Array`, etc.).
380pub(crate) fn fetch_json(url: &str) -> PerlResult<PerlValue> {
381    let s = http_get_body(url)?;
382    let v: JsonValue = serde_json::from_str(&s)
383        .map_err(|e| PerlError::runtime(format!("fetch_json: {}", e), 0))?;
384    Ok(json_to_perl(v))
385}
386
387fn http_get_body(url: &str) -> PerlResult<String> {
388    ureq::get(url)
389        .call()
390        .map_err(|e| PerlError::runtime(format!("fetch: {}", e), 0))?
391        .into_string()
392        .map_err(|e| PerlError::runtime(format!("fetch: {}", e), 0))
393}
394
395fn perl_hash_lookup(v: &PerlValue, key: &str) -> Option<PerlValue> {
396    v.hash_get(key)
397        .or_else(|| v.as_hash_ref().and_then(|r| r.read().get(key).cloned()))
398}
399
400fn perl_opt_lookup(opts: Option<&PerlValue>, key: &str) -> Option<PerlValue> {
401    let o = opts?;
402    perl_hash_lookup(o, key)
403}
404
405fn perl_opt_bool(opts: Option<&PerlValue>, key: &str) -> bool {
406    perl_opt_lookup(opts, key).is_some_and(|v| v.is_true())
407}
408
409fn perl_opt_u64(opts: Option<&PerlValue>, key: &str) -> Option<u64> {
410    perl_opt_lookup(opts, key).map(|v| v.to_int().max(0) as u64)
411}
412
413fn body_bytes_from_perl(v: &PerlValue) -> Vec<u8> {
414    if let Some(b) = v.as_bytes_arc() {
415        return b.as_ref().clone();
416    }
417    v.to_string().into_bytes()
418}
419
420fn headers_map_has_content_type(headers_val: &PerlValue) -> bool {
421    if let Some(m) = headers_val.as_hash_map() {
422        return m.keys().any(|k| k.eq_ignore_ascii_case("content-type"));
423    }
424    if let Some(r) = headers_val.as_hash_ref() {
425        return r
426            .read()
427            .keys()
428            .any(|k| k.eq_ignore_ascii_case("content-type"));
429    }
430    false
431}
432
433fn apply_request_headers(
434    mut req: ureq::Request,
435    headers_val: &PerlValue,
436) -> PerlResult<ureq::Request> {
437    let pairs: Vec<(String, String)> = if let Some(m) = headers_val.as_hash_map() {
438        m.iter().map(|(k, v)| (k.clone(), v.to_string())).collect()
439    } else if let Some(r) = headers_val.as_hash_ref() {
440        r.read()
441            .iter()
442            .map(|(k, v)| (k.clone(), v.to_string()))
443            .collect()
444    } else {
445        return Err(PerlError::runtime(
446            "http_request: headers must be a hash or hashref",
447            0,
448        ));
449    };
450    for (k, v) in pairs {
451        req = req.set(&k, &v);
452    }
453    Ok(req)
454}
455
456/// Full HTTP request: `opts` hash(ref) keys: `method` (default GET), `headers`, `body`, `json`
457/// (encodes body, sets `Content-Type` unless already in `headers`), `timeout` / `timeout_secs`
458/// (omit for 30s; `0` disables client timeout), `binary_response` (body as `BYTES` instead of decoded string).
459///
460/// Returns a hashref: `status`, `status_text`, `headers` (hashref, lowercased names), `body`.
461pub(crate) fn http_request(url: &str, opts: Option<&PerlValue>) -> PerlResult<PerlValue> {
462    let method = perl_opt_lookup(opts, "method")
463        .map(|v| v.to_string())
464        .filter(|s| !s.is_empty())
465        .unwrap_or_else(|| "GET".to_string());
466    let method_uc = method.to_ascii_uppercase();
467    let timeout_secs = perl_opt_u64(opts, "timeout_secs").or_else(|| perl_opt_u64(opts, "timeout"));
468    let binary_response = perl_opt_bool(opts, "binary_response");
469
470    let mut req = ureq::request(method_uc.as_str(), url);
471    match timeout_secs {
472        None => {
473            req = req.timeout(Duration::from_secs(30));
474        }
475        Some(0) => {}
476        Some(n) => {
477            req = req.timeout(Duration::from_secs(n));
478        }
479    }
480
481    if let Some(hv) = opts.and_then(|o| perl_hash_lookup(o, "headers")) {
482        req = apply_request_headers(req, &hv)?;
483    }
484
485    let mut body: Vec<u8> = Vec::new();
486    if let Some(o) = opts {
487        if let Some(jv) = perl_hash_lookup(o, "json") {
488            let jstr = json_encode(&jv)?;
489            if let Some(hv) = perl_hash_lookup(o, "headers") {
490                if !headers_map_has_content_type(&hv) {
491                    req = req.set("Content-Type", "application/json; charset=utf-8");
492                }
493            } else {
494                req = req.set("Content-Type", "application/json; charset=utf-8");
495            }
496            body = jstr.into_bytes();
497        } else if let Some(bv) = perl_hash_lookup(o, "body") {
498            body = body_bytes_from_perl(&bv);
499        }
500    }
501
502    let resp = if body.is_empty() {
503        req.call()
504    } else {
505        req.send_bytes(&body)
506    }
507    .map_err(|e| PerlError::runtime(format!("http_request: {}", e), 0))?;
508
509    let status = resp.status();
510    let status_text = resp.status_text().to_string();
511    let mut hdr_map = IndexMap::new();
512    let mut names = resp.headers_names();
513    names.sort();
514    names.dedup();
515    for n in names {
516        let vals: Vec<&str> = resp.all(&n);
517        if !vals.is_empty() {
518            hdr_map.insert(n, PerlValue::string(vals.join(", ")));
519        }
520    }
521    let headers_ref = PerlValue::hash_ref(Arc::new(RwLock::new(hdr_map)));
522
523    let body_val = if binary_response {
524        let mut buf = Vec::new();
525        resp.into_reader()
526            .read_to_end(&mut buf)
527            .map_err(|e| PerlError::runtime(format!("http_request: body read: {}", e), 0))?;
528        PerlValue::bytes(Arc::new(buf))
529    } else {
530        let s = resp
531            .into_string()
532            .map_err(|e| PerlError::runtime(format!("http_request: body: {}", e), 0))?;
533        PerlValue::string(s)
534    };
535
536    let mut out = IndexMap::new();
537    out.insert("status".into(), PerlValue::integer(status as i64));
538    out.insert("status_text".into(), PerlValue::string(status_text));
539    out.insert("headers".into(), headers_ref);
540    out.insert("body".into(), body_val);
541    Ok(PerlValue::hash_ref(Arc::new(RwLock::new(out))))
542}
543
544/// Parse JSON from the `body` field of an [`http_request`] result hashref.
545pub(crate) fn http_response_json_body(res: &PerlValue) -> PerlResult<PerlValue> {
546    let body = perl_hash_lookup(res, "body")
547        .ok_or_else(|| PerlError::runtime("fetch_json: http response missing body", 0))?;
548    let s = if let Some(b) = body.as_bytes_arc() {
549        String::from_utf8_lossy(b.as_ref()).into_owned()
550    } else {
551        body.to_string()
552    };
553    json_decode(&s)
554}
555
556/// Serialize a [`PerlValue`] to a JSON string (arrays, hashes, refs, structs, scalars; not code/refs/IO).
557pub(crate) fn json_encode(v: &PerlValue) -> PerlResult<String> {
558    let j = perl_to_json_value(v)?;
559    serde_json::to_string(&j).map_err(|e| PerlError::runtime(format!("json_encode: {}", e), 0))
560}
561
562/// Parse a JSON string into [`PerlValue`] (same mapping as [`fetch_json`]).
563pub(crate) fn json_decode(s: &str) -> PerlResult<PerlValue> {
564    let v: JsonValue = serde_json::from_str(s.trim())
565        .map_err(|e| PerlError::runtime(format!("json_decode: {}", e), 0))?;
566    Ok(json_to_perl(v))
567}
568
569/// Run a [jq](https://jqlang.org/)-syntax filter (via [jaq](https://github.com/01mf02/jaq)) on JSON
570/// derived from `data` (same encodable shapes as [`json_encode`]).
571///
572/// Returns `undef` if the filter yields no values, a single Perl value if it yields one output,
573/// or an array of values if it yields more than one (e.g. `.items[]`).
574pub(crate) fn json_jq(data: &PerlValue, filter_src: &str) -> PerlResult<PerlValue> {
575    let j = perl_to_json_value(data)?;
576    let input: jaq_json::Val = serde_json::from_value(j)
577        .map_err(|e| PerlError::runtime(format!("json_jq: could not convert input: {}", e), 0))?;
578
579    let arena = jaq_core::load::Arena::default();
580    let defs = jaq_core::defs()
581        .chain(jaq_std::defs())
582        .chain(jaq_json::defs());
583    let loader = jaq_core::load::Loader::new(defs);
584    let file = jaq_core::load::File {
585        code: filter_src,
586        path: (),
587    };
588    let modules = loader
589        .load(&arena, file)
590        .map_err(|e| PerlError::runtime(format!("json_jq: parse/load: {:?}", e), 0))?;
591
592    type JData = JustLut<jaq_json::Val>;
593    let filter = jaq_core::Compiler::default()
594        .with_funs(
595            jaq_core::funs::<JData>()
596                .chain(jaq_std::funs::<JData>())
597                .chain(jaq_json::funs::<JData>()),
598        )
599        .compile(modules)
600        .map_err(|e| PerlError::runtime(format!("json_jq: compile: {:?}", e), 0))?;
601
602    let ctx = jaq_core::Ctx::<JData>::new(&filter.lut, jaq_core::Vars::new([]));
603    let mut results = Vec::new();
604    for x in filter.id.run((ctx, input)) {
605        match jaq_core::unwrap_valr(x) {
606            Ok(v) => results.push(jaq_json_val_to_perl(v)?),
607            Err(e) => {
608                return Err(PerlError::runtime(format!("json_jq: {}", e), 0));
609            }
610        }
611    }
612
613    match results.len() {
614        0 => Ok(PerlValue::UNDEF),
615        1 => Ok(results.pop().expect("one")),
616        _ => Ok(PerlValue::array(results)),
617    }
618}
619
620fn jaq_json_val_to_perl(v: jaq_json::Val) -> PerlResult<PerlValue> {
621    use jaq_json::Val as Jv;
622    match v {
623        Jv::Null => Ok(PerlValue::UNDEF),
624        Jv::Bool(b) => Ok(PerlValue::integer(i64::from(b))),
625        Jv::Num(n) => jaq_num_to_perl(n),
626        Jv::BStr(b) => Ok(PerlValue::string(String::from_utf8_lossy(&b).into_owned())),
627        Jv::TStr(b) => Ok(PerlValue::string(String::from_utf8_lossy(&b).into_owned())),
628        Jv::Arr(a) => {
629            let v = a.as_ref();
630            let mut out = Vec::with_capacity(v.len());
631            for x in v.iter() {
632                out.push(jaq_json_val_to_perl(x.clone())?);
633            }
634            Ok(PerlValue::array(out))
635        }
636        Jv::Obj(o) => {
637            let mut map = IndexMap::new();
638            for (k, val) in o.iter() {
639                map.insert(k.to_string(), jaq_json_val_to_perl(val.clone())?);
640            }
641            Ok(PerlValue::hash_ref(Arc::new(RwLock::new(map))))
642        }
643    }
644}
645
646fn jaq_num_to_perl(n: jaq_json::Num) -> PerlResult<PerlValue> {
647    use jaq_json::Num as Jn;
648    match n {
649        Jn::Int(i) => Ok(PerlValue::integer(i as i64)),
650        Jn::Float(f) => Ok(PerlValue::float(f)),
651        Jn::BigInt(r) => {
652            let bi = (*r).clone();
653            if let Some(i) = bi.to_i64() {
654                Ok(PerlValue::integer(i))
655            } else if let Some(f) = bi.to_f64() {
656                Ok(PerlValue::float(f))
657            } else {
658                Ok(PerlValue::string(bi.to_string()))
659            }
660        }
661        Jn::Dec(s) => {
662            let f: f64 = s.parse().unwrap_or(f64::NAN);
663            Ok(PerlValue::float(f))
664        }
665    }
666}
667
668pub(crate) fn perl_to_json_value(v: &PerlValue) -> PerlResult<JsonValue> {
669    if v.is_undef() {
670        return Ok(JsonValue::Null);
671    }
672    if let Some(n) = v.as_integer() {
673        return Ok(JsonValue::Number(n.into()));
674    }
675    if let Some(f) = v.as_float() {
676        return serde_json::Number::from_f64(f)
677            .map(JsonValue::Number)
678            .ok_or_else(|| PerlError::runtime("json_encode: non-finite float", 0));
679    }
680    if crate::nanbox::is_raw_float_bits(v.0) {
681        let f = f64::from_bits(v.0);
682        return serde_json::Number::from_f64(f)
683            .map(JsonValue::Number)
684            .ok_or_else(|| PerlError::runtime("json_encode: non-finite float", 0));
685    }
686    if let Some(a) = v.as_array_vec() {
687        let mut out = Vec::with_capacity(a.len());
688        for x in &a {
689            out.push(perl_to_json_value(x)?);
690        }
691        return Ok(JsonValue::Array(out));
692    }
693    if let Some(h) = v.as_hash_map() {
694        let mut m = serde_json::Map::new();
695        for (k, val) in h.iter() {
696            m.insert(k.clone(), perl_to_json_value(val)?);
697        }
698        return Ok(JsonValue::Object(m));
699    }
700    if let Some(r) = v.as_array_ref() {
701        let g = r.read();
702        let mut out = Vec::with_capacity(g.len());
703        for x in g.iter() {
704            out.push(perl_to_json_value(x)?);
705        }
706        return Ok(JsonValue::Array(out));
707    }
708    if let Some(r) = v.as_hash_ref() {
709        let g = r.read();
710        let mut m = serde_json::Map::new();
711        for (k, val) in g.iter() {
712            m.insert(k.clone(), perl_to_json_value(val)?);
713        }
714        return Ok(JsonValue::Object(m));
715    }
716    if let Some(r) = v.as_scalar_ref() {
717        return perl_to_json_value(&r.read());
718    }
719    if let Some(a) = v.as_atomic_arc() {
720        return perl_to_json_value(&a.lock().clone());
721    }
722    if let Some(s) = v.as_str() {
723        return Ok(JsonValue::String(s));
724    }
725    if let Some(b) = v.as_bytes_arc() {
726        return Ok(JsonValue::String(String::from_utf8_lossy(&b).into_owned()));
727    }
728    if let Some(si) = v.as_struct_inst() {
729        let mut m = serde_json::Map::new();
730        let values = si.get_values();
731        for (i, field) in si.def.fields.iter().enumerate() {
732            if let Some(fv) = values.get(i) {
733                m.insert(field.name.clone(), perl_to_json_value(fv)?);
734            }
735        }
736        return Ok(JsonValue::Object(m));
737    }
738    if let Some(b) = v.as_blessed_ref() {
739        let inner = b.data.read().clone();
740        return perl_to_json_value(&inner);
741    }
742    if let Some(vals) = v
743        .with_heap(|h| match h {
744            HeapObject::Set(s) => Some(s.values().cloned().collect::<Vec<_>>()),
745            _ => None,
746        })
747        .flatten()
748    {
749        let mut out = Vec::with_capacity(vals.len());
750        for x in vals {
751            out.push(perl_to_json_value(&x)?);
752        }
753        return Ok(JsonValue::Array(out));
754    }
755    if let Some(vals) = v
756        .with_heap(|h| match h {
757            HeapObject::Deque(d) => Some(d.lock().iter().cloned().collect::<Vec<_>>()),
758            _ => None,
759        })
760        .flatten()
761    {
762        let mut out = Vec::with_capacity(vals.len());
763        for x in vals {
764            out.push(perl_to_json_value(&x)?);
765        }
766        return Ok(JsonValue::Array(out));
767    }
768
769    if let Some(df) = v.as_dataframe() {
770        let g = df.lock();
771        let n = g.nrows();
772        let mut rows = Vec::with_capacity(n);
773        for r in 0..n {
774            let mut m = serde_json::Map::new();
775            for (i, col) in g.columns.iter().enumerate() {
776                m.insert(col.clone(), perl_to_json_value(&g.cols[i][r])?);
777            }
778            rows.push(JsonValue::Object(m));
779        }
780        return Ok(JsonValue::Array(rows));
781    }
782
783    Err(PerlError::runtime(
784        format!(
785            "json_encode: value cannot be encoded as JSON ({})",
786            v.type_name()
787        ),
788        0,
789    ))
790}
791
792fn json_to_perl(v: JsonValue) -> PerlValue {
793    match v {
794        JsonValue::Null => PerlValue::UNDEF,
795        JsonValue::Bool(b) => PerlValue::integer(i64::from(b)),
796        JsonValue::Number(n) => {
797            if let Some(i) = n.as_i64() {
798                PerlValue::integer(i)
799            } else if let Some(u) = n.as_u64() {
800                PerlValue::integer(u as i64)
801            } else {
802                PerlValue::float(n.as_f64().unwrap_or(0.0))
803            }
804        }
805        JsonValue::String(s) => PerlValue::string(s),
806        JsonValue::Array(a) => PerlValue::array(a.into_iter().map(json_to_perl).collect()),
807        JsonValue::Object(o) => {
808            let mut map = IndexMap::new();
809            for (k, v) in o {
810                map.insert(k, json_to_perl(v));
811            }
812            PerlValue::hash_ref(Arc::new(RwLock::new(map)))
813        }
814    }
815}
816
817#[cfg(test)]
818mod http_json_tests {
819    use super::*;
820
821    #[test]
822    fn json_to_perl_object_hashref() {
823        let v: JsonValue = serde_json::from_str(r#"{"name":"a","n":1}"#).unwrap();
824        let p = json_to_perl(v);
825        let r = p.as_hash_ref().expect("expected HashRef");
826        let g = r.read();
827        assert_eq!(g.get("name").unwrap().to_string(), "a");
828        assert_eq!(g.get("n").unwrap().to_int(), 1);
829    }
830
831    #[test]
832    fn json_to_perl_array() {
833        let v: JsonValue = serde_json::from_str(r#"[1,"x",null]"#).unwrap();
834        let p = json_to_perl(v);
835        let a = p.as_array_vec().expect("expected Array");
836        assert_eq!(a.len(), 3);
837        assert_eq!(a[0].to_int(), 1);
838        assert_eq!(a[1].to_string(), "x");
839        assert!(a[2].is_undef());
840    }
841
842    #[test]
843    fn json_encode_decode_roundtrip() {
844        let p = PerlValue::array(vec![
845            PerlValue::integer(1),
846            PerlValue::string("x".into()),
847            PerlValue::UNDEF,
848        ]);
849        let s = json_encode(&p).expect("encode");
850        let back = json_decode(&s).expect("decode");
851        let a = back.as_array_vec().expect("array");
852        assert_eq!(a.len(), 3);
853        assert_eq!(a[0].to_int(), 1);
854        assert_eq!(a[1].to_string(), "x");
855        assert!(a[2].is_undef());
856    }
857
858    #[test]
859    fn json_encode_hash_roundtrip() {
860        let mut m = IndexMap::new();
861        m.insert("a".into(), PerlValue::integer(2));
862        let p = PerlValue::hash(m);
863        let s = json_encode(&p).expect("encode");
864        assert!(s.contains("\"a\""));
865        let back = json_decode(&s).expect("decode");
866        let h = back.as_hash_ref().expect("hashref");
867        assert_eq!(h.read().get("a").unwrap().to_int(), 2);
868    }
869
870    #[test]
871    fn json_jq_field_select() {
872        let p = json_decode(r#"{"a":1,"b":{"c":3}}"#).unwrap();
873        let out = json_jq(&p, ".b.c").unwrap();
874        assert_eq!(out.to_int(), 3);
875    }
876
877    #[test]
878    fn json_jq_map_select_multiple_yields_array() {
879        let p = json_decode(r#"[1,2,3,4]"#).unwrap();
880        let out = json_jq(&p, "map(select(. > 2))").unwrap();
881        let a = out.as_array_vec().expect("array");
882        assert_eq!(a.len(), 2);
883        assert_eq!(a[0].to_int(), 3);
884        assert_eq!(a[1].to_int(), 4);
885    }
886
887    #[test]
888    fn test_dataframe_from_path() {
889        let tmp = std::env::temp_dir().join(format!("test_df_{}.csv", std::process::id()));
890        let csv_data = "id,name,val\n1,alice,10.5\n2,bob,20.0\n";
891        std::fs::write(&tmp, csv_data).expect("write csv");
892
893        let df_val = dataframe_from_path(tmp.to_str().unwrap()).expect("dataframe_from_path");
894        let df_lock = df_val.as_dataframe().expect("as_dataframe");
895        let df = df_lock.lock();
896
897        assert_eq!(df.columns, vec!["id", "name", "val"]);
898        assert_eq!(df.cols.len(), 3);
899        assert_eq!(df.cols[0][0].to_string(), "1");
900        assert_eq!(df.cols[1][1].to_string(), "bob");
901        assert_eq!(df.cols[2][0].to_string(), "10.5");
902
903        let _ = std::fs::remove_file(&tmp);
904    }
905}