diff --git a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc index 3ba0d54..6df158b 100644 --- a/docs/modules/ROOT/pages/how-to/bulk-loads.adoc +++ b/docs/modules/ROOT/pages/how-to/bulk-loads.adoc @@ -83,7 +83,7 @@ The `arcadedb` (native) package does not change the collector. In our measuremen == Property shapes in a graph batch (`arcadedb-jvm`) -On the `arcadedb-jvm` package, `GraphBatch.create_vertices` and `GraphBatch.new_edges` load these property values through the bulk path (the same tagged JSON text as `insert_many`): a list, a tuple, a dict, a `datetime`, a `date`, a `Decimal` and `bytes` (also `bytearray`). A value is stored as the one-by-one path stores it. A list or tuple of only `int` and `float` keeps the older path, which crosses it as one `double[]`. +On the `arcadedb-jvm` package, `GraphBatch.create_vertices` and `GraphBatch.new_edges` send rows whose properties are all `int`, `float`, `bool`, `str` or `None` (every row a `dict` with the same keys in the same order, each property one kind) as one typed column per property, with no JSON text, and `create_vertices` returns the RIDs from one `long[]` of bucket ids and positions. The values below, and any other rows, go through the bulk path (the same tagged JSON text as `insert_many`): a list, a tuple, a dict, a `datetime`, a `date`, a `Decimal` and `bytes` (also `bytearray`). A value is stored as the one-by-one path stores it. A list or tuple of only `int` and `float` keeps the older path, which crosses it as one `double[]`. A value that the text cannot carry sends work to the older path on its own: a `NaN`, an integer beyond 64 bits, a `datetime` inside a list or a dict, or a dict key that is not a `str`. `create_vertices` falls back for the chunk of at most 100,000 rows that holds the value, and `new_edges` for the whole call. A value of another type (a `set`, a numpy or pandas value, a subclass of the types above) also takes the older path. Nothing is lost on a fallback, and the data is the same. Only the speed changes. 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..68fc806 100644 --- a/src/arcadedb/core.py +++ b/src/arcadedb/core.py @@ -241,7 +241,8 @@ def _column_to_java(name, values): into a ``long[]``, ``double[]``, or ``boolean[]``; a string or object array and any other sequence of mixed values convert per element into an ``Object[]`` (a ``str`` reuses one Java String per distinct value, ``None`` is a null); a column of nothing but ``str`` and ``None`` crosses instead as one - ``StringColumn`` (``_string_column``). A pandas + ``StringColumn`` (``_string_column``). A bridge ``TemporalColumn``, which ``insert_arrow`` and ``insert_dataframe`` build from a + timestamp or date column (``dataframe._temporal_column``), crosses as it is. A pandas ``Series`` is read through ``to_numpy``, with a nullable dtype's ```` as ``None``. A 2-D ``float32`` array is a column of vectors: it crosses as one ``float[]`` and each document gets one row of it. Kinds that do not cross natively raise ``TypeError`` rather @@ -313,6 +314,8 @@ def _column_to_java(name, values): # a string, bytes, or object array: its elements are Python objects (str, None, ...), converted one by one below values = list(values.astype(object, copy=False)) elif not isinstance(values, (list, tuple)): + if isinstance(values, _java_class("com.arcadedb.python.TemporalColumn")): + return values, int(values.rows()) values = list(values) column = _string_column(values) if column is not None: diff --git a/src/arcadedb/dataframe.py b/src/arcadedb/dataframe.py index 4c7beb2..6fa3de5 100644 --- a/src/arcadedb/dataframe.py +++ b/src/arcadedb/dataframe.py @@ -245,43 +245,46 @@ def _check_declared( ) -def _temporal_to_java(chunk, pa): - """A timestamp or date chunk as a list of ``LocalDateTime``/``LocalDate``, keeping the unit's precision. +def _temporal_column(chunk, pa): + """A timestamp or date chunk as one ``TemporalColumn`` of the bridge, keeping the unit's precision. + + The integer epoch values cross as ONE ``long[]`` (a buffer copy), the null rows as one ``boolean[]`` when there is a null, and the + bridge builds each ``LocalDateTime`` or ``LocalDate`` as it fills the document of that row: one call per column and batch, where a + ``LocalDateTime.ofEpochSecond`` or ``LocalDate.ofEpochDay`` call per cell (and one conversion per element in ``insert_columns``) came + before. ``insert_columns`` takes the column as it is. DATETIME is a UTC wall clock: a timestamp with a zone is already UTC epoch time, and a timestamp without one is taken as the wall clock it shows (the same rule ``Document.set`` applies to a ``datetime``), so both - are read from the integer epoch value. Going through ``datetime`` would cut a nanosecond value to microseconds. + are read from the integer epoch value, in the unit of the column. Going through ``datetime`` would cut a nanosecond value to + microseconds. A date32 is days, a date64 is milliseconds (the bridge divides by 86,400,000, rounding toward negative infinity). """ import numpy as np types = pa.types - mask = chunk.is_null().to_numpy(zero_copy_only=False) if types.is_date(chunk.type): - local_date = jpype.JClass("java.time.LocalDate") - raw = chunk.cast(pa.int64() if types.is_date64(chunk.type) else pa.int32()) - days = raw.fill_null(0).to_numpy(zero_copy_only=False).astype(np.int64) - if types.is_date64(chunk.type): - days = days // 86_400_000 - return [ - None if m else local_date.ofEpochDay(int(d)) for d, m in zip(days, mask) - ] - per_second = {"s": 1, "ms": 1_000, "us": 1_000_000, "ns": 1_000_000_000}[ - chunk.type.unit - ] - values = chunk.cast(pa.int64()).fill_null(0).to_numpy(zero_copy_only=False) - seconds, rest = np.divmod(values.astype(np.int64), per_second) - nanos = rest * (1_000_000_000 // per_second) - local_datetime = jpype.JClass("java.time.LocalDateTime") - utc = jpype.JClass("java.time.ZoneOffset").UTC - return [ - None if m else local_datetime.ofEpochSecond(int(s), int(n), utc) - for s, n, m in zip(seconds, nanos, mask) - ] + date64 = types.is_date64(chunk.type) + raw = chunk.cast(pa.int64() if date64 else pa.int32()) + ticks, per_unit, factory = raw, 86_400_000 if date64 else 1, "dates" + else: + per_second = {"s": 1, "ms": 1_000, "us": 1_000_000, "ns": 1_000_000_000} + ticks = chunk.cast(pa.int64()) + per_unit, factory = per_second[chunk.type.unit], "dateTimes" + values = ticks.fill_null(0).to_numpy(zero_copy_only=False) + nulls = None + if chunk.null_count: + nulls = jpype.JArray(jpype.JBoolean)( + chunk.is_null().to_numpy(zero_copy_only=False) + ) + return getattr(jpype.JClass("com.arcadedb.python.TemporalColumn"), factory)( + jpype.JArray(jpype.JLong)(np.ascontiguousarray(values, dtype=np.int64)), + per_unit, + nulls, + ) def _temporal_to_tagged(chunk, pa): """A timestamp or date chunk as tagged values (``T`` date-time, ``D`` date) for the native backend, with the same - reading rules as :func:`_temporal_to_java`: ISO text of the integer epoch value, so a nanosecond value keeps its + reading rules as :func:`_temporal_column`: ISO text of the integer epoch value, so a nanosecond value keeps its nanoseconds (a Python ``datetime`` would cut them to microseconds). """ import numpy as np @@ -381,7 +384,8 @@ def _column_values( float_vectors: bool = False, ): """One Arrow array as what ``Database.insert_columns`` takes: a numpy array when the column crosses as one - buffer (integers, floats and booleans without nulls), else a list of values (``None`` is a null). + buffer (integers, floats and booleans without nulls), else a list of values (``None`` is a null). On the JVM backend a timestamp or + date column is one bridge ``TemporalColumn`` instead (``_temporal_column``: one ``long[]`` of epoch ticks, no value per cell). ``float_vectors`` (the JVM backend, a property declared ``ARRAY_OF_FLOATS``): a float32 vector column of one length and no null is one 2-D float32 matrix, which ``insert_columns`` sends as one ``float[]``; ``to_pylist`` made one Python float per element and @@ -434,9 +438,7 @@ def _column_values( if matrix is not None: return matrix if kind in ("timestamp", "date"): - return ( - _temporal_to_tagged(chunk, pa) if native else _temporal_to_java(chunk, pa) - ) + return _temporal_to_tagged(chunk, pa) if native else _temporal_column(chunk, pa) if native and kind in ("decimal", "binary"): return _scalars_to_tagged(kind, chunk) if kind == "null": diff --git a/src/arcadedb/graph_batch.py b/src/arcadedb/graph_batch.py index dbd9ba7..d97c7d8 100644 --- a/src/arcadedb/graph_batch.py +++ b/src/arcadedb/graph_batch.py @@ -4,6 +4,8 @@ Pythonic wrapper for ArcadeDB's high-throughput graph ingest API. """ +import itertools +import operator from typing import Any, Iterable, Optional import jpype @@ -12,6 +14,8 @@ from .exceptions import ArcadeDBError from .graph import Document, Vertex from .type_conversion import ( + _FLOAT_NORMAL_MIN, + _np, convert_python_to_java, declared_float_vectors, declared_scalars_ok, @@ -26,6 +30,108 @@ _LOGGER = get_logger(__name__) +_NONE_KIND = type(None) +# what a masked (null) row of a typed column holds: any value of the column's kind, never read +_FILL = {int: 0, float: 0.0, bool: False} + + +def _scalar_column(column): + """One column of exact scalars as ``(bridge column, bridge null flags or None)`` for ``VertexBatcher.createVerticesColumns`` and + ``EdgeBatcher.newEdgesColumns``, or None when the column is not one the typed route carries. + + Carried: every value of one exact type among ``int`` (in 64 bits, a ``long[]``), ``float`` (finite, no subnormal, a ``double[]``), + ``bool`` (a ``boolean[]``) and ``str`` (a ``StringColumn``, with its own null rows), with ``None`` for a null; a column of nothing + but ``None``. Anything else (mixed kinds, a bool beside ints, a subclass, a numpy scalar, a list, a date, a lone surrogate, an + integer beyond 64 bits, a NaN, an infinity or a subnormal) is left to the JSON text path, which stores or refuses it as before. + """ + n = len(column) + kinds = set(map(type, column)) + has_none = _NONE_KIND in kinds + kinds.discard(_NONE_KIND) + if not kinds: + return jpype.JArray(jpype.JBoolean)(_np.zeros(n, dtype=_np.bool_)), ( + jpype.JArray(jpype.JBoolean)(_np.ones(n, dtype=_np.bool_)) + ) + if len(kinds) != 1: + return None + (kind,) = kinds + if kind is str: + from .core import _string_column + + built = _string_column(column) + return None if built is None else (built, None) + if kind not in _FILL: + return None + mask = None + if has_none: + mask = _np.fromiter( + map(operator.is_, column, itertools.repeat(None)), dtype=_np.bool_, count=n + ) + fill = _FILL[kind] + column = [fill if value is None else value for value in column] + if kind is int: + try: + built = jpype.JArray(jpype.JLong)(_np.array(column, dtype=_np.int64)) + except OverflowError: + return None + elif kind is float: + array = _np.array(column, dtype=_np.float64) + if not _np.isfinite(array).all(): + return None + nonzero = array[array != 0.0] + if (_np.abs(nonzero) < _FLOAT_NORMAL_MIN).any(): + return None + built = jpype.JArray(jpype.JDouble)(array) + else: + built = jpype.JArray(jpype.JBoolean)(_np.array(column, dtype=_np.bool_)) + return built, None if mask is None else jpype.JArray(jpype.JBoolean)(mask) + + +def _scalar_columns(rows, reserved=()): + """The columns of ``rows`` for the typed route as ``(names, columns, null flags)`` (lists, one entry per property), or None. + + Carried: every row an exact ``dict`` with the same keys in the same order (str names, none of ``reserved``) and each column one + that ``_scalar_column`` carries. A row that is None or has other keys, another key order or any other value keeps the JSON text + path, which writes each row's own keys in its own order. Needs numpy, as every other typed route of the bindings does. + """ + if _np is None or not rows or set(map(type, rows)) != {dict}: + return None + shapes = set(map(tuple, rows)) + if len(shapes) != 1: + return None + (names,) = shapes + if not set(map(type, names)) <= {str} or any(name in reserved for name in names): + return None + columns = [] + nulls = [] + for name in names: + built = _scalar_column(list(map(operator.itemgetter(name), rows))) + if built is None: + return None + columns.append(built[0]) + nulls.append(built[1]) + return list(names), columns, nulls + + +def _bridge_columns(names, columns, nulls): + """``(String[], Object[], Object[] or None)`` for the bridge: the null flags cross only when some column has a null.""" + flags = ( + jpype.JArray(jpype.JObject)(nulls) + if any(flag is not None for flag in nulls) + else None + ) + return ( + jpype.JArray(jpype.JString)(names), + jpype.JArray(jpype.JObject)(columns), + flags, + ) + + +def _rids_from_pairs(packed): + """The RID texts (``"#12:345"``, what ``str(rid)`` gives) of a ``long[]`` of ``[bucket id, position]`` pairs.""" + numbers = _np.asarray(packed, dtype=_np.int64) + return list(map("#{}:{}".format, numbers[0::2].tolist(), numbers[1::2].tolist())) + class GraphBatch: """Wrapper for Java GraphBatch with builder-backed configuration. @@ -205,6 +311,14 @@ def create_vertices( per-element path stores it. A list of only numbers keeps the property-matrix path, which crosses it as one double[]. + Rows that are all dicts with the same keys in the same order, each property + holding one kind of ``int``, ``float``, ``bool`` or ``str`` (``None`` for a + null), plus the float vectors above, cross as one typed column per property, + and the RIDs come back as one ``long[]``: no JSON text on either side. Any + other rows (a row that is ``None``, rows with other keys, a list, dict, + date-time, date, Decimal or bytes value, a column of mixed kinds) cross as the + JSON text described above. The stored values are the same on both routes. + Large inputs cross in chunks of 100,000 rows on this build. On the native build (the ``arcadedb`` package) the chunk is the batch's ``commit_every``, 1000 unless ``graph_batch()`` got another value, and each chunk is its own transaction. Chunks that already went in stay stored if a later chunk fails. @@ -314,12 +428,28 @@ def dumps(chunk_rows): rids = [] for start in range(0, len(rows), step): # the previous chunk's rest rows, text and matrices go before this chunk's are built: two float[] buffers were alive at once - split = rest = text = None + split = rest = text = typed = names = columns = nulls = packed = None chunk = rows[start : start + step] split = split_float_vectors( self._java_db, type_name, chunk, declared, undeclared=True ) rest = split[0] if split else chunk + typed = _scalar_columns(rest) + if typed is not None: + # rows of exact scalars and float vectors: one typed column per property, the RIDs back as one long[] + names, columns, nulls = typed + if split: + names = names + [str(name) for name in split[1]] + columns = columns + list(split[2]) + nulls = nulls + [None] * len(split[2]) + packed = vertex_batcher.createVerticesColumns( + self._java_graph_batch, + type_name, + *_bridge_columns(names, columns, nulls), + len(chunk), + ) + rids.extend(_rids_from_pairs(packed)) + continue text = None if declared_scalars_ok(rest, declared): try: @@ -448,6 +578,9 @@ def new_edges( return self try: + if properties is not None and not any(properties): + # no edge has a property: the same call as no properties at all (a plain newEdge per edge) + properties = None if properties is None: # One joined string crosses the boundary in a single bulk # copy; a Python list of N strings would be converted @@ -464,6 +597,23 @@ def new_edges( ) return self + typed = self._typed_edge_columns(properties) + if typed is not None: + # properties of exact scalars: the RIDs cross as the joined texts above, the values as typed columns, one per property + sources = [str(r) for r in source_rids] + dests = [str(r) for r in destination_rids] + step = self._BULK_CHUNK + for k, (names, columns, nulls) in enumerate(typed): + start = k * step + edge_batcher.newEdgesColumns( + self._java_graph_batch, + ";".join(sources[start : start + step]), + edge_type, + ";".join(dests[start : start + step]), + *_bridge_columns(names, columns, nulls), + ) + return self + rows = [] tagged = False # a value that needs the tagged text; False keeps the untagged code of before for src, dst, p in zip(source_rids, destination_rids, properties): @@ -521,6 +671,23 @@ def new_edges( f"Failed to buffer batch edges '{edge_type}': {e}" ) from e + def _typed_edge_columns(self, properties): + """The typed columns of every chunk of the edge ``properties``, or None when any chunk is not carried. + + Every chunk is built before the first crosses, so a chunk the typed route cannot carry still sends the whole call down the + JSON text path with nothing buffered (as the tagged text is written before it crosses). A property named ``_src`` or ``_dst`` + stays on that path, where the rows hold the endpoints under those names. + """ + chunks = [] + for start in range(0, len(properties), self._BULK_CHUNK): + columns = _scalar_columns( + properties[start : start + self._BULK_CHUNK], ("_src", "_dst") + ) + if columns is None: + return None + chunks.append(columns) + return chunks + def flush(self) -> "GraphBatch": """Flush buffered edges to disk. 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..ab4c254 100644 --- a/src/java/com/arcadedb/python/DocumentBatcher.java +++ b/src/java/com/arcadedb/python/DocumentBatcher.java @@ -81,17 +81,17 @@ public static long insertManyJson(final Database db, final String typeName, fina /** * Columnar insert: the documents are built Java-side from whole columns, so each column crosses the FFI once (a long[] or - * double[] copied from a numpy buffer, a boolean[], a StringColumn (one UTF-8 byte[] plus the int[] offsets), or an Object[] of - * boxed numbers, Strings of a mixed column, and nulls) instead of one + * double[] copied from a numpy buffer, a boolean[], a StringColumn (one UTF-8 byte[] plus the int[] offsets), a TemporalColumn (one + * long[] of epoch ticks), or an Object[] of boxed numbers, Strings of a mixed column, and nulls) instead of one * JSON text per batch (bindings issue #150: 2.24x over insertManyJson on the first * 2,000,000 TPC-H SF1 line items, nine typed properties, commit every 10,000, same sums and count). * * Same failure contract as insertManyJson (#7882): batch commits and the failure rollback apply only to a transaction * opened here; a caller's own transaction is the caller's to commit or roll back. * - * @param columns one array per name, each of length n: long[], double[], boolean[], Object[], a FloatVectors.Matrix or a - * StringColumn; a null element (or null row of a StringColumn) sets the property to null, as a JSON null does - * on the insertManyJson path + * @param columns one array per name, each of length n: long[], double[], boolean[], Object[], a FloatVectors.Matrix, a + * StringColumn or a TemporalColumn; a null element (or null row of a StringColumn or TemporalColumn) sets the + * property to null, as a JSON null does on the insertManyJson path */ public static long insertColumns(final Database db, final String typeName, final String[] names, final Object[] columns, final int n, final int commitEvery) { @@ -210,7 +210,7 @@ private static MutableDocument newRecord(final Database db, final String typeNam return vertex ? db.newVertex(typeName) : db.newDocument(typeName); } - private static void checkColumns(final String[] names, final Object[] columns, final int n) { + static void checkColumns(final String[] names, final Object[] columns, final int n) { final int k = names.length; if (columns.length != k) throw new IllegalArgumentException("columns has " + columns.length + " arrays for " + k + " names"); @@ -229,9 +229,11 @@ else if (col instanceof FloatVectors.Matrix m) len = m.rows(); else if (col instanceof StringColumn s) len = s.rows(); + else if (col instanceof TemporalColumn t) + len = t.rows(); else throw new IllegalArgumentException("column '" + names[c] + "' is a " + (col == null ? "null" : col.getClass().getName()) - + ", not a long[], double[], boolean[], Object[], FloatVectors.Matrix or StringColumn"); + + ", not a long[], double[], boolean[], Object[], FloatVectors.Matrix, StringColumn or TemporalColumn"); if (len != n) throw new IllegalArgumentException("column '" + names[c] + "' has " + len + " values for " + n + " rows"); } @@ -251,6 +253,8 @@ else if (col instanceof FloatVectors.Matrix m) v = m.row(i); else if (col instanceof StringColumn s) v = s.get(i); + else if (col instanceof TemporalColumn t) + v = t.get(i); else v = ((Object[]) col)[i]; doc.set(names[c], v); diff --git a/src/java/com/arcadedb/python/EdgeBatcher.java b/src/java/com/arcadedb/python/EdgeBatcher.java index e215296..dd3ab52 100644 --- a/src/java/com/arcadedb/python/EdgeBatcher.java +++ b/src/java/com/arcadedb/python/EdgeBatcher.java @@ -42,6 +42,31 @@ public static void newEdgesJoined(final GraphBatch batch, final String joinedSou newEdges(batch, sources, edgeType, destinations); } + /** + * Bulk edges WITH properties as whole columns: the RIDs cross as the two joined texts of {@link #newEdgesJoined} and the property + * values as typed columns (see {@link VertexBatcher#createVerticesColumns}, which has the layout of {@code columns} and + * {@code nulls} and the boxing of an integer), one value per edge in each. Every edge gets the same property names, in the order of + * {@code names}; an empty {@code names} buffers edges without properties. The engine call is the one {@link #newEdgesJson} makes. + */ + public static void newEdgesColumns(final GraphBatch batch, final String joinedSourceRids, final String edgeType, + final String joinedDestinationRids, final String[] names, final Object[] columns, final Object[] nulls) { + final String[] sources = joinedSourceRids.split(";"); + final String[] destinations = joinedDestinationRids.split(";"); + if (sources.length != destinations.length) + throw new IllegalArgumentException( + "sourceRids and destinationRids must have the same length (" + sources.length + " vs " + destinations.length + ")"); + final int n = sources.length; + DocumentBatcher.checkColumns(names, columns, n); + VertexBatcher.checkNulls(names, nulls, n); + for (int i = 0; i < n; i++) { + final Object[] props = VertexBatcher.flatRow(names, columns, nulls, i); + if (props.length == 0) + batch.newEdge(new RID(sources[i]), edgeType, new RID(destinations[i])); + else + batch.newEdge(new RID(sources[i]), edgeType, new RID(destinations[i]), props); + } + } + /** * Bulk edges WITH properties: one JSON array of rows, each * {"_src": "#1:0", "_dst": "#1:1", ...properties}. Property values are JSON values and the tagged values 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/src/java/com/arcadedb/python/VertexBatcher.java b/src/java/com/arcadedb/python/VertexBatcher.java index 5943f37..1c15c73 100644 --- a/src/java/com/arcadedb/python/VertexBatcher.java +++ b/src/java/com/arcadedb/python/VertexBatcher.java @@ -62,4 +62,71 @@ public static String createVerticesJson(final GraphBatch batch, final String typ } return sb.toString(); } + + /** + * Typed vertex creation: the rows arrive as whole columns (the layout of {@link DocumentBatcher#insertColumns}: a long[], a + * double[], a boolean[], a StringColumn, a FloatVectors.Matrix, ...) and cross once per column instead of as one JSON text, and the + * created RIDs go back as one long[] of 2 * n numbers, {@code [bucket id, position]} per vertex in row order, instead of one joined + * text. The engine call is the one {@link #createVerticesJson} makes, with the same key and value per property, so the stored + * records are the same: {@code names[c]} is the property of column c, in the order of the rows' keys; {@code nulls}, or its element + * c, is null when the column has no null, else one flag per row (a flagged row is set to null, as a JSON null is on the text path). + * A long that fits an int is boxed as an Integer and any other as a Long, because that is what the engine's JSON parser makes of + * the number token, and the text path stored it so. + */ + public static long[] createVerticesColumns(final GraphBatch batch, final String typeName, final String[] names, + final Object[] columns, final Object[] nulls, final int n) { + DocumentBatcher.checkColumns(names, columns, n); + checkNulls(names, nulls, n); + final Object[][] matrix = new Object[n][]; + for (int i = 0; i < n; i++) + matrix[i] = flatRow(names, columns, nulls, i); + final RID[] rids = batch.createVertices(typeName, matrix); + final long[] packed = new long[rids.length * 2]; + for (int i = 0; i < rids.length; i++) { + packed[2 * i] = rids[i].getBucketId(); + packed[2 * i + 1] = rids[i].getPosition(); + } + return packed; + } + + /** One flag array per name or none at all, each as long as the rows. */ + static void checkNulls(final String[] names, final Object[] nulls, final int n) { + if (nulls == null) + return; + if (nulls.length != names.length) + throw new IllegalArgumentException("nulls has " + nulls.length + " entries for " + names.length + " names"); + for (int c = 0; c < nulls.length; c++) + if (nulls[c] != null && (!(nulls[c] instanceof boolean[] flags) || flags.length != n)) + throw new IllegalArgumentException("the null flags of '" + names[c] + "' are not a boolean[] of " + n + " rows"); + } + + /** Row {@code i} as the engine's property varargs: name, value, name, value, ... (empty for a row without properties). */ + static Object[] flatRow(final String[] names, final Object[] columns, final Object[] nulls, final int i) { + final Object[] flat = new Object[names.length * 2]; + for (int c = 0; c < names.length; c++) { + flat[2 * c] = names[c]; + flat[2 * c + 1] = cell(columns[c], nulls == null ? null : (boolean[]) nulls[c], i); + } + return flat; + } + + private static Object cell(final Object column, final boolean[] nulls, final int i) { + if (nulls != null && nulls[i]) + return null; + if (column instanceof long[] a) { + final long v = a[i]; + return v >= Integer.MIN_VALUE && v <= Integer.MAX_VALUE ? (Object) Integer.valueOf((int) v) : (Object) Long.valueOf(v); + } + if (column instanceof double[] a) + return Double.valueOf(a[i]); + if (column instanceof boolean[] a) + return Boolean.valueOf(a[i]); + if (column instanceof StringColumn s) + return s.get(i); + if (column instanceof FloatVectors.Matrix m) + return m.row(i); + if (column instanceof TemporalColumn t) + return t.get(i); + return ((Object[]) column)[i]; + } } diff --git a/tests/conftest.py b/tests/conftest.py index 32ddf8f..dd9dbd1 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") @@ -147,12 +147,15 @@ "test_insert_many_gc_pause.py": "the cyclic collector pause is the JVM build's (the native insert_many allocates nothing the collector tracks)", "test_graph_batch_tagged_shapes.py": "the JVM wheel's graph batch bulk path (bridge VertexBatcher and EdgeBatcher, the property-matrix fallback)", "test_graph_batch_single_pass.py": "the JVM wheel's graph batch bulk path (GraphBatch._create_vertices_json_bulk, the property-matrix fallback)", + "test_graph_batch_typed_jvm.py": "the JVM wheel's typed graph batch route (bridge VertexBatcher.createVerticesColumns and EdgeBatcher.newEdgesColumns, JPype crossings, the Java classes of stored values)", "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)", } if BACKEND == "native" and not os.environ.get("NATIVE_PARITY_ALL"): 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_graph_batch_typed_jvm.py b/tests/test_graph_batch_typed_jvm.py new file mode 100644 index 0000000..648af0c --- /dev/null +++ b/tests/test_graph_batch_typed_jvm.py @@ -0,0 +1,478 @@ +"""GraphBatch.create_vertices and new_edges (the JVM wheel): rows of exact scalars and float vectors cross as typed columns, not as JSON text (#388). + +Before, every chunk of rows crossed as one JSON text that the engine parsed into a JSONObject per row, and the created RIDs came back as one +';'-joined text. Now rows that are all dicts with the same keys in the same order, each property one kind of int, float, bool or str (None +for a null), and the float vectors of a declared or float32 property, cross as one typed column per property +(``VertexBatcher.createVerticesColumns``, ``EdgeBatcher.newEdgesColumns``), and the RIDs of the vertices come back as one ``long[]`` of +bucket id and position pairs. Every other shape keeps the JSON text path (or the property-matrix path), which these tests also pin. + +The parity tests store the same rows twice, once through the typed route and once with it switched off (``_scalar_columns`` answering None, +which is what every other shape gets), and compare what the engine holds: the property names in order, the Java class of each value and +its text. The JPype crossings are counted by wrapping the bridge classes. The JPype bridge does not exist on the native wheel, so +conftest.py does not collect this file there. +""" + +import jpype +import numpy as np +import pytest + +from arcadedb import graph_batch as gbm +from arcadedb import results as results_module +from arcadedb.graph_batch import GraphBatch + +CHUNK = 100 # rows per crossing in these tests: 250 rows are three crossings + + +class _Counting: + """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 bridge calls of a load, by class, and the property-matrix fallbacks.""" + log = {"vertex": [], "edge": [], "matrix": []} + real_jclass = jpype.JClass + + def jclass(name, *args, **kwargs): + cls = real_jclass(name, *args, **kwargs) + if name == "com.arcadedb.python.VertexBatcher": + return _Counting(cls, log["vertex"]) + return cls + + monkeypatch.setattr(jpype, "JClass", jclass) + real_bridge_class = results_module._bridge_class + + def bridge_class(name): + cls = real_bridge_class(name) + if name == "EdgeBatcher" and cls is not None: + return _Counting(cls, log["edge"]) + return cls + + monkeypatch.setattr(results_module, "_bridge_class", bridge_class) + real_matrix = GraphBatch._to_java_property_matrix + + def matrix(self, rows): + log["matrix"].append(len(rows)) + return real_matrix(self, rows) + + monkeypatch.setattr(GraphBatch, "_to_java_property_matrix", matrix) + monkeypatch.setattr(GraphBatch, "_BULK_CHUNK", CHUNK) + return log + + +@pytest.fixture +def db(temp_db): + for name in ("A", "B", "VA", "VB", "VC"): + temp_db.command("sql", f"CREATE VERTEX TYPE {name}") + for name in ("EA", "EB"): + temp_db.command("sql", f"CREATE EDGE TYPE {name}") + for name in ("A", "B"): + temp_db.command("sql", f"CREATE PROPERTY {name}.embedding ARRAY_OF_FLOATS") + return temp_db + + +def _plain(value): + """A stored value as (Java class, text), or the list of the numbers of a float[]; None stays None.""" + if value is None: + return None + get_class = getattr(value, "getClass", None) + if get_class is None: # JPype handed over a Python value + return (type(value).__name__, repr(value)) + name = str(get_class().getName()) + if name.startswith("["): + return (name, [float(x) for x in value]) + return (name, str(value)) + + +def _stored(db, type_name): + """Every record of the type as its (property name, value) pairs in the order the engine holds them, sorted by id.""" + iterator = db._java_db.iterateType(type_name, False) + rows = [] + while iterator.hasNext(): + record = iterator.next().asDocument() + rows.append( + [(str(n), _plain(record.get(n))) for n in record.getPropertyNames()] + ) + rows.sort(key=lambda pairs: int(dict(pairs).get("id", (None, "-1"))[1])) + return rows + + +INTS = [ + 0, + 1, + -1, + 2**31 - 1, + 2**31, + -(2**31), + -(2**31) - 1, + 2**53 + 1, + 2**63 - 1, + -(2**63), +] +_RNG = np.random.default_rng(11) +FLOATS = ( + [ + 0.0, + -0.0, + 1.5, + 0.1, + 1 / 3, + 1e300, + -1e-300, + 2.2250738585072014e-308, + 1e16, + 123456789.123456789, + ] + + [float(_RNG.uniform(-1e6, 1e6)) for _ in range(10)] + + [float(_RNG.random() ** 7) for _ in range(10)] +) +STRS = [ + "", + "a", + "héllo", + "\U0001f600 emoji", + "x" * 1000, + 'quote"s \\ back', + "nul\x00mid", + "line\nbreak", +] + + +def _scalar_rows(n=40): + rows = [] + for i in range(n): + rows.append( + { + "id": i, + "i": INTS[i % len(INTS)], + "f": FLOATS[i % len(FLOATS)], + "s": STRS[i % len(STRS)], + "b": i % 3 == 0, + "ni": None if i % 4 == 0 else i * 7, + "nf": None if i % 5 == 0 else i / 8, + "ns": None if i % 6 == 0 else f"v{i}", + "nb": None if i % 7 == 0 else i % 2 == 0, + "none": None, + } + ) + return rows + + +def _load(db, type_name, rows, **options): + with db.graph_batch(parallel_flush=False, **options) as gb: + return gb.create_vertices(type_name, rows) + + +# --------------------------------------------------------------------------- +# crossings: no JSON text for the shapes the typed route carries + + +def test_scalar_rows_cross_as_typed_columns_one_call_per_chunk(db, bridge): + rows = [ + {"id": i, "name": f"n{i}", "age": i % 90, "score": i / 4} for i in range(250) + ] + rids = _load(db, "A", rows) + assert len(rids) == 250 + assert bridge["vertex"] == ["createVerticesColumns"] * 3 # 100, 100, 50 + assert bridge["matrix"] == [] + + +def test_scalar_rows_with_a_vector_cross_as_typed_columns(db, bridge): + x = np.random.default_rng(3).standard_normal((250, 8)).astype(np.float32) + rows = [ + {"id": i, "views": i % 9, "embedding": jpype.JArray(jpype.JFloat)(x[i])} + for i in range(250) + ] + assert len(_load(db, "A", rows)) == 250 + assert bridge["vertex"] == ["createVerticesColumns"] * 3 + assert bridge["matrix"] == [] + + +def test_an_empty_batch_and_a_batch_of_one(db, bridge): + assert _load(db, "A", []) == [] + assert bridge["vertex"] == [] + one = _load(db, "A", [{"id": 1, "x": "y"}]) + assert len(one) == 1 and one[0].startswith("#") + assert bridge["vertex"] == ["createVerticesColumns"] + + +def test_rows_without_properties_cross_as_zero_columns(db, bridge): + rids = _load(db, "A", [{} for _ in range(7)]) + assert len(rids) == 7 and len(set(rids)) == 7 + assert bridge["vertex"] == ["createVerticesColumns"] + + +# --------------------------------------------------------------------------- +# the RIDs that come back + + +def test_the_rids_are_the_texts_in_row_order(db, bridge): + rows = [{"id": i} for i in range(250)] + rids = _load(db, "A", rows) + assert bridge["vertex"] == ["createVerticesColumns"] * 3 + assert all(type(rid) is str and rid.startswith("#") for rid in rids) + assert len(set(rids)) == 250 + # the engine's own lookup of each returned text: row i must be the vertex that was created from rows[i] + rid_class = jpype.JClass("com.arcadedb.database.RID") + ids = [ + int(str(db._java_db.lookupByRID(rid_class(rid), True).asDocument().get("id"))) + for rid in rids + ] + assert ids == list(range(250)) + + +def test_the_rids_equal_those_of_the_old_route(db, bridge, monkeypatch): + rows = [{"id": i, "s": f"v{i}"} for i in range(250)] + typed = _load(db, "A", rows) + monkeypatch.setattr(gbm, "_scalar_columns", lambda rows, reserved=(): None) + old = _load(db, "B", rows) + assert ( + bridge["vertex"] == ["createVerticesColumns"] * 3 + ["createVerticesJson"] * 3 + ) + assert len(typed) == len(old) == 250 and len(set(typed)) == len(set(old)) == 250 + assert all(type(rid) is str and rid.startswith("#") for rid in typed + old) + + +# --------------------------------------------------------------------------- +# parity: what the engine holds is what the old route stored + + +def test_every_scalar_kind_is_stored_as_the_old_route_stores_it( + db, bridge, monkeypatch +): + rows = _scalar_rows() + _load(db, "A", rows) + assert bridge["vertex"] == ["createVerticesColumns"] + monkeypatch.setattr(gbm, "_scalar_columns", lambda rows, reserved=(): None) + _load(db, "B", rows) + assert bridge["vertex"] == ["createVerticesColumns", "createVerticesJson"] + typed, old = _stored(db, "A"), _stored(db, "B") + assert len(typed) == len(old) == len(rows) + assert typed == old + + +def test_a_float_vector_among_scalars_is_stored_as_the_old_route_stores_it( + db, bridge, monkeypatch +): + x = np.random.default_rng(5).standard_normal((60, 8)).astype(np.float32) + rows = [ + { + "id": i, + "views": i % 11, + "embedding": jpype.JArray(jpype.JFloat)(x[i]), + "note": None if i % 3 else "n", + } + for i in range(60) + ] + _load(db, "A", rows) + monkeypatch.setattr(gbm, "_scalar_columns", lambda rows, reserved=(): None) + _load(db, "B", rows) + assert bridge["vertex"] == ["createVerticesColumns", "createVerticesJson"] + typed, old = _stored(db, "A"), _stored(db, "B") + assert len(typed) == 60 and typed == old + assert dict(typed[3])["embedding"][1] == [float(v) for v in x[3]] + + +def test_a_batch_of_one_and_a_load_over_the_chunk_boundary_agree( + db, bridge, monkeypatch +): + rows = _scalar_rows(250) + _load(db, "A", rows) + _load(db, "VA", rows[:1]) + monkeypatch.setattr(gbm, "_scalar_columns", lambda rows, reserved=(): None) + _load(db, "B", rows) + _load(db, "VB", rows[:1]) + assert _stored(db, "A") == _stored(db, "B") + assert _stored(db, "VA") == _stored(db, "VB") + + +def test_a_null_and_a_missing_key_stay_apart(db, bridge): + # rows with other keys keep the JSON text path: a missing key is no property, a None is a null, on both routes + rows = [{"id": 0, "x": None}, {"id": 1}, {"id": 2, "x": 5}] + _load(db, "A", rows) + assert bridge["vertex"] == ["createVerticesJson"] + stored = _stored(db, "A") + assert len(stored) == 3 + assert "x" not in dict(stored[1]) + assert dict(stored[2])["x"][1] == "5" + + +# --------------------------------------------------------------------------- +# the shapes the typed route keeps off, and where they go instead + + +@pytest.mark.parametrize( + "rows, route", + [ + ([{"id": 0, "a": 1}, {"id": 1}], "createVerticesJson"), # rows with other keys + ( + [{"id": 0, "a": 1, "b": 2}, {"id": 1, "b": 2, "a": 1}], + "createVerticesJson", + ), # another key order + ([{"id": 0}, None], "createVerticesJson"), # a row that is None + ([{"id": i, "l": ["a", i]} for i in range(3)], "createVerticesJson"), # a list + ([{"id": i, "m": {"k": i}} for i in range(3)], "createVerticesJson"), # a dict + ( + [{"id": 0, "x": 1}, {"id": 1, "x": 2.5}], + "createVerticesJson", + ), # ints beside floats + ( + [{"id": 0, "x": 1}, {"id": 1, "x": True}], + "createVerticesJson", + ), # an int beside a bool + ( + [{"id": 0, "x": np.int64(4)}], + None, + ), # a numpy scalar: the property-matrix path + ([{"id": 0, "x": float("nan")}], None), # a NaN + ([{"id": 0, "x": 5e-324}], None), # a subnormal + ], +) +def test_a_shape_the_typed_route_cannot_carry_keeps_the_old_routes( + db, bridge, rows, route +): + rids = _load(db, "A", rows) + assert len(rids) == len(rows) + assert "createVerticesColumns" not in bridge["vertex"] + if route is None: + assert bridge["vertex"] == [] and bridge["matrix"] == [len(rows)] + else: + assert bridge["vertex"] == [route] + + +def test_a_chunk_that_is_not_carried_does_not_move_the_others(db, bridge): + rows = [{"id": i, "v": i} for i in range(250)] + rows[120] = { + "id": 120, + "v": 2.5, + } # chunk 2 holds an int column with a float: that chunk keeps the text + rids = _load(db, "A", rows) + assert len(rids) == 250 + assert bridge["vertex"] == [ + "createVerticesColumns", + "createVerticesJson", + "createVerticesColumns", + ] + stored = {r["id"]: r["v"] for r in db.query("sql", "SELECT id, v FROM A").to_list()} + assert stored[119] == 119 and stored[120] == 2.5 and stored[121] == 121 + + +# --------------------------------------------------------------------------- +# edges + + +def _edge_setup(db, n=250): + vertices = _load(db, "VC", [{"id": i} for i in range(n)]) + return vertices + + +def _edge_props(n): + return [ + { + "id": i, + "w": INTS[i % len(INTS)], + "f": FLOATS[i % len(FLOATS)], + "s": STRS[i % len(STRS)], + "b": i % 3 == 0, + "n": None if i % 4 == 0 else i, + "z": None, + } + for i in range(n) + ] + + +def _new_edges(db, edge_type, sources, destinations, properties=None): + with db.graph_batch(parallel_flush=False) as gb: + gb.new_edges(sources, edge_type, destinations, properties) + + +def test_edge_properties_of_scalars_cross_as_typed_columns(db, bridge): + vertices = _edge_setup(db) + del bridge["vertex"][:] + sources, destinations = vertices, vertices[1:] + vertices[:1] + _new_edges(db, "EA", sources, destinations, _edge_props(250)) + assert bridge["edge"] == ["newEdgesColumns"] * 3 # 100, 100, 50 + assert len(_stored(db, "EA")) == 250 + + +def test_edges_without_a_property_value_in_any_dict_use_the_joined_call(db, bridge): + vertices = _edge_setup(db, 12) + _new_edges(db, "EA", vertices, vertices[1:] + vertices[:1], [{} for _ in vertices]) + _new_edges( + db, "EB", vertices, vertices[1:] + vertices[:1], [None for _ in vertices] + ) + assert bridge["edge"] == ["newEdgesJoined", "newEdgesJoined"] + assert db.query("sql", "SELECT count(*) AS n FROM EA").to_list()[0]["n"] == 12 + assert db.query("sql", "SELECT count(*) AS n FROM EB").to_list()[0]["n"] == 12 + + +def test_edge_properties_are_stored_as_the_old_route_stores_them( + db, bridge, monkeypatch +): + vertices = _edge_setup(db, 250) + sources, destinations = vertices, vertices[1:] + vertices[:1] + properties = _edge_props(250) + _new_edges(db, "EA", sources, destinations, properties) + monkeypatch.setattr(gbm, "_scalar_columns", lambda rows, reserved=(): None) + _new_edges(db, "EB", sources, destinations, properties) + assert bridge["edge"] == ["newEdgesColumns"] * 3 + ["newEdgesJson"] * 3 + typed, old = _stored(db, "EA"), _stored(db, "EB") + assert len(typed) == len(old) == 250 + assert typed == old + + +def test_a_batch_of_one_edge(db, bridge, monkeypatch): + vertices = _edge_setup(db, 2) + _new_edges(db, "EA", vertices[:1], vertices[1:], [{"id": 1, "w": 5}]) + monkeypatch.setattr(gbm, "_scalar_columns", lambda rows, reserved=(): None) + _new_edges(db, "EB", vertices[:1], vertices[1:], [{"id": 1, "w": 5}]) + assert bridge["edge"] == ["newEdgesColumns", "newEdgesJson"] + assert _stored(db, "EA") == _stored(db, "EB") and len(_stored(db, "EA")) == 1 + + +@pytest.mark.parametrize( + "properties", + [ + [{"id": 0, "a": 1}, {"id": 1}, {"id": 2, "a": 3}], # dicts with other keys + [{"id": 0}, None, {"id": 2}], # a None among dicts + [{"id": i, "l": ["a", i]} for i in range(3)], # a list + [ + {"id": 0, "x": 1}, + {"id": 1, "x": 2.5}, + {"id": 2, "x": 3}, + ], # ints beside floats + [ + {"id": i, "_src": "kept"} for i in range(3) + ], # a property named like a row's endpoint + ], +) +def test_edge_shapes_the_typed_route_cannot_carry_keep_the_old_routes( + db, bridge, properties +): + vertices = _edge_setup(db, 3) + _new_edges(db, "EA", vertices, vertices[1:] + vertices[:1], properties) + assert "newEdgesColumns" not in bridge["edge"] + assert len(_stored(db, "EA")) == 3 + + +def test_an_edge_chunk_that_is_not_carried_sends_the_whole_call_the_old_way(db, bridge): + vertices = _edge_setup(db, 250) + properties = [{"id": i, "v": i} for i in range(250)] + properties[120] = {"id": 120, "v": 2.5} + _new_edges(db, "EA", vertices, vertices[1:] + vertices[:1], properties) + assert "newEdgesColumns" not in bridge["edge"] + assert len(_stored(db, "EA")) == 250 diff --git a/tests/test_graph_batch_vector_crossings_jvm.py b/tests/test_graph_batch_vector_crossings_jvm.py index a86b89b..ae000c2 100644 --- a/tests/test_graph_batch_vector_crossings_jvm.py +++ b/tests/test_graph_batch_vector_crossings_jvm.py @@ -5,7 +5,7 @@ builds one ``Object[]`` per row and one ``Object[][]`` for the call (an unwrapped Python loop that was 59 to 61 percent of the wall of the e2 and l3d loads in the untimed ingest profile of 26.11.1). A float32 numpy array on a property that is not declared ``ARRAY_OF_FLOATS`` took the same way. Now ``take_vector_columns`` stacks the column once (numpy reads a Java array through the buffer protocol), one -``float[]`` copy and one ``FloatVectors.Matrix`` cross per property, and one ``VertexBatcher.createVerticesJson`` call per chunk. +``float[]`` copy and one ``FloatVectors.Matrix`` cross per property, and one ``VertexBatcher.createVerticesColumns`` call per chunk. The JPype crossings are counted by wrapping the bridge classes and the Python-side conversion; the stored values are in test_vertex_vector_loads.py. The JPype bridge does not exist on the native wheel (its counts are in test_native_vertex_vector_crossings.py), @@ -129,7 +129,7 @@ def test_a_float32_vector_crosses_as_one_matrix_per_chunk( ), "the property-matrix path ran (a JPype conversion per value and per row)" assert crossings["convert"] == [], "a value was converted one at a time" assert ( - crossings["bridge"] == ["createVerticesJson"] * 3 + crossings["bridge"] == ["createVerticesColumns"] * 3 ) # one crossing per chunk of 100 # one float32 matrix per chunk: `declared` is asked once per call, `Matrix` once per chunk assert crossings["matrix"].count("Matrix") == 3 @@ -148,7 +148,7 @@ def test_two_vector_properties_cross_as_two_matrices_per_chunk(db, crossings): with db.graph_batch(parallel_flush=False) as gb: gb.create_vertices("Declared", rows) assert crossings["fallback"] == [] and crossings["convert"] == [] - assert crossings["bridge"] == ["createVerticesJson"] * 3 + assert crossings["bridge"] == ["createVerticesColumns"] * 3 assert crossings["matrix"].count("Matrix") == 6 @@ -162,7 +162,7 @@ def test_a_ragged_column_still_takes_the_property_matrix_path_for_its_chunk( with db.graph_batch(parallel_flush=False) as gb: assert len(gb.create_vertices("Declared", rows)) == ROWS assert crossings["fallback"] == [100], "only the chunk of row 120" - assert crossings["bridge"] == ["createVerticesJson"] * 2 + assert crossings["bridge"] == ["createVerticesColumns"] * 2 stored = { r["id"]: r["embedding"] for r in db.query("sql", "SELECT id, embedding FROM Declared").to_list() @@ -180,7 +180,7 @@ def test_a_java_double_array_is_a_vector_only_on_a_declared_property(db, crossin gb.create_vertices("Declared", rows) assert ( crossings["fallback"] == [] - and crossings["bridge"] == ["createVerticesJson"] * 3 + and crossings["bridge"] == ["createVerticesColumns"] * 3 ) declared = { r["id"]: r["embedding"] @@ -255,7 +255,7 @@ def test_rows_with_vectors_cross_in_blocks_sized_by_their_width( assert len(gb.create_vertices("Declared", rows)) == ROWS assert crossings["fallback"] == [] and crossings["convert"] == [] assert ( - crossings["bridge"] == ["createVerticesJson"] * 7 + crossings["bridge"] == ["createVerticesColumns"] * 7 ) # 6 blocks of 40, one of 10 assert crossings["matrix"].count("Matrix") == 7 sql = "SELECT id, embedding FROM Declared" @@ -269,12 +269,12 @@ def test_rows_with_vectors_cross_in_blocks_sized_by_their_width( rows = [{"id": 1000 + i, "embedding": arrays[i]} for i in range(25)] with db.graph_batch(parallel_flush=False) as gb: gb.create_vertices("Free", rows) - assert crossings["bridge"] == ["createVerticesJson"] * 3 # 10, 10, 5 + assert crossings["bridge"] == ["createVerticesColumns"] * 3 # 10, 10, 5 # rows without a vector keep the whole _BULK_CHUNK crossings["bridge"].clear() with db.graph_batch(parallel_flush=False) as gb: gb.create_vertices("Free", [{"id": 2000 + i} for i in range(ROWS)]) - assert crossings["bridge"] == ["createVerticesJson"] + assert crossings["bridge"] == ["createVerticesColumns"] def test_vector_width_counts_the_vector_columns_of_the_first_row(db): diff --git a/tests/test_import_database.py b/tests/test_import_database.py index 9f1803f..613141d 100644 --- a/tests/test_import_database.py +++ b/tests/test_import_database.py @@ -67,10 +67,17 @@ def _resource_path(name: str) -> Path: return Path(__file__).resolve().parent / "fixtures" / name +# `parallel` is a reserved word in the SQL grammar, so an IMPORT DATABASE setting of that name can only be written in backticks. For a few +# days the 26.12.1-SNAPSHOT engine refused the quoted name as an unknown importer setting (ArcadeData/arcadedb#11363, fixed by #11441 for +# ArcadeData/arcadedb#11367); every engine accepts it now, as 26.11.1 does, so the two tests below assert OK on all of them. + + def _import_result_ok(result_set): assert result_set is not None row = result_set.one() - assert row.get("result") == "OK" + assert ( + row.get("result") == "OK" + ), row.to_dict() # the whole row: the importer says why on a FAIL return row @@ -188,7 +195,7 @@ def test_import_database_xml_vertices(temp_db_path, sample_xml_path): "sql", ( f"IMPORT DATABASE {_file_url(sample_xml_path)} " - "WITH objectNestLevel = 1, entityType = 'VERTEX'" + "WITH objectNestLevel = 1" ), ) _import_result_ok(result) 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