Storage - STG 104 Photon

eng/versioning/external_dependencies.txt

@@ -25,6 +25,7 @@ com.github.spotbugs:spotbugs;4.8.3
 com.github.spotbugs:spotbugs-annotations;4.8.3
 com.github.spotbugs:spotbugs-maven-plugin;4.8.3.1
 com.google.code.gson:gson;2.14.0
+com.google.flatbuffers:flatbuffers-java;24.3.25
 com.google.guava:guava;33.6.0-jre
 com.h2database:h2;2.2.220
 com.h3xstream.findsecbugs:findsecbugs-plugin;1.9.0

sdk/parents/azure-client-sdk-parent/pom.xml

@@ -602,6 +602,10 @@
 
                 <!-- Used by many libraries to bring in annotations used by Reactor -->
                 <include>com.google.code.findbugs:jsr305:[3.0.2]</include> <!-- {x-include-update;com.google.code.findbugs:jsr305;external_dependency} -->
+
+                <!-- Apache Arrow format parsing. flatbuffers-java is the sole (compile-scoped) dependency used by
+                     azure-storage-blob's vendored Arrow FlatBuffer accessors -->
+                <include>com.google.flatbuffers:flatbuffers-java:[24.3.25]</include> <!-- {x-include-update;com.google.flatbuffers:flatbuffers-java;external_dependency} -->
               </includes>
             </bannedDependencies>
 

sdk/storage/azure-storage-blob/assets.json

@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "java",
   "TagPrefix": "java/storage/azure-storage-blob",
-  "Tag": "java/storage/azure-storage-blob_494b72fca7"
+  "Tag": "java/storage/azure-storage-blob_c4cdc68ef6"
 }

sdk/storage/azure-storage-blob/pom.xml

@@ -138,6 +138,20 @@
       <version>1.17.7</version> <!-- {x-version-update;testdep_net.bytebuddy:byte-buddy-agent;external_dependency} -->
       <scope>test</scope>
     </dependency>
+
+    <!-- Arrow dependency -->
+    <!-- The internal ListBlobs Arrow reader decodes the IPC stream using vendored FlatBuffer metadata accessors
+         (com.azure.storage.blob.implementation.util.arrow) backed by flatbuffers-java, so flatbuffers-java is the sole
+         dependency needed to parse the Arrow format. The arrow-format artifact is deliberately NOT a dependency: it
+         ships Java 11 bytecode and would break the Java 8 baseline build. The reader's behavior is validated against
+         committed golden base64 fixtures (src/test/resources/arrow) rather than the official arrow-vector writer, so no
+         Apache Arrow test dependency is required. -->
+    <dependency>
+      <groupId>com.google.flatbuffers</groupId>
+      <artifactId>flatbuffers-java</artifactId>
+      <version>24.3.25</version> <!-- {x-version-update;com.google.flatbuffers:flatbuffers-java;external_dependency} -->
+    </dependency>
+
   </dependencies>
 
   <profiles>

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/BlobContainerAsyncClient.java

@@ -28,6 +28,8 @@
 import com.azure.storage.blob.implementation.models.EncryptionScope;
 import com.azure.storage.blob.implementation.models.ListBlobsFlatSegmentResponse;
 import com.azure.storage.blob.implementation.models.ListBlobsHierarchySegmentResponse;
+import com.azure.storage.blob.implementation.util.ArrowBlobListDeserializer;
+import com.azure.storage.blob.implementation.util.ArrowBlobListDeserializer.ArrowListBlobsResult;
 import com.azure.storage.blob.implementation.util.BlobConstants;
 import com.azure.storage.blob.implementation.util.BlobSasImplUtil;
 import com.azure.storage.blob.implementation.util.ModelHelper;
@@ -42,6 +44,7 @@
 import com.azure.storage.blob.models.CustomerProvidedKey;
 import com.azure.storage.blob.models.ListBlobsIncludeItem;
 import com.azure.storage.blob.models.ListBlobsOptions;
+import com.azure.storage.blob.models.StorageResponseSerializationFormat;
 import com.azure.storage.blob.models.PublicAccessType;
 import com.azure.storage.blob.models.StorageAccountInfo;
 import com.azure.storage.blob.models.TaggedBlobItem;
