Change range in repeat to track counters per iteration (#3241)

https://issues.apache.org/jira/browse/TINKERPOP-3202

This changeset aligns the semantics of traversals with range() and limit() used in repeat() with the behaviour of the equivalent 'unrolled' traversal without repeat(). This is done by tracking per-iteration counters instead of a global counter. The semantics of range() and limit() without repeat() and with local() were not modified.

Single and nested loop Traverser implementations were also refactored to reduce much code duplication by consolidating common loop logic into a new NL_SL_Traverser base interface for loop-aware traversers.

Author: andreachild

CHANGELOG.asciidoc

@@ -117,6 +117,8 @@ This release also includes changes from <<release-3-7-XXX, 3.7.XXX>>.
 * Renamed `MergeElementStep` to `MergeElementStep` as it is a base class to `mergeV()` and `mergeE()`.
 * Renamed `MergeStep` of `merge()` to `MergeElementStep` for consistency.
 * Modified `RepeatUnrollStrategy` to use a more conservative approach, only unrolling repeat loops containing safe navigation and filtering steps.
+* Modified `limit()`, `skip()`, range()` in `repeat()` to track per-iteration counters.
+* Moved `Traverser` loop logic into new interface `NL_SL_Traverser` and changed loop-supporting `Traverser`s to extend the common interface.
 
 == TinkerPop 3.7.0 (Gremfir Master of the Pan Flute)
 

docs/src/reference/the-traversal.asciidoc

@@ -4075,10 +4075,23 @@ g.V().until(has('name','ripple')).
 <1> do-while semantics stating to do `out()` 2 times.
 <2> while-do semantics stating to break if the traverser is at a vertex named "ripple".
 
-IMPORTANT: There are two modulators for `repeat()`: `until()` and `emit()`. If `until()` comes after `repeat()` it is
-do/while looping. If `until()` comes before `repeat()` it is while/do looping. If `emit()` is placed after `repeat()`,
-it is evaluated on the traversers leaving the repeat-traversal. If `emit()` is placed before `repeat()`, it is
-evaluated on the traversers prior to entering the repeat-traversal.
+[gremlin-groovy,grateful-dead]
+----
+g.V().has('name','JAM').repeat(out('followedBy').limit(2)).times(3) <1>
+g.V().has('name','DRUMS').repeat(__.in('followedBy').range(1,3)).until(loops().is(2)) <2>
+g.V().has('name','HEY BO DIDDLEY').repeat(out('followedBy').skip(5)).times(2) <3>
+----
+
+<1> Starting from the song 'JAM' get 2 songs that have followed, looping 3 times.
+<2> Starting from the song 'DRUMS' get the 2nd and 3rd songs that have preceded, looping twice.
+<3> Starting from the song 'HEY BO DIDDLEY' get the songs that have followed, skipping the first 5 and looping twice.
+
+IMPORTANT: There are three modulators for `repeat()`: `times()`, `until()`, and `emit()`. The most straightforward is 
+`times()`, which indicates the number of times to execute the loop. Conditional loops can be executed using `until()`. 
+If `until()` comes after `repeat()` it is do/while looping. If `until()` comes before `repeat()` it is while/do looping. 
+Emission of traversers from the loop are controlled with `emit()`. If `emit()` is placed after `repeat()`, it is evaluated 
+on the traversers leaving the repeat-traversal. If `emit()` is placed before `repeat()`, it is evaluated on the 
+traversers prior to entering the repeat-traversal.
 
 The `repeat()`-step also supports an "emit predicate", where the predicate for an empty argument `emit()` is
 `true` (i.e. `emit() == emit{true}`). With `emit()`, the traverser is split in two -- the traverser exits the code

docs/src/upgrade/release-3.8.x.asciidoc

@@ -901,15 +901,28 @@ impact.
 
 ===== Examples of Affected Traversals =====
 
-Usage of `limit()` inside `repeat()` is an example of a traversal that will no longer be unrolled. The following results 
-returned from the `modern` graph demonstrate the change of semantics if the `limit()` in `repeat()` were to be unrolled:
+Usage of `inject()` inside `repeat()` is an example of a traversal that will no longer be unrolled. The following results 
+returned from the `modern` graph demonstrate the change of semantics if the `inject()` in `repeat()` were to be unrolled:
 
 [source,text]
 ----
-gremlin> g.V().both().limit(1).both().limit(1)
-==>v[1]
-gremlin> g.withoutStrategies(RepeatUnrollStrategy).V().repeat(both().limit()).times(2)
-gremlin>
+gremlin> g.V().values('name').inject('foo').inject('foo')
+==>foo
+==>foo
+==>marko
+==>vadas
+==>lop
+==>josh
+==>ripple
+==>peter
+gremlin> g.withoutStrategies(RepeatUnrollStrategy).V().values('name').repeat(inject('foo')).times(2)
+==>foo
+==>marko
+==>vadas
+==>lop
+==>josh
+==>ripple
+==>peter
 ----
 
 Another example is the usage of `aggregate()` inside `repeat()`. The following results returned from the `modern` graph 
@@ -973,6 +986,51 @@ g.V().repeat(both()).times(2).dedup()
 
 See: link:https://issues.apache.org/jira/browse/TINKERPOP-3192[TINKERPOP-3192]
 
+==== Modified limit() skip() range() Semantics in repeat()
+
+The semantics of `limit()`, `skip()`, and `range()` steps called with default `Scope` or explicit `Scope.global` inside 
+`repeat()` have been modified to ensure consistent semantics across repeat iterations. Previously, these steps would 
+track global state across iterations, leading to unexpected filtering behavior between loops.
+
+Consider the following examples which demonstrate the unexpected behavior. Note that the examples for version 3.7.4
+disable the `RepeatUnrollStrategy` so that strategy optimization does not replace the `repeat()` traversal with a 
+non-looping equivalent. 3.8.0 examples do not disable the `RepeatUnrollStrategy` as the strategy was modified to be more
+restrictive in this version.
+
+[source,groovy]
+----
+// 3.7.4 - grateful dead graph examples producing no results due to global counters
+gremlin> g.withoutStrategies(RepeatUnrollStrategy).V().has('name','JAM').repeat(out('followedBy').limit(2)).times(2).values('name')
+gremlin>
+gremlin> g.withoutStrategies(RepeatUnrollStrategy).V().has('name','DRUMS').repeat(__.in('followedBy').range(1,3)).times(2).values('name')
+gremlin>
+// 3.7.4 - modern graph examples demonstrating too many results with skip in repeat due to global counters
+gremlin> g.withoutStrategies(RepeatUnrollStrategy).V(1).repeat(out().skip(1)).times(2).values('name')
+==>ripple
+==>lop
+gremlin> g.withoutStrategies(RepeatUnrollStrategy).V(1).out().skip(1).out().skip(1).values('name')
+==>lop
+
+// 3.8.0 - grateful dead graph examples producing results as limit counters tracked per iteration
+gremlin> g.V().has('name','JAM').repeat(out('followedBy').limit(2)).times(2).values('name')
+==>HURTS ME TOO
+==>BLACK THROATED WIND
+gremlin> g.V().has('name','DRUMS').repeat(__.in('followedBy').range(1,3)).times(2).values('name')
+==>DEAL
+==>WOMEN ARE SMARTER
+// 3.8.0 - modern graph examples demonstrating consistent skip semantics
+gremlin> g.V(1).repeat(out().skip(1)).times(2).values('name')
+==>lop
+gremlin> g.V(1).out().skip(1).out().skip(1).values('name')
+==>lop
+----
+
+This change ensures that `limit()`, `skip()`, and `range()` steps called with default `Scope` or explicit `Scope.global` 
+inside `repeat()` are more consistent with manually unrolled traversals. Before upgrading, users should determine if any 
+traversals use `limit()`, skip()`, or `range()` with default `Scope` or explicit `Scope.global` inside `repeat()`. If it 
+is desired that the limit or range should apply across all loops then the `limit()`, `skip()`, or `range()` step should 
+be moved out of the `repeat()` step.
+
 === Upgrading for Providers
 
 ==== Graph System Providers

gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/Traverser.java

@@ -23,8 +23,6 @@
 
 import java.io.Serializable;
 import java.util.Collections;
-import java.util.Comparator;
-import java.util.Optional;
 import java.util.Set;
 import java.util.function.Function;
 
@@ -90,7 +88,9 @@ public default <A> A path(final Pop pop, final String stepLabel) {
      *
      * @return The number of times the traverser has gone through a loop
      */
-    public int loops();
+    default int loops() {
+        throw new UnsupportedOperationException("This traverser does not support loops: " + this.getClass().getCanonicalName());
+    }
 
     /**
      * Return the number of times the traverser has gone through the named looping section of a traversal.
@@ -99,7 +99,9 @@ public default <A> A path(final Pop pop, final String stepLabel) {
      * @return The number of times the traverser has gone through a loop
      * @throws IllegalArgumentException if the loopName is not defined
      */
-    public int loops(final String loopName);
+    default int loops(final String loopName) {
+        throw new UnsupportedOperationException("This traverser does not support loops: " + this.getClass().getCanonicalName());
+    }
 
     /**
      * A traverser may represent a grouping of traversers to allow for more efficient data propagation.
@@ -239,19 +241,22 @@ public interface Admin<T> extends Traverser<T>, Attachable<T> {
          * @param stepLabel the label of the step that is being set-up.
          * @param loopName the user defined name for referencing the loop counter or null if not set
          */
-        public void initialiseLoops(final String stepLabel, final String loopName);
+        default void initialiseLoops(final String stepLabel, final String loopName) {
+        }
 
         /**
          * Increment the number of times the traverser has gone through a looping section of traversal.
          */
-        public void incrLoops();
+        default void incrLoops() {
+        }
 
         /**
          * Set the number of times the traverser has gone through a loop back to 0.
          * When a traverser exits a looping construct, this method should be called.
          * In a nested loop context, the highest stack loop counter should be removed.
          */
-        public void resetLoops();
+        default void resetLoops() {
+        }
 
         /**
          * Get the step id of where the traverser is located.

gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/branch/RepeatStep.java

@@ -237,7 +237,7 @@ protected Iterator<Traverser.Admin<S>> computerAlgorithm() throws NoSuchElementE
             return IteratorUtils.of(start);
         } else {
             start.setStepId(this.repeatTraversal.getStartStep().getId());
-            start.initialiseLoops(start.getStepId(), this.loopName);
+            start.initialiseLoops(this.getId(), this.getLoopName());
             if (doEmit(start, true)) {
                 final Traverser.Admin<S> emitSplit = start.split();
                 emitSplit.resetLoops();

gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/filter/RangeGlobalStep.java

@@ -18,19 +18,25 @@
  */
 package org.apache.tinkerpop.gremlin.process.traversal.step.filter;
 
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BinaryOperator;
+import org.apache.tinkerpop.gremlin.process.traversal.Step;
 import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
 import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
+import org.apache.tinkerpop.gremlin.process.traversal.step.TraversalParent;
+import org.apache.tinkerpop.gremlin.process.traversal.step.branch.RepeatStep;
 import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequirement;
 import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
 import org.apache.tinkerpop.gremlin.process.traversal.util.FastNoSuchElementException;
 import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
 
-import java.io.Serializable;
-import java.util.Collections;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.BinaryOperator;
-
 /**
  * @author Bob Briody (http://bobbriody.com)
  * @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -39,7 +45,10 @@ public final class RangeGlobalStep<S> extends FilterStep<S> implements RangeGlob
 
     private long low;
     private long high;
-    private AtomicLong counter = new AtomicLong(0l);
+    /**
+     * If this range step is used inside a loop there can be multiple counters, otherwise there should only be one
+     */
+    private Map<String, AtomicLong> counters = new HashMap<>();
     private boolean bypass;
 
     public RangeGlobalStep(final Traversal.Admin traversal, final long low, final long high) {
@@ -55,31 +64,37 @@ public RangeGlobalStep(final Traversal.Admin traversal, final long low, final lo
     protected boolean filter(final Traverser.Admin<S> traverser) {
         if (this.bypass) return true;
 
-        if (this.high != -1 && this.counter.get() >= this.high) {
+        final String counterKey = getCounterKey(traverser);
+        final AtomicLong counter = counters.computeIfAbsent(counterKey, k -> new AtomicLong(0L));
+
+        if (this.high != -1 && counter.get() >= this.high) {
+            if (hasRepeatStepParent()) {
+                return false;
+            }
             throw FastNoSuchElementException.instance();
         }
 
         long avail = traverser.bulk();
-        if (this.counter.get() + avail <= this.low) {
+        if (counter.get() + avail <= this.low) {
             // Will not surpass the low w/ this traverser. Skip and filter the whole thing.
-            this.counter.getAndAdd(avail);
+            counter.getAndAdd(avail);
             return false;
         }
 
         // Skip for the low and trim for the high. Both can happen at once.
 
         long toSkip = 0;
-        if (this.counter.get() < this.low) {
-            toSkip = this.low - this.counter.get();
+        if (counter.get() < this.low) {
+            toSkip = this.low - counter.get();
         }
 
         long toTrim = 0;
-        if (this.high != -1 && this.counter.get() + avail >= this.high) {
-            toTrim = this.counter.get() + avail - this.high;
+        if (this.high != -1 && counter.get() + avail >= this.high) {
+            toTrim = counter.get() + avail - this.high;
         }
 
         long toEmit = avail - toSkip - toTrim;
-        this.counter.getAndAdd(toSkip + toEmit);
+        counter.getAndAdd(toSkip + toEmit);
         traverser.setBulk(toEmit);
 
         return true;
@@ -88,7 +103,7 @@ protected boolean filter(final Traverser.Admin<S> traverser) {
     @Override
     public void reset() {
         super.reset();
-        this.counter.set(0l);
+        this.counters.clear();
     }
 
     @Override
@@ -109,7 +124,7 @@ public Long getHighRange() {
     @Override
     public RangeGlobalStep<S> clone() {
         final RangeGlobalStep<S> clone = (RangeGlobalStep<S>) super.clone();
-        clone.counter = new AtomicLong(0l);
+        clone.counters = new HashMap<>();
         return clone;
     }
 
@@ -142,6 +157,43 @@ public void processAllStarts() {
 
     }
 
+    private String getCounterKey(final Traverser.Admin<S> traverser) {
+        final List<String> counterKeyParts = new ArrayList<>();
+        Traversal.Admin<Object, Object> traversal = this.getTraversal();
+        if (hasRepeatStepParent()) {
+            // the range step is inside a loop so we need to track counters per iteration
+            // using a counter key that is composed of the parent steps to the root
+            while (!traversal.isRoot()) {
+                final TraversalParent pt = traversal.getParent();
+                final Step<?, ?> ps = pt.asStep();
+                final String pid = ps.getId();
+                if (traverser.getLoopNames().contains(pid)) {
+                    counterKeyParts.add(pid);
+                    counterKeyParts.add(String.valueOf(traverser.loops(pid)));
+                }
+                traversal = ps.getTraversal();
+            }
+        }
+        // reverse added parts so that it starts from root
+        Collections.reverse(counterKeyParts);
+        counterKeyParts.add(this.getId());
+        if (traverser.getLoopNames().contains(this.getId())) {
+            counterKeyParts.add(String.valueOf(traverser.loops(this.getId())));
+        }
+        return String.join(":", counterKeyParts);
+    }
+
+    private boolean hasRepeatStepParent() {
+        Traversal.Admin<?, ?> traversal = this.getTraversal();
+        while (!traversal.isRoot()) {
+            if (traversal.getParent() instanceof RepeatStep) {
+                return true;
+            }
+            traversal = traversal.getParent().asStep().getTraversal();
+        }
+        return false;
+    }
+
     ////////////////
 
     public static final class RangeBiOperator<S> implements BinaryOperator<TraverserSet<S>>, Serializable {

gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/strategy/optimization/RepeatUnrollStrategy.java

@@ -63,6 +63,10 @@ public final class RepeatUnrollStrategy extends AbstractTraversalStrategy<Traver
 
     private static final RepeatUnrollStrategy INSTANCE = new RepeatUnrollStrategy();
     static final int MAX_BARRIER_SIZE = 2500;
+    /**
+     * When adding more 'allowed' steps, consider if tests are needed that use the step 'withoutStrategies(RepeatUnrollStrategy)' 
+     * to validate the traversal with repeat has the same semantics as the unrolled equivalent
+     */
     private static final Set<Class> ALLOWED_STEP_CLASSES = Set.of(
             VertexStepContract.class,
             EdgeVertexStep.class,