feat(io): add retry support for OpenDAL stores

rust/lance-io/src/object_store/dynamic_opendal.rs

@@ -252,12 +252,29 @@ impl OSObjectStore for DynamicOpenDalStore {
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;
-    use std::sync::Arc;
+    use std::sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    };
 
+    use async_trait::async_trait;
     use opendal::{Operator, services::Memory};
 
     use super::*;
+    use crate::object_store::providers::opendal::finish_opendal_operator;
     use crate::object_store::test_utils::StaticMockStorageOptionsProvider;
+    use crate::object_store::{StorageOptions, StorageOptionsProvider};
+
+    fn build_memory_store(config: HashMap<String, String>) -> Result<OpendalStore> {
+        let storage_options = StorageOptions(config);
+        let operator = Operator::new(Memory::default()).map_err(|e| {
+            lance_core::Error::invalid_input(format!("Failed to create memory operator: {e:?}"))
+        })?;
+        Ok(OpendalStore::new(finish_opendal_operator(
+            operator,
+            storage_options.client_max_retries(),
+        )))
+    }
 
     #[tokio::test]
     async fn test_dynamic_store_caches_by_normalized_config() {
@@ -272,14 +289,7 @@ mod tests {
             HashMap::new(),
             accessor,
             |options| Ok(options.clone()),
-            |_| {
-                let operator = Operator::new(Memory::default()).map_err(|e| {
-                    lance_core::Error::invalid_input(format!(
-                        "Failed to create memory operator: {e:?}"
-                    ))
-                })?;
-                Ok(OpendalStore::new(operator))
-            },
+            build_memory_store,
         );
 
         let first = store
@@ -294,6 +304,55 @@ mod tests {
         assert!(Arc::ptr_eq(&first, &second));
     }
 
+    #[derive(Debug)]
+    struct ChangingRetryConfigProvider {
+        fetch_count: AtomicUsize,
+    }
+
+    #[async_trait]
+    impl StorageOptionsProvider for ChangingRetryConfigProvider {
+        async fn fetch_storage_options(&self) -> Result<Option<HashMap<String, String>>> {
+            let fetch_count = self.fetch_count.fetch_add(1, Ordering::SeqCst);
+            Ok(Some(HashMap::from([
+                (
+                    "client_max_retries".to_string(),
+                    (fetch_count + 1).to_string(),
+                ),
+                ("expires_at_millis".to_string(), "0".to_string()),
+            ])))
+        }
+
+        fn provider_id(&self) -> String {
+            "ChangingRetryConfigProvider".to_string()
+        }
+    }
+
+    #[tokio::test]
+    async fn test_dynamic_store_rebuilds_when_retry_config_changes() {
+        let accessor = Arc::new(StorageOptionsAccessor::with_provider(Arc::new(
+            ChangingRetryConfigProvider {
+                fetch_count: AtomicUsize::new(0),
+            },
+        )));
+        let store = DynamicOpenDalStore::new(
+            "memory",
+            HashMap::new(),
+            accessor,
+            |options| Ok(options.clone()),
+            build_memory_store,
+        );
+
+        let first = store
+            .current_store()
+            .await
+            .expect("first store should build");
+        let second = store
+            .current_store()
+            .await
+            .expect("changed retry config should rebuild store");
+
+        assert!(!Arc::ptr_eq(&first, &second));
+    }
     #[test]
     fn test_merge_options_preserves_protected_base_keys() {
         let accessor = Arc::new(StorageOptionsAccessor::with_provider(Arc::new(
@@ -310,14 +369,7 @@ mod tests {
             ]),
             accessor,
             |options| Ok(options.clone()),
-            |_| {
-                let operator = Operator::new(Memory::default()).map_err(|e| {
-                    lance_core::Error::invalid_input(format!(
-                        "Failed to create memory operator: {e:?}"
-                    ))
-                })?;
-                Ok(OpendalStore::new(operator))
-            },
+            build_memory_store,
         )
         .with_protected_keys(["bucket", "root"]);
 

rust/lance-io/src/object_store/providers.rs

@@ -30,6 +30,17 @@ pub mod goosefs;
 pub mod huggingface;
 pub mod local;
 pub mod memory;
+#[cfg(any(
+    feature = "aws",
+    feature = "azure",
+    feature = "gcp",
+    feature = "goosefs",
+    feature = "huggingface",
+    feature = "oss",
+    feature = "tencent",
+    feature = "tos"
+))]
+pub(in crate::object_store) mod opendal;
 #[cfg(feature = "oss")]
 pub mod oss;
 pub mod shared_memory;

