densa_data/window.rs
1//! Time windows: a table that keeps only its rows within a span of the newest time.
2
3use std::time::Duration;
4
5use arrow_schema::{DataType, Schema, TimeUnit};
6
7use crate::filter::resolve;
8use crate::value::{Kind, Value, ValueFn};
9use crate::view::{Rows, SortColumn, ViewState};
10use crate::DataError;
11
12/// Keeps a [`LiveTable`](crate::LiveTable) to its recent rows: those whose value in a time
13/// column is within a span of the newest. Set it with
14/// [`LiveTable::set_window`](crate::LiveTable::set_window).
15///
16/// The window ends at the largest finite value of the column among the live rows, so it
17/// moves with the data, not with the clock: replayed or delayed data is windowed by its own
18/// times, and nothing expires while no rows arrive. A row expires when the end is more than
19/// the span past its value; rows whose value is null or NaN never do.
20///
21/// ```
22/// use std::sync::Arc;
23/// use std::time::Duration;
24/// use arrow_array::{ArrayRef, Int64Array, RecordBatch, TimestampMillisecondArray};
25/// use densa_data::{LiveTable, Window};
26///
27/// let ticks = |ids: Vec<i64>, times: Vec<i64>| {
28/// RecordBatch::try_from_iter([
29/// ("id", Arc::new(Int64Array::from(ids)) as ArrayRef),
30/// ("time", Arc::new(TimestampMillisecondArray::from(times)) as ArrayRef),
31/// ])
32/// };
33/// let mut table = LiveTable::new(ticks(vec![], vec![])?.schema(), "id")?;
34/// table.set_window(Some(Window::duration("time", Duration::from_secs(60))))?;
35///
36/// table.upsert(ticks(vec![1, 2], vec![0, 30_000])?)?;
37/// let changes = table.upsert(ticks(vec![3], vec![75_000])?)?;
38/// assert_eq!(changes.deleted, 1); // row 1 is 75 s older than the newest row
39/// assert_eq!(table.len(), 2);
40/// # Ok::<(), Box<dyn std::error::Error>>(())
41/// ```
42#[derive(Debug, Clone, PartialEq)]
43pub struct Window {
44 column: String,
45 span: Span,
46}
47
48/// A span of a time column (a `Duration`) or of a number column (in its own units).
49#[derive(Debug, Clone, Copy, PartialEq)]
50pub(crate) enum Span {
51 Duration(Duration),
52 Value(f64),
53}
54
55impl Window {
56 /// A window of `span` over a column of timestamps, dates, times of day or durations.
57 #[must_use]
58 pub fn duration(column: impl Into<String>, span: Duration) -> Self {
59 Self {
60 column: column.into(),
61 span: Span::Duration(span),
62 }
63 }
64
65 /// A window of `span` over a column of numbers (integers, decimals or floats), in the
66 /// column's own units: `60_000.0` keeps a minute of milliseconds. `span` must be zero or
67 /// more. A span within a billionth of a whole number of an integer or decimal column's
68 /// units counts as that number, so float error (`0.29` hundredths) loses no unit.
69 #[must_use]
70 pub fn span(column: impl Into<String>, span: f64) -> Self {
71 Self {
72 column: column.into(),
73 span: Span::Value(span),
74 }
75 }
76
77 /// The time column's name.
78 pub fn column(&self) -> &str {
79 &self.column
80 }
81}
82
83/// How far below the newest value a row may be, in the column's units: whole units for
84/// integer columns (differences are whole, so the span rounds down without changing which
85/// rows stay), the span itself for floats.
86#[derive(Debug, Clone, Copy, PartialEq)]
87pub(crate) enum Reach {
88 Whole(i128),
89 Float(f64),
90}
91
92/// A window bound to its column, with the table's rows in the column's order.
93#[derive(Debug, Clone)]
94pub(crate) struct WindowState {
95 spec: Window,
96 reach: Reach,
97 /// Every live row, oldest first (nulls, then NaN, last).
98 pub(crate) rows: ViewState,
99 column: usize,
100 read: ValueFn,
101}
102
103impl WindowState {
104 pub(crate) fn compile(spec: &Window, schema: &Schema) -> Result<Self, DataError> {
105 let (column, kind, read) = resolve(schema, &spec.column)?;
106 let reach = reach(spec.span, schema.field(column).data_type(), kind)
107 .ok_or_else(|| DataError::WindowSpan(spec.column.clone()))?;
108 Ok(Self {
109 spec: spec.clone(),
110 reach,
111 rows: ViewState::sorted(SortColumn::ascending(column, read)),
112 column,
113 read,
114 })
115 }
116
117 pub(crate) fn spec(&self) -> &Window {
118 &self.spec
119 }
120
121 /// How many of the oldest rows are past the window.
122 pub(crate) fn expired(&self, rows: Rows<'_>) -> usize {
123 let value = |id| {
124 let at = rows.at(id);
125 (self.read)(rows.batches[at.batch].column(self.column).as_ref(), at.row)
126 };
127 // Values below +inf come first (-inf, then numbers); +inf, NaN and nulls after them.
128 // The newest is the last of those. If it is -inf, every row below +inf is -inf, and
129 // `-inf - -inf` is NaN: none is older.
130 let below = self.rows.partition_point(|id| below_infinity(value(id)));
131 let Some(newest) = below.checked_sub(1).and_then(|i| self.rows.get(i)) else {
132 return 0;
133 };
134 let newest = value(newest);
135 self.rows
136 .partition_point(|id| older(value(id), newest, self.reach))
137 }
138}
139
140/// `span` in the units of a column of `data_type`, of `kind`: `None` if they do not fit (a
141/// `Duration` goes with dates, times and timestamps, a number of zero or more with numbers).
142pub(crate) fn reach(span: Span, data_type: &DataType, kind: Kind) -> Option<Reach> {
143 Some(match (span, time_unit(data_type)) {
144 (Span::Duration(span), Some(unit)) => {
145 // Whole units, rounding down. Lossless: at most 2^64 seconds of nanoseconds.
146 Reach::Whole((span.as_nanos() / unit).cast_signed())
147 }
148 (Span::Value(span), None) if span >= 0.0 => match kind {
149 Kind::Int | Kind::UInt => Reach::Whole(whole(span)),
150 Kind::Decimal { scale } => Reach::Whole(whole(span * 10f64.powi(scale.into()))),
151 Kind::Float => Reach::Float(span),
152 Kind::Bool | Kind::Str => return None,
153 },
154 _ => return None,
155 })
156}
157
158const NANOS_PER_DAY: u128 = 86_400 * 1_000_000_000;
159
160/// The nanoseconds in one unit of a column of dates, timestamps, times or durations; `None`
161/// for other types.
162fn time_unit(data_type: &DataType) -> Option<u128> {
163 let unit = match data_type {
164 DataType::Timestamp(unit, _)
165 | DataType::Time32(unit)
166 | DataType::Time64(unit)
167 | DataType::Duration(unit) => unit,
168 DataType::Date32 => return Some(NANOS_PER_DAY),
169 DataType::Date64 => &TimeUnit::Millisecond,
170 _ => return None,
171 };
172 Some(match unit {
173 TimeUnit::Second => 1_000_000_000,
174 TimeUnit::Millisecond => 1_000_000,
175 TimeUnit::Microsecond => 1_000,
176 TimeUnit::Nanosecond => 1,
177 })
178}
179
180/// `span` rounded down to a whole number (saturating: a span past `i128` keeps every row).
181/// A whole number a float error away is taken as it: `0.29 * 100.0` is 28.999999999999996.
182fn whole(span: f64) -> i128 {
183 let nearest = span.round();
184 let span = match (span - nearest).abs() <= NEAR * nearest.abs() {
185 true => nearest,
186 false => span.floor(),
187 };
188 // Saturating cast; `span` is not NaN (checked by the caller's `>= 0.0`).
189 span as i128
190}
191
192/// How close, relative to it, a span must be to a whole number to be taken as it.
193const NEAR: f64 = 1e-9;
194
195fn below_infinity(value: Value<'_>) -> bool {
196 match value {
197 Value::Int(_) | Value::UInt(_) | Value::Decimal(_) => true,
198 Value::Float(v) => v < f64::INFINITY,
199 Value::Null | Value::Bool(_) | Value::Str(_) => false,
200 }
201}
202
203/// Whether `value` is more than `reach` below `newest`, the window's end (a value of the same
204/// column below +inf).
205fn older(value: Value<'_>, newest: Value<'_>, reach: Reach) -> bool {
206 let whole = |v: Value<'_>| match v {
207 Value::Int(v) => Some(i128::from(v)),
208 Value::UInt(v) => Some(i128::from(v)),
209 Value::Decimal(v) => Some(v),
210 _ => None,
211 };
212 match (reach, value, newest) {
213 (Reach::Float(reach), Value::Float(value), Value::Float(newest)) => newest - value > reach,
214 (Reach::Whole(reach), value, newest) => match (whole(value), whole(newest)) {
215 // Saturating: two decimals far apart are as far apart as can be.
216 (Some(value), Some(newest)) => newest.saturating_sub(value) > reach,
217 _ => false,
218 },
219 _ => false,
220 }
221}