Merge branch '3.8-dev'

Author: andreachild

CHANGELOG.asciidoc

@@ -183,6 +183,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

@@ -4079,10 +4079,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

@@ -236,7 +236,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,