[
https://issues.apache.org/jira/browse/HADOOP-19941?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18107734#comment-18107734
]
ASF GitHub Bot commented on HADOOP-19941:
-----------------------------------------
bhattmanish98 commented on code in PR #8611:
URL: https://github.com/apache/hadoop/pull/8611#discussion_r3850199497
##########
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/contract/TestArrowListBlobParser.java:
##########
@@ -0,0 +1,1193 @@
+/**
+ * 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.TimeStampMilliTZVector;
+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 {
+
+ private static final String URL =
"https://account.blob.core.windows.net/container";;
+
+ /** Generous allocator limit used by the functional parsing tests. */
+ private static final long MEMORY_LIMIT = 256L * 1024 * 1024;
+
+ /** Deliberately tiny allocator limit used to force an over-limit failure. */
+ private static final long TINY_MEMORY_LIMIT = 1024L;
+
+ /** Arrow integer bit width used for the content-length column. */
+ private static final int ARROW_INT_BIT_WIDTH = 64;
+
+ /** Row count large enough to exhaust the tiny allocator limit. */
+ private static final int OVER_LIMIT_ROW_COUNT = 2000;
+
+ // Sample content-length values shared between builder rows and assertions.
+ private static final long CONTENT_LENGTH_A = 20L;
+ private static final long CONTENT_LENGTH_B = 30L;
+ private static final long CONTENT_LENGTH_C = 40L;
+ private static final long CONTENT_LENGTH_D = 42L;
+ private static final long CONTENT_LENGTH_SAMPLE = 1234L;
+
+ /**
+ * Verify an Arrow response with a single blob is parsed correctly and all
+ * common blob properties are populated.
+ */
+ @Test
+ public void testSingleBlob() throws Exception {
+ byte[] stream = new ArrowStreamBuilder()
+ .addRow("file1.txt", "0x8DB6668EAB50E67", CONTENT_LENGTH_SAMPLE,
+ "Tue, 06 Jun 2023 08:35:00 GMT",
+ "Tue, 06 Jun 2023 08:34:28 GMT", false)
+ .build();
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.paths()).hasSize(1);
+ BlobListResultEntrySchema entry = result.paths().get(0);
+ assertThat(entry.name()).isEqualTo("file1.txt");
+ assertThat(entry.path().toUri().getPath()).isEqualTo("/file1.txt");
+ assertThat(entry.url()).isEqualTo(URL + "/file1.txt");
+ assertThat(entry.eTag()).isEqualTo("0x8DB6668EAB50E67");
+ assertThat(entry.contentLength()).isEqualTo(CONTENT_LENGTH_SAMPLE);
+ assertThat(entry.lastModified()).isEqualTo("Tue, 06 Jun 2023 08:35:00
GMT");
+ assertThat(entry.creation()).isEqualTo("Tue, 06 Jun 2023 08:34:28 GMT");
+ assertThat(entry.isDirectory()).isFalse();
+ }
+
+ /**
+ * Verify that an ISO-8601 date-time from the Arrow response (as Photon
+ * serializes timestamps, without an explicit zone) is normalized to the same
+ * RFC 1123 GMT representation used by the XML ListBlobs path, and that it
+ * yields the expected epoch when parsed downstream.
+ */
+ @Test
+ public void testArrowIsoDateTimeNormalizedToRfc1123() throws Exception {
+ byte[] stream = new ArrowStreamBuilder()
+ .addRow("file1.txt", "etag", 1L,
+ "2026-07-06T10:31:19", "2026-07-06T10:30:00", false)
+ .build();
+
+ BlobListResultSchema result = parse(stream);
+
+ BlobListResultEntrySchema entry = result.paths().get(0);
+ assertThat(entry.lastModified())
+ .isEqualTo("Mon, 06 Jul 2026 10:31:19 GMT");
+ assertThat(entry.creation())
+ .isEqualTo("Mon, 06 Jul 2026 10:30:00 GMT");
+
+ long arrowEpoch = DateTimeUtils.parseLastModifiedTime(
+ entry.lastModified());
+ long xmlEpoch = DateTimeUtils.parseLastModifiedTime(
+ "Mon, 06 Jul 2026 10:31:19 GMT");
+ assertThat(arrowEpoch).isEqualTo(xmlEpoch);
+ }
+
+ /**
+ * A directory entry (or any entry) may carry no timestamp. Verify that a
+ * missing timestamp is exposed as {@code null} - not the string {@code
"null"}
+ * - through the {@code lastModified()}/{@code creation()} accessors used by
+ * the FileStatus conversion. Previously these accessors wrapped the field in
+ * {@code String.valueOf}, turning an absent value into {@code "null"} that
+ * then failed {@link DateTimeUtils#parseLastModifiedTime(String)} and
logged a
+ * spurious error.
+ */
+ @Test
+ public void testMissingTimestampExposedAsNull() throws Exception {
+ byte[] stream = new ArrowStreamBuilder()
+ .addRow("dir1", "etag", 0L, null, null, true)
+ .build();
+
+ BlobListResultSchema result = parse(stream);
+
+ BlobListResultEntrySchema entry = result.paths().get(0);
+ assertThat(entry.name()).isEqualTo("dir1");
+ assertThat(entry.lastModified()).isNull();
+ assertThat(entry.creation()).isNull();
+ }
+
+ /**
+ * Verify an Arrow response with multiple blobs is parsed correctly.
+ */
+ @Test
+ public void testMultipleBlobs() throws Exception {
+ byte[] stream = new ArrowStreamBuilder()
+ .addRow("a.txt", "etagA", 10L, "lmA", "ctA", false)
+ .addRow("b.txt", "etagB", CONTENT_LENGTH_A, "lmB", "ctB", false)
+ .addRow("dir1", "etagC", 0L, "lmC", "ctC", true)
+ .build();
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.paths()).hasSize(3);
+ assertThat(result.paths().get(0).name()).isEqualTo("a.txt");
+
assertThat(result.paths().get(1).contentLength()).isEqualTo(CONTENT_LENGTH_A);
+ assertThat(result.paths().get(2).isDirectory()).isTrue();
+ }
+
+ /**
+ * Verify an empty Arrow response returns an empty listing.
+ */
+ @Test
+ public void testEmptyResponse() throws Exception {
+ byte[] stream = new ArrowStreamBuilder().build();
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.paths()).isEmpty();
+ assertThat(result.getNextMarker()).isNull();
+ }
+
+ /**
+ * Verify directory entries are populated correctly from an Arrow response.
+ */
+ @Test
+ public void testDirectoryEntry() throws Exception {
+ byte[] stream = new ArrowStreamBuilder()
+ .addRow("mydir", "etag", 0L, "lm", "ct", true)
+ .build();
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.paths()).hasSize(1);
+ assertThat(result.paths().get(0).isDirectory()).isTrue();
+ assertThat(result.paths().get(0).name()).isEqualTo("mydir");
+ }
+
+ /**
+ * Verify a directory name with a trailing slash is normalized (slash
removed)
+ * to match the XML parser behavior.
+ */
+ @Test
+ public void testDirectoryNameTrailingSlashStripped() throws Exception {
+ byte[] stream = new ArrowStreamBuilder()
+ .addRow("mydir/", "etag", 0L, "lm", "ct", true)
+ .build();
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.paths().get(0).name()).isEqualTo("mydir");
+
assertThat(result.paths().get(0).path().toUri().getPath()).isEqualTo("/mydir");
+ }
+
+ /**
+ * Verify an empty-directory marker blob is classified as a directory when
its
+ * only directory indicator is the {@code hdi_isfolder=true} user metadata
+ * (surfaced as a column), even though no {@code IsDirectory}/{@code
+ * ResourceType} column marks it as such. This mirrors the XML parser and is
+ * the scenario that otherwise causes recursive {@code listFiles} on an empty
+ * directory to wrongly report the marker as a file.
+ */
+ @Test
+ public void testMarkerDirectoryViaHdiIsFolderMetadata() throws Exception {
+ Map<String, String> row = new LinkedHashMap<>();
+ row.put(ARROW_COL_NAME, "emptydir");
+ row.put(ARROW_COL_ETAG, "etag");
+ row.put(ARROW_COL_CONTENT_LENGTH, "0");
+ row.put(XML_TAG_HDI_ISFOLDER, "true");
+ byte[] stream = buildStringColumnStream(new ArrayList<Map<String,
String>>() {{
+ add(row);
+ }}, null);
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.paths()).hasSize(1);
+ assertThat(result.paths().get(0).name()).isEqualTo("emptydir");
+ assertThat(result.paths().get(0).isDirectory())
+ .as("marker blob with hdi_isfolder=true must be a directory")
+ .isTrue();
+ }
+
+ /**
+ * Verify a {@code hdi_isfolder} value of {@code false} does not flip a plain
+ * blob into a directory.
+ */
+ @Test
+ public void testHdiIsFolderFalseIsNotDirectory() throws Exception {
+ Map<String, String> row = new LinkedHashMap<>();
+ row.put(ARROW_COL_NAME, "file.txt");
+ row.put(XML_TAG_HDI_ISFOLDER, "false");
+ byte[] stream = buildStringColumnStream(new ArrayList<Map<String,
String>>() {{
+ add(row);
+ }}, null);
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.paths().get(0).isDirectory()).isFalse();
+ }
+
+ /**
+ * Verify the marker is also recognized when the metadata surfaces under the
+ * HTTP header form column name {@code x-ms-meta-hdi_isfolder}.
+ */
+ @Test
+ public void testMarkerDirectoryViaHttpHeaderMetadataColumn() throws
Exception {
+ Map<String, String> row = new LinkedHashMap<>();
+ row.put(ARROW_COL_NAME, "emptydir");
+ row.put(X_MS_META_HDI_ISFOLDER, "true");
+ byte[] stream = buildStringColumnStream(new ArrayList<Map<String,
String>>() {{
+ add(row);
+ }}, null);
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.paths().get(0).isDirectory()).isTrue();
+ }
+
+ /**
+ * Verify an entry with a {@code ResourceType} of {@code directory} is
+ * classified as a directory, matching the XML parser's Properties handling.
+ */
+ @Test
+ public void testResourceTypeDirectory() throws Exception {
+ Map<String, String> row = new LinkedHashMap<>();
+ row.put(ARROW_COL_NAME, "dir1");
+ row.put(ARROW_COL_RESOURCE_TYPE, DIRECTORY);
+ byte[] stream = buildStringColumnStream(new ArrayList<Map<String,
String>>() {{
+ add(row);
+ }}, null);
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.paths().get(0).isDirectory()).isTrue();
+ }
+
+ /**
+ * Verify parsing against the real Blob-endpoint Arrow schema, where the
+ * empty-directory marker's {@code hdi_isfolder} flag is carried inside a
+ * {@code Metadata} map column, {@code Content-Length} is an unsigned 64-bit
+ * integer ({@code UInt8}) and the timestamps are native Arrow
+ * {@code TimeStampSec} vectors. This is the exact shape returned by the
+ * service (see {@code ITestAbfsFileSystemContractGetFileStatus
+ * #testListFilesEmptyDirectoryRecursive}); the earlier flattened-column
+ * assumption never matched a live response.
+ */
+ @Test
+ public void testMarkerDirectoryViaMetadataMapColumn() throws Exception {
+ Map<String, String> markerMetadata = new LinkedHashMap<>();
+ markerMetadata.put(XML_TAG_HDI_ISFOLDER, "true");
+ byte[] stream = new BlobEndpointStreamBuilder()
+ .addRow("emptydir", "0x8DEE", 0L, "2026-07-13T07:06:45",
+ "2026-07-13T07:06:45", "blob", markerMetadata)
+ .addRow("file.txt", "0x8DEF", CONTENT_LENGTH_SAMPLE,
"2026-07-13T07:06:46",
+ "2026-07-13T07:06:46", "blob", Collections.emptyMap())
+ .build();
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.paths()).hasSize(2);
+ BlobListResultEntrySchema marker = result.paths().get(0);
+ assertThat(marker.name()).isEqualTo("emptydir");
+ assertThat(marker.isDirectory())
+ .as("marker blob with Metadata hdi_isfolder=true must be a directory")
+ .isTrue();
+ assertThat(marker.contentLength()).isEqualTo(0L);
+ assertThat(marker.eTag()).isEqualTo("0x8DEE");
+ assertThat(marker.metadata()).containsEntry(XML_TAG_HDI_ISFOLDER, "true");
+ assertThat(marker.lastModified())
+ .as("TimeStampSec value must be normalized to RFC 1123 GMT")
+ .isEqualTo(DateTimeUtils.formatArrowDateTimeToRfc1123(
+ "2026-07-13T07:06:45"));
+
+ BlobListResultEntrySchema file = result.paths().get(1);
+ assertThat(file.name()).isEqualTo("file.txt");
+ assertThat(file.isDirectory())
+ .as("plain blob without hdi_isfolder must be a file")
+ .isFalse();
+ assertThat(file.contentLength())
+ .as("UInt8 content length must be read as a long")
+ .isEqualTo(CONTENT_LENGTH_SAMPLE);
+ }
+
+ /**
+ * Verify an implicit directory - surfaced by the Blob endpoint as a
+ * {@code BlobPrefix} row whose {@code ResourceType} is {@code blobprefix}
and
+ * whose {@code Name} carries a trailing slash - is classified as a
directory,
+ * matching the XML parser which flags every {@code <BlobPrefix>} entry as a
+ * directory. Without this, implicit directories are misclassified as files.
+ */
+ @Test
+ public void testImplicitDirectoryViaBlobPrefixResourceType()
+ throws Exception {
+ Map<String, String> row = new LinkedHashMap<>();
+ row.put(ARROW_COL_NAME, "implicitDir/azcopy/");
+ row.put(ARROW_COL_RESOURCE_TYPE, ARROW_RESOURCE_TYPE_BLOB_PREFIX);
+ byte[] stream = buildStringColumnStream(new ArrayList<Map<String,
String>>() {{
+ add(row);
+ }}, null);
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.paths()).hasSize(1);
+ assertThat(result.paths().get(0).isDirectory())
+ .as("blobprefix (implicit directory) must be a directory")
+ .isTrue();
+ }
+
+ /**
+ * Verify the {@code hdi_isfolder} marker is matched case-insensitively, as
the
+ * Blob service preserves whatever casing the metadata key was set with (e.g.
+ * {@code HDI_ISFOLDER}). The XML parser compares with {@code
equalsIgnoreCase},
+ * so Arrow must too, otherwise directories created with a differently-cased
+ * key are misclassified as files (see
+ * {@code
ITestAzureBlobFileSystemListStatus#testIsDirectoryWithDifferentCases}).
+ */
+ @Test
+ public void testMarkerDirectoryMetadataKeyCaseInsensitive() throws Exception
{
+ Map<String, String> markerMetadata = new LinkedHashMap<>();
+ markerMetadata.put("HDI_ISFOLDER", "true");
+ byte[] stream = new BlobEndpointStreamBuilder()
+ .addRow("emptydir", "0x8DEE", 0L, "2026-07-13T07:06:45",
+ "2026-07-13T07:06:45", "blob", markerMetadata)
+ .build();
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.paths()).hasSize(1);
+ assertThat(result.paths().get(0).isDirectory())
+ .as("hdi_isfolder marker must be matched case-insensitively")
+ .isTrue();
+ }
+
+ /**
+ * Verify the copy-status Properties are propagated from the Arrow columns to
+ * the result entry, matching the XML parser which surfaces {@code CopyId},
+ * {@code CopyStatus}, {@code CopySource}, {@code CopyProgress},
+ * {@code CopyCompletionTime} and {@code CopyStatusDescription}.
+ */
+ @Test
+ public void testCopyPropertiesPopulated() throws Exception {
+ Map<String, String> row = new LinkedHashMap<>();
+ row.put(ARROW_COL_NAME, "copied.txt");
+ row.put(ARROW_COL_COPY_ID, "copy-id-1");
+ row.put(ARROW_COL_COPY_STATUS, "success");
+ row.put(ARROW_COL_COPY_SOURCE, "https://src/blob";);
+ row.put(ARROW_COL_COPY_PROGRESS, "1234/1234");
+ row.put(ARROW_COL_COPY_STATUS_DESCRIPTION, "done");
+ row.put(ARROW_COL_COPY_COMPLETION_TIME, "Mon, 13 Jul 2026 07:06:45 GMT");
+ byte[] stream = buildStringColumnStream(new ArrayList<Map<String,
String>>() {{
+ add(row);
+ }}, null);
+
+ BlobListResultSchema result = parse(stream);
+
+ BlobListResultEntrySchema entry = result.paths().get(0);
+ assertThat(entry.copyId()).isEqualTo("copy-id-1");
+ assertThat(entry.copyStatus()).isEqualTo("success");
+ assertThat(entry.copySourceUrl()).isEqualTo("https://src/blob";);
+ assertThat(entry.copyProgress()).isEqualTo("1234/1234");
+ assertThat(entry.copyStatusDescription()).isEqualTo("done");
+ assertThat(entry.copyCompletionTime())
+ .as("CopyCompletionTime must be parsed to epoch millis")
+ .isEqualTo(DateTimeUtils.parseLastModifiedTime(
+ "Mon, 13 Jul 2026 07:06:45 GMT"));
+ }
+
+ /**
+ * Verify the continuation token is extracted from the Arrow schema custom
+ * metadata.
+ */
+ @Test
+ public void testContinuationToken() throws Exception {
+ byte[] stream = new ArrowStreamBuilder()
+ .withNextMarker("marker-123")
+ .addRow("a.txt", "etag", 10L, "lm", "ct", false)
+ .build();
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.getNextMarker()).isEqualTo("marker-123");
+ assertThat(result.paths()).hasSize(1);
+ }
+
+ /**
+ * Verify that an empty continuation token in the Arrow schema custom
metadata
+ * (as emitted by the service on the terminal page) is normalized to
+ * {@code null}, matching the XML parser which leaves an empty NextMarker
null.
+ */
+ @Test
+ public void testEmptyContinuationTokenNormalizedToNull() throws Exception {
+ byte[] stream = new ArrowStreamBuilder()
+ .withNextMarker("")
+ .addRow("a.txt", "etag", 10L, "lm", "ct", false)
+ .build();
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.getNextMarker()).isNull();
+ assertThat(result.paths()).hasSize(1);
+ }
+
+ /**
+ * Verify a multi-page listing is handled correctly at the parser level: a
+ * non-terminal page carries a next marker signalling continuation while the
+ * terminal page carries none, and the rows from every page are parsed. This
+ * exercises the per-page building blocks the {@code listPath} pagination
loop
+ * relies on to stitch multiple Photon responses into one listing.
+ */
+ @Test
+ public void testMultiPageListingHandledCorrectly() throws Exception {
+ byte[] page1 = new ArrowStreamBuilder()
+ .withNextMarker("page-2-marker")
+ .addRow("a.txt", "etagA", 10L, "lm", "ct", false)
+ .addRow("b.txt", "etagB", CONTENT_LENGTH_A, "lm", "ct", false)
+ .build();
+ byte[] page2 = new ArrowStreamBuilder()
+ .addRow("c.txt", "etagC", CONTENT_LENGTH_B, "lm", "ct", false)
+ .addRow("d.txt", "etagD", CONTENT_LENGTH_C, "lm", "ct", false)
+ .build();
+
+ BlobListResultSchema first = parse(page1);
+ BlobListResultSchema second = parse(page2);
+
+ // Non-terminal page advertises the marker used to fetch the next page.
+ assertThat(first.getNextMarker()).isEqualTo("page-2-marker");
+ assertThat(first.paths()).hasSize(2);
+ // Terminal page carries no marker, ending the pagination loop.
+ assertThat(second.getNextMarker()).isNull();
+ assertThat(second.paths()).hasSize(2);
+
+ List<String> allNames = new ArrayList<>();
+ first.paths().forEach(p -> allNames.add(p.name()));
+ second.paths().forEach(p -> allNames.add(p.name()));
+ assertThat(allNames)
+ .containsExactly("a.txt", "b.txt", "c.txt", "d.txt");
+ }
+
+ /**
+ * Verify special-character blob names (spaces, reserved URL characters,
+ * percent and plus signs, nested path segments) are parsed without loss or
+ * corruption, complementing {@link #testUnicodeBlobName()}. The parser must
+ * expose the blob name exactly as returned in the Arrow payload.
+ */
+ @Test
+ public void testSpecialCharacterBlobNames() throws Exception {
+ String[] names = {
+ "with space.txt",
+ "with+plus.txt",
+ "with%percent.txt",
+ "a&b=c;d,e.txt",
+ "nested/dir/child.txt",
+ "trailing.dots...",
+ "emoji-\uD83D\uDE00.txt",
+ };
+ ArrowStreamBuilder builder = new ArrowStreamBuilder();
+ for (String name : names) {
+ builder.addRow(name, "etag", 1L, "lm", "ct", false);
+ }
+
+ BlobListResultSchema result = parse(builder.build());
+
+ assertThat(result.paths()).hasSize(names.length);
+ for (int i = 0; i < names.length; i++) {
+ assertThat(result.paths().get(i).name())
+ .as("name preserved verbatim for %s", names[i])
+ .isEqualTo(names[i]);
+ }
+ // For names free of URI-reserved delimiters, the derived Path must round
+ // trip to the same relative path used by the XML listing route.
+ assertThat(result.paths().get(0).path().toUri().getPath())
+ .isEqualTo("/with space.txt");
+ assertThat(result.paths().get(4).path().toUri().getPath())
+ .isEqualTo("/nested/dir/child.txt");
+ }
+
+ /**
+ * Verify special characters and Unicode blob names are parsed correctly.
+ */
+ @Test
+ public void testUnicodeBlobName() throws Exception {
+ String name = "文件-Ünïcode &+ space.txt";
+ byte[] stream = new ArrowStreamBuilder()
+ .addRow(name, "etag", 5L, "lm", "ct", false)
+ .build();
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.paths().get(0).name()).isEqualTo(name);
+ }
+
+ /**
+ * Verify additional unknown columns in the Arrow response are ignored safely
+ * and the known columns are still parsed.
+ */
+ @Test
+ public void testUnknownColumnsIgnored() throws Exception {
+ Map<String, String> row = new LinkedHashMap<>();
+ row.put(ARROW_COL_NAME, "a.txt");
+ row.put(ARROW_COL_ETAG, "etag");
+ row.put(ARROW_COL_CONTENT_LENGTH, "42");
+ row.put("SomeUnknownColumn", "ignore-me");
+ row.put("AnotherFutureColumn", "also-ignored");
+ byte[] stream = buildStringColumnStream(new ArrayList<Map<String,
String>>() {{
+ add(row);
+ }}, null);
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.paths()).hasSize(1);
+ assertThat(result.paths().get(0).name()).isEqualTo("a.txt");
+
assertThat(result.paths().get(0).contentLength()).isEqualTo(CONTENT_LENGTH_D);
+ }
+
+ /**
+ * Verify that when the Name column is entirely absent from the Arrow schema
+ * the parser fails loudly rather than silently returning an empty listing.
+ * Name is the one column {@code buildEntry()} cannot proceed without, so a
+ * schema lacking it would otherwise be indistinguishable from a genuinely
+ * empty directory and surface later as missing data in Spark/Hive.
+ */
+ @Test
+ public void testMissingMandatoryNameColumn() throws Exception {
+ Map<String, String> row = new LinkedHashMap<>();
+ row.put(ARROW_COL_ETAG, "etag");
+ row.put(ARROW_COL_CONTENT_LENGTH, "42");
+ byte[] stream = buildStringColumnStream(new ArrayList<Map<String,
String>>() {{
+ add(row);
+ }}, null);
+
+ assertThatThrownBy(() -> parse(stream))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining(ARROW_COL_NAME);
+ }
+
+ /**
+ * Verify a row whose Name column is present but null is skipped (rather than
+ * failing the whole listing), matching the XML parser which ignores an entry
+ * without a name.
+ */
+ @Test
+ public void testNullNameValueRowSkipped() throws Exception {
+ Map<String, String> withName = new LinkedHashMap<>();
+ withName.put(ARROW_COL_NAME, "a.txt");
+ withName.put(ARROW_COL_CONTENT_LENGTH, "42");
+ Map<String, String> withoutName = new LinkedHashMap<>();
+ withoutName.put(ARROW_COL_NAME, null);
+ withoutName.put(ARROW_COL_CONTENT_LENGTH, "10");
+ byte[] stream = buildStringColumnStream(
+ new ArrayList<Map<String, String>>() {{
+ add(withName);
+ add(withoutName);
+ }}, null);
+
+ BlobListResultSchema result = parse(stream);
+
+ assertThat(result.paths()).hasSize(1);
+ assertThat(result.paths().get(0).name()).isEqualTo("a.txt");
+ }
+
+ /**
+ * Verify a corrupted Arrow stream results in an appropriate failure
+ * (IOException) rather than a silent empty result.
+ */
+ @Test
+ public void testCorruptedStreamFails() {
+ byte[] garbage = "this is definitely not an arrow stream".getBytes(
+ StandardCharsets.UTF_8);
+ assertThatThrownBy(() -> parse(garbage)).isInstanceOf(IOException.class);
+ }
+
+ /**
+ * Verify a truncated Arrow stream results in an appropriate failure.
+ */
+ @Test
+ public void testTruncatedStreamFails() throws Exception {
+ byte[] stream = new ArrowStreamBuilder()
+ .addRow("a.txt", "etag", 10L, "lm", "ct", false)
+ .build();
+ byte[] truncated = new byte[stream.length / 2];
Review Comment:
Reworked to truncate at the schema/record-batch boundary (build a
schema-only stream, truncate the full 1-row stream to that length) so the
schema is kept and the record batch is dropped, then assert it throws
IOException. This reliably catches the silent "zero batches → empty listing"
data-loss case.
> 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]
