clippy/rename

Author: abhi-airspace-intelligence

etl-destinations/src/deltalake/table.rs renamed to etl-destinations/src/deltalake/config.rs

@@ -15,10 +15,10 @@ use tokio::sync::Mutex;
 use tracing::{info, trace};
 
 use crate::deltalake::TableRowEncoder;
+use crate::deltalake::config::DeltaTableConfig;
 use crate::deltalake::events::{materialize_events, materialize_events_append_only};
-use crate::deltalake::operations::{append_to_table, merge_to_table};
+use crate::deltalake::operations::append_to_table;
 use crate::deltalake::schema::postgres_to_delta_schema;
-use crate::deltalake::table::DeltaTableConfig;
 
 /// Configuration for Delta Lake destination
 #[derive(Debug, Clone)]
@@ -267,7 +267,7 @@ where
             );
 
             let config = self.config_for_table_name(&table_schema.name.name);
-            let mut table = table.lock().await;
+            let table = table.lock().await;
             todo!();
             // merge_to_table(
             //     table,

etl-destinations/src/deltalake/core.rs

@@ -49,9 +49,8 @@ pub fn materialize_events_append_only<'a>(
 
     let mut upsert_rows: Vec<&PgTableRow> = Vec::new();
     for (_, reg) in crdt_by_key.into_iter() {
-        match reg.val {
-            RowOp::Upsert(row) => upsert_rows.push(row),
-            _ => {}
+        if let RowOp::Upsert(row) = reg.val {
+            upsert_rows.push(row)
         }
     }
 
@@ -125,8 +124,8 @@ pub fn materialize_events<'a>(
 mod tests {
     use super::*;
     use etl::types::{
-        Cell as PgCell, ColumnSchema as PgColumnSchema, DeleteEvent, InsertEvent, PgLsn, TableId,
-        TableName, TableRow as PgTableRow, TableSchema as PgTableSchema, Type, UpdateEvent,
+        Cell as PgCell, ColumnSchema as PgColumnSchema, DeleteEvent, InsertEvent, TableId,
+        TableName, Type as PgType, UpdateEvent,
     };
 
     fn schema_single_pk(table_id: TableId) -> PgTableSchema {
@@ -136,14 +135,14 @@ mod tests {
             vec![
                 PgColumnSchema {
                     name: "id".to_string(),
-                    typ: Type::INT8,
+                    typ: PgType::INT8,
                     modifier: -1,
                     primary: true,
                     nullable: false,
                 },
                 PgColumnSchema {
                     name: "name".to_string(),
-                    typ: Type::TEXT,
+                    typ: PgType::TEXT,
                     modifier: -1,
                     primary: false,
                     nullable: true,
@@ -165,21 +164,21 @@ mod tests {
             vec![
                 PgColumnSchema {
                     name: "tenant_id".to_string(),
-                    typ: Type::INT4,
+                    typ: PgType::INT4,
                     modifier: -1,
                     primary: true,
                     nullable: false,
                 },
                 PgColumnSchema {
                     name: "user_id".to_string(),
-                    typ: Type::INT8,
+                    typ: PgType::INT8,
                     modifier: -1,
                     primary: true,
                     nullable: false,
                 },
                 PgColumnSchema {
                     name: "name".to_string(),
-                    typ: Type::TEXT,
+                    typ: PgType::TEXT,
                     modifier: -1,
                     primary: false,
                     nullable: true,

etl-destinations/src/deltalake/events.rs

@@ -1,9 +1,9 @@
+mod config;
 mod core;
 pub(crate) mod events;
 pub(crate) mod expr;
 mod operations;
 mod schema;
-mod table;
 
 pub use core::{DeltaDestinationConfig, DeltaLakeDestination};
 pub use schema::TableRowEncoder;

etl-destinations/src/deltalake/mod.rs

@@ -4,7 +4,7 @@ use deltalake::{
     writer::{DeltaWriter, RecordBatchWriter},
 };
 
-use crate::deltalake::table::DeltaTableConfig;
+use crate::deltalake::config::DeltaTableConfig;
 
 /// Appends a record batch to a Delta table
 pub async fn append_to_table(

etl-destinations/src/deltalake/operations/append.rs

@@ -4,8 +4,8 @@ use deltalake::{DeltaResult, DeltaTable, datafusion::prelude::Expr};
 use etl::types::{TableRow as PgTableRow, TableSchema as PgTableSchema};
 
 use crate::deltalake::TableRowEncoder;
+use crate::deltalake::config::DeltaTableConfig;
 use crate::deltalake::expr::qualify_primary_keys;
-use crate::deltalake::table::DeltaTableConfig;
 
 pub async fn merge_to_table(
     table: DeltaTable,

etl-destinations/src/deltalake/operations/merge.rs

@@ -3,4 +3,3 @@ mod delete;
 mod merge;
 
 pub use append::append_to_table;
-pub use merge::merge_to_table;

etl-destinations/src/deltalake/operations/mod.rs

@@ -1007,7 +1007,6 @@ mod tests {
     fn test_empty_table_rows() {
         let schema = create_test_schema();
         let result = TableRowEncoder::encode_table_rows(&schema, vec![]);
-        println!("result: {:?}", result);
         assert!(result.is_ok());
         assert!(result.unwrap().num_rows() == 0);
     }

etl-destinations/src/deltalake/schema.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use crate::migrations::migrate_state_store;
 use etl::destination::Destination;
 use etl::destination::memory::MemoryDestination;
@@ -79,8 +81,7 @@ pub async fn start_replicator_with_config(
                 DeltaDestinationConfig {
                     base_uri: base_uri.clone(),
                     storage_options: storage_options.clone(),
-                    partition_columns: partition_columns.clone(),
-                    optimize_after_commits: optimize_after_commits.map(|n| n.try_into().unwrap()),
+                    table_config: HashMap::new(),
                 },
             );