ESQL: add Arrow dataframes output format

docs/changelog/109873.yaml

@@ -0,0 +1,5 @@
+pr: 109873
+summary: "ESQL: add Arrow dataframes output format"
+area: ES|QL
+type: feature
+issues: []

docs/reference/esql/esql-rest.asciidoc

@@ -111,6 +111,9 @@ s|Description
 |{wikipedia}/Smile_(data_interchange_format)[Smile] binary data format similar
 to CBOR
 
+|arrow
+|application/vnd.apache.arrow.stream
+|**Experimental.** https://arrow.apache.org/[Apache Arrow] dataframes, https://arrow.apache.org/docs/format/Columnar.html#ipc-streaming-format[IPC streaming format]
 |===
 
 The `csv` format accepts a formatting URL query attribute, `delimiter`, which

gradle/verification-metadata.xml

@@ -581,6 +581,11 @@
             <sha256 value="baf7d6ea97ce606c53e11b6854ba5f2ce7ef5c24dddf0afa18d1260bd25b002c" origin="Generated by Gradle"/>
          </artifact>
       </component>
+      <component group="com.google.flatbuffers" name="flatbuffers-java" version="23.5.26">
+         <artifact name="flatbuffers-java-23.5.26.jar">
+            <sha256 value="8d10cac2ea9878896077ba437d76fdb1b9a07f55a863c560bb8a024b04103f8b" origin="Generated by Gradle"/>
+         </artifact>
+      </component>
       <component group="com.google.googlejavaformat" name="google-java-format" version="1.16.0">
          <artifact name="google-java-format-1.16.0.jar">
             <sha256 value="0cff5d0230ba20d538f3f70b2aa68bd33f9fdc69768cde07337c563c23eb7c43" origin="Generated by Gradle"/>
@@ -1841,6 +1846,26 @@
             <sha256 value="cd7695b3bfb6964ab71b6a0b31dad60005ae77fe502132364679aacf08f77970" origin="Generated by Gradle"/>
          </artifact>
       </component>
+      <component group="org.apache.arrow" name="arrow-format" version="16.1.0">
+         <artifact name="arrow-format-16.1.0.jar">
+            <sha256 value="ad97e0fc72e193b1de3cbce4818d1ff16e81673fd523d001e8d2774bde40ee6c" origin="Generated by Gradle"/>
+         </artifact>
+      </component>
+      <component group="org.apache.arrow" name="arrow-memory-core" version="16.1.0">
+         <artifact name="arrow-memory-core-16.1.0.jar">
+            <sha256 value="da7af1a1a899bd5a1b6c71284243b9f3c0e1098f0cb10cd7be4b8b455ced79dd" origin="Generated by Gradle"/>
+         </artifact>
+      </component>
+      <component group="org.apache.arrow" name="arrow-memory-unsafe" version="16.1.0">
+         <artifact name="arrow-memory-unsafe-16.1.0.jar">
+            <sha256 value="6534eded25f2c30593416a294c1047f0b017baa9906d98f6f3270737b076c745" origin="Generated by Gradle"/>
+         </artifact>
+      </component>
+      <component group="org.apache.arrow" name="arrow-vector" version="16.1.0">
+         <artifact name="arrow-vector-16.1.0.jar">
+            <sha256 value="c5837b3aa24dfd93759f57bc5759b9a8fbb5bf3912d55994d70cabb904436aab" origin="Generated by Gradle"/>
+         </artifact>
+      </component>
       <component group="org.apache.avro" name="avro" version="1.7.4">
          <artifact name="avro-1.7.4.jar">
             <sha256 value="a01d26e9a5ed0754e8c88dbb373fba896c57df0a0c424185767a3857855bb222" origin="Generated by Gradle"/>
@@ -3177,6 +3202,11 @@
             <sha256 value="e316255bbfcd9fe50d165314b85abb2b33cb2a66a93c491db648e498a82c2de1" origin="Generated by Gradle"/>
          </artifact>
       </component>
+      <component group="org.checkerframework" name="checker-qual" version="3.42.0">
+         <artifact name="checker-qual-3.42.0.jar">
+            <sha256 value="ccaedd33af0b7894d9f2f3b644f4d19e43928e32902e61ac4d10777830f5aac7" origin="Generated by Gradle"/>
+         </artifact>
+      </component>
       <component group="org.checkerframework" name="checker-qual" version="3.5.0">
          <artifact name="checker-qual-3.5.0.jar">
             <sha256 value="729990b3f18a95606fc2573836b6958bcdb44cb52bfbd1b7aa9c339cff35a5a4" origin="Generated by Gradle"/>

server/src/main/java/org/elasticsearch/common/io/stream/RecyclerBytesStreamOutput.java

@@ -32,7 +32,9 @@
 public class RecyclerBytesStreamOutput extends BytesStream implements Releasable {
 
     static final VarHandle VH_BE_INT = MethodHandles.byteArrayViewVarHandle(int[].class, ByteOrder.BIG_ENDIAN);
+    static final VarHandle VH_LE_INT = MethodHandles.byteArrayViewVarHandle(int[].class, ByteOrder.LITTLE_ENDIAN);
     static final VarHandle VH_BE_LONG = MethodHandles.byteArrayViewVarHandle(long[].class, ByteOrder.BIG_ENDIAN);
+    static final VarHandle VH_LE_LONG = MethodHandles.byteArrayViewVarHandle(long[].class, ByteOrder.LITTLE_ENDIAN);
 
     private final ArrayList<Recycler.V<BytesRef>> pages = new ArrayList<>();
     private final Recycler<BytesRef> recycler;
@@ -106,6 +108,17 @@ public void writeInt(int i) throws IOException {
         }
     }
 
