surrealdb-core 3.2.2

A scalable, distributed, collaborative, document-graph database, for the realtime web
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
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
//! Tests for transaction cache invalidation
//!
//! These tests verify that the transaction cache is properly invalidated
//! when entities are added or removed, preventing stale cache data from
//! causing "not found" errors.

// Common test setup helpers
use crate::catalog::providers::{DatabaseProvider, NamespaceProvider, TableProvider};
use crate::catalog::{DatabaseDefinition, DatabaseId, NamespaceDefinition, NamespaceId, TableId};
use crate::dbs::{Capabilities, Session};
use crate::kvs::Datastore;
use crate::kvs::LockType::Optimistic;
use crate::kvs::TransactionType::Write;
use crate::val::TableName;

/// Helper to create a Datastore and write transaction with namespace and database set up
async fn setup_tx_with_ns_db() -> (Datastore, crate::kvs::Transaction, NamespaceId, DatabaseId) {
	let ds = Datastore::builder()
		.with_capabilities(Capabilities::all())
		.build_with_path("memory")
		.await
		.unwrap();
	let tx = ds.transaction(Write, Optimistic).await.unwrap();

	let ns_def = NamespaceDefinition {
		namespace_id: NamespaceId(1),
		name: "test".into(),
		comment: None,
	};
	tx.put_ns(ns_def).await.unwrap();

	let db_def = DatabaseDefinition {
		namespace_id: NamespaceId(1),
		database_id: DatabaseId(1),
		name: "test".into(),
		strict: false,
		comment: None,
		changefeed: None,
	};
	tx.put_db("test", db_def).await.unwrap();

	(ds, tx, NamespaceId(1), DatabaseId(1))
}

/// Test that verifies index is usable after creation
#[tokio::test]
async fn test_index_usable_after_creation() {
	let ds = Datastore::builder()
		.with_capabilities(Capabilities::all())
		.build_with_path("memory")
		.await
		.unwrap();
	let ses = Session::owner().with_ns("test").with_db("test");

	// Setup
	ds.execute("DEFINE NAMESPACE test", &Session::owner(), None).await.unwrap();
	ds.execute("DEFINE DATABASE test", &ses, None).await.unwrap();
	ds.execute("DEFINE TABLE test_table", &ses, None).await.unwrap();

	// Create an index
	let mut res =
		ds.execute("DEFINE INDEX test_idx ON test_table FIELDS name", &ses, None).await.unwrap();
	assert!(res.remove(0).result.is_ok());

	// Insert some data that uses the indexed field
	let mut res = ds.execute("CREATE test_table SET name = 'test'", &ses, None).await.unwrap();
	assert!(res.remove(0).result.is_ok(), "INSERT should succeed with index present");

	// Query using the indexed field
	let mut res =
		ds.execute("SELECT * FROM test_table WHERE name = 'test'", &ses, None).await.unwrap();
	let val = res.remove(0).result.unwrap();
	assert!(val.is_array() && !val.as_array().unwrap().is_empty(), "Query should return results");
}

/// Test that directly verifies cache invalidation within a single transaction
/// when adding an index via put_tb_index.
#[tokio::test]
async fn test_single_tx_cache_invalidation_on_index_put() {
	use crate::catalog::{Index, IndexDefinition, IndexId};

	let (_ds, tx, ns, db) = setup_tx_with_ns_db().await;
	let tb = TableName::from("test_table");

	// Step 1: Populate the cache with an empty index list
	let indexes = tx.all_tb_indexes(ns, db, &tb, None).await.unwrap();
	assert_eq!(indexes.len(), 0, "Initially there should be no indexes");

	// Step 2: Add an index via put_tb_index
	let ix_def = IndexDefinition {
		index_id: IndexId(1),
		name: "test_idx".into(),
		table_name: tb.clone(),
		cols: vec![],
		index: Index::Idx,
		comment: None,
		prepare_remove: false,
	};
	tx.put_tb_index(ns, db, &tb, &ix_def).await.unwrap();

	// Step 3: Query all indexes again — this must see the new index
	let indexes = tx.all_tb_indexes(ns, db, &tb, None).await.unwrap();
	assert_eq!(
		indexes.len(),
		1,
		"After put_tb_index, all_tb_indexes should return the new index (cache must be invalidated)"
	);
	assert_eq!(indexes[0].name, "test_idx");

	tx.cancel().await.unwrap();
}

