+1

CHANGELOG.md

Reviewed

··· 4 4 - Add Pipeline select method 5 5 - Add Pipeline `from_pipelines` constructor for merging pipelines together 6 6 - Remember row order in transform_into 7 7 + - Include source index in errors 7 8 8 9 ## 0.2.0 - 2023 Jan 11 9 10 - Add `Target` struct helper for creating targets, and hide the targets in the `target` module.

+22 -8

src/headers.rs

Reviewed

··· 1 1 - use crate::{Error, Row}; 1 1 + use crate::Row; 2 2 use csv::StringRecordIter; 3 3 use std::collections::BTreeMap; 4 4 + use std::fmt; 4 5 5 6 /// The headers of a CSV file 6 7 #[derive(Debug, Clone, PartialEq)] ··· 8 9 indexes: BTreeMap<String, usize>, 9 10 row: Row, 10 11 } 12 12 + pub enum RenameError { 13 13 + DuplicateColumn(usize), 14 14 + MissingColumn, 15 15 + } 16 16 + impl fmt::Display for RenameError { 17 17 + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 18 18 + match self { 19 19 + RenameError::DuplicateColumn(index) => write!(f, "Duplicate column at index {}", index), 20 20 + RenameError::MissingColumn => write!(f, "Missing column"), 21 21 + } 22 22 + } 23 23 + } 24 24 + 11 25 impl Headers { 12 26 pub fn new() -> Self { 13 27 Headers { ··· 17 31 } 18 32 19 33 /// Returns `Error::MissingColumn` if `from` is non-existant or `Error::DuplicateColumn` the new name already exists 20 20 - pub fn rename(&mut self, from: &str, to: &str) -> Result<(), Error> { 21 21 - if self.contains(to) { 22 22 - return Err(Error::DuplicateColumn(to.to_string())); 34 34 + pub fn rename(&mut self, from: &str, to: &str) -> Result<(), RenameError> { 35 35 + if let Some(index) = self.get_index(to) { 36 36 + return Err(RenameError::DuplicateColumn(index)); 23 37 } 24 38 let index = match self.indexes.remove(from) { 25 39 Some(index) => index, 26 26 - None => return Err(Error::MissingColumn(from.to_string())), 40 40 + None => return Err(RenameError::MissingColumn), 27 41 }; 28 42 self.indexes.insert(to.to_string(), index); 29 43 let mut row_vec: Vec<_> = self.row.into_iter().collect(); ··· 60 74 &self.row 61 75 } 62 76 63 63 - /// Returns `Error::DuplicateColumn` if a column is duplicated 64 64 - pub fn from_row(row: Row) -> Result<Self, Error> { 77 77 + /// If a column is duplicated, errors with the column name 78 78 + pub fn from_row(row: Row) -> Result<Self, String> { 65 79 let mut header = Headers::new(); 66 80 for field in &row { 67 81 let added = header.push_field(field); 68 82 if !added { 69 69 - return Err(Error::DuplicateColumn(field.to_string())); 83 83 + return Err(field.to_string()); 70 84 } 71 85 } 72 86 Ok(header)

+13 -14

src/lib.rs

Reviewed

··· 113 113 pub type RowResult = Result<Row, Error>; 114 114 115 115 #[derive(Debug)] 116 116 - pub enum Error { 117 117 - /// cSV and IO errors are in here 116 116 + pub struct Error { 117 117 + pub source: usize, 118 118 + pub kind: ErrorKind, 119 119 + } 120 120 + impl Error { 121 121 + pub fn new(source: usize, kind: ErrorKind) -> Error { 122 122 + Error { source, kind } 123 123 + } 124 124 + } 125 125 + 126 126 + #[derive(Debug)] 127 127 + pub enum ErrorKind { 128 128 + /// CSV and IO errors are in here. 118 129 Csv(csv::Error), 119 119 - Io(std::io::Error), 120 130 /// The column of this name is missing. 121 131 MissingColumn(String), 122 132 /// This column name appears twice. ··· 126 136 /// Two pipeline sources don't have the same headers. 127 137 MismatchedHeaders(Row, Row), 128 138 } 129 129 - impl From<csv::Error> for Error { 130 130 - fn from(error: csv::Error) -> Error { 131 131 - Error::Csv(error) 132 132 - } 133 133 - } 134 134 - 135 135 - impl From<std::io::Error> for Error { 136 136 - fn from(error: std::io::Error) -> Error { 137 137 - Error::Io(error) 138 138 - } 139 139 - }

