use crate::binding::{binding_memory_bytes, node_memory_bytes, relationship_memory_bytes, Binding};
use crate::expression::evaluate_predicate_with_context;
use crate::pipeline::{
runtime_checkpoint, AccountedBindingBatch, BatchControl, BatchExecutionContext, BindingBatch,
BindingBatchSource,
};
use crate::predicate::{label_ids_for_pattern, node_matches_label_pattern};
use crate::store::{admit_graph_read, AdjacencyReadMemory, GraphExecutionRead, ScanControl};
use crate::traversal::{
visit_bounded_expand_targets, visit_one_hop_relationships_with_context,
visit_zero_hop_expand_target, BoundedExpandSpec, OneHopRelationshipSpec,
};
use crate::{ExecutionLimit, QueryMemoryAccount, QueryMemoryClass};
use hawdb_core::{HawDBError, Result, Value};
use hawdb_plan_cypher::{
GraphEntityKind, GraphMatchNode, GraphMatchProgram, GraphMatchStep, PhysicalPlan,
};
use hawdb_storage::read_view::{
AdmittedNodeRead, AdmittedVec, GraphReadAdmission, GraphReadAllocation,
};
use hawdb_storage::{adjacency::AdjacencyDirection, NodeId, NodeRecord, RelId};
use std::collections::{BTreeMap, BTreeSet};
const MAX_MATCH_STEPS: usize = 32;
pub(crate) fn stream_graph_match(
program: &GraphMatchProgram,
input: Option<&PhysicalPlan>,
source: &mut dyn BindingBatchSource,
store: &dyn GraphExecutionRead,
context: BatchExecutionContext<'_>,
limit: ExecutionLimit,
emit: &mut dyn FnMut(BindingBatch) -> Result<BatchControl>,
) -> Result<BatchControl> {
if program.steps.len() > MAX_MATCH_STEPS {
return Err(HawDBError::Execution(
"MATCH exceeds maximum pattern depth".to_string(),
));
}
for step in &program.steps {
if let GraphMatchStep::Expand {
relationship,
rel_type,
properties,
direction,
min_hops,
max_hops,
..
} = step
&& (*min_hops != 1 || *max_hops != 1)
&& (relationship.is_some()
|| !properties.is_empty()
|| *direction != hawdb_core::RelationshipDirection::Outgoing
|| rel_type.is_empty())
{
return Err(HawDBError::Execution("bounded MATCH expansion requires an outgoing typed pattern without relationship bindings".to_string()));
}
}
if limit.is_reached(0) {
return Ok(BatchControl::Stop);
}
let account = context.operator_account("GraphMatchExec state");
let source_account = context.source_account_retaining("GraphMatchExec source", account.clone());
let output_account = context.memory_ledger.account(
QueryMemoryClass::PipelineBatch,
"GraphMatchExec output",
context.memory.batch_payload_bytes,
);
let mut output = AccountedBindingBatch::with_account(
"GraphMatchExec",
context.memory.batch_rows.get(),
context.memory.batch_payload_bytes,
output_account,
);
let runtime = MatchRuntime {
program,
store,
context,
account: &account,
source_account: &source_account,
};
let mut emitted = 0;
let mut append = |binding: &Binding| {
if output.push_cloned(binding, emit)? == BatchControl::Stop {
return Ok(ScanControl::Stop);
}
emitted += 1;
if output.is_full() && output.emit(emit)? == BatchControl::Stop {
return Ok(ScanControl::Stop);
}
Ok(if limit.is_reached(emitted) {
ScanControl::Stop
} else {
ScanControl::Continue
})
};
let mut visit_input = |input: &Binding| -> Result<BatchControl> {
runtime_checkpoint(context.task_context)?;
let (binding, _leases) = runtime.prepare(input)?;
let mut matched = false;
let control = runtime.visit(0, &binding, &BTreeSet::new(), &mut |row| {
matched = true;
append(row)
})?;
if control == ScanControl::Stop {
return Ok(BatchControl::Stop);
}
if !matched && program.optional {
let _lease = account.reserve(
binding_memory_bytes(&binding).saturating_add(
program
.introduced
.iter()
.map(|name| name.len() + 64)
.sum::<usize>(),
),
)?;
let mut row = binding.clone();
for name in &program.introduced {
row.nodes.remove(name);
row.relationships.remove(name);
row.values.insert(name.clone(), Value::Null);
}
if context.observer.seed_graph_scoring_input().is_some() {
crate::scoring::advance_seed_hop(&mut row, 0, false)?;
}
if append(&row)? == ScanControl::Stop {
return Ok(BatchControl::Stop);
}
}
Ok(BatchControl::Continue)
};
let control = if let Some(input) = input {
source.execute(input, ExecutionLimit::unlimited(), &mut |batch| {
for row in &batch {
if visit_input(row)? == BatchControl::Stop {
return Ok(BatchControl::Stop);
}
}
Ok(BatchControl::Continue)
})?
} else {
visit_input(&Binding::values(BTreeMap::new()))?
};
if output.emit(emit)? == BatchControl::Stop {
return Ok(BatchControl::Stop);
}
Ok(control)
}
struct MatchRuntime<'a> {
program: &'a GraphMatchProgram,
store: &'a dyn GraphExecutionRead,
context: BatchExecutionContext<'a>,
account: &'a QueryMemoryAccount,
source_account: &'a QueryMemoryAccount,
}
impl MatchRuntime<'_> {
fn adjacency_memory(&self) -> AdjacencyReadMemory<'_> {
AdjacencyReadMemory {
budget_bytes: self.context.memory.query_memory_bytes.get(),
account: Some(self.source_account),
}
}
fn prepare(
&self,
input: &Binding,
) -> Result<(Binding, AdmittedVec<Box<dyn GraphReadAllocation>>)> {
let mut admit =
|bytes| admit_graph_read(self.source_account, self.context.task_context, bytes);
let admission = GraphReadAdmission::new(&mut admit);
let mut leases = AdmittedVec::new(&admission)?;
leases.try_push(admit_graph_read(
self.source_account,
self.context.task_context,
binding_memory_bytes(input),
)?)?;
let mut row = input.clone();
for name in &self.program.introduced {
row.nodes.remove(name);
row.relationships.remove(name);
row.values.remove(name);
}
for import in &self.program.imports {
row.nodes.remove(&import.variable);
row.relationships.remove(&import.variable);
let value = input.values.get(&import.column).ok_or_else(|| {
HawDBError::Execution(format!(
"missing graph column '{}' during MATCH",
import.column
))
})?;
if value == &Value::Null {
row.values.insert(import.variable.clone(), Value::Null);
continue;
}
let Value::Map(values) = value else {
return Err(HawDBError::Execution(
"typed graph column requires an entity value".to_string(),
));
};
let id = entity_id(values, "_id")?;
match import.kind {
GraphEntityKind::Node => {
let mut admit_node = |bytes: usize| {
admit_graph_read(
self.source_account,
self.context.task_context,
bytes
.saturating_add(import.variable.len())
.saturating_add(1024),
)
.map(Some)
};
let AdmittedNodeRead::Node(node) = self.store.node_with_allocation(
NodeId(id),
None,
&mut hawdb_storage::read_view::ControlledGraphReadAllocator::new(
&mut admit_node,
self.context.task_context,
),
)?
else {
return Err(HawDBError::Execution(
"bound node disappeared during MATCH".to_string(),
));
};
let (node, allocation) = node.into_parts();
leases.try_push(allocation)?;
row.nodes.insert(import.variable.clone(), node);
}
GraphEntityKind::Relationship => {
let source = NodeId(entity_id(values, "source_id")?);
let mut found = None;
self.store
.visit_ordered_adjacent_relationships_with_allocation(
source,
None,
AdjacencyDirection::Outgoing,
self.adjacency_memory(),
&mut hawdb_storage::read_view::ControlledGraphReadAllocator::new(
&mut |bytes| {
admit_graph_read(
self.source_account,
self.context.task_context,
bytes
.saturating_add(import.variable.len())
.saturating_add(1024),
)
.map(Some)
},
self.context.task_context,
),
&mut |input| {
let (relationship, allocation) = input.into_parts();
runtime_checkpoint(self.context.task_context)?;
if relationship.id == RelId(id) {
found = Some((relationship, allocation));
Ok(ScanControl::Stop)
} else {
Ok(ScanControl::Continue)
}
},
)?;
let (relationship, allocation) = found.ok_or_else(|| {
HawDBError::Execution(
"bound relationship disappeared during MATCH".to_string(),
)
})?;
leases.try_push(allocation)?;
row.relationships
.insert(import.variable.clone(), relationship);
}
}
}
Ok((row, leases))
}
fn visit(
&self,
index: usize,
row: &Binding,
used: &BTreeSet<RelId>,
emit: &mut dyn FnMut(&Binding) -> Result<ScanControl>,
) -> Result<ScanControl> {
runtime_checkpoint(self.context.task_context)?;
let Some(step) = self.program.steps.get(index) else {
if let Some(predicate) = &self.program.predicate
&& !evaluate_predicate_with_context(
predicate,
self.context.catalog,
self.store,
row,
self.context.observer,
self.adjacency_memory(),
self.context.task_context,
)?
{
return Ok(ScanControl::Continue);
}
return emit(row);
};
match step {
GraphMatchStep::Node(pattern) => {
if let Some(node) = row.nodes.get(&pattern.variable) {
return if self.node_matches(pattern, node) {
self.visit(index + 1, row, used, emit)
} else {
Ok(ScanControl::Continue)
};
}
if row.values.get(&pattern.variable) == Some(&Value::Null) {
return Ok(ScanControl::Continue);
}
let labels = label_ids_for_pattern(self.context.catalog, &pattern.label);
let label = labels.as_deref().and_then(|labels| {
if let [label] = labels {
Some(*label)
} else {
None
}
});
self.store.visit_nodes_with_allocation(
label,
&mut hawdb_storage::read_view::ControlledGraphReadAllocator::new(
&mut |bytes| {
crate::store::admit_graph_read(
self.source_account,
self.context.task_context,
bytes
.saturating_add(binding_memory_bytes(row))
.saturating_add(1024 + pattern.variable.len()),
)
.map(Some)
},
self.context.task_context,
),
&mut |input| {
let (node, _allocation) = input.into_parts();
runtime_checkpoint(self.context.task_context)?;
if !self.node_matches(pattern, &node) {
return Ok(ScanControl::Continue);
}
let mut next = row.clone();
next.nodes.insert(pattern.variable.clone(), node);
self.visit(index + 1, &next, used, emit)
},
)
}
GraphMatchStep::Expand {
source,
relationship,
rel_type,
properties,
direction,
min_hops,
max_hops,
target,
} => {
let bounded = *min_hops != 1 || *max_hops != 1;
let Some(source_node) = row.nodes.get(source) else {
if row.values.get(source) == Some(&Value::Null) {
return Ok(ScanControl::Continue);
}
return Err(HawDBError::Execution(format!(
"missing MATCH source '{source}'"
)));
};
let rel_type_id = if rel_type.is_empty() {
None
} else {
let Some(id) = self.context.catalog.rel_type_id(rel_type) else {
if *min_hops == 0 {
let labels = label_ids_for_pattern(self.context.catalog, &target.label);
return visit_zero_hop_expand_target(
self.store,
source_node.id,
labels.as_deref(),
*max_hops,
self.adjacency_memory(),
self.context.task_context,
&mut |node, hop| {
self.visit_target(index, row, used, target, (node, hop), emit)
},
);
}
return Ok(ScanControl::Continue);
};
Some(id)
};
let labels = label_ids_for_pattern(self.context.catalog, &target.label);
if bounded {
return visit_bounded_expand_targets(
self.store,
BoundedExpandSpec {
source: source_node.id,
rel_type_id: rel_type_id.unwrap(),
target_label_ids: labels.as_deref(),
min_hops: *min_hops,
max_hops: *max_hops,
},
self.adjacency_memory(),
self.context.task_context,
&mut |node, hop| {
self.visit_target(index, row, used, target, (node, hop), emit)
},
);
}
let bound_relationship = relationship
.as_ref()
.and_then(|name| row.relationships.get(name))
.map(|record| record.id);
if relationship
.as_ref()
.is_some_and(|name| row.values.get(name) == Some(&Value::Null))
{
return Ok(ScanControl::Continue);
}
visit_one_hop_relationships_with_context(
self.store,
OneHopRelationshipSpec {
source: source_node.id,
rel_type_id,
target_label_ids: labels.as_deref(),
rel_properties: properties,
relationship_scan_filter: None,
direction: *direction,
},
self.adjacency_memory(),
self.context.observer,
self.context.task_context,
&mut |edge, node| {
if used.contains(&edge.id)
|| bound_relationship.is_some_and(|id| id != edge.id)
|| !properties_match(properties, &edge.properties)
{
return Ok(ScanControl::Continue);
}
let _used = self
.account
.reserve(used.len().saturating_add(1).saturating_mul(48))?;
let _lease = self.source_account.reserve(
binding_memory_bytes(row)
.saturating_add(relationship_memory_bytes(&edge)),
)?;
let mut next = row.clone();
let mut used = used.clone();
used.insert(edge.id);
if let Some(variable) = relationship {
next.relationships.insert(variable.clone(), edge);
}
self.visit_target(index, &next, &used, target, (node, 1), emit)
},
)
}
}
}
fn visit_target(
&self,
index: usize,
row: &Binding,
used: &BTreeSet<RelId>,
pattern: &GraphMatchNode,
observed_target: (NodeRecord, usize),
emit: &mut dyn FnMut(&Binding) -> Result<ScanControl>,
) -> Result<ScanControl> {
let (node, hop) = observed_target;
if !self.node_matches(pattern, &node)
|| row
.nodes
.get(&pattern.variable)
.is_some_and(|bound| bound.id != node.id)
|| row.values.get(&pattern.variable) == Some(&Value::Null)
{
return Ok(ScanControl::Continue);
}
let _lease = self.source_account.reserve(
binding_memory_bytes(row)
.saturating_add(node_memory_bytes(&node))
.saturating_add(pattern.variable.len()),
)?;
let mut next = row.clone();
next.nodes.insert(pattern.variable.clone(), node);
if self.context.observer.seed_graph_scoring_input().is_some() {
crate::scoring::advance_seed_hop(&mut next, hop, true)?;
}
self.visit(index + 1, &next, used, emit)
}
fn node_matches(&self, pattern: &GraphMatchNode, node: &NodeRecord) -> bool {
node_matches_label_pattern(
node,
label_ids_for_pattern(self.context.catalog, &pattern.label).as_deref(),
) && properties_match(&pattern.properties, &node.properties)
}
}
fn entity_id(values: &BTreeMap<String, Value>, field: &str) -> Result<u64> {
match values.get(field) {
Some(Value::Int(id)) => Ok(*id as u64),
_ => Err(HawDBError::Execution(format!(
"typed graph value lacks identity field '{field}'"
))),
}
}
fn properties_match(expected: &BTreeMap<String, Value>, actual: &BTreeMap<String, Value>) -> bool {
expected
.iter()
.all(|(name, value)| value != &Value::Null && actual.get(name) == Some(value))
}