{@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/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");
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 ----
/**