HADOOP-19795: ABFS. GetPathStatus Optimization on OpenFileForRead#8212
HADOOP-19795: ABFS. GetPathStatus Optimization on OpenFileForRead#8212manika137 wants to merge 17 commits intoapache:trunkfrom
Conversation
Test Results============================================================
|
|
💔 -1 overall
This message was automatically generated. |
| getClient().getEncryptionType() != EncryptionType.ENCRYPTION_CONTEXT | ||
| || ((VersionedFileStatus) fileStatus).getEncryptionContext() | ||
| != null)) { | ||
| getClient().getEncryptionType() != EncryptionType.ENCRYPTION_CONTEXT |
There was a problem hiding this comment.
additional space changes can be reverted
| contextEncryptionAdapter = new ContextProviderEncryptionAdapter( | ||
| getClient().getEncryptionContextProvider(), getRelativePath(path), | ||
| fileEncryptionContext.getBytes(StandardCharsets.UTF_8)); | ||
| getClient().getEncryptionContextProvider(), getRelativePath(path), |
| encryptionContext.getBytes(StandardCharsets.UTF_8)); | ||
| } | ||
| } else { | ||
| if (parseIsDirectory(resourceType)) { |
There was a problem hiding this comment.
Can be moved to common part as is getting checked in both the cases
| * - 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 { |
There was a problem hiding this comment.
Won't this lead to going ahead and opening the stream without checks? Do we fail later for this case ?
There was a problem hiding this comment.
Yes, the checks with this config would be happening after the first read then. If the read fails there then we throw the appropriate exception
| 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, |
There was a problem hiding this comment.
416 should come from a constant defined in HttpURLConnection class
There was a problem hiding this comment.
We dont have a 416 defined in HttpURLConnection
There was a problem hiding this comment.
we can define it in our constants class then
There was a problem hiding this comment.
Yes, let's define in AbfsHttpConstants
| // Reset Read Type back to normal and set again based on code flow. | ||
| getTracingContext().setReadType(ReadType.NORMAL_READ); | ||
| if (shouldAlwaysReadBufferSize()) { | ||
| if(shouldRestrictGpsOnOpenFile() && isFirstRead()) { |
There was a problem hiding this comment.
nit: space after if
| // Reset Read Type back to normal and set again based on code flow. | ||
| getTracingContext().setReadType(ReadType.NORMAL_READ); | ||
| if (shouldAlwaysReadBufferSize()) { | ||
| if(shouldRestrictGpsOnOpenFile() && isFirstRead()) { |
There was a problem hiding this comment.
add a comment for this condition as well
| private final int footerReadSize; // default buffer size to read when reading footer | ||
| private final int readAheadQueueDepth; // initialized in constructor | ||
| private final String eTag; // eTag of the path when InputStream are created | ||
| private String eTag; // eTag of the path when InputStream are created |
There was a problem hiding this comment.
nit: InputStream is created
| } | ||
| } | ||
|
|
||
| String getRelativePath(final Path path) { |
| tracingContext, | ||
| contextEncryptionAdapter).getResult(); | ||
|
|
||
| String resourceType = |
There was a problem hiding this comment.
use client.checkIsDir method which handles null case as well
| } | ||
| contentLength = Long.parseLong(op.getResult().getResponseHeader(HttpHeaderConfigurations.CONTENT_RANGE). | ||
| split(AbfsHttpConstants.FORWARD_SLASH)[1]); | ||
| eTag = op.getResult().getResponseHeader("ETag"); |
There was a problem hiding this comment.
use extractEtagHeader method
| if (Objects.equals(resourceType, DIRECTORY)) { | ||
| throw directoryReadException(); | ||
| } | ||
| contentLength = Long.parseLong(op.getResult().getResponseHeader(HttpHeaderConfigurations.CONTENT_RANGE). |
There was a problem hiding this comment.
same use extractContentLen method
There was a problem hiding this comment.
We're using content range here instead of content length to extract the actual length. This is to avoid partial reads length. Added a new method for it
| if (ere.getStatusCode() == HttpURLConnection.HTTP_NOT_FOUND) { | ||
| throw new FileNotFoundException(ere.getMessage()); | ||
| int status = ere.getStatusCode(); | ||
| if(ere.getErrorMessage().contains(readOnDirectoryErrorMsg)){ |
There was a problem hiding this comment.
nit: space after if
|
|
||
| } catch (AzureBlobFileSystemException gpsEx) { | ||
| AbfsRestOperationException gpsEre = (AbfsRestOperationException) gpsEx; | ||
| if(gpsEre.getErrorMessage().contains(readOnDirectoryErrorMsg)){ |
There was a problem hiding this comment.
nit: space after if
| } | ||
|
|
||
| // Default: propagate original error | ||
| throw new IOException(ex); |
There was a problem hiding this comment.
should be done once only on line 715
| This is required since contentLength is not available yet to determine prefetch block size. | ||
| */ | ||
| bytesRead = readInternal(getFCursor(), getBuffer(), 0, getBufferSize(), false); | ||
| if(shouldRestrictGpsOnOpenFile() && isFirstRead()) { |
There was a problem hiding this comment.
nit: space after if
|
Sure. Will ensure functionality for all edge cases including file not found with wrong filestatus type or without filestatus. Will document the expectation. |
|
💔 -1 overall
This message was automatically generated. |
| if (getBCursor() == getLimit()) { | ||
| // If EOF, then return -1 | ||
| if (getFCursor() >= getContentLength()) { | ||
| if (!(shouldRestrictGpsOnOpenFile() && isFirstRead()) && getFCursor() >= getContentLength()) { |
There was a problem hiding this comment.
Should we decouple !(shouldRestrictGpsOnOpenFile() && isFirstRead()) from the other conditions?
Currently, if both shouldRestrictGpsOnOpenFile() and isFirstRead() are true, the entire expression evaluates to false. This prevents the function from returning -1 even if getFCursor() >= getContentLength() is true, allowing the execution to proceed incorrectly.
There was a problem hiding this comment.
As discussed, for first reads we have other validation in place
| AbfsRestOperation statusOp = null; | ||
| try { | ||
| // Check if the file already exists by calling GetPathStatus | ||
| //TODO: GBP |
There was a problem hiding this comment.
Remove Todo if it is taken care
| // If the overwrite flag is true, we must verify whether an empty directory exists at the specified path. | ||
| // However, if overwrite is false, we can skip this validation and proceed with blob creation, | ||
| // which will fail with a conflict error if a file or directory already exists at the path. | ||
| //todo: GBP |
| String resourceType = result.getResponseHeader( | ||
| HttpHeaderConfigurations.X_MS_RESOURCE_TYPE); | ||
| return StringUtils.equalsIgnoreCase(resourceType, DIRECTORY); | ||
| return resourceType != null && StringUtils.equalsIgnoreCase(resourceType, DIRECTORY); |
There was a problem hiding this comment.
This check is not needed here as equalsIgnoreCase internally checks if resourceType is null or not.
There was a problem hiding this comment.
As discussed, returns NPE if not present
| } | ||
| } | ||
|
|
||
| private long extractContentLength(AbfsHttpOperation op) { |
| /** ABFS instance to be held by the input stream to avoid GC close. */ | ||
| private final BackReference fsBackRef; | ||
| private final ReadBufferManager readBufferManager; | ||
| private final String readOnDirectoryErrorMsg = "Read operation not permitted on a directory."; |
There was a problem hiding this comment.
We should define this error in AbfsErrors file
| if (getBCursor() == getLimit()) { | ||
| // If EOF, then return -1 | ||
| if (getFCursor() >= getContentLength()) { | ||
| if (!(shouldRestrictGpsOnOpenFile() && isFirstRead()) && getFCursor() >= getContentLength()) { |
There was a problem hiding this comment.
As discussed, for first reads we have other validation in place
| return new AbfsRestOperationException( | ||
| AzureServiceErrorCode.PATH_NOT_FOUND.getStatusCode(), | ||
| AzureServiceErrorCode.PATH_NOT_FOUND.getErrorCode(), | ||
| "openFileForRead must be used with files and not directories", |
There was a problem hiding this comment.
Define this error String in AbfsErrors file
| getClient().getEncryptionType() != EncryptionType.ENCRYPTION_CONTEXT | ||
| || ((VersionedFileStatus) fileStatus).getEncryptionContext() | ||
| != null)) { | ||
| || ((VersionedFileStatus) fileStatus).getEncryptionContext() |
There was a problem hiding this comment.
This change can we reverted
| String.format( | ||
| "Filestatus path [%s] does not match with given path [%s]", | ||
| fileStatus.getPath(), path)); | ||
| "Filestatus path [%s] does not match with given path [%s]", |
There was a problem hiding this comment.
This change can we reverted
| * @param path the {@link Path} to convert; must not be null | ||
| * @return the relative path as a {@link String}; never null | ||
| */ | ||
| String getRelativePath(final Path path) { |
There was a problem hiding this comment.
store class already has this method, can we not define it in a common place?
There was a problem hiding this comment.
Made the method static
| private long extractContentLength(AbfsHttpOperation op) { | ||
| // We need to use content range header instead of content length to take care of partial reads | ||
| String contentRange = op.getResponseHeader(HttpHeaderConfigurations.CONTENT_RANGE); | ||
| if (!StringUtils.isEmpty(contentRange)) { |
There was a problem hiding this comment.
can contentRange header be null ?
There was a problem hiding this comment.
if null, isEmpty method takes care of it and returns true
| if (!StringUtils.isEmpty(contentRange)) { | ||
| contentLength = Long.parseLong(contentRange.split(AbfsHttpConstants.FORWARD_SLASH)[1]); | ||
| } | ||
| else { |
There was a problem hiding this comment.
better to initialize contentLength with 0 and return as is instead of else
| throw ere; | ||
| } | ||
| boolean isHnsEnabled = client.getIsNamespaceEnabled(); | ||
|
|
There was a problem hiding this comment.
extra lines can be removed
|
💔 -1 overall
This message was automatically generated. |
anujmodi2021
left a comment
anujmodi2021
left a comment
There was a problem hiding this comment.
We should add some integration tests as well IMO
| 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, |
There was a problem hiding this comment.
Yes, let's define in AbfsHttpConstants
| if (shouldAlwaysReadBufferSize()) { | ||
|
|
||
| // If restrictGpsOnOpenFile config is enabled, skip prefetch for the first read since contentLength | ||
| // is not available yet to determine prefetch block size. |
There was a problem hiding this comment.
Comment seem to be misleading. prefetch block size is configurable, this should say "yet to determine how much to prefetch" maybe
There was a problem hiding this comment.
Modified the comment
| * | ||
| * @return an {@link AbfsRestOperationException} indicating the operation is not permitted on a directory | ||
| */ | ||
| private IOException directoryReadException() { |
There was a problem hiding this comment.
This should be an UnsupportedOperationException wrapping the server returned exception?
|
💔 -1 overall
This message was automatically generated. |
6 participants
Description of PR
JIRA: https://issues.apache.org/jira/browse/HADOOP-19795
We do a getPathStatus call during file open for read. This call is primarily used to fetch the file’s metadata properties before the actual read begins.
We are now introducing an optional, config-driven read flow that avoids the getPathStatus call during open and instead derives required metadata from the read response itself.
How was this patch tested?
New tests were added and test suite was run. Adding the results in comments below