refactor: isolate factories for storage / log store integrations

Signed-off-by: Robert Pack <[email protected]>

Author: roeap

crates/aws/Cargo.toml

@@ -12,29 +12,43 @@ repository.workspace = true
 rust-version.workspace = true
 
 [dependencies]
+# path dependencies
 deltalake-core = { version = "0.25.0", path = "../core", features = ["cloud"] }
-aws-smithy-runtime-api = { version="1.7" }
-aws-smithy-runtime = { version="1.7", optional = true}
-aws-credential-types = { version="1.2", features = ["hardcoded-credentials"]}
-aws-config = { version = "1.5", default-features = false, features = ["behavior-version-latest","rt-tokio", "credentials-process", "sso"] }
-aws-sdk-dynamodb = {version = "1.45", default-features = false, features = ["behavior-version-latest", "rt-tokio"] }
-aws-sdk-sts = {version = "1.42", default-features = false, features = ["behavior-version-latest", "rt-tokio"] }
-maplit = "1"
 
 # workspace dependencies
 async-trait = { workspace = true }
 bytes = { workspace = true }
 chrono = { workspace = true }
 futures = { workspace = true }
 tracing = { workspace = true }
-object_store = { workspace = true, features = ["aws"]}
+object_store = { workspace = true, features = ["aws"] }
+regex = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
-regex = { workspace = true }
 uuid = { workspace = true, features = ["serde", "v4"] }
 url = { workspace = true }
-backon = { version = "1",default-features = false, features = [ "tokio-sleep" ] }
+
+# crates.io dependencies
+aws-smithy-runtime-api = { version = "1.7" }
+aws-smithy-runtime = { version = "1.7", optional = true }
+aws-credential-types = { version = "1.2", features = ["hardcoded-credentials"] }
+aws-config = { version = "1.5", default-features = false, features = [
+    "behavior-version-latest",
+    "rt-tokio",
+    "credentials-process",
+    "sso",
+] }
+aws-sdk-dynamodb = { version = "1.45", default-features = false, features = [
+    "behavior-version-latest",
+    "rt-tokio",
+] }
+aws-sdk-sts = { version = "1.42", default-features = false, features = [
+    "behavior-version-latest",
+    "rt-tokio",
+] }
+backon = { version = "1", default-features = false, features = ["tokio-sleep"] }
 hyper-tls = { version = "0.5", optional = true }
+maplit = "1"
 
 [dev-dependencies]
 deltalake-core = { path = "../core", features = ["datafusion"] }
