HADOOP-18679. Add API for bulk/paged object deletion

A more minimal design that is easier to use and implement.

Caller creates a BulkOperation; they get the page size of
it and then submit batches to delete of less than that size.

The outcome of each call contains a list of failures.

S3A implementation to show how straightforward it is.

Even with the single entry page size, it is still more
efficient to use this as it doesn't try to recreate a
parent dir or perform any probes to see if it is a directory:
it maps straight to a DELETE call.

Change-Id: Ibe8737e7933fe03d39070e7138a710b50c3d60c2

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/BulkDelete.java

@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
+
+/**
+ * API for bulk deletion of objects/files,
+ * <i>but not directories</i>.
+ * After use, call {@code close()} to release any resources and
+ * to guarantee store IOStatistics are updated.
+ */
+public interface BulkDelete extends IOStatisticsSource, Closeable {
+
+  /**
+   * The maximum number of objects/files to delete in a single request.
+   * @return a number greater than or equal to zero.
+   */
+  int pageSize();
+
+  /**
+   * Base path of a bulk delete operation.
+   * All paths submitted in {@link #bulkDelete(List)} must be under this path.
+   */
+  Path basePath();
+
+  /**
+   * Delete a list of files/objects.
+   * <ul>
+   *   <li>Files must be under the path provided in {@link #basePath()}.</li>
+   *   <li>The size of the list must be equal to or less than the page size
+   *       declared in {@link #pageSize()}.</li>
+   *   <li>Directories are not supported; the outcome of attempting to delete
+   *       directories is undefined (ignored; undetected, listed as failures...).</li>
+   *   <li>The operation is not atomic.</li>
+   *   <li>The operation is treated as idempotent: network failures may
+   *        trigger resubmission of the request -any new objects created under a
+   *        path in the list may then be deleted.</li>
+   *    <li>There is no guarantee that any parent directories exist after this call.
+   *    </li>
+   * </ul>
+   * @param paths list of paths which must be absolute and under the base path.
+   * provided in {@link #basePath()}.
+   * @throws IOException IO problems including networking, authentication and more.
+   * @throws IllegalArgumentException if a path argument is invalid.
+   */
+  BulkDeleteOutcome bulkDelete(List<Path> paths)
+      throws IOException, IllegalArgumentException;
+
+  /**
+   * The outcome: a list of paths which failed to delete.
+   * An empty list means all files were successfully deleted.
+   * There are no guarantees about the ordering of the list.
+   * Reasons for failures are not provided.
+   * File Not Found is not a failure.
+   */
+  class BulkDeleteOutcome {
+
+    /**
+     * List of paths which failed to delete.
+     */
+    private final List<BulkDeleteOutcomeElement> failures;
+
+    /**
+     * Constructor.
+     * @param failures list of failures. This must be non-null.
+     */
+    public BulkDeleteOutcome(final List<BulkDeleteOutcomeElement> failures) {
+      this.failures = failures;
+    }
+
+    /**
+     * Get the list of failures.
+     * @return a possibly empty list of failures.
+     */
+    public List<BulkDeleteOutcomeElement> getFailures() {
+      return failures;
+    }
+  }
+
+  class BulkDeleteOutcomeElement {
+    private final Path path;
+    private final String error;
+    private final Exception exception;
+
+    public BulkDeleteOutcomeElement(
+        final Path path,
+        final String error,
+        final Exception exception) {
+      this.path = path;
+      this.error = error;
+      this.exception = exception;
+    }
+
+    public Path getPath() {
+      return path;
+    }
+
+    public String getError() {
+      return error;
+    }
+
+    public Exception getException() {
+      return exception;
+    }
+  }
+
+}

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/BulkDeleteSource.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs;
+
+import java.io.IOException;
+
+/**
+ * Interface for bulk deletion.
+ * Filesystems which support bulk deletion should implement this interface
+ * and MUST also declare their support in the path capability
+ * {@link CommonPathCapabilities#BULK_DELETE}.
+ * Exporting the interface does not guarantee that the operation is supported;
+ * returning a {@link BulkDelete} object from the call {@link #createBulkDelete(Path)}
+ * is.
+ */
+public interface BulkDeleteSource {
+
+  /**
+   * Create a bulk delete operation.
+   * There is no network IO at this point, simply the creation of
+   * a bulk delete object.
+   * A path must be supplied to ensure that on viewfs and similar filesystems,
+   * @param path path to delete under.
+   * @return the bulk delete.
+   * @throws UnsupportedOperationException the filesystem does not support delete under that path.
+   * @throws IllegalArgumentException path not valid.
+   * @throws IOException problems resolving paths
+   */
+  default BulkDelete createBulkDelete(Path path)
+      throws UnsupportedOperationException, IllegalArgumentException, IOException {
+    throw new UnsupportedOperationException("Bulk delete not supported");
+  }
+
+}

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonPathCapabilities.java

