Skip to content

fix(cypher): group aggregations by evaluated scalar-function keys #1279

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mustafaabidali wants to merge 1 commit into
base: main
Choose a base branch
from
+135 −14
Open

fix(cypher): group aggregations by evaluated scalar-function keys #1279

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 30 additions & 14 deletions src/cypher/cypher.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3716,15 +3716,23 @@ typedef struct {
int64_t *group_node_ids; /* per-item node id when the group var is a node (0 = not) */
} with_agg_t;

/* Build a group key from non-aggregate WITH items */
/* Build a group key from non-aggregate WITH items.
*
* A grouping item is anything that is NOT an aggregate — including computed
* expressions such as labels(n), type(r), toLower(n.prop) and CASE. Testing
* `item->func` alone would misclassify those scalar functions as aggregates
* (they populate `func` too), collapsing every row into a single bogus group.
* Route them through project_item so the *evaluated* value forms the key,
* exactly as the RETURN path does. */
static int with_agg_build_key(cbm_return_clause_t *wc, binding_t *b, char *key, size_t key_sz) {
int kl = 0;
for (int ci = 0; ci < wc->count; ci++) {
if (wc->items[ci].func) {
if (is_aggregate_func(wc->items[ci].func)) {
continue;
}
const char *v = binding_get_virtual(b, wc->items[ci].variable, wc->items[ci].property);
kl += snprintf(key + kl, key_sz - (size_t)kl, "%s|", v);
char vbuf[CBM_SZ_512];
const char *v = project_item(b, &wc->items[ci], vbuf, sizeof(vbuf));
kl += snprintf(key + kl, key_sz - (size_t)kl, "%s|", v ? v : "");
if (kl >= (int)key_sz) {
kl = (int)key_sz - SKIP_ONE;
}
Expand Down Expand Up @@ -3759,16 +3767,18 @@ static int with_agg_find_or_create(with_agg_t **aggs, int *agg_cnt, int *agg_cap
(*aggs)[found].maxs[ci] = -CYP_DBL_MAX;
}
for (int ci = 0; ci < wc->count; ci++) {
if (wc->items[ci].func) {
if (is_aggregate_func(wc->items[ci].func)) {
(*aggs)[found].group_vals[ci] = heap_strdup("0");
continue;
}
const char *v = binding_get_virtual(b, wc->items[ci].variable, wc->items[ci].property);
(*aggs)[found].group_vals[ci] = heap_strdup(v);
char vbuf[CBM_SZ_512];
const char *v = project_item(b, &wc->items[ci], vbuf, sizeof(vbuf));
(*aggs)[found].group_vals[ci] = heap_strdup(v ? v : "");
/* If this group item is a bare node variable, remember its id so the
* carried virtual var can re-fetch any property (group_vals holds only
* the name). */
if (!wc->items[ci].property && wc->items[ci].variable) {
* the name). A computed expression has no node identity to carry. */
if (!wc->items[ci].property && wc->items[ci].variable && !wc->items[ci].func &&
!wc->items[ci].kase && !wc->items[ci].args) {
cbm_node_t *gn = binding_get(b, wc->items[ci].variable);
if (gn) {
(*aggs)[found].group_node_ids[ci] = gn->id;
Expand All @@ -3781,7 +3791,7 @@ static int with_agg_find_or_create(with_agg_t **aggs, int *agg_cnt, int *agg_cap
/* Accumulate aggregation values for a binding */
static void with_agg_accumulate(with_agg_t *agg, cbm_return_clause_t *wc, binding_t *b) {
for (int ci = 0; ci < wc->count; ci++) {
if (!wc->items[ci].func) {
if (!is_aggregate_func(wc->items[ci].func)) {
continue;
}
agg->counts[ci]++;
Expand Down Expand Up @@ -3857,6 +3867,12 @@ static void execute_with_aggregate(cbm_return_clause_t *wc, binding_t *bindings,
int agg_cnt = 0;

for (int bi = 0; bi < bind_count; bi++) {
/* #601: grouping is O(bindings x groups) and each iteration evaluates
* every non-aggregate expression. Abort if we blow the wall-clock
* budget, matching execute_return_agg. */
if ((bi & CYPHER_DEADLINE_CHECK_MASK) == 0 && cypher_deadline_exceeded()) {
break;
}
char key[CBM_SZ_1K] = "";
with_agg_build_key(wc, &bindings[bi], key, sizeof(key));
int found = with_agg_find_or_create(&aggs, &agg_cnt, &agg_cap, wc, &bindings[bi], key);
Expand All @@ -3876,7 +3892,7 @@ static void execute_with_aggregate(cbm_return_clause_t *wc, binding_t *bindings,
for (int ci = 0; ci < wc->count; ci++) {
char name_buf[CBM_SZ_256];
const char *alias = resolve_item_alias(&wc->items[ci], name_buf, sizeof(name_buf));
if (wc->items[ci].func) {
if (is_aggregate_func(wc->items[ci].func)) {
char vbuf[CBM_SZ_64];
if (wc->items[ci].distinct && strcmp(wc->items[ci].func, "COUNT") == 0) {
snprintf(vbuf, sizeof(vbuf), "%d", aggs[a].distinct_n[ci]); /* #239 */
Expand Down Expand Up @@ -4175,7 +4191,7 @@ static void ret_agg_init_group(ret_agg_entry_t *entry, const char *key, int item
/* Accumulate a binding into RETURN aggregation */
static void ret_agg_accumulate(ret_agg_entry_t *entry, cbm_return_clause_t *ret, binding_t *b) {
for (int ci = 0; ci < ret->count; ci++) {
if (!ret->items[ci].func) {
if (!is_aggregate_func(ret->items[ci].func)) {
continue;
}
entry->counts[ci]++;
Expand Down Expand Up @@ -4227,7 +4243,7 @@ static void ret_agg_build_key(cbm_return_clause_t *ret, binding_t *b, char *key,
const char **vals, char valbufs[][CBM_SZ_512]) {
int klen = 0;
for (int ci = 0; ci < ret->count; ci++) {
if (ret->items[ci].func) {
if (is_aggregate_func(ret->items[ci].func)) {
vals[ci] = "0";
continue;
}
Expand All @@ -4251,7 +4267,7 @@ static void ret_agg_emit_row(cbm_return_clause_t *ret, ret_agg_entry_t *agg, res
const char *row[CBM_SZ_32];
char bufs[CBM_SZ_32][CBM_SZ_64];
for (int ci = 0; ci < ret->count; ci++) {
if (!ret->items[ci].func) {
if (!is_aggregate_func(ret->items[ci].func)) {
row[ci] = agg->group_vals[ci];
continue;
}
Expand Down
105 changes: 105 additions & 0 deletions tests/test_cypher.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1346,6 +1346,106 @@ TEST(cypher_exec_unsupported_func_errors_issue373) {
PASS();
}

/* Return the aggregate column of the row whose group column equals group_val,
* so assertions do not depend on group encounter order. */
static const char *agg_lookup(cbm_cypher_result_t *r, const char *group_val) {
for (int i = 0; i < r->row_count; i++) {
if (r->rows[i][0] && strcmp(r->rows[i][0], group_val) == 0) {
return r->rows[i][1];
}
}
return NULL;
}

/* Grouping regression: a non-aggregate computed expression (labels(), type(),
* toLower(), ...) populates cbm_return_item_t.func just like an aggregate does.
* Treating "has func" as "is aggregate" dropped these columns from the GROUP BY
* key and then formatted them through the aggregate formatter, so every row
* collapsed into one group whose group column printed the row count. Fixture:
* 4 Function nodes + 1 Module node, 3 CALLS edges + 1 DEFINES edge. */
TEST(cypher_exec_group_by_labels_func) {
cbm_store_t *s = setup_cypher_store();

cbm_cypher_result_t r = {0};
int rc = cbm_cypher_execute(s, "MATCH (n) RETURN labels(n) AS l, count(n) AS c", "test", 0, &r);
ASSERT_EQ(rc, 0);
ASSERT_EQ(r.row_count, 2);
ASSERT_NOT_NULL(agg_lookup(&r, "[\"Function\"]"));
ASSERT_STR_EQ(agg_lookup(&r, "[\"Function\"]"), "4");
ASSERT_NOT_NULL(agg_lookup(&r, "[\"Module\"]"));
ASSERT_STR_EQ(agg_lookup(&r, "[\"Module\"]"), "1");
cbm_cypher_result_free(&r);

cbm_store_close(s);
PASS();
}

TEST(cypher_exec_group_by_type_func) {
cbm_store_t *s = setup_cypher_store();

cbm_cypher_result_t r = {0};
int rc =
cbm_cypher_execute(s, "MATCH ()-[r]->() RETURN type(r) AS t, count(r) AS c", "test", 0, &r);
ASSERT_EQ(rc, 0);
ASSERT_EQ(r.row_count, 2);
ASSERT_STR_EQ(agg_lookup(&r, "CALLS"), "3");
ASSERT_STR_EQ(agg_lookup(&r, "DEFINES"), "1");
cbm_cypher_result_free(&r);

cbm_store_close(s);
PASS();
}

TEST(cypher_exec_group_by_string_func) {
cbm_store_t *s = setup_cypher_store();

cbm_cypher_result_t r = {0};
int rc = cbm_cypher_execute(s, "MATCH (n) RETURN toLower(n.label) AS l, count(n) AS c", "test",
0, &r);
ASSERT_EQ(rc, 0);
ASSERT_EQ(r.row_count, 2);
ASSERT_STR_EQ(agg_lookup(&r, "function"), "4");
ASSERT_STR_EQ(agg_lookup(&r, "module"), "1");
cbm_cypher_result_free(&r);

cbm_store_close(s);
PASS();
}

/* Same defect existed on the WITH aggregation path, which additionally read the
* group value from the raw binding instead of evaluating the expression. */
TEST(cypher_exec_group_by_func_in_with) {
cbm_store_t *s = setup_cypher_store();

cbm_cypher_result_t r = {0};
int rc = cbm_cypher_execute(s, "MATCH (n) WITH labels(n) AS l, count(n) AS c RETURN l, c",
"test", 0, &r);
ASSERT_EQ(rc, 0);
ASSERT_EQ(r.row_count, 2);
ASSERT_STR_EQ(agg_lookup(&r, "[\"Function\"]"), "4");
ASSERT_STR_EQ(agg_lookup(&r, "[\"Module\"]"), "1");
cbm_cypher_result_free(&r);

cbm_store_close(s);
PASS();
}

/* A plain-property group key must keep working unchanged. */
TEST(cypher_exec_group_by_plain_property_unchanged) {
cbm_store_t *s = setup_cypher_store();

cbm_cypher_result_t r = {0};
int rc = cbm_cypher_execute(s, "MATCH (n) RETURN n.label AS l, count(n) AS c", "test", 0, &r);
ASSERT_EQ(rc, 0);
ASSERT_EQ(r.row_count, 2);
ASSERT_STR_EQ(agg_lookup(&r, "Function"), "4");
ASSERT_STR_EQ(agg_lookup(&r, "Module"), "1");
cbm_cypher_result_free(&r);

cbm_store_close(s);
PASS();
}

/* A recognised function still works, and an unknown one in plain RETURN errors. */
TEST(cypher_exec_unknown_func_return_errors) {
cbm_store_t *s = setup_cypher_store();
Expand Down Expand Up @@ -3240,6 +3340,11 @@ SUITE(cypher) {
RUN_TEST(cypher_exec_where_label_test_issue241);
RUN_TEST(cypher_exec_label_alternation_issue242);
RUN_TEST(cypher_exec_count_distinct_issue239);
RUN_TEST(cypher_exec_group_by_labels_func);
RUN_TEST(cypher_exec_group_by_type_func);
RUN_TEST(cypher_exec_group_by_string_func);
RUN_TEST(cypher_exec_group_by_func_in_with);
RUN_TEST(cypher_exec_group_by_plain_property_unchanged);
RUN_TEST(cypher_exec_unsupported_func_errors_issue373);
RUN_TEST(cypher_exec_unknown_func_return_errors);
RUN_TEST(cypher_exec_inline_props);
Expand Down
Loading