nodedb 0.4.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
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
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230

// SPDX-License-Identifier: BUSL-1.1

//! Regression coverage for the txn-buffering atomicity fix: a documented set
//! of writes (`ArrayOp::{Put, Delete}`, `DocumentOp::{BatchInsert, Merge,
//! UpdateFromJoin}`, plus `CrdtOp` and `VectorOp` variants not exercised by a
//! standalone SQL surface) used to execute IMMEDIATELY against base state
//! inside an explicit `BEGIN ... COMMIT` block — visible before COMMIT, and
//! NOT undone by ROLLBACK. They are now buffered like every other
//! transactional write: ROLLBACK discards them, COMMIT is required to
//! persist them, and — the deliberate trade-off documented in
//! `control/server/shared/write_admission/predicate/txn_buffering.rs` — a
//! read later in the SAME transaction no longer observes the buffered write
//! until COMMIT (read-your-own-writes is lost for these ops specifically).
//!
//! `ClusterArrayOp::{Put, Delete}` has no direct single-node SQL surface
//! (it is the Control-Plane multi-shard array coordinator's own physical
//! plan, never constructed by `plan_sql()`), so it is not covered here.
//! `CrdtOp` and `VectorOp` flipped variants likewise have no direct SQL
//! entry point exercised by the existing pgwire surface at the time of this
//! change, so the fix is pinned via the Array and Document cases below,
//! which exercise the identical `exec_tx_passthrough` COMMIT-replay path.

mod common;

use common::pgwire_harness::TestServer;

/// Query a single-cell 2D array by exact coordinates. Returns the matched
/// rows (0 or 1 for a point query).
async fn array_cell_rows(server: &TestServer, array: &str, row: i64, col: i64) -> Vec<Vec<String>> {
    server
        .query_rows(&format!(
            "SELECT * FROM ARRAY_SLICE('{array}', '{{\"row\":[{row},{row}],\"col\":[{col},{col}]}}', '*', 10)"
        ))
        .await
        .unwrap()
}

async fn setup_array(server: &TestServer, array: &str) {
    server
        .exec(&format!(
            "CREATE ARRAY {array} \
             DIMS (row INT64, col INT64) \
             ATTRS (value FLOAT64) \
             TILE_EXTENTS (10, 10)"
        ))
        .await
        .unwrap();
}

/// `BEGIN; INSERT INTO ARRAY; ROLLBACK` must leave the cell ABSENT. On the
/// pre-fix predicate, `ArrayOp::Put` classified as an unbuffered write: the
/// insert executed immediately against base state and SURVIVED the
/// ROLLBACK — this assertion is exactly what caught that bug.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn array_insert_rollback_discards_cell() {
    let server = TestServer::start().await;
    setup_array(&server, "arr_atomic_rb").await;

    server.exec("BEGIN").await.unwrap();
    server
        .exec("INSERT INTO ARRAY arr_atomic_rb COORDS (1, 1) VALUES (7.0)")
        .await
        .unwrap();
    server.client.simple_query("ROLLBACK").await.unwrap();

    let rows = array_cell_rows(&server, "arr_atomic_rb", 1, 1).await;
    assert!(
        rows.is_empty(),
        "ROLLBACK must discard a buffered ArrayOp::Put; found cell: {rows:?}"
    );
}

/// `BEGIN; INSERT INTO ARRAY; COMMIT` must leave the cell PRESENT: buffering
/// the write does not silently drop it, COMMIT still durably applies it.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn array_insert_commit_persists_cell() {
    let server = TestServer::start().await;
    setup_array(&server, "arr_atomic_commit").await;

    server.exec("BEGIN").await.unwrap();
    server
        .exec("INSERT INTO ARRAY arr_atomic_commit COORDS (2, 2) VALUES (9.0)")
        .await
        .unwrap();
    server.client.simple_query("COMMIT").await.unwrap();

    let rows = array_cell_rows(&server, "arr_atomic_commit", 2, 2).await;
    assert_eq!(
        rows.len(),
        1,
        "COMMIT must durably persist a buffered ArrayOp::Put; got {rows:?}"
    );
}

