Implement proper support for datatypes

Author: abhi-airspace-intelligence

Cargo.toml

@@ -35,6 +35,7 @@ clap = { version = "4.5.42", default-features = false }
 config = { version = "0.14", default-features = false }
 const-oid = { version = "0.9.6", default-features = false }
 constant_time_eq = { version = "0.4.2" }
+delta_kernel = { version = "0.13.0", default-features = false, features = ["arrow-conversion"] }
 deltalake = { version = "0.27.0", default-features = false, features = ["rustls", "datafusion", "s3"] }
 fail = { version = "0.5.1", default-features = false }
 futures = { version = "0.3.31", default-features = false }

README.md

@@ -104,4 +104,3 @@ Apache‑2.0. See `LICENSE` for details.
 <p align="center">
   Made with ❤️ by the <a href="https://supabase.com">Supabase</a> team
 </p>
-

etl-destinations/Cargo.toml

@@ -13,13 +13,14 @@ bigquery = [
     "dep:tokio",
 ]
 iceberg = ["dep:iceberg", "dep:iceberg-catalog-rest"]
-deltalake = ["dep:deltalake", "dep:tokio", "dep:tracing"]
+deltalake = ["dep:deltalake", "dep:delta_kernel", "dep:tokio", "dep:tracing"]
 
 [dependencies]
 etl = { workspace = true }
 etl-postgres = { workspace = true }
 chrono = { workspace = true }
 
+delta_kernel = { workspace = true, optional = true, features = ["arrow-conversion"] }
 deltalake = { workspace = true, optional = true, features = ["rustls", "datafusion"] }
 futures = { workspace = true, optional = true }
 gcp-bigquery-client = { workspace = true, optional = true, features = [

etl-destinations/src/delta/client.rs

@@ -69,7 +69,10 @@ impl DeltaLakeClient {
 
     /// Open a Delta table at `table_uri`.
     pub async fn open_table(&self, table_uri: &str) -> DeltaResult<Arc<DeltaTable>> {
-        let table = open_table(table_uri).await?;
+        let table = self
+            .get_table_with_storage_options(table_uri)?
+            .load()
+            .await?;
         Ok(Arc::new(table))
     }
 

etl-destinations/src/delta/encoding.rs

@@ -1,8 +1,7 @@
 mod client;
 mod core;
-mod encoding;
 mod schema;
 
 pub use client::DeltaLakeClient;
 pub use core::{DeltaDestinationConfig, DeltaLakeDestination};
-pub use encoding::TableRowEncoder;
+pub use schema::TableRowEncoder;