upgrade

Signed-off-by: tison <wander4096@gmail.com>

Author: tisonkun

integrations/object_store/Cargo.toml

@@ -25,7 +25,7 @@ homepage = "https://opendal.apache.org/"
 license = "Apache-2.0"
 repository = "https://github.com/apache/opendal"
 rust-version = "1.85"
-version = "0.55.0"
+version = "0.56.0"
 
 [features]
 send_wrapper = ["dep:send_wrapper"]

integrations/object_store/src/service/deleter.rs

@@ -18,8 +18,8 @@
 use std::sync::Arc;
 
 use futures::stream::{self, StreamExt};
-use object_store::ObjectStore;
 use object_store::path::Path as ObjectStorePath;
+use object_store::{ObjectStore, ObjectStoreExt};
 use opendal::raw::oio::BatchDeleteResult;
 use opendal::raw::*;
 use opendal::*;
@@ -44,7 +44,7 @@ impl oio::BatchDelete for ObjectStoreDeleter {
 
     async fn delete_batch(&self, paths: Vec<(String, OpDelete)>) -> Result<BatchDeleteResult> {
         // convert paths to stream, then use [`ObjectStore::delete_stream`] to delete them in batch
-        let stream = stream::iter(paths.iter())
+        let stream = stream::iter(paths.clone())
             .map(|(path, _)| Ok::<_, object_store::Error>(ObjectStorePath::from(path.as_str())))
             .boxed();
         let results = self.store.delete_stream(stream).collect::<Vec<_>>().await;

integrations/object_store/src/service/mod.rs

@@ -166,7 +166,7 @@ mod tests {
         let builder = ObjectStoreBuilder::new(store);
 
         let backend = builder.build().expect("build should succeed");
-        assert!(backend.info().scheme() == OBJECT_STORE_SCHEME);
+        assert_eq!(backend.info().scheme(), OBJECT_STORE_SCHEME);
     }
 
     #[tokio::test]

integrations/object_store/src/store.rs

@@ -29,7 +29,6 @@ use futures::TryStreamExt;
 use futures::stream::BoxStream;
 use mea::mutex::Mutex;
 use mea::oneshot;
-use object_store::ListResult;
 use object_store::MultipartUpload;
 use object_store::ObjectMeta;
 use object_store::ObjectStore;
@@ -41,6 +40,7 @@ use object_store::path::Path;
 use object_store::{GetOptions, UploadPart};
 use object_store::{GetRange, GetResultPayload};
 use object_store::{GetResult, PutMode};
+use object_store::{ListResult, RenameOptions};
 use opendal::Buffer;
 use opendal::Writer;
 use opendal::options::CopyOptions;
@@ -227,23 +227,6 @@ impl ObjectStore for OpendalStore {
         Ok(PutResult { e_tag, version })
     }
 
-    async fn put_multipart(
-        &self,
-        location: &Path,
-    ) -> object_store::Result<Box<dyn MultipartUpload>> {
-        let decoded_location = percent_decode_path(location.as_ref());
-        let writer = self
-            .inner
-            .writer_with(&decoded_location)
-            .concurrent(8)
-            .into_send()
-            .await
-            .map_err(|err| format_object_store_error(err, location.as_ref()))?;
-        let upload = OpendalMultipartUpload::new(writer, location.clone());
-
-        Ok(Box::new(upload))
-    }
-
     async fn put_multipart_opts(
         &self,
         location: &Path,
@@ -430,15 +413,27 @@ impl ObjectStore for OpendalStore {
         })
     }
 
-    async fn delete(&self, location: &Path) -> object_store::Result<()> {
-        let decoded_location = percent_decode_path(location.as_ref());
-        self.inner
-            .delete(&decoded_location)
-            .into_send()
-            .await
-            .map_err(|err| format_object_store_error(err, location.as_ref()))?;
-
-        Ok(())
+    fn delete_stream(
+        &self,
+        locations: BoxStream<'static, object_store::Result<Path>>,
+    ) -> BoxStream<'static, object_store::Result<Path>> {
+        // TODO: use batch delete to optimize performance
+        let client = self.inner.clone();
+        locations
+            .then(move |location| {
+                let client = client.clone();
+                async move {
+                    let location = location?;
+                    let decoded_location = percent_decode_path(location.as_ref());
+                    client
+                        .delete(&decoded_location)
+                        .into_send()
+                        .await
+                        .map_err(|err| format_object_store_error(err, location.as_ref()))?;
+                    Ok(location)
+                }
+            })
+            .boxed()
     }
 
     fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, object_store::Result<ObjectMeta>> {
@@ -576,11 +571,23 @@ impl ObjectStore for OpendalStore {
         })
     }
 
-    async fn copy(&self, from: &Path, to: &Path) -> object_store::Result<()> {
-        self.copy_request(from, to, false).await
+    async fn copy_opts(
+        &self,
+        from: &Path,
+        to: &Path,
+        options: object_store::CopyOptions,
+    ) -> object_store::Result<()> {
+        let if_not_exists = matches!(options.mode, object_store::CopyMode::Create);
+        self.copy_request(from, to, if_not_exists).await
     }
 
-    async fn rename(&self, from: &Path, to: &Path) -> object_store::Result<()> {
+    async fn rename_opts(
+        &self,
+        from: &Path,
+        to: &Path,
+        // TODO: if we need to support rename options in the future
+        _options: RenameOptions,
+    ) -> object_store::Result<()> {
         self.inner
             .rename(
                 &percent_decode_path(from.as_ref()),
@@ -592,10 +599,6 @@ impl ObjectStore for OpendalStore {
 
         Ok(())
     }
-
-    async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> object_store::Result<()> {
-        self.copy_request(from, to, true).await
-    }
 }
 
 /// `MultipartUpload`'s impl based on `Writer` in opendal
@@ -691,7 +694,7 @@ impl Debug for OpendalMultipartUpload {
 mod tests {
     use bytes::Bytes;
     use object_store::path::Path;
-    use object_store::{ObjectStore, WriteMultipart};
+    use object_store::{ObjectStore, ObjectStoreExt, WriteMultipart};
     use opendal::services;
     use rand::prelude::*;
     use std::sync::Arc;

integrations/object_store/tests/behavior/delete.rs

@@ -18,7 +18,7 @@
 use crate::utils::build_trail;
 use bytes::Bytes;
 use libtest_mimic::Trial;
-use object_store::ObjectStore;
+use object_store::ObjectStoreExt;
 use object_store::path::Path;
 use object_store_opendal::OpendalStore;
 

integrations/object_store/tests/behavior/get.rs

@@ -19,7 +19,7 @@ use crate::utils::{build_trail, new_file_path};
 use anyhow::Result;
 use bytes::Bytes;
 use libtest_mimic::Trial;
-use object_store::{GetOptions, GetRange, ObjectStore};
+use object_store::{GetOptions, GetRange, ObjectStore, ObjectStoreExt};
 use object_store_opendal::OpendalStore;
 
 pub fn tests(store: &OpendalStore, tests: &mut Vec<Trial>) {

integrations/object_store/tests/behavior/put.rs

@@ -19,7 +19,7 @@ use crate::utils::{build_trail, new_file_path};
 use anyhow::Result;
 use bytes::Bytes;
 use libtest_mimic::Trial;
-use object_store::{ObjectStore, PutMode, UpdateVersion};
+use object_store::{ObjectStore, ObjectStoreExt, PutMode, UpdateVersion};
 use object_store_opendal::OpendalStore;
 
 pub fn tests(store: &OpendalStore, tests: &mut Vec<Trial>) {