mirror of
https://github.com/DeusData/codebase-memory-mcp.git
synced 2026-11-04 10:48:20 +00:00
perf(cypher): stream query execution chunk by chunk from a paged node scan
On the kernel index (8.5 M nodes, 1.9 M CALLS edges) `MATCH (a)-[:CALLS]->(b)
RETURN count(*)` peaked at 18-25 GB RSS (killed in a 12 GB container): the
store materialised every start node with all its strings (4.9 GB), every one
was deep-copied into a 1.6 KB binding, and every expanded row was materialised
again before projection.
cypher: execute_single now runs the start nodes through the pipeline in chunks
of 1024 (pattern-0 hops, late WHERE, WITH, RETURN). Each hop emits a binding's
expansions in input order and every consumer (RETURN rows / star / default
projection / aggregation, WITH aggregation / projection) keeps its state
between chunks, so the concatenated output is the single-pass output: same
rows, row order, first-seen group order, ORDER BY ties, DISTINCT, LIMIT and
max_rows truncation, OPTIONAL fallback rows, warnings. Queries with additional
MATCH patterns keep one pass (their join strategy depends on the first binding
of the whole set). Start nodes and fetched neighbours move their strings into
the binding instead of being deep-copied and freed.
store: cbm_store_node_scan_open/_next/_close hand out the rows of
cbm_store_find_nodes / _by_label a page at a time. Pages are keyset queries
along the index the eager query walks (idx_nodes_file: NULL file_path rows by
id, then the rest of the last file by id, then the files after it;
idx_nodes_label: by id), each reset before it returns, so no read lock is held
between pages. open verifies the eager plan with EXPLAIN QUERY PLAN and falls
back to the eager rows for any other plan.
Kernel, same index, medians of 3 (macOS arm64):
count(*) peak RSS 24.9 GB -> 256 MB, 23.0 s -> 13.0 s
top-callee peak RSS 12.4 GB -> 1.26 GB, 11.7 s -> 10.9 s
Responses byte-identical to the baseline on django, redis and the kernel
(19 Cypher + 24 tool cases each); the streamed scan equals the eager scan
field by field, in order, for every project and label of all three indexes.
A scan that fails part-way fails the whole query ("query aborted: reading the graph failed") instead of answering from the pages read so far:
The streamed start-node scan answered from the rows of the pages it had read
when a later page failed (a SQLite step error or a page re-prepare failure):
a partial result that looked complete. A failed page, a scan that cannot
open, or a failed eager fallback now marks the start source failed;
execute_chunk returns an error and cbm_cypher_execute reports
"query aborted: reading the graph failed (<store error>)" with no rows. The
store error is logged as cypher.scan_failed.
The test seam cbm_store_node_scan_test_fail_page_after (CBM_ENABLE_TEST_SEAMS)
fails one page after a given number of successful pages; the new test fails
the second page of a label scan (rows and count(*)) and a later page of the
all-nodes scan, and asserts an error result with no rows.
Signed-off-by: Martin Vogel <martin.vogel.tech@gmail.com>
This commit is contained in:
Martin Vogel committed
1 parent
a372a1fa5e
commit
181837343e
7 files changed
+1308
-244
No files matched your search
@@ -24,7 +24,7 @@ src/cli/config_text_edit.c 72
|
||||
src/cli/config_toml_edit.c 73
|
||||
src/cli/config_yaml_edit.c 117
|
||||
src/cli/hook_augment.c 31
|
||||
src/cypher/cypher.c 128
|
||||
src/cypher/cypher.c 125
|
||||
src/daemon/application.c 107
|
||||
src/daemon/bootstrap.c 13
|
||||
src/daemon/daemon.c 31
|
||||
|
||||
+623
-243
File diff suppressed because it is too large.
Load diff
@@ -359,6 +359,9 @@ void cbm_cypher_test_set_deadline_ms(int64_t budget_ms);
|
||||
/* Override only the deadline clock on this thread; NULL restores real time. */
|
||||
typedef uint64_t (*cbm_cypher_test_clock_fn)(void);
|
||||
void cbm_cypher_test_set_deadline_clock(cbm_cypher_test_clock_fn clock_fn);
|
||||
/* Start nodes per streamed execution chunk on this thread; 0 restores the
|
||||
* default. INT_MAX runs the whole start set as one chunk (a single pass). */
|
||||
void cbm_cypher_test_set_stream_chunk(int start_nodes);
|
||||
#endif
|
||||
|
||||
/* Worst-case binding slot count for a node cross-join. Computes the count in
|
||||
|
||||
@@ -3348,6 +3348,280 @@ int cbm_store_find_nodes_by_file(cbm_store_t *s, const char *project, const char
|
||||
project, file_path, out, count);
|
||||
}
|
||||
|
||||
/* ── Streamed node scan ──────────────────────────────────────────────
|
||||
* cbm_store_find_nodes / cbm_store_find_nodes_by_label materialise every row
|
||||
* at once: 8.5 M kernel nodes with all their strings are 4.9 GB before the
|
||||
* caller sees the first one. The scan below hands the same rows out a page at
|
||||
* a time, and resets each page's statement before returning it, so no read
|
||||
* lock outlives a page: a writer waits behind one page, as behind one eager
|
||||
* lookup, never behind the caller's whole query.
|
||||
*
|
||||
* Pages continue by keyset along the index the eager query walks, so they
|
||||
* concatenate to the eager order:
|
||||
* all nodes idx_nodes_file (project, file_path): the NULL file_path rows by
|
||||
* id (index order puts NULL first), then by (file_path, id). A
|
||||
* page continues with the rest of its last row's file (by id),
|
||||
* then with the files after it: two exact index seeks, where a
|
||||
* (file_path, id) row-value bound seeks the file only and would
|
||||
* rescan a large file from its start on every page;
|
||||
* one label idx_nodes_label (project, label): by id.
|
||||
* open checks that plan with EXPLAIN QUERY PLAN on the eager query itself.
|
||||
* Any other plan (a changed schema, planner statistics) falls back to the
|
||||
* eager scan handed out in batches: the same rows, without the saving. */
|
||||
#define NODE_SCAN_COLS \
|
||||
"SELECT id, project, label, name, qualified_name, file_path, start_line, end_line, " \
|
||||
"properties FROM nodes "
|
||||
|
||||
typedef enum {
|
||||
NODE_SCAN_LABEL, /* label = ?2 AND id >= ?3 */
|
||||
NODE_SCAN_NULL_FILE, /* file_path IS NULL AND id >= ?2 */
|
||||
NODE_SCAN_FILE_FIRST, /* file_path IS NOT NULL: the first file page */
|
||||
NODE_SCAN_FILE_SAME, /* file_path = ?2 AND id > ?3: the rest of the last file */
|
||||
NODE_SCAN_FILE_AFTER, /* file_path > ?2: the files after it */
|
||||
NODE_SCAN_STMTS,
|
||||
} node_scan_stmt_t;
|
||||
|
||||
enum { NODE_SCAN_COL_FILE = 5 }; /* file_path in NODE_SCAN_COLS */
|
||||
|
||||
static const char *const node_scan_sql[NODE_SCAN_STMTS] = {
|
||||
NODE_SCAN_COLS "WHERE project = ?1 AND label = ?2 AND id >= ?3 ORDER BY id LIMIT ?4;",
|
||||
NODE_SCAN_COLS "WHERE project = ?1 AND file_path IS NULL AND id >= ?2 ORDER BY id LIMIT ?3;",
|
||||
NODE_SCAN_COLS "WHERE project = ?1 AND file_path IS NOT NULL ORDER BY file_path, id LIMIT ?2;",
|
||||
NODE_SCAN_COLS "WHERE project = ?1 AND file_path = ?2 AND id > ?3 ORDER BY id LIMIT ?4;",
|
||||
NODE_SCAN_COLS "WHERE project = ?1 AND file_path > ?2 ORDER BY file_path, id LIMIT ?3;",
|
||||
};
|
||||
|
||||
struct cbm_store_node_scan {
|
||||
cbm_store_t *store;
|
||||
char *project;
|
||||
char *label;
|
||||
bool paged;
|
||||
node_scan_stmt_t phase; /* LABEL, NULL_FILE, FILE_SAME or FILE_AFTER */
|
||||
bool keyed; /* last_id (and for files last_file) hold the previous page's end */
|
||||
int64_t last_id;
|
||||
sqlite3_value *last_file;
|
||||
sqlite3_stmt *stmts[NODE_SCAN_STMTS];
|
||||
bool done;
|
||||
bool eager_failed; /* fallback: the eager scan itself failed; next reports it once */
|
||||
cbm_node_t *eager; /* fallback: the eager scan's rows, handed out in order */
|
||||
int eager_count;
|
||||
int eager_next;
|
||||
};
|
||||
|
||||
#ifdef CBM_ENABLE_TEST_SEAMS
|
||||
static atomic_int node_scan_test_fail_countdown = ATOMIC_VAR_INIT(CBM_NOT_FOUND);
|
||||
|
||||
void cbm_store_node_scan_test_fail_page_after(int successful_pages) {
|
||||
atomic_store(&node_scan_test_fail_countdown, successful_pages);
|
||||
}
|
||||
#endif
|
||||
|
||||
/* True when SQLite runs `sql` as the single index search `plan`. */
|
||||
static bool node_scan_plan_is(cbm_store_t *s, const char *sql, const char *plan) {
|
||||
char eqp[CBM_SZ_512];
|
||||
snprintf(eqp, sizeof(eqp), "EXPLAIN QUERY PLAN %s", sql);
|
||||
sqlite3_stmt *stmt = NULL;
|
||||
if (sqlite3_prepare_v2(s->db, eqp, CBM_NOT_FOUND, &stmt, NULL) != SQLITE_OK) {
|
||||
sqlite3_finalize(stmt);
|
||||
return false;
|
||||
}
|
||||
int rows = 0;
|
||||
bool match = false;
|
||||
while (sqlite3_step(stmt) == SQLITE_ROW) {
|
||||
const char *detail = (const char *)sqlite3_column_text(stmt, ST_COL_3);
|
||||
match = rows == 0 && detail && strcmp(detail, plan) == 0;
|
||||
rows++;
|
||||
}
|
||||
sqlite3_finalize(stmt);
|
||||
return match && rows == SKIP_ONE;
|
||||
}
|
||||
|
||||
cbm_store_node_scan_t *cbm_store_node_scan_open(cbm_store_t *s, const char *project,
|
||||
const char *label) {
|
||||
if (!s || !s->db) {
|
||||
return NULL;
|
||||
}
|
||||
cbm_store_node_scan_t *scan = cbm_calloc(CBM_MEM_CLASS_STORE, sizeof(*scan));
|
||||
if (!scan) {
|
||||
return NULL;
|
||||
}
|
||||
scan->store = s;
|
||||
scan->project = cbm_mem_strdup(CBM_MEM_CLASS_STORE, project);
|
||||
scan->label = cbm_mem_strdup(CBM_MEM_CLASS_STORE, label);
|
||||
if ((project && !scan->project) || (label && !scan->label)) {
|
||||
cbm_store_node_scan_close(scan);
|
||||
return NULL;
|
||||
}
|
||||
scan->phase = label ? NODE_SCAN_LABEL : NODE_SCAN_NULL_FILE; /* NULL file_path rows first */
|
||||
if (label) {
|
||||
scan->paged =
|
||||
node_scan_plan_is(s, NODE_SCAN_COLS "WHERE project = ?1 AND label = ?2;",
|
||||
"SEARCH nodes USING INDEX idx_nodes_label (project=? AND label=?)");
|
||||
} else {
|
||||
scan->paged = node_scan_plan_is(s, NODE_SCAN_COLS "WHERE project = ?1;",
|
||||
"SEARCH nodes USING INDEX idx_nodes_file (project=?)");
|
||||
}
|
||||
if (!scan->paged) {
|
||||
int rc = label ? cbm_store_find_nodes_by_label(s, project, label, &scan->eager,
|
||||
&scan->eager_count)
|
||||
: cbm_store_find_nodes(s, project, &scan->eager, &scan->eager_count);
|
||||
scan->eager_failed = rc != CBM_STORE_OK;
|
||||
}
|
||||
return scan;
|
||||
}
|
||||
|
||||
bool cbm_store_node_scan_paged(const cbm_store_node_scan_t *scan) {
|
||||
return scan && scan->paged;
|
||||
}
|
||||
|
||||
/* The current phase has no rows left. The NULL file_path rows hand over to
|
||||
* the files, the rest of a file to the files after it; a label's rows and
|
||||
* the files after the last one end the scan. */
|
||||
static void node_scan_phase_done(cbm_store_node_scan_t *scan) {
|
||||
if (scan->phase == NODE_SCAN_NULL_FILE) {
|
||||
scan->keyed = false;
|
||||
scan->phase = NODE_SCAN_FILE_AFTER;
|
||||
} else if (scan->phase == NODE_SCAN_FILE_SAME) {
|
||||
scan->phase = NODE_SCAN_FILE_AFTER; /* after last_file */
|
||||
} else {
|
||||
scan->done = true;
|
||||
}
|
||||
}
|
||||
|
||||
static void node_scan_bind(cbm_store_node_scan_t *scan, sqlite3_stmt *stmt, node_scan_stmt_t which,
|
||||
int max) {
|
||||
int col = SKIP_ONE;
|
||||
bind_text(stmt, col++, scan->project);
|
||||
if (which == NODE_SCAN_LABEL) {
|
||||
bind_text(stmt, col++, scan->label);
|
||||
}
|
||||
if (which == NODE_SCAN_FILE_SAME || which == NODE_SCAN_FILE_AFTER) {
|
||||
sqlite3_bind_value(stmt, col++, scan->last_file);
|
||||
}
|
||||
if (which == NODE_SCAN_LABEL || which == NODE_SCAN_NULL_FILE) {
|
||||
sqlite3_bind_int64(stmt, col++, scan->keyed ? scan->last_id + SKIP_ONE : INT64_MIN);
|
||||
} else if (which == NODE_SCAN_FILE_SAME) {
|
||||
sqlite3_bind_int64(stmt, col++, scan->last_id);
|
||||
}
|
||||
sqlite3_bind_int(stmt, col, max);
|
||||
}
|
||||
|
||||
/* One page of the current phase into out[0..max). A short page ends the
|
||||
* phase; the statement is reset before returning either way. */
|
||||
static int node_scan_page(cbm_store_node_scan_t *scan, cbm_node_t *out, int max, int *got) {
|
||||
*got = 0;
|
||||
cbm_store_t *s = scan->store;
|
||||
node_scan_stmt_t which = scan->phase;
|
||||
if (which == NODE_SCAN_FILE_AFTER && !scan->keyed) {
|
||||
which = NODE_SCAN_FILE_FIRST;
|
||||
}
|
||||
bool after_id =
|
||||
which == NODE_SCAN_LABEL || which == NODE_SCAN_NULL_FILE || which == NODE_SCAN_FILE_SAME;
|
||||
if (after_id && scan->keyed && scan->last_id == INT64_MAX) {
|
||||
node_scan_phase_done(scan); /* nothing can follow the largest id */
|
||||
return CBM_STORE_OK;
|
||||
}
|
||||
#ifdef CBM_ENABLE_TEST_SEAMS
|
||||
if (graph_compare_test_countdown_fires(&node_scan_test_fail_countdown)) {
|
||||
snprintf(s->errbuf, sizeof(s->errbuf), "node scan aborted: injected page failure");
|
||||
return CBM_STORE_ERR;
|
||||
}
|
||||
#endif
|
||||
read_batch_renew(s);
|
||||
sqlite3_stmt **slot = &scan->stmts[which];
|
||||
if (!*slot &&
|
||||
sqlite3_prepare_v2(s->db, node_scan_sql[which], CBM_NOT_FOUND, slot, NULL) != SQLITE_OK) {
|
||||
store_set_error_sqlite(s, "node scan prepare");
|
||||
return CBM_STORE_ERR;
|
||||
}
|
||||
sqlite3_stmt *stmt = *slot;
|
||||
node_scan_bind(scan, stmt, which, max);
|
||||
|
||||
int n = 0;
|
||||
int rc = SQLITE_ROW;
|
||||
while (n < max && (rc = sqlite3_step(stmt)) == SQLITE_ROW) {
|
||||
scan_node(stmt, &out[n]);
|
||||
n++;
|
||||
}
|
||||
bool failed = n < max && rc != SQLITE_DONE;
|
||||
if (n == max) { /* the statement still sits on the page's last row */
|
||||
scan->keyed = true;
|
||||
scan->last_id = out[n - SKIP_ONE].id;
|
||||
if (which == NODE_SCAN_FILE_FIRST || which == NODE_SCAN_FILE_AFTER) {
|
||||
/* a new last file: continue with the rest of it */
|
||||
sqlite3_value_free(scan->last_file);
|
||||
scan->last_file = sqlite3_value_dup(sqlite3_column_value(stmt, NODE_SCAN_COL_FILE));
|
||||
failed = !scan->last_file;
|
||||
scan->phase = NODE_SCAN_FILE_SAME;
|
||||
}
|
||||
}
|
||||
sqlite3_reset(stmt);
|
||||
if (failed) {
|
||||
store_set_error_sqlite(s, "node scan aborted");
|
||||
for (int i = 0; i < n; i++) {
|
||||
cbm_node_free_fields(&out[i]);
|
||||
}
|
||||
return CBM_STORE_ERR;
|
||||
}
|
||||
if (n < max) {
|
||||
node_scan_phase_done(scan);
|
||||
}
|
||||
*got = n;
|
||||
return CBM_STORE_OK;
|
||||
}
|
||||
|
||||
int cbm_store_node_scan_next(cbm_store_node_scan_t *scan, cbm_node_t *out, int max, int *count) {
|
||||
*count = 0;
|
||||
if (!scan || max <= 0) {
|
||||
return CBM_STORE_OK;
|
||||
}
|
||||
if (scan->eager_failed) {
|
||||
scan->eager_failed = false; /* reported once; the scan then reads as exhausted */
|
||||
return CBM_STORE_ERR;
|
||||
}
|
||||
if (!scan->paged) {
|
||||
int left = scan->eager_count - scan->eager_next;
|
||||
int n = left < max ? left : max;
|
||||
if (n > 0) {
|
||||
memcpy(out, scan->eager + scan->eager_next, (size_t)n * sizeof(cbm_node_t));
|
||||
memset(scan->eager + scan->eager_next, 0, (size_t)n * sizeof(cbm_node_t));
|
||||
scan->eager_next += n;
|
||||
}
|
||||
*count = n;
|
||||
return CBM_STORE_OK;
|
||||
}
|
||||
int n = 0;
|
||||
while (n < max && !scan->done) {
|
||||
int got = 0;
|
||||
if (node_scan_page(scan, out + n, max - n, &got) != CBM_STORE_OK) {
|
||||
/* A failed page yields no rows from this call and ends the scan;
|
||||
* cbm_store_error() names the failure. */
|
||||
for (int i = 0; i < n; i++) {
|
||||
cbm_node_free_fields(&out[i]);
|
||||
}
|
||||
scan->done = true;
|
||||
return CBM_STORE_ERR;
|
||||
}
|
||||
n += got;
|
||||
}
|
||||
*count = n;
|
||||
return CBM_STORE_OK;
|
||||
}
|
||||
|
||||
void cbm_store_node_scan_close(cbm_store_node_scan_t *scan) {
|
||||
if (!scan) {
|
||||
return;
|
||||
}
|
||||
for (int i = 0; i < NODE_SCAN_STMTS; i++) {
|
||||
sqlite3_finalize(scan->stmts[i]);
|
||||
}
|
||||
sqlite3_value_free(scan->last_file);
|
||||
cbm_store_free_nodes(scan->eager, scan->eager_count); /* rows not handed out */
|
||||
cbm_free(CBM_MEM_CLASS_STORE, scan->project);
|
||||
cbm_free(CBM_MEM_CLASS_STORE, scan->label);
|
||||
cbm_free(CBM_MEM_CLASS_STORE, scan);
|
||||
}
|
||||
|
||||
typedef struct {
|
||||
cbm_store_cancel_fn cancel;
|
||||
void *context;
|
||||
|
||||
@@ -625,6 +625,29 @@ int cbm_store_find_nodes_by_label(cbm_store_t *s, const char *project, const cha
|
||||
int cbm_store_find_nodes_by_file(cbm_store_t *s, const char *project, const char *file_path,
|
||||
cbm_node_t **out, int *count);
|
||||
|
||||
/* Streamed node scan: the rows of cbm_store_find_nodes (label NULL) or
|
||||
* cbm_store_find_nodes_by_label, in the same order, a batch at a time. Pages
|
||||
* are keyset queries along the eager query's index, and no read lock is held
|
||||
* between calls. open returns NULL on error. next fills out[0..max) and sets
|
||||
* *count; a count of 0 means the scan is exhausted. The caller frees each
|
||||
* returned node with cbm_node_free_fields. A failed read returns
|
||||
* CBM_STORE_ERR with no rows from that call and ends the scan;
|
||||
* cbm_store_error() names the failure. paged is false when the eager query's
|
||||
* plan is not the one the pages reproduce: the scan then hands out the eager
|
||||
* scan's rows (same order, no memory saving). Close the scan before closing
|
||||
* the store. */
|
||||
typedef struct cbm_store_node_scan cbm_store_node_scan_t;
|
||||
cbm_store_node_scan_t *cbm_store_node_scan_open(cbm_store_t *s, const char *project,
|
||||
const char *label);
|
||||
int cbm_store_node_scan_next(cbm_store_node_scan_t *scan, cbm_node_t *out, int max, int *count);
|
||||
bool cbm_store_node_scan_paged(const cbm_store_node_scan_t *scan);
|
||||
void cbm_store_node_scan_close(cbm_store_node_scan_t *scan);
|
||||
#ifdef CBM_ENABLE_TEST_SEAMS
|
||||
/* One-shot fault: after this many successful pages (of any scan), the next
|
||||
* page fails like a SQLite step error; -1 disables the seam. */
|
||||
void cbm_store_node_scan_test_fail_page_after(int successful_pages);
|
||||
#endif
|
||||
|
||||
/* Return a stable, paginated outline for one exact repository-relative file.
|
||||
* File/folder/container nodes are excluded. labels may be NULL when
|
||||
* label_count is zero; otherwise labels are exact-match filters. The query is
|
||||
|
||||
@@ -5831,6 +5831,267 @@ TEST(cypher_grouping_keeps_first_seen_groups) {
|
||||
PASS();
|
||||
}
|
||||
|
||||
/* ── Streamed execution: chunks of start nodes give the single pass ──
|
||||
*
|
||||
* execute_single runs the start nodes through the pipeline in chunks rather
|
||||
* than materialising every binding first (kernel count(*): 18 GB RSS). The
|
||||
* chunked run must be indistinguishable from one pass over the whole start
|
||||
* set: same rows in the same order, first-seen group order, ORDER BY ties,
|
||||
* DISTINCT, LIMIT / max_rows truncation and OPTIONAL fallback rows. Each query
|
||||
* runs as one chunk (the reference) and in chunks of 1, 7, 64 and the
|
||||
* production default, so chunk boundaries fall inside groups and inside
|
||||
* every start node's expansion order. */
|
||||
|
||||
static int stream_run(cbm_store_t *s, const char *query, int max_rows, int chunk,
|
||||
cbm_cypher_result_t *out) {
|
||||
memset(out, 0, sizeof(*out));
|
||||
cbm_cypher_test_set_stream_chunk(chunk);
|
||||
int rc = cbm_cypher_execute(s, query, "test", max_rows, out);
|
||||
cbm_cypher_test_set_stream_chunk(0);
|
||||
return rc;
|
||||
}
|
||||
|
||||
static bool stream_same_str(const char *a, const char *b) {
|
||||
return (!a && !b) || (a && b && strcmp(a, b) == 0);
|
||||
}
|
||||
|
||||
static bool stream_results_equal(const cbm_cypher_result_t *a, const cbm_cypher_result_t *b) {
|
||||
if (a->col_count != b->col_count || a->row_count != b->row_count ||
|
||||
a->truncated != b->truncated || !stream_same_str(a->warning, b->warning) ||
|
||||
!stream_same_str(a->error, b->error)) {
|
||||
return false;
|
||||
}
|
||||
for (int c = 0; c < a->col_count; c++) {
|
||||
if (strcmp(a->columns[c], b->columns[c]) != 0) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
for (int r = 0; r < a->row_count; r++) {
|
||||
for (int c = 0; c < a->col_count; c++) {
|
||||
if (strcmp(a->rows[r][c], b->rows[r][c]) != 0) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/* True when every chunk size reproduces the single-pass result, which is
|
||||
* left in *ref for the caller's own checks (caller frees). */
|
||||
static bool stream_matches_single_pass(cbm_store_t *s, const char *query, int max_rows,
|
||||
cbm_cypher_result_t *ref) {
|
||||
static const int chunks[] = {1, 7, 64, 0 /* production default */};
|
||||
int ref_rc = stream_run(s, query, max_rows, INT_MAX, ref);
|
||||
bool same = ref_rc == 0;
|
||||
for (size_t i = 0; same && i < sizeof(chunks) / sizeof(chunks[0]); i++) {
|
||||
cbm_cypher_result_t got;
|
||||
int rc = stream_run(s, query, max_rows, chunks[i], &got);
|
||||
same = rc == ref_rc && stream_results_equal(ref, &got);
|
||||
if (!same) {
|
||||
printf(" chunk %d differs from the single pass: %s\n", chunks[i], query);
|
||||
}
|
||||
cbm_cypher_result_free(&got);
|
||||
}
|
||||
return same;
|
||||
}
|
||||
|
||||
/* Every pipeline consumer across chunk boundaries: unsorted rows, global and
|
||||
* grouped aggregates (groups fed from many chunks), DISTINCT, late WHERE,
|
||||
* collect, WITH aggregation and WITH sorting, RETURN * with and without a
|
||||
* WITH, the default projection, edge variables and variable-length hops. */
|
||||
TEST(cypher_stream_chunks_match_single_pass) {
|
||||
static const char *const queries[] = {
|
||||
"MATCH (a)-[:CALLS]->(b) RETURN a.name, b.qualified_name",
|
||||
"MATCH (a)-[:CALLS]->(b) RETURN count(*)",
|
||||
"MATCH (a)-[:CALLS]->(b:Function) RETURN b.qualified_name, count(a) AS c ORDER BY c DESC",
|
||||
"MATCH (a:Function)-[:CALLS]->(b) RETURN b.name, count(DISTINCT a.name) AS c",
|
||||
"MATCH (a)-[:CALLS]->(b) RETURN DISTINCT b.name",
|
||||
"MATCH (a)-[:CALLS]->(b) WHERE b.start_line > 20 RETURN a.name, b.name, b.start_line",
|
||||
"MATCH (a)-[:CALLS]->(b) RETURN b.name, collect(a.name) AS cs",
|
||||
("MATCH (a:Function)-[:CALLS]->(b) WITH b, count(a) AS c WHERE c > 1 "
|
||||
"RETURN b.name, c ORDER BY c DESC LIMIT 20"),
|
||||
"MATCH (a)-[:CALLS]->(b) WITH a.name AS n, b.start_line AS l ORDER BY l DESC RETURN n, l",
|
||||
"MATCH (a)-[:CALLS]->(b) WITH DISTINCT b.name AS n RETURN n",
|
||||
"MATCH (a)-[:CALLS]->(b) RETURN *",
|
||||
"MATCH (a)-[:CALLS]->(b) WITH a.name AS n RETURN *",
|
||||
"MATCH (a)-[:CALLS]->(b)",
|
||||
"MATCH (a)-[r:CALLS]->(b) RETURN type(r), a.name ORDER BY a.name",
|
||||
"MATCH (a)-[:CALLS*1..2]->(b) RETURN a.name, b.name",
|
||||
};
|
||||
cbm_store_t *s = setup_reference_store();
|
||||
for (size_t i = 0; i < sizeof(queries) / sizeof(queries[0]); i++) {
|
||||
cbm_cypher_result_t ref;
|
||||
ASSERT_TRUE(stream_matches_single_pass(s, queries[i], 0, &ref));
|
||||
ASSERT_GT(ref.row_count, 0);
|
||||
ASSERT_FALSE(ref.truncated);
|
||||
cbm_cypher_result_free(&ref);
|
||||
}
|
||||
cbm_store_close(s);
|
||||
PASS();
|
||||
}
|
||||
|
||||
/* OPTIONAL fallback rows (a start node with no match keeps its row, target
|
||||
* unbound) sit between the expanded rows of their neighbours in every chunk
|
||||
* layout; the multi-pattern form keeps its single pass. */
|
||||
TEST(cypher_stream_optional_across_chunks) {
|
||||
cbm_store_t *s = setup_reference_store();
|
||||
cbm_cypher_result_t ref;
|
||||
ASSERT_TRUE(stream_matches_single_pass(
|
||||
s, "OPTIONAL MATCH (a)<-[:CALLS]-(b) RETURN a.name, b.name", 0, &ref));
|
||||
int fallback = 0;
|
||||
for (int r = 0; r < ref.row_count; r++) {
|
||||
fallback += ref.rows[r][1][0] == '\0' ? 1 : 0;
|
||||
}
|
||||
ASSERT_GT(fallback, 0); /* uncalled nodes keep a row */
|
||||
ASSERT_GT(ref.row_count, fallback); /* and called nodes expand */
|
||||
cbm_cypher_result_free(&ref);
|
||||
|
||||
ASSERT_TRUE(stream_matches_single_pass(
|
||||
s, "OPTIONAL MATCH (a)<-[:CALLS]-(b) RETURN a.qualified_name, count(b) AS c", 0, &ref));
|
||||
ASSERT_EQ(ref.row_count, REF_NODES);
|
||||
cbm_cypher_result_free(&ref);
|
||||
|
||||
ASSERT_TRUE(stream_matches_single_pass(
|
||||
s, "MATCH (a:Function) OPTIONAL MATCH (a)<-[:CALLS]-(b) RETURN a.name, count(b)", 0, &ref));
|
||||
ASSERT_GT(ref.row_count, 0);
|
||||
cbm_cypher_result_free(&ref);
|
||||
cbm_store_close(s);
|
||||
PASS();
|
||||
}
|
||||
|
||||
/* A LIMIT or the max_rows budget fills up in one chunk while later chunks
|
||||
* still arrive: they must add nothing, and truncation is reported exactly
|
||||
* when the single pass reports it (engine budget yes, explicit LIMIT no). */
|
||||
TEST(cypher_stream_limit_and_max_rows_across_chunks) {
|
||||
cbm_store_t *s = setup_reference_store();
|
||||
cbm_cypher_result_t ref;
|
||||
ASSERT_TRUE(stream_matches_single_pass(
|
||||
s, "MATCH (a)-[:CALLS]->(b) RETURN a.name, b.name LIMIT 25", 0, &ref));
|
||||
ASSERT_EQ(ref.row_count, 25);
|
||||
ASSERT_FALSE(ref.truncated);
|
||||
cbm_cypher_result_free(&ref);
|
||||
|
||||
ASSERT_TRUE(stream_matches_single_pass(
|
||||
s, "MATCH (a)-[:CALLS]->(b) RETURN DISTINCT b.name LIMIT 30", 0, &ref));
|
||||
ASSERT_EQ(ref.row_count, 30);
|
||||
cbm_cypher_result_free(&ref);
|
||||
|
||||
ASSERT_TRUE(stream_matches_single_pass(
|
||||
s, "MATCH (a)-[:CALLS]->(b) RETURN a.name, b.name SKIP 5 LIMIT 10", 0, &ref));
|
||||
ASSERT_EQ(ref.row_count, 10);
|
||||
cbm_cypher_result_free(&ref);
|
||||
|
||||
static const char *const budgeted[] = {
|
||||
"MATCH (a)-[:CALLS]->(b) RETURN a.name, b.name",
|
||||
"MATCH (a)-[:CALLS]->(b) RETURN a.name, b.name ORDER BY b.name",
|
||||
"MATCH (a)-[:CALLS]->(b) RETURN *",
|
||||
"MATCH (a)-[:CALLS]->(b)",
|
||||
};
|
||||
for (size_t i = 0; i < sizeof(budgeted) / sizeof(budgeted[0]); i++) {
|
||||
ASSERT_TRUE(stream_matches_single_pass(s, budgeted[i], 30, &ref));
|
||||
ASSERT_EQ(ref.row_count, 30);
|
||||
ASSERT_TRUE(ref.truncated);
|
||||
cbm_cypher_result_free(&ref);
|
||||
}
|
||||
cbm_store_close(s);
|
||||
PASS();
|
||||
}
|
||||
|
||||
/* A start-node scan that fails part-way fails the whole query: the rows read
|
||||
* before the failure must not answer it as if they were all of them. Pages
|
||||
* hold 7 rows at chunk 7; the store seam fails a page after earlier pages
|
||||
* delivered rows (the all-nodes scan's first page is its empty NULL
|
||||
* file_path pass, hence one more page before the fault). */
|
||||
TEST(cypher_stream_scan_failure_fails_query) {
|
||||
static const struct {
|
||||
const char *query;
|
||||
int pages_before_fault;
|
||||
} cases[] = {
|
||||
{"MATCH (a:Function)-[:CALLS]->(b) RETURN a.name, b.name", 1},
|
||||
{"MATCH (a:Function)-[:CALLS]->(b) RETURN count(*)", 1},
|
||||
{"MATCH (n) RETURN n.name", 2},
|
||||
};
|
||||
cbm_store_t *s = setup_reference_store();
|
||||
cbm_cypher_result_t r;
|
||||
for (size_t i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) {
|
||||
ASSERT_EQ(stream_run(s, cases[i].query, 0, 7, &r), 0);
|
||||
ASSERT_GT(r.row_count, 0);
|
||||
cbm_cypher_result_free(&r);
|
||||
|
||||
cbm_store_node_scan_test_fail_page_after(cases[i].pages_before_fault);
|
||||
int rc = stream_run(s, cases[i].query, 0, 7, &r);
|
||||
cbm_store_node_scan_test_fail_page_after(-1);
|
||||
ASSERT_TRUE(rc != 0);
|
||||
ASSERT_NOT_NULL(r.error);
|
||||
ASSERT_TRUE(strstr(r.error, "reading the graph failed") != NULL);
|
||||
ASSERT_TRUE(strstr(r.error, "injected page failure") != NULL);
|
||||
ASSERT_EQ(r.row_count, 0);
|
||||
cbm_cypher_result_free(&r);
|
||||
}
|
||||
/* The failure does not outlive its query. */
|
||||
ASSERT_EQ(stream_run(s, cases[0].query, 0, 7, &r), 0);
|
||||
ASSERT_GT(r.row_count, 0);
|
||||
ASSERT_NULL(r.error);
|
||||
cbm_cypher_result_free(&r);
|
||||
cbm_store_close(s);
|
||||
PASS();
|
||||
}
|
||||
|
||||
/* The production chunk boundary itself: a start set of 2.5 chunks, a ring of
|
||||
* calls plus five hubs whose caller groups straddle both boundaries. */
|
||||
TEST(cypher_stream_default_chunk_boundary) {
|
||||
enum { STREAM_NODES = 2600, STREAM_HUBS = 5 };
|
||||
cbm_store_t *s = cbm_store_open_memory();
|
||||
cbm_store_upsert_project(s, "test", "/tmp/test");
|
||||
static int64_t ids[STREAM_NODES];
|
||||
for (int i = 0; i < STREAM_NODES; i++) {
|
||||
char name[32];
|
||||
char qn[32];
|
||||
snprintf(name, sizeof(name), "fn_%d", i % 211);
|
||||
snprintf(qn, sizeof(qn), "test.s%d", i);
|
||||
cbm_node_t n = {.project = "test",
|
||||
.label = "Function",
|
||||
.name = name,
|
||||
.qualified_name = qn,
|
||||
.file_path = "ring.c",
|
||||
.start_line = i % 37};
|
||||
ids[i] = cbm_store_upsert_node(s, &n);
|
||||
}
|
||||
for (int i = 0; i < STREAM_NODES; i++) {
|
||||
cbm_edge_t ring = {.project = "test",
|
||||
.source_id = ids[i],
|
||||
.target_id = ids[(i + 1) % STREAM_NODES],
|
||||
.type = "CALLS"};
|
||||
cbm_edge_t hub = {.project = "test",
|
||||
.source_id = ids[i],
|
||||
.target_id = ids[i % STREAM_HUBS],
|
||||
.type = "CALLS"};
|
||||
cbm_store_insert_edge(s, &ring);
|
||||
if (i >= STREAM_HUBS) { /* no hub self-loops */
|
||||
cbm_store_insert_edge(s, &hub);
|
||||
}
|
||||
}
|
||||
static const char *const queries[] = {
|
||||
"MATCH (a)-[:CALLS]->(b) RETURN count(*)",
|
||||
"MATCH (a)-[:CALLS]->(b) RETURN a.qualified_name, b.qualified_name",
|
||||
"MATCH (a:Function)-[:CALLS]->(b) RETURN b.qualified_name, count(a) AS c ORDER BY c DESC",
|
||||
"MATCH (a)-[:CALLS]->(b) WITH b.name AS n, count(DISTINCT a.name) AS c RETURN n, c",
|
||||
"MATCH (a)-[:CALLS]->(b) RETURN DISTINCT b.name LIMIT 300",
|
||||
};
|
||||
for (size_t i = 0; i < sizeof(queries) / sizeof(queries[0]); i++) {
|
||||
cbm_cypher_result_t ref;
|
||||
ASSERT_TRUE(stream_matches_single_pass(s, queries[i], 0, &ref));
|
||||
ASSERT_GT(ref.row_count, 0);
|
||||
cbm_cypher_result_free(&ref);
|
||||
}
|
||||
cbm_cypher_result_t ref;
|
||||
ASSERT_TRUE(stream_matches_single_pass(s, queries[0], 0, &ref));
|
||||
ASSERT_STR_EQ(ref.rows[0][0], "5195"); /* 2600 ring + 2595 hub edges */
|
||||
cbm_cypher_result_free(&ref);
|
||||
cbm_store_close(s);
|
||||
PASS();
|
||||
}
|
||||
|
||||
/* ══════════════════════════════════════════════════════════════════ */
|
||||
|
||||
SUITE(cypher) {
|
||||
@@ -6077,4 +6338,10 @@ SUITE(cypher) {
|
||||
RUN_TEST(cypher_order_by_is_the_stable_reference_order);
|
||||
RUN_TEST(cypher_distinct_keeps_first_occurrences);
|
||||
RUN_TEST(cypher_grouping_keeps_first_seen_groups);
|
||||
/* Streamed execution: chunked start set == single pass */
|
||||
RUN_TEST(cypher_stream_chunks_match_single_pass);
|
||||
RUN_TEST(cypher_stream_optional_across_chunks);
|
||||
RUN_TEST(cypher_stream_limit_and_max_rows_across_chunks);
|
||||
RUN_TEST(cypher_stream_default_chunk_boundary);
|
||||
RUN_TEST(cypher_stream_scan_failure_fails_query);
|
||||
}
|
||||
@@ -402,6 +402,122 @@ TEST(store_node_find_by_label) {
|
||||
PASS();
|
||||
}
|
||||
|
||||
static bool scan_str_same(const char *a, const char *b) {
|
||||
return (!a && !b) || (a && b && strcmp(a, b) == 0);
|
||||
}
|
||||
|
||||
static bool scan_node_same(const cbm_node_t *a, const cbm_node_t *b) {
|
||||
return a->id == b->id && scan_str_same(a->project, b->project) &&
|
||||
scan_str_same(a->label, b->label) && scan_str_same(a->name, b->name) &&
|
||||
scan_str_same(a->qualified_name, b->qualified_name) &&
|
||||
scan_str_same(a->file_path, b->file_path) && a->start_line == b->start_line &&
|
||||
a->end_line == b->end_line && scan_str_same(a->properties_json, b->properties_json);
|
||||
}
|
||||
|
||||
/* The streamed scan gives the eager scan's rows, field for field and in the
|
||||
* same order, at any batch size, while other lookups (including the eager
|
||||
* scans' own cached statements) run between batches. Returns the row count,
|
||||
* or -1 on the first difference; *paged tells whether the scan paged. */
|
||||
static int scan_matches_eager(cbm_store_t *s, const char *project, const char *label, int batch,
|
||||
bool *paged) {
|
||||
cbm_node_t *eager = NULL;
|
||||
int n = 0;
|
||||
int rc = label ? cbm_store_find_nodes_by_label(s, project, label, &eager, &n)
|
||||
: cbm_store_find_nodes(s, project, &eager, &n);
|
||||
if (rc != CBM_STORE_OK) {
|
||||
return -1;
|
||||
}
|
||||
cbm_store_node_scan_t *scan = cbm_store_node_scan_open(s, project, label);
|
||||
*paged = cbm_store_node_scan_paged(scan);
|
||||
cbm_node_t buf[64];
|
||||
int seen = 0;
|
||||
bool same = scan != NULL && batch <= 64;
|
||||
int got = 0;
|
||||
while (same && cbm_store_node_scan_next(scan, buf, batch, &got) == CBM_STORE_OK && got > 0) {
|
||||
for (int i = 0; i < got; i++, seen++) {
|
||||
same = same && seen < n && scan_node_same(&buf[i], &eager[seen]);
|
||||
cbm_node_free_fields(&buf[i]);
|
||||
}
|
||||
cbm_node_t *other = NULL;
|
||||
int other_n = 0;
|
||||
(void)cbm_store_find_nodes(s, project, &other, &other_n);
|
||||
cbm_store_free_nodes(other, other_n);
|
||||
(void)cbm_store_find_nodes_by_label(s, project, "Class", &other, &other_n);
|
||||
cbm_store_free_nodes(other, other_n);
|
||||
}
|
||||
/* An exhausted scan stays exhausted. */
|
||||
same = same && cbm_store_node_scan_next(scan, buf, batch, &got) == CBM_STORE_OK && got == 0;
|
||||
cbm_store_node_scan_close(scan);
|
||||
cbm_store_free_nodes(eager, n);
|
||||
return same && seen == n ? n : -1;
|
||||
}
|
||||
|
||||
/* Pages of 1, 3 and 64 rows end inside files, at file boundaries and inside
|
||||
* the NULL file_path rows, which the eager all-nodes order puts first. The
|
||||
* upsert API stores "" for a missing path; other writers may store NULL. */
|
||||
TEST(store_node_scan_streams_eager_scan_order) {
|
||||
enum { SCAN_NODES = 60 };
|
||||
cbm_store_t *s = cbm_store_open_memory();
|
||||
cbm_store_upsert_project(s, "test", "/tmp/test");
|
||||
cbm_store_upsert_project(s, "other", "/tmp/other");
|
||||
static const char *const labels[] = {"Function", "Class", "Method"};
|
||||
static const char *const files[] = {"b.c", "a.c", "", "c.c"};
|
||||
int in_test = 0;
|
||||
int test_functions = 0;
|
||||
int other_methods = 0;
|
||||
for (int i = 0; i < SCAN_NODES; i++) {
|
||||
bool other = i % 4 == 3;
|
||||
char name[32];
|
||||
char qn[64];
|
||||
char props[64];
|
||||
snprintf(name, sizeof(name), "n%d", (i * 17) % 40);
|
||||
snprintf(qn, sizeof(qn), "%s.n%d", other ? "other" : "test", i);
|
||||
snprintf(props, sizeof(props), "{\"complexity\":%d}", i);
|
||||
cbm_node_t n = {.project = other ? "other" : "test",
|
||||
.label = labels[i % 3],
|
||||
.name = name,
|
||||
.qualified_name = qn,
|
||||
.file_path = files[(i / 2) % 4],
|
||||
.start_line = i,
|
||||
.end_line = i + 5,
|
||||
.properties_json = i % 5 ? props : NULL};
|
||||
ASSERT_GT(cbm_store_upsert_node(s, &n), 0);
|
||||
in_test += other ? 0 : 1;
|
||||
test_functions += !other && i % 3 == 0 ? 1 : 0;
|
||||
other_methods += other && i % 3 == 2 ? 1 : 0;
|
||||
}
|
||||
ASSERT_EQ(cbm_store_exec(s, "UPDATE nodes SET file_path = NULL WHERE start_line % 7 = 0;"), 0);
|
||||
cbm_node_t *eager = NULL;
|
||||
int eager_n = 0;
|
||||
ASSERT_EQ(cbm_store_find_nodes(s, "test", &eager, &eager_n), CBM_STORE_OK);
|
||||
ASSERT_TRUE(eager_n > 0 && eager[0].file_path == NULL); /* NULL rows come first */
|
||||
cbm_store_free_nodes(eager, eager_n);
|
||||
|
||||
static const int batches[] = {1, 3, 64};
|
||||
bool paged = false;
|
||||
for (size_t b = 0; b < sizeof(batches) / sizeof(batches[0]); b++) {
|
||||
ASSERT_EQ(scan_matches_eager(s, "test", NULL, batches[b], &paged), in_test);
|
||||
ASSERT_TRUE(paged);
|
||||
ASSERT_EQ(scan_matches_eager(s, "test", "Function", batches[b], &paged), test_functions);
|
||||
ASSERT_TRUE(paged);
|
||||
ASSERT_EQ(scan_matches_eager(s, "other", NULL, batches[b], &paged),
|
||||
SCAN_NODES - in_test); /* project filter */
|
||||
ASSERT_EQ(scan_matches_eager(s, "other", "Method", batches[b], &paged), other_methods);
|
||||
ASSERT_EQ(scan_matches_eager(s, "test", "NoSuchLabel", batches[b], &paged), 0);
|
||||
ASSERT_EQ(scan_matches_eager(s, "missing", NULL, batches[b], &paged), 0);
|
||||
}
|
||||
|
||||
/* Without the node indexes the eager queries take other plans, which the
|
||||
* pages would not reproduce: the scan hands out the eager rows instead. */
|
||||
ASSERT_EQ(cbm_store_drop_indexes(s), 0);
|
||||
ASSERT_EQ(scan_matches_eager(s, "test", NULL, 3, &paged), in_test);
|
||||
ASSERT_FALSE(paged);
|
||||
ASSERT_EQ(scan_matches_eager(s, "test", "Function", 3, &paged), test_functions);
|
||||
ASSERT_FALSE(paged);
|
||||
cbm_store_close(s);
|
||||
PASS();
|
||||
}
|
||||
|
||||
TEST(store_node_find_by_file) {
|
||||
cbm_store_t *s = cbm_store_open_memory();
|
||||
cbm_store_upsert_project(s, "test", "/tmp/test");
|
||||
@@ -2516,6 +2632,7 @@ SUITE(store_nodes) {
|
||||
RUN_TEST(store_node_crud);
|
||||
RUN_TEST(store_node_dedup);
|
||||
RUN_TEST(store_node_find_by_label);
|
||||
RUN_TEST(store_node_scan_streams_eager_scan_order);
|
||||
RUN_TEST(store_node_find_by_file);
|
||||
RUN_TEST(store_file_outline_is_filtered_stable_bounded_and_cancellable_issue469);
|
||||
RUN_TEST(store_file_outline_fails_closed_on_text_budget_issue469);
|
||||
|
||||
Reference in new issue
Block a user