fmt

Author: Rickard Lundin

src/chunked/arrow_converter.rs

@@ -25,7 +25,7 @@
 // Last updated: 2024-05-15
 //
 
-use crate::chunked::trimmer::{ColumnTrimmer, trimmer_factory};
+use crate::chunked::trimmer::{trimmer_factory, ColumnTrimmer};
 use arrow::array::{ArrayRef, BooleanBuilder, Int32Builder, Int64Builder, StringBuilder};
 use arrow::record_batch::RecordBatch;
 use log::{debug, info};
@@ -43,25 +43,28 @@ use std::path::PathBuf;
 use std::str::from_utf8_unchecked;
 use std::sync::Arc;
 
-use super::{arrow_file_output, ColumnBuilder, Converter, FnFindLastLineBreak, FnLineBreakLen, Stats, trimmer};
+use super::{
+    arrow_file_output, trimmer, ColumnBuilder, Converter, FnFindLastLineBreak, FnLineBreakLen,
+    Stats,
+};
+use crate::chunked;
+use crate::chunked::threaded_file_output::{ipc_file_out, parquet_file_out};
 use crate::datatype::DataType;
 use crate::schema;
-use crate::{chunked};
 use arrow::datatypes::{Field, Schema, SchemaRef};
 use atomic_counter::{AtomicCounter, ConsistentCounter};
 use crossbeam::atomic::AtomicConsume;
 use libc::bsearch;
 use ordered_channel::Sender;
 use parquet::errors::{ParquetError, Result};
 use parquet::file::metadata::FileMetaData;
+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 std::thread::JoinHandle;
 use std::time::{Duration, Instant};
 use thread::spawn;
-use rayon::join;
-use crate::chunked::threaded_file_output::{ipc_file_out, parquet_file_out};
 //use ordered_channel::Sender;
 //use crossbeam::channel::{Receiver, Sender};
 
@@ -232,16 +235,14 @@ impl<'a> Converter<'a> for Slice2Arrow<'a> {
     fn setup(&mut self) -> JoinHandle<Result<Stats>> {
         let schema = self.masterbuilders.schema_factory();
         let _outfile = self.masterbuilders.outfile.clone();
-        
-        let mut pfo: Box<dyn arrow_file_output> = Box::new(ipc_file_out { sender: None } );
-        let (sender,joinhandler)=pfo.setup(schema,_outfile);
-            
+
+        let mut pfo: Box<dyn arrow_file_output> = Box::new(ipc_file_out { sender: None });
+        let (sender, joinhandler) = pfo.setup(schema, _outfile);
+
         self.masterbuilders.sender = Some(sender);
-       joinhandler
-        
+        joinhandler
     }
 
