Vortex on Ice

package com.example.geo;

public class GeoLocation {
	public static native String formatLocation(
            long latitude,
            long longitude);
}

#include <jni.h>

JNIEXPORT jstring JNICALL Java_com_example_geo_GeoLocation_formatLocation
  (JNIEnv *, jclass, jlong, jlong);

use jni::objects::{JClass, JString};
use jni::sys::{jlong, jstring};
use jni::JNIEnv;

// `no_mangle` ensures that the compiler leaves the symbols name intact.
#[unsafe(no_mangle)]
pub extern "system" fn Java_com_example_geo_GeoLocator_getAddress(
    env: JNIEnv,
    _class: JClass,
    lat: jlong,
    lon: jlong,
) -> jstring {
    let rust_string = format!("({lat}, {lon})", lat, lon);
    let output = env
          .new_string(rust_string)
          .expect("Couldn't create java string");
    output.into_raw()
}

[lib]
crate-type = ["cdylib"]

public class NativeFile {
    private final URL url;

	public NativeFile(URL url) {
		this.url = url;
	}

	public void open() throws IOException {
		// Open a connection to Azure Data Lake Store using our URL
		// and properties.
	}

	// Create a new file scan, pushing the given Iceberg
    // predicate down to Vortex.
	public Scan newScanner(Expression predicate) {
		// ?
	}
}

// A Scan is just an iterator that yields Arrow columnar batches
public class Scan implements Iterator<ArrowColumnVector> {
	public Scan(ArrayIterator iterator) {
		this.iterator = iterator
	}

	@Override
	public void hasNext() {
		return iterator.hasNext();
	}

	@Override
	public ArrowColumnVector next() {
		return iterator.next();
	}
}

public final class NativeFileMethods {
    // This class only defines static methods, and should never
    // be instantiated by a call to `new`. Making its constructor
    // private prevents it from accidentally being used in ways
    // we didn't intend.
    private NativeFileMethods() {}

    /**
     * Open a file using the native library with the provided URI
     * and options.
     *
     * @param uri     The URI of the file to open.
     *                e.g. "file://path/to/file".
     * @param options A map of options to provide for opening
     *                the file.
     * @return A native pointer to the opened file.
     *         This will be 0 if the open call failed.
     */
    public static native long open(
        String uri,
        Map<String, String> options
    );
}

/// Native implementation for dev.vortex.jni.NativeFileMethods.open()
#[unsafe(no_mangle)]
pub extern "system" fn Java_dev_vortex_jni_NativeFileMethods_open(
    mut env: JNIEnv,
    _class: JClass,
    uri: JString,
    options: JObject,
) -> jlong {
    // try_or_throw is a helper that runs a closure that returns a Result,
	// and if it returns an error it throws it as a RuntimeExcpetion
	// back to the JVM.
    try_or_throw(&mut env, |env| {
		// Copy the string from the Java heap to Rust heap
        let file_path: String = env.get_string(&uri)?.into();

        let Ok(url) = Url::parse(&file_path) else {
            throw_runtime!("Invalid URL: {file_path}");
        };

		// Some helper to make the ObjectStore for the URL
        let store = make_object_store(&url)?;

        // block_on is a helper to execute a future in a blocking fashion
        // on the current thread.
        let open_file = block_on(
            "VortexOpenOptions.open()",
            VortexOpenOptions::file().open_object_store(
                &store,
                url.path()
            ),
        )?;

		// Put the VortexFile onto the native heap, and return
		// the raw pointer to the caller as a Java long.
        Ok(Box::new(open_file).into_raw())
    })
}

    /**
     * Close the file associated with the given pointer.
     *
     * @param pointer The native pointer to a file.
     */
    public static native void close(long pointer);

#[unsafe(no_mangle)]
pub extern "system" fn Java_dev_vortex_jni_NativeFileMethods_close(
    _env: JNIEnv,
    _class: JClass,
    pointer: jlong,
) {
	// SAFETY: the pointer MUST be a value returned from a previous
	// call to NativeFileMethods.open() and be non-zero.
    drop(unsafe { Box::from_raw(pointer as *mut VortexFile) });
}

public final class JNIFile {
	private OptionalLong pointer;

	public JNIFile(String uri) {
		this.pointer = OptionalLong.of(NativeFileMethods.open(uri));
	}
}

public final class JNIFile implements AutoCloseable {
	private OptionalLong pointer;

	public JNIFile(String uri) {
		this.pointer = OptionalLong.of(NativeFileMethods.open(uri));
	}

	public long getRowCount() {
		// pointer.getAsLong() will throw NoSuchElementException if
		// this method is called after close was called.
		return NativeFileMethods.rowCount(pointer.getAsLong());
	}

	@Override
	public void close() {
		// This will throw an exception if close has
        // already been called.
		NativeFileMethods.close(pointer.getAsLong());

		// Clear the pointer handle, preventing
        // access to the freed memory.
		pointer = OptionalLong.empty();
	}
}

java.base@17.0.14/sun.nio.ch.FileDispatcherImpl.write0(Native Method)
java.base@17.0.14/sun.nio.ch.FileDispatcherImpl.write(FileDispatcherImpl.java:62)
java.base@17.0.14/sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:132)
java.base@17.0.14/sun.nio.ch.IOUtil.write(IOUtil.java:76)
java.base@17.0.14/sun.nio.ch.IOUtil.write(IOUtil.java:67)
java.base@17.0.14/sun.nio.ch.FileChannelImpl.write(FileChannelImpl.java:288) => holding Monitor(java.lang.Object@439399089)
java.base@17.0.14/sun.nio.ch.FileChannelImpl.transferToTrustedChannel(FileChannelImpl.java:605)
java.base@17.0.14/sun.nio.ch.FileChannelImpl.transferTo(FileChannelImpl.java:699)
app//org.apache.spark.util.Utils$.copyFileStreamNIO(Utils.scala:346)
app//org.apache.spark.util.Utils.copyFileStreamNIO(Utils.scala)
app//org.apache.spark.shuffle.sort.UnsafeShuffleWriter.mergeSpillsWithTransferTo(UnsafeShuffleWriter.java:465)
app//org.apache.spark.shuffle.sort.UnsafeShuffleWriter.mergeSpillsUsingStandardWriter(UnsafeShuffleWriter.java:321)
app//org.apache.spark.shuffle.sort.UnsafeShuffleWriter.mergeSpills(UnsafeShuffleWriter.java:288)
app//org.apache.spark.shuffle.sort.UnsafeShuffleWriter.closeAndWriteOutput(UnsafeShuffleWriter.java:224)
app//org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:182)
app//org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
app//org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:104)
app//org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:54)
app//org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:166)
app//org.apache.spark.scheduler.Task.run(Task.scala:141)
app//org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:620)
app//org.apache.spark.executor.Executor$TaskRunner$$Lambda$2794/0x000072c48cea47e0.apply(Unknown Source)
app//org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64)
app//org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61)
app//org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94)
app//org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:623)
java.base@17.0.14/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
java.base@17.0.14/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
java.base@17.0.14/java.lang.Thread.run(Thread.java:840)