gnitz-store 0.1.8

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
148
149
150
151
152
153
154

//! A delta read, in instructions per row of the band it covers.
//!
//! `cd crates && cargo test -p gnitz-store --release delta_read_bench -- --ignored --nocapture --test-threads=1`

use std::hint::black_box;

use crate::test_support::{cmp_const, fed_view, map_of, relation_fixture, RelationFixture, TID};
use gnitz_expr::{CmpOp, LogicalProgram};
use gnitz_wire::{PkKeys, ReadBound, ReadSink, ReadSpec, SinkKind, TypeCode};
use gnitz_zset::repr::BatchBuilder;
use gnitz_zset::schema::{SchemaColumn, SchemaDescriptor};

/// `id U64 PK` and `payload` I64 columns.
fn schema(payload: usize) -> SchemaDescriptor {
    let mut cols = vec![SchemaColumn::new(TypeCode::U64, false)];
    cols.extend((0..payload).map(|_| SchemaColumn::new(TypeCode::I64, false)));
    SchemaDescriptor::new(&cols, &[0])
}

/// A fed view `id | c0 = id % 512 | c1 = round | c2 = id` whose feed holds
/// `rounds` rounds, each the ids `0..per_round`.
fn fixture(rounds: u64, per_round: u64) -> RelationFixture {
    let s = schema(3);
    let kind = fed_view(1 << 32);
    let mut r = relation_fixture(kind, s, &[], []);
    for round in 2..2 + rounds {
        let mut bb = BatchBuilder::new(&s);
        for id in 0..per_round {
            bb.begin_row(id as u128, 1);
            bb.put_u64(id % 512);
            bb.put_u64(round);
            bb.put_u64(id);
            bb.end_row();
        }
        r.ingest_at(TID, bb.finish(), Some(round), false).unwrap();
    }
    r
}

#[test]
#[ignore = "benchmark; run with --release --ignored --nocapture --test-threads=1"]
fn delta_read_bench() {
    let counter = gnitz_foundation::perf::Counter::instructions();
    let rows = |map| ReadSink { map, kind: SinkKind::Rows { cut: None } };
    let slim = || map_of(LogicalProgram::copy_cols(&[3]), &schema(1));
    let lt = |lit| cmp_const(CmpOp::Lt, 1, lit).to_blob_bytes();
    let keys = |n: u64| {
        let ks: Vec<[u8; 8]> = (0..n).map(|i| i.to_be_bytes()).collect();
        ReadBound::PkSet(PkKeys::from_keys(8, ks.iter().map(|k| k.as_slice())))
    };
    let spec = |bound, predicate, sink| ReadSpec { bound, predicate, sink };
    for (rounds, per_round) in [(1u64, 262144u64), (64, 4096), (4096, 64)] {
        let r = fixture(rounds, per_round);
        let band = rounds * per_round;
        println!("--- {rounds} rounds of {per_round} rows");
        let cells: Vec<(&str, ReadSpec, usize)> = vec![
            ("the view's rows", ReadSpec::all_rows(ReadBound::None), 3),
            (
                "c0 < 256, one row in two",
                spec(ReadBound::None, lt(256), rows(None)),
                3,
            ),
            ("c0 < 1, one row in 512", spec(ReadBound::None, lt(1), rows(None)), 3),
            ("one column", spec(ReadBound::None, vec![], rows(slim())), 1),
            ("one key", spec(keys(1), vec![], rows(None)), 3),
            ("64 keys", spec(keys(64), vec![], rows(None)), 3),
            ("64 keys, c0 < 256", spec(keys(64), lt(256), rows(None)), 3),
        ];
        for (label, spec, payload) in cells {
            let layout = schema(payload).layout_digest();
            let run = || r.delta_read(TID, 1, spec.clone(), layout).unwrap();
            black_box(run());
            let (reply, instructions) = counter.measure(run);
            println!(
                "delta_read_bench {label:<28} {:>7} rows out {:>8.1} instr/band row",
                reply.len(),
                instructions as f64 / band as f64
            );
        }
    }
    // A read's fixed cost: one round of one row under a predicate and a map.
    let r = fixture(1, 1);
    let spec = spec(ReadBound::None, lt(256), rows(slim()));
    let layout = schema(1).layout_digest();
    let run = || r.delta_read(TID, 1, spec.clone(), layout).unwrap();
    black_box(run());
    const CALLS: u64 = 1000;
    let (out, instructions) = counter.measure(|| (0..CALLS).map(|_| run().len()).sum::<usize>());
    println!(
        "delta_read_bench one row, predicate + map       {out:>7} rows out {:>8.1} instr/call",
        instructions as f64 / CALLS as f64
    );
}

/// A fed view over `keys` U64 key columns, then `c0 = id % 512`, then `s` where
/// `string`, every other one out of line; its feed holds `rounds` rounds of the
/// ids `0..per_round`.
fn keys_fixture(keys: usize, string: bool, rounds: u64, per_round: u64) -> (RelationFixture, SchemaDescriptor) {
    let c = SchemaColumn::new;
    let mut cols = vec![c(TypeCode::U64, false); keys];
    cols.push(c(TypeCode::I64, false));
    if string {
        cols.push(c(TypeCode::String, false));
    }
    let pk: Vec<u32> = (0..keys as u32).collect();
    let s = SchemaDescriptor::new(&cols, &pk);
    let kind = fed_view(1 << 32);
    let mut r = relation_fixture(kind, s, &[], []);
    for round in 2..2 + rounds {
        let mut bb = BatchBuilder::new(&s);
        for id in 0..per_round {
            let mut key = vec![7u128; keys];
            key[0] = id as u128;
            bb.begin_row_natives(&key, 1);
            bb.put_u64(id % 512);
            if string {
                match id % 2 {
                    0 => bb.put_string(&format!("a-fairly-long-out-of-line-value-{id}")),
                    _ => bb.put_string("short"),
                }
            }
            bb.end_row();
        }
        r.ingest_at(TID, bb.finish(), Some(round), false).unwrap();
    }
    (r, s)
}

#[test]
#[ignore = "benchmark; run with --release --ignored --nocapture --test-threads=1"]
fn delta_read_keys_bench() {
    let counter = gnitz_foundation::perf::Counter::instructions();
    for (keys, string) in [(1, false), (3, false), (4, false), (1, true), (3, true)] {
        let lt = |lit| cmp_const(CmpOp::Lt, keys as u32, lit).to_blob_bytes();
        for (rounds, per_round) in [(1u64, 262144u64), (64, 4096)] {
            let (r, s) = keys_fixture(keys, string, rounds, per_round);
            let band = rounds * per_round;
            for (label, predicate) in [("whole", vec![]), ("half", lt(256)), ("1/512", lt(1))] {
                let spec = ReadSpec {
                    predicate,
                    ..ReadSpec::all_rows(ReadBound::None)
                };
                let run = || r.delta_read(TID, 1, spec.clone(), s.layout_digest()).unwrap();
                black_box(run());
                let (reply, instructions) = counter.measure(run);
                println!(
                    "delta_read_keys_bench {keys} keys string={string:<5} {rounds:>2} rounds {label:<6} {:>7} rows out {:>8.1} instr/band row",
                    reply.len(),
                    instructions as f64 / band as f64
                );
            }
        }
    }
}