Continue refactor

Signed-off-by: Abhi Agarwal <abhi@airspace-intelligence.com>

Author: abhi-airspace-intelligence

etl-destinations/src/arrow/encoding.rs

@@ -7,15 +7,14 @@ use arrow::{
         TimestampMicrosecondBuilder,
     },
     datatypes::{
-        DataType, Date32Type, Field, FieldRef, Float32Type, Float64Type, Int16Type, Int32Type,
+        DataType, Date32Type, FieldRef, Float32Type, Float64Type, Int16Type, Int32Type,
         Int64Type, Schema, Time64MicrosecondType, TimeUnit, TimestampMicrosecondType, UInt32Type,
     },
     error::ArrowError,
 };
 use chrono::{NaiveDate, NaiveTime};
 use etl::types::{
     ArrayCell, Cell, DATE_FORMAT, TIME_FORMAT, TIMESTAMP_FORMAT, TableRow,
-    TableSchema as PgTableSchema, Type as PgType,
 };
 
 pub const UNIX_EPOCH: NaiveDate =
@@ -27,7 +26,7 @@ const UUID_BYTE_WIDTH: i32 = 16;
 
 /// Extract numeric precision from Postgres atttypmod
 /// Based on: https://stackoverflow.com/questions/72725508/how-to-calculate-numeric-precision-and-other-vals-from-atttypmod
-fn extract_numeric_precision(atttypmod: i32) -> u8 {
+pub fn extract_numeric_precision(atttypmod: i32) -> u8 {
     if atttypmod == -1 {
         // No limit specified, use maximum precision
         38
@@ -39,13 +38,13 @@ fn extract_numeric_precision(atttypmod: i32) -> u8 {
 
 /// Extract numeric scale from Postgres atttypmod
 /// Based on: https://stackoverflow.com/questions/72725508/how-to-calculate-numeric-precision-and-other-vals-from-atttypmod
-fn extract_numeric_scale(atttypmod: i32) -> i8 {
+pub fn extract_numeric_scale(atttypmod: i32) -> u8 {
     if atttypmod == -1 {
         // No limit specified, use reasonable default scale
         18
     } else {
         let scale = (atttypmod - 4) & 65535;
-        std::cmp::min(scale as i8, 38) // Cap at reasonable scale
+        std::cmp::min(scale as u8, 38) // Cap at reasonable scale
     }
 }
 

etl-destinations/src/deltalake/core.rs

@@ -15,7 +15,7 @@ use std::sync::Arc;
 use tokio::sync::Mutex;
 use tracing::{info, trace};
 
-use crate::deltalake::TableRowEncoder;
+use crate::arrow::rows_to_record_batch;
 use crate::deltalake::config::DeltaTableConfig;
 use crate::deltalake::events::{materialize_events, materialize_events_append_only};
 use crate::deltalake::maintenance::TableMaintenanceState;
@@ -355,32 +355,21 @@ where
 
         let table = self.table_handle(table_id).await?;
 
-        let table_schema = self
-            .store
-            .get_table_schema(table_id)
-            .await?
-            .ok_or_else(|| {
-                etl_error!(
-                    ErrorKind::MissingTableSchema,
-                    "Table schema not found",
-                    format!("Schema for table {} not found in store", table_id.0)
-                )
-            })?;
-
         let row_length = table_rows.len();
         trace!("Writing {} rows to Delta table", row_length);
+            
+        let config = self.config_for_table_name(&table_schema.name.name);
+        let mut table_guard = table.lock().await;
+        let schema = table_guard.snapshot().schema();
 
         let record_batch =
-            TableRowEncoder::encode_table_rows(&table_schema, table_rows).map_err(|e| {
+            rows_to_record_batch(table_rows.iter(), table_schema.clone()).map_err(|e| {
                 etl_error!(
                     ErrorKind::ConversionError,
                     "Failed to encode table rows",
                     format!("Error converting to Arrow: {}", e)
                 )
             })?;
-
-        let config = self.config_for_table_name(&table_schema.name.name);
-        let mut table_guard = table.lock().await;
         append_to_table(&mut table_guard, config.as_ref(), record_batch)
             .await
             .map_err(|e| {

etl-destinations/src/deltalake/expr.rs

@@ -1,7 +1,5 @@
 // Utilities related to constructing DataFusion expressions
 
-use crate::deltalake::schema::TableRowEncoder;
-use crate::deltalake::schema::cell_to_scalar_value_for_arrow;
 use deltalake::datafusion::common::Column;
 use deltalake::datafusion::prelude::{Expr, lit};
 use etl::error::EtlResult;

etl-destinations/src/deltalake/mod.rs

@@ -9,4 +9,3 @@ pub(crate) mod util;
 
 pub use config::DeltaTableConfig;
 pub use core::{DeltaDestinationConfig, DeltaLakeDestination};
-pub use schema::TableRowEncoder;

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

@@ -5,7 +5,6 @@ use deltalake::operations::merge::MergeBuilder;
 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;