anmolanmol1234 commented on code in PR #8611: URL: https://github.com/apache/hadoop/pull/8611#discussion_r3842748865
########## 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 { Review Comment: CopyCompletionTime is fed as a VarChar RFC 1123 string, but the real Blob endpoint sends a native Arrow timestamp (as it does for Last-Modified). Please add a BlobEndpointStreamBuilder variant so readTimestampAsRfc1123() → parseLastModifiedTime() is verified on the shape we'll actually receive. -- 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]
