insert_many: date-times, Decimals, bytes, numpy scalars and pandas NA stay on the bulk path; an uncarriable value moves its own row, not the chunk

src/arcadedb/core.py

@@ -32,8 +32,11 @@
 from .transactions import TransactionContext
 from .type_conversion import (
     _is_numpy_bool,
+    bulk_row_ok,
     convert_python_to_java,
+    declared_float_vectors,
     json_bulk_dumps,
+    json_bulk_dumps_tagged,
     split_float_vectors,
 )
 from .vector import to_java_float_array
@@ -690,10 +693,17 @@ def insert_many(
         to one JSON string and looped Java-side (``DocumentBatcher``),
         avoiding the per-row JNI cost that caps ``new_document``-loop ingest.
         Memory stays bounded by the chunk whatever the row count, so a
-        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. A property declared
+        generator of any length can be loaded (#294). A value that sits directly
+        in a row may be JSON-representable (str/int/float/bool/None, nested
+        lists/dicts), a ``datetime`` (also ``pandas.Timestamp`` and
+        ``numpy.datetime64``), ``date``, ``Decimal``, ``bytes``/``bytearray``,
+        a numpy scalar, or ``pandas.NA``/``pandas.NaT`` (stored as null); they
+        cross inside the JSON as tagged values and are stored as the per-row path
+        stores them. A row with any other value (a date-time or Decimal inside a
+        list or dict, an integer beyond 64 bits, NaN or Infinity, a non-str dict
+        key, a lone surrogate, a ``datetime.time``) goes down the per-row path,
+        that row only: the rows around it stay on the bulk path and every row is
+        inserted in the order given. 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
@@ -817,30 +827,61 @@ def _is_vertex_type(self, type_name):
         )
 
     def _insert_many_chunk(self, type_name, chunk):
-        """Insert one chunk of insert_many rows inside the open transaction."""
+        """Insert one chunk of insert_many rows inside the open transaction.
+
+        The chunk crosses as one JSON text. A value the text cannot carry unchanged (NaN or Infinity, an integer beyond 64 bits, a date-time
+        or Decimal inside a list or map, a lone surrogate, a non-str key) sends its own row down the per-row path; the rows around it
+        still go in bulk, and every row is inserted in the order given. Dates, date-times, Decimals, bytes, numpy scalars and
+        pandas NA and NaT sitting directly in a row cross inside the text as tagged values (see json_bulk_dumps_tagged).
+        """
         # 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(split[0] if split else chunk)
+            payload = json_bulk_dumps_tagged(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,
-            # NaN or Infinity, a non-str dict key, a lone surrogate, which the
-            # engine's JSON parser would store as a different value): this
-            # chunk goes row by row (vertices for a vertex type, as the Java
-            # helper does).
-            new_record = (
-                self.new_vertex
-                if self._is_vertex_type(type_name)
-                else self.new_document
-            )
-            for row in chunk:
-                doc = new_record(type_name)
-                for k, v in row.items():
-                    doc.set(k, v)
-                doc.save()
+            self._insert_many_runs(type_name, chunk)
             return
+        self._insert_many_bulk(type_name, payload, split)
+
+    def _insert_many_runs(self, type_name, chunk):
+        """A chunk with a row the JSON text cannot carry: runs of carriable rows go in bulk, the other rows one by one, in order."""
+        skip = declared_float_vectors(self._java_db, type_name)
+        ok = [bulk_row_ok(row, skip) for row in chunk]
+        start = 0
+        n = len(chunk)
+        while start < n:
+            end = start
+            while end < n and ok[end] == ok[start]:
+                end += 1
+            run = chunk[start:end]
+            start = end
+            if not ok[end - 1]:
+                self._insert_many_per_row(type_name, run)
+                continue
+            split = split_float_vectors(self._java_db, type_name, run)
+            try:
+                payload = json_bulk_dumps_tagged(split[0] if split else run)
+            except (
+                TypeError,
+                ValueError,
+            ):  # a vector column the matrix did not take: this run only goes row by row
+                self._insert_many_per_row(type_name, run)
+                continue
+            self._insert_many_bulk(type_name, payload, split)
+
+    def _insert_many_per_row(self, type_name, rows):
+        # vertices for a vertex type, as the bulk helper does
+        new_record = (
+            self.new_vertex if self._is_vertex_type(type_name) else self.new_document
+        )
+        for row in rows:
+            doc = new_record(type_name)
+            for k, v in row.items():
+                doc.set(k, v)
+            doc.save()
+
+    def _insert_many_bulk(self, type_name, payload, split):
         try:
             # commitEvery 0 inside a transaction that is already open: the
             # helper only inserts, and the transaction stays with insert_many.
@@ -862,32 +903,31 @@ def _insert_many_parallel(self, type_name, first, rows):
         chunk = first
         try:
             while chunk:
+                bad = []
                 split = split_float_vectors(self._java_db, type_name, chunk)
                 try:
-                    payload = json_bulk_dumps(split[0] if split else chunk)
+                    payload = json_bulk_dumps_tagged(split[0] if split else chunk)
                 except (TypeError, ValueError):
+                    # Only the rows the JSON text cannot carry unchanged leave the writers. The writers do not keep the
+                    # order of the rows, so the rows are split by kind, not by run.
+                    skip = declared_float_vectors(self._java_db, type_name)
+                    good = []
+                    for row in chunk:
+                        (good if bulk_row_ok(row, skip) else bad).append(row)
                     payload = None
-                if payload is None:
-                    # A chunk the JSON text cannot carry unchanged is written
-                    # synchronously, row by row, in its own transaction, after
-                    # the writers have drained, as the whole load used to be.
-                    self._java_db.async_().waitCompletion()
-                    was_active = self.is_transaction_active()
-                    try:
-                        if not was_active:
-                            self.begin()
-                        self._insert_many_chunk(type_name, chunk)
-                        if not was_active:
-                            self.commit()
-                    except BaseException:
-                        if not was_active:
-                            try:
-                                if self.is_transaction_active():
-                                    self.rollback()
-                            except Exception:  # nosec B110 - best-effort rollback
-                                pass
-                        raise
-                else:
+                    if good:
+                        split = split_float_vectors(self._java_db, type_name, good)
+                        try:
+                            payload = json_bulk_dumps_tagged(
+                                split[0] if split else good
+                            )
+                        except (
+                            TypeError,
+                            ValueError,
+                        ):  # a vector column the matrix did not take
+                            payload = None
+                            bad = chunk
+                if payload is not None:
                     try:
                         if split:
                             failures = batcher.insertManyJsonParallel(
@@ -905,6 +945,26 @@ def _insert_many_parallel(self, type_name, first, rows):
                     # assumed (ArcadeData/arcadedb#8478). Read after the final
                     # waitCompletion, when every writer is done.
                     chunk_failures.append(failures)
+                if bad:
+                    # Rows the JSON text cannot carry are written synchronously,
+                    # row by row, in their own transaction, after the writers
+                    # have drained.
+                    self._java_db.async_().waitCompletion()
+                    was_active = self.is_transaction_active()
+                    try:
+                        if not was_active:
+                            self.begin()
+                        self._insert_many_per_row(type_name, bad)
+                        if not was_active:
+                            self.commit()
+                    except BaseException:
+                        if not was_active:
+                            try:
+                                if self.is_transaction_active():
+                                    self.rollback()
+                            except Exception:  # nosec B110 - best-effort rollback
+                                pass
+                        raise
                 n += len(chunk)
                 chunk = list(itertools.islice(rows, _INSERT_MANY_CHUNK))
         except ArcadeDBError:

src/arcadedb/type_conversion.py

@@ -659,8 +659,8 @@ def json_bulk_scalar_ok(value: Any) -> bool:
 def _json_bulk_check(value: Any) -> None:
     """Raise ValueError for a value JSON would change on the way to the engine.
 
-    Walks nested lists and dicts. A type json.dumps cannot encode at all is
-    left to it (TypeError), which is what insert_many's fallback already handles.
+    Walks nested lists and dicts. A type json.dumps cannot encode at all raises
+    TypeError, which is what insert_many's fallback already handles.
     """
     if value is None or isinstance(value, bool):
         return
@@ -687,6 +687,10 @@ def _json_bulk_check(value: Any) -> None:
     elif isinstance(value, (list, tuple)):
         for item in value:
             _json_bulk_check(item)
+    else:
+        raise TypeError(
+            f"{type(value).__name__} cannot go through the JSON bulk path inside a list or map"
+        )
 
 
 def json_bulk_dumps(rows: Any) -> str:
@@ -724,6 +728,150 @@ def json_bulk_dumps(rows: Any) -> str:
     return json.dumps(rows, allow_nan=False)
 
 
+# ---------------------------------------------------------------------------
+# insert_many: values the plain JSON text cannot carry, as TAGGED values
+#
+# A date-time, date, Decimal, bytes, numpy scalar, pandas NA or NaT used to make json_bulk_dumps raise, and one such value sent the
+# whole 10,000-row chunk down the per-row path. Only DocumentBatcher.fill (insert_many, sync and parallel) decodes the tags, so only
+# insert_many uses json_bulk_dumps_tagged; the graph batches keep json_bulk_dumps. A tag is {"\u0001": [kind, payload]}: T LocalDateTime
+# (UTC wall clock), D LocalDate, n BigDecimal, b byte[] (base64): the objects convert_python_to_java builds for the per-row path.
+# Only a value that sits directly in a row is tagged; inside a list or map it still sends its row to the per-row path.
+
+_TAG = "\u0001"
+_TAGGED_KINDS = frozenset((datetime, date, Decimal, bytes, bytearray))
+_PANDAS_NULLS = frozenset(("NAType", "NaTType"))
+
+
+def _is_pandas_null(kind) -> bool:
+    return kind.__name__ in _PANDAS_NULLS and kind.__module__.startswith("pandas.")
+
+
+def _tag_datetime(value: datetime):
+    if value.utcoffset() is not None:
+        value = value.astimezone(timezone.utc).replace(tzinfo=None)
+    utc = value
+    if (
+        type(utc) is not datetime
+    ):  # pandas.Timestamp keeps nanoseconds in isoformat(); the per-row path keeps microseconds
+        utc = datetime(
+            utc.year,
+            utc.month,
+            utc.day,
+            utc.hour,
+            utc.minute,
+            utc.second,
+            utc.microsecond,
+        )
+    return {_TAG: ["T", utc.isoformat()]}
+
+
+def bulk_tag_default(value: Any) -> Any:
+    """json.dumps hook of json_bulk_dumps_tagged: a value that sits directly in a row, as a tagged value or a plain number or null.
+
+    Raises TypeError for a value with no tagged form and ValueError for one the bulk path would change (a NaN, an infinity, a
+    non-finite Decimal, an integer beyond 64 bits), so the caller sends that row down the per-row path, which stores it exactly or raises.
+    """
+    kind = type(value)
+    if _is_pandas_null(kind):
+        return None
+    if isinstance(value, datetime):
+        return _tag_datetime(value)
+    if isinstance(value, date):
+        return {_TAG: ["D", value.isoformat()]}
+    if isinstance(value, Decimal):
+        if not value.is_finite():
+            raise ValueError(
+                "a non-finite Decimal cannot go through the JSON bulk path"
+            )
+        return {_TAG: ["n", str(value)]}
+    if isinstance(value, (bytes, bytearray)):
+        import base64
+
+        return {_TAG: ["b", base64.b64encode(bytes(value)).decode("ascii")]}
+    if _np is not None and isinstance(value, _np.generic):
+        if isinstance(value, _np.bool_):
+            return bool(value)
+        if isinstance(value, _np.integer):
+            number = int(value)
+            if not _INT64_MIN <= number <= _INT64_MAX:
+                raise ValueError(
+                    f"an integer beyond 64 bits cannot go through the JSON bulk path: {number}"
+                )
+            return number
+        if kind in (_np.float64, _np.float32, _np.float16):
+            number = float(value)
+            if not math.isfinite(number):
+                raise ValueError(
+                    "NaN and Infinity cannot go through the JSON bulk path"
+                )
+            return number
+        if isinstance(value, _np.datetime64):
+            if _np.isnat(value):
+                return None
+            moment = value.astype("datetime64[us]").item()
+            if isinstance(moment, datetime):
+                return _tag_datetime(moment)
+            raise ValueError(
+                "a datetime64 outside the datetime range cannot go through the JSON bulk path"
+            )
+    raise TypeError(f"{kind.__name__} cannot go through the JSON bulk path")
+
+
+def json_bulk_dumps_tagged(rows: Any) -> str:
+    """json_bulk_dumps for insert_many: also lets a date-time, date, Decimal, bytes, numpy scalar, pandas NA or NaT that sits directly in a row
+    through, written by bulk_tag_default. Raises TypeError or ValueError exactly where json_bulk_dumps does for every other value
+    (a datetime inside a list or map included), so the caller can send that row down the per-row path.
+    """
+    int_min, int_max = _INT64_MIN, _INT64_MAX
+    tagged = _TAGGED_KINDS
+    for row in rows:
+        for key in row:
+            if type(key) is not str:
+                raise ValueError("a non-str key cannot go through the JSON bulk path")
+        for value in row.values():
+            kind = type(value)
+            if kind is str:
+                if not value.isascii() and not _str_survives_json(value):
+                    raise ValueError(
+                        "a string with a lone surrogate cannot go through the JSON bulk path"
+                    )
+            elif kind is int:
+                if not int_min <= value <= int_max:
+                    raise ValueError(
+                        f"an integer beyond 64 bits cannot go through the JSON bulk path: {value}"
+                    )
+            elif kind is float or kind is bool or value is None:
+                continue
+            elif kind in tagged:
+                continue  # bulk_tag_default checks it as json.dumps writes it
+            elif kind is dict and _TAG in value:
+                raise ValueError(
+                    "a map keyed by the tag marker cannot go through the JSON bulk path"
+                )
+            elif (
+                isinstance(value, (datetime, date, Decimal, bytes, bytearray))
+                or (_np is not None and isinstance(value, _np.generic))
+                or _is_pandas_null(kind)
+            ):
+                continue  # a subclass (pandas.Timestamp), a numpy scalar, pandas NA or NaT
+            else:
+                _json_bulk_check(
+                    value
+                )  # nested lists and dicts, and subclasses of the types above
+    return json.dumps(rows, allow_nan=False, default=bulk_tag_default)
+
+
+def bulk_row_ok(row: Any, skip=()) -> bool:
+    """Whether one insert_many row can go through json_bulk_dumps_tagged; `skip` names properties that cross some other way (float vectors)."""
+    if skip:
+        row = {k: v for k, v in row.items() if k not in skip}
+    try:
+        json_bulk_dumps_tagged((row,))
+    except (TypeError, ValueError):
+        return False
+    return True
+
+
 _FLOAT_VECTORS = None  # the bridge's FloatVectors class once resolved; False when the bridge jar lacks it