index/commands: stream whole-index scans; add a reverse leaf iterator

A whole-index read used to materialize every candidate before the caller saw
the first one. At the tens-of-GB target that is a list of every offset in the
collection -- ~160 MB for a countDocuments({}) over 20 million documents --
which defeats the point of moving storage to disk. Cursors are M1, but
*streaming a scan* has to exist now.

`Candidates` is the one loop candidates arrive through, whatever produced them:
a plan's materialized lookups, or the index read end to end. That keeps this
file's governing invariant -- an index only generates candidates, the full
filter is re-applied to every one -- in a single place. A narrowed plan still
materializes, because its multikey/$in dedupe genuinely needs the whole set and
is bounded by selectivity.

`RevIter` walks `Node.prev`, which has always been maintained and which nothing
had ever read: a descending scan materialized the whole index and reversed the
list. `find({}).sort({_id:-1}).limit(20)` becomes O(20).

The unfiltered fallback now walks the _id_ index instead of the docs map. That
is ordered rather than hash-ordered, and it does not depend on a structure that
is about to be deleted.

`Plan.full_scan()` refuses multikey indexes, since one document contributes
several entries there and a stream cannot dedupe what `search` did. The check is
currently redundant -- the planner refuses to order a multikey index anyway --
and is kept because the two guards protect different things. Stated precisely
in both places after checking: the commands.zig test reddens only when *both*
guards are removed, which is what that test actually pins.

--

This also broke e2e6's compaction check, and the fix there is the more
interesting half.

The check required peak/final > 1.4 and got 1.28. The final size was identical
to the byte (2,398,065 vs 2,398,064) -- compaction reclaimed exactly as before
-- and only the peak moved. Isolated to one variable: changing just the order
updateMany({}) walks its matches moves peak/final between 1.65 and 1.28, because
compaction can also fire from the once-per-second TTL monitor and whether one
lands inside the batch shifts the peak a long way while leaving the outcome
unchanged. The threshold was measuring the schedule.

Replaced with `peak > final`, which measures the shape instead: an append-only
log grows monotonically, so its maximum *is* its final size, and a file that
was ever larger than it ended can only have been rewritten.

Worth recording why the obvious alternative does not work. An absolute size
bound cannot distinguish a working compactor here: the payload is one repeated
character, so ~48 MB of records LZ4-compress to ~3 MB whether or not anything is
reclaimed -- with compaction disabled entirely the file still ends at 3.1 MB. I
first wrote the comment claiming that bound was the strong one, then measured it
and found the opposite; `peak > final` is what goes red.

src/commands.zig

