fmt

Author: Rickard Lundin

src/chunked/arrow_converter.rs

@@ -50,6 +50,7 @@ use super::{
 };
 use crate::chunked;
 use crate::chunked::threaded_file_output::{ipc_file_out, output_factory, parquet_file_out};
+use crate::cli::Targets;
 use crate::datatype::DataType;
 use crate::schema;
 use arrow::datatypes::{Field, Schema, SchemaRef};
@@ -66,7 +67,6 @@ use std::thread;
 use std::thread::JoinHandle;
 use std::time::{Duration, Instant};
 use thread::spawn;
-use crate::cli::Targets;
 //use ordered_channel::Sender;
 //use crossbeam::channel::{Receiver, Sender};
 
@@ -78,7 +78,7 @@ pub(crate) struct Slice2Arrow<'a> {
     pub(crate) fn_line_break_len: FnLineBreakLen,
     pub(crate) masterbuilders: MasterBuilders,
     pub(crate) consistent_counter: ConsistentCounter,
-    pub(crate) target: Targets
+    pub(crate) target: Targets,
 }
 
 pub(crate) struct MasterBuilders {
@@ -236,10 +236,13 @@ impl<'a> Converter<'a> for Slice2Arrow<'a> {
     }
 
     fn setup(&mut self) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
-        output_factory(self.target.clone(), self.masterbuilders.schema_factory(), self.masterbuilders.outfile.clone())
+        output_factory(
+            self.target.clone(),
+            self.masterbuilders.schema_factory(),
+            self.masterbuilders.outfile.clone(),
+        )
     }
 
-
     fn shutdown(&mut self) {
         //        converter.shutdown();
         let schema = Schema::new(vec![Field::new(
@@ -251,7 +254,6 @@ 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);
-
     }
 }
 

src/chunked/mod.rs

@@ -161,7 +161,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)->(Sender<RecordBatch>, JoinHandle<Result<Stats>>);
+    fn setup(&mut self) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>);
     fn shutdown(&mut self);
 }
 

src/chunked/residual_slicer.rs

@@ -89,7 +89,7 @@ impl<'a> Slicer<'a> for ResidualSlicer<'a> {
             .build_global()
             .unwrap();
 
-        let threaded_writer=converter.setup();
+        let threaded_writer = converter.setup();
 
         loop {
             let cr = ir.next().unwrap();

src/chunked/self_converter.rs

@@ -30,6 +30,7 @@ use log::info;
 use rayon::iter::{IndexedParallelIterator, IntoParallelRefIterator, ParallelIterator};
 
 use arrow::array::RecordBatch;
+use ordered_channel::Sender;
 use parquet::errors::{ParquetError, Result};
 use parquet::file::metadata::FileMetaData;
 use parquet::format;
@@ -38,7 +39,6 @@ use std::io::Write;
 use std::sync::mpsc::SyncSender;
 use std::thread::JoinHandle;
 use std::time::Duration;
-use ordered_channel::Sender;
 
 use super::{Converter, FnFindLastLineBreak};
 
@@ -75,7 +75,6 @@ impl<'a> Converter<'a> for SampleSliceAggregator<'a> {
         todo!()
     }
 
-
     fn shutdown(&mut self) {
         todo!()
     }