zrx_stream/stream/operator/
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//! 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, Report};
34use zrx_scheduler::effect::Item;
35use zrx_scheduler::{Id, Value};
36use zrx_store::StoreMut;
37
38use crate::stream::function::SelectFn;
39use crate::stream::value::Collection;
40use crate::stream::Stream;
41
42use super::Operator;
43
44// ----------------------------------------------------------------------------
45// Structs
46// ----------------------------------------------------------------------------
47
48/// Reduce operator.
49struct Reduce<I, T, F, U> {
50    /// Identifier.
51    id: I,
52    /// Operator function.
53    function: F,
54    /// Store of items.
55    store: HashMap<I, T>,
56    /// Capture types.
57    marker: PhantomData<U>,
58}
59
60// ----------------------------------------------------------------------------
61// Implementations
62// ----------------------------------------------------------------------------
63
64impl<I, T> Stream<I, T>
65where
66    I: Id,
67    T: Value + Clone + Eq,
68{
69    pub fn reduce<F, U>(&self, id: I, f: F) -> Stream<I, U>
70    where
71        F: SelectFn<I, dyn Collection<I, T>, Option<U>>,
72        U: Value,
73    {
74        self.workflow.add_operator(
75            [self.id],
76            Reduce {
77                id,
78                function: f,
79                store: HashMap::default(),
80                marker: PhantomData,
81            },
82        )
83    }
84}
85
86// ----------------------------------------------------------------------------
87// Trait implementations
88// ----------------------------------------------------------------------------
89
90impl<I, T, F, U> Operator<I, T> for Reduce<I, T, F, U>
91where
92    I: Id,
93    T: Value + Clone + Eq,
94    F: SelectFn<I, dyn Collection<I, T>, Option<U>>,
95    U: Value,
96{
97    type Item<'a> = Item<&'a I, Option<&'a T>>;
98
99    /// Handles the given item.
100    ///
101    /// Reductions should be used only sparingly, as they require to store all
102    /// items that are flowing through the stream in the operator, because the
103    /// reduction is computed on the entire store. This makes sure that the
104    /// differential semantics of the stream are preserved.
105    ///
106    /// If we'd provide an operator for differential reductions (also known as
107    /// scanning), the user would be responsible for ensuring the differential
108    /// invariant, which might lead to subtle, hard to detect bugs. There are
109    /// several other operators that provide case-by-case scan-like semantics,
110    /// which are almost always a better choice than using a reduction, as they
111    /// are much more efficient and easier to reason about. When this operator
112    /// is used incorrectly, it might lead to unbounded memory consumption,
113    /// so use it with care.
114    #[cfg_attr(
115        feature = "tracing",
116        tracing::instrument(level = "debug", skip_all, fields(id = %item.id))
117    )]
118    fn handle(&mut self, item: Self::Item<'_>) -> impl IntoOutputs<I> {
119        let has_changed = if let Some(data) = item.data {
120            self.store.insert_if_changed(item.id, data)
121        } else {
122            self.store.remove(item.id).is_some()
123        };
124
125        // If the store has changed, we pass it to the operator function in
126        // order to compute a new output value. The operator function returns
127        // an option to indicate the presence or abscence of a value for the
128        // identifier. If nothing has changed, nothing is emitted.
129        if has_changed {
130            self.function.execute(&self.id, &self.store).map(|report| {
131                report.map(|data| Some(Item::new(self.id.clone(), data)))
132            })
133        } else {
134            Ok(Report::new(None))
135        }
136    }
137
138    /// Returns the descriptor.
139    #[inline]
140    fn descriptor(&self) -> Descriptor {
141        Descriptor::builder() // fmt
142            .property(Property::Flush)
143            .build()
144    }
145}
zrx_stream/stream/operator/reduce.rs

zrx_stream/stream/operator/
reduce.rs
