diff --git a/crates/lance-graph-python/src/graph.rs b/crates/lance-graph-python/src/graph.rs index eadcabd..f3706cf 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,9 @@ impl CypherQuery { /// /// Raises /// ------ + /// NotImplementedError + /// 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/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/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 18545b6..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), @@ -193,18 +247,22 @@ 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), }, - "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), - ), + "abs" | "round" | "floor" | "ceil" | "sqrt" => { + super::math_functions::call(name, args.iter().map(to_df_value_expr).collect()) + } + "left" | "right" | "substring" | "replace" | "split" | "trim" | "ltrim" + | "rtrim" | "size" | "reverse" | "last" | "tail" | "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]), @@ -461,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) @@ -472,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); } @@ -516,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/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..a11afea 100644 --- a/crates/lance-graph/src/datafusion_planner/mod.rs +++ b/crates/lance-graph/src/datafusion_planner/mod.rs @@ -20,6 +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; @@ -29,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 f617431..b99f200 100644 --- a/crates/lance-graph/src/query.rs +++ b/crates/lance-graph/src/query.rs @@ -377,11 +377,21 @@ 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 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::tree_node::{Transformed, TreeNode}; + 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, utils::merge_schema, Expr, LogicalPlan}; use std::sync::Arc; let dialect = dialect.unwrap_or_default(); @@ -405,6 +415,55 @@ impl CypherQuery { location: snafu::Location::new(file!(), line!(), column!()), })?; + // 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", + function.name() + ); + } + } + 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(); let unparser = dialect_unparser.as_unparser(); @@ -1997,6 +2056,76 @@ 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'")); + } + + #[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/crates/lance-graph/src/semantic.rs b/crates/lance-graph/src/semantic.rs index b8cf649..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 { @@ -438,7 +438,9 @@ 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" | "trim" | "ltrim" | "rtrim" | "reverse" | "last" + | "tail" | "keys" | "labels" => { if args.len() != 1 { return Err(GraphError::PlanError { message: format!( @@ -450,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 { @@ -468,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 { @@ -506,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. 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!()), @@ -514,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)?; @@ -715,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, @@ -741,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) { @@ -1594,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/api-reference.md b/docs/api-reference.md index d28bb6b..4972240 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, 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/cypher.md b/docs/cypher.md index 98bbd25..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,16 +259,109 @@ 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) | -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)`. +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, +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}] +``` + +`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. `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 @@ -292,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 57f60c3..85a5b1c 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,22 @@ 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`, 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. + +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`. + +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. + !!! 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_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_math_functions.py b/python/python/tests/test_math_functions.py new file mode 100644 index 0000000..49d90a2 --- /dev/null +++ b/python/python/tests/test_math_functions.py @@ -0,0 +1,145 @@ +# 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").to_pylist()[ + 0 + ] + assert math.isnan(result["root"]) + assert math.copysign(1, result["zero"]) == 1 + + +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") + + +@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( + 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]) 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 8806d8f..4954ba2 100644 --- a/python/python/tests/test_to_sql.py +++ b/python/python/tests/test_to_sql.py @@ -5,7 +5,140 @@ import pyarrow as pa import pytest -from lance_csr import CypherQuery, GraphConfig +from lance_csr import CypherQuery, GraphConfig, SqlDialect, 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()), + "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()), + } + ) + } + return config, datasets + + +@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)", + "sqrt(9)", + "round(null)", + "abs(null)", + "sqrt(null)", + "floor(null)", + "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): + 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]), + ("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( + 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() + + +@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( + "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