From 4296bdf80a5f9110a9fe14282de4d1424b06b10f Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 2 Nov 2026 20:06:23 +0900 Subject: [PATCH 1/4] native: save() of a record refuses to overwrite a record that changed after it was read (#172) The engine keeps no record version, and native save() wrote the Python snapshot back with UPDATE .. MERGE, reverting another transaction's committed values (the JVM build raises ConcurrentModificationException). The fingerprint of the values a record held as read is taken lazily: at modify(), at the first set() of a saved record, or at save(), from the values then held (the read-time snapshot), and cached on the record. The read path takes none (a first design fingerprinted the stored bytes in Java for every record row and slowed reads 5 to 8 percent in a timed A/B, see the issue). save() compares it with the values stored now in the transaction of the update, raises a retryable ConcurrentModificationException error when they differ (nothing written), and moves the fingerprint to the stored values after its own write (UPDATE .. RETURN AFTER). No Java change. Tests: the refusal on both builds (test_graph_api.py), and on the native build that no read path fingerprints and that a late modify() still compares the values read. Co-Authored-By: Claude Opus 5.5 --- src/arcadedb/_native/records.py | 116 +++++++++++++++- tests/conftest.py | 2 + tests/test_graph_api.py | 96 +++++++++++++ tests/test_native_save_lazy_fingerprint.py | 153 +++++++++++++++++++++ 4 files changed, 361 insertions(+), 6 deletions(-) create mode 100644 tests/test_native_save_lazy_fingerprint.py diff --git a/src/arcadedb/_native/records.py b/src/arcadedb/_native/records.py index 6d81502..96e1c41 100644 --- a/src/arcadedb/_native/records.py +++ b/src/arcadedb/_native/records.py @@ -5,7 +5,9 @@ ``DELETE``, ``CREATE EDGE``). They subclass the JPype wrapper classes so ``isinstance`` checks keep their meaning. """ -from typing import Any, Dict, List, Optional +import hashlib +from collections.abc import Mapping +from typing import Any, Dict, List from ..exceptions import ArcadeDBError from ..graph import Document as _BaseDocument @@ -27,6 +29,59 @@ def _java_object(): _UNSAVED = "#-1:-1" +_SCALARS = frozenset((str, int, float, bool, type(None))) + + +def _canon(value): + """``value`` as plain nested lists, tuples and scalars that compare and print the same for two equal values, whatever order a + dict or a set was built in.""" + if type(value) in _SCALARS: + return value + if isinstance(value, Mapping): + return ( + "m", + sorted(((_canon(k), _canon(v)) for k, v in value.items()), key=repr), + ) + if isinstance(value, (list, tuple)): + return [_canon(v) for v in value] + if isinstance(value, (set, frozenset)): + return ("s", sorted((_canon(v) for v in value), key=repr)) + tolist = getattr(value, "tolist", None) # a numpy array + if tolist is not None: + return ("a", str(getattr(value, "dtype", "")), _canon(tolist())) + return value # datetime, date, Decimal, RID and bytes print by value + + +def _fingerprint(props: Dict[str, Any]) -> bytes: + """A fingerprint of the values a record holds. Called only when a record is about to be changed (``modify()``, the first ``set()`` of a + saved record, ``save()``), never when one is read: the read path keeps the values the record already holds, and this runs on them + later. Two reads of one stored record decode to equal values, so equal fingerprints mean an unchanged record. + """ + return hashlib.blake2b(repr(_canon(props)).encode(), digest_size=16).digest() + + +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,7 +93,9 @@ class _Mixin: _CREATE = "INSERT INTO {t} CONTENT :row" _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, token=None + ): self._java_document = _java_object() self._database = database self._db = database @@ -46,6 +103,9 @@ 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 + # fingerprint 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 snapshot. save() refuses to overwrite a changed record + self._token = token self._property_names_cache = None # -- reading @@ -99,6 +159,10 @@ 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._token is None and self._rid: + self._token = _fingerprint( + self._props + ) # a saved record's snapshot, before the first change to it self._props[name] = value return self @@ -115,11 +179,39 @@ 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", f"UPDATE {self._rid} MERGE :row", {"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. The values this object held as read are fingerprinted (once, at modify() or here) and compared + with the values stored now, in the transaction of the update (one is opened when none is active); a stored record that differs + raises ConcurrentModificationException and nothing is written. The fingerprint then moves to the stored values after this save, + so the object's own save is not another transaction's change.""" + token = self._token + if token is None: + token = self._token = _fingerprint(self._props) + own = not db.is_transaction_active() + if own: + db.begin() + try: + stored = db.query("sql", f"SELECT FROM {self._rid}").first() + if stored is not None and _fingerprint(stored._props) != token: + raise _concurrent_modification(self._rid, self._type) + after = db.command( + "sql", + f"UPDATE {self._rid} MERGE :row RETURN AFTER", + {"row": self._props}, + ).first() + if own: + db.commit() + except BaseException: + if own and db.is_transaction_active(): + db.rollback() + raise + self._token = _fingerprint(after._props) if after is not None else None + def _adopt(self, row) -> None: if row is None or not row.get_rid(): raise ArcadeDBError( @@ -127,6 +219,7 @@ def _adopt(self, row) -> None: ) self._rid = row.get_rid() self._props = row.to_dict() + self._token = None # the stored values are the ones now held: fingerprinted when the first change or the next save needs it def delete(self): if self._rid is None: @@ -135,7 +228,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._token is None: + self._token = _fingerprint( + self._props + ) # the values held now are the ones read: cached here, used by save() + return type(self)( + self._db, + self._type, + self._rid, + self._props, + mutable=True, + token=self._token, + ) def __repr__(self) -> str: return f"<{type(self).__name__} rid={self.get_rid()} type={self._type}>" diff --git a/tests/conftest.py b/tests/conftest.py index c5961c2..d26585e 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 lazy fingerprint of a record that save() compares with the stored one (arcadedb._native.records), counted through its seam + collect_ignore.append("test_native_save_lazy_fingerprint.py") if not os.path.isfile( os.path.join(os.path.dirname(os.path.dirname(__file__)), "docs", "index.md") ): diff --git a/tests/test_graph_api.py b/tests/test_graph_api.py index 3b54991..5c6d6a9 100644 --- a/tests/test_graph_api.py +++ b/tests/test_graph_api.py @@ -4,6 +4,8 @@ Tests Document, Vertex, and Edge wrapper classes and their methods. """ +import pytest + import arcadedb from arcadedb.graph import Document, Edge, Vertex @@ -538,3 +540,97 @@ def test_delete_via_wrapper_on_fresh_lookup(temp_db_path): # Verify deleted items_after = list(db.query("sql", "SELECT FROM `Item`")) 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 _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", f"UPDATE {rid} SET b = 2") + + 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", f"UPDATE {rid} SET b = 2") + + 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_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) diff --git a/tests/test_native_save_lazy_fingerprint.py b/tests/test_native_save_lazy_fingerprint.py new file mode 100644 index 0000000..2ad1d40 --- /dev/null +++ b/tests/test_native_save_lazy_fingerprint.py @@ -0,0 +1,153 @@ +"""Native build: save() of a record refuses to overwrite a record that changed after it was read, and the check costs a read nothing. + +The fingerprint of the values a record held as read is taken lazily, at ``modify()``, at the first ``set()`` of a saved record or at +``save()``, never when a read decodes a row into a record. The seam is ``arcadedb._native.records._fingerprint``: these tests count its +calls. The refusal itself is also tested in test_graph_api.py, which runs on both builds. +""" + +import pytest + +import arcadedb +from arcadedb._native import records + + +@pytest.fixture +def fingerprints(monkeypatch): + """The list of calls to the fingerprint function (it still computes the real value).""" + calls = [] + real = records._fingerprint + + def counting(props): + calls.append(dict(props)) + return real(props) + + monkeypatch.setattr(records, "_fingerprint", counting) + return calls + + +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 _stored(db, rid): + row = list(db.query("sql", f"SELECT a, b FROM {rid}"))[0] + return row.get("a"), row.get("b") + + +def test_reads_never_compute_the_fingerprint(temp_db_path, fingerprints): + with arcadedb.create_database(temp_db_path) as db: + rid = _create(db) + fingerprints.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 fingerprints == [] # nothing was fingerprinted by any read path + + mutable = by_rid.modify() + assert ( + len(fingerprints) == 1 + ) # modify() takes it, once, from the values the record holds as read + assert fingerprints[0] == by_rid.to_dict() + by_rid.modify() + assert len(fingerprints) == 1 # and caches it on the record + + mutable.set("a", 10) # the snapshot is already there: no work + assert len(fingerprints) == 1 + with db.transaction(): + mutable.save() # the stored values to compare with, and the stored values after the write + assert len(fingerprints) == 3 + assert _stored(db, rid) == (10, 1) + + +def test_modify_then_save_after_a_concurrent_change_still_raises( + temp_db_path, fingerprints +): + 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", f"UPDATE {rid} SET b = 2") + 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 + + # outside a transaction too, and a fresh read saves + with pytest.raises(arcadedb.ArcadeDBError): + mutable.save() + assert _stored(db, rid) == (1, 2) + fresh = db.lookup_by_rid(rid).modify() + fresh.set("a", 10) + 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 fingerprint 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", f"UPDATE {rid} SET b = 2") + 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, fingerprints +): + """A record this object created has no fingerprint 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 ( + fingerprints == [] + ) # creating and saving a new record fingerprints nothing + rid = str(created.get_identity()) + with db.transaction(): + db.command("sql", f"UPDATE {rid} SET b = 2") + created.set("a", 9) + with pytest.raises(arcadedb.ArcadeDBError): + with db.transaction(): + created.save() + assert _stored(db, rid) == (1, 2) From 7008eb4f176b03ecf24c74a5582eafdb89d6cec5 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 2 Nov 2026 20:08:12 +0900 Subject: [PATCH 2/4] native: keep the f-string statements of the save check out of bandit's B608 (#172) The statements are built from a record's own id (or one a test read); the nosec sits on the one line that builds each. Co-Authored-By: Claude Opus 5.5 --- src/arcadedb/_native/records.py | 5 ++++- tests/test_graph_api.py | 9 +++++++-- tests/test_native_save_lazy_fingerprint.py | 16 ++++++++++++---- 3 files changed, 23 insertions(+), 7 deletions(-) diff --git a/src/arcadedb/_native/records.py b/src/arcadedb/_native/records.py index 96e1c41..c32028e 100644 --- a/src/arcadedb/_native/records.py +++ b/src/arcadedb/_native/records.py @@ -196,7 +196,10 @@ def _save_update(self, db) -> None: if own: db.begin() try: - stored = db.query("sql", f"SELECT FROM {self._rid}").first() + select_sql = ( + f"SELECT FROM {self._rid}" # nosec B608 - _rid is the record's own id + ) + stored = db.query("sql", select_sql).first() if stored is not None and _fingerprint(stored._props) != token: raise _concurrent_modification(self._rid, self._type) after = db.command( diff --git a/tests/test_graph_api.py b/tests/test_graph_api.py index 5c6d6a9..3ad32a5 100644 --- a/tests/test_graph_api.py +++ b/tests/test_graph_api.py @@ -553,6 +553,11 @@ def _create_doc_ab(db): 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") @@ -565,7 +570,7 @@ def test_save_refuses_to_overwrite_a_record_changed_after_it_was_read(temp_db_pa mutable = db.lookup_by_rid(rid).modify() with db.transaction(): # the stored record changes after the read - db.command("sql", f"UPDATE {rid} SET b = 2") + db.command("sql", _bump_b_sql(rid)) mutable.set("a", 10) with pytest.raises(arcadedb.ArcadeDBError) as excinfo: @@ -586,7 +591,7 @@ def test_save_refuses_a_stale_record_from_a_query_result(temp_db_path): row = list(db.query("sql", "SELECT FROM D"))[0] mutable = row.get_element().modify() with db.transaction(): - db.command("sql", f"UPDATE {rid} SET b = 2") + db.command("sql", _bump_b_sql(rid)) mutable.set("a", 10) with pytest.raises(arcadedb.ArcadeDBError): diff --git a/tests/test_native_save_lazy_fingerprint.py b/tests/test_native_save_lazy_fingerprint.py index 2ad1d40..ea33492 100644 --- a/tests/test_native_save_lazy_fingerprint.py +++ b/tests/test_native_save_lazy_fingerprint.py @@ -40,8 +40,16 @@ def _create(db): 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): - row = list(db.query("sql", f"SELECT a, b FROM {rid}"))[0] + 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") @@ -88,7 +96,7 @@ def test_modify_then_save_after_a_concurrent_change_still_raises( with ( db.transaction() ): # another transaction changes the stored record after the read - db.command("sql", f"UPDATE {rid} SET b = 2") + db.command("sql", _bump_b_sql(rid)) mutable.set("a", 10) with pytest.raises(arcadedb.ArcadeDBError) as excinfo: with db.transaction(): @@ -118,7 +126,7 @@ def test_a_change_between_the_read_and_modify_is_not_lost_when_modify_comes_late read = db.lookup_by_rid(rid) with db.transaction(): - db.command("sql", f"UPDATE {rid} SET b = 2") + db.command("sql", _bump_b_sql(rid)) mutable = ( read.modify() ) # late: the stored record is already newer than the values held @@ -145,7 +153,7 @@ def test_a_record_created_here_then_changed_elsewhere_is_refused( ) # creating and saving a new record fingerprints nothing rid = str(created.get_identity()) with db.transaction(): - db.command("sql", f"UPDATE {rid} SET b = 2") + db.command("sql", _bump_b_sql(rid)) created.set("a", 9) with pytest.raises(arcadedb.ArcadeDBError): with db.transaction(): From 3492acbf37fd291396bd96dd2abf54bf2c1fcc72 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Tue, 3 Nov 2026 00:45:57 +0900 Subject: [PATCH 3/4] native: save() of an existing record needs a transaction, as on the JVM build (#172) _save_update opened and committed a transaction of its own when none was active, so a native save() outside a transaction worked while main and the JVM package (Document.save() -> MutableDocument.save()) raise "Transaction not begun". It now opens none: outside a transaction, with auto-transaction off (the engine default, tracked by NativeDatabase._auto_transaction because the native library has no getter), it raises ArcadeDBError before reading or writing anything. The fingerprint check and its ConcurrentModificationException path are unchanged inside the caller's transaction. Tests: test_save_of_an_existing_record_outside_a_transaction_is_refused runs on both builds; the native-only fingerprint test no longer expects a save outside a transaction to succeed. Co-Authored-By: Claude Opus 5.5 --- src/arcadedb/_native/database.py | 2 + src/arcadedb/_native/records.py | 47 +++++++++++----------- tests/test_graph_api.py | 19 +++++++++ tests/test_native_save_lazy_fingerprint.py | 8 ++-- 4 files changed, 47 insertions(+), 29 deletions(-) diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index 3c876fe..2711f6b 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 @@ -1231,6 +1232,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 c32028e..7fb7084 100644 --- a/src/arcadedb/_native/records.py +++ b/src/arcadedb/_native/records.py @@ -186,33 +186,32 @@ 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. The values this object held as read are fingerprinted (once, at modify() or here) and compared - with the values stored now, in the transaction of the update (one is opened when none is active); a stored record that differs - raises ConcurrentModificationException and nothing is written. The fingerprint then moves to the stored values after this save, - so the object's own save is not another transaction's change.""" + with the values stored now, in the caller's transaction; a stored record that differs raises ConcurrentModificationException and + nothing is written. The fingerprint then moves to the stored values after this save, so the object's own save is not another + transaction's change. + + 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)" + ) token = self._token if token is None: token = self._token = _fingerprint(self._props) - own = not db.is_transaction_active() - if own: - db.begin() - try: - select_sql = ( - f"SELECT FROM {self._rid}" # nosec B608 - _rid is the record's own id - ) - stored = db.query("sql", select_sql).first() - if stored is not None and _fingerprint(stored._props) != token: - raise _concurrent_modification(self._rid, self._type) - after = db.command( - "sql", - f"UPDATE {self._rid} MERGE :row RETURN AFTER", - {"row": self._props}, - ).first() - if own: - db.commit() - except BaseException: - if own and db.is_transaction_active(): - db.rollback() - raise + select_sql = ( + f"SELECT FROM {self._rid}" # nosec B608 - _rid is the record's own id + ) + stored = db.query("sql", select_sql).first() + if stored is not None and _fingerprint(stored._props) != token: + raise _concurrent_modification(self._rid, self._type) + after = db.command( + "sql", + f"UPDATE {self._rid} MERGE :row RETURN AFTER", + {"row": self._props}, + ).first() self._token = _fingerprint(after._props) if after is not None else None def _adopt(self, row) -> None: diff --git a/tests/test_graph_api.py b/tests/test_graph_api.py index 3ad32a5..f29ed6b 100644 --- a/tests/test_graph_api.py +++ b/tests/test_graph_api.py @@ -607,6 +607,25 @@ def test_save_refuses_a_stale_record_from_a_query_result(temp_db_path): 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: diff --git a/tests/test_native_save_lazy_fingerprint.py b/tests/test_native_save_lazy_fingerprint.py index ea33492..21dd511 100644 --- a/tests/test_native_save_lazy_fingerprint.py +++ b/tests/test_native_save_lazy_fingerprint.py @@ -107,13 +107,11 @@ def test_modify_then_save_after_a_concurrent_change_still_raises( ) # run_in_transaction retries it assert _stored(db, rid) == (1, 2) # nothing of the refused save is stored - # outside a transaction too, and a fresh read saves - with pytest.raises(arcadedb.ArcadeDBError): - mutable.save() - assert _stored(db, rid) == (1, 2) + # a fresh read saves, in a transaction fresh = db.lookup_by_rid(rid).modify() fresh.set("a", 10) - fresh.save() + with db.transaction(): + fresh.save() assert _stored(db, rid) == (10, 2) From 5cd0a912680337a08ffed894b4a808d15afcb0bf Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Tue, 3 Nov 2026 03:21:45 +0900 Subject: [PATCH 4/4] native: encode the save fingerprint straight to bytes (#172) The stale-save fingerprint canonicalised the record into nested lists and printed it with repr before hashing: about 390 us for a record with 20 properties and a 768-float vector, taken three times per save(). It now appends a type tag and a fixed-width or length-prefixed form of each value to one bytearray (a float or int list is packed in one array call) and hashes that once: about 23 us. The equivalence classes are the ones the old form had (dict and set order, tuple == list, int/float/bool apart, temporals, RID, Decimal, bytes by value); a float ndarray is now equal to the list of its doubles. When the token is taken and where it comes from (RETURN AFTER) are unchanged. Co-Authored-By: Claude Opus 5.5 --- src/arcadedb/_native/records.py | 177 ++++++++++++++++++--- tests/test_native_save_lazy_fingerprint.py | 152 ++++++++++++++++++ 2 files changed, 308 insertions(+), 21 deletions(-) diff --git a/src/arcadedb/_native/records.py b/src/arcadedb/_native/records.py index 7fb7084..b7dccab 100644 --- a/src/arcadedb/_native/records.py +++ b/src/arcadedb/_native/records.py @@ -6,7 +6,13 @@ """ import hashlib +import struct +import sys +from array import array from collections.abc import Mapping +from datetime import date, datetime +from decimal import Decimal +from operator import itemgetter from typing import Any, Dict, List from ..exceptions import ArcadeDBError @@ -29,35 +35,164 @@ def _java_object(): _UNSAVED = "#-1:-1" -_SCALARS = frozenset((str, int, float, bool, type(None))) - - -def _canon(value): - """``value`` as plain nested lists, tuples and scalars that compare and print the same for two equal values, whatever order a - dict or a set was built in.""" - if type(value) in _SCALARS: - return value - if isinstance(value, Mapping): - return ( - "m", - sorted(((_canon(k), _canon(v)) for k, v in value.items()), key=repr), - ) - if isinstance(value, (list, tuple)): - return [_canon(v) for v in value] - if isinstance(value, (set, frozenset)): - return ("s", sorted((_canon(v) for v in value), key=repr)) - tolist = getattr(value, "tolist", None) # a numpy array +_pack_d = struct.Struct(" None: + data = text.encode("utf-8", "surrogatepass") + out += tag + out += _pack_n(len(data)) + out += data + + +def _array(code: str, tag: bytes, values, out: bytearray) -> None: + packed = array(code, values) + if _BIG_ENDIAN: + packed.byteswap() + out += tag + out += _pack_n(len(packed)) + out += packed.tobytes() + + +def _enc_seq(value, out: bytearray) -> None: + """A list or tuple. A list of only floats (a vector) or only ints that fit 64 bits is packed in one call; any other list is its elements + one by one. Which form a list takes depends only on its values.""" + kinds = set(map(type, value)) + if kinds == {float}: + # every NaN is the same bits only when the elements came from one decoder: a vector read twice decodes alike + _array("d", b"V", value, out) + return + if kinds == {int}: + try: + _array("q", b"J", value, out) + return + except OverflowError: + pass + out += b"l" + out += _pack_n(len(value)) + for item in value: + _enc(item, out) + + +def _enc(value, out: bytearray) -> None: + """Append a canonical byte form of ``value`` to ``out``: a type tag, then the value (a length first for anything of variable size). + + Equal values give equal bytes whatever order a dict or a set was built in; a tuple is a list; int, float and bool differ from each + other (``1``, ``1.0`` and ``True`` are three values); a float ndarray is the list of its doubles, so a vector read back as a + numpy float32 array equals the same vector read as a list of Python floats (float32 widens to the same double); a NumPy scalar + is the Python scalar. A type with no form here (a ``datetime`` subclass, say) is its type name and ``repr``. + """ + t = type(value) + if t is str: + _text(b"s", value, out) + elif t is float: + out += b"d" + out += _NAN if value != value else _pack_d(value) + elif t is int: + if -9223372036854775808 <= value <= 9223372036854775807: + out += b"i" + out += _pack_q(value) + else: + data = value.to_bytes((value.bit_length() + 8) // 8, "little", signed=True) + out += b"I" + out += _pack_n(len(data)) + out += data + elif t is bool: + out += b"t" if value else b"f" + elif value is None: + out += b"N" + elif t is list or t is tuple: + _enc_seq(value, out) + elif t is dict: + _enc_map(value, out) + elif t is datetime: + _text(b"z", value.isoformat(), out) + elif t is date: + _text(b"D", value.isoformat(), out) + elif t is Decimal: + _text(b"n", str(value), out) + elif t is RID: + _text(b"R", value.text, out) + elif t is bytes: + out += b"b" + out += _pack_n(len(value)) + out += value + elif isinstance(value, Mapping): + _enc_map(value, out) + elif isinstance(value, (list, tuple)): + _enc_seq(value, out) + elif isinstance(value, (set, frozenset)): + parts = [] + for item in value: + buf = bytearray() + _enc(item, buf) + parts.append(bytes(buf)) + parts.sort() + out += b"e" + out += _pack_n(len(parts)) + for part in parts: + out += part + else: + _enc_other(value, t, out) + + +def _enc_map(value, out: bytearray) -> None: + out += b"m" + out += _pack_n(len(value)) + if all(type(k) is str for k in value): + for k, v in sorted(value.items(), key=_first): + _text(b"s", k, out) + _enc(v, out) + return + parts = [] + for k, v in value.items(): + buf = bytearray() + _enc(k, buf) + _enc(v, buf) + parts.append(bytes(buf)) + parts.sort() + for part in parts: + out += part + + +def _enc_other(value, t, out: bytearray) -> None: + tolist = getattr(value, "tolist", None) # a numpy array or scalar if tolist is not None: - return ("a", str(getattr(value, "dtype", "")), _canon(tolist())) - return value # datetime, date, Decimal, RID and bytes print by value + dtype = getattr(value, "dtype", None) + if ( + getattr(dtype, "kind", "") == "f" + and getattr(value, "ndim", 0) == 1 + and value.size + ): + # a float vector: its doubles in one call (float32 widens to the double tolist() gives) + doubles = value.astype(" bytes: """A fingerprint of the values a record holds. Called only when a record is about to be changed (``modify()``, the first ``set()`` of a saved record, ``save()``), never when one is read: the read path keeps the values the record already holds, and this runs on them later. Two reads of one stored record decode to equal values, so equal fingerprints mean an unchanged record. + + The values are encoded straight to bytes (see ``_enc``) and hashed once. Building a canonical copy of the values and printing it cost + about 500 us for a record with a 768-float vector, three times per ``save()``; the packed form costs a few tens of microseconds. """ - return hashlib.blake2b(repr(_canon(props)).encode(), digest_size=16).digest() + out = bytearray() + _enc(props, out) + return hashlib.blake2b(out, digest_size=16).digest() def _concurrent_modification(rid: str, type_name: str) -> ArcadeDBError: diff --git a/tests/test_native_save_lazy_fingerprint.py b/tests/test_native_save_lazy_fingerprint.py index 21dd511..cdc100d 100644 --- a/tests/test_native_save_lazy_fingerprint.py +++ b/tests/test_native_save_lazy_fingerprint.py @@ -5,6 +5,10 @@ calls. The refusal itself is also tested in test_graph_api.py, which runs on both builds. """ +from datetime import date, datetime +from decimal import Decimal + +import numpy as np import pytest import arcadedb @@ -157,3 +161,151 @@ def test_a_record_created_here_then_changed_elsewhere_is_refused( with db.transaction(): created.save() assert _stored(db, rid) == (1, 2) + + +# ---- the fingerprint itself: a direct byte encoding, equal for equal values whatever path read them + + +def _medium_record(): + """About what a medium record holds: 20 mixed properties and a 768-float vector (float32 values, as the engine stores them).""" + rng = np.random.default_rng(7) + props = { + "i": 12, + "big": 2**40, + "d": 3.5, + "neg_zero": -0.0, + "t": True, + "f": False, + "n": None, + "s": "text with é and 日", + "empty": "", + "day": date(2020, 1, 2), + "at": datetime(2020, 1, 2, 3, 4, 5, 678000), + "dec": Decimal("12.50"), + "ints": [1, 2, 3], + "mixed": [1, "a", 2.5, None], + "strs": ["x", "y"], + "map": {"b": 1, "a": [0.5, 0.25], "c": {"z": "q"}}, + "doubles": [0.1, 0.2], + "blob": b"\x00\x01", + "neg": -7, + "zero": 0, + } + props["vec"] = rng.random(768).astype(np.float32).tolist() + assert len(props) == 21 + return props + + +def test_every_property_and_every_vector_element_changes_the_fingerprint(): + props = _medium_record() + base = records._fingerprint(props) + assert records._fingerprint(dict(props)) == base + seen = {base} + for name in props: + if name == "vec": + continue + changed = dict(props) + value = props[name] + if isinstance(value, bool): + changed[name] = not value + elif value is None: + changed[name] = 0 + elif isinstance(value, (int, float)): + changed[name] = value + 1 + elif isinstance(value, str): + changed[name] = value + "!" + elif isinstance(value, (datetime, date)): + changed[name] = value.replace(year=2021) + elif isinstance(value, Decimal): + changed[name] = Decimal("12.5") # same number, other scale: another value + elif isinstance(value, bytes): + changed[name] = value + b"\x02" + elif isinstance(value, list): + changed[name] = value + [value[0]] + else: + changed[name] = {**value, "new": 1} + fingerprint = records._fingerprint(changed) + assert fingerprint != base, name + seen.add(fingerprint) + for index in (0, 1, 383, 767): + changed = dict(props) + changed["vec"] = list(props["vec"]) + changed["vec"][index] = changed["vec"][index] + 0.5 + fingerprint = records._fingerprint(changed) + assert fingerprint != base, index + seen.add(fingerprint) + shorter = {**props, "vec": props["vec"][:-1]} + assert records._fingerprint(shorter) != base + assert len(seen) == len(props) + 4 + + +def test_the_fingerprint_does_not_depend_on_order_or_on_list_versus_array(): + props = _medium_record() + base = records._fingerprint(props) + reordered = dict(reversed(list(props.items()))) + reordered["map"] = { + "c": {"z": "q"}, + "a": [0.5, 0.25], + "b": 1, + } + assert records._fingerprint(reordered) == base + # a vector read as a numpy float32 array is the vector read as a list of Python floats + as_array = {**props, "vec": np.asarray(props["vec"], dtype=np.float32)} + assert records._fingerprint(as_array) == base + assert ( + records._fingerprint({**props, "ints": (1, 2, 3)}) == base + ) # a tuple is a list + # the types stay apart: 1, 1.0 and True are three values; so are a list, a set, a map, a string and bytes + kinds = [1, 1.0, True, "1", [1], {1}, {"1": 1}, b"1"] + assert len({records._fingerprint({"k": k}) for k in kinds}) == len(kinds) + assert records._fingerprint({"k": {1, 2}}) == records._fingerprint({"k": {2, 1}}) + assert records._fingerprint({"k": float("nan")}) == records._fingerprint( + {"k": float("nan")} + ) + assert records._fingerprint({"k": 0.0}) != records._fingerprint({"k": -0.0}) + assert records._fingerprint({"k": 2**70}) != records._fingerprint({"k": 2**70 + 1}) + assert records._fingerprint({"k": [2**70, 1]}) != records._fingerprint( + {"k": [2**70, 2]} + ) + + +def test_one_record_has_one_fingerprint_on_every_read_path(temp_db_path): + """lookup_by_rid, a SELECT and the row UPDATE .. RETURN AFTER gives back decode to the same fingerprint, and a change to the stored + record (one vector element) changes it.""" + props = _medium_record() + vec = props.pop("vec") + with arcadedb.create_database(temp_db_path) as db: + 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") + for name, value in {**props, "vec": vec}.items(): + vertex.set(name, value) + vertex.save() + rid = str(vertex.get_identity()) + + by_rid = db.lookup_by_rid(rid) + select_sql = ( + f"SELECT FROM {rid}" # nosec B608 - rid is a record id the test read + ) + selected = db.query("sql", select_sql).first() + with db.transaction(): + after = db.command( + "sql", + f"UPDATE {rid} MERGE :row RETURN AFTER", # nosec B608 - rid is a record id the test read + {"row": {"i": props["i"]}}, + ).first() + one = records._fingerprint(by_rid._props) + assert records._fingerprint(selected._props) == one + assert records._fingerprint(after._props) == one + assert len(by_rid._props["vec"]) == 768 + + changed = list(by_rid._props["vec"]) + changed[500] = changed[500] + 0.5 + with db.transaction(): + db.command( + "sql", + f"UPDATE {rid} SET vec = :v", # nosec B608 - rid is a record id the test read + {"v": changed}, + ).close() + assert records._fingerprint(db.lookup_by_rid(rid)._props) != one