Get stuff working

Author: abhi-airspace-intelligence

Cargo.toml

@@ -35,7 +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" }
-deltalake = { version = "0.27.0", default-features = false, features = ["rustls"] }
+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 }
 gcp-bigquery-client = { git = "https://github.com/lquerel/gcp-bigquery-client.git", default-features = false, rev = "391a162642bd0af6d55e2310ef46adf96f21228c" }

etl-api/src/configs/destination.rs

@@ -14,6 +14,7 @@ const DEFAULT_MAX_CONCURRENT_STREAMS: usize = 8;
 
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 #[serde(rename_all = "snake_case")]
+#[non_exhaustive]
 pub enum FullApiDestinationConfig {
     Memory,
     BigQuery {
@@ -30,6 +31,16 @@ pub enum FullApiDestinationConfig {
         #[serde(skip_serializing_if = "Option::is_none")]
         max_concurrent_streams: Option<usize>,
     },
+    DeltaLake {
+        #[schema(example = "s3://my-bucket/my-path")]
+        base_uri: String,
+        #[schema(example = "s3://my-bucket/my-path")]
+        warehouse: Option<String>,
+        #[schema(example = "[\"date\"]")]
+        partition_columns: Option<Vec<String>>,
+        #[schema(example = 100)]
+        optimize_after_commits: Option<u64>,
+    },
 }
 
 impl From<StoredDestinationConfig> for FullApiDestinationConfig {
@@ -49,6 +60,17 @@ impl From<StoredDestinationConfig> for FullApiDestinationConfig {
                 max_staleness_mins,
                 max_concurrent_streams: Some(max_concurrent_streams),
             },
+            StoredDestinationConfig::DeltaLake {
+                base_uri,
+                warehouse,
+                partition_columns,
+                optimize_after_commits,
+            } => Self::DeltaLake {
+                base_uri,
+                warehouse,
+                partition_columns,
+                optimize_after_commits,
+            },
         }
     }
 }
@@ -64,6 +86,12 @@ pub enum StoredDestinationConfig {
         max_staleness_mins: Option<u16>,
         max_concurrent_streams: usize,
     },
+    DeltaLake {
+        base_uri: String,
+        warehouse: Option<String>,
+        partition_columns: Option<Vec<String>>,
+        optimize_after_commits: Option<u64>,
+    },
 }
 
 impl StoredDestinationConfig {
@@ -83,6 +111,17 @@ impl StoredDestinationConfig {
                 max_staleness_mins,
                 max_concurrent_streams,
             },
+            Self::DeltaLake {
+                base_uri,
+                warehouse,
+                partition_columns,
+                optimize_after_commits,
+            } => DestinationConfig::DeltaLake {
+                base_uri,
+                warehouse,
+                partition_columns,
+                optimize_after_commits,
+            },
         }
     }
 }
@@ -105,6 +144,17 @@ impl From<FullApiDestinationConfig> for StoredDestinationConfig {
                 max_concurrent_streams: max_concurrent_streams
                     .unwrap_or(DEFAULT_MAX_CONCURRENT_STREAMS),
             },
+            FullApiDestinationConfig::DeltaLake {
+                base_uri,
+                warehouse,
+                partition_columns,
+                optimize_after_commits,
+            } => Self::DeltaLake {
+                base_uri,
+                warehouse,
+                partition_columns,
+                optimize_after_commits,
+            },
         }
     }
 }
@@ -136,12 +186,26 @@ impl Encrypt<EncryptedStoredDestinationConfig> for StoredDestinationConfig {
                     max_concurrent_streams,
                 })
             }
+            Self::DeltaLake {
+                base_uri,
+                warehouse,
+                partition_columns,
+                optimize_after_commits,
+            } => {
+                Ok(EncryptedStoredDestinationConfig::DeltaLake {
+                    base_uri,
+                    warehouse,
+                    partition_columns,
+                    optimize_after_commits,
+                })
+            }
         }
     }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
+#[non_exhaustive]
 pub enum EncryptedStoredDestinationConfig {
     Memory,
     BigQuery {
@@ -151,7 +215,13 @@ pub enum EncryptedStoredDestinationConfig {
         max_staleness_mins: Option<u16>,
         max_concurrent_streams: usize,
     },
-}
+    DeltaLake {
+        base_uri: String,
+        warehouse: Option<String>,
+        partition_columns: Option<Vec<String>>,
+        optimize_after_commits: Option<u64>,
+    },
+        }
 
 impl Store for EncryptedStoredDestinationConfig {}
 
@@ -182,6 +252,17 @@ impl Decrypt<StoredDestinationConfig> for EncryptedStoredDestinationConfig {
                     max_concurrent_streams,
                 })
             }
+            Self::DeltaLake {
+                base_uri,
+                warehouse,
+                partition_columns,
+                optimize_after_commits,
+            } => Ok(StoredDestinationConfig::DeltaLake {
+                base_uri,
+                warehouse,
+                partition_columns,
+                optimize_after_commits,
+            }),     
         }
     }
 }

etl-config/src/shared/destination.rs

@@ -8,6 +8,7 @@ use crate::SerializableSecretString;
 /// Each variant corresponds to a different supported destination system.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
+#[non_exhaustive]
 pub enum DestinationConfig {
     /// In-memory destination for ephemeral or test data.
     Memory,
@@ -41,4 +42,12 @@ pub enum DestinationConfig {
         /// - and the configured batch size.
         max_concurrent_streams: usize,
     },
+    DeltaLake {
+        base_uri: String,
+        warehouse: Option<String>,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        partition_columns: Option<Vec<String>>,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        optimize_after_commits: Option<u64>,
+    },
 }

etl-destinations/Cargo.toml

@@ -13,14 +13,14 @@ bigquery = [
     "dep:tokio",
 ]
 iceberg = ["dep:iceberg", "dep:iceberg-catalog-rest"]
-deltalake = ["dep:deltalake"]
+deltalake = ["dep:deltalake", "dep:tokio", "dep:tracing"]
 
 [dependencies]
 etl = { workspace = true }
 etl-postgres = { workspace = true }
 chrono = { workspace = true }
 
-deltalake = { workspace = true, optional = true, features = ["rustls"] }
+deltalake = { workspace = true, optional = true, features = ["rustls", "datafusion"] }
 futures = { workspace = true, optional = true }
 gcp-bigquery-client = { workspace = true, optional = true, features = [
     "rust-tls",
@@ -40,6 +40,7 @@ tracing = { workspace = true, optional = true, default-features = true }
 [dev-dependencies]
 etl = { workspace = true, features = ["test-utils"] }
 etl-telemetry = { workspace = true }
+deltalake = { workspace = true, features = ["rustls", "datafusion", "s3"] }
 
 base64 = { workspace = true }
 chrono = { workspace = true }