From a815a57a62877f2b6c85c75dc7dbd3ae0a97257f Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sun, 25 Oct 2026 21:32:16 +0900 Subject: [PATCH 1/3] Native: get_stats, get_metadata and PQ search work, AsyncExecutor.transaction runs its block; the module docstring says what is true VectorIndex.get_stats reads the schema:index: row (the counters, not the row's own columns). get_metadata and get_quantization ask the engine through one new glue entry point (arcadedb_n_vindex, op metadata), so an index this session did not create reports the same keys as on the JVM build, defaults included. find_nearest_approximate runs the engine's PQ search through the same entry point (op approximate): no statement runs it. warm_up runs the first search that loads a persisted graph. AsyncExecutor.transaction is a new op of arcadedb_n_async: the engine begins the transaction on a writer thread, Python runs the block in the upcall, and a Python exception is thrown on the Java side so the writer reports it. The block runs in the upcall of #67, and a Python exception comes back through its out slot as {"error": ...}, which the Java side throws so the writer rolls back and reports it. Tests first, shared by both builds: tests/test_vector_index_introspection.py and tests/test_async_transaction.py (18 tests on the JVM build, 17 of 18 on native, see the PR). The two get_java_* calls stay JVM-only and are documented as such. The module docstring of _native/__init__.py listed far more as missing than is true. Co-Authored-By: Claude Sonnet 5.5 --- docs/modules/ROOT/pages/installation.adoc | 2 +- docs/modules/ROOT/pages/which-package.adoc | 6 +- .../org/humemai/nativelib/NativeExtras.java | 108 +++++++++ src/arcadedb/_native/__init__.py | 24 +- src/arcadedb/_native/extras.py | 25 +- src/arcadedb/_native/lib.py | 1 + src/arcadedb/_native/schema_objects.py | 191 ++++++++++----- tests/test_async_transaction.py | 97 ++++++++ tests/test_vector_index_introspection.py | 227 ++++++++++++++++++ 9 files changed, 608 insertions(+), 73 deletions(-) create mode 100644 tests/test_async_transaction.py create mode 100644 tests/test_vector_index_introspection.py diff --git a/docs/modules/ROOT/pages/installation.adoc b/docs/modules/ROOT/pages/installation.adoc index 733a505..4f9456f 100644 --- a/docs/modules/ROOT/pages/installation.adoc +++ b/docs/modules/ROOT/pages/installation.adoc @@ -147,7 +147,7 @@ Each Python process starts its own JVM, and each JVM sees the whole container li == Native build notes (`{native-pkg-name}`) -The native package is the same engine compiled to a GraalVM native-image shared library. It needs no JVM and no JPype, so the JVM options above do not apply to it. A few API calls raise `NotOnNativeYet` (a `NotImplementedError`) instead of misbehaving. The list is in xref:which-package.adoc[Which package?]. Report problems in the {repo-url}/issues[issue tracker] and say which package you use. +The native package is the same engine compiled to a GraalVM native-image shared library. It needs no JVM and no JPype, so the JVM options above do not apply to it. The calls that hand out a Java object (`get_java_database`, `get_java_document`) raise `NotOnNativeYet` (a `NotImplementedError`) instead of misbehaving. The list is in xref:which-package.adoc[Which package?]. Report problems in the {repo-url}/issues[issue tracker] and say which package you use. == Next steps diff --git a/docs/modules/ROOT/pages/which-package.adoc b/docs/modules/ROOT/pages/which-package.adoc index 75bf4c8..4cb20af 100644 --- a/docs/modules/ROOT/pages/which-package.adoc +++ b/docs/modules/ROOT/pages/which-package.adoc @@ -82,8 +82,8 @@ The API is the same. The runtime is not. These are the differences you can plan | Linux x86_64 (glibc 2.34 or later), Linux aarch64, macOS arm64, Windows x64. The Windows wheel is the newest: it is built in CI, and its test run is the least proven of the four. | Linux x86_64 (glibc 2.34 or later), Linux aarch64, macOS arm64, Windows x64. -| Java-object APIs (`get_java_database`, `get_java_*`) -| Not available. +| Java-object APIs (`get_java_database`, `get_java_document`, `get_java_*`) +| Not available. They raise `NotOnNativeYet`. | Available. | JVM options (`jvm_kwargs`, `start_jvm`, `-X` and `-D` flags) @@ -91,7 +91,7 @@ The API is the same. The runtime is not. These are the differences you can plan | Available. See xref:installation.adoc#jvm-options[JVM options]. | A few API calls -| Raise `NotOnNativeYet` (a `NotImplementedError`): `AsyncExecutor.transaction`, the approximate search `find_nearest_approximate` over product quantization, `get_metadata` of a foreign index, and `get_stats`. +| None, except the Java-object calls above, which raise `NotOnNativeYet` (a `NotImplementedError`). | Work. |=== diff --git a/native/src/main/java/org/humemai/nativelib/NativeExtras.java b/native/src/main/java/org/humemai/nativelib/NativeExtras.java index 9903789..7fc3797 100644 --- a/native/src/main/java/org/humemai/nativelib/NativeExtras.java +++ b/native/src/main/java/org/humemai/nativelib/NativeExtras.java @@ -25,6 +25,7 @@ import com.arcadedb.database.RID; import com.arcadedb.database.async.DatabaseAsyncExecutor; import com.arcadedb.database.async.ErrorCallback; +import com.arcadedb.database.async.OkCallback; import com.arcadedb.database.async.NewEdgeCallback; import com.arcadedb.engine.ErrorRecordCallback; import com.arcadedb.engine.WALFile; @@ -39,6 +40,11 @@ import com.arcadedb.function.CypherFunctionRegistry; import com.arcadedb.function.FunctionRegistry; import com.arcadedb.function.sql.DefaultSQLFunctionFactory; +import com.arcadedb.index.IndexInternal; +import com.arcadedb.index.TypeIndex; +import com.arcadedb.index.vector.LSMVectorIndex; +import com.arcadedb.schema.LSMVectorIndexMetadata; +import com.arcadedb.utility.Pair; import com.arcadedb.function.sql.SQLFunctionAbstract; import com.arcadedb.graph.Edge; import com.arcadedb.graph.GraphBatch; @@ -63,8 +69,10 @@ import java.util.ArrayList; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; @@ -269,6 +277,81 @@ public static CCharPointer dbSet(final IsolateThread thread, final long h, final } } + // ---- vector indexes: what no statement reports (the engine's configuration of an index) and the PQ search no statement runs + + private static String q(final String text) { + return text == null ? "null" : "\"" + text.replace("\\", "\\\\").replace("\"", "\\\"") + "\""; + } + + /** + * ops: metadata {type, prop} | approximate {type, prop, vector, k, allowed}. The index is the one the type has on the property; its bucket + * indexes are searched in turn (the caller merges). metadata describes the first bucket index, as the JPype wrapper does. + */ + @CEntryPoint(name = "arcadedb_n_vindex") + @SuppressWarnings("unchecked") + public static CCharPointer vindex(final IsolateThread thread, final long dbh, final CCharPointer op, final CCharPointer argsJson) { + try { + final Database db = ArcadeNative.db(dbh); + final Map a = args(argsJson); + final String type = (String) a.get("type"), prop = (String) a.get("prop"); + final TypeIndex typeIndex = db.getSchema().getType(type).getPolymorphicIndexByProperties(prop); + if (typeIndex == null) + throw new IllegalArgumentException("No vector index on " + type + "[" + prop + "]"); + final List parts = new ArrayList<>(); + for (final IndexInternal index : typeIndex.getIndexesOnBuckets()) + if (index instanceof LSMVectorIndex lsm) + parts.add(lsm); + if (parts.isEmpty()) + throw new IllegalArgumentException("Underlying index is not an LSM vector index"); + final StringBuilder sb = new StringBuilder("{\"ok\":true"); + switch (ArcadeNative.str(op)) { + case "metadata" -> { + final LSMVectorIndex first = parts.get(0); + final LSMVectorIndexMetadata m = first.getMetadata(); + sb.append(",\"bucket\":").append(q(first.getName())).append(",\"type_name\":").append(q(first.getTypeName())); + sb.append(",\"dimensions\":").append(m.dimensions).append(",\"similarity_function\":").append(q(String.valueOf(m.similarityFunction))); + sb.append(",\"id_property\":").append(q(m.idPropertyName)).append(",\"quantization\":").append(q(String.valueOf(m.quantizationType))); + sb.append(",\"max_connections\":").append(m.maxConnections).append(",\"beam_width\":").append(m.beamWidth); + sb.append(",\"ef_search\":").append(m.efSearch).append(",\"graph_build_cache_size\":").append(m.graphBuildCacheSize); + sb.append(",\"mutations_before_rebuild\":").append(m.mutationsBeforeRebuild).append(",\"store_vectors_in_graph\":").append(m.storeVectorsInGraph); + sb.append(",\"add_hierarchy\":").append(m.addHierarchy).append(",\"build_state\":").append(q(m.buildState)); + sb.append(",\"pq_subspaces\":").append(m.pqSubspaces).append(",\"pq_clusters\":").append(m.pqClusters); + sb.append(",\"pq_center_globally\":").append(m.pqCenterGlobally).append(",\"pq_training_limit\":").append(m.pqTrainingLimit); + } + case "approximate" -> { + final List vector = (List) a.get("vector"); + final float[] query = new float[vector.size()]; + for (int i = 0; i < query.length; i++) + query[i] = ((Number) vector.get(i)).floatValue(); + final int k = intOf(a.get("k"), 10); + Set allowed = null; + if (a.get("allowed") instanceof List rids) { + allowed = new HashSet<>(); + for (final Object rid : rids) + allowed.add(new RID(String.valueOf(rid))); + } + sb.append(",\"hits\":["); + boolean firstHit = true; + for (final LSMVectorIndex lsm : parts) { + final List> hits = allowed != null ? lsm.findNeighborsFromVectorApproximate(query, k, allowed) + : lsm.findNeighborsFromVectorApproximate(query, k); + for (final Pair hit : hits) { + if (!firstHit) + sb.append(','); + firstHit = false; + sb.append("[\"").append(hit.getFirst()).append("\",").append(hit.getSecond()).append(']'); + } + } + sb.append(']'); + } + default -> throw new IllegalArgumentException("unknown vector index op " + ArcadeNative.str(op)); + } + return ArcadeNative.out(sb.append('}').toString()); + } catch (final Throwable t) { + return ArcadeNative.err(t); + } + } + // ---- GraphBatch private record BatchEntry(GraphBatch batch, Database db) { @@ -543,6 +626,31 @@ public void onError(final Exception e) { else ex.query(lang, text, callback); } + case "transaction" -> { + // a block that runs on an async writer inside a transaction: Python runs it in the upcall; its failure is thrown here so the + // writer rolls back, retries a NeedRetryException and reports to the error callback (ok / err: per-call callbacks, slot: one writer) + final long okCb = a.get("ok") instanceof Number n ? n.longValue() : 0L; + final Database.TransactionScope scope = () -> { + final String reply = upcall(cb, event("tx", "")); + if (reply != null && reply.startsWith("{\"error\"")) + throw new RuntimeException(String.valueOf(((Map) Json.parse(reply)).get("error"))); + }; + final int retries = intOf(a.get("retries"), 1); + final OkCallback ok = okCb == 0 ? null : () -> { + try { + fire("transaction ok callback", okCb, event("ok", "")); + } catch (final Throwable e) { + swallowed("transaction ok callback", e); // the writer goes on + } + }; + final ErrorCallback error = errCb == 0 ? null : errorCallback(errCb); + if (a.get("retries") == null && ok == null && error == null && a.get("slot") == null) + ex.transaction(scope); + else if (a.get("slot") instanceof Number slot) + ex.transaction(scope, retries, ok, error, slot.intValue()); + else + ex.transaction(scope, retries, ok, error); + } case "on_ok" -> ex.onOk(() -> { try { fire("on_ok callback", cb, event("ok", "")); diff --git a/src/arcadedb/_native/__init__.py b/src/arcadedb/_native/__init__.py index b413293..47939e3 100644 --- a/src/arcadedb/_native/__init__.py +++ b/src/arcadedb/_native/__init__.py @@ -1,21 +1,25 @@ -"""The native backend: the engine as a GraalVM native-image library, no JVM and no JPype. +"""The native backend: the engine as a GraalVM native-image shared library (``libarcadedb``) that Python loads with ``ctypes``. No JVM, no JPype. -It is the only backend of the `arcadedb` wheel. In a source checkout or in the `arcadedb-jvm` wheel, ``ARCADEDB_BACKEND=native`` -selects it (with ``ARCADEDB_NATIVE_LIB`` pointing at a library you built); the default there is JPype. The engine runs as a GraalVM native-image shared -library (``libarcadedb``) that Python loads with ``ctypes``: no JVM, no JPype. The surface is the DSL-first core: -databases, ``query``/``command`` for SQL and openCypher, transactions, results and the in-process server. +It is the only backend of the ``arcadedb`` wheel. In a source checkout or in the ``arcadedb-jvm`` wheel, ``ARCADEDB_BACKEND=native`` selects it +(with ``ARCADEDB_NATIVE_LIB`` pointing at a library you built); the default there is JPype. The public API is the JVM build's, and the same +test suite (``tests/``) runs against both. -What is not on native yet (each raises :class:`NotOnNativeYet`): ``GraphBatch``, the -async executor, the Java-object APIs (``Document``/``Vertex``/``Edge`` wrappers, ``new_vertex``, ``new_document``, -``schema``, ``get_java_database``), the columnar/Arrow readers other than a simple row conversion, importers and -exporters that go through Java objects, and every JVM setting (``jvm_kwargs``, ``start_jvm``). +On native: databases and the factory, ``query``/``command`` (SQL and openCypher), transactions and ``run_in_transaction``, results as rows, +columns and Arrow, pandas, polars and numpy frames, the ``Document``/``Vertex``/``Edge`` wrappers, the schema (types, properties, indexes), +vector indexes (``find_nearest``, the PQ ``find_nearest_approximate``, ``get_metadata``, ``get_stats``), ``insert_many``, ``GraphBatch``, +the ``AsyncExecutor`` (``transaction`` included), Python functions in SQL and record callbacks, importers and exporters, and the in-process +server. + +Not on native, by design (each raises :class:`NotOnNativeYet`): the Java objects of the JVM build. These are ``get_java_database``, +``get_java_document``, and the Java object a ``Schema``, type or index method returns where the native wrapper has no such method. +There is no JVM to start or tune: ``jvm_kwargs`` is ignored with a warning. """ import os class NotOnNativeYet(NotImplementedError): - """The feature needs the Java object API, which the native backend does not expose (yet).""" + """The call hands out a Java object of the JVM build (``get_java_database``, ``get_java_document``, a Schema method's Java object), which the native build does not have.""" # The backend used when ARCADEDB_BACKEND is unset. "jpype" here, which is the arcadedb-jvm wheel; the build of the native diff --git a/src/arcadedb/_native/extras.py b/src/arcadedb/_native/extras.py index 072184e..b1042e6 100644 --- a/src/arcadedb/_native/extras.py +++ b/src/arcadedb/_native/extras.py @@ -693,8 +693,29 @@ def on_error(message): "scan_type", type=type_name, polymorphic=bool(polymorphic), cb=cb, err=err ) - def transaction(self, *args, **kwargs): - not_yet("AsyncExecutor.transaction") + def transaction( + self, + tx_block: Callable[[], None], + retries: Optional[int] = None, + ok_callback: Optional[Callable[[], None]] = None, + error_callback: Optional[Callable[[Exception], None]] = None, + slot: Optional[int] = None, + ): + """Run ``tx_block`` on an async writer inside a transaction (the engine begins it and commits or rolls it back, retrying a conflict ``retries`` times).""" + + def run(message): + tx_block() + + args = {"cb": _register(run)} + if retries is not None: + args["retries"] = int(retries) + if ok_callback is not None: + args["ok"] = _register(lambda message: ok_callback()) + if error_callback is not None: + args["err"] = self._error_callback(error_callback) + if slot is not None: + args["slot"] = int(slot) + self._op("transaction", **args) def wait_completion(self, timeout_ms: Optional[int] = None): if timeout_ms is None: diff --git a/src/arcadedb/_native/lib.py b/src/arcadedb/_native/lib.py index 95cc7a3..1683f58 100644 --- a/src/arcadedb/_native/lib.py +++ b/src/arcadedb/_native/lib.py @@ -176,6 +176,7 @@ class _Params(ctypes.Structure): "arcadedb_n_gb_create": [P, L, S], "arcadedb_n_gb": [P, L, S, S], "arcadedb_n_async": [P, L, S, S], + "arcadedb_n_vindex": [P, L, S, S], "arcadedb_n_import": [P, L, S], "arcadedb_n_export": [P, L, S, S, I, S], "arcadedb_n_cb": [P, S, S], diff --git a/src/arcadedb/_native/schema_objects.py b/src/arcadedb/_native/schema_objects.py index 455b6fc..3589ad1 100644 --- a/src/arcadedb/_native/schema_objects.py +++ b/src/arcadedb/_native/schema_objects.py @@ -101,14 +101,30 @@ def getTypeName(self): def getPropertyNames(self): return [self._prop] + def engine_info(self) -> dict: + """The engine's configuration of the index (the defaults it chose included), read once: it does not change after the index exists.""" + info = self.__dict__.get("_engine_info") + if info is None: + from . import lib + from .extras import _call, _enc + + info = self._engine_info = _call( + "Failed to read the vector index metadata", + "arcadedb_n_vindex", + self._db._h, + lib.b("metadata"), + lib.b(_enc({"type": self._type, "prop": self._prop})), + ) + return info + def getMetadata(self): - """Just what the PRODUCT-readiness check reads: the session's quantization and pq_clusters (engine default 256).""" + """Just what the PRODUCT-readiness check reads: the index's quantization and pq_clusters.""" from types import SimpleNamespace - meta = self._db._vector_meta.get((self._type, self._prop), {}) + info = self.engine_info() return SimpleNamespace( - quantizationType=str(meta.get("quantization") or "NONE"), - pqClusters=int(meta.get("pqClusters", 256)), + quantizationType=str(info["quantization"] or "NONE"), + pqClusters=int(info["pq_clusters"]), ) def getIdPropertyName(self): @@ -120,14 +136,15 @@ def countEntries(self): return int(row.get("c")) def warmUp(self): - return None + # the engine loads a persisted graph on the first search: run that search now + self.buildVectorGraphNow() def buildVectorGraphNow(self): # the engine builds the graph on the first search: one search now, on a vector of the index's dimension - dims = self._db._vector_dimensions.get((self._type, self._prop)) + dims = self.engine_info()["dimensions"] if dims: self._db.query( - "sql", "SELECT vectorNeighbors(?, ?, 1)", self._name, [0.0] * dims + "sql", "SELECT vectorNeighbors(?, ?, 1)", self._name, [1.0] * dims ).close() @@ -151,7 +168,7 @@ def _run_search( self, *, query_vector, k, allowed_rids, approximate, ef_search=None ): if approximate: - not_yet("VectorIndex.find_nearest_approximate over PRODUCT quantization") + return self._run_approximate(query_vector, k, allowed_rids) from .records import _wrap_dict vector = ( @@ -195,11 +212,59 @@ def _run_search( out.append((record, float(row.get("distance")))) return self._sort_results(out)[:k] + def _run_approximate(self, query_vector, k, allowed_rids): + """The PQ-scored search of the engine (no statement runs it): the engine returns (RID, distance), the records are loaded here.""" + from . import lib + from .extras import _call, _enc + + vector = ( + query_vector.tolist() + if hasattr(query_vector, "tolist") + else list(query_vector) + ) + reply = _call( + "Approximate search failed", + "arcadedb_n_vindex", + self._database._h, + lib.b("approximate"), + lib.b( + _enc( + { + "type": self._shim._type, + "prop": self._shim._prop, + "vector": vector, + "k": int(k), + "allowed": ( + [_rid_text(r) for r in allowed_rids] + if allowed_rids + else None + ), + } + ) + ), + ) + out = [ + (self._database.lookup_by_rid(rid), float(distance)) + for rid, distance in reply["hits"] + ] + return self._sort_results(out)[:k] + def find_nearest_approximate(self, query_vector, k=10, allowed_rids=None): - """PQ approximate search needs PRODUCT quantization; the search itself has no statement form (not on native yet).""" + """PQ approximate search over PRODUCT quantization, the engine's own (``findNeighborsFromVectorApproximate``).""" if self.get_quantization() != "PRODUCT": raise ArcadeDBError("Approximate search requires quantization=PRODUCT (PQ)") - not_yet("VectorIndex.find_nearest_approximate over PRODUCT quantization") + self._ensure_product_quantization_ready() + try: + return self._run_search( + query_vector=query_vector, + k=k, + allowed_rids=allowed_rids, + approximate=True, + ) + except ArcadeDBError: + raise + except Exception as e: + raise ArcadeDBError(f"Approximate search failed: {e}") from e def _lookup_query_vector_by_key(self, key): type_name, prop = self._shim._type, self._shim._prop @@ -217,60 +282,72 @@ def _lookup_query_vector_by_key(self, key): return row.get("query_vector") def get_stats(self): - not_yet("VectorIndex.get_stats") + """The engine's live counters of the primary bucket index, from its ``schema:index:`` row (the row's own columns left out).""" + bucket = self._shim.engine_info()["bucket"] + rows = self._database.query( + "sql", + f"SELECT FROM schema:index:`{bucket}`", # nosec B608 - a backtick-quoted index name the engine reported + ).to_list() + if not rows: + raise ArcadeDBError( + f"Failed to read vector index stats: no index {bucket!r}" + ) + return {k: v for k, v in rows[0].items() if k not in _INDEX_ROW_COLUMNS} def get_metadata(self): - """The configuration this session created the index with (no engine-chosen values: the image has no accessor for them).""" - meta = self._database._vector_meta.get((self._shim._type, self._shim._prop)) - if meta is None: - not_yet("VectorIndex.get_metadata of an index this session did not create") - names = { - "dimensions": "dimensions", - "similarity": "similarity_function", - "idPropertyName": "id_property", - "quantization": "quantization", - "maxConnections": "max_connections", - "beamWidth": "beam_width", - "graphBuildCacheSize": "graph_build_cache_size", - "mutationsBeforeRebuild": "mutations_before_rebuild", - "storeVectorsInGraph": "store_vectors_in_graph", - "addHierarchy": "add_hierarchy", - "pqSubspaces": "pq_subspaces", - "pqClusters": "pq_clusters", - "pqCenterGlobally": "pq_center_globally", - "pqTrainingLimit": "pq_training_limit", - "rerankOversample": "rerank_oversample", - "fusedPQ": "fused_pq", - } - buckets = [ - r["name"] - for r in self._database.query( - "sql", "SELECT FROM schema:indexes WHERE indexType = 'LSM_VECTOR'" - ).to_list() - if r["typeName"] == self._shim._type - and r["properties"] == [[self._shim._prop]] - and "fileId" in r - ] - out = { + """The engine's configuration of the index: what this session created and what an index opened from disk carries, defaults included.""" + info = self._shim.engine_info() + return { "index_name": self._shim._name, - "bucket_index_name": buckets[0] if buckets else None, - "type_name": self._shim._type, + "bucket_index_name": info["bucket"], + "type_name": info["type_name"], "vector_property": self._shim._prop, + **{k: info[k] for k in _METADATA_KEYS}, } - out.update({names[k]: v for k, v in meta.items() if k in names}) - if "similarity_function" in out: - out["similarity_function"] = str(out["similarity_function"]).upper() - if "quantization" in out: - out["quantization"] = str(out["quantization"]).upper() - return out def get_quantization(self): - meta = self._database._vector_meta.get((self._shim._type, self._shim._prop)) - if meta is None: - not_yet( - "VectorIndex.get_quantization of an index this session did not create" - ) - return str(meta.get("quantization") or "NONE") + return str(self._shim.engine_info()["quantization"] or "NONE") + + +# the descriptive columns of a schema:index row; every other column is a counter of IndexInternal.getStats() +_INDEX_ROW_COLUMNS = frozenset( + { + "name", + "indexType", + "typeName", + "properties", + "keyTypes", + "unique", + "automatic", + "compacting", + "valid", + "supportsOrderedIterations", + "nullStrategy", + "associatedBucketId", + "fileId", + "size", + "upgradeWarning", + "typeIndexName", + } +) +_METADATA_KEYS = ( + "dimensions", + "similarity_function", + "id_property", + "quantization", + "max_connections", + "beam_width", + "ef_search", + "graph_build_cache_size", + "mutations_before_rebuild", + "store_vectors_in_graph", + "add_hierarchy", + "build_state", + "pq_subspaces", + "pq_clusters", + "pq_center_globally", + "pq_training_limit", +) def _check_vector_arguments( diff --git a/tests/test_async_transaction.py b/tests/test_async_transaction.py new file mode 100644 index 0000000..94c70cf --- /dev/null +++ b/tests/test_async_transaction.py @@ -0,0 +1,97 @@ +"""``AsyncExecutor.transaction``: a block that runs on an async writer inside a transaction, on the JVM wheel and the native wheel.""" + +import threading + +import pytest + +import arcadedb + +TIMEOUT = 30 + + +@pytest.fixture +def items_db(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Item") + temp_db.command("sql", "CREATE PROPERTY Item.n INTEGER") + return temp_db + + +def _count(db): + return int(db.query("sql", "SELECT count(*) AS c FROM Item").first().get("c")) + + +def _insert(db, n): + db.command("sql", "INSERT INTO Item SET n = ?", n) + + +def test_the_block_runs_in_a_transaction_and_commits(items_db): + async_exec = items_db.async_executor() + ran_on = [] + + def block(): + ran_on.append(threading.current_thread()) + _insert(items_db, 1) + _insert(items_db, 2) + + async_exec.transaction(block) + async_exec.wait_completion(TIMEOUT * 1000) + async_exec.close() + + assert _count(items_db) == 2 + # an async writer ran it, not the calling thread + assert len(ran_on) == 1 and ran_on[0] is not threading.current_thread() + + +def test_the_ok_callback_fires_after_the_commit(items_db): + async_exec = items_db.async_executor() + done = threading.Event() + seen_count = [] + + async_exec.transaction( + lambda: _insert(items_db, 7), + retries=3, + ok_callback=lambda: (seen_count.append(_count(items_db)), done.set()), + ) + assert done.wait(TIMEOUT) + async_exec.wait_completion(TIMEOUT * 1000) + async_exec.close() + + assert seen_count == [1] + assert _count(items_db) == 1 + + +def test_a_failing_block_rolls_back_and_reaches_the_error_callback(items_db): + async_exec = items_db.async_executor() + failed = threading.Event() + errors = [] + + def block(): + _insert(items_db, 1) + raise ValueError("boom") + + def on_error(exc): + errors.append(exc) + failed.set() + + async_exec.transaction(block, retries=1, error_callback=on_error) + assert failed.wait(TIMEOUT) + async_exec.wait_completion(TIMEOUT * 1000) + async_exec.close() + + assert len(errors) == 1 and isinstance(errors[0], Exception) + assert _count(items_db) == 0 + + +def test_blocks_on_one_slot_run_in_submission_order(items_db): + async_exec = items_db.async_executor() + order = [] + + for i in range(5): + async_exec.transaction( + lambda i=i: (order.append(i), _insert(items_db, i)), retries=1, slot=0 + ) + async_exec.wait_completion(TIMEOUT * 1000) + async_exec.close() + + assert order == [0, 1, 2, 3, 4] + assert _count(items_db) == 5 diff --git a/tests/test_vector_index_introspection.py b/tests/test_vector_index_introspection.py new file mode 100644 index 0000000..d6f58a9 --- /dev/null +++ b/tests/test_vector_index_introspection.py @@ -0,0 +1,227 @@ +"""Vector index introspection and PRODUCT-quantization search: the same behaviour on the JVM wheel and the native wheel. + +``get_stats`` (live counters), ``get_metadata`` and ``get_quantization`` of an index that this session did not create (a +reopened database), and ``find_nearest_approximate`` over PRODUCT quantization. The tests use no Java object, so they run +on both backends. +""" + +import pytest + +import arcadedb + +# keys the metadata dict has on both wheels (the engine's configuration of the index) +METADATA_KEYS = { + "index_name", + "bucket_index_name", + "type_name", + "vector_property", + "dimensions", + "similarity_function", + "id_property", + "quantization", + "max_connections", + "beam_width", + "ef_search", + "graph_build_cache_size", + "mutations_before_rebuild", + "store_vectors_in_graph", + "add_hierarchy", + "build_state", + "pq_subspaces", + "pq_clusters", + "pq_center_globally", + "pq_training_limit", +} + + +def _make_type(db): + db.command("sql", "CREATE VERTEX TYPE Doc") + db.command("sql", "CREATE PROPERTY Doc.name STRING") + db.command("sql", "CREATE PROPERTY Doc.embedding ARRAY_OF_FLOATS") + + +def _insert(db, rows): + with db.transaction(): + for name, vector in rows: + db.command( + "sql", + "INSERT INTO Doc SET name = ?, embedding = ?", + name, + arcadedb.to_java_float_array(vector), + ) + + +UNIT_VECTORS = [("x", [1.0, 0.0, 0.0]), ("y", [0.0, 1.0, 0.0]), ("z", [0.0, 0.0, 1.0])] +# as many vectors as PQ clusters: each vector is its own centroid, so the approximate answer is the exact one +PQ_VECTORS = UNIT_VECTORS + [("w", [-1.0, 0.0, 0.0])] + + +class TestGetStats: + def test_counters_start_at_zero_and_track_inserts(self, temp_db): + _make_type(temp_db) + index = temp_db.create_vector_index("Doc", "embedding", dimensions=3) + + stats = index.get_stats() + assert isinstance(stats, dict) and stats + assert stats["totalVectors"] == 0 + assert isinstance(stats["dimensions"], int) + assert stats["vectorFetchFromDocuments"] >= 0 + + _insert(temp_db, UNIT_VECTORS[:2]) + assert index.get_stats()["totalVectors"] == 2 + + def test_values_are_plain_scalars_and_the_listing_columns_are_not_counters( + self, temp_db + ): + _make_type(temp_db) + index = temp_db.create_vector_index("Doc", "embedding", dimensions=3) + _insert(temp_db, UNIT_VECTORS) + + stats = index.get_stats() + assert all( + v is None or isinstance(v, (int, float, bool, str)) for v in stats.values() + ) + # a schema:index row also names the index and describes it; get_stats is the counters only + for descriptive in ("name", "indexType", "typeName", "properties", "fileId"): + assert descriptive not in stats + + def test_searches_move_the_search_counter(self, temp_db): + _make_type(temp_db) + index = temp_db.create_vector_index("Doc", "embedding", dimensions=3) + _insert(temp_db, UNIT_VECTORS) + + before = index.get_stats()["searchOperations"] + index.find_nearest([1.0, 0.0, 0.0], k=1) + assert index.get_stats()["searchOperations"] > before + + +class TestGetMetadataOfAReopenedIndex: + @staticmethod + def _create(path, **kwargs): + with arcadedb.create_database(path) as db: + _make_type(db) + index = db.create_vector_index( + "Doc", "embedding", dimensions=4, id_property="name", **kwargs + ) + _insert( + db, + [ + ("a", [1.0, 0.0, 0.0, 0.0]), + ("b", [0.0, 1.0, 0.0, 0.0]), + ("c", [0.0, 0.0, 1.0, 0.0]), + ], + ) + return index.get_metadata() + + def test_metadata_after_reopen_equals_the_creating_session(self, temp_db_path): + created = self._create( + temp_db_path, + distance_function="EUCLIDEAN", + max_connections=16, + beam_width=64, + add_hierarchy=False, + ) + assert set(created) == METADATA_KEYS + + with arcadedb.open_database(temp_db_path) as db: + index = db.schema.get_vector_index("Doc", "embedding") + reopened = index.get_metadata() + + assert set(reopened) == METADATA_KEYS + assert reopened["dimensions"] == 4 + assert reopened["similarity_function"] == "EUCLIDEAN" + assert reopened["id_property"] == "name" + assert reopened["max_connections"] == 16 + assert reopened["beam_width"] == 64 + assert reopened["add_hierarchy"] is False + # the build state is runtime state; everything else is the index's configuration + created.pop("build_state") + reopened.pop("build_state") + assert reopened == created + + def test_product_quantization_settings_survive_a_reopen(self, temp_db_path): + self._create(temp_db_path, quantization="PRODUCT", pq_clusters=2) + + with arcadedb.open_database(temp_db_path) as db: + index = db.schema.get_vector_index("Doc", "embedding") + assert index.get_quantization() == "PRODUCT" + meta = index.get_metadata() + assert meta["quantization"] == "PRODUCT" + assert meta["pq_clusters"] == 2 + + @pytest.mark.parametrize("quantization", ["NONE", "INT8", "BINARY"]) + def test_quantization_of_a_reopened_index(self, temp_db_path, quantization): + self._create(temp_db_path, quantization=quantization) + with arcadedb.open_database(temp_db_path) as db: + index = db.schema.get_vector_index("Doc", "embedding") + assert index.get_quantization() == quantization + assert index.get_metadata()["quantization"] == quantization + + def test_the_default_quantization_is_the_same_after_a_reopen(self, temp_db_path): + created = self._create(temp_db_path) + with arcadedb.open_database(temp_db_path) as db: + index = db.schema.get_vector_index("Doc", "embedding") + assert index.get_quantization() == created["quantization"] + + +class TestFindNearestApproximate: + @staticmethod + def _pq_index(db, rows): + _make_type(db) + index = db.create_vector_index( + "Doc", "embedding", dimensions=3, quantization="PRODUCT", pq_clusters=4 + ) + _insert(db, rows) + index.build_graph_now() + return index + + def test_returns_the_nearest_record_and_its_distance(self, temp_db): + index = self._pq_index(temp_db, PQ_VECTORS) + + results = index.find_nearest_approximate([0.9, 0.1, 0.0], k=1) + + assert len(results) == 1 + record, distance = results[0] + assert record.get("name") == "x" + assert isinstance(distance, float) + + def test_k_results_come_back_nearest_first(self, temp_db): + index = self._pq_index(temp_db, PQ_VECTORS) + + results = index.find_nearest_approximate([0.9, 0.1, 0.0], k=3) + + assert [r.get("name") for r, _ in results][0] == "x" + assert len(results) == 3 + distances = [d for _, d in results] + assert distances == sorted(distances) + + def test_allowed_rids_restrict_the_answer(self, temp_db): + index = self._pq_index(temp_db, PQ_VECTORS) + rids = { + row.get("name"): row.get("rid") + for row in temp_db.query("sql", "SELECT name, @rid AS rid FROM Doc") + } + + results = index.find_nearest_approximate( + [0.9, 0.1, 0.0], k=3, allowed_rids=[str(rids["y"]), str(rids["z"])] + ) + + names = {r.get("name") for r, _ in results} + assert names <= {"y", "z"} and names + + def test_a_typeindex_wrapper_from_the_schema_searches_too(self, temp_db): + self._pq_index(temp_db, PQ_VECTORS) + index = temp_db.schema.get_vector_index("Doc", "embedding") + + results = index.find_nearest_approximate([0.0, 0.9, 0.1], k=1) + + assert [r.get("name") for r, _ in results] == ["y"] + + def test_without_product_quantization_it_raises(self, temp_db): + _make_type(temp_db) + index = temp_db.create_vector_index("Doc", "embedding", dimensions=3) + _insert(temp_db, UNIT_VECTORS) + index.build_graph_now() + + with pytest.raises(arcadedb.ArcadeDBError): + index.find_nearest_approximate([0.9, 0.1, 0.0], k=1) From 0af1225e6a4e8cd966366967b2fca00a2f56a41c Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sun, 25 Oct 2026 21:32:16 +0900 Subject: [PATCH 2/3] Native: AsyncExecutor.transaction releases its callbacks once the executor has finished its work AsyncExecutor.transaction registers up to three callbacks per call (block, ok, error). They are kept in the executor's _cb_ids (declared, never used so far) and forgotten when wait_completion has returned or the executor is closed, because a retry calls the block again. The other async operations still register callbacks that are never released; that is unchanged. The reply of a failing block needs nothing new here: with #67 it is written into the out slot as library memory, which outlives the Python thread state of an async writer, so the Java side reads {"error": ...} and the writer rolls back and reaches the error callback. Co-Authored-By: Claude Sonnet 5.5 --- src/arcadedb/_native/extras.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/src/arcadedb/_native/extras.py b/src/arcadedb/_native/extras.py index b1042e6..e27a1c7 100644 --- a/src/arcadedb/_native/extras.py +++ b/src/arcadedb/_native/extras.py @@ -713,10 +713,18 @@ def run(message): args["ok"] = _register(lambda message: ok_callback()) if error_callback is not None: args["err"] = self._error_callback(error_callback) + # a retry calls the block again, so the callbacks live until the executor has finished its work (wait_completion, close) + self._cb_ids.extend(v for k, v in args.items() if k in ("cb", "ok", "err")) if slot is not None: args["slot"] = int(slot) self._op("transaction", **args) + def _release_callbacks(self): + """Forget the callbacks of the finished ``transaction`` calls.""" + ids, self._cb_ids = self._cb_ids, [] + for cid in ids: + _forget(cid) + def wait_completion(self, timeout_ms: Optional[int] = None): if timeout_ms is None: self._op("wait") @@ -729,6 +737,7 @@ def wait_completion(self, timeout_ms: Optional[int] = None): raise TimeoutError( f"Async operations did not complete within {timeout_ms}ms" ) + self._release_callbacks() # nothing is pending any more def on_ok(self, callback: Callable[[], None]): self._op("on_ok", cb=_register(lambda message: callback())) @@ -745,6 +754,7 @@ def close(self): self._op("close") finally: self._closed = True + self._release_callbacks() owner, self._owner = self._owner, None if owner is not None: owner._discard_async_executor(self) From ae9cf586d98031f8bedc7ecedad6f1d51975fadb Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Sun, 25 Oct 2026 21:32:16 +0900 Subject: [PATCH 3/3] Native get_metadata: rerank_oversample and fused_pq follow the engine, as on the JVM build The base added these two keys to the JVM get_metadata when the engine has them (hasattr). The image links engine 26.10.2, which has not, so the native dict carries them only when the glue reports them; the shared test accepts them as optional. The glue reports them with the move to 26.11.1. Co-Authored-By: Claude Sonnet 5.5 --- src/arcadedb/_native/schema_objects.py | 3 +++ tests/test_vector_index_introspection.py | 8 ++++++-- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/src/arcadedb/_native/schema_objects.py b/src/arcadedb/_native/schema_objects.py index 3589ad1..3bd9046 100644 --- a/src/arcadedb/_native/schema_objects.py +++ b/src/arcadedb/_native/schema_objects.py @@ -303,6 +303,8 @@ def get_metadata(self): "type_name": info["type_name"], "vector_property": self._shim._prop, **{k: info[k] for k in _METADATA_KEYS}, + # present when the engine the image links has them, as on the JVM build (26.10.2 has not; the glue emits them from 26.11.1 on) + **{k: info[k] for k in _OPTIONAL_METADATA_KEYS if k in info}, } def get_quantization(self): @@ -330,6 +332,7 @@ def get_quantization(self): "typeIndexName", } ) +_OPTIONAL_METADATA_KEYS = ("rerank_oversample", "fused_pq") _METADATA_KEYS = ( "dimensions", "similarity_function", diff --git a/tests/test_vector_index_introspection.py b/tests/test_vector_index_introspection.py index d6f58a9..89a254d 100644 --- a/tests/test_vector_index_introspection.py +++ b/tests/test_vector_index_introspection.py @@ -56,6 +56,10 @@ def _insert(db, rows): PQ_VECTORS = UNIT_VECTORS + [("w", [-1.0, 0.0, 0.0])] +# the engine reports these two only from 26.11.1 on, on both builds +OPTIONAL_METADATA_KEYS = {"rerank_oversample", "fused_pq"} + + class TestGetStats: def test_counters_start_at_zero_and_track_inserts(self, temp_db): _make_type(temp_db) @@ -121,13 +125,13 @@ def test_metadata_after_reopen_equals_the_creating_session(self, temp_db_path): beam_width=64, add_hierarchy=False, ) - assert set(created) == METADATA_KEYS + assert set(created) - OPTIONAL_METADATA_KEYS == METADATA_KEYS with arcadedb.open_database(temp_db_path) as db: index = db.schema.get_vector_index("Doc", "embedding") reopened = index.get_metadata() - assert set(reopened) == METADATA_KEYS + assert set(reopened) - OPTIONAL_METADATA_KEYS == METADATA_KEYS assert reopened["dimensions"] == 4 assert reopened["similarity_function"] == "EUCLIDEAN" assert reopened["id_property"] == "name"