use super::*;
use crate::graph::parallel::{self, ParallelInterrupt};
const CEILING_PUBLISH_STRIDE: usize = 256;
fn repeats_a_node_variable(pattern: &Pattern) -> bool {
let mut seen: Vec<&str> = Vec::new();
for element in &pattern.elements {
let PatternElement::Node(node) = element else {
continue;
};
let Some(var) = node.variable.as_deref() else {
continue;
};
if seen.contains(&var) {
return true;
}
seen.push(var);
}
false
}
fn dedup_interchangeable_partials(
matches: Vec<PatternMatch>,
indices: Vec<NodeIndex>,
collapsible: bool,
) -> (Vec<PatternMatch>, Vec<NodeIndex>) {
if !collapsible {
return (matches, indices);
}
let mut seen = HashSet::with_capacity(indices.len());
let mut kept_matches = Vec::with_capacity(indices.len());
let mut kept_indices = Vec::with_capacity(indices.len());
for (partial, index) in matches.into_iter().zip(indices) {
if seen.insert(index) {
kept_matches.push(partial);
kept_indices.push(index);
}
}
(kept_matches, kept_indices)
}
struct HopPlan<'p> {
edge: &'p EdgePattern,
node: &'p NodePattern,
anonymous_path_var: Option<String>,
track_fixed_trail: bool,
is_last_hop: bool,
limit: Option<usize>,
var_length_cap_safe: bool,
}
#[derive(Clone, Copy, PartialEq, Eq)]
enum CapPass {
Capped,
Uncapped,
}
impl<'a> PatternExecutor<'a> {
fn seed_start_nodes(
&self,
pattern: &Pattern,
first_node: &NodePattern,
has_edges: bool,
source_cap: Option<usize>,
) -> Result<Vec<NodeIndex>, String> {
let first_is_prebound = first_node
.variable
.as_ref()
.map(|v| self.pre_bindings.get(v).is_some())
.unwrap_or(false);
let mut initial_nodes = if !first_is_prebound
&& has_edges
&& first_node.node_type.is_none()
&& first_node.properties.is_none()
{
let edge_conn_types: Option<Vec<InternedKey>> =
if let Some(PatternElement::Edge(ep)) = pattern.elements.get(1) {
if ep.var_length.is_none() {
match ep.conn_filter() {
ConnTypeFilter::Any => None,
ConnTypeFilter::One(key) => Some(vec![key]),
ConnTypeFilter::AnyOf(keys) => Some(keys),
}
} else {
None
}
} else {
None
};
let is_outgoing = if let Some(PatternElement::Edge(ep)) = pattern.elements.get(1) {
ep.direction == EdgeDirection::Outgoing
} else {
false
};
if let (Some(conn_types), true) = (edge_conn_types, is_outgoing) {
let mut union: Vec<NodeIndex> = Vec::new();
let mut complete = true;
for ct in conn_types {
match self
.graph
.graph
.sources_for_conn_type_bounded(ct, source_cap)
{
Some(sources) => {
if source_cap.is_some_and(|cap| sources.len() >= cap) {
self.note_cap_truncated();
}
union.extend(sources.into_iter().map(|s| NodeIndex::new(s as usize)));
}
None => {
complete = false;
break;
}
}
}
if complete {
union.sort_unstable();
union.dedup();
union
} else {
self.find_matching_nodes(first_node)?
}
} else {
self.find_matching_nodes(first_node)?
}
} else {
self.find_matching_nodes(first_node)?
};
if let Some(cap) = source_cap {
if initial_nodes.len() > cap {
if has_edges {
self.note_cap_truncated();
}
initial_nodes.truncate(cap);
}
}
Ok(initial_nodes)
}
pub fn execute(&self, pattern: &Pattern) -> Result<Vec<PatternMatch>, String> {
self.take_cap_truncated();
let matches = self.execute_pass(pattern, CapPass::Capped)?;
if self.max_matches.is_some_and(|max| matches.len() < max) && self.take_cap_truncated() {
return self.execute_pass(pattern, CapPass::Uncapped);
}
Ok(matches)
}
fn plan_hop<'p>(
&self,
edge_pattern: &'p EdgePattern,
node_pattern: &'p NodePattern,
element_index: usize,
is_last_hop: bool,
earlier_relationship_state: bool,
pass: CapPass,
) -> HopPlan<'p> {
HopPlan {
edge: edge_pattern,
node: node_pattern,
anonymous_path_var: (edge_pattern.variable.is_none()
&& edge_pattern.needs_path_info
&& edge_pattern.var_length.is_some())
.then(|| format!("__anon_vlpath_{element_index}")),
track_fixed_trail: edge_pattern.var_length.is_none() && edge_pattern.needs_path_info,
is_last_hop,
limit: if is_last_hop {
self.max_matches
} else {
match pass {
CapPass::Capped => self.max_matches.map(|m| m.saturating_mul(50).max(1000)),
CapPass::Uncapped => None,
}
},
var_length_cap_safe: edge_pattern.var_length.is_some()
&& !earlier_relationship_state
&& !self
.distinct_target_var
.as_deref()
.is_some_and(|dtv| node_pattern.variable.as_deref() == Some(dtv)),
}
}
fn execute_pass(&self, pattern: &Pattern, pass: CapPass) -> Result<Vec<PatternMatch>, String> {
if pattern.elements.is_empty() {
return Ok(Vec::new());
}
let first_node = match &pattern.elements[0] {
PatternElement::Node(np) => np,
_ => {
return Err(
"Pattern must start with a node in parentheses. Example: (n:Person) or ()"
.to_string(),
)
}
};
let has_edges = pattern.elements.len() > 1;
let source_cap = if has_edges {
match pass {
CapPass::Capped => self.max_matches.map(|m| m.saturating_mul(100).max(1000)),
CapPass::Uncapped => None,
}
} else {
self.max_matches
};
let initial_nodes = self.seed_start_nodes(pattern, first_node, has_edges, source_cap)?;
let mut seeds_pending = has_edges && self.max_matches.is_some();
let mut matches: Vec<PatternMatch> = if seeds_pending {
Vec::new()
} else {
initial_nodes
.iter()
.map(|&idx| self.seed_match(first_node, idx))
.collect()
};
let mut current_indices: Vec<NodeIndex> = initial_nodes;
let mut visited = VisitedStamps::default();
let mut distinct_seen: HashSet<NodeIndex> = if self.distinct_target_var.is_some() {
HashSet::with_capacity(current_indices.len())
} else {
HashSet::new()
};
let repeats_a_node_variable = repeats_a_node_variable(pattern);
let mut relationship_state_recorded = false;
let mut i = 1;
while i < pattern.elements.len() {
let is_last_hop = i + 2 >= pattern.elements.len();
if let Some(msg) = self.interrupt_reason() {
return Err(msg);
}
let edge_pattern = match &pattern.elements[i] {
PatternElement::Edge(ep) => ep,
_ => return Err("Expected edge pattern after node. Use -[:TYPE]-> for outgoing, <-[:TYPE]- for incoming.".to_string()),
};
i += 1;
if i >= pattern.elements.len() {
return Err("Edge pattern must be followed by a node pattern. Example: ()-[:KNOWS]->(n:Person)".to_string());
}
let node_pattern = match &pattern.elements[i] {
PatternElement::Node(np) => np,
_ => return Err("Expected node pattern after edge. Complete the pattern with a node: ()-[:EDGE]->(node)".to_string()),
};
let earlier_relationship_state = relationship_state_recorded;
let hop = self.plan_hop(
edge_pattern,
node_pattern,
i,
is_last_hop,
earlier_relationship_state,
pass,
);
relationship_state_recorded |= hop.track_fixed_trail
|| hop.edge.variable.is_some()
|| hop.anonymous_path_var.is_some();
let (mut new_matches, mut new_indices) = if !seeds_pending
&& matches.len() >= EXPANSION_RAYON_THRESHOLD
&& self.max_matches.is_none()
{
self.expand_hop_parallel(&matches, ¤t_indices, &hop)?
} else {
self.expand_hop_sequential(
&matches,
¤t_indices,
&hop,
seeds_pending.then_some(first_node),
&mut distinct_seen,
&mut visited,
)?
};
if let Some(msg) = self.interrupt_reason() {
return Err(msg);
}
if let Some(max) = hop.limit {
if new_matches.len() > max {
if !is_last_hop {
self.note_cap_truncated();
}
new_matches.truncate(max);
new_indices.truncate(max);
}
}
let collapsible = self.distinct_target_var.is_some()
&& !relationship_state_recorded
&& !repeats_a_node_variable
&& i + 1 < pattern.elements.len()
&& node_pattern.variable.is_none();
(matches, current_indices) =
dedup_interchangeable_partials(new_matches, new_indices, collapsible);
seeds_pending = false;
i += 1;
}
Ok(matches)
}
fn expand_hop_parallel(
&self,
matches: &[PatternMatch],
current_indices: &[NodeIndex],
hop: &HopPlan<'_>,
) -> Result<(Vec<PatternMatch>, Vec<NodeIndex>), String> {
let interrupt = ParallelInterrupt::new(|| self.interrupt_reason());
let produced = std::sync::atomic::AtomicUsize::new(0);
let results: Vec<(PatternMatch, NodeIndex)> = parallel::install(|| {
matches
.par_iter()
.zip(current_indices.par_iter())
.map_init(
|| (VisitedStamps::default(), 0usize),
|(visited, unpublished), (current_match, &source_idx)| {
if interrupt.check_each().is_err() {
return Vec::new();
}
let Some(expansions) = interrupt.capture(self.expand_from_node(
source_idx,
hop.edge,
hop.node,
None,
self.bound_target(hop.node, current_match),
visited,
)) else {
return Vec::new();
};
let kept: Vec<_> = expansions
.into_iter()
.filter_map(|(target_idx, edge_binding)| {
if reuses_bound_relationship(current_match, &edge_binding) {
return None;
}
if !self.target_satisfies_bindings(
hop.node,
current_match,
target_idx,
) {
return None;
}
Some((
self.extend_match(current_match, hop, edge_binding, target_idx),
target_idx,
))
})
.collect();
*unpublished += kept.len();
if *unpublished >= CEILING_PUBLISH_STRIDE {
let held = produced
.fetch_add(*unpublished, std::sync::atomic::Ordering::Relaxed)
+ *unpublished;
*unpublished = 0;
if interrupt.capture(self.check_match_ceiling(held)).is_none() {
return Vec::new();
}
}
kept
},
)
.flatten()
.collect()
});
interrupt.finish()?;
self.check_match_ceiling(results.len())?;
let needs_dedup = hop.is_last_hop
&& self
.distinct_target_var
.as_ref()
.is_some_and(|dtv| hop.node.variable.as_deref() == Some(dtv.as_str()));
if needs_dedup {
let mut seen_targets = HashSet::new();
Ok(results
.into_iter()
.filter(|(_, target_idx)| seen_targets.insert(*target_idx))
.unzip())
} else {
Ok(results.into_iter().unzip())
}
}
fn expand_hop_sequential(
&self,
matches: &[PatternMatch],
current_indices: &[NodeIndex],
hop: &HopPlan<'_>,
lazy_seed: Option<&NodePattern>,
distinct_seen: &mut HashSet<NodeIndex>,
visited: &mut VisitedStamps,
) -> Result<(Vec<PatternMatch>, Vec<NodeIndex>), String> {
let mut new_matches = Vec::new();
let mut new_indices = Vec::new();
let mut expand_count: usize = 0;
let hop_limit = hop.limit;
for (position, &source_idx) in current_indices.iter().enumerate() {
if hop_limit.is_some_and(|max| new_matches.len() >= max) {
break;
}
let pending_seed;
let current_match = if let Some(first_node) = lazy_seed {
pending_seed = self.seed_match(first_node, source_idx);
&pending_seed
} else if let Some(m) = matches.get(position) {
m
} else {
break;
};
let mut remaining = hop_limit.map(|max| max.saturating_sub(new_matches.len()));
let hint = self.bound_target(hop.node, current_match);
if hop.edge.var_length.is_some() && !(hop.var_length_cap_safe && hint.is_none()) {
remaining = None;
}
let expansions =
self.expand_from_node(source_idx, hop.edge, hop.node, remaining, hint, visited)?;
for (target_idx, edge_binding) in expansions {
if reuses_bound_relationship(current_match, &edge_binding) {
continue;
}
expand_count += 1;
if expand_count.is_multiple_of(1024) {
if let Some(msg) = self.interrupt_reason() {
return Err(msg);
}
self.check_match_ceiling(new_matches.len())?;
}
if hop_limit.is_some_and(|max| new_matches.len() >= max) {
break;
}
if !self.target_satisfies_bindings(hop.node, current_match, target_idx) {
continue;
}
if hop.is_last_hop {
if let Some(ref dtv) = self.distinct_target_var {
if hop.node.variable.as_deref() == Some(dtv.as_str())
&& (self
.distinct_prior
.is_some_and(|prior| prior.contains(&target_idx))
|| !distinct_seen.insert(target_idx))
{
continue;
}
}
}
new_matches.push(self.extend_match(current_match, hop, edge_binding, target_idx));
new_indices.push(target_idx);
}
}
if !hop.is_last_hop && hop_limit.is_some_and(|max| new_matches.len() >= max) {
self.note_cap_truncated();
}
self.check_match_ceiling(new_matches.len())?;
Ok((new_matches, new_indices))
}
#[inline]
fn target_satisfies_bindings(
&self,
node_pattern: &NodePattern,
current_match: &PatternMatch,
target_idx: NodeIndex,
) -> bool {
let Some(ref var) = node_pattern.variable else {
return true;
};
if let Some(&bound_idx) = self.pre_bindings.get(var) {
if target_idx != bound_idx {
return false;
}
}
let already_bound = current_match.bindings.iter().find_map(|(name, binding)| {
if name == var {
match binding {
MatchBinding::Node { index, .. } | MatchBinding::NodeRef(index) => Some(*index),
_ => None,
}
} else {
None
}
});
already_bound.is_none_or(|bound_idx| target_idx == bound_idx)
}
#[inline]
fn extend_match(
&self,
current_match: &PatternMatch,
hop: &HopPlan<'_>,
edge_binding: MatchBinding,
target_idx: NodeIndex,
) -> PatternMatch {
let mut new_match = current_match.clone();
if hop.track_fixed_trail {
extend_fixed_trail(&mut new_match, &edge_binding);
}
if let Some(ref var) = hop.edge.variable {
new_match.bindings.push((var.clone(), edge_binding));
} else if let Some(ref internal_var) = hop.anonymous_path_var {
new_match
.bindings
.push((internal_var.clone(), edge_binding));
}
if let Some(ref var) = hop.node.variable {
new_match
.bindings
.push((var.clone(), self.node_to_binding(target_idx)));
}
new_match
}
#[inline]
fn seed_match(&self, first_node: &NodePattern, idx: NodeIndex) -> PatternMatch {
let mut pm = PatternMatch {
bindings: Vec::new(),
exact_path: None,
};
if let Some(ref var) = first_node.variable {
pm.bindings.push((var.clone(), self.node_to_binding(idx)));
}
pm
}
}
#[cfg(test)]
mod tests {
use super::repeats_a_node_variable;
use crate::graph::core::pattern_matching::parse_pattern;
#[test]
fn a_node_variable_written_twice_is_reported() {
for text in [
"(a)-[:A]->(b)-[:B]->(a)",
"(a)-[:A]->()-[:B]->(a)",
"(a:N)-[:A]->(a)",
] {
let pattern = parse_pattern(text).unwrap_or_else(|e| panic!("{text}: {e}"));
assert!(repeats_a_node_variable(&pattern), "{text}");
}
}
#[test]
fn distinct_and_anonymous_variables_are_not_repeats() {
for text in [
"(a)-[:A]->(b)-[:B]->(c)",
"(a)-[:A]->()-[:B]->(b)",
"()-[:A]->()-[:B]->()",
] {
let pattern = parse_pattern(text).unwrap_or_else(|e| panic!("{text}: {e}"));
assert!(!repeats_a_node_variable(&pattern), "{text}");
}
}
}