Stg104/photon

eng/versioning/external_dependencies.txt

@@ -162,6 +162,10 @@ commons-io:commons-io;2.17.0
 net.java.dev.jna:jna-platform;5.17.0
 net.oneandone.reflections8:reflections8;0.11.7
 net.jonathangiles.tools:dependencyChecker-maven-plugin;1.0.6
+org.apache.arrow:arrow-format;18.1.0
+org.apache.arrow:arrow-memory-core;18.1.0
+org.apache.arrow:arrow-memory-unsafe;18.1.0
+org.apache.arrow:arrow-vector;18.1.0
 org.apache.commons:commons-collections4;4.4
 org.apache.commons:commons-text;1.10.0
 org.apache.maven:maven-plugin-api;3.9.11
@@ -186,6 +190,7 @@ org.apache.maven.plugins:maven-shade-plugin;3.6.0
 org.apache.maven.plugins:maven-site-plugin;3.21.0
 org.apache.maven.plugins:maven-source-plugin;3.3.1
 org.apache.maven.plugins:maven-surefire-plugin;3.5.3
+org.checkerframework:checker-qual;3.42.0
 org.codehaus.mojo:animal-sniffer-maven-plugin;1.24
 org.codehaus.mojo:build-helper-maven-plugin;3.6.1
 org.codehaus.mojo:exec-maven-plugin;3.5.1

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

@@ -602,6 +602,11 @@
 
                 <!-- 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} -->
+
+                <include>org.apache.arrow:arrow-memory-core:[18.1.0]</include> <!-- {x-include-update;org.apache.arrow:arrow-memory-core;external_dependency} -->
+                <include>org.apache.arrow:arrow-memory-unsafe:[18.1.0]</include> <!-- {x-include-update;org.apache.arrow:arrow-memory-unsafe;external_dependency} -->
+                <include>org.apache.arrow:arrow-vector:[18.1.0]</include> <!-- {x-include-update;org.apache.arrow:arrow-vector;external_dependency} -->
+                <include>org.checkerframework:checker-qual:[3.42.0]</include> <!-- {x-include-update;org.checkerframework:checker-qual;external_dependency} -->
               </includes>
             </bannedDependencies>
 

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

@@ -56,6 +56,8 @@
       --add-reads com.azure.storage.blob=com.azure.core.http.okhttp
       --add-reads com.azure.storage.blob=com.azure.core.http.jdk.httpclient
       --add-reads com.azure.storage.blob=com.azure.core.http.vertx
+      --add-opens java.base/java.nio=ALL-UNNAMED <!-- Arrow's BufferAllocator uses sun.misc.Unsafe to allocate direct byte buffers, which requires this open. -->
+      --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED <!-- Arrow's BufferAllocator uses sun.misc.Unsafe to allocate direct byte buffers, which requires this open. -->
     </javaModulesSurefireArgLine>
     <parallelizeLiveTests>concurrent</parallelizeLiveTests>
 
@@ -138,8 +140,40 @@
       <version>1.17.7</version> <!-- {x-version-update;testdep_net.bytebuddy:byte-buddy-agent;external_dependency} -->
       <scope>test</scope>
     </dependency>
+
+    <!-- Arrow dependencies -->
+    <dependency>
+      <groupId>org.apache.arrow</groupId>
+      <artifactId>arrow-vector</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.arrow</groupId>
+      <artifactId>arrow-memory-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.arrow</groupId>
+      <artifactId>arrow-memory-unsafe</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.checkerframework</groupId>
+      <artifactId>checker-qual</artifactId>
+      <version>3.42.0</version>
+    </dependency>
+
   </dependencies>
 
+  <dependencyManagement>
+    <dependencies>
+      <dependency>
+        <groupId>org.apache.arrow</groupId>
+        <artifactId>arrow-bom</artifactId>
+        <version>18.1.0</version>
+        <type>pom</type>
+        <scope>import</scope>
+      </dependency>
+    </dependencies>
+  </dependencyManagement>
+
   <profiles>
     <profile>
       <id>inject-sas-service-version</id>

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

@@ -28,6 +28,7 @@
 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.BlobConstants;
 import com.azure.storage.blob.implementation.util.BlobSasImplUtil;
 import com.azure.storage.blob.implementation.util.ModelHelper;
@@ -1125,11 +1126,18 @@ PagedFlux<BlobItem> listBlobsFlatWithOptionalTimeout(ListBlobsOptions options, S
                     finalOptions = new ListBlobsOptions().setMaxResultsPerPage(pageSize)
                         .setPrefix(options.getPrefix())
                         .setDetails(options.getDetails());
+                    if (Boolean.TRUE.equals(options.isApacheArrowEnabled())) {
+                        finalOptions.setApacheArrowEnabled(true).setEndBefore(options.getEndBefore());
+                    }
                 }
             } else {
                 finalOptions = options;
             }
 
+            if (finalOptions != null && Boolean.TRUE.equals(finalOptions.isApacheArrowEnabled())) {
+                return listBlobsFlatSegmentArrow(marker, finalOptions, timeout);
+            }
+
             return listBlobsFlatSegment(marker, finalOptions, timeout).map(response -> {
                 List<BlobItem> value = response.getValue().getSegment() == null
                     ? Collections.emptyList()
@@ -1177,6 +1185,62 @@ 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 bais = new java.io.ByteArrayInputStream(bytes);
+
+                    if (contentType != null && contentType.contentEquals("application/vnd.apache.arrow.stream")) {
+                        ArrowBlobListDeserializer.ArrowListBlobsResult arrowResult
+                            = ArrowBlobListDeserializer.deserialize(bais);
+
+                        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(bais));
+
+                            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 +1366,18 @@ PagedFlux<BlobItem> listBlobsHierarchyWithOptionalTimeout(String delimiter, List
                         .setPrefix(options.getPrefix())
                         .setDetails(options.getDetails())
                         .setStartFrom(options.getStartFrom());
+                    if (Boolean.TRUE.equals(options.isApacheArrowEnabled())) {
+                        finalOptions.setApacheArrowEnabled(true).setEndBefore(options.getEndBefore());
+                    }
                 }
             } else {
                 finalOptions = options;
             }
+
+            if (finalOptions != null && Boolean.TRUE.equals(finalOptions.isApacheArrowEnabled())) {
+                return listBlobsHierarchySegmentArrow(marker, delimiter, finalOptions, timeout);
+            }
+
             return listBlobsHierarchySegment(marker, delimiter, finalOptions, timeout).map(response -> {
                 BlobHierarchyListSegment segment = response.getValue().getSegment();
                 List<BlobItem> value;
@@ -1344,6 +1416,71 @@ 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 bais = new java.io.ByteArrayInputStream(bytes);
+
+                    if (contentType != null && contentType.contentEquals("application/vnd.apache.arrow.stream")) {
+                        ArrowBlobListDeserializer.ArrowListBlobsResult arrowResult
+                            = ArrowBlobListDeserializer.deserialize(bais);
+
+                        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(bais));
+
+                            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>.