dev/MultiforaDB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

db: drop the docs hashmap; the _id_ index is the lookup

Browse Source 
The last structure holding the engine to RAM. At the target scale it cost 64-100
bytes per document -- 10+ GB at 100M documents -- and PLAN D4 rules it out for
exactly that reason.

`_id_` was already an ordered B+tree over the canonical `bson.encode_key`, and
since the leaf payload became a slab offset it has held everything the map did.
So the internal key changes from `serialize_value` to `encode_key` throughout,
`lookup_exact` replaces `docs.get`, and the tree's ordered walk replaces the map's
hash-order iteration in `create_index`, `rebuild_index`, the rebuild and the TTL
sweep -- which reads the slab sequentially where the map read it scattered.
`Collection.doc_count` remains, because the compaction trigger wants a count the
tree cannot give in O(1).

This is what unblocked the milestone's central claim, and the mechanism is worth
naming. Opening from a checkpoint had to rebuild the map, and rebuilding it meant
reading *every document* to recover its `_id` -- which faulted the entire database
in and made "RSS = working set" impossible no matter what else was true. Deleting
the map deleted that scan.

Measured, 512 MB of documents in 16 KB records, reopening from a checkpoint:

  RSS after reopen        523 MB  ->  50 MB
  after 4 point lookups   523 MB  ->  51 MB

The remaining 50 MB is the working set: index pages plus the un-checkpointed log
tail being replayed. A checkpoint immediately before shutdown would shrink it
further; the point is that it tracks what is touched rather than what is stored.

--

Replay now maintains `_id_` as it goes, always, not just after a checkpoint. It
is no longer an optimisation: the tree is the only way the next record can find
the document it supersedes. Secondaries still wait for the bulk build.

And PLAN amendment A4's migration hazard is handled where it actually bites. A
database written before `_id_` was canonical could hold two documents whose `_id`s
compare equal -- int32 1 and int64 1 -- and replaying it now keeps only the later
one. That is MongoDB's semantics and a one-way migration, so replay compares the
superseded document's `_id` bytes with the incoming record's and says so out loud
when they differ, naming the namespace.
This commit is contained in:
Alexey Shakhmatov
2026-08-03 21:58:21 +03:00
parent 148e03ac9f
commit b20ae92cbf
1 changed files with 170 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

267
src/db.zig
View File

