diff --git a/native/src/main/java/org/humemai/nativelib/ArcadeNative.java b/native/src/main/java/org/humemai/nativelib/ArcadeNative.java
index 9b5acb91..1703b965 100644
--- a/native/src/main/java/org/humemai/nativelib/ArcadeNative.java
+++ b/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);
     }
diff --git a/native/src/main/java/org/humemai/nativelib/BulkOps.java b/native/src/main/java/org/humemai/nativelib/BulkOps.java
index 37cedf90..00798947 100644
--- a/native/src/main/java/org/humemai/nativelib/BulkOps.java
+++ b/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
diff --git a/native/src/main/java/org/humemai/nativelib/NativeExtras.java b/native/src/main/java/org/humemai/nativelib/NativeExtras.java
index a62574ad..404181b5 100644
--- a/native/src/main/java/org/humemai/nativelib/NativeExtras.java
+++ b/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);
diff --git a/src/arcadedb/_native/bulk.py b/src/arcadedb/_native/bulk.py
index d244c48c..c16d225b 100644
--- a/src/arcadedb/_native/bulk.py
+++ b/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):
diff --git a/src/arcadedb/_native/columnar.py b/src/arcadedb/_native/columnar.py
index f288043c..d001b3da 100644
--- a/src/arcadedb/_native/columnar.py
+++ b/src/arcadedb/_native/columnar.py
@@ -2,19 +2,29 @@
 
 A column crosses as raw memory, not as JSON text: the image reads each buffer in place and builds the documents without a parser. This
 module only decides how a column is laid out and hands the addresses over; it needs no native library, so the layouts are unit-tested