@@ -51,7 +65,7 @@ integration_test = []
 native-tls = [
     "aws-config/client-hyper",
     "aws-smithy-runtime/connector-hyper-0-14-x",
-    "hyper-tls"
+    "hyper-tls",
 ]
 rustls = [
     "aws-config/client-hyper",

crates/aws/src/constants.rs

@@ -74,19 +74,19 @@ pub const AWS_S3_GET_INTERNAL_SERVER_ERROR_RETRIES: &str =
 /// The web identity token file to use when using a web identity provider.
 ///
 /// NOTE: web identity related options are set in the environment when
-/// creating an instance of [crate::storage::s3::S3StorageOptions].
+/// creating an instance of [S3StorageOptions](crate::storage::S3StorageOptions).
 /// See also <https://docs.rs/rusoto_sts/0.47.0/rusoto_sts/struct.WebIdentityProvider.html#method.from_k8s_env>.
 pub const AWS_WEB_IDENTITY_TOKEN_FILE: &str = "AWS_WEB_IDENTITY_TOKEN_FILE";
 /// The role name to use for web identity.
 ///
 /// NOTE: web identity related options are set in the environment when
-/// creating an instance of [crate::storage::s3::S3StorageOptions].
+/// creating an instance of [S3StorageOptions](crate::storage::S3StorageOptions).
 /// See also <https://docs.rs/rusoto_sts/0.47.0/rusoto_sts/struct.WebIdentityProvider.html#method.from_k8s_env>.
 pub const AWS_ROLE_ARN: &str = "AWS_ROLE_ARN";
 /// The role session name to use for web identity.
 ///
 /// NOTE: web identity related options are set in the environment when
-/// creating an instance of [crate::storage::s3::S3StorageOptions].
+/// creating an instance of [S3StorageOptions](crate::storage::S3StorageOptions).
 /// See also <https://docs.rs/rusoto_sts/0.47.0/rusoto_sts/struct.WebIdentityProvider.html#method.from_k8s_env>.
 pub const AWS_ROLE_SESSION_NAME: &str = "AWS_ROLE_SESSION_NAME";
 /// Allow http connections - mainly useful for integration tests
@@ -109,7 +109,7 @@ pub const AWS_FORCE_CREDENTIAL_LOAD: &str = "AWS_FORCE_CREDENTIAL_LOAD";
 
 /// The list of option keys owned by the S3 module.
 /// Option keys not contained in this list will be added to the `extra_opts`
-/// field of [crate::storage::s3::S3StorageOptions].
+/// field of [S3StorageOptions](crate::storage::S3StorageOptions).
 #[allow(deprecated)]
 pub const S3_OPTS: &[&str] = &[
     AWS_ENDPOINT_URL,

crates/aws/src/credentials.rs

@@ -255,7 +255,7 @@ fn assume_session_name(options: &HashMap<String, String>) -> String {
 }
 
 /// Take a set of [StorageOptions] and produce an appropriate AWS SDK [SdkConfig]
-/// for use with various AWS SDK APIs, such as in our [crate::logstore::S3DynamoDbLogStore]
+/// for use with various AWS SDK APIs, such as in our [S3DynamoDbLogStore](crate::logstore::S3DynamoDbLogStore)
 pub async fn resolve_credentials(options: &HashMap<String, String>) -> DeltaResult<SdkConfig> {
     let default_provider = DefaultCredentialsChain::builder().build().await;
 

crates/aws/src/lib.rs

@@ -1,7 +1,7 @@
 //! AWS S3 and similar tooling for delta-rs
 //!
-//! This module also contains the [S3DynamoDbLogStore] implementation for concurrent writer support
-//! with AWS S3 specifically.
+//! This module also contains the [S3DynamoDbLogStore](crate::logstore::S3DynamoDbLogStore)
+//! implementation for concurrent writer support with AWS S3 specifically.
 
 pub mod constants;
 mod credentials;
@@ -10,6 +10,7 @@ pub mod logstore;
 #[cfg(feature = "native-tls")]
 mod native;
 pub mod storage;
+
 use aws_config::Region;
 use aws_config::SdkConfig;
 pub use aws_credential_types::provider::SharedCredentialsProvider;
@@ -26,8 +27,10 @@ use aws_sdk_dynamodb::{
     Client,
 };
 use deltalake_core::logstore::object_store::aws::AmazonS3ConfigKey;
-use deltalake_core::logstore::{default_logstore, logstores, LogStore, LogStoreFactory};
-use deltalake_core::logstore::{factories, ObjectStoreRef, StorageConfig};
+use deltalake_core::logstore::{
+    default_logstore, logstore_factories, object_store_factories, LogStore, LogStoreFactory,
+    ObjectStoreRef, StorageConfig,
+};
 use deltalake_core::{DeltaResult, Path};
 use errors::{DynamoDbConfigError, LockClientError};
 use regex::Regex;
@@ -83,14 +86,16 @@ impl LogStoreFactory for S3LogStoreFactory {
     }
 }
 
-/// Register an [ObjectStoreFactory] for common S3 [Url] schemes
+/// Register an [ObjectStoreFactory] for common S3 url schemes.
+///
+/// [ObjectStoreFactory]: deltalake_core::logstore::ObjectStoreFactory
 pub fn register_handlers(_additional_prefixes: Option<Url>) {
     let object_stores = Arc::new(S3ObjectStoreFactory::default());
     let log_stores = Arc::new(S3LogStoreFactory::default());
     for scheme in ["s3", "s3a"].iter() {
         let url = Url::parse(&format!("{scheme}://")).unwrap();
-        factories().insert(url.clone(), object_stores.clone());
-        logstores().insert(url.clone(), log_stores.clone());
+        object_store_factories().insert(url.clone(), object_stores.clone());
+        logstore_factories().insert(url.clone(), log_stores.clone());
     }
 }
 

crates/aws/src/storage.rs

@@ -23,6 +23,7 @@ use tracing::log::*;
 use url::Url;
 
 use crate::constants;
+use crate::credentials::AWSForObjectStore;
 use crate::errors::DynamoDbConfigError;
 
 const STORE_NAME: &str = "DeltaS3ObjectStore";
@@ -42,30 +43,28 @@ impl ObjectStoreFactory for S3ObjectStoreFactory {
         let options = self.with_env_s3(storage_options);
 
         // All S3-likes should start their builder the same way
-        let mut builder = AmazonS3Builder::new().with_url(url.to_string());
-
+        let mut builder = AmazonS3Builder::new()
+            .with_url(url.to_string())
+            .with_retry(retry.clone());
         for (key, value) in options.iter() {
             if let Ok(key) = AmazonS3ConfigKey::from_str(&key.to_ascii_lowercase()) {
                 builder = builder.with_config(key, value.clone());
             }
         }
 
+        let s3_options = S3StorageOptions::from_map(&options)?;
+        if let Some(ref sdk_config) = s3_options.sdk_config {
+            builder =
+                builder.with_credentials(Arc::new(AWSForObjectStore::new(sdk_config.clone())));
+        }
+
         let (_, path) =
             ObjectStoreScheme::parse(url).map_err(|e| DeltaTableError::GenericError {
                 source: Box::new(e),
             })?;
         let prefix = Path::parse(path)?;
 
-        let s3_options: S3StorageOptions = S3StorageOptions::from_map(&options)?;
-
-        if let Some(ref sdk_config) = s3_options.sdk_config {
-            builder = builder.with_credentials(Arc::new(
-                crate::credentials::AWSForObjectStore::new(sdk_config.clone()),
-            ));
-        }
-
-        let inner = builder.with_retry(retry.clone()).build()?;
-        let store = aws_storage_handler(inner, &s3_options)?;
+        let store = aws_storage_handler(builder.build()?, &s3_options)?;
         debug!("Initialized the object store: {store:?}");
 
         Ok((store, prefix))
@@ -154,7 +153,7 @@ impl PartialEq for S3StorageOptions {
 }
 
 impl S3StorageOptions {
-    /// Creates an instance of S3StorageOptions from the given HashMap.
+    /// Creates an instance of [`S3StorageOptions`] from the given HashMap.
     pub fn from_map(options: &HashMap<String, String>) -> DeltaResult<S3StorageOptions> {
         let extra_opts: HashMap<String, String> = options
             .iter()
@@ -407,7 +406,7 @@ impl ObjectStore for S3StorageBackend {
     }
 }
 
-/// Storage option keys to use when creating [crate::storage::s3::S3StorageOptions].
+/// Storage option keys to use when creating [`S3StorageOptions`].
 ///
 /// The same key should be used whether passing a key in the hashmap or setting it as an environment variable.
 /// Provided keys may include configuration for the S3 backend and also the optional DynamoDb lock used for atomic rename.

crates/aws/tests/repair_s3_rename_test.rs

@@ -3,8 +3,8 @@
 use bytes::Bytes;
 use deltalake_aws::storage::S3StorageBackend;
 use deltalake_core::logstore::object_store::{
-    DynObjectStore, Error as ObjectStoreError, GetOptions, GetResult, ListResult, MultipartId,
-    ObjectMeta, PutOptions, PutResult, Result as ObjectStoreResult,
+    DynObjectStore, Error as ObjectStoreError, GetOptions, GetResult, ListResult, ObjectMeta,
+    PutOptions, PutResult, Result as ObjectStoreResult,
 };
 use deltalake_core::{DeltaTableBuilder, ObjectStore, Path};
 use deltalake_test::utils::IntegrationContext;
@@ -13,7 +13,6 @@ use object_store::{MultipartUpload, PutMultipartOpts, PutPayload};
 use serial_test::serial;
 use std::ops::Range;
 use std::sync::{Arc, Mutex};
-use tokio::io::AsyncWrite;
 use tokio::task::JoinHandle;
 use tokio::time::Duration;
 

crates/azure/src/lib.rs

@@ -2,8 +2,10 @@ use std::collections::HashMap;
 use std::str::FromStr;
 use std::sync::Arc;
 
-use deltalake_core::logstore::{default_logstore, logstores, LogStore, LogStoreFactory};
-use deltalake_core::logstore::{factories, ObjectStoreFactory, ObjectStoreRef, StorageConfig};
+use deltalake_core::logstore::{
+    default_logstore, logstore_factories, object_store_factories, LogStore, LogStoreFactory,
+    ObjectStoreFactory, ObjectStoreRef, StorageConfig,
+};
 use deltalake_core::{DeltaResult, DeltaTableError, Path};
 use object_store::azure::{AzureConfigKey, MicrosoftAzureBuilder};
 use object_store::{ObjectStoreScheme, RetryConfig};
@@ -41,18 +43,20 @@ impl ObjectStoreFactory for AzureFactory {
     ) -> 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());
+        for (key, value) in config.iter() {
+            builder = builder.with_config(*key, value.clone());
+        }
+        let store = builder.build()?;
+
         let (_, path) =
             ObjectStoreScheme::parse(url).map_err(|e| DeltaTableError::GenericError {
                 source: Box::new(e),
             })?;
         let prefix = Path::parse(path)?;
 
-        let mut builder = MicrosoftAzureBuilder::new().with_url(url.to_string());
-        for (key, value) in config.iter() {
-            builder = builder.with_config(*key, value.clone());
-        }
-        let store = builder.with_retry(retry.clone()).build()?;
-
         Ok((Arc::new(store), prefix))
     }
 }
@@ -73,7 +77,7 @@ pub fn register_handlers(_additional_prefixes: Option<Url>) {
     let factory = Arc::new(AzureFactory {});
     for scheme in ["az", "adl", "azure", "abfs", "abfss"].iter() {
         let url = Url::parse(&format!("{scheme}://")).unwrap();
-        factories().insert(url.clone(), factory.clone());
-        logstores().insert(url.clone(), factory.clone());
+        object_store_factories().insert(url.clone(), factory.clone());
+        logstore_factories().insert(url.clone(), factory.clone());
     }
 }

crates/azure/tests/integration.rs

@@ -46,7 +46,6 @@ async fn test_concurrency_azure() -> TestResult {
 // comment](https://github.com/delta-io/delta-rs/pull/1564#issuecomment-1721048753) and we should
 // figure out a way to re-enable this test at least in the GitHub Actions CI environment
 #[ignore]
-#[cfg(feature = "azure")]
 #[tokio::test]
 #[serial]
 async fn test_object_store_onelake() -> TestResult {
@@ -60,7 +59,6 @@ async fn test_object_store_onelake() -> TestResult {
 // comment](https://github.com/delta-io/delta-rs/pull/1564#issuecomment-1721048753) and we should
 // figure out a way to re-enable this test at least in the GitHub Actions CI environment
 #[ignore]
-#[cfg(feature = "azure")]
 #[tokio::test]
 #[serial]
 async fn test_object_store_onelake_abfs() -> TestResult {

crates/catalog-unity/src/lib.rs

@@ -9,7 +9,7 @@ compile_error!(
 
 use datafusion_common::DataFusionError;
 use deltalake_core::logstore::{
-    default_logstore, logstores, object_store::RetryConfig, LogStore, LogStoreFactory,
+    default_logstore, logstore_factories, object_store::RetryConfig, LogStore, LogStoreFactory,
     StorageConfig,
 };
 use reqwest::header::{HeaderValue, InvalidHeaderValue, AUTHORIZATION};
@@ -35,7 +35,7 @@ use deltalake_core::{
 
 use crate::client::retry::*;
 use deltalake_core::logstore::{
-    factories, str_is_truthy, IORuntime, ObjectStoreFactory, ObjectStoreRef,
+    object_store_factories, str_is_truthy, IORuntime, ObjectStoreFactory, ObjectStoreRef,
 };
 pub mod client;
 pub mod credential;
@@ -879,10 +879,9 @@ impl LogStoreFactory for UnityCatalogFactory {
 /// Register an [ObjectStoreFactory] for common UnityCatalogFactory [Url] schemes
 pub fn register_handlers(_additional_prefixes: Option<Url>) {
     let factory = Arc::new(UnityCatalogFactory::default());
-    let scheme = "uc";
-    let url = Url::parse(&format!("{scheme}://")).unwrap();
-    factories().insert(url.clone(), factory.clone());
-    logstores().insert(url.clone(), factory.clone());
+    let url = Url::parse(&format!("uc://")).unwrap();
+    object_store_factories().insert(url.clone(), factory.clone());
+    logstore_factories().insert(url.clone(), factory.clone());
 }
 
 #[async_trait::async_trait]

crates/core/Cargo.toml

@@ -12,7 +12,7 @@ repository.workspace = true
 rust-version.workspace = true
 
 [package.metadata.docs.rs]
-features = ["datafusion", "json", "unity-experimental"]
+features = ["datafusion", "json"]
 
 [dependencies]
 delta_kernel.workspace = true

crates/core/src/data_catalog/storage/mod.rs

@@ -13,7 +13,7 @@ use futures::TryStreamExt;
 use object_store::ObjectStore;
 
 use crate::errors::DeltaResult;
-use crate::logstore::{storage::*, StorageConfig};
+use crate::logstore::{store_for, StorageConfig};
 use crate::open_table_with_storage_options;
 use crate::table::builder::ensure_table_uri;
 

crates/core/src/delta_datafusion/planner.rs

@@ -6,7 +6,7 @@
 //! #[derive(Clone)]
 //! struct MergeMetricExtensionPlanner {}
 //!
-//! #[async_trait]
+//! #[macro@async_trait]
 //! impl ExtensionPlanner for MergeMetricExtensionPlanner {
 //!     async fn plan_extension(
 //!         &self,

crates/core/src/lib.rs

@@ -68,7 +68,6 @@
 // #![deny(missing_docs)]
 #![allow(rustdoc::invalid_html_tags)]
 #![allow(clippy::nonminimal_bool)]
-
 pub mod data_catalog;
 pub mod errors;
 pub mod kernel;
@@ -86,6 +85,7 @@ pub mod delta_datafusion;
 pub mod writer;
 
 use std::collections::HashMap;
+use std::sync::OnceLock;
 
 pub use self::data_catalog::{DataCatalog, DataCatalogError};
 pub use self::errors::*;
@@ -96,14 +96,14 @@ pub use self::table::config::TableProperty;
 pub use self::table::DeltaTable;
 pub use object_store::{path::Path, Error as ObjectStoreError, ObjectMeta, ObjectStore};
 pub use operations::DeltaOps;
-use std::sync::OnceLock;
+
+pub use protocol::checkpoints;
 
 // convenience exports for consumers to avoid aligning crate versions
 pub use arrow;
 #[cfg(feature = "datafusion")]
 pub use datafusion;
 pub use parquet;
-pub use protocol::checkpoints;
 
 /// Creates and loads a DeltaTable from the given path with current metadata.
 /// Infers the storage backend to use from the scheme in the given table path.