sbol_db_postgres/repo/
accel.rs

1//! The SynBioHub query accelerator for Postgres: per-graph derived indexes that
2//! answer the fixed SynBioHub query templates with point lookups and range scans
3//! instead of graph-pattern evaluation.
4//!
5//! The indexes are derived from a graph's triples (via the backend-neutral
6//! [`build_accel_index`]) and maintained synchronously on the write path:
7//! [`AccelRepository::refresh_graph`] rebuilds a graph's indexes inside the
8//! write's own transaction, scanning the triples through that transaction's
9//! connection so the indexes it writes reflect the triples the same transaction
10//! wrote. Indexes and triples therefore commit together, and reads (which never
11//! rebuild) always see indexes consistent with the committed triples.
12
13use std::collections::{HashMap, HashSet};
14
15use sbol_db_core::DomainError;
16use sbol_db_storage::{
17    build_accel_index, generate_metadata_rows, generate_rows, integer, AccelSolutions,
18    AcceleratedQuery, FacetKind, Field, MetaRecord, Scope, TermValue,
19};
20use sqlx::{QueryBuilder, Row};
21
22use crate::repo::db_err;
23use crate::repo::triple::TripleRepository;
24use crate::PgPool;
25
26const FK_TYPES: i16 = 1;
27const FK_ROLES: i16 = 2;
28const FK_CREATORS: i16 = 3;
29
30const XSD_STRING: &str = "http://www.w3.org/2001/XMLSchema#string";
31
32/// Derives, maintains, and queries the accelerator indexes for the Postgres
33/// backend.
34#[derive(Clone)]
35pub struct AccelRepository {
36    pool: PgPool,
37    triples: TripleRepository,
38}
39
40impl AccelRepository {
41    pub fn new(pool: PgPool, triples: TripleRepository) -> Self {
42        Self { pool, triples }
43    }
44
45    /// Answer a recognized query from the graph's accelerator indexes, which the
46    /// write path keeps in sync with the committed triples (see
47    /// [`Self::refresh_graph`]).
48    pub async fn run(&self, query: &AcceleratedQuery) -> Result<AccelSolutions, DomainError> {
49        match query {
50            AcceleratedQuery::ObjectList {
51                graph,
52                scope,
53                projection,
54                offset,
55                limit,
56                subject_prefix,
57            } => {
58                self.object_list(
59                    graph,
60                    scope,
61                    projection,
62                    *offset,
63                    *limit,
64                    subject_prefix.as_deref(),
65                )
66                .await
67            }
68            AcceleratedQuery::Count {
69                graph,
70                scope,
71                var,
72                subject_prefix,
73            } => {
74                self.count(graph, scope, var, subject_prefix.as_deref())
75                    .await
76            }
77            AcceleratedQuery::Facet { graph, kind, var } => self.facet(graph, *kind, var).await,
78            AcceleratedQuery::ObjectMetadata {
79                graph,
80                subject,
81                projection,
82                required,
83            } => {
84                self.object_metadata(graph, subject, projection, required)
85                    .await
86            }
87        }
88    }
89
90    /// Rebuild a graph's accelerator indexes from its triples, inside the
91    /// caller's write transaction (atomic with the triple write). The triple
92    /// scan runs through `conn`, so it sees the triples the same transaction
93    /// just wrote; the rebuilt indexes are deleted and reinserted on `conn` and
94    /// commit together with the triples. Callers invoke this after every write
95    /// that changes a graph's triples.
96    pub async fn refresh_graph(
97        &self,
98        conn: &mut sqlx::PgConnection,
99        graph: &str,
100    ) -> Result<(), DomainError> {
101        let triples = self.triples.scan_graph_in_conn(conn, graph).await?;
102        let index = build_accel_index(&triples);
103
104        let sort_of: HashMap<&str, &str> = index
105            .objects
106            .iter()
107            .map(|o| (o.iri.as_str(), o.meta.sort_key()))
108            .collect();
109        let root_set: HashSet<(&str, &str)> = index
110            .root_members
111            .iter()
112            .map(|(c, m)| (c.as_str(), m.as_str()))
113            .collect();
114
115        let mut obj_iri: Vec<String> = Vec::new();
116        let mut obj_sort: Vec<String> = Vec::new();
117        let mut obj_top: Vec<bool> = Vec::new();
118        let mut obj_meta: Vec<String> = Vec::new();
119        let mut ty_type: Vec<String> = Vec::new();
120        let mut ty_iri: Vec<String> = Vec::new();
121        let mut ty_sort: Vec<String> = Vec::new();
122        let mut facet_kind: Vec<i16> = Vec::new();
123        let mut facet_value: Vec<String> = Vec::new();
124        let mut facet_seen: HashSet<(i16, &str)> = HashSet::new();
125        for obj in &index.objects {
126            let iri = obj.iri.as_str();
127            let m = &obj.meta;
128            let sort = m.sort_key();
129            obj_iri.push(iri.to_owned());
130            obj_sort.push(sort.to_owned());
131            obj_top.push(m.top_level);
132            obj_meta.push(serde_json::to_string(m).map_err(db_err)?);
133            for t in &m.types {
134                ty_type.push(t.clone());
135                ty_iri.push(iri.to_owned());
136                ty_sort.push(sort.to_owned());
137            }
138            if m.top_level {
139                for t in &m.types {
140                    if facet_seen.insert((FK_TYPES, t.as_str())) {
141                        facet_kind.push(FK_TYPES);
142                        facet_value.push(t.clone());
143                    }
144                }
145                for r in &m.roles {
146                    if facet_seen.insert((FK_ROLES, r.as_str())) {
147                        facet_kind.push(FK_ROLES);
148                        facet_value.push(r.clone());
149                    }
150                }
151                for c in &m.creators {
152                    if facet_seen.insert((FK_CREATORS, c.as_str())) {
153                        facet_kind.push(FK_CREATORS);
154                        facet_value.push(c.clone());
155                    }
156                }
157            }
158        }
159
160        let mut mem_coll: Vec<String> = Vec::new();
161        let mut mem_iri: Vec<String> = Vec::new();
162        let mut mem_sort: Vec<String> = Vec::new();
163        let mut mem_root: Vec<bool> = Vec::new();
164        for (collection, member) in &index.members {
165            mem_coll.push(collection.clone());
166            mem_iri.push(member.clone());
167            mem_sort.push(
168                sort_of
169                    .get(member.as_str())
170                    .copied()
171                    .unwrap_or("")
172                    .to_owned(),
173            );
174            mem_root.push(root_set.contains(&(collection.as_str(), member.as_str())));
175        }
176
177        for table in ["accel_object", "accel_type", "accel_member", "accel_facet"] {
178            sqlx::query(&format!("DELETE FROM {table} WHERE graph_iri = $1"))
179                .bind(graph)
180                .execute(&mut *conn)
181                .await
182                .map_err(db_err)?;
183        }
184        if !obj_iri.is_empty() {
185            sqlx::query(
186                "INSERT INTO accel_object (graph_iri, iri, sort_key, top_level, meta)
187                 SELECT $1, iri, sort_key, top_level, meta
188                 FROM UNNEST($2::text[], $3::text[], $4::bool[], $5::text[])
189                    AS u(iri, sort_key, top_level, meta)",
190            )
191            .bind(graph)
192            .bind(&obj_iri)
193            .bind(&obj_sort)
194            .bind(&obj_top)
195            .bind(&obj_meta)
196            .execute(&mut *conn)
197            .await
198            .map_err(db_err)?;
199        }
200        if !ty_iri.is_empty() {
201            sqlx::query(
202                "INSERT INTO accel_type (graph_iri, type_iri, iri, sort_key)
203                 SELECT $1, type_iri, iri, sort_key
204                 FROM UNNEST($2::text[], $3::text[], $4::text[])
205                    AS u(type_iri, iri, sort_key)
206                 ON CONFLICT DO NOTHING",
207            )
208            .bind(graph)
209            .bind(&ty_type)
210            .bind(&ty_iri)
211            .bind(&ty_sort)
212            .execute(&mut *conn)
213            .await
214            .map_err(db_err)?;
215        }
216        if !mem_iri.is_empty() {
217            sqlx::query(
218                "INSERT INTO accel_member (graph_iri, collection_iri, member_iri, sort_key, is_root)
219                 SELECT $1, collection_iri, member_iri, sort_key, is_root
220                 FROM UNNEST($2::text[], $3::text[], $4::text[], $5::bool[])
221                    AS u(collection_iri, member_iri, sort_key, is_root)
222                 ON CONFLICT DO NOTHING",
223            )
224            .bind(graph)
225            .bind(&mem_coll)
226            .bind(&mem_iri)
227            .bind(&mem_sort)
228            .bind(&mem_root)
229            .execute(&mut *conn)
230            .await
231            .map_err(db_err)?;
232        }
233        if !facet_value.is_empty() {
234            sqlx::query(
235                "INSERT INTO accel_facet (graph_iri, kind, value)
236                 SELECT $1, kind, value
237                 FROM UNNEST($2::smallint[], $3::text[]) AS u(kind, value)
238                 ON CONFLICT DO NOTHING",
239            )
240            .bind(graph)
241            .bind(&facet_kind)
242            .bind(&facet_value)
243            .execute(&mut *conn)
244            .await
245            .map_err(db_err)?;
246        }
247        Ok(())
248    }
249
250    async fn object_list(
251        &self,
252        graph: &str,
253        scope: &Scope,
254        projection: &[(String, sbol_db_storage::Field)],
255        offset: usize,
256        limit: Option<usize>,
257        subject_prefix: Option<&str>,
258    ) -> Result<AccelSolutions, DomainError> {
259        let candidates = self
260            .scope_candidates(graph, scope, subject_prefix, offset, limit)
261            .await?;
262
263        let vars: Vec<String> = projection.iter().map(|(v, _)| v.clone()).collect();
264        // Generate rows in displayId order, dedup as we go, and stop once we have
265        // enough for the requested page (objects are visited in order, and each
266        // yields at least one row, so a row's position is fixed by its object).
267        let target = limit.map(|l| offset + l);
268        let mut seen = HashSet::new();
269        let mut rows: Vec<Vec<Option<TermValue>>> = Vec::new();
270        let mut object_rows = Vec::new();
271        for (iri, meta) in &candidates {
272            object_rows.clear();
273            generate_rows(iri, meta, projection, &mut object_rows);
274            for row in object_rows.drain(..) {
275                if seen.insert(format!("{row:?}")) {
276                    rows.push(row);
277                }
278            }
279            if target.is_some_and(|t| rows.len() >= t) {
280                break;
281            }
282        }
283        let rows = rows
284            .into_iter()
285            .skip(offset)
286            .take(limit.unwrap_or(usize::MAX))
287            .collect();
288        Ok(AccelSolutions { vars, rows })
289    }
290
291    /// The objects in `scope`, in displayId order, each with its metadata. A
292    /// member with no metadata (an external reference) is returned with a default
293    /// (empty) record, so it still yields a subject-only row. When a row limit is
294    /// set, the scan is capped to `offset + limit` objects: every object yields at
295    /// least one distinct row, so that many objects suffice to fill the page.
296    async fn scope_candidates(
297        &self,
298        graph: &str,
299        scope: &Scope,
300        subject_prefix: Option<&str>,
301        offset: usize,
302        limit: Option<usize>,
303    ) -> Result<Vec<(String, MetaRecord)>, DomainError> {
304        let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new("");
305        match scope {
306            Scope::TopLevel => {
307                qb.push("SELECT iri, meta FROM accel_object WHERE graph_iri = ");
308                qb.push_bind(graph.to_owned());
309                qb.push(" AND top_level");
310                if let Some(p) = subject_prefix {
311                    qb.push(" AND iri LIKE ").push_bind(like_prefix(p));
312                }
313                qb.push(" ORDER BY sort_key COLLATE \"C\", iri COLLATE \"C\"");
314            }
315            Scope::ByType(t) => {
316                qb.push(
317                    "SELECT ty.iri AS iri, o.meta AS meta FROM accel_type ty \
318                     LEFT JOIN accel_object o \
319                       ON o.graph_iri = ty.graph_iri AND o.iri = ty.iri \
320                     WHERE ty.graph_iri = ",
321                );
322                qb.push_bind(graph.to_owned());
323                qb.push(" AND ty.type_iri = ").push_bind(t.clone());
324                if let Some(p) = subject_prefix {
325                    qb.push(" AND ty.iri LIKE ").push_bind(like_prefix(p));
326                }
327                qb.push(" ORDER BY ty.sort_key COLLATE \"C\", ty.iri COLLATE \"C\"");
328            }
329            Scope::Collection {
330                collection,
331                root_only,
332            } => {
333                qb.push(
334                    "SELECT m.member_iri AS iri, o.meta AS meta FROM accel_member m \
335                     LEFT JOIN accel_object o \
336                       ON o.graph_iri = m.graph_iri AND o.iri = m.member_iri \
337                     WHERE m.graph_iri = ",
338                );
339                qb.push_bind(graph.to_owned());
340                qb.push(" AND m.collection_iri = ")
341                    .push_bind(collection.clone());
342                if *root_only {
343                    qb.push(" AND m.is_root");
344                }
345                if let Some(p) = subject_prefix {
346                    qb.push(" AND m.member_iri LIKE ").push_bind(like_prefix(p));
347                }
348                qb.push(" ORDER BY m.sort_key COLLATE \"C\", m.member_iri COLLATE \"C\"");
349            }
350        }
351        if let Some(l) = limit {
352            qb.push(" LIMIT ").push_bind((offset + l) as i64);
353        }
354
355        let rows = qb.build().fetch_all(&self.pool).await.map_err(db_err)?;
356        let mut out = Vec::with_capacity(rows.len());
357        for row in rows {
358            let iri: String = row.try_get("iri").map_err(db_err)?;
359            let meta_json: Option<String> = row.try_get("meta").map_err(db_err)?;
360            let meta = match meta_json {
361                Some(j) => serde_json::from_str(&j).map_err(db_err)?,
362                None => MetaRecord::default(),
363            };
364            out.push((iri, meta));
365        }
366        Ok(out)
367    }
368
369    /// Fetch one object's metadata projection by primary key. A missing or
370    /// metadata-less object yields no rows (the required title cannot bind).
371    async fn object_metadata(
372        &self,
373        graph: &str,
374        subject: &str,
375        projection: &[(String, Field)],
376        required: &[bool],
377    ) -> Result<AccelSolutions, DomainError> {
378        let vars: Vec<String> = projection.iter().map(|(v, _)| v.clone()).collect();
379        let meta_json: Option<String> =
380            sqlx::query_scalar("SELECT meta FROM accel_object WHERE graph_iri = $1 AND iri = $2")
381                .bind(graph)
382                .bind(subject)
383                .fetch_optional(&self.pool)
384                .await
385                .map_err(db_err)?;
386        let mut rows = Vec::new();
387        if let Some(json) = meta_json {
388            let meta: MetaRecord = serde_json::from_str(&json).map_err(db_err)?;
389            generate_metadata_rows(subject, &meta, projection, required, &mut rows);
390        }
391        Ok(AccelSolutions { vars, rows })
392    }
393
394    async fn count(
395        &self,
396        graph: &str,
397        scope: &Scope,
398        var: &str,
399        subject_prefix: Option<&str>,
400    ) -> Result<AccelSolutions, DomainError> {
401        let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new("");
402        match scope {
403            Scope::TopLevel => {
404                qb.push("SELECT COUNT(*) FROM accel_object WHERE graph_iri = ");
405                qb.push_bind(graph.to_owned());
406                qb.push(" AND top_level");
407                if let Some(p) = subject_prefix {
408                    qb.push(" AND iri LIKE ").push_bind(like_prefix(p));
409                }
410            }
411            Scope::ByType(t) => {
412                qb.push("SELECT COUNT(*) FROM accel_type WHERE graph_iri = ");
413                qb.push_bind(graph.to_owned());
414                qb.push(" AND type_iri = ").push_bind(t.clone());
415                if let Some(p) = subject_prefix {
416                    qb.push(" AND iri LIKE ").push_bind(like_prefix(p));
417                }
418            }
419            Scope::Collection {
420                collection,
421                root_only,
422            } => {
423                qb.push("SELECT COUNT(*) FROM accel_member WHERE graph_iri = ");
424                qb.push_bind(graph.to_owned());
425                qb.push(" AND collection_iri = ")
426                    .push_bind(collection.clone());
427                if *root_only {
428                    qb.push(" AND is_root");
429                }
430                if let Some(p) = subject_prefix {
431                    qb.push(" AND member_iri LIKE ").push_bind(like_prefix(p));
432                }
433            }
434        }
435        let n: i64 = qb
436            .build_query_scalar()
437            .fetch_one(&self.pool)
438            .await
439            .map_err(db_err)?;
440        Ok(AccelSolutions {
441            vars: vec![var.to_owned()],
442            rows: vec![vec![Some(integer(n as u64))]],
443        })
444    }
445
446    async fn facet(
447        &self,
448        graph: &str,
449        kind: FacetKind,
450        var: &str,
451    ) -> Result<AccelSolutions, DomainError> {
452        let tag = match kind {
453            FacetKind::Types => FK_TYPES,
454            FacetKind::Roles => FK_ROLES,
455            FacetKind::Creators => FK_CREATORS,
456        };
457        let values = sqlx::query_scalar::<_, String>(
458            "SELECT value FROM accel_facet WHERE graph_iri = $1 AND kind = $2 \
459             ORDER BY value COLLATE \"C\"",
460        )
461        .bind(graph)
462        .bind(tag)
463        .fetch_all(&self.pool)
464        .await
465        .map_err(db_err)?;
466        let rows = values
467            .into_iter()
468            .map(|value| {
469                let term = match kind {
470                    FacetKind::Creators => TermValue::Literal {
471                        value,
472                        datatype: XSD_STRING.to_owned(),
473                        language: None,
474                    },
475                    _ => TermValue::Iri(value),
476                };
477                vec![Some(term)]
478            })
479            .collect();
480        Ok(AccelSolutions {
481            vars: vec![var.to_owned()],
482            rows,
483        })
484    }
485}
486
487/// A `LIKE` pattern matching everything that starts with `prefix`, escaping the
488/// `LIKE` metacharacters (`%`, `_`, `\`) so an IRI prefix is matched literally.
489fn like_prefix(prefix: &str) -> String {
490    let mut out = String::with_capacity(prefix.len() + 1);
491    for ch in prefix.chars() {
492        if matches!(ch, '%' | '_' | '\\') {
493            out.push('\\');
494        }
495        out.push(ch);
496    }
497    out.push('%');
498    out
499}
sbol_db_postgres/repo/accel.rs

sbol_db_postgres/repo/
accel.rs
