laminar_sql/
temporal.rs

1//! Canonical temporal-join planning contract.
2
3use std::ops::ControlFlow;
4
5use sqlparser::ast::{Statement, TableFactor, TableVersion, Visit, Visitor};
6
7/// Hard parser limit for probes produced from one left row.
8pub const MAX_TEMPORAL_PROBES_PER_ROW: usize = 256;
9
10/// Hard parser limit for the absolute value of a probe horizon (365 days).
11pub const MAX_TEMPORAL_PROBE_HORIZON_MS: i64 = 365 * 24 * 60 * 60 * 1_000;
12
13/// Count every temporal table version in a statement, including nested queries.
14#[must_use]
15pub fn temporal_table_version_count(statement: &Statement) -> usize {
16    #[derive(Default)]
17    struct Counter(usize);
18
19    impl Visitor for Counter {
20        type Break = ();
21
22        fn pre_visit_table_factor(
23            &mut self,
24            table_factor: &TableFactor,
25        ) -> ControlFlow<Self::Break> {
26            if matches!(
27                table_factor,
28                TableFactor::Table {
29                    version: Some(TableVersion::ForSystemTimeAsOf(_)),
30                    ..
31                }
32            ) {
33                self.0 += 1;
34            }
35            ControlFlow::Continue(())
36        }
37    }
38
39    let mut counter = Counter::default();
40    let _ = statement.visit(&mut counter);
41    counter.0
42}
43
44/// Temporal joins supported by the managed state machine.
45#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub enum TemporalJoinKind {
47    /// Emit only matching left rows.
48    Inner,
49    /// Emit every left row, null-extending misses.
50    Left,
51}
52
53/// Validated target-time schedule for one temporal-join state machine.
54#[derive(Debug, Clone, PartialEq, Eq)]
55pub struct TemporalProbeSchedule {
56    offsets_ms: Vec<i64>,
57    multi_horizon: bool,
58}
59
60impl TemporalProbeSchedule {
61    /// One lookup at the left row's event time.
62    #[must_use]
63    pub fn as_of() -> Self {
64        Self {
65            offsets_ms: vec![0],
66            multi_horizon: false,
67        }
68    }
69
70    /// Build a validated explicit schedule.
71    ///
72    /// # Errors
73    /// Returns an error for an empty list, duplicate offsets, or parser-limit violations.
74    pub fn list(offsets_ms: Vec<i64>) -> Result<Self, String> {
75        if offsets_ms.is_empty() {
76            return Err("temporal probe LIST requires at least one offset".into());
77        }
78        if offsets_ms.len() > MAX_TEMPORAL_PROBES_PER_ROW {
79            return Err(format!(
80                "temporal probe schedule exceeds the limit of {MAX_TEMPORAL_PROBES_PER_ROW} offsets per row"
81            ));
82        }
83        for (index, offset) in offsets_ms.iter().enumerate() {
84            validate_horizon(*offset)?;
85            if offsets_ms[..index].contains(offset) {
86                return Err(format!("temporal probe LIST repeats offset {offset}ms"));
87            }
88        }
89        Ok(Self {
90            offsets_ms,
91            multi_horizon: true,
92        })
93    }
94
95    /// Build a validated inclusive range schedule.
96    ///
97    /// # Errors
98    /// Returns an error for invalid bounds, a non-positive step, a range that is not
99    /// exactly divisible by its step, or parser-limit violations.
100    pub fn range(start_ms: i64, end_ms: i64, step_ms: i64) -> Result<Self, String> {
101        validate_horizon(start_ms)?;
102        validate_horizon(end_ms)?;
103        if start_ms > end_ms {
104            return Err("temporal probe RANGE start must not exceed its end".into());
105        }
106        if step_ms <= 0 {
107            return Err("temporal probe RANGE STEP must be positive".into());
108        }
109        let span = end_ms
110            .checked_sub(start_ms)
111            .ok_or_else(|| "temporal probe RANGE span overflows signed milliseconds".to_string())?;
112        if span % step_ms != 0 {
113            return Err(
114                "temporal probe RANGE end must be reachable by whole STEP increments".into(),
115            );
116        }
117        let count = span
118            .checked_div(step_ms)
119            .and_then(|steps| steps.checked_add(1))
120            .ok_or_else(|| "temporal probe RANGE size overflows".to_string())?;
121        let count = usize::try_from(count)
122            .map_err(|_| "temporal probe RANGE size cannot be represented".to_string())?;
123        if count > MAX_TEMPORAL_PROBES_PER_ROW {
124            return Err(format!(
125                "temporal probe schedule exceeds the limit of {MAX_TEMPORAL_PROBES_PER_ROW} offsets per row"
126            ));
127        }
128        let mut offsets_ms = Vec::with_capacity(count);
129        let mut offset = start_ms;
130        for index in 0..count {
131            offsets_ms.push(offset);
132            if index + 1 < count {
133                offset = offset.checked_add(step_ms).ok_or_else(|| {
134                    "temporal probe RANGE expansion overflows milliseconds".to_string()
135                })?;
136            }
137        }
138        Ok(Self {
139            offsets_ms,
140            multi_horizon: true,
141        })
142    }
143
144    /// Number of target-time probes produced per left row.
145    #[must_use]
146    pub fn len(&self) -> usize {
147        self.offsets_ms.len()
148    }
149
150    /// Whether this schedule contains no probes.
151    #[must_use]
152    pub fn is_empty(&self) -> bool {
153        self.offsets_ms.is_empty()
154    }
155
156    /// Deterministic offsets in milliseconds.
157    #[must_use]
158    pub fn offsets_ms(&self) -> &[i64] {
159        &self.offsets_ms
160    }
161
162    /// Whether the syntax requested typed `offset_ms` and `probe_time` outputs.
163    #[must_use]
164    pub const fn is_multi_horizon(&self) -> bool {
165        self.multi_horizon
166    }
167}
168
169fn validate_horizon(offset_ms: i64) -> Result<(), String> {
170    let magnitude = offset_ms
171        .checked_abs()
172        .ok_or_else(|| "temporal probe offset magnitude overflows".to_string())?;
173    if magnitude > MAX_TEMPORAL_PROBE_HORIZON_MS {
174        return Err(format!(
175            "temporal probe offset exceeds the parser horizon limit of {MAX_TEMPORAL_PROBE_HORIZON_MS}ms"
176        ));
177    }
178    Ok(())
179}
180
181#[cfg(test)]
182mod tests {
183    use super::*;
184
185    use sqlparser::parser::Parser;
186
187    #[test]
188    fn schedules_preserve_list_order_and_expand_ranges() {
189        let list = TemporalProbeSchedule::list(vec![5_000, -1_000, 0]).unwrap();
190        assert_eq!(list.offsets_ms(), [5_000, -1_000, 0]);
191
192        let range = TemporalProbeSchedule::range(-1_000, 1_000, 500).unwrap();
193        assert_eq!(range.offsets_ms(), [-1_000, -500, 0, 500, 1_000]);
194    }
195
196    #[test]
197    fn schedules_reject_ambiguous_or_excessive_expansion() {
198        assert!(TemporalProbeSchedule::list(vec![0, 0]).is_err());
199        assert!(TemporalProbeSchedule::range(0, 1_001, 1_000).is_err());
200        assert!(TemporalProbeSchedule::range(0, 256_000, 1_000).is_err());
201    }
202
203    #[test]
204    fn finds_temporal_versions_in_nested_queries() {
205        let statements = Parser::parse_sql(
206            &crate::parser::dialect::LaminarDialect::default(),
207            "WITH q AS (SELECT * FROM l JOIN r FOR SYSTEM_TIME AS OF l.ts ON l.k = r.k) SELECT * FROM q",
208        )
209        .unwrap();
210        assert_eq!(temporal_table_version_count(&statements[0]), 1);
211    }
212}
laminar_sql/temporal.rs

laminar_sql/
temporal.rs
