Native insert_columns(parallel=True) takes the columnar entry (closes #325)

native/src/main/java/org/humemai/nativelib/ArcadeNative.java

@@ -561,7 +561,8 @@ public static CCharPointer insertCols(final IsolateThread thread, final long h,
    * 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}.
+   * The caller keeps every buffer alive and unchanged until the call returns. Returns {"ok":true,"n":rows}; with "parallel":true in the
+   * header the records go to the async executor and the reply also carries "fid" (see BulkOps.insertColumnsReply).
    */
   @CEntryPoint(name = "arcadedb_n_insert_colp")
   public static CCharPointer insertColp(final IsolateThread thread, final long h, final CCharPointer type, final CCharPointer header,
@@ -572,7 +573,7 @@ public static CCharPointer insertColp(final IsolateThread thread, final long h,
       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) + "}");
+      return out(BulkOps.insertColumnsReply(db(h), str(type), str(header), table));
     } catch (final Throwable t) {
       return err(t);
     }

native/src/main/java/org/humemai/nativelib/BulkOps.java

@@ -10,6 +10,9 @@
 
 import com.arcadedb.database.Database;
 import com.arcadedb.database.MutableDocument;
+import com.arcadedb.database.async.DatabaseAsyncExecutor;
+import com.arcadedb.database.async.ErrorCallback;
+import com.arcadedb.schema.EdgeType;
 import com.arcadedb.schema.VertexType;
 
 import org.graalvm.nativeimage.c.type.CTypeConversion;
@@ -55,13 +58,25 @@ static long insertColumns(final Database db, final String type, final String hea
       spec[SLOTS * c + 1] = bytes;
       pos += bytes;
     }
-    return insertColumns(db, type, h, spec);
+    return insertColumns(db, type, h, spec, null);
   }
 
-  /** The pointer form (arcadedb_n_insert_colp): the table holds SLOTS longs per column, see {@link #insertColumns(Database, String, Map, long[])}. */
+  /**
+   * The pointer form (arcadedb_n_insert_colp): the table holds SLOTS longs per column, see {@link #insertColumns(Database, String, Map, long[], ErrorCallback)}.
+   * Returns the reply text {"ok":true,"n":rows}. A header with "parallel":true must also carry "fid", a failure token that the caller made
+   * with the "token" operation of arcadedb_n_async: the records go to the database's async executor, the call does not wait for the writers
+   * and neither opens, commits nor rolls back a transaction ("commit" does not apply), exactly as insert_parallel does for rows. The
+   * caller waits for the writers and reads (and so removes) the token with the "failures" operation, also when this call throws or never
+   * returns to it: the token is the caller's, so this call never drops it.
+   */
   @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);
+  static String insertColumnsReply(final Database db, final String type, final String header, final long[] spec) {
+    final Map<String, Object> h = (Map<String, Object>) Json.parse(header);
+    if (!Boolean.TRUE.equals(h.get("parallel")))
+      return "{\"ok\":true,\"n\":" + insertColumns(db, type, h, spec, null) + "}";
+    if (!(h.get("fid") instanceof Number fid))
+      throw new IllegalArgumentException("a parallel columnar insert needs a failure token (header \"fid\", from the \"token\" operation)");
+    return "{\"ok\":true,\"n\":" + insertColumns(db, type, h, spec, NativeExtras.FailureToken.of(fid.longValue()).onError) + "}";
   }
 
   /**
@@ -72,9 +87,13 @@ static long insertColumns(final Database db, final String type, final String hea
    * 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.
+   * onError null: the synchronous mode above. Not null (header "parallel":true): every record goes to db.async().createRecord with that
+   * callback and no transaction is touched, as DocumentBatcher.insertColumnsParallel does; an edge type is refused there, as it is by
+   * insert_parallel. The columns are validated before the first record is handed over in both modes.
    */
   @SuppressWarnings("unchecked")
-  private static long insertColumns(final Database db, final String type, final Map<String, Object> h, final long[] spec) {
+  private static long insertColumns(final Database db, final String type, final Map<String, Object> h, final long[] spec,
+      final ErrorCallback onError) {
     final long nRows = ((Number) h.get("n")).longValue();
     if (nRows < 0 || nRows > Integer.MAX_VALUE)
       throw new IllegalArgumentException("row count out of range: " + nRows);
@@ -137,45 +156,33 @@ private static long insertColumns(final Database db, final String type, final Ma
       }
     }
     final byte[] scratch = new byte[longest];
-    final boolean vertex = db.getSchema().getType(type) instanceof VertexType;
+    final Object schemaType = db.getSchema().getType(type);
+    final boolean vertex = schemaType instanceof VertexType;
+    if (onError != null) {
+      if (schemaType instanceof EdgeType)
+        throw new IllegalArgumentException("Cannot bulk-insert into '" + type
+            + "' because it is an edge type; load the vertices, then create the edges (graph_batch().new_edges() or CREATE EDGE)");
+      final DatabaseAsyncExecutor async = db.async();
+      int handed = 0;
+      try {
+        for (; handed < n; handed++) {
+          final MutableDocument doc = vertex ? db.newVertex(type) : db.newDocument(type);
+          setRow(doc, handed, names, kinds, dims, data, nulls, blobs, scratch);
+          async.createRecord(doc, null, onError);
+        }
+      } catch (final Throwable t) {
+        // the records handed over before the failure are the writers' now, whatever this call does
+        throw new IllegalStateException("after handing " + handed + " of " + n + " rows to the async writers (those may have been stored): " + t, t);
+      }
+      return n;
+    }
     final boolean wasActive = db.isTransactionActive();
     if (!wasActive)
       db.begin();
     try {
       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], 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 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);