+28 -10

src/pipeline.rs

Reviewed

··· 4 4 }; 5 5 use crate::target::{StringTarget, Target}; 6 6 use crate::transform::Transform; 7 7 - use crate::{Error, Row, RowResult}; 7 7 + use crate::{Error, ErrorKind, Row, RowResult}; 8 8 use csv::{Reader, ReaderBuilder, StringRecordsIntoIter}; 9 9 use linked_hash_map::LinkedHashMap; 10 10 use std::borrow::BorrowMut; ··· 14 14 /// The main thing 15 15 pub struct Pipeline<'a> { 16 16 pub headers: Headers, 17 17 + pub(crate) source: usize, 17 18 iterator: Box<dyn Iterator<Item = RowResult> + 'a>, 18 19 } 19 20 20 21 impl<'a> Pipeline<'a> { 21 22 pub fn from_reader<R: io::Read + 'a>(mut reader: Reader<R>) -> Result<Self, Error> { 22 23 let headers_row = reader.headers().unwrap().clone(); 23 23 - let row_iterator = RowIter::from_records(reader.into_records()); 24 24 + let row_iterator = RowIter::from_records(0, reader.into_records()); 24 25 Ok(Pipeline { 25 25 - headers: Headers::from_row(headers_row)?, 26 26 + headers: match Headers::from_row(headers_row) { 27 27 + Ok(headers) => headers, 28 28 + Err(duplicated_col) => { 29 29 + return Err(Error::new(0, ErrorKind::DuplicateColumn(duplicated_col))) 30 30 + } 31 31 + }, 32 32 + source: 0, 26 33 iterator: Box::new(row_iterator), 27 34 }) 28 35 } ··· 67 74 }; 68 75 Pipeline { 69 76 headers: headers.clone(), 77 77 + source: 0, 70 78 iterator: Box::new(PipelinesChain { 71 79 pipelines, 72 80 current: current.map(|p| p.build()), ··· 164 172 iterator: self.iterator, 165 173 f: get_value, 166 174 name: col.to_string(), 175 175 + source: self.source, 167 176 index: self.headers.get_index(col), 168 177 }); 169 178 self ··· 189 198 self.iterator = Box::new(Select { 190 199 iterator: self.iterator, 191 200 columns: columns.into_iter().map(String::from).collect(), 201 201 + source: self.source, 192 202 headers: self.headers.clone(), 193 203 }); 194 204 self.headers = Headers::from_row(new_header_row).unwrap(); ··· 213 223 pub fn rename_col(mut self, from: &str, to: &str) -> Self { 214 224 match self.headers.rename(from, to) { 215 225 Ok(()) => (), 216 216 - Err(e) => panic!("{:?}", e), 226 226 + Err(e) => panic!("Error renaming column in source {}: {}", self.source, e), 217 227 }; 218 228 self 219 229 } ··· 282 292 let names: Vec<_> = hashers.iter().map(|hasher| hasher.name()).collect(); 283 293 Pipeline { 284 294 headers: Headers::from_row(Row::from(names)).unwrap(), 295 295 + source: self.source, 285 296 iterator: Box::new(TransformInto { 286 297 iterator: self.iterator, 287 298 groups: LinkedHashMap::new(), 288 299 hashers: get_transformers(), 289 300 get_transformers, 301 301 + source: self.source, 290 302 headers: self.headers.clone(), 291 303 }), 292 304 } ··· 312 324 name: name.to_string(), 313 325 iterator: self.iterator, 314 326 f, 327 327 + source: self.source, 315 328 headers: self.headers.clone(), 316 329 }); 317 330 self ··· 334 347 /// assert_eq!(csv, "A,B\n1,2\n"); 335 348 /// ``` 336 349 pub fn flush(mut self, target: impl Target + 'a) -> Self { 337 337 - let flush = Flush::new(self.iterator, target, self.headers.clone()); 350 350 + let flush = Flush::new(self.iterator, target, self.source, self.headers.clone()); 338 351 self.iterator = Box::new(flush); 339 352 self 340 353 } ··· 405 418 406 419 pub struct RowIter<R: io::Read> { 407 420 inner: StringRecordsIntoIter<R>, 421 421 + source: usize, 408 422 } 409 423 impl<R: io::Read> RowIter<R> { 410 410 - pub fn from_records(records: StringRecordsIntoIter<R>) -> Self { 411 411 - RowIter { inner: records } 424 424 + pub fn from_records(source: usize, records: StringRecordsIntoIter<R>) -> Self { 425 425 + RowIter { 426 426 + source, 427 427 + inner: records, 428 428 + } 412 429 } 413 430 } 414 431 impl<R: io::Read> Iterator for RowIter<R> { ··· 417 434 fn next(&mut self) -> Option<Self::Item> { 418 435 self.inner.next().map(|result| { 419 436 result.map_err(|err| { 420 420 - return Error::from(err); 437 437 + return Error::new(self.source, ErrorKind::Csv(err)); 421 438 }) 422 439 }) 423 440 } ··· 433 450 .collect_into_string() 434 451 .unwrap_err(); 435 452 436 436 - match err { 437 437 - Error::MismatchedHeaders(h1, h2) => { 453 453 + assert_eq!(err.source, 2); 454 454 + match err.kind { 455 455 + ErrorKind::MismatchedHeaders(h1, h2) => { 438 456 assert_eq!(h1, Row::from(vec!["A", "B"])); 439 457 assert_eq!(h2, Row::from(vec!["ID", "Country"])); 440 458 }

