gnitz-store 0.1.4

The Z-set store of the gnitz database: LSM storage, relation registry and read executor
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

use super::enforce_unique_pk;
use crate::test_support::{
    make_batch_bytes, make_batch_opk, make_schema_pk_u64_payload_string, opk_pk, payload0_i64, pk_payload_schema,
    scratch_table, zset_of,
};
use gnitz_wire::TypeCode;

/// `(key index, weight, payload)`.
type Row = (usize, i64, i64);

/// A key's `(key index, net weight, live payload)` after a step.
type Live = (usize, i64, Option<i64>);

/// A push, the effective batch it must become, and each named key's state once
/// that is ingested.
type Step = (&'static str, &'static [Row], &'static [Row], &'static [Live]);

const SCRIPT: &[Step] = &[
    ("fresh +2", &[(1, 2, 100)], &[(1, 1, 100)], &[(1, 1, Some(100))]),
    (
        "upsert +2",
        &[(1, 2, 200)],
        &[(1, -1, 100), (1, 1, 200)],
        &[(1, 1, Some(200))],
    ),
    ("delete -3", &[(1, -3, 0)], &[(1, -1, 200)], &[(1, 0, None)]),
    (
        "delete of tombstoned and absent keys",
        &[(1, -1, 0), (2, -1, 0)],
        &[],
        &[(1, 0, None), (2, 0, None)],
    ),
    (
        "fresh + - +",
        &[(2, 1, 10), (2, -1, 10), (2, 1, 20)],
        &[(2, 1, 20)],
        &[(2, 1, Some(20))],
    ),
    (
        "fresh + +",
        &[(3, 1, 10), (3, 1, 20)],
        &[(3, 1, 20)],
        &[(3, 1, Some(20))],
    ),
    ("fresh + -", &[(4, 1, 10), (4, -1, 10)], &[], &[(4, 0, None)]),
    (
        "stored - +",
        &[(2, -1, 0), (2, 1, 30)],
        &[(2, -1, 20), (2, 1, 30)],
        &[(2, 1, Some(30))],
    ),
    (
        "stored + +",
        &[(2, 1, 40), (2, 1, 50)],
        &[(2, -1, 30), (2, 1, 50)],
        &[(2, 1, Some(50))],
    ),
    (
        "stored + -",
        &[(2, 1, 60), (2, -1, 60)],
        &[(2, -1, 50)],
        &[(2, 0, None)],
    ),
    (
        "zero row",
        &[(5, 1, 1), (6, 0, 2), (7, 1, 3)],
        &[(5, 1, 1), (7, 1, 3)],
        &[(5, 1, Some(1)), (6, 0, None), (7, 1, Some(3))],
    ),
    (
        "fresh passthrough",
        &[(10, 1, 1), (8, 1, 2), (9, 1, 3)],
        &[(10, 1, 1), (8, 1, 2), (9, 1, 3)],
        &[(8, 1, Some(2)), (9, 1, Some(3)), (10, 1, Some(1))],
    ),
];

/// One script of pushes, run at every PK shape. Every key's net weight stays in
/// `{0, 1}`, and the effective batch carries no cancelling pair.
#[test]
fn enforce_unique_pk_holds_at_every_pk_shape() {
    type Key = fn(usize) -> Vec<u128>;
    let cases: [(&str, &[TypeCode], Key); 4] = [
        ("u64", &[TypeCode::U64], |i| vec![i as u128]),
        // Negative keys, which OPK sign-flips.
        ("signed i64", &[TypeCode::I64], |i| vec![-(i as i64) as u64 as u128]),
        ("narrow u8", &[TypeCode::U8], |i| vec![i as u128]),
        ("wide 3xu64", &[TypeCode::U64; 3], |i| {
            vec![i as u128, i as u128 + 1, i as u128 + 2]
        }),
    ];

    for (name, pk_types, key) in cases {
        let schema = pk_payload_schema(pk_types);
        let dir = tempfile::tempdir().unwrap();
        let mut pt = scratch_table(dir.path(), schema);
        let k: Vec<Vec<u8>> = (0..=10).map(|i| opk_pk(&schema, &key(i))).collect();
        let batch = |rows: &[Row]| {
            let rows: Vec<(&[u8], i64, i64)> = rows.iter().map(|&(i, w, v)| (&k[i][..], w, v)).collect();
            make_batch_opk(&schema, &rows)
        };

        for &(step, push, want, live) in SCRIPT {
            let what = format!("{name}: {step}");
            let eff = enforce_unique_pk(&pt, batch(push));
            assert_eq!(zset_of(&eff, &schema), zset_of(&batch(want), &schema), "{what}");
            assert_eq!(eff.len(), want.len(), "{what}: effective row count");
            pt.ingest_borrowed_batch(&eff).unwrap();
            for &(i, net, payload) in live {
                let (w, row) = pt.live_row_at(&k[i]);
                assert_eq!(
                    (w, row.as_ref().map(|r| r.source()).map(|(src, i)| payload0_i64(src, i))),
                    (net, payload),
                    "{what}: key {i}"
                );
            }
        }
    }
}

/// Heap strings read back intact, from kept rows and from a stored row's
/// retraction alike.
#[test]
fn enforce_unique_pk_keeps_heap_strings_intact() {
    let schema = make_schema_pk_u64_payload_string();
    let dir = tempfile::tempdir().unwrap();
    let mut pt = scratch_table(dir.path(), schema);
    let long1: &[u8] = b"the stored payload, long enough for the heap";
    let long2: &[u8] = b"the replacing payload, also on the heap";
    let long3: &[u8] = b"a fresh key's payload, on the heap as well";

    let seed = enforce_unique_pk(&pt, make_batch_bytes(&schema, &[(1, 1, long1)]));
    pt.ingest_owned_batch(seed).unwrap();

    let eff = enforce_unique_pk(
        &pt,
        make_batch_bytes(&schema, &[(1, 1, long2), (2, -1, b""), (3, 1, long3)]),
    );
    assert_eq!(
        zset_of(&eff, &schema),
        zset_of(
            &make_batch_bytes(&schema, &[(1, 1, long2), (1, -1, long1), (3, 1, long3)]),
            &schema
        ),
    );
    assert_eq!(eff.len(), 3);
}