Upgrade arrow java, rust, and datafusion versions (#55)

jimexist · web-flow · commit 12e3fbed1219 · 2022-05-17T18:23:53.000+08:00
* upgrade arrow and datafusion

* bump toml version

* ExecutionContext =&gt; SessionContext

* fix more

* data frame not a trait anymore

* dataframe not dyn now

* fix arc

* remove implementation 'org.apache.arrow:arrow-c-data:8.0.0'
diff --git a/README.md b/README.md
@@ -35,8 +35,8 @@ To test it out, you can use this piece of demo code:
 package com.me;
 
 import org.apache.arrow.datafusion.DataFrame;
-import org.apache.arrow.datafusion.ExecutionContext;
-import org.apache.arrow.datafusion.ExecutionContexts;
+import org.apache.arrow.datafusion.SessionContext;
+import org.apache.arrow.datafusion.SessionContexts;
 
 public class DataFusionDemo {
 
@@ -178,7 +178,7 @@ INFO: Created user preferences directory.
 jshell> import org.apache.arrow.datafusion.*
 
 jshell> var context = ExecutionContexts.create()
-context ==> org.apache.arrow.datafusion.DefaultExecutionContext@4229bb3f
+context ==> org.apache.arrow.datafusion.DefaultSessionContext@4229bb3f
 
 jshell> var df = context.sql("select 1.1 + cos(2.0)").join()
 df ==> org.apache.arrow.datafusion.DefaultDataFrame@1a18644
diff --git a/datafusion-examples/build.gradle b/datafusion-examples/build.gradle
@@ -14,8 +14,8 @@ dependencies {
     implementation project(':datafusion-java')
     implementation 'org.slf4j:slf4j-api:1.7.36'
     implementation 'ch.qos.logback:logback-classic:1.2.11'
-    implementation 'org.apache.arrow:arrow-format:7.0.0'
-    implementation 'org.apache.arrow:arrow-vector:7.0.0'
+    implementation 'org.apache.arrow:arrow-format:8.0.0'
+    implementation 'org.apache.arrow:arrow-vector:8.0.0'
 }
 
 application {
diff --git a/datafusion-examples/src/main/java/org/apache/arrow/datafusion/examples/ExampleMain.java b/datafusion-examples/src/main/java/org/apache/arrow/datafusion/examples/ExampleMain.java
@@ -4,8 +4,8 @@
 import java.nio.file.Paths;
 import java.util.concurrent.CompletableFuture;
 import org.apache.arrow.datafusion.DataFrame;
-import org.apache.arrow.datafusion.ExecutionContext;
-import org.apache.arrow.datafusion.ExecutionContexts;
+import org.apache.arrow.datafusion.SessionContext;
+import org.apache.arrow.datafusion.SessionContexts;
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.memory.RootAllocator;
 import org.apache.arrow.vector.BigIntVector;
@@ -20,7 +20,7 @@ public class ExampleMain {
   private static final Logger logger = LoggerFactory.getLogger(ExampleMain.class);
 
   public static void main(String[] args) throws Exception {
-    try (ExecutionContext context = ExecutionContexts.create();
+    try (SessionContext context = SessionContexts.create();
         BufferAllocator allocator = new RootAllocator()) {
       loadConstant(context).join();
 
@@ -64,7 +64,7 @@ private static void consumeReader(ArrowReader reader) {
     }
   }
 
-  private static CompletableFuture<Void> loadConstant(ExecutionContext context) {
+  private static CompletableFuture<Void> loadConstant(SessionContext context) {
     return context
         .sql("select 1 + 2")
         .thenComposeAsync(
diff --git a/datafusion-java/build.gradle b/datafusion-java/build.gradle
@@ -9,9 +9,9 @@ plugins {
 
 dependencies {
     implementation 'org.slf4j:slf4j-api:1.7.36'
-    implementation 'org.apache.arrow:arrow-format:7.0.0'
-    implementation 'org.apache.arrow:arrow-vector:7.0.0'
-    runtimeOnly 'org.apache.arrow:arrow-memory-unsafe:7.0.0'
+    implementation 'org.apache.arrow:arrow-format:8.0.0'
+    implementation 'org.apache.arrow:arrow-vector:8.0.0'
+    runtimeOnly 'org.apache.arrow:arrow-memory-unsafe:8.0.0'
 }
 
 spotless {
diff --git a/datafusion-java/src/main/java/org/apache/arrow/datafusion/DefaultDataFrame.java b/datafusion-java/src/main/java/org/apache/arrow/datafusion/DefaultDataFrame.java
@@ -11,9 +11,9 @@
 class DefaultDataFrame extends AbstractProxy implements DataFrame {
 
   private static final Logger logger = LoggerFactory.getLogger(DefaultDataFrame.class);
-  private final ExecutionContext context;
+  private final SessionContext context;
 
-  DefaultDataFrame(ExecutionContext context, long pointer) {
+  DefaultDataFrame(SessionContext context, long pointer) {
     super(pointer);
     this.context = context;
   }
diff --git a/datafusion-java/src/main/java/org/apache/arrow/datafusion/DefaultSessionContext.java b/datafusion-java/src/main/java/org/apache/arrow/datafusion/DefaultSessionContext.java
@@ -6,9 +6,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-class DefaultExecutionContext extends AbstractProxy implements ExecutionContext {
+class DefaultSessionContext extends AbstractProxy implements SessionContext {
 
-  private static final Logger logger = LoggerFactory.getLogger(DefaultExecutionContext.class);
+  private static final Logger logger = LoggerFactory.getLogger(DefaultSessionContext.class);
 
   static native void querySql(
       long runtime, long context, String sql, ObjectResultCallback callback);
@@ -31,8 +31,7 @@ public CompletableFuture<DataFrame> sql(String sql) {
           if (null != errMessage && !errMessage.equals("")) {
             future.completeExceptionally(new RuntimeException(errMessage));
           } else {
-            DefaultDataFrame frame =
-                new DefaultDataFrame(DefaultExecutionContext.this, dataframeId);
+            DefaultDataFrame frame = new DefaultDataFrame(DefaultSessionContext.this, dataframeId);
             future.complete(frame);
           }
         });
@@ -80,14 +79,14 @@ public Runtime getRuntime() {
 
   private final TokioRuntime runtime;
 
-  DefaultExecutionContext(long pointer) {
+  DefaultSessionContext(long pointer) {
     super(pointer);
     this.runtime = TokioRuntime.create();
     registerChild(runtime);
   }
 
   @Override
   void doClose(long pointer) throws Exception {
-    ExecutionContexts.destroyExecutionContext(pointer);
+    SessionContexts.destroySessionContext(pointer);
   }
 }
diff --git a/datafusion-java/src/main/java/org/apache/arrow/datafusion/ExecutionContexts.java b/datafusion-java/src/main/java/org/apache/arrow/datafusion/ExecutionContexts.java
diff --git a/datafusion-java/src/main/java/org/apache/arrow/datafusion/SessionContext.java b/datafusion-java/src/main/java/org/apache/arrow/datafusion/SessionContext.java
@@ -3,8 +3,8 @@
 import java.nio.file.Path;
 import java.util.concurrent.CompletableFuture;
 
-/** An execution context holds resources and is the entrance for obtaining {@link DataFrame} */
-public interface ExecutionContext extends AutoCloseable, NativeProxy {
+/** A session context holds resources and is the entrance for obtaining {@link DataFrame} */
+public interface SessionContext extends AutoCloseable, NativeProxy {
 
   /** Obtain the {@link DataFrame} by running the {@code sql} against the datafusion library */
   CompletableFuture<DataFrame> sql(String sql);
diff --git a/datafusion-java/src/main/java/org/apache/arrow/datafusion/SessionContexts.java b/datafusion-java/src/main/java/org/apache/arrow/datafusion/SessionContexts.java
@@ -0,0 +1,19 @@
+package org.apache.arrow.datafusion;
+
+public class SessionContexts {
+
+  private SessionContexts() {}
+
+  static native long createSessionContext();
+
+  static native void destroySessionContext(long pointer);
+
+  static {
+    JNILoader.load();
+  }
+
+  public static SessionContext create() {
+    long pointer = createSessionContext();
+    return new DefaultSessionContext(pointer);
+  }
+}
diff --git a/datafusion-jni/Cargo.toml b/datafusion-jni/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "datafusion_jni"
-version = "0.7.1"
+version = "0.11.0"
 homepage = "https://github.com/apache/arrow-datafusion"
 repository = "https://github.com/apache/arrow-datafusion"
 authors = ["Apache Arrow <dev@arrow.apache.org>"]
@@ -12,8 +12,8 @@ edition = "2021"
 [dependencies]
 jni = "^0.19.0"
 tokio = "^1.18.0"
-arrow = "^9.0"
-datafusion = "^7.0"
+arrow = "^13.0"
+datafusion = "^8.0"
 
 [lib]
 crate_type = ["cdylib"]
diff --git a/datafusion-jni/src/context.rs b/datafusion-jni/src/context.rs
@@ -1,12 +1,12 @@
-use datafusion::execution::context::ExecutionContext;
-use datafusion::prelude::CsvReadOptions;
+use datafusion::execution::context::SessionContext;
+use datafusion::prelude::{CsvReadOptions, ParquetReadOptions};
 use jni::objects::{JClass, JObject, JString, JValue};
 use jni::sys::jlong;
 use jni::JNIEnv;
 use tokio::runtime::Runtime;
 
 #[no_mangle]
-pub extern "system" fn Java_org_apache_arrow_datafusion_DefaultExecutionContext_registerCsv(
+pub extern "system" fn Java_org_apache_arrow_datafusion_DefaultSessionContext_registerCsv(
     env: JNIEnv,
     _class: JClass,
     runtime: jlong,
@@ -24,7 +24,7 @@ pub extern "system" fn Java_org_apache_arrow_datafusion_DefaultExecutionContext_
         .get_string(path)
         .expect("Couldn't get name as string!")
         .into();
-    let context = unsafe { &mut *(pointer as *mut ExecutionContext) };
+    let context = unsafe { &mut *(pointer as *mut SessionContext) };
     runtime.block_on(async {
         let register_result = context
             .register_csv(&name, &path, CsvReadOptions::new())
@@ -44,7 +44,7 @@ pub extern "system" fn Java_org_apache_arrow_datafusion_DefaultExecutionContext_
 }
 
 #[no_mangle]
-pub extern "system" fn Java_org_apache_arrow_datafusion_DefaultExecutionContext_registerParquet(
+pub extern "system" fn Java_org_apache_arrow_datafusion_DefaultSessionContext_registerParquet(
     env: JNIEnv,
     _class: JClass,
     runtime: jlong,
@@ -62,9 +62,11 @@ pub extern "system" fn Java_org_apache_arrow_datafusion_DefaultExecutionContext_
         .get_string(path)
         .expect("Couldn't get name as string!")
         .into();
-    let context = unsafe { &mut *(pointer as *mut ExecutionContext) };
+    let context = unsafe { &mut *(pointer as *mut SessionContext) };
     runtime.block_on(async {
-        let register_result = context.register_parquet(&name, &path).await;
+        let register_result = context
+            .register_parquet(&name, &path, ParquetReadOptions::default())
+            .await;
         let err_message: JValue = match register_result {
             Ok(_) => JValue::Void,
             Err(err) => {
@@ -80,7 +82,7 @@ pub extern "system" fn Java_org_apache_arrow_datafusion_DefaultExecutionContext_
 }
 
 #[no_mangle]
-pub extern "system" fn Java_org_apache_arrow_datafusion_DefaultExecutionContext_querySql(
+pub extern "system" fn Java_org_apache_arrow_datafusion_DefaultSessionContext_querySql(
     env: JNIEnv,
     _class: JClass,
     runtime: jlong,
@@ -93,7 +95,7 @@ pub extern "system" fn Java_org_apache_arrow_datafusion_DefaultExecutionContext_
         .get_string(sql)
         .expect("Couldn't get sql as string!")
         .into();
-    let context = unsafe { &mut *(pointer as *mut ExecutionContext) };
+    let context = unsafe { &mut *(pointer as *mut SessionContext) };
     runtime.block_on(async {
         let query_result = context.sql(&sql).await;
         match query_result {
@@ -123,19 +125,19 @@ pub extern "system" fn Java_org_apache_arrow_datafusion_DefaultExecutionContext_
     });
 }
 #[no_mangle]
-pub extern "system" fn Java_org_apache_arrow_datafusion_ExecutionContexts_destroyExecutionContext(
+pub extern "system" fn Java_org_apache_arrow_datafusion_SessionContexts_destroySessionContext(
     _env: JNIEnv,
     _class: JClass,
     pointer: jlong,
 ) {
-    let _ = unsafe { Box::from_raw(pointer as *mut ExecutionContext) };
+    let _ = unsafe { Box::from_raw(pointer as *mut SessionContext) };
 }
 
 #[no_mangle]
-pub extern "system" fn Java_org_apache_arrow_datafusion_ExecutionContexts_createExecutionContext(
+pub extern "system" fn Java_org_apache_arrow_datafusion_SessionContexts_createSessionContext(
     _env: JNIEnv,
     _class: JClass,
 ) -> jlong {
-    let context = ExecutionContext::new();
+    let context = SessionContext::new();
     Box::into_raw(Box::new(context)) as jlong
 }
diff --git a/datafusion-jni/src/dataframe.rs b/datafusion-jni/src/dataframe.rs
@@ -18,7 +18,7 @@ pub extern "system" fn Java_org_apache_arrow_datafusion_DataFrames_collectDatafr
     callback: JObject,
 ) {
     let runtime = unsafe { &mut *(runtime as *mut Runtime) };
-    let dataframe = unsafe { &mut *(dataframe as *mut Arc<dyn DataFrame>) };
+    let dataframe = unsafe { &mut *(dataframe as *mut Arc<DataFrame>) };
     let schema = dataframe.schema().into();
     runtime.block_on(async {
         let batches = dataframe
@@ -59,7 +59,7 @@ pub extern "system" fn Java_org_apache_arrow_datafusion_DataFrames_showDataframe
     callback: JObject,
 ) {
     let runtime = unsafe { &mut *(runtime as *mut Runtime) };
-    let dataframe = unsafe { &mut *(dataframe as *mut Arc<dyn DataFrame>) };
+    let dataframe = unsafe { &mut *(dataframe as *mut Arc<DataFrame>) };
     runtime.block_on(async {
         let r = dataframe.show().await;
         let err_message: JValue = match r {
@@ -87,5 +87,5 @@ pub extern "system" fn Java_org_apache_arrow_datafusion_DataFrames_destroyDataFr
     _class: JClass,
     pointer: jlong,
 ) {
-    let _ = unsafe { Box::from_raw(pointer as *mut Arc<dyn DataFrame>) };
+    let _ = unsafe { Box::from_raw(pointer as *mut Arc<DataFrame>) };
 }