@@ -760,27 +760,41 @@ fn scan_sorted(
     // superseded-but-parseable bytes that the re-applied filter might accept.
     // Loud beats silent: a wrong answer a test can see beats a missing
     // candidate nothing can.
-    if (try index.plan(ctx.gpa, &coll.id_index, coll.indexes.items, filter, sort)) |p| {
+    // Candidates arrive as a stream so that a whole-index read never
-        var plan = p;
+    // materializes: at the tens-of-GB target a `countDocuments({})` would
-        defer plan.deinit(ctx.gpa);
+    // otherwise build a list of every offset in the collection before the
-        var offs: std.ArrayListUnmanaged(u64) = .empty;
+    // first one is examined. A narrowed plan still materializes, because its
-        defer offs.deinit(ctx.gpa);
+    // multikey/$in dedupe genuinely needs the whole set, and it is bounded by
-        try plan.search(ctx.gpa, &offs);
+    // selectivity.
+    var offs: std.ArrayListUnmanaged(u64) = .empty;
+    defer offs.deinit(ctx.gpa);
+    var cands: index.Candidates = undefined;
+    var plan_opt = try index.plan(ctx.gpa, &coll.id_index, coll.indexes.items, filter, sort);
+    defer if (plan_opt) |*p| p.deinit(ctx.gpa);
+
+    if (plan_opt) |*plan| {
         if (sorted) |flag| flag.* = plan.provides_sort;
         if (plan.provides_sort) lim = limit;
-        for (offs.items) |off| {
+        if (plan.full_scan()) {
-            if (!try query.matches_bytes(ctx.gpa, filter, coll.doc_bytes(off))) continue;
+            cands = if (plan.backward)
-            if (out) |list| try list.append(ctx.gpa, off);
+                .{ .scan_rev = plan.index.iter_reverse() }
-            n += 1;
+            else
-            if (lim != 0 and n >= lim) break;
+                .{ .scan = plan.index.iter() };
+        } else {
+            try plan.search(ctx.gpa, &offs);
+            cands = .{ .list = .{ .items = offs.items } };
         }
-        return n;
+    } else {
+        // No usable predicate: every document, in _id order. The docs map was
+        // the fallback here, and its iteration order was the hash's; walking
+        // the _id_ index instead is ordered, streams, and does not depend on a
+        // structure that is going away.
+        cands = .{ .scan = coll.id_index.iter() };
     }
 
-    var it = coll.docs.iterator();
+    while (cands.next()) |off| {
-    while (it.next()) |entry| {
+        if (!try query.matches_bytes(ctx.gpa, filter, coll.doc_bytes(off))) continue;
-        if (!try query.matches_bytes(ctx.gpa, filter, coll.doc_bytes(entry.value_ptr.*))) continue;
+        if (out) |list| try list.append(ctx.gpa, off);
-        if (out) |list| try list.append(ctx.gpa, entry.value_ptr.*);
         n += 1;
         if (lim != 0 and n >= lim) break;
     }
@@ -1096,8 +1110,13 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
         _ = try scan_matching(ctx, db_name, coll_name, filter, 0, &offs);
         stages = stages[1..];
     } else {
-        var it = coll.docs.iterator();
+        // Every document, in _id order. A pipeline materializes its stream
-        while (it.next()) |entry| try offs.append(ctx.gpa, entry.value_ptr.*);
+        // anyway (stages need random access to the window), so this one stays a
+        // list -- but it comes from the _id_ index rather than the docs map,
+        // which is ordered and does not depend on a structure that is going
+        // away. Streaming the whole pipeline is M1's cursor work.
+        var it = coll.id_index.iter();
+        while (it.next()) |e| try offs.append(ctx.gpa, e.off);
     }
 
     var start: usize = 0;
@@ -2384,6 +2403,65 @@ test "count_only_pipeline accepts only shapes a count can answer" {
     try testing.expect(try count_only_pipeline(&reply, &.{}) == null);
 }
 
+test "a sorted full scan over a multikey index returns each document once" {
+    // scan_sorted streams a whole-index read instead of materializing it, which
+    // is what keeps a countDocuments({}) from building a list of every offset in
+    // the collection. But one document contributes several entries to a multikey
+    // index, so walking that index end to end yields it once per array element.
+    // The materializing path deduped; a stream cannot, so Plan.full_scan()
+    // refuses multikey indexes and this shape keeps materializing.
+    //
+    // The shape is `find({}).sort({tags: 1})`: no filter, so the only reason to
+    // use an index at all is that it supplies the order.
+    //
+    // Mutation check, and it is worth stating precisely because the obvious
+    // version of it does nothing: two independent guards refuse this, so
+    // removing either one alone leaves the test green. `index_provides_sort`
+    // returns null for a multikey index, and `Plan.full_scan` refuses one
+    // again. Remove *both* and each document comes back three times. So this
+    // test pins the pair, not either guard -- which is the useful property,
+    // since it is the behaviour that matters rather than which check delivers
+    // it.
+    var threaded: std.Io.Threaded = .init_single_threaded;
+    defer threaded.deinit();
+    const io = threaded.io();
+    var tdb = try TestDb.init(io);
+    defer tdb.deinit();
+
+    try dispatch_insert(&tdb, io, "mk", &.{
+        .{ .doc = &.{
+            .{ .key = "_id", .value = .{ .int32 = 1 } },
+            .{ .key = "tags", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 2 }, .{ .int32 = 3 } } } },
+        } },
+        .{ .doc = &.{
+            .{ .key = "_id", .value = .{ .int32 = 2 } },
+            .{ .key = "tags", .value = .{ .array = &.{ .{ .int32 = 4 }, .{ .int32 = 5 }, .{ .int32 = 6 } } } },
+        } },
+    });
+    try dispatch_create_index(&tdb, io, "mk", .{ .doc = &.{
+        .{ .key = "key", .value = .{ .doc = &.{.{ .key = "tags", .value = .{ .int32 = 1 } }} } },
+        .{ .key = "name", .value = .{ .string = "tags_1" } },
+    } });
+
+    var ctx = tdb.ctx(io);
+    var msg = try parse_fake_msg("find", .{ .string = "mk" }, &.{
+        .{ .key = "filter", .value = .{ .doc = &.{} } },
+        .{ .key = "sort", .value = .{ .doc = &.{.{ .key = "tags", .value = .{ .int32 = 1 } }} } },
+    });
+    defer msg.deinit();
+    var reply = wire.Reply.init(testing.allocator);
+    defer reply.deinit();
+    try dispatch(&ctx, &msg, &reply);
+    try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply.pairs.items, "ok").?.double);
+    const cursor = bson.get_pair(reply.pairs.items, "cursor") orelse return error.TestUnexpectedResult;
+    const batch = switch (bson.get_pair(cursor.doc, "firstBatch") orelse return error.TestUnexpectedResult) {
+        .array => |arr| arr,
+        else => return error.TestUnexpectedResult,
+    };
+    // Two documents, not six.
+    try testing.expectEqual(@as(usize, 2), batch.len);
+}
+
 test "a command with no collection name errors and holds no lock" {
     // Regression for a leaked catalog lock. dispatch resolved the namespace
     // *after* taking the catalog lock, with `orelse return` -- and a plain

src/index.zig

@@ -706,6 +706,55 @@ pub const Index = struct {
         return .{ .ix = self, .leaf = self.first_leaf, .slot = 0 };
     }
 
+    /// Reverse ordered iteration. Leaves are doubly linked and `prev` has
+    /// always been maintained -- nothing walked it until now, so a descending
+    /// scan had to materialize every candidate and reverse the list. This turns
+    /// `find({}).sort({_id: -1}).limit(20)` from O(collection) into O(20).
+    pub const RevIter = struct {
+        ix: *const Index,
+        leaf: u32,
+        /// One past the slot to yield next, so 0 means this leaf is done.
+        slot: u32,
+
+        pub fn next(self: *RevIter) ?EntryRef {
+            const ix = self.ix;
+            while (self.leaf != 0) {
+                if (self.slot > 0) {
+                    self.slot -= 1;
+                    return .{
+                        .key = ix.key_of(self.leaf, self.slot),
+                        .off = ix.off_of(self.leaf, self.slot),
+                    };
+                }
+                const prev = ix.page(self.leaf).prev;
+                self.leaf = prev;
+                if (prev != 0) self.slot = ix.page(prev).count;
+            }
+            return null;
+        }
+    };
+
+    pub fn iter_reverse(self: *const Index) RevIter {
+        const last = self.descend_last();
+        return .{ .ix = self, .leaf = last, .slot = self.page(last).count };
+    }
+
+    /// The rightmost leaf. An internal node's children are `first_child`
+    /// followed by one per separator, so the last child is the last slot's
+    /// `extra` (or `first_child` when the node holds no separators -- which
+    /// removal can leave behind, since it never rebalances).
+    fn descend_last(self: *const Index) u32 {
+        var node_id = self.root;
+        while (self.page(node_id).is_leaf == 0) {
+            const node = self.page(node_id);
+            node_id = if (node.count == 0)
+                node.first_child
+            else
+                get_slot(node, node.count - 1).extra;
+        }
+        return node_id;
+    }
+
     /// Ordered iteration starting at the first entry whose key is not less
     /// than `prefix` (prefix semantics). Used by the equality/range
     /// searches and the TTL sweep's datetime band.
@@ -1681,6 +1730,33 @@ fn default_name(gpa: std.mem.Allocator, key_pairs: []const bson.Pair) ![]u8 {
 /// independent, and sorting handles direction itself.
 /// Entry ordering. No tie-break beyond the key: entries no longer carry a
 /// document, and within one document's batch equal keys are deduped anyway.
+/// A stream of candidate slab offsets, however they were produced: a plan's
+/// materialized lookups, or the index read end to end. The point is that the
+/// consumer is one loop, so the governing invariant of this file -- an index
+/// only generates candidates, the full filter is re-applied to every one --
+/// lives in exactly one place regardless of which shape produced them.
+///
+/// A `.scan`/`.scan_rev` holds an iterator positioned in the tree, so it must
+/// be drained (or dropped) before this collection is written to. The offsets it
+/// yields are values and are unaffected by any later mutation.
+pub const Candidates = union(enum) {
+    scan: Index.Iter,
+    scan_rev: Index.RevIter,
+    list: struct { items: []const u64, i: usize = 0 },
+
+    pub fn next(self: *Candidates) ?u64 {
+        switch (self.*) {
+            .scan => |*it| return if (it.next()) |e| e.off else null,
+            .scan_rev => |*it| return if (it.next()) |e| e.off else null,
+            .list => |*l| {
+                if (l.i >= l.items.len) return null;
+                defer l.i += 1;
+                return l.items[l.i];
+            },
+        }
+    }
+};
+
 pub fn compare_entries(a: Entry, b: Entry) std.math.Order {
     return std.mem.order(u8, a.key, b.key);
 }
@@ -1839,6 +1915,30 @@ pub const Plan = struct {
     /// That order is the reverse of the index's.
     backward: bool = false,
 
+    /// Whether this plan reads the whole index with no narrowing, which is
+    /// the case a streaming leaf walk can answer without materializing
+    /// anything. At the M0 scale that matters: a `countDocuments({})` over
+    /// 20 million documents would otherwise build a 160 MB list of offsets
+    /// before the caller sees the first one.
+    ///
+    /// Multikey indexes are excluded. One document contributes several entries
+    /// there, so a full walk yields it more than once -- `search` dedupes, a
+    /// stream cannot.
+    ///
+    /// That check is currently redundant: `index_provides_sort` refuses
+    /// multikey, and a run-0 plan with no range is only formed when it supplies
+    /// the sort, so no multikey plan reaches here today. Kept deliberately --
+    /// the two guards protect different things, and if the planner ever learns
+    /// to order a multikey index, streaming must not silently start returning
+    /// duplicates. Verified by removing both: the commands.zig test
+    /// "a sorted full scan over a multikey index returns each document once"
+    /// goes red only then, which is the honest statement of what pins this.
+    pub fn full_scan(self: *const Plan) bool {
+        if (self.index.multikey) return false;
+        if (self.lo != null or self.hi != null) return false;
+        return self.lookup_keys.items.len == 1 and self.lookup_keys.items[0].len == 0;
+    }
+
     pub fn deinit(self: *Plan, gpa: std.mem.Allocator) void {
         for (self.lookup_keys.items) |k| gpa.free(k);
         self.lookup_keys.deinit(gpa);
@@ -2272,6 +2372,107 @@ test "unique conflict across documents, replace of own entries allowed" {
     try expect_offs(gpa, &ix, &.{.{ .int32 = 10 }}, &.{});
 }
 
+test "iter_reverse yields every entry in exact reverse order" {
+    // Node.prev has always been maintained and nothing walked it, so a
+    // descending scan materialized the whole index and reversed the list. The
+    // interesting cases are structural, not arithmetic: enough entries to build
+    // several leaves and an interior level, so descend_last has to follow the
+    // last separator's child rather than first_child, and a run of removals
+    // afterwards because removal never rebalances -- it can leave an internal
+    // node with no separators at all, which is the branch descend_last would
+    // otherwise get wrong.
+    //
+    // Mutation checks: start iter_reverse at `first_leaf` and the order is
+    // wrong from the first entry; make descend_last follow `first_child`
+    // unconditionally and it silently misses everything to the right.
+    const gpa = testing.allocator;
+    var ix = try simple_index(gpa, &.{"a"}, false, false);
+    defer ix.deinit(gpa);
+
+    const n = 400;
+    for (0..n) |i| {
+        const d = try bytes_of(gpa, &.{
+            .{ .key = "_id", .value = .{ .int32 = @intCast(i + 1) } },
+            .{ .key = "a", .value = .{ .int32 = @intCast(i + 1) } },
+        });
+        defer gpa.free(d);
+        _ = try ix.add_doc(gpa, d, @intCast(i + 1), false);
+    }
+    try testing.expect(ix.depth >= 1); // an interior level exists
+
+    var fwd: std.ArrayListUnmanaged(u64) = .empty;
+    defer fwd.deinit(gpa);
+    var it = ix.iter();
+    while (it.next()) |e| try fwd.append(gpa, e.off);
+    try testing.expectEqual(@as(usize, n), fwd.items.len);
+
+    var rev: std.ArrayListUnmanaged(u64) = .empty;
+    defer rev.deinit(gpa);
+    var rit = ix.iter_reverse();
+    while (rit.next()) |e| try rev.append(gpa, e.off);
+
+    try testing.expectEqual(fwd.items.len, rev.items.len);
+    for (fwd.items, 0..) |off, i| {
+        try testing.expectEqual(off, rev.items[rev.items.len - 1 - i]);
+    }
+
+    // Now churn: delete most of it and re-check, since removal reshapes the
+    // interior without rebalancing.
+    for (0..n) |i| {
+        if (i % 4 == 0) continue;
+        ix.remove_off(@intCast(i + 1));
+    }
+    fwd.clearRetainingCapacity();
+    rev.clearRetainingCapacity();
+    var it2 = ix.iter();
+    while (it2.next()) |e| try fwd.append(gpa, e.off);
+    var rit2 = ix.iter_reverse();
+    while (rit2.next()) |e| try rev.append(gpa, e.off);
+    try testing.expectEqual(fwd.items.len, rev.items.len);
+    for (fwd.items, 0..) |off, i| {
+        try testing.expectEqual(off, rev.items[rev.items.len - 1 - i]);
+    }
+}
+
+test "Candidates streams the same offsets a materialized plan would" {
+    // The streaming path and the materializing path must be interchangeable,
+    // because scan_sorted picks between them on a property of the plan. Compare
+    // them directly rather than trusting that.
+    //
+    // The multikey hazard full_scan() guards against lives in commands.zig's
+    // "a sorted full scan over a multikey index returns each document once",
+    // not here -- this index is not multikey. See Plan.full_scan for why that
+    // test only reddens when *both* multikey guards are removed.
+    const gpa = testing.allocator;
+    var ix = try simple_index(gpa, &.{"a"}, false, false);
+    defer ix.deinit(gpa);
+    for (0..50) |i| {
+        const d = try bytes_of(gpa, &.{
+            .{ .key = "_id", .value = .{ .int32 = @intCast(i + 1) } },
+            .{ .key = "a", .value = .{ .int32 = @intCast(i % 7) } },
+        });
+        defer gpa.free(d);
+        _ = try ix.add_doc(gpa, d, @intCast(i + 1), false);
+    }
+
+    var streamed: std.ArrayListUnmanaged(u64) = .empty;
+    defer streamed.deinit(gpa);
+    var c: Candidates = .{ .scan = ix.iter() };
+    while (c.next()) |off| try streamed.append(gpa, off);
+
+    var listed: std.ArrayListUnmanaged(u64) = .empty;
+    defer listed.deinit(gpa);
+    try ix.lookup_eq(gpa, &.{}, &listed);
+
+    var c2: Candidates = .{ .list = .{ .items = listed.items } };
+    var from_list: std.ArrayListUnmanaged(u64) = .empty;
+    defer from_list.deinit(gpa);
+    while (c2.next()) |off| try from_list.append(gpa, off);
+
+    try testing.expectEqualSlices(u64, listed.items, streamed.items);
+    try testing.expectEqualSlices(u64, listed.items, from_list.items);
+}
+
 test "lookup_exact matches whole keys only" {
     // The point of exact byte equality rather than cmp_prefix: this answers
     // "is this document present", which is what the engine will ask once the

tests/e2e/e2e6.js

@@ -288,10 +288,26 @@ async function phase2(client, { users, aliceId }) {
   // reclaimed it). ~48 MB of records are written here and ~12 MB of it
   // survives, so without compaction the file would end near 48 MB.
   //
-  // Both bounds are relative to the live size on purpose: compaction now
+  // `peak > final` is the assertion with teeth, and it is worth saying why,
-  // triggers on the share of the log that is garbage rather than on bytes
+  // because the intuitive alternative does not work here. The log is
-  // appended, so the absolute peak depends on when that share crosses the
+  // LZ4-compressed and this payload is one repeated character, so ~48 MB of
-  // threshold and is not a stable number to assert on.
+  // records compress to ~3 MB whether or not anything is reclaimed -- with
+  // compaction disabled entirely the file still ends at only 3.1 MB. So no
+  // absolute size bound distinguishes a working compactor from a broken one at
+  // this scale; the 24 MB below is a sanity guard, nothing more.
+  //
+  // What does distinguish them is the shape: an append-only log grows
+  // monotonically, so its maximum *is* its final size. A file that was ever
+  // larger than it ended can only have been rewritten. Verified by disabling
+  // compaction: final 3.1 MB against a peak of 2.7 MB, and this check goes red.
+  //
+  // The previous form required peak > final * 1.4, which measured the schedule
+  // rather than the engine: compaction also fires from the once-per-second TTL
+  // monitor, so whether one lands inside this batch moves the peak a long way
+  // while leaving the result identical. Measured by changing only the order
+  // updateMany({}) walks its matches -- hash order gave 1.65, _id order 1.28,
+  // and the final sizes differed by one byte. Any threshold between those two
+  // fails for a reason that has nothing to do with compaction.
   let peakSize = fs.statSync(DBFILE).size;
   const watcher = setInterval(() => {
     const s = fs.statSync(DBFILE).size;
@@ -305,6 +321,9 @@ async function phase2(client, { users, aliceId }) {
   clearInterval(watcher);
 
   const logSize = fs.statSync(DBFILE).size;
+  if (process.env.E2E6_DEBUG || process.env.E2E6_PEAK) {
+    console.log(`DEBUG phase2 peak=${peakSize} final=${logSize} ratio=${(peakSize / logSize).toFixed(2)}`);
+  }
   if (process.env.E2E6_DEBUG) {
     const dbg = await users.findOne({ name: 'alice' });
     console.log('DEBUG phase2 after replace+delete alice _id:', String(dbg._id), 'same:', String(dbg._id) === String(aliceId));
@@ -312,7 +331,7 @@ async function phase2(client, { users, aliceId }) {
   }
   check(
     'compaction reclaimed junk (file ~ live size)',
-    logSize < 24 * 1024 * 1024 && peakSize > logSize * 1.4,
+    logSize < 24 * 1024 * 1024 && peakSize > logSize,
     `file peaked at ${(peakSize / 1e6).toFixed(1)}MB, ended at ${(logSize / 1e6).toFixed(1)}MB after ~48MB of records were written (~12MB live)`,
   );
   check('bulk survivors intact', (await bulk.findOne({ _id: 1999 })).payload === payload2);