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

//! Defines an operator which negates the input value

use crate::{
    algebra::NegByRef,
    circuit::{
        Circuit, OwnershipPreference, Scope, Stream,
        operator_traits::{Operator, UnaryOperator},
    },
};
use std::{borrow::Cow, marker::PhantomData, ops::Neg};

impl<C, D> Stream<C, D>
where
    D: Clone + 'static + Neg<Output = D> + NegByRef,
    C: Circuit,
{
    /// Returns a stream with the same type as `self` in which each value is
    /// negated. Negating an indexed Z-set negates all the weights.
    #[track_caller]
    pub fn neg(&self) -> Stream<C, D> {
        let negated = self
            .circuit()
            .add_unary_operator(UnaryMinus::new(), &self.try_sharded_version());

        // If the input stream is sharded then the negated stream is sharded
        negated.mark_sharded_if(self);
        negated
    }
}

pub struct UnaryMinus<T> {
    phantom: PhantomData<T>,
}

impl<T> UnaryMinus<T> {
    pub fn new() -> Self {
        Self {
            phantom: PhantomData,
        }
    }
}

impl<T> Operator for UnaryMinus<T>
where
    T: 'static,
{
    fn name(&self) -> Cow<'static, str> {
        Cow::from("UnaryMinus")
    }

    fn fixedpoint(&self, _scope: Scope) -> bool {
        true
    }
}

impl<T> Default for UnaryMinus<T> {
    fn default() -> Self {
        Self::new()
    }
}

impl<T> UnaryOperator<T, T> for UnaryMinus<T>
where
    T: 'static + NegByRef + Neg<Output = T> + Clone,
{
    async fn eval(&mut self, input: &T) -> T {
        input.neg_by_ref()
    }

    async fn eval_owned(&mut self, i: T) -> T {
        i.neg()
    }

    fn input_preference(&self) -> OwnershipPreference {
        OwnershipPreference::PREFER_OWNED
    }
}

#[cfg(test)]
mod test {
    use crate::{
        Circuit, RootCircuit, algebra::HasZero, operator::Generator, typed_batch::OrdZSet, zset,
    };

    #[test]
    fn zset_sum() {
        let build_circuit = |circuit: &RootCircuit| {
            let mut s = <OrdZSet<u64> as HasZero>::zero();
            let source = circuit.add_source(Generator::new(move || {
                let res = s.clone();
                s = s.merge(&zset! { 5 => 1, 6 => 2 });
                res
            }));
            source
                .neg()
                .plus(&source)
                .inspect(|s| assert_eq!(s, &<OrdZSet<_> as HasZero>::zero()));
            Ok(source)
        };

        let circuit = RootCircuit::build(move |circuit| build_circuit(circuit))
            .unwrap()
            .0;

        for _ in 0..100 {
            circuit.transaction().unwrap();
        }
    }
}