From 8ae3075b7278311118dcabe0adc80869f8394bd3 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Thu, 12 Nov 2026 09:33:21 +0900 Subject: [PATCH] Test: a vector search finds the vector committed in the call before, in the same asyncio task and in another one, on both wheels (refs #306) An agent that remembers things inserts a few vectors per turn and searches right after, from one task or several. The new test runs 200 committed inserts per test, each followed by a search, while the graph rebuilds in the background (the corpus starts at 300 vectors, so the rebuild threshold of 100 pending changes is crossed): once in the task that wrote, once in readers that get the id from a queue after the commit. Every call runs in a worker thread through asyncio.to_thread, as the concurrency page says, so an insert and the next search of one task can run on different threads. ef_search covers the corpus, so a miss would mean the write was not seen and not an approximate walk. The file uses no Java object and runs on the JVM build and the native build. The vectors page says what the agent may rely on (committed writes are found by the next search, from any task), and the concurrency page gets the anchor the new sentence links to. Item 2 of #306 only: the asyncio facade waits for a decision on its names, see the issue. Co-Authored-By: Claude Sonnet 5.5 --- .../ROOT/pages/how-to/concurrency.adoc | 1 + docs/modules/ROOT/pages/how-to/vectors.adoc | 3 + tests/test_vector_read_your_writes.py | 120 ++++++++++++++++++ 3 files changed, 124 insertions(+) create mode 100644 tests/test_vector_read_your_writes.py diff --git a/docs/modules/ROOT/pages/how-to/concurrency.adoc b/docs/modules/ROOT/pages/how-to/concurrency.adoc index aadfb05..696b18b 100644 --- a/docs/modules/ROOT/pages/how-to/concurrency.adoc +++ b/docs/modules/ROOT/pages/how-to/concurrency.adoc @@ -190,6 +190,7 @@ def add_one(key): run_repeatable_read(db, lambda: add_one(3)) ---- +[[asyncio]] == asyncio Call the bindings from asyncio through `asyncio.to_thread` or `loop.run_in_executor`. A call made directly on the event loop blocks the loop for as long as the call takes. diff --git a/docs/modules/ROOT/pages/how-to/vectors.adoc b/docs/modules/ROOT/pages/how-to/vectors.adoc index 741bf83..e7c13da 100644 --- a/docs/modules/ROOT/pages/how-to/vectors.adoc +++ b/docs/modules/ROOT/pages/how-to/vectors.adoc @@ -507,6 +507,9 @@ print(notes.get_size()) An agent that writes one record per turn and searches right after needs no setting: the rebuild runs in the background and the searches keep running while it does. +// Pinned by tests/test_vector_read_your_writes.py, on both packages: 200 inserts, each followed by a search, from the writing task and from other tasks. +A search finds every vector whose transaction committed before the search started. Insert in one call and search in the next, from the same task or from another one, and the new vector is in the result, also while the graph rebuilds in the background. From asyncio, run each call with `asyncio.to_thread` as xref:how-to/concurrency.adoc#asyncio[Concurrency] describes: the insert and the search of one task can run on different threads and the search still finds the vector. This holds for committed writes only. It says nothing about a search made while the writing transaction is still open. + If the tail of the search latency matters more than the agent's total time, you can move the rebuild to a moment of your choosing: raise `mutations_before_rebuild` far above the number of writes, set the database setting `arcadedb.vectorIndex.maxDeltaScanRatio` to 0, and call `build_graph_now()` right before a step in which the agent does not search, such as its call to the language model. The table compares the two on {jvm-pkg-name} and {native-pkg-name} 26.11.1.dev0: 768-dimensional COSINE vectors, 10,000 records at the start, one insert and four k = 10 searches per turn at efSearch 100, `build_graph_now()` every 1,000 turns, two efficiency cores of a laptop, the first 3,000 turns. Each ratio is the median over paired runs of the same package; the two packages ran separately and are not compared with each other. diff --git a/tests/test_vector_read_your_writes.py b/tests/test_vector_read_your_writes.py new file mode 100644 index 0000000..8a42f90 --- /dev/null +++ b/tests/test_vector_read_your_writes.py @@ -0,0 +1,120 @@ +"""A vector search finds every vector whose transaction committed before the search started (ArcadeData/arcadedb-python#306). + +An agent that remembers things writes a few vectors per turn and searches right after, from one asyncio task or from several. The +engine buffers the writes of a built index and rebuilds the graph in the background, so this pins what the agent may rely on: an +insert in one call and a search in the next call finds the vector, in the same task and in another one, also while the graph rebuilds. + +The calls are the ones the concurrency page tells asyncio code to use: each blocking call runs in a worker thread through +``asyncio.to_thread``, so one task's insert and its next search can run on different threads. The corpus starts at BASE vectors, so the +rebuild threshold is max(100, 20 percent of the graph) = 100 pending changes, and the 200 inserts of a test cross it: the searches meet the +delta buffer first and the rebuilt graph later. ``ef_search`` is as large as the corpus on purpose, so a miss cannot be the approximate +walk of a large graph: every vector is the nearest neighbour of itself, at distance 0, and a different result means the write was not seen. +The file uses no Java object, so it runs on the JVM build and on the native build alike. +""" + +import asyncio +import math +import random + +import numpy as np +import pytest + +import arcadedb + +DIM = 8 +BASE = 300 +WRITERS = 4 +PER_WRITER = 50 +TOTAL = WRITERS * PER_WRITER +READERS = 3 +EF_SEARCH = 400 # at least the number of vectors the corpus reaches (BASE + TOTAL = 500 is walked almost whole) + + +def _vec(i): + """A deterministic unit vector whose direction is unique to `i` (see tests/test_vector_delta_visibility.py for why a unit vector).""" + rnd = random.Random(i) # nosec B311 - deterministic test corpus, not security + v = [rnd.uniform(-1.0, 1.0) for _ in range(DIM)] + norm = math.sqrt(sum(x * x for x in v)) or 1.0 + return np.asarray([x / norm for x in v], dtype=np.float32) + + +def _insert(db, i): + """One committed insert, retried on a conflict with another writer (what the concurrency page recommends).""" + + def write(): + db.command("sql", "INSERT INTO Doc SET vid = ?, embedding = ?", i, _vec(i)) + + db.run_in_transaction(write, retries=60) + + +def _nearest_ids(index, i): + hits = index.find_nearest(_vec(i), k=5, ef_search=EF_SEARCH) + return [record.get("vid") for record, _distance in hits] + + +def _count(db): + return db.query("sql", "SELECT count(*) AS n FROM Doc").first().get("n") + + +@pytest.fixture +def corpus(tmp_path): + db = arcadedb.create_database(str(tmp_path / "read_your_writes")) + db.command("sql", "CREATE VERTEX TYPE Doc") + db.command("sql", "CREATE PROPERTY Doc.embedding ARRAY_OF_FLOATS") + with db.transaction(): + for i in range(BASE): + db.command("sql", "INSERT INTO Doc SET vid = ?, embedding = ?", i, _vec(i)) + index = db.create_vector_index("Doc", "embedding", dimensions=DIM) + yield db, index + db.drop() + + +def test_a_task_finds_the_vector_it_committed_in_the_call_before(corpus): + db, index = corpus + missed = [] + + async def agent(n): + for step in range(PER_WRITER): + i = BASE + n * PER_WRITER + step + await asyncio.to_thread(_insert, db, i) + ids = await asyncio.to_thread(_nearest_ids, index, i) + if ids[:1] != [i]: + missed.append((i, ids)) + + async def main(): + await asyncio.gather(*(agent(n) for n in range(WRITERS))) + + asyncio.run(main()) + assert ( + missed == [] + ), f"{len(missed)} of {TOTAL} committed vectors were not found by the next search: {missed[:5]}" + assert _count(db) == BASE + TOTAL + + +def test_another_task_finds_the_vector_once_the_writing_task_has_committed_it(corpus): + db, index = corpus + missed = [] + + async def writer(queue): + for step in range(TOTAL): + i = BASE + step + await asyncio.to_thread(_insert, db, i) + await queue.put(i) # announced only after the commit returned + for _ in range(READERS): + await queue.put(None) + + async def reader(queue): + while (i := await queue.get()) is not None: + ids = await asyncio.to_thread(_nearest_ids, index, i) + if ids[:1] != [i]: + missed.append((i, ids)) + + async def main(): + queue = asyncio.Queue() + await asyncio.gather(writer(queue), *(reader(queue) for _ in range(READERS))) + + asyncio.run(main()) + assert ( + missed == [] + ), f"{len(missed)} of {TOTAL} committed vectors were not found by another task: {missed[:5]}" + assert _count(db) == BASE + TOTAL