refactor: make put-if-absent default for S3 stores

Cargo.toml

@@ -41,7 +41,7 @@ arrow-ord = { version = "53" }
 arrow-row = { version = "53" }
 arrow-schema = { version = "53" }
 arrow-select = { version = "53" }
-object_store = { version = "0.11" }
+object_store = { version = "0.11.2" }
 parquet = { version = "53" }
 
 # datafusion
@@ -69,6 +69,7 @@ thiserror = { version = "2" }
 url = { version = "2" }
 urlencoding = "2.1.3"
 uuid = { version = "1" }
+path-tree = { version = "0.8.1"} # pin to 0.8.1 due to nightly features
 
 # runtime / async
 async-trait = { version = "0.1" }

README.md

@@ -135,9 +135,9 @@ of features outlined in the Delta [protocol][protocol] is also [tracked](#protoc
 | Storage              |  Rust   | Python  | Comment                                                          |
 | -------------------- | :-----: | :-----: | ---------------------------------------------------------------- |
 | Local                | ![done] | ![done] |                                                                  |
-| S3 - AWS             | ![done] | ![done] | requires lock for concurrent writes                              |
-| S3 - MinIO           | ![done] | ![done] | No lock required when using `AmazonS3ConfigKey::ConditionalPut` with `storage_options = {"conditional_put":"etag"}` |
-| S3 - R2              | ![done] | ![done] | No lock required when using `AmazonS3ConfigKey::ConditionalPut` with `storage_options = {"conditional_put":"etag"}` |
+| S3 - AWS             | ![done] | ![done] |                                                                  |
+| S3 - MinIO           | ![done] | ![done] |                                                                  |
+| S3 - R2              | ![done] | ![done] |                                                                  |
 | Azure Blob           | ![done] | ![done] |                                                                  |
 | Azure ADLS Gen2      | ![done] | ![done] |                                                                  |
 | Microsoft OneLake    | ![done] | ![done] |                                                                  |

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,
@@ -40,6 +40,7 @@ use std::{
 };
 use storage::{S3ObjectStoreFactory, S3StorageOptions};
 use tracing::debug;
+use tracing::warn;
 use url::Url;
 
 #[derive(Clone, Debug, Default)]
@@ -54,19 +55,6 @@ impl LogStoreFactory for S3LogStoreFactory {
     ) -> DeltaResult<Arc<dyn LogStore>> {
         let store = url_prefix_handler(store, Path::parse(location.path())?);
 
-        // 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())
-        }) {
-            debug!("S3LogStoreFactory has been asked to create a default LogStore where the underlying store has Conditonal Put enabled - no locking provider required");
-            return Ok(default_logstore(store, location, options));
-        }
-
         if options.0.keys().any(|key| {
             let key = key.to_ascii_lowercase();
             [
@@ -76,22 +64,21 @@ impl LogStoreFactory for S3LogStoreFactory {
             .contains(&key.as_str())
         }) {
             debug!("S3LogStoreFactory has been asked to create a LogStore where the underlying store has copy-if-not-exists enabled - no locking provider required");
+            warn!("Most S3 object store support conditional put, remove copy_if_not_exists parameter to use a more performant conditional put.");
             return Ok(logstore::default_s3_logstore(store, location, options));
         }
 
         let s3_options = S3StorageOptions::from_map(&options.0)?;
-
-        if s3_options.locking_provider.as_deref() != Some("dynamodb") {
-            debug!("S3LogStoreFactory has been asked to create a LogStore without the dynamodb locking provider");
-            return Ok(logstore::default_s3_logstore(store, location, options));
+        if s3_options.locking_provider.as_deref() == Some("dynamodb") {
+            debug!("S3LogStoreFactory has been asked to create a LogStore with the dynamodb locking provider");
+            return Ok(Arc::new(logstore::S3DynamoDbLogStore::try_new(
+                location.clone(),
+                options.clone(),
+                &s3_options,
+                store,
+            )?));
         }
-
-        Ok(Arc::new(logstore::S3DynamoDbLogStore::try_new(
-            location.clone(),
-            options.clone(),
-            &s3_options,
-            store,
-        )?))
+        Ok(default_logstore(store, location, &options))
     }
 }
 
@@ -777,7 +764,22 @@ mod tests {
         let logstore = factory
             .with_options(Arc::new(store), &url, &StorageOptions::from(HashMap::new()))
             .unwrap();
-        assert_eq!(logstore.name(), "S3LogStore");
+        assert_eq!(logstore.name(), "DefaultLogStore");
+    }
+
+    #[test]
+    #[serial]
+    fn test_logstore_factory_with_locking_provider() {
+        let factory = S3LogStoreFactory::default();
+        let store = InMemory::new();
+        let url = Url::parse("s3://test-bucket").unwrap();
+        unsafe {
+            std::env::set_var(crate::constants::AWS_S3_LOCKING_PROVIDER, "dynamodb");
+        }
+        let logstore = factory
+            .with_options(Arc::new(store), &url, &StorageOptions::from(HashMap::new()))
+            .unwrap();
+        assert_eq!(logstore.name(), "S3DynamoDbLogStore");
     }
 
     #[test]

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
     }
 }
@@ -110,24 +125,19 @@ fn aws_storage_handler(
     store: ObjectStoreRef,
     options: &StorageOptions,
 ) -> DeltaResult<ObjectStoreRef> {
-    // If the copy-if-not-exists env var is set or ConditionalPut is set, we don't need to instantiate a locking client or check for allow-unsafe-rename.
-    if options
-        .0
-        .contains_key(AmazonS3ConfigKey::CopyIfNotExists.as_ref())
-        || options
-            .0
-            .contains_key(AmazonS3ConfigKey::ConditionalPut.as_ref())
+    let s3_options = S3StorageOptions::from_map(&options.0)?;
+    // Nearly all S3 Object stores support conditional put, so we change the default to always returning an S3 Object store
+    // unless explicitly passing a locking provider key or allow_unsafe_rename. Then we will pass it to the old S3StorageBackend.
+    if s3_options.locking_provider.as_deref() == Some("dynamodb") || s3_options.allow_unsafe_rename
     {
-        Ok(store)
-    } else {
-        let s3_options = S3StorageOptions::from_map(&options.0)?;
-
         let store = S3StorageBackend::try_new(
             store,
             Some("dynamodb") == s3_options.locking_provider.as_deref()
                 || s3_options.allow_unsafe_rename,
         )?;
         Ok(Arc::new(store))
+    } else {
+        Ok(store)
     }
 }
 
@@ -777,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");
+                }
             }
         });
     }
@@ -804,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");
+                }
             }
         });
     }

python/pyproject.toml

@@ -4,7 +4,6 @@ build-backend = "maturin"
 
 [project]
 name = "deltalake"
-version = "0.22.4"
 description = "Native Delta Lake Python binding based on delta-rs with Pandas integration"
 readme = "README.md"
 license = { file = "licenses/deltalake_license.txt" }
@@ -18,6 +17,7 @@ classifiers = [
     "Programming Language :: Python :: 3.12",
 ]
 dependencies = ["pyarrow>=16"]
+dynamic = ["version"]
 
 [project.optional-dependencies]
 pandas = ["pandas"]