feat: utilize StorageConfig for the log store's parse_url_opts

The parse_url_opts trait method has been a little bit of a thorn lately
since we have wanted to pass more rich configuration options for some
log store implementations through it.

This change utilizes the StorageConfig struct which already contains
most of the options that are necessary and gives a reasonable way to
extend the struct depending on what our future needs are.

This _does_ however represent a public API change, so I have bumped all
the necessary crate versions to reflect that.

NOTE: The `handle` argument _might_ now be redundant since StorageConfig
has a `runtime` field inside it.

Closes #3589

Signed-off-by: R. Tyler Croy <[email protected]>

Author: rtyler

crates/aws/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltalake-aws"
-version = "0.10.0"
+version = "0.11.0"
 authors.workspace = true
 keywords.workspace = true
 readme.workspace = true
@@ -13,7 +13,7 @@ rust-version.workspace = true
 
 [dependencies]
 # path dependencies
-deltalake-core = { version = "0.27.0", path = "../core" }
+deltalake-core = { version = "0.28.0", path = "../core" , features = ["cloud"]}
 
 # workspace dependencies
 async-trait = { workspace = true }

crates/aws/src/storage.rs

@@ -13,13 +13,14 @@ use deltalake_core::logstore::object_store::{
     GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore, ObjectStoreScheme,
     PutMultipartOpts, PutOptions, PutPayload, PutResult, Result as ObjectStoreResult,
 };
-use deltalake_core::logstore::{config::str_is_truthy, ObjectStoreFactory, ObjectStoreRef};
+use deltalake_core::logstore::{
+    config::str_is_truthy, ObjectStoreFactory, ObjectStoreRef, StorageConfig,
+};
 use deltalake_core::{DeltaResult, DeltaTableError, ObjectStoreError, Path};
 use futures::stream::BoxStream;
 use futures::Future;
 use object_store::aws::AmazonS3;
 use object_store::client::SpawnedReqwestConnector;
-use object_store::RetryConfig;
 use tokio::runtime::Handle;
 use tracing::log::*;
 use url::Url;
