fix(#6264): fourteen more algo.* iteration knobs get the minimum and the checkpoint #6216 gave three

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

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

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.
+   * <p>
+   * 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.
+   * <p>
+   * 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.
+   * <p>
+   * 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);

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.
+ * <p>
+ * 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.
+ * </p>
+ * <p>
+ * The hook exists rather than a direct dependency on the query layer's guard because {@code com.arcadedb.graph.olap}
+ * sits <em>below</em> {@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.
+ * </p>
+ *
+ * @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();
+}

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.
    * <p>
    * 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}.
-   * <p>
-   * 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}.
+   * <p>
+   * 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.
+   * <p>
+   * 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 <em>negative</em> 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.
+   * </p>
+   */
+  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.
    * <p>