diff --git a/README.md b/README.md index 3b05be7..fa5af6f 100644 --- a/README.md +++ b/README.md @@ -68,7 +68,7 @@ A row at a time is a boundary crossing a cell, and a hundred crossings cost abou ## Getting rows in -Two ways, and which one you want follows from whether the database exists yet. +Two ways, and which one you want follows from whether the database exists yet. There is a third below for the rows that should not go in at all. A loader builds one out of whole columns. It is the fastest way values get in and, while the engine has no DDL, it is the only way a table comes into being at all: @@ -111,6 +111,45 @@ The first two lines are the copy: 0.68 ns a row over a hundred thousand rows is The last line is the one to read twice. A statement per row parses, plans, runs and commits per row, and none of that work says anything the row before it did not already say. That is what an appender is for. +## Querying memory you already hold + +The third way is not to get the rows in at all. A frame names columns your program is already holding as a table of one connection, and statements read your buffers where they lie: + +```java +LongBuffer ids = ByteBuffer.allocateDirect(3 * 8).order(nativeOrder()).asLongBuffer(); +ids.put(new long[] {1, 2, 3}).flip(); + +try (Frame people = Frame.of("Person", 3)) { + people.column("id", ids); + conn.register(people); + try (Result r = conn.query("MATCH (p:Person) RETURN sum(p.id) AS total")) { + System.out.println(r.row(0).getLong(0)); + } +} +``` + +Nothing is copied at registration and nothing is copied at read. A scan builds vectors pointing straight at your buffers wherever the layouts agree, and the layouts that agree are the ones Arrow and this engine both keep: 64-bit signed integers, doubles, one bit a row for a boolean, and characters end to end with offsets cutting them up. A narrower integer, an unsigned one, a single-precision float and Arrow's microseconds against the nanoseconds this engine keeps time in are widened a value at a time as a statement reaches them, so a frame of a hundred columns costs you the one the statement named. + +Every buffer has to be direct, and one that is not is refused rather than copied. Everywhere else in this API a heap buffer costs a memcpy and nothing else, because the call reads it and is finished. A frame keeps the pointer for as long as it is registered, so a copy here would mean the engine reading a copy of your data for the rest of the frame's life, which is a frame that is not a frame. + +The buffers stay yours and the library never writes one, but a direct buffer is looked after for you: the frame holds a reference to everything handed to it and lets go at the moment the engine says it has finished, which is after the last statement reading the frame ends and is neither the unregister that preceded it nor the close. So a buffer cannot be collected out from under a running statement. The optional release callback runs at that same moment, and it is there for the other kind of buffer, one over memory you allocated yourself or one a lock has to be taken to let go of. + +A frame is read only and has no edges. A statement that would write to one is refused, a name a stored table already holds is refused, a name another frame holds replaces that frame, and registering inside a transaction is refused because a table appearing halfway through one is not something the transaction could then be rolled back over. + +A million rows, three columns, the same laptop: + +| How | Per row | +|---|---| +| describing them and registering them | 0.51 ns | +| the same million rows through a loader, write included | 630 ns | +| `sum(p.id)` over a frame | 1.06 ns | +| `sum(p.id)` over the same numbers in a database | 1.14 ns | +| the same over a 32-bit column, which is widened as it is read | 1.32 ns | +| a string comparison over a frame | 2.26 ns | +| the same over a database | 3.89 ns | + +The first two lines are the whole point. Half a millisecond to make a million rows queryable against six hundred to write them down, and the three lines after that say the query is not paying for it afterwards. + ## How it binds The Foreign Function and Memory API is the primary path. The downcall handles are written by hand against `zu.h` rather than generated with `jextract`, because the C ABI here is around seventy functions with a stable shape, and a hand-written layer is where the interesting decisions live: which calls are `Linker.Option.critical` because they are short pure accessors, where the out-parameter scratch space comes from so that a query does not allocate, and how a `zu_error` becomes a typed Java exception exactly once. There is no native code in this repository beyond `libzu` itself. diff --git a/zudb-bench/src/main/java/dev/zudb/bench/FrameBench.java b/zudb-bench/src/main/java/dev/zudb/bench/FrameBench.java new file mode 100644 index 0000000..3b16bec --- /dev/null +++ b/zudb-bench/src/main/java/dev/zudb/bench/FrameBench.java @@ -0,0 +1,195 @@ +package dev.zudb.bench; + +import dev.zudb.Connection; +import dev.zudb.Database; +import dev.zudb.Frame; +import dev.zudb.Loader; +import dev.zudb.Result; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.IntBuffer; +import java.nio.LongBuffer; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OperationsPerInvocation; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Warmup; + +/** + * What it costs to query a million rows the program is already holding, + * against what the same million rows cost once they are in a database. + * + *

