nodedb_sql/planner/join/
plan.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! JOIN planning entry: builds left + right scans (or array-TVF arms),
4//! attaches projection/filters/aggregation.
5
6use sqlparser::ast::{self, Select};
7
8use super::array_arm;
9use super::constraint::extract_join_spec;
10use super::derived_side::plan_derived_side;
11use crate::error::{Result, SqlError};
12use crate::functions::registry::FunctionRegistry;
13use crate::planner::lateral::plan::{LateralJoinArgs, plan_lateral_join};
14use crate::planner::lateral::subquery::{
15    is_lateral_derived, lateral_alias_from_factor, subquery_from_factor,
16};
17use crate::resolver::columns::{ColumnNaming, TableScope};
18use crate::types::*;
19
20pub fn plan_join_from_select(
21    select: &Select,
22    scope: &TableScope,
23    catalog: &dyn SqlCatalog,
24    functions: &FunctionRegistry,
25    temporal: crate::TemporalScope,
26) -> Result<Option<SqlPlan>> {
27    let from = &select.from[0];
28
29    // Every clause below reads the join's merged rows, which key each field
30    // `qualifier.column`, so a column reference names its owning relation's
31    // qualifier. A LATERAL body reads its outer row through its own
32    // correlation, so a join with a LATERAL item keeps references as written.
33    let has_lateral = from
34        .joins
35        .iter()
36        .any(|join_item| is_lateral_derived(&join_item.relation));
37    let join_row_scope;
38    let scope = if has_lateral {
39        scope
40    } else {
41        join_row_scope = scope.clone().with_column_naming(ColumnNaming::JoinRow);
42        &join_row_scope
43    };
44
45    // Left side: an ARRAY_* TVF, a derived table, or a named table.
46    let left_plan =
47        if let Some(plan) = array_arm::try_plan_relation(&from.relation, catalog, temporal)? {
48            plan
49        } else if let Some(plan) =
50            plan_derived_side(&from.relation, scope, catalog, functions, temporal)?
51        {
52            plan
53        } else {
54            scan_for_relation(&from.relation, scope, temporal)?
55        };
56
57    let outer_alias = scan_alias_from_relation(&from.relation)?;
58
59    let mut current_plan = left_plan;
60
61    for join_item in &from.joins {
62        // Detect LATERAL derived subquery on the right side.
63        if is_lateral_derived(&join_item.relation) {
64            let lateral_alias =
65                lateral_alias_from_factor(&join_item.relation)?.ok_or_else(|| {
66                    SqlError::Unsupported {
67                        detail: "LATERAL subquery requires an alias (e.g. LATERAL (...) AS x)"
68                            .into(),
69                    }
70                })?;
71            let subquery = subquery_from_factor(&join_item.relation)
72                .expect("is_lateral_derived guarantees Derived variant");
73            let left_join = is_left_join_operator(&join_item.join_operator);
74            let projection = super::super::select::convert_projection(&select.projection, scope)?;
75            return Ok(Some(plan_lateral_join(LateralJoinArgs {
76                outer_plan: current_plan,
77                outer_alias,
78                subquery,
79                lateral_alias: &lateral_alias,
80                left_join,
81                outer_projection: projection,
82                outer_scope: scope,
83                catalog,
84                functions,
85                temporal,
86            })?));
87        }
88
89        // Right side: array TVF, derived table, or named table.
90        let right_plan = if let Some(plan) =
91            array_arm::try_plan_relation(&join_item.relation, catalog, temporal)?
92        {
93            plan
94        } else if let Some(plan) =
95            plan_derived_side(&join_item.relation, scope, catalog, functions, temporal)?
96        {
97            plan
98        } else {
99            scan_for_relation(&join_item.relation, scope, temporal)?
100        };
101
102        let right_ids = right_side_identifiers(&join_item.relation)?;
103        let (join_type, mut on_keys, condition) = extract_join_spec(
104            &join_item.join_operator,
105            scope,
106            right_ids.first().map(String::as_str),
107        )?;
108
109        // Orient equi-keys to FROM order: `on.0` must reference the left input
110        // and `on.1` the right input. The ON clause may write the operands in
111        // either order (`ON right.k = left.k`); the physical join builds the
112        // hash index on the right key, so a reversed pair would index the wrong
113        // side and match zero rows.
114        super::constraint::orient_keys_to_sides(&mut on_keys, &right_ids);
115
116        current_plan = SqlPlan::Join {
117            left: Box::new(current_plan),
118            right: Box::new(right_plan),
119            on: on_keys,
120            join_type,
121            condition,
122            limit: None,
123            projection: Vec::new(),
124            filters: Vec::new(),
125        };
126    }
127
128    let (subquery_joins, effective_where) = if let Some(expr) = &select.selection {
129        let extraction =
130            super::super::subquery::extract_subqueries(expr, scope, catalog, functions, temporal)?;
131        (extraction.joins, extraction.remaining_where)
132    } else {
133        (Vec::new(), None)
134    };
135
136    let projection = super::super::select::convert_projection(&select.projection, scope)?;
137    let filters = match &effective_where {
138        Some(expr) => super::super::select::convert_where_to_filters(expr, scope)?,
139        None => Vec::new(),
140    };
141
142    for sq in subquery_joins {
143        current_plan = SqlPlan::Join {
144            left: Box::new(current_plan),
145            right: Box::new(sq.inner_plan),
146            on: sq.on,
147            join_type: sq.join_type,
148            condition: None,
149            limit: None,
150            projection: Vec::new(),
151            filters: Vec::new(),
152        };
153    }
154
155    if super::super::select::has_aggregation(select, functions) {
156        // The WHERE predicates filter the joined rows before grouping.
157        if let SqlPlan::Join {
158            filters: ref mut filt,
159            ..
160        } = current_plan
161        {
162            *filt = filters;
163        }
164        // A join aggregate is always the generic aggregate, so ORDER BY
165        // need not be inspected for aggregates of its own.
166        let clauses = super::super::aggregate_clauses::plan_aggregate_clauses(
167            select,
168            None,
169            &[],
170            scope,
171            functions,
172        )?;
173        return Ok(Some(SqlPlan::Aggregate {
174            input: Box::new(current_plan),
175            group_by: clauses.group_by,
176            group_by_aliases: clauses.group_by_aliases,
177            output_order: clauses.output_order,
178            aggregates: clauses.aggregates,
179            having: clauses.having,
180            limit: super::super::aggregate::AGGREGATE_NO_LIMIT,
181            grouping_sets: clauses.grouping_sets,
182            sort_keys: Vec::new(),
183        }));
184    }
185
186    if let SqlPlan::Join {
187        projection: ref mut proj,
188        filters: ref mut filt,
189        ..
190    } = current_plan
191    {
192        *proj = projection;
193        *filt = filters;
194    }
195    Ok(Some(current_plan))
196}
197
198/// Collect the normalized identifiers (alias and/or table name) by which a
199/// right-side join relation can be referenced in an ON clause. Used to orient
200/// equi-keys so the right-side operand is always `on.1`.
201fn right_side_identifiers(factor: &ast::TableFactor) -> Result<Vec<String>> {
202    let mut ids = Vec::new();
203    if let Some((name, alias)) = crate::parser::normalize::table_name_from_factor(factor)? {
204        if let Some(alias) = alias {
205            ids.push(alias);
206        }
207        ids.push(name);
208    } else if let Some(alias) = derived_alias(factor)? {
209        ids.push(alias);
210    }
211    Ok(ids)
212}
213
214/// Extract an alias (or table name) from a named-table `TableFactor`, or the
215/// alias of a derived table.
216fn scan_alias_from_relation(factor: &ast::TableFactor) -> Result<Option<String>> {
217    match crate::parser::normalize::table_name_from_factor(factor)? {
218        Some((name, alias)) => Ok(Some(alias.unwrap_or(name))),
219        None => derived_alias(factor),
220    }
221}
222
223/// The alias of a derived-table `TableFactor`, `None` for any other factor.
224fn derived_alias(factor: &ast::TableFactor) -> Result<Option<String>> {
225    match factor {
226        ast::TableFactor::Derived {
227            alias: Some(alias), ..
228        } => crate::reserved::check_ast_identifier(&alias.name).map(Some),
229        _ => Ok(None),
230    }
231}
232
233/// True when the join operator represents a LEFT join variant.
234fn is_left_join_operator(op: &ast::JoinOperator) -> bool {
235    matches!(
236        op,
237        ast::JoinOperator::Left(_) | ast::JoinOperator::LeftOuter(_)
238    )
239}
240
241/// Build a `SqlPlan::Scan` for a named-table TableFactor.
242///
243/// The statement's `FOR SYSTEM_TIME` / `FOR VALID_TIME` scope applies to
244/// every joined relation, so each side must be bitemporal.
245fn scan_for_relation(
246    rel: &ast::TableFactor,
247    scope: &TableScope,
248    temporal: crate::TemporalScope,
249) -> Result<SqlPlan> {
250    let (rel_name, rel_alias) =
251        crate::parser::normalize::table_name_from_factor(rel)?.ok_or_else(|| {
252            SqlError::Unsupported {
253                detail: "non-table JOIN target".into(),
254            }
255        })?;
256    let table = scope
257        .tables
258        .values()
259        .find(|t| t.name == rel_name || t.alias.as_deref() == Some(&rel_name))
260        .ok_or_else(|| SqlError::UnknownTable {
261            name: rel_name.clone(),
262        })?;
263    temporal.require_bitemporal(&table.name, table.info.bitemporal)?;
264    Ok(SqlPlan::Scan {
265        collection: table.name.clone(),
266        alias: rel_alias.or_else(|| table.alias.clone()),
267        engine: table.info.engine,
268        filters: Vec::new(),
269        projection: Vec::new(),
270        sort_keys: Vec::new(),
271        limit: None,
272        offset: 0,
273        distinct: false,
274        window_functions: Vec::new(),
275        temporal,
276    })
277}
nodedb_sql/planner/join/plan.rs

nodedb_sql/planner/join/
plan.rs
