kmp_adapter_embedded/adapter/
bounded_adjacency.rs

1use kmp_domain::{
2    AdjacencyPage, AdjacencyRequest, BoundedRelationReader, NodeRelationProjection, PortError,
3    RelationDirection, RelationPosition,
4};
5
6use super::{
7    engine::{Key, ReadTx, Table},
8    serdes::decode_explanation,
9    store::EmbeddedKernelStore,
10};
11
12/// Reusable inside one traversal's transaction; no second snapshot is opened.
13pub(super) fn read_page(
14    tx: &dyn ReadTx,
15    request: &AdjacencyRequest,
16) -> Result<AdjacencyPage, PortError> {
17    let table = match request.direction() {
18        RelationDirection::Outgoing => Table::Relations,
19        RelationDirection::Incoming => Table::RelationsByTarget,
20    };
21    let after = request
22        .after()
23        .map(|p| (p.neighbor.as_str(), p.relation.as_str()));
24    if request
25        .relation_type()
26        .is_some_and(|kind| request.after().is_some_and(|p| p.relation != kind))
27    {
28        return Err(PortError::InvalidState(
29            "adjacency position belongs to another relation type".into(),
30        ));
31    }
32    let rows = tx.scan_str3_page(
33        table,
34        request.node_id(),
35        after,
36        request.limit(),
37        request.relation_type(),
38    )?;
39    let exhausted = rows.len() < request.limit() as usize;
40    let next = if exhausted {
41        None
42    } else {
43        rows.last()
44            .map(|((_, neighbor, relation), _)| RelationPosition {
45                neighbor: neighbor.clone(),
46                relation: relation.clone(),
47            })
48    };
49    let mut edges = Vec::with_capacity(rows.len());
50    for ((first, second, relation_type), value) in rows {
51        let (source_node_id, target_node_id, raw) = match request.direction() {
52            RelationDirection::Outgoing => (first, second, value),
53            RelationDirection::Incoming => {
54                let raw = tx
55                    .get(Table::Relations, Key::Str3(&second, &first, &relation_type))?
56                    .ok_or_else(|| {
57                        PortError::InvalidState(
58                            "adjacency index points at a missing relation".into(),
59                        )
60                    })?;
61                (second, first, raw)
62            }
63        };
64        edges.push(NodeRelationProjection {
65            source_node_id,
66            target_node_id,
67            relation_type,
68            explanation: decode_explanation(&raw)?,
69        });
70    }
71    Ok(AdjacencyPage {
72        edges,
73        next,
74        exhausted,
75    })
76}
77
78impl BoundedRelationReader for EmbeddedKernelStore {
79    async fn read_adjacency(&self, request: &AdjacencyRequest) -> Result<AdjacencyPage, PortError> {
80        let request = request.clone();
81        self.run(move |store| {
82            let tx = store.begin_read()?;
83            read_page(tx.as_ref(), &request)
84        })
85        .await
86    }
87}
88
89#[cfg(test)]
90#[path = "bounded_adjacency_tests.rs"]
91mod tests;
kmp_adapter_embedded/adapter/bounded_adjacency.rs

kmp_adapter_embedded/adapter/
bounded_adjacency.rs
