TINKERPOP-3196 Split bulked traversers for LocalStep

Local was behaving as a per traverser traversal when it should be a per
object traversal. This means that there should be no bulked traversers
that are added as starts to local traversals.

Co-authored-by: Cole-Greer <[email protected]>

Author: kenhuuu

CHANGELOG.asciidoc

@@ -35,6 +35,7 @@ This release also includes changes from <<release-3-7-XXX, 3.7.XXX>>.
 * Renamed `none()` step to `discard()`.
 * Repurposed `none()` step as a list filtering step with the signature `none(P)`.
 * Modified mathematical operators to prevent overflows in steps such as `sum()` and 'sack()' to prefer promotion to the next highest number type.
+* Modified `local()` to be "object-local" rather than "traverser-local".
 * Added `DateTime` ontop of the existing 'datetime' grammar.
 * Added `UUID()` and `UUID(value)` to grammar.
 * Deprecated the `UnifiedChannelizer`.

docs/src/dev/provider/gremlin-semantics.asciidoc

@@ -1589,6 +1589,36 @@ See: link:https://github.com/apache/tinkerpop/tree/x.y.z/gremlin-core/src/main/j
 link:https://github.com/apache/tinkerpop/tree/x.y.z/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/LengthLocalStep.java[source (local)],
 link:https://tinkerpop.apache.org/docs/x.y.z/reference/#length-step[reference]
 
+[[local-step]]
+=== local()
+
+*Description:* Executes the provided traversal in an object-local manner.
+
+*Syntax:* `local(Traversal localTraversal)`
+
+[width="100%",options="header"]
+|=========================================================
+|Start Step |Mid Step |Modulated |Domain |Range
+|N |Y |N |`any` |`any`
+|=========================================================
+
+*Arguments:*
+
+* `localTraversal` - The traversal that processes each single-object traverser individually.
+
+*Modulation:*
+
+None
+
+*Considerations:*
+
+The `local()` step enforces object-local execution. As a branching step with local children, it implements strict lazy
+evaluation by passing a single traverser at a time to the local traversal (bulk of exactly one, if bulking is supported)
+and resetting the traversal to clean state between executions.
+
+See: link:https://github.com/apache/tinkerpop/tree/x.y.z/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/LocalStep.java[source],
+link:https://tinkerpop.apache.org/docs/x.y.z/reference/#local-step[reference]
+
 [[intersect-step]]
 === intersect()
 

docs/src/reference/the-traversal.asciidoc

@@ -2557,30 +2557,15 @@ in an object-local traversal. As such, the `order().by()` and the `limit()` refe
 stream as a whole.
 
 Local Step is quite similar in functionality to <<general-steps,Flat Map Step>> where it can often be confused.