-
     fn shutdown(&mut self) {
         //        converter.shutdown();
         let schema = Schema::new(vec![Field::new(

src/chunked/mod.rs

@@ -25,26 +25,26 @@
 // Last updated: 2024-05-15
 //
 
-use std::fmt::Debug;
 use arrow::array::{ArrayRef, RecordBatch};
 use parquet::format;
+use std::fmt::Debug;
 
+use arrow::datatypes::SchemaRef;
+use ordered_channel::Sender;
 use std::fs;
 use std::path::PathBuf;
 use std::sync::mpsc::SyncSender;
 use std::thread::JoinHandle;
 use std::time::Duration;
-use arrow::datatypes::SchemaRef;
-use ordered_channel::Sender;
 
 use self::residual_slicer::SLICER_IN_CHUNK_SIZE;
 use parquet::errors::{ParquetError, Result};
 
 pub(crate) mod arrow_converter;
 pub(crate) mod residual_slicer;
 pub(crate) mod self_converter;
-mod trimmer;
 mod threaded_file_output;
+mod trimmer;
 
 pub(crate) struct ChunkAndResidue {
     pub(crate) chunk: Box<[u8; SLICER_IN_CHUNK_SIZE]>,
@@ -172,5 +172,9 @@ pub trait ColumnBuilder {
 }
 
 pub trait arrow_file_output {
-    fn setup(&mut self,schema:SchemaRef,outfile:PathBuf)-> (Sender<RecordBatch>, JoinHandle<Result<Stats>>);
+    fn setup(
+        &mut self,
+        schema: SchemaRef,
+        outfile: PathBuf,
+    ) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>);
 }

src/chunked/threaded_file_output.rs

@@ -25,7 +25,7 @@
 // Last updated: 2024-05-15
 //
 
-use crate::chunked::trimmer::{ColumnTrimmer, trimmer_factory};
+use crate::chunked::trimmer::{trimmer_factory, ColumnTrimmer};
 use arrow::array::{ArrayRef, BooleanBuilder, Int32Builder, Int64Builder, StringBuilder};
 use arrow::record_batch::RecordBatch;
 use log::{debug, info};
@@ -43,11 +43,16 @@ use std::path::PathBuf;
 use std::str::from_utf8_unchecked;
 use std::sync::Arc;
 
-use super::{arrow_file_output, ColumnBuilder, Converter, FnFindLastLineBreak, FnLineBreakLen, Stats, trimmer};
+use super::{
+    arrow_file_output, trimmer, ColumnBuilder, Converter, FnFindLastLineBreak, FnLineBreakLen,
+    Stats,
+};
+use crate::chunked;
 use crate::datatype::DataType;
 use crate::schema;
-use crate::{chunked};
 use arrow::datatypes::{Field, Schema, SchemaRef};
+use arrow_ipc::writer::IpcWriteOptions;
+use arrow_ipc::CompressionType;
 use atomic_counter::{AtomicCounter, ConsistentCounter};
 use crossbeam::atomic::AtomicConsume;
 use libc::bsearch;
@@ -60,31 +65,29 @@ use std::thread;
 use std::thread::JoinHandle;
 use std::time::{Duration, Instant};
 use thread::spawn;
-use arrow_ipc::CompressionType;
-use arrow_ipc::writer::IpcWriteOptions;
 use Compression::SNAPPY;
 
 pub(crate) struct parquet_file_out {
     pub(crate) sender: Option<Sender<RecordBatch>>,
 }
 
 impl arrow_file_output for parquet_file_out {
-    fn setup(&mut self, schema: SchemaRef, outfile: PathBuf) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
-
+    fn setup(
+        &mut self,
+        schema: SchemaRef,
+        outfile: PathBuf,
+    ) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
         let _out_file = fs::OpenOptions::new()
             .create(true)
             .append(true)
             .open(outfile)
             .expect("aaa");
 
-        let props = WriterProperties::builder()
-            .set_compression(SNAPPY)
-            .build();
+        let props = WriterProperties::builder().set_compression(SNAPPY).build();
 
         let mut writer: ArrowWriter<File> =
             ArrowWriter::try_new(_out_file, schema, Some(props.clone())).unwrap();
 
-
         let (sender, mut receiver) = bounded::<RecordBatch>(100);
 
         let t: JoinHandle<Result<Stats>> = thread::spawn(move || {
@@ -115,33 +118,39 @@ impl arrow_file_output for parquet_file_out {
                 builder_write_duration: Default::default(),
             })
         });
-        (sender,t)
+        (sender, t)
     }
-
 }
 
 pub struct ipc_file_out {
     pub(crate) sender: Option<Sender<RecordBatch>>,
-
 }
 
 impl arrow_file_output for ipc_file_out {
-    fn setup(&mut self, schema: SchemaRef, outfile: PathBuf) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
-
+    fn setup(
+        &mut self,
+        schema: SchemaRef,
+        outfile: PathBuf,
+    ) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
         let _out_file = fs::OpenOptions::new()
             .create(true)
             .append(true)
             .open(outfile)
             .expect("aaa");
 
+        let p = IpcWriteOptions::try_with_compression(
+            Default::default(),
+            Some(CompressionType::LZ4_FRAME),
+        );
+
+        let mut writer = arrow_ipc::writer::FileWriter::try_new_with_options(
+            _out_file,
+            &schema,
+            Default::default(),
+        )
+        .expect("TODO: panic message");
 
-        let p= IpcWriteOptions::try_with_compression(Default::default(),Some(CompressionType::LZ4_FRAME) );
-        
-        let mut writer=arrow_ipc::writer::FileWriter::try_new_with_options(_out_file, &schema, Default::default()).expect("TODO: panic message");
-        
-        let props = WriterProperties::builder()
-            .set_compression(SNAPPY)
-            .build();
+        let props = WriterProperties::builder().set_compression(SNAPPY).build();
 
         let (sender, mut receiver) = bounded::<RecordBatch>(1000);
 
@@ -173,6 +182,6 @@ impl arrow_file_output for ipc_file_out {
                 builder_write_duration: Default::default(),
             })
         });
-        (sender,t)
+        (sender, t)
     }
 }

src/chunked/trimmer.rs

@@ -27,8 +27,8 @@
 
 use std::cmp::min;
 
-use padder::{Alignment, Symbol};
 use crate::chunked::trimmer;
+use padder::{Alignment, Symbol};
 
 use crate::datatype::DataType;