use petgraph::Graph;
use petgraph::prelude::*;
use petgraph::visit::NodeIndexable;
use rayon::prelude::*;
use rustc_hash::{FxHashMap, FxHashSet};
use crate::prelude::*;
const DEFAULT_TOLERANCE: f64 = 1e-6;
#[derive(Debug)]
pub struct PageRankWorkingMemory<T> {
ranks: Vec<T>,
new_ranks: Vec<T>,
}
impl<T> PageRankWorkingMemory<T>
where
T: BixverseFloat,
{
pub fn new() -> Self {
Self {
ranks: Vec::new(),
new_ranks: Vec::new(),
}
}
fn ensure_capacity(&mut self, node_count: usize) {
if self.ranks.len() < node_count {
self.ranks.resize(node_count, T::zero());
self.new_ranks.resize(node_count, T::zero());
}
}
}
impl<T> Default for PageRankWorkingMemory<T>
where
T: BixverseFloat,
{
fn default() -> Self {
Self {
ranks: Vec::new(),
new_ranks: Vec::new(),
}
}
}
#[derive(Clone)]
pub struct PageRankGraph<T> {
node_count: usize,
in_edges_flat: Vec<usize>,
in_edge_weights_flat: Vec<T>,
in_edges_offsets: Vec<usize>,
out_weight_sums: Vec<T>,
}
#[allow(dead_code)]
impl<T> PageRankGraph<T>
where
T: BixverseFloat + std::iter::Sum,
{
pub fn from_petgraph(graph: Graph<&str, T>) -> Self {
let node_count = graph.node_count();
let mut out_edges: Vec<Vec<(usize, T)>> = vec![Vec::new(); node_count];
let mut in_edges: Vec<Vec<(usize, T)>> = vec![Vec::new(); node_count];
for i in 0..node_count {
let node_id = graph.from_index(i);
for edge in graph.edges(node_id) {
let target_idx = graph.to_index(edge.target());
let weight = *edge.weight();
out_edges[i].push((target_idx, weight));
in_edges[target_idx].push((i, weight));
}
}
let mut in_edges_flat = Vec::new();
let mut in_edge_weights_flat = Vec::new();
let mut in_edges_offsets = Vec::with_capacity(node_count + 1);
in_edges_offsets.push(0);
for node_in_edges in &in_edges {
for &(node_idx, weight) in node_in_edges {
in_edges_flat.push(node_idx);
in_edge_weights_flat.push(weight);
}
in_edges_offsets.push(in_edges_flat.len());
}
let out_weight_sums: Vec<T> = out_edges
.iter()
.map(|edges| edges.iter().map(|(_, weight)| *weight).sum())
.collect();
Self {
node_count,
in_edges_flat,
in_edge_weights_flat,
in_edges_offsets,
out_weight_sums,
}
}
pub fn from_strings(
nodes: &[String],
from: &[String],
to: &[String],
weights: Option<&[T]>,
undirected: bool,
) -> Self {
assert_same_len!(from, to);
if let Some(weights) = weights {
assert_same_len!(from, weights);
}
let node_count = nodes.len();
let mut name_to_idx = FxHashMap::default();
for (idx, name) in nodes.iter().enumerate() {
name_to_idx.insert(name, idx);
}
let mut out_edges: Vec<Vec<(usize, T)>> = vec![Vec::new(); node_count];
let mut in_edges: Vec<Vec<(usize, T)>> = vec![Vec::new(); node_count];
for (i, (from_name, to_name)) in from.iter().zip(to.iter()).enumerate() {
let from_idx = *name_to_idx.get(from_name).unwrap();
let to_idx = *name_to_idx.get(to_name).unwrap();
let weight = weights.map(|w| w[i]).unwrap_or(T::zero());
out_edges[from_idx].push((to_idx, weight));
in_edges[to_idx].push((from_idx, weight));
if undirected {
out_edges[to_idx].push((from_idx, weight));
in_edges[from_idx].push((to_idx, weight));
}
}
let mut in_edges_flat = Vec::new();
let mut in_edge_weights_flat = Vec::new();
let mut in_edges_offsets = Vec::with_capacity(node_count + 1);
in_edges_offsets.push(0);
for node_in_edges in &in_edges {
for &(node_idx, weight) in node_in_edges {
in_edges_flat.push(node_idx);
in_edge_weights_flat.push(weight);
}
in_edges_offsets.push(in_edges_flat.len());
}
let out_weight_sums: Vec<T> = out_edges
.iter()
.map(|edges| edges.iter().map(|(_, weight)| *weight).sum())
.collect();
Self {
node_count,
in_edges_flat,
in_edge_weights_flat,
in_edges_offsets,
out_weight_sums,
}
}
#[inline]
fn in_edges(&self, node: usize) -> (&[usize], &[T]) {
let start = self.in_edges_offsets[node];
let end = self.in_edges_offsets[node + 1];
(
&self.in_edges_flat[start..end],
&self.in_edge_weights_flat[start..end],
)
}
}
pub fn personalised_page_rank<T>(
graph: Graph<&str, T>,
damping_factor: T,
personalisation_vector: &[T],
nb_iter: usize,
tol: Option<T>,
) -> Vec<T>
where
T: BixverseFloat + std::iter::Sum,
{
let node_count = graph.node_count();
if node_count == 0 {
return vec![];
}
assert!(
T::zero() <= damping_factor && damping_factor <= T::one(),
"Damping factor should be between 0 and 1."
);
assert_eq!(
personalisation_vector.len(),
node_count,
"Personalisation vector length must match node count."
);
let tolerance = tol.unwrap_or(T::from_f64(DEFAULT_TOLERANCE).unwrap());
let mut out_edges: Vec<Vec<(usize, T)>> = vec![Vec::new(); node_count];
let mut in_edges: Vec<Vec<(usize, T)>> = vec![Vec::new(); node_count];
for (i, out_edge_vec) in out_edges.iter_mut().enumerate().take(node_count) {
let node_id = graph.from_index(i);
for edge in graph.edges(node_id) {
let target_idx = graph.to_index(edge.target());
let weight = *edge.weight();
out_edge_vec.push((target_idx, weight));
in_edges[target_idx].push((i, weight));
}
}
let out_weight_sums: Vec<T> = out_edges
.iter()
.map(|edges| edges.iter().map(|(_, weight)| *weight).sum())
.collect();
let mut ranks: Vec<T> = personalisation_vector.to_vec();
let teleport_factor = T::one() - damping_factor;
for _ in 0..nb_iter {
let new_ranks: Vec<T> = (0..node_count)
.into_par_iter()
.map(|v| {
let teleport_prob = teleport_factor * personalisation_vector[v];
let link_prob = in_edges[v]
.iter()
.map(|&(w, edge_weight)| {
if out_weight_sums[w] > T::zero() {
damping_factor * ranks[w] * edge_weight / out_weight_sums[w]
} else {
damping_factor * ranks[w] * personalisation_vector[v]
}
})
.sum::<T>();
teleport_prob + link_prob
})
.collect();
let squared_norm_2 = new_ranks
.par_iter()
.zip(&ranks)
.map(|(new, old)| (*new - *old) * (*new - *old))
.sum::<T>();
ranks = new_ranks;
if squared_norm_2 <= tolerance {
break;
}
}
let sum: T = ranks.iter().copied().sum();
ranks.iter_mut().for_each(|x| *x /= sum);
ranks
}
pub fn personalised_page_rank_optimised<T>(
graph: &PageRankGraph<T>,
damping_factor: T,
personalisation_vector: &[T],
nb_iter: usize,
tolerance: T,
working_memory: &mut PageRankWorkingMemory<T>,
) -> Vec<T>
where
T: BixverseFloat + std::iter::Sum,
{
let node_count = graph.node_count;
working_memory.ensure_capacity(node_count);
let ranks = &mut working_memory.ranks;
let new_ranks = &mut working_memory.new_ranks;
ranks[..node_count].copy_from_slice(personalisation_vector);
let teleport_factor = T::one() - damping_factor;
for _ in 0..nb_iter {
new_ranks[..node_count]
.par_iter_mut()
.enumerate()
.for_each(|(v, new_rank)| {
let teleport_prob = teleport_factor * personalisation_vector[v];
let (in_nodes, in_weights) = graph.in_edges(v);
let link_prob: T = in_nodes
.iter()
.zip(in_weights.iter())
.map(|(&w, &edge_weight)| {
if graph.out_weight_sums[w] > T::zero() {
damping_factor * ranks[w] * edge_weight / graph.out_weight_sums[w]
} else {
damping_factor * ranks[w] * personalisation_vector[v]
}
})
.sum();
*new_rank = teleport_prob + link_prob;
});
let squared_norm_2: T = new_ranks[..node_count]
.par_iter()
.zip(&ranks[..node_count])
.map(|(new, old)| {
let diff = *new - *old;
diff * diff
})
.sum();
std::mem::swap(ranks, new_ranks);
if squared_norm_2 <= tolerance {
break;
}
}
let sum: T = ranks[..node_count].iter().cloned().sum();
if sum > T::zero() {
ranks[..node_count].iter_mut().for_each(|x| *x /= sum);
}
ranks[..node_count].to_vec()
}
pub fn constrained_personalised_page_rank<T>(
graph: &Graph<NodeData, EdgeData<T>>,
damping_factor: T,
personalisation_vector: &[T],
nb_iter: usize,
tol: Option<T>,
sink_node_types: Option<&FxHashSet<String>>,
constrained_edge_types: Option<&FxHashSet<String>>,
) -> Vec<T>
where
T: BixverseFloat + std::iter::Sum,
{
let node_count = graph.node_count();
if node_count == 0 {
return vec![];
}
assert!(
T::zero() <= damping_factor && damping_factor <= T::one(),
"Damping factor should be between 0 and 1."
);
assert_eq!(
personalisation_vector.len(),
node_count,
"Personalization vector length must match node count."
);
let tolerance = tol.unwrap_or(T::from_f64(DEFAULT_TOLERANCE).unwrap());
let binding = FxHashSet::default();
let sink_types = sink_node_types.unwrap_or(&binding);
let constrained_types = constrained_edge_types.unwrap_or(&binding);
let mut out_edges: Vec<Vec<(usize, T, bool)>> = vec![Vec::new(); node_count];
let mut in_edges: Vec<Vec<(usize, T, bool)>> = vec![Vec::new(); node_count];
for node_idx in graph.node_indices() {
let node_idx_usize = graph.to_index(node_idx);
let node_data = &graph[node_idx];
if sink_types.contains(node_data.node_type) {
continue;
}
for edge in graph.edges(node_idx) {
let edge_data = edge.weight();
let target_idx = graph.to_index(edge.target());
let is_sink_edge = constrained_types.contains(edge_data.edge_type);
out_edges[node_idx_usize].push((target_idx, *edge_data.weight, is_sink_edge));
in_edges[target_idx].push((node_idx_usize, *edge_data.weight, is_sink_edge));
}
}
let out_degrees: Vec<T> = out_edges
.iter()
.map(|edges| {
let total_weight: T = edges.iter().map(|(_, weight, _)| *weight).sum();
total_weight
})
.collect();
let mut flowable_ranks: Vec<T> = personalisation_vector.to_vec();
let mut absorbed_ranks: Vec<T> = vec![T::zero(); node_count];
let teleport_factor = T::one() - damping_factor;
for _ in 0..nb_iter {
let (new_flowable, new_absorbed): (Vec<T>, Vec<T>) = (0..node_count)
.into_par_iter()
.map(|v| {
let teleport_mass = teleport_factor * personalisation_vector[v];
let mut flowable = teleport_mass;
let mut absorbed = T::zero();
for &(w, edge_weight, is_sink_edge) in &in_edges[v] {
if out_degrees[w] > T::zero() {
let flow =
damping_factor * flowable_ranks[w] * edge_weight / out_degrees[w];
if is_sink_edge {
absorbed += flow;
} else {
flowable += flow;
}
} else {
flowable += damping_factor * flowable_ranks[w] * personalisation_vector[v];
}
}
(flowable, absorbed)
})
.unzip();
let total_absorbed_mass: T = new_absorbed.iter().copied().sum();
let final_flowable: Vec<T> = new_flowable
.into_iter()
.enumerate()
.map(|(v, flowable)| flowable + total_absorbed_mass * personalisation_vector[v])
.collect();
let total_ranks: Vec<T> = final_flowable
.iter()
.zip(&new_absorbed)
.map(|(f, a)| *f + *a)
.collect();
let squared_norm_2 = total_ranks
.par_iter()
.zip(flowable_ranks.par_iter().zip(&absorbed_ranks))
.map(|(new_total, (old_flow, old_abs))| {
let old_total = *old_flow + *old_abs;
(*new_total - old_total) * (*new_total - old_total)
})
.sum::<T>();
flowable_ranks = final_flowable;
absorbed_ranks = new_absorbed;
if squared_norm_2 <= tolerance {
break;
}
}
let mut ranks: Vec<T> = flowable_ranks
.iter()
.zip(&absorbed_ranks)
.map(|(f, a)| *f + *a)
.collect();
let sum: T = ranks.iter().copied().sum();
if sum > T::zero() {
ranks.iter_mut().for_each(|x| *x /= sum);
}
ranks
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_optimised_page_rank_matches_reference() {
let mut graph = Graph::<&str, f64>::new();
let n0 = graph.add_node("A");
let n1 = graph.add_node("B");
let n2 = graph.add_node("C");
graph.add_edge(n0, n1, 1.0);
graph.add_edge(n0, n2, 1.0);
let p_vec: Vec<f64> = vec![1.0 / 3.0, 1.0 / 3.0, 1.0 / 3.0];
let reference = personalised_page_rank(graph, 0.85, &p_vec, 100, None);
let nodes = vec!["A".to_string(), "B".to_string(), "C".to_string()];
let from = vec!["A".to_string(), "A".to_string()];
let to = vec!["B".to_string(), "C".to_string()];
let weights = vec![1.0, 1.0];
let pr_graph = PageRankGraph::from_strings(&nodes, &from, &to, Some(&weights), false);
let mut memory = PageRankWorkingMemory::new();
let optimised =
personalised_page_rank_optimised(&pr_graph, 0.85, &p_vec, 100, 1e-6, &mut memory);
assert_eq!(reference.len(), 3);
assert_eq!(optimised.len(), 3);
for (i, (r, o)) in reference.iter().zip(optimised.iter()).enumerate() {
assert!(
(r - o).abs() < 1e-6,
"node {i}: reference {r} vs optimised {o}"
);
}
assert!((optimised[1] - optimised[2]).abs() < 1e-6);
assert!((optimised.iter().sum::<f64>() - 1.0).abs() < 1e-6);
}
}