nx_value_parse_csv.nx source
↩ module page · 513 lines · 18418 B
1// nx_value_parse_csv.nx -- bytes -> NxValue (ARRAY of OBJECTs).
2//
3// module: nishi-core.data.value_parse_csv
4// depends: nishi-core.io.syscalls, nishi-core.data.value
5// disk_kb: 5
6// capability: CORE_IO
7// wired_status: FULLY_WIRED
8//
9// license_tier: PUBLIC_NISHI_SUBSTRATE
10// genealogy_id: rfc_4180_csv_canonical +
11// postel_robustness_principle +
12// nishi_build_the_system_cardinal_2026
13//
14// Brick #2b of the bits-up ingestion stack. Parallel to
15// nx_value_parse_json but for delimiter-separated values.
16//
17// Input: CSV/TSV bytes (RFC 4180-ish; tolerant to BOM, CRLF/LF, and
18// unterminated final row).
19// Output: NxValue of kind ARRAY whose items are OBJECTs.
20// Each OBJECT's keys come from the header row;
21// each OBJECT's values are STRING (typing happens later).
22//
23// Configurable via NxCsvOpts:
24// delimiter: byte; default 0x2C (',') -- pass 0x09 ('\t') for TSV
25// quote_char: byte; default 0x22 ('"'); 0 to disable quote handling
26// has_header: 1 = first row is field names (default)
27// 0 = synthesize "col_0","col_1",... keys
28// max_rows: hard cap on data rows parsed (per JPL Rule 2)
29//
30// Composition: nx_ingest_batch reads bytes, hands to either
31// nx_value_parse_json OR nx_value_parse_csv based on extension --
32// same downstream NxValue tree, same flat emit.
33
34import "syscalls.nx"
35import "nx_value.nx"
36
37// ===== Verdict ====================================================
38
39const NX_CSV_OK: i64 = 1
40const NX_CSV_BAD_ARGS: i64 = 2
41const NX_CSV_NO_HEADER: i64 = 3
42const NX_CSV_TOO_MANY_FIELDS: i64 = 4
43const NX_CSV_UNTERMINATED_QTE: i64 = 5
44const NX_CSV_MAX_ROWS_HIT: i64 = 6 // not an error -- truncated
45
46func nx_csv_verdict_name(v: i64) -> *u8 {
47 if v == NX_CSV_OK { return "OK" }
48 if v == NX_CSV_BAD_ARGS { return "BAD_ARGS" }
49 if v == NX_CSV_NO_HEADER { return "NO_HEADER" }
50 if v == NX_CSV_TOO_MANY_FIELDS { return "TOO_MANY_FIELDS" }
51 if v == NX_CSV_UNTERMINATED_QTE { return "UNTERMINATED_QUOTE" }
52 if v == NX_CSV_MAX_ROWS_HIT { return "MAX_ROWS_HIT" }
53 return "UNKNOWN"
54}
55
56// ===== Options ====================================================
57
58struct NxCsvOpts {
59 delimiter: i64,
60 quote_char: i64,
61 has_header: i64,
62 max_rows: i64,
63 max_fields_per_row: i64,
64}
65
66const NX_CSV_OPTS_BYTES: i64 = 40
67
68func nx_csv_opts_default(out: *NxCsvOpts) {
69 out.delimiter = 0x2C // ','
70 out.quote_char = 0x22 // '"'
71 out.has_header = 1
72 out.max_rows = 1000000
73 out.max_fields_per_row = 4096
74}
75
76func nx_csv_opts_tsv(out: *NxCsvOpts) {
77 out.delimiter = 0x09 // '\t'
78 out.quote_char = 0x22 // '"' (rare in TSV but allowed)
79 out.has_header = 1
80 out.max_rows = 1000000
81 out.max_fields_per_row = 4096
82}
83
84// ===== Parser state ===============================================
85
86struct NxCsvParse {
87 src: *u8,
88 len: i64,
89 pos: i64,
90 verdict: i64,
91 opts: *NxCsvOpts,
92 // Scratch buffer for an in-place unquote (only used when a
93 // field contains "" escapes).
94 scratch: *u8,
95 scratch_cap: i64,
96}
97
98const NX_CSV_PARSE_BYTES: i64 = 56
99
100// ===== Skip optional BOM ==========================================
101
102func nx_csv_skip_bom(p: *NxCsvParse) {
103 if p.len < 3 { return }
104 let b0: i64 = p.src[0] as i64
105 let b1: i64 = p.src[1] as i64
106 let b2: i64 = p.src[2] as i64
107 if b0 == 0xEF {
108 if b1 == 0xBB {
109 if b2 == 0xBF { p.pos = 3 }
110 }
111 }
112}
113
114// ===== Field extraction ===========================================
115//
116// Reads one field starting at p.pos. On return:
117// *out_ptr / *out_len = span pointer + length
118// *out_field_end_kind:
119// 1 = delimiter ahead (more fields in this row)
120// 2 = line end (CR / LF) -- row complete
121// 3 = end of input
122//
123// Quoted fields: if first non-empty char is quote_char, consume up
124// to the matching closing quote (with "" escape collapse via
125// scratch buffer when needed). Otherwise raw until delimiter or
126// line end.
127
128func nx_csv_is_line_break(c: i64) -> i64 {
129 if c == 0x0A { return 1 }
130 if c == 0x0D { return 1 }
131 return 0
132}
133
134const NX_CSV_END_DELIMITER: i64 = 1
135const NX_CSV_END_LINE: i64 = 2
136const NX_CSV_END_INPUT: i64 = 3
137
138func nx_csv_read_field(
139 p: *NxCsvParse,
140 out_ptr: **u8,
141 out_len: *i64,
142 out_end_kind: *i64
143) -> i64 {
144 *out_ptr = 0 as *u8
145 *out_len = 0
146 *out_end_kind = NX_CSV_END_INPUT
147 if p.pos >= p.len { return 0 }
148
149 let delim: i64 = p.opts.delimiter
150 let quote: i64 = p.opts.quote_char
151 let start: i64 = p.pos
152
153 // ----- Quoted field --------------------------------------------
154 if quote != 0 {
155 if p.src[p.pos] as i64 == quote {
156 p.pos = p.pos + 1
157 let content_start: i64 = p.pos
158 // Walk, watching for "" (escaped quote inside).
159 var iter: i64 = 0
160 var verdict: i64 = 0
161 var has_escape: i64 = 0
162 while verdict == 0 && iter < 1048576 {
163 if p.pos >= p.len {
164 p.verdict = NX_CSV_UNTERMINATED_QTE
165 return 0
166 }
167 let c: i64 = p.src[p.pos] as i64
168 // Flat single-pass decision: advance / close / escape
169 var lookahead_is_quote: i64 = 0
170 if c == quote {
171 if p.pos + 1 < p.len {
172 if p.src[p.pos + 1] as i64 == quote { lookahead_is_quote = 1 }
173 }
174 }
175 if c != quote {
176 p.pos = p.pos + 1
177 iter = iter + 1
178 }
179 if c == quote {
180 if lookahead_is_quote == 1 {
181 has_escape = 1
182 p.pos = p.pos + 2
183 iter = iter + 1
184 }
185 if lookahead_is_quote == 0 { verdict = 1 }
186 }
187 }
188 // p.pos at closing quote
189 let raw_end: i64 = p.pos
190 p.pos = p.pos + 1 // past closing quote
191
192 if has_escape == 0 {
193 *out_ptr = (p.src as i64 + content_start) as *u8
194 *out_len = raw_end - content_start
195 }
196 if has_escape == 1 {
197 // Decode "" -> " into scratch buffer.
198 let need: i64 = raw_end - content_start
199 if p.scratch_cap < need {
200 p.scratch = sys_mmap(need + 16)
201 p.scratch_cap = need + 16
202 }
203 var src_i: i64 = content_start
204 var dst_i: i64 = 0
205 var de_iter: i64 = 0
206 var de_verdict: i64 = 0
207 while de_verdict == 0 && de_iter < 1048576 {
208 if src_i >= raw_end { de_verdict = 1 }
209 if de_verdict == 0 {
210 let c2: i64 = p.src[src_i] as i64
211 if c2 == quote {
212 // "" -> " (outer walk guarantees pairs inside raw_end)
213 p.scratch[dst_i] = quote as u8
214 dst_i = dst_i + 1
215 src_i = src_i + 2
216 de_iter = de_iter + 1
217 }
218 if c2 != quote {
219 p.scratch[dst_i] = c2 as u8
220 dst_i = dst_i + 1
221 src_i = src_i + 1
222 de_iter = de_iter + 1
223 }
224 }
225 }
226 *out_ptr = p.scratch
227 *out_len = dst_i
228 }
229 }
230 }
231
232 // ----- Unquoted field (or after closing quote) ----------------
233 if *out_ptr == 0 as *u8 {
234 // Raw scan
235 let s2: i64 = p.pos
236 var s_iter: i64 = 0
237 var s_verdict: i64 = 0
238 while s_verdict == 0 && s_iter < 1048576 {
239 if p.pos >= p.len { s_verdict = 1 }
240 if s_verdict == 0 {
241 let c: i64 = p.src[p.pos] as i64
242 if c == delim { s_verdict = 1 }
243 if s_verdict == 0 {
244 if nx_csv_is_line_break(c) == 1 { s_verdict = 1 }
245 }
246 if s_verdict == 0 {
247 p.pos = p.pos + 1
248 s_iter = s_iter + 1
249 }
250 }
251 }
252 *out_ptr = (p.src as i64 + s2) as *u8
253 *out_len = p.pos - s2
254 }
255
256 // ----- Determine end-kind -------------------------------------
257 if p.pos >= p.len {
258 *out_end_kind = NX_CSV_END_INPUT
259 return 1
260 }
261 let cend: i64 = p.src[p.pos] as i64
262 if cend == delim {
263 p.pos = p.pos + 1
264 *out_end_kind = NX_CSV_END_DELIMITER
265 return 1
266 }
267 if nx_csv_is_line_break(cend) == 1 {
268 // Consume CR + optional LF (CRLF -> one line break).
269 if cend == 0x0D {
270 p.pos = p.pos + 1
271 if p.pos < p.len {
272 if p.src[p.pos] as i64 == 0x0A { p.pos = p.pos + 1 }
273 }
274 }
275 if cend == 0x0A { p.pos = p.pos + 1 }
276 *out_end_kind = NX_CSV_END_LINE
277 return 1
278 }
279 *out_end_kind = NX_CSV_END_INPUT
280 return 1
281}
282
283// ===== Read one row into parallel field arrays ====================
284//
285// Returns count of fields read. Sets *out_row_end_kind to either
286// NX_CSV_END_LINE or NX_CSV_END_INPUT depending on what terminated
287// the row. Bounded NX_CSV_*_max_fields_per_row.
288
289func nx_csv_read_row(
290 p: *NxCsvParse,
291 fields_ptr: **u8,
292 fields_len: *i64,
293 cap: i64,
294 out_row_end: *i64
295) -> i64 {
296 var n: i64 = 0
297 var iter: i64 = 0
298 var verdict: i64 = 0
299 *out_row_end = NX_CSV_END_INPUT
300 while verdict == 0 && iter < cap {
301 var fp: *u8 = 0 as *u8
302 var fl: i64 = 0
303 var ek: i64 = 0
304 let got: i64 = nx_csv_read_field(p, &fp, &fl, &ek)
305 if p.verdict != NX_CSV_OK { return 0 }
306 if got == 0 { verdict = 1 }
307 if verdict == 0 {
308 fields_ptr[n] = fp
309 fields_len[n] = fl
310 n = n + 1
311 if ek == NX_CSV_END_DELIMITER {
312 iter = iter + 1
313 }
314 if ek != NX_CSV_END_DELIMITER {
315 *out_row_end = ek
316 verdict = 1
317 }
318 }
319 }
320 if iter >= cap { p.verdict = NX_CSV_TOO_MANY_FIELDS }
321 return n
322}
323
324// ===== Top-level entry point ======================================
325//
326// Returns *NxValue (ARRAY of OBJECTs) or NULL on error.
327
328func nx_value_parse_csv(
329 src: *u8,
330 len: i64,
331 opts: *NxCsvOpts,
332 out_verdict: *i64
333) -> *NxValue {
334 *out_verdict = NX_CSV_BAD_ARGS
335 if src == 0 as *u8 { return 0 as *NxValue }
336 if len <= 0 { return 0 as *NxValue }
337
338 let opts_raw: *u8 = sys_mmap(NX_CSV_OPTS_BYTES)
339 let opts_use: *NxCsvOpts = opts_raw as *NxCsvOpts
340 if opts == 0 as *NxCsvOpts { nx_csv_opts_default(opts_use) }
341 if opts != 0 as *NxCsvOpts {
342 opts_use.delimiter = opts.delimiter
343 opts_use.quote_char = opts.quote_char
344 opts_use.has_header = opts.has_header
345 opts_use.max_rows = opts.max_rows
346 opts_use.max_fields_per_row = opts.max_fields_per_row
347 }
348
349 let p_raw: *u8 = sys_mmap(NX_CSV_PARSE_BYTES)
350 let p: *NxCsvParse = p_raw as *NxCsvParse
351 p.src = src
352 p.len = len
353 p.pos = 0
354 p.verdict = NX_CSV_OK
355 p.opts = opts_use
356 p.scratch = 0 as *u8
357 p.scratch_cap = 0
358
359 nx_csv_skip_bom(p)
360
361 // ----- Read header row ---------------------------------------
362 let max_f: i64 = opts_use.max_fields_per_row
363 let hdr_ptr_raw: *u8 = sys_mmap(max_f * 8)
364 let hdr_ptr: **u8 = hdr_ptr_raw as **u8
365 let hdr_len_raw: *u8 = sys_mmap(max_f * 8)
366 let hdr_len: *i64 = hdr_len_raw as *i64
367
368 var n_fields: i64 = 0
369 if opts_use.has_header == 1 {
370 var hek: i64 = 0
371 n_fields = nx_csv_read_row(p, hdr_ptr, hdr_len, max_f, &hek)
372 if p.verdict != NX_CSV_OK {
373 *out_verdict = p.verdict
374 return 0 as *NxValue
375 }
376 if n_fields == 0 {
377 *out_verdict = NX_CSV_NO_HEADER
378 return 0 as *NxValue
379 }
380 }
381
382 // ----- Read data rows -----------------------------------------
383 let max_r: i64 = opts_use.max_rows
384 let rows_raw: *u8 = sys_mmap(max_r * 8)
385 let rows: **NxValue = rows_raw as **NxValue
386 var n_rows: i64 = 0
387
388 let row_ptr_raw: *u8 = sys_mmap(max_f * 8)
389 let row_ptr: **u8 = row_ptr_raw as **u8
390 let row_len_raw: *u8 = sys_mmap(max_f * 8)
391 let row_len: *i64 = row_len_raw as *i64
392
393 var r_iter: i64 = 0
394 var r_verdict: i64 = 0
395 while r_verdict == 0 && r_iter < max_r {
396 if p.pos >= p.len { r_verdict = 1 }
397 if r_verdict == 0 {
398 // Skip empty lines.
399 if nx_csv_is_line_break(p.src[p.pos] as i64) == 1 {
400 if p.src[p.pos] as i64 == 0x0D {
401 p.pos = p.pos + 1
402 if p.pos < p.len {
403 if p.src[p.pos] as i64 == 0x0A { p.pos = p.pos + 1 }
404 }
405 }
406 if p.pos < p.len {
407 if p.src[p.pos] as i64 == 0x0A { p.pos = p.pos + 1 }
408 }
409 }
410 if p.pos >= p.len { r_verdict = 1 }
411 if r_verdict == 0 {
412 var rek: i64 = 0
413 let m: i64 = nx_csv_read_row(p, row_ptr, row_len, max_f, &rek)
414 if p.verdict != NX_CSV_OK {
415 *out_verdict = p.verdict
416 return 0 as *NxValue
417 }
418 if m == 0 { r_verdict = 1 }
419 if r_verdict == 0 {
420 // Build OBJECT for this row.
421 var nk: i64 = m
422 if opts_use.has_header == 1 { nk = n_fields }
423 // Synthesize numeric keys when no header.
424 var keys_use: **u8 = hdr_ptr
425 var klens_use: *i64 = hdr_len
426 if opts_use.has_header == 0 {
427 let sk_raw: *u8 = sys_mmap(m * 8)
428 let sk: **u8 = sk_raw as **u8
429 let skl_raw: *u8 = sys_mmap(m * 8)
430 let skl: *i64 = skl_raw as *i64
431 var ci: i64 = 0
432 var c_iter: i64 = 0
433 var c_verdict: i64 = 0
434 while c_verdict == 0 && c_iter < m {
435 if ci >= m { c_verdict = 1 }
436 if c_verdict == 0 {
437 let buf: *u8 = sys_mmap(16)
438 // "col_" + decimal(ci)
439 buf[0] = 0x63; buf[1] = 0x6F; buf[2] = 0x6C; buf[3] = 0x5F
440 var x: i64 = ci
441 var dn: i64 = 0
442 var d_iter: i64 = 0
443 var d_verdict: i64 = 0
444 while d_verdict == 0 && d_iter < 10 {
445 let dgt: i64 = x % 10
446 buf[4 + dn] = (0x30 + dgt) as u8
447 dn = dn + 1
448 x = x / 10
449 if x == 0 { d_verdict = 1 }
450 d_iter = d_iter + 1
451 }
452 // Reverse digits in place
453 var lo: i64 = 4
454 var hi: i64 = 4 + dn - 1
455 var v_iter: i64 = 0
456 var v_verdict: i64 = 0
457 while v_verdict == 0 && v_iter < 16 {
458 if lo >= hi { v_verdict = 1 }
459 if v_verdict == 0 {
460 let t: i64 = buf[lo] as i64
461 buf[lo] = buf[hi]
462 buf[hi] = t as u8
463 lo = lo + 1; hi = hi - 1
464 v_iter = v_iter + 1
465 }
466 }
467 sk[ci] = buf
468 skl[ci] = 4 + dn
469 ci = ci + 1
470 c_iter = c_iter + 1
471 }
472 }
473 keys_use = sk
474 klens_use = skl
475 }
476 let vals_raw: *u8 = sys_mmap(nk * 8)
477 let vals: **NxValue = vals_raw as **NxValue
478 var fi: i64 = 0
479 var f_iter: i64 = 0
480 var f_verdict: i64 = 0
481 while f_verdict == 0 && f_iter < nk {
482 if fi >= nk { f_verdict = 1 }
483 if f_verdict == 0 {
484 var sp: *u8 = 0 as *u8
485 var sl: i64 = 0
486 if fi < m {
487 sp = row_ptr[fi]
488 sl = row_len[fi]
489 }
490 if fi >= m {
491 sp = (p.src as i64) as *u8
492 sl = 0
493 }
494 vals[fi] = nx_value_new_string(sp, sl)
495 fi = fi + 1
496 f_iter = f_iter + 1
497 }
498 }
499 rows[n_rows] = nx_value_new_object(keys_use, klens_use, vals, nk)
500 n_rows = n_rows + 1
501 r_iter = r_iter + 1
502 if rek == NX_CSV_END_INPUT { r_verdict = 1 }
503 }
504 }
505 }
506 }
507
508 *out_verdict = NX_CSV_OK
509 if r_iter >= max_r {
510 if p.pos < p.len { *out_verdict = NX_CSV_MAX_ROWS_HIT }
511 }
512 return nx_value_new_array(rows, n_rows)
513}