Merge pull request #46 from firelink-data/feature/lake_outputs

Feature/lake outputs

Author: Ignalina

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 = [
@@ -44,11 +44,18 @@ libc = "0.2.154"
 arrow = "51.0.0"
 parquet = "51.0.0"
 arrow-ipc = "51.0.0"
+deltalake =  { version="0.17.3", features = ["datafusion"]}
+datafusion = "38.0.0"
+
+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"
+futures = "0.3.30"
 [dev-dependencies]
 glob = "0.3.1"
 

README.md

@@ -19,14 +19,14 @@
 [![CD](https://github.com/firelink-data/evolution/actions/workflows/cd.yml/badge.svg)](https://github.com/firelink-data/evolution/actions/workflows/cd.yml)
 [![Tests](https://github.com/firelink-data/evolution/actions/workflows/tests.yml/badge.svg)](https://github.com/firelink-data/evolution/actions/workflows/tests.yml)
 
-🦖 *Evolve your fixed-length data files into Apache Parquet, fully parallelized!*
+🦖 *Evolve your fixed-length data files using Apache Arrow , fully parallelized!*
 
 </div>
 
 
 ## 🔎 Overview
 
-This repository hosts the **evolution** program which both allows you to convert existing fixed-length files into other data formats, but also allows you to create large amounts of mocked data blazingly fast. The program supports full parallelism and utilizes SIMD techniques, when possible, for highly efficient parsing of data. 
+This repository hosts the **evolution** program which both allows you to convert existing fixed-length files into dataformats like parquet or lake targets like Delta , but also allows you to create large amounts of mocked data blazingly fast. The program supports full parallelism and utilizes SIMD techniques, when possible, for highly efficient parsing of data. 
 
 To get started, follow the installation, schema setup, and example usage sections below in this README. Happy hacking! 👋🥳
 

src/chunked/arrow_converter.rs

@@ -45,14 +45,15 @@ use std::str::from_utf8_unchecked;
 use std::sync::Arc;
 
 use super::{
-    arrow_file_output, trimmer, ColumnBuilder, Converter, FnFindLastLineBreak, FnLineBreakLen,
+    trimmer, ColumnBuilder, Converter, FnFindLastLineBreak, FnLineBreakLen, RecordBatchOutput,
     Stats,
 };
 use crate::chunked;
-use crate::chunked::threaded_file_output::{ipc_file_out, output_factory, parquet_file_out};
+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,9 +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 std::thread::JoinHandle;
 use std::time::{Duration, Instant};
 use thread::spawn;
+use tokio::runtime;
+use tokio::runtime::Runtime;
+use tokio::task::JoinHandle;
 //use ordered_channel::Sender;
 //use crossbeam::channel::{Receiver, Sender};
 
@@ -79,6 +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) struct MasterBuilders {
@@ -190,7 +194,7 @@ impl<'a> Converter<'a> for Slice2Arrow<'a> {
 
         let start_parse = Instant::now();
         let offset: usize = self.consistent_counter.get();
-
+        let mut local_consistent_counter: ConsistentCounter = ConsistentCounter::new(0);
         let arc_slices = Arc::new(&slices);
         self.masterbuilders
             .builders
@@ -213,40 +217,39 @@ impl<'a> Converter<'a> for Slice2Arrow<'a> {
                             br.push(bb.finish());
                         }
                         let record_batch = RecordBatch::try_from_iter(br).unwrap();
-
                         let _ = self
                             .masterbuilders
                             .sender
                             .clone()
                             .unwrap()
                             .send(offset + i, record_batch);
+                        let offset: usize = local_consistent_counter.add(1);
                     }
                 }
             });
 
         for ii in slices.iter() {
             bytes_in += ii.len();
         }
-        let offset: usize = self
-            .consistent_counter
-            .add(self.masterbuilders.builders.len());
+        let offset: usize = self.consistent_counter.add(local_consistent_counter.get());
 
         parse_duration = start_parse.elapsed();
         (bytes_in, bytes_out, parse_duration, builder_write_duration)
     }
 
-    fn setup(&mut self) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
+    fn setup(&mut self, rt: &runtime::Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
         let o = output_factory(
             self.target.clone(),
+            self.fixed_schema.clone(),
             self.masterbuilders.schema_factory(),
             self.masterbuilders.outfile.clone(),
+            rt,
         );
         self.masterbuilders.sender = Some(o.0.clone());
         o
     }
 
-    fn shutdown(&mut self) {
-        //        converter.shutdown();
+    fn shutdown(&mut self, rt: &runtime::Runtime, jh: JoinHandle<Result<Stats>>) {
         let schema = Schema::new(vec![Field::new(
             "id",
             arrow::datatypes::DataType::Int32,
@@ -256,6 +259,10 @@ 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 {
+            info!("...spawned blockiiiiing");
+            jh.await.unwrap()
+        });
     }
 }
 

src/chunked/mod.rs

@@ -34,16 +34,19 @@ 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 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;
 
 pub(crate) mod arrow_converter;
+pub(crate) mod recordbatch_output;
 pub(crate) mod residual_slicer;
 pub(crate) mod self_converter;
-pub(crate) mod threaded_file_output;
 mod trimmer;
 
 pub(crate) struct ChunkAndResidue {
@@ -61,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,
@@ -161,8 +164,8 @@ 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 shutdown(&mut self);
+    fn setup(&mut self, rt: &Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>);
+    fn shutdown(&mut self, rt: &Runtime, jh: JoinHandle<Result<Stats>>);
 }
 
 pub trait ColumnBuilder {
@@ -171,10 +174,12 @@ pub trait ColumnBuilder {
     //    fn name(&  self) -> &String;
 }
 
-pub trait arrow_file_output {
+pub trait RecordBatchOutput {
     fn setup(
         &mut self,
         schema: SchemaRef,
+        fixed_schema: FixedSchema,
         outfile: PathBuf,
+        rt: &Runtime,
     ) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>);
 }