Native GraphBatch: float32 vectors in a binary side buffer (arcadedb_n_gb_bin)

docs/modules/ROOT/pages/how-to/vectors.adoc

@@ -110,6 +110,10 @@ print(db.count_type("Chunk"))
 200
 ----
 
+`GraphBatch.create_vertices` rows and the properties of `new_edge` and `new_edges` take the same arrays. On the `arcadedb` (native) package, a non-empty one-dimensional `float32` array in a row or an edge property crosses in a binary side buffer, and the stored values are bit-identical to the ones the text path gave. A list or a `float64` array still crosses as number text, so make the array with `np.asarray(v, dtype=np.float32)` before the load. A laptop A/B on the native wheel (100,000 vertices for each load, `ab.py --noise`, ABBA order, 12 clean pairs for each figure, clean A/A control) gave this wall time of the new path against the old one (95 % interval of the median pair ratio): 0.517 [0.515, 0.520] at 128 dimensions, 0.218 [0.217, 0.219] at 96, and 0.271 [0.270, 0.272] at 64. A load without vectors was unchanged (1.009 [1.003, 1.012]).
+
+On the `arcadedb-jvm` package, `create_vertices` already sends the values of a property declared `ARRAY_OF_FLOATS` as one binary float32 matrix when every row holds a list or array of numbers of one length. So `float32` arrays in a property declared `ARRAY_OF_FLOATS` take the binary path on both packages.
+
 == Search
 
 === SQL

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

