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

bindings/python/docs/api/database.md

@@ -521,12 +521,20 @@ faster still (2.24x on the same rows).
 
 - `type_name` (str): Target document type (must exist)
 - `rows` (iterable of dict): One dict per document, read lazily one chunk at
-  a time; values must be JSON-representable (str/int/float/bool/None, nested
-  lists/dicts). A chunk with other types (e.g. `datetime`, `bytes`) falls back
-  transparently to the per-row path, and so do values the JSON text would change on the way
-  to the engine: an integer beyond 64 bits, NaN or Infinity, a dict key that
-  is not a `str`, and a string with a lone surrogate. The per-row path stores
-  such a value exactly or raises, as `Document.set` does.
+  a time. A value that sits directly in a row is JSON-representable
+  (str/int/float/bool/None, nested lists/dicts), or one of these, which cross
+  inside the JSON as tagged values and are stored as `INSERT ... CONTENT` stores
+  them: `datetime` (naive: the wall clock as it shows; aware: converted to UTC;
+  `pandas.Timestamp` too), `date`, `Decimal`, `bytes`/`bytearray`, a numpy
+  scalar (`np.int64`, `np.float32`, `np.bool_`, ...), `np.datetime64` (as a naive
+  `datetime`; `NaT` is null) and `pandas.NA`/`pandas.NaT` (null; the statement
+  refuses these). A chunk with such a value takes the JSON entry, not the columnar
+  one. A row with any other value 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
+  `datetime` or `Decimal` inside a list or dict, an integer beyond 64 bits, NaN or
+  Infinity, a dict key that is not a `str`, a string with a lone surrogate, a
+  `datetime.time`. The per-row path stores such a value exactly or raises, as
+  `Document.set` does.
 - `commit_every` (int): Transaction batch size in synchronous mode
   (0 = single transaction; ignored when a transaction is already open).
 - `parallel` (bool): Route rows through the async executor's parallel

bindings/python/docs/development/testing.md

@@ -81,6 +81,7 @@ asserts, so the list below stops at the file.
 | `test_bulk_insert.py` | `Database.insert_many`, `AsyncExecutor.create_record`, vector columns, numpy `append_samples`, and what a failed call leaves behind |
 | `test_insert_columns.py` | `Database.insert_columns`: typed values, nulls, the parallel mode, the rollback contract, and bad input refused before anything is written |
 | `test_bulk_json_paths.py` | Values the JSON text of a bulk call cannot carry unchanged keep the call off the bulk path, so every entry point stores the same value or raises |
+| `test_native_mixed_chunk.py` | Native: `insert_many` keeps a chunk on the JSON bulk entry for date-times, dates, Decimals, bytes, numpy scalars, pandas NA and NaT, and None; the stored values equal `INSERT ... CONTENT`'s; only a row with an uncarriable value goes to the statement |
 | `test_graph_batch.py` | `GraphBatch` buffering, flushing, failure handling, and the caller's transaction |
 | `test_graph.py` | `GraphBatch.new_edges` and `create_vertices` bulk paths |
 | `test_import_database.py` | SQL `IMPORT DATABASE` across CSV (documents, graphs, and time series), XML, Neo4j, Word2Vec, and RDF |

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

@@ -11,7 +11,7 @@
 
 from ..exceptions import ArcadeDBError
 from ..transactions import TransactionContext
-from ..type_conversion import json_bulk_dumps
+from ..type_conversion import bulk_row_ok, json_bulk_dumps_tagged
 from . import lib, not_yet
 from .bulk import insert_rows_columnar
 from .codec import _default as codec_default
