HADOOP-19795: ABFS. GetPathStatus Optimization on OpenFileForRead

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

@@ -683,6 +683,10 @@ public class AbfsConfiguration{
           DefaultValue = DEFAULT_AZURE_READ_POLICY)
   private String abfsReadPolicy;
 
+  @BooleanConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_RESTRICT_GPS_ON_OPENFILE,
+  DefaultValue = DEFAULT_FS_AZURE_RESTRICT_GPS_ON_OPENFILE)
+  private boolean restrictGpsOnOpenFile;
+
   private String clientProvidedEncryptionKey;
   private String clientProvidedEncryptionKeySHA;
 
@@ -1445,6 +1449,14 @@ public String getAbfsReadPolicy() {
     return abfsReadPolicy;
   }
 
+/**
+   * Indicates whether GPS restriction on open file is enabled.
+   * @return true if GPS restriction is enabled on open file, false otherwise.
+   */
+  public boolean shouldRestrictGpsOnOpenFile() {
+    return restrictGpsOnOpenFile;
+  }
+
   /**
    * Enum config to allow user to pick format of x-ms-client-request-id header
    * @return tracingContextFormat config if valid, else default ALL_ID_FORMAT

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

@@ -152,6 +152,7 @@
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.CHAR_STAR;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.CHAR_UNDERSCORE;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.DIRECTORY;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.EMPTY_STRING;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.FILE;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ROOT_PATH;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.SINGLE_WHITE_SPACE;
@@ -564,7 +565,7 @@ public Hashtable<String, String> getPathStatus(final Path path,
 
   /**
    * Creates an object of {@link ContextEncryptionAdapter}
-   * from a file path. It calls {@link  org.apache.hadoop.fs.azurebfs.services.AbfsClient
+   * from a file path. It calls {@link org.apache.hadoop.fs.azurebfs.services.AbfsClient
    * #getPathStatus(String, boolean, TracingContext, EncryptionAdapter)} method to get
    * contextValue (x-ms-encryption-context) from the server. The contextValue is passed
    * to the constructor of EncryptionAdapter to create the required object of
@@ -878,6 +879,20 @@ public AbfsInputStream openFileForRead(final Path path,
         tracingContext);
   }
 
+  /**
+   * Creates an exception indicating that openFileForRead was called on a directory.
+   *
+   * @return AbfsRestOperationException with PATH_NOT_FOUND error code and a message
+   * indicating that openFileForRead must be used with files and not directories.
+   */
+  private AbfsRestOperationException openFileForReadDirectoryException() {
+    return new AbfsRestOperationException(
+            AzureServiceErrorCode.PATH_NOT_FOUND.getStatusCode(),
+            AzureServiceErrorCode.PATH_NOT_FOUND.getErrorCode(),
+            "openFileForRead must be used with files and not directories",
+            null);
+  }
+
   public AbfsInputStream openFileForRead(Path path,
       final Optional<OpenFileParameters> parameters,
       final FileSystem.Statistics statistics, TracingContext tracingContext)
@@ -890,66 +905,79 @@ public AbfsInputStream openFileForRead(Path path,
       FileStatus fileStatus = parameters.map(OpenFileParameters::getStatus)
           .orElse(null);
       String relativePath = getRelativePath(path);
-      String resourceType, eTag;
-      long contentLength;
+      String resourceType, eTag = EMPTY_STRING;
+      long contentLength = 0;
       ContextEncryptionAdapter contextEncryptionAdapter = NoContextEncryptionAdapter.getInstance();
       /*
       * GetPathStatus API has to be called in case of:
-      *   1.  fileStatus is null or not an object of VersionedFileStatus: as eTag
-      *       would not be there in the fileStatus object.
+      *   1.  restrictGpsOnOpenFile config is disabled AND fileStatus is null or not
+      *       an object of VersionedFileStatus: as eTag would not be there in the fileStatus object.
       *   2.  fileStatus is an object of VersionedFileStatus and the object doesn't
       *       have encryptionContext field when client's encryptionType is
       *       ENCRYPTION_CONTEXT.
       */
       if ((fileStatus instanceof VersionedFileStatus) && (
-          getClient().getEncryptionType() != EncryptionType.ENCRYPTION_CONTEXT
-              || ((VersionedFileStatus) fileStatus).getEncryptionContext()
-              != null)) {
+              getClient().getEncryptionType() != EncryptionType.ENCRYPTION_CONTEXT
+                      || ((VersionedFileStatus) fileStatus).getEncryptionContext()
+                      != null)) {
         path = path.makeQualified(this.uri, path);
         Preconditions.checkArgument(fileStatus.getPath().equals(path),
-            String.format(
-                "Filestatus path [%s] does not match with given path [%s]",
-                fileStatus.getPath(), path));
+                String.format(
+                        "Filestatus path [%s] does not match with given path [%s]",
+                        fileStatus.getPath(), path));
         resourceType = fileStatus.isFile() ? FILE : DIRECTORY;
         contentLength = fileStatus.getLen();
         eTag = ((VersionedFileStatus) fileStatus).getVersion();
         final String encryptionContext
