1use 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
16pub 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
45fn 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
60fn 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
79pub 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 fn plan_explain(&self, e: BoundExplain) -> Result<Vec<LogicalOperator>, PlannerError> {
128 let inner_plan = self.plan(*e.inner)?;
129 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 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 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 fn plan_merge(&self, m: BoundMerge) -> Result<Vec<LogicalOperator>, PlannerError> {
293 Ok(self.build_merge_operators(&m))
294 }
295
296 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 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 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 let mut order_by: Option<Vec<BoundOrderByItem>> = None;
405 let mut limit: Option<u64> = None;
406 let mut skip: Option<u64> = None;
407 let mut skip_next_node = false;
409 let mut available_vars: HashSet<String> = HashSet::new();
413 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 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 let skip_current_node_scan = skip_next_node;
446 skip_next_node = false;
447
448 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 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 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 if !target_var.is_empty() {
511 available_vars.insert(target_var);
512 }
513 skip_next_node = true;
515 continue;
516 }
517
518 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 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 if !dst_var.is_empty() {
563 available_vars.insert(dst_var);
564 }
565
566 skip_next_node = true;
568 continue;
569 }
570 }
571
572 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 } }
595
596 BoundClause::BoundWhere(w) => {
597 if !delete_exprs.is_empty() {
598 delete_exprs.push(LogicalOperator::Filter(LogicalFilter {
602 expression: w.expression.expression,
603 children: Vec::new(),
604 cardinality: 0,
605 }));
606 } else {
607 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 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 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 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 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 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 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 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 delete_exprs.push(LogicalOperator::OptionalMatch(LogicalOptionalMatch {
833 left: Box::new(left_op),
834 right: Box::new(right_op),
835 cardinality: 0,
836 }));
837 }
838 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 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 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 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 let delete_ops: Vec<LogicalOperator> = std::mem::take(&mut delete_exprs);
972
973 let mut result: Vec<LogicalOperator> = Vec::new();
975
976 if scan_ops.is_empty() {
977 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 result.push(scan_ops.into_iter().next().unwrap());
989 } else {
990 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 result.extend(std::mem::take(&mut extend_ops));
998
999 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 result.extend(delete_ops);
1012
1013 append_return_tail(&mut result, projection, order_by, distinct, limit, skip);
1015
1016 Ok(result)
1017 }
1018}
1019
1020fn 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 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 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 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 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 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 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()); 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 assert!(plan.iter().any(|op| matches!(op, LogicalOperator::Extend(_))));
1310 assert!(plan.iter().any(|op| matches!(op, LogicalOperator::ScanNode(_))));
1312 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 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}