zrx_stream/stream/operator/
delta_reduce.rs

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

zrx_stream/stream/operator/
delta_reduce.rs
