anmolanmol1234 commented on code in PR #8611: URL: https://github.com/apache/hadoop/pull/8611#discussion_r3842760260
########## 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]; + System.arraycopy(stream, 0, truncated, 0, truncated.length); + + assertThatThrownBy(() -> parse(truncated)).isInstanceOf(IOException.class); + } + + /** + * Verify that an Arrow response whose parsing would exceed the configured + * allocator memory limit fails with an {@link IOException} rather than + * exhausting off-heap memory. + */ + @Test + public void testOverAllocatorLimitFails() throws Exception { + ArrowStreamBuilder builder = new ArrowStreamBuilder(); + for (int i = 0; i < OVER_LIMIT_ROW_COUNT; i++) { + builder.addRow("some-reasonably-long-blob-name-" + i, "etag-" + i, + 1024L, "last-modified", "creation-time", false); + } + byte[] stream = builder.build(); + + assertThatThrownBy(() -> { + try (InputStream in = new ByteArrayInputStream(stream)) { + new ArrowListBlobParser(URL, TINY_MEMORY_LIMIT).parse(in); + } + }).isInstanceOf(IOException.class); + } + + /** + * Verify the Arrow parser is immune to a set thread-interrupt flag. Because + * the parser must not adapt the buffered body with an interruptible NIO + * channel, parsing a valid stream on a thread whose interrupt status is set + * must still succeed (rather than fail with a + * {@link java.nio.channels.ClosedByInterruptException}), matching the XML SAX + * parser which reads a plain stream and ignores interrupts. The interrupt + * status is preserved and restored so callers can still observe it. + */ + @Test + public void testParseSurvivesInterruptedThread() throws Exception { + Map<String, String> row = new LinkedHashMap<>(); + row.put(ARROW_COL_NAME, "file.txt"); + row.put(ARROW_COL_CONTENT_LENGTH, "10"); + byte[] stream = buildStringColumnStream(new ArrayList<Map<String, String>>() {{ + add(row); + }}, null); + + Thread.currentThread().interrupt(); + try { + BlobListResultSchema result = parse(stream); + + assertThat(result.paths()).hasSize(1); + assertThat(result.paths().get(0).name()).isEqualTo("file.txt"); + assertThat(Thread.currentThread().isInterrupted()) + .as("parser must not clear the caller's interrupt status") + .isTrue(); + } finally { + // Clear the interrupt flag so it cannot leak into other tests. + Thread.interrupted(); + } + } + + /** + * A timezone-aware Arrow timestamp vector exposes an epoch number (not a + * date-time object) via {@code getObject()}. Verify such a column is still + * normalized to the RFC 1123 GMT string the XML path produces, rather than + * passing the raw epoch through as a bogus modification time. The service + * could switch to a TZ column without that being a breaking change. + */ + @Test + public void testTimeZoneTimestampNormalizedToRfc1123() throws Exception { + long epochMillis = LocalDateTime.parse("2026-07-06T10:31:19") + .toEpochSecond(ZoneOffset.UTC) * 1000L; + byte[] stream = buildTimeZoneTimestampStream("file.txt", epochMillis); + + BlobListResultSchema result = parse(stream); + + BlobListResultEntrySchema entry = result.paths().get(0); + assertThat(entry.lastModified()).isEqualTo("Mon, 06 Jul 2026 10:31:19 GMT"); + assertThat(DateTimeUtils.parseLastModifiedTime(entry.lastModified())) + .isEqualTo(DateTimeUtils.parseLastModifiedTime( + "Mon, 06 Jul 2026 10:31:19 GMT")); + } + + /** + * Verify a single Arrow stream carrying two record batches is fully drained: + * the {@code while (loadNextBatch())} loop must iterate for every batch and + * the column references resolved once before the loop must stay valid across + * batches. + */ + @Test + public void testTwoBatchesInOneStream() throws Exception { + byte[] stream = buildTwoBatchNameStream(); + + BlobListResultSchema result = parse(stream); + + assertThat(result.paths()).hasSize(3); + assertThat(result.paths().get(0).name()).isEqualTo("batch1-a"); + assertThat(result.paths().get(1).name()).isEqualTo("batch1-b"); + assertThat(result.paths().get(2).name()).isEqualTo("batch2-a"); + } + + /** + * An unsigned 64-bit ({@code UInt8}) content length above {@code Long.MAX_VALUE} + * comes back negative from {@code getObject()}. Verify such a malformed length + * is rejected so a FileStatus can never surface with a negative length; the + * entry falls back to the default zero length instead. + */ + @Test + public void testNegativeContentLengthRejected() throws Exception { + byte[] stream = new BlobEndpointStreamBuilder() + .addRow("blob.txt", "etag", -1L, "2026-07-06T10:31:19", + "2026-07-06T10:30:00", "blob", new LinkedHashMap<String, String>()) + .build(); + + BlobListResultSchema result = parse(stream); + + assertThat(result.paths().get(0).contentLength()).isEqualTo(0L); + } + + /** + * Verify a non-numeric textual content length is ignored (leaving the default + * zero length) rather than failing parsing, matching the XML path's handling + * of an unparseable {@code Content-Length}. + */ + @Test + public void testNonNumericContentLengthIgnored() throws Exception { + Map<String, String> row = new LinkedHashMap<>(); + row.put(ARROW_COL_NAME, "blob.txt"); + row.put(ARROW_COL_CONTENT_LENGTH, "not-a-number"); + 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).contentLength()).isEqualTo(0L); + } + + /** + * Verify a directory marker carried as a flattened {@code hdi_isfolder} + * column with the numeric-true value {@code "1"} is recognized as a directory. + */ + @Test + public void testHdiIsFolderNumericTrueColumn() throws Exception { + Map<String, String> row = new LinkedHashMap<>(); + row.put(ARROW_COL_NAME, "dir1"); + row.put(XML_TAG_HDI_ISFOLDER, "1"); + 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()).isTrue(); + } + + /** + * Verify a blob name consisting solely of {@code "/"} is handled without + * error. The trailing slash is stripped (as for any directory name), leaving + * an entry rooted at the container root. + */ + @Test + public void testNameThatIsOnlySlash() throws Exception { + Map<String, String> row = new LinkedHashMap<>(); + row.put(ARROW_COL_NAME, "/"); + 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()).isEmpty(); + } + + private BlobListResultSchema parse(byte[] stream) throws IOException { + try (InputStream in = new ByteArrayInputStream(stream)) { + return new ArrowListBlobParser(URL, MEMORY_LIMIT).parse(in); + } + } Review Comment: parse() normalizes a null/empty marker to null, but a whitespace-only " " marker passes straight through and would be sent back as a continuation token. Please add a test for the whitespace case. -- 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]
