feat(db2-migrate): the transcode — old log to new shape, record by record
- wo_wal_migrate rewrites the log without touching db state: no id maps, no indexes, no keys-resident logic — the new log replays through the machinery that already exists and is already tested - cids remap by name, INCLUDING the ones embedded inside stored owned values (an owned value carries a cid on the wire); the embed closure guarantees every nested class is shape-unchanged, so only numbers move - surviving fields go to their new slot, deleted fields' values are freed, added fields take the kind's zero value straight from enc_val(0) - a delta on a deleted field is SPLICED out: an offset map (old record start -> new) rewrites every back pointer, and the dropped delta maps to its own target so later deltas step over it - temp + fsync + rename, compaction's own crash discipline; a stale temp is discarded at start; a torn tail bounds the intact prefix exactly as replay does - fixed en route: early `goto corrupt` jumped over initializers, so the handler freed uninitialized memory — declarations hoisted above the first jump - six end-to-end tests: add, delete (ASan watches the freed Text), reorder with owned fixup, delta splice on a keys-resident chain, poison-bites-only-with-records, corrupt input - test_wal 5951 pass, 0 fail Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> (cherry picked from commit b69092a99206e1dcdf6f2dcdb02146939b0564c6)
This commit is contained in:
parent
df09158b7f
commit
c6e158c6a1
3 changed files with 727 additions and 1 deletions
|
|
@ -1293,6 +1293,298 @@ fail:
|
|||
* update would have. The caller (wo_wal_replay_ex) still owns dropping the
|
||||
* recreated row's payload back to the log, the same way it does for
|
||||
* INSERT/UPDATE. */
|
||||
/* ---- databasev2 12: the migration transcode ---------------------------- */
|
||||
|
||||
/* record-start offset -> record-start offset in the new log. Open-addressed;
|
||||
* keys stored +1 so offset 0 (a legacy log's first record) is representable. */
|
||||
typedef struct migmap {
|
||||
uint64_t *k, *v;
|
||||
size_t cap, len;
|
||||
} migmap;
|
||||
static int migmap_put(migmap *m, uint64_t key, uint64_t val) {
|
||||
if (m->len * 2 >= m->cap) {
|
||||
size_t nc = m->cap ? m->cap * 2 : 1024;
|
||||
uint64_t *nk = calloc(nc, sizeof *nk);
|
||||
uint64_t *nv = malloc(nc * sizeof *nv);
|
||||
if (!nk || !nv) {
|
||||
free(nk);
|
||||
free(nv);
|
||||
return -1;
|
||||
}
|
||||
for (size_t i = 0; i < m->cap; i++) {
|
||||
if (!m->k[i]) continue;
|
||||
size_t j = (size_t)(m->k[i] * 0x9E3779B97F4A7C15ull) & (nc - 1);
|
||||
while (nk[j]) j = (j + 1) & (nc - 1);
|
||||
nk[j] = m->k[i];
|
||||
nv[j] = m->v[i];
|
||||
}
|
||||
free(m->k);
|
||||
free(m->v);
|
||||
m->k = nk;
|
||||
m->v = nv;
|
||||
m->cap = nc;
|
||||
}
|
||||
size_t j = (size_t)((key + 1) * 0x9E3779B97F4A7C15ull) & (m->cap - 1);
|
||||
while (m->k[j] && m->k[j] != key + 1) j = (j + 1) & (m->cap - 1);
|
||||
if (!m->k[j]) m->len++;
|
||||
m->k[j] = key + 1;
|
||||
m->v[j] = val;
|
||||
return 0;
|
||||
}
|
||||
static int migmap_get(const migmap *m, uint64_t key, uint64_t *val) {
|
||||
if (!m->cap) return -1;
|
||||
size_t j = (size_t)((key + 1) * 0x9E3779B97F4A7C15ull) & (m->cap - 1);
|
||||
while (m->k[j]) {
|
||||
if (m->k[j] == key + 1) {
|
||||
*val = m->v[j];
|
||||
return 0;
|
||||
}
|
||||
j = (j + 1) & (m->cap - 1);
|
||||
}
|
||||
return -1;
|
||||
}
|
||||
|
||||
/* remap the class ids EMBEDDED in a decoded value tree (owned values carry a
|
||||
* cid on the wire). The tree's classes are all UNCHANGED — the embed closure
|
||||
* poisoned anything else before a record could reach here — so only the
|
||||
* NUMBERS move; shapes and kinds are the same on both sides. */
|
||||
static int mig_fixup_cids(uint64_t v, uint8_t kind, const wo_schema *oldsc,
|
||||
const wo_mig_plan *plan) {
|
||||
if (!v) return 0;
|
||||
switch (kind) {
|
||||
case WO_K_OWNED: {
|
||||
db_rec *rec = (db_rec *)(uintptr_t)v;
|
||||
uint32_t oc = rec->class_id;
|
||||
if (oc >= plan->old_class_cnt || plan->classes[oc].new_cid == WO_SCHEMA_NONE)
|
||||
return -1;
|
||||
const wo_schema_class *k = &oldsc->classes[oc];
|
||||
for (uint32_t i = 0; i < k->field_cnt; i++)
|
||||
if (mig_fixup_cids(rec->slots[i], k->fields[i].kind, oldsc, plan) != 0)
|
||||
return -1;
|
||||
rec->class_id = plan->classes[oc].new_cid;
|
||||
return 0;
|
||||
}
|
||||
case WO_K_MULTI: {
|
||||
db_multi *m = (db_multi *)(uintptr_t)v;
|
||||
if (m->elem_kind != WO_K_OWNED && m->elem_kind != WO_K_MULTI &&
|
||||
m->elem_kind != WO_K_MAP)
|
||||
return 0;
|
||||
for (uint32_t i = 0; i < m->len; i++)
|
||||
if (mig_fixup_cids(m->items[i], m->elem_kind, oldsc, plan) != 0) return -1;
|
||||
return 0;
|
||||
}
|
||||
case WO_K_MAP: {
|
||||
db_map *m = (db_map *)(uintptr_t)v;
|
||||
for (uint32_t i = 0; i < m->len; i++) {
|
||||
if (mig_fixup_cids(m->kv[2 * i], m->key_kind, oldsc, plan) != 0) return -1;
|
||||
if (mig_fixup_cids(m->kv[2 * i + 1], m->val_kind, oldsc, plan) != 0)
|
||||
return -1;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
default: return 0;
|
||||
}
|
||||
}
|
||||
|
||||
int wo_wal_migrate(const char *path, wo_db *db, const wo_schema *oldsc,
|
||||
const wo_mig_plan *plan, const wo_schema *newsc,
|
||||
uint64_t prealloc, char **err_out) {
|
||||
if (err_out) *err_out = NULL;
|
||||
int rc = -1;
|
||||
int ofd = open(path, O_RDONLY);
|
||||
if (ofd < 0) return -1;
|
||||
|
||||
/* decode with the OLD shapes: a classdesc shim per stored class, kinds
|
||||
* lifted from the schema record. dec_val touches nothing in the db but
|
||||
* classes/class_cnt, so a stack shim is the whole "old database". */
|
||||
wo_db olddb;
|
||||
memset(&olddb, 0, sizeof olddb);
|
||||
wo_classdesc *oldcls = calloc(oldsc->class_cnt ? oldsc->class_cnt : 1,
|
||||
sizeof *oldcls);
|
||||
uint8_t **oldkinds = calloc(oldsc->class_cnt ? oldsc->class_cnt : 1,
|
||||
sizeof *oldkinds);
|
||||
if (!oldcls || !oldkinds) goto out_nolog;
|
||||
for (uint32_t c = 0; c < oldsc->class_cnt; c++) {
|
||||
const wo_schema_class *k = &oldsc->classes[c];
|
||||
oldkinds[c] = malloc(k->field_cnt ? k->field_cnt : 1);
|
||||
if (!oldkinds[c]) goto out_nolog;
|
||||
for (uint32_t f = 0; f < k->field_cnt; f++) oldkinds[c][f] = k->fields[f].kind;
|
||||
oldcls[c].field_cnt = k->field_cnt;
|
||||
oldcls[c].kinds = oldkinds[c];
|
||||
oldcls[c].flags = k->flags;
|
||||
}
|
||||
olddb.classes = oldcls;
|
||||
olddb.class_cnt = oldsc->class_cnt;
|
||||
|
||||
char tmp[4096];
|
||||
if ((size_t)snprintf(tmp, sizeof tmp, "%s%s", path, WO_WAL_TMP_SUFFIX) >= sizeof tmp)
|
||||
goto out_nolog;
|
||||
(void)unlink(tmp);
|
||||
wo_wal nw;
|
||||
if (wo_wal_open(&nw, tmp, prealloc) != 0) goto out_nolog;
|
||||
|
||||
migmap map = {0};
|
||||
uint64_t *vals = NULL;
|
||||
|
||||
/* the new log's first record: the compiled shape */
|
||||
{
|
||||
uint8_t *sp;
|
||||
uint32_t slen;
|
||||
if (wo_schema_encode(newsc, &sp, &slen) != 0) goto out;
|
||||
wbuf b = {0};
|
||||
wput(&b, sp, slen);
|
||||
free(sp);
|
||||
int src = stage(&nw, &b);
|
||||
free(b.b);
|
||||
if (src != 0) goto out;
|
||||
}
|
||||
|
||||
uint32_t pending = 0;
|
||||
for (uint64_t off = 0;;) {
|
||||
uint32_t len;
|
||||
uint8_t *payload;
|
||||
if (scan_record(ofd, off, &len, &payload) != 0) break; /* intact prefix ends */
|
||||
rbuf r = {payload, payload + len, 0};
|
||||
uint8_t kind = rd_u8(&r);
|
||||
uint64_t new_start = nw.off + (uint64_t)nw.len;
|
||||
/* declared before any `goto corrupt` can fire, so the handler never
|
||||
* frees an object whose initializer was jumped over */
|
||||
wbuf b = {0};
|
||||
const wo_schema_class *ok = NULL;
|
||||
const wo_classdesc *nk = NULL;
|
||||
if (kind == WO_WAL_SCHEMA) {
|
||||
free(payload);
|
||||
off += 8u + len + 4u;
|
||||
continue; /* replaced by the head record above */
|
||||
}
|
||||
uint32_t cid = rd_u32(&r);
|
||||
uint64_t id = rd_u64(&r);
|
||||
if (r.bad || cid >= plan->old_class_cnt) goto corrupt;
|
||||
const wo_mig_class *mc = &plan->classes[cid];
|
||||
if (mc->new_cid == WO_SCHEMA_NONE) {
|
||||
/* the poison bites: a record of this class actually exists */
|
||||
if (err_out && mc->poison) *err_out = strdup(mc->poison);
|
||||
rc = -2;
|
||||
free(payload);
|
||||
goto out;
|
||||
}
|
||||
ok = &oldsc->classes[cid];
|
||||
nk = &db->classes[mc->new_cid];
|
||||
if (kind == WO_WAL_INSERT || kind == WO_WAL_UPDATE) {
|
||||
vals = calloc(ok->field_cnt ? ok->field_cnt : 1, sizeof *vals);
|
||||
if (!vals) goto corrupt;
|
||||
for (uint32_t f = 0; f < ok->field_cnt; f++)
|
||||
if (dec_val(&r, &olddb, ok->fields[f].kind, &vals[f]) != 0) goto corrupt;
|
||||
if ((size_t)(r.end - r.p) != 0) goto corrupt;
|
||||
for (uint32_t f = 0; f < ok->field_cnt; f++)
|
||||
if (mig_fixup_cids(vals[f], ok->fields[f].kind, oldsc, plan) != 0)
|
||||
goto corrupt;
|
||||
wput_u8(&b, kind);
|
||||
wput_u32(&b, mc->new_cid);
|
||||
wput_u64(&b, id);
|
||||
for (uint32_t nf = 0; nf < nk->field_cnt; nf++) {
|
||||
uint64_t v = 0; /* an added field: the kind's zero value */
|
||||
for (uint32_t f = 0; f < ok->field_cnt; f++)
|
||||
if (mc->fmap[f] == (int32_t)nf) {
|
||||
v = vals[f];
|
||||
break;
|
||||
}
|
||||
enc_val(&b, db->classes, nk->kinds[nf], v);
|
||||
}
|
||||
/* post-fixup the tree speaks NEW cids, so the compiled db frees it */
|
||||
for (uint32_t f = 0; f < ok->field_cnt; f++)
|
||||
wo_db_val_free(db, ok->fields[f].kind, vals[f]);
|
||||
free(vals);
|
||||
vals = NULL;
|
||||
} else if (kind == WO_WAL_DELTA) {
|
||||
uint32_t fidx = rd_u32(&r);
|
||||
uint64_t back = rd_u64(&r);
|
||||
if (r.bad || fidx >= ok->field_cnt) goto corrupt;
|
||||
uint64_t nback;
|
||||
if (migmap_get(&map, back, &nback) != 0) goto corrupt;
|
||||
if (mc->fmap[fidx] < 0) {
|
||||
/* a delta on a deleted field: splice the chain around it —
|
||||
* anything pointing at THIS record re-points to where this
|
||||
* record itself pointed */
|
||||
free(payload);
|
||||
if (migmap_put(&map, off, nback) != 0) goto out;
|
||||
off += 8u + len + 4u;
|
||||
continue;
|
||||
}
|
||||
uint64_t v = 0;
|
||||
if (dec_val(&r, &olddb, ok->fields[fidx].kind, &v) != 0) goto corrupt;
|
||||
if ((size_t)(r.end - r.p) != 0 ||
|
||||
mig_fixup_cids(v, ok->fields[fidx].kind, oldsc, plan) != 0) {
|
||||
wo_db_val_free(db, ok->fields[fidx].kind, v);
|
||||
goto corrupt;
|
||||
}
|
||||
wput_u8(&b, WO_WAL_DELTA);
|
||||
wput_u32(&b, mc->new_cid);
|
||||
wput_u64(&b, id);
|
||||
wput_u32(&b, (uint32_t)mc->fmap[fidx]);
|
||||
wput_u64(&b, nback);
|
||||
enc_val(&b, db->classes, nk->kinds[mc->fmap[fidx]], v);
|
||||
wo_db_val_free(db, ok->fields[fidx].kind, v);
|
||||
} else if (kind == WO_WAL_REMOVE) {
|
||||
wput_u8(&b, WO_WAL_REMOVE);
|
||||
wput_u32(&b, mc->new_cid);
|
||||
wput_u64(&b, id);
|
||||
} else {
|
||||
goto corrupt;
|
||||
}
|
||||
free(payload);
|
||||
payload = NULL;
|
||||
{
|
||||
int src = stage(&nw, &b);
|
||||
free(b.b);
|
||||
if (src != 0) goto out;
|
||||
}
|
||||
if (migmap_put(&map, off, new_start) != 0) goto out;
|
||||
if (++pending >= WO_WAL_COMPACT_FLUSH) {
|
||||
if (wal_write_nosync(&nw) != 0) goto out;
|
||||
pending = 0;
|
||||
}
|
||||
off += 8u + len + 4u;
|
||||
continue;
|
||||
corrupt:
|
||||
free(payload);
|
||||
free(b.b);
|
||||
if (vals && ok) {
|
||||
for (uint32_t f = 0; f < ok->field_cnt; f++)
|
||||
wo_db_val_free(db, ok->fields[f].kind, vals[f]);
|
||||
free(vals);
|
||||
vals = NULL;
|
||||
}
|
||||
rc = -1;
|
||||
goto out;
|
||||
}
|
||||
|
||||
if (wal_write_nosync(&nw) != 0) goto out;
|
||||
if (fsync(nw.fd) != 0) goto out;
|
||||
wo_wal_close(&nw);
|
||||
if (rename(tmp, path) != 0) {
|
||||
(void)unlink(tmp);
|
||||
goto out_closed;
|
||||
}
|
||||
sync_parent_dir(path);
|
||||
rc = 0;
|
||||
goto out_closed;
|
||||
|
||||
out:
|
||||
wo_wal_close(&nw);
|
||||
(void)unlink(tmp);
|
||||
out_closed:
|
||||
free(map.k);
|
||||
free(map.v);
|
||||
out_nolog:
|
||||
if (oldkinds)
|
||||
for (uint32_t c = 0; c < oldsc->class_cnt; c++) free(oldkinds[c]);
|
||||
free(oldkinds);
|
||||
free(oldcls);
|
||||
close(ofd);
|
||||
return rc;
|
||||
}
|
||||
|
||||
static int apply_delta(wo_db *db, uint32_t cid, uint64_t id, rbuf *r) {
|
||||
uint32_t field_idx = rd_u32(r);
|
||||
uint64_t back_off = rd_u64(r);
|
||||
|
|
|
|||
|
|
@ -192,6 +192,21 @@ typedef struct wo_mig_plan {
|
|||
int wo_schema_diff(const wo_schema *oldsc, const wo_schema *newsc, wo_mig_plan *plan);
|
||||
void wo_mig_plan_free(wo_mig_plan *plan);
|
||||
|
||||
/* databasev2 12: rewrite the log at `path` from its stored shape to the
|
||||
* compiled one — a record-level transcode, no db state touched: cids remap by
|
||||
* name (embedded owned values included), surviving fields move to their new
|
||||
* slot, deleted fields' values are freed, added fields take the kind's zero
|
||||
* value, and a delta chain whose field vanished is spliced around. The new
|
||||
* log is written the way compaction writes one (temp, fsync, rename), so a
|
||||
* crash anywhere leaves the old log intact and the next boot re-migrates.
|
||||
* `db` supplies the COMPILED classes for encoding; nothing is inserted.
|
||||
* Returns 0 on success, -1 on I/O or corruption, -2 when a record of a
|
||||
* poisoned class was met — *err_out (malloc'd, caller frees) then carries the
|
||||
* poison text. */
|
||||
int wo_wal_migrate(const char *path, wo_db *db, const wo_schema *oldsc,
|
||||
const wo_mig_plan *plan, const wo_schema *newsc,
|
||||
uint64_t prealloc, char **err_out);
|
||||
|
||||
/* Adopt `sc` as this log's compiled schema (encoded and owned by the wal). */
|
||||
int wo_wal_set_schema(wo_wal *w, const wo_schema *sc);
|
||||
/* A fresh, empty log gets the schema as its first record — durable before
|
||||
|
|
|
|||
|
|
@ -1008,13 +1008,426 @@ static void test_should_compact_absolute_and_ceiling(void) {
|
|||
T_EQ(wo_wal_should_compact(2048, 1024, floor_b, 2), 0); /* not yet */
|
||||
}
|
||||
|
||||
/* ---- databasev2 12: the boot diff --------------------------------------- */
|
||||
/* ---- databasev2 12: the migration transcode ----------------------------- */
|
||||
|
||||
#define SF(nm, k) {(const uint8_t *)nm, (uint32_t)(sizeof nm - 1), k, WO_SCHEMA_NONE, WO_SCHEMA_NONE}
|
||||
#define SFC(nm, k, fc) {(const uint8_t *)nm, (uint32_t)(sizeof nm - 1), k, fc, WO_SCHEMA_NONE}
|
||||
#define SC(nm, fl, arr) {(const uint8_t *)nm, (uint32_t)(sizeof nm - 1), fl, \
|
||||
(uint32_t)(sizeof arr / sizeof arr[0]), arr}
|
||||
|
||||
|
||||
/* every migrate test speaks both sides: a classdesc array for the engine and
|
||||
* a wo_schema for the diff, built from the same literals */
|
||||
static const uint8_t mig_nt_kinds[] = {WO_K_SCALAR, WO_K_TEXT};
|
||||
static const wo_classdesc MIG_NT[] = {
|
||||
{.name = 0, .flags = 0, .field_cnt = 2, .kinds = mig_nt_kinds},
|
||||
};
|
||||
static const uint8_t mig_nte_kinds[] = {WO_K_SCALAR, WO_K_TEXT, WO_K_SCALAR};
|
||||
static const wo_classdesc MIG_NTE[] = {
|
||||
{.name = 0, .flags = 0, .field_cnt = 3, .kinds = mig_nte_kinds},
|
||||
};
|
||||
static const uint8_t mig_n_kinds[] = {WO_K_SCALAR};
|
||||
static const wo_classdesc MIG_N[] = {
|
||||
{.name = 0, .flags = 0, .field_cnt = 1, .kinds = mig_n_kinds},
|
||||
};
|
||||
|
||||
static wo_schema_field mig_sf_n[] = {SF("n", WO_K_SCALAR)};
|
||||
static wo_schema_field mig_sf_nt[] = {SF("n", WO_K_SCALAR), SF("t", WO_K_TEXT)};
|
||||
static wo_schema_field mig_sf_nte[] = {SF("n", WO_K_SCALAR), SF("t", WO_K_TEXT),
|
||||
SF("extra", WO_K_SCALAR)};
|
||||
|
||||
static db_text *mig_text(const char *sz) {
|
||||
size_t n = strlen(sz);
|
||||
db_text *t = malloc(sizeof(db_text) + n);
|
||||
t->len = (uint32_t)n;
|
||||
memcpy(t->bytes, sz, n);
|
||||
return t;
|
||||
}
|
||||
|
||||
/* REORDER + OWNED FIXUP: the classes swap declaration order and one of them
|
||||
* embeds the other by value. The record's outer cid AND the cid inside the
|
||||
* stored owned value must both be renumbered — the outer one alone would
|
||||
* decode the embedded value against the wrong class. */
|
||||
static void test_migrate_reorder_owned(void) {
|
||||
char path[128];
|
||||
snprintf(path, sizeof path, "%s/migreorder.wal", g_dir);
|
||||
static const uint8_t x_kinds[] = {WO_K_SCALAR};
|
||||
static const uint8_t c_kinds[] = {WO_K_OWNED, WO_K_SCALAR};
|
||||
static const wo_classdesc OLD_XC[] = {
|
||||
{.name = 0, .flags = 0, .field_cnt = 1, .kinds = x_kinds},
|
||||
{.name = 0, .flags = 0, .field_cnt = 2, .kinds = c_kinds},
|
||||
};
|
||||
static const wo_classdesc NEW_CX[] = {
|
||||
{.name = 0, .flags = 0, .field_cnt = 2, .kinds = c_kinds},
|
||||
{.name = 0, .flags = 0, .field_cnt = 1, .kinds = x_kinds},
|
||||
};
|
||||
static wo_schema_field sx[] = {SF("v", WO_K_SCALAR)};
|
||||
static wo_schema_field sc_old[] = {SFC("part", WO_K_OWNED, 0), SF("m", WO_K_SCALAR)};
|
||||
static wo_schema_field sc_new[] = {SFC("part", WO_K_OWNED, 1), SF("m", WO_K_SCALAR)};
|
||||
wo_schema_class oc[] = {SC("X", 0, sx), SC("C", 0, sc_old)};
|
||||
wo_schema oldsc = {2, oc, NULL};
|
||||
wo_schema_class nc[] = {SC("C", 0, sc_new), SC("X", 0, sx)};
|
||||
wo_schema newsc = {2, nc, NULL};
|
||||
|
||||
{
|
||||
wo_db db;
|
||||
T_EQ(wo_db_init(&db, OLD_XC, 2, 0, 1), 0);
|
||||
wo_wal w;
|
||||
T_EQ(wo_wal_open(&w, path, 1 << 16), 0);
|
||||
db_row *rx = wo_row_create_raw(&db, 0, 3); /* an X row, old cid 0 */
|
||||
rx->slots[0] = 7;
|
||||
T_EQ(wo_row_raw_commit(&db, 0, rx), 0);
|
||||
T_EQ(wo_wal_append_insert(&w, &db, 0, 3), 0);
|
||||
db_rec *part = malloc(sizeof(db_rec) + 8); /* embedded X, old cid 0 */
|
||||
part->class_id = 0;
|
||||
part->_pad = 0;
|
||||
part->slots[0] = 42;
|
||||
db_row *rc = wo_row_create_raw(&db, 1, 5); /* a C row, old cid 1 */
|
||||
rc->slots[0] = (uint64_t)(uintptr_t)part;
|
||||
rc->slots[1] = 9;
|
||||
T_EQ(wo_row_raw_commit(&db, 1, rc), 0);
|
||||
T_EQ(wo_wal_append_insert(&w, &db, 1, 5), 0);
|
||||
T_EQ(wo_wal_commit(&w), 0);
|
||||
wo_wal_close(&w);
|
||||
wo_db_destroy(&db);
|
||||
}
|
||||
wo_db db2;
|
||||
T_EQ(wo_db_init(&db2, NEW_CX, 2, 0, 1), 0);
|
||||
wo_mig_plan pl;
|
||||
T_EQ(wo_schema_diff(&oldsc, &newsc, &pl), 0);
|
||||
T_EQ(pl.identity, 0);
|
||||
T_EQ(pl.classes[0].new_cid, 1u);
|
||||
T_EQ(pl.classes[1].new_cid, 0u);
|
||||
T_CHECK(pl.classes[0].poison == NULL && pl.classes[1].poison == NULL);
|
||||
T_EQ(wo_wal_migrate(path, &db2, &oldsc, &pl, &newsc, 1 << 16, NULL), 0);
|
||||
wo_mig_plan_free(&pl);
|
||||
T_EQ(wo_wal_replay(path, &db2), 2);
|
||||
db_row *rx = wo_row_ptr(&db2, 1, 3); /* X lives at cid 1 now */
|
||||
T_CHECK(rx != NULL && rx->slots[0] == 7);
|
||||
db_row *rc = wo_row_ptr(&db2, 0, 5); /* C lives at cid 0 now */
|
||||
T_CHECK(rc != NULL && rc->slots[1] == 9);
|
||||
db_rec *part = (db_rec *)(uintptr_t)rc->slots[0];
|
||||
T_CHECK(part != NULL && part->class_id == 1 && part->slots[0] == 42);
|
||||
wo_db_destroy(&db2);
|
||||
}
|
||||
|
||||
/* DELTA SPLICE: a keys-resident row's chain carries deltas on a field that is
|
||||
* being DELETED. The spliced chain must still fold — later deltas re-point
|
||||
* around the dropped ones — and the surviving field's latest value wins. */
|
||||
static void test_migrate_delta_splice(void) {
|
||||
char path[128];
|
||||
snprintf(path, sizeof path, "%s/migsplice.wal", g_dir);
|
||||
static wo_schema_field sk_old[] = {SF("n", WO_K_SCALAR), SF("label", WO_K_TEXT)};
|
||||
static wo_schema_field sk_new[] = {SF("n", WO_K_SCALAR)};
|
||||
wo_schema_class oc[] = {SC("K", WO_CLASSF_RESIDENT_KEYS, sk_old)};
|
||||
wo_schema oldsc = {1, oc, NULL};
|
||||
wo_schema_class nc[] = {SC("K", WO_CLASSF_RESIDENT_KEYS, sk_new)};
|
||||
wo_schema newsc = {1, nc, NULL};
|
||||
static const uint8_t knew_kinds[] = {WO_K_SCALAR};
|
||||
static const wo_classdesc KNEW[] = {
|
||||
{.name = 0,
|
||||
.flags = WO_CLASSF_RESIDENT_KEYS,
|
||||
.field_cnt = 1,
|
||||
.kinds = knew_kinds},
|
||||
};
|
||||
|
||||
wo_rt rt;
|
||||
T_EQ(wo_rt_init(&rt, 1 << 20, KEYS_CLASSES, 1), 0);
|
||||
uint64_t id;
|
||||
{
|
||||
wo_db db;
|
||||
T_EQ(wo_db_init(&db, KEYS_CLASSES, 1, 0, 1), 0);
|
||||
wo_wal w;
|
||||
T_EQ(wo_wal_open(&w, path, 1 << 18), 0);
|
||||
db.rt = &rt;
|
||||
rt.wal = &w;
|
||||
rt.db = &db;
|
||||
const char *msg = "";
|
||||
wo_str *sa = wo_str_new(&rt, "a", 1);
|
||||
uint64_t vals[2] = {1, (uint64_t)(uintptr_t)sa};
|
||||
id = wo_row_insert(&db, 0, vals, &msg, NULL);
|
||||
T_CHECK(id != 0);
|
||||
uint64_t off = wo_wal_next_offset(&w);
|
||||
T_EQ(wo_wal_append_insert(&w, &db, 0, id), 0);
|
||||
T_EQ(wo_wal_commit(&w), 0);
|
||||
T_EQ(wo_row_drop_payload(&db, 0, id, off), 0);
|
||||
/* the chain: n=2, label="x" (doomed), n=3 — the last delta's back
|
||||
pointer crosses the doomed one */
|
||||
chain_update(&db, &w, 0, id, 0, 2);
|
||||
wo_str *sx = wo_str_new(&rt, "x", 1);
|
||||
chain_update(&db, &w, 0, id, 1, (uint64_t)(uintptr_t)sx);
|
||||
chain_update(&db, &w, 0, id, 0, 3);
|
||||
wo_wal_close(&w);
|
||||
wo_db_destroy(&db);
|
||||
}
|
||||
|
||||
wo_db db2;
|
||||
T_EQ(wo_db_init(&db2, KNEW, 1, 0, 1), 0);
|
||||
wo_mig_plan pl;
|
||||
T_EQ(wo_schema_diff(&oldsc, &newsc, &pl), 0);
|
||||
T_CHECK(pl.classes[0].poison == NULL && pl.classes[0].fmap[1] == -1);
|
||||
T_EQ(wo_wal_migrate(path, &db2, &oldsc, &pl, &newsc, 1 << 18, NULL), 0);
|
||||
wo_mig_plan_free(&pl);
|
||||
|
||||
/* replay the migrated log the way boot does for a keys table, then read
|
||||
the row back through the fold: the chain must resolve to n=3 */
|
||||
db2.rt = &rt;
|
||||
rt.wal = NULL;
|
||||
rt.db = &db2;
|
||||
T_CHECK(wo_wal_replay(path, &db2) >= 0);
|
||||
wo_wal w2;
|
||||
T_EQ(wo_wal_open(&w2, path, 1 << 18), 0);
|
||||
rt.wal = &w2;
|
||||
const char *msg = "";
|
||||
db_row *r = wo_row_borrow(&db2, 0, id, &msg);
|
||||
T_CHECK(r != NULL && r->slots[0] == 3);
|
||||
wo_row_release(&db2, 0, r);
|
||||
wo_wal_close(&w2);
|
||||
wo_db_destroy(&db2);
|
||||
wo_rt_destroy(&rt);
|
||||
}
|
||||
|
||||
/* ADD: a two-field log boots a three-field binary — rows survive, the new
|
||||
* field reads the kind's zero, the head record states the NEW shape, and a
|
||||
* stale compaction temp lying beside the log is discarded, not appended to */
|
||||
static void test_migrate_add_field(void) {
|
||||
char path[128];
|
||||
snprintf(path, sizeof path, "%s/migadd.wal", g_dir);
|
||||
wo_schema_class oc[] = {SC("row", 0, mig_sf_nt)};
|
||||
wo_schema oldsc = {1, oc, NULL};
|
||||
wo_schema_class nc[] = {SC("row", 0, mig_sf_nte)};
|
||||
wo_schema newsc = {1, nc, NULL};
|
||||
|
||||
/* the OLD program writes its log, schema record at the head */
|
||||
{
|
||||
wo_db db;
|
||||
T_EQ(wo_db_init(&db, MIG_NT, 1, 0, 1), 0);
|
||||
wo_wal w;
|
||||
T_EQ(wo_wal_open(&w, path, 1 << 16), 0);
|
||||
T_EQ(wo_wal_set_schema(&w, &oldsc), 0);
|
||||
T_EQ(wo_wal_ensure_schema(&w), 0);
|
||||
db_row *r1 = wo_row_create_raw(&db, 0, 1);
|
||||
r1->slots[0] = 7;
|
||||
r1->slots[1] = (uint64_t)(uintptr_t)mig_text("abc");
|
||||
T_EQ(wo_row_raw_commit(&db, 0, r1), 0);
|
||||
T_EQ(wo_wal_append_insert(&w, &db, 0, 1), 0);
|
||||
db_row *r2 = wo_row_create_raw(&db, 0, 2);
|
||||
r2->slots[0] = 9;
|
||||
r2->slots[1] = 0;
|
||||
T_EQ(wo_row_raw_commit(&db, 0, r2), 0);
|
||||
T_EQ(wo_wal_append_insert(&w, &db, 0, 2), 0);
|
||||
T_EQ(wo_wal_commit(&w), 0);
|
||||
T_EQ(wo_row_remove(&db, 0, 2), 0);
|
||||
T_EQ(wo_wal_append_remove(&w, 0, 2), 0);
|
||||
T_EQ(wo_wal_commit(&w), 0);
|
||||
wo_wal_close(&w);
|
||||
wo_db_destroy(&db);
|
||||
}
|
||||
/* a stale temp beside the log: never authoritative, must be discarded */
|
||||
{
|
||||
char tmp[160];
|
||||
snprintf(tmp, sizeof tmp, "%s.compact", path);
|
||||
FILE *f = fopen(tmp, "w");
|
||||
T_CHECK(f != NULL);
|
||||
fputs("stale-not-a-record", f);
|
||||
fclose(f);
|
||||
}
|
||||
|
||||
/* the NEW binary migrates it at boot */
|
||||
wo_db db3;
|
||||
T_EQ(wo_db_init(&db3, MIG_NTE, 1, 0, 1), 0);
|
||||
wo_mig_plan pl;
|
||||
T_EQ(wo_schema_diff(&oldsc, &newsc, &pl), 0);
|
||||
T_EQ(pl.identity, 0);
|
||||
T_CHECK(pl.classes[0].poison == NULL);
|
||||
char *err = NULL;
|
||||
T_EQ(wo_wal_migrate(path, &db3, &oldsc, &pl, &newsc, 1 << 16, &err), 0);
|
||||
T_CHECK(err == NULL);
|
||||
wo_mig_plan_free(&pl);
|
||||
|
||||
/* head record: the new three-field shape */
|
||||
uint8_t *sp;
|
||||
uint32_t slen;
|
||||
T_EQ(wo_wal_read_schema(path, &sp, &slen), 0);
|
||||
wo_schema *head = wo_schema_decode(sp, slen);
|
||||
T_CHECK(head != NULL && head->classes[0].field_cnt == 3);
|
||||
wo_schema_free(head);
|
||||
free(sp);
|
||||
|
||||
/* replay: row 1 intact with a zero-valued third field, row 2 gone */
|
||||
T_EQ(wo_wal_replay(path, &db3), 3); /* insert, insert, remove */
|
||||
db_row *r = wo_row_ptr(&db3, 0, 1);
|
||||
T_CHECK(r != NULL && r->slots[0] == 7);
|
||||
db_text *t = (db_text *)(uintptr_t)r->slots[1];
|
||||
T_CHECK(t != NULL && t->len == 3 && memcmp(t->bytes, "abc", 3) == 0);
|
||||
T_EQ(r->slots[2], 0u);
|
||||
T_CHECK(wo_row_ptr(&db3, 0, 2) == NULL);
|
||||
wo_db_destroy(&db3);
|
||||
}
|
||||
|
||||
/* DELETE: the Text column's stored values are freed (ASan holds the leash)
|
||||
* and the surviving field lands in its new slot */
|
||||
static void test_migrate_delete_field(void) {
|
||||
char path[128];
|
||||
snprintf(path, sizeof path, "%s/migdel.wal", g_dir);
|
||||
wo_schema_class oc[] = {SC("row", 0, mig_sf_nt)};
|
||||
wo_schema oldsc = {1, oc, NULL};
|
||||
wo_schema_class nc[] = {SC("row", 0, mig_sf_n)};
|
||||
wo_schema newsc = {1, nc, NULL};
|
||||
{
|
||||
wo_db db;
|
||||
T_EQ(wo_db_init(&db, MIG_NT, 1, 0, 1), 0);
|
||||
wo_wal w;
|
||||
T_EQ(wo_wal_open(&w, path, 1 << 16), 0);
|
||||
for (uint64_t id = 1; id <= 20; id++) {
|
||||
db_row *r = wo_row_create_raw(&db, 0, id);
|
||||
r->slots[0] = id * 10;
|
||||
r->slots[1] = (uint64_t)(uintptr_t)mig_text("payload-to-drop");
|
||||
T_EQ(wo_row_raw_commit(&db, 0, r), 0);
|
||||
T_EQ(wo_wal_append_insert(&w, &db, 0, id), 0);
|
||||
}
|
||||
T_EQ(wo_wal_commit(&w), 0);
|
||||
wo_wal_close(&w);
|
||||
wo_db_destroy(&db);
|
||||
}
|
||||
wo_db db2;
|
||||
T_EQ(wo_db_init(&db2, MIG_N, 1, 0, 1), 0);
|
||||
wo_mig_plan pl;
|
||||
T_EQ(wo_schema_diff(&oldsc, &newsc, &pl), 0);
|
||||
T_CHECK(pl.classes[0].poison == NULL && pl.classes[0].fmap[1] == -1);
|
||||
T_EQ(wo_wal_migrate(path, &db2, &oldsc, &pl, &newsc, 1 << 16, NULL), 0);
|
||||
wo_mig_plan_free(&pl);
|
||||
T_EQ(wo_wal_replay(path, &db2), 20);
|
||||
for (uint64_t id = 1; id <= 20; id++) {
|
||||
db_row *r = wo_row_ptr(&db2, 0, id);
|
||||
T_CHECK(r != NULL && r->slots[0] == id * 10);
|
||||
}
|
||||
wo_db_destroy(&db2);
|
||||
}
|
||||
|
||||
/* POISON BITES ONLY WITH RECORDS: a retyped class with no stored rows never
|
||||
* blocks the boot; the same retype WITH a row refuses and names the field */
|
||||
static void test_migrate_poison_needs_records(void) {
|
||||
char path[128];
|
||||
snprintf(path, sizeof path, "%s/migpoison.wal", g_dir);
|
||||
static const uint8_t two_kinds0[] = {WO_K_SCALAR, WO_K_TEXT};
|
||||
static const uint8_t two_kinds1[] = {WO_K_SCALAR};
|
||||
static const wo_classdesc TWO[] = {
|
||||
{.name = 0, .flags = 0, .field_cnt = 2, .kinds = two_kinds0},
|
||||
{.name = 0, .flags = 0, .field_cnt = 1, .kinds = two_kinds1},
|
||||
};
|
||||
static const uint8_t two_kinds1f[] = {WO_K_FLOAT};
|
||||
static const wo_classdesc TWO_NEW[] = {
|
||||
{.name = 0, .flags = 0, .field_cnt = 2, .kinds = two_kinds0},
|
||||
{.name = 0, .flags = 0, .field_cnt = 1, .kinds = two_kinds1f},
|
||||
};
|
||||
wo_schema_field b_old[] = {SF("x", WO_K_SCALAR)};
|
||||
wo_schema_field b_new[] = {SF("x", WO_K_FLOAT)};
|
||||
wo_schema_class oc[] = {SC("A", 0, mig_sf_nt), SC("B", 0, b_old)};
|
||||
wo_schema oldsc = {2, oc, NULL};
|
||||
/* the new side also ADDS a field to A, so the plan is not identity and
|
||||
the transcode genuinely runs */
|
||||
wo_schema_class nc[] = {SC("A", 0, mig_sf_nte), SC("B", 0, b_new)};
|
||||
wo_schema newsc = {2, nc, NULL};
|
||||
|
||||
/* log 1: rows of A only */
|
||||
{
|
||||
wo_db db;
|
||||
T_EQ(wo_db_init(&db, TWO, 2, 0, 1), 0);
|
||||
wo_wal w;
|
||||
T_EQ(wo_wal_open(&w, path, 1 << 16), 0);
|
||||
db_row *r = wo_row_create_raw(&db, 0, 1);
|
||||
r->slots[0] = 1;
|
||||
r->slots[1] = 0;
|
||||
T_EQ(wo_row_raw_commit(&db, 0, r), 0);
|
||||
T_EQ(wo_wal_append_insert(&w, &db, 0, 1), 0);
|
||||
T_EQ(wo_wal_commit(&w), 0);
|
||||
wo_wal_close(&w);
|
||||
wo_db_destroy(&db);
|
||||
}
|
||||
wo_mig_plan pl;
|
||||
T_EQ(wo_schema_diff(&oldsc, &newsc, &pl), 0);
|
||||
T_CHECK(pl.classes[1].poison != NULL); /* B is poisoned... */
|
||||
{
|
||||
wo_db dbn;
|
||||
T_EQ(wo_db_init(&dbn, TWO_NEW, 2, 0, 1), 0);
|
||||
char *err = NULL;
|
||||
T_EQ(wo_wal_migrate(path, &dbn, &oldsc, &pl, &newsc, 1 << 16, &err), 0);
|
||||
T_CHECK(err == NULL); /* ...but nothing of B is stored: boots fine */
|
||||
wo_db_destroy(&dbn);
|
||||
}
|
||||
/* log 2: now with a B record — the poison bites and names the field */
|
||||
{
|
||||
wo_db db;
|
||||
T_EQ(wo_db_init(&db, TWO, 2, 0, 1), 0);
|
||||
wo_wal w;
|
||||
T_EQ(wo_wal_open(&w, path, 1 << 16), 0); /* migrated log: reopen fresh */
|
||||
char p2[144];
|
||||
snprintf(p2, sizeof p2, "%s2", path);
|
||||
wo_wal w2;
|
||||
T_EQ(wo_wal_open(&w2, p2, 1 << 16), 0);
|
||||
db_row *rb = wo_row_create_raw(&db, 1, 4);
|
||||
rb->slots[0] = 11;
|
||||
T_EQ(wo_row_raw_commit(&db, 1, rb), 0);
|
||||
T_EQ(wo_wal_append_insert(&w2, &db, 1, 4), 0);
|
||||
T_EQ(wo_wal_commit(&w2), 0);
|
||||
wo_wal_close(&w2);
|
||||
wo_wal_close(&w);
|
||||
wo_db_destroy(&db);
|
||||
wo_db dbn;
|
||||
T_EQ(wo_db_init(&dbn, TWO_NEW, 2, 0, 1), 0);
|
||||
char *err = NULL;
|
||||
T_EQ(wo_wal_migrate(p2, &dbn, &oldsc, &pl, &newsc, 1 << 16, &err), -2);
|
||||
T_CHECK(err != NULL && strstr(err, "`x`") != NULL);
|
||||
free(err);
|
||||
wo_db_destroy(&dbn);
|
||||
}
|
||||
wo_mig_plan_free(&pl);
|
||||
}
|
||||
|
||||
/* CORRUPT INPUT: a torn tail ends the intact prefix — the transcode takes
|
||||
* the prefix (same rule as replay), never the tear */
|
||||
static void test_migrate_corrupt_input(void) {
|
||||
char path[128];
|
||||
snprintf(path, sizeof path, "%s/migcorrupt.wal", g_dir);
|
||||
wo_schema_class oc[] = {SC("row", 0, mig_sf_nt)};
|
||||
wo_schema oldsc = {1, oc, NULL};
|
||||
wo_schema_class nc[] = {SC("row", 0, mig_sf_nte)};
|
||||
wo_schema newsc = {1, nc, NULL};
|
||||
{
|
||||
wo_db db;
|
||||
T_EQ(wo_db_init(&db, MIG_NT, 1, 0, 1), 0);
|
||||
wo_wal w;
|
||||
T_EQ(wo_wal_open(&w, path, 0), 0);
|
||||
for (uint64_t id = 1; id <= 3; id++) {
|
||||
db_row *r = wo_row_create_raw(&db, 0, id);
|
||||
r->slots[0] = id;
|
||||
r->slots[1] = 0;
|
||||
T_EQ(wo_row_raw_commit(&db, 0, r), 0);
|
||||
T_EQ(wo_wal_append_insert(&w, &db, 0, id), 0);
|
||||
}
|
||||
T_EQ(wo_wal_commit(&w), 0);
|
||||
/* tear the LAST record's tail byte */
|
||||
off_t end = lseek(w.fd, 0, SEEK_END);
|
||||
T_CHECK(end > 4);
|
||||
uint8_t junk = 0xFF;
|
||||
T_EQ((int)pwrite(w.fd, &junk, 1, end - 1), 1);
|
||||
wo_wal_close(&w);
|
||||
wo_db_destroy(&db);
|
||||
}
|
||||
wo_db db2;
|
||||
T_EQ(wo_db_init(&db2, MIG_NTE, 1, 0, 1), 0);
|
||||
wo_mig_plan pl;
|
||||
T_EQ(wo_schema_diff(&oldsc, &newsc, &pl), 0);
|
||||
T_EQ(wo_wal_migrate(path, &db2, &oldsc, &pl, &newsc, 0, NULL), 0);
|
||||
wo_mig_plan_free(&pl);
|
||||
T_EQ(wo_wal_replay(path, &db2), 2); /* rows 1 and 2; the torn third is gone */
|
||||
T_CHECK(wo_row_ptr(&db2, 0, 1) != NULL && wo_row_ptr(&db2, 0, 3) == NULL);
|
||||
wo_db_destroy(&db2);
|
||||
}
|
||||
|
||||
/* ---- databasev2 12: the boot diff --------------------------------------- */
|
||||
|
||||
static void test_schema_diff_verdicts(void) {
|
||||
/* base: A { n: scalar, t: text }, B { part: owned->A } */
|
||||
wo_schema_field a_f[] = {SF("n", WO_K_SCALAR), SF("t", WO_K_TEXT)};
|
||||
|
|
@ -2945,6 +3358,12 @@ int main(void) {
|
|||
test_keys_resident_update_field();
|
||||
test_keys_resident_update_indexed();
|
||||
test_keys_resident_indexed_across_flatten();
|
||||
test_migrate_reorder_owned();
|
||||
test_migrate_delta_splice();
|
||||
test_migrate_add_field();
|
||||
test_migrate_delete_field();
|
||||
test_migrate_poison_needs_records();
|
||||
test_migrate_corrupt_input();
|
||||
test_schema_diff_verdicts();
|
||||
test_schema_roundtrip();
|
||||
test_schema_fresh_log();
|
||||
|
|
|
|||
Loading…
Reference in a new issue