diff --git a/.bumpversion.toml b/.bumpversion.toml index 789c46b..01fee30 100644 --- a/.bumpversion.toml +++ b/.bumpversion.toml @@ -1,5 +1,5 @@ [tool.bumpversion] -current_version = "0.3.0" +current_version = "0.3.1" parse = "(?P\\d+)\\.(?P\\d+)\\.(?P\\d+)(-(?Palpha|beta|rc)\\.(?P\\d+))?" serialize = [ "{major}.{minor}.{patch}-{pre_label}.{pre_n}", diff --git a/Cargo.lock b/Cargo.lock index c11b0c0..dd3b648 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3947,7 +3947,7 @@ dependencies = [ [[package]] name = "lance-csr" -version = "0.3.0" +version = "0.3.1" dependencies = [ "ahash", "arrow", @@ -3976,7 +3976,7 @@ dependencies = [ [[package]] name = "lance-csr-benches" -version = "0.3.0" +version = "0.3.1" dependencies = [ "arrow-array", "arrow-schema", @@ -3991,7 +3991,7 @@ dependencies = [ [[package]] name = "lance-csr-catalog" -version = "0.3.0" +version = "0.3.1" dependencies = [ "arrow-schema", "async-trait", diff --git a/crates/lance-csr-benches/Cargo.toml b/crates/lance-csr-benches/Cargo.toml index 6ebe260..0d907f4 100644 --- a/crates/lance-csr-benches/Cargo.toml +++ b/crates/lance-csr-benches/Cargo.toml @@ -1,12 +1,12 @@ [package] name = "lance-csr-benches" -version = "0.3.0" +version = "0.3.1" edition = "2021" publish = false description = "Benchmarks for lance-csr (not published)" [dependencies] -lance-csr = { path = "../lance-csr", version = "0.3.0" } +lance-csr = { path = "../lance-csr", version = "0.3.1" } lance = "13.0.0" arrow-array = "58" arrow-schema = "58" diff --git a/crates/lance-csr-catalog/Cargo.toml b/crates/lance-csr-catalog/Cargo.toml index c15ef7b..4de57e7 100644 --- a/crates/lance-csr-catalog/Cargo.toml +++ b/crates/lance-csr-catalog/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "lance-csr-catalog" -version = "0.3.0" +version = "0.3.1" edition = "2021" license = "Apache-2.0" authors = ["The lance-csr Authors"] diff --git a/crates/lance-csr-python/Cargo.lock b/crates/lance-csr-python/Cargo.lock index 8645b15..3042552 100644 --- a/crates/lance-csr-python/Cargo.lock +++ b/crates/lance-csr-python/Cargo.lock @@ -3832,7 +3832,7 @@ dependencies = [ [[package]] name = "lance-csr" -version = "0.3.0" +version = "0.3.1" dependencies = [ "ahash", "arrow", @@ -3859,7 +3859,7 @@ dependencies = [ [[package]] name = "lance-csr-catalog" -version = "0.3.0" +version = "0.3.1" dependencies = [ "arrow-schema", "async-trait", @@ -3872,7 +3872,7 @@ dependencies = [ [[package]] name = "lance-csr-python" -version = "0.3.0" +version = "0.3.1" dependencies = [ "arrow", "arrow-array", diff --git a/crates/lance-csr-python/Cargo.toml b/crates/lance-csr-python/Cargo.toml index 8767642..321318f 100644 --- a/crates/lance-csr-python/Cargo.toml +++ b/crates/lance-csr-python/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "lance-csr-python" -version = "0.3.0" +version = "0.3.1" edition = "2021" authors = ["The lance-csr Authors"] publish = false diff --git a/crates/lance-csr/Cargo.toml b/crates/lance-csr/Cargo.toml index 12b0f61..71308d2 100644 --- a/crates/lance-csr/Cargo.toml +++ b/crates/lance-csr/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "lance-csr" -version = "0.3.0" +version = "0.3.1" edition = "2021" license = "Apache-2.0" authors = ["The lance-csr Authors"] @@ -32,7 +32,7 @@ datafusion-expr = "54" datafusion-sql = "54" datafusion-functions-aggregate = "54" futures = "0.3" -lance-csr-catalog = { path = "../lance-csr-catalog", version = "0.3.0" } +lance-csr-catalog = { path = "../lance-csr-catalog", version = "0.3.1" } lance = "13.0.0" lance-linalg = "13.0.0" lance-namespace = "13.0.0" diff --git a/crates/lance-csr/src/datafusion_planner/builder/aggregate_ops.rs b/crates/lance-csr/src/datafusion_planner/builder/aggregate_ops.rs index e18d4fe..d758f07 100644 --- a/crates/lance-csr/src/datafusion_planner/builder/aggregate_ops.rs +++ b/crates/lance-csr/src/datafusion_planner/builder/aggregate_ops.rs @@ -36,6 +36,27 @@ fn split_aggregates(expr: &mut ValueExpression, aggregates: &mut Vec<(String, Va }); } +/// `count(v)` lowers to `count(v__id)`, which only nodes and relationships have. For a +/// value without an id column (a path, or an UNWIND or WITH value), count the value itself. +fn count_values_without_ids( + args: &mut [datafusion::logical_expr::Expr], + schema: &datafusion::common::DFSchema, +) { + for arg in args { + let datafusion::logical_expr::Expr::Column(column) = arg else { + continue; + }; + let Some(variable) = column.name.strip_suffix("__id") else { + continue; + }; + if !schema.has_column_with_unqualified_name(&column.name) + && schema.has_column_with_unqualified_name(variable) + { + *arg = col(datafusion::common::Column::new_unqualified(variable)); + } + } +} + impl DataFusionPlanner { pub(crate) fn build_project_with_aggregates( &self, @@ -81,11 +102,14 @@ impl DataFusionPlanner { split_aggregates(&mut rewritten, &mut aggregates); for (column, aggregate) in &aggregates[first..] { 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 { + // collect() gathers in the order of a preceding `WITH ... ORDER BY`. if f.func.name() == "array_agg" && !f.params.distinct { f.params.order_by = ordering.to_vec(); } + if f.func.name() == "count" { + count_values_without_ids(&mut f.params.args, &schema); + } } agg_exprs.push(aggregate.alias(column)); } diff --git a/crates/lance-csr/src/datafusion_planner/builder/basic_ops.rs b/crates/lance-csr/src/datafusion_planner/builder/basic_ops.rs index cd4b6f5..0408c43 100644 --- a/crates/lance-csr/src/datafusion_planner/builder/basic_ops.rs +++ b/crates/lance-csr/src/datafusion_planner/builder/basic_ops.rs @@ -346,10 +346,12 @@ impl DataFusionPlanner { .map(|item| { let bound = super::super::expression::bind_value(&item.expression, input_plan.schema()); + let asc = matches!(item.direction, crate::ast::SortDirection::Ascending); + // openCypher orders null after every value: last ascending, first descending. SortExpr { expr: self.unresolved_value_expr(&bound), - asc: matches!(item.direction, crate::ast::SortDirection::Ascending), - nulls_first: true, + asc, + nulls_first: !asc, } }) .collect() diff --git a/crates/lance-csr/src/datafusion_planner/expression.rs b/crates/lance-csr/src/datafusion_planner/expression.rs index 7eba94a..4617199 100644 --- a/crates/lance-csr/src/datafusion_planner/expression.rs +++ b/crates/lance-csr/src/datafusion_planner/expression.rs @@ -327,13 +327,8 @@ pub(crate) fn to_df_value_expr(expr: &ValueExpression) -> Expr { // Use __id as a null-sensitive column // This ensures optional matches with NULL variables are not counted // - // LIMITATION: This assumes all bindings expose a __id column. - // This is true for ScanByLabel and Expand, but NOT for: - // - UNWIND-created variables (e.g., UNWIND [1,2,3] AS x) - // - WITH-projected aliases (e.g., WITH 1 AS x) - // Those cases will produce a runtime "column not found" error. - // TODO: Consider using COUNT(1) for non-node variables, or add - // semantic validation to reject COUNT(variable) for non-node types. + // Paths, UNWIND and WITH values have no __id column; the + // aggregate builder, which sees the schema, counts the value instead. // Use qualify_column for consistent case-insensitive normalization col(qualify_column(v, "id")) } diff --git a/crates/lance-csr/tests/test_datafusion_pipeline.rs b/crates/lance-csr/tests/test_datafusion_pipeline.rs index 56570d5..d2e762a 100644 --- a/crates/lance-csr/tests/test_datafusion_pipeline.rs +++ b/crates/lance-csr/tests/test_datafusion_pipeline.rs @@ -2786,24 +2786,24 @@ async fn test_count_with_grouping() { .unwrap(); // NULL city: 1 person (David) - assert!(city_col.is_null(0)); - assert_eq!(count_col.value(0), 1); + assert!(city_col.is_null(4)); + assert_eq!(count_col.value(4), 1); // Chicago: 1 person (Charlie) - assert_eq!(city_col.value(1), "Chicago"); - assert_eq!(count_col.value(1), 1); + assert_eq!(city_col.value(0), "Chicago"); + assert_eq!(count_col.value(0), 1); // New York: 1 person (Alice) - assert_eq!(city_col.value(2), "New York"); - assert_eq!(count_col.value(2), 1); + assert_eq!(city_col.value(1), "New York"); + assert_eq!(count_col.value(1), 1); // San Francisco: 1 person (Bob) - assert_eq!(city_col.value(3), "San Francisco"); - assert_eq!(count_col.value(3), 1); + assert_eq!(city_col.value(2), "San Francisco"); + assert_eq!(count_col.value(2), 1); // Seattle: 1 person (Eve) - assert_eq!(city_col.value(4), "Seattle"); - assert_eq!(count_col.value(4), 1); + assert_eq!(city_col.value(3), "Seattle"); + assert_eq!(count_col.value(3), 1); } #[tokio::test] @@ -3148,21 +3148,21 @@ async fn test_sum_with_grouping() { .downcast_ref::() .unwrap(); - // Verify grouping results (ordered by city, NULL comes first) - assert!(city_col.is_null(0)); // David: 40 (NULL city) - assert_eq!(sum_col.value(0), 40); + // Verify grouping results (ordered by city, NULL sorts last, as in openCypher + assert!(city_col.is_null(4)); // David: 40 (NULL city) + assert_eq!(sum_col.value(4), 40); - assert_eq!(city_col.value(1), "Chicago"); // Charlie: 30 - assert_eq!(sum_col.value(1), 30); + assert_eq!(city_col.value(0), "Chicago"); // Charlie: 30 + assert_eq!(sum_col.value(0), 30); - assert_eq!(city_col.value(2), "New York"); // Alice: 25 - assert_eq!(sum_col.value(2), 25); + assert_eq!(city_col.value(1), "New York"); // Alice: 25 + assert_eq!(sum_col.value(1), 25); - assert_eq!(city_col.value(3), "San Francisco"); // Bob: 35 - assert_eq!(sum_col.value(3), 35); + assert_eq!(city_col.value(2), "San Francisco"); // Bob: 35 + assert_eq!(sum_col.value(2), 35); - assert_eq!(city_col.value(4), "Seattle"); // Eve: 28 - assert_eq!(sum_col.value(4), 28); + assert_eq!(city_col.value(3), "Seattle"); // Eve: 28 + assert_eq!(sum_col.value(3), 28); } #[tokio::test] @@ -3297,21 +3297,21 @@ async fn test_avg_with_grouping() { .downcast_ref::() .unwrap(); - // Verify grouping results (ordered by city, NULL comes first) - assert!(city_col.is_null(0)); // David: 40 (NULL city) - assert_eq!(avg_col.value(0), 40.0); + // Verify grouping results (ordered by city, NULL sorts last, as in openCypher + assert!(city_col.is_null(4)); // David: 40 (NULL city) + assert_eq!(avg_col.value(4), 40.0); - assert_eq!(city_col.value(1), "Chicago"); // Charlie: 30 - assert_eq!(avg_col.value(1), 30.0); + assert_eq!(city_col.value(0), "Chicago"); // Charlie: 30 + assert_eq!(avg_col.value(0), 30.0); - assert_eq!(city_col.value(2), "New York"); // Alice: 25 - assert_eq!(avg_col.value(2), 25.0); + assert_eq!(city_col.value(1), "New York"); // Alice: 25 + assert_eq!(avg_col.value(1), 25.0); - assert_eq!(city_col.value(3), "San Francisco"); // Bob: 35 - assert_eq!(avg_col.value(3), 35.0); + assert_eq!(city_col.value(2), "San Francisco"); // Bob: 35 + assert_eq!(avg_col.value(2), 35.0); - assert_eq!(city_col.value(4), "Seattle"); // Eve: 28 - assert_eq!(avg_col.value(4), 28.0); + assert_eq!(city_col.value(3), "Seattle"); // Eve: 28 + assert_eq!(avg_col.value(3), 28.0); } #[tokio::test] @@ -3473,31 +3473,31 @@ async fn test_min_max_with_grouping() { .downcast_ref::() .unwrap(); - // ORDER BY p.city, NULL comes first per your other tests - assert!(city_col_min.is_null(0)); // David city NULL - assert!(city_col_max.is_null(0)); - assert_eq!(min_col_min.value(0), 40); - assert_eq!(min_col_max.value(0), 40); + // ORDER BY p.city, NULL sorts last, as in openCypher + assert!(city_col_min.is_null(4)); // David city NULL + assert!(city_col_max.is_null(4)); + assert_eq!(min_col_min.value(4), 40); + assert_eq!(min_col_max.value(4), 40); - assert_eq!(city_col_min.value(1), "Chicago"); // Charlie - assert_eq!(city_col_max.value(1), "Chicago"); - assert_eq!(min_col_min.value(1), 30); - assert_eq!(min_col_max.value(1), 30); + assert_eq!(city_col_min.value(0), "Chicago"); // Charlie + assert_eq!(city_col_max.value(0), "Chicago"); + assert_eq!(min_col_min.value(0), 30); + assert_eq!(min_col_max.value(0), 30); - assert_eq!(city_col_min.value(2), "New York"); // Alice - assert_eq!(city_col_max.value(2), "New York"); - assert_eq!(min_col_min.value(2), 25); - assert_eq!(min_col_max.value(2), 25); + assert_eq!(city_col_min.value(1), "New York"); // Alice + assert_eq!(city_col_max.value(1), "New York"); + assert_eq!(min_col_min.value(1), 25); + assert_eq!(min_col_max.value(1), 25); - assert_eq!(city_col_min.value(3), "San Francisco"); // Bob - assert_eq!(city_col_max.value(3), "San Francisco"); - assert_eq!(min_col_min.value(3), 35); - assert_eq!(min_col_max.value(3), 35); + assert_eq!(city_col_min.value(2), "San Francisco"); // Bob + assert_eq!(city_col_max.value(2), "San Francisco"); + assert_eq!(min_col_min.value(2), 35); + assert_eq!(min_col_max.value(2), 35); - assert_eq!(city_col_min.value(4), "Seattle"); // Eve - assert_eq!(city_col_max.value(4), "Seattle"); - assert_eq!(min_col_min.value(4), 28); - assert_eq!(min_col_max.value(4), 28); + assert_eq!(city_col_min.value(3), "Seattle"); // Eve + assert_eq!(city_col_max.value(3), "Seattle"); + assert_eq!(min_col_min.value(3), 28); + assert_eq!(min_col_max.value(3), 28); } // ============================================================================ @@ -4823,16 +4823,14 @@ async fn test_unwind_simple_list() { } } -/// Test that COUNT(x) where x comes from UNWIND fails gracefully -/// This documents a known limitation: COUNT(variable) assumes variable__id exists, -/// but UNWIND-created variables don't have __id columns. +/// COUNT(x) where x comes from UNWIND counts the values: UNWIND variables have no +/// `x__id` column, so the aggregate counts the value column itself. #[tokio::test] -async fn test_count_unwind_variable_known_limitation() { +async fn test_count_unwind_variable() { let config = create_graph_config(); let person_batch = create_person_dataset(); - // COUNT(x) where x comes from UNWIND - x has no __id column - let query = CypherQuery::new("UNWIND [1, 2, 3] AS x RETURN COUNT(x)") + let query = CypherQuery::new("UNWIND [1, 2, 3] AS x RETURN COUNT(x) AS n") .unwrap() .with_config(config); @@ -4841,15 +4839,15 @@ async fn test_count_unwind_variable_known_limitation() { let result = query .execute(datasets, Some(ExecutionStrategy::DataFusion)) - .await; - - // This currently fails because COUNT(x) looks for x__id which doesn't exist - // If this starts passing in the future, update the test to verify correct count (3) - assert!( - result.is_err(), - "COUNT(unwind_variable) should fail - no __id column. If this passes, \ - the limitation has been fixed and test should verify COUNT returns 3" - ); + .await + .unwrap(); + let n = result + .column_by_name("n") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(n.value(0), 3); } #[tokio::test] diff --git a/docs/cypher.md b/docs/cypher.md index cd2443f..d0452fb 100644 --- a/docs/cypher.md +++ b/docs/cypher.md @@ -215,7 +215,7 @@ used by IC14. The tables below specify its supported surface. | `UNION`, `UNION ALL` | Combine independent queries with matching output names/order and compatible Arrow types. | | `CALL { ... RETURN ... }` | Read-only subquery. Import outer variables with a leading plain `WITH`; see below. | | `RETURN [DISTINCT]` | With `AS` aliases. Without reading clauses, expressions run on one implicit row. | -| `ORDER BY`, `SKIP`, `LIMIT` | `SKIP` and `LIMIT` take nonnegative integer literals or parameters. | +| `ORDER BY`, `SKIP`, `LIMIT` | Nulls sort last in ascending order and first in descending order, as in openCypher. `SKIP` and `LIMIT` take nonnegative integer literals or parameters. | !!! failure "Not supported" diff --git a/python/pyproject.toml b/python/pyproject.toml index 36acc3c..0d859c7 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "lance-csr" -version = "0.3.0" +version = "0.3.1" dependencies = [ "pyarrow>=14", "pylance>=13.0.0,<14", diff --git a/python/python/tests/test_call_subqueries.py b/python/python/tests/test_call_subqueries.py index 87668e5..9d4a498 100644 --- a/python/python/tests/test_call_subqueries.py +++ b/python/python/tests/test_call_subqueries.py @@ -316,7 +316,7 @@ def test_call_union_coerces_numeric_and_null_output_schema(graph): "UNWIND [1, 2] AS x CALL { WITH x RETURN x AS n UNION ALL " "RETURN $value AS n } RETURN n ORDER BY n" ) - for value, expected in [(2.5, [1.0, 2.0, 2.5, 2.5]), (None, [None, None, 1, 2])]: + for value, expected in [(2.5, [1.0, 2.0, 2.5, 2.5]), (None, [1, 2, None, None])]: table = engine.execute(query, {"value": value}) expected_type = pa.float64() if value is not None else pa.int64() assert table.schema.field("n").type == expected_type diff --git a/python/python/tests/test_order_nulls.py b/python/python/tests/test_order_nulls.py new file mode 100644 index 0000000..b2bff01 --- /dev/null +++ b/python/python/tests/test_order_nulls.py @@ -0,0 +1,34 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The lance-csr Authors + +"""ORDER BY places nulls as openCypher does: last ascending, first descending.""" + +import pyarrow as pa +import pytest +from lance_csr import CypherEngine, GraphConfig + +CONFIG = ( + GraphConfig.builder() + .node_table("Person", "id") + .rel_table("KNOWS", "Person", "Person", "src", "dst") + .build() +) +TABLES = { + "Person": pa.table({"id": [1, 2, 3, 4], "name": ["B", None, "A", "D"]}), + "KNOWS": pa.table({"src": [1], "dst": [2]}), +} + + +@pytest.mark.parametrize("use_csr", [False, True]) +@pytest.mark.parametrize( + ("order", "expected"), + [ + ("ORDER BY name", ["A", "B", "D", None]), + ("ORDER BY name ASC", ["A", "B", "D", None]), + ("ORDER BY name DESC", [None, "D", "B", "A"]), + ], +) +def test_order_by_null_placement(use_csr, order, expected): + engine = CypherEngine(CONFIG, TABLES, use_csr) + rows = engine.execute(f"MATCH (p:Person) RETURN p.name AS name {order}") + assert [row["name"] for row in rows.to_pylist()] == expected diff --git a/python/python/tests/test_path_values.py b/python/python/tests/test_path_values.py index 41bee22..6a4baf4 100644 --- a/python/python/tests/test_path_values.py +++ b/python/python/tests/test_path_values.py @@ -647,3 +647,43 @@ def test_path_where_on_start_node(use_csr, query, expected): "KNOWS": pa.table({"src": [1, 2], "dst": [2, 3]}), } assert CypherEngine(config, data, use_csr).execute(query).to_pylist() == expected + + +@pytest.mark.parametrize("use_csr", [False, True]) +@pytest.mark.parametrize( + ("query", "expected"), + [ + # Parallel edges 2->3 make two distinct paths. + ( + "MATCH p = (a:Person)-[:KNOWS]->(b:Person) RETURN count(DISTINCT p) AS n", + [{"n": 3}], + ), + ("MATCH p = (a:Person)-[:KNOWS]->(b:Person) RETURN count(p) AS n", [{"n": 3}]), + ( + "UNWIND [1, 2, 2, null] AS x RETURN count(x) AS n, count(DISTINCT x) AS d", + [{"n": 3, "d": 2}], + ), + ( + "MATCH (p:Person) WITH p.name AS name " + "RETURN count(name) AS n, count(DISTINCT name) AS d", + [{"n": 3, "d": 3}], + ), + ( + "MATCH (p:Person) RETURN count(p) AS n, count(DISTINCT p) AS d", + [{"n": 4, "d": 4}], + ), + ], +) +def test_count_of_paths_and_values(use_csr, query, expected): + # count(v) uses v's id column for nodes and relationships, and the value otherwise. + config = ( + GraphConfig.builder() + .node_table("Person", "id") + .rel_table("KNOWS", "Person", "Person", "src", "dst") + .build() + ) + data = { + "Person": pa.table({"id": [1, 2, 3, 4], "name": ["A", "B", "C", None]}), + "KNOWS": pa.table({"src": [1, 2, 2], "dst": [2, 3, 3]}), + } + assert CypherEngine(config, data, use_csr).execute(query).to_pylist() == expected diff --git a/python/python/tests/test_timeout.py b/python/python/tests/test_timeout.py index e931ccf..31633a7 100644 --- a/python/python/tests/test_timeout.py +++ b/python/python/tests/test_timeout.py @@ -72,8 +72,11 @@ def test_no_deadline_and_normal_execution_errors(engine): def test_execution_error_in_a_timed_stream_is_preserved(engine): reader = engine.execute_batches("MATCH (p:Person) RETURN 1 / p.id AS n", timeout=10) + # Only the batch holding p.id = 0 fails, and partitions finish in any order, so + # earlier reads may succeed; the error must still surface before the stream ends. with pytest.raises(pa.ArrowInvalid) as raised: - reader.read_next_batch() + while True: + reader.read_next_batch() assert "Query timeout exceeded" not in str(raised.value) with pytest.raises(StopIteration): reader.read_next_batch() diff --git a/python/uv.lock b/python/uv.lock index 29a88c4..579cb90 100644 --- a/python/uv.lock +++ b/python/uv.lock @@ -40,7 +40,7 @@ wheels = [ [[package]] name = "lance-csr" -version = "0.3.0" +version = "0.3.1" source = { editable = "." } dependencies = [ { name = "pyarrow" },