3.4 degree centrality (#719)

* [wip] degree centrality

* test incoming degrees

* both centrality

* weighted degree centrality

* weighted outgoing

* weighted test

* fix graphview degree

* temporarily only run this on heavy and huge

Author: mneedham

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

@@ -0,0 +1,109 @@
+package org.neo4j.graphalgo.impl;
+
+import org.neo4j.graphalgo.api.Graph;
+import org.neo4j.graphalgo.core.utils.ParallelUtil;
+import org.neo4j.graphalgo.core.utils.ProgressLogger;
+import org.neo4j.graphalgo.impl.betweenness.BetweennessCentrality;
+import org.neo4j.graphdb.Direction;
+
+import java.util.ArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+public class DegreeCentrality extends Algorithm<DegreeCentrality> {
+    private final int nodeCount;
+    private Direction direction;
+    private Graph graph;
+    private final ExecutorService executor;
+    private final int concurrency;
+    private volatile AtomicInteger nodeQueue = new AtomicInteger();
+    private int[] degrees;
+
+    public DegreeCentrality(
+            Graph graph,
+            ExecutorService executor,
+            int concurrency,
+            Direction direction
+    ) {
+
+        this.graph = graph;
+        this.executor = executor;
+        this.concurrency = concurrency;
+        nodeCount = Math.toIntExact(graph.nodeCount());
+        this.direction = direction;
+        degrees = new int[nodeCount];
+    }
+
+    public DegreeCentrality compute() {
+        nodeQueue.set(0);
+        final ArrayList<Future<?>> futures = new ArrayList<>();
+        for (int i = 0; i < concurrency; i++) {
+            futures.add(executor.submit(new DegreeTask()));
+        }
+        ParallelUtil.awaitTermination(futures);
+        return this;
+    }
+
+    @Override
+    public DegreeCentrality me() {
+        return this;
+    }
+
+    @Override
+    public DegreeCentrality release() {
+        graph = null;
+        return null;
+    }
+
+    private class DegreeTask implements Runnable {
+        @Override
+        public void run() {
+            for (; ; ) {
+                final int nodeId = nodeQueue.getAndIncrement();
+                if (nodeId >= nodeCount || !running()) {
+                    return;
+                }
+
+                int degree = graph.degree(nodeId, direction);
+                degrees[nodeId] = degree;
+
+            }
+        }
+    }
+
+    public int[] degrees() {
+        return degrees;
+    }
+
+    public Stream<Result> resultStream() {
+        return IntStream.range(0, nodeCount)
+                .mapToObj(nodeId ->
+                        new Result(graph.toOriginalNodeId(nodeId), degrees[nodeId]));
+    }
+
+    /**
+     * Result class used for streaming
+     */
+    public static final class Result {
+
+        public final long nodeId;
+        public final double centrality;
+
+        public Result(long nodeId, double centrality) {
+            this.nodeId = nodeId;
+            this.centrality = centrality;
+        }
+
+        @Override
+        public String toString() {
+            return "Result{" +
+                    "nodeId=" + nodeId +
+                    ", centrality=" + centrality +
+                    '}';
+        }
+    }
+
+}

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

@@ -0,0 +1,93 @@
+package org.neo4j.graphalgo.impl;
+
+import org.neo4j.graphalgo.api.Graph;
+import org.neo4j.graphalgo.api.WeightedRelationshipConsumer;
+import org.neo4j.graphalgo.core.utils.ParallelUtil;
+import org.neo4j.graphdb.Direction;
+
+import java.util.ArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+public class WeightedDegreeCentrality extends Algorithm<WeightedDegreeCentrality> {
+    private final int nodeCount;
+    private Direction direction;
+    private Graph graph;
+    private final ExecutorService executor;
+    private final int concurrency;
+    private volatile AtomicInteger nodeQueue = new AtomicInteger();
+    private int[] degrees;
+
+    public WeightedDegreeCentrality(
+            Graph graph,
+            ExecutorService executor,
+            int concurrency,
+            Direction direction
+    ) {
+
+        this.graph = graph;
+        this.executor = executor;
+        this.concurrency = concurrency;
+        nodeCount = Math.toIntExact(graph.nodeCount());
+        this.direction = direction;
+        degrees = new int[nodeCount];
+    }
+
+    public WeightedDegreeCentrality compute() {
+        nodeQueue.set(0);
+        final ArrayList<Future<?>> futures = new ArrayList<>();
+        for (int i = 0; i < concurrency; i++) {
+            futures.add(executor.submit(new DegreeTask()));
+        }
+        ParallelUtil.awaitTermination(futures);
+        return this;
+    }
+
+    @Override
+    public WeightedDegreeCentrality me() {
+        return this;
+    }
+
+    @Override
+    public WeightedDegreeCentrality release() {
+        graph = null;
+        return null;
+    }
+
+    private class DegreeTask implements Runnable {
+        @Override
+        public void run() {
+            for (; ; ) {
+                final int nodeId = nodeQueue.getAndIncrement();
+                if (nodeId >= nodeCount || !running()) {
+                    return;
+                }
+
+                int[] weightedDegree = new int[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;
+                    return true;
+                });
+
+                degrees[nodeId] = weightedDegree[0];
+
+            }
+        }
+    }
+
+    public int[] degrees() {
+        return degrees;
+    }
+
+    public Stream<DegreeCentrality.Result> resultStream() {
+        return IntStream.range(0, nodeCount)
+                .mapToObj(nodeId ->
+                        new DegreeCentrality.Result(graph.toOriginalNodeId(nodeId), degrees[nodeId]));
+    }
+
+}

core/src/main/java/org/neo4j/graphalgo/core/neo4jview/GraphView.java

@@ -181,6 +181,10 @@ public int degree(int nodeId, Direction direction) {
             try (NodeCursor nc = transaction.cursors().allocateNodeCursor()) {
                 transaction.dataRead().singleNode(toOriginalNodeId(nodeId), nc);
                 if (nc.next()) {
+                    if (direction == Direction.BOTH || (direction == Direction.OUTGOING && loadAsUndirected) ) {
+                        return rels(transaction).degreeOf(Direction.BOTH, nc);
+                    }
+
                     return rels(transaction).degreeOf(direction, nc);
                 }
                 return 0;