rust/lance-io/src/object_store/providers/aws.rs

@@ -40,6 +40,8 @@ use crate::object_store::{
 use lance_core::error::{Error, Result};
 use lance_core::utils::parse::str_is_truthy;
 
+use super::opendal::finish_opendal_operator;
+
 #[derive(Default, Debug)]
 pub struct AwsStoreProvider;
 
@@ -179,6 +181,7 @@ impl AwsStoreProvider {
 
         let operator = Operator::from_iter::<S3>(config_map)
             .map_err(|e| Error::invalid_input(format!("Failed to create S3 operator: {:?}", e)))?;
+        let operator = finish_opendal_operator(operator, storage_options.client_max_retries());
 
         Ok(Arc::new(OpendalStore::new(operator)) as Arc<dyn OSObjectStore>)
     }

rust/lance-io/src/object_store/providers/azure.rs

@@ -28,6 +28,8 @@ use crate::object_store::{
 use lance_core::error::{Error, Result};
 use lance_core::utils::parse::str_is_truthy;
 
+use super::opendal::finish_opendal_operator;
+
 #[derive(Default, Debug)]
 pub struct AzureBlobStoreProvider;
 
@@ -93,6 +95,7 @@ impl AzureBlobStoreProvider {
         // Start with all storage options as the config map
         // OpenDAL will handle environment variables through its default credentials chain
         let mut config_map = Self::normalize_opendal_azure_options(&storage_options.0);
+        let max_retries = storage_options.client_max_retries();
 
         match base_path.scheme() {
             "az" => {
@@ -108,9 +111,14 @@ impl AzureBlobStoreProvider {
                     config_map.insert("root".to_string(), format!("/{}", prefix));
                 }
 
-                Operator::from_iter::<Azblob>(config_map).map_err(|e| {
-                    Error::invalid_input(format!("Failed to create Azure Blob operator: {:?}", e))
-                })
+                Operator::from_iter::<Azblob>(config_map)
+                    .map_err(|e| {
+                        Error::invalid_input(format!(
+                            "Failed to create Azure Blob operator: {:?}",
+                            e
+                        ))
+                    })
+                    .map(|operator| finish_opendal_operator(operator, max_retries))
             }
             "abfss" => {
                 let filesystem = base_path.username();
@@ -136,12 +144,14 @@ impl AzureBlobStoreProvider {
                     config_map.insert("root".to_string(), format!("/{}", root_path));
                 }
 
-                Operator::from_iter::<Azdls>(config_map).map_err(|e| {
-                    Error::invalid_input(format!(
-                        "Failed to create Azure DFS (ADLS Gen2) operator: {:?}",
-                        e
-                    ))
-                })
+                Operator::from_iter::<Azdls>(config_map)
+                    .map_err(|e| {
+                        Error::invalid_input(format!(
+                            "Failed to create Azure DFS (ADLS Gen2) operator: {:?}",
+                            e
+                        ))
+                    })
+                    .map(|operator| finish_opendal_operator(operator, max_retries))
             }
             _ => Err(Error::invalid_input(format!(
                 "Unsupported Azure scheme: {}",

rust/lance-io/src/object_store/providers/gcp.rs

@@ -29,6 +29,9 @@ use crate::object_store::{
 };
 use lance_core::error::{Error, Result};
 use lance_core::utils::parse::str_is_truthy;
+
+use super::opendal::finish_opendal_operator;
+
 #[derive(Default, Debug)]
 pub struct GcsStoreProvider;
 
@@ -228,6 +231,7 @@ impl GcsStoreProvider {
 
         let operator = Operator::from_iter::<Gcs>(config_map)
             .map_err(|e| Error::invalid_input(format!("Failed to create GCS operator: {:?}", e)))?;
+        let operator = finish_opendal_operator(operator, storage_options.client_max_retries());
 
         Ok(Arc::new(OpendalStore::new(operator)) as Arc<dyn OSObjectStore>)
     }

rust/lance-io/src/object_store/providers/goosefs.rs

@@ -14,6 +14,8 @@ use crate::object_store::{
 };
 use lance_core::error::{Error, Result};
 
+use super::opendal::finish_opendal_operator;
+
 /// Default GooseFS Master gRPC port.
 const DEFAULT_GOOSEFS_PORT: u16 = 9200;
 
@@ -385,7 +387,8 @@ impl GooseFsStoreProvider {
 impl ObjectStoreProvider for GooseFsStoreProvider {
     async fn new_store(&self, base_path: Url, params: &ObjectStoreParams) -> Result<ObjectStore> {
         let block_size = params.block_size.unwrap_or(DEFAULT_CLOUD_BLOCK_SIZE);
-        let storage_options = StorageOptions(params.storage_options().cloned().unwrap_or_default());
+        let storage_options =
+            StorageOptions::new(params.storage_options().cloned().unwrap_or_default());
 
         Self::validate_storage_option_keys(&storage_options)?;
 
@@ -444,6 +447,7 @@ impl ObjectStoreProvider for GooseFsStoreProvider {
         let operator = Operator::from_iter::<GooseFs>(config_map).map_err(|e| {
             Error::invalid_input(format!("Failed to create GooseFS operator: {:?}", e))
         })?;
+        let operator = finish_opendal_operator(operator, storage_options.client_max_retries());
 
         // Wrap as object_store::ObjectStore via OpendalStore bridge
         let opendal_store = Arc::new(OpendalStore::new(operator));

rust/lance-io/src/object_store/providers/huggingface.rs

@@ -18,6 +18,8 @@ use crate::object_store::{
 };
 use lance_core::error::{Error, Result};
 
+use super::opendal::finish_opendal_operator;
+
 /// Hugging Face object store provider backed by OpenDAL.
 #[derive(Default, Debug)]
 pub struct HuggingfaceStoreProvider;
@@ -142,10 +144,20 @@ fn normalize_hf_config(options: &HashMap<String, String>) -> Result<HashMap<Stri
         config_map.insert("enable_resolve_cache".to_string(), enabled.clone());
     }
 
+    if let Some((_, max_retries)) = options
+        .iter()
+        .find(|(key, _)| key.eq_ignore_ascii_case("client_max_retries"))
+    {
+        config_map.insert("client_max_retries".to_string(), max_retries.clone());
+    }
+
     Ok(config_map)
 }
 
 fn build_hf_store(config_map: HashMap<String, String>) -> Result<OpendalStore> {
+    let storage_options = StorageOptions(config_map);
+    let max_retries = storage_options.client_max_retries();
+    let config_map = &storage_options.0;
     let repo_type = config_map
         .get("repo_type")
         .ok_or_else(|| Error::invalid_input("Huggingface repo_type is required"))?;
@@ -178,6 +190,7 @@ fn build_hf_store(config_map: HashMap<String, String>) -> Result<OpendalStore> {
     let operator = Operator::new(builder).map_err(|e| {
         Error::invalid_input(format!("Failed to create Huggingface operator: {:?}", e))
     })?;
+    let operator = finish_opendal_operator(operator, max_retries);
 
     Ok(OpendalStore::new(operator))
 }
@@ -190,7 +203,8 @@ impl ObjectStoreProvider for HuggingfaceStoreProvider {
         } = parse_hf_url(&base_path)?;
 
         let block_size = params.block_size.unwrap_or(DEFAULT_CLOUD_BLOCK_SIZE);
-        let storage_options = StorageOptions(params.storage_options().cloned().unwrap_or_default());
+        let storage_options =
+            StorageOptions::new(params.storage_options().cloned().unwrap_or_default());
         let download_retry_count = storage_options.download_retry_count();
 
         let mut base_options = build_hf_base_options(&repo_type, &repo_id, &storage_options);
@@ -414,6 +428,21 @@ mod tests {
         assert!(err.to_string().contains("Expected true or false"));
     }
 
+    #[test]
+    fn storage_option_preserves_client_max_retries() {
+        let config = normalize_hf_config(&build_hf_base_options(
+            "dataset",
+            "acme/repo",
+            &crate::object_store::StorageOptions(HashMap::from([(
+                "CLIENT_MAX_RETRIES".to_string(),
+                "5".to_string(),
+            )])),
+        ))
+        .unwrap();
+
+        assert_eq!(config.get("client_max_retries").unwrap(), "5");
+    }
+
     #[test]
     fn storage_option_download_mode_rejects_invalid_value() {
         let err = normalize_hf_config(&build_hf_base_options(