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

@@ -343,6 +343,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;
@@ -1205,7 +1209,7 @@ public boolean getRenameResilience() {
     return renameResilience;
   }
 
-  void setRenameResilience(boolean actualResilience) {
-    renameResilience = actualResilience;
+  public boolean isPaginatedDeleteEnabled() {
+    return isPaginatedDeleteEnabled;
   }
 }

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

@@ -120,8 +120,34 @@ public final class AbfsHttpConstants {
   public static final char CHAR_EQUALS = '=';
   public static final char CHAR_STAR = '*';
   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";
+
+  /**
+   * Specifies the version of the REST protocol used for processing the request.
+   * Versions should be added in enum list in ascending chronological order.
+   * Latest one should be added last in the list.
+   * When upgrading the version for whole driver, update the getCurrentVersion;
+   */
+  public enum ApiVersion {
+
+    DEC_12_2019("2019-12-12"),
+    APR_10_2021("2021-04-10"),
+    AUG_03_2023("2023-08-03");
+
+    private final String xMsApiVersion;
+
+    ApiVersion(String xMsApiVersion) {
+      this.xMsApiVersion = xMsApiVersion;
+    }
+
+    @Override
+    public String toString() {
+      return xMsApiVersion;
+    }
+
+    public static ApiVersion getCurrentVersion() {
+      return DEC_12_2019;
+    }
+  }
 
   /**
    * Value that differentiates categories of the http_status.

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

@@ -262,6 +262,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

@@ -120,6 +120,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

@@ -92,7 +92,7 @@ public class AbfsClient implements Closeable {
 
   private final URL baseUrl;
   private final SharedKeyCredentials sharedKeyCredentials;
-  private String xMsVersion = DECEMBER_2019_API_VERSION;
+  private ApiVersion xMsVersion = ApiVersion.getCurrentVersion();
   private final ExponentialRetryPolicy retryPolicy;
   private final String filesystem;
   private final AbfsConfiguration abfsConfiguration;
@@ -139,7 +139,7 @@ private AbfsClient(final URL baseUrl,
 
     if (encryptionContextProvider != null) {
       this.encryptionContextProvider = encryptionContextProvider;
-      xMsVersion = APRIL_2021_API_VERSION; // will be default once server change deployed
+      xMsVersion = ApiVersion.APR_10_2021; // will be default once server change deployed
       encryptionType = EncryptionType.ENCRYPTION_CONTEXT;
     } else if (abfsConfiguration.getEncodedClientProvidedEncryptionKey() != null) {
       clientProvidedEncryptionKey =
@@ -233,9 +233,9 @@ AbfsThrottlingIntercept getIntercept() {
     return intercept;
   }
 
-  List<AbfsHttpHeader> createDefaultHeaders() {
+  List<AbfsHttpHeader> createDefaultHeaders(ApiVersion xMsVersion) {
     final List<AbfsHttpHeader> requestHeaders = new ArrayList<AbfsHttpHeader>();
-    requestHeaders.add(new AbfsHttpHeader(X_MS_VERSION, xMsVersion));
+    requestHeaders.add(new AbfsHttpHeader(X_MS_VERSION, xMsVersion.toString()));
     requestHeaders.add(new AbfsHttpHeader(ACCEPT, APPLICATION_JSON
             + COMMA + SINGLE_WHITE_SPACE + APPLICATION_OCTET_STREAM));
     requestHeaders.add(new AbfsHttpHeader(ACCEPT_CHARSET,
@@ -305,7 +305,7 @@ AbfsUriQueryBuilder createDefaultUriQueryBuilder() {
 
   public AbfsRestOperation createFilesystem(TracingContext tracingContext)
       throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
 
     final AbfsUriQueryBuilder abfsUriQueryBuilder = new AbfsUriQueryBuilder();
     abfsUriQueryBuilder.addQuery(QUERY_PARAM_RESOURCE, FILESYSTEM);
@@ -320,7 +320,7 @@ public AbfsRestOperation createFilesystem(TracingContext tracingContext)
 
   public AbfsRestOperation setFilesystemProperties(final String properties,
       TracingContext tracingContext) throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
     // JDK7 does not support PATCH, so to work around the issue we will use
     // PUT and specify the real method in the X-Http-Method-Override header.
     requestHeaders.add(new AbfsHttpHeader(X_HTTP_METHOD_OVERRIDE,
@@ -345,7 +345,7 @@ public AbfsRestOperation setFilesystemProperties(final String properties,
   public AbfsRestOperation listPath(final String relativePath, final boolean recursive, final int listMaxResults,
                                     final String continuation, TracingContext tracingContext)
       throws IOException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
 
     final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
     abfsUriQueryBuilder.addQuery(QUERY_PARAM_RESOURCE, FILESYSTEM);
@@ -367,7 +367,7 @@ public AbfsRestOperation listPath(final String relativePath, final boolean recur
   }
 
   public AbfsRestOperation getFilesystemProperties(TracingContext tracingContext) throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
 
     final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
     abfsUriQueryBuilder.addQuery(QUERY_PARAM_RESOURCE, FILESYSTEM);
@@ -383,7 +383,7 @@ public AbfsRestOperation getFilesystemProperties(TracingContext tracingContext)
   }
 
   public AbfsRestOperation deleteFilesystem(TracingContext tracingContext) throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
 
     final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
     abfsUriQueryBuilder.addQuery(QUERY_PARAM_RESOURCE, FILESYSTEM);
@@ -434,7 +434,7 @@ public AbfsRestOperation createPath(final String path,
       final ContextEncryptionAdapter contextEncryptionAdapter,
       final TracingContext tracingContext)
       throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
     if (isFile) {
       addEncryptionKeyRequestHeaders(path, requestHeaders, true,
           contextEncryptionAdapter, tracingContext);
@@ -495,7 +495,7 @@ public AbfsRestOperation createPath(final String path,
   }
 
   public AbfsRestOperation acquireLease(final String path, int duration, TracingContext tracingContext) throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
 
     requestHeaders.add(new AbfsHttpHeader(X_MS_LEASE_ACTION, ACQUIRE_LEASE_ACTION));
     requestHeaders.add(new AbfsHttpHeader(X_MS_LEASE_DURATION, Integer.toString(duration)));
@@ -515,7 +515,7 @@ public AbfsRestOperation acquireLease(final String path, int duration, TracingCo
 
   public AbfsRestOperation renewLease(final String path, final String leaseId,
       TracingContext tracingContext) throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
 
     requestHeaders.add(new AbfsHttpHeader(X_MS_LEASE_ACTION, RENEW_LEASE_ACTION));
     requestHeaders.add(new AbfsHttpHeader(X_MS_LEASE_ID, leaseId));
@@ -534,7 +534,7 @@ public AbfsRestOperation renewLease(final String path, final String leaseId,
 
   public AbfsRestOperation releaseLease(final String path,
       final String leaseId, TracingContext tracingContext) throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
 
     requestHeaders.add(new AbfsHttpHeader(X_MS_LEASE_ACTION, RELEASE_LEASE_ACTION));
     requestHeaders.add(new AbfsHttpHeader(X_MS_LEASE_ID, leaseId));
@@ -553,7 +553,7 @@ public AbfsRestOperation releaseLease(final String path,
 
   public AbfsRestOperation breakLease(final String path,
       TracingContext tracingContext) throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
 
     requestHeaders.add(new AbfsHttpHeader(X_MS_LEASE_ACTION, BREAK_LEASE_ACTION));
     requestHeaders.add(new AbfsHttpHeader(X_MS_LEASE_BREAK_PERIOD, DEFAULT_LEASE_BREAK_PERIOD));
@@ -601,7 +601,7 @@ public AbfsClientRenameResult renamePath(
           boolean isMetadataIncompleteState,
           boolean isNamespaceEnabled)
       throws IOException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
 
     final boolean hasEtag = !isEmpty(sourceEtag);
 
@@ -797,7 +797,7 @@ public AbfsRestOperation append(final String path, final byte[] buffer,
       AppendRequestParameters reqParams, final String cachedSasToken,
       ContextEncryptionAdapter contextEncryptionAdapter, TracingContext tracingContext)
       throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
     addEncryptionKeyRequestHeaders(path, requestHeaders, false,
         contextEncryptionAdapter, tracingContext);
     if (reqParams.isExpectHeaderEnabled()) {
@@ -929,7 +929,7 @@ public AbfsRestOperation flush(final String path, final long position,
       final String cachedSasToken, final String leaseId,
       ContextEncryptionAdapter contextEncryptionAdapter, TracingContext tracingContext)
       throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
     addEncryptionKeyRequestHeaders(path, requestHeaders, false,
         contextEncryptionAdapter, tracingContext);
     // JDK7 does not support PATCH, so to workaround the issue we will use
@@ -962,7 +962,7 @@ public AbfsRestOperation flush(final String path, final long position,
   public AbfsRestOperation setPathProperties(final String path, final String properties,
                                              final TracingContext tracingContext, final ContextEncryptionAdapter contextEncryptionAdapter)
       throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
     addEncryptionKeyRequestHeaders(path, requestHeaders, false,
         contextEncryptionAdapter, tracingContext);
     // JDK7 does not support PATCH, so to workaround the issue we will use
@@ -990,7 +990,7 @@ public AbfsRestOperation getPathStatus(final String path,
       final boolean includeProperties, final TracingContext tracingContext,
       final ContextEncryptionAdapter contextEncryptionAdapter)
       throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
 
     final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
     String operation = SASTokenProvider.GET_PROPERTIES_OPERATION;
@@ -1027,7 +1027,7 @@ public AbfsRestOperation read(final String path,
       String cachedSasToken,
       ContextEncryptionAdapter contextEncryptionAdapter,
       TracingContext tracingContext) throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
     addEncryptionKeyRequestHeaders(path, requestHeaders, false,
         contextEncryptionAdapter, tracingContext);
     requestHeaders.add(new AbfsHttpHeader(RANGE,
@@ -1053,12 +1053,22 @@ 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 List<AbfsHttpHeader> requestHeaders
+        = (isPaginatedDelete(tracingContext, recursive)
+        && xMsVersion.compareTo(ApiVersion.AUG_03_2023) < 0)
+        ? createDefaultHeaders(ApiVersion.AUG_03_2023)
+        : createDefaultHeaders(xMsVersion);
     final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
+
+    if (isPaginatedDelete(tracingContext, recursive)) {
+      // 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;
@@ -1130,7 +1140,7 @@ public AbfsRestOperation deleteIdempotencyCheckOp(final AbfsRestOperation op) {
   public AbfsRestOperation setOwner(final String path, final String owner, final String group,
                                     TracingContext tracingContext)
       throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
     // JDK7 does not support PATCH, so to workaround the issue we will use
     // PUT and specify the real method in the X-Http-Method-Override header.
     requestHeaders.add(new AbfsHttpHeader(X_HTTP_METHOD_OVERRIDE,
@@ -1160,7 +1170,7 @@ public AbfsRestOperation setOwner(final String path, final String owner, final S
   public AbfsRestOperation setPermission(final String path, final String permission,
                                          TracingContext tracingContext)
       throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
     // JDK7 does not support PATCH, so to workaround the issue we will use
     // PUT and specify the real method in the X-Http-Method-Override header.
     requestHeaders.add(new AbfsHttpHeader(X_HTTP_METHOD_OVERRIDE,
@@ -1190,7 +1200,7 @@ public AbfsRestOperation setAcl(final String path, final String aclSpecString,
   public AbfsRestOperation setAcl(final String path, final String aclSpecString, final String eTag,
                                   TracingContext tracingContext)
       throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
     // JDK7 does not support PATCH, so to workaround the issue we will use
     // PUT and specify the real method in the X-Http-Method-Override header.
     requestHeaders.add(new AbfsHttpHeader(X_HTTP_METHOD_OVERRIDE,
@@ -1223,7 +1233,7 @@ public AbfsRestOperation getAclStatus(final String path, TracingContext tracingC
 
   public AbfsRestOperation getAclStatus(final String path, final boolean useUPN,
                                         TracingContext tracingContext) throws AzureBlobFileSystemException {
-    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+    final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders(xMsVersion);
 
     final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
     abfsUriQueryBuilder.addQuery(HttpQueryParams.QUERY_PARAM_ACTION, AbfsHttpConstants.GET_ACCESS_CONTROL);
@@ -1261,7 +1271,7 @@ public AbfsRestOperation checkAccess(String path, String rwx, TracingContext tra
         AbfsRestOperationType.CheckAccess,
         AbfsHttpConstants.HTTP_METHOD_HEAD,
         url,
-        createDefaultHeaders());
+        createDefaultHeaders(xMsVersion));
     op.execute(tracingContext);
     return op;
   }
@@ -1401,6 +1411,16 @@ private synchronized Boolean getIsNamespaceEnabled(TracingContext tracingContext
     return isNamespaceEnabled;
   }
 
+  protected Boolean getIsPaginatedDeleteEnabled() {
+    return abfsConfiguration.isPaginatedDeleteEnabled();
+  }
+
+  private Boolean isPaginatedDelete(TracingContext tracingContext,
+      boolean isRecursiveDelete) throws AzureBlobFileSystemException {
+    return getIsPaginatedDeleteEnabled()
+        && getIsNamespaceEnabled(tracingContext) && isRecursiveDelete;
+  }
+
   public AuthType getAuthType() {
     return authType;
   }
@@ -1501,7 +1521,7 @@ protected AbfsCounters getAbfsCounters() {
     return abfsCounters;
   }
 
-  public String getxMsVersion() {
+  public ApiVersion getxMsVersion() {
     return xMsVersion;
   }
 

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 "";
+  }
 }