chore: bump opendal to 0.54

LDeakin · LDeakin · commit 37848eb46e29 · 2025-07-17T22:53:37.000+10:00
diff --git a/Cargo.toml b/Cargo.toml
@@ -67,7 +67,7 @@ version = "0.4.2"
 path = "zarrs_object_store"
 
 [workspace.dependencies.zarrs_opendal]
-version = "0.7.2"
+version = "0.8.0"
 path = "zarrs_opendal"
 
 [workspace.dependencies.zarrs_zip]
@@ -78,7 +78,7 @@ path = "zarrs_zip"
 version = "0.12"
 
 [workspace.dependencies.opendal]
-version = "0.53.0"
+version = "0.54.0"
 
 [workspace.dependencies.zip]
 version = "3.0.0"
diff --git a/zarrs_opendal/CHANGELOG.md b/zarrs_opendal/CHANGELOG.md
@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## Changed
+- **Breaking**: Bump `opendal` to 0.54
+
+## Removed
+- **Breaking**: Remove `OpendalStore`
+  - Use `AsyncOpendalStore` with the `AsyncToSyncStorageAdapter` instead
+
 ## [0.7.2] - 2025-05-16
 
 ## Changed
diff --git a/zarrs_opendal/Cargo.toml b/zarrs_opendal/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "zarrs_opendal"
-version = "0.7.2"
+version = "0.8.0"
 authors = ["Lachlan Deakin <ljdgit@gmail.com>"]
 edition = "2021"
 rust-version = "1.80"
@@ -18,7 +18,7 @@ workspace = true
 [dependencies]
 async-trait = "0.1.74"
 futures = "0.3.29"
-opendal = { version = ">=0.51,<0.54", default-features = false }
+opendal = { version = ">=0.54,<0.55", default-features = false, features=["blocking"] }
 zarrs_storage = { workspace = true, features = ["async"] }
 
 [dev-dependencies]
diff --git a/zarrs_opendal/doc/version_compatibility_matrix.md b/zarrs_opendal/doc/version_compatibility_matrix.md
@@ -1,5 +1,6 @@
 | [zarrs_opendal] | [opendal] | [zarrs] ([zarrs_storage]) |
 | --------------- | --------- | ------------------------- |
+| 0.8             | 0.54-0.54 | 0.18+ (0.3.x)             |
 | 0.7             | 0.51-0.53 | 0.18+ (0.3.x)             |
 | 0.6             | 0.51-0.52 | 0.18+ (0.3.x)             |
 | 0.5             | 0.51-0.51 | 0.18+ (0.3.x)             |
diff --git a/zarrs_opendal/src/async.rs b/zarrs_opendal/src/async.rs
@@ -1,3 +1,5 @@
+use futures::stream::TryStreamExt;
+use futures::{future, StreamExt};
 use opendal::Operator;
 
 use zarrs_storage::{
@@ -12,13 +14,24 @@ use crate::{handle_result, handle_result_notfound};
 /// An asynchronous store backed by an [`opendal::Operator`].
 pub struct AsyncOpendalStore {
     operator: Operator,
+    concurrent_stat_requests: usize,
 }
 
 impl AsyncOpendalStore {
     /// Create a new [`AsyncOpendalStore`].
     #[must_use]
     pub fn new(operator: Operator) -> Self {
-        Self { operator }
+        Self {
+            operator,
+            concurrent_stat_requests: 32,
+        }
+    }
+
+    /// Set the number of concurrent stat requests.
+    #[must_use]
+    pub fn with_concurrent_stat_requests(mut self, concurrent_stat_requests: usize) -> Self {
+        self.concurrent_stat_requests = concurrent_stat_requests;
+        self
     }
 }
 
@@ -157,43 +170,24 @@ impl AsyncListableStorageTraits for AsyncOpendalStore {
     }
 
     async fn size_prefix(&self, prefix: &StorePrefix) -> Result<u64, StorageError> {
-        let Some(files) = handle_result_notfound(
+        let lister = handle_result(
             self.operator
-                .list_with(prefix.as_str())
+                .lister_with(prefix.as_str())
                 .recursive(true)
                 .await,
-        )?
-        else {
-            return Ok(0);
-        };
-
-        if self
-            .operator
-            .info()
-            .full_capability()
-            .list_has_content_length
-        {
-            let size = files
-                .into_iter()
-                .filter_map(|entry| {
-                    if entry.metadata().is_file() {
-                        Some(entry.metadata().content_length())
-                    } else {
-                        None
-                    }
-                })
-                .sum::<u64>();
-            Ok(size)
-        } else {
-            // TODO: concurrent
-            let mut size = 0;
-            for entry in files {
-                let meta = handle_result(self.operator.stat(entry.path()).await)?;
-                if meta.is_file() {
-                    size += meta.content_length();
-                }
-            }
-            Ok(size)
-        }
+        )?;
+
+        let total_size = lister
+            .try_filter(|entry| future::ready(entry.metadata().mode().is_file()))
+            .map(move |entry_result| async move {
+                let entry = handle_result(entry_result)?;
+                let metadata = handle_result(self.operator.stat(entry.path()).await)?;
+                Ok::<_, StorageError>(metadata.content_length())
+            })
+            .buffer_unordered(self.concurrent_stat_requests)
+            .try_fold(0, |acc, size| future::ready(Ok(acc + size)))
+            .await?;
+
+        Ok(total_size)
     }
 }
diff --git a/zarrs_opendal/src/lib.rs b/zarrs_opendal/src/lib.rs
@@ -31,10 +31,8 @@
 //! - the MIT license [LICENSE-MIT](https://docs.rs/crate/zarrs_opendal/latest/source/LICENCE-MIT) or <http://opensource.org/licenses/MIT>, at your option.
 
 mod r#async;
-mod sync;
 
 pub use r#async::AsyncOpendalStore;
-pub use sync::OpendalStore;
 
 pub use opendal;
 
diff --git a/zarrs_opendal/src/sync.rs b/zarrs_opendal/src/sync.rs
diff --git a/zarrs_opendal/tests/sync.rs b/zarrs_opendal/tests/sync.rs