Two things are being measured here and they answer different questions. + * {@code register} is how long it takes to make a million rows queryable, + * which is the number to read against a load: a load writes a file and a frame + * writes nothing. The scans are what a statement costs afterwards, and the + * point of those is that a frame is not a slower way of reading, it is the + * same read against somebody else's memory. + */ +@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"}) +public class FrameBench { + + /** How many rows both sides carry. */ + private static final int ROWS = 1_000_000; + + private Path dir; + + private LongBuffer ids; + private IntBuffer narrow; + private IntBuffer offsets; + private ByteBuffer characters; + + private Database memory; + private Connection frames; + private Connection spare; + private Frame frame; + + private Database stored; + private Connection disk; + + @Setup + public void fill() throws IOException { + dir = Files.createTempDirectory("zu-frame-bench"); + + ids = direct(ROWS * Long.BYTES).asLongBuffer(); + narrow = direct(ROWS * Integer.BYTES).asIntBuffer(); + List names = new ArrayList<>(ROWS); + for (int i = 0; i < ROWS; i++) { + ids.put(i); + narrow.put(i); + names.add("n" + i); + } + ids.flip(); + narrow.flip(); + + byte[] all = String.join("", names).getBytes(StandardCharsets.UTF_8); + characters = direct(all.length); + characters.put(all).flip(); + offsets = direct((ROWS + 1) * Integer.BYTES).asIntBuffer(); + int at = 0; + offsets.put(0); + for (String name : names) { + at += name.length(); + offsets.put(at); + } + offsets.flip(); + + memory = Database.memory(); + frames = memory.connect(); + spare = memory.connect(); + frame = describe(); + frames.register(frame); + + Path path = dir.resolve("people.zu"); + try (Loader loader = Loader.create(path)) { + loader.table("Person", "Knows", ROWS); + loader.column("id", copy()); + loader.column("name", names); + loader.finish(); + } + stored = Database.open(path); + disk = stored.connect(); + } + + @TearDown + public void clean() throws IOException { + disk.close(); + stored.close(); + frames.close(); + spare.close(); + frame.close(); + memory.close(); + Temp.deleteTree(dir); + } + + /** + * Describing a million rows and naming them as a table, which is the whole + * of what a frame costs before a statement can read it. + */ + @Benchmark + @OperationsPerInvocation(ROWS) + public void register() { + try (Frame one = describe()) { + spare.register(one); + spare.unregister("Person"); + } + } + + /** A scan of the lane, which is the column the engine reads as it stands. */ + @Benchmark + @OperationsPerInvocation(ROWS) + public long scanFrame() { + return sum(frames, "MATCH (p:Person) RETURN sum(p.id)"); + } + + /** The same scan over the same numbers, once they are in a database. */ + @Benchmark + @OperationsPerInvocation(ROWS) + public long scanStored() { + return sum(disk, "MATCH (p:Person) RETURN sum(p.id)"); + } + + /** A scan of a 32-bit column, which is widened a value at a time as it is read. */ + @Benchmark + @OperationsPerInvocation(ROWS) + public long scanNarrowFrame() { + return sum(frames, "MATCH (p:Person) RETURN sum(p.small)"); + } + + /** A scan of the string column, whose characters never move either way. */ + @Benchmark + @OperationsPerInvocation(ROWS) + public long scanFrameStrings() { + return sum(frames, "MATCH (p:Person) WHERE p.name = 'n999999' RETURN count(p)"); + } + + /** The same, out of the database. */ + @Benchmark + @OperationsPerInvocation(ROWS) + public long scanStoredStrings() { + return sum(disk, "MATCH (p:Person) WHERE p.name = 'n999999' RETURN count(p)"); + } + + private Frame describe() { + Frame one = Frame.of("Person", ROWS); + one.column("id", ids.duplicate()); + one.column("small", narrow.duplicate()); + one.strings("name", offsets.duplicate(), characters.duplicate()); + return one; + } + + private static long sum(Connection conn, String statement) { + try (Result r = conn.query(statement)) { + return r.row(0).getLong(0); + } + } + + private long[] copy() { + long[] out = new long[ROWS]; + ids.duplicate().get(out); + return out; + } + + private static ByteBuffer direct(int bytes) { + return ByteBuffer.allocateDirect(bytes).order(ByteOrder.nativeOrder()); + } +} 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 7717441..d2e7e39 100644 --- a/zudb-ffm/src/main/java/dev/zudb/ffm/Abi.java +++ b/zudb-ffm/src/main/java/dev/zudb/ffm/Abi.java @@ -135,6 +135,18 @@ final class Abi { final MethodHandle appenderClose; final MethodHandle appenderFree; + final MethodHandle frameNew; + final MethodHandle frameColInt; + final MethodHandle frameColFloat; + final MethodHandle frameColBool; + final MethodHandle frameColStr; + final MethodHandle frameColView; + final MethodHandle frameFree; + final MethodHandle connRegister; + final MethodHandle connUnregister; + final MethodHandle connRegisteredCount; + final MethodHandle connRegisteredName; + final MethodHandle valueType; final MethodHandle valueBool; final MethodHandle valueI64; @@ -316,6 +328,74 @@ final class Abi { h("zu_appender_close", FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, ADDRESS)); appenderFree = h("zu_appender_free", FunctionDescriptor.ofVoid(ADDRESS)); + frameNew = + h( + "zu_frame_new", + FunctionDescriptor.of( + JAVA_INT, ADDRESS, SIZE_T, JAVA_LONG, ADDRESS, ADDRESS, ADDRESS, ADDRESS)); + frameColInt = + h( + "zu_frame_col_int", + FunctionDescriptor.of( + JAVA_INT, + ADDRESS, + ADDRESS, + SIZE_T, + ADDRESS, + JAVA_LONG, + JAVA_INT, + JAVA_INT, + JAVA_LONG, + JAVA_INT, + ADDRESS)); + frameColFloat = + h( + "zu_frame_col_float", + FunctionDescriptor.of( + JAVA_INT, ADDRESS, ADDRESS, SIZE_T, ADDRESS, JAVA_LONG, JAVA_INT, ADDRESS)); + frameColBool = + h( + "zu_frame_col_bool", + FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, SIZE_T, ADDRESS, JAVA_LONG, ADDRESS)); + frameColStr = + h( + "zu_frame_col_str", + FunctionDescriptor.of( + JAVA_INT, + ADDRESS, + ADDRESS, + SIZE_T, + ADDRESS, + JAVA_INT, + ADDRESS, + SIZE_T, + JAVA_LONG, + ADDRESS)); + frameColView = + h( + "zu_frame_col_view", + FunctionDescriptor.of( + JAVA_INT, + ADDRESS, + ADDRESS, + SIZE_T, + ADDRESS, + ADDRESS, + ADDRESS, + SIZE_T, + JAVA_LONG, + ADDRESS)); + frameFree = h("zu_frame_free", FunctionDescriptor.ofVoid(ADDRESS)); + connRegister = h("zu_conn_register", FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, ADDRESS)); + connUnregister = + h( + "zu_conn_unregister", + FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, SIZE_T, ADDRESS, ADDRESS)); + connRegisteredCount = + h("zu_conn_registered_count", FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS)); + connRegisteredName = + h("zu_conn_registered_name", FunctionDescriptor.of(ADDRESS, ADDRESS, JAVA_LONG, ADDRESS)); + valueType = critical("zu_value_type", FunctionDescriptor.of(JAVA_INT, ADDRESS)); valueBool = h("zu_value_bool", FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS)); valueI64 = h("zu_value_i64", FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS)); 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 8cc5342..d4cb887 100644 --- a/zudb-ffm/src/main/java/dev/zudb/ffm/FfmBinding.java +++ b/zudb-ffm/src/main/java/dev/zudb/ffm/FfmBinding.java @@ -958,6 +958,263 @@ public void appenderFree(long appender) { } } + // ---- frames ---- + + @Override + public long frameNew(String name, long rows, Runnable release) { + Scratch s = Scratch.get(); + MemorySegment sl = s.slots(); + MemorySegment n = s.utf8(name); + clear(sl); + Release callback = release == null ? null : Release.of(release); + MemorySegment stub = callback == null ? MemorySegment.NULL : callback.stub(); + boolean made = false; + try { + int st = + (int) + abi.frameNew.invokeExact( + n, + n.byteSize(), + rows, + MemorySegment.NULL, + stub, + sl.asSlice(OUT, 8), + sl.asSlice(ERR, 8)); + check("zu_frame_new", st, sl); + made = true; + return sl.get(ADDRESS, OUT).address(); + } catch (Throwable t) { + throw fail("zu_frame_new", t); + } finally { + if (!made && callback != null) { + // Nothing holds the stub now, and nothing will ever call it. + callback.abandon(); + } + } + } + + @Override + public void frameColumnInts( + long frame, + String name, + java.nio.Buffer values, + long count, + int bits, + boolean signed, + long scale, + int temporal) { + MemorySegment v = lent(values, name); + Scratch scratch = Scratch.get(); + MemorySegment sl = scratch.slots(); + clear(sl); + try (Arena arena = Arena.ofConfined()) { + MemorySegment n = utf8(arena, name); + int st = + (int) + abi.frameColInt.invokeExact( + ptr(frame), + n, + n.byteSize(), + v, + count, + bits, + signed ? 1 : 0, + scale, + temporal, + sl.asSlice(ERR, 8)); + check("zu_frame_col_int", st, sl); + } catch (Throwable t) { + throw fail("zu_frame_col_int", t); + } + } + + @Override + public void frameColumnFloats(long frame, String name, java.nio.Buffer values, long count, + int bits) { + MemorySegment v = lent(values, name); + Scratch scratch = Scratch.get(); + MemorySegment sl = scratch.slots(); + clear(sl); + try (Arena arena = Arena.ofConfined()) { + MemorySegment n = utf8(arena, name); + int st = + (int) + abi.frameColFloat.invokeExact( + ptr(frame), n, n.byteSize(), v, count, bits, sl.asSlice(ERR, 8)); + check("zu_frame_col_float", st, sl); + } catch (Throwable t) { + throw fail("zu_frame_col_float", t); + } + } + + @Override + public void frameColumnBooleans(long frame, String name, java.nio.Buffer bitmap, long count) { + MemorySegment b = lent(bitmap, name); + Scratch scratch = Scratch.get(); + MemorySegment sl = scratch.slots(); + clear(sl); + try (Arena arena = Arena.ofConfined()) { + MemorySegment n = utf8(arena, name); + int st = + (int) + abi.frameColBool.invokeExact( + ptr(frame), n, n.byteSize(), b, count, sl.asSlice(ERR, 8)); + check("zu_frame_col_bool", st, sl); + } catch (Throwable t) { + throw fail("zu_frame_col_bool", t); + } + } + + @Override + public void frameColumnStrings( + long frame, String name, java.nio.Buffer offsets, boolean wide, java.nio.Buffer data, + long count) { + MemorySegment o = lent(offsets, name); + MemorySegment d = lent(data, name); + Scratch scratch = Scratch.get(); + MemorySegment sl = scratch.slots(); + clear(sl); + try (Arena arena = Arena.ofConfined()) { + MemorySegment n = utf8(arena, name); + int st = + (int) + abi.frameColStr.invokeExact( + ptr(frame), + n, + n.byteSize(), + o, + wide ? 1 : 0, + d, + d.byteSize(), + count, + sl.asSlice(ERR, 8)); + check("zu_frame_col_str", st, sl); + } catch (Throwable t) { + throw fail("zu_frame_col_str", t); + } + } + + @Override + public void frameColumnViews( + long frame, String name, java.nio.Buffer views, List data, long count) { + MemorySegment v = lent(views, name); + int buffers = data.size(); + Scratch scratch = Scratch.get(); + MemorySegment sl = scratch.slots(); + clear(sl); + try (Arena arena = Arena.ofConfined()) { + MemorySegment n = utf8(arena, name); + MemorySegment pointers = arena.allocate(ADDRESS, Math.max(buffers, 1)); + MemorySegment lengths = arena.allocate(Abi.SIZE_T, Math.max(buffers, 1)); + for (int i = 0; i < buffers; i++) { + MemorySegment one = lent(data.get(i), name); + pointers.setAtIndex(ADDRESS, i, one); + size(lengths, i, one.byteSize()); + } + int st = + (int) + abi.frameColView.invokeExact( + ptr(frame), + n, + n.byteSize(), + v, + pointers, + lengths, + (long) buffers, + count, + sl.asSlice(ERR, 8)); + check("zu_frame_col_view", st, sl); + } catch (Throwable t) { + throw fail("zu_frame_col_view", t); + } + } + + @Override + public void frameFree(long frame) { + try { + abi.frameFree.invokeExact(ptr(frame)); + } catch (Throwable t) { + throw fail("zu_frame_free", t); + } + } + + @Override + public void connRegister(long conn, long frame) { + Scratch s = Scratch.get(); + MemorySegment sl = s.slots(); + clear(sl); + try { + int st = (int) abi.connRegister.invokeExact(ptr(conn), ptr(frame), sl.asSlice(ERR, 8)); + check("zu_conn_register", st, sl); + } catch (Throwable t) { + throw fail("zu_conn_register", t); + } + } + + @Override + public boolean connUnregister(long conn, String name) { + Scratch s = Scratch.get(); + MemorySegment sl = s.slots(); + MemorySegment n = s.utf8(name); + clear(sl); + try { + int st = + (int) + abi.connUnregister.invokeExact( + ptr(conn), n, n.byteSize(), sl.asSlice(OUT, 4), sl.asSlice(ERR, 8)); + check("zu_conn_unregister", st, sl); + return sl.get(JAVA_INT, OUT) != 0; + } catch (Throwable t) { + throw fail("zu_conn_unregister", t); + } + } + + @Override + public long connRegisteredCount(long conn) { + return counter(abi.connRegisteredCount, "zu_conn_registered_count", conn); + } + + @Override + public String connRegisteredName(long conn, long index) { + Scratch s = Scratch.get(); + MemorySegment sl = s.slots(); + try { + MemorySegment out = + (MemorySegment) abi.connRegisteredName.invokeExact(ptr(conn), index, sl.asSlice(LEN, 8)); + return utf8(out.address(), sl.get(JAVA_LONG, LEN)); + } catch (Throwable t) { + throw fail("zu_conn_registered_name", t); + } + } + + /** + * A buffer the engine may keep rather than read once. + * + *