@@ -181,4 +181,10 @@ private CommonPathCapabilities() {
    */
   public static final String DIRECTORY_LISTING_INCONSISTENT =
       "fs.capability.directory.listing.inconsistent";
+
+  /**
+   * Capability string to probe for bulk delete: {@value}.
+   */
+  public static final String BULK_DELETE = "fs.capability.bulk.delete";
+
 }

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

@@ -105,6 +105,8 @@
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.BulkDelete;
+import org.apache.hadoop.fs.BulkDeleteSource;
 import org.apache.hadoop.fs.CommonPathCapabilities;
 import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.fs.CreateFlag;
@@ -121,6 +123,7 @@
 import org.apache.hadoop.fs.s3a.auth.delegation.DelegationTokenProvider;
 import org.apache.hadoop.fs.s3a.impl.AWSCannedACL;
 import org.apache.hadoop.fs.s3a.impl.AWSHeaders;
+import org.apache.hadoop.fs.s3a.impl.BulkDeleteOperation;
 import org.apache.hadoop.fs.s3a.impl.BulkDeleteRetryHandler;
 import org.apache.hadoop.fs.s3a.impl.ChangeDetectionPolicy;
 import org.apache.hadoop.fs.s3a.impl.ConfigurationHelper;
@@ -284,7 +287,8 @@
 @InterfaceStability.Evolving
 public class S3AFileSystem extends FileSystem implements StreamCapabilities,
     AWSPolicyProvider, DelegationTokenProvider, IOStatisticsSource,
-    AuditSpanSource<AuditSpanS3A>, ActiveThreadSpanSource<AuditSpanS3A> {
+    AuditSpanSource<AuditSpanS3A>, ActiveThreadSpanSource<AuditSpanS3A>,
+    BulkDeleteSource {
 
   /**
    * Default blocksize as used in blocksize and FS status queries.
@@ -3406,6 +3410,13 @@ private void removeKeysS3(
       // exit fast if there are no keys to delete
       return;
     }
+    if (keysToDelete.size() == 1) {
+      // single object is a single delete call.
+      // this is more informative in server logs and may be more efficient..
+      deleteObject(keysToDelete.get(0).key());
+      noteDeleted(1, deleteFakeDir);
+      return;
+    }
     for (ObjectIdentifier objectIdentifier : keysToDelete) {
       blockRootDelete(objectIdentifier.key());
     }
@@ -5476,7 +5487,11 @@ public boolean hasPathCapability(final Path path, final String capability)
     case STORE_CAPABILITY_DIRECTORY_MARKER_AWARE:
       return true;
 
-      // multi object delete flag
+    // this is always true, even if multi object
+    // delete is disabled -the page size is simply reduced to 1.
+    case CommonPathCapabilities.BULK_DELETE:
+      return true;
+
     case ENABLE_MULTI_DELETE:
       return enableMultiObjectsDelete;
 
@@ -5851,4 +5866,50 @@ public boolean isMultipartUploadEnabled() {
     return isMultipartUploadEnabled;
   }
 
+  @Override
+  public BulkDelete createBulkDelete(final Path path)
+      throws IllegalArgumentException, IOException {
+
+    final Path p = makeQualified(path);
+    final AuditSpanS3A span = createSpan("bulkdelete", p.toString(), null);
+    return new BulkDeleteOperation(
+        createStoreContext(),
+        createBulkDeleteCallbacks(span),
+        p,
+        enableMultiObjectsDelete ? 1: pageSize,
+        span);
+  }
+
+  /**
+   * Override point for mocking.
+   * @param span span for operations.
+   * @return an instance of the Bulk Delette callbacks.
+   */
+  protected BulkDeleteOperation.BulkDeleteOperationCallbacks createBulkDeleteCallbacks(
+      final AuditSpanS3A span) {
+    return new BulkDeleteOperationCallbacksImpl(span);
+  }
+
+  /**
+   * Callbacks for the bulk delete operation.
+   */
+  protected class BulkDeleteOperationCallbacksImpl implements
+      BulkDeleteOperation.BulkDeleteOperationCallbacks {
+
+    /** span for operations. */
+    private final AuditSpan span;
+
+    protected BulkDeleteOperationCallbacksImpl(AuditSpan span) {
+      this.span = span;
+    }
+
+    @Override
+    @Retries.RetryTranslated
+    public void bulkDelete(final List<ObjectIdentifier> keys)
+        throws MultiObjectDeleteException, IOException, IllegalArgumentException {
+      span.activate();
+      once("bulkDelete", "", () ->
+          S3AFileSystem.this.removeKeys(keys, false));
+    }
+  }
 }