feat(algorithm): support single source shortest path algorithm

computer-algorithm/src/main/java/org/apache/hugegraph/computer/algorithm/path/shortest/SingleSourceShortestPath.java

@@ -18,8 +18,10 @@
 package org.apache.hugegraph.computer.algorithm.path.shortest;
 
 import java.util.Arrays;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 import org.apache.commons.lang3.StringUtils;
@@ -28,7 +30,7 @@
 import org.apache.hugegraph.computer.core.graph.edge.Edge;
 import org.apache.hugegraph.computer.core.graph.id.Id;
 import org.apache.hugegraph.computer.core.graph.value.DoubleValue;
-import org.apache.hugegraph.computer.core.graph.value.IdList;
+import org.apache.hugegraph.computer.core.graph.value.IdSet;
 import org.apache.hugegraph.computer.core.graph.value.Value;
 import org.apache.hugegraph.computer.core.graph.vertex.Vertex;
 import org.apache.hugegraph.computer.core.worker.Computation;
@@ -55,12 +57,17 @@
 
     /**
      * target vertex id.
-     * 1. single target
-     * 2. multiple target: comma separated
+     * 1. single target: one vertex id
+     * 2. multiple target: multiple vertex ids separated by comma
      * 3. all: *
      */
     private String targetId;
 
+    /**
+     * cache of targetId
+     */
+    private Set<String> targetIdSet;
+
     /**
      * target quantity type
      */
@@ -80,9 +87,9 @@
 
     //****************** global data ******************//
     /**
-     * reach target
+     * reached targets
      */
