Add support for the Arrow IPC/Feather file format

datafusion-java/build.gradle

@@ -13,6 +13,7 @@ dependencies {
     api 'org.apache.arrow:arrow-vector:13.0.0'
     implementation 'org.apache.arrow:arrow-c-data:13.0.0'
     runtimeOnly 'org.apache.arrow:arrow-memory-unsafe:13.0.0'
+    testImplementation 'org.apache.arrow:arrow-compression:13.0.0'
     testImplementation 'org.junit.jupiter:junit-jupiter:5.8.1'
     testImplementation 'org.apache.hadoop:hadoop-client:3.3.5'
     testImplementation 'org.apache.hadoop:hadoop-common:3.3.5'

datafusion-java/src/main/java/org/apache/arrow/datafusion/ArrowFormat.java

@@ -0,0 +1,14 @@
+package org.apache.arrow.datafusion;
+
+/** The Apache Arrow IPC file format configuration. This format is also known as Feather V2 */
+public class ArrowFormat extends AbstractProxy implements FileFormat {
+  /** Create a new ArrowFormat with default options */
+  public ArrowFormat() {
+    super(FileFormats.createArrow());
+  }
+
+  @Override
+  void doClose(long pointer) {
+    FileFormats.destroyFileFormat(pointer);
+  }
+}

datafusion-java/src/main/java/org/apache/arrow/datafusion/FileFormats.java

@@ -4,6 +4,8 @@ class FileFormats {
 
   private FileFormats() {}
 
+  static native long createArrow();
+
   static native long createCsv();
 
   static native long createParquet();

datafusion-java/src/test/java/org/apache/arrow/datafusion/TestListingTable.java

@@ -2,19 +2,29 @@
 
 import static org.junit.jupiter.api.Assertions.*;
 
+import java.io.FileOutputStream;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.arrow.compression.CommonsCompressionFactory;
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.memory.RootAllocator;
 import org.apache.arrow.vector.BigIntVector;
+import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.VarCharVector;
 import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.compression.CompressionCodec;
+import org.apache.arrow.vector.compression.CompressionUtil;
+import org.apache.arrow.vector.compression.NoCompressionCodec;
+import org.apache.arrow.vector.ipc.ArrowFileWriter;
 import org.apache.arrow.vector.ipc.ArrowReader;
+import org.apache.arrow.vector.ipc.message.IpcOption;
+import org.apache.arrow.vector.types.pojo.Field;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
@@ -71,6 +81,106 @@ public void testParquetListingTable(@TempDir Path tempDir) throws Exception {
     }
   }
 
+  @Test
+  public void testArrowListingTable(@TempDir Path tempDir) throws Exception {
+    try (SessionContext context = SessionContexts.create();
+        BufferAllocator allocator = new RootAllocator()) {
+      Path dataDir = tempDir.resolve("data");
+      Files.createDirectories(dataDir);
+
+      Path arrowFilePath0 = dataDir.resolve("0.arrow");
+      Path arrowFilePath1 = dataDir.resolve("1.arrow");
+
+      // Write data files in Arrow IPC (Feather V2) file format
+      try (BigIntVector xVector = new BigIntVector("x", allocator);
+          BigIntVector yVector = new BigIntVector("y", allocator)) {
+        List<FieldVector> vectors = Arrays.asList(xVector, yVector);
+
+        for (int i = 0; i < 2; i++) {
+          xVector.setSafe(i, i * 2 + 1);
+          yVector.setSafe(i, i * 2 + 2);
+        }
+        xVector.setValueCount(2);
+        yVector.setValueCount(2);
+        writeArrowFile(arrowFilePath0, vectors, false);
+
+        xVector.reset();
+        yVector.reset();
+        for (int i = 0; i < 2; i++) {
+          xVector.setSafe(i, i * 2 + 1);
+          yVector.setSafe(i, i * 2 + 12);
+        }
+        xVector.setValueCount(2);
+        yVector.setValueCount(2);
+        writeArrowFile(arrowFilePath1, vectors, false);
+      }
+
+      try (ArrowFormat format = new ArrowFormat();
+          ListingOptions listingOptions =
+              ListingOptions.builder(format).withFileExtension(".arrow").build();
+          ListingTableConfig tableConfig =
+              ListingTableConfig.builder(dataDir)
+                  .withListingOptions(listingOptions)
+                  .build(context)
+                  .join();
+          ListingTable listingTable = new ListingTable(tableConfig)) {
+        context.registerTable("test", listingTable);
+        testQuery(context, allocator);
+      }
+    }
+  }
+
+  @Test
+  public void testCompressedArrowIpc(@TempDir Path tempDir) throws Exception {
+    try (SessionContext context = SessionContexts.create();
+        BufferAllocator allocator = new RootAllocator()) {
+      Path dataDir = tempDir.resolve("data");
+      Files.createDirectories(dataDir);
+      Path arrowFilePath0 = dataDir.resolve("0.arrow");
+
+      // Data needs to be reasonably large otherwise compression is not used
+      int numRows = 10_000;
+
+      // Write data files in compressed Arrow IPC (Feather V2) file format
+      try (BigIntVector xVector = new BigIntVector("x", allocator)) {
+        for (int i = 0; i < numRows; i++) {
+          xVector.setSafe(i, i * 2 + 1);
+        }
+        xVector.setValueCount(numRows);
+        List<FieldVector> vectors = Arrays.asList(xVector);
+        writeArrowFile(arrowFilePath0, vectors, true);
+      }
+
+      try (ArrowFormat format = new ArrowFormat();
+          ListingOptions listingOptions =
+              ListingOptions.builder(format).withFileExtension(".arrow").build();
+          ListingTableConfig tableConfig =
+              ListingTableConfig.builder(dataDir)
+                  .withListingOptions(listingOptions)
+                  .build(context)
+                  .join();
+          ListingTable listingTable = new ListingTable(tableConfig)) {
+        context.registerTable("test", listingTable);
+        try (ArrowReader reader =
+            context
+                .sql("SELECT x FROM test")
+                .thenComposeAsync(df -> df.collect(allocator))
+                .join()) {
+
+          int globalRow = 0;
+          VectorSchemaRoot root = reader.getVectorSchemaRoot();
+          while (reader.loadNextBatch()) {
+            BigIntVector xValues = (BigIntVector) root.getVector(0);
+            for (int row = 0; row < root.getRowCount(); ++row, ++globalRow) {
+              assertEquals(globalRow * 2 + 1, xValues.get(row));
+            }
+          }
+          assertEquals(numRows, globalRow);
+        }
+      }
+    }
+  }
+
   @Test
   public void testDisableCollectStat(@TempDir Path tempDir) throws Exception {
     try (SessionContext context = SessionContexts.create();
@@ -150,6 +260,30 @@ private static Path[] writeParquetFiles(Path dataDir) throws Exception {
     return new Path[] {parquetFilePath0, parquetFilePath1};
   }
 
+  private static void writeArrowFile(Path filePath, List<FieldVector> vectors, boolean compressed)
+      throws Exception {
+    List<Field> fields = vectors.stream().map(v -> v.getField()).collect(Collectors.toList());
+    CompressionUtil.CodecType codec =
+        compressed ? CompressionUtil.CodecType.ZSTD : CompressionUtil.CodecType.NO_COMPRESSION;
+    CompressionCodec.Factory compressionFactory =
+        compressed ? new CommonsCompressionFactory() : NoCompressionCodec.Factory.INSTANCE;
+    try (VectorSchemaRoot root = new VectorSchemaRoot(fields, vectors);
+        FileOutputStream output = new FileOutputStream(filePath.toString());
+        ArrowFileWriter writer =
+            new ArrowFileWriter(
+                root,
+                null,
+                output.getChannel(),
+                null,
+                IpcOption.DEFAULT,
+                compressionFactory,
+                codec)) {
+      writer.start();
+      writer.writeBatch();
+      writer.end();
+    }
+  }
+
   private static void testQuery(SessionContext context, BufferAllocator allocator)
       throws Exception {
     try (ArrowReader reader =

datafusion-jni/Cargo.toml

@@ -12,8 +12,8 @@ edition = "2021"
 [dependencies]
 jni = "^0.21.0"
 tokio = "^1.32.0"
-arrow = { version = "^36.0", features = ["ffi"] }
-datafusion = "^22.0"
+arrow = { version = "^39.0", features = ["ffi", "ipc_compression"] }
+datafusion = "^25.0"
 futures = "0.3.28"
 
 [lib]

datafusion-jni/src/file_formats.rs

@@ -1,3 +1,4 @@
+use datafusion::datasource::file_format::arrow::ArrowFormat;
 use datafusion::datasource::file_format::csv::CsvFormat;
 use datafusion::datasource::file_format::parquet::ParquetFormat;
 use datafusion::datasource::file_format::FileFormat;
@@ -28,6 +29,15 @@ pub extern "system" fn Java_org_apache_arrow_datafusion_FileFormats_createParque
     Box::into_raw(Box::new(format)) as jlong
 }
 
+#[no_mangle]
+pub extern "system" fn Java_org_apache_arrow_datafusion_FileFormats_createArrow(
+    _env: JNIEnv,
+    _class: JClass,
+) -> jlong {
+    let format: Arc<dyn FileFormat> = Arc::new(ArrowFormat::default());
+    Box::into_raw(Box::new(format)) as jlong
+}
+
 #[no_mangle]
 pub extern "system" fn Java_org_apache_arrow_datafusion_FileFormats_destroyFileFormat(
     _env: JNIEnv,