laminar_core/operator/window/
mod.rs

1//! Window assignment, emit strategies, and CDC types for stream processing.
2//!
3//! - [`TumblingWindowAssigner`]: Fixed-size, non-overlapping windows
4//! - [`EmitStrategy`]: Controls when window results are emitted
5//! - [`ChangelogRecord`]: CDC records with Z-set weights
6
7use super::Event;
8use smallvec::SmallVec;
9use std::time::Duration;
10
11/// Strategy for when window results should be emitted.
12#[derive(Debug, Clone, PartialEq, Eq, Default)]
13pub enum EmitStrategy {
14    /// Emit when watermark passes window end (default, most efficient).
15    #[default]
16    OnWatermark,
17    /// Emit intermediate results at fixed intervals, plus final on watermark.
18    Periodic(Duration),
19    /// Emit after every state change (lowest latency, highest overhead).
20    OnUpdate,
21    /// Emit only when window closes. Append-only safe, no retractions.
22    /// SQL: `EMIT ON WINDOW CLOSE`
23    OnWindowClose,
24    /// Emit changelog records with Z-set weights for CDC pipelines.
25    /// SQL: `EMIT CHANGES`
26    Changelog,
27    /// Suppress all intermediate results, emit only finalized.
28    /// SQL: `EMIT FINAL`
29    Final,
30}
31
32impl EmitStrategy {
33    /// Returns true if this strategy requires periodic timer registration.
34    #[must_use]
35    pub fn needs_periodic_timer(&self) -> bool {
36        matches!(self, Self::Periodic(_))
37    }
38
39    /// Returns the periodic interval if this is a periodic strategy.
40    #[must_use]
41    pub fn periodic_interval(&self) -> Option<Duration> {
42        match self {
43            Self::Periodic(d) => Some(*d),
44            _ => None,
45        }
46    }
47
48    /// Returns true if results should be emitted on every update.
49    #[must_use]
50    pub fn emits_on_update(&self) -> bool {
51        matches!(self, Self::OnUpdate)
52    }
53
54    /// Returns true if this strategy emits intermediate results before window close.
55    #[must_use]
56    pub fn emits_intermediate(&self) -> bool {
57        matches!(self, Self::OnUpdate | Self::Periodic(_))
58    }
59
60    /// Returns true if this strategy requires changelog/Z-set support.
61    #[must_use]
62    pub fn requires_changelog(&self) -> bool {
63        matches!(self, Self::Changelog)
64    }
65
66    /// Returns true if safe for append-only sinks (no retractions).
67    #[must_use]
68    pub fn is_append_only_compatible(&self) -> bool {
69        matches!(self, Self::OnWindowClose | Self::Final)
70    }
71
72    /// Returns true if late data should generate retractions.
73    #[must_use]
74    pub fn generates_retractions(&self) -> bool {
75        matches!(self, Self::OnWatermark | Self::OnUpdate | Self::Changelog)
76    }
77
78    /// Returns true if this strategy should suppress intermediate emissions.
79    #[must_use]
80    pub fn suppresses_intermediate(&self) -> bool {
81        matches!(self, Self::OnWindowClose | Self::Final)
82    }
83
84    /// Returns true if late data should be dropped entirely.
85    #[must_use]
86    pub fn drops_late_data(&self) -> bool {
87        matches!(self, Self::Final)
88    }
89}
90
91/// Unique identifier for a window (start inclusive, end exclusive, milliseconds).
92#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
93pub struct WindowId {
94    /// Window start timestamp (inclusive, milliseconds).
95    pub start: i64,
96    /// Window end timestamp (exclusive, milliseconds).
97    pub end: i64,
98}
99
100impl WindowId {
101    /// Creates a new window ID.
102    #[must_use]
103    pub fn new(start: i64, end: i64) -> Self {
104        Self { start, end }
105    }
106
107    /// Window duration in milliseconds.
108    #[must_use]
109    pub fn duration_ms(&self) -> i64 {
110        self.end - self.start
111    }
112
113    /// Converts to a 16-byte big-endian key for state storage.
114    #[inline]
115    #[must_use]
116    pub fn to_key(&self) -> super::TimerKey {
117        super::TimerKey::from(self.to_key_inline())
118    }
119
120    /// Stack-allocated 16-byte key (zero-allocation hot path).
121    #[inline]
122    #[must_use]
123    pub fn to_key_inline(&self) -> [u8; 16] {
124        let mut key = [0u8; 16];
125        key[..8].copy_from_slice(&self.start.to_be_bytes());
126        key[8..16].copy_from_slice(&self.end.to_be_bytes());
127        key
128    }
129
130    /// Parses from a 16-byte big-endian key. Returns `None` if wrong length.
131    #[must_use]
132    pub fn from_key(key: &[u8]) -> Option<Self> {
133        if key.len() != 16 {
134            return None;
135        }
136        let start = i64::from_be_bytes(key[0..8].try_into().ok()?);
137        let end = i64::from_be_bytes(key[8..16].try_into().ok()?);
138        Some(Self { start, end })
139    }
140}
141
142/// Window assignment results. Inline storage for up to 4 windows (avoids heap).
143pub type WindowIdVec = SmallVec<[WindowId; 4]>;
144
145/// A window boundary cannot be represented by the timestamp type.
146#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
147#[error("window boundaries [{start}, {end}) for timestamp {timestamp} do not fit in i64")]
148pub struct WindowAssignmentError {
149    timestamp: i64,
150    start: i128,
151    end: i128,
152}
153
154impl WindowAssignmentError {
155    pub(crate) const fn new(timestamp: i64, start: i128, end: i128) -> Self {
156        Self {
157            timestamp,
158            start,
159            end,
160        }
161    }
162}
163
164/// CDC operation type with Z-set weights.
165#[derive(Debug, Clone, Copy, PartialEq, Eq)]
166pub enum CdcOperation {
167    /// +1 weight
168    Insert,
169    /// -1 weight
170    Delete,
171    /// -1 weight (retraction before update)
172    UpdateBefore,
173    /// +1 weight (new value after update)
174    UpdateAfter,
175}
176
177impl CdcOperation {
178    /// Z-set weight: +1 for inserts, -1 for deletes.
179    #[must_use]
180    pub fn weight(&self) -> i32 {
181        match self {
182            Self::Insert | Self::UpdateAfter => 1,
183            Self::Delete | Self::UpdateBefore => -1,
184        }
185    }
186
187    /// Returns true for insert-type operations.
188    #[must_use]
189    pub fn is_insert(&self) -> bool {
190        matches!(self, Self::Insert | Self::UpdateAfter)
191    }
192
193    /// Returns true for delete-type operations.
194    #[must_use]
195    pub fn is_delete(&self) -> bool {
196        matches!(self, Self::Delete | Self::UpdateBefore)
197    }
198
199    /// Compact u8 encoding for storage.
200    #[inline]
201    #[must_use]
202    pub fn to_u8(self) -> u8 {
203        match self {
204            Self::Insert => 0,
205            Self::Delete => 1,
206            Self::UpdateBefore => 2,
207            Self::UpdateAfter => 3,
208        }
209    }
210
211    /// Decode from u8. Returns `None` for out-of-range values.
212    #[inline]
213    #[must_use]
214    pub fn from_u8(value: u8) -> Option<Self> {
215        match value {
216            0 => Some(Self::Insert),
217            1 => Some(Self::Delete),
218            2 => Some(Self::UpdateBefore),
219            3 => Some(Self::UpdateAfter),
220            _ => None,
221        }
222    }
223}
224
225/// Changelog record with Z-set weight for CDC pipelines.
226#[derive(Debug, Clone)]
227pub struct ChangelogRecord {
228    /// The CDC operation type.
229    pub operation: CdcOperation,
230    /// Z-set weight (+1 for insert, -1 for delete).
231    pub weight: i32,
232    /// Timestamp when this change was emitted.
233    pub emit_timestamp: i64,
234    /// The event data.
235    pub event: Event,
236}
237
238impl ChangelogRecord {
239    /// Creates an insert record (+1 weight).
240    #[must_use]
241    pub fn insert(event: Event, emit_timestamp: i64) -> Self {
242        Self {
243            operation: CdcOperation::Insert,
244            weight: 1,
245            emit_timestamp,
246            event,
247        }
248    }
249
250    /// Creates a delete record (-1 weight).
251    #[must_use]
252    pub fn delete(event: Event, emit_timestamp: i64) -> Self {
253        Self {
254            operation: CdcOperation::Delete,
255            weight: -1,
256            emit_timestamp,
257            event,
258        }
259    }
260
261    /// Creates an update retraction pair (`UpdateBefore`, `UpdateAfter`).
262    #[must_use]
263    pub fn update(old_event: Event, new_event: Event, emit_timestamp: i64) -> (Self, Self) {
264        let before = Self {
265            operation: CdcOperation::UpdateBefore,
266            weight: -1,
267            emit_timestamp,
268            event: old_event,
269        };
270        let after = Self {
271            operation: CdcOperation::UpdateAfter,
272            weight: 1,
273            emit_timestamp,
274            event: new_event,
275        };
276        (before, after)
277    }
278
279    /// Creates a record from raw parts.
280    #[must_use]
281    pub fn new(operation: CdcOperation, event: Event, emit_timestamp: i64) -> Self {
282        Self {
283            operation,
284            weight: operation.weight(),
285            emit_timestamp,
286            event,
287        }
288    }
289
290    /// Returns true for insert-type records.
291    #[must_use]
292    pub fn is_insert(&self) -> bool {
293        self.operation.is_insert()
294    }
295
296    /// Returns true for delete-type records.
297    #[must_use]
298    pub fn is_delete(&self) -> bool {
299        self.operation.is_delete()
300    }
301}
302
303/// Trait for assigning events to windows.
304pub trait WindowAssigner: Send {
305    /// Assigns a timestamp to one or more windows.
306    fn assign_windows(&self, timestamp: i64) -> WindowIdVec;
307
308    /// Maximum timestamp assignable to a window ending at `window_end`.
309    fn max_timestamp(&self, window_end: i64) -> i64 {
310        window_end - 1
311    }
312}
313
314/// Tumbling window assigner: fixed-size, non-overlapping windows aligned to epoch.
315#[derive(Debug, Clone)]
316pub struct TumblingWindowAssigner {
317    size_ms: i64,
318    offset_ms: i64,
319}
320
321impl TumblingWindowAssigner {
322    /// # Panics
323    /// Panics if size is zero.
324    #[must_use]
325    pub fn new(size: Duration) -> Self {
326        let size_ms = i64::try_from(size.as_millis()).expect("Window size must fit in i64");
327        assert!(size_ms > 0, "Window size must be positive");
328        Self {
329            size_ms,
330            offset_ms: 0,
331        }
332    }
333
334    /// # Panics
335    /// Panics if `size_ms` is zero or negative.
336    #[must_use]
337    pub fn from_millis(size_ms: i64) -> Self {
338        assert!(size_ms > 0, "Window size must be positive");
339        Self {
340            size_ms,
341            offset_ms: 0,
342        }
343    }
344
345    /// Sets window offset in milliseconds for timezone-aligned windows.
346    #[must_use]
347    pub fn with_offset_ms(mut self, offset_ms: i64) -> Self {
348        self.offset_ms = offset_ms;
349        self
350    }
351
352    /// Window size in milliseconds.
353    #[must_use]
354    pub fn size_ms(&self) -> i64 {
355        self.size_ms
356    }
357
358    /// Window offset in milliseconds.
359    #[must_use]
360    pub fn offset_ms(&self) -> i64 {
361        self.offset_ms
362    }
363
364    /// O(1) window assignment. Floor-divides timestamp into window boundaries.
365    ///
366    /// # Panics
367    /// Panics if either boundary falls outside the `i64` timestamp range.
368    #[inline]
369    #[must_use]
370    pub fn assign(&self, timestamp: i64) -> WindowId {
371        self.try_assign(timestamp)
372            .expect("tumbling window boundaries must fit in i64")
373    }
374
375    /// Assigns a timestamp, rejecting boundaries outside the `i64` timestamp range.
376    ///
377    /// # Errors
378    ///
379    /// Returns an error when either window boundary cannot be represented as `i64`.
380    #[inline]
381    pub fn try_assign(&self, timestamp: i64) -> Result<WindowId, WindowAssignmentError> {
382        if let Some(adjusted) = timestamp.checked_sub(self.offset_ms) {
383            let quotient = adjusted.div_euclid(self.size_ms);
384            if let Some((start, end)) = quotient
385                .checked_mul(self.size_ms)
386                .and_then(|start| start.checked_add(self.offset_ms))
387                .and_then(|start| start.checked_add(self.size_ms).map(|end| (start, end)))
388            {
389                return Ok(WindowId::new(start, end));
390            }
391        }
392
393        self.assign_wide(timestamp)
394    }
395
396    #[cold]
397    fn assign_wide(&self, timestamp: i64) -> Result<WindowId, WindowAssignmentError> {
398        let size_ms = i128::from(self.size_ms);
399        let offset_ms = i128::from(self.offset_ms);
400        let adjusted = i128::from(timestamp) - offset_ms;
401        let window_start = adjusted.div_euclid(size_ms) * size_ms + offset_ms;
402        let window_end = window_start + size_ms;
403
404        match (i64::try_from(window_start), i64::try_from(window_end)) {
405            (Ok(start), Ok(end)) => Ok(WindowId::new(start, end)),
406            _ => Err(WindowAssignmentError::new(
407                timestamp,
408                window_start,
409                window_end,
410            )),
411        }
412    }
413}
414
415impl WindowAssigner for TumblingWindowAssigner {
416    #[inline]
417    fn assign_windows(&self, timestamp: i64) -> WindowIdVec {
418        let mut windows = WindowIdVec::new();
419        windows.push(self.assign(timestamp));
420        windows
421    }
422}
423
424#[cfg(test)]
425mod tests;
laminar_core/operator/window/mod.rs

laminar_core/operator/window/
mod.rs
