Implement Leiden stream procedure

Co-authored-by: Ioannis Panagiotas <ioannis.panagiotas@neotechnology.com>

Author: vnickolov

algo/src/main/java/org/neo4j/gds/leiden/LeidenAlgorithmFactory.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j 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.gds.leiden;
+
+import org.neo4j.gds.GraphAlgorithmFactory;
+import org.neo4j.gds.api.Graph;
+import org.neo4j.gds.core.utils.progress.tasks.ProgressTracker;
+
+public class LeidenAlgorithmFactory<CONFIG extends LeidenBaseConfig> extends GraphAlgorithmFactory<Leiden, CONFIG> {
+    @Override
+    public Leiden build(Graph graph, CONFIG configuration, ProgressTracker progressTracker) {
+
+        double gamma = configuration.gamma() / graph.relationshipCount();
+
+        return new Leiden(
+            graph,
+            configuration.maxLevels(),
+            gamma,
+            configuration.theta(),
+            configuration.randomSeed().orElse(0L),
+            progressTracker
+        );
+    }
+
+    @Override
+    public String taskName() {
+        return "Leiden";
+    }
+}

algo/src/main/java/org/neo4j/gds/leiden/LeidenBaseConfig.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j 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.gds.leiden;
+
+import org.immutables.value.Value;
+import org.neo4j.gds.config.AlgoBaseConfig;
+import org.neo4j.gds.config.RandomSeedConfig;
+import org.neo4j.gds.config.RelationshipWeightConfig;
+
+public interface LeidenBaseConfig extends
+    AlgoBaseConfig,
+    RelationshipWeightConfig,
+    RandomSeedConfig {
+
+    @Value.Default
+    default double gamma() {
+        return 1.0;
+    }
+
+    @Value.Default
+    default double theta() {
+        return 0.01;
+    }
+
+    @Value.Default
+    default int maxLevels() {
+        return 10;
+    }
+}

algo/src/main/java/org/neo4j/gds/leiden/LeidenStreamConfig.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j 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.gds.leiden;
+
+
+import org.neo4j.gds.annotation.Configuration;
+import org.neo4j.gds.annotation.ValueClass;
+import org.neo4j.gds.core.CypherMapWrapper;
+
+@ValueClass
+@Configuration
+@SuppressWarnings("immutables:subtype")
+public interface LeidenStreamConfig extends LeidenBaseConfig {
+    static LeidenStreamConfig of(CypherMapWrapper userInput) {
+        return new LeidenStreamConfigImpl(userInput);
+    }
+}

doc/asciidoc/operations-reference/appendix-a-graph-algos.adoc

@@ -309,4 +309,6 @@ include::../algorithms/algorithm-tiers.adoc[]
 | `gds.alpha.kmeans.mutate`
 | `gds.alpha.kmeans.stats`
 | `gds.alpha.kmeans.stream`
+.1+<.^| Leiden
+| `gds.alpha.leiden.stream`
 |===

open-packaging/src/test/java/org/neo4j/gds/OpenGdsProcedureSmokeTest.java

