Merge pull request #40 from firelink-data/feature/threaded_revolver

added timing output

Author: Ignalina

src/chunked/arrow_converter.rs

@@ -46,6 +46,7 @@ use crate::datatype::DataType;
 use crate::schema;
 use crate::trimmer::{trimmer_factory, ColumnTrimmer};
 use crate::{chunked, trimmer};
+use std::time::{Duration, Instant};
 
 pub(crate) struct Slice2Arrow<'a> {
     //    pub(crate) file_out: File,
@@ -163,9 +164,13 @@ impl<'a> Converter<'a> for Slice2Arrow<'a> {
         self.fn_line_break
     }
 
-    fn process(&mut self, slices: Vec<&'a [u8]>) -> (usize, usize) {
+    fn process(&mut self, slices: Vec<&'a [u8]>) -> (usize, usize, Duration, Duration) {
         let mut bytes_in: usize = 0;
         let mut bytes_out: usize = 0;
+        let mut parse_duration: Duration = Duration::new(0, 0);
+        let mut builder_write_duration: Duration = Duration::new(0, 0);
+
+        let start_parse = Instant::now();
 
         let arc_slices = Arc::new(&slices);
         self.masterbuilders
@@ -190,6 +195,7 @@ impl<'a> Converter<'a> for Slice2Arrow<'a> {
         for ii in slices.iter() {
             bytes_in += ii.len();
         }
+        parse_duration = start_parse.elapsed();
 
         for b in self.masterbuilders.builders.iter_mut() {
             let mut br: Vec<(&str, ArrayRef)> = vec![];
@@ -198,14 +204,17 @@ impl<'a> Converter<'a> for Slice2Arrow<'a> {
                 br.push(bb.finish());
             }
 
+            let start_builder_write = Instant::now();
             let batch = RecordBatch::try_from_iter(br).unwrap();
 
             self.writer.write(&batch).expect("Error Writing batch");
             bytes_out += self.writer.bytes_written();
-            debug!("Batch write: accumulated bytes_written {}", bytes_out);
+
+            builder_write_duration += start_builder_write.elapsed();
         }
+        debug!("Batch write: accumulated bytes_written {}", bytes_out);
 
-        (bytes_in, bytes_out)
+        (bytes_in, bytes_out, parse_duration, builder_write_duration)
     }
 
     fn finish(&mut self) -> parquet::errors::Result<format::FileMetaData> {

src/chunked/mod.rs

@@ -29,6 +29,7 @@ use arrow::array::ArrayRef;
 use parquet::format;
 
 use std::fs;
+use std::time::Duration;
 
 use self::residual_slicer::SLICER_IN_CHUNK_SIZE;
 
@@ -50,8 +51,11 @@ pub(crate) trait Slicer<'a> {
 pub(crate) struct Stats {
     pub(crate) bytes_in: usize,
     pub(crate) bytes_out: usize,
-
     pub(crate) num_rows: i64,
+
+    pub(crate) read_duration: Duration,
+    pub(crate) parse_duration: Duration,
+    pub(crate) builder_write_duration: Duration,
 }
 
 pub(crate) type FnLineBreakLen = fn() -> usize;
@@ -147,7 +151,7 @@ pub(crate) trait Converter<'a> {
     fn get_line_break_handler(&self) -> FnFindLastLineBreak<'a>;
 
     //    fn process(& mut self, slices: Vec< &'a[u8]>) -> usize;
-    fn process(&mut self, slices: Vec<&'a [u8]>) -> (usize, usize);
+    fn process(&mut self, slices: Vec<&'a [u8]>) -> (usize, usize, Duration, Duration);
     fn finish(&mut self) -> parquet::errors::Result<format::FileMetaData>;
     fn get_finish_bytes_written(&mut self) -> usize;
 }

src/chunked/residual_slicer.rs

@@ -31,10 +31,11 @@ use std::cmp;
 use std::fs;
 use std::fs::File;
 use std::io::{BufReader, Read};
+use std::time::{Duration, Instant};
 
 use super::{ChunkAndResidue, Converter, FnFindLastLineBreak, IterRevolver, Slicer, Stats};
 
-pub(crate) const SLICER_IN_CHUNK_SIZE: usize = 1024 * 1024;
+pub(crate) const SLICER_IN_CHUNK_SIZE: usize = 1024 * 2024;
 pub(crate) const SLICER_MAX_RESIDUE_SIZE: usize = SLICER_IN_CHUNK_SIZE;
 
 pub(crate) const IN_MAX_CHUNKS: usize = 2;
@@ -61,6 +62,9 @@ impl<'a> Slicer<'a> for ResidualSlicer<'a> {
 
         let mut bytes_in = 0;
         let mut bytes_out = 0;
+        let mut read_duration_tot: Duration = Duration::new(0, 0);
+        let mut parse_duration_tot: Duration = Duration::new(0, 0);
+        let mut builder_write_duration_tot: Duration = Duration::new(0, 0);
 
         let mut remaining_file_length = infile.metadata().unwrap().len() as usize;
 
@@ -90,6 +94,8 @@ impl<'a> Slicer<'a> for ResidualSlicer<'a> {
 
             let chunk_len_effective_read: usize;
 
+            let start_read = Instant::now();
+
             (residue_len, chunk_len_effective_read, slices) = read_chunk_and_slice(
                 self.fn_find_last_nl,
                 &mut residue,
@@ -100,10 +106,14 @@ impl<'a> Slicer<'a> for ResidualSlicer<'a> {
                 chunk_len_toread,
             );
 
+            read_duration_tot += start_read.elapsed();
+
             remaining_file_length -= chunk_len_effective_read;
-            let (bin, bout) = converter.process(slices);
+            let (bin, bout, parse_duration, builder_write_duration) = converter.process(slices);
             bytes_in += bin;
             bytes_out += bout;
+            parse_duration_tot += parse_duration;
+            builder_write_duration_tot += builder_write_duration;
 
             if remaining_file_length == 0 {
                 break;
@@ -115,18 +125,26 @@ impl<'a> Slicer<'a> for ResidualSlicer<'a> {
         if 0 != residue_len {
             slices = residual_to_slice(&residue, &mut cr.chunk, residue_len);
 
-            let (bin, bout) = converter.process(slices);
+            let (bin, bout, parse_duration, builder_write_duration) = converter.process(slices);
             bytes_in += bin;
             bytes_out += bout;
+            parse_duration_tot += parse_duration;
+            builder_write_duration_tot += builder_write_duration;
         }
 
-        info!("Bytes in= {} out= {}", bytes_in, bytes_out);
+        info!(
+            "Bytes in= {}\n out= {}\nparse duration= {:?}\n \n builder write_duration {:?}\n",
+            bytes_in, bytes_out, parse_duration_tot, builder_write_duration_tot
+        );
 
         match converter.finish() {
             Ok(x) => Result::Ok(Stats {
                 bytes_in,
                 bytes_out: converter.get_finish_bytes_written(),
                 num_rows: x.num_rows,
+                read_duration: read_duration_tot,
+                parse_duration: parse_duration_tot,
+                builder_write_duration: builder_write_duration_tot,
             }),
             Err(_x) => Result::Err("Could not produce Parquet"),
         }

src/chunked/self_converter.rs

@@ -30,6 +30,7 @@ use rayon::iter::{IndexedParallelIterator, IntoParallelRefIterator, ParallelIter
 
 use std::fs::File;
 use std::io::Write;
+use std::time::Duration;
 
 use super::{Converter, FnFindLastLineBreak};
 
@@ -46,8 +47,9 @@ impl<'a> Converter<'a> for SampleSliceAggregator<'a> {
         self.fn_line_break
     }
 
-    fn process(&mut self, slices: Vec<&'a [u8]>) -> (usize, usize) {
+    fn process(&mut self, slices: Vec<&'a [u8]>) -> (usize, usize, Duration, Duration) {
         let mut bytes_processed: usize = 0;
+        let duration = Duration::new(0, 0);
 
         slices
             .par_iter()
@@ -58,7 +60,7 @@ impl<'a> Converter<'a> for SampleSliceAggregator<'a> {
             let l = val.len();
             bytes_processed += l;
         }
-        (bytes_processed, 0)
+        (bytes_processed, 0, duration, duration)
     }
 
     fn finish(&mut self) -> Result<parquet::format::FileMetaData, parquet::errors::ParquetError> {

src/cli.rs

@@ -302,8 +302,8 @@ impl Cli {
                     slicer_instance.slice_and_convert(converter_instance, _in_file, n_threads)?;
 
                 info!(
-                    "Operation successful inbytes={} out bytes={} num of rows={}",
-                    stats.bytes_in, stats.bytes_out, stats.num_rows
+                    "Operation successful \ninbytes={} \nout bytes={} \nnum of rows={} \nread duration={:?} \nparse duration {:?}\nwrite duration={:?}",
+                    stats.bytes_in, stats.bytes_out, stats.num_rows,stats.read_duration,stats.parse_duration,stats.builder_write_duration
                 );
             }
             Commands::Mock {

src/trimmer.rs

@@ -27,7 +27,6 @@
 
 use std::cmp::min;
 
-use crate::converter::Converter;
 use padder::{Alignment, Symbol};
 
 use crate::datatype::DataType;

src/writer.rs

@@ -326,7 +326,7 @@ mod tests_writer {
             .build()
             .unwrap();
 
-        flfw.write(&vec![0u8; 64]).unwrap();
+        flfw.write(&[0u8; 64]).unwrap();
         fs::remove_file(path).unwrap();
     }