clippy

Author: Rickard Lundin

src/chunked/recordbatch_output.rs

@@ -76,11 +76,11 @@ use deltalake::kernel::{PrimitiveType, StructField, StructType};
 use deltalake::parquet::basic::ZstdLevel;
 use deltalake::writer::{DeltaWriter, RecordBatchWriter};
 use deltalake::*;
+use deltalake_core::writer::WriteMode;
 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;
@@ -116,23 +116,25 @@ pub(crate) struct DeltaOut {
 
 impl DeltaOut {
     async fn deltasetup(schema: FixedSchema, out: PathBuf) -> Result<DeltaTable, DeltaTableError> {
-        
         info!("Using the location of: {:?}", out);
 
-        let table_path =out.to_str().unwrap();
+        let table_path = out.to_str().unwrap();
 
         let maybe_table = deltalake::open_table(&table_path).await;
-//        let hardoded_partition_cols = vec!["id".to_owned()];
+        //        let hardoded_partition_cols = vec!["id".to_owned()];
         let mut table = match maybe_table {
             Ok(table) => table,
             Err(DeltaTableError::NotATable(_)) => {
-                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)
+                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_partition_columns(hardoded_partition_cols)
+                    //                    with_partition_columns(hardoded_partition_cols)
                     .with_columns(fxschema_to_delta)
                     .await
                     .unwrap()
@@ -149,12 +151,16 @@ impl DeltaOut {
         outfile: PathBuf,
     ) -> (Result<Stats>) {
         let mut table = Self::deltasetup(fixed_schema, outfile).await.unwrap();
-// TODO @TEDDY AND @WILLE THESE MAX SETTINGS DO NOTHING ?
-        let writer_properties = WriterProperties::builder().set_max_row_group_size(300).set_write_batch_size(300).set_writer_version(WriterVersion::PARQUET_2_0)
+        // TODO @TEDDY AND @WILLE THESE MAX SETTINGS DO NOTHING ?
+        let writer_properties = WriterProperties::builder()
+            .set_max_row_group_size(300)
+            .set_write_batch_size(300)
+            .set_writer_version(WriterVersion::PARQUET_2_0)
             .set_compression(Compression::ZSTD(ZstdLevel::try_new(3).unwrap()))
             .build();
-//        table.metadata().unwrap().partition_columns
-        let mut writer = RecordBatchWriter::for_table(&table).expect("Failed to make RecordBatchWriter")
+        //        table.metadata().unwrap().partition_columns
+        let mut writer = RecordBatchWriter::for_table(&table)
+            .expect("Failed to make RecordBatchWriter")
             .with_writer_properties(writer_properties);
 
         let metadata = table
@@ -164,7 +170,7 @@ impl DeltaOut {
             &metadata.schema().expect("failed to get schema"),
         )
         .expect("Failed to convert to arrow schema");
-        let mut adds_tot:usize=0;
+        let mut adds_tot: usize = 0;
 
         'outer: loop {
             let mut message = receiver.recv();
@@ -174,35 +180,40 @@ impl DeltaOut {
                     if (rb.num_rows() == 0) {
                         break 'outer;
                     }
-                    let nrb=rb.with_schema(Arc::new(arrow_schema.clone())).unwrap();
+                    let nrb = rb.with_schema(Arc::new(arrow_schema.clone())).unwrap();
                     //                    let nrb=rb;
-                    info!("nrb rows={} nrb col len {} batchcount={} usize={}",nrb.num_rows(),nrb.columns().len(),writer.buffered_record_batch_count(),writer.buffer_len());
+                    info!(
+                        "nrb rows={} nrb col len {} batchcount={} usize={}",
+                        nrb.num_rows(),
+                        nrb.columns().len(),
+                        writer.buffered_record_batch_count(),
+                        writer.buffer_len()
+                    );
                     writer.write(nrb).await.expect("writing");
-                    if (writer.buffered_record_batch_count()>200) {
-                        writer
-                            .flush()
-                            .await
-                            .expect("Failed to flush write");
+                    if (writer.buffered_record_batch_count() > 200) {
+                        writer.flush().await.expect("Failed to flush write");
                     }
-//                    writer.write_with_mode(nrb,WriteMode::MergeSchema);
-
+                    //                    writer.write_with_mode(nrb,WriteMode::MergeSchema);
                 }
                 Err(e) => {
                     info!("got RecvError in channel , break to outer");
                     break 'outer;
                 }
             }
         }
-        info!("closing the writer for parquet {}",writer.buffered_record_batch_count());
+        info!(
+            "closing the writer for parquet {}",
+            writer.buffered_record_batch_count()
+        );
 
         let adds = writer
             .flush_and_commit(&mut table)
             .await
             .expect("Failed to flush write");
-        adds_tot+=adds as usize;
+        adds_tot += adds as usize;
 
         info!("{} adds written", adds_tot);
-        
+
         Ok(Stats {
             bytes_in: 0,
             bytes_out: 0,

src/schema.rs

@@ -31,11 +31,11 @@ use padder::{Alignment, Symbol};
 use rand::rngs::ThreadRng;
 use serde::{Deserialize, Serialize};
 
+use log::info;
 use std::fs;
 use std::io;
 use std::path::PathBuf;
 use std::sync::Arc;
-use log::info;
 
 use crate::builder::{
     BooleanColumnBuilder, ColumnBuilder, Float16ColumnBuilder, Float32ColumnBuilder,
@@ -312,7 +312,7 @@ impl FixedSchema {
                 )
             })
             .collect();
-    info!("as_arrow_schema= {:?}",fields);
+        info!("as_arrow_schema= {:?}", fields);
         Schema::new(fields)
     }