[
https://issues.apache.org/jira/browse/HADOOP-19946?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18141863#comment-18141863
]
ASF GitHub Bot commented on HADOOP-19946:
-----------------------------------------
anmolanmol1234 commented on code in PR #8830:
URL: https://github.com/apache/hadoop/pull/8830#discussion_r4522796900
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java:
##########
@@ -588,11 +673,294 @@ protected int readInternal(final long position, final
byte[] b, final int offset
// got nothing from read-ahead, do our own read now
TracingContext tc = new TracingContext(tracingContext);
tc.setReadType(ReadType.MISSEDCACHE_READ);
- receivedBytes = readRemote(position, b, offset, length, tc);
+
+ String endpoint = findEndpointForPosition(position, length);
+ receivedBytes = readRemote(position, b, offset, length, tc, endpoint);
return receivedBytes;
} else {
LOG.debug("read ahead disabled, reading remote");
- return readRemote(position, b, offset, length, new
TracingContext(tracingContext));
+ return readRemote(position, b, offset, length,
+ new TracingContext(tracingContext),
+ findEndpointForPosition(position, length));
+ }
+ }
+
+ /**
+ * Get blob layout for a file based on start and end position.
+ * In case the layout is present in multiple pages, this method will
internally
+ * fetch all the pages and construct the layout.
+ *
+ * @param start start position of the read
+ * @param end end position of the read
+ * @return blob layout for the file
+ * @throws IllegalStateException if the client is not an AbfsBlobClient
+ */
+ private BlobLayoutResponse getBlobLayout(final long start, final long end,
+ final TracingContext tracingContext) throws AzureBlobFileSystemException
{
+ BlobLayoutResponse fullLayout = new BlobLayoutResponse();
+ TracingContext context = new TracingContext(tracingContext);
+ context.setOperation(FSOperationType.GET_BLOB_LAYOUT);
+ String nextMarker = null;
+ do {
+ AbfsRestOperation op = ((AbfsBlobClient) client).getBlobLayout(path,
+ start, end, eTag, nextMarker, context);
+ try {
+ InputStream stream = op.getResult().getListResultStream();
+ stream.reset();
+
+ SAXParserFactory factory = SAXParserFactory.newInstance();
+ SAXParser parser = factory.newSAXParser();
+ BlobLayoutXmlParser handler = new BlobLayoutXmlParser();
+ parser.parse(stream, handler);
+
+ BlobLayoutResponse currPage = handler.getResponse();
+ fullLayout.addBlobLayoutResponse(currPage);
+ nextMarker = currPage.getNextMarker();
+ } catch (Exception ex) {
+ throw new AbfsRestOperationException(-1, "",
+ "Failed to parse blob layout response", ex);
+ }
+ }
+ while (!StringUtils.isEmpty(nextMarker));
+ return fullLayout;
+ }
+
+ /**
+ * Get Blob layout for the range. This method will first try to get the
layout
+ * from cache and if not present, it will call get blob layout API and update
+ * the cache and return the request data to the caller method. This method
fetch
+ * additional layout to save future calls to get blob layout API.
+ * @param start start position
+ * @param end end position
+ * @param tracingContext tracing context
+ * @return List of blob ranges for the requested range
+ */
+ private List<BlobLayout.BlobRange> getBlobRanges(long start, long end,
+ TracingContext tracingContext) {
+ if (!isDataLocalityCheckEnabled) {
+ return null;
+ }
+
+ TracingContext tracingContext1 = new TracingContext(tracingContext);
+ tracingContext1.setOperation(FSOperationType.GET_BLOB_LAYOUT);
+ List<BlobLayout.BlobRange> gaps = layoutCache.getGaps(eTag, start, end);
+ AbfsCounters abfsCounters = client.getAbfsCounters();
+
+ Set<CompletableFuture<Void>> dependencies = new HashSet<>();
+ if (gaps == null) {
+ // This case will come when data is not distributed across layouts.
+ // In this case, we need to use host URL to fetch the data instead of
+ // iterating through layouts.
+ if (abfsCounters != null) {
+ abfsCounters.incrementCounter(AbfsStatistic.LAYOUT_NOT_PRESENT, 1);
+ }
+ return null;
+ } else if (!gaps.isEmpty()) {
+ if (abfsCounters != null) {
+ abfsCounters.incrementCounter(AbfsStatistic.LAYOUT_CACHE_MISS, 1);
+ }
+ for (BlobLayout.BlobRange gap : gaps) {
+ // Determine the optimal range to fetch using bridge logic
+ BlobLayout.BlobRange bridge = layoutCache.getBridgeGap(
+ eTag, gap.start(), MAX_FETCH_LIMIT);
+
+ long fStart = (bridge != null) ? bridge.start() : gap.start();
+ long fEnd = (bridge != null) ? bridge.end() :
+ Math.min(contentLength, gap.start() + MAX_FETCH_LIMIT) - 1;
+
+ // Atomic operation: registers if absent, returns existing if present
+ dependencies.add(
+ registerAndFetch(start, end, fStart, fEnd, tracingContext1));
+ }
+ } else {
+ if (abfsCounters != null) {
+ abfsCounters.incrementCounter(AbfsStatistic.LAYOUT_CACHE_HIT, 1);
+ }
+ }
+
+ if (!dependencies.isEmpty()) {
+ try {
+ CompletableFuture.allOf(dependencies.toArray(new CompletableFuture[0]))
Review Comment:
Data locality is an optimisation, but this blocks the calling thread for up
to FS_AZURE_BLOB_LAYOUT_FETCH_TIMEOUT_MILLIS, which defaults to 10 seconds. In
readInternal it is also called once per read-ahead window inside the while
loop. On failure or timeout nothing is remembered, because the promise is
removed and the gap is retried. With a slow or degraded layout call, every read
pays the full wait again. Please use a much shorter default wait, and add a
short per-ETag backoff or negative cache after a timeout or failure. That would
keep a layout problem from becoming a read-latency problem.
> ABFS: Enable data locality aware reads for Azure Blob Storage
> -------------------------------------------------------------
>
> Key: HADOOP-19946
> URL: https://issues.apache.org/jira/browse/HADOOP-19946
> Project: Hadoop Common
> Issue Type: New Feature
> Components: fs/azure
> Reporter: Manish Bhatt
> Assignee: Manish Bhatt
> Priority: Major
> Labels: pull-request-available
>
> Introduce data locality support in ABFS to improve read performance for Azure
> Blob Storage by routing read requests to the endpoint that hosts the required
> data range. The goal is to reduce latency, improve throughput, and make read
> operations more efficient for large files and distributed workloads. This
> work adds blob layout discovery, range-to-endpoint mapping, locality-aware
> range selection, and caching of layout metadata to avoid repeated lookups. It
> also enables segmented read-ahead aligned with locality boundaries and adds
> safeguards for concurrency, eviction, timeout handling, and failure
> propagation needed for reliable production-scale reads.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
