Add triangles stream in compute facade

Author: IoannisPanagiotas

algorithms-compute-facade/src/main/java/org/neo4j/gds/community/CommunityComputeFacade.java

@@ -28,7 +28,9 @@
 import org.neo4j.gds.approxmaxkcut.ApproxMaxKCutResult;
 import org.neo4j.gds.async.AsyncAlgorithmCaller;
 import org.neo4j.gds.beta.pregel.ImmutablePregelResult;
+import org.neo4j.gds.beta.pregel.NodeValue;
 import org.neo4j.gds.beta.pregel.PregelResult;
+import org.neo4j.gds.beta.pregel.PregelSchema;
 import org.neo4j.gds.cliqueCounting.CliqueCounting;
 import org.neo4j.gds.cliqueCounting.CliqueCountingResult;
 import org.neo4j.gds.cliquecounting.CliqueCountingParameters;
@@ -82,11 +84,14 @@
 import org.neo4j.gds.triangle.LocalClusteringCoefficientResult;
 import org.neo4j.gds.triangle.TriangleCountParameters;
 import org.neo4j.gds.triangle.TriangleCountResult;
+import org.neo4j.gds.triangle.TriangleResult;
+import org.neo4j.gds.triangle.TriangleStream;
 import org.neo4j.gds.wcc.Wcc;
 import org.neo4j.gds.wcc.WccParameters;
 
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.stream.Stream;
 
 public class CommunityComputeFacade {
 
@@ -596,6 +601,29 @@ CompletableFuture<TimedAlgorithmResult<TriangleCountResult>> triangleCount(
         );
     }
 
+    CompletableFuture<TimedAlgorithmResult<Stream<TriangleResult>>> triangles(
+        Graph graph,
+        TriangleCountParameters parameters,
+        JobId jobId) {
+
+        if (graph.isEmpty()) {
+            return CompletableFuture.completedFuture(TimedAlgorithmResult.empty( Stream.empty()));
+        }
+
+        var algorithm = TriangleStream.create(
+            graph,
+            DefaultPool.INSTANCE,
+            parameters.concurrency(),
+            parameters.labelFilter(),
+            terminationFlag
+        );
+
+        return algorithmCaller.run(
+            algorithm::compute,
+            jobId
+        );
+    }
+
     CompletableFuture<TimedAlgorithmResult<DisjointSetStruct>> wcc(
         Graph graph,
         WccParameters parameters,
@@ -628,6 +656,7 @@ CompletableFuture<TimedAlgorithmResult<DisjointSetStruct>> wcc(
             jobId
         );
     }
