use crate::error::{Error, Result};
use crate::limits::Limits;
pub const MODEL_VERSION: u8 = 1;
pub const MAX_MODEL_RECORDS: u32 = 1 << 24;
pub const MAX_MODEL_FIELDS: u64 = 1 << 24;
pub const DELIM_COMMA: u8 = b',';
pub const DELIM_TAB: u8 = b'\t';
pub const TERM_LF: u8 = 0;
pub const TERM_CRLF: u8 = 1;
pub const TERM_CR: u8 = 2;
pub const QUOTE: u8 = b'"';
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Dialect {
pub delimiter: u8,
pub quote: u8,
pub terminator: u8,
pub bom_len: u8,
pub mixed_terminators: bool,
}
impl Dialect {
pub const fn delimiter_name(&self) -> &'static str {
match self.delimiter {
DELIM_TAB => "tab",
_ => "comma",
}
}
pub const fn terminator_name(&self) -> &'static str {
match self.terminator {
TERM_CRLF => "crlf",
TERM_CR => "cr",
_ => "lf",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CsvField {
pub start: u64,
pub end: u64,
pub quoted: bool,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CsvRecord {
pub start: u64,
pub end: u64,
pub fields: Vec<CsvField>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StreamRecord {
pub start: u64,
pub end: u64,
pub terminator: u8,
pub fields: Vec<CsvField>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CsvModel {
pub doc_len: u64,
pub dialect: Dialect,
pub has_header: bool,
pub records: Vec<CsvRecord>,
}
impl CsvModel {
pub fn node(&self, index: u32) -> Option<&CsvRecord> {
self.records.get(index as usize)
}
pub fn row_count(&self) -> usize {
self.records.len()
}
pub fn header(&self) -> Option<&CsvRecord> {
self.records.first()
}
pub fn encode(&self) -> Vec<u8> {
let mut out = Vec::with_capacity(32 + self.records.len() * 24);
out.extend_from_slice(b"CSVM");
out.push(MODEL_VERSION);
out.push(self.dialect.delimiter);
out.push(self.dialect.quote);
out.push(self.dialect.terminator);
out.push(self.dialect.bom_len);
let mut flags = 0u8;
if self.has_header {
flags |= 1;
}
if self.dialect.mixed_terminators {
flags |= 2;
}
out.push(flags);
out.extend_from_slice(&self.doc_len.to_le_bytes());
out.extend_from_slice(&(self.records.len() as u32).to_le_bytes());
for r in &self.records {
out.extend_from_slice(&r.start.to_le_bytes());
out.extend_from_slice(&r.end.to_le_bytes());
out.extend_from_slice(&(r.fields.len() as u32).to_le_bytes());
for f in &r.fields {
out.extend_from_slice(&f.start.to_le_bytes());
out.extend_from_slice(&f.end.to_le_bytes());
out.push(u8::from(f.quoted));
}
}
out
}
pub fn decode(bytes: &[u8]) -> Result<CsvModel> {
let mut r = BinReader::new(bytes);
if r.bytes(4)? != b"CSVM" {
return Err(corrupt("bad model magic"));
}
if r.u8()? != MODEL_VERSION {
return Err(corrupt("unsupported model version"));
}
let delimiter = r.u8()?;
if delimiter != DELIM_COMMA && delimiter != DELIM_TAB {
return Err(corrupt("unknown delimiter"));
}
let quote = r.u8()?;
if quote != QUOTE {
return Err(corrupt("unknown quote character"));
}
let terminator = r.u8()?;
if terminator > TERM_CR {
return Err(corrupt("unknown terminator"));
}
let bom_len = r.u8()?;
if bom_len > 3 {
return Err(corrupt("implausible BOM length"));
}
let flags = r.u8()?;
if flags & !3 != 0 {
return Err(corrupt("unknown model flags"));
}
let doc_len = r.u64()?;
let count = r.u32()?;
if count > MAX_MODEL_RECORDS {
return Err(corrupt("model record count is implausible"));
}
let mut records = Vec::with_capacity(count as usize);
let mut total_fields = 0u64;
for _ in 0..count {
let start = r.u64()?;
let end = r.u64()?;
if start > end || end > doc_len {
return Err(corrupt("record span is outside the document"));
}
let fn_ = r.u32()?;
total_fields = total_fields.saturating_add(fn_ as u64);
if total_fields > MAX_MODEL_FIELDS {
return Err(corrupt("model field count is implausible"));
}
let mut fields = Vec::with_capacity(fn_ as usize);
for _ in 0..fn_ {
let fs = r.u64()?;
let fe = r.u64()?;
let quoted = r.u8()?;
if quoted > 1 {
return Err(corrupt("invalid quoted flag"));
}
if fs > fe || fs < start || fe > end {
return Err(corrupt("field span is outside its record"));
}
fields.push(CsvField {
start: fs,
end: fe,
quoted: quoted == 1,
});
}
records.push(CsvRecord { start, end, fields });
}
if !r.at_end() {
return Err(corrupt("model has trailing bytes"));
}
if bom_len as u64 > doc_len {
return Err(corrupt("BOM length exceeds the document"));
}
Ok(CsvModel {
doc_len,
dialect: Dialect {
delimiter,
quote,
terminator,
bom_len,
mixed_terminators: flags & 2 != 0,
},
has_header: flags & 1 != 0,
records,
})
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CsvMatch {
pub record: u32,
pub column: u32,
pub quoted: bool,
pub start: u64,
pub end: u64,
pub text: String,
}
pub fn detect(source: &[u8], limits: Limits) -> bool {
if looks_like_other(source, limits) {
return false;
}
sniff_dialect(source, limits).is_ok()
}
fn looks_like_other(source: &[u8], limits: Limits) -> bool {
let _ = limits;
if source.starts_with(b"%PDF-") {
return true;
}
if source.starts_with(b"PK\x03\x04")
|| source.starts_with(b"PK\x05\x06")
|| source.starts_with(b"PK\x07\x08")
{
return true;
}
#[cfg(feature = "json")]
if crate::adapter::json::detect(source, limits) {
return true;
}
#[cfg(feature = "yaml")]
if crate::adapter::yaml::detect(source, limits) {
return true;
}
false
}
pub fn sniff_dialect(source: &[u8], limits: Limits) -> Result<Dialect> {
if source.is_empty() {
return Err(corrupt("empty input is not a CSV table"));
}
if source.len() as u64 > limits.max_csv_document_bytes {
return Err(Error::resource_limit(format!(
"CSV source is {} bytes, above the {}-byte document cap",
source.len(),
limits.max_csv_document_bytes
)));
}
let bom_len = if source.starts_with(&[0xEF, 0xBB, 0xBF]) {
3
} else {
0
};
for delimiter in [DELIM_COMMA, DELIM_TAB] {
if let Some(dialect) = try_dialect(source, delimiter, bom_len, limits) {
return Ok(dialect);
}
}
Err(corrupt(
"source is not a CSV table under a comma or tab delimiter",
))
}
fn try_dialect(source: &[u8], delimiter: u8, bom_len: u8, limits: Limits) -> Option<Dialect> {
let cap = limits.max_csv_sampled_records_for_detection.max(2) as usize;
let base = Dialect {
delimiter,
quote: QUOTE,
terminator: TERM_LF,
bom_len,
mixed_terminators: false,
};
let mut scanner = Scanner::new(source, base, limits);
let mut counts: Vec<u64> = Vec::new();
for _ in 0..cap {
match scanner.next_record(true) {
Ok(Some(rec)) => counts.push(rec.nfields),
Ok(None) => break,
Err(_) => return None,
}
}
if counts.len() < 2 {
return None;
}
let mut best = 0u64;
let mut best_freq = 0usize;
for &c in &counts {
let freq = counts.iter().filter(|&&x| x == c).count();
if freq > best_freq || (freq == best_freq && c > best) {
best = c;
best_freq = freq;
}
}
if best < 2 || best_freq * 2 <= counts.len() || counts[0] != best {
return None;
}
Some(Dialect {
terminator: scanner.terminator,
mixed_terminators: scanner.mixed_terminators,
..base
})
}
pub fn parse(source: &[u8], limits: Limits, build: bool) -> Result<CsvModel> {
let dialect = sniff_dialect(source, limits)?;
let mut scanner = Scanner::new(source, dialect, limits);
let mut records: Vec<CsvRecord> = Vec::new();
let mut total_fields = 0u64;
while let Some(rec) = scanner.next_record(!build)? {
total_fields = total_fields.saturating_add(rec.nfields);
if total_fields > MAX_MODEL_FIELDS {
return Err(Error::resource_limit(format!(
"CSV table exceeds the {MAX_MODEL_FIELDS}-field model cap"
)));
}
if records.len() as u64 >= limits.max_csv_rows as u64 {
return Err(Error::resource_limit(format!(
"CSV table exceeds the {}-record cap",
limits.max_csv_rows
)));
}
if build {
records.push(CsvRecord {
start: rec.start,
end: rec.end,
fields: rec.fields,
});
}
}
Ok(CsvModel {
doc_len: source.len() as u64,
dialect,
has_header: true,
records,
})
}
pub fn build_csv_model(source: &[u8], limits: Limits) -> Result<Vec<u8>> {
Ok(parse(source, limits, true)?.encode())
}
pub fn record_at(
source: &[u8],
dialect: Dialect,
index: u32,
limits: Limits,
) -> Result<StreamRecord> {
if source.len() as u64 > limits.max_csv_document_bytes {
return Err(Error::resource_limit(format!(
"CSV source is {} bytes, above the {}-byte document cap",
source.len(),
limits.max_csv_document_bytes
)));
}
if index as u64 >= limits.max_csv_rows as u64 {
return Err(Error::resource_limit(format!(
"CSV record index {index} is at or above the {}-record cap",
limits.max_csv_rows
)));
}
let mut scanner = Scanner::new(source, dialect, limits);
let mut i: u32 = 0;
loop {
match scanner.next_record(i != index)? {
Some(rec) => {
if i == index {
return Ok(StreamRecord {
start: rec.start,
end: rec.end,
terminator: rec.terminator,
fields: rec.fields,
});
}
i += 1;
}
None => {
return Err(Error::unsupported_feature(format!(
"CSV table has no record {index}"
)));
}
}
}
}
pub fn field_bytes<'a>(source: &'a [u8], field: &CsvField) -> Result<&'a [u8]> {
let s = usize::try_from(field.start).map_err(|_| corrupt("span overflow"))?;
let e = usize::try_from(field.end).map_err(|_| corrupt("span overflow"))?;
source
.get(s..e)
.ok_or_else(|| corrupt("field span is outside the source"))
}
pub fn record_bytes<'a>(source: &'a [u8], record: &StreamRecord) -> Result<&'a [u8]> {
let s = usize::try_from(record.start).map_err(|_| corrupt("span overflow"))?;
let e = usize::try_from(record.end).map_err(|_| corrupt("span overflow"))?;
source
.get(s..e)
.ok_or_else(|| corrupt("record span is outside the source"))
}
pub fn decode_field(source: &[u8], field: &CsvField) -> Result<String> {
let tok = field_bytes(source, field)?;
if !field.quoted {
return Ok(String::from_utf8_lossy(tok).into_owned());
}
if tok.len() < 2 || tok[0] != QUOTE || tok[tok.len() - 1] != QUOTE {
return Err(corrupt("quoted field is not a well-formed quoted token"));
}
let inner = &tok[1..tok.len() - 1];
let mut out: Vec<u8> = Vec::with_capacity(inner.len());
let mut i = 0usize;
while i < inner.len() {
if inner[i] == QUOTE && inner.get(i + 1) == Some("E) {
out.push(QUOTE);
i += 2;
} else {
out.push(inner[i]);
i += 1;
}
}
Ok(String::from_utf8_lossy(&out).into_owned())
}
pub fn decode_record_fields(source: &[u8], record: &StreamRecord) -> Result<Vec<String>> {
record
.fields
.iter()
.map(|f| decode_field(source, f))
.collect()
}
pub fn canonical_text(
source: &[u8],
dialect: Dialect,
limits: Limits,
max_out: u64,
) -> Result<String> {
let mut scanner = Scanner::new(source, dialect, limits);
let mut out = String::new();
let mut first = true;
while let Some(rec) = scanner.next_record(false)? {
if !first {
out.push('\n');
}
first = false;
for (i, f) in rec.fields.iter().enumerate() {
if i > 0 {
out.push(dialect.delimiter as char);
}
out.push_str(&String::from_utf8_lossy(field_bytes(source, f)?));
if out.len() as u64 > max_out {
return Err(Error::resource_limit(format!(
"CSV text projection exceeds the {max_out}-byte budget"
)));
}
}
}
Ok(out)
}
pub fn find(
source: &[u8],
dialect: Dialect,
pattern: &str,
limits: Limits,
max_out: u64,
) -> Result<Vec<CsvMatch>> {
let mut scanner = Scanner::new(source, dialect, limits);
let mut out: Vec<CsvMatch> = Vec::new();
let mut estimated: u64 = 0;
let mut record: u32 = 0;
while let Some(rec) = scanner.next_record(false)? {
for (col, f) in rec.fields.iter().enumerate() {
let text = decode_field(source, f)?;
if text.contains(pattern) {
estimated = estimated.saturating_add(32 + text.len() as u64);
if estimated > max_out {
return Err(Error::resource_limit(format!(
"CSV find exceeded the {max_out}-byte budget"
)));
}
out.push(CsvMatch {
record,
column: col as u32,
quoted: f.quoted,
start: f.start,
end: f.end,
text,
});
}
}
record = record.saturating_add(1);
if record as u64 > MAX_MODEL_RECORDS as u64 {
return Err(Error::resource_limit(
"CSV find exceeded the record scan cap",
));
}
}
Ok(out)
}
struct RawRecord {
start: u64,
end: u64,
terminator: u8,
nfields: u64,
fields: Vec<CsvField>,
}
struct Scanner<'a> {
b: &'a [u8],
at: usize,
dialect: Dialect,
limits: Limits,
terminator: u8,
terminator_seen: bool,
mixed_terminators: bool,
}
impl<'a> Scanner<'a> {
fn new(b: &'a [u8], dialect: Dialect, limits: Limits) -> Self {
Scanner {
b,
at: dialect.bom_len as usize,
dialect,
limits,
terminator: dialect.terminator,
terminator_seen: false,
mixed_terminators: false,
}
}
#[inline]
fn peek(&self) -> Option<u8> {
self.b.get(self.at).copied()
}
fn consume_eol(&mut self) -> u8 {
let tag = if self.b[self.at] == b'\r' {
if self.b.get(self.at + 1) == Some(&b'\n') {
self.at += 2;
TERM_CRLF
} else {
self.at += 1;
TERM_CR
}
} else {
self.at += 1;
TERM_LF
};
if !self.terminator_seen {
self.terminator = tag;
self.terminator_seen = true;
} else if tag != self.terminator {
self.mixed_terminators = true;
}
tag
}
fn next_record(&mut self, count_only: bool) -> Result<Option<RawRecord>> {
if self.at >= self.b.len() {
return Ok(None);
}
let d = self.dialect.delimiter;
let q = self.dialect.quote;
let start = self.at as u64;
let mut fields: Vec<CsvField> = Vec::new();
let mut nfields: u64 = 0;
let end: u64;
let terminator: u8;
loop {
let field_start = self.at;
let quoted;
if self.peek() == Some(q) {
quoted = true;
self.at += 1;
loop {
let c = self
.peek()
.ok_or_else(|| corrupt("unterminated quoted field"))?;
if c == q {
if self.b.get(self.at + 1) == Some(&q) {
self.at += 2; } else {
self.at += 1; break;
}
} else {
self.at += 1; }
}
} else {
quoted = false;
while let Some(c) = self.peek() {
if c == d || c == b'\r' || c == b'\n' {
break;
}
self.at += 1;
}
}
let field_end = self.at;
if !count_only {
if field_end - field_start > self.limits.max_csv_field_bytes as usize {
return Err(Error::resource_limit(format!(
"CSV field exceeds the {}-byte cap",
self.limits.max_csv_field_bytes
)));
}
if nfields >= self.limits.max_csv_cols as u64 {
return Err(Error::resource_limit(format!(
"CSV record exceeds the {}-column cap",
self.limits.max_csv_cols
)));
}
fields.push(CsvField {
start: field_start as u64,
end: field_end as u64,
quoted,
});
}
nfields += 1;
match self.peek() {
Some(c) if c == d => {
self.at += 1;
}
Some(b'\r') | Some(b'\n') => {
end = field_end as u64;
terminator = self.consume_eol();
break;
}
Some(_) => {
return Err(corrupt(
"unexpected byte after a closing quote (expected a delimiter or end of record)",
));
}
None => {
end = self.at as u64;
terminator = TERM_LF;
break;
}
}
}
if !count_only && end - start > self.limits.max_csv_record_bytes as u64 {
return Err(Error::resource_limit(format!(
"CSV record exceeds the {}-byte cap",
self.limits.max_csv_record_bytes
)));
}
Ok(Some(RawRecord {
start,
end,
terminator,
nfields,
fields: if count_only { Vec::new() } else { fields },
}))
}
}
fn corrupt(msg: &str) -> Error {
Error::invalid_csv_structure(format!("malformed CSV: {msg}"))
}
struct BinReader<'a> {
b: &'a [u8],
at: usize,
}
impl<'a> BinReader<'a> {
fn new(b: &'a [u8]) -> Self {
BinReader { b, at: 0 }
}
fn bytes(&mut self, n: usize) -> Result<&'a [u8]> {
let end = self
.at
.checked_add(n)
.ok_or_else(|| corrupt("reader overflow"))?;
let s = self
.b
.get(self.at..end)
.ok_or_else(|| corrupt("reader underrun"))?;
self.at = end;
Ok(s)
}
fn u8(&mut self) -> Result<u8> {
Ok(self.bytes(1)?[0])
}
fn u32(&mut self) -> Result<u32> {
let b = self.bytes(4)?;
Ok(u32::from_le_bytes([b[0], b[1], b[2], b[3]]))
}
fn u64(&mut self) -> Result<u64> {
let b = self.bytes(8)?;
Ok(u64::from_le_bytes([
b[0], b[1], b[2], b[3], b[4], b[5], b[6], b[7],
]))
}
fn at_end(&self) -> bool {
self.at == self.b.len()
}
}
#[cfg(test)]
mod tests {
use super::*;
fn dialect(source: &[u8]) -> Dialect {
sniff_dialect(source, Limits::DEFAULT).unwrap()
}
#[test]
fn detects_tables_and_rejects_plain_text() {
assert!(detect(b"a,b\nc,d\n", Limits::DEFAULT));
assert!(detect(b"a\tb\nc\td\n", Limits::DEFAULT));
assert!(!detect(b"hello\nworld\n", Limits::DEFAULT));
assert!(!detect(b"a,b,c", Limits::DEFAULT));
assert!(!detect(
b"Hello, world.\nThis is a test.\n",
Limits::DEFAULT
));
assert!(!detect(b"", Limits::DEFAULT));
assert!(!detect(b"{\"a\":1,\"b\":2}", Limits::DEFAULT));
}
#[test]
fn reports_dialect() {
let d = dialect(b"a,b\r\nc,d\r\n");
assert_eq!(d.delimiter, DELIM_COMMA);
assert_eq!(d.terminator, TERM_CRLF);
assert_eq!(d.bom_len, 0);
let d = dialect(b"a\tb\nc\td\n");
assert_eq!(d.delimiter_name(), "tab");
let d = dialect(b"\xEF\xBB\xBFa,b\nc,d\n");
assert_eq!(d.bom_len, 3);
}
#[test]
fn preserves_quotes_embedded_delimiters_and_newlines() {
let src = b"a,\"b,c\",d\n\"x\ny\",2,3\n";
let m = parse(src, Limits::DEFAULT, true).unwrap();
assert_eq!(m.records.len(), 2);
assert_eq!(m.records[0].fields.len(), 3);
let f = m.records[0].fields[1];
assert!(f.quoted);
assert_eq!(field_bytes(src, &f).unwrap(), b"\"b,c\"");
assert_eq!(decode_field(src, &f).unwrap(), "b,c");
assert_eq!(
record_bytes(src, &stream(m.records[1].clone())).unwrap(),
b"\"x\ny\",2,3"
);
}
fn stream(r: CsvRecord) -> StreamRecord {
StreamRecord {
start: r.start,
end: r.end,
terminator: TERM_LF,
fields: r.fields,
}
}
#[test]
fn unescapes_doubled_quotes() {
let src = b"\"a\"\"b\",c\n1,2\n";
let d = dialect(src);
let r = record_at(src, d, 0, Limits::DEFAULT).unwrap();
assert_eq!(decode_field(src, &r.fields[0]).unwrap(), "a\"b");
assert_eq!(field_bytes(src, &r.fields[0]).unwrap(), b"\"a\"\"b\"");
}
#[test]
fn ragged_rows_are_preserved() {
let src = b"a,b,c\n1,2\n3,4,5\n";
let m = parse(src, Limits::DEFAULT, true).unwrap();
assert_eq!(m.records[0].fields.len(), 3);
assert_eq!(m.records[1].fields.len(), 2);
assert_eq!(m.records[2].fields.len(), 3);
}
#[test]
fn model_roundtrips_and_fails_closed() {
let src = b"a,b\r\n\"c\",d\r\n";
let m = parse(src, Limits::DEFAULT, true).unwrap();
let enc = m.encode();
assert_eq!(CsvModel::decode(&enc).unwrap(), m);
let mut bad = enc.clone();
bad[0] = b'X';
assert!(CsvModel::decode(&bad).is_err());
let mut truncated = enc;
truncated.truncate(truncated.len() - 1);
assert!(CsvModel::decode(&truncated).is_err());
}
#[test]
fn column_cap_declines_typed() {
let mut line = String::new();
for _ in 0..5000 {
line.push_str("x,");
}
line.push_str("x\n");
let src = format!("{line}{line}").into_bytes();
let e = parse(&src, Limits::STRICT, true).unwrap_err();
assert_eq!(e.class(), crate::error::ErrorClass::ResourceLimit);
}
#[test]
fn random_bytes_never_panic() {
let mut x: u64 = 0x9E37_79B9_7F4A_7C15;
for _ in 0..128 {
let mut buf = vec![0u8; 512];
for b in buf.iter_mut() {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
*b = (x & 0xFF) as u8;
}
let _ = detect(&buf, Limits::DEFAULT);
if let Ok(d) = sniff_dialect(&buf, Limits::DEFAULT) {
let _ = canonical_text(&buf, d, Limits::STRICT, 1 << 20);
let _ = build_csv_model(&buf, Limits::STRICT);
}
}
}
}