fmt

Author: Rickard Lundin

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "evolution"
-version = "1.0.0"
+version = "1.1.0"
 edition = "2021"
 description = "🦖 Evolve your fixed-length data files into Apache Parquet, fully parallelized!"
 authors = [
@@ -49,7 +49,7 @@ deltalake-core = "0.17.3"
 padder = { version = "1.2.0", features = ["serde"] }
 ordered-channel="1.1.0"
 atomic-counter = "1.0.1"
-crossbeam-channel = "0.5.12"
+crossbeam-channel = "0.5.13"
 tracing = "0.1.40"
 tokio = "1.37.0"
 [dev-dependencies]

src/chunked/arrow_converter.rs

@@ -45,14 +45,15 @@ use std::str::from_utf8_unchecked;
 use std::sync::Arc;
 
 use super::{
-    RecordBatchOutput, trimmer, ColumnBuilder, Converter, FnFindLastLineBreak, FnLineBreakLen,
+    trimmer, ColumnBuilder, Converter, FnFindLastLineBreak, FnLineBreakLen, RecordBatchOutput,
     Stats,
 };
 use crate::chunked;
-use crate::chunked::recordbatch_output::{IpcFileOut, output_factory, ParquetFileOut};
+use crate::chunked::recordbatch_output::{output_factory, IpcFileOut, ParquetFileOut};
 pub use crate::cli::Targets;
 use crate::datatype::DataType;
 use crate::schema;
+use crate::schema::FixedSchema;
 use arrow::datatypes::{Field, Schema, SchemaRef};
 use atomic_counter::{AtomicCounter, ConsistentCounter};
 use crossbeam::atomic::AtomicConsume;
@@ -64,12 +65,11 @@ use rayon::join;
 use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT};
 use std::sync::mpsc::{sync_channel, Receiver, RecvError, SyncSender};
 use std::thread;
-use tokio::task::JoinHandle;
 use std::time::{Duration, Instant};
 use thread::spawn;
 use tokio::runtime;
 use tokio::runtime::Runtime;
-use crate::schema::FixedSchema;
+use tokio::task::JoinHandle;
 //use ordered_channel::Sender;
 //use crossbeam::channel::{Receiver, Sender};
 
@@ -82,7 +82,7 @@ pub(crate) struct Slice2Arrow<'a> {
     pub(crate) masterbuilders: MasterBuilders,
     pub(crate) consistent_counter: ConsistentCounter,
     pub(crate) target: Targets,
-    pub(crate) fixed_schema: FixedSchema
+    pub(crate) fixed_schema: FixedSchema,
 }
 
 pub(crate) struct MasterBuilders {
@@ -245,13 +245,13 @@ impl<'a> Converter<'a> for Slice2Arrow<'a> {
             self.fixed_schema.clone(),
             self.masterbuilders.schema_factory(),
             self.masterbuilders.outfile.clone(),
-            rt
+            rt,
         );
         self.masterbuilders.sender = Some(o.0.clone());
         o
     }
-    
-    fn shutdown(&mut self,rt: &runtime::Runtime,jh:JoinHandle<Result<Stats>>) {
+
+    fn shutdown(&mut self, rt: &runtime::Runtime, jh: JoinHandle<Result<Stats>>) {
         let schema = Schema::new(vec![Field::new(
             "id",
             arrow::datatypes::DataType::Int32,
@@ -261,8 +261,7 @@ impl<'a> Converter<'a> for Slice2Arrow<'a> {
         let emptyrb = arrow::record_batch::RecordBatch::new_empty(Arc::new(schema));
         let c = self.consistent_counter.get();
         let _ = &self.masterbuilders.sender.clone().unwrap().send(c, emptyrb);
-        rt.spawn_blocking(|| async {jh.await.unwrap() });
-        
+        rt.spawn_blocking(|| async { jh.await.unwrap() });
     }
 }
 

src/chunked/mod.rs

@@ -34,19 +34,19 @@ use ordered_channel::Sender;
 use std::fs;
 use std::path::PathBuf;
 use std::sync::mpsc::SyncSender;
-use tokio::task::JoinHandle;
-use tokio::runtime::Runtime;
 use std::time::Duration;
+use tokio::runtime::Runtime;
+use tokio::task::JoinHandle;
 
 use self::residual_slicer::SLICER_IN_CHUNK_SIZE;
+use crate::schema::FixedSchema;
 use parquet::errors::{ParquetError, Result};
 use tokio::runtime;
-use crate::schema::FixedSchema;
 
 pub(crate) mod arrow_converter;
+pub(crate) mod recordbatch_output;
 pub(crate) mod residual_slicer;
 pub(crate) mod self_converter;
-pub(crate) mod recordbatch_output;
 mod trimmer;
 
 pub(crate) struct ChunkAndResidue {
@@ -64,7 +64,7 @@ pub(crate) struct Stats {
     pub(crate) bytes_in: usize,
     pub(crate) bytes_out: usize,
     pub(crate) num_rows: i64,
-
+    pub(crate) adds: i64,
     pub(crate) read_duration: Duration,
     pub(crate) parse_duration: Duration,
     pub(crate) builder_write_duration: Duration,
@@ -165,7 +165,7 @@ pub(crate) trait Converter<'a> {
     //    fn process(& mut self, slices: Vec< &'a[u8]>) -> usize;
     fn process(&mut self, slices: Vec<&'a [u8]>) -> (usize, usize, Duration, Duration);
     fn setup(&mut self, rt: &Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>);
-    fn shutdown(&mut self,rt: &Runtime,jh:JoinHandle<Result<Stats>>);
+    fn shutdown(&mut self, rt: &Runtime, jh: JoinHandle<Result<Stats>>);
 }
 
 pub trait ColumnBuilder {
@@ -180,6 +180,6 @@ pub trait RecordBatchOutput {
         schema: SchemaRef,
         fixed_schema: FixedSchema,
         outfile: PathBuf,
-        rt: &Runtime
+        rt: &Runtime,
     ) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>);
 }