-            = ((VersionedFileStatus) fileStatus).getEncryptionContext();
+                = ((VersionedFileStatus) fileStatus).getEncryptionContext();
         if (getClient().getEncryptionType() == EncryptionType.ENCRYPTION_CONTEXT) {
           contextEncryptionAdapter = new ContextProviderEncryptionAdapter(
-              getClient().getEncryptionContextProvider(), getRelativePath(path),
-              encryptionContext.getBytes(StandardCharsets.UTF_8));
+                  getClient().getEncryptionContextProvider(), getRelativePath(path),
+                  encryptionContext.getBytes(StandardCharsets.UTF_8));
         }
-      } else {
+        if (parseIsDirectory(resourceType)) {
+          throw openFileForReadDirectoryException();
+        }
+      }
+      /*
+       *  If file created with ENCRYPTION_CONTEXT, irrespective of whether isRestrictGpsOnOpenFile config is enabled or not,
+       *  GetPathStatus API has to be called to get the encryptionContext from the response header
+       */
+      else if (getClient().getEncryptionType() == EncryptionType.ENCRYPTION_CONTEXT
+              || !getAbfsConfiguration().shouldRestrictGpsOnOpenFile()) {
+
         AbfsHttpOperation op = getClient().getPathStatus(relativePath, false,
-            tracingContext, null).getResult();
-        resourceType = getClient().checkIsDir(op) ? DIRECTORY : FILE;
-        contentLength = extractContentLength(op);
-        eTag = op.getResponseHeader(HttpHeaderConfigurations.ETAG);
+                tracingContext, null).getResult();
         /*
          * For file created with ENCRYPTION_CONTEXT, client shall receive
          * encryptionContext from header field: X_MS_ENCRYPTION_CONTEXT.
          */
         if (getClient().getEncryptionType() == EncryptionType.ENCRYPTION_CONTEXT) {
           final String fileEncryptionContext = op.getResponseHeader(
-              HttpHeaderConfigurations.X_MS_ENCRYPTION_CONTEXT);
+                  X_MS_ENCRYPTION_CONTEXT);
           if (fileEncryptionContext == null) {
             LOG.debug("EncryptionContext missing in GetPathStatus response");
             throw new PathIOException(path.toString(),
-                "EncryptionContext not present in GetPathStatus response headers");
+                    "EncryptionContext not present in GetPathStatus response headers");
           }
           contextEncryptionAdapter = new ContextProviderEncryptionAdapter(
-              getClient().getEncryptionContextProvider(), getRelativePath(path),
-              fileEncryptionContext.getBytes(StandardCharsets.UTF_8));
+                  getClient().getEncryptionContextProvider(), getRelativePath(path),
+                  fileEncryptionContext.getBytes(StandardCharsets.UTF_8));
         }
-      }
+        resourceType = getClient().checkIsDir(op) ? DIRECTORY : FILE;
+        contentLength = extractContentLength(op);
+        eTag = op.getResponseHeader(HttpHeaderConfigurations.ETAG);
 
-      if (parseIsDirectory(resourceType)) {
-        throw new AbfsRestOperationException(
-            AzureServiceErrorCode.PATH_NOT_FOUND.getStatusCode(),
-            AzureServiceErrorCode.PATH_NOT_FOUND.getErrorCode(),
-            "openFileForRead must be used with files and not directories",
-            null);
+        if (parseIsDirectory(resourceType)) {
+          throw openFileForReadDirectoryException();
+        }
+      }
+      /* The only remaining case is:
+       * - restrictGpsOnOpenFile config is enabled with null FileStatus and encryptionType not as ENCRYPTION_CONTEXT
+       * In this case, we don't need to call GetPathStatus API.
+       */
+      else {
+        // do nothing
       }
 
       perfInfo.registerSuccess(true);
