HADOOP-1593. [ABFS] Add vectored read support in ABFS driver #8400
Open
anmolanmol1234 wants to merge 33 commits intoapache:trunkfrom
Open
HADOOP-1593. [ABFS] Add vectored read support in ABFS driver #8400anmolanmol1234 wants to merge 33 commits intoapache:trunkfrom
anmolanmol1234 wants to merge 33 commits intoapache:trunkfrom
Conversation
… HADOOP-15963_poc
… HADOOP-15963_poc
… HADOOP-15963_poc
… HADOOP-15963_poc
… HADOOP-15963_poc
… HADOOP-15963_poc
|
🎊 +1 overall
This message was automatically generated. |
|
🎊 +1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
|
🎊 +1 overall
This message was automatically generated. |
|
🎊 +1 overall
This message was automatically generated. |
manika137
reviewed
Apr 11, 2026
| * completed in doneReading once actual bytes are known. | ||
| * Short-read safety is enforced there via per-unit coverage check. | ||
| */ | ||
| return true; |
Contributor
There was a problem hiding this comment.
this is not only the READING_IN_PROGRESS case right- incase the buffer is UNAVAILABLE (queued but not picked up by any thread yet) it would also land here.
Contributor
Author
There was a problem hiding this comment.
yes, that will also land here, will update the comments. In case of doneReading we validate that it has read the needed data for the vectored case.
manika137
reviewed
Apr 11, 2026
| * applied later in doneReading before dispatching completion. | ||
| */ | ||
| long end = existing.getOffset() + ( | ||
| existing.getStatus() == ReadBufferStatus.AVAILABLE |
Contributor
There was a problem hiding this comment.
So we would wait and attach the buffer even if its not yet picked up by thread yet (unavailable state)?
Contributor
Author
There was a problem hiding this comment.
doneReading is where the actual handling of vectored reads happens, so if the buffer is already queued it is safe to attach the buffer here and later do the handling in doneReading.
manika137
reviewed
Apr 11, 2026
| return System.identityHashCode(range); | ||
| } | ||
| } | ||
| } No newline at end of file |
manika137
reviewed
Apr 11, 2026
| @@ -1078,7 +1105,7 @@ public int minSeekForVectorReads() { | |||
| */ | |||
| @Override | |||
| public int maxReadSizeForVectorReads() { | |||
Contributor
There was a problem hiding this comment.
the method mentions the read size but we're returning the max gap size
Contributor
Author
There was a problem hiding this comment.
this is the name of the method in the superclass
Contributor
There was a problem hiding this comment.
But why are we returning the seek size here? Why not actual read size?
manika137
reviewed
Apr 11, 2026
| buffer.getPath(), r.getOffset(), destOffset, length, left); | ||
|
|
||
| if (left < 0) { | ||
| LOG.error("fanOut: pending bytes went negative possible duplicate write:" |
Contributor
There was a problem hiding this comment.
nit: non-printable character
manika137
reviewed
Apr 11, 2026
| if (end >= unit.getOffset() + unit.getLength()) { | ||
| existing.setBufferType(BufferType.VECTORED); | ||
| existing.addVectoredUnit(unit); | ||
| existing.setAllocator(allocator); |
Contributor
There was a problem hiding this comment.
could we have multiple readvector calls with overlapping ranges that could reset the allocator here? Or isAlreadyQueued section wont go through for overlapping ranges?
Contributor
Author
There was a problem hiding this comment.
Overlapping ranges are not allowed in vectored reads. validateAndSortRanges in VectoredReadUtils class takes care of this.
manika137
reviewed
Apr 11, 2026
| return bufferManager; | ||
| } | ||
|
|
||
| VectoredReadHandler getVectoredReadHandler() { |
manika137
reviewed
Apr 11, 2026
| long bufferEnd = bufferStart + bytesRead; | ||
|
|
||
| /* Iterate over all combined logical units mapped to this buffer */ | ||
| for (CombinedFileRange unit : units) { |
Contributor
There was a problem hiding this comment.
could the following scenario be possible: while we fan-out here, we have another vectorRead call come in overlapping the ranges shared by this buffer and getting attached to the same buffer as a unit
Contributor
Author
There was a problem hiding this comment.
Overlapping ranges are not allowed, List<? extends FileRange> sortedRanges = VectoredReadUtils.validateAndSortRanges(
ranges, Optional.of(fileLength)); this takes care of that
manika137
reviewed
Apr 11, 2026
| */ | ||
| if (isAlreadyQueued(stream.getETag(), unit.getOffset())) { | ||
| ReadBuffer existing = findQueuedBuffer(stream, unit.getOffset()); | ||
| if (existing != null && existing.getStream().getETag() != null && stream.getETag() |
Contributor
There was a problem hiding this comment.
same doubt as RBMV1, do we wait for UNAVAILABLE state readbuffers too?
Contributor
Author
There was a problem hiding this comment.
yes addressed above
manika137
reviewed
Apr 11, 2026
| * @param abfsConfiguration the configuration to set for the ReadBufferManagerV2. | ||
| */ | ||
| public static void setReadBufferManagerConfigs(final int readAheadBlockSize, | ||
| public static void setReadBufferManagerConfigs(int readAheadBlockSize, |
|
💔 -1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
| return isFirstByteConsumed() && isLastByteConsumed(); | ||
| } | ||
|
|
||
| void addVectoredUnit(CombinedFileRange u) { |
Contributor
There was a problem hiding this comment.
Java doc for all the newly created methods
| // Allocator used for vectored fan-out; captured at queue time */ | ||
| private IntFunction<ByteBuffer> allocator; | ||
| // Tracks whether fanOut has already been executed | ||
| private final AtomicInteger fanOutDone = new AtomicInteger(0); |
Contributor
There was a problem hiding this comment.
Would it be better to keep fanOutDone as AtomicBoolean instead of AtomicInteger? We don't have to compare the value in isFanOutDone() in that case.
| ReadBuffer findInList(final Collection<ReadBuffer> buffers, | ||
| final AbfsInputStream stream, long requestedOffset) { | ||
| for (ReadBuffer buffer : buffers) { | ||
| if (buffer.getStream() == stream |
Contributor
There was a problem hiding this comment.
can buffer be null, if yes it will result into null pointer exception here.
Contributor
Author
There was a problem hiding this comment.
No buffer can't be null here
Contributor
Author
|
@steveloughran @mukund-thakur requesting you to kindly review the PR. Thanks |
|
🎊 +1 overall
This message was automatically generated. |
Contributor
steveloughran
left a comment
steveloughran
left a comment
There was a problem hiding this comment.
have a look at #7105 / HADOOP-19105. Improve resilience in vector reads. https://issues.apache.org/jira/browse/HADOOP-19105
to see why I now think trying to merge ranges is a PITA.
do make sure that your error handling explicitly releases any allocated buffers before raising exceptions; the PositionedReadable interface was extended to support this
| @@ -1184,7 +1209,7 @@ ReadBufferManager getReadBufferManager() { | |||
| */ | |||
| @Override | |||
| public int minSeekForVectorReads() { | |||
Contributor
There was a problem hiding this comment.
I'm actually going to recommend not coalescing vectors unless you have a good strategy to deal with partial failures, memory releases, retries etc.
I did try to do that in s3a and gave up, even after extending the api to allow a "release" operation to be passed (which parquet passes down FWIW).
We've never seen failures with s3 in production and coalesced ranges as parquet/orc rowgroups are too far apart. So it's better to focus on retry and recovery there than range coalescing
Contributor
Author
There was a problem hiding this comment.
In case of Azure, we do 4 MB buffer reads, so if don't do coalescing, we are wasting a lot of read data per range and hence we found that merging ranges which fall in one buffer i.e. 4 MB is giving better results
| * Performs a vectored direct read by fetching multiple non-contiguous | ||
| * ranges in a single operation. | ||
| */ | ||
| VECTORED_DIRECT_READ("VDR"), |
Contributor
There was a problem hiding this comment.
azure does multirange? nice
Contributor
Author
There was a problem hiding this comment.
Here this direct read is when we are not able to queue a vectored read in the read ahead queue due to read ahead buffers not being available, we do a direct readRemote call
Contributor
Author
Have tried to take care of this in the VectoredReadHandler fanOut and directRead methods, can you please check if you see any concerns with the code there. |
| @@ -1078,7 +1105,7 @@ public int minSeekForVectorReads() { | |||
| */ | |||
| @Override | |||
| public int maxReadSizeForVectorReads() { | |||
Contributor
There was a problem hiding this comment.
But why are we returning the seek size here? Why not actual read size?
| CombinedFileRange u = it.next(); | ||
| if ((u.getOffset() + u.getLength()) > actualEnd) { | ||
| it.remove(); | ||
| LOGGER.debug( |
Contributor
There was a problem hiding this comment.
I think we should use Trace logs at most places and debug only at critical spots and errors?
| buffer.setBuffer(bufferPool[bufferIndex]); | ||
| buffer.setBufferindex(bufferIndex); | ||
|
|
||
| getReadAheadQueue().add(buffer); |
Contributor
There was a problem hiding this comment.
Please add trce logging here and other places as done for normal queuing and RBMV1
| "ReadBufferWorker completed prefetch for file: {} with eTag: {}, for offset: {}, queued by stream: {}, with status: {} and bytes read: {}", | ||
| buffer.getPath(), buffer.getETag(), buffer.getOffset(), | ||
| buffer.getStream().hashCode(), result, bytesActuallyRead); | ||
| if (LOGGER.isTraceEnabled()) { |
Contributor
There was a problem hiding this comment.
Please use printTraceLog(), it has check inside method
| final long requestedOffset) { | ||
| for (ReadBuffer buffer : list) { | ||
| if (eTag.equals(buffer.getETag())) { | ||
| if (eTag != null && eTag.equals(buffer.getETag())) { |
Contributor
There was a problem hiding this comment.
We can use Objects.equals(). It will have null check
| // Buffer is returned if the requestedOffset is at or above buffer's | ||
| // offset but less than buffer's length or the actual requestedLength | ||
| if (eTag.equals(buffer.getETag()) | ||
| if (eTag != null && eTag.equals(buffer.getETag()) |
| <include>**/azurebfs/ITestSmallWriteOptimization.java</include> | ||
| <include>**/azurebfs/services/ITestReadBufferManager.java</include> | ||
| <include>**/azurebfs/ITestAbfsStreamStatistics*.java</include> | ||
| <include>**/azurebfs/services/ITestVectoredRead.java</include> |
Contributor
There was a problem hiding this comment.
Why they can't run in parallel?
Contributor
Author
There was a problem hiding this comment.
So, we are taking dependency on buffer states, running in parallel leads to buffer moving out of queues and hence moved it to sequential runs
|
🎊 +1 overall
This message was automatically generated. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
6 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This PR introduces vectored read support in the Azure Blob File System (ABFS) driver to improve read performance for workloads that issue multiple small, non-contiguous read requests.
Vectored reads enable batching of multiple read ranges into fewer network calls, reducing request overhead and improving throughput—especially beneficial for analytics engines like Spark.
Current ABFS read implementation performs sequential, independent read operations for each requested range. This leads to:
Increased number of network calls
Higher latency for small/random reads
Inefficient utilization of bandwidth
Vectored I/O addresses these issues by coalescing multiple read requests into a single or fewer backend calls.