Skip to content

Native: columnar bulk insert by address for strings, nulls and vectors (no JSON text) #72

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tae898 merged 2 commits into from
Oct 25, 2026
Merged

Native: columnar bulk insert by address for strings, nulls and vectors (no JSON text) #72

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6fd1624
Native: columnar bulk insert by address for strings, nulls and vector…
tae898 Oct 25, 2026
032122d
Merge native-two-dist (the stack: #71, #67, #69, #70) into native-col…
tae898 Oct 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 24 additions & 3 deletions native/src/main/java/org/humemai/nativelib/ArcadeNative.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.graalvm.nativeimage.c.type.CCharPointer;
import org.graalvm.nativeimage.c.type.CFloatPointer;
import org.graalvm.nativeimage.c.type.CLongPointer;
import org.graalvm.word.Pointer;
import org.graalvm.word.WordFactory;
import org.graalvm.nativeimage.c.type.CTypeConversion;

Expand Down Expand Up @@ -510,13 +511,33 @@ private static CCharPointer replyOf(final ResultSet rs, final int max, final int
return p;
}

/** Columnar insert (see BulkOps): header JSON, one buffer holding the columns. Returns {"ok":true,"n":rows}. */
/** Columnar insert (see BulkOps): header JSON, one buffer holding the columns back to back. Returns {"ok":true,"n":rows}. */
@CEntryPoint(name = "arcadedb_n_insert_cols")
public static CCharPointer insertCols(final IsolateThread thread, final long h, final CCharPointer type, final CCharPointer header,
final CCharPointer buf, final long bufBytes) {
try {
final java.nio.ByteBuffer bb = CTypeConversion.asByteBuffer(buf, (int) bufBytes);
return out("{\"ok\":true,\"n\":" + BulkOps.insertColumns(db(h), str(type), str(header), bb) + "}");
return out("{\"ok\":true,\"n\":" + BulkOps.insertColumns(db(h), str(type), str(header), (Pointer) buf, bufBytes) + "}");
} catch (final Throwable t) {
return err(t);
}
}

/**
* Columnar insert by pointer: no copy of the columns into one buffer. spec points at 6 longs per column (data address and bytes, aux
* address and bytes, null-bitmap address and bytes, see BulkOps.insertColumns) and nspec is their count. Kinds q d b f as
* arcadedb_n_insert_cols, plus i (int32), e (float32) and s (Arrow UTF-8 string: int32 offsets and the bytes they index), and null bitmaps.
* The caller keeps every buffer alive and unchanged until the call returns. Returns {"ok":true,"n":rows}.
*/
@CEntryPoint(name = "arcadedb_n_insert_colp")
public static CCharPointer insertColp(final IsolateThread thread, final long h, final CCharPointer type, final CCharPointer header,
final CCharPointer spec, final int nspec) {
try {
if (nspec < 0 || nspec % BulkOps.SLOTS != 0)
throw new IllegalArgumentException("address table of " + nspec + " entries is not a multiple of " + BulkOps.SLOTS);
final long[] table = new long[nspec];
for (int i = 0; i < nspec; i++)
table[i] = ((Pointer) spec).readLong(WordFactory.signed(8L * i));
return out("{\"ok\":true,\"n\":" + BulkOps.insertColumns(db(h), str(type), str(header), table) + "}");
} catch (final Throwable t) {
return err(t);
}
Expand Down
162 changes: 124 additions & 38 deletions native/src/main/java/org/humemai/nativelib/BulkOps.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,75 +2,141 @@
* SPDX-License-Identifier: Apache-2.0
*
* Phase 7: columnar bulk insert through one crossing. The documents are built here from whole columns that Python lays out
* in one native-endian buffer (int64, float64, bool bytes, float32 vector matrix), as DocumentBatcher.insertColumns does on
* the JPype path from long[], double[], boolean[] and Object[] columns.
* in native-endian buffers (int64, float64, bool bytes, int32, float32, float32 vector matrix, UTF-8 strings as Arrow offsets plus
* bytes, optional Arrow null bitmaps), as DocumentBatcher.insertColumns does on the JPype path from long[], double[], boolean[] and
* Object[] columns. Values are read straight from the caller's buffers; no JSON text, no parser, no intermediate arrays.
*/
package org.humemai.nativelib;

import com.arcadedb.database.Database;
import com.arcadedb.database.MutableDocument;
import com.arcadedb.schema.VertexType;

import org.graalvm.nativeimage.c.type.CTypeConversion;
import org.graalvm.word.Pointer;
import org.graalvm.word.WordFactory;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;

final class BulkOps {
private BulkOps() {
}

/** Slots per column in the address table: data (address, bytes), aux (address, bytes), nulls (address, bytes). */
static final int SLOTS = 6;

/**
* header: {"n":rows,"commit":every,"cols":[[name,kind,dim],..]} with kind q (int64), d (float64), b (bool byte) or f
* (float32 matrix, dim floats per row); the buffer holds the columns in that order. Same transaction contract as
* DocumentBatcher.insertColumns: batch commits and the rollback on failure apply only to a transaction opened here.
* The original single-buffer form (arcadedb_n_insert_cols): the columns of kind q, d, b or f lie back to back in one buffer, in
* header order. The address table of the pointer form is derived from the layout, so one loop builds the documents for both.
*/
@SuppressWarnings("unchecked")
static long insertColumns(final Database db, final String type, final String header, final ByteBuffer raw) {
static long insertColumns(final Database db, final String type, final String header, final Pointer buf, final long bufBytes) {
final Map<String, Object> h = (Map<String, Object>) Json.parse(header);
final int n = ((Number) h.get("n")).intValue();
final long n = ((Number) h.get("n")).longValue();
final List<Object> cols = (List<Object>) h.get("cols");
final long[] spec = new long[SLOTS * cols.size()];
long pos = 0;
for (int c = 0; c < cols.size(); c++) {
final List<Object> col = (List<Object>) cols.get(c);
final char kind = ((String) col.get(1)).charAt(0);
final long dim = ((Number) col.get(2)).longValue();
final long bytes = switch (kind) {
case 'q', 'd' -> 8 * n;
case 'b' -> n;
case 'f' -> 4 * n * dim;
default -> throw new IllegalArgumentException("unknown column kind " + kind);
};
if (pos + bytes > bufBytes)
throw new IllegalArgumentException("column '" + col.get(0) + "' ends at byte " + (pos + bytes) + ", the buffer has " + bufBytes);
spec[SLOTS * c] = buf.rawValue() + pos;
spec[SLOTS * c + 1] = bytes;
pos += bytes;
}
return insertColumns(db, type, h, spec);
}

/** The pointer form (arcadedb_n_insert_colp): the table holds SLOTS longs per column, see {@link #insertColumns(Database, String, Map, long[])}. */
@SuppressWarnings("unchecked")
static long insertColumns(final Database db, final String type, final String header, final long[] spec) {
return insertColumns(db, type, (Map<String, Object>) Json.parse(header), spec);
}

/**
* header: {"n":rows,"commit":every,"cols":[[name,kind,dim],..]}. Kinds: q (int64), d (float64), b (bool byte), i (int32), e (float32),
* f (float32 matrix, dim floats per row), s (UTF-8 string: int32 offsets[n+1] in the data slot, the bytes they index in the aux slot, as
* an Arrow string array; offsets[0] need not be 0). spec: per column six longs, [data address, data bytes, aux address, aux bytes, null
* bitmap address, null bitmap bytes]; the bitmap is Arrow's (bit i of byte i/8, least significant first; 1 = present), an address of 0
* means no nulls. The memory belongs to the caller for the duration of the call; nothing is kept. Every length is checked here before
* the first document is built, so a wrong table is an exception and nothing is written. Same transaction contract as
* DocumentBatcher.insertColumns: batch commits and the rollback on failure apply only to a transaction opened here.
*/
@SuppressWarnings("unchecked")
private static long insertColumns(final Database db, final String type, final Map<String, Object> h, final long[] spec) {
final long nRows = ((Number) h.get("n")).longValue();
if (nRows < 0 || nRows > Integer.MAX_VALUE)
throw new IllegalArgumentException("row count out of range: " + nRows);
final int n = (int) nRows;
final int commitEvery = ((Number) h.get("commit")).intValue();
final List<Object> cols = (List<Object>) h.get("cols");
final int k = cols.size();
if (spec.length != SLOTS * k)
throw new IllegalArgumentException("address table has " + spec.length + " entries for " + k + " columns, expected " + SLOTS * k);
final String[] names = new String[k];
final char[] kinds = new char[k];
final Object[] data = new Object[k];
final int[] dims = new int[k];
raw.order(ByteOrder.nativeOrder());
int pos = 0;
final long[] data = new long[k]; // addresses: a Word value cannot live in an array or be compared with null
final long[] nulls = new long[k]; // 0 = no nulls
final ByteBuffer[] blobs = new ByteBuffer[k];
int longest = 0;
for (int c = 0; c < k; c++) {
final List<Object> col = (List<Object>) cols.get(c);
names[c] = (String) col.get(0);
kinds[c] = ((String) col.get(1)).charAt(0);
dims[c] = ((Number) col.get(2)).intValue();
final int bytes = switch (kinds[c]) {
case 'q', 'd' -> 8 * n;
final long dataAddr = spec[SLOTS * c], dataBytes = spec[SLOTS * c + 1];
final long auxAddr = spec[SLOTS * c + 2], auxBytes = spec[SLOTS * c + 3];
final long nullAddr = spec[SLOTS * c + 4], nullBytes = spec[SLOTS * c + 5];
final long need = switch (kinds[c]) {
case 'q', 'd' -> 8L * n;
case 'b' -> n;
case 'f' -> 4 * n * dims[c];
case 'i', 'e' -> 4L * n;
case 'f' -> {
if (dims[c] <= 0)
throw new IllegalArgumentException("column '" + names[c] + "': a vector column needs a dimension above 0");
yield 4L * n * dims[c];
}
case 's' -> 4L * (n + 1L);
default -> throw new IllegalArgumentException("unknown column kind " + kinds[c]);
};
final ByteBuffer part = raw.slice(pos, bytes).order(ByteOrder.nativeOrder());
pos += bytes;
switch (kinds[c]) {
case 'q' -> {
final long[] a = new long[n];
part.asLongBuffer().get(a);
data[c] = a;
}
case 'd' -> {
final double[] a = new double[n];
part.asDoubleBuffer().get(a);
data[c] = a;
}
case 'b' -> {
final boolean[] a = new boolean[n];
for (int i = 0; i < n; i++)
a[i] = part.get(i) != 0;
data[c] = a;
if (dataBytes < need || (need > 0 && dataAddr == 0))
throw new IllegalArgumentException("column '" + names[c] + "' has " + dataBytes + " bytes, " + need + " needed");
data[c] = dataAddr;
if (nullAddr != 0) {
if (nullBytes < (n + 7L) / 8)
throw new IllegalArgumentException("column '" + names[c] + "' has a null bitmap of " + nullBytes + " bytes, " + (n + 7L) / 8 + " needed");
nulls[c] = nullAddr;
}
if (kinds[c] == 's') {
if (auxBytes < 0 || auxBytes > Integer.MAX_VALUE || (auxBytes > 0 && auxAddr == 0))
throw new IllegalArgumentException("column '" + names[c] + "': the string bytes are missing or beyond 2 GiB");
blobs[c] = auxBytes == 0 ? ByteBuffer.allocate(0) : CTypeConversion.asByteBuffer(WordFactory.pointer(auxAddr), (int) auxBytes);
final Pointer offsets = WordFactory.pointer(data[c]);
int prev = offsets.readInt(WordFactory.signed(0));
if (prev < 0)
throw new IllegalArgumentException("column '" + names[c] + "': negative string offset");
for (int i = 1; i <= n; i++) {
final int off = offsets.readInt(WordFactory.signed(4L * i));
if (off < prev || off > auxBytes)
throw new IllegalArgumentException("column '" + names[c] + "': string offsets are not increasing within the " + auxBytes + " bytes (row " + (i - 1) + ")");
longest = Math.max(longest, off - prev);
prev = off;
}
default -> data[c] = part.asFloatBuffer();
}
}
final byte[] scratch = new byte[longest];
final boolean vertex = db.getSchema().getType(type) instanceof VertexType;
final boolean wasActive = db.isTransactionActive();
if (!wasActive)
Expand All @@ -79,13 +145,33 @@ static long insertColumns(final Database db, final String type, final String hea
for (int i = 0; i < n; i++) {
final MutableDocument doc = vertex ? db.newVertex(type) : db.newDocument(type);
for (int c = 0; c < k; c++) {
final Pointer base = WordFactory.pointer(data[c]);
if (nulls[c] != 0 && ((WordFactory.<Pointer>pointer(nulls[c]).readByte(WordFactory.signed(i >> 3)) >> (i & 7)) & 1) == 0) {
doc.set(names[c], null);
continue;
}
switch (kinds[c]) {
case 'q' -> doc.set(names[c], ((long[]) data[c])[i]);
case 'd' -> doc.set(names[c], ((double[]) data[c])[i]);
case 'b' -> doc.set(names[c], ((boolean[]) data[c])[i]);
case 'q' -> doc.set(names[c], base.readLong(WordFactory.signed(8L * i)));
case 'd' -> doc.set(names[c], base.readDouble(WordFactory.signed(8L * i)));
case 'b' -> doc.set(names[c], base.readByte(WordFactory.signed(i)) != 0);
case 'i' -> doc.set(names[c], base.readInt(WordFactory.signed(4L * i)));
case 'e' -> doc.set(names[c], base.readFloat(WordFactory.signed(4L * i)));
case 's' -> {
final int from = base.readInt(WordFactory.signed(4L * i));
final int len = base.readInt(WordFactory.signed(4L * (i + 1L))) - from;
if (len == 0)
doc.set(names[c], "");
else {
blobs[c].get(from, scratch, 0, len);
doc.set(names[c], new String(scratch, 0, len, StandardCharsets.UTF_8));
}
}
default -> {
final float[] v = new float[dims[c]];
((java.nio.FloatBuffer) data[c]).get(i * dims[c], v);
final int dim = dims[c];
final float[] v = new float[dim];
final long at = 4L * i * dim;
for (int j = 0; j < dim; j++)
v[j] = base.readFloat(WordFactory.signed(at + 4L * j));
doc.set(names[c], v);
}
}
Expand Down
Loading
Loading