-without one. A column that cannot be laid out (a nested value, a mixed-type column, a date-time, an integer past 64 bits, a string with
-a lone surrogate) makes :func:`pack` return ``None`` and the caller uses the JSON path for that call; nothing is half sent.
+without one. A column that cannot be laid out (a nested value, a mixed-type column, a numpy datetime64, an integer past 64 bits, a string
+with a lone surrogate) makes :func:`pack` return ``None`` and the caller uses the JSON path for that call; nothing is half sent.
 
 Kinds (the image's, ``BulkOps.insertColumns``): ``q`` int64, ``d`` float64, ``b`` bool byte, ``i`` int32, ``e`` float32, ``f`` float32
 matrix (``dim`` per row), ``s`` UTF-8 string (``int32 offsets[n+1]`` and the bytes they index: Arrow's string layout). A column may
 carry a null bitmap (Arrow's: bit ``i`` of byte ``i // 8``, least significant first, 1 = present).
 
+Temporal, decimal and binary kinds (the values the JSON path sends as tagged values, so the stored value is the same object): ``T`` a
+date-time, int64 ticks since 1970-01-01T00:00:00 of the UTC wall clock, ``dim`` ticks per second (1, 1000, 10**6 or 10**9: an Arrow
+timestamp keeps its unit, a Python ``datetime`` is microseconds), read as ``LocalDateTime``; ``D`` a date, int32 days since the epoch,
+read as ``LocalDate``; ``n`` a decimal, the string layout holding ``str(Decimal)``, read as ``BigDecimal``; ``x`` bytes, the string
+layout, read as ``byte[]``. A time-zone-aware ``datetime`` or Arrow timestamp is its UTC wall clock and a naive one is taken as it is,
+the rule of ``_tag_datetime``; nothing here decides a time zone.
+
 ``i`` and ``e`` and the float matrix from a list of lists are chosen only for a property declared ``INTEGER``, ``FLOAT`` and
 ``ARRAY_OF_FLOATS``: the engine would convert to those types anyway, so the stored value is the one the JSON path stores. For an
 undeclared property an ``int32`` column crosses as ``q`` (a Long) and a ``float32`` column as ``d`` (a Double), as the JSON path stores
 them.
 """
 
+from datetime import date, datetime, timezone
+from decimal import Decimal
+
 from .schema_objects import (  # the schema-entry reader shared with NativeType (#195)
     _name_type,
 )
@@ -27,6 +37,16 @@
 _INT64_MAX = 2**63 - 1
 _INT32_MAX = 2**31 - 1
 _NoneType = type(None)
+_EPOCH_ORDINAL = date(1970, 1, 1).toordinal()
+_UTC = timezone.utc
+_BLOB_KINDS = (
+    "snx"  # the kinds laid out as int32 offsets[n + 1] plus the bytes they index
+)
+_TICKS = {"s": 1, "ms": 1_000, "us": 1_000_000, "ns": 1_000_000_000}
+# the seconds that java.time.LocalDateTime.ofEpochSecond takes (years -999999999 to 999999999): beyond them it throws in the image, in the
+# middle of a batch. Only a timestamp[s] can pass them: an int64 of milliseconds is at most 9.2e15 seconds
+_LDT_MIN_SECONDS = -31_557_014_167_219_200
+_LDT_MAX_SECONDS = 31_556_889_864_403_199
 
 
 class Packed:
@@ -39,8 +59,8 @@ def __init__(self, name, kind, dim, array, aux=None, nulls=None, keep=()):
         self.kind = kind
         self.dim = dim
         self.n = int(array.shape[0]) - (
-            1 if kind == "s" else 0
-        )  # a string column's array is its n + 1 offsets
+            1 if kind in _BLOB_KINDS else 0
+        )  # a string, decimal or bytes column's array is its n + 1 offsets
         self.data = (array.ctypes.data, array.nbytes)
         self.aux = aux or (0, 0)
         self.nulls = (nulls.ctypes.data, nulls.nbytes) if nulls is not None else (0, 0)
@@ -166,6 +186,35 @@ def _string_column(name, texts, nulls=None):
     return Packed(name, "s", 0, offsets, aux=aux, nulls=nulls, keep=(blob_array, data))
 
 
+def _blob_column(name, kind, items, nulls=None):
+    """A list of bytes (a null already replaced by b'') as an offsets-plus-bytes column of ``kind``, or None past 2 GiB."""
+    n = len(items)
+    data = b"".join(items)
+    if len(data) > _INT32_MAX:
+        return None
+    lengths = _np.fromiter(map(len, items), dtype=_np.int64, count=n)
+    offsets = _np.zeros(n + 1, dtype=_np.int32)
+    _np.cumsum(lengths, out=offsets[1:], dtype=_np.int32)
+    blob_array = (
+        _np.frombuffer(data, dtype=_np.uint8) if data else _np.zeros(0, dtype=_np.uint8)
+    )
+    aux = (blob_array.ctypes.data, blob_array.nbytes) if data else (0, 0)
+    return Packed(name, kind, 0, offsets, aux=aux, nulls=nulls, keep=(blob_array, data))
+
+
+def _epoch_micros(value):
+    """Microseconds since 1970-01-01T00:00:00 of the UTC wall clock of a ``datetime`` (a naive one is taken as it is, as ``_tag_datetime``)."""
+    if value.utcoffset() is not None:
+        value = value.astimezone(_UTC)
+    seconds = (
+        (value.toordinal() - _EPOCH_ORDINAL) * 86400
+        + value.hour * 3600
+        + value.minute * 60
+        + value.second
+    )
+    return seconds * 1_000_000 + value.microsecond
+
+
 def _list_column(name, values, declared):
     """A list or tuple of Python values as a column: one type (None allowed) or a list of equal-length float arrays; else None."""
     n = len(values)
@@ -203,6 +252,33 @@ def _list_column(name, values, declared):
             return None  # an integer beyond 64 bits
         kind = "q" if only is int else "d" if only is float else "b"
         return Packed(name, kind, 0, arr, nulls=nulls)
+    if issubclass(only, datetime):
+        if only.__name__ == "NaTType":
+            return None  # the JSON path stores it as a null
+        try:
+            ticks = [0 if v is None else _epoch_micros(v) for v in values]
+        except (OverflowError, ValueError):
+            return None  # the JSON path raises what it raised before
+        return Packed(
+            name, "T", 1_000_000, _np.array(ticks, dtype=_np.int64), nulls=nulls
+        )
+    if only is date:
+        days = [0 if v is None else v.toordinal() - _EPOCH_ORDINAL for v in values]
+        return Packed(name, "D", 0, _np.array(days, dtype=_np.int32), nulls=nulls)
+    if only is Decimal:
+        texts = []
+        for v in values:
+            if v is None:
+                texts.append(b"")
+            elif v.is_finite():
+                texts.append(str(v).encode("ascii"))
+            else:
+                return None  # the JSON path sends a non-finite Decimal down the per-row path
+        return _blob_column(name, "n", texts, nulls)
+    if only is bytes or only is bytearray:
+        return _blob_column(
+            name, "x", values if present is None else [v or b"" for v in values], nulls
+        )
     if only is list and declared.get(name) == "ARRAY_OF_FLOATS" and present is None:
         try:
             arr = _np.array(values, dtype=_np.float32)
@@ -248,10 +324,11 @@ def _arrow_column(name, arr, declared):
         if arr.null_count
         else None
     )
-    if types.is_string(t) or types.is_large_string(t):
+    is_text = types.is_string(t) or types.is_large_string(t)
+    if is_text or types.is_binary(t) or types.is_large_binary(t):
         bufs = arr.buffers()
         off_buf, data_buf = bufs[1], bufs[2]
-        if types.is_string(t):
+        if types.is_string(t) or types.is_binary(t):
             offsets = _np.frombuffer(off_buf, dtype=_np.int32)[
                 arr.offset : arr.offset + n + 1
             ]
@@ -266,8 +343,42 @@ def _arrow_column(name, arr, declared):
         size = 0 if data_buf is None else min(data_buf.size, _INT32_MAX)
         aux = (data_buf.address, size) if size else (0, 0)
         return Packed(
-            name, "s", 0, offsets, aux=aux, nulls=nulls, keep=(arr, off_buf, data_buf)
+            name,
+            "s" if is_text else "x",
+            0,
+            offsets,
+            aux=aux,
+            nulls=nulls,
+            keep=(arr, off_buf, data_buf),
+        )
+    if types.is_timestamp(t):
+        values = arr.cast(pa.int64()).fill_null(0).to_numpy(zero_copy_only=False)
+        if t.unit == "s" and values.size:
+            if values.min() < _LDT_MIN_SECONDS or values.max() > _LDT_MAX_SECONDS:
+                return None  # the row path raises its usual error for it, before anything is stored
+        return Packed(
+            name,
+            "T",
+            _TICKS[t.unit],
+            _np.ascontiguousarray(values, dtype=_np.int64),
+            nulls=nulls,
+        )
+    if types.is_date(t):
+        if types.is_date64(t):  # milliseconds since the epoch, floor-divided to days
+            raw = arr.cast(pa.int64()).fill_null(0).to_numpy(zero_copy_only=False)
+            days = raw // 86_400_000
+            if days.size and (
+                days.min() < -(_INT32_MAX + 1) or days.max() > _INT32_MAX
+            ):
+                return None  # int32 days would wrap: the row path raises its usual error for it
+        else:
+            days = arr.cast(pa.int32()).fill_null(0).to_numpy(zero_copy_only=False)
+        return Packed(
+            name, "D", 0, _np.ascontiguousarray(days, dtype=_np.int32), nulls=nulls
         )
+    if types.is_decimal(t):
+        texts = [b"" if v is None else str(v).encode("ascii") for v in arr.to_pylist()]
+        return _blob_column(name, "n", texts, nulls)
     if types.is_integer(t) or types.is_floating(t) or types.is_boolean(t):
         if arr.null_count:
             arr = arr.fill_null(False if types.is_boolean(t) else 0)
@@ -343,17 +454,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):
diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py
index f30dc526..3ee0f51b 100644
--- a/src/arcadedb/_native/database.py
+++ b/src/arcadedb/_native/database.py
@@ -877,7 +877,8 @@ def insert_many(
         a newDocument/set/save loop inside the image, as ``DocumentBatcher.insertManyJson`` on the JPype path). A ``datetime`` (also
         ``pandas.Timestamp`` and ``numpy.datetime64``), ``date``, ``Decimal``, ``bytes``/``bytearray``, numpy scalar, ``pandas.NA`` or
         ``pandas.NaT`` sitting directly in a row crosses in that text as a tagged value (``{"\\u0001": [kind, payload]}``, decoded by
-        ``Json.java``), so it does not move the chunk off this path; such a chunk loses the columnar entry. A row with a value the JSON
+        ``Json.java``), so it does not move the chunk off this path; but when a column holds only date-times, dates, Decimals or
+        bytes (None is a null) the chunk goes as columns, the same stored objects. A row with a value the JSON
         text cannot carry unchanged (NaN or Infinity, an integer beyond 64 bits, a date-time or decimal inside a list or map, a non-str
         key, a lone surrogate, a ``datetime.time``) goes through ``INSERT INTO t CONTENT :row`` on its own, in one crossing for a run of
         such rows; the rows around it keep the bulk path and every row is inserted in the order given. Inside the caller's
@@ -995,20 +996,50 @@ def _command_many(self, language: str, command: str, rows) -> int:
             raise _error("Command failed", reply)
         return reply["n"]
 
-    def _parallel_load_failed(self, type_name, error) -> ArcadeDBError:
-        """The error for a parallel load that failed while handing rows to the writers: waits for the queue to drain first, so
-        the caller sees a final state, and says that the rows handed over before the failure may have been stored.
+    def _drain_parallel(self, tokens):
+        """Waits for the async writers, then reads, and so removes, the failure token of every parallel load handed in. Returns
+        ``(problem, failed, first)``: the error of the wait or of a read (or None), the failed records the writers reported on all the
+        tokens and the first failure's text. The tokens are read whatever the wait did: an unread token keeps its entries for good.
         """
+        from . import extras
+
+        problem = None
         try:
             self._async().wait_completion()
-        except Exception:  # nosec B110 - the original error is the one to report
-            pass
+        except Exception as e:
+            problem = e
+        failed, first = 0, None
+        for token in tokens:
+            try:
+                result = extras.failures(self, token)
+            except Exception as e:
+                problem = problem or e
+                continue
+            if result["count"] and first is None:
+                first = result["first"]
+            failed += result["count"]
+        return problem, failed, first
+
+    def _abandon_parallel(self, tokens):
+        """A parallel load that is being interrupted: waits for the writers (they run on) and removes the tokens; the error that is on its
+        way out is the one to report.
+        """
+        self._drain_parallel(tokens)
+
+    def _parallel_load_failed(self, type_name, error, tokens=()) -> ArcadeDBError:
+        """The error for a parallel load that failed while handing rows to the writers: waits for the queue to drain first, so
+        the caller sees a final state, and says that the rows handed over before the failure may have been stored.
+        """
+        _, failed, _ = self._drain_parallel(tokens)
         text = str(error)
         prefix = "Failed to bulk-insert: "
         if text.startswith(prefix):
             text = text[len(prefix) :]
+        reported = (
+            f"; the writers had reported {failed} failed record(s)" if failed else ""
+        )
         return ArcadeDBError(
-            f"Failed to bulk-insert into '{type_name}': {text}; the rows handed to the "
+            f"Failed to bulk-insert into '{type_name}': {text}{reported}; the rows handed to the "
             f"parallel writers before the failure may have been stored (the call waited "
             f"for them before raising)"
         )
@@ -1030,26 +1061,28 @@ def _insert_many_parallel(self, type_name, rows) -> int:
                     tokens.append(extras.insert_parallel(self, type_name, chunk)["fid"])
                 except ArcadeDBError as e:
                     # as the JPype path: the rows handed to the writers before the failure are written whatever this call does
-                    raise self._parallel_load_failed(type_name, e) from e
+                    raise self._parallel_load_failed(type_name, e, tokens) from e
                 n += len(chunk)
         except ArcadeDBError:
             raise
         except BaseException:
-            try:
-                self._async().wait_completion()
-            except Exception:  # nosec B110 - the original error is the one to report
-                pass
+            self._abandon_parallel(tokens)
             raise
+        return self._finish_parallel(type_name, tokens, n)
+
+    def _finish_parallel(self, type_name, tokens, n) -> int:
+        """The end of a parallel load: wait for the async writers, then read the failure count of every token and raise when a record failed."""
         try:
-            self._async().wait_completion()
-        except Exception as e:
-            raise ArcadeDBError(f"Failed to bulk-insert into '{type_name}': {e}") from e
-        n_failed, first = 0, None
-        for token in tokens:
-            result = extras.failures(self, token)
-            if result["count"] and first is None:
-                first = result["first"]
-            n_failed += result["count"]
+            problem, n_failed, first = self._drain_parallel(tokens)
+        except BaseException:
+            self._abandon_parallel(
+                tokens
+            )  # a ^C in the wait: finish it, and remove the tokens
+            raise
+        if problem is not None:
+            raise ArcadeDBError(
+                f"Failed to bulk-insert into '{type_name}': {problem}"
+            ) from problem
         if n_failed:
             raise ArcadeDBError(
                 f"Failed to bulk-insert into '{type_name}': the parallel writers reported {n_failed} failed record(s) "
@@ -1314,7 +1347,11 @@ def export_to_csv(
     def insert_columns(
         self, type_name, columns, commit_every: int = 10_000, parallel: bool = False
     ) -> int:
-        """Bulk insert from whole columns (dict of name -> numpy array or list), as rows through :meth:`insert_many`."""
+        """Bulk insert from whole columns (dict of name -> numpy array or list): columns the buffers can carry cross in one call
+        (``arcadedb_n_insert_colp``), any other column set goes as rows through :meth:`insert_many`. With ``parallel=True`` the
+        same one call hands the records to the async writers (no transaction, ``commit_every`` does not apply, the call waits for the
+        writers and raises when a record failed), as ``insert_many(parallel=True)`` does for rows.
+        """
         self._check_not_closed()
         items = list(columns.items()) if hasattr(columns, "items") else None
         if not items:
@@ -1330,7 +1367,13 @@ def insert_columns(
         known = {n: len(v) for n, v in items if hasattr(v, "__len__")}
         if len(set(known.values())) > 1:
             raise ValueError(f"columns differ in length: {known}")
-        if not parallel:
+        if parallel:
+            from .bulk import insert_columns_parallel
+
+            done = insert_columns_parallel(self, type_name, columns)
+            if done is not None:
+                return done
+        else:
             from .bulk import insert_columns_fast
 
             done = insert_columns_fast(self, type_name, columns, commit_every)
diff --git a/src/arcadedb/_native/extras.py b/src/arcadedb/_native/extras.py
index 5f8a4702..439fb9ef 100644
--- a/src/arcadedb/_native/extras.py
+++ b/src/arcadedb/_native/extras.py
@@ -915,6 +915,19 @@ def insert_parallel(database, type_name: str, rows) -> dict:
     )
 
 
+def failure_token(database) -> int:
+    """A new failure token for one parallel load: the caller makes it before the load, hands its id to the load and reads it with
+    :func:`failures` after the final wait, so it holds the id whatever happens to the load (an interrupt included).
+    """
+    return _call(
+        "Failed to make a failure token",
+        "arcadedb_n_async",
+        database._h,
+        b"token",
+        lib.b(_enc({})),
+    )["fid"]
+
+
 def failures(database, token: int) -> dict:
     return _call(
         "Failed to read the parallel writers",
diff --git a/src/arcadedb/dataframe.py b/src/arcadedb/dataframe.py
index 9d3ee0b7..37c67a29 100644
--- a/src/arcadedb/dataframe.py
+++ b/src/arcadedb/dataframe.py
@@ -325,8 +325,10 @@ def _column_values(name: str, chunk, pa, native: bool = False, arrow: bool = Fal
     """One Arrow array as what ``Database.insert_columns`` takes: a numpy array when the column crosses as one
     buffer (integers, floats and booleans without nulls), else a list of values (``None`` is a null).
 
-    ``arrow`` (the native backend's document load): a string column, and a numeric one with nulls, stay Arrow arrays, which
-    the columnar entry reads in place (the offsets, the bytes and the validity bitmap) instead of as Python values.
+    ``arrow`` (the native backend's document load): a string, binary, timestamp, date or decimal column, and a numeric one with
+    nulls, stay Arrow arrays, which the columnar entry reads in place (the offsets, the bytes, the integer ticks and the validity
+    bitmap) instead of as Python values; ``_insert_documents_native`` turns them into tagged values for the row path when a batch
+    cannot go columnar.
     """
     types = pa.types
     if types.is_dictionary(chunk.type):  # a pandas categorical or a Polars Categorical
@@ -361,6 +363,8 @@ def _column_values(name: str, chunk, pa, native: bool = False, arrow: bool = Fal
                 values = values.astype("float64")  # float16/32 widen exactly
             return values
         return chunk.to_pylist()
+    if arrow and kind in ("timestamp", "date", "decimal", "binary"):
+        return chunk
     if kind in ("timestamp", "date"):
         return (
             _temporal_to_tagged(chunk, pa) if native else _temporal_to_java(chunk, pa)
@@ -421,8 +425,12 @@ def insert_table(
     native = _is_native(db)
     written = 0
     # a document load inside the caller's own transaction joins it (insert_columns commits
-    # nothing then); a vertex load refuses to run inside one
-    in_caller_transaction = (not is_vertex) and bool(db.is_transaction_active())
+    # nothing then); a vertex load refuses to run inside one. parallel=True never joins it: the
+    # async writers commit the records themselves, so a rollback of the caller's transaction does
+    # not remove them
+    in_caller_transaction = (
+        (not is_vertex) and (not parallel) and bool(db.is_transaction_active())
+    )
     graph_batch = db.graph_batch() if is_vertex else None
     try:
         for batch in table.to_batches(max_chunksize=batch_size):
@@ -434,7 +442,7 @@ def insert_table(
                     batch.column(i),
                     pa,
                     native,
-                    arrow=native and not is_vertex and not parallel,
+                    arrow=native and not is_vertex,
                 )
                 for i, name in enumerate(names)
             }
@@ -459,6 +467,13 @@ def insert_table(
                         f"{exc}; the {written} row(s) of the batches before this one "
                         f"were written in your transaction, not committed"
                     ) from exc
+                if written and parallel:
+                    # the async writers commit on their own, inside the caller's transaction or not
+                    raise ArcadeDBError(
+                        f"{exc}; the {written} row(s) of the batches before this one "
+                        f"were handed to the async writers, which commit on their own: "
+                        f"a rollback of your transaction does not remove them"
+                    ) from exc
                 if written:
                     raise ArcadeDBError(
                         f"{exc}; the {written} row(s) of the batches before this one "
@@ -485,6 +500,20 @@ def _insert_vertices(
     )
 
 
+def _arrow_row_values(col) -> list:
+    """A column kept as an Arrow array, as the values the row path takes: Python values, or tagged values for a timestamp, date,
+    decimal or binary column (the form it had before the columnar entry took them, so the stored value is unchanged).
+    """
+    import pyarrow as pa
+
+    kind = _kind(pa, col.type)
+    if kind in ("timestamp", "date"):
+        return _temporal_to_tagged(col, pa)
+    if kind in ("decimal", "binary"):
+        return _scalars_to_tagged(kind, col)
+    return col.to_pylist()
+
+
 def _insert_documents_native(
     db,
     type_name: str,
@@ -497,9 +526,20 @@ def _insert_documents_native(
     """One batch of documents on the native backend. When every column can cross as memory (numbers, bools, strings, nulls,
     float vectors for an ``ARRAY_OF_FLOATS`` property) the batch goes through the columnar entry, no Python values and no JSON
     text; otherwise the columns become one dict per row and go to ``insert_many`` (one crossing per 10,000 rows). Either way
-    the batch is one transaction, or joins the caller's.
+    the batch is one transaction, or joins the caller's. With ``parallel`` the same columnar entry hands the records to the async
+    writers (no transaction), and the call waits for them and raises when a record failed, as ``insert_many(parallel=True)`` does.
     """
-    if not parallel:
+    if parallel:
+        from ._native.bulk import insert_columns_parallel
+
+        if (
+            insert_columns_parallel(
+                db, type_name, columns, {} if declared is None else declared
+            )
+            is not None
+        ):
+            return
+    else:
         from ._native.bulk import insert_columns_fast
 
         if (
@@ -520,7 +560,7 @@ def _insert_documents_native(
         if hasattr(
             col, "to_pylist"
         ):  # a column kept as an Arrow array for the columnar entry
-            lists.append(col.to_pylist())
+            lists.append(_arrow_row_values(col))
         else:
             # jvm-payload: not-a-crossing -- no JVM here: insert_many serialises the rows to JSON text
             lists.append(col.tolist() if hasattr(col, "tolist") else col)
diff --git a/tests/conftest.py b/tests/conftest.py
index ec3147c7..d7d95ce1 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -47,6 +47,14 @@
     collect_ignore.append("test_native_vector_hit_lazy.py")
     # the columnar insert's stored records are compared between the JSON and the columnar entry points of the native library
     collect_ignore.append("test_native_columnar_insert.py")
+    # insert_columns(parallel=True) takes the columnar entry and stores what the row path stores (arcadedb_n_insert_colp with "parallel")
+    collect_ignore.append("test_native_columnar_parallel.py")
+    # the review of the native parallel loads: no silent loss, an interrupt waits for the writers and reads the failure token
+    collect_ignore.append("test_native_columnar_parallel_review_fixes.py")
+    # a column of date-times, dates, Decimals or bytes crosses as a column (kinds T D n x), observed on the native library
+    collect_ignore.append("test_native_columnar_temporal.py")
+    # the review of the temporal kinds: a value the image cannot take is refused in Python, never wrapped or thrown mid-batch
+    collect_ignore.append("test_native_columnar_temporal_review_fixes.py")
     # the rows of an insert_many chunk with date-times, Decimals, bytes and numpy scalars are told apart by the entry points they cross
     collect_ignore.append("test_native_mixed_chunk.py")
     # the native build's ingest defaults (insert_many and graph_batch commit size), observed on the native library
diff --git a/tests/test_columnar_temporal_lists.py b/tests/test_columnar_temporal_lists.py
new file mode 100644
index 00000000..9a9d62b7
--- /dev/null
+++ b/tests/test_columnar_temporal_lists.py
@@ -0,0 +1,114 @@
+"""A column of date-times, dates, Decimals or bytes stores the same values on both wheels (ArcadeData/arcadedb-python #316).
+
+On the native wheel such a column used to move the whole batch off the columnar entry; now ``insert_columns`` and ``insert_many`` send
+it as a column (``tests/test_native_columnar_temporal.py`` proves the entry and compares with the row path). Here the expected values
+are literals, so the same file on the JVM wheel and on the native wheel says that both store them: a naive date-time as it is, a
+time-zone-aware one as its UTC wall clock, a date-time before 1970, a null, a date, a Decimal and bytes.
+"""
+
+import datetime as dt
+import decimal
+
+import pytest
+
+UTC9 = dt.timezone(dt.timedelta(hours=9))
+TS = [
+    dt.datetime(2020, 1, 2, 3, 4, 5, 123456),
+    dt.datetime(1969, 12, 31, 23, 59, 59, 999999),
+    dt.datetime(1900, 1, 1, 0, 0, 0, 1),
+    dt.datetime(2400, 2, 29, 23, 59, 59),
+    None,
+]
+TS_STORED = TS  # a naive date-time is the wall clock
+AWARE = [
+    dt.datetime(2020, 1, 2, 12, 0, 0, 5, tzinfo=UTC9),
+    dt.datetime(1969, 12, 31, 23, 0, 0, tzinfo=dt.timezone.utc),
+    dt.datetime(2020, 6, 1, 0, 0, 0, tzinfo=dt.timezone(dt.timedelta(hours=-7))),
+    None,
+    dt.datetime(2000, 1, 1, tzinfo=dt.timezone.utc),
+]
+AWARE_STORED = [
+    dt.datetime(2020, 1, 2, 3, 0, 0, 5),
+    dt.datetime(1969, 12, 31, 23, 0, 0),
+    dt.datetime(2020, 6, 1, 7, 0, 0),
+    None,
+    dt.datetime(2000, 1, 1),
+]
+DAYS = [
+    dt.date(2020, 2, 29),
+    dt.date(1960, 1, 1),
+    None,
+    dt.date(1970, 1, 1),
+    dt.date(1969, 12, 31),
+]
+AMOUNTS = [
+    decimal.Decimal("12345.6789"),
+    decimal.Decimal("-0.50"),
+    decimal.Decimal("1E+3"),
+    None,
+    decimal.Decimal("0"),
+]
+BLOBS = [b"\x00\xff abc\x7f", b"", None, bytearray(b"ab\x80"), b"z"]
+
+
+def _type(db, name):
+    db.command("sql", f"CREATE DOCUMENT TYPE {name}")
+    db.command("sql", f"CREATE PROPERTY {name}.id INTEGER")
+    db.command("sql", f"CREATE PROPERTY {name}.ts DATETIME_MICROS")
+    db.command("sql", f"CREATE PROPERTY {name}.day DATE")
+    db.command("sql", f"CREATE PROPERTY {name}.amount DECIMAL")
+    db.command("sql", f"CREATE PROPERTY {name}.blob BINARY")
+
+
+def _stored(db, name, column):
+    sql = f"SELECT id, {column} AS v FROM {name} ORDER BY id"  # nosec B608 - test-owned names
+    return [r.get("v") for r in db.query("sql", sql).to_list()]
+
+
+def _bytes(values):
+    # a BINARY property reads back as the values of a Java byte[] (signed, -128..127) or as bytes; & 0xFF takes both to the same bytes
+    return [None if v is None else bytes(b & 0xFF for b in v) for v in values]
+
+
+@pytest.mark.parametrize("path", ["insert_columns", "insert_many"])
+def test_naive_and_aware_datetimes_dates_decimals_and_bytes_are_stored_the_same(
+    temp_db, path
+):
+    _type(temp_db, "Ev")
+    n = len(TS)
+    columns = {
+        "id": list(range(n)),
+        "ts": TS,
+        "day": DAYS,
+        "amount": AMOUNTS,
+        "blob": BLOBS,
+    }
+    if path == "insert_columns":
+        assert temp_db.insert_columns("Ev", columns) == n
+    else:
+        rows = [dict(zip(columns, vals)) for vals in zip(*columns.values())]
+        assert temp_db.insert_many("Ev", rows) == n
+    assert _stored(temp_db, "Ev", "ts") == TS_STORED
+    assert _stored(temp_db, "Ev", "day") == DAYS
+    assert _stored(temp_db, "Ev", "amount") == AMOUNTS
+    assert _bytes(_stored(temp_db, "Ev", "blob")) == _bytes(BLOBS)
+
+
+def test_time_zone_aware_datetimes_are_stored_as_their_utc_wall_clock(temp_db):
+    _type(temp_db, "Zoned")
+    n = len(AWARE)
+    assert temp_db.insert_columns("Zoned", {"id": list(range(n)), "ts": AWARE}) == n
+    assert _stored(temp_db, "Zoned", "ts") == AWARE_STORED
+
+
+def test_a_parallel_load_of_datetimes_stores_the_same_values(temp_db):
+    _type(temp_db, "Par")
+    n = len(TS)
+    assert (
+        temp_db.insert_columns(
+            "Par", {"id": list(range(n)), "ts": TS, "day": DAYS}, parallel=True
+        )
+        == n
+    )
+    assert _stored(temp_db, "Par", "ts") == TS_STORED
+    assert _stored(temp_db, "Par", "day") == DAYS
diff --git a/tests/test_native_columnar_insert.py b/tests/test_native_columnar_insert.py
index 1ec2984c..e7296960 100644
--- a/tests/test_native_columnar_insert.py
+++ b/tests/test_native_columnar_insert.py
@@ -315,15 +315,18 @@ def test_rows_with_different_keys_fall_back_to_json(temp_db, spy, monkeypatch):
     _fallback_case(temp_db, spy, monkeypatch, rows, ROWS)
 
 
-def test_a_datetime_leaves_the_columnar_entry_but_stays_on_the_json_entry(
-    temp_db, spy, monkeypatch
-):
-    """columnar.pack refuses a date-time, so the chunk takes the JSON entry, which carries it as a tagged value
-    (test_native_mixed_chunk.py compares what it stores with INSERT ... CONTENT)."""
+def test_a_datetime_column_stays_on_the_columnar_entry(temp_db, spy, monkeypatch):
+    """A column of date-times is the kind T of the columnar entry (#316): the stored value is the one the JSON entry stores as a tagged
+    value (tests/test_native_columnar_temporal.py has the full comparison, test_native_mixed_chunk.py the rows that stay on JSON).
+    """
     when = datetime.datetime(2024, 5, 6, 7, 8, 9)
     rows = [{"id": i, "txt": "t%d" % i, "when": when} for i in range(10)]
-    _fallback_case(temp_db, spy, monkeypatch, rows, ROWS)
-    assert spy.count(MANY) == 0
+    _types(temp_db, "Ref", "Col")
+    _json_reference(temp_db, "Ref", rows, monkeypatch)
+    spy.calls.clear()
+    temp_db.insert_many("Col", rows)
+    assert spy.count(COLP) == 1 and spy.count(ROWS) == 0 and spy.count(MANY) == 0
+    assert _stored(temp_db, "Col") == _stored(temp_db, "Ref")
 
 
 def test_values_json_cannot_carry_still_go_row_by_row_in_one_crossing(
@@ -344,9 +347,7 @@ def test_an_integer_past_64_bits_is_refused_on_both_paths(temp_db, spy):
         temp_db.insert_columns("Col", {"id": np.array([2**63], dtype=np.uint64)})
 
 
-def test_a_dataframe_with_a_column_the_buffers_cannot_carry_falls_back(
-    temp_db, spy, monkeypatch
-):
+def test_a_dataframe_with_a_timestamp_column_stays_columnar(temp_db, spy, monkeypatch):
     import pandas as pd
     import pyarrow as pa
 
@@ -362,7 +363,7 @@ def test_a_dataframe_with_a_column_the_buffers_cannot_carry_falls_back(
         }
     )
     temp_db.insert_arrow("Col", table)
-    assert spy.count(COLP) == 0
+    assert spy.count(COLP) == 1 and spy.count(ROWS) == 0
     assert pd is not None
     rows = temp_db.query("sql", "SELECT id, txt FROM Col ORDER BY id").to_list()
     assert [(r["id"], r["txt"]) for r in rows] == [(1, "a"), (2, "b"), (3, "c")]
diff --git a/tests/test_native_columnar_pack.py b/tests/test_native_columnar_pack.py
index cda0a1f7..e8925661 100644
--- a/tests/test_native_columnar_pack.py
+++ b/tests/test_native_columnar_pack.py
@@ -190,7 +190,11 @@ def test_a_bitmap_longer_than_a_byte():
         [1, 2**63],  # past 64 bits
         [{"a": 1}, {"a": 2}],  # nested
         [[1], [2]],  # a list without a declared ARRAY_OF_FLOATS
-        [b"ab", b"cd"],
+        [
+            b"ab",
+            "cd",
+        ],  # bytes and text (a column of bytes alone crosses as kind x since #316)
+        [b"ab", bytearray(b"cd")],  # two bytes-like types
         [np.int64(1), np.int64(2)],  # numpy scalars in a list are not Python numbers
     ],
 )
@@ -198,16 +202,107 @@ def test_columns_the_buffer_cannot_carry_unchanged_are_refused(column):
     assert columnar.pack({"c": column}, NO_TYPES) is None
 
 
-def test_datetimes_decimals_and_object_arrays_of_them_are_refused():
+def test_numpy_datetime64_arrays_and_mixed_temporal_columns_are_refused():
     import datetime
     import decimal
 
-    assert columnar.pack({"d": [datetime.datetime(2020, 1, 1)]}, NO_TYPES) is None
-    assert columnar.pack({"d": [decimal.Decimal("1.5")]}, NO_TYPES) is None
     assert (
         columnar.pack({"d": np.array(["2020-01-01"], dtype="datetime64[D]")}, NO_TYPES)
         is None
     )
+    assert (
+        columnar.pack(
+            {"d": [datetime.datetime(2020, 1, 1), datetime.date(2020, 1, 1)]}, NO_TYPES
+        )
+        is None
+    )
+    assert (
+        columnar.pack({"d": [decimal.Decimal("1"), decimal.Decimal("NaN")]}, NO_TYPES)
+        is None
+    )
+    assert columnar.pack({"d": [b"a", "b"]}, NO_TYPES) is None
+
+
+def _ticks(col, dtype):
+    return np.frombuffer(_bytes(col.data), dtype=dtype).tolist()
+
+
+def test_python_datetimes_are_utc_wall_clock_microseconds():
+    import datetime
+
+    utc9 = datetime.timezone(datetime.timedelta(hours=9))
+    values = [
+        datetime.datetime(1970, 1, 1),
+        datetime.datetime(1969, 12, 31, 23, 59, 59, 999999),
+        datetime.datetime(2020, 1, 2, 12, 0, 0, 5, tzinfo=utc9),  # 03:00:00.000005 UTC
+        None,
+    ]
+    rows, cols = columnar.pack({"t": values}, NO_TYPES)
+    col = cols[0]
+    assert rows == 4 and col.kind == "T" and col.dim == 1_000_000
+    assert _ticks(col, np.int64) == [0, -1, 1_577_934_000_000_005, 0]
+    assert _present(col) == [True, True, True, False]
+
+
+def test_python_dates_are_epoch_days():
+    import datetime
+
+    rows, cols = columnar.pack(
+        {"d": [datetime.date(1970, 1, 2), datetime.date(1969, 12, 31), None]}, NO_TYPES
+    )
+    assert cols[0].kind == "D" and _ticks(cols[0], np.int32) == [1, -1, 0]
+    assert _present(cols[0]) == [True, True, False]
+
+
+def test_decimals_and_bytes_use_the_string_layout():
+    import decimal
+
+    rows, cols = columnar.pack(
+        {
+            "a": [decimal.Decimal("1E+3"), None, decimal.Decimal("-0.50")],
+            "b": [b"\x00\xff", b"", None],
+            "c": [bytearray(b"xy"), bytearray(b"z"), bytearray(b"")],
+        },
+        NO_TYPES,
+    )
+    a, b, c = cols
+    assert [x.kind for x in cols] == ["n", "x", "x"]
+    offsets = np.frombuffer(_bytes(a.data), dtype=np.int32).tolist()
+    assert offsets == [0, 4, 4, 9] and _bytes(a.aux) == b"1E+3-0.50"
+    offsets = np.frombuffer(_bytes(b.data), dtype=np.int32).tolist()
+    assert offsets == [0, 2, 2, 2] and _bytes(b.aux) == b"\x00\xff"
+    assert _bytes(c.aux) == b"xyz" and _present(a) == [True, False, True]
+
+
+def test_arrow_timestamps_dates_decimals_and_binary():
+    import decimal
+
+    import pyarrow as pa
+
+    stamps = pa.array([1, None, -2], pa.int64()).cast(
+        pa.timestamp("ms", tz="Asia/Seoul")
+    )
+    days = pa.array([3, -4, None], pa.int32()).cast(pa.date32())
+    day_ms = pa.array([86_400_000 * 2, -1, None], pa.int64()).cast(pa.date64())
+    amounts = pa.array([decimal.Decimal("1.50"), None], pa.decimal128(10, 2))
+    blobs = pa.array([b"ab", None, b"c"])
+    for name, col, expected in (
+        ("t", stamps, ("T", 1_000)),
+        ("d", days, ("D", 0)),
+        ("m", day_ms, ("D", 0)),
+    ):
+        packed = columnar.pack({name: col}, NO_TYPES)[1][0]
+        assert (packed.kind, packed.dim) == expected
+    assert _ticks(columnar.pack({"t": stamps}, NO_TYPES)[1][0], np.int64) == [1, 0, -2]
+    assert _ticks(columnar.pack({"d": days}, NO_TYPES)[1][0], np.int32) == [3, -4, 0]
+    # a date64 is floor-divided to days: 2 days, then -1 ms is the day before the epoch
+    assert _ticks(columnar.pack({"m": day_ms}, NO_TYPES)[1][0], np.int32) == [2, -1, 0]
+    n = columnar.pack({"n": amounts}, NO_TYPES)[1][0]
+    assert n.kind == "n" and _bytes(n.aux) == b"1.50"
+    x = columnar.pack({"x": blobs}, NO_TYPES)[1][0]
+    assert (
+        x.kind == "x" and _bytes(x.aux) == b"abc" and _present(x) == [True, False, True]
+    )
 
 
 def test_ragged_columns_and_empty_input_are_left_to_the_row_path():
@@ -233,6 +328,11 @@ def test_the_address_table_and_header_describe_the_columns():
         "cols": [["id", "q", 0], ["s", "s", 0]],
     }
     assert json.loads(columnar.header_of(rows, 2**40, packed))["commit"] == 2**31 - 1
+    # "parallel" and the failure token are in the header only when a token is given (the image hands the records to the async writers then)
+    parallel = json.loads(columnar.header_of(rows, 10_000, packed, fid=41))
+    assert parallel["parallel"] is True and parallel["fid"] == 41
+    plain = json.loads(columnar.header_of(rows, 10_000, packed))
+    assert "parallel" not in plain and "fid" not in plain
 
 
 def test_rows_become_columns_only_when_every_row_has_the_same_keys():
diff --git a/tests/test_native_columnar_parallel.py b/tests/test_native_columnar_parallel.py
new file mode 100644
index 00000000..90d6f79a
--- /dev/null
+++ b/tests/test_native_columnar_parallel.py
@@ -0,0 +1,242 @@
+"""Native ``insert_columns(parallel=True)`` takes the columnar entry and stores what the row path of ``insert_many(parallel=True)`` stores.
+
+Before, a parallel call skipped the columnar fast path: the columns became one dict per row, then one JSON text per chunk of 10,000
+rows for ``arcadedb_n_async insert_parallel``. Now the same ``arcadedb_n_insert_colp`` crossing that the synchronous call uses carries
+``"parallel": true`` and the image hands the records to the async writers, as ``DocumentBatcher.insertColumnsParallel`` does on the
+JVM wheel. Needs the native library, so conftest.py collects this file on the native backend only. Each test loads the same data
+through the row path (the private reference: ``insert_many(parallel=True)`` with dict rows) and through the columns, compares the
+stored records value by value and type by type, and spies on the entry points so a silent fall back cannot pass.
+"""
+
+import numpy as np
+import pytest
+
+from arcadedb._native import bulk, extras, lib  # noqa: E402
+from arcadedb.exceptions import ArcadeDBError  # noqa: E402
+
+COLP, ROWS = "arcadedb_n_insert_colp", "arcadedb_n_insert_rows"
+
+
+class Spy:
+    """Counts the crossings by name and the calls that hand a chunk of dict rows to the parallel writers."""
+
+    def __init__(self, monkeypatch):
+        self.calls, self.row_chunks = [], 0
+        real_call, real_rows = lib.call, extras.insert_parallel
+
+        def call(name, *args):
+            self.calls.append(name)
+            return real_call(name, *args)
+
+        def insert_parallel(database, type_name, rows):
+            self.row_chunks += 1
+            return real_rows(database, type_name, rows)
+
+        monkeypatch.setattr(lib, "call", call)
+        monkeypatch.setattr(extras, "insert_parallel", insert_parallel)
+
+    def count(self, name):
+        return self.calls.count(name)
+
+
+@pytest.fixture
+def spy(monkeypatch):
+    return Spy(monkeypatch)
+
+
+def _count(db, name):
+    sql = f"SELECT count(*) AS n FROM {name}"  # nosec B608 - test-owned type name
+    return db.query("sql", sql).to_list()[0]["n"]
+
+
+def _types(db, *names, kind="DOCUMENT", buckets=3, unique=False):
+    for name in names:
+        db.command("sql", f"CREATE {kind} TYPE {name} BUCKETS {buckets}")
+        db.command("sql", f"CREATE PROPERTY {name}.id INTEGER")
+        db.command("sql", f"CREATE PROPERTY {name}.val DOUBLE")
+        db.command("sql", f"CREATE PROPERTY {name}.flag BOOLEAN")
+        db.command("sql", f"CREATE PROPERTY {name}.txt STRING")
+        db.command("sql", f"CREATE PROPERTY {name}.vec ARRAY_OF_FLOATS")
+        if unique:
+            db.command("sql", f"CREATE INDEX ON {name} (id) UNIQUE")
+
+
+def _stored(db, name):
+    """Every record of a type as (value, python type) pairs per property, ordered by id (the writers store in no fixed order)."""
+    sql = f"SELECT FROM {name} ORDER BY id"  # nosec B608 - test-owned type name
+    return [
+        {k: (v, type(v).__name__) for k, v in row.items() if not k.startswith("@")}
+        for row in db.query("sql", sql).to_list()
+    ]
+
+
+def _columns(n, dim=4, seed=11):
+    rng = np.random.default_rng(seed)
+    ids = np.arange(n, dtype=np.int64)
+    return {
+        "id": ids,
+        "val": (ids * 37 % 10007) / 7.0,
+        "flag": ids % 3 == 0,
+        "txt": ["row-%d-%s" % (k, "é" * (k % 5)) for k in range(n)],
+        "vec": rng.standard_normal((n, dim), dtype=np.float32),
+    }
+
+
+def _rows(columns):
+    names = list(columns)
+    lists = [c.tolist() if hasattr(c, "tolist") else c for c in columns.values()]
+    return [dict(zip(names, vals)) for vals in zip(*lists)]
+
+
+def test_a_parallel_call_uses_the_columnar_entry_and_stores_the_row_paths_records(
+    temp_db, spy
+):
+    _types(temp_db, "Ref", "Col")
+    columns = _columns(3_100)
+    assert temp_db.insert_many("Ref", _rows(columns), parallel=True) == 3_100
+    assert spy.row_chunks == 1 and spy.count(COLP) == 0  # the reference is the row path
+
+    spy.calls.clear()
+    spy.row_chunks = 0
+    assert temp_db.insert_columns("Col", columns, parallel=True) == 3_100
+    assert spy.count(COLP) == 1, spy.calls  # one crossing for the whole call
+    assert spy.row_chunks == 0 and spy.count(ROWS) == 0  # no dict rows, no JSON text
+
+    expected = _stored(temp_db, "Ref")
+    assert len(expected) == 3_100 and _count(temp_db, "Col") == 3_100
+    assert expected[7]["id"][1] == "int" and expected[7]["vec"][1] == "list"
+    assert _stored(temp_db, "Col") == expected
+
+
+def test_nulls_and_awkward_strings_are_stored_as_the_row_path_stores_them(temp_db, spy):
+    _types(temp_db, "Ref", "Col")
+    columns = {
+        "id": list(range(8)),
+        "txt": ["", "a", None, "snow☃man", "x" * 300, None, "\n\t", "z"],
+        "val": [1.5, None, 2.5, 1e300, -1e-300, 0.0, 3.0, None],
+    }
+    temp_db.insert_many("Ref", _rows(columns), parallel=True)
+    spy.calls.clear()
+    temp_db.insert_columns("Col", columns, parallel=True)
+    assert spy.count(COLP) == 1
+    got, want = _stored(temp_db, "Col"), _stored(temp_db, "Ref")
+    assert got == want and got[2]["txt"][0] is None
+
+
+def test_a_vertex_type_gets_vertices_in_parallel(temp_db, spy):
+    _types(temp_db, "Ref", "Col", kind="VERTEX")
+    columns = _columns(1_500)
+    temp_db.insert_many("Ref", _rows(columns), parallel=True)
+    spy.calls.clear()
+    temp_db.insert_columns("Col", columns, parallel=True)
+    assert spy.count(COLP) == 1
+    assert _stored(temp_db, "Col") == _stored(temp_db, "Ref")
+    kinds = temp_db.query(
+        "sql", "SELECT FROM schema:types WHERE name = 'Col'"
+    ).to_list()
+    assert kinds[0]["type"] == "vertex"
+
+
+def test_a_record_the_writers_reject_is_reported_after_the_wait(temp_db, spy):
+    _types(temp_db, "Col", unique=True)
+    ids = np.arange(400, dtype=np.int64)
+    ids[300] = ids[7]  # a duplicate key
+    with pytest.raises(ArcadeDBError, match="failed record"):
+        temp_db.insert_columns("Col", {"id": ids, "txt": ["t"] * 400}, parallel=True)
+    assert spy.count(COLP) == 1 and spy.row_chunks == 0
+    stored = _count(temp_db, "Col")
+    assert 0 < stored <= 400  # the call had waited for the writers: nothing lands later
+
+
+def test_a_value_the_declared_type_refuses_raises_and_the_call_is_final(temp_db, spy):
+    _types(temp_db, "Col")
+    values = [str(i) for i in range(1_000)]
+    values[700] = "not a number"
+    with pytest.raises(
+        ArcadeDBError, match="may have been stored|failed record"
+    ) as err:
+        temp_db.insert_columns("Col", {"id": values}, parallel=True)
+    assert "rolled back" not in str(err.value)
+    assert spy.count(COLP) == 1
+    stored = _count(temp_db, "Col")
+    temp_db.async_executor().wait_completion()
+    assert _count(temp_db, "Col") == stored <= 1_000
+
+
+def test_an_edge_type_is_refused_as_before(temp_db):
+    temp_db.command("sql", "CREATE EDGE TYPE Link")
+    with pytest.raises(ArcadeDBError, match="edge type"):
+        temp_db.insert_columns("Link", {"id": [1, 2, 3]}, parallel=True)
+
+
+def test_a_missing_type_is_the_schemas_error(temp_db):
+    with pytest.raises(ArcadeDBError, match="NoSuchType"):
+        temp_db.insert_columns("NoSuchType", {"id": [1, 2, 3]}, parallel=True)
+
+
+def test_a_column_the_buffers_cannot_carry_still_goes_through_the_row_path(
+    temp_db, spy
+):
+    _types(temp_db, "Col")
+    columns = {"id": [1, 2, 3, 4], "txt": ["a", 2, "c", 4.5]}  # a mixed-type column
+    assert temp_db.insert_columns("Col", columns, parallel=True) == 4
+    assert spy.count(COLP) == 0 and spy.row_chunks == 1
+    assert _count(temp_db, "Col") == 4
+
+
+def test_the_switch_that_turns_the_columnar_path_off_applies_to_parallel(
+    temp_db, spy, monkeypatch
+):
+    _types(temp_db, "Col")
+    monkeypatch.setattr(bulk, "_ENABLED", False)
+    assert temp_db.insert_columns("Col", _columns(50), parallel=True) == 50
+    assert spy.count(COLP) == 0 and spy.row_chunks == 1
+    assert _count(temp_db, "Col") == 50
+
+
+def test_no_rows_is_zero_and_writes_nothing(temp_db, spy):
+    _types(temp_db, "Col")
+    assert temp_db.insert_columns("Col", {"id": []}, parallel=True) == 0
+    assert spy.count(COLP) == 0 and _count(temp_db, "Col") == 0
+
+
+def test_the_synchronous_call_is_unchanged(temp_db, spy):
+    _types(temp_db, "Col", unique=True)
+    assert temp_db.insert_columns("Col", _columns(500), commit_every=100) == 500
+    assert spy.count(COLP) == 1 and spy.row_chunks == 0
+    assert _count(temp_db, "Col") == 500
+
+
+def test_insert_arrow_parallel_takes_the_columnar_entry(temp_db, spy, monkeypatch):
+    import pyarrow as pa
+
+    _types(temp_db, "Ref", "ByTable")
+    columns = _columns(2_000)
+    table = pa.table(
+        {
+            "id": columns["id"],
+            "val": pa.array(
+                [None if i % 7 == 0 else v for i, v in enumerate(columns["val"])]
+            ),
+            "txt": pa.array(columns["txt"]),
+        }
+    )
+    with (
+        monkeypatch.context() as m
+    ):  # the private reference: the row path of the old code
+        m.setattr(bulk, "_ENABLED", False)
+        assert (
+            temp_db.insert_arrow("Ref", table, batch_size=700, parallel=True) == 2_000
+        )
+    assert spy.count(COLP) == 0 and spy.row_chunks >= 1
+
+    spy.calls.clear()
+    spy.row_chunks = 0
+    assert (
+        temp_db.insert_arrow("ByTable", table, batch_size=700, parallel=True) == 2_000
+    )
+    assert spy.count(COLP) == 3 and spy.row_chunks == 0  # one crossing per batch
+
+    expected = _stored(temp_db, "Ref")
+    assert len(expected) == 2_000 and expected[0]["val"][0] is None
+    assert _stored(temp_db, "ByTable") == expected
diff --git a/tests/test_native_columnar_parallel_review_fixes.py b/tests/test_native_columnar_parallel_review_fixes.py
new file mode 100644
index 00000000..70f17bc3
--- /dev/null
+++ b/tests/test_native_columnar_parallel_review_fixes.py
@@ -0,0 +1,203 @@
+"""Review of the native parallel loads (#328): nothing is lost silently, an interrupt leaves no writer running and no token behind.
+
+``insert_columns(parallel=True)`` hands the records to the async writers in one crossing and waits for them before it returns. These
+tests pin what a caller relies on: a record that a writer rejects is counted (``stored + reported failed >= rows``) and no key is stored twice
+(a writer that meets a duplicate abandons its whole open batch, and reports it); a load at the writers' batch boundary stores every row and nothing lands after the call returns;
+a ^C during the crossing, or between the chunks of the row path, waits for the writers and reads (so removes) the failure token; a wait
+that fails still reads the tokens; and the message of a failed parallel batch inside a caller's transaction does not claim that a
+rollback removes rows the writers committed. Needs the native library: conftest.py collects this file on the native backend only.
+"""
+
+import re
+import time
+
+import numpy as np
+import pytest
+
+from arcadedb._native import database as native_database
+from arcadedb._native import extras
+from arcadedb.exceptions import ArcadeDBError
+
+COLP = "arcadedb_n_insert_colp"
+
+
+def _count(db, name):
+    sql = f"SELECT count(*) AS n FROM {name}"  # nosec B608 - test-owned type name
+    return db.query("sql", sql).to_list()[0]["n"]
+
+
+def _type(db, name="Col", unique=False):
+    db.command("sql", f"CREATE DOCUMENT TYPE {name} BUCKETS 3")
+    db.command("sql", f"CREATE PROPERTY {name}.id INTEGER")
+    db.command("sql", f"CREATE PROPERTY {name}.txt STRING")
+    if unique:
+        db.command("sql", f"CREATE INDEX ON {name} (id) UNIQUE")
+
+
+class TokenSpy:
+    """The failure tokens that were made and the ones that were read (a read removes the token's entries in the image)."""
+
+    def __init__(self, monkeypatch):
+        self.made, self.read = [], []
+        real_make, real_read = extras.failure_token, extras.failures
+        real_rows = extras.insert_parallel
+
+        def make(database):
+            fid = real_make(database)
+            self.made.append(fid)
+            return fid
+
+        def failures(database, token):
+            self.read.append(token)
+            return real_read(database, token)
+
+        def rows(database, type_name, chunk):
+            fid = real_rows(database, type_name, chunk)
+            self.made.append(fid["fid"])
+            return fid
+
+        monkeypatch.setattr(extras, "failure_token", make)
+        monkeypatch.setattr(extras, "failures", failures)
+        monkeypatch.setattr(extras, "insert_parallel", rows)
+
+
+@pytest.fixture
+def tokens(monkeypatch):
+    return TokenSpy(monkeypatch)
+
+
+def _failed_count(error):
+    return int(re.search(r"reported (\d+) failed record", str(error)).group(1))
+
+
+def test_a_duplicate_key_is_reported_and_no_row_is_stored_twice(temp_db):
+    _type(temp_db, unique=True)
+    n, dupes = 3_000, 50
+    ids = np.arange(n, dtype=np.int64)
+    ids[n - dupes :] = np.arange(dupes)  # the last 50 repeat the first 50 keys
+    with pytest.raises(ArcadeDBError, match="failed record") as err:
+        temp_db.insert_columns("Col", {"id": ids, "txt": ["t"] * n}, parallel=True)
+    stored_ids = [r["id"] for r in temp_db.query("sql", "SELECT id FROM Col").to_list()]
+    stored = len(stored_ids)
+    assert (
+        stored + _failed_count(err.value) >= n
+    )  # every row is a stored row or a reported one
+    # A writer that meets a duplicate key rolls back its WHOLE open batch (up to commit_every rows), so the valid rows of that batch are
+    # abandoned and reported too: `stored` can be below the number of distinct keys, never above it.
+    assert stored <= len(set(ids.tolist()))
+    assert set(stored_ids) <= set(ids.tolist())  # only ids that were sent
+    assert len(stored_ids) == len(set(stored_ids))  # none stored twice
+
+
+def test_a_load_at_the_writers_batch_boundary_stores_every_row_and_nothing_later(
+    temp_db,
+):
+    _type(temp_db)
+    threads = temp_db.async_executor().get_parallel_level()
+    n = threads * 10_240 + 1  # one row past a full batch for every writer
+    ids = np.arange(n, dtype=np.int64)
+    assert (
+        temp_db.insert_columns("Col", {"id": ids, "txt": ["t"] * n}, parallel=True) == n
+    )
+    stored = _count(temp_db, "Col")
+    assert stored == n
+    time.sleep(0.3)
+    temp_db.async_executor().wait_completion()
+    assert _count(temp_db, "Col") == n  # nothing landed after the call returned
+    distinct = temp_db.query("sql", "SELECT count(DISTINCT id) AS c FROM Col")
+    assert distinct.to_list()[0]["c"] == n
+
+
+def test_an_interrupt_in_the_crossing_waits_for_the_writers_and_reads_the_token(
+    temp_db, tokens, monkeypatch
+):
+    _type(temp_db)
+    real_call = native_database._call
+
+    def interrupted(prefix, name, *args):
+        reply = real_call(prefix, name, *args)
+        if name == COLP:
+            raise KeyboardInterrupt  # the records are handed over; the reply never reaches the caller
+        return reply
+
+    monkeypatch.setattr(native_database, "_call", interrupted)
+    n = 30_000
+    with pytest.raises(KeyboardInterrupt):
+        temp_db.insert_columns(
+            "Col", {"id": np.arange(n, dtype=np.int64), "txt": ["t"] * n}, parallel=True
+        )
+    assert (
+        temp_db.async_executor().is_processing() is False
+    )  # waited: no writer runs on
+    assert len(tokens.made) == 1 and tokens.read == tokens.made  # and the token is gone
+    assert (
+        _count(temp_db, "Col") == n
+    )  # the rows handed over were all written, none half
+
+
+def test_an_interrupt_between_the_chunks_of_the_row_path_waits_and_reads_the_tokens(
+    temp_db, tokens, monkeypatch
+):
+    _type(temp_db)
+    real_rows = extras.insert_parallel
+    seen = []
+
+    def second_chunk_interrupted(database, type_name, chunk):
+        if seen:
+            raise KeyboardInterrupt
+        seen.append(1)
+        return real_rows(database, type_name, chunk)
+
+    monkeypatch.setattr(extras, "insert_parallel", second_chunk_interrupted)
+    rows = [{"id": i, "txt": "t"} for i in range(25_000)]
+    with pytest.raises(KeyboardInterrupt):
+        temp_db.insert_many("Col", rows, parallel=True)
+    assert temp_db.async_executor().is_processing() is False
+    assert _count(temp_db, "Col") == 10_000  # the first chunk, whole
+    assert len(tokens.made) == 1 and tokens.read == tokens.made
+
+
+def test_a_wait_that_fails_still_reads_the_tokens(temp_db, tokens, monkeypatch):
+    _type(temp_db)
+    real_wait = extras.NativeAsyncExecutor.wait_completion
+    state = {"raised": False}
+
+    def failing_wait(self, timeout_ms=None):
+        if not state["raised"]:
+            state["raised"] = True
+            raise RuntimeError("the wait failed")
+        return real_wait(self, timeout_ms)
+
+    monkeypatch.setattr(extras.NativeAsyncExecutor, "wait_completion", failing_wait)
+    n = 2_000
+    with pytest.raises(ArcadeDBError, match="the wait failed"):
+        temp_db.insert_columns(
+            "Col", {"id": np.arange(n, dtype=np.int64), "txt": ["t"] * n}, parallel=True
+        )
+    assert len(tokens.made) == 1 and tokens.read == tokens.made
+    temp_db.async_executor().wait_completion()
+    assert _count(temp_db, "Col") == n
+
+
+def test_a_failed_parallel_batch_in_a_caller_transaction_does_not_promise_a_rollback(
+    temp_db,
+):
+    import pyarrow as pa
+
+    _type(temp_db, unique=True)
+    ids = list(range(300))
+    ids[250] = 7  # a duplicate key in the third batch of 100
+    table = pa.table({"id": pa.array(ids, pa.int64()), "txt": ["t"] * 300})
+    temp_db.begin()
+    try:
+        with pytest.raises(ArcadeDBError) as err:
+            temp_db.insert_arrow("Col", table, batch_size=100, parallel=True)
+        message = str(err.value)
+        assert "not committed" not in message and "your transaction" in message
+        assert "async writers" in message and "does not remove them" in message
+    finally:
+        temp_db.rollback()
+    temp_db.async_executor().wait_completion()
+    assert (
+        _count(temp_db, "Col") >= 200
+    )  # the writers committed the first two batches: the rollback did not remove them
diff --git a/tests/test_native_columnar_temporal.py b/tests/test_native_columnar_temporal.py
new file mode 100644
index 00000000..7ae3b545
--- /dev/null
+++ b/tests/test_native_columnar_temporal.py
@@ -0,0 +1,326 @@
+"""Native: a column of date-times, dates, Decimals or bytes stays on the columnar entry and stores what the row path stores (#316).
+
+Before, one such column made ``columnar.pack`` return None, so every column of the batch went through one dict per row and a JSON text
+with tagged values. Now the kinds T, D, n and x of ``arcadedb_n_insert_colp`` carry them and the image builds the same objects
+(``LocalDateTime``, ``LocalDate``, ``BigDecimal``, ``byte[]``) that ``Json.decodeTag`` builds for the tagged values. Each test loads the
+same data through the row path (the private reference: the columnar path switched off, so the chunk goes as tagged JSON) and through
+the columns, compares every stored value and its type, and spies on the entry points so a silent fall back cannot pass. Needs the
+native library, so conftest.py collects this file on the native backend only.
+"""
+
+import datetime as dt
+import decimal
+
+import pytest
+
+from arcadedb._native import bulk, columnar, lib  # noqa: E402
+
+COLP, ROWS, MANY = (
+    "arcadedb_n_insert_colp",
+    "arcadedb_n_insert_rows",
+    "arcadedb_n_exec_many",
+)
+UTC9 = dt.timezone(dt.timedelta(hours=9))
+
+
+class Spy:
+    def __init__(self, monkeypatch):
+        self.calls = []
+        real = lib.call
+
+        def call(name, *args):
+            self.calls.append(name)
+            return real(name, *args)
+
+        monkeypatch.setattr(lib, "call", call)
+
+    def count(self, name):
+        return self.calls.count(name)
+
+
+@pytest.fixture
+def spy(monkeypatch):
+    return Spy(monkeypatch)
+
+
+def _type(db, name, declared=True):
+    db.command("sql", f"CREATE DOCUMENT TYPE {name}")
+    db.command("sql", f"CREATE PROPERTY {name}.id INTEGER")
+    if declared:
+        for prop, ty in (
+            ("ts", "DATETIME_MICROS"),
+            ("ns", "DATETIME_NANOS"),
+            ("day", "DATE"),
+            ("amount", "DECIMAL"),
+            ("blob", "BINARY"),
+        ):
+            db.command("sql", f"CREATE PROPERTY {name}.{prop} {ty}")
+
+
+def _stored(db, name):
+    """Every record by id as {property: (python type name, repr)}, so a changed type is as visible as a changed value."""
+    sql = f"SELECT FROM {name} ORDER BY id"  # nosec B608 - test-owned type name
+    return [
+        {
+            k: (type(v).__name__, repr(v))
+            for k, v in sorted(rec.items())
+            if not k.startswith("@") and k != "rid"
+        }
+        for rec in db.query("sql", sql).to_list()
+    ]
+
+
+def _rows(columns):
+    return [dict(zip(columns, vals)) for vals in zip(*columns.values())]
+
+
+def _reference(db, name, rows, monkeypatch):
+    with monkeypatch.context() as m:
+        m.setattr(bulk, "_ENABLED", False)
+        db.insert_many(name, rows)
+
+
+def _columns(n=60):
+    base = dt.datetime(1969, 12, 31, 23, 59, 59, 999999)
+    return {
+        "id": list(range(n)),
+        "ts": [
+            (
+                None
+                if i % 11 == 0
+                else (base + dt.timedelta(seconds=i * 86_399, microseconds=i * 7))
+            )
+            for i in range(n)
+        ],
+        "day": [
+            None if i % 13 == 0 else dt.date(1960, 1, 1) + dt.timedelta(days=i * 401)
+            for i in range(n)
+        ],
+        "amount": [
+            (
+                None
+                if i % 9 == 0
+                else decimal.Decimal(f"{i * 1234567}.{i % 100:02d}")
+                * (-1 if i % 2 else 1)
+            )
+            for i in range(n)
+        ],
+        "blob": [
+            None if i % 8 == 0 else bytes(range(i % 256)) + b"\x00\xff"
+            for i in range(n)
+        ],
+    }
+
+
+@pytest.mark.parametrize("declared", [True, False], ids=["declared", "schemaless"])
+def test_a_batch_with_temporal_decimal_and_bytes_columns_stays_columnar(
+    temp_db, spy, monkeypatch, declared
+):
+    for name in ("Ref", "ByRows", "ByColumns"):
+        _type(temp_db, name, declared)
+    columns = _columns()
+    _reference(temp_db, "Ref", _rows(columns), monkeypatch)
+    assert (
+        spy.count(COLP) == 0 and spy.count(ROWS) == 1
+    )  # the reference is the JSON path
+    spy.calls.clear()
+    assert temp_db.insert_many("ByRows", _rows(columns)) == 60
+    assert spy.count(COLP) == 1 and spy.count(ROWS) == 0 and spy.count(MANY) == 0
+    spy.calls.clear()
+    assert temp_db.insert_columns("ByColumns", columns) == 60
+    assert spy.count(COLP) == 1 and spy.count(ROWS) == 0
+    expected = _stored(temp_db, "Ref")
+    assert len(expected) == 60 and expected[1]["ts"][0] == "datetime"
+    assert _stored(temp_db, "ByRows") == expected
+    assert _stored(temp_db, "ByColumns") == expected
+
+
+def test_time_zone_aware_and_naive_datetimes_follow_the_json_rule(
+    temp_db, spy, monkeypatch
+):
+    """An aware date-time is its UTC wall clock, a naive one is taken as it is (_tag_datetime): the column does not decide a zone."""
+    _type(temp_db, "Ref")
+    _type(temp_db, "Col")
+    values = [
+        dt.datetime(2020, 1, 2, 12, 0, 0, 5, tzinfo=UTC9),
+        dt.datetime(1969, 12, 31, 23, 0, 0, tzinfo=dt.timezone.utc),
+        dt.datetime(2020, 6, 1, 0, 0, 0, tzinfo=dt.timezone(dt.timedelta(hours=-7))),
+    ]
+    rows = [{"id": i, "ts": v} for i, v in enumerate(values)]
+    _reference(temp_db, "Ref", rows, monkeypatch)
+    spy.calls.clear()
+    temp_db.insert_many("Col", rows)
+    assert spy.count(COLP) == 1
+    assert _stored(temp_db, "Col") == _stored(temp_db, "Ref")
+
+
+def test_pandas_timestamps_keep_microseconds_as_the_json_path_does(
+    temp_db, spy, monkeypatch
+):
+    import pandas as pd
+
+    _type(temp_db, "Ref")
+    _type(temp_db, "Col")
+    stamps = [
+        pd.Timestamp("2026-05-06 07:08:09.123456789"),
+        pd.Timestamp("1960-05-06 07:08:09.123456", tz="Asia/Seoul"),
+    ]
+    rows = [{"id": i, "ts": v} for i, v in enumerate(stamps)]
+    _reference(temp_db, "Ref", rows, monkeypatch)
+    spy.calls.clear()
+    temp_db.insert_many("Col", rows)
+    assert spy.count(COLP) == 1 and spy.count(ROWS) == 0
+    assert _stored(temp_db, "Col") == _stored(temp_db, "Ref")
+
+
+@pytest.mark.parametrize(
+    "column, colp_runs",
+    [
+        ([dt.datetime(2020, 1, 1), dt.date(2020, 1, 1)], 0),  # a date-time and a date
+        # A non-finite Decimal: the column is refused (BigDecimal("NaN") would throw mid-batch in the image), so the NaN row goes to the
+        # statement on its own, as before. The row around it is a run of carriable rows: it goes in bulk, and the columnar entry
+        # carries a finite Decimal now, so that one run crosses on arcadedb_n_insert_colp (before, it was the JSON entry).
+        ([decimal.Decimal("1.5"), decimal.Decimal("NaN")], 1),
+        ([b"a", "b"], 0),  # bytes and text
+    ],
+    ids=["datetime-and-date", "nan-decimal", "bytes-and-text"],
+)
+def test_a_column_the_buffers_cannot_carry_unchanged_still_falls_back(
+    temp_db, spy, monkeypatch, column, colp_runs
+):
+    assert columnar.pack({"x": column}, columnar.Declared(known={})) is None
+    _type(temp_db, "Ref", declared=False)
+    _type(temp_db, "Col", declared=False)
+    rows = [{"id": i, "x": v} for i, v in enumerate(column)]
+    try:
+        _reference(temp_db, "Ref", rows, monkeypatch)
+    except (
+        Exception
+    ):  # the statement path may refuse it; then the new path must refuse it the same way
+        with pytest.raises(Exception):
+            temp_db.insert_many("Col", rows)
+        assert spy.count(COLP) == colp_runs
+        return
+    spy.calls.clear()
+    temp_db.insert_many("Col", rows)
+    assert spy.count(COLP) == colp_runs
+    assert _stored(temp_db, "Col") == _stored(temp_db, "Ref")
+
+
+def _arrow_table(pa):
+    import numpy as np
+
+    n = 40
+    ids = np.arange(n, dtype=np.int64)
+    epoch_us = (ids - 20) * 1_000_003_000 + 7  # microseconds, before and after 1970
+    mask = ids % 6 == 0
+
+    def stamps(unit, tz=None):
+        raw = {
+            "s": epoch_us // 1_000_000,
+            "ms": epoch_us // 1_000,
+            "us": epoch_us,
+            "ns": epoch_us * 1_000,
+        }[unit]
+        return pa.array(raw.astype(np.int64), mask=mask).cast(pa.timestamp(unit, tz=tz))
+
+    return pa.table(
+        {
+            "id": ids,
+            "ts": stamps("us"),
+            "ns": stamps("ns"),
+            "sec": stamps("s"),
+            "zoned": stamps("ms", tz="Asia/Seoul"),
+            "day": pa.array(
+                ((ids - 10) * 397).astype(np.int32), mask=ids % 5 == 0
+            ).cast(pa.date32()),
+            "amount": pa.array(
+                [
+                    None if i % 7 == 0 else decimal.Decimal(f"{i}.{i % 100:02d}")
+                    for i in range(n)
+                ],
+                type=pa.decimal128(20, 2),
+            ),
+            "blob": pa.array(
+                [None if i % 4 == 0 else bytes(range(i)) for i in range(n)]
+            ),
+        }
+    )
+
+
+def test_arrow_timestamps_dates_decimals_and_binary_stay_columnar(
+    temp_db, spy, monkeypatch
+):
+    import pyarrow as pa
+
+    for name in ("Ref", "Col"):
+        _type(temp_db, name)
+        for prop, ty in (("sec", "DATETIME_SECOND"), ("zoned", "DATETIME")):
+            temp_db.command("sql", f"CREATE PROPERTY {name}.{prop} {ty}")
+    table = _arrow_table(pa)
+    with (
+        monkeypatch.context() as m
+    ):  # the reference: tagged values through the JSON path
+        m.setattr(bulk, "_ENABLED", False)
+        assert temp_db.insert_arrow("Ref", table) == 40
+    assert spy.count(COLP) == 0
+    spy.calls.clear()
+    assert temp_db.insert_arrow("Col", table) == 40
+    assert spy.count(COLP) == 1 and spy.count(ROWS) == 0 and spy.count(MANY) == 0
+    expected = _stored(temp_db, "Ref")
+    assert len(expected) == 40 and expected[1]["ns"][0] == "datetime"
+    assert _stored(temp_db, "Col") == expected
+
+
+def test_an_arrow_batch_with_one_list_column_falls_back_with_the_same_values(
+    temp_db, spy, monkeypatch
+):
+    """A list<string> column cannot cross as a buffer: the whole batch takes the row path, and the temporal columns reach it as tagged values."""
+    import pyarrow as pa
+
+    for name in ("Ref", "Col"):
+        _type(temp_db, name)
+        for prop, ty in (("sec", "DATETIME_SECOND"), ("zoned", "DATETIME")):
+            temp_db.command("sql", f"CREATE PROPERTY {name}.{prop} {ty}")
+    table = _arrow_table(pa).append_column(
+        "tags", pa.array([["a", "b"]] * 40, type=pa.list_(pa.string()))
+    )
+    with monkeypatch.context() as m:
+        m.setattr(bulk, "_ENABLED", False)
+        temp_db.insert_arrow("Ref", table)
+    spy.calls.clear()
+    temp_db.insert_arrow("Col", table)
+    assert spy.count(COLP) == 0
+    assert _stored(temp_db, "Col") == _stored(temp_db, "Ref")
+
+
+def test_a_parallel_load_of_datetimes_is_one_columnar_crossing(temp_db, spy):
+    _type(temp_db, "Par")
+    columns = _columns(300)
+    assert temp_db.insert_columns("Par", columns, parallel=True) == 300
+    assert spy.count(COLP) == 1 and spy.count(ROWS) == 0
+    assert temp_db.query("sql", "SELECT count(*) AS n FROM Par").first().get("n") == 300
+
+
+def test_the_image_refuses_a_wrong_ticks_per_second_and_writes_nothing(temp_db):
+    import json
+
+    _type(temp_db, "Col")
+    rows, packed = columnar.pack(
+        {"id": [1, 2], "ts": [dt.datetime(2020, 1, 1)] * 2}, columnar.Declared(known={})
+    )
+    assert packed[1].kind == "T" and packed[1].dim == 1_000_000
+    packed[1].dim = 7
+    spec = columnar.spec_of(packed)
+    reply = json.loads(
+        lib.call(
+            COLP,
+            temp_db._h,
+            lib.b("Col"),
+            lib.b(columnar.header_of(rows, 10_000, packed)),
+            spec,
+            len(spec),
+        )
+    )
+    assert not reply["ok"] and "ticks per second" in reply["error"]
+    assert temp_db.query("sql", "SELECT count(*) AS n FROM Col").first().get("n") == 0
diff --git a/tests/test_native_columnar_temporal_review_fixes.py b/tests/test_native_columnar_temporal_review_fixes.py
new file mode 100644
index 00000000..34898f9c
--- /dev/null
+++ b/tests/test_native_columnar_temporal_review_fixes.py
@@ -0,0 +1,190 @@
+"""Review of the temporal columnar kinds (#333): a value the image cannot take is refused in Python, before the crossing, and never wraps.
+
+``T`` and ``D`` columns carry integers that the image turns into ``LocalDateTime`` and ``LocalDate``. A ``date64`` past int32 days used
+to wrap in the int32 cast and a ``timestamp[s]`` past the years of ``java.time`` threw ``DateTimeException`` in the middle of a batch. Now
+``columnar.pack`` returns None for both, so the call takes the row path, which raises what it raised before (nothing stored, no
+transaction left open on the synchronous call; ``may have been stored`` for a parallel call, which has handed rows over by then). Also
+pinned: a chunk with a date-time column and a float column holding NaN goes columnar now and stores the same NaN that ``INSERT INTO t
+CONTENT :row`` stores (both end in ``MutableDocument.putProperty``). Needs the native library: conftest.py collects this file on the
+native backend only.
+"""
+
+import datetime as dt
+import decimal
+import math
+
+import numpy as np
+import pytest
+
+from arcadedb._native import columnar, lib  # noqa: E402
+from arcadedb.exceptions import ArcadeDBError
+
+pa = pytest.importorskip("pyarrow")
+
+COLP = "arcadedb_n_insert_colp"
+DAY_MS = 86_400_000
+LDT_MIN_SECONDS, LDT_MAX_SECONDS = -31_557_014_167_219_200, 31_556_889_864_403_199
+NO_TYPES = columnar.Declared(known={})
+
+
+class Spy:
+    def __init__(self, monkeypatch):
+        self.calls = []
+        real = lib.call
+
+        def call(name, *args):
+            self.calls.append(name)
+            return real(name, *args)
+
+        monkeypatch.setattr(lib, "call", call)
+
+
+def _count(db, name):
+    sql = f"SELECT count(*) AS n FROM {name}"  # nosec B608 - test-owned type name
+    return db.query("sql", sql).first().get("n")
+
+
+def _kinds(columns):
+    packed = columnar.pack(columns, NO_TYPES)
+    return None if packed is None else [c.kind for c in packed[1]]
+
+
+def test_a_date64_past_int32_days_is_not_laid_out_and_never_wraps():
+    wraps = pa.array([DAY_MS * (2**31 + 5)], pa.int64()).cast(pa.date64())
+    assert columnar.pack({"d": wraps}, NO_TYPES) is None
+    below = pa.array([DAY_MS * (-(2**31) - 1)], pa.int64()).cast(pa.date64())
+    assert columnar.pack({"d": below}, NO_TYPES) is None
+    one_bad = pa.array([0, DAY_MS, DAY_MS * 2**33], pa.int64()).cast(pa.date64())
+    assert columnar.pack({"d": one_bad}, NO_TYPES) is None  # one value is enough
+
+
+def test_a_date64_at_the_int32_limits_stays_columnar_and_exact():
+    edge = pa.array(
+        [DAY_MS * (2**31 - 1), DAY_MS * -(2**31), None], pa.int64()
+    ).cast(pa.date64())
+    rows, packed = columnar.pack({"d": edge}, NO_TYPES)
+    assert rows == 3 and packed[0].kind == "D"
+    import ctypes
+
+    import numpy as np
+
+    addr, length = packed[0].data
+    days = np.frombuffer(ctypes.string_at(addr, length), dtype=np.int32)
+    assert days.tolist() == [2**31 - 1, -(2**31), 0]
+
+
+def test_a_timestamp_past_the_years_of_java_time_is_not_laid_out():
+    assert (
+        columnar.pack({"t": pa.array([0, 2**62], pa.timestamp("s"))}, NO_TYPES) is None
+    )
+    over = pa.array([LDT_MAX_SECONDS + 1], pa.timestamp("s"))
+    assert columnar.pack({"t": over}, NO_TYPES) is None
+    under = pa.array([LDT_MIN_SECONDS - 1], pa.timestamp("s"))
+    assert columnar.pack({"t": under}, NO_TYPES) is None
+
+
+def test_a_timestamp_at_the_limits_stays_columnar():
+    edge = pa.array([LDT_MAX_SECONDS, LDT_MIN_SECONDS, None], pa.timestamp("s"))
+    assert _kinds({"t": edge}) == ["T"]
+    # an int64 of milliseconds, microseconds or nanoseconds is always inside the years of java.time
+    for unit in ("ms", "us", "ns"):
+        big = pa.array([2**63 - 1, -(2**63)], pa.timestamp(unit))
+        assert _kinds({"t": big}) == ["T"], unit
+
+
+@pytest.mark.parametrize("text", ["NaN", "-NaN", "sNaN", "Infinity", "-Infinity"])
+def test_a_non_finite_decimal_is_never_laid_out(text):
+    # new BigDecimal("NaN") throws in the image, in the middle of a batch: the column must stay in Python, in every container
+    bad = decimal.Decimal(text)
+    assert columnar.pack({"x": [decimal.Decimal("1.5"), bad]}, NO_TYPES) is None
+    assert columnar.pack({"x": [None, bad, None]}, NO_TYPES) is None
+    as_object = np.array([decimal.Decimal("1"), bad], dtype=object)
+    assert columnar.pack({"x": as_object}, NO_TYPES) is None
+    finite = columnar.pack({"x": [decimal.Decimal("1.5"), None]}, NO_TYPES)
+    assert finite is not None and finite[1][0].kind == "n"  # the finite column crosses
+
+
+def _type(db, name):
+    db.command("sql", f"CREATE DOCUMENT TYPE {name}")
+    db.command("sql", f"CREATE PROPERTY {name}.id INTEGER")
+
+
+def test_a_timestamp_the_image_cannot_take_stores_nothing_and_leaves_no_transaction(
+    temp_db,
+):
+    _type(temp_db, "Col")
+    table = pa.table(
+        {
+            "id": pa.array([1, 2], pa.int64()),
+            "t": pa.array([0, 2**62], pa.timestamp("s")),
+        }
+    )
+    with pytest.raises(ArcadeDBError):
+        temp_db.insert_arrow("Col", table)
+    assert not temp_db.is_transaction_active()
+    assert _count(temp_db, "Col") == 0
+
+
+def test_a_date_the_image_cannot_take_stores_nothing_and_leaves_no_transaction(temp_db):
+    _type(temp_db, "Col")
+    wraps = pa.array([0, DAY_MS * (2**31 + 5)], pa.int64()).cast(pa.date64())
+    table = pa.table({"id": pa.array([1, 2], pa.int64()), "d": wraps})
+    with pytest.raises(ArcadeDBError):
+        temp_db.insert_arrow("Col", table)
+    assert not temp_db.is_transaction_active()
+    assert _count(temp_db, "Col") == 0
+
+
+def test_a_parallel_load_of_such_a_timestamp_says_that_rows_may_have_been_stored(
+    temp_db,
+):
+    _type(temp_db, "Col")
+    table = pa.table(
+        {
+            "id": pa.array([1, 2], pa.int64()),
+            "t": pa.array([0, 2**62], pa.timestamp("s")),
+        }
+    )
+    with pytest.raises(ArcadeDBError, match="may have been stored"):
+        temp_db.insert_arrow("Col", table, parallel=True)
+    temp_db.async_executor().wait_completion()
+    assert (
+        _count(temp_db, "Col") <= 1
+    )  # the row before the bad one may have been handed over
+
+
+def test_a_datetime_chunk_with_nan_stores_the_nan_the_statement_path_stores(
+    temp_db, monkeypatch
+):
+    for name in ("Col", "Ref"):
+        _type(temp_db, name)
+        temp_db.command("sql", f"CREATE PROPERTY {name}.ts DATETIME_MICROS")
+        temp_db.command("sql", f"CREATE PROPERTY {name}.val DOUBLE")
+    rows = [
+        {
+            "id": i,
+            "ts": dt.datetime(2020, 1, 1, 0, 0, i),
+            "val": float("nan") if i % 2 else 1.5 * i,
+        }
+        for i in range(10)
+    ]
+    spy = Spy(monkeypatch)
+    temp_db.insert_many("Col", rows)
+    assert (
+        spy.calls.count(COLP) == 1
+    )  # the date-time column no longer forces the JSON path
+    sql = "INSERT INTO `Ref` CONTENT :row"
+    temp_db.begin()
+    temp_db._command_many("sql", sql, [{"row": row} for row in rows])
+    temp_db.commit()
+
+    def stored(name):
+        select = f"SELECT FROM {name} ORDER BY id"  # nosec B608 - test-owned type name
+        return [
+            (r["id"], type(r["val"]).__name__, repr(r["val"]), repr(r["ts"]))
+            for r in temp_db.query("sql", select).to_list()
+        ]
+
+    got, want = stored("Col"), stored("Ref")
+    assert got == want
+    assert all(math.isnan(float(v[2])) for v in got if v[0] % 2)
diff --git a/tests/test_native_mixed_chunk.py b/tests/test_native_mixed_chunk.py
index db4162cf..cf409cb6 100644
--- a/tests/test_native_mixed_chunk.py
+++ b/tests/test_native_mixed_chunk.py
@@ -169,7 +169,8 @@ def test_a_pandas_timestamp_is_stored_as_the_statement_path_stores_it(
         _statement_path(db, "Ref", rows)
         spy.clear()
         db.insert_many("Bulk", rows)
-        assert spy.count(MANY) == 0 and spy.count(ROWS) == 1
+        # one column of Timestamps is the kind T of the columnar entry (#316): microseconds, as the tagged path keeps them
+        assert spy.count(MANY) == 0 and spy.count(ROWS) == 0 and spy.count(COLP) == 1
         assert _stored(db, "Bulk") == _stored(db, "Ref")
 
 
@@ -249,8 +250,10 @@ def test_an_uncarriable_value_sends_its_own_row_to_the_statement_not_the_chunk(
         spy.clear()
         db.insert_many("Bulk", rows)
         assert spy.count(MANY) == 1, "one statement run for the one uncarriable row"
+        # the rows before and after it stay on the bulk entry; its columnar form carries the date-time column now, so each run
+        # crosses on arcadedb_n_insert_colp (before it was the JSON entry, arcadedb_n_insert_rows)
         assert (
-            spy.count(ROWS) == 2
+            spy.count(COLP) == 2 and spy.count(ROWS) == 0
         ), "the rows before and after it stay on the bulk entry"
         assert _stored(db, "Bulk") == _stored(db, "Ref")