diff --git a/src/db.zig b/src/db.zig index bf6fbf6..be696d3 100644 --- a/src/db.zig +++ b/src/db.zig @@ -34,7 +34,10 @@ pub const Collection = struct { /// Removed documents leave garbage bytes until a rebuild rewrites them. A /// checkpoint must never renumber these offsets: every index leaf holds one /// (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. pager: *pgr.Pager, /// 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 { var self: Collection = .{ - .docs = .empty, + .doc_count = 0, .pager = pager, .slab_extents = .empty, .slab_tail = 0, @@ -346,20 +349,15 @@ pub const Engine = struct { // never be smaller than this one's -- and a u64 underflow here would // read as an astronomically large live count, permanently suppressing // compaction rather than crashing. - assert_msg(self.live_docs >= coll.docs.count(), "dropping a collection would underflow the engine's live count"); - self.live_docs -= coll.docs.count(); - self.dead_docs += coll.docs.count(); + assert_msg(self.live_docs >= coll.doc_count, "dropping a collection would underflow the engine's live count"); + self.live_docs -= coll.doc_count; + self.dead_docs += coll.doc_count; coll.id_index.deinit(self.gpa); for (coll.indexes.items) |ix| { ix.deinit(self.gpa); self.gpa.destroy(ix); } 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 // later, so a fallback to the previous image still finds them intact. 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 /// regenerating them from it, which is far cheaper than scanning. - fn evict_doc(self: *Engine, coll: *Collection, id_key: []const u8) void { - const old = coll.docs.fetchRemove(id_key) orelse return; + /// `id_enc` is `bson.encode_key` of the document's `_id` -- the canonical + /// 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 // index removal, so the slice is safe for the call. - const old_bytes = coll.doc_bytes(old.value); - // Entries are keyed by the document's slab offset now, which is exactly - // what the map just gave us. - 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); + const old_bytes = coll.doc_bytes(off); + coll.id_index.remove_doc(self.gpa, old_bytes, off); + for (coll.indexes.items) |ix| ix.remove_doc(self.gpa, old_bytes, off); // 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"); self.live_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) --------------------- @@ -654,10 +664,12 @@ pub const Engine = struct { var id_arena = std.heap.ArenaAllocator.init(self.gpa); defer id_arena.deinit(); 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. - const id_key = try bson.serialize_value(self.gpa, id_value); - var stored = false; - errdefer if (!stored) self.gpa.free(id_key); + // The canonical encoding, because that is what `_id_` is keyed on. It + // used to be serialize_value, for a hashmap that no longer exists. + var id_enc_list: std.ArrayListUnmanaged(u8) = .empty; + 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; // 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 // document's *current* slab offset, since that is what its existing // 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| { if (!b.ix.unique) continue; 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); // 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 // slab and record the offset. Infallible from here. const off = coll.slab_append(doc_bytes); - try coll.docs.put(self.gpa, id_key, off); self.live_docs += 1; + coll.doc_count += 1; for (built_list.items) |*b| { if (b.built.multikey) b.ix.multikey = true; b.ix.insert_entries(&b.built, off); } - stored = true; // The write is published; anything the reservations above did not claim // is dead. Leaving it promised would grow the file on every write. self.pager.release_reservation(); @@ -754,15 +765,17 @@ pub const Engine = struct { coll_name: []const u8, id: bson.Value, ) !bool { - const id_key = try bson.serialize_value(self.gpa, id); - defer self.gpa.free(id_key); - return self.remove(db_name, coll_name, id_key); + var enc: std.ArrayListUnmanaged(u8) = .empty; + defer enc.deinit(self.gpa); + 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 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 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 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 // never compacts, because only upsert and ttl_sweep used to check. self.note_compact(); @@ -793,10 +806,11 @@ pub const Engine = struct { self: *Engine, db_name: []const u8, coll_name: []const u8, - id_key: []const u8, + /// `bson.encode_key` of the `_id`. + id_enc: []const u8, ) ?[]const u8 { 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); } @@ -857,9 +871,9 @@ pub const Engine = struct { // document into a sorted array memmoves the tail every time, which // is what made this quadratic. On any failure the deferred // 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| { - 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); self.pager.release_reservation(); @@ -985,8 +999,9 @@ pub const Engine = struct { var removed: usize = 0; for (offs.items) |off| { 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); - if (try self.remove(db_name, coll_name, id_key)) removed += 1; + var enc: std.ArrayListUnmanaged(u8) = .empty; + try bson.encode_key(id_value, arena.allocator(), &enc); + if (try self.remove(db_name, coll_name, enc.items)) removed += 1; } 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 /// it was copied to. Named rather than anonymous so the rebuild and the /// 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 /// pages. @@ -1199,18 +1214,19 @@ pub const Engine = struct { // so a later scan reads it sequentially. var moved: std.ArrayListUnmanaged(Moved) = .empty; 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| { - 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); const new_off = coll.slab_append(bytes); 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. - for (moved.items) |m| coll.docs.putAssumeCapacity(m.key, m.off); // Rebuild every index from the new offsets, bulk-packed. 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. fn rebuild_index(self: *Engine, coll: *Collection, ix: *index.Index) !void { if (ix.count() > 0) return; // defensive - var doc_it = coll.docs.iterator(); - while (doc_it.next()) |doc_entry| { - ix.append_doc_entries(self.gpa, coll.doc_bytes(doc_entry.value_ptr.*), doc_entry.value_ptr.*) catch |err| switch (err) { + // The `_id_` tree is the enumeration of live documents now. It is also + // ordered, so this reads the slab sequentially where the hashmap read it + // 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 => { std.debug.print( "multiforadb: WARNING: index '{s}' cannot index an existing " ++ @@ -1441,11 +1460,12 @@ pub const Engine = struct { }; try coll.indexes.append(self.gpa, boxed); } - // The docs map is still the authoritative _id -> offset lookup, - // and it is not in the catalog on purpose: the commit that drops - // it would only have to delete the format again. Rebuilt from the - // _id_ tree instead, which the data file already holds. - try self.rebuild_docs_map(coll); + // Nothing else to rebuild. The `_id_` tree *is* the lookup, and + // it is already in the file -- which is the whole reason an open + // no longer touches a single document page. The version of this + // that kept a hashmap had to read every document here to recover + // 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); } - /// Rebuild the _id -> offset hashmap by walking the _id_ tree. fn rebuild_docs_map(self: *Engine, coll: *Collection) !void { var arena = std.heap.ArenaAllocator.init(self.gpa); 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", .{}); return; }; - const id_key = try bson.serialize_value(self.gpa, id_value); - var key_owned = false; - defer if (!key_owned) self.gpa.free(id_key); + var id_enc_list: std.ArrayListUnmanaged(u8) = .empty; + defer id_enc_list.deinit(self.gpa); + 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 // 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) { 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); defer self.gpa.free(doc_bytes); try coll.slab_reserve(self.gpa, doc_bytes.len); const off = coll.slab_append(doc_bytes); - try coll.docs.put(self.gpa, id_key, off); 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) { self.index_doc_on_replay(coll, doc_bytes, off); + } else { + self.index_one(&coll.id_index, doc_bytes, off); } self.pager.release_reservation(); // 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 // 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 => {}, } } @@ -1795,7 +1829,7 @@ test "insert, query, remove" { engine.unlock(); // 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); try engine.lock(); 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); defer engine2.deinit(); 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); 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); try testing.expectEqualStrings("alice", (try bson.get_at(gpa, engine2.get_doc("app", "users", id_key1).?, "name")).?.string); engine2.unlock(); @@ -1971,11 +2005,11 @@ test "auto _id generation survives reopen" { var engine2 = try Engine.open(gpa, io, tmp.path); defer engine2.deinit(); const coll = engine2.get_collection("app", "no_ids").?; - var it = coll.docs.iterator(); + var it = coll.id_index.iter(); var count: usize = 0; while (it.next()) |entry| { 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.expectEqual(@as(usize, 1), count); @@ -2023,7 +2057,7 @@ test "compaction rewrites log and keeps data" { defer engine3.deinit(); try engine3.lock(); 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); 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(); if (e.get_collection("app", "users")) |coll| { var n: usize = 0; - var it = coll.docs.iterator(); + var it = coll.id_index.iter(); while (it.next()) |_| n += 1; // A reader must never observe more docs than can exist. 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(); defer engine.unlock_read(); 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| { - 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); 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(); defer reopened.unlock_read(); 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| { - 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); 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(); try testing.expect(engine.pager.loaded.generation >= 1); 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()); // And the sequence continued from the watermark rather than restarting. 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(); defer engine.unlock(); const coll = engine.get_collection("app", "users").?; - try testing.expectEqual(@as(usize, 46), coll.docs.count()); - const id_key = try bson.serialize_value(gpa, .{ .int32 = 500 }); + try testing.expectEqual(@as(usize, 46), coll.id_index.count()); + const id_key = try id_key_for(gpa, .{ .int32 = 500 }); defer gpa.free(id_key); 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(); defer engine.unlock(); 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()); - 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); 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); 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. try testing.expect(coll.slab_used < before_used); 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()); // 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(); i = 0; 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); var enc: std.ArrayListUnmanaged(u8) = .empty; defer enc.deinit(gpa); try bson.encode_key(.{ .int32 = i }, gpa, &enc); 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 index and the map must agree, which catches either one being left - // 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. + // 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 + // rebuild that left stale offsets behind shows up here and nowhere else. 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); } @@ -2811,7 +2841,7 @@ test "drop_collection frees indexes; log without index records replays" { var engine2 = try Engine.open(gpa, io, tmp.path); defer engine2.deinit(); 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); try testing.expect(engine2.get_doc("app", "plain", id_key) != null); // 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); } 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()); // 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, 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, 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. @@ -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)); // 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, 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 @@ -2901,16 +2931,16 @@ test "ttl_sweep deletes expired documents and the deletion survives reopen" { try engine2.lock(); defer engine2.unlock(); 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(?i64, 60), coll.indexes.items[0].ttl); try testing.expectEqual(@as(usize, 3), coll.indexes.items[0].count()); 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); 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); 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 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", "events").?.docs.count()); - try testing.expectEqual(@as(usize, 1), engine.get_collection("other", "plain").?.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").?.doc_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.