sbol_db_postgres/repo/
pagerank.rs

1//! The object PageRank store over Postgres.
2//!
3//! Scores live in `object_pagerank`. A rebuild replaces the whole table in one
4//! transaction (`TRUNCATE` then a single `UNNEST` insert), so readers only ever
5//! see a complete ranking.
6
7use std::collections::HashMap;
8
9use async_trait::async_trait;
10use sbol_db_core::DomainError;
11use sbol_db_storage::{PageRankStore, RankRow};
12use sqlx::Row;
13
14use crate::repo::db_err;
15use crate::PgPool;
16
17#[derive(Clone)]
18pub struct PgPageRankStore {
19    pool: PgPool,
20}
21
22impl PgPageRankStore {
23    pub fn new(pool: PgPool) -> Self {
24        Self { pool }
25    }
26}
27
28#[async_trait]
29impl PageRankStore for PgPageRankStore {
30    async fn rank_of(&self, iri: &str) -> Result<Option<f64>, DomainError> {
31        let row = sqlx::query("SELECT score FROM object_pagerank WHERE iri = $1")
32            .bind(iri)
33            .fetch_optional(&self.pool)
34            .await
35            .map_err(db_err)?;
36        match row {
37            Some(r) => Ok(Some(r.try_get::<f64, _>("score").map_err(db_err)?)),
38            None => Ok(None),
39        }
40    }
41
42    async fn ranks_for(&self, iris: &[String]) -> Result<HashMap<String, f64>, DomainError> {
43        if iris.is_empty() {
44            return Ok(HashMap::new());
45        }
46        let rows =
47            sqlx::query("SELECT iri, score FROM object_pagerank WHERE iri = ANY($1::text[])")
48                .bind(iris)
49                .fetch_all(&self.pool)
50                .await
51                .map_err(db_err)?;
52        let mut out = HashMap::with_capacity(rows.len());
53        for r in rows {
54            let iri: String = r.try_get("iri").map_err(db_err)?;
55            let score: f64 = r.try_get("score").map_err(db_err)?;
56            out.insert(iri, score);
57        }
58        Ok(out)
59    }
60
61    async fn all_ranks(&self) -> Result<Vec<RankRow>, DomainError> {
62        let rows = sqlx::query("SELECT iri, score FROM object_pagerank")
63            .fetch_all(&self.pool)
64            .await
65            .map_err(db_err)?;
66        let mut out = Vec::with_capacity(rows.len());
67        for r in rows {
68            let iri: String = r.try_get("iri").map_err(db_err)?;
69            let score: f64 = r.try_get("score").map_err(db_err)?;
70            out.push(RankRow { iri, score });
71        }
72        Ok(out)
73    }
74
75    async fn replace_all_ranks(&self, ranks: Vec<RankRow>) -> Result<(), DomainError> {
76        let mut tx = self.pool.begin().await.map_err(db_err)?;
77        sqlx::query("TRUNCATE object_pagerank")
78            .execute(&mut *tx)
79            .await
80            .map_err(db_err)?;
81        if !ranks.is_empty() {
82            let iris: Vec<String> = ranks.iter().map(|r| r.iri.clone()).collect();
83            let scores: Vec<f64> = ranks.iter().map(|r| r.score).collect();
84            sqlx::query(
85                r#"
86                INSERT INTO object_pagerank (iri, score)
87                SELECT * FROM UNNEST($1::text[], $2::float8[])
88                ON CONFLICT (iri) DO UPDATE SET score = EXCLUDED.score
89                "#,
90            )
91            .bind(&iris)
92            .bind(&scores)
93            .execute(&mut *tx)
94            .await
95            .map_err(db_err)?;
96        }
97        tx.commit().await.map_err(db_err)?;
98        Ok(())
99    }
100}
sbol_db_postgres/repo/pagerank.rs

sbol_db_postgres/repo/
pagerank.rs