@@ -1015,6 +1043,7 @@ AZURE_FOOTER_READ_BUFFER_SIZE, getAbfsConfiguration().getFooterReadBufferSize())
         .withStreamStatistics(new AbfsInputStreamStatisticsImpl())
         .withShouldReadBufferSizeAlways(getAbfsConfiguration().shouldReadBufferSizeAlways())
         .withReadAheadBlockSize(getAbfsConfiguration().getReadAheadBlockSize())
+        .shouldRestrictGpsOnOpenFile(getAbfsConfiguration().shouldRestrictGpsOnOpenFile())
         .withBufferedPreadDisabled(bufferedPreadDisabled)
         .withEncryptionAdapter(contextEncryptionAdapter)
         .withAbfsBackRef(fsBackRef)

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

@@ -664,5 +664,11 @@ public static String containerProperty(String property, String fsName, String ac
    */
   public static final String FS_AZURE_TAIL_LATENCY_MAX_RETRY_COUNT = "fs.azure.tail.latency.max.retry.count";
 
+  /**
+   * If true, restricts GPS (getPathStatus) calls on openFileforRead
+   * Default: false
+   */
+  public static final String FS_AZURE_RESTRICT_GPS_ON_OPENFILE = "fs.azure.restrict.gps.on.openfile";
+
   private ConfigurationKeys() {}
 }

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

@@ -442,6 +442,7 @@ public final class FileSystemConfigurations {
   public static final int MIN_FS_AZURE_TAIL_LATENCY_ANALYSIS_WINDOW_GRANULARITY = 1;
   public static final int DEFAULT_FS_AZURE_TAIL_LATENCY_PERCENTILE_COMPUTATION_INTERVAL_MILLIS = 500;
   public static final int DEFAULT_FS_AZURE_TAIL_LATENCY_MAX_RETRY_COUNT = 1;
+  public static final boolean DEFAULT_FS_AZURE_RESTRICT_GPS_ON_OPENFILE = false;
 
   private FileSystemConfigurations() {}
 }

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

@@ -39,6 +39,7 @@ public final class HttpHeaderConfigurations {
   public static final String CONTENT_MD5 = "Content-MD5";
   public static final String CONTENT_TYPE = "Content-Type";
   public static final String RANGE = "Range";
+  public static final String CONTENT_RANGE = "Content-Range";
   public static final String TRANSFER_ENCODING = "Transfer-Encoding";
   public static final String USER_AGENT = "User-Agent";
   public static final String X_HTTP_METHOD_OVERRIDE = "X-HTTP-Method-Override";

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

@@ -66,6 +66,8 @@ public enum AzureServiceErrorCode {
   INVALID_APPEND_OPERATION("InvalidAppendOperation", HttpURLConnection.HTTP_CONFLICT, null),
   UNAUTHORIZED_BLOB_OVERWRITE("UnauthorizedBlobOverwrite", HttpURLConnection.HTTP_FORBIDDEN,
           "This request is not authorized to perform blob overwrites."),
+  INVALID_RANGE("InvalidRange", 416,
+          "The range specified is invalid for the current size of the resource."),
   UNKNOWN(null, -1, null);
 
   private final String errorCode;

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

@@ -68,7 +68,7 @@ protected int readOneBlock(final byte[] b, final int off, final int len) throws
     // If buffer is empty, then fill the buffer.
     if (getBCursor() == getLimit()) {
       // If EOF, then return -1
-      if (getFCursor() >= getContentLength()) {
+      if (!(shouldRestrictGpsOnOpenFile() && isFirstRead()) && getFCursor() >= getContentLength()) {
         return -1;
       }
 
@@ -83,7 +83,11 @@ protected int readOneBlock(final byte[] b, final int off, final int len) throws
 
       // Reset Read Type back to normal and set again based on code flow.
       getTracingContext().setReadType(ReadType.NORMAL_READ);
-      if (shouldAlwaysReadBufferSize()) {
+      if(shouldRestrictGpsOnOpenFile() && isFirstRead()) {
+        LOG.debug("RestrictGpsOnOpenFile is enabled. Skip readahead for first read.");
+        bytesRead = readInternal(getFCursor(), getBuffer(), 0, getBufferSize(), true);
+      }
+      else if (shouldAlwaysReadBufferSize()) {
         bytesRead = readInternal(getFCursor(), getBuffer(), 0, getBufferSize(), false);
       } else {
         // Enable readAhead when reading sequentially