/// `BEGIN; DELETE FROM ARRAY; ROLLBACK` must leave the pre-existing cell
/// PRESENT (the delete must not have executed against base state).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn array_delete_rollback_restores_cell() {
    let server = TestServer::start().await;
    setup_array(&server, "arr_atomic_del_rb").await;
    server
        .exec("INSERT INTO ARRAY arr_atomic_del_rb COORDS (3, 3) VALUES (5.0)")
        .await
        .unwrap();

    server.exec("BEGIN").await.unwrap();
    server
        .exec("DELETE FROM ARRAY arr_atomic_del_rb WHERE COORDS IN ((3, 3))")
        .await
        .unwrap();
    server.client.simple_query("ROLLBACK").await.unwrap();

    let rows = array_cell_rows(&server, "arr_atomic_del_rb", 3, 3).await;
    assert_eq!(
        rows.len(),
        1,
        "ROLLBACK must discard a buffered ArrayOp::Delete; cell must survive, got {rows:?}"
    );
}

/// Pin the RYOW-loss trade-off explicitly: a write buffered by this fix
/// (`ArrayOp::Put`) is NOT staged into the per-transaction overlay, so a read
/// later in the SAME transaction does not observe it until COMMIT. This is a
/// deliberate, documented behavior change (module doc on
/// `txn_buffering.rs`), not a regression — asserted here so it is proven by
/// a test rather than discovered later.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn array_insert_in_txn_not_visible_to_same_txn_read() {
    let server = TestServer::start().await;
    setup_array(&server, "arr_atomic_ryow").await;

    server.exec("BEGIN").await.unwrap();
    server
        .exec("INSERT INTO ARRAY arr_atomic_ryow COORDS (4, 4) VALUES (1.0)")
        .await
        .unwrap();

    let rows = array_cell_rows(&server, "arr_atomic_ryow", 4, 4).await;
    assert!(
        rows.is_empty(),
        "a buffered ArrayOp::Put must NOT be visible to a read in the same \
         transaction before COMMIT (RYOW loss is the documented trade-off \
         for closing the atomicity gap); got {rows:?}"
    );

    server.client.simple_query("ROLLBACK").await.unwrap();
}

// ── Document MERGE: rollback discards, commit persists ─────────────────────

async fn create_merge_schema(server: &TestServer) {
    server
        .exec(
            "CREATE COLLECTION merge_atomic_target (\
                id TEXT PRIMARY KEY, \
                name TEXT, \
                score INT) WITH (engine='document_strict')",
        )
        .await
        .unwrap();
    server
        .exec(
            "CREATE COLLECTION merge_atomic_source (\
                id TEXT PRIMARY KEY, \
                name TEXT, \
                score INT) WITH (engine='document_strict')",
        )
        .await
        .unwrap();
    server
        .exec("INSERT INTO merge_atomic_target (id, name, score) VALUES ('a', 'alpha', 10)")
        .await
        .unwrap();
    server
        .exec("INSERT INTO merge_atomic_source (id, name, score) VALUES ('a', 'ALPHA_UPD', 99)")
        .await
        .unwrap();
}

const MERGE_SQL: &str = "MERGE INTO merge_atomic_target t \
     USING merge_atomic_source s ON t.id = s.id \
     WHEN MATCHED THEN UPDATE SET name = s.name, score = s.score";

/// `BEGIN; MERGE; ROLLBACK` must leave the target row UNCHANGED. On the
/// pre-fix predicate, `DocumentOp::Merge` classified as an unbuffered write:
/// the merge executed immediately against base state and SURVIVED the
/// ROLLBACK — this assertion is exactly what caught that bug.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn document_merge_rollback_discards_changes() {
    let server = TestServer::start().await;
    create_merge_schema(&server).await;

    server.exec("BEGIN").await.unwrap();
    server.exec(MERGE_SQL).await.unwrap();
    server.client.simple_query("ROLLBACK").await.unwrap();

    let rows = server
        .query_rows("SELECT name, score FROM merge_atomic_target WHERE id = 'a'")
        .await
        .unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(
        rows[0],
        vec!["alpha".to_string(), "10".to_string()],
        "ROLLBACK must discard a buffered DocumentOp::Merge; got {rows:?}"
    );
}

/// `BEGIN; MERGE; COMMIT` must leave the target row updated: buffering the
/// merge does not silently drop it, COMMIT still durably applies it.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn document_merge_commit_persists_changes() {
    let server = TestServer::start().await;
    create_merge_schema(&server).await;

    server.exec("BEGIN").await.unwrap();
    server.exec(MERGE_SQL).await.unwrap();
    server.client.simple_query("COMMIT").await.unwrap();

    let rows = server
        .query_rows("SELECT name, score FROM merge_atomic_target WHERE id = 'a'")
        .await
        .unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(
        rows[0],
        vec!["ALPHA_UPD".to_string(), "99".to_string()],
        "COMMIT must durably persist a buffered DocumentOp::Merge; got {rows:?}"
    );
}