Weighted PageRank (#718)

* pr bm

* [WIP] Weighted PageRank

* call weighted from the procedure - also adapt integration tests

* Cleaner way of calling weighted vs unweighted

* Implemented for Huge graphs

* docs for weighted PR

* update description

* mention weightProperty in docs

* configurable default

* add default to docs

* making PageRank and WeightedPageRank similar to make it easier to abstract

* move PageRank stuff into its own package

* abstract ComputeStep creation

* ComputeStep as interface

* abstract on ComputeStepFactory

* no longer need this class

* common base class

* reuse code for Huge PageRank as well

* calculate the weighted degree centrality up front and then reuse it

* updating pr benchmarks!

* weight benchmark

* make pr benchmarks compile + fix link

* exclude negative weights on Weighted PageRank

* don't need to compute this for non weighted PageRank

* better named interface

* add refactoring to be done

* introduce DegreeComputer so we have the same level of abstraction in PageRankVariant

* merge left over

* pr bm

* inline all the things in ComputeStep

* much faster with the class as RelationshipConsumer

* update weighted to use consumer too

* run all

* try with all variations of iterations

* add weighted to LDBC

Author: mneedham

algo/src/main/java/org/neo4j/graphalgo/PageRankProc.java

@@ -28,9 +28,9 @@
 import org.neo4j.graphalgo.core.utils.TerminationFlag;
 import org.neo4j.graphalgo.core.utils.paged.AllocationTracker;
 import org.neo4j.graphalgo.core.write.Exporter;
-import org.neo4j.graphalgo.impl.PageRankResult;
+import org.neo4j.graphalgo.impl.pagerank.PageRankResult;
 import org.neo4j.graphalgo.impl.Algorithm;
-import org.neo4j.graphalgo.impl.PageRankAlgorithm;
+import org.neo4j.graphalgo.impl.pagerank.PageRankAlgorithm;
 import org.neo4j.graphalgo.results.PageRankScore;
 import org.neo4j.graphdb.Direction;
 import org.neo4j.graphdb.Node;
@@ -58,6 +58,8 @@ public final class PageRankProc {
     public static final Integer DEFAULT_ITERATIONS = 20;
     public static final String DEFAULT_SCORE_PROPERTY = "pagerank";
 
+    public static final String CONFIG_WEIGHT_KEY = "weightProperty";
+
     @Context
     public GraphDatabaseAPI api;
 
@@ -69,7 +71,7 @@ public final class PageRankProc {
 
     @Procedure(value = "algo.pageRank", mode = Mode.WRITE)
     @Description("CALL algo.pageRank(label:String, relationship:String, " +
-            "{iterations:5, dampingFactor:0.85, write: true, writeProperty:'pagerank', concurrency:4}) " +
+            "{iterations:5, dampingFactor:0.85, weightProperty: null, write: true, writeProperty:'pagerank', concurrency:4}) " +
             "YIELD nodes, iterations, loadMillis, computeMillis, writeMillis, dampingFactor, write, writeProperty" +
             " - calculates page rank and potentially writes back")
     public Stream<PageRankScore.Stats> pageRank(
@@ -79,17 +81,19 @@ public Stream<PageRankScore.Stats> pageRank(
 
         ProcedureConfiguration configuration = ProcedureConfiguration.create(config);
 
+        final String weightPropertyKey = configuration.getString(CONFIG_WEIGHT_KEY, null);
+
         PageRankScore.Stats.Builder statsBuilder = new PageRankScore.Stats.Builder();
         AllocationTracker tracker = AllocationTracker.create();
-        final Graph graph = load(label, relationship, tracker, configuration.getGraphImpl(), statsBuilder, configuration);
+        final Graph graph = load(label, relationship, tracker, configuration.getGraphImpl(), statsBuilder, configuration, weightPropertyKey);
 
         if(graph.nodeCount() == 0) {
             graph.release();
             return Stream.of(statsBuilder.build());
         }
 
         TerminationFlag terminationFlag = TerminationFlag.wrap(transaction);
-        PageRankResult scores = evaluate(graph, tracker, terminationFlag, configuration, statsBuilder);
+        PageRankResult scores = evaluate(graph, tracker, terminationFlag, configuration, statsBuilder, weightPropertyKey);
 
         log.info("PageRank: overall memory usage: %s", tracker.getUsageString());
 
@@ -100,7 +104,7 @@ public Stream<PageRankScore.Stats> pageRank(
 
     @Procedure(value = "algo.pageRank.stream", mode = Mode.READ)
     @Description("CALL algo.pageRank.stream(label:String, relationship:String, " +
-            "{iterations:20, dampingFactor:0.85, concurrency:4}) " +
+            "{iterations:20, dampingFactor:0.85, weightProperty: null, concurrency:4}) " +
             "YIELD node, score - calculates page rank and streams results")
     public Stream<PageRankScore> pageRankStream(
             @Name(value = "label", defaultValue = "") String label,
@@ -109,17 +113,19 @@ public Stream<PageRankScore> pageRankStream(
 
             ProcedureConfiguration configuration = ProcedureConfiguration.create(config);
 
+        final String weightPropertyKey = configuration.getString(CONFIG_WEIGHT_KEY, null);
+
         PageRankScore.Stats.Builder statsBuilder = new PageRankScore.Stats.Builder();
         AllocationTracker tracker = AllocationTracker.create();
-        final Graph graph = load(label, relationship, tracker, configuration.getGraphImpl(), statsBuilder, configuration);
+        final Graph graph = load(label, relationship, tracker, configuration.getGraphImpl(), statsBuilder, configuration, weightPropertyKey);
 
         if(graph.nodeCount() == 0) {
             graph.release();
             return Stream.empty();
         }
 
         TerminationFlag terminationFlag = TerminationFlag.wrap(transaction);
-        PageRankResult scores = evaluate(graph, tracker, terminationFlag, configuration, statsBuilder);
+        PageRankResult scores = evaluate(graph, tracker, terminationFlag, configuration, statsBuilder, weightPropertyKey);
 
         log.info("PageRank: overall memory usage: %s", tracker.getUsageString());
 
@@ -152,11 +158,13 @@ private Graph load(
             String relationship,
             AllocationTracker tracker,
             Class<? extends GraphFactory> graphFactory,
-            PageRankScore.Stats.Builder statsBuilder, ProcedureConfiguration configuration) {
+            PageRankScore.Stats.Builder statsBuilder,
+            ProcedureConfiguration configuration,
+            String weightPropertyKey) {
         GraphLoader graphLoader = new GraphLoader(api, Pools.DEFAULT)
                 .init(log, label, relationship, configuration)
                 .withAllocationTracker(tracker)
-                .withoutRelationshipWeights();
+                .withOptionalRelationshipWeightsFromProperty(weightPropertyKey, configuration.getWeightPropertyDefaultValue(0.0));
 
         Direction direction = configuration.getDirection(Direction.OUTGOING);
         if (direction == Direction.BOTH) {
@@ -178,7 +186,8 @@ private PageRankResult evaluate(
             AllocationTracker tracker,
             TerminationFlag terminationFlag,
             ProcedureConfiguration configuration,
-            PageRankScore.Stats.Builder statsBuilder) {
+            PageRankScore.Stats.Builder statsBuilder,
+            String weightPropertyKey) {
 
         double dampingFactor = configuration.get(CONFIG_DAMPING, DEFAULT_DAMPING);
         int iterations = configuration.getIterations(DEFAULT_ITERATIONS);
@@ -189,14 +198,29 @@ private PageRankResult evaluate(
 
         List<Node> sourceNodes = configuration.get("sourceNodes", new ArrayList<>());
         LongStream sourceNodeIds = sourceNodes.stream().mapToLong(Node::getId);
-        PageRankAlgorithm prAlgo = PageRankAlgorithm.of(
-                tracker,
-                graph,
-                dampingFactor,
-                sourceNodeIds,
-                Pools.DEFAULT,
-                concurrency,
-                batchSize);
+
+        PageRankAlgorithm prAlgo;
+        if(weightPropertyKey != null) {
+            prAlgo = PageRankAlgorithm.weightedOf(
+                    tracker,
+                    graph,
+                    dampingFactor,
+                    sourceNodeIds,
+                    Pools.DEFAULT,
+                    concurrency,
+                    batchSize);
+        } else {
+            prAlgo = PageRankAlgorithm.of(
+                    tracker,
+                    graph,
+                    dampingFactor,
+                    sourceNodeIds,
+                    Pools.DEFAULT,
+                    concurrency,
+                    batchSize);
+        }
+
+
         Algorithm<?> algo = prAlgo
                 .algorithm()
                 .withLog(log)

algo/src/main/java/org/neo4j/graphalgo/impl/WeightedDegreeCentrality.java

@@ -3,9 +3,12 @@
 import org.neo4j.graphalgo.api.Graph;
 import org.neo4j.graphalgo.api.WeightedRelationshipConsumer;
 import org.neo4j.graphalgo.core.utils.ParallelUtil;
+import org.neo4j.graphalgo.core.utils.Pools;
+import org.neo4j.graphalgo.impl.pagerank.HugeComputeStep;
 import org.neo4j.graphdb.Direction;
 
 import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -19,30 +22,35 @@ public class WeightedDegreeCentrality extends Algorithm<WeightedDegreeCentrality
     private final ExecutorService executor;
     private final int concurrency;
     private volatile AtomicInteger nodeQueue = new AtomicInteger();
-    private int[] degrees;
+    private double[] degrees;
 
     public WeightedDegreeCentrality(
             Graph graph,
             ExecutorService executor,
             int concurrency,
             Direction direction
     ) {
+        if (concurrency <= 0) {
+            concurrency = Pools.DEFAULT_QUEUE_SIZE;
+        }
 
         this.graph = graph;
         this.executor = executor;
         this.concurrency = concurrency;
         nodeCount = Math.toIntExact(graph.nodeCount());
         this.direction = direction;
-        degrees = new int[nodeCount];
+        degrees = new double[nodeCount];
     }
 
     public WeightedDegreeCentrality compute() {
         nodeQueue.set(0);
-        final ArrayList<Future<?>> futures = new ArrayList<>();
+
+        List<DegreeTask> tasks = new ArrayList<>();
         for (int i = 0; i < concurrency; i++) {
-            futures.add(executor.submit(new DegreeTask()));
+            tasks.add(new DegreeTask());
         }
-        ParallelUtil.awaitTermination(futures);
+        ParallelUtil.runWithConcurrency(concurrency, tasks, executor);
+
         return this;
     }
 
@@ -66,11 +74,11 @@ public void run() {
                     return;
                 }
 
-                int[] weightedDegree = new int[1];
+                double[] weightedDegree = new double[1];
                 graph.forEachRelationship(nodeId, direction, (sourceNodeId, targetNodeId, relationId, weight) -> {
-                    double v = graph.weightOf(targetNodeId, sourceNodeId);
-                    System.out.println(sourceNodeId + ", " + targetNodeId + " -> " + weight + ", " + v);
-                    weightedDegree[0] += weight;
+                    if(weight > 0) {
+                        weightedDegree[0] += weight;
+                    }
                     return true;
                 });
 
@@ -80,7 +88,7 @@ public void run() {
         }
     }
 
-    public int[] degrees() {
+    public double[] degrees() {
         return degrees;
     }
 

algo/src/main/java/org/neo4j/graphalgo/impl/pagerank/BaseComputeStep.java

@@ -0,0 +1,154 @@
+package org.neo4j.graphalgo.impl.pagerank;
+
+import org.neo4j.graphalgo.api.Degrees;
+import org.neo4j.graphalgo.api.RelationshipIterator;
+import org.neo4j.graphalgo.api.RelationshipWeights;
+import org.neo4j.graphdb.Direction;
+
+import java.util.Arrays;
+import java.util.stream.IntStream;
+
+import static org.neo4j.graphalgo.core.utils.ArrayUtil.binaryLookup;
+
+public abstract class BaseComputeStep implements ComputeStep {
+    private static final int S_INIT = 0;
+    private static final int S_CALC = 1;
+    private static final int S_SYNC = 2;
+
+    private int state;
+
+    int[] starts;
+    private int[] lengths;
+    private int[] sourceNodeIds;
+    final RelationshipIterator relationshipIterator;
+    final Degrees degrees;
+
+    private final double alpha;
+    private final double dampingFactor;
+
+    private double[] pageRank;
+    double[] deltas;
+    int[][] nextScores;
+    private int[][] prevScores;
+
+    private final int partitionSize;
+    final int startNode;
+    final int endNode;
+
+    BaseComputeStep(
+            double dampingFactor,
+            int[] sourceNodeIds,
+            RelationshipIterator relationshipIterator,
+            Degrees degrees,
+            int partitionSize,
+            int startNode) {
+        this.dampingFactor = dampingFactor;
+        this.alpha = 1.0 - dampingFactor;
+        this.sourceNodeIds = sourceNodeIds;
+        this.relationshipIterator = relationshipIterator;
+        this.degrees = degrees;
+        this.partitionSize = partitionSize;
+        this.startNode = startNode;
+        this.endNode = startNode + partitionSize;
+        state = S_INIT;
+    }
+
+    public void setStarts(int starts[], int[] lengths) {
+        this.starts = starts;
+        this.lengths = lengths;
+    }
+
+    @Override
+    public void run() {
+        if (state == S_CALC) {
+            singleIteration();
+            state = S_SYNC;
+        } else if (state == S_SYNC) {
+            synchronizeScores(combineScores());
+            state = S_CALC;
+        } else if (state == S_INIT) {
+            initialize();
+            state = S_CALC;
+        }
+    }
+
+    private void initialize() {
+        this.nextScores = new int[starts.length][];
+        Arrays.setAll(nextScores, i -> new int[lengths[i]]);
+
+        double[] partitionRank = new double[partitionSize];
+
+        if(sourceNodeIds.length == 0) {
+            Arrays.fill(partitionRank, alpha);
+        } else {
+            Arrays.fill(partitionRank,0);
+
+            int[] partitionSourceNodeIds = IntStream.of(sourceNodeIds)
+                    .filter(sourceNodeId -> sourceNodeId >= startNode && sourceNodeId < endNode)
+                    .toArray();
+
+            for (int sourceNodeId : partitionSourceNodeIds) {
+                partitionRank[sourceNodeId - this.startNode] = alpha;
+            }
+        }
+
+
+        this.pageRank = partitionRank;
+        this.deltas = Arrays.copyOf(partitionRank, partitionSize);
+    }
+
+    abstract void singleIteration();
+
+    public void prepareNextIteration(int[][] prevScores) {
+        this.prevScores = prevScores;
+    }
+
+    private int[] combineScores() {
+        assert prevScores != null;
+        assert prevScores.length >= 1;
+        int[][] prevScores = this.prevScores;
+
+        int length = prevScores.length;
+        int[] allScores = prevScores[0];
+        for (int i = 1; i < length; i++) {
+            int[] scores = prevScores[i];
+            for (int j = 0; j < scores.length; j++) {
+                allScores[j] += scores[j];
+                scores[j] = 0;
+            }
+        }
+
+        return allScores;
+    }
+
+    private void synchronizeScores(int[] allScores) {
+        double dampingFactor = this.dampingFactor;
+        double[] pageRank = this.pageRank;
+
+        int length = allScores.length;
+        for (int i = 0; i < length; i++) {
+            int sum = allScores[i];
+
+            double delta = dampingFactor * (sum / 100_000.0);
+            pageRank[i] += delta;
+            deltas[i] = delta;
+            allScores[i] = 0;
+        }
+    }
+
+    @Override
+    public int[][] nextScores() {
+        return nextScores;
+    }
+
+    @Override
+    public double[] pageRank() {
+        return pageRank;
+    }
+
+    @Override
+    public int[] starts() {
+        return starts;
+    }
+
+}

algo/src/main/java/org/neo4j/graphalgo/impl/pagerank/ComputeStep.java

@@ -0,0 +1,13 @@
+package org.neo4j.graphalgo.impl.pagerank;
+
+public interface ComputeStep extends Runnable {
+    int[][] nextScores();
+
+    double[] pageRank();
+
+    int[] starts();
+
+    void setStarts(int[] startArray, int[] lengthArray);
+
+    void prepareNextIteration(int[][] score);
+}