use super::*;
use crate::graph_match::{
import_identity, visit_correlated_matches, CorrelatedInput, ImportIdentity, MatchContext,
ResolvedImport,
};
use crate::store::{GraphExecutionRead, ScanControl};
use hawdb_plan_cypher::Predicate;
use hawdb_plan_cypher::{
GraphBindingImport, GraphEntityKind, GraphMatchProgram, GraphMatchStep, PatternProjection,
};
use hawdb_storage::read_view::{AdmittedNodeRead, AdmittedRelationshipRead};
use hawdb_storage::{NodeId, RelId, RelRecord};
use std::borrow::Cow;
use std::collections::BTreeSet;
#[derive(Clone, Copy)]
pub(crate) struct ExpressionGraph<'a> {
pub(crate) store: &'a dyn GraphExecutionRead,
pub(crate) context: MatchContext<'a>,
pub(crate) account: Option<&'a QueryMemoryAccount>,
}
impl<'a> ExpressionGraph<'a> {
pub(crate) fn new(
store: &'a dyn GraphExecutionRead,
context: crate::pipeline::BatchExecutionContext<'a>,
) -> Self {
Self {
store,
context: context.into(),
account: None,
}
}
pub(crate) fn with_account(self, account: &'a QueryMemoryAccount) -> Self {
Self {
account: Some(account),
..self
}
}
}
pub(super) struct Scope<'a, 'w> {
pub(super) catalog: &'a Catalog,
pub(super) binding: &'a Binding,
pub(super) locals: Option<&'a ListLocals<'a>>,
pub(super) account: Option<&'w QueryMemoryAccount>,
pub(super) work: &'w ListWork,
}
fn graph_access<'g>(
what: &str,
graph: Option<&'g ExpressionGraph<'g>>,
) -> Result<&'g ExpressionGraph<'g>> {
graph.ok_or_else(|| {
HawDBError::Execution(format!(
"{what} require graph access, which this expression position does not provide"
))
})
}
enum Charged<'a> {
Shared(&'a QueryMemoryAccount),
Private(QueryMemoryAccount),
}
impl Charged<'_> {
fn account(&self) -> &QueryMemoryAccount {
match self {
Self::Shared(account) => account,
Self::Private(account) => account,
}
}
}
fn charged<'a>(
account: Option<&'a QueryMemoryAccount>,
graph: &'a ExpressionGraph<'a>,
) -> Charged<'a> {
match account.or(graph.account) {
Some(account) => Charged::Shared(account),
None => {
let budget = graph.context.query_memory_bytes;
Charged::Private(crate::QueryMemoryLedger::new(budget).account(
crate::QueryMemoryClass::BlockingState,
"pattern expression",
budget,
))
}
}
}
pub(super) fn evaluate<'a>(
pattern: &'a PatternProjection,
scope: Scope<'a, '_>,
graph: Option<&ExpressionGraph<'_>>,
) -> Result<ProjectedValue<'a>> {
let graph = graph_access("pattern expressions", graph)?;
let charged = charged(scope.account, graph);
let account = charged.account();
crate::pipeline::runtime_checkpoint(graph.context.task_context)?;
scope.work.charge(1)?;
let mut row = Binding::values(BTreeMap::new());
for name in &pattern.graph_variables {
let copied = match graph_variable_kind(&pattern.program, name) {
Some(GraphEntityKind::Node) => {
kind_matches(scope.binding, name, GraphEntityKind::Node)?;
row.nodes.insert_shared(&scope.binding.nodes, name)
}
Some(GraphEntityKind::Relationship) => {
kind_matches(scope.binding, name, GraphEntityKind::Relationship)?;
row.relationships
.insert_shared(&scope.binding.relationships, name)
}
None => {
row.nodes.insert_shared(&scope.binding.nodes, name)
|| row
.relationships
.insert_shared(&scope.binding.relationships, name)
}
};
if copied {
continue;
}
if scope.binding.values.get(name) != Some(&Value::Null) {
return Err(HawDBError::Execution(format!(
"missing variable '{name}' during pattern evaluation"
)));
}
row.values.insert(name.clone(), Value::Null);
}
let mut identities = Vec::new();
let mut _arguments = account.reserve(0)?;
for (column, argument) in &pattern.arguments {
let value = evaluate_scoped(
argument,
scope.catalog,
scope.binding,
scope.locals,
Some(account),
scope.work,
Some(graph),
)?;
if let Some(import) = identity_import(pattern, &pattern.program, column)
&& let Ok(Some(identity)) = import_identity(value.as_ref(), import)
{
identities.push(ImportIdentity {
column: column.as_str(),
identity,
});
continue;
}
let value = value.into_owned_accounted(&mut _arguments)?;
row.values.insert(column.clone(), value);
}
let mut _seeds = account.reserve(0)?;
let runs = match &pattern.seeds {
None => vec![(&pattern.program, Cow::Borrowed(&row), identities)],
Some(seeds) => {
let by_identity = identity_import(pattern, &seeds.program, &seeds.column).is_some();
let mut runs = Vec::new();
let mut seen = BTreeSet::new();
for expression in &seeds.expressions {
if by_identity {
let Some(id) = seed_identity(expression, &scope, graph, account)? else {
continue;
};
if !seen.insert(id) {
continue;
}
let mut seeded = identities.clone();
seeded.push(ImportIdentity {
column: seeds.column.as_str(),
identity: ResolvedImport {
id: id.0,
source: None,
},
});
runs.push((&seeds.program, Cow::Borrowed(&row), seeded));
continue;
}
let computed = evaluate_scoped(
expression,
scope.catalog,
scope.binding,
scope.locals,
Some(account),
scope.work,
Some(graph),
)?;
let payload = crate::binding::value_memory_bytes(computed.as_ref())
.saturating_sub(std::mem::size_of::<Value>());
let value = computed.into_owned();
_seeds.grow(payload)?;
let Some(id) = seed_node_id(graph.store, &value)? else {
continue;
};
if !seen.insert(id) {
continue;
}
let mut seeded = row.clone();
seeded.values.insert(seeds.column.clone(), value);
runs.push((&seeds.program, Cow::Owned(seeded), identities.clone()));
}
runs
}
};
with_column_frames(
&pattern.columns,
scope.binding,
scope.locals,
&mut |locals| {
let mut found = false;
let mut output = pattern
.projection
.as_ref()
.map(|_| list::ListBuilder::new(Some(account), scope.work, 0))
.transpose()?;
for (program, row, identities) in &runs {
if found {
break;
}
visit_correlated_matches(
program,
graph.store,
graph.context,
account,
CorrelatedInput { row, identities },
&|| scope.work.charge(1),
&mut |matched| {
let evaluate = |expression| {
evaluate_scoped(
expression,
scope.catalog,
matched,
locals,
Some(account),
scope.work,
Some(graph),
)
};
if let Some(predicate) = &pattern.predicate {
match evaluate(predicate)?.as_ref() {
Value::Bool(true) => {}
Value::Bool(false) | Value::Null => {
return Ok(ScanControl::Continue)
}
value => {
return Err(HawDBError::Execution(format!(
"pattern WHERE requires a boolean value, got {:?}",
description(value, Some(account))
)))
}
}
}
match (&pattern.projection, output.as_mut()) {
(Some(projection), Some(output)) => {
output.push_projected(evaluate(projection)?)?;
Ok(ScanControl::Continue)
}
_ => {
found = true;
Ok(ScanControl::Stop)
}
}
},
)?;
}
Ok(match output {
Some(output) => output.finish(),
None => ProjectedValue::scalar(Value::Bool(found)),
})
},
)
}
fn identity_import<'p>(
pattern: &PatternProjection,
program: &'p GraphMatchProgram,
column: &str,
) -> Option<&'p GraphBindingImport> {
if !pattern.identity_columns.iter().any(|name| name == column) {
return None;
}
program
.imports
.iter()
.find(|import| import.column == column)
}
fn seed_identity(
expression: &ProjectionExpression,
scope: &Scope<'_, '_>,
graph: &ExpressionGraph<'_>,
account: &QueryMemoryAccount,
) -> Result<Option<NodeId>> {
if let Some(node) = slot_node(expression, scope.binding) {
return Ok(Some(node.id));
}
let value = evaluate_scoped(
expression,
scope.catalog,
scope.binding,
scope.locals,
Some(account),
scope.work,
Some(graph),
)?;
seed_node_id(graph.store, value.as_ref())
}
fn slot_node<'b>(
expression: &ProjectionExpression,
binding: &'b Binding,
) -> Option<&'b NodeRecord> {
match expression {
ProjectionExpression::Variable { variable } => binding.nodes.get(variable),
ProjectionExpression::RelationshipEndpoint {
endpoint,
variable,
endpoints: Some(endpoints),
} => {
let relationship = binding.relationships.get(variable)?;
let id = match endpoint {
hawdb_core::RelationshipEndpoint::Start => relationship.source,
hawdb_core::RelationshipEndpoint::End => relationship.target,
};
endpoints
.iter()
.filter_map(|name| binding.nodes.get(name))
.find(|node| node.id == id)
}
_ => None,
}
}
fn seed_node_id(store: &dyn GraphExecutionRead, value: &Value) -> Result<Option<NodeId>> {
let Value::Map(entries) = value else {
return Ok(None);
};
let (Some(Value::Int(id)), true) = (entries.get("_id"), entries.contains_key("labels")) else {
return Ok(None);
};
let Ok(id) = u64::try_from(*id) else {
return Ok(None);
};
let id = NodeId(id);
Ok(store.node_owned(id)?.map(|_| id))
}
fn graph_variable_kind(program: &GraphMatchProgram, name: &str) -> Option<GraphEntityKind> {
program.steps.iter().find_map(|step| match step {
GraphMatchStep::Node(node) if node.variable == name => Some(GraphEntityKind::Node),
GraphMatchStep::Expand {
source,
relationship,
target,
..
} => {
if source == name || target.variable == name {
Some(GraphEntityKind::Node)
} else if relationship.as_deref() == Some(name) {
Some(GraphEntityKind::Relationship)
} else {
None
}
}
_ => None,
})
}
fn kind_matches(binding: &Binding, name: &str, kind: GraphEntityKind) -> Result<()> {
let other = match kind {
GraphEntityKind::Node => binding.relationships.contains_key(name),
GraphEntityKind::Relationship => binding.nodes.contains_key(name),
};
if other {
let expected = match kind {
GraphEntityKind::Node => "a node",
GraphEntityKind::Relationship => "a relationship",
};
return Err(HawDBError::Execution(format!(
"pattern variable '{name}' must be bound to {expected}"
)));
}
Ok(())
}
fn with_column_frames<'p, R>(
columns: &'p [String],
binding: &'p Binding,
parent: Option<&'p ListLocals<'p>>,
visit: &mut dyn FnMut(Option<&ListLocals<'_>>) -> Result<R>,
) -> Result<R> {
let Some((column, rest)) = columns.split_first() else {
return visit(parent);
};
match binding.values.get(column) {
Some(value) => {
let frame = ListLocals {
name: column,
value,
parent,
};
with_column_frames(rest, binding, Some(&frame), visit)
}
None => with_column_frames(rest, binding, parent, visit),
}
}
pub(super) fn relationship_value_endpoint<'a>(
endpoint: hawdb_core::RelationshipEndpoint,
relationship: &Value,
visibility: Option<&Predicate>,
catalog: &Catalog,
account: Option<&QueryMemoryAccount>,
graph: Option<&ExpressionGraph<'_>>,
) -> Result<ProjectedValue<'a>> {
let name = endpoint.function_name();
let fields = match relationship {
Value::Null => return Ok(ProjectedValue::scalar(Value::Null)),
Value::Map(fields) => fields,
value => {
return Err(HawDBError::Execution(format!(
"{name}() requires a relationship, got {:?}",
description(value, account)
)))
}
};
let field = |key: &str| match fields.get(key) {
Some(Value::Int(id)) if *id >= 0 => Some(*id as u64),
_ => None,
};
let not_a_relationship = || {
HawDBError::Execution(format!(
"{name}() requires a relationship, got a map that names no stored relationship"
))
};
let (Some(id), Some(source), Some(target)) =
(field("_id"), field("source_id"), field("target_id"))
else {
return Err(not_a_relationship());
};
let what = format!("{name}() of a relationship value");
let graph = graph_access(&what, graph)?;
let charged = charged(account, graph);
let account = charged.account();
let Some(stored) = stored_relationship(graph, account, RelId(id), NodeId(source))?
.filter(|stored| stored.target == NodeId(target))
else {
if visibility.is_some() {
return Ok(ProjectedValue::scalar(Value::Null));
}
return Err(not_a_relationship());
};
let node = match endpoint {
hawdb_core::RelationshipEndpoint::Start => stored.source,
hawdb_core::RelationshipEndpoint::End => stored.target,
};
read_node(name, node, visibility, catalog, account, graph)
}
fn stored_relationship(
graph: &ExpressionGraph<'_>,
account: &QueryMemoryAccount,
id: RelId,
source: NodeId,
) -> Result<Option<RelRecord>> {
let task_context = graph.context.task_context;
let read = graph.store.projected_relationship_with_allocation(
id,
&BTreeSet::new(),
&mut hawdb_storage::read_view::ControlledGraphReadAllocator::new(
&mut |bytes| crate::store::admit_graph_read(account, task_context, bytes).map(Some),
task_context,
),
)?;
match read {
AdmittedRelationshipRead::Relationship(relationship) => {
let (relationship, _allocation) = relationship.into_parts();
Ok((relationship.source == source).then_some(relationship))
}
AdmittedRelationshipRead::Missing => Ok(None),
AdmittedRelationshipRead::Stopped => Err(HawDBError::Execution(
"relationship point read stopped before admission".into(),
)),
}
}
pub(super) fn stored_node<'a>(
name: &str,
id: NodeId,
catalog: &Catalog,
account: Option<&QueryMemoryAccount>,
graph: Option<&ExpressionGraph<'_>>,
) -> Result<ProjectedValue<'a>> {
let what = format!("{name}() of a relationship without its pattern endpoints");
let graph = graph_access(&what, graph)?;
let charged = charged(account, graph);
read_node(name, id, None, catalog, charged.account(), graph)
}
fn read_node<'a>(
name: &str,
id: NodeId,
visibility: Option<&Predicate>,
catalog: &Catalog,
account: &QueryMemoryAccount,
graph: &ExpressionGraph<'_>,
) -> Result<ProjectedValue<'a>> {
let task_context = graph.context.task_context;
let mut admit =
|bytes: usize| crate::store::admit_graph_read(account, task_context, bytes).map(Some);
let read = graph.store.node_with_allocation(
id,
None,
&mut hawdb_storage::read_view::ControlledGraphReadAllocator::new(&mut admit, task_context),
)?;
let AdmittedNodeRead::Node(node) = read else {
return Err(HawDBError::Execution(format!(
"{name}() endpoint node {} disappeared during the query",
id.0
)));
};
let (node, _read) = node.into_parts();
if let Some(visibility) = visibility {
let mut binding = Binding::values(BTreeMap::new());
binding.nodes.insert(
hawdb_plan_cypher::ENDPOINT_VISIBILITY_VARIABLE.to_string(),
node.clone(),
);
let visible = crate::expression::evaluate_predicate_with_context(
visibility,
catalog,
graph.store,
&binding,
graph.context.observer,
crate::store::AdjacencyReadMemory {
budget_bytes: graph.context.query_memory_bytes.get(),
account: Some(account),
},
task_context,
)?;
if !visible {
return Ok(ProjectedValue::scalar(Value::Null));
}
}
allocated(node_value_bytes(&node, catalog), Some(account), || {
node_value(&node, catalog)
})
}