@@ -1118,18 +1121,35 @@ PagedFlux<BlobItem> listBlobsFlatWithOptionalTimeout(ListBlobsOptions options, S
         Duration timeout) {
         BiFunction<String, Integer, Mono<PagedResponse<BlobItem>>> func = (marker, pageSize) -> {
             ListBlobsOptions finalOptions;
+            /*
+             If pageSize was not set in a .byPage(int) method, the page size from options will be preserved.
+             Otherwise, prefer the new value.
+             */
             if (pageSize != null) {
                 if (options == null) {
                     finalOptions = new ListBlobsOptions().setMaxResultsPerPage(pageSize);
                 } else {
+                    // Note that this prefers the value passed to .byPage(int) over the value on the options
                     finalOptions = new ListBlobsOptions().setMaxResultsPerPage(pageSize)
                         .setPrefix(options.getPrefix())
-                        .setDetails(options.getDetails());
+                        .setDetails(options.getDetails())
+                        .setStartFrom(options.getStartFrom());
+                    if (ModelHelper.resolveSerializationFormat(options.getStorageResponseSerializationFormat())
+                        == StorageResponseSerializationFormat.ARROW) {
+                        finalOptions.setStorageResponseSerializationFormat(StorageResponseSerializationFormat.ARROW)
+                            .setEndBefore(options.getEndBefore());
+                    }
                 }
             } else {
                 finalOptions = options;
             }
 
+            if (finalOptions != null
+                && ModelHelper.resolveSerializationFormat(finalOptions.getStorageResponseSerializationFormat())
+                    == StorageResponseSerializationFormat.ARROW) {
+                return listBlobsFlatSegmentArrow(marker, finalOptions, timeout);
+            }
+
             return listBlobsFlatSegment(marker, finalOptions, timeout).map(response -> {
                 List<BlobItem> value = response.getValue().getSegment() == null
                     ? Collections.emptyList()
@@ -1177,6 +1197,61 @@ PagedFlux<BlobItem> listBlobsFlatWithOptionalTimeout(ListBlobsOptions options, S
             timeout);
     }
 
+    private Mono<PagedResponse<BlobItem>> listBlobsFlatSegmentArrow(String marker, ListBlobsOptions options,
+        Duration timeout) {
+        options = options == null ? new ListBlobsOptions() : options;
+
+        ArrayList<ListBlobsIncludeItem> include
+            = options.getDetails().toList().isEmpty() ? null : options.getDetails().toList();
+
+        ListBlobsOptions finalOptions = options;
+        return StorageImplUtils.applyOptionalTimeout(this.azureBlobStorage.getContainers()
+            .listBlobFlatSegmentApacheArrowWithResponseAsync(containerName, finalOptions.getPrefix(), marker,
+                finalOptions.getMaxResultsPerPage(), include, null, finalOptions.getStartFrom(),
+                finalOptions.getEndBefore(), null, Context.NONE),
+            timeout).flatMap(response -> {
+                String contentType = response.getHeaders().getValue(com.azure.core.http.HttpHeaderName.CONTENT_TYPE);
+
+                return FluxUtil.collectBytesInByteBufferStream(response.getValue()).map(bytes -> {
+                    java.io.ByteArrayInputStream inputStream = new java.io.ByteArrayInputStream(bytes);
+
+                    if (contentType != null && contentType.contentEquals("application/vnd.apache.arrow.stream")) {
+                        ArrowListBlobsResult arrowResult = ArrowBlobListDeserializer.deserialize(inputStream);
+
+                        List<BlobItem> value = arrowResult.getBlobItems()
+                            .stream()
+                            .map(ModelHelper::populateBlobItem)
+                            .collect(Collectors.toList());
+
+                        return (PagedResponse<BlobItem>) new PagedResponseBase<>(response.getRequest(),
+                            response.getStatusCode(), response.getHeaders(), value, arrowResult.getNextMarker(),
+                            response.getDeserializedHeaders());
+                    } else {
+                        // XML fallback
+                        try {
+                            ListBlobsFlatSegmentResponse xmlResponse
+                                = ListBlobsFlatSegmentResponse.fromXml(com.azure.xml.XmlReader.fromStream(inputStream));
+
+                            List<BlobItem> value = xmlResponse.getSegment() == null
+                                ? Collections.emptyList()
+                                : xmlResponse.getSegment()
+                                    .getBlobItems()
+                                    .stream()
+                                    .map(ModelHelper::populateBlobItem)
+                                    .collect(Collectors.toList());
+
+                            return (PagedResponse<BlobItem>) new PagedResponseBase<>(response.getRequest(),
+                                response.getStatusCode(), response.getHeaders(), value, xmlResponse.getNextMarker(),
+                                null);
+                        } catch (javax.xml.stream.XMLStreamException e) {
+                            throw LOGGER
+                                .logExceptionAsError(new RuntimeException("Failed to parse XML fallback response", e));
+                        }
+                    }
+                });
+            });
+    }
+
     /**
      * Returns a reactive Publisher emitting all the blobs and directories (prefixes) under the given directory
      * (prefix). Directories will have {@link BlobItem#isPrefix()} set to true.
@@ -1302,10 +1377,22 @@ PagedFlux<BlobItem> listBlobsHierarchyWithOptionalTimeout(String delimiter, List
                         .setPrefix(options.getPrefix())
                         .setDetails(options.getDetails())
                         .setStartFrom(options.getStartFrom());
+                    if (ModelHelper.resolveSerializationFormat(options.getStorageResponseSerializationFormat())
+                        == StorageResponseSerializationFormat.ARROW) {
+                        finalOptions.setStorageResponseSerializationFormat(StorageResponseSerializationFormat.ARROW)
+                            .setEndBefore(options.getEndBefore());
+                    }
                 }
             } else {
                 finalOptions = options;
             }
+
+            if (finalOptions != null
+                && ModelHelper.resolveSerializationFormat(finalOptions.getStorageResponseSerializationFormat())
+                    == StorageResponseSerializationFormat.ARROW) {
+                return listBlobsHierarchySegmentArrow(marker, delimiter, finalOptions, timeout);
+            }
+
             return listBlobsHierarchySegment(marker, delimiter, finalOptions, timeout).map(response -> {
                 BlobHierarchyListSegment segment = response.getValue().getSegment();
                 List<BlobItem> value;
@@ -1344,6 +1431,70 @@ PagedFlux<BlobItem> listBlobsHierarchyWithOptionalTimeout(String delimiter, List
             timeout);
     }
 
+    private Mono<PagedResponse<BlobItem>> listBlobsHierarchySegmentArrow(String marker, String delimiter,
+        ListBlobsOptions options, Duration timeout) {
+        options = options == null ? new ListBlobsOptions() : options;
+        if (options.getDetails().getRetrieveSnapshots()) {
+            throw LOGGER.logExceptionAsError(
+                new UnsupportedOperationException("Including snapshots in a hierarchical listing is not supported."));
+        }
+
+        ArrayList<ListBlobsIncludeItem> include
+            = options.getDetails().toList().isEmpty() ? null : options.getDetails().toList();
+
+        ListBlobsOptions finalOptions = options;
+        return StorageImplUtils
+            .applyOptionalTimeout(this.azureBlobStorage.getContainers()
+                .listBlobHierarchySegmentApacheArrowWithResponseAsync(containerName, delimiter,
+                    finalOptions.getPrefix(), marker, finalOptions.getMaxResultsPerPage(), include, null,
+                    finalOptions.getStartFrom(), finalOptions.getEndBefore(), null, Context.NONE),
+                timeout)
+            .flatMap(response -> {
+                String contentType = response.getHeaders().getValue(com.azure.core.http.HttpHeaderName.CONTENT_TYPE);
+
+                return FluxUtil.collectBytesInByteBufferStream(response.getValue()).map(bytes -> {
+                    java.io.ByteArrayInputStream inputStream = new java.io.ByteArrayInputStream(bytes);
+
+                    if (contentType != null && contentType.contentEquals("application/vnd.apache.arrow.stream")) {
+                        ArrowListBlobsResult arrowResult = ArrowBlobListDeserializer.deserialize(inputStream);
+
+                        List<BlobItem> value = arrowResult.getBlobItems()
+                            .stream()
+                            .map(ModelHelper::populateBlobItem)
+                            .collect(Collectors.toList());
+
+                        return (PagedResponse<BlobItem>) new PagedResponseBase<>(response.getRequest(),
+                            response.getStatusCode(), response.getHeaders(), value, arrowResult.getNextMarker(),
+                            response.getDeserializedHeaders());
+                    } else {
+                        // XML fallback
+                        try {
+                            ListBlobsHierarchySegmentResponse xmlResponse = ListBlobsHierarchySegmentResponse
+                                .fromXml(com.azure.xml.XmlReader.fromStream(inputStream));
+
+                            BlobHierarchyListSegment segment = xmlResponse.getSegment();
+                            List<BlobItem> value = new ArrayList<>();
+                            if (segment != null) {
+                                segment.getBlobItems()
+                                    .forEach(item -> value.add(BlobItemConstructorProxy.create(item)));
+                                segment.getBlobPrefixes()
+                                    .forEach(prefix -> value
+                                        .add(new BlobItem().setName(ModelHelper.toBlobNameString(prefix.getName()))
+                                            .setIsPrefix(true)));
+                            }
+
+                            return (PagedResponse<BlobItem>) new PagedResponseBase<>(response.getRequest(),
+                                response.getStatusCode(), response.getHeaders(), value, xmlResponse.getNextMarker(),
+                                null);
+                        } catch (javax.xml.stream.XMLStreamException e) {
+                            throw LOGGER
+                                .logExceptionAsError(new RuntimeException("Failed to parse XML fallback response", e));
+                        }
+                    }
+                });
+            });
+    }
+
     /**
      * Returns a reactive Publisher emitting the blobs in this container whose tags match the query expression. For more
      * information, including information on the query syntax, see the <a href="https://docs.microsoft.com/rest/api/storageservices/find-blobs-by-tags">Azure Docs</a>.