/// Test that directly verifies cache invalidation within a single transaction
/// when removing an index via del_tb_index.
#[tokio::test]
async fn test_single_tx_cache_invalidation_on_index_delete() {
	use crate::catalog::{Index, IndexDefinition, IndexId};

	let (_ds, tx, ns, db) = setup_tx_with_ns_db().await;
	let tb = TableName::from("test_table");

	// Add an index
	let ix_def = IndexDefinition {
		index_id: IndexId(1),
		name: "test_idx".into(),
		table_name: tb.clone(),
		cols: vec![],
		index: Index::Idx,
		comment: None,
		prepare_remove: false,
	};
	tx.put_tb_index(ns, db, &tb, &ix_def).await.unwrap();

	// Populate the cache with the list containing one index
	let indexes = tx.all_tb_indexes(ns, db, &tb, None).await.unwrap();
	assert_eq!(indexes.len(), 1, "Should have one index");

	// Remove the index
	tx.del_tb_index(ns, db, &tb, "test_idx").await.unwrap();

	// Query again — must see empty list
	let indexes = tx.all_tb_indexes(ns, db, &tb, None).await.unwrap();
	assert_eq!(
		indexes.len(),
		0,
		"After del_tb_index, all_tb_indexes should return empty list (cache must be invalidated)"
	);

	// Also verify individual cache entry is invalidated
	let ix = tx.get_tb_index(ns, db, &tb, "test_idx", None).await.unwrap();
	assert!(ix.is_none(), "After del_tb_index, get_tb_index should return None");

	tx.cancel().await.unwrap();
}

/// Test that directly verifies cache invalidation within a single transaction
/// when adding a field via put_tb_field.
#[tokio::test]
async fn test_single_tx_cache_invalidation_on_field_put() {
	use std::str::FromStr;

	use crate::catalog::FieldDefinition;
	use crate::expr::Idiom;

	let (_ds, tx, ns, db) = setup_tx_with_ns_db().await;
	let tb = TableName::from("test_table");

	// Step 1: Populate the cache with an empty field list
	let fields = tx.all_tb_fields(ns, db, &tb, None).await.unwrap();
	assert_eq!(fields.len(), 0, "Initially there should be no fields");

	// Step 2: Add a field via put_tb_field
	let fd_def = FieldDefinition {
		name: Idiom::from_str("name").unwrap(),
		table: tb.clone(),
		..Default::default()
	};
	tx.put_tb_field(ns, db, &tb, &fd_def).await.unwrap();

	// Step 3: Query all fields again — this must see the new field
	let fields = tx.all_tb_fields(ns, db, &tb, None).await.unwrap();
	assert_eq!(
		fields.len(),
		1,
		"After put_tb_field, all_tb_fields should return the new field (cache must be invalidated)"
	);

	tx.cancel().await.unwrap();
}

/// Test that verifies multiple sequential index operations work correctly
#[tokio::test]
async fn test_multiple_index_operations_cache_consistency() {
	let ds = Datastore::builder()
		.with_capabilities(Capabilities::all())
		.build_with_path("memory")
		.await
		.unwrap();
	let ses = Session::owner().with_ns("test").with_db("test");

	// Setup
	ds.execute("DEFINE NAMESPACE test", &Session::owner(), None).await.unwrap();
	ds.execute("DEFINE DATABASE test", &ses, None).await.unwrap();
	ds.execute("DEFINE TABLE test_table", &ses, None).await.unwrap();

	// Add first index
	let mut res =
		ds.execute("DEFINE INDEX idx1 ON test_table FIELDS field1", &ses, None).await.unwrap();
	assert!(res.remove(0).result.is_ok());

	// Add second index
	let mut res =
		ds.execute("DEFINE INDEX idx2 ON test_table FIELDS field2", &ses, None).await.unwrap();
	assert!(res.remove(0).result.is_ok());

	// Verify both indexes exist
	let mut res = ds.execute("INFO FOR TABLE test_table", &ses, None).await.unwrap();
	let val = res.remove(0).result.unwrap();
	let info = format!("{:?}", val);
	assert!(info.contains("idx1") && info.contains("idx2"), "Both indexes should be visible");

	// Remove first index
	let mut res = ds.execute("REMOVE INDEX idx1 ON test_table", &ses, None).await.unwrap();
	assert!(res.remove(0).result.is_ok());

	// Verify only second index remains
	let mut res = ds.execute("INFO FOR TABLE test_table", &ses, None).await.unwrap();
	let val = res.remove(0).result.unwrap();
	let info = format!("{:?}", val);
	assert!(
		!info.contains("idx1") && info.contains("idx2"),
		"Only idx2 should remain after removing idx1"
	);
}

