miradb-core 0.0.3

Mira storage engine: OTLP -> Arrow encoding, immutable block writer, zero-copy mmap reader
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
461
462
//! Block-level "might contain this" filters.
//!
//! Block names prune by time, which answers every question that has a time
//! bound and none of the ones that do not. Two queries have no useful time
//! bound at all, and both of them read every block on disk without a filter:
//!
//! * *"Every span of trace `0000…6eaa`."* You do not know when the trace
//!   happened — that is why you are looking it up.
//! * *"Any record with `k8s.pod.name = api-7f9`."* A predicate that matches
//!   nothing has no early exit: `limit` never fills, so the scan runs to the
//!   end of retention to prove a negative.
//!
//! A sidecar per block, written next to the Arrow tables at publish and read
//! before any of them, turns both into "open the blocks that can answer".
//! Measured on this machine, each query against its own corpus (section 11):
//!
//! ```text
//!                                       blocks   rows scanned   time
//! 4.1 GB / 25M spans / 84 blocks
//! every span of one trace, without         84      25,000,000   14.3 s
//!                          with             1         212,992    250 ms cold, 20 warm
//! 6.4 GB / 25M logs / 69 blocks
//! absent attribute value,  without         69      24,961,024   10.4 s
//!                          with             0               0    71 ms cold, 5.7 warm
//! ```
//!
//! [`TRACE_IDX`] costs 65 KB per block, 0.13% of the data. [`ATTR_IDX`] is
//! sized by the block's *distinct* `(key, value)` pairs, so on that corpus it is
//! 70 bytes per block and it grows only where pruning pays best.
//!
//! Both go through a splitmix64 finalizer before Kirsch-Mitzenmacher double
//! hashing, and that is not optional. W3C only requires a trace id to be
//! non-zero, and plenty of real ones are structured rather than random: X-Ray
//! puts an epoch in the first four bytes, a counter-derived id leaves whole
//! bytes constant. The bit index is `(h1 + i*h2) & mask`, so it reads only the
//! *low* bits of the halves; ids that vary in their middle bytes would map every
//! trace in a block onto the same handful of bits and the filter would answer
//! "maybe" to everything. Five ops per half buys immunity to that.
//!
//! Everything here fails open. A short, corrupt or unrecognized file means
//! "scan the block", never "skip it" — a false positive costs one wasted block
//! read, a false negative silently loses data from a correct query.

use std::collections::HashSet;

use arrow_array::{Array, FixedSizeBinaryArray};

/// Filenames inside a published block. Part of the on-disk format.
pub const TRACE_IDX: &str = "trace.idx";
pub const ATTR_IDX: &str = "attr.idx";

const MAGIC: [u8; 4] = *b"MBLM";
const VERSION: u8 = 1;
/// magic 4 | version 1 | k 1 | flags 1 | pad 1 | words 4 | crc32 4
const HEADER: usize = 16;

/// Set in [`ATTR_IDX`] when the block holds at least one `double`-typed
/// attribute.
///
/// Doubles are the one type not indexed. The canonical text of a float is not
/// unique — `200`, `200.0` and `2e2` are the same number and three different
/// strings, and a query writing one of the forms the writer did not would be
/// pruned away. So the block declares that it has some, and any query whose
/// value could be read as a number scans it. Blocks with no double attributes,
/// which is most of them, pay nothing for the rule.
pub const HAS_DOUBLE: u8 = 1;

/// Bits per inserted key. 10 with k=7 is the textbook ~0.8% false positive
/// rate.
const BITS_PER_KEY: usize = 10;
const K: u32 = 7;

