nodedb_sql/planner/lateral/
plan.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! LATERAL join planning: classify correlation shape and emit the appropriate
4//! `SqlPlan::LateralTopK` or `SqlPlan::LateralLoop` variant.
5
6use sqlparser::ast;
7
8use super::correlation::analyse_lateral_where;
9use super::subquery::{
10    extract_inner_alias, extract_inner_collection, inner_non_correlated_filters, limit_from_query,
11    reject_lateral_offset,
12};
13use crate::error::{Result, SqlError};
14use crate::functions::registry::FunctionRegistry;
15use crate::planner::sort::{sort_ascending, sort_nulls_first};
16use crate::resolver::ColumnScope;
17use crate::resolver::columns::{ResolvedTable, TableScope};
18use crate::resolver::expr::convert_expr;
19use crate::temporal::TemporalScope;
20use crate::types::*;
21
22/// The maximum outer-row count allowed for `LateralLoop` queries.
23pub const LATERAL_LOOP_CAP: usize = 100_000;
24
25/// Parameters for [`plan_lateral_join`].
26pub struct LateralJoinArgs<'a> {
27    /// Plan for the driving (outer) side.
28    pub outer_plan: SqlPlan,
29    /// Alias or name of the outer table for correlation detection.
30    pub outer_alias: Option<String>,
31    /// The LATERAL inner subquery.
32    pub subquery: &'a ast::Query,
33    /// Alias given to the LATERAL in the SQL (e.g. `x` in `LATERAL (...) x`).
34    pub lateral_alias: &'a str,
35    /// True when the enclosing join is LEFT JOIN LATERAL (outer rows
36    /// preserved when inner produces no rows).
37    pub left_join: bool,
38    /// SELECT list projection to apply after the lateral.
39    pub outer_projection: Vec<Projection>,
40    /// The enclosing query's column namespace. The inner body nests inside it
41    /// so a correlated reference to an outer relation resolves.
42    pub outer_scope: &'a TableScope,
43    pub catalog: &'a dyn SqlCatalog,
44    pub functions: &'a FunctionRegistry,
45    pub temporal: TemporalScope,
46}
47
48/// Plan a LATERAL subquery join.
49///
50/// Called when the right side of a JOIN (or a comma-separated FROM item) is a
51/// `TableFactor::Derived { lateral: true, .. }`.
52pub fn plan_lateral_join(args: LateralJoinArgs<'_>) -> Result<SqlPlan> {
53    let LateralJoinArgs {
54        outer_plan,
55        outer_alias,
56        subquery,
57        lateral_alias,
58        left_join,
59        outer_projection,
60        outer_scope,
61        catalog,
62        functions,
63        temporal,
64    } = args;
65    let select = match subquery.body.as_ref() {
66        sqlparser::ast::SetExpr::Select(s) => s,
67        _ => {
68            return Err(SqlError::Unsupported {
69                detail: "LATERAL subquery body must be a SELECT".into(),
70            });
71        }
72    };
73
74    let outer_alias_str = outer_alias.as_deref().unwrap_or("").to_string();
75
76    let analysis = analyse_lateral_where(subquery, &outer_alias_str);
77
78    // The outer operand of a correlation predicate is stripped from the inner
79    // WHERE before conversion, so it is checked here or nowhere.
80    let outer_qualifier = outer_alias.as_deref();
81    for key in &analysis.equi_keys {
82        outer_scope.check_name(outer_qualifier, &key.outer_col)?;
83    }
84    for (_, outer_col) in &analysis.non_equi {
85        outer_scope.check_name(outer_qualifier, outer_col)?;
86    }
87
88    // Determine if this is the equi-correlated + TopK shape:
89    //   - At least one equi-key correlation.
90    //   - A LIMIT k on the subquery.
91    //   - No non-equi correlations (those require LateralLoop).
92    let has_equi = !analysis.equi_keys.is_empty();
93    let inner_limit = limit_from_query(subquery)?;
94    reject_lateral_offset(subquery)?;
95    if let Some(inner_limit) = inner_limit.filter(|_| has_equi && analysis.non_equi.is_empty()) {
96        plan_lateral_top_k(LateralTopKPlanArgs {
97            outer_plan,
98            outer_alias,
99            select,
100            subquery,
101            equi_keys: analysis.equi_keys,
102            inner_limit,
103            lateral_alias,
104            left_join,
105            outer_projection,
106            outer_scope,
107            catalog,
108            functions,
109            temporal,
110        })
111    } else if has_equi && analysis.non_equi.is_empty() {
112        // Equi-correlated, no LIMIT: rewrite as a regular hash join.
113        //
114        // The equi correlations become the join keys. The inner side is a bare
115        // scan of the inner collection carrying only the residual (equi-stripped)
116        // WHERE; the outer alias never leaks into inner name resolution. The join
117        // executor scans the inner collection by name, so the join key columns
118        // are available on the merged rows.
119        let inner_plan = build_inner_scan(InnerScanArgs {
120            select,
121            residual_where: analysis.remaining,
122            lateral_alias,
123            catalog,
124            temporal,
125            outer_scope,
126        })?;
127        let equi_on: Vec<(String, String)> = analysis
128            .equi_keys
129            .into_iter()
130            .map(|c| (c.outer_col, c.inner_col))
131            .collect();
132        Ok(SqlPlan::Join {
133            left: Box::new(outer_plan),
134            right: Box::new(inner_plan),
135            on: equi_on,
136            join_type: if left_join {
137                JoinType::Left
138            } else {
139                JoinType::Inner
140            },
141            condition: None,
142            limit: None,
143            projection: outer_projection,
144            filters: Vec::new(),
145        })
146    } else {
147        // General correlation — LateralLoop.
148        //
149        // The inner side is a bare scan of the inner collection carrying the
150        // residual WHERE (`analysis.remaining`): non-correlated predicates plus
151        // any non-equi correlated predicates (e.g. `e.log_time > u.created_at`).
152        // The latter lower to `*Column` scan filters (`GtColumn`, ...) whose
153        // outer operand is bound per outer row by the Data Plane executor via
154        // `bind_outer_values`. Equi correlations are applied separately through
155        // `correlation_predicates`, so they are excluded from `remaining` and
156        // not duplicated here. The subquery is not routed through `plan_query`
157        // because its WHERE references the outer alias, which is not resolvable
158        // in the inner FROM scope.
159        let inner_plan = build_inner_scan(InnerScanArgs {
160            select,
161            residual_where: analysis.remaining,
162            lateral_alias,
163            catalog,
164            temporal,
165            outer_scope,
166        })?;
167        let correlation_predicates: Vec<(String, String)> = analysis
168            .equi_keys
169            .iter()
170            .map(|c| (c.inner_col.clone(), c.outer_col.clone()))
171            .collect();
172        Ok(SqlPlan::LateralLoop(LateralLoopPlan {
173            outer: Box::new(outer_plan),
174            outer_alias,
175            inner: Box::new(inner_plan),
176            correlation_predicates,
177            lateral_alias: lateral_alias.to_string(),
178            projection: outer_projection,
179            outer_row_cap: LATERAL_LOOP_CAP,
180            left_join,
181        }))
182    }
183}
184
185/// Parameters for [`build_inner_scan`].
186struct InnerScanArgs<'a> {
187    select: &'a sqlparser::ast::Select,
188    residual_where: Option<ast::Expr>,
189    lateral_alias: &'a str,
190    catalog: &'a dyn SqlCatalog,
191    temporal: TemporalScope,
192    outer_scope: &'a TableScope,
193}
194
195/// Build the inner `SqlPlan::Scan` for a LATERAL join.
196///
197/// The scan carries the residual WHERE as filters. Correlated non-equi
198/// predicates survive as column-vs-column comparisons and lower to
199/// runtime-bound `*Column` filters downstream; the executor binds the outer
200/// operand per outer row.
201fn build_inner_scan(args: InnerScanArgs<'_>) -> Result<SqlPlan> {
202    let InnerScanArgs {
203        select,
204        residual_where,
205        lateral_alias,
206        catalog,
207        temporal,
208        outer_scope,
209    } = args;
210    let inner_collection = extract_inner_collection(select)?;
211    let inner_alias = extract_inner_alias(select)?;
212    let inner_info = catalog
213        .resolve_relation(nodedb_types::DatabaseId::DEFAULT, &inner_collection)?
214        .ok_or_else(|| SqlError::UnknownTable {
215            name: inner_collection.clone(),
216        })?;
217    temporal.require_bitemporal(&inner_collection, inner_info.bitemporal)?;
218    // A correlated residual predicate names the outer relation, so the inner
219    // scope nests inside it.
220    let inner_scope = TableScope::single(ResolvedTable {
221        name: inner_collection.clone(),
222        alias: inner_alias,
223        info: inner_info.clone(),
224    })?
225    .nested_in(outer_scope.clone());
226    let filters = match &residual_where {
227        Some(expr) => crate::planner::select::convert_where_to_filters(expr, &inner_scope)?,
228        None => Vec::new(),
229    };
230    // The lateral subquery's output relation is named by the LATERAL alias, and
231    // the inner table alias is private to the subquery. Downstream the scan
232    // alias qualifies the inner columns on a merged join row, so it carries the
233    // LATERAL alias and `x.a` resolves the way `LateralLoop` and `LateralTopK`
234    // already name their inner columns.
235    Ok(SqlPlan::Scan {
236        collection: inner_collection,
237        alias: Some(lateral_alias.to_string()),
238        engine: inner_info.engine,
239        filters,
240        projection: Vec::new(),
241        sort_keys: Vec::new(),
242        limit: None,
243        offset: 0,
244        distinct: false,
245        window_functions: Vec::new(),
246        temporal,
247    })
248}
249
250/// Parameters for [`plan_lateral_top_k`].
251struct LateralTopKPlanArgs<'a> {
252    outer_plan: SqlPlan,
253    outer_alias: Option<String>,
254    select: &'a sqlparser::ast::Select,
255    subquery: &'a ast::Query,
256    equi_keys: Vec<super::correlation::CorrelationEq>,
257    inner_limit: usize,
258    lateral_alias: &'a str,
259    left_join: bool,
260    outer_projection: Vec<Projection>,
261    outer_scope: &'a TableScope,
262    catalog: &'a dyn SqlCatalog,
263    functions: &'a FunctionRegistry,
264    temporal: TemporalScope,
265}
266
267/// Plan the `LateralTopK` variant: equi-correlated + ORDER BY + LIMIT k.
268fn plan_lateral_top_k(args: LateralTopKPlanArgs<'_>) -> Result<SqlPlan> {
269    let LateralTopKPlanArgs {
270        outer_plan,
271        outer_alias,
272        select,
273        subquery,
274        equi_keys,
275        inner_limit,
276        lateral_alias,
277        left_join,
278        outer_projection,
279        outer_scope,
280        catalog,
281        functions,
282        temporal,
283    } = args;
284    // Build a bare inner Scan without correlation filters (those are injected
285    // at runtime per outer row).
286    let inner_collection = extract_inner_collection(select)?;
287    catalog
288        .resolve_relation(nodedb_types::DatabaseId::DEFAULT, &inner_collection)?
289        .ok_or_else(|| SqlError::UnknownTable {
290            name: inner_collection.clone(),
291        })?;
292    // The Top-K plan does not retain the inner alias, but it must still reject
293    // malformed aliases before expressions referencing them are lowered.
294    let _inner_alias = extract_inner_alias(select)?;
295    let inner_scope = TableScope::resolve_from(catalog, functions, temporal, &select.from)?
296        .nested_in(outer_scope.clone());
297    let inner_filters =
298        inner_non_correlated_filters(select, outer_alias.as_deref().unwrap_or(""), &inner_scope)?;
299
300    // Extract ORDER BY from the inner subquery.
301    // For LATERAL inner scans we only need simple column-expression sort keys;
302    // the full search-trigger machinery (vector/hybrid search) is not applicable
303    // here, so we convert expressions directly.
304    let inner_order_by = if let Some(order_by) = &subquery.order_by {
305        match &order_by.kind {
306            ast::OrderByKind::Expressions(exprs) => exprs
307                .iter()
308                .map(|o| {
309                    Ok(SortKey {
310                        expr: convert_expr(&o.expr, &ColumnScope::Relations(&inner_scope))?,
311                        ascending: sort_ascending(&o.options)?,
312                        nulls_first: sort_nulls_first(&o.options)?,
313                    })
314                })
315                .collect::<Result<Vec<_>>>()?,
316            ast::OrderByKind::All(_) => Vec::new(),
317        }
318    } else {
319        Vec::new()
320    };
321
322    let correlation_keys: Vec<(String, String)> = equi_keys
323        .into_iter()
324        .map(|c| (c.outer_col, c.inner_col))
325        .collect();
326
327    Ok(SqlPlan::LateralTopK(LateralTopKPlan {
328        outer: Box::new(outer_plan),
329        outer_alias,
330        inner_collection,
331        inner_filters,
332        inner_order_by,
333        inner_limit,
334        correlation_keys,
335        lateral_alias: lateral_alias.to_string(),
336        projection: outer_projection,
337        left_join,
338    }))
339}
nodedb_sql/planner/lateral/plan.rs

nodedb_sql/planner/lateral/
plan.rs
