Pass storage_options directly instead of reading from env

Author: abhi-airspace-intelligence

etl-api/src/configs/destination.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use etl_config::SerializableSecretString;
 use etl_config::shared::DestinationConfig;
 use secrecy::ExposeSecret;
@@ -34,10 +36,10 @@ pub enum FullApiDestinationConfig {
     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 = "{\"aws_access_key_id\": \"https://my-endpoint.com\"}")]
+        storage_options: Option<HashMap<String, String>>,
+        #[schema(example = "{\"my_table\": [\"date\"]}")]
+        partition_columns: Option<HashMap<String, Vec<String>>>,
         #[schema(example = 100)]
         optimize_after_commits: Option<u64>,
     },
@@ -62,12 +64,12 @@ impl From<StoredDestinationConfig> for FullApiDestinationConfig {
             },
             StoredDestinationConfig::DeltaLake {
                 base_uri,
-                warehouse,
+                storage_options,
                 partition_columns,
                 optimize_after_commits,
             } => Self::DeltaLake {
                 base_uri,
-                warehouse,
+                storage_options,
                 partition_columns,
                 optimize_after_commits,
             },
@@ -88,8 +90,8 @@ pub enum StoredDestinationConfig {
     },
     DeltaLake {
         base_uri: String,
-        warehouse: Option<String>,
-        partition_columns: Option<Vec<String>>,
+        storage_options: Option<HashMap<String, String>>,
+        partition_columns: Option<HashMap<String, Vec<String>>>,
         optimize_after_commits: Option<u64>,
     },
 }
@@ -113,12 +115,12 @@ impl StoredDestinationConfig {
             },
             Self::DeltaLake {
                 base_uri,
-                warehouse,
+                storage_options,
                 partition_columns,
                 optimize_after_commits,
             } => DestinationConfig::DeltaLake {
                 base_uri,
-                warehouse,
+                storage_options,
                 partition_columns,
                 optimize_after_commits,
             },
@@ -146,12 +148,12 @@ impl From<FullApiDestinationConfig> for StoredDestinationConfig {
             },
             FullApiDestinationConfig::DeltaLake {
                 base_uri,
-                warehouse,
+                storage_options,
                 partition_columns,
                 optimize_after_commits,
             } => Self::DeltaLake {
                 base_uri,
-                warehouse,
+                storage_options,
                 partition_columns,
                 optimize_after_commits,
             },
@@ -188,12 +190,12 @@ impl Encrypt<EncryptedStoredDestinationConfig> for StoredDestinationConfig {
             }
             Self::DeltaLake {
                 base_uri,
-                warehouse,
+                storage_options,
                 partition_columns,
                 optimize_after_commits,
             } => Ok(EncryptedStoredDestinationConfig::DeltaLake {
                 base_uri,
-                warehouse,
+                storage_options,
                 partition_columns,
                 optimize_after_commits,
             }),
@@ -215,8 +217,8 @@ pub enum EncryptedStoredDestinationConfig {
     },
     DeltaLake {
         base_uri: String,
-        warehouse: Option<String>,
-        partition_columns: Option<Vec<String>>,
+        storage_options: Option<HashMap<String, String>>,
+        partition_columns: Option<HashMap<String, Vec<String>>>,
         optimize_after_commits: Option<u64>,
     },
 }
@@ -252,12 +254,12 @@ impl Decrypt<StoredDestinationConfig> for EncryptedStoredDestinationConfig {
             }
             Self::DeltaLake {
                 base_uri,
-                warehouse,
+                storage_options,
                 partition_columns,
                 optimize_after_commits,
             } => Ok(StoredDestinationConfig::DeltaLake {
                 base_uri,
-                warehouse,
+                storage_options,
                 partition_columns,
                 optimize_after_commits,
             }),

etl-config/src/shared/destination.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use serde::{Deserialize, Serialize};
 
 use crate::SerializableSecretString;