/// Build a filter over a `FixedSizeBinary(16)` column. `None` when there is
/// nothing to index, in which case no file is written and the reader's
/// fail-open path scans the block.
///
/// The column holds one row per *span*, and a trace is eight of them. Sizing on
/// the row count would make every filter eight times bigger than the key set it
/// holds, and the read path pays that on every block it probes. Runs of the
/// same id collapse — spans of a trace arrive in one export and land adjacent —
/// which is a 16-byte compare per row against a hash set that would cost an
/// allocation and a probe. Interleaved traces just fall back to the row count,
/// which is the size we would have had anyway.
pub fn build(ids: &FixedSizeBinaryArray) -> Option<Vec<u8>> {
    if ids.value_length() != 16 {
        return None;
    }
    let keys = || Runs {
        ids,
        i: 0,
        prev: None,
    };
    let n = keys().count();
    encode(n, 0, keys().map(halves))
}

/// A set of hashed keys, deduplicated so the filter is sized by what it holds.
///
/// The trace filter gets its dedup from adjacency and needs none of this. An
/// attribute filter cannot: `http.route = GET /cart` recurs on a third of the
/// rows in a block and nowhere near adjacently, so without a set the filter
/// would be sized for a million rows holding five values.
#[derive(Default)]
pub struct Keys {
    seen: HashSet<(u64, u64)>,
    flags: u8,
    full: bool,
}

/// Beyond this many distinct keys the filter stops being a rounding error on
/// the block — 1M keys is 1.25 MB — and a block that diverse prunes little
/// anyway. Past it we write nothing, which the reader reads as "scan me".
const MAX_KEYS: usize = 1 << 20;

impl Keys {
    pub fn insert(&mut self, h: (u64, u64)) {
        if self.seen.len() < MAX_KEYS {
            self.seen.insert(h);
        } else {
            self.full = true;
        }
    }

    pub fn flag(&mut self, bit: u8) {
        self.flags |= bit;
    }

    pub fn build(&self) -> Option<Vec<u8>> {
        if self.full {
            return None;
        }
        encode(self.seen.len(), self.flags, self.seen.iter().copied())
    }
}

fn encode(n: usize, flags: u8, keys: impl Iterator<Item = (u64, u64)>) -> Option<Vec<u8>> {
    if n == 0 {
        return None;
    }
    let words = ((n * BITS_PER_KEY).div_ceil(64)).next_power_of_two();
    let mut bits = vec![0u64; words];
    // A power-of-two word count makes the modulo an `and`, which matters: this
    // runs seven times per key on the seal path.
    let mask = (words as u64 * 64) - 1;
    for (h1, h2) in keys {
        for i in 0..K {
            let bit = h1.wrapping_add((i as u64).wrapping_mul(h2)) & mask;
            bits[bit as usize / 64] |= 1 << (bit % 64);
        }
    }

    let mut out = Vec::with_capacity(HEADER + words * 8);
    out.extend_from_slice(&MAGIC);
    out.push(VERSION);
    out.push(K as u8);
    out.push(flags);
    out.push(0);
    out.extend_from_slice(&(words as u32).to_le_bytes());
    let body: Vec<u8> = bits.iter().flat_map(|w| w.to_le_bytes()).collect();
    out.extend_from_slice(&crc32fast::hash(&body).to_le_bytes());
    out.extend_from_slice(&body);
    Some(out)
}

/// A filter checked out, with its header validated once.
pub struct Filter<'a> {
    body: &'a [u8],
    k: u32,
    mask: u64,
    pub flags: u8,
}

