HADOOP-19139.No GetPathStatus for opening AbfsInputStream

hadoop-common-project/hadoop-common/src/site/markdown/filesystem/filesystem.md

@@ -817,7 +817,9 @@ Implementations without a compliant call SHOULD throw `UnsupportedOperationExcep
 
 This is a critical precondition. Implementations of some FileSystems (e.g.
 Object stores) could shortcut one round trip by postponing their HTTP GET
-operation until the first `read()` on the returned `FSDataInputStream`.
+operation until the first `read()` on the returned `FSDataInputStream`. This
+lazy open may not surface file non-existence or access permission failures
+until the first `read()` of the actual data.
 However, much client code does depend on the existence check being performed
 at the time of the `open()` operation. Implementations MUST check for the
 presence of the file at the time of creation. This does not imply that

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

@@ -367,6 +367,14 @@ public class AbfsConfiguration{
       FS_AZURE_ENABLE_PAGINATED_DELETE, DefaultValue = DEFAULT_ENABLE_PAGINATED_DELETE)
   private boolean isPaginatedDeleteEnabled;
 
+  @BooleanConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_INPUT_STREAM_LAZY_OPEN_OPTIMIZATION_ENABLED,
+      DefaultValue = DEFAULT_HEAD_OPTIMIZATION_INPUT_STREAM)
+  private boolean isInputStreamLazyOptimizationEnabled;
+
+  @BooleanConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_PREFETCH_ON_FIRST_READ_ENABLED,
+      DefaultValue = DEFAULT_PREFETCH_READAHEAD_ON_FIRST_READ)
+  private boolean isPrefetchOnFirstReadEnabled;
+
   private String clientProvidedEncryptionKey;
   private String clientProvidedEncryptionKeySHA;
 
@@ -1247,6 +1255,14 @@ public boolean getIsChecksumValidationEnabled() {
     return isChecksumValidationEnabled;
   }
 
+  public boolean isInputStreamLazyOptimizationEnabled() {
+    return isInputStreamLazyOptimizationEnabled;
+  }
+
+  public boolean isPrefetchOnFirstReadEnabled() {
+    return isPrefetchOnFirstReadEnabled;
+  }
+
   @VisibleForTesting
   public void setIsChecksumValidationEnabled(boolean isChecksumValidationEnabled) {
     this.isChecksumValidationEnabled = isChecksumValidationEnabled;

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

@@ -814,16 +814,17 @@ public AbfsInputStream openFileForRead(Path path,
       LOG.debug("openFileForRead filesystem: {} path: {}",
           client.getFileSystem(), path);
 
-      FileStatus fileStatus = parameters.map(OpenFileParameters::getStatus)
+      final FileStatus fileStatus = parameters.map(OpenFileParameters::getStatus)
           .orElse(null);
       String relativePath = getRelativePath(path);
-      String resourceType, eTag;
-      long contentLength;
+      String resourceType = null, eTag = null;
+      long contentLength = -1;
       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.
+      *       would not be there in the fileStatus object. This shall be called
+      *       only if inputStream's lazy optimization is disabled.
       *   2.  fileStatus is an object of VersionedFileStatus and the object doesn't
       *       have encryptionContext field when client's encryptionType is
       *       ENCRYPTION_CONTEXT.
@@ -848,32 +849,37 @@ public AbfsInputStream openFileForRead(Path path,
               encryptionContext.getBytes(StandardCharsets.UTF_8));
         }
       } else {
-        AbfsHttpOperation op = client.getPathStatus(relativePath, false,
-            tracingContext, null).getResult();
-        resourceType = op.getResponseHeader(
-            HttpHeaderConfigurations.X_MS_RESOURCE_TYPE);
-        contentLength = Long.parseLong(
-            op.getResponseHeader(HttpHeaderConfigurations.CONTENT_LENGTH));
-        eTag = op.getResponseHeader(HttpHeaderConfigurations.ETAG);
-        /*
-         * For file created with ENCRYPTION_CONTEXT, client shall receive
-         * encryptionContext from header field: X_MS_ENCRYPTION_CONTEXT.
-         */
-        if (client.getEncryptionType() == EncryptionType.ENCRYPTION_CONTEXT) {
-          final String fileEncryptionContext = op.getResponseHeader(
-              HttpHeaderConfigurations.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");
+        if (client.getEncryptionType() == EncryptionType.ENCRYPTION_CONTEXT
+            || !abfsConfiguration.isInputStreamLazyOptimizationEnabled()) {
+          final AbfsHttpOperation op = client.getPathStatus(relativePath, false,
+              tracingContext, null).getResult();
+          resourceType = op.getResponseHeader(
+              HttpHeaderConfigurations.X_MS_RESOURCE_TYPE);
+          contentLength = Long.parseLong(
+              op.getResponseHeader(HttpHeaderConfigurations.CONTENT_LENGTH));
+          eTag = op.getResponseHeader(HttpHeaderConfigurations.ETAG);
+
+          /*
+           * For file created with ENCRYPTION_CONTEXT, client shall receive
+           * encryptionContext from header field: X_MS_ENCRYPTION_CONTEXT.
+           */
+          if (client.getEncryptionType() == EncryptionType.ENCRYPTION_CONTEXT) {
+            final String fileEncryptionContext = op.getResponseHeader(
+                HttpHeaderConfigurations.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");
+            }
+            contextEncryptionAdapter = new ContextProviderEncryptionAdapter(
+                client.getEncryptionContextProvider(), getRelativePath(path),
+                fileEncryptionContext.getBytes(StandardCharsets.UTF_8));
           }
-          contextEncryptionAdapter = new ContextProviderEncryptionAdapter(
-              client.getEncryptionContextProvider(), getRelativePath(path),
-              fileEncryptionContext.getBytes(StandardCharsets.UTF_8));
         }
       }
 
-      if (parseIsDirectory(resourceType)) {
+      if ((fileStatus != null || !abfsConfiguration.isInputStreamLazyOptimizationEnabled())
+          && parseIsDirectory(resourceType)) {
         throw new AbfsRestOperationException(
             AzureServiceErrorCode.PATH_NOT_FOUND.getStatusCode(),
             AzureServiceErrorCode.PATH_NOT_FOUND.getErrorCode(),
@@ -916,6 +922,8 @@ AZURE_FOOTER_READ_BUFFER_SIZE, getAbfsConfiguration().getFooterReadBufferSize())
             .withBufferedPreadDisabled(bufferedPreadDisabled)
             .withEncryptionAdapter(contextEncryptionAdapter)
             .withAbfsBackRef(fsBackRef)
+            .withPrefetchTriggerOnFirstRead(
+                abfsConfiguration.isPrefetchOnFirstReadEnabled())
             .build();
   }
 

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

@@ -88,6 +88,7 @@ public final class AbfsHttpConstants {
   public static final String HTTP_HEADER_PREFIX = "x-ms-";
   public static final String HASH = "#";
   public static final String TRUE = "true";
+  public static final String FALSE = "false";
 
   public static final String PLUS_ENCODE = "%20";
   public static final String FORWARD_SLASH_ENCODE = "%2F";
@@ -165,5 +166,7 @@ public static ApiVersion getCurrentVersion() {
    */
   public static final Integer HTTP_STATUS_CATEGORY_QUOTIENT = 100;
 
+  public static final Integer READ_PATH_REQUEST_NOT_SATISFIABLE = 416;
+
   private AbfsHttpConstants() {}
 }

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

@@ -309,5 +309,21 @@ public static String accountProperty(String property, String account) {
    * @see FileSystem#openFile(org.apache.hadoop.fs.Path)
    */
   public static final String FS_AZURE_BUFFERED_PREAD_DISABLE = "fs.azure.buffered.pread.disable";
+
+  /**
+   * Enable lazy opening of an inputStream. Lazy opening would not call HEAD call
+   * to get file metadata before creating inputStream. ReadPath API of server
+   * would give the contentLength and eTag which would be used in subsequent calls
+   * for if-match headers.
+   */
+  public static final String
+      FS_AZURE_INPUT_STREAM_LAZY_OPEN_OPTIMIZATION_ENABLED = "fs.azure.input.stream.lazy.open.optimization.enabled";
+
+  /**
+   * Enable prefetch on the first read to {@link org.apache.hadoop.fs.azurebfs.services.AbfsInputStream}.
+   * If disabled, first call would not trigger prefetch. Prefetch would be switched on
+   * after first read call.
+   */
+  public static final String FS_AZURE_PREFETCH_ON_FIRST_READ_ENABLED = "fs.azure.prefetch.on.first.read.enabled";
   private ConfigurationKeys() {}
 }

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

@@ -161,5 +161,9 @@ public final class FileSystemConfigurations {
    */
   public static final int RATE_LIMIT_DEFAULT = 10_000;
 
+  public static final boolean DEFAULT_HEAD_OPTIMIZATION_INPUT_STREAM = false;
+
+  public static final boolean DEFAULT_PREFETCH_READAHEAD_ON_FIRST_READ = true;
+
   private FileSystemConfigurations() {}
 }

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

@@ -34,6 +34,7 @@ public final class HttpHeaderConfigurations {
   public static final String IF_MATCH = "If-Match";
   public static final String IF_NONE_MATCH = "If-None-Match";
   public static final String CONTENT_LENGTH = "Content-Length";
+  public static final String CONTENT_RANGE = "Content-Range";
   public static final String CONTENT_ENCODING = "Content-Encoding";
   public static final String CONTENT_LANGUAGE = "Content-Language";
   public static final String CONTENT_MD5 = "Content-MD5";

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/exceptions/AbfsRestOperationException.java

@@ -34,6 +34,7 @@ public class AbfsRestOperationException extends AzureBlobFileSystemException {
   private final int statusCode;
   private final AzureServiceErrorCode errorCode;
   private final String errorMessage;
+  private final AbfsHttpOperation abfsHttpOperation;
 
   public AbfsRestOperationException(
       final int statusCode,
@@ -45,6 +46,7 @@ public AbfsRestOperationException(
     this.statusCode = statusCode;
     this.errorCode = AzureServiceErrorCode.getAzureServiceCode(this.statusCode, errorCode);
     this.errorMessage = errorMessage;
+    this.abfsHttpOperation = null;
   }
 
   public AbfsRestOperationException(
@@ -55,6 +57,7 @@ public AbfsRestOperationException(
       final AbfsHttpOperation abfsHttpOperation) {
     super(formatMessage(abfsHttpOperation), innerException);
 
+    this.abfsHttpOperation = abfsHttpOperation;
     this.statusCode = statusCode;
     this.errorCode = AzureServiceErrorCode.getAzureServiceCode(this.statusCode, errorCode);
     this.errorMessage = errorMessage;
@@ -66,6 +69,7 @@ public AbfsRestOperationException(final HttpException innerException) {
     this.statusCode = innerException.getHttpErrorCode();
     this.errorCode = AzureServiceErrorCode.UNKNOWN;
     this.errorMessage = innerException.getMessage();
+    this.abfsHttpOperation = null;
   }
 
   public int getStatusCode() {
@@ -80,6 +84,10 @@ public String getErrorMessage() {
     return this.errorMessage;
   }
 
+  public AbfsHttpOperation getAbfsHttpOperation() {
+    return this.abfsHttpOperation;
+  }
+
   private static String formatMessage(final AbfsHttpOperation abfsHttpOperation) {
     // HEAD request response doesn't have StorageErrorCode, StorageErrorMessage.
     if (abfsHttpOperation.getMethod().equals("HEAD")) {

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

@@ -36,6 +36,7 @@
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsInvalidChecksumException;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsDriverException;
@@ -1099,7 +1100,9 @@ public AbfsRestOperation read(final String path,
     AbfsHttpHeader rangeHeader = new AbfsHttpHeader(RANGE,
         String.format("bytes=%d-%d", position, position + bufferLength - 1));
     requestHeaders.add(rangeHeader);
-    requestHeaders.add(new AbfsHttpHeader(IF_MATCH, eTag));
+    if (!StringUtils.isEmpty(eTag)) {
+      requestHeaders.add(new AbfsHttpHeader(IF_MATCH, eTag));
+    }
 
     // Add request header to fetch MD5 Hash of data returned by server.
     if (isChecksumValidationEnabled(requestHeaders, rangeHeader, bufferLength)) {