Migrate PageRank write to facade

algo/src/main/java/org/neo4j/gds/algorithms/centrality/CentralityAlgorithmsMutateBusinessFacade.java

@@ -32,7 +32,6 @@
 import org.neo4j.gds.degree.DegreeCentralityMutateConfig;
 import org.neo4j.gds.harmonic.HarmonicCentralityMutateConfig;
 import org.neo4j.gds.pagerank.PageRankMutateConfig;
-import org.neo4j.gds.pagerank.PageRankResult;
 import org.neo4j.gds.result.CentralityStatistics;
 
 import java.util.function.Supplier;
@@ -133,27 +132,43 @@ public NodePropertyMutateResult<PageRankSpecificFields> pageRank(
         var intermediateResult = runWithTiming(
             () -> centralityAlgorithmsFacade.pageRank(graphName, configuration)
         );
+        var algorithmResult = intermediateResult.algorithmResult;
+        return algorithmResult.result().map(result -> {
+            // 2. Construct NodePropertyValues from the algorithm result
+            // 2.1 Should we measure some post-processing here?
+            var nodePropertyValues = result.nodePropertyValues();
 
-        CentralityFunctionSupplier<PageRankResult> centralityFunctionSupplier = CentralityAlgorithmResult::centralityScoreProvider;
-        SpecificFieldsWithCentralityDistributionSupplier<PageRankResult, PageRankSpecificFields> specificFieldsSupplier = (r, c) -> new PageRankSpecificFields(
-            r.iterations(),
-            r.didConverge(),
-            c
-        );
-        Supplier<PageRankSpecificFields> emptyASFSupplier = () -> PageRankSpecificFields.EMPTY;
+            // 3. Go and mutate the graph store
+            var addNodePropertyResult = mutateNodePropertyService.mutate(
+                configuration.mutateProperty(),
+                nodePropertyValues,
+                configuration.nodeLabelIdentifiers(algorithmResult.graphStore()),
+                algorithmResult.graph(), algorithmResult.graphStore()
+            );
 
-        NodePropertyValuesMapper<PageRankResult> nodePropertyValuesMapper = PageRankResult::nodePropertyValues;
 
-        return mutateNodeProperty(
-            intermediateResult.algorithmResult,
-            configuration,
-            centralityFunctionSupplier,
-            nodePropertyValuesMapper,
-            specificFieldsSupplier,
-            shouldComputeCentralityDistribution,
-            intermediateResult.computeMilliseconds,
-            emptyASFSupplier
-        );
+            var pageRankDistribution = PageRankDistributionComputer.computeDistribution(
+                result,
+                configuration,
+                shouldComputeCentralityDistribution
+            );
+
+            var specificFields = new PageRankSpecificFields(
+                result.iterations(),
+                result.didConverge(),
+                pageRankDistribution.centralitySummary
+            );
+
+            return NodePropertyMutateResult.<PageRankSpecificFields>builder()
+                .computeMillis(intermediateResult.computeMilliseconds)
+                .postProcessingMillis(pageRankDistribution.postProcessingMillis)
+                .nodePropertiesWritten(addNodePropertyResult.nodePropertiesAdded())
+                .mutateMillis(addNodePropertyResult.mutateMilliseconds())
+                .configuration(configuration)
+                .algorithmSpecificFields(specificFields)
+                .build();
+        }).orElseGet(() -> NodePropertyMutateResult.empty(PageRankSpecificFields.EMPTY, configuration));
+
     }
 
 

algo/src/main/java/org/neo4j/gds/algorithms/centrality/CentralityAlgorithmsStatsBusinessFacade.java

@@ -31,7 +31,6 @@
 import org.neo4j.gds.core.concurrency.DefaultPool;
 import org.neo4j.gds.degree.DegreeCentralityStatsConfig;
 import org.neo4j.gds.harmonic.HarmonicCentralityStatsConfig;
-import org.neo4j.gds.pagerank.PageRankResult;
 import org.neo4j.gds.pagerank.PageRankStatsConfig;
 import org.neo4j.gds.result.CentralityStatistics;
 
@@ -128,21 +127,30 @@ public StatsResult<PageRankSpecificFields> pageRank(
             () -> centralityAlgorithmsFacade.pageRank(graphName, configuration)
         );
 
