laminar_sql/translator/join_translator/
mod.rs

1//! Join operator configuration builder
2//!
3//! Translates parsed join analysis into operator configurations
4//! for stream-stream joins and lookup joins.
5
6use std::time::Duration;
7
8use crate::parser::join_parser::{JoinAnalysis, JoinType, MultiJoinAnalysis};
9use crate::temporal::{TemporalJoinKind, TemporalProbeSchedule};
10
11/// Configuration for stream-stream join operator
12#[derive(Debug, Clone)]
13pub struct StreamJoinConfig {
14    /// SQL join kind.
15    pub join_type: JoinType,
16    /// Ordered left-side equality key columns.
17    pub left_keys: Vec<String>,
18    /// Ordered right-side equality key columns.
19    pub right_keys: Vec<String>,
20    /// Left side time column for interval matching
21    pub left_time_column: String,
22    /// Right side time column for interval matching
23    pub right_time_column: String,
24    /// Left side table name
25    pub left_table: String,
26    /// Right side table name
27    pub right_table: String,
28    /// Time bound for joining (max time difference between events)
29    pub time_bound: Duration,
30}
31
32/// Configuration for lookup join operator
33#[derive(Debug, Clone)]
34pub struct LookupJoinConfig {
35    /// Stream side key column
36    pub stream_key: String,
37    /// Lookup table key column
38    pub lookup_key: String,
39    /// Join type
40    pub join_type: LookupJoinType,
41    /// Cache TTL for lookup results
42    pub cache_ttl: Duration,
43}
44
45/// Lookup join types
46#[derive(Debug, Clone, Copy, PartialEq, Eq)]
47pub enum LookupJoinType {
48    /// Stream event required, lookup optional
49    Inner,
50    /// Stream event always emitted, lookup optional
51    Left,
52}
53
54/// Configuration for temporal join operator (FOR SYSTEM_TIME AS OF).
55#[derive(Debug, Clone)]
56pub struct TemporalJoinTranslatorConfig {
57    /// Left input relation.
58    pub left_table: String,
59    /// Versioned right input relation.
60    pub right_table: String,
61    /// Ordered left-side equality key columns.
62    pub left_key_columns: Vec<String>,
63    /// Ordered right-side equality key columns.
64    pub right_key_columns: Vec<String>,
65    /// Left event-time column.
66    pub left_time_column: String,
67    /// Explicit right event-time/version column.
68    pub right_time_column: String,
69    /// INNER or LEFT output semantics.
70    pub join_kind: TemporalJoinKind,
71    /// One canonical target-time schedule.
72    pub probe_schedule: TemporalProbeSchedule,
73    /// Alias exposing multi-horizon `offset_ms` and `probe_time` columns.
74    pub probe_alias: Option<String>,
75}
76
77/// Union type for join operator configurations
78#[derive(Debug, Clone)]
79pub enum JoinOperatorConfig {
80    /// Stream-stream join
81    StreamStream(StreamJoinConfig),
82    /// Lookup join
83    Lookup(LookupJoinConfig),
84    /// Temporal join (FOR SYSTEM_TIME AS OF)
85    Temporal(TemporalJoinTranslatorConfig),
86}
87
88impl std::fmt::Display for LookupJoinType {
89    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
90        match self {
91            LookupJoinType::Inner => write!(f, "INNER"),
92            LookupJoinType::Left => write!(f, "LEFT"),
93        }
94    }
95}
96
97impl std::fmt::Display for JoinType {
98    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
99        match self {
100            JoinType::Inner => write!(f, "INNER"),
101            JoinType::Left => write!(f, "LEFT"),
102            JoinType::Right => write!(f, "RIGHT"),
103            JoinType::Full => write!(f, "FULL"),
104            JoinType::LeftSemi => write!(f, "LEFT SEMI"),
105            JoinType::LeftAnti => write!(f, "LEFT ANTI"),
106            JoinType::RightSemi => write!(f, "RIGHT SEMI"),
107            JoinType::RightAnti => write!(f, "RIGHT ANTI"),
108        }
109    }
110}
111
112impl std::fmt::Display for StreamJoinConfig {
113    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
114        write!(f, "{} JOIN ON ", self.join_type)?;
115        for (index, (left, right)) in self.left_keys.iter().zip(&self.right_keys).enumerate() {
116            if index != 0 {
117                write!(f, " AND ")?;
118            }
119            write!(
120                f,
121                "{}.{} = {}.{}",
122                self.left_table, left, self.right_table, right
123            )?;
124        }
125        write!(
126            f,
127            " (bound: {}s, time: {} ~ {})",
128            self.time_bound.as_secs(),
129            self.left_time_column,
130            self.right_time_column,
131        )
132    }
133}
134
135impl std::fmt::Display for LookupJoinConfig {
136    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
137        write!(
138            f,
139            "{} LOOKUP JOIN ON stream.{} = lookup.{} (cache_ttl: {}s)",
140            self.join_type,
141            self.stream_key,
142            self.lookup_key,
143            self.cache_ttl.as_secs()
144        )
145    }
146}
147
148impl std::fmt::Display for TemporalJoinTranslatorConfig {
149    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
150        write!(f, "{:?} TEMPORAL JOIN ON ", self.join_kind)?;
151        for (index, (left, right)) in self
152            .left_key_columns
153            .iter()
154            .zip(&self.right_key_columns)
155            .enumerate()
156        {
157            if index != 0 {
158                write!(f, " AND ")?;
159            }
160            write!(
161                f,
162                "{}.{} = {}.{}",
163                self.left_table, left, self.right_table, right
164            )?;
165        }
166        write!(
167            f,
168            " ({} -> {}, probes: {})",
169            self.left_time_column,
170            self.right_time_column,
171            self.probe_schedule.len(),
172        )
173    }
174}
175
176impl std::fmt::Display for JoinOperatorConfig {
177    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
178        match self {
179            JoinOperatorConfig::StreamStream(c) => write!(f, "{c}"),
180            JoinOperatorConfig::Lookup(c) => write!(f, "{c}"),
181            JoinOperatorConfig::Temporal(c) => write!(f, "{c}"),
182        }
183    }
184}
185
186impl JoinOperatorConfig {
187    /// Create from join analysis.
188    ///
189    /// # Errors
190    /// Returns an error when the analyzed join has no supported, complete runtime contract.
191    pub fn from_analysis(analysis: &JoinAnalysis) -> Result<Self, String> {
192        if analysis.is_temporal_join() {
193            let join_kind = match analysis.join_type {
194                JoinType::Inner => TemporalJoinKind::Inner,
195                JoinType::Left => TemporalJoinKind::Left,
196                unsupported => {
197                    return Err(format!(
198                        "temporal joins support only INNER or LEFT joins; {unsupported:?} is unsupported"
199                    ));
200                }
201            };
202            let left_time_column = analysis.left_time_column.clone().ok_or_else(|| {
203                "temporal joins require an explicit left event-time column".to_string()
204            })?;
205            let right_time_column = analysis.right_time_column.clone().ok_or_else(|| {
206                "temporal joins require an explicit right event-time column from the source contract"
207                    .to_string()
208            })?;
209            let probe_schedule = analysis
210                .temporal_probe_schedule
211                .clone()
212                .ok_or_else(|| "temporal join probe schedule is missing".to_string())?;
213            if probe_schedule.is_multi_horizon() && analysis.temporal_probe_alias.is_none() {
214                return Err("multi-horizon temporal probes require an output alias".into());
215            }
216            let (left_key_columns, right_key_columns) = ordered_key_columns(analysis);
217            validate_temporal_key_columns(&left_key_columns, &right_key_columns)?;
218            return Ok(JoinOperatorConfig::Temporal(TemporalJoinTranslatorConfig {
219                left_table: analysis.left_table.clone(),
220                right_table: analysis.right_table.clone(),
221                left_key_columns,
222                right_key_columns,
223                left_time_column,
224                right_time_column,
225                join_kind,
226                probe_schedule,
227                probe_alias: analysis.temporal_probe_alias.clone(),
228            }));
229        }
230
231        if analysis.is_lookup_join {
232            if !analysis.additional_key_columns.is_empty() {
233                return Err(
234                    "lookup joins support exactly one equality key; composite predicates are not implemented"
235                        .to_string(),
236                );
237            }
238            let join_type = match analysis.join_type {
239                JoinType::Inner => LookupJoinType::Inner,
240                JoinType::Left => LookupJoinType::Left,
241                unsupported => {
242                    return Err(format!(
243                        "lookup joins support only INNER or LEFT joins; {unsupported:?} is unsupported"
244                    ));
245                }
246            };
247            Ok(JoinOperatorConfig::Lookup(LookupJoinConfig {
248                stream_key: analysis.left_key_column.clone(),
249                lookup_key: analysis.right_key_column.clone(),
250                join_type,
251                cache_ttl: Duration::from_secs(300), // Default 5 min
252            }))
253        } else {
254            let time_bound = analysis.time_bound.ok_or_else(|| {
255                "stream-stream joins require an explicit finite time bound".to_string()
256            })?;
257            if time_bound.is_zero() {
258                return Err("stream-stream joins require a positive finite time bound".to_string());
259            }
260            let (left_keys, right_keys) = ordered_key_columns(analysis);
261            Ok(JoinOperatorConfig::StreamStream(StreamJoinConfig {
262                join_type: analysis.join_type,
263                left_keys,
264                right_keys,
265                left_time_column: analysis.left_time_column.clone().unwrap_or_default(),
266                right_time_column: analysis.right_time_column.clone().unwrap_or_default(),
267                left_table: analysis.left_table.clone(),
268                right_table: analysis.right_table.clone(),
269                time_bound,
270            }))
271        }
272    }
273
274    /// Create from a multi-join analysis, producing one config per join step.
275    ///
276    /// # Errors
277    /// Returns an error when any join step has no supported, complete runtime contract.
278    pub fn from_multi_analysis(multi: &MultiJoinAnalysis) -> Result<Vec<Self>, String> {
279        multi.joins.iter().map(Self::from_analysis).collect()
280    }
281
282    /// Check if this is a stream-stream join.
283    #[must_use]
284    pub fn is_stream_stream(&self) -> bool {
285        matches!(self, JoinOperatorConfig::StreamStream(_))
286    }
287
288    /// Check if this is a lookup join.
289    #[must_use]
290    pub fn is_lookup(&self) -> bool {
291        matches!(self, JoinOperatorConfig::Lookup(_))
292    }
293
294    /// Check if this is a temporal join.
295    #[must_use]
296    pub fn is_temporal(&self) -> bool {
297        matches!(self, JoinOperatorConfig::Temporal(_))
298    }
299
300    /// Get the ordered left-side equality key columns.
301    #[must_use]
302    pub fn left_keys(&self) -> &[String] {
303        match self {
304            JoinOperatorConfig::StreamStream(config) => &config.left_keys,
305            JoinOperatorConfig::Lookup(config) => std::slice::from_ref(&config.stream_key),
306            JoinOperatorConfig::Temporal(config) => &config.left_key_columns,
307        }
308    }
309
310    /// Get the ordered right-side equality key columns.
311    #[must_use]
312    pub fn right_keys(&self) -> &[String] {
313        match self {
314            JoinOperatorConfig::StreamStream(config) => &config.right_keys,
315            JoinOperatorConfig::Lookup(config) => std::slice::from_ref(&config.lookup_key),
316            JoinOperatorConfig::Temporal(config) => &config.right_key_columns,
317        }
318    }
319}
320
321fn ordered_key_columns(analysis: &JoinAnalysis) -> (Vec<String>, Vec<String>) {
322    let mut left = Vec::with_capacity(1 + analysis.additional_key_columns.len());
323    let mut right = Vec::with_capacity(1 + analysis.additional_key_columns.len());
324    left.push(analysis.left_key_column.clone());
325    right.push(analysis.right_key_column.clone());
326    for (left_column, right_column) in &analysis.additional_key_columns {
327        left.push(left_column.clone());
328        right.push(right_column.clone());
329    }
330    (left, right)
331}
332
333fn validate_temporal_key_columns(left: &[String], right: &[String]) -> Result<(), String> {
334    if left.is_empty()
335        || left.len() != right.len()
336        || left.iter().chain(right).any(String::is_empty)
337    {
338        return Err(
339            "temporal join equality keys must be non-empty and have matching cardinality".into(),
340        );
341    }
342    Ok(())
343}
344
345impl StreamJoinConfig {
346    /// Create a new stream-stream join configuration.
347    #[must_use]
348    pub fn new(
349        join_type: JoinType,
350        left_keys: Vec<String>,
351        right_keys: Vec<String>,
352        time_bound: Duration,
353    ) -> Self {
354        Self {
355            join_type,
356            left_keys,
357            right_keys,
358            left_time_column: String::new(),
359            right_time_column: String::new(),
360            left_table: String::new(),
361            right_table: String::new(),
362            time_bound,
363        }
364    }
365}
366
367impl LookupJoinConfig {
368    /// Create a new lookup join configuration.
369    #[must_use]
370    pub fn new(
371        stream_key: String,
372        lookup_key: String,
373        join_type: LookupJoinType,
374        cache_ttl: Duration,
375    ) -> Self {
376        Self {
377            stream_key,
378            lookup_key,
379            join_type,
380            cache_ttl,
381        }
382    }
383
384    /// Create an inner lookup join configuration.
385    #[must_use]
386    pub fn inner(stream_key: String, lookup_key: String) -> Self {
387        Self::new(
388            stream_key,
389            lookup_key,
390            LookupJoinType::Inner,
391            Duration::from_secs(300),
392        )
393    }
394
395    /// Create a left lookup join configuration.
396    #[must_use]
397    pub fn left(stream_key: String, lookup_key: String) -> Self {
398        Self::new(
399            stream_key,
400            lookup_key,
401            LookupJoinType::Left,
402            Duration::from_secs(300),
403        )
404    }
405
406    /// Set the cache TTL.
407    #[must_use]
408    pub fn with_cache_ttl(mut self, ttl: Duration) -> Self {
409        self.cache_ttl = ttl;
410        self
411    }
412}
413
414#[cfg(test)]
415mod tests;
laminar_sql/translator/join_translator/mod.rs

laminar_sql/translator/join_translator/
mod.rs
