use crate::graph::{Entity, GraphStore};
use crate::storage::{self, filter_sql};
use crate::types::{ReadFilter, RecordKind, WriteReceipt};
use crate::{Error, Result};
use petgraph::algo::{astar, connected_components, tarjan_scc};
use petgraph::graph::{DiGraph, Graph, NodeIndex};
use rusqlite::{params, params_from_iter, Connection};
use std::collections::{BTreeMap, HashMap, HashSet};
const SAME_AS: &str = "sys:same_as";
#[derive(Default)]
struct UnionFind { parent: HashMap<i64, i64> }
impl UnionFind {
fn find(&mut self, x: i64) -> i64 {
if !self.parent.contains_key(&x) { self.parent.insert(x, x); return x; }
let mut root = x;
while self.parent[&root] != root { root = self.parent[&root]; }
let mut cursor = x;
while self.parent[&cursor] != root { let next = self.parent[&cursor]; self.parent.insert(cursor, root); cursor = next; }
root
}
fn union(&mut self, a: i64, b: i64) {
let (ra, rb) = (self.find(a), self.find(b));
if ra != rb { self.parent.insert(ra, rb); }
}
}
pub struct GraphView {
graph: Graph<i64, String>,
index: HashMap<i64, NodeIndex>,
canonical: HashMap<i64, i64>,
}
fn snapshot(conn: &Connection, filter: &ReadFilter) -> Result<(Vec<i64>, Vec<(i64, i64, String)>)> {
let (cond, values) = filter_sql(filter, &[RecordKind::Entity], false)?;
let mut stmt = conn.prepare(&format!(
"SELECT e.record_id FROM entities e JOIN records r ON r.id=e.record_id WHERE {cond}"
))?;
let nodes = stmt
.query_map(params_from_iter(values), |row| row.get::<_, i64>(0))?
.collect::<rusqlite::Result<Vec<_>>>()?;
let (cond, values) = filter_sql(filter, &[RecordKind::Relation], false)?;
let mut stmt = conn.prepare(&format!(
"SELECT rel.subject_id, rel.object_id, s.text, pr.is_symmetric, inv.text \
FROM relations rel JOIN records r ON r.id=rel.record_id \
JOIN strings s ON s.id=rel.predicate_id \
LEFT JOIN predicate_rules pr ON pr.predicate_id=rel.predicate_id \
LEFT JOIN strings inv ON inv.id=pr.inverse_predicate_id WHERE {cond}"
))?;
let mut edges: Vec<(i64, i64, String)> = Vec::new();
let mut rows = stmt.query(params_from_iter(values))?;
while let Some(row) = rows.next()? {
let subject: i64 = row.get(0)?;
let object: i64 = row.get(1)?;
let predicate: String = row.get(2)?;
let symmetric: Option<i64> = row.get(3)?;
let inverse: Option<String> = row.get(4)?;
edges.push((subject, object, predicate.clone()));
if symmetric == Some(1) {
edges.push((object, subject, predicate));
} else if let Some(inverse) = inverse {
edges.push((object, subject, inverse));
}
}
Ok((nodes, edges))
}
fn canonical_ids(nodes: &[i64], edges: &[(i64, i64, String)]) -> HashMap<i64, i64> {
let mut uf = UnionFind::default();
for (s, o, pred) in edges {
if pred == SAME_AS { uf.union(*s, *o); }
}
let mut canonical: HashMap<i64, i64> = HashMap::new();
for id in nodes.iter().copied().chain(edges.iter().flat_map(|(s, o, _)| [*s, *o])) {
let representative = uf.find(id);
canonical.insert(id, representative);
}
canonical
}
impl GraphStore {
pub fn build_graph(&self, filter: &ReadFilter) -> Result<GraphView> {
let state = self.0.read()?;
let (nodes, edges) = snapshot(state.conn(), filter)?;
let canonical = canonical_ids(&nodes, &edges);
let mut graph = Graph::<i64, String>::new();
let mut index: HashMap<i64, NodeIndex> = HashMap::new();
for raw in nodes.iter().copied().chain(edges.iter().flat_map(|(s, o, _)| [*s, *o])) {
let representative = canonical[&raw];
if !index.contains_key(&representative) {
let i = graph.add_node(representative);
index.insert(representative, i);
}
}
for (s, o, predicate) in edges {
if predicate == SAME_AS { continue; }
graph.add_edge(index[&canonical[&s]], index[&canonical[&o]], predicate);
}
Ok(GraphView { graph, index, canonical })
}
pub fn strongly_connected(&self, filter: &ReadFilter) -> Result<Vec<Vec<i64>>> {
let (nodes, edges) = {
let state = self.0.read()?;
snapshot(state.conn(), filter)?
};
let canonical = canonical_ids(&nodes, &edges);
let mut graph = DiGraph::<i64, ()>::new();
let mut index: HashMap<i64, NodeIndex> = HashMap::new();
for raw in nodes.iter().copied().chain(edges.iter().flat_map(|(s, o, _)| [*s, *o])) {
let representative = canonical[&raw];
if !index.contains_key(&representative) {
let i = graph.add_node(representative);
index.insert(representative, i);
}
}
for (s, o, _pred) in edges {
let (si, oi) = (index[&canonical[&s]], index[&canonical[&o]]);
if si != oi { graph.add_edge(si, oi, ()); }
}
Ok(tarjan_scc(&graph)
.into_iter()
.filter(|c| c.len() > 1)
.map(|c| c.into_iter().map(|n| graph[n]).collect())
.collect())
}
pub fn ego(&self, root: i64, depth: usize, filter: &ReadFilter, limit: usize) -> Result<Vec<Entity>> {
let ids = self.build_graph(filter)?.ego_ids(root, depth, limit);
let state = self.0.read()?;
let mut loaded: BTreeMap<i64, Entity> = storage::load_many(state.conn(), &ids, filter)?;
Ok(ids.iter().filter_map(|id| loaded.remove(id)).collect())
}
pub fn path(&self, from: i64, to: i64, filter: &ReadFilter) -> Result<Option<Vec<Entity>>> {
let Some(ids) = self.build_graph(filter)?.path_ids(from, to) else {
return Ok(None);
};
let state = self.0.read()?;
let mut loaded: BTreeMap<i64, Entity> = storage::load_many(state.conn(), &ids, filter)?;
Ok(Some(ids.iter().filter_map(|id| loaded.remove(id)).collect()))
}
pub fn set_predicate_rule(&self, predicate: &str, inverse: Option<&str>, symmetric: bool) -> Result<WriteReceipt<()>> {
storage::validate_identity("predicate", predicate)?;
if symmetric && inverse.is_some() {
return Err(Error::Validation("a symmetric predicate must not declare an inverse".into()));
}
if let Some(inverse) = inverse { storage::validate_identity("inverse predicate", inverse)?; }
self.0.mutate_meta(|tx| {
let predicate_id = storage::term_id(tx, predicate)?;
let inverse_id = match inverse { Some(text) => Some(storage::term_id(tx, text)?), None => None };
tx.execute("INSERT INTO predicate_rules(predicate_id,inverse_predicate_id,is_symmetric) VALUES (?1,?2,?3)
ON CONFLICT(predicate_id) DO UPDATE SET inverse_predicate_id=excluded.inverse_predicate_id,is_symmetric=excluded.is_symmetric",
params![predicate_id, inverse_id, i64::from(symmetric)])?;
Ok(())
})
}
}
impl GraphView {
pub fn node_count(&self) -> usize {
self.graph.node_count()
}
pub fn edge_count(&self) -> usize {
self.graph.edge_count()
}
pub fn ego_ids(&self, root: i64, depth: usize, limit: usize) -> Vec<i64> {
let root = self.canonical.get(&root).copied().unwrap_or(root);
let start = match self.index.get(&root) {
Some(&i) => i,
None => return Vec::new(),
};
let mut seen: HashSet<NodeIndex> = HashSet::new();
seen.insert(start);
let mut frontier = vec![start];
let mut out: Vec<i64> = Vec::new();
for _ in 0..depth {
let mut next = Vec::new();
for n in &frontier {
for nb in self.graph.neighbors(*n) {
if seen.insert(nb) {
out.push(self.graph[nb]);
next.push(nb);
if out.len() >= limit {
return out;
}
}
}
}
if next.is_empty() {
break;
}
frontier = next;
}
out
}
pub fn path_ids(&self, from: i64, to: i64) -> Option<Vec<i64>> {
let from = self.canonical.get(&from).copied().unwrap_or(from);
let to = self.canonical.get(&to).copied().unwrap_or(to);
let start = self.index.get(&from)?;
let goal = self.index.get(&to)?;
let (_cost, path) = astar(&self.graph, *start, |n| n == *goal, |_e| 1i32, |_n| 0i32)?;
Some(path.into_iter().map(|n| self.graph[n]).collect())
}
pub fn component_count(&self) -> usize {
connected_components(&self.graph)
}
}
#[cfg(test)]
mod tests {
use crate::graph::{EntityInput, GraphBatch, RelationInput};
use crate::types::{ReadFilter, RecordInput};
use crate::KnowledgeBase;
use std::collections::BTreeMap;
fn ent(name: &str) -> EntityInput {
EntityInput {
record: RecordInput::default(),
name: name.into(),
entity_type: "person".into(),
aliases: Vec::new(),
attributes: BTreeMap::new(),
summary: String::new(),
}
}
fn rel(s: i64, p: &str, o: i64) -> RelationInput {
RelationInput {
record: RecordInput::default(),
subject_id: s,
predicate: p.into(),
object_id: o,
confidence: 1.0,
reason: String::new(),
}
}
#[test]
fn ego_path_components() {
let dir = tempfile::tempdir().unwrap();
let kb = KnowledgeBase::open(dir.path()).unwrap();
let ents = kb
.graph()
.apply_batch(&GraphBatch { entities: vec![ent("A"), ent("B"), ent("C")], ..Default::default() })
.unwrap()
.value
.entities;
let (a, b, c) = (ents[0].header.id, ents[1].header.id, ents[2].header.id);
kb.graph()
.apply_batch(&GraphBatch { relations: vec![rel(a, "knows", b), rel(b, "knows", c)], ..Default::default() })
.unwrap();
let view = kb.graph().build_graph(&ReadFilter::default()).unwrap();
assert_eq!(view.node_count(), 3);
assert_eq!(view.edge_count(), 2);
assert_eq!(view.component_count(), 1);
assert_eq!(view.ego_ids(a, 1, 100), vec![b]);
let mut ego = view.ego_ids(a, 2, 100);
ego.sort();
let mut expect = vec![b, c];
expect.sort();
assert_eq!(ego, expect);
assert_eq!(view.path_ids(a, c), Some(vec![a, b, c]));
let names: Vec<String> = kb
.graph()
.ego(a, 2, &ReadFilter::default(), 100)
.unwrap()
.into_iter()
.map(|e| e.name)
.collect();
assert!(names.contains(&"B".to_string()) && names.contains(&"C".to_string()));
let hop: Vec<String> = kb
.graph()
.path(a, c, &ReadFilter::default())
.unwrap()
.unwrap()
.into_iter()
.map(|e| e.name)
.collect();
assert_eq!(hop, vec!["A", "B", "C"]);
kb.graph()
.apply_batch(&GraphBatch { relations: vec![rel(b, "knows", a)], ..Default::default() })
.unwrap();
let scc = kb.graph().strongly_connected(&ReadFilter::default()).unwrap();
assert_eq!(scc.len(), 1);
let mut ring = scc[0].clone();
ring.sort();
let mut exp = vec![a, b];
exp.sort();
assert_eq!(ring, exp);
kb.graph()
.apply_batch(&GraphBatch { entities: vec![ent("D")], ..Default::default() })
.unwrap();
let view = kb.graph().build_graph(&ReadFilter::default()).unwrap();
assert_eq!(view.node_count(), 4);
assert_eq!(view.component_count(), 2);
}
#[test]
fn filter_isolates_namespaces() {
let dir = tempfile::tempdir().unwrap();
let kb = KnowledgeBase::open(dir.path()).unwrap();
let mut ea = ent("A");
ea.record.namespace = "a".into();
let mut eb = ent("B");
eb.record.namespace = "a".into();
let mut ec = ent("C");
ec.record.namespace = "b".into();
let got = kb
.graph()
.apply_batch(&GraphBatch { entities: vec![ea, eb, ec], ..Default::default() })
.unwrap()
.value
.entities;
let (a, b) = (got[0].header.id, got[1].header.id);
let mut r = rel(a, "knows", b);
r.record.namespace = "a".into();
kb.graph()
.apply_batch(&GraphBatch { relations: vec![r], ..Default::default() })
.unwrap();
let fa = ReadFilter { namespace: "a".into(), scopes: vec!["public".into()], tags: vec![], note_ids: vec![] };
let fb = ReadFilter { namespace: "b".into(), scopes: vec!["public".into()], tags: vec![], note_ids: vec![] };
let va = kb.graph().build_graph(&fa).unwrap();
assert_eq!((va.node_count(), va.edge_count()), (2, 1));
let vb = kb.graph().build_graph(&fb).unwrap();
assert_eq!((vb.node_count(), vb.edge_count()), (1, 0));
}
#[test]
fn virtual_inverse_and_symmetric_edges_enable_reverse_queries() {
let dir = tempfile::tempdir().unwrap();
let kb = KnowledgeBase::open(dir.path()).unwrap();
let filter = ReadFilter::default();
let ents = kb
.graph()
.apply_batch(&GraphBatch { entities: vec![ent("张伟"), ent("张父"), ent("甲"), ent("乙")], ..Default::default() })
.unwrap()
.value
.entities;
let (son, father, jia, yi) = (ents[0].header.id, ents[1].header.id, ents[2].header.id, ents[3].header.id);
kb.graph()
.apply_batch(&GraphBatch { relations: vec![rel(son, "父亲", father), rel(jia, "同事", yi)], ..Default::default() })
.unwrap();
let view = kb.graph().build_graph(&filter).unwrap();
assert!(view.ego_ids(father, 1, 100).is_empty());
assert_eq!(view.path_ids(father, son), None);
kb.graph().set_predicate_rule("父亲", Some("子女"), false).unwrap();
let view = kb.graph().build_graph(&filter).unwrap();
assert_eq!(view.ego_ids(father, 1, 100), vec![son]);
assert_eq!(view.path_ids(father, son), Some(vec![father, son]));
kb.graph().set_predicate_rule("同事", None, true).unwrap();
let view = kb.graph().build_graph(&filter).unwrap();
assert_eq!(view.ego_ids(yi, 1, 100), vec![jia]);
assert!(kb.graph().set_predicate_rule("同事", Some("同事"), true).is_err());
}
#[test]
fn same_as_contracts_alias_nodes() {
let dir = tempfile::tempdir().unwrap();
let kb = KnowledgeBase::open(dir.path()).unwrap();
let filter = ReadFilter::default();
let ents = kb
.graph()
.apply_batch(&GraphBatch { entities: vec![ent("乙"), ent("乙先生"), ent("导师")], ..Default::default() })
.unwrap()
.value
.entities;
let (lin, alias, mentor) = (ents[0].header.id, ents[1].header.id, ents[2].header.id);
kb.graph()
.apply_batch(&GraphBatch { relations: vec![rel(lin, "sys:same_as", alias), rel(alias, "导师", mentor)], ..Default::default() })
.unwrap();
let view = kb.graph().build_graph(&filter).unwrap();
assert_eq!(view.node_count(), 2);
assert_eq!(view.edge_count(), 1);
assert_eq!(view.ego_ids(lin, 1, 100), vec![mentor]);
assert_eq!(view.ego_ids(alias, 1, 100), vec![mentor]);
let path = view.path_ids(lin, mentor).unwrap();
assert_eq!(path.len(), 2);
assert!(path.contains(&mentor));
}
}