feat(jni): bridge caller-provided Java I/O into VortexReadAt/VortexWrite

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

@@ -5,6 +5,7 @@
 
 import com.google.common.base.Preconditions;
 import dev.vortex.VortexCleaner;
+import dev.vortex.io.NativeReadable;
 import dev.vortex.jni.NativeDataSource;
 import java.util.Arrays;
 import java.util.Collections;
@@ -71,6 +72,51 @@ public static DataSource open(Session session, List<String> uris, Map<String, St
         return new DataSource(session, pointer);
     }
 
+    /** Open a single file through a caller-provided byte source. See {@link #open(Session, List, int)}. */
+    public static DataSource open(Session session, NativeReadable readable) {
+        return open(session, List.of(readable), 0);
+    }
+
+    /** Open files through caller-provided byte sources with the default read concurrency. */
+    public static DataSource open(Session session, List<NativeReadable> readables) {
+        return open(session, readables, 0);
+    }
+
+    /**
+     * Open one or more files through caller-provided byte sources instead of native storage clients. Every read the
+     * scan performs becomes an upcall into the corresponding {@link NativeReadable}, so this is the integration point
+     * for external I/O abstractions (for example Iceberg's {@code FileIO}). Each file is identified by
+     * {@link NativeReadable#name()}, which must be unique within {@code readables}.
+     *
+     * <p>The readables must remain open until this data source and all scans created from it are closed; native code
+     * never closes them.
+     *
+     * @param session open session
+     * @param readables byte sources
+     * @param readConcurrency maximum in-flight {@code readFully} calls across all files of this data source; {@code 0}
+     *     selects the default. Each in-flight read occupies one native thread and typically one stream on its readable.
+     */
+    public static DataSource open(Session session, List<NativeReadable> readables, int readConcurrency) {
+        Objects.requireNonNull(session, "session");
+        Objects.requireNonNull(readables, "readables");
+        Preconditions.checkArgument(!readables.isEmpty(), "at least one readable is required");
+        Preconditions.checkArgument(readConcurrency >= 0, "readConcurrency must not be negative");
+        Object[] readableArray = readables.toArray();
+        String[] names = new String[readableArray.length];
+        long[] lengths = new long[readableArray.length];
+        for (int i = 0; i < readableArray.length; i++) {
+            Preconditions.checkArgument(readableArray[i] != null, "readables must not contain null values");
+            NativeReadable readable = readables.get(i);
+            names[i] = readable.name();
+            Preconditions.checkArgument(names[i] != null, "readable at index %s returned a null name", i);
+            lengths[i] = readable.length();
+            Preconditions.checkArgument(lengths[i] >= 0, "readable for %s reported negative length", names[i]);
+        }
+        long pointer =
+                NativeDataSource.openFiles(session.nativePointer(), readableArray, names, lengths, readConcurrency);
+        return new DataSource(session, pointer);
+    }
+
     /** Arrow schema of the data source (and of scans produced from it). */
     public Schema arrowSchema(BufferAllocator allocator) {
         try (ArrowSchema schema = ArrowSchema.allocateNew(allocator)) {

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

@@ -5,6 +5,7 @@
 
 import com.google.common.base.Preconditions;
 import dev.vortex.VortexCleaner;
+import dev.vortex.io.NativeWritable;
 import dev.vortex.jni.NativeWriter;
 import java.io.IOException;
 import java.util.Map;
@@ -66,6 +67,33 @@ public static VortexWriter create(
         }
     }
 
+    /**
+     * Create a writer that streams the file into a caller-provided byte sink instead of a native storage client. This
+     * is the integration point for external I/O abstractions (for example Iceberg's {@code PositionOutputStream}).
+     *
+     * <p>The native side writes and flushes the sink but never closes it: after {@link #close()} returns, all bytes
+     * have been written and flushed, and the caller must close the sink to finalize the file.
+     */
+    public static VortexWriter create(
+            Session session, NativeWritable writable, Schema arrowSchema, BufferAllocator allocator)
+            throws IOException {
+        Objects.requireNonNull(session, "session");
+        Objects.requireNonNull(writable, "writable");
+        Objects.requireNonNull(arrowSchema, "arrowSchema");
+        Objects.requireNonNull(allocator, "allocator");
+        ArrowSchema ffi = ArrowSchema.allocateNew(allocator);
+        try {
+            Data.exportSchema(allocator, arrowSchema, null, ffi);
+            long ptr = NativeWriter.createStream(session.nativePointer(), writable, ffi.memoryAddress());
+            if (ptr <= 0) {
+                throw new IOException("failed to create stream writer (ptr=" + ptr + ")");
+            }
+            return new VortexWriter(ptr);
+        } finally {
+            ffi.close();
+        }
+    }
+
     /** Write a batch directly from Arrow C Data Interface addresses. */
     public void writeBatch(long arrowArrayAddr, long arrowSchemaAddr) throws IOException {
         Preconditions.checkState(!closed.get(), "writer already closed");

java/vortex-jni/src/main/java/dev/vortex/io/NativeReadable.java

@@ -0,0 +1,47 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+package dev.vortex.io;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * A random-access byte source supplied by the caller and read from native code.
+ *
+ * <p>Implementations bridge external I/O abstractions — for example Iceberg's {@code FileIO} input streams — into the
+ * Vortex native reader. The native side invokes {@link #readFully(long, ByteBuffer)} from its own worker threads,
+ * potentially many calls concurrently, so implementations must be safe for concurrent positional reads (typically by
+ * opening one underlying stream per concurrent call, or delegating to a positional-read API).
+ *
+ * <p>Lifecycle: the creator owns this object. Native code never calls {@link #close()}; close it only after every data
+ * source or scan built on top of it has been closed.
+ */
+public interface NativeReadable extends Closeable {
+    /**
+     * Unique name of this source, typically its original location (URI or file path). Used to key per-session caches
+     * and in native error messages, so it must be stable and unique across sources; it must not contain glob characters
+     * ({@code *?[}).
+     */
+    String name();
+
+    /** Total length of the source in bytes. Must be cheap; called once when the source is registered. */
+    long length();
+
+    /**
+     * Read exactly {@code buffer.remaining()} bytes starting at absolute {@code position} into {@code buffer}.
+     *
+     * <p>Unlike {@link java.io.InputStream#read}, short reads are not permitted: implementations must either fill the
+     * buffer completely ({@code buffer.remaining() == 0} on return) or throw.
+     *
+     * <p>{@code buffer} is a <em>direct</em> {@link ByteBuffer} wrapping native memory owned by the caller, so bytes
+     * written to it land in the native reader without an extra copy. It is valid only for the duration of this call:
+     * implementations must not retain a reference to it after returning. Implementations backed by {@code byte[]} APIs
+     * can bulk-{@link ByteBuffer#put(byte[], int, int) put} into it.
+     *
+     * @throws java.io.EOFException if the range extends past the end of the source
+     * @throws IOException if the underlying storage fails
+     */
+    void readFully(long position, ByteBuffer buffer) throws IOException;
+}

java/vortex-jni/src/main/java/dev/vortex/io/NativeWritable.java

@@ -0,0 +1,26 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+package dev.vortex.io;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ * A sequential byte sink supplied by the caller and written to from native code.
+ *
+ * <p>Implementations bridge external output abstractions — for example Iceberg's {@code PositionOutputStream} — into
+ * the Vortex native writer. Writes are sequential and single-threaded: the native side never issues concurrent
+ * {@code write} calls against the same instance.
+ *
+ * <p>Lifecycle: the creator owns this object. Native code calls {@link #write} and {@link #flush} but never
+ * {@link #close()}; after {@link dev.vortex.api.VortexWriter#close()} returns, all bytes have been written and flushed,
+ * and the creator must close the underlying stream to finalize it.
+ */
+public interface NativeWritable extends Closeable {
+    /** Append {@code length} bytes from {@code buffer} starting at {@code offset}. */
+    void write(byte[] buffer, int offset, int length) throws IOException;
+
+    /** Flush buffered bytes to the underlying storage. */
+    void flush() throws IOException;
+}

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

@@ -22,6 +22,21 @@ private NativeDataSource() {}
      */
     public static native long open(long sessionPointer, String[] uris, Map<String, String> options);
 
+    /**
+     * Open a data source over caller-provided {@link dev.vortex.io.NativeReadable} objects. All reads become upcalls
+     * into the supplied readables; no native storage client is created.
+     *
+     * @param sessionPointer pointer from {@link NativeSession#newSession()}
+     * @param readables one {@link dev.vortex.io.NativeReadable} per file
+     * @param names unique name per file (from {@link dev.vortex.io.NativeReadable#name()}), parallel to
+     *     {@code readables}
+     * @param lengths file sizes in bytes, parallel to {@code readables}
+     * @param readConcurrency maximum in-flight {@code readFully} upcalls across all files of this data source;
+     *     {@code <= 0} selects the default
+     */
+    public static native long openFiles(
+            long sessionPointer, Object[] readables, String[] names, long[] lengths, int readConcurrency);
+
     /** Free a data source pointer. */
     public static native void free(long pointer);
 

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

@@ -18,6 +18,12 @@ private NativeWriter() {}
     public static native long create(
             long sessionPointer, String uri, long arrowSchemaAddress, Map<String, String> options);
 
+    /**
+     * Open a writer that streams the file into a caller-provided {@link dev.vortex.io.NativeWritable}. The native side
+     * writes and flushes but never closes the writable; the caller must close it after {@link #close}.
+     */
+    public static native long createStream(long sessionPointer, Object writable, long arrowSchemaAddress);
+
     /**
      * Write a batch directly from Arrow C Data Interface addresses.
      *