laminar_sql/parser/join_parser/
mod.rs

1//! Join query analysis and extraction
2//!
3//! This module analyzes JOIN clauses to extract:
4//! - Join type (INNER, LEFT, RIGHT, FULL)
5//! - Key columns for join condition
6//! - Time bounds for stream-stream joins
7//! - Detection of lookup joins vs stream-stream joins
8
9use std::time::Duration;
10
11use sqlparser::ast::{
12    BinaryOperator, Expr, JoinConstraint, JoinOperator, ObjectName, ObjectNamePart, Select,
13    TableFactor, TableVersion,
14};
15
16use super::window_rewriter::WindowRewriter;
17use super::ParseError;
18use crate::temporal::TemporalProbeSchedule;
19
20mod temporal_probe;
21
22pub(crate) use temporal_probe::parse_temporal_probe_query;
23
24/// Join type classification
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum JoinType {
27    /// INNER JOIN
28    Inner,
29    /// LEFT \[OUTER\] JOIN
30    Left,
31    /// RIGHT \[OUTER\] JOIN
32    Right,
33    /// FULL \[OUTER\] JOIN
34    Full,
35    /// LEFT SEMI JOIN — emit left rows with at least one match
36    LeftSemi,
37    /// LEFT ANTI JOIN — emit left rows with no match
38    LeftAnti,
39    /// RIGHT SEMI JOIN — emit right rows with at least one match
40    RightSemi,
41    /// RIGHT ANTI JOIN — emit right rows with no match
42    RightAnti,
43}
44
45/// Unresolved time column refs from a BETWEEN clause.
46#[derive(Debug, Clone)]
47struct RawTimeCols {
48    expr_qualifier: String,
49    expr_col: String,
50    low_qualifier: String,
51    low_col: String,
52}
53
54#[derive(Debug, Clone, Copy, PartialEq, Eq)]
55enum JoinSide {
56    Left,
57    Right,
58}
59
60#[derive(Debug, Clone, Copy)]
61struct JoinSides<'a> {
62    left_table: &'a str,
63    right_table: &'a str,
64    left_alias: Option<&'a str>,
65    right_alias: Option<&'a str>,
66}
67
68impl JoinSides<'_> {
69    fn resolve_qualifier(&self, qualifier: &str, context: &str) -> Result<JoinSide, ParseError> {
70        let is_left =
71            qualifier == self.left_table || self.left_alias.is_some_and(|alias| qualifier == alias);
72        let is_right = qualifier == self.right_table
73            || self.right_alias.is_some_and(|alias| qualifier == alias);
74
75        match (is_left, is_right) {
76            (true, false) => Ok(JoinSide::Left),
77            (false, true) => Ok(JoinSide::Right),
78            (true, true) => Err(ParseError::StreamingError(format!(
79                "{context} must use unambiguous left and right join input names; qualifier '{qualifier}' names both inputs"
80            ))),
81            (false, false) => Err(ParseError::StreamingError(format!(
82                "{context} must use unambiguous left and right join input names; qualifier '{qualifier}' names neither input"
83            ))),
84        }
85    }
86}
87
88/// Resolve BETWEEN time columns to `(left_time_col, right_time_col)` using
89/// table qualifiers.
90fn resolve_time_cols(
91    raw: &RawTimeCols,
92    left_table: &str,
93    right_table: &str,
94    left_alias: Option<&str>,
95    right_alias: Option<&str>,
96) -> Result<(String, String), ParseError> {
97    let sides = JoinSides {
98        left_table,
99        right_table,
100        left_alias,
101        right_alias,
102    };
103    let expr_side = sides.resolve_qualifier(&raw.expr_qualifier, "streaming interval timestamp")?;
104    let low_side = sides.resolve_qualifier(&raw.low_qualifier, "streaming interval timestamp")?;
105
106    match (expr_side, low_side) {
107        (JoinSide::Right, JoinSide::Left) => Ok((raw.low_col.clone(), raw.expr_col.clone())),
108        (JoinSide::Left, JoinSide::Right) => Err(ParseError::StreamingError(
109            "streaming interval joins require the right timestamp BETWEEN the left timestamp and left timestamp + interval"
110                .to_string(),
111        )),
112        _ => Err(ParseError::StreamingError(
113            "streaming interval join timestamps must reference opposite join inputs".to_string(),
114        )),
115    }
116}
117
118/// Analysis result for a JOIN clause
119#[derive(Debug, Clone, PartialEq, Eq)]
120pub struct JoinAnalysis {
121    /// Type of join (inner, left, right, full)
122    pub join_type: JoinType,
123    /// Left side table name
124    pub left_table: String,
125    /// Right side table name
126    pub right_table: String,
127    /// Left side key column
128    pub left_key_column: String,
129    /// Right side key column
130    pub right_key_column: String,
131    /// Time bound for stream-stream joins (None for lookup joins)
132    pub time_bound: Option<Duration>,
133    /// Whether this is a lookup join (no time bound)
134    pub is_lookup_join: bool,
135    /// Left side alias (if any)
136    pub left_alias: Option<String>,
137    /// Right side alias (if any)
138    pub right_alias: Option<String>,
139    /// Left side time column for a bounded join
140    pub left_time_column: Option<String>,
141    /// Right side time column for a bounded join
142    pub right_time_column: Option<String>,
143    /// Target-time schedule for a temporal join.
144    pub temporal_probe_schedule: Option<TemporalProbeSchedule>,
145    /// Pseudo-table alias exposing `offset_ms` and `probe_time` for multi-horizon probes.
146    pub temporal_probe_alias: Option<String>,
147    /// Additional key columns for composite join keys (beyond the primary key pair)
148    pub additional_key_columns: Vec<(String, String)>,
149}
150
151impl JoinAnalysis {
152    /// Create a stream-stream join analysis
153    #[must_use]
154    pub fn stream_stream(
155        left_table: String,
156        right_table: String,
157        left_key: String,
158        right_key: String,
159        time_bound: Duration,
160        join_type: JoinType,
161    ) -> Self {
162        Self {
163            join_type,
164            left_table,
165            right_table,
166            left_key_column: left_key,
167            right_key_column: right_key,
168            time_bound: Some(time_bound),
169            is_lookup_join: false,
170            left_alias: None,
171            right_alias: None,
172            left_time_column: None,
173            right_time_column: None,
174            temporal_probe_schedule: None,
175            temporal_probe_alias: None,
176            additional_key_columns: vec![],
177        }
178    }
179
180    /// Create a lookup join analysis
181    #[must_use]
182    pub fn lookup(
183        left_table: String,
184        right_table: String,
185        left_key: String,
186        right_key: String,
187        join_type: JoinType,
188    ) -> Self {
189        Self {
190            join_type,
191            left_table,
192            right_table,
193            left_key_column: left_key,
194            right_key_column: right_key,
195            time_bound: None,
196            is_lookup_join: true,
197            left_alias: None,
198            right_alias: None,
199            left_time_column: None,
200            right_time_column: None,
201            temporal_probe_schedule: None,
202            temporal_probe_alias: None,
203            additional_key_columns: vec![],
204        }
205    }
206
207    /// Create a temporal join analysis (FOR SYSTEM_TIME AS OF).
208    #[must_use]
209    pub fn temporal(
210        left_table: String,
211        right_table: String,
212        left_key: String,
213        right_key: String,
214        left_time_column: String,
215        join_type: JoinType,
216    ) -> Self {
217        Self {
218            join_type,
219            left_table,
220            right_table,
221            left_key_column: left_key,
222            right_key_column: right_key,
223            time_bound: None,
224            is_lookup_join: false,
225            left_alias: None,
226            right_alias: None,
227            left_time_column: Some(left_time_column),
228            right_time_column: None,
229            temporal_probe_schedule: Some(TemporalProbeSchedule::as_of()),
230            temporal_probe_alias: None,
231            additional_key_columns: vec![],
232        }
233    }
234
235    /// True if this step has a `BETWEEN` bound or `FOR SYSTEM_TIME AS OF`.
236    #[must_use]
237    pub fn is_bounded(&self) -> bool {
238        self.time_bound.is_some() || self.is_temporal_join()
239    }
240
241    /// Whether this step uses versioned event-time lookup state.
242    #[must_use]
243    pub fn is_temporal_join(&self) -> bool {
244        self.temporal_probe_schedule.is_some()
245    }
246}
247
248/// Analyze a SELECT statement for join information.
249///
250/// # Errors
251///
252/// Returns `ParseError::StreamingError` if:
253/// - Join constraint is not supported
254/// - Cannot extract key columns
255pub fn analyze_join(select: &Select) -> Result<Option<JoinAnalysis>, ParseError> {
256    let from = &select.from;
257    if from.is_empty() {
258        return Ok(None);
259    }
260
261    let first_table = &from[0];
262    if first_table.joins.is_empty() {
263        return Ok(None);
264    }
265
266    // Extract left table information
267    let left_table = extract_table_name(&first_table.relation)?;
268    let left_alias = extract_table_alias(&first_table.relation);
269
270    // Analyze the first join
271    let join = &first_table.joins[0];
272    let right_table = extract_table_name(&join.relation)?;
273    let right_alias = extract_table_alias(&join.relation);
274
275    let join_type = map_join_operator(&join.join_operator)?;
276    let sides = JoinSides {
277        left_table: &left_table,
278        right_table: &right_table,
279        left_alias: left_alias.as_deref(),
280        right_alias: right_alias.as_deref(),
281    };
282
283    // Check for temporal join (FOR SYSTEM_TIME AS OF)
284    if let Some(left_time_col) = extract_temporal_left_time(&join.relation, &sides)? {
285        let (left_key, right_key, additional, time_bound, time_cols) =
286            analyze_join_constraint(&join.join_operator, &sides)?;
287        if time_bound.is_some() || time_cols.is_some() {
288            return Err(ParseError::StreamingError(
289                "temporal joins do not accept an additional time-bound predicate".into(),
290            ));
291        }
292        let mut analysis = JoinAnalysis::temporal(
293            left_table,
294            right_table,
295            left_key,
296            right_key,
297            left_time_col,
298            join_type,
299        );
300        analysis.left_alias = left_alias;
301        analysis.right_alias = right_alias;
302        analysis.additional_key_columns = additional;
303        return Ok(Some(analysis));
304    }
305
306    // Analyze the join constraint
307    let (left_key, right_key, additional, time_bound, time_cols) =
308        analyze_join_constraint(&join.join_operator, &sides)?;
309
310    let mut analysis = if let Some(tb) = time_bound {
311        JoinAnalysis::stream_stream(left_table, right_table, left_key, right_key, tb, join_type)
312    } else {
313        JoinAnalysis::lookup(left_table, right_table, left_key, right_key, join_type)
314    };
315
316    analysis.left_alias.clone_from(&left_alias);
317    analysis.right_alias.clone_from(&right_alias);
318    analysis.additional_key_columns = additional;
319
320    if let Some(ref raw) = time_cols {
321        let (lt, rt) = resolve_time_cols(
322            raw,
323            &analysis.left_table,
324            &analysis.right_table,
325            left_alias.as_deref(),
326            right_alias.as_deref(),
327        )?;
328        analysis.left_time_column = Some(lt);
329        analysis.right_time_column = Some(rt);
330    }
331
332    Ok(Some(analysis))
333}
334
335/// Extract table name from a TableFactor.
336fn extract_table_name(factor: &TableFactor) -> Result<String, ParseError> {
337    match factor {
338        TableFactor::Table { name, .. } => match name.0.as_slice() {
339            [ObjectNamePart::Identifier(ident)] => Ok(ident.value.clone()),
340            _ => Err(ParseError::StreamingError(
341                "streaming joins require single-part relation names; qualify the relation in the catalog before planning the join"
342                    .to_string(),
343            )),
344        },
345        TableFactor::Derived { alias, .. } => {
346            if let Some(alias) = alias {
347                Ok(alias.name.value.clone())
348            } else {
349                Err(ParseError::StreamingError(
350                    "Derived table without alias not supported".to_string(),
351                ))
352            }
353        }
354        _ => Err(ParseError::StreamingError(
355            "Unsupported table factor type".to_string(),
356        )),
357    }
358}
359
360/// Extract and bind the left event-time column in `FOR SYSTEM_TIME AS OF`.
361fn extract_temporal_left_time(
362    factor: &TableFactor,
363    sides: &JoinSides<'_>,
364) -> Result<Option<String>, ParseError> {
365    if let TableFactor::Table {
366        version: Some(TableVersion::ForSystemTimeAsOf(expr)),
367        ..
368    } = factor
369    {
370        let Expr::CompoundIdentifier(parts) = expr else {
371            return Err(ParseError::StreamingError(
372                "FOR SYSTEM_TIME AS OF requires a qualified left event-time column".into(),
373            ));
374        };
375        let [qualifier, column] = parts.as_slice() else {
376            return Err(ParseError::StreamingError(
377                "FOR SYSTEM_TIME AS OF requires a two-part left event-time column".into(),
378            ));
379        };
380        if sides.resolve_qualifier(&qualifier.value, "temporal AS OF timestamp")? != JoinSide::Left
381        {
382            return Err(ParseError::StreamingError(
383                "FOR SYSTEM_TIME AS OF must reference the left join input's event-time column"
384                    .into(),
385            ));
386        }
387        Ok(Some(column.value.clone()))
388    } else {
389        Ok(None)
390    }
391}
392
393/// Extract table alias from a TableFactor.
394fn extract_table_alias(factor: &TableFactor) -> Option<String> {
395    match factor {
396        TableFactor::Table { alias, .. } => alias.as_ref().map(|a| a.name.value.clone()),
397        TableFactor::Derived { alias, .. } => alias.as_ref().map(|a| a.name.value.clone()),
398        _ => None,
399    }
400}
401
402/// Map sqlparser `JoinOperator` to our `JoinType`.
403fn map_join_operator(op: &JoinOperator) -> Result<JoinType, ParseError> {
404    Ok(match op {
405        JoinOperator::Inner(_) | JoinOperator::Join(_) | JoinOperator::StraightJoin(_) => {
406            JoinType::Inner
407        }
408        JoinOperator::Left(_) | JoinOperator::LeftOuter(_) => JoinType::Left,
409        JoinOperator::LeftSemi(_) | JoinOperator::Semi(_) => JoinType::LeftSemi,
410        JoinOperator::LeftAnti(_) | JoinOperator::Anti(_) => JoinType::LeftAnti,
411        JoinOperator::AsOf { .. } => {
412            return Err(ParseError::StreamingError(
413                "ASOF JOIN is unsupported; use a bounded JOIN with an explicit event-time interval"
414                    .to_string(),
415            ));
416        }
417        JoinOperator::Right(_) | JoinOperator::RightOuter(_) => JoinType::Right,
418        JoinOperator::RightSemi(_) => JoinType::RightSemi,
419        JoinOperator::RightAnti(_) => JoinType::RightAnti,
420        JoinOperator::FullOuter(_) => JoinType::Full,
421        // CrossJoin, CrossApply, OuterApply are rejected by get_join_constraint()
422        _ => JoinType::Inner,
423    })
424}
425
426/// Analyze join constraint to extract key columns, additional key columns,
427/// time bound, and optional time column pair.
428#[allow(clippy::type_complexity)]
429fn analyze_join_constraint(
430    op: &JoinOperator,
431    sides: &JoinSides<'_>,
432) -> Result<
433    (
434        String,
435        String,
436        Vec<(String, String)>,
437        Option<Duration>,
438        Option<RawTimeCols>,
439    ),
440    ParseError,
441> {
442    let constraint = get_join_constraint(op)?;
443
444    match constraint {
445        JoinConstraint::On(expr) => {
446            let (key_pairs, time_bound, time_cols) = analyze_on_expression(expr, sides)?;
447            if key_pairs.is_empty() {
448                return Ok((String::new(), String::new(), vec![], time_bound, time_cols));
449            }
450            let (first_left, first_right) = key_pairs[0].clone();
451            let additional = key_pairs[1..].to_vec();
452            Ok((first_left, first_right, additional, time_bound, time_cols))
453        }
454        JoinConstraint::Using(cols) => {
455            if cols.is_empty() {
456                return Err(ParseError::StreamingError(
457                    "USING clause requires at least one column".to_string(),
458                ));
459            }
460            // First column is the primary key pair
461            let first_col = extract_using_column(&cols[0])?;
462            // Remaining columns are additional key pairs
463            let additional: Vec<(String, String)> = cols[1..]
464                .iter()
465                .map(|c| {
466                    let col = extract_using_column(c)?;
467                    Ok((col.clone(), col))
468                })
469                .collect::<Result<_, ParseError>>()?;
470            Ok((first_col.clone(), first_col, additional, None, None))
471        }
472        JoinConstraint::Natural => Err(ParseError::StreamingError(
473            "NATURAL JOIN not supported for streaming".to_string(),
474        )),
475        JoinConstraint::None => Err(ParseError::StreamingError(
476            "JOIN without condition not supported for streaming".to_string(),
477        )),
478    }
479}
480
481fn extract_using_column(name: &ObjectName) -> Result<String, ParseError> {
482    match name.0.as_slice() {
483        [ObjectNamePart::Identifier(ident)] => Ok(ident.value.clone()),
484        _ => Err(ParseError::StreamingError(
485            "streaming JOIN USING keys must be single column identifiers".to_string(),
486        )),
487    }
488}
489
490/// Get the JoinConstraint from a JoinOperator.
491fn get_join_constraint(op: &JoinOperator) -> Result<&JoinConstraint, ParseError> {
492    match op {
493        JoinOperator::Inner(constraint)
494        | JoinOperator::Join(constraint)
495        | JoinOperator::Left(constraint)
496        | JoinOperator::LeftOuter(constraint)
497        | JoinOperator::Right(constraint)
498        | JoinOperator::RightOuter(constraint)
499        | JoinOperator::FullOuter(constraint)
500        | JoinOperator::LeftSemi(constraint)
501        | JoinOperator::RightSemi(constraint)
502        | JoinOperator::LeftAnti(constraint)
503        | JoinOperator::RightAnti(constraint)
504        | JoinOperator::Semi(constraint)
505        | JoinOperator::Anti(constraint)
506        | JoinOperator::StraightJoin(constraint)
507        | JoinOperator::AsOf { constraint, .. } => Ok(constraint),
508        JoinOperator::CrossJoin(_) | JoinOperator::CrossApply | JoinOperator::OuterApply => Err(
509            ParseError::StreamingError("CROSS JOIN not supported for streaming".to_string()),
510        ),
511    }
512}
513
514/// Analyze ON expression to extract all key column pairs, time bound,
515/// and optional time column pair for stream-stream joins.
516#[allow(clippy::type_complexity)]
517fn analyze_on_expression(
518    expr: &Expr,
519    sides: &JoinSides<'_>,
520) -> Result<(Vec<(String, String)>, Option<Duration>, Option<RawTimeCols>), ParseError> {
521    // Handle compound expressions (AND)
522    match expr {
523        Expr::BinaryOp {
524            left,
525            op: BinaryOperator::And,
526            right,
527        } => {
528            let (mut key_pairs, left_bound, left_cols) = analyze_on_expression(left, sides)?;
529            let (right_keys, right_bound, right_cols) = analyze_on_expression(right, sides)?;
530
531            if left_bound.is_some() && right_bound.is_some() {
532                return Err(ParseError::StreamingError(
533                    "streaming interval joins require exactly one time-bound predicate".to_string(),
534                ));
535            }
536
537            key_pairs.extend(right_keys);
538            Ok((
539                key_pairs,
540                left_bound.or(right_bound),
541                left_cols.or(right_cols),
542            ))
543        }
544        // Equality condition: a.col = b.col
545        Expr::BinaryOp {
546            left,
547            op: BinaryOperator::Eq,
548            right,
549        } => Ok((
550            vec![orient_equality_columns(
551                left,
552                right,
553                sides,
554                "join equality",
555            )?],
556            None,
557            None,
558        )),
559        // BETWEEN clause for time bound: p.ts BETWEEN o.ts AND o.ts + INTERVAL
560        Expr::Between {
561            expr: between_expr,
562            negated,
563            low,
564            high,
565        } => {
566            if *negated {
567                return Err(ParseError::StreamingError(
568                    "NOT BETWEEN is not supported for streaming interval joins".to_string(),
569                ));
570            }
571
572            let (expr_qualifier, expr_col) = extract_qualified_column_ref(between_expr)
573                .ok_or_else(|| {
574                    ParseError::StreamingError(
575                        "streaming interval join timestamps must be qualified column references"
576                            .to_string(),
577                    )
578                })?;
579            let (low_qualifier, low_col) = extract_qualified_column_ref(low).ok_or_else(|| {
580                ParseError::StreamingError(
581                    "streaming interval join timestamps must be qualified column references"
582                        .to_string(),
583                )
584            })?;
585            let time_bound = extract_strict_interval_bound(high, &low_qualifier, &low_col)?;
586
587            Ok((
588                vec![],
589                Some(time_bound),
590                Some(RawTimeCols {
591                    expr_qualifier,
592                    expr_col,
593                    low_qualifier,
594                    low_col,
595                }),
596            ))
597        }
598        Expr::Nested(inner) => analyze_on_expression(inner, sides),
599        _ => Err(ParseError::StreamingError(format!(
600            "Unsupported join condition expression: {expr:?}"
601        ))),
602    }
603}
604
605fn extract_qualified_column_ref(expr: &Expr) -> Option<(String, String)> {
606    match strip_nested(expr) {
607        Expr::CompoundIdentifier(parts) if parts.len() == 2 => {
608            Some((parts[0].value.clone(), parts[1].value.clone()))
609        }
610        _ => None,
611    }
612}
613
614fn orient_equality_columns(
615    expression_left: &Expr,
616    expression_right: &Expr,
617    sides: &JoinSides<'_>,
618    context: &str,
619) -> Result<(String, String), ParseError> {
620    let (left_qualifier, left_column) =
621        extract_qualified_column_ref(expression_left).ok_or_else(|| {
622            ParseError::StreamingError(format!(
623                "Cannot extract column references from {context}; operands must be qualified column references"
624            ))
625        })?;
626    let (right_qualifier, right_column) = extract_qualified_column_ref(expression_right)
627        .ok_or_else(|| {
628            ParseError::StreamingError(format!(
629                "Cannot extract column references from {context}; operands must be qualified column references"
630            ))
631        })?;
632    let left_side = sides.resolve_qualifier(&left_qualifier, context)?;
633    let right_side = sides.resolve_qualifier(&right_qualifier, context)?;
634
635    match (left_side, right_side) {
636        (JoinSide::Left, JoinSide::Right) => Ok((left_column, right_column)),
637        (JoinSide::Right, JoinSide::Left) => Ok((right_column, left_column)),
638        _ => Err(ParseError::StreamingError(format!(
639            "{context} must compare one left-input column with one right-input column"
640        ))),
641    }
642}
643
644fn strip_nested(mut expr: &Expr) -> &Expr {
645    while let Expr::Nested(inner) = expr {
646        expr = inner;
647    }
648    expr
649}
650
651/// Parse the only admitted interval upper bound: the exact lower timestamp
652/// column plus a positive interval.
653fn extract_strict_interval_bound(
654    high: &Expr,
655    low_qualifier: &str,
656    low_col: &str,
657) -> Result<Duration, ParseError> {
658    let Expr::BinaryOp { left, op, right } = strip_nested(high) else {
659        return Err(ParseError::StreamingError(
660            "streaming interval join upper bound must be the left timestamp plus an interval"
661                .to_string(),
662        ));
663    };
664    if !matches!(op, BinaryOperator::Plus) {
665        return Err(ParseError::StreamingError(
666            "streaming interval join upper bound must use addition".to_string(),
667        ));
668    }
669
670    let Some((high_qualifier, high_col)) = extract_qualified_column_ref(left) else {
671        return Err(ParseError::StreamingError(
672            "streaming interval join upper bound must repeat the qualified lower timestamp"
673                .to_string(),
674        ));
675    };
676    if high_qualifier != low_qualifier || high_col != low_col {
677        return Err(ParseError::StreamingError(
678            "streaming interval join upper bound must use the same timestamp as its lower bound"
679                .to_string(),
680        ));
681    }
682
683    let interval = strip_nested(right);
684    if !matches!(interval, Expr::Interval(_)) {
685        return Err(ParseError::StreamingError(
686            "streaming interval join upper bound must end with an INTERVAL".to_string(),
687        ));
688    }
689    let duration = WindowRewriter::parse_interval_to_duration(interval)?;
690    if duration.is_zero() {
691        return Err(ParseError::StreamingError(
692            "streaming interval joins require a positive finite time bound".to_string(),
693        ));
694    }
695    Ok(duration)
696}
697
698/// Check if a SELECT contains a join.
699#[must_use]
700pub fn has_join(select: &Select) -> bool {
701    !select.from.is_empty() && !select.from[0].joins.is_empty()
702}
703
704/// Count the number of joins in a SELECT.
705#[must_use]
706pub fn count_joins(select: &Select) -> usize {
707    select
708        .from
709        .iter()
710        .map(|table_with_joins| table_with_joins.joins.len())
711        .sum()
712}
713
714/// Analysis result for multi-way JOINs (e.g., `A JOIN B ... JOIN C ...`).
715///
716/// Each step represents one left-deep join: step 0 joins the base table with
717/// the first right table, step 1 joins the result with the next right table, etc.
718#[derive(Debug, Clone)]
719pub struct MultiJoinAnalysis {
720    /// Ordered join steps (left-to-right)
721    pub joins: Vec<JoinAnalysis>,
722    /// All referenced tables in order (base table first, then each right table)
723    pub tables: Vec<String>,
724}
725
726impl MultiJoinAnalysis {
727    /// Number of join steps.
728    #[must_use]
729    pub fn len(&self) -> usize {
730        self.joins.len()
731    }
732
733    /// Whether there are no join steps.
734    #[must_use]
735    pub fn is_empty(&self) -> bool {
736        self.joins.is_empty()
737    }
738
739    /// Whether this is a single join (backward-compatible case).
740    #[must_use]
741    pub fn is_single(&self) -> bool {
742        self.joins.len() == 1
743    }
744
745    /// The first join step (convenience for single-join queries).
746    #[must_use]
747    pub fn first(&self) -> Option<&JoinAnalysis> {
748        self.joins.first()
749    }
750}
751
752/// Analyze a SELECT statement for all join steps (multi-way).
753///
754/// Returns `None` if the query has no joins. For a single join this
755/// returns a `MultiJoinAnalysis` with one step, making it backward
756/// compatible with `analyze_join()`.
757///
758/// # Errors
759///
760/// Returns `ParseError::StreamingError` if any join constraint is
761/// not supported or key columns cannot be extracted.
762pub fn analyze_joins(select: &Select) -> Result<Option<MultiJoinAnalysis>, ParseError> {
763    let from = &select.from;
764    if from.is_empty() {
765        return Ok(None);
766    }
767
768    let first_table = &from[0];
769    if first_table.joins.is_empty() {
770        return Ok(None);
771    }
772
773    // Extract base table
774    let base_table = extract_table_name(&first_table.relation)?;
775    let base_alias = extract_table_alias(&first_table.relation);
776
777    let mut join_steps = Vec::with_capacity(first_table.joins.len());
778    let mut tables = vec![base_table.clone()];
779
780    // Track the left table name for left-deep chaining
781    let mut prev_left_table = base_table;
782    let mut prev_left_alias = base_alias;
783
784    for join in &first_table.joins {
785        let right_table = extract_table_name(&join.relation)?;
786        let right_alias = extract_table_alias(&join.relation);
787        tables.push(right_table.clone());
788
789        let join_type = map_join_operator(&join.join_operator)?;
790        let sides = JoinSides {
791            left_table: &prev_left_table,
792            right_table: &right_table,
793            left_alias: prev_left_alias.as_deref(),
794            right_alias: right_alias.as_deref(),
795        };
796
797        if let Some(left_time_col) = extract_temporal_left_time(&join.relation, &sides)? {
798            // Temporal join: right side has FOR SYSTEM_TIME AS OF
799            let (left_key, right_key, additional, time_bound, time_cols) =
800                analyze_join_constraint(&join.join_operator, &sides)?;
801            if time_bound.is_some() || time_cols.is_some() {
802                return Err(ParseError::StreamingError(
803                    "temporal joins do not accept an additional time-bound predicate".into(),
804                ));
805            }
806
807            let mut analysis = JoinAnalysis::temporal(
808                prev_left_table.clone(),
809                right_table.clone(),
810                left_key,
811                right_key,
812                left_time_col,
813                join_type,
814            );
815            analysis.left_alias.clone_from(&prev_left_alias);
816            analysis.right_alias = right_alias;
817            analysis.additional_key_columns = additional;
818            join_steps.push(analysis);
819        } else {
820            // Regular join (inner, left, right, full)
821            let (left_key, right_key, additional, time_bound, time_cols) =
822                analyze_join_constraint(&join.join_operator, &sides)?;
823
824            let mut analysis = if let Some(tb) = time_bound {
825                JoinAnalysis::stream_stream(
826                    prev_left_table.clone(),
827                    right_table.clone(),
828                    left_key,
829                    right_key,
830                    tb,
831                    join_type,
832                )
833            } else {
834                JoinAnalysis::lookup(
835                    prev_left_table.clone(),
836                    right_table.clone(),
837                    left_key,
838                    right_key,
839                    join_type,
840                )
841            };
842            analysis.left_alias.clone_from(&prev_left_alias);
843            analysis.right_alias.clone_from(&right_alias);
844            analysis.additional_key_columns = additional;
845
846            if let Some(ref raw) = time_cols {
847                let (lt, rt) = resolve_time_cols(
848                    raw,
849                    &analysis.left_table,
850                    &analysis.right_table,
851                    prev_left_alias.as_deref(),
852                    right_alias.as_deref(),
853                )?;
854                analysis.left_time_column = Some(lt);
855                analysis.right_time_column = Some(rt);
856            }
857            join_steps.push(analysis);
858        }
859
860        // Next step's left table is this step's right table (left-deep)
861        prev_left_table = right_table;
862        prev_left_alias = extract_table_alias(&join.relation);
863    }
864
865    Ok(Some(MultiJoinAnalysis {
866        joins: join_steps,
867        tables,
868    }))
869}
870
871#[cfg(test)]
872mod tests;
laminar_sql/parser/join_parser/mod.rs

laminar_sql/parser/join_parser/
mod.rs
