diff --git a/CHANGELOG.md b/CHANGELOG.md index 2fe67d5..23136cb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ - Add Pipeline select method - Add Pipeline `from_pipelines` constructor for merging pipelines together - Remember row order in transform_into +- Include source index in errors ## 0.2.0 - 2023 Jan 11 - Add `Target` struct helper for creating targets, and hide the targets in the `target` module. diff --git a/src/headers.rs b/src/headers.rs index 02850f6..eb39c9e 100644 --- a/src/headers.rs +++ b/src/headers.rs @@ -1,6 +1,7 @@ -use crate::{Error, Row}; +use crate::Row; use csv::StringRecordIter; use std::collections::BTreeMap; +use std::fmt; /// The headers of a CSV file #[derive(Debug, Clone, PartialEq)] @@ -8,6 +9,19 @@ pub struct Headers { indexes: BTreeMap, row: Row, } +pub enum RenameError { + DuplicateColumn(usize), + MissingColumn, +} +impl fmt::Display for RenameError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + RenameError::DuplicateColumn(index) => write!(f, "Duplicate column at index {}", index), + RenameError::MissingColumn => write!(f, "Missing column"), + } + } +} + impl Headers { pub fn new() -> Self { Headers { @@ -17,13 +31,13 @@ impl Headers { } /// Returns `Error::MissingColumn` if `from` is non-existant or `Error::DuplicateColumn` the new name already exists - pub fn rename(&mut self, from: &str, to: &str) -> Result<(), Error> { - if self.contains(to) { - return Err(Error::DuplicateColumn(to.to_string())); + pub fn rename(&mut self, from: &str, to: &str) -> Result<(), RenameError> { + if let Some(index) = self.get_index(to) { + return Err(RenameError::DuplicateColumn(index)); } let index = match self.indexes.remove(from) { Some(index) => index, - None => return Err(Error::MissingColumn(from.to_string())), + None => return Err(RenameError::MissingColumn), }; self.indexes.insert(to.to_string(), index); let mut row_vec: Vec<_> = self.row.into_iter().collect(); @@ -60,13 +74,13 @@ impl Headers { &self.row } - /// Returns `Error::DuplicateColumn` if a column is duplicated - pub fn from_row(row: Row) -> Result { + /// If a column is duplicated, errors with the column name + pub fn from_row(row: Row) -> Result { let mut header = Headers::new(); for field in &row { let added = header.push_field(field); if !added { - return Err(Error::DuplicateColumn(field.to_string())); + return Err(field.to_string()); } } Ok(header) diff --git a/src/lib.rs b/src/lib.rs index 3a3f2f2..147115c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -113,10 +113,20 @@ pub type Row = csv::StringRecord; pub type RowResult = Result; #[derive(Debug)] -pub enum Error { - /// cSV and IO errors are in here +pub struct Error { + pub source: usize, + pub kind: ErrorKind, +} +impl Error { + pub fn new(source: usize, kind: ErrorKind) -> Error { + Error { source, kind } + } +} + +#[derive(Debug)] +pub enum ErrorKind { + /// CSV and IO errors are in here. Csv(csv::Error), - Io(std::io::Error), /// The column of this name is missing. MissingColumn(String), /// This column name appears twice. @@ -126,14 +136,3 @@ pub enum Error { /// Two pipeline sources don't have the same headers. MismatchedHeaders(Row, Row), } -impl From for Error { - fn from(error: csv::Error) -> Error { - Error::Csv(error) - } -} - -impl From for Error { - fn from(error: std::io::Error) -> Error { - Error::Io(error) - } -} diff --git a/src/pipeline.rs b/src/pipeline.rs index 83092ad..4bc80ee 100644 --- a/src/pipeline.rs +++ b/src/pipeline.rs @@ -4,7 +4,7 @@ use crate::pipeline_iterators::{ }; use crate::target::{StringTarget, Target}; use crate::transform::Transform; -use crate::{Error, Row, RowResult}; +use crate::{Error, ErrorKind, Row, RowResult}; use csv::{Reader, ReaderBuilder, StringRecordsIntoIter}; use linked_hash_map::LinkedHashMap; use std::borrow::BorrowMut; @@ -14,15 +14,22 @@ use std::path::Path; /// The main thing pub struct Pipeline<'a> { pub headers: Headers, + pub(crate) source: usize, iterator: Box + 'a>, } impl<'a> Pipeline<'a> { pub fn from_reader(mut reader: Reader) -> Result { let headers_row = reader.headers().unwrap().clone(); - let row_iterator = RowIter::from_records(reader.into_records()); + let row_iterator = RowIter::from_records(0, reader.into_records()); Ok(Pipeline { - headers: Headers::from_row(headers_row)?, + headers: match Headers::from_row(headers_row) { + Ok(headers) => headers, + Err(duplicated_col) => { + return Err(Error::new(0, ErrorKind::DuplicateColumn(duplicated_col))) + } + }, + source: 0, iterator: Box::new(row_iterator), }) } @@ -67,6 +74,7 @@ impl<'a> Pipeline<'a> { }; Pipeline { headers: headers.clone(), + source: 0, iterator: Box::new(PipelinesChain { pipelines, current: current.map(|p| p.build()), @@ -164,6 +172,7 @@ impl<'a> Pipeline<'a> { iterator: self.iterator, f: get_value, name: col.to_string(), + source: self.source, index: self.headers.get_index(col), }); self @@ -189,6 +198,7 @@ impl<'a> Pipeline<'a> { self.iterator = Box::new(Select { iterator: self.iterator, columns: columns.into_iter().map(String::from).collect(), + source: self.source, headers: self.headers.clone(), }); self.headers = Headers::from_row(new_header_row).unwrap(); @@ -213,7 +223,7 @@ impl<'a> Pipeline<'a> { pub fn rename_col(mut self, from: &str, to: &str) -> Self { match self.headers.rename(from, to) { Ok(()) => (), - Err(e) => panic!("{:?}", e), + Err(e) => panic!("Error renaming column in source {}: {}", self.source, e), }; self } @@ -282,11 +292,13 @@ impl<'a> Pipeline<'a> { let names: Vec<_> = hashers.iter().map(|hasher| hasher.name()).collect(); Pipeline { headers: Headers::from_row(Row::from(names)).unwrap(), + source: self.source, iterator: Box::new(TransformInto { iterator: self.iterator, groups: LinkedHashMap::new(), hashers: get_transformers(), get_transformers, + source: self.source, headers: self.headers.clone(), }), } @@ -312,6 +324,7 @@ impl<'a> Pipeline<'a> { name: name.to_string(), iterator: self.iterator, f, + source: self.source, headers: self.headers.clone(), }); self @@ -334,7 +347,7 @@ impl<'a> Pipeline<'a> { /// assert_eq!(csv, "A,B\n1,2\n"); /// ``` pub fn flush(mut self, target: impl Target + 'a) -> Self { - let flush = Flush::new(self.iterator, target, self.headers.clone()); + let flush = Flush::new(self.iterator, target, self.source, self.headers.clone()); self.iterator = Box::new(flush); self } @@ -405,10 +418,14 @@ impl<'a> Iterator for PipelineIter<'a> { pub struct RowIter { inner: StringRecordsIntoIter, + source: usize, } impl RowIter { - pub fn from_records(records: StringRecordsIntoIter) -> Self { - RowIter { inner: records } + pub fn from_records(source: usize, records: StringRecordsIntoIter) -> Self { + RowIter { + source, + inner: records, + } } } impl Iterator for RowIter { @@ -417,7 +434,7 @@ impl Iterator for RowIter { fn next(&mut self) -> Option { self.inner.next().map(|result| { result.map_err(|err| { - return Error::from(err); + return Error::new(self.source, ErrorKind::Csv(err)); }) }) } @@ -433,8 +450,9 @@ fn from_pipelines_mismatch() { .collect_into_string() .unwrap_err(); - match err { - Error::MismatchedHeaders(h1, h2) => { + assert_eq!(err.source, 2); + match err.kind { + ErrorKind::MismatchedHeaders(h1, h2) => { assert_eq!(h1, Row::from(vec!["A", "B"])); assert_eq!(h2, Row::from(vec!["ID", "Country"])); } diff --git a/src/pipeline_iterators.rs b/src/pipeline_iterators.rs index a359bc8..1f0380a 100644 --- a/src/pipeline_iterators.rs +++ b/src/pipeline_iterators.rs @@ -1,7 +1,7 @@ use super::headers::Headers; use crate::target::Target; use crate::transform::{compute_hash, Transform}; -use crate::{Error, Pipeline, PipelineIter, Row, RowResult}; +use crate::{Error, ErrorKind, Pipeline, PipelineIter, Row, RowResult}; use linked_hash_map::{Entry, LinkedHashMap}; pub struct PipelinesChain<'a, P> { @@ -17,32 +17,38 @@ where type Item = RowResult; fn next(&mut self) -> Option { - match &mut self.current { - Some(current) => match current.next() { - Some(row) => return Some(row), - None => {} - }, + // If current is None, iteration is done + match self.current.as_mut()?.next() { + Some(mut row) => { + if let Err(e) = row.as_mut() { + e.source = self.index; + } + return Some(row); + } None => {} }; + // If current was done, go to the next pipeline match self.pipelines.next() { Some(pipeline) => { - if pipeline.headers.get_row() != self.headers.get_row() { - return Some(Err(Error::MismatchedHeaders( - self.headers.get_row().to_owned(), - pipeline.headers.get_row().to_owned(), + self.index += 1; + self.current = Some(pipeline.build()); + let current = self.current.as_mut().unwrap(); + if current.headers.get_row() != self.headers.get_row() { + return Some(Err(Error::new( + self.index, + ErrorKind::MismatchedHeaders( + self.headers.get_row().to_owned(), + current.headers.get_row().to_owned(), + ), ))); } - self.current = Some(pipeline.build()); - self.index += 1; } None => { self.current = None; + return None; } } - match self.current { - Some(ref mut current) => current.next(), - None => None, - } + self.next() } } @@ -101,6 +107,7 @@ pub struct MapCol Result> { pub iterator: I, pub f: F, pub name: String, + pub source: usize, pub index: Option, } impl Iterator for MapCol @@ -118,11 +125,21 @@ where let mut row_vec: Vec<_> = row.into_iter().collect(); let index = match self.index { Some(index) => index, - None => return Some(Err(Error::MissingColumn(self.name.clone()))), + None => { + return Some(Err(Error::new( + self.source, + ErrorKind::MissingColumn(self.name.clone()), + ))) + } }; let field = match row_vec.get_mut(index) { Some(field) => field, - None => return Some(Err(Error::MissingColumn(self.name.clone()))), + None => { + return Some(Err(Error::new( + self.source, + ErrorKind::MissingColumn(self.name.clone()), + ))) + } }; let new_value = match (self.f)(field) { Ok(value) => value, @@ -136,6 +153,7 @@ where pub struct Select { pub iterator: I, pub columns: Vec, + pub source: usize, pub headers: Headers, } impl Iterator for Select @@ -153,7 +171,12 @@ where for col in &self.columns { let field = match self.headers.get_field(&row, col) { Some(field) => field, - None => return Some(Err(Error::MissingColumn(col.clone()))), + None => { + return Some(Err(Error::new( + self.source, + ErrorKind::MissingColumn(col.clone()), + ))) + } }; selection.push(field); } @@ -169,6 +192,7 @@ where pub groups: LinkedHashMap>>, pub hashers: Vec>, pub get_transformers: F, + pub source: usize, pub headers: Headers, } impl Iterator for TransformInto @@ -188,7 +212,7 @@ where }; let hash = match compute_hash(&self.hashers, &self.headers, &row) { Ok(hash) => hash, - Err(e) => return Some(Err(e)), + Err(e) => return Some(Err(Error::new(self.source, e))), }; match self.groups.entry(hash) { @@ -203,7 +227,7 @@ where for reducer in group_row { let result = reducer.add_row(&self.headers, &row); if let Err(e) = result { - return Some(Err(e)); + return Some(Err(Error::new(self.source, e))); } } } @@ -247,6 +271,7 @@ pub struct ValidateCol { pub name: String, pub iterator: I, pub f: F, + pub source: usize, pub headers: Headers, } impl Iterator for ValidateCol @@ -263,7 +288,12 @@ where }; let field = match self.headers.get_field(&row, &self.name) { Some(field) => field, - None => return Some(Err(Error::MissingColumn(self.name.clone()))), + None => { + return Some(Err(Error::new( + self.source, + ErrorKind::MissingColumn(self.name.clone()), + ))) + } }; match (self.f)(&field) { Ok(()) => Some(Ok(row)), @@ -275,14 +305,16 @@ where pub struct Flush { pub iterator: I, pub target: T, + pub source: usize, /// `None` if headers have been written, `Some` otherwise headers: Option, } impl Flush { - pub fn new(iterator: I, target: T, headers: Headers) -> Self { + pub fn new(iterator: I, target: T, source: usize, headers: Headers) -> Self { Self { iterator, target, + source, headers: Some(headers), } } @@ -298,7 +330,7 @@ where if let Some(headers) = &self.headers { match self.target.write_headers(headers) { Ok(()) => self.headers = None, - Err(e) => return Some(Err(e)), + Err(e) => return Some(Err(Error::new(self.source, ErrorKind::Csv(e)))), } } @@ -308,7 +340,7 @@ where }; let r = match self.target.write_row(&row) { Ok(()) => Some(Ok(row)), - Err(e) => Some(Err(e)), + Err(e) => return Some(Err(Error::new(self.source, ErrorKind::Csv(e)))), }; r } diff --git a/src/target.rs b/src/target.rs index 88ec28e..d5e3894 100644 --- a/src/target.rs +++ b/src/target.rs @@ -1,4 +1,4 @@ -use crate::{Error, Headers, Row}; +use crate::{Headers, Row}; use csv::WriterBuilder; use std::fs::{self, File}; use std::io; @@ -6,8 +6,8 @@ use std::path::PathBuf; pub trait Target { /// Useful for initializations - fn write_headers(&mut self, headers: &Headers) -> Result<(), Error>; - fn write_row(&mut self, row: &Row) -> Result<(), Error>; + fn write_headers(&mut self, headers: &Headers) -> Result<(), csv::Error>; + fn write_row(&mut self, row: &Row) -> Result<(), csv::Error>; } pub struct PathTarget { @@ -23,7 +23,7 @@ impl PathTarget { } } impl Target for PathTarget { - fn write_headers(&mut self, headers: &Headers) -> Result<(), Error> { + fn write_headers(&mut self, headers: &Headers) -> Result<(), csv::Error> { if let Some(parent) = self.path.parent() { fs::create_dir_all(parent)?; } @@ -31,7 +31,7 @@ impl Target for PathTarget { self.writer = Some(csv::Writer::from_path(&self.path)?); self.write_row(headers.get_row()) } - fn write_row(&mut self, row: &Row) -> Result<(), Error> { + fn write_row(&mut self, row: &Row) -> Result<(), csv::Error> { self.writer.as_mut().unwrap().write_record(row)?; Ok(()) } @@ -46,13 +46,13 @@ impl StdoutTarget { } } impl Target for StdoutTarget { - fn write_headers(&mut self, headers: &Headers) -> Result<(), Error> { + fn write_headers(&mut self, headers: &Headers) -> Result<(), csv::Error> { let writer = WriterBuilder::new().from_writer(io::stdout()); self.writer = Some(writer); self.write_row(headers.get_row())?; Ok(()) } - fn write_row(&mut self, row: &Row) -> Result<(), Error> { + fn write_row(&mut self, row: &Row) -> Result<(), csv::Error> { self.writer.as_mut().unwrap().write_record(row)?; Ok(()) } @@ -67,12 +67,12 @@ impl StderrTarget { } } impl Target for StderrTarget { - fn write_headers(&mut self, headers: &Headers) -> Result<(), Error> { + fn write_headers(&mut self, headers: &Headers) -> Result<(), csv::Error> { let writer = WriterBuilder::new().from_writer(io::stderr()); self.writer = Some(writer); self.write_row(headers.get_row()) } - fn write_row(&mut self, row: &Row) -> Result<(), Error> { + fn write_row(&mut self, row: &Row) -> Result<(), csv::Error> { self.writer.as_mut().unwrap().write_record(row)?; Ok(()) } @@ -106,10 +106,10 @@ impl<'a> StringTarget<'a> { } } impl<'a> Target for StringTarget<'a> { - fn write_headers(&mut self, headers: &Headers) -> Result<(), Error> { + fn write_headers(&mut self, headers: &Headers) -> Result<(), csv::Error> { self.write_row(headers.get_row()) } - fn write_row(&mut self, row: &Row) -> Result<(), Error> { + fn write_row(&mut self, row: &Row) -> Result<(), csv::Error> { self.writer.write_record(row)?; Ok(()) } diff --git a/src/transform.rs b/src/transform.rs index 6734611..a15bf54 100644 --- a/src/transform.rs +++ b/src/transform.rs @@ -1,4 +1,4 @@ -use crate::{Error, Headers, Row}; +use crate::{ErrorKind, Headers, Row}; use core::fmt::Display; use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher}; @@ -13,7 +13,7 @@ pub trait Transform { _hasher: &mut DefaultHasher, _headers: &Headers, _row: &Row, - ) -> Result<(), Error> { + ) -> Result<(), ErrorKind> { Ok(()) } @@ -21,7 +21,7 @@ pub trait Transform { fn name(&self) -> String; /// Combine the row with the value - fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error>; + fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), ErrorKind>; /// Turn the current value to a string fn value(&self) -> String; @@ -66,7 +66,7 @@ impl Transformer { /// Reduce the values from this column into a single value using a closure pub fn reduce<'a, R, V>(self, reduce: R, init: V) -> Box where - R: FnMut(V, &str) -> Result + 'a, + R: FnMut(V, &str) -> Result + 'a, V: Display + Clone + 'a, { Box::new(Reduce { @@ -84,10 +84,15 @@ struct KeepUnique { value: String, } impl Transform for KeepUnique { - fn hash(&self, hasher: &mut DefaultHasher, headers: &Headers, row: &Row) -> Result<(), Error> { + fn hash( + &self, + hasher: &mut DefaultHasher, + headers: &Headers, + row: &Row, + ) -> Result<(), ErrorKind> { let field = headers .get_field(row, &self.from_col) - .ok_or(Error::MissingColumn(self.from_col.clone()))?; + .ok_or(ErrorKind::MissingColumn(self.from_col.clone()))?; field.hash(hasher); Ok(()) } @@ -96,10 +101,10 @@ impl Transform for KeepUnique { self.name.clone() } - fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> { + fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), ErrorKind> { self.value = headers .get_field(row, &self.from_col) - .ok_or(Error::MissingColumn(self.from_col.clone()))? + .ok_or(ErrorKind::MissingColumn(self.from_col.clone()))? .to_string(); Ok(()) } @@ -113,7 +118,7 @@ pub(crate) fn compute_hash<'a>( transformers: &Vec>, headers: &Headers, row: &Row, -) -> Result { +) -> Result { let mut hasher = DefaultHasher::new(); for transformer in transformers { let result = transformer.hash(&mut hasher, &headers, &row); @@ -132,13 +137,13 @@ struct Reduce { } impl Transform for Reduce where - F: FnMut(V, &str) -> Result, + F: FnMut(V, &str) -> Result, V: Display + Clone, { - fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> { + fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), ErrorKind> { let field = headers .get_field(row, &self.from_col) - .ok_or(Error::MissingColumn(self.from_col.clone()))? + .ok_or(ErrorKind::MissingColumn(self.from_col.clone()))? .to_string(); self.value = (self.reduce)(self.value.clone(), &field)?; Ok(()) @@ -162,14 +167,14 @@ impl Transform for Sum where V: Display + AddAssign + FromStr + Clone, { - fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> { + fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), ErrorKind> { let field = headers .get_field(row, &self.from_col) - .ok_or(Error::MissingColumn(self.from_col.clone()))? + .ok_or(ErrorKind::MissingColumn(self.from_col.clone()))? .to_string(); let new: V = match field.parse() { Ok(v) => v, - Err(_) => return Err(Error::InvalidField(field)), + Err(_) => return Err(ErrorKind::InvalidField(field)), }; self.value += new; Ok(())