@@ -39,16 +40,15 @@ impl ObjectStoreFactory for S3ObjectStoreFactory {
     fn parse_url_opts(
         &self,
         url: &Url,
-        storage_options: &HashMap<String, String>,
-        retry: &RetryConfig,
+        config: &StorageConfig,
         handle: Option<Handle>,
     ) -> DeltaResult<(ObjectStoreRef, Path)> {
-        let options = self.with_env_s3(storage_options);
+        let options = self.with_env_s3(&config.raw);
 
         // All S3-likes should start their builder the same way
         let mut builder = AmazonS3Builder::new()
             .with_url(url.to_string())
-            .with_retry(retry.clone());
+            .with_retry(config.retry.clone());
 
         if let Some(handle) = handle {
             builder = builder.with_http_connector(SpawnedReqwestConnector::new(handle));

crates/azure/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltalake-azure"
-version = "0.10.0"
+version = "0.11.0"
 authors.workspace = true
 keywords.workspace = true
 readme.workspace = true
@@ -12,7 +12,7 @@ repository.workspace = true
 rust-version.workspace = true
 
 [dependencies]
-deltalake-core = { version = "0.27.0", path = "../core" }
+deltalake-core = { version = "0.28.0", path = "../core", features = ["cloud"] }
 
 # workspace depenndecies
 async-trait = { workspace = true }
@@ -26,7 +26,7 @@ regex = { workspace = true }
 url = { workspace = true }
 
 [dev-dependencies]
-deltalake-core = { version = "0.27.0", path = "../core", features = [
+deltalake-core = { version = "0.28.0", path = "../core", features = [
     "datafusion",
 ] }
 chrono = { workspace = true }

crates/azure/src/lib.rs

@@ -9,7 +9,7 @@ use deltalake_core::logstore::{
 use deltalake_core::{DeltaResult, DeltaTableError, Path};
 use object_store::azure::{AzureConfigKey, MicrosoftAzureBuilder};
 use object_store::client::SpawnedReqwestConnector;
-use object_store::{ObjectStoreScheme, RetryConfig};
+use object_store::ObjectStoreScheme;
 use tokio::runtime::Handle;
 use url::Url;
 
@@ -40,15 +40,13 @@ impl ObjectStoreFactory for AzureFactory {
     fn parse_url_opts(
         &self,
         url: &Url,
-        options: &HashMap<String, String>,
-        retry: &RetryConfig,
+        config: &StorageConfig,
         handle: Option<Handle>,
     ) -> DeltaResult<(ObjectStoreRef, Path)> {
-        let config = config::AzureConfigHelper::try_new(options.as_azure_options())?.build()?;
-
         let mut builder = MicrosoftAzureBuilder::new()
             .with_url(url.to_string())
-            .with_retry(retry.clone());
+            .with_retry(config.retry.clone());
+        let config = config::AzureConfigHelper::try_new(config.raw.as_azure_options())?.build()?;
 
         if let Some(handle) = handle {
             builder = builder.with_http_connector(SpawnedReqwestConnector::new(handle));
@@ -95,3 +93,21 @@ pub fn register_handlers(_additional_prefixes: Option<Url>) {
         logstore_factories().insert(url.clone(), factory.clone());
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::HashMap;
+
+    #[test]
+    fn test_as_azure_options() {
+        use object_store::azure::AzureConfigKey;
+        let mut options = HashMap::default();
+        let key = "AZURE_STORAGE_ACCOUNT_KEY".to_string();
+        let value = "value".to_string();
+        options.insert(key, value.clone());
+
+        let converted = options.as_azure_options();
+        assert_eq!(converted.get(&AzureConfigKey::AccessKey), Some(&value));
+    }
+}

crates/catalog-glue/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltalake-catalog-glue"
-version = "0.11.0"
+version = "0.12.0"
 authors.workspace = true
 keywords.workspace = true
 readme.workspace = true
@@ -15,7 +15,7 @@ rust-version.workspace = true
 async-trait = { workspace = true }
 aws-config = "1"
 aws-sdk-glue = "=1.78.0"
-deltalake-core = { version = "0.27.0", path = "../core" }
+deltalake-core = { version = "0.28.0", path = "../core" }
 thiserror = { workspace = true }
 
 [dev-dependencies]

crates/catalog-unity/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltalake-catalog-unity"
-version = "0.11.0"
+version = "0.12.0"
 authors.workspace = true
 keywords.workspace = true
 readme.workspace = true
@@ -20,10 +20,10 @@ thiserror.workspace = true
 futures.workspace = true
 chrono.workspace = true
 tracing.workspace = true
-deltalake-core = { version = "0.27.0", path = "../core" }
-deltalake-aws = { version = "0.10.0", path = "../aws", optional = true }
-deltalake-azure = { version = "0.10.0", path = "../azure", optional = true }
-deltalake-gcp = { version = "0.11.0", path = "../gcp", optional = true }
+deltalake-core = { version = "0.28.0", path = "../core" }
+deltalake-aws = { version = "0.11.0", path = "../aws", optional = true }
+deltalake-azure = { version = "0.11.0", path = "../azure", optional = true }
+deltalake-gcp = { version = "0.12.0", path = "../gcp", optional = true }
 reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json", "http2"] }
 reqwest-retry = "0.7"
 reqwest-middleware = { version = "0.4.0", features = ["json"] }

crates/catalog-unity/src/lib.rs

@@ -839,24 +839,24 @@ impl ObjectStoreFactory for UnityCatalogFactory {
     fn parse_url_opts(
         &self,
         table_uri: &Url,
-        options: &HashMap<String, String>,
-        _retry: &RetryConfig,
+        config: &StorageConfig,
         handle: Option<Handle>,
     ) -> DeltaResult<(ObjectStoreRef, Path)> {
         let (table_path, temp_creds) = UnityCatalogBuilder::execute_uc_future(
             UnityCatalogBuilder::get_uc_location_and_token(table_uri.as_str()),
         )??;
 
-        let mut storage_options = options.clone();
+        let mut storage_options = config.raw.clone();
         storage_options.extend(temp_creds);
 
         // TODO(roeap): we should not have to go through the table here.
         // ideally we just create the right storage ...
         let mut builder = DeltaTableBuilder::from_uri(&table_path);
 
-        if let Some(handle) = handle {
-            builder = builder.with_io_runtime(IORuntime::RT(handle));
+        if let Some(runtime) = &config.runtime {
+            builder = builder.with_io_runtime(runtime.clone());
         }
+
         if !storage_options.is_empty() {
             builder = builder.with_storage_options(storage_options.clone());
         }

crates/core/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltalake-core"
-version = "0.27.0"
+version = "0.28.0"
 authors.workspace = true
 keywords.workspace = true
 readme.workspace = true
@@ -15,7 +15,7 @@ rust-version.workspace = true
 features = ["datafusion", "json"]
 
 [dependencies]
-deltalake-derive = { version = "0.27.0", path = "../derive" }
+deltalake-derive = { version = "0.28.0", path = "../derive" }
 
 delta_kernel.workspace = true
 

crates/core/src/logstore/config.rs

@@ -300,12 +300,13 @@ pub fn str_is_truthy(val: &str) -> bool {
 mod tests {
     use super::*;
     use maplit::hashmap;
-    use object_store::RetryConfig;
     use std::time::Duration;
 
     // Test retry config parsing
+    #[cfg(feature = "cloud")]
     #[test]
     fn test_retry_config_from_options() {
+        use object_store::RetryConfig;
         let options = hashmap! {
             "max_retries".to_string() => "100".to_string() ,
             "retry_timeout".to_string()  => "300s".to_string() ,
@@ -324,8 +325,10 @@ mod tests {
     }
 
     // Test ParseResult functionality
+    #[cfg(feature = "cloud")]
     #[test]
     fn test_parse_result_handling() {
+        use object_store::RetryConfig;
         let options = hashmap! {
             "retry_timeout".to_string() => "300s".to_string(),
             "max_retries".to_string() => "not_a_number".to_string(),

crates/core/src/logstore/factories.rs

@@ -4,8 +4,6 @@ use std::{
 };
 
 use dashmap::DashMap;
-#[cfg(feature = "cloud")]
-use object_store::RetryConfig;
 use object_store::{path::Path, DynObjectStore};
 use tokio::runtime::Handle;
 use url::Url;
@@ -26,20 +24,10 @@ pub trait ObjectStoreFactory: Send + Sync {
     /// corresponding to the path segment of the URL.
     ///
     /// The store should __NOT__ apply the decorations via the passed `options`
-    #[cfg(feature = "cloud")]
     fn parse_url_opts(
         &self,
         url: &Url,
-        options: &HashMap<String, String>,
-        retry: &RetryConfig,
-        handle: Option<Handle>,
-    ) -> DeltaResult<(ObjectStoreRef, Path)>;
-
-    #[cfg(not(feature = "cloud"))]
-    fn parse_url_opts(
-        &self,
-        url: &Url,
-        options: &HashMap<String, String>,
+        config: &StorageConfig,
         handle: Option<Handle>,
     ) -> DeltaResult<(ObjectStoreRef, Path)>;
 }
@@ -51,11 +39,10 @@ impl ObjectStoreFactory for DefaultObjectStoreFactory {
     fn parse_url_opts(
         &self,
         url: &Url,
-        options: &HashMap<String, String>,
-        #[cfg(feature = "cloud")] _retry: &RetryConfig,
+        config: &StorageConfig,
         handle: Option<Handle>,
     ) -> DeltaResult<(ObjectStoreRef, Path)> {
-        let (mut store, path) = default_parse_url_opts(url, options)?;
+        let (mut store, path) = default_parse_url_opts(url, &config.raw)?;
 
         if let Some(handle) = handle {
             store = Arc::new(DeltaIOStorageBackend::new(store, handle)) as Arc<DynObjectStore>;
@@ -102,11 +89,7 @@ where
     let scheme = Url::parse(&format!("{}://", url.scheme())).unwrap();
     let storage_config = StorageConfig::parse_options(options)?;
     if let Some(factory) = object_store_factories().get(&scheme) {
-        #[cfg(feature = "cloud")]
-        let (store, _prefix) =
-            factory.parse_url_opts(url, &storage_config.raw, &storage_config.retry, None)?;
-        #[cfg(not(feature = "cloud"))]
-        let (store, _prefix) = factory.parse_url_opts(url, &storage_config.raw, None)?;
+        let (store, _prefix) = factory.parse_url_opts(url, &storage_config, None)?;
         let store = storage_config.decorate_store(store, url)?;
         Ok(Arc::new(store))
     } else {
@@ -176,3 +159,6 @@ pub fn logstore_factories() -> LogStoreFactoryRegistry {
         })
         .clone()
 }
+
+#[cfg(test)]
+mod tests {}