From 64c1b1ab4a6a9ed5a904d132c4352ed556a87e35 Mon Sep 17 00:00:00 2001
From: "R. Tyler Croy" <rtyler@brokenco.de>
Date: Tue, 31 Dec 2024 23:35:59 +0000
Subject: [PATCH] fix: ensure defaulted options are set prior to ObjectStore
 creation

The options handling must occur before the ObjectStore is actually
created in order for the object_store crate to properly inherit the
conditional_put configuration setting.

Otherwise an OSError is returned (object_store NotImplemented error)

Signed-off-by: R. Tyler Croy <rtyler@brokenco.de>
---
 crates/aws/src/lib.rs     | 24 +-----------------------
 crates/aws/src/storage.rs | 30 +++++++++++++++++++++++++-----
 2 files changed, 26 insertions(+), 28 deletions(-)

diff --git a/crates/aws/src/lib.rs b/crates/aws/src/lib.rs
index d85ed78236..e32aefcf8b 100644
--- a/crates/aws/src/lib.rs
+++ b/crates/aws/src/lib.rs
@@ -26,11 +26,11 @@ use aws_sdk_dynamodb::{
     Client,
 };
 use deltalake_core::logstore::{default_logstore, logstores, LogStore, LogStoreFactory};
+use deltalake_core::storage::object_store::aws::AmazonS3ConfigKey;
 use deltalake_core::storage::{factories, url_prefix_handler, ObjectStoreRef, StorageOptions};
 use deltalake_core::{DeltaResult, Path};
 use errors::{DynamoDbConfigError, LockClientError};
 use lazy_static::lazy_static;
-use object_store::aws::AmazonS3ConfigKey;
 use regex::Regex;
 use std::{
     collections::HashMap,
@@ -78,28 +78,6 @@ impl LogStoreFactory for S3LogStoreFactory {
                 store,
             )?));
         }
-
-        // All S3-like Object Stores use conditional put, object-store crate however still requires you to explicitly
-        // set this behaviour. We will however assume, when a locking provider/copy-if-not-exists keys are not provided
-        // that PutIfAbsent is supported.
-        // With conditional put in S3-like API we can use the deltalake default logstore which use PutIfAbsent
-        let inner = if !options.0.keys().any(|key| {
-            let key = key.to_ascii_lowercase();
-            [
-                AmazonS3ConfigKey::ConditionalPut.as_ref(),
-                "conditional_put",
-            ]
-            .contains(&key.as_str())
-        }) {
-            let mut inner = options.0.clone();
-            inner.insert("aws_conditional_put".to_string(), "etag".to_string());
-            inner
-        } else {
-            options.0.clone()
-        };
-
-        let options = StorageOptions::from(inner);
-        debug!("S3LogStoreFactory has been asked to create a default LogStore where the underlying store has Conditonal Put enabled - no locking provider required");
         Ok(default_logstore(store, location, &options))
     }
 }
diff --git a/crates/aws/src/storage.rs b/crates/aws/src/storage.rs
index 329ef6023b..9c6473e234 100644
--- a/crates/aws/src/storage.rs
+++ b/crates/aws/src/storage.rs
@@ -60,6 +60,21 @@ impl S3ObjectStoreFactory {
                 }
             }
         }
+
+        // All S3-like Object Stores use conditional put, object-store crate however still requires you to explicitly
+        // set this behaviour. We will however assume, when a locking provider/copy-if-not-exists keys are not provided
+        // that PutIfAbsent is supported.
+        // With conditional put in S3-like API we can use the deltalake default logstore which use PutIfAbsent
+        if !options.0.keys().any(|key| {
+            let key = key.to_ascii_lowercase();
+            [
+                AmazonS3ConfigKey::ConditionalPut.as_ref(),
+                "conditional_put",
+            ]
+            .contains(&key.as_str())
+        }) {
+            options.0.insert("conditional_put".into(), "etag".into());
+        }
         options
     }
 }
@@ -772,10 +787,13 @@ mod tests {
             let combined_options =
                 S3ObjectStoreFactory {}.with_env_s3(&StorageOptions(raw_options));
 
-            assert_eq!(combined_options.0.len(), 4);
+            // Four and then the conditional_put built-in
+            assert_eq!(combined_options.0.len(), 5);
 
-            for v in combined_options.0.values() {
-                assert_eq!(v, "env_key");
+            for (key, v) in combined_options.0 {
+                if key != "conditional_put" {
+                    assert_eq!(v, "env_key");
+                }
             }
         });
     }
@@ -799,8 +817,10 @@ mod tests {
             let combined_options =
                 S3ObjectStoreFactory {}.with_env_s3(&StorageOptions(raw_options));
 
-            for v in combined_options.0.values() {
-                assert_eq!(v, "options_key");
+            for (key, v) in combined_options.0 {
+                if key != "conditional_put" {
+                    assert_eq!(v, "options_key");
+                }
             }
         });
     }