Add numeric scalar functions with strict Cypher types

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))]

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",

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

crates/lance-graph/src/ast.rs

@@ -443,6 +443,23 @@ pub enum ValueExpression {
         filter: Option<Box<BooleanExpression>>,
         map: Option<Box<ValueExpression>>,
     },
+    /// 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<ValueExpression>,
+        variable: String,
+        list: Box<ValueExpression>,
+        expression: Box<ValueExpression>,
+    },
+    /// `all/any/none/single(variable IN list WHERE predicate)`.
+    ListPredicate {
+        kind: String,
+        variable: String,
+        list: Box<ValueExpression>,
+        predicate: Box<ValueExpression>,
+    },
     /// Vector literal: [0.1, 0.2, 0.3]
     /// Represents an inline vector for similarity search
     VectorLiteral(Vec<f32>),
@@ -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,

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::<Result<Vec<_>>>()?;
+        LogicalPlanBuilder::from(aggregate)
             .project(final_projection)
             .map_err(|e| self.plan_error("Failed to project after aggregate", e))?
             .build()

crates/lance-graph/src/datafusion_planner/builder/basic_ops.rs

@@ -86,8 +86,7 @@ impl DataFusionPlanner {
         predicate: &crate::ast::BooleanExpression,
     ) -> Result<LogicalPlan> {
         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::<Result<Vec<_>>>()?;
             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 {

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)])

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())