impl<'a> Filter<'a> {
    /// `None` for anything unreadable, which every caller must treat as "scan
    /// the block".
    pub fn open(file: &'a [u8]) -> Option<Filter<'a>> {
        if file.len() < HEADER || file[..4] != MAGIC || file[4] != VERSION {
            return None;
        }
        let k = file[5] as u32;
        let flags = file[6];
        let words = u32::from_le_bytes(file[8..12].try_into().expect("4 bytes")) as usize;
        let crc = u32::from_le_bytes(file[12..16].try_into().expect("4 bytes"));
        let body = &file[HEADER..];
        if k == 0 || !words.is_power_of_two() || body.len() != words * 8 {
            return None;
        }
        // The one check that costs something — 65 KB of CRC against a block
        // read of tens of megabytes. Skipping a block on the word of a corrupt
        // filter is the one outcome worth paying to avoid.
        if crc32fast::hash(body) != crc {
            return None;
        }
        Some(Filter {
            body,
            k,
            mask: (words as u64 * 64) - 1,
            flags,
        })
    }

    pub fn may_contain(&self, (h1, h2): (u64, u64)) -> bool {
        for i in 0..self.k {
            let bit = h1.wrapping_add((i as u64).wrapping_mul(h2)) & self.mask;
            let word = u64::from_le_bytes(
                self.body[(bit as usize / 64) * 8..][..8]
                    .try_into()
                    .expect("slice of 8"),
            );
            if word & (1 << (bit % 64)) == 0 {
                return false;
            }
        }
        true
    }
}

/// Could this block contain `id`? Anything unreadable answers yes.
pub fn may_contain(file: &[u8], id: &[u8; 16]) -> bool {
    Filter::open(file).is_none_or(|f| f.may_contain(halves(id)))
}

/// Hash one `(key, canonical value)` pair for [`ATTR_IDX`].
///
/// The value is the *text* of the scalar, not its typed bytes, and that is what
/// makes one probe enough. A query scalar is compared against whatever type the
/// column happens to hold — `{attr: http.status_code, eq: "200"}` matches a
/// stored integer `200`, because the comparison parses rather than requiring the
/// query to know the SDK's choice. Indexing the decimal text makes the filter
/// agree with that rule instead of quietly disagreeing with it, which would be a
/// false negative and therefore a lost row.
pub fn attr_hash(key: &str, value: &[u8]) -> (u64, u64) {
    let k = crate::identity::hash64(key.as_bytes());
    let v = crate::identity::hash64(value);
    (
        mix(k ^ v.rotate_left(17)),
        mix(k.wrapping_mul(0x9e37_79b9_7f4a_7c15) ^ v) | 1,
    )
}

/// Non-null values, with adjacent duplicates dropped.
struct Runs<'a> {
    ids: &'a FixedSizeBinaryArray,
    i: usize,
    prev: Option<&'a [u8]>,
}

impl<'a> Iterator for Runs<'a> {
    type Item = &'a [u8];

    fn next(&mut self) -> Option<&'a [u8]> {
        while self.i < self.ids.len() {
            let i = self.i;
            self.i += 1;
            if self.ids.is_null(i) {
                continue;
            }
            let v = self.ids.value(i);
            if self.prev != Some(v) {
                self.prev = Some(v);
                return Some(v);
            }
        }
        None
    }
}

fn halves(id: &[u8]) -> (u64, u64) {
    let h1 = mix(u64::from_le_bytes(id[..8].try_into().expect("16-byte id")));
    // Odd, so the probe sequence walks the whole filter instead of landing on
    // the same bit whenever the second half happens to be even.
    let h2 = mix(u64::from_le_bytes(
        id[8..16].try_into().expect("16-byte id"),
    )) | 1;
    (h1, h2)
}

/// splitmix64's finalizer: every input bit reaches every output bit.
fn mix(mut x: u64) -> u64 {
    x ^= x >> 30;
    x = x.wrapping_mul(0xbf58_476d_1ce4_e5b9);
    x ^= x >> 27;
    x = x.wrapping_mul(0x94d0_49bb_1331_11eb);
    x ^ (x >> 31)
}

#[cfg(test)]
mod tests {
    use super::*;
    use arrow_array::FixedSizeBinaryArray;

    fn id(n: u64) -> [u8; 16] {
        let mut b = [0u8; 16];
        // Not a counter: the filter's whole premise is that trace ids are
        // uniform, so a test over 0,1,2,… would measure a case that cannot
        // happen and hide a probe sequence that clusters.
        b[..8].copy_from_slice(&crate::identity::hash64(&n.to_le_bytes()).to_le_bytes());
        b[8..].copy_from_slice(&crate::identity::hash64(&(!n).to_le_bytes()).to_le_bytes());
        b
    }

