#[cfg(feature = "algorithms")]
use std::cmp::Reverse;
#[cfg(any(feature = "algorithms", feature = "embeddings-index"))]
use std::collections::HashMap;
#[cfg(feature = "algorithms")]
use std::collections::HashSet;
use crate::error::{NopalError, Result};
use crate::query::nql::executor::result::{QueryResult, Row};
use crate::query::nql::parser::ast::{Expression, GroupByClause, Projection, Query};
use crate::types::{Node, PropertyValue};
#[cfg(feature = "algorithms")]
use crate::graph::Direction;
use crate::graph::Graph;
#[cfg(feature = "algorithms")]
use crate::types::NodeId;
#[cfg(feature = "algorithms")]
#[derive(Default)]
pub struct AlgoResults {
pub pagerank: Option<HashMap<NodeId, f64>>,
pub betweenness: Option<HashMap<NodeId, f64>>,
pub clustering: Option<HashMap<NodeId, f64>>,
pub degree: Option<HashMap<NodeId, f64>>,
pub community: Option<HashMap<NodeId, usize>>,
pub leiden: Option<HashMap<NodeId, usize>>,
}
#[cfg(not(feature = "algorithms"))]
#[derive(Default)]
pub struct AlgoResults {}
#[cfg(feature = "algorithms")]
pub async fn precompute_for_query(
graph: &Graph,
nodes: &[Node],
query: &Query,
) -> Result<AlgoResults> {
precompute_algorithms(graph, nodes, query).await
}
#[cfg(not(feature = "algorithms"))]
pub async fn precompute_for_query(
_graph: &Graph,
_nodes: &[Node],
_query: &Query,
) -> Result<AlgoResults> {
Ok(AlgoResults::default())
}
#[cfg(feature = "algorithms")]
pub fn lookup_algo_value(
name: &str,
node_id: &NodeId,
cache: &AlgoResults,
) -> PropertyValue {
let lower = name.to_lowercase();
match lower.as_str() {
"pagerank" => cache.pagerank.as_ref()
.and_then(|m| m.get(node_id))
.map(|v| PropertyValue::Float(*v))
.unwrap_or(PropertyValue::Null),
"betweenness" => cache.betweenness.as_ref()
.and_then(|m| m.get(node_id))
.map(|v| PropertyValue::Float(*v))
.unwrap_or(PropertyValue::Null),
"clustering" => cache.clustering.as_ref()
.and_then(|m| m.get(node_id))
.map(|v| PropertyValue::Float(*v))
.unwrap_or(PropertyValue::Null),
"degree" => cache.degree.as_ref()
.and_then(|m| m.get(node_id))
.map(|v| PropertyValue::Float(*v))
.unwrap_or(PropertyValue::Null),
"community" | "community_fast" => cache.community.as_ref()
.and_then(|m| m.get(node_id))
.map(|v| PropertyValue::Int(*v as i64))
.unwrap_or(PropertyValue::Null),
"leiden" => cache.leiden.as_ref()
.and_then(|m| m.get(node_id))
.map(|v| PropertyValue::Int(*v as i64))
.unwrap_or(PropertyValue::Null),
_ => PropertyValue::Null,
}
}
#[cfg(not(feature = "algorithms"))]
pub fn lookup_algo_value(
_name: &str,
_node_id: &crate::NodeId,
_cache: &AlgoResults,
) -> PropertyValue {
PropertyValue::Null
}
struct QueryContext {
#[cfg_attr(not(feature = "algorithms"), allow(dead_code))]
algo: AlgoResults,
#[cfg(feature = "embeddings-index")]
emb_indices: HashMap<String, std::sync::Arc<crate::embeddings::HnswIndex>>,
}
impl QueryContext {
fn new(algo: AlgoResults) -> Self {
Self {
algo,
#[cfg(feature = "embeddings-index")]
emb_indices: HashMap::new(),
}
}
}
pub fn has_aggregations(projections: &[Projection]) -> bool {
projections.iter().any(|p| match p {
Projection::Expression { expr, .. } => is_aggregation_or_algorithm_expr(expr),
_ => false,
})
}
pub fn has_real_aggregations(projections: &[Projection]) -> bool {
projections.iter().any(|p| match p {
Projection::Expression { expr, .. } => expr.is_aggregation(),
_ => false,
})
}
pub fn has_algorithm_projections(projections: &[Projection]) -> bool {
projections.iter().any(|p| match p {
Projection::Expression { expr, .. } => expr.is_algorithm(),
_ => false,
})
}
fn is_aggregation_or_algorithm_expr(expr: &Expression) -> bool {
match expr {
Expression::FunctionCall { name, .. } => {
if matches!(name.as_str(), "count" | "sum" | "avg" | "min" | "max") {
return true;
}
if matches!(name.as_str(), "pagerank" | "betweenness" | "clustering" | "degree" | "community" | "community_fast" | "leiden" | "shortestPath") {
return true;
}
#[cfg(feature = "embeddings")]
if matches!(name.as_str(), "embedding_similarity" | "knn_nodes") {
return true;
}
false
}
_ => false,
}
}
pub async fn execute_aggregations(
graph: &Graph,
nodes: Vec<Node>,
query: &Query,
variable: &str,
) -> Result<QueryResult> {
let algo_results = precompute_algorithms(graph, &nodes, query).await?;
#[cfg(feature = "embeddings-index")]
let mut ctx = QueryContext::new(algo_results);
#[cfg(not(feature = "embeddings-index"))]
let ctx = QueryContext::new(algo_results);
#[cfg(feature = "embeddings-index")]
for projection in &query.find.projections {
if let Projection::Expression { expr, .. } = projection
&& let Expression::FunctionCall { name, args } = expr
&& name == "knn_nodes"
&& args.len() == 3
&& let Expression::Literal(PropertyValue::String(model)) = &args[2]
&& !ctx.emb_indices.contains_key(model)
&& let Ok(idx) = graph.get_or_build_embedding_index(model).await
{
ctx.emb_indices.insert(model.clone(), idx);
}
}
if let Some(group_by) = &query.group_by {
execute_grouped_aggregation(graph, nodes, query, variable, group_by, &ctx).await
} else {
let implicit_group_exprs: Vec<Expression> = query.find.projections.iter()
.filter_map(|projection| {
if let Projection::Expression { expr, .. } = projection
&& !is_aggregation_or_algorithm_expr(expr)
&& matches!(expr, Expression::Property { .. }) {
return Some(expr.clone());
}
None
})
.collect();
if implicit_group_exprs.is_empty() {
execute_simple_aggregation(graph, nodes, query, variable, &ctx).await
} else {
let implicit_group_by = GroupByClause {
expressions: implicit_group_exprs,
};
execute_grouped_aggregation(graph, nodes, query, variable, &implicit_group_by, &ctx).await
}
}
}
async fn execute_simple_aggregation(
graph: &Graph,
nodes: Vec<Node>,
query: &Query,
variable: &str,
ctx: &QueryContext,
) -> Result<QueryResult> {
let mut columns = Vec::new();
let mut row = Row::new();
for projection in &query.find.projections {
if let Projection::Expression { expr, alias } = projection {
let (key, value) = evaluate_aggregation(graph, expr, &nodes, variable, alias, ctx).await?;
columns.push(key.clone());
row.set(key, value);
}
}
let mut result = QueryResult::new(columns);
result.add_row(row);
Ok(result)
}
async fn execute_grouped_aggregation(
graph: &Graph,
nodes: Vec<Node>,
query: &Query,
variable: &str,
group_by: &crate::query::nql::parser::ast::GroupByClause,
ctx: &QueryContext,
) -> Result<QueryResult> {
let mut groups: std::collections::BTreeMap<String, Vec<Node>> = std::collections::BTreeMap::new();
for node in nodes {
let group_key = group_by.expressions.iter()
.map(|expr| evaluate_group_key(expr, &node, variable))
.collect::<Result<Vec<_>>>()?
.join("|");
groups.entry(group_key).or_default().push(node);
}
let node_id_col = format!("{}.id", variable);
let mut columns = vec![node_id_col.clone()];
for expr in &group_by.expressions {
if let Expression::Property { variable: v, property } = expr {
let raw_col = format!("{}.{}", v, property);
if raw_col == node_id_col {
continue;
}
let col_name = query.find.projections.iter()
.find_map(|p| {
if let Projection::Expression { expr: pexpr, alias: Some(a) } = p
&& let Expression::Property { variable: pv, property: pp } = pexpr
&& pv == v && pp == property
{
Some(a.clone())
} else {
None
}
})
.unwrap_or(raw_col);
columns.push(col_name);
}
}
for projection in &query.find.projections {
if let Projection::Expression { expr, alias } = projection
&& is_aggregation_or_algorithm_expr(expr)
&& let Expression::FunctionCall { name, .. } = expr {
let col_name = alias.clone().unwrap_or_else(|| name.clone());
columns.push(col_name);
}
}
let mut result = QueryResult::new(columns);
for (group_key, group_nodes) in groups {
let mut row = Row::new();
if let Some(first_node) = group_nodes.first() {
row.set(node_id_col.clone(), PropertyValue::String(first_node.id.to_string()));
}
let key_parts: Vec<&str> = group_key.split('|').collect();
for (i, expr) in group_by.expressions.iter().enumerate() {
if let Expression::Property { variable: v, property } = expr {
let raw_key = format!("{}.{}", v, property);
if raw_key == node_id_col {
continue;
}
let row_key = query.find.projections.iter()
.find_map(|p| {
if let Projection::Expression { expr: pexpr, alias: Some(a) } = p
&& let Expression::Property { variable: pv, property: pp } = pexpr
&& pv == v && pp == property
{
Some(a.clone())
} else {
None
}
})
.unwrap_or(raw_key);
let value = key_parts.get(i)
.map(|s| s.to_string())
.unwrap_or_else(|| "null".to_string());
row.set(row_key, PropertyValue::String(value));
}
}
for projection in &query.find.projections {
if let Projection::Expression { expr, alias } = projection
&& is_aggregation_or_algorithm_expr(expr) {
let (key, value) = evaluate_aggregation(graph, expr, &group_nodes, variable, alias, ctx).await?;
row.set(key, value);
}
}
result.add_row(row);
}
Ok(result)
}
fn evaluate_group_key(expr: &Expression, node: &Node, variable: &str) -> Result<String> {
match expr {
Expression::Property { variable: var, property } => {
if var != variable {
return Err(NopalError::QueryExecutionError(
format!("Unknown variable in GROUP BY: {}", var)
));
}
if property == "label" {
return Ok(node.label.clone());
}
if property == "id" {
return Ok(node.id.to_string());
}
Ok(node.properties.get(property)
.map(|v| match v {
PropertyValue::String(s) => s.clone(),
PropertyValue::Int(i) => i.to_string(),
PropertyValue::Float(f) => f.to_string(),
PropertyValue::Bool(b) => b.to_string(),
PropertyValue::Null => "null".to_string(),
PropertyValue::Bytes(_) => "<bytes>".to_string(),
PropertyValue::List(_) => "<list>".to_string(),
PropertyValue::Object(_) => "<object>".to_string(),
})
.unwrap_or_else(|| "null".to_string()))
}
_ => Err(NopalError::QueryExecutionError(
"GROUP BY only supports property expressions".into()
)),
}
}
#[allow(unused_variables)]
async fn evaluate_aggregation(
graph: &Graph,
expr: &Expression,
nodes: &[Node],
variable: &str,
alias: &Option<String>,
ctx: &QueryContext,
) -> Result<(String, PropertyValue)> {
match expr {
Expression::FunctionCall { name, args } => {
let result_key = alias.clone().unwrap_or_else(|| name.clone());
match name.as_str() {
"count" => {
let count = nodes.len() as i64;
Ok((result_key, PropertyValue::Int(count)))
}
"sum" => {
if args.is_empty() {
return Err(NopalError::QueryExecutionError(
"sum() requires an argument".into()
));
}
let sum = sum_property(nodes, &args[0], variable)?;
Ok((result_key, PropertyValue::Float(sum)))
}
"avg" => {
if args.is_empty() {
return Err(NopalError::QueryExecutionError(
"avg() requires an argument".into()
));
}
let sum = sum_property(nodes, &args[0], variable)?;
let avg = sum / nodes.len() as f64;
Ok((result_key, PropertyValue::Float(avg)))
}
"min" => {
if args.is_empty() {
return Err(NopalError::QueryExecutionError(
"min() requires an argument".into()
));
}
let min = min_property(nodes, &args[0], variable)?;
Ok((result_key, min))
}
"max" => {
if args.is_empty() {
return Err(NopalError::QueryExecutionError(
"max() requires an argument".into()
));
}
let max = max_property(nodes, &args[0], variable)?;
Ok((result_key, max))
}
#[cfg(feature = "algorithms")]
"pagerank" => {
let ranks = ctx.algo.pagerank.as_ref().ok_or_else(|| NopalError::QueryExecutionError("pagerank algorithm not precomputed".into()))?;
Ok((result_key, PropertyValue::Float(average_algo_scores(nodes, ranks))))
}
#[cfg(feature = "algorithms")]
"betweenness" => {
let scores = ctx.algo.betweenness.as_ref().ok_or_else(|| NopalError::QueryExecutionError("betweenness algorithm not precomputed".into()))?;
Ok((result_key, PropertyValue::Float(average_algo_scores(nodes, scores))))
}
#[cfg(feature = "algorithms")]
"clustering" => {
let coeffs = ctx.algo.clustering.as_ref().ok_or_else(|| NopalError::QueryExecutionError("clustering algorithm not precomputed".into()))?;
Ok((result_key, PropertyValue::Float(average_algo_scores(nodes, coeffs))))
}
#[cfg(feature = "algorithms")]
"degree" => {
let degrees = ctx.algo.degree.as_ref().ok_or_else(|| NopalError::QueryExecutionError("degree algorithm not precomputed".into()))?;
Ok((result_key, PropertyValue::Float(average_algo_scores(nodes, degrees))))
}
#[cfg(feature = "algorithms")]
"community" | "community_fast" => {
let communities = ctx.algo.community.as_ref().ok_or_else(|| NopalError::QueryExecutionError("community algorithm not precomputed".into()))?;
Ok((result_key, PropertyValue::Float(average_algo_scores_usize(nodes, communities))))
}
#[cfg(feature = "algorithms")]
"leiden" => {
let communities = ctx.algo.leiden.as_ref().ok_or_else(|| NopalError::QueryExecutionError(
"leiden algorithm not precomputed — asegúrate de tener feature `algorithms`".into()
))?;
Ok((result_key, PropertyValue::Float(average_algo_scores_usize(nodes, communities))))
}
#[cfg(feature = "algorithms")]
"shortestPath" => {
execute_shortest_path(graph, args, &result_key).await
}
#[cfg(not(feature = "algorithms"))]
"pagerank" | "betweenness" | "clustering" | "degree" | "community" | "community_fast" | "leiden" | "shortestPath" => {
Err(NopalError::QueryExecutionError(
format!("{}() requires feature `algorithms` (enable `--features core` or `--features algorithms`)", name)
))
}
#[cfg(feature = "embeddings")]
"embedding_similarity" => {
execute_embedding_similarity(graph, args, nodes, &result_key).await
}
#[cfg(feature = "embeddings-index")]
"knn_nodes" => {
execute_knn_nodes(graph, args, nodes, &result_key, ctx).await
}
_ => Err(NopalError::QueryExecutionError(
format!("Unknown aggregation function: {}", name)
))
}
}
_ => Err(NopalError::QueryExecutionError(
"Not an aggregation expression".into()
))
}
}
fn sum_property(nodes: &[Node], expr: &Expression, variable: &str) -> Result<f64> {
match expr {
Expression::Property { variable: var, property } => {
if var != variable {
return Err(NopalError::QueryExecutionError(
format!("Unknown variable: {}", var)
));
}
let mut sum = 0.0;
for node in nodes {
if let Some(value) = node.properties.get(property) {
sum += match value {
PropertyValue::Int(i) => *i as f64,
PropertyValue::Float(f) => *f,
_ => 0.0,
};
}
}
Ok(sum)
}
_ => Err(NopalError::QueryExecutionError(
"Aggregation argument must be a property".into()
))
}
}
fn min_property(nodes: &[Node], expr: &Expression, variable: &str) -> Result<PropertyValue> {
match expr {
Expression::Property { variable: var, property } => {
if var != variable {
return Err(NopalError::QueryExecutionError(
format!("Unknown variable: {}", var)
));
}
let mut min: Option<PropertyValue> = None;
for node in nodes {
if let Some(value) = node.properties.get(property)
&& min.as_ref().is_none_or(|m| is_less_than(value, m)) {
min = Some(value.clone());
}
}
min.ok_or_else(|| NopalError::QueryExecutionError(
"No values found for min()".into()
))
}
_ => Err(NopalError::QueryExecutionError(
"Aggregation argument must be a property".into()
))
}
}
fn max_property(nodes: &[Node], expr: &Expression, variable: &str) -> Result<PropertyValue> {
match expr {
Expression::Property { variable: var, property } => {
if var != variable {
return Err(NopalError::QueryExecutionError(
format!("Unknown variable: {}", var)
));
}
let mut max: Option<PropertyValue> = None;
for node in nodes {
if let Some(value) = node.properties.get(property)
&& max.as_ref().is_none_or(|m| is_greater_than(value, m)) {
max = Some(value.clone());
}
}
max.ok_or_else(|| NopalError::QueryExecutionError(
"No values found for max()".into()
))
}
_ => Err(NopalError::QueryExecutionError(
"Aggregation argument must be a property".into()
))
}
}
fn is_less_than(a: &PropertyValue, b: &PropertyValue) -> bool {
match (a, b) {
(PropertyValue::Int(a), PropertyValue::Int(b)) => a < b,
(PropertyValue::Float(a), PropertyValue::Float(b)) => a < b,
(PropertyValue::Int(a), PropertyValue::Float(b)) => (*a as f64) < *b,
(PropertyValue::Float(a), PropertyValue::Int(b)) => *a < (*b as f64),
_ => false,
}
}
fn is_greater_than(a: &PropertyValue, b: &PropertyValue) -> bool {
match (a, b) {
(PropertyValue::Int(a), PropertyValue::Int(b)) => a > b,
(PropertyValue::Float(a), PropertyValue::Float(b)) => a > b,
(PropertyValue::Int(a), PropertyValue::Float(b)) => (*a as f64) > *b,
(PropertyValue::Float(a), PropertyValue::Int(b)) => *a > (*b as f64),
_ => false,
}
}
#[cfg(feature = "algorithms")]
async fn precompute_algorithms(
graph: &Graph,
nodes: &[Node],
query: &Query,
) -> Result<AlgoResults> {
let mut results = AlgoResults::default();
let mut needed: std::collections::HashSet<String> = std::collections::HashSet::new();
for projection in &query.find.projections {
if let Projection::Expression { expr, .. } = projection {
collect_algorithm_names(expr, &mut needed);
}
}
if let Some(filter) = &query.filter {
collect_algorithm_names(&filter.condition, &mut needed);
}
if let Some(having) = &query.having {
collect_algorithm_names(&having.condition, &mut needed);
}
if needed.is_empty() && !has_aggregations(&query.find.projections) {
return Ok(results);
}
use crate::graph::view::Subgraph;
let subgraph = Subgraph::from_nodes(graph, nodes);
for name in &needed {
match name.as_str() {
"pagerank" if results.pagerank.is_none() => {
use crate::algorithms::pagerank::{PageRank, PageRankConfig};
let pr = PageRank::new(PageRankConfig { damping: 0.85, max_iterations: 100, tolerance: 1e-6, parallel: false });
results.pagerank = Some(pr.compute(graph).await?);
}
"betweenness" if results.betweenness.is_none() => {
use crate::algorithms::betweenness::{BetweennessCentrality, BetweennessConfig};
let bc = BetweennessCentrality::new(BetweennessConfig { normalize: true, parallel: false });
results.betweenness = Some(bc.compute(&subgraph).await?);
}
"clustering" if results.clustering.is_none() => {
use crate::algorithms::clustering::{ClusteringCoefficient, ClusteringConfig};
let cc = ClusteringCoefficient::new(ClusteringConfig { weighted: false });
results.clustering = Some(cc.compute(&subgraph).await?);
}
"degree" if results.degree.is_none() => {
use crate::algorithms::degree::{DegreeCentrality, DegreeConfig, DegreeType};
let dc = DegreeCentrality::new(DegreeConfig { degree_type: DegreeType::Total, normalize: false });
results.degree = Some(dc.compute(graph).await?);
}
"community" if results.community.is_none() => {
results.community = Some(get_or_compute_community_exact(graph).await?);
}
"community_fast" if results.community.is_none() => {
results.community = Some(compute_community_fast_map(graph, nodes).await?);
}
"leiden" if results.leiden.is_none() => {
results.leiden = Some(get_or_compute_leiden(graph).await?);
}
_ => {}
}
}
Ok(results)
}
#[cfg(feature = "algorithms")]
fn collect_algorithm_names(expr: &Expression, out: &mut std::collections::HashSet<String>) {
match expr {
Expression::FunctionCall { name, args } => {
let lower = name.to_lowercase();
if matches!(lower.as_str(),
"pagerank" | "betweenness" | "clustering" | "degree"
| "community" | "community_fast" | "leiden")
{
out.insert(lower);
}
for a in args {
collect_algorithm_names(a, out);
}
}
Expression::BinaryOp { left, right, .. } => {
collect_algorithm_names(left, out);
collect_algorithm_names(right, out);
}
Expression::UnaryOp { expr, .. } => collect_algorithm_names(expr, out),
_ => {}
}
}
#[cfg(not(feature = "algorithms"))]
async fn precompute_algorithms(
_graph: &Graph,
_nodes: &[Node],
_query: &Query,
) -> Result<AlgoResults> {
Ok(AlgoResults::default())
}
#[cfg(feature = "algorithms")]
fn average_algo_scores(nodes: &[Node], scores: &HashMap<NodeId, f64>) -> f64 {
if nodes.is_empty() { return 0.0; }
let sum: f64 = nodes.iter().filter_map(|n| scores.get(&n.id).copied()).sum();
sum / nodes.len() as f64
}
#[cfg(feature = "algorithms")]
fn average_algo_scores_usize(nodes: &[Node], scores: &HashMap<NodeId, usize>) -> f64 {
if nodes.is_empty() { return 0.0; }
let sum: f64 = nodes.iter().filter_map(|n| scores.get(&n.id).map(|&v| v as f64)).sum();
sum / nodes.len() as f64
}
#[cfg(feature = "algorithms")]
async fn get_or_compute_community_exact(graph: &Graph) -> Result<HashMap<uuid::Uuid, usize>> {
use crate::algorithms::community::LouvainCommunity;
let current_topology = graph.topology_version();
if let Some((cached_topology, assignments)) = graph.get_cached_community_partition_exact().await
&& cached_topology == current_topology {
return Ok(assignments);
}
let lc = LouvainCommunity::with_defaults();
let assignments = lc.detect(graph).await?;
graph
.set_cached_community_partition_exact(current_topology, assignments.clone())
.await;
Ok(assignments)
}
#[cfg(feature = "algorithms")]
async fn get_or_compute_leiden(graph: &Graph) -> Result<HashMap<uuid::Uuid, usize>> {
use crate::algorithms::community::LeidenCommunity;
let current_topology = graph.topology_version();
if let Some((cached_topology, assignments)) = graph.get_cached_leiden_partition().await
&& cached_topology == current_topology
{
return Ok(assignments);
}
let leiden = LeidenCommunity::with_defaults();
let assignments = leiden.detect(graph).await?;
graph.set_cached_leiden_partition(current_topology, assignments.clone()).await;
Ok(assignments)
}
#[cfg(feature = "algorithms")]
async fn compute_community_fast_map(
graph: &Graph,
nodes: &[Node],
) -> Result<HashMap<NodeId, usize>> {
if nodes.is_empty() {
return Ok(HashMap::new());
}
let selected_ids: HashSet<uuid::Uuid> = nodes.iter().map(|n| n.id).collect();
let mut adjacency: HashMap<uuid::Uuid, Vec<uuid::Uuid>> = HashMap::new();
for node in nodes {
let mut local = graph.neighbors(node.id, Direction::Both).await?
.into_iter()
.filter(|nid| selected_ids.contains(nid))
.collect::<Vec<_>>();
local.sort_unstable();
local.dedup();
adjacency.insert(node.id, local);
}
let mut labels: HashMap<uuid::Uuid, usize> = HashMap::new();
for (idx, node) in nodes.iter().enumerate() {
labels.insert(node.id, idx);
}
for _ in 0..3 {
let mut changed = false;
let mut next_labels = labels.clone();
for node in nodes {
let neighbors = adjacency
.get(&node.id)
.map(|n| n.as_slice())
.unwrap_or(&[]);
if neighbors.is_empty() {
continue;
}
let mut counts: HashMap<usize, usize> = HashMap::new();
for neighbor in neighbors {
if let Some(label) = labels.get(neighbor) {
*counts.entry(*label).or_insert(0) += 1;
}
}
if counts.is_empty() {
continue;
}
let best_label = counts
.into_iter()
.max_by_key(|(label, count)| (*count, Reverse(*label)))
.map(|(label, _)| label)
.unwrap_or_else(|| labels[&node.id]);
if best_label != labels[&node.id] {
changed = true;
next_labels.insert(node.id, best_label);
}
}
labels = next_labels;
if !changed {
break;
}
}
let mut compact_map: HashMap<usize, usize> = HashMap::new();
let mut next_compact = 0usize;
for node in nodes {
let label = labels[&node.id];
compact_map.entry(label).or_insert_with(|| {
let current = next_compact;
next_compact += 1;
current
});
}
let result: HashMap<NodeId, usize> = nodes
.iter()
.map(|n| (n.id, compact_map[&labels[&n.id]]))
.collect();
Ok(result)
}
#[cfg(feature = "algorithms")]
async fn execute_shortest_path(
graph: &Graph,
args: &[Expression],
result_key: &str,
) -> Result<(String, PropertyValue)> {
use crate::algorithms::shortest_path::{ShortestPath, ShortestPathConfig};
use uuid::Uuid;
if args.len() != 2 {
return Err(NopalError::QueryExecutionError(
"shortestPath() requires exactly two arguments (source_id, target_id)".into(),
));
}
let source_str = match &args[0] {
Expression::Literal(PropertyValue::String(s)) => s.clone(),
_ => return Err(NopalError::QueryExecutionError(
"shortestPath() arguments must be string literals (node UUIDs)".into(),
)),
};
let target_str = match &args[1] {
Expression::Literal(PropertyValue::String(s)) => s.clone(),
_ => return Err(NopalError::QueryExecutionError(
"shortestPath() arguments must be string literals (node UUIDs)".into(),
)),
};
let source = Uuid::parse_str(&source_str).map_err(|_| {
NopalError::QueryExecutionError(format!("Invalid node UUID for source: {}", source_str))
})?;
let target = Uuid::parse_str(&target_str).map_err(|_| {
NopalError::QueryExecutionError(format!("Invalid node UUID for target: {}", target_str))
})?;
let sp = ShortestPath::new(ShortestPathConfig::default());
let result = sp.find_path(graph, source, target).await?;
let distance = match result {
Some(path_result) => path_result.distance,
None => -1.0, };
Ok((result_key.to_string(), PropertyValue::Float(distance)))
}
#[cfg(feature = "embeddings")]
async fn execute_embedding_similarity(
graph: &Graph,
args: &[Expression],
nodes: &[Node],
result_key: &str,
) -> Result<(String, PropertyValue)> {
if args.len() != 3 {
return Err(NopalError::QueryExecutionError(
"embedding_similarity(n, \"ref-uuid\", \"model\") requires 3 arguments".into(),
));
}
let ref_uuid_str = match &args[1] {
Expression::Literal(PropertyValue::String(s)) => s.clone(),
_ => return Err(NopalError::QueryExecutionError(
"embedding_similarity: second argument must be a UUID string".into(),
)),
};
let model = match &args[2] {
Expression::Literal(PropertyValue::String(s)) => s.clone(),
_ => return Err(NopalError::QueryExecutionError(
"embedding_similarity: third argument must be a model name string".into(),
)),
};
let ref_id = uuid::Uuid::parse_str(&ref_uuid_str).map_err(|_| {
NopalError::QueryExecutionError(format!(
"embedding_similarity: invalid UUID '{}'", ref_uuid_str
))
})?;
let ref_emb = graph.get_node_embedding(ref_id, &model).await
.map_err(|_| NopalError::QueryExecutionError(format!(
"embedding_similarity: no embedding found for reference node '{}' model '{}'",
ref_uuid_str, model
)))?;
let mut total = 0.0_f64;
let mut count = 0_usize;
for node in nodes {
if let Ok(emb) = graph.get_node_embedding(node.id, &model).await {
total += emb.cosine_similarity(&ref_emb) as f64;
count += 1;
}
}
let avg = if count > 0 { total / count as f64 } else { 0.0 };
Ok((result_key.to_string(), PropertyValue::Float(avg)))
}
#[cfg(feature = "embeddings-index")]
async fn execute_knn_nodes(
graph: &Graph,
args: &[Expression],
nodes: &[Node],
result_key: &str,
ctx: &QueryContext,
) -> Result<(String, PropertyValue)> {
if args.len() != 3 {
return Err(NopalError::QueryExecutionError(
"knn_nodes(n, k, \"model\") requires 3 arguments".into(),
));
}
let k = match &args[1] {
Expression::Literal(PropertyValue::Int(k)) => *k as usize,
_ => return Err(NopalError::QueryExecutionError(
"knn_nodes: second argument must be an integer k".into(),
)),
};
let model = match &args[2] {
Expression::Literal(PropertyValue::String(s)) => s.clone(),
_ => return Err(NopalError::QueryExecutionError(
"knn_nodes: third argument must be a model name string".into(),
)),
};
let Some(node) = nodes.first() else {
return Ok((result_key.to_string(), PropertyValue::String("[]".into())));
};
let emb = match graph.get_node_embedding(node.id, &model).await {
Ok(e) => e,
Err(_) => return Ok((result_key.to_string(), PropertyValue::String("[]".into()))),
};
let index = ctx.emb_indices.get(&model).ok_or_else(|| {
NopalError::QueryExecutionError(format!(
"knn_nodes: no embedding index for model '{}' (not precomputed)",
model
))
})?;
let results = index.search_knn(&emb.vector, k)?;
let json = format!(
"[{}]",
results.iter()
.map(|(id, _dist)| format!("\"{}\"", id))
.collect::<Vec<_>>()
.join(",")
);
Ok((result_key.to_string(), PropertyValue::String(json)))
}