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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

// Drives the writer, so it needs the `write` feature.
#![cfg(feature = "write")]

//! The encoder version marker: the one thing that turns a silent
//! disagreement between the writing and reading encoder into an error.

use std::collections::BTreeMap;

use dendro::archive::{Archive, ArchiveMut, SourceMeta, WalRow};
use dendro::keys;
use dendro::read;
use dendro::segment::{EncodeResult, Segment, SegmentEncoder};
use dendro::writer::Writer;
use dendro::Error;

struct Tags(&'static str);
impl SegmentEncoder for Tags {
    fn encode(&self, _stream: &str, rows: &[WalRow]) -> EncodeResult {
        if rows.is_empty() {
            return Ok(None);
        }
        let ts: Vec<String> = rows.iter().map(|r| r.ts.to_string()).collect();
        Ok(Some(Segment {
            bytes: ts.join(",").into_bytes(),
            rows: rows.len() as u64,
            first_ts: rows[0].ts,
            last_ts: rows[rows.len() - 1].ts,
            index: None,
        }))
    }
    fn version(&self) -> Option<&str> {
        if self.0.is_empty() {
            None
        } else {
            Some(self.0)
        }
    }
}

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

fn row(ts: i64) -> WalRow {
    WalRow {
        stream: "s".to_string(),
        ts,
        wall_offset: 0,
        row: vec![1],
    }
}

fn written_with(path: &std::path::Path, version: &'static str) -> i64 {
    let (archive, mut w) = Writer::single(path, Box::new(Tags(version)), source()).unwrap();
    let id = w.source_id();
    w.wal(vec![row(1)]).unwrap();
    archive.finalize_single(w, (1, 0)).unwrap();
    id
}

#[test]
fn the_writing_encoders_version_is_recorded_and_a_mismatch_is_refused() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("e.dendro");
    let id = written_with(&path, "v1");
    let db = Archive::open(&path).unwrap();
    assert_eq!(
        db.source_metadata(id)
            .unwrap()
            .get(keys::ENCODER)
            .map(String::as_str),
        Some("v1")
    );

    // The same version reads; a different one is refused on every path.
    assert!(read::read_archive(&db, &Tags("v1")).is_ok());
    let is_mismatch = |e: Error| matches!(e, Error::EncoderMismatch { wrote, reading, .. } if wrote == "v1" && reading == "v2");
    assert!(is_mismatch(
        read::read_archive(&db, &Tags("v2")).unwrap_err()
    ));
    assert!(is_mismatch(
        read::stream_segments(&db, id, "s", &Tags("v2")).unwrap_err()
    ));
    assert!(is_mismatch(
        read::probe(&db, id, "s", &Tags("v2")).unwrap_err()
    ));
    assert!(is_mismatch(
        read::stream_range(&db, id, "s", 0, 10, &Tags("v2")).unwrap_err()
    ));

    // An encoder that does not version itself is never checked.
    assert!(read::read_archive(&db, &Tags("")).is_ok());
    drop(db);

    // Nor can a different version resume the source, or copy it (the tail
    // is re-encoded on the way across).
    let mut archive = Writer::open(&path, Box::new(Tags("v2"))).unwrap();
    assert!(is_mismatch(archive.resume_source(id, 100).unwrap_err()));
    drop(archive);
    let src = Archive::open(&path).unwrap();
    let mut dst = ArchiveMut::create(&dir.path().join("copy.dendro")).unwrap();
    let err = dst
        .transaction(|tx| {
            dendro::rewrite::copy_sources_into(
                &src,
                tx,
                &dendro::rewrite::CopySpec::everything(),
                &Tags("v2"),
            )
        })
        .unwrap_err();
    assert!(is_mismatch(err));
}

/// A source from before the key, or written by an unversioned encoder, has
/// nothing to compare and is read by anything.
#[test]
fn an_unversioned_source_is_read_by_any_encoder() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("e.dendro");
    let id = written_with(&path, "");
    let db = Archive::open(&path).unwrap();
    assert!(!db.source_metadata(id).unwrap().contains_key(keys::ENCODER));
    assert!(read::read_archive(&db, &Tags("v9")).is_ok());
}