feat(cypher): run the official ldbc ic queries as opencypher text

README.md

@@ -23,8 +23,11 @@ DataFusion as extension operators, falling back to a join for any hop the CSR ca
   and `refresh()`es to new dataset versions.
 - **Smarter planning**: patterns start at their most selective node, and DataFusion gets column statistics to size
   its joins.
-- **Much wider Cypher**, enough for all 14 LDBC Interactive complex queries: `OPTIONAL MATCH`, `EXISTS`, chained
-  `WITH`, relationship alternation, `shortestPath` and more.
+- **Much wider Cypher**, enough to run the official LDBC Interactive complex queries (IC14 without `reduce()`,
+  which openCypher leaves out): `OPTIONAL MATCH`, `EXISTS` and pattern predicates, chained `WITH`, `UNWIND`,
+  list and pattern comprehensions, maps, relationship alternation, `shortestPath`, `$parameters` and more,
+  following openCypher's
+  [standardisation scope](https://github.com/opencypher/openCypher/blob/master/docs/standardisation-scope.adoc).
 - **Correctness fixes** to behavior inherited from lance-graph (for example, undirected patterns dropped incoming
   edges), each with a regression test.
 - **Current dependencies** (Lance 13, DataFusion 54, Arrow 58, PyO3 0.28), with Delta Lake and Unity Catalog support

crates/lance-graph-python/src/graph.rs

@@ -824,6 +824,12 @@ fn python_to_json(value: &Bound<'_, PyAny>) -> PyResult<JsonValue> {
         ))
     } else if let Ok(s) = value.extract::<String>() {
         Ok(JsonValue::String(s))
+    } else if value.hasattr("isoformat")? {
+        // datetime and date: ISO 8601 text, which DataFusion casts when comparing with
+        // timestamp and date columns.
+        Ok(JsonValue::String(
+            value.call_method0("isoformat")?.extract::<String>()?,
+        ))
     } else {
         Err(PyValueError::new_err("Unsupported parameter type"))
     }
@@ -1877,9 +1883,19 @@ impl CypherEngine {
             .clone()
     }
 