-            }
-          }
-        }
+        setRow(doc, i, names, kinds, dims, data, nulls, blobs, scratch);
         doc.save();
         if (!wasActive && commitEvery > 0 && (i + 1) % commitEvery == 0) {
           db.commit();
@@ -197,6 +204,44 @@ private static long insertColumns(final Database db, final String type, final Ma
     return n;
   }
 
+  /** Sets the properties of row i on doc from the columns: the one place that reads a value out of a buffer. */
+  private static void setRow(final MutableDocument doc, final int i, final String[] names, final char[] kinds, final int[] dims,
+      final long[] data, final long[] nulls, final ByteBuffer[] blobs, final byte[] scratch) {
+    final int k = names.length;
+    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], 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 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);
+        }
+      }
+    }
+  }
+
   /**
    * Documents (or vertices of a vertex type) from rows [{property: value},..] in the UTF-8 JSON text b[0, n): one record per row through
    * newDocument/newVertex, set per property and save, as DocumentBatcher.insertManyJson does on the JPype path; no transaction is opened

native/src/main/java/org/humemai/nativelib/NativeExtras.java

@@ -527,6 +527,51 @@ else if (opName.equals("create_vertices"))
   private static final Map<Long, AtomicLong> ASYNC_FAILS = new ConcurrentHashMap<>();
   private static final Map<Long, AtomicReference<Throwable>> ASYNC_FIRST = new ConcurrentHashMap<>();
 
+  /**
+   * A failure token for one parallel load: the id the caller reads (and so removes) with the "failures" operation, and the callback that
+   * counts. The caller creates it before the load ("token" operation) and passes the id in the load's header, so that it holds the id
+   * whatever happens to the call, an interrupt included, and always has the means to wait for the writers and remove the entries.
+   */
+  static final class FailureToken {
+    final long fid;
+    final AtomicLong count;
+    final AtomicReference<Throwable> first;
+    final ErrorCallback onError;
+
+    private FailureToken(final long fid, final AtomicLong count, final AtomicReference<Throwable> first) {
+      this.fid = fid;
+      this.count = count;
+      this.first = first;
+      this.onError = t -> {
+        count.incrementAndGet();
+        first.compareAndSet(null, t);
+      };
+    }
+
+    /** A new token, registered for the "failures" operation. */
+    static FailureToken create() {
+      final FailureToken token = new FailureToken(ArcadeNative.IDS.getAndIncrement(), new AtomicLong(), new AtomicReference<>());
+      ASYNC_FAILS.put(token.fid, token.count);
+      ASYNC_FIRST.put(token.fid, token.first);
+      return token;
+    }
+
+    /** The token that "token" created, for the load that its id was passed to. */
+    static FailureToken of(final long fid) {
+      final AtomicLong count = ASYNC_FAILS.get(fid);
+      final AtomicReference<Throwable> first = ASYNC_FIRST.get(fid);
+      if (count == null || first == null)
+        throw new IllegalArgumentException("unknown failure token " + fid + " (already read, or never created)");
+      return new FailureToken(fid, count, first);
+    }
+
+    /** The load failed before it returned: the caller gets the error, not a token to read failures from. */
+    void drop() {
+      ASYNC_FAILS.remove(fid);
+      ASYNC_FIRST.remove(fid);
+    }
+  }
+
   private static String replyRows(final ResultSet rs) {
     final StringBuilder sb = new StringBuilder("{\"ok\":true,\"rs\":0,");
     NativeCodec.appendBatch(sb, rs, Integer.MAX_VALUE);
@@ -824,15 +869,9 @@ else if (a.get("slot") instanceof Number slot)
         ex.appendSamples((String) a.get("type"), stamps, columns);
       }
       case "insert_parallel" -> {
-        final long fid = ArcadeNative.IDS.getAndIncrement();
-        final AtomicLong count = new AtomicLong();
-        final AtomicReference<Throwable> first = new AtomicReference<>();
-        ASYNC_FAILS.put(fid, count);
-        ASYNC_FIRST.put(fid, first);
-        final ErrorCallback onError = t -> {
-          count.incrementAndGet();
-          first.compareAndSet(null, t);
-        };
+        final FailureToken token = FailureToken.create();
+        final long fid = token.fid;
+        final ErrorCallback onError = token.onError;
         final String type = (String) a.get("type");
         try {
           final List<Object> rows = (List<Object>) a.get("rows");
@@ -846,12 +885,14 @@ else if (a.get("slot") instanceof Number slot)
             ex.createRecord(doc, null, onError);
           }
         } catch (final Throwable t) {
-          ASYNC_FAILS.remove(fid); // the caller gets the error, not a token to read failures from
-          ASYNC_FIRST.remove(fid);
+          token.drop(); // the caller gets the error, not a token to read failures from
           throw t;
         }
         return ArcadeNative.out("{\"ok\":true,\"fid\":" + fid + "}");
       }
