fix: ListingSchemaProvider not forwarding custom storage_options

crates/azure/Cargo.toml

@@ -12,15 +12,17 @@ repository.workspace = true
 rust-version.workspace = true
 
 [dependencies]
-deltalake-core = { version = "0.21.0", path = "../core" }
+deltalake-core = { version = "0.21.0", path = "../core", features = [
+    "datafusion",
+] }
 lazy_static = "1"
 
 # workspace depenndecies
 async-trait = { workspace = true }
 bytes = { workspace = true }
 futures = { workspace = true }
 tracing = { workspace = true }
-object_store = { workspace = true, features = ["azure"]}
+object_store = { workspace = true, features = ["azure"] }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 regex = { workspace = true }

crates/azure/tests/integration.rs

@@ -1,6 +1,10 @@
 #![cfg(feature = "integration_test")]
 
+use std::assert_eq;
+use std::collections::HashMap;
+
 use bytes::Bytes;
+use deltalake_core::data_catalog::storage::ListingSchemaProvider;
 use deltalake_core::DeltaTableBuilder;
 use deltalake_test::read::read_table_paths;
 use deltalake_test::{test_concurrent_writes, test_read_tables, IntegrationContext, TestResult};
@@ -89,3 +93,42 @@ async fn read_write_test_onelake(context: &IntegrationContext, path: &Path) -> T
 
     Ok(())
 }
+
+#[test]
+#[serial]
+fn list_delta_tables_using_listing_provider_with_missing_account_name() -> TestResult {
+    let context = IntegrationContext::new(Box::new(MsftIntegration::default()))?;
+    // Removing the the envs set by the `IntegrationContext (az_cli::prepare_env())` to illustrate the issue if e.g. account_name is not set from custom `storage_options`, but still preserving the use of the `IntegrationContext`
+    std::env::remove_var("AZURE_STORAGE_USE_EMULATOR");
+    std::env::remove_var("AZURE_STORAGE_ACCOUNT_NAME");
+    std::env::remove_var("AZURE_STORAGE_TOKEN");
+    std::env::remove_var("AZURE_STORAGE_ACCOUNT_KEY");
+
+    let storage_options = HashMap::<String, String>::new();
+    if let Err(read_error) =
+        ListingSchemaProvider::try_new(&context.root_uri(), Some(storage_options))
+    {
+        assert_eq!(read_error.to_string(), "Failed to read delta log object: Generic MicrosoftAzure error: Account must be specified".to_string());
+    };
+    Ok(())
+}
+
+#[tokio::test]
+#[serial]
+async fn list_delta_tables_using_listing_provider_with_account_name() -> TestResult {
+    let context = IntegrationContext::new(Box::new(MsftIntegration::default()))?;
+    // Removing the the envs set by the `IntegrationContext (az_cli::prepare_env())` to illustrate the issue if e.g. account_name is not set from custom `storage_options`, but still preserving the use of the `IntegrationContext`
+    std::env::remove_var("AZURE_STORAGE_USE_EMULATOR");
+    std::env::remove_var("AZURE_STORAGE_ACCOUNT_NAME");
+    std::env::remove_var("AZURE_STORAGE_TOKEN");
+    std::env::remove_var("AZURE_STORAGE_ACCOUNT_KEY");
+
+    let mut storage_options = HashMap::<String, String>::new();
+    storage_options.insert("account_name".to_string(), "test_account".to_string());
+    let schema = ListingSchemaProvider::try_new(&context.root_uri(), Some(storage_options));
+    assert!(
+        schema.is_ok(),
+        "Capable of reading the storage options. Fails if e.g. `account_name` is missing"
+    );
+    Ok(())
+}

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

@@ -47,9 +47,9 @@ impl ListingSchemaProvider {
         storage_options: Option<HashMap<String, String>>,
     ) -> DeltaResult<Self> {
         let uri = ensure_table_uri(root_uri)?;
-        let storage_options = storage_options.unwrap_or_default().into();
+        let storage_options: StorageOptions = storage_options.unwrap_or_default().into();
         // We already parsed the url, so unwrapping is safe.
-        let store = store_for(&uri)?;
+        let store = store_for(&uri, &storage_options)?;
         Ok(Self {
             authority: uri.to_string(),
             store,

crates/core/src/storage/mod.rs

@@ -395,10 +395,10 @@ pub fn factories() -> FactoryRegistry {
 }
 
 /// Simpler access pattern for the [FactoryRegistry] to get a single store
-pub fn store_for(url: &Url) -> DeltaResult<ObjectStoreRef> {
+pub fn store_for(url: &Url, storage_options: &StorageOptions) -> DeltaResult<ObjectStoreRef> {
     let scheme = Url::parse(&format!("{}://", url.scheme())).unwrap();
     if let Some(factory) = factories().get(&scheme) {
-        let (store, _prefix) = factory.parse_url_opts(url, &StorageOptions::default())?;
+        let (store, _prefix) = factory.parse_url_opts(url, storage_options)?;
         Ok(store)
     } else {
         Err(DeltaTableError::InvalidTableLocation(url.clone().into()))

crates/core/tests/fs_common/mod.rs

@@ -5,7 +5,7 @@ use deltalake_core::kernel::{
 use deltalake_core::operations::create::CreateBuilder;
 use deltalake_core::operations::transaction::CommitBuilder;
 use deltalake_core::protocol::{DeltaOperation, SaveMode};
-use deltalake_core::storage::{GetResult, ObjectStoreResult};
+use deltalake_core::storage::{GetResult, ObjectStoreResult, StorageOptions};
 use deltalake_core::DeltaTable;
 use object_store::path::Path as StorePath;
 use object_store::{
@@ -152,7 +152,7 @@ impl SlowStore {
         _options: impl Into<deltalake_core::storage::StorageOptions> + Clone,
     ) -> deltalake_core::DeltaResult<Self> {
         Ok(Self {
-            inner: deltalake_core::storage::store_for(&location)?,
+            inner: deltalake_core::storage::store_for(&location, &StorageOptions::default())?,
         })
     }
 }