Fix bug with degree centrality when processing multiple batches (#865)

Author: mneedham

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

@@ -40,7 +40,6 @@ public class DegreeCentrality extends Algorithm<DegreeCentrality> implements Deg
     private Graph graph;
     private final ExecutorService executor;
     private final int concurrency;
-    private volatile AtomicInteger nodeQueue = new AtomicInteger();
 
     private double[] degrees;
 
@@ -70,12 +69,15 @@ public void compute() {
         this.starts = new long[taskCount];
         this.partitions = new double[taskCount][batchSize];
 
+        long startNode = 0L;
         for (int i = 0; i < taskCount; i++) {
+            starts[i] = startNode;
             if(weighted) {
                 tasks.add(new WeightedDegreeTask(starts[i], partitions[i]));
             } else {
                 tasks.add(new DegreeTask(starts[i], partitions[i]));
             }
+            startNode += batchSize;
         }
         ParallelUtil.runWithConcurrency(concurrency, tasks, executor);
     }

tests/src/test/java/org/neo4j/graphalgo/algo/DegreeProcIssue848IntegrationTest.java

@@ -0,0 +1,136 @@
+/**
+ * Copyright (c) 2017 "Neo4j, Inc." <http://neo4j.com>
+ *
+ * This file is part of Neo4j Graph Algorithms <http://github.com/neo4j-contrib/neo4j-graph-algorithms>.
+ *
+ * Neo4j Graph Algorithms is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.graphalgo.algo;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.neo4j.graphalgo.DegreeCentralityProc;
+import org.neo4j.graphalgo.TestDatabaseCreator;
+import org.neo4j.graphdb.Label;
+import org.neo4j.graphdb.Result;
+import org.neo4j.graphdb.Transaction;
+import org.neo4j.internal.kernel.api.exceptions.KernelException;
+import org.neo4j.kernel.impl.proc.Procedures;
+import org.neo4j.kernel.internal.GraphDatabaseAPI;
+
+import java.util.*;
+import java.util.function.Consumer;
+
+import static org.junit.Assert.*;
+
+@RunWith(Parameterized.class)
+public class DegreeProcIssue848IntegrationTest {
+
+    private static GraphDatabaseAPI db;
+
+    private static final String DB_CYPHER = "" +
+            "UNWIND range(1,10001) as s\n" +
+            "CREATE (:Node{id:s});\n";
+
+    @AfterClass
+    public static void tearDown() throws Exception {
+        if (db != null) db.shutdown();
+    }
+
+    @BeforeClass
+    public static void setup() throws KernelException {
+        db = TestDatabaseCreator.createTestDatabase();
+        try (Transaction tx = db.beginTx()) {
+            db.execute(DB_CYPHER).close();
+            tx.success();
+        }
+
+        db.getDependencyResolver()
+                .resolveDependency(Procedures.class)
+                .registerProcedure(DegreeCentralityProc.class);
+    }
+
+    @Parameterized.Parameters(name = "{0}")
+    public static Collection<Object[]> data() {
+        return Arrays.asList(
+                new Object[]{"Heavy"},
+                new Object[]{"Light"},
+                new Object[]{"Kernel"},
+                new Object[]{"Huge"}
+        );
+    }
+
+    @Parameterized.Parameter
+    public String graphImpl;
+
+    @Test
+    public void multipleBatches() throws Exception {
+        final Map<Long, Double> actual = new HashMap<>();
+        runQuery(
+                "CALL algo.degree.stream(\"Node\", \"REL\", {direction: \"incoming\"})\n" +
+                        "YIELD nodeId, score\n",
+                row -> actual.put(
+                        (Long)row.get("nodeId"),
+                        (Double) row.get("score")));
+
+        Map<Long, Double> expected = new HashMap<>();
+        for (long i = 0; i < 10001; i++) {
+             expected.put(i, 0.0);
+        }
+
+        assertMapEquals(expected, actual);
+    }
+
+    private static void assertMapEquals(
+            Map<Long, Double> expected,
+            Map<Long, Double> actual) {
+        assertEquals("number of elements", expected.size(), actual.size());
+        HashSet<Long> expectedKeys = new HashSet<>(expected.keySet());
+        for (Map.Entry<Long, Double> entry : actual.entrySet()) {
+            assertTrue(
+                    "unknown key " + entry.getKey(),
+                    expectedKeys.remove(entry.getKey()));
+            assertEquals(
+                    "value for " + entry.getKey(),
+                    expected.get(entry.getKey()),
+                    entry.getValue(),
+                    0.1);
+        }
+        for (Long expectedKey : expectedKeys) {
+            fail("missing key " + expectedKey);
+        }
+    }
+
+    private static void runQuery(
+            String query,
+            Consumer<Result.ResultRow> check) {
+        runQuery(query, new HashMap<>(), check);
+    }
+
+    private static void runQuery(
+            String query,
+            Map<String, Object> params,
+            Consumer<Result.ResultRow> check) {
+        try (Result result = db.execute(query, params)) {
+            result.accept(row -> {
+                check.accept(row);
+                return true;
+            });
+        }
+    }
+
+}