reifydb_flow/window/
span.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::fmt::Debug;
5
6use reifydb_codec::row::operator::StateCodec;
7use reifydb_core::metrics::heap::HeapSize;
8use reifydb_macro::operator_state;
9use reifydb_value::value::datetime::DateTime;
10
11use crate::operator::state::seal::coord::{Coord, IsZero};
12
13pub type SlotCoord<S> = <S as Slot>::Coord;
14
15pub trait WindowAnchor: Slot<Coord = Self> + Coord {}
16
17impl<T> WindowAnchor for T where T: Slot<Coord = T> + Coord {}
18
19pub type SlotSpan<S> = <<S as Slot>::Coord as Coord>::Span;
20
21pub trait Slot: Copy + Ord + Debug + StateCodec {
22	type Coord: Coord;
23
24	fn order_key(&self) -> Self::Coord;
25
26	fn from_order_key(coord: Self::Coord) -> Self;
27}
28
29impl Slot for DateTime {
30	type Coord = DateTime;
31
32	fn order_key(&self) -> DateTime {
33		*self
34	}
35
36	fn from_order_key(coord: DateTime) -> Self {
37		coord
38	}
39}
40
41#[operator_state]
42#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
43pub struct WindowSpan<T> {
44	pub start: T,
45	pub end: T,
46}
47
48impl<T: HeapSize> HeapSize for WindowSpan<T> {
49	fn heap_size(&self) -> usize {
50		self.start.heap_size() + self.end.heap_size()
51	}
52}
53
54impl<C> WindowSpan<C>
55where
56	C: Coord,
57{
58	#[inline]
59	pub fn for_coord(coord: C, span: C::Span) -> Self {
60		assert!(!span.is_zero(), "WindowSpan::for_coord: span must be > 0");
61		let start = coord.floor_to(span);
62		Self {
63			start,
64			end: start.add_span(span),
65		}
66	}
67
68	#[inline]
69	pub fn new(start: C, end: C) -> Self {
70		assert!(start < end, "WindowSpan::new: start ({start:?}) must be < end ({end:?})");
71		Self {
72			start,
73			end,
74		}
75	}
76
77	#[inline]
78	pub fn duration(&self) -> C::Span {
79		self.end.span_since(self.start)
80	}
81
82	#[inline]
83	pub fn contains(&self, coord: C) -> bool {
84		coord >= self.start && coord < self.end
85	}
86
87	#[inline]
88	pub fn next(&self) -> Self {
89		let span = self.duration();
90		Self {
91			start: self.end,
92			end: self.end.add_span(span),
93		}
94	}
95}
96
97#[cfg(test)]
98mod tests {
99	use reifydb_value::{
100		factory::time::{at_millis, millis},
101		value::duration::Duration,
102	};
103
104	use super::*;
105	use crate::operator::state::seal::policy::{is_sealed, seal_horizon};
106
107	#[test]
108	fn for_coord_aligns_datetime_to_span() {
109		let coord = DateTime::from_ymd_hms(2024, 1, 15, 10, 30, 25).unwrap();
110		let one_second = Duration::from_seconds(1).unwrap();
111		let one_minute = Duration::from_seconds(60).unwrap();
112
113		// A sub-minute (1s) window must stay 1s, not round up to a minute.
114		let sec = WindowSpan::for_coord(coord, one_second);
115		assert_eq!(sec.start, DateTime::from_ymd_hms(2024, 1, 15, 10, 30, 25).unwrap());
116		assert_eq!(sec.end, DateTime::from_ymd_hms(2024, 1, 15, 10, 30, 26).unwrap());
117		assert_eq!(sec.duration(), one_second);
118
119		// A 1m window aligns the coord down to the minute boundary.
120		let min = WindowSpan::for_coord(coord, one_minute);
121		assert_eq!(min.start, DateTime::from_ymd_hms(2024, 1, 15, 10, 30, 0).unwrap());
122		assert_eq!(min.end, DateTime::from_ymd_hms(2024, 1, 15, 10, 31, 0).unwrap());
123		assert!(min.contains(coord));
124		assert!(!min.contains(min.end));
125	}
126
127	#[test]
128	fn contains_is_half_open() {
129		let span = WindowSpan::new(at_millis(100), at_millis(200));
130		assert!(span.contains(at_millis(100)));
131		assert!(span.contains(at_millis(199)));
132		assert!(!span.contains(at_millis(200)));
133		assert!(!span.contains(at_millis(99)));
134	}
135
136	#[test]
137	fn boundary_coord_belongs_to_next_window() {
138		// An event at exactly window_end must not be claimed by the current window.
139		let cur = WindowSpan::for_coord(at_millis(60), millis(60));
140		let nxt = cur.next();
141		assert!(!cur.contains(at_millis(120)));
142		assert!(nxt.contains(at_millis(120)));
143		assert_eq!(nxt, WindowSpan::new(at_millis(120), at_millis(180)));
144	}
145
146	#[test]
147	#[should_panic(expected = "span must be > 0")]
148	fn zero_duration_panics() {
149		WindowSpan::for_coord(at_millis(10), Duration::zero());
150	}
151
152	#[test]
153	#[should_panic(expected = "must be <")]
154	fn empty_span_panics() {
155		WindowSpan::new(at_millis(100), at_millis(100));
156	}
157
158	#[test]
159	fn a_time_coordinate_can_only_have_a_duration_subtracted_from_it() {
160		// A seal horizon is watermark - lateness; with both sides a bare u64 nothing stopped a
161		// millisecond span reaching a nanosecond coordinate, yielding a horizon a million times too
162		// small. Pairing a coordinate with its own Span makes the wrong subtraction fail to compile.
163		let watermark = DateTime::from_epoch_millis(6_060_000).expect("representable instant");
164		let one_minute = Duration::from_seconds(60).expect("representable span");
165
166		assert_eq!(
167			watermark.saturating_sub_span(one_minute),
168			DateTime::from_epoch_millis(6_000_000).expect("representable"),
169			"a minute behind the watermark is a minute, not a million times less"
170		);
171		assert_eq!(<DateTime as Coord>::span_millis(one_minute), Some(60_000));
172	}
173
174	#[test]
175	fn a_coordinate_survives_the_round_trip_through_its_storage_encoding() {
176		// to_order/from_order are the persisted expiry-index encoding. They must be exact inverses:
177		// a lossy round trip would move a window's anchor and either seal it early or strand it.
178		let coord = DateTime::from_epoch_millis(1_234_567).expect("representable");
179		assert_eq!(<DateTime as Coord>::from_order(coord.to_order()), coord);
180	}
181
182	#[test]
183	fn a_seal_horizon_leaves_the_window_exactly_at_the_boundary_admissible() {
184		// The boundary is load-bearing in both directions. A window whose start sits exactly one lateness
185		// span behind the watermark is still reachable by a late event, so sealing it would discard a
186		// legitimate retraction; sealing nothing would let state grow without bound.
187		let watermark = DateTime::from_epoch_millis(6_060_000).expect("representable");
188		let horizon = seal_horizon(watermark, Duration::from_seconds(60).expect("representable"));
189
190		let at_boundary = DateTime::from_epoch_millis(6_000_000).expect("representable");
191		let before_boundary = DateTime::from_epoch_millis(5_999_999).expect("representable");
192
193		assert!(!is_sealed(at_boundary, horizon), "the boundary window is still live");
194		assert!(is_sealed(before_boundary, horizon), "anything older is sealed");
195	}
196}
reifydb_flow/window/span.rs

reifydb_flow/window/
span.rs