/// Test cache invalidation for put_ns (namespace list cache).
#[tokio::test]
async fn test_single_tx_cache_invalidation_on_ns_put() {
	let ds = Datastore::builder()
		.with_capabilities(Capabilities::all())
		.build_with_path("memory")
		.await
		.unwrap();
	let tx = ds.transaction(Write, Optimistic).await.unwrap();

	// Populate the cache with an empty namespace list
	let nss = tx.all_ns(None).await.unwrap();
	assert_eq!(nss.len(), 0, "Initially there should be no namespaces");

	// Add a namespace
	let ns_def = NamespaceDefinition {
		namespace_id: NamespaceId(1),
		name: "test".into(),
		comment: None,
	};
	tx.put_ns(ns_def).await.unwrap();

	// Query again — must see the new namespace
	let nss = tx.all_ns(None).await.unwrap();
	assert_eq!(
		nss.len(),
		1,
		"After put_ns, all_ns should return the new namespace (cache must be invalidated)"
	);

	tx.cancel().await.unwrap();
}

/// Test cache invalidation for put_db and del_db (database list cache).
#[tokio::test]
async fn test_single_tx_cache_invalidation_on_db_put_and_del() {
	let ds = Datastore::builder()
		.with_capabilities(Capabilities::all())
		.build_with_path("memory")
		.await
		.unwrap();
	let tx = ds.transaction(Write, Optimistic).await.unwrap();

	let ns_def = NamespaceDefinition {
		namespace_id: NamespaceId(1),
		name: "test".into(),
		comment: None,
	};
	tx.put_ns(ns_def).await.unwrap();

	// Populate the cache with an empty database list
	let dbs = tx.all_db(NamespaceId(1), None).await.unwrap();
	assert_eq!(dbs.len(), 0, "Initially there should be no databases");

	// Add a database
	let db_def = DatabaseDefinition {
		namespace_id: NamespaceId(1),
		database_id: DatabaseId(1),
		name: "testdb".into(),
		strict: false,
		comment: None,
		changefeed: None,
	};
	tx.put_db("test", db_def).await.unwrap();

	// Query again — must see the new database
	let dbs = tx.all_db(NamespaceId(1), None).await.unwrap();
	assert_eq!(
		dbs.len(),
		1,
		"After put_db, all_db should return the new database (cache must be invalidated)"
	);

	// Delete the database (deferred: removes the catalog entry + invalidates
	// the cache now; the data prefix is reclaimed in the background).
	tx.del_db_deferred("test", "testdb", false).await.unwrap();

	// Query again — must see empty list
	let dbs = tx.all_db(NamespaceId(1), None).await.unwrap();
	assert_eq!(
		dbs.len(),
		0,
		"After del_db_deferred, all_db should return empty list (cache must be invalidated)"
	);

	tx.cancel().await.unwrap();
}

/// Test cache invalidation for put_tb and del_tb (table list cache).
#[tokio::test]
async fn test_single_tx_cache_invalidation_on_tb_put_and_del() {
	use crate::catalog::TableDefinition;

	let (_ds, tx, ns, db) = setup_tx_with_ns_db().await;

	// Populate the cache with an empty table list
	let tbs = tx.all_tb(ns, db, None).await.unwrap();
	assert_eq!(tbs.len(), 0, "Initially there should be no tables");

	// Add a table
	let tb_def = TableDefinition::new(ns, db, TableId(1), TableName::from("test_table"));
	tx.put_tb("test", "test", &tb_def).await.unwrap();

	// Query again — must see the new table
	let tbs = tx.all_tb(ns, db, None).await.unwrap();
	assert_eq!(
		tbs.len(),
		1,
		"After put_tb, all_tb should return the new table (cache must be invalidated)"
	);

	// Delete the table
	tx.del_tb("test", "test", &TableName::from("test_table")).await.unwrap();

	// Query again — must see empty list
	let tbs = tx.all_tb(ns, db, None).await.unwrap();
	assert_eq!(
		tbs.len(),
		0,
		"After del_tb, all_tb should return empty list (cache must be invalidated)"
	);

	tx.cancel().await.unwrap();
}

