feat: experimental objectstore wrapper with IO runtime

crates/aws/src/storage.rs

@@ -9,7 +9,8 @@ use deltalake_core::storage::object_store::{
     ObjectStore, PutOptions, PutResult, Result as ObjectStoreResult,
 };
 use deltalake_core::storage::{
-    limit_store_handler, str_is_truthy, ObjectStoreFactory, ObjectStoreRef, StorageOptions,
+    deltars_store_handler, limit_store_handler, str_is_truthy, ObjectStoreFactory, ObjectStoreRef,
+    StorageOptions,
 };
 use deltalake_core::{DeltaResult, ObjectStoreError, Path};
 use futures::stream::BoxStream;
@@ -79,25 +80,34 @@ impl ObjectStoreFactory for S3ObjectStoreFactory {
             }),
         )?;
 
-        let store = limit_store_handler(inner, &options);
-
-        // If the copy-if-not-exists env var 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())
-        {
-            Ok((store, prefix))
-        } else {
-            let s3_options = S3StorageOptions::from_map(&storage_options.0)?;
+        let store = deltars_store_handler(
+            aws_storage_handler(limit_store_handler(inner, &options), &options)?,
+            &storage_options,
+        );
 
-            let store = S3StorageBackend::try_new(
-                store,
-                Some("dynamodb") == s3_options.locking_provider.as_deref()
-                    || s3_options.allow_unsafe_rename,
-            )?;
+        Ok((store, prefix))
+    }
+}
 
-            Ok((Arc::new(store), prefix))
-        }
+fn aws_storage_handler(
+    store: ObjectStoreRef,
+    options: &StorageOptions,
+) -> DeltaResult<ObjectStoreRef> {
+    // If the copy-if-not-exists env var 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())
+    {
+        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))
     }
 }
 

crates/azure/src/lib.rs

@@ -4,8 +4,8 @@ use std::sync::Arc;
 
 use deltalake_core::logstore::{default_logstore, logstores, LogStore, LogStoreFactory};
 use deltalake_core::storage::{
-    factories, limit_store_handler, url_prefix_handler, ObjectStoreFactory, ObjectStoreRef,
-    StorageOptions,
+    deltars_store_handler, factories, limit_store_handler, url_prefix_handler, ObjectStoreFactory,
+    ObjectStoreRef, StorageOptions,
 };
 use deltalake_core::{DeltaResult, Path};
 use object_store::azure::AzureConfigKey;
@@ -44,7 +44,10 @@ impl ObjectStoreFactory for AzureFactory {
     ) -> DeltaResult<(ObjectStoreRef, Path)> {
         let config = config::AzureConfigHelper::try_new(options.as_azure_options())?.build()?;
         let (inner, prefix) = parse_url_opts(url, config)?;
-        let store = limit_store_handler(url_prefix_handler(inner, prefix.clone()), options);
+        let store = deltars_store_handler(
+            limit_store_handler(url_prefix_handler(inner, prefix.clone()), options),
+            options,
+        );
         Ok((store, prefix))
     }
 }

crates/core/src/operations/writer.rs

@@ -1,16 +1,20 @@
 //! Abstractions and implementations for writing data to delta tables
 
 use std::collections::HashMap;
+use std::sync::{Arc, OnceLock};
 
 use arrow_array::RecordBatch;
 use arrow_schema::{ArrowError, SchemaRef as ArrowSchemaRef};
 use bytes::Bytes;
 use delta_kernel::expressions::Scalar;
+use futures::future::BoxFuture;
+use futures::{FutureExt, TryFutureExt};
 use indexmap::IndexMap;
 use object_store::{path::Path, ObjectStore};
 use parquet::arrow::ArrowWriter;
 use parquet::basic::Compression;
 use parquet::file::properties::WriterProperties;
+use tokio::runtime::Runtime;
 use tracing::debug;
 
 use crate::crate_version;
@@ -369,6 +373,7 @@ impl PartitionWriter {
 
         // write file to object store
         self.object_store.put(&path, buffer.into()).await?;
+
         self.files_written.push(
             create_add(
                 &self.config.partition_values,