Merge branch '3.8-dev'

Author: andreachild

docs/src/recipes/appendix.asciidoc

@@ -311,7 +311,7 @@ g.addV('person').property('name','alice').as('alice').
       times(3).emit().
     select(all, "x").
     project("name","score").
-      by(tail(local, 1).select("a").values("name")).
+      by(tail(local, 1).unfold().select("a").values("name")).
       by(unfold().
          sack(assign).by(select("b")).
          sack(mult).by(select("c")).

docs/src/recipes/recommendation.asciidoc

@@ -199,7 +199,7 @@ g.V().has("person","name","alice").as("alice").
       where(within("self")).count()).as("g").
       select(values).
       order(local).
-        by(desc).limit(local, 1).as("m").
+        by(desc).limit(local, 1).unfold().as("m").
       select("g").unfold().
       where(select(values).as("m")).select(keys)
 ----
@@ -217,7 +217,7 @@ g.V().has("person","name","alice").as("alice").
       where(within("self")).count()).as("g").
       select(values).
       order(local).
-        by(desc).limit(local, 1).as("m").
+        by(desc).limit(local, 1).unfold().as("m").
       select("g").unfold().
       where(select(values).as("m")).select(keys).
       out("bought").where(without("self"))
@@ -235,7 +235,7 @@ g.V().has("person","name","alice").as("alice").
       where(within("self")).count()).as("g").
       select(values).
       order(local).
-        by(desc).limit(local, 1).as("m").
+        by(desc).limit(local, 1).unfold().as("m").
       select("g").unfold().
       where(select(values).as("m")).select(keys).
       out("bought").where(without("self")).

docs/src/recipes/shortest-path.asciidoc

@@ -187,7 +187,7 @@ g.withSideEffect('v', []).                            <1>
   unfold().
   map(unfold().as('v_or_e').
       coalesce(V().where(eq('v_or_e')).aggregate(local,'v'),
-               select('v').tail(local, 1).bothE().where(eq('v_or_e'))).
+               select('v').tail(local, 1).unfold().bothE().where(eq('v_or_e'))).
       values('name','weight').
       fold()).
   group().

docs/src/reference/the-traversal.asciidoc

@@ -3315,13 +3315,13 @@ contains additional information that will be applied if an existing vertex is fo
 ----
 g.inject([[(T.id):400],[(T.label):'Dog',name:'Pixel',age:1],[updated:'2022-02-1']]).as('m').
   mergeV(select('m').limit(local,1).unfold()). <1>
-  option(Merge.onCreate, select('m').range(local,1,2)). <2>
-  option(Merge.onMatch, select('m').tail(local)) <3>
+  option(Merge.onCreate, select('m').range(local,1,2).unfold()). <2>
+  option(Merge.onMatch, select('m').tail(local).unfold()) <3>
 g.V(400).valueMap().with(WithOptions.tokens)
 g.inject([[(T.id):400],[(T.label):'Dog',name:'Pixel',age:1],[updated:'2022-02-1']]).as('m').
   mergeV(select('m').limit(local,1).unfold()).
-  option(Merge.onCreate, select('m').range(local,1,2)).
-  option(Merge.onMatch, select('m').tail(local))  <4>
+  option(Merge.onCreate, select('m').range(local,1,2).unfold()).
+  option(Merge.onMatch, select('m').tail(local).unfold())  <4>
 g.V(400).valueMap().with(WithOptions.tokens)  <5>
 ----
 

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

@@ -46,9 +46,18 @@ public final class RangeGlobalStep<S> extends FilterStep<S> implements RangeGlob
     private long low;
     private long high;
     /**
-     * If this range step is used inside a loop there can be multiple counters, otherwise there should only be one
+     * Flag to indicate if the step is inside a repeat loop. Can be null if the value has not been initialized yet as 
+     * the traversal has not been finalized.
      */
