diff --git a/src/cypher/cypher.c b/src/cypher/cypher.c index 3cc31f9d7..db48eaacf 100644 --- a/src/cypher/cypher.c +++ b/src/cypher/cypher.c @@ -12,6 +12,8 @@ #include "foundation/platform.h" #include "foundation/limits.h" #include "foundation/log.h" +#include "foundation/hash_table.h" +#include "foundation/mem_core.h" enum { CYP_BUF_16 = 16, @@ -2837,6 +2839,104 @@ static void rb_add_row(result_builder_t *rb, const char **values) { rb->rows[rb->row_count++] = row; } +/* ── DISTINCT while projecting (#1364) ───────────────────────────── + * + * RETURN DISTINCT used to project up to the engine row cap and only then drop + * duplicates, so a value first seen past the cap never reached the result: on + * a 150k-edge index `RETURN DISTINCT type(r)` lost its one HAS_BRANCH edge. + * The projection loops now de-duplicate as they go, so the cap counts DISTINCT + * rows. Memory stays bounded by the cap: only admitted rows are remembered; + * rows past the cap are probed (no insert) solely to decide `truncated`. */ +typedef struct { + CBMHashTable *seen; /* row key -> the same heap key; NULL = not DISTINCT */ +} rb_distinct_t; + +static void rb_distinct_init(rb_distinct_t *d, bool distinct) { + d->seen = distinct ? cbm_ht_create(CBM_SZ_64) : NULL; +} + +static void rb_distinct_free_key(const char *key, void *value, void *userdata) { + (void)key; + (void)userdata; + cbm_free(CBM_MEM_CLASS_OTHER, value); +} + +static void rb_distinct_free(rb_distinct_t *d) { + if (d->seen) { + cbm_ht_foreach(d->seen, rb_distinct_free_key, NULL); + cbm_ht_free(d->seen); + d->seen = NULL; + } +} + +/* Length-prefixed row key (":" per column), so no value content + * can make two different rows collide. NULL projects as "" (as rb_add_row). */ +static char *rb_distinct_row_key(const char **vals, int n) { + size_t total = SKIP_ONE; + for (int i = 0; i < n; i++) { + total += (vals[i] ? strlen(vals[i]) : 0) + CBM_SZ_32; + } + char *key = cbm_alloc(CBM_MEM_CLASS_OTHER, total); + if (!key) { + return NULL; + } + size_t pos = 0; + key[0] = '\0'; + for (int i = 0; i < n; i++) { + const char *v = vals[i] ? vals[i] : ""; + int w = snprintf(key + pos, total - pos, "%zu:%s", strlen(v), v); + if (w > 0) { + pos += (size_t)w; + } + } + return key; +} + +/* Admission for one projected row. Returns false for a row DISTINCT already + * holds. On true, *key_out is the key to hand to rb_distinct_keep (NULL when + * not DISTINCT, or when the key allocation failed — the row is then admitted + * without de-duplication rather than dropped). */ +static bool rb_distinct_is_new(const rb_distinct_t *d, const char **vals, int n, char **key_out) { + *key_out = NULL; + if (!d->seen) { + return true; + } + char *key = rb_distinct_row_key(vals, n); + if (!key) { + return true; + } + if (cbm_ht_has(d->seen, key)) { + cbm_free(CBM_MEM_CLASS_OTHER, key); + return false; + } + *key_out = key; + return true; +} + +static void rb_distinct_keep(rb_distinct_t *d, char *key) { + if (key) { + cbm_ht_set(d->seen, key, key); + } +} + +/* Add one projected row under a row cap, de-duplicating when DISTINCT. + * Returns false once a NEW row arrives with the cap already full: the caller + * stops there, and flags truncation when the cap is the engine budget. A + * duplicate past the cap is neither kept nor evidence of truncation. */ +static bool rb_add_row_capped(result_builder_t *rb, rb_distinct_t *d, const char **vals, int cap) { + char *key = NULL; + if (!rb_distinct_is_new(d, vals, rb->col_count, &key)) { + return true; + } + if (rb->row_count >= cap) { + cbm_free(CBM_MEM_CLASS_OTHER, key); + return false; + } + rb_distinct_keep(d, key); + rb_add_row(rb, vals); + return true; +} + /* ── Main execution ─────────────────────────────────────────────── */ /* Hard ceiling: queries returning more than this trigger an error instead of data. @@ -4359,14 +4459,20 @@ static void execute_return_star_after_with(cbm_query_t *q, binding_t *bindings, cols[i] = resolve_item_alias(&wc->items[i], name_bufs[i], sizeof(name_bufs[i])); } rb_set_columns(rb, cols, col_n); - for (int bi = 0; bi < bind_count && rb->row_count < max_rows; bi++) { + rb_distinct_t seen; + rb_distinct_init(&seen, q->ret && q->ret->distinct); + for (int bi = 0; bi < bind_count; bi++) { const char *vals[CYP_MAX_VARS]; for (int i = 0; i < col_n; i++) { cbm_node_t *vn = binding_get(&bindings[bi], cols[i]); vals[i] = vn && vn->name ? vn->name : ""; } - rb_add_row(rb, vals); + if (!rb_add_row_capped(rb, &seen, vals, max_rows)) { + g_cypher_truncated = true; + break; + } } + rb_distinct_free(&seen); } static void execute_return_star(cbm_query_t *q, binding_t *bindings, int bind_count, int max_rows, @@ -4381,19 +4487,22 @@ static void execute_return_star(cbm_query_t *q, binding_t *bindings, int bind_co int projection_cap = max_rows; bool cap_is_engine_budget = true; cbm_return_clause_t *ret = q->ret; - if (ret && ret->limit >= 0 && ret->limit <= max_rows && !ret->distinct && - ret->order_key_count == 0 && ret->skip <= 0) { - projection_cap = ret->limit; + if (ret && ret->limit >= 0 && ret->limit <= max_rows && ret->order_key_count == 0 && + ret->skip <= 0) { + projection_cap = ret->limit; /* DISTINCT: the cap counts distinct rows (#1364) */ cap_is_engine_budget = false; } - for (int bi = 0; bi < bind_count && rb->row_count < projection_cap; bi++) { + rb_distinct_t seen; + rb_distinct_init(&seen, ret && ret->distinct); + for (int bi = 0; bi < bind_count; bi++) { const char *vals[CBM_SZ_128]; project_star_row(&bindings[bi], vars, vc, vals); - rb_add_row(rb, vals); - } - if (cap_is_engine_budget && bind_count > projection_cap) { - g_cypher_truncated = true; + if (!rb_add_row_capped(rb, &seen, vals, projection_cap)) { + g_cypher_truncated = g_cypher_truncated || cap_is_engine_budget; + break; + } } + rb_distinct_free(&seen); } /* Format an aggregate value into buf based on function name */ @@ -4662,22 +4771,25 @@ static void execute_return_simple(cbm_return_clause_t *ret, binding_t *bindings, int max_rows, result_builder_t *rb) { int proj_cap = max_rows; bool cap_is_engine_budget = true; - if (ret->limit >= 0 && !ret->distinct && ret->order_key_count == 0 && ret->skip <= 0) { - proj_cap = ret->limit; + if (ret->limit >= 0 && ret->order_key_count == 0 && ret->skip <= 0) { + proj_cap = ret->limit; /* DISTINCT: the cap counts distinct rows (#1364) */ cap_is_engine_budget = false; } - for (int bi = 0; bi < bind_count && rb->row_count < proj_cap; bi++) { + rb_distinct_t seen; + rb_distinct_init(&seen, ret->distinct); + for (int bi = 0; bi < bind_count; bi++) { const char *vals[CBM_SZ_32]; char func_bufs[CBM_SZ_32][CBM_SZ_512]; for (int ci = 0; ci < ret->count; ci++) { vals[ci] = project_item(&bindings[bi], &ret->items[ci], func_bufs[ci], sizeof(func_bufs[ci])); } - rb_add_row(rb, vals); - } - if (cap_is_engine_budget && bind_count > proj_cap) { - g_cypher_truncated = true; + if (!rb_add_row_capped(rb, &seen, vals, proj_cap)) { + g_cypher_truncated = g_cypher_truncated || cap_is_engine_budget; + break; + } } + rb_distinct_free(&seen); } /* Build default 3-column headers (name, qualified_name, label) per variable */ @@ -5019,7 +5131,9 @@ static void execute_return_clause(cbm_query_t *q, cbm_return_clause_t *ret, bind } } - if (ret->distinct) { + /* The simple and star projections de-duplicate while projecting (#1364); + * only aggregate rows still need the post-pass. */ + if (ret->distinct && has_agg && !ret->star) { rb_apply_distinct(rb); } rb_apply_order_by(rb, ret); diff --git a/tests/test_cypher.c b/tests/test_cypher.c index abd334683..f0d9c2724 100644 --- a/tests/test_cypher.c +++ b/tests/test_cypher.c @@ -1269,6 +1269,99 @@ TEST(cypher_exec_aggregate_sees_all_edges_beyond_expansion_cap_issue1196) { PASS(); } +/* #1364: RETURN DISTINCT de-duplicated only AFTER the engine row cap, so a + * value that first appears past the cap was never seen: on a 150k-edge index + * `RETURN DISTINCT type(r)` returned 21 of 22 types (HAS_BRANCH, one edge, + * missing). The cap must count DISTINCT rows. Fixture: 20 CALLS edges, then + * one HAS_BRANCH edge from the last-inserted node; max_rows=5 puts the rare + * type far past the raw-row cap. */ +TEST(cypher_return_distinct_counts_distinct_rows_against_cap_issue1364) { + cbm_store_t *s = cbm_store_open_memory(); + ASSERT_NOT_NULL(s); + ASSERT_EQ(cbm_store_upsert_project(s, "test", "/tmp/test"), CBM_STORE_OK); + + int64_t ids[22]; + for (int i = 0; i < 22; i++) { + char name[32]; + char qn[64]; + snprintf(name, sizeof(name), "fn_%02d", i); + snprintf(qn, sizeof(qn), "test.%s", name); + cbm_node_t n = {.project = "test", + .label = "Function", + .name = name, + .qualified_name = qn, + .file_path = "a.py"}; + ids[i] = cbm_store_upsert_node(s, &n); + ASSERT_GT(ids[i], 0); + } + for (int i = 0; i < 20; i++) { + cbm_edge_t e = { + .project = "test", .source_id = ids[i], .target_id = ids[i + 1], .type = "CALLS"}; + cbm_store_insert_edge(s, &e); + } + cbm_edge_t rare = { + .project = "test", .source_id = ids[21], .target_id = ids[0], .type = "HAS_BRANCH"}; + cbm_store_insert_edge(s, &rare); + + /* Plain DISTINCT: both types, complete (2 distinct rows << cap of 5). */ + cbm_cypher_result_t r = {0}; + int rc = cbm_cypher_execute(s, "MATCH ()-[r]->() RETURN DISTINCT type(r) AS t", "test", 5, &r); + ASSERT_EQ(rc, 0); + ASSERT_EQ(r.row_count, 2); + bool saw_calls = false; + bool saw_branch = false; + for (int i = 0; i < r.row_count; i++) { + saw_calls = saw_calls || strcmp(r.rows[i][0], "CALLS") == 0; + saw_branch = saw_branch || strcmp(r.rows[i][0], "HAS_BRANCH") == 0; + } + ASSERT_TRUE(saw_calls); + ASSERT_TRUE(saw_branch); + ASSERT_FALSE(r.truncated); + cbm_cypher_result_free(&r); + + /* DISTINCT + ORDER BY shares the cap. */ + memset(&r, 0, sizeof(r)); + rc = cbm_cypher_execute(s, "MATCH ()-[r]->() RETURN DISTINCT type(r) AS t ORDER BY t DESC", + "test", 5, &r); + ASSERT_EQ(rc, 0); + ASSERT_EQ(r.row_count, 2); + ASSERT_STR_EQ(r.rows[0][0], "HAS_BRANCH"); + ASSERT_STR_EQ(r.rows[1][0], "CALLS"); + ASSERT_FALSE(r.truncated); + cbm_cypher_result_free(&r); + + /* DISTINCT + ORDER BY + SKIP/LIMIT: SKIP counts distinct rows. */ + memset(&r, 0, sizeof(r)); + rc = cbm_cypher_execute( + s, "MATCH ()-[r]->() RETURN DISTINCT type(r) AS t ORDER BY t SKIP 1 LIMIT 1", "test", 5, + &r); + ASSERT_EQ(rc, 0); + ASSERT_EQ(r.row_count, 1); + ASSERT_STR_EQ(r.rows[0][0], "HAS_BRANCH"); + ASSERT_FALSE(r.truncated); + cbm_cypher_result_free(&r); + + /* An explicit LIMIT on DISTINCT counts distinct rows too. */ + memset(&r, 0, sizeof(r)); + rc = cbm_cypher_execute(s, "MATCH ()-[r]->() RETURN DISTINCT type(r) AS t LIMIT 2", "test", 5, + &r); + ASSERT_EQ(rc, 0); + ASSERT_EQ(r.row_count, 2); + ASSERT_FALSE(r.truncated); + cbm_cypher_result_free(&r); + + /* More distinct rows than the cap: still bounded and reported truncated. */ + memset(&r, 0, sizeof(r)); + rc = cbm_cypher_execute(s, "MATCH (a)-[r]->(b) RETURN DISTINCT b.name", "test", 5, &r); + ASSERT_EQ(rc, 0); + ASSERT_EQ(r.row_count, 5); + ASSERT_TRUE(r.truncated); + cbm_cypher_result_free(&r); + + cbm_store_close(s); + PASS(); +} + /* #874: coalesce(var.prop, literal) in WHERE — null-safe numeric filters * for audit queries over OPTIONAL graph properties. The parser rejected the * call outright ("unexpected operator"); RETURN-side coalesce already @@ -4925,6 +5018,7 @@ SUITE(cypher) { RUN_TEST(cypher_exec_where_eq); RUN_TEST(cypher_exec_unlabeled_where_beyond_result_limit_issue1196); RUN_TEST(cypher_exec_aggregate_sees_all_edges_beyond_expansion_cap_issue1196); + RUN_TEST(cypher_return_distinct_counts_distinct_rows_against_cap_issue1364); RUN_TEST(cypher_exec_varlength_path_semantics_issue797); RUN_TEST(cypher_exec_where_coalesce_issue874); RUN_TEST(cypher_exec_where_regex);