krabka_client_streams/dsl/
windows.rs

1//! Time windows, the `Windowed<K>` output key, and a windowed output serde.
2use bytes::{BufMut, Bytes, BytesMut};
3use krabka_units::prelude::*;
4
5use crate::processor::serde::{Serde, SerdeAssociate, SerdeError};
6
7/// A time window in epoch milliseconds.
8///
9/// Time windows ([`TimeWindows`]) are half-open `[start, end)`. Session windows
10/// ([`SessionWindows`]) are inclusive `[start, end]`, and both bounds are
11/// observed record timestamps. The producing operator carries the
12/// interpretation. This struct does not encode it.
13#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
14pub struct Window {
15    pub start: i64,
16    pub end: i64,
17}
18
19/// An aggregation key tagged with its window. This is the output key of a
20/// windowed aggregation (`KTable<Windowed<K>, V>`).
21#[derive(Debug, Clone, PartialEq, Eq, Hash)]
22pub struct Windowed<K> {
23    pub key: K,
24    pub window: Window,
25}
26
27/// Tumbling and hopping time windows, which are epoch-aligned.
28///
29/// `advance == size` is tumbling, and `advance < size` is hopping. `grace` adds
30/// to the changelog retention and to the
31/// [`Suppressed::until_window_closes`] timing when the resulting table is
32/// suppressed.
33///
34/// The window bounds these windows produce are epoch-millisecond *instants* and
35/// stay `i64`. The size, the hop, and the grace are *extents* and are [`Time`]
36/// quantities.
37///
38/// [`Suppressed::until_window_closes`]: crate::dsl::Suppressed::until_window_closes
39#[derive(Debug, Clone, Copy)]
40pub struct TimeWindows {
41    pub size: Time,
42    pub advance: Time,
43    pub grace: Time,
44}
45
46/// The finest window a millisecond timeline can express.
47///
48/// `windows_for` works in epoch milliseconds and divides by the hop, so a
49/// sub-millisecond size or advance would round to zero and cause a division by
50/// zero. A [`Time`] quantity can express such a value, so the constructors must
51/// reject it.
52const MIN_RESOLUTION: Time = millis(1);
53
54impl TimeWindows {
55    /// Tumbling window of `size`, with `advance == size` and grace 0.
56    #[must_use]
57    /// # Panics
58    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
59    pub fn of_size(size: Time) -> Self {
60        assert!(size >= MIN_RESOLUTION, "window size must be >= 1ms");
61        Self {
62            size,
63            advance: size,
64            grace: Time::ZERO,
65        }
66    }
67    /// Hopping window that advances by `advance`, where
68    /// `0 < advance <= size`.
69    #[must_use]
70    /// # Panics
71    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
72    pub fn advance_by(mut self, advance: Time) -> Self {
73        assert!(
74            advance >= MIN_RESOLUTION && advance <= self.size,
75            "1ms <= advance <= size"
76        );
77        self.advance = advance;
78        self
79    }
80    /// Set the grace period used to accept out-of-order records and size
81    /// changelog retention.
82    #[must_use]
83    /// # Panics
84    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
85    pub fn grace(mut self, grace: Time) -> Self {
86        assert!(grace >= Time::ZERO, "grace must be >= 0");
87        self.grace = grace;
88        self
89    }
90    /// The window starts that a timestamp `t` falls into, as in the JVM
91    /// `TimeWindows.windowsFor`.
92    ///
93    /// `t` and the returned starts are epoch-millisecond instants. The size and
94    /// the hop cross into that coordinate space here.
95    #[must_use]
96    pub fn windows_for(&self, t: i64) -> Vec<i64> {
97        let size_ms = self.size.millis_i64();
98        let advance_ms = self.advance.millis_i64();
99        let mut start = (std::cmp::max(0, t - size_ms + advance_ms) / advance_ms) * advance_ms;
100        let mut out = Vec::new();
101        while start <= t {
102            out.push(start);
103            start += advance_ms;
104        }
105        out
106    }
107}
108
109/// A symmetric or asymmetric join window. A record at `t` matches the other
110/// side's records with a timestamp in `[t - before, t + after]`.
111///
112/// `JoinWindows::of` is symmetric, so `before == after`. `.before` and `.after`
113/// make the window asymmetric.
114///
115/// `before` and `after` are extents measured from the joining record, not
116/// absolute bounds, so both are [`Time`].
117#[derive(Debug, Clone, Copy)]
118pub struct JoinWindows {
119    pub before: Time,
120    pub after: Time,
121    pub grace: Time,
122}
123
124impl JoinWindows {
125    /// Symmetric window of `time_difference` before and after, with grace 0.
126    #[must_use]
127    /// # Panics
128    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
129    pub fn of(time_difference: Time) -> Self {
130        assert!(
131            time_difference >= Time::ZERO,
132            "time difference must be >= 0"
133        );
134        Self {
135            before: time_difference,
136            after: time_difference,
137            grace: Time::ZERO,
138        }
139    }
140    #[must_use]
141    /// # Panics
142    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
143    pub fn before(mut self, before: Time) -> Self {
144        assert!(before >= Time::ZERO, "before must be >= 0");
145        self.before = before;
146        self
147    }
148    #[must_use]
149    /// # Panics
150    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
151    pub fn after(mut self, after: Time) -> Self {
152        assert!(after >= Time::ZERO, "after must be >= 0");
153        self.after = after;
154        self
155    }
156    #[must_use]
157    /// # Panics
158    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
159    pub fn grace(mut self, grace: Time) -> Self {
160        assert!(grace >= Time::ZERO, "grace must be >= 0");
161        self.grace = grace;
162        self
163    }
164    /// Window size, which is `before + after`. It is the store retention
165    /// basis.
166    #[must_use]
167    pub fn size(&self) -> Time {
168        self.before + self.after
169    }
170}
171
172/// Session windows. The records for a key form one session while they stay
173/// within `gap` of each other. `gap` is the inactivity gap.
174///
175/// A session window `[start, end]` is defined by the data and is not
176/// epoch-aligned. `grace` adds to the changelog retention and to the
177/// [`Suppressed::until_window_closes`] timing when the resulting table is
178/// suppressed.
179///
180/// [`Suppressed::until_window_closes`]: crate::dsl::Suppressed::until_window_closes
181#[derive(Debug, Clone, Copy)]
182pub struct SessionWindows {
183    pub gap: Time,
184    pub grace: Time,
185}
186
187impl SessionWindows {
188    /// Inactivity gap of `gap`, with grace 0. `gap` must be greater than
189    /// zero.
190    #[must_use]
191    /// # Panics
192    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
193    pub fn of_inactivity_gap(gap: Time) -> Self {
194        assert!(gap > Time::ZERO, "session gap must be > 0");
195        Self {
196            gap,
197            grace: Time::ZERO,
198        }
199    }
200    /// Set the grace period used to accept out-of-order records and size
201    /// changelog retention.
202    #[must_use]
203    /// # Panics
204    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
205    pub fn grace(mut self, grace: Time) -> Self {
206        assert!(grace >= Time::ZERO, "grace must be >= 0");
207        self.grace = grace;
208        self
209    }
210}
211
212/// Sliding windows (KIP-450).
213///
214/// A record at time `t` belongs to every window of the fixed size
215/// `time_difference` (`W`) that contains it, that is to the windows
216/// `[ws, ws + W]` with `ws ∈ [t - W, t]`. Windows are **inclusive on both ends**
217/// and **data-defined**, not epoch-aligned, so there is no `windows_for`. A scan
218/// of the window store finds the affected windows. `grace` allows out-of-order
219/// records up to `W + grace` behind stream time, and it feeds the changelog
220/// retention.
221#[derive(Debug, Clone, Copy)]
222pub struct SlidingWindows {
223    /// Window size `W`. The window is `[start, start + time_difference]`,
224    /// inclusive on both ends.
225    pub time_difference: Time,
226    pub grace: Time,
227}
228
229impl SlidingWindows {
230    /// Time difference of `time_difference` with no grace.
231    #[must_use]
232    /// # Panics
233    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
234    pub fn of_time_difference_with_no_grace(time_difference: Time) -> Self {
235        assert!(
236            time_difference >= Time::ZERO,
237            "time difference must be >= 0"
238        );
239        Self {
240            time_difference,
241            grace: Time::ZERO,
242        }
243    }
244    /// Time difference and grace period.
245    #[must_use]
246    /// # Panics
247    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
248    pub fn of_time_difference_and_grace(time_difference: Time, grace: Time) -> Self {
249        assert!(
250            time_difference >= Time::ZERO,
251            "time difference must be >= 0"
252        );
253        assert!(grace >= Time::ZERO, "grace must be >= 0");
254        Self {
255            time_difference,
256            grace,
257        }
258    }
259}
260
261/// `Serde<Windowed<K>>` that produces the JVM session **output-topic** format
262/// `inner_key_bytes ‖ end:8B BE ‖ start:8B BE`.
263///
264/// The bytes hold both bounds. This differs from `TimeWindowedSerde`, which
265/// encodes only the start and derives `end`.
266#[derive(Debug, Clone, Copy)]
267pub struct SessionWindowedSerde<KS> {
268    inner: KS,
269}
270
271impl<KS> SessionWindowedSerde<KS> {
272    #[must_use]
273    pub fn new(inner: KS) -> Self {
274        Self { inner }
275    }
276}
277
278impl<K, KS> Serde<Windowed<K>> for SessionWindowedSerde<KS>
279where
280    K: Send + Sync + 'static,
281    KS: Serde<K>,
282{
283    fn serialize(&self, topic: &str, value: &Windowed<K>) -> Bytes {
284        let kb = self.inner.serialize(topic, &value.key);
285        let mut b = BytesMut::with_capacity(kb.len() + 16);
286        b.extend_from_slice(&kb);
287        b.put_i64(value.window.end);
288        b.put_i64(value.window.start);
289        b.freeze()
290    }
291    fn deserialize(&self, topic: &str, bytes: &[u8]) -> Result<Windowed<K>, SerdeError> {
292        if bytes.len() < 16 {
293            return Err(SerdeError(format!(
294                "session key too short: {}",
295                bytes.len()
296            )));
297        }
298        let split = bytes.len() - 16;
299        let key = self.inner.deserialize(topic, &bytes[..split])?;
300        let end = i64::from_be_bytes(bytes[split..split + 8].try_into().expect("8 bytes"));
301        let start = i64::from_be_bytes(bytes[split + 8..].try_into().expect("8 bytes"));
302        Ok(Windowed {
303            key,
304            window: Window { start, end },
305        })
306    }
307}
308impl<KS: SerdeAssociate> SerdeAssociate for SessionWindowedSerde<KS> {
309    type Target = Windowed<KS::Target>;
310}
311
312/// `Serde<Windowed<K>>` that produces the JVM **output-topic** format
313/// `inner_key_bytes ‖ windowStart : 8-byte BE`, with no end and no seqnum.
314///
315/// The serde carries the window `size`, so `deserialize` can rebuild
316/// `end = start + size`.
317#[derive(Debug, Clone, Copy)]
318pub struct TimeWindowedSerde<KS> {
319    inner: KS,
320    size: Time,
321}
322
323impl<KS> TimeWindowedSerde<KS> {
324    #[must_use]
325    pub fn new(inner: KS, size: Time) -> Self {
326        Self { inner, size }
327    }
328}
329
330impl<K, KS> Serde<Windowed<K>> for TimeWindowedSerde<KS>
331where
332    K: Send + Sync + 'static,
333    KS: Serde<K>,
334{
335    fn serialize(&self, topic: &str, value: &Windowed<K>) -> Bytes {
336        let kb = self.inner.serialize(topic, &value.key);
337        let mut b = BytesMut::with_capacity(kb.len() + 8);
338        b.extend_from_slice(&kb);
339        b.put_i64(value.window.start);
340        b.freeze()
341    }
342    fn deserialize(&self, topic: &str, bytes: &[u8]) -> Result<Windowed<K>, SerdeError> {
343        if bytes.len() < 8 {
344            return Err(SerdeError(format!(
345                "windowed key too short: {}",
346                bytes.len()
347            )));
348        }
349        let split = bytes.len() - 8;
350        let key = self.inner.deserialize(topic, &bytes[..split])?;
351        let start = i64::from_be_bytes(bytes[split..].try_into().expect("8 bytes"));
352        Ok(Windowed {
353            key,
354            window: Window {
355                start,
356                end: start + self.size.millis_i64(),
357            },
358        })
359    }
360}
361
362impl<KS: SerdeAssociate> SerdeAssociate for TimeWindowedSerde<KS> {
363    type Target = Windowed<KS::Target>;
364}
365
366#[cfg(test)]
367mod tests {
368    use assert2::check;
369
370    use super::*;
371
372    #[test]
373    fn sub_millisecond_windows_are_rejected() {
374        // `windows_for` divides by the hop in whole milliseconds, so anything
375        // finer than a millisecond would divide by zero rather than window.
376        check!(std::panic::catch_unwind(|| TimeWindows::of_size(micros(100))).is_err());
377        check!(
378            std::panic::catch_unwind(|| TimeWindows::of_size(millis(10)).advance_by(micros(500)))
379                .is_err()
380        );
381        // One millisecond is the finest the timeline can express, and works.
382        let finest = TimeWindows::of_size(millis(1));
383        check!(finest.windows_for(0) == vec![0]);
384    }
385
386    #[test]
387    fn windows_for_tumbling_one_window() {
388        let w = TimeWindows::of_size(millis(10));
389        check!(w.windows_for(0) == vec![0]);
390        check!(w.windows_for(9) == vec![0]);
391        check!(w.windows_for(10) == vec![10]);
392        check!(w.windows_for(25) == vec![20]);
393    }
394
395    #[test]
396    fn windows_for_hopping_overlaps() {
397        let w = TimeWindows::of_size(millis(10)).advance_by(millis(5));
398        check!(w.windows_for(12) == vec![5, 10]); // start0 = max(0,12-10+5)/5*5 = 5
399        check!(w.windows_for(0) == vec![0]);
400    }
401
402    #[test]
403    fn join_windows_before_after_size() {
404        let w = JoinWindows::of(millis(10));
405        check!((w.before, w.after, w.grace) == (millis(10), millis(10), Time::ZERO));
406        check!(w.size() == millis(20));
407        let a = JoinWindows::of(millis(10))
408            .before(millis(3))
409            .after(millis(7))
410            .grace(millis(5));
411        check!((a.before, a.after, a.grace) == (millis(3), millis(7), millis(5)));
412        check!(a.size() == millis(10));
413    }
414
415    #[test]
416    fn time_windowed_serde_round_trips_output_format() {
417        use crate::processor::serde::{Serde, StringSerde};
418        let s = TimeWindowedSerde::new(StringSerde, millis(10));
419        let wk = Windowed {
420            key: "k".to_string(),
421            window: Window { start: 20, end: 30 },
422        };
423        let b = s.serialize("t", &wk);
424        check!(b.len() == 9); // "k"(1) ‖ 20i64 BE(8)
425        check!(b[1..9] == 20i64.to_be_bytes());
426        let back = s.deserialize("t", &b).unwrap();
427        check!(back.key == "k");
428        check!(back.window == Window { start: 20, end: 30 }); // end = start + size
429    }
430
431    #[test]
432    fn session_windows_gap_and_grace() {
433        let w = SessionWindows::of_inactivity_gap(secs(60));
434        check!((w.gap, w.grace) == (secs(60), Time::ZERO));
435        let g = SessionWindows::of_inactivity_gap(secs(60)).grace(millis(5));
436        check!((g.gap, g.grace) == (secs(60), millis(5)));
437    }
438
439    #[test]
440    fn session_windowed_serde_round_trips_end_then_start() {
441        use crate::processor::serde::{Serde, StringSerde};
442        let s = SessionWindowedSerde::new(StringSerde);
443        let wk = Windowed {
444            key: "k".to_string(),
445            window: Window { start: 5, end: 9 },
446        };
447        let b = s.serialize("t", &wk);
448        check!(b.len() == 17); // "k"(1) ‖ end:8 ‖ start:8
449        check!(b[1..9] == 9i64.to_be_bytes()); // end first
450        check!(b[9..17] == 5i64.to_be_bytes()); // start second
451        let back = s.deserialize("t", &b).unwrap();
452        check!(back.key == "k");
453        check!(back.window == Window { start: 5, end: 9 });
454    }
455
456    #[test]
457    fn sliding_windows_constructors() {
458        let w = SlidingWindows::of_time_difference_with_no_grace(millis(100));
459        check!((w.time_difference, w.grace) == (millis(100), Time::ZERO));
460        let g = SlidingWindows::of_time_difference_and_grace(millis(100), millis(50));
461        check!((g.time_difference, g.grace) == (millis(100), millis(50)));
462    }
463
464    #[test]
465    #[should_panic(expected = "time difference must be >= 0")]
466    fn sliding_windows_rejects_negative_difference() {
467        let _ = SlidingWindows::of_time_difference_with_no_grace(Time::from_millis(-1));
468    }
469}
krabka_client_streams/dsl/windows.rs

krabka_client_streams/dsl/
windows.rs
