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
162
163
164
165
166
167
168
169
170
171
172
173
174
175
// SPDX-License-Identifier: BUSL-1.1
//! In-transaction MATCH pattern reads observe the transaction's own staged
//! graph edge writes/deletes (read-your-own-writes).
//!
//! `GRAPH NEIGHBORS`/`Hop`/traversal already merged the per-transaction
//! `GraphTxnOverlay` for RYOW, but the MATCH pattern engine read committed CSR
//! only: a staged `GRAPH INSERT EDGE` was invisible to a MATCH in the same
//! transaction, and a staged `GRAPH DELETE EDGE` was still traversed. This
//! suite locks in the MATCH-side merge: a staged edge PUT appears in an in-txn
//! MATCH and disappears on ROLLBACK; a staged edge DELETE disappears from an
//! in-txn MATCH and reappears on ROLLBACK; and an autocommit MATCH (no
//! transaction) is unchanged (committed-CSR-only).
//!
//! Every staged edge here is a SELF-LOOP (`FROM == TO == node`), exactly as the
//! sibling `sql_transactions_graph_overlay` / `sql_transactions_graph_edge_delete_overlay`
//! suites do: inside an explicit `BEGIN..COMMIT` only a SINGLE-HOME
//! (both-endpoints-on-one-vShard) edge write stages through the per-task gate;
//! a distinct-endpoint (dual-home) edge write in a transaction is rejected as
//! cross-shard. A self-loop `n -[:l]-> n` still yields a MATCH row `{x:n, y:n}`,
//! which is sufficient to exercise the staged-PUT-visible and
//! staged-DELETE-hidden merge on the MATCH read path.
//!
//! NOTE on multi-hop through a STAGED-ONLY intermediate node: driving that at
//! the SQL layer needs distinct-endpoint staged edges (`a->b`, `b->c` with `b`
//! brand-new), which the single-home staging constraint above rejects inside a
//! transaction. That path (a MATCH walking through a staged-only node with no
//! durable CSR id) is covered by the `overlay_expand` unit test
//! `staged_only_node_expands_as_source` instead, where the overlay is exercised
//! without the SQL staging gate.
mod common;
use common::pgwire_harness::TestServer;
use tokio_postgres::SimpleQueryMessage;
async fn create_collection(server: &TestServer) {
server.exec("CREATE COLLECTION g_match").await.unwrap();
}
/// `GRAPH INSERT EDGE` self-loop `node -[:label]-> node` in collection `g_match`.
async fn insert_self_loop(server: &TestServer, node: &str, label: &str) -> Result<(), String> {
server
.client
.simple_query(&format!(
"GRAPH INSERT EDGE IN 'g_match' FROM '{node}' TO '{node}' TYPE '{label}'"
))
.await
.map(|_| ())
.map_err(|e| e.to_string())
}
/// `GRAPH DELETE EDGE` self-loop `node -[:label]-> node` in collection `g_match`.
async fn delete_self_loop(server: &TestServer, node: &str, label: &str) -> Result<(), String> {
server
.client
.simple_query(&format!(
"GRAPH DELETE EDGE IN 'g_match' FROM '{node}' TO '{node}' TYPE '{label}'"
))
.await
.map(|_| ())
.map_err(|e| e.to_string())
}
/// Run `MATCH (x)-[:l]->(y) RETURN x, y` and return the set of `x` node ids
/// (self-loops satisfy `x == y`).
async fn match_sources(server: &TestServer, label: &str) -> Vec<String> {
let msgs = server
.client
.simple_query(&format!("MATCH (x)-[:{label}]->(y) RETURN x, y"))
.await
.expect("MATCH should succeed");
let mut out = Vec::new();
for msg in msgs {
if let SimpleQueryMessage::Row(row) = msg
&& let Some(x) = row.get(0)
{
out.push(x.to_string());
}
}
out
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn in_tx_match_sees_staged_edge_put_and_rollback_discards() {
let server = TestServer::start().await;
create_collection(&server).await;
// Committed edge, before any transaction.
insert_self_loop(&server, "base_put", "l").await.unwrap();
server.exec("BEGIN").await.unwrap();
insert_self_loop(&server, "staged_put", "l")
.await
.expect("in-tx GRAPH INSERT EDGE should stage at statement time");
// Read-your-own-writes: the in-tx MATCH must include the staged edge.
// Pre-fix, MATCH read committed CSR only and this was absent.
let in_tx = match_sources(&server, "l").await;
assert!(
in_tx.contains(&"staged_put".to_string()),
"in-tx MATCH must observe the transaction's own staged edge PUT \
(read-your-own-writes), got: {in_tx:?}"
);
assert!(
in_tx.contains(&"base_put".to_string()),
"a pre-existing committed edge must remain visible in-tx, got: {in_tx:?}"
);
server.client.simple_query("ROLLBACK").await.unwrap();
let after = match_sources(&server, "l").await;
assert!(
!after.contains(&"staged_put".to_string()),
"a rolled-back staged edge must not appear in MATCH, got: {after:?}"
);
assert!(
after.contains(&"base_put".to_string()),
"the committed base edge must survive the rollback, got: {after:?}"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn in_tx_match_hides_staged_edge_delete_and_rollback_restores() {
let server = TestServer::start().await;
create_collection(&server).await;
// Committed edge, before any transaction.
insert_self_loop(&server, "del_target", "l").await.unwrap();
assert!(
match_sources(&server, "l")
.await
.contains(&"del_target".to_string()),
"the committed edge must be visible before the transaction"
);
server.exec("BEGIN").await.unwrap();
delete_self_loop(&server, "del_target", "l")
.await
.expect("in-tx GRAPH DELETE EDGE should stage at statement time");
// Read-your-own-writes: the staged tombstone must hide the edge from an
// in-tx MATCH. Pre-fix, MATCH still traversed the durable edge.
let in_tx = match_sources(&server, "l").await;
assert!(
!in_tx.contains(&"del_target".to_string()),
"in-tx MATCH must NOT observe an edge deleted in the same transaction \
(read-your-own-writes), got: {in_tx:?}"
);
server.client.simple_query("ROLLBACK").await.unwrap();
// The delete was staged, never applied — ROLLBACK restores the edge.
let after = match_sources(&server, "l").await;
assert!(
after.contains(&"del_target".to_string()),
"a rolled-back edge delete must leave the edge visible to MATCH, got: {after:?}"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn autocommit_match_is_committed_only_unchanged() {
let server = TestServer::start().await;
create_collection(&server).await;
insert_self_loop(&server, "auto_edge", "l").await.unwrap();
// No transaction: MATCH resolves the active TxnId to None, so behaviour is
// committed-CSR-only exactly as before the overlay merge.
let rows = match_sources(&server, "l").await;
assert!(
rows.contains(&"auto_edge".to_string()),
"an autocommit MATCH must see the committed edge, got: {rows:?}"
);
}