diff --git a/docs/release-26.9.1.md b/docs/release-26.9.1.md index 8a60330c76..59c2c55b67 100644 --- a/docs/release-26.9.1.md +++ b/docs/release-26.9.1.md @@ -2234,6 +2234,46 @@ purpose ("more seeds than nodes" reads as "as many as exist" and is clamped to t nameless `NegativeArraySizeException`. [#6216](https://github.com/ArcadeData/arcadedb/issues/6216) + +## The same iteration-knob guard, applied to the fourteen `algo.*` procedures #6216 left out of scope (#6264) + +[#6216](https://github.com/ArcadeData/arcadedb/issues/6216) established what an iteration-shaped knob needs - +a domain minimum rejected by name, and a checkpoint inside the loop it drives - and gave both to the three +procedures its parent review had named. Fourteen more carried the identical defect, untouched: +`algo.pageRank`, `algo.personalizedPageRank`, `algo.articleRank`, `algo.eigenvector`, `algo.hits`, +`algo.katz`, `algo.louvain`, `algo.leiden`, `algo.labelPropagation`, `algo.slpa`, `algo.simRank`, +`algo.fastrp`, `algo.hashgnn` and `algo.graphsage`. Every one extracted its knob with a plain +`extractInt(n, "maxIterations")`, and none contained a single checkpoint. + +So `CALL algo.pageRank({maxIterations: 0})` returned the *uniform initial rank vector* as though it were a +PageRank result, `algo.louvain` returned every node in its own community, and `algo.fastrp` the untouched +random projection, and `algo.graphsage` the untouched random-Gaussian initial features presented as trained +embeddings. The silent half is the more serious one: an un-iterated centrality is not obviously wrong to a +caller, unlike an exception. All fourteen now reject a value below 1 with a message naming the procedure, the +parameter and the value. + +For time there is still no honest ceiling to pick, so a large value is not forbidden but made abortable. Each +iteration loop calls the shared `WorkGuard`, which observes a thread interrupt and the +`arcadedb.command.timeout` deadline; a per-node checkpoint inside each pass bounds the abort latency below a +whole sweep of the graph, throttled to once every 1024 nodes where a node's own work is small and unthrottled +where it is already O(n). Six of the fourteen have no convergence test at all - the CSR PageRank kernel, +`algo.simRank`, `algo.fastrp`, `algo.hashgnn`, `algo.graphsage` and `algo.slpa` - so the knob alone decided +when they stopped and nothing could end the run early. + +Two of them hand the work to `GraphAlgorithms`, which lives below the query layer and knew nothing about +deadlines. Rather than couple the OLAP kernels to the query engine, `GraphAlgorithms.pageRank` and +`GraphAlgorithms.labelPropagation` gained an overload taking a `WorkCheckpoint`, a one-method interface in +`com.arcadedb.graph.olap` that the procedures satisfy with a method reference to their own guard. Existing +callers are unchanged and get a checkpoint that never aborts. + +`algo.slpa` needed one thing more. Alone among the fourteen its `iterations` buys heap as well as time: every +node keeps a label-memory row of `iterations + 1` ints, so `{iterations: 1000000}` on a 10k-node graph asks +for 40 GB, and at `Integer.MAX_VALUE` the `iterations + 1` wrapped to `Integer.MIN_VALUE` and died as a bare +`NegativeArraySizeException` naming nothing. The footprint is now estimated in saturating `long` arithmetic +and checked against the same `arcadedb.cypher.algoMaxWalkMemory` budget the walk buffers use, before the +first row is allocated. + +[#6264](https://github.com/ArcadeData/arcadedb/issues/6264) ## A placeholder whose content had to spill into chunks is no longer returned twice by a scan (#6196) `SELECT FROM Doc` could return the same record twice, under two different RIDs, and `count(@rid)` counted it diff --git a/engine/src/main/java/com/arcadedb/GlobalConfiguration.java b/engine/src/main/java/com/arcadedb/GlobalConfiguration.java index b5a99e71a2..15745c3cc5 100644 --- a/engine/src/main/java/com/arcadedb/GlobalConfiguration.java +++ b/engine/src/main/java/com/arcadedb/GlobalConfiguration.java @@ -815,15 +815,16 @@ client error before any element is generated. Negative number means no limit (th CYPHER_ALGO_MAX_WALK_MEMORY("arcadedb.cypher.algoMaxWalkMemory", SCOPE.DATABASE, """ - Maximum heap, in bytes, that a single call to an OpenCypher random-walk algorithm procedure may reserve for \ - its walk buffers: algo.node2vec materialises walksPerNode x nodeCount walks of walkLength steps each, and \ - algo.randomWalk a single walk of steps entries. Those knobs have no graph-derived ceiling to clamp against - \ + Maximum heap, in bytes, that a single call to an OpenCypher algorithm procedure may reserve for the per-node \ + int buffers one of its knobs sizes: algo.node2vec materialises walksPerNode x nodeCount walks of walkLength \ + steps each, algo.randomWalk a single walk of steps entries, and algo.slpa one label-memory row of iterations \ + entries per node. Those knobs have no graph-derived ceiling to clamp against - \ unlike a top-k bound, which is capped by the node count - so a large but perfectly in-range int would \ otherwise reach the allocator unchecked, or wrap the int product on the way there and surface as a \ - NegativeArraySizeException from inside the walk generator. The estimate is computed in saturating long \ + NegativeArraySizeException from inside the algorithm. The estimate is computed in saturating long \ arithmetic and checked BEFORE anything is allocated: a call over the budget is rejected as a client error \ naming the knobs that produced the estimate and this setting. Negative number means no limit. When left at \ - the default it auto-scales with the JVM max heap (one eighth of it, never below 64MB), so the walk buffers \ + the default it auto-scales with the JVM max heap (one eighth of it, never below 64MB), so the buffers \ of a legitimate large run stay a fraction of the heap they share with the rest of the query.""", Long.class, 64 * 1024 * 1024L, null, value -> { // Auto-scale the default with the JVM max heap: one eighth of it, never below the 64MB floor so that a diff --git a/engine/src/main/java/com/arcadedb/graph/olap/GraphAlgorithms.java b/engine/src/main/java/com/arcadedb/graph/olap/GraphAlgorithms.java index 8285443c01..a4deecbcb7 100644 --- a/engine/src/main/java/com/arcadedb/graph/olap/GraphAlgorithms.java +++ b/engine/src/main/java/com/arcadedb/graph/olap/GraphAlgorithms.java @@ -163,6 +163,27 @@ public static void awaitFutures(final Future[] futures, final int count) { */ public static double[] pageRank(final GraphAnalyticalView view, final double damping, final int iterations, final DIRECTION direction, final String... edgeTypes) { + return pageRank(view, damping, iterations, direction, WorkCheckpoint.NONE, edgeTypes); + } + + /** + * {@link #pageRank(GraphAnalyticalView, double, int, DIRECTION, String...)} with a cooperative abort hook. + *

+ * This kernel has no convergence test - it always runs the full {@code iterations} count - and that count comes + * straight from a caller-supplied knob, so without a checkpoint {@code algo.pageRank({maxIterations: 2000000000})} + * spins with nothing able to stop it. The hook is called once per power iteration, which bounds abort latency by + * one sweep of the graph and costs one virtual call per O(n + m) of work. + *

+ * One sweep is deliberately coarser than the ~1024-node latency the inline OLTP loops of the {@code algo.*} + * procedures achieve, and it is as fine as this kernel can be: the per-node work here runs inside + * {@link #parallelForRange}, on worker threads that would not observe an interrupt aimed at the calling thread, + * and throwing out of a chunk closure would leave its siblings running. So the checkpoint stays on the calling + * thread, between the parallel phases. + * + * @param checkpoint called between iterations; throws to abort. {@link WorkCheckpoint#NONE} to run unbounded + */ + public static double[] pageRank(final GraphAnalyticalView view, final double damping, + final int iterations, final DIRECTION direction, final WorkCheckpoint checkpoint, final String... edgeTypes) { final int n = view.getNodeMapping().size(); if (n == 0) return new double[0]; @@ -224,6 +245,7 @@ public static double[] pageRank(final GraphAnalyticalView view, final double dam danglingNodes[dIdx++] = u; for (int iter = 0; iter < iterations; iter++) { + checkpoint.check(); final double base = (1.0 - damping) / n; final double[] currentRank = rank; final double[] nextRank = next; @@ -1117,6 +1139,22 @@ private static double getWeight(final int fwdIdx, final double[] doubleData, fin */ public static int[] labelPropagation(final GraphAnalyticalView view, final int maxIters, final String... edgeTypes) { + return labelPropagation(view, maxIters, WorkCheckpoint.NONE, edgeTypes); + } + + /** + * {@link #labelPropagation(GraphAnalyticalView, int, String...)} with a cooperative abort hook. + *

+ * The convergence test ({@code break} once no label moved) is not a bound on {@code maxIters}: a graph that + * oscillates between two labellings never converges, so the caller-supplied knob is what decides when the run + * ends. The hook is called once per iteration, which bounds abort latency by one sweep of the graph - coarser than + * the inline OLTP loops of the {@code algo.*} procedures, and as fine as this kernel can be, for the reason given + * on {@link #pageRank(GraphAnalyticalView, double, int, DIRECTION, WorkCheckpoint, String...)}. + * + * @param checkpoint called between iterations; throws to abort. {@link WorkCheckpoint#NONE} to run unbounded + */ + public static int[] labelPropagation(final GraphAnalyticalView view, final int maxIters, + final WorkCheckpoint checkpoint, final String... edgeTypes) { final int n = view.getNodeMapping().size(); if (n == 0) return new int[0]; @@ -1161,6 +1199,7 @@ public static int[] labelPropagation(final GraphAnalyticalView view, final int m final int maxDeg = maxDegree; for (int iter = 0; iter < maxIters; iter++) { + checkpoint.check(); System.arraycopy(labels, 0, newLabels, 0, n); final AtomicBoolean anyChanged = new AtomicBoolean(false); diff --git a/engine/src/main/java/com/arcadedb/graph/olap/WorkCheckpoint.java b/engine/src/main/java/com/arcadedb/graph/olap/WorkCheckpoint.java new file mode 100644 index 0000000000..9ed852945a --- /dev/null +++ b/engine/src/main/java/com/arcadedb/graph/olap/WorkCheckpoint.java @@ -0,0 +1,50 @@ +/* + * Copyright © 2021-present Arcade Data Ltd (info@arcadedata.com) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * SPDX-FileCopyrightText: 2021-present Arcade Data Ltd (info@arcadedata.com) + * SPDX-License-Identifier: Apache-2.0 + */ +package com.arcadedb.graph.olap; + +/** + * Cooperative abort hook that an iterative {@link GraphAlgorithms} kernel calls once per iteration. + *

+ * The iteration count of a kernel such as {@link GraphAlgorithms#pageRank} comes straight from a caller-supplied + * knob ({@code algo.pageRank({maxIterations: ...})}), and for time there is no honest ceiling to pick: how long a + * run may legitimately take is a property of the graph, the hardware and the caller's patience, not of the + * parameter. So a large value is not forbidden, it is made abortable - and the kernel, which knows nothing about + * queries, timeouts or threads, asks this hook whether it should still be running. + *

+ *

+ * The hook exists rather than a direct dependency on the query layer's guard because {@code com.arcadedb.graph.olap} + * sits below {@code com.arcadedb.query}: the OpenCypher procedures pass their own guard as a method + * reference, and the kernels stay free of any knowledge of it. + *

+ * + * @author Luca Garulli (l.garulli@arcadedata.com) + */ +@FunctionalInterface +public interface WorkCheckpoint { + /** Checkpoint for a caller with nothing to cancel: never aborts, and the JIT folds it away. */ + WorkCheckpoint NONE = () -> { + }; + + /** + * Throws to abort the algorithm, or returns to let it continue. Called from the calling thread between two + * iterations, never from inside a parallel chunk, so an implementation does not have to be thread-safe and the + * exception it throws propagates straight to the caller of the kernel. + */ + void check(); +} diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AbstractAlgoProcedure.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AbstractAlgoProcedure.java index 12df0f74ae..b0faa773bc 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AbstractAlgoProcedure.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AbstractAlgoProcedure.java @@ -253,9 +253,9 @@ protected int extractCount(final Number value, final String paramName) { // ── Work bounds ────────────────────────────────────────────────────────── /** - * Heap cost of one row of a walk matrix on top of its payload: 16-byte array header, 4-byte length, 4 bytes of - * padding and the 8-byte reference the enclosing array holds. Walk rows are typically short, so this overhead - * is a real part of the footprint rather than a rounding error. + * Heap cost of one row of a per-node {@code int} matrix on top of its payload: 16-byte array header, 4-byte + * length, 4 bytes of padding and the 8-byte reference the enclosing array holds. Such rows are typically short, + * so this overhead is a real part of the footprint rather than a rounding error. *

* A heuristic for a budget check, not a guarantee: the true figure moves with the JVM's object layout * (compressed oops on or off, alignment). It does not need to be exact - it only has to keep the estimate in @@ -269,22 +269,39 @@ protected int extractCount(final Number value, final String paramName) { /** * Rejects, before a single byte is allocated, a random-walk buffer whose estimated heap footprint exceeds * {@link GlobalConfiguration#CYPHER_ALGO_MAX_WALK_MEMORY}. - *

- * The knobs that size these buffers ({@code walksPerNode}, {@code walkLength}, {@code steps}) have no - * graph-derived ceiling to clamp against, so they are bounded here by the resource they actually consume - * rather than by a guessed per-knob maximum: the budget scales with the JVM heap and is tunable, and the - * estimate is computed in saturating {@code long} arithmetic, so a product that would wrap {@code int} is - * caught here instead of surfacing as a {@code NegativeArraySizeException} from inside the walk generator. * * @param db database whose configuration carries the budget * @param estimatedBytes estimated footprint, computed with {@link #saturatingProduct(long, long)} * @param detail breakdown of the knobs that produced the estimate, for the error message + * + * @see #checkBufferBudget(Database, long, String, String) */ protected void checkWalkBudget(final Database db, final long estimatedBytes, final String detail) { + checkBufferBudget(db, estimatedBytes, "random walk buffer", detail); + } + + /** + * Rejects, before a single byte is allocated, a per-node buffer whose estimated heap footprint exceeds + * {@link GlobalConfiguration#CYPHER_ALGO_MAX_WALK_MEMORY}. + *

+ * The knobs that size these buffers ({@code walksPerNode}, {@code walkLength}, {@code steps}, SLPA's + * {@code iterations}) have no graph-derived ceiling to clamp against, so they are bounded here by the resource + * they actually consume rather than by a guessed per-knob maximum: the budget scales with the JVM heap and is + * tunable, and the estimate is computed in saturating {@code long} arithmetic, so a product that would wrap + * {@code int} is caught here instead of surfacing as a {@code NegativeArraySizeException} from inside the + * allocator. + * + * @param db database whose configuration carries the budget + * @param estimatedBytes estimated footprint, computed with {@link #saturatingProduct(long, long)} + * @param what name of the buffer, for the error message ("random walk buffer", ...) + * @param detail breakdown of the knobs that produced the estimate, for the error message + */ + protected void checkBufferBudget(final Database db, final long estimatedBytes, final String what, + final String detail) { final long budget = db.getConfiguration().getValueAsLong(GlobalConfiguration.CYPHER_ALGO_MAX_WALK_MEMORY); if (budget < 0 || estimatedBytes <= budget) return; - throw new IllegalArgumentException(getName() + "(): the random walk buffer would need " + throw new IllegalArgumentException(getName() + "(): the " + what + " would need " + (estimatedBytes == Long.MAX_VALUE ? "over " + Long.MAX_VALUE : Long.toString(estimatedBytes)) + " bytes (" + detail + "), more than the " + budget + " bytes allowed. Set " + GlobalConfiguration.CYPHER_ALGO_MAX_WALK_MEMORY.getKey() + " to raise the limit"); @@ -299,6 +316,24 @@ protected static long saturatingProduct(final long a, final long b) { } } + /** + * Adds two non-negative longs, saturating at {@link Long#MAX_VALUE} instead of wrapping. + *

+ * The companion to {@link #saturatingProduct(long, long)}, and required wherever a footprint estimate mixes the + * two: a saturated product plus a per-row overhead wraps to a large negative number, and a negative + * estimate passes {@link #checkBufferBudget} unconditionally - the budget check would be silently disabled by + * exactly the input it exists to refuse. No current caller can reach that (every estimate here is bounded by an + * {@code int}-sized count), so this closes the shape rather than an instance. + *

+ */ + protected static long saturatingSum(final long a, final long b) { + try { + return Math.addExact(a, b); + } catch (final ArithmeticException e) { + return Long.MAX_VALUE; + } + } + /** * Cooperative abort check for the CPU-bound loops of the algorithm procedures. *

diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoArticleRank.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoArticleRank.java index 17820c6f8d..20076a6a5a 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoArticleRank.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoArticleRank.java @@ -104,25 +104,26 @@ public Stream execute(final Object[] args, final Result inputRow, final final double dampingFactor = config != null && config.get("dampingFactor") instanceof Number num ? num.doubleValue() : 0.85; final int maxIterations = config != null && config.get("maxIterations") instanceof Number num ? - extractInt(num, "maxIterations") : 20; + extractInt(num, "maxIterations", 1) : 20; final double tolerance = config != null && config.get("tolerance") instanceof Number num ? num.doubleValue() : 0.0001; final Database db = context.getDatabase(); + final WorkGuard guard = newWorkGuard(context); // Try CSR-accelerated path final GraphTraversalProvider provider = findProvider(db, null); if (provider != null) { context.setVariable(CommandContext.CSR_ACCELERATED_VAR, true); - return executeWithCSR(provider, dampingFactor, maxIterations, tolerance); + return executeWithCSR(provider, dampingFactor, maxIterations, tolerance, guard); } // Fall back to OLTP path - return executeWithOLTP(db, dampingFactor, maxIterations, tolerance); + return executeWithOLTP(db, dampingFactor, maxIterations, tolerance, guard); } private Stream executeWithCSR(final GraphTraversalProvider provider, - final double dampingFactor, final int maxIterations, final double tolerance) { + final double dampingFactor, final int maxIterations, final double tolerance, final WorkGuard guard) { final int n = provider.getNodeCount(); if (n == 0) return Stream.empty(); @@ -151,6 +152,9 @@ private Stream executeWithCSR(final GraphTraversalProvider provider, scores[i] = initialScore; for (int iter = 0; iter < maxIterations; iter++) { + // maxIterations is a caller-supplied knob and the tolerance break only fires if the graph converges, so the + // outer loop carries the checkpoint. One iteration is O(n + m), which swallows a flag test whole. + guard.check(); final double[] newScores = new double[n]; double dangling = 0.0; @@ -162,6 +166,8 @@ private Stream executeWithCSR(final GraphTraversalProvider provider, if (hasView) { for (int i = 0; i < n; i++) { + // A single iteration walks the whole graph, so on a large one the checkpoint belongs inside the pass too. + guard.checkPeriodically(i); final int start = outView.offset(i); final int end = outView.offsetEnd(i); if (start == end) @@ -172,6 +178,8 @@ private Stream executeWithCSR(final GraphTraversalProvider provider, } } else { for (int i = 0; i < n; i++) { + // The fallback branch of the same pass - the checkpoint belongs in whichever one runs. + guard.checkPeriodically(i); final int[] neighbors = outNeighborsFallback[i]; if (neighbors.length == 0) continue; @@ -202,7 +210,7 @@ private Stream executeWithCSR(final GraphTraversalProvider provider, } private Stream executeWithOLTP(final Database db, final double dampingFactor, - final int maxIterations, final double tolerance) { + final int maxIterations, final double tolerance, final WorkGuard guard) { final List vertices = new ArrayList<>(); final Iterator iter = getAllVertices(db, null); while (iter.hasNext()) @@ -227,6 +235,9 @@ private Stream executeWithOLTP(final Database db, final double dampingFa scores[i] = initialScore; for (int iter2 = 0; iter2 < maxIterations; iter2++) { + // Same knob and same checkpoint as the CSR path above: the tolerance break only fires if the graph + // converges, so maxIterations is what ends the run and the guard is what can abort it. + guard.check(); final double[] newScores = new double[n]; double dangling = 0.0; for (int i = 0; i < n; i++) @@ -234,6 +245,8 @@ private Stream executeWithOLTP(final Database db, final double dampingFa dangling += scores[i]; for (int i = 0; i < n; i++) { + // A single iteration walks the whole graph, so on a large one the checkpoint belongs inside the pass too. + guard.checkPeriodically(i); final Vertex v = vertices.get(i); final double denom = outDegrees[i] + avgOutDeg; for (final Edge edge : v.getEdges(Vertex.DIRECTION.OUT)) { diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoEigenvectorCentrality.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoEigenvectorCentrality.java index 1a66505948..aef414c4d6 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoEigenvectorCentrality.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoEigenvectorCentrality.java @@ -80,10 +80,11 @@ public Stream execute(final Object[] args, final Result inputRow, final final String[] relTypes = args.length > 0 ? extractRelTypes(args[0]) : null; final Vertex.DIRECTION dir = args.length > 1 ? parseDirection(extractString(args[1], "direction")) : Vertex.DIRECTION.BOTH; - final int maxIterations = args.length > 2 ? extractInt((Number) args[2], "maxIterations") : 20; + final int maxIterations = args.length > 2 ? extractInt((Number) args[2], "maxIterations", 1) : 20; final double tolerance = args.length > 3 ? ((Number) args[3]).doubleValue() : 1e-6; final Database db = context.getDatabase(); + final WorkGuard guard = newWorkGuard(context); final GraphData graph = loadGraph(db, null, relTypes, context); @@ -99,8 +100,13 @@ public Stream execute(final Object[] args, final Result inputRow, final scores[i] = 1.0; for (int iteration = 0; iteration < maxIterations; iteration++) { + // maxIterations is a caller-supplied knob and the tolerance break only fires if the graph converges, so the + // outer loop carries the checkpoint. One iteration is O(n + m), which swallows a flag test whole. + guard.check(); // newScores[v] = sum of scores[u] for all neighbors u of v for (int v = 0; v < n; v++) { + // A single iteration walks the whole graph, so on a large one the checkpoint belongs inside the pass too. + guard.checkPeriodically(v); double sum = 0.0; for (final int u : adj[v]) sum += scores[u]; diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoFastRP.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoFastRP.java index 92a0c68356..7afb993168 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoFastRP.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoFastRP.java @@ -98,7 +98,7 @@ public Stream execute(final Object[] args, final Result inputRow, final final Map config = args.length > 0 ? extractMap(args[0], "config") : null; final int dimensions = config != null && config.get("dimensions") instanceof Number n ? extractEmbeddingDimension(n, "dimensions") : 128; - final int iterations = config != null && config.get("iterations") instanceof Number n ? extractInt(n, "iterations") : 4; + final int iterations = config != null && config.get("iterations") instanceof Number n ? extractInt(n, "iterations", 1) : 4; final double normStrength = config != null && config.get("normalization") instanceof Number n ? n.doubleValue() : 0.0; final double selfInfluence = config != null && config.get("selfInfluence") instanceof Number n ? n.doubleValue() : 0.0; final long seed = config != null && config.get("seed") instanceof Number n ? n.longValue() : -1L; @@ -106,6 +106,7 @@ public Stream execute(final Object[] args, final Result inputRow, final final Vertex.DIRECTION dir = parseDirection(config != null ? (String) config.get("direction") : null); final Database db = context.getDatabase(); + final WorkGuard guard = newWorkGuard(context); final GraphData graph = loadGraph(db, null, relTypes, context); final int n = graph.nodeCount; @@ -137,7 +138,12 @@ else if (r == 1) // Iterative neighbourhood propagation final double[][] newEmbed = new double[n][dimensions]; for (int iter = 0; iter < iterations; iter++) { + // iterations is a caller-supplied knob and this kernel has no convergence test at all, so it always runs the + // full count: the guard is the only thing that can end a run the caller no longer wants. + guard.check(); for (int i = 0; i < n; i++) { + // A single propagation walks the whole graph, so on a large one the checkpoint belongs inside it too. + guard.checkPeriodically(i); final int deg = degree[i]; // Self contribution for (int d = 0; d < dimensions; d++) @@ -159,8 +165,10 @@ else if (r == 1) normalizeL2(newEmbed[i]); } // Swap buffers - for (int i = 0; i < n; i++) + for (int i = 0; i < n; i++) { + guard.checkPeriodically(i); System.arraycopy(newEmbed[i], 0, embed[i], 0, dimensions); + } } return IntStream.range(0, n).mapToObj(i -> { diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoGraphSAGE.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoGraphSAGE.java index d8a29ff233..fcafb027e8 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoGraphSAGE.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoGraphSAGE.java @@ -95,12 +95,13 @@ public Stream execute(final Object[] args, final Result inputRow, final final Map config = args.length > 0 ? extractMap(args[0], "config") : null; final int outDim = config != null && config.get("embeddingDimension") instanceof Number n ? extractEmbeddingDimension(n, "embeddingDimension") : 64; - final int layers = config != null && config.get("layers") instanceof Number n ? extractInt(n, "layers") : 2; + final int layers = config != null && config.get("layers") instanceof Number n ? extractInt(n, "layers", 1) : 2; final long seed = config != null && config.get("seed") instanceof Number n ? n.longValue() : -1L; final String[] relTypes = config != null ? extractRelTypes(config.get("relTypes")) : null; final Vertex.DIRECTION dir = parseDirection(config != null ? (String) config.get("direction") : null); final Database db = context.getDatabase(); + final WorkGuard guard = newWorkGuard(context); final GraphData graph = loadGraph(db, null, relTypes, context); final int n = graph.nodeCount; @@ -132,6 +133,11 @@ public Stream execute(final Object[] args, final Result inputRow, final // Layer-wise aggregation for (int layer = 0; layer < layers; layer++) { + // layers is a caller-supplied knob and this kernel has no convergence test at all, so it always runs the full + // count: the guard is the only thing that can end a run the caller no longer wants. The Xavier initialisation + // below carries no checkpoint of its own because both its bounds are embedding dimensions, capped at + // MAX_EMBEDDING_DIMENSION - it is bounded work whatever the caller asks for, unlike the per-node loop. + guard.check(); // Random projection: (2*curDim) → outDim with Xavier initialisation final int concatDim = curDim * 2; final double projScale = Math.sqrt(2.0 / (concatDim + outDim)); @@ -145,6 +151,8 @@ public Stream execute(final Object[] args, final Result inputRow, final final double[] concat = new double[concatDim]; for (int i = 0; i < n; i++) { + // A single layer walks the whole graph, so on a large one the checkpoint belongs inside the layer too. + guard.checkPeriodically(i); // Mean aggregation over neighbours final int deg = degree[i]; Arrays.fill(agg, 0.0); diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoHITS.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoHITS.java index d6f0c786e8..e4c3a2a43a 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoHITS.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoHITS.java @@ -80,10 +80,11 @@ public Stream execute(final Object[] args, final Result inputRow, final validateArgs(args); final String[] relTypes = args.length > 0 ? extractRelTypes(args[0]) : null; - final int maxIterations = args.length > 1 ? extractInt((Number) args[1], "maxIterations") : 20; + final int maxIterations = args.length > 1 ? extractInt((Number) args[1], "maxIterations", 1) : 20; final double tolerance = args.length > 2 ? ((Number) args[2]).doubleValue() : 1e-6; final Database db = context.getDatabase(); + final WorkGuard guard = newWorkGuard(context); final GraphData graph = loadGraph(db, null, relTypes, context); @@ -105,8 +106,13 @@ public Stream execute(final Object[] args, final Result inputRow, final } for (int iter2 = 0; iter2 < maxIterations; iter2++) { + // maxIterations is a caller-supplied knob and the tolerance break only fires if the graph converges, so the + // outer loop carries the checkpoint. One iteration is O(n + m), which swallows a flag test whole. + guard.check(); // Update auth[v] = sum of hub[u] for all in-neighbors u of v for (int v = 0; v < n; v++) { + // A single iteration walks the whole graph, so on a large one the checkpoint belongs inside the pass too. + guard.checkPeriodically(v); double sum = 0.0; for (final int u : adjIn[v]) sum += hub[u]; @@ -115,6 +121,7 @@ public Stream execute(final Object[] args, final Result inputRow, final // Update hub[v] = sum of auth[w] for all out-neighbors w of v for (int v = 0; v < n; v++) { + guard.checkPeriodically(v); double sum = 0.0; for (final int w : adjOut[v]) sum += newAuth[w]; diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoHashGNN.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoHashGNN.java index a63ffc5cc8..8800122c31 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoHashGNN.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoHashGNN.java @@ -94,12 +94,13 @@ public Stream execute(final Object[] args, final Result inputRow, final final Map config = args.length > 0 ? extractMap(args[0], "config") : null; final int embDim = config != null && config.get("embeddingDimension") instanceof Number n ? extractEmbeddingDimension(n, "embeddingDimension") : 128; - final int iterations = config != null && config.get("iterations") instanceof Number n ? extractInt(n, "iterations") : 4; + final int iterations = config != null && config.get("iterations") instanceof Number n ? extractInt(n, "iterations", 1) : 4; final long seed = config != null && config.get("seed") instanceof Number n ? n.longValue() : -1L; final String[] relTypes = config != null ? extractRelTypes(config.get("relTypes")) : null; final Vertex.DIRECTION dir = parseDirection(config != null ? (String) config.get("direction") : null); final Database db = context.getDatabase(); + final WorkGuard guard = newWorkGuard(context); final GraphData graph = loadGraph(db, null, relTypes, context); final int n = graph.nodeCount; @@ -141,7 +142,12 @@ public Stream execute(final Object[] args, final Result inputRow, final // Iterative message passing: OR-combine neighbour features, then MinHash-reduce final boolean[][] newFeatures = new boolean[n][numFeatures]; for (int iter = 0; iter < iterations; iter++) { + // iterations is a caller-supplied knob and this kernel has no convergence test at all, so it always runs the + // full count: the guard is the only thing that can end a run the caller no longer wants. + guard.check(); for (int i = 0; i < n; i++) { + // A single message-passing round walks the whole graph, so on a large one the checkpoint belongs inside it. + guard.checkPeriodically(i); System.arraycopy(features[i], 0, newFeatures[i], 0, numFeatures); for (final int j : adj[i]) { for (int f = 0; f < numFeatures; f++) @@ -149,8 +155,10 @@ public Stream execute(final Object[] args, final Result inputRow, final } } // Swap - for (int i = 0; i < n; i++) + for (int i = 0; i < n; i++) { + guard.checkPeriodically(i); System.arraycopy(newFeatures[i], 0, features[i], 0, numFeatures); + } } // Compute MinHash signature → float embedding diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoKatz.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoKatz.java index eb9357217e..bf4c1ecc10 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoKatz.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoKatz.java @@ -85,10 +85,11 @@ public Stream execute(final Object[] args, final Result inputRow, final final String[] relTypes = args.length > 0 ? extractRelTypes(args[0]) : null; final double alpha = args.length > 1 ? ((Number) args[1]).doubleValue() : 0.005; - final int maxIterations = args.length > 2 ? extractInt((Number) args[2], "maxIterations") : 100; + final int maxIterations = args.length > 2 ? extractInt((Number) args[2], "maxIterations", 1) : 100; final double tolerance = args.length > 3 ? ((Number) args[3]).doubleValue() : 1e-6; final Database db = context.getDatabase(); + final WorkGuard guard = newWorkGuard(context); final GraphData graph = loadGraph(db, null, relTypes, context); @@ -104,8 +105,13 @@ public Stream execute(final Object[] args, final Result inputRow, final Arrays.fill(scores, 1.0); for (int iter2 = 0; iter2 < maxIterations; iter2++) { + // maxIterations is a caller-supplied knob and the tolerance break only fires if the graph converges, so the + // outer loop carries the checkpoint. One iteration is O(n + m), which swallows a flag test whole. + guard.check(); // newScores[i] = alpha * sum_{j in inNeighbors(i)} scores[j] + 1 for (int i = 0; i < n; i++) { + // A single iteration walks the whole graph, so on a large one the checkpoint belongs inside the pass too. + guard.checkPeriodically(i); double sum = 0.0; for (final int j : adjIn[i]) sum += scores[j]; diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoLabelPropagation.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoLabelPropagation.java index b09987b4cc..49e495b6c6 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoLabelPropagation.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoLabelPropagation.java @@ -100,29 +100,33 @@ public Stream execute(final Object[] args, final Result inputRow, final final Map config = args.length > 0 ? extractMap(args[0], "config") : null; final int maxIterations = config != null && config.get("maxIterations") instanceof Number n ? - extractInt(n, "maxIterations") : 10; + extractInt(n, "maxIterations", 1) : 10; final Database db = context.getDatabase(); + final WorkGuard guard = newWorkGuard(context); // Try CSR-accelerated path: delegate to native label propagation on CSR arrays final GraphTraversalProvider provider = findProvider(db, null); if (provider instanceof GraphAnalyticalView gav) { context.setVariable(CommandContext.CSR_ACCELERATED_VAR, true); - return executeWithCSR(context, gav, maxIterations); + return executeWithCSR(context, gav, maxIterations, guard); } // Fall back to OLTP path final String directionStr = config != null && config.get("direction") instanceof String s ? s : "BOTH"; final Vertex.DIRECTION direction = parseDirection(directionStr); - return executeWithOLTP(db, maxIterations, direction); + return executeWithOLTP(db, maxIterations, direction, guard); } - private Stream executeWithCSR(final CommandContext context, final GraphAnalyticalView gav, final int maxIterations) { + private Stream executeWithCSR(final CommandContext context, final GraphAnalyticalView gav, + final int maxIterations, final WorkGuard guard) { final int n = gav.getNodeCount(); if (n == 0) return Stream.empty(); - final int[] labels = GraphAlgorithms.labelPropagation(gav, maxIterations); + // The kernel's "nothing moved" break only fires if the labelling settles - a graph that oscillates between two + // labellings never converges - so maxIterations is what ends the run, and the guard is what can abort it. + final int[] labels = GraphAlgorithms.labelPropagation(gav, maxIterations, guard::check); context.setVariable(CommandContext.RESULT_COUNT_HINT_VAR, (long) n); return IntStream.range(0, n).mapToObj(i -> { @@ -134,7 +138,7 @@ private Stream executeWithCSR(final CommandContext context, final GraphA } private Stream executeWithOLTP(final Database db, final int maxIterations, - final Vertex.DIRECTION direction) { + final Vertex.DIRECTION direction, final WorkGuard guard) { final List vertices = new ArrayList<>(); final Iterator vertIter = getAllVertices(db, null); while (vertIter.hasNext()) @@ -155,10 +159,15 @@ private Stream executeWithOLTP(final Database db, final int maxIteration // Synchronous label propagation: compute all new labels, then apply for (int iter = 0; iter < maxIterations; iter++) { + // maxIterations is a caller-supplied knob and the "nothing moved" break only fires if the labelling settles, + // so the outer loop carries the checkpoint. One iteration is O(n + m), which swallows a flag test whole. + guard.check(); final int[] newLabel = new int[n]; boolean changed = false; for (int i = 0; i < n; i++) { + // A single iteration walks the whole graph, so on a large one the checkpoint belongs inside the pass too. + guard.checkPeriodically(i); final int[] neighbors = adj[i]; if (neighbors.length == 0) { newLabel[i] = label[i]; diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoLeiden.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoLeiden.java index 2cb6fb6a2a..c1055fbb77 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoLeiden.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoLeiden.java @@ -81,10 +81,11 @@ public Stream execute(final Object[] args, final Result inputRow, final validateArgs(args); final String[] relTypes = args.length > 0 ? extractRelTypes(args[0]) : null; - final int maxIterations = args.length > 1 && args[1] instanceof Number n ? extractInt(n, "maxIterations") : 10; + final int maxIterations = args.length > 1 && args[1] instanceof Number n ? extractInt(n, "maxIterations", 1) : 10; final double resolution = args.length > 2 && args[2] instanceof Number n ? n.doubleValue() : 1.0; final Database db = context.getDatabase(); + final WorkGuard guard = newWorkGuard(context); final GraphData graph = loadGraph(db, null, relTypes, context); @@ -116,10 +117,15 @@ public Stream execute(final Object[] args, final Result inputRow, final communityDegree[i] = degree[i]; for (int iter2 = 0; iter2 < maxIterations; iter2++) { + // maxIterations is a caller-supplied knob and the "nothing moved" break only fires if the graph settles, so + // the outer loop carries the checkpoint. One iteration is two passes over every edge. + guard.check(); boolean changed = false; // Phase 1: Local moves (greedy modularity optimization) for (int i = 0; i < n; i++) { + // A single pass walks the whole graph, so on a large one the checkpoint belongs inside the pass too. + guard.checkPeriodically(i); final int currentComm = community[i]; final long ki = degree[i]; @@ -161,6 +167,7 @@ public Stream execute(final Object[] args, final Result inputRow, final if (changed) { boolean refined = false; for (int i = 0; i < n; i++) { + guard.checkPeriodically(i); final int currentComm = community[i]; final long ki = degree[i]; diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoLouvain.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoLouvain.java index 51a1feb634..ce45b593b2 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoLouvain.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoLouvain.java @@ -98,12 +98,13 @@ public Stream execute(final Object[] args, final Result inputRow, final final Map config = args.length > 0 ? extractMap(args[0], "config") : null; final int maxIterations = config != null && config.get("maxIterations") instanceof Number n ? - extractInt(n, "maxIterations") : 10; + extractInt(n, "maxIterations", 1) : 10; final double tolerance = config != null && config.get("tolerance") instanceof Number n ? n.doubleValue() : 0.0001; final String weightProperty = config != null ? (String) config.get("weightProperty") : null; final Database db = context.getDatabase(); + final WorkGuard guard = newWorkGuard(context); final List vertices = new ArrayList<>(); final Iterator vertIter = getAllVertices(db, null); while (vertIter.hasNext()) @@ -147,9 +148,15 @@ public Stream execute(final Object[] args, final Result inputRow, final // Phase 1: Modularity optimization for (int iter = 0; iter < maxIterations; iter++) { + // maxIterations is a caller-supplied knob and the "nothing moved" break only fires if the graph settles, so + // the outer loop carries the checkpoint. One iteration walks every edge, plus a modularity pass. + guard.check(); boolean changed = false; for (int i = 0; i < n; i++) { + // One node is already O(deg x n) - getCommunityDegree() scans every node once per candidate community - + // so the per-node checkpoint is unthrottled: 1024 of these would be 1024 whole-graph scans of latency. + guard.check(); final Vertex v = vertices.get(i); final int currentCommunity = community[i]; diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoNode2Vec.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoNode2Vec.java index 0b78d12294..eaae6f1033 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoNode2Vec.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoNode2Vec.java @@ -144,8 +144,12 @@ public Stream execute(final Object[] args, final Result inputRow, final // Sized in long arithmetic: `n * walksPerNode` wraps int for a large walksPerNode, which used to size the // walk matrix with a negative or (for an exact multiple of 2^32) far too small a value. final long totalWalksAsLong = saturatingProduct(n, walksPerNode); - // Per walk: one matrix row of walkLength ints, plus one entry of the walkOrder shuffle array. - final long bytesPerWalk = WALK_ROW_OVERHEAD_BYTES + WALK_ENTRY_BYTES + WALK_ENTRY_BYTES * walkLen; + // Per walk: one matrix row of walkLength ints, plus one entry of the walkOrder shuffle array. Saturating + // throughout, like the estimate itself: a footprint that mixes a saturated product with a plain addition wraps + // to a negative number, and a negative estimate passes the budget check unconditionally. walkLen is int-bounded + // so this one cannot reach that today - it is written this way so the shape is the same at every call site. + final long bytesPerWalk = saturatingSum(WALK_ROW_OVERHEAD_BYTES + WALK_ENTRY_BYTES, + saturatingProduct(WALK_ENTRY_BYTES, walkLen)); checkWalkBudget(db, saturatingProduct(totalWalksAsLong, bytesPerWalk), "walksPerNode=" + walksPerNode + " x walkLength=" + walkLen + " over " + n + " nodes"); if (totalWalksAsLong > Integer.MAX_VALUE) diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoPageRank.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoPageRank.java index 7190f7608c..edf6f1bcfc 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoPageRank.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoPageRank.java @@ -103,7 +103,7 @@ public Stream execute(final Object[] args, final Result inputRow, final final double dampingFactor = config != null && config.get("dampingFactor") instanceof Number n ? n.doubleValue() : 0.85; final int maxIterations = config != null && config.get("maxIterations") instanceof Number n ? - extractInt(n, "maxIterations") : 20; + extractInt(n, "maxIterations", 1) : 20; final double tolerance = config != null && config.get("tolerance") instanceof Number n ? n.doubleValue() : 0.0001; final String weightProperty = config != null ? (String) config.get("weightProperty") : null; @@ -112,25 +112,28 @@ public Stream execute(final Object[] args, final Result inputRow, final "IN".equalsIgnoreCase(dirStr) ? Vertex.DIRECTION.IN : Vertex.DIRECTION.OUT; final Database db = context.getDatabase(); + final WorkGuard guard = newWorkGuard(context); // Try CSR-accelerated path (only for unweighted PageRank) final GraphTraversalProvider provider = weightProperty == null ? findProvider(db, null) : null; if (provider instanceof GraphAnalyticalView gav) { context.setVariable(CommandContext.CSR_ACCELERATED_VAR, true); - return executeWithCSR(context, gav, dampingFactor, maxIterations, direction); + return executeWithCSR(context, gav, dampingFactor, maxIterations, direction, guard); } // Fall back to OLTP path - return executeWithOLTP(db, dampingFactor, maxIterations, tolerance, weightProperty, direction); + return executeWithOLTP(db, dampingFactor, maxIterations, tolerance, weightProperty, direction, guard); } private Stream executeWithCSR(final CommandContext context, final GraphAnalyticalView gav, - final double dampingFactor, final int maxIterations, final Vertex.DIRECTION direction) { + final double dampingFactor, final int maxIterations, final Vertex.DIRECTION direction, final WorkGuard guard) { final int n = gav.getNodeCount(); if (n == 0) return Stream.empty(); - final double[] scores = GraphAlgorithms.pageRank(gav, dampingFactor, maxIterations, direction); + // The CSR kernel has no convergence test at all, so maxIterations alone decides when it stops: the guard is + // the only thing that can end a run the caller no longer wants. + final double[] scores = GraphAlgorithms.pageRank(gav, dampingFactor, maxIterations, direction, guard::check); // Set result count hint for CallStep count-only optimization context.setVariable(CommandContext.RESULT_COUNT_HINT_VAR, (long) n); @@ -145,7 +148,7 @@ private Stream executeWithCSR(final CommandContext context, final GraphA private Stream executeWithOLTP(final Database db, final double dampingFactor, final int maxIterations, final double tolerance, final String weightProperty, - final Vertex.DIRECTION direction) { + final Vertex.DIRECTION direction, final WorkGuard guard) { final List vertices = new ArrayList<>(); final Iterator vertIter = getAllVertices(db, null); while (vertIter.hasNext()) @@ -216,6 +219,9 @@ private Stream executeWithOLTP(final Database db, final double dampingFa scores[i] = initialScore; for (int iter = 0; iter < maxIterations; iter++) { + // maxIterations is a caller-supplied knob and the tolerance break only fires if the graph converges, so the + // outer loop carries the checkpoint. One iteration is O(n + m), which swallows a flag test whole. + guard.check(); final double[] newScores = new double[n]; double dangling = 0.0; @@ -225,6 +231,8 @@ private Stream executeWithOLTP(final Database db, final double dampingFa } for (int i = 0; i < n; i++) { + // A single iteration walks the whole graph, so on a large one the checkpoint belongs inside the pass too. + guard.checkPeriodically(i); final int[] neighbors = outNeighbors[i]; if (neighbors.length == 0) continue; diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoPersonalizedPageRank.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoPersonalizedPageRank.java index 0115ecd6c2..a2a5e6496b 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoPersonalizedPageRank.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoPersonalizedPageRank.java @@ -85,10 +85,11 @@ public Stream execute(final Object[] args, final Result inputRow, final final Vertex sourceVertex = extractVertex(args[0], "sourceNode"); final String[] relTypes = args.length > 1 ? extractRelTypes(args[1]) : null; final double dampingFactor = args.length > 2 && args[2] instanceof Number n ? n.doubleValue() : 0.85; - final int maxIterations = args.length > 3 && args[3] instanceof Number n ? extractInt(n, "maxIterations") : 20; + final int maxIterations = args.length > 3 && args[3] instanceof Number n ? extractInt(n, "maxIterations", 1) : 20; final double tolerance = args.length > 4 && args[4] instanceof Number n ? n.doubleValue() : 1e-6; final Database db = context.getDatabase(); + final WorkGuard guard = newWorkGuard(context); // Try CSR-accelerated path final GraphTraversalProvider provider = findProvider(db, relTypes); @@ -96,16 +97,17 @@ public Stream execute(final Object[] args, final Result inputRow, final final int sourceIdx = provider.getNodeId(sourceVertex.getIdentity()); if (sourceIdx >= 0) { context.setVariable(CommandContext.CSR_ACCELERATED_VAR, true); - return executeWithCSR(provider, sourceIdx, relTypes, dampingFactor, maxIterations, tolerance); + return executeWithCSR(provider, sourceIdx, relTypes, dampingFactor, maxIterations, tolerance, guard); } } // Fall back to OLTP path - return executeWithOLTP(db, sourceVertex, relTypes, dampingFactor, maxIterations, tolerance); + return executeWithOLTP(db, sourceVertex, relTypes, dampingFactor, maxIterations, tolerance, guard); } private Stream executeWithCSR(final GraphTraversalProvider provider, final int sourceIdx, - final String[] relTypes, final double dampingFactor, final int maxIterations, final double tolerance) { + final String[] relTypes, final double dampingFactor, final int maxIterations, final double tolerance, + final WorkGuard guard) { final int n = provider.getNodeCount(); if (n == 0) return Stream.empty(); @@ -126,6 +128,9 @@ private Stream executeWithCSR(final GraphTraversalProvider provider, fin rank[sourceIdx] = 1.0; for (int iter = 0; iter < maxIterations; iter++) { + // maxIterations is a caller-supplied knob and the tolerance break only fires if the graph converges, so the + // outer loop carries the checkpoint. One iteration is O(n + m), which swallows a flag test whole. + guard.check(); final double[] newRank = new double[n]; double dangling = 0.0; for (int i = 0; i < n; i++) @@ -135,6 +140,8 @@ private Stream executeWithCSR(final GraphTraversalProvider provider, fin if (hasView) { final int[] inNbrs = inView.neighbors(); for (int i = 0; i < n; i++) { + // A single iteration walks the whole graph, so on a large one the checkpoint belongs inside the pass too. + guard.checkPeriodically(i); double incoming = 0.0; for (int k = inView.offset(i), end = inView.offsetEnd(i); k < end; k++) { final int j = inNbrs[k]; @@ -146,6 +153,8 @@ private Stream executeWithCSR(final GraphTraversalProvider provider, fin } } else { for (int i = 0; i < n; i++) { + // The fallback branch of the same pass - the checkpoint belongs in whichever one runs. + guard.checkPeriodically(i); double incoming = 0.0; for (final int j : inAdjFallback[i]) if (outDegree[j] > 0) @@ -174,7 +183,7 @@ private Stream executeWithCSR(final GraphTraversalProvider provider, fin } private Stream executeWithOLTP(final Database db, final Vertex sourceVertex, final String[] relTypes, - final double dampingFactor, final int maxIterations, final double tolerance) { + final double dampingFactor, final int maxIterations, final double tolerance, final WorkGuard guard) { final GraphData graph = loadGraph(db, null, relTypes); @@ -198,6 +207,9 @@ private Stream executeWithOLTP(final Database db, final Vertex sourceVer rank[sourceIdx] = 1.0; for (int iter2 = 0; iter2 < maxIterations; iter2++) { + // Same knob and same checkpoint as the CSR path above: the tolerance break only fires if the graph + // converges, so maxIterations is what ends the run and the guard is what can abort it. + guard.check(); final double[] newRank = new double[n]; double dangling = 0.0; for (int i = 0; i < n; i++) @@ -205,6 +217,8 @@ private Stream executeWithOLTP(final Database db, final Vertex sourceVer dangling += rank[i]; for (int i = 0; i < n; i++) { + // A single iteration walks the whole graph, so on a large one the checkpoint belongs inside the pass too. + guard.checkPeriodically(i); double incoming = 0.0; for (final int j : inAdj[i]) if (outDegree[j] > 0) diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoSLPA.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoSLPA.java index 454f82a37f..fc6310b0f3 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoSLPA.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoSLPA.java @@ -94,7 +94,7 @@ public Stream execute(final Object[] args, final Result inputRow, final final Map config = args.length > 0 ? extractMap(args[0], "config") : null; final int iterations = config != null && config.get("iterations") instanceof Number num ? - extractInt(num, "iterations") : 20; + extractInt(num, "iterations", 1) : 20; final double threshold = config != null && config.get("threshold") instanceof Number num ? num.doubleValue() : 0.1; final long seedVal = config != null && config.get("seed") instanceof Number num ? @@ -104,6 +104,7 @@ public Stream execute(final Object[] args, final Result inputRow, final final String[] relTypes = config != null ? extractRelTypes(config.get("relTypes")) : null; final Database db = context.getDatabase(); + final WorkGuard guard = newWorkGuard(context); final GraphData graph = loadGraph(db, null, relTypes, context); @@ -114,13 +115,28 @@ public Stream execute(final Object[] args, final Result inputRow, final final int[][] adj = graph.adjacency(Vertex.DIRECTION.BOTH); // Memory: memory[v] is a list of labels heard by v (including its initial label) - // Using int[] lists backed by arrays for performance + // Using int[] lists backed by arrays for performance. + // + // Unlike the other iteration knobs, SLPA's `iterations` buys heap as well as time: every node keeps one row of + // `iterations + 1` ints, so the matrix is nodeCount x (iterations + 1) and a value that merely looks large - + // {iterations: 1000000} on a 10k-node graph is 40 GB - reaches the allocator with nothing between it and the + // heap. The footprint is estimated in saturating long arithmetic and checked against the same budget the walk + // buffers use, BEFORE the first row is allocated; `iterations + 1` is computed in long because at + // Integer.MAX_VALUE the int form wraps to Integer.MIN_VALUE and died as a bare NegativeArraySizeException. + final long rowCapacity = iterations + 1L; + checkBufferBudget(db, + saturatingProduct(n, saturatingSum(saturatingProduct(rowCapacity, WALK_ENTRY_BYTES), WALK_ROW_OVERHEAD_BYTES)), + "label memory", "iterations=" + iterations + " over " + n + " nodes"); + if (rowCapacity > Integer.MAX_VALUE) + throw new IllegalArgumentException(getName() + "(): iterations=" + iterations + " needs " + rowCapacity + + " label entries per node, more than the " + Integer.MAX_VALUE + " a Java array can hold"); + final int[][] memory = new int[n][]; final int[] memorySize = new int[n]; // Each node starts with a unique label equal to its index for (int i = 0; i < n; i++) { - memory[i] = new int[iterations + 1]; + memory[i] = new int[(int) rowCapacity]; memory[i][0] = i; memorySize[i] = 1; } @@ -131,13 +147,20 @@ public Stream execute(final Object[] args, final Result inputRow, final order[i] = i; for (int t = 0; t < iterations; t++) { + // iterations is a caller-supplied knob and this kernel has no convergence test at all, so it always runs the + // full count: the guard is the only thing that can end a run the caller no longer wants. + guard.check(); // Shuffle node order each round for (int i = n - 1; i > 0; i--) { + guard.checkPeriodically(i); final int j = rng.nextInt(i + 1); final int tmp = order[i]; order[i] = order[j]; order[j] = tmp; } - for (final int listener : order) { + for (int idx = 0; idx < n; idx++) { + // A single round walks the whole graph, so on a large one the checkpoint belongs inside the round too. + guard.checkPeriodically(idx); + final int listener = order[idx]; if (adj[listener].length == 0) continue; diff --git a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoSimRank.java b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoSimRank.java index 85fdaec277..89ed5a42ad 100644 --- a/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoSimRank.java +++ b/engine/src/main/java/com/arcadedb/query/opencypher/procedures/algo/AlgoSimRank.java @@ -24,6 +24,7 @@ import com.arcadedb.query.sql.executor.Result; import com.arcadedb.query.sql.executor.ResultInternal; +import java.util.Arrays; import java.util.List; import java.util.stream.Stream; @@ -85,9 +86,10 @@ public Stream execute(final Object[] args, final Result inputRow, final final Vertex nodeB = extractVertex(args[1], "nodeB"); final String[] relTypes = args.length > 2 ? extractRelTypes(args[2]) : null; final double decayFactor = args.length > 3 ? ((Number) args[3]).doubleValue() : 0.8; - final int maxIterations = args.length > 4 ? extractInt((Number) args[4], "maxIterations") : 5; + final int maxIterations = args.length > 4 ? extractInt((Number) args[4], "maxIterations", 1) : 5; final Database db = context.getDatabase(); + final WorkGuard guard = newWorkGuard(context); final GraphData graph = loadGraph(db, null, relTypes, context); @@ -119,11 +121,20 @@ public Stream execute(final Object[] args, final Result inputRow, final sim[i][i] = 1.0; for (int iter2 = 0; iter2 < maxIterations; iter2++) { - for (int i = 0; i < n; i++) - for (int j = 0; j < n; j++) - newSim[i][j] = i == j ? 1.0 : 0.0; + // maxIterations is a caller-supplied knob and this kernel has no convergence test at all, so it always runs + // the full count: the guard is the only thing that can end a run the caller no longer wants. + guard.check(); + for (int i = 0; i < n; i++) { + guard.checkPeriodically(i); + Arrays.fill(newSim[i], 0.0); + newSim[i][i] = 1.0; + } for (int u = 0; u < n; u++) { + // Unthrottled, unlike the per-node checkpoints elsewhere in this package: the loop it guards is + // O(n^2 x deg^2), so n flag tests are at worst a 1/n fraction of it however sparse the graph is, while + // throttling to one test every 1024 nodes would leave the abort latency proportional to n itself. + guard.check(); for (int v = u + 1; v < n; v++) { final int[] inU = adjIn[u]; final int[] inV = adjIn[v]; diff --git a/engine/src/test/java/com/arcadedb/query/opencypher/procedures/algo/Issue6264AlgoIterationKnobGuardTest.java b/engine/src/test/java/com/arcadedb/query/opencypher/procedures/algo/Issue6264AlgoIterationKnobGuardTest.java new file mode 100644 index 0000000000..dbb3011e56 --- /dev/null +++ b/engine/src/test/java/com/arcadedb/query/opencypher/procedures/algo/Issue6264AlgoIterationKnobGuardTest.java @@ -0,0 +1,500 @@ +/* + * Copyright © 2021-present Arcade Data Ltd (info@arcadedata.com) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * SPDX-FileCopyrightText: 2021-present Arcade Data Ltd (info@arcadedata.com) + * SPDX-License-Identifier: Apache-2.0 + */ +package com.arcadedb.query.opencypher.procedures.algo; + +import com.arcadedb.GlobalConfiguration; +import com.arcadedb.database.Database; +import com.arcadedb.database.DatabaseFactory; +import com.arcadedb.graph.MutableVertex; +import com.arcadedb.graph.Vertex; +import com.arcadedb.graph.olap.GraphAlgorithms; +import com.arcadedb.graph.olap.GraphAnalyticalView; +import com.arcadedb.graph.olap.WorkCheckpoint; +import com.arcadedb.query.sql.executor.Result; +import com.arcadedb.query.sql.executor.ResultSet; +import com.arcadedb.utility.StallAwareStopwatch; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assertions.within; + +/** + * Regression tests for issue #6264 - the iteration-shaped knob of the fourteen {@code algo.*} procedures that + * #6216 left out of scope. The issue named thirteen; {@code algo.graphsage}'s {@code layers} is a fourteenth with + * exactly the same shape, found in review. + *

+ * #6216 established that such a knob needs two things, and gave both to {@code algo.node2vec}, + * {@code algo.maxKCut} and {@code algo.influenceMaximization} only: + *

+ * SLPA's {@code iterations} additionally buys heap rather than only time (one row of {@code iterations + 1} ints + * per node), so it is priced against the same budget the walk buffers use. + * + * @author Luca Garulli (l.garulli@arcadedata.com) + */ +class Issue6264AlgoIterationKnobGuardTest { + private Database database; + + @BeforeEach + void setup() { + final DatabaseFactory factory = new DatabaseFactory("./target/databases/test-issue-6264-algo-iteration-knobs"); + if (factory.exists()) + factory.open().drop(); + database = factory.create(); + database.getSchema().createVertexType("Node"); + database.getSchema().createEdgeType("LINK"); + + // Directed cycle A→B→C→D→A: no node is a sink, so the iterative kernels always have work to do. + database.transaction(() -> { + final MutableVertex a = database.newVertex("Node").set("name", "A").save(); + final MutableVertex b = database.newVertex("Node").set("name", "B").save(); + final MutableVertex c = database.newVertex("Node").set("name", "C").save(); + final MutableVertex d = database.newVertex("Node").set("name", "D").save(); + a.newEdge("LINK", b, true, (Object[]) null).save(); + b.newEdge("LINK", c, true, (Object[]) null).save(); + c.newEdge("LINK", d, true, (Object[]) null).save(); + d.newEdge("LINK", a, true, (Object[]) null).save(); + }); + } + + @AfterEach + void teardown() { + // A test that arms the interrupt flag must not leave it set for whatever runs next on this thread. + Thread.interrupted(); + if (database != null) + database.drop(); + } + + // ── The parameter domains: fourteen knobs, one minimum ────────────────── + + /** + * Every one of the fourteen extracted its knob with a plain {@code extractInt(n, name)} - no minimum - so a + * non-positive value was absorbed in silence and came back as a result. The value is the trip count of a loop: + * zero trips is not a cheaper answer, it is the initial state of the algorithm returned as its output. + */ + @ParameterizedTest(name = "{0}({1})") + @CsvSource(delimiter = '|', value = { + "algo.pagerank | maxIterations | CALL algo.pagerank({maxIterations: 0}) YIELD node RETURN node", + "algo.articlerank | maxIterations | CALL algo.articlerank({maxIterations: 0}) YIELD node RETURN node", + "algo.personalizedPageRank | maxIterations | MATCH (a:Node {name: 'A'}) CALL algo.personalizedPageRank(a, 'LINK', 0.85, 0) YIELD nodeId RETURN nodeId", + "algo.eigenvector | maxIterations | CALL algo.eigenvector('LINK', 'BOTH', 0) YIELD node RETURN node", + "algo.hits | maxIterations | CALL algo.hits('LINK', 0) YIELD node RETURN node", + "algo.katz | maxIterations | CALL algo.katz('LINK', 0.005, 0) YIELD nodeId RETURN nodeId", + "algo.louvain | maxIterations | CALL algo.louvain({maxIterations: 0}) YIELD node RETURN node", + "algo.leiden | maxIterations | CALL algo.leiden('LINK', 0) YIELD nodeId RETURN nodeId", + "algo.labelpropagation | maxIterations | CALL algo.labelpropagation({maxIterations: 0}) YIELD node RETURN node", + "algo.simRank | maxIterations | MATCH (a:Node {name: 'A'}), (b:Node {name: 'C'}) CALL algo.simRank(a, b, 'LINK', 0.8, 0) YIELD similarity RETURN similarity", + "algo.slpa | iterations | CALL algo.slpa({iterations: 0}) YIELD node RETURN node", + "algo.fastrp | iterations | CALL algo.fastrp({dimensions: 8, iterations: 0}) YIELD node RETURN node", + "algo.hashgnn | iterations | CALL algo.hashgnn({embeddingDimension: 8, iterations: 0}) YIELD node RETURN node", + "algo.graphsage | layers | CALL algo.graphsage({embeddingDimension: 8, layers: 0}) YIELD node RETURN node" }) + void everyIterationKnobRejectsZero(final String procedure, final String knob, final String query) { + assertThatThrownBy(() -> drain(query)) + .as("%s must refuse %s 0 by name instead of returning its own initial state as a result", procedure, knob) + .hasStackTraceContaining(procedure + "(): " + knob + " must be at least 1, got 0"); + } + + /** + * A negative count reached the same loop and behaved exactly like zero, so it needs the same refusal: the two + * differ only in that a negative one cannot even be read as "do nothing on purpose". + *

+ * Three of the fourteen, sampled rather than exhaustive, and deliberately so: what distinguishes a negative + * value from zero lives entirely in {@code extractInt(value, name, minimum)}, which all fourteen share, and the + * per-procedure half - that the knob is extracted with a minimum at all - is what the zero case above covers + * for every one of them. Fourteen more rows here would re-test one shared comparison fourteen times. + */ + @ParameterizedTest(name = "{0}({1})") + @CsvSource(delimiter = '|', value = { + "algo.pagerank | maxIterations | CALL algo.pagerank({maxIterations: -7}) YIELD node RETURN node", + "algo.hits | maxIterations | CALL algo.hits('LINK', -7) YIELD node RETURN node", + "algo.fastrp | iterations | CALL algo.fastrp({dimensions: 8, iterations: -7}) YIELD node RETURN node" }) + void anIterationKnobRejectsANegativeCount(final String procedure, final String knob, final String query) { + assertThatThrownBy(() -> drain(query)) + .hasStackTraceContaining(procedure + "(): " + knob + " must be at least 1, got -7"); + } + + /** + * Over-reach guard for the minimum. The risk a domain check carries is refusing a run that is merely small, so + * every knob is exercised at exactly its minimum and has to produce a full result set - the boundary is + * inclusive, and one iteration is a legitimate, if crude, run of each of these algorithms. + */ + @ParameterizedTest(name = "{0}") + @CsvSource(delimiter = '|', value = { + "algo.pagerank | 4 | CALL algo.pagerank({maxIterations: 1}) YIELD node RETURN node", + "algo.articlerank | 4 | CALL algo.articlerank({maxIterations: 1}) YIELD node RETURN node", + "algo.personalizedPageRank | 4 | MATCH (a:Node {name: 'A'}) CALL algo.personalizedPageRank(a, 'LINK', 0.85, 1) YIELD nodeId RETURN nodeId", + "algo.eigenvector | 4 | CALL algo.eigenvector('LINK', 'BOTH', 1) YIELD node RETURN node", + "algo.hits | 4 | CALL algo.hits('LINK', 1) YIELD node RETURN node", + "algo.katz | 4 | CALL algo.katz('LINK', 0.005, 1) YIELD nodeId RETURN nodeId", + "algo.louvain | 4 | CALL algo.louvain({maxIterations: 1}) YIELD node RETURN node", + "algo.leiden | 4 | CALL algo.leiden('LINK', 1) YIELD nodeId RETURN nodeId", + "algo.labelpropagation | 4 | CALL algo.labelpropagation({maxIterations: 1}) YIELD node RETURN node", + "algo.simRank | 1 | MATCH (a:Node {name: 'A'}), (b:Node {name: 'C'}) CALL algo.simRank(a, b, 'LINK', 0.8, 1) YIELD similarity RETURN similarity", + "algo.slpa | 4 | CALL algo.slpa({iterations: 1, seed: 1}) YIELD node RETURN node", + "algo.fastrp | 4 | CALL algo.fastrp({dimensions: 8, iterations: 1, seed: 1}) YIELD node RETURN node", + "algo.hashgnn | 4 | CALL algo.hashgnn({embeddingDimension: 8, iterations: 1, seed: 1}) YIELD node RETURN node", + "algo.graphsage | 4 | CALL algo.graphsage({embeddingDimension: 8, layers: 1, seed: 1}) YIELD node RETURN node" }) + void everyIterationKnobAcceptsItsMinimum(final String procedure, final int expectedRows, final String query) { + assertThat(drain(query)).as("%s must still run at the smallest legal setting", procedure).hasSize(expectedRows); + } + + // ── Cooperative abort: the checkpoint inside the loop ──────────────────── + + /** + * The interrupt half of the guard, on all fourteen. This is the deterministic one: the flag is armed before the + * call, so the very first checkpoint the procedure reaches has to observe it, whatever the machine's speed. + *

+ * The assertion is on the guard's own message rather than merely "something was thrown", because only + * {@code WorkGuard.check()} produces it - nothing else in the query path reports an interrupt as + * "{@code algo.x() has been interrupted}". Without the checkpoint the run simply completes and returns rows, + * which is what this test is here to fail on. + */ + @Timeout(120) + @ParameterizedTest(name = "{0}") + @CsvSource(delimiter = '|', value = { + "algo.pagerank | CALL algo.pagerank({maxIterations: 2000000000, tolerance: 0.0}) YIELD node RETURN node", + "algo.articlerank | CALL algo.articlerank({maxIterations: 2000000000, tolerance: 0.0}) YIELD node RETURN node", + "algo.personalizedPageRank | MATCH (a:Node {name: 'A'}) CALL algo.personalizedPageRank(a, 'LINK', 0.85, 2000000000, 0.0) YIELD nodeId RETURN nodeId", + "algo.eigenvector | CALL algo.eigenvector('LINK', 'BOTH', 2000000000, 0.0) YIELD node RETURN node", + "algo.hits | CALL algo.hits('LINK', 2000000000, 0.0) YIELD node RETURN node", + "algo.katz | CALL algo.katz('LINK', 0.005, 2000000000, 0.0) YIELD nodeId RETURN nodeId", + "algo.louvain | CALL algo.louvain({maxIterations: 2000000000}) YIELD node RETURN node", + "algo.leiden | CALL algo.leiden('LINK', 2000000000) YIELD nodeId RETURN nodeId", + "algo.labelpropagation | CALL algo.labelpropagation({maxIterations: 2000000000}) YIELD node RETURN node", + "algo.simRank | MATCH (a:Node {name: 'A'}), (b:Node {name: 'C'}) CALL algo.simRank(a, b, 'LINK', 0.8, 2000000000) YIELD similarity RETURN similarity", + "algo.slpa | CALL algo.slpa({iterations: 1000000, seed: 1}) YIELD node RETURN node", + "algo.fastrp | CALL algo.fastrp({dimensions: 8, iterations: 2000000000, seed: 1}) YIELD node RETURN node", + "algo.hashgnn | CALL algo.hashgnn({embeddingDimension: 8, iterations: 2000000000, seed: 1}) YIELD node RETURN node", + "algo.graphsage | CALL algo.graphsage({embeddingDimension: 8, layers: 2000000000, seed: 1}) YIELD node RETURN node" }) + void everyIterationLoopAbortsOnInterrupt(final String procedure, final String query) { + Thread.currentThread().interrupt(); + try { + assertThatThrownBy(() -> drain(query)) + .as("%s must abort at its checkpoint instead of running the whole knob out", procedure) + .hasStackTraceContaining(procedure + "() has been interrupted"); + assertThat(Thread.currentThread().isInterrupted()) + .as("the flag is consumed, so the pooled query thread is not left interrupted for the next task") + .isFalse(); + } finally { + Thread.interrupted(); + } + } + + /** + * The deadline half of the guard: {@code arcadedb.command.timeout}, which before #6216 only the SQL SELECT + * planner honoured. + *

+ * The list is twelve of the fourteen rather than all: {@code algo.louvain} and {@code algo.leiden} both stop as + * soon as no node changes community, and on this four-node cycle they settle in a handful of microseconds, so a + * deadline test on them would assert nothing about a run that is over before the clock is read. Their checkpoint + * is the same {@code WorkGuard.check()} call the other twelve use - it observes the deadline and the interrupt + * in one place - and it is covered above. + */ + @Timeout(120) + @ParameterizedTest(name = "{0}") + @CsvSource(delimiter = '|', value = { + "algo.pagerank | CALL algo.pagerank({maxIterations: 2000000000, tolerance: 0.0}) YIELD node RETURN node", + "algo.articlerank | CALL algo.articlerank({maxIterations: 2000000000, tolerance: 0.0}) YIELD node RETURN node", + "algo.personalizedPageRank | MATCH (a:Node {name: 'A'}) CALL algo.personalizedPageRank(a, 'LINK', 0.85, 2000000000, 0.0) YIELD nodeId RETURN nodeId", + "algo.eigenvector | CALL algo.eigenvector('LINK', 'BOTH', 2000000000, 0.0) YIELD node RETURN node", + "algo.hits | CALL algo.hits('LINK', 2000000000, 0.0) YIELD node RETURN node", + "algo.katz | CALL algo.katz('LINK', 0.005, 2000000000, 0.0) YIELD nodeId RETURN nodeId", + "algo.labelpropagation | CALL algo.labelpropagation({maxIterations: 2000000000}) YIELD node RETURN node", + "algo.simRank | MATCH (a:Node {name: 'A'}), (b:Node {name: 'C'}) CALL algo.simRank(a, b, 'LINK', 0.8, 2000000000) YIELD similarity RETURN similarity", + "algo.slpa | CALL algo.slpa({iterations: 1000000, seed: 1}) YIELD node RETURN node", + "algo.fastrp | CALL algo.fastrp({dimensions: 8, iterations: 2000000000, seed: 1}) YIELD node RETURN node", + "algo.hashgnn | CALL algo.hashgnn({embeddingDimension: 8, iterations: 2000000000, seed: 1}) YIELD node RETURN node", + "algo.graphsage | CALL algo.graphsage({embeddingDimension: 8, layers: 2000000000, seed: 1}) YIELD node RETURN node" }) + void everyIterationLoopHonoursTheCommandTimeout(final String procedure, final String query) { + database.getConfiguration().setValue(GlobalConfiguration.COMMAND_TIMEOUT, 1L); + + assertThatThrownBy(() -> drain(query)) + .as("%s must give up at the command deadline instead of running the whole knob out", procedure) + .hasStackTraceContaining(procedure + "() exceeded the " + GlobalConfiguration.COMMAND_TIMEOUT.getKey()); + } + + /** + * The checkpoint has to sit inside the per-node scan as well as around the iteration loop, and this is + * the only test that can tell the two apart. + *

+ * {@code maxIterations: 1} means the outer checkpoint runs exactly once, before any work has happened, so it + * cannot be what fires. SimRank is O(n² x deg²) per iteration, so on an 800-node graph of degree 400 a + * single iteration takes half a minute - and with only the outer checkpoint the call runs it to completion and + * returns a similarity, no exception at all. It needs its own database because {@code algo.simRank} loads the + * whole graph. + *

+ * The deadline is 1.5 s rather than 1 ms on purpose: the guard starts its clock before the graph is loaded, and + * a millisecond deadline would already have passed by the time the outer checkpoint runs, making this pass for + * the wrong reason. 1.5 s is comfortably longer than loading 800 nodes (measured at ~0.5 s) and far shorter + * than one iteration. + */ + @Test + @Tag("slow") + @Timeout(300) + void simRankHonoursTheCommandTimeoutInsideASingleIteration() { + final DatabaseFactory factory = new DatabaseFactory("./target/databases/test-issue-6264-simrank-single-pass"); + if (factory.exists()) + factory.open().drop(); + final Database dense = factory.create(); + try { + dense.getSchema().createVertexType("Node"); + dense.getSchema().createEdgeType("LINK"); + + final int nodeCount = 800; + final int degree = 400; + dense.transaction(() -> { + final List nodes = new ArrayList<>(nodeCount); + for (int i = 0; i < nodeCount; i++) + nodes.add(dense.newVertex("Node").set("idx", i).save()); + for (int i = 0; i < nodeCount; i++) + for (int k = 1; k <= degree; k++) + nodes.get(i).newEdge("LINK", nodes.get((i + k) % nodeCount), true, (Object[]) null).save(); + }); + + dense.getConfiguration().setValue(GlobalConfiguration.COMMAND_TIMEOUT, 1_500L); + + final StallAwareStopwatch stopwatch = StallAwareStopwatch.start(); + assertThatThrownBy(() -> { + final ResultSet rs = dense.query("opencypher", "MATCH (a:Node {idx: 0}), (b:Node {idx: 400}) " + + "CALL algo.simRank(a, b, 'LINK', 0.8, 1) YIELD similarity RETURN similarity"); + while (rs.hasNext()) + rs.next(); + }).as("one iteration longer than the deadline must be abortable from inside, not only between iterations") + .hasStackTraceContaining(GlobalConfiguration.COMMAND_TIMEOUT.getKey()); + + // The bound comes from measurement, not taste: the run gives up ~1.5 s after the deadline is armed, and the + // same call with the per-node checkpoint removed grinds through the whole iteration in ~29 s and returns a + // similarity instead of throwing. 8 s sits about 4x above the passing case and 3.5x below the failing one, + // and both figures scale together on a slower runner. If this ever flakes, raise the bound rather than + // dropping the assertion: the exception alone is already meaningful here (an unguarded run does not throw at + // all), but the elapsed time is what says the abort came from inside the pass rather than after it. + stopwatch.assertStayedUnder(8_000L, "the deadline observed inside one iteration, not after it"); + } finally { + dense.drop(); + } + } + + // ── The CSR kernels behind algo.pageRank and algo.labelPropagation ─────── + + /** + * {@code algo.pageRank} hands a CSR-backed graph straight to {@link GraphAlgorithms#pageRank}, which lives below + * the query layer and knew nothing about deadlines. That kernel has no convergence test at all, so it always ran + * the full {@code maxIterations}: the knob alone decided when it stopped, and nothing could interrupt it. + */ + @Test + void thePageRankKernelCallsTheCheckpointOncePerIteration() { + final GraphAnalyticalView gav = GraphAnalyticalView.builder(database) + .withVertexTypes("Node").withEdgeTypes("LINK").build(); + + final AtomicInteger calls = new AtomicInteger(); + GraphAlgorithms.pageRank(gav, 0.85, 7, Vertex.DIRECTION.OUT, calls::incrementAndGet, "LINK"); + + assertThat(calls.get()).as("one checkpoint per power iteration bounds abort latency by one graph sweep") + .isEqualTo(7); + } + + @Test + void thePageRankKernelPropagatesAnAbortFromTheCheckpoint() { + final GraphAnalyticalView gav = GraphAnalyticalView.builder(database) + .withVertexTypes("Node").withEdgeTypes("LINK").build(); + + final AtomicInteger calls = new AtomicInteger(); + final WorkCheckpoint abortOnThird = () -> { + if (calls.incrementAndGet() == 3) + throw new IllegalStateException("aborted by the caller"); + }; + + assertThatThrownBy(() -> GraphAlgorithms.pageRank(gav, 0.85, 1000, Vertex.DIRECTION.OUT, abortOnThird, "LINK")) + .isInstanceOf(IllegalStateException.class).hasMessage("aborted by the caller"); + assertThat(calls.get()).as("the kernel stops at the checkpoint rather than finishing the run").isEqualTo(3); + } + + @Test + void theLabelPropagationKernelPropagatesAnAbortFromTheCheckpoint() { + final GraphAnalyticalView gav = GraphAnalyticalView.builder(database) + .withVertexTypes("Node").withEdgeTypes("LINK").build(); + + final AtomicInteger calls = new AtomicInteger(); + final WorkCheckpoint abortOnSecond = () -> { + if (calls.incrementAndGet() == 2) + throw new IllegalStateException("aborted by the caller"); + }; + + assertThatThrownBy(() -> GraphAlgorithms.labelPropagation(gav, 1000, abortOnSecond, "LINK")) + .isInstanceOf(IllegalStateException.class).hasMessage("aborted by the caller"); + assertThat(calls.get()).isEqualTo(2); + } + + /** + * The same abort end to end, through Cypher, on the CSR path rather than the OLTP one. + *

+ * The query deliberately leaves {@code tolerance} at its default: the OLTP path converges on this graph within a + * handful of iterations and returns, so a run that still has to be aborted proves the CSR kernel - which ignores + * tolerance entirely - is the one that executed. + */ + @Test + @Timeout(120) + void pageRankOnACSRBackedGraphHonoursTheCommandTimeout() { + GraphAnalyticalView.builder(database).withVertexTypes("Node").withEdgeTypes("LINK").build(); + database.getConfiguration().setValue(GlobalConfiguration.COMMAND_TIMEOUT, 1L); + + assertThatThrownBy(() -> drain("CALL algo.pagerank({maxIterations: 2000000000}) YIELD node RETURN node")) + .as("the CSR kernel has no convergence test, so only the checkpoint can end this run") + .hasStackTraceContaining("algo.pagerank() exceeded the " + GlobalConfiguration.COMMAND_TIMEOUT.getKey()); + } + + // ── SLPA: an iteration knob that buys heap as well as time ─────────────── + + /** + * Alone among the fourteen, SLPA's {@code iterations} sizes an allocation: every node keeps a label-memory row + * of {@code iterations + 1} ints, so the matrix is {@code nodeCount x (iterations + 1)} and a value that merely + * looks large reaches the allocator with nothing between it and the heap. + */ + @Test + void slpaRejectsALabelMemoryLargerThanTheBudget() { + // The budget is set rather than left at its default because the default auto-scales with the JVM heap: on a + // large-heap runner a value big enough to exceed it would be one this test then has to allocate to prove + // nothing. 4 nodes x 1000001 ints is 16 MB against a 1 MB budget, and neither is ever reserved. + database.getConfiguration().setValue(GlobalConfiguration.CYPHER_ALGO_MAX_WALK_MEMORY, 1024L * 1024L); + + assertThatThrownBy(() -> drain("CALL algo.slpa({iterations: 1000000}) YIELD node RETURN node")) + .as("a label memory over the budget must be refused before the first row is allocated") + .hasStackTraceContaining("label memory") + .hasStackTraceContaining("iterations=1000000 over 4 nodes") + .hasStackTraceContaining(GlobalConfiguration.CYPHER_ALGO_MAX_WALK_MEMORY.getKey()); + } + + @Test + void slpaRejectsMoreLabelEntriesThanAJavaArrayCanHoldEvenWithTheBudgetDisabled() { + // The budget is what normally catches an oversized matrix, but it explicitly accepts "negative = no limit", + // and `iterations + 1` at Integer.MAX_VALUE wrapped to Integer.MIN_VALUE: a bare NegativeArraySizeException + // naming nothing. The capacity is computed in long and refused on its own account. + database.getConfiguration().setValue(GlobalConfiguration.CYPHER_ALGO_MAX_WALK_MEMORY, -1L); + + assertThatThrownBy(() -> drain("CALL algo.slpa({iterations: 2147483647}) YIELD node RETURN node")) + .hasStackTraceContaining("2147483648 label entries per node, more than the 2147483647 a Java array can hold"); + } + + @Test + void slpaRunsWhenTheLabelMemoryFitsTheBudget() { + // Over-reach guard: the same shape of call, under the budget, must be untouched by the check. + database.getConfiguration().setValue(GlobalConfiguration.CYPHER_ALGO_MAX_WALK_MEMORY, 1024L * 1024L); + + assertThat(drain("CALL algo.slpa({iterations: 50, seed: 3}) YIELD node, communities RETURN node, communities")) + .hasSize(4); + } + + // ── Over-reach: the results themselves are unchanged ───────────────────── + + /** + * The change adds a rejection and a checkpoint to every iterative kernel in the package, so the risk it carries + * is a wrong answer rather than a refused one. PageRank at its defaults is the sharpest available check: on a + * directed cycle every node is symmetric, so all four scores must be equal and sum to 1, which is only true if + * the iteration loop ran unaltered. + */ + @Test + void pageRankStillConvergesToTheRightAnswerWithTheCheckpointInPlace() { + final List results = drain("CALL algo.pagerank() YIELD node, score RETURN node, score"); + + assertThat(results).hasSize(4); + double sum = 0.0; + for (final Result r : results) { + final double score = ((Number) r.getProperty("score")).doubleValue(); + assertThat(score).as("every node of a directed cycle carries the same rank").isCloseTo(0.25, within(1e-6)); + sum += score; + } + assertThat(sum).isCloseTo(1.0, within(1e-6)); + } + + /** + * The one place this PR changes what a kernel computes rather than only when it stops: SimRank's per-iteration + * reset of the n x n similarity matrix became an {@code Arrays.fill} plus the diagonal, instead of an + * element-by-element write of {@code i == j ? 1.0 : 0.0}. The two are equivalent, and this pins the value that + * proves it. + *

+ * The fixture is a hub pointing at two leaves, so the leaves share their only in-neighbour and + * {@code sim(A, B) = decay x sim(hub, hub) = 0.8} - a value the four-node cycle cannot produce, since there + * every SimRank of two distinct nodes is 0 and a broken reset would go unnoticed. + *

+ * The assertion runs at more than one iteration count on purpose. The reset exists only for the diagonal (the + * {@code u < v} loop writes every off-diagonal cell itself), and the buffers are swapped each round, so dropping + * it leaves {@code sim(hub, hub)} at 0 from the second iteration onwards: one iteration still returns 0.8 and + * only three reveals the difference. + */ + @Test + void simRankStillComputesTheSameSimilarityAfterTheMatrixResetRefactor() { + final DatabaseFactory factory = new DatabaseFactory("./target/databases/test-issue-6264-simrank-shared-parent"); + if (factory.exists()) + factory.open().drop(); + final Database shared = factory.create(); + try { + shared.getSchema().createVertexType("Node"); + shared.getSchema().createEdgeType("LINK"); + shared.transaction(() -> { + final MutableVertex hub = shared.newVertex("Node").set("name", "H").save(); + final MutableVertex a = shared.newVertex("Node").set("name", "A").save(); + final MutableVertex b = shared.newVertex("Node").set("name", "B").save(); + hub.newEdge("LINK", a, true, (Object[]) null).save(); + hub.newEdge("LINK", b, true, (Object[]) null).save(); + }); + + for (final int iterations : new int[] { 1, 3 }) { + final ResultSet rs = shared.query("opencypher", "MATCH (a:Node {name: 'A'}), (b:Node {name: 'B'}) " + + "CALL algo.simRank(a, b, 'LINK', 0.8, " + iterations + ") YIELD similarity RETURN similarity"); + assertThat(rs.hasNext()).isTrue(); + assertThat(((Number) rs.next().getProperty("similarity")).doubleValue()) + .as("two nodes sharing their only in-neighbour are decay-similar, at %d iterations", iterations) + .isCloseTo(0.8, within(1e-9)); + } + } finally { + shared.drop(); + } + } + + // ── Helpers ────────────────────────────────────────────────────────────── + + private List drain(final String query) { + final ResultSet rs = database.query("opencypher", query); + final List results = new ArrayList<>(); + while (rs.hasNext()) + results.add(rs.next()); + return results; + } +}