diff --git a/crates/lance-graph/README.md b/crates/lance-graph/README.md index e5172fb..591531c 100644 --- a/crates/lance-graph/README.md +++ b/crates/lance-graph/README.md @@ -19,6 +19,13 @@ Replace a directly constructed `Some(n)` with `Some(ast::RowCount::Literal(n))`; use `ast::RowCount::Parameter(name)`. Resolve parameter values before building a logical plan, using the normal `CypherQuery` parameter-binding methods. Existing Cypher text with literal pagination is unchanged. +## Rust AST migration for list folds + +`ast::ValueExpression` adds `Reduce`, `ListPredicate` and `LocalVariable` variants. Update exhaustive +matches and visitors to traverse the fold input and body while respecting iterator and accumulator +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. + ## Quick Start ```rust diff --git a/crates/lance-graph/src/ast.rs b/crates/lance-graph/src/ast.rs index 7d4e183..d6834ad 100644 --- a/crates/lance-graph/src/ast.rs +++ b/crates/lance-graph/src/ast.rs @@ -443,6 +443,23 @@ pub enum ValueExpression { filter: Option>, map: Option>, }, + /// A lexically scoped variable in a reduce or list predicate body. + LocalVariable(String), + /// `reduce(accumulator = initial, variable IN list | expression)`. + Reduce { + accumulator: String, + initial: Box, + variable: String, + list: Box, + expression: Box, + }, + /// `all/any/none/single(variable IN list WHERE predicate)`. + ListPredicate { + kind: String, + variable: String, + list: Box, + predicate: Box, + }, /// Vector literal: [0.1, 0.2, 0.3] /// Represents an inline vector for similarity search VectorLiteral(Vec), @@ -662,7 +679,24 @@ pub fn transform_value(expr: &mut ValueExpression, f: &mut dyn FnMut(&mut ValueE } transform_value(map, f); } + ValueExpression::Reduce { + initial, + list, + expression, + .. + } => { + transform_value(initial, f); + transform_value(list, f); + transform_value(expression, f); + } + ValueExpression::ListPredicate { + list, predicate, .. + } => { + transform_value(list, f); + transform_value(predicate, f); + } ValueExpression::Variable(_) + | ValueExpression::LocalVariable(_) | ValueExpression::Property(_) | ValueExpression::Literal(_) | ValueExpression::Parameter(_) @@ -671,6 +705,51 @@ pub fn transform_value(expr: &mut ValueExpression, f: &mut dyn FnMut(&mut ValueE f(expr); } +// Mark only the body's local references; the input and initializer retain outer scope. +pub(crate) fn mark_locals(expr: &mut ValueExpression, names: &[&str]) { + transform_value(expr, &mut |value| match value { + ValueExpression::Variable(name) if names.iter().any(|n| n.eq_ignore_ascii_case(name)) => { + *value = ValueExpression::LocalVariable(name.to_lowercase()); + } + ValueExpression::Property(p) + if names.iter().any(|n| n.eq_ignore_ascii_case(&p.variable)) => + { + *value = ValueExpression::Field { + base: Box::new(ValueExpression::LocalVariable(p.variable.to_lowercase())), + field: p.property.clone(), + }; + } + _ => {} + }); +} + +/// Expressions that need relational planning rather than scalar lambda execution. +pub(crate) fn needs_relational_evaluation(expr: &ValueExpression) -> bool { + fn boolean_has_pattern(expr: &BooleanExpression) -> bool { + match expr { + BooleanExpression::PatternExists { .. } => true, + BooleanExpression::And(a, b) | BooleanExpression::Or(a, b) => { + boolean_has_pattern(a) || boolean_has_pattern(b) + } + BooleanExpression::Not(b) => boolean_has_pattern(b), + _ => false, + } + } + let mut found = false; + transform_value(&mut expr.clone(), &mut |value| { + found |= match value { + ValueExpression::PatternComprehension { .. } + | ValueExpression::ListComprehension { .. } => true, + ValueExpression::Boolean(b) => boolean_has_pattern(b), + ValueExpression::Case { whens, .. } => { + whens.iter().any(|(b, _)| boolean_has_pattern(b)) + } + _ => false, + }; + }); + found +} + /// [`transform_value`] over every value in a predicate. pub fn transform_boolean_values( expr: &mut BooleanExpression, diff --git a/crates/lance-graph/src/datafusion_planner/builder/aggregate_ops.rs b/crates/lance-graph/src/datafusion_planner/builder/aggregate_ops.rs index 0cf71fc..d03293d 100644 --- a/crates/lance-graph/src/datafusion_planner/builder/aggregate_ops.rs +++ b/crates/lance-graph/src/datafusion_planner/builder/aggregate_ops.rs @@ -14,8 +14,8 @@ use datafusion::logical_expr::{col, LogicalPlan, LogicalPlanBuilder}; /// first and apply the outer expression after. Cypher's `collect` gives `[]`, not null, /// when nothing was collected. fn split_aggregates(expr: &mut ValueExpression, aggregates: &mut Vec<(String, ValueExpression)>) { - match expr { - ValueExpression::AggregateFunction { name, .. } => { + crate::ast::transform_value(expr, &mut |expr| { + if let ValueExpression::AggregateFunction { name, .. } = expr { let column = format!("__agg_{}", aggregates.len()); let is_collect = name.eq_ignore_ascii_case("collect"); let aggregate = std::mem::replace(expr, ValueExpression::Variable(column.clone())); @@ -33,23 +33,7 @@ fn split_aggregates(expr: &mut ValueExpression, aggregates: &mut Vec<(String, Va }; } } - ValueExpression::ScalarFunction { args, .. } => args - .iter_mut() - .for_each(|a| split_aggregates(a, aggregates)), - ValueExpression::Arithmetic { left, right, .. } => { - split_aggregates(left, aggregates); - split_aggregates(right, aggregates); - } - ValueExpression::Case { otherwise, whens } => { - for (_, then) in whens { - split_aggregates(then, aggregates); - } - if let Some(e) = otherwise { - split_aggregates(e, aggregates); - } - } - _ => {} - } + }); } impl DataFusionPlanner { @@ -96,7 +80,7 @@ impl DataFusionPlanner { let first = aggregates.len(); split_aggregates(&mut rewritten, &mut aggregates); for (column, aggregate) in &aggregates[first..] { - let mut aggregate = super::super::expression::to_df_value_expr(aggregate); + let mut aggregate = self.value_expr(aggregate, &schema)?; // collect() gathers in the order of a preceding `WITH ... ORDER BY`. if let datafusion::logical_expr::Expr::AggregateFunction(f) = &mut aggregate { if f.func.name() == "array_agg" && !f.params.distinct { @@ -108,7 +92,7 @@ impl DataFusionPlanner { outer.push(super::super::expression::to_df_value_expr(&rewritten).alias(alias)); } else { // Group expressions: use raw expression for grouping, no alias - group_exprs.push(super::super::expression::to_df_value_expr(&p.expression)); + group_exprs.push(self.value_expr(&p.expression, &schema)?); } } @@ -145,9 +129,15 @@ impl DataFusionPlanner { } } - LogicalPlanBuilder::from(input_plan) + let aggregate = LogicalPlanBuilder::from(input_plan) .aggregate(group_exprs, agg_exprs) - .map_err(|e| self.plan_error("Failed to build aggregate", e))? + .and_then(|b| b.build()) + .map_err(|e| self.plan_error("Failed to build aggregate", e))?; + let final_projection = final_projection + .into_iter() + .map(|expr| self.resolve_expr(expr, aggregate.schema())) + .collect::>>()?; + LogicalPlanBuilder::from(aggregate) .project(final_projection) .map_err(|e| self.plan_error("Failed to project after aggregate", e))? .build() diff --git a/crates/lance-graph/src/datafusion_planner/builder/basic_ops.rs b/crates/lance-graph/src/datafusion_planner/builder/basic_ops.rs index c10b116..5688010 100644 --- a/crates/lance-graph/src/datafusion_planner/builder/basic_ops.rs +++ b/crates/lance-graph/src/datafusion_planner/builder/basic_ops.rs @@ -86,8 +86,7 @@ impl DataFusionPlanner { predicate: &crate::ast::BooleanExpression, ) -> Result { let input_plan = self.build_operator(ctx, input)?; - let predicate = super::super::expression::bind_boolean(predicate, input_plan.schema()); - let expr = super::super::expression::to_df_boolean_expr(&predicate); + let expr = self.boolean_expr(predicate, input_plan.schema())?; LogicalPlanBuilder::from(input_plan) .filter(expr) .map_err(|e| self.plan_error("Failed to build filter", e))? @@ -119,7 +118,16 @@ impl DataFusionPlanner { } = input { let input_plan = self.build_operator(ctx, inner)?; - let ordering = self.sort_exprs(&input_plan, sort_items); + let ordering = self + .sort_exprs(&input_plan, sort_items) + .into_iter() + .map(|sort| { + Ok(SortExpr { + expr: self.resolve_expr(sort.expr, input_plan.schema())?, + ..sort + }) + }) + .collect::>>()?; return self.build_project_with_aggregates(input_plan, projections, &ordering); } let input_plan = self.build_operator(ctx, input)?; @@ -174,8 +182,7 @@ impl DataFusionPlanner { exprs.extend(columns); continue; } - let bound = super::super::expression::bind_value(&p.expression, &schema); - let expr = super::super::expression::to_df_value_expr(&bound); + let expr = self.value_expr(&p.expression, &schema)?; // Apply alias if provided, otherwise use Cypher dot notation // Normalize alias to lowercase for case-insensitive behavior exprs.push(if let Some(alias) = &p.alias { @@ -219,6 +226,8 @@ impl DataFusionPlanner { .sort(sort_exprs) .map_err(|e| self.plan_error("Failed to build sort", e))? .build() + // The sort builder first restores input columns omitted by RETURN. + .and_then(|plan| plan.resolve_lambda_variables().map(|p| p.data)) .map_err(|e| self.plan_error("Failed to build plan", e)) } @@ -288,8 +297,7 @@ impl DataFusionPlanner { }; // Convert expression to DataFusion Expr - let bound = super::super::expression::bind_value(expression, input_plan.schema()); - let df_expr = super::super::expression::to_df_value_expr(&bound); + let df_expr = self.value_expr(expression, input_plan.schema())?; // A list of node ids: unnest under a temporary name, then join each id back to the // node table so the alias carries the node's columns like any matched node. let unnested = match label { diff --git a/crates/lance-graph/src/datafusion_planner/builder/comprehension_ops.rs b/crates/lance-graph/src/datafusion_planner/builder/comprehension_ops.rs index 43b77d2..d7f2e63 100644 --- a/crates/lance-graph/src/datafusion_planner/builder/comprehension_ops.rs +++ b/crates/lance-graph/src/datafusion_planner/builder/comprehension_ops.rs @@ -9,14 +9,13 @@ use arrow_schema::DataType; use datafusion::common::Column; use datafusion::functions::core::expr_fn::coalesce; use datafusion::functions_aggregate::array_agg::array_agg; -use datafusion::functions_nested::expr_fn::{array_element, cardinality, gen_series, make_array}; +use datafusion::functions_nested::expr_fn::{array_element, array_length, gen_series, make_array}; use datafusion::functions_window::expr_fn::row_number; use datafusion::logical_expr::{ - col, lit, Cast, Expr, ExprFunctionExt, JoinType, LogicalPlan, LogicalPlanBuilder, + col, lit, Cast, Expr, ExprFunctionExt, ExprSchemable, JoinType, LogicalPlan, LogicalPlanBuilder, }; use super::super::analysis::PlanningContext; -use super::super::expression::{bind_boolean, bind_value, to_df_boolean_expr, to_df_value_expr}; use crate::ast::{BooleanExpression, ValueExpression}; use crate::case_insensitive::qualify_column; use crate::datafusion_planner::DataFusionPlanner; @@ -54,8 +53,17 @@ impl DataFusionPlanner { .map_err(error("Failed to number comprehension rows"))? .build() .map_err(error("Failed to number comprehension rows"))?; - let list = to_df_value_expr(&bind_value(list, numbered.schema())); - let length = int64(coalesce(vec![cardinality(list.clone()), lit(0_u64)])); + let list = self.value_expr(list, numbered.schema())?; + let list = if list + .get_type(numbered.schema()) + .map_err(error("Failed to type comprehension list"))? + == DataType::Null + { + datafusion::logical_expr::cast(list, DataType::new_list(DataType::Null, true)) + } else { + list + }; + let length = int64(coalesce(vec![array_length(list.clone()), lit(0_u64)])); let last = datafusion::functions::core::expr_fn::greatest(vec![length.clone(), lit(1_i64)]); let positions = gen_series(lit(1_i64), last, lit(1_i64)); let exploded = LogicalPlanBuilder::from(numbered.clone()) @@ -130,9 +138,9 @@ impl DataFusionPlanner { // Padding rows (position past the list's end) are never collected. let mut kept = col(&pos).lt_eq(length); if let Some(keep) = keep { - kept = kept.and(to_df_boolean_expr(&bind_boolean(keep, elements.schema()))); + kept = kept.and(self.boolean_expr(keep, elements.schema())?); } - let value = to_df_value_expr(&bind_value(map, elements.schema())); + let value = self.value_expr(map, elements.schema())?; let collected = array_agg(value) .filter(kept) .order_by(vec![col(&pos).sort(true, false)]) diff --git a/crates/lance-graph/src/datafusion_planner/builder/join_builder.rs b/crates/lance-graph/src/datafusion_planner/builder/join_builder.rs index 7fc3ac3..4777885 100644 --- a/crates/lance-graph/src/datafusion_planner/builder/join_builder.rs +++ b/crates/lance-graph/src/datafusion_planner/builder/join_builder.rs @@ -50,11 +50,13 @@ impl DataFusionPlanner { .project(right_exprs) .and_then(|right| right.build()) .and_then(|right| { - LogicalPlanBuilder::from(left).join( + LogicalPlanBuilder::from(left).join_detailed( right, datafusion::logical_expr::JoinType::Left, (keys.to_vec(), right_keys), None, + // Captured scalar nulls identify the same correlation binding. + datafusion::common::NullEquality::NullEqualsNull, ) }) .and_then(|b| b.build()) diff --git a/crates/lance-graph/src/datafusion_planner/expression.rs b/crates/lance-graph/src/datafusion_planner/expression.rs index 555528a..919cfc6 100644 --- a/crates/lance-graph/src/datafusion_planner/expression.rs +++ b/crates/lance-graph/src/datafusion_planner/expression.rs @@ -19,6 +19,34 @@ use datafusion_functions_aggregate::min_max::max; use datafusion_functions_aggregate::min_max::min; use datafusion_functions_aggregate::sum::sum; +impl super::DataFusionPlanner { + pub(crate) fn resolve_expr( + &self, + expr: Expr, + schema: &datafusion::common::DFSchema, + ) -> crate::error::Result { + expr.resolve_lambda_variables(schema) + .map(|e| e.data) + .map_err(|e| self.plan_error("Failed to resolve expression scope", e)) + } + + pub(crate) fn value_expr( + &self, + expr: &ValueExpression, + schema: &datafusion::common::DFSchema, + ) -> crate::error::Result { + self.resolve_expr(to_df_value_expr(&bind_value(expr, schema)), schema) + } + + pub(crate) fn boolean_expr( + &self, + expr: &BooleanExpression, + schema: &datafusion::common::DFSchema, + ) -> crate::error::Result { + self.resolve_expr(to_df_boolean_expr(&bind_boolean(expr, schema)), schema) + } +} + /// Helper function to create LIKE expressions with consistent settings fn create_like_expr(expression: &ValueExpression, pattern: &str, case_insensitive: bool) -> Expr { Expr::Like(datafusion::logical_expr::Like { @@ -122,6 +150,32 @@ pub(crate) fn to_df_value_expr(expr: &ValueExpression) -> Expr { col(qualify_column(&prop.variable, &prop.property)) } VE::Variable(v) => col(v.to_lowercase()), + VE::LocalVariable(v) => Expr::LambdaVariable( + datafusion::logical_expr::expr::LambdaVariable::new(v.to_lowercase(), None), + ), + VE::Reduce { + accumulator, + initial, + variable, + list, + expression, + } => super::list_folds::reduce( + to_df_value_expr(list), + to_df_value_expr(initial), + vec![accumulator.clone(), variable.clone()], + to_df_value_expr(expression), + ), + VE::ListPredicate { + kind, + variable, + list, + predicate, + } => super::list_folds::predicate( + kind, + to_df_value_expr(list), + variable.clone(), + to_df_value_expr(predicate), + ), VE::Literal(PV::String(s)) => lit(s.clone()), VE::Literal(PV::Integer(i)) => lit(*i), VE::Literal(PV::Float(f)) => lit(*f), @@ -197,7 +251,9 @@ pub(crate) fn to_df_value_expr(expr: &ValueExpression) -> Expr { super::math_functions::call(name, args.iter().map(to_df_value_expr).collect()) } "left" | "right" | "substring" | "replace" | "split" | "trim" | "ltrim" - | "rtrim" | "size" | "reverse" | "last" | "tail" => { + | "rtrim" | "size" | "reverse" | "last" | "tail" | "keys" | "__entity_keys" + | "labels" | "__predicate_all" | "__predicate_any" | "__predicate_none" + | "__predicate_single" => { super::scalar_functions::call(name, args.iter().map(to_df_value_expr).collect()) } "range" => { @@ -463,6 +519,27 @@ pub(crate) fn bind_value( schema: &datafusion::common::DFSchema, ) -> ValueExpression { let has = |name: &str| schema.has_column_with_unqualified_name(name); + let entity_properties = |var: &str| { + let var = var.to_lowercase(); + let prefix = format!("{var}__"); + if has(&var) { + return Vec::new(); + } + schema + .fields() + .iter() + .filter_map(|f| f.name().strip_prefix(&prefix)) + .map(|property| { + ( + property.to_string(), + ValueExpression::Property(crate::ast::PropertyRef { + variable: var.clone(), + property: property.to_string(), + }), + ) + }) + .collect::>() + }; let mut expr = expr.clone(); crate::ast::transform_value(&mut expr, &mut |e| match e { ValueExpression::Property(p) @@ -474,28 +551,23 @@ pub(crate) fn bind_value( field: p.property.clone(), }; } + ValueExpression::ScalarFunction { name, args } if name.eq_ignore_ascii_case("keys") => { + if let [ValueExpression::Variable(var)] = args.as_slice() { + let properties = entity_properties(var); + if !properties.is_empty() { + *e = ValueExpression::ScalarFunction { + name: "__entity_keys".to_string(), + args: vec![ValueExpression::Map(properties)], + }; + } + } + } ValueExpression::Map(entries) => { for (_, value) in entries.iter_mut() { let ValueExpression::Variable(var) = value else { continue; }; - let var = var.to_lowercase(); - let prefix = format!("{var}__"); - if has(&var) { - continue; - } - let properties: Vec<(String, ValueExpression)> = schema - .fields() - .iter() - .filter_map(|f| f.name().strip_prefix(&prefix)) - .map(|property| { - let column = ValueExpression::Property(crate::ast::PropertyRef { - variable: var.clone(), - property: property.to_string(), - }); - (property.to_string(), column) - }) - .collect(); + let properties = entity_properties(var); if !properties.is_empty() { *value = ValueExpression::Map(properties); } @@ -518,34 +590,13 @@ pub(crate) fn bind_boolean( /// Check if a ValueExpression contains an aggregate function pub(crate) fn contains_aggregate(expr: &ValueExpression) -> bool { - use crate::ast::ValueExpression as VE; - match expr { - VE::AggregateFunction { .. } => true, - VE::ScalarFunction { args, .. } => args.iter().any(contains_aggregate), - VE::Map(entries) => entries.iter().any(|(_, v)| contains_aggregate(v)), - VE::Field { base, .. } => contains_aggregate(base), - VE::Arithmetic { left, right, .. } => contains_aggregate(left) || contains_aggregate(right), - VE::Case { whens, otherwise } => { - whens.iter().any(|(_, then)| contains_aggregate(then)) - || otherwise.as_deref().is_some_and(contains_aggregate) - } - VE::VectorDistance { left, right, .. } => { - contains_aggregate(left) || contains_aggregate(right) - } - VE::VectorSimilarity { left, right, .. } => { - contains_aggregate(left) || contains_aggregate(right) - } - _ => false, - } + let mut found = false; + crate::ast::transform_value(&mut expr.clone(), &mut |value| { + found |= matches!(value, ValueExpression::AggregateFunction { .. }); + }); + found } -/// Convert a ValueExpression to Cypher dot notation for column naming -/// -/// This generates user-friendly column names following Cypher conventions: -/// - Property references: `p.name` (variable.property) -/// - Functions: `function_name(arg)` with simplified argument representation -/// - Other expressions: Use the expression as-is -/// /// This is used when no explicit alias is provided in RETURN clauses. pub(crate) fn to_cypher_column_name(expr: &ValueExpression) -> String { use crate::ast::ValueExpression as VE; diff --git a/crates/lance-graph/src/datafusion_planner/list_folds.rs b/crates/lance-graph/src/datafusion_planner/list_folds.rs new file mode 100644 index 0000000..9f3a8b7 --- /dev/null +++ b/crates/lance-graph/src/datafusion_planner/list_folds.rs @@ -0,0 +1,328 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The lance-csr Authors + +//! Cypher ordered folds and three-valued list predicates, using DataFusion lambdas. + +use std::sync::Arc; + +use arrow::array::{ + new_empty_array, new_null_array, Array, ArrayRef, AsArray, BooleanArray, BooleanBuilder, +}; +use arrow::compute::{cast, concat, take_arrays}; +use arrow_schema::{DataType, Field, FieldRef}; +use datafusion::common::utils::{adjust_offsets_for_slice, list_values, list_values_row_number}; +use datafusion::common::{exec_err, plan_err, Result}; +use datafusion::logical_expr::expr::{HigherOrderFunction, Lambda}; +use datafusion::logical_expr::{ + ColumnarValue, Expr, HigherOrderFunctionArgs, HigherOrderReturnFieldArgs, HigherOrderSignature, + HigherOrderUDF, HigherOrderUDFImpl, LambdaParametersProgress, ValueOrLambda, Volatility, +}; + +pub(super) fn reduce(list: Expr, initial: Expr, params: Vec, body: Expr) -> Expr { + Expr::HigherOrderFunction(HigherOrderFunction::new( + Arc::new(HigherOrderUDF::from(CypherFold::new("reduce"))), + vec![list, initial, Expr::Lambda(Lambda::new(params, body))], + )) +} + +pub(super) fn predicate(kind: &str, list: Expr, variable: String, body: Expr) -> Expr { + Expr::HigherOrderFunction(HigherOrderFunction::new( + Arc::new(HigherOrderUDF::from(CypherFold::new(kind))), + vec![list, Expr::Lambda(Lambda::new(vec![variable], body))], + )) +} + +#[derive(Debug, PartialEq, Eq, Hash)] +struct CypherFold { + name: String, + signature: HigherOrderSignature, +} + +impl CypherFold { + fn new(name: &str) -> Self { + let mut arguments = vec![ValueOrLambda::Value(())]; + if name == "reduce" { + arguments.push(ValueOrLambda::Value(())); + } + arguments.push(ValueOrLambda::Lambda(())); + Self { + name: name.to_string(), + signature: HigherOrderSignature::exact(arguments, Volatility::Immutable), + } + } +} + +fn list_type(data_type: &DataType) -> Result { + match data_type { + DataType::List(_) | DataType::LargeList(_) => Ok(data_type.clone()), + DataType::ListView(f) | DataType::FixedSizeList(f, _) => Ok(DataType::List(f.clone())), + DataType::LargeListView(f) => Ok(DataType::LargeList(f.clone())), + DataType::Null => Ok(DataType::new_list(DataType::Null, true)), + other => plan_err!("reduce/list predicate requires a list, got {other}"), + } +} + +impl HigherOrderUDFImpl for CypherFold { + fn name(&self) -> &str { + &self.name + } + fn signature(&self) -> &HigherOrderSignature { + &self.signature + } + + fn coerce_value_types(&self, types: &[DataType]) -> Result> { + let mut result = types.to_vec(); + result[0] = list_type(&types[0])?; + Ok(result) + } + + fn lambda_parameters( + &self, + step: usize, + fields: &[ValueOrLambda>], + ) -> Result { + let ValueOrLambda::Value(list) = &fields[0] else { + return plan_err!("expected list value"); + }; + let normalized = list_type(list.data_type())?; + let (DataType::List(element) | DataType::LargeList(element)) = normalized else { + unreachable!() + }; + if self.name != "reduce" { + return Ok(LambdaParametersProgress::Complete(vec![vec![element]])); + } + let ValueOrLambda::Value(initial) = &fields[1] else { + return plan_err!("expected initial value"); + }; + let ValueOrLambda::Lambda(body) = &fields[2] else { + return plan_err!("expected reduce body"); + }; + match body { + None => Ok(LambdaParametersProgress::Partial(vec![Some(vec![ + initial.clone(), + element, + ])])), + Some(body) => { + // One rebinding permits integer-to-float promotion and a null initializer. + if step > 2 { + return plan_err!("reduce accumulator type does not converge"); + } + Ok(LambdaParametersProgress::Complete(vec![vec![ + body.clone(), + element, + ]])) + } + } + } + + fn coerce_values_for_lambdas( + &self, + fields: &[ValueOrLambda], + ) -> Result>> { + if self.name != "reduce" { + return Ok(None); + } + let [ValueOrLambda::Value(list), ValueOrLambda::Value(initial), ValueOrLambda::Lambda(body)] = + fields + else { + return plan_err!("invalid reduce arguments"); + }; + if initial != body + && *initial != DataType::Null + && !(initial.is_numeric() && body.is_numeric()) + { + return plan_err!("reduce initial value ({initial}) and accumulator ({body}) must have compatible types"); + } + Ok(Some(vec![list.clone(), body.clone()])) + } + + fn return_field_from_args(&self, args: HigherOrderReturnFieldArgs) -> Result { + let Some(ValueOrLambda::Lambda(body)) = args.arg_fields.last() else { + return plan_err!("expected lambda body"); + }; + if self.name == "reduce" { + return Ok(Arc::new(body.as_ref().clone().with_nullable(true))); + } + if !matches!(body.data_type(), DataType::Boolean | DataType::Null) { + return plan_err!( + "{} predicate must return boolean, got {}", + self.name, + body.data_type() + ); + } + Ok(Arc::new(Field::new("", DataType::Boolean, true))) + } + + fn invoke_with_args(&self, args: HigherOrderFunctionArgs) -> Result { + let ValueOrLambda::Value(list) = &args.args[0] else { + return exec_err!("expected list value"); + }; + let Some(ValueOrLambda::Lambda(lambda)) = args.args.last() else { + return exec_err!("expected lambda body"); + }; + let list = list.to_array(args.number_rows)?; + let ranges: Vec<(usize, usize)> = match list.data_type() { + DataType::List(_) => list + .as_list::() + .value_offsets() + .windows(2) + .map(|w| (w[0] as usize, w[1] as usize)) + .collect(), + DataType::LargeList(_) => list + .as_list::() + .value_offsets() + .windows(2) + .map(|w| (w[0] as usize, w[1] as usize)) + .collect(), + other => return exec_err!("expected list, got {other}"), + }; + if self.name == "reduce" { + let ValueOrLambda::Value(initial) = &args.args[1] else { + return exec_err!("expected initial value"); + }; + let initial = cast(&initial.to_array(args.number_rows)?, args.return_type())?; + let values = match list.data_type() { + DataType::List(_) => list.as_list::().values(), + DataType::LargeList(_) => list.as_list::().values(), + _ => unreachable!(), + }; + let mut output: Vec = Vec::with_capacity(list.len()); + for (row, (start, end)) in ranges.into_iter().enumerate() { + if list.is_null(row) { + output.push(new_null_array(args.return_type(), 1)); + continue; + } + let mut accumulator = initial.slice(row, 1); + // Each step sees the previous step's result, in list order. + for index in start..end { + let state = || Ok(accumulator.clone()); + let element = || Ok(values.slice(index, 1)); + accumulator = lambda + .evaluate(&[&state, &element], |captures| { + Ok(captures.iter().map(|a| a.slice(row, 1)).collect()) + })? + .into_array(1)?; + if accumulator.data_type() != args.return_type() { + return exec_err!( + "reduce accumulator type changed from {} to {}", + args.return_type(), + accumulator.data_type() + ); + } + } + output.push(accumulator); + } + let result = if output.is_empty() { + new_empty_array(args.return_type()) + } else { + concat(&output.iter().map(|a| a.as_ref()).collect::>())? + }; + return Ok(ColumnarValue::Array(result)); + } + let values = list_values(&list)?; + let predicate = if values.is_empty() { + new_empty_array(&DataType::Boolean) + } else { + let parameter = || Ok(values.clone()); + cast( + &lambda + .evaluate(&[¶meter], |captures| { + Ok(take_arrays( + captures, + &list_values_row_number(&list)?, + None, + )?) + })? + .into_array(values.len())?, + &DataType::Boolean, + )? + }; + let predicate = predicate.as_boolean(); + let ranges: Vec<(usize, usize)> = match list.data_type() { + DataType::List(_) => adjust_offsets_for_slice(list.as_list::()) + .windows(2) + .map(|w| (w[0] as usize, w[1] as usize)) + .collect(), + DataType::LargeList(_) => adjust_offsets_for_slice(list.as_list::()) + .windows(2) + .map(|w| (w[0] as usize, w[1] as usize)) + .collect(), + _ => unreachable!(), + }; + let mut output = BooleanBuilder::with_capacity(list.len()); + for (row, (start, end)) in ranges.into_iter().enumerate() { + output.append_option(if list.is_null(row) { + None + } else { + predicate_result(&self.name, predicate, start, end) + }); + } + Ok(ColumnarValue::Array(Arc::new(output.finish()))) + } +} + +pub(super) fn predicate_result( + kind: &str, + values: &BooleanArray, + start: usize, + end: usize, +) -> Option { + let mut trues = 0; + let mut falses = 0; + let mut nulls = 0; + for i in start..end { + if values.is_null(i) { + nulls += 1; + } else if values.value(i) { + trues += 1; + } else { + falses += 1; + } + } + match kind { + "all" if falses > 0 => Some(false), + "any" if trues > 0 => Some(true), + "none" if trues > 0 => Some(false), + "single" if trues > 1 => Some(false), + _ if nulls > 0 => None, + "all" | "none" => Some(true), + "any" => Some(false), + "single" => Some(trues == 1), + _ => unreachable!(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn list_predicates_use_three_valued_logic() { + for (values, expected) in [ + (vec![], [Some(true), Some(false), Some(true), Some(false)]), + (vec![None], [None, None, None, None]), + ( + vec![Some(true), None], + [None, Some(true), Some(false), None], + ), + (vec![Some(false), None], [Some(false), None, None, None]), + ( + vec![Some(true), Some(true), None], + [None, Some(true), Some(false), Some(false)], + ), + ( + vec![Some(true), Some(false)], + [Some(false), Some(true), Some(false), Some(true)], + ), + ] { + let values = BooleanArray::from(values); + for (kind, want) in ["all", "any", "none", "single"].iter().zip(expected) { + assert_eq!( + predicate_result(kind, &values, 0, values.len()), + want, + "{kind}: {values:?}" + ); + } + } + } +} diff --git a/crates/lance-graph/src/datafusion_planner/mod.rs b/crates/lance-graph/src/datafusion_planner/mod.rs index c5f9caf..a11afea 100644 --- a/crates/lance-graph/src/datafusion_planner/mod.rs +++ b/crates/lance-graph/src/datafusion_planner/mod.rs @@ -20,6 +20,7 @@ mod config_helpers; mod cypher_udf; mod expression; mod join_ops; +mod list_folds; mod math_functions; mod scalar_functions; mod scan_ops; diff --git a/crates/lance-graph/src/datafusion_planner/scalar_functions.rs b/crates/lance-graph/src/datafusion_planner/scalar_functions.rs index fd9511a..dbf283c 100644 --- a/crates/lance-graph/src/datafusion_planner/scalar_functions.rs +++ b/crates/lance-graph/src/datafusion_planner/scalar_functions.rs @@ -48,6 +48,11 @@ impl ScalarUDFImpl for CypherScalar { for (i, t) in types.iter().enumerate() { let valid = match self.name.as_str() { "range" => t.is_integer(), + "__predicate_all" | "__predicate_any" | "__predicate_none" + | "__predicate_single" => list_field(t) + .is_some_and(|f| matches!(f.data_type(), DataType::Boolean | DataType::Null)), + "keys" | "__entity_keys" => matches!(t, DataType::Struct(_)), + "labels" => false, "size" | "reverse" => is_string(t) || list_field(t).is_some(), "last" | "tail" => list_field(t).is_some(), "left" | "right" | "substring" if i > 0 => t.is_integer(), @@ -58,9 +63,14 @@ impl ScalarUDFImpl for CypherScalar { } } Ok(match self.name.as_str() { + "__predicate_all" | "__predicate_any" | "__predicate_none" | "__predicate_single" => { + DataType::Boolean + } "size" => DataType::Int64, "range" => DataType::List(Arc::new(Field::new("item", DataType::Int64, true))), - "split" => DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))), + "split" | "keys" | "__entity_keys" | "labels" => { + DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))) + } "last" => list_field(&types[0]) .map(|f| f.data_type().clone()) .unwrap_or(DataType::Null), @@ -81,6 +91,55 @@ impl ScalarUDFImpl for CypherScalar { args.number_rows, ))); } + if let Some(kind) = self.name.strip_prefix("__predicate_") { + let input = args.args[0].to_array(args.number_rows)?; + let mut result = arrow::array::BooleanBuilder::with_capacity(input.len()); + for row in 0..input.len() { + if input.is_null(row) { + result.append_null(); + continue; + } + let values = match input.data_type() { + DataType::List(_) => input.as_list::().value(row), + DataType::LargeList(_) => input.as_list::().value(row), + DataType::FixedSizeList(_, _) => input.as_fixed_size_list().value(row), + _ => unreachable!("validated list predicate input"), + }; + let values = cast(&values, &DataType::Boolean)?; + result.append_option(super::list_folds::predicate_result( + kind, + values.as_boolean(), + 0, + values.len(), + )); + } + return Ok(ColumnarValue::Array(Arc::new(result.finish()))); + } + if matches!(self.name.as_str(), "keys" | "__entity_keys") { + let input = args.args[0].to_array(args.number_rows)?; + let map = input.as_struct(); + let nulls: Vec<_> = map + .columns() + .iter() + .map(|column| column.logical_nulls()) + .collect(); + let entity = self.name == "__entity_keys"; + let mut result = ListBuilder::new(StringBuilder::new()); + for row in 0..input.len() { + let missing = |i: usize| nulls[i].as_ref().is_some_and(|nulls| nulls.is_null(row)); + if input.is_null(row) || (entity && (0..map.num_columns()).all(missing)) { + result.append(false); + continue; + } + for (i, field) in map.fields().iter().enumerate() { + if !entity || !missing(i) { + result.values().append_value(field.name()); + } + } + result.append(true); + } + return Ok(ColumnarValue::Array(Arc::new(result.finish()))); + } if self.name == "range" { for arg in &mut args.args { *arg = ColumnarValue::Array(cast_with_options( @@ -345,6 +404,37 @@ mod tests { .await } + #[tokio::test] + async fn cypher_entity_map_keys() { + for (expression, expected) in [ + ("keys({x: null, y: 1})", vec!["x", "y"]), + ("keys({})", vec![]), + ] { + let out = run(expression).await.unwrap(); + let keys = out.column(0).as_list::().value(0); + assert_eq!( + keys.as_string::() + .iter() + .map(Option::unwrap) + .collect::>(), + expected + ); + } + for expression in ["keys(null)", "labels(null)"] { + assert_eq!( + run(expression) + .await + .unwrap() + .column(0) + .logical_null_count(), + 1 + ); + } + for expression in ["keys(1)", "labels(1)"] { + assert!(run(expression).await.is_err()); + } + } + #[tokio::test] async fn cypher_scalar_unicode_and_slicing() { for (expression, expected) in [ diff --git a/crates/lance-graph/src/logical_plan.rs b/crates/lance-graph/src/logical_plan.rs index 4359fc0..533eaba 100644 --- a/crates/lance-graph/src/logical_plan.rs +++ b/crates/lance-graph/src/logical_plan.rs @@ -470,6 +470,39 @@ impl<'a> LogicalPlanner<'a> { }); } } + ValueExpression::ScalarFunction { name, args } + if name.eq_ignore_ascii_case("labels") => + { + if let ValueExpression::Variable(var) = &args[0] { + if let Some(mapping) = self + .variables + .get(var) + .and_then(|label| self.config.get_node_mapping(label)) + { + *expr = ValueExpression::Case { + whens: vec![( + BooleanExpression::IsNull(ValueExpression::Property(PropertyRef { + variable: var.clone(), + property: mapping.id_field.clone(), + })), + ValueExpression::Literal(PropertyValue::Null), + )], + otherwise: Some(Box::new(ValueExpression::ScalarFunction { + name: "__list".to_string(), + args: vec![ValueExpression::Literal(PropertyValue::String( + mapping.label.clone(), + ))], + })), + }; + return; + } + } + args.iter_mut().for_each(|arg| self.resolve_value(arg)); + } + // keys(node/relationship) binds the entity's property columns, not its ID. + ValueExpression::ScalarFunction { name, args } + if name.eq_ignore_ascii_case("keys") + && matches!(args.as_slice(), [ValueExpression::Variable(_)]) => {} // Path functions take the path variable itself. ValueExpression::ScalarFunction { name, args } if matches!(name.to_lowercase().as_str(), "length" | "nodes") => {} @@ -490,6 +523,22 @@ impl<'a> LogicalPlanner<'a> { self.resolve_value(e); } } + ValueExpression::Reduce { + initial, + list, + expression, + .. + } => { + self.resolve_value(initial); + self.resolve_value(list); + self.resolve_value(expression); + } + ValueExpression::ListPredicate { + list, predicate, .. + } => { + self.resolve_value(list); + self.resolve_value(predicate); + } ValueExpression::Boolean(b) => self.resolve_boolean(b), // A node inside a map keeps its properties (see `bind_value`), not just its id. ValueExpression::Map(entries) => entries @@ -744,6 +793,15 @@ impl<'a> LogicalPlanner<'a> { expr: &mut ValueExpression, ) -> Result { match expr { + ValueExpression::ScalarFunction { name, args } + if name.eq_ignore_ascii_case("labels") + && matches!(args.as_slice(), [ValueExpression::Variable(var)] if self.variables.get(var).is_some_and(|label| label.contains('|'))) => + { + Err(GraphError::UnsupportedFeature { + feature: "labels() on a mixed-label binding is not supported".to_string(), + location: Location::new(file!(), line!(), column!()), + }) + } ValueExpression::Case { whens, otherwise } => { for (condition, then) in whens.iter_mut() { input = self.rewrite_exists(input, condition)?; @@ -765,6 +823,80 @@ impl<'a> LogicalPlanner<'a> { input = self.rewrite_exists_in_value(input, left)?; self.rewrite_exists_in_value(input, right) } + ValueExpression::Reduce { + initial, + list, + expression, + .. + } => { + input = self.rewrite_exists_in_value(input, initial)?; + input = self.rewrite_exists_in_value(input, list)?; + self.rewrite_exists_in_value(input, expression) + } + ValueExpression::ListPredicate { + kind, + variable, + list, + predicate, + } if crate::ast::needs_relational_evaluation(predicate) => { + input = self.rewrite_exists_in_value(input, list)?; + let original_list = (**list).clone(); + let mut map = (**predicate).clone(); + // The relational comprehension supplies this iterator as a column. Restore + // inner lambda bindings after replacing references to the outer iterator. + crate::ast::transform_value(&mut map, &mut |value| match value { + ValueExpression::LocalVariable(name) if name == variable => { + *value = ValueExpression::Variable(name.clone()); + } + ValueExpression::Field { base, field } if matches!(base.as_ref(), ValueExpression::Variable(name) if name == variable) => + { + *value = ValueExpression::Property(PropertyRef::new( + variable.clone(), + field.clone(), + )); + } + ValueExpression::Reduce { + accumulator, + variable, + expression, + .. + } => { + crate::ast::mark_locals(expression, &[accumulator, variable]); + } + ValueExpression::ListPredicate { + variable, + predicate, + .. + } => { + crate::ast::mark_locals(predicate, &[variable]); + } + _ => {} + }); + let mut comprehension = ValueExpression::ListComprehension { + variable: variable.clone(), + list: list.clone(), + filter: None, + map: Some(Box::new(map)), + }; + input = self.rewrite_exists_in_value(input, &mut comprehension)?; + *expr = ValueExpression::Case { + whens: vec![( + BooleanExpression::IsNull(original_list), + ValueExpression::Literal(PropertyValue::Null), + )], + otherwise: Some(Box::new(ValueExpression::ScalarFunction { + name: format!("__predicate_{kind}"), + args: vec![comprehension], + })), + }; + Ok(input) + } + ValueExpression::ListPredicate { + list, predicate, .. + } => { + input = self.rewrite_exists_in_value(input, list)?; + self.rewrite_exists_in_value(input, predicate) + } ValueExpression::Boolean(b) => self.rewrite_exists(input, b), ValueExpression::Map(entries) => { for (_, v) in entries.iter_mut() { @@ -1007,9 +1139,6 @@ impl<'a> LogicalPlanner<'a> { } } } - if shared.is_empty() { - return Err(unsupported("without a variable bound by an earlier clause")); - } let mut keys = Vec::with_capacity(shared.len()); for var in &shared { let label = &self.variables[var]; @@ -1026,12 +1155,29 @@ impl<'a> LogicalPlanner<'a> { let local = crate::pattern_order::pattern_variables(&clause.patterns); let mut scalars: Vec = names .into_iter() - .filter(|n| !self.variables.contains_key(*n) && !local.contains(*n)) + .filter(|n| !local.contains(*n) && !shared.iter().any(|v| v == *n)) .map(str::to_string) .collect(); scalars.sort(); + for variable in &scalars { + // Binding variables distinguish iterator contexts. Untracked aggregate + // captures can be re-evaluated in a different order by the seed plan. + if !self.variables.contains_key(variable) && !shared.is_empty() { + continue; + } + keys.push( + self.variables + .get(variable) + .and_then(|label| self.config.get_node_mapping(label)) + .map(|mapping| qualify_column(variable, &mapping.id_field)) + .unwrap_or_else(|| variable.to_lowercase()), + ); + } carried.extend(scalars); } + if carried.is_empty() { + return Err(unsupported("without a variable bound by an earlier clause")); + } let seed = LogicalOperator::Distinct { input: Box::new(LogicalOperator::Project { input: Box::new(seed_source.unwrap_or_else(|| base.clone())), diff --git a/crates/lance-graph/src/parameter_substitution.rs b/crates/lance-graph/src/parameter_substitution.rs index d0a0df4..38587ef 100644 --- a/crates/lance-graph/src/parameter_substitution.rs +++ b/crates/lance-graph/src/parameter_substitution.rs @@ -299,6 +299,22 @@ fn substitute_in_value_expression( } substitute_in_value_expression(map, parameters)?; } + ValueExpression::Reduce { + initial, + list, + expression, + .. + } => { + substitute_in_value_expression(initial, parameters)?; + substitute_in_value_expression(list, parameters)?; + substitute_in_value_expression(expression, parameters)?; + } + ValueExpression::ListPredicate { + list, predicate, .. + } => { + substitute_in_value_expression(list, parameters)?; + substitute_in_value_expression(predicate, parameters)?; + } ValueExpression::ListComprehension { list, filter, map, .. } => { diff --git a/crates/lance-graph/src/parser.rs b/crates/lance-graph/src/parser.rs index 260240b..435eb26 100644 --- a/crates/lance-graph/src/parser.rs +++ b/crates/lance-graph/src/parser.rs @@ -492,6 +492,11 @@ fn boolean_primary_expression(input: &str) -> IResult<&str, BooleanExpression> { where_clause: None, }), comparison_expression, + map(list_predicate, |left| BooleanExpression::Comparison { + left, + operator: ComparisonOperator::Equal, + right: ValueExpression::Literal(PropertyValue::Boolean(true)), + }), ))(input) } @@ -669,6 +674,8 @@ fn basic_value_expression(input: &str) -> IResult<&str, ValueExpression> { list_comprehension, list_literal, // Then a list of any expressions [a.x, 'y'] (or []) map_literal, + reduce_expression, + list_predicate, parse_parameter, // Try $parameter function_call, // Regular function calls map(property_value, ValueExpression::Literal), // Try literals BEFORE property references @@ -733,6 +740,62 @@ fn list_comprehension(input: &str) -> IResult<&str, ValueExpression> { )) } +fn reduce_expression(input: &str) -> IResult<&str, ValueExpression> { + let (input, _) = tuple((tag_no_case("reduce"), multispace0, char('('), multispace0))(input)?; + let (input, accumulator) = identifier(input)?; + let (input, initial) = + preceded(tuple((multispace0, char('='), multispace0)), expression)(input)?; + let (input, variable) = preceded(comma_ws, identifier)(input)?; + let (input, list) = preceded( + tuple((multispace1, tag_no_case("IN"), multispace1)), + value_expression, + )(input)?; + let (input, mut body) = + preceded(tuple((multispace0, char('|'), multispace0)), expression)(input)?; + let (input, _) = tuple((multispace0, char(')')))(input)?; + mark_locals(&mut body, &[accumulator, variable]); + Ok(( + input, + ValueExpression::Reduce { + accumulator: accumulator.to_lowercase(), + initial: Box::new(initial), + variable: variable.to_lowercase(), + list: Box::new(list), + expression: Box::new(body), + }, + )) +} + +fn list_predicate(input: &str) -> IResult<&str, ValueExpression> { + let (input, kind) = alt(( + tag_no_case("all"), + tag_no_case("any"), + tag_no_case("none"), + tag_no_case("single"), + ))(input)?; + let (input, variable) = + preceded(tuple((multispace0, char('('), multispace0)), identifier)(input)?; + let (input, list) = preceded( + tuple((multispace1, tag_no_case("IN"), multispace1)), + value_expression, + )(input)?; + let (input, mut predicate) = preceded( + tuple((multispace1, tag_no_case("WHERE"), multispace1)), + expression, + )(input)?; + let (input, _) = tuple((multispace0, char(')')))(input)?; + mark_locals(&mut predicate, &[variable]); + Ok(( + input, + ValueExpression::ListPredicate { + kind: kind.to_lowercase(), + variable: variable.to_lowercase(), + list: Box::new(list), + predicate: Box::new(predicate), + }, + )) +} + // `{key: value, ...}` fn map_literal(input: &str) -> IResult<&str, ValueExpression> { let entry = tuple(( @@ -2529,4 +2592,31 @@ mod tests { ValueExpression::Variable("nullCount".into()) ); } + #[test] + fn reduce_keeps_inputs_in_outer_scope_and_marks_local_properties() { + let q = + parse_cypher_query("WITH {n:1} AS x RETURN reduce(s=x.n, x IN [x] | s+x.n)").unwrap(); + let ValueExpression::Reduce { + initial, + list, + expression, + .. + } = &q.return_clause.items[0].expression + else { + panic!("expected reduce"); + }; + assert!(matches!(initial.as_ref(), ValueExpression::Property(p) if p.variable == "x")); + let mut input_locals = 0; + transform_value(&mut list.as_ref().clone(), &mut |v| { + input_locals += usize::from(matches!(v, ValueExpression::LocalVariable(_))) + }); + assert_eq!(input_locals, 0); + let mut body_locals = Vec::new(); + transform_value(&mut expression.as_ref().clone(), &mut |v| { + if let ValueExpression::LocalVariable(n) = v { + body_locals.push(n.clone()); + } + }); + assert_eq!(body_locals, ["s", "x"]); + } } diff --git a/crates/lance-graph/src/pattern_order.rs b/crates/lance-graph/src/pattern_order.rs index b5333e4..2d4aade 100644 --- a/crates/lance-graph/src/pattern_order.rs +++ b/crates/lance-graph/src/pattern_order.rs @@ -258,7 +258,24 @@ fn value_variables<'a>(expr: &'a ValueExpression, out: &mut HashSet<&'a str>) { value_variables(m, out); } } - ValueExpression::Literal(_) + ValueExpression::Reduce { + initial, + list, + expression, + .. + } => { + value_variables(initial, out); + value_variables(list, out); + value_variables(expression, out); + } + ValueExpression::ListPredicate { + list, predicate, .. + } => { + value_variables(list, out); + value_variables(predicate, out); + } + ValueExpression::LocalVariable(_) + | ValueExpression::Literal(_) | ValueExpression::Parameter(_) | ValueExpression::VectorLiteral(_) => {} } diff --git a/crates/lance-graph/src/semantic.rs b/crates/lance-graph/src/semantic.rs index ac30dff..bb2d176 100644 --- a/crates/lance-graph/src/semantic.rs +++ b/crates/lance-graph/src/semantic.rs @@ -425,7 +425,7 @@ impl SemanticAnalyzer { ValueExpression::Literal(_) => { // Literals are always valid } - ValueExpression::Variable(var) => { + ValueExpression::Variable(var) | ValueExpression::LocalVariable(var) => { // Use case-insensitive lookup if !self.variables.contains_key_ci(var) { return Err(GraphError::PlanError { @@ -440,7 +440,7 @@ impl SemanticAnalyzer { match function_name.as_str() { "tolower" | "lower" | "toupper" | "upper" | "abs" | "round" | "floor" | "ceil" | "sqrt" | "trim" | "ltrim" | "rtrim" | "reverse" | "last" - | "tail" => { + | "tail" | "keys" | "labels" => { if args.len() != 1 { return Err(GraphError::PlanError { message: format!( @@ -521,7 +521,7 @@ impl SemanticAnalyzer { // Unknown scalar function - reject early with helpful error return Err(GraphError::UnsupportedFeature { feature: format!( - "Cypher function '{}' is not implemented. Supported scalar functions: toLower, lower, toUpper, upper, toString, coalesce, date_part, to_epoch_ms, length, nodes, size, range, head, last, tail, reverse, left, right, substring, replace, split, trim, lTrim, rTrim, abs, round, floor, ceil, sqrt, toInteger, toFloat, date, datetime, duration.inSeconds. Supported aggregate functions: COUNT, SUM, AVG, MIN, MAX, COLLECT.", + "Cypher function '{}' is not implemented. Supported scalar functions: toLower, lower, toUpper, upper, toString, coalesce, date_part, to_epoch_ms, length, nodes, size, range, head, last, tail, keys, labels, reverse, left, right, substring, replace, split, trim, lTrim, rTrim, abs, round, floor, ceil, sqrt, toInteger, toFloat, date, datetime, duration.inSeconds. Supported aggregate functions: COUNT, SUM, AVG, MIN, MAX, COLLECT.", name ), location: snafu::Location::new(file!(), line!(), column!()), @@ -529,6 +529,34 @@ impl SemanticAnalyzer { } } + if function_name == "labels" { + if let ValueExpression::Variable(var) = &args[0] { + if let Some(info) = self.variables.get_ci(var) { + if info + .labels + .iter() + .map(|label| label.to_lowercase()) + .collect::>() + .len() + > 1 + { + return Err(GraphError::UnsupportedFeature { + feature: "labels() on a mixed-label binding is not supported" + .to_string(), + location: snafu::Location::new(file!(), line!(), column!()), + }); + } + if info.variable_type == VariableType::Relationship { + return Err(GraphError::PlanError { + message: "labels() requires a node, not a relationship" + .to_string(), + location: snafu::Location::new(file!(), line!(), column!()), + }); + } + } + } + } + // Validate arguments recursively for arg in args { self.analyze_value_expression(arg)?; @@ -730,6 +758,32 @@ impl SemanticAnalyzer { self.variables = before; result?; } + ValueExpression::Reduce { + accumulator, + initial, + variable, + list, + expression, + } => { + self.analyze_value_expression(initial)?; + self.analyze_value_expression(list)?; + if accumulator.eq_ignore_ascii_case(variable) { + return Err(GraphError::PlanError { + message: "reduce accumulator and element variables must differ".to_string(), + location: snafu::Location::new(file!(), line!(), column!()), + }); + } + self.analyze_lambda(&[accumulator, variable], expression, false)?; + } + ValueExpression::ListPredicate { + variable, + list, + predicate, + .. + } => { + self.analyze_value_expression(list)?; + self.analyze_lambda(&[variable], predicate, true)?; + } ValueExpression::ListComprehension { variable, list, @@ -756,6 +810,34 @@ impl SemanticAnalyzer { Ok(()) } + fn analyze_lambda( + &mut self, + names: &[&String], + expression: &ValueExpression, + allow_patterns: bool, + ) -> Result<()> { + let before = self.variables.clone(); + for name in names { + self.variables.remove(&name.to_lowercase()); + self.register_projection_alias(name); + } + let mut unsupported = + !allow_patterns && crate::ast::needs_relational_evaluation(expression); + crate::ast::transform_value(&mut expression.clone(), &mut |value| { + unsupported |= matches!(value, ValueExpression::AggregateFunction { .. }); + }); + let result = if unsupported { + Err(GraphError::UnsupportedFeature { + feature: "graph expressions/comprehensions inside a reduce body, or aggregates inside a lambda body; compute the list before the body".to_string(), + location: snafu::Location::new(file!(), line!(), column!()), + }) + } else { + self.analyze_value_expression(expression) + }; + self.variables = before; + result + } + fn register_projection_alias(&mut self, alias: &str) { // Use case-insensitive lookup and store normalized key if self.variables.contains_key_ci(alias) { diff --git a/docs/cypher.md b/docs/cypher.md index ff68643..e3aa541 100644 --- a/docs/cypher.md +++ b/docs/cypher.md @@ -179,7 +179,8 @@ implemented; `ExecutionStrategy.LanceNative` is reserved and raises an error. The Graph Data Council (GDC), formerly known as the Linked Data Benchmark Council (LDBC), maintains the [GDC SNB Interactive](https://ldbcouncil.org/benchmarks/snb/) complex queries. lance-csr implements a large -read-only subset of openCypher, enough to run all 14 (IC14 without `reduce()`). +read-only subset of openCypher. The tables below specify its supported surface; relationship-path +operations needed by the unchanged IC14 query are not yet supported. ### Clauses @@ -260,7 +261,9 @@ RETURN length(p), nodes(p) | Aggregates | `count` (including `count(*)` and `count(DISTINCT x)`), `sum`, `avg`, `min`, `max`, `collect` (including `DISTINCT`; skips nulls) | | Strings | `toLower` / `lower`, `toUpper` / `upper`, `toString`, `left`, `right`, `substring`, `replace`, `split`, `trim`, `lTrim`, `rTrim`, `reverse`, `size` | | Conversion | `toInteger`, `toFloat`, `coalesce` | -| Lists | `size`, `head`, `last`, `tail`, `reverse`, `range(start, end [, step])` (inclusive) | +| Properties | `keys(map)`, `keys(node)`, `keys(relationship)`, `labels(node)` | +| Lists | `size`, `head`, `last`, `tail`, `reverse`, `range(start, end [, step])` (inclusive), `reduce` | +| List predicates | `all`, `any`, `none`, `single` | | Math | `abs`, `round`, `floor`, `ceil`, `sqrt` | | Temporal | `date`, `datetime`, `date_part('month', d)`, `to_epoch_ms`, `duration.inSeconds(a, b).seconds` | | Paths | `length(p)`, `nodes(p)` on `shortestPath` variables | @@ -303,9 +306,62 @@ result = engine.execute("RETURN abs(-5) AS magnitude, round(-1.5) AS rounded, sq print(result.to_pylist()) # [{'magnitude': 5, 'rounded': -1.0, 'root': 3.0}] ``` +`keys` returns the field names in a map, including keys with null values. For a node or relationship, +it returns its exposed, non-null property names, including the ID or endpoint columns already exposed +by the engine. Property names use the engine's normalized column spelling. Key order is not guaranteed. +`keys({})` returns an empty list. `labels(node)` returns the canonical label configured for that node, +even if the query spells it differently. Both functions preserve aliases carried through `WITH` and +return null for a missing entity from `OPTIONAL MATCH`. + +!!! note "Labels on mixed-label bindings" + + `labels(p)` supports a binding with one configured label. A pattern such as + `(p:Person:City)` is rejected with a mixed-label error, including after `WITH p AS x`. + The engine does not retain each row's label after this union. It never guesses a label from an ID + that might occur in both tables. + +```python +result = engine.execute(""" + MATCH (p:Person {name: 'Alice'}) WITH p AS person + RETURN labels(person) AS labels, size(keys(person)) AS properties +""") +print(result.to_pylist()) # [{'labels': ['Person'], 'properties': 3}] +``` + Any other function raises `ValueError` with `Cypher function 'name' is not implemented`. This is a -bounded function set, not a claim of full Cypher conformance. Remaining functions include `keys`, -`labels`, `reduce`, the list predicates `any` / `all` / `none` / `single`, and `relationships(p)`. +bounded function set, not a claim of full Cypher conformance. `relationships(p)` is not supported. + +### Ordered reduce and list predicates + +`reduce(total = initial, item IN list | expression)` evaluates the expression once per element in +list order. Each step sees the previous accumulator value. An empty list returns the initial value; +a null list returns null. Null elements are passed to the expression, so use `coalesce` if you want +to replace them. The accumulator and iterator names are local to the body: the initial value and +list still refer to outer variables. The accumulator must keep a compatible Arrow type; integer +initial values can promote to floating point when the expression requires it. + +```python +result = engine.execute(""" + RETURN reduce(total = 0, item IN [1, 2, 3] | total * 10 + item) AS ordered, + all(item IN [1, 2, 3] WHERE item > 0) AS positive, + single(item IN [1, 2, 3] WHERE item = 2) AS exactly_one +""") +print(result.to_pylist()) # [{'ordered': 123, 'positive': True, 'exactly_one': True}] +``` + +The list predicates use `function(item IN list WHERE predicate)` syntax. `all` requires every +predicate result to be true, `any` requires at least one, `none` requires no true results, and +`single` requires exactly one. For an empty list, `all` and `none` are true; `any` and `single` are +false. A null list returns null. A null predicate result makes the answer null unless a known result +already decides it: false decides `all`, true decides `any` and `none`, and two true results decide +`single`. Predicates can inspect null elements explicitly with `item IS NULL`. + +Both forms can capture outer values and nest scalar folds or predicates. List predicates also +support correlated graph predicates such as `EXISTS { MATCH ... }`. A reduce input can be a list +comprehension, pattern comprehension, or `collect(...)`; its body supports scalar expressions but +rejects graph patterns, comprehensions, and aggregation. Compute such a list before the body. +These semantics follow the pinned [Cypher 4.4 list reference](https://neo4j.com/docs/cypher-manual/4.4/functions/list/#functions-reduce) +and [predicate reference](https://neo4j.com/docs/cypher-manual/4.4/functions/predicate/). ### Example: a little of everything diff --git a/python/python/tests/test_entity_functions.py b/python/python/tests/test_entity_functions.py new file mode 100644 index 0000000..e891b26 --- /dev/null +++ b/python/python/tests/test_entity_functions.py @@ -0,0 +1,137 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The lance-csr Authors + +"""Entity functions retain graph binding information and map null semantics.""" + +import pyarrow as pa +import pytest +from lance_csr import CypherEngine, GraphConfig + + +@pytest.fixture(params=[False, True], ids=["join", "csr"]) +def engine(request): + config = ( + GraphConfig.builder() + .node_table("Person", "person_id") + .node_table("City", "person_id") + .rel_table("KNOWS", "Person", "Person", "src", "dst") + .rel_table("VISITS", "Person", "City", "src", "dst") + .build() + ) + return CypherEngine( + config, + { + "Person": pa.table( + { + "person_id": [1, 2, 3], + "name": ["Alice", None, "Carol"], + "age": [30, None, 0], + "missing": pa.nulls(3), + } + ), + "City": pa.table({"person_id": [1], "name": ["London"]}), + "KNOWS": pa.table({"src": [1, 2], "dst": [2, 3], "since": [2020, None]}), + "VISITS": pa.table({"src": [1], "dst": [1]}), + }, + use_csr=request.param, + ) + + +def test_node_keys_labels_and_with_alias(engine): + result = engine.execute( + "MATCH (p:person) WITH p AS person " + "RETURN person.person_id AS id, keys(person) AS keys, labels(person) AS labels " + "ORDER BY id" + ).to_pylist() + assert [sorted(row["keys"]) for row in result] == [ + ["age", "name", "person_id"], + ["person_id"], + ["age", "name", "person_id"], + ] + assert [row["labels"] for row in result] == [["Person"]] * 3 + + +def test_optional_node_and_relationship_keys(engine): + result = engine.execute( + "MATCH (p:Person) OPTIONAL MATCH (p)-[r:KNOWS]->(q:Person) " + "RETURN p.person_id AS id, keys(q) AS node_keys, labels(q) AS labels, " + "keys(r) AS edge_keys ORDER BY id" + ).to_pylist() + for row in result: + for key in ("node_keys", "edge_keys"): + if row[key] is not None: + row[key] = sorted(row[key]) + assert result == [ + { + "id": 1, + "node_keys": ["person_id"], + "labels": ["Person"], + "edge_keys": ["dst", "since", "src"], + }, + { + "id": 2, + "node_keys": ["age", "name", "person_id"], + "labels": ["Person"], + "edge_keys": ["dst", "src"], + }, + {"id": 3, "node_keys": None, "labels": None, "edge_keys": None}, + ] + + +def test_relationship_keys_with_alias(engine): + result = engine.execute( + "MATCH (:Person)-[r:KNOWS]->(:Person) WITH r AS edge " + "RETURN edge.src AS src, keys(edge) AS keys ORDER BY src" + ).to_pylist() + assert [sorted(row["keys"]) for row in result] == [ + ["dst", "since", "src"], + ["dst", "src"], + ] + + +@pytest.mark.parametrize("value", [{}, {"MixedKey": None, "name": "Alice"}, None]) +def test_map_keys_keep_null_values(engine, value): + result = engine.execute( + "WITH $value AS map RETURN keys(map) AS keys", {"value": value} + ) + keys = result.to_pylist()[0]["keys"] + if value is None: + assert keys is None + else: + assert sorted(keys) == sorted(value) + + +def test_labels_null_and_keys_literal(engine): + result = engine.execute( + "WITH null AS value RETURN labels(value) AS labels, keys({x: null}) AS keys" + ).to_pylist() + assert result == [{"labels": None, "keys": ["x"]}] + + +@pytest.mark.parametrize( + "query", + [ + "MATCH (p:Person:City) RETURN labels(p) AS labels", + "MATCH (p:Person:City) WITH p AS x RETURN labels(x) AS labels", + "MATCH (:Person)-[:KNOWS|VISITS]->(p) RETURN labels(p) AS labels", + ], +) +def test_labels_reject_mixed_label_bindings(engine, query): + with pytest.raises((ValueError, NotImplementedError), match="mixed-label binding"): + engine.execute(query) + + +@pytest.mark.parametrize( + "query", + [ + "RETURN keys(1) AS keys", + "RETURN labels(1) AS labels", + "RETURN labels({x: 1}) AS labels", + "MATCH (:Person)-[r:KNOWS]->(:Person) RETURN labels(r) AS labels", + "RETURN keys() AS keys", + "RETURN labels(null, null) AS labels", + ], +) +def test_entity_function_invalid_arguments(engine, query): + with pytest.raises((ValueError, RuntimeError)): + engine.execute(query) diff --git a/python/python/tests/test_list_folds.py b/python/python/tests/test_list_folds.py new file mode 100644 index 0000000..906b196 --- /dev/null +++ b/python/python/tests/test_list_folds.py @@ -0,0 +1,240 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The lance-csr Authors + +"""Ordered reduce, lexical scope, and Cypher's three-valued list predicates.""" + +import itertools + +import pyarrow as pa +import pytest +from lance_csr import CypherEngine, GraphConfig + + +@pytest.fixture(params=[False, True], ids=["join", "csr"]) +def engine(request): + config = ( + GraphConfig.builder() + .node_table("Person", "id") + .rel_table("KNOWS", "Person", "Person", "src", "dst") + .build() + ) + return CypherEngine( + config, + { + "Person": pa.table( + { + "id": [1, 2, 3], + "items": pa.array([[1, None, 3], None, []], pa.list_(pa.int64())), + "large": pa.array([[1, 2], None, []], pa.large_list(pa.int64())), + "pair": pa.array([[1, 2], None, [3, 4]], pa.list_(pa.int64(), 2)), + } + ), + "KNOWS": pa.table({"src": [1, 1, 2], "dst": [2, 3, 3]}), + }, + use_csr=request.param, + ) + + +@pytest.mark.parametrize( + ("expression", "expected"), + [ + ("reduce(s=0, x IN [1,2,3] | s*10+x)", 123), + ("reduce(s=9, x IN [] | s+x)", 9), + ("reduce(s=9, x IN null | s+x)", None), + ("reduce(s=0, x IN [1,null,3] | s+x)", None), + ("reduce(s=0, x IN [1,null,3] | s+coalesce(x,0))", 4), + ("reduce(s=null, x IN [1,2,3] | coalesce(s,0)+x)", 6), + ("reduce(s=0, x IN [1.5,2.5] | s+x)", 4.0), + ("reduce(s=1, x IN [2,3] | s/x)", 0), + ("reduce(s='猫😀猫', x IN ['猫','😀'] | replace(s,x,'a'))", "aaa"), + ("reduce(s={n:0}, x IN [1,2] | {n:s.n+x}).n", 3), + ("reduce(s=0, x IN [{n:1},{n:2}] | s+x.n)", 3), + ("reduce(s=0, xs IN [[1,2],[3]] | s+reduce(t=0, x IN xs | t+x))", 6), + ("[xs IN [[1,2],[3]] | reduce(s=0,x IN xs | s+x)]", [3, 3]), + ("reduce(s=0, x IN [1,2] | s+reduce(s=10, y IN [x] | s+y))", 23), + ("reduce(s=0, x IN [1,2] | s+reduce(t=s, y IN [x] | t+y))", 4), + ( + "reduce(s=0, x IN [1,2] | s+CASE WHEN any(y IN [x] WHERE y=2) " + "THEN 10 ELSE x END)", + 11, + ), + ("any(x IN [null] WHERE x IS NULL)", True), + ("all(x IN [null] WHERE x IS NULL)", True), + ("single(x IN [null] WHERE x IS NULL)", True), + ("none(x IN [null] WHERE x IS NULL)", False), + ("any(x IN [[1,2],[3]] WHERE size(x)=2)", True), + ("all(x IN [{n:1},{n:2}] WHERE x.n>0)", True), + ("all(x IN [1,2] WHERE any(x IN [3,4] WHERE x>2))", True), + ], +) +def test_fold_expression(engine, expression, expected): + assert engine.execute(f"RETURN {expression} AS result").to_pylist() == [ + {"result": expected} + ] + + +def test_predicate_truth_tables(engine): + """Exhaust true/false/null combinations through length three.""" + lists = [None] + [ + list(values) + for length in range(4) + for values in itertools.product([True, False, None], repeat=length) + ] + expected = [] + for values in lists: + if values is None: + expected.append(dict.fromkeys(["a", "b", "c", "d"])) + continue + yes = values.count(True) + no = values.count(False) + unknown = None in values + expected.append( + { + "a": False if no else None if unknown else True, + "b": True if yes else None if unknown else False, + "c": False if yes else None if unknown else True, + "d": False if yes > 1 else None if unknown else yes == 1, + } + ) + for values, want in zip(lists, expected, strict=True): + result = engine.execute( + "RETURN all(x IN $xs WHERE x) AS a, any(x IN $xs WHERE x) AS b, " + "none(x IN $xs WHERE x) AS c, single(x IN $xs WHERE x) AS d", + {"xs": values}, + ).to_pylist() + assert result == [want], values + + +def test_captured_properties_scope_and_arrow_lists(engine): + result = engine.execute( + "MATCH (p:Person) RETURN p.id AS id, " + "reduce(p=p.id, x IN p.items | p+coalesce(x,0)) AS reduced, " + "all(x IN p.items WHERE x(q:Person)|q.id] | s+x) AS result" + ).to_pylist() == [{"result": 5}] + assert engine.execute( + "MATCH (p:Person {id:1}) RETURN " + "[x IN [2,3] | reduce(w=0.0, v IN [(p)-[:KNOWS]->(q:Person) " + "WHERE q.id=x | 0.5] | w+v)] AS result" + ).to_pylist() == [{"result": [0.5, 0.5]}] + + +def test_graph_predicates_reuse_correlated_comprehensions(engine): + result = engine.execute( + "MATCH (p:Person) RETURN p.id AS id, " + "any(x IN [2,3] WHERE EXISTS {MATCH (p)-[:KNOWS]->(q:Person) " + "WHERE q.id=x}) AS connected " + "ORDER BY id" + ).to_pylist() + assert result == [ + {"id": 1, "connected": True}, + {"id": 2, "connected": True}, + {"id": 3, "connected": False}, + ] + for value, expected in [(None, None), ([], True)]: + assert engine.execute( + "MATCH (p:Person {id:1}) RETURN " + "all(x IN $xs WHERE EXISTS {MATCH (p)-[:KNOWS]->(q:Person) " + "WHERE q.id=x}) AS result", + {"xs": value}, + ).to_pylist() == [{"result": expected}] + + +def test_graph_predicate_null_and_duplicate_capture_bindings(engine): + result = engine.execute( + "MATCH (p:Person {id:1}) RETURN " + "[x IN [2,2,null,99] | any(y IN [x] WHERE EXISTS {" + "MATCH (p)-[:KNOWS]->(q:Person) WHERE q.id=y OR y IS NULL})] AS result" + ).to_pylist() + assert result == [{"result": [True, True, True, False]}] + # No shared graph variable: correlation consists only of the scalar iterator. + assert engine.execute( + "RETURN all(x IN [1,2,3] WHERE EXISTS {MATCH (p:Person) " + "WHERE p.id=x}) AS result" + ).to_pylist() == [{"result": True}] + + +def test_pattern_reduce_correlates_captured_struct_and_duplicate_rows(engine): + assert engine.execute( + "WITH [{id:2},{id:2},{id:3}] AS values UNWIND values AS captured " + "RETURN captured.id AS id, reduce(s=0, x IN " + "[(a:Person)-[:KNOWS]->(b:Person) WHERE b.id=captured.id | 1] | s+x) AS n " + "ORDER BY id" + ).to_pylist() == [{"id": 2, "n": 1}, {"id": 2, "n": 1}, {"id": 3, "n": 2}] + + +@pytest.mark.parametrize( + "query", + [ + "RETURN reduce(s=0, s IN [1] | s)", + "RETURN reduce(s=0, x IN 1 | s+x)", + "RETURN any(x IN 1 WHERE true)", + "RETURN all(x IN [1] WHERE x)", + "RETURN reduce(s=0, x IN [1] | s+undefined)", + "RETURN reduce(s=0, x IN [1] | s+x) AS result, x", + "RETURN any(x IN [1] WHERE x=1) AS result, x", + "RETURN reduce(s=0, x IN [1] | toString(x))", + "RETURN reduce(s=0, x IN [1] | size([y IN [x] | y]))", + "MATCH (p:Person) RETURN reduce(s=0,x IN [1] | " + "CASE WHEN EXISTS {MATCH (p)-[:KNOWS]->(:Person)} THEN s+x ELSE s END)", + ], +) +def test_invalid_fold_scope_and_types(engine, query): + with pytest.raises((ValueError, RuntimeError, NotImplementedError)): + engine.execute(query) + + +@pytest.mark.parametrize("use_csr", [False, True], ids=["join", "csr"]) +def test_sliced_lists_and_unreachable_null_storage(use_csr): + config = GraphConfig.builder().node_table("Person", "id").build() + items = pa.ListArray.from_arrays( + pa.array([0, 2, 4], type=pa.int32()), + pa.array([0, 0, 1, 2]), + mask=pa.array([True, False]), + ) + people = pa.table({"id": [1, 2], "items": items}) + query = ( + "MATCH (p:Person) RETURN p.id AS id, " + "reduce(s=12,x IN p.items | s/x) AS value, " + "all(x IN p.items WHERE 12/x>0) AS predicate ORDER BY id" + ) + expected = [ + {"id": 1, "value": None, "predicate": None}, + {"id": 2, "value": 6, "predicate": True}, + ] + for table, want in [(people, expected), (people.slice(1), expected[1:])]: + local = CypherEngine(config, {"Person": table}, use_csr=use_csr) + assert local.execute(query).to_pylist() == want + + +def test_fold_in_sort_and_unwind(engine): + assert engine.execute( + "MATCH (p:Person) RETURN p.id AS id ORDER BY reduce(s=0,x IN [p.id] | s-x)" + ).to_pylist() == [{"id": 3}, {"id": 2}, {"id": 1}] + assert engine.execute( + "UNWIND [reduce(s=0,x IN [1,2] | s+x)] AS value RETURN value" + ).to_pylist() == [{"value": 3}]