HADOOP-19941. ABFS: Support Photon (Apache Arrow) ListBlobs on the Blob endpoint - #8611
HADOOP-19941. ABFS: Support Photon (Apache Arrow) ListBlobs on the Blob endpoint#8611bhattmanish98 wants to merge 7 commits into
Conversation
…oint Add config-gated support for consuming ListBlobs responses in the Apache Arrow (Photon) format on the ABFS Blob endpoint, with automatic, transparent fallback to the existing XML path and no public API changes. Behaviour - New config fs.azure.photon.enabled (default false, opt-in). When enabled, ABFS advertises Arrow on ListBlobs via an Accept header (application/vnd.apache.arrow.stream, application/xml); the service may honour it or fall back to XML. - Response handling selects a parser by the response Content-Type: an Arrow content type routes to the new Arrow parser, otherwise the existing XML SAX parser is used. Both produce the same BlobListResultSchema, so all downstream processing (rename-pending filtering, pagination, directory rectification) is unchanged. - Malformed Arrow responses surface as AbfsDriverException through the existing error-handling model rather than silently degrading. Parser - ResponseParserFactory chooses between XmlListBlobResponseParser and ArrowListBlobParser behind the ListBlobResponseParser interface. - ArrowListBlobParser reads the Arrow IPC stream and reaches full parity with the XML parser: blob user metadata (Metadata map column), hdi_isfolder=true directory markers (case-insensitive; empty mkdir directories), implicit directories (BlobPrefix rows / ResourceType blobprefix|directory), copy properties, native timestamp (TimeStampSec) and unsigned length (UInt8) vectors normalized to the XML representation, and continuation via the schema NextMarker custom metadata (empty normalized to null). - The Arrow allocator memory limit is configurable, and Arrow/XML timestamps are normalized to a single RFC 1123 GMT representation so both paths yield identical FileStatus values and epochs. - Parsing is immune to thread interrupts: instead of letting ArrowStreamReader wrap the (already fully buffered) body in an interruptible NIO channel - which would abort with ClosedByInterruptException when the caller's interrupt flag is set, e.g. task cancellation - it reads through a non-interruptible channel, matching the interrupt-tolerant XML SAX path. Telemetry - Adds Photon metrics: request count, response (Arrow served) count, fallback (XML returned) count, parse-failure count, and an end-to-end listing-latency duration tracker, wired through AbfsCountersImpl and emitted from listPath. Tests - Unit tests for parser selection (TestResponseParserFactory), Arrow parsing scenarios including metadata/directory markers, blobprefix, native vector types, multi-page, special characters, allocator-limit and malformed streams, and interrupt tolerance (TestArrowListBlobParser), Photon request-header application (TestAbfsBlobClientPhotonHeaders), and metrics (TestPhotonList BlobMetrics). - Integration tests for XML/Arrow parity, fallback, pagination and metrics (ITestAbfsPhotonListStatus) plus additions to the existing list-status ITest. - Documentation updated in blobEndpoint.md; Arrow INFO logs quieted in the test log4j configuration. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
…tBlobs
Address code review feedback on the Photon (Apache Arrow) ListBlobs support:
* ArrowListBlobParser: cap the per-read heap staging buffer in the
non-interruptible channel adapter at 8 KB instead of allocating a buffer
sized to the reader's full request. ArrowStreamReader reads record-batch
bodies through direct (off-heap) ByteBuffers, which take the staging path,
so unbounded requests caused large transient allocations and GC pressure
during listings. Arrow's readFully() loops on partial reads, so capping the
chunk is safe and mirrors the JDK Channels.newChannel transfer-size cap.
* ResponseParserFactory.isArrowResponse(): match the negotiated Arrow IPC
stream media type (application/vnd.apache.arrow.stream), tolerating
parameters such as charset, instead of a loose substring check for "arrow".
This prevents unrelated content types from being misrouted to the Arrow
parser. Removed the now-unused CONTENT_TYPE_ARROW_TOKEN constant and added
a regression test for content types that merely contain "arrow".
* pom.xml: remove the redundant --add-opens=java.base/java.nio=ALL-UNNAMED
from the per-profile surefire argLine entries; it is already contributed by
the shared ${maven-surefire-plugin.argLine} property.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
…oton ListBlobs * ITestAzureBlobFileSystemListStatus: make the list-parse failure assertion tolerant of the XML-fallback case. The wrapping error message is derived from the parser selected by the response Content-Type, not from the Photon config, so a Photon-enabled account that returns XML yields ERR_BLOB_LIST_PARSING. Assert the message contains either the Arrow or XML parsing failure string to avoid flakiness across accounts/service versions. * DateTimeUtils: attach the caught DateTimeException as the log cause when an Arrow timestamp fails to parse, preserving the stack/details for debugging instead of only logging the raw value. * ArrowListBlobParser: reuse a single fixed-size staging buffer per non-interruptible channel instance for direct-ByteBuffer reads instead of allocating a temporary array on every read, removing avoidable allocation/GC pressure during listing. Parsing drives the channel single-threaded, so the shared buffer is safe. * AbfsHttpConstants: document that bumping ApiVersion.getCurrentVersion() to JUN_06_2026 is an intentional global default change required by the Photon (Arrow) ListBlobs contract, validated across all endpoint/operation combinations and therefore safe as the driver-wide default. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
|
💔 -1 overall
This message was automatically generated. |
- Remove redundant lastModifiedTime()/creationTime() accessors from BlobListResultEntrySchema; use lastModified()/creation() consistently. - Fix BC_VACUOUS_INSTANCEOF in ArrowListBlobParser.readMetadata by using the typed List<?> returned by MapVector.getObject with a null check. - Replace checkstyle MagicNumber literals with named constants in DateTimeUtils and the Photon Arrow tests. - Anchor javac/java to the running JDK (java.home) in TestAggregateMetricsManager to avoid toolchain-version mismatch. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
|
🎊 +1 overall
This message was automatically generated. |
| */ | ||
| @VisibleForTesting | ||
| boolean applyPhotonRequestHeadersIfEnabled( | ||
| final List<AbfsHttpHeader> requestHeaders) { |
There was a problem hiding this comment.
No HNS gating. Service returns 409 Conflict for Arrow listing on HNS accounts. A 409 never reaches ResponseParserFactory, so Content-Type fallback can't help — every listing fails hard.
There was a problem hiding this comment.
Good catch. Gated Photon on !getIsNamespaceEnabled() in applyPhotonRequestHeadersIfEnabled, so HNS accounts never advertise Arrow and stay on the XML path. Added testAcceptHeaderUnchangedOnHnsAccount and documented the restriction in blobEndpoint.md.
| several java.base packages but not java.nio, so append it here to ensure | ||
| the default (no-profile) `mvn test` run can execute Arrow-based tests. | ||
| --> | ||
| <maven-surefire-plugin.argLine>-Xmx4096m -Xss4m -XX:+HeapDumpOnOutOfMemoryError ${extraJavaTestArgs} --add-opens=java.base/java.nio=ALL-UNNAMED</maven-surefire-plugin.argLine> |
There was a problem hiding this comment.
The --add-opens flag here only applies when running tests. Real clusters (Spark, Hive) won't have it, so Arrow will fail to initialise on Java 17 and listings will break — even though CI passes. Can we document the flag in blobEndpoint.md and fall back to XML if Arrow can't start?
There was a problem hiding this comment.
Documented the --add-opens=java.base/java.nio=ALL-UNNAMED requirement for Java 17+ in blobEndpoint.md (driver JVM, e.g. Spark/Hive executors, not just tests), with guidance to keep fs.azure.photon.enabled=false when the flag can't be guaranteed. A true runtime "Arrow-init-failed → re-request XML" fallback isn't possible once the service has already returned an Arrow body, so the safe control is the config gate + the flag.
| retryListOp.execute(tracingContext); | ||
| listResponseData = parseListPathResults(retryListOp.getResult(), uri); | ||
| listResponseData.setOp(retryListOp); | ||
| updatePhotonRequestMetric(photonRequested); |
There was a problem hiding this comment.
Metric invariant is broken on the rename-recovery retry path. updatePhotonRequestMetric() is called once before op.execute(), but the rename-recovery retry issues a second ListBlobs request whose response still goes through parseListPathResultsWithMetrics(). So PHOTON_RESPONSE_COUNT + PHOTON_FALLBACK_COUNT can exceed PHOTON_REQUEST_COUNT. Please increment the request counter for the retry call as well (or move the increment into a helper used by both calls).
There was a problem hiding this comment.
Right. The retry issues a second ListBlobs whose response is also classified, so I now call updatePhotonRequestMetric(photonRequested) before retryListOp.execute(...). RESPONSE + FALLBACK can no longer exceed REQUEST.
| final AbfsRestOperation op, final URI uri, final boolean photonRequested) | ||
| throws AzureBlobFileSystemException { | ||
| final AbfsHttpOperation result = op.getResult(); | ||
| updatePhotonResponseMetrics(photonRequested, result); |
There was a problem hiding this comment.
Response/fallback classification also counts non-200 responses. This is invoked before the status code is checked, so an error response (404/409/etc., which carries an XML error body) increments PHOTON_FALLBACK_COUNT and makes the "service fell back to XML" signal unusable for rollout decisions. Classify only when result.getStatusCode() == HTTP_OK.
There was a problem hiding this comment.
updatePhotonResponseMetrics now returns early unless result.getStatusCode() == HTTP_OK, so non-200 error bodies no longer inflate PHOTON_FALLBACK_COUNT.
| if (!getAbfsConfiguration().isPhotonEnabled()) { | ||
| return false; | ||
| } | ||
| requestHeaders.removeIf(header -> ACCEPT.equals(header.getName())); |
There was a problem hiding this comment.
use equalsIgnoreCase
There was a problem hiding this comment.
Changed the removeIf to ACCEPT.equalsIgnoreCase(header.getName()).
| writer.start(); | ||
| writer.writeBatch(); | ||
| writer.end(); | ||
| return out.toByteArray(); |
There was a problem hiding this comment.
return out.toByteArray() is evaluated before writer.close(). It only works because end() flushes; please move ByteArrayOutputStream outside the try-with-resources and return after the block so the stream is probably complete
There was a problem hiding this comment.
Moved ByteArrayOutputStream outside the try-with-resources and return out.toByteArray() after the block, so the stream is guaranteed flushed/closed.
| if (value == null) { | ||
| return null; | ||
| } | ||
| return DateTimeUtils.formatArrowDateTimeToRfc1123(value.toString()); |
There was a problem hiding this comment.
The Javadoc assumes getObject() yields an ISO-8601 local date-time. That's true for TimeStampSecVector / TimeStampMilliVector. But for the TZ variants (TimeStampSecTZVector, TimeStampMilliTZVector, TimeStampMicroTZVector), Arrow's getObject() returns a Long epoch value, not a LocalDateTime.
So value.toString() becomes "1783936279", which formatArrowDateTimeToRfc1123() won't recognize as ISO-8601 and will pass through unchanged. DateTimeUtils.parseLastModifiedTime() then fails on it, and every FileStatus gets a wrong or zero modification time — silently, with no exception and no metric.
There was a problem hiding this comment.
readTimestampAsRfc1123 now detects a Number (TZ vectors' epoch) and converts it via the column's ArrowType.Timestamp unit (sec/milli/micro/nano) → Instant → new DateTimeUtils.formatInstantToRfc1123. Fixed the Javadoc and added testTimeZoneTimestampNormalizedToRfc1123.
| String value = readString((VarCharVector) vector, row); | ||
| if (value != null && !value.isEmpty()) { | ||
| String trimmed = value.trim(); | ||
| return Boolean.parseBoolean(trimmed) || "1".equals(trimmed); |
There was a problem hiding this comment.
use constants for strings
There was a problem hiding this comment.
Extracted NUMERIC_TRUE = "1".
| } | ||
| Object value = vector.getObject(row); | ||
| if (value instanceof Number) { | ||
| return ((Number) value).longValue(); |
There was a problem hiding this comment.
UInt8Vector.getObject() returns a Long holding the raw bit pattern, so an unsigned value above Long.MAX_VALUE comes back negative and is passed straight into entry.setContentLength(). The Long.parseLong text fallback below accepts "-1" just as readily. Could we reject negatives in setContentLength() (or use getObjectNoOverflow() for the UInt8Vector case), so a malformed response can't produce a FileStatus with a negative length?
There was a problem hiding this comment.
setContentLength now rejects negatives (unsigned overflow / "-1") with a LOG.debug, leaving the default 0 rather than a negative FileStatus length. Added testNegativeContentLengthRejected.
| try { | ||
| return Long.parseLong(text); | ||
| } catch (NumberFormatException ignored) { | ||
| // Leave unset if the value is not numeric. |
There was a problem hiding this comment.
A malformed length is swallowed here, so the entry falls back to 0 and surfaces as a zero-byte FileStatus — readers would treat the blob as empty rather than fail. Since a non-numeric length is a data-integrity signal rather than a benign absence, could we at least LOG.debug it, and confirm this matches what the XML path does with an unparseable ?
There was a problem hiding this comment.
Added LOG.debug on the NumberFormatException in readLong. This matches the XML path, which also leaves an unparseable unset (→0). Added testNonNumericContentLengthIgnored.
| } | ||
|
|
||
| while (reader.loadNextBatch()) { | ||
| BatchColumns columns = BatchColumns.resolve(root); |
There was a problem hiding this comment.
varChar() returns null when a column is missing or present-but-not-a-VarCharVector, so if that happens for Name, buildEntry() returns null for every row and parse() completes normally with zero entries — no exception, and PHOTON_PARSE_FAILURE_COUNT never fires. Callers can't distinguish that from a genuinely empty directory, so the failure surfaces later as missing data in Spark/Hive rather than as a listing error. Could we throw here when columns.name == null? Name is the one column buildEntry() can't proceed without, so the "missing columns are tolerated" rule shouldn't extend to it.
There was a problem hiding this comment.
Now throws IOException (with ERR_ARROW_LIST_PARSING) when columns.name == null. Updated testMissingMandatoryNameColumn to expect the throw and added testNullNameValueRowSkipped to confirm a present-but-null name value is still row-skipped.
| } | ||
|
|
||
| while (reader.loadNextBatch()) { | ||
| BatchColumns columns = BatchColumns.resolve(root); |
There was a problem hiding this comment.
NIT: resolve() allocates a HashMap and rescans every field vector on each batch, but ArrowStreamReader loads successive batches into the same VectorSchemaRoot and the same FieldVector instances, so the resolved references stay valid for the whole stream. Could we hoist this above the while loop? On a multi-batch listing this is per-batch overhead on the path whose whole purpose is to be faster than XML.
There was a problem hiding this comment.
Resolved once before the while loop (same VectorSchemaRoot/FieldVector instances persist across batches). Added testTwoBatchesInOneStream to exercise the multi-batch path.
| || metadata.isNull(row)) { | ||
| return result; | ||
| } | ||
| List<?> entries = metadata.getObject(row); |
There was a problem hiding this comment.
Optimization: MapVector.getObject() materialises each row's entries as JsonStringHashMap instances, so a page with metadata allocates a map per key/value pair plus a boxed Text per string — on a 5000-blob listing that's the dominant allocation cost, and the instanceof Map / toString() unwrapping below then discards all of it. Given the perf motivation for this path, worth considering reading the underlying key/value VarCharVectors directly via the map's offset buffer instead.
There was a problem hiding this comment.
Thanks — you're right that MapVector.getObject(row) materialises a JsonStringArrayList of JsonStringHashMaps plus boxed Text per key/value, and that the instanceof Map / toString() unwrapping then throws all of it away. I dug into the trade-off:
Pros of the direct key/value VarCharVector read
- Eliminates the intermediate allocations on the metadata path: for R rows with M entries each, we drop the per-row list, the R×M JsonStringHashMaps, and the R×2M boxed Text objects.
- Lower allocation rate → less young-gen GC churn → marginally better throughput on large, metadata-heavy listings.
- No extra off-heap cost; we'd read straight from the existing map offset buffer + child key/value vectors.
Cons / limits
- It doesn't reduce retained/peak heap. The data we actually keep — the returned Map<String,String> + key/value Strings per entry — is identical either way and lives until FileStatus conversion. So this is a transient-garbage optimization, not a footprint reduction. To shrink retained memory we'd have to stop materialising the full metadata map (e.g. extract only hdi_isfolder), which breaks XML parity.
- The win scales with M. Typical blobs carry 0–2 user-metadata entries, so in the common case we're only skipping ~one tiny map per row — negligible. It's meaningful only when listings are both very large and metadata-rich.
- Higher complexity/risk. It couples the parser to Arrow internals (MapVector → StructVector → child key/value VarCharVectors, per-row ranges via the offset buffer) with hand-rolled null/empty-map and multi-batch handling — precisely the kind of low-level code that's easy to get subtly wrong. And since we still need all keys for XML parity, we don't even get to short-circuit the traversal.
Proposal: keep the current readable getObject() version for this PR (correctness + parity), and track the offset-buffer read as a follow-up gated on a benchmark — if a profile of a large (≥50k), metadata-heavy listing shows readMetadata as a real hotspot/GC driver, I'll switch it and post the before/after numbers. Happy to file the JIRA and link it here. WDYT?
| * @return an absolute path to the tool inside the running JDK, or {@code name}. | ||
| */ | ||
| private static String jdkTool(String name) { | ||
| String javaHome = System.getProperty("java.home"); |
There was a problem hiding this comment.
Added JAVA_HOME_PROPERTY and JDK_BIN_DIR constants.
| private static String jdkTool(String name) { | ||
| String javaHome = System.getProperty("java.home"); | ||
| if (javaHome != null && !javaHome.isEmpty()) { | ||
| File tool = new File(new File(javaHome, "bin"), name); |
| * Unit tests for {@link ArrowListBlobParser}, the Photon (Apache Arrow based) | ||
| * ListBlobs response parser. | ||
| */ | ||
| public class TestArrowListBlobParser { |
There was a problem hiding this comment.
Few tests which can be added
-
Timestamps with a timezone. All the tests use TimeStampSecVector, which has no zone. The TZ variants (TimeStampSecTZVector, TimeStampMilliTZVector) hand back a plain epoch number instead of a date object, so readTimestampAsRfc1123() would turn it into a string like "1783936279" and pass it straight through. Every file would end up with the wrong modified time and nothing would fail. The service could switch to a TZ column without that being a breaking change, so worth a test.
-
Two batches in one stream. testMultiPageListingHandledCorrectly() parses two separate streams, each with one batch, so the while (loadNextBatch()) loop never actually loops and the per-batch column lookup is never re-run.
-
Smaller ones: a UInt8 length big enough to go negative, a length that isn't a number, hdi_isfolder = "1", CopyCompletionTime sent as a real timestamp vector rather than a string, and a name that's just "/".
There was a problem hiding this comment.
Added: TZ timestamp, two batches in one stream, negative UInt8 length, non-numeric length, hdi_isfolder="1" column, and a name of just "/".
Blob endpoint / Photon (Apache Arrow) ListBlobs review fixes: - AbfsBlobClient: gate Photon on non-HNS accounts (the Blob endpoint rejects an Arrow ListBlobs request on HNS with 409, which the XML fallback cannot recover), use equalsIgnoreCase for the ACCEPT header removal, count the rename-recovery retry as a Photon request so the request/response/fallback metric invariant holds, and classify the response/fallback metric only on HTTP 200. - ArrowListBlobParser: fail loudly when the mandatory Name column is absent, resolve batch columns once outside the batch loop, handle timezone-aware Arrow timestamp vectors (epoch values) when normalizing to RFC 1123, reject negative (unsigned-overflow) content lengths, log malformed length values, and use a named constant for the numeric-true value. - DateTimeUtils: add formatInstantToRfc1123 for the epoch timestamp path. - blobEndpoint.md: document the non-HNS restriction and the Java 17+ --add-opens=java.base/java.nio requirement for Arrow. - Tests: fix ByteArrayOutputStream scoping, use constants for JDK tool resolution, and add Arrow parser coverage (TZ timestamp, two batches in one stream, negative and non-numeric length, hdi_isfolder="1", name "/"). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
3 participants
Description of PR
Adds config-gated support for consuming ListBlobs responses in the Apache Arrow (Photon) format on the ABFS Blob endpoint, with automatic, transparent fallback to the existing XML path and no public API changes.
JIRA: HADOOP-19941
Behaviour
fs.azure.photon.enabled(defaultfalse, opt-in). When enabled, ABFS advertises Arrow on ListBlobs via anAcceptheader (application/vnd.apache.arrow.stream, application/xml); the service may honour it or fall back to XML.Content-Type— Arrow routes to the new Arrow parser, otherwise the existing XML SAX parser is used. Both produce the sameBlobListResultSchema, so all downstream processing (rename-pending filtering, pagination, directory rectification) is unchanged.AbfsDriverExceptionthrough the existing error-handling model rather than silently degrading.Implementation
ResponseParserFactorychooses betweenXmlListBlobResponseParserandArrowListBlobParserbehind theListBlobResponseParserinterface.ArrowListBlobParserreaches full parity with the XML parser: blob user metadata (Metadata map column),hdi_isfolder=truedirectory markers (case-insensitive; emptymkdirdirectories), implicit directories (BlobPrefix/ResourceTypeblobprefix|directory), copy properties, nativeTimeStampSec/UInt8vectors, and continuation via schemaNextMarkermetadata.FileStatusvalues; Arrow allocator memory limit is configurable.ArrowStreamReader's interruptible NIO channel, avoidingClosedByInterruptExceptionon interrupted threads — matching the XML path.Telemetry
Adds Photon metrics: request count, response (Arrow served) count, fallback (XML returned) count, parse-failure count, and an end-to-end listing-latency duration tracker.
How was this patch tested?
blobEndpoint.md.For code changes:
org.apache.arrow:arrow-vector, Apache-2.0)