HADOOP-18656: [ABFS] Adding Support for Paginated Delete for Large Directories in HNS Account

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsConfiguration.java

@@ -338,6 +338,10 @@ public class AbfsConfiguration{
           FS_AZURE_ABFS_RENAME_RESILIENCE, DefaultValue = DEFAULT_ENABLE_ABFS_RENAME_RESILIENCE)
   private boolean renameResilience;
 
+  @BooleanConfigurationValidatorAnnotation(ConfigurationKey =
+      FS_AZURE_ENABLE_PAGINATED_DELETE, DefaultValue = DEFAULT_ENABLE_PAGINATED_DELETE)
+  private boolean isPaginatedDeleteEnabled;
+
   private String clientProvidedEncryptionKey;
 
   private String clientProvidedEncryptionKeySHA;
@@ -1191,7 +1195,12 @@ public boolean getRenameResilience() {
     return renameResilience;
   }
 
-  void setRenameResilience(boolean actualResilience) {
-    renameResilience = actualResilience;
+  public boolean isPaginatedDeleteEnabled() {
+    return isPaginatedDeleteEnabled;
+  }
+
+  @VisibleForTesting
+  public void setIsPaginatedDeleteEnabled(boolean isPaginatedDeleteEnabled) {
+    this.isPaginatedDeleteEnabled = isPaginatedDeleteEnabled;
   }
 }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/AbfsHttpConstants.java

@@ -121,6 +121,7 @@ public final class AbfsHttpConstants {
   public static final char CHAR_PLUS = '+';
   public static final String DECEMBER_2019_API_VERSION = "2019-12-12";
   public static final String APRIL_2021_API_VERSION = "2021-04-10";
+  public static final String AUGUST_2023_API_VERSION = "2023-08-03";
 
   /**
    * Value that differentiates categories of the http_status.

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/ConfigurationKeys.java

@@ -247,6 +247,11 @@ public final class ConfigurationKeys {
   /** Add extra resilience to rename failures, at the expense of performance. */
   public static final String FS_AZURE_ABFS_RENAME_RESILIENCE = "fs.azure.enable.rename.resilience";
 
+  /**
+   * Specify whether paginated behavior is to be expected or not in delete path.
+   */
+  public static final String FS_AZURE_ENABLE_PAGINATED_DELETE = "fs.azure.enable.paginated.delete";
+
   public static String accountProperty(String property, String account) {
     return property + "." + account;
   }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/FileSystemConfigurations.java

@@ -119,6 +119,7 @@ public final class FileSystemConfigurations {
   public static final int STREAM_ID_LEN = 12;
   public static final boolean DEFAULT_ENABLE_ABFS_LIST_ITERATOR = true;
   public static final boolean DEFAULT_ENABLE_ABFS_RENAME_RESILIENCE = true;
+  public static final boolean DEFAULT_ENABLE_PAGINATED_DELETE = false;
 
   /**
    * Limit of queued block upload operations before writes

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/HttpQueryParams.java

@@ -40,6 +40,7 @@ public final class HttpQueryParams {
   public static final String QUERY_PARAM_CLOSE = "close";
   public static final String QUERY_PARAM_UPN = "upn";
   public static final String QUERY_PARAM_BLOBTYPE = "blobtype";
+  public static final String QUERY_PARAM_PAGINATED = "paginated";
 
   //query params for SAS
   public static final String QUERY_PARAM_SAOID = "saoid";

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java

@@ -1053,12 +1053,24 @@ public AbfsRestOperation read(final String path,
     return op;
   }
 
-  public AbfsRestOperation deletePath(final String path, final boolean recursive, final String continuation,
+  public AbfsRestOperation deletePath(final String path, final boolean recursive,
+                                      final String continuation,
                                       TracingContext tracingContext)
           throws AzureBlobFileSystemException {
     final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
-
     final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
+
+    if (abfsConfiguration.isPaginatedDeleteEnabled() && recursive) {
+      // Change the x-ms-version to "2023-08-03" if its less than that.
+      if (xMsVersion.compareTo(AUGUST_2023_API_VERSION) < 0) {
+        requestHeaders.removeIf(header -> header.getName().equalsIgnoreCase(X_MS_VERSION));
+        requestHeaders.add(new AbfsHttpHeader(X_MS_VERSION, AUGUST_2023_API_VERSION));
+      }
+
+      // Add paginated query parameter
+      abfsUriQueryBuilder.addQuery(QUERY_PARAM_PAGINATED, TRUE);
+    }
+
     abfsUriQueryBuilder.addQuery(QUERY_PARAM_RECURSIVE, String.valueOf(recursive));
     abfsUriQueryBuilder.addQuery(QUERY_PARAM_CONTINUATION, continuation);
     String operation = recursive ? SASTokenProvider.DELETE_RECURSIVE_OPERATION : SASTokenProvider.DELETE_OPERATION;

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/AbfsClientUtils.java

@@ -18,6 +18,8 @@
 
 package org.apache.hadoop.fs.azurebfs.services;
 
+import java.util.List;
+
 import org.apache.hadoop.fs.azurebfs.extensions.EncryptionContextProvider;
 
 public final class AbfsClientUtils {
@@ -31,4 +33,13 @@ public static void setIsNamespaceEnabled(final AbfsClient abfsClient, final Bool
   public static void setEncryptionContextProvider(final AbfsClient abfsClient, final EncryptionContextProvider provider) {
     abfsClient.setEncryptionContextProvider(provider);
   }
+
+  public static String getHeaderValue(List<AbfsHttpHeader> reqHeaders, String headerName) {
+    for (AbfsHttpHeader header : reqHeaders) {
+      if (header.getName().equals(headerName)) {
+        return header.getValue();
+      }
+    }
+    return "";
+  }
 }

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/ITestAbfsClient.java

@@ -410,7 +410,7 @@ public static AbfsClient getMockAbfsClient(AbfsClient baseAbfsClientInstance,
     return client;
   }
 
-  private static AbfsClient setAbfsClientField(
+  static AbfsClient setAbfsClientField(
       final AbfsClient client,
       final String fieldName,
       Object fieldObject) throws Exception {