Skip to main content

rivet/format/
csv.rs

1use std::io::Write;
2
3use arrow::array::Time64MicrosecondArray;
4use arrow::array::types::Decimal128Type;
5use arrow::array::*;
6use arrow::datatypes::{DataType, SchemaRef, TimeUnit};
7use arrow::record_batch::RecordBatch;
8
9use crate::error::Result;
10use crate::types::decimal::scaled_i128_to_decimal_str;
11
12pub struct CsvFormat;
13
14pub struct CsvFormatWriter {
15    writer: Box<dyn Write + Send>,
16    bytes_written: u64,
17}
18
19impl super::Format for CsvFormat {
20    fn create_writer(
21        &self,
22        schema: &SchemaRef,
23        mut writer: Box<dyn Write + Send>,
24    ) -> Result<Box<dyn super::FormatWriter + Send>> {
25        // Fail loud: arrays and other nested/wide Arrow types have no CSV cell
26        // representation. Reject them up front, naming the column, instead of
27        // silently writing empty values for every row — `format: parquet` or
28        // excluding the column from the query is the fix.
29        if let Some(field) = schema
30            .fields()
31            .iter()
32            .find(|f| !csv_serializable(f.data_type()))
33        {
34            anyhow::bail!(
35                "CSV cannot serialize column '{}' (Arrow type {:?}); use `format: parquet` \
36                 or drop the column from the query",
37                field.name(),
38                field.data_type()
39            );
40        }
41        // RFC-4180 quote each column NAME with the SAME rule the Utf8 data arm
42        // applies to cells (quote on , " CR LF, double embedded quotes). Data
43        // cells were quoted but the header was a raw `join(",")`, so a curated-
44        // query alias with a comma (`... AS "Amount, USD"`) or quote split the
45        // header across columns and silently mis-aligned EVERY downstream reader
46        // from the data. Built once (not the per-cell hot path), so plain String.
47        let header = schema
48            .fields()
49            .iter()
50            .map(|f| {
51                let n = f.name();
52                if n.bytes().any(|b| matches!(b, b',' | b'"' | b'\n' | b'\r')) {
53                    format!("\"{}\"", n.replace('"', "\"\""))
54                } else {
55                    n.clone()
56                }
57            })
58            .collect::<Vec<String>>()
59            .join(",");
60        let header_bytes = header.len() as u64 + 1; // +1 for newline
61        writeln!(writer, "{}", header)?;
62        Ok(Box::new(CsvFormatWriter {
63            writer,
64            bytes_written: header_bytes,
65        }))
66    }
67
68    fn file_extension(&self) -> &str {
69        "csv"
70    }
71}
72
73impl super::FormatWriter for CsvFormatWriter {
74    fn write_batch(&mut self, batch: &RecordBatch) -> Result<()> {
75        let mut buf = Vec::with_capacity(batch.num_rows() * batch.num_columns() * 8);
76        for row_idx in 0..batch.num_rows() {
77            for col_idx in 0..batch.num_columns() {
78                if col_idx > 0 {
79                    buf.push(b',');
80                }
81                write_csv_value(&mut buf, batch.column(col_idx), row_idx)?;
82            }
83            buf.push(b'\n');
84        }
85        self.bytes_written += buf.len() as u64;
86        self.writer.write_all(&buf)?;
87        Ok(())
88    }
89
90    fn finish(self: Box<Self>) -> Result<()> {
91        Ok(())
92    }
93
94    fn bytes_written(&self) -> u64 {
95        self.bytes_written
96    }
97}
98
99/// Arrow types `write_csv_value` can serialize. Everything else — lists,
100/// structs, maps, `Decimal256`, non-UUID fixed binary, … — has no CSV cell
101/// representation and is rejected at writer creation rather than silently
102/// emitted as an empty value.
103pub(crate) fn csv_serializable(dt: &DataType) -> bool {
104    matches!(
105        dt,
106        DataType::Boolean
107            | DataType::Int16
108            | DataType::Int32
109            | DataType::Int64
110            | DataType::UInt64
111            | DataType::Decimal128(_, _)
112            | DataType::Float32
113            | DataType::Float64
114            | DataType::Utf8
115            | DataType::Binary
116            | DataType::FixedSizeBinary(16)
117            | DataType::Date32
118            | DataType::Time64(TimeUnit::Microsecond)
119            | DataType::Timestamp(TimeUnit::Microsecond, _)
120    )
121}
122
123/// Lowercase hex-encode `bytes` into `writer`. Replaces a per-byte
124/// `write!("{:02x}")` (which drove `core::fmt` through the `dyn Write` vtable
125/// once per byte) with a table lookup batched through a stack buffer — zero
126/// allocation, byte-identical output. On a binary/blob column this is the
127/// difference between N format-machinery calls and a handful of `write_all`s.
128fn write_lower_hex(writer: &mut dyn Write, bytes: &[u8]) -> Result<()> {
129    const HEX: &[u8; 16] = b"0123456789abcdef";
130    let mut chunk = [0u8; 1024];
131    for slab in bytes.chunks(chunk.len() / 2) {
132        let mut n = 0;
133        for &b in slab {
134            chunk[n] = HEX[(b >> 4) as usize];
135            chunk[n + 1] = HEX[(b & 0x0f) as usize];
136            n += 2;
137        }
138        writer.write_all(&chunk[..n])?;
139    }
140    Ok(())
141}
142
143fn write_csv_value(writer: &mut dyn Write, array: &dyn Array, idx: usize) -> Result<()> {
144    if array.is_null(idx) {
145        return Ok(());
146    }
147
148    match array.data_type() {
149        DataType::Boolean => {
150            let arr = array
151                .as_any()
152                .downcast_ref::<BooleanArray>()
153                .expect("DataType/Array mismatch");
154            write!(writer, "{}", arr.value(idx))?;
155        }
156        DataType::Int16 => {
157            let arr = array
158                .as_any()
159                .downcast_ref::<Int16Array>()
160                .expect("DataType/Array mismatch");
161            write!(writer, "{}", arr.value(idx))?;
162        }
163        DataType::Int32 => {
164            let arr = array
165                .as_any()
166                .downcast_ref::<Int32Array>()
167                .expect("DataType/Array mismatch");
168            write!(writer, "{}", arr.value(idx))?;
169        }
170        DataType::Int64 => {
171            let arr = array
172                .as_any()
173                .downcast_ref::<Int64Array>()
174                .expect("DataType/Array mismatch");
175            write!(writer, "{}", arr.value(idx))?;
176        }
177        DataType::UInt64 => {
178            let arr = array
179                .as_any()
180                .downcast_ref::<UInt64Array>()
181                .expect("DataType/Array mismatch");
182            write!(writer, "{}", arr.value(idx))?;
183        }
184        DataType::Decimal128(_, scale) => {
185            let arr = array.as_primitive::<Decimal128Type>();
186            let text = scaled_i128_to_decimal_str(arr.value(idx), *scale);
187            writer.write_all(text.as_bytes())?;
188        }
189        DataType::Float32 => {
190            let arr = array
191                .as_any()
192                .downcast_ref::<Float32Array>()
193                .expect("DataType/Array mismatch");
194            write!(writer, "{}", arr.value(idx))?;
195        }
196        DataType::Float64 => {
197            let arr = array
198                .as_any()
199                .downcast_ref::<Float64Array>()
200                .expect("DataType/Array mismatch");
201            write!(writer, "{}", arr.value(idx))?;
202        }
203        DataType::Utf8 => {
204            let arr = array
205                .as_any()
206                .downcast_ref::<StringArray>()
207                .expect("DataType/Array mismatch");
208            let val = arr.value(idx);
209            // RFC 4180: quote on delimiter, quote, LF — and CR, which readers
210            // in universal-newline mode treat as a record terminator. One pass
211            // instead of four `contains` scans per cell.
212            if val
213                .bytes()
214                .any(|b| matches!(b, b',' | b'"' | b'\n' | b'\r'))
215            {
216                writer.write_all(b"\"")?;
217                let mut rest = val;
218                while let Some(pos) = rest.find('"') {
219                    writer.write_all(&rest.as_bytes()[..pos])?;
220                    writer.write_all(b"\"\"")?;
221                    rest = &rest[pos + 1..];
222                }
223                writer.write_all(rest.as_bytes())?;
224                writer.write_all(b"\"")?;
225            } else {
226                writer.write_all(val.as_bytes())?;
227            }
228        }
229        DataType::Binary => {
230            let arr = array
231                .as_any()
232                .downcast_ref::<BinaryArray>()
233                .expect("DataType/Array mismatch");
234            write_lower_hex(writer, arr.value(idx))?;
235        }
236        // FixedSizeBinary today only carries 16-byte UUIDs (see
237        // `RivetType::Uuid` → `DataType::FixedSizeBinary(16)` in
238        // `src/types/mapping.rs`). CSV has no native binary cell; emit the
239        // canonical hyphenated lowercase form so downstream readers can
240        // recognise it as a UUID rather than 16 bytes of mojibake. Any
241        // future FixedSizeBinary use that is not a UUID should branch on
242        // the size argument before reaching this arm.
243        DataType::FixedSizeBinary(16) => {
244            let arr = array
245                .as_any()
246                .downcast_ref::<FixedSizeBinaryArray>()
247                .expect("DataType/Array mismatch");
248            let val = arr.value(idx);
249            let mut bytes = [0u8; 16];
250            bytes.copy_from_slice(val);
251            write!(writer, "{}", uuid::Uuid::from_bytes(bytes).to_hyphenated())?;
252        }
253        DataType::Date32 => {
254            let arr = array
255                .as_any()
256                .downcast_ref::<Date32Array>()
257                .expect("DataType/Array mismatch");
258            let days = arr.value(idx);
259            // `Date32` is "days since 1970-01-01"; a pathological value near
260            // i32::MAX overflows `NaiveDate + Duration` and panics in chrono.
261            // Fall back to checked arithmetic and emit an empty cell on
262            // overflow — matches the null-cell convention for unserialisable
263            // values elsewhere in this writer.
264            let epoch = chrono::NaiveDate::from_ymd_opt(1970, 1, 1).expect("epoch is valid");
265            let date =
266                chrono::Duration::try_days(days as i64).and_then(|d| epoch.checked_add_signed(d));
267            if let Some(date) = date {
268                write!(writer, "{}", date)?;
269            }
270        }
271        DataType::Time64(TimeUnit::Microsecond) => {
272            let arr = array
273                .as_any()
274                .downcast_ref::<Time64MicrosecondArray>()
275                .expect("DataType/Array mismatch");
276            let micros = arr.value(idx);
277            // MySQL TIME is a SIGNED duration (-838:59:59 .. 838:59:59), so micros
278            // can be negative. Extract the sign ONCE and format the magnitude: the
279            // old truncating `/`+`%` on a negative value emitted a minus PER FIELD
280            // ("-1:-30:00", "00:00:00.-00001", "00:00:-1.-500000"), corrupt in every
281            // CSV/time reader. Hours are `{:02}` MINIMUM-width, so a >24h duration
282            // (up to 838h) still prints in full.
283            let neg = micros < 0;
284            let abs = micros.unsigned_abs();
285            let secs = abs / 1_000_000;
286            let frac_us = abs % 1_000_000;
287            write!(
288                writer,
289                "{}{:02}:{:02}:{:02}.{:06}",
290                if neg { "-" } else { "" },
291                secs / 3600,
292                (secs % 3600) / 60,
293                secs % 60,
294                frac_us
295            )?;
296        }
297        DataType::Timestamp(TimeUnit::Microsecond, tz) => {
298            let arr = array
299                .as_any()
300                .downcast_ref::<TimestampMicrosecondArray>()
301                .expect("DataType/Array mismatch");
302            // rivet's type model keeps the naive-vs-instant distinction that
303            // every other output (Parquet `isAdjustedToUTC`, the warehouse
304            // target types) preserves: a source TIMESTAMPTZ maps to
305            // `Timestamp(_, Some("UTC"))`, a naive TIMESTAMP to `None`. Without
306            // a marker on the instant, CSV renders BOTH identically — a consumer
307            // loading the instant into a tz-aware column on a NON-UTC session
308            // then reads the UTC wall-clock as local and shifts every value (the
309            // session-state class, on the CSV encode leg). Emit a trailing `Z`
310            // for the instant (rivet normalises to UTC) so it is unambiguous;
311            // the naive value stays bare (its wall-clock IS the value).
312            let is_instant = tz.is_some();
313            let micros = arr.value(idx);
314            // FLOOR division (div_euclid/rem_euclid), NOT truncating `/`+`%`: for
315            // a pre-1970 (negative) micros with a sub-second fraction, `micros %
316            // 1_000_000` is NEGATIVE, so `* 1_000 as u32` wraps to a huge value,
317            // `from_timestamp` returns None, and the cell was emitted EMPTY —
318            // silent data loss for every sub-second timestamp before the epoch
319            // (Parquet keeps them). Euclidean rem is always in [0, 1_000_000), so
320            // nsecs stays in range: micros=-1 → secs=-1, nsecs=999_999_000 →
321            // 1969-12-31T23:59:59.999999.
322            let secs = micros.div_euclid(1_000_000);
323            let nsecs = (micros.rem_euclid(1_000_000) * 1_000) as u32;
324            if let Some(dt) = chrono::DateTime::from_timestamp(secs, nsecs) {
325                use chrono::{Datelike as _, Timelike as _};
326                let y = dt.year();
327                // Fast path: emit the `%Y-%m-%dT%H:%M:%S%.6f` form by hand to
328                // skip chrono's per-value strftime-string re-parse. `{:04}`
329                // matches `%Y`'s zero-pad-to-4 exactly for years 0..=9999;
330                // outside that range (pre-CE / 5-digit years from pathological
331                // micros) fall back to chrono so the output stays identical.
332                if (0..=9999).contains(&y) {
333                    write!(
334                        writer,
335                        "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}.{:06}",
336                        y,
337                        dt.month(),
338                        dt.day(),
339                        dt.hour(),
340                        dt.minute(),
341                        dt.second(),
342                        dt.nanosecond() / 1_000
343                    )?;
344                } else {
345                    write!(writer, "{}", dt.format("%Y-%m-%dT%H:%M:%S%.6f"))?;
346                }
347                if is_instant {
348                    writer.write_all(b"Z")?;
349                }
350            }
351        }
352        other => {
353            // Defensive: `create_writer` rejects unsupported types up front, so
354            // this should be unreachable. Bail rather than silently skip if a
355            // new type slips through.
356            anyhow::bail!(
357                "CSV: no serializer for Arrow type {other:?} (column should have been rejected at writer creation)"
358            );
359        }
360    }
361
362    Ok(())
363}
364
365#[cfg(test)]
366mod tests {
367    use super::*;
368    use arrow::datatypes::{DataType, Field, Schema, TimeUnit};
369    use std::sync::Arc;
370
371    // Helper: render one cell to a String using write_csv_value.
372    fn cell<A: Array + 'static>(array: A, idx: usize) -> String {
373        let mut buf = Vec::new();
374        write_csv_value(&mut buf, &array, idx).unwrap();
375        String::from_utf8(buf).unwrap()
376    }
377
378    // Helper: render a null cell from any typed array.
379    fn null_cell(dt: DataType) -> String {
380        use arrow::array::new_null_array;
381        let arr = new_null_array(&dt, 1);
382        let mut buf = Vec::new();
383        write_csv_value(&mut buf, arr.as_ref(), 0).unwrap();
384        String::from_utf8(buf).unwrap()
385    }
386
387    // ── mutation-W5 gap closures ─────────────────────────────────────────────
388
389    #[test]
390    fn time64_renders_exact_wall_clock() {
391        // W5: the µs → hh:mm:ss.ffffff arithmetic (/, %) had no golden — a
392        // mutated divisor silently corrupts every TIME cell in CSV output.
393        use arrow::array::Time64MicrosecondArray;
394        assert_eq!(
395            cell(Time64MicrosecondArray::from(vec![86_399_999_999_i64]), 0),
396            "23:59:59.999999"
397        );
398        assert_eq!(
399            cell(Time64MicrosecondArray::from(vec![3_661_000_001_i64]), 0),
400            "01:01:01.000001"
401        );
402        assert_eq!(
403            cell(Time64MicrosecondArray::from(vec![0_i64]), 0),
404            "00:00:00.000000"
405        );
406    }
407
408    // MySQL TIME is a SIGNED duration; the batch decoder emits negative micros.
409    // The old truncating `/`+`%` put a minus on EVERY field ("-1:-30:00",
410    // "00:00:00.-00001") — corrupt in any reader. One leading sign, magnitude
411    // formatted; a >24h duration still prints in full.
412    #[test]
413    fn negative_time_renders_one_leading_sign_not_a_minus_per_field() {
414        use arrow::array::Time64MicrosecondArray;
415        for (micros, want) in [
416            (-5_400_000_000_i64, "-01:30:00.000000"),      // -1h30m
417            (-1_i64, "-00:00:00.000001"),                  // -1 microsecond
418            (-1_500_000_i64, "-00:00:01.500000"),          // -1.5s
419            (-3_020_399_000_000_i64, "-838:59:59.000000"), // MySQL TIME min (>24h)
420            (5_400_000_000_i64, "01:30:00.000000"),        // positive unaffected
421        ] {
422            assert_eq!(
423                cell(Time64MicrosecondArray::from(vec![micros]), 0),
424                want,
425                "micros={micros}"
426            );
427        }
428    }
429
430    #[test]
431    fn write_batch_layout_and_byte_accounting_are_exact() {
432        // W5: `col_idx > 0` mutating to `>=` puts a LEADING comma on every row
433        // (corrupt CSV); the header `len + 1` and `bytes_written +=` mutants
434        // desync the rotation accounting. Pin the exact output AND the count.
435        use arrow::array::Int64Array;
436        use std::sync::{Arc as SArc, Mutex};
437
438        #[derive(Clone)]
439        struct SharedBuf(SArc<Mutex<Vec<u8>>>);
440        impl std::io::Write for SharedBuf {
441            fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
442                self.0.lock().unwrap().extend_from_slice(b);
443                Ok(b.len())
444            }
445            fn flush(&mut self) -> std::io::Result<()> {
446                Ok(())
447            }
448        }
449
450        let schema: SchemaRef = Arc::new(Schema::new(vec![
451            Field::new("a", DataType::Int64, false),
452            Field::new("b", DataType::Int64, false),
453        ]));
454        let sink = SharedBuf(SArc::new(Mutex::new(Vec::new())));
455        use crate::format::Format as _;
456        let mut w = CsvFormat
457            .create_writer(&schema, Box::new(sink.clone()))
458            .unwrap();
459        let batch = RecordBatch::try_new(
460            schema,
461            vec![
462                Arc::new(Int64Array::from(vec![1, 2])),
463                Arc::new(Int64Array::from(vec![10, 20])),
464            ],
465        )
466        .unwrap();
467        w.write_batch(&batch).unwrap();
468
469        let text = String::from_utf8(sink.0.lock().unwrap().clone()).unwrap();
470        assert_eq!(
471            text, "a,b\n1,10\n2,20\n",
472            "exact CSV layout (no leading commas)"
473        );
474        assert_eq!(
475            w.bytes_written(),
476            text.len() as u64,
477            "byte accounting must equal the physical output"
478        );
479    }
480
481    // ── null handling ────────────────────────────────────────────────────────
482
483    #[test]
484    fn null_value_writes_empty_string() {
485        assert_eq!(null_cell(DataType::Int64), "");
486        assert_eq!(null_cell(DataType::Utf8), "");
487        assert_eq!(null_cell(DataType::Boolean), "");
488    }
489
490    // ── scalars ─────────────────────────────────────────────────────────────
491
492    #[test]
493    fn bool_true_writes_true() {
494        assert_eq!(cell(BooleanArray::from(vec![true]), 0), "true");
495    }
496
497    #[test]
498    fn bool_false_writes_false() {
499        assert_eq!(cell(BooleanArray::from(vec![false]), 0), "false");
500    }
501
502    #[test]
503    fn int16_value() {
504        assert_eq!(cell(Int16Array::from(vec![42i16]), 0), "42");
505    }
506
507    #[test]
508    fn int32_negative() {
509        assert_eq!(cell(Int32Array::from(vec![-7i32]), 0), "-7");
510    }
511
512    #[test]
513    fn decimal128_writes_exact_text() {
514        let arr = Decimal128Array::from(vec![10i128])
515            .with_precision_and_scale(18, 2)
516            .unwrap();
517        assert_eq!(cell(arr, 0), "0.10");
518        let scaled =
519            crate::types::decimal::decimal_str_to_scaled_i128("999999999999.99", 2).unwrap();
520        let arr = Decimal128Array::from(vec![scaled])
521            .with_precision_and_scale(18, 2)
522            .unwrap();
523        assert_eq!(cell(arr, 0), "999999999999.99");
524    }
525
526    #[test]
527    fn int64_large() {
528        assert_eq!(
529            cell(Int64Array::from(vec![9_999_999_999i64]), 0),
530            "9999999999"
531        );
532    }
533
534    #[test]
535    fn float32_value() {
536        let result = cell(Float32Array::from(vec![1.5f32]), 0);
537        assert!(result.starts_with("1.5"), "got: {result}");
538    }
539
540    #[test]
541    fn float64_value() {
542        let result = cell(Float64Array::from(vec![std::f64::consts::PI]), 0);
543        assert!(result.starts_with("3.14"), "got: {result}");
544    }
545
546    // Characterization: float NaN/±Infinity are IEEE-754 values a float column
547    // can legitimately hold (unlike `decimal`, whose Arrow `Decimal128` has no
548    // NaN bit pattern — see the NUMERIC NaN/infinity reject in
549    // `postgres::arrow_convert::build_array`). They are preserved natively in
550    // Parquet; in CSV we emit the Rust float literal (`NaN` / `inf` / `-inf`)
551    // rather than an empty cell, because writing empty would silently conflate
552    // a real NaN/Inf with NULL — corruption — whereas a recognizable literal
553    // round-trips into every major loader's float parser. This pins that
554    // contract so a future arrow/std change can't silently alter it. The CSV
555    // literal is documented in docs/type-mapping.md.
556    #[test]
557    fn float_special_values_emit_literals_not_empty() {
558        assert_eq!(cell(Float64Array::from(vec![f64::NAN]), 0), "NaN");
559        assert_eq!(cell(Float64Array::from(vec![f64::INFINITY]), 0), "inf");
560        assert_eq!(cell(Float64Array::from(vec![f64::NEG_INFINITY]), 0), "-inf");
561        assert_eq!(cell(Float32Array::from(vec![f32::NAN]), 0), "NaN");
562        assert_eq!(cell(Float32Array::from(vec![f32::INFINITY]), 0), "inf");
563        // -0.0 keeps its sign (a real IEEE-754 distinction), never becomes "0".
564        assert_eq!(cell(Float64Array::from(vec![-0.0f64]), 0), "-0");
565    }
566
567    // ── string escaping ──────────────────────────────────────────────────────
568
569    #[test]
570    fn plain_string_no_quoting() {
571        assert_eq!(cell(StringArray::from(vec!["hello"]), 0), "hello");
572    }
573
574    #[test]
575    fn string_with_comma_is_quoted() {
576        assert_eq!(cell(StringArray::from(vec!["a,b"]), 0), "\"a,b\"");
577    }
578
579    #[test]
580    fn string_with_double_quote_is_escaped() {
581        // say "hi" → opening " + say  + "" + hi + "" + closing " = "say ""hi"""
582        let result = cell(StringArray::from(vec![r#"say "hi""#]), 0);
583        assert_eq!(result, r#""say ""hi""""#);
584    }
585
586    #[test]
587    fn string_with_newline_is_quoted() {
588        let result = cell(StringArray::from(vec!["line1\nline2"]), 0);
589        assert!(
590            result.starts_with('"') && result.ends_with('"'),
591            "got: {result}"
592        );
593        assert!(result.contains("line1\nline2"), "got: {result}");
594    }
595
596    // ROAST-RED csv-cr-quote: the quote predicate checks ',', '"' and '\n' but
597    // not '\r', so a value containing a lone CR is emitted unquoted — RFC 4180
598    // requires quoting CR, and Excel/Python universal-newline readers split the
599    // row on the bare CR.
600    // Asserts CORRECT behavior; expected to FAIL until the fix lands.
601    #[test]
602    fn roast_string_with_carriage_return_is_quoted() {
603        let result = cell(StringArray::from(vec!["a\rb"]), 0);
604        assert_eq!(
605            result, "\"a\rb\"",
606            "lone CR must force quoting per RFC 4180, but got unquoted cell {result:?}"
607        );
608    }
609
610    // ── binary ───────────────────────────────────────────────────────────────
611
612    #[test]
613    fn binary_is_written_as_hex() {
614        let arr = BinaryArray::from_vec(vec![&[0xDE, 0xAD, 0xBE, 0xEF][..]]);
615        assert_eq!(cell(arr, 0), "deadbeef");
616    }
617
618    #[test]
619    fn binary_empty_writes_empty() {
620        let arr = BinaryArray::from_vec(vec![&[][..]]);
621        assert_eq!(cell(arr, 0), "");
622    }
623
624    // ── Date32 ───────────────────────────────────────────────────────────────
625
626    #[test]
627    fn date32_epoch_is_1970_01_01() {
628        assert_eq!(cell(Date32Array::from(vec![0i32]), 0), "1970-01-01");
629    }
630
631    #[test]
632    fn date32_positive_offset() {
633        // 365 days after epoch = 1971-01-01
634        assert_eq!(cell(Date32Array::from(vec![365i32]), 0), "1971-01-01");
635    }
636
637    // ── Timestamp(Microsecond) ───────────────────────────────────────────────
638
639    #[test]
640    fn timestamp_micros_formats_as_iso() {
641        // 2023-01-01T00:00:00.000000 = 1672531200_000000 micros since epoch
642        let micros: i64 = 1_672_531_200 * 1_000_000;
643        let _schema = Arc::new(Schema::new(vec![Field::new(
644            "ts",
645            DataType::Timestamp(TimeUnit::Microsecond, None),
646            true,
647        )]));
648        let arr = TimestampMicrosecondArray::from(vec![micros]);
649        let result = cell(arr, 0);
650        assert!(result.starts_with("2023-01-01T"), "got: {result}");
651        assert!(result.contains("00:00:00"), "got: {result}");
652    }
653
654    // rivet keeps the naive-vs-instant distinction in every other output
655    // (Parquet isAdjustedToUTC, the warehouse target types). CSV must too: a
656    // naive TIMESTAMP renders bare (its wall-clock IS the value), an instant
657    // TIMESTAMPTZ (Timestamp(_, Some("UTC"))) gets a trailing `Z` — else a
658    // consumer loading the instant on a NON-UTC session reads the UTC wall-clock
659    // as local and shifts every value.
660    #[test]
661    fn timestamp_marks_instant_utc_but_leaves_naive_bare() {
662        // 2024-06-15T03:00:00.000000 UTC.
663        let micros: i64 = 1_718_420_400 * 1_000_000;
664        let naive = TimestampMicrosecondArray::from(vec![micros]);
665        assert_eq!(
666            cell(naive, 0),
667            "2024-06-15T03:00:00.000000",
668            "a naive TIMESTAMP renders bare — no tz marker"
669        );
670        let instant = TimestampMicrosecondArray::from(vec![micros]).with_timezone("UTC");
671        assert_eq!(
672            cell(instant, 0),
673            "2024-06-15T03:00:00.000000Z",
674            "an instant TIMESTAMPTZ renders an explicit UTC `Z`"
675        );
676    }
677
678    // ── write_batch via CsvFormat ────────────────────────────────────────────
679
680    #[test]
681    fn csv_format_write_batch_tracks_bytes_and_succeeds() {
682        use crate::format::Format;
683
684        let schema = Arc::new(Schema::new(vec![
685            Field::new("id", DataType::Int64, false),
686            Field::new("name", DataType::Utf8, true),
687        ]));
688        let batch = arrow::record_batch::RecordBatch::try_new(
689            schema.clone(),
690            vec![
691                Arc::new(Int64Array::from(vec![1i64, 2])),
692                Arc::new(StringArray::from(vec![Some("alice"), None])),
693            ],
694        )
695        .unwrap();
696
697        // Pass Vec by value — avoids the &mut T 'static lifetime requirement.
698        let fmt = CsvFormat;
699        let mut writer = fmt
700            .create_writer(&schema, Box::new(Vec::<u8>::new()))
701            .unwrap();
702        writer.write_batch(&batch).unwrap();
703        // Header "id,name\n" + rows "1,alice\n" + "2,\n" = at least 18 bytes
704        assert!(
705            writer.bytes_written() > 10,
706            "expected >10 bytes, got {}",
707            writer.bytes_written()
708        );
709        writer.finish().unwrap();
710    }
711
712    // Regression: a column NAME containing a comma/quote must be RFC-4180 quoted
713    // in the header, exactly as data cells are — else a curated-query alias like
714    // `... AS "Amount, USD"` split the header and silently mis-aligned every
715    // downstream reader from the data columns. RED before the header quoter.
716    #[test]
717    fn csv_header_quotes_a_column_name_with_a_comma_or_quote() {
718        use crate::format::Format;
719        let schema = Arc::new(Schema::new(vec![
720            Field::new("Amount, USD", DataType::Int64, false), // comma in the name
721            Field::new("a\"b", DataType::Utf8, true),          // quote in the name
722            Field::new("plain", DataType::Utf8, true),         // untouched
723        ]));
724        let dir = tempfile::tempdir().unwrap();
725        let path = dir.path().join("out.csv");
726        let w = CsvFormat
727            .create_writer(&schema, Box::new(std::fs::File::create(&path).unwrap()))
728            .unwrap();
729        w.finish().unwrap();
730        let out = std::fs::read_to_string(&path).unwrap();
731        let header = out.lines().next().unwrap();
732        // comma-name quoted; quote-name quoted with the `"` doubled; plain bare —
733        // and crucially still exactly THREE header columns, the comma not a split.
734        assert_eq!(header, r#""Amount, USD","a""b",plain"#);
735    }
736
737    // ── fail loud on types CSV can't represent ───────────────────────────────
738
739    #[test]
740    fn csv_rejects_array_columns_loudly() {
741        use crate::format::Format;
742        let schema = Arc::new(Schema::new(vec![
743            Field::new("id", DataType::Int64, false),
744            Field::new(
745                "tags",
746                DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))),
747                true,
748            ),
749        ]));
750        let Err(err) = CsvFormat.create_writer(&schema, Box::new(Vec::<u8>::new())) else {
751            panic!("CSV must reject array columns, not silently drop them");
752        };
753        let msg = format!("{err:#}");
754        assert!(msg.contains("tags"), "error must name the column: {msg}");
755        assert!(msg.to_lowercase().contains("csv"), "{msg}");
756    }
757
758    /// Consistency guard: every type `csv_serializable` admits must actually be
759    /// handled by `write_csv_value` (not hit its `other => bail` fallthrough).
760    /// Keeps the whitelist and the writer in lock-step so one can't drift.
761    #[test]
762    fn every_serializable_type_is_actually_written() {
763        use crate::format::Format;
764        let cols: Vec<(&str, ArrayRef)> = vec![
765            ("b", Arc::new(BooleanArray::from(vec![true]))),
766            ("i16", Arc::new(Int16Array::from(vec![1i16]))),
767            ("i32", Arc::new(Int32Array::from(vec![1i32]))),
768            ("i64", Arc::new(Int64Array::from(vec![1i64]))),
769            ("u64", Arc::new(UInt64Array::from(vec![1u64]))),
770            (
771                "dec",
772                Arc::new(
773                    Decimal128Array::from(vec![100i128])
774                        .with_precision_and_scale(18, 2)
775                        .unwrap(),
776                ),
777            ),
778            ("f32", Arc::new(Float32Array::from(vec![1.0f32]))),
779            ("f64", Arc::new(Float64Array::from(vec![1.0f64]))),
780            ("s", Arc::new(StringArray::from(vec!["x"]))),
781            ("bin", Arc::new(BinaryArray::from_vec(vec![&[1u8][..]]))),
782            (
783                "uuid",
784                Arc::new(
785                    FixedSizeBinaryArray::try_from_iter(std::iter::once(vec![0u8; 16])).unwrap(),
786                ),
787            ),
788            ("d", Arc::new(Date32Array::from(vec![0i32]))),
789            ("t", Arc::new(Time64MicrosecondArray::from(vec![0i64]))),
790            ("ts", Arc::new(TimestampMicrosecondArray::from(vec![0i64]))),
791        ];
792        let fields: Vec<Field> = cols
793            .iter()
794            .map(|(n, a)| Field::new(*n, a.data_type().clone(), true))
795            .collect();
796        // Sanity: each column's type is on the whitelist.
797        for f in &fields {
798            assert!(
799                csv_serializable(f.data_type()),
800                "test type {:?} not in csv_serializable",
801                f.data_type()
802            );
803        }
804        let schema = Arc::new(Schema::new(fields));
805        let arrays: Vec<ArrayRef> = cols.into_iter().map(|(_, a)| a).collect();
806        let batch = RecordBatch::try_new(schema.clone(), arrays).unwrap();
807        let mut w = CsvFormat
808            .create_writer(&schema, Box::new(Vec::<u8>::new()))
809            .unwrap();
810        w.write_batch(&batch)
811            .expect("every serializable type must write without hitting the fallthrough");
812    }
813
814    // ── perf-wave byte-identity guards ───────────────────────────────────────
815    // The hex + timestamp fast paths must stay byte-for-byte identical to the
816    // `{:02x}`-per-byte / `dt.format(...)` forms they replaced.
817
818    #[test]
819    fn binary_hex_matches_per_byte_format_for_all_byte_values() {
820        // Every byte 0..=255 plus the empty slice — the table+chunk encoder
821        // must equal the old per-byte `{:02x}`.
822        let all: Vec<u8> = (0..=255u8).collect();
823        for case in [&all[..], &[][..], &[0x00, 0xff, 0x10, 0x0a]] {
824            let expected: String = case.iter().map(|b| format!("{b:02x}")).collect();
825            let got = cell(BinaryArray::from_vec(vec![case]), 0);
826            assert_eq!(got, expected, "hex mismatch for {case:?}");
827        }
828    }
829
830    #[test]
831    fn binary_hex_spans_chunk_boundary() {
832        // Larger than the 512-byte slab so the chunked write_all path is hit.
833        let big: Vec<u8> = (0..2000u32).map(|i| (i % 256) as u8).collect();
834        let expected: String = big.iter().map(|b| format!("{b:02x}")).collect();
835        let got = cell(BinaryArray::from_vec(vec![&big[..]]), 0);
836        assert_eq!(got, expected);
837    }
838
839    #[test]
840    fn timestamp_fast_path_matches_chrono_format() {
841        // Representative micros: epoch, sub-second, end-of-day, a far-future
842        // in-range year, and a value whose year leaves the 0..=9999 fast path
843        // (must hit the chrono fallback and still match). The recompute uses the
844        // SAME euclidean split the product now uses, so it is a self-consistency
845        // check of the fast-path-vs-chrono rendering — the pre-1970 correctness
846        // is pinned separately below with a HARD-CODED oracle.
847        let cases: [i64; 6] = [
848            0,
849            1_700_000_000_123_456,        // 2023-… with micros
850            1_000_000 * 86_399 + 999_999, // 1970-01-01T23:59:59.999999
851            -62_135_596_800_000_000,      // 0001-01-01T00:00:00 (negative, fast path)
852            253_402_300_799_000_000,      // 9999-12-31T23:59:59 (still fast path)
853            300_000_000_000_000_000,      // year > 9999 → chrono fallback
854        ];
855        for micros in cases {
856            let got = cell(TimestampMicrosecondArray::from(vec![micros]), 0);
857            let secs = micros.div_euclid(1_000_000);
858            let nsecs = (micros.rem_euclid(1_000_000) * 1_000) as u32;
859            let expected = match chrono::DateTime::from_timestamp(secs, nsecs) {
860                Some(dt) => format!("{}", dt.format("%Y-%m-%dT%H:%M:%S%.6f")),
861                None => String::new(),
862            };
863            assert_eq!(got, expected, "timestamp mismatch for micros={micros}");
864        }
865    }
866
867    // Regression (true oracle, NOT the recomputed self-oracle above): a pre-1970
868    // sub-second timestamp must render its real value, never an empty cell. With
869    // the old truncating `/`+`%` split the negative remainder wrapped `nsecs`,
870    // `from_timestamp` returned None, and the cell was emitted EMPTY — 100%
871    // silent loss for this whole class while Parquet kept the value. Hard-coded
872    // expected strings so the test can never re-derive the bug.
873    #[test]
874    fn pre_1970_subsecond_timestamp_is_not_dropped_to_an_empty_cell() {
875        let expect: [(i64, &str); 4] = [
876            (-1, "1969-12-31T23:59:59.999999"), // one micro before the epoch
877            (-500_000, "1969-12-31T23:59:59.500000"), // half a second before
878            (-86_400_000_000, "1969-12-31T00:00:00.000000"), // whole day before (whole second)
879            (-125_125_000, "1969-12-31T23:57:54.875000"), // multi-second, sub-second frac
880        ];
881        for (micros, want) in expect {
882            let got = cell(TimestampMicrosecondArray::from(vec![micros]), 0);
883            assert_eq!(
884                got, want,
885                "pre-1970 micros={micros} must render {want:?}, not an empty/other cell"
886            );
887        }
888    }
889}