async writer working but no speed difference

Author: Rickard Lundin

src/chunked/arrow_converter.rs

@@ -252,7 +252,6 @@ impl<'a> Converter<'a> for Slice2Arrow<'a> {
     }
 
     fn shutdown(&mut self,rt: &runtime::Runtime,jh:JoinHandle<Result<Stats>>) {
-        //        converter.shutdown();
         let schema = Schema::new(vec![Field::new(
             "id",
             arrow::datatypes::DataType::Int32,
@@ -262,12 +261,8 @@ 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 { });
+        rt.spawn_blocking(|| async {jh.await.unwrap() });
 
-        //     jh.spawn_blocking(async move {threaded_writer.1.await;});
-//        threaded_writer.1.await.expect("The task being joined has panicked");;
-
-
     }
 }
 

src/chunked/recordbatch_output.rs

@@ -29,15 +29,14 @@ use std::env::VarError;
 use crate::chunked::trimmer::{trimmer_factory, ColumnTrimmer};
 use arrow::array::{ArrayRef, BooleanBuilder, Int32Builder, Int64Builder, StringBuilder};
 use ordered_channel::bounded;
-use parquet::arrow::ArrowWriter;
+use parquet::arrow::{ArrowWriter, AsyncArrowWriter};
 use parquet::basic::Compression;
 use parquet::file::properties::WriterProperties;
 use parquet::format;
 use rayon::iter::IndexedParallelIterator;
 use rayon::prelude::*;
 
-use std::fs;
-use std::fs::File;
+use tokio::fs::File;
 use std::path::{Path, PathBuf};
 use std::str::from_utf8_unchecked;
 use std::sync::Arc;
@@ -208,24 +207,21 @@ pub(crate) struct ParquetFileOut {
 }
 impl ParquetFileOut {
     pub(crate) async fn myParquet(mut receiver:  Receiver<RecordBatch>, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf)->(Result<Stats>) {
-//        Self::deltasetup(fixed_schema).await.unwrap();
-        let _out_file = fs::OpenOptions::new()
+        let _out_file = File::options()
             .create(true)
             .append(true)
-            .open(outfile)
-            .expect("aaa");
-
+            .open(outfile).await.unwrap();
+        
         let props = WriterProperties::builder().set_compression(SNAPPY).build();
-
-        let mut writer: ArrowWriter<File> =
-            ArrowWriter::try_new(_out_file, schema, Some(props.clone())).unwrap();
+        let mut writer: AsyncArrowWriter<File> =
+            AsyncArrowWriter::try_new(_out_file, schema, Some(props.clone())).unwrap();
 
             'outer: loop {
                 let mut message = receiver.recv();
 
                 match message {
                     Ok(rb) => {
-                        writer.write(&rb).expect("Error Writing batch");
+                        writer.write(&rb).await.expect("Error Writing batch");
                         if (rb.num_rows() == 0) {
                             break 'outer;
                         }
@@ -237,7 +233,7 @@ impl ParquetFileOut {
                 }
             }
             info!("closing the writer for parquet");
-            writer.finish();
+            writer.close();
             Ok(Stats {
                 bytes_in: 0,
                 bytes_out: 0,