From 43fecac215de2895bd49750ee3e406b1e1aba2ff Mon Sep 17 00:00:00 2001 From: David Hughes Date: Sun, 25 Oct 2026 11:36:53 -0700 Subject: [PATCH 1/6] feat(cypher): add string and list scalar functions Implement the separate string/list submission for #20 on the numeric-function prerequisite. Add Unicode string operations, list extraction/reversal, outer list size, stepped ranges, strict argument kinds, and empty quoted literals. Update the unsupported-function regression now that replace is supported. Validation: Python static checks, formatting, strict docs, and spelling pass. Exact-source Rust and Python runtime checks are pending and tracked separately; old combined-source evidence is preserved. Submitted by David Hughes (@stugorf), with AI Collaboration. --- .../src/datafusion_planner/expression.rs | 23 +- .../lance-graph/src/datafusion_planner/mod.rs | 1 + .../datafusion_planner/scalar_functions.rs | 398 ++++++++++++++++++ crates/lance-graph/src/parser.rs | 6 +- crates/lance-graph/src/semantic.rs | 26 +- .../tests/test_datafusion_pipeline.rs | 9 +- docs/cypher.md | 31 +- python/python/tests/test_scalar_functions.py | 176 +++++++- 8 files changed, 629 insertions(+), 41 deletions(-) create mode 100644 crates/lance-graph/src/datafusion_planner/scalar_functions.rs diff --git a/crates/lance-graph/src/datafusion_planner/expression.rs b/crates/lance-graph/src/datafusion_planner/expression.rs index e57229e..555528a 100644 --- a/crates/lance-graph/src/datafusion_planner/expression.rs +++ b/crates/lance-graph/src/datafusion_planner/expression.rs @@ -196,18 +196,17 @@ pub(crate) fn to_df_value_expr(expr: &ValueExpression) -> Expr { "abs" | "round" | "floor" | "ceil" | "sqrt" => { super::math_functions::call(name, args.iter().map(to_df_value_expr).collect()) } - "size" => Expr::Cast(datafusion::logical_expr::Cast::new( - Box::new(datafusion::functions_nested::expr_fn::cardinality( - to_df_value_expr(&args[0]), - )), - arrow_schema::DataType::Int64, - )), - // Cypher's range includes its upper bound, like DataFusion's gen_series. - "range" => datafusion::functions_nested::expr_fn::gen_series( - to_df_value_expr(&args[0]), - to_df_value_expr(&args[1]), - lit(1_i64), - ), + "left" | "right" | "substring" | "replace" | "split" | "trim" | "ltrim" + | "rtrim" | "size" | "reverse" | "last" | "tail" => { + super::scalar_functions::call(name, args.iter().map(to_df_value_expr).collect()) + } + "range" => { + let mut args: Vec<_> = args.iter().map(to_df_value_expr).collect(); + if args.len() == 2 { + args.push(lit(1_i64)); + } + super::scalar_functions::call(name, args) + } // list[i] (0-based in Cypher, 1-based in DataFusion's array_element). "__index" => datafusion::functions_nested::expr_fn::array_element( to_df_value_expr(&args[0]), diff --git a/crates/lance-graph/src/datafusion_planner/mod.rs b/crates/lance-graph/src/datafusion_planner/mod.rs index 9f66573..8be2325 100644 --- a/crates/lance-graph/src/datafusion_planner/mod.rs +++ b/crates/lance-graph/src/datafusion_planner/mod.rs @@ -21,6 +21,7 @@ mod cypher_udf; mod expression; mod join_ops; mod math_functions; +mod scalar_functions; mod scan_ops; mod udf; pub mod vector_ops; diff --git a/crates/lance-graph/src/datafusion_planner/scalar_functions.rs b/crates/lance-graph/src/datafusion_planner/scalar_functions.rs new file mode 100644 index 0000000..3763b67 --- /dev/null +++ b/crates/lance-graph/src/datafusion_planner/scalar_functions.rs @@ -0,0 +1,398 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The lance-csr Authors + +//! Cypher scalar semantics: strict argument kinds, Unicode characters, and nulls. +//! List operations reuse DataFusion's Arrow kernels; string slicing differs from SQL. + +use std::sync::Arc; + +use arrow::array::{ + new_null_array, Array, AsArray, Int64Array, ListBuilder, StringArray, StringBuilder, +}; +use arrow::compute::cast; +use arrow_schema::{DataType, Field}; +use datafusion::common::{exec_err, plan_err, ScalarValue}; +use datafusion::error::Result; +use datafusion::functions_nested::{extract, length, range, reverse}; +use datafusion::logical_expr::{ + ColumnarValue, Expr, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, Volatility, +}; + +#[derive(Debug, PartialEq, Eq, Hash)] +struct CypherScalar { + name: String, + signature: Signature, +} + +fn is_string(t: &DataType) -> bool { + matches!(t, DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View) +} + +fn list_field(t: &DataType) -> Option<&Arc> { + match t { + DataType::List(f) | DataType::LargeList(f) | DataType::FixedSizeList(f, _) => Some(f), + _ => None, + } +} + +impl ScalarUDFImpl for CypherScalar { + fn name(&self) -> &str { + &self.name + } + + fn signature(&self) -> &Signature { + &self.signature + } + + fn return_type(&self, types: &[DataType]) -> Result { + for (i, t) in types.iter().enumerate() { + let valid = match self.name.as_str() { + "range" => t.is_integer(), + "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(), + _ => is_string(t), + }; + if !valid && !t.is_null() { + return plan_err!("{} does not accept {t} for argument {}", self.name, i + 1); + } + } + Ok(match self.name.as_str() { + "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))), + "last" => list_field(&types[0]) + .map(|f| f.data_type().clone()) + .unwrap_or(DataType::Null), + "tail" => match &types[0] { + DataType::FixedSizeList(f, _) => DataType::List(f.clone()), + t => t.clone(), + }, + "reverse" if list_field(&types[0]).is_some() => types[0].clone(), + _ => DataType::Utf8, + }) + } + + fn invoke_with_args(&self, mut args: ScalarFunctionArgs) -> Result { + let output_type = args.return_type().clone(); + if args.args[0].data_type().is_null() && self.name != "range" { + return Ok(ColumnarValue::Array(new_null_array( + &output_type, + args.number_rows, + ))); + } + if self.name == "range" { + for arg in &mut args.args { + *arg = + ColumnarValue::Array(cast(&arg.to_array(args.number_rows)?, &DataType::Int64)?); + } + return range::gen_series_udf().invoke_with_args(args); + } + if list_field(&args.args[0].data_type()).is_some() { + let udf = match self.name.as_str() { + "size" => length::array_length_udf(), + "reverse" => reverse::array_reverse_udf(), + "last" => { + if let DataType::FixedSizeList(field, _) = args.args[0].data_type() { + args.args[0] = ColumnarValue::Array(cast( + &args.args[0].to_array(args.number_rows)?, + &DataType::List(field), + )?); + } + args.args + .push(ColumnarValue::Scalar(ScalarValue::Int64(Some(-1)))); + extract::array_element_udf() + } + "tail" => { + args.args[0] = ColumnarValue::Array(cast( + &args.args[0].to_array(args.number_rows)?, + &output_type, + )?); + args.args.extend( + [2, -1].map(|n| ColumnarValue::Scalar(ScalarValue::Int64(Some(n)))), + ); + extract::array_slice_udf() + } + _ => unreachable!("validated list function"), + }; + let rows = args.number_rows; + args.arg_fields = args + .args + .iter() + .map(|arg| Arc::new(Field::new("arg", arg.data_type(), true))) + .collect(); + args.return_field = Arc::new(Field::new( + udf.name(), + udf.return_type( + &args + .args + .iter() + .map(ColumnarValue::data_type) + .collect::>(), + )?, + true, + )); + let result = udf.invoke_with_args(args)?; + return if self.name == "size" { + Ok(ColumnarValue::Array(cast( + &result.to_array(rows)?, + &DataType::Int64, + )?)) + } else { + Ok(result) + }; + } + let arrays = args + .args + .iter() + .enumerate() + .map(|(i, arg)| { + let t = if i > 0 && matches!(self.name.as_str(), "left" | "right" | "substring") { + DataType::Int64 + } else { + DataType::Utf8 + }; + cast(&arg.to_array(args.number_rows)?, &t) + }) + .collect::, _>>()?; + let strings = arrays[0].as_string::(); + if self.name == "size" { + return Ok(ColumnarValue::Array(Arc::new( + strings + .iter() + .map(|s| s.map(|s| s.chars().count() as i64)) + .collect::(), + ))); + } + if self.name == "split" { + let delimiters = arrays[1].as_string::(); + let mut result = ListBuilder::new(StringBuilder::new()); + for (s, delimiter) in strings.iter().zip(delimiters.iter()) { + match (s, delimiter) { + (Some(s), Some(delimiter)) => { + if s.is_empty() { + result.values().append_value(""); + } else if delimiter.is_empty() { + for c in s.chars() { + result.values().append_value(c.to_string()); + } + } else { + for part in s.split(delimiter) { + result.values().append_value(part); + } + } + result.append(true); + } + _ => result.append(false), + } + } + return Ok(ColumnarValue::Array(Arc::new(result.finish()))); + } + let result = strings + .iter() + .enumerate() + .map(|(row, s)| { + let Some(s) = s else { return Ok(None) }; + let value = match self.name.as_str() { + "trim" => s.trim().to_string(), + "ltrim" => s.trim_start().to_string(), + "rtrim" => s.trim_end().to_string(), + "reverse" => s.chars().rev().collect(), + "replace" => { + if arrays[1].is_null(row) || arrays[2].is_null(row) { + return Ok(None); + } + s.replace( + arrays[1].as_string::().value(row), + arrays[2].as_string::().value(row), + ) + } + "left" | "right" | "substring" => { + let index = |i: usize| -> Result { + if arrays[i].is_null(row) { + return exec_err!( + "{} requires a non-null integer for argument {}", + self.name, + i + 1 + ); + } + let n = arrays[i] + .as_primitive::() + .value(row); + if n < 0 { + return exec_err!( + "{} requires a non-negative integer for argument {}", + self.name, + i + 1 + ); + } + Ok(n as usize) + }; + let n = index(1)?; + let (start, count) = match self.name.as_str() { + "left" => (0, n), + "right" => (s.chars().count().saturating_sub(n), n), + _ => ( + n, + if arrays.len() == 3 { + index(2)? + } else { + usize::MAX + }, + ), + }; + s.chars().skip(start).take(count).collect() + } + _ => unreachable!("validated string function"), + }; + Ok(Some(value)) + }) + .collect::>()?; + Ok(ColumnarValue::Array(Arc::new(result))) + } +} + +pub(super) fn call(name: &str, args: Vec) -> Expr { + ScalarUDF::new_from_impl(CypherScalar { + name: name.to_lowercase(), + signature: Signature::any(args.len(), Volatility::Immutable), + }) + .call(args) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{config::GraphConfig, query::CypherQuery}; + use arrow::array::{ArrayRef, RecordBatch}; + use std::collections::HashMap; + + async fn run(expression: &str) -> crate::error::Result { + let batch = + RecordBatch::try_from_iter([("id", Arc::new(Int64Array::from(vec![1])) as ArrayRef)]) + .unwrap(); + CypherQuery::new(&format!("RETURN {expression} AS value"))? + .with_config( + GraphConfig::builder() + .node_table("Person", "id") + .build() + .unwrap(), + ) + .execute(HashMap::from([("Person".to_string(), batch)]), None) + .await + } + + #[tokio::test] + async fn cypher_scalar_unicode_and_slicing() { + for (expression, expected) in [ + ("left('a猫😀z', 3)", "a猫😀"), + ("right('a猫😀z', 2)", "😀z"), + ("substring('a猫😀z', 1, 2)", "猫😀"), + ("substring('a猫😀z', 2)", "😀z"), + ("substring('x', 99, 2)", ""), + ("right('x', 0)", ""), + ("left('x', 99)", "x"), + ("reverse('a猫😀z')", "z😀猫a"), + ("trim(' 猫 ')", "猫"), + ("lTrim(' 猫 ')", "猫 "), + ("rTrim(' 猫 ')", " 猫"), + ("replace('a猫a', 'a', '😀')", "😀猫😀"), + ] { + let out = run(expression).await.unwrap(); + assert_eq!( + out.column(0).as_string::().value(0), + expected, + "{expression}" + ); + } + let out = run("size('a猫😀z')").await.unwrap(); + assert_eq!( + out.column(0) + .as_primitive::() + .value(0), + 4 + ); + } + + #[tokio::test] + async fn cypher_scalar_list_operations() { + for (expression, expected) in [ + ("reverse([1, 2, 3])", vec![3, 2, 1]), + ("tail([1, 2, 3])", vec![2, 3]), + ("tail([1])", vec![]), + ("range(5, 1, -2)", vec![5, 3, 1]), + ("range(1, 5, -2)", vec![]), + ("range(0, 4, 2)", vec![0, 2, 4]), + ] { + let out = run(expression).await.unwrap(); + let values = out.column(0).as_list::().value(0); + let values: Vec<_> = values + .as_primitive::() + .values() + .to_vec(); + assert_eq!(values, expected, "{expression}"); + } + for (expression, expected) in [("last([1, 2, 3])", 3), ("size([[1, 2], [3]])", 2)] { + let out = run(expression).await.unwrap(); + assert_eq!( + out.column(0) + .as_primitive::() + .value(0), + expected + ); + } + for (expression, expected) in [ + ("split('a,,b,', ',')", vec!["a", "", "b", ""]), + ("split('猫😀', '')", vec!["猫", "😀"]), + ("split('', ',')", vec![""]), + ] { + let out = run(expression).await.unwrap(); + let values = out.column(0).as_list::().value(0); + assert_eq!( + values + .as_string::() + .iter() + .map(Option::unwrap) + .collect::>(), + expected + ); + } + } + + #[tokio::test] + async fn cypher_scalar_nulls_and_invalid_arguments() { + for expression in [ + "size(null)", + "reverse(null)", + "last(null)", + "tail(null)", + "last([])", + "left(null, null)", + "substring(null, -1)", + "replace('x', null, '')", + "split('x', null)", + "range(null, 3)", + ] { + let out = run(expression).await.unwrap(); + assert_eq!(out.column(0).logical_null_count(), 1, "{expression}"); + } + for expression in [ + "left('x', -1)", + "right('x', null)", + "substring('x', -1)", + "substring('x', 0, null)", + "range(1, 3, 0)", + "range(1.5, 3)", + "size(1)", + "reverse(1)", + "tail('abc')", + "last(2)", + "left('x', 1.5)", + "split('x', 1)", + "substring('x')", + "range(1)", + ] { + assert!(run(expression).await.is_err(), "{expression} should fail"); + } + } +} diff --git a/crates/lance-graph/src/parser.rs b/crates/lance-graph/src/parser.rs index 2f6035b..260240b 100644 --- a/crates/lance-graph/src/parser.rs +++ b/crates/lance-graph/src/parser.rs @@ -10,7 +10,7 @@ use crate::ast::*; use crate::error::{GraphError, Result}; use nom::{ branch::alt, - bytes::complete::{tag, tag_no_case, take_while1}, + bytes::complete::{tag, tag_no_case, take_while, take_while1}, character::complete::{char, digit0, digit1, multispace0, multispace1, one_of, satisfy}, combinator::{map, map_res, not, opt, peek, recognize}, multi::{many0, many1, separated_list0, separated_list1}, @@ -1382,14 +1382,14 @@ fn string_literal(input: &str) -> IResult<&str, String> { fn double_quoted_string(input: &str) -> IResult<&str, String> { let (input, _) = char('"')(input)?; - let (input, content) = take_while1(|c| c != '"')(input)?; + let (input, content) = take_while(|c| c != '"')(input)?; let (input, _) = char('"')(input)?; Ok((input, content.to_string())) } fn single_quoted_string(input: &str) -> IResult<&str, String> { let (input, _) = char('\'')(input)?; - let (input, content) = take_while1(|c| c != '\'')(input)?; + let (input, content) = take_while(|c| c != '\'')(input)?; let (input, _) = char('\'')(input)?; Ok((input, content.to_string())) } diff --git a/crates/lance-graph/src/semantic.rs b/crates/lance-graph/src/semantic.rs index f0a2227..ac30dff 100644 --- a/crates/lance-graph/src/semantic.rs +++ b/crates/lance-graph/src/semantic.rs @@ -439,7 +439,8 @@ impl SemanticAnalyzer { // Validate arity and known functions match function_name.as_str() { "tolower" | "lower" | "toupper" | "upper" | "abs" | "round" | "floor" - | "ceil" | "sqrt" => { + | "ceil" | "sqrt" | "trim" | "ltrim" | "rtrim" | "reverse" | "last" + | "tail" => { if args.len() != 1 { return Err(GraphError::PlanError { message: format!( @@ -451,10 +452,11 @@ impl SemanticAnalyzer { }); } } - "date_part" | "to_epoch_ms" | "length" | "nodes" | "size" | "range" - | "__index" => { + "date_part" | "to_epoch_ms" | "length" | "nodes" | "size" | "left" + | "right" | "split" | "replace" | "__index" => { let want = match function_name.as_str() { - "date_part" | "range" | "__index" => 2, + "date_part" | "__index" | "left" | "right" | "split" => 2, + "replace" => 3, _ => 1, }; if args.len() != want { @@ -469,6 +471,18 @@ impl SemanticAnalyzer { }); } } + "substring" | "range" => { + if !(2..=3).contains(&args.len()) { + return Err(GraphError::PlanError { + message: format!( + "{} requires 2 or 3 arguments, got {}", + name, + args.len() + ), + location: snafu::Location::new(file!(), line!(), column!()), + }); + } + } "__list" => {} "head" | "tointeger" | "tofloat" | "date" | "datetime" => { if args.len() != 1 { @@ -507,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, abs, round, floor, ceil, sqrt. 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, 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!()), @@ -1595,7 +1609,7 @@ mod tests { #[test] fn test_unimplemented_scalar_function_fails_validation() { let expr = ValueExpression::ScalarFunction { - name: "replace".to_string(), + name: "not_a_function".to_string(), args: vec![ValueExpression::Property(PropertyRef::new("n", "name"))], }; let result = analyze_return_with_match("n", "Person", expr).unwrap(); diff --git a/crates/lance-graph/tests/test_datafusion_pipeline.rs b/crates/lance-graph/tests/test_datafusion_pipeline.rs index 2b2041f..b256185 100644 --- a/crates/lance-graph/tests/test_datafusion_pipeline.rs +++ b/crates/lance-graph/tests/test_datafusion_pipeline.rs @@ -4747,7 +4747,7 @@ async fn test_unimplemented_scalar_function_errors() { .unwrap(); let query = CypherQuery::new( - "MATCH (p:Person) RETURN p.name AS name, replace(p.name, 'A', 'a') AS replaced", + "MATCH (p:Person) RETURN p.name AS name, not_a_function(p.name) AS result", ) .unwrap() .with_config(config); @@ -4758,10 +4758,13 @@ async fn test_unimplemented_scalar_function_errors() { let err = query .execute(datasets, Some(ExecutionStrategy::DataFusion)) .await - .expect_err("replace() should error until implemented"); + .expect_err("unknown functions must fail explicitly"); let message = err.to_string().to_lowercase(); - assert!(message.contains("replace"), "unexpected error: {err}"); + assert!( + message.contains("not_a_function"), + "unexpected error: {err}" + ); assert!( message.contains("not implemented") || message.contains("unsupported"), "unexpected error: {err}" diff --git a/docs/cypher.md b/docs/cypher.md index aaa1bfc..6e0128f 100644 --- a/docs/cypher.md +++ b/docs/cypher.md @@ -258,14 +258,36 @@ RETURN length(p), nodes(p) | Group | Functions | | --- | --- | | Aggregates | `count` (including `count(*)` and `count(DISTINCT x)`), `sum`, `avg`, `min`, `max`, `collect` (including `DISTINCT`; skips nulls) | -| Strings | `toLower` / `lower`, `toUpper` / `upper`, `toString` | +| Strings | `toLower` / `lower`, `toUpper` / `upper`, `toString`, `left`, `right`, `substring`, `replace`, `split`, `trim`, `lTrim`, `rTrim`, `reverse`, `size` | | Conversion | `toInteger`, `toFloat`, `coalesce` | -| Lists | `size`, `head`, `range(start, end)` (inclusive, no step) | +| Lists | `size`, `head`, `last`, `tail`, `reverse`, `range(start, end [, step])` (inclusive) | | 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 | | Vectors | `vector_distance(a, b, metric)`, `vector_similarity(a, b, metric)`; see [Vector search](vector-search.md) | +The string and list functions follow the [Cypher v9 reference](https://s3.amazonaws.com/artifacts.opencypher.org/openCypher9.pdf), +within the list representation described above. `size` counts Unicode characters in a string or the +outer elements of a list. `left(text, length)`, `right(text, length)` and +`substring(text, start [, length])` count characters; the substring start is zero-based. Lengths and +starts must be non-negative integers. A null input string returns null; a null start or length on a +non-null string raises an error. `trim`, `lTrim` and `rTrim` remove whitespace, and `replace` replaces +all literal matches. `split` preserves empty fields; an empty delimiter splits into characters. +Splitting an empty string returns `['']`. + +`last([])` returns null; `tail([])` and `reverse([])` return empty lists. `reverse` also accepts strings. +`range(start, end [, step])` includes the end when reached, defaults to step 1, and accepts negative +steps. A step pointing away from the end gives an empty list; zero steps and non-integer arguments +raise an error. Null arguments return null. + +```python +result = engine.execute(""" + RETURN substring('a猫😀z', 1, 2) AS text, + reverse(range(1, 5, 2)) AS numbers +""") +print(result.to_pylist()) # [{'text': '猫😀', 'numbers': [5, 3, 1]}] +``` + `abs` preserves integer results; the other math functions return floating point values. Integer `abs` raises an error on absolute-value overflow or unsigned inputs outside Int64 range. The other math functions convert numeric inputs, including UInt64, to Float64. `round` accepts one argument, @@ -281,9 +303,8 @@ print(result.to_pylist()) # [{'magnitude': 5, 'rounded': -1.0, 'root': 3.0}] ``` 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 most -string and list functions, `reduce`, the list predicates `any` / `all` / `none` / `single`, and -`relationships(p)`. +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)`. ### Example: a little of everything diff --git a/python/python/tests/test_scalar_functions.py b/python/python/tests/test_scalar_functions.py index 83eb1b5..4e24460 100644 --- a/python/python/tests/test_scalar_functions.py +++ b/python/python/tests/test_scalar_functions.py @@ -1,22 +1,174 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright The Lance Authors +"""Cypher v9 scalar semantics, including values from Arrow-backed graph rows.""" + import pyarrow as pa import pytest -from lance_csr import CypherQuery, GraphConfig +from lance_csr import CypherEngine, GraphConfig + + +@pytest.fixture(params=[False, True], ids=["join", "csr"]) +def engine(request): + config = GraphConfig.builder().node_table("Person", "id").build() + people = pa.table( + { + "id": [1, 2, 3], + "text": pa.array(["a猫😀z", None, ""], type=pa.large_string()), + "items": pa.array([[1, None, 3], None, []], type=pa.list_(pa.int64())), + "pair": pa.array([[1, 2], None, [3, 4]], type=pa.list_(pa.int64(), 2)), + } + ) + return CypherEngine(config, {"Person": people}, use_csr=request.param) + + +@pytest.mark.parametrize( + ("expression", "expected"), + [ + ("left('a猫😀z', 3)", "a猫😀"), + ("right('a猫😀z', 2)", "😀z"), + ("substring('a猫😀z', 1, 2)", "猫😀"), + ("substring('a猫😀z', 2)", "😀z"), + ("substring('x', 99, 2)", ""), + ("substring('x', 0, 0)", ""), + ("left('x', 0)", ""), + ("right('x', 0)", ""), + ("left('x', 99)", "x"), + ("right('x', 99)", "x"), + ("size('a猫😀z')", 4), + ("size('')", 0), + ('size("")', 0), + ("reverse('a猫😀z')", "z😀猫a"), + ("replace('a猫a', 'a', '😀')", "😀猫😀"), + ("replace('abc', 'x', 'y')", "abc"), + ("split('a,,b,', ',')", ["a", "", "b", ""]), + ("split('猫😀', '')", ["猫", "😀"]), + ("split('', ',')", [""]), + ("split('', '')", [""]), + ("trim(' 猫 ')", "猫"), + ("lTrim(' 猫 ')", "猫 "), + ("rTrim(' 猫 ')", " 猫"), + ("last([1, null, 3])", 3), + ("last([1, null])", None), + ("last([])", None), + ("tail([1, null, 3])", [None, 3]), + ("tail([1])", []), + ("tail([])", []), + ("reverse([1, null, 3])", [3, None, 1]), + ("reverse([])", []), + ("reverse([[1, 2], [3]])", [[3], [1, 2]]), + ("size([[1, 2], [3]])", 2), + ("range(0, 4)", [0, 1, 2, 3, 4]), + ("range(0, 4, 2)", [0, 2, 4]), + ("range(5, 1, -2)", [5, 3, 1]), + ("range(1, 5, -2)", []), + ("range(5, 1, 2)", []), + ("range(1, 1, -2)", [1]), + ("range(null, 4)", None), + ("range(1, null)", None), + ("range(1, 4, null)", None), + ("left(null, null)", None), + ("substring(null, -1)", None), + ("replace('x', null, '')", None), + ("split('x', null)", None), + ] + + [ + (f"{name}(null)", None) + for name in ( + "trim", + "lTrim", + "rTrim", + "reverse", + "size", + "last", + "tail", + ) + ], +) +def test_scalar_value(engine, expression, expected): + result = engine.execute(f"RETURN {expression} AS value") + assert result.to_pylist() == [{"value": expected}] + + +@pytest.mark.parametrize( + "expression", + [ + "left('x', -1)", + "right('x', null)", + "substring('x', -1)", + "substring('x', 0, null)", + "range(1, 3, 0)", + "range(1.5, 3)", + "size(1)", + "reverse(1)", + "tail('abc')", + "last(2)", + "left('x', 1.5)", + "split('x', 1)", + "substring('x')", + "range(1)", + ], +) +def test_scalar_invalid_arguments(engine, expression): + with pytest.raises((ValueError, RuntimeError)): + engine.execute(f"RETURN {expression} AS value") + + +def test_scalar_arrow_columns_and_nulls(engine): + result = engine.execute( + "MATCH (p:Person) RETURN p.id AS id, " + "substring(p.text, 1, 2) AS sub, size(p.text) AS size, " + "reverse(p.items) AS reversed, " + "tail(p.pair) AS tail, last(p.pair) AS last ORDER BY id" + ) + assert result.to_pylist() == [ + { + "id": 1, + "sub": "猫😀", + "size": 4, + "reversed": [3, None, 1], + "tail": [2], + "last": 2, + }, + { + "id": 2, + "sub": None, + "size": None, + "reversed": None, + "tail": None, + "last": None, + }, + { + "id": 3, + "sub": "", + "size": 0, + "reversed": [], + "tail": [4], + "last": 4, + }, + ] + assert result.schema.field("size").type == pa.int64() + +def test_scalar_whitespace(engine): + result = engine.execute( + "RETURN trim($text) AS text", {"text": "\t\n 猫\u2003"} + ).to_pylist()[0] + assert result["text"] == "猫" -def test_unimplemented_scalar_function_errors() -> None: - cfg = GraphConfig.builder().node_table("Person", "name").build() - datasets = {"Person": pa.table({"name": ["Alice", "BOB", "CaSeY"]})} - query = CypherQuery( - "MATCH (p:Person) " - "RETURN p.name AS name, replace(p.name, 'A', 'a') AS replaced " - "ORDER BY name", - ).with_config(cfg) +def test_unimplemented_scalar_function_errors(engine): + with pytest.raises(ValueError, match="not_a_function.*is not implemented"): + engine.execute("RETURN not_a_function('x') AS value") - with pytest.raises(Exception) as excinfo: - query.execute(datasets) - assert "replace" in str(excinfo.value).lower() +def test_scalar_varying_arguments(engine): + result = engine.execute( + "MATCH (p:Person) RETURN p.id AS id, left(p.text, p.id) AS prefix, " + "range(0, p.id, 2) AS numbers ORDER BY id" + ) + assert result.to_pylist() == [ + {"id": 1, "prefix": "a", "numbers": [0]}, + {"id": 2, "prefix": None, "numbers": [0, 2]}, + {"id": 3, "prefix": "", "numbers": [0, 2]}, + ] From 23d00ee2c7d23e364a97097fa4bd0542d75ee234 Mon Sep 17 00:00:00 2001 From: David Hughes Date: Sun, 25 Oct 2026 12:50:38 -0700 Subject: [PATCH 2/6] fix(cypher): reject unsigned range argument overflow Return an error when an Arrow UInt64 range endpoint or step exceeds Int64 instead of silently converting it to null. Preserve actual null arguments. Add Rust and CSR/join Python regressions and document the boundary. --- .../datafusion_planner/scalar_functions.rs | 41 +++++++++++++++++-- docs/cypher.md | 3 +- python/python/tests/test_scalar_functions.py | 20 +++++++++ 3 files changed, 60 insertions(+), 4 deletions(-) diff --git a/crates/lance-graph/src/datafusion_planner/scalar_functions.rs b/crates/lance-graph/src/datafusion_planner/scalar_functions.rs index 3763b67..f46e54a 100644 --- a/crates/lance-graph/src/datafusion_planner/scalar_functions.rs +++ b/crates/lance-graph/src/datafusion_planner/scalar_functions.rs @@ -9,7 +9,7 @@ use std::sync::Arc; use arrow::array::{ new_null_array, Array, AsArray, Int64Array, ListBuilder, StringArray, StringBuilder, }; -use arrow::compute::cast; +use arrow::compute::{cast, cast_with_options, CastOptions}; use arrow_schema::{DataType, Field}; use datafusion::common::{exec_err, plan_err, ScalarValue}; use datafusion::error::Result; @@ -83,8 +83,14 @@ impl ScalarUDFImpl for CypherScalar { } if self.name == "range" { for arg in &mut args.args { - *arg = - ColumnarValue::Array(cast(&arg.to_array(args.number_rows)?, &DataType::Int64)?); + *arg = ColumnarValue::Array(cast_with_options( + &arg.to_array(args.number_rows)?, + &DataType::Int64, + &CastOptions { + safe: false, + ..Default::default() + }, + )?); } return range::gen_series_udf().invoke_with_args(args); } @@ -267,6 +273,35 @@ mod tests { use arrow::array::{ArrayRef, RecordBatch}; use std::collections::HashMap; + #[tokio::test] + async fn range_rejects_unsigned_overflow() { + let batch = RecordBatch::try_from_iter(vec![ + ("id", Arc::new(Int64Array::from(vec![1])) as ArrayRef), + ( + "large", + Arc::new(arrow::array::UInt64Array::from(vec![1_u64 << 63])) as ArrayRef, + ), + ]) + .unwrap(); + let config = GraphConfig::builder() + .node_table("Person", "id") + .build() + .unwrap(); + for expression in [ + "range(p.large, 1)", + "range(1, p.large)", + "range(1, 2, p.large)", + ] { + let error = CypherQuery::new(&format!("MATCH (p:Person) RETURN {expression} AS value")) + .unwrap() + .with_config(config.clone()) + .execute(HashMap::from([("Person".to_string(), batch.clone())]), None) + .await + .unwrap_err(); + assert!(error.to_string().contains("cast"), "{expression}: {error}"); + } + } + async fn run(expression: &str) -> crate::error::Result { let batch = RecordBatch::try_from_iter([("id", Arc::new(Int64Array::from(vec![1])) as ArrayRef)]) diff --git a/docs/cypher.md b/docs/cypher.md index 6e0128f..ff68643 100644 --- a/docs/cypher.md +++ b/docs/cypher.md @@ -278,7 +278,8 @@ Splitting an empty string returns `['']`. `last([])` returns null; `tail([])` and `reverse([])` return empty lists. `reverse` also accepts strings. `range(start, end [, step])` includes the end when reached, defaults to step 1, and accepts negative steps. A step pointing away from the end gives an empty list; zero steps and non-integer arguments -raise an error. Null arguments return null. +raise an error. Unsigned Arrow arguments outside Int64 range also raise an error; they are not +converted to null. Null arguments return null. ```python result = engine.execute(""" diff --git a/python/python/tests/test_scalar_functions.py b/python/python/tests/test_scalar_functions.py index 4e24460..b82afb3 100644 --- a/python/python/tests/test_scalar_functions.py +++ b/python/python/tests/test_scalar_functions.py @@ -8,6 +8,26 @@ from lance_csr import CypherEngine, GraphConfig +@pytest.mark.parametrize("use_csr", [False, True]) +@pytest.mark.parametrize( + "expression", ["range(p.value, 1)", "range(1, p.value)", "range(1, 2, p.value)"] +) +def test_range_unsigned_overflow_and_null(use_csr, expression): + config = GraphConfig.builder().node_table("Person", "id").build() + query = f"MATCH (p:Person) RETURN {expression} AS value" + for value in (None, 2**63): + engine = CypherEngine( + config, + {"Person": pa.table({"id": [1], "value": pa.array([value], pa.uint64())})}, + use_csr=use_csr, + ) + if value is None: + assert engine.execute(query).to_pylist() == [{"value": None}] + else: + with pytest.raises((ValueError, RuntimeError), match="cast|Cast"): + engine.execute(query) + + @pytest.fixture(params=[False, True], ids=["join", "csr"]) def engine(request): config = GraphConfig.builder().node_table("Person", "id").build() From d7093ea932ad217f99f3b721d6b469cfce1aaaba Mon Sep 17 00:00:00 2001 From: David Hughes Date: Sun, 25 Oct 2026 13:06:34 -0700 Subject: [PATCH 3/6] fix(cypher): preserve native list SQL exports Translate Cypher list size and inclusive range wrappers into faithful native SQL expressions before applying the inherited registry guard. Add string/null and List/LargeList/FixedSizeList SQL coverage and document runtime-only string exports. Submitted by David Hughes (@stugorf), with AI Collaboration. --- .../lance-graph/src/datafusion_planner/mod.rs | 1 + .../datafusion_planner/scalar_functions.rs | 28 +++++++++++++++++++ crates/lance-graph/src/query.rs | 12 ++++++-- docs/sql.md | 7 +++-- python/python/tests/test_to_sql.py | 20 +++++++++++++ 5 files changed, 63 insertions(+), 5 deletions(-) diff --git a/crates/lance-graph/src/datafusion_planner/mod.rs b/crates/lance-graph/src/datafusion_planner/mod.rs index 8be2325..c5f9caf 100644 --- a/crates/lance-graph/src/datafusion_planner/mod.rs +++ b/crates/lance-graph/src/datafusion_planner/mod.rs @@ -31,6 +31,7 @@ mod test_fixtures; // Re-export public types pub use analysis::{PlanningContext, QueryAnalysis, RelationshipInstance}; +pub(crate) use scalar_functions::sql_expr; use crate::config::GraphConfig; use crate::error::Result; diff --git a/crates/lance-graph/src/datafusion_planner/scalar_functions.rs b/crates/lance-graph/src/datafusion_planner/scalar_functions.rs index f46e54a..fd9511a 100644 --- a/crates/lance-graph/src/datafusion_planner/scalar_functions.rs +++ b/crates/lance-graph/src/datafusion_planner/scalar_functions.rs @@ -266,6 +266,34 @@ pub(super) fn call(name: &str, args: Vec) -> Expr { .call(args) } +/// SQL equivalents for the list operations supported before Cypher-specific wrappers. +pub(crate) fn sql_expr( + function: &datafusion::logical_expr::expr::ScalarFunction, + schema: &datafusion::common::DFSchema, +) -> Result> { + use datafusion::logical_expr::{expr_fn::cast, ExprSchemable}; + + if !function.func.inner().is::() { + return Ok(None); + } + let args = &function.args; + Ok(match function.name() { + "size" if list_field(&args[0].get_type(schema)?).is_some() => Some(cast( + length::array_length_udf().call(args.clone()), + DataType::Int64, + )), + "range" => Some( + range::gen_series_udf().call( + args.iter() + .cloned() + .map(|arg| cast(arg, DataType::Int64)) + .collect(), + ), + ), + _ => None, + }) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/lance-graph/src/query.rs b/crates/lance-graph/src/query.rs index 895326d..b99f200 100644 --- a/crates/lance-graph/src/query.rs +++ b/crates/lance-graph/src/query.rs @@ -388,10 +388,10 @@ impl CypherQuery { dialect: Option, ) -> Result { use datafusion::common::tree_node::{Transformed, TreeNode}; - use datafusion::common::{not_impl_err, ScalarValue}; + use datafusion::common::{not_impl_err, DFSchema, ScalarValue}; use datafusion::logical_expr::expr_rewriter::NamePreserver; use datafusion::logical_expr::registry::FunctionRegistry; - use datafusion::logical_expr::{lit, Expr}; + use datafusion::logical_expr::{lit, utils::merge_schema, Expr, LogicalPlan}; use std::sync::Arc; let dialect = dialect.unwrap_or_default(); @@ -418,11 +418,19 @@ impl CypherQuery { // An SQL function with the same name need not implement Cypher semantics. let optimized_plan = optimized_plan .transform_up_with_subqueries(|plan| { + let schema = if let LogicalPlan::TableScan(scan) = &plan { + DFSchema::try_from_qualified_schema(scan.table_name.clone(), &scan.source.schema())? + } else { + merge_schema(&plan.inputs()) + }; let names = NamePreserver::new(&plan); plan.map_expressions(|expr| { let name = names.save(&expr); let rewritten = expr.transform_up(|expr| { if let Expr::ScalarFunction(function) = &expr { + if let Some(native) = crate::datafusion_planner::sql_expr(function, &schema)? { + return Ok(Transformed::yes(native)); + } if !ctx.udf(function.name()).is_ok_and(|registered| registered == function.func) { return not_impl_err!( "SQL export of Cypher function '{}' is unsupported; execute the Cypher query directly", diff --git a/docs/sql.md b/docs/sql.md index ed70018..85a5b1c 100644 --- a/docs/sql.md +++ b/docs/sql.md @@ -84,8 +84,8 @@ print(query.to_sql(graph, SqlDialect.PostgreSql)) Export is experimental and supports a subset of Cypher execution. Runtime-only Cypher functions raise `NotImplementedError` when no faithful SQL expression is available. This includes math wrappers such -as `round` and `abs` when they depend on table values. Cypher `round(-1.5)` produces `-1.0`; -DataFusion SQL `round(-1.5)` produces `-2.0`. +as `round` and `abs`, and string functions such as `left` and string `size`, when they depend on +table values. Cypher `round(-1.5)` produces `-1.0`; DataFusion SQL `round(-1.5)` produces `-2.0`. Integer `abs` also normalizes its result to Int64 in Cypher, while SQL preserves the input width. Use `CypherQuery.execute` or `CypherEngine.execute` for these expressions. @@ -93,7 +93,8 @@ Constant-folded typed nulls use DataFusion `arrow_cast` in the Default dialect s result types are preserved. Other dialects raise `NotImplementedError` for typed-null literals; a bare Cypher `null` remains an untyped SQL `NULL`. -Existing native SQL functions remain exportable. Constant expressions that the optimizer +List `size` and inclusive `range` retain SQL export support through native expressions, including +null arguments, integer casts and the optional range step. Constant expressions that the optimizer evaluates completely can also export their resulting literals. These checks establish DataFusion semantics; other SQL dialects still require validation on the destination engine. diff --git a/python/python/tests/test_to_sql.py b/python/python/tests/test_to_sql.py index b1daa89..4c4e4d5 100644 --- a/python/python/tests/test_to_sql.py +++ b/python/python/tests/test_to_sql.py @@ -18,6 +18,15 @@ def scalar_export_env(): "number": pa.array([-1.5, None, 2.0], pa.float32()), "integer": pa.array([-100, None, 2], pa.int8()), "text": pa.array(["a猫😀z", None, ""], pa.large_string()), + "items": pa.array([[1, None], None, []], pa.list_(pa.int64())), + "nested": pa.array( + [[[1, 2], [3]], None, []], pa.list_(pa.list_(pa.int64())) + ), + "large_items": pa.array( + [[1, None], None, []], pa.large_list(pa.int64()) + ), + "pair": pa.array([[1, 2], None, [3, 4]], pa.list_(pa.int64(), 2)), + "end": pa.array([3, None, -1], pa.int8()), } ) } @@ -27,6 +36,13 @@ def scalar_export_env(): @pytest.mark.parametrize( "expression", [ + "size(p.items)", + "size(p.large_items)", + "size(p.pair)", + "size(p.nested)", + "range(1, p.end)", + "range(p.end, 1, -1)", + "size(range(1, p.end))", "toUpper(p.text)", "round(-1.5)", "abs(-100)", @@ -38,6 +54,7 @@ def scalar_export_env(): "ceil(null)", "toUpper(null)", "null", + "left('a猫😀z', 3)", ], ) def test_sql_export_preserves_scalar_values_and_types(scalar_export_env, expression): @@ -56,6 +73,9 @@ def test_sql_export_preserves_scalar_values_and_types(scalar_export_env, express [ ("round(p.number)", "round", [-1.0, None, 2.0]), ("abs(p.integer)", "abs", [100, None, 2]), + ("left(p.text, 3)", "left", ["a猫😀", None, ""]), + ("size(p.text)", "size", [4, None, 0]), + ("reverse(p.text)", "reverse", ["z😀猫a", None, ""]), ], ) def test_sql_export_rejects_runtime_only_cypher_functions( From d44596967609bce35fad34acafe68753a2913383 Mon Sep 17 00:00:00 2001 From: David Hughes Date: Sun, 25 Oct 2026 14:32:20 -0700 Subject: [PATCH 4/6] test(cypher): cover null string and list SQL types Compare real Cypher and SqlEngine results for null string/list functions and typed empty-list extraction, preserving exact Arrow output types after constant folding. Submitted by David Hughes (@stugorf), with AI Collaboration. --- python/python/tests/test_to_sql.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/python/python/tests/test_to_sql.py b/python/python/tests/test_to_sql.py index 4c4e4d5..4954ba2 100644 --- a/python/python/tests/test_to_sql.py +++ b/python/python/tests/test_to_sql.py @@ -55,6 +55,17 @@ def scalar_export_env(): "toUpper(null)", "null", "left('a猫😀z', 3)", + "left(null, 1)", + "right(null, 1)", + "substring(null, 0)", + "replace(null, 'x', 'y')", + "split(null, ',')", + "trim(null)", + "reverse(null)", + "size(null)", + "range(null, 3)", + "last(range(1, 0))", + "tail(null)", ], ) def test_sql_export_preserves_scalar_values_and_types(scalar_export_env, expression): From d31152b9d237fa159b00404596f6cd27d3203653 Mon Sep 17 00:00:00 2001 From: David Hughes Date: Mon, 26 Oct 2026 09:51:20 -0700 Subject: [PATCH 5/6] Add map/entity keys and node labels (#44) * feat(cypher): add map and entity keys with node labels Implement keys for maps and exposed entity property columns, preserving null map fields while omitting absent entity properties. Resolve labels from canonical configured node bindings and preserve optional nulls and WITH aliases. Reject mixed-label bindings explicitly. Validation: 344 Rust library tests, 32 focused Python cases, all 615 Python tests, workspace all-target clippy, Python lint, executable documentation, strict docs build, spelling, formatting, and whitespace checks pass. Part of #20; mixed-label labels() remains an explicit engine boundary. Submitted by David Hughes (@stugorf), with AI Collaboration. * Add ordered reduce and list predicates (#45) * feat(cypher): add ordered reduce and list predicates Part of #20. Keep fold variables local, preserve list order and null semantics, and correlate graph predicates with iterator bindings. Document supported forms and explicit body limits. * docs(cypher): explain list fold AST migration Document the exhaustive-match and local-scope requirements for Rust consumers of the new fold expression variants. Submitted by David Hughes (@stugorf), with AI Collaboration. --- crates/lance-graph/README.md | 7 + crates/lance-graph/src/ast.rs | 79 +++++ .../builder/aggregate_ops.rs | 36 +- .../datafusion_planner/builder/basic_ops.rs | 22 +- .../builder/comprehension_ops.rs | 22 +- .../builder/join_builder.rs | 4 +- .../src/datafusion_planner/expression.rs | 139 +++++--- .../src/datafusion_planner/list_folds.rs | 328 ++++++++++++++++++ .../lance-graph/src/datafusion_planner/mod.rs | 1 + .../datafusion_planner/scalar_functions.rs | 92 ++++- crates/lance-graph/src/logical_plan.rs | 154 +++++++- .../lance-graph/src/parameter_substitution.rs | 16 + crates/lance-graph/src/parser.rs | 90 +++++ crates/lance-graph/src/pattern_order.rs | 19 +- crates/lance-graph/src/semantic.rs | 88 ++++- docs/cypher.md | 64 +++- python/python/tests/test_entity_functions.py | 137 ++++++++ python/python/tests/test_list_folds.py | 240 +++++++++++++ 18 files changed, 1443 insertions(+), 95 deletions(-) create mode 100644 crates/lance-graph/src/datafusion_planner/list_folds.rs create mode 100644 python/python/tests/test_entity_functions.py create mode 100644 python/python/tests/test_list_folds.py 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}] From dd23c1a7bc4e8ad40ea942c2a6b80cfd1a3a281f Mon Sep 17 00:00:00 2001 From: David Hughes Date: Mon, 26 Oct 2026 09:51:29 -0700 Subject: [PATCH 6/6] fix(query): resolve shadowing output aliases in sort (#49) Resolve all RETURN expressions before applying scalar aliases to the ORDER BY binding scope. Preserve whole-node projection behavior. Add focused planner and CSR/join Python regressions for issue #21. Document alias scope in the Cypher guide. Validation: Rust formatting, Python lint, strict docs build, spelling, and whitespace checks pass. Published 0.2.0 reproduction: 14 failures, 4 existing passes. Current-source runtime tests remain pending the coordinated build. Implemented by David Hughes with assistance from Codex. --- crates/lance-graph/src/logical_plan.rs | 49 ++++++++++++++ docs/cypher.md | 8 +-- python/python/tests/test_alias_shadowing.py | 74 +++++++++++++++++++++ 3 files changed, 127 insertions(+), 4 deletions(-) create mode 100644 python/python/tests/test_alias_shadowing.py diff --git a/crates/lance-graph/src/logical_plan.rs b/crates/lance-graph/src/logical_plan.rs index 533eaba..9b7822c 100644 --- a/crates/lance-graph/src/logical_plan.rs +++ b/crates/lance-graph/src/logical_plan.rs @@ -1532,6 +1532,17 @@ impl<'a> LogicalPlanner<'a> { } } + // RETURN aliases shadow input entities in ORDER BY, after every item is bound. + // Whole-node aliases expand to `alias.property`, so keep their entity binding. + self.variables.retain(|variable, _| { + !projections.iter().any(|projection| { + projection + .alias + .as_ref() + .is_some_and(|alias| alias.eq_ignore_ascii_case(variable)) + }) + }); + let mut plan = LogicalOperator::Project { input: Box::new(input), projections, @@ -1676,6 +1687,44 @@ mod tests { use super::*; use crate::{parser::parse_cypher_query, NodeMapping}; + #[test] + fn return_alias_shadows_node_only_after_projection() { + let config = GraphConfig::builder() + .node_table("Person", "id") + .build() + .unwrap(); + let query = + parse_cypher_query("MATCH (p:Person) RETURN p.name AS P, p.id AS id ORDER BY p") + .unwrap(); + let LogicalOperator::Sort { input, sort_items } = + LogicalPlanner::new(&config).plan(&query).unwrap() + else { + panic!("expected sort"); + }; + assert_eq!( + sort_items[0].expression, + ValueExpression::Variable("p".into()) + ); + let LogicalOperator::Project { projections, .. } = *input else { + panic!("expected projection"); + }; + assert_eq!( + projections[1].expression, + ValueExpression::Property(PropertyRef { + variable: "p".into(), + property: "id".into(), + }) + ); + + let query = parse_cypher_query("MATCH (p:Person) RETURN p AS p ORDER BY p").unwrap(); + let LogicalOperator::Sort { sort_items, .. } = + LogicalPlanner::new(&config).plan(&query).unwrap() + else { + panic!("expected sort"); + }; + assert_eq!(sort_items[0].expression, projections[1].expression); + } + #[test] fn test_relationship_query_logical_plan_structure() { let query_text = r#"MATCH (p:Person {name: "Alice"})-[:KNOWS]->(f:Person) WHERE f.age > 30 RETURN f.name"#; diff --git a/docs/cypher.md b/docs/cypher.md index e3aa541..6cd4476 100644 --- a/docs/cypher.md +++ b/docs/cypher.md @@ -386,10 +386,10 @@ print(result.to_pylist()) Keywords, function names, variables, labels, relationship types and property names are all case-insensitive. `match (P:person) return p.NAME` works, and the output column keeps the spelling you wrote. -!!! bug "Known issue: an alias that shadows a variable" - - `RETURN p.name AS p` reuses the variable name `p` as an output alias, which can fail to plan once the - query also aggregates or uses `OPTIONAL MATCH`. Pick a different alias, such as `AS name`. +An output alias may reuse an input variable's name. Expressions in the same `RETURN` still use the input +variable; the following `ORDER BY` uses the output alias. For example, +`RETURN p.name AS p, p.id AS id ORDER BY p` returns names and IDs, sorted by name. This also works with +aggregation and `OPTIONAL MATCH`. ## Errors diff --git a/python/python/tests/test_alias_shadowing.py b/python/python/tests/test_alias_shadowing.py new file mode 100644 index 0000000..39f2b99 --- /dev/null +++ b/python/python/tests/test_alias_shadowing.py @@ -0,0 +1,74 @@ +"""Output aliases shadow input nodes only after their expressions are evaluated.""" + +import pytest +from lance_csr import CypherEngine +from test_engine_csr import CONFIG, KNOWS, PEOPLE + + +@pytest.mark.parametrize("use_csr", [False, True]) +@pytest.mark.parametrize( + ("query", "columns", "expected"), + [ + ( + "MATCH (p:Person) RETURN p.name AS p ORDER BY p", + ["p"], + [(name,) for name in "abcdef"], + ), + ( + "MATCH (p:Person) RETURN p.name AS p, count(*) AS n ORDER BY p", + ["p", "n"], + [(name, 1) for name in "abcdef"], + ), + ( + "MATCH (p:Person) OPTIONAL MATCH (p)-[:KNOWS]->(f:Person) " + "RETURN p.name AS p, f.name AS f ORDER BY p, f", + ["p", "f"], + [ + ("a", "b"), + ("a", "b"), + ("a", "c"), + ("b", "c"), + ("c", "c"), + ("d", None), + ("e", "a"), + ("f", None), + ], + ), + ( + "MATCH (p:Person) OPTIONAL MATCH (p)-[:KNOWS]->(f:Person) " + "RETURN p.name AS p, count(f) AS n ORDER BY p", + ["p", "n"], + list(zip("abcdef", [3, 1, 1, 0, 1, 0], strict=True)), + ), + ( + "MATCH (p:Person) RETURN p.name AS p, p.id AS id ORDER BY p", + ["p", "id"], + list(zip("abcdef", range(1, 7), strict=True)), + ), + ( + "MATCH (p:Person) RETURN p.name AS P ORDER BY p DESC", + ["p"], + [(name,) for name in "fedcba"], + ), + ( + "MATCH (p:Person) RETURN p.name AS p ORDER BY toUpper(p) DESC", + ["p"], + [(name,) for name in "fedcba"], + ), + ( + "MATCH (p:Person) WITH p.name AS p ORDER BY p RETURN p", + ["p"], + [(name,) for name in "abcdef"], + ), + ( + "MATCH (p:Person) RETURN p AS p ORDER BY p", + ["p.id"], + [(key,) for key in range(1, 7)], + ), + ], +) +def test_output_alias_shadows_node(use_csr, query, columns, expected): + engine = CypherEngine(CONFIG, {"Person": PEOPLE, "KNOWS": KNOWS}, use_csr=use_csr) + result = engine.execute(query) + assert result.column_names == columns + assert [tuple(row.values()) for row in result.to_pylist()] == expected