Native columnar insert: date-time, date, decimal and bytes columns stay columnar (refs #316)

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,14 +10,21 @@
 
 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;
 import org.graalvm.word.Pointer;
 import org.graalvm.word.WordFactory;
 
+import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
 import java.util.List;
 import java.util.Map;
 
@@ -55,26 +62,45 @@ 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) + "}";
   }
 
   /**
    * 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
+   * an Arrow string array; offsets[0] need not be 0). The kinds that the JSON rows carry as tagged values (Json.decodeTag), read as the same
+   * objects: T (int64 ticks since 1970-01-01T00:00:00 of the UTC wall clock, dim = ticks per second: 1, 1000, 10^6 or 10^9, read as a
+   * LocalDateTime), D (int32 days since the epoch, a LocalDate), n (a decimal as text in the string layout, a BigDecimal) and x (bytes in
+   * the string layout, a byte[]), so a column of these is stored as the row path stores it. 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.
+   * 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);
@@ -102,13 +128,18 @@ private static long insertColumns(final Database db, final String type, final Ma
       final long need = switch (kinds[c]) {
         case 'q', 'd' -> 8L * n;
         case 'b' -> n;
-        case 'i', 'e' -> 4L * n;
+        case 'i', 'e', 'D' -> 4L * n;
+        case 'T' -> {
+          if (dims[c] != 1 && dims[c] != 1_000 && dims[c] != 1_000_000 && dims[c] != 1_000_000_000)
+            throw new IllegalArgumentException("column '" + names[c] + "': a date-time column has 1, 1000, 10^6 or 10^9 ticks per second, not " + dims[c]);
+          yield 8L * 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);
+        case 's', 'n', 'x' -> 4L * (n + 1L);
         default -> throw new IllegalArgumentException("unknown column kind " + kinds[c]);
       };
       if (dataBytes < need || (need > 0 && dataAddr == 0))
@@ -119,7 +150,7 @@ private static long insertColumns(final Database db, final String type, final Ma
           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 (kinds[c] == 's' || kinds[c] == 'n' || kinds[c] == 'x') {
         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);
@@ -137,45 +168,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 +216,63 @@ 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 'T' -> {
+          final long ticks = base.readLong(WordFactory.signed(8L * i));
+          final long perSecond = dims[c];
+          doc.set(names[c], LocalDateTime.ofEpochSecond(Math.floorDiv(ticks, perSecond),
+              (int) (Math.floorMod(ticks, perSecond) * (1_000_000_000L / perSecond)), ZoneOffset.UTC));
+        }
+        case 'D' -> doc.set(names[c], LocalDate.ofEpochDay(base.readInt(WordFactory.signed(4L * i))));
+        case 'n' -> {
+          final int from = base.readInt(WordFactory.signed(4L * i));
+          final int len = base.readInt(WordFactory.signed(4L * (i + 1L))) - from;
+          blobs[c].get(from, scratch, 0, len);
+          doc.set(names[c], new BigDecimal(new String(scratch, 0, len, StandardCharsets.US_ASCII)));
+        }
+        case 'x' -> {
+          final int from = base.readInt(WordFactory.signed(4L * i));
+          final byte[] bytes = new byte[base.readInt(WordFactory.signed(4L * (i + 1L))) - from];
+          blobs[c].get(from, bytes, 0, bytes.length);
+          doc.set(names[c], bytes);
+        }
+        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);