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

docs/modules/ROOT/pages/how-to/bulk-loads.adoc

@@ -53,6 +53,10 @@ Rows that are already committed stay committed when a later chunk fails. Only th
 
 Because the two packages have different defaults, the same program can keep different rows after the same failure: each keeps its rows up to its own last commit. If the outcome of a failed load must not depend on the package, pass the same `commit_every` on both, or `commit_every=0`.
 
+== Values the bulk path carries
+
+On both packages, `insert_many` keeps rows with `datetime`, `date`, `Decimal`, `bytes`, numpy scalar, or pandas `NA` and `NaT` values on the bulk path. On the `arcadedb` (native) package, a row with a value the bulk text cannot carry (NaN or infinity, an integer beyond 64 bits, a date-time or `Decimal` inside a list or dict, or a `datetime.time`) goes in with its own statement, in the order given.
+
 == Change the commit size
 
 [source,python]

src/arcadedb/_native/database.py

@@ -12,7 +12,7 @@
 from .._params import refuse_surplus_positionals
 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 interrupts, lib, needs_jvm
 from .bulk import insert_rows_columnar
 from .codec import _prep as _prep_row
@@ -754,8 +754,13 @@ def insert_many(
         """Bulk insert. Rows go in chunks of at most 10,000 (a chunk never crosses a commit), 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 map, 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). The default is 1000 on
         this build (the JVM build's is 10,000). Rows that are already committed stay committed if a later chunk fails, so a smaller
         ``commit_every`` keeps fewer rows of a failed load (only the transaction that was open is rolled back).
@@ -796,31 +801,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:

tests/conftest.py

@@ -43,6 +43,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 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 native library's start-up output (#477) can only be checked with the native library loaded

tests/test_insert_many_mixed_values.py

@@ -21,9 +21,8 @@
 # The call counts below read Database.new_document, the per-row path of the JVM wheel (one JPype crossing per row). The native build has its
 # own per-row path (INSERT ... CONTENT) that never calls it, so only the stored values are compared there.
 JVM = arcadedb.__backend__ == "jvm"
-# pandas NA/NaT and numpy datetime64 are tagged by the JVM wheel's bulk path. The native build refuses them as parameters (TypeError;
-# BACKLOG-bindings.md N-04/J-05, native column): the two tests below assert that refusal there, which turns red the day native accepts them.
-# An xfail would count as an uncounted skip in the native parity gate (native/wheel/parity_wheel.py), so the refusal is asserted instead.
+# pandas NA/NaT and numpy datetime64 are tagged by the bulk path of both builds (the native build since ArcadeData/arcadedb-python#218; before that it
+# refused them as parameters with a TypeError, and these two tests asserted that refusal so that they would turn red the day native accepted them).
 
 UTC9 = dt.timezone(dt.timedelta(hours=9))
 
@@ -175,10 +174,6 @@ def test_pandas_na_and_nat_are_nulls_on_the_bulk_path(temp_db_path, new_document
     ]
     with arcadedb.create_database(temp_db_path) as db:
         _create(db, "Bulk", False)
-        if not JVM:
-            with pytest.raises(TypeError):
-                db.insert_many("Bulk", rows)
-            return
         db.insert_many("Bulk", rows)
         assert new_documents == []
         got = db.query("sql", "SELECT FROM Bulk ORDER BY id").to_list()
@@ -196,10 +191,6 @@ def test_a_numpy_datetime64_is_the_datetime_the_per_row_path_would_store(
     rows = [{"id": 0, "ts": np.datetime64("2026-05-06T07:08:09.123456")}]
     with arcadedb.create_database(temp_db_path) as db:
         _create(db, "Bulk", False)
-        if not JVM:
-            with pytest.raises(TypeError):
-                db.insert_many("Bulk", rows)
-            return
         db.insert_many("Bulk", rows)
         assert new_documents == []
         got = db.query("sql", "SELECT FROM Bulk").first()

tests/test_native_columnar_insert.py

@@ -315,12 +315,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):