added datafusion

Author: Rickard Lundin

Cargo.toml

@@ -44,8 +44,11 @@ libc = "0.2.154"
 arrow = "51.0.0"
 parquet = "51.0.0"
 arrow-ipc = "51.0.0"
-deltalake = "0.17.3"
+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"

src/chunked/recordbatch_output.rs

@@ -29,7 +29,7 @@ use arrow::array::{ArrayRef, BooleanBuilder, Int32Builder, Int64Builder, StringB
 use ordered_channel::bounded;
 use parquet::arrow::ArrowWriter;
 use parquet::basic::Compression;
-use parquet::file::properties::WriterProperties;
+use parquet::file::properties::{WriterProperties, WriterVersion};
 use parquet::format;
 use rayon::iter::IndexedParallelIterator;
 use rayon::prelude::*;
@@ -78,6 +78,7 @@ use std::time::{Duration, Instant};
 use thread::spawn;
 use tokio::task::JoinHandle;
 use Compression::SNAPPY;
+use deltalake_core::writer::WriteMode;
 //use deltalake::writer::test_utils::create_initialized_table;
 use tokio::runtime::Builder;
 use tokio::time::sleep;
@@ -122,12 +123,13 @@ impl DeltaOut {
         let mut table = match maybe_table {
             Ok(table) => table,
             Err(DeltaTableError::NotATable(_)) => {
-                info!("It doesn't look like our delta table has been created");
+                let fxschema_to_delta =schema.into_delta_columns();
+                info!("It doesn't look like our delta table has been created\n {:?}",fxschema_to_delta);
                 DeltaOps::try_from_uri(table_path)
                     .await
                     .unwrap()
                     .create()
-                    .with_columns(schema.into_delta_columns())
+                    .with_columns(fxschema_to_delta)
                     .await
                     .unwrap()
             }
@@ -144,32 +146,38 @@ impl DeltaOut {
     ) -> (Result<Stats>) {
         let mut table = Self::deltasetup(fixed_schema, outfile).await.unwrap();
 
-        let writer_properties = WriterProperties::builder()
+        let writer_properties = WriterProperties::builder().set_writer_version(WriterVersion::PARQUET_2_0)
             .set_compression(Compression::ZSTD(ZstdLevel::try_new(3).unwrap()))
             .build();
 
+
         let mut writer = RecordBatchWriter::for_table(&table)
             .expect("Failed to make RecordBatchWriter")
             .with_writer_properties(writer_properties);
 
+        
+
         let metadata = table
             .metadata()
             .expect("Failed to get metadata for the table");
         let arrow_schema = <deltalake::arrow::datatypes::Schema as TryFrom<&StructType>>::try_from(
             &metadata.schema().expect("failed to get schema"),
         )
         .expect("Failed to convert to arrow schema");
-        let arrow_schema_ref = Arc::new(arrow_schema);
 
         'outer: loop {
             let mut message = receiver.recv();
 
             match message {
                 Ok(rb) => {
-                    if (rb.num_rows() == 1) {
+                    if (rb.num_rows() == 0) {
                         break 'outer;
                     }
-                    writer.write(rb).await.expect("Error Writing batch");
+                    let nrb=rb.with_schema(Arc::new(arrow_schema.clone())).unwrap();
+                    info!("nrb rows={} ",nrb.num_rows());
+//                    writer.write(nrb).await.expect("Error Writing batch");
+                    writer.write_with_mode(nrb,WriteMode::MergeSchema);
+
                 }
                 Err(e) => {
                     info!("got RecvError in channel , break to outer");

src/cli.rs

@@ -295,7 +295,6 @@ impl Cli {
                             schema.to_path_buf(),
                             n_threads as i16,
                         );
-                        let sc = master_builders.schema_factory();
 
                         let s2a: Box<Slice2Arrow> = Box::new(Slice2Arrow {
                             fn_line_break: find_last_nl,

src/schema.rs

@@ -35,6 +35,7 @@ use std::fs;
 use std::io;
 use std::path::PathBuf;
 use std::sync::Arc;
+use log::info;
 
 use crate::builder::{
     BooleanColumnBuilder, ColumnBuilder, Float16ColumnBuilder, Float32ColumnBuilder,
@@ -311,7 +312,7 @@ impl FixedSchema {
                 )
             })
             .collect();
-
+    info!("as_arrow_schema= {:?}",fields);
         Schema::new(fields)
     }