{@code
+ * try (BufferAllocator allocator = new RootAllocator();
+ * ArrowReader reader = Arrow.query(allocator, conn, "MATCH (p:Person) RETURN p.id AS id")) {
+ * while (reader.loadNextBatch()) {
+ * BigIntVector ids = (BigIntVector) reader.getVectorSchemaRoot().getVector(0);
+ * for (int i = 0; i < ids.getValueCount(); i++) {
+ * sum += ids.get(i);
+ * }
+ * }
+ * }
+ * }
+ *
+ * Nothing on this path is proportional to the answer. The arrays that cross + * are the buffers the engine's executor filled, at the addresses it filled + * them at, and what an export costs is the schema, the stream and the pointers + * in it. A million rows and ten thousand cost about the same. + * + *
That is also why an export spends the result. Once the buffers have left, + * there is nothing on this side to read a second time, so the {@code Result} + * handed to any of these is closed by the call and every buffer a columnar + * reader borrowed from it before now belongs to the Arrow consumer. Closing it + * again afterwards is the no-op it always was, so a try-with-resources around + * it is still the right shape. + * + *
The reader owns what it was given and releases the stream when it closes, + * which releases the result the stream was made from. Close the reader. + * + *
A result the engine had to build across its rows, which is anything with + * an {@code ORDER BY}, has no buffers to move and is read into buffers of its + * own on the way out. That is the fallback working rather than the fast path + * failing, and it is still one pass and still correct. + */ +public final class Arrow { + + private Arrow() {} + + /** + * Runs a statement and hands back its answer as Arrow. + * + * @param allocator what the Arrow side allocates from + * @param conn the connection + * @param statement the text + * @return the reader, which the caller closes + */ + public static ArrowReader query(BufferAllocator allocator, Connection conn, String statement) { + Objects.requireNonNull(conn, "conn"); + Result result = conn.query(statement); + try { + return reader(allocator, result); + } catch (RuntimeException | Error e) { + result.close(); + throw e; + } + } + + /** + * A result already in hand, as Arrow, in batches of {@link + * Result#DEFAULT_BATCH} rows. + * + * @param allocator what the Arrow side allocates from + * @param result the result, which this call spends + * @return the reader, which the caller closes + */ + public static ArrowReader reader(BufferAllocator allocator, Result result) { + return reader(allocator, result, 0); + } + + /** + * The same, with the batch size named. + * + * @param allocator what the Arrow side allocates from + * @param result the result, which this call spends + * @param rowsPerBatch how many rows a consumer sees at a time, or zero for + * {@link Result#DEFAULT_BATCH}. The batches are slices of arrays that + * are already in memory, so this is about what a consumer likes to work + * in and not about what gets allocated + * @return the reader, which the caller closes + */ + public static ArrowReader reader(BufferAllocator allocator, Result result, long rowsPerBatch) { + Objects.requireNonNull(allocator, "allocator"); + Objects.requireNonNull(result, "result"); + ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator); + try { + result.exportArrow(stream.memoryAddress(), rowsPerBatch); + return Data.importArrayStream(allocator, stream); + } catch (RuntimeException | Error e) { + // A refusal leaves the struct as it was allocated, which is + // released, so this frees the memory it sits in and calls nothing. + // An import that failed leaves a live stream, and this is what + // releases it. + stream.close(); + throw e; + } + } +} diff --git a/zudb-arrow/src/main/java/dev/zudb/arrow/package-info.java b/zudb-arrow/src/main/java/dev/zudb/arrow/package-info.java new file mode 100644 index 0000000..f489236 --- /dev/null +++ b/zudb-arrow/src/main/java/dev/zudb/arrow/package-info.java @@ -0,0 +1,14 @@ +/** + * A zu result as Arrow, over the C Data Interface. + * + *
One class, {@link dev.zudb.arrow.Arrow}, and three static methods on it. + * Everything else a program needs on this path is arrow-java's own, because + * what comes back is an {@link org.apache.arrow.vector.ipc.ArrowReader} and + * every Arrow consumer on the JVM already takes one. + * + *
This lives in an artifact of its own so that the client keeps its + * dependencies at none. A program reading rows or columns has no reason to + * carry arrow-java, and a program that wants Arrow adds one line to a build + * file. + */ +package dev.zudb.arrow; diff --git a/zudb-arrow/src/main/java/module-info.java b/zudb-arrow/src/main/java/module-info.java new file mode 100644 index 0000000..5597237 --- /dev/null +++ b/zudb-arrow/src/main/java/module-info.java @@ -0,0 +1,21 @@ +/** + * A zu result as an Arrow reader. + * + *
This module is where arrow-java is named and the only place in this + * client that names it. A program that reads rows or columns depends on {@code + * dev.zudb} and carries nothing of Arrow; a program that wants Arrow adds this + * and gets the reader every Arrow consumer on the JVM already takes. + */ +module dev.zudb.arrow { + // Transitive, all three of them, because they are the types on the + // three methods this module has: a caller passes an allocator and a + // result and is handed a reader, so a caller that reads this module + // reads those as well or cannot call it at all. + requires transitive dev.zudb; + requires transitive org.apache.arrow.memory.core; + requires transitive org.apache.arrow.vector; + + requires org.apache.arrow.c; + + exports dev.zudb.arrow; +} diff --git a/zudb-arrow/src/test/java/dev/zudb/arrow/ArrowTest.java b/zudb-arrow/src/test/java/dev/zudb/arrow/ArrowTest.java new file mode 100644 index 0000000..7746652 --- /dev/null +++ b/zudb-arrow/src/test/java/dev/zudb/arrow/ArrowTest.java @@ -0,0 +1,329 @@ +package dev.zudb.arrow; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import dev.zudb.Connection; +import dev.zudb.Database; +import dev.zudb.Loader; +import dev.zudb.Result; +import dev.zudb.Statement; +import dev.zudb.Value; +import dev.zudb.ZuClosedException; +import dev.zudb.ZuException; +import dev.zudb.ZuProgrammingException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.BigIntVector; +import org.apache.arrow.vector.Float8Vector; +import org.apache.arrow.vector.VarCharVector; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.complex.StructVector; +import org.apache.arrow.vector.ipc.ArrowReader; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * A result crossing into Arrow, over the C Data Interface. + * + *
What these are for is that the values arrive intact, that the batching is + * what was asked for, and that the result is spent exactly when the engine + * spends it: after an export, whether the export worked or not, and not before + * a call this client refused on its own. + * + *
The allocator is closed after every test, which fails the test if a buffer
+ * was left behind, so leaking the stream or the reader is caught here rather
+ * than showing up as memory a long-running program never gets back.
+ */
+class ArrowTest {
+
+ @TempDir Path dir;
+
+ private BufferAllocator allocator;
+
+ @BeforeAll
+ static void engine() {
+ Libzu.require();
+ }
+
+ @BeforeEach
+ void allocator() {
+ allocator = new RootAllocator();
+ }
+
+ @AfterEach
+ void balanced() {
+ // Closing a RootAllocator with anything outstanding throws, so this is
+ // the assertion that every reader above released what it was handed.
+ allocator.close();
+ }
+
+ @Test
+ void aStoredColumnCrossesAsArrow() throws Exception {
+ Path path = dir.resolve("people.zu");
+ try (Loader loader = Loader.create(path)) {
+ loader.table("Person", "Knows", 3);
+ loader.column("id", 1L, 2L, 3L);
+ loader.column("name", "ada", "grace", "alan");
+ loader.finish();
+ }
+
+ // A projection of stored values with nothing above it, which is the plan
+ // whose columns the executor fills. The arrays that cross are those
+ // buffers, so nothing here is proportional to the row count.
+ try (Database db = Database.open(path);
+ Connection conn = db.connect();
+ ArrowReader reader =
+ Arrow.query(allocator, conn, "MATCH (p:Person) RETURN p.id AS id, p.name AS name")) {
+ VectorSchemaRoot root = reader.getVectorSchemaRoot();
+ assertEquals(List.of("id", "name"), root.getSchema().getFields().stream().map(f -> f.getName()).toList());
+
+ List The same twenty lines as the provider module's, and a copy rather than a
+ * shared artifact on purpose: a test fixture published so that two test suites
+ * can agree on where a file is would be a third artifact to version, and this
+ * is a question with one right answer that nobody is going to change twice.
+ *
+ * These tests link against a real engine, so they are skipped rather than
+ * failed when there is not one to link against. A checkout with no build of
+ * the engine beside it is an ordinary state for this repository to be in, and
+ * a red suite for it would train everybody to ignore a red suite.
+ *
+ * Point them at one with {@code -Dzu.library=/path/to/libzu.dylib}, or set
+ * {@code ZU_LIBRARY}. A sibling checkout of the engine with a release build in
+ * it is found on its own.
+ */
+final class Libzu {
+
+ private Libzu() {}
+
+ private static final Path FOUND = locate();
+
+ /** Skips the calling test when there is no engine to call. */
+ static void require() {
+ assumeTrue(FOUND != null, "no libzu: set -Dzu.library to run these");
+ if (System.getProperty("zu.library") == null) {
+ System.setProperty("zu.library", FOUND.toString());
+ }
+ }
+
+ private static Path locate() {
+ String named = System.getProperty("zu.library");
+ if (named == null || named.isBlank()) {
+ named = System.getenv("ZU_LIBRARY");
+ }
+ if (named != null && !named.isBlank()) {
+ Path p = Paths.get(named);
+ return Files.isRegularFile(p) ? p : null;
+ }
+ String name = System.mapLibraryName("zu");
+ Path here = Paths.get("").toAbsolutePath();
+ for (Path root = here; root != null; root = root.getParent()) {
+ for (String sibling : new String[] {"zu", "zu-dx", "zu-g0"}) {
+ Path candidate = root.resolveSibling(sibling).resolve("target/release").resolve(name);
+ if (Files.isRegularFile(candidate)) {
+ return candidate;
+ }
+ }
+ }
+ return null;
+ }
+}
diff --git a/zudb-bench/pom.xml b/zudb-bench/pom.xml
index 71197b2..8e5760a 100644
--- a/zudb-bench/pom.xml
+++ b/zudb-bench/pom.xml
@@ -35,12 +35,25 @@
Every benchmark below runs the statement as well, because an export + * spends its result and there is no way to export the same one twice. That + * makes these four comparable to each other and not to {@link ReadBench}, + * which reads a result it was handed and never runs anything. + * + *
The table is on disk and the statement is a scan of stored values with + * nothing above it, which is the plan whose columns the executor fills. That + * is the case worth measuring: the arrays that cross into Arrow are those + * buffers, so the export is a schema and a handful of pointers and does not + * grow with the row count. Summing through the reader afterwards is the JVM + * walking memory the engine wrote, which is what a real consumer does and + * what makes {@code exportAndSum} more than a pointer swap. + */ +@State(Scope.Benchmark) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Warmup(iterations = 3, time = 2) +@Measurement(iterations = 5, time = 2) +@Fork(value = 1, jvmArgs = {"--enable-native-access=ALL-UNNAMED", "--add-opens=java.base/java.nio=ALL-UNNAMED", "--sun-misc-unsafe-memory-access=allow"}) +public class ArrowBench { + + /** + * How many rows the table holds. A constant rather than a parameter because + * the per-row score is scaled by it, and JMH wants that scale as a literal + * in an annotation. + */ + private static final int ROWS = 100_000; + + private static final String SCAN = "MATCH (r:Row) RETURN r.id AS id"; + + private Path dir; + private Database db; + private Connection conn; + private BufferAllocator allocator; + + @Setup + public void build() throws IOException { + dir = Files.createTempDirectory("zu-arrow-bench"); + Path path = dir.resolve("rows.zu"); + try (Loader loader = Loader.create(path)) { + loader.table("Row", "Near", ROWS); + long[] ids = new long[ROWS]; + for (int i = 0; i < ROWS; i++) { + ids[i] = i; + } + loader.column("id", ids); + loader.finish(); + } + db = Database.open(path); + conn = db.connect(); + allocator = new RootAllocator(); + } + + @TearDown + public void close() throws IOException { + allocator.close(); + conn.close(); + db.close(); + Temp.deleteTree(dir); + } + + /** The statement and nothing else, so the three below split in two. */ + @Benchmark + @OperationsPerInvocation(ROWS) + public long queryOnly() { + try (Result r = conn.query(SCAN)) { + return r.rows(); + } + } + + /** The statement, the export, and a sum over every batch the reader gives back. */ + @Benchmark + @OperationsPerInvocation(ROWS) + public long exportAndSum() throws IOException { + long total = 0; + try (ArrowReader reader = Arrow.query(allocator, conn, SCAN)) { + while (reader.loadNextBatch()) { + BigIntVector id = (BigIntVector) reader.getVectorSchemaRoot().getVector(0); + for (int i = 0, n = id.getValueCount(); i < n; i++) { + total += id.get(i); + } + } + } + return total; + } + + /** The statement and the same sum over the borrowed column, which stays here. */ + @Benchmark + @OperationsPerInvocation(ROWS) + public long borrowAndSum() { + long total = 0; + try (Result r = conn.query(SCAN)) { + LongBuffer b = r.longs(0); + for (int i = 0, n = b.remaining(); i < n; i++) { + total += b.get(i); + } + } + return total; + } + + /** The statement and the same sum a row at a time, which is what it costs to not do either. */ + @Benchmark + @OperationsPerInvocation(ROWS) + public long rowAtATime() { + long total = 0; + try (Result r = conn.query(SCAN)) { + for (Row row : r) { + total += row.getLong(0); + } + } + return total; + } +} diff --git a/zudb-ffm/src/main/java/dev/zudb/ffm/Abi.java b/zudb-ffm/src/main/java/dev/zudb/ffm/Abi.java index 34950d4..c7467e0 100644 --- a/zudb-ffm/src/main/java/dev/zudb/ffm/Abi.java +++ b/zudb-ffm/src/main/java/dev/zudb/ffm/Abi.java @@ -113,6 +113,8 @@ final class Abi { final MethodHandle chunkColNodeOffset; final MethodHandle chunkColValid; + final MethodHandle resultArrow; + final MethodHandle loaderCreate; final MethodHandle loaderTable; final MethodHandle loaderEdges; @@ -279,6 +281,11 @@ final class Abi { "zu_result_chunk_col_valid", FunctionDescriptor.of(JAVA_INT, ADDRESS, JAVA_LONG, JAVA_INT, ADDRESS)); + resultArrow = + h( + "zu_result_arrow", + FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, JAVA_LONG, ADDRESS, ADDRESS)); + loaderCreate = h("zu_loader_create", FunctionDescriptor.of(JAVA_INT, ADDRESS, SIZE_T, ADDRESS, ADDRESS)); loaderTable = diff --git a/zudb-ffm/src/main/java/dev/zudb/ffm/FfmBinding.java b/zudb-ffm/src/main/java/dev/zudb/ffm/FfmBinding.java index d51ccb2..778cf5f 100644 --- a/zudb-ffm/src/main/java/dev/zudb/ffm/FfmBinding.java +++ b/zudb-ffm/src/main/java/dev/zudb/ffm/FfmBinding.java @@ -597,6 +597,32 @@ public ByteBuffer chunkValid(long result, long chunk, int col, long rows) { return p == 0 ? null : buffer(p, rows, 1); } + @Override + public void resultArrow(long conn, long result, long rowsPerBatch, long stream) { + Scratch s = Scratch.get(); + MemorySegment sl = s.slots(); + clear(sl); + // The engine takes the result through a pointer to it and writes null + // back, so the handle has to be somewhere it can write rather than in a + // register. The slot it lands in is read for nothing afterwards: the API + // module treats the result as gone whatever this answered, because the + // engine nulled it on the refusal path too. + sl.set(ADDRESS, OUT, ptr(result)); + try { + int st = + (int) + abi.resultArrow.invokeExact( + ptr(conn), + sl.asSlice(OUT, 8), + rowsPerBatch, + ptr(stream), + sl.asSlice(ERR, 8)); + check("zu_result_arrow", st, sl); + } catch (Throwable t) { + throw fail("zu_result_arrow", t); + } + } + @Override public long resultCell(long result, long row, int col) { Scratch s = Scratch.get(); diff --git a/zudb/src/main/java/dev/zudb/Connection.java b/zudb/src/main/java/dev/zudb/Connection.java index 0f4f14e..312c896 100644 --- a/zudb/src/main/java/dev/zudb/Connection.java +++ b/zudb/src/main/java/dev/zudb/Connection.java @@ -120,7 +120,7 @@ public static Connection memory() { * @return the result, which the caller closes */ public Result query(String statement) { - return new Result(zu, zu.query(open(), statement)); + return new Result(zu, zu.query(open(), statement), this); } /** @@ -144,7 +144,7 @@ public void execute(String statement) { * @return the statement, which the caller closes */ public Statement prepare(String statement) { - return new Statement(zu, zu.prepare(open(), statement)); + return new Statement(zu, zu.prepare(open(), statement), this); } /** @@ -435,4 +435,17 @@ private long open() { } return h; } + + /** + * The handle, or zero once this connection has closed, for the one call + * that has something to say either way. + * + *
Exporting a result to Arrow reads node table names out of the catalog + * this connection holds, and a result outlives its connection on purpose, + * so a connection that has gone is not a failure there. It costs the names, + * which the export says in its own documentation, and nothing else. + */ + long lend() { + return handle.get(); + } } diff --git a/zudb/src/main/java/dev/zudb/Result.java b/zudb/src/main/java/dev/zudb/Result.java index 4646701..9660cfa 100644 --- a/zudb/src/main/java/dev/zudb/Result.java +++ b/zudb/src/main/java/dev/zudb/Result.java @@ -25,12 +25,14 @@ * outlive is {@link #close()}, and that includes every buffer the columnar * readers handed back and every string that came out of a row. * - *
There are three ways to read it, in the order you reach for them. + *
There are four ways to read it, in the order you reach for them. * {@link #stream()} for a row at a time, which is what most code wants. * {@link #longs(int)} and the three beside it for a whole column, borrowed * from the engine rather than copied, which is what a million rows wants. * {@link #chunks()} for a whole column read a chunk at a time, which is what - * a million rows wants when you are not going to read all of them. + * a million rows wants when you are not going to read all of them. And + * {@link #exportArrow(long, long)} for handing the whole thing to something + * else entirely, which spends the result rather than reading it. * *
{@code
* try (Result r = conn.query("MATCH (p:Person) RETURN p.name AS name")) {
@@ -40,16 +42,24 @@
*/
public final class Result implements AutoCloseable, Iterable {
+ /**
+ * How many rows a consumer sees per Arrow batch when nobody names a number,
+ * which is what {@link #exportArrow(long)} asks the engine for.
+ */
+ public static final long DEFAULT_BATCH = 65536;
+
private final ZuBinding zu;
private final AtomicLong handle;
+ private final Connection conn;
private final long rows;
private final int columns;
private final List names;
private final Map byName;
- Result(ZuBinding zu, long handle) {
+ Result(ZuBinding zu, long handle, Connection conn) {
this.zu = zu;
this.handle = new AtomicLong(handle);
+ this.conn = conn;
this.rows = zu.resultRows(handle);
this.columns = zu.resultCols(handle);
List found = new ArrayList<>(columns);
@@ -332,12 +342,15 @@ public Chunk chunk(long index) {
/**
* Every chunk, in order.
*
- * Which of these to use is a question of size. A point read wants a
- * whole column, because the answer is small and one call beats a loop.
- * Every large answer wants chunks, because the whole-column call converts
- * all of it before returning any of it and keeps the conversion until the
- * result is freed: reading the first hundred rows of a million-row column
- * and stopping pays for the other 999,900.
+ *
Which of these to use is a question of size, and only for the columns
+ * the engine did not fill. On those, the whole-column call converts all of
+ * the column before returning any of it and keeps the conversion until the
+ * result is freed, so reading the first hundred rows of a million-row
+ * column and stopping pays for the other 999,900. On a column the engine
+ * filled, which is every plan whose projection is a scan of stored values,
+ * both calls are views of the buffer it wrote and neither converts
+ * anything, so the choice is about the shape of the reading loop and
+ * nothing else.
*
* @return the chunks
*/
@@ -346,6 +359,76 @@ public Stream chunks() {
return java.util.stream.LongStream.range(0, count).mapToObj(this::chunk);
}
+ // ---- arrow ----
+
+ /**
+ * Hands the whole result to an Arrow consumer through the C Data Interface
+ * and spends it, in batches of {@link #DEFAULT_BATCH} rows.
+ *
+ * @param stream the address of an {@code ArrowArrayStream} the caller owns
+ * and has not initialised
+ */
+ public void exportArrow(long stream) {
+ exportArrow(stream, 0);
+ }
+
+ /**
+ * Hands the whole result to an Arrow consumer through the C Data Interface
+ * and spends it.
+ *
+ * This is the low-level door, and most programs want {@code zudb-arrow}
+ * rather than this: that module wraps this call in the {@code ArrowReader}
+ * arrow-java already knows how to read, and it is a separate artifact so
+ * that a program with no use for Arrow does not carry the dependency. What
+ * is here is what that module needs and what a program with its own Arrow
+ * bindings can use instead.
+ *
+ *
The two arguments are checked here rather than by the engine, so a call
+ * refused for a stream that is nowhere or a batch of fewer than no rows
+ * handed nothing over and spent nothing: that result is still there to read
+ * the ordinary way, or to export once the argument is right.
+ *
+ *
Nothing on this path is a copy. The arrays that cross are the buffers
+ * the executor filled, at the addresses it filled them at, which is why the
+ * result is spent: after the buffers have left there is nothing here to
+ * read a second time. So this result is closed whatever the call answered,
+ * including a refusal, every buffer a columnar reader handed out before it
+ * now belongs to the Arrow consumer, and closing it again afterwards is the
+ * no-op it always was.
+ *
+ *
A node column names its table out of the catalog the connection holds.
+ * A connection that has already closed is not a failure here, and the
+ * export then names a table after its id, which is still an answer for a
+ * program that kept a result and let its connection go.
+ *
+ * @param stream the address of an {@code ArrowArrayStream} the caller owns
+ * and has not initialised, written only on success and released through
+ * its own release callback rather than by anything here
+ * @param rowsPerBatch how many rows a consumer sees at a time, or zero for
+ * {@link #DEFAULT_BATCH}. The batches are slices of arrays that are
+ * already in memory, so this is about what a consumer likes to work in
+ * and not about what gets allocated
+ * @throws ZuProgrammingException if the stream address is zero, the batch
+ * is negative, or a column holds something Arrow has no type for
+ * @throws ZuClosedException if this result is already closed
+ */
+ public void exportArrow(long stream, long rowsPerBatch) {
+ long h = open();
+ if (stream == 0) {
+ throw new ZuProgrammingException(
+ Diagnostic.misuse(Status.MISUSE, "the stream to export into is null"));
+ }
+ if (rowsPerBatch < 0) {
+ throw new ZuProgrammingException(
+ Diagnostic.misuse(Status.MISUSE, "a batch of fewer than no rows: " + rowsPerBatch));
+ }
+ // The handle goes before the call rather than after it. The engine nulls
+ // the result on every path it takes, refusals included, so a result this
+ // still thought it owned would be one a later close would free twice.
+ handle.set(0);
+ zu.resultArrow(conn == null ? 0 : conn.lend(), h, rowsPerBatch, stream);
+ }
+
/**
* Whether this result has been closed.
*
diff --git a/zudb/src/main/java/dev/zudb/Statement.java b/zudb/src/main/java/dev/zudb/Statement.java
index 2d47cda..d64628b 100644
--- a/zudb/src/main/java/dev/zudb/Statement.java
+++ b/zudb/src/main/java/dev/zudb/Statement.java
@@ -37,10 +37,12 @@ public final class Statement implements AutoCloseable {
private final ZuBinding zu;
private final AtomicLong handle;
+ private final Connection conn;
- Statement(ZuBinding zu, long handle) {
+ Statement(ZuBinding zu, long handle, Connection conn) {
this.zu = zu;
this.handle = new AtomicLong(handle);
+ this.conn = conn;
}
/**
@@ -241,7 +243,7 @@ public Statement bindNull(String name) {
* @return the result, which the caller closes
*/
public Result execute() {
- return new Result(zu, zu.execute(open()));
+ return new Result(zu, zu.execute(open()), conn);
}
/**
diff --git a/zudb/src/main/java/dev/zudb/spi/ZuBinding.java b/zudb/src/main/java/dev/zudb/spi/ZuBinding.java
index bb90b60..d08f3df 100644
--- a/zudb/src/main/java/dev/zudb/spi/ZuBinding.java
+++ b/zudb/src/main/java/dev/zudb/spi/ZuBinding.java
@@ -543,6 +543,32 @@ public interface ZuBinding {
*/
ByteBuffer chunkValid(long result, long chunk, int col, long rows);
+ // ---- arrow ----
+
+ /**
+ * Hands a whole result to an Arrow consumer through the C Data Interface,
+ * and spends it.
+ *
+ *
Nothing here is a copy. The arrays that cross are the buffers the
+ * engine's executor filled, at the addresses it filled them at, which is
+ * why the result is spent: once the buffers have left there is nothing on
+ * this side to read a second time. The engine writes null back through the
+ * handle on every path, including a refusal, so an implementation must
+ * treat the result as gone whatever this call answered and the API module
+ * must not free it afterwards.
+ *
+ * @param conn the connection the result was produced on, or zero. It is
+ * what a node column's table name is read out of, and a result whose
+ * connection has closed still exports, naming tables by their ids
+ * @param result the result, which is gone when this returns
+ * @param rowsPerBatch how many rows a consumer sees at a time, or zero for
+ * the engine's own
+ * @param stream the address of an {@code ArrowArrayStream} the caller owns
+ * and has not initialised, written only on success and released through
+ * its own release callback rather than by anything here
+ */
+ void resultArrow(long conn, long result, long rowsPerBatch, long stream);
+
// ---- values ----
/**
From 0caf1bcb8ff869a3d95129f60c9eb3e48abc60d2 Mon Sep 17 00:00:00 2001
From: Tam Nguyen Duc <1218621+tamnd@users.noreply.github.com>
Date: Thu, 20 Aug 2026 12:34:00 +0700
Subject: [PATCH 2/2] The ABI this client speaks is 0.12
zu_result_arrow is what the engine added the revision for, so a client
that calls it is a client written against 0.12 and should say so. CI
reads the macro out of the engine's own header and compares, which is
the step that has been red on main since the engine bumped.
---
zudb/src/main/java/dev/zudb/Zu.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/zudb/src/main/java/dev/zudb/Zu.java b/zudb/src/main/java/dev/zudb/Zu.java
index eced10c..13111e3 100644
--- a/zudb/src/main/java/dev/zudb/Zu.java
+++ b/zudb/src/main/java/dev/zudb/Zu.java
@@ -37,7 +37,7 @@ public final class Zu {
* has every symbol this client calls, which is the mismatch that actually
* bites, and it names the missing one.
*/
- public static final String ABI_VERSION = "0.11";
+ public static final String ABI_VERSION = "0.12";
private static final Logger LOG = System.getLogger("dev.zudb");