Transform almost there · kasper.space/csv-pipeline@950d1ab

+43 -5

src/lib.rs

··· 7 7 pub use headers::Headers; 8 8 pub use pipeline::{Pipeline, PipelineIter}; 9 9 pub use target::{PathTarget, StderrTarget, StdoutTarget, StringTarget, Target}; 10 - pub use transform::{Transform, Transformer}; 10 + pub use transform::Transformer; 11 11 12 12 pub type Row = csv::StringRecord; 13 13 pub type RowResult = Result<Row, Error>; ··· 33 33 } 34 34 35 35 #[test] 36 + fn transform() { 37 + let source = "\ 38 + Person,Score\n\ 39 + A,1\n\ 40 + A,8\n\ 41 + B,3\n\ 42 + B,4\n"; 43 + let reader = csv::ReaderBuilder::new() 44 + .flexible(true) 45 + .from_reader(source.as_bytes()); 46 + let csv_str = Pipeline::from_reader(reader) 47 + .unwrap() 48 + .map(|_headers, row| { 49 + println!("{row:?}"); 50 + Ok(row) 51 + }) 52 + .transform_into([ 53 + Transformer::new("Person").keep_unique(), 54 + Transformer::new("Score").reduce( 55 + |accumulator, current| { 56 + let score: u64 = current.parse().unwrap(); 57 + Ok(accumulator + score) 58 + }, 59 + 0, 60 + ), 61 + ]) 62 + .map(|_headers, row| { 63 + println!("{row:?}"); 64 + Ok(row) 65 + }) 66 + .collect_into_string() 67 + .unwrap(); 68 + 69 + assert_eq!( 70 + csv_str, 71 + "Person,Score\n\ 72 + A,9\n\ 73 + B,7\n" 74 + ); 75 + } 76 + 77 + #[test] 36 78 fn test_pipeline() { 37 79 let csv_str = Pipeline::from_path("test/Countries.csv") 38 80 .unwrap() ··· 42 84 _ => Ok("Unknown"), 43 85 } 44 86 }) 45 - .transform_into([ 46 - Transformer::new("ID").keep_unique(), 47 - Transformer::new("Country").sum(0), 48 - ]) 49 87 .rename_col("Country", "COUNTRY") 50 88 .map_col("COUNTRY", |id_str| Ok(id_str.to_uppercase())) 51 89 .collect_into_string()

+30 -4

src/pipeline.rs

