use super::super::schema::FileSpec;
use super::super::table::{push_row, RawCsv};
use super::knobs::{
first_duplicate, get_bool, get_string, get_string_list, get_string_map, get_usize, Extra,
};
use super::{FormatSpec, Source};
use std::fs::File;
use std::io::{BufRead, BufReader, Read};
use std::path::PathBuf;
pub const ACCEPTED_FILE_KEYS_DELIMITED: &[&str] = &[
"path",
"format",
"delimiter",
"quote",
"header",
"columns",
"skip_lines",
"comment_prefix",
"line_suffix",
"encoding",
"prefix_strip",
];
pub const KNOB_KEYS_DELIMITED: &[&str] = &[
"delimiter",
"quote",
"header",
"columns",
"skip_lines",
"comment_prefix",
"line_suffix",
"encoding",
"prefix_strip",
];
pub const FORMAT: FormatSpec = FormatSpec {
name: "delimited",
accepted_keys: ACCEPTED_FILE_KEYS_DELIMITED,
knob_keys: KNOB_KEYS_DELIMITED,
validate_entry,
};
fn validate_entry(name: &str, file: &FileSpec) -> Result<(), String> {
DelimitedConfig::from_spec(name, file).map(|_| ())
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Encoding {
Utf8,
Latin1,
}
#[derive(Clone, Debug)]
pub struct DelimitedConfig {
delimiter: String,
quote: Option<u8>,
columns: Option<Vec<String>>,
skip_lines: usize,
comment_prefix: Option<String>,
line_suffix: Option<String>,
encoding: Encoding,
prefix_strip: Vec<(String, String)>,
}
impl DelimitedConfig {
pub fn from_spec(name: &str, file: &FileSpec) -> Result<Self, String> {
let extra = &file.extra;
let delimiter = get_string(name, extra, "delimiter")?.ok_or_else(|| {
format!(
"files '{name}': a 'delimited' input needs a 'delimiter' — the text between two \
fields, e.g. \"\\t\" or \"\\t|\\t\"."
)
})?;
if delimiter.is_empty() {
return Err(format!(
"files '{name}': 'delimiter' is empty — there is no text between two fields to \
split on."
));
}
let quote = Self::parse_quote(name, extra, &delimiter)?;
let header = get_bool(name, extra, "header")?.unwrap_or(true);
let columns = Self::parse_columns(name, extra, header)?;
let comment_prefix = Self::non_empty(name, extra, "comment_prefix")?;
let line_suffix = Self::non_empty(name, extra, "line_suffix")?;
let prefix_strip = Self::parse_prefix_strip(name, extra)?;
Ok(Self {
delimiter,
quote,
columns,
skip_lines: get_usize(name, extra, "skip_lines", "lines")?.unwrap_or(0),
comment_prefix,
line_suffix,
encoding: Self::parse_encoding(name, extra)?,
prefix_strip,
})
}
fn non_empty(name: &str, extra: &Extra, key: &str) -> Result<Option<String>, String> {
match get_string(name, extra, key)? {
Some(s) if s.is_empty() => Err(format!(
"files '{name}': '{key}' is empty — drop the key instead, an empty one matches \
every line."
)),
other => Ok(other),
}
}
fn parse_quote(name: &str, extra: &Extra, delimiter: &str) -> Result<Option<u8>, String> {
let Some(quote) = get_string(name, extra, "quote")? else {
return Ok(None);
};
let bytes = quote.as_bytes();
if bytes.len() != 1 {
return Err(format!(
"files '{name}': 'quote' must be a single ASCII character, but it is '{quote}'."
));
}
if delimiter.len() > 1 {
return Err(format!(
"files '{name}': 'quote' cannot be used with the multi-character delimiter \
'{}' — a file split on a multi-character separator is read line by line, with \
no quoting. Drop 'quote'.",
delimiter.escape_debug()
));
}
Ok(Some(bytes[0]))
}
fn parse_columns(
name: &str,
extra: &Extra,
header: bool,
) -> Result<Option<Vec<String>>, String> {
let columns = get_string_list(name, extra, "columns")?;
match (&columns, header) {
(Some(_), true) => Err(format!(
"files '{name}': 'columns' is set together with \"header\": true — the file's \
own header row would name the columns and 'columns' would be ignored. Set \
\"header\": false to name them here."
)),
(None, false) => Err(format!(
"files '{name}': \"header\": false needs 'columns' — a file with no header row \
has nothing to name its columns."
)),
(Some(c), false) if c.is_empty() => Err(format!(
"files '{name}': 'columns' is empty — a headerless file needs one name per \
column."
)),
(Some(c), false) => match first_duplicate(c) {
Some(dup) => Err(format!(
"files '{name}': 'columns' names '{dup}' twice — a spec asking for that \
column would always read the first one."
)),
None => Ok(columns),
},
(None, true) => Ok(None),
}
}
fn parse_encoding(name: &str, extra: &Extra) -> Result<Encoding, String> {
let Some(raw) = get_string(name, extra, "encoding")? else {
return Ok(Encoding::Utf8);
};
let normalised: String = raw
.chars()
.filter(|c| *c != '-' && *c != '_' && *c != ' ')
.flat_map(char::to_lowercase)
.collect();
match normalised.as_str() {
"utf8" => Ok(Encoding::Utf8),
"latin1" | "iso88591" => Ok(Encoding::Latin1),
_ => Err(format!(
"files '{name}': unknown encoding '{raw}' — this build decodes 'utf-8' (the \
default) and 'latin-1'."
)),
}
}
fn parse_prefix_strip(name: &str, extra: &Extra) -> Result<Vec<(String, String)>, String> {
let entries = get_string_map(name, extra, "prefix_strip")?.unwrap_or_default();
for (column, prefix) in &entries {
if prefix.is_empty() {
return Err(format!(
"files '{name}': 'prefix_strip' entry '{column}' is empty — every cell starts \
with an empty prefix, so there is nothing to strip."
));
}
}
Ok(entries)
}
fn single_byte_delimiter(&self) -> Option<u8> {
match self.delimiter.as_bytes() {
[b] => Some(*b),
_ => None,
}
}
}
pub struct DelimitedFile {
path: PathBuf,
display: String,
config: DelimitedConfig,
}
impl DelimitedFile {
pub fn new(path: PathBuf, display: String, config: DelimitedConfig) -> Self {
Self {
path,
display,
config,
}
}
fn open(&self) -> Result<(RowReader, Vec<String>), String> {
let file =
File::open(&self.path).map_err(|e| format!("delimited open {}: {e}", self.display))?;
let lines = LineFilter::new(BufReader::new(file), &self.config, &self.display);
let engine = match self.config.single_byte_delimiter() {
Some(delimiter) => {
let mut builder = ::csv::ReaderBuilder::new();
builder
.delimiter(delimiter)
.has_headers(false)
.flexible(true);
if let Some(quote) = self.config.quote {
builder.quote(quote);
}
RowEngine::Csv {
reader: builder.from_reader(FilteredRead::new(lines)),
record: ::csv::StringRecord::new(),
}
}
None => RowEngine::Lines {
lines,
delimiter: self.config.delimiter.clone(),
current: String::new(),
},
};
let mut reader = RowReader {
engine,
display: self.display.clone(),
};
let headers = match &self.config.columns {
Some(columns) => columns.clone(),
None if reader.advance()? => reader.fields().map(str::to_string).collect(),
None => Vec::new(),
};
Ok((reader, headers))
}
fn strip_slots(&self, headers: &[String]) -> Vec<Option<String>> {
let mut slots = vec![None; headers.len()];
for (column, prefix) in &self.config.prefix_strip {
if let Some(i) = headers.iter().position(|h| h == column) {
slots[i] = Some(prefix.clone());
}
}
slots
}
}
fn strip_cell<'a>(cell: &'a str, slot: Option<&Option<String>>) -> &'a str {
match slot {
Some(Some(prefix)) => cell.strip_prefix(prefix.as_str()).unwrap_or(cell),
_ => cell,
}
}
impl Source for DelimitedFile {
fn display_name(&self) -> &str {
&self.display
}
fn size_hint(&self) -> Option<u64> {
std::fs::metadata(&self.path).ok().map(|m| m.len())
}
fn can_chunk(&self) -> bool {
true
}
fn read_all(&self) -> Result<RawCsv, String> {
let (mut reader, headers) = self.open()?;
let slots = self.strip_slots(&headers);
let n_cols = headers.len();
let mut rows = Vec::new();
let mut nulls = Vec::new();
while reader.advance()? {
push_stripped_row(reader.fields(), n_cols, &slots, &mut rows, &mut nulls);
}
let row_ids: Vec<usize> = (1..=rows.len()).collect();
Ok(RawCsv {
headers,
rows,
nulls,
row_ids,
})
}
fn chunks(
&self,
chunk_size: usize,
) -> Result<Box<dyn Iterator<Item = Result<RawCsv, String>> + '_>, String> {
let (mut reader, headers) = self.open()?;
let slots = self.strip_slots(&headers);
let n_cols = headers.len();
let mut next_row_id = 1usize;
let iter = std::iter::from_fn(move || {
let mut rows = Vec::with_capacity(chunk_size);
let mut nulls = Vec::with_capacity(chunk_size);
let mut row_ids = Vec::with_capacity(chunk_size);
for _ in 0..chunk_size {
match reader.advance() {
Ok(true) => {
push_stripped_row(reader.fields(), n_cols, &slots, &mut rows, &mut nulls);
row_ids.push(next_row_id);
next_row_id += 1;
}
Ok(false) => break,
Err(e) => return Some(Err(e)),
}
}
if rows.is_empty() {
return None;
}
debug_assert_eq!(
row_ids.len(),
rows.len(),
"every chunk row carries its source row number"
);
Some(Ok(RawCsv {
headers: headers.clone(),
rows,
nulls,
row_ids,
}))
});
Ok(Box::new(iter))
}
fn scan_columns(
&self,
columns: &[&str],
visit: &mut dyn FnMut(usize, &str) -> bool,
) -> Result<(), String> {
let (mut reader, headers) = self.open()?;
let indices: Vec<Option<usize>> = columns
.iter()
.map(|name| headers.iter().position(|h| h == name))
.collect();
if indices.iter().all(Option::is_none) {
return Ok(());
}
let slots = self.strip_slots(&headers);
while reader.advance()? {
for (slot, idx) in indices.iter().enumerate() {
let Some(idx) = idx else { continue };
let Some(cell) = reader.field(*idx) else {
continue;
};
let cell = strip_cell(cell, slots.get(*idx));
if cell.is_empty() {
continue;
}
if !visit(slot, cell) {
return Ok(());
}
}
}
Ok(())
}
}
fn push_stripped_row<'a>(
fields: impl Iterator<Item = &'a str>,
n_cols: usize,
slots: &[Option<String>],
rows: &mut Vec<Vec<String>>,
nulls: &mut Vec<Vec<bool>>,
) {
if slots.iter().all(Option::is_none) {
push_row(fields, n_cols, rows, nulls);
return;
}
push_row(
fields
.enumerate()
.map(|(i, cell)| strip_cell(cell, slots.get(i))),
n_cols,
rows,
nulls,
);
}
struct RowReader {
engine: RowEngine,
display: String,
}
enum RowEngine {
Csv {
reader: ::csv::Reader<FilteredRead<BufReader<File>>>,
record: ::csv::StringRecord,
},
Lines {
lines: LineFilter<BufReader<File>>,
delimiter: String,
current: String,
},
}
impl RowReader {
fn advance(&mut self) -> Result<bool, String> {
let display = &self.display;
match &mut self.engine {
RowEngine::Csv { reader, record } => reader
.read_record(record)
.map_err(|e| csv_row_error(display, e)),
RowEngine::Lines { lines, current, .. } => match lines.next_line()? {
Some(line) => {
*current = line;
Ok(true)
}
None => Ok(false),
},
}
}
fn fields(&self) -> Fields<'_> {
match &self.engine {
RowEngine::Csv { record, .. } => Fields::Csv(record.iter()),
RowEngine::Lines {
delimiter, current, ..
} => Fields::Lines(current.split(delimiter.as_str())),
}
}
fn field(&self, idx: usize) -> Option<&str> {
match &self.engine {
RowEngine::Csv { record, .. } => record.get(idx),
RowEngine::Lines {
delimiter, current, ..
} => current.split(delimiter.as_str()).nth(idx),
}
}
}
fn csv_row_error(display: &str, e: ::csv::Error) -> String {
if let ::csv::ErrorKind::Io(io) = e.kind() {
if io.kind() == std::io::ErrorKind::InvalidData {
return io.to_string();
}
}
format!("delimited row {display}: {e}")
}
enum Fields<'a> {
Csv(::csv::StringRecordIter<'a>),
Lines(std::str::Split<'a, &'a str>),
}
impl<'a> Iterator for Fields<'a> {
type Item = &'a str;
fn next(&mut self) -> Option<&'a str> {
match self {
Fields::Csv(it) => it.next(),
Fields::Lines(it) => it.next(),
}
}
}
const UTF8_BOM: &[u8] = &[0xEF, 0xBB, 0xBF];
struct LineFilter<R: BufRead> {
inner: R,
encoding: Encoding,
skip_lines: usize,
comment_prefix: Option<String>,
line_suffix: Option<String>,
display: String,
physical: usize,
buf: Vec<u8>,
}
impl<R: BufRead> LineFilter<R> {
fn new(inner: R, config: &DelimitedConfig, display: &str) -> Self {
Self {
inner,
encoding: config.encoding,
skip_lines: config.skip_lines,
comment_prefix: config.comment_prefix.clone(),
line_suffix: config.line_suffix.clone(),
display: display.to_string(),
physical: 0,
buf: Vec::new(),
}
}
fn next_line(&mut self) -> Result<Option<String>, String> {
loop {
self.buf.clear();
let read = self
.inner
.read_until(b'\n', &mut self.buf)
.map_err(|e| format!("delimited read {}: {e}", self.display))?;
if read == 0 {
return Ok(None);
}
self.physical += 1;
let mut bytes: &[u8] = &self.buf;
if let Some(rest) = bytes.strip_suffix(b"\n") {
bytes = rest;
}
if let Some(rest) = bytes.strip_suffix(b"\r") {
bytes = rest;
}
if self.physical == 1 {
if let Some(rest) = bytes.strip_prefix(UTF8_BOM) {
bytes = rest;
}
}
if self.physical <= self.skip_lines {
continue;
}
if bytes.is_empty() {
continue;
}
let line = self.decode(bytes)?;
if let Some(prefix) = &self.comment_prefix {
if line.starts_with(prefix.as_str()) {
continue;
}
}
return Ok(Some(self.trim_suffix(line)));
}
}
fn decode(&self, bytes: &[u8]) -> Result<String, String> {
match self.encoding {
Encoding::Utf8 => std::str::from_utf8(bytes).map(str::to_string).map_err(|e| {
format!(
"delimited {} line {}: {e} — declare \"encoding\": \"latin-1\" if the \
file is not UTF-8.",
self.display, self.physical
)
}),
Encoding::Latin1 => Ok(bytes.iter().map(|b| *b as char).collect()),
}
}
fn trim_suffix(&self, mut line: String) -> String {
if let Some(suffix) = &self.line_suffix {
if line.ends_with(suffix.as_str()) {
line.truncate(line.len() - suffix.len());
}
}
line
}
}
struct FilteredRead<R: BufRead> {
lines: LineFilter<R>,
pending: Vec<u8>,
pos: usize,
done: bool,
}
impl<R: BufRead> FilteredRead<R> {
fn new(lines: LineFilter<R>) -> Self {
Self {
lines,
pending: Vec::new(),
pos: 0,
done: false,
}
}
}
impl<R: BufRead> Read for FilteredRead<R> {
fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
if self.pos >= self.pending.len() {
if self.done {
return Ok(0);
}
match self.lines.next_line() {
Err(e) => return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, e)),
Ok(None) => {
self.done = true;
return Ok(0);
}
Ok(Some(line)) => {
self.pending.clear();
self.pending.extend_from_slice(line.as_bytes());
self.pending.push(b'\n');
self.pos = 0;
}
}
}
let n = (self.pending.len() - self.pos).min(out.len());
out[..n].copy_from_slice(&self.pending[self.pos..self.pos + n]);
self.pos += n;
Ok(n)
}
}
#[cfg(test)]
mod delimited_tests {
use super::*;
use std::io::Write;
fn spec(json: &str) -> FileSpec {
serde_json::from_str(json).expect("files entry parses")
}
fn config(json: &str) -> DelimitedConfig {
DelimitedConfig::from_spec("t", &spec(json)).expect("a valid declaration")
}
fn config_err(json: &str) -> String {
DelimitedConfig::from_spec("t", &spec(json)).expect_err("an invalid declaration")
}
fn file(content: &[u8], json: &str) -> (tempfile::NamedTempFile, DelimitedFile) {
let mut f = tempfile::NamedTempFile::new().unwrap();
f.write_all(content).unwrap();
f.flush().unwrap();
let src = DelimitedFile::new(
f.path().to_path_buf(),
"sample.txt".to_string(),
config(json),
);
(f, src)
}
type Shape = (Vec<String>, Vec<Vec<String>>, Vec<Vec<bool>>, Vec<usize>);
fn shape(raw: &RawCsv) -> Shape {
(
raw.headers.clone(),
raw.rows.clone(),
raw.nulls.clone(),
raw.row_ids.clone(),
)
}
#[test]
fn both_engines_produce_the_same_table() {
let single = b"a\tb\tc\n1\t2\t3\n4\t\t6\n";
let multi = "a<->b<->c\n1<->2<->3\n4<-><->6\n";
let (_f1, csv_engine) = file(single, r#"{"path": "x", "delimiter": "\t"}"#);
let (_f2, line_engine) = file(multi.as_bytes(), r#"{"path": "x", "delimiter": "<->"}"#);
assert_eq!(
shape(&csv_engine.read_all().unwrap()),
shape(&line_engine.read_all().unwrap())
);
let raw = csv_engine.read_all().unwrap();
assert_eq!(raw.headers, vec!["a", "b", "c"]);
assert_eq!(raw.rows[1], vec!["4", "", "6"]);
assert_eq!(raw.nulls[1], vec![false, true, false]);
assert_eq!(raw.row_ids, vec![1, 2]);
}
#[test]
fn the_line_suffix_is_stripped_once_and_makes_no_phantom_column() {
let content = "1\t|\troot\t|\t0\t|\n2\t|\tBacteria\t|\t1\t|\n";
let (_f, src) = file(
content.as_bytes(),
r#"{"path": "x", "delimiter": "\t|\t", "line_suffix": "\t|",
"header": false, "columns": ["id", "name", "rank"]}"#,
);
let raw = src.read_all().unwrap();
assert_eq!(raw.headers, vec!["id", "name", "rank"]);
assert_eq!(raw.rows.len(), 2);
assert_eq!(raw.rows[0], vec!["1", "root", "0"]);
assert_eq!(raw.rows[1], vec!["2", "Bacteria", "1"]);
}
#[test]
fn skip_lines_drops_a_preamble_the_header_sits_under() {
let (_f, src) = file(
b"BugSigDB, License: CC BY 4.0\na,b\n1,2\n3,4\n",
r#"{"path": "x", "delimiter": ",", "skip_lines": 1}"#,
);
let raw = src.read_all().unwrap();
assert_eq!(raw.headers, vec!["a", "b"]);
assert_eq!(raw.rows, vec![vec!["1", "2"], vec!["3", "4"]]);
assert_eq!(raw.row_ids, vec![1, 2]);
}
#[test]
fn comment_prefix_drops_a_line_anywhere_in_the_file() {
let (_f, src) = file(
b"# licence\na,b\n1,2\n# midway\n3,4\n",
r##"{"path": "x", "delimiter": ",", "comment_prefix": "#"}"##,
);
let raw = src.read_all().unwrap();
assert_eq!(raw.headers, vec!["a", "b"]);
assert_eq!(raw.rows, vec![vec!["1", "2"], vec!["3", "4"]]);
assert_eq!(raw.row_ids, vec![1, 2]);
}
#[test]
fn a_blank_line_is_not_a_row_in_either_engine() {
let (_f1, csv_engine) = file(b"a,b\n1,2\n\n3,4\n\n", r#"{"path": "x", "delimiter": ","}"#);
let raw = csv_engine.read_all().unwrap();
assert_eq!(raw.rows, vec![vec!["1", "2"], vec!["3", "4"]]);
assert_eq!(raw.row_ids, vec![1, 2]);
let (_f2, line_engine) = file(
b"a::b\n1::2\n\n3::4\n\n",
r#"{"path": "x", "delimiter": "::"}"#,
);
assert_eq!(shape(&line_engine.read_all().unwrap()), shape(&raw));
}
#[test]
fn a_utf8_bom_is_stripped_by_both_engines() {
let mut single = UTF8_BOM.to_vec();
single.extend_from_slice(b"id\tname\n1\tAlice\n");
let (_f1, csv_engine) = file(&single, r#"{"path": "x", "delimiter": "\t"}"#);
assert_eq!(csv_engine.read_all().unwrap().headers, vec!["id", "name"]);
let mut multi = UTF8_BOM.to_vec();
multi.extend_from_slice("id::name\n1::Alice\n".as_bytes());
let (_f2, line_engine) = file(&multi, r#"{"path": "x", "delimiter": "::"}"#);
assert_eq!(line_engine.read_all().unwrap().headers, vec!["id", "name"]);
}
#[test]
fn prefix_strip_removes_a_prefix_from_its_own_column_only() {
let content = "path:map00010\tcpd:C00022\npath:map00020\tC00024\n";
let (_f, src) = file(
content.as_bytes(),
r#"{"path": "x", "delimiter": "\t", "header": false,
"columns": ["pathway", "compound"],
"prefix_strip": {"pathway": "path:", "compound": "cpd:"}}"#,
);
let raw = src.read_all().unwrap();
assert_eq!(raw.rows[0], vec!["map00010", "C00022"]);
assert_eq!(raw.rows[1], vec!["map00020", "C00024"]);
}
#[test]
fn latin1_decodes_the_high_bytes() {
let content = b"name\nAndr\xe9\n";
let (_f, src) = file(
content,
r#"{"path": "x", "delimiter": ",", "encoding": "latin-1"}"#,
);
assert_eq!(src.read_all().unwrap().rows[0], vec!["André"]);
let (_f2, utf8) = file(content, r#"{"path": "x", "delimiter": ","}"#);
let err = utf8.read_all().err().expect("the same bytes are not UTF-8");
assert!(err.contains("line 2"), "{err}");
assert!(err.contains("latin-1"), "{err}");
}
#[test]
fn a_short_row_is_padded_and_extra_fields_are_dropped_by_both_engines() {
let (_f1, csv_engine) = file(b"a,b,c\n1\n1,2,3,4\n", r#"{"path": "x", "delimiter": ","}"#);
let raw = csv_engine.read_all().unwrap();
assert_eq!(raw.rows[0], vec!["1", "", ""]);
assert_eq!(raw.nulls[0], vec![false, true, true]);
assert_eq!(raw.rows[1], vec!["1", "2", "3"]);
let (_f2, line_engine) = file(
"a::b::c\n1\n1::2::3::4\n".as_bytes(),
r#"{"path": "x", "delimiter": "::"}"#,
);
assert_eq!(shape(&line_engine.read_all().unwrap()), shape(&raw));
}
#[test]
fn crlf_endings_load_through_both_engines() {
let (_f1, csv_engine) = file(b"a,b\r\n1,2\r\n", r#"{"path": "x", "delimiter": ","}"#);
let raw = csv_engine.read_all().unwrap();
assert_eq!(raw.headers, vec!["a", "b"]);
assert_eq!(raw.rows, vec![vec!["1", "2"]]);
let (_f2, line_engine) = file(b"a::b\r\n1::2\r\n", r#"{"path": "x", "delimiter": "::"}"#);
assert_eq!(shape(&line_engine.read_all().unwrap()), shape(&raw));
}
#[test]
fn quoting_is_the_csv_engines_and_the_line_engine_has_none() {
let (_f1, csv_engine) = file(
b"a,b\n\"one, two\",3\n",
r#"{"path": "x", "delimiter": ","}"#,
);
assert_eq!(
csv_engine.read_all().unwrap().rows[0],
vec!["one, two", "3"]
);
let (_f2, line_engine) = file(
"a::b::c\n\"one::two\"::3\n".as_bytes(),
r#"{"path": "x", "delimiter": "::"}"#,
);
assert_eq!(
line_engine.read_all().unwrap().rows[0],
vec!["\"one", "two\"", "3"]
);
}
#[test]
fn chunks_can_be_called_twice_and_yield_identical_batches() {
let mut content = String::from("a<->b\n");
for i in 0..25 {
content.push_str(&format!("{i}<->{i}\n"));
}
let (_f, src) = file(content.as_bytes(), r#"{"path": "x", "delimiter": "<->"}"#);
let batches = |src: &DelimitedFile| -> Vec<(Vec<usize>, Vec<Vec<String>>)> {
src.chunks(10)
.unwrap()
.map(|c| {
let c = c.unwrap();
(c.row_ids.clone(), c.rows.clone())
})
.collect()
};
let first = batches(&src);
assert_eq!(first.len(), 3);
assert_eq!(first[2].0, vec![21, 22, 23, 24, 25]);
assert_eq!(first, batches(&src));
}
#[test]
fn chunks_and_read_all_agree() {
let mut content = String::from("a\tb\n");
for i in 0..17 {
content.push_str(&format!("{i}\t{i}\n"));
}
let (_f, src) = file(content.as_bytes(), r#"{"path": "x", "delimiter": "\t"}"#);
let whole = src.read_all().unwrap();
let mut streamed_rows = Vec::new();
let mut streamed_ids = Vec::new();
for chunk in src.chunks(5).unwrap() {
let chunk = chunk.unwrap();
assert_eq!(chunk.headers, whole.headers);
streamed_rows.extend(chunk.rows);
streamed_ids.extend(chunk.row_ids);
}
assert_eq!(streamed_rows, whole.rows);
assert_eq!(streamed_ids, whole.row_ids);
}
#[test]
fn scan_columns_sees_what_read_all_holds() {
let content = "id::name::score\n1::Alice::3\n2::::4\n3::Cara::\n";
let (_f, src) = file(content.as_bytes(), r#"{"path": "x", "delimiter": "::"}"#);
let raw = src.read_all().unwrap();
let wanted = ["score", "name", "absent"];
let mut expected = Vec::new();
for r in 0..raw.row_count() {
for (slot, name) in wanted.iter().enumerate() {
let Some(idx) = raw.col_index(name) else {
continue;
};
if !raw.nulls[r][idx] {
expected.push((slot, raw.rows[r][idx].clone()));
}
}
}
let mut seen = Vec::new();
src.scan_columns(&wanted, &mut |slot, cell| {
seen.push((slot, cell.to_string()));
true
})
.unwrap();
assert_eq!(seen, expected);
assert_eq!(seen.len(), 4, "{seen:?}");
}
#[test]
fn scan_columns_strips_the_prefixes_too() {
let content = "cpd:1\ncpd:2\n";
let (_f, src) = file(
content.as_bytes(),
r#"{"path": "x", "delimiter": "\t", "header": false, "columns": ["compound"],
"prefix_strip": {"compound": "cpd:"}}"#,
);
let mut seen = Vec::new();
src.scan_columns(&["compound"], &mut |_, cell| {
seen.push(cell.to_string());
true
})
.unwrap();
assert_eq!(seen, vec!["1", "2"]);
}
#[test]
fn scan_columns_stops_when_the_sink_says_stop() {
let (_f, src) = file(b"a\n1\n2\n3\n", r#"{"path": "x", "delimiter": ","}"#);
let mut visits = 0usize;
src.scan_columns(&["a"], &mut |_, _| {
visits += 1;
false
})
.unwrap();
assert_eq!(visits, 1);
}
#[test]
fn a_file_that_is_not_there_reports_its_declared_name() {
let src = DelimitedFile::new(
PathBuf::from("/nonexistent/not-here.dmp"),
"nodes.dmp".to_string(),
config(r#"{"path": "x", "delimiter": "\t"}"#),
);
assert_eq!(src.size_hint(), None);
let err = src.read_all().err().expect("a missing file fails to read");
assert!(err.contains("delimited open nodes.dmp"), "{err}");
}
#[test]
fn a_declaration_without_a_delimiter_says_what_it_is_for() {
let err = config_err(r#"{"path": "x"}"#);
assert!(err.contains("needs a 'delimiter'"), "{err}");
}
#[test]
fn every_knob_is_checked_before_a_file_is_opened() {
for (json, expected) in [
(r#"{"delimiter": ""}"#, "'delimiter' is empty"),
(r#"{"delimiter": 9}"#, "'delimiter' must be a string"),
(r#"{"delimiter": ",", "quote": "''"}"#, "single ASCII"),
(
r#"{"delimiter": "\t|\t", "quote": "\"", "header": false, "columns": ["a"]}"#,
"cannot be used with the multi-character delimiter",
),
(
r#"{"delimiter": ",", "columns": ["a"]}"#,
"together with \"header\": true",
),
(r#"{"delimiter": ",", "header": false}"#, "needs 'columns'"),
(
r#"{"delimiter": ",", "header": false, "columns": []}"#,
"'columns' is empty",
),
(
r#"{"delimiter": ",", "header": false, "columns": ["a", "a"]}"#,
"names 'a' twice",
),
(
r#"{"delimiter": ",", "header": false, "columns": "a"}"#,
"must be a list of column names",
),
(
r#"{"delimiter": ",", "header": false, "columns": [1]}"#,
"every entry of 'columns' must be a column name",
),
(
r#"{"delimiter": ",", "header": 1}"#,
"must be true or false",
),
(r#"{"delimiter": ",", "skip_lines": -1}"#, "not negative"),
(
r#"{"delimiter": ",", "skip_lines": "2"}"#,
"must be a whole number of lines",
),
(
r#"{"delimiter": ",", "comment_prefix": ""}"#,
"'comment_prefix' is empty",
),
(
r#"{"delimiter": ",", "line_suffix": ""}"#,
"'line_suffix' is empty",
),
(r#"{"delimiter": ",", "encoding": "utf-16"}"#, "'latin-1'"),
(
r#"{"delimiter": ",", "prefix_strip": ["a"]}"#,
"must be an object of column name",
),
(
r#"{"delimiter": ",", "prefix_strip": {"a": 1}}"#,
"entry 'a' must be text",
),
(
r#"{"delimiter": ",", "prefix_strip": {"a": ""}}"#,
"entry 'a' is empty",
),
] {
let err = config_err(json);
assert!(err.contains(expected), "{json}\n got: {err}");
assert!(err.starts_with("files 't':"), "{json}\n got: {err}");
}
}
#[test]
fn the_encoding_name_is_spelled_liberally() {
for spelling in ["utf-8", "UTF8", "utf_8"] {
let json = format!(r#"{{"delimiter": ",", "encoding": "{spelling}"}}"#);
assert_eq!(config(&json).encoding, Encoding::Utf8, "{spelling}");
}
for spelling in ["latin-1", "Latin1", "ISO-8859-1"] {
let json = format!(r#"{{"delimiter": ",", "encoding": "{spelling}"}}"#);
assert_eq!(config(&json).encoding, Encoding::Latin1, "{spelling}");
}
}
#[test]
fn the_engine_is_chosen_by_the_delimiters_width() {
assert_eq!(
config(r#"{"delimiter": "\t"}"#).single_byte_delimiter(),
Some(b'\t')
);
assert_eq!(
config(r#"{"delimiter": "\t|\t", "header": false, "columns": ["a"]}"#)
.single_byte_delimiter(),
None
);
assert_eq!(
config(r#"{"delimiter": "→", "header": false, "columns": ["a"]}"#)
.single_byte_delimiter(),
None
);
}
}