-        CentralityFunctionSupplier<PageRankResult> centralityFunctionSupplier = CentralityAlgorithmResult::centralityScoreProvider;
-        SpecificFieldsWithCentralityDistributionSupplier<PageRankResult, PageRankSpecificFields> specificFieldsSupplier =
-            (r, c) -> new PageRankSpecificFields(r.iterations(), r.didConverge(), c);
+        var algorithmResult = intermediateResult.algorithmResult;
+        return algorithmResult.result().map(result -> {
+            // 2. Construct NodePropertyValues from the algorithm result
+            // 2.1 Should we measure some post-processing here?
+            var nodePropertyValues = result.nodePropertyValues();
 
-        Supplier<PageRankSpecificFields> emptyASFSupplier = () -> PageRankSpecificFields.EMPTY;
+            var pageRankDistribution = PageRankDistributionComputer.computeDistribution(
+                result,
+                configuration,
+                shouldComputeCentralityDistribution
+            );
 
-        return statsResult(
-            intermediateResult.algorithmResult,
-            configuration,
-            centralityFunctionSupplier,
-            specificFieldsSupplier,
-            shouldComputeCentralityDistribution,
-            intermediateResult.computeMilliseconds,
-            emptyASFSupplier
-        );
+            var specificFields = new PageRankSpecificFields(
+                result.iterations(),
+                result.didConverge(),
+                pageRankDistribution.centralitySummary
+            );
+
+            return StatsResult.<PageRankSpecificFields>builder()
+                .computeMillis(intermediateResult.computeMilliseconds)
+                .postProcessingMillis(pageRankDistribution.postProcessingMillis)
+                .algorithmSpecificFields(specificFields)
+                .build();
+        }).orElseGet(() -> StatsResult.empty(PageRankSpecificFields.EMPTY));
 
     }
 

algo/src/main/java/org/neo4j/gds/algorithms/centrality/CentralityAlgorithmsWriteBusinessFacade.java

@@ -24,13 +24,15 @@
 import org.neo4j.gds.algorithms.centrality.specificfields.AlphaHarmonicSpecificFields;
 import org.neo4j.gds.algorithms.centrality.specificfields.CentralityStatisticsSpecificFields;
 import org.neo4j.gds.algorithms.centrality.specificfields.DefaultCentralitySpecificFields;
+import org.neo4j.gds.algorithms.centrality.specificfields.PageRankSpecificFields;
 import org.neo4j.gds.algorithms.writeservices.WriteNodePropertyService;
 import org.neo4j.gds.betweenness.BetweennessCentralityWriteConfig;
 import org.neo4j.gds.closeness.ClosenessCentralityWriteConfig;
 import org.neo4j.gds.config.AlgoBaseConfig;
 import org.neo4j.gds.config.ArrowConnectionInfo;
 import org.neo4j.gds.core.concurrency.DefaultPool;
 import org.neo4j.gds.degree.DegreeCentralityWriteConfig;
+import org.neo4j.gds.pagerank.PageRankWriteConfig;
 import org.neo4j.gds.harmonic.DeprecatedTieredHarmonicCentralityWriteConfig;
 import org.neo4j.gds.harmonic.HarmonicCentralityWriteConfig;
 import org.neo4j.gds.result.CentralityStatistics;
@@ -119,6 +121,56 @@ public NodePropertyWriteResult<DefaultCentralitySpecificFields> closenessCentral
         );
     }
 
+    public NodePropertyWriteResult<PageRankSpecificFields> pageRank(
+        String graphName,
+        PageRankWriteConfig configuration,
+        boolean shouldComputeCentralityDistribution
+    ) {
+        // 1. Run the algorithm and time the execution
+        var intermediateResult = runWithTiming(
+            () -> centralityAlgorithmsFacade.pageRank(graphName, configuration)
+        );
+
+        var algorithmResult = intermediateResult.algorithmResult;
+        return algorithmResult.result().map(result -> {
+            // 2. Construct NodePropertyValues from the algorithm result
+            // 2.1 Should we measure some post-processing here?
+            var nodePropertyValues = result.nodePropertyValues();
+
+            // 3. Write to database
+            var writeNodePropertyResult = writeNodePropertyService.write(
+                algorithmResult.graph(),
+                algorithmResult.graphStore(),
+                nodePropertyValues,
+                configuration.writeConcurrency(),
+                configuration.writeProperty(),
+                "PageRankWrite",
+                configuration.arrowConnectionInfo()
+            );
+
+            var pageRankDistribution = PageRankDistributionComputer.computeDistribution(
+                result,
+                configuration,
+                shouldComputeCentralityDistribution
+            );
+
+            var specificFields = new PageRankSpecificFields(
+                result.iterations(),
+                result.didConverge(),
+                pageRankDistribution.centralitySummary
+            );
+
+            return NodePropertyWriteResult.<PageRankSpecificFields>builder()
+                .computeMillis(intermediateResult.computeMilliseconds)
+                .postProcessingMillis(pageRankDistribution.postProcessingMillis)
+                .nodePropertiesWritten(writeNodePropertyResult.nodePropertiesWritten())
+                .writeMillis(writeNodePropertyResult.writeMilliseconds())
+                .configuration(configuration)
+                .algorithmSpecificFields(specificFields)
+                .build();
+        }).orElseGet(() -> NodePropertyWriteResult.empty(PageRankSpecificFields.EMPTY, configuration));
+    }
+
     public NodePropertyWriteResult<DefaultCentralitySpecificFields> harmonicCentrality(
         String graphName,
         HarmonicCentralityWriteConfig configuration,
@@ -173,8 +225,7 @@ public NodePropertyWriteResult<AlphaHarmonicSpecificFields> alphaHarmonicCentral
 
     }
 