-    private Map<String, AtomicLong> counters = new HashMap<>();
+    private Boolean insideLoop;
+    /**
+     * Single counter if this range step is not inside a loop
+     */
+    private AtomicLong singleCounter = new AtomicLong(0);
+    /**
+     * If this range step is used inside a loop there can be multiple loop counters
+     */
+    private Map<String, AtomicLong> loopCounters = new HashMap<>();
     private boolean bypass;
 
     public RangeGlobalStep(final Traversal.Admin traversal, final long low, final long high) {
@@ -64,11 +73,10 @@ public RangeGlobalStep(final Traversal.Admin traversal, final long low, final lo
     protected boolean filter(final Traverser.Admin<S> traverser) {
         if (this.bypass) return true;
 
-        final String counterKey = getCounterKey(traverser);
-        final AtomicLong counter = counters.computeIfAbsent(counterKey, k -> new AtomicLong(0L));
+        final AtomicLong counter = getCounter(traverser);
 
         if (this.high != -1 && counter.get() >= this.high) {
-            if (hasRepeatStepParent()) {
+            if (isInsideLoop()) {
                 return false;
             }
             throw FastNoSuchElementException.instance();
@@ -103,7 +111,8 @@ protected boolean filter(final Traverser.Admin<S> traverser) {
     @Override
     public void reset() {
         super.reset();
-        this.counters.clear();
+        this.singleCounter.set(0);
+        this.loopCounters.clear();
     }
 
     @Override
@@ -124,7 +133,8 @@ public Long getHighRange() {
     @Override
     public RangeGlobalStep<S> clone() {
         final RangeGlobalStep<S> clone = (RangeGlobalStep<S>) super.clone();
-        clone.counters = new HashMap<>();
+        clone.singleCounter = new AtomicLong(0);
+        clone.loopCounters = new HashMap<>();
         return clone;
     }
 
@@ -157,10 +167,32 @@ public void processAllStarts() {
 
     }
 
+    private AtomicLong getCounter(final Traverser.Admin<S> traverser) {
+        if (isInsideLoop()) {
+            final String counterKey = getCounterKey(traverser);
+            return loopCounters.computeIfAbsent(counterKey, k -> new AtomicLong(0L));
+        } else {
+            return this.singleCounter;
+        }
+    }
+
+    /**
+     * This will initialize the insideLoop flag if it hasn't been set by analyzing the traversal up to the root and 
+     * should only be called after the traversal has been finalized.
+     * 
+     * @return if the step is being used inside a repeat loop.
+     */
+    private boolean isInsideLoop() {
+        if (this.insideLoop == null) {
+            this.insideLoop = hasRepeatStepParent();
+        }
+        return this.insideLoop;
+    }
+
     private String getCounterKey(final Traverser.Admin<S> traverser) {
         final List<String> counterKeyParts = new ArrayList<>();
         Traversal.Admin<Object, Object> traversal = this.getTraversal();
-        if (hasRepeatStepParent()) {
+        if (isInsideLoop()) {
             // 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()) {

gremlin-tools/gremlin-benchmark/src/main/java/org/apache/tinkerpop/gremlin/process/RangeTraversalBenchmark.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.tinkerpop.gremlin.process;
+
+import java.util.List;
+import org.apache.tinkerpop.benchmark.util.AbstractGraphBenchmark;
+import org.apache.tinkerpop.gremlin.LoadGraphWith;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.openjdk.jmh.annotations.Benchmark;
+
+@LoadGraphWith(LoadGraphWith.GraphData.GRATEFUL)
+public class RangeTraversalBenchmark extends AbstractGraphBenchmark {
+
+    @Override
+    protected int getWarmupIterations() {
+        return 1;
+    }
+
+    @Override
+    protected int getForks() {
+        return 1;
+    }
+
+    @Benchmark
+    public List<Vertex> limit() {
+        return g.V().hasLabel("artist").limit(10).toList();
+    }
+
+    @Benchmark
+    public List<Vertex> range() {
+        return g.V().hasLabel("artist").range(5, 20).toList();
+    }
+
+    @Benchmark
+    public List<Vertex> skip() {
+        return g.V().hasLabel("artist").skip(5).toList();
+    }
+}