insert_dataframe and insert_arrow: a vertex type loads through the columnar insert (refs #319)

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/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

@@ -1,12 +1,14 @@
 """Columnar bulk insert through one crossing (``arcadedb_n_insert_colp``): whole columns go over by address, with no JSON text. Numpy
 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.
+vector columns, date-times, dates, decimals, bytes and Arrow arrays are laid out by :mod:`.columnar`; anything else (nested values,
+mixed-type columns, 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; 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):