··· 1 1 use super::headers::Headers; 2 - use crate::pipeline_iterators::{AddCol, Flush, MapCol, MapRow}; 2 + use crate::pipeline_iterators::{AddCol, Flush, MapCol, MapRow, TransformInto}; 3 3 use crate::target::Target; 4 4 use crate::transform::Transform; 5 5 use crate::{Error, Row, RowResult, StringTarget}; 6 6 use csv::{Reader, ReaderBuilder, StringRecordsIntoIter}; 7 7 use std::borrow::BorrowMut; 8 - use std::fs::File; 8 + use std::collections::HashMap; 9 9 use std::io; 10 10 use std::path::Path; 11 11 ··· 15 15 } 16 16 17 17 impl<'a> Pipeline<'a> { 18 - pub fn from_reader(mut reader: Reader<File>) -> Result<Self, Error> { 18 + pub fn from_reader<R: io::Read + 'a>(mut reader: Reader<R>) -> Result<Self, Error> { 19 19 let headers_row = reader.headers().unwrap().clone(); 20 20 let row_iterator = RowIter::from_records(reader.into_records()); 21 21 Ok(Pipeline { ··· 187 187 self 188 188 } 189 189 190 - pub fn transform_into<T>(self, transformers: T) -> Self 190 + /// Group and reduce rows 191 + /// 192 + /// ## Example 193 + /// 194 + /// ``` 195 + /// use csv_pipeline::{Pipeline, StringTarget}; 196 + /// 197 + /// let csv = Pipeline::from_path("test/AB.csv") 198 + /// .unwrap() 199 + /// .rename_cols(|i, name| { 200 + /// match name { 201 + /// "A" => "X", 202 + /// name => name, 203 + /// } 204 + /// }) 205 + /// .collect_into_string() 206 + /// .unwrap(); 207 + /// 208 + /// assert_eq!(csv, "X,B\n1,2\n"); 209 + /// ``` 210 + pub fn transform_into<T>(mut self, transformers: T) -> Self 191 211 where 192 212 T: IntoIterator<Item = Box<dyn Transform>>, 193 213 { 214 + self.iterator = Box::new(TransformInto { 215 + iterator: self.iterator, 216 + groups: HashMap::new(), 217 + transformers: transformers.into_iter().collect(), 218 + headers: self.headers.clone(), 219 + }); 194 220 self 195 221 } 196 222

+36 -17

src/pipeline_iterators.rs

··· 1 - use std::collections::hash_map::{DefaultHasher, Entry}; 2 - use std::collections::HashMap; 3 - use std::hash::Hasher; 4 - 5 1 use super::headers::Headers; 6 2 use crate::target::Target; 7 - use crate::transform::compute_hash; 8 - use crate::{Error, Row, RowResult, Transform}; 3 + use crate::transform::{compute_hash, Reduce, Transform}; 4 + use crate::{Error, Row, RowResult}; 5 + use std::collections::hash_map::Entry; 6 + use std::collections::HashMap; 9 7 10 8 pub struct AddCol<'a, I, F: FnMut(&Headers, &Row) -> Result<&'a str, Error>> { 11 9 pub iterator: I, ··· 94 92 } 95 93 } 96 94 97 - pub struct TransformInto<I> { 95 + pub struct TransformInto<'a, I> { 98 96 pub iterator: I, 99 - pub groups: HashMap<u64, Vec<Box<dyn Transform>>>, 100 - pub transformers: Vec<Box<dyn Transform>>, 97 + pub groups: HashMap<u64, Vec<Box<dyn Reduce + 'a>>>, 98 + pub transformers: Vec<Box<dyn Transform + 'a>>, 101 99 pub headers: Headers, 102 100 } 103 - impl<I> Iterator for TransformInto<I> 101 + impl<'a, I> TransformInto<'a, I> { 102 + pub fn ensure_group(&'a mut self, hash: u64) { 103 + if let Entry::Vacant(entry) = self.groups.entry(hash) { 104 + let reducers = self 105 + .transformers 106 + .iter_mut() 107 + .map(|transformer| transformer.new_reducer()) 108 + .collect(); 109 + entry.insert(reducers); 110 + }; 111 + } 112 + } 113 + impl<'a, I> Iterator for TransformInto<'a, I> 104 114 where 105 115 I: Iterator<Item = RowResult>, 106 116 { 107 117 type Item = RowResult; 108 118 109 119 fn next(&mut self) -> Option<Self::Item> { 110 - // First run iterator into hashmap, then return rows from the hashmap 111 120 // If any error rows are found, they are returned first 112 121 if let Some(row_result) = self.iterator.next() { 122 + // First run iterator into hashmap 113 123 let row = match row_result { 114 124 Ok(row) => row, 115 125 Err(e) => return Some(Err(e)), ··· 118 128 Ok(hash) => hash, 119 129 Err(e) => return Some(Err(e)), 120 130 }; 121 - let group_row = self.groups.entry(hash).or_default(); 122 - for transformer in self.transformers { 123 - let result = transformer.add_row(&self.headers, &row); 131 + self.ensure_group(hash); 132 + let group_row = self.groups.get_mut(&hash).unwrap(); 133 + for reducer in group_row { 134 + let result = reducer.add_row(&self.headers, &row); 124 135 if let Err(e) = result { 125 136 return Some(Err(e)); 126 137 } 127 138 } 128 - group_row.push_field(row); 139 + return self.next(); 140 + } 141 + // Finally, return rows from the hashmap 142 + if let Some(key) = self.groups.keys().next().copied() { 143 + let reducers = self.groups.remove(&key).unwrap(); 144 + let fields: Vec<_> = reducers.iter().map(|reducer| reducer.value()).collect(); 145 + let row = Row::from(fields); 146 + println!("x {row:?}"); 147 + println!("- {}", reducers.len()); 148 + Some(Ok(row)) 129 149 } else { 150 + None 130 151 } 131 - 132 - x 133 152 } 134 153 } 135 154

+103 -55

src/transform.rs

··· 1 1 use crate::{Error, Headers, Row}; 2 + use core::fmt::Display; 2 3 use std::collections::hash_map::DefaultHasher; 3 4 use std::hash::{Hash, Hasher}; 4 - use std::ops::AddAssign; 5 - use std::str::FromStr; 6 5 7 6 pub trait Transform { 8 7 /// Add the row to the hasher to group this row separately from others 9 - fn hash(&self, hasher: &mut DefaultHasher, headers: &Headers, row: &Row) -> Result<(), Error> { 8 + fn hash( 9 + &self, 10 + _hasher: &mut DefaultHasher, 11 + _headers: &Headers, 12 + _row: &Row, 13 + ) -> Result<(), Error> { 10 14 Ok(()) 11 15 } 12 16 17 + fn new_reducer<'a>(&'a mut self) -> Box<dyn Reduce + 'a>; 18 + 19 + /// Get the resulting column name 20 + fn name(&self) -> String; 21 + } 22 + 23 + pub trait Reduce { 13 24 /// Combine the row with the value 14 25 fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error>; 15 26 16 27 /// Turn the current value to a string 17 28 fn value(&self) -> String; 18 - 19 - /// Get the resulting column name 20 - fn name(&self) -> String; 21 29 } 22 30 23 31 pub struct Transformer { 24 - col_name: String, 32 + name: String, 25 33 from_col: String, 26 34 } 27 - impl Transformer { 35 + impl<'a> Transformer { 28 36 pub fn new(col_name: &str) -> Self { 29 37 Self { 30 - col_name: col_name.to_string(), 38 + name: col_name.to_string(), 31 39 from_col: col_name.to_string(), 32 40 } 33 41 } 34 42 pub fn keep_unique(self) -> Box<dyn Transform> { 35 43 Box::new(KeepUnique { 36 - col_name: self.col_name, 44 + name: self.name, 37 45 from_col: self.from_col, 38 46 value: "".to_string(), 39 47 }) 40 48 } 41 - pub fn sum<'a, N: AddAssign + FromStr + ToString + 'a>( 42 - self, 43 - init: N, 44 - ) -> Box<dyn Transform + 'a> { 45 - Box::new(Sum { 46 - col_name: self.col_name, 49 + } 50 + impl Transformer { 51 + pub fn reduce<'a, R, V>(self, reduce: R, init: V) -> Box<dyn Transform + 'a> 52 + where 53 + R: FnMut(V, &str) -> Result<V, Error> + 'a, 54 + V: Display + Clone + 'a, 55 + { 56 + Box::new(ClosureTransformer { 57 + name: self.name, 47 58 from_col: self.from_col, 59 + reduce, 48 60 value: init, 49 61 }) 50 62 } 51 63 } 52 64 53 - pub fn compute_hash( 54 - transformers: &[Box<dyn Transform>], 55 - headers: &Headers, 56 - row: &Row, 57 - ) -> Result<u64, Error> { 58 - let mut hasher = DefaultHasher::new(); 59 - for transformer in transformers { 60 - let result = transformer.hash(&mut hasher, &headers, &row); 61 - if let Err(e) = result { 62 - return Err(e); 63 - } 65 + struct ClosureTransformer<F, V> { 66 + name: String, 67 + from_col: String, 68 + reduce: F, 69 + value: V, 70 + } 71 + impl<F, V> Transform for ClosureTransformer<F, V> 72 + where 73 + F: FnMut(V, &str) -> Result<V, Error>, 74 + V: Display + Clone, 75 + { 76 + fn new_reducer<'a>(&'a mut self) -> Box<dyn Reduce + 'a> { 77 + Box::new(Closure { 78 + from_col: self.from_col.clone(), 79 + reduce: &mut self.reduce, 80 + value: self.value.clone(), 81 + }) 82 + } 83 + 84 + fn name(&self) -> String { 85 + self.name.clone() 64 86 } 65 - Ok(hasher.finish()) 66 87 } 67 88 68 - pub struct KeepUnique { 69 - col_name: String, 89 + struct Closure<'a, F, V> { 90 + from_col: String, 91 + reduce: &'a mut F, 92 + value: V, 93 + } 94 + impl<F, V> Reduce for Closure<'_, F, V> 95 + where 96 + F: FnMut(V, &str) -> Result<V, Error>, 97 + V: Display + Clone, 98 + { 99 + fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> { 100 + let field = headers 101 + .get_field(row, &self.from_col) 102 + .ok_or(Error::MissingColumn(self.from_col.clone()))? 103 + .to_string(); 104 + (self.reduce)(self.value.clone(), &field)?; 105 + todo!() 106 + } 107 + 108 + fn value(&self) -> String { 109 + self.value.to_string() 110 + } 111 + } 112 + 113 + struct KeepUnique { 114 + name: String, 70 115 from_col: String, 71 116 value: String, 72 117 } ··· 74 119 fn hash(&self, hasher: &mut DefaultHasher, headers: &Headers, row: &Row) -> Result<(), Error> { 75 120 let field = headers 76 121 .get_field(row, &self.from_col) 77 - .ok_or(Error::MissingColumn(self.value.clone()))?; 122 + .ok_or(Error::MissingColumn(self.from_col.clone()))?; 78 123 field.hash(hasher); 79 124 Ok(()) 80 125 } 126 + fn new_reducer<'a>(&'a mut self) -> Box<dyn Reduce + 'a> { 127 + Box::new(Self { 128 + name: self.name.clone(), 129 + from_col: self.name.clone(), 130 + value: "".to_string(), 131 + }) 132 + } 133 + 134 + fn name(&self) -> String { 135 + self.name.clone() 136 + } 137 + } 138 + impl Reduce for KeepUnique { 81 139 fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> { 82 140 self.value = headers 83 141 .get_field(row, &self.from_col) 84 - .ok_or(Error::MissingColumn(self.value.clone()))? 142 + .ok_or(Error::MissingColumn(self.from_col.clone()))? 85 143 .to_string(); 86 144 Ok(()) 87 145 } 146 + 88 147 fn value(&self) -> String { 89 148 self.value.clone() 90 149 } 91 - fn name(&self) -> String { 92 - self.col_name.clone() 93 - } 94 150 } 95 151 96 - pub struct Sum<N> { 97 - col_name: String, 98 - from_col: String, 99 - value: N, 100 - } 101 - impl<N: AddAssign + FromStr + ToString> Transform for Sum<N> { 102 - fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> { 103 - let field = headers.get_field(row, &self.col_name).unwrap(); 104 - let number = match N::from_str(field) { 105 - Ok(number) => number, 106 - Err(e) => return Err(Error::InvalidField(field.to_string())), 107 - }; 108 - self.value += number; 109 - Ok(()) 152 + pub fn compute_hash<'a>( 153 + transformers: &Vec<Box<dyn Transform + 'a>>, 154 + headers: &Headers, 155 + row: &Row, 156 + ) -> Result<u64, Error> { 157 + let mut hasher = DefaultHasher::new(); 158 + for transformer in transformers { 159 + let result = transformer.hash(&mut hasher, &headers, &row); 160 + if let Err(e) = result { 161 + return Err(e); 162 + } 110 163 } 111 - fn value(&self) -> String { 112 - self.value.to_string() 113 - } 114 - fn name(&self) -> String { 115 - self.col_name.clone() 116 - } 164 + Ok(hasher.finish()) 117 165 }

Configure Feed

Configure Feed