Native: columnar bulk insert by address for strings, nulls and vectors (no JSON text)

bindings/python/docs/development/bridge.md

@@ -44,7 +44,7 @@ crossing per batch**, receiving a bulk payload it can decode at C speed: the
 | `ResultSet.to_columns()` / fast `to_dataframe()` / `to_arrow()` | `ColumnBatcher` |
 | `Database.insert_many()` | `DocumentBatcher` |
 | `Database.insert_columns()` | `DocumentBatcher` |
-| `Database.insert_dataframe()` / `insert_arrow()` | `DocumentBatcher` (document types, through `insert_columns()`), `VertexBatcher` (vertex types, through `GraphBatch.create_vertices()`); the Arrow-to-column conversion is Python (`dataframe.py`); on the native backend the same checks run and a document batch goes through `insert_many()` (`arcadedb_n_insert_rows`), a vertex batch through `GraphBatch.create_vertices()` |
+| `Database.insert_dataframe()` / `insert_arrow()` | `DocumentBatcher` (document types, through `insert_columns()`), `VertexBatcher` (vertex types, through `GraphBatch.create_vertices()`); the Arrow-to-column conversion is Python (`dataframe.py`); on the native backend the same checks run and a document batch whose columns are numbers, bools, strings (nulls allowed) or float vectors of an `ARRAY_OF_FLOATS` property goes to `arcadedb_n_insert_colp` by address (no JSON text, `_native/columnar.py`), any other document batch through `insert_many()` (`arcadedb_n_insert_rows`), a vertex batch through `GraphBatch.create_vertices()` |
 | `ResultSet.to_polars()` | `ColumnBatcher` (through `to_arrow()`) |
 | `AsyncExecutor.append_samples()` (numpy numeric-column boxing) | `DocumentBatcher` |
 | `AsyncExecutor.append_samples(..., primitive=True)` | `TimeSeriesBatcher` |

bindings/python/src/arcadedb_embedded/_native/bulk.py

@@ -1,96 +1,71 @@
-"""Columnar bulk insert through one crossing (``arcadedb_n_insert_cols``): numpy int/float/bool columns and float32 vector
-columns (a 2-D array or a list of equal-length 1-D float arrays) go over as one buffer; anything else returns None and the
-caller falls back to the per-row path. The documents are built Java-side as ``DocumentBatcher.insertColumns`` builds them on JPype.
+"""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.
 """
 
-import json
 import os
 
-from . import lib
-
-try:
-    import numpy as _np
-except ImportError:  # pragma: no cover
-    _np = None
+from . import columnar, lib
 
 _ENABLED = os.environ.get("ARCADEDB_NATIVE_BULK", "1") != "0"
 
 
-def insert_columns_fast(db, type_name, columns, commit_every):
-    if (
-        not _ENABLED
-        or _np is None
-        or lib._lib is None
-        or not hasattr(lib._lib, "arcadedb_n_insert_cols")
-    ):
-        return None
-    cols, parts, n = [], [], None
-    for name, values in columns.items():
-        if hasattr(values, "to_numpy") and not isinstance(values, (list, tuple)):
-            return None
-        kind, dim = None, 0
-        if isinstance(values, _np.ndarray):
-            k = values.dtype.kind
-            if (
-                values.ndim == 1
-                and k in "iu"
-                and (
-                    k == "i"
-                    or not values.size
-                    or int(values.max()) <= _np.iinfo(_np.int64).max
-                )
-            ):
-                kind, arr = "q", _np.ascontiguousarray(values, dtype=_np.int64)
-            elif values.ndim == 1 and k == "f":
-                kind, arr = "d", _np.ascontiguousarray(values, dtype=_np.float64)
-            elif values.ndim == 1 and k == "b":
-                kind, arr = "b", _np.ascontiguousarray(values, dtype=_np.uint8)
-            elif values.ndim == 2 and k == "f" and values.shape[1] > 0:
-                kind, arr, dim = (
-                    "f",
-                    _np.ascontiguousarray(values, dtype=_np.float32),
-                    int(values.shape[1]),
-                )
-        elif (
-            isinstance(values, (list, tuple))
-            and values
-            and all(
-                isinstance(v, _np.ndarray)
-                and v.ndim == 1
-                and v.dtype.kind == "f"
-                and v.shape == values[0].shape
-                for v in values
-            )
-            and values[0].shape[0] > 0
-        ):
-            kind, arr, dim = (
-                "f",
-                _np.ascontiguousarray(_np.stack(values), dtype=_np.float32),
-                int(values[0].shape[0]),
-            )
-        if kind is None:
-            return None
-        rows = int(arr.shape[0])
-        if n is not None and rows != n:
-            return None  # the per-row path reports the ragged columns
-        n = rows
-        cols.append([name, kind, dim])
-        parts.append(arr.tobytes())
-    if not n:
-        return None
-    header = json.dumps(
-        {"n": n, "commit": min(int(commit_every), 2**31 - 1), "cols": cols}
+def _usable():
+    return (
+        _ENABLED
+        and columnar._np is not None
+        and lib._lib is not None
+        and hasattr(lib._lib, "arcadedb_n_insert_colp")
     )
-    buf = b"".join(parts)
+
+
+def _send(db, type_name, packed, rows, commit_every, prefix):
     from .database import _call
 
+    spec = columnar.spec_of(packed)
     reply = _call(
-        "Failed to bulk-insert",
-        "arcadedb_n_insert_cols",
+        prefix,
+        "arcadedb_n_insert_colp",
         db._h,
         lib.b(type_name),
-        lib.b(header),
-        buf,
-        len(buf),
-    )
+        lib.b(columnar.header_of(rows, commit_every, packed)),
+        spec,
+        len(spec),
+    )  # `packed` holds the arrays the addresses point into until the call returns
     return int(reply["n"])
+
+
+def insert_columns_fast(
+    db, type_name, columns, commit_every, declared=None, prefix="Failed to bulk-insert"
+):
+    """Insert ``columns`` (name -> column) in one crossing and return the row count, or None when a column cannot be sent this way.
+
+    ``declared``: a dict of declared property types when the caller already knows them (the DataFrame path), else they are read from the
+    schema if a column needs them.
+    """
+    if not _usable():
+        return None
+    packed = columnar.pack(columns, columnar.Declared(db, type_name, declared))
+    if packed is None:
+        return None
+    rows, cols = packed
+    return _send(db, type_name, cols, rows, commit_every, prefix)
+
+
+def insert_rows_columnar(db, type_name, chunk):
+    """One ``insert_many`` chunk (a list of dicts with the same keys) through the columnar entry inside the caller's transaction (no
+    commit here). Returns True when it was sent, False when the chunk needs the JSON path.
+    """
+    if not _usable():
+        return False
+    columns = columnar.rows_to_columns(chunk)
+    if columns is None:
+        return False
+    packed = columnar.pack(columns, columnar.Declared(db, type_name))
+    if packed is None:
+        return False
+    rows, cols = packed
+    _send(db, type_name, cols, rows, 0, f"Failed to bulk-insert into '{type_name}'")
+    return True