-
-    <RESULT extends CentralityAlgorithmResult, CONFIG extends AlgoBaseConfig> NodePropertyWriteResult<DefaultCentralitySpecificFields> writeToDatabase(
+    private <RESULT extends CentralityAlgorithmResult, CONFIG extends AlgoBaseConfig> NodePropertyWriteResult<DefaultCentralitySpecificFields> writeToDatabase(
         AlgorithmComputationResult<RESULT> algorithmResult,
         CONFIG configuration,
         boolean shouldComputeCentralityDistribution,
@@ -185,12 +236,12 @@ <RESULT extends CentralityAlgorithmResult, CONFIG extends AlgoBaseConfig> NodePr
         Optional<ArrowConnectionInfo> arrowConnectionInfo
     ) {
 
-        CentralityFunctionSupplier<RESULT> centralityFunctionSupplier = (r) -> r.centralityScoreProvider();
+        CentralityFunctionSupplier<RESULT> centralityFunctionSupplier = CentralityAlgorithmResult::centralityScoreProvider;
         SpecificFieldsWithCentralityDistributionSupplier<RESULT, DefaultCentralitySpecificFields> specificFieldsSupplier = (r, c) -> new DefaultCentralitySpecificFields(
             c);
         Supplier<DefaultCentralitySpecificFields> emptyASFSupplier = () -> DefaultCentralitySpecificFields.EMPTY;
 
-        NodePropertyValuesMapper<RESULT> nodePropertyValuesMapper = (r) -> r.nodePropertyValues();
+        NodePropertyValuesMapper<RESULT> nodePropertyValuesMapper = CentralityAlgorithmResult::nodePropertyValues;
 
         return writeToDatabase(
             algorithmResult,

algo/src/main/java/org/neo4j/gds/algorithms/centrality/PageRankDistribution.java

@@ -0,0 +1,33 @@
+/*
+ * 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.algorithms.centrality;
+
+import java.util.Collections;
+import java.util.Map;
+
+class PageRankDistribution {
+    final Map<String, Object> centralitySummary;
+    final long postProcessingMillis;
+
+    PageRankDistribution(Map<String, Object> centralitySummary, long postProcessingMillis) {
+        this.centralitySummary = Collections.unmodifiableMap(centralitySummary);
+        this.postProcessingMillis = postProcessingMillis;
+    }
+}

algo/src/main/java/org/neo4j/gds/algorithms/centrality/PageRankDistributionComputer.java

@@ -0,0 +1,69 @@
+/*
+ * 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.algorithms.centrality;
+
+import org.neo4j.gds.core.concurrency.DefaultPool;
+import org.neo4j.gds.pagerank.PageRankConfig;
+import org.neo4j.gds.pagerank.PageRankResult;
+import org.neo4j.gds.result.CentralityStatistics;
+import org.neo4j.gds.scaling.LogScaler;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.neo4j.gds.utils.StringFormatting.toUpperCaseWithLocale;
+
+final class PageRankDistributionComputer {
+    private static final String HISTOGRAM_ERROR_KEY = "Error";
+
+    private PageRankDistributionComputer() {}
+
+    static PageRankDistribution computeDistribution(
+        PageRankResult result,
+        PageRankConfig configuration,
+        boolean shouldComputeCentralityDistribution
+    ) {
+        Map<String, Object> centralitySummary = new HashMap<>();
+        long postProcessingMillis = 0;
+        if (shouldComputeCentralityDistribution) {
+            var usingLogScaler = configuration.scaler().type().equals(LogScaler.TYPE);
+            if (usingLogScaler) {
+                centralitySummary.put(
+                    HISTOGRAM_ERROR_KEY,
+                    "Unable to create histogram when using scaler of type " + toUpperCaseWithLocale(LogScaler.TYPE)
+                );
+            } else {
+                // Compute result statistics
+                var centralityStatistics = CentralityStatistics.centralityStatistics(
+                    result.nodeCount(),
+                    result.centralityScoreProvider(),
+                    DefaultPool.INSTANCE,
+                    configuration.concurrency(),
+                    true
+                );
+
+                centralitySummary = CentralityStatistics.centralitySummary(centralityStatistics.histogram());
+                postProcessingMillis = centralityStatistics.computeMilliseconds();
+            }
+        }
+        return new PageRankDistribution(centralitySummary, postProcessingMillis);
+    }
+
+}

algo/src/main/java/org/neo4j/gds/pagerank/PageRankResult.java

@@ -47,6 +47,10 @@ public boolean didConverge() {
         return didConverge;
     }
 
+    public long nodeCount() {
+        return scores.size();
+    }
+
     @Override
     public NodePropertyValues nodePropertyValues() {
         return NodePropertyValuesAdapter.adapt(scores);
@@ -56,4 +60,5 @@ public NodePropertyValues nodePropertyValues() {
     public LongToDoubleFunction centralityScoreProvider() {
         return scores::get;
     }
+
 }