index/db: locate index entries by regeneration, not by scanning

remove_id walks every entry in the index comparing ids, so evicting one
document cost O(index size) per index -- on a 65,536-document collection
that is 65,536 id comparisons to remove a single entry, and it ran on
every update and every delete.

Entry generation is a pure function of the document, so remove_doc
regenerates the entries the document contributed and binary-searches for
each one. evict_doc now removes the document from the docs map first and
hands the document itself to the index, while both it and the map key are
still alive.

  updateMany({k: 7}, {$inc}) over 65,536 documents, measured A/B:
    15.4ms -> 5.5ms   (MongoDB 8.3.7: 6.1ms)

It is infallible by construction. Regeneration allocates and can fail,
and a document the index could not key contributed nothing to remove; in
either case it falls back to the scan, which is always correct. It also
falls back if the regenerated entries are not all found, so a
disagreement degrades to slow rather than leaving a stale index. That
last guard is defensive only -- regeneration is deterministic, so the
test below does not reach it.

The equivalence is checked directly rather than by example: two identical
indexes are built over documents exercising multikey arrays with repeats,
missing fields and both sparse settings, then emptied document by
document in shuffled order -- one through remove_doc, one through
remove_id -- asserting the entry arrays stay byte-identical at every
step. Verified it fails when removal order is reversed, which is the way
positional removal actually breaks.

Insertion still memmoves the tail. That, and ordered leaf iteration, are
what the tree is still for.

Note for later: the updateOne({_id}) and deleteOne({_id}) paths are slow
here for an unrelated reason -- an integer _id is rejected by
value_fast_path_safe, so each one is a full collection scan. The encoded
keys already make that guard unnecessary; removing it belongs with the
_id index.

Verified: 78 unit tests under ReleaseFast and ReleaseSafe, e2e
29/16/17/3/2, the crash pair, e2e6 72/72.

src/db.zig