    fn filter(n: u64) -> Vec<u8> {
        let ids: Vec<[u8; 16]> = (0..n).map(id).collect();
        let arr = FixedSizeBinaryArray::try_from_iter(ids.iter().map(|v| v.as_slice())).unwrap();
        build(&arr).unwrap()
    }

    /// Everything inserted must be found — a false negative is a lost trace —
    /// and the false positive rate has to be near the 0.8% the sizing promises,
    /// or the filter is a 16 KB file that skips nothing.
    #[test]
    fn no_false_negatives_and_few_false_positives() {
        let n = 10_000u64;
        let f = filter(n);
        // The size the two "skips nothing" comments quote: 10 bits a key,
        // rounded up to a power-of-two word count, plus the header.
        assert_eq!(f.len(), HEADER + 2048 * 8);
        for i in 0..n {
            assert!(may_contain(&f, &id(i)), "false negative at {i}");
        }
        let probes = 100_000u64;
        let fp = (n..n + probes).filter(|&i| may_contain(&f, &id(i))).count();
        let rate = fp as f64 / probes as f64;
        assert!(rate < 0.02, "false positive rate {rate}");
    }

    /// Every way the file can be wrong has to answer "scan it". This is the
    /// property the whole thing rests on: a filter is an optimization, and an
    /// optimization that can lose data is a bug.
    #[test]
    fn a_damaged_filter_says_maybe() {
        let f = filter(1_000);
        let probe = id(999_999);
        assert!(!may_contain(&f, &probe), "test needs a known-absent id");

        assert!(may_contain(&[], &probe), "empty");
        assert!(may_contain(&f[..HEADER - 1], &probe), "truncated header");
        assert!(may_contain(&f[..f.len() - 8], &probe), "truncated body");

        let mut bad = f.clone();
        bad[0] = b'X';
        assert!(may_contain(&bad, &probe), "wrong magic");

        let mut bad = f.clone();
        bad[4] = VERSION + 1;
        assert!(may_contain(&bad, &probe), "future version");

        // A word count of zero is the one header field that is not merely
        // wrong but *unrepresentable*: `mask` is `words * 64 - 1`, so a filter
        // that got this far with no bits at all would underflow — a panic in
        // debug, a `u64::MAX` mask and an out-of-bounds body index in release.
        // The power-of-two check is what stands between a single zeroed byte
        // and that, so it is worth its own case rather than being folded into
        // "truncated body".
        let mut bad = f.clone();
        bad[8..12].copy_from_slice(&0u32.to_le_bytes());
        assert!(may_contain(&bad, &probe), "no bits at all");

        // Not a power of two, so the `and` that stands in for the modulo would
        // address words the body does not have.
        let mut bad = f.clone();
        bad[8..12].copy_from_slice(&3u32.to_le_bytes());
        assert!(may_contain(&bad, &probe), "word count not a power of two");

        // k = 0 probes nothing and therefore answers "no" to everything, which
        // is the one answer a filter is never allowed to get wrong.
        let mut bad = f.clone();
        bad[5] = 0;
        assert!(may_contain(&bad, &probe), "no hash functions");

        // A single flipped bit in the bitmap is exactly the case a checksum
        // exists for: it turns a "no" into a wrong "no" with nothing else to
        // notice it.
        let mut bad = f;
        bad[HEADER + 3] ^= 0x40;
        assert!(may_contain(&bad, &probe), "corrupt body");
    }

