From 9390021b1e26fa33af6cd5f6228f0bee2e37d0e4 Mon Sep 17 00:00:00 2001 From: Aleksey Shakhmatov Date: Mon, 3 Aug 2026 20:05:38 +0300 Subject: [PATCH] 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 | 114 +++++++++++++++++++++----- src/index.zig | 201 ++++++++++++++++++++++++++++++++++++++++++++++ tests/e2e/e2e6.js | 29 +++++-- 3 files changed, 321 insertions(+), 23 deletions(-) diff --git a/src/commands.zig b/src/commands.zig index 41552bb..7a3bb36 100644 --- a/src/commands.zig +++ b/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| { - var plan = p; - defer plan.deinit(ctx.gpa); - var offs: std.ArrayListUnmanaged(u64) = .empty; - defer offs.deinit(ctx.gpa); - try plan.search(ctx.gpa, &offs); + // Candidates arrive as a stream so that a whole-index read never + // materializes: at the tens-of-GB target a `countDocuments({})` would + // otherwise build a list of every offset in the collection before the + // first one is examined. A narrowed plan still materializes, because its + // multikey/$in dedupe genuinely needs the whole set, and it is bounded by + // 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 (!try query.matches_bytes(ctx.gpa, filter, coll.doc_bytes(off))) continue; - if (out) |list| try list.append(ctx.gpa, off); - n += 1; - if (lim != 0 and n >= lim) break; + if (plan.full_scan()) { + cands = if (plan.backward) + .{ .scan_rev = plan.index.iter_reverse() } + else + .{ .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 (it.next()) |entry| { - if (!try query.matches_bytes(ctx.gpa, filter, coll.doc_bytes(entry.value_ptr.*))) continue; - if (out) |list| try list.append(ctx.gpa, entry.value_ptr.*); + while (cands.next()) |off| { + if (!try query.matches_bytes(ctx.gpa, filter, coll.doc_bytes(off))) continue; + if (out) |list| try list.append(ctx.gpa, off); 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(); - while (it.next()) |entry| try offs.append(ctx.gpa, entry.value_ptr.*); + // Every document, in _id order. A pipeline materializes its stream + // 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 diff --git a/src/index.zig b/src/index.zig index e73d412..d65385f 100644 --- a/src/index.zig +++ b/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 diff --git a/tests/e2e/e2e6.js b/tests/e2e/e2e6.js index 0f87ec4..b7571fa 100644 --- a/tests/e2e/e2e6.js +++ b/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 - // triggers on the share of the log that is garbage rather than on bytes - // appended, so the absolute peak depends on when that share crosses the - // threshold and is not a stable number to assert on. + // `peak > final` is the assertion with teeth, and it is worth saying why, + // because the intuitive alternative does not work here. The log is + // LZ4-compressed and this payload is one repeated character, so ~48 MB of + // 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);