feat(core): support batch+parallel edges traverse

hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/EdgesQueryIterator.java

@@ -0,0 +1,64 @@
+/*
+ * 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.hugegraph.backend.query;
+
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.tx.GraphTransaction;
+import org.apache.hugegraph.type.define.Directions;
+
+public class EdgesQueryIterator implements Iterator<Query> {
+    private final List<Id> labels;
+    private final Directions directions;
+    private final long limit;
+    private final Iterator<Id> sources;
+
+    public EdgesQueryIterator(Iterator<Id> sources,
+                              Directions directions,
+                              List<Id> labels,
+                              long limit) {
+        this.sources = sources;
+        this.labels = labels;
+        this.directions = directions;
+        // Traverse NO_LIMIT 和 Query.NO_LIMIT 不同
+        this.limit = limit < 0 ? Query.NO_LIMIT : limit;
+    }
+
+    @Override
+    public boolean hasNext() {
+        return sources.hasNext();
+    }
+
+    @Override
+    public Query next() {
+        Id sourceId = this.sources.next();
+        ConditionQuery query = GraphTransaction.constructEdgesQuery(sourceId,
+                                                                    this.directions,
+                                                                    this.labels);
+        if (this.limit != Query.NO_LIMIT) {
+            query.limit(this.limit);
+            query.capacity(this.limit);
+        } else {
+            query.capacity(Query.NO_CAPACITY);
+        }
+        return query;
+    }
+
+}

hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/HugeTraverser.java

@@ -17,6 +17,8 @@
 
 package org.apache.hugegraph.traversal.algorithm;
 
+import java.io.Closeable;
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -37,6 +39,7 @@
 import org.apache.hugegraph.backend.id.Id;
 import org.apache.hugegraph.backend.query.Aggregate;
 import org.apache.hugegraph.backend.query.ConditionQuery;
+import org.apache.hugegraph.backend.query.EdgesQueryIterator;
 import org.apache.hugegraph.backend.query.Query;
 import org.apache.hugegraph.backend.query.QueryResults;
 import org.apache.hugegraph.backend.tx.GraphTransaction;
@@ -66,6 +69,7 @@
 import org.apache.hugegraph.util.collection.ObjectIntMapping;
 import org.apache.hugegraph.util.collection.ObjectIntMappingFactory;
 import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.apache.tinkerpop.gremlin.structure.util.CloseableIterator;
 import org.slf4j.Logger;
 
 import com.google.common.collect.ImmutableList;
@@ -465,6 +469,13 @@ private Iterator<Edge> edgesOfVertex(Id source, EdgeStep edgeStep,
         return edgeStep.skipSuperNodeIfNeeded(edges);
     }
 
+    public EdgesIterator edgesOfVertices(Iterator<Id> sources,
+                                         Directions dir,
+                                         List<Id> labelIds,
+                                         long degree) {
+        return new EdgesIterator(new EdgesQueryIterator(sources, dir, labelIds, degree));
+    }
+
     public Iterator<Edge> edgesOfVertex(Id source, Steps steps) {
         List<Id> edgeLabels = steps.edgeLabels();
         ConditionQuery cq = GraphTransaction.constructEdgesQuery(
@@ -1004,4 +1015,31 @@ public Set<Edge> getEdges(Iterator<Id> vertexIter) {
             return edges;
         }
     }
+
+    public class EdgesIterator implements Iterator<Iterator<Edge>>, Closeable {
+        private final Iterator<Iterator<Edge>> currentIt;
+
+        public EdgesIterator(EdgesQueryIterator queryIterator) {
+            List<Iterator<Edge>> iteratorList = new ArrayList<>();
+            while (queryIterator.hasNext()) {
+                iteratorList.add(graph().edges(queryIterator.next()));
+            }
+            this.currentIt = iteratorList.iterator();
+        }
+
+        @Override
+        public boolean hasNext() {
+            return this.currentIt.hasNext();
+        }
+
+        @Override
+        public Iterator<Edge> next() {
+            return this.currentIt.next();
+        }
+
+        @Override
+        public void close() throws IOException {
+            CloseableIterator.closeIterator(currentIt);
+        }
+    }
 }

hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/KneighborTraverser.java

@@ -17,11 +17,11 @@
 
 package org.apache.hugegraph.traversal.algorithm;
 
-import java.util.Iterator;
 import java.util.Set;
 import java.util.function.Consumer;
 
 import org.apache.hugegraph.HugeGraph;
+import org.apache.hugegraph.backend.id.EdgeId;
 import org.apache.hugegraph.backend.id.Id;
 import org.apache.hugegraph.structure.HugeEdge;
 import org.apache.hugegraph.traversal.algorithm.records.KneighborRecords;
@@ -48,25 +48,27 @@
 
         Id labelId = this.getEdgeLabelId(label);
 
-        Set<Id> latest = newSet();
-        Set<Id> all = newSet();
+        KneighborRecords records = new KneighborRecords(true, sourceV, true);
 
-        latest.add(sourceV);
-        this.vertexIterCounter.addAndGet(1L);
+        Consumer<EdgeId> consumer = edgeId -> {
+            if (this.reachLimit(limit, records.size())) {
+                return;
+            }
+            records.addPath(edgeId.ownerVertexId(), edgeId.otherVertexId());
+        };
 
         while (depth-- > 0) {
-            long remaining = limit == NO_LIMIT ? NO_LIMIT : limit - all.size();
-            latest = this.adjacentVertices(sourceV, latest, dir, labelId,
-                                           all, degree, remaining);
-            all.addAll(latest);
-            this.vertexIterCounter.addAndGet(1L);
-            this.edgeIterCounter.addAndGet(latest.size());
-            if (reachLimit(limit, all.size())) {
+            records.startOneLayer(true);
+            bfsQuery(records.keys(), dir, labelId, degree, NO_LIMIT, consumer);
+            records.finishOneLayer();
+            if (reachLimit(limit, records.size())) {
                 break;
             }
         }
 
-        return all;
+        this.vertexIterCounter.addAndGet(records.size());
+
+        return records.idSet(limit);
     }
 
     public KneighborRecords customizedKneighbor(Id source, Steps steps,
@@ -76,33 +78,29 @@
         checkPositive(maxDepth, "k-neighbor max_depth");
         checkLimit(limit);
 
-        boolean concurrent = maxDepth >= this.concurrentDepth();
-
-        KneighborRecords records = new KneighborRecords(concurrent,
+        KneighborRecords records = new KneighborRecords(true,
                                                         source, true);
 
-        Consumer<Id> consumer = v -> {
+        Consumer<Edge> consumer = edge -> {
             if (this.reachLimit(limit, records.size())) {
                 return;
             }
-            Iterator<Edge> edges = edgesOfVertex(v, steps);
-            this.vertexIterCounter.addAndGet(1L);
-            while (!this.reachLimit(limit, records.size()) && edges.hasNext()) {
-                HugeEdge edge = (HugeEdge) edges.next();
-                Id target = edge.id().otherVertexId();
-                records.addPath(v, target);
-
-                records.edgeResults().addEdge(v, target, edge);
-
-                this.edgeIterCounter.addAndGet(1L);
-            }
+            EdgeId edgeId = ((HugeEdge) edge).id();
+            records.addPath(edgeId.ownerVertexId(), edgeId.otherVertexId());
+            records.edgeResults().addEdge(edgeId.ownerVertexId(), edgeId.otherVertexId(), edge);
         };
 
         while (maxDepth-- > 0) {
             records.startOneLayer(true);
-            traverseIds(records.keys(), consumer, concurrent);
+            bfsQuery(records.keys(), steps, NO_LIMIT, consumer);
             records.finishOneLayer();
+            if (this.reachLimit(limit, records.size())) {
+                break;
+            }
         }
+
+        this.vertexIterCounter.addAndGet(records.size());
+
         return records;
     }
 

hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/KoutTraverser.java

@@ -18,12 +18,15 @@
 package org.apache.hugegraph.traversal.algorithm;
 
 import java.util.Iterator;
+import java.util.List;
 import java.util.Set;
 import java.util.function.Consumer;
 
 import org.apache.hugegraph.HugeException;
 import org.apache.hugegraph.HugeGraph;
+import org.apache.hugegraph.backend.id.EdgeId;
 import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.query.Query;
 import org.apache.hugegraph.structure.HugeEdge;
 import org.apache.hugegraph.traversal.algorithm.records.KoutRecords;
 import org.apache.hugegraph.traversal.algorithm.steps.Steps;
@@ -57,34 +60,45 @@ public Set<Id> kout(Id sourceV, Directions dir, String label,
 
         Id labelId = this.getEdgeLabelId(label);
 
-        Set<Id> latest = newIdSet();
-        latest.add(sourceV);
+        Set<Id> sources = newIdSet();
+        Set<Id> neighbors = newIdSet();
+        Set<Id> visited = nearest ? newIdSet() : null;
 
-        Set<Id> all = newIdSet();
-        all.add(sourceV);
+        neighbors.add(sourceV);
+
+        ConcurrentVerticesConsumer consumer;
+
+        long remaining = capacity == NO_LIMIT ? NO_LIMIT : capacity - 1;
 
-        long remaining = capacity == NO_LIMIT ?
-                         NO_LIMIT : capacity - latest.size();
-        this.vertexIterCounter.addAndGet(1L);
         while (depth-- > 0) {
             // Just get limit nodes in last layer if limit < remaining capacity
             if (depth == 0 && limit != NO_LIMIT &&
                 (limit < remaining || remaining == NO_LIMIT)) {
                 remaining = limit;
             }
-            if (nearest) {
-                latest = this.adjacentVertices(sourceV, latest, dir, labelId,
-                                               all, degree, remaining);
-                all.addAll(latest);
-            } else {
-                latest = this.adjacentVertices(sourceV, latest, dir, labelId,
-                                               null, degree, remaining);
+
+            if (visited != null) {
+                visited.addAll(neighbors);
             }
-            this.vertexIterCounter.addAndGet(1L);
-            this.edgeIterCounter.addAndGet(latest.size());
+
+            // swap sources and neighbors
+            Set<Id> tmp = neighbors;
+            neighbors = sources;
+            sources = tmp;
+
+            // start
+            consumer = new ConcurrentVerticesConsumer(sourceV, visited, remaining, neighbors);
+
+            this.vertexIterCounter.addAndGet(sources.size());
+            this.edgeIterCounter.addAndGet(neighbors.size());
+
+            bfsQuery(sources.iterator(), dir, labelId, degree, capacity, consumer);
+
+            sources.clear();
+
             if (capacity != NO_LIMIT) {
                 // Update 'remaining' value to record remaining capacity
-                remaining -= latest.size();
+                remaining -= neighbors.size();
 
                 if (remaining <= 0 && depth > 0) {
                     throw new HugeException(
@@ -94,7 +108,7 @@ public Set<Id> kout(Id sourceV, Directions dir, String label,
             }
         }
 
-        return latest;
+        return neighbors;
     }
 
     public KoutRecords customizedKout(Id source, Steps steps,
@@ -107,33 +121,25 @@ public KoutRecords customizedKout(Id source, Steps steps,
         checkLimit(limit);
         long[] depth = new long[1];
         depth[0] = maxDepth;
-        boolean concurrent = maxDepth >= this.concurrentDepth();
 
-        KoutRecords records = new KoutRecords(concurrent, source, nearest, 0);
+        KoutRecords records = new KoutRecords(true, source, nearest, 0);
 
-        Consumer<Id> consumer = v -> {
+        Consumer<Edge> consumer = edge -> {
             if (this.reachLimit(limit, depth[0], records.size())) {
                 return;
             }
-            Iterator<Edge> edges = edgesOfVertex(v, steps);
-            this.vertexIterCounter.addAndGet(1L);
-            while (!this.reachLimit(limit, depth[0], records.size()) &&
-                   edges.hasNext()) {
-                HugeEdge edge = (HugeEdge) edges.next();
-                Id target = edge.id().otherVertexId();
-                records.addPath(v, target);
-                this.checkCapacity(capacity, records.accessed(), depth[0]);
-
-                records.edgeResults().addEdge(v, target, edge);
-
-                this.edgeIterCounter.addAndGet(1L);
-            }
+            EdgeId edgeId = ((HugeEdge) edge).id();
+            records.addPath(edgeId.ownerVertexId(), edgeId.otherVertexId());
+            records.edgeResults().addEdge(edgeId.ownerVertexId(), edgeId.otherVertexId(), edge);
         };
 
         while (depth[0]-- > 0) {
+            List<Id> sources = records.ids(Query.NO_LIMIT);
             records.startOneLayer(true);
-            this.traverseIds(records.keys(), consumer, concurrent);
+            bfsQuery(sources.iterator(), steps, capacity, consumer);
+            this.vertexIterCounter.addAndGet(sources.size());
             records.finishOneLayer();
+            checkCapacity(capacity, records.accessed(), depth[0]);
         }
         return records;
     }