diff --git a/src/arcadedb/core.py b/src/arcadedb/core.py index 342fd22..c233326 100644 --- a/src/arcadedb/core.py +++ b/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: diff --git a/src/arcadedb/type_conversion.py b/src/arcadedb/type_conversion.py index b552319..d18f1a6 100644 --- a/src/arcadedb/type_conversion.py +++ b/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 diff --git a/src/java/com/arcadedb/python/DocumentBatcher.java b/src/java/com/arcadedb/python/DocumentBatcher.java index fa36a70..a3598e6 100644 --- a/src/java/com/arcadedb/python/DocumentBatcher.java +++ b/src/java/com/arcadedb/python/DocumentBatcher.java @@ -15,9 +15,12 @@ * the JSON or as a column. The boxDoubles/boxLongs * helpers below serve AsyncExecutor.append_samples' numpy fast path. * - * JSON-representable property values only (str/int/float/bool/null and - * nested lists/maps thereof) — the Python side falls back to the per-row - * new_document path for anything else (e.g. datetime, bytes). + * JSON-representable property values (str/int/float/bool/null and nested + * lists/maps thereof), plus a date-time, date, Decimal or byte[] sitting + * directly in a row, which cross as tagged values (see untag). The Python side + * sends a row with any other value (NaN, an integer beyond 64 bits, a date-time + * inside a list or map) down the per-row path (new_document, or new_vertex for + * a vertex type), that row only. * * Every entry point loads a document type or a vertex type: a vertex type gets * real vertices (db.newVertex), as INSERT INTO does on the SQL path and on the @@ -287,12 +290,35 @@ public static Object[] boxLongs(final long[] a) { return out; } + /** + * A tagged value from the Python side, {"\u0001": [kind, payload]}: T LocalDateTime (UTC wall clock), D LocalDate, n BigDecimal, + * b byte[] (base64): the objects the per-row path (type_conversion.convert_python_to_java) builds. Anything else is returned unchanged. + */ + private static Object untag(final Object v) { + if (v instanceof Map m && m.size() == 1 && m.get("\u0001") instanceof java.util.List t && t.size() == 2 + && t.get(0) instanceof String kind && t.get(1) instanceof String s) { + switch (kind) { + case "T": + return java.time.LocalDateTime.parse(s); + case "D": + return java.time.LocalDate.parse(s); + case "n": + return new java.math.BigDecimal(s); + case "b": + return java.util.Base64.getDecoder().decode(s); + default: + throw new IllegalArgumentException("unknown value tag " + kind); + } + } + return v; + } + private static void fill(final MutableDocument doc, final JSONObject row, final int i, final String[] vectorNames, final Object[] vectors) { // toMap() turns nested objects and arrays into plain Map and List. Storing the parsed JSONArray itself reads back // as a JSONArray inside the transaction that wrote it, and only becomes a list once the record is serialized. for (final Map.Entry entry : row.toMap().entrySet()) - doc.set(entry.getKey(), entry.getValue()); + doc.set(entry.getKey(), untag(entry.getValue())); if (vectorNames != null) for (int c = 0; c < vectorNames.length; c++) doc.set(vectorNames[c], ((FloatVectors.Matrix) vectors[c]).row(i)); diff --git a/tests/test_insert_many_mixed_values.py b/tests/test_insert_many_mixed_values.py new file mode 100644 index 0000000..f0a2eef --- /dev/null +++ b/tests/test_insert_many_mixed_values.py @@ -0,0 +1,372 @@ +"""insert_many keeps a chunk on the bulk JSON path when its rows hold a date-time, date, Decimal, bytes, a numpy scalar, or a null. + +Those values used to make ``json.dumps`` raise, and one such value sent the whole 10,000-row chunk down the per-row path +(``new_document`` + ``set`` per property + ``save``). They now cross inside the JSON text as tagged values that +``DocumentBatcher.fill`` turns back into the objects the per-row path builds. Each test stores the same rows through the per-row path and +through ``insert_many`` and compares the stored values and their types, and counts ``new_document`` calls to prove which path ran. +A value the bulk path still cannot carry (NaN, an integer beyond 64 bits, a datetime inside a list, a lone surrogate, a ``time``) sends only +its own row down the per-row path, not the 10,000 rows around it. +""" + +import datetime as dt +import decimal + +import pytest + +import arcadedb as arcadedb +from arcadedb.core import Database + +np = pytest.importorskip("numpy") + +# 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. + +UTC9 = dt.timezone(dt.timedelta(hours=9)) + +# name -> value; every one of these is stored by the per-row path (Document.set), so the bulk path 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, +} + +# a declared property per value, so the engine's conversion to the declared type is compared too +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", +} + + +@pytest.fixture +def new_documents(monkeypatch): + """Count the per-row path: it is the only caller of Database.new_document while insert_many runs.""" + calls = [] + real = Database.new_document + + def counting(self, *args, **kwargs): + calls.append(args[0] if args else kwargs.get("type_name")) + return real(self, *args, **kwargs) + + monkeypatch.setattr(Database, "new_document", counting) + return calls + + +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 _per_row(db, type_name, rows): + """The reference: the per-row path, exactly as insert_many used to fall back to it.""" + db.begin() + for row in rows: + doc = db.new_document(type_name) + for k, v in row.items(): + doc.set(k, v) + doc.save() + 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 = [] + sql = f"SELECT FROM {type_name} ORDER BY id" # nosec B608 - test-owned type name + for rec in db.query("sql", sql).to_list(): + 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_per_row_path_stores_it( + temp_db_path, new_documents, declared +): + rows = _rows(list(VALUES)) + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Ref", declared) + _create(db, "Bulk", declared) + _per_row(db, "Ref", rows) + new_documents.clear() + db.insert_many("Bulk", rows) + assert new_documents == [], "the chunk left the bulk path" + 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, new_documents, declared): + row = {"id": 0, **{k: v for k, v in VALUES.items()}} + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Ref", declared) + _create(db, "Bulk", declared) + _per_row(db, "Ref", [row]) + new_documents.clear() + db.insert_many("Bulk", [row]) + assert new_documents == [] + assert _stored(db, "Bulk") == _stored(db, "Ref") + + +def test_a_pandas_timestamp_is_stored_as_the_per_row_path_stores_it( + temp_db_path, new_documents +): + pd = pytest.importorskip("pandas") + naive = pd.Timestamp("2026-05-06 07:08:09.123456789") + aware = pd.Timestamp("2026-05-06 07:08:09.123456", tz="Asia/Seoul") + rows = [{"id": 0, "ts": naive}, {"id": 1, "ts": aware}] + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Ref", False) + _create(db, "Bulk", False) + _per_row(db, "Ref", rows) + new_documents.clear() + db.insert_many("Bulk", rows) + assert new_documents == [] + assert _stored(db, "Bulk") == _stored(db, "Ref") + + +def test_pandas_na_and_nat_are_nulls_on_the_bulk_path(temp_db_path, new_documents): + """The per-row path refuses pd.NA and pd.NaT (TypeError / ValueError); insert_dataframe stores them as nulls, and so does the bulk path.""" + 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) + 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() + 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_the_datetime_the_per_row_path_would_store( + temp_db_path, new_documents +): + """The per-row path raises TypeError for np.datetime64; the bulk path stores the UTC wall clock, like a naive datetime.""" + 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() + assert got.get("ts") == dt.datetime(2026, 5, 6, 7, 8, 9, 123456) + + +def test_none_in_every_column_stays_on_the_bulk_path(temp_db_path, new_documents): + 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 new_documents == [] + assert db.count_type("Bulk") == 50 + + +def test_a_big_chunk_with_one_datetime_is_one_bulk_call_not_10000_rows( + temp_db_path, new_documents +): + 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 new_documents == [] + assert db.count_type("Bulk") == n + + +# values the bulk path still cannot carry: only the row that holds one goes per row +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")], + "surrogate": "a\ud800b", +} + + +@pytest.mark.parametrize( + "name", ["nan", "nested_dt", "dict_dt", "decimal_in_list", "surrogate"] +) +def test_an_uncarriable_value_sends_its_own_row_per_row_not_the_chunk( + temp_db_path, new_documents, 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) + try: + _per_row(db, "Ref", rows) + except ( + Exception + ): # nosec B110 - a lone surrogate may be refused; the bulk path must refuse it the same way + db.rollback() if db.is_transaction_active() else None + new_documents.clear() + with pytest.raises(Exception): + db.insert_many("Bulk", rows) + return + new_documents.clear() + db.insert_many("Bulk", rows) + if JVM: + assert new_documents == [ + "Bulk" + ], "only the row with the uncarriable value goes per row" + assert _stored(db, "Bulk") == _stored(db, "Ref") + + +def test_rows_keep_their_insert_order_around_a_per_row_run(temp_db_path, new_documents): + 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") # per-row rows between and at the ends of bulk runs + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Bulk", False) + db.insert_many("Bulk", rows) + if JVM: + assert len(new_documents) == 5 + # a scan returns records in insertion (RID) order + 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_parallel_insert_many_carries_the_same_values(temp_db_path, new_documents): + rows = _rows(list(VALUES)) + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Ref", False) + _create(db, "Par", False) + _per_row(db, "Ref", rows) + new_documents.clear() + db.insert_many("Par", rows, parallel=True) + assert new_documents == [] + assert _stored(db, "Par") == _stored(db, "Ref") + + +def test_parallel_insert_many_sends_only_the_uncarriable_row_per_row( + temp_db_path, new_documents +): + rows = [{"id": i, "ts": dt.datetime(2026, 1, 1)} for i in range(500)] + rows[250]["v"] = float("nan") + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Par", False) + db.insert_many("Par", rows, parallel=True) + if JVM: + assert new_documents == ["Par"] + assert db.count_type("Par") == 500 + + +def test_a_user_dict_that_looks_like_a_tag_is_not_decoded(temp_db_path): + """The bulk JSON marks a date-time or a Decimal as {"\\u0001": [kind, payload]}; a user's own map with that key stays a map.""" + lookalike = {"\u0001": ["T", "2026-01-01T00:00:00"]} + rows = [{"id": 0, "m": lookalike, "ts": dt.datetime(2026, 1, 1)}] + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Ref", False) + _create(db, "Bulk", False) + _per_row(db, "Ref", rows) + db.insert_many("Bulk", rows) + assert _stored(db, "Bulk") == _stored(db, "Ref") + + +@pytest.mark.parametrize("parallel", [False, True], ids=["serial", "parallel"]) +def test_a_vertex_type_stays_vertices_when_a_row_goes_per_row(temp_db_path, parallel): + """A bulk load into a vertex type creates vertices; so must the rows that fall back to the per-row path. + + One row carries NaN (not carriable by the JSON text) and the rest carry a datetime (carried as a tagged value). + A CREATE EDGE needs vertices at both ends, so it refuses a record that came out as a plain document. + """ + n = 50 + rows = [{"id": i, "ts": dt.datetime(2026, 1, 1, 0, 0, i % 60)} for i in range(n)] + rows[20]["v"] = float("nan") + with arcadedb.create_database(temp_db_path) as db: + db.command("sql", "CREATE VERTEX TYPE Vtx") + db.command("sql", "CREATE EDGE TYPE Knows") + assert db.insert_many("Vtx", rows, parallel=parallel) == n + with db.transaction(): + db.command( + "sql", + "CREATE EDGE Knows FROM (SELECT FROM Vtx WHERE id = 20) " + "TO (SELECT FROM Vtx WHERE id = 21)", + ) + assert db.query("sql", "SELECT count(*) AS n FROM Knows").to_list()[0]["n"] == 1 + assert db.query("sql", "SELECT count(*) AS n FROM Vtx").to_list()[0]["n"] == n