uqa_graph/
centrality.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Vertex centrality measures: `PageRank`, `HITS`, betweenness. Each
8//! operator runs against a [`GraphStore`] and returns a
9//! [`GraphPostingList`] keyed on vertex id with a calibrated score.
10
11use std::collections::{BTreeMap, VecDeque};
12
13use uqa_core::{DocId, Payload, PostingEntry, PostingList, Value, VertexId};
14
15use crate::posting_list::{GraphPayload, GraphPostingList};
16use crate::store::{GraphStore, GraphStoreError, GraphStoreResult};
17
18const MAX_EXACT_F64_INTEGER: u64 = 9_007_199_254_740_992;
19
20fn usize_as_f64(value: usize, context: &str) -> GraphStoreResult<f64> {
21    let value = u64::try_from(value)
22        .map_err(|_| GraphStoreError::CorruptGraph(format!("{context} exceeds the u64 range")))?;
23    u64_as_f64(value, context)
24}
25
26fn u64_as_f64(value: u64, context: &str) -> GraphStoreResult<f64> {
27    if value <= MAX_EXACT_F64_INTEGER {
28        Ok(value as f64)
29    } else {
30        Err(GraphStoreError::CorruptGraph(format!(
31            "{context} {value} exceeds the exact f64 integer range"
32        )))
33    }
34}
35
36/// `PageRank` centrality (power iteration with damping).
37///
38/// Iterates `new_rank[v] = (1 - d)/N + d * sum(rank[u] / out_deg(u))`
39/// over in-neighbors `u`, until the L1 delta drops below
40/// `tolerance` or `max_iterations` is reached. Final scores are
41/// min-max normalized to `[0, 1]`.
42pub struct PageRank<'a> {
43    pub graph: &'a str,
44    pub damping: f64,
45    pub max_iterations: u32,
46    pub tolerance: f64,
47}
48
49impl<'a> PageRank<'a> {
50    pub fn new(graph: &'a str) -> Self {
51        Self {
52            graph,
53            damping: 0.85,
54            max_iterations: 100,
55            tolerance: 1e-6,
56        }
57    }
58
59    pub fn damping(mut self, d: f64) -> Self {
60        self.damping = d;
61        self
62    }
63
64    pub fn max_iterations(mut self, k: u32) -> Self {
65        self.max_iterations = k;
66        self
67    }
68
69    pub fn tolerance(mut self, t: f64) -> Self {
70        self.tolerance = t;
71        self
72    }
73
74    pub fn execute<G: GraphStore>(&self, store: &G) -> GraphStoreResult<GraphPostingList> {
75        if !self.damping.is_finite() || !(0.0..=1.0).contains(&self.damping) {
76            return Err(GraphStoreError::InvalidMutation(format!(
77                "PageRank damping must be finite and in [0, 1], got {}",
78                self.damping
79            )));
80        }
81        if !self.tolerance.is_finite() || self.tolerance < 0.0 {
82            return Err(GraphStoreError::InvalidMutation(format!(
83                "PageRank tolerance must be finite and non-negative, got {}",
84                self.tolerance
85            )));
86        }
87        let vertices: Vec<VertexId> = store.vertex_ids_in_graph(self.graph)?.into_iter().collect();
88        let n = vertices.len();
89        if n == 0 {
90            return Ok(GraphPostingList::new());
91        }
92        if n == 1 {
93            return single_vertex_result(vertices[0], 1.0, &vertices, self.graph);
94        }
95
96        let n_f64 = usize_as_f64(n, "PageRank vertex count")?;
97        let mut rank: BTreeMap<VertexId, f64> =
98            vertices.iter().map(|v| (*v, 1.0 / n_f64)).collect();
99        let mut out_degree: BTreeMap<VertexId, usize> = BTreeMap::new();
100        let mut in_neighbors: BTreeMap<VertexId, Vec<VertexId>> = BTreeMap::new();
101        for v in &vertices {
102            out_degree.insert(*v, store.out_edge_ids(*v, self.graph)?.len());
103            let mut ins: Vec<VertexId> = Vec::new();
104            for eid in store.in_edge_ids(*v, self.graph)? {
105                let edge = store.get_edge(eid).ok_or_else(|| {
106                    GraphStoreError::CorruptGraph(format!("missing PageRank edge {eid}"))
107                })?;
108                ins.push(edge.source_id);
109            }
110            in_neighbors.insert(*v, ins);
111        }
112
113        let d = self.damping;
114        for _ in 0..self.max_iterations {
115            let mut new_rank: BTreeMap<VertexId, f64> = BTreeMap::new();
116            for v in &vertices {
117                let mut incoming = 0.0;
118                if let Some(ins) = in_neighbors.get(v) {
119                    for u in ins {
120                        let deg = *out_degree.get(u).unwrap_or(&0);
121                        if deg > 0 {
122                            incoming += rank[u] / usize_as_f64(deg, "PageRank out-degree")?;
123                        }
124                    }
125                }
126                new_rank.insert(*v, (1.0 - d) / n_f64 + d * incoming);
127            }
128            let delta: f64 = vertices.iter().map(|v| (new_rank[v] - rank[v]).abs()).sum();
129            rank = new_rank;
130            if delta < self.tolerance {
131                break;
132            }
133        }
134
135        let normalized = min_max_normalize(&rank, &vertices)?;
136        build_score_result(&vertices, &normalized, self.graph, &BTreeMap::new())
137    }
138}
139
140/// `HITS` centrality (hub / authority mutual reinforcement).
141///
142/// Authority of `v` is the sum of hub scores of in-neighbors; hub is
143/// the sum of authority scores of out-neighbors. Each round normalizes
144/// by L2 norm. Final hub and authority scores are min-max normalized
145/// to `[0, 1]`. The payload's `score` is the authority; the per-entry
146/// fields carry both `hub_score` and `authority_score`.
147pub struct HITS<'a> {
148    pub graph: &'a str,
149    pub max_iterations: u32,
150    pub tolerance: f64,
151}
152
153impl<'a> HITS<'a> {
154    pub fn new(graph: &'a str) -> Self {
155        Self {
156            graph,
157            max_iterations: 100,
158            tolerance: 1e-6,
159        }
160    }
161
162    pub fn max_iterations(mut self, k: u32) -> Self {
163        self.max_iterations = k;
164        self
165    }
166
167    pub fn tolerance(mut self, t: f64) -> Self {
168        self.tolerance = t;
169        self
170    }
171
172    pub fn execute<G: GraphStore>(&self, store: &G) -> GraphStoreResult<GraphPostingList> {
173        if !self.tolerance.is_finite() || self.tolerance < 0.0 {
174            return Err(GraphStoreError::InvalidMutation(format!(
175                "HITS tolerance must be finite and non-negative, got {}",
176                self.tolerance
177            )));
178        }
179        let vertices: Vec<VertexId> = store.vertex_ids_in_graph(self.graph)?.into_iter().collect();
180        if vertices.is_empty() {
181            return Ok(GraphPostingList::new());
182        }
183
184        let mut hub: BTreeMap<VertexId, f64> = vertices.iter().map(|v| (*v, 1.0)).collect();
185        let mut auth: BTreeMap<VertexId, f64> = vertices.iter().map(|v| (*v, 1.0)).collect();
186        let mut in_neighbors: BTreeMap<VertexId, Vec<VertexId>> = BTreeMap::new();
187        let mut out_neighbors: BTreeMap<VertexId, Vec<VertexId>> = BTreeMap::new();
188        for v in &vertices {
189            let mut ins = Vec::new();
190            for eid in store.in_edge_ids(*v, self.graph)? {
191                let edge = store.get_edge(eid).ok_or_else(|| {
192                    GraphStoreError::CorruptGraph(format!("missing HITS edge {eid}"))
193                })?;
194                ins.push(edge.source_id);
195            }
196            in_neighbors.insert(*v, ins);
197            let mut outs = Vec::new();
198            for eid in store.out_edge_ids(*v, self.graph)? {
199                let edge = store.get_edge(eid).ok_or_else(|| {
200                    GraphStoreError::CorruptGraph(format!("missing HITS edge {eid}"))
201                })?;
202                outs.push(edge.target_id);
203            }
204            out_neighbors.insert(*v, outs);
205        }
206
207        for _ in 0..self.max_iterations {
208            let mut new_auth: BTreeMap<VertexId, f64> = BTreeMap::new();
209            for v in &vertices {
210                let s = in_neighbors[v].iter().map(|u| hub[u]).sum::<f64>();
211                new_auth.insert(*v, s);
212            }
213            let mut new_hub: BTreeMap<VertexId, f64> = BTreeMap::new();
214            for v in &vertices {
215                let s = out_neighbors[v].iter().map(|w| new_auth[w]).sum::<f64>();
216                new_hub.insert(*v, s);
217            }
218            let auth_norm = new_auth.values().map(|x| x * x).sum::<f64>().sqrt();
219            let hub_norm = new_hub.values().map(|x| x * x).sum::<f64>().sqrt();
220            if auth_norm > 0.0 {
221                for v in &vertices {
222                    let value = new_auth.get_mut(v).ok_or_else(|| {
223                        GraphStoreError::CorruptGraph(format!(
224                            "missing HITS authority state for vertex {v}"
225                        ))
226                    })?;
227                    *value /= auth_norm;
228                }
229            }
230            if hub_norm > 0.0 {
231                for v in &vertices {
232                    let value = new_hub.get_mut(v).ok_or_else(|| {
233                        GraphStoreError::CorruptGraph(format!(
234                            "missing HITS hub state for vertex {v}"
235                        ))
236                    })?;
237                    *value /= hub_norm;
238                }
239            }
240            let delta: f64 = vertices
241                .iter()
242                .map(|v| (new_auth[v] - auth[v]).abs() + (new_hub[v] - hub[v]).abs())
243                .sum();
244            auth = new_auth;
245            hub = new_hub;
246            if delta < self.tolerance {
247                break;
248            }
249        }
250
251        let auth_n = min_max_normalize(&auth, &vertices)?;
252        let hub_n = min_max_normalize(&hub, &vertices)?;
253        let mut extra_fields: BTreeMap<VertexId, BTreeMap<String, Value>> = BTreeMap::new();
254        for v in &vertices {
255            let mut m: BTreeMap<String, Value> = BTreeMap::new();
256            m.insert("hub_score".into(), Value::Float(hub_n[v]));
257            m.insert("authority_score".into(), Value::Float(auth_n[v]));
258            extra_fields.insert(*v, m);
259        }
260        build_score_result(&vertices, &auth_n, self.graph, &extra_fields)
261    }
262}
263
264/// Betweenness centrality via Brandes algorithm.
265///
266/// For unweighted directed graphs, the per-vertex betweenness is
267/// `sum over s != v != t of (sigma_st(v) / sigma_st)`. Scores are
268/// normalized by `(N-1)*(N-2)` and clamped into `[0, 1]`.
269pub struct BetweennessCentrality<'a> {
270    pub graph: &'a str,
271}
272
273impl<'a> BetweennessCentrality<'a> {
274    pub fn new(graph: &'a str) -> Self {
275        Self { graph }
276    }
277
278    pub fn execute<G: GraphStore>(&self, store: &G) -> GraphStoreResult<GraphPostingList> {
279        let vertices: Vec<VertexId> = store.vertex_ids_in_graph(self.graph)?.into_iter().collect();
280        let n = vertices.len();
281        if n == 0 {
282            return Ok(GraphPostingList::new());
283        }
284        if n == 1 {
285            return single_vertex_result(vertices[0], 0.0, &vertices, self.graph);
286        }
287
288        let vertex_index: BTreeMap<VertexId, usize> = vertices
289            .iter()
290            .enumerate()
291            .map(|(idx, vertex_id)| (*vertex_id, idx))
292            .collect();
293        let mut out_neighbors: Vec<Vec<usize>> = vec![Vec::new(); n];
294        for (idx, vertex_id) in vertices.iter().enumerate() {
295            for eid in store.out_edge_ids(*vertex_id, self.graph)? {
296                let edge = store.get_edge(eid).ok_or_else(|| {
297                    GraphStoreError::CorruptGraph(format!("missing betweenness edge {eid}"))
298                })?;
299                if let Some(target_idx) = vertex_index.get(&edge.target_id) {
300                    out_neighbors[idx].push(*target_idx);
301                }
302            }
303        }
304
305        let mut cb = vec![0.0; n];
306        for s in 0..n {
307            let mut stack: Vec<usize> = Vec::with_capacity(n);
308            let mut predecessors: Vec<Vec<usize>> = vec![Vec::new(); n];
309            let mut sigma = vec![0u64; n];
310            sigma[s] = 1;
311            let mut dist = vec![-1i64; n];
312            dist[s] = 0;
313            let mut queue: VecDeque<usize> = VecDeque::new();
314            queue.push_back(s);
315            while let Some(v) = queue.pop_front() {
316                stack.push(v);
317                for &w in &out_neighbors[v] {
318                    if dist[w] < 0 {
319                        dist[w] = dist[v].checked_add(1).ok_or_else(|| {
320                            GraphStoreError::CorruptGraph(
321                                "betweenness path distance exceeds bigint range".into(),
322                            )
323                        })?;
324                        queue.push_back(w);
325                    }
326                    if dist[w] == dist[v] + 1 {
327                        sigma[w] = sigma[w].checked_add(sigma[v]).ok_or_else(|| {
328                            GraphStoreError::CorruptGraph(
329                                "betweenness shortest-path count exceeds u64".into(),
330                            )
331                        })?;
332                        predecessors[w].push(v);
333                    }
334                }
335            }
336            let mut delta = vec![0.0; n];
337            while let Some(w) = stack.pop() {
338                for &v in &predecessors[w] {
339                    if sigma[w] > 0 {
340                        let contrib = (u64_as_f64(sigma[v], "betweenness path count")?
341                            / u64_as_f64(sigma[w], "betweenness path count")?)
342                            * (1.0 + delta[w]);
343                        delta[v] += contrib;
344                    }
345                }
346                if w != s {
347                    cb[w] += delta[w];
348                }
349            }
350        }
351
352        let normalization_count = (n - 1).checked_mul(n - 2).ok_or_else(|| {
353            GraphStoreError::CorruptGraph("betweenness normalization count overflow".into())
354        })?;
355        let normalization = usize_as_f64(normalization_count, "betweenness normalization")?;
356        if normalization > 0.0 {
357            for value in &mut cb {
358                *value /= normalization;
359            }
360        }
361        let cb: BTreeMap<VertexId, f64> = vertices
362            .iter()
363            .zip(cb)
364            .map(|(vertex_id, score)| (*vertex_id, score.clamp(0.0, 1.0)))
365            .collect();
366        build_score_result(&vertices, &cb, self.graph, &BTreeMap::new())
367    }
368}
369
370fn min_max_normalize(
371    scores: &BTreeMap<VertexId, f64>,
372    vertices: &[VertexId],
373) -> GraphStoreResult<BTreeMap<VertexId, f64>> {
374    let min_s = scores.values().copied().fold(f64::INFINITY, f64::min);
375    let max_s = scores.values().copied().fold(f64::NEG_INFINITY, f64::max);
376    if max_s - min_s > 0.0 {
377        vertices
378            .iter()
379            .map(|v| {
380                scores
381                    .get(v)
382                    .copied()
383                    .map(|score| (*v, (score - min_s) / (max_s - min_s)))
384                    .ok_or_else(|| {
385                        GraphStoreError::CorruptGraph(format!(
386                            "missing centrality score for vertex {v}"
387                        ))
388                    })
389            })
390            .collect()
391    } else {
392        Ok(vertices.iter().map(|v| (*v, 1.0)).collect())
393    }
394}
395
396fn single_vertex_result(
397    vid: VertexId,
398    score: f64,
399    vertices: &[VertexId],
400    graph: &str,
401) -> GraphStoreResult<GraphPostingList> {
402    let entry = PostingEntry::new(vid, Payload::with_score(score));
403    let mut graph_payloads: BTreeMap<DocId, GraphPayload> = BTreeMap::new();
404    graph_payloads.insert(
405        vid,
406        GraphPayload {
407            subgraph_vertices: vertices.to_vec(),
408            subgraph_edges: Vec::new(),
409            graph_name: graph.to_string(),
410            score_override: Some(score),
411        },
412    );
413    GraphPostingList::try_from_parts(
414        PostingList::from_sorted_unchecked(vec![entry]),
415        graph_payloads,
416    )
417    .map_err(Into::into)
418}
419
420fn build_score_result(
421    vertices: &[VertexId],
422    scores: &BTreeMap<VertexId, f64>,
423    graph: &str,
424    extra_fields: &BTreeMap<VertexId, BTreeMap<String, Value>>,
425) -> GraphStoreResult<GraphPostingList> {
426    let mut entries: Vec<PostingEntry> = Vec::with_capacity(vertices.len());
427    let mut graph_payloads: BTreeMap<DocId, GraphPayload> = BTreeMap::new();
428    let mut sorted = vertices.to_vec();
429    sorted.sort_unstable();
430    for vid in &sorted {
431        let score = *scores.get(vid).ok_or_else(|| {
432            GraphStoreError::CorruptGraph(format!("missing centrality score for vertex {vid}"))
433        })?;
434        let mut payload = Payload::with_score(score);
435        if let Some(fields) = extra_fields.get(vid) {
436            payload.fields = fields.clone();
437        }
438        entries.push(PostingEntry::new(*vid, payload));
439        graph_payloads.insert(
440            *vid,
441            GraphPayload {
442                subgraph_vertices: sorted.clone(),
443                subgraph_edges: Vec::new(),
444                graph_name: graph.to_string(),
445                score_override: Some(score),
446            },
447        );
448    }
449    GraphPostingList::try_from_parts(PostingList::from_sorted_unchecked(entries), graph_payloads)
450        .map_err(Into::into)
451}
uqa_graph/centrality.rs

uqa_graph/
centrality.rs