@@ -62,6 +62,8 @@ class OpenGdsProcedureSmokeTest extends BaseProcTest {
         "gds.alpha.closeness.harmonic.write",
         "gds.alpha.closeness.harmonic.stream",
 
+        "gds.alpha.leiden.stream",
+
         "gds.alpha.kmeans.mutate",
         "gds.alpha.kmeans.stats",
         "gds.alpha.kmeans.stream",
@@ -455,7 +457,7 @@ void countShouldMatch() {
         );
 
         // If you find yourself updating this count, please also update the count in SmokeTest.kt
-        int expectedCount = 314;
+        int expectedCount = 315;
         assertEquals(
             expectedCount,
             registeredProcedures.size(),

proc/community/src/main/java/org/neo4j/gds/leiden/LeidenStreamProc.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j 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.gds.leiden;
+
+import org.neo4j.gds.AlgoBaseProc;
+import org.neo4j.gds.AlgorithmFactory;
+import org.neo4j.gds.core.CypherMapWrapper;
+import org.neo4j.gds.core.utils.paged.HugeLongArray;
+import org.neo4j.gds.executor.ComputationResultConsumer;
+import org.neo4j.gds.executor.ProcedureExecutor;
+import org.neo4j.procedure.Description;
+import org.neo4j.procedure.Name;
+import org.neo4j.procedure.Procedure;
+
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static org.neo4j.procedure.Mode.READ;
+
+public class LeidenStreamProc extends AlgoBaseProc<Leiden, HugeLongArray, LeidenStreamConfig, LeidenStreamProc.StreamResult> {
+    // Config
+    // relationshipWeightProperty
+    // maxIterations
+
+    // Output
+    // nodeId
+    // communityId
+
+    static final String DESCRIPTION =
+        "Leiden is a community detection algorithm, which guarantees that communities are well connected";
+
+    @Procedure(name = "gds.alpha.leiden.stream", mode = READ)
+    @Description(DESCRIPTION)
+    public Stream<StreamResult> stream(
+        @Name(value = "graphName") String graphName,
+        @Name(value = "configuration", defaultValue = "{}") Map<String, Object> configuration
+    ) {
+        var streamSpec = new LeidenStreamSpec();
+
+        return new ProcedureExecutor<>(
+            streamSpec,
+            executionContext()
+        ).compute(graphName, configuration, true, true);
+    }
+
+    @Override
+    public AlgorithmFactory<?, Leiden, LeidenStreamConfig> algorithmFactory() {
+        return new LeidenStreamSpec().algorithmFactory();
+    }
+
+    @Override
+    public ComputationResultConsumer<Leiden, HugeLongArray, LeidenStreamConfig, Stream<StreamResult>> computationResultConsumer() {
+        return new LeidenStreamSpec().computationResultConsumer();
+    }
+
+    @Override
+    protected LeidenStreamConfig newConfig(String username, CypherMapWrapper config) {
+        return new LeidenStreamSpec().newConfigFunction().apply(username, config);
+    }
+
+    public static class StreamResult {
+
+        public final long nodeId;
+        public final long communityId;
+
+        StreamResult(long nodeId, long communityId) {
+            this.nodeId = nodeId;
+            this.communityId = communityId;
+        }
+    }
+}

proc/community/src/main/java/org/neo4j/gds/leiden/LeidenStreamSpec.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j 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.gds.leiden;
+
+import org.neo4j.gds.core.utils.paged.HugeLongArray;
+import org.neo4j.gds.executor.AlgorithmSpec;
+import org.neo4j.gds.executor.ComputationResultConsumer;
+import org.neo4j.gds.executor.GdsCallable;
+import org.neo4j.gds.executor.NewConfigFunction;
+
+import java.util.stream.LongStream;
+import java.util.stream.Stream;
+
+import static org.neo4j.gds.executor.ExecutionMode.STREAM;
+import static org.neo4j.gds.leiden.LeidenStreamProc.DESCRIPTION;
+
+@GdsCallable(name = "gds.alpha.leiden.stream", description = DESCRIPTION, executionMode = STREAM)
+public class LeidenStreamSpec implements AlgorithmSpec<Leiden, HugeLongArray, LeidenStreamConfig, Stream<LeidenStreamProc.StreamResult>, LeidenAlgorithmFactory<LeidenStreamConfig>> {
+    @Override
+    public String name() {
+        return "LeidenStream";
+    }
+
+    @Override
+    public LeidenAlgorithmFactory<LeidenStreamConfig> algorithmFactory() {
+        return new LeidenAlgorithmFactory<>();
+    }
+
+    @Override
+    public NewConfigFunction<LeidenStreamConfig> newConfigFunction() {
+        return (__, config) -> LeidenStreamConfig.of(config);
+    }
+
+    @Override
+    public ComputationResultConsumer<Leiden, HugeLongArray, LeidenStreamConfig, Stream<LeidenStreamProc.StreamResult>> computationResultConsumer() {
+        return (computationResult, executionContext) -> {
+            var graph = computationResult.graph();
+            var communities = computationResult.result();
+            return LongStream.range(0, graph.nodeCount())
+                .mapToObj(nodeId -> new LeidenStreamProc.StreamResult(graph.toOriginalNodeId(nodeId), communities.get(nodeId)));
+        };
+    }
+}

proc/community/src/test/java/org/neo4j/gds/leiden/LeidenStreamProcTest.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j 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.gds.leiden;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.neo4j.gds.BaseProcTest;
+import org.neo4j.gds.catalog.GraphProjectProc;
+import org.neo4j.gds.extension.Neo4jGraph;
+
+import java.util.HashSet;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class LeidenStreamProcTest extends BaseProcTest {
+
+    @Neo4jGraph
+    private static final String DB_CYPHER =
+        "CREATE " +
+        "  (a0:Node)," +
+        "  (a1:Node)," +
+        "  (a2:Node)," +
+        "  (a3:Node)," +
+        "  (a4:Node)," +
+        "  (a5:Node)," +
+        "  (a6:Node)," +
+        "  (a7:Node)," +
+        "  (a0)-[:R {weight: 1.0}]->(a1)," +
+        "  (a0)-[:R {weight: 1.0}]->(a2)," +
+        "  (a0)-[:R {weight: 1.0}]->(a3)," +
+        "  (a0)-[:R {weight: 1.0}]->(a4)," +
+        "  (a2)-[:R {weight: 1.0}]->(a3)," +
+        "  (a2)-[:R {weight: 1.0}]->(a4)," +
+        "  (a3)-[:R {weight: 1.0}]->(a4)," +
+        "  (a1)-[:R {weight: 1.0}]->(a5)," +
+        "  (a1)-[:R {weight: 1.0}]->(a6)," +
+        "  (a1)-[:R {weight: 1.0}]->(a7)," +
+        "  (a5)-[:R {weight: 1.0}]->(a6)," +
+        "  (a5)-[:R {weight: 1.0}]->(a7)," +
+        "  (a6)-[:R {weight: 1.0}]->(a7)";
+
+    @BeforeEach
+    void setUp() throws Exception {
+        registerProcedures(
+            GraphProjectProc.class,
+            LeidenStreamProc.class
+        );
+
+        runQuery("CALL gds.graph.project('leiden', '*', '*')");
+    }
+
+    @Test
+    void stream() {
+        runQuery("CALL gds.alpha.leiden.stream('leiden')", result -> {
+            assertThat(result.columns()).containsExactlyInAnyOrder("nodeId", "communityId");
+            long resultRowCount = 0;
+            var communities = new HashSet<Long>();
+            while(result.hasNext()) {
+                var next = result.next();
+                assertThat(next.get("nodeId")).isInstanceOf(Long.class);
+                assertThat(next.get("communityId")).isInstanceOf(Long.class);
+                communities.add((Long) next.get("communityId"));
+                resultRowCount++;
+            }
+            assertThat(resultRowCount).isEqualTo(8L);
+            assertThat(communities).hasSize(2);
+            return  true;
+        });
+    }
+}