[TINKERPOP-3202] Improve performance of RangeGlobalStep (#3251)

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

Follow up to #3241 to improve performance of RangeGlobalStep by using single counter if not inside repeat and changing parent step check for RepeatStep to occur just once instead of for each call to filter. This change is for performance only and does not affect any semantics.

Author: andreachild

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();
+    }
+}