Revamp JanusGraphMultiQueryStrategy for better parent step usage

This commit improves JanusGraphMultiQueryStrategy to better support multi-query compatible parent steps.

1. This commit brings better support for `repeat` step by introducing next itaration registration process.
Previously `repeat` children steps was getting traversers registered from the beginning of all outer repeat steps which
could result in duplicate or unnecesary retrievals for the first batch. Moreover, next iterations were not considered.
This commit changes the approach to register only with the beginning and end of the first reaching `repeat` step.

2. Instead of using JanusGraphMultiQueryStep in the local traversal, this commit changes the approach to use
the most outer eligible parent's JanusGraphMultiQueryStep. This improves usage of the existing parent steps
by properly registering batches instead of registering batches considering local elements only.

3. This commit adds support to almost all known TinkerPop Parent steps.
The exception is `match` step. We didn't have proper outter start registration for `match` step previously and now as well.

Fixes #3733
Fixes #3735
Fixes #2996

Signed-off-by: Oleksandr Porunov <alexandr.porunov@gmail.com>

janusgraph-backend-testutils/src/main/java/org/janusgraph/graphdb/JanusGraphTest.java

@@ -4671,6 +4671,96 @@ public void testMultiQueryMetricsWhenReadingFromBackend() {
         assertTrue(queryProfilerAnnotationIsPresent(t, QueryProfiler.MULTIQUERY_ANNOTATION));
     }
 
+    private void addTestAdjacentVertices(Vertex vertex, int levelVerticesAmount, int depth){
+        for (int i = 0; i < levelVerticesAmount; ++i) {
+            Vertex adjacentVertex = graph.addVertex();
+            adjacentVertex.property("foo", "bar");
+            adjacentVertex.property("depth", depth);
+            if(depth==2){
+                adjacentVertex.property("someProp", "val");
+            }
+            vertex.addEdge("knows", adjacentVertex);
+            if(depth>0){
+                addTestAdjacentVertices(adjacentVertex, levelVerticesAmount, depth-1);
+            }
+        }
+    }
+
+    @Test
+    public void testLimitBatchSizeForRepeatStep() {
+
+        JanusGraphVertex a = graph.addVertex();
+        a.property("depth", Integer.MAX_VALUE);
+        addTestAdjacentVertices(a,  10,4);
+
+        //TODO: below query doesn't leverage RepeatStep optimization due to being too complicated.
+        // It seems when `until` is placed before `repeat` with `emit` at the end the evaluation of the query
+        // is split into smaller branches and the multi-query optimization is applied only on those small branches
+        // instead of be applied on multiple branches (like with all other `repeat` cases).
+        // Ideally we should support below cases as well, but could be done as a separate task in case there is
+        // demand for such `repeat` usage.
+//        clopen(option(USE_MULTIQUERY), true, option(LIMITED_BATCH), true);
+//        JanusGraphVertex b = graph.addVertex();
+//        graph.traversal().V(a).until(__.in("knows").is(b)).repeat(__.out("knows")).emit(__.in()).count().profile().next();
+        graph.tx().rollback();
+        //graph.traversal().V(a).out().out().project("a", "b").by(__.id()).by(__.values()).toList();
+        //graph.traversal().V(a).out("knows").repeat(__.repeat(__.has("depth", P.gte(8)).out("knows")).emit()).emit().count().next();
+        TraversalMetrics repeatUntilProfile = graph.traversal().V(a).repeat(__.out("knows")).until(__.in().has("someProp", "val")).count().profile().next();
+        System.out.println(repeatUntilProfile);
+
+        graph.tx().rollback();
+        TraversalMetrics untilRepeatProfile = graph.traversal().V(a).until(__.in().has("someProp", "val")).repeat(__.out("knows")).count().profile().next();
+        System.out.println(untilRepeatProfile);
+
+        graph.tx().rollback();
+        TraversalMetrics emitRepeatProfile = graph.traversal().V(a).emit().repeat(__.out("knows")).count().profile().next();
+        System.out.println(emitRepeatProfile);
+
+        graph.tx().rollback();
+        TraversalMetrics repeatEmitProfile = graph.traversal().V(a).repeat(__.out("knows")).emit().count().profile().next();
+        System.out.println(repeatEmitProfile);
+
+        graph.tx().rollback();
+        TraversalMetrics repeatNonTrueEmitProfile = graph.traversal().V(a).repeat(__.out("knows")).emit(__.in().has("someProp", "val")).count().profile().next();
+        System.out.println(repeatNonTrueEmitProfile);
+
+        graph.tx().rollback();
+        TraversalMetrics nonTrueEmitRepeatProfile = graph.traversal().V(a).emit(__.in().has("someProp", "val")).repeat(__.out("knows")).count().profile().next();
+        System.out.println(nonTrueEmitRepeatProfile);
+
+        graph.tx().rollback();
+        TraversalMetrics untilRepeatNonEmptyEmitProfile = graph.traversal().V(a).until(__.in().has("someProp", "val")).repeat(__.out("knows")).emit(__.in().has("someProp", "val")).count().profile().next();
+        System.out.println(untilRepeatNonEmptyEmitProfile);
+
+        graph.tx().rollback();
+        TraversalMetrics untilNonEmptyEmitRepeatProfile = graph.traversal().V(a).until(__.in().has("someProp", "val")).emit(__.in().has("someProp", "val")).repeat(__.out("knows")).count().profile().next();
+        System.out.println(untilNonEmptyEmitRepeatProfile);
+
+        graph.tx().rollback();
+        TraversalMetrics nonEmptyEmitRepeatUntilProfile = graph.traversal().V(a).emit(__.in().has("someProp", "val")).repeat(__.out("knows")).until(__.in().has("someProp", "val")).count().profile().next();
+        System.out.println(nonEmptyEmitRepeatUntilProfile);
+
+        graph.tx().rollback();
+        TraversalMetrics repeatNonEmptyEmitUntilProfile = graph.traversal().V(a).repeat(__.out("knows")).emit(__.in().has("someProp", "val")).until(__.in().has("someProp", "val")).count().profile().next();
+        System.out.println(repeatNonEmptyEmitUntilProfile);
+
+        graph.tx().rollback();
+        TraversalMetrics untilRepeatEmitProfile = graph.traversal().V(a).until(__.in().has("someProp", "val")).repeat(__.out("knows")).emit().count().profile().next();
+        System.out.println(untilRepeatEmitProfile);
+
+        graph.tx().rollback();
+        TraversalMetrics untilEmitRepeatProfile = graph.traversal().V(a).until(__.in().has("someProp", "val")).emit().repeat(__.out("knows")).count().profile().next();
+        System.out.println(untilEmitRepeatProfile);
+
+        graph.tx().rollback();
+        TraversalMetrics emitRepeatUntilProfile = graph.traversal().V(a).emit().repeat(__.out("knows")).until(__.in().has("someProp", "val")).count().profile().next();
+        System.out.println(emitRepeatUntilProfile);
+
+        graph.tx().rollback();
+        TraversalMetrics repeatEmitUntilProfile = graph.traversal().V(a).repeat(__.out("knows")).emit().until(__.in().has("someProp", "val")).count().profile().next();
+        System.out.println(repeatEmitUntilProfile);
+    }
+
     @Test
     public void testLimitBatchSizeForMultiQuery() {
         int numV = 100;