+      case "token" -> {
+        return ArcadeNative.out("{\"ok\":true,\"fid\":" + FailureToken.create().fid + "}");
+      }
       case "failures" -> {
         final long fid = ((Number) a.get("fid")).longValue();
         final AtomicLong count = ASYNC_FAILS.remove(fid);

src/arcadedb/_native/bulk.py

@@ -2,11 +2,13 @@
 ints, floats and bools, lists of one value type (None is a null), strings (UTF-8 bytes plus offsets, as Arrow lays them out), float32
 vector columns and Arrow arrays are laid out by :mod:`.columnar`; anything else (nested values, mixed-type columns, date-times,
 decimals, integers past 64 bits) returns None and the caller uses the per-row JSON path. The documents are built image-side as
-``DocumentBatcher.insertColumns`` builds them on JPype. ``ARCADEDB_NATIVE_BULK=0`` turns the columnar path off.
+``DocumentBatcher.insertColumns`` builds them on JPype; with ``parallel`` they are handed to the async writers instead of a transaction
+(``DocumentBatcher.insertColumnsParallel``). ``ARCADEDB_NATIVE_BULK=0`` turns the columnar path off.
 """
 
 import os
 
+from ..exceptions import ArcadeDBError
 from . import columnar, lib
 
 _ENABLED = os.environ.get("ARCADEDB_NATIVE_BULK", "1") != "0"
@@ -21,20 +23,19 @@ def _usable():
     )
 
 
-def _send(db, type_name, packed, rows, commit_every, prefix):
+def _send(db, type_name, packed, rows, commit_every, prefix, fid=None):
     from .database import _call
 
     spec = columnar.spec_of(packed)
-    reply = _call(
+    return _call(
         prefix,
         "arcadedb_n_insert_colp",
         db._h,
         lib.b(type_name),
-        lib.b(columnar.header_of(rows, commit_every, packed)),
+        lib.b(columnar.header_of(rows, commit_every, packed, fid)),
         spec,
         len(spec),
     )  # `packed` holds the arrays the addresses point into until the call returns
-    return int(reply["n"])
 
 
 def insert_columns_fast(
@@ -51,7 +52,32 @@ def insert_columns_fast(
     if packed is None:
         return None
     rows, cols = packed
-    return _send(db, type_name, cols, rows, commit_every, prefix)
+    return int(_send(db, type_name, cols, rows, commit_every, prefix)["n"])
+
+
+def insert_columns_parallel(db, type_name, columns, declared=None):
+    """Hand ``columns`` (name -> column) to the async writers in one crossing, wait for them and return the row count, or None when a
+    column cannot be sent this way (nothing was handed over then). A record that a writer rejects raises
+    (``Database._finish_parallel``), as for ``insert_many(parallel=True)``, and so does an interrupt, after the writers have finished:
+    the failure token is made before the crossing, so the call always has the id to wait for and to read.
+    """
+    if not _usable():
+        return None
+    packed = columnar.pack(columns, columnar.Declared(db, type_name, declared))
+    if packed is None:
+        return None
+    rows, cols = packed
+    from . import extras
+
+    fid = extras.failure_token(db)
+    try:
+        _send(db, type_name, cols, rows, 0, "Failed to bulk-insert", fid=fid)
+    except ArcadeDBError as e:
+        raise db._parallel_load_failed(type_name, e, [fid]) from e
+    except BaseException:
+        db._abandon_parallel([fid])  # a ^C: the writers still run, so wait for them
+        raise
+    return db._finish_parallel(type_name, [fid], rows)
 
 
 def insert_rows_columnar(db, type_name, chunk):

src/arcadedb/_native/columnar.py

@@ -343,17 +343,22 @@ def spec_of(packed):
     return (ctypes.c_int64 * len(flat))(*flat)
 
 
-def header_of(rows, commit_every, packed):
-    """The JSON header text: row count, commit interval and ``[name, kind, dim]`` per column."""
+def header_of(rows, commit_every, packed, fid=None):
+    """The JSON header text: row count, commit interval and ``[name, kind, dim]`` per column; with a failure token ``fid`` it also says
+    ``"parallel": true``, which hands the records to the async executor (the image then ignores the commit interval and counts the
+    failed records on that token).
+    """
     import json
 
-    return json.dumps(
-        {
-            "n": rows,
-            "commit": min(int(commit_every), 2**31 - 1),
-            "cols": [[c.name, c.kind, c.dim] for c in packed],
-        }
-    )
+    header = {
+        "n": rows,
+        "commit": min(int(commit_every), 2**31 - 1),
+        "cols": [[c.name, c.kind, c.dim] for c in packed],
+    }
+    if fid is not None:
+        header["parallel"] = True
+        header["fid"] = fid
+    return json.dumps(header)
 
 
 def rows_to_columns(rows):