-`local()` propagates the traverser through the internal traversal as is without splitting/cloning it. Thus, its
-a “global traversal” with local processing. Its use is subtle and primarily finds application in compilation
-optimizations (i.e. when writing `TraversalStrategy` implementations. As another example consider:
+The primary distinction between these steps is that while `local()` preserves the path history of traversers as they
+pass through its child traversal, `flatMap()` does not. As another example consider:
 
 [gremlin-groovy,modern]
 ----
-g.V().both().barrier().flatMap(groupCount().by("name"))
-g.V().both().barrier().local(groupCount().by("name"))
+g.V().local(outE().inV()).path()
+g.V().flatMap(outE().inV()).path()
 ----
 
-Use of `local()` is often a mistake. This is especially true when its argument contains a reducing step. For example,
-let's say the requirement was to count the number of properties per `Vertex` in:
-
-[gremlin-groovy,modern]
-----
-g.V().both().local(properties('name','age').count())  <1>
-g.V().both().map(properties('name','age').count()) <2>
-----
-
-<1> The output here seems impossible because no single vertex in the "modern" graph can have more than two properties
-given the "name" and "age" filters, but because the counting is happening object-local the counting is occurring unique
-to each object rather than each global traverser.
-<2> Replacing `local()` with `map()` returns the result desired by the requirement.
-
 WARNING: The anonymous traversal of `local()` processes the current object "locally." In OLAP, where the atomic unit
 of computing is the vertex and its local "star graph," it is important that the anonymous traversal does not leave
 the confines of the vertex's star graph. In other words, it can not traverse to an adjacent vertex's properties or edges.

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

@@ -425,6 +425,34 @@ compatibility.
 
 See: link:https://issues.apache.org/jira/browse/TINKERPOP-3161[TINKERPOP-3161]
 
+==== Split bulked traversers for `local()`
+
+Prior to 3.8.0, local() exhibited "traverser-local" semantics, where the local traversal would apply independently to
+each individual `Traverser`. This often led to confusion, especially in the presence of reducing barrier steps, as
+bulked traversers would cause multiple objects to be processed at once. local() has been updated to automatically split
+any bulked traversers and thus now exhibits true "object-local" semantics.
+
+[source,groovy]
+----
+// 3.7.4
+gremlin> g.V().out().barrier().local(count())
+==>3
+==>1
+==>1
+==>1
+
+// 3.8.0
+gremlin> g.V().out().barrier().local(count())
+==>1
+==>1
+==>1
+==>1
+==>1
+==>1
+----
+
+See: link:https://issues.apache.org/jira/browse/TINKERPOP-3196[TINKERPOP-3196]
+
 ==== Removal of P.getOriginalValue()
 
 `P.getOriginalValue()` has been removed as it was not offering much value and was often confused with `P.getValue()`.

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

@@ -23,6 +23,7 @@
 import org.apache.tinkerpop.gremlin.process.traversal.step.TraversalParent;
 import org.apache.tinkerpop.gremlin.process.traversal.step.util.AbstractStep;
 import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequirement;
+import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.EmptyTraverser;
 import org.apache.tinkerpop.gremlin.process.traversal.util.FastNoSuchElementException;
 import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
 
@@ -38,6 +39,7 @@ public final class LocalStep<S, E> extends AbstractStep<S, E> implements Travers
 
     private Traversal.Admin<S, E> localTraversal;
     private boolean first = true;
+    private Traverser.Admin<S> currentStart = EmptyTraverser.instance();
 
     public LocalStep(final Traversal.Admin traversal, final Traversal.Admin<S, E> localTraversal) {
         super(traversal);
@@ -58,20 +60,34 @@ public Set<TraverserRequirement> getRequirements() {
     protected Traverser.Admin<E> processNextStart() throws NoSuchElementException {
         if (this.first) {
             this.first = false;
-            this.localTraversal.addStart(this.starts.next());
+            this.localTraversal.addStart(nextStart());
         }
         while (true) {
             if (this.localTraversal.hasNext())
                 return this.localTraversal.nextTraverser();
-            else if (this.starts.hasNext()) {
+            else if (hasStartRemaining()) {
                 this.localTraversal.reset();
-                this.localTraversal.addStart(this.starts.next());
+                this.localTraversal.addStart(nextStart());
             } else {
                 throw FastNoSuchElementException.instance();
             }
         }
     }
 
+    private boolean hasStartRemaining() {
+        return (currentStart.bulk() > 0L) || this.starts.hasNext();
+    }
+
+    private Traverser.Admin<S> nextStart() throws NoSuchElementException {
+        if (currentStart.bulk() == 0L) {
+            currentStart = starts.next();
+        }
+        final Traverser.Admin<S> split = currentStart.split();
+        split.setBulk(1L);
+        currentStart.setBulk(currentStart.bulk() - 1L);
+        return split;
+    }
+
     @Override
     public void reset() {
         super.reset();

gremlin-test/src/main/resources/org/apache/tinkerpop/gremlin/test/features/branch/Choose.feature

@@ -337,10 +337,14 @@ Feature: Step - choose()
     When iterated to list
     Then the result should be unordered
       | result |
-      | l[marko,marko] |
-      | l[vadas,vadas] |
-      | l[josh,josh] |
-      | l[peter,peter] |
+      | l[marko] |
+      | l[marko] |
+      | l[vadas] |
+      | l[vadas] |
+      | l[josh] |
+      | l[josh] |
+      | l[peter] |
+      | l[peter] |
 
   @GraphComputerVerificationMidVNotSupported
   Scenario: g_unionXV_VXhasLabelXpersonX_barrier_mapXchooseXageX_optionXbetweenX26_30X_name_foldX_optionXnone_name_foldXX

gremlin-test/src/main/resources/org/apache/tinkerpop/gremlin/test/features/branch/Local.feature

@@ -183,4 +183,91 @@ Feature: Step - local()
       | m[{"name":"marko","project":"lop"}]   |
       | m[{"name":"josh","project":"lop"}]    |
       | m[{"name":"peter","project":"lop"}]   |
-      | m[{"name":"josh","project":"ripple"}] |
+      | m[{"name":"josh","project":"ripple"}] |
+
+  # Barrier should show that bulked traversers don't affect local() traversal
+  Scenario: g_V_in_barrier_localXcountX
+    Given the modern graph
+    And the traversal of
+      """
+      g.V().in().barrier().local(__.count())
+      """
+    When iterated to list
+    Then the result should be unordered
+      | result |
+      | d[1].l |
+      | d[1].l |
+      | d[1].l |
+      | d[1].l |
+      | d[1].l |
+      | d[1].l |
+
+  # Path of traversers isn't hidden in local
+  @GraphComputerVerificationStarGraphExceeded
+  Scenario: g_V_localXout_in_simplePathX_path
+    Given the modern graph
+    And the traversal of
+      """
+      g.V().local(__.out().in().simplePath()).path()
+      """
+    When iterated to list
+    Then the result should be unordered
+      | result |
+      | p[v[josh],v[lop],v[marko]] |
+      | p[v[josh],v[lop],v[peter]] |
+      | p[v[marko],v[lop],v[josh]] |
+      | p[v[marko],v[lop],v[peter]] |
+      | p[v[peter],v[lop],v[marko]] |
+      | p[v[peter],v[lop],v[josh]] |
+  
+  # Traverser's sack value should be carried over from local traversal
+  Scenario: g_withSackX0LX_V_in_barrier_localXsackXsumX_byXageXX_sack
+    Given the modern graph
+    And the traversal of
+      """
+      g.withSack(0L).V().in().barrier().local(__.sack(sum).by("age")).sack()
+      """
+    When iterated to list
+    Then the result should be unordered
+      | result |
+      | d[29].l |
+      | d[29].l |
+      | d[29].l |
+      | d[32].l |
+      | d[32].l |
+      | d[35].l |
+
+  # Nested local should return proper local count
+  Scenario: g_V_localXout_localXcountXX
+    Given the modern graph
+    And the traversal of
+      """
+      g.V().local(__.out().local(__.count()))
+      """
+    When iterated to list
+    Then the result should be unordered
+      | result |
+      | d[1].l |
+      | d[1].l |
+      | d[1].l |
+      | d[1].l |
+      | d[1].l |
+      | d[1].l |
+
+  # Local should be applied to union's global child
+  Scenario: g_V_unionXoutE_count_localXinE_countXX
+    Given the modern graph
+    And the traversal of
+      """
+      g.V().union(__.outE().count(), __.local(inE().count()))
+      """
+    When iterated to list
+    Then the result should be unordered
+      | result |
+      | d[6].l |
+      | d[0].l |
+      | d[1].l |
+      | d[3].l |
+      | d[1].l |
+      | d[1].l |
+      | d[0].l |