fix: FFI/Java bindings improvements (#2644)

Implement some small fixes for Java

- Handle dropping field DTypes when creating Struct DType, less
error-prone than relying on user to call free
- Expose nullability + null_count over FFI and in Java bindings

Author: a10y

java/vortex-jni/src/main/java/dev/vortex/api/Array.java

@@ -26,6 +26,8 @@ public interface Array extends AutoCloseable {
 
     boolean getNull(int index);
 
+    int getNullCount();
+
     byte getByte(int index);
 
     short getShort(int index);
@@ -43,4 +45,7 @@ public interface Array extends AutoCloseable {
     String getUTF8(int index);
 
     byte[] getBinary(int index);
+
+    @Override
+    void close();
 }

java/vortex-jni/src/main/java/dev/vortex/api/ArrayStream.java

@@ -26,4 +26,7 @@ public interface ArrayStream extends AutoCloseable {
      * It is an error to call this method if a previous invocation returned false.
      */
     boolean next();
+
+    @Override
+    void close();
 }

java/vortex-jni/src/main/java/dev/vortex/impl/NativeArray.java

@@ -84,8 +84,14 @@ public Array slice(int start, int stop) {
 
     @Override
     public boolean getNull(int index) {
-        // check validity of the array
-        return false;
+        checkNotNull(inner, "inner");
+        return FFI.FFIArray_is_null(inner, index);
+    }
+
+    @Override
+    public int getNullCount() {
+        checkNotNull(inner, "inner");
+        return FFI.FFIArray_null_count(inner);
     }
 
     @Override

java/vortex-jni/src/main/java/dev/vortex/jni/FFI.java

@@ -37,6 +37,10 @@ public final class FFI {
 
     public static native FFIArray FFIArray_slice(FFIArray array, int start, int stop);
 
+    public static native boolean FFIArray_is_null(FFIArray array, int index);
+
+    public static native int FFIArray_null_count(FFIArray array);
+
     public static native FFIArray FFIArray_get_field(FFIArray array, int index);
 
     public static native byte FFIArray_get_u8(FFIArray array, int index);

java/vortex-jni/src/test/java/dev/vortex/jni/FFITest.java

@@ -17,7 +17,6 @@
 
 import dev.vortex.api.Array;
 import dev.vortex.spark.SparkTypes;
-import java.io.IOException;
 import org.apache.spark.sql.types.Decimal;
 import org.apache.spark.sql.vectorized.ColumnVector;
 import org.apache.spark.sql.vectorized.ColumnarArray;
@@ -36,7 +35,7 @@ public VortexColumnVector(Array array) {
     public void close() {
         try {
             array.close();
-        } catch (IOException e) {
+        } catch (Exception e) {
             throw new RuntimeException("Failed to close Vortex Array", e);
         }
     }

java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexColumnVector.java

@@ -24,7 +24,6 @@
 import dev.vortex.api.ScanOptions;
 import dev.vortex.impl.NativeFile;
 import dev.vortex.spark.VortexFilePartition;
-import java.io.IOException;
 import java.util.Objects;
 import org.apache.spark.sql.connector.read.PartitionReader;
 import org.apache.spark.sql.vectorized.ColumnarBatch;
@@ -72,7 +71,7 @@ void initNativeResources() {
     }
 
     @Override
-    public void close() throws IOException {
+    public void close() {
         checkState(Objects.nonNull(file), "File was closed");
         checkState(Objects.nonNull(arrayStream), "ArrayStream was closed");
 

java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java

@@ -19,8 +19,10 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
+@Disabled
 final class VortexScanTest {
     private static final Path TPCH_ROOT = Paths.get("/Volumes/Code/vortex/bench-vortex/data/tpch/1/vortex_compressed");
 

java/vortex-spark/src/test/java/dev/vortex/spark/VortexScanTest.java

@@ -82,6 +82,27 @@ pub unsafe extern "C" fn FFIArray_slice(
     Box::into_raw(Box::new(FFIArray { inner: sliced }))
 }
 
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn FFIArray_is_null(array: *const FFIArray, index: u32) -> bool {
+    let array = &*array;
+    array
+        .inner
+        .is_invalid(index as usize)
+        .vortex_expect("FFIArray_is_null: is_invalid")
+}
+
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn FFIArray_null_count(array: *const FFIArray) -> u32 {
+    let array = &*array;
+    array
+        .inner
+        .as_ref()
+        .invalid_count()
+        .vortex_expect("FFIArray_null_count: invalid count")
+        .try_into()
+        .vortex_expect("FFIArray_null_count: invalid count to u32")
+}
+
 macro_rules! ffiarray_get_ptype {
     ($ptype:ident) => {
         paste::paste! {

vortex-ffi/src/array.rs

@@ -68,7 +68,8 @@ pub unsafe extern "C" fn DType_new_struct(
     for i in 0..len {
         let name_ptr = *names.add(i as usize);
         let name: Arc<str> = CStr::from_ptr(name_ptr).to_string_lossy().into();
-        let dtype = (**dtypes.add(i as usize)).clone();
+        let dtype = Box::from_raw(*dtypes.add(i as usize));
+        let dtype = *dtype;
 
         rust_names.push(name);
         field_dtypes.push(dtype);
@@ -301,8 +302,6 @@ mod tests {
             let dtypes = [name, age];
 
             let person = DType_new_struct(names.as_ptr(), dtypes.as_ptr(), 2, false);
-            DType_free(name);
-            DType_free(age);
 
             assert_eq!(DType_get(person), DTYPE_STRUCT);
             assert_eq!(DType_field_count(person), 2);