Add optional weight to pagerank

python/python/raphtory/algorithms/__init__.pyi

@@ -289,6 +289,7 @@ def pagerank(
     max_diff: Optional[float] = None,
     use_l2_norm: bool = True,
     damping_factor: float = 0.85,
+    weight: Optional[str] = None,
 ) -> NodeStateF64:
     """
     Pagerank -- pagerank centrality value of the nodes in a graph
@@ -305,6 +306,7 @@ def pagerank(
             is less than the max diff value given.
         use_l2_norm (bool): Flag for choosing the norm to use for convergence checks, True for l2 norm, False for l1 norm. Defaults to True.
         damping_factor (float): The damping factor for the PageRank calculation. Defaults to 0.85.
+        weight (Optional[str]): Edge property key to use as weight. If None, all edges have weight 1.0.
 
     Returns:
         NodeStateF64: Mapping of nodes to their pagerank value.

python/tests/test_base_install/test_graphdb/test_algorithms.py

@@ -312,6 +312,36 @@ def test_page_rank():
     assert actual == expected
 
 
+def test_weighted_page_rank():
+    g = Graph()
+    g.add_edge(0, 1, 2, {"weight": 0.37})
+    g.add_edge(0, 1, 3, {"weight": 4.2})
+    g.add_edge(0, 2, 1, {"weight": 0.9})
+    g.add_edge(0, 2, 4, {"weight": 1.7})
+    g.add_edge(0, 3, 1, {"weight": 2.6})
+    g.add_edge(0, 3, 2, {"weight": 0.05})
+    g.add_edge(0, 4, 3, {"weight": 3.3})
+    g.add_edge(0, 4, 1, {"weight": 0.8})
+
+    actual = algorithms.pagerank(g, iter_count=1000, max_diff=1e-10, weight="weight")
+    for node, expected in [("1", 0.42499), ("2", 0.07353), ("3", 0.42311), ("4", 0.07837)]:
+        assert abs(actual[node] - expected) < 1e-5, f"node {node}: {actual[node]} != {expected}"
+
+
+def test_weighted_page_rank_none_matches_unweighted():
+    g = Graph()
+    g.add_edge(0, 1, 2, {"weight": 1.0})
+    g.add_edge(0, 1, 4, {"weight": 1.0})
+    g.add_edge(0, 2, 3, {"weight": 1.0})
+    g.add_edge(0, 3, 1, {"weight": 1.0})
+    g.add_edge(0, 4, 1, {"weight": 1.0})
+
+    unweighted = algorithms.pagerank(g, iter_count=1000)
+    weighted = algorithms.pagerank(g, iter_count=1000, weight="weight")
+    for node in ["1", "2", "3", "4"]:
+        assert abs(unweighted[node] - weighted[node]) < 1e-5, f"node {node} differs"
+
+
 def test_temporal_reachability():
     g = gen_graph()
 

raphtory-benchmark/benches/algobench.rs

@@ -1,7 +1,7 @@
 use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion, SamplingMode};
 use raphtory::{
     algorithms::{
-        centrality::pagerank::unweighted_page_rank,
+        centrality::pagerank::page_rank,
         components::weakly_connected_components,
         metrics::clustering_coefficient::{
             global_clustering_coefficient::global_clustering_coefficient,
@@ -87,7 +87,7 @@ pub fn graphgen_large_pagerank(c: &mut Criterion) {
         &graph,
         |b, graph| {
             b.iter(|| {
-                let result = unweighted_page_rank(graph, Some(100), None, None, true, None);
+                let result = page_rank(graph, None, Some(100), None, None, true, None);
                 black_box(result);
             });
         },

raphtory-graphql/schema.graphql

@@ -1184,7 +1184,7 @@ type Graph {
 }
 
 type GraphAlgorithmPlugin {
-	pagerank(iterCount: Int!, threads: Int, tol: Float): [PagerankOutput!]!
+	pagerank(iterCount: Int!, threads: Int, tol: Float, weight: String): [PagerankOutput!]!
 	shortest_path(source: String!, targets: [String!]!, direction: String): [ShortestPathOutput!]!
 }
 

raphtory-graphql/src/model/plugins/algorithms.rs

@@ -9,7 +9,7 @@ use itertools::Itertools;
 use ordered_float::OrderedFloat;
 use raphtory::{
     algorithms::{
-        centrality::pagerank::unweighted_page_rank,
+        centrality::pagerank::page_rank,
         pathing::dijkstra::dijkstra_single_source_shortest_paths,
     },
     prelude::NodeViewOps,
@@ -70,6 +70,7 @@ impl<'a> Operation<'a, GraphAlgorithmPlugin> for Pagerank {
             ("iterCount", TypeRef::named_nn(TypeRef::INT)), // _nn stands for not null
             ("threads", TypeRef::named(TypeRef::INT)),      // this one though might be null
             ("tol", TypeRef::named(TypeRef::FLOAT)),
+            ("weight", TypeRef::named(TypeRef::STRING)),
         ]
     }
 
@@ -96,8 +97,10 @@ fn apply_pagerank<'b>(
         .get("damping_factor")
         .map(|v| v.f64())
         .transpose()?;
-    let binding = unweighted_page_rank(
+    let weight = ctx.args.get("weight").map(|v| v.string()).transpose()?;
+    let binding = page_rank(
         &entry_point.graph,
+        weight.as_deref(),
         Some(iter_count),
         threads,
         tol,

raphtory/src/algorithms/centrality/pagerank.rs

@@ -3,29 +3,30 @@ use crate::{
     db::{
         api::{
             state::NodeState,
-            view::{NodeViewOps, StaticGraphViewOps},
+            view::{EdgeViewOps, NodeViewOps, StaticGraphViewOps},
         },
         task::{
             context::Context,
             task::{ATask, Job, Step},
             task_runner::TaskRunner,
         },
     },
-    prelude::GraphViewOps,
+    prelude::{GraphViewOps, PropertiesOps},
 };
 use num_traits::abs;
+use raphtory_api::core::entities::properties::prop::PropUnwrap;
 
 #[derive(Clone, Debug, Default)]
 struct PageRankState {
     score: f64,
-    out_degree: usize,
+    weighted_out_degree: f64,
 }
 
 impl PageRankState {
     fn new(num_nodes: usize) -> Self {
         Self {
             score: 1f64 / num_nodes as f64,
-            out_degree: 0,
+            weighted_out_degree: 0f64,
         }
     }
 
@@ -40,6 +41,7 @@ impl PageRankState {
 /// # Arguments
 ///
 /// - `g`: A GraphView object
+/// - `weight`: Edge property key to use as weight. If None, all edges have weight 1.0.
 /// - `iter_count`: Number of iterations to run the algorithm for
 /// - `threads`: Number of threads to use for parallel execution
 /// - `tol`: The tolerance value for convergence
@@ -50,8 +52,9 @@ impl PageRankState {
 ///
 /// An [AlgorithmResult] object containing the mapping from node ID to the PageRank score of the node
 ///
-pub fn unweighted_page_rank<G: StaticGraphViewOps>(
+pub fn page_rank<G: StaticGraphViewOps>(
     g: &G,
+    weight: Option<&str>,
     iter_count: Option<usize>,
     threads: Option<usize>,
     tol: Option<f64>,
@@ -76,38 +79,53 @@ pub fn unweighted_page_rank<G: StaticGraphViewOps>(
 
     ctx.global_agg_reset(total_sink_contribution);
 
-    let step1 = ATask::new(move |s| {
-        let out_degree = s.out_degree();
-        let state: &mut PageRankState = s.get_mut();
-        state.out_degree = out_degree;
-        Step::Continue
+    let weight_key: Option<String> = weight.map(|s| s.to_string());
+
+    let step1 = ATask::new({
+        let weight_key = weight_key.clone();
+        move |s| {
+            let weighted_out_degree = s.out_edges().iter().fold(0.0f64, |acc, edge| {
+                weight_key
+                    .as_ref()
+                    .and_then(|key| edge.properties().get(key))
+                    .and_then(|p| p.as_f64())
+                    .unwrap_or(1.0)
+                    + acc
+            });
+            let state: &mut PageRankState = s.get_mut();
+            state.weighted_out_degree = weighted_out_degree;
+            Step::Continue
+        }
     });
 
     let step2: ATask<G, ComputeStateVec, PageRankState, _> = ATask::new(move |s| {
-        // reset score
         {
             let state: &mut PageRankState = s.get_mut();
             state.reset();
         }
 
-        for t in s.in_neighbours() {
-            let prev = t.prev();
-
-            s.get_mut().score += prev.score / prev.out_degree as f64;
+        for edge in s.in_edges() {
+            let w = weight_key
+                .as_ref()
+                .and_then(|key| edge.properties().get(key))
+                .and_then(|p| p.as_f64())
+                .unwrap_or(1.0);
+            let nbr = edge.nbr();
+            let prev = nbr.prev();
+            if prev.weighted_out_degree > 0.0 {
+                s.get_mut().score += prev.score * w / prev.weighted_out_degree;
+            }
         }
 
         s.get_mut().score *= damp;
-
         s.get_mut().score += teleport_prob;
         Step::Continue
     });
 
     let step3 = ATask::new(move |s| {
         let state: &mut PageRankState = s.get_mut();
-
-        if state.out_degree == 0 {
+        if state.weighted_out_degree == 0.0 {
             let curr = s.prev().score;
-
             let ts_contrib = factor * curr;
             s.global_update(&total_sink_contribution, ts_contrib);
         }

raphtory/src/python/packages/algorithms.rs

@@ -6,7 +6,7 @@ use crate::{
         centrality::{
             betweenness::betweenness_centrality as betweenness_rs,
             degree_centrality::degree_centrality as degree_centrality_rs, hits::hits as hits_rs,
-            pagerank::unweighted_page_rank,
+            pagerank::page_rank,
         },
         community_detection::{
             label_propagation::label_propagation as label_propagation_rs,
@@ -268,20 +268,23 @@ pub fn out_component(
 ///         is less than the max diff value given.
 ///     use_l2_norm (bool): Flag for choosing the norm to use for convergence checks, True for l2 norm, False for l1 norm. Defaults to True.
 ///     damping_factor (float): The damping factor for the PageRank calculation. Defaults to 0.85.
+///     weight (Optional[str]): Edge property key to use as weight. If None, all edges have weight 1.0.
 ///
 /// Returns:
 ///     NodeStateF64: Mapping of nodes to their pagerank value.
 #[pyfunction]
-#[pyo3(signature = (graph, iter_count=20, max_diff=None, use_l2_norm=true, damping_factor=0.85))]
+#[pyo3(signature = (graph, iter_count=20, max_diff=None, use_l2_norm=true, damping_factor=0.85, weight=None))]
 pub fn pagerank(
     graph: &PyGraphView,
     iter_count: usize,
     max_diff: Option<f64>,
     use_l2_norm: bool,
     damping_factor: Option<f64>,
+    weight: Option<&str>,
 ) -> NodeState<'static, f64, DynamicGraph> {
-    unweighted_page_rank(
+    page_rank(
         &graph.graph,
+        weight,
         Some(iter_count),
         None,
         max_diff,