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