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::{BTreeMap, HashMap, HashSet};
14
15use sbol_db_core::DomainError;
16use sbol_db_storage::{
17    build_catalog_projection, generate_metadata_rows, generate_rows, integer,
18    merge_resource_occurrences, AccelSolutions, AcceleratedQuery, CatalogSequenceRecord,
19    CursorPage, FacetKind, Field, MetaRecord, ResourceOccurrence, ResourceQuery, ResourceRecord,
20    Scope, SequenceQuery, TermValue,
21};
22use sqlx::{QueryBuilder, Row};
23
24use crate::repo::db_err;
25use crate::repo::triple::TripleRepository;
26use crate::PgPool;
27
28const FK_TYPES: i16 = 1;
29const FK_ROLES: i16 = 2;
30const FK_CREATORS: i16 = 3;
31
32const XSD_STRING: &str = "http://www.w3.org/2001/XMLSchema#string";
33/// Bound the Postgres bind message while rebuilding production-scale graphs.
34/// A graph can derive millions of type/member rows; sending every array in one
35/// `UNNEST` risks PostgreSQL's one-gigabyte protocol/message ceiling.
36const INSERT_CHUNK: usize = 5_000;
37
38/// Derives, maintains, and queries the accelerator indexes for the Postgres
39/// backend.
40#[derive(Clone)]
41pub struct AccelRepository {
42    pool: PgPool,
43    triples: TripleRepository,
44}
45
46impl AccelRepository {
47    pub fn new(pool: PgPool, triples: TripleRepository) -> Self {
48        Self { pool, triples }
49    }
50
51    /// Answer a recognized query from the graph's accelerator indexes, which the
52    /// write path keeps in sync with the committed triples (see
53    /// [`Self::refresh_graph`]).
54    pub async fn run(&self, query: &AcceleratedQuery) -> Result<AccelSolutions, DomainError> {
55        match query {
56            AcceleratedQuery::ObjectList {
57                graph,
58                scope,
59                projection,
60                offset,
61                limit,
62                subject_prefix,
63            } => {
64                self.object_list(
65                    graph,
66                    scope,
67                    projection,
68                    *offset,
69                    *limit,
70                    subject_prefix.as_deref(),
71                )
72                .await
73            }
74            AcceleratedQuery::Count {
75                graph,
76                scope,
77                var,
78                subject_prefix,
79            } => {
80                self.count(graph, scope, var, subject_prefix.as_deref())
81                    .await
82            }
83            AcceleratedQuery::Facet { graph, kind, var } => self.facet(graph, *kind, var).await,
84            AcceleratedQuery::FacetCounts {
85                graph,
86                kind,
87                value_var,
88                count_var,
89            } => self.facet_counts(graph, *kind, value_var, count_var).await,
90            AcceleratedQuery::ObjectMetadata {
91                graph,
92                subject,
93                projection,
94                required,
95            } => {
96                self.object_metadata(graph, subject, projection, required)
97                    .await
98            }
99        }
100    }
101
102    /// Every occurrence of one resource, ordered by graph IRI. The graph-local
103    /// accelerator metadata is the universal RDF-derived resource catalog.
104    pub async fn resource_occurrences(
105        &self,
106        iri: &str,
107    ) -> Result<Vec<ResourceOccurrence>, DomainError> {
108        let rows = sqlx::query(
109            "SELECT graph_iri, meta FROM accel_object WHERE iri = $1 ORDER BY graph_iri",
110        )
111        .bind(iri)
112        .fetch_all(&self.pool)
113        .await
114        .map_err(db_err)?;
115        rows.into_iter()
116            .map(|row| {
117                let graph_iri: String = row.try_get("graph_iri").map_err(db_err)?;
118                let meta: String = row.try_get("meta").map_err(db_err)?;
119                Ok(ResourceOccurrence {
120                    graph_iri,
121                    resource_iri: iri.to_owned(),
122                    meta: serde_json::from_str(&meta).map_err(db_err)?,
123                })
124            })
125            .collect()
126    }
127
128    pub async fn resource(&self, iri: &str) -> Result<Option<ResourceRecord>, DomainError> {
129        let occurrences = self.resource_occurrences(iri).await?;
130        Ok(merge_resource_occurrences(iri, &occurrences))
131    }
132
133    /// Globally keyset-page unique resource IRIs, applying filters before the
134    /// limit. Only the returned page is expanded to occurrence metadata.
135    pub async fn resources(
136        &self,
137        query: &ResourceQuery,
138    ) -> Result<CursorPage<ResourceRecord>, DomainError> {
139        let limit = query.limit.clamp(1, 500) as i64;
140        let mut qb: QueryBuilder<sqlx::Postgres> =
141            QueryBuilder::new("SELECT DISTINCT o.iri FROM accel_object o WHERE TRUE");
142        if let Some(after) = &query.after {
143            qb.push(" AND o.iri > ").push_bind(after.clone());
144        }
145        if let Some(graph) = &query.graph_iri {
146            qb.push(" AND o.graph_iri = ").push_bind(graph.clone());
147        }
148        if let Some(class) = &query.class {
149            qb.push(
150                " AND EXISTS (SELECT 1 FROM accel_type ty \
151                 WHERE ty.graph_iri = o.graph_iri AND ty.iri = o.iri AND ty.type_iri = ",
152            )
153            .push_bind(class.clone())
154            .push(")");
155        }
156        if let Some(role) = &query.role {
157            qb.push(
158                " AND EXISTS (SELECT 1 FROM accel_role r \
159                 WHERE r.graph_iri = o.graph_iri AND r.iri = o.iri AND r.role_iri = ",
160            )
161            .push_bind(role.clone())
162            .push(")");
163        }
164        if let Some(text) = query.text.as_deref().filter(|text| !text.is_empty()) {
165            let needle = format!("%{text}%");
166            qb.push(" AND (o.iri ILIKE ")
167                .push_bind(needle.clone())
168                .push(" OR o.meta ILIKE ")
169                .push_bind(needle)
170                .push(")");
171        }
172        qb.push(" ORDER BY o.iri LIMIT ").push_bind(limit + 1);
173        let mut iris: Vec<String> = qb
174            .build_query_scalar()
175            .fetch_all(&self.pool)
176            .await
177            .map_err(db_err)?;
178        let has_more = iris.len() > limit as usize;
179        iris.truncate(limit as usize);
180        let next_cursor = has_more.then(|| iris.last().cloned()).flatten();
181        let items = self.resources_for_iris(&iris).await?;
182        Ok(CursorPage { items, next_cursor })
183    }
184
185    /// Expand a page of resource identities in one query. Keeping this separate
186    /// from the keyset query avoids one database round trip per list item.
187    pub async fn resources_for_iris(
188        &self,
189        iris: &[String],
190    ) -> Result<Vec<ResourceRecord>, DomainError> {
191        if iris.is_empty() {
192            return Ok(Vec::new());
193        }
194        let mut qb: QueryBuilder<sqlx::Postgres> =
195            QueryBuilder::new("SELECT iri, graph_iri, meta FROM accel_object WHERE iri IN (");
196        {
197            let mut values = qb.separated(", ");
198            for iri in iris {
199                values.push_bind(iri);
200            }
201        }
202        qb.push(") ORDER BY iri, graph_iri");
203        let rows = qb.build().fetch_all(&self.pool).await.map_err(db_err)?;
204        let mut by_iri: BTreeMap<String, Vec<ResourceOccurrence>> = BTreeMap::new();
205        for row in rows {
206            let iri: String = row.try_get("iri").map_err(db_err)?;
207            let graph_iri: String = row.try_get("graph_iri").map_err(db_err)?;
208            let meta: String = row.try_get("meta").map_err(db_err)?;
209            by_iri
210                .entry(iri.clone())
211                .or_default()
212                .push(ResourceOccurrence {
213                    graph_iri,
214                    resource_iri: iri,
215                    meta: serde_json::from_str(&meta).map_err(db_err)?,
216                });
217        }
218        Ok(iris
219            .iter()
220            .filter_map(|iri| {
221                by_iri
222                    .get(iri)
223                    .and_then(|occurrences| merge_resource_occurrences(iri, occurrences))
224            })
225            .collect())
226    }
227
228    pub async fn sequence(&self, iri: &str) -> Result<Option<CatalogSequenceRecord>, DomainError> {
229        let graph_count: i64 = sqlx::query_scalar(
230            "SELECT count(DISTINCT graph_iri) FROM accel_type WHERE iri = $1 \
231             AND type_iri IN ('http://sbols.org/v2#Sequence', 'http://sbols.org/v3#Sequence')",
232        )
233        .bind(iri)
234        .fetch_one(&self.pool)
235        .await
236        .map_err(db_err)?;
237        if graph_count == 0 {
238            return Ok(None);
239        }
240        let rows = sqlx::query(
241            r#"
242            SELECT predicate_iri::text AS predicate_iri,
243                   object_iri::text AS object_iri, object_literal
244            FROM sbol_triples
245            WHERE subject_iri = $1
246              AND predicate_iri IN (
247                'http://sbols.org/v2#elements', 'http://sbols.org/v3#elements',
248                'http://sbols.org/v2#encoding', 'http://sbols.org/v3#encoding'
249              )
250            ORDER BY graph_iri NULLS LAST, id
251            "#,
252        )
253        .bind(iri)
254        .fetch_all(&self.pool)
255        .await
256        .map_err(db_err)?;
257        let mut elements = None;
258        let mut encoding_iri = None;
259        for row in rows {
260            let predicate: String = row.try_get("predicate_iri").map_err(db_err)?;
261            if predicate.ends_with("#elements") && elements.is_none() {
262                elements = row.try_get("object_literal").map_err(db_err)?;
263            } else if predicate.ends_with("#encoding") && encoding_iri.is_none() {
264                encoding_iri = row.try_get("object_iri").map_err(db_err)?;
265            }
266        }
267        Ok(Some(CatalogSequenceRecord {
268            iri: iri.to_owned(),
269            graph_count: graph_count as u64,
270            alphabet: catalog_alphabet(encoding_iri.as_deref()),
271            encoding_iri,
272            elements,
273        }))
274    }
275
276    pub async fn sequences(
277        &self,
278        query: &SequenceQuery,
279    ) -> Result<CursorPage<CatalogSequenceRecord>, DomainError> {
280        let limit = query.limit.clamp(1, 500) as i64;
281        let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(
282            "SELECT DISTINCT ty.iri FROM accel_type ty WHERE ty.type_iri IN \
283             ('http://sbols.org/v2#Sequence', 'http://sbols.org/v3#Sequence')",
284        );
285        if let Some(after) = &query.after {
286            qb.push(" AND ty.iri > ").push_bind(after.clone());
287        }
288        if let Some(text) = query.text.as_deref().filter(|value| !value.is_empty()) {
289            qb.push(
290                " AND EXISTS (SELECT 1 FROM accel_object o WHERE o.iri = ty.iri \
291                     AND (o.iri ILIKE ",
292            )
293            .push_bind(format!("%{text}%"))
294            .push(" OR o.meta ILIKE ")
295            .push_bind(format!("%{text}%"))
296            .push("))");
297        }
298        qb.push(" ORDER BY ty.iri LIMIT ").push_bind(limit + 1);
299        let mut iris: Vec<String> = qb
300            .build_query_scalar()
301            .fetch_all(&self.pool)
302            .await
303            .map_err(db_err)?;
304        let has_more = iris.len() > limit as usize;
305        iris.truncate(limit as usize);
306        let next_cursor = has_more.then(|| iris.last().cloned()).flatten();
307        let items = self.sequences_for_iris(&iris).await?;
308        Ok(CursorPage { items, next_cursor })
309    }
310
311    /// Expand sequence counts and values with two set queries, independent of
312    /// page size. The first value in graph/id order remains the representative.
313    async fn sequences_for_iris(
314        &self,
315        iris: &[String],
316    ) -> Result<Vec<CatalogSequenceRecord>, DomainError> {
317        if iris.is_empty() {
318            return Ok(Vec::new());
319        }
320        let mut count_qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(
321            "SELECT iri, count(DISTINCT graph_iri) AS graph_count FROM accel_type \
322             WHERE type_iri IN ('http://sbols.org/v2#Sequence', 'http://sbols.org/v3#Sequence') \
323             AND iri IN (",
324        );
325        {
326            let mut values = count_qb.separated(", ");
327            for iri in iris {
328                values.push_bind(iri);
329            }
330        }
331        count_qb.push(") GROUP BY iri");
332        let count_rows = count_qb
333            .build()
334            .fetch_all(&self.pool)
335            .await
336            .map_err(db_err)?;
337        let mut records: HashMap<String, (u64, Option<String>, Option<String>)> = HashMap::new();
338        for row in count_rows {
339            let iri: String = row.try_get("iri").map_err(db_err)?;
340            let graph_count: i64 = row.try_get("graph_count").map_err(db_err)?;
341            records.insert(iri, (graph_count as u64, None, None));
342        }
343
344        let mut value_qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(
345            "SELECT subject_iri::text AS subject_iri, predicate_iri::text AS predicate_iri, \
346                    object_iri::text AS object_iri, object_literal \
347             FROM sbol_triples WHERE predicate_iri IN (\
348               'http://sbols.org/v2#elements', 'http://sbols.org/v3#elements', \
349               'http://sbols.org/v2#encoding', 'http://sbols.org/v3#encoding'\
350             ) AND subject_iri IN (",
351        );
352        {
353            let mut values = value_qb.separated(", ");
354            for iri in iris {
355                values.push_bind(iri);
356            }
357        }
358        value_qb.push(") ORDER BY subject_iri, graph_iri NULLS LAST, id");
359        let value_rows = value_qb
360            .build()
361            .fetch_all(&self.pool)
362            .await
363            .map_err(db_err)?;
364        for row in value_rows {
365            let iri: String = row.try_get("subject_iri").map_err(db_err)?;
366            let predicate: String = row.try_get("predicate_iri").map_err(db_err)?;
367            let Some((_, elements, encoding)) = records.get_mut(&iri) else {
368                continue;
369            };
370            if predicate.ends_with("#elements") && elements.is_none() {
371                *elements = row.try_get("object_literal").map_err(db_err)?;
372            } else if predicate.ends_with("#encoding") && encoding.is_none() {
373                *encoding = row.try_get("object_iri").map_err(db_err)?;
374            }
375        }
376
377        Ok(iris
378            .iter()
379            .filter_map(|iri| {
380                records
381                    .remove(iri)
382                    .map(
383                        |(graph_count, elements, encoding_iri)| CatalogSequenceRecord {
384                            iri: iri.clone(),
385                            graph_count,
386                            alphabet: catalog_alphabet(encoding_iri.as_deref()),
387                            encoding_iri,
388                            elements,
389                        },
390                    )
391            })
392            .collect())
393    }
394
395    /// Rebuild a graph's accelerator indexes from its triples, inside the
396    /// caller's write transaction (atomic with the triple write). The triple
397    /// scan runs through `conn`, so it sees the triples the same transaction
398    /// just wrote; the rebuilt indexes are deleted and reinserted on `conn` and
399    /// commit together with the triples. Callers invoke this after every write
400    /// that changes a graph's triples.
401    pub async fn refresh_graph(
402        &self,
403        conn: &mut sqlx::PgConnection,
404        graph: &str,
405    ) -> Result<(), DomainError> {
406        let triples = self.triples.scan_graph_in_conn(conn, graph).await?;
407        let index = build_catalog_projection(&triples);
408
409        let sort_of: HashMap<&str, &str> = index
410            .objects
411            .iter()
412            .map(|o| (o.iri.as_str(), o.meta.sort_key()))
413            .collect();
414        let root_set: HashSet<(&str, &str)> = index
415            .root_members
416            .iter()
417            .map(|(c, m)| (c.as_str(), m.as_str()))
418            .collect();
419
420        let mut obj_iri: Vec<String> = Vec::new();
421        let mut obj_sort: Vec<String> = Vec::new();
422        let mut obj_top: Vec<bool> = Vec::new();
423        let mut obj_meta: Vec<String> = Vec::new();
424        let mut ty_type: Vec<String> = Vec::new();
425        let mut ty_iri: Vec<String> = Vec::new();
426        let mut ty_sort: Vec<String> = Vec::new();
427        let mut role_role: Vec<String> = Vec::new();
428        let mut role_iri: Vec<String> = Vec::new();
429        let mut role_sort: Vec<String> = Vec::new();
430        let mut facet_counts: HashMap<(i16, String), i64> = HashMap::new();
431        for obj in &index.objects {
432            let iri = obj.iri.as_str();
433            let m = &obj.meta;
434            let sort = m.sort_key();
435            obj_iri.push(iri.to_owned());
436            obj_sort.push(sort.to_owned());
437            obj_top.push(m.top_level);
438            obj_meta.push(serde_json::to_string(m).map_err(db_err)?);
439            for t in &m.types {
440                ty_type.push(t.clone());
441                ty_iri.push(iri.to_owned());
442                ty_sort.push(sort.to_owned());
443            }
444            for role in &m.roles {
445                role_role.push(role.clone());
446                role_iri.push(iri.to_owned());
447                role_sort.push(sort.to_owned());
448            }
449            if m.top_level {
450                let mut object_facets: HashSet<(i16, &str)> = HashSet::new();
451                for t in &m.types {
452                    object_facets.insert((FK_TYPES, t.as_str()));
453                }
454                for r in &m.roles {
455                    object_facets.insert((FK_ROLES, r.as_str()));
456                }
457                for c in &m.creators {
458                    object_facets.insert((FK_CREATORS, c.as_str()));
459                }
460                for (kind, value) in object_facets {
461                    *facet_counts.entry((kind, value.to_owned())).or_default() += 1;
462                }
463            }
464        }
465        let mut facets: Vec<(i16, String, i64)> = facet_counts
466            .into_iter()
467            .map(|((kind, value), count)| (kind, value, count))
468            .collect();
469        facets.sort();
470        let facet_kind: Vec<i16> = facets.iter().map(|(kind, _, _)| *kind).collect();
471        let facet_value: Vec<String> = facets.iter().map(|(_, value, _)| value.clone()).collect();
472        let facet_count: Vec<i64> = facets.iter().map(|(_, _, count)| *count).collect();
473
474        let mut mem_coll: Vec<String> = Vec::new();
475        let mut mem_iri: Vec<String> = Vec::new();
476        let mut mem_sort: Vec<String> = Vec::new();
477        let mut mem_root: Vec<bool> = Vec::new();
478        for (collection, member) in &index.members {
479            mem_coll.push(collection.clone());
480            mem_iri.push(member.clone());
481            mem_sort.push(
482                sort_of
483                    .get(member.as_str())
484                    .copied()
485                    .unwrap_or("")
486                    .to_owned(),
487            );
488            mem_root.push(root_set.contains(&(collection.as_str(), member.as_str())));
489        }
490
491        for table in [
492            "accel_object",
493            "accel_type",
494            "accel_role",
495            "accel_member",
496            "accel_facet",
497        ] {
498            sqlx::query(&format!("DELETE FROM {table} WHERE graph_iri = $1"))
499                .bind(graph)
500                .execute(&mut *conn)
501                .await
502                .map_err(db_err)?;
503        }
504        for start in (0..obj_iri.len()).step_by(INSERT_CHUNK) {
505            let end = (start + INSERT_CHUNK).min(obj_iri.len());
506            sqlx::query(
507                "INSERT INTO accel_object (graph_iri, iri, sort_key, top_level, meta)
508                 SELECT $1, iri, sort_key, top_level, meta
509                 FROM UNNEST($2::text[], $3::text[], $4::bool[], $5::text[])
510                    AS u(iri, sort_key, top_level, meta)",
511            )
512            .bind(graph)
513            .bind(&obj_iri[start..end])
514            .bind(&obj_sort[start..end])
515            .bind(&obj_top[start..end])
516            .bind(&obj_meta[start..end])
517            .execute(&mut *conn)
518            .await
519            .map_err(db_err)?;
520        }
521        for start in (0..ty_iri.len()).step_by(INSERT_CHUNK) {
522            let end = (start + INSERT_CHUNK).min(ty_iri.len());
523            sqlx::query(
524                "INSERT INTO accel_type (graph_iri, type_iri, iri, sort_key)
525                 SELECT $1, type_iri, iri, sort_key
526                 FROM UNNEST($2::text[], $3::text[], $4::text[])
527                    AS u(type_iri, iri, sort_key)
528                 ON CONFLICT DO NOTHING",
529            )
530            .bind(graph)
531            .bind(&ty_type[start..end])
532            .bind(&ty_iri[start..end])
533            .bind(&ty_sort[start..end])
534            .execute(&mut *conn)
535            .await
536            .map_err(db_err)?;
537        }
538        for start in (0..role_iri.len()).step_by(INSERT_CHUNK) {
539            let end = (start + INSERT_CHUNK).min(role_iri.len());
540            sqlx::query(
541                "INSERT INTO accel_role (graph_iri, role_iri, iri, sort_key)
542                 SELECT $1, role_iri, iri, sort_key
543                 FROM UNNEST($2::text[], $3::text[], $4::text[])
544                    AS u(role_iri, iri, sort_key)
545                 ON CONFLICT DO NOTHING",
546            )
547            .bind(graph)
548            .bind(&role_role[start..end])
549            .bind(&role_iri[start..end])
550            .bind(&role_sort[start..end])
551            .execute(&mut *conn)
552            .await
553            .map_err(db_err)?;
554        }
555        for start in (0..mem_iri.len()).step_by(INSERT_CHUNK) {
556            let end = (start + INSERT_CHUNK).min(mem_iri.len());
557            sqlx::query(
558                "INSERT INTO accel_member (graph_iri, collection_iri, member_iri, sort_key, is_root)
559                 SELECT $1, collection_iri, member_iri, sort_key, is_root
560                 FROM UNNEST($2::text[], $3::text[], $4::text[], $5::bool[])
561                    AS u(collection_iri, member_iri, sort_key, is_root)
562                 ON CONFLICT DO NOTHING",
563            )
564            .bind(graph)
565            .bind(&mem_coll[start..end])
566            .bind(&mem_iri[start..end])
567            .bind(&mem_sort[start..end])
568            .bind(&mem_root[start..end])
569            .execute(&mut *conn)
570            .await
571            .map_err(db_err)?;
572        }
573        for start in (0..facet_value.len()).step_by(INSERT_CHUNK) {
574            let end = (start + INSERT_CHUNK).min(facet_value.len());
575            sqlx::query(
576                "INSERT INTO accel_facet (graph_iri, kind, value, subject_count)
577                 SELECT $1, kind, value, subject_count
578                 FROM UNNEST($2::smallint[], $3::text[], $4::bigint[])
579                    AS u(kind, value, subject_count)
580                 ON CONFLICT DO NOTHING",
581            )
582            .bind(graph)
583            .bind(&facet_kind[start..end])
584            .bind(&facet_value[start..end])
585            .bind(&facet_count[start..end])
586            .execute(&mut *conn)
587            .await
588            .map_err(db_err)?;
589        }
590        Ok(())
591    }
592
593    async fn object_list(
594        &self,
595        graph: &str,
596        scope: &Scope,
597        projection: &[(String, sbol_db_storage::Field)],
598        offset: usize,
599        limit: Option<usize>,
600        subject_prefix: Option<&str>,
601    ) -> Result<AccelSolutions, DomainError> {
602        let candidates = self
603            .scope_candidates(graph, scope, subject_prefix, offset, limit)
604            .await?;
605
606        let vars: Vec<String> = projection.iter().map(|(v, _)| v.clone()).collect();
607        // Generate rows in displayId order, dedup as we go, and stop once we have
608        // enough for the requested page (objects are visited in order, and each
609        // yields at least one row, so a row's position is fixed by its object).
610        let target = limit.map(|l| offset + l);
611        let mut seen = HashSet::new();
612        let mut rows: Vec<Vec<Option<TermValue>>> = Vec::new();
613        let mut object_rows = Vec::new();
614        for (iri, meta) in &candidates {
615            object_rows.clear();
616            generate_rows(iri, meta, projection, &mut object_rows);
617            for row in object_rows.drain(..) {
618                if seen.insert(format!("{row:?}")) {
619                    rows.push(row);
620                }
621            }
622            if target.is_some_and(|t| rows.len() >= t) {
623                break;
624            }
625        }
626        let rows = rows
627            .into_iter()
628            .skip(offset)
629            .take(limit.unwrap_or(usize::MAX))
630            .collect();
631        Ok(AccelSolutions { vars, rows })
632    }
633
634    /// The objects in `scope`, in displayId order, each with its metadata. A
635    /// member with no metadata (an external reference) is returned with a default
636    /// (empty) record, so it still yields a subject-only row. When a row limit is
637    /// set, the scan is capped to `offset + limit` objects: every object yields at
638    /// least one distinct row, so that many objects suffice to fill the page.
639    async fn scope_candidates(
640        &self,
641        graph: &str,
642        scope: &Scope,
643        subject_prefix: Option<&str>,
644        offset: usize,
645        limit: Option<usize>,
646    ) -> Result<Vec<(String, MetaRecord)>, DomainError> {
647        let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new("");
648        match scope {
649            Scope::TopLevel => {
650                qb.push("SELECT iri, meta FROM accel_object WHERE graph_iri = ");
651                qb.push_bind(graph.to_owned());
652                qb.push(" AND top_level");
653                if let Some(p) = subject_prefix {
654                    qb.push(" AND iri LIKE ").push_bind(like_prefix(p));
655                }
656                qb.push(
657                    " ORDER BY left(sort_key, 128) COLLATE \"C\", sort_key COLLATE \"C\", \
658                              left(iri, 128) COLLATE \"C\", iri COLLATE \"C\"",
659                );
660            }
661            Scope::ByType(t) => {
662                qb.push(
663                    "SELECT ty.iri AS iri, o.meta AS meta FROM accel_type ty \
664                     LEFT JOIN accel_object o \
665                       ON o.graph_iri = ty.graph_iri AND o.iri = ty.iri \
666                     WHERE ty.graph_iri = ",
667                );
668                qb.push_bind(graph.to_owned());
669                qb.push(" AND ty.type_iri = ").push_bind(t.clone());
670                if let Some(p) = subject_prefix {
671                    qb.push(" AND ty.iri LIKE ").push_bind(like_prefix(p));
672                }
673                qb.push(
674                    " ORDER BY left(ty.sort_key, 128) COLLATE \"C\", ty.sort_key COLLATE \"C\", \
675                              left(ty.iri, 128) COLLATE \"C\", ty.iri COLLATE \"C\"",
676                );
677            }
678            Scope::TopLevelByType(t) => {
679                qb.push(
680                    "SELECT ty.iri AS iri, o.meta AS meta FROM accel_type ty \
681                     JOIN accel_object o \
682                       ON o.graph_iri = ty.graph_iri AND o.iri = ty.iri \
683                     WHERE ty.graph_iri = ",
684                );
685                qb.push_bind(graph.to_owned());
686                qb.push(" AND ty.type_iri = ").push_bind(t.clone());
687                qb.push(" AND o.top_level");
688                if let Some(p) = subject_prefix {
689                    qb.push(" AND ty.iri LIKE ").push_bind(like_prefix(p));
690                }
691                qb.push(
692                    " ORDER BY left(ty.sort_key, 128) COLLATE \"C\", ty.sort_key COLLATE \"C\", \
693                              left(ty.iri, 128) COLLATE \"C\", ty.iri COLLATE \"C\"",
694                );
695            }
696            Scope::TopLevelByRole(role) => {
697                qb.push(
698                    "SELECT r.iri AS iri, o.meta AS meta FROM accel_role r \
699                     JOIN accel_object o \
700                       ON o.graph_iri = r.graph_iri AND o.iri = r.iri \
701                     WHERE r.graph_iri = ",
702                );
703                qb.push_bind(graph.to_owned());
704                qb.push(" AND r.role_iri = ").push_bind(role.clone());
705                qb.push(" AND o.top_level");
706                if let Some(p) = subject_prefix {
707                    qb.push(" AND r.iri LIKE ").push_bind(like_prefix(p));
708                }
709                qb.push(
710                    " ORDER BY left(r.sort_key, 128) COLLATE \"C\", r.sort_key COLLATE \"C\", \
711                              left(r.iri, 128) COLLATE \"C\", r.iri COLLATE \"C\"",
712                );
713            }
714            Scope::TopLevelByTypeAndRole { object_type, role } => {
715                qb.push(
716                    "SELECT r.iri AS iri, o.meta AS meta FROM accel_role r \
717                     JOIN accel_type ty \
718                       ON ty.graph_iri = r.graph_iri AND ty.iri = r.iri \
719                     JOIN accel_object o \
720                       ON o.graph_iri = r.graph_iri AND o.iri = r.iri \
721                     WHERE r.graph_iri = ",
722                );
723                qb.push_bind(graph.to_owned());
724                qb.push(" AND r.role_iri = ").push_bind(role.clone());
725                qb.push(" AND ty.type_iri = ")
726                    .push_bind(object_type.clone());
727                qb.push(" AND o.top_level");
728                if let Some(p) = subject_prefix {
729                    qb.push(" AND r.iri LIKE ").push_bind(like_prefix(p));
730                }
731                qb.push(
732                    " ORDER BY left(r.sort_key, 128) COLLATE \"C\", r.sort_key COLLATE \"C\", \
733                              left(r.iri, 128) COLLATE \"C\", r.iri COLLATE \"C\"",
734                );
735            }
736            Scope::RootByType(t) => {
737                qb.push(
738                    "SELECT ty.iri AS iri, o.meta AS meta FROM accel_type ty \
739                     LEFT JOIN accel_object o \
740                       ON o.graph_iri = ty.graph_iri AND o.iri = ty.iri \
741                     WHERE ty.graph_iri = ",
742                );
743                qb.push_bind(graph.to_owned());
744                qb.push(" AND ty.type_iri = ").push_bind(t.clone());
745                qb.push(
746                    " AND NOT EXISTS (SELECT 1 FROM accel_member m \
747                       WHERE m.graph_iri = ty.graph_iri AND m.member_iri = ty.iri)",
748                );
749                if let Some(p) = subject_prefix {
750                    qb.push(" AND ty.iri LIKE ").push_bind(like_prefix(p));
751                }
752                qb.push(
753                    " ORDER BY left(ty.sort_key, 128) COLLATE \"C\", ty.sort_key COLLATE \"C\", \
754                              left(ty.iri, 128) COLLATE \"C\", ty.iri COLLATE \"C\"",
755                );
756            }
757            Scope::RootTopLevelByType(t) => {
758                qb.push(
759                    "SELECT ty.iri AS iri, o.meta AS meta FROM accel_type ty \
760                     JOIN accel_object o \
761                       ON o.graph_iri = ty.graph_iri AND o.iri = ty.iri \
762                     WHERE ty.graph_iri = ",
763                );
764                qb.push_bind(graph.to_owned());
765                qb.push(" AND ty.type_iri = ").push_bind(t.clone());
766                qb.push(" AND o.top_level");
767                qb.push(
768                    " AND NOT EXISTS (SELECT 1 FROM accel_member m \
769                       WHERE m.graph_iri = ty.graph_iri AND m.member_iri = ty.iri)",
770                );
771                if let Some(p) = subject_prefix {
772                    qb.push(" AND ty.iri LIKE ").push_bind(like_prefix(p));
773                }
774                qb.push(
775                    " ORDER BY left(ty.sort_key, 128) COLLATE \"C\", ty.sort_key COLLATE \"C\", \
776                              left(ty.iri, 128) COLLATE \"C\", ty.iri COLLATE \"C\"",
777                );
778            }
779            Scope::Collection {
780                collection,
781                root_only,
782            } => {
783                qb.push(
784                    "SELECT m.member_iri AS iri, o.meta AS meta FROM accel_member m \
785                     LEFT JOIN accel_object o \
786                       ON o.graph_iri = m.graph_iri AND o.iri = m.member_iri \
787                     WHERE m.graph_iri = ",
788                );
789                qb.push_bind(graph.to_owned());
790                qb.push(" AND m.collection_iri = ")
791                    .push_bind(collection.clone());
792                if *root_only {
793                    qb.push(" AND m.is_root");
794                }
795                if let Some(p) = subject_prefix {
796                    qb.push(" AND m.member_iri LIKE ").push_bind(like_prefix(p));
797                }
798                qb.push(
799                    " ORDER BY left(m.sort_key, 128) COLLATE \"C\", m.sort_key COLLATE \"C\", \
800                              left(m.member_iri, 128) COLLATE \"C\", m.member_iri COLLATE \"C\"",
801                );
802            }
803        }
804        if let Some(l) = limit {
805            qb.push(" LIMIT ").push_bind((offset + l) as i64);
806        }
807
808        let rows = qb.build().fetch_all(&self.pool).await.map_err(db_err)?;
809        let mut out = Vec::with_capacity(rows.len());
810        for row in rows {
811            let iri: String = row.try_get("iri").map_err(db_err)?;
812            let meta_json: Option<String> = row.try_get("meta").map_err(db_err)?;
813            let meta = match meta_json {
814                Some(j) => serde_json::from_str(&j).map_err(db_err)?,
815                None => MetaRecord::default(),
816            };
817            out.push((iri, meta));
818        }
819        Ok(out)
820    }
821
822    /// Fetch one object's metadata projection by primary key. A missing or
823    /// metadata-less object yields no rows (the required title cannot bind).
824    async fn object_metadata(
825        &self,
826        graph: &str,
827        subject: &str,
828        projection: &[(String, Field)],
829        required: &[bool],
830    ) -> Result<AccelSolutions, DomainError> {
831        let vars: Vec<String> = projection.iter().map(|(v, _)| v.clone()).collect();
832        let meta_json: Option<String> =
833            sqlx::query_scalar("SELECT meta FROM accel_object WHERE graph_iri = $1 AND iri = $2")
834                .bind(graph)
835                .bind(subject)
836                .fetch_optional(&self.pool)
837                .await
838                .map_err(db_err)?;
839        let mut rows = Vec::new();
840        if let Some(json) = meta_json {
841            let meta: MetaRecord = serde_json::from_str(&json).map_err(db_err)?;
842            generate_metadata_rows(subject, &meta, projection, required, &mut rows);
843        }
844        Ok(AccelSolutions { vars, rows })
845    }
846
847    async fn count(
848        &self,
849        graph: &str,
850        scope: &Scope,
851        var: &str,
852        subject_prefix: Option<&str>,
853    ) -> Result<AccelSolutions, DomainError> {
854        let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new("");
855        match scope {
856            Scope::TopLevel => {
857                qb.push("SELECT COUNT(*) FROM accel_object WHERE graph_iri = ");
858                qb.push_bind(graph.to_owned());
859                qb.push(" AND top_level");
860                if let Some(p) = subject_prefix {
861                    qb.push(" AND iri LIKE ").push_bind(like_prefix(p));
862                }
863            }
864            Scope::ByType(t) => {
865                qb.push("SELECT COUNT(*) FROM accel_type WHERE graph_iri = ");
866                qb.push_bind(graph.to_owned());
867                qb.push(" AND type_iri = ").push_bind(t.clone());
868                if let Some(p) = subject_prefix {
869                    qb.push(" AND iri LIKE ").push_bind(like_prefix(p));
870                }
871            }
872            Scope::TopLevelByType(t) => {
873                qb.push(
874                    "SELECT COUNT(*) FROM accel_type ty \
875                     JOIN accel_object o \
876                       ON o.graph_iri = ty.graph_iri AND o.iri = ty.iri \
877                     WHERE ty.graph_iri = ",
878                );
879                qb.push_bind(graph.to_owned());
880                qb.push(" AND ty.type_iri = ").push_bind(t.clone());
881                qb.push(" AND o.top_level");
882                if let Some(p) = subject_prefix {
883                    qb.push(" AND ty.iri LIKE ").push_bind(like_prefix(p));
884                }
885            }
886            Scope::TopLevelByRole(role) => {
887                qb.push(
888                    "SELECT COUNT(*) FROM accel_role r \
889                     JOIN accel_object o \
890                       ON o.graph_iri = r.graph_iri AND o.iri = r.iri \
891                     WHERE r.graph_iri = ",
892                );
893                qb.push_bind(graph.to_owned());
894                qb.push(" AND r.role_iri = ").push_bind(role.clone());
895                qb.push(" AND o.top_level");
896                if let Some(p) = subject_prefix {
897                    qb.push(" AND r.iri LIKE ").push_bind(like_prefix(p));
898                }
899            }
900            Scope::TopLevelByTypeAndRole { object_type, role } => {
901                qb.push(
902                    "SELECT COUNT(*) FROM accel_role r \
903                     JOIN accel_type ty \
904                       ON ty.graph_iri = r.graph_iri AND ty.iri = r.iri \
905                     JOIN accel_object o \
906                       ON o.graph_iri = r.graph_iri AND o.iri = r.iri \
907                     WHERE r.graph_iri = ",
908                );
909                qb.push_bind(graph.to_owned());
910                qb.push(" AND r.role_iri = ").push_bind(role.clone());
911                qb.push(" AND ty.type_iri = ")
912                    .push_bind(object_type.clone());
913                qb.push(" AND o.top_level");
914                if let Some(p) = subject_prefix {
915                    qb.push(" AND r.iri LIKE ").push_bind(like_prefix(p));
916                }
917            }
918            Scope::RootByType(t) => {
919                qb.push("SELECT COUNT(*) FROM accel_type ty WHERE ty.graph_iri = ");
920                qb.push_bind(graph.to_owned());
921                qb.push(" AND ty.type_iri = ").push_bind(t.clone());
922                qb.push(
923                    " AND NOT EXISTS (SELECT 1 FROM accel_member m \
924                       WHERE m.graph_iri = ty.graph_iri AND m.member_iri = ty.iri)",
925                );
926                if let Some(p) = subject_prefix {
927                    qb.push(" AND ty.iri LIKE ").push_bind(like_prefix(p));
928                }
929            }
930            Scope::RootTopLevelByType(t) => {
931                qb.push(
932                    "SELECT COUNT(*) FROM accel_type ty \
933                     JOIN accel_object o \
934                       ON o.graph_iri = ty.graph_iri AND o.iri = ty.iri \
935                     WHERE ty.graph_iri = ",
936                );
937                qb.push_bind(graph.to_owned());
938                qb.push(" AND ty.type_iri = ").push_bind(t.clone());
939                qb.push(" AND o.top_level");
940                qb.push(
941                    " AND NOT EXISTS (SELECT 1 FROM accel_member m \
942                       WHERE m.graph_iri = ty.graph_iri AND m.member_iri = ty.iri)",
943                );
944                if let Some(p) = subject_prefix {
945                    qb.push(" AND ty.iri LIKE ").push_bind(like_prefix(p));
946                }
947            }
948            Scope::Collection {
949                collection,
950                root_only,
951            } => {
952                qb.push("SELECT COUNT(*) FROM accel_member WHERE graph_iri = ");
953                qb.push_bind(graph.to_owned());
954                qb.push(" AND collection_iri = ")
955                    .push_bind(collection.clone());
956                if *root_only {
957                    qb.push(" AND is_root");
958                }
959                if let Some(p) = subject_prefix {
960                    qb.push(" AND member_iri LIKE ").push_bind(like_prefix(p));
961                }
962            }
963        }
964        let n: i64 = qb
965            .build_query_scalar()
966            .fetch_one(&self.pool)
967            .await
968            .map_err(db_err)?;
969        Ok(AccelSolutions {
970            vars: vec![var.to_owned()],
971            rows: vec![vec![Some(integer(n as u64))]],
972        })
973    }
974
975    async fn facet(
976        &self,
977        graph: &str,
978        kind: FacetKind,
979        var: &str,
980    ) -> Result<AccelSolutions, DomainError> {
981        let tag = match kind {
982            FacetKind::Types => FK_TYPES,
983            FacetKind::Roles => FK_ROLES,
984            FacetKind::Creators => FK_CREATORS,
985        };
986        let values = sqlx::query_scalar::<_, String>(
987            "SELECT value FROM accel_facet WHERE graph_iri = $1 AND kind = $2 \
988             ORDER BY value COLLATE \"C\"",
989        )
990        .bind(graph)
991        .bind(tag)
992        .fetch_all(&self.pool)
993        .await
994        .map_err(db_err)?;
995        let rows = values
996            .into_iter()
997            .map(|value| {
998                let term = match kind {
999                    FacetKind::Creators => TermValue::Literal {
1000                        value,
1001                        datatype: XSD_STRING.to_owned(),
1002                        language: None,
1003                    },
1004                    _ => TermValue::Iri(value),
1005                };
1006                vec![Some(term)]
1007            })
1008            .collect();
1009        Ok(AccelSolutions {
1010            vars: vec![var.to_owned()],
1011            rows,
1012        })
1013    }
1014
1015    async fn facet_counts(
1016        &self,
1017        graph: &str,
1018        kind: FacetKind,
1019        value_var: &str,
1020        count_var: &str,
1021    ) -> Result<AccelSolutions, DomainError> {
1022        let (tag, iri_value) = match kind {
1023            FacetKind::Types => (FK_TYPES, true),
1024            FacetKind::Roles => (FK_ROLES, true),
1025            FacetKind::Creators => (FK_CREATORS, false),
1026        };
1027        let rows = sqlx::query(
1028            "SELECT value, subject_count FROM accel_facet \
1029             WHERE graph_iri = $1 AND kind = $2 ORDER BY value COLLATE \"C\"",
1030        )
1031        .bind(graph)
1032        .bind(tag)
1033        .fetch_all(&self.pool)
1034        .await
1035        .map_err(db_err)?;
1036        let rows = rows
1037            .into_iter()
1038            .map(|row| {
1039                let value: String = row.try_get("value").map_err(db_err)?;
1040                let count: i64 = row.try_get("subject_count").map_err(db_err)?;
1041                let value = if iri_value {
1042                    TermValue::Iri(value)
1043                } else {
1044                    TermValue::Literal {
1045                        value,
1046                        datatype: XSD_STRING.to_owned(),
1047                        language: None,
1048                    }
1049                };
1050                Ok(vec![Some(value), Some(integer(count as u64))])
1051            })
1052            .collect::<Result<Vec<_>, DomainError>>()?;
1053        Ok(AccelSolutions {
1054            vars: vec![value_var.to_owned(), count_var.to_owned()],
1055            rows,
1056        })
1057    }
1058}
1059
1060/// A `LIKE` pattern matching everything that starts with `prefix`, escaping the
1061/// `LIKE` metacharacters (`%`, `_`, `\`) so an IRI prefix is matched literally.
1062fn like_prefix(prefix: &str) -> String {
1063    let mut out = String::with_capacity(prefix.len() + 1);
1064    for ch in prefix.chars() {
1065        if matches!(ch, '%' | '_' | '\\') {
1066            out.push('\\');
1067        }
1068        out.push(ch);
1069    }
1070    out.push('%');
1071    out
1072}
1073
1074fn catalog_alphabet(encoding: Option<&str>) -> Option<String> {
1075    let encoding = encoding?.to_ascii_lowercase();
1076    Some(
1077        if encoding.contains("protein") || encoding.contains("amino") {
1078            "PROTEIN"
1079        } else if encoding.contains("rna") {
1080            "RNA"
1081        } else if encoding.contains("dna")
1082            || encoding.contains("naseq")
1083            || encoding.contains("1207")
1084        {
1085            "DNA"
1086        } else {
1087            "OTHER"
1088        }
1089        .to_owned(),
1090    )
1091}
sbol_db_postgres/repo/accel.rs

sbol_db_postgres/repo/
accel.rs