@@ -664,8 +664,13 @@ def insert_many(
         """Bulk insert. Rows go in chunks of 10,000, one crossing per chunk. A chunk of rows with the same keys whose columns each hold one
         value type (ints, floats, bools, strings, equal-length float lists for an ``ARRAY_OF_FLOATS`` property; None is a null) goes as
         columns by address (``arcadedb_n_insert_colp``, no JSON text); any other chunk goes as one JSON text (``arcadedb_n_insert_rows``:
-        a newDocument/set/save loop inside the image, as ``DocumentBatcher.insertManyJson`` on the JPype path); a chunk with a value the
-        JSON text cannot carry unchanged goes through ``INSERT INTO t CONTENT :row`` for each row, in one crossing. Inside the caller's
+        a newDocument/set/save loop inside the image, as ``DocumentBatcher.insertManyJson`` on the JPype path). A ``datetime`` (also
+        ``pandas.Timestamp`` and ``numpy.datetime64``), ``date``, ``Decimal``, ``bytes``/``bytearray``, numpy scalar, ``pandas.NA`` or
+        ``pandas.NaT`` sitting directly in a row crosses in that text as a tagged value (``{"\\u0001": [kind, payload]}``, decoded by ``Json.java``),
+        so it does not move the chunk off this path; such a chunk loses the columnar entry. A row with a value the JSON text cannot carry
+        unchanged (NaN or Infinity, an integer beyond 64 bits, a date-time or decimal inside a list or dict, a non-str key, a lone
+        surrogate, a ``datetime.time``) goes through ``INSERT INTO t CONTENT :row`` on its own, in one crossing for a run of such rows; the
+        rows around it keep the bulk path and every row is inserted in the order given. Inside the caller's
         transaction nothing is committed; otherwise every ``commit_every`` rows are one transaction (0: all rows in one).
         """
         self._check_not_closed()
@@ -704,31 +709,59 @@ def insert_many(
             raise
 
     def _insert_chunk(self, type_name: str, sql: str, chunk) -> None:
+        """One chunk of ``insert_many`` inside the open transaction; see :meth:`insert_many` for the paths.
+
+        A row with a value the JSON 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) goes through ``INSERT INTO t CONTENT :row`` on its own; the rows around it still go in bulk, and every row
+        is inserted in the order given.
+        """
         if lib.INSERT_ROWS:
             if insert_rows_columnar(self, type_name, chunk):
                 return
             try:
-                payload = json_bulk_dumps(chunk)
+                payload = json_bulk_dumps_tagged(chunk)
             except (TypeError, ValueError):
-                payload = None
-            if payload is not None:
-                reply = json.loads(
-                    lib.call(
-                        "arcadedb_n_insert_rows",
-                        self._h,
-                        lib.b(type_name),
-                        lib.b(payload),
-                    )
-                )
-                if not reply.get("ok"):
-                    raise _error(f"Failed to bulk-insert into '{type_name}'", reply)
+                self._insert_chunk_runs(type_name, sql, chunk)
                 return
+            self._insert_rows_json(type_name, payload)
+            return
+        self._insert_chunk_per_row(sql, chunk)
+
+    def _insert_rows_json(self, type_name: str, payload: str) -> None:
+        reply = json.loads(
+            lib.call(
+                "arcadedb_n_insert_rows",
+                self._h,
+                lib.b(type_name),
+                lib.b(payload),
+            )
+        )
+        if not reply.get("ok"):
+            raise _error(f"Failed to bulk-insert into '{type_name}'", reply)
+
+    def _insert_chunk_runs(self, type_name: str, sql: str, chunk) -> None:
+        """A chunk with a row the JSON text cannot carry: runs of carriable rows go in bulk, the other rows one statement each, in order."""
+        ok = [bulk_row_ok(row) for row in chunk]
+        n = len(chunk)
+        start = 0
+        while start < n:
+            end = start
+            while end < n and ok[end] == ok[start]:
+                end += 1
+            run = chunk[start:end]
+            carriable = ok[start]
+            start = end
+            if not carriable:
+                self._insert_chunk_per_row(sql, run)
+            elif not insert_rows_columnar(self, type_name, run):
+                self._insert_rows_json(type_name, json_bulk_dumps_tagged(run))
+
+    def _insert_chunk_per_row(self, sql: str, rows) -> None:
         if lib.EXEC_MANY:
-            # values JSON text cannot carry unchanged (a date-time, a decimal, NaN, a big integer): the same statement per
-            # row, but all rows of the chunk in one crossing (arcadedb_n_exec_many), not one call per row
-            self._command_many("sql", sql, [{"row": row} for row in chunk])
+            # the same statement per row, but all rows in one crossing (arcadedb_n_exec_many), not one call per row
+            self._command_many("sql", sql, [{"row": row} for row in rows])
             return
-        for row in chunk:
+        for row in rows:
             self.command("sql", sql, {"row": row}).close()
 
     def _command_many(self, language: str, command: str, rows) -> int:

bindings/python/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
@@ -672,6 +677,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:
@@ -707,3 +716,135 @@ 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)
+
+
+# ---------------------------------------------------------------------------
+# 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):
+    """The tagged form of a datetime, written as codec._default writes it (a pandas.Timestamp keeps its nanoseconds)."""
+    if value.utcoffset() is not None:
+        value = value.astimezone(timezone.utc).replace(tzinfo=None)
+    return {_TAG: ["T", value.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

bindings/python/tests/conftest.py

@@ -29,6 +29,8 @@
     collect_ignore.append("test_native_udf.py")
     # the columnar insert's stored records are compared between the JSON and the columnar entry points of the native library
     collect_ignore.append("test_native_columnar_insert.py")
+    # the rows of an insert_many chunk with date-times, Decimals, bytes and numpy scalars are told apart by the entry points they cross
+    collect_ignore.append("test_native_mixed_chunk.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")
 if not os.path.isdir(os.path.join(os.path.dirname(os.path.dirname(__file__)), "docs")):

bindings/python/tests/test_native_columnar_insert.py

@@ -298,12 +298,24 @@ def test_rows_with_different_keys_fall_back_to_json(temp_db, spy, monkeypatch):
     _fallback_case(temp_db, spy, monkeypatch, rows, ROWS)
 
 
-def test_values_json_cannot_carry_still_go_row_by_row_in_one_crossing(
+def test_a_datetime_leaves_the_columnar_entry_but_stays_on_the_json_entry(
     temp_db, spy, monkeypatch
 ):
+    """columnar.pack refuses a date-time, so the chunk takes the JSON entry, which carries it as a tagged value
+    (test_native_mixed_chunk.py compares what it stores with INSERT ... CONTENT)."""
     when = datetime.datetime(2024, 5, 6, 7, 8, 9)
     rows = [{"id": i, "txt": "t%d" % i, "when": when} for i in range(10)]
+    _fallback_case(temp_db, spy, monkeypatch, rows, ROWS)
+    assert spy.count(MANY) == 0
+
+
+def test_values_json_cannot_carry_still_go_row_by_row_in_one_crossing(
+    temp_db, spy, monkeypatch
+):
+    when = datetime.datetime(2024, 5, 6, 7, 8, 9)
+    rows = [{"id": i, "txt": "t%d" % i, "whens": [when]} for i in range(10)]
     _fallback_case(temp_db, spy, monkeypatch, rows, MANY)
+    assert spy.count(ROWS) == 0
 
 
 def test_an_integer_past_64_bits_is_refused_on_both_paths(temp_db, spy):