From 00910315420b254bd6945b679dc3dfd23c938869 Mon Sep 17 00:00:00 2001 From: Martin Vogel Date: Mon, 2 Nov 2026 22:04:45 +0100 Subject: [PATCH] perf(store): write resolver evidence straight into b-tree pages Writing evidence row by row through SQL was the largest single cost of an index: 12.9 s of writes on django, 35.6 s on kubernetes. Rows now go straight to b-tree pages and digests run in parallel, the proof pool keeps a memo of the texts of a session, and hot keys no longer go through snprintf, which takes a process-wide locale lock on macOS. Writes take under 0.4 s on django and under 0.9 s on kubernetes. Databases shrink: kubernetes 4,364 to 4,162 MB, django 932 to 869 MB. Nothing readers use is removed; the same rows are stored. Signed-off-by: Martin Vogel --- internal/cbm/resolve/policy.c | 19 +- internal/cbm/resolve/proof.c | 141 ++++- internal/cbm/resolve/proof.h | 12 + internal/cbm/sqlite_writer.c | 227 +++++++- internal/cbm/sqlite_writer.h | 23 + scripts/memory-core-baseline.txt | 4 +- src/foundation/fast_format.h | 30 ++ src/graph_buffer/graph_buffer.c | 50 +- src/graph_buffer/graph_buffer.h | 8 + src/pipeline/pipeline.c | 91 +++- src/pipeline/pipeline_internal.h | 2 + src/simhash/minhash.c | 15 +- src/store/store_resolver.c | 24 +- src/store/store_resolver.h | 7 +- src/store/store_resolver_evidence.c | 790 +++++++++++++++++++++++----- src/store/store_resolver_evidence.h | 20 +- tests/test_resolver_generation.c | 2 +- tests/test_sqlite_writer.c | 117 ++++ 18 files changed, 1387 insertions(+), 195 deletions(-) create mode 100644 src/foundation/fast_format.h diff --git a/internal/cbm/resolve/policy.c b/internal/cbm/resolve/policy.c index 292454af5..58309b4d8 100644 --- a/internal/cbm/resolve/policy.c +++ b/internal/cbm/resolve/policy.c @@ -1,6 +1,7 @@ #include "policy.h" #include "proof.h" #include "foundation/mem_core.h" +#include "foundation/fast_format.h" #include #include @@ -9,11 +10,19 @@ char *cbm_resolve_properties_tail(const cbm_resolve_record_t *record, int32_t *line) { const char *json = record->properties_json ? record->properties_json : "{}"; char head[160]; - int head_length = snprintf( - head, sizeof(head), - "{\"strategy\":\"shared_resolver\",\"resolver_rule\":%d,\"start_byte\":%u,\"end_byte\":%u," - "\"source_origin\":%u", - (int)record->result.rule, record->start_byte, record->end_byte, record->source_origin); + char *head_end = head; +#define HEAD_TEXT(literal) head_end = cbm_put_text(head_end, literal, sizeof(literal) - 1) + HEAD_TEXT("{\"strategy\":\"shared_resolver\",\"resolver_rule\":"); + head_end = cbm_put_u64(head_end, (uint64_t)record->result.rule); + HEAD_TEXT(",\"start_byte\":"); + head_end = cbm_put_u64(head_end, record->start_byte); + HEAD_TEXT(",\"end_byte\":"); + head_end = cbm_put_u64(head_end, record->end_byte); + HEAD_TEXT(",\"source_origin\":"); + head_end = cbm_put_u64(head_end, record->source_origin); +#undef HEAD_TEXT + *head_end = '\0'; + int head_length = (int)(head_end - head); const char *rest = json; size_t before = strlen(json); const char *after = NULL; diff --git a/internal/cbm/resolve/proof.c b/internal/cbm/resolve/proof.c index 989bf513c..f37d0db1c 100644 --- a/internal/cbm/resolve/proof.c +++ b/internal/cbm/resolve/proof.c @@ -5,7 +5,7 @@ #include #include -#define PROOF_MAX_DEPTH 256u +#define PROOF_MAX_DEPTH CBM_RESOLVE_PROOF_MAX_DEPTH struct cbm_resolve_proof_entry { cbm_resolve_proof_t node; @@ -20,9 +20,18 @@ typedef struct { const cbm_resolve_proof_t *source; const cbm_resolve_proof_t *destination; } proof_memo_slot_t; +/* A source pool stores each string once, so one clone session sees the same few source + * addresses again and again: project, file, scope, two 64-character identities. */ +typedef struct { + const char *source; + const char *destination; + size_t length; +} proof_text_memo_t; +enum { PROOF_TEXT_MEMO_SLOTS = 1 << 12 }; typedef struct { proof_memo_slot_t *slots; size_t capacity, count; + proof_text_memo_t *texts; cbm_resolve_proof_pool_t *pool; cbm_resolve_proof_visit_fn visit; void *context; @@ -145,6 +154,8 @@ static int compare_payload(const cbm_resolve_proof_view_t *a, const cbm_resolve_ count = proof_strings(a, as); (void)proof_strings(b, bs); for (size_t i = 0; i < count; i++) { + if (as[i] == bs[i]) + continue; int order = strcmp(text_or_empty(as[i]), text_or_empty(bs[i])); if (order) return order; @@ -238,18 +249,38 @@ static bool grow_texts(cbm_resolve_proof_pool_t *pool) { /* A fact repeats its project, file, scope and two content identities, and a * project holds millions of facts: every distinct string is stored once per * pool. Identity stays full-string comparison; sharing is only storage. */ +/* This table's hash is never seen outside it, so it reads a word at a time. */ +static uint64_t text_hash(const char *text) { + size_t length = strlen(text); + uint64_t hash = UINT64_C(0x9E3779B97F4A7C15) ^ length; + for (; length >= sizeof(uint64_t); length -= sizeof(uint64_t), text += sizeof(uint64_t)) { + uint64_t word; + memcpy(&word, text, sizeof(word)); + hash = (hash ^ word) * UINT64_C(0xFF51AFD7ED558CCD); + hash ^= hash >> 29; + } + uint64_t tail = 0; + memcpy(&tail, text, length); + hash = (hash ^ tail) * UINT64_C(0xC4CEB9FE1A85EC53); + return hash ^ (hash >> 32); +} +/* Facts repeat their source's strings, already this pool's own. A string in this cache is + * in the pool's arena for the pool's whole life, so no other text can share its address. */ +static size_t own_recent_slot(const char *text) { + uintptr_t address = (uintptr_t)text; + return (size_t)(address ^ (address >> 11)) & (CBM_RESOLVE_PROOF_RECENT_TEXTS - 1); +} static bool own_text(cbm_resolve_proof_pool_t *pool, const char **field) { if (!*field || !**field) { *field = NULL; return true; } + const char *const *recent = &pool->recent_texts[own_recent_slot(*field)]; + if (*recent == *field) + return true; if (!grow_texts(pool)) return proof_fail(pool, CBM_RESOLVE_ANALYSIS_FAILED); - uint64_t hash = UINT64_C(14695981039346656037); - for (const unsigned char *at = (const unsigned char *)*field; *at; at++) { - hash ^= *at; - hash *= UINT64_C(1099511628211); - } + uint64_t hash = text_hash(*field); cbm_resolve_proof_text_t *slot = text_slot(pool->texts, pool->text_capacity, *field, hash); if (!slot->text) { const char *copy = cbm_arena_strdup(pool->arena, *field); @@ -259,6 +290,7 @@ static bool own_text(cbm_resolve_proof_pool_t *pool, const char **field) { pool->text_count++; } *field = slot->text; + pool->recent_texts[own_recent_slot(slot->text)] = slot->text; return true; } static bool own_payload(cbm_resolve_proof_pool_t *pool, cbm_resolve_proof_view_t *v) { @@ -292,10 +324,56 @@ static bool proof_charge(cbm_resolve_proof_pool_t *pool, const cbm_resolve_proof return false; return true; } -static const cbm_resolve_proof_t *intern(cbm_resolve_proof_pool_t *pool, - cbm_resolve_proof_view_t value, bool fingerprinted) { +/* Exactly what proof_charge and own_payload do for one string, with each one's work on a + * repeated source address answered by the session. A clone session's source strings stay + * alive and unchanged until it ends, so an address names one text for all of it. */ +static bool session_own_text(cbm_resolve_proof_pool_t *pool, proof_text_memo_t *memo, + const char **field) { + const char *source = *field; + if (!source || !*source) { + *field = NULL; + return proof_step(pool, 1); + } + uintptr_t address = (uintptr_t)source; + proof_text_memo_t *slot = &memo[(address ^ (address >> 12)) & (PROOF_TEXT_MEMO_SLOTS - 1)]; + if (slot->source != source) { + const char *owned = source; + if (!own_text(pool, &owned)) + return false; + *slot = + (proof_text_memo_t){.source = source, .destination = owned, .length = strlen(owned)}; + } + *field = slot->destination; + return proof_step(pool, slot->length + 1); +} +static bool session_own_payload(cbm_resolve_proof_pool_t *pool, proof_text_memo_t *memo, + cbm_resolve_proof_view_t *v) { + const char **fields[] = {&v->source.project, + &v->source.file, + &v->source.module, + &v->source.scope, + &v->source.content_identity, + &v->source.configuration_identity, + &v->key.lookup_domain, + &v->key.lookup_key, + &v->key.declaration_identity, + &v->key.binding_identity, + &v->key.type_identity, + &v->key.detail, + &v->outcome.target, + &v->outcome.reason}; + for (size_t i = 0; i < sizeof(fields) / sizeof(fields[0]); i++) + if (!session_own_text(pool, memo, fields[i])) + return false; + return true; +} +static const cbm_resolve_proof_t *intern_with(cbm_resolve_proof_pool_t *pool, + cbm_resolve_proof_view_t value, bool fingerprinted, + proof_text_memo_t *memo) { if (!proof_step(pool, 1) || - !(fingerprinted ? proof_charge(pool, &value) : proof_hash(pool, &value)) || + !(memo ? session_own_payload(pool, memo, &value) + : fingerprinted ? proof_charge(pool, &value) + : proof_hash(pool, &value)) || !grow_pool(pool)) return NULL; size_t slot = value.fingerprint & (pool->capacity - 1); @@ -310,7 +388,7 @@ static const cbm_resolve_proof_t *intern(cbm_resolve_proof_pool_t *pool, return NULL; } cbm_resolve_proof_entry_t *entry = cbm_arena_calloc(pool->arena, sizeof(*entry)); - if (!entry || !own_payload(pool, &value)) { + if (!entry || (!memo && !own_payload(pool, &value))) { proof_fail(pool, CBM_RESOLVE_ANALYSIS_FAILED); return NULL; } @@ -331,6 +409,10 @@ static const cbm_resolve_proof_t *intern(cbm_resolve_proof_pool_t *pool, pool->count++; return &entry->node; } +static const cbm_resolve_proof_t *intern(cbm_resolve_proof_pool_t *pool, + cbm_resolve_proof_view_t value, bool fingerprinted) { + return intern_with(pool, value, fingerprinted, NULL); +} static bool leaf_kind(cbm_resolve_proof_kind_t kind) { return kind >= CBM_PROOF_SOURCE_FACT && kind <= CBM_PROOF_REGISTERED_DECLARATION; } @@ -455,7 +537,7 @@ static const cbm_resolve_proof_t *walk_node(proof_walk_t *walk, const cbm_resolv if (ok && walk->pool) { cbm_resolve_proof_view_t value = proof->value; value.premises = premises; - result = intern(walk->pool, value, true); + result = intern_with(walk->pool, value, true, walk->texts); } else if (ok && walk->visit(proof, walk->context)) { result = proof; } @@ -463,8 +545,8 @@ static const cbm_resolve_proof_t *walk_node(proof_walk_t *walk, const cbm_resolv memo_slot(walk, proof)->destination = result; return result; } -bool cbm_resolve_proof_clone_begin(cbm_resolve_proof_clone_session_t *session, - cbm_resolve_proof_pool_t *destination) { +static bool clone_begin(cbm_resolve_proof_clone_session_t *session, + cbm_resolve_proof_pool_t *destination, bool many_roots) { if (!session) return false; *session = (cbm_resolve_proof_clone_session_t){.destination = destination}; @@ -475,9 +557,17 @@ bool cbm_resolve_proof_clone_begin(cbm_resolve_proof_clone_session_t *session, return proof_fail(destination, CBM_RESOLVE_ANALYSIS_FAILED); walk->pool = destination; walk->limit = destination->node_limit; + /* Without it a clone simply hashes each string, so a failure here changes nothing. */ + walk->texts = + many_roots ? cbm_calloc(CBM_MEM_CLASS_EXTRACT, PROOF_TEXT_MEMO_SLOTS * sizeof(*walk->texts)) + : NULL; session->state = walk; return true; } +bool cbm_resolve_proof_clone_begin(cbm_resolve_proof_clone_session_t *session, + cbm_resolve_proof_pool_t *destination) { + return clone_begin(session, destination, true); +} const cbm_resolve_proof_t *cbm_resolve_proof_clone_root(cbm_resolve_proof_clone_session_t *session, const cbm_resolve_proof_t *proof) { if (!session || !session->state || !proof_step(session->destination, 1)) @@ -496,6 +586,7 @@ void cbm_resolve_proof_clone_end(cbm_resolve_proof_clone_session_t *session) { proof_walk_t *walk = session->state; if (walk) { cbm_free(CBM_MEM_CLASS_EXTRACT, walk->slots); + cbm_free(CBM_MEM_CLASS_EXTRACT, walk->texts); cbm_free(CBM_MEM_CLASS_EXTRACT, walk); } *session = (cbm_resolve_proof_clone_session_t){0}; @@ -506,7 +597,7 @@ const cbm_resolve_proof_t *cbm_resolve_proof_clone(cbm_resolve_proof_pool_t *des cbm_resolve_proof_pool_owns(destination, proof)) return proof_step(destination, 1) ? proof : NULL; cbm_resolve_proof_clone_session_t session; - if (!cbm_resolve_proof_clone_begin(&session, destination)) + if (!clone_begin(&session, destination, false)) return NULL; const cbm_resolve_proof_t *result = cbm_resolve_proof_clone_root(&session, proof); cbm_resolve_proof_clone_end(&session); @@ -604,11 +695,12 @@ static void derived_finish(cbm_resolve_proof_view_t *value, const cbm_resolve_pr value->outcome.candidate_count = 0; } } -bool cbm_resolve_proof_compose(cbm_resolve_proof_view_t *out, cbm_resolve_proof_kind_t kind, - const cbm_resolve_proof_source_t *source, - const cbm_resolve_proof_key_t *key, - const cbm_resolve_result_t *outcome, - const cbm_resolve_proof_t **premises, size_t count, bool closed) { +bool cbm_resolve_proof_compose_payload(cbm_resolve_proof_view_t *out, cbm_resolve_proof_kind_t kind, + const cbm_resolve_proof_source_t *source, + const cbm_resolve_proof_key_t *key, + const cbm_resolve_result_t *outcome, + const cbm_resolve_proof_t **premises, size_t count, + bool closed) { if (!out || !key || !outcome || !premises || !count) return false; *out = derived_begin(kind, source, key, outcome, closed); @@ -617,7 +709,16 @@ bool cbm_resolve_proof_compose(cbm_resolve_proof_view_t *out, cbm_resolve_proof_ return false; derived_finish(out, premises, count); out->outcome.proof = NULL; - return proof_hash(NULL, out); + return true; +} +bool cbm_resolve_proof_compose(cbm_resolve_proof_view_t *out, cbm_resolve_proof_kind_t kind, + const cbm_resolve_proof_source_t *source, + const cbm_resolve_proof_key_t *key, + const cbm_resolve_result_t *outcome, + const cbm_resolve_proof_t **premises, size_t count, bool closed) { + return cbm_resolve_proof_compose_payload(out, kind, source, key, outcome, premises, count, + closed) && + proof_hash(NULL, out); } const cbm_resolve_proof_t *cbm_resolve_proof_derive(cbm_resolve_proof_pool_t *pool, cbm_resolve_proof_kind_t kind, diff --git a/internal/cbm/resolve/proof.h b/internal/cbm/resolve/proof.h index f107977e6..139d2bed9 100644 --- a/internal/cbm/resolve/proof.h +++ b/internal/cbm/resolve/proof.h @@ -6,6 +6,9 @@ #include #include +/* No fact is deeper than this, and each is deeper than all its premises. */ +#define CBM_RESOLVE_PROOF_MAX_DEPTH 256u + /* Immutable captured-source identity. Empty fields explicitly mean unknown; * a caller must not recover them by re-reading a potentially changed file. */ typedef struct { @@ -74,6 +77,7 @@ struct cbm_resolve_proof { cbm_resolve_proof_view_t value; }; +enum { CBM_RESOLVE_PROOF_RECENT_TEXTS = 64 }; typedef struct cbm_resolve_proof_entry cbm_resolve_proof_entry_t; typedef struct cbm_resolve_proof_text cbm_resolve_proof_text_t; typedef struct cbm_resolve_proof_pool { @@ -82,6 +86,7 @@ typedef struct cbm_resolve_proof_pool { size_t capacity, count; cbm_resolve_proof_text_t *texts; size_t text_capacity, text_count; + const char *recent_texts[CBM_RESOLVE_PROOF_RECENT_TEXTS]; uint64_t work, work_limit; size_t node_limit, premise_limit; cbm_resolve_analysis_t failure; @@ -141,6 +146,13 @@ bool cbm_resolve_proof_compose(cbm_resolve_proof_view_t *out, cbm_resolve_proof_ const cbm_resolve_proof_key_t *key, const cbm_resolve_result_t *outcome, const cbm_resolve_proof_t **premises, size_t count, bool closed); +/* The same view with no fingerprint, for a reader that only needs its fields. */ +bool cbm_resolve_proof_compose_payload(cbm_resolve_proof_view_t *out, cbm_resolve_proof_kind_t kind, + const cbm_resolve_proof_source_t *source, + const cbm_resolve_proof_key_t *key, + const cbm_resolve_result_t *outcome, + const cbm_resolve_proof_t **premises, size_t count, + bool closed); /* The ordering of cbm_resolve_proof_compare, for a view that has no node yet. */ int cbm_resolve_proof_view_compare(const cbm_resolve_proof_view_t *a, const cbm_resolve_proof_view_t *b); diff --git a/internal/cbm/sqlite_writer.c b/internal/cbm/sqlite_writer.c index d684ff3a0..d53f587ee 100644 --- a/internal/cbm/sqlite_writer.c +++ b/internal/cbm/sqlite_writer.c @@ -28,6 +28,7 @@ #include #include #include +#include "foundation/mem_core.h" #include #ifdef _WIN32 @@ -1081,6 +1082,13 @@ static uint8_t *build_index_entry_unique_2int_2text_rowid(int64_t v1, int64_t v2 // Ensure leaves array has capacity for one more entry. // Returns false on allocation failure. +static uint8_t *sep_cell_copy(const uint8_t *cell, int length) { + uint8_t *copy = (uint8_t *)malloc((size_t)length); + if (copy) + memcpy(copy, cell, (size_t)length); + return copy; +} + static bool pb_ensure_leaf_cap(PageBuilder *pb) { if (pb->leaf_count < pb->leaf_cap) { return true; @@ -1309,8 +1317,7 @@ static bool pb_promote_and_flush(PageBuilder *pb, uint8_t **cells, int *cell_len return false; } pb->leaves[pb->leaf_count].max_key = 0; - pb->leaves[pb->leaf_count].sep_cell = (uint8_t *)malloc(cell_lens[prev_idx]); - memcpy(pb->leaves[pb->leaf_count].sep_cell, cells[prev_idx], cell_lens[prev_idx]); + pb->leaves[pb->leaf_count].sep_cell = sep_cell_copy(cells[prev_idx], cell_lens[prev_idx]); pb->leaves[pb->leaf_count].sep_cell_len = cell_lens[prev_idx]; // Un-add the last cell — it's promoted to the interior separator. @@ -1382,8 +1389,7 @@ static uint32_t write_index_btree(FILE *fp, uint32_t *next_page, uint8_t **cells } pb.leaves[pb.leaf_count].max_key = 0; int last = count - SKIP_ONE; - pb.leaves[pb.leaf_count].sep_cell = (uint8_t *)malloc(cell_lens[last]); - memcpy(pb.leaves[pb.leaf_count].sep_cell, cells[last], cell_lens[last]); + pb.leaves[pb.leaf_count].sep_cell = sep_cell_copy(cells[last], cell_lens[last]); pb.leaves[pb.leaf_count].sep_cell_len = cell_lens[last]; pb_flush_leaf(&pb); } @@ -1777,9 +1783,12 @@ static uint32_t build_node_index_sorted(FILE *fp, uint32_t *next_page, CBMDumpNo // --- Main entry point --- /* Write context passed to sub-phases of cbm_write_db. */ +enum { WRITER_MAX_EXTRA_TABLES = 8 }; typedef struct { FILE *fp; uint32_t next_page; + MasterEntry extra_tables[WRITER_MAX_EXTRA_TABLES]; + int extra_table_count; char final_path[CBM_SZ_4K]; char temp_path[CBM_SZ_4K]; const char *project; @@ -2324,7 +2333,15 @@ static int write_db_after_nodes(write_db_ctx_t *w, uint32_t nodes_root) { }; int master_count = sizeof(master) / sizeof(master[0]); - int rc2 = write_master_page1(fp, master, master_count, next_page); + MasterEntry *all = (MasterEntry *)cbm_alloc( + CBM_MEM_CLASS_DUMP, sizeof(master) + (size_t)w->extra_table_count * sizeof(*all)); + if (!all) { + return discard_writer_output(w, ERR_WRITE_FAILED); + } + memcpy(all, master, sizeof(master)); + memcpy(all + master_count, w->extra_tables, (size_t)w->extra_table_count * sizeof(*all)); + int rc2 = write_master_page1(fp, all, master_count + w->extra_table_count, next_page); + cbm_free(CBM_MEM_CLASS_DUMP, all); if (rc2 != 0) { return discard_writer_output(w, rc2); } @@ -2375,6 +2392,204 @@ cbm_db_writer_t *cbm_writer_open(const char *path) { return w; } +/* A row's serial types go first in its record, so the values wait in a buffer until the + * row ends. Nothing per row is allocated: this runs millions of times. */ +enum { TABLE_MAX_COLUMNS = 64, TABLE_INITIAL_ROW = 1024 }; +struct cbm_writer_table { + cbm_db_writer_t *w; + PageBuilder pb; + const char *name, *sql; + int64_t serials[TABLE_MAX_COLUMNS]; + int columns; + uint8_t *values; + size_t values_used, values_capacity; + uint8_t *cell; + size_t cell_capacity; + int last_cell_length; /* The last cell on the current leaf, should it be promoted. */ + bool failed; +}; + +static bool table_room(uint8_t **buffer, size_t *capacity, size_t needed) { + if (needed <= *capacity) + return true; + size_t next = *capacity ? *capacity : TABLE_INITIAL_ROW; + while (next < needed) + next *= GROWTH_FACTOR; + uint8_t *grown = (uint8_t *)cbm_realloc(CBM_MEM_CLASS_DUMP, *buffer, next); + if (!grown) + return false; + *buffer = grown; + *capacity = next; + return true; +} + +cbm_writer_table_t *cbm_writer_table_begin(cbm_db_writer_t *w, const char *name, + const char *create_sql) { + if (!w || w->err || w->node_rows_written || w->wc.extra_table_count >= WRITER_MAX_EXTRA_TABLES) + return NULL; + cbm_writer_table_t *t = (cbm_writer_table_t *)cbm_calloc(CBM_MEM_CLASS_DUMP, sizeof(*t)); + if (!t) + return NULL; + t->w = w; + t->name = name; + t->sql = create_sql; + pb_init(&t->pb, w->wc.fp, w->nodes_pb.next_page, true); + return t; +} + +void cbm_writer_row_begin(cbm_writer_table_t *t) { + t->columns = 0; + t->values_used = 0; +} + +static void table_column(cbm_writer_table_t *t, int64_t serial, const void *data, size_t length) { + if (t->columns >= TABLE_MAX_COLUMNS || + !table_room(&t->values, &t->values_capacity, t->values_used + length)) { + t->failed = true; + return; + } + t->serials[t->columns++] = serial; + if (length) + memcpy(t->values + t->values_used, data, length); + t->values_used += length; +} + +void cbm_writer_row_null(cbm_writer_table_t *t) { + table_column(t, 0, NULL, 0); +} + +void cbm_writer_row_int(cbm_writer_table_t *t, int64_t value) { + int64_t serial = int_serial_type(value); + int bytes = int_storage_bytes((int)serial); + uint8_t encoded[SERIAL_SIZE_INT64]; + put_int_be(encoded, value, bytes); + table_column(t, serial, encoded, (size_t)bytes); +} + +void cbm_writer_row_text(cbm_writer_table_t *t, const char *text) { + size_t length = strlen(text); + table_column(t, text_serial_type((int)length), text, length); +} + +void cbm_writer_row_blob(cbm_writer_table_t *t, const void *data, int length) { + table_column(t, (int64_t)length * BLOB_SERIAL_MUL + BLOB_SERIAL_BASE, data, (size_t)length); +} + +/* Written leaves come from the shared page counter, since the tables interleave. */ +static void table_flush_leaf(cbm_writer_table_t *t) { + uint32_t *next_page = &t->w->nodes_pb.next_page; + t->pb.next_page = *next_page; + pb_flush_leaf(&t->pb); + *next_page = t->pb.next_page; +} + +static bool table_note_separator(cbm_writer_table_t *t, const uint8_t *cell, int length) { + if (!pb_ensure_leaf_cap(&t->pb)) + return false; + uint8_t *separator = sep_cell_copy(cell, length); + if (!separator) + return false; + PageRef *leaf = &t->pb.leaves[t->pb.leaf_count]; + leaf->max_key = 0; + leaf->sep_cell = separator; + leaf->sep_cell_len = length; + return true; +} + +static bool table_add_cell(cbm_writer_table_t *t, const uint8_t *cell, int length) { + if (!pb_cell_fits(&t->pb, length) && t->pb.cell_count > 0) { + /* The leaf's last cell moves up as the separator, as in write_index_btree. */ + int last = t->last_cell_length; + if (!table_note_separator(t, t->pb.page + t->pb.content_offset, last)) + return false; + t->pb.cell_count--; + t->pb.content_offset += last; + t->pb.ptr_offset -= CELL_PTR_SIZE; + table_flush_leaf(t); + } + if (!pb_cell_fits(&t->pb, length)) + return false; + pb_add_cell(&t->pb, cell, length); + t->last_cell_length = length; + return true; +} + +bool cbm_writer_row_end(cbm_writer_table_t *t) { + if (t->failed) + return false; + int serial_bytes = 0; + for (int i = 0; i < t->columns; i++) + serial_bytes += varint_len(t->serials[i]); + int header = serial_bytes + varint_len(serial_bytes + VARINT_MIN_LEN); + if (varint_len(header) != header - serial_bytes) + header = serial_bytes + varint_len(header); + size_t payload = (size_t)header + t->values_used; + if (payload > INT32_MAX || + !table_room(&t->cell, &t->cell_capacity, payload + VARINT_BUF_SIZE)) { + t->failed = true; + return false; + } + int at = put_varint(t->cell, (int64_t)payload); + at += put_varint(t->cell + at, header); + for (int i = 0; i < t->columns; i++) + at += put_varint(t->cell + at, t->serials[i]); + memcpy(t->cell + at, t->values, t->values_used); + at += (int)t->values_used; + bool ok; + if ((int64_t)payload > INDEX_OVERFLOW_MAX_LOCAL) { + uint8_t *whole = cells_alloc((size_t)at); + int length = at; + if (whole) { + memcpy(whole, t->cell, (size_t)at); + whole = overflowize_index_cell(t->pb.fp, &t->w->nodes_pb.next_page, whole, &length); + } + ok = whole && table_add_cell(t, whole, length); + if (whole) + cells_free(whole); + } else { + ok = table_add_cell(t, t->cell, at); + } + t->failed = !ok; + return ok; +} + +bool cbm_writer_table_end(cbm_writer_table_t *t) { + if (!t) + return false; + bool ok = !t->failed; + uint32_t *next_page = &t->w->nodes_pb.next_page; + uint32_t root = 0; + if (ok && t->pb.cell_count > 0) { + /* Nothing reads the last leaf's separator, but the interior builder frees it. */ + ok = table_note_separator(t, t->pb.page + t->pb.content_offset, t->last_cell_length); + if (ok) + table_flush_leaf(t); + } + if (ok) { + t->pb.next_page = *next_page; + if (t->pb.leaf_count == 0) + root = write_empty_index_leaf(t->pb.fp, next_page); + else if (t->pb.leaf_count == SKIP_ONE) + root = t->pb.leaves[0].page_num; + else { + root = pb_build_interior(&t->pb, true); + *next_page = t->pb.next_page; + } + ok = root != 0; + } + if (ok) { + t->w->wc.extra_tables[t->w->wc.extra_table_count++] = + (MasterEntry){"table", t->name, t->name, root, t->sql}; + } else { + t->w->err = ERR_WRITE_FAILED; + } + pb_free(&t->pb); + cbm_free(CBM_MEM_CLASS_DUMP, t->values); + cbm_free(CBM_MEM_CLASS_DUMP, t->cell); + cbm_free(CBM_MEM_CLASS_DUMP, t); + return ok; +} + int cbm_writer_append_nodes(cbm_db_writer_t *w, const CBMDumpNode *nodes, int count) { if (!w) { return CBM_NOT_FOUND; @@ -2412,6 +2627,8 @@ int cbm_writer_finalize(cbm_db_writer_t *w, const char *project, const char *roo uint32_t nodes_root = 0; if (err == 0) { if (w->node_rows_written == 0) { + /* Streamed tables may already hold pages past the first. */ + w->wc.next_page = w->nodes_pb.next_page; pb_free(&w->nodes_pb); nodes_root = write_table_btree(w->wc.fp, &w->wc.next_page, NULL, NULL, NULL, 0, false); } else { diff --git a/internal/cbm/sqlite_writer.h b/internal/cbm/sqlite_writer.h index b92a13719..9b4213554 100644 --- a/internal/cbm/sqlite_writer.h +++ b/internal/cbm/sqlite_writer.h @@ -1,6 +1,7 @@ #ifndef CBM_SQLITE_WRITER_H #define CBM_SQLITE_WRITER_H +#include #include // --- Input structs (flat, borrowed strings) --- @@ -90,6 +91,28 @@ int cbm_writer_finalize(cbm_db_writer_t *w, const char *project, const char *roo CBMDumpEdge *edges, int edge_count, CBMDumpVector *vectors, int vector_count, CBMDumpTokenVec *token_vecs, int token_vec_count); +// --- WITHOUT ROWID tables, streamed --- +// +// A table of the caller's own, written straight to pages like the rest of this file, +// for rows too many to send through SQL one at a time. Rows must arrive in primary-key +// order with the key columns first, as SQLite itself stores such a table. Several tables +// may be open at once, between open and the first append_nodes. create_sql is what +// sqlite_master will say and must outlive finalize; it decides everything SQLite will +// assume about the rows. +typedef struct cbm_writer_table cbm_writer_table_t; + +cbm_writer_table_t *cbm_writer_table_begin(cbm_db_writer_t *w, const char *name, + const char *create_sql); +void cbm_writer_row_begin(cbm_writer_table_t *t); +void cbm_writer_row_null(cbm_writer_table_t *t); +void cbm_writer_row_int(cbm_writer_table_t *t, int64_t value); +void cbm_writer_row_text(cbm_writer_table_t *t, const char *text); +void cbm_writer_row_blob(cbm_writer_table_t *t, const void *data, int length); +// False once a write has failed; the failure is sticky and finalize reports it too. +bool cbm_writer_row_end(cbm_writer_table_t *t); +// Frees t. Every table begun must be ended before the first append_nodes. +bool cbm_writer_table_end(cbm_writer_table_t *t); + // Byte offset of 1-based page `page_num` in the output file. 64-bit on every // platform, so databases larger than 2 GiB are addressed correctly on Windows. int64_t cbm_writer_page_offset(uint32_t page_num); diff --git a/scripts/memory-core-baseline.txt b/scripts/memory-core-baseline.txt index 235bc6bf8..91a795bd9 100644 --- a/scripts/memory-core-baseline.txt +++ b/scripts/memory-core-baseline.txt @@ -12,7 +12,7 @@ internal/cbm/lsp/c_lsp.c 4 internal/cbm/lsp/perl_lsp.c 21 internal/cbm/macro_table.c 1 internal/cbm/service_patterns.c 2 -internal/cbm/sqlite_writer.c 74 +internal/cbm/sqlite_writer.c 73 src/cli/activation_transaction.c 78 src/cli/agent_clients.c 73 src/cli/agent_profiles.c 7 @@ -68,7 +68,7 @@ src/pipeline/pass_importance.c 4 src/pipeline/pass_infrascan.c 2 src/pipeline/pass_k8s.c 6 src/pipeline/pass_lsp_cross.c 46 -src/pipeline/pass_parallel.c 29 +src/pipeline/pass_parallel.c 28 src/pipeline/pass_pkgmap.c 63 src/pipeline/pass_semantic.c 3 src/pipeline/pass_tests.c 3 diff --git a/src/foundation/fast_format.h b/src/foundation/fast_format.h new file mode 100644 index 000000000..6aee615a0 --- /dev/null +++ b/src/foundation/fast_format.h @@ -0,0 +1,30 @@ +#ifndef CBM_FAST_FORMAT_H +#define CBM_FAST_FORMAT_H + +#include +#include +#include + +/* Composing a key once per edge or occurrence. snprintf reads the locale on every call, + * and on macOS that takes a process-wide lock, so with all workers running it spent more + * time waiting than formatting. These write the same digits as %llu, without a NUL, and + * return the new end; the caller has already sized the buffer. */ +#define CBM_PUT_U64_MAX 20 + +static inline char *cbm_put_u64(char *out, uint64_t value) { + char digits[CBM_PUT_U64_MAX]; + size_t at = sizeof(digits); + do { + digits[--at] = (char)('0' + value % 10); + value /= 10; + } while (value); + memcpy(out, digits + at, sizeof(digits) - at); + return out + (sizeof(digits) - at); +} + +static inline char *cbm_put_text(char *out, const char *text, size_t length) { + memcpy(out, text, length); + return out + length; +} + +#endif diff --git a/src/graph_buffer/graph_buffer.c b/src/graph_buffer/graph_buffer.c index 6d286a9ff..db2e2d5ef 100644 --- a/src/graph_buffer/graph_buffer.c +++ b/src/graph_buffer/graph_buffer.c @@ -9,6 +9,7 @@ * The buffer frees everything in cbm_gbuf_free(). */ #include "foundation/constants.h" +#include "foundation/fast_format.h" enum { GB_ERR = -1, @@ -708,10 +709,28 @@ static void make_edge_key(char *buf, size_t bufsz, int64_t src, int64_t tgt, con return; } } + size_t type_length = strlen(type); + if (src >= 0 && tgt >= 0 && type_length + 2 * CBM_PUT_U64_MAX + 3 <= bufsz) { + char *at = cbm_put_u64(buf, (uint64_t)src); + *at++ = ':'; + at = cbm_put_u64(at, (uint64_t)tgt); + *at++ = ':'; + at = cbm_put_text(at, type, type_length); + *at = '\0'; + return; + } snprintf(buf, bufsz, "%lld:%lld:%s", (long long)src, (long long)tgt, type); } static void make_src_type_key(char *buf, size_t bufsz, int64_t src, const char *type) { + size_t type_length = strlen(type); + if (src >= 0 && type_length + CBM_PUT_U64_MAX + 2 <= bufsz) { + char *at = cbm_put_u64(buf, (uint64_t)src); + *at++ = ':'; + at = cbm_put_text(at, type, type_length); + *at = '\0'; + return; + } snprintf(buf, bufsz, "%lld:%s", (long long)src, type); } @@ -2480,6 +2499,28 @@ int cbm_gbuf_delete_edges_by_type(cbm_gbuf_t *gb, const char *type) { return gbuf_delete_edges_if(gb, gbuf_edge_is_type, type); } +typedef struct { + const char *const *types; + size_t count; +} gbuf_type_set_t; + +static bool gbuf_edge_in_types(const cbm_gbuf_t *gb, const cbm_gbuf_edge_t *e, const void *ud) { + (void)gb; + const gbuf_type_set_t *set = ud; + for (size_t i = 0; i < set->count; i++) + if (strcmp(e->type, set->types[i]) == 0) + return true; + return false; +} + +int cbm_gbuf_delete_edges_by_types(cbm_gbuf_t *gb, const char *const *types, size_t count) { + if (!gb || !types) { + return CBM_NOT_FOUND; + } + gbuf_type_set_t set = {types, count}; + return gbuf_delete_edges_if(gb, gbuf_edge_in_types, &set); +} + typedef struct { const char *type; const char *label; @@ -3039,6 +3080,11 @@ static void release_and_remap_vectors(cbm_gbuf_t *gb, const int64_t *temp_to_fin } int cbm_gbuf_dump_to_sqlite(cbm_gbuf_t *gb, const char *path) { + return cbm_gbuf_dump_to_sqlite_with(gb, path, NULL, NULL); +} + +int cbm_gbuf_dump_to_sqlite_with(cbm_gbuf_t *gb, const char *path, cbm_gbuf_dump_tables_fn tables, + void *context) { if (!gb || !path) { return CBM_NOT_FOUND; } @@ -3090,11 +3136,11 @@ int cbm_gbuf_dump_to_sqlite(cbm_gbuf_t *gb, const char *path) { return CBM_NOT_FOUND; } + int rc = tables && !tables(w, context) ? CBM_NOT_FOUND : 0; CBM_PROF_START(t_append); enum { DUMP_PARTITION_NODES = 1 << 16 }; bool free_heavy = false; - int rc = 0; - for (int off = 0; off < node_idx; off += DUMP_PARTITION_NODES) { + for (int off = 0; rc == 0 && off < node_idx; off += DUMP_PARTITION_NODES) { int chunk = node_idx - off; if (chunk > DUMP_PARTITION_NODES) { chunk = DUMP_PARTITION_NODES; diff --git a/src/graph_buffer/graph_buffer.h b/src/graph_buffer/graph_buffer.h index 118cd6002..ff67beebf 100644 --- a/src/graph_buffer/graph_buffer.h +++ b/src/graph_buffer/graph_buffer.h @@ -9,6 +9,7 @@ #ifndef CBM_GRAPH_BUFFER_H #define CBM_GRAPH_BUFFER_H +#include "sqlite_writer.h" #include #include #include @@ -230,6 +231,8 @@ int cbm_gbuf_edge_count_by_type(const cbm_gbuf_t *gb, const char *type); /* Delete all edges of a type. */ int cbm_gbuf_delete_edges_by_type(cbm_gbuf_t *gb, const char *type); +/* Several types in one walk of the edges and one rebuild of their indexes. */ +int cbm_gbuf_delete_edges_by_types(cbm_gbuf_t *gb, const char *const *types, size_t count); /* Delete the edges of `type` whose source node has label `label`. */ int cbm_gbuf_delete_edges_from_label(cbm_gbuf_t *gb, const char *type, const char *label); @@ -257,6 +260,11 @@ int cbm_gbuf_store_token_vector(cbm_gbuf_t *gb, const char *token, const uint8_t * Assigns sequential final IDs and remaps edge references. * Returns 0 on success, -1 on error. */ int cbm_gbuf_dump_to_sqlite(cbm_gbuf_t *gb, const char *path); +/* As above. tables runs once the file is open and before any node reaches it, to add + * tables of the caller's own; false fails the dump. */ +typedef bool (*cbm_gbuf_dump_tables_fn)(cbm_db_writer_t *writer, void *context); +int cbm_gbuf_dump_to_sqlite_with(cbm_gbuf_t *gb, const char *path, cbm_gbuf_dump_tables_fn tables, + void *context); /* Flush the buffer to an existing store via the store API. * Deletes existing project data first. Returns 0 on success. */ diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index 528a332d0..ca84bfb34 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -20,6 +20,8 @@ enum { CBM_DIR_PERMS = 0755, PL_RING = 4, PL_RING_MASK = 3, PL_SEQ_PASSES = 9 }; #include "pipeline/pipeline.h" #include "pipeline/artifact.h" #include "pipeline/pipeline_internal.h" +#include "foundation/fast_format.h" +#include "store/store_resolver_evidence.h" #include "pipeline/doc_links.h" #include "pipeline/lsp_surface.h" #include "pipeline/lsp_resolve.h" @@ -1451,14 +1453,28 @@ static char *unresolved_publication_key(const char *file, const char *scope, con cbm_utf8_lossy_dup(CBM_MEM_CLASS_RESOLVE, spelling, strlen(spelling), NULL); char *key = NULL; if (clean_file && clean_scope && clean_spelling) { - size_t cap = strlen(clean_file) + strlen(clean_scope) + strlen(clean_spelling) + - strlen(content) + strlen(configuration) + 192; + const char *const parts[] = {clean_file, clean_scope, clean_spelling, content, + configuration}; + size_t cap = 192; + for (size_t i = 0; i < sizeof(parts) / sizeof(parts[0]); i++) + cap += strlen(parts[i]); key = cbm_alloc(CBM_MEM_CLASS_RESOLVE, cap); - if (key) - snprintf(key, cap, "%zu:%s|%zu:%s|%zu:%s|%zu:%s|%zu:%s|%u:%u:%u", strlen(clean_file), - clean_file, strlen(clean_scope), clean_scope, strlen(clean_spelling), - clean_spelling, strlen(content), content, strlen(configuration), configuration, - start, end, origin); + if (key) { + char *at = key; + for (size_t i = 0; i < sizeof(parts) / sizeof(parts[0]); i++) { + size_t length = strlen(parts[i]); + at = cbm_put_u64(at, length); + *at++ = ':'; + at = cbm_put_text(at, parts[i], length); + *at++ = '|'; + } + at = cbm_put_u64(at, start); + *at++ = ':'; + at = cbm_put_u64(at, end); + *at++ = ':'; + at = cbm_put_u64(at, origin); + *at = '\0'; + } } cbm_free(CBM_MEM_CLASS_RESOLVE, clean_file); cbm_free(CBM_MEM_CLASS_RESOLVE, clean_scope); @@ -3776,11 +3792,16 @@ static bool promote_mode_to_existing_coverage(cbm_pipeline_t *p) { return promoted; } +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS +/* The evidence takes the SQL route, so a test can compare what the two routes store. */ +bool cbm_pipeline_test_resolver_stage_off; +#endif /* Defined below, next to the other publication helpers. */ static char *create_staging_path(const char *final_path); static char *create_staging_path_impl(const char *final_path, cbm_pipeline_t *frozen); static int publish_staged_impl(char *path, const cbm_pipeline_generation_t *generation, - bool fts_wholesale, bool healthy, cbm_pipeline_t *frozen); + bool fts_wholesale, bool healthy, cbm_pipeline_t *frozen, + const cbm_resolver_stage_t *resolver_stage); static int finalize_staged_impl(char *path, const char *destination, atomic_int *cancelled, bool healthy, cbm_pipeline_t *frozen); @@ -4104,6 +4125,33 @@ static int prepare_existing_generation_for_replace(const char *db_path, return 0; } +/* A generation headed for a new file writes its evidence with the file. An incomplete one + * writes none here: publishing refuses it as it always did. */ +typedef struct { + const cbm_pipeline_generation_t *generation; + cbm_resolver_stage_t *stage; +} resolver_stage_write_t; + +static bool resolver_stage_tables(cbm_db_writer_t *writer, void *context) { + resolver_stage_write_t *write = context; + const cbm_pipeline_generation_t *generation = write->generation; + if (!generation->gbuf || !cbm_gbuf_resolutions_complete(generation->gbuf)) + return true; +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS + if (cbm_pipeline_test_resolver_stage_off) + return true; +#endif + size_t count = 0, file_count = 0, summary_count = 0; + const cbm_resolve_record_t *rows = cbm_gbuf_resolutions(generation->gbuf, &count); + const cbm_resolve_file_t *files = cbm_gbuf_resolver_files(generation->gbuf, &file_count); + const cbm_resolve_summary_t *summaries = + cbm_gbuf_resolver_summaries(generation->gbuf, &summary_count); + write->stage = + cbm_store_resolver_stage_write(writer, generation->project, rows, count, files, file_count, + summaries, summary_count, generation->workers); + return write->stage != NULL; +} + static int publish_generation_impl(const cbm_pipeline_generation_t *generation, cbm_pipeline_t *frozen) { if (!generation || !generation->gbuf || !generation->final_db_path || !generation->project || @@ -4133,26 +4181,34 @@ static int publish_generation_impl(const cbm_pipeline_generation_t *generation, return CBM_PIPELINE_PERSIST_FAILED; } - int dump_rc = cbm_gbuf_dump_to_sqlite(generation->gbuf, stage_path); + resolver_stage_write_t stage_write = {.generation = generation}; + int dump_rc = cbm_gbuf_dump_to_sqlite_with(generation->gbuf, stage_path, resolver_stage_tables, + &stage_write); if (dump_rc != 0) { + cbm_store_resolver_stage_free(stage_write.stage); discard_generation_frozen(stage_path, frozen); free(stage_path); return CBM_PIPELINE_PERSIST_FAILED; } #if defined(CBM_INCREMENTAL_TEST_API) && CBM_INCREMENTAL_TEST_API if (cbm_pipeline_persist_test_take_failure_after_stage_dump()) { + cbm_store_resolver_stage_free(stage_write.stage); discard_generation_frozen(stage_path, frozen); free(stage_path); return CBM_PIPELINE_PERSIST_FAILED; } #endif if (generation->cancelled && atomic_load(generation->cancelled)) { + cbm_store_resolver_stage_free(stage_write.stage); discard_generation_frozen(stage_path, frozen); free(stage_path); return CBM_PIPELINE_ABORT_PRESERVE_DB; } - return publish_staged_impl(stage_path, generation, true, false, frozen); + int published = + publish_staged_impl(stage_path, generation, true, false, frozen, stage_write.stage); + cbm_store_resolver_stage_free(stage_write.stage); + return published; } /* Write the generation's doc_link_unresolved rows (plus the error marker when @@ -4188,7 +4244,7 @@ static int publish_doc_links(cbm_store_t *store, const cbm_pipeline_generation_t * the nodes it created. */ static int publish_staged_impl(char *stage_path, const cbm_pipeline_generation_t *generation, bool fts_wholesale, bool destination_known_healthy, - cbm_pipeline_t *frozen) { + cbm_pipeline_t *frozen, const cbm_resolver_stage_t *resolver_stage) { struct timespec t_pub; cbm_clock_gettime(CLOCK_MONOTONIC, &t_pub); cbm_store_t *store = cbm_store_open_path(stage_path); @@ -4221,7 +4277,11 @@ static int publish_staged_impl(char *stage_path, const cbm_pipeline_generation_t } /* Occurrence outcomes are part of this exact graph generation. They * publish atomically with it, including references that produced no edge. */ - if (ok && generation->gbuf) { + if (ok && generation->gbuf && resolver_stage) { + ok = cbm_gbuf_resolutions_complete(generation->gbuf) && + cbm_store_resolver_publish_stage(store, generation->project, resolver_stage) == + CBM_STORE_OK; + } else if (ok && generation->gbuf) { size_t count = 0; const cbm_resolve_record_t *rows = cbm_gbuf_resolutions(generation->gbuf, &count); size_t file_count = 0, summary_count = 0; @@ -4230,8 +4290,8 @@ static int publish_staged_impl(char *stage_path, const cbm_pipeline_generation_t cbm_gbuf_resolver_summaries(generation->gbuf, &summary_count); ok = cbm_gbuf_resolutions_complete(generation->gbuf) && cbm_store_resolver_publish_generation(store, generation->project, rows, count, files, - file_count, summaries, - summary_count) == CBM_STORE_OK; + file_count, summaries, summary_count, + generation->workers) == CBM_STORE_OK; } /* ADRs are NOT written here: they live in the ".adr.db" sidecar, which * this rebuild never touches, so there is nothing to re-apply. */ @@ -4328,7 +4388,7 @@ int cbm_pipeline_publish_generation(const cbm_pipeline_generation_t *generation) int cbm_pipeline_publish_staged(char *path, const cbm_pipeline_generation_t *generation, bool fts_wholesale, bool healthy) { - return publish_staged_impl(path, generation, fts_wholesale, healthy, NULL); + return publish_staged_impl(path, generation, fts_wholesale, healthy, NULL, NULL); } char *cbm_pipeline_create_staging_path(const char *final_path) { @@ -4517,6 +4577,7 @@ static int dump_and_persist_hashes(cbm_pipeline_t *p, const cbm_file_hash_t *bas } cbm_pipeline_generation_t generation = { .gbuf = p->gbuf, + .workers = cbm_pipeline_worker_count(p), .final_db_path = db_path, .project = p->project_name, .cancelled = p->cancelled, diff --git a/src/pipeline/pipeline_internal.h b/src/pipeline/pipeline_internal.h index ca578ce4d..2862b7e91 100644 --- a/src/pipeline/pipeline_internal.h +++ b/src/pipeline/pipeline_internal.h @@ -1793,6 +1793,8 @@ typedef struct { * proxy buffer has no section text or function tokens to recompute * them); a row whose node is gone resolves to nothing when read. */ bool doc_candidates_in_place; + /* Threads for the digests of the resolver evidence; below 2 means this thread. */ + int workers; } cbm_pipeline_generation_t; /* Serialize and fully populate a sibling staging database, then atomically diff --git a/src/simhash/minhash.c b/src/simhash/minhash.c index d8680c8cc..4943bd4a1 100644 --- a/src/simhash/minhash.c +++ b/src/simhash/minhash.c @@ -9,6 +9,7 @@ */ #include "simhash/minhash.h" #include "foundation/constants.h" +#include "foundation/fast_format.h" #include "foundation/mem_core.h" #include "foundation/log.h" /* Inline all xxHash functions — avoids separate compilation unit. */ @@ -248,11 +249,19 @@ static int hash_trigrams(const char **tokens, int token_count, cbm_minhash_t *ou continue; } - int len = snprintf(trigram_buf, sizeof(trigram_buf), "%s|%s|%s", tokens[i], - tokens[i + SKIP_ONE], tokens[i + TRIGRAM_WINDOW]); - if (len <= 0 || (size_t)len >= sizeof(trigram_buf)) { + /* This ran under snprintf's process-wide locale lock on macOS, once per trigram. */ + size_t first = strlen(tokens[i]), second = strlen(tokens[i + SKIP_ONE]), + third = strlen(tokens[i + TRIGRAM_WINDOW]); + if (first + second + third + TRIGRAM_WINDOW >= sizeof(trigram_buf)) { continue; } + char *trigram_end = cbm_put_text(trigram_buf, tokens[i], first); + *trigram_end++ = '|'; + trigram_end = cbm_put_text(trigram_end, tokens[i + SKIP_ONE], second); + *trigram_end++ = '|'; + trigram_end = cbm_put_text(trigram_end, tokens[i + TRIGRAM_WINDOW], third); + *trigram_end = '\0'; + int len = (int)(trigram_end - trigram_buf); /* A repeated trigram is the same string with the same weight: its * update would lower the same minima to the same values. Only a first diff --git a/src/store/store_resolver.c b/src/store/store_resolver.c index 9e145d329..3336a68e0 100644 --- a/src/store/store_resolver.c +++ b/src/store/store_resolver.c @@ -250,7 +250,8 @@ static int resolver_write_generation(cbm_store_t *store, const char *project, const cbm_resolve_record_t *rows, size_t count, const cbm_resolve_file_t *files, size_t file_count, const cbm_resolve_summary_t *summaries, size_t summary_count, - bool commit_batches) { + bool commit_batches, int workers, + const cbm_resolver_stage_t *stage) { if (!store || !project || (!rows && count)) return CBM_STORE_ERR; sqlite3 *db = cbm_store_get_db(store); @@ -259,9 +260,11 @@ static int resolver_write_generation(cbm_store_t *store, const char *project, "rules_version INTEGER NOT NULL);"; if (!db || sqlite3_exec(db, "SAVEPOINT resolver_replace", NULL, NULL, NULL) != SQLITE_OK) return CBM_STORE_ERR; - bool ok = sqlite3_exec(db, schema, NULL, NULL, NULL) == SQLITE_OK && - cbm_store_resolver_evidence(db, project, rows, count, files, file_count, summaries, - summary_count, commit_batches); + bool ok = + sqlite3_exec(db, schema, NULL, NULL, NULL) == SQLITE_OK && + (stage ? cbm_store_resolver_stage_finish(db, project, stage) + : cbm_store_resolver_evidence(db, project, rows, count, files, file_count, summaries, + summary_count, commit_batches, workers)); sqlite3_stmt *stmt = NULL; if (ok) ok = sqlite3_prepare_v2(db, "INSERT OR REPLACE INTO resolver_generations VALUES(?,?)", -1, @@ -291,16 +294,23 @@ int cbm_store_resolver_replace_generation(cbm_store_t *store, const char *projec const cbm_resolve_summary_t *summaries, size_t summary_count) { return resolver_write_generation(store, project, rows, count, files, file_count, summaries, - summary_count, false); + summary_count, false, 1, NULL); } int cbm_store_resolver_publish_generation(cbm_store_t *store, const char *project, const cbm_resolve_record_t *rows, size_t count, const cbm_resolve_file_t *files, size_t file_count, const cbm_resolve_summary_t *summaries, - size_t summary_count) { + size_t summary_count, int workers) { return resolver_write_generation(store, project, rows, count, files, file_count, summaries, - summary_count, true); + summary_count, true, workers, NULL); +} + +int cbm_store_resolver_publish_stage(cbm_store_t *store, const char *project, + const cbm_resolver_stage_t *stage) { + return stage ? resolver_write_generation(store, project, NULL, 0, NULL, 0, NULL, 0, true, 1, + stage) + : CBM_STORE_ERR; } static bool resolver_count_outcomes(sqlite3 *db, const char *project, diff --git a/src/store/store_resolver.h b/src/store/store_resolver.h index 0b54130c4..5dc1f9307 100644 --- a/src/store/store_resolver.h +++ b/src/store/store_resolver.h @@ -16,11 +16,16 @@ int cbm_store_resolver_replace_generation(cbm_store_t *store, const char *projec /* The same write for a private staging store that a failure discards. Rows commit * in batches, so the WAL stays small instead of holding a second copy of them. A * partial generation can never pass for a whole one: the receipt is written last. */ +/* The rest of a generation whose rows were written into the file as it was built. */ +typedef struct cbm_resolver_stage cbm_resolver_stage_t; +int cbm_store_resolver_publish_stage(cbm_store_t *store, const char *project, + const cbm_resolver_stage_t *stage); +/* workers: threads for the digests of a big write; 1 or less runs on this thread. */ int cbm_store_resolver_publish_generation(cbm_store_t *store, const char *project, const cbm_resolve_record_t *rows, size_t count, const cbm_resolve_file_t *files, size_t file_count, const cbm_resolve_summary_t *summaries, - size_t summary_count); + size_t summary_count, int workers); /* Fixed-size schema/receipt check, independent of occurrence count. Missing or * subsequently modified generation metadata is unavailable and needs rebuild. */ int cbm_store_resolver_current(cbm_store_t *store, const char *project, bool *present); diff --git a/src/store/store_resolver_evidence.c b/src/store/store_resolver_evidence.c index 94039ce44..2ad5672bc 100644 --- a/src/store/store_resolver_evidence.c +++ b/src/store/store_resolver_evidence.c @@ -9,7 +9,10 @@ #include "resolve/policy.h" #include "resolve/proof.h" #include "resolve/snapshot.h" +#include "pipeline/worker_pool.h" +#include "sqlite_writer.h" #include +#include #include #include #include @@ -171,7 +174,15 @@ enum { VERIFIED_SOURCE_SLOTS = 1 << 12, /* One line per this many stored rows keeps a long write visibly alive: the * worker's log is the supervisor's no-progress heartbeat. */ - PROGRESS_ROWS = 1 << 20 + PROGRESS_ROWS = 1 << 20, + /* Composed facts are digested this many at a time, in order, and forgotten. */ + FACT_BATCH = 1 << 16, + /* Fewer jobs than this are not worth starting threads for. */ + FACT_INLINE_JOBS = 256, + ROOT_CHUNK_ROWS = 1 << 12, + FACT_TEXT_MEMO_SLOTS = 1 << 12, + /* A new file's first project: see cbm_resolver_stage. */ + DIRECT_PID = 1 }; typedef struct { @@ -196,22 +207,46 @@ typedef struct { const char *content, *file, *project, *configuration; uint32_t origin; } verified_source_t; +/* A string's length and whether it needs escaping, by address: every string a fact + * holds outlives the write, and a project repeats the same few in every fact. */ +typedef struct { + const char *pointer; + size_t length; + bool plain; +} fact_text_memo_t; typedef struct fact_text { char *bytes; size_t length, capacity; bool plain; + fact_text_memo_t *memo; } fact_text_t; +/* What is left of a write once its rows are in pages: everything the two remaining + * tables need, so the fact table and its digests can go before the graph is written. */ +struct cbm_resolver_stage { + char *table_sql[4]; + const cbm_resolve_summary_t **summaries; + uint32_t *summary_proofs; + size_t summary_count; + const cbm_resolve_file_t **files; + char (*derived)[65]; + size_t file_count; +}; typedef struct { sqlite3 *db; const char *project; sqlite3_int64 pid; sqlite3_stmt *text, *fact, *support, *occurrence; + /* Set instead of the statements when rows go straight into a new file's pages. */ + cbm_writer_table_t *text_table, *fact_table, *support_table, *occurrence_table; evidence_id_t *ids; size_t capacity, id_count; - size_t count; /* Facts written, which is also the last fact id. */ + size_t count; /* Facts planned, which is also the last fact id. */ + uintptr_t *order; /* Fact id - 1 -> its node, or its row with the low bit set. */ + size_t order_capacity; + const char **valid_text; + int workers; unsigned char *digests; size_t digest_capacity, digest_count; - unsigned char last_digest[DIGEST_BYTES]; dict_slot_t *dict; size_t dict_capacity, dict_count; dict_cache_t *cache; @@ -224,7 +259,6 @@ typedef struct { bool commit_batches; const cbm_resolve_file_t *const *files; size_t file_count; - struct fact_text fact_text; } evidence_write_t; static const char *text(const char *value) { @@ -266,6 +300,26 @@ static bool finish_row(evidence_write_t *w, sqlite3_stmt *s, bool ok, const char return ok; } +static bool table_finish(evidence_write_t *w, cbm_writer_table_t *table, bool ok, + const char *stage) { + ok = ok && cbm_writer_row_end(table); + if (ok) { + cbm_resolve_stats_add(CBM_RESOLVE_STAT_STORED_RECORDS, 1); + ok = note_rows(w, stage); + } + return ok; +} +/* A NULL text id means the empty string, as bind_optional writes it. */ +static bool table_text_id(cbm_writer_table_t *table, sqlite3_int64 id) { + if (id < 0) + return false; + if (id) + cbm_writer_row_int(table, id); + else + cbm_writer_row_null(table); + return true; +} + /* ── String dictionary ───────────────────────────────────────────── */ static uint64_t text_hash(const char *value) { @@ -323,10 +377,20 @@ static sqlite3_int64 dict_id(evidence_write_t *w, const char *value, bool stable if (!owned) return -1; sqlite3_int64 id = (sqlite3_int64)++w->dict_count; - bool ok = sqlite3_bind_int64(w->text, 1, w->pid) == SQLITE_OK && - sqlite3_bind_int64(w->text, 2, id) == SQLITE_OK && bind_text(w->text, 3, owned); - if (!finish_row(w, w->text, ok, "text")) - return -1; + if (w->text_table) { + cbm_writer_row_begin(w->text_table); + cbm_writer_row_int(w->text_table, w->pid); + cbm_writer_row_int(w->text_table, id); + cbm_writer_row_text(w->text_table, owned); + if (!table_finish(w, w->text_table, true, "text")) + return -1; + } else { + bool ok = sqlite3_bind_int64(w->text, 1, w->pid) == SQLITE_OK && + sqlite3_bind_int64(w->text, 2, id) == SQLITE_OK && + bind_text(w->text, 3, owned); + if (!finish_row(w, w->text, ok, "text")) + return -1; + } *slot = (dict_slot_t){.text = owned, .hash = hash, .id = id}; } if (cached) @@ -356,13 +420,13 @@ static size_t pointer_hash(const cbm_resolve_proof_t *node) { uintptr_t value = (uintptr_t)node >> 3; return (size_t)(value ^ (value >> 17)); } -static evidence_id_t *id_slot(evidence_write_t *w, const cbm_resolve_proof_t *node) { +static evidence_id_t *id_slot(const evidence_write_t *w, const cbm_resolve_proof_t *node) { size_t at = pointer_hash(node) & (w->capacity - 1); while (w->ids[at].node && w->ids[at].node != node) at = (at + 1) & (w->capacity - 1); return &w->ids[at]; } -static evidence_id_t *find_id(evidence_write_t *w, const cbm_resolve_proof_t *node) { +static evidence_id_t *find_id(const evidence_write_t *w, const cbm_resolve_proof_t *node) { if (!node || !w->capacity) return NULL; evidence_id_t *entry = id_slot(w, node); @@ -405,10 +469,8 @@ static bool grow_ids(evidence_write_t *w) { static const unsigned char *id_digest(const evidence_write_t *w, const evidence_id_t *entry) { return w->digests + (size_t)entry->digest * DIGEST_BYTES; } -/* Call grow_ids first. */ -static void hold_id(evidence_write_t *w, const cbm_resolve_proof_t *node, uint32_t id, - const unsigned char *digest) { - memcpy(w->digests + w->digest_count * DIGEST_BYTES, digest, DIGEST_BYTES); +/* Call grow_ids first. The digest slot is filled once every node has its id. */ +static void hold_id(evidence_write_t *w, const cbm_resolve_proof_t *node, uint32_t id) { *id_slot(w, node) = (evidence_id_t){.node = node, .id = id, .digest = (uint32_t)w->digest_count}; w->digest_count++; @@ -537,16 +599,26 @@ static void fact_append(fact_text_t *t, const char *bytes, size_t length) { memcpy(t->bytes + t->length, bytes, length); t->length += length; } +static fact_text_memo_t *fact_memo(fact_text_t *t, const char *value) { + uintptr_t address = (uintptr_t)value; + fact_text_memo_t *slot = &t->memo[(address ^ (address >> 13)) & (FACT_TEXT_MEMO_SLOTS - 1)]; + if (slot->pointer != value) { + size_t length = 0; + bool plain = true; + for (const unsigned char *p = (const unsigned char *)value; *p; p++, length++) + plain = plain && *p >= 0x20 && *p <= 0x7e && *p != '"' && *p != '\\'; + *slot = (fact_text_memo_t){.pointer = value, .length = length, .plain = plain}; + } + return slot; +} static void fact_str(fact_text_t *t, const char *key, size_t key_length, const char *value) { - size_t length = 0; - for (const unsigned char *p = (const unsigned char *)text(value); *p; p++, length++) { - if (*p < 0x20 || *p > 0x7e || *p == '"' || *p == '\\') { - t->plain = false; - return; - } + const fact_text_memo_t *known = fact_memo(t, text(value)); + if (!known->plain) { + t->plain = false; + return; } fact_append(t, key, key_length); - fact_append(t, text(value), length); + fact_append(t, known->pointer, known->length); fact_append(t, "\"", 1); } static void fact_num(fact_text_t *t, const char *key, size_t key_length, uint64_t value) { @@ -623,17 +695,27 @@ static bool note_type_digest(evidence_write_t *w, const cbm_resolve_proof_view_t } /* resolver_facts builds its payload from these strings, and a payload must stay * strict JSON, as fact_json's was. Source text that is not UTF-8 is stored as - * fact_json would have written it. A plain fact has already been checked. */ -static bool bind_fact_text(evidence_write_t *w, sqlite3_stmt *s, int col, const char *value) { - if (w->fact_text.plain || !value || cbm_utf8_valid(value, strlen(value))) - return bind_id(w, s, col, value); + * fact_json would have written it. Each address needs checking once. */ +static sqlite3_int64 fact_text_id(evidence_write_t *w, const char *value) { + if (!value || !*value) + return dict_id(w, value, true); + const char **checked = &w->valid_text[dict_cache_at(value)]; + if (*checked == value) + return dict_id(w, value, true); + if (cbm_utf8_valid(value, strlen(value))) { + *checked = value; + return dict_id(w, value, true); + } size_t length = 0; char *clean = cbm_utf8_lossy_dup(CBM_MEM_CLASS_STORE, value, strlen(value), &length); if (!clean) - return false; - bool ok = bind_optional(s, col, dict_id(w, clean, false)); + return -1; + sqlite3_int64 id = dict_id(w, clean, false); cbm_free(CBM_MEM_CLASS_STORE, clean); - return ok; + return id; +} +static bool bind_fact_text(evidence_write_t *w, sqlite3_stmt *s, int col, const char *value) { + return bind_optional(s, col, fact_text_id(w, value)); } static bool bind_fact(evidence_write_t *w, const cbm_resolve_proof_view_t *v, size_t id, const unsigned char *digest) { @@ -668,47 +750,108 @@ static bool bind_fact(evidence_write_t *w, const cbm_resolve_proof_view_t *v, si bind_fact_text(w, s, 30, v->outcome.reason) && sqlite3_bind_int(s, 31, v->predecessors_closed ? 1 : 0) == SQLITE_OK; } -/* node is NULL for a composed fact, which only its row will ever cite. */ -static bool write_fact(evidence_write_t *w, const cbm_resolve_proof_view_t *v, - const cbm_resolve_proof_t *node, size_t *written) { - if (!v || !source_matches(w, &v->source) || w->count >= UINT32_MAX || (node && !grow_ids(w))) - return false; - bool plain = fact_plain_text(&w->fact_text, v); +static bool table_fact_text(evidence_write_t *w, const char *value) { + return table_text_id(w->fact_table, fact_text_id(w, value)); +} +/* The same columns as bind_fact, for a row going straight into a page. */ +static bool table_fact(evidence_write_t *w, const cbm_resolve_proof_view_t *v, size_t id, + const unsigned char *digest) { + cbm_writer_table_t *t = w->fact_table; + cbm_writer_row_begin(t); + cbm_writer_row_int(t, w->pid); + cbm_writer_row_int(t, (sqlite3_int64)id); + cbm_writer_row_int(t, (int)v->kind); + cbm_writer_row_blob(t, digest, DIGEST_BYTES); + bool ok = table_fact_text(w, v->source.project) && table_fact_text(w, v->source.file) && + table_fact_text(w, v->source.module) && table_fact_text(w, v->source.scope) && + table_fact_text(w, v->source.content_identity) && + table_fact_text(w, v->source.configuration_identity); + cbm_writer_row_int(t, v->source.source_origin); + ok = ok && table_fact_text(w, v->key.lookup_domain) && table_fact_text(w, v->key.lookup_key) && + table_fact_text(w, v->key.declaration_identity) && + table_fact_text(w, v->key.binding_identity) && table_fact_text(w, v->key.type_identity) && + table_fact_text(w, v->key.detail); + cbm_writer_row_int(t, v->key.start_byte); + cbm_writer_row_int(t, v->key.end_byte); + cbm_writer_row_int(t, v->key.origin); + cbm_writer_row_int(t, v->key.flags); + cbm_writer_row_int(t, v->key.lookup_closed ? 1 : 0); + cbm_writer_row_int(t, (int)v->outcome.binding); + cbm_writer_row_int(t, (int)v->outcome.domain); + cbm_writer_row_int(t, (int)v->outcome.analysis); + cbm_writer_row_int(t, (int)v->outcome.rule); + cbm_writer_row_int(t, v->outcome.facts); + ok = ok && table_fact_text(w, v->outcome.target); + cbm_writer_row_int(t, (sqlite3_int64)v->outcome.candidate_count); + ok = ok && table_fact_text(w, v->outcome.reason); + cbm_writer_row_int(t, v->predecessors_closed ? 1 : 0); + return table_finish(w, t, ok, "facts"); +} + +/* What a fact is digested from: its text, then each premise's digest. Reads only what + * planning has fixed, so any number of these run at once. */ +static bool fact_digest(const evidence_write_t *w, const cbm_resolve_proof_view_t *v, + fact_text_t *fact_text, unsigned char digest[DIGEST_BYTES]) { + bool plain = fact_plain_text(fact_text, v); char *json = plain ? NULL : fact_json(v); if (!plain && !json) return false; - uint32_t id = (uint32_t)(w->count + 1); cbm_sha256_ctx hash; cbm_sha256_init(&hash); - cbm_resolve_hash_text(&hash, plain ? w->fact_text.bytes : json); + cbm_resolve_hash_text(&hash, plain ? fact_text->bytes : json); + cbm_free(CBM_MEM_CLASS_OTHER, json); + for (size_t i = 0; i < v->premise_count; i++) { + const evidence_id_t *premise = find_id(w, v->premises[i]); + if (!premise) + return false; + char hex[65]; + digest_hex(id_digest(w, premise), hex); + cbm_resolve_hash_text(&hash, hex); + } + cbm_sha256_final(&hash, digest); + return true; +} +static void fact_text_end(fact_text_t *fact_text) { + cbm_free(CBM_MEM_CLASS_STORE, fact_text->bytes); + cbm_free(CBM_MEM_CLASS_STORE, fact_text->memo); +} +static bool fact_text_begin(fact_text_t *fact_text) { + *fact_text = (fact_text_t){0}; + fact_text->memo = + cbm_calloc(CBM_MEM_CLASS_STORE, FACT_TEXT_MEMO_SLOTS * sizeof(*fact_text->memo)); + return fact_text->memo != NULL; +} + +/* One fact's row and its support rows. The digest is already known. */ +static bool write_fact_row(evidence_write_t *w, const cbm_resolve_proof_view_t *v, size_t id, + const unsigned char *digest) { bool ok = true; for (size_t i = 0; ok && i < v->premise_count; i++) { - evidence_id_t *premise = find_id(w, v->premises[i]); - ok = premise != NULL; - if (ok) { - char hex[65]; - digest_hex(id_digest(w, premise), hex); - cbm_resolve_hash_text(&hash, hex); - ok = finish_row(w, w->support, - sqlite3_bind_int64(w->support, 1, w->pid) == SQLITE_OK && - sqlite3_bind_int64(w->support, 2, (sqlite3_int64)id) == SQLITE_OK && - sqlite3_bind_int64(w->support, 3, (sqlite3_int64)i) == SQLITE_OK && - sqlite3_bind_int64(w->support, 4, (sqlite3_int64)premise->id) == - SQLITE_OK, - "supports"); + const evidence_id_t *premise = find_id(w, v->premises[i]); + if (w->support_table) { + ok = premise != NULL; + if (ok) { + cbm_writer_row_begin(w->support_table); + cbm_writer_row_int(w->support_table, w->pid); + cbm_writer_row_int(w->support_table, (sqlite3_int64)id); + cbm_writer_row_int(w->support_table, (sqlite3_int64)i); + cbm_writer_row_int(w->support_table, (sqlite3_int64)premise->id); + ok = table_finish(w, w->support_table, true, "supports"); + } + continue; } + ok = premise && + finish_row(w, w->support, + sqlite3_bind_int64(w->support, 1, w->pid) == SQLITE_OK && + sqlite3_bind_int64(w->support, 2, (sqlite3_int64)id) == SQLITE_OK && + sqlite3_bind_int64(w->support, 3, (sqlite3_int64)i) == SQLITE_OK && + sqlite3_bind_int64(w->support, 4, (sqlite3_int64)premise->id) == + SQLITE_OK, + "supports"); } - cbm_sha256_final(&hash, w->last_digest); - cbm_free(CBM_MEM_CLASS_OTHER, json); - ok = ok && note_type_digest(w, v, w->last_digest) && - finish_row(w, w->fact, bind_fact(w, v, id, w->last_digest), "facts"); - if (ok) { - if (node) - hold_id(w, node, id, w->last_digest); - w->count++; - *written = id; - } - return ok; + return ok && note_type_digest(w, v, digest) && + (w->fact_table ? table_fact(w, v, id, digest) + : finish_row(w, w->fact, bind_fact(w, v, id, digest), "facts")); } /* Each distinct fact after its premises, so a fact id always follows the ids it @@ -716,7 +859,24 @@ static bool write_fact(evidence_write_t *w, const cbm_resolve_proof_view_t *v, * would cost a gigabyte on the largest repositories. Recursion is bounded because * a node's depth is checked when it is made, and a cycle cannot be built from * immutable nodes. */ -static bool write_facts_from(evidence_write_t *w, const cbm_resolve_proof_t *node) { +static bool plan_fact(evidence_write_t *w, uintptr_t ref, size_t *id) { + if (w->count >= UINT32_MAX) + return false; + if (w->count == w->order_capacity) { + size_t next = w->order_capacity ? w->order_capacity * 2 : 1024; + if (next > SIZE_MAX / sizeof(*w->order)) + return false; + uintptr_t *order = cbm_realloc(CBM_MEM_CLASS_STORE, w->order, next * sizeof(*order)); + if (!order) + return false; + w->order = order; + w->order_capacity = next; + } + w->order[w->count++] = ref; + *id = w->count; + return true; +} +static bool plan_facts_from(evidence_write_t *w, const cbm_resolve_proof_t *node) { cbm_resolve_stats_add(CBM_RESOLVE_STAT_PROPAGATED_FACTS, 1); if (find_id(w, node)) return true; @@ -724,10 +884,13 @@ static bool write_facts_from(evidence_write_t *w, const cbm_resolve_proof_t *nod if (!v || (v->premise_count && !v->premises)) return false; for (size_t i = 0; i < v->premise_count; i++) - if (!write_facts_from(w, v->premises[i])) + if (!plan_facts_from(w, v->premises[i])) return false; - size_t written = 0; - return write_fact(w, v, node, &written); + size_t id = 0; + if (!source_matches(w, &v->source) || !grow_ids(w) || !plan_fact(w, (uintptr_t)node, &id)) + return false; + hold_id(w, node, (uint32_t)id); + return true; } /* Roots are sorted structurally, so fact ids do not depend on occurrence order. A @@ -760,6 +923,17 @@ static bool compose_fact(composed_fact_t *fact, const cbm_resolve_record_t *row, return cbm_resolve_proof_compose(&fact->view, CBM_PROOF_OCCURRENCE_BINDING, &source, &key, &outcome, fact->premises, count, true); } +/* The same fact when only its payload will be read: a root's fingerprint is already known. */ +static bool compose_fact_payload(composed_fact_t *fact, const cbm_resolve_record_t *row, + const char *project) { + cbm_resolve_proof_source_t source; + cbm_resolve_proof_key_t key; + cbm_resolve_result_t outcome; + size_t count = 0; + cbm_resolve_record_envelope(row, project, &source, &key, &outcome, fact->premises, &count); + return cbm_resolve_proof_compose_payload(&fact->view, CBM_PROOF_OCCURRENCE_BINDING, &source, + &key, &outcome, fact->premises, count, true); +} /* Composition of these rows already succeeded when their fingerprints were taken. */ static const cbm_resolve_proof_view_t *root_view(const fact_root_t *root, composed_fact_t *fact) { if (!root_is_row(root)) @@ -781,8 +955,8 @@ static int compare_roots(const void *left, const void *right) { /* A composed root that compares equal to its neighbour is one fact with it, as * interning would have made them: the first writes it and the others take its id. * Two nodes are the same fact only if they are the same node. */ -static bool write_root(evidence_write_t *w, const fact_root_t *root, const fact_root_t *previous, - const cbm_resolve_record_t *rows, size_t *last_id) { +static bool plan_root(evidence_write_t *w, const fact_root_t *root, const fact_root_t *previous, + const cbm_resolve_record_t *rows, size_t *last_id) { bool same = previous && (root_is_row(root) || root_is_row(previous)) && !compare_roots(previous, root); if (!root_is_row(root)) { @@ -790,30 +964,148 @@ static bool write_root(evidence_write_t *w, const fact_root_t *root, const fact_ if (same && !find_id(w, node)) { if (!grow_ids(w)) return false; - hold_id(w, node, (uint32_t)*last_id, w->last_digest); + hold_id(w, node, (uint32_t)*last_id); return true; } - if (!write_facts_from(w, node)) + if (!plan_facts_from(w, node)) return false; - const evidence_id_t *entry = find_id(w, node); - *last_id = entry->id; - memcpy(w->last_digest, id_digest(w, entry), DIGEST_BYTES); + *last_id = find_id(w, node)->id; return true; } const cbm_resolve_record_t *row = root_row(root); if (!same) { composed_fact_t fact; - if (!compose_fact(&fact, row, w->project)) + if (!compose_fact_payload(&fact, row, w->project)) return false; for (size_t i = 0; i < fact.view.premise_count; i++) - if (!write_facts_from(w, fact.view.premises[i])) + if (!plan_facts_from(w, fact.view.premises[i])) return false; - if (!write_fact(w, &fact.view, NULL, last_id)) + if (!source_matches(w, &fact.view.source) || !plan_fact(w, root->ref, last_id)) return false; } w->row_facts[row - rows] = (uint32_t)*last_id; return true; } + +/* ── Digests ─────────────────────────────────────────────────────── */ + +enum { PROOF_DEPTH_LEVELS = CBM_RESOLVE_PROOF_MAX_DEPTH }; + +/* Workers take jobs from one counter and stop on the first failure. */ +typedef struct { + const evidence_write_t *w; + const evidence_id_t **nodes; /* Node digests: one depth at a time. */ + size_t count; + const uintptr_t *refs; /* Composed digests: one batch at a time. */ + composed_fact_t *facts; + unsigned char *digests; + _Atomic size_t next; + _Atomic bool failed; +} digest_pass_t; + +static bool digest_job(digest_pass_t *pass, size_t job, fact_text_t *fact_text) { + const evidence_write_t *w = pass->w; + if (pass->nodes) { + const evidence_id_t *entry = pass->nodes[job]; + return fact_digest(w, cbm_resolve_proof_view(entry->node), fact_text, + w->digests + (size_t)entry->digest * DIGEST_BYTES); + } + if (!(pass->refs[job] & 1)) + return true; + const cbm_resolve_record_t *row = + (const cbm_resolve_record_t *)(pass->refs[job] & ~(uintptr_t)1); + return compose_fact_payload(&pass->facts[job], row, w->project) && + fact_digest(w, &pass->facts[job].view, fact_text, pass->digests + job * DIGEST_BYTES); +} +static void digest_worker(int worker, void *context) { + (void)worker; + digest_pass_t *pass = context; + fact_text_t fact_text; + if (!fact_text_begin(&fact_text)) { + atomic_store(&pass->failed, true); + return; + } + for (;;) { + size_t job = atomic_fetch_add_explicit(&pass->next, 1, memory_order_relaxed); + if (job >= pass->count || atomic_load_explicit(&pass->failed, memory_order_relaxed)) + break; + if (!digest_job(pass, job, &fact_text)) + atomic_store(&pass->failed, true); + } + fact_text_end(&fact_text); +} +static bool digest_run(const evidence_write_t *w, digest_pass_t *pass) { + atomic_init(&pass->next, 0); + atomic_init(&pass->failed, false); + if (w->workers <= 1 || pass->count < FACT_INLINE_JOBS) { + digest_worker(0, pass); + } else { + cbm_parallel_for_opts_t options = {.max_workers = w->workers, .force_pthreads = false}; + cbm_parallel_for(w->workers, digest_worker, pass, options); + } + return !atomic_load(&pass->failed); +} +/* Every node's digest, shallowest first: a premise is always shallower than the fact + * citing it, so one depth at a time finds its premises done. */ +static bool digest_nodes(evidence_write_t *w) { + if (!w->id_count) + return true; + size_t counts[PROOF_DEPTH_LEVELS + 1] = {0}; + for (size_t i = 0; i < w->capacity; i++) { + if (w->ids[i].node) { + uint32_t depth = cbm_resolve_proof_view(w->ids[i].node)->depth; + if (depth > PROOF_DEPTH_LEVELS) + return false; + counts[depth]++; + } + } + const evidence_id_t **sorted = cbm_alloc(CBM_MEM_CLASS_STORE, w->id_count * sizeof(*sorted)); + if (!sorted) + return false; + size_t starts[PROOF_DEPTH_LEVELS + 2] = {0}; + for (size_t depth = 0; depth <= PROOF_DEPTH_LEVELS; depth++) + starts[depth + 1] = starts[depth] + counts[depth]; + size_t placed[PROOF_DEPTH_LEVELS + 1]; + memcpy(placed, starts, sizeof(placed)); + for (size_t i = 0; i < w->capacity; i++) + if (w->ids[i].node) + sorted[placed[cbm_resolve_proof_view(w->ids[i].node)->depth]++] = &w->ids[i]; + bool ok = true; + for (size_t depth = 0; ok && depth <= PROOF_DEPTH_LEVELS; depth++) { + if (!counts[depth]) + continue; + digest_pass_t pass = {.w = w, .nodes = sorted + starts[depth], .count = counts[depth]}; + ok = digest_run(w, &pass); + } + cbm_free(CBM_MEM_CLASS_STORE, sorted); + return ok; +} +/* Rows in fact-id order. A composed fact's digest is needed once, for its own row, so a + * batch is composed and digested together and dropped when its rows are written. */ +static bool write_planned_facts(evidence_write_t *w) { + composed_fact_t *facts = cbm_alloc(CBM_MEM_CLASS_STORE, FACT_BATCH * sizeof(*facts)); + unsigned char *digests = cbm_alloc(CBM_MEM_CLASS_STORE, (size_t)FACT_BATCH * DIGEST_BYTES); + bool ok = facts && digests; + for (size_t first = 0; ok && first < w->count; first += FACT_BATCH) { + size_t count = w->count - first < FACT_BATCH ? w->count - first : FACT_BATCH; + digest_pass_t pass = { + .w = w, .count = count, .refs = w->order + first, .facts = facts, .digests = digests}; + ok = digest_run(w, &pass); + for (size_t i = 0; ok && i < count; i++) { + uintptr_t ref = w->order[first + i]; + if (ref & 1) { + ok = write_fact_row(w, &facts[i].view, first + i + 1, digests + i * DIGEST_BYTES); + } else { + const cbm_resolve_proof_t *node = (const cbm_resolve_proof_t *)ref; + ok = write_fact_row(w, cbm_resolve_proof_view(node), first + i + 1, + id_digest(w, find_id(w, node))); + } + } + } + cbm_free(CBM_MEM_CLASS_STORE, facts); + cbm_free(CBM_MEM_CLASS_STORE, digests); + return ok; +} static bool write_facts(evidence_write_t *w, fact_root_t *roots, size_t count, const cbm_resolve_record_t *rows) { fact_root_project = w->project; @@ -824,12 +1116,12 @@ static bool write_facts(evidence_write_t *w, fact_root_t *roots, size_t count, qsort(roots, count, sizeof(*roots), compare_roots); size_t last_id = 0; for (size_t i = 0; i < count; i++) - if (!write_root(w, &roots[i], i ? &roots[i - 1] : NULL, rows, &last_id)) + if (!plan_root(w, &roots[i], i ? &roots[i - 1] : NULL, rows, &last_id)) return false; - return true; + return digest_nodes(w) && write_planned_facts(w); } /* A row's occurrence fact, or the node it already has. */ -static bool collect_row_roots(evidence_write_t *w, const cbm_resolve_record_t *row, +static bool collect_row_roots(const evidence_write_t *w, const cbm_resolve_record_t *row, fact_root_t *roots, size_t *count) { if (row->envelope_pending) { composed_fact_t fact; @@ -846,6 +1138,63 @@ static bool collect_row_roots(evidence_write_t *w, const cbm_resolve_record_t *r .ref = (uintptr_t)row->binding_proof}; return true; } +/* Fingerprinting a composed fact hashes all its text. A chunk of rows owns a known stretch + * of roots, so the chunks need no coordination and the roots keep their row order. */ +typedef struct { + const evidence_write_t *w; + const cbm_resolve_record_t *rows; + size_t row_count; + fact_root_t *roots; + const size_t *chunk_starts; + _Atomic size_t next; + _Atomic bool failed; +} root_pass_t; + +static void root_worker(int worker, void *context) { + (void)worker; + root_pass_t *pass = context; + size_t chunks = (pass->row_count + ROOT_CHUNK_ROWS - 1) / ROOT_CHUNK_ROWS; + for (;;) { + size_t chunk = atomic_fetch_add_explicit(&pass->next, 1, memory_order_relaxed); + if (chunk >= chunks) + break; + size_t first = chunk * ROOT_CHUNK_ROWS; + size_t last = + first + ROOT_CHUNK_ROWS < pass->row_count ? first + ROOT_CHUNK_ROWS : pass->row_count; + size_t at = pass->chunk_starts[chunk]; + for (size_t i = first; i < last; i++) + if (!collect_row_roots(pass->w, &pass->rows[i], pass->roots, &at)) + atomic_store(&pass->failed, true); + } +} +static bool collect_all_row_roots(const evidence_write_t *w, const cbm_resolve_record_t *rows, + size_t count, fact_root_t *roots, size_t *root_count) { + if (!count) + return true; + size_t chunks = (count + ROOT_CHUNK_ROWS - 1) / ROOT_CHUNK_ROWS; + size_t *starts = cbm_alloc(CBM_MEM_CLASS_STORE, chunks * sizeof(*starts)); + if (!starts) + return false; + size_t at = *root_count; + for (size_t i = 0; i < count; i++) { + if (i % ROOT_CHUNK_ROWS == 0) + starts[i / ROOT_CHUNK_ROWS] = at; + at += (rows[i].envelope_pending || rows[i].result.proof) + (rows[i].binding_proof != NULL); + } + root_pass_t pass = { + .w = w, .rows = rows, .row_count = count, .roots = roots, .chunk_starts = starts}; + atomic_init(&pass.next, 0); + atomic_init(&pass.failed, false); + if (w->workers <= 1 || chunks == 1) { + root_worker(0, &pass); + } else { + cbm_parallel_for_opts_t options = {.max_workers = w->workers, .force_pthreads = false}; + cbm_parallel_for(w->workers, root_worker, &pass, options); + } + cbm_free(CBM_MEM_CLASS_STORE, starts); + *root_count = at; + return !atomic_load(&pass.failed); +} /* ── Generation reset ────────────────────────────────────────────── */ @@ -870,17 +1219,20 @@ static bool project_id(sqlite3 *db, const char *project, sqlite3_int64 *pid) { sqlite3_finalize(s); return ok; } -static bool reset_tables(sqlite3 *db, const char *project, sqlite3_int64 *pid) { +static bool reset_tables(sqlite3 *db, const char *project, sqlite3_int64 *pid, bool rows_written) { char *schema = schema_sql(); bool created = schema && sqlite3_exec(db, schema, NULL, NULL, NULL) == SQLITE_OK; cbm_free(CBM_MEM_CLASS_STORE, schema); if (!created || !project_id(db, project, pid)) return false; + /* Those rows were written for the first project of a new file. */ + if (rows_written && *pid != DIRECT_PID) + return false; const char *by_pid[] = {"DELETE FROM resolver_text WHERE pid=?", "DELETE FROM resolver_occurrence_rows WHERE pid=?", "DELETE FROM resolver_fact_rows WHERE pid=?", "DELETE FROM resolver_support_rows WHERE pid=?"}; - for (size_t i = 0; i < sizeof(by_pid) / sizeof(by_pid[0]); i++) { + for (size_t i = 0; !rows_written && i < sizeof(by_pid) / sizeof(by_pid[0]); i++) { sqlite3_stmt *s = NULL; bool ok = sqlite3_prepare_v2(db, by_pid[i], -1, &s, NULL) == SQLITE_OK && sqlite3_bind_int64(s, 1, *pid) == SQLITE_OK && sqlite3_step(s) == SQLITE_DONE; @@ -944,6 +1296,38 @@ static bool bind_occurrence(evidence_write_t *w, const cbm_resolve_record_t *row sqlite3_bind_int64(s, 23, proof) == SQLITE_OK && sqlite3_bind_int64(s, 24, binding ? (sqlite3_int64)binding->id : 0) == SQLITE_OK; } +static bool table_occurrence(evidence_write_t *w, const cbm_resolve_record_t *row, size_t ordinal, + sqlite3_int64 proof, const evidence_id_t *binding) { + cbm_writer_table_t *t = w->occurrence_table; + const cbm_resolve_result_t *r = &row->result; + int32_t line = -1; + cbm_writer_row_begin(t); + cbm_writer_row_int(t, w->pid); + cbm_writer_row_int(t, (sqlite3_int64)ordinal); + bool ok = table_text_id(t, dict_id(w, row->file_path, true)) && + table_text_id(t, dict_id(w, row->scope, true)) && + table_text_id(t, dict_id(w, row->relation, true)) && + table_text_id(t, dict_id(w, row->spelling, true)); + cbm_writer_row_int(t, row->start_byte); + cbm_writer_row_int(t, (sqlite3_int64)row->end_byte - row->start_byte); + cbm_writer_row_int(t, (int)row->source_origin); + cbm_writer_row_int(t, (int)r->binding); + cbm_writer_row_int(t, (int)r->domain); + cbm_writer_row_int(t, (int)r->analysis); + cbm_writer_row_int(t, (int)r->rule); + cbm_writer_row_int(t, r->facts); + ok = ok && table_text_id(t, dict_id(w, r->target, true)); + cbm_writer_row_int(t, (sqlite3_int64)r->candidate_count); + ok = ok && table_text_id(t, dict_id(w, r->reason, true)); + cbm_writer_row_int(t, row->emitted ? 1 : 0); + ok = ok && table_text_id(t, properties_id(w, row, &line)); + cbm_writer_row_int(t, line); + ok = ok && table_text_id(t, dict_id(w, row->content_identity, true)) && + table_text_id(t, dict_id(w, row->configuration_identity, true)); + cbm_writer_row_int(t, proof); + cbm_writer_row_int(t, binding ? (sqlite3_int64)binding->id : 0); + return table_finish(w, t, ok, "occurrences"); +} static bool write_occurrences(evidence_write_t *w, const cbm_resolve_record_t *rows, size_t count) { for (size_t i = 0; i < count; i++) { evidence_id_t *node = rows[i].envelope_pending ? NULL : find_id(w, rows[i].result.proof); @@ -959,11 +1343,13 @@ static bool write_occurrences(evidence_write_t *w, const cbm_resolve_record_t *r .source_origin = rows[i].source_origin}; /* A claimed emitted row must satisfy the same admission as its edge. * A failed metadata write aborts publication, never fabricates zero. */ - if (rows[i].proof_borrowed || - (rows[i].emitted && - (!cbm_resolve_record_admit(&rows[i]) || !proof || !source_matches(w, &source))) || - !finish_row(w, w->occurrence, bind_occurrence(w, &rows[i], i, proof, binding), - "occurrences")) + if (rows[i].proof_borrowed || (rows[i].emitted && (!cbm_resolve_record_admit(&rows[i]) || + !proof || !source_matches(w, &source)))) + return false; + if (w->occurrence_table + ? !table_occurrence(w, &rows[i], i, proof, binding) + : !finish_row(w, w->occurrence, bind_occurrence(w, &rows[i], i, proof, binding), + "occurrences")) return false; } return true; @@ -981,30 +1367,26 @@ static int compare_summaries(const void *a, const void *b) { order = left->kind < right->kind ? -1 : 1; return order; } -static bool write_summaries(evidence_write_t *w, const cbm_resolve_summary_t *const *rows, - size_t count) { +static bool write_summaries(evidence_write_t *w, const cbm_resolver_stage_t *stage) { sqlite3_stmt *s = NULL; bool ok = sqlite3_prepare_v2( w->db, "INSERT INTO resolver_derived_bindings VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", -1, &s, NULL) == SQLITE_OK; - for (size_t i = 0; ok && i < count; i++) { - const cbm_resolve_summary_t *row = rows[i]; - evidence_id_t *proof = find_id(w, row->result.proof); - ok = !row->proof_borrowed && - finish_row(w, s, - bind_text(s, 1, w->project) && bind_text(s, 2, row->file_path) && - bind_text(s, 3, row->key) && - sqlite3_bind_int64(s, 4, row->kind) == SQLITE_OK && - sqlite3_bind_int(s, 5, row->module) == SQLITE_OK && - sqlite3_bind_int(s, 6, row->closed) == SQLITE_OK && - sqlite3_bind_int(s, 7, row->result.binding) == SQLITE_OK && - sqlite3_bind_int(s, 8, row->result.domain) == SQLITE_OK && - sqlite3_bind_int(s, 9, row->result.analysis) == SQLITE_OK && - bind_text(s, 10, row->result.target) && - sqlite3_bind_int64(s, 11, proof ? (sqlite3_int64)proof->id : 0) == - SQLITE_OK && - bind_text(s, 12, row->payload_json), - "derived_bindings"); + for (size_t i = 0; ok && i < stage->summary_count; i++) { + const cbm_resolve_summary_t *row = stage->summaries[i]; + ok = finish_row( + w, s, + bind_text(s, 1, w->project) && bind_text(s, 2, row->file_path) && + bind_text(s, 3, row->key) && sqlite3_bind_int64(s, 4, row->kind) == SQLITE_OK && + sqlite3_bind_int(s, 5, row->module) == SQLITE_OK && + sqlite3_bind_int(s, 6, row->closed) == SQLITE_OK && + sqlite3_bind_int(s, 7, row->result.binding) == SQLITE_OK && + sqlite3_bind_int(s, 8, row->result.domain) == SQLITE_OK && + sqlite3_bind_int(s, 9, row->result.analysis) == SQLITE_OK && + bind_text(s, 10, row->result.target) && + sqlite3_bind_int64(s, 11, (sqlite3_int64)stage->summary_proofs[i]) == SQLITE_OK && + bind_text(s, 12, row->payload_json), + "derived_bindings"); } sqlite3_finalize(s); return ok; @@ -1036,16 +1418,28 @@ static void hash_type_states(evidence_write_t *w, cbm_sha256_ctx *hash, const ch cbm_resolve_hash_text(hash, hex); } } -static bool write_files(evidence_write_t *w, const cbm_resolve_summary_t *const *rows, - size_t count) { - sqlite3_stmt *s = NULL; - bool ok = sqlite3_prepare_v2( - w->db, "INSERT INTO resolver_file_summaries VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", -1, - &s, NULL) == SQLITE_OK; +/* Each file's digest of its derived facts and each summary's fact id: all that the two + * remaining tables take from the facts. */ +static bool decide_stage(evidence_write_t *w, cbm_resolver_stage_t *stage) { + const cbm_resolve_summary_t *const *rows = stage->summaries; + size_t count = stage->summary_count; + stage->summary_proofs = + count ? cbm_alloc(CBM_MEM_CLASS_STORE, count * sizeof(*stage->summary_proofs)) : NULL; + stage->derived = stage->file_count ? cbm_alloc(CBM_MEM_CLASS_STORE, + stage->file_count * sizeof(*stage->derived)) + : NULL; + if ((count && !stage->summary_proofs) || (stage->file_count && !stage->derived)) + return false; + for (size_t i = 0; i < count; i++) { + if (rows[i]->proof_borrowed) + return false; + const evidence_id_t *proof = find_id(w, rows[i]->result.proof); + stage->summary_proofs[i] = proof ? proof->id : 0; + } if (w->type_count > 1) qsort(w->types, w->type_count, sizeof(*w->types), compare_type_digest); size_t at = 0; - for (size_t i = 0; ok && i < w->file_count; i++) { + for (size_t i = 0; i < w->file_count; i++) { const cbm_resolve_file_t *file = w->files[i]; cbm_sha256_ctx hash; cbm_sha256_init(&hash); @@ -1070,8 +1464,17 @@ static bool write_files(evidence_write_t *w, const cbm_resolve_summary_t *const cbm_resolve_hash_text(&hash, hex); } hash_type_states(w, &hash, file->file_path); - char derived[65]; - cbm_resolve_hash_finish(&hash, derived); + cbm_resolve_hash_finish(&hash, stage->derived[i]); + } + return true; +} +static bool write_files(evidence_write_t *w, const cbm_resolver_stage_t *stage) { + sqlite3_stmt *s = NULL; + bool ok = sqlite3_prepare_v2( + w->db, "INSERT INTO resolver_file_summaries VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", -1, + &s, NULL) == SQLITE_OK; + for (size_t i = 0; ok && i < stage->file_count; i++) { + const cbm_resolve_file_t *file = stage->files[i]; ok = finish_row( w, s, bind_text(s, 1, w->project) && bind_text(s, 2, file->file_path) && @@ -1082,7 +1485,7 @@ static bool write_files(evidence_write_t *w, const cbm_resolve_summary_t *const bind_text(s, 7, file->configuration_identity) && sqlite3_bind_int64(s, 8, (sqlite3_int64)file->raw_length) == SQLITE_OK && sqlite3_bind_int64(s, 9, (sqlite3_int64)file->preprocessed_length) == SQLITE_OK && - bind_text(s, 10, file->extracted_summary) && bind_text(s, 11, derived) && + bind_text(s, 10, file->extracted_summary) && bind_text(s, 11, stage->derived[i]) && sqlite3_bind_int(s, 12, file->complete) == SQLITE_OK, "file_summaries"); } @@ -1103,29 +1506,102 @@ static bool prepare_writer(evidence_write_t *w) { "?,?,?,?,?,?,?,?,?,?,?,?,?)", -1, &w->occurrence, NULL) == SQLITE_OK; } -bool cbm_store_resolver_evidence(sqlite3 *db, const char *project, const cbm_resolve_record_t *rows, - size_t count, const cbm_resolve_file_t *files, size_t file_count, - const cbm_resolve_summary_t *summaries, size_t summary_count, - bool commit_batches) { - if (!db || !project || (!rows && count) || (!files && file_count) || +void cbm_store_resolver_stage_free(cbm_resolver_stage_t *stage) { + if (!stage) + return; + for (size_t i = 0; i < sizeof(stage->table_sql) / sizeof(stage->table_sql[0]); i++) + cbm_free(CBM_MEM_CLASS_STORE, stage->table_sql[i]); + cbm_free(CBM_MEM_CLASS_STORE, stage->summaries); + cbm_free(CBM_MEM_CLASS_STORE, stage->summary_proofs); + cbm_free(CBM_MEM_CLASS_STORE, stage->files); + cbm_free(CBM_MEM_CLASS_STORE, stage->derived); + cbm_free(CBM_MEM_CLASS_STORE, stage); +} +/* A table's CREATE as sqlite_schema keeps it: without IF NOT EXISTS or the ';'. */ +static char *table_create_sql(const char *table) { + static const char opening[] = "CREATE TABLE IF NOT EXISTS "; + size_t name_length = strlen(table); + const char *at = resolver_schema; + while ((at = strstr(at, opening)) != NULL) { + const char *name = at + sizeof(opening) - 1; + if (!strncmp(name, table, name_length) && name[name_length] == '(') { + const char *end = strchr(name, ';'); + if (!end) + return NULL; + size_t length = (size_t)(end - name); + char *out = cbm_alloc(CBM_MEM_CLASS_STORE, sizeof("CREATE TABLE ") + length); + if (out) { + memcpy(out, "CREATE TABLE ", sizeof("CREATE TABLE ") - 1); + memcpy(out + sizeof("CREATE TABLE ") - 1, name, length); + out[sizeof("CREATE TABLE ") - 1 + length] = '\0'; + } + return out; + } + at = name; + } + return NULL; +} +static bool begin_tables(evidence_write_t *w, cbm_db_writer_t *direct, + cbm_resolver_stage_t *stage) { + static const char *const names[] = {"resolver_text", "resolver_fact_rows", + "resolver_support_rows", "resolver_occurrence_rows"}; + cbm_writer_table_t **tables[] = {&w->text_table, &w->fact_table, &w->support_table, + &w->occurrence_table}; + for (size_t i = 0; i < sizeof(names) / sizeof(names[0]); i++) { + stage->table_sql[i] = table_create_sql(names[i]); + if (!stage->table_sql[i]) + return false; + *tables[i] = cbm_writer_table_begin(direct, names[i], stage->table_sql[i]); + if (!*tables[i]) + return false; + } + return true; +} +static bool end_tables(evidence_write_t *w) { + cbm_writer_table_t **tables[] = {&w->text_table, &w->fact_table, &w->support_table, + &w->occurrence_table}; + bool ok = true; + for (size_t i = 0; i < sizeof(tables) / sizeof(tables[0]); i++) { + if (*tables[i]) + ok = cbm_writer_table_end(*tables[i]) && ok; + *tables[i] = NULL; + } + return ok; +} +/* One write, with its rows going to the statements on db or to direct's pages. In the + * second case db is NULL, and *out takes what the store still has to write. */ +static bool evidence_run(sqlite3 *db, cbm_db_writer_t *direct, const char *project, + const cbm_resolve_record_t *rows, size_t count, + const cbm_resolve_file_t *files, size_t file_count, + const cbm_resolve_summary_t *summaries, size_t summary_count, + bool commit_batches, int workers, cbm_resolver_stage_t **out) { + if ((!db && !direct) || !project || (!rows && count) || (!files && file_count) || (!summaries && summary_count) || count > (SIZE_MAX - summary_count) / 2 || file_count > SIZE_MAX / sizeof(void *) || summary_count > SIZE_MAX / sizeof(void *)) return false; size_t root_capacity = summary_count; for (size_t i = 0; i < count; i++) - root_capacity += (rows[i].result.proof != NULL) + (rows[i].binding_proof != NULL); + root_capacity += + (rows[i].envelope_pending || rows[i].result.proof) + (rows[i].binding_proof != NULL); if (root_capacity > SIZE_MAX / sizeof(fact_root_t)) return false; fact_root_t *roots = root_capacity ? cbm_alloc(CBM_MEM_CLASS_STORE, root_capacity * sizeof(*roots)) : NULL; + cbm_resolver_stage_t *stage = cbm_calloc(CBM_MEM_CLASS_STORE, sizeof(*stage)); const cbm_resolve_file_t **ordered_files = file_count ? cbm_alloc(CBM_MEM_CLASS_STORE, file_count * sizeof(*ordered_files)) : NULL; const cbm_resolve_summary_t **ordered_summaries = summary_count ? cbm_alloc(CBM_MEM_CLASS_STORE, summary_count * sizeof(*ordered_summaries)) : NULL; dict_cache_t *cache = cbm_calloc(CBM_MEM_CLASS_STORE, DICT_CACHE_SLOTS * sizeof(*cache)); - bool ok = (!root_capacity || roots) && (!file_count || ordered_files) && + bool ok = (!root_capacity || roots) && stage && (!file_count || ordered_files) && (!summary_count || ordered_summaries) && cache; + if (stage) { + stage->summaries = ordered_summaries; + stage->summary_count = summary_count; + stage->files = ordered_files; + stage->file_count = file_count; + } size_t root_count = 0; for (size_t i = 0; ok && i < summary_count; i++) { ordered_summaries[i] = &summaries[i]; @@ -1143,8 +1619,12 @@ bool cbm_store_resolver_evidence(sqlite3 *db, const char *project, const cbm_res evidence_write_t *writer = cbm_calloc(CBM_MEM_CLASS_STORE, sizeof(*writer)); if (!writer) { cbm_free(CBM_MEM_CLASS_STORE, roots); - cbm_free(CBM_MEM_CLASS_STORE, ordered_files); - cbm_free(CBM_MEM_CLASS_STORE, ordered_summaries); + if (stage) { + cbm_store_resolver_stage_free(stage); + } else { + cbm_free(CBM_MEM_CLASS_STORE, ordered_files); + cbm_free(CBM_MEM_CLASS_STORE, ordered_summaries); + } cbm_free(CBM_MEM_CLASS_STORE, cache); return false; } @@ -1154,21 +1634,29 @@ bool cbm_store_resolver_evidence(sqlite3 *db, const char *project, const cbm_res w->files = ordered_files; w->file_count = file_count; w->cache = cache; + w->valid_text = cbm_calloc(CBM_MEM_CLASS_STORE, DICT_CACHE_SLOTS * sizeof(*w->valid_text)); + ok = ok && w->valid_text; w->commit_batches = commit_batches; + w->workers = workers; cbm_arena_init(&w->owned); w->row_facts = count ? cbm_calloc(CBM_MEM_CLASS_STORE, count * sizeof(*w->row_facts)) : NULL; #if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS held_digests = 0; #endif ok = ok && (!count || w->row_facts); - for (size_t i = 0; ok && i < count; i++) - ok = collect_row_roots(w, &rows[i], roots, &root_count); - ok = ok && reset_tables(db, project, &w->pid) && prepare_writer(w) && - write_facts(w, roots, root_count, rows); + ok = ok && collect_all_row_roots(w, rows, count, roots, &root_count); + if (direct) { + w->pid = DIRECT_PID; + ok = ok && begin_tables(w, direct, stage); + } else { + ok = ok && reset_tables(db, project, &w->pid, false) && prepare_writer(w); + } + ok = ok && write_facts(w, roots, root_count, rows); cbm_free(CBM_MEM_CLASS_STORE, roots); - ok = ok && write_occurrences(w, rows, count) && - write_summaries(w, ordered_summaries, summary_count) && - write_files(w, ordered_summaries, summary_count); + ok = ok && write_occurrences(w, rows, count) && decide_stage(w, stage); + ok = end_tables(w) && ok; + if (!direct) + ok = ok && write_summaries(w, stage) && write_files(w, stage); sqlite3_finalize(w->text); sqlite3_finalize(w->fact); sqlite3_finalize(w->support); @@ -1180,12 +1668,48 @@ bool cbm_store_resolver_evidence(sqlite3 *db, const char *project, const cbm_res cbm_free(CBM_MEM_CLASS_STORE, w->digests); cbm_free(CBM_MEM_CLASS_STORE, w->dict); cbm_free(CBM_MEM_CLASS_STORE, w->types); - cbm_free(CBM_MEM_CLASS_STORE, w->fact_text.bytes); + cbm_free(CBM_MEM_CLASS_STORE, w->order); + cbm_free(CBM_MEM_CLASS_STORE, w->valid_text); cbm_free(CBM_MEM_CLASS_STORE, w->row_facts); cbm_free(CBM_MEM_CLASS_STORE, cache); cbm_arena_destroy(&w->owned); cbm_free(CBM_MEM_CLASS_STORE, writer); - cbm_free(CBM_MEM_CLASS_STORE, ordered_files); - cbm_free(CBM_MEM_CLASS_STORE, ordered_summaries); + if (ok && out) + *out = stage; + else + cbm_store_resolver_stage_free(stage); + return ok; +} +bool cbm_store_resolver_evidence(sqlite3 *db, const char *project, const cbm_resolve_record_t *rows, + size_t count, const cbm_resolve_file_t *files, size_t file_count, + const cbm_resolve_summary_t *summaries, size_t summary_count, + bool commit_batches, int workers) { + return db && evidence_run(db, NULL, project, rows, count, files, file_count, summaries, + summary_count, commit_batches, workers, NULL); +} +cbm_resolver_stage_t *cbm_store_resolver_stage_write(cbm_db_writer_t *writer, const char *project, + const cbm_resolve_record_t *rows, size_t count, + const cbm_resolve_file_t *files, + size_t file_count, + const cbm_resolve_summary_t *summaries, + size_t summary_count, int workers) { + cbm_resolver_stage_t *stage = NULL; + return writer && evidence_run(NULL, writer, project, rows, count, files, file_count, summaries, + summary_count, false, workers, &stage) + ? stage + : NULL; +} +bool cbm_store_resolver_stage_finish(sqlite3 *db, const char *project, + const cbm_resolver_stage_t *stage) { + if (!db || !project || !stage) + return false; + evidence_write_t *w = cbm_calloc(CBM_MEM_CLASS_STORE, sizeof(*w)); + if (!w) + return false; + w->db = db; + w->project = project; + bool ok = reset_tables(db, project, &w->pid, true) && write_summaries(w, stage) && + write_files(w, stage); + cbm_free(CBM_MEM_CLASS_STORE, w); return ok; } diff --git a/src/store/store_resolver_evidence.h b/src/store/store_resolver_evidence.h index 49ff2404b..5e317a0c3 100644 --- a/src/store/store_resolver_evidence.h +++ b/src/store/store_resolver_evidence.h @@ -1,6 +1,7 @@ #ifndef CBM_STORE_RESOLVER_EVIDENCE_H #define CBM_STORE_RESOLVER_EVIDENCE_H #include "resolve/record.h" +#include "sqlite_writer.h" #include /* Called inside the writer's savepoint. All roots must be owned by one canonical * generation DAG; the caller's graph stays alive through write. With commit_batches @@ -10,7 +11,24 @@ bool cbm_store_resolver_evidence(sqlite3 *db, const char *project, const cbm_resolve_record_t *rows, size_t count, const cbm_resolve_file_t *files, size_t file_count, const cbm_resolve_summary_t *summaries, size_t summary_count, - bool commit_batches); + bool commit_batches, int workers); +/* The same write when the store is a new file still being built: facts, supports, + * occurrences and their text go straight into writer's pages, tens of millions of rows + * that SQL would take one at a time. The tables are ordinary and the views over them + * read back the same values. NULL on failure, which also fails the file. What is + * returned holds no fact and no digest, only what the store then inserts, and it + * borrows the same rows as this call until freed. */ +typedef struct cbm_resolver_stage cbm_resolver_stage_t; +cbm_resolver_stage_t *cbm_store_resolver_stage_write(cbm_db_writer_t *writer, const char *project, + const cbm_resolve_record_t *rows, size_t count, + const cbm_resolve_file_t *files, + size_t file_count, + const cbm_resolve_summary_t *summaries, + size_t summary_count, int workers); +/* The rest of the write, on the finished file. */ +bool cbm_store_resolver_stage_finish(sqlite3 *db, const char *project, + const cbm_resolver_stage_t *stage); +void cbm_store_resolver_stage_free(cbm_resolver_stage_t *stage); /* True when sql, as stored in sqlite_schema, is exactly how the writer defines this * logical view. A view of another name or definition is not this generation's data. */ bool cbm_store_resolver_view_definition_current(const char *sql); diff --git a/tests/test_resolver_generation.c b/tests/test_resolver_generation.c index 1565a9371..eec8bace9 100644 --- a/tests/test_resolver_generation.c +++ b/tests/test_resolver_generation.c @@ -1239,7 +1239,7 @@ TEST(resolver_generation_publish_keeps_the_wal_small) { ASSERT_EQ(sqlite3_exec(cbm_store_get_db(store), "PRAGMA wal_autocheckpoint=8", NULL, NULL, NULL), SQLITE_OK); cbm_store_resolver_test_set_progress_rows(256); - int rc = cbm_store_resolver_publish_generation(store, "proj", rows, ROWS, NULL, 0, NULL, 0); + int rc = cbm_store_resolver_publish_generation(store, "proj", rows, ROWS, NULL, 0, NULL, 0, 1); cbm_store_resolver_test_set_progress_rows(0); bool present = false; int current = cbm_store_resolver_current(store, "proj", &present); diff --git a/tests/test_sqlite_writer.c b/tests/test_sqlite_writer.c index cbeb2dfc3..c0c6ecf52 100644 --- a/tests/test_sqlite_writer.c +++ b/tests/test_sqlite_writer.c @@ -1063,7 +1063,124 @@ TEST(sw_publish_failure_reports_the_rename_not_the_cleanup) { PASS(); } +static bool sw_scalar(sqlite3 *db, const char *sql, long long *out) { + sqlite3_stmt *stmt = NULL; + bool ok = sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) == SQLITE_OK && + sqlite3_step(stmt) == SQLITE_ROW; + if (ok) + *out = sqlite3_column_int64(stmt, 0); + sqlite3_finalize(stmt); + return ok; +} + +/* Streamed tables interleave their pages and take every value width, NULL, a blob, text past + * the local payload limit (the writer spills it) and enough rows for interior pages. Tables + * begun before nodes share the file with them, and SQLite must find every row by key. */ +TEST(sw_streamed_tables_read_back) { + enum { ROWS = 200000, HUGE_ROW = 7, HUGE_TEXT = 200000 }; + char path[256]; + ASSERT_EQ(make_temp_db(path, sizeof(path)), 0); + cbm_db_writer_t *w = cbm_writer_open(path); + ASSERT_NOT_NULL(w); + static const char first_sql[] = + "CREATE TABLE stream_first(k INTEGER NOT NULL,o INTEGER NOT NULL,v TEXT,b BLOB,n INTEGER," + "PRIMARY KEY(k,o)) WITHOUT ROWID"; + static const char second_sql[] = + "CREATE TABLE stream_second(k INTEGER PRIMARY KEY,v INTEGER NOT NULL) WITHOUT ROWID"; + static const char empty_sql[] = + "CREATE TABLE stream_empty(k INTEGER PRIMARY KEY) WITHOUT ROWID"; + cbm_writer_table_t *first = cbm_writer_table_begin(w, "stream_first", first_sql); + cbm_writer_table_t *second = cbm_writer_table_begin(w, "stream_second", second_sql); + cbm_writer_table_t *empty = cbm_writer_table_begin(w, "stream_empty", empty_sql); + ASSERT_NOT_NULL(first); + ASSERT_NOT_NULL(second); + ASSERT_NOT_NULL(empty); + char *huge = malloc(HUGE_TEXT + 1); + ASSERT_NOT_NULL(huge); + memset(huge, 'x', HUGE_TEXT); + huge[HUGE_TEXT] = '\0'; + bool written = true; + for (int i = 0; written && i < ROWS; i++) { + char small_text[32]; + snprintf(small_text, sizeof(small_text), "row_%d", i); + unsigned char digest[32] = {(unsigned char)i}; + cbm_writer_row_begin(first); + cbm_writer_row_int(first, 1); + cbm_writer_row_int(first, i); + cbm_writer_row_text(first, i == HUGE_ROW ? huge : small_text); + cbm_writer_row_blob(first, digest, sizeof(digest)); + if (i % 3) + cbm_writer_row_int(first, (long long)i * (i % 2 ? 100000 : -1)); + else + cbm_writer_row_null(first); + written = cbm_writer_row_end(first); + cbm_writer_row_begin(second); + cbm_writer_row_int(second, i); + cbm_writer_row_int(second, i % 2); + written = written && cbm_writer_row_end(second); + } + free(huge); + ASSERT_TRUE(written); + ASSERT_TRUE(cbm_writer_table_end(first)); + ASSERT_TRUE(cbm_writer_table_end(second)); + ASSERT_TRUE(cbm_writer_table_end(empty)); + CBMDumpNode node = {.id = 1, + .project = "test", + .label = "Function", + .name = "n", + .qualified_name = "test.n", + .file_path = "f.c", + .start_line = 1, + .end_line = 2, + .properties = "{}"}; + ASSERT_EQ(cbm_writer_append_nodes(w, &node, 1), 0); + ASSERT_EQ(cbm_writer_finalize(w, "test", "/tmp/test", "2026-07-07T00:00:00Z", &node, 1, NULL, 0, + NULL, 0, NULL, 0), + 0); + + sqlite3 *db = NULL; + ASSERT_EQ(sqlite3_open(path, &db), SQLITE_OK); + sqlite3_stmt *stmt = NULL; + ASSERT_EQ(sqlite3_prepare_v2(db, "PRAGMA integrity_check", -1, &stmt, NULL), SQLITE_OK); + ASSERT_EQ(sqlite3_step(stmt), SQLITE_ROW); + ASSERT_STR_EQ((const char *)sqlite3_column_text(stmt, 0), "ok"); + sqlite3_finalize(stmt); + long long value = 0; + ASSERT_TRUE(sw_scalar(db, "SELECT count(*) FROM stream_first", &value)); + ASSERT_EQ(value, ROWS); + ASSERT_TRUE(sw_scalar(db, "SELECT count(*) FROM stream_second", &value)); + ASSERT_EQ(value, ROWS); + ASSERT_TRUE(sw_scalar(db, "SELECT count(*) FROM stream_empty", &value)); + ASSERT_EQ(value, 0); + ASSERT_TRUE(sw_scalar(db, "SELECT count(*) FROM nodes", &value)); + ASSERT_EQ(value, 1); + ASSERT_TRUE(sw_scalar(db, "SELECT length(v) FROM stream_first WHERE k=1 AND o=7", &value)); + ASSERT_EQ(value, HUGE_TEXT); + ASSERT_TRUE(sw_scalar(db, "SELECT n FROM stream_first WHERE k=1 AND o=199999", &value)); + ASSERT_EQ(value, 199999LL * 100000); + ASSERT_TRUE(sw_scalar(db, "SELECT n FROM stream_first WHERE k=1 AND o=100000", &value)); + ASSERT_EQ(value, -100000); + ASSERT_TRUE(sw_scalar(db, "SELECT count(*) FROM stream_first WHERE k=1 AND o=99999 AND n IS NULL", + &value)); + ASSERT_EQ(value, 1); + ASSERT_TRUE(sw_scalar(db, "SELECT length(b) FROM stream_first WHERE k=1 AND o=5", &value)); + ASSERT_EQ(value, 32); + ASSERT_TRUE(sw_scalar(db, "SELECT v FROM stream_second WHERE k=123457", &value)); + ASSERT_EQ(value, 1); + /* Ordinary tables: SQLite can go on writing to them. */ + ASSERT_EQ(sqlite3_exec(db, "INSERT INTO stream_empty VALUES(5);" + "INSERT INTO stream_second VALUES(9999999,1);" + "DELETE FROM stream_second WHERE k<100;" + "PRAGMA integrity_check;", + NULL, NULL, NULL), + SQLITE_OK); + sqlite3_close(db); + cbm_unlink(path); + PASS(); +} + SUITE(sqlite_writer) { + RUN_TEST(sw_streamed_tables_read_back); RUN_TEST(sw_minimal_data); RUN_TEST(sw_imports_local_name_unique); RUN_TEST(sw_scale_and_indexes);