Skip to main content

akar_planner/
planner.rs

1//! Query planner — converts bound statements into logical query plans.
2//!
3//! Builds a tree of logical operators from the bound AST:
4//! - MATCH → ScanNode / ScanRel
5//! - Multiple MATCH patterns combined via join tree (HashJoin / CrossProduct)
6//! - WHERE → Filter (applied after joins)
7//! - RETURN → Projection (topmost operator)
8
9use crate::join_order::{build_join_tree, build_wcoj_intersect, flatten_join_plan};
10use crate::logical_operator::*;
11use akar_binder::bound_statement::*;
12use akar_common::error::PlannerError;
13use akar_parser::ast::Expression;
14use std::collections::HashSet;
15
16/// Whether an ORDER BY key expression is produced by the projection output.
17///
18/// A key is covered when it matches a projection item's alias or is identical
19/// to one of the projected expressions, or when the projection returns the bare
20/// node variable and the key accesses a property of that variable. Keys that are
21/// not covered (e.g. `m.access_count` in `RETURN m.id, m.label ORDER BY
22/// m.access_count`) must sort on the pre-projection columns (P53.37).
23pub fn projection_covers_sort_key(projected: &[BoundExpression], key: &Expression) -> bool {
24    for be in projected {
25        if let Some(alias) = &be.alias {
26            if sort_key_matches_name(key, alias) {
27                return true;
28            }
29        }
30        if &be.expression == key {
31            return true;
32        }
33    }
34    if let Expression::PropertyAccess(obj, _) = key {
35        if let Expression::Variable(var) = &**obj {
36            let bare = Expression::Variable(var.clone());
37            if projected.iter().any(|be| be.expression == bare) {
38                return true;
39            }
40        }
41    }
42    false
43}
44
45/// Whether a sort key refers to the column named by `name` (`x` or `node.prop`).
46fn sort_key_matches_name(key: &Expression, name: &str) -> bool {
47    match key {
48        Expression::Variable(v) => v == name,
49        Expression::PropertyAccess(obj, prop) => {
50            if let Expression::Variable(var) = &**obj {
51                format!("{var}.{prop}") == name
52            } else {
53                false
54            }
55        }
56        _ => false,
57    }
58}
59
60/// AND-combine a newly bound WHERE predicate with any previously accumulated
61/// filter (e.g. an implicit one generated by the binder from inline
62/// node-properties) — otherwise the earlier predicate would be silently
63/// dropped (P48.17 BUG-A). With no prior filter, the new predicate is kept
64/// verbatim.
65fn combine_filter_expr(prev: Option<BoundExpression>, next: BoundExpression) -> BoundExpression {
66    let Some(prev) = prev else { return next };
67    BoundExpression {
68        expression: Expression::BinaryOp(
69            akar_parser::ast::BinaryOp::And,
70            Box::new(prev.expression),
71            Box::new(next.expression),
72        ),
73        resolved_type: akar_common::types::LogicalTypeID::Bool,
74        is_constant: prev.is_constant && next.is_constant,
75        alias: None,
76    }
77}
78
79/// The query planner transforms bound statements into logical query plans.
80pub struct QueryPlanner;
81
82impl QueryPlanner {
83    pub fn new() -> Self {
84        Self
85    }
86
87    pub fn plan(&self, statement: BoundStatement) -> Result<Vec<LogicalOperator>, PlannerError> {
88        match statement {
89            BoundStatement::BoundQuery(query) => self.plan_query(query),
90            BoundStatement::BoundCopyFrom(c) => self.plan_copy_from(c),
91            BoundStatement::BoundUnion(u) => self.plan_union(u),
92            BoundStatement::BoundMerge(m) => self.plan_merge(m),
93            BoundStatement::BoundExplain(e) => self.plan_explain(e),
94            BoundStatement::BoundCreateNodeTable(t) => self.plan_create_node_table(t),
95            BoundStatement::BoundCreateRelTable(t) => self.plan_create_rel_table(t),
96            BoundStatement::BoundDropTable(t) => self.plan_drop_table(t),
97            BoundStatement::BoundAlterTable(a) => self.plan_alter_table(a),
98            BoundStatement::BoundCreateIndex(idx) => self.plan_create_index(idx),
99            BoundStatement::BoundDropIndex(idx) => self.plan_drop_index(idx),
100            BoundStatement::BoundCreateVectorIndex(idx) => self.plan_create_vector_index(idx),
101            BoundStatement::BoundCreateSequence(s) => self.plan_create_sequence(s),
102            BoundStatement::BoundDropSequence(s) => self.plan_drop_sequence(s),
103            BoundStatement::BoundCreateDml(c) => self.plan_create_dml(c),
104            BoundStatement::BoundExportDatabase(e) => self.plan_export_database(e),
105            BoundStatement::BoundImportDatabase(i) => self.plan_import_database(i),
106            BoundStatement::BoundCreateFtsIndex(c) => {
107                Ok(vec![LogicalOperator::CreateFtsIndex(LogicalCreateFtsIndex {
108                    index_name: c.index_name,
109                    table_name: c.table_name,
110                    column_name: c.column_name,
111                    if_not_exists: c.if_not_exists,
112                    docs_table: c.docs_table,
113                    terms_table: c.terms_table,
114                    posting_table: c.posting_table,
115                    cardinality: 1,
116                })])
117            }
118            BoundStatement::BoundStandaloneCall(c) => self.plan_standalone_call(c),
119            _ => Ok(Vec::new()),
120        }
121    }
122
123    /// Plan an EXPLAIN statement.
124    ///
125    /// Plans the inner statement first, then wraps the result in a
126    /// LogicalExplain operator that will serialize the plan tree to text.
127    fn plan_explain(&self, e: BoundExplain) -> Result<Vec<LogicalOperator>, PlannerError> {
128        let inner_plan = self.plan(*e.inner)?;
129        // Take the last operator of the inner plan as the tree root to explain
130        let inner_op = if inner_plan.is_empty() {
131            return Err("Cannot EXPLAIN an empty plan".into());
132        } else if inner_plan.len() == 1 {
133            inner_plan.into_iter().next().unwrap()
134        } else {
135            // Wrap multi-operator pipeline in a projection root
136            LogicalOperator::Projection(LogicalProjection {
137                expressions: Vec::new(),
138                children: inner_plan,
139                cardinality: 0,
140            })
141        };
142
143        Ok(vec![LogicalOperator::Explain(LogicalExplain {
144            inner: Box::new(inner_op),
145            explain_type: e.explain_type,
146            cardinality: 1,
147        })])
148    }
149
150    fn plan_copy_from(&self, c: BoundCopyFrom) -> Result<Vec<LogicalOperator>, PlannerError> {
151        Ok(vec![LogicalOperator::CopyFrom(LogicalCopyFrom {
152            table_name: c.table_name,
153            table_id: c.table_id,
154            file_path: c.file_path,
155            options: c.options,
156            cardinality: 0,
157        })])
158    }
159
160    fn plan_standalone_call(&self, c: BoundStandaloneCall) -> Result<Vec<LogicalOperator>, PlannerError> {
161        Ok(vec![LogicalOperator::StandaloneCall(LogicalStandaloneCall {
162            function_name: c.function_name,
163            args: c.args,
164            cardinality: 1,
165        })])
166    }
167
168    // ==================== DDL Planning ====================
169
170    fn plan_create_node_table(&self, t: BoundCreateNodeTable) -> Result<Vec<LogicalOperator>, PlannerError> {
171        Ok(vec![LogicalOperator::CreateNodeTable(LogicalCreateNodeTable {
172            name: t.name,
173            columns: t.columns,
174            primary_key: t.primary_key,
175            cardinality: 1,
176        })])
177    }
178
179    fn plan_create_rel_table(&self, t: BoundCreateRelTable) -> Result<Vec<LogicalOperator>, PlannerError> {
180        Ok(vec![LogicalOperator::CreateRelTable(LogicalCreateRelTable {
181            name: t.name,
182            from: t.from,
183            to: t.to,
184            columns: t.columns,
185            cardinality: 1,
186        })])
187    }
188
189    fn plan_drop_table(&self, t: BoundDropTable) -> Result<Vec<LogicalOperator>, PlannerError> {
190        Ok(vec![LogicalOperator::DropTable(LogicalDropTable {
191            name: t.name,
192            cardinality: 1,
193        })])
194    }
195
196    fn plan_alter_table(&self, a: BoundAlterTable) -> Result<Vec<LogicalOperator>, PlannerError> {
197        Ok(vec![LogicalOperator::AlterTable(LogicalAlterTable {
198            table_name: a.table_name,
199            action: a.action,
200            cardinality: 1,
201        })])
202    }
203
204    fn plan_create_index(&self, idx: BoundCreateIndex) -> Result<Vec<LogicalOperator>, PlannerError> {
205        Ok(vec![LogicalOperator::CreateIndex(LogicalCreateIndex {
206            index_type: idx.index_type,
207            index_name: idx.index_name,
208            table_name: idx.table_name,
209            column_name: idx.column_name,
210            cardinality: 1,
211        })])
212    }
213
214    fn plan_drop_index(&self, idx: BoundDropIndex) -> Result<Vec<LogicalOperator>, PlannerError> {
215        Ok(vec![LogicalOperator::DropIndex(LogicalDropIndex {
216            index_name: idx.index_name,
217            table_name: idx.table_name,
218            cardinality: 1,
219        })])
220    }
221
222    fn plan_create_vector_index(&self, idx: BoundCreateVectorIndex) -> Result<Vec<LogicalOperator>, PlannerError> {
223        Ok(vec![LogicalOperator::CreateVectorIndex(LogicalCreateVectorIndex {
224            index_name: idx.index_name,
225            table_name: idx.table_name,
226            column_name: idx.column_name,
227            metric: idx.metric,
228            dimensions: idx.dimensions,
229            cardinality: 1,
230        })])
231    }
232
233    fn plan_create_sequence(&self, s: BoundCreateSequence) -> Result<Vec<LogicalOperator>, PlannerError> {
234        Ok(vec![LogicalOperator::CreateSequence(LogicalCreateSequence {
235            name: s.name,
236            if_not_exists: s.if_not_exists,
237            or_replace: s.or_replace,
238            start_with: s.start_with,
239            increment: s.increment,
240            min_value: s.min_value,
241            max_value: s.max_value,
242            cycle: s.cycle,
243            cardinality: 1,
244        })])
245    }
246
247    fn plan_drop_sequence(&self, s: BoundDropSequence) -> Result<Vec<LogicalOperator>, PlannerError> {
248        Ok(vec![LogicalOperator::DropSequence(LogicalDropSequence {
249            name: s.name,
250            if_exists: s.if_exists,
251            cardinality: 1,
252        })])
253    }
254
255    fn plan_create_dml(&self, c: BoundCreateDml) -> Result<Vec<LogicalOperator>, PlannerError> {
256        let first_node = c.patterns.iter().find_map(|p| p.node.clone());
257        let (table_name, table_id, properties) = match first_node {
258            Some(n) => (n.table_name, n.table_id, n.properties),
259            None => (String::new(), 0, Vec::new()),
260        };
261        Ok(vec![LogicalOperator::CreateDml(LogicalCreateDml {
262            table_name,
263            table_id,
264            properties,
265            cardinality: 1,
266        })])
267    }
268
269    fn plan_export_database(&self, e: BoundExportDatabase) -> Result<Vec<LogicalOperator>, PlannerError> {
270        Ok(vec![LogicalOperator::ExportDatabase(LogicalExportDatabase {
271            file_path: e.file_path,
272            file_type: e.file_type,
273            schema_only: e.schema_only,
274            options: e.options,
275            cardinality: 1,
276        })])
277    }
278
279    fn plan_import_database(&self, i: BoundImportDatabase) -> Result<Vec<LogicalOperator>, PlannerError> {
280        Ok(vec![LogicalOperator::ImportDatabase(LogicalImportDatabase {
281            file_path: i.file_path,
282            query: i.query,
283            index_query: i.index_query,
284            cardinality: 1,
285        })])
286    }
287
288    /// Plan a MERGE statement.
289    ///
290    /// Converts the bound merge into a `LogicalMerge` operator with
291    /// ON MATCH SET and ON CREATE SET as `LogicalSet` sub-operators.
292    fn plan_merge(&self, m: BoundMerge) -> Result<Vec<LogicalOperator>, PlannerError> {
293        Ok(self.build_merge_operators(&m))
294    }
295
296    /// Build the update operator(s) for a MERGE clause. Edge MERGE
297    /// (`MERGE (a)-[r:R]->(b)`, P53.20) emits a `LogicalMergeRel`; otherwise a
298    /// node `LogicalMerge` is produced.
299    fn build_merge_operators(&self, m: &BoundMerge) -> Vec<LogicalOperator> {
300        let on_match: Vec<LogicalSet> = m
301            .on_match
302            .iter()
303            .map(|item| LogicalSet {
304                table_name: item.table_name.clone(),
305                table_id: item.table_id,
306                is_node: item.is_node,
307                items: vec![SetItem {
308                    column_name: item.column_name.clone(),
309                    column_idx: item.column_idx,
310                    value: item.value.clone(),
311                }],
312                emit_count: false,
313                cardinality: 0,
314            })
315            .collect();
316
317        let on_create: Vec<LogicalSet> = m
318            .on_create
319            .iter()
320            .map(|item| LogicalSet {
321                table_name: item.table_name.clone(),
322                table_id: item.table_id,
323                is_node: item.is_node,
324                items: vec![SetItem {
325                    column_name: item.column_name.clone(),
326                    column_idx: item.column_idx,
327                    value: item.value.clone(),
328                }],
329                emit_count: false,
330                cardinality: 0,
331            })
332            .collect();
333
334        if let Some(edge) = m.patterns.iter().find_map(|p| p.edge.clone()) {
335            return vec![LogicalOperator::MergeRel(LogicalMergeRel {
336                rel_table_name: edge.table_name,
337                rel_table_id: edge.table_id,
338                edge_var: edge.variable.unwrap_or_default(),
339                src_node_var: edge.src_var,
340                dst_node_var: edge.dst_var,
341                properties: edge.properties,
342                on_match,
343                on_create,
344                cardinality: 0,
345            })];
346        }
347
348        vec![LogicalOperator::Merge(LogicalMerge {
349            table_name: m.table_name.clone(),
350            table_id: m.table_id,
351            properties: m.properties.clone(),
352            on_match,
353            on_create,
354            cardinality: 0,
355        })]
356    }
357
358    /// Plan a UNION or UNION ALL statement.
359    ///
360    /// Plans left and right sub-queries independently, then wraps each
361    /// side's pipeline (potentially multiple operators) into a synthetic
362    /// projection root so that `LogicalUnion` can store them as tree children.
363    fn plan_union(&self, u: BoundUnion) -> Result<Vec<LogicalOperator>, PlannerError> {
364        let left_plan = self.plan_query(*u.left)?;
365        let right_plan = self.plan_query(*u.right)?;
366
367        let left_op = if left_plan.len() == 1 {
368            left_plan.into_iter().next().unwrap()
369        } else {
370            // Wrap multi-operator pipeline in a projection root
371            LogicalOperator::Projection(LogicalProjection {
372                expressions: Vec::new(),
373                children: left_plan,
374                cardinality: 0,
375            })
376        };
377
378        let right_op = if right_plan.len() == 1 {
379            right_plan.into_iter().next().unwrap()
380        } else {
381            LogicalOperator::Projection(LogicalProjection {
382                expressions: Vec::new(),
383                children: right_plan,
384                cardinality: 0,
385            })
386        };
387
388        Ok(vec![LogicalOperator::Union(LogicalUnion {
389            left: Box::new(left_op),
390            right: Box::new(right_op),
391            all: u.all,
392            cardinality: 0,
393        })])
394    }
395
396    pub fn plan_query(&self, query: BoundQuery) -> Result<Vec<LogicalOperator>, PlannerError> {
397        let mut scan_ops: Vec<LogicalOperator> = Vec::new();
398        let mut filter_expr: Option<BoundExpression> = None;
399        let mut projection: Option<LogicalProjection> = None;
400        let mut distinct = false;
401        let mut delete_exprs: Vec<LogicalOperator> = Vec::new();
402        let mut extend_ops: Vec<LogicalOperator> = Vec::new();
403        // ORDER BY / LIMIT / SKIP from RETURN clause
404        let mut order_by: Option<Vec<BoundOrderByItem>> = None;
405        let mut limit: Option<u64> = None;
406        let mut skip: Option<u64> = None;
407        // Flag to skip destination node pattern consumed by RecursiveExtend or Extend
408        let mut skip_next_node = false;
409        // Node variables already bound to a scan in the current pipeline.
410        // Prevents duplicate scans for a shared variable across comma patterns
411        // (P48.1): `MATCH (a)-[:r1]->(b), (b)-[:r3]->(c)` must not scan `b` twice.
412        let mut available_vars: HashSet<String> = HashSet::new();
413        // A terminal SET (no trailing RETURN) reports "N rows updated" via
414        // column 0; with a trailing RETURN the SET must pass through the matched
415        // rows (0 rows when nothing matched) instead of a phantom count row
416        // (P53.39 / kairos P59.1).
417        let has_return = query.clauses.iter().any(|c| matches!(c, BoundClause::BoundReturn(_)));
418
419        for clause in query.clauses {
420            match clause {
421                BoundClause::BoundMatch(mut m) => {
422                    let mut fts_to_assign = m.fts_query.as_ref().map(|fq| LogicalFtsScan {
423                        index_name: fq.index_name.clone(),
424                        query_string: fq.query_string.clone(),
425                        docs_table: fq.docs_table.clone(),
426                        terms_table: fq.terms_table.clone(),
427                        posting_table: fq.posting_table.clone(),
428                        table_name: fq.table_name.clone(),
429                        column_name: fq.column_name.clone(),
430                        cardinality: 0,
431                    });
432                    let patterns: Vec<BoundPattern> = std::mem::take(&mut m.patterns);
433
434                    // WCOJ pass: `MATCH (a)-[:r1]->(b), (a)-[:r2]->(c)` becomes a single
435                    // Intersect that probes the shared node once across all build sides.
436                    // Triangle queries additionally get closure-edge Extend+Filter ops.
437                    if let Some((wcoj_op, wcoj_trailing)) = build_wcoj_intersect(&patterns) {
438                        scan_ops.push(wcoj_op);
439                        extend_ops.extend(wcoj_trailing);
440                    } else {
441                        let mut patterns_iter = patterns.into_iter().peekable();
442                        let _skip_next_node = false;
443                        while let Some(pattern) = patterns_iter.next() {
444                            // If the previous pattern consumed this dest node, skip the node scan
445                            let skip_current_node_scan = skip_next_node;
446                            skip_next_node = false;
447
448                            // Check if this pattern has a var-length edge → create RecursiveExtend
449                            if let Some(ref edge) = pattern.edge {
450                                let is_var_length = edge.lower_bound.is_some() || edge.upper_bound.is_some();
451                                if is_var_length {
452                                    let lb = edge.lower_bound.unwrap_or(0);
453                                    let ub = edge.upper_bound.unwrap_or(1);
454                                    let direction = match edge.direction {
455                                        akar_parser::ast::EdgeDirection::LeftToRight => {
456                                            akar_common::enums::ExtendDirection::Fwd
457                                        }
458                                        akar_parser::ast::EdgeDirection::RightToLeft => {
459                                            akar_common::enums::ExtendDirection::Bwd
460                                        }
461                                        akar_parser::ast::EdgeDirection::Both => {
462                                            akar_common::enums::ExtendDirection::Both
463                                        }
464                                    };
465
466                                    // Scan source node
467                                    let node_var = &pattern.node_variable;
468                                    let var_len_src_bound =
469                                        node_var.as_ref().is_some_and(|v| available_vars.contains(v));
470                                    if !skip_current_node_scan && !var_len_src_bound {
471                                        if let Some(label) = pattern.node_label {
472                                            scan_ops.push(LogicalOperator::ScanNode(LogicalScanNode {
473                                                table_name: label,
474                                                table_id: pattern.node_table_id.unwrap_or(0),
475                                                alias: node_var.clone(),
476                                                columns: Vec::new(),
477                                                cardinality: 0,
478                                                fts_query: fts_to_assign.take(),
479                                                predicate: None,
480                                            }));
481                                            if let Some(v) = node_var {
482                                                available_vars.insert(v.clone());
483                                            }
484                                        }
485                                    }
486
487                                    // Create RecursiveExtend (consumes destination node pattern)
488                                    let rel_table_ids = edge.rel_table_id.map_or(vec![], |id| vec![id]);
489                                    let rel_labels = edge.label.as_ref().map_or(vec![], |l| vec![l.clone()]);
490                                    let target_var = patterns_iter
491                                        .peek()
492                                        .and_then(|p| p.node_variable.clone())
493                                        .unwrap_or_default();
494                                    scan_ops.push(LogicalOperator::RecursiveExtend(LogicalRecursiveExtend {
495                                        source_var: node_var.clone().unwrap_or_default(),
496                                        source_table_id: pattern.node_table_id.unwrap_or(0),
497                                        edge_var: edge.variable.clone(),
498                                        target_var: target_var.clone(),
499                                        rel_table_ids,
500                                        rel_labels,
501                                        lower_bound: lb,
502                                        upper_bound: ub,
503                                        direction,
504                                        semantic: akar_common::enums::PathSemantic::Walk,
505                                        weight_property: None,
506                                        cost_output_name: None,
507                                        cardinality: 0,
508                                    }));
509                                    // The destination node is produced by the RecursiveExtend (P48.1).
510                                    if !target_var.is_empty() {
511                                        available_vars.insert(target_var);
512                                    }
513                                    // Skip the destination node pattern scan
514                                    skip_next_node = true;
515                                    continue;
516                                }
517
518                                // Regular (non-var-length) edge → create Extend
519                                // Scan the source node (clone what we need before pattern is moved)
520                                let src_node_var = pattern.node_variable.clone();
521                                let src_already_bound =
522                                    src_node_var.as_ref().is_some_and(|v| available_vars.contains(v));
523                                if !skip_current_node_scan && !src_already_bound {
524                                    if let Some(label) = &pattern.node_label {
525                                        scan_ops.push(LogicalOperator::ScanNode(LogicalScanNode {
526                                            table_name: label.clone(),
527                                            table_id: pattern.node_table_id.unwrap_or(0),
528                                            alias: src_node_var.clone(),
529                                            columns: Vec::new(),
530                                            cardinality: 0,
531                                            fts_query: fts_to_assign.take(),
532                                            predicate: None,
533                                        }));
534                                        if let Some(v) = &src_node_var {
535                                            available_vars.insert(v.clone());
536                                        }
537                                    }
538                                }
539
540                                // Create Extend which replaces ScanRel + destination ScanNode
541                                if let Some(rel_label) = &edge.label {
542                                    let dest_pattern = patterns_iter.peek();
543                                    let dst_var =
544                                        dest_pattern.and_then(|p| p.node_variable.clone()).unwrap_or_default();
545                                    let dst_table_name =
546                                        dest_pattern.and_then(|p| p.node_label.clone()).unwrap_or_default();
547                                    let dst_table_id = dest_pattern.and_then(|p| p.node_table_id).unwrap_or(0);
548
549                                    extend_ops.push(LogicalOperator::Extend(LogicalExtend {
550                                        rel_table_name: rel_label.clone(),
551                                        rel_table_id: edge.rel_table_id.unwrap_or(0),
552                                        rel_var: edge.variable.clone().unwrap_or_default(),
553                                        bound_node_var: src_node_var.unwrap_or_default(),
554                                        direction: edge.direction.clone(),
555                                        dst_node_var: dst_var.clone(),
556                                        dst_table_name,
557                                        dst_table_id,
558                                        cardinality: 0,
559                                    }));
560                                    // The destination node is produced by this Extend — it becomes
561                                    // available to later patterns without a fresh scan (P48.1).
562                                    if !dst_var.is_empty() {
563                                        available_vars.insert(dst_var);
564                                    }
565
566                                    // Skip the destination node pattern scan
567                                    skip_next_node = true;
568                                    continue;
569                                }
570                            }
571
572                            // Regular (non-var-length) pattern without edge: Scan node only
573                            if !skip_current_node_scan {
574                                if let Some(label) = pattern.node_label {
575                                    let var = pattern.node_variable.clone();
576                                    if !var.as_ref().is_some_and(|v| available_vars.contains(v)) {
577                                        scan_ops.push(LogicalOperator::ScanNode(LogicalScanNode {
578                                            table_name: label,
579                                            table_id: pattern.node_table_id.unwrap_or(0),
580                                            alias: var.clone(),
581                                            columns: Vec::new(),
582                                            cardinality: 0,
583                                            fts_query: fts_to_assign.take(),
584                                            predicate: None,
585                                        }));
586                                        if let Some(v) = var {
587                                            available_vars.insert(v);
588                                        }
589                                    }
590                                }
591                            }
592                        }
593                    } // end else (regular pattern loop)
594                }
595
596                BoundClause::BoundWhere(w) => {
597                    if !delete_exprs.is_empty() {
598                        // WHERE appearing after a WITH/UNWIND/update clause filters the
599                        // in-flight pipeline, not the scan. Push it as a Filter on top of
600                        // the clauses accumulated so far (P53.14).
601                        delete_exprs.push(LogicalOperator::Filter(LogicalFilter {
602                            expression: w.expression.expression,
603                            children: Vec::new(),
604                            cardinality: 0,
605                        }));
606                    } else {
607                        // Combine with any prior WHERE clause — see `combine_filter_expr`
608                        // (P48.17 BUG-A).
609                        filter_expr = Some(combine_filter_expr(filter_expr.take(), w.expression));
610                    }
611                }
612                BoundClause::BoundReturn(r) => {
613                    distinct = r.distinct;
614                    order_by = r.order_by;
615                    limit = r.limit;
616                    skip = r.skip;
617                    projection = Some(LogicalProjection {
618                        expressions: r.expressions,
619                        children: Vec::new(),
620                        cardinality: 0,
621                    });
622                }
623                BoundClause::BoundWith(r) => {
624                    delete_exprs.push(LogicalOperator::Projection(LogicalProjection {
625                        expressions: r.expressions,
626                        children: Vec::new(),
627                        cardinality: 0,
628                    }));
629                }
630                BoundClause::BoundOptionalMatch(om) => {
631                    // P53.25: Detect the bound-edge-probe shape — an optional
632                    // pattern that is a single edge between two node variables
633                    // already bound by the required side:
634                    //   `OPTIONAL MATCH (a)-[existing:Connected]-(b)`
635                    // where `a`/`b` come from the mandatory MATCH and carry no
636                    // label/property constraints here. Such patterns are executed
637                    // by probing the relationship adjacency per input row
638                    // (OptionalExtend) rather than scanning + outer-joining, so
639                    // the compound `a._id`/`b._id` endpoints survive.
640                    let probe_edge_shape = om.patterns.len() == 2
641                        && om.patterns[0].edge.as_ref().is_some_and(|e| {
642                            e.variable.is_some()
643                                && e.label.as_ref().is_some_and(|l| !l.is_empty())
644                                && e.properties.is_empty()
645                                && e.lower_bound.is_none()
646                                && e.upper_bound.is_none()
647                        })
648                        && om.patterns[1].edge.is_none();
649
650                    let both_endpoints_bound = probe_edge_shape
651                        && om.patterns.iter().all(|p| {
652                            p.node_label.is_none()
653                                && p.properties.is_empty()
654                                && p.node_variable.as_ref().is_some_and(|v| available_vars.contains(v))
655                        });
656
657                    // P53.40 (kairos Finding #22): anonymous labeled destination —
658                    // `(m)-[r:Connected]-(:Memory)` with `m` bound and no dst
659                    // variable downstream. The rel table's endpoint schema
660                    // guarantees the destination node type, so probing all edges
661                    // incident on the source is equivalent to the full pattern;
662                    // each match fans out one row per edge (zero → one NULL row).
663                    // The previous fallback routed this through the generic
664                    // OptionalMatch merge, which cross-producted when no column
665                    // name is shared and duplicated every left row.
666                    let anon_dst_fanout = probe_edge_shape
667                        && om.patterns[0].node_label.is_none()
668                        && om.patterns[0].properties.is_empty()
669                        && om.patterns[0]
670                            .node_variable
671                            .as_ref()
672                            .is_some_and(|v| available_vars.contains(v))
673                        && om.patterns[1].node_variable.is_none()
674                        && om.patterns[1].properties.is_empty()
675                        && om.patterns[1].node_label.is_some();
676
677                    let edge_probe = both_endpoints_bound || anon_dst_fanout;
678
679                    // Build the current required-side pipeline (left child)
680                    let mut left_pipeline: Vec<LogicalOperator> = Vec::new();
681                    if !scan_ops.is_empty() {
682                        if scan_ops.len() == 1 {
683                            left_pipeline.push(scan_ops.into_iter().next().unwrap());
684                        } else {
685                            let join_plan = build_join_tree(scan_ops, filter_expr.as_ref());
686                            let flattened = flatten_join_plan(&join_plan);
687                            left_pipeline.extend(flattened);
688                        }
689                    }
690                    if let Some(expr) = filter_expr.take() {
691                        left_pipeline.push(LogicalOperator::Filter(LogicalFilter {
692                            expression: expr.expression,
693                            children: Vec::new(),
694                            cardinality: 0,
695                        }));
696                    }
697                    if let Some(proj) = projection.take() {
698                        left_pipeline.push(LogicalOperator::Projection(proj));
699                    }
700                    if edge_probe {
701                        let edge = om.patterns[0].edge.as_ref().unwrap();
702                        let src_var = om.patterns[0].node_variable.clone().unwrap();
703                        // Anonymous destination (P53.40 fan-out): empty string
704                        // tells the physical operator to emit every incident
705                        // edge instead of probing a specific pair.
706                        let dst_var = om.patterns[1].node_variable.clone().unwrap_or_default();
707                        delete_exprs.push(LogicalOperator::OptionalExtend(LogicalOptionalExtend {
708                            children: left_pipeline,
709                            rel_table_name: edge.label.clone().unwrap(),
710                            rel_table_id: edge.rel_table_id.unwrap_or(0),
711                            rel_var: edge.variable.clone().unwrap(),
712                            src_node_var: src_var,
713                            dst_node_var: dst_var,
714                            direction: edge.direction.clone(),
715                            cardinality: 0,
716                        }));
717                    } else {
718                        let left_op = if left_pipeline.len() == 1 {
719                            left_pipeline.into_iter().next().unwrap()
720                        } else if left_pipeline.is_empty() {
721                            // Empty left side — use a dummy scan
722                            LogicalOperator::ScanNode(LogicalScanNode {
723                                table_name: String::new(),
724                                table_id: 0,
725                                alias: None,
726                                columns: Vec::new(),
727                                cardinality: 0,
728                                fts_query: None,
729                                predicate: None,
730                            })
731                        } else {
732                            LogicalOperator::Projection(LogicalProjection {
733                                expressions: Vec::new(),
734                                children: left_pipeline,
735                                cardinality: 0,
736                            })
737                        };
738
739                        // Build the optional-side pipeline (right child)
740                        let mut right_ops: Vec<LogicalOperator> = Vec::new();
741                        for pattern in &om.patterns {
742                            if let Some(label) = &pattern.node_label {
743                                right_ops.push(LogicalOperator::ScanNode(LogicalScanNode {
744                                    table_name: label.clone(),
745                                    table_id: pattern.node_table_id.unwrap_or(0),
746                                    alias: pattern.node_variable.clone(),
747                                    columns: Vec::new(),
748                                    cardinality: 0,
749                                    fts_query: None,
750                                    predicate: None,
751                                }));
752                            }
753                            if let Some(edge) = &pattern.edge
754                                && let Some(rel_label) = &edge.label
755                            {
756                                right_ops.push(LogicalOperator::ScanRel(LogicalScanRel {
757                                    table_name: rel_label.clone(),
758                                    table_id: edge.rel_table_id.unwrap_or(0),
759                                    direction: edge.direction.clone(),
760                                    cardinality: 0,
761                                }));
762                            }
763                        }
764                        // Apply inline node/edge property predicates to the optional
765                        // side, mirroring the implicit WHERE the binder generates
766                        // for MATCH. Without this, `OPTIONAL MATCH (m:T {id: 999})`
767                        // scans every T row (predicate silently dropped) and the
768                        // left-outer merge degenerates into a cross product.
769                        let mut inline_exprs: Vec<Expression> = Vec::new();
770                        for pattern in &om.patterns {
771                            if let Some(node_var) = &pattern.node_variable {
772                                for (key, val_expr) in &pattern.properties {
773                                    inline_exprs.push(Expression::BinaryOp(
774                                        akar_parser::ast::BinaryOp::Equal,
775                                        Box::new(Expression::PropertyAccess(
776                                            Box::new(Expression::Variable(node_var.clone())),
777                                            key.clone(),
778                                        )),
779                                        Box::new(val_expr.clone()),
780                                    ));
781                                }
782                            }
783                            if let Some(edge) = &pattern.edge
784                                && let Some(edge_var) = &edge.variable
785                            {
786                                for (key, val_expr) in &edge.properties {
787                                    inline_exprs.push(Expression::BinaryOp(
788                                        akar_parser::ast::BinaryOp::Equal,
789                                        Box::new(Expression::PropertyAccess(
790                                            Box::new(Expression::Variable(edge_var.clone())),
791                                            key.clone(),
792                                        )),
793                                        Box::new(val_expr.clone()),
794                                    ));
795                                }
796                            }
797                        }
798                        if !inline_exprs.is_empty() {
799                            let combined = inline_exprs.into_iter().reduce(|acc, e| {
800                                Expression::BinaryOp(akar_parser::ast::BinaryOp::And, Box::new(acc), Box::new(e))
801                            });
802                            right_ops.push(LogicalOperator::Filter(LogicalFilter {
803                                expression: combined.unwrap(),
804                                children: Vec::new(),
805                                cardinality: 0,
806                            }));
807                        }
808                        let right_op = if right_ops.len() == 1 {
809                            right_ops.into_iter().next().unwrap()
810                        } else if right_ops.is_empty() {
811                            LogicalOperator::ScanNode(LogicalScanNode {
812                                table_name: String::new(),
813                                table_id: 0,
814                                alias: None,
815                                columns: Vec::new(),
816                                cardinality: 0,
817                                fts_query: None,
818                                predicate: None,
819                            })
820                        } else {
821                            LogicalOperator::Projection(LogicalProjection {
822                                expressions: Vec::new(),
823                                children: right_ops,
824                                cardinality: 0,
825                            })
826                        };
827
828                        // Create the OptionalMatch tree node.
829                        // The left side is the entire pipeline built so far (scans + filter + projection).
830                        // The right side is the optional pattern scans.
831                        // Push to delete_exprs so it gets appended at the end of the pipeline.
832                        delete_exprs.push(LogicalOperator::OptionalMatch(LogicalOptionalMatch {
833                            left: Box::new(left_op),
834                            right: Box::new(right_op),
835                            cardinality: 0,
836                        }));
837                    }
838                    // Reset pipeline state — subsequent clauses (DELETE, SET, etc.) build fresh
839                    scan_ops = Vec::new();
840                    filter_expr = None;
841                    projection = None;
842                    distinct = false;
843                }
844                BoundClause::BoundDelete(d) => {
845                    for item in &d.items {
846                        delete_exprs.push(LogicalOperator::Delete(LogicalDelete {
847                            table_name: item.table_name.clone(),
848                            table_id: item.table_id,
849                            primary_key_column: item.primary_key_column.clone(),
850                            is_node: item.is_node,
851                            detach: d.detach,
852                            cardinality: 0,
853                        }));
854                    }
855                }
856                BoundClause::BoundUnwind(u) => {
857                    // Treat UNWIND as a scan operator: it produces the row
858                    // variable that later MATCH clauses join against. Routing it
859                    // through the scan list (instead of delete_exprs) lets the
860                    // join tree combine UNWIND rows with node scans, and keeps
861                    // implicit WHERE predicates from MATCH inline properties in
862                    // the scan filter instead of the top-level pipeline (P53.25).
863                    scan_ops.push(LogicalOperator::Unwind(LogicalUnwind {
864                        expression: u.expression.clone(),
865                        variable: u.variable.clone(),
866                        cardinality: 0,
867                    }));
868                    if !u.variable.is_empty() {
869                        available_vars.insert(u.variable.clone());
870                    }
871                }
872                BoundClause::BoundSet(s) => {
873                    // Merge every item of a single SET clause into one operator
874                    // (grouped by target table). All items then evaluate against
875                    // the same pre-update snapshot (P53.17) — chaining one
876                    // operator per item would feed items 2+ the previous item's
877                    // count chunk, losing the scan rows (`SET a=123.0, b=b+1`
878                    // left `b` at its old value).
879                    let mut groups: Vec<(String, u64, bool, Vec<SetItem>)> = Vec::new();
880                    for item in &s.items {
881                        let key = (item.table_name.clone(), item.table_id, item.is_node);
882                        match groups
883                            .iter_mut()
884                            .find(|(n, id, n2, _)| *n == key.0 && *id == key.1 && *n2 == key.2)
885                        {
886                            Some((_, _, _, items)) => items.push(SetItem {
887                                column_name: item.column_name.clone(),
888                                column_idx: item.column_idx,
889                                value: item.value.clone(),
890                            }),
891                            None => groups.push((
892                                key.0,
893                                key.1,
894                                key.2,
895                                vec![SetItem {
896                                    column_name: item.column_name.clone(),
897                                    column_idx: item.column_idx,
898                                    value: item.value.clone(),
899                                }],
900                            )),
901                        }
902                    }
903                    for (table_name, table_id, is_node, items) in groups {
904                        delete_exprs.push(LogicalOperator::Set(LogicalSet {
905                            table_name,
906                            table_id,
907                            is_node,
908                            items,
909                            emit_count: !has_return,
910                            cardinality: 0,
911                        }));
912                    }
913                }
914                BoundClause::BoundCreate(c) => {
915                    let mut patterns_iter = c.patterns.into_iter().peekable();
916                    while let Some(pattern) = patterns_iter.next() {
917                        let node_var = pattern.node_variable.clone().unwrap_or_default();
918
919                        if c.new_variables.iter().any(|v| v.name == node_var) {
920                            delete_exprs.push(LogicalOperator::CreateNode(LogicalCreateNode {
921                                table_name: pattern.node_label.clone().unwrap_or_default(),
922                                table_id: pattern.node_table_id.unwrap_or(0),
923                                out_var_name: node_var.clone(),
924                                properties: pattern.properties.clone(),
925                                cardinality: 0,
926                            }));
927                        }
928
929                        if let Some(edge) = pattern.edge {
930                            let dest_var = patterns_iter
931                                .peek()
932                                .and_then(|p| p.node_variable.clone())
933                                .unwrap_or_default();
934                            let (src_node_name, dst_node_name) = match edge.direction {
935                                akar_parser::ast::EdgeDirection::RightToLeft => (dest_var, node_var.clone()),
936                                _ => (node_var.clone(), dest_var),
937                            };
938
939                            delete_exprs.push(LogicalOperator::CreateRel(LogicalCreateRel {
940                                table_name: edge.label.clone().unwrap_or_default(),
941                                table_id: edge.rel_table_id.unwrap_or(0),
942                                src_node_name,
943                                dst_node_name,
944                                properties: edge.properties.clone(),
945                                cardinality: 0,
946                            }));
947                        }
948                    }
949                }
950                BoundClause::BoundForeach(f) => {
951                    // Plan FOREACH sub-statements
952                    let mut sub_plans = Vec::new();
953                    for sub_stmt in &f.sub_statements {
954                        let plan = self.plan(sub_stmt.clone())?;
955                        sub_plans.push(plan);
956                    }
957                    delete_exprs.push(LogicalOperator::Foreach(LogicalForeach {
958                        variable: f.variable.clone(),
959                        expression: f.expression.clone(),
960                        sub_plans,
961                        cardinality: 0,
962                    }));
963                }
964                BoundClause::BoundMerge(m) => {
965                    delete_exprs.extend(self.build_merge_operators(&m));
966                }
967            }
968        }
969
970        // Collect delete/set clauses (added after the main pipeline)
971        let delete_ops: Vec<LogicalOperator> = std::mem::take(&mut delete_exprs);
972
973        // Build operator pipeline bottom-up
974        let mut result: Vec<LogicalOperator> = Vec::new();
975
976        if scan_ops.is_empty() {
977            // All scans live inside delete_ops children (e.g. an OPTIONAL
978            // MATCH pipeline resets `scan_ops`). The RETURN tail must still be
979            // emitted — this fast path previously dropped ORDER BY and LIMIT
980            // entirely (P53.40).
981            result.extend(delete_ops);
982            append_return_tail(&mut result, projection, order_by, distinct, limit, skip);
983            return Ok(result);
984        }
985
986        if scan_ops.len() == 1 {
987            // Single scan — no join needed
988            result.push(scan_ops.into_iter().next().unwrap());
989        } else {
990            // Multiple scans — build join tree with greedy ordering
991            let join_plan = build_join_tree(scan_ops, filter_expr.as_ref());
992            let flattened = flatten_join_plan(&join_plan);
993            result.extend(flattened);
994        }
995
996        // Append extend operators (replace ScanRel in pipeline)
997        result.extend(std::mem::take(&mut extend_ops));
998
999        // Apply filter on top of scans/joins/extends
1000        if let Some(expr) = filter_expr {
1001            result.push(LogicalOperator::Filter(LogicalFilter {
1002                expression: expr.expression,
1003                children: Vec::new(),
1004                cardinality: 0,
1005            }));
1006        }
1007
1008        // Append update/delete/set/create/unwind/with-projection operators.
1009        // These run before the RETURN projection so that updates are visible to
1010        // the returned expressions (P53.14): `MATCH ... SET ... RETURN ...`.
1011        result.extend(delete_ops);
1012
1013        // Project as topmost, then ORDER BY / LIMIT (shared tail, P53.40)
1014        append_return_tail(&mut result, projection, order_by, distinct, limit, skip);
1015
1016        Ok(result)
1017    }
1018}
1019
1020/// Append the RETURN projection tail shared by BOTH plan-assembly paths:
1021/// the optional below-projection sort (P53.37), the projection itself, the
1022/// DISTINCT dedup aggregate, the above-projection sort, and LIMIT/SKIP.
1023///
1024/// Extracted because the `scan_ops.is_empty()` fast path (e.g. after an
1025/// OPTIONAL MATCH resets the scan state) previously returned early with only
1026/// `[delete_ops..., Projection]` — silently dropping ORDER BY and LIMIT for
1027/// every such query (P53.40, kairos Finding #22).
1028fn append_return_tail(
1029    result: &mut Vec<LogicalOperator>,
1030    projection: Option<LogicalProjection>,
1031    mut order_by: Option<Vec<BoundOrderByItem>>,
1032    distinct: bool,
1033    limit: Option<u64>,
1034    skip: Option<u64>,
1035) {
1036    if let Some(proj) = projection {
1037        let group_by = if distinct {
1038            Some(
1039                proj.expressions
1040                    .iter()
1041                    .map(|be| be.expression.clone())
1042                    .collect::<Vec<Expression>>(),
1043            )
1044        } else {
1045            None
1046        };
1047
1048        // When ORDER BY references a column the projection does not output
1049        // (e.g. `RETURN m.id, m.label ORDER BY m.access_count`), the sort key
1050        // cannot be evaluated against the projected (pruned) chunk. Push the
1051        // sort below the projection so it runs against the full pre-projection
1052        // columns (P53.37). It stays on top when every key is covered by the
1053        // projection output (alias or identical expression), or when DISTINCT
1054        // deduplicates above (sorting before dedup would lose the order).
1055        let order_by_below_projection = match &order_by {
1056            Some(items) => {
1057                group_by.is_none()
1058                    && !items
1059                        .iter()
1060                        .all(|item| projection_covers_sort_key(&proj.expressions, &item.expression.expression))
1061            }
1062            None => false,
1063        };
1064
1065        if order_by_below_projection {
1066            if let Some(items) = order_by.take() {
1067                let sort_keys: Vec<(Expression, bool)> = items
1068                    .iter()
1069                    .map(|item| (item.expression.expression.clone(), item.ascending))
1070                    .collect();
1071                result.push(LogicalOperator::OrderBy(LogicalOrderBy {
1072                    sort_keys,
1073                    children: Vec::new(),
1074                    cardinality: 0,
1075                }));
1076            }
1077        }
1078
1079        result.push(LogicalOperator::Projection(proj));
1080        // DISTINCT is implemented as a hash aggregate with group-by keys and no aggregate functions
1081        if let Some(gb) = group_by {
1082            result.push(LogicalOperator::Aggregate(LogicalAggregate {
1083                group_by: gb,
1084                aggregates: Vec::new(),
1085                children: Vec::new(),
1086                cardinality: 0,
1087            }));
1088        }
1089    }
1090
1091    // Insert ORDER BY operator if present
1092    if let Some(items) = order_by {
1093        let sort_keys: Vec<(Expression, bool)> = items
1094            .iter()
1095            .map(|item| (item.expression.expression.clone(), item.ascending))
1096            .collect();
1097        result.push(LogicalOperator::OrderBy(LogicalOrderBy {
1098            sort_keys,
1099            children: Vec::new(),
1100            cardinality: 0,
1101        }));
1102    }
1103
1104    // Insert LIMIT/SKIP operator if present
1105    if limit.is_some() || skip.is_some() {
1106        result.push(LogicalOperator::Limit(LogicalLimit {
1107            limit: limit.unwrap_or(u64::MAX),
1108            offset: skip.unwrap_or(0),
1109            children: Vec::new(),
1110            cardinality: 0,
1111        }));
1112    }
1113}
1114
1115impl Default for QueryPlanner {
1116    fn default() -> Self {
1117        Self::new()
1118    }
1119}
1120
1121#[cfg(test)]
1122mod tests {
1123    use super::*;
1124    use akar_binder::Binder;
1125    use akar_catalog::{Catalog, CatalogColumn};
1126    use akar_common::types::LogicalTypeID;
1127    use akar_parser::parse;
1128    use std::sync::Arc;
1129
1130    fn setup_binder() -> Binder {
1131        let mut catalog = Catalog::new();
1132        catalog.create_node_table(
1133            "Person".into(),
1134            vec![
1135                CatalogColumn {
1136                    compression: akar_common::enums::CompressionType::Uncompressed,
1137                    name: "name".into(),
1138                    logical_type: LogicalTypeID::String,
1139                    is_primary_key: true,
1140                    default_value: None,
1141                },
1142                CatalogColumn {
1143                    compression: akar_common::enums::CompressionType::Uncompressed,
1144                    name: "age".into(),
1145                    logical_type: LogicalTypeID::Int64,
1146                    is_primary_key: false,
1147                    default_value: None,
1148                },
1149            ],
1150        );
1151        catalog.create_rel_table(
1152            "Knows".into(),
1153            0,
1154            0,
1155            vec![CatalogColumn {
1156                compression: akar_common::enums::CompressionType::Uncompressed,
1157                name: "since".into(),
1158                logical_type: LogicalTypeID::Int64,
1159                is_primary_key: false,
1160                default_value: None,
1161            }],
1162        );
1163        Binder::new(Arc::new(std::sync::Mutex::new(catalog)))
1164    }
1165
1166    #[test]
1167    fn test_plan_match_return() {
1168        let binder = setup_binder();
1169        let sql = "MATCH (a:Person) RETURN a.name";
1170        let stmt = parse(sql).unwrap();
1171        let bound = binder.bind(stmt).unwrap();
1172        let planner = QueryPlanner::new();
1173        let plan = planner.plan(bound).unwrap();
1174        assert!(!plan.is_empty());
1175
1176        // Should have ScanNode + Projection
1177        let scan_count = plan
1178            .iter()
1179            .filter(|op| matches!(op, LogicalOperator::ScanNode(_)))
1180            .count();
1181        let proj_count = plan
1182            .iter()
1183            .filter(|op| matches!(op, LogicalOperator::Projection(_)))
1184            .count();
1185        assert_eq!(scan_count, 1);
1186        assert_eq!(proj_count, 1);
1187    }
1188
1189    #[test]
1190    fn test_plan_optional_match_keeps_order_by_and_limit() {
1191        // P53.40 / kairos Finding #22: the `scan_ops.is_empty()` assembly fast
1192        // path (hit whenever OPTIONAL MATCH resets the scan state) previously
1193        // returned early and dropped ORDER BY and LIMIT entirely.
1194        let binder = setup_binder();
1195        let sql = "MATCH (a:Person) OPTIONAL MATCH (a)-[r:Knows]-(:Person) \
1196                   RETURN a.name AS id, COUNT(r) AS c ORDER BY id LIMIT 5";
1197        let stmt = parse(sql).unwrap();
1198        let bound = binder.bind(stmt).unwrap();
1199        let planner = QueryPlanner::new();
1200        let plan = planner.plan(bound).unwrap();
1201        assert!(
1202            plan.iter().any(|op| matches!(op, LogicalOperator::OrderBy(_))),
1203            "planner must emit OrderBy above the RETURN projection, got: {plan:?}"
1204        );
1205        assert!(
1206            plan.iter().any(|op| matches!(op, LogicalOperator::Limit(_))),
1207            "planner must emit Limit above the RETURN projection, got: {plan:?}"
1208        );
1209    }
1210
1211    #[test]
1212    fn test_plan_return_only_projection() {
1213        let binder = setup_binder();
1214        let sql = "RETURN 1";
1215        let stmt = parse(sql).unwrap();
1216        let bound = binder.bind(stmt).unwrap();
1217        let planner = QueryPlanner::new();
1218        let plan = planner.plan(bound).unwrap();
1219
1220        assert_eq!(plan.len(), 1);
1221        assert!(matches!(plan[0], LogicalOperator::Projection(_)));
1222    }
1223
1224    #[test]
1225    fn test_plan_unwind_then_projection_without_scan() {
1226        let binder = setup_binder();
1227        let sql = "UNWIND [1, 2, 3] AS x RETURN x";
1228        let stmt = parse(sql).unwrap();
1229        let bound = binder.bind(stmt).unwrap();
1230        let planner = QueryPlanner::new();
1231        let plan = planner.plan(bound).unwrap();
1232
1233        assert_eq!(plan.len(), 2);
1234        assert!(matches!(plan[0], LogicalOperator::Unwind(_)));
1235        assert!(matches!(plan[1], LogicalOperator::Projection(_)));
1236    }
1237
1238    #[test]
1239    fn test_plan_match_where_return() {
1240        let binder = setup_binder();
1241        let sql = "MATCH (a:Person) WHERE a.age > 25 RETURN a.name";
1242        let stmt = parse(sql).unwrap();
1243        let bound = binder.bind(stmt).unwrap();
1244        let planner = QueryPlanner::new();
1245        let plan = planner.plan(bound).unwrap();
1246        assert!(!plan.is_empty());
1247
1248        let scan_count = plan
1249            .iter()
1250            .filter(|op| matches!(op, LogicalOperator::ScanNode(_)))
1251            .count();
1252        let filter_count = plan
1253            .iter()
1254            .filter(|op| matches!(op, LogicalOperator::Filter(_)))
1255            .count();
1256        let proj_count = plan
1257            .iter()
1258            .filter(|op| matches!(op, LogicalOperator::Projection(_)))
1259            .count();
1260        assert_eq!(scan_count, 1);
1261        assert_eq!(filter_count, 1);
1262        assert_eq!(proj_count, 1);
1263    }
1264
1265    #[test]
1266    fn test_plan_ddl_empty() {
1267        let binder = Binder::new(Arc::new(std::sync::Mutex::new(Catalog::new())));
1268        let sql = "CREATE NODE TABLE City(name STRING, PRIMARY KEY (name))";
1269        let stmt = parse(sql).unwrap();
1270        let bound = binder.bind(stmt).unwrap();
1271        let planner = QueryPlanner::new();
1272        let plan = planner.plan(bound).unwrap();
1273        assert!(!plan.is_empty()); // DDL now produces a logical plan
1274        match &plan[0] {
1275            LogicalOperator::CreateNodeTable(ct) => {
1276                assert_eq!(ct.name, "City");
1277            }
1278            _ => panic!("Expected CreateNodeTable"),
1279        }
1280    }
1281
1282    #[test]
1283    fn test_plan_scan_node_fields() {
1284        let binder = setup_binder();
1285        let sql = "MATCH (a:Person) RETURN a";
1286        let stmt = parse(sql).unwrap();
1287        let bound = binder.bind(stmt).unwrap();
1288        let planner = QueryPlanner::new();
1289        let plan = planner.plan(bound).unwrap();
1290
1291        match &plan[0] {
1292            LogicalOperator::ScanNode(s) => {
1293                assert_eq!(s.table_name, "Person");
1294                assert_eq!(s.alias, Some("a".into()));
1295            }
1296            _ => panic!("Expected ScanNode"),
1297        }
1298    }
1299
1300    #[test]
1301    fn test_plan_rel_pattern() {
1302        let binder = setup_binder();
1303        let sql = "MATCH (a:Person)-[r:Knows]->(b:Person) RETURN a, b";
1304        let stmt = parse(sql).unwrap();
1305        let bound = binder.bind(stmt).unwrap();
1306        let planner = QueryPlanner::new();
1307        let plan = planner.plan(bound).unwrap();
1308        // Should have an Extend operator replacing the relationship scan + join
1309        assert!(plan.iter().any(|op| matches!(op, LogicalOperator::Extend(_))));
1310        // Should also have ScanNode for the source node
1311        assert!(plan.iter().any(|op| matches!(op, LogicalOperator::ScanNode(_))));
1312        // Should NOT have ScanRel (replaced by Extend)
1313        assert!(!plan.iter().any(|op| matches!(op, LogicalOperator::ScanRel(_))));
1314    }
1315
1316    #[test]
1317    fn test_plan_order() {
1318        let binder = setup_binder();
1319        let sql = "MATCH (a:Person) WHERE a.age > 25 RETURN a.name";
1320        let stmt = parse(sql).unwrap();
1321        let bound = binder.bind(stmt).unwrap();
1322        let planner = QueryPlanner::new();
1323        let plan = planner.plan(bound).unwrap();
1324
1325        // Order should be: scans → filter → projection
1326        let positions: Vec<&str> = plan
1327            .iter()
1328            .map(|op| match op {
1329                LogicalOperator::ScanNode(_) => "scan",
1330                LogicalOperator::Filter(_) => "filter",
1331                LogicalOperator::Projection(_) => "proj",
1332                _ => "other",
1333            })
1334            .collect();
1335
1336        let scan_pos = positions.iter().position(|&p| p == "scan").unwrap();
1337        let filter_pos = positions.iter().position(|&p| p == "filter").unwrap();
1338        let proj_pos = positions.iter().position(|&p| p == "proj").unwrap();
1339
1340        assert!(scan_pos < filter_pos);
1341        assert!(filter_pos < proj_pos);
1342    }
1343}