Native append_samples: numeric columns cross by address, not as .tolist() and JSON text

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

@@ -574,8 +574,67 @@ private static boolean isVertexType(final Database db, final String typeName) {
    * cb, err} | on_ok {cb} | on_error {cb} | append_samples {type, ts, cols} | insert_parallel {type, rows} | failures {fid}.
    */
   @CEntryPoint(name = "arcadedb_n_async")
-  @SuppressWarnings("unchecked")
   public static CCharPointer async(final IsolateThread thread, final long dbh, final CCharPointer op, final CCharPointer argsJson) {
+    return asyncImpl(dbh, op, argsJson, 0L, 0);
+  }
+
+  /**
+   * Same as arcadedb_n_async plus an address table (pointer and count of longs, TS_SLOTS per slot: data address, element count) that the
+   * append_samples arguments point into, as arcadedb_n_insert_colp reads its columns: "ts" and every entry of "cols" is either a JSON array
+   * of values, as for arcadedb_n_async, or the kind of a column in the table, "q" (int64) or "d" (float64), slot 0 being the timestamps and
+   * slot 1 + i the column i. "n" is the number of samples. The caller keeps every buffer alive and unchanged until the call returns.
+   */
+  @CEntryPoint(name = "arcadedb_n_async_bin")
+  public static CCharPointer asyncBin(final IsolateThread thread, final long dbh, final CCharPointer op, final CCharPointer argsJson,
+      final CCharPointer table, final int nTable) {
+    return asyncImpl(dbh, op, argsJson, table.rawValue(), nTable);
+  }
+
+  /** Longs per slot of the address table of arcadedb_n_async_bin: the data address and the element count. */
+  private static final int TS_SLOTS = 2;
+
+  /** The data address of a table slot, checked to hold n elements of the kind: "q" for the timestamps, "q" or "d" for a column. */
+  private static long sampleAddress(final long table, final int nTable, final int slot, final Object kind, final long n, final boolean timestamps) {
+    final String what = timestamps ? "the timestamps" : "column " + (slot - 1);
+    if (!"q".equals(kind) && (timestamps || !"d".equals(kind)))
+      throw new IllegalArgumentException(what + " has the unknown kind " + kind);
+    if (n < 0 || n > Integer.MAX_VALUE)
+      throw new IllegalArgumentException("sample count out of range: " + n);
+    if (table == 0 || nTable < TS_SLOTS * (slot + 1))
+      throw new IllegalArgumentException(what + " is not in the address table of " + nTable + " entries");
+    final Pointer t = WordFactory.pointer(table);
+    final long address = t.readLong(WordFactory.signed(8L * TS_SLOTS * slot));
+    final long count = t.readLong(WordFactory.signed(8L * (TS_SLOTS * slot + 1)));
+    if (count != n || (n > 0 && address == 0))
+      throw new IllegalArgumentException(what + " has " + count + " values in the address table, " + n + " samples");
+    return address;
+  }
+
+  /** A column of the table as the boxed values the text path gives: a Long for "q", a Double for "d" (a NaN is the canonical NaN of the text). */
+  private static Object[] sampleColumn(final long address, final boolean doubles, final int n) {
+    final Pointer p = WordFactory.pointer(address);
+    final Object[] values = new Object[n];
+    for (int i = 0; i < n; i++) {
+      final long bits = p.readLong(WordFactory.signed(8L * i));
+      if (doubles) {
+        final double d = Double.longBitsToDouble(bits);
+        values[i] = d != d ? Double.NaN : d;
+      } else
+        values[i] = bits;
+    }
+    return values;
+  }
+
+  private static long[] sampleLongs(final long address, final int n) {
+    final Pointer p = WordFactory.pointer(address);
+    final long[] values = new long[n];
+    for (int i = 0; i < n; i++)
+      values[i] = p.readLong(WordFactory.signed(8L * i));
+    return values;
+  }
+
+  @SuppressWarnings("unchecked")
+  private static CCharPointer asyncImpl(final long dbh, final CCharPointer op, final CCharPointer argsJson, final long table, final int nTable) {
     try {
       final DatabaseInternal db = (DatabaseInternal) ArcadeNative.db(dbh);
       final DatabaseAsyncExecutor ex = db.async();
@@ -726,14 +785,26 @@ else if (a.get("slot") instanceof Number slot)
               event("scan_error", "\"rid\":\"" + rid + "\"," + errorBody(t))));
       }
       case "append_samples" -> {
-        final List<Object> ts = (List<Object>) a.get("ts");
-        final long[] stamps = new long[ts.size()];
-        for (int i = 0; i < stamps.length; i++)
-          stamps[i] = ((Number) ts.get(i)).longValue();
+        final Object tsArg = a.get("ts");
+        final long n = a.get("n") instanceof Number num ? num.longValue() : -1L;
+        final long[] stamps;
+        if (tsArg instanceof String kind)
+          stamps = sampleLongs(sampleAddress(table, nTable, 0, kind, n, true), (int) n);
+        else {
+          final List<Object> ts = (List<Object>) tsArg;
+          stamps = new long[ts.size()];
+          for (int i = 0; i < stamps.length; i++)
+            stamps[i] = ((Number) ts.get(i)).longValue();
+        }
         final List<Object> cols = (List<Object>) a.get("cols");
         final Object[][] columns = new Object[cols.size()][];
-        for (int i = 0; i < columns.length; i++)
-          columns[i] = ((List<Object>) cols.get(i)).toArray();
+        for (int i = 0; i < columns.length; i++) {
+          final Object col = cols.get(i);
+          if (col instanceof String kind)
+            columns[i] = sampleColumn(sampleAddress(table, nTable, i + 1, kind, n, false), "d".equals(kind), (int) n);
+          else
+            columns[i] = ((List<Object>) col).toArray();
+        }
         ex.appendSamples((String) a.get("type"), stamps, columns);
       }
       case "insert_parallel" -> {

src/arcadedb/_native/codec.py

@@ -222,6 +222,47 @@ def _dump(obj, binary):
     return text, buf, state[1]
 
 
+_INT64_MAX = 2**63 - 1
+_NS_DATETIME = None if _np is None else _np.dtype("datetime64[ns]")
+
+
+def sample_array(values, timestamps=False):
+    """The int64 or float64 array that stands for a column of ``append_samples`` in the address table of ``arcadedb_n_async_bin``, or
+    None when the column keeps the JSON text it has always had. It is the array whose ``.tolist()`` the text path writes, value for
+    value: an int, uint (uint64 only inside the int64 range) or bool array is int64, a float array narrower than long double is float64
+    (a float32 widens exactly), and, for the timestamps only, a ``datetime64[ns]`` array without NaT is its int64 count (what ``.tolist()``
+    gives for that unit). Anything else, a list, a subclass of ndarray, a 2-D or empty array, a float array as timestamps, other
+    ``datetime64`` units (``int()`` refuses their datetimes on the text path), a ``datetime64[ns]`` of non-native byte order, returns None.
+    """
+    if (
+        _np is None
+        or type(values) is not _np.ndarray
+        or values.ndim != 1
+        or not values.size
+    ):
+        return None
+    dtype = values.dtype
+    kind = dtype.kind
+    if kind in "ib":
+        return _np.ascontiguousarray(values, dtype=_np.int64)
+    if kind == "u":
+        if dtype.itemsize == 8 and int(values.max()) > _INT64_MAX:
+            return None
+        return _np.ascontiguousarray(values, dtype=_np.int64)
+    if kind == "f":
+        if timestamps or dtype.itemsize > 8:
+            return None
+        return _np.ascontiguousarray(values, dtype=_np.float64)
+    if kind == "M" and timestamps and dtype == _NS_DATETIME:
+        counts = _np.ascontiguousarray(values.view(_np.int64))
+        if (
+            counts == _np.iinfo(_np.int64).min
+        ).any():  # NaT: the text path raises on its None
+            return None
+        return counts
+    return None
+
+
 def row_default(value):
     """json.dumps hook for the rows of a bulk load: a NumPy float array that is not float32 is the list it would have been
     (a float64 array stored on an undeclared property keeps its doubles, as on the JPype path); the rest as :func:`_default`.

src/arcadedb/_native/extras.py

@@ -20,7 +20,7 @@
 from ..exceptions import ArcadeDBError
 from ..graph_batch import GraphBatch as _BaseGraphBatch
 from . import interrupts, lib, not_yet
-from .codec import _PLAIN, _default, _prep, loads
+from .codec import _PLAIN, _default, _np, _prep, loads, sample_array
 
 # -------------------------------------------------------------------------------------------------------- upcall
 
@@ -605,20 +605,69 @@ def plain(values):
                 values = values.astype("int64")
             return values.tolist() if hasattr(values, "tolist") else list(values)
 
-        stamps = [int(t) for t in plain(timestamps)]
+        # numpy int and float columns (and an integer timestamp array) cross by address, the rest as the JSON text they always had
+        ts_array = sample_array(timestamps, True) if lib.TS_BIN else None
+        if ts_array is not None:
+            stamps, count = None, len(ts_array)
+        else:
+            stamps = [int(t) for t in plain(timestamps)]
+            count = len(stamps)
         if primitive:
-            # the engine's primitive TimeSeriesBatch is a JVM-boundary optimisation (no boxing per value); here the rows cross
-            # as one JSON text either way, so the columns take the same path, with the primitive path's length check
+            # the engine's primitive TimeSeriesBatch is a JVM-boundary optimisation (no boxing per value); here the columns
+            # take the same path as without it, with the primitive path's length check
             for index, values in enumerate(column_values):
-                if len(values) != len(stamps):
+                if len(values) != count:
                     raise ValueError(
-                        f"column {index} has {len(values)} values for {len(stamps)} timestamps"
+                        f"column {index} has {len(values)} values for {count} timestamps"
                     )
-        self._op(
-            "append_samples",
-            type=type_name,
-            ts=stamps,
-            cols=[plain(c) for c in column_values],
+        arrays = []
+        if lib.TS_BIN:
+            for values in column_values:
+                array = sample_array(values)
+                # a column of another length stays text: the engine pads or cuts it as it always did
+                arrays.append(
+                    array if array is not None and len(array) == count else None
+                )
+        if ts_array is None and not any(a is not None for a in arrays):
+            self._op(
+                "append_samples",
+                type=type_name,
+                ts=stamps,
+                cols=[plain(c) for c in column_values],
+            )
+            return
+        # the address table: per slot (0 the timestamps, 1 + i the column i) the data address and the element count of an array that
+        # stays alive until the call returns; a column that is text has no entry (zeros) and its values are in the JSON
+        table = _np.zeros(2 * (1 + len(column_values)), dtype=_np.int64)
+        if ts_array is not None:
+            table[0], table[1] = ts_array.ctypes.data, count
+        cols = []
+        for index, (values, array) in enumerate(zip(column_values, arrays)):
+            if array is None:
+                cols.append(plain(values))
+            else:
+                table[2 * (index + 1)], table[2 * (index + 1) + 1] = (
+                    array.ctypes.data,
+                    count,
+                )
+                cols.append("d" if array.dtype.kind == "f" else "q")
+        _call(
+            "Async executor failed",
+            "arcadedb_n_async_bin",
+            self._db._h,
+            lib.b("append_samples"),
+            lib.b(
+                _enc(
+                    {
+                        "type": type_name,
+                        "n": count,
+                        "ts": "q" if stamps is None else stamps,
+                        "cols": cols,
+                    }
+                )
+            ),
+            table.ctypes.data,
+            len(table),
         )
 
     def _edge_callback(self, callback):

src/arcadedb/_native/lib.py

@@ -31,6 +31,8 @@
 EXEC_MANY = os.environ.get("ARCADEDB_NATIVE_EXEC_MANY", "1") != "0"
 # ARCADEDB_NATIVE_NEXT_BIN: the batches after the first read through arcadedb_n_next_bin, float arrays of 8 or more elements in a float32 tail (as flag 4 of arcadedb_n_exec does for the first batch), not as number text
 NEXT_BIN = os.environ.get("ARCADEDB_NATIVE_NEXT_BIN", "1") != "0"
+# ARCADEDB_NATIVE_TS_BIN: the numpy int and float columns (and an integer timestamp array) of append_samples cross by address in arcadedb_n_async_bin, not as .tolist() and JSON text
+TS_BIN = os.environ.get("ARCADEDB_NATIVE_TS_BIN", "1") != "0"
 
 # the directory the engine log, logging.properties and the crash files go to (None until the library is loaded)
 LOG_DIR_USED = None
@@ -243,6 +245,7 @@ class _Params(ctypes.Structure):
             "arcadedb_n_gb_create": [P, L, S],
             "arcadedb_n_gb": [P, L, S, S],
             "arcadedb_n_async": [P, L, S, S],
+            "arcadedb_n_async_bin": [P, L, S, S, P, I],
             "arcadedb_n_vindex": [P, L, S, S],
             "arcadedb_n_import": [P, L, S],
             "arcadedb_n_export": [P, L, S, S, I, S],
@@ -266,7 +269,7 @@ class _Params(ctypes.Structure):
                 )
                 else ctypes.c_longlong if name == "arcadedb_n_cmd" else ctypes.c_void_p
             )
-        global BINARY_PARAMS, CMD_PATH, CLOSE_BATCH, INSERT_ROWS, EXEC_MANY, NEXT_BIN
+        global BINARY_PARAMS, CMD_PATH, CLOSE_BATCH, INSERT_ROWS, EXEC_MANY, NEXT_BIN, TS_BIN
         if not hasattr(
             lib, "arcadedb_n_exec_bin"
         ):  # a library built before the binary path
@@ -283,6 +286,10 @@ class _Params(ctypes.Structure):
             lib, "arcadedb_n_next_bin"
         ):  # a library built before the binary tail of later batches
             NEXT_BIN = False
+        if not hasattr(
+            lib, "arcadedb_n_async_bin"
+        ):  # a library built before the address table of append_samples
+            TS_BIN = False
         _isolate = isolate
         _creator = threading.get_ident()
         _local.thread = _Thread(thread, owned=False)

tests/conftest.py

@@ -63,6 +63,8 @@
     collect_ignore.append("test_native_tx_crossings.py")
     # counts the statement texts the native backend hands to the engine (the JPype backend has no such layer)
     collect_ignore.append("test_native_statement_texts.py")
+    # the address table of append_samples (arcadedb_n_async_bin), observed on the native library
+    collect_ignore.append("test_native_timeseries_binary.py")
 if not os.path.isfile(
     os.path.join(os.path.dirname(os.path.dirname(__file__)), "docs", "index.md")
 ):