zrx_stream/stream/operator/
delta_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//! Delta 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;
33use zrx_scheduler::effect::Item;
34use zrx_scheduler::{Id, Value};
35use zrx_store::StoreMutRef;
36
37use crate::stream::function::SelectFn;
38use crate::stream::value::{Collection, Delta};
39use crate::stream::Stream;
40
41use super::Operator;
42
43// ----------------------------------------------------------------------------
44// Structs
45// ----------------------------------------------------------------------------
46
47/// Delta reduce operator.
48struct DeltaReduce<I, T, F, U> {
49    /// Operator function.
50    function: F,
51    /// Store of items.
52    store: HashMap<I, HashMap<I, T>>,
53    /// Type marker.
54    marker: PhantomData<U>,
55}
56
57// ----------------------------------------------------------------------------
58// Implementations
59// ----------------------------------------------------------------------------
60
61impl<I, T> Stream<I, Delta<I, T>>
62where
63    I: Id,
64    T: Value + Clone + Eq,
65{
66    pub fn delta_reduce<F, U>(&self, f: F) -> Stream<I, U>
67    where
68        F: SelectFn<I, dyn Collection<I, T>, Option<U>>,
69        U: Value,
70    {
71        self.workflow.add_operator(
72            [self.id],
73            DeltaReduce {
74                function: f,
75                store: HashMap::default(),
76                marker: PhantomData,
77            },
78        )
79    }
80}
81
82// ----------------------------------------------------------------------------
83// Trait implementations
84// ----------------------------------------------------------------------------
85
86impl<I, T, F, U> Operator<I, Delta<I, T>> for DeltaReduce<I, T, F, U>
87where
88    I: Id,
89    T: Value + Clone + Eq,
90    F: SelectFn<I, dyn Collection<I, T>, Option<U>>,
91    U: Value,
92{
93    type Item<'a> = Item<&'a I, &'a Delta<I, T>>;
94
95    /// Handles the given item.
96    ///
97    /// This operator keeps track of the current state of deltas of items that
98    /// are associated with each identifier. When a new delta is received, it
99    /// updates the internal store accordingly, applying the insertions and
100    /// deletions, and passes the updated store to the reduction function.
101    #[cfg_attr(
102        feature = "tracing",
103        tracing::instrument(level = "debug", skip_all, fields(id = %item.id))
104    )]
105    fn handle(&mut self, item: Self::Item<'_>) -> impl IntoOutputs<I> {
106        let store = self.store.get_or_insert_default(item.id);
107
108        // Update internal store (chunk) associated with the item's identifier,
109        // applying the insertions and deletions as part of the delta of items
110        // to it. This allows us to keep track of the current state of items
111        // associated with the identifier.
112        for part in item.data {
113            if let Some(data) = &part.data {
114                store.insert(part.id.clone(), data.clone());
115            } else {
116                store.remove(&part.id);
117            }
118        }
119
120        // Since we assume that delta of items are differential by design, we
121        // do not need to check if something has actually changed
122        self.function.execute(item.id, store).map(|report| {
123            report.map(|data| Some(Item::new(item.id.clone(), data)))
124        })
125    }
126
127    /// Returns the descriptor.
128    #[inline]
129    fn descriptor(&self) -> Descriptor {
130        Descriptor::builder() // fmt
131            .property(Property::Stable)
132            .property(Property::Flush)
133            .build()
134    }
135}
zrx_stream/stream/operator/delta_reduce.rs

zrx_stream/stream/operator/
delta_reduce.rs
