HADOOP-19801. Scope optimized non-empty directory deletes to the delete request.

Author: deepujain

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java

@@ -1780,6 +1780,20 @@ private Constants() {
   public static final String DELETE_NON_EMPTY_DIRECTORY_ENABLED =
       "fs.s3a.delete.non-empty-directory.enabled";
 
+  /**
+   * Custom delete header required by some S3-compatible stores to delete a
+   * non-empty directory key (prefix) in one request.
+   * Value: {@value}.
+   */
+  public static final String DELETE_NON_EMPTY_DIRECTORY_HEADER =
+      "x-amz-delete-contents";
+
+  /**
+   * Value sent with {@link #DELETE_NON_EMPTY_DIRECTORY_HEADER}.
+   * Value: {@value}.
+   */
+  public static final String DELETE_NON_EMPTY_DIRECTORY_HEADER_VALUE = "true";
+
   /**
    * Default value of {@link #DELETE_NON_EMPTY_DIRECTORY_ENABLED}: {@value}.
    */

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -2623,9 +2623,20 @@ public void deleteObjectAtPath(final Path path,
         final String key,
         final boolean isFile)
         throws IOException {
+      deleteObjectAtPath(path, key, isFile, false);
+    }
+
+    @Override
+    @Retries.RetryTranslated
+    public void deleteObjectAtPath(final Path path,
+        final String key,
+        final boolean isFile,
+        final boolean deleteContents)
+        throws IOException {
       auditSpan.activate();
       once("delete", path.toString(), () ->
-          S3AFileSystem.this.deleteObjectAtPath(path, key, isFile));
+          S3AFileSystem.this.deleteObjectAtPath(
+              path, key, isFile, deleteContents));
     }
 
     @Override
@@ -3227,12 +3238,24 @@ void deleteObjectAtPath(Path f,
       String key,
       boolean isFile)
       throws SdkException, IOException {
+    deleteObjectAtPath(f, key, isFile, false);
+  }
+
+  @Retries.RetryRaw
+  void deleteObjectAtPath(Path f,
+      String key,
+      boolean isFile,
+      boolean deleteContents)
+      throws SdkException, IOException {
     if (isFile) {
       instrumentation.fileDeleted(1);
     } else {
       instrumentation.directoryDeleted();
     }
-    deleteObject(key);
+    incrementWriteOperations();
+    getStore().deleteObject(getRequestFactory()
+        .newDeleteObjectRequestBuilder(key, deleteContents)
+        .build());
   }
 
   /**

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/api/RequestFactory.java

@@ -245,7 +245,20 @@ ListObjectsV2Request.Builder newListObjectsV2RequestBuilder(String key,
    * @param key object to delete
    * @return the request builder.
    */