+    @Override
+    public void writeIntLE(int i) throws IOException {
+        if (4 > (pageSize - currentPageOffset)) {
+            super.writeIntLE(i);
+        } else {
+            BytesRef currentPage = pages.get(pageIndex).v();
+            VH_LE_INT.set(currentPage.bytes, currentPage.offset + currentPageOffset, i);
+            currentPageOffset += 4;
+        }
+    }
+
     @Override
     public void writeLong(long i) throws IOException {
         if (8 > (pageSize - currentPageOffset)) {
@@ -117,6 +130,17 @@ public void writeLong(long i) throws IOException {
         }
     }
 
+    @Override
+    public void writeLongLE(long i) throws IOException {
+        if (8 > (pageSize - currentPageOffset)) {
+            super.writeLongLE(i);
+        } else {
+            BytesRef currentPage = pages.get(pageIndex).v();
+            VH_LE_LONG.set(currentPage.bytes, currentPage.offset + currentPageOffset, i);
+            currentPageOffset += 8;
+        }
+    }
+
     @Override
     public void writeWithSizePrefix(Writeable writeable) throws IOException {
         // TODO: do this without copying the bytes from tmp by calling writeBytes and just use the pages in tmp directly through

server/src/main/java/org/elasticsearch/common/io/stream/StreamOutput.java

@@ -190,6 +190,15 @@ public void writeInt(int i) throws IOException {
         writeBytes(buffer, 0, 4);
     }
 
+    /**
+     * Writes an int as four bytes, least significant bytes first.
+     */
+    public void writeIntLE(int i) throws IOException {
+        final byte[] buffer = scratch.get();
+        ByteUtils.writeIntLE(i, buffer, 0);
+        writeBytes(buffer, 0, 4);
+    }
+
     /**
      * Writes an int in a variable-length format.  Writes between one and
      * five bytes.  Smaller values take fewer bytes.  Negative numbers
@@ -243,6 +252,15 @@ public void writeLong(long i) throws IOException {
         writeBytes(buffer, 0, 8);
     }
 
+    /**
+     * Writes a long as eight bytes.
+     */
+    public void writeLongLE(long i) throws IOException {
+        final byte[] buffer = scratch.get();
+        ByteUtils.writeLongLE(i, buffer, 0);
+        writeBytes(buffer, 0, 8);
+    }
+
     /**
      * Writes a non-negative long in a variable-length format. Writes between one and ten bytes. Smaller values take fewer bytes. Negative
      * numbers use ten bytes and trip assertions (if running in tests) so prefer {@link #writeLong(long)} or {@link #writeZLong(long)} for
@@ -442,6 +460,10 @@ public void writeDouble(double v) throws IOException {
         writeLong(Double.doubleToLongBits(v));
     }
 
+    public void writeDoubleLE(double v) throws IOException {
+        writeLongLE(Double.doubleToLongBits(v));
+    }
+
     public void writeOptionalDouble(@Nullable Double v) throws IOException {
         if (v == null) {
             writeBoolean(false);

x-pack/plugin/esql/arrow/build.gradle

@@ -0,0 +1,61 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+apply plugin: 'elasticsearch.build'
+
+dependencies {
+  compileOnly project(':server')
+  compileOnly project(':x-pack:plugin:esql:compute')
+  compileOnly project(':x-pack:plugin:esql-core')
+  compileOnly project(':x-pack:plugin:mapper-version')
+  implementation('org.apache.arrow:arrow-vector:16.1.0')
+  implementation('org.apache.arrow:arrow-format:16.1.0')
+  implementation('org.apache.arrow:arrow-memory-core:16.1.0')
+  implementation('org.checkerframework:checker-qual:3.42.0')
+  implementation('com.google.flatbuffers:flatbuffers-java:23.5.26')
+  // Needed for the json arrow serialization, and loaded even if we don't use it.
+  implementation("com.fasterxml.jackson.core:jackson-annotations:${versions.jackson}")
+  implementation("com.fasterxml.jackson.core:jackson-core:${versions.jackson}")
+  implementation("com.fasterxml.jackson.core:jackson-databind:${versions.jackson}")
+  implementation("org.slf4j:slf4j-api:${versions.slf4j}")
+  runtimeOnly "org.slf4j:slf4j-nop:${versions.slf4j}"
+
+  testImplementation project(':test:framework')
+  testImplementation('org.apache.arrow:arrow-memory-unsafe:16.1.0')
+}
+
+tasks.named("dependencyLicenses").configure {
+  mapping from: /jackson-.*/, to: 'jackson'
+  mapping from: /arrow-.*/, to: 'arrow'
+  mapping from: /slf4j-.*/, to: 'slf4j'
+}
+
+tasks.named("thirdPartyAudit").configure {
+  ignoreViolations(
+    // uses sun.misc.Unsafe. Only used in tests.
+    'org.apache.arrow.memory.util.hash.SimpleHasher',
+    'org.apache.arrow.memory.util.hash.MurmurHasher',
+    'org.apache.arrow.memory.util.MemoryUtil',
+    'org.apache.arrow.memory.util.MemoryUtil$1',
+    'org.apache.arrow.vector.DecimalVector',
+    'org.apache.arrow.vector.BaseFixedWidthVector',
+    'org.apache.arrow.vector.util.DecimalUtility',
+    'org.apache.arrow.vector.Decimal256Vector',
+    'org.apache.arrow.vector.util.VectorAppender',
+    'org.apache.arrow.memory.ArrowBuf',
+    'org.apache.arrow.vector.BitVectorHelper',
+    'org.apache.arrow.memory.util.ByteFunctionHelpers',
+  )
+  ignoreMissingClasses(
+    'org.apache.commons.codec.binary.Hex'
+  )
+}
+
+test {
+  jvmArgs('--add-opens=java.base/java.nio=ALL-UNNAMED')
+}