diff --git a/native/src/main/java/org/humemai/nativelib/ArcadeNative.java b/native/src/main/java/org/humemai/nativelib/ArcadeNative.java index 9b5acb91..7a6e31e1 100644 --- a/native/src/main/java/org/humemai/nativelib/ArcadeNative.java +++ b/native/src/main/java/org/humemai/nativelib/ArcadeNative.java @@ -337,6 +337,9 @@ private static String batchOf(final long id, final ResultSet rs, final int max) sb.append(id).append(','); final boolean drained = NativeCodec.appendBatch(sb, rs, max); if (drained) { + // before the close: the facets of a FACETS query are computed when its rows end, and the reply is the only place Python can read + // them from once the cursor is gone + appendFacets(sb, rs); RESULT_SETS.remove(id); closeQuietly(rs); sb.replace(mark, mark + String.valueOf(id).length(), "0"); @@ -344,6 +347,47 @@ private static String batchOf(final long id, final ResultSet rs, final int max) return sb.append('}').toString(); } + /** + * Appends ,"facets":{..} (see FacetsReply) for a drained FACETS query, nothing for any other statement. A failure to compute them is + * the member "ferr" (the usual error object), raised by Python when the facets are read, not when the rows are. + */ + private static void appendFacets(final StringBuilder sb, final ResultSet rs) { + final int mark = sb.length(); + try { + sb.append(",\"facets\":"); + if (!FacetsReply.append(sb, rs)) + sb.setLength(mark); + } catch (final Throwable t) { + sb.setLength(mark); + sb.append(",\"ferr\":").append(errorJson(t)); + } + } + + /** + * The facets of an open cursor whose rows were not all read: {"ok":true,"facets":{..}} and the cursor is closed (the engine reads the + * matches the rows did not reach and ends the rows), or {"ok":true,"facets":null} and the cursor stays open when the statement has + * no FACETS clause. + */ + @CEntryPoint(name = "arcadedb_n_facets") + public static CCharPointer facets(final IsolateThread thread, final long rsId) { + ResultSet rs = null; + try { + rs = RESULT_SETS.get(rsId); + if (rs == null) + throw new IllegalArgumentException("unknown result set " + rsId); + final StringBuilder sb = new StringBuilder("{\"ok\":true,\"facets\":"); + if (!FacetsReply.append(sb, rs)) + return out("{\"ok\":true,\"facets\":null}"); + RESULT_SETS.remove(rsId); + closeQuietly(rs); + return out(sb.append('}').toString()); + } catch (final Throwable t) { + RESULT_SETS.remove(rsId); + closeQuietly(rs); + return err(t); + } + } + private static void closeQuietly(final ResultSet rs) { if (rs != null) try { diff --git a/native/src/main/java/org/humemai/nativelib/FacetsReply.java b/native/src/main/java/org/humemai/nativelib/FacetsReply.java new file mode 100644 index 00000000..4127a9ba --- /dev/null +++ b/native/src/main/java/org/humemai/nativelib/FacetsReply.java @@ -0,0 +1,203 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The facets of a result set (ResultSet.getFacets(), columnar faceting, engine 26.12.1) as the JSON object that + * ResultSet.facets returns in Python: + * + * {"meta":{"exact":b,"patched":n,"lag_ms":n},"":{"type":..,"source":..,"reason":..,"buckets":[..],"other_count":n,"missing":n}} + * + * Every value of the data (a term, a bound, the min and max of a STATS facet, a cursor) is written as NativeCodec writes a row's value, + * so NaN and the infinities, dates and decimals reach Python as they do in a row (the engine's own JSON writers turn NaN into null). + * + * The engine classes are read by reflection, never named at compile time: the repository also builds against engines before 26.12.1 + * (the pull request gate links 26.11.1), which have neither ResultSet.getFacets() nor com.arcadedb.query.facet. On such an engine + * GET_FACETS is null and no statement has facets. The classes read here are listed in reachability-metadata.json. + */ +package org.humemai.nativelib; + +import com.arcadedb.query.sql.executor.ResultSet; + +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +final class FacetsReply { + private static final String UNAVAILABLE = "unavailable"; + + /** ResultSet.getFacets(), null on an engine that has none. */ + private static final Method GET_FACETS = find(ResultSet.class, "getFacets"); + /** FacetResults.get(String), null on an engine that has none. */ + private static final Method GET_FACET = find("com.arcadedb.query.facet.FacetResults", "get", String.class); + + /** The public no-argument methods read so far, by class and name. */ + private static final ClassValue> METHODS = new ClassValue<>() { + @Override + protected Map computeValue(final Class type) { + return new ConcurrentHashMap<>(); + } + }; + + private FacetsReply() { + } + + private static Method find(final String type, final String name, final Class... parameters) { + try { + return find(Class.forName(type), name, parameters); + } catch (final ReflectiveOperationException | LinkageError e) { + return null; + } + } + + private static Method find(final Class type, final String name, final Class... parameters) { + try { + return type.getMethod(name, parameters); + } catch (final ReflectiveOperationException | LinkageError e) { + return null; + } + } + + /** The cursor whose getFacets() answers: a Prefetched (the lazy cursors) holds the engine's result set as its tail. */ + private static ResultSet engineSet(final ResultSet rs) { + return rs instanceof Prefetched prefetched ? prefetched.tail() : rs; + } + + /** + * Appends the facets of rs and returns true; returns false, appending nothing, for a statement that has none. Computing them reads + * the matches the rows did not reach and ends the rows (read the rows first, as the engine documents). + */ + static boolean append(final StringBuilder sb, final ResultSet rs) throws Throwable { + if (GET_FACETS == null || GET_FACET == null) + return false; + final Object facets; + try { + facets = GET_FACETS.invoke(engineSet(rs)); + } catch (final InvocationTargetException e) { + throw e.getCause(); + } + if (facets == null) + return false; + appendFacets(sb, facets); + return true; + } + + /** The public no-argument method {@code name} of the object, called. A failure of the method itself is rethrown as it is. */ + private static Object call(final Object target, final String name) throws Throwable { + final Map methods = METHODS.get(target.getClass()); + Method method = methods.get(name); + if (method == null) { + method = target.getClass().getMethod(name); + methods.put(name, method); + } + try { + return method.invoke(target); + } catch (final InvocationTargetException e) { + throw e.getCause(); + } + } + + private static void appendFacets(final StringBuilder sb, final Object facets) throws Throwable { + sb.append("{\"meta\":{\"exact\":").append(call(facets, "exact")).append(",\"patched\":").append(call(facets, "patched")) + .append(",\"lag_ms\":").append(call(facets, "lagMs")).append('}'); + for (final Object alias : (Set) call(facets, "aliases")) { + sb.append(','); + NativeCodec.quote(sb, (String) alias); + sb.append(':'); + final Object result; + try { + result = GET_FACET.invoke(facets, alias); + } catch (final InvocationTargetException e) { + throw e.getCause(); + } + appendFacet(sb, result); + } + sb.append('}'); + } + + private static void appendFacet(final StringBuilder sb, final Object result) throws Throwable { + final String kind = (String) call(result, "kind"); + final String source = (String) call(result, "source"); + final String reason = (String) call(result, "reason"); + sb.append("{\"type\":"); + NativeCodec.quote(sb, kind); + sb.append(",\"source\":"); + NativeCodec.quote(sb, source); + if (reason != null) { + sb.append(",\"reason\":"); + NativeCodec.quote(sb, reason); + } + // an unavailable facet has no buckets, no counts and no cursor, only its reason + if (UNAVAILABLE.equals(source)) { + sb.append('}'); + return; + } + final Object stats = call(result, "stats"); + if (stats != null) { + sb.append(",\"count\":").append(call(stats, "count")).append(",\"missing\":").append(call(stats, "missing")); + appendMember(sb, "min", call(stats, "min")); + appendMember(sb, "max", call(stats, "max")); + appendMember(sb, "sum", call(stats, "sum")); + appendMember(sb, "avg", call(stats, "avg")); + } else { + final Object distinct = call(result, "distinct"); + if (distinct != null) + sb.append(",\"value\":").append(call(distinct, "value")).append(",\"exact\":").append(call(distinct, "exact")); + else { + sb.append(",\"buckets\":["); + boolean first = true; + for (final Object bucket : (List) call(result, "buckets")) { + if (!first) + sb.append(','); + first = false; + appendBucket(sb, bucket); + } + sb.append(']'); + // the counts of the values not returned: a terms or hierarchy facet has them, a range or histogram has every interval + if (!"range".equals(kind) && !"histogram".equals(kind)) + sb.append(",\"other_count\":").append(call(result, "otherCount")); + } + sb.append(",\"missing\":").append(call(result, "missing")); + } + if ((Boolean) call(result, "hasAfter")) + appendMember(sb, "after", call(result, "after")); + sb.append('}'); + } + + private static void appendBucket(final StringBuilder sb, final Object bucket) throws Throwable { + final String key = (String) call(bucket, "key"); + sb.append('{'); + if (key != null) { + sb.append("\"key\":"); + NativeCodec.quote(sb, key); + if ((Boolean) call(bucket, "isDefault")) + sb.append(",\"default\":true"); + else { + final Object from = call(bucket, "from"); + if (from != null) + appendMember(sb, "from", from); + final Object to = call(bucket, "to"); + if (to != null) + appendMember(sb, "to", to); + } + } else if ((Boolean) call(bucket, "isMissing")) + // the records without a value + sb.append("\"value\":null,\"missing\":true"); + else { + sb.append("\"value\":"); + NativeCodec.append(sb, call(bucket, "value")); + } + sb.append(",\"count\":").append(call(bucket, "count")); + // the extra members of a hierarchy bucket (parent, depth, orphan): the text of a JSON object, written without its braces + final String attributes = String.valueOf(call(bucket, "attributes")); + if (attributes.length() > 2 && attributes.charAt(0) == '{') + sb.append(',').append(attributes, 1, attributes.length() - 1); + sb.append('}'); + } + + private static void appendMember(final StringBuilder sb, final String name, final Object value) { + sb.append(",\"").append(name).append("\":"); + NativeCodec.append(sb, value); + } +} diff --git a/native/src/main/java/org/humemai/nativelib/Prefetched.java b/native/src/main/java/org/humemai/nativelib/Prefetched.java index 6929c023..e90fd048 100644 --- a/native/src/main/java/org/humemai/nativelib/Prefetched.java +++ b/native/src/main/java/org/humemai/nativelib/Prefetched.java @@ -54,4 +54,12 @@ public Result next() { public void close() { tail.close(); } + + /** + * The engine's result set behind the rows pulled so far: its getFacets() answers for this cursor (FacetsReply reads it there). It is + * not overridden here because the interface method does not exist on engines before 26.12.1, which this class must compile against. + */ + ResultSet tail() { + return tail; + } } diff --git a/native/src/main/resources/META-INF/native-image/org.humemai/arcadedb-native-lib/reachability-metadata.json b/native/src/main/resources/META-INF/native-image/org.humemai/arcadedb-native-lib/reachability-metadata.json index 8d9e5e08..0f8d99d8 100644 --- a/native/src/main/resources/META-INF/native-image/org.humemai/arcadedb-native-lib/reachability-metadata.json +++ b/native/src/main/resources/META-INF/native-image/org.humemai/arcadedb-native-lib/reachability-metadata.json @@ -642,6 +642,26 @@ { "type": "com.arcadedb.postgres.PostgresProtocolPlugin" }, + { + "type": "com.arcadedb.query.facet.DistinctValue", + "allPublicMethods": true + }, + { + "type": "com.arcadedb.query.facet.FacetBucket", + "allPublicMethods": true + }, + { + "type": "com.arcadedb.query.facet.FacetResult", + "allPublicMethods": true + }, + { + "type": "com.arcadedb.query.facet.FacetResults", + "allPublicMethods": true + }, + { + "type": "com.arcadedb.query.facet.FacetStats", + "allPublicMethods": true + }, { "type": "com.arcadedb.query.opencypher.query.OpenCypherQueryEngineFactory", "methods": [ @@ -4886,6 +4906,10 @@ "allPublicMethods": true, "allDeclaredFields": true }, + { + "type": "com.arcadedb.query.sql.executor.ResultSet", + "allPublicMethods": true + }, { "type": "com.arcadedb.query.sql.parser.Alias", "allDeclaredFields": true @@ -5238,6 +5262,18 @@ "type": "com.arcadedb.query.sql.parser.ExpressionStatement", "allDeclaredFields": true }, + { + "type": "com.arcadedb.query.sql.parser.FacetItem", + "allDeclaredFields": true + }, + { + "type": "com.arcadedb.query.sql.parser.FacetRangeItem", + "allDeclaredFields": true + }, + { + "type": "com.arcadedb.query.sql.parser.FacetsClause", + "allDeclaredFields": true + }, { "type": "com.arcadedb.query.sql.parser.FieldMatchPathItem", "allDeclaredFields": true diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index bd0107e9..8dce7ab4 100644 --- a/src/arcadedb/_native/database.py +++ b/src/arcadedb/_native/database.py @@ -215,6 +215,10 @@ def __init__(self, database, reply: dict, tagged: bool = True): # the engine computes lazily: an error met while the first rows were read is raised when the rows before it # have been consumed, as it is on the JPype path (the cursor is already finished on the Java side) self._pending_error = reply.get("rerr") + # the facets of a FACETS query ride on the reply that drains the cursor (the engine computes them when its rows end, and the + # cursor is closed with that reply); ferr is the error of computing them, raised when they are read + self._facets = reply.get("facets") + self._facets_error = reply.get("ferr") @classmethod def _lazy(cls, database, rs_id: int) -> "NativeResultSet": @@ -228,6 +232,8 @@ def _lazy(cls, database, rs_id: int) -> "NativeResultSet": self._closed = False self._exhausted = False self._pending_error = None + self._facets = None + self._facets_error = None return self # -- state @@ -282,8 +288,66 @@ def _fill(self) -> bool: self._buf.extend(zip(reply["rows"], reply["meta"])) self._more = not reply["drained"] self._pending_error = reply.get("rerr") + if "facets" in reply: + self._facets = reply["facets"] + if "ferr" in reply: + self._facets_error = reply["ferr"] return bool(self._buf) or self._pending_error is not None + @property + def facets(self) -> Optional[Dict[str, Any]]: + """The facets of a ``SELECT ... FACETS`` query, or None for a statement without a ``FACETS`` clause. + + A plain dict. ``meta`` holds ``exact`` (False when the counts may lag the committed data), ``patched`` (records read from + the record itself to correct the projection) and ``lag_ms``. Every other key is a facet, by its alias (the property name + when none is written), in the order of the clause. A facet has ``type`` (``terms``, ``range``, ``histogram``, + ``hierarchy``, ``distinct`` or ``stats``) and ``source`` (``index``, ``scan`` or ``unavailable``). An unavailable facet + (for example one that needed a fallback scan the database refused) has a ``reason`` and nothing else, it is not dropped. + Terms, hierarchy, range and histogram facets have ``buckets``, a list of dicts with ``count`` and ``value`` (terms), or + ``key`` with ``from`` and ``to`` (an unbounded end is left out), or ``key`` with ``default`` (the values in no range); the + bucket of the records without a value has ``value`` None and ``missing`` True. Terms and hierarchy facets add + ``other_count`` (the records of the values not returned); every facet but a stats one has ``missing`` (the records + without a value). A stats facet has ``count``, ``missing``, ``min``, ``max``, ``sum`` and ``avg``, a distinct facet + ``value`` and ``exact``, and a facet with an ``AFTER`` cursor has ``after``. Values are the Python values a row holds, so + NaN and the infinities are floats. + + The facets cover every record the ``WHERE`` clause matched, whatever ``SKIP`` and ``LIMIT`` kept. The engine computes them + when the rows are exhausted, so read the rows first if you need both. Reading ``facets`` before the rows works and gives the + same facets, but it makes the engine compute them over the matches the rows did not reach, and whether rows that were not + fetched yet can still be read afterwards is unspecified. The facets stay available after the rows are read or the result + set is exhausted. + + Raises ArcadeDBError for a result set closed before its end, and for one whose database was closed before its end. + """ + if self._facets_error is not None: + raise _read_error(self._facets_error) + if self._facets is not None: + return self._facets + if not self._readable(): + return None # read to the end: the last reply carried the facets, and a statement without them has none + if not (self._more and self._rs): + return None # every row is fetched and no cursor is left to ask + reply, _ = loads_tagged(lib.call("arcadedb_n_facets", self._rs)) + if not reply.get("ok"): + # the cursor is gone on the Java side + self._more = False + self._rs = 0 + self._closed = True + self._exhausted = True + self._buf.clear() + raise _read_error(reply) + facets = reply.get("facets") + if facets is None: + return ( + None # no FACETS clause: the cursor stays open, the rows are untouched + ) + # the facets are computed and the engine ended the rows: the cursor was closed with the call + self._facets = facets + self._more = False + self._rs = 0 + self._finish() + return facets + def _finish(self) -> None: self._exhausted = True self.close() diff --git a/src/arcadedb/_native/lib.py b/src/arcadedb/_native/lib.py index 58b261ba..3e49d465 100644 --- a/src/arcadedb/_native/lib.py +++ b/src/arcadedb/_native/lib.py @@ -233,6 +233,7 @@ class _Params(ctypes.Structure): "arcadedb_n_next": [P, L, I], "arcadedb_n_next_bin": [P, L, I], "arcadedb_n_rs_close": [P, L], + "arcadedb_n_facets": [P, L], "arcadedb_n_cmd": [P, L, I, S, S, S, S, L, I], "arcadedb_n_last_error": [P], "arcadedb_n_rs_close_many": [P, P, I], diff --git a/src/arcadedb/results.py b/src/arcadedb/results.py index ebb344d0..25c5b0d8 100644 --- a/src/arcadedb/results.py +++ b/src/arcadedb/results.py @@ -137,6 +137,90 @@ def _bridge_class(name): return cls +def _facet_values(java_object, names): + """``{name: value}`` of the named accessors of a facet object, each value as a row's value (None stays None).""" + return { + name: convert_java_to_python(getattr(java_object, name)()) for name in names + } + + +def _facet_bucket(java_bucket) -> Dict[str, Any]: + """One bucket of a facet as the dict ``ResultSet.facets`` documents.""" + key = java_bucket.key() + bucket: Dict[str, Any] = {} + if key is not None: + bucket["key"] = str(key) + if bool(java_bucket.isDefault()): + bucket["default"] = True + else: + # JPype renames a Java method called like a Python keyword: from() is from_() + java_from = getattr(java_bucket, "from_", None) or getattr( + java_bucket, "from" + ) + for name, java_value in (("from", java_from()), ("to", java_bucket.to())): + if java_value is not None: + bucket[name] = convert_java_to_python(java_value) + elif bool(java_bucket.isMissing()): + bucket["value"] = None + bucket["missing"] = True + else: + bucket["value"] = convert_java_to_python(java_bucket.value()) + bucket["count"] = int(java_bucket.count()) + # the extra members of a hierarchy bucket (parent, depth, orphan): the text of a JSON object + extra = json.loads(str(java_bucket.attributes().toString())) + for name, value in extra.items(): + bucket.setdefault(name, value) + return bucket + + +def _facet_result(java_result) -> Dict[str, Any]: + """One facet (``FacetResult``) as the dict ``ResultSet.facets`` documents.""" + kind = str(java_result.kind()) + source = str(java_result.source()) + out: Dict[str, Any] = {"type": kind, "source": source} + reason = java_result.reason() + if reason is not None: + out["reason"] = str(reason) + if source == "unavailable": + return out # no buckets, no counts and no cursor, only the reason + stats = java_result.stats() + if stats is not None: + out["count"] = int(stats.count()) + out["missing"] = int(stats.missing()) + out.update(_facet_values(stats, ("min", "max", "sum", "avg"))) + else: + distinct = java_result.distinct() + if distinct is not None: + out["value"] = int(distinct.value()) + out["exact"] = bool(distinct.exact()) + else: + out["buckets"] = [_facet_bucket(b) for b in java_result.buckets()] + if kind not in ("range", "histogram"): + out["other_count"] = int(java_result.otherCount()) + out["missing"] = int(java_result.missing()) + if bool(java_result.hasAfter()): + out["after"] = convert_java_to_python(java_result.after()) + return out + + +def _facets_to_python(java_facets) -> Dict[str, Any]: + """The engine's ``FacetResults`` as the dict ``ResultSet.facets`` documents. + + Read through the engine's own accessors, not through the bridge jar: the bridge is compiled against the engine jars of + each build, and an engine before 26.12.1 has no ``FacetResults`` class to compile against. + """ + out: Dict[str, Any] = { + "meta": { + "exact": bool(java_facets.exact()), + "patched": int(java_facets.patched()), + "lag_ms": int(java_facets.lagMs()), + } + } + for alias in java_facets.aliases(): + out[str(alias)] = _facet_result(java_facets.get(alias)) + return out + + def _read_error(exc: BaseException) -> ArcadeDBError: """The ArcadeDBError for a Java exception raised while rows were read. @@ -345,6 +429,8 @@ def __init__(self, java_result_set, database=None): self._database = database # strong reference, see the class docstring self._closed = False self._exhausted = False + self._facets = None + self._facets_read = False def __iter__(self) -> Iterator["Result"]: return self @@ -374,6 +460,52 @@ def _readable(self) -> bool: "it had not returned are gone. Run the query again to read them." ) + @property + def facets(self) -> Optional[Dict[str, Any]]: + """The facets of a ``SELECT ... FACETS`` query, or None for a statement without a ``FACETS`` clause. + + A plain dict. ``meta`` holds ``exact`` (False when the counts may lag the committed data), ``patched`` (records read from + the record itself to correct the projection) and ``lag_ms``. Every other key is a facet, by its alias (the property name + when none is written), in the order of the clause. A facet has ``type`` (``terms``, ``range``, ``histogram``, + ``hierarchy``, ``distinct`` or ``stats``) and ``source`` (``index``, ``scan`` or ``unavailable``). An unavailable facet + (for example one that needed a fallback scan the database refused) has a ``reason`` and nothing else, it is not dropped. + Terms, hierarchy, range and histogram facets have ``buckets``, a list of dicts with ``count`` and ``value`` (terms), or + ``key`` with ``from`` and ``to`` (an unbounded end is left out), or ``key`` with ``default`` (the values in no range); the + bucket of the records without a value has ``value`` None and ``missing`` True. Terms and hierarchy facets add + ``other_count`` (the records of the values not returned); every facet but a stats one has ``missing`` (the records + without a value). A stats facet has ``count``, ``missing``, ``min``, ``max``, ``sum`` and ``avg``, a distinct facet + ``value`` and ``exact``, and a facet with an ``AFTER`` cursor has ``after``. Values are the Python values a row holds, so + NaN and the infinities are floats. + + The facets cover every record the ``WHERE`` clause matched, whatever ``SKIP`` and ``LIMIT`` kept. The engine computes them + when the rows are exhausted, so read the rows first if you need both. Reading ``facets`` before the rows works and gives the + same facets, but it makes the engine compute them over the matches the rows did not reach, and whether rows that were not + fetched yet can still be read afterwards is unspecified. The facets stay available after the rows are read or the result + set is exhausted. + + Raises ArcadeDBError for a result set closed before its end, and for one whose database was closed before its end. + """ + if self._facets_read: + return self._facets + readable = self._readable() + try: + java_facets = self._java_result_set.getFacets() + except AttributeError: + return None # an engine before 26.12.1 has no facets + except JException as exc: + raise _read_error(exc) from exc + if java_facets is None: + return None # no FACETS clause: the rows are untouched + try: + self._facets = _facets_to_python(java_facets) + except JException as exc: + raise _read_error(exc) from exc + self._facets_read = True + if readable: + # the engine computed the facets over every match and may have ended the rows: they are not read again + self._finish() + return self._facets + def _finish(self) -> None: self._exhausted = True self.close() diff --git a/tests/test_result_set_facets.py b/tests/test_result_set_facets.py new file mode 100644 index 00000000..9aeb558f --- /dev/null +++ b/tests/test_result_set_facets.py @@ -0,0 +1,514 @@ +"""The facets of a result set: ``ResultSet.facets`` of a ``SELECT ... FACETS`` query (ArcadeDB 26.12.1, issue #290). + +The engine computes the facets next to the rows (``ResultSet.getFacets()``); both packages expose them as a plain dict: +``meta`` (``exact``, ``patched``, ``lag_ms``) and one entry per facet alias. This file is the proof for both: it runs on +``arcadedb-jvm`` and on the native ``arcadedb``, with only the public Python API. + +The engine of a lane is whatever the lane links (see tests/test_columnar_engine_smoke.py): an engine older than 26.12.1 must +refuse the ``FACETS`` clause with its own syntax error and a plain query has no facets; a newer one must return them. The version +is read through the test's database (``SELECT version()``), so the same check runs on both builds and nothing is skipped. +""" + +import math +from datetime import datetime + +import pytest + +from arcadedb import ArcadeDBError, create_database + +FACETS_SINCE = (26, 12, 1) + +CATEGORIES = ["a", "b", "a", "c", "a", "b", "a", "d", "a", "c"] +# a: 5, b: 2, c: 2, d: 1 +CATEGORY_COUNTS = {"a": 5, "b": 2, "c": 2, "d": 1} +# ids 0-3 in January 2024, 4-6 in February, 7-9 in April: three calendar months with rows, March empty +MONTH_OF = [1, 1, 1, 1, 2, 2, 2, 4, 4, 4] + +TERMS_Q = "SELECT id, category FROM FacetDoc ORDER BY id FACETS category TERMS TOP 2" +PAGED_Q = ( + "SELECT id FROM FacetDoc ORDER BY id FACETS category TERMS TOP 10 SKIP 2 LIMIT 3" +) +RANGES_Q = ( + "SELECT id FROM FacetDoc ORDER BY id " + "FACETS s: score RANGES ([0, 50) AS 'low', [50, *) AS 'high')" +) +RANGES_DEFAULT_Q = ( + "SELECT id FROM FacetDoc ORDER BY id " + "FACETS s: score RANGES ([0, 30) AS 'low') DEFAULT 'rest'" +) +HISTOGRAM_Q = ( + "SELECT id FROM FacetDoc ORDER BY id " + "FACETS h: ts HISTOGRAM CALENDAR '1M' TIMEZONE 'UTC'" +) +FILTER_Q = ( + "SELECT id, category FROM FacetDoc ORDER BY id " + "FACETS c: category TERMS TOP 10 FILTER category = 'a', " + "s: score RANGES ([0, 50) AS 'low', [50, *) AS 'high')" +) +STATS_Q = ( + "SELECT id FROM FacetDoc ORDER BY id FACETS s: score STATS, d: category DISTINCT" +) +RANGES_NAMED_Q = ( + "SELECT id FROM FacetDoc ORDER BY id " + "FACETS s: score RANGES ([:lo, :hi) AS 'mid')" +) +RANGES_POSITIONAL_Q = ( + "SELECT id FROM FacetDoc ORDER BY id FACETS s: score RANGES ([?, ?) AS 'mid')" +) +INCLUDE_NAMED_Q = ( + "SELECT id FROM FacetDoc ORDER BY id FACETS c: category TERMS INCLUDE [:x, :y]" +) +INCLUDE_POSITIONAL_Q = ( + "SELECT id FROM FacetDoc ORDER BY id FACETS c: category TERMS INCLUDE [?, ?]" +) + + +@pytest.fixture +def test_db(tmp_path): + """Create a temporary test database.""" + db = create_database(str(tmp_path / "test_result_set_facets_db")) + yield db + db.drop() + + +def _engine_version(db): + """The engine's x.y.z through SQL `SELECT version()`; a SNAPSHOT or a build suffix is dropped.""" + raw = str(db.query("sql", "SELECT version() AS v").first().get("v")) + parts = [ + int(p) if p.isdigit() else 0 + for p in raw.split()[0].split("-")[0].split(".")[:3] + ] + return tuple(parts + [0] * (3 - len(parts))) + + +def _older_engine_refuses(db, sql, *args): + """True when the engine predates the ``FACETS`` clause: the statement is then refused with its syntax error, and the caller + has nothing more to check on this engine (an engine without the clause reads FACETS as a table alias). + """ + if _engine_version(db) >= FACETS_SINCE: + return False + with pytest.raises(Exception, match="SQL syntax error"): + db.query("sql", sql, *args).to_list() + return True + + +def _make_docs(db): + db.command("sql", "CREATE DOCUMENT TYPE FacetDoc") + db.command("sql", "CREATE PROPERTY FacetDoc.id INTEGER") + db.command("sql", "CREATE PROPERTY FacetDoc.category STRING") + db.command("sql", "CREATE PROPERTY FacetDoc.score INTEGER") + db.command("sql", "CREATE PROPERTY FacetDoc.ts DATETIME") + with db.transaction(): + for i, category in enumerate(CATEGORIES): + db.command( + "sql", + "INSERT INTO FacetDoc SET id = ?, category = ?, score = ?, ts = ?", + i, + category, + 10 * (i + 1), + datetime(2024, MONTH_OF[i], 10 + i, 12, 0, 0), + ) + + +def _index_docs(db): + """The COLUMNAR indexes, on an engine that has them (the older branch of every test returned before this).""" + for prop in ("category", "score", "ts"): + db.command("sql", f"CREATE INDEX ON FacetDoc ({prop}) COLUMNAR") + + +def _docs(db): + _make_docs(db) + if _engine_version(db) >= FACETS_SINCE: + _index_docs(db) + + +def _counts(facet): + """A terms facet as {value: count}, the bucket of the records without a value left out.""" + return {b["value"]: b["count"] for b in facet["buckets"] if not b.get("missing")} + + +def _group_by_counts(db, where=""): + rows = db.query( + "sql", + f"SELECT category, count(*) AS n FROM FacetDoc {where} GROUP BY category", # nosec B608 - test-owned text + ).to_list() + return {r["category"]: r["n"] for r in rows} + + +def _same_float(actual, expected): + assert isinstance(actual, float), actual + if math.isnan(expected): + assert math.isnan(actual) + else: + assert actual == expected + + +class TestResultSetFacets: + def test_plain_query_has_no_facets(self, test_db): + _make_docs(test_db) + + rs = test_db.query("sql", "SELECT FROM FacetDoc ORDER BY id") + assert rs.facets is None + # asking did not end the rows of a statement that has no facets + assert [r["id"] for r in rs.to_list()] == list(range(10)) + assert rs.facets is None + + rs = test_db.command("sql", "SELECT FROM FacetDoc ORDER BY id") + assert rs.facets is None + assert len(rs.to_list()) == 10 + + def test_facets_of_a_closed_result_set_raise(self, test_db): + _make_docs(test_db) + + rs = test_db.query("sql", "SELECT FROM FacetDoc") + rs.close() + with pytest.raises(ArcadeDBError, match="closed before"): + rs.facets + + def test_older_engine_refuses_the_clause(self, test_db): + _make_docs(test_db) + + if _engine_version(test_db) >= FACETS_SINCE: + # the newer branch of every test below: the clause is accepted + rs = test_db.query("sql", TERMS_Q) + assert len(rs.to_list()) == 10 + assert rs.facets is not None + return + assert _older_engine_refuses(test_db, TERMS_Q) + + def test_terms(self, test_db): + _docs(test_db) + if _older_engine_refuses(test_db, TERMS_Q): + return + + rs = test_db.query("sql", TERMS_Q) + rows = rs.to_list() # the rows first, then the facets + facets = rs.facets + + assert [r["id"] for r in rows] == list(range(10)) + assert list(facets) == ["meta", "category"] + meta = facets["meta"] + assert set(meta) == {"exact", "patched", "lag_ms"} + assert isinstance(meta["exact"], bool) + assert isinstance(meta["patched"], int) and meta["patched"] >= 0 + assert isinstance(meta["lag_ms"], int) and meta["lag_ms"] >= 0 + + category = facets["category"] + assert category["type"] == "terms" + assert category["source"] in ("index", "scan") + assert "reason" not in category + # TOP 2: the biggest value, then one of the two with two records; the counts of the rest are other_count + assert category["buckets"][0] == {"value": "a", "count": 5} + assert len(category["buckets"]) == 2 + assert category["buckets"][1]["count"] == 2 + assert category["other_count"] == 3 + assert category["missing"] == 0 + + def test_facets_cover_every_match_whatever_skip_and_limit_kept(self, test_db): + _docs(test_db) + if _older_engine_refuses(test_db, PAGED_Q): + return + + rs = test_db.query("sql", PAGED_Q) + rows = rs.to_list() + facets = rs.facets + + assert [r["id"] for r in rows] == [2, 3, 4] + assert _counts(facets["category"]) == CATEGORY_COUNTS + + def test_ranges(self, test_db): + _docs(test_db) + if _older_engine_refuses(test_db, RANGES_Q): + return + + rs = test_db.query("sql", RANGES_Q) + rs.to_list() + score = rs.facets["s"] + + assert score["type"] == "range" + assert score["source"] in ("index", "scan") + assert "other_count" not in score # every interval is returned + low, high = score["buckets"] + assert (low["key"], low["from"], low["to"], low["count"]) == ("low", 0, 50, 4) + # an unbounded end is left out + assert (high["key"], high["from"], high["count"]) == ("high", 50, 6) + assert "to" not in high + assert score["missing"] == 0 + + def test_ranges_default_bucket(self, test_db): + _docs(test_db) + if _older_engine_refuses(test_db, RANGES_DEFAULT_Q): + return + + rs = test_db.query("sql", RANGES_DEFAULT_Q) + rs.to_list() + buckets = rs.facets["s"]["buckets"] + + assert [(b["key"], b["count"]) for b in buckets] == [("low", 2), ("rest", 8)] + assert buckets[1]["default"] is True + assert "default" not in buckets[0] + + def test_calendar_histogram(self, test_db): + _docs(test_db) + if _older_engine_refuses(test_db, HISTOGRAM_Q): + return + + rs = test_db.query("sql", HISTOGRAM_Q) + rs.to_list() + hist = rs.facets["h"] + + assert hist["type"] == "histogram" + assert hist["source"] in ("index", "scan") + # one bucket per month, in time order (an empty month, March, may or may not be listed) + full = [b for b in hist["buckets"] if b["count"]] + assert [b["count"] for b in full] == [4, 3, 3] + assert all(b["key"] and "from" in b and "to" in b for b in full) + assert hist["missing"] == 0 + + def test_disjunctive_filter(self, test_db): + _docs(test_db) + if _older_engine_refuses(test_db, FILTER_Q): + return + + rs = test_db.query("sql", FILTER_Q) + rows = rs.to_list() + facets = rs.facets + + # the rows keep only what every FILTER passes + assert [r["id"] for r in rows] == [0, 2, 4, 6, 8] + # the facet that owns the filter ignores it (disjunctive): every category is still counted + assert _counts(facets["c"]) == CATEGORY_COUNTS + # the other facets count what the filter passes: the five records of category a (scores 10, 30, 50, 70, 90) + assert [(b["key"], b["count"]) for b in facets["s"]["buckets"]] == [ + ("low", 2), + ("high", 3), + ] + + def test_stats_and_distinct(self, test_db): + _docs(test_db) + if _older_engine_refuses(test_db, STATS_Q): + return + + rs = test_db.query("sql", STATS_Q) + rs.to_list() + facets = rs.facets + + stats = facets["s"] + assert stats["type"] == "stats" + assert "buckets" not in stats + assert (stats["count"], stats["missing"]) == (10, 0) + assert (stats["min"], stats["max"], stats["sum"]) == (10, 100, 550) + assert stats["avg"] == 55 + distinct = facets["d"] + assert distinct["type"] == "distinct" + assert (distinct["value"], distinct["exact"]) == (4, True) + + @pytest.mark.parametrize( + "name, values", + [ + ("OddNan", [1.5, math.nan]), + ("OddPositive", [1.5, math.inf]), + ("OddNegative", [1.5, -math.inf]), + ], + ) + def test_non_finite_values_are_floats(self, test_db, name, values): + sql = f"SELECT FROM {name} FACETS s: x STATS" # nosec B608 - test-owned name + test_db.command("sql", f"CREATE DOCUMENT TYPE {name}") + test_db.command("sql", f"CREATE PROPERTY {name}.x DOUBLE") + if _older_engine_refuses(test_db, sql): + return + test_db.command("sql", f"CREATE INDEX ON {name} (x) COLUMNAR") + with test_db.transaction(): + for value in values: + test_db.command("sql", f"INSERT INTO {name} SET x = :x", {"x": value}) + + rs = test_db.query("sql", sql) + assert len(rs.to_list()) == 2 + stats = rs.facets["s"] + + assert stats["type"] == "stats" and stats["count"] == 2 + # the engine's own JSON writers turn these into null: here they are the floats a row holds + _same_float(stats["sum"], sum(values)) + _same_float(stats["avg"], sum(values) / 2) + if math.isnan(values[1]): + assert stats["min"] == 1.5 + assert isinstance(stats["max"], float) + elif values[1] > 0: + assert (stats["min"], stats["max"]) == (1.5, math.inf) + else: + assert (stats["min"], stats["max"]) == (-math.inf, 1.5) + + def test_named_parameters_in_ranges_and_include(self, test_db): + _docs(test_db) + params = {"lo": 30, "hi": 70} + if _older_engine_refuses(test_db, RANGES_NAMED_Q, params): + return + + for run in (test_db.query, test_db.command): + rs = run("sql", RANGES_NAMED_Q, params) + assert len(rs.to_list()) == 10 + (mid,) = rs.facets["s"]["buckets"] + # [30, 70): 30, 40, 50 and 60 + assert (mid["key"], mid["from"], mid["to"], mid["count"]) == ( + "mid", + 30, + 70, + 4, + ) + + rs = run("sql", INCLUDE_NAMED_Q, {"x": "a", "y": "c"}) + rs.to_list() + assert _counts(rs.facets["c"]) == {"a": 5, "c": 2} + + def test_positional_parameters_in_ranges_and_include(self, test_db): + _docs(test_db) + if _older_engine_refuses(test_db, RANGES_POSITIONAL_Q, 30, 70): + return + + for run in (test_db.query, test_db.command): + rs = run("sql", RANGES_POSITIONAL_Q, 30, 70) + rs.to_list() + (mid,) = rs.facets["s"]["buckets"] + assert (mid["key"], mid["count"]) == ("mid", 4) + + rs = run("sql", INCLUDE_POSITIONAL_Q, "a", "c") + rs.to_list() + assert _counts(rs.facets["c"]) == {"a": 5, "c": 2} + + def test_rows_first_then_facets(self, test_db): + _docs(test_db) + if _older_engine_refuses(test_db, TERMS_Q): + return + + rs = test_db.query("sql", TERMS_Q) + seen = [row.get("id") for row in rs] # row by row, to the end + facets = rs.facets + + assert seen == list(range(10)) + assert facets is not None and facets["category"]["buckets"][0]["count"] == 5 + # the same dict on every call, and the rows stay read + assert rs.facets == facets + assert rs.to_list() == [] + + def test_facets_first_give_the_same_facets(self, test_db): + """Facets read before any row are those of reading the rows first. Whether rows not fetched yet can still be read after + that is unspecified (read the rows first if both are needed), so no row is asserted here.""" + _docs(test_db) + if _older_engine_refuses(test_db, TERMS_Q): + return + + rs = test_db.query("sql", TERMS_Q) + facets = rs.facets # before any row is read + + assert facets["category"]["buckets"][0] == {"value": "a", "count": 5} + assert rs.facets == facets + rows_first = test_db.query("sql", TERMS_Q) + rows_first.to_list() + # meta is how far the projection lagged, which can differ between two runs + assert {k: v for k, v in facets.items() if k != "meta"} == { + k: v for k, v in rows_first.facets.items() if k != "meta" + } + + def test_facets_of_a_lazy_command_cursor(self, test_db): + _docs(test_db) + if _older_engine_refuses(test_db, TERMS_Q): + return + + rs = test_db.command("sql", TERMS_Q) + facets = rs.facets + assert facets["category"]["buckets"][0] == {"value": "a", "count": 5} + + rs = test_db.command("sql", TERMS_Q) + assert len(rs.to_list()) == 10 + assert rs.facets == facets + + @pytest.mark.parametrize("run", ["query", "command"]) + @pytest.mark.parametrize("rows_first", [True, False]) + def test_facets_of_a_cursor_with_rows_left(self, test_db, run, rows_first): + """More rows than the first batch of either build reads, so the cursor is still open when the facets are asked.""" + # no ORDER BY: the rows stream, so asking for the facets first has rows left to read past + sql = "SELECT id FROM Big FACETS c: category TERMS TOP 5" + test_db.command("sql", "CREATE DOCUMENT TYPE Big") + test_db.command("sql", "CREATE PROPERTY Big.id INTEGER") + test_db.command("sql", "CREATE PROPERTY Big.category STRING") + if _older_engine_refuses(test_db, sql): + return + with test_db.transaction(): + for i in range(600): + test_db.command( + "sql", "INSERT INTO Big SET id = ?, category = ?", i, f"x{i % 3}" + ) + test_db.command("sql", "CREATE INDEX ON Big (category) COLUMNAR") + + rs = getattr(test_db, run)("sql", sql) + if rows_first: + assert sorted(r["id"] for r in rs.to_list()) == list(range(600)) + facets = rs.facets + else: + facets = rs.facets + + assert _counts(facets["c"]) == {"x0": 200, "x1": 200, "x2": 200} + + def test_facets_equal_group_by(self, test_db): + _docs(test_db) + sql = "SELECT id FROM FacetDoc WHERE score >= 30 ORDER BY id FACETS category TERMS TOP 10" + if _older_engine_refuses(test_db, sql): + return + + rs = test_db.query("sql", TERMS_Q.replace("TOP 2", "TOP 10")) + rs.to_list() + assert _counts(rs.facets["category"]) == _group_by_counts(test_db) + + # with a WHERE: the facets cover the matches, as GROUP BY does + rs = test_db.query("sql", sql) + rows = rs.to_list() + assert len(rows) == 8 + assert _counts(rs.facets["category"]) == _group_by_counts( + test_db, "WHERE score >= 30" + ) + + def test_unavailable_facet_is_surfaced(self, test_db): + """A HIERARCHY facet over a type whose subtype alone has a hierarchy: the records of the parent type have none, so the facet + is unavailable and says why, while the rows and the other facets are answered (the engine's HierarchyFacetIntegrationTest). + """ + sql = "SELECT FROM Media FACETS t: tags HIERARCHY TOP 100, n: tags TERMS TOP 5" + test_db.command("sql", "CREATE VERTEX TYPE FacetTag") + test_db.command("sql", "CREATE PROPERTY FacetTag.path STRING") + test_db.command("sql", "CREATE INDEX ON FacetTag (path) NOTUNIQUE") + test_db.command("sql", "CREATE EDGE TYPE FacetParentOf") + test_db.command("sql", "CREATE DOCUMENT TYPE Media") + test_db.command("sql", "CREATE PROPERTY Media.tags LIST OF STRING") + test_db.command("sql", "CREATE DOCUMENT TYPE Photo EXTENDS Media") + if _older_engine_refuses(test_db, sql): + return + test_db.command( + "sql", + "CREATE INDEX ON Photo (tags BY ITEM) COLUMNAR METADATA " + '{"hierarchy": {"vertexType": "FacetTag", "key": "path", ' + '"parentEdge": "FacetParentOf"}}', + ) + with test_db.transaction(): + test_db.command("sql", "CREATE VERTEX FacetTag SET path = 't0'") + test_db.command("sql", "CREATE VERTEX FacetTag SET path = 't0/t1'") + test_db.command( + "sql", + "CREATE EDGE FacetParentOf FROM (SELECT FROM FacetTag WHERE path = 't0') " + "TO (SELECT FROM FacetTag WHERE path = 't0/t1')", + ) + for _ in range(3): + test_db.command("sql", "INSERT INTO Media SET tags = ['t0/t1']") + test_db.command("sql", "INSERT INTO Photo SET tags = ['t0/t1']") + + rs = test_db.query("sql", sql) + assert len(rs.to_list()) == 6 + facets = rs.facets + + hierarchy = facets["t"] + assert hierarchy["type"] == "hierarchy" + assert hierarchy["source"] == "unavailable" + assert "Media.tags" in hierarchy["reason"] + assert "buckets" not in hierarchy + # the other facet of the same query is answered + assert facets["n"]["type"] == "terms" + assert facets["n"]["source"] in ("index", "scan") + assert _counts(facets["n"]) == {"t0/t1": 6} diff --git a/tests/test_schema.py b/tests/test_schema.py index f6eaa844..29f46ec7 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -461,7 +461,8 @@ def test_drop_index(self, test_db): indexes = schema.get_indexes() index_name = None for idx in indexes: - if "User" in idx.getTypeName() and "temp" in idx.getName(): + # the type-level index: a per-bucket index (`User_0_`) cannot be dropped alone + if idx.getName() == "User[temp]": index_name = idx.getName() break @@ -491,8 +492,9 @@ def test_drop_existing_index_force(self, test_db): schema.create_vertex_type("User") schema.create_property("User", "temp", PropertyType.STRING) schema.create_index("User", ["temp"]) + # the type-level index `User[temp]`: the engine refuses to drop one of its per-bucket indexes (`User_0_`) alone name = next( - i.getName() for i in schema.get_indexes() if i.getTypeName() == "User" + i.getName() for i in schema.get_indexes() if i.getName() == "User[temp]" ) schema.drop_index(name, force=True)