diff --git a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc index 148c077..2b4d2dc 100644 --- a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc +++ b/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] diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index 8fb321f..1625984 100644 --- a/src/arcadedb/_native/database.py +++ b/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: diff --git a/tests/conftest.py b/tests/conftest.py index 01ab332..9ffd624 100644 --- a/tests/conftest.py +++ b/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 diff --git a/tests/test_insert_many_mixed_values.py b/tests/test_insert_many_mixed_values.py index f0a2eef..ce4949c 100644 --- a/tests/test_insert_many_mixed_values.py +++ b/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() diff --git a/tests/test_native_columnar_insert.py b/tests/test_native_columnar_insert.py index a3591d5..1ec2984 100644 --- a/tests/test_native_columnar_insert.py +++ b/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): diff --git a/tests/test_native_mixed_chunk.py b/tests/test_native_mixed_chunk.py new file mode 100644 index 0000000..db4162c --- /dev/null +++ b/tests/test_native_mixed_chunk.py @@ -0,0 +1,332 @@ +"""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 numpy as np +import pandas as pd +import pytest + +import arcadedb +from arcadedb._native import lib + +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 = [] + select = f"SELECT FROM {type_name} ORDER BY id" # nosec B608 - test-owned type name + for rec in db.query("sql", select).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_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 +): + 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.""" + 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") + + +def test_pandas_na_and_nat_store_what_a_none_stores_on_the_statement_path( + temp_db_path, spy +): + """pd.NA and pd.NaT are refused by the statement path, so the reference is the same rows with None in their place.""" + with_na = [ + {"id": 0, "a": pd.NA, "b": pd.NaT, "c": 1}, + {"id": 1, "a": 2, "b": dt.datetime(2026, 5, 6), "c": pd.NA}, + ] + with_none = [ + {k: (None if v is pd.NA or v is pd.NaT else v) for k, v in r.items()} + for r in with_na + ] + with arcadedb.create_database(temp_db_path) as db: + _create(db, "Ref", False) + _create(db, "Bulk", False) + _statement_path(db, "Ref", with_none) + spy.clear() + db.insert_many("Bulk", with_na) + assert spy.count(MANY) == 0 and spy.count(ROWS) == 1 + assert _stored(db, "Bulk") == _stored(db, "Ref")