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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
