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);
+}