This is the one place a copy is refused instead of made. Everywhere + * else a heap buffer costs a memcpy and nothing else, because the call reads + * it and is done. A frame keeps the pointer for as long as it is registered, + * and a heap buffer has no address anything outside the JVM can keep, so a + * copy here would mean the engine reading a copy for the rest of the frame's + * life. That is a frame that is not a frame, and quietly making one is worse + * than saying so. + */ + private static MemorySegment lent(java.nio.Buffer buffer, String name) { + if (buffer == null) { + throw Diagnostic.misuse(Status.MISUSE, "column " + name + " of a frame has no buffer") + .toException(); + } + if (!buffer.isDirect()) { + throw Diagnostic.misuse( + Status.MISUSE, + "column " + + name + + " of a frame is on the heap, and a frame is read where it lies rather than" + + " copied, so it wants a buffer from ByteBuffer.allocateDirect") + .toException(); + } + return MemorySegment.ofBuffer(buffer); + } + /** One of the {@code (handle, uint64_t *out)} calls that cannot fail with an error. */ private long counter(java.lang.invoke.MethodHandle mh, String what, long handle) { Scratch s = Scratch.get(); diff --git a/zudb-ffm/src/main/java/dev/zudb/ffm/Release.java b/zudb-ffm/src/main/java/dev/zudb/ffm/Release.java new file mode 100644 index 0000000..8b7ba03 --- /dev/null +++ b/zudb-ffm/src/main/java/dev/zudb/ffm/Release.java @@ -0,0 +1,141 @@ +package dev.zudb.ffm; + +import static java.lang.foreign.ValueLayout.ADDRESS; + +import java.lang.System.Logger; +import java.lang.System.Logger.Level; +import java.lang.foreign.Arena; +import java.lang.foreign.FunctionDescriptor; +import java.lang.foreign.Linker; +import java.lang.foreign.MemorySegment; +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; + +/** + * The one place this binding hands the engine a pointer to Java code rather + * than the other way round. + * + *

A frame's release callback is how a host learns the engine has finished + * with the buffers it lent: it runs once, on a thread of the library's, after + * the last statement reading the frame ends. Reaching Java from there needs an + * upcall stub, which is executable memory with a lifetime of its own. + * + *