-  DeleteObjectRequest.Builder newDeleteObjectRequestBuilder(String key);
+  default DeleteObjectRequest.Builder newDeleteObjectRequestBuilder(String key) {
+    return newDeleteObjectRequestBuilder(key, false);
+  }
+
+  /**
+   * Create a request builder to delete a single object.
+   * @param key object to delete
+   * @param deleteContents should the request include the custom non-empty
+   *                       directory delete header?
+   * @return the request builder.
+   */
+  DeleteObjectRequest.Builder newDeleteObjectRequestBuilder(
+      String key,
+      boolean deleteContents);
 
   /**
    * Create a request builder to delete objects in bulk.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/DeleteOperation.java

@@ -252,18 +252,19 @@ public Boolean execute() throws IOException {
       }
       if (status.isEmptyDirectory() == Tristate.TRUE) {
         LOG.debug("deleting empty directory {}", path);
-        deleteObjectAtPath(path, key, false);
+        deleteObjectAtPath(path, key, false, false);
       } else if (deleteNonEmptyDirectoryEnabled) {
-        LOG.debug("deleting non-empty directory {} with single request (endpoint supports it)", path);
-        deleteObjectAtPath(path, key, false);
+        LOG.debug("deleting non-empty directory {} with single request "
+            + "(endpoint supports it)", path);
+        deleteObjectAtPath(path, key, false, true);
       } else {
         deleteDirectoryTree(path, key);
       }
 
     } else {
       // simple file.
       LOG.debug("deleting simple file {}", path);
-      deleteObjectAtPath(path, key, true);
+      deleteObjectAtPath(path, key, true, false);
     }
     LOG.debug("Deleted {} objects", filesDeleted);
     return true;
@@ -363,7 +364,8 @@ private void queueForDeletion(
    */
   private void queueForDeletion(final String key,
       boolean isDirMarker) throws IOException {
-    LOG.debug("Adding object to delete: \"{}\"", key);
+    LOG.debug("Adding {} to delete: \"{}\"",
+        isDirMarker ? "dir marker" : "file", key);
     keys.add(new DeleteEntry(key, isDirMarker));
     if (keys.size() == pageSize) {
       submitNextBatch();
@@ -408,11 +410,12 @@ private void resetDeleteList() {
   private void deleteObjectAtPath(
       final Path path,
       final String key,
-      final boolean isFile)
+      final boolean isFile,
+      final boolean deleteContents)
       throws IOException {
     LOG.debug("delete: {} {}", (isFile ? "file" : "dir marker"), key);
     filesDeleted++;
-    callbacks.deleteObjectAtPath(path, key, isFile);
+    callbacks.deleteObjectAtPath(path, key, isFile, deleteContents);
   }
 
   /**

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/OperationCallbacks.java

@@ -98,6 +98,24 @@ void deleteObjectAtPath(Path path,
       boolean isFile)
       throws IOException;
 
+  /**
+   * Delete an object, optionally enabling the custom non-empty directory
+   * delete header on stores which support it.
+   * @param path path to delete
+   * @param key key of entry
+   * @param isFile is the path a file (used for instrumentation only)
+   * @param deleteContents should the request include the custom delete header?
+   * @throws IOException from invoker signature only -should not be raised.
+   */
+  @Retries.RetryTranslated
+  default void deleteObjectAtPath(Path path,
+      String key,
+      boolean isFile,
+      boolean deleteContents)
+      throws IOException {
+    deleteObjectAtPath(path, key, isFile);
+  }
+
   /**
    * Recursive list of files and directory markers.
    *

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/RequestFactoryImpl.java

@@ -65,6 +65,8 @@
 import static org.apache.commons.lang3.StringUtils.isEmpty;
 import static org.apache.commons.lang3.StringUtils.isNotEmpty;
 import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_PART_UPLOAD_TIMEOUT;
+import static org.apache.hadoop.fs.s3a.Constants.DELETE_NON_EMPTY_DIRECTORY_HEADER;
+import static org.apache.hadoop.fs.s3a.Constants.DELETE_NON_EMPTY_DIRECTORY_HEADER_VALUE;
 import static org.apache.hadoop.fs.s3a.Constants.IF_NONE_MATCH_STAR;
 import static org.apache.hadoop.fs.s3a.S3AEncryptionMethods.SSE_C;
 import static org.apache.hadoop.fs.s3a.S3AEncryptionMethods.UNKNOWN_ALGORITHM;
@@ -724,7 +726,22 @@ public ListObjectsV2Request.Builder newListObjectsV2RequestBuilder(
 
   @Override
   public DeleteObjectRequest.Builder newDeleteObjectRequestBuilder(String key) {
-    return prepareRequest(DeleteObjectRequest.builder().bucket(bucket).key(key));
+    return newDeleteObjectRequestBuilder(key, false);
+  }
+
+  @Override
+  public DeleteObjectRequest.Builder newDeleteObjectRequestBuilder(
+      final String key,
+      final boolean deleteContents) {
+    DeleteObjectRequest.Builder builder =
+        prepareRequest(DeleteObjectRequest.builder().bucket(bucket).key(key));
+    if (deleteContents) {
+      builder.overrideConfiguration(
+          override -> override.putHeader(
+              DELETE_NON_EMPTY_DIRECTORY_HEADER,
+              DELETE_NON_EMPTY_DIRECTORY_HEADER_VALUE));
+    }
+    return builder;
   }
 
   @Override

hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/index.md

@@ -578,8 +578,10 @@ Here are some the S3A properties for use in production.
   <value>false</value>
   <description>When true, recursive deletion of a non-empty directory uses a single delete
     request for the directory key (prefix) instead of listing and deleting contained
-    objects first. Only enable this for S3-compatible endpoints that support
-    deleting non-empty directories (path prefix) in one request (e.g. VAST).
+    objects first. S3A sends the request-scoped header
+    <code>x-amz-delete-contents: true</code> only on that delete call. Only enable
+    this for S3-compatible endpoints that support deleting non-empty directories
+    (path prefix) in one request (e.g. VAST).
   </description>
 </property>
 
@@ -1732,6 +1734,8 @@ to the number of files (objects) in the directory.
 When `fs.s3a.delete.non-empty-directory.enabled=true`, only one delete request is sent for
 the directory (prefix). The S3 endpoint has to support this feature. Depending on the
 S3 endpoint implementation of this feature, deletes might be synchronous or asynchronous.
+S3A sends the custom header `x-amz-delete-contents: true` only on that delete request,
+so there is no need to configure it globally with `fs.s3a.client.s3.custom.headers`.
 
 The [VAST S3 endpoint](https://kb.vastdata.com/documentation/docs/using-trash-folder-for-s3-objects-6)
 supports such deletes.

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ADeleteNonEmptyDirectoryCapability.java

@@ -20,15 +20,13 @@
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.contract.AbstractContractDeleteTest;
-import org.apache.hadoop.fs.contract.AbstractFSContract;
-import org.apache.hadoop.fs.contract.s3a.S3AContract;
 import org.apache.hadoop.test.tags.IntegrationTest;
 
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 import static org.apache.hadoop.fs.s3a.Constants.DELETE_NON_EMPTY_DIRECTORY_ENABLED;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.disableFilesystemCaching;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.getTestBucketName;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
 
@@ -38,22 +36,19 @@
  */
 @IntegrationTest
 public class ITestS3ADeleteNonEmptyDirectoryCapability extends
-    AbstractContractDeleteTest {
-
-  @Override
-  protected AbstractFSContract createContract(Configuration conf) {
-    return new S3AContract(conf);
-  }
+    AbstractS3ATestBase {
 
   /**
    * Test that when the option is disabled (default), the capability is false.
    */
   @Test
   public void testCapabilityDisabledByDefault() throws Throwable {
-    Assertions.assertThat(getFileSystem().hasPathCapability(new Path("/"),
-        DELETE_NON_EMPTY_DIRECTORY_ENABLED))
-        .describedAs("path capability when option not set")
-        .isFalse();
+    try (S3AFileSystem fs = createCapabilityTestFileSystem(false)) {
+      Assertions.assertThat(fs.hasPathCapability(new Path("/"),
+          DELETE_NON_EMPTY_DIRECTORY_ENABLED))
+          .describedAs("path capability when option not set")
+          .isFalse();
+    }
   }
 
   /**
@@ -62,16 +57,23 @@ public void testCapabilityDisabledByDefault() throws Throwable {
    */
   @Test
   public void testCapabilityWhenEnabled() throws Throwable {
-    Configuration conf = new Configuration(getFileSystem().getConf());
-    removeBaseAndBucketOverrides(getTestBucketName(conf), conf,
-        DELETE_NON_EMPTY_DIRECTORY_ENABLED);
-    conf.setBoolean(DELETE_NON_EMPTY_DIRECTORY_ENABLED, true);
-    try (S3AFileSystem fs = new S3AFileSystem()) {
-      fs.initialize(getFileSystem().getUri(), conf);
+    try (S3AFileSystem fs = createCapabilityTestFileSystem(true)) {
       Assertions.assertThat(fs.hasPathCapability(new Path("/"),
           DELETE_NON_EMPTY_DIRECTORY_ENABLED))
           .describedAs("path capability when option enabled")
           .isTrue();
     }
   }
+
+  private S3AFileSystem createCapabilityTestFileSystem(
+      final boolean enabled) throws Exception {
+    Configuration conf = new Configuration(getFileSystem().getConf());
+    removeBaseAndBucketOverrides(getTestBucketName(conf), conf,
+        DELETE_NON_EMPTY_DIRECTORY_ENABLED);
+    disableFilesystemCaching(conf);
+    conf.setBoolean(DELETE_NON_EMPTY_DIRECTORY_ENABLED, enabled);
+    S3AFileSystem fs = new S3AFileSystem();
+    fs.initialize(getFileSystem().getUri(), conf);
+    return fs;
+  }
 }

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/MockS3AFileSystem.java

@@ -358,10 +358,19 @@ void deleteObjectAtPath(Path f,
       String key,
       boolean isFile)
       throws SdkException, IOException {
+    deleteObjectAtPath(f, key, isFile, false);
+  }
+
+  @Override
+  void deleteObjectAtPath(Path f,
+      String key,
+      boolean isFile,
+      boolean deleteContents)
+      throws SdkException, IOException {
     mock.getS3AInternals()
             .getAmazonS3Client("test")
             .deleteObject(getRequestFactory()
-            .newDeleteObjectRequestBuilder(key)
+            .newDeleteObjectRequestBuilder(key, deleteContents)
             .build());
   }
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestRequestFactory.java

@@ -37,6 +37,7 @@
 import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
 import software.amazon.awssdk.services.s3.model.CopyObjectRequest;
 import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
 import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
 import org.assertj.core.api.Assertions;
 import org.slf4j.Logger;
@@ -56,6 +57,8 @@
 import org.apache.hadoop.test.AbstractHadoopTestBase;
 
 import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_PART_UPLOAD_TIMEOUT;
+import static org.apache.hadoop.fs.s3a.Constants.DELETE_NON_EMPTY_DIRECTORY_HEADER;
+import static org.apache.hadoop.fs.s3a.Constants.DELETE_NON_EMPTY_DIRECTORY_HEADER_VALUE;
 import static org.apache.hadoop.fs.s3a.impl.PutObjectOptions.defaultOptions;
 import static org.apache.hadoop.test.LambdaTestUtils.intercept;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -147,6 +150,30 @@ public void testRequestFactoryWithProcessor() throws Throwable {
         .isEqualTo(requestsAnalyzed);
   }
 
+  @Test
+  public void testDeleteObjectRequestWithDeleteContentsHeader()
+      throws Throwable {
+    RequestFactory factory = RequestFactoryImpl.builder()
+        .withBucket("bucket")
+        .build();
+
+    DeleteObjectRequest request =
+        factory.newDeleteObjectRequestBuilder("path", true).build();
+    Assertions.assertThat(request.overrideConfiguration())
+        .describedAs("override config for delete request")
+        .isPresent();
+    Assertions.assertThat(request.overrideConfiguration().get().headers())
+        .containsEntry(
+            DELETE_NON_EMPTY_DIRECTORY_HEADER,
+            Collections.singletonList(DELETE_NON_EMPTY_DIRECTORY_HEADER_VALUE));
+
+    DeleteObjectRequest defaultRequest =
+        factory.newDeleteObjectRequestBuilder("path").build();
+    Assertions.assertThat(defaultRequest.overrideConfiguration())
+        .describedAs("override config for default delete request")
+        .isEmpty();
+  }
+
   private final class CountRequests
       implements RequestFactoryImpl.PrepareRequest {