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 2026/08/25 09:50:50 UTC
(opendal) 01/01: feat(services): support conditional delete for Azure services
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/azure-delete-with-if-match
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 9ee557f81439eb0245c6fa5dbe00e2a24ca286e5
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Aug 25 17:50:03 2026 +0800
feat(services): support conditional delete for Azure services
---
core/services/azblob/src/backend.rs | 1 +
core/services/azblob/src/core.rs | 20 ++++++---
core/services/azblob/src/deleter.rs | 24 +++++-----
core/services/azdls/src/backend.rs | 1 +
core/services/azdls/src/core.rs | 18 +++++++-
core/services/azdls/src/deleter.rs | 6 ++-
core/tests/behavior/async_delete.rs | 89 ++++++++++++++++++++++++++++++++++---
7 files changed, 129 insertions(+), 30 deletions(-)
diff --git a/core/services/azblob/src/backend.rs b/core/services/azblob/src/backend.rs
index 7f269ee38..0e80ed458 100644
--- a/core/services/azblob/src/backend.rs
+++ b/core/services/azblob/src/backend.rs
@@ -413,6 +413,7 @@ impl Builder for AzblobBuilder {
write_with_user_metadata: true,
delete: true,
+ delete_with_if_match: true,
delete_max_size: Some(AZBLOB_BATCH_LIMIT),
copy: true,
diff --git a/core/services/azblob/src/core.rs b/core/services/azblob/src/core.rs
index c70ad3a36..0ef723eb4 100644
--- a/core/services/azblob/src/core.rs
+++ b/core/services/azblob/src/core.rs
@@ -725,9 +725,14 @@ impl AzblobCore {
self.send(ctx, req).await
}
- fn azblob_delete_blob_request(&self, path: &str) -> Result<Request<Buffer>> {
- Request::delete(self.build_path_url(path))
- .header(CONTENT_LENGTH, 0)
+ fn azblob_delete_blob_request(&self, path: &str, args: &OpDelete) -> Result<Request<Buffer>> {
+ let mut req = Request::delete(self.build_path_url(path));
+
+ if let Some(if_match) = args.if_match() {
+ req = req.header(IF_MATCH, if_match);
+ }
+
+ req.header(CONTENT_LENGTH, 0)
.extension(Operation::Delete)
.extension(ServiceOperation("DeleteBlob"))
.body(Buffer::new())
@@ -738,8 +743,9 @@ impl AzblobCore {
&self,
ctx: &OperationContext,
path: &str,
+ args: &OpDelete,
) -> Result<Response<Buffer>> {
- let req = self.azblob_delete_blob_request(path)?;
+ let req = self.azblob_delete_blob_request(path, args)?;
let req = self.sign(ctx, req).await?;
self.send(ctx, req).await
}
@@ -817,7 +823,7 @@ impl AzblobCore {
pub async fn azblob_batch_delete(
&self,
ctx: &OperationContext,
- paths: &[String],
+ batch: &[(String, OpDelete)],
) -> Result<Response<Buffer>> {
let url = format!(
"{}/{}?restype=container&comp=batch",
@@ -826,8 +832,8 @@ impl AzblobCore {
let mut multipart = Multipart::new();
- for (idx, path) in paths.iter().enumerate() {
- let req = self.azblob_delete_blob_request(path)?;
+ for (idx, (path, args)) in batch.iter().enumerate() {
+ let req = self.azblob_delete_blob_request(path, args)?;
let req = self.batch_sign(ctx, req).await?;
multipart = multipart.part(
diff --git a/core/services/azblob/src/deleter.rs b/core/services/azblob/src/deleter.rs
index 06e365381..a0a89aebe 100644
--- a/core/services/azblob/src/deleter.rs
+++ b/core/services/azblob/src/deleter.rs
@@ -37,8 +37,11 @@ impl AzblobDeleter {
}
impl oio::BatchDelete for AzblobDeleter {
- async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
- let resp = self.core.azblob_delete_blob(&self.ctx, &path).await?;
+ async fn delete_once(&self, path: String, args: OpDelete) -> Result<()> {
+ let resp = self
+ .core
+ .azblob_delete_blob(&self.ctx, &path, &args)
+ .await?;
let status = resp.status();
@@ -50,10 +53,8 @@ impl oio::BatchDelete for AzblobDeleter {
async fn delete_batch(&self, batch: Vec<(String, OpDelete)>) -> Result<BatchDeleteResult> {
// TODO: Add remove version support.
- let paths = batch.into_iter().map(|(p, _)| p).collect::<Vec<_>>();
-
// construct and complete batch request
- let resp = self.core.azblob_batch_delete(&self.ctx, &paths).await?;
+ let resp = self.core.azblob_batch_delete(&self.ctx, &batch).await?;
// check response status
if resp.status() != StatusCode::ACCEPTED {
@@ -75,26 +76,23 @@ impl oio::BatchDelete for AzblobDeleter {
Multipart::new().with_boundary(&boundary).parse(bs)?;
let parts = multipart.into_parts();
- if paths.len() != parts.len() {
+ if batch.len() != parts.len() {
return Err(Error::new(
ErrorKind::Unexpected,
- "invalid batch response, paths and response parts don't match",
+ "invalid batch response, requests and response parts don't match",
));
}
let mut batched_result = BatchDeleteResult::default();
- for (i, part) in parts.into_iter().enumerate() {
+ for (part, (path, args)) in parts.into_iter().zip(batch) {
let resp = part.into_response();
- let path = paths[i].clone();
// deleting not existing objects is ok
if resp.status() == StatusCode::ACCEPTED || resp.status() == StatusCode::NOT_FOUND {
- batched_result.succeeded.push((path, OpDelete::default()));
+ batched_result.succeeded.push((path, args));
} else {
- batched_result
- .failed
- .push((path, OpDelete::default(), parse_error(resp)));
+ batched_result.failed.push((path, args, parse_error(resp)));
}
}
diff --git a/core/services/azdls/src/backend.rs b/core/services/azdls/src/backend.rs
index 59819d356..011541188 100644
--- a/core/services/azdls/src/backend.rs
+++ b/core/services/azdls/src/backend.rs
@@ -337,6 +337,7 @@ impl Builder for AzdlsBuilder {
create_dir: true,
delete: true,
+ delete_with_if_match: true,
delete_with_recursive: true,
rename: true,
diff --git a/core/services/azdls/src/core.rs b/core/services/azdls/src/core.rs
index d757784c1..a63b226d3 100644
--- a/core/services/azdls/src/core.rs
+++ b/core/services/azdls/src/core.rs
@@ -421,6 +421,7 @@ impl AzdlsCore {
&self,
ctx: &OperationContext,
path: &str,
+ args: &OpDelete,
) -> Result<Response<Buffer>> {
let p = build_abs_path(&self.root, path)
.trim_end_matches('/')
@@ -433,7 +434,13 @@ impl AzdlsCore {
percent_encode_path(&p)
);
- let req = Request::delete(&url)
+ let mut req = Request::delete(&url);
+
+ if let Some(if_match) = args.if_match() {
+ req = req.header(IF_MATCH, if_match);
+ }
+
+ let req = req
.extension(Operation::Delete)
.extension(ServiceOperation("DeletePath"))
.body(Buffer::new())
@@ -447,6 +454,7 @@ impl AzdlsCore {
&self,
ctx: &OperationContext,
path: &str,
+ args: &OpDelete,
) -> Result<Response<Buffer>> {
let p = build_abs_path(&self.root, path)
.trim_end_matches('/')
@@ -472,7 +480,13 @@ impl AzdlsCore {
url = url.push("continuation", &percent_encode_path(&continuation));
}
- let req = Request::delete(url.finish())
+ let mut req = Request::delete(url.finish());
+
+ if let Some(if_match) = args.if_match() {
+ req = req.header(IF_MATCH, if_match);
+ }
+
+ let req = req
.extension(Operation::Delete)
.extension(ServiceOperation("RecursiveDeletePath"))
.body(Buffer::new())
diff --git a/core/services/azdls/src/deleter.rs b/core/services/azdls/src/deleter.rs
index 67a4bf430..219dc8c1d 100644
--- a/core/services/azdls/src/deleter.rs
+++ b/core/services/azdls/src/deleter.rs
@@ -38,9 +38,11 @@ impl AzdlsDeleter {
impl oio::OneShotDelete for AzdlsDeleter {
async fn delete_once(&self, path: String, args: OpDelete) -> Result<()> {
let resp = if args.recursive() {
- self.core.azdls_recursive_delete(&self.ctx, &path).await?
+ self.core
+ .azdls_recursive_delete(&self.ctx, &path, &args)
+ .await?
} else {
- self.core.azdls_delete(&self.ctx, &path).await?
+ self.core.azdls_delete(&self.ctx, &path, &args).await?
};
let status = resp.status();
diff --git a/core/tests/behavior/async_delete.rs b/core/tests/behavior/async_delete.rs
index 9effe5d8f..e074d0fbc 100644
--- a/core/tests/behavior/async_delete.rs
+++ b/core/tests/behavior/async_delete.rs
@@ -52,7 +52,8 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
tests.extend(async_trials!(
op,
test_delete_with_if_match_match,
- test_delete_with_if_match_mismatch
+ test_delete_with_if_match_mismatch,
+ test_batch_delete_with_if_match
));
}
}
@@ -441,8 +442,8 @@ pub async fn test_delete_with_if_match_match(op: Operator) -> Result<()> {
Ok(())
}
-/// Delete with a non-matching `If-Match` ETag should fail with
-/// [`ErrorKind::ConditionNotMatch`] and leave the object intact.
+/// Delete with a stale `If-Match` ETag should fail with
+/// [`ErrorKind::ConditionNotMatch`] and leave the replacement intact.
pub async fn test_delete_with_if_match_mismatch(op: Operator) -> Result<()> {
if !op.info().capability().delete_with_if_match {
return Ok(());
@@ -451,14 +452,90 @@ pub async fn test_delete_with_if_match_mismatch(op: Operator) -> Result<()> {
let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
op.write(&path, content).await.expect("write must succeed");
+ let stale_etag = op
+ .stat(&path)
+ .await
+ .expect("stat must succeed")
+ .etag()
+ .expect("etag must be present")
+ .to_string();
+ let replacement = "replacement generation";
+ op.write(&path, replacement)
+ .await
+ .expect("replacement write must succeed");
+
let err = op
.delete_with(&path)
- .if_match("\"this-etag-does-not-match\"")
+ .if_match(&stale_etag)
.await
- .expect_err("delete must fail when etag mismatches");
+ .expect_err("stale ETag must not delete the replacement");
assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+ assert_eq!(op.read(&path).await?.to_bytes(), replacement.as_bytes());
+
+ op.delete(&path).await?;
+
+ Ok(())
+}
+
+/// Batch delete should apply each entry's `If-Match` condition independently.
+pub async fn test_batch_delete_with_if_match(op: Operator) -> Result<()> {
+ let mut cap = op.info().capability();
+ if cap.delete_max_size.unwrap_or(1) <= 1 {
+ return Ok(());
+ }
+
+ cap.delete_max_size = Some(2);
+ let op = op.layer(CapabilityOverrideLayer::new(move |_| cap));
+
+ let (matching_path, matching_content, _) = TEST_FIXTURE.new_file(op.clone());
+ op.write(&matching_path, matching_content)
+ .await
+ .expect("write must succeed");
+ let matching_etag = op
+ .stat(&matching_path)
+ .await
+ .expect("stat must succeed")
+ .etag()
+ .expect("etag must be present")
+ .to_string();
+
+ let (stale_path, stale_content, _) = TEST_FIXTURE.new_file(op.clone());
+ op.write(&stale_path, stale_content)
+ .await
+ .expect("write must succeed");
+ let stale_etag = op
+ .stat(&stale_path)
+ .await
+ .expect("stat must succeed")
+ .etag()
+ .expect("etag must be present")
+ .to_string();
+ let replacement = "replacement generation";
+ op.write(&stale_path, replacement)
+ .await
+ .expect("replacement write must succeed");
+
+ let err = op
+ .delete_iter([
+ (
+ matching_path.clone(),
+ OpDelete::new().with_if_match(&matching_etag),
+ ),
+ (
+ stale_path.clone(),
+ OpDelete::new().with_if_match(&stale_etag),
+ ),
+ ])
+ .await
+ .expect_err("batch delete must report the stale ETag");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+ assert!(!op.exists(&matching_path).await?);
+ assert_eq!(
+ op.read(&stale_path).await?.to_bytes(),
+ replacement.as_bytes()
+ );
- assert!(op.exists(&path).await?);
+ op.delete(&stale_path).await?;
Ok(())
}