use super::{
access_path::index_seek_from_filter,
costing::{
estimate_node_cartesian_product_cost, estimate_physical_plan_cost,
push_optional_relationship_count_sum_cost_decision,
},
logical_rewrite::{rewrite_logical_plan, LogicalRewriteOutput},
selected_trace::selected_plan_trace,
stages::{
ACCESS_PATH_SELECTION_STAGE, DIRECT_PHYSICAL_FALLBACK_STAGE, LOGICAL_GROUPING_STAGE,
PHYSICAL_SEARCH_STAGE, PLAN_FINALIZATION_STAGE, SELECTED_PLAN_COSTING_STAGE,
},
LogicalPlanRoot, LoweringReadyLogicalPlanRoot, OptimizationSearchReport, OptimizerCatalog,
OptimizerConfig, OptimizerTrace, PhysicalPlan, PhysicalPlanRoot, StageStats,
};
use crate::{
GroupId, Memo, OptimizerContext, OptimizerSearchDirective, OptimizerSearchDirectiveError,
RuleEvent, RuleOutcome, SelectedPlanTrace, StageTrace,
};
use hawdb_core::Value;
use hawdb_cypher::RelationshipDirection;
use hawdb_plan::{
AggregateFunction, AggregateTarget, GraphExpansionBudget, LogicalPlan, NodeProjectionAccess,
Predicate, Projection, ProjectionExpression, SortItem, SortKey,
};
use std::collections::{BTreeMap, BTreeSet};
mod access;
mod ddl;
mod join;
mod mutation;
mod procedure;
mod simple;
mod traversal;
#[cfg(test)]
mod tests;
type GraphMemo = Memo<GroupExpr>;
const GRAPH_EXPANSION_FANOUT_PER_SEED_HOP: usize = 32;
const GRAPH_EXPANSION_MAX_CANDIDATES: usize = 4_096;
const GRAPH_EXPANSION_PAYLOAD_BYTE_LIMIT: usize = 4 * 1024 * 1024;
#[derive(Debug)]
struct GroupExpr {
logical: LogicalPlan,
children: Vec<GroupId>,
}
#[derive(Debug, Default, Clone)]
pub struct CascadesOptimizer {
context: OptimizerContext,
}
impl CascadesOptimizer {
pub fn new(config: OptimizerConfig) -> Self {
Self {
context: OptimizerContext::from_config(config),
}
}
pub fn with_context(context: OptimizerContext) -> Self {
Self { context }
}
pub fn context(&self) -> &OptimizerContext {
&self.context
}
pub fn refresh_trace_for_physical_plan(
&self,
trace: &mut OptimizerTrace,
plan: &PhysicalPlan,
catalog: &OptimizerCatalog,
) {
refresh_selected_plan_trace(trace, selected_plan_trace(plan, catalog, &self.context));
}
pub fn optimize(&self, logical: &LogicalPlan) -> PhysicalPlan {
self.optimize_root(&LogicalPlanRoot::new(logical.clone()))
.into_parts()
.0
}
pub fn optimize_with_trace(&self, logical: &LogicalPlan) -> (PhysicalPlan, OptimizerTrace) {
self.optimize_root(&LogicalPlanRoot::new(logical.clone()))
.into_parts()
}
pub fn optimize_with_catalog(
&self,
logical: &LogicalPlan,
catalog: &OptimizerCatalog,
) -> (PhysicalPlan, OptimizerTrace) {
self.optimize_root_with_catalog(&LogicalPlanRoot::new(logical.clone()), catalog)
.into_parts()
}
pub fn optimize_root(&self, root: &LogicalPlanRoot) -> PhysicalPlanRoot {
self.optimize_root_with_catalog(root, &OptimizerCatalog::optimistic())
}
pub fn optimize_root_with_catalog(
&self,
root: &LogicalPlanRoot,
catalog: &OptimizerCatalog,
) -> PhysicalPlanRoot {
self.optimize_lowering_ready_root_with_catalog_and_directive(
&root.clone().into_lowering_ready(),
catalog,
OptimizerSearchDirective::Auto,
)
.expect("automatic optimizer search cannot reject its directive")
}
pub fn optimize_lowering_ready_root_with_catalog(
&self,
root: &LoweringReadyLogicalPlanRoot,
catalog: &OptimizerCatalog,
) -> PhysicalPlanRoot {
self.optimize_lowering_ready_root_with_catalog_and_directive(
root,
catalog,
OptimizerSearchDirective::Auto,
)
.expect("automatic optimizer search cannot reject its directive")
}
pub fn optimize_root_with_catalog_and_directive(
&self,
root: &LogicalPlanRoot,
catalog: &OptimizerCatalog,
directive: OptimizerSearchDirective,
) -> Result<PhysicalPlanRoot, OptimizerSearchDirectiveError> {
self.optimize_lowering_ready_root_with_catalog_and_directive(
&root.clone().into_lowering_ready(),
catalog,
directive,
)
}
pub fn optimize_lowering_ready_root_with_catalog_and_directive(
&self,
root: &LoweringReadyLogicalPlanRoot,
catalog: &OptimizerCatalog,
directive: OptimizerSearchDirective,
) -> Result<PhysicalPlanRoot, OptimizerSearchDirectiveError> {
let rewrite = rewrite_logical_plan(root.plan());
let logical = rewrite.plan();
let required_groups = logical_group_count(logical);
let max_groups = self.context.optimizer_config().max_groups;
if directive == OptimizerSearchDirective::Memo && required_groups > max_groups {
return Err(OptimizerSearchDirectiveError::MemoGroupBudgetExceeded {
required_groups,
max_groups,
});
}
if directive == OptimizerSearchDirective::DirectFallback
|| (directive == OptimizerSearchDirective::Auto && required_groups > max_groups)
{
let mut decisions = Vec::new();
let mut stage_events = Vec::new();
let plan = lower_logical(
logical,
LoweringChildren::Direct,
catalog,
&self.context,
&mut decisions,
&mut stage_events,
);
let plan = finalize_physical_plan(plan, &mut decisions, &mut stage_events);
let mut report = if directive == OptimizerSearchDirective::DirectFallback {
OptimizationSearchReport::forced_direct_fallback(required_groups)
} else {
OptimizationSearchReport::direct_fallback(required_groups, max_groups)
};
record_logical_rewrite(&mut report, &rewrite);
report.push_stage_event(
LOGICAL_GROUPING_STAGE.trace(StageStats::new(1, required_groups)),
);
for event in stage_events {
report.push_stage_event(event);
}
report.push_stage_event(
DIRECT_PHYSICAL_FALLBACK_STAGE.trace(StageStats::new(required_groups, 1)),
);
report.extend_decisions(decisions);
let selected = selected_plan_trace(&plan, catalog, &self.context);
report.push_stage_event(SELECTED_PLAN_COSTING_STAGE.trace(StageStats::new(1, 1)));
report.record_selected_plan_cost(selected.cost);
return Ok(PhysicalPlanRoot::new(plan, report.into_trace(selected)));
}
let mut memo = GraphMemo::default();
let root = insert_logical_group(&mut memo, logical);
let mut decisions = Vec::new();
let mut stage_events = Vec::new();
let plan = best_physical(
&memo,
root,
catalog,
&self.context,
&mut decisions,
&mut stage_events,
);
let plan = finalize_physical_plan(plan, &mut decisions, &mut stage_events);
let mut report = OptimizationSearchReport::memo(memo.group_count());
record_logical_rewrite(&mut report, &rewrite);
report
.push_stage_event(LOGICAL_GROUPING_STAGE.trace(StageStats::new(1, memo.group_count())));
let (applied_rules, skipped_rules) = stage_rule_counts(&stage_events);
for event in stage_events {
report.push_stage_event(event);
}
report.push_stage_event(PHYSICAL_SEARCH_STAGE.trace(
StageStats::new(memo.group_count(), 1).with_rule_counts(applied_rules, skipped_rules),
));
report.extend_decisions(decisions);
let selected = selected_plan_trace(&plan, catalog, &self.context);
report.push_stage_event(SELECTED_PLAN_COSTING_STAGE.trace(StageStats::new(1, 1)));
report.record_selected_plan_cost(selected.cost);
if directive == OptimizerSearchDirective::Memo {
report.push_decision("selected memo search: explicit optimizer search directive");
}
Ok(PhysicalPlanRoot::new(plan, report.into_trace(selected)))
}
}
fn refresh_selected_plan_trace(trace: &mut OptimizerTrace, selected: SelectedPlanTrace) {
let selected_cost = selected.cost.with_cardinality_floor();
let selected_cost_breakdown = selected.cost_breakdown.with_cardinality_floor();
let cost_detail = format!(
"estimated_rows={} cost={}",
selected_cost.estimated_rows, selected_cost.cost
);
let cost_decision = format!("selected physical plan cost: {cost_detail}");
if let Some(decision) = trace
.decisions
.iter_mut()
.find(|decision| decision.starts_with("selected physical plan cost: "))
{
*decision = cost_decision;
} else {
trace.decisions.push(cost_decision);
}
let cost_event = RuleEvent::selected("physical plan cost", cost_detail);
if let Some(event) = trace.rule_events.iter_mut().find(|event| {
event.rule() == "physical plan cost" && event.outcome() == RuleOutcome::Selected
}) {
*event = cost_event;
} else {
trace.rule_events.push(cost_event);
}
trace.query_digest = selected.query_digest;
trace.selected_plan = selected.explain;
trace.selected_plan_fingerprint = selected.fingerprint;
trace.selected_plan_cost = selected_cost;
trace.selected_plan_cost_breakdown = selected_cost_breakdown;
trace.selected_plan_properties = selected.properties;
trace.selected_plan_cardinality_estimates = selected.cardinality_estimates;
trace.selected_plan_operator_counts = selected.operator_counts;
trace.selected_plan_class_counts = selected.class_counts;
}
pub(super) fn record_logical_rewrite(
report: &mut OptimizationSearchReport,
rewrite: &LogicalRewriteOutput,
) {
report.push_stage_event(rewrite.trace().clone());
if let Some(warning) = rewrite.warning() {
report.push_warning(warning);
}
for event in rewrite.events().iter().cloned() {
report.push_rule_event(event);
}
}
fn insert_logical_group(memo: &mut GraphMemo, logical: &LogicalPlan) -> GroupId {
let expr = GroupExpr::from_logical(logical, memo);
memo.insert_group(expr)
}
fn best_physical(
memo: &GraphMemo,
root: GroupId,
catalog: &OptimizerCatalog,
optimizer_context: &OptimizerContext,
decisions: &mut Vec<String>,
stage_events: &mut Vec<StageTrace>,
) -> PhysicalPlan {
let group = memo.group(root).expect("memo group id should exist");
group
.first_expression()
.expect("memo group should contain at least one expression")
.to_physical(memo, catalog, optimizer_context, decisions, stage_events)
}
fn select_bounded_sort_plan(
items: Vec<SortItem>,
offset: usize,
limit: usize,
input: PhysicalPlan,
catalog: &OptimizerCatalog,
decisions: &mut Vec<String>,
) -> PhysicalPlan {
let top_n = PhysicalPlan::TopNExec {
items: items.clone(),
offset,
limit,
input: Box::new(input.clone()),
};
let sort_limit = PhysicalPlan::LimitExec {
offset,
limit: Some(limit),
input: Box::new(PhysicalPlan::SortExec {
items,
input: Box::new(input),
}),
};
let top_n_cost = estimate_physical_plan_cost(&top_n, catalog);
let sort_limit_cost = estimate_physical_plan_cost(&sort_limit, catalog);
if top_n_cost.cost < sort_limit_cost.cost {
decisions.push(format!(
"choose TopN for bounded sort: offset={offset} limit={limit} top_n_cost={} sort_limit_cost={}",
top_n_cost.cost, sort_limit_cost.cost
));
top_n
} else {
decisions.push(format!(
"keep Sort + Limit for bounded sort: offset={offset} limit={limit} top_n_cost={} sort_limit_cost={}",
top_n_cost.cost, sort_limit_cost.cost
));
sort_limit
}
}
fn stage_rule_counts(stage_events: &[StageTrace]) -> (usize, usize) {
stage_events
.iter()
.fold((0, 0), |(applied, skipped), event| {
let stats = event.stats();
(
applied.saturating_add(stats.applied_rules),
skipped.saturating_add(stats.skipped_rules),
)
})
}
impl GroupExpr {
fn from_logical(logical: &LogicalPlan, memo: &mut GraphMemo) -> Self {
match logical {
LogicalPlan::GraphMatch { input, .. } => Self {
logical: logical.clone(),
children: input
.iter()
.map(|input| insert_logical_group(memo, input))
.collect(),
},
LogicalPlan::CreateNodeLabel { .. }
| LogicalPlan::CreateRelationshipType { .. }
| LogicalPlan::CreateNodeTable { .. }
| LogicalPlan::CreateRelationshipTable { .. }
| LogicalPlan::CreateProperty { .. }
| LogicalPlan::AlterTableState { .. }
| LogicalPlan::AlterPropertyState { .. }
| LogicalPlan::CreateIndex { .. }
| LogicalPlan::CreateCompositeIndex { .. }
| LogicalPlan::CreateRangeIndex { .. }
| LogicalPlan::CreateFullTextIndex { .. }
| LogicalPlan::CreateUniqueConstraint { .. }
| LogicalPlan::CreateNodePropertyExistsConstraint { .. }
| LogicalPlan::CreateRelationshipUniqueConstraint { .. }
| LogicalPlan::CreateRelationshipPropertyExistsConstraint { .. }
| LogicalPlan::ProjectGraph { .. }
| LogicalPlan::GraphAlgorithm { .. }
| LogicalPlan::VectorSeed { .. }
| LogicalPlan::CreateNode { .. }
| LogicalPlan::UnwindMutation { .. }
| LogicalPlan::MergeNode { .. }
| LogicalPlan::MergeRelationship { .. }
| LogicalPlan::MergeMatchedRelationship { .. }
| LogicalPlan::MergeRelationshipFromMatchedRelationship { .. }
| LogicalPlan::MergeRelationshipToMatchedTarget { .. }
| LogicalPlan::MergeRelationshipFromMatchedTarget { .. }
| LogicalPlan::CreateMatchedRelationship { .. }
| LogicalPlan::SetNodeProperty { .. }
| LogicalPlan::SetNodeProperties { .. }
| LogicalPlan::SetNodePropertiesReturn { .. }
| LogicalPlan::SetRelationshipProperty { .. }
| LogicalPlan::SetRelationshipProperties { .. }
| LogicalPlan::DeleteNode { .. }
| LogicalPlan::DeleteRelationship { .. }
| LogicalPlan::DeleteRelationshipTargetNodes { .. }
| LogicalPlan::CreateRelationship { .. }
| LogicalPlan::NodeScan { .. }
| LogicalPlan::OptionalRelationshipCountSum { .. }
| LogicalPlan::ThreadRepairStats { .. }
| LogicalPlan::ShortestPath { .. } => Self {
logical: logical.clone(),
children: Vec::new(),
},
LogicalPlan::Limit { limit: Some(0), .. } => Self {
logical: logical.clone(),
children: Vec::new(),
},
LogicalPlan::NodeCartesianProduct { left, right } => Self {
logical: logical.clone(),
children: vec![
insert_logical_group(memo, left),
insert_logical_group(memo, right),
],
},
LogicalPlan::Limit {
limit: Some(_),
input,
..
} if matches!(input.as_ref(), LogicalPlan::Sort { .. }) => {
let LogicalPlan::Sort { input, .. } = input.as_ref() else {
unreachable!("guard requires a sort input");
};
Self {
logical: logical.clone(),
children: vec![insert_logical_group(memo, input)],
}
}
LogicalPlan::Expand { input, .. }
| LogicalPlan::NodeColumnLookup { input, .. }
| LogicalPlan::OptionalDegree { input, .. }
| LogicalPlan::Filter { input, .. }
| LogicalPlan::Project { input, .. }
| LogicalPlan::Aggregate { input, .. }
| LogicalPlan::Distinct { input }
| LogicalPlan::Sort { input, .. }
| LogicalPlan::Limit { input, .. } => Self {
logical: logical.clone(),
children: vec![insert_logical_group(memo, input)],
},
}
}
fn to_physical(
&self,
memo: &GraphMemo,
catalog: &OptimizerCatalog,
optimizer_context: &OptimizerContext,
decisions: &mut Vec<String>,
stage_events: &mut Vec<StageTrace>,
) -> PhysicalPlan {
lower_logical(
&self.logical,
LoweringChildren::Memo {
memo,
children: &self.children,
},
catalog,
optimizer_context,
decisions,
stage_events,
)
}
}
fn push_vector_seed_metadata_filter(
input: &mut PhysicalPlan,
predicate: &hawdb_plan::Predicate,
decisions: &mut Vec<String>,
) {
if let hawdb_plan::Predicate::And(predicates) = predicate {
for predicate in predicates {
push_vector_seed_metadata_filter(input, predicate, decisions);
}
return;
}
let hawdb_plan::Predicate::PropertyEq {
variable,
property,
value,
} = predicate
else {
return;
};
let Some(filter_field) = vector_seed_metadata_field(property) else {
return;
};
let Some(filter_value) = vector_seed_metadata_value(value) else {
return;
};
if attach_vector_seed_metadata_filter(input, variable, filter_field, filter_value) {
decisions.push(format!(
"push descriptor-safe vector seed filter {variable}.{property} before candidate generation"
));
}
}
fn attach_vector_seed_metadata_filter(
plan: &mut PhysicalPlan,
variable: &str,
field: &str,
value: String,
) -> bool {
match plan {
PhysicalPlan::NodeColumnLookupExec {
variable: lookup_variable,
input,
..
} if lookup_variable == variable => {
attach_metadata_filter_to_vector_seed(input, field, value)
}
PhysicalPlan::AdjacencyExpandExec {
source_variable,
input,
..
} if source_variable == variable => {
attach_vector_seed_metadata_filter(input, variable, field, value)
}
_ => false,
}
}
fn attach_metadata_filter_to_vector_seed(
plan: &mut PhysicalPlan,
field: &str,
value: String,
) -> bool {
let PhysicalPlan::VectorSeedScan {
metadata_filters,
vector_plan,
..
} = plan
else {
return false;
};
if metadata_filters
.get(field)
.is_some_and(|existing| existing != &value)
{
return false;
}
metadata_filters.insert(field.to_string(), value);
attach_vector_filter_field(vector_plan, field);
true
}
fn attach_vector_filter_field(plan: &mut hawdb_plan::VectorPhysicalPlan, field: &str) {
match plan {
hawdb_plan::VectorPhysicalPlan::Filter { fields } => {
if !fields.iter().any(|existing| existing == field) {
fields.push(field.to_string());
fields.sort();
}
}
hawdb_plan::VectorPhysicalPlan::VectorCandidateScan { input, .. }
| hawdb_plan::VectorPhysicalPlan::ResidualFilter { input, .. }
| hawdb_plan::VectorPhysicalPlan::RawVectorRerank { input, .. }
| hawdb_plan::VectorPhysicalPlan::TopK { input, .. } => {
attach_vector_filter_field(input, field);
}
}
}
fn vector_seed_metadata_field(property: &str) -> Option<&str> {
match property {
"id" => Some("external_id"),
"kind" | "external_id" | "source_id" | "space_id" | "unit_type" | "lifecycle_state"
| "importance" | "confidence" | "created_at" | "updated_at" | "event_start"
| "event_end" | "is_latest" => Some(property),
_ => None,
}
}
fn vector_seed_metadata_value(value: &Value) -> Option<String> {
match value {
Value::String(value) => Some(value.clone()),
Value::Bool(value) => Some(value.to_string()),
Value::Int(value) => Some(value.to_string()),
Value::Float(value) if value.is_finite() => Some(value.to_string()),
Value::Uuid(value) => Some(value.to_string()),
Value::Null | Value::Float(_) | Value::Binary(_) | Value::List(_) | Value::Map(_) => None,
}
}
fn vector_seed_top_k(plan: &PhysicalPlan) -> Option<usize> {
match plan {
PhysicalPlan::VectorSeedScan { vector_plan, .. } => vector_plan_top_k(vector_plan),
PhysicalPlan::NodeColumnLookupExec { input, .. }
| PhysicalPlan::AdjacencyExpandExec { input, .. }
| PhysicalPlan::AdjacencyExistsExec { input, .. }
| PhysicalPlan::FilterExec { input, .. }
| PhysicalPlan::ProjectExec { input, .. }
| PhysicalPlan::LimitExec { input, .. } => vector_seed_top_k(input),
_ => None,
}
}
fn vector_plan_top_k(plan: &hawdb_plan::VectorPhysicalPlan) -> Option<usize> {
match plan {
hawdb_plan::VectorPhysicalPlan::TopK { limit, .. } => Some(*limit),
hawdb_plan::VectorPhysicalPlan::VectorCandidateScan { input, .. }
| hawdb_plan::VectorPhysicalPlan::ResidualFilter { input, .. }
| hawdb_plan::VectorPhysicalPlan::RawVectorRerank { input, .. } => vector_plan_top_k(input),
hawdb_plan::VectorPhysicalPlan::Filter { .. } => None,
}
}
fn graph_expansion_budget(top_k: usize, max_hops: usize) -> GraphExpansionBudget {
let candidate_limit = top_k
.max(1)
.saturating_mul(max_hops.max(1))
.saturating_mul(GRAPH_EXPANSION_FANOUT_PER_SEED_HOP)
.min(GRAPH_EXPANSION_MAX_CANDIDATES)
.max(top_k);
GraphExpansionBudget {
candidate_limit,
payload_byte_limit: GRAPH_EXPANSION_PAYLOAD_BYTE_LIMIT,
}
}
fn logical_group_count(logical: &LogicalPlan) -> usize {
match logical {
LogicalPlan::GraphMatch { input, .. } => {
1 + input.as_deref().map(logical_group_count).unwrap_or(0)
}
LogicalPlan::Limit { limit: Some(0), .. } => 1,
LogicalPlan::Limit {
limit: Some(_),
input,
..
} if matches!(input.as_ref(), LogicalPlan::Sort { .. }) => {
let LogicalPlan::Sort { input, .. } = input.as_ref() else {
unreachable!("guard requires a sort input");
};
1 + logical_group_count(input)
}
LogicalPlan::Expand { input, .. }
| LogicalPlan::NodeColumnLookup { input, .. }
| LogicalPlan::OptionalDegree { input, .. }
| LogicalPlan::Filter { input, .. }
| LogicalPlan::Project { input, .. }
| LogicalPlan::Aggregate { input, .. }
| LogicalPlan::Distinct { input }
| LogicalPlan::Sort { input, .. }
| LogicalPlan::Limit { input, .. } => 1 + logical_group_count(input),
LogicalPlan::OptionalRelationshipCountSum { .. }
| LogicalPlan::ThreadRepairStats { .. }
| LogicalPlan::ShortestPath { .. } => 1,
LogicalPlan::CreateNodeLabel { .. }
| LogicalPlan::CreateRelationshipType { .. }
| LogicalPlan::CreateNodeTable { .. }
| LogicalPlan::CreateRelationshipTable { .. }
| LogicalPlan::CreateProperty { .. }
| LogicalPlan::AlterTableState { .. }
| LogicalPlan::AlterPropertyState { .. }
| LogicalPlan::CreateIndex { .. }
| LogicalPlan::CreateCompositeIndex { .. }
| LogicalPlan::CreateRangeIndex { .. }
| LogicalPlan::CreateFullTextIndex { .. }
| LogicalPlan::CreateUniqueConstraint { .. }
| LogicalPlan::CreateNodePropertyExistsConstraint { .. }
| LogicalPlan::CreateRelationshipUniqueConstraint { .. }
| LogicalPlan::CreateRelationshipPropertyExistsConstraint { .. }
| LogicalPlan::ProjectGraph { .. }
| LogicalPlan::GraphAlgorithm { .. }
| LogicalPlan::VectorSeed { .. }
| LogicalPlan::CreateNode { .. }
| LogicalPlan::UnwindMutation { .. }
| LogicalPlan::MergeNode { .. }
| LogicalPlan::MergeRelationship { .. }
| LogicalPlan::MergeMatchedRelationship { .. }
| LogicalPlan::MergeRelationshipFromMatchedRelationship { .. }
| LogicalPlan::MergeRelationshipToMatchedTarget { .. }
| LogicalPlan::MergeRelationshipFromMatchedTarget { .. }
| LogicalPlan::CreateMatchedRelationship { .. }
| LogicalPlan::SetNodeProperty { .. }
| LogicalPlan::SetNodeProperties { .. }
| LogicalPlan::SetNodePropertiesReturn { .. }
| LogicalPlan::SetRelationshipProperty { .. }
| LogicalPlan::SetRelationshipProperties { .. }
| LogicalPlan::DeleteNode { .. }
| LogicalPlan::DeleteRelationship { .. }
| LogicalPlan::DeleteRelationshipTargetNodes { .. }
| LogicalPlan::CreateRelationship { .. }
| LogicalPlan::NodeScan { .. } => 1,
LogicalPlan::NodeCartesianProduct { left, right } => {
1 + logical_group_count(left) + logical_group_count(right)
}
}
}
enum LoweringChildren<'a> {
Memo {
memo: &'a GraphMemo,
children: &'a [GroupId],
},
Direct,
}
impl LoweringChildren<'_> {
fn lower(
&self,
logical: &LogicalPlan,
index: usize,
catalog: &OptimizerCatalog,
optimizer_context: &OptimizerContext,
decisions: &mut Vec<String>,
stage_events: &mut Vec<StageTrace>,
) -> PhysicalPlan {
match self {
Self::Memo { memo, children } => best_physical(
memo,
children[index],
catalog,
optimizer_context,
decisions,
stage_events,
),
Self::Direct => lower_logical(
logical,
Self::Direct,
catalog,
optimizer_context,
decisions,
stage_events,
),
}
}
}
fn lower_logical(
logical: &LogicalPlan,
children: LoweringChildren<'_>,
catalog: &OptimizerCatalog,
optimizer_context: &OptimizerContext,
decisions: &mut Vec<String>,
stage_events: &mut Vec<StageTrace>,
) -> PhysicalPlan {
if let Some(plan) = select_exact_count_fast_path(logical, decisions, stage_events) {
return plan;
}
if let Some(plan) = simple::lower_simple_logical(logical, optimizer_context) {
return plan;
}
match logical {
LogicalPlan::GraphMatch { program, input } => PhysicalPlan::GraphMatchExec {
program: program.clone(),
input: input.as_ref().map(|input| {
Box::new(children.lower(
input,
0,
catalog,
optimizer_context,
decisions,
stage_events,
))
}),
},
LogicalPlan::NodeCartesianProduct { left, right } => {
let left = children.lower(left, 0, catalog, optimizer_context, decisions, stage_events);
let right = children.lower(
right,
1,
catalog,
optimizer_context,
decisions,
stage_events,
);
let (left, right) =
order_single_row_cartesian_product_children(catalog, decisions, left, right);
push_cartesian_product_cost_decision(catalog, decisions, &left, &right);
PhysicalPlan::NodeCartesianProductExec {
left: Box::new(left),
right: Box::new(right),
}
}
LogicalPlan::NodeColumnLookup {
variable,
label,
property,
column,
optional,
input,
} => PhysicalPlan::NodeColumnLookupExec {
variable: variable.clone(),
label: label.clone(),
property: property.clone(),
column: column.clone(),
optional: *optional,
input: Box::new(children.lower(
input,
0,
catalog,
optimizer_context,
decisions,
stage_events,
)),
},
LogicalPlan::Expand {
source_variable,
source_label,
rel_variable,
rel_type,
rel_properties,
direction,
target_variable,
target_label,
min_hops,
max_hops,
optional,
input,
} => {
push_expand_estimate_decision(
catalog,
decisions,
ExpandEstimateRequest {
source_label,
rel_type,
rel_properties,
target_label,
min_hops: *min_hops,
max_hops: *max_hops,
},
);
let input = children.lower(
input,
0,
catalog,
optimizer_context,
decisions,
stage_events,
);
let graph_budget =
vector_seed_top_k(&input).map(|top_k| graph_expansion_budget(top_k, *max_hops));
if let Some(graph_budget) = graph_budget {
decisions.push(format!(
"bound vector-seeded graph expansion to {} candidates and {} payload bytes",
graph_budget.candidate_limit, graph_budget.payload_byte_limit
));
}
PhysicalPlan::AdjacencyExpandExec {
source_variable: source_variable.clone(),
source_label: source_label.clone(),
rel_variable: rel_variable.clone(),
rel_type: rel_type.clone(),
rel_properties: rel_properties.clone(),
direction: *direction,
target_variable: target_variable.clone(),
target_label: target_label.clone(),
min_hops: *min_hops,
max_hops: *max_hops,
optional: *optional,
graph_budget,
input: Box::new(input),
}
}
LogicalPlan::OptionalDegree {
source_variable,
rel_type,
rel_properties,
direction,
target_label,
target_properties,
alias,
input,
} => PhysicalPlan::OptionalDegreeExec {
source_variable: source_variable.clone(),
rel_type: rel_type.clone(),
rel_properties: rel_properties.clone(),
direction: *direction,
target_label: target_label.clone(),
target_properties: target_properties.clone(),
alias: alias.clone(),
input: Box::new(children.lower(
input,
0,
catalog,
optimizer_context,
decisions,
stage_events,
)),
},
LogicalPlan::OptionalRelationshipCountSum {
variable,
label,
properties,
legs,
output,
} => {
push_optional_relationship_count_sum_cost_decision(
catalog, decisions, label, properties, legs,
);
PhysicalPlan::OptionalRelationshipCountSumExec {
variable: variable.clone(),
label: label.clone(),
properties: properties.clone(),
legs: legs.clone(),
output: output.clone(),
}
}
LogicalPlan::Filter { predicate, input } => {
if let Some(plan) =
index_seek_from_filter(predicate, input, catalog, decisions, stage_events)
{
plan
} else if let Some(plan) = access::source_segment_scan_from_filter(predicate, input) {
decisions.push(
"choose SourceSegmentScan for storage-prunable Source filter".to_string(),
);
PhysicalPlan::FilterExec {
predicate: predicate.clone(),
input: Box::new(plan),
}
} else if let Predicate::BoundRelationshipExists {
source_variable,
rel_type,
direction,
target_variable,
} = predicate
{
decisions.push(
"lower bound relationship existence predicate to AdjacencyExistsExec"
.to_string(),
);
PhysicalPlan::AdjacencyExistsExec {
source_variable: source_variable.clone(),
rel_type: rel_type.clone(),
direction: *direction,
target_variable: target_variable.clone(),
input: Box::new(children.lower(
input,
0,
catalog,
optimizer_context,
decisions,
stage_events,
)),
}
} else {
let mut input = children.lower(
input,
0,
catalog,
optimizer_context,
decisions,
stage_events,
);
push_vector_seed_metadata_filter(&mut input, predicate, decisions);
input = join::lower_property_join(input, predicate, decisions);
PhysicalPlan::FilterExec {
predicate: predicate.clone(),
input: Box::new(input),
}
}
}
LogicalPlan::Project { items, input } => PhysicalPlan::ProjectExec {
items: items.clone(),
input: Box::new(children.lower(
input,
0,
catalog,
optimizer_context,
decisions,
stage_events,
)),
},
LogicalPlan::Aggregate {
group_keys,
items,
input,
} => PhysicalPlan::AggregateExec {
group_keys: group_keys.clone(),
items: items.clone(),
input: Box::new(children.lower(
input,
0,
catalog,
optimizer_context,
decisions,
stage_events,
)),
},
LogicalPlan::Distinct { input } => PhysicalPlan::DistinctExec {
input: Box::new(children.lower(
input,
0,
catalog,
optimizer_context,
decisions,
stage_events,
)),
},
LogicalPlan::Sort { items, input } => PhysicalPlan::SortExec {
items: items.clone(),
input: Box::new(children.lower(
input,
0,
catalog,
optimizer_context,
decisions,
stage_events,
)),
},
LogicalPlan::Limit {
offset,
limit,
input,
} => {
if *limit == Some(0) {
PhysicalPlan::EmptyExec
} else if let (Some(limit), LogicalPlan::Sort { items, input }) =
(limit, input.as_ref())
{
let input = children.lower(
input,
0,
catalog,
optimizer_context,
decisions,
stage_events,
);
select_bounded_sort_plan(items.clone(), *offset, *limit, input, catalog, decisions)
} else {
PhysicalPlan::LimitExec {
offset: *offset,
limit: *limit,
input: Box::new(children.lower(
input,
0,
catalog,
optimizer_context,
decisions,
stage_events,
)),
}
}
}
_ => unreachable!("leaf logical plans are lowered before child planning"),
}
}
fn select_exact_count_fast_path(
logical: &LogicalPlan,
decisions: &mut Vec<String>,
stage_events: &mut Vec<StageTrace>,
) -> Option<PhysicalPlan> {
if let Some(plan) = select_node_count_fast_path(logical, decisions, stage_events) {
return Some(plan);
}
select_relationship_count_fast_path(logical, decisions, stage_events)
}
fn select_node_count_fast_path(
logical: &LogicalPlan,
decisions: &mut Vec<String>,
stage_events: &mut Vec<StageTrace>,
) -> Option<PhysicalPlan> {
let LogicalPlan::Aggregate {
group_keys,
items,
input,
} = logical
else {
return None;
};
let [item] = items.as_slice() else {
return None;
};
let LogicalPlan::NodeScan { variable, label } = input.as_ref() else {
return None;
};
let exact_node_count = group_keys.is_empty()
&& item.function == AggregateFunction::Count
&& match &item.target {
AggregateTarget::All => !item.distinct,
AggregateTarget::Variable(target) => target == variable,
AggregateTarget::Property { .. }
| AggregateTarget::Column(_)
| AggregateTarget::ColumnProperty { .. } => false,
};
if !exact_node_count {
return None;
}
decisions.push(format!(
"selected implementation:node_count_fast_path: use exact label count for {label}"
));
stage_events
.push(ACCESS_PATH_SELECTION_STAGE.trace(StageStats::new(1, 1).with_rule_counts(1, 0)));
Some(PhysicalPlan::NodeCountExec {
label: label.clone(),
output: item.name.clone(),
})
}
fn select_relationship_count_fast_path(
logical: &LogicalPlan,
decisions: &mut Vec<String>,
stage_events: &mut Vec<StageTrace>,
) -> Option<PhysicalPlan> {
let LogicalPlan::Aggregate {
group_keys,
items,
input,
} = logical
else {
return None;
};
let [item] = items.as_slice() else {
return None;
};
let LogicalPlan::Expand {
source_variable,
source_label,
rel_variable,
rel_type,
rel_properties,
direction,
target_label,
min_hops,
max_hops,
optional,
input,
..
} = input.as_ref()
else {
return None;
};
let LogicalPlan::NodeScan { variable, label } = input.as_ref() else {
return None;
};
let exact_relationship_count = group_keys.is_empty()
&& item.function == AggregateFunction::Count
&& !item.distinct
&& source_variable == variable
&& source_label.is_empty()
&& label.is_empty()
&& target_label.is_empty()
&& rel_properties.is_empty()
&& *min_hops == 1
&& *max_hops == 1
&& !optional
&& matches!(
direction,
RelationshipDirection::Outgoing | RelationshipDirection::Incoming
)
&& match &item.target {
AggregateTarget::All => true,
AggregateTarget::Variable(target) => rel_variable.as_ref() == Some(target),
AggregateTarget::Property { .. }
| AggregateTarget::Column(_)
| AggregateTarget::ColumnProperty { .. } => false,
};
if !exact_relationship_count {
return None;
}
decisions.push(format!(
"selected implementation:relationship_count_fast_path: use exact relationship count for {rel_type}"
));
stage_events
.push(ACCESS_PATH_SELECTION_STAGE.trace(StageStats::new(1, 1).with_rule_counts(1, 0)));
Some(PhysicalPlan::RelationshipCountExec {
rel_type: rel_type.clone(),
output: item.name.clone(),
})
}
fn finalize_physical_plan(
plan: PhysicalPlan,
decisions: &mut Vec<String>,
stage_events: &mut Vec<StageTrace>,
) -> PhysicalPlan {
let event_count_before = stage_events.len();
let finalized = finalize_required_node_properties(plan, decisions, stage_events);
if !stage_events[event_count_before..]
.iter()
.any(|event| event.name() == PLAN_FINALIZATION_STAGE.name())
{
stage_events.push(PLAN_FINALIZATION_STAGE.trace(StageStats::new(1, 1)));
}
finalized
}
fn finalize_required_node_properties(
plan: PhysicalPlan,
decisions: &mut Vec<String>,
stage_events: &mut Vec<StageTrace>,
) -> PhysicalPlan {
match plan {
PhysicalPlan::ProjectExec { items, input } => {
select_node_projection_scan(&items, &input, decisions, stage_events)
.unwrap_or(PhysicalPlan::ProjectExec { items, input })
}
PhysicalPlan::AggregateExec {
group_keys,
items,
input,
} => {
let input = select_node_aggregate_required_scan(
&group_keys,
&items,
&input,
decisions,
stage_events,
)
.map(Box::new)
.unwrap_or(input);
PhysicalPlan::AggregateExec {
group_keys,
items,
input,
}
}
PhysicalPlan::LimitExec {
offset,
limit,
input,
} => PhysicalPlan::LimitExec {
offset,
limit,
input: Box::new(finalize_required_node_properties(
*input,
decisions,
stage_events,
)),
},
PhysicalPlan::DistinctExec { input } => PhysicalPlan::DistinctExec {
input: Box::new(finalize_required_node_properties(
*input,
decisions,
stage_events,
)),
},
PhysicalPlan::SortExec { items, input }
if sort_items_use_only_projected_columns(&items) =>
{
PhysicalPlan::SortExec {
items,
input: Box::new(finalize_required_node_properties(
*input,
decisions,
stage_events,
)),
}
}
PhysicalPlan::TopNExec {
items,
offset,
limit,
input,
} if sort_items_use_only_projected_columns(&items) => PhysicalPlan::TopNExec {
items,
offset,
limit,
input: Box::new(finalize_required_node_properties(
*input,
decisions,
stage_events,
)),
},
plan => plan,
}
}
fn sort_items_use_only_projected_columns(items: &[SortItem]) -> bool {
items
.iter()
.all(|item| matches!(item.key, SortKey::Column(_)))
}
fn select_node_projection_scan(
items: &[Projection],
input: &PhysicalPlan,
decisions: &mut Vec<String>,
stage_events: &mut Vec<StageTrace>,
) -> Option<PhysicalPlan> {
let (source, predicate) = match input {
PhysicalPlan::FilterExec { predicate, input } => (input.as_ref(), Some(predicate)),
source => (source, None),
};
let (variable, label, access) = node_projection_access(source)?;
let mut required_properties = BTreeSet::new();
if !items.iter().all(|item| {
collect_projection_properties(&item.expression, variable, &mut required_properties)
}) || predicate.is_some_and(|predicate| {
!collect_predicate_properties(predicate, variable, &mut required_properties)
}) {
return None;
}
collect_access_properties(&access, &mut required_properties);
let required_properties = required_properties.into_iter().collect::<Vec<_>>();
decisions.push(format!(
"selected implementation:node_projection_scan: access={} decode {} required properties for {variable}:{label}",
access.kind_name(),
required_properties.len(),
));
stage_events.push(PLAN_FINALIZATION_STAGE.trace(StageStats::new(1, 1).with_rule_counts(1, 0)));
Some(PhysicalPlan::NodeProjectionScanExec {
variable: variable.clone(),
label: label.clone(),
access,
required_properties,
predicate: predicate.cloned(),
items: items.to_vec(),
})
}
fn select_node_aggregate_required_scan(
group_keys: &[Projection],
items: &[hawdb_plan::Aggregation],
input: &PhysicalPlan,
decisions: &mut Vec<String>,
stage_events: &mut Vec<StageTrace>,
) -> Option<PhysicalPlan> {
let (source, predicate) = match input {
PhysicalPlan::FilterExec { predicate, input } => (input.as_ref(), Some(predicate)),
source => (source, None),
};
let (variable, label, access) = node_projection_access(source)?;
let mut required_properties = BTreeSet::new();
if !group_keys.iter().all(|item| {
collect_projection_properties(&item.expression, variable, &mut required_properties)
}) || !items
.iter()
.all(|item| collect_aggregate_properties(item, variable, &mut required_properties))
|| predicate.is_some_and(|predicate| {
!collect_predicate_properties(predicate, variable, &mut required_properties)
})
{
return None;
}
collect_access_properties(&access, &mut required_properties);
let required_properties = required_properties.into_iter().collect::<Vec<_>>();
decisions.push(format!(
"selected implementation:node_aggregate_required_scan: access={} decode {} required properties for {variable}:{label}",
access.kind_name(),
required_properties.len(),
));
stage_events.push(PLAN_FINALIZATION_STAGE.trace(StageStats::new(1, 1).with_rule_counts(1, 0)));
Some(PhysicalPlan::NodeProjectionScanExec {
variable: variable.clone(),
label: label.clone(),
access,
required_properties,
predicate: predicate.cloned(),
items: Vec::new(),
})
}
fn collect_aggregate_properties(
item: &hawdb_plan::Aggregation,
variable: &str,
required: &mut BTreeSet<String>,
) -> bool {
match (&item.function, &item.target) {
(AggregateFunction::Count, AggregateTarget::All) => true,
(AggregateFunction::Count, AggregateTarget::Variable(current)) => current == variable,
(
AggregateFunction::Count
| AggregateFunction::Min
| AggregateFunction::Max
| AggregateFunction::Avg
| AggregateFunction::Collect,
AggregateTarget::Property {
variable: current,
property,
},
) if current == variable => {
required.insert(property.clone());
true
}
_ => false,
}
}
fn node_projection_access(plan: &PhysicalPlan) -> Option<(&String, &String, NodeProjectionAccess)> {
match plan {
PhysicalPlan::SeqNodeScan { variable, label } => {
Some((variable, label, NodeProjectionAccess::LabelScan))
}
PhysicalPlan::IndexNodeSeek {
variable,
label,
property,
value,
} => Some((
variable,
label,
NodeProjectionAccess::PropertyValues {
property: property.clone(),
values: vec![value.clone()],
},
)),
PhysicalPlan::IndexNodeMultiSeek {
variable,
label,
property,
values,
} => Some((
variable,
label,
NodeProjectionAccess::PropertyValues {
property: property.clone(),
values: values.clone(),
},
)),
PhysicalPlan::IndexNodeUnionSeek {
variable,
label,
branches,
} => Some((
variable,
label,
NodeProjectionAccess::PropertyUnion {
branches: branches.clone(),
},
)),
PhysicalPlan::IndexNodeCompositeSeek {
variable,
label,
predicates,
} => Some((
variable,
label,
NodeProjectionAccess::CompositeEquality {
predicates: predicates.clone(),
},
)),
PhysicalPlan::IndexNodeCompositeRangeSeek {
variable,
label,
seek,
} => Some((
variable,
label,
NodeProjectionAccess::CompositeRange { seek: seek.clone() },
)),
PhysicalPlan::IndexNodeRangeSeek {
variable,
label,
property,
lower,
upper,
} => Some((
variable,
label,
NodeProjectionAccess::PropertyRange {
property: property.clone(),
lower: lower.clone(),
upper: upper.clone(),
},
)),
PhysicalPlan::IndexNodeTextSeek {
variable,
label,
property,
query,
} => Some((
variable,
label,
NodeProjectionAccess::FullText {
property: property.clone(),
query: query.clone(),
},
)),
_ => None,
}
}
fn collect_access_properties(access: &NodeProjectionAccess, required: &mut BTreeSet<String>) {
match access {
NodeProjectionAccess::LabelScan => {}
NodeProjectionAccess::PropertyValues { property, .. }
| NodeProjectionAccess::PropertyRange { property, .. }
| NodeProjectionAccess::FullText { property, .. } => {
required.insert(property.clone());
}
NodeProjectionAccess::CompositeEquality { predicates } => {
required.extend(predicates.iter().map(|(property, _)| property.clone()));
}
NodeProjectionAccess::CompositeRange { seek } => {
required.extend(
seek.equality_prefix
.iter()
.map(|(property, _)| property.clone()),
);
required.insert(seek.range_property.clone());
}
NodeProjectionAccess::PropertyUnion { branches } => {
required.extend(branches.iter().map(|branch| branch.property.clone()));
}
}
}
fn collect_predicate_properties(
predicate: &Predicate,
variable: &str,
required: &mut BTreeSet<String>,
) -> bool {
match predicate {
Predicate::And(predicates) | Predicate::Or(predicates) => predicates
.iter()
.all(|predicate| collect_predicate_properties(predicate, variable, required)),
Predicate::Not(predicate) => collect_predicate_properties(predicate, variable, required),
Predicate::ConstantBool(_) => true,
Predicate::IdEq {
variable: current, ..
}
| Predicate::IdNotEq {
variable: current, ..
}
| Predicate::IdCompare {
variable: current, ..
}
| Predicate::IdIn {
variable: current, ..
} => current == variable,
Predicate::PropertyEq {
variable: current,
property,
..
}
| Predicate::PropertyNotEq {
variable: current,
property,
..
}
| Predicate::PropertyCompare {
variable: current,
property,
..
}
| Predicate::PropertyListContains {
variable: current,
property,
..
}
| Predicate::PropertyListContainsLower {
variable: current,
property,
..
}
| Predicate::PropertyContains {
variable: current,
property,
..
}
| Predicate::PropertyStartsWith {
variable: current,
property,
..
}
| Predicate::PropertyEndsWith {
variable: current,
property,
..
}
| Predicate::PropertyRegexMatch {
variable: current,
property,
..
}
| Predicate::PropertyIsNull {
variable: current,
property,
}
| Predicate::PropertyIsNotNull {
variable: current,
property,
}
| Predicate::PropertyIn {
variable: current,
property,
..
} => {
if current != variable {
return false;
}
required.insert(property.clone());
true
}
Predicate::ExpressionEq { expression, value }
| Predicate::ExpressionNotEq { expression, value }
| Predicate::ExpressionContains { expression, value } => {
collect_projection_properties(expression, variable, required)
&& collect_projection_properties(value, variable, required)
}
Predicate::ExpressionCompare {
expression, value, ..
} => {
collect_projection_properties(expression, variable, required)
&& collect_projection_properties(value, variable, required)
}
Predicate::RelationshipExists { .. } | Predicate::BoundRelationshipExists { .. } => false,
}
}
fn collect_projection_properties(
expression: &ProjectionExpression,
variable: &str,
required: &mut BTreeSet<String>,
) -> bool {
match expression {
ProjectionExpression::Case { .. }
| ProjectionExpression::Binary { .. }
| ProjectionExpression::Not(_)
| ProjectionExpression::IsNull { .. } => expression
.all_children(|child| collect_projection_properties(child, variable, required)),
ProjectionExpression::Variable { .. }
| ProjectionExpression::RelationshipType { .. }
| ProjectionExpression::Column(_)
| ProjectionExpression::ColumnProperty { .. }
| ProjectionExpression::ColumnDefaultIfNullOrEq { .. }
| ProjectionExpression::ColumnValueDefaultIfNull { .. }
| ProjectionExpression::ColumnValueCasePropertyNotNullOrEq { .. }
| ProjectionExpression::CaseColumnSearchRank(_) => false,
ProjectionExpression::Id { variable: current } => current == variable,
ProjectionExpression::Literal(_) => true,
ProjectionExpression::Property {
variable: current,
property,
}
| ProjectionExpression::DatePart {
variable: current,
property,
..
}
| ProjectionExpression::DefaultIfNullOrEq {
variable: current,
property,
..
}
| ProjectionExpression::DefaultIfNull {
variable: current,
property,
..
}
| ProjectionExpression::CasePropertyNotNullOrEq {
variable: current,
property,
..
}
| ProjectionExpression::CasePropertyEqualsRank {
variable: current,
property,
..
}
| ProjectionExpression::CaseLowerPropertyDefault {
variable: current,
property,
..
} => {
if current != variable {
return false;
}
required.insert(property.clone());
true
}
ProjectionExpression::Coalesce(expressions) => expressions
.iter()
.all(|expression| collect_projection_properties(expression, variable, required)),
ProjectionExpression::Left { expression, .. } | ProjectionExpression::Lower(expression) => {
collect_projection_properties(expression, variable, required)
}
ProjectionExpression::CaseCoalesceDifferenceFloorZero {
variable: current,
terms,
} => {
if current != variable {
return false;
}
required.extend(terms.iter().map(|term| term.property.clone()));
true
}
ProjectionExpression::CaseEntitySearchRank(rank) => {
if rank.variable != variable {
return false;
}
required.insert(rank.name_property.clone());
required.insert(rank.aliases_property.clone());
true
}
}
}
struct ExpandEstimateRequest<'a> {
source_label: &'a str,
rel_type: &'a str,
rel_properties: &'a BTreeMap<String, Value>,
target_label: &'a str,
min_hops: usize,
max_hops: usize,
}
fn push_expand_estimate_decision(
catalog: &OptimizerCatalog,
decisions: &mut Vec<String>,
request: ExpandEstimateRequest<'_>,
) {
let estimate = catalog.estimate_expand_rows(
request.source_label,
request.rel_type,
request.rel_properties,
request.target_label,
request.min_hops,
request.max_hops,
);
let path_count = estimate
.path_count
.map(|count| count.to_string())
.unwrap_or_else(|| "unknown".to_string());
let hop_rows = estimate
.hop_estimates
.iter()
.map(|estimate| {
format!(
"{}:{}:{}",
estimate.hop,
if estimate.exact { "exact" } else { "fallback" },
estimate.rows
)
})
.collect::<Vec<_>>()
.join(",");
decisions.push(format!(
"estimate AdjacencyExpand for {}-[:{}*{}..{}]->{}: path_count={path_count} rel_count={} rel_type_sources={} average_fanout={} rel_property_distinct_product={} hop_rows=[{}] estimated_rows={}",
request.source_label,
request.rel_type,
request.min_hops,
request.max_hops,
request.target_label,
estimate.rel_count,
estimate.source_count,
estimate.average_fanout,
estimate.property_distinct_product,
hop_rows,
estimate.estimated_rows
));
}
fn order_single_row_cartesian_product_children(
catalog: &OptimizerCatalog,
decisions: &mut Vec<String>,
left: PhysicalPlan,
right: PhysicalPlan,
) -> (PhysicalPlan, PhysicalPlan) {
let left_cost = estimate_physical_plan_cost(&left, catalog);
let right_cost = estimate_physical_plan_cost(&right, catalog);
if !cartesian_product_leaves_are_single_row(catalog, &left)
|| !cartesian_product_leaves_are_single_row(catalog, &right)
{
decisions.push(format!(
"keep NodeCartesianProduct input order: left_rows={} right_rows={} reason=non_single_row_input",
left_cost.estimated_rows, right_cost.estimated_rows
));
return (left, right);
}
let mut leaves = Vec::new();
collect_cartesian_product_leaves(left, &mut leaves);
collect_cartesian_product_leaves(right, &mut leaves);
let mut keyed_leaves = leaves
.into_iter()
.map(|plan| {
let cost = estimate_physical_plan_cost(&plan, catalog);
let fingerprint = plan.instance_fingerprint();
(cost, fingerprint, plan)
})
.collect::<Vec<_>>();
let original_order = keyed_leaves
.iter()
.map(|(cost, fingerprint, _)| format!("{}:{}", cost.cost, fingerprint))
.collect::<Vec<_>>()
.join("|");
keyed_leaves.sort_by(
|(left_cost, left_fingerprint, _), (right_cost, right_fingerprint, _)| {
(left_cost.cost, left_fingerprint).cmp(&(right_cost.cost, right_fingerprint))
},
);
let ordered = keyed_leaves
.iter()
.map(|(cost, fingerprint, _)| format!("{}:{}", cost.cost, fingerprint))
.collect::<Vec<_>>()
.join("|");
if original_order == ordered {
decisions.push(format!(
"keep NodeCartesianProduct single-row input order: inputs={} order={ordered}",
keyed_leaves.len()
));
} else {
decisions.push(format!(
"order NodeCartesianProduct single-row inputs: inputs={} original={original_order} ordered={ordered}",
keyed_leaves.len()
));
}
let mut ordered_plans = keyed_leaves.into_iter().map(|(_, _, plan)| plan);
let left = ordered_plans
.next()
.expect("cartesian product has left input");
let right = rebuild_cartesian_product(ordered_plans);
(left, right)
}
fn cartesian_product_leaves_are_single_row(
catalog: &OptimizerCatalog,
plan: &PhysicalPlan,
) -> bool {
match plan {
PhysicalPlan::NodeCartesianProductExec { left, right } => {
cartesian_product_leaves_are_single_row(catalog, left)
&& cartesian_product_leaves_are_single_row(catalog, right)
}
plan => estimate_physical_plan_cost(plan, catalog).estimated_rows == 1,
}
}
fn collect_cartesian_product_leaves(plan: PhysicalPlan, leaves: &mut Vec<PhysicalPlan>) {
match plan {
PhysicalPlan::NodeCartesianProductExec { left, right } => {
collect_cartesian_product_leaves(*left, leaves);
collect_cartesian_product_leaves(*right, leaves);
}
plan => leaves.push(plan),
}
}
fn rebuild_cartesian_product(plans: impl IntoIterator<Item = PhysicalPlan>) -> PhysicalPlan {
let mut plans = plans.into_iter();
let first = plans
.next()
.expect("cartesian product rebuild requires at least one input");
plans.fold(first, |left, right| {
PhysicalPlan::NodeCartesianProductExec {
left: Box::new(left),
right: Box::new(right),
}
})
}
fn push_cartesian_product_cost_decision(
catalog: &OptimizerCatalog,
decisions: &mut Vec<String>,
left: &PhysicalPlan,
right: &PhysicalPlan,
) {
let left_cost = estimate_physical_plan_cost(left, catalog);
let right_cost = estimate_physical_plan_cost(right, catalog);
let cost = estimate_node_cartesian_product_cost(left_cost, right_cost);
decisions.push(format!(
"estimate NodeCartesianProduct: left_rows={} right_rows={} output_rows={} left_cost={} right_cost={} cost={}",
left_cost.estimated_rows,
right_cost.estimated_rows,
cost.estimated_rows,
left_cost.cost,
right_cost.cost,
cost.cost
));
}