diff --git a/docs/modules/ROOT/pages/how-to/concurrency.adoc b/docs/modules/ROOT/pages/how-to/concurrency.adoc index 79ce818c..82a6afea 100644 --- a/docs/modules/ROOT/pages/how-to/concurrency.adoc +++ b/docs/modules/ROOT/pages/how-to/concurrency.adoc @@ -77,7 +77,7 @@ What the runs show: * Reading the value, computing in Python and writing it back lost updates under the default isolation level, also when every conflict was retried: with `run_in_transaction` 1,137 of 12,000 on the JVM build and 201 of 12,000 on the native build, with the own loop 2,883 and 331. In this probe the read is a `SELECT` and the write is a separate `UPDATE` with the computed value. * The same read-compute-write under `REPEATABLE_READ` with the own retry loop lost nothing: 0 lost updates in each of the 3 runs, one per build. -Based on that, do the arithmetic in the statement and retry the transaction. If the new value needs a computation that SQL cannot do, start the transaction with `BEGIN ISOLATION REPEATABLE_READ` and retry as in <>. The evidence for each is one run per build, as in the table. Do not read a value, compute in Python and write it back under the default isolation level. +Based on that, do the arithmetic in the statement and retry the transaction. If the new value needs a computation that SQL cannot do, start the transaction with `BEGIN ISOLATION REPEATABLE_READ` and retry as in <>. The evidence for each is one run per build, as in the table. Do not read a value, compute in Python and write it back under the default isolation level. The table does not cover the record object's `save()` on the native build: see <>. [source,python] ---- @@ -88,6 +88,26 @@ db.run_in_transaction( ) ---- +[[record-save]] +=== `save()` of a record object refuses a lost update + +`record.modify().set(...).save()` on a record that exists is a read-compute-write: the object holds the values it was read with, and `save()` writes the changed properties back. If another transaction changed the stored record after the object was read, that write used to overwrite the change without an error. The JVM build already refused it. The `arcadedb` (native) build now refuses it too: `save()` raises `ArcadeDBError` with `ConcurrentModificationException` in the message, and writes nothing. + +The engine keeps no record version, so the native build compares values. Inside the transaction of the update, `save()` reads the stored record and compares it with the values the object held as read. A record that differs is a lost update. A record that another transaction changed and then changed back is not detected. A `save()` of a new record has nothing to compare and is not checked. A second `save()` of the same object does not conflict with the first, because after a successful save the object holds the values it wrote. + +`save()` of an existing record is a write, so it needs an open transaction on both builds (`with db.transaction():`, or `begin()`). Outside one it raises `ArcadeDBError` ("Transaction not begun"). To retry after a conflict, read the record again and repeat the change in a new transaction: `db.run_in_transaction(fn, retries=...)` does this when `fn` does the lookup, `modify()`, `set()` and `save()` itself. A retry that reuses the old object fails again, because the object still holds the old values. + +[source,python] +---- +def rename(rid, name): + rec = db.lookup_by_rid(rid).modify() # read inside the retried function + rec.set("name", name).save() + +db.run_in_transaction(lambda: rename(rid, "new name"), retries=60) +---- + +The check costs on the write side only. A laptop A/B on the native wheel (10 clean interleaved pairs for each figure, with a clean A/A control, reported as the median of the pair ratios of the new build against main, where above 1 means slower) measured one `lookup_by_rid`, `modify()`, `set()` and `save()` of a record in a transaction: 1.241 (95% interval 1.239 to 1.244) for a record with 5 properties and 1.426 (95% interval 1.412 to 1.437) for a record with 20 properties and a 768-float vector. Each `save()` now runs one extra `SELECT`, and when a changed value is not one the engine stores exactly as sent, the `UPDATE` also returns the stored record. All 14 read figures (lookups by RID and by key, `first()`, a 10,000-row `SELECT`, a two-hop `expand()`, a two-hop Cypher query, each on both record sizes, and two projection controls) stayed between 0.991 and 1.004 of main, so reads are unchanged. If you need atomic arithmetic, an `UPDATE ... SET n = n + 1` in the statement does not use this path. + [[retry-loop]] === A retry loop for REPEATABLE_READ diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index 8fb321f2..6dd47514 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -418,6 +418,7 @@ def __init__(self, handle: int, owned_by_server: bool = False): self._vector_meta: dict = {} self._vector_dimensions: dict = {} self._vector_id_properties: dict = {} + self._auto_transaction = False # the engine default; set_auto_transaction() is the only way it changes # -- lifecycle @@ -1263,6 +1264,7 @@ def set_auto_transaction(self, enabled: bool): from .extras import db_set db_set(self, "auto_transaction", "true" if enabled else "false") + self._auto_transaction = bool(enabled) def _column_values(name, values): diff --git a/src/arcadedb/_native/records.py b/src/arcadedb/_native/records.py index f790adec..65a442ca 100644 --- a/src/arcadedb/_native/records.py +++ b/src/arcadedb/_native/records.py @@ -5,7 +5,13 @@ ``DELETE``, ``CREATE EDGE``). They subclass the JPype wrapper classes so ``isinstance`` checks keep their meaning. """ -from typing import Any, Dict, List, Optional +import copy +import math +from array import array +from collections.abc import Mapping +from datetime import date, datetime +from decimal import Decimal +from typing import Any, Dict, List from ..exceptions import ArcadeDBError from ..graph import Document as _BaseDocument @@ -27,6 +33,182 @@ def _java_object(): _UNSAVED = "#-1:-1" +_MISSING = object() +_IMMUTABLE = frozenset( + (str, int, float, bool, type(None), bytes, date, datetime, Decimal, RID) +) +_PLAIN = frozenset((str, int, bool, type(None))) +_FLOAT_ONLY = frozenset((float,)) + + +def _snap(value): + """A copy of ``value`` that nothing the caller holds can change, in the form a read of the stored value gives: a list for a list, a tuple and + a float or int array (``tolist()`` widens float32 to the same double a read gives), a Python scalar for a NumPy scalar, a set for a set. + ``set()`` writes into the record's own dict and a caller can change a nested list in place, so the values as read are copied, not kept. + """ + t = type(value) + if t in _IMMUTABLE: + return value + if t is list or t is tuple: + if set(map(type, value)) <= _IMMUTABLE: + return list(value) + return [_snap(item) for item in value] + if t is dict: + if set(map(type, value.values())) <= _IMMUTABLE: + return dict(value) + return {key: _snap(item) for key, item in value.items()} + tolist = getattr(value, "tolist", None) # a numpy array or scalar + if tolist is not None: + listed = tolist() + return listed if type(listed) is not list else _snap(listed) + if isinstance(value, Mapping): + return {key: _snap(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_snap(item) for item in value] + if isinstance(value, (set, frozenset)): + return {_snap(item) for item in value} + return copy.deepcopy(value) + + +def _same_float(a: float, b: float) -> bool: + if a == b: + return a != 0.0 or math.copysign(1.0, a) == math.copysign( + 1.0, b + ) # 0.0 and -0.0 are two values + return a != a and b != b # NaN is the value NaN, equal to itself + + +def _same_after_commit(a, b) -> bool: + """Two values of different types are one value in one case: a read inside the transaction that wrote a set, or bytes, gives the object just + written, and a read after the commit gives what the engine stored, a list (a set's elements, bytes as signed ints). The stored row is + ``a`` and the baseline ``b``, so a list against a set or bytes.""" + if type(a) is not list: + return False + if type(b) is set: + return ( + len(a) == len(b) and set(a) == b and set(map(type, a)) == set(map(type, b)) + ) + if type(b) is bytes: + return a == [x - 256 if x > 127 else x for x in b] + return False + + +def _same(a, b) -> bool: + """Whether the stored value ``a`` and the baseline ``b`` are one value: the same type (``1``, ``1.0`` and ``True`` are three values) and equal, a NaN equal to a NaN, ``0.0`` not + equal to ``-0.0``, a Decimal equal only at the same scale. Both are decoded values (a read never gives an array), so a list is compared as a + list; a float list, a vector, is compared in one call (its bytes when a zero is among the values). + """ + t = type(a) + if t is not type(b): + return _same_after_commit(a, b) + if t is float: + return _same_float(a, b) + if t is list: + if len(a) != len(b): + return False + kinds = set(map(type, a)) + if kinds != set(map(type, b)): + return False + if kinds == _FLOAT_ONLY: + if ( + a == b + ): # in C; equal floats of one type are one value unless a zero is among them (0.0 == -0.0) + return ( + 0.0 not in a or array("d", a).tobytes() == array("d", b).tobytes() + ) + return all( + map(_same_float, a, b) + ) # a NaN is not equal to itself in the list comparison + if len(kinds) <= 1 and kinds <= _PLAIN: + return a == b + return all(map(_same, a, b)) + if t is dict: + if len(a) != len(b): + return False + for key, item in a.items(): + other = b.get(key, _MISSING) + if other is _MISSING or not _same(item, other): + return False + return True + if t is set: + return a == b and set(map(type, a)) == set(map(type, b)) + if t is Decimal: + return (a == b and a.as_tuple() == b.as_tuple()) or (a.is_nan() and b.is_nan()) + if t is datetime: + return a == b and a.tzinfo == b.tzinfo + return bool(a == b) + + +def _exact_value(value) -> bool: + """Whether the engine stores this (snapped) value so that a read gives back the same value, whatever the property's declared type allows. + A kind that a type can round, truncate or turn into another kind is not on the list: a float the float32 of a FLOAT or ARRAY_OF_FLOATS + property would round (a float is exact only when float32 holds it), a datetime (milliseconds, or seconds, by the property's precision; + a time zone is dropped), a set (read back as a list), bytes (read back as a list of ints), a RID. + """ + t = type(value) + if t is str or t is bool or value is None: + return True + if t is int: + return -9223372036854775808 <= value <= 9223372036854775807 + if t is float: + return ( + array("f", (value,))[0] == value + ) # false for NaN, and for a double float32 rounds + if t is date: + return True + if t is Decimal: + return value.is_finite() + if t is list: + kinds = set(map(type, value)) + if kinds == _FLOAT_ONLY: + return array("f", value).tolist() == value + return all(map(_exact_value, value)) + if t is dict: + return all(type(key) is str for key in value) and all( + map(_exact_value, value.values()) + ) + return False + + +def _stored_as_sent(new, old) -> bool: + """Whether a changed value (``new``, snapped) is stored exactly as sent, so that the values just written can be the next baseline without a + read. The engine converts a value to the declared type of its property (an int sent to a DOUBLE property reads back as 3.0, a str to a + DATE one as a date), and the record object does not know the declaration. What it does know is what a read gave for this property before: + a value of that same type is one the property holds as it is, so only a value of the type this property was last read as, with a kind that + is exact (``_exact_value``), counts. A new property, a ``None`` before and a list that changes its element kinds do not: they are read + back.""" + if type(new) is not type(old) or old is None: + return False + if type(new) is list: + if not new: + return True + if not old or set(map(type, new)) != set(map(type, old)): + return False + return _exact_value(new) + + +def _concurrent_modification(rid: str, type_name: str) -> ArcadeDBError: + """The error the engine raises when a record changed after it was read, in the form of a failed native reply (so that + ``run_in_transaction`` retries it, as it retries the JVM build's).""" + from .database import _error + + cls = "com.arcadedb.exception.ConcurrentModificationException" + message = ( + f"Record {rid} was modified by a concurrent transaction after it was read. " + "Please retry the operation" + ) + return _error( + f"Failed to save a {type_name} record", + { + "ok": False, + "error": message, + "cls": cls, + "chain": [[cls, message]], + "retry": "need_retry", + "retry_ms": 0, + }, + ) + def _q(name: str) -> str: return "`" + name.replace("`", "``") + "`" @@ -38,10 +220,14 @@ class _Mixin: _CREATE = "INSERT INTO {t} CONTENT :row" # the record id goes over as a parameter (a RID, passed back it is a link), never in the text: the engine caches a parsed statement by its # text and keeps a record id in a FROM target as part of it, so a pasted id would make one cache entry per record - _UPDATE = "UPDATE :rid MERGE :row" + _SELECT = "SELECT FROM :rid" + _UPDATE = "UPDATE :rid MERGE :row RETURN AFTER" + _UPDATE_SENT = "UPDATE :rid MERGE :row" # no read-back: used when every changed value is stored exactly as sent (_stored_as_sent) _DELETE = "DELETE FROM :rid" - def __init__(self, database, type_name, rid=None, props=None, mutable=False): + def __init__( + self, database, type_name, rid=None, props=None, mutable=False, base=None + ): self._java_document = _java_object() self._database = database self._db = database @@ -49,6 +235,10 @@ def __init__(self, database, type_name, rid=None, props=None, mutable=False): self._rid = rid or None self._props: Dict[str, Any] = dict(props or {}) self._mutable = mutable or not rid + # a copy of the values this record held as read (None: not taken yet). A read does not take it: modify(), the first set() of a saved + # record or save() does, from the values then held, which are the read-time values. save() refuses to overwrite a changed record. + # The dict is never changed in place (a save makes a new one), so the objects modify() hands it to can share it + self._base = base self._property_names_cache = None # -- reading @@ -102,10 +292,22 @@ def set(self, name: str, value: Any): raise AttributeError( f"{type(self).__name__} is immutable. Call .modify() first to get a mutable version." ) + if self._base is None and self._rid: + self._base = _snap( + self._props + ) # a saved record's values as read, before the first change to it self._props[name] = value return self def save(self): + """Create the record, or write the changed properties of a saved one. + + A saved record is written back only if the stored record still has the values this object was read with. Otherwise it raises + ArcadeDBError ("ConcurrentModificationException" in the message, retryable by ``run_in_transaction``) and writes nothing: read the + record again and repeat the change. The check compares values (the engine keeps no version), so a change that was changed back is + not seen. Saving an existing record needs an active transaction, like every write; outside one it raises ArcadeDBError + ("Transaction not begun"). Creating a new record is not checked. + """ db = self._db if self._rid is None: from .extras import veto_count @@ -118,11 +320,59 @@ def save(self): return self # a before_create callback vetoed it: nothing is written, no error (as on the JPype backend) self._adopt(row) else: - db.command( - "sql", self._UPDATE, {"rid": RID(self._rid), "row": self._props} - ).close() + self._save_update(db) return self + def _save_update(self, db) -> None: + """UPDATE .. MERGE of a saved record, refused when the stored record is no longer the one this object was read from. + + The engine keeps no record version. A copy of the values this object held as read (taken once, at modify() or here) is compared by + value with the values stored now, in the caller's transaction; a stored record that differs raises ConcurrentModificationException and + nothing is written. The copy then becomes the values this save wrote, so the object's own save is not another transaction's change: + when every changed value is one the engine stores exactly as sent (``_stored_as_sent``) the statement returns no row and the values + sent are the new copy; otherwise the statement returns the stored row (RETURN AFTER) and that row is. + + 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: + raise ArcadeDBError( + "Failed to save record: com.arcadedb.exception.TransactionException: Transaction not begun " + "(wrap the write in `with db.transaction():` or call begin() first)" + ) + base = self._base + if base is None: + base = self._base = _snap(self._props) + stored = db.query("sql", self._SELECT, {"rid": RID(self._rid)}).first() + if stored is not None and not _same(stored._props, base): + raise _concurrent_modification(self._rid, self._type) + sent = dict(base) + exact = True + for name, value in self._props.items(): + old = base.get(name, _MISSING) + if old is not _MISSING and _same(value, old): + continue # as read: the copy of it stays + value = _snap(value) + if old is not _MISSING and _same(value, old): + continue # a tuple or an array of what was read + sent[name] = value + if exact and (old is _MISSING or not _stored_as_sent(value, old)): + exact = False + from .extras import veto_count + + vetoes = veto_count() + params = {"rid": RID(self._rid), "row": self._props} + if exact: + row = db.command("sql", self._UPDATE_SENT, params).first() + if row is not None and row.get("count") and veto_count() == vetoes: + self._base = sent + else: + after = db.command("sql", self._UPDATE, params).first() + if ( + after is not None + ): # a veto or a record that is gone gives no row: nothing was written, the copy stays + self._base = after._props + def _adopt(self, row) -> None: if row is None or not row.get_rid(): raise ArcadeDBError( @@ -130,6 +380,7 @@ def _adopt(self, row) -> None: ) self._rid = row.get_rid() self._props = row.to_dict() + self._base = None # the stored values are the ones now held: copied when the first change or the next save needs it def delete(self): if self._rid is None: @@ -138,7 +389,18 @@ def delete(self): def modify(self): self._check_open() - return type(self)(self._db, self._type, self._rid, self._props, mutable=True) + if self._rid and self._base is None: + self._base = _snap( + self._props + ) # the values held now are the ones read: copied here, used by save() + return type(self)( + self._db, + self._type, + self._rid, + self._props, + mutable=True, + base=self._base, + ) def __repr__(self) -> str: return f"<{type(self).__name__} rid={self.get_rid()} type={self._type}>" diff --git a/src/arcadedb/graph.py b/src/arcadedb/graph.py index 4641788d..198465d1 100644 --- a/src/arcadedb/graph.py +++ b/src/arcadedb/graph.py @@ -129,6 +129,8 @@ def save(self) -> "Document": Raises: ArcadeDBError: If the engine refuses the write (no transaction, a duplicate key, a read-only database, a ``before_*`` callback that raised). + ArcadeDBError: If another transaction changed the record after it was read + (``ConcurrentModificationException`` in the message; the native build raises the same error). """ try: self._java_document.save() diff --git a/tests/conftest.py b/tests/conftest.py index 7669916d..24cbc673 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -56,6 +56,8 @@ collect_ignore.append("test_native_point_read.py") # the vector provider the native glue picks by default (ARCADEDB_VECTOR_PROVIDER, #153), observed on the native library 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) 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) diff --git a/tests/test_graph_api.py b/tests/test_graph_api.py index 7ba4264d..235b31f4 100644 --- a/tests/test_graph_api.py +++ b/tests/test_graph_api.py @@ -542,6 +542,124 @@ def test_delete_via_wrapper_on_fresh_lookup(temp_db_path): assert len(items_after) == 0, "Item should be deleted via .delete()" +def _create_doc_ab(db): + """A document type D with one record a=1, b=1; returns its RID.""" + db.command("sql", "CREATE DOCUMENT TYPE D") + with db.transaction(): + doc = db.new_document("D") + doc.set("a", 1) + doc.set("b", 1) + doc.save() + return str(doc.get_identity()) + + +def _bump_b_sql(rid): + """The statement that changes b of the record at ``rid`` (a RID the test read).""" + return f"UPDATE {rid} SET b = 2" # nosec B608 - rid is a record id the test read + + +def _stored_ab(db): + row = list(db.query("sql", "SELECT a, b FROM D"))[0] + return row.get("a"), row.get("b") + + +def test_save_refuses_to_overwrite_a_record_changed_after_it_was_read(temp_db_path): + """A record object read before another transaction changed the stored record must not write its snapshot back (a lost update).""" + with arcadedb.create_database(temp_db_path) as db: + rid = _create_doc_ab(db) + + mutable = db.lookup_by_rid(rid).modify() + with db.transaction(): # the stored record changes after the read + db.command("sql", _bump_b_sql(rid)) + + mutable.set("a", 10) + with pytest.raises(arcadedb.ArcadeDBError) as excinfo: + with db.transaction(): + mutable.save() + message = str(excinfo.value) + assert "oncurrent" in message or rid in message + + # nothing of the refused save is stored, and the other transaction's value is kept + assert _stored_ab(db) == (1, 2) + + +def test_save_refuses_a_stale_record_from_a_query_result(temp_db_path): + """The same rule for a record that was born from a query row.""" + with arcadedb.create_database(temp_db_path) as db: + rid = _create_doc_ab(db) + + row = list(db.query("sql", "SELECT FROM D"))[0] + mutable = row.get_element().modify() + with db.transaction(): + db.command("sql", _bump_b_sql(rid)) + + mutable.set("a", 10) + with pytest.raises(arcadedb.ArcadeDBError): + with db.transaction(): + mutable.save() + assert _stored_ab(db) == (1, 2) + + # a fresh read sees the new state and saves + fresh = db.lookup_by_rid(rid).modify() + fresh.set("a", 10) + with db.transaction(): + fresh.save() + assert _stored_ab(db) == (10, 2) + + +def test_save_of_an_existing_record_outside_a_transaction_is_refused(temp_db_path): + """Both builds: save() is a write, so it needs an active transaction (auto-transaction is off by default) and opens none of its own.""" + with arcadedb.create_database(temp_db_path) as db: + rid = _create_doc_ab(db) + + mutable = db.lookup_by_rid(rid).modify() + mutable.set("a", 10) + with pytest.raises(arcadedb.ArcadeDBError) as excinfo: + mutable.save() + assert "Transaction not begun" in str(excinfo.value) + assert not db.is_transaction_active() # save() left no transaction behind + assert _stored_ab(db) == (1, 1) # and wrote nothing + + # the same object saves inside a transaction + with db.transaction(): + mutable.save() + assert _stored_ab(db) == (10, 1) + + +def test_two_saves_in_a_row_by_one_record_object_both_succeed(temp_db_path): + """The object's own earlier save is not another transaction's change: the second save, in the same and in a later transaction, goes through.""" + with arcadedb.create_database(temp_db_path) as db: + rid = _create_doc_ab(db) + + mutable = db.lookup_by_rid(rid).modify() + with db.transaction(): + mutable.set("a", 2) + mutable.save() + mutable.set("b", 3) + mutable.save() + assert _stored_ab(db) == (2, 3) + + with db.transaction(): + mutable.set("a", 4) + mutable.save() + assert _stored_ab(db) == (4, 3) + + +def test_record_saved_for_the_first_time_can_be_saved_again(temp_db_path): + """A record created by this object and saved, then changed and saved in a later transaction.""" + with arcadedb.create_database(temp_db_path) as db: + db.command("sql", "CREATE DOCUMENT TYPE D") + with db.transaction(): + created = db.new_document("D") + created.set("a", 7) + created.set("b", 7) + created.save() + with db.transaction(): + created.set("a", 8) + created.save() + assert _stored_ab(db) == (8, 7) + + def _chain(db, count): """``count`` Person vertices linked a -> b -> c ... by Knows edges (``w`` = the position of the edge).""" db.command("sql", "CREATE VERTEX TYPE Person") diff --git a/tests/test_native_save_baseline.py b/tests/test_native_save_baseline.py new file mode 100644 index 00000000..6e0236ab --- /dev/null +++ b/tests/test_native_save_baseline.py @@ -0,0 +1,558 @@ +"""Native build: save() of a record refuses to overwrite a record that changed after it was read, and the check is cheap. + +A record keeps a copy of the values it held as read, taken lazily (at ``modify()`` or at the first ``set()`` of a saved record, never when +a read decodes a row), and ``save()`` compares the stored row with that copy by value. After the UPDATE the values just sent are the new +copy when the engine stores them exactly as sent, so the row is not read back; otherwise the UPDATE returns the stored row. The seams the +tests count are ``arcadedb._native.records._snap`` (the copy) and the statement texts handed to the engine. The refusal itself is also +tested in test_graph_api.py, which runs on both builds. +""" + +import copy +import hashlib +import math +from datetime import date, datetime, timezone +from decimal import Decimal + +import numpy as np +import pytest + +import arcadedb +from arcadedb._native import records +from arcadedb._native.database import NativeDatabase + + +@pytest.fixture +def snaps(monkeypatch): + """The list of calls to the copy function (it still computes the real copy).""" + calls = [] + real = records._snap + + def counting(value): + calls.append(value) + return real(value) + + monkeypatch.setattr(records, "_snap", counting) + return calls + + +@pytest.fixture +def hashes(monkeypatch): + """Every hash the save path could compute: blake2b anywhere, and the record fingerprint of earlier versions of this module (counted when a + version has one).""" + calls = [] + real_hash = hashlib.blake2b + + def counting_hash(*args, **kwargs): + calls.append(("blake2b", args)) + return real_hash(*args, **kwargs) + + monkeypatch.setattr(hashlib, "blake2b", counting_hash) + real_fingerprint = getattr(records, "_fingerprint", None) + if real_fingerprint is not None: + + def counting_fingerprint(props): + calls.append(("fingerprint", props)) + return real_fingerprint(props) + + monkeypatch.setattr(records, "_fingerprint", counting_fingerprint) + return calls + + +@pytest.fixture +def texts(monkeypatch): + """Every statement text handed to the engine, in order, while the fixture is active.""" + seen = [] + original = NativeDatabase._exec + + def recording(self, flags, prefix, language, command, args): + seen.append(command) + return original(self, flags, prefix, language, command, args) + + monkeypatch.setattr(NativeDatabase, "_exec", recording) + return seen + + +def _create(db): + db.command("sql", "CREATE VERTEX TYPE V") + db.command("sql", "CREATE EDGE TYPE E") + with db.transaction(): + a = db.new_vertex("V") + a.set("a", 1) + a.set("b", 1) + a.save() + b = db.new_vertex("V") + b.set("a", 2) + b.save() + a.new_edge("E", b, w=1.5) + return str(a.get_identity()) + + +def _bump_b_sql(rid): + """The statement that changes b of the record at ``rid`` (a RID the test read).""" + return f"UPDATE {rid} SET b = 2" # nosec B608 - rid is a record id the test read + + +def _stored(db, rid): + select_sql = ( + f"SELECT a, b FROM {rid}" # nosec B608 - rid is a record id the test read + ) + row = list(db.query("sql", select_sql))[0] + return row.get("a"), row.get("b") + + +def test_reads_never_copy_the_values(temp_db_path, snaps): + with arcadedb.create_database(temp_db_path) as db: + rid = _create(db) + snaps.clear() + + rows = list(db.query("sql", "SELECT FROM V")) + records_read = [row.get_element() for row in rows] + by_rid = db.lookup_by_rid(rid) + vertex = records_read[0] + edges = vertex.get_out_edges("E") + vertex.get_in_edges("E") + ends = [edge.get_in() for edge in edges] + [edge.get_out() for edge in edges] + cypher = [ + row.get_element() for row in db.query("opencypher", "MATCH (n:V) RETURN n") + ] + assert len(records_read) == 2 and by_rid is not None and ends and cypher + assert snaps == [] # nothing was copied by any read path + + mutable = by_rid.modify() + assert len(snaps) == 1 # modify() copies, once, the values the record holds + assert snaps[0] == by_rid.to_dict() + by_rid.modify() + assert len(snaps) == 1 # and caches the copy on the record + + mutable.set("a", 10) # the copy is already there: no work + assert len(snaps) == 1 + with db.transaction(): + mutable.save() + assert _stored(db, rid) == (10, 1) + + +def test_no_hash_is_computed_on_the_save_path(temp_db_path, hashes): + with arcadedb.create_database(temp_db_path) as db: + rid = _create(db) + mutable = db.lookup_by_rid(rid).modify() + mutable.set("a", 10) + with db.transaction(): + mutable.save() + mutable.set("b", 5) + mutable.save() + again = db.lookup_by_rid(rid).modify() + again.set("name", "x") # a new property: the read-back path + with db.transaction(): + again.save() + assert _stored(db, rid) == (10, 5) + assert hashes == [] + + +def test_modify_then_save_after_a_concurrent_change_still_raises(temp_db_path): + with arcadedb.create_database(temp_db_path) as db: + rid = _create(db) + + mutable = db.lookup_by_rid(rid).modify() + with ( + db.transaction() + ): # another transaction changes the stored record after the read + db.command("sql", _bump_b_sql(rid)) + mutable.set("a", 10) + with pytest.raises(arcadedb.ArcadeDBError) as excinfo: + with db.transaction(): + mutable.save() + assert "ConcurrentModificationException" in str(excinfo.value) + assert ( + excinfo.value.native["retry"] == "need_retry" + ) # run_in_transaction retries it + assert _stored(db, rid) == (1, 2) # nothing of the refused save is stored + + # a fresh read saves, in a transaction + fresh = db.lookup_by_rid(rid).modify() + fresh.set("a", 10) + with db.transaction(): + fresh.save() + assert _stored(db, rid) == (10, 2) + + +def test_a_change_between_the_read_and_modify_is_not_lost_when_modify_comes_late( + temp_db_path, +): + """The copy is of the values read, not of what is stored when modify() runs.""" + with arcadedb.create_database(temp_db_path) as db: + rid = _create(db) + + read = db.lookup_by_rid(rid) + with db.transaction(): + db.command("sql", _bump_b_sql(rid)) + mutable = ( + read.modify() + ) # late: the stored record is already newer than the values held + mutable.set("a", 10) + with pytest.raises(arcadedb.ArcadeDBError): + with db.transaction(): + mutable.save() + assert _stored(db, rid) == (1, 2) + + +def test_a_record_created_here_then_changed_elsewhere_is_refused(temp_db_path, snaps): + """A record this object created has no copy until its first change: taken before that change, from the stored values.""" + with arcadedb.create_database(temp_db_path) as db: + db.command("sql", "CREATE DOCUMENT TYPE D") + with db.transaction(): + created = db.new_document("D") + created.set("a", 1) + created.set("b", 1) + created.save() + assert snaps == [] # creating and saving a new record copies nothing + rid = str(created.get_identity()) + with db.transaction(): + db.command("sql", _bump_b_sql(rid)) + created.set("a", 9) + with pytest.raises(arcadedb.ArcadeDBError): + with db.transaction(): + created.save() + assert _stored(db, rid) == (1, 2) + + +# ---- the copy of the values as read: nothing the caller does to the record changes it + + +def _vector(n=768, seed=7): + return np.random.default_rng(seed).random(n).astype(np.float32).tolist() + + +def _rich_record(db): + """A vertex with a scalar, a list, a map and a 768-float vector (float32 values, as the engine stores them); returns its RID.""" + db.command("sql", "CREATE VERTEX TYPE M") + db.command("sql", "CREATE PROPERTY M.vec ARRAY_OF_FLOATS") + with db.transaction(): + vertex = db.new_vertex("M") + vertex.set("name", "n") + vertex.set("tags", ["x", "y"]) + vertex.set("meta", {"k": [1, 2], "z": {"q": "r"}}) + vertex.set("vec", _vector()) + vertex.save() + return str(vertex.get_identity()) + + +def test_the_copy_is_not_changed_by_set_or_by_changing_a_nested_value_in_place( + temp_db_path, +): + with arcadedb.create_database(temp_db_path) as db: + rid = _rich_record(db) + read = db.lookup_by_rid(rid) + mutable = read.modify() + as_read = copy.deepcopy(mutable.to_dict()) + assert mutable._base == as_read + assert mutable._base is not mutable._props + for name in ("tags", "meta", "vec"): + assert mutable._base[name] is not mutable._props[name] # copied, not shared + + mutable.set("name", "changed") # set() writes into the record's own dict + mutable.set("tags", ["other"]) + assert mutable._base == as_read + + mutable.get("meta")["k"].append(3) # a nested list changed in place + mutable.get("meta")["z"]["new"] = 1 + mutable.get("vec")[5] = 0.5 + assert mutable.get("vec")[5] == 0.5 and mutable._props["meta"]["k"] == [1, 2, 3] + assert mutable._base == as_read + assert read.modify()._base == as_read # the read object's copy is untouched too + + # the changes in place are the record's changes: the save writes them, and the object's next save is not a conflict + with db.transaction(): + mutable.save() + fresh = db.lookup_by_rid(rid) + assert fresh.get("meta") == {"k": [1, 2, 3], "z": {"q": "r", "new": 1}} + assert fresh.get("vec")[5] == 0.5 and fresh.get("tags") == ["other"] + mutable.get("tags").append("again") + with db.transaction(): + mutable.save() + assert db.lookup_by_rid(rid).get("tags") == ["other", "again"] + + +def test_a_concurrent_change_of_a_nested_vector_is_detected(temp_db_path): + with arcadedb.create_database(temp_db_path) as db: + rid = _rich_record(db) + mutable = db.lookup_by_rid(rid).modify() + other = _vector() + other[767] = float(np.float32(other[767] + 0.25)) # one element, the last + with db.transaction(): + db.command( + "sql", + f"UPDATE {rid} SET vec = :v", # nosec B608 - rid is a record id the test read + {"v": other}, + ).close() + mutable.set("name", "mine") + with pytest.raises(arcadedb.ArcadeDBError) as excinfo: + with db.transaction(): + mutable.save() + assert "ConcurrentModificationException" in str(excinfo.value) + fresh = db.lookup_by_rid(rid) + assert fresh.get("name") == "n" and fresh.get("vec")[767] == other[767] + + # and a nested map changed by another transaction + mutable = db.lookup_by_rid(rid).modify() + with db.transaction(): + db.command( + "sql", + f"UPDATE {rid} SET meta = :m", # nosec B608 - rid is a record id the test read + {"m": {"k": [1, 2], "z": {"q": "R"}}}, + ).close() + mutable.set("name", "mine") + with pytest.raises(arcadedb.ArcadeDBError): + with db.transaction(): + mutable.save() + + +def test_a_vetoed_update_leaves_the_copy_as_it_was(temp_db_path): + """A before_update veto writes nothing and returns no error; the next save, after the veto is lifted, is not a conflict.""" + with arcadedb.create_database(temp_db_path) as db: + db.command("sql", "CREATE DOCUMENT TYPE D") + with db.transaction(): + doc = db.new_document("D") + doc.set("name", "a") + doc.save() + rid = str(doc.get_identity()) + veto = db.on("before_update", lambda r: r.get("name") != "frozen", "D") + mutable = db.lookup_by_rid(rid).modify() + mutable.set("name", "frozen") + with db.transaction(): + mutable.save() + assert db.lookup_by_rid(rid).get("name") == "a" # nothing written + veto.remove() + mutable.set("name", "thawed") + with db.transaction(): + mutable.save() + assert db.lookup_by_rid(rid).get("name") == "thawed" + + +# ---- which values are stored exactly as sent (the parity test behind records._stored_as_sent) + +_F32 = _vector(8, seed=3) +_NOT_F32 = [0.1, 0.2, 0.3] +_MS = datetime(2021, 1, 2, 3, 4, 5, 678000) +_US = datetime(2021, 1, 2, 3, 4, 5, 678901) + +# (id, declared type of the property or None, value stored first, value sent by set(), True when the read-back is skipped) +CASES = [ + ("str", "STRING", "a", "héllo 日本", True), + ("str_empty", "STRING", "a", "", True), + ("bool", "BOOLEAN", True, False, True), + ("int", "INTEGER", 1, 2, True), + ("int_negative_long", "LONG", 1, -(2**40), True), + ("int_untyped", None, 1, 2**40, True), + ("float_f32_exact", "DOUBLE", 0.5, 0.25, True), + ("float_neg_zero", "DOUBLE", 0.5, -0.0, True), + ("float_untyped_f32_exact", None, 0.5, 0.75, True), + ("float_double", "DOUBLE", 0.5, 0.1, False), # fine in a DOUBLE, rounded by a FLOAT + ("float_in_FLOAT_property", "FLOAT", 0.5, 0.1, False), # reads back 0.10000000149 + ("float_nan", "DOUBLE", 0.5, float("nan"), False), + ("float_inf", "DOUBLE", 0.5, float("inf"), True), + ("date", "DATE", date(2020, 1, 2), date(2021, 3, 4), True), + ("datetime_ms", "DATETIME", datetime(2020, 1, 2, 3, 4, 5), _MS, False), + ("datetime_us", "DATETIME", datetime(2020, 1, 2, 3, 4, 5), _US, False), + ("decimal", "DECIMAL", Decimal("1.5"), Decimal("12.50"), True), + ("set_into_list_property", "LIST", [1], {1, 2, 3}, False), # reads back as a list + ( + "bytes", + "BINARY", + b"\x01", + b"\x00\x01\xff", + False, + ), # reads back as a list of ints + ("bytes_untyped", None, 1, b"\x00\x01\xff", False), + ("set_untyped", None, 1, {1, 2, 3}, False), + ( + "datetime_aware_untyped", + None, + 1, + datetime(2021, 1, 2, 3, 4, 5, 678000, tzinfo=timezone.utc), + False, + ), + ("datetime_untyped", None, datetime(2020, 1, 2), _US, False), + ("numpy_float32_scalar", "DOUBLE", 0.5, np.float32(0.25), True), + ("numpy_float64_scalar_not_f32", "DOUBLE", 0.5, np.float64(0.1), False), + ("numpy_int64_scalar", "LONG", 1, np.int64(5), True), + ("numpy_bool", "BOOLEAN", True, np.bool_(False), True), + ("list_float32_values", "ARRAY_OF_FLOATS", _F32, _F32[::-1], True), + ("list_float32_values_untyped", None, _F32, _F32[::-1], True), + ("list_double_values", "ARRAY_OF_FLOATS", _F32, _NOT_F32, False), # read as float32 + ("list_double_values_in_LIST", "LIST", [0.5, 0.25], _NOT_F32, False), + ( + "ndarray_float32", + "ARRAY_OF_FLOATS", + _F32, + np.asarray(_F32[::-1], dtype=np.float32), + True, + ), + ( + "ndarray_float64_f32_values", + "ARRAY_OF_FLOATS", + _F32, + np.asarray(_F32[::-1], dtype=np.float64), + True, + ), + ("ndarray_float64_not_f32", "ARRAY_OF_FLOATS", _F32, np.asarray(_NOT_F32), False), + ("ndarray_int", "LIST", [1, 2], np.array([3, 4, 5]), True), + ("list_ints", "LIST", [1], [1, 2, 3], True), + ("list_strs", "LIST", ["a"], ["b", "c"], True), + ("list_empty", "LIST", [1], [], True), + ("list_from_empty", "LIST", [], [1], False), # the element kind was never read + ("list_other_element_kind", "LIST", [1, 2], ["a"], False), + ("list_with_none", "LIST", [1, None], [None, 2], True), + ("tuple", "LIST", [1], (1, 2), True), + ("list_of_maps", "LIST", [{"a": 1}], [{"a": 1}, {"b": [1, 2]}], True), + ("map", "MAP", {"a": 1}, {"b": 1, "a": [0.5, 0.25], "c": {"z": "q"}}, True), + ("map_with_datetime", "MAP", {"a": 1}, {"d": _MS}, False), + ("map_with_set", "MAP", {"a": 1}, {"s": {1, 2}}, False), + ( + "list_of_datetimes", + "LIST", + [datetime(2020, 1, 2)], + [datetime(2021, 1, 2)], + False, + ), + ("int_into_DOUBLE_property", "DOUBLE", 1.5, 3, False), # reads back as 3.0 + ("float_into_INTEGER_property", "INTEGER", 1, 3.0, False), # reads back as 3 + ("str_into_INTEGER_property", "INTEGER", 1, "3", False), + ("str_into_DATE_property", "DATE", date(2020, 1, 2), "2021-01-02", False), + ("int_into_STRING_property", "STRING", "a", 5, False), + ("none_before", "STRING", None, "x", False), + ("none_after", "STRING", "a", None, False), +] + + +def _updates(texts): + return [text for text in texts if text.startswith("UPDATE")] + + +@pytest.mark.parametrize("case", CASES, ids=[c[0] for c in CASES]) +def test_a_value_is_stored_as_sent_exactly_when_the_read_back_is_skipped( + temp_db_path, texts, case +): + """For every kind: the baseline after a save equals what a fresh read returns (so the same object's next save is not a conflict), and the + statement has no RETURN AFTER exactly for the kinds the engine stores as sent.""" + _, declared, first, sent, skips_read_back = case + with arcadedb.create_database(temp_db_path) as db: + db.command("sql", "CREATE DOCUMENT TYPE D") + if declared: + db.command("sql", f"CREATE PROPERTY D.p {declared}") + with db.transaction(): + doc = db.new_document("D") + doc.set("o", 1) + doc.set("p", first) + doc.save() + rid = str(doc.get_identity()) + + mutable = db.lookup_by_rid(rid).modify() + mutable.set("p", sent) + del texts[:] + with db.transaction(): + mutable.save() + updates = _updates(texts) + assert len(updates) == 1 + assert ("RETURN AFTER" not in updates[0]) is skips_read_back, updates[0] + + stored = db.lookup_by_rid(rid)._props + assert sorted(stored) == ["o", "p"] + assert records._same( + stored, mutable._base + ), "the baseline differs from what a read returns" + + # the same object saves again, twice (a baseline that differed from the stored row would be refused here) + mutable.set("o", 2) + with db.transaction(): + mutable.save() + mutable.set("o", 3) + mutable.save() + assert db.lookup_by_rid(rid).get("o") == 3 + assert records._same(db.lookup_by_rid(rid)._props, mutable._base) + + +def test_a_new_property_is_read_back_and_then_known(temp_db_path, texts): + with arcadedb.create_database(temp_db_path) as db: + rid = _create(db) + mutable = db.lookup_by_rid(rid).modify() + mutable.set( + "fresh", "x" + ) # no value of this property was read: its type is unknown + del texts[:] + with db.transaction(): + mutable.save() + assert "RETURN AFTER" in _updates(texts)[0] + mutable.set("fresh", "y") # now it was read as a str: exact + del texts[:] + with db.transaction(): + mutable.save() + assert "RETURN AFTER" not in _updates(texts)[0] + assert db.lookup_by_rid(rid).get("fresh") == "y" + + +def test_a_stored_vector_reads_as_a_list_of_python_floats(temp_db_path): + """The codec decodes a float vector to a list (``tolist()`` of the float32 buffer), never to an array: the copy and the comparison + work on lists, and an array that is sent is copied as the list it reads back as.""" + with arcadedb.create_database(temp_db_path) as db: + rid = _rich_record(db) + stored = db.lookup_by_rid(rid).get("vec") + assert type(stored) is list and set(map(type, stored)) == {float} + array = np.asarray(stored, dtype=np.float32) + snapped = records._snap(array) + assert type(snapped) is list and snapped == stored + assert records._same(snapped, stored) + mutable = db.lookup_by_rid(rid).modify() + mutable.set("vec", array[::-1].copy()) + with db.transaction(): + mutable.save() + assert db.lookup_by_rid(rid).get("vec") == stored[::-1] + assert records._same(mutable._base["vec"], stored[::-1]) + + +# ---- the two helpers on their own + + +def test_same_is_strict_about_types_and_special_floats(): + same = records._same + assert same(1, 1) and not same(1, 1.0) and not same(1, True) and not same(0, False) + assert same(float("nan"), float("nan")) + assert not same(0.0, -0.0) and same(-0.0, -0.0) + assert same([1.0, float("nan")], [1.0, float("nan")]) + assert not same([0.0, 1.0], [-0.0, 1.0]) + assert not same([1, True], [True, 1]) and not same([1, 2], [1.0, 2.0]) + assert not same([1, 2], [1, 2, 3]) and not same([1, 2], (1, 2)) + vec = _vector() + changed = list(vec) + changed[500] += 0.5 + assert same(vec, list(vec)) and not same(vec, changed) + assert same({"a": [1, {"b": 2}], "c": 1}, {"c": 1, "a": [1, {"b": 2}]}) + assert not same({"a": 1}, {"a": 1, "b": 2}) and not same({"a": 1}, {"b": 1}) + assert same({1, 2}, {2, 1}) and not same({1}, {True}) + assert same(Decimal("1.50"), Decimal("1.50")) + assert not same(Decimal("1.5"), Decimal("1.50")) + assert same(date(2020, 1, 2), date(2020, 1, 2)) + assert not same(date(2020, 1, 2), datetime(2020, 1, 2)) + assert not same(None, 0) and same(None, None) and same("a", "a") + assert not same(math.inf, -math.inf) and same(math.inf, math.inf) + + +def test_snap_copies_and_normalises(): + snap = records._snap + nested = {"l": [1, [2, 3]], "m": {"k": [4]}, "s": {1, 2}, "t": (1, 2)} + copied = snap(nested) + assert copied == {"l": [1, [2, 3]], "m": {"k": [4]}, "s": {1, 2}, "t": [1, 2]} + assert copied["l"] is not nested["l"] and copied["l"][1] is not nested["l"][1] + assert copied["m"]["k"] is not nested["m"]["k"] and copied["s"] is not nested["s"] + nested["l"][1].append(9) + nested["m"]["k"].append(9) + assert copied["l"] == [1, [2, 3]] and copied["m"] == {"k": [4]} + # arrays and NumPy scalars are the lists and scalars a read gives + array = np.asarray([0.5, 0.25], dtype=np.float32) + assert snap(array) == [0.5, 0.25] and type(snap(array)[0]) is float + assert snap(np.float32(0.5)) == 0.5 and type(snap(np.float32(0.5))) is float + assert type(snap(np.int64(3))) is int and type(snap(np.bool_(True))) is bool + assert snap(np.array([[1, 2], [3, 4]])) == [[1, 2], [3, 4]] + held = snap(array) + array[1] = 8 # the copy is not the array + assert held == [0.5, 0.25] + values = [datetime(2020, 1, 2), date(2020, 1, 2), Decimal("1"), b"x", None, "s"] + assert snap(values) == values and snap(values) is not values