diff --git a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc index 3ba0d54..73d4e7b 100644 --- a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc +++ b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc @@ -57,6 +57,8 @@ Because the two packages have different defaults, the same program can keep diff On both packages, `insert_many` keeps rows with `datetime`, `date`, `Decimal`, `bytes`, numpy scalar, or pandas `NA` and `NaT` values on the bulk path. On the `arcadedb` (native) package, a row with a value the bulk text cannot carry (NaN or infinity, an integer beyond 64 bits, a date-time or `Decimal` inside a list or dict, or a `datetime.time`) goes in with its own statement, in the order given. +On the `arcadedb-jvm` package, a chunk of 10,000 rows crosses as typed columns, with no JSON text, when every row is a `dict` with the same keys in the same order and each column holds `None` and values of one type: `int` within 64 bits, `float` (NaN and infinity included), `bool`, `str`, `date`, or a `datetime` without a time zone. The stored values are the same as the JSON text stores: a whole number is an `Integer` when it fits an int and a `Long` beyond, and `None` is a null property. Any other chunk keeps the JSON text, as before: a row that lacks a key or lists the keys in another order, a column that mixes types (an `int` beside a `float`), a `Decimal`, `bytes`, numpy or pandas value, a `datetime` with a time zone, a list, tuple or dict, and a float vector for a property declared `ARRAY_OF_FLOATS`. A chunk is judged as a whole, so one such row sends its 10,000 rows down the JSON route. The `arcadedb` (native) package already sends the rows of one shape through its columnar entry. + == Change the commit size [source,python] 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") 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)" diff --git a/src/arcadedb/core.py b/src/arcadedb/core.py index a050b72..a3dba6d 100644 --- a/src/arcadedb/core.py +++ b/src/arcadedb/core.py @@ -8,7 +8,7 @@ import itertools import operator from collections.abc import Mapping -from datetime import date +from datetime import date, datetime from decimal import Decimal from os import PathLike from typing import Any, Callable, List, Optional @@ -33,6 +33,8 @@ from .results import ResultSet from .transactions import TransactionContext from .type_conversion import ( + _INT64_MAX, + _INT64_MIN, _is_numpy_bool, bulk_row_ok, convert_python_to_java, @@ -241,7 +243,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 +316,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: @@ -331,6 +336,129 @@ def _column_to_java(name, values): return jpype.JArray(jpype.JObject)(converted), len(converted) +# insert_many: the kinds of column a chunk can cross as typed columns (see _typed_chunk_columns). The exact types only: a subclass +# (pandas.Timestamp, numpy.float64) and every other type (Decimal, bytes, list, dict, tuple) keep the JSON text. +_TYPED_KINDS = { + int: "int", + float: "float", + bool: "bool", + str: "str", + date: "date", + datetime: "datetime", +} +_DICT_ONLY = frozenset((dict,)) +_STR_ONLY = frozenset((str,)) +_NO_ZONE = frozenset((None,)) +_EPOCH_ORDINAL = date(1970, 1, 1).toordinal() +_EPOCH_DATE = date(1970, 1, 1) +_EPOCH_DATETIME = datetime(1970, 1, 1) + + +def _typed_kind(values): + """``(kind, nullable)`` of one column of an insert_many chunk when it can cross as a typed column, else None. + + The column holds ``None`` and values of ONE exact type of ``_TYPED_KINDS`` (``"null"`` when there is nothing but ``None``). + A mixed column (an int beside a float, a bool beside an int) would store each value as its own type on the JSON route, which one + typed array cannot do; an int beyond 64 bits, a date-time with a zone and a type the table does not list keep the JSON route. + """ + kinds = set(map(type, values)) + nullable = _NONE_TYPE in kinds + kinds.discard(_NONE_TYPE) + if not kinds: + return "null", True + if len(kinds) != 1: + return None + kind = _TYPED_KINDS.get(next(iter(kinds))) + if kind == "int": + present = [v for v in values if v is not None] if nullable else values + if min(present) < _INT64_MIN or max(present) > _INT64_MAX: + return None + elif kind == "datetime": + present = [v for v in values if v is not None] if nullable else values + if set(map(operator.attrgetter("tzinfo"), present)) != _NO_ZONE: + return None + return (kind, nullable) if kind else None + + +def _typed_java_column(name, values, kind, nullable): + """One column of an insert_many chunk as the Java column ``DocumentBatcher.insertManyColumns`` takes, or None. + + A whole column of int, float or bool without a null is one numpy array and one buffer copy (``long[]``, ``double[]``, + ``boolean[]``); a column of str and None is one ``StringColumn``; date and naive date-time columns are one ``TemporalColumn`` + (a ``long[]`` of epoch days or microseconds, and one ``boolean[]`` when there is a null); a numeric column with a null converts + per element into an ``Object[]``, as ``insert_columns`` does, and a column of nothing but None is an ``Object[]`` of nulls. + """ + n = len(values) + if kind == "null": + return jpype.JArray(jpype.JObject)([None] * n) + if kind == "str": + return _string_column(values) + if kind == "date" or kind == "datetime": + mask = None + if nullable: + mask = jpype.JArray(jpype.JBoolean)( + _np.fromiter( + map(operator.is_, values, itertools.repeat(None)), + dtype=_np.bool_, + count=n, + ) + ) + filler = _EPOCH_DATE if kind == "date" else _EPOCH_DATETIME + values = [filler if v is None else v for v in values] + temporal = _java_class("com.arcadedb.python.TemporalColumn") + if kind == "date": + ticks = ( + _np.fromiter(map(date.toordinal, values), dtype=_np.int64, count=n) + - _EPOCH_ORDINAL + ) + return temporal.dates(jpype.JArray(jpype.JLong)(ticks), 1, mask) + ticks = _np.array(values, dtype="datetime64[us]").astype(_np.int64) + return temporal.dateTimes(jpype.JArray(jpype.JLong)(ticks), 1_000_000, mask) + if nullable: + return _column_to_java(name, values)[0] + dtype = {"int": _np.int64, "float": _np.float64, "bool": _np.bool_}[kind] + try: + array = _np.fromiter(values, dtype=dtype, count=n) + except (OverflowError, ValueError, TypeError): + return None + return _column_to_java(name, array)[0] + + +def _typed_chunk_columns(chunk): + """``(names, columns)`` for one chunk of insert_many rows that can cross as typed columns, else None (the chunk keeps the JSON text). + + The chunk qualifies when every row is a ``dict`` with the same str keys in the same order (a row that lacks a key, has another + one, or lists them in another order is a different shape, and a record keeps its properties in the order they were set), and every + column holds one kind of value (``_typed_kind``). Whether a column qualifies is decided from the types alone before any bridge + array is built, so a chunk that cannot cross this way costs the census and nothing else. The result is stored as the JSON text + stores it: ``DocumentBatcher.insertManyColumns`` stores a whole number as an Integer when it fits an int, as the text parser + does, and ``None`` is a null property, never a missing one. + """ + if _np is None or not chunk or set(map(type, chunk)) != _DICT_ONLY: + return None + names = tuple(chunk[0]) + if not names or set(map(type, names)) != _STR_ONLY: + return None + if len(set(map(tuple, chunk))) != 1: + return None + lists = [] + shapes = [] + for name in names: + values = list(map(operator.itemgetter(name), chunk)) + shape = _typed_kind(values) + if shape is None: + return None + lists.append(values) + shapes.append(shape) + columns = [] + for name, values, (kind, nullable) in zip(names, lists, shapes): + column = _typed_java_column(name, values, kind, nullable) + if column is None: + return None + columns.append(column) + return names, columns + + def _wrap_java_record(java_record, database=None): """Wrap a Java record in the matching Python class (Vertex/Edge/Document).""" if java_record is None: @@ -760,11 +888,17 @@ def insert_many( ) -> int: """Bulk-insert documents with one FFI crossing per chunk of rows. - The iterable is read 10,000 rows at a time; each chunk is serialized - to one JSON string and looped Java-side (``DocumentBatcher``), - avoiding the per-row JNI cost that caps ``new_document``-loop ingest. + The iterable is read 10,000 rows at a time; each chunk crosses once and + is looped Java-side (``DocumentBatcher``), avoiding the per-row JNI cost + that caps ``new_document``-loop ingest. A chunk whose rows are dicts with + the same keys in the same order, and whose columns each hold ``None`` and + values of one type (``int`` within 64 bits, ``float``, ``bool``, ``str``, + ``date``, or a ``datetime`` without a zone), crosses as typed columns, with + no text (``DocumentBatcher.insertManyColumns``); a whole number is stored + as an Integer when it fits an int and as a Long beyond, which is what the + JSON text stored. Any other chunk is serialized to one JSON string. Memory stays bounded by the chunk whatever the row count, so a - generator of any length can be loaded (#294). A value that sits directly + generator of any length can be loaded (#294). On the JSON route, a value that sits directly in a row may be JSON-representable (str/int/float/bool/None, nested lists/dicts), a ``datetime`` (also ``pandas.Timestamp`` and ``numpy.datetime64``), ``date``, ``Decimal``, ``bytes``/``bytearray``, @@ -939,6 +1073,12 @@ def _insert_many_chunk(self, type_name, chunk): still go in bulk, and every row is inserted in the order given. Dates, date-times, Decimals, bytes, numpy scalars and pandas NA and NaT sitting directly in a row cross inside the text as tagged values (see json_bulk_dumps_tagged). """ + # A chunk of rows with the same keys whose columns each hold one kind of value crosses as typed columns, with no text + # (see _typed_chunk_columns); every other chunk keeps the JSON text below, as it was. + typed = _typed_chunk_columns(chunk) + if typed is not None: + self._insert_many_typed(type_name, typed, len(chunk)) + return # Float vectors for ARRAY_OF_FLOATS properties cross as one binary matrix # beside the JSON, not as text (see split_float_vectors). split = split_float_vectors(self._java_db, type_name, chunk) @@ -949,6 +1089,21 @@ def _insert_many_chunk(self, type_name, chunk): return self._insert_many_bulk(type_name, payload, split) + def _insert_many_typed(self, type_name, typed, n): + """One chunk of insert_many as typed columns inside the open transaction (commitEvery 0: the transaction stays with insert_many).""" + names, columns = typed + try: + _java_class("com.arcadedb.python.DocumentBatcher").insertManyColumns( + self._java_db, + type_name, + jpype.JArray(jpype.JString)(names), + jpype.JArray(jpype.JObject)(columns), + n, + 0, + ) + except Exception as e: + raise ArcadeDBError(f"Failed to bulk-insert into '{type_name}': {e}") from e + def _insert_many_runs(self, type_name, chunk): """A chunk with a row the JSON text cannot carry: runs of carriable rows go in bulk, the other rows one by one, in order.""" skip = declared_float_vectors(self._java_db, type_name) @@ -1008,6 +1163,22 @@ def _insert_many_parallel(self, type_name, first, rows): chunk = first try: while chunk: + typed = _typed_chunk_columns(chunk) + if typed is not None: + try: + failures = batcher.insertManyColumnsParallel( + self._java_db, + type_name, + jpype.JArray(jpype.JString)(typed[0]), + jpype.JArray(jpype.JObject)(typed[1]), + len(chunk), + ) + except Exception as e: + raise self._parallel_load_failed(type_name, e) from e + chunk_failures.append(failures) + n += len(chunk) + chunk = list(itertools.islice(rows, _INSERT_MANY_CHUNK)) + continue bad = [] split = split_float_vectors(self._java_db, type_name, chunk) try: 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/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() diff --git a/src/java/com/arcadedb/python/DocumentBatcher.java b/src/java/com/arcadedb/python/DocumentBatcher.java index 0edeb45..5ed7424 100644 --- a/src/java/com/arcadedb/python/DocumentBatcher.java +++ b/src/java/com/arcadedb/python/DocumentBatcher.java @@ -11,7 +11,8 @@ * the async executor's parallel bucket writers (insertManyJsonParallel; the * Python insert_many wrapper waits for completion itself, then reads the * failures the writers reported). insertColumns takes whole columns instead of - * JSON (Database.insert_columns). Float vectors for ARRAY_OF_FLOATS properties cross as binary matrices (FloatVectors), beside + * JSON (Database.insert_columns); insertManyColumns is the same for a chunk of insert_many rows whose columns each hold one kind of + * value, and stores every number as the JSON text route did (see narrowed). Float vectors for ARRAY_OF_FLOATS properties cross as binary matrices (FloatVectors), beside * the JSON or as a column. The boxDoubles/boxLongs * helpers below serve AsyncExecutor.append_samples' numpy fast path. * @@ -81,25 +82,40 @@ 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) { + return insertColumns(db, typeName, names, columns, n, commitEvery, false); + } + + /** + * The columns of one insert_many chunk (every row has the same keys in the same order and each column holds one kind of value): as + * {@link #insertColumns}, except that a whole number is stored as the JSON text route stored it, an Integer when it fits an int and a + * Long beyond, so the type of a schemaless property does not depend on which route carried the chunk (see {@link #narrowed}). + */ + public static long insertManyColumns(final Database db, final String typeName, final String[] names, final Object[] columns, + final int n, final int commitEvery) { + return insertColumns(db, typeName, names, columns, n, commitEvery, true); + } + + private static long insertColumns(final Database db, final String typeName, final String[] names, final Object[] columns, + final int n, final int commitEvery, final boolean narrowIntegers) { checkColumns(names, columns, n); final boolean vertex = n > 0 && isVertexType(db, typeName); writeChunks(db, n, commitEvery, i -> { final MutableDocument doc = newRecord(db, typeName, vertex); - fillRow(doc, names, columns, i); + fillRow(doc, names, columns, i, narrowIntegers); doc.save(); }); return n; @@ -180,13 +196,24 @@ private static void backoffBeforeRetry(final int attempt) { */ public static AsyncFailures insertColumnsParallel(final Database db, final String typeName, final String[] names, final Object[] columns, final int n) { + return insertColumnsParallel(db, typeName, names, columns, n, false); + } + + /** The parallel twin of {@link #insertManyColumns}. */ + public static AsyncFailures insertManyColumnsParallel(final Database db, final String typeName, final String[] names, + final Object[] columns, final int n) { + return insertColumnsParallel(db, typeName, names, columns, n, true); + } + + private static AsyncFailures insertColumnsParallel(final Database db, final String typeName, final String[] names, + final Object[] columns, final int n, final boolean narrowIntegers) { checkColumns(names, columns, n); final AsyncFailures failures = new AsyncFailures(); final ErrorCallback onError = failures::record; final boolean vertex = n > 0 && isVertexType(db, typeName); for (int i = 0; i < n; i++) { final MutableDocument doc = newRecord(db, typeName, vertex); - fillRow(doc, names, columns, i); + fillRow(doc, names, columns, i, narrowIntegers); db.async().createRecord(doc, null, onError); } return failures; @@ -229,20 +256,35 @@ 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"); } } - private static void fillRow(final MutableDocument doc, final String[] names, final Object[] columns, final int i) { + /** + * The value a JSON number token becomes in the engine (JSONObject.parseNumber): an Integer when a whole number fits an int, a Long + * beyond. The columns of a numpy long[] hold every whole number as a long, which would store a Long where the text route stored an + * Integer, so the insert_many columns narrow it back. Written with returns, not a conditional: {@code c ? (int) v : v} would promote + * both arms to long. + */ + private static Object narrowed(final long v) { + if (v >= Integer.MIN_VALUE && v <= Integer.MAX_VALUE) + return (int) v; + return v; + } + + private static void fillRow(final MutableDocument doc, final String[] names, final Object[] columns, final int i, + final boolean narrowIntegers) { for (int c = 0; c < names.length; c++) { final Object col = columns[c]; - final Object v; + Object v; if (col instanceof long[] a) - v = a[i]; + v = narrowIntegers ? narrowed(a[i]) : (Object) a[i]; else if (col instanceof double[] a) v = a[i]; else if (col instanceof boolean[] a) @@ -251,8 +293,13 @@ else if (col instanceof FloatVectors.Matrix m) v = m.row(i); else if (col instanceof StringColumn s) v = s.get(i); - else + else if (col instanceof TemporalColumn t) + v = t.get(i); + else { v = ((Object[]) col)[i]; + if (narrowIntegers && v instanceof Long l) + v = narrowed(l); + } 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); + } + } +} diff --git a/tests/conftest.py b/tests/conftest.py index 32ddf8f..7f1ea24 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") @@ -150,10 +150,13 @@ "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)", + "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)", + "test_insert_many_typed_columns.py": "the typed-column route of the JVM wheel's insert_many (the bridge's DocumentBatcher.insertManyColumns, core._typed_chunk_columns, the Java classes of the stored values)", } if BACKEND == "native" and not os.environ.get("NATIVE_PARITY_ALL"): collect_ignore.extend(NATIVE_NOT_APPLICABLE) 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") diff --git a/tests/test_insert_many_typed_columns.py b/tests/test_insert_many_typed_columns.py new file mode 100644 index 0000000..5b2250b --- /dev/null +++ b/tests/test_insert_many_typed_columns.py @@ -0,0 +1,619 @@ +"""insert_many (the JVM wheel): a chunk whose rows have the same keys and whose columns each hold one kind of value crosses as typed columns. + +Before, every chunk of 10,000 rows became one JSON text (``json_bulk_dumps_tagged``), which the engine parsed again into one JSON tree and +copied key by key (``DocumentBatcher.insertManyJson``). Now a homogeneous chunk takes ``DocumentBatcher.insertManyColumns``: one numpy +buffer per int, float or bool column, one UTF-8 buffer per string column, one ``TemporalColumn`` per date or date-time column, and no text. +Every other chunk keeps the JSON text exactly as it was. + +The stored data has to be the same on both routes, so each parity test stores the same rows twice, once with the typed route switched off +(``core._typed_chunk_columns`` returns None: the old route) and once as shipped, and compares every record property by property +through SQL: the property names in the order the record lists them, the Java class of every value (an Integer is not a Long), and the +value. The route that ran is read from the calls made on the bridge class. The Java types are read through the Java database +(``get_java_database``), so this file has no native counterpart (the native wheel carries homogeneous chunks through its own columnar +entry, ``arcadedb._native.bulk.insert_rows_columnar``). +""" + +import datetime as dt +import decimal +import math + +import jpype +import numpy as np +import pytest + +from arcadedb import core +from arcadedb.exceptions import ArcadeDBError + +BRIDGE = "com.arcadedb.python.DocumentBatcher" +UTC9 = dt.timezone(dt.timedelta(hours=9)) + + +class _Spy: + """Forwards every call of a Java class and records its name.""" + + def __init__(self, target, log): + self._target, self._log = target, log + + def __getattr__(self, name): + attribute = getattr(self._target, name) + if not callable(attribute): + return attribute + + def call(*args): + self._log.append(name) + return attribute(*args) + + return call + + +@pytest.fixture +def bridge(monkeypatch): + """The names of the DocumentBatcher calls of the test, in order.""" + log = [] + real = core._java_class + + def spy(name): + cls = real(name) + return _Spy(cls, log) if name == BRIDGE else cls + + monkeypatch.setattr(core, "_java_class", spy) + return log + + +def _old_route(monkeypatch): + """The route before this change: no chunk is typed, every chunk is one JSON text.""" + monkeypatch.setattr(core, "_typed_chunk_columns", lambda chunk: None) + + +def _describe(value): + if value is None: + return ("null",) + if isinstance(value, str): + return ("str", value) + if isinstance(value, jpype.JArray): + return (str(value.getClass().getName()), repr(list(value))) + if hasattr(value, "getClass"): + return (str(value.getClass().getName()), str(value)) + return ("py:" + type(value).__name__, repr(value)) + + +def _stored(db, type_name): + """Every record by id: [(property, (Java class, text of the value)), ...] in the order the record lists its properties.""" + result = db.get_java_database().query( + "sql", f"SELECT FROM {type_name} ORDER BY id" # nosec B608 - test-owned name + ) + records = [] + while result.hasNext(): + record = result.next() + records.append( + [ + (str(name), _describe(record.getProperty(name))) + for name in record.getPropertyNames() + ] + ) + return records + + +DECLARED = { + "i": "LONG", + "f": "DOUBLE", + "b": "BOOLEAN", + "s": "STRING", + "d": "DATE", + "t": "DATETIME", +} + + +def _make(db, name, declared=False, kind="DOCUMENT", buckets=None): + suffix = f" BUCKETS {buckets}" if buckets else "" + db.command("sql", f"CREATE {kind} TYPE {name}{suffix}") + db.command("sql", f"CREATE PROPERTY {name}.id INTEGER") + if declared: + for prop, ty in DECLARED.items(): + db.command("sql", f"CREATE PROPERTY {name}.{prop} {ty}") + + +INT_EDGES = [ + 0, + 1, + -1, + 2**31 - 1, + 2**31, + -(2**31), + -(2**31) - 1, + 2**53 + 1, + -(2**53) - 1, + 2**63 - 1, + -(2**63), +] +FLOAT_EDGES = [ + 0.0, + -0.0, + 0.1, + 1.0, + 1e300, + -1e-300, + 5e-324, + 2.2250738585072014e-308, + 1.7976931348623157e308, + float("nan"), + float("inf"), + float("-inf"), +] +STRINGS = [ + "", + "a", + 'quo"te', + "back\\slash", + "new\nline", + "tab\t", + "emoji \U0001f600", + "é ü 中", + "\x01tag", + "x" * 5000, +] +DATES = [ + dt.date(1970, 1, 1), + dt.date(1969, 12, 31), + dt.date(2026, 11, 14), + dt.date(1, 1, 1), + dt.date(9999, 12, 31), + dt.date(2000, 2, 29), +] +DATETIMES = [ + dt.datetime(1970, 1, 1), + dt.datetime(1969, 12, 31, 23, 59, 59, 999999), + dt.datetime(2026, 11, 14, 12, 34, 56, 123456), + dt.datetime(2000, 2, 29, 0, 0, 0, 1), + dt.datetime(1, 1, 1), + dt.datetime(9999, 12, 31, 23, 59, 59, 999999), +] + + +def _one_column(values, name="v"): + return [{"id": i, name: v} for i, v in enumerate(values)] + + +def _cycle(values, n): + return [values[i % len(values)] for i in range(n)] + + +def _all_kinds(nulls): + n = 12 + rows = [] + for i in range(n): + null = nulls and i % 3 == 1 + rows.append( + { + "id": i, + "i": None if null else INT_EDGES[i % len(INT_EDGES)], + "f": None if null else FLOAT_EDGES[i % len(FLOAT_EDGES)], + "b": None if null else i % 2 == 0, + "s": None if null else STRINGS[i % len(STRINGS)], + "d": None if null else DATES[i % len(DATES)], + "t": None if null else DATETIMES[i % len(DATETIMES)], + } + ) + return rows + + +# name -> (rows, typed): the rows the chunk holds and the route they take +CASES = { + "ints": (_one_column(INT_EDGES), True), + "floats_nan_inf_subnormal": (_one_column(FLOAT_EDGES), True), + "bools": (_one_column([True, False, True, True]), True), + "strings": (_one_column(STRINGS), True), + "dates": (_one_column(DATES), True), + "datetimes": (_one_column(DATETIMES), True), + "all_kinds": (_all_kinds(False), True), + "all_kinds_with_nulls": (_all_kinds(True), True), + "null_in_every_row": (_one_column([None, None, None]), True), + "null_beside_ints": (_one_column([1, None, 2**40, None]), True), + "null_beside_floats": (_one_column([1.5, None, float("nan")]), True), + "null_beside_strings": (_one_column(["a", None, "b"]), True), + "single_row": (_one_column([7]), True), + # shapes the typed route cannot carry: the chunk keeps the JSON text and the per-row path for the rows the text cannot carry + "other_key_order": ( + [ + {"id": 0, "a": 1, "b": 2}, + {"id": 1, "b": 3, "a": 4}, + {"id": 2, "a": 5, "b": 6}, + ], + False, + ), + "key_missing_in_a_row": ([{"id": 0, "a": 1}, {"id": 1}, {"id": 2, "a": 3}], False), + "null_in_one_row_missing_in_another": ( + [{"id": 0, "a": None}, {"id": 1}, {"id": 2, "a": 2}], + False, + ), + "an_extra_key_in_a_row": ([{"id": 0, "a": 1}, {"id": 1, "a": 2, "b": 3}], False), + "int_beside_float": (_one_column([1, 2.5, 3]), False), + "bool_beside_int": (_one_column([True, 1, False]), False), + "str_beside_int": (_one_column(["a", 1]), False), + "date_beside_datetime": (_one_column([DATES[0], DATETIMES[0]]), False), + "date_inside_a_list": (_one_column([[DATES[2]], [DATES[0]]]), False), + "datetime_inside_a_list": (_one_column([[DATETIMES[2]], [1]]), False), + "decimal_inside_a_list": (_one_column([[decimal.Decimal("1.5")], [2]]), False), + "decimal": ( + _one_column([decimal.Decimal("12345.6789"), decimal.Decimal("-0.50")]), + False, + ), + "bytes": (_one_column([b"\x00\xff abc\x7f", b""]), False), + "numpy_scalars": ( + _one_column([np.int64(2**40 + 5), np.int64(-3), np.int64(7)]), + False, + ), + "numpy_float_scalars": (_one_column([np.float64(0.1), np.float64(2.5)]), False), + "aware_datetime": ( + _one_column([dt.datetime(2026, 5, 6, 7, 8, 9, 654321, tzinfo=UTC9)]), + False, + ), + "nested_lists_and_maps": ( + _one_column([[1, 2, [3, 4]], {"k": [1, {"z": 2}], "n": None}]), + False, + ), + "tuple": (_one_column([(1, 2, 3), (4, 5, 6)]), False), + "list_of_ints_and_a_float": (_one_column([[1, 2.5], [3, 4]]), False), +} + + +# every case on schemaless properties, and the rows of every kind on declared ones too (the engine converts to the declared type) +PAIRS = [(case, False) for case in CASES] + [ + ("all_kinds", True), + ("all_kinds_with_nulls", True), +] + + +@pytest.mark.parametrize( + "case, declared", + PAIRS, + ids=[f"{c}-{'declared' if d else 'free'}" for c, d in PAIRS], +) +def test_the_typed_route_stores_what_the_json_route_stores( + temp_db, monkeypatch, bridge, case, declared +): + rows, typed = CASES[case] + _make(temp_db, "Old", declared) + _make(temp_db, "New", declared) + with monkeypatch.context() as old: + _old_route(old) + bridge.clear() + assert temp_db.insert_many("Old", rows) == len(rows) + old_calls = list(bridge) + assert "insertManyColumns" not in old_calls + bridge.clear() + assert temp_db.insert_many("New", rows) == len(rows) + if typed: + assert bridge == ["insertManyColumns"], f"{case} left the typed route" + else: + # the chunk keeps the route it always had, call for call (the JSON text, and the per-row path for rows the text cannot carry) + assert ( + bridge == old_calls + ), f"{case} took the typed route, which cannot carry it" + new, old = _stored(temp_db, "New"), _stored(temp_db, "Old") + assert len(new) == len(rows) + assert new == old + + +def test_a_whole_number_is_an_integer_when_it_fits_an_int_and_a_long_beyond( + temp_db, bridge +): + """The classes themselves, not only old against new: what the JSON text stored (JSONObject.parseNumber).""" + _make(temp_db, "T") + temp_db.insert_many( + "T", _one_column([5, 2**31 - 1, 2**31, -(2**31), -(2**31) - 1, 2**62]) + ) + assert bridge == ["insertManyColumns"] + classes = [dict(r)["v"][0] for r in _stored(temp_db, "T")] + assert classes == [ + "java.lang.Integer", + "java.lang.Integer", + "java.lang.Long", + "java.lang.Integer", + "java.lang.Long", + "java.lang.Long", + ] + + +def test_insert_columns_still_stores_a_long_for_every_whole_number(temp_db): + """insert_columns shares fillRow with the new entry point; only insertManyColumns narrows (a regression test, green before this change).""" + _make(temp_db, "C") + temp_db.insert_columns("C", {"id": np.arange(3, dtype=np.int64), "v": [1, 2, 3]}) + for record in _stored(temp_db, "C"): + assert dict(record)["id"][0] == "java.lang.Long" + + +def test_a_chunk_that_takes_the_old_route_builds_no_bridge_array( + temp_db, monkeypatch, bridge +): + """A fallback chunk costs the census and nothing else: no column is built before the shape is known to qualify.""" + built = [] + real = core._typed_java_column + + def spy(*args): + built.append(args[0]) + return real(*args) + + monkeypatch.setattr(core, "_typed_java_column", spy) + _make(temp_db, "T") + # the first column is fine, the second holds a Decimal: nothing is built for either + temp_db.insert_many("T", [{"id": i, "v": decimal.Decimal(i)} for i in range(5)]) + assert built == [] + assert bridge == ["insertManyJson"] + + +def test_a_homogeneous_load_crosses_once_per_chunk_with_no_text_and_no_per_value_conversion( + temp_db, monkeypatch, bridge +): + monkeypatch.setattr(core, "_INSERT_MANY_CHUNK", 1000) + + def no_text(*args, **kwargs): + raise AssertionError("a homogeneous chunk went through the JSON text") + + monkeypatch.setattr(core, "json_bulk_dumps_tagged", no_text) + converted = [] + real = core.convert_python_to_java + + def counting(value): + converted.append(type(value).__name__) + return real(value) + + monkeypatch.setattr(core, "convert_python_to_java", counting) + _make(temp_db, "T") + rows = [ + { + "id": i, + "n": i * 3, + "x": i * 0.5, + "ok": i % 2 == 0, + "name": f"row {i}", + "day": DATES[i % len(DATES)], + "at": DATETIMES[i % len(DATETIMES)], + } + for i in range(2500) + ] + assert temp_db.insert_many("T", rows) == 2500 + assert bridge == ["insertManyColumns"] * 3 # chunks of 1000, 1000, 500 + assert converted == [], "a value was converted one at a time" + assert temp_db.query("sql", "SELECT count(*) AS c FROM T").first().get("c") == 2500 + record = dict( + _stored(temp_db, "T")[1235] + ) # 1235 % 6 == 5: the last date and date-time of the lists above + assert record == { + "id": ("java.lang.Integer", "1235"), + "n": ("java.lang.Integer", "3705"), + "x": ("java.lang.Double", "617.5"), + "ok": ("java.lang.Boolean", "false"), + "name": ("str", "row 1235"), + "day": ("java.time.LocalDate", "2000-02-29"), + "at": ("java.time.LocalDateTime", DATETIMES[5].isoformat()), + } + + +def test_chunks_of_both_routes_keep_the_order_of_the_rows(temp_db, monkeypatch, bridge): + monkeypatch.setattr(core, "_INSERT_MANY_CHUNK", 4) + _make(temp_db, "T", buckets=1) + rows = _one_column(list(range(12))) + rows[5] = {"id": 5, "v": decimal.Decimal(5)} # the middle chunk keeps the text + assert temp_db.insert_many("T", rows) == 12 + assert bridge == ["insertManyColumns", "insertManyJson", "insertManyColumns"] + ids = [r.get("id") for r in temp_db.query("sql", "SELECT id FROM T").to_list()] + assert ids == list(range(12)) + + +def test_the_commit_boundary_still_splits_a_chunk(temp_db, bridge): + _make(temp_db, "T") + assert temp_db.insert_many("T", _one_column(list(range(25))), commit_every=10) == 25 + assert ( + bridge == ["insertManyColumns"] * 3 + ) # 10, 10, 5: a chunk never straddles a commit + assert len(_stored(temp_db, "T")) == 25 + + +def test_an_empty_batch_and_a_batch_of_one(temp_db, bridge): + _make(temp_db, "T") + assert temp_db.insert_many("T", []) == 0 + assert bridge == [] + assert temp_db.insert_many("T", [{"id": 1, "v": 2}]) == 1 + assert bridge == ["insertManyColumns"] + assert len(_stored(temp_db, "T")) == 1 + + +def test_a_generator_is_read_chunk_by_chunk(temp_db, monkeypatch, bridge): + monkeypatch.setattr(core, "_INSERT_MANY_CHUNK", 100) + _make(temp_db, "T") + n = temp_db.insert_many("T", ({"id": i, "v": i} for i in range(250))) + assert n == 250 + assert bridge == ["insertManyColumns"] * 3 + + +@pytest.mark.parametrize("kind", ["DOCUMENT", "VERTEX"]) +def test_documents_and_vertices_are_the_same_on_both_routes( + temp_db, monkeypatch, bridge, kind +): + _make(temp_db, "Old", kind=kind) + _make(temp_db, "New", kind=kind) + rows = _one_column(INT_EDGES) + with monkeypatch.context() as old: + _old_route(old) + temp_db.insert_many("Old", rows) + temp_db.insert_many("New", rows) + assert _stored(temp_db, "New") == _stored(temp_db, "Old") + for name in ("Old", "New"): + sql = f"SELECT FROM {name}" # nosec B608 - test-owned name + result = temp_db.get_java_database().query("sql", sql) + n = 0 + while result.hasNext(): + assert result.next().isVertex() == (kind == "VERTEX") + n += 1 + assert n == len(rows) + + +def test_an_edge_type_is_refused_as_before(temp_db, monkeypatch): + temp_db.command("sql", "CREATE VERTEX TYPE V") + temp_db.command("sql", "CREATE EDGE TYPE E") + rows = _one_column([1, 2]) + messages = [] + for route in ("old", "new"): + with monkeypatch.context() as m: + if route == "old": + _old_route(m) + with pytest.raises(ArcadeDBError) as error: + temp_db.insert_many("E", rows) + messages.append(str(error.value)) + assert messages[0] == messages[1] + assert "edge type" in messages[0] + + +def test_a_duplicate_key_fails_the_call_and_rolls_it_back_on_both_routes( + temp_db, monkeypatch, bridge +): + outcomes = [] + for route in ("old", "new"): + name = f"T_{route}" + temp_db.command("sql", f"CREATE DOCUMENT TYPE {name}") + temp_db.command("sql", f"CREATE PROPERTY {name}.id INTEGER") + temp_db.command("sql", f"CREATE INDEX ON {name} (id) UNIQUE") + rows = _one_column([1, 2, 3, 4]) + [ + {"id": 2, "v": 9} + ] # the id 2 is there twice + with monkeypatch.context() as m: + if route == "old": + _old_route(m) + bridge.clear() + with pytest.raises( + Exception + ) as error: # noqa: B017 - the class is compared below + temp_db.insert_many(name, rows) + outcomes.append( + ( + type(error.value).__name__, + temp_db.query("sql", f"SELECT count(*) AS c FROM {name}") # nosec B608 + .first() + .get("c"), + list(bridge), + ) + ) + assert outcomes[0][:2] == outcomes[1][:2] + assert outcomes[1][1] == 0, "the call rolled back the transaction it opened" + assert outcomes[0][2] == ["insertManyJson"] + assert outcomes[1][2] == ["insertManyColumns"] + + +@pytest.mark.parametrize( + "values", + [[1, 2**64], [-(2**63) - 1, 1], ["ok", "a\ud800b"]], + ids=["beyond_int64", "below_int64", "lone_surrogate"], +) +def test_a_value_no_route_can_store_fails_as_before( + temp_db, monkeypatch, bridge, values +): + outcomes = [] + for route in ("old", "new"): + name = f"T_{route}" + _make(temp_db, name) + with monkeypatch.context() as m: + if route == "old": + _old_route(m) + bridge.clear() + try: + temp_db.insert_many(name, _one_column(values)) + raised = None + except Exception as error: # noqa: BLE001 - the same class is the claim + raised = type(error).__name__ + outcomes.append( + (raised, len(_stored(temp_db, name)), "insertManyColumns" in bridge) + ) + assert outcomes[0][:2] == outcomes[1][:2] + assert ( + outcomes[1][2] is False + ), "a value the text refuses must not take the typed route" + + +@pytest.mark.parametrize("case", ["ints", "all_kinds_with_nulls", "decimal"]) +def test_parallel_mode_stores_what_the_json_route_stores( + temp_db, monkeypatch, bridge, case +): + rows, typed = CASES[case] + _make(temp_db, "Old") + _make(temp_db, "New") + with monkeypatch.context() as old: + _old_route(old) + bridge.clear() + assert temp_db.insert_many("Old", rows, parallel=True) == len(rows) + assert "insertManyJsonParallel" in bridge + bridge.clear() + assert temp_db.insert_many("New", rows, parallel=True) == len(rows) + assert bridge[0] == ( + "insertManyColumnsParallel" if typed else "insertManyJsonParallel" + ) + assert _stored(temp_db, "New") == _stored(temp_db, "Old") + + +def test_a_caller_transaction_is_left_to_the_caller(temp_db, bridge): + _make(temp_db, "T") + with pytest.raises(RuntimeError): + with temp_db.transaction(): + temp_db.insert_many("T", _one_column(list(range(30))), commit_every=10) + raise RuntimeError("roll back") + assert bridge == ["insertManyColumns"] + assert temp_db.query("sql", "SELECT count(*) AS c FROM T").first().get("c") == 0 + + +@pytest.mark.parametrize( + "values, expected", + [ + ([1, 2, 3], ("int", False)), + ([1, None], ("int", True)), + ([2**63], None), + ([-(2**63) - 1, None], None), + ([1, 2.0], None), + ([True, False], ("bool", False)), + ([True, 1], None), + ([None, None], ("null", True)), + (["a", None], ("str", True)), + ([1.5, math.nan], ("float", False)), + ([dt.date(2020, 1, 2)], ("date", False)), + ([dt.datetime(2020, 1, 2)], ("datetime", False)), + ([dt.datetime(2020, 1, 2), None], ("datetime", True)), + ([dt.datetime(2020, 1, 2, tzinfo=dt.timezone.utc)], None), + ([dt.date(2020, 1, 2), dt.datetime(2020, 1, 2)], None), + ([decimal.Decimal(1)], None), + ([b"x"], None), + ([[1, 2]], None), + ([{"a": 1}], None), + ([np.int64(1)], None), + ([np.float64(1)], None), + ], +) +def test_which_columns_qualify(values, expected): + assert core._typed_kind(values) == expected + + +@pytest.mark.parametrize( + "chunk", + [ + [], + [{}, {}], + [{1: 1}], + [{"a": 1}, [("a", 2)]], + [{"a": 1, "b": 2}, {"b": 2, "a": 1}], + [{"a": 1}, {"a": 1, "b": 2}], + [{"a": 1}, {"b": 1}], + [{"a": 1}, {"a": 2.5}], + [{"a": [1]}, {"a": [2]}], + ], + ids=[ + "empty", + "no_keys", + "non_str_key", + "not_a_dict", + "key_order", + "missing_key", + "other_key", + "mixed_kind", + "list", + ], +) +def test_which_chunks_qualify(chunk): + assert core._typed_chunk_columns(chunk) is None diff --git a/tests/test_native_one_row_crossings.py b/tests/test_native_one_row_crossings.py index 73f32a9..d24cde4 100644 --- a/tests/test_native_one_row_crossings.py +++ b/tests/test_native_one_row_crossings.py @@ -7,7 +7,12 @@ 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: +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. """ @@ -27,7 +32,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 +44,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 @@ -190,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() diff --git a/tests/test_native_tx_crossings.py b/tests/test_native_tx_crossings.py index 7f9f618..78a1635 100644 --- a/tests/test_native_tx_crossings.py +++ b/tests/test_native_tx_crossings.py @@ -1,16 +1,21 @@ -"""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. """ +import contextlib import json import pytest @@ -20,20 +25,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(name, *args): - self.calls.append(_OPS[args[1]] if name == "arcadedb_n_tx" else name) - return real(name, *args) + def counting_tx(handle, op): + self.calls.append(_OPS[op]) + return real_tx(handle, op) - monkeypatch.setattr(lib, "call", counting) + def counting_call(name, *args): + self.calls.append(name) + return real_call(name, *args) + + monkeypatch.setattr(lib, "tx", counting_tx) + monkeypatch.setattr(lib, "call", counting_call) @property def tx(self): @@ -43,6 +60,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,8 +113,19 @@ def db(temp_db_path): database.transaction() ): # the thread's transaction context exists from here on, as it does in a running application pass - yield database - database.drop() + lib.flush_closes() # a cursor closed earlier is not part of a call measured below + 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 @@ -60,6 +133,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 +197,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 +266,23 @@ def test_a_nested_block_that_succeeds_commits_with_the_caller(db): assert _count(db) == 2 +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) + with db.transaction(): + assert db._transaction_stack_mark() == (True, base + 1) + db.command("sql", "INSERT INTO R SET k = 1") + assert db._transaction_stack_mark() == (True, base) + 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 +333,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 +372,101 @@ 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() + # 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() 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 + # 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) + two = lib.tx(handle, 0) + 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) # 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 + + +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_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_an_error_reply_is_decoded_into_the_same_error(db, monkeypatch): +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 +475,36 @@ 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) + 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: + 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() 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