reifydb-engine 0.4.12

Query execution and processing engine for ReifyDB
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

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2025 ReifyDB

use reifydb_catalog::change::apply_system_change;
use reifydb_core::{
	delta::Delta,
	interface::{
		catalog::{id::NamespaceId, shape::ShapeId},
		cdc::SystemChange,
	},
};
use reifydb_engine::test_harness::TestEngine;
use reifydb_transaction::transaction::{Transaction, admin::AdminTransaction, replica::ReplicaTransaction};
use reifydb_type::value::identity::IdentityId;

#[test]
fn test_row_ttl_sync_to_materialized_catalog() {
	let engine = TestEngine::new();
	let catalog = engine.catalog();

	// 1. Create a namespace and table with TTL
	engine.admin("CREATE NAMESPACE test");
	engine.admin(r#"
		CREATE TABLE test::users { id: int4 } WITH {
			ttl: { duration: '1h', on: created, mode: drop }
		};
	"#);

	// 2. Check if TTL is in MaterializedCatalog immediately
	let mut txn = engine.begin_admin(IdentityId::system()).unwrap();
	let ns_id = NamespaceId(16385); // 'test' namespace
	let table = catalog
		.find_table_by_name(&mut Transaction::Admin(&mut txn), ns_id, "users")
		.unwrap()
		.expect("table not found");
	let shape = ShapeId::Table(table.id);

	let ttl = catalog.materialized.find_row_ttl(shape).expect("TTL not found in materialized catalog");
	assert_eq!(ttl.duration_nanos, 3_600_000_000_000);
}

#[test]
fn test_row_ttl_replication_sync() {
	let primary = TestEngine::new();
	let replica = TestEngine::new();
	let replica_catalog = replica.catalog();

	// 1. Start transaction on primary
	let mut txn = primary.begin_admin(IdentityId::system()).unwrap();

	// 2. Create table with TTL
	let r = txn.rql("CREATE NAMESPACE test", Default::default());
	if let Some(e) = r.error {
		panic!("{e:?}");
	}
	let r = txn.rql(
		"CREATE TABLE test::users { id: int4 } WITH { ttl: { duration: '1m', on: created, mode: drop } }",
		Default::default(),
	);
	if let Some(e) = r.error {
		panic!("{e:?}");
	}

	// 3. Capture changes
	let changes = deltas_to_system_changes(&txn);

	// 4. Commit primary
	let version = txn.commit().unwrap();

	// 5. Apply to replica
	let mut replica_txn = ReplicaTransaction::new(replica.multi_owned(), version).unwrap();
	for change in &changes {
		apply_system_change(&replica_catalog, &mut Transaction::Replica(&mut replica_txn), change).unwrap();
	}
	replica_txn.commit_at_version().unwrap();

	// 6. Verify replica materialized catalog has the TTL
	// Namespace ID should be 16385
	let mut q_txn = replica.begin_admin(IdentityId::system()).unwrap();
	let table = replica_catalog
		.find_table_by_name(&mut Transaction::Admin(&mut q_txn), NamespaceId(16385), "users")
		.unwrap()
		.expect("table not found on replica");
	let shape = ShapeId::Table(table.id);

	let ttl = replica_catalog
		.materialized
		.find_row_ttl(shape)
		.expect("TTL not found in replica materialized catalog");
	assert_eq!(ttl.duration_nanos, 60_000_000_000);
}

fn deltas_to_system_changes(txn: &AdminTransaction) -> Vec<SystemChange> {
	txn.pending_writes()
		.clone()
		.into_iter_insertion_order()
		.filter_map(|(_, pending)| match pending.delta {
			Delta::Set {
				key,
				row,
			} => Some(SystemChange::Insert {
				key,
				post: row,
			}),
			Delta::Unset {
				key,
				row,
			} => Some(SystemChange::Delete {
				key,
				pre: Some(row),
			}),
			Delta::Remove {
				key,
			} => Some(SystemChange::Delete {
				key,
				pre: None,
			}),
			Delta::Drop {
				key: _,
			} => None,
		})
		.collect()
}