-    fn parse(&self, query: &str, state: &EngineState) -> PyResult<RustCypherQuery> {
+    fn parse(
+        &self,
+        query: &str,
+        params: Option<&Bound<'_, PyDict>>,
+        state: &EngineState,
+    ) -> PyResult<RustCypherQuery> {
+        let mut parameters = HashMap::new();
+        for (key, value) in params.into_iter().flat_map(|p| p.iter()) {
+            parameters.insert(key.extract::<String>()?, python_to_json(&value)?);
+        }
         let query = RustCypherQuery::new(query)
             .map_err(graph_error_to_pyerr)?
+            .with_parameters(parameters)
             .with_config(self.config.clone())
             .with_fan_out(state.fan_out());
         Ok(if self.use_csr {
@@ -1939,15 +1955,30 @@ impl CypherEngine {
     ///
     /// The whole result is held in memory; use ``execute_batches`` for large outputs.
     ///
+    /// Parameters
+    /// ----------
+    /// query : str
+    ///     The Cypher query
+    /// params : dict, optional
+    ///     Values for ``$name`` parameters in the query: numbers, strings, booleans,
+    ///     ``None``, and ``datetime``/``date`` values (compared as timestamps and dates).
+    ///     ``execute_batches`` and ``explain`` take the same argument.
+    ///
     /// Raises
     /// ------
     /// ValueError
     ///     If the query is invalid
     /// RuntimeError
     ///     If query execution fails
-    fn execute(&self, py: Python, query: &str) -> PyResult<Py<PyAny>> {
+    #[pyo3(signature = (query, params = None))]
+    fn execute(
+        &self,
+        py: Python,
+        query: &str,
+        params: Option<&Bound<'_, PyDict>>,
+    ) -> PyResult<Py<PyAny>> {
         let state = self.snapshot();
-        let cypher_query = self.parse(query, &state)?;
+        let cypher_query = self.parse(query, params, &state)?;
         let result_batch = RT
             .block_on(
                 Some(py),
@@ -1965,11 +1996,17 @@ impl CypherEngine {
     /// stream with bounded memory; joins, sorts and aggregations may still buffer.
     /// The reader keeps its snapshot alive, so a concurrent ``refresh()`` does not
     /// affect it. Planning errors raise immediately; execution errors raise while reading.
-    fn execute_batches(&self, py: Python, query: &str) -> PyResult<Py<PyAny>> {
+    #[pyo3(signature = (query, params = None))]
+    fn execute_batches(
+        &self,
+        py: Python,
+        query: &str,
+        params: Option<&Bound<'_, PyDict>>,
+    ) -> PyResult<Py<PyAny>> {
         use arrow::pyarrow::IntoPyArrow;
 
         let state = self.snapshot();
-        let cypher_query = self.parse(query, &state)?;
+        let cypher_query = self.parse(query, params, &state)?;
         let stream = RT
             .block_on(
                 Some(py),
@@ -1987,9 +2024,15 @@ impl CypherEngine {
     }
 
     /// Return the query plan, including the physical plan with pushed-down scans.
-    fn explain(&self, py: Python, query: &str) -> PyResult<String> {
+    #[pyo3(signature = (query, params = None))]
+    fn explain(
+        &self,
+        py: Python,
+        query: &str,
+        params: Option<&Bound<'_, PyDict>>,
+    ) -> PyResult<String> {
         let state = self.snapshot();
-        let cypher_query = self.parse(query, &state)?;
+        let cypher_query = self.parse(query, params, &state)?;
         RT.block_on(
             Some(py),
             cypher_query
@@ -2111,7 +2154,7 @@ impl CypherEngine {
         vector_search: &VectorSearch,
     ) -> PyResult<Py<PyAny>> {
         let state = self.snapshot();
-        let cypher_query = self.parse(query, &state)?;
+        let cypher_query = self.parse(query, None, &state)?;
         let vs = vector_search.inner.clone();
 
         // Execute query to get candidates, then apply vector reranking

crates/lance-graph/src/ast.rs

@@ -238,6 +238,9 @@ pub enum PropertyValue {
     Parameter(String),
     /// Property reference (e.g., node.property)
     Property(PropertyRef),
+    /// A variable from an earlier clause, as an inline property value (`{id: tagId}`).
+    /// Like `Property` values, it is moved into the clause's WHERE before planning.
+    Variable(String),
 }
 
 /// Reference to a property of a node or relationship
@@ -284,6 +287,12 @@ pub enum BooleanExpression {
         expression: ValueExpression,
         list: Vec<ValueExpression>,
     },
+    /// `x IN list` where the list is any expression, such as a collected alias
+    /// (`friend IN friends`) rather than a literal `[...]`.
+    Member {
+        expression: ValueExpression,
+        list: ValueExpression,
+    },
     /// LIKE pattern matching (case-sensitive)
     Like {
         expression: ValueExpression,
@@ -389,6 +398,30 @@ pub enum ValueExpression {
     },
     /// Parameter reference for query parameters (e.g., $query_vector)
     Parameter(String),
+    /// A predicate used as a value: `path IS NULL`, `NOT (a)-[:R]-(b) AS isNew`.
+    Boolean(Box<BooleanExpression>),
+    /// A map literal `{key: value, ...}`; a node as a value inside it keeps its properties.
+    Map(Vec<(String, ValueExpression)>),
+    /// `value.field` on something other than a node or relationship variable: a map key,
+    /// or a component of a date or timestamp (`birthday.month`).
+    Field {
+        base: Box<ValueExpression>,
+        field: String,
+    },
+    /// `[(a)-[:R]->(b) WHERE predicate | value]`: the values for each match of the pattern
+    /// from this row's bound variables.
+    PatternComprehension {
+        pattern: Box<PathPattern>,
+        filter: Option<Box<BooleanExpression>>,
+        map: Box<ValueExpression>,
+    },
+    /// `[x IN list WHERE predicate | value]`, both parts optional.
+    ListComprehension {
+        variable: String,
+        list: Box<ValueExpression>,
+        filter: Option<Box<BooleanExpression>>,
+        map: Option<Box<ValueExpression>>,
+    },
     /// Vector literal: [0.1, 0.2, 0.3]
     /// Represents an inline vector for similarity search
     VectorLiteral(Vec<f32>),
@@ -410,7 +443,8 @@ pub fn classify_function(name: &str) -> FunctionType {
     match name.to_lowercase().as_str() {
         "count" | "sum" | "avg" | "min" | "max" | "collect" => FunctionType::Aggregate,
         "tolower" | "lower" | "toupper" | "upper" | "coalesce" | "date_part" | "to_epoch_ms"
-        | "length" | "nodes" | "size" | "range" | "__index" | "__list" | "tostring" => {
+        | "length" | "nodes" | "size" | "range" | "__index" | "__list" | "tostring" | "head"
+        | "tointeger" | "tofloat" | "date" | "datetime" | "duration.inseconds" => {
             FunctionType::Scalar
         }
         // Vector functions are handled separately as special variants
@@ -562,6 +596,94 @@ impl PropertyRef {
     }
 }
 
+/// Apply `f` to every value inside `expr`, children before parents, including the values
+/// inside predicates. Patterns of EXISTS subqueries are left alone.
+pub fn transform_value(expr: &mut ValueExpression, f: &mut dyn FnMut(&mut ValueExpression)) {
+    match expr {
+        ValueExpression::ScalarFunction { args, .. }
+        | ValueExpression::AggregateFunction { args, .. } => {
+            args.iter_mut().for_each(|a| transform_value(a, f))
+        }
+        ValueExpression::Arithmetic { left, right, .. }
+        | ValueExpression::VectorDistance { left, right, .. }
+        | ValueExpression::VectorSimilarity { left, right, .. } => {
+            transform_value(left, f);
+            transform_value(right, f);
+        }
+        ValueExpression::Case { whens, otherwise } => {
+            for (condition, then) in whens {
+                transform_boolean_values(condition, f);
+                transform_value(then, f);
+            }
+            if let Some(e) = otherwise {
+                transform_value(e, f);
+            }
+        }
+        ValueExpression::Boolean(b) => transform_boolean_values(b, f),
+        ValueExpression::Map(entries) => {
+            entries.iter_mut().for_each(|(_, v)| transform_value(v, f))
+        }
+        ValueExpression::Field { base, .. } => transform_value(base, f),
+        ValueExpression::ListComprehension {
+            list, filter, map, ..
+        } => {
+            transform_value(list, f);
+            if let Some(b) = filter {
+                transform_boolean_values(b, f);
+            }
+            if let Some(m) = map {
+                transform_value(m, f);
+            }
+        }
+        ValueExpression::PatternComprehension { filter, map, .. } => {
+            if let Some(b) = filter {
+                transform_boolean_values(b, f);
+            }
+            transform_value(map, f);
+        }
+        ValueExpression::Variable(_)
+        | ValueExpression::Property(_)
+        | ValueExpression::Literal(_)
+        | ValueExpression::Parameter(_)
+        | ValueExpression::VectorLiteral(_) => {}
+    }
+    f(expr);
+}
+
+/// [`transform_value`] over every value in a predicate.
+pub fn transform_boolean_values(
+    expr: &mut BooleanExpression,
+    f: &mut dyn FnMut(&mut ValueExpression),
+) {
+    match expr {
+        BooleanExpression::Comparison { left, right, .. } => {
+            transform_value(left, f);
+            transform_value(right, f);
+        }
+        BooleanExpression::And(a, b) | BooleanExpression::Or(a, b) => {
+            transform_boolean_values(a, f);
+            transform_boolean_values(b, f);
+        }
+        BooleanExpression::Not(inner) => transform_boolean_values(inner, f),
+        BooleanExpression::In { expression, list } => {
+            transform_value(expression, f);
+            list.iter_mut().for_each(|v| transform_value(v, f));
+        }
+        BooleanExpression::Member { expression, list } => {
+            transform_value(expression, f);
+            transform_value(list, f);
+        }
+        BooleanExpression::Like { expression, .. }
+        | BooleanExpression::ILike { expression, .. }
+        | BooleanExpression::Contains { expression, .. }
+        | BooleanExpression::StartsWith { expression, .. }
+        | BooleanExpression::EndsWith { expression, .. }
+        | BooleanExpression::IsNull(expression)
+        | BooleanExpression::IsNotNull(expression) => transform_value(expression, f),
+        BooleanExpression::Exists(_) | BooleanExpression::PatternExists { .. } => {}
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/lance-graph/src/datafusion_planner/analysis.rs

@@ -38,13 +38,16 @@ pub struct RelationshipInstance {
 pub struct PlanningContext<'a> {
     pub analysis: &'a QueryAnalysis,
     pub(crate) relationship_instance_idx: HashMap<String, usize>,
+    /// Element rows of the list comprehensions being built, innermost last.
+    pub(crate) comprehension_elements: Vec<datafusion::logical_expr::LogicalPlan>,
 }
 
 impl<'a> PlanningContext<'a> {
     pub fn new(analysis: &'a QueryAnalysis) -> Self {
         Self {
             analysis,
             relationship_instance_idx: HashMap::new(),
+            comprehension_elements: Vec::new(),
         }
     }
 
@@ -206,10 +209,19 @@ fn analyze_operator(
             analyze_operator(left, analysis, rel_counter)?;
             analyze_operator(right, analysis, rel_counter)?;
         }
-        LogicalOperator::Unwind { input, .. } => {
+        LogicalOperator::Unwind {
+            input,
+            alias,
+            label,
+            ..
+        } => {
             if let Some(op) = input {
                 analyze_operator(op, analysis, rel_counter)?;
             }
+            if let Some(label) = label {
+                analysis.var_to_label.insert(alias.clone(), label.clone());
+                analysis.required_datasets.insert(label.clone());
+            }
         }
         LogicalOperator::ShortestPath { input, .. } => {
             analyze_operator(input, analysis, rel_counter)?;
@@ -227,6 +239,24 @@ fn analyze_operator(
                 analyze_operator(input, analysis, rel_counter)?;
             }
         }
+        // Same order as `build_list_comprehension`: the rows, then the element plan.
+        LogicalOperator::ListComprehension {
+            input,
+            variable,
+            label,
+            element,
+            ..
+        } => {
+            analyze_operator(input, analysis, rel_counter)?;
+            if let Some(label) = label {
+                analysis
+                    .var_to_label
+                    .insert(variable.clone(), label.clone());
+                analysis.required_datasets.insert(label.clone());
+            }
+            analyze_operator(element, analysis, rel_counter)?;
+        }
+        LogicalOperator::ComprehensionElement => {}
     }
     Ok(())
 }