From bb76c31a25d810ecc01b136b86c616d3b2829d3d Mon Sep 17 00:00:00 2001 From: beinan Date: Wed, 28 Jan 2026 21:50:59 +0000 Subject: [PATCH 01/11] fix(query): error on unsupported Cypher functions --- crates/lance-graph/src/query.rs | 57 +++++++++++++++- crates/lance-graph/src/semantic.rs | 67 ++++++++++++++++--- .../lance-graph/src/simple_executor/expr.rs | 42 +++++++++++- python/python/tests/test_scalar_functions.py | 24 +++++++ 4 files changed, 178 insertions(+), 12 deletions(-) create mode 100644 python/python/tests/test_scalar_functions.py diff --git a/crates/lance-graph/src/query.rs b/crates/lance-graph/src/query.rs index 281a9fa09..82c78c3f8 100644 --- a/crates/lance-graph/src/query.rs +++ b/crates/lance-graph/src/query.rs @@ -721,7 +721,13 @@ impl CypherQuery { // Phase 1: Semantic Analysis let mut analyzer = SemanticAnalyzer::new(config.clone()); - analyzer.analyze(&self.ast)?; + let semantic = analyzer.analyze(&self.ast)?; + if !semantic.errors.is_empty() { + return Err(GraphError::PlanError { + message: format!("Semantic analysis failed:\n{}", semantic.errors.join("\n")), + location: snafu::Location::new(file!(), line!(), column!()), + }); + } // Phase 2: Graph Logical Plan let mut logical_planner = LogicalPlanner::new(); @@ -879,12 +885,23 @@ impl CypherQuery { datasets: HashMap, ) -> Result { use arrow::compute::concat_batches; + use crate::semantic::SemanticAnalyzer; use datafusion::datasource::MemTable; use datafusion::prelude::*; use std::sync::Arc; // Require a config for now, even if we don't fully exploit it yet - let _config = self.require_config()?; + let config = self.require_config()?.clone(); + + // Ensure we don't silently ignore unsupported features (e.g. scalar functions). + let mut analyzer = SemanticAnalyzer::new(config); + let semantic = analyzer.analyze(&self.ast)?; + if !semantic.errors.is_empty() { + return Err(GraphError::PlanError { + message: format!("Semantic analysis failed:\n{}", semantic.errors.join("\n")), + location: snafu::Location::new(file!(), line!(), column!()), + }); + } if datasets.is_empty() { return Err(GraphError::PlanError { @@ -2249,4 +2266,40 @@ mod tests { "expected unsupported feature error, got {err:?}" ); } + + #[tokio::test] + async fn test_execute_fails_on_semantic_error() { + use arrow_array::RecordBatch; + use arrow_schema::{DataType, Field, Schema}; + use std::sync::Arc; + use std::collections::HashMap; + + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Int64, false), + ])); + let batch = RecordBatch::new_empty(schema); + let mut datasets = HashMap::new(); + datasets.insert("Person".to_string(), batch); + + let cfg = GraphConfig::builder() + .with_node_label("Person", "id") + .build() + .unwrap(); + + // Query referencing undefined variable 'x' + let query = CypherQuery::new("MATCH (n:Person) RETURN x.name") + .unwrap() + .with_config(cfg); + + let result = query.execute_simple(datasets).await; + + assert!(result.is_err()); + match result { + Err(GraphError::PlanError { message, .. }) => { + assert!(message.contains("Semantic analysis failed")); + assert!(message.contains("Undefined variable: 'x'")); + } + _ => panic!("Expected PlanError with semantic failure message, got {:?}", result), + } + } } diff --git a/crates/lance-graph/src/semantic.rs b/crates/lance-graph/src/semantic.rs index 926fc0c97..17fa910d1 100644 --- a/crates/lance-graph/src/semantic.rs +++ b/crates/lance-graph/src/semantic.rs @@ -53,7 +53,6 @@ pub enum ScopeType { /// Semantic analysis result with validated and enriched AST #[derive(Debug, Clone)] pub struct SemanticResult { - pub query: CypherQuery, pub variables: HashMap, pub errors: Vec, pub warnings: Vec, @@ -106,6 +105,14 @@ impl SemanticAnalyzer { } } + // Phase 4: Variable discovery in post-WITH MATCH clauses (query chaining) + self.current_scope = ScopeType::Match; + for match_clause in &query.post_with_match_clauses { + if let Err(e) = self.analyze_match_clause(match_clause) { + errors.push(format!("Post-WITH MATCH clause error: {}", e)); + } + } + // Phase 4: Validate post-WITH WHERE clause if present if let Some(post_where) = &query.post_with_where_clause { self.current_scope = ScopeType::PostWithWhere; @@ -135,7 +142,6 @@ impl SemanticAnalyzer { self.validate_types(&mut errors); Ok(SemanticResult { - query: query.clone(), variables: self.variables.clone(), errors, warnings, @@ -343,14 +349,17 @@ impl SemanticAnalyzer { } } ValueExpression::Function { name, args } => { + let function_name = name.to_lowercase(); + // Validate function-specific arity and signature rules - match name.to_lowercase().as_str() { + match function_name.as_str() { + // Aggregations "count" | "sum" | "avg" | "min" | "max" | "collect" => { if args.len() != 1 { return Err(GraphError::PlanError { message: format!( "{} requires exactly 1 argument, got {}", - name.to_uppercase(), + function_name.to_uppercase(), args.len() ), location: snafu::Location::new(file!(), line!(), column!()), @@ -359,25 +368,46 @@ impl SemanticAnalyzer { // Additional validation for SUM, AVG, MIN, MAX: they require properties, not bare variables // Only COUNT and COLLECT allow bare variables (COUNT(*), COUNT(p), COLLECT(p)) - if matches!(name.to_lowercase().as_str(), "sum" | "avg" | "min" | "max") { + if matches!(function_name.as_str(), "sum" | "avg" | "min" | "max") { if let Some(ValueExpression::Variable(v)) = args.first() { return Err(GraphError::PlanError { message: format!( "{}({}) is invalid - {} requires a property like {}({}.property). You cannot {} a node/entity.", - name.to_uppercase(), v, name.to_uppercase(), name.to_uppercase(), v, name.to_lowercase() + function_name.to_uppercase(), v, function_name.to_uppercase(), function_name.to_uppercase(), v, function_name ), location: snafu::Location::new(file!(), line!(), column!()), }); } } } + // Scalar string functions + "tolower" | "lower" | "toupper" | "upper" => { + if args.len() != 1 { + return Err(GraphError::PlanError { + message: format!( + "{} requires exactly 1 argument, got {}", + function_name.to_uppercase(), + args.len() + ), + location: snafu::Location::new(file!(), line!(), column!()), + }); + } + } + // Unknown/unimplemented scalar function _ => { - // Other functions - no validation yet + return Err(GraphError::UnsupportedFeature { + feature: format!("Cypher function '{}' is not implemented", function_name), + location: snafu::Location::new(file!(), line!(), column!()), + }); } } - // Validate arguments recursively + // Validate arguments recursively. + // Special-case COUNT(*) where '*' isn't a real variable. for arg in args { + if function_name == "count" && matches!(arg, ValueExpression::Variable(v) if v == "*") { + continue; + } self.analyze_value_expression(arg)?; } } @@ -453,6 +483,21 @@ impl SemanticAnalyzer { Ok(()) } + fn register_projection_alias(&mut self, alias: &str) { + if self.variables.contains_key(alias) { + return; + } + + let var_info = VariableInfo { + name: alias.to_string(), + variable_type: VariableType::Property, + labels: vec![], + properties: HashSet::new(), + defined_in: self.current_scope.clone(), + }; + self.variables.insert(alias.to_string(), var_info); + } + /// Validate property reference fn validate_property_reference(&self, prop_ref: &PropertyRef) -> Result<()> { if !self.variables.contains_key(&prop_ref.variable) { @@ -468,6 +513,9 @@ impl SemanticAnalyzer { fn analyze_return_clause(&mut self, return_clause: &ReturnClause) -> Result<()> { for item in &return_clause.items { self.analyze_value_expression(&item.expression)?; + if let Some(alias) = &item.alias { + self.register_projection_alias(alias); + } } Ok(()) } @@ -477,6 +525,9 @@ impl SemanticAnalyzer { // Validate WITH item expressions (similar to RETURN) for item in &with_clause.items { self.analyze_value_expression(&item.expression)?; + if let Some(alias) = &item.alias { + self.register_projection_alias(alias); + } } // Validate ORDER BY within WITH if present if let Some(order_by) = &with_clause.order_by { diff --git a/crates/lance-graph/src/simple_executor/expr.rs b/crates/lance-graph/src/simple_executor/expr.rs index ec1fb694f..130ac3fe9 100644 --- a/crates/lance-graph/src/simple_executor/expr.rs +++ b/crates/lance-graph/src/simple_executor/expr.rs @@ -163,12 +163,50 @@ pub(crate) fn to_df_value_expr_simple( expr: &crate::ast::ValueExpression, ) -> datafusion::logical_expr::Expr { use crate::ast::ValueExpression as VE; - use datafusion::logical_expr::{col, lit}; + use datafusion::functions::string::{lower, upper}; + use datafusion::logical_expr::{col, lit, BinaryExpr, Expr, Operator}; match expr { VE::Property(prop) => col(&prop.property), VE::Variable(v) => col(v), VE::Literal(v) => to_df_literal(v), - VE::Function { .. } | VE::Arithmetic { .. } => lit(0), + VE::Function { name, args } => match name.to_lowercase().as_str() { + "tolower" | "lower" => { + if args.len() == 1 { + lower().call(vec![to_df_value_expr_simple(&args[0])]) + } else { + Expr::Literal(datafusion::scalar::ScalarValue::Null, None) + } + } + "toupper" | "upper" => { + if args.len() == 1 { + upper().call(vec![to_df_value_expr_simple(&args[0])]) + } else { + Expr::Literal(datafusion::scalar::ScalarValue::Null, None) + } + } + _ => Expr::Literal(datafusion::scalar::ScalarValue::Null, None), + }, + VE::Arithmetic { + left, + operator, + right, + } => { + use crate::ast::ArithmeticOperator as AO; + let l = to_df_value_expr_simple(left); + let r = to_df_value_expr_simple(right); + let op = match operator { + AO::Add => Operator::Plus, + AO::Subtract => Operator::Minus, + AO::Multiply => Operator::Multiply, + AO::Divide => Operator::Divide, + AO::Modulo => Operator::Modulo, + }; + Expr::BinaryExpr(BinaryExpr { + left: Box::new(l), + op, + right: Box::new(r), + }) + } VE::VectorDistance { .. } => lit(0.0f32), VE::VectorSimilarity { .. } => lit(1.0f32), VE::Parameter(_) => lit(0), diff --git a/python/python/tests/test_scalar_functions.py b/python/python/tests/test_scalar_functions.py new file mode 100644 index 000000000..bcf82b3a3 --- /dev/null +++ b/python/python/tests/test_scalar_functions.py @@ -0,0 +1,24 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The Lance Authors + +import pyarrow as pa +import pytest + +from lance_graph import CypherQuery, GraphConfig + + +def test_unimplemented_scalar_function_errors() -> None: + cfg = GraphConfig.builder().with_node_label("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) + + with pytest.raises(Exception) as excinfo: + query.execute(datasets) + + assert "replace" in str(excinfo.value).lower() + From 145d3f14eb240507f8e82b7415dee77a22f5c6fa Mon Sep 17 00:00:00 2001 From: beinan Date: Wed, 28 Jan 2026 21:53:22 +0000 Subject: [PATCH 02/11] test: format scalar function test --- python/python/tests/test_scalar_functions.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/python/python/tests/test_scalar_functions.py b/python/python/tests/test_scalar_functions.py index bcf82b3a3..70028a9c3 100644 --- a/python/python/tests/test_scalar_functions.py +++ b/python/python/tests/test_scalar_functions.py @@ -14,11 +14,10 @@ def test_unimplemented_scalar_function_errors() -> None: query = CypherQuery( "MATCH (p:Person) " "RETURN p.name AS name, replace(p.name, 'A', 'a') AS replaced " - "ORDER BY name" + "ORDER BY name", ).with_config(cfg) with pytest.raises(Exception) as excinfo: query.execute(datasets) assert "replace" in str(excinfo.value).lower() - From 348b898413e14f5e9311a03f304000d4518df482 Mon Sep 17 00:00:00 2001 From: beinan Date: Wed, 28 Jan 2026 21:57:21 +0000 Subject: [PATCH 03/11] style: sort imports in scalar function test --- python/python/tests/test_scalar_functions.py | 1 - 1 file changed, 1 deletion(-) diff --git a/python/python/tests/test_scalar_functions.py b/python/python/tests/test_scalar_functions.py index 70028a9c3..8d3a08990 100644 --- a/python/python/tests/test_scalar_functions.py +++ b/python/python/tests/test_scalar_functions.py @@ -3,7 +3,6 @@ import pyarrow as pa import pytest - from lance_graph import CypherQuery, GraphConfig From c9703196fa59ffe6d92623db0070e186af9afe8d Mon Sep 17 00:00:00 2001 From: beinan Date: Wed, 28 Jan 2026 22:00:38 +0000 Subject: [PATCH 04/11] style(rust): appease rustfmt --- crates/lance-graph/src/query.rs | 13 +++++++------ crates/lance-graph/src/semantic.rs | 9 +++++++-- 2 files changed, 14 insertions(+), 8 deletions(-) diff --git a/crates/lance-graph/src/query.rs b/crates/lance-graph/src/query.rs index 82c78c3f8..391bbe4d8 100644 --- a/crates/lance-graph/src/query.rs +++ b/crates/lance-graph/src/query.rs @@ -884,8 +884,8 @@ impl CypherQuery { &self, datasets: HashMap, ) -> Result { - use arrow::compute::concat_batches; use crate::semantic::SemanticAnalyzer; + use arrow::compute::concat_batches; use datafusion::datasource::MemTable; use datafusion::prelude::*; use std::sync::Arc; @@ -2271,12 +2271,10 @@ mod tests { async fn test_execute_fails_on_semantic_error() { use arrow_array::RecordBatch; use arrow_schema::{DataType, Field, Schema}; - use std::sync::Arc; use std::collections::HashMap; + use std::sync::Arc; - let schema = Arc::new(Schema::new(vec![ - Field::new("id", DataType::Int64, false), - ])); + let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)])); let batch = RecordBatch::new_empty(schema); let mut datasets = HashMap::new(); datasets.insert("Person".to_string(), batch); @@ -2299,7 +2297,10 @@ mod tests { assert!(message.contains("Semantic analysis failed")); assert!(message.contains("Undefined variable: 'x'")); } - _ => panic!("Expected PlanError with semantic failure message, got {:?}", result), + _ => panic!( + "Expected PlanError with semantic failure message, got {:?}", + result + ), } } } diff --git a/crates/lance-graph/src/semantic.rs b/crates/lance-graph/src/semantic.rs index 17fa910d1..dd34ae722 100644 --- a/crates/lance-graph/src/semantic.rs +++ b/crates/lance-graph/src/semantic.rs @@ -396,7 +396,10 @@ impl SemanticAnalyzer { // Unknown/unimplemented scalar function _ => { return Err(GraphError::UnsupportedFeature { - feature: format!("Cypher function '{}' is not implemented", function_name), + feature: format!( + "Cypher function '{}' is not implemented", + function_name + ), location: snafu::Location::new(file!(), line!(), column!()), }); } @@ -405,7 +408,9 @@ impl SemanticAnalyzer { // Validate arguments recursively. // Special-case COUNT(*) where '*' isn't a real variable. for arg in args { - if function_name == "count" && matches!(arg, ValueExpression::Variable(v) if v == "*") { + if function_name == "count" + && matches!(arg, ValueExpression::Variable(v) if v == "*") + { continue; } self.analyze_value_expression(arg)?; From 16fe424954c56256342d8237fc45c7356e4f42ae Mon Sep 17 00:00:00 2001 From: beinan Date: Wed, 28 Jan 2026 22:14:19 +0000 Subject: [PATCH 05/11] test(query): cover scalar function semantics --- .../tests/test_datafusion_pipeline.rs | 194 +++++------------- 1 file changed, 48 insertions(+), 146 deletions(-) diff --git a/crates/lance-graph/tests/test_datafusion_pipeline.rs b/crates/lance-graph/tests/test_datafusion_pipeline.rs index b97257522..54fa6bec2 100644 --- a/crates/lance-graph/tests/test_datafusion_pipeline.rs +++ b/crates/lance-graph/tests/test_datafusion_pipeline.rs @@ -4508,184 +4508,86 @@ async fn test_with_post_match_chaining() { } // ============================================================================ -// UNWIND Tests +// Scalar Function / Semantic Validation Regression Tests // ============================================================================ #[tokio::test] -async fn test_unwind_simple_list() { - let config = create_graph_config(); - // We need at least one dataset to initialize the catalog/context even if not used in query +async fn test_unimplemented_scalar_function_errors() { let person_batch = create_person_dataset(); + let config = GraphConfig::builder() + .with_node_label("Person", "id") + .build() + .unwrap(); - let query = CypherQuery::new("UNWIND [1, 2, 3] AS x RETURN x") - .unwrap() - .with_config(config); + let query = CypherQuery::new( + "MATCH (p:Person) RETURN p.name AS name, replace(p.name, 'A', 'a') AS replaced", + ) + .unwrap() + .with_config(config); let mut datasets = HashMap::new(); datasets.insert("Person".to_string(), person_batch); - let result = query + let err = query .execute(datasets, Some(ExecutionStrategy::DataFusion)) .await - .unwrap(); + .expect_err("replace() should error until implemented"); - assert_eq!(result.num_rows(), 3); - assert_eq!(result.num_columns(), 1); - - // Check the column type - UNWIND returns different types based on implementation - let col = result.column(0); - let data_type = col.data_type(); - - // Try to get values as Int64, Float32, or Float64 - if let Some(int_values) = col.as_any().downcast_ref::() { - let result_values: Vec = (0..result.num_rows()) - .map(|i| int_values.value(i)) - .collect(); - assert_eq!(result_values, vec![1, 2, 3]); - } else if let Some(float_values) = col.as_any().downcast_ref::() { - let result_values: Vec = (0..result.num_rows()) - .map(|i| float_values.value(i)) - .collect(); - assert_eq!(result_values, vec![1.0, 2.0, 3.0]); - } else if let Some(float_values) = col.as_any().downcast_ref::() { - let result_values: Vec = (0..result.num_rows()) - .map(|i| float_values.value(i)) - .collect(); - assert_eq!(result_values, vec![1.0, 2.0, 3.0]); - } else { - panic!("Unexpected column type: {:?}", data_type); - } + let message = err.to_string().to_lowercase(); + assert!(message.contains("replace"), "unexpected error: {err}"); + assert!( + message.contains("not implemented") || message.contains("unsupported"), + "unexpected error: {err}" + ); } #[tokio::test] -async fn test_unwind_after_match() { - let config = create_graph_config(); +async fn test_tolower_works_in_simple_executor() { let person_batch = create_person_dataset(); + let config = GraphConfig::builder() + .with_node_label("Person", "id") + .build() + .unwrap(); - // Alice, Bob - // UNWIND [10, 20] -> Cartesian product: (Alice, 10), (Alice, 20), (Bob, 10), (Bob, 20) - let query = CypherQuery::new("MATCH (p:Person) UNWIND [10, 20] AS x RETURN p.name, x") - .unwrap() - .with_config(config); + let query = CypherQuery::new( + "MATCH (p:Person) RETURN p.name AS name, tolower(p.name) AS lowered ORDER BY name", + ) + .unwrap() + .with_config(config); let mut datasets = HashMap::new(); datasets.insert("Person".to_string(), person_batch); let result = query - .execute(datasets, Some(ExecutionStrategy::DataFusion)) + .execute(datasets, Some(ExecutionStrategy::Simple)) .await .unwrap(); - assert_eq!(result.num_rows(), 10); // 5 people * 2 values = 10 rows - assert_eq!(result.num_columns(), 2); - let names = result - .column(0) + .column_by_name("name") + .unwrap() .as_any() .downcast_ref::() .unwrap(); - - // Try Int64, Float32, or Float64 for the unwound values - let mut rows: Vec<(String, i32)> = if let Some(int_values) = - result.column(1).as_any().downcast_ref::() - { - (0..result.num_rows()) - .map(|i| (names.value(i).to_string(), int_values.value(i) as i32)) - .collect() - } else if let Some(float_values) = result - .column(1) - .as_any() - .downcast_ref::() - { - (0..result.num_rows()) - .map(|i| (names.value(i).to_string(), float_values.value(i) as i32)) - .collect() - } else if let Some(float_values) = result.column(1).as_any().downcast_ref::() { - (0..result.num_rows()) - .map(|i| (names.value(i).to_string(), float_values.value(i) as i32)) - .collect() - } else { - panic!( - "Unexpected column type for unwound values: {:?}", - result.column(1).data_type() - ); - }; - - rows.sort(); - - let expected = vec![ - ("Alice".to_string(), 10), - ("Alice".to_string(), 20), - ("Bob".to_string(), 10), - ("Bob".to_string(), 20), - ("Charlie".to_string(), 10), - ("Charlie".to_string(), 20), - ("David".to_string(), 10), - ("David".to_string(), 20), - ("Eve".to_string(), 10), - ("Eve".to_string(), 20), - ]; - - assert_eq!(rows, expected); -} - -#[tokio::test] -async fn test_unwind_then_match() { - let config = create_graph_config(); - let person_batch = create_person_dataset(); - - // UNWIND [1, 2] usually yields Float64Array in current parser/planner pipeline - // p.id is Int64 - // We expect DataFusion to handle comparison (maybe with cast) - let query = CypherQuery::new("UNWIND [1, 2] AS target_id MATCH (p:Person) WHERE p.id = target_id RETURN p.name, target_id") + let lowered = result + .column_by_name("lowered") .unwrap() - .with_config(config); - - let mut datasets = HashMap::new(); - datasets.insert("Person".to_string(), person_batch); - - let result = query - .execute(datasets, Some(ExecutionStrategy::DataFusion)) - .await - .unwrap(); - - // We expect 2 matches: Alice (id=1) and Bob (id=2) - assert_eq!(result.num_rows(), 2); - - let names = result - .column(0) .as_any() .downcast_ref::() .unwrap(); - // target_id from Unwind might be Int64, Float32, or Float64 - let mut rows: Vec<(String, i32)> = - if let Some(int_ids) = result.column(1).as_any().downcast_ref::() { - (0..result.num_rows()) - .map(|i| (names.value(i).to_string(), int_ids.value(i) as i32)) - .collect() - } else if let Some(float_ids) = result - .column(1) - .as_any() - .downcast_ref::() - { - (0..result.num_rows()) - .map(|i| (names.value(i).to_string(), float_ids.value(i) as i32)) - .collect() - } else if let Some(float_ids) = result.column(1).as_any().downcast_ref::() { - (0..result.num_rows()) - .map(|i| (names.value(i).to_string(), float_ids.value(i) as i32)) - .collect() - } else { - panic!( - "Unexpected column type for target_id: {:?}", - result.column(1).data_type() - ); - }; - - rows.sort(); - - let expected = vec![("Alice".to_string(), 1), ("Bob".to_string(), 2)]; - - assert_eq!(rows, expected); + let got: Vec<(String, String)> = (0..result.num_rows()) + .map(|i| (names.value(i).to_string(), lowered.value(i).to_string())) + .collect(); + + assert_eq!( + got, + vec![ + ("Alice".to_string(), "alice".to_string()), + ("Bob".to_string(), "bob".to_string()), + ("Charlie".to_string(), "charlie".to_string()), + ("David".to_string(), "david".to_string()), + ("Eve".to_string(), "eve".to_string()), + ] + ); } From 19345ef7c958ae6551c6f6a7e02f7931a6648a6a Mon Sep 17 00:00:00 2001 From: beinan Date: Wed, 28 Jan 2026 22:21:37 +0000 Subject: [PATCH 06/11] chore: add TODO for hard error on unknown functions --- crates/lance-graph/src/datafusion_planner/expression.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/crates/lance-graph/src/datafusion_planner/expression.rs b/crates/lance-graph/src/datafusion_planner/expression.rs index 954b54654..9372afed5 100644 --- a/crates/lance-graph/src/datafusion_planner/expression.rs +++ b/crates/lance-graph/src/datafusion_planner/expression.rs @@ -220,6 +220,10 @@ pub(crate) fn to_df_value_expr(expr: &ValueExpression) -> Expr { _ => { // Unsupported function - return NULL which coerces to any type // This prevents type coercion errors in both string and numeric contexts + // + // TODO(#107): Now that semantic analysis rejects unknown functions, consider + // upgrading this to a hard internal error (e.g. `unreachable!()` or returning + // a planner/execution error) to catch validator regressions early. Expr::Literal(datafusion::scalar::ScalarValue::Null, None) } } From 1c17037f80581a6d19c00548c962a44508fcb8c6 Mon Sep 17 00:00:00 2001 From: beinan Date: Wed, 28 Jan 2026 22:51:40 +0000 Subject: [PATCH 07/11] fix(simple): respect RETURN aliases --- crates/lance-graph/src/query.rs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/crates/lance-graph/src/query.rs b/crates/lance-graph/src/query.rs index 391bbe4d8..a7e15ae12 100644 --- a/crates/lance-graph/src/query.rs +++ b/crates/lance-graph/src/query.rs @@ -975,7 +975,14 @@ impl CypherQuery { .return_clause .items .iter() - .map(|item| to_df_value_expr_simple(&item.expression)) + .map(|item| { + let expr = to_df_value_expr_simple(&item.expression); + if let Some(alias) = &item.alias { + expr.alias(alias) + } else { + expr + } + }) .collect(); if !proj_exprs.is_empty() { df = df.select(proj_exprs).map_err(|e| GraphError::PlanError { From dd2fc8858e0fbbb3dcca08fcd1c20a6efb17a33a Mon Sep 17 00:00:00 2001 From: beinan Date: Thu, 29 Jan 2026 06:42:07 +0000 Subject: [PATCH 08/11] test: reorganize simple executor coverage --- crates/lance-graph/src/semantic.rs | 31 ++++++++ .../lance-graph/src/simple_executor/expr.rs | 42 +++++++++++ .../tests/test_datafusion_pipeline.rs | 51 +------------ .../tests/test_simple_executor_pipeline.rs | 73 +++++++++++++++++++ 4 files changed, 147 insertions(+), 50 deletions(-) create mode 100644 crates/lance-graph/tests/test_simple_executor_pipeline.rs diff --git a/crates/lance-graph/src/semantic.rs b/crates/lance-graph/src/semantic.rs index dd34ae722..293ba0b50 100644 --- a/crates/lance-graph/src/semantic.rs +++ b/crates/lance-graph/src/semantic.rs @@ -1187,6 +1187,37 @@ mod tests { ); } + #[test] + fn test_count_star_passes_validation() { + // COUNT(*) should be allowed (special-cased in semantic analysis) + let expr = ValueExpression::Function { + name: "count".to_string(), + args: vec![ValueExpression::Variable("*".to_string())], + }; + let result = analyze_return_with_match("n", "Person", expr).unwrap(); + assert!( + result.errors.is_empty(), + "Expected COUNT(*) to pass semantic validation, got: {:?}", + result.errors + ); + } + + #[test] + fn test_unimplemented_scalar_function_fails_validation() { + let expr = ValueExpression::Function { + name: "replace".to_string(), + args: vec![ValueExpression::Property(PropertyRef::new("n", "name"))], + }; + let result = analyze_return_with_match("n", "Person", expr).unwrap(); + assert!( + result.errors + .iter() + .any(|e| e.to_lowercase().contains("not implemented")), + "Expected semantic validation to reject unimplemented function, got: {:?}", + result.errors + ); + } + #[test] fn test_sum_with_variable_fails_validation() { let expr = ValueExpression::Function { diff --git a/crates/lance-graph/src/simple_executor/expr.rs b/crates/lance-graph/src/simple_executor/expr.rs index 130ac3fe9..0ddc2c73c 100644 --- a/crates/lance-graph/src/simple_executor/expr.rs +++ b/crates/lance-graph/src/simple_executor/expr.rs @@ -213,3 +213,45 @@ pub(crate) fn to_df_value_expr_simple( VE::VectorLiteral(_) => lit(0.0f32), } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::ast::{ArithmeticOperator, PropertyRef, ValueExpression}; + use datafusion::logical_expr::Expr; + use datafusion::scalar::ScalarValue; + + #[test] + fn test_simple_expr_unknown_function_returns_null() { + let expr = ValueExpression::Function { + name: "replace".to_string(), + args: vec![ValueExpression::Property(PropertyRef::new("p", "name"))], + }; + let df_expr = to_df_value_expr_simple(&expr); + assert!(matches!(df_expr, Expr::Literal(ScalarValue::Null, _))); + } + + #[test] + fn test_simple_expr_lower_wrong_arity_returns_null() { + let expr = ValueExpression::Function { + name: "lower".to_string(), + args: vec![ + ValueExpression::Property(PropertyRef::new("p", "name")), + ValueExpression::Property(PropertyRef::new("p", "name")), + ], + }; + let df_expr = to_df_value_expr_simple(&expr); + assert!(matches!(df_expr, Expr::Literal(ScalarValue::Null, _))); + } + + #[test] + fn test_simple_expr_arithmetic_builds_binary_expr() { + let expr = ValueExpression::Arithmetic { + left: Box::new(ValueExpression::Variable("x".to_string())), + operator: ArithmeticOperator::Add, + right: Box::new(ValueExpression::Literal(crate::ast::PropertyValue::Integer(1))), + }; + let df_expr = to_df_value_expr_simple(&expr); + assert!(matches!(df_expr, Expr::BinaryExpr(_))); + } +} diff --git a/crates/lance-graph/tests/test_datafusion_pipeline.rs b/crates/lance-graph/tests/test_datafusion_pipeline.rs index 54fa6bec2..330f703e5 100644 --- a/crates/lance-graph/tests/test_datafusion_pipeline.rs +++ b/crates/lance-graph/tests/test_datafusion_pipeline.rs @@ -4541,53 +4541,4 @@ async fn test_unimplemented_scalar_function_errors() { ); } -#[tokio::test] -async fn test_tolower_works_in_simple_executor() { - let person_batch = create_person_dataset(); - let config = GraphConfig::builder() - .with_node_label("Person", "id") - .build() - .unwrap(); - - let query = CypherQuery::new( - "MATCH (p:Person) RETURN p.name AS name, tolower(p.name) AS lowered ORDER BY name", - ) - .unwrap() - .with_config(config); - - let mut datasets = HashMap::new(); - datasets.insert("Person".to_string(), person_batch); - - let result = query - .execute(datasets, Some(ExecutionStrategy::Simple)) - .await - .unwrap(); - - let names = result - .column_by_name("name") - .unwrap() - .as_any() - .downcast_ref::() - .unwrap(); - let lowered = result - .column_by_name("lowered") - .unwrap() - .as_any() - .downcast_ref::() - .unwrap(); - - let got: Vec<(String, String)> = (0..result.num_rows()) - .map(|i| (names.value(i).to_string(), lowered.value(i).to_string())) - .collect(); - - assert_eq!( - got, - vec![ - ("Alice".to_string(), "alice".to_string()), - ("Bob".to_string(), "bob".to_string()), - ("Charlie".to_string(), "charlie".to_string()), - ("David".to_string(), "david".to_string()), - ("Eve".to_string(), "eve".to_string()), - ] - ); -} +// NOTE: Simple executor tests live in `tests/test_simple_executor_pipeline.rs`. diff --git a/crates/lance-graph/tests/test_simple_executor_pipeline.rs b/crates/lance-graph/tests/test_simple_executor_pipeline.rs new file mode 100644 index 000000000..097b1166b --- /dev/null +++ b/crates/lance-graph/tests/test_simple_executor_pipeline.rs @@ -0,0 +1,73 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The Lance Authors + +use std::{collections::HashMap, sync::Arc}; + +use arrow_array::{Int64Array, RecordBatch, StringArray}; +use arrow_schema::{DataType, Field, Schema}; +use lance_graph::{CypherQuery, ExecutionStrategy, GraphConfig}; + +fn create_person_batch() -> RecordBatch { + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Int64, false), + Field::new("name", DataType::Utf8, false), + ])); + + let ids = Arc::new(Int64Array::from(vec![1, 2, 3, 4, 5])); + let names = Arc::new(StringArray::from(vec![ + "Alice", "Bob", "Charlie", "David", "Eve", + ])); + + RecordBatch::try_new(schema, vec![ids, names]).unwrap() +} + +#[tokio::test] +async fn test_tolower_works_in_simple_executor() { + let person_batch = create_person_batch(); + let config = GraphConfig::builder() + .with_node_label("Person", "id") + .build() + .unwrap(); + + let query = CypherQuery::new( + "MATCH (p:Person) RETURN p.name AS name, tolower(p.name) AS lowered ORDER BY name", + ) + .unwrap() + .with_config(config); + + let mut datasets = HashMap::new(); + datasets.insert("Person".to_string(), person_batch); + + let result = query + .execute(datasets, Some(ExecutionStrategy::Simple)) + .await + .unwrap(); + + let names = result + .column_by_name("name") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let lowered = result + .column_by_name("lowered") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + + let got: Vec<(String, String)> = (0..result.num_rows()) + .map(|i| (names.value(i).to_string(), lowered.value(i).to_string())) + .collect(); + + assert_eq!( + got, + vec![ + ("Alice".to_string(), "alice".to_string()), + ("Bob".to_string(), "bob".to_string()), + ("Charlie".to_string(), "charlie".to_string()), + ("David".to_string(), "david".to_string()), + ("Eve".to_string(), "eve".to_string()), + ] + ); +} From 633c3e9d4903d86720b32b5ea37b8842afef2522 Mon Sep 17 00:00:00 2001 From: beinan Date: Thu, 29 Jan 2026 07:13:08 +0000 Subject: [PATCH 09/11] fix: update semantic for post-WITH reading clauses --- crates/lance-graph/src/semantic.rs | 20 ++++++++++++++----- .../lance-graph/src/simple_executor/expr.rs | 4 +++- 2 files changed, 18 insertions(+), 6 deletions(-) diff --git a/crates/lance-graph/src/semantic.rs b/crates/lance-graph/src/semantic.rs index 293ba0b50..edea213fe 100644 --- a/crates/lance-graph/src/semantic.rs +++ b/crates/lance-graph/src/semantic.rs @@ -105,11 +105,20 @@ impl SemanticAnalyzer { } } - // Phase 4: Variable discovery in post-WITH MATCH clauses (query chaining) + // Phase 4: Variable discovery in post-WITH READING clauses (query chaining) self.current_scope = ScopeType::Match; - for match_clause in &query.post_with_match_clauses { - if let Err(e) = self.analyze_match_clause(match_clause) { - errors.push(format!("Post-WITH MATCH clause error: {}", e)); + for clause in &query.post_with_reading_clauses { + match clause { + ReadingClause::Match(match_clause) => { + if let Err(e) = self.analyze_match_clause(match_clause) { + errors.push(format!("Post-WITH MATCH clause error: {}", e)); + } + } + ReadingClause::Unwind(unwind_clause) => { + if let Err(e) = self.analyze_unwind_clause(unwind_clause) { + errors.push(format!("Post-WITH UNWIND clause error: {}", e)); + } + } } } @@ -1210,7 +1219,8 @@ mod tests { }; let result = analyze_return_with_match("n", "Person", expr).unwrap(); assert!( - result.errors + result + .errors .iter() .any(|e| e.to_lowercase().contains("not implemented")), "Expected semantic validation to reject unimplemented function, got: {:?}", diff --git a/crates/lance-graph/src/simple_executor/expr.rs b/crates/lance-graph/src/simple_executor/expr.rs index 0ddc2c73c..bcdd2915f 100644 --- a/crates/lance-graph/src/simple_executor/expr.rs +++ b/crates/lance-graph/src/simple_executor/expr.rs @@ -249,7 +249,9 @@ mod tests { let expr = ValueExpression::Arithmetic { left: Box::new(ValueExpression::Variable("x".to_string())), operator: ArithmeticOperator::Add, - right: Box::new(ValueExpression::Literal(crate::ast::PropertyValue::Integer(1))), + right: Box::new(ValueExpression::Literal( + crate::ast::PropertyValue::Integer(1), + )), }; let df_expr = to_df_value_expr_simple(&expr); assert!(matches!(df_expr, Expr::BinaryExpr(_))); From 136dde894e6f1d67673ea0780907f83cb950d53f Mon Sep 17 00:00:00 2001 From: beinan Date: Thu, 29 Jan 2026 18:48:40 +0000 Subject: [PATCH 10/11] test: restore UNWIND coverage --- .../tests/test_datafusion_pipeline.rs | 184 ++++++++++++++++++ 1 file changed, 184 insertions(+) diff --git a/crates/lance-graph/tests/test_datafusion_pipeline.rs b/crates/lance-graph/tests/test_datafusion_pipeline.rs index 330f703e5..3ff9bdd68 100644 --- a/crates/lance-graph/tests/test_datafusion_pipeline.rs +++ b/crates/lance-graph/tests/test_datafusion_pipeline.rs @@ -4542,3 +4542,187 @@ async fn test_unimplemented_scalar_function_errors() { } // NOTE: Simple executor tests live in `tests/test_simple_executor_pipeline.rs`. + +// ============================================================================ +// UNWIND Tests +// ============================================================================ + +#[tokio::test] +async fn test_unwind_simple_list() { + let config = create_graph_config(); + // We need at least one dataset to initialize the catalog/context even if not used in query + let person_batch = create_person_dataset(); + + let query = CypherQuery::new("UNWIND [1, 2, 3] AS x RETURN x") + .unwrap() + .with_config(config); + + let mut datasets = HashMap::new(); + datasets.insert("Person".to_string(), person_batch); + + let result = query + .execute(datasets, Some(ExecutionStrategy::DataFusion)) + .await + .unwrap(); + + assert_eq!(result.num_rows(), 3); + assert_eq!(result.num_columns(), 1); + + // Check the column type - UNWIND returns different types based on implementation + let col = result.column(0); + let data_type = col.data_type(); + + // Try to get values as Int64, Float32, or Float64 + if let Some(int_values) = col.as_any().downcast_ref::() { + let result_values: Vec = (0..result.num_rows()) + .map(|i| int_values.value(i)) + .collect(); + assert_eq!(result_values, vec![1, 2, 3]); + } else if let Some(float_values) = col.as_any().downcast_ref::() { + let result_values: Vec = (0..result.num_rows()) + .map(|i| float_values.value(i)) + .collect(); + assert_eq!(result_values, vec![1.0, 2.0, 3.0]); + } else if let Some(float_values) = col.as_any().downcast_ref::() { + let result_values: Vec = (0..result.num_rows()) + .map(|i| float_values.value(i)) + .collect(); + assert_eq!(result_values, vec![1.0, 2.0, 3.0]); + } else { + panic!("Unexpected column type: {:?}", data_type); + } +} + +#[tokio::test] +async fn test_unwind_after_match() { + let config = create_graph_config(); + let person_batch = create_person_dataset(); + + // Alice, Bob + // UNWIND [10, 20] -> Cartesian product: (Alice, 10), (Alice, 20), (Bob, 10), (Bob, 20) + let query = CypherQuery::new("MATCH (p:Person) UNWIND [10, 20] AS x RETURN p.name, x") + .unwrap() + .with_config(config); + + let mut datasets = HashMap::new(); + datasets.insert("Person".to_string(), person_batch); + + let result = query + .execute(datasets, Some(ExecutionStrategy::DataFusion)) + .await + .unwrap(); + + assert_eq!(result.num_rows(), 10); // 5 people * 2 values = 10 rows + assert_eq!(result.num_columns(), 2); + + let names = result + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + + // Try Int64, Float32, or Float64 for the unwound values + let mut rows: Vec<(String, i32)> = if let Some(int_values) = + result.column(1).as_any().downcast_ref::() + { + (0..result.num_rows()) + .map(|i| (names.value(i).to_string(), int_values.value(i) as i32)) + .collect() + } else if let Some(float_values) = result + .column(1) + .as_any() + .downcast_ref::() + { + (0..result.num_rows()) + .map(|i| (names.value(i).to_string(), float_values.value(i) as i32)) + .collect() + } else if let Some(float_values) = result.column(1).as_any().downcast_ref::() { + (0..result.num_rows()) + .map(|i| (names.value(i).to_string(), float_values.value(i) as i32)) + .collect() + } else { + panic!( + "Unexpected column type for unwound values: {:?}", + result.column(1).data_type() + ); + }; + + rows.sort(); + + let expected = vec![ + ("Alice".to_string(), 10), + ("Alice".to_string(), 20), + ("Bob".to_string(), 10), + ("Bob".to_string(), 20), + ("Charlie".to_string(), 10), + ("Charlie".to_string(), 20), + ("David".to_string(), 10), + ("David".to_string(), 20), + ("Eve".to_string(), 10), + ("Eve".to_string(), 20), + ]; + + assert_eq!(rows, expected); +} + +#[tokio::test] +async fn test_unwind_then_match() { + let config = create_graph_config(); + let person_batch = create_person_dataset(); + + // UNWIND [1, 2] usually yields Float64Array in current parser/planner pipeline + // p.id is Int64 + // We expect DataFusion to handle comparison (maybe with cast) + let query = CypherQuery::new("UNWIND [1, 2] AS target_id MATCH (p:Person) WHERE p.id = target_id RETURN p.name, target_id") + .unwrap() + .with_config(config); + + let mut datasets = HashMap::new(); + datasets.insert("Person".to_string(), person_batch); + + let result = query + .execute(datasets, Some(ExecutionStrategy::DataFusion)) + .await + .unwrap(); + + // We expect 2 matches: Alice (id=1) and Bob (id=2) + assert_eq!(result.num_rows(), 2); + + let names = result + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + + // target_id from Unwind might be Int64, Float32, or Float64 + let mut rows: Vec<(String, i32)> = + if let Some(int_ids) = result.column(1).as_any().downcast_ref::() { + (0..result.num_rows()) + .map(|i| (names.value(i).to_string(), int_ids.value(i) as i32)) + .collect() + } else if let Some(float_ids) = result + .column(1) + .as_any() + .downcast_ref::() + { + (0..result.num_rows()) + .map(|i| (names.value(i).to_string(), float_ids.value(i) as i32)) + .collect() + } else if let Some(float_ids) = result.column(1).as_any().downcast_ref::() + { + (0..result.num_rows()) + .map(|i| (names.value(i).to_string(), float_ids.value(i) as i32)) + .collect() + } else { + panic!( + "Unexpected column type for target_id: {:?}", + result.column(1).data_type() + ); + }; + + rows.sort(); + + let expected = vec![("Alice".to_string(), 1), ("Bob".to_string(), 2)]; + + assert_eq!(rows, expected); +} From 36923492cddb29878ff3f2b30fab4333e144bc1c Mon Sep 17 00:00:00 2001 From: beinan Date: Thu, 29 Jan 2026 19:54:57 +0000 Subject: [PATCH 11/11] style: format unwind test --- crates/lance-graph/tests/test_datafusion_pipeline.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/crates/lance-graph/tests/test_datafusion_pipeline.rs b/crates/lance-graph/tests/test_datafusion_pipeline.rs index 3ff9bdd68..ae439332c 100644 --- a/crates/lance-graph/tests/test_datafusion_pipeline.rs +++ b/crates/lance-graph/tests/test_datafusion_pipeline.rs @@ -4708,8 +4708,7 @@ async fn test_unwind_then_match() { (0..result.num_rows()) .map(|i| (names.value(i).to_string(), float_ids.value(i) as i32)) .collect() - } else if let Some(float_ids) = result.column(1).as_any().downcast_ref::() - { + } else if let Some(float_ids) = result.column(1).as_any().downcast_ref::() { (0..result.num_rows()) .map(|i| (names.value(i).to_string(), float_ids.value(i) as i32)) .collect()