feat(services): support conditional delete for Azure services

core/core/src/types/delete/deleter.rs

@@ -102,6 +102,9 @@ impl Deleter {
         if input.recursive {
             op = op.with_recursive(true);
         }
+        if let Some(if_match) = &input.if_match {
+            op = op.with_if_match(if_match);
+        }
 
         self.deleter.delete(&input.path, op).await?;
         Ok(())

core/core/src/types/delete/input.rs

@@ -28,6 +28,8 @@ pub struct DeleteInput {
     pub version: Option<String>,
     /// Whether to perform recursive deletion.
     pub recursive: bool,
+    /// Delete the path only when its ETag matches this value.
+    pub if_match: Option<String>,
 }
 
 /// IntoDeleteInput is a helper trait that makes it easier for users to play with `Deleter`.
@@ -80,6 +82,9 @@ impl IntoDeleteInput for (String, OpDelete) {
         if let Some(version) = args.version() {
             input.version = Some(version.to_string());
         }
+        if let Some(if_match) = args.if_match() {
+            input.if_match = Some(if_match.to_string());
+        }
         input
     }
 }
@@ -101,3 +106,19 @@ impl IntoDeleteInput for Entry {
         input
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_op_delete_input_preserves_if_match() {
+        let input = (
+            "path".to_string(),
+            OpDelete::new().with_if_match("\"etag\""),
+        )
+            .into_delete_input();
+
+        assert_eq!(input.if_match.as_deref(), Some("\"etag\""));
+    }
+}

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,

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(

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)));
             }
         }
 

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,

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())

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();

core/services/s3/src/core.rs

@@ -2222,7 +2222,9 @@ mod error {
             | "ExceedBucketQPSLimit"
             | "ExceedBucketRateLimit" => Some((ErrorKind::RateLimited, true)),
             "InvalidRange" => Some((ErrorKind::RangeNotSatisfied, false)),
-            "PreconditionFailed" => Some((ErrorKind::ConditionNotMatch, false)),
+            "PreconditionFailed" | "412 Precondition Failed" => {
+                Some((ErrorKind::ConditionNotMatch, false))
+            }
             _ => None,
         }
     }
@@ -2287,6 +2289,10 @@ mod error {
                 parse_s3_error_code("PreconditionFailed"),
                 Some((ErrorKind::ConditionNotMatch, false))
             );
+            assert_eq!(
+                parse_s3_error_code("412 Precondition Failed"),
+                Some((ErrorKind::ConditionNotMatch, false))
+            );
         }
     }
 }

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(())
 }