dendro 0.2.1

A segmented-parquet archive with a write-ahead log, in a single SQLite file
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

//! One encoder contract for the three places the crate runs an encoder.
//!
//! The writer refused an over-claiming encoder at seal and a copy refused it
//! on the way across; the read path accepted it. So a reader and the next
//! seal could disagree about a live tail — the symmetry the crate depends on,
//! broken silently.

use std::collections::BTreeMap;

use dendro::archive::{ArchiveMut, SourceMeta, WalRow};
use dendro::read;
use dendro::segment::{materialize, EncodeResult, Segment, SegmentEncoder};
use dendro::Error;

fn source() -> SourceMeta {
    SourceMeta {
        labels: BTreeMap::new(),
        metadata: BTreeMap::new(),
        clock_anchor_wall_ns: 0,
    }
}

fn rows(ts: &[i64]) -> Vec<WalRow> {
    ts.iter()
        .map(|&ts| WalRow {
            stream: "s".to_string(),
            ts,
            wall_offset: 0,
            row: vec![1],
        })
        .collect()
}

/// Keeps the first and last row and drops what is between: the shape a
/// `(rows, first_ts, last_ts)` triple cannot catch and counting can.
struct DropsMiddle;
impl SegmentEncoder for DropsMiddle {
    fn encode(&self, _stream: &str, rows: &[WalRow]) -> EncodeResult {
        if rows.len() < 3 {
            return Ok(None);
        }
        Ok(Some(Segment {
            bytes: b"first,last".to_vec(),
            rows: 2,
            first_ts: rows[0].ts,
            last_ts: rows[rows.len() - 1].ts,
            index: None,
        }))
    }
}

/// Claims one more row than it was given.
struct OverClaims;
impl SegmentEncoder for OverClaims {
    fn encode(&self, _stream: &str, rows: &[WalRow]) -> EncodeResult {
        if rows.is_empty() {
            return Ok(None);
        }
        Ok(Some(Segment {
            bytes: b"x".to_vec(),
            rows: rows.len() as u64 + 1,
            first_ts: rows[0].ts,
            last_ts: rows[rows.len() - 1].ts,
            index: None,
        }))
    }
}

struct Panics;
impl SegmentEncoder for Panics {
    fn encode(&self, _stream: &str, _rows: &[WalRow]) -> EncodeResult {
        panic!("no");
    }
}

#[test]
fn the_shared_check_refuses_a_hole_an_overclaim_and_a_panic() {
    let input = rows(&[1, 2, 3]);
    assert!(matches!(
        materialize(&DropsMiddle, "s", &input).unwrap_err(),
        Error::EncoderContract { .. }
    ));
    assert!(matches!(
        materialize(&OverClaims, "s", &input).unwrap_err(),
        Error::EncoderContract { .. }
    ));
    match materialize(&Panics, "s", &input).unwrap_err() {
        Error::Encoder { source, .. } => assert!(source.to_string().contains("panicked: no")),
        other => panic!("{other}"),
    }
    // Empty input never reaches the encoder — a panicking one is not called.
    assert!(materialize(&Panics, "s", &[]).unwrap().is_none());
}

/// The read path runs the same check: an encoder the seal would refuse is
/// refused when a reader materializes a live tail, rather than handing back
/// bytes the next seal will disagree with.
#[test]
fn a_reader_refuses_what_the_seal_would_refuse() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("m.dendro");
    let mut db = ArchiveMut::create(&path).unwrap();
    let id = db.insert_source(&source()).unwrap();
    db.insert_wal_rows(id, &rows(&[1, 2, 3])).unwrap();

    let err = read::stream_segments(&db, id, "s", &DropsMiddle).unwrap_err();
    assert!(matches!(err, Error::EncoderContract { .. }), "{err}");
    let err = read::read_archive(&db, &OverClaims).unwrap_err();
    assert!(matches!(err, Error::EncoderContract { .. }), "{err}");
}