+
     CompletableFuture<TimedAlgorithmResult<PregelResult>> sllpa(
         Graph graph,
         SpeakerListenerLPAConfig configuration,
@@ -636,7 +665,8 @@ CompletableFuture<TimedAlgorithmResult<PregelResult>> sllpa(
     ) {
 
         if (graph.isEmpty()) {
-            return CompletableFuture.completedFuture(TimedAlgorithmResult.empty(ImmutablePregelResult.of(null, 0, false)));
+            var empty = NodeValue.of(new PregelSchema.Builder().build(),0,new Concurrency(1));
+            return CompletableFuture.completedFuture(TimedAlgorithmResult.empty(ImmutablePregelResult.of(empty, 0, false)));
         }
 
         var progressTracker = progressTrackerFactory.create(

algorithms-compute-facade/src/test/java/org/neo4j/gds/community/CommunityComputeFacadeEmptyGraphTest.java

@@ -41,9 +41,7 @@
 import org.neo4j.gds.kcore.KCoreDecompositionParameters;
 import org.neo4j.gds.kcore.KCoreDecompositionResult;
 import org.neo4j.gds.kmeans.KmeansParameters;
-import org.neo4j.gds.labelpropagation.LabelPropagationParameters;
 import org.neo4j.gds.labelpropagation.LabelPropagationResult;
-import org.neo4j.gds.leiden.LeidenParameters;
 import org.neo4j.gds.leiden.LeidenResult;
 import org.neo4j.gds.louvain.LouvainParameters;
 import org.neo4j.gds.louvain.LouvainResult;
@@ -215,7 +213,7 @@ void labelPropagation(){
 
         var future = facade.labelPropagation(
             graph,
-            mock(LabelPropagationParameters.class),
+            null,
             jobIdMock,
             false
         );
@@ -249,7 +247,7 @@ void leiden(){
 
         var future = facade.leiden(
             graph,
-            mock(LeidenParameters.class),
+            null,
             jobIdMock,
             false
         );
@@ -345,6 +343,21 @@ void triangleCount(){
         verifyNoInteractions(algorithmCallerMock);
     }
 
+    @Test
+    void triangles() {
+        var future = facade.triangles(
+            graph,
+            mock(TriangleCountParameters.class),
+            jobIdMock
+        );
+        var result = future.join();
+        assertThat(result.result()).isEmpty();
+
+        verifyNoInteractions(progressTrackerFactoryMock);
+        verifyNoInteractions(algorithmCallerMock);
+    }
+
+
     @Test
     void wcc(){
 

algorithms-compute-facade/src/test/java/org/neo4j/gds/community/CommunityComputeFacadeTest.java

@@ -26,6 +26,7 @@
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.neo4j.gds.Orientation;
 import org.neo4j.gds.ProgressTrackerFactory;
+import org.neo4j.gds.api.Graph;
 import org.neo4j.gds.approxmaxkcut.ApproxMaxKCutParameters;
 import org.neo4j.gds.async.AsyncAlgorithmCaller;
 import org.neo4j.gds.cliquecounting.CliqueCountingMode;
@@ -38,7 +39,6 @@
 import org.neo4j.gds.extension.GdlGraph;
 import org.neo4j.gds.extension.IdFunction;
 import org.neo4j.gds.extension.Inject;
-import org.neo4j.gds.extension.TestGraph;
 import org.neo4j.gds.hdbscan.HDBScanParameters;
 import org.neo4j.gds.k1coloring.K1ColoringParameters;
 import org.neo4j.gds.kcore.KCoreDecompositionParameters;
@@ -89,7 +89,7 @@ class CommunityComputeFacadeTest {
         """;
 
     @Inject
-    private TestGraph graph;
+    private Graph graph;
 
     @Inject
     private IdFunction idFunction;
@@ -120,7 +120,7 @@ void maxKCut() {
                 new Concurrency(4),
                 10_000,
                 Optional.empty(),
-                List.of(),
+                List.of(0L,0L),
                 false,
                 false
             ),
@@ -180,9 +180,9 @@ void hdbscan(){
             graph,
             new HDBScanParameters(
                 new Concurrency(4),
-                1,
+                2,
                 3,
-                1,
+                2,
                 "prop2"
             ),
             jobIdMock,
@@ -416,13 +416,33 @@ void triangleCount(){
         assertThat(results.computeMillis()).isNotNegative();
     }
 
+    @Test
+    void triangles() {
+        var future = facade.triangles(
+            graph,
+            new TriangleCountParameters(new Concurrency(4), 100,List.of()),
+            jobIdMock
+        );
+
+        var results = future.join();
+        long a = idFunction.of("a");
+        long b = idFunction.of("b");
+        long c = idFunction.of("c");
+
+        assertThat(results.result()).isNotEmpty()
+            .anySatisfy(r -> {
+                long[] triangleArray = new long[]{r.nodeA,r.nodeB,r.nodeC};
+                assertThat(triangleArray).containsExactlyInAnyOrder(a,b,c);
+            });
+    }
+
     @Test
     void wcc(){
         var future = facade.wcc(
             graph,
             new WccParameters(
                 0,
-                null,
+                Optional.empty(),
                 new Concurrency(4)
             ),
             jobIdMock,

pregel/src/main/java/org/neo4j/gds/beta/pregel/NodeValue.java

@@ -22,15 +22,15 @@
 import org.jetbrains.annotations.Nullable;
 import org.neo4j.gds.api.DefaultValue;
 import org.neo4j.gds.api.nodeproperties.ValueType;
-import org.neo4j.gds.core.concurrency.Concurrency;
-import org.neo4j.gds.core.concurrency.ParallelUtil;
-import org.neo4j.gds.termination.TerminationFlag;
-import org.neo4j.gds.mem.MemoryEstimation;
-import org.neo4j.gds.mem.MemoryEstimations;
 import org.neo4j.gds.collections.ha.HugeDoubleArray;
 import org.neo4j.gds.collections.ha.HugeLongArray;
 import org.neo4j.gds.collections.ha.HugeObjectArray;
+import org.neo4j.gds.core.concurrency.Concurrency;
+import org.neo4j.gds.core.concurrency.ParallelUtil;
 import org.neo4j.gds.mem.Estimate;
+import org.neo4j.gds.mem.MemoryEstimation;
+import org.neo4j.gds.mem.MemoryEstimations;
+import org.neo4j.gds.termination.TerminationFlag;
 import org.neo4j.gds.utils.StringFormatting;
 import org.neo4j.gds.utils.StringJoining;
 import org.neo4j.gds.values.FloatingPointValue;
@@ -57,7 +57,7 @@ public abstract class NodeValue {
             .collect(Collectors.toMap(Element::propertyKey, Element::propertyType));
     }
 
-    static NodeValue of(PregelSchema schema, long nodeCount, Concurrency concurrency) {
+    public static NodeValue of(PregelSchema schema, long nodeCount, Concurrency concurrency) {
         var properties = schema.elements()
             .stream()
             .collect(Collectors.toMap(