@@ -84,9 +84,15 @@ static void rows(final byte[] b, final int len, final RowSink sink) {
    * Parses a top-level object whose value for streamKey is an array of objects: those rows go to the sink one by one and the key maps to
    * the row count (Integer) instead of the list. Every other key is parsed as usual. A streamKey value that is not an array stays as parsed.
    */
-  @SuppressWarnings("unchecked")
   static Map<String, Object> object(final byte[] b, final int len, final String streamKey, final RowSink sink) {
-    final Json j = new Json(b, len, 0L, 0);
+    return object(b, len, streamKey, sink, 0L, 0);
+  }
+
+  /** As above; buf: the float32 side buffer that {"\u0001":["B",offset,count]} values point into (see parse), 0 for none. */
+  @SuppressWarnings("unchecked")
+  static Map<String, Object> object(final byte[] b, final int len, final String streamKey, final RowSink sink, final long buf,
+      final long bufFloats) {
+    final Json j = new Json(b, len, buf, bufFloats);
     j.ws();
     if (j.i >= j.n || j.s[j.i] != '{')
       throw new IllegalArgumentException("expected a JSON object");

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

@@ -67,6 +67,7 @@
 import org.graalvm.nativeimage.c.function.CFunctionPointer;
 import org.graalvm.nativeimage.c.function.InvokeCFunctionPointer;
 import org.graalvm.nativeimage.c.type.CCharPointer;
+import org.graalvm.nativeimage.c.type.CFloatPointer;
 import org.graalvm.nativeimage.c.type.CTypeConversion;
 import org.graalvm.word.Pointer;
 import org.graalvm.word.WordFactory;
@@ -446,8 +447,23 @@ public static CCharPointer gbCreate(final IsolateThread thread, final long dbh,
    * new_edges {src:[..], type, dst:[..], props:[..]|null}; flush; close; stats.
    */
   @CEntryPoint(name = "arcadedb_n_gb")
-  @SuppressWarnings("unchecked")
   public static CCharPointer gb(final IsolateThread thread, final long h, final CCharPointer op, final CCharPointer argsJson) {
+    return gbImpl(h, op, argsJson, 0L, 0);
+  }
+
+  /**
+   * Same as arcadedb_n_gb plus a float32 side buffer (pointer + count of floats) that values {"\u0001":["B",offset,count]} in argsJson read
+   * from, as arcadedb_n_exec_bin does: a float[] property of create_vertices rows or new_edges props, with no number text.
+   */
+  @CEntryPoint(name = "arcadedb_n_gb_bin")
+  public static CCharPointer gbBin(final IsolateThread thread, final long h, final CCharPointer op, final CCharPointer argsJson,
+      final CFloatPointer buf, final long bufFloats) {
+    return gbImpl(h, op, argsJson, buf.rawValue(), bufFloats);
+  }
+
+  @SuppressWarnings("unchecked")
+  private static CCharPointer gbImpl(final long h, final CCharPointer op, final CCharPointer argsJson, final long buf,
+      final long bufFloats) {
     try {
       final BatchEntry entry = BATCHES.get(h);
       if (entry == null)
@@ -460,9 +476,9 @@ public static CCharPointer gb(final IsolateThread thread, final long h, final CC
       if (chars.n == 0)
         a = new HashMap<>();
       else if (opName.equals("create_vertices"))
-        a = Json.object(chars.b, chars.n, "rows", row -> flats.add(varargs(row)));
+        a = Json.object(chars.b, chars.n, "rows", row -> flats.add(varargs(row)), buf, bufFloats);
       else
-        a = (Map<String, Object>) Json.parse(chars.b, chars.n, 0L, 0);
+        a = (Map<String, Object>) Json.parse(chars.b, chars.n, buf, bufFloats);
       switch (opName) {
       case "create_vertex" -> {
         final MutableVertex v = batch.createVertex((String) a.get("type"), varargs(a.get("props")));

src/arcadedb/_native/codec.py

@@ -222,6 +222,35 @@ def _dump(obj, binary):
     return text, buf, state[1]
 
 
+def _gb_default(value, state):
+    """json.dumps hook for a GraphBatch op in binary mode: a one-dimensional non-empty float32 ndarray goes to the side buffer (the
+    ``["B", offset, count]`` tag, a float[] in Java, the value the ``"F"`` text gives); every other value as :func:`_default`
+    writes it, so a list, a float64 array or a 2-D array keeps the text it has always had.
+    """
+    if (
+        _np is not None
+        and isinstance(value, _np.ndarray)
+        and value.dtype == _np.float32
+        and value.ndim == 1
+        and value.size
+    ):
+        return _default(value, state)
+    return _default(value)
+
+
+def encode_gb_args(args):
+    """The ``args`` of ``arcadedb_n_gb_bin``: ``(text, float32 side buffer bytes or None, float count)``, text as :func:`_prep` and
+    :func:`_default` write it for ``arcadedb_n_gb`` except that the float32 vectors of the rows and props are in the buffer.
+    """
+    state = [[], 0]
+    text = json.dumps(_prep(args), default=lambda v: _gb_default(v, state))
+    arrays = state[0]
+    if not arrays:
+        return text, None, 0
+    buf = arrays[0].tobytes() if len(arrays) == 1 else _np.concatenate(arrays).tobytes()
+    return text, buf, state[1]
+
+
 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, _prep, encode_gb_args, loads
 
 # -------------------------------------------------------------------------------------------------------- upcall
 
@@ -164,6 +164,20 @@ def _to_java_wal_flush_name(cls, mode: str) -> str:
         return cls._VALID_WAL_FLUSH_MODES[normalized]
 
     def _op(self, prefix: str, op: str, **args) -> dict:
+        if lib.GB_BIN:
+            # float32 vectors cross in a binary side buffer; a library without the entry keeps the text path below
+            text, buf, count = encode_gb_args(args)
+            if buf is not None:
+                return _call(
+                    prefix,
+                    "arcadedb_n_gb_bin",
+                    self._gb,
+                    lib.b(op),
+                    lib.b(text),
+                    buf,
+                    count,
+                )
+            return _call(prefix, "arcadedb_n_gb", self._gb, lib.b(op), lib.b(text))
         return _call(prefix, "arcadedb_n_gb", self._gb, lib.b(op), lib.b(_enc(args)))
 
     def new_vertex(self, type_name: str):
@@ -233,6 +247,9 @@ def _create_vertices_json_bulk(self, type_name: str, rows):
     def create_vertices(self, type_name: str, count_or_properties) -> list:
         """The RIDs of the new vertices. Rows go in chunks of the batch's ``commit_every`` (1000 unless graph_batch() got another value;
         0: 10,000), one crossing and one transaction per chunk: chunks that already went in stay stored if a later chunk fails.
+
+        A non-empty one-dimensional ``float32`` numpy array in a row crosses in a binary side buffer; a list or a ``float64`` array
+        crosses as number text (``new_edge`` and ``new_edges`` props follow the same rule).
         """
         self._check_not_closed()
         db = self._db

src/arcadedb/_native/lib.py

@@ -32,6 +32,9 @@
 # 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_GB_BIN: the float32 vectors of a GraphBatch op (create_vertices rows, new_edges props) go through arcadedb_n_gb_bin in a float32 side buffer, not as number text
+GB_BIN = os.environ.get("ARCADEDB_NATIVE_GB_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
 
@@ -242,6 +245,7 @@ class _Params(ctypes.Structure):
             "arcadedb_n_lookup": [P, L, S, S],
             "arcadedb_n_gb_create": [P, L, S],
             "arcadedb_n_gb": [P, L, S, S],
+            "arcadedb_n_gb_bin": [P, L, S, S, S, L],
             "arcadedb_n_async": [P, L, S, S],
             "arcadedb_n_vindex": [P, L, S, S],
             "arcadedb_n_import": [P, L, S],
@@ -266,7 +270,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, GB_BIN
         if not hasattr(
             lib, "arcadedb_n_exec_bin"
         ):  # a library built before the binary path
@@ -283,6 +287,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_gb_bin"
+        ):  # a library built before the side buffer of GraphBatch ops
+            GB_BIN = False
         _isolate = isolate
         _creator = threading.get_ident()
         _local.thread = _Thread(thread, owned=False)

tests/conftest.py

@@ -45,6 +45,8 @@
     collect_ignore.append("test_native_columnar_insert.py")
     # the native build's ingest defaults (insert_many and graph_batch commit size), observed on the native library
     collect_ignore.append("test_native_ingest_defaults.py")
+    # the float32 side buffer of the native GraphBatch entry (arcadedb_n_gb_bin), observed on the native library
+    collect_ignore.append("test_native_graph_batch_binary_vectors.py")
     # the native library's start-up output (#477) can only be checked with the native library loaded
     collect_ignore.append("test_native_startup_runtime.py")
     # the XML entity limits of an import, lifted by the native glue (the JVM build lifts them with JVM system properties)

tests/test_float_vector_bulk.py

@@ -291,6 +291,23 @@ def test_numpy_rows(self, vec_db, dtype):
             )
         _assert_float32_of(self._ids(vec_db), x)
 
+    @pytest.mark.parametrize("dim", [64, 96, 128])
+    def test_float32_rows_are_stored_bit_equal(self, vec_db, dim):
+        # the lanes' shapes (SIFT 128, Deep 96, e2 64): float32 values a 6-digit text would not hold, -0.0 and the extremes included.
+        # The native build sends them in a binary side buffer (test_native_graph_batch_binary_vectors.py), the JVM build as a matrix
+        x = np.random.default_rng(dim).standard_normal((N, dim)).astype(np.float32)
+        x[0, 0] = np.float32(-0.0)
+        x[1, 0] = np.finfo(np.float32).max
+        x[1, 1] = np.finfo(np.float32).tiny
+        x[1, 2] = np.finfo(np.float32).smallest_subnormal
+        with vec_db.graph_batch() as gb:
+            gb.create_vertices("Vx", [{"id": i, "embedding": x[i]} for i in range(N)])
+        stored = self._ids(vec_db)
+        assert sorted(stored) == list(range(N))
+        for i in range(N):
+            got = stored[i].astype(np.float32)
+            assert got.view(np.uint32).tolist() == x[i].view(np.uint32).tolist()
+
     def test_one_value_json_cannot_carry_sends_only_its_chunk_the_typed_way(
         self, vec_db
     ):