3.4 louvain backward (#737)

* intermediate communities must be explicitly requested on streaming side

* write side

Author: mneedham

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

@@ -43,8 +43,9 @@
  */
 public class LouvainProc {
 
-    public static final String CONFIG_CLUSTER_PROPERTY = "writeProperty";
-    public static final String DEFAULT_CLUSTER_PROPERTY = "communities";
+    public static final String INTERMEDIATE_COMMUNITIES_WRITE_PROPERTY = "intermediateCommunitiesWriteProperty";
+    public static final String DEFAULT_CLUSTER_PROPERTY = "community";
+    public static final String INCLUDE_INTERMEDIATE_COMMUNITIES = "includeIntermediateCommunities";
 
     @Context
     public GraphDatabaseAPI api;
@@ -93,7 +94,7 @@ public Stream<LouvainResult> louvain(
         }
 
         if (configuration.isWriteFlag()) {
-            builder.timeWrite(() -> write(graph, louvain.getDendrogram(), configuration));
+            builder.timeWrite(() -> write(graph, louvain.getDendrogram(), louvain.getCommunityIds(), configuration));
         }
 
         return Stream.of(builder.build());
@@ -125,7 +126,7 @@ public Stream<Louvain.StreamingResult> louvainStream(
             return Stream.empty();
         }
 
-        return louvain.dendrogramStream();
+        return louvain.dendrogramStream(configuration.get(INCLUDE_INTERMEDIATE_COMMUNITIES, false));
     }
 
     public Graph graph(ProcedureConfiguration config) {
@@ -138,16 +139,18 @@ public Graph graph(ProcedureConfiguration config) {
                 .load(config.getGraphImpl());
     }
 
-    private void write(Graph graph, int[][] communities, ProcedureConfiguration configuration) {
+    private void write(Graph graph, int[][] allCommunities, int[] finalCommunities, ProcedureConfiguration configuration) {
         log.debug("Writing results");
+        boolean includeIntermediateCommunities = configuration.get(INCLUDE_INTERMEDIATE_COMMUNITIES, false);
 
         new LouvainCommunityExporter(
                 api,
                 Pools.DEFAULT,
                 configuration.getConcurrency(),
                 graph,
-                communities[0].length,
-                configuration.getWriteProperty(DEFAULT_CLUSTER_PROPERTY))
-                .export(communities);
+                allCommunities[0].length,
+                configuration.getWriteProperty(DEFAULT_CLUSTER_PROPERTY),
+                configuration.get(INTERMEDIATE_COMMUNITIES_WRITE_PROPERTY, "communities"))
+                .export(allCommunities, finalCommunities, includeIntermediateCommunities);
     }
 }

algo/src/main/java/org/neo4j/graphalgo/impl/louvain/Louvain.java

@@ -209,15 +209,18 @@ public Stream<Result> resultStream() {
                 .mapToObj(i -> new Result(i, communities[i]));
     }
 
-    public Stream<StreamingResult> dendrogramStream() {
+    public Stream<StreamingResult> dendrogramStream(boolean includeIntermediateCommunities) {
         return IntStream.range(0, rootNodeCount)
                 .mapToObj(i -> {
-                    List<Long> communities = new ArrayList<>(dendrogram.length);
-                    for (int[] community : dendrogram) {
-                        communities.add((long) community[i]);
+                    List<Long> communitiesList = null;
+                    if (includeIntermediateCommunities) {
+                        communitiesList = new ArrayList<>(dendrogram.length);
+                        for (int[] community : dendrogram) {
+                            communitiesList.add((long) community[i]);
+                        }
                     }
 
-                    return new StreamingResult(root.toOriginalNodeId(i), communities);
+                    return new StreamingResult(root.toOriginalNodeId(i), communitiesList, communities[i]);
                 });
     }
 
@@ -272,11 +275,13 @@ public Result(long id, long community) {
     public static final class StreamingResult {
         public final long nodeId;
         public final List<Long> communities;
+        public final long community;
 
-        public StreamingResult(long nodeId, List<Long> communities) {
+        public StreamingResult(long nodeId, List<Long> communities, long community) {
 
             this.nodeId = nodeId;
             this.communities = communities;
+            this.community = community;
         }
     }
 }

algo/src/main/java/org/neo4j/graphalgo/impl/louvain/LouvainCommunityExporter.java

@@ -40,62 +40,96 @@ public class LouvainCommunityExporter extends StatementApi {
     private final int concurrency;
     private final IdMapping mapping;
     private final int nodeCount;
+    private final int intermediateCommunitiesPropertyId;
     private Integer propertyId;
 
     public LouvainCommunityExporter(GraphDatabaseAPI api,
                                     ExecutorService pool,
                                     int concurrency,
                                     IdMapping mapping,
                                     int nodeCount,
-                                    String propertyName) {
+                                    String propertyName,
+                                    String intermediateCommunitiesPropertyName) {
         super(api);
         this.pool = pool;
         this.concurrency = concurrency;
         this.mapping = mapping;
         this.nodeCount = nodeCount;
 
-            propertyId = applyInTransaction(statement -> statement.tokenWrite().propertyKeyGetOrCreateForName(propertyName));
-
+        propertyId = applyInTransaction(statement -> statement.tokenWrite().propertyKeyGetOrCreateForName(propertyName));
+        intermediateCommunitiesPropertyId = applyInTransaction(statement -> statement.tokenWrite().propertyKeyGetOrCreateForName(intermediateCommunitiesPropertyName));
     }
 
-    public void export(int[][] communities) {
+    public void export(int[][] communities, int[] finalCommunities, boolean includeIntermediateCommunities) {
         final Collection<PrimitiveIntIterable> batchIterables = ParallelUtil.batchIterables(concurrency, nodeCount);
         final ArrayList<Runnable> tasks = new ArrayList<>();
-        batchIterables.forEach(it -> tasks.add(new NodeBatchExporter(it, communities)));
+        batchIterables.forEach(it -> tasks.add(new NodeBatchExporter(it, communities, finalCommunities, includeIntermediateCommunities)));
         ParallelUtil.run(tasks, pool);
     }
 
     private class NodeBatchExporter implements Runnable {
 
         private final PrimitiveIntIterable iterable;
-        private final int[][] communities;
+        private final int[][] allCommunities;
+        private final int[] finalCommunities;
+        private final boolean includeIntermediateCommunities;
 
-        private NodeBatchExporter(PrimitiveIntIterable iterable, int[][] communities) {
+        private NodeBatchExporter(PrimitiveIntIterable iterable, int[][] allCommunities, int[] finalCommunities, boolean includeIntermediateCommunities) {
             this.iterable = iterable;
-            this.communities = communities;
+            this.allCommunities = allCommunities;
+            this.finalCommunities = finalCommunities;
+            this.includeIntermediateCommunities = includeIntermediateCommunities;
         }
 
         @Override
         public void run() {
+            if (includeIntermediateCommunities) {
+                writeEverything();
+            } else {
+                onlyWriteFinalCommunities();
+            }
+        }
+
+        private void writeEverything() {
             acceptInTransaction(statement -> {
                 final Write dataWriteOperations = statement.dataWrite();
                 for(PrimitiveIntIterator it = iterable.iterator(); it.hasNext(); ) {
                     final int id = it.next();
                     // build int array
-                    final int[] data = new int[communities.length];
+                    final int[] data = new int[allCommunities.length];
                     for (int i = 0; i < data.length; i++) {
                         try {
-                            data[i] = communities[i][id];
+                            data[i] = allCommunities[i][id];
                         } catch (Exception e) {
                             throw e; // TODO
                         }
                     }
+
                     dataWriteOperations.nodeSetProperty(
                             mapping.toOriginalNodeId(id),
                             propertyId,
+                            Values.intValue(finalCommunities[id]));
+
+                    dataWriteOperations.nodeSetProperty(
+                            mapping.toOriginalNodeId(id),
+                            intermediateCommunitiesPropertyId,
                             Values.intArray(data));
                 }
             });
         }
+
+        private void onlyWriteFinalCommunities() {
+            acceptInTransaction(statement -> {
+                final Write dataWriteOperations = statement.dataWrite();
+                for(PrimitiveIntIterator it = iterable.iterator(); it.hasNext(); ) {
+                    final int id = it.next();
+
+                    dataWriteOperations.nodeSetProperty(
+                            mapping.toOriginalNodeId(id),
+                            propertyId,
+                            Values.intValue(finalCommunities[id]));
+                }
+            });
+        }
     }
 }

doc/asciidoc/louvain.adoc

@@ -151,6 +151,8 @@ YIELD nodes, communityCount, iterations, loadMillis, computeMillis, writeMillis
 | write          | boolean | true           | yes      | Specifies if the result should be written back as a node property
 | writeProperty  | string  | 'community'    | yes      | The property name written back to the ID of the community that particular node belongs to
 | defaultValue   | float   | null           | yes      | The default value of the weight in case it is missing or invalid
+| includeIntermediateCommunities          | boolean | false           | yes      | Specifies whether an array of intermediate communities should be returned
+| intermediateCommunitiesWriteProperty  | string  | 'communities'    | yes      | The property name written back to the ID of the intermediate communities that particular node belongs to
 | concurrency    | int     | available CPUs | yes      | The number of concurrent threads
 | graph          | string  | 'heavy'        | yes      | Use 'heavy' when describing the subset of the graph with label and relationship-type parameter. Use 'cypher' for describing the subset with cypher node-statement and relationship-statement
 |===

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

@@ -28,7 +28,10 @@
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
+import static junit.framework.TestCase.assertNull;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
@@ -114,6 +117,32 @@ public void test() {
     @Test
     public void testStream() {
         final String cypher = "CALL algo.louvain.stream('', '', {concurrency:1}) " +
+                "YIELD nodeId, community, communities";
+        final IntIntScatterMap testMap = new IntIntScatterMap();
+        DB.execute(cypher).accept(row -> {
+            final long community = (long) row.get("community");
+            System.out.println(community);
+            testMap.addTo((int) community, 1);
+            return false;
+        });
+        assertEquals(3, testMap.size());
+    }
+
+    @Test
+    public void testStreamNoIntermediateCommunitiesByDefault() {
+        final String cypher = "CALL algo.louvain.stream('', '', {concurrency:1}) " +
+                "YIELD nodeId, community, communities";
+        final IntIntScatterMap testMap = new IntIntScatterMap();
+        DB.execute(cypher).accept(row -> {
+            Object communities = row.get("communities");
+            assertNull(communities);
+            return false;
+        });
+    }
+
+    @Test
+    public void testStreamIncludingIntermediateCommunities() {
+        final String cypher = "CALL algo.louvain.stream('', '', {concurrency:1, includeIntermediateCommunities: true}) " +
                 "YIELD nodeId, communities";
         final IntIntScatterMap testMap = new IntIntScatterMap();
         DB.execute(cypher).accept(row -> {
@@ -125,6 +154,56 @@ public void testStream() {
         assertEquals(3, testMap.size());
     }
 
+    @Test
+    public void testWrite() {
+        final String cypher = "CALL algo.louvain('', '', {concurrency:1})";
+        final IntIntScatterMap testMap = new IntIntScatterMap();
+        DB.execute(cypher);
+
+        String readQuery = "MATCH (n) RETURN n.community AS community";
+
+        DB.execute(readQuery).accept(row -> {
+            final int community = (int) row.get("community");
+            testMap.addTo(community, 1);
+            return true;
+        });
+
+        assertEquals(3, testMap.size());
+    }
+
+    @Test
+    public void testWriteIncludingIntermediateCommunities() {
+        final String cypher = "CALL algo.louvain('', '', {concurrency:1, includeIntermediateCommunities: true})";
+        final IntIntScatterMap testMap = new IntIntScatterMap();
+        DB.execute(cypher);
+
+        String readQuery = "MATCH (n) RETURN n.communities AS communities";
+
+        DB.execute(readQuery).accept(row -> {
+            final long community = ((int[]) row.get("communities"))[0];
+            testMap.addTo((int) community, 1);
+            return true;
+        });
+
+        assertEquals(3, testMap.size());
+    }
+
+    @Test
+    public void testWriteNoIntermediateCommunitiesByDefault() {
+        final String cypher = "CALL algo.louvain('', '', {concurrency:1})";
+        DB.execute(cypher);
+
+        final AtomicLong testInteger = new AtomicLong(0);
+        String readQuery = "MATCH (n) WHERE not(exists(n.communities)) RETURN count(*) AS count";
+        DB.execute(readQuery).accept(row -> {
+            long count = (long) row.get("count");
+            testInteger.set(count);
+            return false;
+        });
+
+        assertEquals(9, testInteger.get());
+    }
+
     @Test
     public void testWithLabelRel() {
         final String cypher = "CALL algo.louvain('Node', 'TYPE', {concurrency:1}) " +

tests/src/test/java/org/neo4j/graphalgo/impl/LouvainMultiLevelTest.java

@@ -137,6 +137,8 @@ public void testComplex() throws Exception {
             }
             System.out.println("level " + i + ": " + Arrays.toString(dendogram[i - 1]));
         }
+
         assertArrayEquals(new int[]{0, 0, 0, 1, 1, 1, 2, 2, 2}, dendogram[0]);
+        assertArrayEquals(new int[]{0, 0, 0, 1, 1, 1, 2, 2, 2}, algorithm.getCommunityIds());
     }
 }