@@ -34,7 +34,10 @@ pub const Collection = struct {
/// Removed documents leave garbage bytes until a rebuild rewrites them. A /// Removed documents leave garbage bytes until a rebuild rewrites them. A
/// checkpoint must never renumber these offsets: every index leaf holds one /// checkpoint must never renumber these offsets: every index leaf holds one
/// (PLAN amendment A3). /// (PLAN amendment A3).
docs: std.StringHashMapUnmanaged(u64), /// Live documents. The `_id_` index is the lookup now, so this is only a
/// count -- kept because the compaction trigger and `collStats` want it and
/// the tree cannot answer it in O(1).
doc_count: u64,
/// The data file this collection's documents live in. /// The data file this collection's documents live in.
pager: *pgr.Pager, pager: *pgr.Pager,
/// Extents owned by this collection's slab, in allocation order. /// Extents owned by this collection's slab, in allocation order.
@@ -76,7 +79,7 @@ pub const Collection = struct {
fn init(gpa: std.mem.Allocator, pager: *pgr.Pager) !Collection { fn init(gpa: std.mem.Allocator, pager: *pgr.Pager) !Collection {
var self: Collection = .{ var self: Collection = .{
.docs = .empty, .doc_count = 0,
.pager = pager, .pager = pager,
.slab_extents = .empty, .slab_extents = .empty,
.slab_tail = 0, .slab_tail = 0,
@@ -346,20 +349,15 @@ pub const Engine = struct {
// never be smaller than this one's -- and a u64 underflow here would // never be smaller than this one's -- and a u64 underflow here would
// read as an astronomically large live count, permanently suppressing // read as an astronomically large live count, permanently suppressing
// compaction rather than crashing. // compaction rather than crashing.
assert_msg(self.live_docs >= coll.docs.count(), "dropping a collection would underflow the engine's live count"); assert_msg(self.live_docs >= coll.doc_count, "dropping a collection would underflow the engine's live count");
self.live_docs -= coll.docs.count(); self.live_docs -= coll.doc_count;
self.dead_docs += coll.docs.count(); self.dead_docs += coll.doc_count;
coll.id_index.deinit(self.gpa); coll.id_index.deinit(self.gpa);
for (coll.indexes.items) |ix| { for (coll.indexes.items) |ix| {
ix.deinit(self.gpa); ix.deinit(self.gpa);
self.gpa.destroy(ix); self.gpa.destroy(ix);
} }
coll.indexes.deinit(self.gpa); coll.indexes.deinit(self.gpa);
var doc_it = coll.docs.iterator();
while (doc_it.next()) |doc_entry| {
self.gpa.free(doc_entry.key_ptr.*);
}
coll.docs.deinit(self.gpa);
// Give the slab's pages back. They become reusable two generations // Give the slab's pages back. They become reusable two generations
// later, so a fallback to the previous image still finds them intact. // later, so a fallback to the previous image still finds them intact.
for (coll.slab_extents.items) |e| { for (coll.slab_extents.items) |e| {
@@ -388,21 +386,33 @@ pub const Engine = struct {
/// ///
/// The document itself is handed to the index: entries are located by /// The document itself is handed to the index: entries are located by
/// regenerating them from it, which is far cheaper than scanning. /// regenerating them from it, which is far cheaper than scanning.
fn evict_doc(self: *Engine, coll: *Collection, id_key: []const u8) void { /// `id_enc` is `bson.encode_key` of the document's `_id` -- the canonical
const old = coll.docs.fetchRemove(id_key) orelse return; /// encoding, which is what the `_id_` index is keyed on.
fn evict_doc(self: *Engine, coll: *Collection, id_enc: []const u8) void {
const off = coll.id_index.lookup_exact(id_enc) orelse return;
// Resolve the bytes before any mutation; the slab is untouched by // Resolve the bytes before any mutation; the slab is untouched by
// index removal, so the slice is safe for the call. // index removal, so the slice is safe for the call.
const old_bytes = coll.doc_bytes(old.value); const old_bytes = coll.doc_bytes(off);
// Entries are keyed by the document's slab offset now, which is exactly coll.id_index.remove_doc(self.gpa, old_bytes, off);
// what the map just gave us. for (coll.indexes.items) |ix| ix.remove_doc(self.gpa, old_bytes, off);
coll.id_index.remove_doc(self.gpa, old_bytes, old.value);
for (coll.indexes.items) |ix| ix.remove_doc(self.gpa, old_bytes, old.value);
self.gpa.free(old.key);
// This document's log record (and its slab bytes) just became garbage. // This document's log record (and its slab bytes) just became garbage.
// The fetchRemove above succeeded, so a live document was counted.
assert_msg(self.live_docs >= 1, "evicting a document would underflow the engine's live count"); assert_msg(self.live_docs >= 1, "evicting a document would underflow the engine's live count");
self.live_docs -= 1; self.live_docs -= 1;
self.dead_docs += 1; self.dead_docs += 1;
assert_msg(coll.doc_count >= 1, "evicting a document would underflow the collection's count");
coll.doc_count -= 1;
}
/// `bson.encode_key` of a stored document's `_id`, owned by the caller.
fn id_enc_of(gpa: std.mem.Allocator, doc_bytes: []const u8) ![]u8 {
var arena = std.heap.ArenaAllocator.init(gpa);
defer arena.deinit();
const id_value = (try bson.get_at(arena.allocator(), doc_bytes, "_id")) orelse
return error.MissingId;
var enc: std.ArrayListUnmanaged(u8) = .empty;
errdefer enc.deinit(gpa);
try bson.encode_key(id_value, gpa, &enc);
return enc.toOwnedSlice(gpa);
} }
// -- commands (callers must hold the matching lock) --------------------- // -- commands (callers must hold the matching lock) ---------------------
@@ -654,10 +664,12 @@ pub const Engine = struct {
var id_arena = std.heap.ArenaAllocator.init(self.gpa); var id_arena = std.heap.ArenaAllocator.init(self.gpa);
defer id_arena.deinit(); defer id_arena.deinit();
const id_value = (try bson.get_at(id_arena.allocator(), doc_bytes, "_id")) orelse unreachable; const id_value = (try bson.get_at(id_arena.allocator(), doc_bytes, "_id")) orelse unreachable;
// Ownership of the key moves to the map once `stored` is set. // The canonical encoding, because that is what `_id_` is keyed on. It
const id_key = try bson.serialize_value(self.gpa, id_value); // used to be serialize_value, for a hashmap that no longer exists.
var stored = false; var id_enc_list: std.ArrayListUnmanaged(u8) = .empty;
errdefer if (!stored) self.gpa.free(id_key); defer id_enc_list.deinit(self.gpa);
try bson.encode_key(id_value, self.gpa, &id_enc_list);
const id_enc = id_enc_list.items;
coll.dup_index = null; coll.dup_index = null;
// 1. Build entries for every index. ParallelArrays escapes here, // 1. Build entries for every index. ParallelArrays escapes here,
@@ -696,7 +708,7 @@ pub const Engine = struct {
// same-_id collision this must catch. For a replace it is the // same-_id collision this must catch. For a replace it is the
// document's *current* slab offset, since that is what its existing // document's *current* slab offset, since that is what its existing
// entries carry -- the new offset does not exist yet. // entries carry -- the new offset does not exist yet.
const exclude: ?u64 = if (mode == .replace) coll.docs.get(id_key) else null; const exclude: ?u64 = if (mode == .replace) coll.id_index.lookup_exact(id_enc) else null;
for (built_list.items) |*b| { for (built_list.items) |*b| {
if (!b.ix.unique) continue; if (!b.ix.unique) continue;
b.ix.check_unique(b.built.entries.items, exclude) catch { b.ix.check_unique(b.built.entries.items, exclude) catch {
@@ -727,18 +739,17 @@ pub const Engine = struct {
try self.log_append(.upsert, db_name, coll_name, doc_bytes); try self.log_append(.upsert, db_name, coll_name, doc_bytes);
// 6. Replace drops the old document (and its index entries). // 6. Replace drops the old document (and its index entries).
if (mode == .replace) self.evict_doc(coll, id_key); if (mode == .replace) self.evict_doc(coll, id_enc);
// 7. Publish the document and its entries: copy the bytes into the // 7. Publish the document and its entries: copy the bytes into the
// slab and record the offset. Infallible from here. // slab and record the offset. Infallible from here.
const off = coll.slab_append(doc_bytes); const off = coll.slab_append(doc_bytes);
try coll.docs.put(self.gpa, id_key, off);
self.live_docs += 1; self.live_docs += 1;
coll.doc_count += 1;
for (built_list.items) |*b| { for (built_list.items) |*b| {
if (b.built.multikey) b.ix.multikey = true; if (b.built.multikey) b.ix.multikey = true;
b.ix.insert_entries(&b.built, off); b.ix.insert_entries(&b.built, off);
} }
stored = true;
// The write is published; anything the reservations above did not claim // The write is published; anything the reservations above did not claim
// is dead. Leaving it promised would grow the file on every write. // is dead. Leaving it promised would grow the file on every write.
self.pager.release_reservation(); self.pager.release_reservation();
@@ -754,15 +765,17 @@ pub const Engine = struct {
coll_name: []const u8, coll_name: []const u8,
id: bson.Value, id: bson.Value,
) !bool { ) !bool {
const id_key = try bson.serialize_value(self.gpa, id); var enc: std.ArrayListUnmanaged(u8) = .empty;
defer self.gpa.free(id_key); defer enc.deinit(self.gpa);
return self.remove(db_name, coll_name, id_key); try bson.encode_key(id, self.gpa, &enc);
return self.remove(db_name, coll_name, enc.items);
} }
fn remove(self: *Engine, db_name: []const u8, coll_name: []const u8, id_key: []const u8) !bool { /// `id_enc` is `bson.encode_key` of the `_id`.
fn remove(self: *Engine, db_name: []const u8, coll_name: []const u8, id_enc: []const u8) !bool {
const db = self.dbs.get(db_name) orelse return false; const db = self.dbs.get(db_name) orelse return false;
const coll = db.collections.get(coll_name) orelse return false; const coll = db.collections.get(coll_name) orelse return false;
const off = coll.docs.get(id_key) orelse return false; const off = coll.id_index.lookup_exact(id_enc) orelse return false;
// Log (and sync) the delete before removing it from memory, so the // Log (and sync) the delete before removing it from memory, so the
// log always describes at least as much as the in-memory state. // log always describes at least as much as the in-memory state.
@@ -777,7 +790,7 @@ pub const Engine = struct {
try bson.write_doc(&id_pairs, self.gpa, &id_doc); try bson.write_doc(&id_pairs, self.gpa, &id_doc);
try self.log_append(.delete, db_name, coll_name, id_doc.items); try self.log_append(.delete, db_name, coll_name, id_doc.items);
self.evict_doc(coll, id_key); self.evict_doc(coll, id_enc);
// Deletes grow the log too. Without this a delete-heavy workload // Deletes grow the log too. Without this a delete-heavy workload
// never compacts, because only upsert and ttl_sweep used to check. // never compacts, because only upsert and ttl_sweep used to check.
self.note_compact(); self.note_compact();
@@ -793,10 +806,11 @@ pub const Engine = struct {
self: *Engine, self: *Engine,
db_name: []const u8, db_name: []const u8,
coll_name: []const u8, coll_name: []const u8,
id_key: []const u8, /// `bson.encode_key` of the `_id`.
id_enc: []const u8,
) ?[]const u8 { ) ?[]const u8 {
const coll = self.get_collection(db_name, coll_name) orelse return null; const coll = self.get_collection(db_name, coll_name) orelse return null;
const off = coll.docs.get(id_key) orelse return null; const off = coll.id_index.lookup_exact(id_enc) orelse return null;
return coll.doc_bytes(off); return coll.doc_bytes(off);
} }
@@ -857,9 +871,9 @@ pub const Engine = struct {
// document into a sorted array memmoves the tail every time, which // document into a sorted array memmoves the tail every time, which
// is what made this quadratic. On any failure the deferred // is what made this quadratic. On any failure the deferred
// ix.deinit frees every appended key. Nothing is persisted. // ix.deinit frees every appended key. Nothing is persisted.
var doc_it = coll.docs.iterator(); var doc_it = coll.id_index.iter();
while (doc_it.next()) |entry| { while (doc_it.next()) |entry| {
try ix.append_doc_entries(self.gpa, coll.doc_bytes(entry.value_ptr.*), entry.value_ptr.*); try ix.append_doc_entries(self.gpa, coll.doc_bytes(entry.off), entry.off);
} }
_ = try ix.finish_bulk(self.gpa, true); _ = try ix.finish_bulk(self.gpa, true);
self.pager.release_reservation(); self.pager.release_reservation();
@@ -985,8 +999,9 @@ pub const Engine = struct {
var removed: usize = 0; var removed: usize = 0;
for (offs.items) |off| { for (offs.items) |off| {
const id_value = (try bson.get_at(arena.allocator(), coll.doc_bytes(off), "_id")) orelse continue; const id_value = (try bson.get_at(arena.allocator(), coll.doc_bytes(off), "_id")) orelse continue;
const id_key = try bson.serialize_value(arena.allocator(), id_value); var enc: std.ArrayListUnmanaged(u8) = .empty;
if (try self.remove(db_name, coll_name, id_key)) removed += 1; try bson.encode_key(id_value, arena.allocator(), &enc);
if (try self.remove(db_name, coll_name, enc.items)) removed += 1;
} }
return removed; return removed;
} }
@@ -1130,7 +1145,7 @@ pub const Engine = struct {
/// One document's new home during a rebuild: its docs-map key and the offset /// One document's new home during a rebuild: its docs-map key and the offset
/// it was copied to. Named rather than anonymous so the rebuild and the /// it was copied to. Named rather than anonymous so the rebuild and the
/// repack agree on the type. /// repack agree on the type.
const Moved = struct { key: []const u8, off: u64 }; const Moved = struct { off: u64 };
/// Reclaim what a checkpoint cannot: dead document bytes and abandoned /// Reclaim what a checkpoint cannot: dead document bytes and abandoned
/// pages. /// pages.
@@ -1199,18 +1214,19 @@ pub const Engine = struct {
// so a later scan reads it sequentially. // so a later scan reads it sequentially.
var moved: std.ArrayListUnmanaged(Moved) = .empty; var moved: std.ArrayListUnmanaged(Moved) = .empty;
defer moved.deinit(self.gpa); defer moved.deinit(self.gpa);
try moved.ensureTotalCapacity(self.gpa, coll.docs.count()); try moved.ensureTotalCapacity(self.gpa, @intCast(coll.doc_count));
var it = coll.docs.iterator(); // The `_id_` tree is the enumeration, and it is in key order -- so the
// new slab ends up ordered and a later scan reads it sequentially.
var it = coll.id_index.iter();
while (it.next()) |entry| { while (it.next()) |entry| {
const bytes = doc_bytes_in(self.pager, entry.value_ptr.*); const bytes = doc_bytes_in(self.pager, entry.off);
try coll.slab_reserve(self.gpa, bytes.len); try coll.slab_reserve(self.gpa, bytes.len);
const new_off = coll.slab_append(bytes); const new_off = coll.slab_append(bytes);
self.pager.release_reservation(); self.pager.release_reservation();
try moved.append(self.gpa, .{ .key = entry.key_ptr.*, .off = new_off }); try moved.append(self.gpa, .{ .off = new_off });
} }
// Republish the offsets. // Republish the offsets.
for (moved.items) |m| coll.docs.putAssumeCapacity(m.key, m.off);
// Rebuild every index from the new offsets, bulk-packed. // Rebuild every index from the new offsets, bulk-packed.
try self.repack_index(coll, &coll.id_index, moved.items); try self.repack_index(coll, &coll.id_index, moved.items);
@@ -1270,9 +1286,12 @@ pub const Engine = struct {
/// opens, leaving dropIndexes as an in-band recovery path. /// opens, leaving dropIndexes as an in-band recovery path.
fn rebuild_index(self: *Engine, coll: *Collection, ix: *index.Index) !void { fn rebuild_index(self: *Engine, coll: *Collection, ix: *index.Index) !void {
if (ix.count() > 0) return; // defensive if (ix.count() > 0) return; // defensive
var doc_it = coll.docs.iterator(); // The `_id_` tree is the enumeration of live documents now. It is also
while (doc_it.next()) |doc_entry| { // ordered, so this reads the slab sequentially where the hashmap read it
ix.append_doc_entries(self.gpa, coll.doc_bytes(doc_entry.value_ptr.*), doc_entry.value_ptr.*) catch |err| switch (err) { // in hash order.
var doc_it = coll.id_index.iter();
while (doc_it.next()) |entry| {
ix.append_doc_entries(self.gpa, coll.doc_bytes(entry.off), entry.off) catch |err| switch (err) {
error.ParallelArrays => { error.ParallelArrays => {
std.debug.print( std.debug.print(
"multiforadb: WARNING: index '{s}' cannot index an existing " ++ "multiforadb: WARNING: index '{s}' cannot index an existing " ++
@@ -1441,11 +1460,12 @@ pub const Engine = struct {
}; };
try coll.indexes.append(self.gpa, boxed); try coll.indexes.append(self.gpa, boxed);
} }
// The docs map is still the authoritative _id -> offset lookup, // Nothing else to rebuild. The `_id_` tree *is* the lookup, and
// and it is not in the catalog on purpose: the commit that drops // it is already in the file -- which is the whole reason an open
// it would only have to delete the format again. Rebuilt from the // no longer touches a single document page. The version of this
// _id_ tree instead, which the data file already holds. // that kept a hashmap had to read every document here to recover
try self.rebuild_docs_map(coll); // its `_id`, and that alone faulted the entire database in.
coll.doc_count = coll.id_index.count();
} }
} }
} }
@@ -1526,7 +1546,6 @@ pub const Engine = struct {
ix.insert_entries(&built, off); ix.insert_entries(&built, off);
} }
/// Rebuild the _id -> offset hashmap by walking the _id_ tree.
fn rebuild_docs_map(self: *Engine, coll: *Collection) !void { fn rebuild_docs_map(self: *Engine, coll: *Collection) !void {
var arena = std.heap.ArenaAllocator.init(self.gpa); var arena = std.heap.ArenaAllocator.init(self.gpa);
defer arena.deinit(); defer arena.deinit();
@@ -1702,9 +1721,10 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an
std.debug.print("multiforadb: log record without _id, skipping\n", .{}); std.debug.print("multiforadb: log record without _id, skipping\n", .{});
return; return;
}; };
const id_key = try bson.serialize_value(self.gpa, id_value); var id_enc_list: std.ArrayListUnmanaged(u8) = .empty;
var key_owned = false; defer id_enc_list.deinit(self.gpa);
defer if (!key_owned) self.gpa.free(id_key); bson.encode_key(id_value, self.gpa, &id_enc_list) catch return;
const id_enc = id_enc_list.items;
// Engine.seq used to restart at 0 on every open, which was harmless while // Engine.seq used to restart at 0 on every open, which was harmless while
// the log was always replayed in full and fatal the moment a watermark // the log was always replayed in full and fatal the moment a watermark
@@ -1714,16 +1734,30 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an
switch (record.type) { switch (record.type) {
storage.record_type_upsert => { storage.record_type_upsert => {
self.evict_doc(coll, id_key); // A record that supersedes one already present. Worth reporting when
// the two `_id`s are not byte-identical, because that means the
// database predates `_id_` being canonical and held two documents
// whose ids compare equal -- int32 1 and int64 1, say. One of them is
// being dropped here, which is MongoDB's semantics but is also silent
// data loss for an existing file (PLAN amendment A4).
if (coll.id_index.lookup_exact(id_enc)) |old_off| {
warn_on_equal_id_collision(self.gpa, coll, old_off, doc, record);
}
self.evict_doc(coll, id_enc);
const doc_bytes = try serialize_doc(self.gpa, doc); const doc_bytes = try serialize_doc(self.gpa, doc);
defer self.gpa.free(doc_bytes); defer self.gpa.free(doc_bytes);
try coll.slab_reserve(self.gpa, doc_bytes.len); try coll.slab_reserve(self.gpa, doc_bytes.len);
const off = coll.slab_append(doc_bytes); const off = coll.slab_append(doc_bytes);
try coll.docs.put(self.gpa, id_key, off);
self.live_docs += 1; self.live_docs += 1;
key_owned = true; coll.doc_count += 1;
// The `_id_` entry is added *now*, not after replay: it is the only
// way the next record can find this document to supersede it. The
// secondaries can still wait for the bulk build, unless this open
// came from a checkpoint that already populated them.
if (self.replay_maintains_indexes) { if (self.replay_maintains_indexes) {
self.index_doc_on_replay(coll, doc_bytes, off); self.index_doc_on_replay(coll, doc_bytes, off);
} else {
self.index_one(&coll.id_index, doc_bytes, off);
} }
self.pager.release_reservation(); self.pager.release_reservation();
// The _id_ entry is added after replay, in build_all_indexes, // The _id_ entry is added after replay, in build_all_indexes,
@@ -1737,7 +1771,7 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an
// must always open). The commit that drops the docs map is where // must always open). The commit that drops the docs map is where
// that stops being true -- see PLAN amendment A4. // that stops being true -- see PLAN amendment A4.
}, },
storage.record_type_delete => self.evict_doc(coll, id_key), storage.record_type_delete => self.evict_doc(coll, id_enc),
else => {}, else => {},
} }
} }
@@ -1795,7 +1829,7 @@ test "insert, query, remove" {
engine.unlock(); engine.unlock();
// find by id // find by id
const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = 2 }); const id_key = try id_key_for(gpa, bson.Value{ .int32 = 2 });
defer gpa.free(id_key); defer gpa.free(id_key);
try engine.lock(); try engine.lock();
const found = engine.get_doc("app", "users", id_key).?; const found = engine.get_doc("app", "users", id_key).?;
@@ -1931,10 +1965,10 @@ test "reopen replays log" {
var engine2 = try Engine.open(gpa, io, tmp.path); var engine2 = try Engine.open(gpa, io, tmp.path);
defer engine2.deinit(); defer engine2.deinit();
try engine2.lock(); try engine2.lock();
const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = 2 }); const id_key = try id_key_for(gpa, bson.Value{ .int32 = 2 });
defer gpa.free(id_key); defer gpa.free(id_key);
try testing.expect(engine2.get_doc("app", "users", id_key) == null); try testing.expect(engine2.get_doc("app", "users", id_key) == null);
const id_key1 = try bson.serialize_value(gpa, bson.Value{ .int32 = 1 }); const id_key1 = try id_key_for(gpa, bson.Value{ .int32 = 1 });
defer gpa.free(id_key1); defer gpa.free(id_key1);
try testing.expectEqualStrings("alice", (try bson.get_at(gpa, engine2.get_doc("app", "users", id_key1).?, "name")).?.string); try testing.expectEqualStrings("alice", (try bson.get_at(gpa, engine2.get_doc("app", "users", id_key1).?, "name")).?.string);
engine2.unlock(); engine2.unlock();
@@ -1971,11 +2005,11 @@ test "auto _id generation survives reopen" {
var engine2 = try Engine.open(gpa, io, tmp.path); var engine2 = try Engine.open(gpa, io, tmp.path);
defer engine2.deinit(); defer engine2.deinit();
const coll = engine2.get_collection("app", "no_ids").?; const coll = engine2.get_collection("app", "no_ids").?;
var it = coll.docs.iterator(); var it = coll.id_index.iter();
var count: usize = 0; var count: usize = 0;
while (it.next()) |entry| { while (it.next()) |entry| {
count += 1; count += 1;
const b = coll.doc_bytes(entry.value_ptr.*); const b = coll.doc_bytes(entry.off);
try testing.expect((try bson.get_at(gpa, b, "_id")).?.object_id.len == 12); try testing.expect((try bson.get_at(gpa, b, "_id")).?.object_id.len == 12);
} }
try testing.expectEqual(@as(usize, 1), count); try testing.expectEqual(@as(usize, 1), count);
@@ -2023,7 +2057,7 @@ test "compaction rewrites log and keeps data" {
defer engine3.deinit(); defer engine3.deinit();
try engine3.lock(); try engine3.lock();
for (1..6) |i| { for (1..6) |i| {
const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = @intCast(i) }); const id_key = try id_key_for(gpa, bson.Value{ .int32 = @intCast(i) });
defer gpa.free(id_key); defer gpa.free(id_key);
try testing.expect(engine3.get_doc("app", "users", id_key) != null); try testing.expect(engine3.get_doc("app", "users", id_key) != null);
} }
@@ -2076,7 +2110,7 @@ test "concurrent readers and writers on a threaded Io" {
defer e.unlock_read(); defer e.unlock_read();
if (e.get_collection("app", "users")) |coll| { if (e.get_collection("app", "users")) |coll| {
var n: usize = 0; var n: usize = 0;
var it = coll.docs.iterator(); var it = coll.id_index.iter();
while (it.next()) |_| n += 1; while (it.next()) |_| n += 1;
// A reader must never observe more docs than can exist. // A reader must never observe more docs than can exist.
if (n > @as(usize, @intCast(total))) return error.Canceled; if (n > @as(usize, @intCast(total))) return error.Canceled;
@@ -2095,9 +2129,9 @@ test "concurrent readers and writers on a threaded Io" {
try engine.lock_read(); try engine.lock_read();
defer engine.unlock_read(); defer engine.unlock_read();
const coll = engine.get_collection("app", "users") orelse return error.TestUnexpectedResult; const coll = engine.get_collection("app", "users") orelse return error.TestUnexpectedResult;
try testing.expectEqual(@as(usize, @intCast(total)), coll.docs.count()); try testing.expectEqual(@as(usize, @intCast(total)), coll.id_index.count());
for (1..total + 1) |i| { for (1..total + 1) |i| {
const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = @intCast(i) }); const id_key = try id_key_for(gpa, bson.Value{ .int32 = @intCast(i) });
defer gpa.free(id_key); defer gpa.free(id_key);
try testing.expect(engine.get_doc("app", "users", id_key) != null); try testing.expect(engine.get_doc("app", "users", id_key) != null);
} }
@@ -2239,9 +2273,9 @@ test "concurrent writers compacting: the log survives a reopen" {
try reopened.lock_read(); try reopened.lock_read();
defer reopened.unlock_read(); defer reopened.unlock_read();
const coll = reopened.get_collection("app", "users") orelse return error.TestUnexpectedResult; const coll = reopened.get_collection("app", "users") orelse return error.TestUnexpectedResult;
try testing.expectEqual(@as(usize, @intCast(total)), coll.docs.count()); try testing.expectEqual(@as(usize, @intCast(total)), coll.id_index.count());
for (1..total + 1) |i| { for (1..total + 1) |i| {
const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = @intCast(i) }); const id_key = try id_key_for(gpa, bson.Value{ .int32 = @intCast(i) });
defer gpa.free(id_key); defer gpa.free(id_key);
try testing.expect(reopened.get_doc("app", "users", id_key) != null); try testing.expect(reopened.get_doc("app", "users", id_key) != null);
} }
@@ -2498,7 +2532,7 @@ test "a checkpoint lets the next open skip the log it covers" {
defer engine.unlock(); defer engine.unlock();
try testing.expect(engine.pager.loaded.generation >= 1); try testing.expect(engine.pager.loaded.generation >= 1);
const coll = engine.get_collection("app", "users").?; const coll = engine.get_collection("app", "users").?;
try testing.expectEqual(@as(usize, 45), coll.docs.count()); try testing.expectEqual(@as(usize, 45), coll.id_index.count());
try testing.expectEqual(@as(usize, 45), coll.id_index.count()); try testing.expectEqual(@as(usize, 45), coll.id_index.count());
// And the sequence continued from the watermark rather than restarting. // And the sequence continued from the watermark rather than restarting.
try testing.expect(engine.seq >= engine.pager.loaded.seq); try testing.expect(engine.seq >= engine.pager.loaded.seq);
@@ -2522,8 +2556,8 @@ test "a checkpoint lets the next open skip the log it covers" {
try engine.lock(); try engine.lock();
defer engine.unlock(); defer engine.unlock();
const coll = engine.get_collection("app", "users").?; const coll = engine.get_collection("app", "users").?;
try testing.expectEqual(@as(usize, 46), coll.docs.count()); try testing.expectEqual(@as(usize, 46), coll.id_index.count());
const id_key = try bson.serialize_value(gpa, .{ .int32 = 500 }); const id_key = try id_key_for(gpa, .{ .int32 = 500 });
defer gpa.free(id_key); defer gpa.free(id_key);
try testing.expect(engine.get_doc("app", "users", id_key) != null); try testing.expect(engine.get_doc("app", "users", id_key) != null);
} }
@@ -2597,12 +2631,12 @@ test "a checkpoint reclaims the log and the data survives" {
try engine.lock(); try engine.lock();
defer engine.unlock(); defer engine.unlock();
const coll = engine.get_collection("app", "users").?; const coll = engine.get_collection("app", "users").?;
try testing.expectEqual(@as(usize, 201), coll.docs.count());
try testing.expectEqual(@as(usize, 201), coll.id_index.count()); try testing.expectEqual(@as(usize, 201), coll.id_index.count());
const id_key = try bson.serialize_value(gpa, .{ .int32 = 999 }); try testing.expectEqual(@as(usize, 201), coll.id_index.count());
const id_key = try id_key_for(gpa, .{ .int32 = 999 });
defer gpa.free(id_key); defer gpa.free(id_key);
try testing.expect(engine.get_doc("app", "users", id_key) != null); try testing.expect(engine.get_doc("app", "users", id_key) != null);
const first_key = try bson.serialize_value(gpa, .{ .int32 = 0 }); const first_key = try id_key_for(gpa, .{ .int32 = 0 });
defer gpa.free(first_key); defer gpa.free(first_key);
try testing.expect(engine.get_doc("app", "users", first_key) != null); try testing.expect(engine.get_doc("app", "users", first_key) != null);
} }
@@ -2662,7 +2696,7 @@ test "a rebuild reclaims dead document bytes and keeps every index valid" {
// 60 live documents occupy less than the 120 writes that produced them. // 60 live documents occupy less than the 120 writes that produced them.
try testing.expect(coll.slab_used < before_used); try testing.expect(coll.slab_used < before_used);
try testing.expect(coll.slab_used > 0); try testing.expect(coll.slab_used > 0);
try testing.expectEqual(@as(usize, 60), coll.docs.count()); try testing.expectEqual(@as(usize, 60), coll.id_index.count());
try testing.expectEqual(@as(usize, 60), coll.id_index.count()); try testing.expectEqual(@as(usize, 60), coll.id_index.count());
// The assertions that matter. Content alone proves nothing here: the old // The assertions that matter. Content alone proves nothing here: the old
@@ -2674,20 +2708,16 @@ test "a rebuild reclaims dead document bytes and keeps every index valid" {
defer engine.unlock(); defer engine.unlock();
i = 0; i = 0;
while (i < 60) : (i += 1) { while (i < 60) : (i += 1) {
const id_key = try bson.serialize_value(gpa, .{ .int32 = i }); const id_key = try id_key_for(gpa, .{ .int32 = i });
defer gpa.free(id_key); defer gpa.free(id_key);
var enc: std.ArrayListUnmanaged(u8) = .empty; var enc: std.ArrayListUnmanaged(u8) = .empty;
defer enc.deinit(gpa); defer enc.deinit(gpa);
try bson.encode_key(.{ .int32 = i }, gpa, &enc); try bson.encode_key(.{ .int32 = i }, gpa, &enc);
const from_index = coll.id_index.lookup_exact(enc.items) orelse return error.TestUnexpectedResult; const from_index = coll.id_index.lookup_exact(enc.items) orelse return error.TestUnexpectedResult;
const from_map = coll.docs.get(id_key) orelse return error.TestUnexpectedResult; // The offset must be in the new slab. With the hashmap gone this is the
// whole check: there is no second structure to disagree with, so a
// The index and the map must agree, which catches either one being left // rebuild that left stale offsets behind shows up here and nowhere else.
// behind by the rebuild.
try testing.expectEqual(from_map, from_index);
// And the offset must be in the new slab, which catches both being left
// behind together.
try testing.expect(offset_in_extents(coll, from_index)); try testing.expect(offset_in_extents(coll, from_index));
try testing.expect(std.mem.indexOf(u8, coll.doc_bytes(from_index), "b@x.io") != null); try testing.expect(std.mem.indexOf(u8, coll.doc_bytes(from_index), "b@x.io") != null);
} }
@@ -2811,7 +2841,7 @@ test "drop_collection frees indexes; log without index records replays" {
var engine2 = try Engine.open(gpa, io, tmp.path); var engine2 = try Engine.open(gpa, io, tmp.path);
defer engine2.deinit(); defer engine2.deinit();
try engine2.lock(); try engine2.lock();
const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = 2 }); const id_key = try id_key_for(gpa, bson.Value{ .int32 = 2 });
defer gpa.free(id_key); defer gpa.free(id_key);
try testing.expect(engine2.get_doc("app", "plain", id_key) != null); try testing.expect(engine2.get_doc("app", "plain", id_key) != null);
// Pre-existing limitation (documented in the README): drop_collection // Pre-existing limitation (documented in the README): drop_collection
@@ -2875,12 +2905,12 @@ test "ttl_sweep deletes expired documents and the deletion survives reopen" {
try engine.insert("app", "sessions", &doc, &env.gen); try engine.insert("app", "sessions", &doc, &env.gen);
} }
const coll = engine.get_collection("app", "sessions").?; const coll = engine.get_collection("app", "sessions").?;
try testing.expectEqual(@as(usize, 6), coll.docs.count()); try testing.expectEqual(@as(usize, 6), coll.id_index.count());
try testing.expectEqual(@as(usize, 6), coll.indexes.items[0].count()); try testing.expectEqual(@as(usize, 6), coll.indexes.items[0].count());
// The cutoff is inclusive: doc 2 goes with doc 1. // The cutoff is inclusive: doc 2 goes with doc 1.
try testing.expectEqual(@as(usize, 2), try engine.ttl_sweep(now_ms)); try testing.expectEqual(@as(usize, 2), try engine.ttl_sweep(now_ms));
try testing.expectEqual(@as(usize, 4), coll.docs.count()); try testing.expectEqual(@as(usize, 4), coll.id_index.count());
try testing.expectEqual(@as(usize, 4), coll.indexes.items[0].count()); try testing.expectEqual(@as(usize, 4), coll.indexes.items[0].count());
try testing.expectEqual(@as(usize, 0), try index_count(gpa, &engine, "app", "sessions", "expireAt_1", .{ .datetime = now_ms - 120_000 })); try testing.expectEqual(@as(usize, 0), try index_count(gpa, &engine, "app", "sessions", "expireAt_1", .{ .datetime = now_ms - 120_000 }));
// The string and the missing field are untouched by any sweep. // The string and the missing field are untouched by any sweep.
@@ -2891,7 +2921,7 @@ test "ttl_sweep deletes expired documents and the deletion survives reopen" {
try testing.expectEqual(@as(usize, 0), try engine.ttl_sweep(now_ms)); try testing.expectEqual(@as(usize, 0), try engine.ttl_sweep(now_ms));
// An hour later doc 3 has expired too; doc 4 still has not. // An hour later doc 3 has expired too; doc 4 still has not.
try testing.expectEqual(@as(usize, 1), try engine.ttl_sweep(now_ms + 3_000_000)); try testing.expectEqual(@as(usize, 1), try engine.ttl_sweep(now_ms + 3_000_000));
try testing.expectEqual(@as(usize, 3), coll.docs.count()); try testing.expectEqual(@as(usize, 3), coll.id_index.count());
} }
// Sweeps go through `remove`, so they are logged: the deletions hold // Sweeps go through `remove`, so they are logged: the deletions hold
@@ -2901,16 +2931,16 @@ test "ttl_sweep deletes expired documents and the deletion survives reopen" {
try engine2.lock(); try engine2.lock();
defer engine2.unlock(); defer engine2.unlock();
const coll = engine2.get_collection("app", "sessions").?; const coll = engine2.get_collection("app", "sessions").?;
try testing.expectEqual(@as(usize, 3), coll.docs.count()); try testing.expectEqual(@as(usize, 3), coll.id_index.count());
try testing.expectEqual(@as(usize, 1), coll.indexes.items.len); try testing.expectEqual(@as(usize, 1), coll.indexes.items.len);
try testing.expectEqual(@as(?i64, 60), coll.indexes.items[0].ttl); try testing.expectEqual(@as(?i64, 60), coll.indexes.items[0].ttl);
try testing.expectEqual(@as(usize, 3), coll.indexes.items[0].count()); try testing.expectEqual(@as(usize, 3), coll.indexes.items[0].count());
for ([_]i32{ 1, 2, 3 }) |id| { for ([_]i32{ 1, 2, 3 }) |id| {
const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = id }); const id_key = try id_key_for(gpa, bson.Value{ .int32 = id });
defer gpa.free(id_key); defer gpa.free(id_key);
try testing.expect(engine2.get_doc("app", "sessions", id_key) == null); try testing.expect(engine2.get_doc("app", "sessions", id_key) == null);
} }
const alive = try bson.serialize_value(gpa, bson.Value{ .int32 = 4 }); const alive = try id_key_for(gpa, bson.Value{ .int32 = 4 });
defer gpa.free(alive); defer gpa.free(alive);
try testing.expect(engine2.get_doc("app", "sessions", alive) != null); try testing.expect(engine2.get_doc("app", "sessions", alive) != null);
} }
@@ -2964,9 +2994,52 @@ test "ttl_sweep spans collections and several TTL indexes on one collection" {
try engine.insert("other", "plain", &plain, &env.gen); try engine.insert("other", "plain", &plain, &env.gen);
try testing.expectEqual(@as(usize, 2), try engine.ttl_sweep(now_ms)); try testing.expectEqual(@as(usize, 2), try engine.ttl_sweep(now_ms));
try testing.expectEqual(@as(usize, 0), engine.get_collection("app", "sessions").?.docs.count()); try testing.expectEqual(@as(usize, 0), engine.get_collection("app", "sessions").?.doc_count);
try testing.expectEqual(@as(usize, 0), engine.get_collection("app", "events").?.docs.count()); try testing.expectEqual(@as(usize, 0), engine.get_collection("app", "events").?.doc_count);
try testing.expectEqual(@as(usize, 1), engine.get_collection("other", "plain").?.docs.count()); try testing.expectEqual(@as(usize, 1), engine.get_collection("other", "plain").?.doc_count);
}
/// Report a replay eviction whose `_id` bytes differ from the incoming record's.
///
/// `_id_` is keyed on the canonical `bson.encode_key`, under which int32 1,
/// int64 1 and double 1.0 are one key -- as they are in MongoDB. A database
/// written before that could legitimately hold two such documents, and replaying
/// it now drops one. That is the intended semantics and a one-way migration, so
/// it has to be said out loud rather than discovered.
///
/// Best effort by design: this runs during replay, where nothing may refuse to
/// start.
fn warn_on_equal_id_collision(
gpa: std.mem.Allocator,
coll: *const Collection,
old_off: u64,
doc: *const bson.Document,
record: storage.Record,
) void {
var arena = std.heap.ArenaAllocator.init(gpa);
defer arena.deinit();
const a = arena.allocator();
const old_id = (bson.get_at(a, coll.doc_bytes(old_off), "_id") catch return) orelse return;
const new_id = doc.get("_id") orelse return;
const old_bytes = bson.serialize_value(a, old_id) catch return;
const new_bytes = bson.serialize_value(a, new_id) catch return;
if (std.mem.eql(u8, old_bytes, new_bytes)) return; // an ordinary replace
std.debug.print(
"multiforadb: WARNING: {s}.{s} holds two documents whose _id values compare " ++
"equal but were stored differently; keeping the later one. This is a " ++
"one-way migration -- _id uniqueness is canonical now, as in MongoDB.\n",
.{ record.db, record.coll },
);
}
/// The canonical `_id` key the engine looks documents up by, owned by the caller.
/// Tests used `bson.serialize_value` when a hashmap keyed on it; `_id_` is keyed
/// on `bson.encode_key`, which is the canonical encoding.
fn id_key_for(gpa: std.mem.Allocator, v: bson.Value) ![]u8 {
var enc: std.ArrayListUnmanaged(u8) = .empty;
errdefer enc.deinit(gpa);
try bson.encode_key(v, gpa, &enc);
return enc.toOwnedSlice(gpa);
} }
/// Whether an offset falls inside one of the collection's current slab extents. /// Whether an offset falls inside one of the collection's current slab extents.