Skip to main content

powdb_query/
planner.rs

1use crate::ast::*;
2use crate::parser::{parse, ParseError};
3use crate::plan::*;
4use powdb_storage::stored_json_path::StoredJsonPathV1;
5
6#[derive(Debug, Clone, PartialEq, Eq)]
7enum RangeTarget {
8    Column(String),
9    JsonPath(StoredJsonPathV1),
10}
11
12/// (target, lower_bound, upper_bound) — used by range-index extraction.
13type RangeBound = (RangeTarget, Option<(Expr, bool)>, Option<(Expr, bool)>);
14
15/// Plan-phase error — wraps ParseError for the full lex→parse→plan chain.
16#[derive(Debug)]
17pub enum PlanError {
18    /// Error originated in the parser (or lexer, via ParseError::Lex).
19    Parse(ParseError),
20    /// The parsed query is structurally valid but cannot be planned safely.
21    Semantic(String),
22}
23
24impl PlanError {
25    /// Convenience: human-readable message for any variant.
26    pub fn message(&self) -> String {
27        self.to_string()
28    }
29}
30
31impl std::fmt::Display for PlanError {
32    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
33        match self {
34            Self::Parse(e) => write!(f, "{e}"),
35            Self::Semantic(message) => write!(f, "{message}"),
36        }
37    }
38}
39
40impl std::error::Error for PlanError {
41    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
42        match self {
43            Self::Parse(e) => Some(e),
44            Self::Semantic(_) => None,
45        }
46    }
47}
48
49impl From<ParseError> for PlanError {
50    fn from(e: ParseError) -> Self {
51        PlanError::Parse(e)
52    }
53}
54
55pub fn plan(input: &str) -> Result<PlanNode, PlanError> {
56    let stmt = parse(input)?;
57    plan_statement(stmt)
58}
59
60pub fn plan_statement(stmt: Statement) -> Result<PlanNode, PlanError> {
61    match stmt {
62        Statement::Query(q) => plan_query(q),
63        Statement::Insert(ins) => plan_insert(ins),
64        Statement::UpdateQuery(upd) => plan_update(upd),
65        Statement::DeleteQuery(del) => plan_delete(del),
66        Statement::CreateType(ct) => plan_create_type(ct),
67        Statement::AlterTable(at) => Ok(PlanNode::AlterTable {
68            table: at.table,
69            action: at.action,
70        }),
71        Statement::DropTable(dt) => Ok(PlanNode::DropTable {
72            name: dt.table,
73            if_exists: dt.if_exists,
74        }),
75        Statement::CreateView(cv) => Ok(PlanNode::CreateView {
76            name: cv.name,
77            query_text: cv.query_text,
78        }),
79        Statement::RefreshView(rv) => Ok(PlanNode::RefreshView { name: rv.name }),
80        Statement::DropView(dv) => Ok(PlanNode::DropView {
81            name: dv.name,
82            if_exists: dv.if_exists,
83        }),
84        Statement::ListTypes => Ok(PlanNode::ListTypes),
85        Statement::Describe(table) => Ok(PlanNode::Describe { table }),
86        Statement::Union(u) => {
87            let left = plan_statement(*u.left)?;
88            let right = plan_statement(*u.right)?;
89            Ok(PlanNode::Union {
90                left: Box::new(left),
91                right: Box::new(right),
92                all: u.all,
93            })
94        }
95        Statement::Upsert(ups) => plan_upsert(ups),
96        Statement::Begin => Ok(PlanNode::Begin),
97        Statement::Commit => Ok(PlanNode::Commit),
98        Statement::Rollback => Ok(PlanNode::Rollback),
99        Statement::Explain(inner) => {
100            let inner_plan = plan_statement(*inner)?;
101            Ok(PlanNode::Explain {
102                input: Box::new(inner_plan),
103            })
104        }
105    }
106}
107
108fn plan_query(mut q: QueryExpr) -> Result<PlanNode, PlanError> {
109    // Mission E1.2: if the query has joins, build a left-deep nested-loop
110    // plan. Correctness first — hash-join optimization is E1.3. We also
111    // don't try to fold an IndexScan under a joined query yet (the
112    // leaf-level fast paths all match on `PlanNode::SeqScan { .. }`
113    // literally, so mixing them into a join plan would silently break).
114    if !q.joins.is_empty() {
115        return plan_joined_query(q);
116    }
117    let source_aliases = std::collections::HashSet::from([q.source.clone()]);
118    // Try to fold `filter .col = literal` into an IndexScan. The executor
119    // decides at run time whether the column actually has an index — if not,
120    // it transparently falls back to a sequential scan with the same predicate,
121    // so this rewrite is always safe.
122    //
123    // We only rewrite the *simple* eq case: `filter .col = literal`. Conjunctions
124    // like `filter .col = 1 and .other > 5` fall through to SeqScan + Filter.
125    // Extending this to split conjunctions is a future optimization.
126    let ordered_expr_scan = try_extract_ordered_expr_index_scan(&q);
127    let (source, filter) = if let Some(scan) = ordered_expr_scan {
128        // The ordered expression node owns these clauses and executes them in
129        // index order. Clear them so the generic pipeline does not wrap a
130        // second Sort/Offset/Limit around the speculative node.
131        q.order = None;
132        q.limit = None;
133        q.offset = None;
134        (scan, None)
135    } else {
136        match q.filter {
137            Some(pred) => match try_extract_eq_index_key(&q.source, &pred) {
138                Some(index_scan) => (index_scan, None),
139                None => match try_extract_range_index_keys(&q.source, &pred) {
140                    Some(range_scan) => (range_scan, None),
141                    None => (
142                        PlanNode::SeqScan {
143                            table: q.source.clone(),
144                        },
145                        Some(pred),
146                    ),
147                },
148            },
149            None => (
150                PlanNode::SeqScan {
151                    table: q.source.clone(),
152                },
153                None,
154            ),
155        }
156    };
157    let mut node = source;
158
159    if let Some(pred) = filter {
160        node = PlanNode::Filter {
161            input: Box::new(node),
162            predicate: pred,
163        };
164    }
165
166    // Mission E2b: GROUP BY path — insert GroupBy + Project before
167    // order/limit/offset/distinct.
168    if let Some(group) = q.group_by {
169        let mut grouped_order = q.order;
170        let mut proj_fields: Vec<ProjectField> = q
171            .projection
172            .map(|proj| {
173                proj.into_iter()
174                    .map(|pf| ProjectField {
175                        alias: pf.alias,
176                        expr: pf.expr,
177                    })
178                    .collect()
179            })
180            .unwrap_or_default();
181        let mut having = group.having;
182        let aggregates = extract_aggregates(&mut proj_fields, &mut having, &source_aliases)?;
183        rewrite_group_order_keys(grouped_order.as_mut(), &proj_fields, &group.keys);
184        rewrite_group_key_references(&mut proj_fields, &mut having, &group.keys);
185
186        node = PlanNode::GroupBy {
187            input: Box::new(node),
188            keys: group.keys,
189            aggregates,
190            having,
191        };
192
193        if !proj_fields.is_empty() {
194            node = PlanNode::Project {
195                input: Box::new(node),
196                fields: proj_fields,
197            };
198        }
199
200        if let Some(order) = grouped_order {
201            node = PlanNode::Sort {
202                input: Box::new(node),
203                keys: order
204                    .keys
205                    .into_iter()
206                    .map(|k| SortKey {
207                        expr: k.expr,
208                        descending: k.descending,
209                    })
210                    .collect(),
211            };
212        }
213        // Offset must be applied *before* Limit: skip M rows, then take N.
214        // Plan shape is Limit(Offset(...)), so Offset is built first (inner)
215        // and Limit wraps it (outer).
216        if let Some(off) = q.offset {
217            node = PlanNode::Offset {
218                input: Box::new(node),
219                count: off,
220            };
221        }
222        if let Some(lim) = q.limit {
223            node = PlanNode::Limit {
224                input: Box::new(node),
225                count: lim,
226            };
227        }
228        if q.distinct {
229            node = PlanNode::Distinct {
230                input: Box::new(node),
231            };
232        }
233        return Ok(node);
234    }
235
236    if let Some(order) = q.order {
237        node = PlanNode::Sort {
238            input: Box::new(node),
239            keys: order
240                .keys
241                .into_iter()
242                .map(|k| SortKey {
243                    expr: k.expr,
244                    descending: k.descending,
245                })
246                .collect(),
247        };
248    }
249
250    // Offset must be applied *before* Limit: skip M rows, then take N.
251    // Plan shape is Limit(Offset(...)), so Offset is built first (inner)
252    // and Limit wraps it (outer).
253    if let Some(off) = q.offset {
254        node = PlanNode::Offset {
255            input: Box::new(node),
256            count: off,
257        };
258    }
259
260    if let Some(lim) = q.limit {
261        node = PlanNode::Limit {
262            input: Box::new(node),
263            count: lim,
264        };
265    }
266
267    if let Some(proj) = q.projection {
268        let mut fields: Vec<ProjectField> = proj
269            .into_iter()
270            .map(|pf| ProjectField {
271                alias: pf.alias,
272                expr: pf.expr,
273            })
274            .collect();
275        let windows = extract_windows(&mut fields);
276        if !windows.is_empty() {
277            node = PlanNode::Window {
278                input: Box::new(node),
279                windows,
280            };
281        }
282        node = PlanNode::Project {
283            input: Box::new(node),
284            fields,
285        };
286    }
287
288    if q.distinct {
289        node = PlanNode::Distinct {
290            input: Box::new(node),
291        };
292    }
293
294    if let Some(agg) = q.aggregation {
295        let provenance_alias = symmetric_provenance_alias(
296            agg.function,
297            agg.argument.as_ref(),
298            agg.mode,
299            &source_aliases,
300        )?;
301        node = PlanNode::Aggregate {
302            input: Box::new(node),
303            function: agg.function,
304            argument: agg.argument,
305            mode: agg.mode,
306            provenance_alias,
307        };
308    }
309
310    Ok(node)
311}
312
313/// Build a left-deep nested-loop join plan for a query with 1+ join clauses.
314///
315/// The plan shape for `T1 as a [inner|left|cross] join T2 as b on <pred> ...` is:
316///
317///   Project? (optional, from q.projection)
318///   └─ Offset? / Limit? / Sort?
319///      └─ Filter? (the top-level q.filter, using qualified columns)
320///         └─ NestedLoopJoin { kind, on }
321///            ├─ AliasScan { T1, a }
322///            └─ AliasScan { T2, b }
323///
324/// Multi-join chains extend left-deep: a third join adds a second
325/// `NestedLoopJoin` on top, with the first join's output as its `left`.
326///
327/// Aliases default to the source table name when the query didn't write
328/// `as <name>` explicitly — that way users can always write `T.field`
329/// without being forced to alias every source.
330///
331/// RightOuter is rewritten into LeftOuter with inputs swapped — the two
332/// differ only in which side survives non-matching rows, and swapping
333/// inputs lets the executor ship a single LeftOuter path.
334fn plan_joined_query(mut q: QueryExpr) -> Result<PlanNode, PlanError> {
335    let primary_alias = q.alias.clone().unwrap_or_else(|| q.source.clone());
336    let mut aliases = std::collections::HashSet::new();
337    aliases.insert(primary_alias.clone());
338    let mut node = PlanNode::AliasScan {
339        table: q.source.clone(),
340        alias: primary_alias,
341    };
342
343    for join in q.joins {
344        let right_alias = join.alias.unwrap_or_else(|| join.source.clone());
345        if !aliases.insert(right_alias.clone()) {
346            return Err(ParseError::Syntax {
347                message: format!(
348                    "duplicate source alias `{right_alias}` in join; every joined source needs a unique alias"
349                ),
350            }
351            .into());
352        }
353        let right = PlanNode::AliasScan {
354            table: join.source,
355            alias: right_alias,
356        };
357        match join.kind {
358            JoinKind::Inner | JoinKind::LeftOuter | JoinKind::Cross => {
359                node = PlanNode::NestedLoopJoin {
360                    left: Box::new(node),
361                    right: Box::new(right),
362                    on: join.on,
363                    kind: join.kind,
364                };
365            }
366            JoinKind::RightOuter => {
367                // `a RIGHT OUTER JOIN b ON <p>` ≡ `b LEFT OUTER JOIN a ON <p>`.
368                node = PlanNode::NestedLoopJoin {
369                    left: Box::new(right),
370                    right: Box::new(node),
371                    on: join.on,
372                    kind: JoinKind::LeftOuter,
373                };
374            }
375        }
376    }
377
378    if let Some(pred) = q.filter {
379        node = PlanNode::Filter {
380            input: Box::new(node),
381            predicate: pred,
382        };
383    }
384
385    if q.group_by.is_none() {
386        if let Some(order) = q.order.take() {
387            node = PlanNode::Sort {
388                input: Box::new(node),
389                keys: order
390                    .keys
391                    .into_iter()
392                    .map(|k| SortKey {
393                        expr: k.expr,
394                        descending: k.descending,
395                    })
396                    .collect(),
397            };
398        }
399    }
400
401    // Mission E2b: GROUP BY path for joined queries.
402    if let Some(group) = q.group_by {
403        let mut grouped_order = q.order;
404        let mut proj_fields: Vec<ProjectField> = q
405            .projection
406            .map(|proj| {
407                proj.into_iter()
408                    .map(|pf| ProjectField {
409                        alias: pf.alias,
410                        expr: pf.expr,
411                    })
412                    .collect()
413            })
414            .unwrap_or_default();
415        let mut having = group.having;
416        let aggregates = extract_aggregates(&mut proj_fields, &mut having, &aliases)?;
417        rewrite_group_order_keys(grouped_order.as_mut(), &proj_fields, &group.keys);
418        rewrite_group_key_references(&mut proj_fields, &mut having, &group.keys);
419
420        node = PlanNode::GroupBy {
421            input: Box::new(node),
422            keys: group.keys,
423            aggregates,
424            having,
425        };
426
427        if !proj_fields.is_empty() {
428            node = PlanNode::Project {
429                input: Box::new(node),
430                fields: proj_fields,
431            };
432        }
433        if let Some(order) = grouped_order {
434            node = PlanNode::Sort {
435                input: Box::new(node),
436                keys: order
437                    .keys
438                    .into_iter()
439                    .map(|key| SortKey {
440                        expr: key.expr,
441                        descending: key.descending,
442                    })
443                    .collect(),
444            };
445        }
446        // LIMIT/OFFSET operate on grouped result rows, never on the joined
447        // input. Applying either before GroupBy truncates source rows and can
448        // silently change aggregate values. Offset remains inside Limit so
449        // execution skips M grouped rows before taking N.
450        if let Some(off) = q.offset {
451            node = PlanNode::Offset {
452                input: Box::new(node),
453                count: off,
454            };
455        }
456        if let Some(lim) = q.limit {
457            node = PlanNode::Limit {
458                input: Box::new(node),
459                count: lim,
460            };
461        }
462        if q.distinct {
463            node = PlanNode::Distinct {
464                input: Box::new(node),
465            };
466        }
467        return Ok(node);
468    }
469
470    // Offset must be applied *before* Limit: skip M rows, then take N.
471    // Plan shape is Limit(Offset(...)), so Offset is built first (inner)
472    // and Limit wraps it (outer).
473    if let Some(off) = q.offset {
474        node = PlanNode::Offset {
475            input: Box::new(node),
476            count: off,
477        };
478    }
479
480    if let Some(lim) = q.limit {
481        node = PlanNode::Limit {
482            input: Box::new(node),
483            count: lim,
484        };
485    }
486
487    if let Some(proj) = q.projection {
488        let mut fields: Vec<ProjectField> = proj
489            .into_iter()
490            .map(|pf| ProjectField {
491                alias: pf.alias,
492                expr: pf.expr,
493            })
494            .collect();
495        let windows = extract_windows(&mut fields);
496        if !windows.is_empty() {
497            node = PlanNode::Window {
498                input: Box::new(node),
499                windows,
500            };
501        }
502        node = PlanNode::Project {
503            input: Box::new(node),
504            fields,
505        };
506    }
507
508    if q.distinct {
509        node = PlanNode::Distinct {
510            input: Box::new(node),
511        };
512    }
513
514    if let Some(agg) = q.aggregation {
515        let provenance_alias =
516            symmetric_provenance_alias(agg.function, agg.argument.as_ref(), agg.mode, &aliases)?;
517        node = PlanNode::Aggregate {
518            input: Box::new(node),
519            function: agg.function,
520            argument: agg.argument,
521            mode: agg.mode,
522            provenance_alias,
523        };
524    }
525
526    Ok(node)
527}
528
529fn plan_insert(ins: InsertExpr) -> Result<PlanNode, PlanError> {
530    Ok(PlanNode::Insert {
531        table: ins.target,
532        rows: ins.rows,
533        returning: ins.returning,
534    })
535}
536
537fn plan_update(upd: UpdateExpr) -> Result<PlanNode, PlanError> {
538    // Mirror the read-side IndexScan fold: when the update filter is a simple
539    // `.col = literal`, emit `Update(IndexScan)` so the executor's index-lookup
540    // mutation fast path fires. The executor falls back to a scan if the
541    // column happens to lack an index, so this is always safe.
542    let source = match upd.filter {
543        Some(pred) => match try_extract_eq_index_key(&upd.source, &pred) {
544            Some(index_scan) => index_scan,
545            None => match try_extract_range_index_keys(&upd.source, &pred) {
546                Some(range_scan) => range_scan,
547                None => PlanNode::Filter {
548                    input: Box::new(PlanNode::SeqScan {
549                        table: upd.source.clone(),
550                    }),
551                    predicate: pred,
552                },
553            },
554        },
555        None => PlanNode::SeqScan {
556            table: upd.source.clone(),
557        },
558    };
559    Ok(PlanNode::Update {
560        input: Box::new(source),
561        table: upd.source,
562        assignments: upd.assignments,
563        returning: upd.returning,
564    })
565}
566
567fn plan_delete(del: DeleteExpr) -> Result<PlanNode, PlanError> {
568    let source = match del.filter {
569        Some(pred) => match try_extract_eq_index_key(&del.source, &pred) {
570            Some(index_scan) => index_scan,
571            None => match try_extract_range_index_keys(&del.source, &pred) {
572                Some(range_scan) => range_scan,
573                None => PlanNode::Filter {
574                    input: Box::new(PlanNode::SeqScan {
575                        table: del.source.clone(),
576                    }),
577                    predicate: pred,
578                },
579            },
580        },
581        None => PlanNode::SeqScan {
582            table: del.source.clone(),
583        },
584    };
585    Ok(PlanNode::Delete {
586        input: Box::new(source),
587        table: del.source,
588        returning: del.returning,
589    })
590}
591
592fn plan_upsert(ups: UpsertExpr) -> Result<PlanNode, PlanError> {
593    Ok(PlanNode::Upsert {
594        table: ups.target,
595        key_column: ups.key_column,
596        assignments: ups.assignments,
597        on_conflict: ups.on_conflict,
598    })
599}
600
601fn plan_create_type(ct: CreateTypeExpr) -> Result<PlanNode, PlanError> {
602    let fields = ct
603        .fields
604        .into_iter()
605        .map(|f| crate::plan::CreateField {
606            name: f.name,
607            type_name: f.type_name,
608            required: f.required,
609            unique: f.unique,
610            default: f.default,
611            auto: f.auto,
612        })
613        .collect();
614    Ok(PlanNode::CreateTable {
615        name: ct.name,
616        fields,
617        if_not_exists: ct.if_not_exists,
618    })
619}
620
621/// If the predicate is a simple `.field = literal` (or `literal = .field`),
622/// return a corresponding IndexScan plan node. Otherwise return None so the
623/// caller can fall through to SeqScan + Filter.
624///
625/// The executor decides at run time whether the named column actually has a
626/// B-tree index — if not, IndexScan transparently falls back to a scan +
627/// equality filter on that column. That means this rewrite is always safe
628/// regardless of schema/index state; it just unlocks the fast path when an
629/// index happens to exist.
630fn try_extract_eq_index_key(table: &str, pred: &Expr) -> Option<PlanNode> {
631    let (lhs, op, rhs) = match pred {
632        Expr::BinaryOp(lhs, op, rhs) => (lhs.as_ref(), *op, rhs.as_ref()),
633        _ => return None,
634    };
635    if op != BinOp::Eq {
636        return None;
637    }
638    match (lhs, rhs) {
639        (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some(PlanNode::ExprIndexScan {
640            table: table.to_string(),
641            path: stored_json_path(path)?,
642            key: rhs.clone(),
643        }),
644        (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some(PlanNode::ExprIndexScan {
645            table: table.to_string(),
646            path: stored_json_path(path)?,
647            key: lhs.clone(),
648        }),
649        (Expr::Field(name), Expr::Literal(_)) => Some(PlanNode::IndexScan {
650            table: table.to_string(),
651            column: name.clone(),
652            key: rhs.clone(),
653        }),
654        (Expr::Literal(_), Expr::Field(name)) => Some(PlanNode::IndexScan {
655            table: table.to_string(),
656            column: name.clone(),
657            key: lhs.clone(),
658        }),
659        _ => None,
660    }
661}
662
663fn stored_json_path(expr: &Expr) -> Option<StoredJsonPathV1> {
664    JsonPathIdentityV1::from_expr(expr)?.bind_table_local(None)
665}
666
667/// Extract a single range bound from a simple inequality predicate.
668/// Returns `(column, lower_bound, upper_bound)` where at most one bound is set.
669fn extract_single_bound(pred: &Expr) -> Option<RangeBound> {
670    let (lhs, op, rhs) = match pred {
671        Expr::BinaryOp(lhs, op, rhs) => (lhs.as_ref(), *op, rhs.as_ref()),
672        _ => return None,
673    };
674    match op {
675        // .col > literal  →  lower=(literal, exclusive)
676        BinOp::Gt => match (lhs, rhs) {
677            (Expr::Field(name), Expr::Literal(_)) => Some((
678                RangeTarget::Column(name.clone()),
679                Some((rhs.clone(), false)),
680                None,
681            )),
682            (Expr::Literal(_), Expr::Field(name)) => {
683                // literal > .col  →  col < literal  →  upper=(literal, exclusive)
684                Some((
685                    RangeTarget::Column(name.clone()),
686                    None,
687                    Some((lhs.clone(), false)),
688                ))
689            }
690            (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
691                RangeTarget::JsonPath(stored_json_path(path)?),
692                Some((rhs.clone(), false)),
693                None,
694            )),
695            (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
696                RangeTarget::JsonPath(stored_json_path(path)?),
697                None,
698                Some((lhs.clone(), false)),
699            )),
700            _ => None,
701        },
702        // .col >= literal  →  lower=(literal, inclusive)
703        BinOp::Gte => match (lhs, rhs) {
704            (Expr::Field(name), Expr::Literal(_)) => Some((
705                RangeTarget::Column(name.clone()),
706                Some((rhs.clone(), true)),
707                None,
708            )),
709            (Expr::Literal(_), Expr::Field(name)) => Some((
710                RangeTarget::Column(name.clone()),
711                None,
712                Some((lhs.clone(), true)),
713            )),
714            (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
715                RangeTarget::JsonPath(stored_json_path(path)?),
716                Some((rhs.clone(), true)),
717                None,
718            )),
719            (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
720                RangeTarget::JsonPath(stored_json_path(path)?),
721                None,
722                Some((lhs.clone(), true)),
723            )),
724            _ => None,
725        },
726        // .col < literal  →  upper=(literal, exclusive)
727        BinOp::Lt => match (lhs, rhs) {
728            (Expr::Field(name), Expr::Literal(_)) => Some((
729                RangeTarget::Column(name.clone()),
730                None,
731                Some((rhs.clone(), false)),
732            )),
733            (Expr::Literal(_), Expr::Field(name)) => Some((
734                RangeTarget::Column(name.clone()),
735                Some((lhs.clone(), false)),
736                None,
737            )),
738            (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
739                RangeTarget::JsonPath(stored_json_path(path)?),
740                None,
741                Some((rhs.clone(), false)),
742            )),
743            (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
744                RangeTarget::JsonPath(stored_json_path(path)?),
745                Some((lhs.clone(), false)),
746                None,
747            )),
748            _ => None,
749        },
750        // .col <= literal  →  upper=(literal, inclusive)
751        BinOp::Lte => match (lhs, rhs) {
752            (Expr::Field(name), Expr::Literal(_)) => Some((
753                RangeTarget::Column(name.clone()),
754                None,
755                Some((rhs.clone(), true)),
756            )),
757            (Expr::Literal(_), Expr::Field(name)) => Some((
758                RangeTarget::Column(name.clone()),
759                Some((lhs.clone(), true)),
760                None,
761            )),
762            (path @ Expr::JsonPath { .. }, Expr::Literal(_)) => Some((
763                RangeTarget::JsonPath(stored_json_path(path)?),
764                None,
765                Some((rhs.clone(), true)),
766            )),
767            (Expr::Literal(_), path @ Expr::JsonPath { .. }) => Some((
768                RangeTarget::JsonPath(stored_json_path(path)?),
769                Some((lhs.clone(), true)),
770                None,
771            )),
772            _ => None,
773        },
774        _ => None,
775    }
776}
777
778/// If the predicate is an inequality or a conjunction of two inequalities
779/// on the same indexed column, return a RangeScan plan node.
780/// Handles: `.col > lit`, `.col >= lit`, `.col < lit`, `.col <= lit`,
781/// and AND-conjunctions like `.col >= low AND .col <= high` (BETWEEN pattern).
782fn try_extract_range_index_keys(table: &str, pred: &Expr) -> Option<PlanNode> {
783    // Case 1: AND conjunction — try to merge two bounds on the same column.
784    if let Expr::BinaryOp(lhs, BinOp::And, rhs) = pred {
785        if let (Some((col1, s1, e1)), Some((col2, s2, e2))) =
786            (extract_single_bound(lhs), extract_single_bound(rhs))
787        {
788            if col1 == col2 {
789                let start = s1.or(s2);
790                let end = e1.or(e2);
791                if start.is_some() || end.is_some() {
792                    return Some(range_scan_for_target(table, col1, start, end));
793                }
794            }
795        }
796    }
797
798    // Case 2: single inequality.
799    if let Some((col, start, end)) = extract_single_bound(pred) {
800        return Some(range_scan_for_target(table, col, start, end));
801    }
802
803    None
804}
805
806fn range_scan_for_target(
807    table: &str,
808    target: RangeTarget,
809    start: Option<(Expr, bool)>,
810    end: Option<(Expr, bool)>,
811) -> PlanNode {
812    match target {
813        RangeTarget::Column(column) => PlanNode::RangeScan {
814            table: table.to_string(),
815            column,
816            start,
817            end,
818        },
819        RangeTarget::JsonPath(path) => PlanNode::ExprRangeScan {
820            table: table.to_string(),
821            path,
822            start,
823            end,
824        },
825    }
826}
827
828/// Fold only the exact, semantics-preserving single-table shape that can stream
829/// directly from one expression index. Anything involving filters, joins,
830/// grouping, distinct, aggregation, windows, multiple sort keys, or non-integer
831/// slice expressions retains the generic Sort pipeline.
832fn try_extract_ordered_expr_index_scan(query: &QueryExpr) -> Option<PlanNode> {
833    if query.alias.is_some()
834        || !query.joins.is_empty()
835        || query.filter.is_some()
836        || query.group_by.is_some()
837        || query.distinct
838        || query.aggregation.is_some()
839        || query.projection.as_ref().is_some_and(|fields| {
840            fields
841                .iter()
842                .any(|field| matches!(field.expr, Expr::Window { .. }))
843        })
844    {
845        return None;
846    }
847    let order = query.order.as_ref()?;
848    let [key] = order.keys.as_slice() else {
849        return None;
850    };
851    let path = stored_json_path(&key.expr)?;
852    let limit = query.limit.as_ref()?;
853    if !matches!(limit, Expr::Literal(Literal::Int(value)) if *value >= 0) {
854        return None;
855    }
856    if !query
857        .offset
858        .as_ref()
859        .is_none_or(|offset| matches!(offset, Expr::Literal(Literal::Int(value)) if *value >= 0))
860    {
861        return None;
862    }
863    Some(PlanNode::OrderedExprIndexScan {
864        table: query.source.clone(),
865        path,
866        descending: key.descending,
867        limit: limit.clone(),
868        offset: query.offset.clone(),
869    })
870}
871
872/// Walk projection fields, replacing every `Expr::Window { .. }` with
873/// `Expr::Field("__win_N")` and collecting the corresponding `WindowDef`
874/// descriptors. Returns the list of window definitions to insert as a
875/// `PlanNode::Window` before the `Project` node.
876fn extract_windows(proj_fields: &mut [ProjectField]) -> Vec<WindowDef> {
877    let mut defs = Vec::new();
878    let mut counter = 0usize;
879    for f in proj_fields.iter_mut() {
880        if let Expr::Window {
881            function,
882            args,
883            mode,
884            partition_by,
885            order_by,
886        } = &f.expr
887        {
888            let output_name = format!("__win_{counter}");
889            defs.push(WindowDef {
890                function: *function,
891                args: args.clone(),
892                mode: *mode,
893                partition_by: partition_by.clone(),
894                order_by: order_by
895                    .iter()
896                    .map(|k| SortKey {
897                        expr: k.expr.clone(),
898                        descending: k.descending,
899                    })
900                    .collect(),
901                output_name: output_name.clone(),
902            });
903            f.expr = Expr::Field(output_name);
904            counter += 1;
905        }
906    }
907    defs
908}
909
910/// Walk projection fields and HAVING expression, replacing every
911/// `Expr::FunctionCall(func, Field(col))` with `Expr::Field("__agg_N")`
912/// and collecting the corresponding `GroupAgg` descriptors. Deduplicates:
913/// if the same (func, field) pair appears in both projection and HAVING,
914/// they share a single `GroupAgg` entry.
915fn extract_aggregates(
916    proj_fields: &mut [ProjectField],
917    having: &mut Option<Expr>,
918    source_aliases: &std::collections::HashSet<String>,
919) -> Result<Vec<GroupAgg>, PlanError> {
920    let mut aggs: Vec<GroupAgg> = Vec::new();
921    let mut counter = 0usize;
922    for f in proj_fields.iter_mut() {
923        rewrite_agg_expr(&mut f.expr, &mut aggs, &mut counter, source_aliases)?;
924    }
925    if let Some(h) = having {
926        rewrite_agg_expr(h, &mut aggs, &mut counter, source_aliases)?;
927    }
928    Ok(aggs)
929}
930
931fn rewrite_group_key_references(
932    fields: &mut [ProjectField],
933    having: &mut Option<Expr>,
934    keys: &[GroupKey],
935) {
936    for field in fields {
937        rewrite_group_key_expr(&mut field.expr, keys);
938    }
939    if let Some(having) = having {
940        rewrite_group_key_expr(having, keys);
941    }
942}
943
944fn rewrite_group_order_keys(
945    order: Option<&mut OrderClause>,
946    projection: &[ProjectField],
947    keys: &[GroupKey],
948) {
949    let Some(order) = order else {
950        return;
951    };
952    for order_key in &mut order.keys {
953        let Some(group_key) = keys.iter().find(|key| key.expr == order_key.expr) else {
954            continue;
955        };
956        let projected_name = projection
957            .iter()
958            .find(|field| field.expr == group_key.expr)
959            .and_then(|field| field.alias.clone())
960            .unwrap_or_else(|| group_key.output_name());
961        order_key.expr = Expr::Field(projected_name);
962    }
963}
964
965fn rewrite_group_key_expr(expr: &mut Expr, keys: &[GroupKey]) {
966    if let Some(key) = keys.iter().find(|key| key.expr == *expr) {
967        *expr = Expr::Field(key.output_name());
968        return;
969    }
970    match expr {
971        // Aggregate arguments run against input rows and have already been
972        // extracted before this pass, so a survivor must not be rebound to a
973        // grouped output column.
974        Expr::FunctionCall(..) => {}
975        Expr::BinaryOp(left, _, right) | Expr::Coalesce(left, right) => {
976            rewrite_group_key_expr(left, keys);
977            rewrite_group_key_expr(right, keys);
978        }
979        Expr::UnaryOp(_, inner) | Expr::Cast(inner, _) => rewrite_group_key_expr(inner, keys),
980        Expr::ScalarFunc(_, args) => {
981            for arg in args {
982                rewrite_group_key_expr(arg, keys);
983            }
984        }
985        Expr::InList { expr, list, .. } => {
986            rewrite_group_key_expr(expr, keys);
987            for item in list {
988                rewrite_group_key_expr(item, keys);
989            }
990        }
991        Expr::Case { whens, else_expr } => {
992            for (condition, result) in whens {
993                rewrite_group_key_expr(condition, keys);
994                rewrite_group_key_expr(result, keys);
995            }
996            if let Some(expr) = else_expr {
997                rewrite_group_key_expr(expr, keys);
998            }
999        }
1000        _ => {}
1001    }
1002}
1003
1004fn rewrite_agg_expr(
1005    expr: &mut Expr,
1006    aggs: &mut Vec<GroupAgg>,
1007    counter: &mut usize,
1008    source_aliases: &std::collections::HashSet<String>,
1009) -> Result<(), PlanError> {
1010    match expr {
1011        Expr::FunctionCall(func, inner, mode) => {
1012            let output = find_or_insert_agg(aggs, *func, inner, *mode, counter, source_aliases)?;
1013            *expr = Expr::Field(output);
1014        }
1015        Expr::BinaryOp(l, _, r) => {
1016            rewrite_agg_expr(l, aggs, counter, source_aliases)?;
1017            rewrite_agg_expr(r, aggs, counter, source_aliases)?;
1018        }
1019        Expr::UnaryOp(_, inner) => rewrite_agg_expr(inner, aggs, counter, source_aliases)?,
1020        Expr::Coalesce(l, r) => {
1021            rewrite_agg_expr(l, aggs, counter, source_aliases)?;
1022            rewrite_agg_expr(r, aggs, counter, source_aliases)?;
1023        }
1024        Expr::InList { expr: e, list, .. } => {
1025            rewrite_agg_expr(e, aggs, counter, source_aliases)?;
1026            for item in list {
1027                rewrite_agg_expr(item, aggs, counter, source_aliases)?;
1028            }
1029        }
1030        Expr::InSubquery { expr: e, .. } => {
1031            rewrite_agg_expr(e, aggs, counter, source_aliases)?;
1032        }
1033        _ => {}
1034    }
1035    Ok(())
1036}
1037
1038fn find_or_insert_agg(
1039    aggs: &mut Vec<GroupAgg>,
1040    func: AggFunc,
1041    argument: &Expr,
1042    mode: AggregateMode,
1043    counter: &mut usize,
1044    source_aliases: &std::collections::HashSet<String>,
1045) -> Result<String, PlanError> {
1046    for existing in aggs.iter() {
1047        if existing.function == func && existing.argument == *argument && existing.mode == mode {
1048            return Ok(existing.output_name.clone());
1049        }
1050    }
1051    let provenance_alias = symmetric_provenance_alias(func, Some(argument), mode, source_aliases)?;
1052    let output_name = format!("__agg_{counter}");
1053    aggs.push(GroupAgg {
1054        function: func,
1055        argument: argument.clone(),
1056        mode,
1057        provenance_alias,
1058        output_name: output_name.clone(),
1059    });
1060    *counter += 1;
1061    Ok(output_name)
1062}
1063
1064fn symmetric_provenance_alias(
1065    function: AggFunc,
1066    argument: Option<&Expr>,
1067    mode: AggregateMode,
1068    source_aliases: &std::collections::HashSet<String>,
1069) -> Result<Option<String>, PlanError> {
1070    if mode == AggregateMode::Raw
1071        || source_aliases.len() < 2
1072        || !matches!(function, AggFunc::Sum | AggFunc::Avg | AggFunc::Count)
1073        || (function == AggFunc::Count
1074            && argument.is_none_or(|argument| matches!(argument, Expr::Field(name) if name == "*")))
1075    {
1076        return Ok(None);
1077    }
1078    let Some(argument) = argument else {
1079        return Err(symmetric_aggregate_error(
1080            function,
1081            "does not reference a source row",
1082        ));
1083    };
1084
1085    let mut qualified = std::collections::HashSet::new();
1086    let mut has_unqualified = false;
1087    collect_expression_sources(argument, &mut qualified, &mut has_unqualified);
1088
1089    for alias in &qualified {
1090        if !source_aliases.contains(alias) {
1091            return Err(symmetric_aggregate_error(
1092                function,
1093                &format!("references unknown source alias '{alias}'"),
1094            ));
1095        }
1096    }
1097    if has_unqualified {
1098        if source_aliases.len() != 1 {
1099            return Err(symmetric_aggregate_error(
1100                function,
1101                "contains an ambiguous unqualified field",
1102            ));
1103        }
1104        qualified.extend(source_aliases.iter().cloned());
1105    }
1106    match qualified.len() {
1107        1 => Ok(qualified.into_iter().next()),
1108        0 => Err(symmetric_aggregate_error(
1109            function,
1110            "does not reference a source row",
1111        )),
1112        _ => Err(symmetric_aggregate_error(
1113            function,
1114            "references multiple source aliases",
1115        )),
1116    }
1117}
1118
1119fn symmetric_aggregate_error(function: AggFunc, reason: &str) -> PlanError {
1120    let name = format!("{function:?}").to_lowercase();
1121    PlanError::Semantic(format!(
1122        "symmetric {name} expression {reason}; reference exactly one source alias or use {name}(raw ...)"
1123    ))
1124}
1125
1126fn collect_expression_sources(
1127    expr: &Expr,
1128    qualified: &mut std::collections::HashSet<String>,
1129    has_unqualified: &mut bool,
1130) {
1131    match expr {
1132        Expr::Field(name) if name != "*" => *has_unqualified = true,
1133        Expr::QualifiedField { qualifier, .. } => {
1134            qualified.insert(qualifier.clone());
1135        }
1136        Expr::BinaryOp(left, _, right) | Expr::Coalesce(left, right) => {
1137            collect_expression_sources(left, qualified, has_unqualified);
1138            collect_expression_sources(right, qualified, has_unqualified);
1139        }
1140        Expr::UnaryOp(_, inner) | Expr::Cast(inner, _) | Expr::JsonPath { base: inner, .. } => {
1141            collect_expression_sources(inner, qualified, has_unqualified);
1142        }
1143        Expr::ScalarFunc(_, args) => {
1144            for argument in args {
1145                collect_expression_sources(argument, qualified, has_unqualified);
1146            }
1147        }
1148        Expr::InList { expr, list, .. } => {
1149            collect_expression_sources(expr, qualified, has_unqualified);
1150            for item in list {
1151                collect_expression_sources(item, qualified, has_unqualified);
1152            }
1153        }
1154        Expr::InSubquery { expr, .. } => {
1155            collect_expression_sources(expr, qualified, has_unqualified);
1156        }
1157        Expr::Case { whens, else_expr } => {
1158            for (condition, result) in whens {
1159                collect_expression_sources(condition, qualified, has_unqualified);
1160                collect_expression_sources(result, qualified, has_unqualified);
1161            }
1162            if let Some(expr) = else_expr {
1163                collect_expression_sources(expr, qualified, has_unqualified);
1164            }
1165        }
1166        Expr::Window {
1167            args,
1168            partition_by,
1169            order_by,
1170            ..
1171        } => {
1172            for expr in args.iter().chain(partition_by) {
1173                collect_expression_sources(expr, qualified, has_unqualified);
1174            }
1175            for key in order_by {
1176                collect_expression_sources(&key.expr, qualified, has_unqualified);
1177            }
1178        }
1179        Expr::FunctionCall(_, inner, _) => {
1180            collect_expression_sources(inner, qualified, has_unqualified);
1181        }
1182        Expr::ExistsSubquery { .. }
1183        | Expr::Field(_)
1184        | Expr::Literal(_)
1185        | Expr::Param(_)
1186        | Expr::ValueLit(_)
1187        | Expr::Null => {}
1188    }
1189}
1190
1191#[cfg(test)]
1192mod tests {
1193    use super::*;
1194    use crate::plan::PlanNode;
1195
1196    #[test]
1197    fn test_plan_simple_scan() {
1198        let plan = plan("User").unwrap();
1199        assert!(matches!(plan, PlanNode::SeqScan { table } if table == "User"));
1200    }
1201
1202    #[test]
1203    fn test_plan_filter() {
1204        let plan = plan("User filter .age > 30").unwrap();
1205        assert!(matches!(plan, PlanNode::RangeScan { .. }));
1206    }
1207
1208    #[test]
1209    fn test_plan_filter_with_projection() {
1210        let plan = plan("User filter .age > 30 { name, email }").unwrap();
1211        assert!(matches!(plan, PlanNode::Project { .. }));
1212    }
1213
1214    #[test]
1215    fn test_plan_insert() {
1216        let plan = plan(r#"insert User { name := "Alice", age := 30 }"#).unwrap();
1217        assert!(matches!(plan, PlanNode::Insert { .. }));
1218    }
1219
1220    #[test]
1221    fn test_plan_order_limit() {
1222        let plan = plan("User order .name limit 10").unwrap();
1223        match plan {
1224            PlanNode::Limit { input, .. } => {
1225                assert!(matches!(*input, PlanNode::Sort { .. }));
1226            }
1227            _ => panic!("expected Limit(Sort(SeqScan))"),
1228        }
1229    }
1230
1231    #[test]
1232    fn test_plan_count() {
1233        let plan = plan("count(User)").unwrap();
1234        assert!(matches!(plan, PlanNode::Aggregate { .. }));
1235    }
1236
1237    #[test]
1238    fn single_source_aggregates_do_not_request_provenance() {
1239        for query in [
1240            "sum(User { .amount })",
1241            "avg(User { .amount })",
1242            "count(User { .amount })",
1243        ] {
1244            match plan(query).unwrap() {
1245                PlanNode::Aggregate {
1246                    provenance_alias, ..
1247                } => assert!(
1248                    provenance_alias.is_none(),
1249                    "unexpected provenance for {query}"
1250                ),
1251                other => panic!("expected Aggregate for {query}, got {other:?}"),
1252            }
1253        }
1254
1255        match plan("User group .dept { total: sum(.amount) }").unwrap() {
1256            PlanNode::Project { input, .. } => match *input {
1257                PlanNode::GroupBy { aggregates, .. } => {
1258                    assert!(aggregates[0].provenance_alias.is_none());
1259                }
1260                other => panic!("expected GroupBy, got {other:?}"),
1261            },
1262            other => panic!("expected Project(GroupBy), got {other:?}"),
1263        }
1264    }
1265
1266    #[test]
1267    fn join_provenance_is_limited_to_fanout_sensitive_aggregates() {
1268        let base = "Account as a join Entry as e on a.id = e.account_id group a.dept";
1269        for (function, expects_provenance) in [
1270            ("sum(a.balance)", true),
1271            ("avg(a.balance)", true),
1272            ("count(a.balance)", true),
1273            ("min(a.balance)", false),
1274            ("max(a.balance)", false),
1275            ("count(distinct a.balance)", false),
1276            ("count(*)", false),
1277        ] {
1278            let query = format!("{base} {{ value: {function} }}");
1279            match plan(&query).unwrap() {
1280                PlanNode::Project { input, .. } => match *input {
1281                    PlanNode::GroupBy { aggregates, .. } => assert_eq!(
1282                        aggregates[0].provenance_alias.as_deref(),
1283                        expects_provenance.then_some("a"),
1284                        "unexpected provenance selection for {function}"
1285                    ),
1286                    other => panic!("expected GroupBy for {function}, got {other:?}"),
1287                },
1288                other => panic!("expected Project(GroupBy) for {function}, got {other:?}"),
1289            }
1290        }
1291    }
1292
1293    #[test]
1294    fn test_plan_eq_becomes_index_scan() {
1295        // `filter .col = literal` should fold into an IndexScan — the executor
1296        // falls back to a scan if the column happens to lack an index.
1297        let plan = plan("User filter .id = 42").unwrap();
1298        match plan {
1299            PlanNode::IndexScan { table, column, key } => {
1300                assert_eq!(table, "User");
1301                assert_eq!(column, "id");
1302                assert!(matches!(key, Expr::Literal(Literal::Int(42))));
1303            }
1304            other => panic!("expected IndexScan, got {other:?}"),
1305        }
1306    }
1307
1308    #[test]
1309    fn test_plan_eq_reversed_becomes_index_scan() {
1310        // Literal-on-the-left form should fold the same way.
1311        let plan = plan(r#"User filter "NYC" = .city"#).unwrap();
1312        assert!(matches!(plan, PlanNode::IndexScan { .. }));
1313    }
1314
1315    #[test]
1316    fn json_path_equality_and_reversed_equality_are_speculative_expression_scans() {
1317        for query in ["Post filter .data->age = 21", "Post filter 21 = .data->age"] {
1318            match plan(query).unwrap() {
1319                PlanNode::ExprIndexScan { table, path, key } => {
1320                    assert_eq!(table, "Post");
1321                    assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
1322                    assert!(matches!(key, Expr::Literal(Literal::Int(21))));
1323                }
1324                other => panic!("expected ExprIndexScan for `{query}`, got {other:?}"),
1325            }
1326        }
1327    }
1328
1329    #[test]
1330    fn json_path_range_and_same_path_compound_bounds_are_speculative_scans() {
1331        for query in ["Post filter .data->age > 18", "Post filter 18 < .data->age"] {
1332            match plan(query).unwrap() {
1333                PlanNode::ExprRangeScan {
1334                    path, start, end, ..
1335                } => {
1336                    assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
1337                    assert!(start.is_some());
1338                    assert!(end.is_none());
1339                }
1340                other => panic!("expected ExprRangeScan for `{query}`, got {other:?}"),
1341            }
1342        }
1343
1344        match plan("Post filter .data->age >= 18 and .data->age < 65").unwrap() {
1345            PlanNode::ExprRangeScan {
1346                path, start, end, ..
1347            } => {
1348                assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
1349                assert_eq!(start, Some((Expr::Literal(Literal::Int(18)), true)));
1350                assert_eq!(end, Some((Expr::Literal(Literal::Int(65)), false)));
1351            }
1352            other => panic!("expected bounded ExprRangeScan, got {other:?}"),
1353        }
1354
1355        assert!(matches!(
1356            plan("Post filter .data->age >= 18 and .data->score < 65").unwrap(),
1357            PlanNode::Filter { .. }
1358        ));
1359    }
1360
1361    #[test]
1362    fn exact_single_path_order_limit_uses_ordered_expression_scan() {
1363        match plan("Post order .data->age desc limit 10 offset 2 { .id }").unwrap() {
1364            PlanNode::Project { input, .. } => match *input {
1365                PlanNode::OrderedExprIndexScan {
1366                    table,
1367                    path,
1368                    descending,
1369                    limit,
1370                    offset,
1371                } => {
1372                    assert_eq!(table, "Post");
1373                    assert_eq!(path.canonical_text(), "v1:.data->\"age\"");
1374                    assert!(descending);
1375                    assert_eq!(limit, Expr::Literal(Literal::Int(10)));
1376                    assert_eq!(offset, Some(Expr::Literal(Literal::Int(2))));
1377                }
1378                other => panic!("expected OrderedExprIndexScan, got {other:?}"),
1379            },
1380            other => panic!("expected Project(OrderedExprIndexScan), got {other:?}"),
1381        }
1382    }
1383
1384    #[test]
1385    fn incompatible_path_order_shapes_keep_generic_sort() {
1386        for query in [
1387            "Post order .data->age",
1388            "Post order .data->age, .id limit 10",
1389            "Post filter .data->active = true order .data->age limit 10",
1390            "Post order .data->age limit .id",
1391        ] {
1392            let planned = plan(query).unwrap();
1393            assert!(
1394                !plan_contains_ordered_expr_scan(&planned),
1395                "`{query}` must remain on the generic pipeline: {planned:?}"
1396            );
1397        }
1398    }
1399
1400    fn plan_contains_ordered_expr_scan(plan: &PlanNode) -> bool {
1401        match plan {
1402            PlanNode::OrderedExprIndexScan { .. } => true,
1403            PlanNode::Filter { input, .. }
1404            | PlanNode::Project { input, .. }
1405            | PlanNode::Sort { input, .. }
1406            | PlanNode::Limit { input, .. }
1407            | PlanNode::Offset { input, .. }
1408            | PlanNode::Aggregate { input, .. }
1409            | PlanNode::Distinct { input }
1410            | PlanNode::GroupBy { input, .. }
1411            | PlanNode::Update { input, .. }
1412            | PlanNode::Delete { input, .. }
1413            | PlanNode::Window { input, .. }
1414            | PlanNode::Explain { input } => plan_contains_ordered_expr_scan(input),
1415            PlanNode::NestedLoopJoin { left, right, .. } | PlanNode::Union { left, right, .. } => {
1416                plan_contains_ordered_expr_scan(left) || plan_contains_ordered_expr_scan(right)
1417            }
1418            _ => false,
1419        }
1420    }
1421
1422    #[test]
1423    fn test_plan_non_eq_stays_filter() {
1424        // `>` now emits a RangeScan instead of SeqScan+Filter.
1425        let plan = plan("User filter .age > 30").unwrap();
1426        match plan {
1427            PlanNode::RangeScan {
1428                column, start, end, ..
1429            } => {
1430                assert_eq!(column, "age");
1431                assert!(start.is_some(), "expected lower bound");
1432                assert!(end.is_none(), "expected no upper bound");
1433                let (_, inclusive) = start.unwrap();
1434                assert!(!inclusive, "expected exclusive lower bound for >");
1435            }
1436            other => panic!("expected RangeScan, got {other:?}"),
1437        }
1438    }
1439
1440    #[test]
1441    fn test_plan_index_scan_with_projection() {
1442        // Projection on top of an IndexScan should layer correctly.
1443        let plan = plan("User filter .id = 1 { .name }").unwrap();
1444        match plan {
1445            PlanNode::Project { input, .. } => {
1446                assert!(matches!(*input, PlanNode::IndexScan { .. }));
1447            }
1448            other => panic!("expected Project(IndexScan), got {other:?}"),
1449        }
1450    }
1451
1452    #[test]
1453    fn test_plan_update_by_pk_becomes_index_scan() {
1454        // `.id = literal` update should fold to Update(IndexScan), not
1455        // Update(Filter(SeqScan)).
1456        let plan = plan("User filter .id = 42 update { age := 31 }").unwrap();
1457        match plan {
1458            PlanNode::Update { input, .. } => {
1459                assert!(
1460                    matches!(*input, PlanNode::IndexScan { .. }),
1461                    "expected Update(IndexScan), got {input:?}"
1462                );
1463            }
1464            other => panic!("expected Update, got {other:?}"),
1465        }
1466    }
1467
1468    #[test]
1469    fn test_plan_update_range_stays_range_scan() {
1470        let plan = plan("User filter .age > 30 update { age := 31 }").unwrap();
1471        match plan {
1472            PlanNode::Update { input, .. } => {
1473                assert!(
1474                    matches!(*input, PlanNode::RangeScan { .. }),
1475                    "expected Update(RangeScan), got {input:?}"
1476                );
1477            }
1478            other => panic!("expected Update, got {other:?}"),
1479        }
1480    }
1481
1482    #[test]
1483    fn test_plan_delete_by_pk_becomes_index_scan() {
1484        let plan = plan("User filter .id = 7 delete").unwrap();
1485        match plan {
1486            PlanNode::Delete { input, .. } => {
1487                assert!(matches!(*input, PlanNode::IndexScan { .. }));
1488            }
1489            other => panic!("expected Delete, got {other:?}"),
1490        }
1491    }
1492
1493    #[test]
1494    fn test_plan_inner_join_builds_nested_loop() {
1495        // Mission E1.2: a join query should plan to NestedLoopJoin with
1496        // AliasScan leaves on both sides.
1497        let plan = plan("User as u join Order as o on u.id = o.user_id").unwrap();
1498        match plan {
1499            PlanNode::NestedLoopJoin {
1500                left,
1501                right,
1502                on,
1503                kind,
1504            } => {
1505                assert_eq!(kind, JoinKind::Inner);
1506                assert!(on.is_some());
1507                assert!(matches!(*left, PlanNode::AliasScan { .. }));
1508                assert!(matches!(*right, PlanNode::AliasScan { .. }));
1509            }
1510            other => panic!("expected NestedLoopJoin, got {other:?}"),
1511        }
1512    }
1513
1514    #[test]
1515    fn duplicate_join_aliases_are_rejected_before_execution() {
1516        let err = plan("A as x join A as x on x.id = x.id").unwrap_err();
1517        assert!(
1518            err.to_string().contains("duplicate source alias `x`"),
1519            "unexpected error: {err}"
1520        );
1521    }
1522
1523    #[test]
1524    fn test_plan_right_join_rewritten_as_left_with_swapped_inputs() {
1525        let plan = plan("User as u right join Order as o on u.id = o.user_id").unwrap();
1526        match plan {
1527            PlanNode::NestedLoopJoin {
1528                left, right, kind, ..
1529            } => {
1530                assert_eq!(kind, JoinKind::LeftOuter);
1531                // Swapped: Order is now on the left, User on the right.
1532                match *left {
1533                    PlanNode::AliasScan { table, .. } => assert_eq!(table, "Order"),
1534                    other => panic!("expected AliasScan(Order), got {other:?}"),
1535                }
1536                match *right {
1537                    PlanNode::AliasScan { table, .. } => assert_eq!(table, "User"),
1538                    other => panic!("expected AliasScan(User), got {other:?}"),
1539                }
1540            }
1541            other => panic!("expected NestedLoopJoin, got {other:?}"),
1542        }
1543    }
1544
1545    #[test]
1546    fn test_plan_multi_join_is_left_deep() {
1547        // Three sources → two NestedLoopJoins, left-deep.
1548        let plan = plan(
1549            "User as u join Order as o on u.id = o.user_id \
1550             join Product as p on o.product_id = p.id",
1551        )
1552        .unwrap();
1553        match plan {
1554            PlanNode::NestedLoopJoin { left, right, .. } => {
1555                // Outer (Product) join: right is AliasScan(Product)
1556                match *right {
1557                    PlanNode::AliasScan { table, .. } => assert_eq!(table, "Product"),
1558                    other => panic!("expected AliasScan(Product), got {other:?}"),
1559                }
1560                // Outer.left is inner (Order) NestedLoopJoin
1561                assert!(matches!(*left, PlanNode::NestedLoopJoin { .. }));
1562            }
1563            other => panic!("expected NestedLoopJoin, got {other:?}"),
1564        }
1565    }
1566
1567    #[test]
1568    fn test_plan_join_with_filter_tail_wraps_filter_on_top() {
1569        let plan =
1570            plan("User as u join Order as o on u.id = o.user_id filter o.total > 100").unwrap();
1571        match plan {
1572            PlanNode::Filter { input, .. } => {
1573                assert!(matches!(*input, PlanNode::NestedLoopJoin { .. }));
1574            }
1575            other => panic!("expected Filter(NestedLoopJoin), got {other:?}"),
1576        }
1577    }
1578
1579    #[test]
1580    fn test_plan_group_by_builds_groupby_node() {
1581        let plan = plan("User group .status { .status, n: count(.name) }").unwrap();
1582        // Should be Project(GroupBy(SeqScan)).
1583        match plan {
1584            PlanNode::Project { input, fields } => {
1585                assert_eq!(fields.len(), 2);
1586                match *input {
1587                    PlanNode::GroupBy {
1588                        input: inner,
1589                        keys,
1590                        aggregates,
1591                        having,
1592                    } => {
1593                        assert!(matches!(*inner, PlanNode::SeqScan { .. }));
1594                        assert_eq!(
1595                            keys,
1596                            vec![GroupKey {
1597                                expr: Expr::Field("status".into()),
1598                                output_name: "status".into(),
1599                            }]
1600                        );
1601                        assert_eq!(aggregates.len(), 1);
1602                        assert_eq!(aggregates[0].function, AggFunc::Count);
1603                        assert_eq!(aggregates[0].argument, Expr::Field("name".into()));
1604                        assert!(having.is_none());
1605                    }
1606                    other => panic!("expected GroupBy, got {other:?}"),
1607                }
1608            }
1609            other => panic!("expected Project, got {other:?}"),
1610        }
1611    }
1612
1613    #[test]
1614    fn test_plan_joined_group_applies_order_offset_limit_after_grouping() {
1615        let plan = plan(
1616            "User as u join Order as o on u.id = o.user_id \
1617             group u.status { u.status, n: count(*) } order n desc offset 1 limit 2",
1618        )
1619        .unwrap();
1620
1621        let PlanNode::Limit { input, .. } = plan else {
1622            panic!("expected Limit at the grouped-result boundary");
1623        };
1624        let PlanNode::Offset { input, .. } = *input else {
1625            panic!("expected Offset below Limit");
1626        };
1627        let PlanNode::Sort { input, .. } = *input else {
1628            panic!("expected Sort below Offset");
1629        };
1630        let PlanNode::Project { input, .. } = *input else {
1631            panic!("expected Project below Sort");
1632        };
1633        let PlanNode::GroupBy { input, .. } = *input else {
1634            panic!("expected GroupBy below Project");
1635        };
1636        assert!(
1637            matches!(*input, PlanNode::NestedLoopJoin { .. }),
1638            "joined rows must flow into GroupBy before result limiting"
1639        );
1640    }
1641
1642    #[test]
1643    fn test_plan_group_by_having_rewrites_agg_in_having() {
1644        let plan = plan("User group .status having count(.name) > 1 { .status }").unwrap();
1645        match plan {
1646            PlanNode::Project { input, .. } => {
1647                match *input {
1648                    PlanNode::GroupBy {
1649                        having, aggregates, ..
1650                    } => {
1651                        // The planner should have extracted count(.name) into
1652                        // aggregates and rewritten the HAVING to reference __agg_0.
1653                        assert_eq!(aggregates.len(), 1);
1654                        assert_eq!(aggregates[0].output_name, "__agg_0");
1655                        let h = having.expect("having should be Some");
1656                        match h {
1657                            Expr::BinaryOp(l, BinOp::Gt, _) => {
1658                                assert!(
1659                                    matches!(*l, Expr::Field(ref name) if name == "__agg_0"),
1660                                    "expected Field(__agg_0), got {l:?}"
1661                                );
1662                            }
1663                            other => panic!("expected BinaryOp, got {other:?}"),
1664                        }
1665                    }
1666                    other => panic!("expected GroupBy, got {other:?}"),
1667                }
1668            }
1669            other => panic!("expected Project, got {other:?}"),
1670        }
1671    }
1672
1673    #[test]
1674    fn test_plan_window_inserts_window_node_before_project() {
1675        let plan = plan("User { .name, rn: row_number() over (order .age) }").unwrap();
1676        // Expected shape: Project(Window(SeqScan))
1677        match plan {
1678            PlanNode::Project { input, fields } => {
1679                assert_eq!(fields.len(), 2);
1680                // The window expr should have been replaced with Field("__win_0")
1681                assert!(
1682                    matches!(&fields[1].expr, Expr::Field(name) if name == "__win_0"),
1683                    "expected Field(__win_0), got {:?}",
1684                    fields[1].expr
1685                );
1686                match *input {
1687                    PlanNode::Window {
1688                        input: inner,
1689                        windows,
1690                    } => {
1691                        assert_eq!(windows.len(), 1);
1692                        assert_eq!(windows[0].output_name, "__win_0");
1693                        assert!(matches!(*inner, PlanNode::SeqScan { .. }));
1694                    }
1695                    other => panic!("expected Window, got {other:?}"),
1696                }
1697            }
1698            other => panic!("expected Project, got {other:?}"),
1699        }
1700    }
1701
1702    #[test]
1703    fn test_plan_multiple_windows() {
1704        let plan = plan(
1705            "User { .name, rn: row_number() over (order .age), s: sum(.salary) over (partition .dept order .salary) }"
1706        ).unwrap();
1707        match plan {
1708            PlanNode::Project { input, fields } => {
1709                assert_eq!(fields.len(), 3);
1710                assert!(matches!(&fields[1].expr, Expr::Field(name) if name == "__win_0"));
1711                assert!(matches!(&fields[2].expr, Expr::Field(name) if name == "__win_1"));
1712                match *input {
1713                    PlanNode::Window { windows, .. } => {
1714                        assert_eq!(windows.len(), 2);
1715                        assert_eq!(windows[0].output_name, "__win_0");
1716                        assert_eq!(windows[1].output_name, "__win_1");
1717                    }
1718                    other => panic!("expected Window, got {other:?}"),
1719                }
1720            }
1721            other => panic!("expected Project, got {other:?}"),
1722        }
1723    }
1724
1725    #[test]
1726    fn test_plan_no_window_without_over() {
1727        // Plain aggregate in projection should not create a Window node.
1728        let plan = plan("User group .dept { .dept, total: sum(.salary) }").unwrap();
1729        match plan {
1730            PlanNode::Project { input, .. } => {
1731                // Input should be GroupBy, not Window.
1732                assert!(
1733                    matches!(*input, PlanNode::GroupBy { .. }),
1734                    "expected GroupBy under Project, got {:?}",
1735                    input
1736                );
1737            }
1738            other => panic!("expected Project, got {other:?}"),
1739        }
1740    }
1741
1742    #[test]
1743    fn test_plan_explain_wraps_inner() {
1744        let plan = plan("explain User filter .age > 30").unwrap();
1745        match plan {
1746            PlanNode::Explain { input } => {
1747                assert!(
1748                    matches!(*input, PlanNode::RangeScan { .. }),
1749                    "expected Explain(RangeScan), got {:?}",
1750                    input
1751                );
1752            }
1753            other => panic!("expected Explain, got {other:?}"),
1754        }
1755    }
1756
1757    #[test]
1758    fn test_plan_explain_simple_scan() {
1759        let plan = plan("explain User").unwrap();
1760        match plan {
1761            PlanNode::Explain { input } => {
1762                assert!(matches!(*input, PlanNode::SeqScan { .. }));
1763            }
1764            other => panic!("expected Explain(SeqScan), got {other:?}"),
1765        }
1766    }
1767
1768    #[test]
1769    fn test_plan_explain_join() {
1770        let plan = plan("explain User as u join Order as o on u.id = o.user_id").unwrap();
1771        match plan {
1772            PlanNode::Explain { input } => {
1773                assert!(matches!(*input, PlanNode::NestedLoopJoin { .. }));
1774            }
1775            other => panic!("expected Explain(NestedLoopJoin), got {other:?}"),
1776        }
1777    }
1778}