zrx_stream/stream/operator/
reduce.rs

1// Copyright (c) Zensical LLC <https://zensical.org>
2
3// SPDX-License-Identifier: MIT
4// Third-party contributions licensed under CLA
5
6// Permission is hereby granted, free of charge, to any person obtaining a copy
7// of this software and associated documentation files (the "Software"), to
8// deal in the Software without restriction, including without limitation the
9// rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
10// sell copies of the Software, and to permit persons to whom the Software is
11// furnished to do so, subject to the following conditions:
12
13// The above copyright notice and this permission notice shall be included in
14// all copies or substantial portions of the Software.
15
16// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18// FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE
19// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
21// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
22// IN THE SOFTWARE.
23
24// ----------------------------------------------------------------------------
25
26//! Reduce operator.
27
28use ahash::HashMap;
29use std::marker::PhantomData;
30use zrx_scheduler::action::descriptor::Property;
31use zrx_scheduler::action::output::IntoOutputs;
32use zrx_scheduler::action::{Descriptor, Report};
33use zrx_scheduler::effect::Item;
34use zrx_scheduler::{Id, Value};
35use zrx_store::StoreMut;
36
37use crate::stream::function::SelectFn;
38use crate::stream::value::Collection;
39use crate::stream::Stream;
40
41use super::Operator;
42
43// ----------------------------------------------------------------------------
44// Structs
45// ----------------------------------------------------------------------------
46
47/// Reduce operator.
48struct Reduce<I, T, F, U> {
49    /// Identifier.
50    id: I,
51    /// Operator function.
52    function: F,
53    /// Store of items.
54    store: HashMap<I, T>,
55    /// Type marker.
56    marker: PhantomData<U>,
57}
58
59// ----------------------------------------------------------------------------
60// Implementations
61// ----------------------------------------------------------------------------
62
63impl<I, T> Stream<I, T>
64where
65    I: Id,
66    T: Value + Clone + Eq,
67{
68    pub fn reduce<F, U>(&self, id: I, f: F) -> Stream<I, U>
69    where
70        F: SelectFn<I, dyn Collection<I, T>, Option<U>>,
71        U: Value,
72    {
73        self.workflow.add_operator(
74            [self.id],
75            Reduce {
76                id,
77                function: f,
78                store: HashMap::default(),
79                marker: PhantomData,
80            },
81        )
82    }
83}
84
85// ----------------------------------------------------------------------------
86// Trait implementations
87// ----------------------------------------------------------------------------
88
89impl<I, T, F, U> Operator<I, T> for Reduce<I, T, F, U>
90where
91    I: Id,
92    T: Value + Clone + Eq,
93    F: SelectFn<I, dyn Collection<I, T>, Option<U>>,
94    U: Value,
95{
96    type Item<'a> = Item<&'a I, Option<&'a T>>;
97
98    /// Handles the given item.
99    ///
100    /// Reductions should be used only sparingly, as they require to store all
101    /// items that are flowing through the stream in the operator, because the
102    /// reduction is computed on the entire store. This makes sure that the
103    /// differential semantics of the stream are preserved.
104    ///
105    /// If we'd provide an operator for differential reductions (also known as
106    /// scanning), the user would be responsible for ensuring the differential
107    /// invariant, which might lead to subtle, hard to detect bugs. There are
108    /// several other operators that provide case-by-case scan-like semantics,
109    /// which are almost always a better choice than using a reduction, as they
110    /// are much more efficient and easier to reason about. When this operator
111    /// is used incorrectly, it might lead to unbounded memory consumption,
112    /// so use it with care.
113    #[cfg_attr(
114        feature = "tracing",
115        tracing::instrument(level = "debug", skip_all, fields(id = %item.id))
116    )]
117    fn handle(&mut self, item: Self::Item<'_>) -> impl IntoOutputs<I> {
118        let has_changed = if let Some(data) = item.data {
119            self.store.insert_if_changed(item.id, data)
120        } else {
121            self.store.remove(item.id).is_some()
122        };
123
124        // If the store has changed, we pass it to the operator function in
125        // order to compute a new output value. The operator function returns
126        // an option to indicate the presence or abscence of a value for the
127        // identifier. If nothing has changed, nothing is emitted.
128        if has_changed {
129            self.function.execute(&self.id, &self.store).map(|report| {
130                report.map(|data| Some(Item::new(self.id.clone(), data)))
131            })
132        } else {
133            Ok(Report::new(None))
134        }
135    }
136
137    /// Returns the descriptor.
138    #[inline]
139    fn descriptor(&self) -> Descriptor {
140        Descriptor::builder() // fmt
141            .property(Property::Flush)
142            .build()
143    }
144}
zrx_stream/stream/operator/reduce.rs

zrx_stream/stream/operator/
reduce.rs
