reifydb-engine 0.9.1

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

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

use reifydb_test_harness::engine::TestEngine;
use reifydb_value::{params, value::frame::frame::Frame};

fn collect_n(frames: &[Frame]) -> Vec<i32> {
	let mut out: Vec<i32> =
		frames.iter().flat_map(|f| f.rows().map(|r| r.get::<i32>("n").unwrap().unwrap())).collect();
	out.sort();
	out
}

#[test]
fn bulk_inserted_partitioned_series_rows_are_reachable_through_the_partitioned_read_path() {
	// The bulk path had no partitioned branch and wrote every row under an unpartitioned series
	// key, while every reader of a partitioned series scans the partitioned keyspace. The rows
	// were durable and unreachable at the same time, which reads as silent data loss.
	let t = TestEngine::new();
	let identity = TestEngine::identity();

	t.admin("CREATE NAMESPACE test");
	t.admin(
		"CREATE SERIES test::s { ts: int8, region: utf8, n: int4 } WITH { key: ts, partition: { by: { region } } }",
	);

	let mut builder = t.bulk_insert(identity);
	builder.series("test::s")
		.row(params! { ts: 10, region: "us", n: 1 })
		.row(params! { ts: 20, region: "eu", n: 2 })
		.row(params! { ts: 30, region: "us", n: 3 })
		.done();
	let result = builder.execute().unwrap();

	assert_eq!(result.series[0].inserted, 3);
	assert_eq!(collect_n(&t.query("FROM test::s")), vec![1, 2, 3], "an object-wide scan must see every row");
	assert_eq!(
		collect_n(&t.query("FROM test::s FILTER region == \"us\"")),
		vec![1, 3],
		"a pruned partition scan must find the rows the bulk path wrote into that partition"
	);
	assert_eq!(collect_n(&t.query("FROM test::s FILTER region == \"eu\"")), vec![2]);
}

#[test]
fn bulk_inserted_partitioned_series_rows_answer_a_pushed_key_span() {
	// The pruned scan bounds the key span inside the partition prefix, so a row written under the
	// wrong prefix or with a mis-sequenced key falls outside the window rather than failing loudly.
	let t = TestEngine::new();
	let identity = TestEngine::identity();

	t.admin("CREATE NAMESPACE test");
	t.admin(
		"CREATE SERIES test::s { ts: int8, region: utf8, n: int4 } WITH { key: ts, partition: { by: { region } } }",
	);

	let mut builder = t.bulk_insert(identity);
	builder.series("test::s")
		.row(params! { ts: 10, region: "us", n: 1 })
		.row(params! { ts: 20, region: "us", n: 2 })
		.row(params! { ts: 30, region: "us", n: 3 })
		.row(params! { ts: 20, region: "eu", n: 4 })
		.done();
	builder.execute().unwrap();

	assert_eq!(
		collect_n(&t.query("FROM test::s FILTER region == \"us\" and ts >= 20")),
		vec![2, 3],
		"the pushed span must keep exactly the in-range rows of the pruned partition"
	);
	assert_eq!(
		collect_n(&t.query("FROM test::s FILTER region == \"us\" and ts >= 20 and ts <= 20")),
		vec![2],
		"a single-key span must not pick up the other partition's row at the same key"
	);
}

#[test]
fn bulk_inserted_unpartitioned_series_rows_stay_reachable() {
	// The partitioned branch is selected off the series definition, so a series without partition
	// columns must keep taking the plain series key.
	let t = TestEngine::new();
	let identity = TestEngine::identity();

	t.admin("CREATE NAMESPACE test");
	t.admin("CREATE SERIES test::s { ts: int8, n: int4 } WITH { key: ts }");

	let mut builder = t.bulk_insert(identity);
	builder.series("test::s").row(params! { ts: 10, n: 1 }).row(params! { ts: 20, n: 2 }).done();
	builder.execute().unwrap();

	assert_eq!(collect_n(&t.query("FROM test::s")), vec![1, 2]);
	assert_eq!(collect_n(&t.query("FROM test::s FILTER ts >= 20")), vec![2]);
}