palimpsest-dataflow 0.1.1

A Postgres WAL-backed live query sync engine.
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

#![allow(clippy::all, clippy::nursery, clippy::pedantic)]

use timely::dataflow::operators::generic::OperatorInfo;
use timely::progress::{frontier::AntichainRef, Antichain};

use palimpsest_dataflow::trace::cursor::Cursor;
use palimpsest_dataflow::trace::implementations::{ValBatcher, ValBuilder, ValSpine};
use palimpsest_dataflow::trace::{Batcher, Trace, TraceReader};

type IntegerTrace = ValSpine<u64, u64, usize, i64>;
type IntegerBuilder = ValBuilder<u64, u64, usize, i64>;

fn get_trace() -> ValSpine<u64, u64, usize, i64> {
    let op_info = OperatorInfo::new(0, 0, [].into());
    let mut trace = IntegerTrace::new(op_info, None, None);
    {
        let mut batcher = ValBatcher::<u64, u64, usize, i64>::new(None, 0);

        batcher.push_container(&mut vec![((1, 2), 0, 1), ((2, 3), 1, 1), ((2, 3), 2, -1)]);

        let batch_ts = &[1, 2, 3];
        let batches = batch_ts
            .iter()
            .map(move |i| batcher.seal::<IntegerBuilder>(Antichain::from_elem(*i)));
        for b in batches {
            trace.insert(b);
        }
    }
    trace
}

#[test]
fn test_trace() {
    let mut trace = get_trace();

    let (mut cursor1, storage1) = trace.cursor_through(AntichainRef::new(&[1])).unwrap();
    let vec_1 = cursor1.to_vec(&storage1, |k| k.clone(), |v| v.clone());
    assert_eq!(vec_1, vec![((1, 2), vec![(0, 1)])]);

    let (mut cursor2, storage2) = trace.cursor_through(AntichainRef::new(&[2])).unwrap();
    let vec_2 = cursor2.to_vec(&storage2, |k| k.clone(), |v| v.clone());
    println!("--> {:?}", vec_2);
    assert_eq!(vec_2, vec![((1, 2), vec![(0, 1)]), ((2, 3), vec![(1, 1)]),]);

    let (mut cursor3, storage3) = trace.cursor_through(AntichainRef::new(&[3])).unwrap();
    let vec_3 = cursor3.to_vec(&storage3, |k| k.clone(), |v| v.clone());
    assert_eq!(
        vec_3,
        vec![((1, 2), vec![(0, 1)]), ((2, 3), vec![(1, 1), (2, -1)]),]
    );

    let (mut cursor4, storage4) = trace.cursor();
    let vec_4 = cursor4.to_vec(&storage4, |k| k.clone(), |v| v.clone());
    assert_eq!(vec_4, vec_3);
}