That lifetime is the awkward part. The stub has to outlive the frame, + * because the callback is the last thing to happen and may happen after the + * frame was freed, so it cannot hang off the frame. It cannot free itself + * either: closing the arena a stub lives in from inside a call through that + * same stub is closing the ground you are standing on. So a spent arena goes + * on a queue and the next stub to be made closes it, which costs nothing, needs + * no thread of ours, and bounds the outstanding stubs at the number of frames + * whose callbacks have not run yet. + */ +final class Release { + + private static final Logger LOG = System.getLogger("dev.zudb"); + + private static final FunctionDescriptor DESCRIPTOR = FunctionDescriptor.ofVoid(ADDRESS); + + private static final MethodHandle RUN = run(); + + /** Arenas whose callback has been and gone, waiting for somebody else to close them. */ + private static final Queue SPENT = new ConcurrentLinkedQueue<>(); + + private final Runnable body; + private final Arena arena; + + private Release(Runnable body, Arena arena) { + this.body = body; + this.arena = arena; + } + + private MemorySegment stub; + + /** + * A {@code void (*)(void *)} the engine can call, bound to this runnable. + * + * @param body what to run when the engine is finished with the frame + * @return the release, which frees itself by way of the queue above + */ + @SuppressWarnings("restricted") + static Release of(Runnable body) { + sweep(); + // Shared rather than confined: the callback arrives on a thread of the + // library's, and a confined arena would refuse the call it is there to + // serve. + Arena arena = Arena.ofShared(); + try { + Release release = new Release(body, arena); + release.stub = Linker.nativeLinker().upcallStub(RUN.bindTo(release), DESCRIPTOR, arena); + return release; + } catch (RuntimeException | Error e) { + arena.close(); + throw e; + } + } + + /** + * The function pointer. + * + * @return the stub + */ + MemorySegment stub() { + return stub; + } + + /** + * Says the callback will never be called, for the frame that failed to be + * made at all, so that the stub goes the way a spent one goes. + */ + void abandon() { + SPENT.add(arena); + } + + /** + * Called by the engine. + * + *

Nothing may be thrown out of here. An exception crossing an upcall + * takes the whole JVM down, and a host's release callback is exactly the + * kind of code that throws: it takes a lock, or a runtime's interpreter + * lock, and lets go of buffers. So it is logged and swallowed, which leaves + * the process alive and the mistake findable. + * + * @param owner the pointer passed at creation, which this binding does not + * use because the runnable already knows what it owns + */ + @SuppressWarnings("unused") + private void run(MemorySegment owner) { + try { + body.run(); + } catch (Throwable t) { + LOG.log(Level.ERROR, "a zu frame release callback threw, which nothing above it can see", t); + } finally { + SPENT.add(arena); + } + } + + /** Closes the arenas of every callback that has already run. */ + private static void sweep() { + for (Arena arena = SPENT.poll(); arena != null; arena = SPENT.poll()) { + try { + arena.close(); + } catch (RuntimeException e) { + // A shared arena refuses to close while a thread is still inside a + // call into it. Put it back and let the next frame try. + SPENT.add(arena); + return; + } + } + } + + private static MethodHandle run() { + try { + return MethodHandles.lookup() + .findVirtual(Release.class, "run", MethodType.methodType(void.class, MemorySegment.class)); + } catch (ReflectiveOperationException e) { + throw new ExceptionInInitializerError(e); + } + } +} diff --git a/zudb-ffm/src/test/java/dev/zudb/ffm/FrameTest.java b/zudb-ffm/src/test/java/dev/zudb/ffm/FrameTest.java new file mode 100644 index 0000000..958cc18 --- /dev/null +++ b/zudb-ffm/src/test/java/dev/zudb/ffm/FrameTest.java @@ -0,0 +1,488 @@ +package dev.zudb.ffm; + +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.Frame; +import dev.zudb.Loader; +import dev.zudb.Result; +import dev.zudb.Value; +import dev.zudb.ZuException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.DoubleBuffer; +import java.nio.FloatBuffer; +import java.nio.IntBuffer; +import java.nio.LongBuffer; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * Querying memory the test already holds, without any of it getting into a + * database. + */ +class FrameTest { + + @TempDir Path dir; + + @BeforeAll + static void engine() { + Libzu.require(); + } + + @Test + void aColumnOfLongsIsQueriedWhereItLies() { + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame frame = Frame.of("Person", 3)) { + frame.column("id", longs(1, 2, 3)); + conn.register(frame); + try (Result r = conn.query("MATCH (p:Person) RETURN p.id AS id ORDER BY id")) { + assertEquals(List.of(1L, 2L, 3L), r.stream().map(row -> row.getLong(0)).toList()); + } + } + } + + @Test + void everyWidthAndSignComesBackAsTheNumberItIs() { + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame frame = Frame.of("Wide", 2)) { + frame.column("big", longs(1, -2)); + frame.column("mid", ints(3, -4)); + frame.column("small", shorts((short) 5, (short) -6)); + frame.integers("tiny", bytes((byte) 7, (byte) 8), 2, 8, false, 1, null); + frame.integers("unsigned", ints(9, 10), 2, 32, false, 1, null); + conn.register(frame); + try (Result r = + conn.query("MATCH (w:Wide) RETURN w.big, w.mid, w.small, w.tiny, w.unsigned")) { + assertEquals(1L, r.row(0).getLong(0)); + assertEquals(3L, r.row(0).getLong(1)); + assertEquals(5L, r.row(0).getLong(2)); + assertEquals(7L, r.row(0).getLong(3)); + assertEquals(9L, r.row(0).getLong(4)); + assertEquals(-2L, r.row(1).getLong(0)); + assertEquals(-4L, r.row(1).getLong(1)); + assertEquals(-6L, r.row(1).getLong(2)); + assertEquals(8L, r.row(1).getLong(3)); + assertEquals(10L, r.row(1).getLong(4)); + } + } + } + + @Test + void bothWidthsOfFloatComeBack() { + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame frame = Frame.of("Reading", 2)) { + frame.column("wide", doubles(1.5, 2.5)); + frame.column("narrow", floats(0.5f, 0.25f)); + conn.register(frame); + try (Result r = conn.query("MATCH (x:Reading) RETURN x.wide, x.narrow")) { + assertEquals(1.5, r.row(0).getDouble(0)); + assertEquals(0.5, r.row(0).getDouble(1)); + assertEquals(2.5, r.row(1).getDouble(0)); + assertEquals(0.25, r.row(1).getDouble(1)); + } + } + } + + @Test + void aBitmapIsAColumnOfBooleans() { + // 0b00001101: true, false, true, true. + ByteBuffer bitmap = ByteBuffer.allocateDirect(1); + bitmap.put((byte) 0b1101).flip(); + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame frame = Frame.of("Flag", 4)) { + frame.booleans("on", bitmap, 4); + conn.register(frame); + try (Result r = conn.query("MATCH (f:Flag) RETURN f.on")) { + assertEquals( + List.of(true, false, true, true), + r.stream().map(row -> row.getBoolean(0)).toList()); + } + } + } + + @Test + void charactersEndToEndAreAColumnOfStrings() { + List words = List.of("ada", "", "grace", "éàü", "a".repeat(300)); + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame frame = Frame.of("Word", words.size())) { + frame.strings("text", offsets32(words), characters(words)); + conn.register(frame); + try (Result r = conn.query("MATCH (w:Word) RETURN w.text")) { + assertEquals(words, r.stream().map(row -> row.getString(0)).toList()); + } + } + } + + @Test + void sixtyFourBitOffsetsAreTheSameColumn() { + List words = List.of("ada", "grace", "alan"); + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame frame = Frame.of("Word", words.size())) { + frame.strings("text", offsets64(words), characters(words)); + conn.register(frame); + try (Result r = conn.query("MATCH (w:Word) RETURN w.text")) { + assertEquals(words, r.stream().map(row -> row.getString(0)).toList()); + } + } + } + + @Test + void aViewColumnIsReadWithoutItsCharactersMoving() { + // Short strings live in the view itself, long ones point into the data + // buffer, and both have to come back the same. + List words = List.of("ada", "a string too long to live in sixteen bytes"); + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame frame = Frame.of("Word", words.size())) { + ByteBuffer data = characters(words); + frame.views("text", views(words), List.of(data)); + conn.register(frame); + try (Result r = conn.query("MATCH (w:Word) RETURN w.text")) { + assertEquals(words, r.stream().map(row -> row.getString(0)).toList()); + } + } + } + + @Test + void aDateGoesInAsTheDaysItIsAndComesBackAsADate() { + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame frame = Frame.of("Event", 2)) { + frame.dates("on", ints(19782, 0)); + conn.register(frame); + try (Result r = conn.query("MATCH (e:Event) RETURN e.on")) { + Value.Temporal first = r.row(0).getTemporal(0); + assertEquals(Value.Temporal.Kind.DATE, first.kind()); + assertEquals(19782L, first.count()); + } + } + } + + @Test + void arrowMicrosecondsAreScaledToTheNanosecondsThisEngineCountsIn() { + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame frame = Frame.of("Event", 1)) { + frame.timestamps("at", longs(1_700_000_000_000_000L)); + conn.register(frame); + try (Result r = conn.query("MATCH (e:Event) RETURN e.at")) { + Value.Temporal at = r.row(0).getTemporal(0); + assertEquals(Value.Temporal.Kind.LOCAL_DATETIME, at.kind()); + assertEquals(1_700_000_000_000_000_000L, at.count()); + } + } + } + + @Test + void oneFrameServesAsManyConnectionsAsThereAreThreadsToQueryFromIt() { + try (Database db = Database.memory(); + Connection first = db.connect(); + Connection second = db.connect(); + Frame frame = Frame.of("Person", 2)) { + frame.column("id", longs(10, 20)); + first.register(frame); + second.register(frame); + try (Result a = first.query("MATCH (p:Person) RETURN sum(p.id)"); + Result b = second.query("MATCH (p:Person) RETURN sum(p.id)")) { + assertEquals(30L, a.row(0).getLong(0)); + assertEquals(30L, b.row(0).getLong(0)); + } + } + } + + @Test + void theConnectionSaysWhatIsRegisteredOnIt() { + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame people = Frame.of("Person", 1); + Frame places = Frame.of("Place", 1)) { + people.column("id", longs(1)); + places.column("id", longs(2)); + assertEquals(0, conn.registeredCount()); + conn.register(people); + conn.register(places); + assertEquals(2, conn.registeredCount()); + assertEquals(List.of("Person", "Place"), conn.registeredNames()); + assertTrue(conn.unregister("Person")); + assertEquals(List.of("Place"), conn.registeredNames()); + assertFalse(conn.unregister("Person")); + } + } + + @Test + void theReleaseCallbackSaysWhenTheEngineIsFinished() throws InterruptedException { + CountDownLatch released = new CountDownLatch(1); + try (Database db = Database.memory(); + Connection conn = db.connect()) { + try (Frame frame = Frame.of("Person", 2, released::countDown)) { + frame.column("id", longs(1, 2)); + conn.register(frame); + conn.query("MATCH (p:Person) RETURN p.id").close(); + assertEquals(1, released.getCount(), "nothing has let go of the columns yet"); + conn.unregister("Person"); + } + assertTrue( + released.await(5, TimeUnit.SECONDS), "the callback never ran after the frame went away"); + } + } + + @Test + void aBufferNothingElseRefersToIsNotCollectedOutFromUnderTheEngine() { + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame frame = Frame.of("Person", 3)) { + // Nothing but the frame refers to this buffer once the call returns, and + // a direct buffer that becomes unreachable has its memory freed by a + // cleaner. If the frame did not hold on to it the engine would be + // pointing at memory that had been handed back. + frame.column("id", longs(1, 2, 3)); + conn.register(frame); + for (int i = 0; i < 10; i++) { + System.gc(); + ByteBuffer.allocateDirect(1 << 20); + } + try (Result r = conn.query("MATCH (p:Person) RETURN sum(p.id)")) { + assertEquals(6L, r.row(0).getLong(0)); + } + } + } + + @Test + void aFrameNoConnectionEverSawStillLetsGoOfWhatItHeld() throws InterruptedException { + CountDownLatch released = new CountDownLatch(1); + try (Frame frame = Frame.of("Person", 1, released::countDown)) { + frame.column("id", longs(1)); + } + assertTrue(released.await(5, TimeUnit.SECONDS), "the callback never ran"); + } + + @Test + void aHeapBufferIsRefusedRatherThanCopied() { + try (Frame frame = Frame.of("Person", 3)) { + ZuException e = + assertThrows(ZuException.class, () -> frame.column("id", LongBuffer.wrap(new long[3]))); + assertTrue(e.getMessage().contains("allocateDirect"), e.getMessage()); + } + } + + @Test + void aColumnOfTheWrongLengthIsRefusedAtThatColumn() { + try (Frame frame = Frame.of("Person", 3)) { + assertThrows(ZuException.class, () -> frame.column("id", longs(1, 2))); + } + } + + @Test + void aNameAStoredTableHoldsIsRefused() { + Path path = dir.resolve("people.zu"); + try (Loader loader = Loader.create(path)) { + loader.table("Person", "Knows", 1); + loader.column("id", 1L); + loader.finish(); + } + try (Database db = Database.open(path); + Connection conn = db.connect(); + Frame frame = Frame.of("Person", 1)) { + frame.column("id", longs(9)); + assertThrows(ZuException.class, () -> conn.register(frame)); + } + } + + @Test + void aFrameReplacesAFrameOfTheSameName() { + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame first = Frame.of("Person", 1); + Frame second = Frame.of("Person", 1)) { + first.column("id", longs(1)); + second.column("id", longs(2)); + conn.register(first); + conn.register(second); + assertEquals(1, conn.registeredCount()); + try (Result r = conn.query("MATCH (p:Person) RETURN p.id")) { + assertEquals(2L, r.row(0).getLong(0)); + } + } + } + + @Test + void registeringInsideATransactionIsRefused() { + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame frame = Frame.of("Person", 1)) { + frame.column("id", longs(1)); + conn.begin(); + try { + assertThrows(ZuException.class, () -> conn.register(frame)); + } finally { + conn.rollback(); + } + } + } + + @Test + void aStatementThatWouldWriteToAFrameIsRefused() { + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame frame = Frame.of("Person", 1)) { + frame.column("id", longs(1)); + conn.register(frame); + assertThrows(ZuException.class, () -> conn.execute("MATCH (p:Person) SET p.id = 5")); + } + } + + @Test + void aClosedFrameSaysSoRatherThanCrashing() { + Frame frame = Frame.of("Person", 1); + frame.close(); + assertTrue(frame.isClosed()); + assertThrows(ZuException.class, () -> frame.column("id", longs(1))); + } + + @Test + void aSliceOfABufferIsTheSliceAndNotTheWholeThing() { + LongBuffer all = longs(1, 2, 3, 4, 5); + all.position(1).limit(4); + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame frame = Frame.of("Person", 3)) { + frame.column("id", all.slice()); + conn.register(frame); + try (Result r = conn.query("MATCH (p:Person) RETURN p.id AS id ORDER BY id")) { + assertEquals(List.of(2L, 3L, 4L), r.stream().map(row -> row.getLong(0)).toList()); + } + } + } + + @Test + void aFrameOfManyRowsIsReadWholeAndCorrectly() { + int rows = 200_000; + LongBuffer ids = direct(rows * 8).asLongBuffer(); + for (int i = 0; i < rows; i++) { + ids.put(i); + } + ids.flip(); + try (Database db = Database.memory(); + Connection conn = db.connect(); + Frame frame = Frame.of("Person", rows)) { + frame.column("id", ids); + conn.register(frame); + try (Result r = conn.query("MATCH (p:Person) RETURN count(p), sum(p.id)")) { + assertEquals(rows, r.row(0).getLong(0)); + assertEquals((long) rows * (rows - 1) / 2, r.row(0).getLong(1)); + } + } + } + + // ---- direct buffers, which is all a frame takes ---- + + private static ByteBuffer direct(int bytes) { + return ByteBuffer.allocateDirect(bytes).order(ByteOrder.nativeOrder()); + } + + private static LongBuffer longs(long... values) { + LongBuffer b = direct(values.length * 8).asLongBuffer(); + return b.put(values).flip(); + } + + private static IntBuffer ints(int... values) { + IntBuffer b = direct(values.length * 4).asIntBuffer(); + return b.put(values).flip(); + } + + private static java.nio.ShortBuffer shorts(short... values) { + java.nio.ShortBuffer b = direct(values.length * 2).asShortBuffer(); + return b.put(values).flip(); + } + + private static ByteBuffer bytes(byte... values) { + ByteBuffer b = direct(values.length); + return b.put(values).flip(); + } + + private static DoubleBuffer doubles(double... values) { + DoubleBuffer b = direct(values.length * 8).asDoubleBuffer(); + return b.put(values).flip(); + } + + private static FloatBuffer floats(float... values) { + FloatBuffer b = direct(values.length * 4).asFloatBuffer(); + return b.put(values).flip(); + } + + /** The characters of a column of strings, end to end, which is Arrow's data buffer. */ + private static ByteBuffer characters(List words) { + byte[] all = + words.stream().collect(Collectors.joining()).getBytes(StandardCharsets.UTF_8); + ByteBuffer b = direct(Math.max(all.length, 1)); + b.put(all).flip(); + return b; + } + + private static IntBuffer offsets32(List words) { + IntBuffer b = direct((words.size() + 1) * 4).asIntBuffer(); + int at = 0; + b.put(0); + for (String w : words) { + at += w.getBytes(StandardCharsets.UTF_8).length; + b.put(at); + } + return b.flip(); + } + + private static LongBuffer offsets64(List words) { + LongBuffer b = direct((words.size() + 1) * 8).asLongBuffer(); + long at = 0; + b.put(0); + for (String w : words) { + at += w.getBytes(StandardCharsets.UTF_8).length; + b.put(at); + } + return b.flip(); + } + + /** + * Arrow's Utf8View: four bytes of length, then either the characters + * themselves when there are twelve or fewer of them, or a four byte prefix + * and the buffer and offset they are at. + */ + private static ByteBuffer views(List words) { + ByteBuffer b = direct(words.size() * 16); + int at = 0; + for (String w : words) { + byte[] utf8 = w.getBytes(StandardCharsets.UTF_8); + b.putInt(utf8.length); + if (utf8.length <= 12) { + b.put(utf8); + for (int i = utf8.length; i < 12; i++) { + b.put((byte) 0); + } + } else { + b.put(utf8, 0, 4); + b.putInt(0); + b.putInt(at); + } + at += utf8.length; + } + return b.flip(); + } +} diff --git a/zudb/src/main/java/dev/zudb/Connection.java b/zudb/src/main/java/dev/zudb/Connection.java index 5bc9463..0d9cc10 100644 --- a/zudb/src/main/java/dev/zudb/Connection.java +++ b/zudb/src/main/java/dev/zudb/Connection.java @@ -1,6 +1,8 @@ package dev.zudb; import dev.zudb.spi.ZuBinding; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Supplier; @@ -84,6 +86,65 @@ public Appender appender(String table) { return new Appender(zu, zu.appenderOpen(open(), table)); } + /** + * Names a frame as a table of this connection, so that statements read the + * caller's own columns where they lie. + * + *

Nothing is copied here or at read. The frame is not spent either, so + * the same columns may be registered on as many connections as there are + * threads to query them from. + * + *

This is where everything a frame's description could get wrong is + * settled: alignment, an unsigned value too large for the signed lane, a + * scale that would overflow, an offset that leaves its buffer. After it + * returns, a read of the frame cannot fail. + * + * @param frame the columns and what they are called + * @throws ZuException if a stored table already holds the name, or if a + * transaction is running, since a table appearing halfway through one is + * not something the transaction could be rolled back over + */ + public void register(Frame frame) { + zu.connRegister(open(), frame.handle()); + } + + /** + * Drops a registered frame. + * + *

A statement already running keeps the frame it started with, and the + * release callback waits for it. + * + * @param name the table name it was registered under + * @return whether there was one under that name + */ + public boolean unregister(String name) { + return zu.connUnregister(open(), name); + } + + /** + * How many frames are registered. + * + * @return the count + */ + public long registeredCount() { + return zu.connRegisteredCount(open()); + } + + /** + * What every registered frame is called, sorted. + * + * @return the names, which is a list of its own and not a view of anything + */ + public List registeredNames() { + long h = open(); + long count = zu.connRegisteredCount(h); + List names = new ArrayList<>((int) count); + for (long i = 0; i < count; i++) { + names.add(zu.connRegisteredName(h, i)); + } + return names; + } + /** * A second connection on the database this one is already on, made without * a path. diff --git a/zudb/src/main/java/dev/zudb/Frame.java b/zudb/src/main/java/dev/zudb/Frame.java new file mode 100644 index 0000000..6c26931 --- /dev/null +++ b/zudb/src/main/java/dev/zudb/Frame.java @@ -0,0 +1,405 @@ +package dev.zudb; + +import dev.zudb.spi.ZuBinding; +import java.nio.Buffer; +import java.nio.ByteBuffer; +import java.nio.DoubleBuffer; +import java.nio.FloatBuffer; +import java.nio.IntBuffer; +import java.nio.LongBuffer; +import java.nio.ShortBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Columns your program already holds, named as a table and queried where they + * lie. + * + *

This is the other direction from {@link Loader}. A loader takes your + * columns and writes a database out of them. A frame writes nothing at all: + * you describe the memory you are already holding, register it on a + * {@link Connection}, and statements read your buffers directly. Nothing is + * copied at registration and nothing is copied at read, so a frame of ten + * million rows is ready in the time it takes to walk its columns. + * + * 

{@code
+ * LongBuffer ids = ByteBuffer.allocateDirect(3 * 8).order(ByteOrder.nativeOrder()).asLongBuffer();
+ * ids.put(new long[] {1, 2, 3}).flip();
+ * try (Frame frame = Frame.of("Person", 3)) {
+ *   frame.column("id", ids);
+ *   conn.register(frame);
+ *   try (Result r = conn.query("MATCH (p:Person) RETURN sum(p.id) AS total")) {
+ *     System.out.println(r.row(0).getLong(0));
+ *   }
+ * }
+ * }
+ * + *

Every buffer has to be direct, and one that is not is refused rather than + * copied. Everywhere else in this API a heap buffer costs a memcpy and nothing + * else, because the call reads it and is finished. A frame keeps the pointer + * for as long as it is registered, so a copy here would mean the engine + * reading a copy of your data for the rest of the frame's life. That is a + * frame that is not a frame, and it is better said out loud than done quietly. + * + *

The buffers stay yours and this library never writes one. What it asks is + * that each stays where it is, unwritten and unfreed, until the engine is + * finished with it, which is after the last statement reading the frame ends + * and is neither the unregister that preceded it nor {@link #close()}. + * + *

For a direct buffer that is looked after for you. A frame keeps a + * reference to everything handed to it and lets go only when the engine says + * it is finished, so a buffer cannot be collected out from under a running + * statement and you do not have to keep a field alive to prevent it. The + * optional release callback is for the other kind of buffer: one over memory + * you allocated yourself, or one a lock has to be taken to let go of. It runs + * once, on a thread of the library's, at that same moment. + * + *

The order is fixed: make the frame, describe one column at a time, then + * register it. A column whose count does not match the frame's row count is + * refused at that column, where you still know which one you were describing. + * Everything else that can go wrong is settled at + * {@link Connection#register(Frame)}: alignment, an unsigned value too large + * for the signed lane, a scale that would overflow, an offset that leaves its + * buffer. A read of a registered frame cannot fail, which is what lets a scan + * be a loop. + * + *

A frame is described once and registered as often as you like, on as many + * connections as you like. It is read only and has no edges: a statement that + * would insert into, set on or delete from a registered name is refused, a + * name a stored table already holds is refused, and a name another frame holds + * replaces that frame. + * + *

Which lane a column takes decides what it costs to read. Sixty-four + * signed bits, doubles, one bit a row for a boolean, and characters end to end + * with offsets cutting them up are what this engine keeps natively, and those + * cost nothing. A narrower integer, an unsigned one, a single-precision float + * and Arrow's microseconds against the nanoseconds this engine keeps time in + * are widened a value at a time as a statement reaches them, so they cost + * something, but only for the columns the statement actually named. + */ +public final class Frame implements AutoCloseable { + + /** What the C ABI calls a column of numbers and nothing else. */ + private static final int PLAIN = -1; + + private final ZuBinding zu; + private final AtomicLong handle; + private final String name; + private final long rows; + private final List lent; + + private Frame(ZuBinding zu, long handle, String name, long rows, List lent) { + this.zu = zu; + this.handle = new AtomicLong(handle); + this.name = name; + this.rows = rows; + this.lent = lent; + } + + /** + * A frame of buffers a garbage collector already looks after. + * + * @param name what the table is called in a statement + * @param rows how many rows every column of it carries + * @return the frame, which the caller closes + */ + public static Frame of(String name, long rows) { + return of(name, rows, null); + } + + /** + * A frame whose buffers something has to be told about. + * + * @param name what the table is called in a statement + * @param rows how many rows every column of it carries + * @param release run once, on a thread of the library's, after the last + * statement reading this frame ends, which is where a host that has to + * take a lock to let go of what it lent takes it + * @return the frame, which the caller closes + */ + public static Frame of(String name, long rows, Runnable release) { + ZuBinding zu = Zu.binding(); + // The list is what keeps the caller's buffers reachable, and the callback + // is what holds the list, so a direct buffer nothing else refers to lives + // exactly as long as the engine may still read it. That is why there is a + // callback even when the caller asked for none. + List lent = new ArrayList<>(); + Runnable body = new Keep(lent, release); + return new Frame(zu, zu.frameNew(name, rows, body), name, rows, lent); + } + + /** What the engine calls when it has finished, and what holds the buffers until then. */ + private static final class Keep implements Runnable { + + private final List lent; + private final Runnable body; + + Keep(List lent, Runnable body) { + this.lent = lent; + this.body = body; + } + + @Override + public void run() { + lent.clear(); + if (body != null) { + body.run(); + } + } + } + + /** + * A column of 64-bit signed integers, which is the lane and costs nothing. + * + * @param column the column + * @param values one a row, between the buffer's position and its limit + * @return this frame + */ + public Frame column(String column, LongBuffer values) { + return integers(column, values, values.remaining(), 64, true, 1, null); + } + + /** + * A column of 32-bit signed integers, widened as a statement reaches them. + * + * @param column the column + * @param values one a row, between the buffer's position and its limit + * @return this frame + */ + public Frame column(String column, IntBuffer values) { + return integers(column, values, values.remaining(), 32, true, 1, null); + } + + /** + * A column of 16-bit signed integers, widened as a statement reaches them. + * + * @param column the column + * @param values one a row, between the buffer's position and its limit + * @return this frame + */ + public Frame column(String column, ShortBuffer values) { + return integers(column, values, values.remaining(), 16, true, 1, null); + } + + /** + * A column of doubles, which is the lane and costs nothing. + * + * @param column the column + * @param values one a row, between the buffer's position and its limit + * @return this frame + */ + public Frame column(String column, DoubleBuffer values) { + zu.frameColumnFloats(open(), column, values, values.remaining(), 64); + return keep(values); + } + + /** + * A column of single-precision floats, widened as a statement reaches them. + * + * @param column the column + * @param values one a row, between the buffer's position and its limit + * @return this frame + */ + public Frame column(String column, FloatBuffer values) { + zu.frameColumnFloats(open(), column, values, values.remaining(), 32); + return keep(values); + } + + /** + * A column of integers in whatever width and sign the host holds them in. + * + *

This is the full form the six calls above are shorthand for, and it is + * what an Arrow array of any integer type maps onto. + * + * @param column the column + * @param values the buffer, which has to be direct + * @param count how many values are in it, which has to be the frame's row + * count + * @param bits 8, 16, 32 or 64 + * @param signed whether they are signed + * @param scale what one value is multiplied by to reach the unit its + * meaning counts in, so 1 for an integer and a date, and 1000 for the + * microseconds Arrow keeps a time or a timestamp in + * @param kind what the counts mean, or null for a column of numbers and + * nothing else + * @return this frame + */ + public Frame integers( + String column, + Buffer values, + long count, + int bits, + boolean signed, + long scale, + Value.Temporal.Kind kind) { + zu.frameColumnInts( + open(), column, values, count, bits, signed, scale, kind == null ? PLAIN : kind.value()); + return keep(values); + } + + /** + * A column of booleans, one bit a row, low bit of the first byte first, + * which is Arrow's bitmap and this engine's alike. + * + *

A host holding a slice with a bit offset of its own owes the shift + * before it gets here: a bitmap that starts partway into a byte is not a + * thing a pointer can say. + * + * @param column the column + * @param bitmap the bits, which has to be direct + * @param count how many rows are in it, which the bitmap cannot say + * @return this frame + */ + public Frame booleans(String column, ByteBuffer bitmap, long count) { + zu.frameColumnBooleans(open(), column, bitmap, count); + return keep(bitmap); + } + + /** + * A column of strings as Arrow's Utf8 keeps them. + * + *

The characters never move, either now or at read: what a scan builds + * is a view pointing back into your data buffer. + * + * @param column the column + * @param offsets where each string starts, of which there are one more than + * there are rows, the last being how much of data is used + * @param data the characters, end to end + * @return this frame + */ + public Frame strings(String column, IntBuffer offsets, ByteBuffer data) { + zu.frameColumnStrings(open(), column, offsets, false, data, offsets.remaining() - 1L); + return keep(offsets).keep(data); + } + + /** + * A column of strings as Arrow's LargeUtf8 keeps them, which is the same + * with 64-bit offsets. + * + * @param column the column + * @param offsets where each string starts, of which there are one more than + * there are rows + * @param data the characters, end to end + * @return this frame + */ + public Frame strings(String column, LongBuffer offsets, ByteBuffer data) { + zu.frameColumnStrings(open(), column, offsets, true, data, offsets.remaining() - 1L); + return keep(offsets).keep(data); + } + + /** + * A column of strings as Arrow's Utf8View keeps them, sixteen bytes a row + * over one or more data buffers. + * + *

A short string in that layout is already this engine's own view, byte + * for byte, so this is the cheapest string column there is. + * + * @param column the column + * @param views sixteen bytes a row + * @param data the buffers they point into + * @return this frame + */ + public Frame views(String column, ByteBuffer views, List data) { + List buffers = new ArrayList<>(data); + zu.frameColumnViews(open(), column, views, buffers, views.remaining() / 16L); + keep(views); + lent.addAll(buffers); + return this; + } + + /** + * A column of dates as Arrow's Date32 keeps them, days since the epoch. + * + * @param column the column + * @param epochDays one a row + * @return this frame + */ + public Frame dates(String column, IntBuffer epochDays) { + return integers( + column, epochDays, epochDays.remaining(), 32, true, 1, Value.Temporal.Kind.DATE); + } + + /** + * A column of timestamps as Arrow keeps them at microsecond precision, + * scaled to the nanoseconds this engine counts time in. + * + * @param column the column + * @param epochMicros one a row + * @return this frame + */ + public Frame timestamps(String column, LongBuffer epochMicros) { + return integers( + column, + epochMicros, + epochMicros.remaining(), + 64, + true, + 1000, + Value.Temporal.Kind.LOCAL_DATETIME); + } + + /** + * What the table is called in a statement. + * + * @return the name + */ + public String name() { + return name; + } + + /** + * How many rows every column of it carries. + * + * @return the count + */ + public long rows() { + return rows; + } + + /** + * Whether this frame has been closed. + * + * @return true once {@link #close()} has run + */ + public boolean isClosed() { + return handle.get() == 0; + } + + /** + * Ends the frame. + * + *

This is not what tells you the engine has let go of your buffers. A + * statement that started before this may still be reading them, and the + * release callback is what runs when the last one has finished. Closing + * twice does nothing the second time. + */ + @Override + public void close() { + long h = handle.getAndSet(0); + if (h != 0) { + zu.frameFree(h); + } + } + + long handle() { + return open(); + } + + /** + * Holds on to a buffer the engine now points at, so that a collector cannot + * free it while a statement is reading it. + */ + private Frame keep(Buffer buffer) { + lent.add(buffer); + return this; + } + + private long open() { + long h = handle.get(); + if (h == 0) { + throw new ZuClosedException(Diagnostic.misuse(Status.MISUSE_CLOSED, "this frame is closed")); + } + return h; + } +} diff --git a/zudb/src/main/java/dev/zudb/spi/ZuBinding.java b/zudb/src/main/java/dev/zudb/spi/ZuBinding.java index ba8429e..7db49f3 100644 --- a/zudb/src/main/java/dev/zudb/spi/ZuBinding.java +++ b/zudb/src/main/java/dev/zudb/spi/ZuBinding.java @@ -1,6 +1,7 @@ package dev.zudb.spi; import dev.zudb.Diagnostic; +import java.nio.Buffer; import java.nio.ByteBuffer; import java.nio.DoubleBuffer; import java.nio.IntBuffer; @@ -811,4 +812,146 @@ public interface ZuBinding { * @param appender the appender */ void appenderFree(long appender); + + // ---- frames ---- + + /** + * Describes a table of columns the host already holds, which the engine + * will read where they lie. + * + *

Every buffer handed to a {@code frameColumn} call below has to be a + * direct one. A frame keeps the pointer rather than the values, and a heap + * buffer has no address anything outside the JVM can keep. A provider + * refuses one rather than copying, because a copy would quietly undo the + * only thing a frame is for. + * + * @param name what the table is called in a statement + * @param rows how many rows every column of it carries + * @param release run once, on a thread of the library's, after the last + * statement reading this frame ends, or null for a host whose buffers + * outlive the process + * @return the handle + */ + long frameNew(String name, long rows, Runnable release); + + /** + * A column of integers, or of a temporal counted as integers. + * + * @param frame the frame + * @param name the column + * @param values the buffer, which has to be direct + * @param count how many values are in it + * @param bits 8, 16, 32 or 64, where 64 signed at scale 1 is the lane this + * engine reads natively and the column that costs nothing at all + * @param signed whether they are signed + * @param scale what one value is multiplied by to reach the unit its + * meaning counts in, so 1 for an integer and a date and 1000 for the + * microseconds Arrow keeps a timestamp in + * @param temporal one of the {@code ZU_TEMPORAL_} kinds, or -1 for a column + * of numbers and nothing else + */ + void frameColumnInts( + long frame, + String name, + Buffer values, + long count, + int bits, + boolean signed, + long scale, + int temporal); + + /** + * A column of floating point numbers. + * + * @param frame the frame + * @param name the column + * @param values the buffer, which has to be direct + * @param count how many values are in it + * @param bits 32 or 64, where 64 is the lane + */ + void frameColumnFloats(long frame, String name, Buffer values, long count, int bits); + + /** + * A column of booleans, one bit a row, low bit of the first byte first, + * which is Arrow's bitmap and this engine's alike. + * + * @param frame the frame + * @param name the column + * @param bitmap the buffer, which has to be direct + * @param count how many rows are in it, which the bitmap cannot say + */ + void frameColumnBooleans(long frame, String name, Buffer bitmap, long count); + + /** + * A column of strings as Arrow keeps them, characters end to end with + * offsets cutting them up. + * + * @param frame the frame + * @param name the column + * @param offsets the offsets, which has to be direct, and of which there + * are count + 1 + * @param wide false for 32-bit offsets, which is Arrow's Utf8, and true for + * 64-bit, which is its LargeUtf8 + * @param data the characters, which has to be direct + * @param count how many rows are in it + */ + void frameColumnStrings( + long frame, String name, Buffer offsets, boolean wide, Buffer data, long count); + + /** + * A column of strings as Arrow's Utf8View keeps them, sixteen bytes a row + * over one or more data buffers. + * + * @param frame the frame + * @param name the column + * @param views the views, which has to be direct + * @param data the buffers they point into, each of which has to be direct + * @param count how many rows are in it + */ + void frameColumnViews(long frame, String name, Buffer views, List data, long count); + + /** + * Releases a frame. The handle is the caller's on every path, so this is + * what ends one whether or not it was ever registered. + * + * @param frame the frame + */ + void frameFree(long frame); + + /** + * Registers a frame as a table of this connection. Does not spend the + * frame, which may be registered on as many connections as you like. + * + * @param conn the connection + * @param frame the frame + */ + void connRegister(long conn, long frame); + + /** + * Drops a registered frame. + * + * @param conn the connection + * @param name the table name it was registered under + * @return whether there was one under that name + */ + boolean connUnregister(long conn, String name); + + /** + * How many frames are registered, which is also the call that refreshes + * the names {@link #connRegisteredName(long, long)} hands out. + * + * @param conn the connection + * @return the count + */ + long connRegisteredCount(long conn); + + /** + * One registered name, in the sorted order the last count call read them + * in. + * + * @param conn the connection + * @param index from nought + * @return the name, or null out of range + */ + String connRegisteredName(long conn, long index); }