[
https://issues.apache.org/jira/browse/HADOOP-19941?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18102403#comment-18102403
]
ASF GitHub Bot commented on HADOOP-19941:
-----------------------------------------
bhattmanish98 commented on code in PR #8611:
URL: https://github.com/apache/hadoop/pull/8611#discussion_r3728047203
##########
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsBlobClientPhotonHeaders.java:
##########
@@ -0,0 +1,182 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowStreamWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.junit.jupiter.api.Test;
+
+import org.apache.hadoop.fs.azurebfs.AbfsConfiguration;
+import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsDriverException;
+
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.APPLICATION_APACHE_ARROW_STREAM;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.APPLICATION_XML;
+import static
org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.ACCEPT;
+import static
org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.CONTENT_TYPE;
+import static
org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_ARROW_LIST_PARSING;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.withSettings;
+
+/**
+ * Unit tests for the Photon (Apache Arrow based ListBlob) request header
+ * behavior of {@link AbfsBlobClient#applyPhotonRequestHeadersIfEnabled(List)}.
+ */
+public class TestAbfsBlobClientPhotonHeaders {
+
+ /** Row count large enough to exhaust the tiny allocator limit. */
+ private static final int OVER_LIMIT_ROW_COUNT = 2000;
+
+ private AbfsBlobClient clientWithPhoton(final boolean photonEnabled) {
+ AbfsConfiguration configuration = mock(AbfsConfiguration.class);
+ doReturn(photonEnabled).when(configuration).isPhotonEnabled();
+ AbfsBlobClient client = mock(AbfsBlobClient.class,
+ withSettings().defaultAnswer(CALLS_REAL_METHODS));
+ doReturn(configuration).when(client).getAbfsConfiguration();
+ return client;
+ }
+
+ private List<AbfsHttpHeader> defaultHeaders() {
+ List<AbfsHttpHeader> headers = new ArrayList<>();
+ headers.add(new AbfsHttpHeader(ACCEPT, "application/json,
application/xml"));
+ return headers;
+ }
+
+ private String acceptValue(final List<AbfsHttpHeader> headers) {
+ return headers.stream()
+ .filter(header -> ACCEPT.equals(header.getName()))
+ .map(AbfsHttpHeader::getValue)
+ .findFirst()
+ .orElse(null);
+ }
+
+ /**
+ * Verify Arrow request headers are added when Photon is enabled.
+ */
+ @Test
+ public void testAcceptHeaderOverriddenWhenPhotonEnabled() {
+ AbfsBlobClient client = clientWithPhoton(true);
+ List<AbfsHttpHeader> headers = defaultHeaders();
+
+ boolean photonRequested =
client.applyPhotonRequestHeadersIfEnabled(headers);
+
+ assertThat(photonRequested)
+ .as("Arrow should be requested when Photon is enabled")
+ .isTrue();
+ long acceptCount = headers.stream()
+ .filter(header -> ACCEPT.equals(header.getName()))
+ .count();
+ assertThat(acceptCount).isEqualTo(1);
+ assertThat(acceptValue(headers))
+ .isEqualTo(APPLICATION_APACHE_ARROW_STREAM + ", " + APPLICATION_XML);
+ }
+
+ /**
+ * Verify the existing Accept header is left unchanged when Photon is
disabled.
+ */
+ @Test
+ public void testAcceptHeaderUnchangedWhenPhotonDisabled() {
+ AbfsBlobClient client = clientWithPhoton(false);
+ List<AbfsHttpHeader> headers = defaultHeaders();
+
+ boolean photonRequested =
client.applyPhotonRequestHeadersIfEnabled(headers);
+
+ assertThat(photonRequested)
+ .as("Arrow should not be requested when Photon is disabled")
+ .isFalse();
+ assertThat(acceptValue(headers))
+ .isEqualTo("application/json, application/xml");
+ }
+
+ /**
+ * Verify that an Arrow ListBlobs response whose parsing exceeds the
configured
+ * allocator memory limit surfaces an {@link AbfsDriverException} carrying
the
+ * Arrow parsing error message, exercising the Arrow branch of
+ * {@link AbfsBlobClient#parseListPathResults} end to end.
+ */
+ @Test
+ public void testArrowOverAllocatorLimitSurfacesDriverException()
+ throws Exception {
+ byte[] arrowStream = buildArrowNameStream(OVER_LIMIT_ROW_COUNT);
+
+ AbfsConfiguration configuration = mock(AbfsConfiguration.class);
+ doReturn(1024L).when(configuration).getPhotonArrowMemoryLimit();
+
+ AbfsBlobClient client = mock(AbfsBlobClient.class,
+ withSettings().defaultAnswer(CALLS_REAL_METHODS));
+ doReturn(configuration).when(client).getAbfsConfiguration();
+ doReturn(new URL("https://account.blob.core.windows.net/container";))
+ .when(client).getBaseUrl();
+
+ AbfsHttpOperation result = mock(AbfsHttpOperation.class);
+ doReturn(APPLICATION_APACHE_ARROW_STREAM)
+ .when(result).getResponseHeaderIgnoreCase(CONTENT_TYPE);
+ doReturn(new ByteArrayInputStream(arrowStream))
+ .when(result).getListResultStream();
+
+ AbfsDriverException ex = intercept(AbfsDriverException.class,
+ () -> client.parseListPathResults(result, null));
+ assertThat(ex.getErrorMessage()).contains(ERR_ARROW_LIST_PARSING);
+ }
+
+ /**
+ * Build a valid single-column ("Name") Arrow IPC stream carrying the given
+ * number of rows, used to drive the allocator limit test.
+ */
+ private static byte[] buildArrowNameStream(final int rows) throws
IOException {
+ Field nameField = new Field("Name",
+ FieldType.nullable(new ArrowType.Utf8()), null);
+ Schema schema = new Schema(Collections.singletonList(nameField));
+ try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE);
+ VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator);
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ ArrowStreamWriter writer = new ArrowStreamWriter(root, null, out)) {
+ VarCharVector nameVector = (VarCharVector) root.getVector("Name");
+ nameVector.allocateNew(rows);
+ for (int i = 0; i < rows; i++) {
+ nameVector.setSafe(i, ("some-reasonably-long-blob-name-" + i)
+ .getBytes(StandardCharsets.UTF_8));
+ }
+ root.setRowCount(rows);
+ writer.start();
+ writer.writeBatch();
+ writer.end();
+ return out.toByteArray();
Review Comment:
Moved ByteArrayOutputStream outside the try-with-resources and return
out.toByteArray() after the block, so the stream is guaranteed flushed/closed.
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/services/ArrowListBlobParser.java:
##########
@@ -0,0 +1,590 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.contracts.services;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.BitVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.complex.MapVector;
+import org.apache.arrow.vector.ipc.ArrowStreamReader;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.utils.DateTimeUtils;
+
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_ACL;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CONTENT_LENGTH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_COMPLETION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_ID;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_PROGRESS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_SOURCE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS_DESCRIPTION;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CREATION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_ETAG;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_GROUP;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_IS_DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_LAST_MODIFIED;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_METADATA;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_NAME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_OWNER;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_PERMISSIONS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_RESOURCE_TYPE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_METADATA_NEXT_MARKER;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_RESOURCE_TYPE_BLOB_PREFIX;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.FORWARD_SLASH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ROOT_PATH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.XML_TAG_HDI_ISFOLDER;
+import static
org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_META_HDI_ISFOLDER;
+import static
org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_ARROW_LIST_PARSING;
+
+/**
+ * Photon response parser implementation for Apache Arrow based ListBlobs
+ * responses. The parser reads the Arrow IPC stream returned by the service,
+ * iterates through the record batches and converts each row into a
+ * {@link BlobListResultEntrySchema}, producing a {@link BlobListResultSchema}
+ * that keeps the downstream FileStatus conversion path unchanged.
+ * <p>
+ * Arrow specific objects are confined to this class; the output contract is
+ * {@link BlobListResultSchema} so existing listing behavior is preserved.
+ * <p>
+ * Column matching is case-insensitive. Unknown columns are ignored and missing
+ * columns are treated as absent values so that additive schema changes do not
+ * break parsing. The continuation token is read from the Arrow schema custom
+ * metadata (key {@code NextMarker}).
+ */
+public class ArrowListBlobParser implements ListBlobResponseParser {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ArrowListBlobParser.class);
+
+ /**
+ * Upper bound, in bytes, on the heap staging buffer allocated per read when
+ * copying into a direct (non-array-backed) {@link ByteBuffer}. Caps
transient
+ * allocations for large reader requests; the reader simply issues additional
+ * reads for anything beyond this chunk.
+ */
+ private static final int NON_INTERRUPTIBLE_READ_CHUNK = 8192;
+
+ /**
+ * Base URL for which the ListBlobs API is called, used to build the absolute
+ * URL for each entry (mirrors the XML parser behavior).
+ */
+ private final String url;
+
+ /**
+ * Maximum off-heap (direct) memory in bytes the Arrow allocator may use
while
+ * parsing a single response. Bounds the memory an oversized or malformed
+ * response can consume; exceeding it fails the parse with an
+ * {@link org.apache.arrow.memory.OutOfMemoryException} that is surfaced as
an
+ * {@link IOException}.
+ */
+ private final long memoryLimitBytes;
+
+ /**
+ * @param url base URL for which the ListBlobs API is called.
+ * @param memoryLimitBytes maximum off-heap memory in bytes the Arrow
allocator
+ * may use while parsing a single response.
+ */
+ public ArrowListBlobParser(final String url, final long memoryLimitBytes) {
+ this.url = url;
+ this.memoryLimitBytes = memoryLimitBytes;
+ }
+
+ @Override
+ public BlobListResultSchema parse(final InputStream responseStream)
+ throws IOException {
+ BlobListResultSchema listResultSchema = new BlobListResultSchema();
+ try (BufferAllocator allocator = new RootAllocator(memoryLimitBytes);
+ ArrowStreamReader reader =
+ new ArrowStreamReader(nonInterruptibleChannel(responseStream),
+ allocator)) {
+ VectorSchemaRoot root = reader.getVectorSchemaRoot();
+ Schema schema = root.getSchema();
+
+ // Continuation token is carried in the Arrow schema custom metadata.
+ // The service emits an empty string when there is no continuation token,
+ // whereas the XML path leaves it null; normalize an absent or empty
+ // marker to null so both formats produce identical values.
+ Map<String, String> customMetadata = schema.getCustomMetadata();
+ if (customMetadata != null) {
+ String nextMarker = customMetadata.get(ARROW_METADATA_NEXT_MARKER);
+ listResultSchema.setNextMarker(
+ (nextMarker == null || nextMarker.isEmpty()) ? null : nextMarker);
+ }
+
+ while (reader.loadNextBatch()) {
+ BatchColumns columns = BatchColumns.resolve(root);
+ int rowCount = root.getRowCount();
+ for (int row = 0; row < rowCount; row++) {
+ BlobListResultEntrySchema entry = buildEntry(columns, row);
+ if (entry != null) {
+ listResultSchema.addBlobListEntry(entry);
+ }
+ }
+ }
+ LOG.debug("Photon ListBlobs parsed {} blobs with {} as continuation
token",
+ listResultSchema.paths().size(), listResultSchema.getNextMarker());
+ return listResultSchema;
+ } catch (IOException ex) {
+ throw ex;
+ } catch (RuntimeException ex) {
+ // Wrap Arrow runtime failures (e.g. corrupted stream) as IOException so
+ // callers can surface them through the existing error-handling model.
+ throw new IOException(ex);
+ }
+ }
+
+ @Override
+ public String getParsingErrorMessage() {
+ return ERR_ARROW_LIST_PARSING;
+ }
+
+ /**
+ * Wrap an {@link InputStream} in a {@link ReadableByteChannel} that is
+ * <em>not</em> interruptible.
+ *
+ * <p>{@link ArrowStreamReader}, when handed an {@code InputStream},
internally
+ * adapts it with {@link java.nio.channels.Channels#newChannel(InputStream)},
+ * which returns a channel extending
+ * {@link java.nio.channels.spi.AbstractInterruptibleChannel}. If the parsing
+ * thread's interrupt flag is set (e.g. Hadoop task cancellation, speculative
+ * execution kills, or an executor {@code shutdownNow()}), that channel
aborts
+ * the read with a {@link java.nio.channels.ClosedByInterruptException},
+ * turning a benign interrupt into a hard listing failure. The XML SAX parser
+ * reads from a plain {@code InputStream} and is immune to this, so the Arrow
+ * path must not be more fragile.</p>
+ *
+ * <p>The list response body is already fully buffered in memory before it
+ * reaches the parser, so an interruptible channel offers no benefit here;
this
+ * plain adapter simply reads from the buffer without consulting the thread's
+ * interrupt status.</p>
+ *
+ * @param in the buffered response stream.
+ * @return a non-interruptible channel over {@code in}.
+ */
+ private static ReadableByteChannel nonInterruptibleChannel(
+ final InputStream in) {
+ return new ReadableByteChannel() {
+ private volatile boolean open = true;
+ // Fixed staging buffer reused across reads for direct ByteBuffers. Arrow
+ // parsing drives this channel from a single thread, so a per-instance
+ // buffer is safe and avoids per-read allocation/GC churn.
+ private final byte[] staging = new byte[NON_INTERRUPTIBLE_READ_CHUNK];
+
+ @Override
+ public int read(final ByteBuffer dst) throws IOException {
+ final int toRead = dst.remaining();
+ if (toRead == 0) {
+ return 0;
+ }
+ if (dst.hasArray()) {
+ final int n = in.read(dst.array(),
+ dst.arrayOffset() + dst.position(), toRead);
+ if (n > 0) {
+ dst.position(dst.position() + n);
+ }
+ return n;
+ }
+ // For a direct (non-array-backed) ByteBuffer we must stage bytes in a
+ // heap buffer first. ArrowStreamReader can request very large reads,
so
+ // read at most one staging-buffer worth per call and let the reader
+ // issue further reads for the remainder (its readFully loops on
partial
+ // reads). This bounds allocation and reuses a single buffer.
+ final int chunk = Math.min(toRead, staging.length);
+ final int n = in.read(staging, 0, chunk);
+ if (n > 0) {
+ dst.put(staging, 0, n);
+ }
+ return n;
+ }
+
+ @Override
+ public boolean isOpen() {
+ return open;
+ }
+
+ @Override
+ public void close() throws IOException {
+ open = false;
+ in.close();
+ }
+ };
+ }
+
+ /**
+ * Convert a single Arrow row into a {@link BlobListResultEntrySchema}. Rows
+ * without a usable name are skipped.
+ */
+ private BlobListResultEntrySchema buildEntry(
+ final BatchColumns columns, final int row) {
+ String name = readString(columns.name, row);
+ if (name == null || name.isEmpty()) {
+ return null;
+ }
+ // Directory names may carry a trailing slash; strip it to match XML
parser.
+ if (name.endsWith(FORWARD_SLASH)) {
+ name = name.substring(0, name.length() - 1);
+ }
+
+ BlobListResultEntrySchema entry = new BlobListResultEntrySchema();
+ entry.setName(name);
+ entry.setPath(new Path(ROOT_PATH + name));
+ entry.setUrl(url + ROOT_PATH + name);
+
+ entry.setETag(readString(columns.etag, row));
+ // The Arrow (Photon) response serializes timestamps as native Arrow
+ // timestamp vectors (e.g. Creation-Time / Last-Modified as TimeStampSec),
+ // whose object form is an ISO-8601 local date-time (e.g.
2026-07-06T10:31:19)
+ // in UTC. Normalize to RFC 1123 GMT so both paths produce identical values
+ // and DateTimeUtils.parseLastModifiedTime yields the correct time.
+ entry.setLastModifiedTime(readTimestampAsRfc1123(columns.lastModified,
row));
+ entry.setCreationTime(readTimestampAsRfc1123(columns.creationTime, row));
+ entry.setOwner(readString(columns.owner, row));
+ entry.setGroup(readString(columns.group, row));
+ entry.setPermission(readString(columns.permission, row));
+ entry.setAcl(readString(columns.acl, row));
+
+ // Blob user metadata (including the hdi_isfolder directory marker) is
+ // carried in a single Arrow map column; mirror the XML parser which
+ // populates the metadata map from the <Metadata> element.
+ Map<String, String> metadata = readMetadata(columns.metadata, row);
+ if (!metadata.isEmpty()) {
+ entry.setMetadata(metadata);
+ }
+
+ setContentLength(entry, columns.contentLength, row);
+ setCopyProperties(entry, columns, row);
+ entry.setIsDirectory(resolveIsDirectory(columns, metadata, row));
+ return entry;
+ }
+
+ /**
+ * Populate the copy-related properties, mirroring the XML parser's handling
+ * of the {@code Properties} copy elements. Absent columns leave the
+ * corresponding field unset, exactly as the XML path does when the elements
+ * are missing.
+ */
+ private void setCopyProperties(final BlobListResultEntrySchema entry,
+ final BatchColumns columns, final int row) {
+ entry.setCopyId(readString(columns.copyId, row));
+ entry.setCopyStatus(readString(columns.copyStatus, row));
+ entry.setCopySourceUrl(readString(columns.copySource, row));
+ entry.setCopyProgress(readString(columns.copyProgress, row));
+ entry.setCopyStatusDescription(
+ readString(columns.copyStatusDescription, row));
+ // XML stores the completion time as epoch millis parsed from an RFC 1123
+ // string; normalize the Arrow timestamp to the same RFC 1123 form first so
+ // both paths yield identical epoch values.
+ String copyCompletion =
+ readTimestampAsRfc1123(columns.copyCompletionTime, row);
+ if (copyCompletion != null) {
+ entry.setCopyCompletionTime(
+ DateTimeUtils.parseLastModifiedTime(copyCompletion));
+ }
+ }
+
+ /**
+ * Populate the content length from the content-length column. The Blob
+ * endpoint surfaces it as a native unsigned 64-bit integer
+ * ({@code UInt8Vector}); other numeric vector types and a numeric string
+ * column are also tolerated so additive schema changes do not break parsing.
+ */
+ private void setContentLength(final BlobListResultEntrySchema entry,
+ final FieldVector contentLength, final int row) {
+ Long value = readLong(contentLength, row);
+ if (value != null) {
+ entry.setContentLength(value);
+ }
+ }
+
+ /**
+ * Determine whether the row represents a directory. Mirrors the XML parser
+ * ({@code BlobListXmlParser}), which flags an entry as a directory when any
of
+ * the following hold:
+ * <ul>
+ * <li>the entry is a {@code BlobPrefix} - surfaced in the Arrow response
as
+ * a {@code ResourceType} of {@code blobprefix} - i.e. an implicit
+ * directory;</li>
+ * <li>a {@code ResourceType} of {@code directory};</li>
+ * <li>an explicit boolean {@code IsDirectory} indicator;</li>
+ * <li>the {@code hdi_isfolder} user metadata being {@code true}.</li>
+ * </ul>
+ * The implicit-directory ({@code blobprefix}) and {@code hdi_isfolder} cases
+ * are essential: an implicit directory has no marker blob at all, while an
+ * empty directory created by {@code mkdir} is a zero-byte marker blob whose
+ * only directory indicator is the {@code hdi_isfolder=true} metadata entry.
+ * The metadata key is matched case-insensitively, matching the XML parser
+ * ({@code equalsIgnoreCase}). Without honoring these, such directories are
+ * misclassified as files, diverging from the XML listing path.
+ */
+ private boolean resolveIsDirectory(final BatchColumns columns,
+ final Map<String, String> metadata, final int row) {
+ String resourceType = readString(columns.resourceType, row);
+ if (DIRECTORY.equalsIgnoreCase(resourceType)
+ || ARROW_RESOURCE_TYPE_BLOB_PREFIX.equalsIgnoreCase(resourceType)) {
+ return true;
+ }
+ if (readBoolean(columns.isDirectory, row)) {
+ return true;
+ }
+ String hdiIsFolder = metadataValueIgnoreCase(metadata,
XML_TAG_HDI_ISFOLDER);
+ if (hdiIsFolder != null && Boolean.parseBoolean(hdiIsFolder.trim())) {
+ return true;
+ }
+ // Fallback for a schema that flattens the marker metadata into a dedicated
+ // column instead of the metadata map.
+ return readBoolean(columns.hdiIsFolder, row);
+ }
+
+ /**
+ * Case-insensitive lookup of a metadata value by key, mirroring the XML
+ * parser's {@code equalsIgnoreCase} matching of the {@code hdi_isfolder}
+ * marker (the service preserves the casing used when the metadata was set).
+ */
+ private static String metadataValueIgnoreCase(
+ final Map<String, String> metadata, final String key) {
+ String value = metadata.get(key);
+ if (value != null || metadata.containsKey(key)) {
+ return value;
+ }
+ for (Map.Entry<String, String> entry : metadata.entrySet()) {
+ if (key.equalsIgnoreCase(entry.getKey())) {
+ return entry.getValue();
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Read the user metadata key/value pairs from the Arrow map column for the
+ * given row. Keys are stored verbatim (e.g. {@code hdi_isfolder}) to match
the
+ * XML parser. Returns an empty (mutable) map when the column is absent or
the
+ * value is null.
+ */
+ private Map<String, String> readMetadata(final MapVector metadata,
+ final int row) {
+ Map<String, String> result = new HashMap<>();
+ if (metadata == null || row >= metadata.getValueCount()
+ || metadata.isNull(row)) {
+ return result;
+ }
+ List<?> entries = metadata.getObject(row);
+ if (entries == null) {
+ return result;
+ }
+ for (Object element : entries) {
+ if (element instanceof Map) {
+ Map<?, ?> keyValue = (Map<?, ?>) element;
+ Object key = keyValue.get(MapVector.KEY_NAME);
+ Object value = keyValue.get(MapVector.VALUE_NAME);
+ if (key != null) {
+ result.put(key.toString(), value == null ? null : value.toString());
+ }
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Read a timestamp column for the given row and convert it to the RFC 1123
+ * GMT representation used by the XML path. Native Arrow timestamp vectors
+ * expose an ISO-8601 local date-time via {@link FieldVector#getObject(int)};
+ * a textual (VarChar) timestamp column is also tolerated. Returns
+ * {@code null} when the column is absent or the value is null.
+ */
+ private String readTimestampAsRfc1123(final FieldVector vector,
+ final int row) {
+ if (vector == null || row >= vector.getValueCount()
+ || vector.isNull(row)) {
+ return null;
+ }
+ Object value = vector.getObject(row);
+ if (value == null) {
+ return null;
+ }
+ return DateTimeUtils.formatArrowDateTimeToRfc1123(value.toString());
Review 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.
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/services/ArrowListBlobParser.java:
##########
@@ -0,0 +1,590 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.contracts.services;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.BitVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.complex.MapVector;
+import org.apache.arrow.vector.ipc.ArrowStreamReader;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.utils.DateTimeUtils;
+
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_ACL;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CONTENT_LENGTH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_COMPLETION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_ID;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_PROGRESS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_SOURCE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS_DESCRIPTION;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CREATION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_ETAG;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_GROUP;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_IS_DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_LAST_MODIFIED;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_METADATA;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_NAME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_OWNER;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_PERMISSIONS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_RESOURCE_TYPE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_METADATA_NEXT_MARKER;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_RESOURCE_TYPE_BLOB_PREFIX;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.FORWARD_SLASH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ROOT_PATH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.XML_TAG_HDI_ISFOLDER;
+import static
org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_META_HDI_ISFOLDER;
+import static
org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_ARROW_LIST_PARSING;
+
+/**
+ * Photon response parser implementation for Apache Arrow based ListBlobs
+ * responses. The parser reads the Arrow IPC stream returned by the service,
+ * iterates through the record batches and converts each row into a
+ * {@link BlobListResultEntrySchema}, producing a {@link BlobListResultSchema}
+ * that keeps the downstream FileStatus conversion path unchanged.
+ * <p>
+ * Arrow specific objects are confined to this class; the output contract is
+ * {@link BlobListResultSchema} so existing listing behavior is preserved.
+ * <p>
+ * Column matching is case-insensitive. Unknown columns are ignored and missing
+ * columns are treated as absent values so that additive schema changes do not
+ * break parsing. The continuation token is read from the Arrow schema custom
+ * metadata (key {@code NextMarker}).
+ */
+public class ArrowListBlobParser implements ListBlobResponseParser {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ArrowListBlobParser.class);
+
+ /**
+ * Upper bound, in bytes, on the heap staging buffer allocated per read when
+ * copying into a direct (non-array-backed) {@link ByteBuffer}. Caps
transient
+ * allocations for large reader requests; the reader simply issues additional
+ * reads for anything beyond this chunk.
+ */
+ private static final int NON_INTERRUPTIBLE_READ_CHUNK = 8192;
+
+ /**
+ * Base URL for which the ListBlobs API is called, used to build the absolute
+ * URL for each entry (mirrors the XML parser behavior).
+ */
+ private final String url;
+
+ /**
+ * Maximum off-heap (direct) memory in bytes the Arrow allocator may use
while
+ * parsing a single response. Bounds the memory an oversized or malformed
+ * response can consume; exceeding it fails the parse with an
+ * {@link org.apache.arrow.memory.OutOfMemoryException} that is surfaced as
an
+ * {@link IOException}.
+ */
+ private final long memoryLimitBytes;
+
+ /**
+ * @param url base URL for which the ListBlobs API is called.
+ * @param memoryLimitBytes maximum off-heap memory in bytes the Arrow
allocator
+ * may use while parsing a single response.
+ */
+ public ArrowListBlobParser(final String url, final long memoryLimitBytes) {
+ this.url = url;
+ this.memoryLimitBytes = memoryLimitBytes;
+ }
+
+ @Override
+ public BlobListResultSchema parse(final InputStream responseStream)
+ throws IOException {
+ BlobListResultSchema listResultSchema = new BlobListResultSchema();
+ try (BufferAllocator allocator = new RootAllocator(memoryLimitBytes);
+ ArrowStreamReader reader =
+ new ArrowStreamReader(nonInterruptibleChannel(responseStream),
+ allocator)) {
+ VectorSchemaRoot root = reader.getVectorSchemaRoot();
+ Schema schema = root.getSchema();
+
+ // Continuation token is carried in the Arrow schema custom metadata.
+ // The service emits an empty string when there is no continuation token,
+ // whereas the XML path leaves it null; normalize an absent or empty
+ // marker to null so both formats produce identical values.
+ Map<String, String> customMetadata = schema.getCustomMetadata();
+ if (customMetadata != null) {
+ String nextMarker = customMetadata.get(ARROW_METADATA_NEXT_MARKER);
+ listResultSchema.setNextMarker(
+ (nextMarker == null || nextMarker.isEmpty()) ? null : nextMarker);
+ }
+
+ while (reader.loadNextBatch()) {
+ BatchColumns columns = BatchColumns.resolve(root);
+ int rowCount = root.getRowCount();
+ for (int row = 0; row < rowCount; row++) {
+ BlobListResultEntrySchema entry = buildEntry(columns, row);
+ if (entry != null) {
+ listResultSchema.addBlobListEntry(entry);
+ }
+ }
+ }
+ LOG.debug("Photon ListBlobs parsed {} blobs with {} as continuation
token",
+ listResultSchema.paths().size(), listResultSchema.getNextMarker());
+ return listResultSchema;
+ } catch (IOException ex) {
+ throw ex;
+ } catch (RuntimeException ex) {
+ // Wrap Arrow runtime failures (e.g. corrupted stream) as IOException so
+ // callers can surface them through the existing error-handling model.
+ throw new IOException(ex);
+ }
+ }
+
+ @Override
+ public String getParsingErrorMessage() {
+ return ERR_ARROW_LIST_PARSING;
+ }
+
+ /**
+ * Wrap an {@link InputStream} in a {@link ReadableByteChannel} that is
+ * <em>not</em> interruptible.
+ *
+ * <p>{@link ArrowStreamReader}, when handed an {@code InputStream},
internally
+ * adapts it with {@link java.nio.channels.Channels#newChannel(InputStream)},
+ * which returns a channel extending
+ * {@link java.nio.channels.spi.AbstractInterruptibleChannel}. If the parsing
+ * thread's interrupt flag is set (e.g. Hadoop task cancellation, speculative
+ * execution kills, or an executor {@code shutdownNow()}), that channel
aborts
+ * the read with a {@link java.nio.channels.ClosedByInterruptException},
+ * turning a benign interrupt into a hard listing failure. The XML SAX parser
+ * reads from a plain {@code InputStream} and is immune to this, so the Arrow
+ * path must not be more fragile.</p>
+ *
+ * <p>The list response body is already fully buffered in memory before it
+ * reaches the parser, so an interruptible channel offers no benefit here;
this
+ * plain adapter simply reads from the buffer without consulting the thread's
+ * interrupt status.</p>
+ *
+ * @param in the buffered response stream.
+ * @return a non-interruptible channel over {@code in}.
+ */
+ private static ReadableByteChannel nonInterruptibleChannel(
+ final InputStream in) {
+ return new ReadableByteChannel() {
+ private volatile boolean open = true;
+ // Fixed staging buffer reused across reads for direct ByteBuffers. Arrow
+ // parsing drives this channel from a single thread, so a per-instance
+ // buffer is safe and avoids per-read allocation/GC churn.
+ private final byte[] staging = new byte[NON_INTERRUPTIBLE_READ_CHUNK];
+
+ @Override
+ public int read(final ByteBuffer dst) throws IOException {
+ final int toRead = dst.remaining();
+ if (toRead == 0) {
+ return 0;
+ }
+ if (dst.hasArray()) {
+ final int n = in.read(dst.array(),
+ dst.arrayOffset() + dst.position(), toRead);
+ if (n > 0) {
+ dst.position(dst.position() + n);
+ }
+ return n;
+ }
+ // For a direct (non-array-backed) ByteBuffer we must stage bytes in a
+ // heap buffer first. ArrowStreamReader can request very large reads,
so
+ // read at most one staging-buffer worth per call and let the reader
+ // issue further reads for the remainder (its readFully loops on
partial
+ // reads). This bounds allocation and reuses a single buffer.
+ final int chunk = Math.min(toRead, staging.length);
+ final int n = in.read(staging, 0, chunk);
+ if (n > 0) {
+ dst.put(staging, 0, n);
+ }
+ return n;
+ }
+
+ @Override
+ public boolean isOpen() {
+ return open;
+ }
+
+ @Override
+ public void close() throws IOException {
+ open = false;
+ in.close();
+ }
+ };
+ }
+
+ /**
+ * Convert a single Arrow row into a {@link BlobListResultEntrySchema}. Rows
+ * without a usable name are skipped.
+ */
+ private BlobListResultEntrySchema buildEntry(
+ final BatchColumns columns, final int row) {
+ String name = readString(columns.name, row);
+ if (name == null || name.isEmpty()) {
+ return null;
+ }
+ // Directory names may carry a trailing slash; strip it to match XML
parser.
+ if (name.endsWith(FORWARD_SLASH)) {
+ name = name.substring(0, name.length() - 1);
+ }
+
+ BlobListResultEntrySchema entry = new BlobListResultEntrySchema();
+ entry.setName(name);
+ entry.setPath(new Path(ROOT_PATH + name));
+ entry.setUrl(url + ROOT_PATH + name);
+
+ entry.setETag(readString(columns.etag, row));
+ // The Arrow (Photon) response serializes timestamps as native Arrow
+ // timestamp vectors (e.g. Creation-Time / Last-Modified as TimeStampSec),
+ // whose object form is an ISO-8601 local date-time (e.g.
2026-07-06T10:31:19)
+ // in UTC. Normalize to RFC 1123 GMT so both paths produce identical values
+ // and DateTimeUtils.parseLastModifiedTime yields the correct time.
+ entry.setLastModifiedTime(readTimestampAsRfc1123(columns.lastModified,
row));
+ entry.setCreationTime(readTimestampAsRfc1123(columns.creationTime, row));
+ entry.setOwner(readString(columns.owner, row));
+ entry.setGroup(readString(columns.group, row));
+ entry.setPermission(readString(columns.permission, row));
+ entry.setAcl(readString(columns.acl, row));
+
+ // Blob user metadata (including the hdi_isfolder directory marker) is
+ // carried in a single Arrow map column; mirror the XML parser which
+ // populates the metadata map from the <Metadata> element.
+ Map<String, String> metadata = readMetadata(columns.metadata, row);
+ if (!metadata.isEmpty()) {
+ entry.setMetadata(metadata);
+ }
+
+ setContentLength(entry, columns.contentLength, row);
+ setCopyProperties(entry, columns, row);
+ entry.setIsDirectory(resolveIsDirectory(columns, metadata, row));
+ return entry;
+ }
+
+ /**
+ * Populate the copy-related properties, mirroring the XML parser's handling
+ * of the {@code Properties} copy elements. Absent columns leave the
+ * corresponding field unset, exactly as the XML path does when the elements
+ * are missing.
+ */
+ private void setCopyProperties(final BlobListResultEntrySchema entry,
+ final BatchColumns columns, final int row) {
+ entry.setCopyId(readString(columns.copyId, row));
+ entry.setCopyStatus(readString(columns.copyStatus, row));
+ entry.setCopySourceUrl(readString(columns.copySource, row));
+ entry.setCopyProgress(readString(columns.copyProgress, row));
+ entry.setCopyStatusDescription(
+ readString(columns.copyStatusDescription, row));
+ // XML stores the completion time as epoch millis parsed from an RFC 1123
+ // string; normalize the Arrow timestamp to the same RFC 1123 form first so
+ // both paths yield identical epoch values.
+ String copyCompletion =
+ readTimestampAsRfc1123(columns.copyCompletionTime, row);
+ if (copyCompletion != null) {
+ entry.setCopyCompletionTime(
+ DateTimeUtils.parseLastModifiedTime(copyCompletion));
+ }
+ }
+
+ /**
+ * Populate the content length from the content-length column. The Blob
+ * endpoint surfaces it as a native unsigned 64-bit integer
+ * ({@code UInt8Vector}); other numeric vector types and a numeric string
+ * column are also tolerated so additive schema changes do not break parsing.
+ */
+ private void setContentLength(final BlobListResultEntrySchema entry,
+ final FieldVector contentLength, final int row) {
+ Long value = readLong(contentLength, row);
+ if (value != null) {
+ entry.setContentLength(value);
+ }
+ }
+
+ /**
+ * Determine whether the row represents a directory. Mirrors the XML parser
+ * ({@code BlobListXmlParser}), which flags an entry as a directory when any
of
+ * the following hold:
+ * <ul>
+ * <li>the entry is a {@code BlobPrefix} - surfaced in the Arrow response
as
+ * a {@code ResourceType} of {@code blobprefix} - i.e. an implicit
+ * directory;</li>
+ * <li>a {@code ResourceType} of {@code directory};</li>
+ * <li>an explicit boolean {@code IsDirectory} indicator;</li>
+ * <li>the {@code hdi_isfolder} user metadata being {@code true}.</li>
+ * </ul>
+ * The implicit-directory ({@code blobprefix}) and {@code hdi_isfolder} cases
+ * are essential: an implicit directory has no marker blob at all, while an
+ * empty directory created by {@code mkdir} is a zero-byte marker blob whose
+ * only directory indicator is the {@code hdi_isfolder=true} metadata entry.
+ * The metadata key is matched case-insensitively, matching the XML parser
+ * ({@code equalsIgnoreCase}). Without honoring these, such directories are
+ * misclassified as files, diverging from the XML listing path.
+ */
+ private boolean resolveIsDirectory(final BatchColumns columns,
+ final Map<String, String> metadata, final int row) {
+ String resourceType = readString(columns.resourceType, row);
+ if (DIRECTORY.equalsIgnoreCase(resourceType)
+ || ARROW_RESOURCE_TYPE_BLOB_PREFIX.equalsIgnoreCase(resourceType)) {
+ return true;
+ }
+ if (readBoolean(columns.isDirectory, row)) {
+ return true;
+ }
+ String hdiIsFolder = metadataValueIgnoreCase(metadata,
XML_TAG_HDI_ISFOLDER);
+ if (hdiIsFolder != null && Boolean.parseBoolean(hdiIsFolder.trim())) {
+ return true;
+ }
+ // Fallback for a schema that flattens the marker metadata into a dedicated
+ // column instead of the metadata map.
+ return readBoolean(columns.hdiIsFolder, row);
+ }
+
+ /**
+ * Case-insensitive lookup of a metadata value by key, mirroring the XML
+ * parser's {@code equalsIgnoreCase} matching of the {@code hdi_isfolder}
+ * marker (the service preserves the casing used when the metadata was set).
+ */
+ private static String metadataValueIgnoreCase(
+ final Map<String, String> metadata, final String key) {
+ String value = metadata.get(key);
+ if (value != null || metadata.containsKey(key)) {
+ return value;
+ }
+ for (Map.Entry<String, String> entry : metadata.entrySet()) {
+ if (key.equalsIgnoreCase(entry.getKey())) {
+ return entry.getValue();
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Read the user metadata key/value pairs from the Arrow map column for the
+ * given row. Keys are stored verbatim (e.g. {@code hdi_isfolder}) to match
the
+ * XML parser. Returns an empty (mutable) map when the column is absent or
the
+ * value is null.
+ */
+ private Map<String, String> readMetadata(final MapVector metadata,
+ final int row) {
+ Map<String, String> result = new HashMap<>();
+ if (metadata == null || row >= metadata.getValueCount()
+ || metadata.isNull(row)) {
+ return result;
+ }
+ List<?> entries = metadata.getObject(row);
+ if (entries == null) {
+ return result;
+ }
+ for (Object element : entries) {
+ if (element instanceof Map) {
+ Map<?, ?> keyValue = (Map<?, ?>) element;
+ Object key = keyValue.get(MapVector.KEY_NAME);
+ Object value = keyValue.get(MapVector.VALUE_NAME);
+ if (key != null) {
+ result.put(key.toString(), value == null ? null : value.toString());
+ }
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Read a timestamp column for the given row and convert it to the RFC 1123
+ * GMT representation used by the XML path. Native Arrow timestamp vectors
+ * expose an ISO-8601 local date-time via {@link FieldVector#getObject(int)};
+ * a textual (VarChar) timestamp column is also tolerated. Returns
+ * {@code null} when the column is absent or the value is null.
+ */
+ private String readTimestampAsRfc1123(final FieldVector vector,
+ final int row) {
+ if (vector == null || row >= vector.getValueCount()
+ || vector.isNull(row)) {
+ return null;
+ }
+ Object value = vector.getObject(row);
+ if (value == null) {
+ return null;
+ }
+ return DateTimeUtils.formatArrowDateTimeToRfc1123(value.toString());
+ }
+
+ /**
+ * Read a numeric column for the given row as a {@code long}. Supports native
+ * Arrow integer vectors (whose object form is a {@link Number}) and a
numeric
+ * {@link VarCharVector}. Returns {@code null} when the column is absent, the
+ * value is null, or a textual value is not parseable.
+ */
+ private Long readLong(final FieldVector vector, final int row) {
+ if (vector == null || row >= vector.getValueCount()
+ || vector.isNull(row)) {
+ return null;
+ }
+ Object value = vector.getObject(row);
+ if (value instanceof Number) {
+ return ((Number) value).longValue();
Review Comment:
setContentLength now rejects negatives (unsigned overflow / "-1") with a
LOG.debug, leaving the default 0 rather than a negative FileStatus length.
Added testNegativeContentLengthRejected.
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/services/ArrowListBlobParser.java:
##########
@@ -0,0 +1,590 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.contracts.services;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.BitVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.complex.MapVector;
+import org.apache.arrow.vector.ipc.ArrowStreamReader;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.utils.DateTimeUtils;
+
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_ACL;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CONTENT_LENGTH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_COMPLETION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_ID;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_PROGRESS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_SOURCE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS_DESCRIPTION;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CREATION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_ETAG;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_GROUP;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_IS_DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_LAST_MODIFIED;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_METADATA;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_NAME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_OWNER;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_PERMISSIONS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_RESOURCE_TYPE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_METADATA_NEXT_MARKER;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_RESOURCE_TYPE_BLOB_PREFIX;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.FORWARD_SLASH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ROOT_PATH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.XML_TAG_HDI_ISFOLDER;
+import static
org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_META_HDI_ISFOLDER;
+import static
org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_ARROW_LIST_PARSING;
+
+/**
+ * Photon response parser implementation for Apache Arrow based ListBlobs
+ * responses. The parser reads the Arrow IPC stream returned by the service,
+ * iterates through the record batches and converts each row into a
+ * {@link BlobListResultEntrySchema}, producing a {@link BlobListResultSchema}
+ * that keeps the downstream FileStatus conversion path unchanged.
+ * <p>
+ * Arrow specific objects are confined to this class; the output contract is
+ * {@link BlobListResultSchema} so existing listing behavior is preserved.
+ * <p>
+ * Column matching is case-insensitive. Unknown columns are ignored and missing
+ * columns are treated as absent values so that additive schema changes do not
+ * break parsing. The continuation token is read from the Arrow schema custom
+ * metadata (key {@code NextMarker}).
+ */
+public class ArrowListBlobParser implements ListBlobResponseParser {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ArrowListBlobParser.class);
+
+ /**
+ * Upper bound, in bytes, on the heap staging buffer allocated per read when
+ * copying into a direct (non-array-backed) {@link ByteBuffer}. Caps
transient
+ * allocations for large reader requests; the reader simply issues additional
+ * reads for anything beyond this chunk.
+ */
+ private static final int NON_INTERRUPTIBLE_READ_CHUNK = 8192;
+
+ /**
+ * Base URL for which the ListBlobs API is called, used to build the absolute
+ * URL for each entry (mirrors the XML parser behavior).
+ */
+ private final String url;
+
+ /**
+ * Maximum off-heap (direct) memory in bytes the Arrow allocator may use
while
+ * parsing a single response. Bounds the memory an oversized or malformed
+ * response can consume; exceeding it fails the parse with an
+ * {@link org.apache.arrow.memory.OutOfMemoryException} that is surfaced as
an
+ * {@link IOException}.
+ */
+ private final long memoryLimitBytes;
+
+ /**
+ * @param url base URL for which the ListBlobs API is called.
+ * @param memoryLimitBytes maximum off-heap memory in bytes the Arrow
allocator
+ * may use while parsing a single response.
+ */
+ public ArrowListBlobParser(final String url, final long memoryLimitBytes) {
+ this.url = url;
+ this.memoryLimitBytes = memoryLimitBytes;
+ }
+
+ @Override
+ public BlobListResultSchema parse(final InputStream responseStream)
+ throws IOException {
+ BlobListResultSchema listResultSchema = new BlobListResultSchema();
+ try (BufferAllocator allocator = new RootAllocator(memoryLimitBytes);
+ ArrowStreamReader reader =
+ new ArrowStreamReader(nonInterruptibleChannel(responseStream),
+ allocator)) {
+ VectorSchemaRoot root = reader.getVectorSchemaRoot();
+ Schema schema = root.getSchema();
+
+ // Continuation token is carried in the Arrow schema custom metadata.
+ // The service emits an empty string when there is no continuation token,
+ // whereas the XML path leaves it null; normalize an absent or empty
+ // marker to null so both formats produce identical values.
+ Map<String, String> customMetadata = schema.getCustomMetadata();
+ if (customMetadata != null) {
+ String nextMarker = customMetadata.get(ARROW_METADATA_NEXT_MARKER);
+ listResultSchema.setNextMarker(
+ (nextMarker == null || nextMarker.isEmpty()) ? null : nextMarker);
+ }
+
+ while (reader.loadNextBatch()) {
+ BatchColumns columns = BatchColumns.resolve(root);
+ int rowCount = root.getRowCount();
+ for (int row = 0; row < rowCount; row++) {
+ BlobListResultEntrySchema entry = buildEntry(columns, row);
+ if (entry != null) {
+ listResultSchema.addBlobListEntry(entry);
+ }
+ }
+ }
+ LOG.debug("Photon ListBlobs parsed {} blobs with {} as continuation
token",
+ listResultSchema.paths().size(), listResultSchema.getNextMarker());
+ return listResultSchema;
+ } catch (IOException ex) {
+ throw ex;
+ } catch (RuntimeException ex) {
+ // Wrap Arrow runtime failures (e.g. corrupted stream) as IOException so
+ // callers can surface them through the existing error-handling model.
+ throw new IOException(ex);
+ }
+ }
+
+ @Override
+ public String getParsingErrorMessage() {
+ return ERR_ARROW_LIST_PARSING;
+ }
+
+ /**
+ * Wrap an {@link InputStream} in a {@link ReadableByteChannel} that is
+ * <em>not</em> interruptible.
+ *
+ * <p>{@link ArrowStreamReader}, when handed an {@code InputStream},
internally
+ * adapts it with {@link java.nio.channels.Channels#newChannel(InputStream)},
+ * which returns a channel extending
+ * {@link java.nio.channels.spi.AbstractInterruptibleChannel}. If the parsing
+ * thread's interrupt flag is set (e.g. Hadoop task cancellation, speculative
+ * execution kills, or an executor {@code shutdownNow()}), that channel
aborts
+ * the read with a {@link java.nio.channels.ClosedByInterruptException},
+ * turning a benign interrupt into a hard listing failure. The XML SAX parser
+ * reads from a plain {@code InputStream} and is immune to this, so the Arrow
+ * path must not be more fragile.</p>
+ *
+ * <p>The list response body is already fully buffered in memory before it
+ * reaches the parser, so an interruptible channel offers no benefit here;
this
+ * plain adapter simply reads from the buffer without consulting the thread's
+ * interrupt status.</p>
+ *
+ * @param in the buffered response stream.
+ * @return a non-interruptible channel over {@code in}.
+ */
+ private static ReadableByteChannel nonInterruptibleChannel(
+ final InputStream in) {
+ return new ReadableByteChannel() {
+ private volatile boolean open = true;
+ // Fixed staging buffer reused across reads for direct ByteBuffers. Arrow
+ // parsing drives this channel from a single thread, so a per-instance
+ // buffer is safe and avoids per-read allocation/GC churn.
+ private final byte[] staging = new byte[NON_INTERRUPTIBLE_READ_CHUNK];
+
+ @Override
+ public int read(final ByteBuffer dst) throws IOException {
+ final int toRead = dst.remaining();
+ if (toRead == 0) {
+ return 0;
+ }
+ if (dst.hasArray()) {
+ final int n = in.read(dst.array(),
+ dst.arrayOffset() + dst.position(), toRead);
+ if (n > 0) {
+ dst.position(dst.position() + n);
+ }
+ return n;
+ }
+ // For a direct (non-array-backed) ByteBuffer we must stage bytes in a
+ // heap buffer first. ArrowStreamReader can request very large reads,
so
+ // read at most one staging-buffer worth per call and let the reader
+ // issue further reads for the remainder (its readFully loops on
partial
+ // reads). This bounds allocation and reuses a single buffer.
+ final int chunk = Math.min(toRead, staging.length);
+ final int n = in.read(staging, 0, chunk);
+ if (n > 0) {
+ dst.put(staging, 0, n);
+ }
+ return n;
+ }
+
+ @Override
+ public boolean isOpen() {
+ return open;
+ }
+
+ @Override
+ public void close() throws IOException {
+ open = false;
+ in.close();
+ }
+ };
+ }
+
+ /**
+ * Convert a single Arrow row into a {@link BlobListResultEntrySchema}. Rows
+ * without a usable name are skipped.
+ */
+ private BlobListResultEntrySchema buildEntry(
+ final BatchColumns columns, final int row) {
+ String name = readString(columns.name, row);
+ if (name == null || name.isEmpty()) {
+ return null;
+ }
+ // Directory names may carry a trailing slash; strip it to match XML
parser.
+ if (name.endsWith(FORWARD_SLASH)) {
+ name = name.substring(0, name.length() - 1);
+ }
+
+ BlobListResultEntrySchema entry = new BlobListResultEntrySchema();
+ entry.setName(name);
+ entry.setPath(new Path(ROOT_PATH + name));
+ entry.setUrl(url + ROOT_PATH + name);
+
+ entry.setETag(readString(columns.etag, row));
+ // The Arrow (Photon) response serializes timestamps as native Arrow
+ // timestamp vectors (e.g. Creation-Time / Last-Modified as TimeStampSec),
+ // whose object form is an ISO-8601 local date-time (e.g.
2026-07-06T10:31:19)
+ // in UTC. Normalize to RFC 1123 GMT so both paths produce identical values
+ // and DateTimeUtils.parseLastModifiedTime yields the correct time.
+ entry.setLastModifiedTime(readTimestampAsRfc1123(columns.lastModified,
row));
+ entry.setCreationTime(readTimestampAsRfc1123(columns.creationTime, row));
+ entry.setOwner(readString(columns.owner, row));
+ entry.setGroup(readString(columns.group, row));
+ entry.setPermission(readString(columns.permission, row));
+ entry.setAcl(readString(columns.acl, row));
+
+ // Blob user metadata (including the hdi_isfolder directory marker) is
+ // carried in a single Arrow map column; mirror the XML parser which
+ // populates the metadata map from the <Metadata> element.
+ Map<String, String> metadata = readMetadata(columns.metadata, row);
+ if (!metadata.isEmpty()) {
+ entry.setMetadata(metadata);
+ }
+
+ setContentLength(entry, columns.contentLength, row);
+ setCopyProperties(entry, columns, row);
+ entry.setIsDirectory(resolveIsDirectory(columns, metadata, row));
+ return entry;
+ }
+
+ /**
+ * Populate the copy-related properties, mirroring the XML parser's handling
+ * of the {@code Properties} copy elements. Absent columns leave the
+ * corresponding field unset, exactly as the XML path does when the elements
+ * are missing.
+ */
+ private void setCopyProperties(final BlobListResultEntrySchema entry,
+ final BatchColumns columns, final int row) {
+ entry.setCopyId(readString(columns.copyId, row));
+ entry.setCopyStatus(readString(columns.copyStatus, row));
+ entry.setCopySourceUrl(readString(columns.copySource, row));
+ entry.setCopyProgress(readString(columns.copyProgress, row));
+ entry.setCopyStatusDescription(
+ readString(columns.copyStatusDescription, row));
+ // XML stores the completion time as epoch millis parsed from an RFC 1123
+ // string; normalize the Arrow timestamp to the same RFC 1123 form first so
+ // both paths yield identical epoch values.
+ String copyCompletion =
+ readTimestampAsRfc1123(columns.copyCompletionTime, row);
+ if (copyCompletion != null) {
+ entry.setCopyCompletionTime(
+ DateTimeUtils.parseLastModifiedTime(copyCompletion));
+ }
+ }
+
+ /**
+ * Populate the content length from the content-length column. The Blob
+ * endpoint surfaces it as a native unsigned 64-bit integer
+ * ({@code UInt8Vector}); other numeric vector types and a numeric string
+ * column are also tolerated so additive schema changes do not break parsing.
+ */
+ private void setContentLength(final BlobListResultEntrySchema entry,
+ final FieldVector contentLength, final int row) {
+ Long value = readLong(contentLength, row);
+ if (value != null) {
+ entry.setContentLength(value);
+ }
+ }
+
+ /**
+ * Determine whether the row represents a directory. Mirrors the XML parser
+ * ({@code BlobListXmlParser}), which flags an entry as a directory when any
of
+ * the following hold:
+ * <ul>
+ * <li>the entry is a {@code BlobPrefix} - surfaced in the Arrow response
as
+ * a {@code ResourceType} of {@code blobprefix} - i.e. an implicit
+ * directory;</li>
+ * <li>a {@code ResourceType} of {@code directory};</li>
+ * <li>an explicit boolean {@code IsDirectory} indicator;</li>
+ * <li>the {@code hdi_isfolder} user metadata being {@code true}.</li>
+ * </ul>
+ * The implicit-directory ({@code blobprefix}) and {@code hdi_isfolder} cases
+ * are essential: an implicit directory has no marker blob at all, while an
+ * empty directory created by {@code mkdir} is a zero-byte marker blob whose
+ * only directory indicator is the {@code hdi_isfolder=true} metadata entry.
+ * The metadata key is matched case-insensitively, matching the XML parser
+ * ({@code equalsIgnoreCase}). Without honoring these, such directories are
+ * misclassified as files, diverging from the XML listing path.
+ */
+ private boolean resolveIsDirectory(final BatchColumns columns,
+ final Map<String, String> metadata, final int row) {
+ String resourceType = readString(columns.resourceType, row);
+ if (DIRECTORY.equalsIgnoreCase(resourceType)
+ || ARROW_RESOURCE_TYPE_BLOB_PREFIX.equalsIgnoreCase(resourceType)) {
+ return true;
+ }
+ if (readBoolean(columns.isDirectory, row)) {
+ return true;
+ }
+ String hdiIsFolder = metadataValueIgnoreCase(metadata,
XML_TAG_HDI_ISFOLDER);
+ if (hdiIsFolder != null && Boolean.parseBoolean(hdiIsFolder.trim())) {
+ return true;
+ }
+ // Fallback for a schema that flattens the marker metadata into a dedicated
+ // column instead of the metadata map.
+ return readBoolean(columns.hdiIsFolder, row);
+ }
+
+ /**
+ * Case-insensitive lookup of a metadata value by key, mirroring the XML
+ * parser's {@code equalsIgnoreCase} matching of the {@code hdi_isfolder}
+ * marker (the service preserves the casing used when the metadata was set).
+ */
+ private static String metadataValueIgnoreCase(
+ final Map<String, String> metadata, final String key) {
+ String value = metadata.get(key);
+ if (value != null || metadata.containsKey(key)) {
+ return value;
+ }
+ for (Map.Entry<String, String> entry : metadata.entrySet()) {
+ if (key.equalsIgnoreCase(entry.getKey())) {
+ return entry.getValue();
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Read the user metadata key/value pairs from the Arrow map column for the
+ * given row. Keys are stored verbatim (e.g. {@code hdi_isfolder}) to match
the
+ * XML parser. Returns an empty (mutable) map when the column is absent or
the
+ * value is null.
+ */
+ private Map<String, String> readMetadata(final MapVector metadata,
+ final int row) {
+ Map<String, String> result = new HashMap<>();
+ if (metadata == null || row >= metadata.getValueCount()
+ || metadata.isNull(row)) {
+ return result;
+ }
+ List<?> entries = metadata.getObject(row);
Review 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?
##########
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/contract/TestArrowListBlobParser.java:
##########
@@ -0,0 +1,974 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.contract;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.channels.Channels;
+import java.nio.charset.StandardCharsets;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.BigIntVector;
+import org.apache.arrow.vector.BitVector;
+import org.apache.arrow.vector.TimeStampSecVector;
+import org.apache.arrow.vector.UInt8Vector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.complex.MapVector;
+import org.apache.arrow.vector.complex.impl.UnionMapWriter;
+import org.apache.arrow.vector.ipc.ArrowStreamWriter;
+import org.apache.arrow.vector.types.TimeUnit;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.arrow.vector.util.Text;
+import org.junit.jupiter.api.Test;
+
+import org.apache.hadoop.fs.azurebfs.contracts.services.ArrowListBlobParser;
+import
org.apache.hadoop.fs.azurebfs.contracts.services.BlobListResultEntrySchema;
+import org.apache.hadoop.fs.azurebfs.contracts.services.BlobListResultSchema;
+import org.apache.hadoop.fs.azurebfs.utils.DateTimeUtils;
+
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CONTENT_LENGTH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_COMPLETION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_ID;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_PROGRESS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_SOURCE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS_DESCRIPTION;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CREATION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_ETAG;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_IS_DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_LAST_MODIFIED;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_METADATA;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_NAME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_RESOURCE_TYPE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_METADATA_NEXT_MARKER;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_RESOURCE_TYPE_BLOB_PREFIX;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.XML_TAG_HDI_ISFOLDER;
+import static
org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_META_HDI_ISFOLDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Unit tests for {@link ArrowListBlobParser}, the Photon (Apache Arrow based)
+ * ListBlobs response parser.
+ */
+public class TestArrowListBlobParser {
Review Comment:
Added: TZ timestamp, two batches in one stream, negative UInt8 length,
non-numeric length, hdi_isfolder="1" column, and a name of just "/".
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/services/ArrowListBlobParser.java:
##########
@@ -0,0 +1,590 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.contracts.services;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.BitVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.complex.MapVector;
+import org.apache.arrow.vector.ipc.ArrowStreamReader;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.utils.DateTimeUtils;
+
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_ACL;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CONTENT_LENGTH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_COMPLETION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_ID;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_PROGRESS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_SOURCE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS_DESCRIPTION;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CREATION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_ETAG;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_GROUP;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_IS_DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_LAST_MODIFIED;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_METADATA;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_NAME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_OWNER;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_PERMISSIONS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_RESOURCE_TYPE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_METADATA_NEXT_MARKER;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_RESOURCE_TYPE_BLOB_PREFIX;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.FORWARD_SLASH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ROOT_PATH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.XML_TAG_HDI_ISFOLDER;
+import static
org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_META_HDI_ISFOLDER;
+import static
org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_ARROW_LIST_PARSING;
+
+/**
+ * Photon response parser implementation for Apache Arrow based ListBlobs
+ * responses. The parser reads the Arrow IPC stream returned by the service,
+ * iterates through the record batches and converts each row into a
+ * {@link BlobListResultEntrySchema}, producing a {@link BlobListResultSchema}
+ * that keeps the downstream FileStatus conversion path unchanged.
+ * <p>
+ * Arrow specific objects are confined to this class; the output contract is
+ * {@link BlobListResultSchema} so existing listing behavior is preserved.
+ * <p>
+ * Column matching is case-insensitive. Unknown columns are ignored and missing
+ * columns are treated as absent values so that additive schema changes do not
+ * break parsing. The continuation token is read from the Arrow schema custom
+ * metadata (key {@code NextMarker}).
+ */
+public class ArrowListBlobParser implements ListBlobResponseParser {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ArrowListBlobParser.class);
+
+ /**
+ * Upper bound, in bytes, on the heap staging buffer allocated per read when
+ * copying into a direct (non-array-backed) {@link ByteBuffer}. Caps
transient
+ * allocations for large reader requests; the reader simply issues additional
+ * reads for anything beyond this chunk.
+ */
+ private static final int NON_INTERRUPTIBLE_READ_CHUNK = 8192;
+
+ /**
+ * Base URL for which the ListBlobs API is called, used to build the absolute
+ * URL for each entry (mirrors the XML parser behavior).
+ */
+ private final String url;
+
+ /**
+ * Maximum off-heap (direct) memory in bytes the Arrow allocator may use
while
+ * parsing a single response. Bounds the memory an oversized or malformed
+ * response can consume; exceeding it fails the parse with an
+ * {@link org.apache.arrow.memory.OutOfMemoryException} that is surfaced as
an
+ * {@link IOException}.
+ */
+ private final long memoryLimitBytes;
+
+ /**
+ * @param url base URL for which the ListBlobs API is called.
+ * @param memoryLimitBytes maximum off-heap memory in bytes the Arrow
allocator
+ * may use while parsing a single response.
+ */
+ public ArrowListBlobParser(final String url, final long memoryLimitBytes) {
+ this.url = url;
+ this.memoryLimitBytes = memoryLimitBytes;
+ }
+
+ @Override
+ public BlobListResultSchema parse(final InputStream responseStream)
+ throws IOException {
+ BlobListResultSchema listResultSchema = new BlobListResultSchema();
+ try (BufferAllocator allocator = new RootAllocator(memoryLimitBytes);
+ ArrowStreamReader reader =
+ new ArrowStreamReader(nonInterruptibleChannel(responseStream),
+ allocator)) {
+ VectorSchemaRoot root = reader.getVectorSchemaRoot();
+ Schema schema = root.getSchema();
+
+ // Continuation token is carried in the Arrow schema custom metadata.
+ // The service emits an empty string when there is no continuation token,
+ // whereas the XML path leaves it null; normalize an absent or empty
+ // marker to null so both formats produce identical values.
+ Map<String, String> customMetadata = schema.getCustomMetadata();
+ if (customMetadata != null) {
+ String nextMarker = customMetadata.get(ARROW_METADATA_NEXT_MARKER);
+ listResultSchema.setNextMarker(
+ (nextMarker == null || nextMarker.isEmpty()) ? null : nextMarker);
+ }
+
+ while (reader.loadNextBatch()) {
+ BatchColumns columns = BatchColumns.resolve(root);
Review 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.
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/services/ArrowListBlobParser.java:
##########
@@ -0,0 +1,590 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.contracts.services;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.BitVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.complex.MapVector;
+import org.apache.arrow.vector.ipc.ArrowStreamReader;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.utils.DateTimeUtils;
+
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_ACL;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CONTENT_LENGTH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_COMPLETION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_ID;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_PROGRESS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_SOURCE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS_DESCRIPTION;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CREATION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_ETAG;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_GROUP;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_IS_DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_LAST_MODIFIED;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_METADATA;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_NAME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_OWNER;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_PERMISSIONS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_RESOURCE_TYPE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_METADATA_NEXT_MARKER;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_RESOURCE_TYPE_BLOB_PREFIX;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.FORWARD_SLASH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ROOT_PATH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.XML_TAG_HDI_ISFOLDER;
+import static
org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_META_HDI_ISFOLDER;
+import static
org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_ARROW_LIST_PARSING;
+
+/**
+ * Photon response parser implementation for Apache Arrow based ListBlobs
+ * responses. The parser reads the Arrow IPC stream returned by the service,
+ * iterates through the record batches and converts each row into a
+ * {@link BlobListResultEntrySchema}, producing a {@link BlobListResultSchema}
+ * that keeps the downstream FileStatus conversion path unchanged.
+ * <p>
+ * Arrow specific objects are confined to this class; the output contract is
+ * {@link BlobListResultSchema} so existing listing behavior is preserved.
+ * <p>
+ * Column matching is case-insensitive. Unknown columns are ignored and missing
+ * columns are treated as absent values so that additive schema changes do not
+ * break parsing. The continuation token is read from the Arrow schema custom
+ * metadata (key {@code NextMarker}).
+ */
+public class ArrowListBlobParser implements ListBlobResponseParser {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ArrowListBlobParser.class);
+
+ /**
+ * Upper bound, in bytes, on the heap staging buffer allocated per read when
+ * copying into a direct (non-array-backed) {@link ByteBuffer}. Caps
transient
+ * allocations for large reader requests; the reader simply issues additional
+ * reads for anything beyond this chunk.
+ */
+ private static final int NON_INTERRUPTIBLE_READ_CHUNK = 8192;
+
+ /**
+ * Base URL for which the ListBlobs API is called, used to build the absolute
+ * URL for each entry (mirrors the XML parser behavior).
+ */
+ private final String url;
+
+ /**
+ * Maximum off-heap (direct) memory in bytes the Arrow allocator may use
while
+ * parsing a single response. Bounds the memory an oversized or malformed
+ * response can consume; exceeding it fails the parse with an
+ * {@link org.apache.arrow.memory.OutOfMemoryException} that is surfaced as
an
+ * {@link IOException}.
+ */
+ private final long memoryLimitBytes;
+
+ /**
+ * @param url base URL for which the ListBlobs API is called.
+ * @param memoryLimitBytes maximum off-heap memory in bytes the Arrow
allocator
+ * may use while parsing a single response.
+ */
+ public ArrowListBlobParser(final String url, final long memoryLimitBytes) {
+ this.url = url;
+ this.memoryLimitBytes = memoryLimitBytes;
+ }
+
+ @Override
+ public BlobListResultSchema parse(final InputStream responseStream)
+ throws IOException {
+ BlobListResultSchema listResultSchema = new BlobListResultSchema();
+ try (BufferAllocator allocator = new RootAllocator(memoryLimitBytes);
+ ArrowStreamReader reader =
+ new ArrowStreamReader(nonInterruptibleChannel(responseStream),
+ allocator)) {
+ VectorSchemaRoot root = reader.getVectorSchemaRoot();
+ Schema schema = root.getSchema();
+
+ // Continuation token is carried in the Arrow schema custom metadata.
+ // The service emits an empty string when there is no continuation token,
+ // whereas the XML path leaves it null; normalize an absent or empty
+ // marker to null so both formats produce identical values.
+ Map<String, String> customMetadata = schema.getCustomMetadata();
+ if (customMetadata != null) {
+ String nextMarker = customMetadata.get(ARROW_METADATA_NEXT_MARKER);
+ listResultSchema.setNextMarker(
+ (nextMarker == null || nextMarker.isEmpty()) ? null : nextMarker);
+ }
+
+ while (reader.loadNextBatch()) {
+ BatchColumns columns = BatchColumns.resolve(root);
+ int rowCount = root.getRowCount();
+ for (int row = 0; row < rowCount; row++) {
+ BlobListResultEntrySchema entry = buildEntry(columns, row);
+ if (entry != null) {
+ listResultSchema.addBlobListEntry(entry);
+ }
+ }
+ }
+ LOG.debug("Photon ListBlobs parsed {} blobs with {} as continuation
token",
+ listResultSchema.paths().size(), listResultSchema.getNextMarker());
+ return listResultSchema;
+ } catch (IOException ex) {
+ throw ex;
+ } catch (RuntimeException ex) {
+ // Wrap Arrow runtime failures (e.g. corrupted stream) as IOException so
+ // callers can surface them through the existing error-handling model.
+ throw new IOException(ex);
+ }
+ }
+
+ @Override
+ public String getParsingErrorMessage() {
+ return ERR_ARROW_LIST_PARSING;
+ }
+
+ /**
+ * Wrap an {@link InputStream} in a {@link ReadableByteChannel} that is
+ * <em>not</em> interruptible.
+ *
+ * <p>{@link ArrowStreamReader}, when handed an {@code InputStream},
internally
+ * adapts it with {@link java.nio.channels.Channels#newChannel(InputStream)},
+ * which returns a channel extending
+ * {@link java.nio.channels.spi.AbstractInterruptibleChannel}. If the parsing
+ * thread's interrupt flag is set (e.g. Hadoop task cancellation, speculative
+ * execution kills, or an executor {@code shutdownNow()}), that channel
aborts
+ * the read with a {@link java.nio.channels.ClosedByInterruptException},
+ * turning a benign interrupt into a hard listing failure. The XML SAX parser
+ * reads from a plain {@code InputStream} and is immune to this, so the Arrow
+ * path must not be more fragile.</p>
+ *
+ * <p>The list response body is already fully buffered in memory before it
+ * reaches the parser, so an interruptible channel offers no benefit here;
this
+ * plain adapter simply reads from the buffer without consulting the thread's
+ * interrupt status.</p>
+ *
+ * @param in the buffered response stream.
+ * @return a non-interruptible channel over {@code in}.
+ */
+ private static ReadableByteChannel nonInterruptibleChannel(
+ final InputStream in) {
+ return new ReadableByteChannel() {
+ private volatile boolean open = true;
+ // Fixed staging buffer reused across reads for direct ByteBuffers. Arrow
+ // parsing drives this channel from a single thread, so a per-instance
+ // buffer is safe and avoids per-read allocation/GC churn.
+ private final byte[] staging = new byte[NON_INTERRUPTIBLE_READ_CHUNK];
+
+ @Override
+ public int read(final ByteBuffer dst) throws IOException {
+ final int toRead = dst.remaining();
+ if (toRead == 0) {
+ return 0;
+ }
+ if (dst.hasArray()) {
+ final int n = in.read(dst.array(),
+ dst.arrayOffset() + dst.position(), toRead);
+ if (n > 0) {
+ dst.position(dst.position() + n);
+ }
+ return n;
+ }
+ // For a direct (non-array-backed) ByteBuffer we must stage bytes in a
+ // heap buffer first. ArrowStreamReader can request very large reads,
so
+ // read at most one staging-buffer worth per call and let the reader
+ // issue further reads for the remainder (its readFully loops on
partial
+ // reads). This bounds allocation and reuses a single buffer.
+ final int chunk = Math.min(toRead, staging.length);
+ final int n = in.read(staging, 0, chunk);
+ if (n > 0) {
+ dst.put(staging, 0, n);
+ }
+ return n;
+ }
+
+ @Override
+ public boolean isOpen() {
+ return open;
+ }
+
+ @Override
+ public void close() throws IOException {
+ open = false;
+ in.close();
+ }
+ };
+ }
+
+ /**
+ * Convert a single Arrow row into a {@link BlobListResultEntrySchema}. Rows
+ * without a usable name are skipped.
+ */
+ private BlobListResultEntrySchema buildEntry(
+ final BatchColumns columns, final int row) {
+ String name = readString(columns.name, row);
+ if (name == null || name.isEmpty()) {
+ return null;
+ }
+ // Directory names may carry a trailing slash; strip it to match XML
parser.
+ if (name.endsWith(FORWARD_SLASH)) {
+ name = name.substring(0, name.length() - 1);
+ }
+
+ BlobListResultEntrySchema entry = new BlobListResultEntrySchema();
+ entry.setName(name);
+ entry.setPath(new Path(ROOT_PATH + name));
+ entry.setUrl(url + ROOT_PATH + name);
+
+ entry.setETag(readString(columns.etag, row));
+ // The Arrow (Photon) response serializes timestamps as native Arrow
+ // timestamp vectors (e.g. Creation-Time / Last-Modified as TimeStampSec),
+ // whose object form is an ISO-8601 local date-time (e.g.
2026-07-06T10:31:19)
+ // in UTC. Normalize to RFC 1123 GMT so both paths produce identical values
+ // and DateTimeUtils.parseLastModifiedTime yields the correct time.
+ entry.setLastModifiedTime(readTimestampAsRfc1123(columns.lastModified,
row));
+ entry.setCreationTime(readTimestampAsRfc1123(columns.creationTime, row));
+ entry.setOwner(readString(columns.owner, row));
+ entry.setGroup(readString(columns.group, row));
+ entry.setPermission(readString(columns.permission, row));
+ entry.setAcl(readString(columns.acl, row));
+
+ // Blob user metadata (including the hdi_isfolder directory marker) is
+ // carried in a single Arrow map column; mirror the XML parser which
+ // populates the metadata map from the <Metadata> element.
+ Map<String, String> metadata = readMetadata(columns.metadata, row);
+ if (!metadata.isEmpty()) {
+ entry.setMetadata(metadata);
+ }
+
+ setContentLength(entry, columns.contentLength, row);
+ setCopyProperties(entry, columns, row);
+ entry.setIsDirectory(resolveIsDirectory(columns, metadata, row));
+ return entry;
+ }
+
+ /**
+ * Populate the copy-related properties, mirroring the XML parser's handling
+ * of the {@code Properties} copy elements. Absent columns leave the
+ * corresponding field unset, exactly as the XML path does when the elements
+ * are missing.
+ */
+ private void setCopyProperties(final BlobListResultEntrySchema entry,
+ final BatchColumns columns, final int row) {
+ entry.setCopyId(readString(columns.copyId, row));
+ entry.setCopyStatus(readString(columns.copyStatus, row));
+ entry.setCopySourceUrl(readString(columns.copySource, row));
+ entry.setCopyProgress(readString(columns.copyProgress, row));
+ entry.setCopyStatusDescription(
+ readString(columns.copyStatusDescription, row));
+ // XML stores the completion time as epoch millis parsed from an RFC 1123
+ // string; normalize the Arrow timestamp to the same RFC 1123 form first so
+ // both paths yield identical epoch values.
+ String copyCompletion =
+ readTimestampAsRfc1123(columns.copyCompletionTime, row);
+ if (copyCompletion != null) {
+ entry.setCopyCompletionTime(
+ DateTimeUtils.parseLastModifiedTime(copyCompletion));
+ }
+ }
+
+ /**
+ * Populate the content length from the content-length column. The Blob
+ * endpoint surfaces it as a native unsigned 64-bit integer
+ * ({@code UInt8Vector}); other numeric vector types and a numeric string
+ * column are also tolerated so additive schema changes do not break parsing.
+ */
+ private void setContentLength(final BlobListResultEntrySchema entry,
+ final FieldVector contentLength, final int row) {
+ Long value = readLong(contentLength, row);
+ if (value != null) {
+ entry.setContentLength(value);
+ }
+ }
+
+ /**
+ * Determine whether the row represents a directory. Mirrors the XML parser
+ * ({@code BlobListXmlParser}), which flags an entry as a directory when any
of
+ * the following hold:
+ * <ul>
+ * <li>the entry is a {@code BlobPrefix} - surfaced in the Arrow response
as
+ * a {@code ResourceType} of {@code blobprefix} - i.e. an implicit
+ * directory;</li>
+ * <li>a {@code ResourceType} of {@code directory};</li>
+ * <li>an explicit boolean {@code IsDirectory} indicator;</li>
+ * <li>the {@code hdi_isfolder} user metadata being {@code true}.</li>
+ * </ul>
+ * The implicit-directory ({@code blobprefix}) and {@code hdi_isfolder} cases
+ * are essential: an implicit directory has no marker blob at all, while an
+ * empty directory created by {@code mkdir} is a zero-byte marker blob whose
+ * only directory indicator is the {@code hdi_isfolder=true} metadata entry.
+ * The metadata key is matched case-insensitively, matching the XML parser
+ * ({@code equalsIgnoreCase}). Without honoring these, such directories are
+ * misclassified as files, diverging from the XML listing path.
+ */
+ private boolean resolveIsDirectory(final BatchColumns columns,
+ final Map<String, String> metadata, final int row) {
+ String resourceType = readString(columns.resourceType, row);
+ if (DIRECTORY.equalsIgnoreCase(resourceType)
+ || ARROW_RESOURCE_TYPE_BLOB_PREFIX.equalsIgnoreCase(resourceType)) {
+ return true;
+ }
+ if (readBoolean(columns.isDirectory, row)) {
+ return true;
+ }
+ String hdiIsFolder = metadataValueIgnoreCase(metadata,
XML_TAG_HDI_ISFOLDER);
+ if (hdiIsFolder != null && Boolean.parseBoolean(hdiIsFolder.trim())) {
+ return true;
+ }
+ // Fallback for a schema that flattens the marker metadata into a dedicated
+ // column instead of the metadata map.
+ return readBoolean(columns.hdiIsFolder, row);
+ }
+
+ /**
+ * Case-insensitive lookup of a metadata value by key, mirroring the XML
+ * parser's {@code equalsIgnoreCase} matching of the {@code hdi_isfolder}
+ * marker (the service preserves the casing used when the metadata was set).
+ */
+ private static String metadataValueIgnoreCase(
+ final Map<String, String> metadata, final String key) {
+ String value = metadata.get(key);
+ if (value != null || metadata.containsKey(key)) {
+ return value;
+ }
+ for (Map.Entry<String, String> entry : metadata.entrySet()) {
+ if (key.equalsIgnoreCase(entry.getKey())) {
+ return entry.getValue();
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Read the user metadata key/value pairs from the Arrow map column for the
+ * given row. Keys are stored verbatim (e.g. {@code hdi_isfolder}) to match
the
+ * XML parser. Returns an empty (mutable) map when the column is absent or
the
+ * value is null.
+ */
+ private Map<String, String> readMetadata(final MapVector metadata,
+ final int row) {
+ Map<String, String> result = new HashMap<>();
+ if (metadata == null || row >= metadata.getValueCount()
+ || metadata.isNull(row)) {
+ return result;
+ }
+ List<?> entries = metadata.getObject(row);
+ if (entries == null) {
+ return result;
+ }
+ for (Object element : entries) {
+ if (element instanceof Map) {
+ Map<?, ?> keyValue = (Map<?, ?>) element;
+ Object key = keyValue.get(MapVector.KEY_NAME);
+ Object value = keyValue.get(MapVector.VALUE_NAME);
+ if (key != null) {
+ result.put(key.toString(), value == null ? null : value.toString());
+ }
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Read a timestamp column for the given row and convert it to the RFC 1123
+ * GMT representation used by the XML path. Native Arrow timestamp vectors
+ * expose an ISO-8601 local date-time via {@link FieldVector#getObject(int)};
+ * a textual (VarChar) timestamp column is also tolerated. Returns
+ * {@code null} when the column is absent or the value is null.
+ */
+ private String readTimestampAsRfc1123(final FieldVector vector,
+ final int row) {
+ if (vector == null || row >= vector.getValueCount()
+ || vector.isNull(row)) {
+ return null;
+ }
+ Object value = vector.getObject(row);
+ if (value == null) {
+ return null;
+ }
+ return DateTimeUtils.formatArrowDateTimeToRfc1123(value.toString());
+ }
+
+ /**
+ * Read a numeric column for the given row as a {@code long}. Supports native
+ * Arrow integer vectors (whose object form is a {@link Number}) and a
numeric
+ * {@link VarCharVector}. Returns {@code null} when the column is absent, the
+ * value is null, or a textual value is not parseable.
+ */
+ private Long readLong(final FieldVector vector, final int row) {
+ if (vector == null || row >= vector.getValueCount()
+ || vector.isNull(row)) {
+ return null;
+ }
+ Object value = vector.getObject(row);
+ if (value instanceof Number) {
+ return ((Number) value).longValue();
+ }
+ if (value != null) {
+ String text = value.toString().trim();
+ if (!text.isEmpty()) {
+ try {
+ return Long.parseLong(text);
+ } catch (NumberFormatException ignored) {
+ // Leave unset if the value is not numeric.
+ }
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Read a boolean-valued column for the given row. Supports both a native
+ * Arrow {@link BitVector} and a {@link VarCharVector} carrying a textual
+ * {@code true}/{@code 1} value (matching how the XML parser interprets the
+ * {@code hdi_isfolder} metadata via {@link Boolean#valueOf(String)}).
Returns
+ * {@code false} when the column is absent or the value is null/empty.
+ */
+ private boolean readBoolean(final FieldVector vector, final int row) {
+ if (vector == null || row >= vector.getValueCount()
+ || vector.isNull(row)) {
+ return false;
+ }
+ if (vector instanceof BitVector) {
+ return ((BitVector) vector).get(row) != 0;
+ }
+ if (vector instanceof VarCharVector) {
+ String value = readString((VarCharVector) vector, row);
+ if (value != null && !value.isEmpty()) {
+ String trimmed = value.trim();
+ return Boolean.parseBoolean(trimmed) || "1".equals(trimmed);
Review Comment:
Extracted NUMERIC_TRUE = "1".
##########
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAggregateMetricsManager.java:
##########
@@ -667,6 +667,31 @@ private void runProgramAndCaptureOutput(String program,
}
}
+ /**
+ * Resolves a JDK tool (e.g. {@code javac}, {@code java}) from the JDK that
is
+ * currently running this test, rather than relying on whichever version is
+ * first on the {@code PATH}. The spawned {@code javac} must read the
compiled
+ * classes of this project (and its dependencies) off the classpath; those
are
+ * emitted by the same JDK that runs the tests. If a mismatched compiler is
+ * picked from the {@code PATH} (e.g. a Java 8 {@code javac} against Java 17
+ * bytecode), compilation fails with "class file has wrong version".
Anchoring
+ * on {@code java.home} keeps the toolchain consistent. Falls back to the
bare
+ * tool name when the JDK layout cannot be resolved.
+ *
+ * @param name the tool name, e.g. {@code javac} or {@code java}.
+ * @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");
Review Comment:
Added JAVA_HOME_PROPERTY and JDK_BIN_DIR constants.
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/services/ArrowListBlobParser.java:
##########
@@ -0,0 +1,590 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.contracts.services;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.BitVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.complex.MapVector;
+import org.apache.arrow.vector.ipc.ArrowStreamReader;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.utils.DateTimeUtils;
+
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_ACL;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CONTENT_LENGTH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_COMPLETION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_ID;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_PROGRESS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_SOURCE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS_DESCRIPTION;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CREATION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_ETAG;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_GROUP;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_IS_DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_LAST_MODIFIED;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_METADATA;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_NAME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_OWNER;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_PERMISSIONS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_RESOURCE_TYPE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_METADATA_NEXT_MARKER;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_RESOURCE_TYPE_BLOB_PREFIX;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.FORWARD_SLASH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ROOT_PATH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.XML_TAG_HDI_ISFOLDER;
+import static
org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_META_HDI_ISFOLDER;
+import static
org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_ARROW_LIST_PARSING;
+
+/**
+ * Photon response parser implementation for Apache Arrow based ListBlobs
+ * responses. The parser reads the Arrow IPC stream returned by the service,
+ * iterates through the record batches and converts each row into a
+ * {@link BlobListResultEntrySchema}, producing a {@link BlobListResultSchema}
+ * that keeps the downstream FileStatus conversion path unchanged.
+ * <p>
+ * Arrow specific objects are confined to this class; the output contract is
+ * {@link BlobListResultSchema} so existing listing behavior is preserved.
+ * <p>
+ * Column matching is case-insensitive. Unknown columns are ignored and missing
+ * columns are treated as absent values so that additive schema changes do not
+ * break parsing. The continuation token is read from the Arrow schema custom
+ * metadata (key {@code NextMarker}).
+ */
+public class ArrowListBlobParser implements ListBlobResponseParser {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ArrowListBlobParser.class);
+
+ /**
+ * Upper bound, in bytes, on the heap staging buffer allocated per read when
+ * copying into a direct (non-array-backed) {@link ByteBuffer}. Caps
transient
+ * allocations for large reader requests; the reader simply issues additional
+ * reads for anything beyond this chunk.
+ */
+ private static final int NON_INTERRUPTIBLE_READ_CHUNK = 8192;
+
+ /**
+ * Base URL for which the ListBlobs API is called, used to build the absolute
+ * URL for each entry (mirrors the XML parser behavior).
+ */
+ private final String url;
+
+ /**
+ * Maximum off-heap (direct) memory in bytes the Arrow allocator may use
while
+ * parsing a single response. Bounds the memory an oversized or malformed
+ * response can consume; exceeding it fails the parse with an
+ * {@link org.apache.arrow.memory.OutOfMemoryException} that is surfaced as
an
+ * {@link IOException}.
+ */
+ private final long memoryLimitBytes;
+
+ /**
+ * @param url base URL for which the ListBlobs API is called.
+ * @param memoryLimitBytes maximum off-heap memory in bytes the Arrow
allocator
+ * may use while parsing a single response.
+ */
+ public ArrowListBlobParser(final String url, final long memoryLimitBytes) {
+ this.url = url;
+ this.memoryLimitBytes = memoryLimitBytes;
+ }
+
+ @Override
+ public BlobListResultSchema parse(final InputStream responseStream)
+ throws IOException {
+ BlobListResultSchema listResultSchema = new BlobListResultSchema();
+ try (BufferAllocator allocator = new RootAllocator(memoryLimitBytes);
+ ArrowStreamReader reader =
+ new ArrowStreamReader(nonInterruptibleChannel(responseStream),
+ allocator)) {
+ VectorSchemaRoot root = reader.getVectorSchemaRoot();
+ Schema schema = root.getSchema();
+
+ // Continuation token is carried in the Arrow schema custom metadata.
+ // The service emits an empty string when there is no continuation token,
+ // whereas the XML path leaves it null; normalize an absent or empty
+ // marker to null so both formats produce identical values.
+ Map<String, String> customMetadata = schema.getCustomMetadata();
+ if (customMetadata != null) {
+ String nextMarker = customMetadata.get(ARROW_METADATA_NEXT_MARKER);
+ listResultSchema.setNextMarker(
+ (nextMarker == null || nextMarker.isEmpty()) ? null : nextMarker);
+ }
+
+ while (reader.loadNextBatch()) {
+ BatchColumns columns = BatchColumns.resolve(root);
+ int rowCount = root.getRowCount();
+ for (int row = 0; row < rowCount; row++) {
+ BlobListResultEntrySchema entry = buildEntry(columns, row);
+ if (entry != null) {
+ listResultSchema.addBlobListEntry(entry);
+ }
+ }
+ }
+ LOG.debug("Photon ListBlobs parsed {} blobs with {} as continuation
token",
+ listResultSchema.paths().size(), listResultSchema.getNextMarker());
+ return listResultSchema;
+ } catch (IOException ex) {
+ throw ex;
+ } catch (RuntimeException ex) {
+ // Wrap Arrow runtime failures (e.g. corrupted stream) as IOException so
+ // callers can surface them through the existing error-handling model.
+ throw new IOException(ex);
+ }
+ }
+
+ @Override
+ public String getParsingErrorMessage() {
+ return ERR_ARROW_LIST_PARSING;
+ }
+
+ /**
+ * Wrap an {@link InputStream} in a {@link ReadableByteChannel} that is
+ * <em>not</em> interruptible.
+ *
+ * <p>{@link ArrowStreamReader}, when handed an {@code InputStream},
internally
+ * adapts it with {@link java.nio.channels.Channels#newChannel(InputStream)},
+ * which returns a channel extending
+ * {@link java.nio.channels.spi.AbstractInterruptibleChannel}. If the parsing
+ * thread's interrupt flag is set (e.g. Hadoop task cancellation, speculative
+ * execution kills, or an executor {@code shutdownNow()}), that channel
aborts
+ * the read with a {@link java.nio.channels.ClosedByInterruptException},
+ * turning a benign interrupt into a hard listing failure. The XML SAX parser
+ * reads from a plain {@code InputStream} and is immune to this, so the Arrow
+ * path must not be more fragile.</p>
+ *
+ * <p>The list response body is already fully buffered in memory before it
+ * reaches the parser, so an interruptible channel offers no benefit here;
this
+ * plain adapter simply reads from the buffer without consulting the thread's
+ * interrupt status.</p>
+ *
+ * @param in the buffered response stream.
+ * @return a non-interruptible channel over {@code in}.
+ */
+ private static ReadableByteChannel nonInterruptibleChannel(
+ final InputStream in) {
+ return new ReadableByteChannel() {
+ private volatile boolean open = true;
+ // Fixed staging buffer reused across reads for direct ByteBuffers. Arrow
+ // parsing drives this channel from a single thread, so a per-instance
+ // buffer is safe and avoids per-read allocation/GC churn.
+ private final byte[] staging = new byte[NON_INTERRUPTIBLE_READ_CHUNK];
+
+ @Override
+ public int read(final ByteBuffer dst) throws IOException {
+ final int toRead = dst.remaining();
+ if (toRead == 0) {
+ return 0;
+ }
+ if (dst.hasArray()) {
+ final int n = in.read(dst.array(),
+ dst.arrayOffset() + dst.position(), toRead);
+ if (n > 0) {
+ dst.position(dst.position() + n);
+ }
+ return n;
+ }
+ // For a direct (non-array-backed) ByteBuffer we must stage bytes in a
+ // heap buffer first. ArrowStreamReader can request very large reads,
so
+ // read at most one staging-buffer worth per call and let the reader
+ // issue further reads for the remainder (its readFully loops on
partial
+ // reads). This bounds allocation and reuses a single buffer.
+ final int chunk = Math.min(toRead, staging.length);
+ final int n = in.read(staging, 0, chunk);
+ if (n > 0) {
+ dst.put(staging, 0, n);
+ }
+ return n;
+ }
+
+ @Override
+ public boolean isOpen() {
+ return open;
+ }
+
+ @Override
+ public void close() throws IOException {
+ open = false;
+ in.close();
+ }
+ };
+ }
+
+ /**
+ * Convert a single Arrow row into a {@link BlobListResultEntrySchema}. Rows
+ * without a usable name are skipped.
+ */
+ private BlobListResultEntrySchema buildEntry(
+ final BatchColumns columns, final int row) {
+ String name = readString(columns.name, row);
+ if (name == null || name.isEmpty()) {
+ return null;
+ }
+ // Directory names may carry a trailing slash; strip it to match XML
parser.
+ if (name.endsWith(FORWARD_SLASH)) {
+ name = name.substring(0, name.length() - 1);
+ }
+
+ BlobListResultEntrySchema entry = new BlobListResultEntrySchema();
+ entry.setName(name);
+ entry.setPath(new Path(ROOT_PATH + name));
+ entry.setUrl(url + ROOT_PATH + name);
+
+ entry.setETag(readString(columns.etag, row));
+ // The Arrow (Photon) response serializes timestamps as native Arrow
+ // timestamp vectors (e.g. Creation-Time / Last-Modified as TimeStampSec),
+ // whose object form is an ISO-8601 local date-time (e.g.
2026-07-06T10:31:19)
+ // in UTC. Normalize to RFC 1123 GMT so both paths produce identical values
+ // and DateTimeUtils.parseLastModifiedTime yields the correct time.
+ entry.setLastModifiedTime(readTimestampAsRfc1123(columns.lastModified,
row));
+ entry.setCreationTime(readTimestampAsRfc1123(columns.creationTime, row));
+ entry.setOwner(readString(columns.owner, row));
+ entry.setGroup(readString(columns.group, row));
+ entry.setPermission(readString(columns.permission, row));
+ entry.setAcl(readString(columns.acl, row));
+
+ // Blob user metadata (including the hdi_isfolder directory marker) is
+ // carried in a single Arrow map column; mirror the XML parser which
+ // populates the metadata map from the <Metadata> element.
+ Map<String, String> metadata = readMetadata(columns.metadata, row);
+ if (!metadata.isEmpty()) {
+ entry.setMetadata(metadata);
+ }
+
+ setContentLength(entry, columns.contentLength, row);
+ setCopyProperties(entry, columns, row);
+ entry.setIsDirectory(resolveIsDirectory(columns, metadata, row));
+ return entry;
+ }
+
+ /**
+ * Populate the copy-related properties, mirroring the XML parser's handling
+ * of the {@code Properties} copy elements. Absent columns leave the
+ * corresponding field unset, exactly as the XML path does when the elements
+ * are missing.
+ */
+ private void setCopyProperties(final BlobListResultEntrySchema entry,
+ final BatchColumns columns, final int row) {
+ entry.setCopyId(readString(columns.copyId, row));
+ entry.setCopyStatus(readString(columns.copyStatus, row));
+ entry.setCopySourceUrl(readString(columns.copySource, row));
+ entry.setCopyProgress(readString(columns.copyProgress, row));
+ entry.setCopyStatusDescription(
+ readString(columns.copyStatusDescription, row));
+ // XML stores the completion time as epoch millis parsed from an RFC 1123
+ // string; normalize the Arrow timestamp to the same RFC 1123 form first so
+ // both paths yield identical epoch values.
+ String copyCompletion =
+ readTimestampAsRfc1123(columns.copyCompletionTime, row);
+ if (copyCompletion != null) {
+ entry.setCopyCompletionTime(
+ DateTimeUtils.parseLastModifiedTime(copyCompletion));
+ }
+ }
+
+ /**
+ * Populate the content length from the content-length column. The Blob
+ * endpoint surfaces it as a native unsigned 64-bit integer
+ * ({@code UInt8Vector}); other numeric vector types and a numeric string
+ * column are also tolerated so additive schema changes do not break parsing.
+ */
+ private void setContentLength(final BlobListResultEntrySchema entry,
+ final FieldVector contentLength, final int row) {
+ Long value = readLong(contentLength, row);
+ if (value != null) {
+ entry.setContentLength(value);
+ }
+ }
+
+ /**
+ * Determine whether the row represents a directory. Mirrors the XML parser
+ * ({@code BlobListXmlParser}), which flags an entry as a directory when any
of
+ * the following hold:
+ * <ul>
+ * <li>the entry is a {@code BlobPrefix} - surfaced in the Arrow response
as
+ * a {@code ResourceType} of {@code blobprefix} - i.e. an implicit
+ * directory;</li>
+ * <li>a {@code ResourceType} of {@code directory};</li>
+ * <li>an explicit boolean {@code IsDirectory} indicator;</li>
+ * <li>the {@code hdi_isfolder} user metadata being {@code true}.</li>
+ * </ul>
+ * The implicit-directory ({@code blobprefix}) and {@code hdi_isfolder} cases
+ * are essential: an implicit directory has no marker blob at all, while an
+ * empty directory created by {@code mkdir} is a zero-byte marker blob whose
+ * only directory indicator is the {@code hdi_isfolder=true} metadata entry.
+ * The metadata key is matched case-insensitively, matching the XML parser
+ * ({@code equalsIgnoreCase}). Without honoring these, such directories are
+ * misclassified as files, diverging from the XML listing path.
+ */
+ private boolean resolveIsDirectory(final BatchColumns columns,
+ final Map<String, String> metadata, final int row) {
+ String resourceType = readString(columns.resourceType, row);
+ if (DIRECTORY.equalsIgnoreCase(resourceType)
+ || ARROW_RESOURCE_TYPE_BLOB_PREFIX.equalsIgnoreCase(resourceType)) {
+ return true;
+ }
+ if (readBoolean(columns.isDirectory, row)) {
+ return true;
+ }
+ String hdiIsFolder = metadataValueIgnoreCase(metadata,
XML_TAG_HDI_ISFOLDER);
+ if (hdiIsFolder != null && Boolean.parseBoolean(hdiIsFolder.trim())) {
+ return true;
+ }
+ // Fallback for a schema that flattens the marker metadata into a dedicated
+ // column instead of the metadata map.
+ return readBoolean(columns.hdiIsFolder, row);
+ }
+
+ /**
+ * Case-insensitive lookup of a metadata value by key, mirroring the XML
+ * parser's {@code equalsIgnoreCase} matching of the {@code hdi_isfolder}
+ * marker (the service preserves the casing used when the metadata was set).
+ */
+ private static String metadataValueIgnoreCase(
+ final Map<String, String> metadata, final String key) {
+ String value = metadata.get(key);
+ if (value != null || metadata.containsKey(key)) {
+ return value;
+ }
+ for (Map.Entry<String, String> entry : metadata.entrySet()) {
+ if (key.equalsIgnoreCase(entry.getKey())) {
+ return entry.getValue();
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Read the user metadata key/value pairs from the Arrow map column for the
+ * given row. Keys are stored verbatim (e.g. {@code hdi_isfolder}) to match
the
+ * XML parser. Returns an empty (mutable) map when the column is absent or
the
+ * value is null.
+ */
+ private Map<String, String> readMetadata(final MapVector metadata,
+ final int row) {
+ Map<String, String> result = new HashMap<>();
+ if (metadata == null || row >= metadata.getValueCount()
+ || metadata.isNull(row)) {
+ return result;
+ }
+ List<?> entries = metadata.getObject(row);
+ if (entries == null) {
+ return result;
+ }
+ for (Object element : entries) {
+ if (element instanceof Map) {
+ Map<?, ?> keyValue = (Map<?, ?>) element;
+ Object key = keyValue.get(MapVector.KEY_NAME);
+ Object value = keyValue.get(MapVector.VALUE_NAME);
+ if (key != null) {
+ result.put(key.toString(), value == null ? null : value.toString());
+ }
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Read a timestamp column for the given row and convert it to the RFC 1123
+ * GMT representation used by the XML path. Native Arrow timestamp vectors
+ * expose an ISO-8601 local date-time via {@link FieldVector#getObject(int)};
+ * a textual (VarChar) timestamp column is also tolerated. Returns
+ * {@code null} when the column is absent or the value is null.
+ */
+ private String readTimestampAsRfc1123(final FieldVector vector,
+ final int row) {
+ if (vector == null || row >= vector.getValueCount()
+ || vector.isNull(row)) {
+ return null;
+ }
+ Object value = vector.getObject(row);
+ if (value == null) {
+ return null;
+ }
+ return DateTimeUtils.formatArrowDateTimeToRfc1123(value.toString());
+ }
+
+ /**
+ * Read a numeric column for the given row as a {@code long}. Supports native
+ * Arrow integer vectors (whose object form is a {@link Number}) and a
numeric
+ * {@link VarCharVector}. Returns {@code null} when the column is absent, the
+ * value is null, or a textual value is not parseable.
+ */
+ private Long readLong(final FieldVector vector, final int row) {
+ if (vector == null || row >= vector.getValueCount()
+ || vector.isNull(row)) {
+ return null;
+ }
+ Object value = vector.getObject(row);
+ if (value instanceof Number) {
+ return ((Number) value).longValue();
+ }
+ if (value != null) {
+ String text = value.toString().trim();
+ if (!text.isEmpty()) {
+ try {
+ return Long.parseLong(text);
+ } catch (NumberFormatException ignored) {
+ // Leave unset if the value is not numeric.
Review Comment:
Added LOG.debug on the NumberFormatException in readLong. This matches the
XML path, which also leaves an unparseable <Content-Length> unset (→0). Added
testNonNumericContentLengthIgnored.
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/services/ArrowListBlobParser.java:
##########
@@ -0,0 +1,590 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.contracts.services;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.BitVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.complex.MapVector;
+import org.apache.arrow.vector.ipc.ArrowStreamReader;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.utils.DateTimeUtils;
+
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_ACL;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CONTENT_LENGTH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_COMPLETION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_ID;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_PROGRESS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_SOURCE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_COPY_STATUS_DESCRIPTION;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_CREATION_TIME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_ETAG;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_GROUP;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_IS_DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_LAST_MODIFIED;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_METADATA;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_NAME;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_OWNER;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_PERMISSIONS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_COL_RESOURCE_TYPE;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_METADATA_NEXT_MARKER;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ARROW_RESOURCE_TYPE_BLOB_PREFIX;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.DIRECTORY;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.FORWARD_SLASH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ROOT_PATH;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.XML_TAG_HDI_ISFOLDER;
+import static
org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_META_HDI_ISFOLDER;
+import static
org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_ARROW_LIST_PARSING;
+
+/**
+ * Photon response parser implementation for Apache Arrow based ListBlobs
+ * responses. The parser reads the Arrow IPC stream returned by the service,
+ * iterates through the record batches and converts each row into a
+ * {@link BlobListResultEntrySchema}, producing a {@link BlobListResultSchema}
+ * that keeps the downstream FileStatus conversion path unchanged.
+ * <p>
+ * Arrow specific objects are confined to this class; the output contract is
+ * {@link BlobListResultSchema} so existing listing behavior is preserved.
+ * <p>
+ * Column matching is case-insensitive. Unknown columns are ignored and missing
+ * columns are treated as absent values so that additive schema changes do not
+ * break parsing. The continuation token is read from the Arrow schema custom
+ * metadata (key {@code NextMarker}).
+ */
+public class ArrowListBlobParser implements ListBlobResponseParser {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ArrowListBlobParser.class);
+
+ /**
+ * Upper bound, in bytes, on the heap staging buffer allocated per read when
+ * copying into a direct (non-array-backed) {@link ByteBuffer}. Caps
transient
+ * allocations for large reader requests; the reader simply issues additional
+ * reads for anything beyond this chunk.
+ */
+ private static final int NON_INTERRUPTIBLE_READ_CHUNK = 8192;
+
+ /**
+ * Base URL for which the ListBlobs API is called, used to build the absolute
+ * URL for each entry (mirrors the XML parser behavior).
+ */
+ private final String url;
+
+ /**
+ * Maximum off-heap (direct) memory in bytes the Arrow allocator may use
while
+ * parsing a single response. Bounds the memory an oversized or malformed
+ * response can consume; exceeding it fails the parse with an
+ * {@link org.apache.arrow.memory.OutOfMemoryException} that is surfaced as
an
+ * {@link IOException}.
+ */
+ private final long memoryLimitBytes;
+
+ /**
+ * @param url base URL for which the ListBlobs API is called.
+ * @param memoryLimitBytes maximum off-heap memory in bytes the Arrow
allocator
+ * may use while parsing a single response.
+ */
+ public ArrowListBlobParser(final String url, final long memoryLimitBytes) {
+ this.url = url;
+ this.memoryLimitBytes = memoryLimitBytes;
+ }
+
+ @Override
+ public BlobListResultSchema parse(final InputStream responseStream)
+ throws IOException {
+ BlobListResultSchema listResultSchema = new BlobListResultSchema();
+ try (BufferAllocator allocator = new RootAllocator(memoryLimitBytes);
+ ArrowStreamReader reader =
+ new ArrowStreamReader(nonInterruptibleChannel(responseStream),
+ allocator)) {
+ VectorSchemaRoot root = reader.getVectorSchemaRoot();
+ Schema schema = root.getSchema();
+
+ // Continuation token is carried in the Arrow schema custom metadata.
+ // The service emits an empty string when there is no continuation token,
+ // whereas the XML path leaves it null; normalize an absent or empty
+ // marker to null so both formats produce identical values.
+ Map<String, String> customMetadata = schema.getCustomMetadata();
+ if (customMetadata != null) {
+ String nextMarker = customMetadata.get(ARROW_METADATA_NEXT_MARKER);
+ listResultSchema.setNextMarker(
+ (nextMarker == null || nextMarker.isEmpty()) ? null : nextMarker);
+ }
+
+ while (reader.loadNextBatch()) {
+ BatchColumns columns = BatchColumns.resolve(root);
Review Comment:
Resolved once before the while loop (same VectorSchemaRoot/FieldVector
instances persist across batches). Added testTwoBatchesInOneStream to exercise
the multi-batch path.
> ABFS: Support Photon (Apache Arrow) based ListBlobs on Blob endpoint with XML
> fallback
> --------------------------------------------------------------------------------------
>
> Key: HADOOP-19941
> URL: https://issues.apache.org/jira/browse/HADOOP-19941
> Project: Hadoop Common
> Issue Type: New Feature
> Components: fs/azure
> Reporter: Manish Bhatt
> Assignee: Manish Bhatt
> Priority: Major
> Labels: pull-request-available
>
> Add config-gated support (fs.azure.photon.enabled, default off) for consuming
> ListBlobs responses in the Apache Arrow (Photon) format on the ABFS Blob
> endpoint. When enabled, ABFS advertises Arrow via an Accept header; the
> response Content-Type selects an Arrow or the existing XML parser, both
> producing identical FileStatus results, so downstream behaviour is unchanged.
> Includes automatic, transparent fallback to XML, full parsing parity
> (metadata, directory markers, implicit directories, copy properties, native
> timestamp/length vectors), interrupt-safe Arrow parsing, and Photon telemetry
> (request, response, fallback, parse-failure counts and listing latency). No
> public API changes. Covered by unit and integration tests.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
