dbsp 0.287.0

Continuous streaming analytics 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
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

//! Source operator that reads data from a CSV file.

// TODO:
// - Error handling (currently we just panic on reader error or deserialization
//   error).
// - Batching (don't read the whole file in one clock cycle)
// - Async implementation (wait for data to become available in the reader)
// - Sharded implementation (currently we feed all data on worker 0).

use crate::utils::Tup2;
use crate::{
    DBData, OrdZSet, Runtime, ZWeight,
    circuit::{
        Scope,
        operator_traits::{Operator, SourceOperator},
    },
};
use csv::Reader as CsvReader;
use serde::Deserialize;
use std::{borrow::Cow, io::Read, marker::PhantomData};

/// A source operator that reads records of type `T` from a CSV file.
///
/// The operator reads the entire file and yields its contents
/// in the first clock cycle as a Z-set with unit weights.
pub struct CsvSource<R, T> {
    reader: CsvReader<R>,
    time: usize,
    _t: PhantomData<fn(&T)>,
}

impl<R, T> CsvSource<R, T>
where
    R: Read,
{
    /// Create a [`CsvSource`] instance from any reader using
    /// default `CsvReader` settings.
    pub fn from_reader(reader: R) -> Self {
        Self::from_csv_reader(CsvReader::from_reader(reader))
    }

    /// Create a [`CsvSource`] from a pre-configured `CsvReader`.
    pub fn from_csv_reader(reader: CsvReader<R>) -> Self {
        Self {
            reader,
            time: 0,
            _t: PhantomData,
        }
    }
}

impl<R, T> Operator for CsvSource<R, T>
where
    R: 'static,
    T: 'static,
{
    fn name(&self) -> Cow<'static, str> {
        Cow::from("CsvSource")
    }
    fn clock_start(&mut self, _scope: Scope) {
        self.time = 0;
    }
    fn fixedpoint(&self, _scope: Scope) -> bool {
        self.time >= 2
    }
}

impl<R, T> SourceOperator<OrdZSet<T>> for CsvSource<R, T>
where
    T: DBData + for<'de> Deserialize<'de> + 'static,
    R: Read + 'static,
{
    async fn eval(&mut self) -> OrdZSet<T> {
        let source = if self.time == 0 && Runtime::worker_index() == 0 {
            let mut data = Vec::<Tup2<Tup2<T, ()>, ZWeight>>::new();

            for x in self.reader.deserialize() {
                data.push(Tup2(Tup2(x.unwrap(), ()), 1));
            }

            OrdZSet::<T>::from_tuples((), data)
        } else {
            OrdZSet::<T>::empty()
        };
        self.time += 1;

        source
    }
}

#[cfg(test)]
mod test {
    use crate::operator::CsvSource;
    use crate::utils::Tup3;
    use crate::{Circuit, OrdZSet, RootCircuit, zset};
    use csv::ReaderBuilder;

    #[test]
    fn test_csv_reader() {
        let circuit = RootCircuit::build(move |circuit| {
            let expected = zset! {
                Tup3(18, 3, 237641) => 1,
                Tup3(237641, 4, 18) => 1,
                Tup3(18, 5, 21) => 1,
                Tup3(18, 5, 22) => 1,
                Tup3(18, 5, 23) => 1,
                Tup3(18, 5, 24) => 1,
                Tup3(18, 5, 25) => 1,
            };
            let csv_data = "\
18,3,237641
237641,4,18
18,5,21
18,5,22
18,5,23
18,5,24
18,5,25
";
            let reader = ReaderBuilder::new()
                .delimiter(b',')
                .has_headers(false)
                .from_reader(csv_data.as_bytes());
            circuit
                .add_source(CsvSource::from_csv_reader(reader))
                .inspect(move |data: &OrdZSet<Tup3<u64, u64, u64>>| assert_eq!(data, &expected));
            Ok(())
        })
        .unwrap()
        .0;

        circuit.transaction().unwrap();
    }
}