krabka_client_streams/dsl/
sliding_windowed_cogrouped.rs

1//! `SlidingWindowedCogroupedStream<K, VOut>`: sliding-windowed KIP-150 cogroup.
2//!
3//! `CogroupedKStream::windowed_by_sliding(SlidingWindows)` builds it. The
4//! terminal `aggregate_explicit` produces a `KTable<Windowed<K>, VOut>` over a
5//! shared window store. That store uses the KIP-450 `2 * timeDifferenceMs`
6//! retention formula.
7
8use std::{any::Any, cell::RefCell, marker::PhantomData, rc::Rc, sync::Arc};
9
10use crate::{
11    dsl::{
12        builder::InternalStreamsBuilder,
13        cogrouped::{
14            CogroupInput, CogroupKind, CogroupSpec, CogroupedKStream, StoreRegistrarFn,
15            lower_cogroup,
16        },
17        config::Materialized,
18        kgrouped::mint_store_name,
19        ktable::KTable,
20        names,
21        sliding_windowed_kgrouped::sliding_suppress_factory,
22        windows::{SlidingWindows, TimeWindowedSerde, Windowed},
23    },
24    processor::serde::Serde,
25};
26
27impl<K, VOut> CogroupedKStream<K, VOut>
28where
29    K: Any + Send + Sync + Clone,
30    VOut: Any + Send + Sync + Clone,
31{
32    /// `windowedBy(SlidingWindows)`: switch to a sliding-windowed cogroup
33    /// (KIP-450).
34    #[must_use]
35    pub fn windowed_by_sliding(
36        self,
37        windows: SlidingWindows,
38    ) -> SlidingWindowedCogroupedStream<K, VOut> {
39        SlidingWindowedCogroupedStream {
40            builder: self.builder,
41            inputs: self.inputs,
42            windows,
43            _pd: PhantomData,
44        }
45    }
46}
47
48/// Handle produced by [`CogroupedKStream::windowed_by_sliding`]. The terminal
49/// sliding-windowed aggregation consumes it.
50pub struct SlidingWindowedCogroupedStream<K, VOut> {
51    builder: Rc<RefCell<InternalStreamsBuilder>>,
52    inputs: Vec<CogroupInput<K, VOut>>,
53    windows: SlidingWindows,
54    _pd: PhantomData<fn() -> (K, VOut)>,
55}
56
57impl<K, VOut> SlidingWindowedCogroupedStream<K, VOut>
58where
59    K: Any + Send + Sync + Clone,
60    VOut: Any + Send + Sync + Clone,
61{
62    /// Sliding-windowed terminal aggregation. It produces a
63    /// `KTable<Windowed<K>, VOut>`.
64    ///
65    /// Store retention uses the KIP-450 formula: `size = 2 * timeDifferenceMs`
66    /// (matching `sliding_windowed_kgrouped.rs::lower_aggregate`).
67    pub fn aggregate_explicit<KS, VS, I>(
68        self,
69        init: I,
70        materialized: impl Into<Materialized<KS, VS>>,
71    ) -> KTable<Windowed<K>, VOut, TimeWindowedSerde<KS>, VS>
72    where
73        KS: Serde<K> + Clone + 'static,
74        VS: Serde<VOut> + Clone + 'static,
75        I: Fn() -> VOut + Send + Sync + 'static,
76    {
77        let materialized = materialized.into();
78        let store_name = mint_store_name(&self.builder, &materialized, names::AGGREGATE_STORE);
79        let suppress_factory = sliding_suppress_factory::<K, VOut, KS, VS>(
80            materialized.key_serde.clone(),
81            materialized.value_serde.clone(),
82            self.windows,
83        );
84        let Materialized {
85            key_serde,
86            value_serde,
87            logging,
88            caching,
89            ..
90        } = materialized;
91        let spec = CogroupSpec::<K, VOut> {
92            kind: CogroupKind::Sliding(self.windows),
93            init: Arc::new(init),
94            merger: None,
95        };
96        let ks = key_serde.clone();
97        let vs = value_serde.clone();
98        let store_for_reg = store_name.clone();
99        // Sliding window retention formula (JVM-exact):
100        // timeDifferenceMs + timeDifferenceMs + gracePeriodMs + 86_400_000
101        // (a sliding window spans [t - timeDiff, t + timeDiff] so the effective
102        // window size for changelog retention is 2 * timeDifferenceMs).
103        let size = self.windows.time_difference * 2.0;
104        // The true sliding window size for the key end is 1 * timeDiff (the
105        // retention basis above is the 2× span).
106        let window_size = self.windows.time_difference;
107        let grace = self.windows.grace;
108        let registrar: StoreRegistrarFn = Box::new(move |state, procs| {
109            state.topology.add_window_store::<K, VOut, KS, VS>(
110                store_for_reg.clone(),
111                ks.clone(),
112                vs.clone(),
113                (size, window_size, grace),
114                procs,
115            );
116            state.topology.mark_store_caching(&store_for_reg, caching);
117        });
118        let merge_id = lower_cogroup::<K, VOut, Windowed<K>>(
119            &self.builder,
120            self.inputs,
121            &store_name,
122            &spec,
123            logging,
124            registrar,
125        );
126        KTable::new(
127            Rc::clone(&self.builder),
128            merge_id,
129            Some(store_name),
130            None,
131            TimeWindowedSerde::new(key_serde, self.windows.time_difference),
132            value_serde,
133        )
134        .with_window_grace(Some(self.windows.grace))
135        .with_suppress_factory(Some(suppress_factory))
136    }
137}
138
139#[cfg(test)]
140mod caching_tests {
141    use assert2::check;
142    use krabka_units::prelude::*;
143
144    use crate::{
145        BufferConfig, I64Serde, Materialized, Produced, SlidingWindows, StringSerde, Suppressed,
146        dsl::{StreamsBuilder, windows::TimeWindowedSerde},
147        store::backend::StoreBackend,
148    };
149
150    #[test]
151    fn sliding_windowed_cogroup_marks_store_cached() {
152        let b = StreamsBuilder::new();
153        let g1 = b.stream::<String, String>(["in1"]).group_by_key();
154        let g2 = b.stream::<String, String>(["in2"]).group_by_key();
155        g1.cogroup::<i64, _>(|_k, v: &String, acc| {
156            acc + i64::try_from(v.len()).unwrap_or(i64::MAX)
157        })
158        .cogroup(g2, |_k, _v: &String, acc| acc + 1)
159        .windowed_by_sliding(SlidingWindows::of_time_difference_with_no_grace(millis(
160            100,
161        )))
162        .aggregate_explicit(
163            || 0i64,
164            Materialized::with(StringSerde, I64Serde).as_store("cg"),
165        )
166        .suppress(Suppressed::until_window_closes(BufferConfig::unbounded()))
167        .to_stream()
168        .to_explicit(
169            "out",
170            Produced::with(TimeWindowedSerde::new(StringSerde, millis(100)), I64Serde),
171        );
172        let built = b.build("app").unwrap();
173        let g = pollster::block_on(built.instantiate(&StoreBackend::InMemory, "app", kibibytes(1)))
174            .unwrap();
175        check!(g.cache_owner.contains_key("cg"));
176    }
177
178    #[test]
179    fn sliding_windowed_cogroup_uncached_when_off() {
180        let b = StreamsBuilder::new();
181        let g1 = b.stream::<String, String>(["in1"]).group_by_key();
182        let g2 = b.stream::<String, String>(["in2"]).group_by_key();
183        g1.cogroup::<i64, _>(|_k, v: &String, acc| {
184            acc + i64::try_from(v.len()).unwrap_or(i64::MAX)
185        })
186        .cogroup(g2, |_k, _v: &String, acc| acc + 1)
187        .windowed_by_sliding(SlidingWindows::of_time_difference_with_no_grace(millis(
188            100,
189        )))
190        .aggregate_explicit(
191            || 0i64,
192            Materialized::with(StringSerde, I64Serde)
193                .as_store("cg")
194                .with_caching(false),
195        )
196        .to_stream()
197        .to_explicit(
198            "out",
199            Produced::with(TimeWindowedSerde::new(StringSerde, millis(100)), I64Serde),
200        );
201        let built = b.build("app").unwrap();
202        let g = pollster::block_on(built.instantiate(&StoreBackend::InMemory, "app", kibibytes(1)))
203            .unwrap();
204        check!(!g.cache_owner.contains_key("cg"));
205    }
206}
krabka_client_streams/dsl/sliding_windowed_cogrouped.rs

krabka_client_streams/dsl/
sliding_windowed_cogrouped.rs
