-
Notifications
You must be signed in to change notification settings - Fork 708
chore(deps): update object_store requirement from 0.12.3 to 0.13.0 in /integrations/object_store #7118
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
chore(deps): update object_store requirement from 0.12.3 to 0.13.0 in /integrations/object_store #7118
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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( | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is now internally implemented by |
||
| &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<()> { | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Replaced with |
||
| 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() | ||
|
Comment on lines
+422
to
+436
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
fn delete_stream(
&self,
locations: BoxStream<'static, Result<Path>>,
) -> BoxStream<'static, Result<Path>> {
let client = Arc::clone(&self.client);
locations
.try_chunks(1_000)
.map(move |locations| {
let client = Arc::clone(&client);
async move {
// Early return the error. We ignore the paths that have already been
// collected into the chunk.
let locations = locations.map_err(|e| e.1)?;
client
.bulk_delete_request(locations)
.await
.map(futures::stream::iter)
}
})
.buffered(20)
.try_flatten()
.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, | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Seems OpenDAL doesn't support different rename mode now. |
||
| ) -> 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 | ||
| } | ||
|
Comment on lines
-596
to
-598
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Moved to |
||
| } | ||
|
|
||
| /// `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; | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
FYI @alamb @tustvold in case you'd like to see how other lib is integrated with object_store 0.13.
And perhaps there is more best pratice to do what gets done here.