zrx_stream/stream/operator/
lift.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//! Lift operator.
27
28use ahash::HashMap;
29use zrx_scheduler::action::descriptor::Property;
30use zrx_scheduler::action::output::IntoOutputs;
31use zrx_scheduler::action::{Descriptor, Report};
32use zrx_scheduler::effect::Item;
33use zrx_scheduler::{Id, Value};
34use zrx_store::behavior::StoreDelta;
35use zrx_store::StoreMutRef;
36
37use crate::stream::function::LiftFn;
38use crate::stream::value::Delta;
39use crate::stream::Stream;
40
41use super::{Operator, OperatorExt};
42
43// ----------------------------------------------------------------------------
44// Structs
45// ----------------------------------------------------------------------------
46
47/// Lift operator.
48struct Lift<F, I, U> {
49    /// Operator function.
50    function: F,
51    /// Store of change sets.
52    store: HashMap<I, HashMap<I, U>>,
53}
54
55// ----------------------------------------------------------------------------
56// Implementations
57// ----------------------------------------------------------------------------
58
59impl<I, T> Stream<I, T>
60where
61    I: Id,
62    T: Value,
63{
64    pub fn lift<F, U>(&self, f: F) -> Stream<I, Delta<I, U>>
65    where
66        F: LiftFn<I, T, U>,
67        U: Value + Clone + Eq,
68    {
69        self.with_operator(Lift {
70            function: f,
71            store: HashMap::default(),
72        })
73    }
74}
75
76// ----------------------------------------------------------------------------
77// Trait implementations
78// ----------------------------------------------------------------------------
79
80impl<I, T, F, U> Operator<I, T> for Lift<F, I, U>
81where
82    I: Id,
83    T: Value,
84    F: LiftFn<I, T, U>,
85    U: Value + Clone + Eq,
86{
87    type Item<'a> = Item<&'a I, Option<&'a T>>;
88
89    /// Handles the given item.
90    ///
91    /// Lifting is an essential operation in stream processing, as it allows an
92    /// input item to be transformed into multiple related output items with a
93    /// user-defined function. This operator always returns a delta of items,
94    /// ensuring that the related items can be processed differentially.
95    #[cfg_attr(
96        feature = "tracing",
97        tracing::instrument(level = "debug", skip_all, fields(id = %item.id))
98    )]
99    fn handle(&mut self, item: Self::Item<'_>) -> impl IntoOutputs<I> {
100        if let Some(data) = item.data {
101            // When new data arrives, we apply the operator function to generate
102            // related items. We then compute the delta between the previous set
103            // of related items and the returned set, ensuring that only changes
104            // are propagated downstream.
105            self.function.execute(item.id, data).map(|report| {
106                report.map(|data| {
107                    let store = self.store.get_or_insert_default(item.id);
108                    let delta = store
109                        .changes(data.into_iter().map(Item::into_parts))
110                        .map(|(id, data)| Item::new(id, data))
111                        .collect();
112
113                    // Return delta of items
114                    Item::new(item.id.clone(), Some(delta))
115                })
116            })
117        } else {
118            // If the incoming item has no data, interpret this as a deletion,
119            // removing any previously stored related items. By emitting a delta
120            // of items instead of a deletion, we can ensure that all downstream
121            // operators can update their internal state accordingly.
122            let store = self.store.remove(item.id);
123            let delta = store.map_or_else(Delta::default, |inner| {
124                inner.into_keys().map(|id| Item::new(id, None)).collect()
125            });
126
127            // Return delta of items
128            Ok(Report::new(Item::new(item.id.clone(), Some(delta))))
129        }
130    }
131
132    /// Returns the descriptor.
133    #[inline]
134    fn descriptor(&self) -> Descriptor {
135        Descriptor::builder() // fmt
136            .property(Property::Stable)
137            .build()
138    }
139}
zrx_stream/stream/operator/lift.rs

zrx_stream/stream/operator/
lift.rs
