From d0a5cd28280eb5b59fea08cc60285365fe40ef9c Mon Sep 17 00:00:00 2001 From: David Hughes Date: Sun, 25 Oct 2026 11:34:17 -0700 Subject: [PATCH 1/6] feat(cypher): add numeric scalar functions Implement the math-function submission for #20. Reuse DataFusion kernels for abs, floor, ceil, and sqrt; enforce numeric arguments and Cypher rounding ties, nulls, return types, and overflow errors. Document the finite supported signatures with executable examples. Validation: 343 Rust library tests and Python static checks pass. Exact-source Python runtime checks and workspace clippy are in progress; release evidence is tracked separately. Submitted by David Hughes (@stugorf), with AI Collaboration. --- crates/lance-graph/Cargo.toml | 1 + .../src/datafusion_planner/expression.rs | 3 + .../src/datafusion_planner/math_functions.rs | 164 ++++++++++++++++++ .../lance-graph/src/datafusion_planner/mod.rs | 1 + crates/lance-graph/src/semantic.rs | 5 +- docs/cypher.md | 20 ++- python/python/tests/test_math_functions.py | 119 +++++++++++++ 7 files changed, 308 insertions(+), 5 deletions(-) create mode 100644 crates/lance-graph/src/datafusion_planner/math_functions.rs create mode 100644 python/python/tests/test_math_functions.py diff --git a/crates/lance-graph/Cargo.toml b/crates/lance-graph/Cargo.toml index b054f60..aa40f55 100644 --- a/crates/lance-graph/Cargo.toml +++ b/crates/lance-graph/Cargo.toml @@ -18,6 +18,7 @@ arrow-schema = "58" async-trait = "0.1" datafusion = { version = "54", default-features = false, features = [ "nested_expressions", + "math_expressions", "regex_expressions", "unicode_expressions", "crypto_expressions", diff --git a/crates/lance-graph/src/datafusion_planner/expression.rs b/crates/lance-graph/src/datafusion_planner/expression.rs index 18545b6..e57229e 100644 --- a/crates/lance-graph/src/datafusion_planner/expression.rs +++ b/crates/lance-graph/src/datafusion_planner/expression.rs @@ -193,6 +193,9 @@ pub(crate) fn to_df_value_expr(expr: &ValueExpression) -> Expr { VE::Variable(path) => col(qualify_column(path, &name.to_lowercase())), _ => Expr::Literal(datafusion::scalar::ScalarValue::Null, None), }, + "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]), diff --git a/crates/lance-graph/src/datafusion_planner/math_functions.rs b/crates/lance-graph/src/datafusion_planner/math_functions.rs new file mode 100644 index 0000000..e82f427 --- /dev/null +++ b/crates/lance-graph/src/datafusion_planner/math_functions.rs @@ -0,0 +1,164 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The lance-csr Authors + +//! Strict Cypher numeric arguments and rounding over DataFusion's math kernels. + +use std::sync::Arc; + +use arrow::array::{AsArray, Float64Array}; +use arrow::compute::{cast_with_options, CastOptions}; +use arrow_schema::{DataType, Field}; +use datafusion::common::plan_err; +use datafusion::error::Result; +use datafusion::functions::math; +use datafusion::logical_expr::{ + ColumnarValue, Expr, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, Volatility, +}; + +#[derive(Debug, PartialEq, Eq, Hash)] +struct CypherMath { + name: String, + signature: Signature, +} + +impl ScalarUDFImpl for CypherMath { + fn name(&self) -> &str { + &self.name + } + + fn signature(&self) -> &Signature { + &self.signature + } + + fn return_type(&self, types: &[DataType]) -> Result { + let input = &types[0]; + if !input.is_integer() && !input.is_floating() && !input.is_null() { + return plan_err!("{} does not accept {input} for argument 1", self.name); + } + Ok(if self.name == "abs" && input.is_integer() { + DataType::Int64 + } else { + DataType::Float64 + }) + } + + fn invoke_with_args(&self, mut args: ScalarFunctionArgs) -> Result { + let output_type = args.return_type().clone(); + let input = cast_with_options( + &args.args[0].to_array(args.number_rows)?, + &output_type, + &CastOptions { + safe: false, + ..Default::default() + }, + )?; + if self.name == "round" { + let rounded = input + .as_primitive::() + .iter() + .map(|n| { + n.map(|n| { + // Cypher rounds ties toward positive infinity and returns positive zero. + let rounded = if n.fract() == -0.5 { + n.ceil() + } else { + n.round() + }; + if rounded == 0.0 { + 0.0 + } else { + rounded + } + }) + }) + .collect::(); + return Ok(ColumnarValue::Array(Arc::new(rounded))); + } + let udf = match self.name.as_str() { + "abs" => math::abs(), + "floor" => math::floor(), + "ceil" => math::ceil(), + "sqrt" => math::sqrt(), + _ => unreachable!("validated math function"), + }; + args.args = vec![ColumnarValue::Array(input)]; + args.arg_fields = vec![Arc::new(Field::new("arg", output_type, true))]; + udf.invoke_with_args(args) + } +} + +pub(super) fn call(name: &str, args: Vec) -> Expr { + ScalarUDF::new_from_impl(CypherMath { + name: name.to_lowercase(), + signature: Signature::any(1, Volatility::Immutable), + }) + .call(args) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{config::GraphConfig, query::CypherQuery}; + use arrow::array::{Array, ArrayRef, Int64Array, 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_math_rounding_and_overflow() { + for (expression, expected) in [ + ("abs(-1.5)", 1.5), + ("round(-1.5)", -1.0), + ("round(1.5)", 2.0), + ("round(4503599627370495.5)", 4503599627370496.0), + ("round(4503599627370497.0)", 4503599627370497.0), + ("floor(-1.5)", -2.0), + ("ceil(-1.5)", -1.0), + ("sqrt(9)", 3.0), + ] { + let out = run(expression).await.unwrap(); + assert_eq!( + out.column(0) + .as_primitive::() + .value(0), + expected, + "{expression}" + ); + } + let out = run("abs(-5)").await.unwrap(); + assert_eq!( + out.column(0) + .as_primitive::() + .value(0), + 5 + ); + assert!(run("abs(-9223372036854775807 - 1)").await.is_err()); + } + + #[tokio::test] + async fn cypher_math_validates_types_arity_and_nulls() { + for name in ["abs", "round", "floor", "ceil", "sqrt"] { + let out = run(&format!("{name}(null)")).await.unwrap(); + assert_eq!(out.column(0).null_count(), 1, "{name}"); + for arg in ["'1'", "true", "[1]", "1, 2"] { + assert!( + run(&format!("{name}({arg})")).await.is_err(), + "{name}({arg})" + ); + } + } + } +} diff --git a/crates/lance-graph/src/datafusion_planner/mod.rs b/crates/lance-graph/src/datafusion_planner/mod.rs index 6c8d8ab..9f66573 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 math_functions; mod scan_ops; mod udf; pub mod vector_ops; diff --git a/crates/lance-graph/src/semantic.rs b/crates/lance-graph/src/semantic.rs index b8cf649..f0a2227 100644 --- a/crates/lance-graph/src/semantic.rs +++ b/crates/lance-graph/src/semantic.rs @@ -438,7 +438,8 @@ impl SemanticAnalyzer { let function_name = name.to_lowercase(); // Validate arity and known functions match function_name.as_str() { - "tolower" | "lower" | "toupper" | "upper" => { + "tolower" | "lower" | "toupper" | "upper" | "abs" | "round" | "floor" + | "ceil" | "sqrt" => { if args.len() != 1 { return Err(GraphError::PlanError { message: format!( @@ -506,7 +507,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. 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, abs, round, floor, ceil, sqrt. Supported aggregate functions: COUNT, SUM, AVG, MIN, MAX, COLLECT.", name ), location: snafu::Location::new(file!(), line!(), column!()), diff --git a/docs/cypher.md b/docs/cypher.md index 98bbd25..60c1d21 100644 --- a/docs/cypher.md +++ b/docs/cypher.md @@ -261,13 +261,27 @@ RETURN length(p), nodes(p) | Strings | `toLower` / `lower`, `toUpper` / `upper`, `toString` | | Conversion | `toInteger`, `toFloat`, `coalesce` | | Lists | `size`, `head`, `range(start, end)` (inclusive, no step) | +| 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) | -Any other function raises `NotImplementedError: Cypher function 'name' is not implemented`. Notably -missing: math functions (`abs`, `round`, `sqrt`, ...), most string and list functions, `reduce`, the list -predicates `any` / `all` / `none` / `single`, and `relationships(p)`. +`abs` preserves integer results; the other math functions return floating point values. Integer +absolute-value overflow raises an error. `round` accepts one argument, with ties toward positive +infinity (`round(-1.5)` is `-1.0`), as specified in the +[Neo4j 4.4 numeric reference](https://neo4j.com/docs/cypher-manual/4.4/functions/mathematical-numeric/). +`sqrt` of a negative number returns NaN. These functions return null for null input and reject +non-numeric values, including strings and booleans. + +```python +result = engine.execute("RETURN abs(-5) AS magnitude, round(-1.5) AS rounded, sqrt(9) AS root") +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)`. ### Example: a little of everything diff --git a/python/python/tests/test_math_functions.py b/python/python/tests/test_math_functions.py new file mode 100644 index 0000000..8fe1db9 --- /dev/null +++ b/python/python/tests/test_math_functions.py @@ -0,0 +1,119 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The lance-csr Authors + +"""Cypher math types, nulls, and rounding through both execution backends.""" + +import math + +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").build() + people = pa.table( + { + "id": [1, 2, 3], + "integer": pa.array([-128, None, 2], type=pa.int8()), + "number": pa.array([-1.5, None, 2.0], type=pa.float32()), + } + ) + return CypherEngine(config, {"Person": people}, use_csr=request.param) + + +@pytest.mark.parametrize( + ("expression", "expected"), + [ + ("abs(-5)", 5), + ("abs(-1.5)", 1.5), + ("round(-1.5)", -1.0), + ("round(1.5)", 2.0), + ("round(4503599627370495.5)", 4503599627370496.0), + ("round(4503599627370497.0)", 4503599627370497.0), + ("floor(-1.5)", -2.0), + ("ceil(-1.5)", -1.0), + ("sqrt(9)", 3.0), + ] + + [(f"{name}(null)", None) for name in ("abs", "round", "floor", "ceil", "sqrt")], +) +def test_math_value(engine, expression, expected): + result = engine.execute(f"RETURN {expression} AS value") + assert result.to_pylist() == [{"value": expected}] + assert result.schema.field("value").type == ( + pa.int64() if isinstance(expected, int) else pa.float64() + ) + + +@pytest.mark.parametrize("name", ["abs", "round", "floor", "ceil", "sqrt"]) +@pytest.mark.parametrize("argument", ["'1'", "true", "[1]", "{x: 1}", "", "1, 2"]) +def test_math_invalid_arguments(engine, name, argument): + with pytest.raises((ValueError, RuntimeError)): + engine.execute(f"RETURN {name}({argument}) AS value") + + +def test_math_integer_overflow(engine): + with pytest.raises((ValueError, RuntimeError), match="overflow"): + engine.execute("RETURN abs(-9223372036854775807 - 1) AS value") + + +def test_math_arrow_columns_and_nulls(engine): + result = engine.execute( + "MATCH (p:Person) RETURN p.id AS id, abs(p.integer) AS magnitude, " + "abs(p.number) AS floating, round(p.number) AS rounded, " + "floor(p.number) AS floored, ceil(p.number) AS ceiling, " + "sqrt(abs(p.integer)) AS root ORDER BY id" + ) + assert result.to_pylist() == [ + { + "id": 1, + "magnitude": 128, + "floating": 1.5, + "rounded": -1.0, + "floored": -2.0, + "ceiling": -1.0, + "root": math.sqrt(128), + }, + { + "id": 2, + "magnitude": None, + "floating": None, + "rounded": None, + "floored": None, + "ceiling": None, + "root": None, + }, + { + "id": 3, + "magnitude": 2, + "floating": 2.0, + "rounded": 2.0, + "floored": 2.0, + "ceiling": 2.0, + "root": math.sqrt(2), + }, + ] + assert result.schema.field("magnitude").type == pa.int64() + for column in ("floating", "rounded", "floored", "ceiling", "root"): + assert result.schema.field(column).type == pa.float64() + + +def test_math_nan_and_signed_zero(engine): + result = engine.execute( + "RETURN sqrt(-1) AS root, round(-0.5) AS zero, round($infinity) AS infinity", + {"infinity": math.inf}, + ).to_pylist()[0] + assert math.isnan(result["root"]) + assert math.copysign(1, result["zero"]) == 1 + assert result["infinity"] == math.inf + + +def test_math_out_of_range_unsigned_integer(): + config = GraphConfig.builder().node_table("Person", "id").build() + engine = CypherEngine( + config, + {"Person": pa.table({"id": [1], "large": pa.array([2**63], type=pa.uint64())})}, + ) + with pytest.raises((ValueError, RuntimeError), match="cast|Cast"): + engine.execute("MATCH (p:Person) RETURN abs(p.large) AS value") From 1297119a346c22b662a0cdbc8fbc383f3f6543ef Mon Sep 17 00:00:00 2001 From: David Hughes Date: Sun, 25 Oct 2026 11:41:43 -0700 Subject: [PATCH 2/6] test(cypher): cover non-finite Arrow math inputs Exercise infinity and NaN through Arrow columns while preserving the existing rejection of non-finite parameter values. Document out-of-range unsigned input errors. Validation: 343 Rust library tests, workspace all-target clippy, 96 focused and 548 total Python tests, Python static checks, strict docs, spelling, and executable documentation pass. Submitted by David Hughes (@stugorf), with AI Collaboration. --- docs/cypher.md | 2 +- python/python/tests/test_math_functions.py | 23 +++++++++++++++++----- 2 files changed, 19 insertions(+), 6 deletions(-) diff --git a/docs/cypher.md b/docs/cypher.md index 60c1d21..f202f62 100644 --- a/docs/cypher.md +++ b/docs/cypher.md @@ -267,7 +267,7 @@ RETURN length(p), nodes(p) | Vectors | `vector_distance(a, b, metric)`, `vector_similarity(a, b, metric)`; see [Vector search](vector-search.md) | `abs` preserves integer results; the other math functions return floating point values. Integer -absolute-value overflow raises an error. `round` accepts one argument, with ties toward positive +absolute-value overflow and unsigned inputs outside Int64 range raise an error. `round` accepts one argument, with ties toward positive infinity (`round(-1.5)` is `-1.0`), as specified in the [Neo4j 4.4 numeric reference](https://neo4j.com/docs/cypher-manual/4.4/functions/mathematical-numeric/). `sqrt` of a negative number returns NaN. These functions return null for null input and reject diff --git a/python/python/tests/test_math_functions.py b/python/python/tests/test_math_functions.py index 8fe1db9..4cb066c 100644 --- a/python/python/tests/test_math_functions.py +++ b/python/python/tests/test_math_functions.py @@ -100,13 +100,11 @@ def test_math_arrow_columns_and_nulls(engine): def test_math_nan_and_signed_zero(engine): - result = engine.execute( - "RETURN sqrt(-1) AS root, round(-0.5) AS zero, round($infinity) AS infinity", - {"infinity": math.inf}, - ).to_pylist()[0] + result = engine.execute("RETURN sqrt(-1) AS root, round(-0.5) AS zero").to_pylist()[ + 0 + ] assert math.isnan(result["root"]) assert math.copysign(1, result["zero"]) == 1 - assert result["infinity"] == math.inf def test_math_out_of_range_unsigned_integer(): @@ -117,3 +115,18 @@ def test_math_out_of_range_unsigned_integer(): ) with pytest.raises((ValueError, RuntimeError), match="cast|Cast"): engine.execute("MATCH (p:Person) RETURN abs(p.large) AS value") + + +def test_math_non_finite_arrow_values(): + config = GraphConfig.builder().node_table("Person", "id").build() + engine = CypherEngine( + config, + {"Person": pa.table({"id": [1, 2], "number": [math.inf, math.nan]})}, + ) + rows = engine.execute( + "MATCH (p:Person) RETURN p.id AS id, round(p.number) AS rounded, " + "abs(p.number) AS magnitude, sqrt(p.number) AS root ORDER BY id" + ).to_pylist() + for name in ("rounded", "magnitude", "root"): + assert rows[0][name] == math.inf + assert math.isnan(rows[1][name]) From f21c5b92e662d7873975461ebbd79a9c0415bc3b Mon Sep 17 00:00:00 2001 From: David Hughes Date: Sun, 25 Oct 2026 13:05:45 -0700 Subject: [PATCH 3/6] fix(cypher): guard math SQL export semantics Reject runtime-only scalar implementations after constant folding instead of exporting same-named SQL functions with different Cypher values or types. Preserve native SQL functions, document NotImplementedError, and test real math SQL comparisons. Submitted by David Hughes (@stugorf), with AI Collaboration. --- crates/lance-graph-python/src/graph.rs | 4 +- crates/lance-graph/src/query.rs | 61 +++++++++++++++++++ docs/api-reference.md | 3 +- docs/sql.md | 13 +++- python/python/tests/test_to_sql.py | 82 +++++++++++++++++++++++++- 5 files changed, 159 insertions(+), 4 deletions(-) diff --git a/crates/lance-graph-python/src/graph.rs b/crates/lance-graph-python/src/graph.rs index eadcabd..85cb999 100644 --- a/crates/lance-graph-python/src/graph.rs +++ b/crates/lance-graph-python/src/graph.rs @@ -588,7 +588,7 @@ impl CypherQuery { Ok(dict.unbind()) } - /// Convert query to SQL + /// Export supported Cypher expressions as SQL /// /// Parameters /// ---------- @@ -604,6 +604,8 @@ impl CypherQuery { /// /// Raises /// ------ + /// NotImplementedError + /// If a runtime-only Cypher function has no faithful SQL translation /// RuntimeError /// If SQL generation fails #[pyo3(signature = (datasets, dialect=None))] diff --git a/crates/lance-graph/src/query.rs b/crates/lance-graph/src/query.rs index f617431..4cb85ff 100644 --- a/crates/lance-graph/src/query.rs +++ b/crates/lance-graph/src/query.rs @@ -377,11 +377,19 @@ impl CypherQuery { /// /// # Returns /// A SQL string representing the query in the specified dialect + /// + /// # Errors + /// Returns `GraphError::UnsupportedFeature` for runtime-only Cypher functions + /// without a faithful SQL translation. Constant expressions may be folded first. pub async fn to_sql( &self, datasets: HashMap, dialect: Option, ) -> Result { + use datafusion::common::not_impl_err; + use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion}; + use datafusion::logical_expr::registry::FunctionRegistry; + use datafusion::logical_expr::Expr; use std::sync::Arc; let dialect = dialect.unwrap_or_default(); @@ -405,6 +413,28 @@ impl CypherQuery { location: snafu::Location::new(file!(), line!(), column!()), })?; + // An SQL function with the same name need not implement Cypher semantics. + optimized_plan + .apply_with_subqueries(|plan| { + plan.apply_expressions(|expr| { + expr.apply(|expr| { + if let Expr::ScalarFunction(function) = expr { + 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", + function.name() + ); + } + } + Ok(TreeNodeRecursion::Continue) + }) + }) + }) + .map_err(|error| GraphError::UnsupportedFeature { + feature: error.to_string(), + location: snafu::Location::new(file!(), line!(), column!()), + })?; + // Unparse to SQL using the specified dialect let dialect_unparser = dialect.unparser(); let unparser = dialect_unparser.as_unparser(); @@ -1997,6 +2027,37 @@ mod tests { assert!(sql.contains("p.name")); } + #[tokio::test] + async fn test_to_sql_rejects_runtime_only_cypher_function() { + let batch = RecordBatch::try_from_iter(vec![ + ( + "id", + Arc::new(arrow_array::Int64Array::from(vec![1])) as arrow_array::ArrayRef, + ), + ( + "number", + Arc::new(arrow_array::Float64Array::from(vec![-1.5])) as arrow_array::ArrayRef, + ), + ]) + .unwrap(); + let query = CypherQuery::new("MATCH (p:Person) RETURN round(p.number) AS value") + .unwrap() + .with_config( + GraphConfig::builder() + .node_table("Person", "id") + .build() + .unwrap(), + ); + let error = query + .to_sql(HashMap::from([("Person".to_string(), batch)]), None) + .await + .unwrap_err(); + assert!(matches!(error, GraphError::UnsupportedFeature { .. })); + assert!(error + .to_string() + .contains("SQL export of Cypher function 'round'")); + } + async fn write_lance_dataset(path: &std::path::Path, batch: arrow_array::RecordBatch) { use arrow_array::{RecordBatch, RecordBatchIterator}; use lance::dataset::{Dataset, WriteParams}; diff --git a/docs/api-reference.md b/docs/api-reference.md index d28bb6b..eaa8fc0 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -187,7 +187,8 @@ A parsed Cypher query. Raises `ValueError` if `query_text` doesn't parse. Builde : The plans, as for `CypherEngine.explain`. `#!python to_sql(datasets: dict, dialect: SqlDialect | None = None) -> str` -: The equivalent SQL, as joins. See [From Cypher to SQL](sql.md#from-cypher-to-sql). +: Exports supported expressions as SQL joins. Raises `NotImplementedError` for runtime-only Cypher + functions without a faithful SQL translation. See [From Cypher to SQL](sql.md#from-cypher-to-sql). `#!python execute_with_vector_rerank(datasets: dict, vector_search: VectorSearch) -> pyarrow.Table` : Runs the query, then reranks by vector similarity. diff --git a/docs/sql.md b/docs/sql.md index 57f60c3..9035a09 100644 --- a/docs/sql.md +++ b/docs/sql.md @@ -59,7 +59,7 @@ The SQL dialect is DataFusion's; see the ## From Cypher to SQL -`CypherQuery.to_sql(datasets, dialect=None)` shows the SQL that a Cypher query is equivalent to. It's handy +`CypherQuery.to_sql(datasets, dialect=None)` exports supported Cypher expressions as SQL. It's handy for understanding a query, or for running a graph pattern on another SQL engine: ```python @@ -82,6 +82,17 @@ print(query.to_sql(graph, SqlDialect.PostgreSql)) `SqlDialect` has `Default` (DataFusion's), `Spark`, `PostgreSql`, `MySql` and `Sqlite`. +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`. +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. + +Existing native SQL functions remain exportable. 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. + !!! info "This is the join path" The generated SQL expresses each hop as a join. It describes what the query computes, not how diff --git a/python/python/tests/test_to_sql.py b/python/python/tests/test_to_sql.py index 8806d8f..83e848c 100644 --- a/python/python/tests/test_to_sql.py +++ b/python/python/tests/test_to_sql.py @@ -5,7 +5,87 @@ import pyarrow as pa import pytest -from lance_csr import CypherQuery, GraphConfig +from lance_csr import CypherQuery, GraphConfig, SqlEngine + + +@pytest.fixture +def scalar_export_env(): + config = GraphConfig.builder().node_table("Person", "id").build() + datasets = { + "Person": pa.table( + { + "id": [1, 2, 3], + "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()), + } + ) + } + return config, datasets + + +@pytest.mark.parametrize( + "expression", + [ + "toUpper(p.text)", + "round(-1.5)", + "abs(-100)", + "sqrt(9)", + ], +) +def test_sql_export_preserves_scalar_values_and_types(scalar_export_env, expression): + config, datasets = scalar_export_env + query = CypherQuery( + f"MATCH (p:Person) RETURN p.id AS id, {expression} AS value ORDER BY id" + ).with_config(config) + cypher = query.execute(datasets) + sql = SqlEngine(datasets).execute(query.to_sql(datasets)) + assert sql.to_pylist() == cypher.to_pylist() + assert sql.schema.types == cypher.schema.types + + +@pytest.mark.parametrize( + ("expression", "function", "expected"), + [ + ("round(p.number)", "round", [-1.0, None, 2.0]), + ("abs(p.integer)", "abs", [100, None, 2]), + ], +) +def test_sql_export_rejects_runtime_only_cypher_functions( + scalar_export_env, expression, function, expected +): + config, datasets = scalar_export_env + query = CypherQuery( + f"MATCH (p:Person) RETURN p.id AS id, {expression} AS value ORDER BY id" + ).with_config(config) + result = query.execute(datasets) + assert result.column("value").to_pylist() == expected + with pytest.raises( + NotImplementedError, match=f"SQL export of Cypher function '{function}'" + ): + query.to_sql(datasets) + + # A same-named SQL function can differ in values or Arrow result width. + engine = SqlEngine(datasets) + if function == "round": + assert engine.execute( + "SELECT round(number) AS value FROM person ORDER BY id" + ).column("value").to_pylist() == [-2.0, None, 2.0] + if function == "abs": + sql = engine.execute("SELECT abs(integer) AS value FROM person ORDER BY id") + assert sql.schema.field("value").type == pa.int8() + assert result.schema.field("value").type == pa.int64() + + +def test_sql_export_checks_filter_functions(scalar_export_env): + config, datasets = scalar_export_env + query = CypherQuery( + "MATCH (p:Person) WHERE round(p.number) = -1 RETURN p.id" + ).with_config(config) + with pytest.raises( + NotImplementedError, match="SQL export of Cypher function 'round'" + ): + query.to_sql(datasets) @pytest.fixture From 92b1b1f21bd6a559c35044ab54f028b6cdeac8e6 Mon Sep 17 00:00:00 2001 From: David Hughes Date: Sun, 25 Oct 2026 14:25:11 -0700 Subject: [PATCH 4/6] docs(cypher): clarify unsigned math conversion Limit the Int64 overflow restriction to integer abs. The remaining math functions convert numeric values, including UInt64, to Float64; cover those Arrow inputs with real execution tests. Submitted by David Hughes (@stugorf), with AI Collaboration. --- docs/cypher.md | 4 +++- python/python/tests/test_math_functions.py | 13 +++++++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/docs/cypher.md b/docs/cypher.md index f202f62..aaa1bfc 100644 --- a/docs/cypher.md +++ b/docs/cypher.md @@ -267,7 +267,9 @@ RETURN length(p), nodes(p) | Vectors | `vector_distance(a, b, metric)`, `vector_similarity(a, b, metric)`; see [Vector search](vector-search.md) | `abs` preserves integer results; the other math functions return floating point values. Integer -absolute-value overflow and unsigned inputs outside Int64 range raise an error. `round` accepts one argument, with ties toward positive +`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, +with ties toward positive infinity (`round(-1.5)` is `-1.0`), as specified in the [Neo4j 4.4 numeric reference](https://neo4j.com/docs/cypher-manual/4.4/functions/mathematical-numeric/). `sqrt` of a negative number returns NaN. These functions return null for null input and reject diff --git a/python/python/tests/test_math_functions.py b/python/python/tests/test_math_functions.py index 4cb066c..49d90a2 100644 --- a/python/python/tests/test_math_functions.py +++ b/python/python/tests/test_math_functions.py @@ -117,6 +117,19 @@ def test_math_out_of_range_unsigned_integer(): engine.execute("MATCH (p:Person) RETURN abs(p.large) AS value") +@pytest.mark.parametrize("function", ["round", "floor", "ceil", "sqrt"]) +def test_math_unsigned_integer_converts_to_float(function): + config = GraphConfig.builder().node_table("Person", "id").build() + engine = CypherEngine( + config, + {"Person": pa.table({"id": [1], "large": pa.array([2**63], type=pa.uint64())})}, + ) + result = engine.execute(f"MATCH (p:Person) RETURN {function}(p.large) AS value") + expected = math.sqrt(2**63) if function == "sqrt" else float(2**63) + assert result.to_pylist() == [{"value": expected}] + assert result.schema.field("value").type == pa.float64() + + def test_math_non_finite_arrow_values(): config = GraphConfig.builder().node_table("Person", "id").build() engine = CypherEngine( From c53f47c2304f09f48019e0d3facdc7ece62c3a83 Mon Sep 17 00:00:00 2001 From: David Hughes Date: Sun, 25 Oct 2026 14:25:11 -0700 Subject: [PATCH 5/6] fix(cypher): preserve typed nulls in SQL export Use DataFusion arrow_cast after constant folding so typed nulls retain their Arrow result types. Preserve untyped nulls and reject typed-null export to other dialects. Verify real SQL/Cypher round trips and document the boundary. Submitted by David Hughes (@stugorf), with AI Collaboration. --- crates/lance-graph-python/src/graph.rs | 3 +- crates/lance-graph/src/query.rs | 84 ++++++++++++++++++++++---- docs/api-reference.md | 2 +- docs/sql.md | 4 ++ python/python/tests/test_to_sql.py | 24 +++++++- 5 files changed, 102 insertions(+), 15 deletions(-) diff --git a/crates/lance-graph-python/src/graph.rs b/crates/lance-graph-python/src/graph.rs index 85cb999..f3706cf 100644 --- a/crates/lance-graph-python/src/graph.rs +++ b/crates/lance-graph-python/src/graph.rs @@ -605,7 +605,8 @@ impl CypherQuery { /// Raises /// ------ /// NotImplementedError - /// If a runtime-only Cypher function has no faithful SQL translation + /// If a runtime-only Cypher function has no faithful SQL translation, or a + /// typed-null literal is exported outside the Default dialect /// RuntimeError /// If SQL generation fails #[pyo3(signature = (datasets, dialect=None))] diff --git a/crates/lance-graph/src/query.rs b/crates/lance-graph/src/query.rs index 4cb85ff..895326d 100644 --- a/crates/lance-graph/src/query.rs +++ b/crates/lance-graph/src/query.rs @@ -380,16 +380,18 @@ impl CypherQuery { /// /// # Errors /// Returns `GraphError::UnsupportedFeature` for runtime-only Cypher functions - /// without a faithful SQL translation. Constant expressions may be folded first. + /// without a faithful SQL translation or typed nulls outside the Default dialect. + /// Constant expressions may be folded first. pub async fn to_sql( &self, datasets: HashMap, dialect: Option, ) -> Result { - use datafusion::common::not_impl_err; - use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion}; + use datafusion::common::tree_node::{Transformed, TreeNode}; + use datafusion::common::{not_impl_err, ScalarValue}; + use datafusion::logical_expr::expr_rewriter::NamePreserver; use datafusion::logical_expr::registry::FunctionRegistry; - use datafusion::logical_expr::Expr; + use datafusion::logical_expr::{lit, Expr}; use std::sync::Arc; let dialect = dialect.unwrap_or_default(); @@ -414,11 +416,13 @@ impl CypherQuery { })?; // An SQL function with the same name need not implement Cypher semantics. - optimized_plan - .apply_with_subqueries(|plan| { - plan.apply_expressions(|expr| { - expr.apply(|expr| { - if let Expr::ScalarFunction(function) = expr { + let optimized_plan = optimized_plan + .transform_up_with_subqueries(|plan| { + 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 !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", @@ -426,14 +430,31 @@ impl CypherQuery { ); } } - Ok(TreeNodeRecursion::Continue) - }) + if let Expr::Literal(value, _) = &expr { + let data_type = value.data_type(); + if value.is_null() && !data_type.is_null() { + let type_name = data_type.to_string(); + if dialect != SqlDialect::Default + || !type_name.parse::().is_ok_and(|parsed| parsed == data_type) + { + return not_impl_err!("SQL export of typed null {data_type} requires a faithful Default-dialect Arrow cast"); + } + // The unparser emits bare NULL for typed nulls, losing their Arrow type. + return Ok(Transformed::yes(datafusion::functions::core::arrow_cast().call(vec![ + lit(ScalarValue::Null), lit(type_name), + ]))); + } + } + Ok(Transformed::no(expr)) + })?; + Ok(rewritten.update_data(|expr| name.restore(expr))) }) }) .map_err(|error| GraphError::UnsupportedFeature { feature: error.to_string(), location: snafu::Location::new(file!(), line!(), column!()), - })?; + })? + .data; // Unparse to SQL using the specified dialect let dialect_unparser = dialect.unparser(); @@ -2058,6 +2079,45 @@ mod tests { .contains("SQL export of Cypher function 'round'")); } + #[tokio::test] + async fn test_to_sql_preserves_constant_null_types() { + let batch = RecordBatch::try_from_iter(vec![( + "id", + Arc::new(arrow_array::Int64Array::from(vec![1])) as arrow_array::ArrayRef, + )]) + .unwrap(); + let datasets = HashMap::from([("Person".to_string(), batch)]); + let config = GraphConfig::builder() + .node_table("Person", "id") + .build() + .unwrap(); + for expression in [ + "round(null)", + "abs(null)", + "sqrt(null)", + "floor(null)", + "ceil(null)", + "toUpper(null)", + "null", + ] { + let query = CypherQuery::new(&format!("MATCH (p:Person) RETURN {expression} AS value")) + .unwrap() + .with_config(config.clone()); + let cypher = query.execute(datasets.clone(), None).await.unwrap(); + let sql_text = query.to_sql(datasets.clone(), None).await.unwrap(); + let sql = crate::sql_query::SqlQuery::new(&sql_text) + .execute(datasets.clone()) + .await + .unwrap(); + assert_eq!( + sql.schema().field(0).data_type(), + cypher.schema().field(0).data_type(), + "{expression}: {sql_text}" + ); + assert_eq!(sql.column(0).logical_null_count(), 1, "{expression}"); + } + } + async fn write_lance_dataset(path: &std::path::Path, batch: arrow_array::RecordBatch) { use arrow_array::{RecordBatch, RecordBatchIterator}; use lance::dataset::{Dataset, WriteParams}; diff --git a/docs/api-reference.md b/docs/api-reference.md index eaa8fc0..4972240 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -188,7 +188,7 @@ A parsed Cypher query. Raises `ValueError` if `query_text` doesn't parse. Builde `#!python to_sql(datasets: dict, dialect: SqlDialect | None = None) -> str` : Exports supported expressions as SQL joins. Raises `NotImplementedError` for runtime-only Cypher - functions without a faithful SQL translation. See [From Cypher to SQL](sql.md#from-cypher-to-sql). + functions without a faithful SQL translation, and for typed-null literals outside the Default dialect. See [From Cypher to SQL](sql.md#from-cypher-to-sql). `#!python execute_with_vector_rerank(datasets: dict, vector_search: VectorSearch) -> pyarrow.Table` : Runs the query, then reranks by vector similarity. diff --git a/docs/sql.md b/docs/sql.md index 9035a09..ed70018 100644 --- a/docs/sql.md +++ b/docs/sql.md @@ -89,6 +89,10 @@ 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. +Constant-folded typed nulls use DataFusion `arrow_cast` in the Default dialect so their Arrow +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 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 83e848c..b1daa89 100644 --- a/python/python/tests/test_to_sql.py +++ b/python/python/tests/test_to_sql.py @@ -5,7 +5,7 @@ import pyarrow as pa import pytest -from lance_csr import CypherQuery, GraphConfig, SqlEngine +from lance_csr import CypherQuery, GraphConfig, SqlDialect, SqlEngine @pytest.fixture @@ -31,6 +31,13 @@ def scalar_export_env(): "round(-1.5)", "abs(-100)", "sqrt(9)", + "round(null)", + "abs(null)", + "sqrt(null)", + "floor(null)", + "ceil(null)", + "toUpper(null)", + "null", ], ) def test_sql_export_preserves_scalar_values_and_types(scalar_export_env, expression): @@ -77,6 +84,21 @@ def test_sql_export_rejects_runtime_only_cypher_functions( assert result.schema.field("value").type == pa.int64() +@pytest.mark.parametrize( + "dialect", + [SqlDialect.Spark, SqlDialect.PostgreSql, SqlDialect.MySql, SqlDialect.Sqlite], +) +def test_sql_export_typed_null_requires_default_dialect(scalar_export_env, dialect): + config, datasets = scalar_export_env + query = CypherQuery("MATCH (p:Person) RETURN round(null) AS value").with_config( + config + ) + with pytest.raises(NotImplementedError, match="SQL export of typed null Float64"): + query.to_sql(datasets, dialect) + bare_null = CypherQuery("MATCH (p:Person) RETURN null AS value").with_config(config) + assert "NULL" in bare_null.to_sql(datasets, dialect) + + def test_sql_export_checks_filter_functions(scalar_export_env): config, datasets = scalar_export_env query = CypherQuery( From 5be628da42c60fc1005340e81f9fa5bdba062231 Mon Sep 17 00:00:00 2001 From: David Hughes Date: Mon, 26 Oct 2026 09:51:45 -0700 Subject: [PATCH 6/6] Add Cypher string and list scalar functions (#43) * 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. * 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. * 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. * 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. * 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. * 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/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 | 160 +++-- .../src/datafusion_planner/list_folds.rs | 328 +++++++++++ .../lance-graph/src/datafusion_planner/mod.rs | 3 + .../datafusion_planner/scalar_functions.rs | 551 ++++++++++++++++++ crates/lance-graph/src/logical_plan.rs | 203 ++++++- .../lance-graph/src/parameter_substitution.rs | 16 + crates/lance-graph/src/parser.rs | 96 ++- crates/lance-graph/src/pattern_order.rs | 19 +- crates/lance-graph/src/query.rs | 12 +- crates/lance-graph/src/semantic.rs | 110 +++- .../tests/test_datafusion_pipeline.rs | 9 +- docs/cypher.md | 98 +++- docs/sql.md | 7 +- python/python/tests/test_alias_shadowing.py | 74 +++ python/python/tests/test_entity_functions.py | 137 +++++ python/python/tests/test_list_folds.py | 240 ++++++++ python/python/tests/test_scalar_functions.py | 196 ++++++- python/python/tests/test_to_sql.py | 31 + 24 files changed, 2322 insertions(+), 138 deletions(-) create mode 100644 crates/lance-graph/src/datafusion_planner/list_folds.rs create mode 100644 crates/lance-graph/src/datafusion_planner/scalar_functions.rs create mode 100644 python/python/tests/test_alias_shadowing.py 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 e57229e..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), @@ -196,18 +250,19 @@ 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" | "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" => { + 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]), @@ -464,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) @@ -475,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); } @@ -519,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 9f66573..a11afea 100644 --- a/crates/lance-graph/src/datafusion_planner/mod.rs +++ b/crates/lance-graph/src/datafusion_planner/mod.rs @@ -20,7 +20,9 @@ mod config_helpers; mod cypher_udf; mod expression; mod join_ops; +mod list_folds; mod math_functions; +mod scalar_functions; mod scan_ops; mod udf; pub mod vector_ops; @@ -30,6 +32,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 new file mode 100644 index 0000000..dbf283c --- /dev/null +++ b/crates/lance-graph/src/datafusion_planner/scalar_functions.rs @@ -0,0 +1,551 @@ +// 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, cast_with_options, CastOptions}; +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(), + "__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(), + _ => 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() { + "__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" | "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), + "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 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( + &arg.to_array(args.number_rows)?, + &DataType::Int64, + &CastOptions { + safe: false, + ..Default::default() + }, + )?); + } + 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) +} + +/// 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::*; + use crate::{config::GraphConfig, query::CypherQuery}; + 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)]) + .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_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 [ + ("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/logical_plan.rs b/crates/lance-graph/src/logical_plan.rs index 4359fc0..9b7822c 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())), @@ -1386,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, @@ -1530,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/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 2f6035b..435eb26 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}, @@ -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(( @@ -1382,14 +1445,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())) } @@ -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/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/crates/lance-graph/src/semantic.rs b/crates/lance-graph/src/semantic.rs index f0a2227..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 { @@ -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" | "keys" | "labels" => { 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, 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!()), @@ -515,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)?; @@ -716,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, @@ -742,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) { @@ -1595,7 +1691,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..6cd4476 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 @@ -258,14 +259,39 @@ 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) | +| 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 | | 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. Unsigned Arrow arguments outside Int64 range also raise an error; they are not +converted to null. 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, @@ -280,10 +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 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. `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 @@ -308,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/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_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 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}] diff --git a/python/python/tests/test_scalar_functions.py b/python/python/tests/test_scalar_functions.py index 83eb1b5..b82afb3 100644 --- a/python/python/tests/test_scalar_functions.py +++ b/python/python/tests/test_scalar_functions.py @@ -1,22 +1,194 @@ # 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.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() + 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]}, + ] diff --git a/python/python/tests/test_to_sql.py b/python/python/tests/test_to_sql.py index b1daa89..4954ba2 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,18 @@ def scalar_export_env(): "ceil(null)", "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): @@ -56,6 +84,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(