1use super::{
16 access_path::index_seek_from_filter,
17 costing::{
18 estimate_node_cartesian_product_cost, estimate_physical_plan_cost,
19 push_optional_relationship_count_sum_cost_decision,
20 },
21 logical_rewrite::{rewrite_logical_plan, LogicalRewriteOutput},
22 properties::satisfies_ordering,
23 selected_trace::selected_plan_trace,
24 stages::{
25 ACCESS_PATH_SELECTION_STAGE, DIRECT_PHYSICAL_FALLBACK_STAGE, LOGICAL_GROUPING_STAGE,
26 PHYSICAL_SEARCH_STAGE, PLAN_FINALIZATION_STAGE, SELECTED_PLAN_COSTING_STAGE,
27 },
28 LogicalPlanRoot, LoweringReadyLogicalPlanRoot, OptimizationSearchReport, OptimizerCatalog,
29 OptimizerConfig, OptimizerTrace, PhysicalPlan, PhysicalPlanRoot, StageStats,
30};
31use crate::{
32 GroupId, Memo, OptimizerContext, OptimizerSearchDirective, OptimizerSearchDirectiveError,
33 RuleEvent, RuleOutcome, SelectedPlanTrace, StageTrace,
34};
35use hawdb_core::Value;
36use hawdb_cypher::RelationshipDirection;
37use hawdb_plan_cypher::{
38 AggregateFunction, AggregateTarget, GraphExpansionBudget, LogicalPlan, NodeProjectionAccess,
39 Predicate, Projection, ProjectionExpression, SortItem, SortKey,
40};
41use std::collections::{BTreeMap, BTreeSet};
42
43mod access;
44mod ddl;
45mod join;
46mod mutation;
47mod procedure;
48mod simple;
49mod traversal;
50
51#[cfg(test)]
52mod tests;
53
54type GraphMemo = Memo<GroupExpr>;
55
56const GRAPH_EXPANSION_FANOUT_PER_SEED_HOP: usize = 64;
57const GRAPH_EXPANSION_MAX_CANDIDATES: usize = 8_192;
58const GRAPH_EXPANSION_PAYLOAD_BYTE_LIMIT: usize = 8 * 1024 * 1024;
59
60#[derive(Debug)]
61struct GroupExpr {
62 logical: LogicalPlan,
63 children: Vec<GroupId>,
64}
65
66#[derive(Debug, Default, Clone)]
67pub struct CascadesOptimizer {
68 context: OptimizerContext,
69}
70
71impl CascadesOptimizer {
72 pub fn new(config: OptimizerConfig) -> Self {
73 Self {
74 context: OptimizerContext::from_config(config),
75 }
76 }
77
78 pub fn with_context(context: OptimizerContext) -> Self {
79 Self { context }
80 }
81
82 pub fn context(&self) -> &OptimizerContext {
83 &self.context
84 }
85
86 pub fn refresh_trace_for_physical_plan(
87 &self,
88 trace: &mut OptimizerTrace,
89 plan: &PhysicalPlan,
90 catalog: &OptimizerCatalog,
91 ) {
92 refresh_selected_plan_trace(trace, selected_plan_trace(plan, catalog, &self.context));
93 }
94
95 pub fn optimize(&self, logical: &LogicalPlan) -> PhysicalPlan {
96 self.optimize_root(&LogicalPlanRoot::new(logical.clone()))
97 .into_parts()
98 .0
99 }
100
101 pub fn optimize_with_trace(&self, logical: &LogicalPlan) -> (PhysicalPlan, OptimizerTrace) {
102 self.optimize_root(&LogicalPlanRoot::new(logical.clone()))
103 .into_parts()
104 }
105
106 pub fn optimize_with_catalog(
107 &self,
108 logical: &LogicalPlan,
109 catalog: &OptimizerCatalog,
110 ) -> (PhysicalPlan, OptimizerTrace) {
111 self.optimize_root_with_catalog(&LogicalPlanRoot::new(logical.clone()), catalog)
112 .into_parts()
113 }
114
115 pub fn optimize_root(&self, root: &LogicalPlanRoot) -> PhysicalPlanRoot {
116 self.optimize_root_with_catalog(root, &OptimizerCatalog::optimistic())
117 }
118
119 pub fn optimize_root_with_catalog(
120 &self,
121 root: &LogicalPlanRoot,
122 catalog: &OptimizerCatalog,
123 ) -> PhysicalPlanRoot {
124 self.optimize_lowering_ready_root_with_catalog_and_directive(
125 &root.clone().into_lowering_ready(),
126 catalog,
127 OptimizerSearchDirective::Auto,
128 )
129 .expect("automatic optimizer search cannot reject its directive")
130 }
131
132 pub fn optimize_lowering_ready_root_with_catalog(
133 &self,
134 root: &LoweringReadyLogicalPlanRoot,
135 catalog: &OptimizerCatalog,
136 ) -> PhysicalPlanRoot {
137 self.optimize_lowering_ready_root_with_catalog_and_directive(
138 root,
139 catalog,
140 OptimizerSearchDirective::Auto,
141 )
142 .expect("automatic optimizer search cannot reject its directive")
143 }
144
145 pub fn optimize_root_with_catalog_and_directive(
146 &self,
147 root: &LogicalPlanRoot,
148 catalog: &OptimizerCatalog,
149 directive: OptimizerSearchDirective,
150 ) -> Result<PhysicalPlanRoot, OptimizerSearchDirectiveError> {
151 self.optimize_lowering_ready_root_with_catalog_and_directive(
152 &root.clone().into_lowering_ready(),
153 catalog,
154 directive,
155 )
156 }
157
158 pub fn optimize_lowering_ready_root_with_catalog_and_directive(
159 &self,
160 root: &LoweringReadyLogicalPlanRoot,
161 catalog: &OptimizerCatalog,
162 directive: OptimizerSearchDirective,
163 ) -> Result<PhysicalPlanRoot, OptimizerSearchDirectiveError> {
164 let rewrite = rewrite_logical_plan(root.plan());
165 let logical = rewrite.plan();
166 let required_groups = logical_group_count(logical);
167 let max_groups = self.context.optimizer_config().max_groups;
168 if directive == OptimizerSearchDirective::Memo && required_groups > max_groups {
169 return Err(OptimizerSearchDirectiveError::MemoGroupBudgetExceeded {
170 required_groups,
171 max_groups,
172 });
173 }
174 if directive == OptimizerSearchDirective::DirectFallback
175 || (directive == OptimizerSearchDirective::Auto && required_groups > max_groups)
176 {
177 let mut decisions = Vec::new();
178 let mut stage_events = Vec::new();
179 let plan = lower_logical(
180 logical,
181 LoweringChildren::Direct,
182 catalog,
183 &self.context,
184 &mut decisions,
185 &mut stage_events,
186 );
187 let plan = finalize_physical_plan(plan, &mut decisions, &mut stage_events);
188 let mut report = if directive == OptimizerSearchDirective::DirectFallback {
189 OptimizationSearchReport::forced_direct_fallback(required_groups)
190 } else {
191 OptimizationSearchReport::direct_fallback(required_groups, max_groups)
192 };
193 record_logical_rewrite(&mut report, &rewrite);
194 report.push_stage_event(
195 LOGICAL_GROUPING_STAGE.trace(StageStats::new(1, required_groups)),
196 );
197 for event in stage_events {
198 report.push_stage_event(event);
199 }
200 report.push_stage_event(
201 DIRECT_PHYSICAL_FALLBACK_STAGE.trace(StageStats::new(required_groups, 1)),
202 );
203 report.extend_decisions(decisions);
204 let selected = selected_plan_trace(&plan, catalog, &self.context);
205 report.push_stage_event(SELECTED_PLAN_COSTING_STAGE.trace(StageStats::new(1, 1)));
206 report.record_selected_plan_cost(selected.cost);
207 return Ok(PhysicalPlanRoot::new(plan, report.into_trace(selected)));
208 }
209 let mut memo = GraphMemo::default();
210 let root = insert_logical_group(&mut memo, logical);
211 let mut decisions = Vec::new();
212 let mut stage_events = Vec::new();
213 let plan = best_physical(
214 &memo,
215 root,
216 catalog,
217 &self.context,
218 &mut decisions,
219 &mut stage_events,
220 );
221 let plan = finalize_physical_plan(plan, &mut decisions, &mut stage_events);
222 let mut report = OptimizationSearchReport::memo(memo.group_count());
223 record_logical_rewrite(&mut report, &rewrite);
224 report
225 .push_stage_event(LOGICAL_GROUPING_STAGE.trace(StageStats::new(1, memo.group_count())));
226 let (applied_rules, skipped_rules) = stage_rule_counts(&stage_events);
227 for event in stage_events {
228 report.push_stage_event(event);
229 }
230 report.push_stage_event(PHYSICAL_SEARCH_STAGE.trace(
231 StageStats::new(memo.group_count(), 1).with_rule_counts(applied_rules, skipped_rules),
232 ));
233 report.extend_decisions(decisions);
234 let selected = selected_plan_trace(&plan, catalog, &self.context);
235 report.push_stage_event(SELECTED_PLAN_COSTING_STAGE.trace(StageStats::new(1, 1)));
236 report.record_selected_plan_cost(selected.cost);
237 if directive == OptimizerSearchDirective::Memo {
238 report.push_decision("selected memo search: explicit optimizer search directive");
239 }
240 Ok(PhysicalPlanRoot::new(plan, report.into_trace(selected)))
241 }
242}
243
244fn refresh_selected_plan_trace(trace: &mut OptimizerTrace, selected: SelectedPlanTrace) {
245 let selected_cost = selected.cost.with_cardinality_floor();
246 let selected_cost_breakdown = selected.cost_breakdown.with_cardinality_floor();
247 let cost_detail = format!(
248 "estimated_rows={} cost={}",
249 selected_cost.estimated_rows, selected_cost.cost
250 );
251 let cost_decision = format!("selected physical plan cost: {cost_detail}");
252 if let Some(decision) = trace
253 .decisions
254 .iter_mut()
255 .find(|decision| decision.starts_with("selected physical plan cost: "))
256 {
257 *decision = cost_decision;
258 } else {
259 trace.decisions.push(cost_decision);
260 }
261 let cost_event = RuleEvent::selected("physical plan cost", cost_detail);
262 if let Some(event) = trace.rule_events.iter_mut().find(|event| {
263 event.rule() == "physical plan cost" && event.outcome() == RuleOutcome::Selected
264 }) {
265 *event = cost_event;
266 } else {
267 trace.rule_events.push(cost_event);
268 }
269
270 trace.query_digest = selected.query_digest;
271 trace.selected_plan = selected.explain;
272 trace.selected_plan_fingerprint = selected.fingerprint;
273 trace.selected_plan_cost = selected_cost;
274 trace.selected_plan_cost_breakdown = selected_cost_breakdown;
275 trace.selected_plan_properties = selected.properties;
276 trace.selected_plan_cardinality_estimates = selected.cardinality_estimates;
277 trace.selected_plan_operator_counts = selected.operator_counts;
278 trace.selected_plan_class_counts = selected.class_counts;
279}
280
281pub(super) fn record_logical_rewrite(
282 report: &mut OptimizationSearchReport,
283 rewrite: &LogicalRewriteOutput,
284) {
285 report.push_stage_event(rewrite.trace().clone());
286 if let Some(warning) = rewrite.warning() {
287 report.push_warning(warning);
288 }
289 for event in rewrite.events().iter().cloned() {
290 report.push_rule_event(event);
291 }
292}
293
294fn insert_logical_group(memo: &mut GraphMemo, logical: &LogicalPlan) -> GroupId {
295 let expr = GroupExpr::from_logical(logical, memo);
296 memo.insert_group(expr)
297}
298
299fn best_physical(
300 memo: &GraphMemo,
301 root: GroupId,
302 catalog: &OptimizerCatalog,
303 optimizer_context: &OptimizerContext,
304 decisions: &mut Vec<String>,
305 stage_events: &mut Vec<StageTrace>,
306) -> PhysicalPlan {
307 let group = memo.group(root).expect("memo group id should exist");
308 group
309 .first_expression()
310 .expect("memo group should contain at least one expression")
311 .to_physical(memo, catalog, optimizer_context, decisions, stage_events)
312}
313
314fn select_bounded_sort_plan(
315 items: Vec<SortItem>,
316 offset: usize,
317 limit: usize,
318 input: PhysicalPlan,
319 catalog: &OptimizerCatalog,
320 decisions: &mut Vec<String>,
321) -> PhysicalPlan {
322 if satisfies_ordering(&input, &items, catalog) {
323 decisions.push("elide bounded sort: access path satisfies required ordering".to_string());
324 return PhysicalPlan::LimitExec {
325 offset,
326 limit: Some(limit),
327 input: Box::new(input),
328 };
329 }
330 let top_n = PhysicalPlan::TopNExec {
331 items: items.clone(),
332 offset,
333 limit,
334 input: Box::new(input.clone()),
335 };
336 let sort_limit = PhysicalPlan::LimitExec {
337 offset,
338 limit: Some(limit),
339 input: Box::new(PhysicalPlan::SortExec {
340 items,
341 input: Box::new(input),
342 }),
343 };
344 let top_n_cost = estimate_physical_plan_cost(&top_n, catalog);
345 let sort_limit_cost = estimate_physical_plan_cost(&sort_limit, catalog);
346 if top_n_cost.cost < sort_limit_cost.cost {
347 decisions.push(format!(
348 "choose TopN for bounded sort: offset={offset} limit={limit} top_n_cost={} sort_limit_cost={}",
349 top_n_cost.cost, sort_limit_cost.cost
350 ));
351 top_n
352 } else {
353 decisions.push(format!(
354 "keep Sort + Limit for bounded sort: offset={offset} limit={limit} top_n_cost={} sort_limit_cost={}",
355 top_n_cost.cost, sort_limit_cost.cost
356 ));
357 sort_limit
358 }
359}
360
361fn stage_rule_counts(stage_events: &[StageTrace]) -> (usize, usize) {
362 stage_events
363 .iter()
364 .fold((0, 0), |(applied, skipped), event| {
365 let stats = event.stats();
366 (
367 applied.saturating_add(stats.applied_rules),
368 skipped.saturating_add(stats.skipped_rules),
369 )
370 })
371}
372
373impl GroupExpr {
374 fn from_logical(logical: &LogicalPlan, memo: &mut GraphMemo) -> Self {
375 match logical {
376 LogicalPlan::GraphMatch { input, .. } => Self {
377 logical: logical.clone(),
378 children: input
379 .iter()
380 .map(|input| insert_logical_group(memo, input))
381 .collect(),
382 },
383 LogicalPlan::CreateNodeLabel { .. }
384 | LogicalPlan::CreateRelationshipType { .. }
385 | LogicalPlan::CreateNodeTable { .. }
386 | LogicalPlan::CreateRelationshipTable { .. }
387 | LogicalPlan::CreateProperty { .. }
388 | LogicalPlan::AlterTableState { .. }
389 | LogicalPlan::AlterPropertyState { .. }
390 | LogicalPlan::CreateIndex { .. }
391 | LogicalPlan::CreateCompositeIndex { .. }
392 | LogicalPlan::CreateRangeIndex { .. }
393 | LogicalPlan::CreateFullTextIndex { .. }
394 | LogicalPlan::CreateUniqueConstraint { .. }
395 | LogicalPlan::CreateNodePropertyExistsConstraint { .. }
396 | LogicalPlan::CreateRelationshipUniqueConstraint { .. }
397 | LogicalPlan::CreateRelationshipPropertyExistsConstraint { .. }
398 | LogicalPlan::ProjectGraph { .. }
399 | LogicalPlan::GraphAlgorithm { .. }
400 | LogicalPlan::VectorSeed { .. }
401 | LogicalPlan::TextSeed { .. }
402 | LogicalPlan::GraphSeed { .. }
403 | LogicalPlan::CreateNode { .. }
404 | LogicalPlan::UnwindMutation { .. }
405 | LogicalPlan::MergeNode { .. }
406 | LogicalPlan::MergeRelationship { .. }
407 | LogicalPlan::MergeMatchedRelationship { .. }
408 | LogicalPlan::MergeRelationshipFromMatchedRelationship { .. }
409 | LogicalPlan::MergeRelationshipToMatchedTarget { .. }
410 | LogicalPlan::MergeRelationshipFromMatchedTarget { .. }
411 | LogicalPlan::CreateMatchedRelationship { .. }
412 | LogicalPlan::SetNodeProperty { .. }
413 | LogicalPlan::SetNodeProperties { .. }
414 | LogicalPlan::SetNodePropertiesReturn { .. }
415 | LogicalPlan::SetRelationshipProperty { .. }
416 | LogicalPlan::SetRelationshipProperties { .. }
417 | LogicalPlan::DeleteNode { .. }
418 | LogicalPlan::DeleteRelationship { .. }
419 | LogicalPlan::DeleteRelationshipTargetNodes { .. }
420 | LogicalPlan::CreateRelationship { .. }
421 | LogicalPlan::NodeScan { .. }
422 | LogicalPlan::OptionalRelationshipCountSum { .. }
423 | LogicalPlan::ThreadRepairStats { .. }
424 | LogicalPlan::ShortestPath { .. } => Self {
425 logical: logical.clone(),
426 children: Vec::new(),
427 },
428 LogicalPlan::Limit { limit: Some(0), .. } => Self {
429 logical: logical.clone(),
430 children: Vec::new(),
431 },
432 LogicalPlan::NodeCartesianProduct { left, right } => Self {
433 logical: logical.clone(),
434 children: vec![
435 insert_logical_group(memo, left),
436 insert_logical_group(memo, right),
437 ],
438 },
439 LogicalPlan::Limit {
440 limit: Some(_),
441 input,
442 ..
443 } if matches!(input.as_ref(), LogicalPlan::Sort { .. }) => {
444 let LogicalPlan::Sort { input, .. } = input.as_ref() else {
445 unreachable!("guard requires a sort input");
446 };
447 Self {
448 logical: logical.clone(),
449 children: vec![insert_logical_group(memo, input)],
450 }
451 }
452 LogicalPlan::Expand { input, .. }
453 | LogicalPlan::NodeColumnLookup { input, .. }
454 | LogicalPlan::OptionalDegree { input, .. }
455 | LogicalPlan::Filter { input, .. }
456 | LogicalPlan::Project { input, .. }
457 | LogicalPlan::Aggregate { input, .. }
458 | LogicalPlan::Distinct { input }
459 | LogicalPlan::Sort { input, .. }
460 | LogicalPlan::Limit { input, .. } => Self {
461 logical: logical.clone(),
462 children: vec![insert_logical_group(memo, input)],
463 },
464 }
465 }
466
467 fn to_physical(
468 &self,
469 memo: &GraphMemo,
470 catalog: &OptimizerCatalog,
471 optimizer_context: &OptimizerContext,
472 decisions: &mut Vec<String>,
473 stage_events: &mut Vec<StageTrace>,
474 ) -> PhysicalPlan {
475 lower_logical(
476 &self.logical,
477 LoweringChildren::Memo {
478 memo,
479 children: &self.children,
480 },
481 catalog,
482 optimizer_context,
483 decisions,
484 stage_events,
485 )
486 }
487}
488
489fn push_seed_metadata_filter(
490 input: &mut PhysicalPlan,
491 predicate: &hawdb_plan_cypher::Predicate,
492 decisions: &mut Vec<String>,
493) {
494 if let hawdb_plan_cypher::Predicate::And(predicates) = predicate {
495 for predicate in predicates {
496 push_seed_metadata_filter(input, predicate, decisions);
497 }
498 return;
499 }
500 let hawdb_plan_cypher::Predicate::PropertyEq {
501 variable,
502 property,
503 value,
504 } = predicate
505 else {
506 return;
507 };
508 let Some(filter_field) = seed_metadata_field(property) else {
509 return;
510 };
511 let Some(filter_value) = seed_metadata_value(value) else {
512 return;
513 };
514 let producer = seed_window(input)
515 .map(|(producer, _)| producer)
516 .unwrap_or("retriever");
517 if attach_seed_metadata_filter(input, variable, filter_field, filter_value) {
518 decisions.push(format!(
519 "push descriptor-safe {producer} seed filter {variable}.{property} before candidate generation"
520 ));
521 }
522}
523
524fn attach_seed_metadata_filter(
525 plan: &mut PhysicalPlan,
526 variable: &str,
527 field: &str,
528 value: String,
529) -> bool {
530 match plan {
531 PhysicalPlan::NodeColumnLookupExec {
532 variable: lookup_variable,
533 input,
534 ..
535 } if lookup_variable == variable => attach_metadata_filter_to_seed(input, field, value),
536 PhysicalPlan::AdjacencyExpandExec {
537 source_variable,
538 input,
539 ..
540 } if source_variable == variable => {
541 attach_seed_metadata_filter(input, variable, field, value)
542 }
543 _ => false,
544 }
545}
546
547fn attach_metadata_filter_to_seed(plan: &mut PhysicalPlan, field: &str, value: String) -> bool {
548 if let PhysicalPlan::TextSeedScan {
549 metadata_filters, ..
550 } = plan
551 {
552 if metadata_filters
553 .get(field)
554 .is_some_and(|existing| existing != &value)
555 {
556 return false;
557 }
558 metadata_filters.insert(field.to_string(), value);
559 return true;
560 }
561 let PhysicalPlan::VectorSeedScan {
562 metadata_filters,
563 vector_plan,
564 ..
565 } = plan
566 else {
567 return false;
568 };
569 if metadata_filters
570 .get(field)
571 .is_some_and(|existing| existing != &value)
572 {
573 return false;
574 }
575 metadata_filters.insert(field.to_string(), value);
576 attach_vector_filter_field(vector_plan, field);
577 true
578}
579
580fn attach_vector_filter_field(plan: &mut hawdb_plan_cypher::VectorPhysicalPlan, field: &str) {
581 match plan {
582 hawdb_plan_cypher::VectorPhysicalPlan::Filter { fields } => {
583 if !fields.iter().any(|existing| existing == field) {
584 fields.push(field.to_string());
585 fields.sort();
586 }
587 }
588 hawdb_plan_cypher::VectorPhysicalPlan::VectorCandidateScan { input, .. }
589 | hawdb_plan_cypher::VectorPhysicalPlan::ResidualFilter { input, .. }
590 | hawdb_plan_cypher::VectorPhysicalPlan::RawVectorRerank { input, .. }
591 | hawdb_plan_cypher::VectorPhysicalPlan::TopK { input, .. } => {
592 attach_vector_filter_field(input, field);
593 }
594 }
595}
596
597fn seed_metadata_field(property: &str) -> Option<&str> {
598 match property {
599 "id" => Some("external_id"),
600 "kind" | "external_id" | "source_id" | "space_id" | "unit_type" | "lifecycle_state"
601 | "importance" | "confidence" | "created_at" | "updated_at" | "event_start"
602 | "event_end" | "is_latest" => Some(property),
603 _ => None,
604 }
605}
606
607fn seed_metadata_value(value: &Value) -> Option<String> {
608 match value {
609 Value::String(value) => Some(value.clone()),
610 Value::Bool(value) => Some(value.to_string()),
611 Value::Int(value) => Some(value.to_string()),
612 Value::Float(value) if value.is_finite() => Some(value.to_string()),
613 Value::Uuid(value) => Some(value.to_string()),
614 Value::Null | Value::Float(_) | Value::Binary(_) | Value::List(_) | Value::Map(_) => None,
615 }
616}
617
618fn seed_window(plan: &PhysicalPlan) -> Option<(&'static str, usize)> {
619 match plan {
620 PhysicalPlan::VectorSeedScan { vector_plan, .. } => {
621 vector_plan_top_k(vector_plan).map(|window| ("vector", window))
622 }
623 PhysicalPlan::TextSeedScan { top_k, .. } => Some(("text", *top_k)),
624 PhysicalPlan::GraphSeedScan { top_k, .. } => Some(("graph", *top_k)),
625 PhysicalPlan::NodeColumnLookupExec { input, .. }
626 | PhysicalPlan::AdjacencyExpandExec { input, .. }
627 | PhysicalPlan::AdjacencyExistsExec { input, .. }
628 | PhysicalPlan::FilterExec { input, .. }
629 | PhysicalPlan::ProjectExec { input, .. }
630 | PhysicalPlan::LimitExec { input, .. } => seed_window(input),
631 _ => None,
632 }
633}
634
635fn vector_plan_top_k(plan: &hawdb_plan_cypher::VectorPhysicalPlan) -> Option<usize> {
636 match plan {
637 hawdb_plan_cypher::VectorPhysicalPlan::TopK { limit, .. } => Some(*limit),
638 hawdb_plan_cypher::VectorPhysicalPlan::VectorCandidateScan { input, .. }
639 | hawdb_plan_cypher::VectorPhysicalPlan::ResidualFilter { input, .. }
640 | hawdb_plan_cypher::VectorPhysicalPlan::RawVectorRerank { input, .. } => {
641 vector_plan_top_k(input)
642 }
643 hawdb_plan_cypher::VectorPhysicalPlan::Filter { .. } => None,
644 }
645}
646
647fn graph_expansion_budget(top_k: usize, max_hops: usize) -> GraphExpansionBudget {
648 let candidate_limit = top_k
649 .max(1)
650 .saturating_mul(max_hops.max(1))
651 .saturating_mul(GRAPH_EXPANSION_FANOUT_PER_SEED_HOP)
652 .min(GRAPH_EXPANSION_MAX_CANDIDATES)
653 .max(top_k);
654 GraphExpansionBudget {
655 candidate_limit,
656 payload_byte_limit: GRAPH_EXPANSION_PAYLOAD_BYTE_LIMIT,
657 }
658}
659
660fn logical_group_count(logical: &LogicalPlan) -> usize {
661 match logical {
662 LogicalPlan::GraphMatch { input, .. } => {
663 1 + input.as_deref().map(logical_group_count).unwrap_or(0)
664 }
665 LogicalPlan::Limit { limit: Some(0), .. } => 1,
666 LogicalPlan::Limit {
667 limit: Some(_),
668 input,
669 ..
670 } if matches!(input.as_ref(), LogicalPlan::Sort { .. }) => {
671 let LogicalPlan::Sort { input, .. } = input.as_ref() else {
672 unreachable!("guard requires a sort input");
673 };
674 1 + logical_group_count(input)
676 }
677 LogicalPlan::Expand { input, .. }
678 | LogicalPlan::NodeColumnLookup { input, .. }
679 | LogicalPlan::OptionalDegree { input, .. }
680 | LogicalPlan::Filter { input, .. }
681 | LogicalPlan::Project { input, .. }
682 | LogicalPlan::Aggregate { input, .. }
683 | LogicalPlan::Distinct { input }
684 | LogicalPlan::Sort { input, .. }
685 | LogicalPlan::Limit { input, .. } => 1 + logical_group_count(input),
686 LogicalPlan::OptionalRelationshipCountSum { .. }
687 | LogicalPlan::ThreadRepairStats { .. }
688 | LogicalPlan::ShortestPath { .. } => 1,
689 LogicalPlan::CreateNodeLabel { .. }
690 | LogicalPlan::CreateRelationshipType { .. }
691 | LogicalPlan::CreateNodeTable { .. }
692 | LogicalPlan::CreateRelationshipTable { .. }
693 | LogicalPlan::CreateProperty { .. }
694 | LogicalPlan::AlterTableState { .. }
695 | LogicalPlan::AlterPropertyState { .. }
696 | LogicalPlan::CreateIndex { .. }
697 | LogicalPlan::CreateCompositeIndex { .. }
698 | LogicalPlan::CreateRangeIndex { .. }
699 | LogicalPlan::CreateFullTextIndex { .. }
700 | LogicalPlan::CreateUniqueConstraint { .. }
701 | LogicalPlan::CreateNodePropertyExistsConstraint { .. }
702 | LogicalPlan::CreateRelationshipUniqueConstraint { .. }
703 | LogicalPlan::CreateRelationshipPropertyExistsConstraint { .. }
704 | LogicalPlan::ProjectGraph { .. }
705 | LogicalPlan::GraphAlgorithm { .. }
706 | LogicalPlan::VectorSeed { .. }
707 | LogicalPlan::TextSeed { .. }
708 | LogicalPlan::GraphSeed { .. }
709 | LogicalPlan::CreateNode { .. }
710 | LogicalPlan::UnwindMutation { .. }
711 | LogicalPlan::MergeNode { .. }
712 | LogicalPlan::MergeRelationship { .. }
713 | LogicalPlan::MergeMatchedRelationship { .. }
714 | LogicalPlan::MergeRelationshipFromMatchedRelationship { .. }
715 | LogicalPlan::MergeRelationshipToMatchedTarget { .. }
716 | LogicalPlan::MergeRelationshipFromMatchedTarget { .. }
717 | LogicalPlan::CreateMatchedRelationship { .. }
718 | LogicalPlan::SetNodeProperty { .. }
719 | LogicalPlan::SetNodeProperties { .. }
720 | LogicalPlan::SetNodePropertiesReturn { .. }
721 | LogicalPlan::SetRelationshipProperty { .. }
722 | LogicalPlan::SetRelationshipProperties { .. }
723 | LogicalPlan::DeleteNode { .. }
724 | LogicalPlan::DeleteRelationship { .. }
725 | LogicalPlan::DeleteRelationshipTargetNodes { .. }
726 | LogicalPlan::CreateRelationship { .. }
727 | LogicalPlan::NodeScan { .. } => 1,
728 LogicalPlan::NodeCartesianProduct { left, right } => {
729 1 + logical_group_count(left) + logical_group_count(right)
730 }
731 }
732}
733
734enum LoweringChildren<'a> {
737 Memo {
738 memo: &'a GraphMemo,
739 children: &'a [GroupId],
740 },
741 Direct,
742}
743
744impl LoweringChildren<'_> {
745 fn lower(
746 &self,
747 logical: &LogicalPlan,
748 index: usize,
749 catalog: &OptimizerCatalog,
750 optimizer_context: &OptimizerContext,
751 decisions: &mut Vec<String>,
752 stage_events: &mut Vec<StageTrace>,
753 ) -> PhysicalPlan {
754 match self {
755 Self::Memo { memo, children } => best_physical(
756 memo,
757 children[index],
758 catalog,
759 optimizer_context,
760 decisions,
761 stage_events,
762 ),
763 Self::Direct => lower_logical(
764 logical,
765 Self::Direct,
766 catalog,
767 optimizer_context,
768 decisions,
769 stage_events,
770 ),
771 }
772 }
773}
774
775fn lower_logical(
776 logical: &LogicalPlan,
777 children: LoweringChildren<'_>,
778 catalog: &OptimizerCatalog,
779 optimizer_context: &OptimizerContext,
780 decisions: &mut Vec<String>,
781 stage_events: &mut Vec<StageTrace>,
782) -> PhysicalPlan {
783 if let Some(plan) = select_exact_count_fast_path(logical, decisions, stage_events) {
784 return plan;
785 }
786 if let Some(plan) = simple::lower_simple_logical(logical, optimizer_context) {
787 return plan;
788 }
789 match logical {
790 LogicalPlan::GraphMatch { program, input } => PhysicalPlan::GraphMatchExec {
791 program: program.clone(),
792 input: input.as_ref().map(|input| {
793 Box::new(children.lower(
794 input,
795 0,
796 catalog,
797 optimizer_context,
798 decisions,
799 stage_events,
800 ))
801 }),
802 },
803 LogicalPlan::NodeCartesianProduct { left, right } => {
804 let left = children.lower(left, 0, catalog, optimizer_context, decisions, stage_events);
805 let right = children.lower(
806 right,
807 1,
808 catalog,
809 optimizer_context,
810 decisions,
811 stage_events,
812 );
813 let (left, right) =
814 order_single_row_cartesian_product_children(catalog, decisions, left, right);
815 push_cartesian_product_cost_decision(catalog, decisions, &left, &right);
816 PhysicalPlan::NodeCartesianProductExec {
817 left: Box::new(left),
818 right: Box::new(right),
819 }
820 }
821 LogicalPlan::NodeColumnLookup {
822 variable,
823 label,
824 property,
825 column,
826 optional,
827 node_visibility_predicate,
828 input,
829 } => PhysicalPlan::NodeColumnLookupExec {
830 variable: variable.clone(),
831 label: label.clone(),
832 property: property.clone(),
833 column: column.clone(),
834 optional: *optional,
835 node_visibility_predicate: node_visibility_predicate.clone(),
836 input: Box::new(children.lower(
837 input,
838 0,
839 catalog,
840 optimizer_context,
841 decisions,
842 stage_events,
843 )),
844 },
845 LogicalPlan::Expand {
846 source_variable,
847 source_label,
848 rel_variable,
849 rel_type,
850 rel_properties,
851 direction,
852 target_variable,
853 target_label,
854 min_hops,
855 max_hops,
856 optional,
857 input,
858 } => {
859 push_expand_estimate_decision(
860 catalog,
861 decisions,
862 ExpandEstimateRequest {
863 source_label,
864 rel_type,
865 rel_properties,
866 target_label,
867 min_hops: *min_hops,
868 max_hops: *max_hops,
869 },
870 );
871 let input = children.lower(
872 input,
873 0,
874 catalog,
875 optimizer_context,
876 decisions,
877 stage_events,
878 );
879 let graph_budget =
880 seed_window(&input).map(|(_, top_k)| graph_expansion_budget(top_k, *max_hops));
881 if let Some(graph_budget) = graph_budget {
882 let producer = seed_window(&input)
883 .map(|(producer, _)| producer)
884 .unwrap_or("retriever");
885 decisions.push(format!(
886 "bound {producer}-seeded graph expansion to {} candidates and {} payload bytes",
887 graph_budget.candidate_limit, graph_budget.payload_byte_limit
888 ));
889 }
890 PhysicalPlan::AdjacencyExpandExec {
891 source_variable: source_variable.clone(),
892 source_label: source_label.clone(),
893 rel_variable: rel_variable.clone(),
894 rel_type: rel_type.clone(),
895 rel_properties: rel_properties.clone(),
896 direction: *direction,
897 target_variable: target_variable.clone(),
898 target_label: target_label.clone(),
899 min_hops: *min_hops,
900 max_hops: *max_hops,
901 optional: *optional,
902 graph_budget,
903 input: Box::new(input),
904 }
905 }
906 LogicalPlan::OptionalDegree {
907 source_variable,
908 rel_type,
909 rel_properties,
910 direction,
911 target_label,
912 target_properties,
913 alias,
914 input,
915 } => PhysicalPlan::OptionalDegreeExec {
916 source_variable: source_variable.clone(),
917 rel_type: rel_type.clone(),
918 rel_properties: rel_properties.clone(),
919 direction: *direction,
920 target_label: target_label.clone(),
921 target_properties: target_properties.clone(),
922 alias: alias.clone(),
923 input: Box::new(children.lower(
924 input,
925 0,
926 catalog,
927 optimizer_context,
928 decisions,
929 stage_events,
930 )),
931 },
932 LogicalPlan::OptionalRelationshipCountSum {
933 variable,
934 label,
935 properties,
936 legs,
937 output,
938 } => {
939 push_optional_relationship_count_sum_cost_decision(
940 catalog, decisions, label, properties, legs,
941 );
942 PhysicalPlan::OptionalRelationshipCountSumExec {
943 variable: variable.clone(),
944 label: label.clone(),
945 properties: properties.clone(),
946 legs: legs.clone(),
947 output: output.clone(),
948 }
949 }
950 LogicalPlan::Filter { predicate, input } => {
951 if let Some(plan) =
952 index_seek_from_filter(predicate, input, catalog, decisions, stage_events)
953 {
954 plan
955 } else if let Some(plan) = access::source_segment_scan_from_filter(predicate, input) {
956 decisions.push(
957 "choose SourceSegmentScan for storage-prunable Source filter".to_string(),
958 );
959 PhysicalPlan::FilterExec {
960 predicate: predicate.clone(),
961 input: Box::new(plan),
962 }
963 } else if let Predicate::BoundRelationshipExists {
964 source_variable,
965 rel_type,
966 direction,
967 target_variable,
968 } = predicate
969 {
970 decisions.push(
971 "lower bound relationship existence predicate to AdjacencyExistsExec"
972 .to_string(),
973 );
974 PhysicalPlan::AdjacencyExistsExec {
975 source_variable: source_variable.clone(),
976 rel_type: rel_type.clone(),
977 direction: *direction,
978 target_variable: target_variable.clone(),
979 input: Box::new(children.lower(
980 input,
981 0,
982 catalog,
983 optimizer_context,
984 decisions,
985 stage_events,
986 )),
987 }
988 } else {
989 let mut input = children.lower(
990 input,
991 0,
992 catalog,
993 optimizer_context,
994 decisions,
995 stage_events,
996 );
997 push_seed_metadata_filter(&mut input, predicate, decisions);
998 input = join::lower_property_join(input, predicate, decisions);
999 PhysicalPlan::FilterExec {
1000 predicate: predicate.clone(),
1001 input: Box::new(input),
1002 }
1003 }
1004 }
1005 LogicalPlan::Project { items, input } => PhysicalPlan::ProjectExec {
1006 items: items.clone(),
1007 input: Box::new(children.lower(
1008 input,
1009 0,
1010 catalog,
1011 optimizer_context,
1012 decisions,
1013 stage_events,
1014 )),
1015 },
1016 LogicalPlan::Aggregate {
1017 group_keys,
1018 items,
1019 input,
1020 } => PhysicalPlan::AggregateExec {
1021 group_keys: group_keys.clone(),
1022 items: items.clone(),
1023 input: Box::new(children.lower(
1024 input,
1025 0,
1026 catalog,
1027 optimizer_context,
1028 decisions,
1029 stage_events,
1030 )),
1031 },
1032 LogicalPlan::Distinct { input } => PhysicalPlan::DistinctExec {
1033 input: Box::new(children.lower(
1034 input,
1035 0,
1036 catalog,
1037 optimizer_context,
1038 decisions,
1039 stage_events,
1040 )),
1041 },
1042 LogicalPlan::Sort { items, input } => {
1043 let input = children.lower(
1044 input,
1045 0,
1046 catalog,
1047 optimizer_context,
1048 decisions,
1049 stage_events,
1050 );
1051 if satisfies_ordering(&input, items, catalog) {
1052 decisions
1053 .push("elide SortExec: access path satisfies required ordering".to_string());
1054 input
1055 } else {
1056 PhysicalPlan::SortExec {
1057 items: items.clone(),
1058 input: Box::new(input),
1059 }
1060 }
1061 }
1062 LogicalPlan::Limit {
1063 offset,
1064 limit,
1065 input,
1066 } => {
1067 if *limit == Some(0) {
1068 PhysicalPlan::EmptyExec
1069 } else if let (Some(limit), LogicalPlan::Sort { items, input }) =
1070 (limit, input.as_ref())
1071 {
1072 let input = children.lower(
1073 input,
1074 0,
1075 catalog,
1076 optimizer_context,
1077 decisions,
1078 stage_events,
1079 );
1080 select_bounded_sort_plan(items.clone(), *offset, *limit, input, catalog, decisions)
1081 } else {
1082 PhysicalPlan::LimitExec {
1083 offset: *offset,
1084 limit: *limit,
1085 input: Box::new(children.lower(
1086 input,
1087 0,
1088 catalog,
1089 optimizer_context,
1090 decisions,
1091 stage_events,
1092 )),
1093 }
1094 }
1095 }
1096 _ => unreachable!("leaf logical plans are lowered before child planning"),
1097 }
1098}
1099
1100fn select_exact_count_fast_path(
1101 logical: &LogicalPlan,
1102 decisions: &mut Vec<String>,
1103 stage_events: &mut Vec<StageTrace>,
1104) -> Option<PhysicalPlan> {
1105 if let Some(plan) = select_node_count_fast_path(logical, decisions, stage_events) {
1106 return Some(plan);
1107 }
1108 select_relationship_count_fast_path(logical, decisions, stage_events)
1109}
1110
1111fn select_node_count_fast_path(
1112 logical: &LogicalPlan,
1113 decisions: &mut Vec<String>,
1114 stage_events: &mut Vec<StageTrace>,
1115) -> Option<PhysicalPlan> {
1116 let LogicalPlan::Aggregate {
1117 group_keys,
1118 items,
1119 input,
1120 } = logical
1121 else {
1122 return None;
1123 };
1124 let [item] = items.as_slice() else {
1125 return None;
1126 };
1127 let LogicalPlan::NodeScan { variable, label } = input.as_ref() else {
1128 return None;
1129 };
1130 let exact_node_count = group_keys.is_empty()
1131 && item.function == AggregateFunction::Count
1132 && match &item.target {
1133 AggregateTarget::All => !item.distinct,
1134 AggregateTarget::Variable(target) => target == variable,
1135 AggregateTarget::Property { .. }
1136 | AggregateTarget::Column(_)
1137 | AggregateTarget::ColumnProperty { .. } => false,
1138 };
1139 if !exact_node_count {
1140 return None;
1141 }
1142
1143 decisions.push(format!(
1144 "selected implementation:node_count_fast_path: use exact label count for {label}"
1145 ));
1146 stage_events
1147 .push(ACCESS_PATH_SELECTION_STAGE.trace(StageStats::new(1, 1).with_rule_counts(1, 0)));
1148 Some(PhysicalPlan::NodeCountExec {
1149 label: label.clone(),
1150 output: item.name.clone(),
1151 })
1152}
1153
1154fn select_relationship_count_fast_path(
1155 logical: &LogicalPlan,
1156 decisions: &mut Vec<String>,
1157 stage_events: &mut Vec<StageTrace>,
1158) -> Option<PhysicalPlan> {
1159 let LogicalPlan::Aggregate {
1160 group_keys,
1161 items,
1162 input,
1163 } = logical
1164 else {
1165 return None;
1166 };
1167 let [item] = items.as_slice() else {
1168 return None;
1169 };
1170 let LogicalPlan::Expand {
1171 source_variable,
1172 source_label,
1173 rel_variable,
1174 rel_type,
1175 rel_properties,
1176 direction,
1177 target_label,
1178 min_hops,
1179 max_hops,
1180 optional,
1181 input,
1182 ..
1183 } = input.as_ref()
1184 else {
1185 return None;
1186 };
1187 let LogicalPlan::NodeScan { variable, label } = input.as_ref() else {
1188 return None;
1189 };
1190 let exact_relationship_count = group_keys.is_empty()
1191 && item.function == AggregateFunction::Count
1192 && !item.distinct
1193 && source_variable == variable
1194 && source_label.is_empty()
1195 && label.is_empty()
1196 && target_label.is_empty()
1197 && rel_properties.is_empty()
1198 && *min_hops == 1
1199 && *max_hops == 1
1200 && !optional
1201 && matches!(
1202 direction,
1203 RelationshipDirection::Outgoing | RelationshipDirection::Incoming
1204 )
1205 && match &item.target {
1206 AggregateTarget::All => true,
1207 AggregateTarget::Variable(target) => rel_variable.as_ref() == Some(target),
1208 AggregateTarget::Property { .. }
1209 | AggregateTarget::Column(_)
1210 | AggregateTarget::ColumnProperty { .. } => false,
1211 };
1212 if !exact_relationship_count {
1213 return None;
1214 }
1215
1216 decisions.push(format!(
1217 "selected implementation:relationship_count_fast_path: use exact relationship count for {rel_type}"
1218 ));
1219 stage_events
1220 .push(ACCESS_PATH_SELECTION_STAGE.trace(StageStats::new(1, 1).with_rule_counts(1, 0)));
1221 Some(PhysicalPlan::RelationshipCountExec {
1222 rel_type: rel_type.clone(),
1223 output: item.name.clone(),
1224 })
1225}
1226
1227fn finalize_physical_plan(
1228 plan: PhysicalPlan,
1229 decisions: &mut Vec<String>,
1230 stage_events: &mut Vec<StageTrace>,
1231) -> PhysicalPlan {
1232 let event_count_before = stage_events.len();
1233 let finalized = finalize_required_node_properties(plan, decisions, stage_events);
1234 if !stage_events[event_count_before..]
1235 .iter()
1236 .any(|event| event.name() == PLAN_FINALIZATION_STAGE.name())
1237 {
1238 stage_events.push(PLAN_FINALIZATION_STAGE.trace(StageStats::new(1, 1)));
1239 }
1240 finalized
1241}
1242
1243fn finalize_required_node_properties(
1244 plan: PhysicalPlan,
1245 decisions: &mut Vec<String>,
1246 stage_events: &mut Vec<StageTrace>,
1247) -> PhysicalPlan {
1248 match plan {
1249 PhysicalPlan::ProjectExec { items, input } => {
1250 select_node_projection_scan(&items, &input, decisions, stage_events)
1251 .unwrap_or(PhysicalPlan::ProjectExec { items, input })
1252 }
1253 PhysicalPlan::AggregateExec {
1254 group_keys,
1255 items,
1256 input,
1257 } => {
1258 let input = select_node_aggregate_required_scan(
1259 &group_keys,
1260 &items,
1261 &input,
1262 decisions,
1263 stage_events,
1264 )
1265 .map(Box::new)
1266 .unwrap_or(input);
1267 PhysicalPlan::AggregateExec {
1268 group_keys,
1269 items,
1270 input,
1271 }
1272 }
1273 PhysicalPlan::LimitExec {
1274 offset,
1275 limit,
1276 input,
1277 } => PhysicalPlan::LimitExec {
1278 offset,
1279 limit,
1280 input: Box::new(finalize_required_node_properties(
1281 *input,
1282 decisions,
1283 stage_events,
1284 )),
1285 },
1286 PhysicalPlan::DistinctExec { input } => PhysicalPlan::DistinctExec {
1287 input: Box::new(finalize_required_node_properties(
1288 *input,
1289 decisions,
1290 stage_events,
1291 )),
1292 },
1293 PhysicalPlan::SortExec { items, input }
1294 if sort_items_use_only_projected_columns(&items) =>
1295 {
1296 PhysicalPlan::SortExec {
1297 items,
1298 input: Box::new(finalize_required_node_properties(
1299 *input,
1300 decisions,
1301 stage_events,
1302 )),
1303 }
1304 }
1305 PhysicalPlan::TopNExec {
1306 items,
1307 offset,
1308 limit,
1309 input,
1310 } if sort_items_use_only_projected_columns(&items) => PhysicalPlan::TopNExec {
1311 items,
1312 offset,
1313 limit,
1314 input: Box::new(finalize_required_node_properties(
1315 *input,
1316 decisions,
1317 stage_events,
1318 )),
1319 },
1320 plan => plan,
1321 }
1322}
1323
1324fn sort_items_use_only_projected_columns(items: &[SortItem]) -> bool {
1325 items
1326 .iter()
1327 .all(|item| matches!(item.key, SortKey::Column(_)))
1328}
1329
1330fn select_node_projection_scan(
1331 items: &[Projection],
1332 input: &PhysicalPlan,
1333 decisions: &mut Vec<String>,
1334 stage_events: &mut Vec<StageTrace>,
1335) -> Option<PhysicalPlan> {
1336 let (source, predicate) = match input {
1337 PhysicalPlan::FilterExec { predicate, input } => (input.as_ref(), Some(predicate)),
1338 source => (source, None),
1339 };
1340 let (variable, label, access) = node_projection_access(source)?;
1341 let mut required_properties = BTreeSet::new();
1342 if !items.iter().all(|item| {
1343 collect_projection_properties(&item.expression, variable, &mut required_properties)
1344 }) || predicate.is_some_and(|predicate| {
1345 !collect_predicate_properties(predicate, variable, &mut required_properties)
1346 }) {
1347 return None;
1348 }
1349 collect_access_properties(&access, &mut required_properties);
1350
1351 let required_properties = required_properties.into_iter().collect::<Vec<_>>();
1352 decisions.push(format!(
1353 "selected implementation:node_projection_scan: access={} decode {} required properties for {variable}:{label}",
1354 access.kind_name(),
1355 required_properties.len(),
1356 ));
1357 stage_events.push(PLAN_FINALIZATION_STAGE.trace(StageStats::new(1, 1).with_rule_counts(1, 0)));
1358 Some(PhysicalPlan::NodeProjectionScanExec {
1359 variable: variable.clone(),
1360 label: label.clone(),
1361 access,
1362 required_properties,
1363 predicate: predicate.cloned(),
1364 items: items.to_vec(),
1365 })
1366}
1367
1368fn select_node_aggregate_required_scan(
1369 group_keys: &[Projection],
1370 items: &[hawdb_plan_cypher::Aggregation],
1371 input: &PhysicalPlan,
1372 decisions: &mut Vec<String>,
1373 stage_events: &mut Vec<StageTrace>,
1374) -> Option<PhysicalPlan> {
1375 let (source, predicate) = match input {
1376 PhysicalPlan::FilterExec { predicate, input } => (input.as_ref(), Some(predicate)),
1377 source => (source, None),
1378 };
1379 let (variable, label, access) = node_projection_access(source)?;
1380 let mut required_properties = BTreeSet::new();
1381 if !group_keys.iter().all(|item| {
1382 collect_projection_properties(&item.expression, variable, &mut required_properties)
1383 }) || !items
1384 .iter()
1385 .all(|item| collect_aggregate_properties(item, variable, &mut required_properties))
1386 || predicate.is_some_and(|predicate| {
1387 !collect_predicate_properties(predicate, variable, &mut required_properties)
1388 })
1389 {
1390 return None;
1391 }
1392 collect_access_properties(&access, &mut required_properties);
1393
1394 let required_properties = required_properties.into_iter().collect::<Vec<_>>();
1395 decisions.push(format!(
1396 "selected implementation:node_aggregate_required_scan: access={} decode {} required properties for {variable}:{label}",
1397 access.kind_name(),
1398 required_properties.len(),
1399 ));
1400 stage_events.push(PLAN_FINALIZATION_STAGE.trace(StageStats::new(1, 1).with_rule_counts(1, 0)));
1401 Some(PhysicalPlan::NodeProjectionScanExec {
1402 variable: variable.clone(),
1403 label: label.clone(),
1404 access,
1405 required_properties,
1406 predicate: predicate.cloned(),
1407 items: Vec::new(),
1408 })
1409}
1410
1411fn collect_aggregate_properties(
1412 item: &hawdb_plan_cypher::Aggregation,
1413 variable: &str,
1414 required: &mut BTreeSet<String>,
1415) -> bool {
1416 match (&item.function, &item.target) {
1417 (AggregateFunction::Count, AggregateTarget::All) => true,
1418 (AggregateFunction::Count, AggregateTarget::Variable(current)) => current == variable,
1419 (
1420 AggregateFunction::Count
1421 | AggregateFunction::Min
1422 | AggregateFunction::Max
1423 | AggregateFunction::Avg
1424 | AggregateFunction::Collect,
1425 AggregateTarget::Property {
1426 variable: current,
1427 property,
1428 },
1429 ) if current == variable => {
1430 required.insert(property.clone());
1431 true
1432 }
1433 _ => false,
1434 }
1435}
1436
1437fn node_projection_access(plan: &PhysicalPlan) -> Option<(&String, &String, NodeProjectionAccess)> {
1438 match plan {
1439 PhysicalPlan::SeqNodeScan { variable, label } => {
1440 Some((variable, label, NodeProjectionAccess::LabelScan))
1441 }
1442 PhysicalPlan::IndexNodeSeek {
1443 variable,
1444 label,
1445 property,
1446 value,
1447 } => Some((
1448 variable,
1449 label,
1450 NodeProjectionAccess::PropertyValues {
1451 property: property.clone(),
1452 values: vec![value.clone()],
1453 },
1454 )),
1455 PhysicalPlan::IndexNodeMultiSeek {
1456 variable,
1457 label,
1458 property,
1459 values,
1460 } => Some((
1461 variable,
1462 label,
1463 NodeProjectionAccess::PropertyValues {
1464 property: property.clone(),
1465 values: values.clone(),
1466 },
1467 )),
1468 PhysicalPlan::IndexNodeUnionSeek {
1469 variable,
1470 label,
1471 branches,
1472 } => Some((
1473 variable,
1474 label,
1475 NodeProjectionAccess::PropertyUnion {
1476 branches: branches.clone(),
1477 },
1478 )),
1479 PhysicalPlan::IndexNodeCompositeSeek {
1480 variable,
1481 label,
1482 predicates,
1483 } => Some((
1484 variable,
1485 label,
1486 NodeProjectionAccess::CompositeEquality {
1487 predicates: predicates.clone(),
1488 },
1489 )),
1490 PhysicalPlan::IndexNodeCompositeRangeSeek {
1491 variable,
1492 label,
1493 seek,
1494 } => Some((
1495 variable,
1496 label,
1497 NodeProjectionAccess::CompositeRange { seek: seek.clone() },
1498 )),
1499 PhysicalPlan::IndexNodeRangeSeek {
1500 variable,
1501 label,
1502 property,
1503 lower,
1504 upper,
1505 } => Some((
1506 variable,
1507 label,
1508 NodeProjectionAccess::PropertyRange {
1509 property: property.clone(),
1510 lower: lower.clone(),
1511 upper: upper.clone(),
1512 },
1513 )),
1514 PhysicalPlan::IndexNodeTextSeek {
1515 variable,
1516 label,
1517 property,
1518 query,
1519 } => Some((
1520 variable,
1521 label,
1522 NodeProjectionAccess::FullText {
1523 property: property.clone(),
1524 query: query.clone(),
1525 },
1526 )),
1527 _ => None,
1528 }
1529}
1530
1531fn collect_access_properties(access: &NodeProjectionAccess, required: &mut BTreeSet<String>) {
1532 match access {
1533 NodeProjectionAccess::LabelScan => {}
1534 NodeProjectionAccess::PropertyValues { property, .. }
1535 | NodeProjectionAccess::PropertyRange { property, .. }
1536 | NodeProjectionAccess::FullText { property, .. } => {
1537 required.insert(property.clone());
1538 }
1539 NodeProjectionAccess::CompositeEquality { predicates } => {
1540 required.extend(predicates.iter().map(|(property, _)| property.clone()));
1541 }
1542 NodeProjectionAccess::CompositeRange { seek } => {
1543 required.extend(
1544 seek.equality_prefix
1545 .iter()
1546 .map(|(property, _)| property.clone()),
1547 );
1548 required.insert(seek.range_property.clone());
1549 }
1550 NodeProjectionAccess::PropertyUnion { branches } => {
1551 required.extend(branches.iter().map(|branch| branch.property.clone()));
1552 }
1553 }
1554}
1555
1556fn collect_predicate_properties(
1557 predicate: &Predicate,
1558 variable: &str,
1559 required: &mut BTreeSet<String>,
1560) -> bool {
1561 match predicate {
1562 Predicate::And(predicates) | Predicate::Or(predicates) => predicates
1563 .iter()
1564 .all(|predicate| collect_predicate_properties(predicate, variable, required)),
1565 Predicate::Not(predicate) => collect_predicate_properties(predicate, variable, required),
1566 Predicate::ConstantBool(_) => true,
1567 Predicate::IdEq {
1568 variable: current, ..
1569 }
1570 | Predicate::IdNotEq {
1571 variable: current, ..
1572 }
1573 | Predicate::IdCompare {
1574 variable: current, ..
1575 }
1576 | Predicate::IdIn {
1577 variable: current, ..
1578 } => current == variable,
1579 Predicate::PropertyEq {
1580 variable: current,
1581 property,
1582 ..
1583 }
1584 | Predicate::PropertyNotEq {
1585 variable: current,
1586 property,
1587 ..
1588 }
1589 | Predicate::PropertyCompare {
1590 variable: current,
1591 property,
1592 ..
1593 }
1594 | Predicate::PropertyListContains {
1595 variable: current,
1596 property,
1597 ..
1598 }
1599 | Predicate::PropertyListContainsLower {
1600 variable: current,
1601 property,
1602 ..
1603 }
1604 | Predicate::PropertyContains {
1605 variable: current,
1606 property,
1607 ..
1608 }
1609 | Predicate::PropertyStartsWith {
1610 variable: current,
1611 property,
1612 ..
1613 }
1614 | Predicate::PropertyEndsWith {
1615 variable: current,
1616 property,
1617 ..
1618 }
1619 | Predicate::PropertyRegexMatch {
1620 variable: current,
1621 property,
1622 ..
1623 }
1624 | Predicate::PropertyIsNull {
1625 variable: current,
1626 property,
1627 }
1628 | Predicate::PropertyIsNotNull {
1629 variable: current,
1630 property,
1631 }
1632 | Predicate::PropertyIn {
1633 variable: current,
1634 property,
1635 ..
1636 } => {
1637 if current != variable {
1638 return false;
1639 }
1640 required.insert(property.clone());
1641 true
1642 }
1643 Predicate::ExpressionEq { expression, value }
1644 | Predicate::ExpressionNotEq { expression, value }
1645 | Predicate::ExpressionContains { expression, value } => {
1646 collect_projection_properties(expression, variable, required)
1647 && collect_projection_properties(value, variable, required)
1648 }
1649 Predicate::ExpressionCompare {
1650 expression, value, ..
1651 } => {
1652 collect_projection_properties(expression, variable, required)
1653 && collect_projection_properties(value, variable, required)
1654 }
1655 Predicate::RelationshipExists { .. } | Predicate::BoundRelationshipExists { .. } => false,
1656 }
1657}
1658
1659fn collect_projection_properties(
1660 expression: &ProjectionExpression,
1661 variable: &str,
1662 required: &mut BTreeSet<String>,
1663) -> bool {
1664 match expression {
1665 ProjectionExpression::Case { .. }
1666 | ProjectionExpression::Binary { .. }
1667 | ProjectionExpression::Not(_)
1668 | ProjectionExpression::IsNull { .. } => expression
1669 .all_children(|child| collect_projection_properties(child, variable, required)),
1670 ProjectionExpression::Variable { .. }
1671 | ProjectionExpression::RelationshipType { .. }
1672 | ProjectionExpression::Column(_)
1673 | ProjectionExpression::ColumnProperty { .. }
1674 | ProjectionExpression::ColumnDefaultIfNullOrEq { .. }
1675 | ProjectionExpression::ColumnValueDefaultIfNull { .. }
1676 | ProjectionExpression::ColumnValueCasePropertyNotNullOrEq { .. }
1677 | ProjectionExpression::CaseColumnSearchRank(_) => false,
1678 ProjectionExpression::Id { variable: current } => current == variable,
1679 ProjectionExpression::Literal(_) => true,
1680 ProjectionExpression::Property {
1681 variable: current,
1682 property,
1683 }
1684 | ProjectionExpression::DatePart {
1685 variable: current,
1686 property,
1687 ..
1688 }
1689 | ProjectionExpression::DefaultIfNullOrEq {
1690 variable: current,
1691 property,
1692 ..
1693 }
1694 | ProjectionExpression::DefaultIfNull {
1695 variable: current,
1696 property,
1697 ..
1698 }
1699 | ProjectionExpression::CasePropertyNotNullOrEq {
1700 variable: current,
1701 property,
1702 ..
1703 }
1704 | ProjectionExpression::CasePropertyEqualsRank {
1705 variable: current,
1706 property,
1707 ..
1708 }
1709 | ProjectionExpression::CaseLowerPropertyDefault {
1710 variable: current,
1711 property,
1712 ..
1713 } => {
1714 if current != variable {
1715 return false;
1716 }
1717 required.insert(property.clone());
1718 true
1719 }
1720 ProjectionExpression::Coalesce(expressions) => expressions
1721 .iter()
1722 .all(|expression| collect_projection_properties(expression, variable, required)),
1723 ProjectionExpression::Left { expression, .. } | ProjectionExpression::Lower(expression) => {
1724 collect_projection_properties(expression, variable, required)
1725 }
1726 ProjectionExpression::CaseCoalesceDifferenceFloorZero {
1727 variable: current,
1728 terms,
1729 } => {
1730 if current != variable {
1731 return false;
1732 }
1733 required.extend(terms.iter().map(|term| term.property.clone()));
1734 true
1735 }
1736 ProjectionExpression::CaseEntitySearchRank(rank) => {
1737 if rank.variable != variable {
1738 return false;
1739 }
1740 required.insert(rank.name_property.clone());
1741 required.insert(rank.aliases_property.clone());
1742 true
1743 }
1744 }
1745}
1746
1747struct ExpandEstimateRequest<'a> {
1748 source_label: &'a str,
1749 rel_type: &'a str,
1750 rel_properties: &'a BTreeMap<String, Value>,
1751 target_label: &'a str,
1752 min_hops: usize,
1753 max_hops: usize,
1754}
1755
1756fn push_expand_estimate_decision(
1757 catalog: &OptimizerCatalog,
1758 decisions: &mut Vec<String>,
1759 request: ExpandEstimateRequest<'_>,
1760) {
1761 let estimate = catalog.estimate_expand_rows(
1762 request.source_label,
1763 request.rel_type,
1764 request.rel_properties,
1765 request.target_label,
1766 request.min_hops,
1767 request.max_hops,
1768 );
1769 let path_count = estimate
1770 .path_count
1771 .map(|count| count.to_string())
1772 .unwrap_or_else(|| "unknown".to_string());
1773 let hop_rows = estimate
1774 .hop_estimates
1775 .iter()
1776 .map(|estimate| {
1777 format!(
1778 "{}:{}:{}",
1779 estimate.hop,
1780 if estimate.exact { "exact" } else { "fallback" },
1781 estimate.rows
1782 )
1783 })
1784 .collect::<Vec<_>>()
1785 .join(",");
1786 decisions.push(format!(
1787 "estimate AdjacencyExpand for {}-[:{}*{}..{}]->{}: path_count={path_count} rel_count={} rel_type_sources={} average_fanout={} rel_property_distinct_product={} hop_rows=[{}] estimated_rows={}",
1788 request.source_label,
1789 request.rel_type,
1790 request.min_hops,
1791 request.max_hops,
1792 request.target_label,
1793 estimate.rel_count,
1794 estimate.source_count,
1795 estimate.average_fanout,
1796 estimate.property_distinct_product,
1797 hop_rows,
1798 estimate.estimated_rows
1799 ));
1800}
1801
1802fn order_single_row_cartesian_product_children(
1803 catalog: &OptimizerCatalog,
1804 decisions: &mut Vec<String>,
1805 left: PhysicalPlan,
1806 right: PhysicalPlan,
1807) -> (PhysicalPlan, PhysicalPlan) {
1808 let left_cost = estimate_physical_plan_cost(&left, catalog);
1809 let right_cost = estimate_physical_plan_cost(&right, catalog);
1810 if !cartesian_product_leaves_are_single_row(catalog, &left)
1811 || !cartesian_product_leaves_are_single_row(catalog, &right)
1812 {
1813 decisions.push(format!(
1814 "keep NodeCartesianProduct input order: left_rows={} right_rows={} reason=non_single_row_input",
1815 left_cost.estimated_rows, right_cost.estimated_rows
1816 ));
1817 return (left, right);
1818 }
1819
1820 let mut leaves = Vec::new();
1821 collect_cartesian_product_leaves(left, &mut leaves);
1822 collect_cartesian_product_leaves(right, &mut leaves);
1823
1824 let mut keyed_leaves = leaves
1825 .into_iter()
1826 .map(|plan| {
1827 let cost = estimate_physical_plan_cost(&plan, catalog);
1828 let fingerprint = plan.instance_fingerprint();
1829 (cost, fingerprint, plan)
1830 })
1831 .collect::<Vec<_>>();
1832
1833 let original_order = keyed_leaves
1834 .iter()
1835 .map(|(cost, fingerprint, _)| format!("{}:{}", cost.cost, fingerprint))
1836 .collect::<Vec<_>>()
1837 .join("|");
1838 keyed_leaves.sort_by(
1839 |(left_cost, left_fingerprint, _), (right_cost, right_fingerprint, _)| {
1840 (left_cost.cost, left_fingerprint).cmp(&(right_cost.cost, right_fingerprint))
1841 },
1842 );
1843 let ordered = keyed_leaves
1844 .iter()
1845 .map(|(cost, fingerprint, _)| format!("{}:{}", cost.cost, fingerprint))
1846 .collect::<Vec<_>>()
1847 .join("|");
1848 if original_order == ordered {
1849 decisions.push(format!(
1850 "keep NodeCartesianProduct single-row input order: inputs={} order={ordered}",
1851 keyed_leaves.len()
1852 ));
1853 } else {
1854 decisions.push(format!(
1855 "order NodeCartesianProduct single-row inputs: inputs={} original={original_order} ordered={ordered}",
1856 keyed_leaves.len()
1857 ));
1858 }
1859
1860 let mut ordered_plans = keyed_leaves.into_iter().map(|(_, _, plan)| plan);
1861 let left = ordered_plans
1862 .next()
1863 .expect("cartesian product has left input");
1864 let right = rebuild_cartesian_product(ordered_plans);
1865 (left, right)
1866}
1867
1868fn cartesian_product_leaves_are_single_row(
1869 catalog: &OptimizerCatalog,
1870 plan: &PhysicalPlan,
1871) -> bool {
1872 match plan {
1873 PhysicalPlan::NodeCartesianProductExec { left, right } => {
1874 cartesian_product_leaves_are_single_row(catalog, left)
1875 && cartesian_product_leaves_are_single_row(catalog, right)
1876 }
1877 plan => estimate_physical_plan_cost(plan, catalog).estimated_rows == 1,
1878 }
1879}
1880
1881fn collect_cartesian_product_leaves(plan: PhysicalPlan, leaves: &mut Vec<PhysicalPlan>) {
1882 match plan {
1883 PhysicalPlan::NodeCartesianProductExec { left, right } => {
1884 collect_cartesian_product_leaves(*left, leaves);
1885 collect_cartesian_product_leaves(*right, leaves);
1886 }
1887 plan => leaves.push(plan),
1888 }
1889}
1890
1891fn rebuild_cartesian_product(plans: impl IntoIterator<Item = PhysicalPlan>) -> PhysicalPlan {
1892 let mut plans = plans.into_iter();
1893 let first = plans
1894 .next()
1895 .expect("cartesian product rebuild requires at least one input");
1896 plans.fold(first, |left, right| {
1897 PhysicalPlan::NodeCartesianProductExec {
1898 left: Box::new(left),
1899 right: Box::new(right),
1900 }
1901 })
1902}
1903
1904fn push_cartesian_product_cost_decision(
1905 catalog: &OptimizerCatalog,
1906 decisions: &mut Vec<String>,
1907 left: &PhysicalPlan,
1908 right: &PhysicalPlan,
1909) {
1910 let left_cost = estimate_physical_plan_cost(left, catalog);
1911 let right_cost = estimate_physical_plan_cost(right, catalog);
1912 let cost = estimate_node_cartesian_product_cost(left_cost, right_cost);
1913 decisions.push(format!(
1914 "estimate NodeCartesianProduct: left_rows={} right_rows={} output_rows={} left_cost={} right_cost={} cost={}",
1915 left_cost.estimated_rows,
1916 right_cost.estimated_rows,
1917 cost.estimated_rows,
1918 left_cost.cost,
1919 right_cost.cost,
1920 cost.cost
1921 ));
1922}