HADOOP-1593. [ABFS] Add vectored read support in ABFS driver #8400
Open
anmolanmol1234 wants to merge 35 commits intoapache:trunkfrom
Open
HADOOP-1593. [ABFS] Add vectored read support in ABFS driver #8400anmolanmol1234 wants to merge 35 commits intoapache:trunkfrom
anmolanmol1234 wants to merge 35 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. |
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?
|
🎊 +1 overall
This message was automatically generated. |
Contributor
Author
|
Hi @mukund-thakur gentle reminder for reviewing the PR |
Contributor
Hi @anmolanmol1234 Thanks for implementing this in ABFS. I skimmed over the changes quickly and I think it is fairly complex thus would be better if someone from ABFS team review and commit this. I haven't touched this code in long time. I would be happy to answer interface related questions. Thanks. |
| @Override | ||
| public int maxReadSizeForVectorReads() { | ||
| return S_2M; | ||
| return client.getAbfsConfiguration().getMaxSeekForVectoredReads(); |
Contributor
There was a problem hiding this comment.
It should be in lines of maxReadSizeForVectorReads
| * Configuration key that defines the maximum gap between adjacent read ranges | ||
| * for merging ranges during vectored reads in ABFS: {@value}. | ||
| */ | ||
| public static final String FS_AZURE_MAX_SEEK_FOR_VECTORED_READS = |
Contributor
There was a problem hiding this comment.
It would be good if you use consistent parameter names.
fs.s3a.vectored.read.max.merged.size is the name in S3A.
| import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ZERO; | ||
| import static org.apache.hadoop.fs.contract.ContractTestUtils.validateVectoredReadResult; | ||
|
|
||
| public class ITestVectoredRead extends AbstractAbfsIntegrationTest { |
Contributor
There was a problem hiding this comment.
Please make sure to run ABFS contract tests ITestAbfsFileSystemContractVectoredRead
Contributor
Author
There was a problem hiding this comment.
yes all tests are passing here
manika137
approved these changes
Apr 29, 2026
Contributor
Author
|
Hi @steveloughran have addressed your concerns, can you take a look once. Thanks |
|
🎊 +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
7 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.