+57 -25

src/pipeline_iterators.rs

Reviewed

··· 1 1 use super::headers::Headers; 2 2 use crate::target::Target; 3 3 use crate::transform::{compute_hash, Transform}; 4 4 - use crate::{Error, Pipeline, PipelineIter, Row, RowResult}; 4 4 + use crate::{Error, ErrorKind, Pipeline, PipelineIter, Row, RowResult}; 5 5 use linked_hash_map::{Entry, LinkedHashMap}; 6 6 7 7 pub struct PipelinesChain<'a, P> { ··· 17 17 type Item = RowResult; 18 18 19 19 fn next(&mut self) -> Option<Self::Item> { 20 20 - match &mut self.current { 21 21 - Some(current) => match current.next() { 22 22 - Some(row) => return Some(row), 23 23 - None => {} 24 24 - }, 20 20 + // If current is None, iteration is done 21 21 + match self.current.as_mut()?.next() { 22 22 + Some(mut row) => { 23 23 + if let Err(e) = row.as_mut() { 24 24 + e.source = self.index; 25 25 + } 26 26 + return Some(row); 27 27 + } 25 28 None => {} 26 29 }; 30 30 + // If current was done, go to the next pipeline 27 31 match self.pipelines.next() { 28 32 Some(pipeline) => { 29 29 - if pipeline.headers.get_row() != self.headers.get_row() { 30 30 - return Some(Err(Error::MismatchedHeaders( 31 31 - self.headers.get_row().to_owned(), 32 32 - pipeline.headers.get_row().to_owned(), 33 33 + self.index += 1; 34 34 + self.current = Some(pipeline.build()); 35 35 + let current = self.current.as_mut().unwrap(); 36 36 + if current.headers.get_row() != self.headers.get_row() { 37 37 + return Some(Err(Error::new( 38 38 + self.index, 39 39 + ErrorKind::MismatchedHeaders( 40 40 + self.headers.get_row().to_owned(), 41 41 + current.headers.get_row().to_owned(), 42 42 + ), 33 43 ))); 34 44 } 35 35 - self.current = Some(pipeline.build()); 36 36 - self.index += 1; 37 45 } 38 46 None => { 39 47 self.current = None; 48 48 + return None; 40 49 } 41 50 } 42 42 - match self.current { 43 43 - Some(ref mut current) => current.next(), 44 44 - None => None, 45 45 - } 51 51 + self.next() 46 52 } 47 53 } 48 54 ··· 101 107 pub iterator: I, 102 108 pub f: F, 103 109 pub name: String, 110 110 + pub source: usize, 104 111 pub index: Option<usize>, 105 112 } 106 113 impl<I, F> Iterator for MapCol<I, F> ··· 118 125 let mut row_vec: Vec<_> = row.into_iter().collect(); 119 126 let index = match self.index { 120 127 Some(index) => index, 121 121 - None => return Some(Err(Error::MissingColumn(self.name.clone()))), 128 128 + None => { 129 129 + return Some(Err(Error::new( 130 130 + self.source, 131 131 + ErrorKind::MissingColumn(self.name.clone()), 132 132 + ))) 133 133 + } 122 134 }; 123 135 let field = match row_vec.get_mut(index) { 124 136 Some(field) => field, 125 125 - None => return Some(Err(Error::MissingColumn(self.name.clone()))), 137 137 + None => { 138 138 + return Some(Err(Error::new( 139 139 + self.source, 140 140 + ErrorKind::MissingColumn(self.name.clone()), 141 141 + ))) 142 142 + } 126 143 }; 127 144 let new_value = match (self.f)(field) { 128 145 Ok(value) => value, ··· 136 153 pub struct Select<I> { 137 154 pub iterator: I, 138 155 pub columns: Vec<String>, 156 156 + pub source: usize, 139 157 pub headers: Headers, 140 158 } 141 159 impl<I> Iterator for Select<I> ··· 153 171 for col in &self.columns { 154 172 let field = match self.headers.get_field(&row, col) { 155 173 Some(field) => field, 156 156 - None => return Some(Err(Error::MissingColumn(col.clone()))), 174 174 + None => { 175 175 + return Some(Err(Error::new( 176 176 + self.source, 177 177 + ErrorKind::MissingColumn(col.clone()), 178 178 + ))) 179 179 + } 157 180 }; 158 181 selection.push(field); 159 182 } ··· 169 192 pub groups: LinkedHashMap<u64, Vec<Box<dyn Transform>>>, 170 193 pub hashers: Vec<Box<dyn Transform>>, 171 194 pub get_transformers: F, 195 195 + pub source: usize, 172 196 pub headers: Headers, 173 197 } 174 198 impl<I, F> Iterator for TransformInto<I, F> ··· 188 212 }; 189 213 let hash = match compute_hash(&self.hashers, &self.headers, &row) { 190 214 Ok(hash) => hash, 191 191 - Err(e) => return Some(Err(e)), 215 215 + Err(e) => return Some(Err(Error::new(self.source, e))), 192 216 }; 193 217 194 218 match self.groups.entry(hash) { ··· 203 227 for reducer in group_row { 204 228 let result = reducer.add_row(&self.headers, &row); 205 229 if let Err(e) = result { 206 206 - return Some(Err(e)); 230 230 + return Some(Err(Error::new(self.source, e))); 207 231 } 208 232 } 209 233 } ··· 247 271 pub name: String, 248 272 pub iterator: I, 249 273 pub f: F, 274 274 + pub source: usize, 250 275 pub headers: Headers, 251 276 } 252 277 impl<I, F> Iterator for ValidateCol<I, F> ··· 263 288 }; 264 289 let field = match self.headers.get_field(&row, &self.name) { 265 290 Some(field) => field, 266 266 - None => return Some(Err(Error::MissingColumn(self.name.clone()))), 291 291 + None => { 292 292 + return Some(Err(Error::new( 293 293 + self.source, 294 294 + ErrorKind::MissingColumn(self.name.clone()), 295 295 + ))) 296 296 + } 267 297 }; 268 298 match (self.f)(&field) { 269 299 Ok(()) => Some(Ok(row)), ··· 275 305 pub struct Flush<I, T> { 276 306 pub iterator: I, 277 307 pub target: T, 308 308 + pub source: usize, 278 309 /// `None` if headers have been written, `Some` otherwise 279 310 headers: Option<Headers>, 280 311 } 281 312 impl<I, T> Flush<I, T> { 282 282 - pub fn new(iterator: I, target: T, headers: Headers) -> Self { 313 313 + pub fn new(iterator: I, target: T, source: usize, headers: Headers) -> Self { 283 314 Self { 284 315 iterator, 285 316 target, 317 317 + source, 286 318 headers: Some(headers), 287 319 } 288 320 } ··· 298 330 if let Some(headers) = &self.headers { 299 331 match self.target.write_headers(headers) { 300 332 Ok(()) => self.headers = None, 301 301 - Err(e) => return Some(Err(e)), 333 333 + Err(e) => return Some(Err(Error::new(self.source, ErrorKind::Csv(e)))), 302 334 } 303 335 } 304 336 ··· 308 340 }; 309 341 let r = match self.target.write_row(&row) { 310 342 Ok(()) => Some(Ok(row)), 311 311 - Err(e) => Some(Err(e)), 343 343 + Err(e) => return Some(Err(Error::new(self.source, ErrorKind::Csv(e)))), 312 344 }; 313 345 r 314 346 }

