Bridge: float vectors cross as one binary buffer, not as JSON text or one JPype call per element (stacked on #29)

src/arcadedb_embedded/core.py

@@ -29,7 +29,12 @@
 from .jvm import start_jvm
 from .results import ResultSet
 from .transactions import TransactionContext
-from .type_conversion import _is_numpy_bool, convert_python_to_java, json_bulk_dumps
+from .type_conversion import (
+    _is_numpy_bool,
+    convert_python_to_java,
+    json_bulk_dumps,
+    split_float_vectors,
+)
 from .vector import to_java_float_array
 
 _LOGGER = get_logger(__name__)
@@ -184,7 +189,8 @@ def _column_to_java(name, values):
     and any other sequence convert per element into an ``Object[]`` (a ``str``
     reuses one Java String per distinct value, ``None`` is a null). A pandas
     ``Series`` is read through ``to_numpy``, with a nullable dtype's ``<NA>``
-    as ``None``. Kinds that do not cross natively raise ``TypeError`` rather
+    as ``None``. A 2-D ``float32`` array is a column of vectors: it crosses as
+    one ``float[]`` and each document gets one row of it. Kinds that do not cross natively raise ``TypeError`` rather
     than being stored as something else.
     """
     if hasattr(values, "to_numpy") and not isinstance(values, (list, tuple)):
@@ -194,9 +200,26 @@ def _column_to_java(name, values):
         else:  # a pandas extension dtype (Int64, string, boolean, category): its <NA> is a null
             values = values.to_numpy(dtype=object, na_value=None)
     if _np is not None and isinstance(values, _np.ndarray):
+        if values.ndim == 2 and values.dtype == _np.float32 and values.shape[1] > 0:
+            # a column of vectors: ONE float[] holds the whole matrix and the
+            # documents get a row each, Java-side (FloatVectors.Matrix)
+            return (
+                _java_class("com.arcadedb.python.FloatVectors$Matrix")(
+                    jpype.JArray(jpype.JFloat)(
+                        _np.ascontiguousarray(values).reshape(-1)
+                    ),
+                    int(values.shape[1]),
+                ),
+                int(values.shape[0]),
+            )
         if values.ndim != 1:
+            hint = (
+                "; a 2-D float32 matrix is accepted as a column of vectors"
+                if values.ndim == 2
+                else ""
+            )
             raise ValueError(
-                f"column {name!r} is {values.ndim}-dimensional; a column is one-dimensional"
+                f"column {name!r} is {values.ndim}-dimensional; a column is one-dimensional{hint}"
             )
         kind = values.dtype.kind
         if kind in "iu":
@@ -664,7 +687,13 @@ def insert_many(
         generator of any length can be loaded (#294). Values must be
         JSON-representable (str/int/float/bool/None and nested lists/dicts);
         a chunk containing other types (e.g. datetime, bytes) falls back
-        transparently to the per-row path.
+        transparently to the per-row path. A property declared
+        ``ARRAY_OF_FLOATS`` whose value in every row of a chunk is a list or a
+        numpy array of numbers of one length crosses as one binary float32
+        matrix beside the JSON instead of as text (``FloatVectors``); the stored
+        values are the list's, since the engine converts a list for such a
+        property to float. A vector that is ragged, missing or null in a row
+        stays in the JSON.
 
         Args:
             type_name: Target document type (must exist).
@@ -770,8 +799,11 @@ def insert_many(
 
     def _insert_many_chunk(self, type_name, chunk):
         """Insert one chunk of insert_many rows inside the open transaction."""
+        # Float vectors for ARRAY_OF_FLOATS properties cross as one binary matrix
+        # beside the JSON, not as text (see split_float_vectors).
+        split = split_float_vectors(self._java_db, type_name, chunk)
         try:
-            payload = json_bulk_dumps(chunk)
+            payload = json_bulk_dumps(split[0] if split else chunk)
         except (TypeError, ValueError):
             # Values the JSON text cannot carry unchanged (numpy integer
             # scalars, which json.dumps rejects; an integer beyond 64 bits,
@@ -787,9 +819,13 @@ def _insert_many_chunk(self, type_name, chunk):
         try:
             # commitEvery 0 inside a transaction that is already open: the
             # helper only inserts, and the transaction stays with insert_many.
-            _java_class("com.arcadedb.python.DocumentBatcher").insertManyJson(
-                self._java_db, type_name, payload, 0, False
-            )
+            batcher = _java_class("com.arcadedb.python.DocumentBatcher")
+            if split:
+                batcher.insertManyJson(
+                    self._java_db, type_name, payload, 0, False, split[1], split[2]
+                )
+            else:
+                batcher.insertManyJson(self._java_db, type_name, payload, 0, False)
         except Exception as e:
             raise ArcadeDBError(f"Failed to bulk-insert into '{type_name}': {e}") from e
 
@@ -801,8 +837,9 @@ def _insert_many_parallel(self, type_name, first, rows):
         chunk = first
         try:
             while chunk:
+                split = split_float_vectors(self._java_db, type_name, chunk)
                 try:
-                    payload = json_bulk_dumps(chunk)
+                    payload = json_bulk_dumps(split[0] if split else chunk)
                 except (TypeError, ValueError):
                     payload = None
                 if payload is None:
@@ -827,9 +864,14 @@ def _insert_many_parallel(self, type_name, first, rows):
                         raise
                 else:
                     try:
-                        failures = batcher.insertManyJsonParallel(
-                            self._java_db, type_name, payload
-                        )
+                        if split:
+                            failures = batcher.insertManyJsonParallel(
+                                self._java_db, type_name, payload, split[1], split[2]
+                            )
+                        else:
+                            failures = batcher.insertManyJsonParallel(
+                                self._java_db, type_name, payload
+                            )
                     except Exception as e:
                         raise self._parallel_load_failed(type_name, e) from e
                     # The writers report a rejected record only through its
@@ -887,7 +929,10 @@ def insert_columns(
         engine): 8.2 s against 18.4 s for ``insert_many`` (2.24x); every arm
         stored the same sums and count. Use it whenever the data already lives
         in columns (a pandas ``DataFrame``, a parquet batch, numpy arrays);
-        ``insert_many`` stays the path for a list of dicts.
+        ``insert_many`` stays the path for a list of dicts. A 2-D ``float32``
+        array is a column of vectors: the whole matrix crosses as one
+        ``float[]`` and each document gets one row of it (an ``ARRAY_OF_FLOATS``
+        property).
 
         Args:
             type_name: Target document type (must exist).

src/arcadedb_embedded/graph_batch.py

@@ -11,7 +11,13 @@
 from ._logging import get_logger, log_swallowed_exception
 from .exceptions import ArcadeDBError
 from .graph import Document, Vertex
-from .type_conversion import convert_python_to_java, json_bulk_scalar_ok
+from .type_conversion import (
+    convert_python_to_java,
+    declared_float_vectors,
+    is_vector_candidate,
+    json_bulk_scalar_ok,
+    split_float_vectors,
+)
 
 _LOGGER = get_logger(__name__)
 
@@ -180,7 +186,10 @@ def create_vertices(
             type_name: Vertex type name.
             count_or_properties: Either an integer vertex count or an iterable of
                 per-vertex property dictionaries. Use `None` or `{}` for vertices
-                without properties when passing an iterable.
+                without properties when passing an iterable. A property declared
+                ``ARRAY_OF_FLOATS`` whose value in every row is a list or a numpy
+                array of numbers of one length crosses as one binary float32
+                matrix instead of JSON text.
         """
         self._check_not_closed()
         # From engine 26.10.1 the engine refuses to run inside a transaction the
@@ -228,23 +237,31 @@ def create_vertices(
 
     def _create_vertices_json_bulk(self, type_name, rows):
         """Bulk path: rows cross as one JSON string, RIDs return as one joined
-        string (two bulk copies instead of per-value marshaling — measured
-        ~2.4x). Returns None when unavailable/unsuitable so the caller falls
-        back to the property-matrix path."""
+        string (two bulk copies instead of per-value marshaling, measured
+        ~2.4x). The float vectors of ARRAY_OF_FLOATS properties cross as one
+        binary matrix beside the JSON instead of as text (split_float_vectors).
+        Returns None when unavailable/unsuitable so the caller falls back to
+        the property-matrix path."""
         try:
             vertex_batcher = jpype.JClass("com.arcadedb.python.VertexBatcher")
+            declared = set(declared_float_vectors(self._java_db, type_name))
         except Exception:
             return None
 
         for row in rows:
             if row:
                 for key, value in row.items():
-                    if type(key) is not str or not json_bulk_scalar_ok(value):
+                    if type(key) is not str:
+                        # a property name that is not a str, which json.dumps would
+                        # write as text ({1: "a"} as "1") where the matrix path
+                        # refuses it
+                        return None
+                    if key in declared and is_vector_candidate(value):
+                        continue  # the vector split below takes it, or its chunk goes the matrix way
+                    if not json_bulk_scalar_ok(value):
                         # e.g. datetime/bytes (the matrix path preserves types), or a
                         # value JSON would change: an integer beyond 64 bits, NaN or
-                        # Infinity, a lone surrogate; or a property name that is not
-                        # a str, which json.dumps would write as text ({1: "a"} as
-                        # "1") where the matrix path refuses it
+                        # Infinity, a lone surrogate
                         return None
 
         import json
@@ -253,17 +270,39 @@ def _create_vertices_json_bulk(self, type_name, rows):
         rids = []
         for start in range(0, len(rows), self._BULK_CHUNK):
             chunk = rows[start : start + self._BULK_CHUNK]
-            joined = str(
-                vertex_batcher.createVerticesJson(
-                    self._java_graph_batch,
-                    type_name,
-                    json.dumps([row or {} for row in chunk]),
+            split = split_float_vectors(self._java_db, type_name, chunk, declared)
+            rest = split[0] if split else chunk
+            if declared and not self._json_bulk_ok(rest):
+                # a list the matrix cannot carry (ragged, a null, not numbers)
+                # is left in this chunk: the property-matrix path takes the chunk
+                java_rids = self._java_graph_batch.createVertices(
+                    type_name, self._to_java_property_matrix(chunk)
                 )
-            )
+                rids.extend(str(rid) for rid in java_rids)
+                continue
+            text = json.dumps([row or {} for row in rest])
+            if split:
+                joined = vertex_batcher.createVerticesJson(
+                    self._java_graph_batch, type_name, text, split[1], split[2]
+                )
+            else:
+                joined = vertex_batcher.createVerticesJson(
+                    self._java_graph_batch, type_name, text
+                )
+            joined = str(joined)
             if joined:
                 rids.extend(joined.split(";"))
         return rids
 
+    @staticmethod
+    def _json_bulk_ok(rows):
+        for row in rows:
+            if row:
+                for value in row.values():
+                    if not json_bulk_scalar_ok(value):
+                        return False
+        return True
+
     def new_edge(
         self,
         source_vertex_or_rid,

src/arcadedb_embedded/type_conversion.py

@@ -13,6 +13,11 @@
 
 import jpype
 
+try:
+    import numpy as _np
+except ImportError:  # numpy is optional
+    _np = None
+
 
 class _JavaTimeTypes(NamedTuple):
     instant: Any
@@ -506,6 +511,16 @@ def convert_python_to_java(value: Any) -> Any:
         # Not a bool subclass: JPype would store it as the Double 1.0 or 0.0.
         return bool(value)
 
+    if _np is not None and isinstance(value, _np.ndarray):
+        # A numpy array reached the JVM as a TypeError ("only integer scalar
+        # arrays can be converted to a scalar index") from JPype. A float32 vector
+        # crosses as ONE float[] (one buffer copy), what an ARRAY_OF_FLOATS
+        # property stores; any other array is the list it would have been
+        # (exact values: a float64 array keeps its doubles).
+        if value.ndim == 1 and value.dtype == _np.float32:
+            return jpype.JArray(jpype.JFloat)(value)
+        return convert_python_to_java(value.tolist())
+
     java_python_types = _get_java_python_types()
 
     if isinstance(value, Decimal):
@@ -707,3 +722,92 @@ def json_bulk_dumps(rows: Any) -> str:
                     value
                 )  # nested lists and dicts, and subclasses of the types above
     return json.dumps(rows, allow_nan=False)
+
+
+_FLOAT_VECTORS = None  # the bridge's FloatVectors class once resolved; False when the bridge jar lacks it
+
+
+def _float_vectors_class():
+    global _FLOAT_VECTORS
+    if _FLOAT_VECTORS is None:
+        try:
+            _FLOAT_VECTORS = jpype.JClass("com.arcadedb.python.FloatVectors")
+        except Exception:  # an older bridge jar: the JSON text carries the vectors
+            _FLOAT_VECTORS = False
+    return _FLOAT_VECTORS or None
+
+
+def declared_float_vectors(java_db, type_name):
+    """The names of the ARRAY_OF_FLOATS properties of `type_name` (inherited ones too), [] when there are none or no numpy."""
+    vectors = _float_vectors_class() if _np is not None else None
+    if vectors is None:
+        return []
+    return [str(name) for name in vectors.declared(java_db, type_name)]
+
+
+def is_vector_candidate(value):
+    """A value split_float_vectors may take out of the JSON: a list or a numpy array."""
+    return type(value) is list or (_np is not None and isinstance(value, _np.ndarray))
+
+
+def split_float_vectors(java_db, type_name, rows, declared=None):
+    """Take the float vectors of `rows` out of the JSON text of a bulk load.
+
+    A list of floats costs Python about 450 ns a number to write as text and the
+    engine about as much again to parse it, so a vector load spent most of its
+    time on the text. For each property of `type_name` that is declared
+    ARRAY_OF_FLOATS, whose value in EVERY row is a list or a numpy array of
+    numbers of one length, the vectors cross as one float32 matrix instead
+    (``FloatVectors.Matrix``, one buffer copy). The engine converts the numbers of a list to float for
+    such a property, so the float32 matrix stores the same values. Every other
+    property, and a column that is missing in a row, ragged, not numeric, or
+    null in a row, stays in the JSON.
+
+    Returns ``(rest_rows, names, matrices)``: `rest_rows` are the rows without
+    those properties, `names` a ``String[]`` and `matrices` an ``Object[]`` of
+    ``FloatVectors.Matrix`` for the bridge; or None when no property qualifies
+    (the caller sends `rows` as they are).
+    """
+    if _np is None or not rows:
+        return None
+    vectors = _float_vectors_class()
+    if vectors is None:
+        return None
+    if declared is None:
+        declared = declared_float_vectors(java_db, type_name)
+    taken = []
+    matrices = []
+    for name in declared:
+        try:
+            column = [row[name] for row in rows]
+        except (KeyError, TypeError):
+            continue
+        if not is_vector_candidate(column[0]):
+            continue
+        try:
+            array = _np.array(column)
+        except (ValueError, TypeError):
+            continue
+        if array.ndim != 2 or array.shape[1] == 0 or array.dtype.kind not in "fiu":
+            continue
+        with _np.errstate(
+            all="ignore"
+        ):  # a double beyond float32 becomes Infinity, as in the engine
+            flat = _np.ascontiguousarray(array, dtype=_np.float32).reshape(-1)
+        taken.append(name)
+        matrices.append(
+            vectors.Matrix(jpype.JArray(jpype.JFloat)(flat), int(array.shape[1]))
+        )
+    if not taken:
+        return None
+    if len(taken) == 1:
+        only = taken[0]
+        rest = [{k: v for k, v in row.items() if k != only} for row in rows]
+    else:
+        gone = set(taken)
+        rest = [{k: v for k, v in row.items() if k not in gone} for row in rows]
+    return (
+        rest,
+        jpype.JArray(jpype.JString)(taken),
+        jpype.JArray(jpype.JObject)(matrices),
+    )

src/arcadedb_embedded/vector.py

@@ -386,10 +386,19 @@ def _build_allowed_rids_set(self, allowed_rids):
         if not allowed_rids:
             return None
 
+        from .graph import _graph_calls
+
+        values = list(allowed_rids)
+        calls = _graph_calls()
+        if calls is not None and all(type(v) is str for v in values):
+            # RIDs given as text: one crossing builds the whole set (a RID
+            # constructor and an add() call per RID took about 5 us each)
+            return calls.ridSet(jpype.JArray(jpype.JString)(values))
+
         HashSet = jpype.JClass("java.util.HashSet")
 
         allowed_rids_set = HashSet()
-        for rid_value in allowed_rids:
+        for rid_value in values:
             allowed_rids_set.add(self._database.to_java_rid(rid_value))
         return allowed_rids_set