@@ -44,9 +46,9 @@ pub enum DestinationConfig {
     },
     DeltaLake {
         base_uri: String,
-        warehouse: Option<String>,
+        storage_options: Option<HashMap<String, String>>,
         #[serde(skip_serializing_if = "Option::is_none")]
-        partition_columns: Option<Vec<String>>,
+        partition_columns: Option<HashMap<String, Vec<String>>>,
         #[serde(skip_serializing_if = "Option::is_none")]
         optimize_after_commits: Option<u64>,
     },

etl-destinations/src/delta/client.rs

@@ -1,30 +1,43 @@
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 
 use super::schema::postgres_to_delta_schema;
 use deltalake::arrow::record_batch::RecordBatch;
-use deltalake::{DeltaOps, DeltaResult, DeltaTable, open_table};
+use deltalake::{DeltaOps, DeltaResult, DeltaTable, DeltaTableBuilder, open_table};
 use etl::types::TableSchema;
 
 /// Client for connecting to Delta Lake tables.
 #[derive(Clone)]
-pub struct DeltaLakeClient {}
+pub struct DeltaLakeClient {
+    storage_options: Option<HashMap<String, String>>,
+}
 
 impl Default for DeltaLakeClient {
     fn default() -> Self {
-        Self::new()
+        Self::new(None)
     }
 }
 
 impl DeltaLakeClient {
     /// Create a new client.
-    pub fn new() -> Self {
-        Self {}
+    pub fn new(storage_options: Option<HashMap<String, String>>) -> Self {
+        Self { storage_options }
+    }
+
+    fn get_table_with_storage_options(&self, table_uri: &str) -> DeltaResult<DeltaTableBuilder> {
+        let mut builder = DeltaTableBuilder::from_valid_uri(table_uri)?;
+        if let Some(storage_options) = &self.storage_options {
+            builder = builder.with_storage_options(storage_options.clone());
+        }
+        Ok(builder)
     }
 
     /// Returns true if a Delta table exists at the given uri/path.
     pub async fn table_exists(&self, table_uri: &str) -> bool {
-        open_table(table_uri).await.is_ok()
+        let Ok(builder) = self.get_table_with_storage_options(table_uri) else {
+            return false;
+        };
+        builder.load().await.is_ok()
     }
 
     /// Create a Delta table at `table_uri` if it doesn't exist, using the provided Postgres schema.
@@ -39,7 +52,12 @@ impl DeltaLakeClient {
 
         let delta_schema = postgres_to_delta_schema(table_schema)?;
 
-        let ops = DeltaOps::try_from_uri(table_uri).await?;
+        let ops = if let Some(storage_options) = &self.storage_options {
+            DeltaOps::try_from_uri_with_storage_options(table_uri, storage_options.clone()).await?
+        } else {
+            DeltaOps::try_from_uri(table_uri).await?
+        };
+
         let table = ops
             .create()
             // TODO(abhi): Figure out how to avoid the clone

etl-destinations/src/delta/core.rs

@@ -18,10 +18,10 @@ use crate::delta::{DeltaLakeClient, TableRowEncoder};
 pub struct DeltaDestinationConfig {
     /// Base URI for Delta table storage (e.g., "s3://bucket/warehouse", "file:///tmp/delta")
     pub base_uri: String,
-    /// Optional warehouse path for organizing tables
-    pub warehouse: Option<String>,
+    /// Optional storage options passed to underlying object store
+    pub storage_options: Option<HashMap<String, String>>,
     /// Columns to use for partitioning (per table)
-    pub partition_columns: Option<Vec<String>>,
+    pub partition_columns: Option<HashMap<String, Vec<String>>>,
     /// Run OPTIMIZE every N commits (None = disabled)
     pub optimize_after_commits: Option<NonZeroU64>,
 }
@@ -30,7 +30,7 @@ impl Default for DeltaDestinationConfig {
     fn default() -> Self {
         Self {
             base_uri: "file:///tmp/delta".to_string(),
-            warehouse: None,
+            storage_options: None,
             partition_columns: None,
             optimize_after_commits: None,
         }
@@ -56,7 +56,7 @@ where
     /// Create a new Delta Lake destination
     pub fn new(store: S, config: DeltaDestinationConfig) -> Self {
         Self {
-            client: DeltaLakeClient::new(),
+            client: DeltaLakeClient::new(config.storage_options.clone()),
             store,
             config,
             table_cache: Arc::new(RwLock::new(HashMap::new())),

etl-destinations/tests/common/delta.rs

@@ -2,6 +2,7 @@ use etl::store::schema::SchemaStore;
 use etl::store::state::StateStore;
 use etl::types::TableName;
 use etl_destinations::delta::{DeltaDestinationConfig, DeltaLakeDestination};
+use std::collections::HashMap;
 use std::env;
 use uuid::Uuid;
 
@@ -62,18 +63,6 @@ impl DeltaLakeDatabase {
         let warehouse_path = random_warehouse_path();
         let s3_base_uri = format!("s3://{}/{}", bucket, warehouse_path);
 
-        // Set up AWS environment for delta-rs to use minio
-        unsafe {
-            env::set_var("AWS_ENDPOINT_URL", &endpoint);
-            env::set_var("AWS_ACCESS_KEY_ID", &access_key);
-            env::set_var("AWS_SECRET_ACCESS_KEY", &secret_key);
-            env::set_var("AWS_REGION", "local-01");
-            env::set_var("AWS_S3_ALLOW_UNSAFE_RENAME", "true");
-            env::set_var("AWS_S3_PATH_STYLE_ACCESS", "true");
-            env::set_var("AWS_USE_HTTPS", "false");
-            env::set_var("AWS_ALLOW_HTTP", "true");
-        }
-
         Self {
             warehouse_path,
             s3_base_uri,
@@ -87,14 +76,27 @@ impl DeltaLakeDatabase {
     /// Creates a [`DeltaLakeDestination`] configured for this database instance.
     ///
     /// Returns a destination suitable for ETL operations, configured with
-    /// the test warehouse location.
+    /// the test warehouse location and appropriate storage options for MinIO.
     pub async fn build_destination<S>(&self, store: S) -> DeltaLakeDestination<S>
     where
         S: StateStore + SchemaStore + Send + Sync,
     {
+        // Create storage options HashMap with AWS-compatible settings for MinIO
+        let mut storage_options = HashMap::new();
+        storage_options.insert("endpoint".to_string(), self.endpoint.clone());
+        storage_options.insert("access_key_id".to_string(), self.access_key.clone());
+        storage_options.insert("secret_access_key".to_string(), self.secret_key.clone());
+        storage_options.insert("region".to_string(), "local-01".to_string());
+        storage_options.insert("allow_http".to_string(), "true".to_string());
+        // Use path-style requests for MinIO compatibility (opposite of virtual hosted style)
+        storage_options.insert(
+            "virtual_hosted_style_request".to_string(),
+            "false".to_string(),
+        );
+
         let config = DeltaDestinationConfig {
             base_uri: self.s3_base_uri.clone(),
-            warehouse: Some(self.warehouse_path.clone()),
+            storage_options: Some(storage_options),
             partition_columns: None,
             optimize_after_commits: None,
         };

etl-replicator/src/core.rs

@@ -70,15 +70,15 @@ pub async fn start_replicator_with_config(
         }
         DestinationConfig::DeltaLake {
             base_uri,
-            warehouse,
+            storage_options,
             partition_columns,
             optimize_after_commits,
         } => {
             let destination = DeltaLakeDestination::new(
                 state_store.clone(),
                 DeltaDestinationConfig {
                     base_uri: base_uri.clone(),
-                    warehouse: warehouse.clone(),
+                    storage_options: storage_options.clone(),
                     partition_columns: partition_columns.clone(),
                     optimize_after_commits: optimize_after_commits.map(|n| n.try_into().unwrap()),
                 },
@@ -121,12 +121,12 @@ fn log_destination_config(config: &DestinationConfig) {
             )
         }
         DestinationConfig::DeltaLake {
-            base_uri: _,
-            warehouse: _,
+            base_uri,
+            storage_options: _,
             partition_columns: _,
             optimize_after_commits: _,
         } => {
-            debug!("using delta lake destination config");
+            debug!(base_uri = base_uri, "using delta lake destination config");
         }
         _ => unimplemented!("destination config not implemented"),
     }