+11 -11

src/target.rs

Reviewed

··· 1 1 - use crate::{Error, Headers, Row}; 1 1 + use crate::{Headers, Row}; 2 2 use csv::WriterBuilder; 3 3 use std::fs::{self, File}; 4 4 use std::io; ··· 6 6 7 7 pub trait Target { 8 8 /// Useful for initializations 9 9 - fn write_headers(&mut self, headers: &Headers) -> Result<(), Error>; 10 10 - fn write_row(&mut self, row: &Row) -> Result<(), Error>; 9 9 + fn write_headers(&mut self, headers: &Headers) -> Result<(), csv::Error>; 10 10 + fn write_row(&mut self, row: &Row) -> Result<(), csv::Error>; 11 11 } 12 12 13 13 pub struct PathTarget { ··· 23 23 } 24 24 } 25 25 impl Target for PathTarget { 26 26 - fn write_headers(&mut self, headers: &Headers) -> Result<(), Error> { 26 26 + fn write_headers(&mut self, headers: &Headers) -> Result<(), csv::Error> { 27 27 if let Some(parent) = self.path.parent() { 28 28 fs::create_dir_all(parent)?; 29 29 } ··· 31 31 self.writer = Some(csv::Writer::from_path(&self.path)?); 32 32 self.write_row(headers.get_row()) 33 33 } 34 34 - fn write_row(&mut self, row: &Row) -> Result<(), Error> { 34 34 + fn write_row(&mut self, row: &Row) -> Result<(), csv::Error> { 35 35 self.writer.as_mut().unwrap().write_record(row)?; 36 36 Ok(()) 37 37 } ··· 46 46 } 47 47 } 48 48 impl Target for StdoutTarget { 49 49 - fn write_headers(&mut self, headers: &Headers) -> Result<(), Error> { 49 49 + fn write_headers(&mut self, headers: &Headers) -> Result<(), csv::Error> { 50 50 let writer = WriterBuilder::new().from_writer(io::stdout()); 51 51 self.writer = Some(writer); 52 52 self.write_row(headers.get_row())?; 53 53 Ok(()) 54 54 } 55 55 - fn write_row(&mut self, row: &Row) -> Result<(), Error> { 55 55 + fn write_row(&mut self, row: &Row) -> Result<(), csv::Error> { 56 56 self.writer.as_mut().unwrap().write_record(row)?; 57 57 Ok(()) 58 58 } ··· 67 67 } 68 68 } 69 69 impl Target for StderrTarget { 70 70 - fn write_headers(&mut self, headers: &Headers) -> Result<(), Error> { 70 70 + fn write_headers(&mut self, headers: &Headers) -> Result<(), csv::Error> { 71 71 let writer = WriterBuilder::new().from_writer(io::stderr()); 72 72 self.writer = Some(writer); 73 73 self.write_row(headers.get_row()) 74 74 } 75 75 - fn write_row(&mut self, row: &Row) -> Result<(), Error> { 75 75 + fn write_row(&mut self, row: &Row) -> Result<(), csv::Error> { 76 76 self.writer.as_mut().unwrap().write_record(row)?; 77 77 Ok(()) 78 78 } ··· 106 106 } 107 107 } 108 108 impl<'a> Target for StringTarget<'a> { 109 109 - fn write_headers(&mut self, headers: &Headers) -> Result<(), Error> { 109 109 + fn write_headers(&mut self, headers: &Headers) -> Result<(), csv::Error> { 110 110 self.write_row(headers.get_row()) 111 111 } 112 112 - fn write_row(&mut self, row: &Row) -> Result<(), Error> { 112 112 + fn write_row(&mut self, row: &Row) -> Result<(), csv::Error> { 113 113 self.writer.write_record(row)?; 114 114 Ok(()) 115 115 }

+20 -15

src/transform.rs

Reviewed

··· 1 1 - use crate::{Error, Headers, Row}; 1 1 + use crate::{ErrorKind, Headers, Row}; 2 2 use core::fmt::Display; 3 3 use std::collections::hash_map::DefaultHasher; 4 4 use std::hash::{Hash, Hasher}; ··· 13 13 _hasher: &mut DefaultHasher, 14 14 _headers: &Headers, 15 15 _row: &Row, 16 16 - ) -> Result<(), Error> { 16 16 + ) -> Result<(), ErrorKind> { 17 17 Ok(()) 18 18 } 19 19 ··· 21 21 fn name(&self) -> String; 22 22 23 23 /// Combine the row with the value 24 24 - fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error>; 24 24 + fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), ErrorKind>; 25 25 26 26 /// Turn the current value to a string 27 27 fn value(&self) -> String; ··· 66 66 /// Reduce the values from this column into a single value using a closure 67 67 pub fn reduce<'a, R, V>(self, reduce: R, init: V) -> Box<dyn Transform + 'a> 68 68 where 69 69 - R: FnMut(V, &str) -> Result<V, Error> + 'a, 69 69 + R: FnMut(V, &str) -> Result<V, ErrorKind> + 'a, 70 70 V: Display + Clone + 'a, 71 71 { 72 72 Box::new(Reduce { ··· 84 84 value: String, 85 85 } 86 86 impl Transform for KeepUnique { 87 87 - fn hash(&self, hasher: &mut DefaultHasher, headers: &Headers, row: &Row) -> Result<(), Error> { 87 87 + fn hash( 88 88 + &self, 89 89 + hasher: &mut DefaultHasher, 90 90 + headers: &Headers, 91 91 + row: &Row, 92 92 + ) -> Result<(), ErrorKind> { 88 93 let field = headers 89 94 .get_field(row, &self.from_col) 90 90 - .ok_or(Error::MissingColumn(self.from_col.clone()))?; 95 95 + .ok_or(ErrorKind::MissingColumn(self.from_col.clone()))?; 91 96 field.hash(hasher); 92 97 Ok(()) 93 98 } ··· 96 101 self.name.clone() 97 102 } 98 103 99 99 - fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> { 104 104 + fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), ErrorKind> { 100 105 self.value = headers 101 106 .get_field(row, &self.from_col) 102 102 - .ok_or(Error::MissingColumn(self.from_col.clone()))? 107 107 + .ok_or(ErrorKind::MissingColumn(self.from_col.clone()))? 103 108 .to_string(); 104 109 Ok(()) 105 110 } ··· 113 118 transformers: &Vec<Box<dyn Transform + 'a>>, 114 119 headers: &Headers, 115 120 row: &Row, 116 116 - ) -> Result<u64, Error> { 121 121 + ) -> Result<u64, ErrorKind> { 117 122 let mut hasher = DefaultHasher::new(); 118 123 for transformer in transformers { 119 124 let result = transformer.hash(&mut hasher, &headers, &row); ··· 132 137 } 133 138 impl<F, V> Transform for Reduce<F, V> 134 139 where 135 135 - F: FnMut(V, &str) -> Result<V, Error>, 140 140 + F: FnMut(V, &str) -> Result<V, ErrorKind>, 136 141 V: Display + Clone, 137 142 { 138 138 - fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> { 143 143 + fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), ErrorKind> { 139 144 let field = headers 140 145 .get_field(row, &self.from_col) 141 141 - .ok_or(Error::MissingColumn(self.from_col.clone()))? 146 146 + .ok_or(ErrorKind::MissingColumn(self.from_col.clone()))? 142 147 .to_string(); 143 148 self.value = (self.reduce)(self.value.clone(), &field)?; 144 149 Ok(()) ··· 162 167 where 163 168 V: Display + AddAssign + FromStr + Clone, 164 169 { 165 165 - fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> { 170 170 + fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), ErrorKind> { 166 171 let field = headers 167 172 .get_field(row, &self.from_col) 168 168 - .ok_or(Error::MissingColumn(self.from_col.clone()))? 173 173 + .ok_or(ErrorKind::MissingColumn(self.from_col.clone()))? 169 174 .to_string(); 170 175 let new: V = match field.parse() { 171 176 Ok(v) => v, 172 172 - Err(_) => return Err(Error::InvalidField(field)), 177 177 + Err(_) => return Err(ErrorKind::InvalidField(field)), 173 178 }; 174 179 self.value += new; 175 180 Ok(())