From c1a78388510abc29fca69f4620628289e65309f1 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 14 Nov 2026 02:43:51 +0900 Subject: [PATCH 01/11] Test: a TypedRows batch with no tagged value is decoded without an object_hook and without reading its side array (refs #315) to_list() and iter_dicts() of the JVM wheel build a json object_hook for every batch and read the side array of every batch, whether or not the batch holds a tagged value. These tests count the hook factory, the object_hook argument of json.loads and the batch elements that are read, for a result of plain values (no hook, no read of the side array) and for every kind of value the bridge tags (the hook stays). They also compare to_list() and iter_dicts() with the decoding that always builds the hook, over single shapes, a mixed result and batch boundaries. They fail on the current decoder (every batch gets a hook and a read of the side array) and pass once the decoder looks for the tag escape in the batch text first. Co-Authored-By: Claude Sonnet 5.5 --- tests/conftest.py | 1 + tests/test_typed_batch_guard.py | 401 ++++++++++++++++++++++++++++++++ 2 files changed, 402 insertions(+) create mode 100644 tests/test_typed_batch_guard.py diff --git a/tests/conftest.py b/tests/conftest.py index 32ddf8f..e7d904c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -153,6 +153,7 @@ "test_async_proxy_cache.py": "the JPype proxy classes of the JVM build's async executor", "test_read_crossings.py": "crossing counts of the JVM build's read path (JPype proxies of the Java objects and the bridge classes)", "test_reader_crossings.py": "crossing counts of the JVM build's record and row readers (JPype proxies of the Java objects and the bridge classes)", + "test_typed_batch_guard.py": "the object_hook and the side array of the JVM build's TypedRows batches (JPype arrays, the bridge class, a spy on json.loads)", "test_jvm_ingest_integrity_faults.py": "rows-dropped faults put into the JVM build's column conversion and bulk vertex path (core._column_to_java, GraphBatch._create_vertices_json_bulk)", } if BACKEND == "native" and not os.environ.get("NATIVE_PARITY_ALL"): diff --git a/tests/test_typed_batch_guard.py b/tests/test_typed_batch_guard.py new file mode 100644 index 0000000..688a587 --- /dev/null +++ b/tests/test_typed_batch_guard.py @@ -0,0 +1,401 @@ +"""A TypedRows batch with no tagged value is decoded without a ``json`` object_hook and without reading its side array (#315). + +``ResultSet.to_list()`` and ``iter_dicts()`` read rows through the bridge's ``TypedRows``: a batch is +``{String json, Object[] side, Boolean drained}``. Java writes a value JSON cannot carry exactly as +``{"\\u0001": [kind, payload]}`` and Python restores it with an ``object_hook``, which reads the side +array. A batch of plain values (strings, integers, finite floats, booleans, nulls, lists, maps) holds no +tag, so the hook and the read of the side array are two JPype operations and a Python call per object +for nothing. The decoder now looks for the tag escape in the batch text first, as ``_loads_typed_row`` +already did for one row. + +Two things are pinned here, and neither may cost the other: + +* the guard fires: a batch without the tag escape is parsed by plain ``json.loads`` and ``batch[1]`` is + not read (the hook factory and the ``object_hook`` argument are counted); +* the guard never fires on a tagged batch: every shape the bridge tags keeps the hook, and the rows are + the ones the always-hooked decoder gives, which is what ``to_list()`` did before. + +Only the JVM build has this reader (``NATIVE_NOT_APPLICABLE`` in ``conftest.py``). +""" + +import contextlib +import json +import types + +import jpype +import pytest + +from arcadedb import results + +# the wide table with every property type (fixture ``db``), the row-by-row reference and the in-memory result sets +from tests.test_typed_rows import _in_memory, db, exact, reference # noqa: F401 + +ESCAPE = "\\u0001" +BATCH = 1000 # results._TYPED_BATCH_ROWS: the rows of one to_list() crossing + + +class _Batch: + """A TypedRows batch that logs which of its three elements are read (0 json, 1 side, 2 drained).""" + + def __init__(self, target, reads): + self._target = target + self._reads = reads + + def __getitem__(self, index): + self._reads.append(index) + return self._target[index] + + +class _Bridge: + """The bridge class TypedRows, with every batch it returns wrapped in a _Batch.""" + + def __init__(self, real, reads): + self._real = real + self._reads = reads + + def nextRows(self, java_result_set, size): + return _Batch(self._real.nextRows(java_result_set, size), self._reads) + + +class _Json: + """The ``json`` module as results.py sees it, logging the keyword arguments of every ``loads``.""" + + def __init__(self, calls): + self._calls = calls + + def loads(self, text, **kwargs): + self._calls.append(kwargs) + return json.loads(text, **kwargs) + + def __getattr__(self, name): + return getattr(json, name) + + +@contextlib.contextmanager +def watching(): + """What the decoder of results.py does while the block runs. + + ``loads``: the keyword arguments of every json.loads call, ``hooks``: the side arrays the hook factory + was built with (one entry per batch decoded with a hook), ``reads``: the batch elements read. + Nothing is recorded outside the block, so the reference rows are computed first. + """ + seen = types.SimpleNamespace(loads=[], hooks=[], reads=[]) + real_hook = results._typed_hook + + def counting_hook(side): + seen.hooks.append(side) + return real_hook(side) + + bridge = _Bridge(results._bridge_class("TypedRows"), seen.reads) + with pytest.MonkeyPatch.context() as patch: + patch.setattr(results, "_typed_hook", counting_hook) + patch.setattr(results, "json", _Json(seen.loads)) + patch.setitem(results._BRIDGE_CLASSES, "TypedRows", bridge) + yield seen + + +def _hooked(seen): + """How many json.loads calls got an object_hook.""" + return sum(1 for kwargs in seen.loads if kwargs.get("object_hook") is not None) + + +def _texts(java_result_set, size): + """The JSON text of every batch TypedRows writes for the result set.""" + typed_rows = jpype.JClass("com.arcadedb.python.TypedRows") + texts = [] + while True: + batch = typed_rows.nextRows(java_result_set, size) + texts.append(str(batch[0])) + if batch[2]: + return texts + + +def _decoded_with_the_hook_always(java_result_set, size): + """The rows decoded as to_list() did before #315: an object_hook on every batch, tagged or not.""" + typed_rows = jpype.JClass("com.arcadedb.python.TypedRows") + rows = [] + while True: + batch = typed_rows.nextRows(java_result_set, size) + rows.extend( + json.loads(str(batch[0]), object_hook=results._typed_hook(batch[1])) + ) + if batch[2]: + return rows + + +def _all(rows): + return [exact(row) for row in rows] + + +def _java(name): + return jpype.JClass(name) + + +def _java_list(*items): + java_list = _java("java.util.ArrayList")() + for item in items: + java_list.add(item) + return java_list + + +def _java_map(**entries): + java_map = _java("java.util.LinkedHashMap")() + for key, value in entries.items(): + java_map.put(key, value) + return java_map + + +def _java_set(*items): + java_set = _java("java.util.LinkedHashSet")() + for item in items: + java_set.add(item) + return java_set + + +def _int_key_map(): + java_map = _java("java.util.HashMap")() + java_map.put(jpype.JInt(1), "one") + return java_map + + +def _tag_key_map(): + java_map = _java("java.util.HashMap")() + java_map.put("\u0001", "tag") + return java_map + + +# Every kind of value the bridge tags: the decoder must keep the hook for a batch that holds one. The +# lambdas build the Java objects when the test runs, as the JVM is not started when this module loads. +TAGGED = { + "decimal": lambda: _java("java.math.BigDecimal")("1.25"), + "decimal_exponent": lambda: _java("java.math.BigDecimal")("1E+3"), + "big_integer": lambda: _java("java.math.BigInteger")( + "123456789012345678901234567890" + ), + "date": lambda: _java("java.time.LocalDate").of(2020, 1, 2), + "date_year_one": lambda: _java("java.time.LocalDate").of(1, 1, 1), + "datetime": lambda: _java("java.time.LocalDateTime").of( + 2021, 2, 3, 4, 5, 6, 789123000 + ), + "instant": lambda: _java("java.time.Instant").parse( + "2021-02-03T04:05:06.789123456Z" + ), + "offset_datetime": lambda: _java("java.time.OffsetDateTime").parse( + "2021-02-03T04:05:06.5+09:00" + ), + "util_date": lambda: _java("java.util.Date")(1612325106789), + "set": lambda: _java_set("a", "b"), + "nan": lambda: float("nan"), + "inf": lambda: float("inf"), + "ninf": lambda: float("-inf"), + "float_array": lambda: jpype.JArray(jpype.JFloat)([0.1, 0.2, 0.3]), + "float_array_nan": lambda: jpype.JArray(jpype.JFloat)([0.5, float("nan")]), + "double_array": lambda: jpype.JArray(jpype.JDouble)([0.25, 0.5]), + "map_with_an_int_key": _int_key_map, + "map_with_the_tag_as_key": _tag_key_map, + "decimal_in_a_list": lambda: _java_list(1, _java("java.math.BigDecimal")("2.5")), + "date_in_a_map": lambda: _java_map( + k=_java_list(_java("java.time.LocalDate").of(1999, 12, 31)) + ), + "nan_in_a_map": lambda: _java_map(a=1, b=_java_list(float("nan"))), +} + +# Values that hold no tag; two of them put the escape into the text anyway (a control character and the +# text of the escape in a string), so the decoder keeps the hook and must still return the plain value. +PLAIN = { + "ints": lambda: 7, + "long": lambda: jpype.JLong(9223372036854775807), + "float": lambda: 0.1, + "negative_zero": lambda: -0.0, + "bool": lambda: True, + "string": lambda: 'q"uote\\ \n\t é 日本 \U0001f600', + "list": lambda: _java_list(1, "a", 2.5, True), + "nested_map": lambda: _java_map(a=1, b=_java_map(c=_java_list(1, 2, 3))), + "int_array": lambda: jpype.JArray(jpype.JInt)([1, 2, 3]), + "byte_array": lambda: jpype.JArray(jpype.JByte)([0, 1, -1]), + "control_character_in_a_string": lambda: "a\u0001b", + "the_escape_text_in_a_string": lambda: "x\\u0001y", +} + + +def _rows(value, count=3): + return [{"v": value} for _ in range(count)] + + +# ---------------------------------------------------------------- a batch with no tagged value + + +@pytest.fixture +def plain(temp_db): + """2,100 rows (three batches) of plain values only.""" + temp_db.command("sql", "CREATE DOCUMENT TYPE Plain") + with temp_db.transaction(): + for i in range(2100): + temp_db.command( + "sql", + "INSERT INTO Plain SET k = :k, s = :s, b = :b, x = :x, z = null, li = :li, ma = :ma", + { + "k": i, + "s": f'row {i} "q" \\ é', + "b": i % 2 == 0, + "x": i / 8, + "li": [i, "a", 0.5], + "ma": {"a": i, "b": {"c": [1, 2, 3]}}, + }, + ) + return temp_db + + +PLAIN_QUERY = "SELECT k, s, b, x, z, li, ma FROM Plain ORDER BY k" + + +def test_to_list_of_untagged_batches_uses_no_hook_and_reads_no_side_array(plain): + result_set = plain.query("sql", PLAIN_QUERY) # held: freeing it closes the Java one + texts = _texts(result_set._java_result_set, BATCH) + assert len(texts) == 3 + assert not any(ESCAPE in text for text in texts), "the premise: nothing is tagged" + expected = _all(reference(plain, PLAIN_QUERY)) + + with watching() as seen: + rows = plain.query("sql", PLAIN_QUERY).to_list() + + assert _all(rows) == expected + assert len(seen.loads) == 3 # one parse per batch + assert _hooked(seen) == 0 + assert seen.hooks == [] + assert set(seen.reads) == {0, 2}, "only the text and the drained flag are read" + + +def test_iter_dicts_of_untagged_rows_uses_no_hook_and_reads_no_side_array(plain): + expected = _all(reference(plain, PLAIN_QUERY)) + + with watching() as seen: + rows = list(plain.query("sql", PLAIN_QUERY).iter_dicts()) + + assert _all(rows) == expected + assert ( + len(seen.loads) == 2100 + 1 + ) # a row per crossing, and the empty batch at the end + assert _hooked(seen) == 0 + assert seen.hooks == [] + assert set(seen.reads) == {0}, "only the text is read" + + +def test_an_empty_result_is_decoded_without_a_hook(plain): + query = "SELECT k FROM Plain WHERE k < 0" + with watching() as seen: + assert plain.query("sql", query).to_list() == [] + assert list(plain.query("sql", query).iter_dicts()) == [] + assert _hooked(seen) == 0 and seen.hooks == [] + + +# ---------------------------------------------------------------- a batch with a tagged value + + +@pytest.mark.parametrize("name", sorted(TAGGED)) +def test_every_tagged_shape_keeps_the_hook(db, name): + rows = _rows(TAGGED[name]()) + assert any( + ESCAPE in text for text in _texts(_in_memory(rows), BATCH) + ), "the bridge writes the tag escape for this value" + expected = _all(r.to_dict() for r in results.ResultSet(_in_memory(rows), db)) + + with watching() as seen: + listed = results.ResultSet(_in_memory(rows), db).to_list() + assert _all(listed) == expected + assert len(seen.hooks) == 1 and _hooked(seen) == 1 # the one batch + assert 1 in seen.reads + + with watching() as seen: + iterated = list(results.ResultSet(_in_memory(rows), db).iter_dicts()) + assert _all(iterated) == expected + assert len(seen.hooks) == 3 and _hooked(seen) == 3 # every row is tagged + assert seen.loads[-1] == {}, "the empty batch at the end holds nothing to restore" + + +def test_the_decision_is_made_for_each_batch(db): + """One tagged row among 2,100 gives one hooked batch of three for to_list(), one hooked row for iter_dicts().""" + decimal = _java("java.math.BigDecimal")("1.5") + rows = [{"n": decimal if i == 1500 else i} for i in range(2100)] + expected = _all(r.to_dict() for r in results.ResultSet(_in_memory(rows), db)) + + with watching() as seen: + listed = results.ResultSet(_in_memory(rows), db).to_list() + assert _all(listed) == expected + assert type(listed[1500]["n"]).__name__ == "Decimal" + assert len(seen.loads) == 3 + assert [kwargs.get("object_hook") is not None for kwargs in seen.loads] == [ + False, + True, + False, + ] + assert len(seen.hooks) == 1 + assert seen.reads.count(1) == 1 + + with watching() as seen: + iterated = list(results.ResultSet(_in_memory(rows), db).iter_dicts()) + assert _all(iterated) == expected + assert len(seen.loads) == 2100 + 1 + assert len(seen.hooks) == 1 and _hooked(seen) == 1 + assert seen.reads.count(1) == 1 + + +# ---------------------------------------------------------------- the rows are the ones the hook gave + + +def _parity(make_result_set): + """to_list() and iter_dicts() against the always-hooked decoding of the same batches. + + ``make_result_set`` returns a new ResultSet on every call. Each one is held while it is read: a + ResultSet that is freed closes its Java result set. + """ + first = make_result_set() + old = _all(_decoded_with_the_hook_always(first._java_result_set, BATCH)) + second = make_result_set() + one_row_at_a_time = _all(_decoded_with_the_hook_always(second._java_result_set, 1)) + assert one_row_at_a_time == old + assert _all(make_result_set().to_list()) == old + assert _all(make_result_set().iter_dicts()) == old + return len(old) + + +@pytest.mark.parametrize("name", sorted(TAGGED) + sorted(PLAIN)) +def test_each_shape_decodes_as_with_the_hook_always(db, name): + value = (TAGGED.get(name) or PLAIN[name])() + assert _parity(lambda: results.ResultSet(_in_memory(_rows(value)), db)) == 3 + + +def test_a_mixed_batch_decodes_as_with_the_hook_always(db): + """Every shape in one result set: tagged and plain values side by side, and a row of its own for each.""" + values = {name: make() for name, make in {**PLAIN, **TAGGED}.items()} + rows = [values] + [{"v": value} for value in values.values()] + assert _parity(lambda: results.ResultSet(_in_memory(rows), db)) == len(rows) + + +def test_the_wide_table_decodes_as_with_the_hook_always(db): + """Every property type, NaN and the infinities, dates at both ends of the range, a Decimal, a vector, a link.""" + assert _parity(lambda: db.query("sql", "SELECT FROM Wide")) == 7 + + +@pytest.mark.parametrize( + "query", + [ + "SELECT i, s FROM Wide WHERE i IS NOT NULL", + "SELECT s FROM Wide WHERE s IS NOT NULL", + "SELECT x, f FROM Wide WHERE x IS NOT NULL", + "SELECT [1, 2] AS a, {'x': {'y': 1}} AS m", + "SELECT s AS `\u0001`, i FROM Wide LIMIT 3", + "SELECT count(*) AS c FROM Wide", + ], +) +def test_query_shapes_decode_as_with_the_hook_always(db, query): + assert _parity(lambda: db.query("sql", query)) >= 1 + + +def test_batches_across_the_boundary_decode_as_with_the_hook_always(db): + """2,500 rows with a tagged value in the first, the middle and the last batch.""" + decimal = _java("java.math.BigDecimal")("1.5") + date = _java("java.time.LocalDate").of(2020, 1, 2) + tagged = {3: decimal, 1700: float("nan"), 2499: date} + rows = [{"n": tagged.get(i, i)} for i in range(2500)] + assert _parity(lambda: results.ResultSet(_in_memory(rows), db)) == 2500 From 650c3eeefb393808339370a1d3efa3c15c4aca26 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 14 Nov 2026 02:45:18 +0900 Subject: [PATCH 02/11] JVM to_list and iter_dicts: parse a TypedRows batch with no tag in its text without an object_hook and without reading its side array (refs #315) Every batch was decoded with json.loads(text, object_hook=_typed_hook(side)) and its side array was read first, also when the batch held only plain values. The hook is a Python call per JSON object and the read of the side array is a JPype operation per batch, which for iter_dicts (one row per crossing) is one per row. TypedRows writes every tag as {"\u0001": [kind, payload]} with the escape spelled out, writes a control character of a string as the same escape, and never writes the raw character, and every side array entry is referred to by a tag in the text. A text without the escape therefore holds no tag and no side index, the hook would return each object unchanged, and plain json.loads gives the same value. The check is a substring scan of a str that already exists (_loads_typed_row did it for one row). A string that holds U+0001 or the text of the escape only keeps the hook, which then changes nothing. No Java change, and the native wheel is untouched. The speed effect is not measured: it waits for a timed A/B of a candidate wheel against 26.11.1.post2. Co-Authored-By: Claude Sonnet 5.5 --- src/arcadedb/results.py | 30 +++++++++++++++++++++++------- 1 file changed, 23 insertions(+), 7 deletions(-) diff --git a/src/arcadedb/results.py b/src/arcadedb/results.py index a17f8bc..20c0ee3 100644 --- a/src/arcadedb/results.py +++ b/src/arcadedb/results.py @@ -121,6 +121,24 @@ def _loads_typed_row(java_row): return json.loads(text, object_hook=_typed_hook(java_row[1])) +def _loads_typed_batch(java_batch): + """The rows of one ``TypedRows.nextRows`` batch, ``{String json, Object[] side, Boolean drained}``. + + The hook, and the read of the side array that builds it, are needed only when the text holds a + tag. TypedRows writes every tag as ``{"\\u0001": [kind, payload]}`` with the escape spelled out + (``tagged``, and the ``r`` row of ``appendRow``), writes a control character of a string as the + same escape (``quote``), and never writes the raw character, and every entry of the side array + is referred to by such a tag. So a text without the escape holds no tag and no side index: the + hook would return every object unchanged, and ``json.loads(text)`` is the same value without it. + A string that holds U+0001, or the text of the escape, also puts the escape into the text; that + only keeps the hook, which then changes nothing. + """ + text = str(java_batch[0]) + if "\\u0001" not in text: + return json.loads(text) + return json.loads(text, object_hook=_typed_hook(java_batch[1])) + + def _loads_names(java_text): """The property names ``ReaderCalls.rowNames`` / ``documentNames`` wrote, as a tuple.""" return tuple(json.loads(str(java_text))) @@ -340,7 +358,9 @@ def _to_list_typed(self, typed_rows) -> List[Dict[str, Any]]: else (RIDs, embedded documents, non-finite floats, ...) by index into a side array of the engine's own objects, so every value comes back as the type ``convert_java_to_python`` gives it. The C ``json`` module - parses the text and one hook call per tagged value restores it. + parses the text and one hook call per tagged value restores it; a batch + with no tag in its text is parsed without the hook and its side array + is not read (``_loads_typed_batch``). """ out: List[Dict[str, Any]] = [] if not self._readable(): @@ -352,9 +372,7 @@ def _to_list_typed(self, typed_rows) -> List[Dict[str, Any]]: ) except JException as exc: raise _read_error(exc) from exc - side = java_batch[1] - rows = json.loads(str(java_batch[0]), object_hook=_typed_hook(side)) - out.extend(rows) + out.extend(_loads_typed_batch(java_batch)) if java_batch[2]: # the result set is drained, and the bridge closed it: no # second call just to see an empty batch. (A batch also ends @@ -392,9 +410,7 @@ def iter_dicts(self, convert_types: bool = True) -> Iterator[Dict[str, Any]]: java_batch = typed_rows.nextRows(self._java_result_set, 1) except JException as exc: raise _read_error(exc) from exc - rows = json.loads( - str(java_batch[0]), object_hook=_typed_hook(java_batch[1]) - ) + rows = _loads_typed_batch(java_batch) if not rows: # nextRows closed the drained result set itself self._drained() From 04fd969c0a8e732c55e6ccccf23514d17f0915f7 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 14 Nov 2026 02:47:57 +0900 Subject: [PATCH 03/11] Tests first: timestamp and date columns of insert_arrow and insert_dataframe on the JVM wheel (refs #315) Round trips of every Arrow temporal type through a real load on both wheels (every timestamp unit with and without a zone, to the nanosecond; date32 and date64 before 1970; nulls; batches; vertex and parallel loads; pandas and Polars frames), and JVM-only tests that count what one load does: no java.time call and no convert_python_to_java per cell, and each timestamp or date column reaching the bridge as one TemporalColumn per batch. The counting tests fail on the code before, which builds one LocalDateTime or LocalDate per cell in Python and converts every element again. Co-Authored-By: Claude Sonnet 5.5 --- tests/conftest.py | 1 + tests/test_arrow_temporal_column_jvm.py | 271 ++++++++++++++++++++++++ tests/test_dataframe_ingest.py | 229 ++++++++++++++++++++ 3 files changed, 501 insertions(+) create mode 100644 tests/test_arrow_temporal_column_jvm.py diff --git a/tests/conftest.py b/tests/conftest.py index e7d904c..3aa8a12 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -150,6 +150,7 @@ "test_graph_batch_vector_crossings_jvm.py": "the JPype crossings of the JVM wheel's create_vertices vector path (the bridge classes, the property-matrix fallback)", "test_insert_columns_string_blob.py": "the bridge's StringColumn and the JVM wheel's core._column_to_java (JPype arrays, a spy on the per-element conversion)", "test_arrow_vector_column_jvm.py": "the JPype crossings of the JVM wheel's Arrow vector column (core.convert_python_to_java, the bridge's FloatVectors.Matrix) and its Python heap", + "test_arrow_temporal_column_jvm.py": "the JPype calls of the JVM wheel's Arrow timestamp and date columns (the bridge's TemporalColumn, core._column_to_java, java.time classes)", "test_async_proxy_cache.py": "the JPype proxy classes of the JVM build's async executor", "test_read_crossings.py": "crossing counts of the JVM build's read path (JPype proxies of the Java objects and the bridge classes)", "test_reader_crossings.py": "crossing counts of the JVM build's record and row readers (JPype proxies of the Java objects and the bridge classes)", diff --git a/tests/test_arrow_temporal_column_jvm.py b/tests/test_arrow_temporal_column_jvm.py new file mode 100644 index 0000000..ef7c6dc --- /dev/null +++ b/tests/test_arrow_temporal_column_jvm.py @@ -0,0 +1,271 @@ +"""insert_arrow and insert_dataframe with timestamp and date columns on the JVM wheel: a column crosses as one array of epoch ticks per batch, +not as one JPype call per cell, and what is stored is unchanged. + +``_column_values`` turned every timestamp or date column into a Python list of ``LocalDateTime`` or ``LocalDate``, one +``ofEpochSecond`` or ``ofEpochDay`` call per cell, and ``_column_to_java`` then converted every element again with +``convert_python_to_java``: a million-row timestamp column was two million JPype calls. Now the column is the integer epoch values in ONE +``long[]`` (plus one ``boolean[]`` when a row is null) wrapped in the bridge's ``TemporalColumn``, built with one call per column and batch, and +``DocumentBatcher`` builds each ``LocalDateTime`` or ``LocalDate`` as it fills the document of that row. + +Two tests fail on the code before (they count the per-cell ``java.time`` calls, the per-element conversions and the columns that reach the bridge +as a ``TemporalColumn``); the others pin what the old route stored, so the new one must store the same: the bridge column against the per-cell +reading it replaced (every unit, negative ticks, nulls) and the refusal of a bad unit and of a timestamp past the years of ``java.time`` +before a row is written. The round trips of every Arrow temporal type through a real load, with the stored wall clock to the nanosecond, +are in ``test_dataframe_ingest.py::TestTemporalColumns``, which runs on both wheels. The JVM build only: it counts JPype calls and builds +bridge objects, so conftest.py does not collect this file on the native wheel. +""" + +from collections import Counter + +import jpype +import numpy as np +import pandas as pd +import pyarrow as pa +import pytest + +from arcadedb import core, type_conversion +from arcadedb.jvm import start_jvm + +TEMPORAL_COLUMN = "com.arcadedb.python.TemporalColumn" +UNITS = {"s": 1, "ms": 1_000, "us": 1_000_000, "ns": 1_000_000_000} +TEMPORAL_COLUMNS = 7 # the five timestamp columns and the two date columns of _table() + + +@pytest.fixture(scope="module", autouse=True) +def _jvm(): + """The tests that build a bridge object without a database need the JVM too (a no-op when it already runs).""" + start_jvm() + + +def _column_class(): + return jpype.JClass(TEMPORAL_COLUMN) + + +def _table(n, start=0): + """n rows: a timestamp column of every unit, one with a zone, a date32 and a date64, a null in every seventh row.""" + idx = np.arange(start, start + n, dtype=np.int64) + null = idx % 7 == 3 + stamp = idx * 1_000 - 5_000_000 + columns = {"idx": pa.array(idx)} + for unit in UNITS: + columns[f"t_{unit}"] = pa.array(stamp, mask=null).cast(pa.timestamp(unit)) + columns["zoned"] = pa.array(stamp, mask=null).cast( + pa.timestamp("us", tz="Asia/Karachi") + ) + columns["d32"] = pa.array(idx - 300, mask=null).cast(pa.int32()).cast(pa.date32()) + columns["d64"] = pa.array((idx - 300) * 86_400_000, mask=null).cast(pa.date64()) + return pa.table(columns) + + +def _make(db, name="Stamp"): + db.command("sql", f"CREATE DOCUMENT TYPE {name}") + db.command("sql", f"CREATE PROPERTY {name}.idx LONG") + + +def _count(db, name): + sql = f"SELECT count(*) AS c FROM {name}" # nosec B608 - test-owned name + return db.query("sql", sql).first().get("c") + + +@pytest.fixture +def crossings(monkeypatch): + """What one load does on the JPype side: the ``java.time`` calls Python makes, the per-value conversions, and the type and length of every column handed to the bridge.""" + # the type conversion caches the classes it fetched first: fetch them now, so the counting proxy below never lands in that cache + type_conversion._get_java_time_types() + type_conversion._get_java_python_types() + log = {"java_time": [], "convert": [], "columns": []} + real_convert = core.convert_python_to_java + real_column = core._column_to_java + real_jclass = jpype.JClass + + class _Counting: + def __init__(self, name): + self._name = name + self._target = real_jclass(name) + + def __getattr__(self, attr): + member = getattr(self._target, attr) + if not callable(member): + return member + + def call(*args): + log["java_time"].append(f"{self._name}.{attr}") + return member(*args) + + return call + + def convert(value): + log["convert"].append(type(value).__name__) + return real_convert(value) + + def column_to_java(name, values): + array, n = real_column(name, values) + log["columns"].append((type(array).__name__.rsplit(".", 1)[-1], n)) + return array, n + + def jclass(name, *args, **kwargs): + if isinstance(name, str) and name.startswith("java.time."): + return _Counting(name) + return real_jclass(name, *args, **kwargs) + + monkeypatch.setattr(core, "convert_python_to_java", convert) + monkeypatch.setattr(core, "_column_to_java", column_to_java) + monkeypatch.setattr(jpype, "JClass", jclass) + return log + + +def _temporal_columns(log): + """The lengths of the columns that reached the bridge as a TemporalColumn.""" + return Counter(n for kind, n in log["columns"] if kind == "TemporalColumn") + + +# ---- the tests that fail on the code before the fix: they count what the load does, never how long it takes + + +def test_a_temporal_column_makes_no_java_time_call_and_no_conversion_per_cell( + temp_db, crossings +): + """Before: one ``ofEpochSecond`` or ``ofEpochDay`` and one ``convert_python_to_java`` per cell. After: neither, for any of the seven columns.""" + n = 600 + _make(temp_db) + assert temp_db.insert_arrow("Stamp", _table(n), batch_size=n) == n + assert crossings["java_time"] == [] + assert crossings["convert"] == [] + assert _temporal_columns(crossings) == {n: TEMPORAL_COLUMNS} + + +def test_each_temporal_column_crosses_as_one_temporal_column_per_batch( + temp_db, crossings +): + """1,300 rows in batches of 500 (500, 500, 300): each of the seven columns is ONE bridge column of the batch's rows, three times.""" + _make(temp_db) + assert temp_db.insert_arrow("Stamp", _table(1_300), batch_size=500) == 1_300 + assert _temporal_columns(crossings) == { + 500: 2 * TEMPORAL_COLUMNS, + 300: TEMPORAL_COLUMNS, + } + + +def test_a_dataframe_and_a_parallel_load_take_the_same_route(temp_db, crossings): + """``insert_dataframe`` reads the frame as an Arrow table and ``parallel=True`` fills the same columns: no per-cell call on either.""" + _make(temp_db) + frame = _table(400).to_pandas() + assert isinstance(frame, pd.DataFrame) + assert temp_db.insert_dataframe("Stamp", frame, batch_size=200) == 400 + _make(temp_db, "Par") + assert ( + temp_db.insert_arrow("Par", _table(400), batch_size=200, parallel=True) == 400 + ) + assert crossings["java_time"] == [] and crossings["convert"] == [] + assert _temporal_columns(crossings) == {200: 4 * TEMPORAL_COLUMNS} + + +# ---- what the bridge column holds + + +def _reference(ticks, per_second): + """The per-cell reading the old route made: ``LocalDateTime.ofEpochSecond`` of the floor quotient, the remainder in nanoseconds.""" + local_datetime = jpype.JClass("java.time.LocalDateTime") + utc = jpype.JClass("java.time.ZoneOffset").UTC + seconds, rest = np.divmod(np.asarray(ticks, dtype=np.int64), per_second) + nanos = rest * (1_000_000_000 // per_second) + return [ + local_datetime.ofEpochSecond(int(s), int(n), utc) + for s, n in zip(seconds, nanos) + ] + + +@pytest.mark.parametrize("unit", list(UNITS)) +def test_a_date_time_column_reads_as_the_per_cell_reading_did(unit): + per_second = UNITS[unit] + limit = min(100_000_000_000 * per_second, 2**62) # about the year 5000 + ticks = np.random.default_rng(7).integers(-limit, limit, size=2_000) + ticks[:4] = [ + -1, + 0, + 1, + -per_second, + ] # one tick before the epoch is a floor, not a truncation + nulls = np.zeros(len(ticks), dtype=np.bool_) + nulls[5::11] = True + column = _column_class().dateTimes( + jpype.JArray(jpype.JLong)(ticks), + per_second, + jpype.JArray(jpype.JBoolean)(nulls), + ) + assert int(column.rows()) == len(ticks) + expected = _reference(ticks, per_second) + for i, want in enumerate(expected): + got = column.get(i) + if nulls[i]: + assert got is None + else: + assert bool(got.equals(want)), (i, int(ticks[i]), str(got), str(want)) + + +def test_a_nanosecond_tick_keeps_its_nanosecond_and_a_null_flag_is_optional(): + ns = 1_577_934_245_123_456_789 # 2020-01-02 03:04:05.123456789 + column = _column_class().dateTimes( + jpype.JArray(jpype.JLong)([ns, -1]), 1_000_000_000, None + ) + assert str(column.get(0)) == "2020-01-02T03:04:05.123456789" + assert str(column.get(1)) == "1969-12-31T23:59:59.999999999" + + +@pytest.mark.parametrize( + "per_day, days", + [(1, [-1, 0, 1, 18_322, -135_140]), (86_400_000, [-86_400_000, 0, 86_400_000])], + ids=["date32", "date64"], +) +def test_a_date_column_reads_the_day(per_day, days): + column = _column_class().dates(jpype.JArray(jpype.JLong)(days), per_day, None) + local_date = jpype.JClass("java.time.LocalDate") + for i, value in enumerate(days): + assert bool(column.get(i).equals(local_date.ofEpochDay(value // per_day))) + + +def test_a_date64_value_is_milliseconds_divided_with_a_floor(): + """The numpy ``//`` the old route used: a millisecond before midnight is still the day before.""" + column = _column_class().dates( + jpype.JArray(jpype.JLong)([86_400_000 - 1, -1]), 86_400_000, None + ) + assert str(column.get(0)) == "1970-01-01" and str(column.get(1)) == "1969-12-31" + + +def test_the_bridge_refuses_what_it_cannot_read(): + longs = jpype.JArray(jpype.JLong)([1, 2]) + for per_second in ( + 0, + -1, + 7, + 3_000_000_000, + ): # none divides a second into whole nanoseconds + with pytest.raises(jpype.JException): + _column_class().dateTimes(longs, per_second, None) + with pytest.raises(jpype.JException): + _column_class().dates(longs, 0, None) + with pytest.raises(jpype.JException): # one null flag for two rows + _column_class().dateTimes(longs, 1_000, jpype.JArray(jpype.JBoolean)([False])) + + +def test_column_to_java_hands_a_temporal_column_over_as_it_is(): + column = _column_class().dates(jpype.JArray(jpype.JLong)([0, 1, 2]), 1, None) + array, n = core._column_to_java("day", column) + assert n == 3 and array is column + + +def test_a_timestamp_past_the_years_of_java_time_is_refused_before_a_row_is_written( + temp_db, +): + """``timestamp[s]`` holds 2**62 seconds, which no ``LocalDateTime`` does: the call raises what it raised before, with the batch unwritten.""" + _make(temp_db) + table = pa.table( + { + "idx": pa.array([0, 1], pa.int64()), + "t": pa.array([0, 2**62], pa.int64()).cast(pa.timestamp("s")), + } + ) + with pytest.raises(jpype.JException): + temp_db.insert_arrow("Stamp", table) + assert _count(temp_db, "Stamp") == 0 + assert not temp_db.is_transaction_active() diff --git a/tests/test_dataframe_ingest.py b/tests/test_dataframe_ingest.py index 449e6d7..6b6c8db 100644 --- a/tests/test_dataframe_ingest.py +++ b/tests/test_dataframe_ingest.py @@ -212,6 +212,235 @@ def test_dates_before_the_epoch(self, temp_db): assert _one_column(temp_db, pa.timestamp("ms"), values) == values +# The engine pattern that shows a stored wall clock to the nanosecond: a read through Result.get returns a Python datetime, +# which holds microseconds and cuts the rest. +_WALL_CLOCK = "yyyy-MM-dd HH:mm:ss.SSSSSSSSS" +_TICKS_PER_SECOND = {"s": 1, "ms": 1_000, "us": 1_000_000, "ns": 1_000_000_000} +# Epoch ticks of each unit: one tick before the epoch (a floor, never a truncation toward zero), the epoch, one tick after +# it, 2020-01-02 03:04:05.123456789 cut to the unit, 1900-01-01 00:00:00 plus one tick, a null, and the year 2255 (the last +# year a nanosecond column holds, so every unit stays inside the range of a DATETIME_NANOS property). +_STAMPS = { + "s": [-1, 0, 1, 1_577_934_245, -2_208_988_799, None, 9_000_000_000], + "ms": [-1, 0, 1, 1_577_934_245_123, -2_208_988_799_999, None, 9_000_000_000_000], + "us": [ + -1, + 0, + 1, + 1_577_934_245_123_456, + -2_208_988_799_999_999, + None, + 9_000_000_000_000_000, + ], + "ns": [ + -1, + 0, + 1, + 1_577_934_245_123_456_789, + -2_208_988_799_999_999_999, + None, + 9_000_000_000_000_000_000, + ], +} +_DAYS = [ + dt.date(1600, 3, 1), + dt.date(1969, 12, 31), + dt.date(1970, 1, 1), + dt.date(2020, 2, 29), + None, + dt.date(2999, 12, 31), +] + + +def _wall_clock(ticks, unit): + """The UTC wall clock of ``ticks`` of ``unit`` since the epoch as ``_WALL_CLOCK`` prints it (a None stays None).""" + if ticks is None: + return None + seconds, rest = divmod(ticks, _TICKS_PER_SECOND[unit]) + moment = dt.datetime(1970, 1, 1) + dt.timedelta(seconds=seconds) + nanos = rest * (1_000_000_000 // _TICKS_PER_SECOND[unit]) + return f"{moment.isoformat(sep=' ', timespec='seconds')}.{nanos:09d}" + + +def _timestamps(ticks, unit, tz=None): + """An Arrow timestamp column of integer ticks (None is a null), so that no Python datetime cuts a nanosecond.""" + return pa.array(ticks, pa.int64()).cast(pa.timestamp(unit, tz=tz)) + + +def _wall_clocks(db, type_name, *columns): + """Each column as the text of its stored wall clock, in the order of ``idx``: {column: [text or None, ...]}.""" + shown = ", ".join(f"{c}.format('{_WALL_CLOCK}') AS {c}" for c in columns) + sql = f"SELECT idx, {shown} FROM {type_name} ORDER BY idx" # nosec B608 - test-controlled + rows = _rows(db, sql) + return {c: [r.get(c) for r in rows] for c in columns} + + +class TestTemporalColumns: + """A timestamp or date column of a load keeps what the Arrow integers say: the unit, the zone rule, the nulls, the dates before 1970. + + The JVM wheel sends such a column to the bridge as one array of epoch ticks and builds the date-times in Java; the native + wheel lays it out as columnar buffers. The expected values come from the integers and the UTC rule, never from a + database read, so both wheels must store the same wall clock. + """ + + @pytest.mark.parametrize( + "tz", [None, "UTC", "Asia/Karachi"], ids=["naive", "utc", "zoned"] + ) + @pytest.mark.parametrize("unit", ["s", "ms", "us", "ns"]) + def test_a_timestamp_keeps_its_wall_clock_to_the_nanosecond( + self, temp_db, unit, tz + ): + """A zoned timestamp is already UTC epoch time and a naive one is the clock it shows: both read the integer.""" + ticks = _STAMPS[unit] + temp_db.command("sql", "CREATE DOCUMENT TYPE Stamp") + temp_db.command("sql", "CREATE PROPERTY Stamp.val DATETIME_NANOS") + table = pa.table( + { + "idx": pa.array(range(len(ticks)), pa.int64()), + "val": _timestamps(ticks, unit, tz), + } + ) + assert temp_db.insert_arrow("Stamp", table) == len(ticks) + assert _wall_clocks(temp_db, "Stamp", "val") == { + "val": [_wall_clock(t, unit) for t in ticks] + } + + @pytest.mark.parametrize("declared", [None, "DATE"], ids=["undeclared", "DATE"]) + @pytest.mark.parametrize( + "arrow_type", [pa.date32(), pa.date64()], ids=["date32", "date64"] + ) + def test_a_date_before_and_after_the_epoch_with_a_null( + self, temp_db, arrow_type, declared + ): + assert _one_column(temp_db, arrow_type, _DAYS, declared) == _DAYS + + @pytest.mark.parametrize( + "kind, parallel", + [("DOCUMENT", False), ("DOCUMENT", True), ("VERTEX", False)], + ids=["document", "document_parallel", "vertex"], + ) + def test_every_temporal_column_of_a_load_in_batches(self, temp_db, kind, parallel): + """23 rows in batches of 5 (the last of 3): six timestamp columns and two date columns, with their nulls in different rows.""" + n = 23 + idx = np.arange(n, dtype=np.int64) + units = { + "t_s": "s", + "t_ms": "ms", + "t_us": "us", + "t_ns": "ns", + "t_zoned": "us", + "t_null": "us", + } + ticks = { + "t_s": (idx * 86_400 - 1_000_000_000).tolist(), + "t_ms": (idx * 1_000_003 - 1_000_000_000_001).tolist(), + "t_us": (idx * 1_000_000_007 - 1_000_000_000_000_001).tolist(), + "t_ns": (idx * 1_000_000_000_011 - 1_000_000_000_000_000_001).tolist(), + "t_zoned": (idx * 777_777_777 - 5).tolist(), + "t_null": [None] * n, + } + for name, every in ( + ("t_s", 4), + ("t_ms", 5), + ("t_us", 6), + ("t_ns", 7), + ("t_zoned", 3), + ): + ticks[name] = [ + None if i % every == 2 else t for i, t in enumerate(ticks[name]) + ] + days = [ + ( + None + if i % 6 == 1 + else dt.date(2000, 1, 1) + dt.timedelta(days=i * 997 - 30_000) + ) + for i in range(n) + ] + temp_db.command("sql", f"CREATE {kind} TYPE Wide") + for name in units: + temp_db.command("sql", f"CREATE PROPERTY Wide.{name} DATETIME_NANOS") + columns = {"idx": pa.array(idx)} + for name, unit in units.items(): + tz = "Asia/Karachi" if name == "t_zoned" else None + columns[name] = _timestamps(ticks[name], unit, tz) + columns["d32"] = pa.array(days, pa.date32()) + columns["d64"] = pa.array(days, pa.date64()) + assert ( + temp_db.insert_arrow( + "Wide", pa.table(columns), batch_size=5, parallel=parallel + ) + == n + ) + assert _count(temp_db, "Wide") == n + assert _wall_clocks(temp_db, "Wide", *units) == { + name: [_wall_clock(t, unit) for t in ticks[name]] + for name, unit in units.items() + } + stored = _rows(temp_db, "SELECT idx, d32, d64 FROM Wide ORDER BY idx") + assert [r.get("d32") for r in stored] == days + assert [r.get("d64") for r in stored] == days + + def test_pandas_and_polars_columns_keep_their_unit_zone_and_nulls(self, temp_db): + """pandas datetime64[ns] with a NaT, the same instants with a zone, and Polars Datetime of two units and a Date.""" + ticks = _STAMPS["ns"] + nat = np.iinfo(np.int64).min + values = np.array([nat if t is None else t for t in ticks], dtype="int64") + naive = pd.Series(values.view("datetime64[ns]")) + pandas_frame = pd.DataFrame( + { + "idx": np.arange(len(ticks), dtype=np.int64), + "t_naive": naive, + "t_zoned": naive.dt.tz_localize("UTC").dt.tz_convert("Etc/GMT-5"), + } + ) + whole_ms = [ + dt.datetime(1969, 12, 31, 23, 59, 59, 999000), + dt.datetime(2020, 1, 2, 3, 4, 5, 123000), + None, + dt.datetime(1900, 1, 1, 0, 0, 0, 1000), + ] + micros = [ + dt.datetime(1969, 12, 31, 23, 59, 59, 999999), + dt.datetime(2020, 1, 2, 3, 4, 5, 123456), + None, + dt.datetime(1900, 1, 1, 0, 0, 0, 1), + ] + polars_frame = pl.DataFrame( + { + "idx": pl.Series(range(4), dtype=pl.Int64), + "t_ms": pl.Series(whole_ms, dtype=pl.Datetime("ms")), + "t_ns": pl.Series(micros, dtype=pl.Datetime("ns")), + "day": pl.Series([m and m.date() for m in micros], dtype=pl.Date), + } + ) + for name, columns in ( + ("FromPandas", ("t_naive", "t_zoned")), + ("FromPolars", ("t_ms", "t_ns")), + ): + temp_db.command("sql", f"CREATE DOCUMENT TYPE {name}") + for column in columns: + temp_db.command( + "sql", f"CREATE PROPERTY {name}.{column} DATETIME_NANOS" + ) + assert temp_db.insert_dataframe("FromPandas", pandas_frame) == len(ticks) + assert temp_db.insert_dataframe("FromPolars", polars_frame) == 4 + shown = [_wall_clock(t, "ns") for t in ticks] + assert _wall_clocks(temp_db, "FromPandas", "t_naive", "t_zoned") == { + "t_naive": shown, + "t_zoned": shown, # the zone is only how pandas shows the instant: it is stored as the UTC wall clock + } + got = _wall_clocks(temp_db, "FromPolars", "t_ms", "t_ns") + assert got["t_ms"] == [ + m and m.isoformat(sep=" ", timespec="milliseconds") + "000000" + for m in whole_ms + ] + assert got["t_ns"] == [ + m and m.isoformat(sep=" ", timespec="microseconds") + "000" for m in micros + ] + days = _rows(temp_db, "SELECT idx, day FROM FromPolars ORDER BY idx") + assert [r.get("day") for r in days] == [m and m.date() for m in micros] + + class TestNulls: def test_every_kind_stores_nulls_as_missing(self, temp_db): temp_db.command("sql", "CREATE DOCUMENT TYPE Nul") From ef382809d54648c053b449b59b4a1942460ca184 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 14 Nov 2026 02:50:26 +0900 Subject: [PATCH 04/11] JVM insert_arrow, insert_dataframe, insert_columns: a timestamp or date column crosses as one long[] of epoch ticks, not one JPype call per cell (refs #315) _temporal_to_java made one LocalDateTime.ofEpochSecond or LocalDate.ofEpochDay call per cell, and insert_columns converted every element of that list again with convert_python_to_java: a million-row timestamp column was two million JPype calls. The column is now the integer epoch values in one long[] (plus one boolean[] when a row is null) wrapped in the new bridge class TemporalColumn (dateTimes(long[] ticks, long ticksPerSecond, boolean[] nulls) and dates(long[] ticks, long ticksPerDay, boolean[] nulls)), and DocumentBatcher builds each LocalDateTime or LocalDate as it fills the document of that row, as StringColumn does for text. The reading rules are the ones of the code before: UTC wall clock, a zoned timestamp is already UTC epoch time, a naive one is the clock it shows, the unit's precision is kept (a nanosecond tick is not cut to microseconds), a tick before the epoch floors, date64 is milliseconds divided by 86,400,000 with a floor. The factory checks the smallest and the largest tick, so a value past the years of java.time still raises before a row is written. The per-cell path is retired: the wheel carries its own bridge jar, so no fallback for an older one. Not measured: the speed claim waits for a timed A/B against 26.11.1.post2. Co-Authored-By: Claude Sonnet 5.5 --- src/arcadedb/core.py | 5 +- src/arcadedb/dataframe.py | 60 ++++----- .../com/arcadedb/python/DocumentBatcher.java | 16 ++- .../com/arcadedb/python/TemporalColumn.java | 118 ++++++++++++++++++ 4 files changed, 163 insertions(+), 36 deletions(-) create mode 100644 src/java/com/arcadedb/python/TemporalColumn.java diff --git a/src/arcadedb/core.py b/src/arcadedb/core.py index a050b72..68fc806 100644 --- a/src/arcadedb/core.py +++ b/src/arcadedb/core.py @@ -241,7 +241,8 @@ def _column_to_java(name, values): into a ``long[]``, ``double[]``, or ``boolean[]``; a string or object array and any other sequence of mixed values convert per element into an ``Object[]`` (a ``str`` reuses one Java String per distinct value, ``None`` is a null); a column of nothing but ``str`` and ``None`` crosses instead as one - ``StringColumn`` (``_string_column``). A pandas + ``StringColumn`` (``_string_column``). A bridge ``TemporalColumn``, which ``insert_arrow`` and ``insert_dataframe`` build from a + timestamp or date column (``dataframe._temporal_column``), crosses as it is. A pandas ``Series`` is read through ``to_numpy``, with a nullable dtype's ```` as ``None``. A 2-D ``float32`` array is a column of vectors: it crosses as one ``float[]`` and each document gets one row of it. Kinds that do not cross natively raise ``TypeError`` rather @@ -313,6 +314,8 @@ def _column_to_java(name, values): # a string, bytes, or object array: its elements are Python objects (str, None, ...), converted one by one below values = list(values.astype(object, copy=False)) elif not isinstance(values, (list, tuple)): + if isinstance(values, _java_class("com.arcadedb.python.TemporalColumn")): + return values, int(values.rows()) values = list(values) column = _string_column(values) if column is not None: diff --git a/src/arcadedb/dataframe.py b/src/arcadedb/dataframe.py index 4c7beb2..6fa3de5 100644 --- a/src/arcadedb/dataframe.py +++ b/src/arcadedb/dataframe.py @@ -245,43 +245,46 @@ def _check_declared( ) -def _temporal_to_java(chunk, pa): - """A timestamp or date chunk as a list of ``LocalDateTime``/``LocalDate``, keeping the unit's precision. +def _temporal_column(chunk, pa): + """A timestamp or date chunk as one ``TemporalColumn`` of the bridge, keeping the unit's precision. + + The integer epoch values cross as ONE ``long[]`` (a buffer copy), the null rows as one ``boolean[]`` when there is a null, and the + bridge builds each ``LocalDateTime`` or ``LocalDate`` as it fills the document of that row: one call per column and batch, where a + ``LocalDateTime.ofEpochSecond`` or ``LocalDate.ofEpochDay`` call per cell (and one conversion per element in ``insert_columns``) came + before. ``insert_columns`` takes the column as it is. DATETIME is a UTC wall clock: a timestamp with a zone is already UTC epoch time, and a timestamp without one is taken as the wall clock it shows (the same rule ``Document.set`` applies to a ``datetime``), so both - are read from the integer epoch value. Going through ``datetime`` would cut a nanosecond value to microseconds. + are read from the integer epoch value, in the unit of the column. Going through ``datetime`` would cut a nanosecond value to + microseconds. A date32 is days, a date64 is milliseconds (the bridge divides by 86,400,000, rounding toward negative infinity). """ import numpy as np types = pa.types - mask = chunk.is_null().to_numpy(zero_copy_only=False) if types.is_date(chunk.type): - local_date = jpype.JClass("java.time.LocalDate") - raw = chunk.cast(pa.int64() if types.is_date64(chunk.type) else pa.int32()) - days = raw.fill_null(0).to_numpy(zero_copy_only=False).astype(np.int64) - if types.is_date64(chunk.type): - days = days // 86_400_000 - return [ - None if m else local_date.ofEpochDay(int(d)) for d, m in zip(days, mask) - ] - per_second = {"s": 1, "ms": 1_000, "us": 1_000_000, "ns": 1_000_000_000}[ - chunk.type.unit - ] - values = chunk.cast(pa.int64()).fill_null(0).to_numpy(zero_copy_only=False) - seconds, rest = np.divmod(values.astype(np.int64), per_second) - nanos = rest * (1_000_000_000 // per_second) - local_datetime = jpype.JClass("java.time.LocalDateTime") - utc = jpype.JClass("java.time.ZoneOffset").UTC - return [ - None if m else local_datetime.ofEpochSecond(int(s), int(n), utc) - for s, n, m in zip(seconds, nanos, mask) - ] + date64 = types.is_date64(chunk.type) + raw = chunk.cast(pa.int64() if date64 else pa.int32()) + ticks, per_unit, factory = raw, 86_400_000 if date64 else 1, "dates" + else: + per_second = {"s": 1, "ms": 1_000, "us": 1_000_000, "ns": 1_000_000_000} + ticks = chunk.cast(pa.int64()) + per_unit, factory = per_second[chunk.type.unit], "dateTimes" + values = ticks.fill_null(0).to_numpy(zero_copy_only=False) + nulls = None + if chunk.null_count: + nulls = jpype.JArray(jpype.JBoolean)( + chunk.is_null().to_numpy(zero_copy_only=False) + ) + return getattr(jpype.JClass("com.arcadedb.python.TemporalColumn"), factory)( + jpype.JArray(jpype.JLong)(np.ascontiguousarray(values, dtype=np.int64)), + per_unit, + nulls, + ) def _temporal_to_tagged(chunk, pa): """A timestamp or date chunk as tagged values (``T`` date-time, ``D`` date) for the native backend, with the same - reading rules as :func:`_temporal_to_java`: ISO text of the integer epoch value, so a nanosecond value keeps its + reading rules as :func:`_temporal_column`: ISO text of the integer epoch value, so a nanosecond value keeps its nanoseconds (a Python ``datetime`` would cut them to microseconds). """ import numpy as np @@ -381,7 +384,8 @@ def _column_values( float_vectors: bool = False, ): """One Arrow array as what ``Database.insert_columns`` takes: a numpy array when the column crosses as one - buffer (integers, floats and booleans without nulls), else a list of values (``None`` is a null). + buffer (integers, floats and booleans without nulls), else a list of values (``None`` is a null). On the JVM backend a timestamp or + date column is one bridge ``TemporalColumn`` instead (``_temporal_column``: one ``long[]`` of epoch ticks, no value per cell). ``float_vectors`` (the JVM backend, a property declared ``ARRAY_OF_FLOATS``): a float32 vector column of one length and no null is one 2-D float32 matrix, which ``insert_columns`` sends as one ``float[]``; ``to_pylist`` made one Python float per element and @@ -434,9 +438,7 @@ def _column_values( if matrix is not None: return matrix if kind in ("timestamp", "date"): - return ( - _temporal_to_tagged(chunk, pa) if native else _temporal_to_java(chunk, pa) - ) + return _temporal_to_tagged(chunk, pa) if native else _temporal_column(chunk, pa) if native and kind in ("decimal", "binary"): return _scalars_to_tagged(kind, chunk) if kind == "null": diff --git a/src/java/com/arcadedb/python/DocumentBatcher.java b/src/java/com/arcadedb/python/DocumentBatcher.java index 0edeb45..8d892a9 100644 --- a/src/java/com/arcadedb/python/DocumentBatcher.java +++ b/src/java/com/arcadedb/python/DocumentBatcher.java @@ -81,17 +81,17 @@ public static long insertManyJson(final Database db, final String typeName, fina /** * Columnar insert: the documents are built Java-side from whole columns, so each column crosses the FFI once (a long[] or - * double[] copied from a numpy buffer, a boolean[], a StringColumn (one UTF-8 byte[] plus the int[] offsets), or an Object[] of - * boxed numbers, Strings of a mixed column, and nulls) instead of one + * double[] copied from a numpy buffer, a boolean[], a StringColumn (one UTF-8 byte[] plus the int[] offsets), a TemporalColumn (one + * long[] of epoch ticks), or an Object[] of boxed numbers, Strings of a mixed column, and nulls) instead of one * JSON text per batch (bindings issue #150: 2.24x over insertManyJson on the first * 2,000,000 TPC-H SF1 line items, nine typed properties, commit every 10,000, same sums and count). * * Same failure contract as insertManyJson (#7882): batch commits and the failure rollback apply only to a transaction * opened here; a caller's own transaction is the caller's to commit or roll back. * - * @param columns one array per name, each of length n: long[], double[], boolean[], Object[], a FloatVectors.Matrix or a - * StringColumn; a null element (or null row of a StringColumn) sets the property to null, as a JSON null does - * on the insertManyJson path + * @param columns one array per name, each of length n: long[], double[], boolean[], Object[], a FloatVectors.Matrix, a + * StringColumn or a TemporalColumn; a null element (or null row of a StringColumn or TemporalColumn) sets the + * property to null, as a JSON null does on the insertManyJson path */ public static long insertColumns(final Database db, final String typeName, final String[] names, final Object[] columns, final int n, final int commitEvery) { @@ -229,9 +229,11 @@ else if (col instanceof FloatVectors.Matrix m) len = m.rows(); else if (col instanceof StringColumn s) len = s.rows(); + else if (col instanceof TemporalColumn t) + len = t.rows(); else throw new IllegalArgumentException("column '" + names[c] + "' is a " + (col == null ? "null" : col.getClass().getName()) - + ", not a long[], double[], boolean[], Object[], FloatVectors.Matrix or StringColumn"); + + ", not a long[], double[], boolean[], Object[], FloatVectors.Matrix, StringColumn or TemporalColumn"); if (len != n) throw new IllegalArgumentException("column '" + names[c] + "' has " + len + " values for " + n + " rows"); } @@ -251,6 +253,8 @@ else if (col instanceof FloatVectors.Matrix m) v = m.row(i); else if (col instanceof StringColumn s) v = s.get(i); + else if (col instanceof TemporalColumn t) + v = t.get(i); else v = ((Object[]) col)[i]; doc.set(names[c], v); diff --git a/src/java/com/arcadedb/python/TemporalColumn.java b/src/java/com/arcadedb/python/TemporalColumn.java new file mode 100644 index 0000000..533fb3c --- /dev/null +++ b/src/java/com/arcadedb/python/TemporalColumn.java @@ -0,0 +1,118 @@ +/* + * Python-bindings bridge: a timestamp or date column of Database.insert_columns that crosses as one long[] of epoch ticks. + * + * Building a LocalDateTime or LocalDate in Python costs one JPype call per cell (and the list of them one conversion per element + * again), so a million-row timestamp column was two million calls. Here the whole column is ONE long[] (the integer epoch value of every + * row, as Arrow holds it), the tick size, and optionally ONE boolean[] marking the null rows. Row i is read when the document of that + * row is built, so the values are short-lived. + * + * A date-time is {@code ticks / ticksPerSecond} whole seconds (rounded toward negative infinity, so a tick before the epoch is the + * last tick of 1969) and the rest as nanoseconds, read as the UTC wall clock: the engine stores a DATETIME as a UTC wall clock, a + * zone-aware Arrow timestamp is already UTC epoch time and a zone-less one is taken as the wall clock it shows. The unit's precision is + * kept: a nanosecond tick is not cut to microseconds. A date is {@code ticks / ticksPerDay} days since the epoch, rounded the same way + * (a date32 is days, a date64 is milliseconds). + * + * Compiled into arcadedb-python-bridge.jar during the wheel build and consumed by DocumentBatcher.insertColumns() and + * insertColumnsParallel(). + */ +package com.arcadedb.python; + +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.ZoneOffset; + +public final class TemporalColumn { + private static final long NANOS_PER_SECOND = 1_000_000_000L; + + private final long[] ticks; + private final boolean[] nulls; + private final boolean date; + /** Ticks in one second (a date-time) or in one day (a date). */ + private final long ticksPerUnit; + /** Nanoseconds in one tick, for a date-time. */ + private final long nanosPerTick; + + private TemporalColumn(final long[] ticks, final boolean[] nulls, final boolean date, final long ticksPerUnit, + final long nanosPerTick) { + if (nulls != null && nulls.length != ticks.length) + throw new IllegalArgumentException("a temporal column has " + nulls.length + " null flags for " + ticks.length + " rows"); + this.ticks = ticks; + this.nulls = nulls; + this.date = date; + this.ticksPerUnit = ticksPerUnit; + this.nanosPerTick = nanosPerTick; + checkRange(); + } + + /** + * A column of date-times. + * + * @param ticks the epoch value of every row, in units of {@code 1 / ticksPerSecond} second since 1970-01-01T00:00:00 + * @param ticksPerSecond 1 (seconds), 1,000, 1,000,000 or 1,000,000,000: any divisor of a second into whole nanoseconds + * @param nulls null for a column without a null, else one flag per row (a flagged row reads as null and its tick is ignored) + * @throws IllegalArgumentException for a tick size that is not a whole number of nanoseconds, or a wrong number of null flags. + * A row past the years a LocalDateTime holds raises the DateTimeException of LocalDateTime itself, + * here, before anything is written + */ + public static TemporalColumn dateTimes(final long[] ticks, final long ticksPerSecond, final boolean[] nulls) { + if (ticksPerSecond < 1 || NANOS_PER_SECOND % ticksPerSecond != 0) + throw new IllegalArgumentException( + "ticks per second must divide 1,000,000,000 into whole nanoseconds, got " + ticksPerSecond); + return new TemporalColumn(ticks, nulls, false, ticksPerSecond, NANOS_PER_SECOND / ticksPerSecond); + } + + /** + * A column of dates. + * + * @param ticks the epoch value of every row, in units of {@code 1 / ticksPerDay} day since 1970-01-01 + * @param ticksPerDay 1 for a date32 (days), 86,400,000 for a date64 (milliseconds) + * @param nulls null for a column without a null, else one flag per row (a flagged row reads as null and its tick is ignored) + * @throws IllegalArgumentException for a tick size below 1, or a wrong number of null flags. A row past the days a LocalDate holds + * raises the DateTimeException of LocalDate itself, here, before anything is written + */ + public static TemporalColumn dates(final long[] ticks, final long ticksPerDay, final boolean[] nulls) { + if (ticksPerDay < 1) + throw new IllegalArgumentException("ticks per day must be at least 1, got " + ticksPerDay); + return new TemporalColumn(ticks, nulls, true, ticksPerDay, 0L); + } + + public int rows() { + return ticks.length; + } + + /** The LocalDateTime or LocalDate of row {@code i}, or null for a null row. */ + public Object get(final int i) { + if (nulls != null && nulls[i]) + return null; + return valueOf(ticks[i]); + } + + private Object valueOf(final long tick) { + if (date) + return LocalDate.ofEpochDay(Math.floorDiv(tick, ticksPerUnit)); + return LocalDateTime.ofEpochSecond(Math.floorDiv(tick, ticksPerUnit), (int) (Math.floorMod(tick, ticksPerUnit) * nanosPerTick), + ZoneOffset.UTC); + } + + /** + * The value of a row grows with its tick, so the smallest and the largest tick of the non-null rows are the two that can fall outside + * the years of java.time: a value past them raises here, as one call per cell did in Python, and nothing of the batch is written. + */ + private void checkRange() { + long min = Long.MAX_VALUE; + long max = Long.MIN_VALUE; + for (int i = 0; i < ticks.length; i++) { + if (nulls != null && nulls[i]) + continue; + final long tick = ticks[i]; + if (tick < min) + min = tick; + if (tick > max) + max = tick; + } + if (min <= max) { + valueOf(min); + valueOf(max); + } + } +} From a17daaed3feac3526c4bd5e0e8a7ed47c61fe555 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 14 Nov 2026 02:58:42 +0900 Subject: [PATCH 05/11] Tests: a native transaction crossing is one foreign call, with no reply text, no free and no JSON decode (refs #315) arcadedb_n_tx returns the transaction state as one integer (bit 0 active, bits 1 to 31 depth, bits 32 to 62 commits, -1 for an error whose text lib.last_error holds), as arcadedb_n_cmd returns a cursor id. These tests pin that and that an error still reaches Python as the same typed exception with the same message: - with db.transaction() is three foreign calls (mark, begin, commit), all arcadedb_n_tx: no strlen, no copy into a str, no arcadedb_free (was twelve); - begin, commit, rollback, is_transaction_active and the mark each make exactly one foreign call; - no JSON decode in a transaction, not even for the mark that reads active and depth; - the layout of the integer, -1 for an unknown handle with the error in last_error, commit without begin, rollback with nothing to roll back, a failed statement in a block, nested with blocks, an injected failure on each entry point, a database closed with a transaction open. They fail on the old code: lib.tx does not exist there, and the foreign-call and decode counts are higher. Co-Authored-By: Claude Sonnet 5.5 --- tests/conftest.py | 2 +- tests/test_native_one_row_crossings.py | 8 +- tests/test_native_tx_crossings.py | 271 +++++++++++++++++++++---- 3 files changed, 241 insertions(+), 40 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 3aa8a12..d59db09 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -93,7 +93,7 @@ collect_ignore.append("test_native_vector_provider.py") # the copy of a record's values as read that save() compares with the stored row (arcadedb._native.records), counted through its seams collect_ignore.append("test_native_save_baseline.py") - # crossing counts of the transaction entry points (lib.call is the native library's ctypes door) + # crossing and foreign-call counts of the transaction entry points (lib.tx and lib.call are the native library's ctypes doors) collect_ignore.append("test_native_tx_crossings.py") # counts the statement texts the native backend hands to the engine (the JPype backend has no such layer) collect_ignore.append("test_native_statement_texts.py") diff --git a/tests/test_native_one_row_crossings.py b/tests/test_native_one_row_crossings.py index 73f32a9..e65e3e9 100644 --- a/tests/test_native_one_row_crossings.py +++ b/tests/test_native_one_row_crossings.py @@ -7,7 +7,7 @@ reads the stored record first: the engine's reply to an UPDATE comes after the write, and the check must refuse before writing anything inside the caller's transaction, so it cannot be read from the update's own result. -The counts are crossings (the ctypes doors ``cmd``, ``call``, ``call_bin``, ``exec_query`` and the batched cursor close), not seconds: +The counts are crossings (the ctypes doors ``cmd``, ``call``, ``call_bin``, ``exec_query``, ``tx`` and the batched cursor close), not seconds: deterministic. The oracle for the values and the errors is the lazy form itself (``db.command(...).first()``), kept for every other caller. """ @@ -27,7 +27,7 @@ class _Crossings: def __init__(self, monkeypatch): self.names = [] - for attr in ("cmd", "call", "call_bin", "exec_query"): + for attr in ("cmd", "call", "call_bin", "exec_query", "tx"): monkeypatch.setattr(lib, attr, self._spy(attr, getattr(lib, attr))) real_close = lib._lib.arcadedb_n_rs_close_many @@ -39,7 +39,9 @@ def close_many(*args): def _spy(self, attr, real): def spy(*args, **kwargs): - self.names.append(args[0] if attr in ("call", "call_bin") else attr) + # the transaction door is named by its entry point, as it was when it went through lib.call + name = "arcadedb_n_tx" if attr == "tx" else attr + self.names.append(args[0] if attr in ("call", "call_bin") else name) return real(*args, **kwargs) return spy diff --git a/tests/test_native_tx_crossings.py b/tests/test_native_tx_crossings.py index 7f9f618..a218d61 100644 --- a/tests/test_native_tx_crossings.py +++ b/tests/test_native_tx_crossings.py @@ -1,11 +1,15 @@ -"""The native wheel spends the fewest library crossings a transaction needs. +"""The native wheel spends the fewest library crossings a transaction needs, and each one is a single foreign call. -Every ``arcadedb_n_tx`` call is one trip into the native library: a ctypes call, a malloc'd JSON reply, ``string_at``, a decode and a -``json.loads``. The reply of begin, commit and rollback already carries the whole transaction state (``active``, ``depth`` and -``commits``), so a caller that needs the state right after one of these calls must not ask for it again with a mark. +Every ``arcadedb_n_tx`` call is one trip into the native library. It used to return a malloc'd JSON text (about 60 bytes) that the +caller then read with ``strlen``, a copy into a ``str`` and ``arcadedb_free``: four foreign calls per crossing, twelve for +``with db.transaction()``. It now returns the transaction state as one integer (bit 0 ``active``, bits 1 to 31 ``depth``, bits 32 to +62 ``commits``, -1 for an error whose text ``lib.last_error`` holds), the way ``arcadedb_n_cmd`` returns a cursor id: one foreign +call, no reply text, no free, no JSON decode. The reply of begin, commit and rollback carries the whole state, so a caller that +needs the state right after one of these calls must not ask for it again with a mark. -The counts are crossings, not seconds: they are deterministic. The behaviour tests next to them pin what the marks are for (a caller's -transaction is never rolled back by a nested block, a retry rolls the failed attempt back), so a lower count can not buy a different result. +The counts are crossings and foreign calls, not seconds: they are deterministic. The behaviour tests next to them pin what the +marks are for (a caller's transaction is never rolled back by a nested block, a retry rolls the failed attempt back) and that an +error reaches Python as the same typed exception with the same message, so a lower count can not buy a different result. Needs the native library, so conftest.py does not collect this file on the JPype backend (a skip there would fail scripts/check_test_skips.py); it runs in the native parity run. @@ -20,20 +24,32 @@ from arcadedb._native import lib _OPS = {0: "begin", 1: "commit", 2: "rollback", 3: "mark"} +_DEPTH_MASK = 0x7FFFFFFF + + +def _fields(state): + """``(active, depth, commits)`` of the integer ``arcadedb_n_tx`` returns (the layout is pinned here, not read from the wheel).""" + return bool(state & 1), (state >> 1) & _DEPTH_MASK, state >> 32 class _Crossings: - """Counts every ``lib.call`` made while it is installed; ``tx`` is the sequence of ``arcadedb_n_tx`` operations.""" + """Counts every ``lib.tx`` and ``lib.call`` made while it is installed; ``tx`` is the sequence of transaction operations.""" def __init__(self, monkeypatch): self.calls = [] - real = lib.call + real_tx = lib.tx + real_call = lib.call + + def counting_tx(handle, op): + self.calls.append(_OPS[op]) + return real_tx(handle, op) - def counting(name, *args): - self.calls.append(_OPS[args[1]] if name == "arcadedb_n_tx" else name) - return real(name, *args) + def counting_call(name, *args): + self.calls.append(name) + return real_call(name, *args) - monkeypatch.setattr(lib, "call", counting) + monkeypatch.setattr(lib, "tx", counting_tx) + monkeypatch.setattr(lib, "call", counting_call) @property def tx(self): @@ -43,6 +59,51 @@ def reset(self): self.calls.clear() +class _ForeignCalls: + """Every foreign (ctypes) call the transaction path can make, by name, while it is installed. + + The entry points of the library are counted through a proxy of the loaded library object, and the two helpers that turn a reply + text into a ``str`` (``strlen`` and ``PyUnicode_FromStringAndSize``) by name. ``arcadedb_free`` is an entry point, so a reply + text that is read and released shows up as ``arcadedb_n_tx``, ``_strlen``, ``_str_from_utf8``, ``arcadedb_free``. + """ + + def __init__(self, monkeypatch): + self.names = [] + names = self.names + real_lib = lib._lib + + class _Proxy: + def __getattr__(self, attr): + function = getattr(real_lib, attr) + + def counted(*args): + names.append(attr) + return function(*args) + + return counted + + monkeypatch.setattr(lib, "_lib", _Proxy()) + for helper in ("_strlen", "_str_from_utf8"): + monkeypatch.setattr( + lib, helper, self._counted(helper, getattr(lib, helper)) + ) + + def _counted(self, name, function): + def counted(*args): + self.names.append(name) + return function(*args) + + return counted + + @property + def entries(self): + # a batched cursor close that an earlier statement left waiting runs before a commit: it belongs to that statement + return [n for n in self.names if n != "arcadedb_n_rs_close_many"] + + def reset(self): + del self.names[:] + + @pytest.fixture def db(temp_db_path): database = arcadedb.create_database(temp_db_path) @@ -51,6 +112,7 @@ def db(temp_db_path): database.transaction() ): # the thread's transaction context exists from here on, as it does in a running application pass + lib.flush_closes() # a cursor closed earlier is not part of a call measured below yield database database.drop() @@ -60,6 +122,11 @@ def crossings(db, monkeypatch): return _Crossings(monkeypatch) +@pytest.fixture +def foreign(db, monkeypatch): + return _ForeignCalls(monkeypatch) + + def _count(db): return db.query("sql", "SELECT count(*) AS c FROM R").first().get("c") @@ -119,6 +186,51 @@ def fn(): ] +# -- foreign calls: a crossing is one call into the library, with no reply text to read and free + + +def test_with_transaction_is_three_foreign_calls(db, foreign): + # was twelve: each of the three crossings (mark, begin, commit) was the entry, strlen, the copy into a str and arcadedb_free + with db.transaction(): + pass + assert foreign.entries == ["arcadedb_n_tx"] * 3 + + +def test_each_transaction_call_is_one_foreign_call_and_frees_nothing(db, foreign): + for call in ( + db.is_transaction_active, + db.begin, + db.is_transaction_active, + db.rollback, + db.begin, + db.commit, + db._transaction_stack_mark, + ): + foreign.reset() + call() + assert foreign.entries == ["arcadedb_n_tx"], call.__name__ + + +def test_a_transaction_that_raises_is_three_foreign_calls(db, foreign): + with pytest.raises(ValueError): + with db.transaction(): + raise ValueError("boom") + assert foreign.entries == ["arcadedb_n_tx"] * 3 + + +def test_a_transaction_never_reads_a_reply_text(db, monkeypatch): + taken = [] + real = lib._take + monkeypatch.setattr(lib, "_take", lambda *a: taken.append(1) or real(*a)) + with db.transaction(): + pass + db.begin() + assert db.is_transaction_active() is True + db.rollback() + assert db.run_in_transaction(lambda: 1) == 1 + assert taken == [] + + # -- what the marks are for: the same results at the lower count @@ -143,6 +255,18 @@ def test_a_nested_block_that_succeeds_commits_with_the_caller(db): assert _count(db) == 2 +def test_nested_with_blocks_push_and_pop_one_level_each(db): + base = db._transaction_stack_mark()[1] + with db.transaction(): + assert db._transaction_stack_mark() == (True, base + 1) + with db.transaction(): + assert db._transaction_stack_mark() == (True, base + 2) + db.command("sql", "INSERT INTO R SET k = 1") + assert db._transaction_stack_mark() == (True, base + 1) + assert db._transaction_stack_mark() == (False, base) + assert _count(db) == 1 + + def test_a_refused_attempt_is_rolled_back_before_the_retry(db): attempts = [] @@ -193,11 +317,11 @@ def test_begin_commit_rollback_return_nothing(db): assert db.commit() is None -# -- decodes: a status the caller does not read is not a JSON decode +# -- decodes: a status the caller does not read is not a JSON decode, and a state the caller reads is not one either class _Decodes: - """Counts every ``json.loads`` while it is installed (the transaction replies are decoded with it).""" + """Counts every ``json.loads`` while it is installed (an error reply is decoded with it, nothing else is).""" def __init__(self, monkeypatch): self.n = 0 @@ -232,40 +356,99 @@ def test_is_transaction_active_decodes_nothing_and_reports_the_state(db, decodes assert decodes.n == 0 -def test_with_transaction_decodes_only_the_mark(db, decodes): - # the mark reads two fields (active, depth) and is decoded; begin and commit read none +def test_with_transaction_decodes_nothing(db, decodes): + # the mark reads two fields (active, depth) out of the integer, begin and commit read none: no JSON anywhere with db.transaction(): pass - assert decodes.n == 1 + assert decodes.n == 0 -def test_a_state_reply_still_carries_depth_and_commits(db): +def test_a_state_still_carries_active_depth_and_commits(db): db.begin() db.begin() assert db._transaction_stack_mark() == (True, 2) db.rollback() db.rollback() assert db._transaction_stack_mark()[0] is False - assert db._begin_state()["depth"] == 1 - assert db._rollback_state()["active"] is False + active, depth, _ = _fields(db._begin_state()) + assert (active, depth) == (True, 1) + active, _, _ = _fields(db._rollback_state()) + assert active is False + + +def test_the_state_comes_back_as_one_integer(db): + handle = db._h + idle = lib.tx(handle, 3) + assert type(idle) is int and idle >= 0 + active, base, _ = _fields(idle) + assert active is False + one = lib.tx(handle, 0) + assert type(one) is int and one >= 0 + assert _fields(one)[:2] == (True, base + 1) + two = lib.tx(handle, 0) + assert _fields(two)[:2] == (True, base + 2) + assert _fields(two)[2] == _fields(lib.tx(handle, 3))[2] # a mark reads the count + assert _fields(lib.tx(handle, 2))[:2] == (True, base + 1) + assert _fields(lib.tx(handle, 2))[:2] == (False, base) + + +# -- errors: the same typed exception with the same message + + +def test_a_failed_call_returns_minus_one_and_the_error_is_in_last_error(db): + assert lib.tx(2**62, 3) == -1 # a handle no database has + error = json.loads(lib.last_error()) + assert error["ok"] is False + assert error["cls"] == "java.lang.IllegalArgumentException" + assert "unknown database handle" in error["error"] + assert error["chain"][0][0] == error["cls"] + + +def test_a_commit_without_begin_raises_the_engines_error_unchanged(db): + with pytest.raises(ArcadeDBError) as caught: + db.commit() + native = caught.value.native + assert native["ok"] is False + assert native["cls"] and "Transaction not begun" in native["error"] + assert ( + str(caught.value) + == f"Failed to commit transaction: {native['cls']}: {native['error']}" + ) + assert caught.value.java_chain == native["chain"] + assert caught.value.java_chain[0] == [native["cls"], native["error"]] + assert db.is_transaction_active() is False -def test_an_error_reply_is_decoded_into_the_same_error(db, monkeypatch): +def test_a_rollback_with_nothing_to_roll_back_is_not_an_error(db): + db.rollback() + assert db.is_transaction_active() is False + + +def test_a_failed_statement_in_a_block_rolls_the_block_back_and_leaves_no_level(db): + db.command("sql", "CREATE PROPERTY R.k INTEGER") + db.command("sql", "CREATE INDEX ON R (k) UNIQUE") + base = db._transaction_stack_mark() + with pytest.raises(ArcadeDBError) as caught: + with db.transaction(): + db.command("sql", "INSERT INTO R (k) VALUES (1), (1)") + assert "DuplicatedKeyException" in str(caught.value) + assert db._transaction_stack_mark() == base + assert _count(db) == 0 + + +def test_a_failure_reply_is_read_from_last_error_into_the_same_error(db, monkeypatch): reply = ( '{"ok":false,"error":"boom","cls":"x.Y","chain":[["x.Y","boom"]],' '"retry":"need_retry","retry_ms":3}' ) - real = lib.call - monkeypatch.setattr( - lib, - "call", - lambda name, *a: reply if name == "arcadedb_n_tx" else real(name, *a), - ) + monkeypatch.setattr(lib, "tx", lambda handle, op: -1) + monkeypatch.setattr(lib, "last_error", lambda: reply) for op, prefix in ( (db.begin, "Failed to begin transaction"), (db.commit, "Failed to commit transaction"), (db.rollback, "Failed to rollback transaction"), (db.is_transaction_active, "Failed to check transaction status"), + (db._transaction_stack_mark, "Failed to read the transaction state"), ): with pytest.raises(ArcadeDBError) as caught: op() @@ -274,12 +457,28 @@ def test_an_error_reply_is_decoded_into_the_same_error(db, monkeypatch): assert caught.value.java_chain == [["x.Y", "boom"]] -def test_a_differently_spaced_ok_reply_is_still_read(db, monkeypatch): - monkeypatch.setattr( - lib, - "call", - lambda name, *a: '{"ok": true, "active": true, "depth": 1, "commits": 0}', - ) - assert db.is_transaction_active() is True - db.begin() - db.commit() +# -- a closed database + + +def test_a_transaction_left_open_is_gone_after_a_close_and_reopen(temp_db_path): + database = arcadedb.create_database(temp_db_path) + database.command("sql", "CREATE DOCUMENT TYPE R") + database.begin() + database.command("sql", "INSERT INTO R SET k = 1") + assert database.is_transaction_active() is True + database.close() + for call in ( + database.begin, + database.commit, + database.rollback, + database.is_transaction_active, + database._transaction_stack_mark, + ): + with pytest.raises(ArcadeDBError, match="Database is closed"): + call() + reopened = arcadedb.open_database(temp_db_path) + try: + assert reopened.is_transaction_active() is False + assert _count(reopened) == 0 + finally: + reopened.drop() From 452e93c4178311fbf0eccb3301b5e1e099ee0e7c Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 14 Nov 2026 03:00:20 +0900 Subject: [PATCH 06/11] Native transactions: arcadedb_n_tx returns the state as one integer, not a JSON text (refs #315) A begin, a commit, a rollback and a mark each crossed into the library and got back a malloc'd JSON text of about 60 bytes that Python read with strlen, a copy into a str and arcadedb_free, and mostly never looked at. with db.transaction() is three crossings (the mark, begin, commit), so twelve foreign calls and three strings built, copied and freed. arcadedb_n_tx now returns a long: bit 0 active, bits 1 to 31 depth, bits 32 to 62 the low 31 bits of the commit count, -1 for an error. It is the way arcadedb_n_cmd returns a cursor id: the error text is kept per thread and read with arcadedb_n_last_error, built by the same errorJson as before, so the typed exception and its message do not change. Each crossing is one foreign call, and with db.transaction() is three. The Python side reads the fields of the integer inline, so the mark, the begin and rollback states the unwind and run_in_transaction read, and is_transaction_active no longer decode anything. Not measured: the speed claim waits for a timed A/B of a candidate wheel against 26.11.1.post2. Co-Authored-By: Claude Sonnet 5.5 --- .../org/humemai/nativelib/ArcadeNative.java | 24 ++++-- src/arcadedb/_native/database.py | 80 ++++++++----------- src/arcadedb/_native/lib.py | 17 +++- 3 files changed, 66 insertions(+), 55 deletions(-) diff --git a/native/src/main/java/org/humemai/nativelib/ArcadeNative.java b/native/src/main/java/org/humemai/nativelib/ArcadeNative.java index 5ccd0c3..2ec6c1f 100644 --- a/native/src/main/java/org/humemai/nativelib/ArcadeNative.java +++ b/native/src/main/java/org/humemai/nativelib/ArcadeNative.java @@ -1,9 +1,10 @@ /* * SPDX-License-Identifier: Apache-2.0 * - * Phase 2: the C entry points behind the Python native backend (arcadedb/_native). Every function returns a + * Phase 2: the C entry points behind the Python native backend (arcadedb/_native). Most functions return a * malloc'd UTF-8 JSON text that the caller releases with arcadedb_free: {"ok":true,...} or * {"ok":false,"error":msg,"cls":java class,"chain":[[class,msg],..],"retry":"need_retry"|"duplicated_key"|null,"retry_ms":n}. + * arcadedb_n_cmd and arcadedb_n_tx return a long (-1 for an error, whose text in that form arcadedb_n_last_error returns). * Databases, result sets and servers are long handles. Any thread attached to the isolate may call any function; a * transaction belongs to the (Java) thread that began it, as in the embedded API. */ @@ -184,9 +185,15 @@ public static CCharPointer info(final IsolateThread thread, final long h) { } } - /** op 0 begin, 1 commit, 2 rollback, 3 mark. Every reply carries {"active":b,"depth":n,"commits":n}. */ + /** + * op 0 begin, 1 commit, 2 rollback, 3 mark. Returns the transaction state after the call as one non-negative long: bit 0 is + * isTransactionActive, bits 1 to 31 the nesting depth, bits 32 to 62 the low 31 bits of the commit count of the thread's + * transaction context (callers compare two counts for equality). -1 on error: the error text is then read with + * arcadedb_n_last_error on the same thread, as for arcadedb_n_cmd. No reply text, so no malloc, no arcadedb_free and no JSON + * decode in the caller. + */ @CEntryPoint(name = "arcadedb_n_tx") - public static CCharPointer tx(final IsolateThread thread, final long h, final int op) { + public static long tx(final IsolateThread thread, final long h, final int op) { try { final Database db = db(h); switch (op) { @@ -196,14 +203,15 @@ public static CCharPointer tx(final IsolateThread thread, final long h, final in default -> { } } - return out(txState(db)); + return txState(db); } catch (final Throwable t) { - return err(t); + LAST_ERROR.set(errorJson(t)); + return -1; } } - private static String txState(final Database db) { - int depth = 0; + private static long txState(final Database db) { + long depth = 0; long commits = 0; if (db instanceof LocalDatabase local) depth = local.getNestedTransactions(); @@ -212,7 +220,7 @@ private static String txState(final Database db) { if (ctx != null) commits = ctx.getCommitCount(); } - return "{\"ok\":true,\"active\":" + db.isTransactionActive() + ",\"depth\":" + depth + ",\"commits\":" + commits + "}"; + return (db.isTransactionActive() ? 1L : 0L) | ((depth & 0x7FFFFFFFL) << 1) | ((commits & 0x7FFFFFFFL) << 32); } // ---- statements diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index c02ac66..fc9ae74 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -74,12 +74,10 @@ def _call(prefix: str, name: str, *args) -> dict: return reply -# Every successful ``arcadedb_n_tx`` reply is ``{"ok":true,"active":,"depth":,"commits":}`` (``ArcadeNative.txState``). A caller -# that reads none of the state (begin, commit, rollback), or only ``active``, tells success and that bool from the text it already holds -# instead of decoding a JSON object. An error reply is ``{"ok":false,...}``: it fails the prefix test and is decoded, as before. -_TX_OK = '{"ok":true,' -_TX_ACTIVE = '{"ok":true,"active":true,' -_TX_IDLE = '{"ok":true,"active":false,' +# ``arcadedb_n_tx`` answers with the transaction state after the call as one int (``ArcadeNative.tx``, no reply text to copy, free or +# decode): bit 0 is ``active``, bits 1 to 31 the nesting ``depth``, bits 32 to 62 the ``commits`` of the transaction context (its low 31 +# bits: callers compare two of them for equality). Callers read the fields inline: ``state & 1``, ``(state >> 1) & 0x7FFFFFFF``, +# ``state >> 32``. -1 is an error: its text, the ``{"ok":false,...}`` reply of every other entry point, is ``lib.last_error()``. def _first_row(database, reply: dict): @@ -793,73 +791,63 @@ def _command_first_row(self, language: str, command: str, *args) -> NativeResult # -- transactions - def _tx_text(self, op: int, prefix: str) -> str: - """One ``arcadedb_n_tx`` crossing: the reply text of a success (not decoded), an ArcadeDBError for a failure.""" + def _tx_state(self, op: int, prefix: str) -> int: + """One ``arcadedb_n_tx`` crossing, one foreign call: the transaction state after the operation as an int (its layout is + described at the top of this module), an ArcadeDBError for a failure. + """ if self._closed: raise ArcadeDBError("Database is closed") if (op == 1 or op == 2) and lib.CLOSE_BATCH: lib.flush_closes() - text = lib.call("arcadedb_n_tx", self._h, op) - if text.startswith(_TX_OK): - return text - reply = json.loads(text) - if not reply.get("ok"): - raise _error(prefix, reply) - return text - - def _tx(self, op: int, prefix: str) -> dict: - """The decoded state reply: for a caller that reads ``depth`` or ``commits``.""" - return json.loads(self._tx_text(op, prefix)) + state = lib.tx(self._h, op) + if state < 0: + raise _error(prefix, json.loads(lib.last_error())) + return state def begin(self): - self._tx_text(0, "Failed to begin transaction") + self._tx_state(0, "Failed to begin transaction") - def _begin_state(self) -> dict: - """begin(), and the state right after it: the reply of ``arcadedb_n_tx`` carries ``active``, ``depth`` and ``commits``.""" - return self._tx(0, "Failed to begin transaction") + def _begin_state(self) -> int: + """begin(), and the state right after it: the int of ``arcadedb_n_tx`` carries ``active``, ``depth`` and ``commits``.""" + return self._tx_state(0, "Failed to begin transaction") def commit(self): - self._tx_text(1, "Failed to commit transaction") + self._tx_state(1, "Failed to commit transaction") def rollback(self): - self._tx_text(2, "Failed to rollback transaction") + self._tx_state(2, "Failed to rollback transaction") - def _rollback_state(self) -> dict: - """rollback(), and the state right after it (the reply carries it, as begin's does): the unwind reads it instead of asking.""" - return self._tx(2, "Failed to rollback transaction") + def _rollback_state(self) -> int: + """rollback(), and the state right after it (the int carries it, as begin's does): the unwind reads it instead of asking.""" + return self._tx_state(2, "Failed to rollback transaction") def transaction(self) -> TransactionContext: return TransactionContext(self) def is_transaction_active(self) -> bool: - text = self._tx_text(3, "Failed to check transaction status") - if text.startswith(_TX_ACTIVE): - return True - if text.startswith(_TX_IDLE): - return False - return json.loads(text)["active"] + return bool(self._tx_state(3, "Failed to check transaction status") & 1) def _transaction_stack_mark(self): - state = self._tx(3, "Failed to read the transaction state") - return state["active"], state["depth"] + state = self._tx_state(3, "Failed to read the transaction state") + return bool(state & 1), (state >> 1) & 0x7FFFFFFF def _transaction_commit_mark(self): """The commit count of the current transaction (the witness that a BATCH boundary published part of its work).""" - return self._tx(3, "Failed to read the transaction state")["commits"] + return self._tx_state(3, "Failed to read the transaction state") >> 32 def _unwind_opened_since(self, mark): """Roll back every level opened since ``mark``, and no other; returns ``(state, first_error)``. - ``state`` is the transaction state after the last crossing (every rollback reply carries it, so no read sits between two + ``state`` is the transaction state int after the last crossing (every rollback carries it, so no read sits between two steps), or None when that crossing failed. A level the caller held at the mark is theirs, as in the JPype ``Database``. """ was_active, depth_before = mark first_error = None - state = self._tx(3, "Failed to read the transaction state") - for _ in range(state["depth"] - depth_before): + state = self._tx_state(3, "Failed to read the transaction state") + for _ in range(((state >> 1) & 0x7FFFFFFF) - depth_before): if state is None: - state = self._tx(3, "Failed to check transaction status") - if not state["active"]: + state = self._tx_state(3, "Failed to check transaction status") + if not state & 1: break try: state = self._rollback_state() @@ -868,8 +856,8 @@ def _unwind_opened_since(self, mark): first_error = first_error or e if not was_active: if state is None: - state = self._tx(3, "Failed to check transaction status") - if state["active"]: + state = self._tx_state(3, "Failed to check transaction status") + if state & 1: try: state = self._rollback_state() except Exception as e: @@ -893,7 +881,7 @@ def run_in_transaction(self, fn, retries: int = 12, backoff_s: float = 0.005): mark = self._transaction_stack_mark() commits_at_start = None try: - commits_at_start = self._begin_state()["commits"] + commits_at_start = self._begin_state() >> 32 result = fn() self.commit() return result @@ -919,7 +907,7 @@ def run_in_transaction(self, fn, retries: int = 12, backoff_s: float = 0.005): # retry, as before); the raise sits outside that try, or its own except would swallow it. try: commits_now = ( - state["commits"] + state >> 32 if state is not None else self._transaction_commit_mark() ) diff --git a/src/arcadedb/_native/lib.py b/src/arcadedb/_native/lib.py index 394bd32..ebff0d4 100644 --- a/src/arcadedb/_native/lib.py +++ b/src/arcadedb/_native/lib.py @@ -259,6 +259,9 @@ class _Params(ctypes.Structure): "arcadedb_n_set_upcall": [P, P], "arcadedb_free": [P, P], } + # the entry points that answer with a number, not with the address of a reply text: a cursor id (arcadedb_n_cmd) and the + # transaction state (arcadedb_n_tx); -1 is an error for both, its text is read with arcadedb_n_last_error + returns_long = ("arcadedb_n_cmd", "arcadedb_n_tx") for name, argtypes in sigs.items(): fn = getattr(lib, name, None) if fn is None: # a library built before this entry point existed @@ -272,7 +275,7 @@ class _Params(ctypes.Structure): "arcadedb_n_rs_close_many", "arcadedb_n_set_upcall", ) - else ctypes.c_longlong if name == "arcadedb_n_cmd" else ctypes.c_void_p + else ctypes.c_longlong if name in returns_long else ctypes.c_void_p ) global BINARY_PARAMS, CMD_PATH, CLOSE_BATCH, INSERT_ROWS, EXEC_MANY, NEXT_BIN, GB_BIN, TS_BIN if not hasattr( @@ -488,7 +491,19 @@ def cmd(handle, flags, language, command, params, buf, count, first) -> int: ) +def tx(handle, op) -> int: + """``arcadedb_n_tx``: op 0 begin, 1 commit, 2 rollback, 3 mark. Returns the transaction state after the call as one int (bit 0 + ``active``, bits 1 to 31 ``depth``, bits 32 to 62 ``commits``, as ``ArcadeNative.tx`` writes it), or -1 for an error (read it + with :func:`last_error`). One foreign call: no reply text to copy, free or decode. + """ + lib = _lib or _load() + t = _local.__dict__.get("thread") + th = t.handle if t is not None else _thread_handle() + return lib.arcadedb_n_tx(th, handle, op) + + def last_error() -> str: + """The error of the last failed :func:`cmd` or :func:`tx` on this thread, as the JSON text of an error reply.""" return call("arcadedb_n_last_error") From 3db17be8b4b0f1b0281b5fe910a1ef5e02b287ee Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 14 Nov 2026 02:59:46 +0900 Subject: [PATCH 07/11] Test: save() of a saved record with auto-transaction on spends two crossings, not three (refs #317) The transaction check of a saved record's save() cannot refuse while auto-transaction is on, yet it still crossed into the library (arcadedb_n_tx) before the SELECT and the UPDATE. These tests pin two crossings there, bare and inside a transaction, with and without the read-back of the row, and pin what must not move: the check is back the moment auto-transaction is off (first, before any read), a stale record is refused under auto-transaction too, and the stored values and the next save's baseline are those of an update inside a transaction for NULL, float, text, date-time, date, Decimal, nested values, a float32 array and a value set unchanged. The two-crossing tests fail on the code before the change (three crossings); the rest pass on both. Co-Authored-By: Claude Sonnet 5.5 --- tests/test_native_one_row_crossings.py | 109 +++++++++++++++++++++++++ 1 file changed, 109 insertions(+) diff --git a/tests/test_native_one_row_crossings.py b/tests/test_native_one_row_crossings.py index e65e3e9..d24cde4 100644 --- a/tests/test_native_one_row_crossings.py +++ b/tests/test_native_one_row_crossings.py @@ -7,6 +7,11 @@ reads the stored record first: the engine's reply to an UPDATE comes after the write, and the check must refuse before writing anything inside the caller's transaction, so it cannot be read from the update's own result. +With auto-transaction on, the transaction check cannot refuse (the engine opens a transaction for a bare write), so ``save()`` does not +ask for the state: two crossings, the SELECT and the UPDATE. With it off (the engine default) the check still comes first, outside a +transaction and inside one. The engine runs the before-update callbacks and the property validation before it looks for a transaction, +and an UPDATE of a record that is gone raises nothing, so the engine's own refusal cannot stand in for the check. + The counts are crossings (the ctypes doors ``cmd``, ``call``, ``call_bin``, ``exec_query``, ``tx`` and the batched cursor close), not seconds: deterministic. The oracle for the values and the errors is the lazy form itself (``db.command(...).first()``), kept for every other caller. """ @@ -192,6 +197,110 @@ def test_an_update_that_reads_its_row_back_is_three_crossings(db, crossings): assert db.lookup_by_rid(rid).get("brand_new") == "value" +@pytest.mark.parametrize("in_transaction", [False, True], ids=["bare", "in_tx"]) +def test_updating_a_saved_record_with_auto_transaction_is_two_crossings( + db, crossings, in_transaction +): + rid = _saved(db, a=1, name="x") + db.set_auto_transaction(True) + record = db.lookup_by_rid(rid).modify() + record.set("a", 2) + with contextlib.ExitStack() as stack: + if in_transaction: + stack.enter_context(db.transaction()) + crossings.reset() + record.save() + # the SELECT of the stale-record check and the UPDATE: the transaction check cannot refuse, so it is not asked (was three) + assert crossings.names == ["exec_query", "exec_query"] + assert db.lookup_by_rid(rid).get("a") == 2 + + +@pytest.mark.parametrize("in_transaction", [False, True], ids=["bare", "in_tx"]) +def test_an_update_that_reads_its_row_back_with_auto_transaction_is_two_crossings( + db, crossings, in_transaction +): + rid = _saved(db, a=1) + db.set_auto_transaction(True) + record = db.lookup_by_rid(rid).modify() + record.set("brand_new", "value") # a new property is read back (RETURN AFTER) + with contextlib.ExitStack() as stack: + if in_transaction: + stack.enter_context(db.transaction()) + crossings.reset() + record.save() + assert crossings.names == ["exec_query", "exec_query"] + assert db.lookup_by_rid(rid).get("brand_new") == "value" + + +def test_with_auto_transaction_off_again_the_transaction_check_comes_back( + db, crossings +): + rid = _saved(db, a=1) + db.set_auto_transaction(True) + db.set_auto_transaction(False) + record = db.lookup_by_rid(rid).modify() + record.set("a", 2) + crossings.reset() + with pytest.raises(ArcadeDBError, match="Transaction not begun"): + record.save() + assert crossings.names == ["arcadedb_n_tx"] # refused before it reads anything + with db.transaction(): + crossings.reset() + record.save() + assert crossings.names == ["arcadedb_n_tx", "exec_query", "exec_query"] + assert db.lookup_by_rid(rid).get("a") == 2 + + +def test_a_stale_record_is_refused_under_auto_transaction_too(db): + rid = _saved(db, a=1, b=1) + mine = db.lookup_by_rid(rid).modify() + with db.transaction(): + bump_sql = f"UPDATE {rid} SET b = 9" # nosec B608 - a record id this test read + db.command("sql", bump_sql).close() + db.set_auto_transaction(True) + mine.set("a", 2) + with pytest.raises(ArcadeDBError, match="ConcurrentModificationException"): + mine.save() + stored = db.lookup_by_rid(rid) + assert (stored.get("a"), stored.get("b")) == (1, 9) # nothing of mine was written + + +UPDATE_CASES = { + "null": None, + "float": 2.5, + "text": "new", + "date and time": dt.datetime(2026, 11, 9, 12, 30, 15, 123000), + "date": dt.date(2026, 11, 9), + "decimal": Decimal("1.50"), + "list and map": {"l": [1, 2, 3], "m": {"k": [1.5, "v"]}}, + "float array": np.arange(16, dtype=np.float32), + "as read": "old", # the value the record already holds: set() of an unchanged value +} + + +@pytest.mark.parametrize("case", list(UPDATE_CASES)) +def test_an_update_under_auto_transaction_stores_what_one_in_a_transaction_stores( + db, case +): + value = UPDATE_CASES[case] + inside, bare = _saved(db, k="old", n=0), _saved(db, k="old", n=0) + record = db.lookup_by_rid(inside).modify() + record.set("k", value) + with db.transaction(): + record.save() + db.set_auto_transaction(True) + held = db.lookup_by_rid(bare).modify() + held.set("k", value) + held.save() + assert _plain(db.lookup_by_rid(bare).to_dict()) == _plain( + db.lookup_by_rid(inside).to_dict() + ) + # the copy the save left is the stored row: the next save of the same object is no conflict + held.set("n", 1) + held.save() + assert db.lookup_by_rid(bare).get("n") == 1 + + def test_a_stale_record_is_still_refused_before_anything_is_written(db): rid = _saved(db, a=1, b=1) mine = db.lookup_by_rid(rid).modify() From e8e0c450f99c9fe88a892686710a2aef4a0c8d38 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 14 Nov 2026 03:00:30 +0900 Subject: [PATCH 08/11] Native: save() of a saved record does not ask for the transaction state when auto-transaction is on (refs #317) _save_update asked is_transaction_active() (an arcadedb_n_tx crossing) first and only then looked at the auto-transaction flag, so the answer was thrown away whenever the flag was on. The operands are swapped: the flag is a Python attribute, the state is a crossing. With auto-transaction on, an update of a saved record is two crossings (the SELECT of the stale-record check and the UPDATE), not three. With it off nothing changes: the check still comes first, outside a transaction and inside one. The check is not replaced by the engine's refusal for the default mode: the engine runs the before-update callbacks and the property validation before it looks for a transaction (read in LocalDatabase.updateRecord of the 26.12.1-SNAPSHOT source), and an UPDATE of a record that is gone raises nothing, so a save outside a transaction would run callbacks, or pass silently, instead of raising "Transaction not begun" before it reads. A Python-side record of the transaction state is not exact either: a SQL COMMIT or ROLLBACK statement ends the transaction without passing the wheel's transaction calls. Co-Authored-By: Claude Sonnet 5.5 --- src/arcadedb/_native/records.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/arcadedb/_native/records.py b/src/arcadedb/_native/records.py index c42d973..9bb732d 100644 --- a/src/arcadedb/_native/records.py +++ b/src/arcadedb/_native/records.py @@ -359,8 +359,11 @@ def _save_update(self, db) -> None: Like every write (and like the JVM build), it needs an active transaction while auto-transaction is off, the engine default: outside one it raises ArcadeDBError ("Transaction not begun") before it reads or writes anything. This method opens no - transaction of its own.""" - if not db.is_transaction_active() and not db._auto_transaction: + transaction of its own. With auto-transaction on the check cannot refuse, so the state is not asked for (one crossing less). + The engine's own refusal cannot stand in for the check with it off: the engine runs the before-update callbacks and the + validation before it looks for a transaction, and an UPDATE of a record that is gone raises nothing. + """ + if not db._auto_transaction and not db.is_transaction_active(): raise ArcadeDBError( "Failed to save record: com.arcadedb.exception.TransactionException: Transaction not begun " "(wrap the write in `with db.transaction():` or call begin() first)" From 55b427a51d2e0b7a6c8c4a95c97c06ded38fb10d Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 14 Nov 2026 03:54:08 +0900 Subject: [PATCH 09/11] Tests: the nesting depth the native mark reports is the engine's, and no failed assert leaves a transaction open (refs #315) CI of the first push (run 43478436168) failed two tests on a wrong assumption of mine: I expected the outer begin to raise the depth by one. LocalDatabase.getNestedTransactions() is the size of the thread's stack of transaction contexts (DatabaseContext.DatabaseContextTL.transactions). The stack always holds the thread's base context once the thread has used the database, so the depth is 1 while idle. The outer begin reuses that base context (depth 1, active). Only a begin while a transaction is active pushes a context (2), the commit or rollback of that level pops it (1, still active), and the outermost end leaves the base context on the stack (1, idle). popIfNotLastTransaction never removes the last one. The native tx entry point returns exactly that number, as the old JSON reply did, so the change is right and the two expectations were not. The fixture now rolls back whatever a test left open before it drops the database, so a failed assert is one failure and not also a "Cannot drop the database in transaction" teardown error. The close-and-reopen test does the same in a finally. Co-Authored-By: Claude Sonnet 5.5 --- tests/test_native_tx_crossings.py | 78 ++++++++++++++++++++----------- 1 file changed, 52 insertions(+), 26 deletions(-) diff --git a/tests/test_native_tx_crossings.py b/tests/test_native_tx_crossings.py index a218d61..78a1635 100644 --- a/tests/test_native_tx_crossings.py +++ b/tests/test_native_tx_crossings.py @@ -15,6 +15,7 @@ scripts/check_test_skips.py); it runs in the native parity run. """ +import contextlib import json import pytest @@ -113,8 +114,18 @@ def db(temp_db_path): ): # the thread's transaction context exists from here on, as it does in a running application pass lib.flush_closes() # a cursor closed earlier is not part of a call measured below - yield database - database.drop() + try: + yield database + finally: + # a test that fails with a transaction open (a begin without its end, a level a direct lib.tx call pushed) must not turn into + # a teardown error as well: the engine refuses to drop a database that has a transaction open on the thread + try: + for _ in range(16): + if not database.is_transaction_active(): + break + database.rollback() + finally: + database.drop() @pytest.fixture @@ -255,14 +266,19 @@ def test_a_nested_block_that_succeeds_commits_with_the_caller(db): assert _count(db) == 2 -def test_nested_with_blocks_push_and_pop_one_level_each(db): +def test_nested_with_blocks_push_one_level_for_the_inner_block_only(db): + # The depth is the size of the thread's stack of transaction contexts (LocalDatabase.getNestedTransactions). The stack always holds + # the thread's base context once the thread has used the database (the fixture did), so it is 1 while idle. The outer begin + # reuses that base context (depth stays 1, active); only a begin while a transaction is active pushes a context (2), and the commit + # of that inner level pops it (1, still active). The outer commit ends the base context and leaves it on the stack (1, idle). base = db._transaction_stack_mark()[1] + assert db._transaction_stack_mark() == (False, 1) with db.transaction(): - assert db._transaction_stack_mark() == (True, base + 1) + assert db._transaction_stack_mark() == (True, base) with db.transaction(): - assert db._transaction_stack_mark() == (True, base + 2) + assert db._transaction_stack_mark() == (True, base + 1) db.command("sql", "INSERT INTO R SET k = 1") - assert db._transaction_stack_mark() == (True, base + 1) + assert db._transaction_stack_mark() == (True, base) assert db._transaction_stack_mark() == (False, base) assert _count(db) == 1 @@ -366,6 +382,7 @@ def test_with_transaction_decodes_nothing(db, decodes): def test_a_state_still_carries_active_depth_and_commits(db): db.begin() db.begin() + # the first begin reuses the thread's base context (depth 1), the second pushes one (2) assert db._transaction_stack_mark() == (True, 2) db.rollback() db.rollback() @@ -382,14 +399,15 @@ def test_the_state_comes_back_as_one_integer(db): assert type(idle) is int and idle >= 0 active, base, _ = _fields(idle) assert active is False + # the first begin reuses the thread's base context (the depth stays), a begin inside it pushes one more (see the nested with test) one = lib.tx(handle, 0) assert type(one) is int and one >= 0 - assert _fields(one)[:2] == (True, base + 1) + assert _fields(one)[:2] == (True, base) two = lib.tx(handle, 0) - assert _fields(two)[:2] == (True, base + 2) + assert _fields(two)[:2] == (True, base + 1) assert _fields(two)[2] == _fields(lib.tx(handle, 3))[2] # a mark reads the count - assert _fields(lib.tx(handle, 2))[:2] == (True, base + 1) - assert _fields(lib.tx(handle, 2))[:2] == (False, base) + assert _fields(lib.tx(handle, 2))[:2] == (True, base) # the inner level is gone + assert _fields(lib.tx(handle, 2))[:2] == (False, base) # the outer one ended # -- errors: the same typed exception with the same message @@ -462,23 +480,31 @@ def test_a_failure_reply_is_read_from_last_error_into_the_same_error(db, monkeyp def test_a_transaction_left_open_is_gone_after_a_close_and_reopen(temp_db_path): database = arcadedb.create_database(temp_db_path) - database.command("sql", "CREATE DOCUMENT TYPE R") - database.begin() - database.command("sql", "INSERT INTO R SET k = 1") - assert database.is_transaction_active() is True - database.close() - for call in ( - database.begin, - database.commit, - database.rollback, - database.is_transaction_active, - database._transaction_stack_mark, - ): - with pytest.raises(ArcadeDBError, match="Database is closed"): - call() - reopened = arcadedb.open_database(temp_db_path) + reopened = None try: + database.command("sql", "CREATE DOCUMENT TYPE R") + database.begin() + database.command("sql", "INSERT INTO R SET k = 1") + assert database.is_transaction_active() is True + database.close() + for call in ( + database.begin, + database.commit, + database.rollback, + database.is_transaction_active, + database._transaction_stack_mark, + ): + with pytest.raises(ArcadeDBError, match="Database is closed"): + call() + reopened = arcadedb.open_database(temp_db_path) assert reopened.is_transaction_active() is False assert _count(reopened) == 0 finally: - reopened.drop() + if reopened is not None: + reopened.drop() + else: + # a failure before the reopen: end what is open and drop, ignoring the refusals of a database already closed + with contextlib.suppress(ArcadeDBError): + database.rollback() + with contextlib.suppress(ArcadeDBError): + database.drop() From 61c200d93b6774ecc4a52d8f752ecd6a8bdba4c4 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 14 Nov 2026 15:49:35 +0900 Subject: [PATCH 10/11] Tests first: get_rid, get_vertex, get_edge and get_element of a row and get_rid of a record are one call each on the JVM build (refs #388) The tests count the calls on a proxy of the Java row or record and of ReaderCalls: the parent makes four calls for Result.get_rid, three for each element getter and two for Document.get_rid, and none on ReaderCalls. A second set compares every answer with the per-call path for a vertex, an edge, a document, a projection and an aggregate. Co-Authored-By: Claude Sonnet 5.5 --- tests/conftest.py | 1 + tests/test_reader_identity_crossings.py | 156 ++++++++++++++++++++++++ 2 files changed, 157 insertions(+) create mode 100644 tests/test_reader_identity_crossings.py diff --git a/tests/conftest.py b/tests/conftest.py index d59db09..90d8379 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -154,6 +154,7 @@ "test_async_proxy_cache.py": "the JPype proxy classes of the JVM build's async executor", "test_read_crossings.py": "crossing counts of the JVM build's read path (JPype proxies of the Java objects and the bridge classes)", "test_reader_crossings.py": "crossing counts of the JVM build's record and row readers (JPype proxies of the Java objects and the bridge classes)", + "test_reader_identity_crossings.py": "crossing counts of get_rid, get_vertex, get_edge and get_element on the JVM build (JPype proxies of the Java row and the bridge class)", "test_typed_batch_guard.py": "the object_hook and the side array of the JVM build's TypedRows batches (JPype arrays, the bridge class, a spy on json.loads)", "test_jvm_ingest_integrity_faults.py": "rows-dropped faults put into the JVM build's column conversion and bulk vertex path (core._column_to_java, GraphBatch._create_vertices_json_bulk)", } diff --git a/tests/test_reader_identity_crossings.py b/tests/test_reader_identity_crossings.py new file mode 100644 index 0000000..1ad23bc --- /dev/null +++ b/tests/test_reader_identity_crossings.py @@ -0,0 +1,156 @@ +"""Crossings of ``Result.get_rid()``, ``get_vertex()``, ``get_edge()``, ``get_element()`` and ``Document.get_rid()`` on the JVM build. + +``Result.get_rid()`` made four JPype calls (``getIdentity``, ``isPresent``, ``get``, ``toString``), each of the three element getters made three +(``getVertex()``, ``isPresent()``, ``get()``), and ``Document.get_rid()`` two. ``ReaderCalls`` now answers each in one call. Every test here pins two +things: the calls into the JVM (counted on a proxy of the Java row or record and of the bridge class, the technique of ``test_reader_crossings.py``; +a call on the Java row itself would show up in the log) and that the answer is the one the per-call path gives (the oracle is that path, written out +below), so a count cannot be bought with a change of behaviour. +""" + +import pytest + +from arcadedb import results +from arcadedb.graph import Document, Edge, Vertex + + +class Counting: + """Counts the method calls made on a Java object or a Java class; a ``Counting`` argument is unwrapped, return values are the real objects.""" + + def __init__(self, target, log, label): + self._target = target + self._log = log + self._label = label + + def __getattr__(self, name): + attr = getattr(self._target, name) + if not callable(attr): + return attr + + def call(*args): + self._log.append(f"{self._label}.{name}") + args = [a._target if isinstance(a, Counting) else a for a in args] + return attr(*args) + + return call + + +def _old_rid(java_row): + """``Result.get_rid()`` before the bridge call.""" + identity = java_row.getIdentity() + return str(identity.get().toString()) if identity.isPresent() else None + + +def _old_element(java_row, getter): + """``getVertex()``, ``getEdge()`` or ``getElement()`` of a Java row, read as the three calls did.""" + optional = getattr(java_row, getter)() + return optional.get() if optional.isPresent() else None + + +@pytest.fixture +def graph_db(temp_db): + temp_db.command("sql", "CREATE VERTEX TYPE V") + temp_db.command("sql", "CREATE EDGE TYPE E") + temp_db.command("sql", "CREATE DOCUMENT TYPE D") + with temp_db.transaction(): + a = temp_db.command("sql", "CREATE VERTEX V SET name = 'a'").first() + b = temp_db.command("sql", "CREATE VERTEX V SET name = 'b'").first() + temp_db.command( + "sql", + "CREATE EDGE E FROM :a TO :b SET w = 1", + {"a": a.get_rid(), "b": b.get_rid()}, + ).close() + temp_db.command("sql", "INSERT INTO D SET k = 1").close() + return temp_db + + +@pytest.fixture +def log(graph_db, monkeypatch): + """The list every counted call is appended to; the bridge class ``ReaderCalls`` is counted.""" + calls = [] + real = results._bridge_class("ReaderCalls") + assert real is not None, "the bridge jar must carry ReaderCalls" + monkeypatch.setitem( + results._BRIDGE_CLASSES, "ReaderCalls", Counting(real, calls, "ReaderCalls") + ) + return calls + + +# a vertex, an edge, a plain document, a projection that is no record, an aggregate +ROWS = { + "vertex": "SELECT FROM V WHERE name = 'a'", + "edge": "SELECT FROM E", + "document": "SELECT FROM D", + "projection": "SELECT 1 AS x", + "aggregate": "SELECT count(*) AS c FROM V", +} + + +def _row(db, kind): + return db.query("sql", ROWS[kind]).first() + + +@pytest.mark.parametrize("kind", ROWS) +def test_get_rid_is_one_call_and_the_rid_the_per_call_path_gives(graph_db, log, kind): + row = _row(graph_db, kind) + want = _old_rid(row._java_result) + row._java_result = Counting(row._java_result, log, "row") + del log[:] + got = row.get_rid() + assert log == ["ReaderCalls.ridOf"] + assert got == want + assert (got is None) == (kind in ("projection", "aggregate")) + + +@pytest.mark.parametrize("kind", ROWS) +@pytest.mark.parametrize( + "method, getter, bridge, wrapper", + [ + ("get_vertex", "getVertex", "vertexOrNull", Vertex), + ("get_edge", "getEdge", "edgeOrNull", Edge), + ("get_element", "getElement", "elementOrNull", Document), + ], +) +def test_the_element_getters_are_one_call_and_return_what_the_per_call_path_does( + graph_db, log, kind, method, getter, bridge, wrapper +): + row = _row(graph_db, kind) + want = _old_element(row._java_result, getter) + row._java_result = Counting(row._java_result, log, "row") + del log[:] + got = getattr(row, method)() + assert log == [f"ReaderCalls.{bridge}"] + if want is None: + assert got is None + return + assert isinstance(got, wrapper) + assert str(got._java_document.getClass().getName()) == str( + want.getClass().getName() + ) + assert str(got.get_rid()) == str(want.getIdentity()) + + +def test_each_kind_of_row_has_the_element_it_should(graph_db): + vertex = _row(graph_db, "vertex") + assert isinstance(vertex.get_vertex(), Vertex) and vertex.get_edge() is None + edge = _row(graph_db, "edge") + assert isinstance(edge.get_edge(), Edge) and edge.get_vertex() is None + assert isinstance(_row(graph_db, "document").get_element(), Document) + assert _row(graph_db, "projection").get_element() is None + + +def test_document_get_rid_is_one_call_and_the_text_str_of_the_identity_gives( + graph_db, log +): + doc = _row(graph_db, "document").get_element() + want = str(doc._java_document.getIdentity()) + doc._java_document = Counting(doc._java_document, log, "doc") + del log[:] + assert doc.get_rid() == want + assert log == ["ReaderCalls.documentRid"] + + +def test_a_record_with_no_identity_yet_reads_as_it_did(graph_db): + with graph_db.transaction(): + doc = graph_db.new_document("D") + # str() of whatever the engine gives an unsaved record: the same text as before, "None" for no identity at all + assert doc.get_rid() == str(doc._java_document.getIdentity()) From 2c64e8f374301289919b47cc37632739abddd6d7 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sat, 14 Nov 2026 15:49:36 +0900 Subject: [PATCH 11/11] JVM: Result.get_rid, get_vertex, get_edge, get_element and Document.get_rid cross once each, through ReaderCalls (refs #388) ReaderCalls.ridOf, vertexOrNull, edgeOrNull, elementOrNull and documentRid answer in one crossing what took getIdentity, isPresent, get and toString for a row id, getVertex/getEdge/getElement, isPresent and get for an element, and getIdentity plus str() for a record id. A record with no identity yet still reads as the text str(None), as before. The wheel carries its own bridge jar, so the per-call route is gone from these five methods. Co-Authored-By: Claude Sonnet 5.5 --- src/arcadedb/graph.py | 3 +- src/arcadedb/results.py | 24 +++++---------- src/java/com/arcadedb/python/ReaderCalls.java | 30 +++++++++++++++++++ 3 files changed, 40 insertions(+), 17 deletions(-) diff --git a/src/arcadedb/graph.py b/src/arcadedb/graph.py index 7feac61..8f739ea 100644 --- a/src/arcadedb/graph.py +++ b/src/arcadedb/graph.py @@ -220,7 +220,8 @@ def to_dict(self, convert_types: bool = True) -> Dict[str, Any]: def get_rid(self) -> str: """Get Record ID.""" - return str(self._java_document.getIdentity()) + # one crossing; a record with no identity yet reads "None", as str() of the identity did + return str(_reader_calls()[0].documentRid(self._java_document)) def get_java_document(self): """Expose the wrapped Java document for internal integrations.""" diff --git a/src/arcadedb/results.py b/src/arcadedb/results.py index 20c0ee3..daa90e2 100644 --- a/src/arcadedb/results.py +++ b/src/arcadedb/results.py @@ -1072,10 +1072,8 @@ def get_rid(self) -> Optional[str]: >>> print(result.get_rid()) #10:5 """ - identity = self._java_result.getIdentity() - if identity.isPresent(): - return str(identity.get().toString()) - return None + rid = _bridge_class("ReaderCalls").ridOf(self._java_result) + return None if rid is None else str(rid) def get_vertex(self) -> Optional[Vertex]: """ @@ -1085,10 +1083,8 @@ def get_vertex(self) -> Optional[Vertex]: Vertex object or None """ self._check_open() - vertex = self._java_result.getVertex() - if vertex.isPresent(): - return Vertex(vertex.get(), self._database) - return None + vertex = _bridge_class("ReaderCalls").vertexOrNull(self._java_result) + return None if vertex is None else Vertex(vertex, self._database) def get_edge(self) -> Optional[Edge]: """ @@ -1098,10 +1094,8 @@ def get_edge(self) -> Optional[Edge]: Edge object or None """ self._check_open() - edge = self._java_result.getEdge() - if edge.isPresent(): - return Edge(edge.get(), self._database) - return None + edge = _bridge_class("ReaderCalls").edgeOrNull(self._java_result) + return None if edge is None else Edge(edge, self._database) def get_element(self) -> Optional[Document]: """ @@ -1111,10 +1105,8 @@ def get_element(self) -> Optional[Document]: Document, Vertex, or Edge object or None """ self._check_open() - element = self._java_result.getElement() - if element.isPresent(): - return Document.wrap(element.get(), self._database) - return None + element = _bridge_class("ReaderCalls").elementOrNull(self._java_result) + return None if element is None else Document.wrap(element, self._database) def get_property_names(self) -> List[str]: """ diff --git a/src/java/com/arcadedb/python/ReaderCalls.java b/src/java/com/arcadedb/python/ReaderCalls.java index dd93799..af90ea3 100644 --- a/src/java/com/arcadedb/python/ReaderCalls.java +++ b/src/java/com/arcadedb/python/ReaderCalls.java @@ -19,6 +19,7 @@ package com.arcadedb.python; import com.arcadedb.database.Document; +import com.arcadedb.database.RID; import com.arcadedb.query.sql.executor.Result; import com.arcadedb.query.sql.executor.ResultSet; @@ -82,6 +83,35 @@ public static Object[] documentJson(final Document doc) { return new Object[] { sb.toString(), side.toArray() }; } + /** + * The record id of a row as text, or null when the row is not a record: the {@code getIdentity()}, {@code isPresent()}, + * {@code get()} and {@code toString()} of {@code Result.get_rid()} in one crossing. + */ + public static String ridOf(final Result row) { + return row.getIdentity().map(Object::toString).orElse(null); + } + + /** The record id of a record as text, or null when it has none yet: the {@code getIdentity()} and the {@code str()} of {@code Document.get_rid()}. */ + public static String documentRid(final Document doc) { + final RID rid = doc.getIdentity(); + return rid == null ? null : rid.toString(); + } + + /** The vertex of a row, or null when the row is not a vertex: {@code getVertex()}, {@code isPresent()} and {@code get()} in one crossing. */ + public static Object vertexOrNull(final Result row) { + return row.getVertex().orElse(null); + } + + /** The edge of a row, or null when the row is not an edge: {@code getEdge()}, {@code isPresent()} and {@code get()} in one crossing. */ + public static Object edgeOrNull(final Result row) { + return row.getEdge().orElse(null); + } + + /** The record of a row, or null when the row is not a record: {@code getElement()}, {@code isPresent()} and {@code get()} in one crossing. */ + public static Object elementOrNull(final Result row) { + return row.getElement().orElse(null); + } + /** The property names of a row as one JSON array of strings, in the order {@code getPropertyNames()} gives them. */ public static String rowNames(final Result row) { return namesJson(row.getPropertyNames());