@@ -131,12 +131,15 @@ pub const Engine = struct {
 
     /// Drop the document stored under `id_key`, freeing it and its key.
     /// No-op when the id is absent. This is the single chokepoint where a
-    /// document dies, so index entries are removed here — before
+    /// document dies, so index entries are removed here — while the
-    /// old.value.deinit() and gpa.free(old.key) — keeping the entry aliasing
+    /// document and the docs map key are both still alive, which is what
-    /// (values into the document arena, id into the docs map key) safe.
+    /// keeps `Entry.id`'s aliasing of that key safe.
+    ///
+    /// 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 {
-        for (coll.indexes.items) |*ix| ix.remove_id(self.gpa, id_key);
         const old = coll.docs.fetchRemove(id_key) orelse return;
+        for (coll.indexes.items) |*ix| ix.remove_doc(self.gpa, old.value, old.key);
         old.value.*.deinit();
         self.gpa.destroy(old.value);
         self.gpa.free(old.key);

src/index.zig

@@ -298,6 +298,52 @@ pub const Index = struct {
         self.entries.items.len = w;
     }
 
+    /// Remove the entries `doc` contributed, locating each by binary search
+    /// instead of scanning the array.
+    ///
+    /// Entry generation is a pure function of the document, so the entries
+    /// to remove can be regenerated and looked up directly. `remove_id`
+    /// walks every entry in the index comparing ids, which made a single
+    /// document update cost O(index size) per index — the dominant cost of
+    /// updateMany and deleteMany on a large collection.
+    ///
+    /// Infallible by construction: regeneration allocates and can fail, and
+    /// a document the index cannot key contributed nothing to remove, so
+    /// either way it falls back to the scan, which is always correct.
+    pub fn remove_doc(self: *Index, gpa: std.mem.Allocator, doc: *const bson.Document, id: []const u8) void {
+        var built = self.build_entries(gpa, doc, id) catch return self.remove_id(gpa, id);
+        defer built.deinit(gpa);
+        // Sparse index that skipped this document: nothing was inserted.
+        if (built.entries.items.len == 0) return;
+
+        // Positions must be removed high-to-low so the earlier ones stay
+        // valid. build_entries returns its entries sorted, and the array is
+        // in the same order, so positions come out ascending already.
+        var found: [max_index_keys]usize = undefined;
+        var n: usize = 0;
+        for (built.entries.items) |want| {
+            const pos = std.sort.lowerBound(Entry, self.entries.items, want, compare_entries);
+            if (pos >= self.entries.items.len) continue;
+            const have = self.entries.items[pos];
+            // compare_entries orders by key then id, so an exact hit means
+            // this is the very entry the document contributed.
+            if (compare_entries(have, want) != .eq) continue;
+            if (n == found.len) return self.remove_id(gpa, id); // more entries than we can track
+            found[n] = pos;
+            n += 1;
+        }
+        // A document that generated entries we cannot find would leave the
+        // index stale, which is worse than being slow.
+        if (n != built.entries.items.len) return self.remove_id(gpa, id);
+
+        var i = n;
+        while (i > 0) {
+            i -= 1;
+            gpa.free(self.entries.items[found[i]].key);
+            _ = self.entries.orderedRemove(found[i]);
+        }
+    }
+
     /// Reject when any of `new_entries` has a key already present under a
     /// different id. Entries with `exclude_id` (the replacing document's
     /// own old entries) are allowed.
@@ -1207,6 +1253,94 @@ test "compound index prefix search and range on the next key" {
     try testing.expect(std.mem.eql(u8, out.items[0], "c1") or std.mem.eql(u8, out.items[0], "c2"));
 }
 
+test "remove_doc leaves the index identical to a full scan removal" {
+    // remove_doc finds entries by regenerating them from the document
+    // instead of scanning for the id. If regeneration ever disagreed with
+    // what was inserted, entries would be left behind and the index would
+    // silently return stale ids -- so check it against the scan directly,
+    // over documents that exercise multikey arrays, missing fields and
+    // duplicate values.
+    const gpa = testing.allocator;
+    var prng = std.Random.DefaultPrng.init(0xDEADBEEF);
+    const rand = prng.random();
+
+    for ([_]bool{ false, true }) |sparse| {
+        var by_doc = try simple_index(gpa, &.{ "a", "b" }, false, sparse);
+        defer by_doc.deinit(gpa);
+        var by_scan = try simple_index(gpa, &.{ "a", "b" }, false, sparse);
+        defer by_scan.deinit(gpa);
+
+        const n = 120;
+        var ids: std.ArrayListUnmanaged([]u8) = .empty;
+        defer {
+            for (ids.items) |x| gpa.free(x);
+            ids.deinit(gpa);
+        }
+        var arrays: [n][3]bson.Value = undefined;
+        var pairs: [n][2]bson.Pair = undefined;
+        var docs: [n]bson.Document = undefined;
+
+        for (0..n) |i| {
+            const id = try std.fmt.allocPrint(gpa, "id{d}", .{i});
+            try ids.append(gpa, id);
+            const shape = rand.intRangeAtMost(u8, 0, 3);
+            const av = rand.intRangeAtMost(i32, 0, 3);
+            var np: usize = 0;
+            switch (shape) {
+                // A plain value.
+                0 => {
+                    pairs[i][0] = .{ .key = "a", .value = .{ .int32 = av } };
+                    pairs[i][1] = .{ .key = "b", .value = .{ .int32 = rand.intRangeAtMost(i32, 0, 3) } };
+                    np = 2;
+                },
+                // Multikey: an array, sometimes with repeats.
+                1 => {
+                    arrays[i] = .{ .{ .int32 = av }, .{ .int32 = av }, .{ .int32 = av + 1 } };
+                    pairs[i][0] = .{ .key = "a", .value = .{ .array = arrays[i][0..3] } };
+                    pairs[i][1] = .{ .key = "b", .value = .{ .int32 = rand.intRangeAtMost(i32, 0, 3) } };
+                    np = 2;
+                },
+                // Missing "b": indexed as null, or skipped when sparse.
+                2 => {
+                    pairs[i][0] = .{ .key = "a", .value = .{ .int32 = av } };
+                    np = 1;
+                },
+                // Missing both.
+                else => np = 0,
+            }
+            docs[i] = doc_of(pairs[i][0..np]);
+            try by_doc.append_doc_entries(gpa, &docs[i], id);
+            try by_scan.append_doc_entries(gpa, &docs[i], id);
+        }
+        _ = try by_doc.finish_bulk(false);
+        _ = try by_scan.finish_bulk(false);
+        try testing.expectEqual(by_scan.entries.items.len, by_doc.entries.items.len);
+
+        // Remove in a shuffled order, so removals interleave rather than
+        // peeling the array from one end.
+        var order: [n]usize = undefined;
+        for (0..n) |i| order[i] = i;
+        rand.shuffle(usize, &order);
+
+        for (order) |i| {
+            by_doc.remove_doc(gpa, &docs[i], ids.items[i]);
+            by_scan.remove_id(gpa, ids.items[i]);
+
+            testing.expectEqual(by_scan.entries.items.len, by_doc.entries.items.len) catch |err| {
+                std.debug.print("sparse={} after removing doc {d}: scan left {d}, remove_doc left {d}\n", .{
+                    sparse, i, by_scan.entries.items.len, by_doc.entries.items.len,
+                });
+                return err;
+            };
+            for (by_doc.entries.items, by_scan.entries.items) |x, y| {
+                try testing.expect(std.mem.eql(u8, x.key, y.key));
+                try testing.expect(std.mem.eql(u8, x.id, y.id));
+            }
+        }
+        try testing.expectEqual(@as(usize, 0), by_doc.entries.items.len);
+    }
+}
+
 test "lookup_range matches a brute-force filter over random data" {
     const gpa = testing.allocator;