Upgrade arrow to 2023 Spring version (#76)

* upgrade jni and tokio

* fix lint

* add unit test and binding

* fix javadoc

* add err case

Author: jimexist

datafusion-examples/src/main/java/org/apache/arrow/datafusion/examples/ExampleMain.java

@@ -55,7 +55,16 @@ public static void main(String[] args) throws Exception {
 
       context
           .sql("select * from test_parquet limit 3")
-          .thenComposeAsync(df -> df.registerTable(context, "test_parquet_limited"))
+          .thenAccept(
+              df -> {
+                try {
+                  boolean previouslyRegistered =
+                      context.registerTable("test_parquet_limited", df.intoView()).isPresent();
+                  assert !previouslyRegistered;
+                } catch (Exception e) {
+                  throw new RuntimeException(e);
+                }
+              })
           .join();
 
       context.sql("select * from test_parquet_limited").thenComposeAsync(DataFrame::show).join();

datafusion-java/build.gradle

@@ -30,7 +30,7 @@ java {
     withSourcesJar()
 
     compileJava {
-        options.compilerArgs += ["-h", "${buildDir}/target/headers"]
+        options.compilerArgs += ["-h", "${layout.buildDirectory.asFile.get()}/target/headers"]
     }
 }
 

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

@@ -45,11 +45,11 @@ public interface DataFrame extends NativeProxy {
   CompletableFuture<Void> writeCsv(Path path);
 
   /**
-   * Register this dataframe as a temporary table.
+   * Converts this DataFrame into a TableProvider that can be registered as a table view using
+   * {@link SessionContext#registerParquet(String, Path)}
    *
-   * @param context SessionContext to register table to
-   * @param name name of the tmp table
-   * @return null
+   * @return the table provider ready to be e.g. {@link SessionContext#registerTable(String,
+   *     TableProvider) registered}.
    */
-  CompletableFuture<Void> registerTable(SessionContext context, String name);
+  TableProvider intoView();
 }

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

@@ -20,6 +20,5 @@ static native void writeParquet(
 
   static native void writeCsv(long runtime, long dataframe, String path, Consumer<String> callback);
 
-  static native void registerTable(
-      long runtime, long dataframe, long context, String name, Consumer<String> callback);
+  static native long intoView(long dataframe);
 }

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

@@ -104,25 +104,11 @@ public CompletableFuture<Void> writeCsv(Path path) {
     return future;
   }
 
-  public CompletableFuture<Void> registerTable(SessionContext ctx, String name) {
-    Runtime runtime = context.getRuntime();
-    long runtimePointer = runtime.getPointer();
+  @Override
+  public TableProvider intoView() {
     long dataframe = getPointer();
-    long contextPointer = ctx.getPointer();
-    CompletableFuture<Void> future = new CompletableFuture<>();
-    DataFrames.registerTable(
-        runtimePointer,
-        dataframe,
-        contextPointer,
-        name,
-        (String errString) -> {
-          if (containsError(errString)) {
-            future.completeExceptionally(new RuntimeException(errString));
-          } else {
-            future.complete(null);
-          }
-        });
-    return future;
+    long tableProviderPointer = DataFrames.intoView(dataframe);
+    return new DefaultTableProvider(tableProviderPointer);
   }
 
   @Override

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

@@ -1,6 +1,7 @@
 package org.apache.arrow.datafusion;
 
 import java.nio.file.Path;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
 import org.slf4j.Logger;
@@ -19,6 +20,8 @@ static native void registerCsv(
   static native void registerParquet(
       long runtime, long context, String name, String path, Consumer<String> callback);
 
+  static native long registerTable(long context, String name, long tableProvider) throws Exception;
+
   @Override
   public CompletableFuture<DataFrame> sql(String sql) {
     long runtime = getRuntime().getPointer();
@@ -64,6 +67,16 @@ public CompletableFuture<Void> registerParquet(String name, Path path) {
     return future;
   }
 
+  @Override
+  public Optional<TableProvider> registerTable(String name, TableProvider tableProvider)
+      throws Exception {
+    long previouslyRegistered = registerTable(getPointer(), name, tableProvider.getPointer());
+    if (previouslyRegistered == 0) {
+      return Optional.empty();
+    }
+    return Optional.of(new DefaultTableProvider(previouslyRegistered));
+  }
+
   private void voidCallback(CompletableFuture<Void> future, String errMessage) {
     if (null != errMessage && !errMessage.isEmpty()) {
       future.completeExceptionally(new RuntimeException(errMessage));

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

@@ -0,0 +1,12 @@
+package org.apache.arrow.datafusion;
+
+class DefaultTableProvider extends AbstractProxy implements TableProvider {
+  DefaultTableProvider(long pointer) {
+    super(pointer);
+  }
+
+  @Override
+  void doClose(long pointer) throws Exception {
+    TableProviders.destroyTableProvider(pointer);
+  }
+}

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

@@ -1,6 +1,7 @@
 package org.apache.arrow.datafusion;
 
 import java.nio.file.Path;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
 /** A session context holds resources and is the entrance for obtaining {@link DataFrame} */
@@ -32,6 +33,17 @@ public interface SessionContext extends AutoCloseable, NativeProxy {
    */
   CompletableFuture<Void> registerParquet(String name, Path path);
 
+  /**
+   * Registers a TableProvider as a table that can be referenced from SQL statements executed
+   * against this context.
+   *
+   * @param name table reference
+   * @param tableProvider table provider
+   * @return as of Arrow 22 this is only {@link Optional#empty()}
+   * @throws Exception when the table is already registered
+   */
+  Optional<TableProvider> registerTable(String name, TableProvider tableProvider) throws Exception;
+
   /**
    * Get the runtime associated with this context
    *

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

@@ -0,0 +1,4 @@
+package org.apache.arrow.datafusion;
+
+/** vague interface that maps to {@code Arc<dyn TableProvider>}. */
+public interface TableProvider extends NativeProxy {}

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

@@ -0,0 +1,8 @@
+package org.apache.arrow.datafusion;
+
+class TableProviders {
+
+  private TableProviders() {}
+
+  static native void destroyTableProvider(long pointer);
+}

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

@@ -1,6 +1,9 @@
 package org.apache.arrow.datafusion;
 
-import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -15,6 +18,23 @@
 import org.junit.jupiter.api.io.TempDir;
 
 public class TestQuery {
+
+  @Test
+  public void testQueryInMemoryTable() throws Exception {
+    try (SessionContext context = SessionContexts.create();
+        BufferAllocator allocator = new RootAllocator()) {
+      DataFrame df = context.sql("SELECT * FROM (VALUES (1, 2), (3, 4)) AS t (x, y)").join();
+      assertFalse(
+          context.registerTable("test", df.intoView()).isPresent(),
+          "there should not be any duplicates");
+      testQuery(context, allocator);
+      assertThrows(
+          Exception.class,
+          () -> context.registerTable("test", df.intoView()),
+          "now there should be duplicates");
+    }
+  }
+
   @Test
   public void testQueryCsv(@TempDir Path tempDir) throws Exception {
     try (SessionContext context = SessionContexts.create();

datafusion-jni/Cargo.toml

@@ -12,8 +12,8 @@ edition = "2021"
 [dependencies]
 jni = "^0.21.0"
 tokio = "^1.32.0"
-arrow = "^22.0"
-datafusion = "^12.0"
+arrow = "^36.0"
+datafusion = "^22.0"
 
 [lib]
 crate_type = ["cdylib"]

datafusion-jni/src/context.rs

@@ -1,8 +1,10 @@
+use datafusion::datasource::TableProvider;
 use datafusion::execution::context::SessionContext;
 use datafusion::prelude::{CsvReadOptions, ParquetReadOptions};
 use jni::objects::{JClass, JObject, JString};
 use jni::sys::jlong;
 use jni::JNIEnv;
+use std::sync::Arc;
 use tokio::runtime::Runtime;
 
 #[no_mangle]
@@ -46,6 +48,34 @@ pub extern "system" fn Java_org_apache_arrow_datafusion_DefaultSessionContext_re
     });
 }
 
+#[no_mangle]
+pub extern "system" fn Java_org_apache_arrow_datafusion_DefaultSessionContext_registerTable(
+    mut env: JNIEnv,
+    _class: JClass,
+    pointer: jlong,
+    name: JString,
+    table_provider: jlong,
+) -> jlong {
+    let name: String = env
+        .get_string(&name)
+        .expect("Couldn't get name as string!")
+        .into();
+    let context = unsafe { &mut *(pointer as *mut SessionContext) };
+    let table_provider = unsafe { &*(table_provider as *const Arc<dyn TableProvider>) };
+    let result = context.register_table(&name, table_provider.clone());
+    match result {
+        // TODO this is to be fixed on datafusion side as duplicates will not be returned
+        // and instead returned as err
+        Ok(Some(v)) => Box::into_raw(Box::new(v)) as jlong,
+        Ok(None) => 0,
+        Err(err) => {
+            env.throw_new("java/lang/Exception", err.to_string())
+                .unwrap();
+            0
+        }
+    }
+}
+
 #[no_mangle]
 pub extern "system" fn Java_org_apache_arrow_datafusion_DefaultSessionContext_registerParquet(
     mut env: JNIEnv,