From c056d946c52c58abe3e90b5cdb3329d235b81602 Mon Sep 17 00:00:00 2001 From: semyonsinchenko Date: Wed, 15 Jul 2026 16:42:23 +0200 Subject: [PATCH] feat: graphx-style pagerank + pregel optimizations --- src/algorithm/centrality/pagerank.rs | 59 ++++++++++++++++++++++------ src/algorithm/pregel.rs | 48 ++++++++++++++-------- 2 files changed, 80 insertions(+), 27 deletions(-) diff --git a/src/algorithm/centrality/pagerank.rs b/src/algorithm/centrality/pagerank.rs index 5473bab..9d33416 100644 --- a/src/algorithm/centrality/pagerank.rs +++ b/src/algorithm/centrality/pagerank.rs @@ -12,6 +12,11 @@ use futures::{StreamExt, TryStreamExt}; /// Column name for pagerank in the Page Rank algorithm pub const PAGERANK: &str = "pagerank"; +/// Column name for the per-iteration pagerank delta used by the incremental +/// (GraphX-style) PageRank. It is an internal state column and is not part of +/// the final output. +pub const PAGERANK_DELTA: &str = "pagerank_delta"; + /// A builder for the PageRank algorithm. /// #[derive(Debug, Clone)] @@ -93,38 +98,70 @@ impl<'a> PageRankBuilder<'a> { let intermediate_dir = self.checkpoint_config.dir.clone().join("_pregel_raw"); let intermediate_uri = format!("{}{}/", store_url.as_str(), intermediate_dir); + // Incremental (GraphX-style) PageRank. + // + // Each vertex carries two extra columns alongside `out_degree`: + // * `pagerank` — accumulated rank, updated as PR += alpha * msgSum + // * `pagerank_delta` — the per-iteration change = alpha * msgSum (= newPR - oldPR) + // + // The message a source sends is its *delta* split over its out-edges + // (`delta / out_degree`), and only sources whose delta is still above `tol` + // participate — the `participates` column drives the source-side participation + // filter in the Pregel engine, so converged vertices stop emitting messages and + // later iterations become cheap. + // + // Because the result is normalized to sum to 1 at the end, the exact additive + // constant does not matter; we only need a non-zero initial delta to bootstrap. + // Seeding both columns with `reset_prob` also makes the first iteration + // numerically identical to the old full-recompute. + // + // `alpha` (the damping factor = 1 - reset_prob) is kept: it cannot be folded into + // normalization. Dropping it collapses the result to the pure random-walk + // stationary distribution, which is wrong whenever the graph has sinks. + let new_delta = lit(alpha) * coalesce(vec![pregel_default_msg(), lit(0.0)]); + let pregel_builder = graph_with_degrees .pregel() .add_vertex_column( PAGERANK, - lit(reset_prob_per_vertices), // All vertices start with a rank of 1/N - lit(reset_prob_per_vertices) - + lit(alpha) * coalesce(vec![pregel_default_msg(), lit(0.0)]), + lit(reset_prob_per_vertices), + col(PAGERANK) + new_delta.clone(), + ) + .add_vertex_column( + PAGERANK_DELTA, + lit(reset_prob_per_vertices), + new_delta.clone(), ) .add_vertex_column("out_degree", col("out_degree"), col("out_degree")) .add_message( - pregel_src(PAGERANK) / pregel_src("out_degree"), + pregel_src(PAGERANK_DELTA) / pregel_src("out_degree"), MessageDirection::SrcToDst, ) .add_aggregate_expr(sum(pregel_default_msg())) + // A vertex participates while its (new) delta is still above `tol`. This is + // intentionally a separate column from the voting column: participation prunes + // message generation every iteration, while voting only decides when to stop. + .with_participation_column( + "participates", + lit(true), + new_delta.clone().gt(lit(self.tol)), + ) .skip_dest_state() .with_checkpoint_store(store_url.clone()) .set_checkpoint_dir(self.checkpoint_config.dir.join("inner_checkpoint")); let num_iterations = if self.max_iter > 0 { + // Fixed iteration budget: keep the participation filter (cheap tail) but do + // not vote for early termination. pregel_builder .max_iterations(self.max_iter) .run(ctx, &intermediate_uri, false) .await? } else { + // Convergence mode: stop once no vertex is still active. The voting column + // is distinct from `participates`; both happen to equal `delta > tol` here. pregel_builder - .with_vertex_voting( - "rank_changed", - abs(col(PAGERANK) - - (lit(reset_prob_per_vertices) - + lit(alpha) * coalesce(vec![pregel_default_msg(), lit(0.0)]))) - .gt(lit(self.tol)), - ) + .with_vertex_voting("active", new_delta.clone().gt(lit(self.tol))) .run(ctx, &intermediate_uri, false) .await? }; diff --git a/src/algorithm/pregel.rs b/src/algorithm/pregel.rs index da46d3a..4f213d5 100644 --- a/src/algorithm/pregel.rs +++ b/src/algorithm/pregel.rs @@ -380,12 +380,35 @@ impl PregelBuilder { // Main Pregel loop while iteration < max_iterations { iteration += 1; - let src_vertices = - current_vertices - .clone() - .select(current_vertices.schema().fields().iter().map(|field| { - col(field.name()).alias(format!("{}_{}", PREGEL_MSG_SRC, field.name())) - }))?; + // Build the source side of the triplets. + // + // When destination state is not needed (`skip_dest_state`) we can push the + // participation filter *before* the join: only participating sources ever + // emit a message, so dropping inactive sources up front shrinks the join + // input. This is the GraphX-style truncation — inactive sources send nothing. + // + // When destination state IS needed we must not pre-filter sources, because + // a triplet stays relevant as long as *either* endpoint participates; that + // OR semantics cannot be expressed by filtering each side independently, so + // we defer to the post-join filter below. + let src_projection: Vec = current_vertices + .schema() + .fields() + .iter() + .map(|field| { + col(field.name()).alias(format!("{}_{}", PREGEL_MSG_SRC, field.name())) + }) + .collect(); + + let src_vertices = if !self.use_dest_sate { + let mut src_state = current_vertices.clone(); + if let Some(participation_column) = &self.participation_column { + src_state = src_state.filter(col(&participation_column.name))?; + } + src_state.select(src_projection.clone())? + } else { + current_vertices.clone().select(src_projection)? + }; let mut triplets = src_vertices.clone().join_on( edges_struct.clone(), @@ -404,20 +427,13 @@ impl PregelBuilder { JoinType::Inner, vec![pregel_dst(VERTEX_ID).eq(pregel_edge(EDGE_DST))], )?; - } - - // If participation_column is present, skip triplets where both src and dst - // are not participating in iteration. - if self.participation_column.is_some() { - let participation_column = self.participation_column.as_ref().unwrap(); - - if self.use_dest_sate { + // Drop triplets where *both* endpoints are inactive; keep a triplet as + // long as the source or the destination still participates. + if let Some(participation_column) = &self.participation_column { triplets = triplets.filter( pregel_src(&participation_column.name) .or(pregel_dst(&participation_column.name)), )?; - } else { - triplets = triplets.filter(pregel_src(&participation_column.name))?; } }