use super::*;
use hawdb_cypher::{OrderExpression as CypherOrderExpression, ProcedureCallKind, YieldItem};
mod vector_index;
pub(super) use vector_index::bind_vector_index;
use vector_index::bind_vector_index_pipeline;
pub(super) fn bind_procedure_pipeline(
query: &QueryPipeline,
parameters: &BTreeMap<String, Value>,
) -> Result<LogicalPlan> {
let ClauseKind::Call { procedure, yields } = &query.clauses[0].kind else {
unreachable!()
};
let mut tail = &query.clauses[1..];
let has_graph_read = tail
.iter()
.any(|clause| matches!(clause.kind, ClauseKind::Match { .. }));
let (mut input, columns) = match &procedure.kind {
ProcedureCallKind::ProjectGraph {
name,
node_labels,
rel_types,
relationship_predicates,
} => {
if !yields.is_empty() || !tail.is_empty() {
return Err(unsupported("project_graph does not yield query rows"));
}
return Ok(LogicalPlan::ProjectGraph {
name: name.clone(),
node_labels: node_labels.clone(),
rel_types: rel_types.clone(),
relationship_predicates: bind_projected_relationship_predicates(
relationship_predicates,
)?,
});
}
ProcedureCallKind::GraphAlgorithm {
algorithm,
graph_name,
options,
} => {
let default_score = match algorithm {
CypherGraphAlgorithmKind::PageRank => "pagerank_score",
CypherGraphAlgorithmKind::Louvain => "louvain_id",
};
let mut score = default_score.to_string();
let requested = graph_algorithm_requested_columns(yields, tail);
let mut identity = GraphAlgorithmIdentity {
node_id: requested.contains("node_id"),
node_label: requested.contains("node_label"),
node_labels: requested.contains("node_labels"),
};
if *algorithm == CypherGraphAlgorithmKind::PageRank
&& requested.contains("rank")
&& !requested.contains(default_score)
{
score = "rank".to_string();
}
let mut legacy_shape = false;
if yields.is_empty()
&& let [clause] = tail
&& let ClauseKind::Return(projection) = &clause.kind
&& let Some(names) = identity_columns(projection)
&& names
.first()
.is_some_and(|name| name.eq_ignore_ascii_case("node"))
&& let Some((legacy_identity, score_name)) =
graph_algorithm_return_shape(*algorithm, &names)
&& (score_name.eq_ignore_ascii_case(default_score)
|| (*algorithm == CypherGraphAlgorithmKind::PageRank
&& score_name.eq_ignore_ascii_case("rank")))
{
identity = GraphAlgorithmIdentity::legacy(legacy_identity);
legacy_shape = true;
score = score_name.to_string();
tail = &[];
}
let mut columns = vec!["node".to_string()];
columns.extend(identity.columns().map(str::to_string));
if *algorithm == CypherGraphAlgorithmKind::Louvain {
columns.push("level".to_string());
}
columns.push(score.clone());
let input = if legacy_shape || identity.is_empty() {
LogicalPlan::GraphAlgorithm {
algorithm: plan_graph_algorithm_kind(*algorithm),
graph_name: graph_name.clone(),
options: bind_graph_algorithm_options(options, parameters)?,
score_column: score,
return_node_identity: identity.node_id,
node_visibility_predicate: None,
}
} else {
LogicalPlan::GraphAlgorithmOutput(Box::new(GraphAlgorithmOutput {
algorithm: plan_graph_algorithm_kind(*algorithm),
graph_name: graph_name.clone(),
options: bind_graph_algorithm_options(options, parameters)?,
score_column: score,
identity,
node_visibility_predicate: None,
}))
};
(input, columns)
}
ProcedureCallKind::GraphSeedSearch(search) => {
return bind_graph_seed_pipeline(search, yields, tail, parameters);
}
ProcedureCallKind::VectorIndexSearch(search) => {
return bind_vector_index_pipeline(search, yields, tail, parameters);
}
ProcedureCallKind::TextSearch(search) => {
if yields.is_empty() && has_graph_read {
return Err(unsupported("text search requires YIELD before MATCH"));
}
let seeded_match = !yields.is_empty() && has_graph_read;
let input = bind_text_seed(search, parameters, seeded_match)?;
(input, vec!["id".to_string(), "score".to_string()])
}
ProcedureCallKind::VectorSearch(search) => {
let input = bind_vector_seed(search, parameters, !yields.is_empty())?;
if yields.is_empty() && has_graph_read {
return Err(unsupported("vector search requires YIELD before MATCH"));
}
if yields.is_empty()
&& let [clause] = tail
&& let ClauseKind::Return(projection) = &clause.kind
&& identity_columns(projection).is_some_and(|names| {
names.len() == 2
&& names[0].eq_ignore_ascii_case("id")
&& names[1].eq_ignore_ascii_case("score")
})
{
tail = &[];
}
(input, vec!["id".to_string(), "score".to_string()])
}
};
let seeded_match = !yields.is_empty()
&& (matches!(procedure.kind, ProcedureCallKind::VectorSearch(_))
|| (matches!(procedure.kind, ProcedureCallKind::TextSearch(_)) && has_graph_read));
let producer = if matches!(procedure.kind, ProcedureCallKind::TextSearch(_)) {
"text"
} else {
"vector"
};
let (next, mut scope) = bind_yields(input, &columns, yields, seeded_match)?;
input = next;
if seeded_match {
let Some((clause, rest)) = tail.split_first() else {
return Err(unsupported(&format!(
"{producer} search YIELD must feed a MATCH read query"
)));
};
let ClauseKind::Match {
optional: false,
patterns,
predicate,
} = &clause.kind
else {
return Err(unsupported(&format!(
"{producer} search YIELD must feed a MATCH read query"
)));
};
let [pattern] = patterns.as_slice() else {
return Err(unsupported(&format!(
"{producer} search requires one seeded MATCH pattern"
)));
};
let hops = tail.iter().fold(0usize, |hops, clause| match &clause.kind {
ClauseKind::Match { patterns, .. } => patterns.iter().fold(hops, |hops, pattern| {
pattern.steps.iter().fold(hops, |hops, step| {
hops.saturating_add(step.relationship.max_hops)
})
}),
_ => hops,
});
if hops > MAX_RETRIEVER_SEEDED_GRAPH_HOPS {
return Err(unsupported(&format!("{producer}-seeded graph expansion supports at most {MAX_RETRIEVER_SEEDED_GRAPH_HOPS} hops")));
}
scope.bind_graph(
&pattern.first.variable,
GraphEntityKind::Node,
&mut Vec::new(),
)?;
input = LogicalPlan::NodeColumnLookup {
variable: pattern.first.variable.clone(),
label: pattern.first.label.clone(),
property: "id".to_string(),
column: "external_id".to_string(),
optional: false,
node_visibility_predicate: None,
input: Box::new(input),
};
if pattern.first.properties.is_empty() && pattern.steps.is_empty() {
if let Some(predicate) = predicate {
input = LogicalPlan::Filter {
predicate: bind_predicate(&predicate.kind, &scope, parameters)?,
input: Box::new(input),
};
}
tail = rest;
}
}
bind_read_clauses(tail, Some(input), scope, parameters)
}
fn identity_columns(projection: &ProjectionClause) -> Option<Vec<String>> {
if projection.distinct
|| projection.predicate.is_some()
|| !projection.order_by.is_empty()
|| projection.offset.is_some()
|| projection.limit.is_some()
{
return None;
}
projection
.items
.iter()
.map(|item| match &item.expression.kind {
ReturnExpressionKind::Value(AstNode {
kind: ScalarExpressionKind::Variable(name),
..
}) if item.alias.is_none() => Some(name.clone()),
_ => None,
})
.collect()
}
fn graph_algorithm_return_shape(
algorithm: CypherGraphAlgorithmKind,
names: &[String],
) -> Option<(bool, &str)> {
let mut index = 1;
let identity = names
.get(index)
.is_some_and(|name| name.eq_ignore_ascii_case("node_id"));
if identity {
if !names
.get(index + 1)
.is_some_and(|name| name.eq_ignore_ascii_case("node_label"))
{
return None;
}
index += 2;
} else if names
.get(index)
.is_some_and(|name| name.eq_ignore_ascii_case("node_label"))
{
return None;
}
if algorithm == CypherGraphAlgorithmKind::Louvain
&& names
.get(index)
.is_some_and(|name| name.eq_ignore_ascii_case("level"))
{
index += 1;
}
(index + 1 == names.len()).then(|| (identity, names[index].as_str()))
}
fn bind_yields(
input: LogicalPlan,
columns: &[String],
yields: &[YieldItem],
preserve_external_id: bool,
) -> Result<(LogicalPlan, Scope)> {
let mut scope = Scope::default();
if yields.is_empty() {
scope.0.extend(
columns
.iter()
.map(|name| (name.clone(), BindingType::Scalar)),
);
return Ok((input, scope));
}
let mut items = Vec::new();
let mut identity = yields.len() == columns.len();
for (index, item) in yields.iter().enumerate() {
let Some(column) = columns
.iter()
.find(|column| column.eq_ignore_ascii_case(&item.name))
else {
return Err(unsupported("YIELD references an unknown procedure column"));
};
let name = item.alias.as_ref().unwrap_or(column);
if scope.0.insert(name.clone(), BindingType::Scalar).is_some()
|| (preserve_external_id && name == "external_id")
{
return Err(unsupported(
"YIELD aliases must be distinct procedure columns",
));
}
identity &= columns.get(index) == Some(name) && name == column;
items.push(Projection {
expression: ProjectionExpression::Column(column.clone()),
name: name.clone(),
});
}
if identity {
return Ok((input, scope));
}
if preserve_external_id {
items.push(Projection {
expression: ProjectionExpression::Column("external_id".to_string()),
name: "external_id".to_string(),
});
}
Ok((
LogicalPlan::Project {
items,
input: Box::new(input),
},
scope,
))
}
fn unsupported(message: &str) -> HawDBError {
HawDBError::Semantic(message.to_string())
}
fn bind_graph_seed_pipeline(
search: &hawdb_cypher::GraphSeedSearch,
yields: &[YieldItem],
tail: &[Clause],
parameters: &BTreeMap<String, Value>,
) -> Result<LogicalPlan> {
let ValueExpressionKind::Parameter(query_parameter) = &search.query.kind else {
return Err(unsupported("graph seed query must be a string parameter"));
};
if !matches!(parameters.get(query_parameter), Some(Value::String(_))) {
return Err(unsupported("graph seed query must be a string parameter"));
}
let Value::String(label) = bind_value(&search.label, parameters)? else {
return Err(unsupported("graph seed label must be a nonempty string"));
};
if label.is_empty() || label.contains('\0') {
return Err(unsupported(
"graph seed label must be a nonempty string without NUL",
));
}
let top_k = search
.top_k
.as_ref()
.map(|value| bind_non_negative_usize(value, parameters, "topK"))
.transpose()?
.unwrap_or(10);
let mut scope = Scope::default();
let mut variable = "node".to_string();
let mut score_column = "score".to_string();
let mut seen = BTreeSet::new();
for item in yields {
let column = item.name.to_ascii_lowercase();
if !seen.insert(column.clone()) {
return Err(unsupported("graph seed YIELD columns must be distinct"));
}
let alias = item.alias.as_ref().unwrap_or(&column);
if alias.contains('\0') || scope.0.contains_key(alias) {
return Err(unsupported(
"graph seed YIELD aliases must be distinct and public",
));
}
match column.as_str() {
"node" => {
variable = alias.clone();
scope.bind_graph(alias, GraphEntityKind::Node, &mut Vec::new())?;
}
"score" => {
score_column = alias.clone();
scope.0.insert(alias.clone(), BindingType::Scalar);
}
_ => return Err(unsupported("unknown graph seed YIELD column")),
}
}
if yields.is_empty() {
scope.bind_graph(&variable, GraphEntityKind::Node, &mut Vec::new())?;
scope.0.insert(score_column.clone(), BindingType::Scalar);
}
let hops = tail.iter().fold(0usize, |hops, clause| match &clause.kind {
ClauseKind::Match { patterns, .. } => patterns.iter().fold(hops, |hops, pattern| {
pattern.steps.iter().fold(hops, |hops, step| {
hops.saturating_add(step.relationship.max_hops)
})
}),
_ => hops,
});
if hops > MAX_RETRIEVER_SEEDED_GRAPH_HOPS {
return Err(unsupported(
"graph-seeded expansion supports at most two hops",
));
}
let input = LogicalPlan::GraphSeed {
query_parameter: query_parameter.clone(),
label,
variable,
score_column,
top_k,
node_visibility_predicate: None,
};
if tail.is_empty() {
let items = scope
.0
.iter()
.map(|(name, kind)| Projection {
name: name.clone(),
expression: match kind {
BindingType::Graph { .. } => ProjectionExpression::Variable {
variable: name.clone(),
},
_ => ProjectionExpression::Column(name.clone()),
},
})
.collect();
return Ok(LogicalPlan::Project {
items,
input: Box::new(input),
});
}
bind_read_clauses(tail, Some(input), scope, parameters)
}
fn graph_algorithm_requested_columns(yields: &[YieldItem], tail: &[Clause]) -> BTreeSet<String> {
if !yields.is_empty() {
return yields
.iter()
.map(|item| item.name.to_ascii_lowercase())
.collect();
}
let mut names = BTreeSet::new();
if let Some(ClauseKind::Return(projection) | ClauseKind::With(projection)) =
tail.first().map(|clause| &clause.kind)
{
for item in &projection.items {
collect_algorithm_return_columns(&item.expression, &mut names);
}
let mut additional = BTreeSet::new();
if let Some(predicate) = &projection.predicate {
collect_predicate_variables(predicate, &mut additional);
}
for item in &projection.order_by {
match &item.expression {
CypherOrderExpression::Value(value) => {
collect_scalar_expression_variables(value, &mut additional)
}
CypherOrderExpression::Column(name)
| CypherOrderExpression::Property { variable: name, .. }
| CypherOrderExpression::Id { variable: name } => {
additional.insert(name.clone());
}
}
}
additional.retain(|name| {
!projection
.items
.iter()
.any(|item| item.alias.as_ref() == Some(name))
});
names.extend(additional);
}
names
}
fn collect_algorithm_return_columns(
expression: &hawdb_cypher::ReturnExpression,
names: &mut BTreeSet<String>,
) {
use hawdb_cypher::AggregateExpression;
match &expression.kind {
ReturnExpressionKind::Value(value) => collect_scalar_expression_variables(value, names),
ReturnExpressionKind::Arithmetic { first, rest } => {
collect_algorithm_return_columns(first, names);
for (_, expression) in rest {
collect_algorithm_return_columns(expression, names);
}
}
ReturnExpressionKind::Aggregate(aggregate) => match aggregate {
AggregateExpression::CountAll => {}
AggregateExpression::CollectTop {
document_id, score, ..
} => {
names.insert(document_id.clone());
names.insert(score.clone());
}
AggregateExpression::MaxVariable { variable }
| AggregateExpression::CountVariable { variable, .. }
| AggregateExpression::CountProperty { variable, .. }
| AggregateExpression::CollectProperty { variable, .. }
| AggregateExpression::CollectVariable { variable, .. }
| AggregateExpression::MinProperty { variable, .. }
| AggregateExpression::MaxProperty { variable, .. }
| AggregateExpression::AvgProperty { variable, .. } => {
names.insert(variable.clone());
}
},
ReturnExpressionKind::Path(_) => {}
}
}