uqa_graph/
incremental_match.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Delta-aware pattern matching (Section 9.3, Paper 2).
8//!
9//! Maintains a set of materialized matches for a [`GraphPattern`] and
10//! refreshes them incrementally on a [`GraphDelta`] application:
11//!
12//! 1. Drop matches that include any vertex affected by the delta.
13//! 2. Re-run the matcher with each pattern variable in turn forced
14//!    to one of the affected vertices, collecting new matches.
15//! 3. Union the new matches into the surviving base set.
16
17use std::collections::BTreeSet;
18
19use uqa_core::{Edge, VertexId};
20
21use crate::delta::{DeltaOp, GraphDelta};
22use crate::operators::GMatch;
23use crate::pattern::{GraphPattern, VertexPredicate};
24use crate::store::{GraphStore, GraphStoreResult};
25
26pub struct IncrementalPatternMatcher {
27    pub pattern: GraphPattern,
28    pub graph: String,
29    pub base_matches: BTreeSet<Vec<VertexId>>,
30}
31
32impl IncrementalPatternMatcher {
33    pub fn new(pattern: GraphPattern, graph: impl Into<String>) -> Self {
34        Self {
35            pattern,
36            graph: graph.into(),
37            base_matches: BTreeSet::new(),
38        }
39    }
40
41    pub fn matches(&self) -> &BTreeSet<Vec<VertexId>> {
42        &self.base_matches
43    }
44
45    /// Initial population of the base match set. Equivalent to a
46    /// one-shot `GMatch` whose results are folded into `base_matches`.
47    pub fn seed<G: GraphStore>(&mut self, store: &G) -> GraphStoreResult<()> {
48        let result = GMatch::new(self.pattern.clone(), &self.graph).execute(store)?;
49        let mut matches = BTreeSet::new();
50        for entry in result.inner().entries() {
51            if let Some(gp) = result.get_graph_payload(entry.doc_id) {
52                let mut vertices = gp.subgraph_vertices.clone();
53                vertices.sort_unstable();
54                vertices.dedup();
55                matches.insert(vertices);
56            }
57        }
58        self.base_matches = matches;
59        Ok(())
60    }
61
62    /// Apply a delta and return the refreshed match set. The store is
63    /// expected to already reflect the delta.
64    pub fn update<G: GraphStore>(
65        &mut self,
66        store: &G,
67        delta: &GraphDelta,
68    ) -> GraphStoreResult<&BTreeSet<Vec<VertexId>>> {
69        if delta
70            .ops()
71            .iter()
72            .any(|operation| matches!(operation, DeltaOp::RemoveEdge(_)))
73        {
74            // A remove-by-id delta does not retain the deleted edge's
75            // endpoints. Negated edge predicates may gain matches anywhere
76            // those endpoints participated, so the only exact refresh after
77            // the store has applied such a delta is a complete re-match.
78            self.seed(store)?;
79            return Ok(&self.base_matches);
80        }
81        let mut affected: BTreeSet<VertexId> = delta.affected_vertex_ids();
82        // Edge add/remove ops also implicate their endpoints, even though
83        // GraphDelta::affected_vertex_ids only sees the endpoints of *added*
84        // edges. For removed edges we look the endpoints up via the store
85        // (the edge has just been deleted; we record any survivor info we
86        // can still resolve).
87        for op in delta.ops() {
88            if let DeltaOp::AddEdge(edge) = op {
89                affected.insert(edge.source_id);
90                affected.insert(edge.target_id);
91            }
92        }
93
94        // Step 1: drop any base match that overlaps an affected vertex.
95        let affected_set = affected.clone();
96        let mut base_matches = self.base_matches.clone();
97        base_matches.retain(|m| !m.iter().any(|v| affected_set.contains(v)));
98
99        // Step 2: for each pattern variable, re-run a constrained match
100        // with that variable bound to one of the affected vertices.
101        let mut new_matches: BTreeSet<Vec<VertexId>> = BTreeSet::new();
102        for vp in &self.pattern.vertex_patterns {
103            let mut constrained_pattern = self.pattern.clone();
104            for cvp in &mut constrained_pattern.vertex_patterns {
105                if cvp.variable == vp.variable {
106                    let affected_for_predicate = affected.clone();
107                    cvp.constraints
108                        .push(VertexPredicate::Custom(std::sync::Arc::new(
109                            move |vertex| affected_for_predicate.contains(&vertex.vertex_id),
110                        )));
111                }
112            }
113            let result = GMatch::new(constrained_pattern, &self.graph).execute(store)?;
114            for entry in result.inner().entries() {
115                if let Some(gp) = result.get_graph_payload(entry.doc_id) {
116                    let mut vertices = gp.subgraph_vertices.clone();
117                    vertices.sort_unstable();
118                    vertices.dedup();
119                    new_matches.insert(vertices);
120                }
121            }
122        }
123
124        base_matches.extend(new_matches);
125        self.base_matches = base_matches;
126        Ok(&self.base_matches)
127    }
128}
129
130/// Convenience helper: count vertices implicated by a delta, using the
131/// store to resolve removed edges back to their endpoints when those
132/// records are still around.
133pub fn implicated_vertices<G: GraphStore>(
134    store: &G,
135    delta: &GraphDelta,
136    graph: &str,
137) -> GraphStoreResult<BTreeSet<VertexId>> {
138    let graph_edge_ids: BTreeSet<_> = store
139        .edges_in_graph(graph)?
140        .into_iter()
141        .map(|edge| edge.edge_id)
142        .collect();
143    let mut out = delta.affected_vertex_ids();
144    for op in delta.ops() {
145        if let DeltaOp::RemoveEdge(eid) = op {
146            if !graph_edge_ids.contains(eid) {
147                continue;
148            }
149            if let Some(Edge {
150                source_id,
151                target_id,
152                ..
153            }) = store.get_edge(*eid).cloned()
154            {
155                out.insert(source_id);
156                out.insert(target_id);
157            }
158        }
159    }
160    Ok(out)
161}
uqa_graph/incremental_match.rs

uqa_graph/
incremental_match.rs