-    private IdList reachTarget;
+    private IdSet reachedTargets;
 
     @Override
     public String category() {
@@ -98,31 +105,27 @@
     public void init(Config config) {
         this.sourceId = config.getString(OPTION_SOURCE_ID, "");
         if (StringUtils.isBlank(this.sourceId)) {
             throw new ComputerException("The param '%s' must not be blank", OPTION_SOURCE_ID);
         }
 
         this.targetId = config.getString(OPTION_TARGET_ID, "");
         if (StringUtils.isBlank(this.targetId)) {
             throw new ComputerException("The param '%s' must not be blank", OPTION_TARGET_ID);
         }
 
-        if (this.targetId.equals("*")) {
-            this.targetQuantityType = QuantityType.ALL;
-        } else if (this.targetId.split(",").length > 1) {
-            // remove spaces
-            this.targetId = Arrays.stream(this.targetId.split(","))
-                                  .map(e -> e.trim())
-                                  .collect(Collectors.joining(","));
-            this.targetQuantityType = QuantityType.MULTIPLE;
-        } else {
-            this.targetQuantityType = QuantityType.SINGLE;
-        }
+        // remove spaces
+        this.targetId = Arrays.stream(this.targetId.split(","))
+                              .map(e -> e.trim())
+                              .collect(Collectors.joining(","));
+        // cache targetId
+        this.targetIdSet = new HashSet<>(Arrays.asList(this.targetId.split(",")));
+        this.targetQuantityType = this.getQuantityType();
 
         this.weightProperty = config.getString(OPTION_WEIGHT_PROPERTY, "");
 
         this.defaultWeight = config.getDouble(OPTION_DEFAULT_WEIGHT, 1);
         if (this.defaultWeight <= 0) {
             throw new ComputerException("The param '%s' must be greater than 0, " +
                                         "actual got '%s'",
                                         OPTION_DEFAULT_WEIGHT, this.defaultWeight);
         }
@@ -142,19 +145,19 @@
         value.zeroDistance(); // source vertex
 
         // single target && source vertex == target vertex
-        if (this.targetQuantityType.equals(QuantityType.SINGLE) &&
+        if (this.targetQuantityType == QuantityType.SINGLE &&
             this.sourceId.equals(this.targetId)) {
             LOG.debug("source vertex {} equals target vertex {}", this.sourceId, this.targetId);
             vertex.inactivate();
             return;
         }
 
         if (vertex.numEdges() <= 0) {
             // isolated vertex
             LOG.debug("source vertex {} can not reach target vertex {}",
                       this.sourceId, this.targetId);
             vertex.inactivate();
             return;
         }
 
         vertex.edges().forEach(edge -> {
@@ -170,9 +173,9 @@
     @Override
     public void compute(ComputationContext context, Vertex vertex,
                         Iterator<SingleSourceShortestPathValue> messages) {
-        if (this.isTarget(vertex) && !this.reachTarget.contains(vertex.id())) {
-            // reach target
-            this.reachTarget.add(vertex.id());
+        if (this.isTarget(vertex) && !this.reachedTargets.contains(vertex.id())) {
+            // reached targets
+            this.reachedTargets.add(vertex.id());
         }
 
         while (messages.hasNext()) {
@@ -187,7 +190,7 @@
             }
 
             // reach all target or nowhere to go
-            if (this.isAllTargetReach(vertex) || vertex.numEdges() <= 0) {
+            if (this.isAllTargetsReached(vertex) || vertex.numEdges() <= 0) {
                 continue;
             }
 
@@ -205,15 +208,28 @@
 
     @Override
     public void beforeSuperstep(WorkerContext context) {
-        this.reachTarget = context.aggregatedValue(
-                SingleSourceShortestPathMaster.SINGLE_SOURCE_SHORTEST_PATH_REACH_TARGET);
+        this.reachedTargets = context.aggregatedValue(
+                SingleSourceShortestPathMaster.SINGLE_SOURCE_SHORTEST_PATH_REACHED_TARGETS);
     }
 
     @Override
     public void afterSuperstep(WorkerContext context) {
         context.aggregateValue(
-                SingleSourceShortestPathMaster.SINGLE_SOURCE_SHORTEST_PATH_REACH_TARGET,
-                this.reachTarget);
+                SingleSourceShortestPathMaster.SINGLE_SOURCE_SHORTEST_PATH_REACHED_TARGETS,
+                this.reachedTargets);
+    }
+
+    /**
+     * get QuantityType by this.targetId
+     */
+    private QuantityType getQuantityType() {
+        if (this.targetId.equals("*")) {
+            return QuantityType.ALL;
+        } else if (this.targetId.contains(",")) {
+            return QuantityType.MULTIPLE;
+        } else {
+            return QuantityType.SINGLE;
+        }
     }
 
     /**
@@ -232,16 +248,16 @@
         Value property = edge.property(this.weightProperty);
         if (property != null) {
             if (!property.isNumber()) {
                 throw new ComputerException("The value of %s must be a numeric value, " +
                                             "actual got '%s'",
                                             this.weightProperty, property.string());
             }
 
             weight = ((DoubleValue) property).doubleValue();
             if (weight <= 0) {
                 throw new ComputerException("The value of %s must be greater than 0, " +
                                             "actual got '%s'",
                                             this.weightProperty, property.string());
             }
         }
         return weight;
@@ -251,42 +267,29 @@
      * determine whether vertex is one of the target
      */
     private boolean isTarget(Vertex vertex) {
-        if (this.targetQuantityType.equals(QuantityType.SINGLE) &&
-            this.idEquals(vertex, this.targetId)) {
-            return true;
-        }
-
-        if (this.targetQuantityType.equals(QuantityType.MULTIPLE)) {
-            for (String targetId : this.targetId.split(",")) {
-                if (this.idEquals(vertex, targetId)) {
-                    return true;
-                }
-            }
-        }
-        return false;
+        return this.targetIdSet.contains(vertex.id().toString());
     }
 
     /**
-     * determine whether reach all target
+     * determine whether all targets reached
      */
-    private boolean isAllTargetReach(Vertex vertex) {
-        if (this.targetQuantityType.equals(QuantityType.SINGLE) &&
+    private boolean isAllTargetsReached(Vertex vertex) {
+        if (this.targetQuantityType == QuantityType.SINGLE &&
             this.idEquals(vertex, this.targetId)) {
             return true;
         }
 
-        if (this.targetQuantityType.equals(QuantityType.MULTIPLE)) {
-            String[] targetIds = this.targetId.split(",");
-            if (targetIds.length == this.reachTarget.size()) {
-                List<String> reachTargets = this.reachTarget.values()
-                                                            .stream().map(Id::toString)
-                                                            .collect(Collectors.toList());
-                for (String targetId : targetIds) {
-                    if (!reachTargets.contains(targetId)) {
+        if (this.targetQuantityType == QuantityType.MULTIPLE) {
+            if (this.targetIdSet.size() == this.reachedTargets.size()) {
+                List<String> reachedTargets = this.reachedTargets.value()
+                                                                 .stream().map(Id::toString)
+                                                                 .collect(Collectors.toList());
+                for (String targetId : this.targetIdSet) {
+                    if (!reachedTargets.contains(targetId)) {
                         return false;
                     }
                 }
                 return true;
             }
         }
 

computer-algorithm/src/main/java/org/apache/hugegraph/computer/algorithm/path/shortest/SingleSourceShortestPathMaster.java

@@ -17,22 +17,22 @@
 
 package org.apache.hugegraph.computer.algorithm.path.shortest;
 
-import org.apache.hugegraph.computer.core.combiner.IdListMergeCombiner;
+import org.apache.hugegraph.computer.core.combiner.IdSetMergeCombiner;
 import org.apache.hugegraph.computer.core.graph.value.ValueType;
 import org.apache.hugegraph.computer.core.master.MasterComputation;
 import org.apache.hugegraph.computer.core.master.MasterComputationContext;
 import org.apache.hugegraph.computer.core.master.MasterContext;
 
 public class SingleSourceShortestPathMaster implements MasterComputation {
 
-    public static final String SINGLE_SOURCE_SHORTEST_PATH_REACH_TARGET =
-            "single_source_shortest_path.reach_target";
+    public static final String SINGLE_SOURCE_SHORTEST_PATH_REACHED_TARGETS =
+            "single_source_shortest_path.reached_targets";
 
     @Override
     public void init(MasterContext context) {
-        context.registerAggregator(SINGLE_SOURCE_SHORTEST_PATH_REACH_TARGET,
-                                   ValueType.ID_LIST,
-                                   IdListMergeCombiner.class);
+        context.registerAggregator(SINGLE_SOURCE_SHORTEST_PATH_REACHED_TARGETS,
+                                   ValueType.ID_SET,
+                                   IdSetMergeCombiner.class);
     }
 
     @Override

computer-api/src/main/java/org/apache/hugegraph/computer/core/combiner/IdSetMergeCombiner.java

@@ -0,0 +1,30 @@
+/*
+ * 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.computer.core.combiner;
+
+import org.apache.hugegraph.computer.core.graph.value.IdSet;
+
+public class IdSetMergeCombiner implements Combiner<IdSet> {
+
+    @Override
+    public void combine(IdSet v1, IdSet v2, IdSet result) {
+        // merge
+        result.addAll(v1);
+        result.addAll(v2);
+    }
+}

computer-api/src/main/java/org/apache/hugegraph/computer/core/graph/value/IdSet.java

@@ -107,4 +107,8 @@
     public int compareTo(Value o) {
         throw new UnsupportedOperationException("compareTo");
     }
+
+    public int size() {
+        return this.values.size();
+    }
 }