Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions crates/lance-graph/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,15 @@ matches and visitors to traverse the fold input and body while respecting iterat
scope. Prefer parsing Cypher text to construct these expressions; the parser resolves local variable
references so they cannot accidentally bind to graph variables with the same name.

## Rust AST migration for path values

Named ordinary paths add `PathPattern.variable`; set it to `None` in existing unnamed-path struct
literals, or `Some(name)` for `p = (a)-[:R]->(b)`. Shortest-path names still use
`ShortestPathSpec.variable`. Exhaustive matches and visitors must handle `LogicalOperator::PathMatch`
by traversing its input and path pattern. It retains ordered node and relationship records, so a path
value must not be replaced by its length. Public `nodes(path)` still returns integer keys; the Cypher
guide describes the separate Arrow record returned by a path value.

## Quick Start

```rust
Expand Down
3 changes: 3 additions & 0 deletions crates/lance-graph/src/ast.rs
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,9 @@ pub struct NodePattern {
/// A path pattern connecting nodes through relationships
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct PathPattern {
/// Named ordinary path (`p = (a)-[:R]->(b)`).
#[serde(default)]
pub variable: Option<String>,
/// Starting node
pub start_node: NodePattern,
/// Relationships and intermediate nodes
Expand Down
39 changes: 39 additions & 0 deletions crates/lance-graph/src/csr_expand.rs
Original file line number Diff line number Diff line change
Expand Up @@ -247,6 +247,23 @@ impl Adjacency {
})
}

/// Inspect one occurrence so cancelled work also stops through long tombstone runs.
fn next_candidate(&self, key: i64, offset: &mut usize) -> Option<Option<u64>> {
let base = self.base_neighbors(key);
let position = *offset;
let neighbor = if position < base.len() {
base[position]
} else {
*self.added.get(&key)?.get(position - base.len())?
};
*offset += 1;
let removed = self
.removed
.get(&key)
.is_some_and(|r| r.binary_search(&position).is_ok());
Some((!removed).then_some(neighbor))
}

/// Live edges: base + appended - removed.
pub fn num_edges(&self) -> u64 {
self.base.csr.num_edges() + self.added_count - self.removed_count
Expand Down Expand Up @@ -827,6 +844,21 @@ impl Neighbors {
)
}

/// Inspect one raw edge, including tombstones and suppressed reverse self-loops.
pub(crate) fn next_candidate(
&self,
key: i64,
position: &mut NeighborPosition,
) -> Option<Option<u64>> {
if let Some(candidate) = self.forward.next_candidate(key, &mut position.forward) {
return Some(candidate);
}
self.reverse
.as_ref()?
.next_candidate(key, &mut position.reverse)
.map(|candidate| candidate.filter(|&n| n as i64 != key))
}

/// Edges visited over all keys (self-loops count once per direction in this estimate).
pub fn num_edges(&self) -> u64 {
self.forward.num_edges() + self.reverse.as_ref().map_or(0, |r| r.num_edges())
Expand Down Expand Up @@ -1569,6 +1601,13 @@ impl ExtensionPlanner for CsrExtensionPlanner {
physical_inputs: &[Arc<dyn ExecutionPlan>],
session_state: &SessionState,
) -> datafusion::common::Result<Option<Arc<dyn ExecutionPlan>>> {
if let Some(node) = node
.as_any()
.downcast_ref::<crate::path_match::PathMatchNode>()
{
return crate::path_match::plan_path_match(node, physical_inputs, session_state)
.map(Some);
}
if let Some(node) = node
.as_any()
.downcast_ref::<crate::shortest_path::ShortestPathNode>()
Expand Down
14 changes: 14 additions & 0 deletions crates/lance-graph/src/datafusion_planner/analysis.rs
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ pub struct RelationshipInstance {
/// Planning context that tracks state during plan building
pub struct PlanningContext<'a> {
pub analysis: &'a QueryAnalysis,
pub(crate) path_sources: crate::path_inventory::SourceBatches,
pub(crate) relationship_instance_idx: HashMap<String, usize>,
/// Element rows of the list comprehensions being built, innermost last.
pub(crate) comprehension_elements: Vec<datafusion::logical_expr::LogicalPlan>,
Expand All @@ -46,6 +47,7 @@ impl<'a> PlanningContext<'a> {
pub fn new(analysis: &'a QueryAnalysis) -> Self {
Self {
analysis,
path_sources: Default::default(),
relationship_instance_idx: HashMap::new(),
comprehension_elements: Vec::new(),
}
Expand Down Expand Up @@ -220,6 +222,18 @@ fn analyze_operator(
analysis.required_datasets.insert(label.clone());
}
}
LogicalOperator::PathMatch { input, path } => {
analyze_operator(input, analysis, rel_counter)?;
for node in
std::iter::once(&path.start_node).chain(path.segments.iter().map(|s| &s.end_node))
{
if let (Some(variable), Some(label)) = (&node.variable, node.labels.first()) {
analysis
.var_to_label
.insert(variable.clone(), label.clone());
}
}
}
LogicalOperator::ShortestPath { input, .. } => {
analyze_operator(input, analysis, rel_counter)?;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ impl DataFusionPlanner {
let mut analysis = ctx.analysis.clone();
Self::collect_labels(input, &mut analysis.var_to_label);
let mut branch = PlanningContext::new(&analysis);
branch.path_sources = ctx.path_sources.clone();
branch.relationship_instance_idx = ctx.relationship_instance_idx.clone();
plans.push(self.build_operator(&mut branch, input)?);
// Branches consume distinct relationship instances, in analysis order.
Expand Down
19 changes: 19 additions & 0 deletions crates/lance-graph/src/datafusion_planner/builder/helpers.rs
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,15 @@ impl DataFusionPlanner {
LogicalOperator::OptionalMatch { optional, .. } => {
Self::collect_variables(optional, vars);
}
LogicalOperator::PathMatch { input, path } => {
Self::collect_variables(input, vars);
vars.extend(
std::iter::once(&path.start_node)
.chain(path.segments.iter().map(|s| &s.end_node))
.filter_map(|n| n.variable.clone()),
);
vars.extend(path.variable.clone());
}
LogicalOperator::ShortestPath {
input,
path_variable,
Expand Down Expand Up @@ -171,6 +180,16 @@ impl DataFusionPlanner {
LogicalOperator::OptionalMatch { optional, .. } => {
Self::collect_labels(optional, labels)
}
LogicalOperator::PathMatch { input, path } => {
Self::collect_labels(input, labels);
for node in std::iter::once(&path.start_node)
.chain(path.segments.iter().map(|s| &s.end_node))
{
if let (Some(variable), Some(label)) = (&node.variable, node.labels.first()) {
labels.insert(variable.clone(), label.clone());
}
}
}
LogicalOperator::ShortestPath { input, .. } => Self::collect_labels(input, labels),
LogicalOperator::Union { .. } => {}
LogicalOperator::ListComprehension { input, .. } => Self::collect_labels(input, labels),
Expand Down
2 changes: 2 additions & 0 deletions crates/lance-graph/src/datafusion_planner/builder/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ mod comprehension_ops;
mod expand_ops;
mod helpers;
mod join_builder;
mod path_ops;

use super::DataFusionPlanner;
use crate::error::Result;
Expand Down Expand Up @@ -137,6 +138,7 @@ impl DataFusionPlanner {
optional,
keys,
} => self.build_optional_match(ctx, input, optional, keys),
LogicalOperator::PathMatch { input, path } => self.build_path_match(ctx, input, path),
LogicalOperator::ShortestPath {
input,
source_variable,
Expand Down
214 changes: 214 additions & 0 deletions crates/lance-graph/src/datafusion_planner/builder/path_ops.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,214 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The lance-csr Authors

use crate::ast::PathPattern;
use crate::case_insensitive::qualify_column;
use crate::datafusion_planner::{analysis::PlanningContext, DataFusionPlanner};
use crate::error::{GraphError, Result};
use crate::logical_plan::LogicalOperator;
use crate::path_inventory::{EdgeSource, NodeSource};
use crate::path_match::{Binding, PathMatchNode, PathSpec, Step};
use datafusion::logical_expr::{Extension, LogicalPlan, LogicalPlanBuilder};
use std::sync::Arc;

impl DataFusionPlanner {
pub(crate) fn build_path_match(
&self,
ctx: &mut PlanningContext,
input: &LogicalOperator,
path: &PathPattern,
) -> Result<LogicalPlan> {
let error = |message: String| GraphError::UnsupportedFeature {
feature: message,
location: snafu::Location::new(file!(), line!(), column!()),
};
let catalog = self.get_catalog()?;
let input = self.build_operator(ctx, input)?;
let mut node_labels = Vec::<String>::new();
let mut edge_labels = Vec::<String>::new();
for segment in &path.segments {
if segment.relationship.types.is_empty() {
return Err(error("named paths need explicit relationship types".into()));
}
for kind in &segment.relationship.types {
if edge_labels.iter().any(|t| t.eq_ignore_ascii_case(kind)) {
continue;
}
let mapping = self.get_relationship_mapping(kind)?;
for label in [&mapping.source_label, &mapping.target_label] {
let label = label.as_ref().ok_or_else(|| {
error("named paths need configured endpoint labels".into())
})?;
if !node_labels.iter().any(|l| l.eq_ignore_ascii_case(label)) {
node_labels.push(label.clone());
}
}
edge_labels.push(mapping.relationship_type.clone());
}
}
let mut plans = vec![input];
let mut nodes = Vec::new();
for label in &node_labels {
let mapping = self
.config
.get_node_mapping(label)
.ok_or_else(|| error(format!("missing node mapping {label}")))?;
let source = catalog
.node_source(label)
.ok_or_else(|| error(format!("missing node table {label}")))?;
let schema = source.schema();
let field = schema
.fields()
.iter()
.find(|f| f.name().eq_ignore_ascii_case(&mapping.id_field))
.ok_or_else(|| error(format!("missing node ID {}", mapping.id_field)))?;
if !crate::csr_expand::is_supported_id_type(field.data_type()) {
return Err(error(format!(
"path node ID '{}' has unsupported type {}",
field.name(),
field.data_type()
)));
}
nodes.push(NodeSource {
label: mapping.label.clone(),
id: mapping.id_field.clone(),
schema,
});
plans.push(
LogicalPlanBuilder::scan(label.to_lowercase(), source, None)
.and_then(|b| b.build())
.map_err(|e| self.plan_error("Failed to scan path nodes", e))?,
);
}
let mut edges = Vec::new();
for label in &edge_labels {
let mapping = self.get_relationship_mapping(label)?;
let source = catalog
.relationship_source(label)
.ok_or_else(|| error(format!("missing relationship table {label}")))?;
let schema = source.schema();
for id in [&mapping.source_id_field, &mapping.target_id_field] {
let field = schema
.fields()
.iter()
.find(|f| f.name().eq_ignore_ascii_case(id))
.ok_or_else(|| error(format!("missing relationship ID {id}")))?;
if !crate::csr_expand::is_supported_id_type(field.data_type()) {
return Err(error(format!(
"path relationship ID '{id}' has unsupported type {}",
field.data_type()
)));
}
}
edges.push(EdgeSource {
label: mapping.relationship_type.clone(),
source_label: node_labels
.iter()
.position(|l| Some(l) == mapping.source_label.as_ref())
.unwrap(),
target_label: node_labels
.iter()
.position(|l| Some(l) == mapping.target_label.as_ref())
.unwrap(),
source_id: mapping.source_id_field.clone(),
target_id: mapping.target_id_field.clone(),
schema,
});
plans.push(
LogicalPlanBuilder::scan(label.to_lowercase(), source, None)
.and_then(|b| b.build())
.map_err(|e| self.plan_error("Failed to scan path edges", e))?,
);
}
let mut bindings = Vec::new();
for node in
std::iter::once(&path.start_node).chain(path.segments.iter().map(|s| &s.end_node))
{
let variable = node
.variable
.clone()
.ok_or_else(|| error("named path endpoint was not assigned a variable".into()))?;
let label = ctx
.analysis
.var_to_label
.get(&variable)
.or_else(|| node.labels.first())
.ok_or_else(|| {
error(format!("path variable '{variable}' needs a resolved label"))
})?;
let label = node_labels
.iter()
.position(|l| l.eq_ignore_ascii_case(label))
.ok_or_else(|| {
error(format!(
"path label '{label}' is incompatible with its relationship types"
))
})?;
bindings.push(Binding {
id_column: qualify_column(&variable, &nodes[label].id),
variable,
label,
properties: node.properties.clone(),
});
}
let shortest = path.shortest.as_ref().map(|s| s.all);
let mut steps = Vec::new();
for segment in &path.segments {
let rel = &segment.relationship;
let min = rel.length.as_ref().and_then(|l| l.min).unwrap_or(1);
let max = rel
.length
.as_ref()
.map(|l| {
l.max.unwrap_or(if shortest.is_some() {
u32::MAX
} else {
crate::MAX_VARIABLE_LENGTH_HOPS
})
})
.unwrap_or(1);
if min > max || (shortest.is_none() && max > crate::MAX_VARIABLE_LENGTH_HOPS) {
return Err(error(format!("invalid named path range {min}..{max}; ordinary variable-length paths allow at most {} hops", crate::MAX_VARIABLE_LENGTH_HOPS)));
}
steps.push(Step {
types: rel
.types
.iter()
.map(|t| {
edge_labels
.iter()
.position(|l| l.eq_ignore_ascii_case(t))
.unwrap()
})
.collect(),
direction: rel.direction.clone(),
min,
max,
relationship: rel.variable.clone(),
properties: rel.properties.clone(),
});
}
let variable = path
.variable
.as_ref()
.or_else(|| path.shortest.as_ref().and_then(|s| s.variable.as_ref()))
.cloned()
.ok_or_else(|| error("rich path needs a variable".into()))?;
let node = PathMatchNode::try_new(
plans,
PathSpec {
variable,
sources: ctx.path_sources.clone(),
nodes,
edges,
bindings,
steps,
shortest,
},
)
.map_err(|e| self.plan_error("Failed to plan path value", e))?;
Ok(LogicalPlan::Extension(Extension {
node: Arc::new(node),
}))
}
}
Loading
Loading