    /// [`halves`] reads bytes 0..8 and 8..16 of whatever it is handed, so a
    /// column of any other width is an index out of bounds — a panic on the
    /// seal path, which takes the process down with `panic = "abort"` and loses
    /// the whole open block. Refusing to index it writes no file, and no file
    /// means "scan the block".
    #[test]
    fn a_column_that_is_not_a_16_byte_id_indexes_nothing() {
        let narrow = [[7u8; 8].as_slice(), [9u8; 8].as_slice()];
        let arr = FixedSizeBinaryArray::try_from_iter(narrow.into_iter()).unwrap();
        assert_eq!(arr.value_length(), 8);
        assert!(build(&arr).is_none());
        assert!(build(&FixedSizeBinaryArray::new_null(32, 4)).is_none());
    }

    /// Past [`MAX_KEYS`] the filter stops being a rounding error on the block,
    /// so the writer gives up and writes nothing. Giving up has to fail *open*:
    /// the reader sees a block with no sidecar, which is the same thing it sees
    /// for a block published before the filter existed, and it scans.
    ///
    /// The boundary is asserted from both sides because the interesting bug is
    /// off-by-one in the other direction — a filter dropped one key early is
    /// invisible, since the answer is still correct and only slower.
    #[test]
    fn too_many_distinct_keys_write_no_filter_rather_than_a_huge_one() {
        let mut keys = Keys::default();
        for i in 0..MAX_KEYS as u64 {
            keys.insert((i, i | 1));
        }
        assert!(keys.build().is_some(), "MAX_KEYS keys still fit");

        keys.insert((u64::MAX, 1));
        assert!(keys.build().is_none(), "one key past the cap");
    }

    /// The ids the load generator emits, and the shape every counter-derived or
    /// timestamp-prefixed id has: constant bytes at both ends, variation in the
    /// middle. Without the finalizer every one of these lands on the same bits
    /// and the filter says "maybe" to everything — which is not a wrong answer,
    /// just a 16 KB file that skips nothing.
    #[test]
    fn structured_ids_still_spread() {
        let structured = |n: u64| {
            let mut b = [0u8; 16];
            b[..8].copy_from_slice(&(n as u32 as u64).to_be_bytes());
            b[8..].copy_from_slice(&(0x5555_5555_5500_0000 | n).to_be_bytes());
            b
        };
        let ids: Vec<[u8; 16]> = (0..10_000u64).map(structured).collect();
        let arr = FixedSizeBinaryArray::try_from_iter(ids.iter().map(|v| v.as_slice())).unwrap();
        let f = build(&arr).unwrap();

        for i in 0..10_000u64 {
            assert!(may_contain(&f, &structured(i)), "false negative at {i}");
        }
        let probes = 100_000u64;
        let fp = (10_000..10_000 + probes)
            .filter(|&i| may_contain(&f, &structured(i)))
            .count();
        let rate = fp as f64 / probes as f64;
        assert!(rate < 0.02, "false positive rate {rate}");
    }

    /// Eight spans per trace is the ordinary shape of a block, and the filter
    /// has to be sized for the traces, not the spans — otherwise every probe on
    /// the read path pays 8x for nothing.
    #[test]
    fn adjacent_duplicates_do_not_inflate_the_filter() {
        let ids: Vec<[u8; 16]> = (0..1_000u64).flat_map(|n| [id(n); 8]).collect();
        let arr = FixedSizeBinaryArray::try_from_iter(ids.iter().map(|v| v.as_slice())).unwrap();
        let fanned = build(&arr).unwrap();
        assert_eq!(fanned.len(), filter(1_000).len());
        for i in 0..1_000u64 {
            assert!(may_contain(&fanned, &id(i)), "false negative at {i}");
        }
    }

    /// An empty column writes no file, and the reader treats a missing one as
    /// "scan the block" — the same path a block from an older writer takes.
    #[test]
    fn nothing_to_index_writes_nothing() {
        assert!(build(&FixedSizeBinaryArray::new_null(16, 0)).is_none());
        assert!(build(&FixedSizeBinaryArray::new_null(16, 100)).is_none());
    }
}