[
https://issues.apache.org/jira/browse/HADOOP-19941?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18099708#comment-18099708
]
ASF GitHub Bot commented on HADOOP-19941:
-----------------------------------------
anmolanmol1234 commented on code in PR #8611:
URL: https://github.com/apache/hadoop/pull/8611#discussion_r3665116934
##########
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:
varChar() returns null when a column is missing or
present-but-not-a-VarCharVector, so if that happens for Name, buildEntry()
returns null for every row and parse() completes normally with zero entries —
no exception, and PHOTON_PARSE_FAILURE_COUNT never fires. Callers can't
distinguish that from a genuinely empty directory, so the failure surfaces
later as missing data in Spark/Hive rather than as a listing error. Could we
throw here when columns.name == null? Name is the one column buildEntry() can't
proceed without, so the "missing columns are tolerated" rule shouldn't extend
to it.
> ABFS: Support Photon (Apache Arrow) based ListBlobs on Blob endpoint with XML
> fallback
> --------------------------------------------------------------------------------------
>
> Key: HADOOP-19941
> URL: https://issues.apache.org/jira/browse/HADOOP-19941
> Project: Hadoop Common
> Issue Type: New Feature
> Components: fs/azure
> Reporter: Manish Bhatt
> Assignee: Manish Bhatt
> Priority: Major
> Labels: pull-request-available
>
> Add config-gated support (fs.azure.photon.enabled, default off) for consuming
> ListBlobs responses in the Apache Arrow (Photon) format on the ABFS Blob
> endpoint. When enabled, ABFS advertises Arrow via an Accept header; the
> response Content-Type selects an Arrow or the existing XML parser, both
> producing identical FileStatus results, so downstream behaviour is unchanged.
> Includes automatic, transparent fallback to XML, full parsing parity
> (metadata, directory markers, implicit directories, copy properties, native
> timestamp/length vectors), interrupt-safe Arrow parsing, and Photon telemetry
> (request, response, fallback, parse-failure counts and listing latency). No
> public API changes. Covered by unit and integration tests.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