/// Test cache invalidation for put_db_param (param list cache).
/// This also validates the pattern used for put_db_function, put_db_module, and put_db_api.
#[tokio::test]
async fn test_single_tx_cache_invalidation_on_param_put() {
	use crate::catalog::ParamDefinition;

	let (_ds, tx, ns, db) = setup_tx_with_ns_db().await;

	// Populate the cache with an empty param list
	let pas = tx.all_db_params(ns, db, None).await.unwrap();
	assert_eq!(pas.len(), 0, "Initially there should be no params");

	// Add a param
	let pa_def = ParamDefinition {
		name: "test_param".into(),
		value: crate::val::Value::Bool(true),
		..Default::default()
	};
	tx.put_db_param(ns, db, &pa_def).await.unwrap();

	// Query again — must see the new param
	let pas = tx.all_db_params(ns, db, None).await.unwrap();
	assert_eq!(
		pas.len(),
		1,
		"After put_db_param, all_db_params should return the new param (cache must be invalidated)"
	);

	tx.cancel().await.unwrap();
}

/// Test that a versioned read of tables does not pollute the current-view cache.
///
/// The LRU cache keys do not incorporate a version, so historical reads must
/// bypass the cache entirely. If they wrote into the cache, a subsequent
/// None (current) read would see stale data.
#[tokio::test]
async fn test_versioned_read_does_not_pollute_table_cache() {
	let ds = Datastore::builder()
		.with_capabilities(Capabilities::all())
		.build_with_path("memory?versioned=true")
		.await
		.unwrap();
	let ses = Session::owner().with_ns("test").with_db("test");

	ds.execute("DEFINE NAMESPACE test", &Session::owner(), None).await.unwrap();
	ds.execute("DEFINE DATABASE test", &ses, None).await.unwrap();
	ds.execute("DEFINE TABLE my_table", &ses, None).await.unwrap();

	let tx = ds.transaction(Write, Optimistic).await.unwrap();
	let ns_def = tx.get_ns_by_name("test", None).await.unwrap().unwrap();
	let db_def = tx.get_db_by_name("test", "test", None).await.unwrap().unwrap();
	let ns = ns_def.namespace_id;
	let db = db_def.database_id;

	let tables = tx.all_tb(ns, db, None).await.unwrap();
	assert_eq!(tables.len(), 1, "Current view should have 1 table");

	let old_tables = tx.all_tb(ns, db, Some(0)).await.unwrap();
	assert_eq!(old_tables.len(), 0, "Historical read at version 0 should be empty");

	let tables_again = tx.all_tb(ns, db, None).await.unwrap();
	assert_eq!(tables_again.len(), 1, "Current view must still have 1 table after versioned read");

	tx.cancel().await.unwrap();
}

/// Test that a versioned read of field definitions does not pollute the current-view cache.
#[tokio::test]
async fn test_versioned_read_does_not_pollute_field_cache() {
	let ds = Datastore::builder()
		.with_capabilities(Capabilities::all())
		.build_with_path("memory?versioned=true")
		.await
		.unwrap();
	let ses = Session::owner().with_ns("test").with_db("test");

	ds.execute("DEFINE NAMESPACE test", &Session::owner(), None).await.unwrap();
	ds.execute("DEFINE DATABASE test", &ses, None).await.unwrap();
	ds.execute("DEFINE TABLE my_table", &ses, None).await.unwrap();
	ds.execute("DEFINE FIELD name ON TABLE my_table TYPE string", &ses, None).await.unwrap();

	let tx = ds.transaction(Write, Optimistic).await.unwrap();
	let ns_def = tx.get_ns_by_name("test", None).await.unwrap().unwrap();
	let db_def = tx.get_db_by_name("test", "test", None).await.unwrap().unwrap();
	let ns = ns_def.namespace_id;
	let db = db_def.database_id;
	let tb = TableName::from("my_table");

	let fields = tx.all_tb_fields(ns, db, &tb, None).await.unwrap();
	assert_eq!(fields.len(), 1, "Current view should have 1 field");

	let old_fields = tx.all_tb_fields(ns, db, &tb, Some(0)).await.unwrap();
	assert_eq!(old_fields.len(), 0, "Historical read at version 0 should be empty");

	let fields_again = tx.all_tb_fields(ns, db, &tb, None).await.unwrap();
	assert_eq!(fields_again.len(), 1, "Current view must still have 1 field after versioned read");

	tx.cancel().await.unwrap();
}