From 374346fb44c3a7e66dabbd628966633cbd5cc707 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sun, 25 Oct 2026 17:36:43 +0900 Subject: [PATCH] Native insert_many: a date-time, Decimal, bytes, numpy scalar or pandas NA stays on the JSON bulk entry; an uncarriable value moves its own row, not the chunk Json.java already decodes the tagged values (T, D, n, b), so a datetime, date, Decimal, bytes, numpy scalar, pandas NA or NaT sitting directly in a row now crosses in the arcadedb_n_insert_rows text instead of sending the whole 10,000-row chunk through INSERT INTO t CONTENT :row for each row. A value the text still cannot carry sends only its own run of rows to the statement; the rows around it keep the bulk entry and the insert order. Python only, no native-image change. Co-Authored-By: Claude Sonnet 5.5 --- bindings/python/docs/api/database.md | 20 +- bindings/python/docs/development/testing.md | 1 + .../src/arcadedb_embedded/_native/database.py | 73 ++-- .../src/arcadedb_embedded/type_conversion.py | 141 ++++++++ bindings/python/tests/conftest.py | 2 + .../tests/test_native_columnar_insert.py | 14 +- .../python/tests/test_native_mixed_chunk.py | 312 ++++++++++++++++++ 7 files changed, 536 insertions(+), 27 deletions(-) create mode 100644 bindings/python/tests/test_native_mixed_chunk.py diff --git a/bindings/python/docs/api/database.md b/bindings/python/docs/api/database.md index 683a0036b94..91b658ec20b 100644 --- a/bindings/python/docs/api/database.md +++ b/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 diff --git a/bindings/python/docs/development/testing.md b/bindings/python/docs/development/testing.md index 8687ae7570c..bc7bebbeb5b 100644 --- a/bindings/python/docs/development/testing.md +++ b/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 | diff --git a/bindings/python/src/arcadedb_embedded/_native/database.py b/bindings/python/src/arcadedb_embedded/_native/database.py index 1461b97842e..eb1be9f849b 100644 --- a/bindings/python/src/arcadedb_embedded/_native/database.py +++ b/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: diff --git a/bindings/python/src/arcadedb_embedded/type_conversion.py b/bindings/python/src/arcadedb_embedded/type_conversion.py index 94c9bae550c..8f046e3459d 100644 --- a/bindings/python/src/arcadedb_embedded/type_conversion.py +++ b/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 diff --git a/bindings/python/tests/conftest.py b/bindings/python/tests/conftest.py index 180a3771d86..3d0e41e7ca1 100644 --- a/bindings/python/tests/conftest.py +++ b/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")): diff --git a/bindings/python/tests/test_native_columnar_insert.py b/bindings/python/tests/test_native_columnar_insert.py index f4ce830e06c..5845ed5cc82 100644 --- a/bindings/python/tests/test_native_columnar_insert.py +++ b/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): diff --git a/bindings/python/tests/test_native_mixed_chunk.py b/bindings/python/tests/test_native_mixed_chunk.py new file mode 100644 index 00000000000..e5b511274be --- /dev/null +++ b/bindings/python/tests/test_native_mixed_chunk.py @@ -0,0 +1,312 @@ +"""Native backend: ``insert_many`` keeps a chunk on the JSON bulk entry (``arcadedb_n_insert_rows``) when its rows hold a date-time, date, +Decimal, bytes, a numpy scalar, a pandas NA or NaT, or a null. + +Before, one such value made ``json.dumps`` raise and the whole 10,000-row chunk went through ``INSERT INTO t CONTENT :row`` for each row +(``arcadedb_n_exec_many``). ``Json.java`` already decodes the tagged values the Python side writes, so the rows now cross in the text as +tagged values. Each test stores the same rows through that statement (``_command_many``, what the chunk used to take) and through +``insert_many`` and compares the stored values and their types; a spy on ``lib.call`` proves which entry ran. A row with a value the text +still cannot carry (NaN, an integer beyond 64 bits, a date-time inside a list or map, a lone surrogate, a ``time``) goes through the +statement on its own; the rows around it keep the bulk entry and the insert order. + +Needs the native library, so conftest.py collects this file on the native backend only (a skip elsewhere would fail +scripts/check_test_skips.py). +""" + +import datetime as dt +import decimal + +import arcadedb_embedded as arcadedb +import pytest +from arcadedb_embedded._native import lib + +np = pytest.importorskip("numpy") + +ROWS, MANY, COLP = ( + "arcadedb_n_insert_rows", + "arcadedb_n_exec_many", + "arcadedb_n_insert_colp", +) +UTC9 = dt.timezone(dt.timedelta(hours=9)) + +# name -> value; the statement path (INSERT ... CONTENT :row) stores every one of these, so the bulk entry has a stored value to match +VALUES = { + "naive_dt": dt.datetime(2026, 5, 6, 7, 8, 9, 123456), + "aware_dt": dt.datetime(2026, 5, 6, 7, 8, 9, 654321, tzinfo=UTC9), + "dt_whole_second": dt.datetime(2026, 5, 6, 7, 8, 9), + "date": dt.date(2026, 5, 6), + "decimal": decimal.Decimal("12345.6789"), + "decimal_exp": decimal.Decimal("1E+3"), + "decimal_neg": decimal.Decimal("-0.50"), + "bytes": b"\x00\xff abc\x7f", + "bytearray": bytearray(b"ab\x80"), + "np_int64": np.int64(2**40 + 5), + "np_int8": np.int8(-3), + "np_uint32": np.uint32(4_000_000_000), + "np_float64": np.float64(0.1), + "np_float32": np.float32(0.1), + "np_bool": np.bool_(True), + "none": None, + "plain_int": 7, +} + +DECLARED = { + "naive_dt": "DATETIME", + "aware_dt": "DATETIME", + "dt_whole_second": "DATETIME", + "date": "DATE", + "decimal": "DECIMAL", + "decimal_exp": "DECIMAL", + "decimal_neg": "DECIMAL", + "bytes": "BINARY", + "np_int64": "LONG", + "np_int8": "INTEGER", + "np_uint32": "LONG", + "np_float64": "DOUBLE", + "np_float32": "FLOAT", + "np_bool": "BOOLEAN", +} + + +class Spy: + def __init__(self, monkeypatch): + self.calls = [] + real = lib.call + + def call(name, *args): + self.calls.append(name) + return real(name, *args) + + monkeypatch.setattr(lib, "call", call) + + def count(self, name): + return self.calls.count(name) + + def clear(self): + self.calls.clear() + + +@pytest.fixture +def spy(monkeypatch): + return Spy(monkeypatch) + + +def _create(db, name, declared): + db.command("sql", f"CREATE DOCUMENT TYPE {name}") + db.command("sql", f"CREATE PROPERTY {name}.id INTEGER") + if declared: + for prop, ty in DECLARED.items(): + db.command("sql", f"CREATE PROPERTY {name}.{prop} {ty}") + + +def _statement_path(db, type_name, rows): + """The reference: INSERT INTO t CONTENT :row for each row, as insert_many fell back to it for a chunk with such a value.""" + sql = f"INSERT INTO `{type_name}` CONTENT :row" # nosec B608 - test-owned type name + db.begin() + db._command_many("sql", sql, [{"row": row} for row in rows]) + db.commit() + + +def _stored(db, type_name): + """Every record by id as {property: (python type name, repr)}, so a changed type is as visible as a changed value.""" + out = [] + for rec in db.query( + "sql", f"SELECT FROM {type_name} ORDER BY id" + ).to_list(): # nosec B608 - test-owned type name + out.append( + { + k: (type(v).__name__, repr(v)) + for k, v in sorted(rec.items()) + if not k.startswith("@") and k != "rid" + } + ) + return out + + +def _rows(names): + return [{"id": i, name: VALUES[name]} for i, name in enumerate(names)] + + +@pytest.mark.parametrize("declared", [True, False], ids=["declared", "schemaless"]) +def test_every_value_kind_is_stored_as_the_statement_path_stores_it( + temp_db_path, spy, declared +): + rows = _rows(list(VALUES)) + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Ref", declared) + _create(db, "Bulk", declared) + _statement_path(db, "Ref", rows) + spy.clear() + db.insert_many("Bulk", rows) + assert spy.count(MANY) == 0, "the chunk left the bulk entry" + assert spy.count(ROWS) == 1 + assert _stored(db, "Bulk") == _stored(db, "Ref") + assert len(_stored(db, "Bulk")) == len(VALUES) + + +@pytest.mark.parametrize("declared", [True, False], ids=["declared", "schemaless"]) +def test_one_row_with_every_kind_at_once(temp_db_path, spy, declared): + row = {"id": 0, **VALUES} + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Ref", declared) + _create(db, "Bulk", declared) + _statement_path(db, "Ref", [row]) + spy.clear() + db.insert_many("Bulk", [row]) + assert spy.count(MANY) == 0 and spy.count(ROWS) == 1 + assert _stored(db, "Bulk") == _stored(db, "Ref") + + +def test_a_pandas_timestamp_is_stored_as_the_statement_path_stores_it( + temp_db_path, spy +): + pd = pytest.importorskip("pandas") + rows = [ + {"id": 0, "ts": pd.Timestamp("2026-05-06 07:08:09.123456789")}, + {"id": 1, "ts": pd.Timestamp("2026-05-06 07:08:09.123456", tz="Asia/Seoul")}, + ] + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Ref", False) + _create(db, "Bulk", False) + _statement_path(db, "Ref", rows) + spy.clear() + db.insert_many("Bulk", rows) + assert spy.count(MANY) == 0 and spy.count(ROWS) == 1 + assert _stored(db, "Bulk") == _stored(db, "Ref") + + +def test_pandas_na_and_nat_are_nulls_on_the_bulk_entry(temp_db_path, spy): + """The statement path refuses pd.NA and pd.NaT; insert_dataframe stores them as nulls, and so does the bulk entry.""" + pd = pytest.importorskip("pandas") + rows = [ + {"id": 0, "a": pd.NA, "b": pd.NaT, "c": 1}, + {"id": 1, "a": 2, "b": pd.Timestamp("2026-05-06"), "c": pd.NA}, + {"id": 2, "a": np.datetime64("NaT", "us"), "b": None, "c": 3}, + ] + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Bulk", False) + db.insert_many("Bulk", rows) + assert spy.count(MANY) == 0 + got = db.query("sql", "SELECT FROM Bulk ORDER BY id").to_list() + assert [(r.get("a"), r.get("b"), r.get("c")) for r in got] == [ + (None, None, 1), + (2, dt.datetime(2026, 5, 6), None), + (None, None, 3), + ] + + +def test_a_numpy_datetime64_is_a_naive_datetime(temp_db_path, spy): + 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) + db.insert_many("Bulk", rows) + assert spy.count(MANY) == 0 + assert db.query("sql", "SELECT FROM Bulk").first().get("ts") == dt.datetime( + 2026, 5, 6, 7, 8, 9, 123456 + ) + + +def test_none_in_every_column_stays_on_the_bulk_entry(temp_db_path, spy): + rows = [ + {"id": i, "a": None, "b": None if i % 2 else 1.5, "c": None} for i in range(50) + ] + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Bulk", False) + db.insert_many("Bulk", rows) + assert spy.count(MANY) == 0 + assert db.count_type("Bulk") == 50 + + +def test_a_big_chunk_with_one_datetime_is_one_bulk_call(temp_db_path, spy): + n = 10_000 + rows = [{"id": i, "x": i * 2} for i in range(n)] + rows[5_000]["ts"] = dt.datetime(2026, 1, 1) + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Bulk", False) + assert db.insert_many("Bulk", rows) == n + assert spy.count(MANY) == 0 and spy.count(ROWS) == 1 + assert db.count_type("Bulk") == n + + +UNCARRIABLE = { + "nan": float("nan"), + "nested_dt": [dt.datetime(2026, 1, 1)], + "dict_dt": {"k": dt.datetime(2026, 1, 1)}, + "decimal_in_list": [decimal.Decimal("1.5")], +} + + +@pytest.mark.parametrize("name", list(UNCARRIABLE)) +def test_an_uncarriable_value_sends_its_own_row_to_the_statement_not_the_chunk( + temp_db_path, spy, name +): + n = 1_000 + rows = [ + {"id": i, "x": i, "ts": dt.datetime(2026, 1, 1, 0, 0, i % 60)} for i in range(n) + ] + rows[500]["v"] = UNCARRIABLE[name] + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Ref", False) + _create(db, "Bulk", False) + _statement_path(db, "Ref", rows) + spy.clear() + db.insert_many("Bulk", rows) + assert spy.count(MANY) == 1, "one statement run for the one uncarriable row" + assert ( + spy.count(ROWS) == 2 + ), "the rows before and after it stay on the bulk entry" + assert _stored(db, "Bulk") == _stored(db, "Ref") + + +def test_rows_keep_their_insert_order_around_a_statement_run(temp_db_path, spy): + n = 300 + rows = [{"id": i, "ts": dt.datetime(2026, 1, 1)} for i in range(n)] + for i in (3, 4, 5, 150, 299): + rows[i]["v"] = float("nan") + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Bulk", False) + db.insert_many("Bulk", rows) + assert spy.count(MANY) == 3 # runs 3-5, 150, 299 + ids = [r.get("id") for r in db.query("sql", "SELECT id FROM Bulk").to_list()] + assert ids == list(range(n)) + + +def test_commit_every_still_commits_in_the_middle_of_mixed_chunks(temp_db_path): + rows = [ + {"id": i, "ts": dt.datetime(2026, 1, 1), "d": decimal.Decimal(i)} + for i in range(2_500) + ] + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Bulk", False) + assert db.insert_many("Bulk", rows, commit_every=1_000) == 2_500 + assert db.count_type("Bulk") == 2_500 + assert not db.is_transaction_active() + + +def test_a_failing_row_rolls_the_open_batch_back(temp_db_path): + rows = [{"id": i, "ts": dt.datetime(2026, 1, 1)} for i in range(20)] + rows[10]["ts"] = object() # no stored form: both paths raise + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Bulk", False) + with pytest.raises(Exception): + db.insert_many("Bulk", rows) + assert db.count_type("Bulk") == 0 + assert not db.is_transaction_active() + + +def test_a_user_dict_that_looks_like_a_tag_is_stored_as_before(temp_db_path): + """Json.java decodes any one-key map keyed by the tag marker, on the statement path and on the bulk entry alike (a quirk of the native + parser, not changed here): the row goes to the statement, as before, and is stored as that path stores it. + """ + rows = [ + { + "id": 0, + "m": {"\u0001": ["T", "2026-01-01T00:00:00"]}, + "ts": dt.datetime(2026, 1, 1), + } + ] + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Ref", False) + _create(db, "Bulk", False) + _statement_path(db, "Ref", rows) + db.insert_many("Bulk", rows) + assert _stored(db, "Bulk") == _stored(db, "Ref")