hyphae/traits/collections/
map_values.rs

1//! Key-preserving projection plan nodes.
2
3use std::{hash::Hash, marker::PhantomData};
4
5use crate::{
6    map_query::{
7        BuildQueryRuntime, MapQuery,
8        properties::{ExactlyOne, PlanProperties, ZeroOrOne},
9    },
10    subscription::SubscriptionGuard,
11    traits::{
12        CellValue,
13        collections::internal::stateless_runtime::{
14            install_filter_map_values_runtime, install_map_values_runtime,
15        },
16    },
17};
18
19impl<S, K, V, U, F> PlanProperties for MapValuesPlan<S, K, V, U, F>
20where
21    S: MapQuery<Key = K, Value = V> + PlanProperties,
22    K: Hash + Eq + CellValue,
23    V: CellValue,
24    U: CellValue,
25    F: Fn(&K, &V) -> U + Send + Sync + 'static,
26{
27    type Cardinality = ExactlyOne;
28    type InputPartition = S::OutputPartition;
29    type OutputPartition = S::OutputPartition;
30}
31
32impl<S, K, V, U, F> PlanProperties for FilterMapValuesPlan<S, K, V, U, F>
33where
34    S: MapQuery<Key = K, Value = V> + PlanProperties,
35    K: Hash + Eq + CellValue,
36    V: CellValue,
37    U: CellValue,
38    F: Fn(&K, &V) -> Option<U> + Send + Sync + 'static,
39{
40    type Cardinality = ZeroOrOne;
41    type InputPartition = S::OutputPartition;
42    type OutputPartition = S::OutputPartition;
43}
44
45/// Exactly-one, key-preserving value projection.
46pub struct MapValuesPlan<S, K, V, U, F>
47where
48    S: MapQuery<Key = K, Value = V>,
49    K: Hash + Eq + CellValue,
50    V: CellValue,
51    U: CellValue,
52    F: Fn(&K, &V) -> U + Send + Sync + 'static,
53{
54    pub(crate) source: S,
55    pub(crate) f: F,
56    _types: PhantomData<fn() -> (K, V, U)>,
57}
58
59impl<S, K, V, U, F> BuildQueryRuntime<K, U> for MapValuesPlan<S, K, V, U, F>
60where
61    S: MapQuery<Key = K, Value = V>,
62    K: Hash + Eq + CellValue,
63    V: CellValue,
64    U: CellValue,
65    F: Fn(&K, &V) -> U + Send + Sync + 'static,
66{
67    fn build_into(
68        self,
69        cx: &mut crate::map_query::compiler::CompileContext,
70        sink: crate::map_query::BoxedMapDiffSink<K, U>,
71    ) -> Vec<SubscriptionGuard> {
72        install_map_values_runtime(cx, self.source, self.f, sink)
73    }
74}
75
76#[allow(private_bounds)]
77impl<S, K, V, U, F> MapQuery for MapValuesPlan<S, K, V, U, F>
78where
79    S: MapQuery<Key = K, Value = V>,
80    K: Hash + Eq + CellValue,
81    V: CellValue,
82    U: CellValue,
83    F: Fn(&K, &V) -> U + Send + Sync + 'static,
84{
85    type Key = K;
86    type Value = U;
87}
88
89impl<S, K, V, U, F> MapValuesPlan<S, K, V, U, F>
90where
91    S: MapQuery<Key = K, Value = V>,
92    K: Hash + Eq + CellValue,
93    V: CellValue,
94    U: CellValue,
95    F: Fn(&K, &V) -> U + Send + Sync + 'static,
96{
97    /// Fuse another exactly-one projection into this kernel.
98    pub fn map_values<W, G>(
99        self,
100        g: G,
101    ) -> MapValuesPlan<S, K, V, W, impl Fn(&K, &V) -> W + Send + Sync + 'static>
102    where
103        W: CellValue,
104        G: Fn(&K, &U) -> W + Send + Sync + 'static,
105    {
106        let f = self.f;
107        MapValuesPlan {
108            source: self.source,
109            f: move |key, value| {
110                let value = f(key, value);
111                g(key, &value)
112            },
113            _types: PhantomData,
114        }
115    }
116
117    /// Fuse a filtering projection into this kernel.
118    pub fn filter_map_values<W, G>(
119        self,
120        g: G,
121    ) -> FilterMapValuesPlan<S, K, V, W, impl Fn(&K, &V) -> Option<W> + Send + Sync + 'static>
122    where
123        W: CellValue,
124        G: Fn(&K, &U) -> Option<W> + Send + Sync + 'static,
125    {
126        let f = self.f;
127        FilterMapValuesPlan {
128            source: self.source,
129            f: move |key, value| {
130                let value = f(key, value);
131                g(key, &value)
132            },
133            _types: PhantomData,
134        }
135    }
136
137    /// Fuse a value predicate after this projection.
138    pub fn select<G>(
139        self,
140        predicate: G,
141    ) -> FilterMapValuesPlan<S, K, V, U, impl Fn(&K, &V) -> Option<U> + Send + Sync + 'static>
142    where
143        G: Fn(&U) -> bool + Send + Sync + 'static,
144    {
145        let f = self.f;
146        FilterMapValuesPlan {
147            source: self.source,
148            f: move |key, value| {
149                let value = f(key, value);
150                predicate(&value).then_some(value)
151            },
152            _types: PhantomData,
153        }
154    }
155
156    /// Fuse a key-aware predicate after this projection.
157    pub fn select_by<G>(
158        self,
159        predicate: G,
160    ) -> FilterMapValuesPlan<S, K, V, U, impl Fn(&K, &V) -> Option<U> + Send + Sync + 'static>
161    where
162        G: Fn(&K, &U) -> bool + Send + Sync + 'static,
163    {
164        let f = self.f;
165        FilterMapValuesPlan {
166            source: self.source,
167            f: move |key, value| {
168                let value = f(key, value);
169                predicate(key, &value).then_some(value)
170            },
171            _types: PhantomData,
172        }
173    }
174}
175
176/// Zero-or-one, key-preserving value projection.
177pub struct FilterMapValuesPlan<S, K, V, U, F>
178where
179    S: MapQuery<Key = K, Value = V>,
180    K: Hash + Eq + CellValue,
181    V: CellValue,
182    U: CellValue,
183    F: Fn(&K, &V) -> Option<U> + Send + Sync + 'static,
184{
185    pub(crate) source: S,
186    pub(crate) f: F,
187    pub(crate) _types: PhantomData<fn() -> (K, V, U)>,
188}
189
190impl<S, K, V, U, F> FilterMapValuesPlan<S, K, V, U, F>
191where
192    S: MapQuery<Key = K, Value = V>,
193    K: Hash + Eq + CellValue,
194    V: CellValue,
195    U: CellValue,
196    F: Fn(&K, &V) -> Option<U> + Send + Sync + 'static,
197{
198    /// Fuse an exactly-one projection after this optional kernel.
199    pub fn map_values<W, G>(
200        self,
201        g: G,
202    ) -> FilterMapValuesPlan<S, K, V, W, impl Fn(&K, &V) -> Option<W> + Send + Sync + 'static>
203    where
204        W: CellValue,
205        G: Fn(&K, &U) -> W + Send + Sync + 'static,
206    {
207        let f = self.f;
208        FilterMapValuesPlan {
209            source: self.source,
210            f: move |key, value| f(key, value).map(|value| g(key, &value)),
211            _types: PhantomData,
212        }
213    }
214
215    /// Fuse another filtering projection into this optional kernel.
216    pub fn filter_map_values<W, G>(
217        self,
218        g: G,
219    ) -> FilterMapValuesPlan<S, K, V, W, impl Fn(&K, &V) -> Option<W> + Send + Sync + 'static>
220    where
221        W: CellValue,
222        G: Fn(&K, &U) -> Option<W> + Send + Sync + 'static,
223    {
224        let f = self.f;
225        FilterMapValuesPlan {
226            source: self.source,
227            f: move |key, value| f(key, value).and_then(|value| g(key, &value)),
228            _types: PhantomData,
229        }
230    }
231
232    /// Fuse a value predicate into this optional kernel.
233    pub fn select<G>(
234        self,
235        predicate: G,
236    ) -> FilterMapValuesPlan<S, K, V, U, impl Fn(&K, &V) -> Option<U> + Send + Sync + 'static>
237    where
238        G: Fn(&U) -> bool + Send + Sync + 'static,
239    {
240        let f = self.f;
241        FilterMapValuesPlan {
242            source: self.source,
243            f: move |key, value| f(key, value).filter(|value| predicate(value)),
244            _types: PhantomData,
245        }
246    }
247
248    /// Fuse a key-aware predicate into this optional kernel.
249    pub fn select_by<G>(
250        self,
251        predicate: G,
252    ) -> FilterMapValuesPlan<S, K, V, U, impl Fn(&K, &V) -> Option<U> + Send + Sync + 'static>
253    where
254        G: Fn(&K, &U) -> bool + Send + Sync + 'static,
255    {
256        let f = self.f;
257        FilterMapValuesPlan {
258            source: self.source,
259            f: move |key, value| f(key, value).filter(|value| predicate(key, value)),
260            _types: PhantomData,
261        }
262    }
263}
264
265impl<S, K, V, U, F> BuildQueryRuntime<K, U> for FilterMapValuesPlan<S, K, V, U, F>
266where
267    S: MapQuery<Key = K, Value = V>,
268    K: Hash + Eq + CellValue,
269    V: CellValue,
270    U: CellValue,
271    F: Fn(&K, &V) -> Option<U> + Send + Sync + 'static,
272{
273    fn build_into(
274        self,
275        cx: &mut crate::map_query::compiler::CompileContext,
276        sink: crate::map_query::BoxedMapDiffSink<K, U>,
277    ) -> Vec<SubscriptionGuard> {
278        install_filter_map_values_runtime(cx, self.source, self.f, sink)
279    }
280}
281
282#[allow(private_bounds)]
283impl<S, K, V, U, F> MapQuery for FilterMapValuesPlan<S, K, V, U, F>
284where
285    S: MapQuery<Key = K, Value = V>,
286    K: Hash + Eq + CellValue,
287    V: CellValue,
288    U: CellValue,
289    F: Fn(&K, &V) -> Option<U> + Send + Sync + 'static,
290{
291    type Key = K;
292    type Value = U;
293}
294
295/// Semantic key-preserving projection operators.
296///
297/// Projection closures must be deterministic, externally side-effect-free,
298/// and nonblocking. They may be invoked repeatedly or concurrently; invocation
299/// count, order, and thread are not API guarantees.
300pub trait MapValuesExt<K, V>: MapQuery<Key = K, Value = V>
301where
302    K: Hash + Eq + CellValue,
303    V: CellValue,
304{
305    /// Map every value while retaining its input key.
306    fn map_values<U, F>(self, f: F) -> MapValuesPlan<Self, K, V, U, F>
307    where
308        U: CellValue,
309        F: Fn(&K, &V) -> U + Send + Sync + 'static,
310    {
311        MapValuesPlan {
312            source: self,
313            f,
314            _types: PhantomData,
315        }
316    }
317
318    /// Map or omit a value while retaining its input key.
319    fn filter_map_values<U, F>(self, f: F) -> FilterMapValuesPlan<Self, K, V, U, F>
320    where
321        U: CellValue,
322        F: Fn(&K, &V) -> Option<U> + Send + Sync + 'static,
323    {
324        FilterMapValuesPlan {
325            source: self,
326            f,
327            _types: PhantomData,
328        }
329    }
330}
331
332impl<K, V, M> MapValuesExt<K, V> for M
333where
334    K: Hash + Eq + CellValue,
335    V: CellValue,
336    M: MapQuery<Key = K, Value = V>,
337{
338}
339
340#[cfg(test)]
341mod tests {
342    use super::*;
343    use crate::CellMap;
344
345    #[test]
346    fn key_preserving_projections_track_membership_and_values() {
347        let source = CellMap::<u64, u64>::new();
348        source.insert(1, 5);
349        source.insert(2, 10);
350
351        let output = source
352            .clone()
353            .filter_map_values(|_key, value| (*value >= 10).then_some(value * 2))
354            .map_values(|_key, value| value + 1)
355            .materialize();
356
357        assert_eq!(output.get_value(&1), None);
358        assert_eq!(output.get_value(&2), Some(21));
359
360        source.insert(1, 12);
361        source.insert(2, 3);
362        assert_eq!(output.get_value(&1), Some(25));
363        assert_eq!(output.get_value(&2), None);
364    }
365}
hyphae/traits/collections/map_values.rs

hyphae/traits/collections/
map_values.rs
