You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2025/01/13 02:19:56 UTC
(opendal) branch main updated: feat(integrations/object_store): implement put_opts and get_opts (#5513)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 03b78f10e feat(integrations/object_store): implement put_opts and get_opts (#5513)
03b78f10e is described below
commit 03b78f10e5515f09cbef5150621884ba9461c74b
Author: meteorgan <me...@gmail.com>
AuthorDate: Mon Jan 13 10:19:50 2025 +0800
feat(integrations/object_store): implement put_opts and get_opts (#5513)
---
integrations/object_store/src/store.rs | 219 ++++++++++--------
integrations/object_store/src/utils.rs | 4 +
integrations/object_store/tests/behavior/get.rs | 282 +++++++++++++++++++++++-
integrations/object_store/tests/behavior/put.rs | 110 ++++++++-
4 files changed, 523 insertions(+), 92 deletions(-)
diff --git a/integrations/object_store/src/store.rs b/integrations/object_store/src/store.rs
index aaef7caa7..b32ec8f96 100644
--- a/integrations/object_store/src/store.rs
+++ b/integrations/object_store/src/store.rs
@@ -17,19 +17,16 @@
use std::fmt::{self, Debug, Display, Formatter};
use std::future::IntoFuture;
-use std::ops::Range;
+use std::io;
use std::sync::Arc;
use crate::utils::*;
use async_trait::async_trait;
-use bytes::Bytes;
use futures::stream::BoxStream;
use futures::FutureExt;
use futures::StreamExt;
use futures::TryStreamExt;
use object_store::path::Path;
-use object_store::GetResult;
-use object_store::GetResultPayload;
use object_store::ListResult;
use object_store::MultipartUpload;
use object_store::ObjectMeta;
@@ -39,6 +36,8 @@ use object_store::PutOptions;
use object_store::PutPayload;
use object_store::PutResult;
use object_store::{GetOptions, UploadPart};
+use object_store::{GetRange, GetResultPayload};
+use object_store::{GetResult, PutMode};
use opendal::Buffer;
use opendal::Writer;
use opendal::{Operator, OperatorInfo};
@@ -104,6 +103,11 @@ impl OpendalStore {
inner: op,
}
}
+
+ /// Get the Operator info.
+ pub fn info(&self) -> &OperatorInfo {
+ self.info.as_ref()
+ }
}
impl Debug for OpendalStore {
@@ -138,29 +142,47 @@ impl From<Operator> for OpendalStore {
#[async_trait]
impl ObjectStore for OpendalStore {
- async fn put(&self, location: &Path, bytes: PutPayload) -> object_store::Result<PutResult> {
- self.inner
- .write(location.as_ref(), Buffer::from_iter(bytes.into_iter()))
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, location.as_ref()))?;
- Ok(PutResult {
- e_tag: None,
- version: None,
- })
- }
-
async fn put_opts(
&self,
- _location: &Path,
- _bytes: PutPayload,
- _opts: PutOptions,
+ location: &Path,
+ bytes: PutPayload,
+ opts: PutOptions,
) -> object_store::Result<PutResult> {
- Err(object_store::Error::NotSupported {
- source: Box::new(opendal::Error::new(
- opendal::ErrorKind::Unsupported,
- "put_opts is not implemented so far",
- )),
+ let mut future_write = self
+ .inner
+ .write_with(location.as_ref(), Buffer::from_iter(bytes.into_iter()));
+ let opts_mode = opts.mode.clone();
+ match opts.mode {
+ PutMode::Overwrite => {}
+ PutMode::Create => {
+ future_write = future_write.if_not_exists(true);
+ }
+ PutMode::Update(update_version) => {
+ let Some(etag) = update_version.e_tag else {
+ Err(object_store::Error::NotSupported {
+ source: Box::new(opendal::Error::new(
+ opendal::ErrorKind::Unsupported,
+ "etag is required for conditional put",
+ )),
+ })?
+ };
+ future_write = future_write.if_match(etag.as_str());
+ }
+ }
+ future_write.into_send().await.map_err(|err| {
+ match format_object_store_error(err, location.as_ref()) {
+ object_store::Error::Precondition { path, source }
+ if opts_mode == PutMode::Create =>
+ {
+ object_store::Error::AlreadyExists { path, source }
+ }
+ e => e,
+ }
+ })?;
+
+ Ok(PutResult {
+ e_tag: None,
+ version: None,
})
}
@@ -193,13 +215,32 @@ impl ObjectStore for OpendalStore {
})
}
- async fn get(&self, location: &Path) -> object_store::Result<GetResult> {
- let meta = self
- .inner
- .stat(location.as_ref())
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, location.as_ref()))?;
+ async fn get_opts(
+ &self,
+ location: &Path,
+ options: GetOptions,
+ ) -> object_store::Result<GetResult> {
+ let meta = {
+ let mut s = self.inner.stat_with(location.as_ref());
+ if let Some(version) = &options.version {
+ s = s.version(version.as_str())
+ }
+ if let Some(if_match) = &options.if_match {
+ s = s.if_match(if_match.as_str());
+ }
+ if let Some(if_none_match) = &options.if_none_match {
+ s = s.if_none_match(if_none_match.as_str());
+ }
+ if let Some(if_modified_since) = options.if_modified_since {
+ s = s.if_modified_since(if_modified_since);
+ }
+ if let Some(if_unmodified_since) = options.if_unmodified_since {
+ s = s.if_unmodified_since(if_unmodified_since);
+ }
+ s.into_send()
+ .await
+ .map_err(|err| format_object_store_error(err, location.as_ref()))?
+ };
let meta = ObjectMeta {
location: location.clone(),
@@ -209,78 +250,76 @@ impl ObjectStore for OpendalStore {
version: meta.version().map(|x| x.to_string()),
};
- let r = self
- .inner
- .reader(location.as_ref())
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, location.as_ref()))?;
+ if options.head {
+ return Ok(GetResult {
+ payload: GetResultPayload::Stream(Box::pin(futures::stream::empty())),
+ range: 0..0,
+ meta,
+ attributes: Default::default(),
+ });
+ }
- let stream = r
- .into_bytes_stream(0..meta.size as u64)
+ let reader = {
+ let mut r = self.inner.reader_with(location.as_ref());
+ if let Some(version) = options.version {
+ r = r.version(version.as_str());
+ }
+ if let Some(if_match) = options.if_match {
+ r = r.if_match(if_match.as_str());
+ }
+ if let Some(if_none_match) = options.if_none_match {
+ r = r.if_none_match(if_none_match.as_str());
+ }
+ if let Some(if_modified_since) = options.if_modified_since {
+ r = r.if_modified_since(if_modified_since);
+ }
+ if let Some(if_unmodified_since) = options.if_unmodified_since {
+ r = r.if_unmodified_since(if_unmodified_since);
+ }
+ r.into_send()
+ .await
+ .map_err(|err| format_object_store_error(err, location.as_ref()))?
+ };
+
+ let read_range = match options.range {
+ Some(GetRange::Bounded(r)) => {
+ if r.start >= r.end || r.start >= meta.size {
+ 0..0
+ } else {
+ let end = r.end.min(meta.size);
+ r.start..end
+ }
+ }
+ Some(GetRange::Offset(r)) => {
+ if r < meta.size {
+ r..meta.size
+ } else {
+ 0..0
+ }
+ }
+ Some(GetRange::Suffix(r)) if r < meta.size => (meta.size - r)..meta.size,
+ _ => 0..meta.size,
+ };
+
+ let stream = reader
+ .into_bytes_stream(read_range.start as u64..read_range.end as u64)
.into_send()
.await
- .map_err(|err| object_store::Error::Generic {
- store: "IoError",
- source: Box::new(err),
- })?
+ .map_err(|err| format_object_store_error(err, location.as_ref()))?
.into_send()
- .map_err(|err| object_store::Error::Generic {
+ .map_err(|err: io::Error| object_store::Error::Generic {
store: "IoError",
source: Box::new(err),
});
Ok(GetResult {
payload: GetResultPayload::Stream(Box::pin(stream)),
- range: 0..meta.size,
+ range: read_range.start..read_range.end,
meta,
attributes: Default::default(),
})
}
- async fn get_opts(
- &self,
- _location: &Path,
- _options: GetOptions,
- ) -> object_store::Result<GetResult> {
- Err(object_store::Error::NotSupported {
- source: Box::new(opendal::Error::new(
- opendal::ErrorKind::Unsupported,
- "get_opts is not implemented so far",
- )),
- })
- }
-
- async fn get_range(&self, location: &Path, range: Range<usize>) -> object_store::Result<Bytes> {
- let bs = self
- .inner
- .read_with(location.as_ref())
- .range(range.start as u64..range.end as u64)
- .into_future()
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, location.as_ref()))?;
-
- Ok(bs.to_bytes())
- }
-
- async fn head(&self, location: &Path) -> object_store::Result<ObjectMeta> {
- let meta = self
- .inner
- .stat(location.as_ref())
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, location.as_ref()))?;
-
- Ok(ObjectMeta {
- location: location.clone(),
- last_modified: meta.last_modified().unwrap_or_default(),
- size: meta.content_length() as usize,
- e_tag: meta.etag().map(|x| x.to_string()),
- version: meta.version().map(|x| x.to_string()),
- })
- }
-
async fn delete(&self, location: &Path) -> object_store::Result<()> {
self.inner
.delete(location.as_ref())
@@ -534,12 +573,12 @@ impl Debug for OpendalMultipartUpload {
#[cfg(test)]
mod tests {
- use std::sync::Arc;
-
+ use bytes::Bytes;
use object_store::path::Path;
use object_store::{ObjectStore, WriteMultipart};
use opendal::services;
use rand::prelude::*;
+ use std::sync::Arc;
use super::*;
diff --git a/integrations/object_store/src/utils.rs b/integrations/object_store/src/utils.rs
index 63357402b..99a0578d1 100644
--- a/integrations/object_store/src/utils.rs
+++ b/integrations/object_store/src/utils.rs
@@ -42,6 +42,10 @@ pub fn format_object_store_error(err: opendal::Error, path: &str) -> object_stor
path: path.to_string(),
source: Box::new(err),
},
+ ErrorKind::ConditionNotMatch => object_store::Error::Precondition {
+ path: path.to_string(),
+ source: Box::new(err),
+ },
kind => object_store::Error::Generic {
store: kind.into_static(),
source: Box::new(err),
diff --git a/integrations/object_store/tests/behavior/get.rs b/integrations/object_store/tests/behavior/get.rs
index 3b241864f..6e7c549da 100644
--- a/integrations/object_store/tests/behavior/get.rs
+++ b/integrations/object_store/tests/behavior/get.rs
@@ -19,13 +19,48 @@ use crate::utils::{build_trail, new_file_path};
use anyhow::Result;
use bytes::Bytes;
use libtest_mimic::Trial;
-use object_store::ObjectStore;
+use object_store::{GetOptions, GetRange, ObjectStore};
use object_store_opendal::OpendalStore;
pub fn tests(store: &OpendalStore, tests: &mut Vec<Trial>) {
tests.push(build_trail("test_basic_get", store, test_basic_get));
tests.push(build_trail("test_head", store, test_head));
tests.push(build_trail("test_get_range", store, test_get_range));
+ tests.push(build_trail(
+ "test_get_opts_with_range",
+ store,
+ test_get_opts_with_range,
+ ));
+ tests.push(build_trail(
+ "test_get_opts_with_invalid_range",
+ store,
+ test_get_opts_with_invalid_range,
+ ));
+ tests.push(build_trail(
+ "test_get_opts_with_version",
+ store,
+ test_get_opts_with_version,
+ ));
+ tests.push(build_trail(
+ "test_get_ops_with_if_match",
+ store,
+ test_get_ops_with_if_match,
+ ));
+ tests.push(build_trail(
+ "test_get_ops_with_if_none_match",
+ store,
+ test_get_ops_with_if_none_match,
+ ));
+ tests.push(build_trail(
+ "test_get_opts_with_if_modified_since",
+ store,
+ test_get_opts_with_if_modified_since,
+ ));
+ tests.push(build_trail(
+ "test_get_opts_with_if_unmodified_since",
+ store,
+ test_get_opts_with_if_unmodified_since,
+ ));
}
pub async fn test_basic_get(store: OpendalStore) -> Result<()> {
@@ -78,3 +113,248 @@ pub async fn test_get_range(store: OpendalStore) -> Result<()> {
store.delete(&location).await?;
Ok(())
}
+
+pub async fn test_get_opts_with_range(store: OpendalStore) -> Result<()> {
+ let location = new_file_path("data").into();
+ let value = Bytes::from_static(b"Hello, world!");
+
+ store.put(&location, value.clone().into()).await?;
+
+ let opts = GetOptions {
+ range: Some((0..5).into()),
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await?;
+ let data = ret.bytes().await?;
+ assert_eq!(Bytes::from_static(b"Hello"), data);
+
+ let opts = GetOptions {
+ range: Some(GetRange::Offset(7)),
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await?;
+ let data = ret.bytes().await?;
+ assert_eq!(Bytes::from_static(b"world!"), data);
+
+ let opts = GetOptions {
+ range: Some(GetRange::Suffix(6)),
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await?;
+ let data = ret.bytes().await?;
+ assert_eq!(Bytes::from_static(b"world!"), data);
+
+ store.delete(&location).await?;
+ Ok(())
+}
+
+pub async fn test_get_opts_with_invalid_range(store: OpendalStore) -> Result<()> {
+ let location = new_file_path("data").into();
+ let value = Bytes::from_static(b"Hello, world!");
+
+ store.put(&location, value.clone().into()).await?;
+
+ // the end of the range is greater than the size of the object
+ let opts = GetOptions {
+ range: Some((0..100).into()),
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await?;
+ assert_eq!(ret.range, 0..value.len());
+ assert_eq!(ret.bytes().await?, value);
+
+ // the offset of the range is greater than the size of the object
+ let opts = GetOptions {
+ range: Some(GetRange::Offset(100)),
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await?;
+ assert_eq!(ret.range, 0..0);
+
+ // the suffix of the range is greater than the size of the object
+ let opts = GetOptions {
+ range: Some(GetRange::Suffix(100)),
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await?;
+ assert_eq!(ret.range, 0..value.len());
+ assert_eq!(ret.bytes().await?, value);
+
+ store.delete(&location).await?;
+ Ok(())
+}
+
+pub async fn test_get_opts_with_version(store: OpendalStore) -> Result<()> {
+ if !store.info().full_capability().read_with_version {
+ return Ok(());
+ }
+
+ let location = new_file_path("data").into();
+ let value = Bytes::from_static(b"Hello, world!");
+ store.put(&location, value.clone().into()).await?;
+ let meta = store.head(&location).await?;
+
+ let opts = GetOptions {
+ version: meta.version,
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await?;
+ let data = ret.bytes().await?;
+ assert_eq!(value, data);
+
+ let another_location = new_file_path("data").into();
+ store
+ .put(
+ &another_location,
+ Bytes::from_static(b"Hello, world!").into(),
+ )
+ .await?;
+ let another_version = store.head(&another_location).await?.version;
+
+ let opts = GetOptions {
+ version: another_version,
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await;
+ assert!(ret.is_err());
+ assert!(matches!(
+ ret.err().unwrap(),
+ object_store::Error::NotFound { .. }
+ ));
+
+ store.delete(&location).await?;
+ store.delete(&another_location).await?;
+ Ok(())
+}
+
+async fn test_get_ops_with_if_match(store: OpendalStore) -> Result<()> {
+ if !store.info().full_capability().read_with_if_match {
+ return Ok(());
+ }
+
+ let location = new_file_path("data").into();
+ let value = Bytes::from_static(b"Hello, world!");
+ store.put(&location, value.clone().into()).await?;
+ let meta = store.head(&location).await?;
+
+ let opts = GetOptions {
+ if_match: Some(meta.e_tag.unwrap()),
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await?;
+ let data = ret.bytes().await?;
+ assert_eq!(value, data);
+
+ let opts = GetOptions {
+ if_match: Some("invalid etag".to_string()),
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await;
+ assert!(ret.is_err());
+ assert!(matches!(
+ ret.err().unwrap(),
+ object_store::Error::Precondition { .. }
+ ));
+
+ store.delete(&location).await?;
+ Ok(())
+}
+
+async fn test_get_ops_with_if_none_match(store: OpendalStore) -> Result<()> {
+ if !store.info().full_capability().read_with_if_none_match {
+ return Ok(());
+ }
+
+ let location = new_file_path("data").into();
+ let value = Bytes::from_static(b"Hello, world!");
+ store.put(&location, value.clone().into()).await?;
+ let meta = store.head(&location).await?;
+
+ let opts = GetOptions {
+ if_none_match: Some("invalid etag".to_string()),
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await?;
+ let data = ret.bytes().await?;
+ assert_eq!(value, data);
+
+ let opts = GetOptions {
+ if_none_match: Some(meta.e_tag.unwrap()),
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await;
+ assert!(ret.is_err());
+ assert!(matches!(
+ ret.err().unwrap(),
+ object_store::Error::Precondition { .. }
+ ));
+
+ store.delete(&location).await?;
+ Ok(())
+}
+
+async fn test_get_opts_with_if_modified_since(store: OpendalStore) -> Result<()> {
+ if !store.info().full_capability().read_with_if_modified_since {
+ return Ok(());
+ }
+
+ let location = new_file_path("data").into();
+ let value = Bytes::from_static(b"Hello, world!");
+ store.put(&location, value.clone().into()).await?;
+ let meta = store.head(&location).await?;
+
+ let opts = GetOptions {
+ if_modified_since: Some(meta.last_modified - std::time::Duration::from_secs(1)),
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await?;
+ let data = ret.bytes().await?;
+ assert_eq!(value, data);
+
+ let opts = GetOptions {
+ if_modified_since: Some(meta.last_modified + std::time::Duration::from_secs(1)),
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await;
+ assert!(ret.is_err());
+ assert!(matches!(
+ ret.err().unwrap(),
+ object_store::Error::Precondition { .. }
+ ));
+
+ store.delete(&location).await?;
+ Ok(())
+}
+
+async fn test_get_opts_with_if_unmodified_since(store: OpendalStore) -> Result<()> {
+ if !store.info().full_capability().read_with_if_unmodified_since {
+ return Ok(());
+ }
+
+ let location = new_file_path("data").into();
+ let value = Bytes::from_static(b"Hello, world!");
+ store.put(&location, value.clone().into()).await?;
+ let meta = store.head(&location).await?;
+
+ let opts = GetOptions {
+ if_unmodified_since: Some(meta.last_modified + std::time::Duration::from_secs(1)),
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await?;
+ let data = ret.bytes().await?;
+ assert_eq!(value, data);
+
+ let opts = GetOptions {
+ if_unmodified_since: Some(meta.last_modified - std::time::Duration::from_secs(1)),
+ ..Default::default()
+ };
+ let ret = store.get_opts(&location, opts).await;
+ assert!(ret.is_err());
+ assert!(matches!(
+ ret.err().unwrap(),
+ object_store::Error::Precondition { .. }
+ ));
+
+ store.delete(&location).await?;
+ Ok(())
+}
diff --git a/integrations/object_store/tests/behavior/put.rs b/integrations/object_store/tests/behavior/put.rs
index bbafd58b1..be428ac33 100644
--- a/integrations/object_store/tests/behavior/put.rs
+++ b/integrations/object_store/tests/behavior/put.rs
@@ -19,11 +19,26 @@ use crate::utils::{build_trail, new_file_path};
use anyhow::Result;
use bytes::Bytes;
use libtest_mimic::Trial;
-use object_store::ObjectStore;
+use object_store::{ObjectStore, PutMode, UpdateVersion};
use object_store_opendal::OpendalStore;
pub fn tests(store: &OpendalStore, tests: &mut Vec<Trial>) {
tests.push(build_trail("test_basic_put", store, test_basic_put));
+ tests.push(build_trail(
+ "test_put_opts_with_overwrite",
+ store,
+ test_put_opts_with_overwrite,
+ ));
+ tests.push(build_trail(
+ "test_put_opts_with_create",
+ store,
+ test_put_opts_with_create,
+ ));
+ tests.push(build_trail(
+ "test_put_opts_with_update",
+ store,
+ test_put_opts_with_update,
+ ));
}
pub async fn test_basic_put(store: OpendalStore) -> Result<()> {
@@ -37,3 +52,96 @@ pub async fn test_basic_put(store: OpendalStore) -> Result<()> {
store.delete(&location).await?;
Ok(())
}
+
+pub async fn test_put_opts_with_overwrite(store: OpendalStore) -> Result<()> {
+ let location = new_file_path("data").into();
+ let value = Bytes::from("Hello, world!");
+ store.put(&location, value.clone().into()).await?;
+
+ let new_value = Bytes::from("Hello, world! 2");
+ let opts = object_store::PutOptions {
+ mode: PutMode::Overwrite,
+ ..Default::default()
+ };
+ store
+ .put_opts(&location, new_value.clone().into(), opts)
+ .await?;
+
+ let data = store.get(&location).await?.bytes().await?;
+ assert_eq!(new_value, data);
+
+ store.delete(&location).await?;
+ Ok(())
+}
+
+pub async fn test_put_opts_with_create(store: OpendalStore) -> Result<()> {
+ if !store.info().full_capability().write_with_if_not_exists {
+ return Ok(());
+ }
+
+ let location = new_file_path("data").into();
+ let value = Bytes::from("Hello, world!");
+
+ let opts = object_store::PutOptions {
+ mode: PutMode::Create,
+ ..Default::default()
+ };
+ let ret = store
+ .put_opts(&location, value.clone().into(), opts.clone())
+ .await;
+ assert!(ret.is_ok());
+
+ let ret = store.put_opts(&location, value.clone().into(), opts).await;
+ assert!(ret.is_err());
+ assert!(matches!(
+ ret.unwrap_err(),
+ object_store::Error::AlreadyExists { .. }
+ ));
+
+ store.delete(&location).await?;
+ Ok(())
+}
+
+pub async fn test_put_opts_with_update(store: OpendalStore) -> Result<()> {
+ if !store.info().full_capability().write_with_if_match {
+ return Ok(());
+ }
+
+ let location = new_file_path("data").into();
+ let value = Bytes::from("Hello, world!");
+ store.put(&location, value.clone().into()).await?;
+ let meta = store.head(&location).await?;
+
+ let new_value = Bytes::from("Hello, world! 2");
+ let put_opts = object_store::PutOptions {
+ mode: PutMode::Update(UpdateVersion {
+ e_tag: meta.e_tag,
+ version: None,
+ }),
+ ..Default::default()
+ };
+ store
+ .put_opts(&location, new_value.clone().into(), put_opts)
+ .await?;
+ let data = store.get(&location).await?.bytes().await?;
+ assert_eq!(new_value, data);
+
+ let put_opts = object_store::PutOptions {
+ mode: PutMode::Update(UpdateVersion {
+ e_tag: Some("invalid etag".to_string()),
+ version: None,
+ }),
+ ..Default::default()
+ };
+ let ret = store
+ .put_opts(&location, value.clone().into(), put_opts)
+ .await;
+ assert!(ret.is_err());
+ assert!(matches!(
+ ret.unwrap_err(),
+ object_store::Error::Precondition { .. }
+ ));
+
+ store.delete(&location).await?;
+ Ok(())
+}