//! Secondary indexes: per-collection, single-field and compound, with
//! `unique` and `sparse` options, persisted through the log and used by the
//! query planner as candidate-id generators.
//!
//! The governing invariant: an index is only ever used to produce a
//! candidate id set; the full filter is re-applied to every candidate. An
//! index that over-approximates is merely slow, never wrong. The entire
//! correctness risk collapses onto one question — can the index ever
//! under-approximate? Every design decision here answers that with *no*:
//!
//! - Entry generation mirrors field_matches (src/query.zig) exactly,
//!   indexing the value at the path plus the elements of any array there,
//!   so whole-array equality can never be missed.
//! - A sparse index is never used for a null-valued component ({a: null}
//!   would otherwise miss docs whose field the sparse index skipped).
//! - The _id_ fast path (the docs map) is only used when the queried value's
//!   compare-equivalence class is serialization-canonical: bson.compare
//!   treats int32 1, int64 1 and double 1.0 as equal, but serialize_value
//!   produces different map keys, so a hash lookup would miss.
//! - Entry insertion is infallible after the log append (capacity is
//!   reserved first), so a document can never be live but unindexed.
//!
//! `Entry.key` is an owned, order-preserving byte encoding of the indexed
//! values (see bson.encode_key), so comparing two entries is a memcmp
//! rather than a walk over Values that each live in a different document's
//! arena. An entry carries no reference to its document: the leaf record's
//! payload is the document's slab offset, so a candidate produced by a lookup
//! *is* the location of the document, and it is a value rather than a slice
//! that tree mutation could invalidate.
//!
//! Entries live in a B+tree (item 1 of the performance roadmap) instead of
//! one sorted array, so inserting into an already-built index costs O(log n)
//! instead of memmoving the array tail. Nodes are fixed 4 KiB pages in a
//! flat u32-addressed `ArrayListUnmanaged(Node)`; leaves are doubly linked
//! for ordered iteration. Records longer than a quarter page spill to an
//! append-only overflow slab (BSON strings reach 16 MB). Deletion does not
//! rebalance: leaves may sit underfull until emptied (an emptied leaf is
//! unlinked and dropped from its parent), and internal nodes may carry one
//! child — both are correct, and a fresh tree is rebuilt from the log on
//! every open anyway.

const std = @import("std");
const bson = @import("bson.zig");
const query = @import("query.zig");
const pgr = @import("pager.zig");
// Always active, including in the default ReleaseFast build. The tree's hot
// inner loops keep std.debug.assert (see assert.zig's module comment); these
// guard the reservation bounds, whose violation is a buffer overrun on a path
// that has already appended to the log and cannot report failure.
const assert_msg = @import("assert.zig").assert_msg;

/// MongoDB's compound index field limit.
pub const max_index_keys: usize = 32;

/// Cap on the $in cartesian product a plan will generate; beyond it the
/// planner falls back to a scan.
const max_combos: u64 = 100;

/// One key in an index spec. `path` is owned by the Index.
pub const IndexKey = struct {
    path: []const u8,
    descending: bool,
};

/// One index entry: the gpa-owned encoded form of a document's values for the
/// index's paths, concatenated column by column (see bson.encode_key).
///
/// Deliberately carries no reference to the document. Every entry one document
/// contributes shares one document, so which document it is belongs on the call
/// that commits the entries, not on each entry -- which also makes it
/// impossible to confuse the offset a replace is *removing* with the one it is
/// *inserting*. It used to hold an `id` slice aliasing the docs map's key, an
/// aliasing that only held because removal happened at the single chokepoint
/// where a document dies (PLAN amendment A3).
pub const Entry = struct {
    key: []const u8,
};

/// One staged entry during a bulk build. Unlike `Entry` this does carry the
/// document, because a bulk batch spans many of them.
const Staged = struct {
    key: []const u8,
    off: u64,
};

/// Entries built for one document, before they are committed to the index.
pub const BuiltEntries = struct {
    entries: std.ArrayListUnmanaged(Entry),
    multikey: bool,

    pub fn deinit(self: *BuiltEntries, gpa: std.mem.Allocator) void {
        for (self.entries.items) |e| gpa.free(e.key);
        self.entries.deinit(gpa);
    }
};

// ---------------------------------------------------------------------------
// B+tree storage
// ---------------------------------------------------------------------------

/// Pages in a standard overflow extent: 1 MiB, enough that a batch of spilled
/// records rarely needs more than one.
const ovf_extent_pages: u32 = (1024 * 1024) / pgr.page_size;

const page_size = 4096;
/// Bytes of node payload: a 32-byte header plus the slotted region.
const page_data = page_size - 32;
/// Records longer than a quarter of a node spill to the overflow slab.
const inline_limit = page_size / 4;
/// Upper bound on the slots one node can hold, since every slot costs at
/// least its own size. Bounds the split scratch.
const max_slots = page_data / slot_size;

/// One slotted-page entry: a key plus either an id length (leaf records) or
/// a child node id (internal separators). The key bytes live either inline
/// in the node's page (`off` into `buf`) or in the overflow slab (`off` into
/// `Index.overflow`).
const Slot = packed struct {
    off: u64,
    key_len: u32,
    extra: u32,
    spill: bool,
    _pad: u31,
};
const slot_size = @sizeOf(Slot);

/// A fixed 4 KiB page. Slots grow from the front of `buf`, record bytes
/// from the back; slots are kept ordered by key, so a node lookup is a
/// binary search over slots. Node ids address `Index.nodes` directly.
const Node = extern struct {
    is_leaf: u32,
    count: u32,
    parent: u32,
    next: u32,
    prev: u32,
    first_child: u32,
    data_start: u32,
    _rsv: u32,
    buf: [page_data]u8,
};

// These layouts are about to become an on-disk format: M0 maps the node arena
// straight out of the data file, with no serialization on page-in, so a page
// written by one build must be readable by the next. Nothing pinned them
// before, which meant a field added to Node or a change in how Zig lays out a
// packed Slot would silently reshape the file. `@sizeOf(Slot)` in particular
// is not obvious from its declaration -- a packed struct's size depends on the
// alignment of its backing integer, so the 160 declared bits round up.
comptime {
    std.debug.assert(page_size == 4096);
    std.debug.assert(@sizeOf(Node) == page_size);
    std.debug.assert(@alignOf(Node) <= page_size);
    std.debug.assert(page_data == page_size - 32);
    std.debug.assert(@offsetOf(Node, "buf") == 32);
    std.debug.assert(@bitSizeOf(Slot) == 160);
    // 160 declared bits is 20 bytes, but the backing integer's 16-byte
    // alignment rounds @sizeOf up to 32 -- so 12 of every 32 slot bytes are
    // padding, and a node holds 127 slots where a 20-byte slot would give it
    // 203. Pinned rather than fixed: narrowing the slot changes the fanout
    // and therefore the on-disk shape of every index, which belongs in the
    // commit that reshapes leaf records, not in a refactor.
    std.debug.assert(slot_size == 32);
    std.debug.assert(max_slots * slot_size <= page_data);
    // Node pages are raw host memory in the data file, so the file is
    // little-endian-only (the log, which is framed field by field, is not).
    std.debug.assert(@import("builtin").cpu.arch.endian() == .little);
}

/// A view of one stored entry yielded by iteration. `key` aliases the tree and
/// is valid only while the tree is not mutated; `off` is a value and stays
/// valid regardless.
pub const EntryRef = struct {
    key: []const u8,
    off: u64,
};

/// A leaf record's payload: the document's slab offset, little-endian, so a
/// record is `key ++ off_le` and every leaf slot's `extra` is exactly this.
/// Fixing the payload width is what keeps every byte-accounting site --
/// `fits`, `record_cost`, `slot_cost`, `balanced_cut`, `repack_keep_prefix` --
/// unchanged by the switch away from id bytes.
const off_len: u32 = @sizeOf(u64);

pub const Index = struct {
    name: []const u8,
    keys: []const IndexKey,
    unique: bool,
    sparse: bool,
    /// expireAfterSeconds when this is a TTL index, else null. Documents
    /// whose indexed value is a datetime older than this many seconds are
    /// deleted by Engine.ttl_sweep (src/db.zig); the index itself only
    /// carries the setting. Always within
    /// [0, max_expire_after_seconds] — parse_spec is the only producer.
    ttl: ?i64,
    multikey: bool,

    // -- the tree ----------------------------------------------------------
    /// The data file the node pages and the overflow slab live in.
    pager: *pgr.Pager,
    /// Node id -> page number. A node is one page, but its id is *not* its page
    /// number, and PLAN amendment A1 explains why: `Node.parent`, `next`, `prev`
    /// and an internal slot's `extra` are back-pointers by id, so copy-on-write
    /// moving a page would force every node referring to it to move too --
    /// COWing one leaf would cascade through the whole leaf level, and one
    /// internal node through its entire subtree.
    ///
    /// With this indirection the table slot is the single owner of a page
    /// number, so copy-on-write has exactly one pointer to fix. It costs one
    /// dependent load per node access and 4 bytes per node -- about 5.6 MB at
    /// 100M documents, against the 64-100 bytes *per document* of the hashmap
    /// this milestone removes.
    node_pages: std.ArrayListUnmanaged(u32),
    /// Spilled records live in the data file, in extents this index owns.
    /// Append-only and never rewritten, so a `Slot.off` into it stays valid for
    /// the life of the tree. Those offsets are *absolute file offsets* now,
    /// which is the same change documents went through -- and the reason
    /// `Slot.off` means two different things depending on `spill` is unchanged,
    /// only the second meaning moved.
    ovf_extents: std.ArrayListUnmanaged(pgr.Extent),
    ovf_tail: u64,
    ovf_end: u64,
    /// Bulk-build staging: entries appended unsorted by append_doc_entries,
    /// sorted and packed into the tree by finish_bulk. The keys are owned
    /// by the staging array until the pack consumes them.
    staging: std.ArrayListUnmanaged(Staged),
    root: u32,
    first_leaf: u32,
    leaf_count: u32,
    /// Internal levels above the leaves: 0 when the root is a leaf.
    depth: u32,
    /// Total entries, maintained incrementally.
    entry_count: usize,
    /// Repack scratch: any single node's record bytes fit here.
    scratch: [page_data]u8,
    /// Promoted-key scratch: inline keys being propagated up a split are
    /// copied here so the reference survives node-array growth.
    promo: [inline_limit]u8,

    pub fn init(
        gpa: std.mem.Allocator,
        pager: *pgr.Pager,
        name: []const u8,
        keys: []const IndexKey,
        unique: bool,
        sparse: bool,
        ttl: ?i64,
    ) !Index {
        var self: Index = .{
            .name = undefined,
            .keys = undefined,
            .unique = unique,
            .sparse = sparse,
            .ttl = ttl,
            .multikey = false,
            .pager = pager,
            .node_pages = .empty,
            .ovf_extents = .empty,
            .ovf_tail = 0,
            .ovf_end = 0,
            .staging = .empty,
            .root = 0,
            .first_leaf = 0,
            .leaf_count = 0,
            .depth = 0,
            .entry_count = 0,
            .scratch = undefined,
            .promo = undefined,
        };
        self.name = try gpa.dupe(u8, name);
        const owned_keys = try gpa.alloc(IndexKey, keys.len);
        var n: usize = 0;
        errdefer {
            for (owned_keys[0..n]) |k| gpa.free(k.path);
            gpa.free(owned_keys);
            gpa.free(self.name);
        }
        while (n < keys.len) : (n += 1) {
            owned_keys[n] = .{ .path = try gpa.dupe(u8, keys[n].path), .descending = keys[n].descending };
        }
        self.keys = owned_keys;
        // Slot 0 is a dummy (0 is the null node id); the root is one empty
        // leaf, so a fresh index is always a valid tree.
        try self.node_pages.ensureUnusedCapacity(gpa, 2);
        try pager.reserve_pages(2);
        self.node_pages.appendAssumeCapacity(pager.alloc_pages_assume_reserved(1));
        self.node_pages.appendAssumeCapacity(pager.alloc_pages_assume_reserved(1));
        self.page_mut(0).* = empty_node(0);
        self.page_mut(1).* = empty_node(1);
        self.root = 1;
        self.first_leaf = 1;
        self.leaf_count = 1;
        return self;
    }

    pub fn deinit(self: *Index, gpa: std.mem.Allocator) void {
        for (self.staging.items) |e| gpa.free(e.key);
        self.staging.deinit(gpa);
        self.node_pages.deinit(gpa);
        self.ovf_extents.deinit(gpa);
        for (self.keys) |k| gpa.free(k.path);
        gpa.free(self.keys);
        gpa.free(self.name);
    }

    pub fn count(self: *const Index) usize {
        return self.entry_count;
    }

    // -- entry generation ---------------------------------------------------

    /// Build the entries one document contributes. Mirrors field_matches
    /// (src/query.zig): the value at each path plus the elements of any
    /// array there, so both `{tags: "a"}` element queries and whole-array
    /// equality on `{tags: ["a","b"]}` are covered. Returns an empty list
    /// for a sparse index when a path yields no values (the document is
    /// skipped); a non-sparse index indexes missing fields as null.
    pub fn build_entries(
        self: *const Index,
        gpa: std.mem.Allocator,
        doc: []const u8,
    ) !BuiltEntries {
        // One arena for the whole call: the collected values and any nested
        // spines the byte walker materializes (whole-array/document values)
        // live here, so nothing leaks. The finished keys are still
        // exact-sized gpa copies that BuiltEntries owns.
        var arena = std.heap.ArenaAllocator.init(gpa);
        defer arena.deinit();
        const a = arena.allocator();

        var per_path: std.ArrayListUnmanaged(std.ArrayListUnmanaged(bson.Value)) = .empty;
        var multikey = false;
        var multi_paths: usize = 0;

        for (self.keys) |k| {
            var values: std.ArrayListUnmanaged(bson.Value) = .empty;
            try query.collect_values_bytes(a, doc, k.path, &values, 0);
            // Index the array itself and each element, like field_matches.
            const direct = values.items.len;
            var i: usize = 0;
            while (i < direct) : (i += 1) {
                if (values.items[i] == .array) {
                    multikey = true;
                    for (values.items[i].array) |elem| try values.append(a, elem);
                }
            }
            if (direct > 1) multikey = true;
            if (values.items.len > 1) multi_paths += 1;
            if (values.items.len == 0) {
                if (self.sparse) return .{ .entries = .empty, .multikey = false };
                try values.append(a, .null);
            }
            try per_path.append(a, values);
        }
        if (multi_paths > 1) return error.ParallelArrays;

        // Cartesian product across paths, then dedupe identical entries.
        var out: std.ArrayListUnmanaged(Entry) = .empty;
        errdefer {
            for (out.items) |e| gpa.free(e.key);
            out.deinit(gpa);
        }
        const nkeys = self.keys.len;
        var limits: [max_index_keys]usize = undefined;
        for (0..nkeys) |ci| limits[ci] = per_path.items[ci].items.len;
        var choice: [max_index_keys]usize = undefined;
        @memset(choice[0..nkeys], 0);
        // One reused buffer; each finished key is copied out to its own
        // exact-sized allocation.
        var enc: std.ArrayListUnmanaged(u8) = .empty;
        while (true) {
            enc.clearRetainingCapacity();
            for (0..nkeys) |ci| try bson.encode_key(per_path.items[ci].items[choice[ci]], a, &enc);
            const key = try gpa.dupe(u8, enc.items);
            errdefer gpa.free(key);
            try out.append(gpa, .{ .key = key });
            if (!advance_choice(choice[0..nkeys], limits[0..nkeys])) break;
        }
        if (out.items.len > 1) {
            std.mem.sort(Entry, out.items, {}, entry_less);
            var w: usize = 1;
            for (out.items[1..]) |e| {
                if (compare_entries(out.items[w - 1], e) != .eq) {
                    out.items[w] = e;
                    w += 1;
                } else {
                    gpa.free(e.key);
                }
            }
            out.items.len = w;
        }
        return .{ .entries = out, .multikey = multikey };
    }

    // -- mutation -----------------------------------------------------------

    /// Reserve everything `insert_entries` of this batch will need, so that
    /// insertion — which runs after the log append — is infallible: node
    /// pages for the worst-case splits of a batch (each entry creates at
    /// most one leaf plus a bounded share of internal splits; batches are
    /// sorted, so new leaves are packed, but reserve the trivial bound),
    /// and the exact overflow bytes (spilled records are copied once).
    pub fn reserve_for(self: *Index, gpa: std.mem.Allocator, entries: []const Entry) !void {
        const n: u64 = entries.len;
        // One entry splits at most one node per level, plus a new root when
        // the old root is the level that splits: `levels + 1` nodes, where
        // `levels == depth + 1`.
        //
        // A batch can also deepen the tree as it goes, and each added level
        // costs one more node per remaining entry -- so the bound has to use
        // the *final* depth. Growing by one level means splitting the root,
        // which means filling it first, and the smallest root a split can
        // leave holds one separator: refilling it to a split takes at least
        // two more arrivals from below, each of which needs a split one
        // level down. So g added levels take at least 2^g entries, and
        // `log2_ceil(n+1) + 1` bounds g.
        //
        // This replaces an earlier `n/8` stand-in for g, which was ~125
        // levels for a 1000-entry batch -- harmless as ArrayList capacity,
        // but it becomes real file growth once the arena is file-backed
        // (~528 MiB of demanded headroom for that batch, against ~70 MiB
        // here). Tightening it further needs an amortized argument rather
        // than this per-entry one, since no single insertion can split a
        // full path twice in a row.
        //
        // Note what does and does not guard this arithmetic: `alloc_node`'s
        // assert catches an overrun of the *actual* capacity, and
        // ensureUnusedCapacity over-allocates geometrically, so a bound
        // that is slightly too small is usually masked here. (Mutation-
        // checked by dropping the reservation entirely, which does fire it
        // -- six tests.) Once the arena is file-backed and the reservation
        // is exact, that assert becomes a real check on this expression.
        const growth: u64 = std.math.log2_int_ceil(u64, n + 1) + 1;
        const extra_nodes: u64 = n * (self.depth + 2 + growth) + 4;
        // Two reservations, because a node now needs both a page in the file and
        // a slot in the id->page table, and the insertion after the log append
        // must not be able to fail on either.
        try self.node_pages.ensureUnusedCapacity(gpa, @intCast(extra_nodes));
        try self.pager.reserve_pages(@intCast(extra_nodes));
        try self.reserve_overflow(gpa, entries);
    }

    /// Reserve the slab bytes `entries` will spill, so that store_record's
    /// append is infallible. A spilled record is copied into the slab once,
    /// when it first enters the tree; moving it between nodes re-uses its
    /// offset.
    ///
    /// `anytype` so it serves both `Entry` (the insert path) and `Staged` (the
    /// bulk path); it only ever reads `.key`.
    fn reserve_overflow(self: *Index, gpa: std.mem.Allocator, entries: anytype) !void {
        var overflow_bytes: u64 = 0;
        for (entries) |e| {
            const rec_len: u64 = e.key.len + off_len;
            if (rec_len > inline_limit) overflow_bytes += rec_len;
        }
        if (overflow_bytes == 0) return;
        // Same rule as the document slab: a checkpoint freezes the page the tail
        // points into, so a frozen tail means starting a fresh extent rather
        // than writing inside the durable image.
        // Same reasoning as the document slab: a recycled extent is below the
        // stable mark and still writable, so ask whether these bytes are in the
        // published image rather than where they sit.
        if (self.pager.is_unpublished_at(self.ovf_tail) and self.ovf_tail + overflow_bytes <= self.ovf_end) return;
        // One extent for the whole batch, or a bespoke one when a single
        // record is larger than the standard extent (a BSON string reaches
        // 16 MB).
        const want_pages: u32 = @intCast(@max(
            ovf_extent_pages,
            (overflow_bytes + pgr.page_size - 1) / pgr.page_size,
        ));
        try self.pager.reserve_pages(want_pages);
        const first = self.pager.alloc_pages_assume_reserved(want_pages);
        try self.ovf_extents.append(gpa, .{ .first = first, .pages = want_pages });
        self.ovf_tail = @as(u64, first) << pgr.page_shift;
        self.ovf_end = self.ovf_tail + (@as(u64, want_pages) << pgr.page_shift);
    }

    /// Insert pre-built entries (maintaining order) and drain the batch.
    /// Infallible: reserve_for must have run first. The tree copies each
    /// key, so ownership stays with the batch and its deinit frees them.
    /// `off` is where the document these entries describe lives in the slab.
    /// Passed here rather than carried on each entry so a replace cannot mix up
    /// the offset it is removing with the one it is inserting.
    pub fn insert_entries(self: *Index, built: *BuiltEntries, off: u64) void {
        for (built.entries.items) |e| {
            self.insert_entry(e.key, off);
        }
    }

    /// Build, check, and insert entries for one document — the whole
    /// entry-commit protocol in one call, used everywhere a single document
    /// joins an index (create, rebuild on open). Writes that must reserve
    /// capacity before a log append use the split build/reserve/insert form
    /// directly.
    ///
    /// With `enforce_unique` false a duplicate is tolerated rather than
    /// rejected (the rebuild path keeps the index and warns); the return
    /// value reports whether that happened.
    pub fn add_doc(
        self: *Index,
        gpa: std.mem.Allocator,
        doc: []const u8,
        off: u64,
        enforce_unique: bool,
    ) !bool {
        var built = try self.build_entries(gpa, doc);
        // Runs on success too: the batch's keys are copied into the tree,
        // so deinit frees exactly what this call allocated.
        defer built.deinit(gpa);
        var duplicate = false;
        if (self.unique) {
            // null, not `off`: this document is not in the tree yet, so it
            // has no own entries to exclude. See check_unique.
            self.check_unique(built.entries.items, null) catch |err| {
                if (enforce_unique) return err;
                duplicate = true;
            };
        }
        if (built.multikey) self.multikey = true;
        try self.reserve_for(gpa, built.entries.items);
        self.insert_entries(&built, off);
        return duplicate;
    }

    /// Append one document's entries to the bulk-build staging array
    /// without maintaining order. Pairs with `finish_bulk`, which sorts the
    /// staging once and packs it into leaves bottom-up.
    ///
    /// Inserting documents one at a time keeps a sorted array memmoved on
    /// every entry, so building an index over n documents would move O(n²)
    /// bytes — that was the whole cost of createIndex on a large
    /// collection. Staging and packing is O(n log n) and no memmove.
    pub fn append_doc_entries(
        self: *Index,
        gpa: std.mem.Allocator,
        doc: []const u8,
        off: u64,
    ) !void {
        var built = try self.build_entries(gpa, doc);
        // Runs on success too: the append below moves the keys into the
        // staging array, leaving only the (now empty) ArrayList buffer.
        defer built.deinit(gpa);
        try self.staging.ensureUnusedCapacity(gpa, built.entries.items.len);
        for (built.entries.items) |e| self.staging.appendAssumeCapacity(.{ .key = e.key, .off = off });
        if (built.multikey) self.multikey = true;
        // Key ownership moved into the staging array.
        built.entries.items.len = 0;
    }

    /// Sort the staging collected by `append_doc_entries`, check a unique
    /// index for duplicate keys (one adjacent-pair scan), and pack the tree.
    ///
    /// With `enforce_unique` false a duplicate is tolerated rather than
    /// rejected, matching `add_doc`; the return value reports whether that
    /// happened.
    /// The error set widened when the overflow slab moved into the data file:
    /// reserving it can now fail on file growth, not only on OOM.
    pub fn finish_bulk(
        self: *Index,
        gpa: std.mem.Allocator,
        enforce_unique: bool,
    ) !bool {
        std.mem.sort(Staged, self.staging.items, {}, staged_less);
        var duplicate = false;
        if (self.unique and self.staging.items.len >= 2) {
            for (self.staging.items[1..], 0..) |cur, prev_i| {
                const prev = self.staging.items[prev_i];
                // Sorting puts equal keys next to each other.
                if (!std.mem.eql(u8, prev.key, cur.key)) continue;
                // A document's own entries were deduped at build time, so
                // equal keys from one document are not a conflict — the same
                // rule as check_unique's exclude.
                if (prev.off == cur.off) continue;
                if (enforce_unique) return error.DuplicateKeyIndex;
                duplicate = true;
            }
        }
        // Only the slab needs reserving up front: pack_tree grows the node
        // array itself, so a bulk build no longer reserves a node per entry
        // when it needs one per leaf (65k entries pack into ~1k leaves).
        try self.reserve_overflow(gpa, self.staging.items);
        try self.pack_tree(gpa);
        return duplicate;
    }

    /// Throw the tree away and start from an empty root, so a rebuild can pack a
    /// fresh one. The old pages go on the free list, which withholds them for two
    /// generations -- the image that still references them stays intact.
    pub fn reset_tree(self: *Index, gpa: std.mem.Allocator) !void {
        for (self.node_pages.items) |p| try self.pager.free_pages(p, 1);
        for (self.ovf_extents.items) |e| try self.pager.free_pages(e.first, e.pages);
        self.node_pages.clearRetainingCapacity();
        self.ovf_extents.clearRetainingCapacity();
        self.ovf_tail = 0;
        self.ovf_end = 0;
        try self.node_pages.ensureUnusedCapacity(gpa, 2);
        try self.pager.reserve_pages(2);
        self.node_pages.appendAssumeCapacity(self.pager.alloc_pages_assume_reserved(1));
        self.node_pages.appendAssumeCapacity(self.pager.alloc_pages_assume_reserved(1));
        self.page_mut(0).* = empty_node(0);
        self.page_mut(1).* = empty_node(1);
        self.root = 1;
        self.first_leaf = 1;
        self.leaf_count = 1;
        self.depth = 0;
        self.entry_count = 0;
        self.multikey = false;
    }

    /// Remove every entry for `id`, in one pass over the leaves. Infallible.
    /// Used directly by remove_id's own callers and as the fallback when
    /// regeneration cannot locate entries.
    pub fn remove_off(self: *Index, off: u64) void {
        var leaf_id = self.first_leaf;
        while (leaf_id != 0) {
            const node = self.page(leaf_id);
            const next = node.next;
            // Slots are removed high-to-low so the indices stay valid.
            var i = node.count;
            while (i > 0) {
                i -= 1;
                if (self.off_of(leaf_id, i) == off) self.leaf_remove(leaf_id, i);
            }
            leaf_id = next;
        }
    }

    /// Remove the entries `doc` contributed, locating each by regeneration
    /// instead of scanning the tree.
    ///
    /// Entry generation is a pure function of the document, so the entries
    /// to remove can be regenerated and looked up directly; each is a
    /// descent plus a leaf-local scan of its key band (equal keys may
    /// legitimately span several leaves). `remove_id` walks every leaf
    /// comparing ids, which made a single document update cost O(index
    /// size) per index.
    ///
    /// 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 u8, off: u64) void {
        var built = self.build_entries(gpa, doc) catch return self.remove_off(off);
        defer built.deinit(gpa);
        // Sparse index that skipped this document: nothing was inserted.
        if (built.entries.items.len == 0) return;
        for (built.entries.items) |want| {
            // A document that generated entries we cannot find would leave
            // the index stale, which is worse than being slow.
            if (!self.remove_entry(want.key, off)) return self.remove_off(off);
        }
    }

    /// 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.
    /// `exclude_id` is the document whose own existing entries do not count as
    /// duplicates -- a replace re-inserts its entries, so its old ones must be
    /// ignored. It must be **null for an insert**, where the document has no
    /// entries yet: passing its id there would make a colliding entry with the
    /// *same* id invisible, which is exactly the case the implicit `_id_` index
    /// has to catch.
    pub fn check_unique(
        self: *const Index,
        new_entries: []const Entry,
        exclude: ?u64,
    ) error{DuplicateKeyIndex}!void {
        for (new_entries) |e| {
            var it = self.seek(e.key);
            while (it.next()) |have| {
                if (cmp_prefix(e.key, have.key) != .eq) break;
                const is_self = exclude != null and have.off == exclude.?;
                if (!is_self) return error.DuplicateKeyIndex;
            }
        }
    }

    /// The id stored under exactly `key`, or null. Exact byte equality rather
    /// than `cmp_prefix`, because this answers "is this document present"
    /// rather than "what is in this key band". On a non-unique index it
    /// returns the first entry of the band.
    pub fn lookup_exact(self: *const Index, key: []const u8) ?u64 {
        var it = self.seek(key);
        const e = it.next() orelse return null;
        if (!std.mem.eql(u8, e.key, key)) return null;
        return e.off;
    }

    // -- search -------------------------------------------------------------

    /// All ids whose key equals `key` (component-wise). For a partial key
    /// (fewer components than the index has) this is a prefix search.
    pub fn lookup_eq(
        self: *const Index,
        gpa: std.mem.Allocator,
        key: []const bson.Value,
        out: *std.ArrayListUnmanaged(u64),
    ) !void {
        var enc: std.ArrayListUnmanaged(u8) = .empty;
        defer enc.deinit(gpa);
        for (key) |v| try bson.encode_key(v, gpa, &enc);
        var it = self.seek(enc.items);
        while (it.next()) |e| {
            if (cmp_prefix(enc.items, e.key) != .eq) break;
            try out.append(gpa, e.off);
        }
    }

    /// All ids whose key starts with `prefix` and whose component at
    /// `prefix.len` falls within [lo, hi]. Range bounds apply to the next
    /// component after the equality prefix.
    pub fn lookup_range(
        self: *const Index,
        gpa: std.mem.Allocator,
        prefix: []const bson.Value,
        lo: ?bson.Value,
        lo_incl: bool,
        hi: ?bson.Value,
        hi_incl: bool,
        out: *std.ArrayListUnmanaged(u64),
    ) !void {
        // Both bounds extend the encoded prefix by one component, so each
        // is a lower-bound seek (encoded bytes are self-delimiting: the
        // extended key orders within the prefix's own band). The scan
        // starts at the lower bound and stops at the first key beyond the
        // band or the upper bound — no slice ends to compute.
        var enc_a: std.ArrayListUnmanaged(u8) = .empty;
        var enc_b: std.ArrayListUnmanaged(u8) = .empty;
        defer enc_a.deinit(gpa);
        defer enc_b.deinit(gpa);
        for (prefix) |v| try bson.encode_key(v, gpa, &enc_a);
        const prefix_len = enc_a.items.len;
        if (lo) |l| try bson.encode_key(l, gpa, &enc_a);
        if (hi) |h| {
            for (prefix) |v| try bson.encode_key(v, gpa, &enc_b);
            try bson.encode_key(h, gpa, &enc_b);
        }
        const prefix_key = enc_a.items[0..prefix_len];
        const lo_key: ?[]const u8 = if (lo != null) enc_a.items else null;
        const hi_key: ?[]const u8 = if (hi != null) enc_b.items else null;

        var it = self.seek(if (lo_key) |lk| lk else prefix_key);
        while (it.next()) |e| {
            if (lo_key) |lk| {
                // Exclusive lower bound: skip entries still at or below lo
                // (an entry whose component equals lo is exactly lo).
                if (!lo_incl and cmp_prefix(lk, e.key) != .lt) continue;
            }
            if (hi_key) |hk| {
                const c = cmp_prefix(hk, e.key);
                if (hi_incl) {
                    if (c == .lt) break; // key > hi
                } else {
                    if (c != .gt) break; // key >= hi
                }
            } else if (cmp_prefix(prefix_key, e.key) != .eq) {
                break; // band ended
            }
            try out.append(gpa, e.off);
        }
    }

    // -- iteration ----------------------------------------------------------

    /// Ordered iteration over the whole index. Entries come out in key
    /// order; equal keys in the order the leaves happen to hold them.
    pub const Iter = struct {
        ix: *const Index,
        leaf: u32,
        slot: u32,

        pub fn next(self: *Iter) ?EntryRef {
            const ix = self.ix;
            while (self.leaf != 0) {
                const node = ix.page(self.leaf);
                if (self.slot < node.count) {
                    const key = ix.key_of(self.leaf, self.slot);
                    const off = ix.off_of(self.leaf, self.slot);
                    self.slot += 1;
                    return .{ .key = key, .off = off };
                }
                self.leaf = node.next;
                self.slot = 0;
            }
            return null;
        }
    };

    pub fn iter(self: *const Index) Iter {
        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.
    pub fn seek(self: *const Index, prefix: []const u8) Iter {
        const b = self.lower_bound(prefix);
        return .{ .ix = self, .leaf = b.leaf, .slot = b.slot };
    }

    // -- serialization ------------------------------------------------------

    /// The canonical spec document bytes
    /// ({v, key, name, unique?, sparse?, expireAfterSeconds?}) stored in the
    /// log and used to rebuild the index on replay.
    pub fn write_spec(
        self: *const Index,
        gpa: std.mem.Allocator,
        out: *std.ArrayListUnmanaged(u8),
    ) !void {
        var arena = std.heap.ArenaAllocator.init(gpa);
        defer arena.deinit();
        var pairs: std.ArrayListUnmanaged(bson.Pair) = .empty;
        try self.spec_pairs(arena.allocator(), &pairs);
        try bson.write_doc(pairs.items, gpa, out);
    }

    /// The spec as pairs in `arena` (values alias this index's own storage,
    /// which outlives any reply). Used by listIndexes.
    pub fn spec_pairs(
        self: *const Index,
        arena: std.mem.Allocator,
        out: *std.ArrayListUnmanaged(bson.Pair),
    ) !void {
        try out.append(arena, .{ .key = "v", .value = .{ .int32 = 2 } });
        const key_pairs = try arena.alloc(bson.Pair, self.keys.len);
        for (self.keys, 0..) |k, i| {
            key_pairs[i] = .{ .key = k.path, .value = if (k.descending) .{ .int32 = -1 } else .{ .int32 = 1 } };
        }
        try out.append(arena, .{ .key = "key", .value = .{ .doc = key_pairs } });
        try out.append(arena, .{ .key = "name", .value = .{ .string = self.name } });
        if (self.unique) try out.append(arena, .{ .key = "unique", .value = .{ .bool = true } });
        if (self.sparse) try out.append(arena, .{ .key = "sparse", .value = .{ .bool = true } });
        // int32 like MongoDB: parse_spec caps the value at
        // max_expire_after_seconds, so the cast always fits.
        if (self.ttl) |secs| try out.append(arena, .{ .key = "expireAfterSeconds", .value = .{ .int32 = @intCast(secs) } });
    }

    pub fn spec_equal(a: *const Index, b: *const Index) bool {
        if (!std.mem.eql(u8, a.name, b.name)) return false;
        if (a.unique != b.unique or a.sparse != b.sparse) return false;
        if (!std.meta.eql(a.ttl, b.ttl)) return false;
        if (a.keys.len != b.keys.len) return false;
        for (a.keys, b.keys) |ka, kb| {
            if (!std.mem.eql(u8, ka.path, kb.path)) return false;
            if (ka.descending != kb.descending) return false;
        }
        return true;
    }

    // -- tree internals -----------------------------------------------------

    const Record = struct {
        key: []const u8,
        /// Set for a leaf record: the document's slab offset. `child` is set
        /// for an internal separator instead. Never both -- the optional is
        /// what tells `store_record` which kind it is holding.
        off: ?u64 = null,
        child: u32 = 0,
        /// When set, key+id already live in the overflow slab at this
        /// offset and are referenced rather than copied (moved leaf
        /// records, promoted separators). Optional rather than a 0
        /// sentinel: offset 0 is a real slab position, held by the first
        /// record that ever spilled -- which would otherwise be copied into
        /// the slab again on every move, past the reserved capacity.
        spill_off: ?u64 = null,
    };

    const StableKey = struct {
        key: []const u8,
        spill_off: ?u64,
    };

    const Split = struct {
        /// The separator to promote, stable across node-array growth (a
        /// slice into the overflow slab, or a copy in the promo buffer).
        key: []const u8,
        spill_off: ?u64,
        right: u32,
    };

    fn empty_node(is_leaf: u32) Node {
        return .{
            .is_leaf = is_leaf,
            .count = 0,
            .parent = 0,
            .next = 0,
            .prev = 0,
            .first_child = 0,
            .data_start = page_data,
            ._rsv = 0,
            .buf = undefined,
        };
    }

    /// Allocate a node id. Infallible: insert paths reserve capacity first;
    /// the pack path reserves via reserve_for before packing.
    ///
    /// The assert is the tripwire for `reserve_for`'s bound. Overrunning the
    /// reservation is not a graceful failure: `appendAssumeCapacity` writes
    /// past the buffer, and in ReleaseFast (this project's default) nothing
    /// else checks. It also cannot be reported to the caller -- this runs
    /// after the log append, on the path whose whole point is that a
    /// document can never be live but unindexed -- so panicking is the only
    /// honest response.
    fn alloc_node(self: *Index) u32 {
        assert_msg(self.node_pages.items.len < self.node_pages.capacity, "node allocation overran reserve_for's bound");
        const p = self.pager.alloc_pages_assume_reserved(1);
        self.node_pages.appendAssumeCapacity(p);
        const id: u32 = @intCast(self.node_pages.items.len - 1);
        self.page_mut(id).* = empty_node(0);
        return id;
    }

    /// Allocate a node id, growing the array. For the pack path, which is
    /// fallible anyway and would otherwise have to reserve one node per
    /// entry when it needs one per leaf.
    fn alloc_node_grow(self: *Index, gpa: std.mem.Allocator) !u32 {
        const p = try self.pager.alloc_pages(1);
        try self.node_pages.append(gpa, p);
        const id: u32 = @intCast(self.node_pages.items.len - 1);
        self.page_mut(id).* = empty_node(0);
        return id;
    }

    // -- arena access -------------------------------------------------------
    //
    // Every read of a node page goes through `page`, every write through
    // `page_mut`, and every overflow-slab read through `ovf`. Nothing else
    // touches `nodes.items` or `overflow.items`. That is not style: M0 moves
    // both stores into an mmap'd data file, where a write to a page belonging
    // to the last durable checkpoint has to copy the page first (PLAN
    // Amendment A1). Funnelling writes through one function is what makes
    // that a change of three bodies instead of sixty call sites, and what
    // lets a debug build enforce "no store below the stable mark".
    //
    // The rule that copy-on-write will impose, worth honouring already: do
    // not hold a `*Node` across a `page_mut` of *the same* id. Under COW the
    // second call can move that id to a fresh page, leaving the first
    // pointer aimed at a page nothing will ever read again. Holding pointers
    // to two *different* ids at once stays fine.
    //
    // The helpers that take a node id rather than a `*Node` -- store_record,
    // remove_record, repack_keep_prefix -- each re-acquire the page, so they
    // are where a caller could break the rule. Every caller currently
    // complies, and where it is not obvious it is because a value was read
    // out first: `insert_rec` passes `node.count` to repack_keep_prefix and
    // never touches `node` again afterwards. Audited as part of introducing
    // these accessors; re-audit when COW lands, and consider passing the
    // `*Node` down so the copy happens once at the top of the call.

    /// The page holding node `id`, for reading.
    inline fn page(self: *const Index, id: u32) *const Node {
        return @ptrCast(self.pager.page(self.node_pages.items[id]));
    }

    /// The page holding node `id`, for writing.
    /// Writable, via copy-on-write: a node inside the published image is copied
    /// to a fresh page and its table slot updated, so the durable bytes are
    /// never disturbed. Infallible in practice because `reserve_for` reserves
    /// the pages a batch can need -- the `catch` here would mean the reservation
    /// was short, which its own assert reports first and more precisely.
    inline fn page_mut(self: *Index, id: u32) *Node {
        const p = self.pager.page_mut_cow(&self.node_pages.items[id]) catch
            @panic("multiforadb: out of pages while writing an index node");
        return @ptrCast(p);
    }

    /// Overflow-slab bytes in `[from, to)`. Slot offsets are u64 because a
    /// spilled record can sit anywhere in the slab; the casts live here so
    /// the callers read as plain slicing.
    inline fn ovf(self: *const Index, from: u64, to: u64) []const u8 {
        return self.pager.bytes(from, @intCast(to - from));
    }

    fn get_slot(node_page: *const Node, i: u32) Slot {
        return std.mem.bytesToValue(Slot, node_page.buf[i * slot_size ..][0..slot_size]);
    }

    fn set_slot(node_page: *Node, i: u32, s: Slot) void {
        std.mem.bytesAsValue(Slot, node_page.buf[i * slot_size ..][0..slot_size]).* = s;
    }

    /// The key bytes of slot `i`, either in the node's page or the slab.
    fn key_of(self: *const Index, node_id: u32, i: u32) []const u8 {
        const node = self.page(node_id);
        const s = get_slot(node, i);
        if (s.spill) return self.ovf(s.off, s.off + s.key_len);
        return node.buf[@intCast(s.off)..@intCast(s.off + s.key_len)];
    }

    /// The id bytes of leaf slot `i`.
    /// The slab offset of leaf slot `i`. Leaves only: an internal separator's
    /// `extra` is a child node id, not a payload length.
    fn off_of(self: *const Index, node_id: u32, i: u32) u64 {
        const node = self.page(node_id);
        std.debug.assert(node.is_leaf == 1);
        const s = get_slot(node, i);
        std.debug.assert(s.extra == off_len);
        const start = s.off + s.key_len;
        const bytes = if (s.spill)
            self.ovf(start, start + off_len)
        else
            node.buf[@intCast(start)..@intCast(start + off_len)];
        return std.mem.readInt(u64, bytes[0..off_len], .little);
    }

    /// Whether a record of `rec_len` bytes fits `node`: the slot plus, when
    /// it stays inline, its bytes. Oversized records spill (slot only).
    fn fits(self: *const Index, node_id: u32, rec_len: u64) bool {
        const node = self.page(node_id);
        const inline_bytes: u64 = if (rec_len > inline_limit) 0 else rec_len;
        return (@as(u64, node.count) + 1) * slot_size + inline_bytes <= node.data_start;
    }

    /// Write `rec` into `node` at slot position `pos` (append when pos ==
    /// count), shifting later slots right to make room. Assumes fit.
    fn store_record(self: *Index, node_id: u32, pos: u32, rec: Record) void {
        const node = self.page_mut(node_id);
        // A leaf record carries the 8-byte offset; an internal separator
        // carries none, and puts its child node id in `extra` instead. The
        // optional is the discriminator: the old convention keyed off a
        // zero-length payload, which silently aliased onto child id 0.
        const payload: u32 = if (rec.off != null) off_len else 0;
        const rec_len: u64 = rec.key.len + payload;
        var s: Slot = .{
            .off = 0,
            .key_len = @intCast(rec.key.len),
            .extra = if (rec.off != null) off_len else rec.child,
            .spill = false,
            ._pad = 0,
        };
        if (rec.spill_off) |off| {
            s.off = off;
            s.spill = true;
        } else if (rec_len > inline_limit) {
            // Tripwire for reserve_overflow, for the same reason as
            // alloc_node's: this append runs after the log append and cannot
            // fail back to the caller.
            assert_msg(self.ovf_tail + rec_len <= self.ovf_end, "spilled record overran reserve_overflow's bound");
            s.off = self.ovf_tail;
            @memcpy(self.pager.bytes_mut(self.ovf_tail, rec.key.len), rec.key);
            self.ovf_tail += rec.key.len;
            if (rec.off) |o| {
                std.mem.writeInt(u64, self.pager.bytes_mut(self.ovf_tail, off_len)[0..off_len], o, .little);
                self.ovf_tail += off_len;
            }
            s.spill = true;
        } else {
            node.data_start -= @intCast(rec_len);
            const dst = node.buf[node.data_start .. node.data_start + @as(usize, @intCast(rec_len))];
            @memcpy(dst[0..rec.key.len], rec.key);
            if (rec.off) |o| std.mem.writeInt(u64, dst[rec.key.len..][0..off_len], o, .little);
            s.off = node.data_start;
        }
        if (pos < node.count) {
            const src = node.buf[pos * slot_size .. node.count * slot_size];
            const dst = node.buf[(pos + 1) * slot_size .. (node.count + 1) * slot_size];
            std.mem.copyBackwards(u8, dst, src);
        }
        set_slot(node, pos, s);
        node.count += 1;
    }

    /// Remove slot `i`, closing the hole its inline bytes leave and
    /// adjusting surviving inline offsets. Infallible.
    fn remove_record(self: *Index, node_id: u32, i: u32) void {
        const node = self.page_mut(node_id);
        const s = get_slot(node, i);
        const is_leaf = node.is_leaf == 1;
        const rec_len: u64 = s.key_len + (if (is_leaf) s.extra else 0);
        if (!s.spill and rec_len > 0) {
            const src = node.buf[@intCast(s.off + rec_len)..page_data];
            const dst = node.buf[@intCast(s.off) .. page_data - @as(usize, @intCast(rec_len))];
            std.mem.copyForwards(u8, dst, src);
            var j: u32 = 0;
            while (j < node.count) : (j += 1) {
                if (j == i) continue;
                var sj = get_slot(node, j);
                if (!sj.spill and sj.off > s.off) sj.off -= rec_len;
                set_slot(node, j, sj);
            }
            // data_start is the low boundary of the live region and is
            // unchanged: the freed bytes sit dead at the top of the page,
            // below the accounting, until the next split repacks it.
        }
        const src_slots = node.buf[(i + 1) * slot_size .. node.count * slot_size];
        const dst_slots = node.buf[i * slot_size .. (node.count - 1) * slot_size];
        std.mem.copyForwards(u8, dst_slots, src_slots);
        node.count -= 1;
        if (node.count == 0) node.data_start = page_data;
    }

    /// Repack `node` keeping only slots [0, k). Record bytes are copied to
    /// the scratch and written back packed from the top of the page; slots
    /// are rewritten in place with their new offsets.
    fn repack_keep_prefix(self: *Index, node_id: u32, k: u32) void {
        const node = self.page_mut(node_id);
        const is_leaf = node.is_leaf == 1;
        // Pass 1: surviving inline records to the scratch, in slot order.
        var scratch_len: usize = 0;
        var j: u32 = 0;
        while (j < k) : (j += 1) {
            const s = get_slot(node, j);
            if (s.spill) continue;
            const rec_len: usize = @intCast(s.key_len + (if (is_leaf) s.extra else 0));
            const src = node.buf[@intCast(s.off)..@intCast(s.off + rec_len)];
            @memcpy(self.scratch[scratch_len .. scratch_len + rec_len], src);
            scratch_len += rec_len;
        }
        // Pass 2: rewrite slots and data, back to front so the scratch is
        // consumed LIFO.
        var cursor: usize = page_data;
        j = k;
        while (j > 0) {
            j -= 1;
            var s = get_slot(node, j);
            if (s.spill) continue;
            const rec_len: usize = @intCast(s.key_len + (if (is_leaf) s.extra else 0));
            cursor -= rec_len;
            scratch_len -= rec_len;
            @memcpy(node.buf[cursor .. cursor + rec_len], self.scratch[scratch_len .. scratch_len + rec_len]);
            s.off = cursor;
            set_slot(node, j, s);
        }
        node.data_start = @intCast(cursor);
        node.count = k;
    }

    /// The key at (node, i) in a form stable across node-array growth.
    fn stable_key(self: *Index, node_id: u32, i: u32) StableKey {
        const node = self.page(node_id);
        const s = get_slot(node, i);
        if (s.spill) return .{ .key = self.ovf(s.off, s.off + s.key_len), .spill_off = s.off };
        const k = self.key_of(node_id, i);
        @memcpy(self.promo[0..k.len], k);
        return .{ .key = self.promo[0..k.len], .spill_off = null };
    }

    /// What one record costs a page: its slot plus, when it stays inline,
    /// its bytes. The unit both `fits` and the split point are measured in.
    fn record_cost(rec_len: u64) u32 {
        return slot_size + @as(u32, if (rec_len > inline_limit) 0 else @intCast(rec_len));
    }

    /// `record_cost` of the record already in slot `i`.
    fn slot_cost(self: *const Index, node_id: u32, i: u32) u32 {
        const node = self.page(node_id);
        const s = get_slot(node, i);
        if (s.spill) return slot_size;
        return slot_size + s.key_len + (if (node.is_leaf == 1) s.extra else 0);
    }

    /// Where to cut a merged run of records so that neither half exceeds
    /// half the total cost by more than one record: the first cut whose
    /// left half would pass half the total. A page's live records plus one
    /// new record cost at most page_data plus one record, so both halves
    /// are then guaranteed to fit a page. Cutting by slot count is not:
    /// every large record can sit on one side of the count midpoint.
    fn balanced_cut(costs: []const u32) u32 {
        var total: u64 = 0;
        for (costs) |c| total += c;
        var acc: u64 = 0;
        var mid: u32 = 0;
        while (mid < costs.len) : (mid += 1) {
            acc += costs[mid];
            if (acc * 2 > total) break;
        }
        return @min(mid, @as(u32, @intCast(costs.len - 1)));
    }

    /// Entry position in a leaf, by (key, id).
    fn leaf_pos(self: *const Index, leaf_id: u32, key: []const u8, off: u64) u32 {
        const node = self.page(leaf_id);
        var lo: u32 = 0;
        var hi: u32 = node.count;
        while (lo < hi) {
            const mid = lo + (hi - lo) / 2;
            const o = std.mem.order(u8, self.key_of(leaf_id, mid), key);
            const less = o == .lt or (o == .eq and self.off_of(leaf_id, mid) < off);
            if (less) lo = mid + 1 else hi = mid;
        }
        return lo;
    }

    /// Separator position in an internal node: after any equal keys, so the
    /// "last separator <= key" descent lands on the newest right child.
    fn separator_pos(self: *const Index, node_id: u32, key: []const u8) u32 {
        const node = self.page(node_id);
        var lo: u32 = 0;
        var hi: u32 = node.count;
        while (lo < hi) {
            const mid = lo + (hi - lo) / 2;
            if (std.mem.order(u8, self.key_of(node_id, mid), key) != .gt) lo = mid + 1 else hi = mid;
        }
        return lo;
    }

    /// The child holding the range `key` sorts into: right of the last
    /// separator that is <= key (equal keys live right of equal separators).
    fn descend_insert(self: *const Index, node_id: u32, key: []const u8) u32 {
        const node = self.page(node_id);
        var lo: u32 = 0;
        var hi: u32 = node.count;
        while (lo < hi) {
            const mid = lo + (hi - lo) / 2;
            if (std.mem.order(u8, self.key_of(node_id, mid), key) != .gt) lo = mid + 1 else hi = mid;
        }
        if (lo == 0) return node.first_child;
        if (lo == node.count) {
            if (node.count == 0) return node.first_child;
            return get_slot(node, node.count - 1).extra;
        }
        return get_slot(node, lo - 1).extra;
    }

    /// The child containing the lower bound of `prefix`: the child left of
    /// the first separator not less than the prefix (prefix semantics), or
    /// the rightmost child when every separator is less.
    fn descend_lower(self: *const Index, node_id: u32, prefix: []const u8) u32 {
        const node = self.page(node_id);
        var lo: u32 = 0;
        var hi: u32 = node.count;
        while (lo < hi) {
            const mid = lo + (hi - lo) / 2;
            if (cmp_prefix(prefix, self.key_of(node_id, mid)) == .gt) lo = mid + 1 else hi = mid;
        }
        if (lo == node.count) {
            // A one-child internal node (deletion does not rebalance) has
            // no separators; its only child is first_child.
            if (node.count == 0) return node.first_child;
            return get_slot(node, node.count - 1).extra;
        }
        if (lo == 0) return node.first_child;
        return get_slot(node, lo - 1).extra;
    }

    /// Position in a leaf of the first slot whose key is not less than
    /// `prefix` (prefix semantics).
    fn leaf_lower(self: *const Index, leaf_id: u32, prefix: []const u8) u32 {
        const node = self.page(leaf_id);
        var lo: u32 = 0;
        var hi: u32 = node.count;
        while (lo < hi) {
            const mid = lo + (hi - lo) / 2;
            if (cmp_prefix(prefix, self.key_of(leaf_id, mid)) == .gt) lo = mid + 1 else hi = mid;
        }
        return lo;
    }

    /// The (leaf, slot) of the first entry whose key is not less than
    /// `prefix`.
    fn lower_bound(self: *const Index, prefix: []const u8) struct { leaf: u32, slot: u32 } {
        var node_id = self.root;
        while (self.page(node_id).is_leaf == 0) node_id = self.descend_lower(node_id, prefix);
        return .{ .leaf = node_id, .slot = self.leaf_lower(node_id, prefix) };
    }

    /// Insert one entry, splitting upward as needed. Infallible.
    fn insert_entry(self: *Index, key: []const u8, off: u64) void {
        if (self.insert_rec(self.root, key, off)) |up| {
            // The root split: a new root with the two halves as children.
            const new_root = self.alloc_node();
            self.page_mut(new_root).first_child = self.root;
            self.page_mut(self.root).parent = new_root;
            self.store_record(new_root, 0, .{ .key = up.key, .child = up.right, .spill_off = up.spill_off });
            self.page_mut(up.right).parent = new_root;
            self.root = new_root;
            self.depth += 1;
        }
    }

    /// Descend and insert; return the split to promote at the level above,
    /// or null when the subtree absorbed the record.
    fn insert_rec(self: *Index, node_id: u32, key: []const u8, off: u64) ?Split {
        const node = self.page(node_id);
        if (node.is_leaf == 1) {
            if (self.fits(node_id, key.len + off_len)) {
                self.store_record(node_id, self.leaf_pos(node_id, key, off), .{ .key = key, .off = off });
                self.entry_count += 1;
                return null;
            }
            // remove_record leaves freed bytes dead at the top of the page,
            // so a churned leaf can run out of room with only a few live
            // records. Reclaim the dead bytes first; a leaf with a handful
            // of <= 1 KiB records always fits after that, so the split below
            // only ever sees a genuinely full leaf with count >= 2.
            self.repack_keep_prefix(node_id, node.count);
            if (self.fits(node_id, key.len + off_len)) {
                self.store_record(node_id, self.leaf_pos(node_id, key, off), .{ .key = key, .off = off });
                self.entry_count += 1;
                return null;
            }
            return self.split_leaf(node_id, key, off);
        }
        const child = self.descend_insert(node_id, key);
        const res = self.insert_rec(child, key, off) orelse return null;
        return self.insert_separator(node_id, res);
    }

    /// Split a full leaf around the record being inserted. The new record
    /// takes its place in the leaf's key order and the merged run is cut
    /// where the two halves come closest to equal cost, so the half the new
    /// record lands in is guaranteed to have room for it. Records equal to
    /// the boundary key may end up on both sides; lookups scan whole key
    /// bands, so that is fine. The right half moves to a new leaf and its
    /// first key is promoted.
    fn split_leaf(self: *Index, leaf_id: u32, key: []const u8, off: u64) Split {
        // insert_rec reclaims dead bytes before giving up on a page, so the
        // page is genuinely full here and holds at least two records --
        // which is what makes both halves below non-empty.
        const old_count = self.page(leaf_id).count;
        std.debug.assert(old_count >= 2);
        const pos = self.leaf_pos(leaf_id, key, off);
        const n = old_count + 1;

        var costs: [max_slots + 1]u32 = undefined;
        for (0..n) |m| {
            costs[m] = if (m == pos)
                record_cost(key.len + off_len)
            else
                self.slot_cost(leaf_id, @intCast(if (m < pos) m else m - 1));
        }
        const mid = @max(1, @min(balanced_cut(costs[0..n]), n - 1));

        const right_id = self.alloc_node();
        {
            const node = self.page_mut(leaf_id);
            const right = self.page_mut(right_id);
            right.is_leaf = 1;
            right.next = node.next;
            right.prev = leaf_id;
            right.parent = node.parent;
        }
        // Move the merged tail; spilled records keep their slab reference.
        var m: u32 = mid;
        while (m < n) : (m += 1) {
            const at: u32 = self.page(right_id).count;
            if (m == pos) {
                self.store_record(right_id, at, .{ .key = key, .off = off });
                continue;
            }
            const src: u32 = if (m < pos) m else m - 1;
            const slot = get_slot(self.page(leaf_id), src);
            self.store_record(right_id, at, .{
                .key = self.key_of(leaf_id, src),
                .off = self.off_of(leaf_id, src),
                .spill_off = if (slot.spill) slot.off else null,
            });
        }
        // Keep the merged head in the left page: its own records, plus the
        // new one when that is the half it belongs to.
        if (pos < mid) {
            self.repack_keep_prefix(leaf_id, mid - 1);
            self.store_record(leaf_id, pos, .{ .key = key, .off = off });
        } else {
            self.repack_keep_prefix(leaf_id, mid);
        }
        // Link the chain.
        const left = self.page_mut(leaf_id);
        if (left.next != 0) self.page_mut(left.next).prev = right_id;
        left.next = right_id;
        self.leaf_count += 1;
        self.entry_count += 1;

        const sep = self.stable_key(right_id, 0);
        return .{ .key = sep.key, .spill_off = sep.spill_off, .right = right_id };
    }

    /// Insert a promoted separator into an internal node, splitting it when
    /// full. Returns the next promotion, or null.
    fn insert_separator(self: *Index, node_id: u32, split: Split) ?Split {
        // The incoming key may live in the promo buffer, which a nested
        // split_internal (below) would overwrite with its own promoted key;
        // spilled keys already live in the immutable slab. Copy inline keys
        // to a stack buffer so they survive.
        var local: [inline_limit]u8 = undefined;
        const key: []const u8 = if (split.spill_off != null) split.key else blk: {
            @memcpy(local[0..split.key.len], split.key);
            break :blk local[0..split.key.len];
        };
        // As in a leaf: a dropped child leaves its separator's bytes dead
        // in the page, so reclaim before believing the page is full. This
        // is also what guarantees at least two separators at the split.
        if (!self.fits(node_id, key.len)) {
            self.repack_keep_prefix(node_id, self.page(node_id).count);
        }
        if (self.fits(node_id, key.len)) {
            self.store_record(node_id, self.separator_pos(node_id, key), .{
                .key = key,
                .child = split.right,
                .spill_off = split.spill_off,
            });
            self.page_mut(split.right).parent = node_id;
            return null;
        }
        return self.split_internal(node_id, key, split.spill_off, split.right);
    }

    /// Split a full internal node around the separator being inserted: the
    /// new separator joins the node's order, the merged run is cut where
    /// the halves come closest to equal cost (see split_leaf), and the
    /// separator at the cut is promoted -- its child becoming the right
    /// node's first child.
    fn split_internal(
        self: *Index,
        node_id: u32,
        key: []const u8,
        spill_off: ?u64,
        child: u32,
    ) Split {
        const old_count = self.page(node_id).count;
        std.debug.assert(old_count >= 2);
        const pos = self.separator_pos(node_id, key);
        const n = old_count + 1;

        var costs: [max_slots + 1]u32 = undefined;
        for (0..n) |m| {
            costs[m] = if (m == pos)
                record_cost(key.len)
            else
                self.slot_cost(node_id, @intCast(if (m < pos) m else m - 1));
        }
        // The cut is promoted rather than kept, so only the right half
        // needs a separator left over.
        const mid = @min(balanced_cut(costs[0..n]), n - 1);

        // The promoted key has to survive the repack below. A spilled key
        // already lives in the immutable slab (and can be far larger than
        // the promo buffer); an inline one is copied into promo.
        const promoted: StableKey = if (mid != pos)
            self.stable_key(node_id, if (mid < pos) mid else mid - 1)
        else if (spill_off != null)
            .{ .key = key, .spill_off = spill_off }
        else blk: {
            @memcpy(self.promo[0..key.len], key);
            break :blk .{ .key = self.promo[0..key.len], .spill_off = null };
        };

        const right_id = self.alloc_node();
        self.page_mut(right_id).is_leaf = 0;
        self.page_mut(right_id).parent = self.page(node_id).parent;
        // The promoted separator's child heads the right node.
        const mid_child: u32 = if (mid == pos)
            child
        else
            get_slot(self.page(node_id), if (mid < pos) mid else mid - 1).extra;
        self.page_mut(right_id).first_child = mid_child;
        self.page_mut(mid_child).parent = right_id;

        var m: u32 = mid + 1;
        while (m < n) : (m += 1) {
            const at: u32 = self.page(right_id).count;
            if (m == pos) {
                self.store_record(right_id, at, .{ .key = key, .child = child, .spill_off = spill_off });
                self.page_mut(child).parent = right_id;
                continue;
            }
            const src: u32 = if (m < pos) m else m - 1;
            const slot = get_slot(self.page(node_id), src);
            self.store_record(right_id, at, .{
                .key = self.key_of(node_id, src),
                .child = slot.extra,
                .spill_off = if (slot.spill) slot.off else null,
            });
            self.page_mut(slot.extra).parent = right_id;
        }
        if (pos < mid) {
            self.repack_keep_prefix(node_id, mid - 1);
            self.store_record(node_id, pos, .{ .key = key, .child = child, .spill_off = spill_off });
            self.page_mut(child).parent = node_id;
        } else {
            self.repack_keep_prefix(node_id, mid);
        }
        return .{ .key = promoted.key, .spill_off = promoted.spill_off, .right = right_id };
    }

    /// Remove the exact entry (key, id). Equal keys may span several
    /// leaves, so the entry is found by scanning its key band from the
    /// lower bound rather than by a straight descent.
    fn remove_entry(self: *Index, key: []const u8, off: u64) bool {
        var it = self.seek(key);
        while (it.next()) |e| {
            const c = std.mem.order(u8, e.key, key);
            if (c == .gt) return false; // band ended, not found
            if (c != .eq) continue;
            if (e.off == off) {
                self.leaf_remove(it.leaf, it.slot - 1);
                return true;
            }
        }
        return false;
    }

    /// Remove leaf slot and, when the leaf empties, unlink it and drop it
    /// from its parent, cascading up while internal nodes empty.
    fn leaf_remove(self: *Index, leaf_id: u32, slot_idx: u32) void {
        self.remove_record(leaf_id, slot_idx);
        self.entry_count -= 1;
        const node = self.page(leaf_id);
        if (node.count > 0) return;
        // Empty leaf: unlink and drop from the parent (unless it is the
        // root, which stays as the empty root leaf).
        if (leaf_id == self.root) return;
        if (node.prev != 0) self.page_mut(node.prev).next = node.next;
        if (node.next != 0) self.page_mut(node.next).prev = node.prev;
        if (leaf_id == self.first_leaf) self.first_leaf = node.next;
        self.leaf_count -= 1;
        var child = leaf_id;
        var parent = node.parent;
        while (parent != 0) {
            self.drop_child(parent, child);
            const pnode = self.page(parent);
            if (pnode.count == 0 and pnode.first_child == 0) {
                if (parent == self.root) {
                    self.replace_root_with_leaf();
                    return;
                }
                child = parent;
                parent = pnode.parent;
            } else {
                return;
            }
        }
    }

    /// Remove `child` from `parent`'s child list. Infallible; `child`'s
    /// node is abandoned in place (its page is simply never referenced
    /// again — node ids are append-only, so this is a leak of at most the
    /// peak tree size, exactly what the old entry array's capacity was).
    fn drop_child(self: *Index, parent_id: u32, child_id: u32) void {
        const pnode = self.page_mut(parent_id);
        if (pnode.first_child == child_id) {
            if (pnode.count > 0) {
                const s0 = get_slot(pnode, 0);
                pnode.first_child = s0.extra;
                self.remove_record(parent_id, 0);
            } else {
                pnode.first_child = 0;
            }
            return;
        }
        var j: u32 = 0;
        while (j < pnode.count) : (j += 1) {
            if (get_slot(pnode, j).extra == child_id) {
                self.remove_record(parent_id, j);
                return;
            }
        }
    }

    /// The internal root emptied: swap in a fresh empty leaf as the root.
    fn replace_root_with_leaf(self: *Index) void {
        // Removal reserves no capacity, so this must not allocate a node:
        // re-use the emptied root page as the empty root leaf.
        self.page_mut(self.root).* = empty_node(1);
        self.first_leaf = self.root;
        self.leaf_count = 1;
        self.depth = 0;
    }

    /// Pack the sorted staging array into a fresh tree: leaves filled in
    /// key order, interior levels built bottom-up. Frees the staging keys
    /// and clears the staging array.
    fn pack_tree(self: *Index, gpa: std.mem.Allocator) !void {
        defer {
            for (self.staging.items) |e| gpa.free(e.key);
            self.staging.clearRetainingCapacity();
        }
        if (self.staging.items.len == 0) return;
        // A fresh tree replaces whatever was here; the old pages are
        // abandoned in place, like any other dropped node.
        self.leaf_count = 0;
        self.depth = 0;

        const Level = struct {
            id: u32,
            first_key: []const u8, // aliases a leaf's page or the slab
        };
        var level: std.ArrayListUnmanaged(Level) = .empty;
        defer level.deinit(gpa);

        // Leaves.
        var prev: u32 = 0;
        var lit = self.staging.items;
        while (lit.len > 0) {
            const leaf = try self.alloc_node_grow(gpa);
            self.page_mut(leaf).is_leaf = 1;
            const first_key = lit[0].key;
            // Fill until the next record would not fit.
            var used: usize = 0;
            var slots: usize = 0;
            var n: usize = 0;
            while (n < lit.len) : (n += 1) {
                const rec_len: u64 = lit[n].key.len + off_len;
                const inline_bytes: usize = if (rec_len > inline_limit) 0 else @intCast(rec_len);
                if ((slots + 1) * slot_size + used + inline_bytes > page_data) break;
                self.store_record(leaf, @intCast(slots), .{ .key = lit[n].key, .off = lit[n].off });
                slots += 1;
                used += inline_bytes;
            }
            self.page_mut(leaf).prev = prev;
            if (prev != 0) self.page_mut(prev).next = leaf;
            prev = leaf;
            self.leaf_count += 1;
            try level.append(gpa, .{ .id = leaf, .first_key = first_key });
            lit = lit[n..];
        }
        self.page_mut(prev).next = 0;
        self.first_leaf = level.items[0].id;

        // Interior levels: group the level below into internal nodes whose
        // separators are the first keys of its members.
        while (level.items.len > 1) {
            var next: std.ArrayListUnmanaged(Level) = .empty;
            defer next.deinit(gpa);
            var i: usize = 0;
            while (i < level.items.len) {
                const node = try self.alloc_node_grow(gpa);
                self.page_mut(node).first_child = level.items[i].id;
                self.page_mut(level.items[i].id).parent = node;
                const first_key = level.items[i].first_key;
                var used: usize = 0;
                var slots: usize = 0;
                var j = i + 1;
                while (j < level.items.len) : (j += 1) {
                    const rec_len: u64 = level.items[j].first_key.len;
                    const inline_bytes: usize = if (rec_len > inline_limit) 0 else @intCast(rec_len);
                    if ((slots + 1) * slot_size + used + inline_bytes > page_data) break;
                    self.store_record(node, @intCast(slots), .{
                        .key = level.items[j].first_key,
                        .child = level.items[j].id,
                    });
                    self.page_mut(level.items[j].id).parent = node;
                    slots += 1;
                    used += inline_bytes;
                }
                try next.append(gpa, .{ .id = node, .first_key = first_key });
                i = j;
            }
            level.deinit(gpa);
            level = next;
            next = .empty; // ownership moved to `level`; the defer frees nothing
            self.depth += 1;
        }
        self.root = level.items[0].id;
        self.page_mut(self.root).parent = 0;
        self.entry_count = self.staging.items.len;
    }
};

/// The index whose key pattern is exactly `key_pairs` (same paths, same
/// order, same directions), or null. Keeps the IndexKey layout — and what
/// counts as a match — inside this module.
pub fn find_by_key_pattern(indexes: []const *Index, key_pairs: []const bson.Pair) ?*const Index {
    for (indexes) |ix| {
        if (ix.keys.len != key_pairs.len) continue;
        var match = true;
        for (ix.keys, key_pairs) |k, kp| {
            if (!std.mem.eql(u8, k.path, kp.key) or k.descending != descending(kp.value)) {
                match = false;
                break;
            }
        }
        if (match) return ix;
    }
    return null;
}

pub const SpecError = error{ InvalidIndexSpec, TtlOnCompoundIndex, InvalidExpireAfterSeconds, OutOfMemory };

/// Parse {key: {...}, name?, unique?, sparse?, expireAfterSeconds?} from a
/// spec document — the form drivers send and the form the log stores.
/// The error set is inferred rather than `SpecError`, because building an index
/// now allocates pages in the data file and so can fail on file growth too.
pub fn parse_spec(gpa: std.mem.Allocator, pager: *pgr.Pager, spec: *const bson.Document) !Index {
    const key_value = bson.get_pair(spec.pairs, "key") orelse return error.InvalidIndexSpec;
    const key_pairs = switch (key_value) {
        .doc => |p| p,
        else => return error.InvalidIndexSpec,
    };
    if (key_pairs.len == 0 or key_pairs.len > max_index_keys) return error.InvalidIndexSpec;
    var keys: [max_index_keys]IndexKey = undefined;
    for (key_pairs, 0..) |p, i| {
        const ok_mag = switch (p.value) {
            .int32 => |n| n == 1 or n == -1,
            .int64 => |n| n == 1 or n == -1,
            .double => |n| n == 1.0 or n == -1.0,
            else => false,
        };
        if (!ok_mag) return error.InvalidIndexSpec;
        keys[i] = .{ .path = p.key, .descending = descending(p.value) };
    }
    var unique = false;
    var sparse = false;
    if (bson.get_pair(spec.pairs, "unique")) |v| unique = query.truthy(v);
    if (bson.get_pair(spec.pairs, "sparse")) |v| sparse = query.truthy(v);
    var ttl: ?i64 = null;
    if (bson.get_pair(spec.pairs, "expireAfterSeconds")) |v| {
        ttl = try expire_after_seconds(v);
        // MongoDB's rule: TTL is a single-field index option. A compound key
        // would leave it ambiguous which component dates the document.
        if (key_pairs.len != 1) return error.TtlOnCompoundIndex;
    }
    const name_value = bson.get_pair(spec.pairs, "name") orelse {
        const nm = try default_name(gpa, key_pairs);
        defer gpa.free(nm);
        return Index.init(gpa, pager, nm, keys[0..key_pairs.len], unique, sparse, ttl);
    };
    const name = switch (name_value) {
        .string => |s| s,
        else => return error.InvalidIndexSpec,
    };
    return Index.init(gpa, pager, name, keys[0..key_pairs.len], unique, sparse, ttl);
}

/// MongoDB's bound on expireAfterSeconds. Keeping it means a TTL always
/// round-trips as an int32, exactly as a real server reports it.
pub const max_expire_after_seconds: i64 = 2147483647;

/// The seconds in an `expireAfterSeconds` option: integral and within
/// [0, max_expire_after_seconds]. 0 is legal (expire at the stored instant);
/// a negative, fractional, out-of-range or non-numeric value is not. A
/// double is accepted because that is what a JavaScript driver sends for a
/// plain number — and the range check runs before the conversion, so
/// @intFromFloat is always defined.
fn expire_after_seconds(v: bson.Value) error{InvalidExpireAfterSeconds}!i64 {
    const secs: i64 = switch (v) {
        .int32 => |n| n,
        .int64 => |n| n,
        .double => |d| blk: {
            if (!std.math.isFinite(d) or @trunc(d) != d) return error.InvalidExpireAfterSeconds;
            if (d < 0 or d > @as(f64, @floatFromInt(max_expire_after_seconds))) return error.InvalidExpireAfterSeconds;
            break :blk @intFromFloat(d);
        },
        else => return error.InvalidExpireAfterSeconds,
    };
    if (secs < 0 or secs > max_expire_after_seconds) return error.InvalidExpireAfterSeconds;
    return secs;
}

/// Whether a key-pattern direction value means descending. The single
/// definition of what -1 means in a key pattern, shared with dropIndexes'
/// key-pattern matching.
pub fn descending(v: bson.Value) bool {
    return switch (v) {
        .int32 => |n| n < 0,
        .int64 => |n| n < 0,
        .double => |n| n < 0,
        else => false,
    };
}

/// MongoDB's default index name: a_1_b_-1.
fn default_name(gpa: std.mem.Allocator, key_pairs: []const bson.Pair) ![]u8 {
    var out: std.ArrayListUnmanaged(u8) = .empty;
    errdefer out.deinit(gpa);
    for (key_pairs, 0..) |p, i| {
        if (i > 0) try out.append(gpa, '_');
        try out.appendSlice(gpa, p.key);
        try out.append(gpa, '_');
        if (descending(p.value)) {
            try out.append(gpa, '-');
        }
        try out.append(gpa, '1');
    }
    return out.toOwnedSlice(gpa);
}

// ---------------------------------------------------------------------------
// Comparison
// ---------------------------------------------------------------------------

/// Encoded-key byte order, tie-broken by the id bytes. This is the total
/// order batches are sorted into before packing.
///
/// bson.encode_key guarantees this reproduces component-wise bson.compare
/// exactly, so ordering is a memcmp. Key direction is deliberately not
/// applied here: uniqueness and candidate generation are direction
/// 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);
}

fn entry_less(_: void, a: Entry, b: Entry) bool {
    return compare_entries(a, b) == .lt;
}

/// Bulk-build ordering: by key, then by document offset so the order is total
/// and the adjacent-pair duplicate scan in finish_bulk sees a document's own
/// equal keys next to each other.
fn staged_less(_: void, a: Staged, b: Staged) bool {
    const o = std.mem.order(u8, a.key, b.key);
    if (o != .eq) return o == .lt;
    return a.off < b.off;
}

/// Order of an encoded search prefix against a stored key, the tree's
/// lookup comparator: `.eq` when the key starts with the prefix, so a
/// partial key matches a whole band; `.lt` when the key sorts strictly
/// before the prefix; `.gt` when after.
///
/// Comparing raw bytes is sound because every column encoding is
/// self-delimiting, so a prefix of the encoded key is exactly the encoding
/// of its leading columns.
fn cmp_prefix(prefix: []const u8, key: []const u8) std.math.Order {
    const n = @min(prefix.len, key.len);
    const o = std.mem.order(u8, prefix[0..n], key[0..n]);
    if (o != .eq) return o;
    // The key ran out first, so it sorts below the prefix.
    if (prefix.len > key.len) return .gt;
    return .eq;
}

/// Advance an odometer of positions, each bounded by the matching `limits`
/// entry. Returns false once it wraps, i.e. the product is exhausted.
fn advance_choice(choice: []usize, limits: []const usize) bool {
    var i = choice.len;
    while (i > 0) {
        i -= 1;
        choice[i] += 1;
        if (choice[i] < limits[i]) return true;
        choice[i] = 0;
    }
    return false;
}

// ---------------------------------------------------------------------------
// Query planning
// ---------------------------------------------------------------------------

const Clause = struct {
    path: []const u8,
    value: bson.Value,
};

/// Flatten top-level pairs and $and members into AND-ed predicates. Every
/// other top-level operator ($or, $nor, ...) is skipped: the full filter is
/// re-applied later, so a usable sibling still yields a valid superset.
fn flatten_clauses(
    gpa: std.mem.Allocator,
    pairs: []const bson.Pair,
    out: *std.ArrayListUnmanaged(Clause),
) !void {
    for (pairs) |p| {
        if (p.key.len > 0 and p.key[0] == '$') {
            if (std.mem.eql(u8, p.key, "$and")) {
                const members = switch (p.value) {
                    .array => |a| a,
                    else => continue,
                };
                for (members) |m| {
                    const mp = switch (m) {
                        .doc => |d| d,
                        else => continue,
                    };
                    try flatten_clauses(gpa, mp, out);
                }
            }
            continue;
        }
        try out.append(gpa, .{ .path = p.key, .value = p.value });
    }
}

/// Per-index-component info extracted from the filter clauses.
const CompInfo = struct {
    eq: ?bson.Value = null,
    in_values: ?[]const bson.Value = null,
    lo: ?bson.Value = null,
    lo_incl: bool = false,
    hi: ?bson.Value = null,
    hi_incl: bool = false,
};

/// Extract the usable constraint from one clause value: a bare non-regex
/// equality, or an operator doc whose operators are all in
/// {$eq, $in, $gt, $gte, $lt, $lte}. Anything else leaves the info
/// untouched (unusable — the full filter is re-applied anyway).
fn analyze_clause(value: bson.Value, info: *CompInfo) void {
    if (value == .doc) {
        const pairs = value.doc;
        if (pairs.len > 0 and !query.all_operator_keys(pairs)) {
            // Bare document equality: {a: {n: 1}} compares the whole doc.
            info.eq = value;
            return;
        }
        for (pairs) |p| {
            if (std.mem.eql(u8, p.key, "$eq")) {
                info.eq = p.value;
            } else if (std.mem.eql(u8, p.key, "$in")) {
                info.in_values = switch (p.value) {
                    .array => |a| a,
                    else => return,
                };
            } else if (std.mem.eql(u8, p.key, "$gt")) {
                info.lo = p.value;
                info.lo_incl = false;
            } else if (std.mem.eql(u8, p.key, "$gte")) {
                info.lo = p.value;
                info.lo_incl = true;
            } else if (std.mem.eql(u8, p.key, "$lt")) {
                info.hi = p.value;
                info.hi_incl = false;
            } else if (std.mem.eql(u8, p.key, "$lte")) {
                info.hi = p.value;
                info.hi_incl = true;
            } else {
                info.eq = null;
                info.in_values = null;
                info.lo = null;
                info.hi = null;
                return;
            }
        }
        return;
    }
    if (value == .regex) return;
    info.eq = value;
}

/// A candidate-generation plan for one index: `lookup_keys` is the
/// cartesian product of the leading equality/$in components (each key has
/// `key_len` Values aliasing the filter), plus an optional range on the
/// next component.
pub const Plan = struct {
    index: *const Index,
    lookup_keys: std.ArrayListUnmanaged([]const bson.Value),
    lo: ?bson.Value,
    lo_incl: bool,
    hi: ?bson.Value,
    hi_incl: bool,
    /// The candidates come out already in the requested sort order, so the
    /// caller can skip sorting and stop as soon as the page is full.
    provides_sort: bool = false,
    /// 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);
    }

    /// How many leading index components the lookup keys pin down. Every key
    /// is built with the same component count, and there is always at least
    /// one (a pure range plan appends a single empty key).
    pub fn key_len(self: *const Plan) usize {
        return self.lookup_keys.items[0].len;
    }

    /// Collect the candidate ids, sorted and deduplicated. Range scans can
    /// return the same id non-adjacently (a doc with {tags: ["a","b"]}
    /// contributes two entries inside one range), so adjacent-dup skipping
    /// would be wrong.
    ///
    /// Duplicates are only possible from a multikey index (one document,
    /// several entries) or from several lookup keys (whose ranges can be the
    /// same key repeated, as in {$in: [1, 1]}); the common single-key lookup
    /// on a non-multikey index skips the pass entirely.
    pub fn search(
        self: *const Plan,
        gpa: std.mem.Allocator,
        out: *std.ArrayListUnmanaged(u64),
    ) !void {
        for (self.lookup_keys.items) |key| {
            if (self.lo == null and self.hi == null) {
                try self.index.lookup_eq(gpa, key, out);
            } else {
                try self.index.lookup_range(gpa, key, self.lo, self.lo_incl, self.hi, self.hi_incl, out);
            }
        }
        // Entries come out of the leaves in key order, so a forward scan is
        // already sorted; a backward one just reads it the other way. This
        // must happen before the dedupe pass below, which sorts by offset
        // and would destroy the order — the conditions that set provides_sort
        // are exactly the ones under which that pass is skipped.
        if (self.provides_sort and self.backward) std.mem.reverse(u64, out.items);

        const may_repeat = self.index.multikey or self.lookup_keys.items.len > 1;
        if (may_repeat and out.items.len > 1) {
            std.mem.sort(u64, out.items, {}, std.sort.asc(u64));
            var w: usize = 1;
            for (out.items[1..]) |id| {
                if (id != out.items[w - 1]) {
                    out.items[w] = id;
                    w += 1;
                }
            }
            out.items.len = w;
        }
    }
};

/// Pick the index (if any) that can generate a superset of the matching
/// documents: the one covering the longest leading run of equality/$in
/// predicates, optionally with a range on the next key. Returns null when
/// nothing usable remains — the caller scans.
///
/// `id_ix` is the collection's implicit `_id_` index (kept separate from
/// the secondary `indexes` list so the listing/drop commands stay
/// unchanged). Every document has an `_id` and the index is not sparse, so
/// a full scan of it cannot miss a document — which is what the sort
/// planner's full-scan plan relies on. The encoded keys are canonical, so
/// `_id` equality on compare-equal values (int32 1, int64 1, double 1.0)
/// finds the same entries.
pub fn plan(
    gpa: std.mem.Allocator,
    id_ix: ?*const Index,
    indexes: []const *Index,
    filter: []const bson.Pair,
    sort: []const query.SortKey,
) !?Plan {
    if (id_ix == null and indexes.len == 0) return null;
    var clauses: std.ArrayListUnmanaged(Clause) = .empty;
    defer clauses.deinit(gpa);
    try flatten_clauses(gpa, filter, &clauses);

    var best: ?Plan = null;
    if (id_ix) |idx| {
        if (try evaluate_index(gpa, idx, clauses.items, sort)) |cand| {
            best = cand;
        }
    }
    for (indexes) |ix| {
        var cand = (try evaluate_index(gpa, ix, clauses.items, sort)) orelse continue;
        if (best) |b| {
            if (plan_better(&cand, &b)) {
                best.?.deinit(gpa);
                best = cand;
            } else {
                cand.deinit(gpa);
            }
        } else {
            best = cand;
        }
    }
    return best;
}

fn plan_better(a: *const Plan, b: *const Plan) bool {
    if (a.key_len() != b.key_len()) return a.key_len() > b.key_len();
    // Same selectivity: providing the sort saves ordering the whole result.
    if (a.provides_sort != b.provides_sort) return a.provides_sort;
    const a_range = a.lo != null or a.hi != null;
    const b_range = b.lo != null or b.hi != null;
    if (a_range != b_range) return a_range;
    return a.lookup_keys.items.len < b.lookup_keys.items.len;
}

/// Whether scanning `ix` in order satisfies `sort`, and if so whether that
/// means reading the tree backwards.
///
/// The sort keys must line up with the index components that follow the
/// equality-pinned prefix: those components are fixed to one value each, so
/// they do not affect the order of what remains. Directions must agree
/// uniformly — every key the same way round, or every key opposite — since
/// the leaves can only be read forwards or backwards.
///
/// A multikey index is excluded: it emits a document once per indexed
/// value, so its order is not an order on documents.
fn index_provides_sort(ix: *const Index, run: usize, sort: []const query.SortKey) ?bool {
    if (sort.len == 0 or ix.multikey) return null;
    if (run + sort.len > ix.keys.len) return null;

    const backward = sort[0].descending != ix.keys[run].descending;
    for (sort, 0..) |sk, i| {
        const k = ix.keys[run + i];
        if (!std.mem.eql(u8, sk.path, k.path)) return null;
        if ((sk.descending != k.descending) != backward) return null;
    }
    return backward;
}

fn evaluate_index(
    gpa: std.mem.Allocator,
    ix: *const Index,
    clauses: []const Clause,
    sort: []const query.SortKey,
) !?Plan {
    const n = ix.keys.len;
    var infos: [max_index_keys]CompInfo = undefined;
    for (0..n) |i| {
        var info = CompInfo{};
        for (clauses) |cl| {
            if (std.mem.eql(u8, cl.path, ix.keys[i].path)) analyze_clause(cl.value, &info);
        }
        infos[i] = info;
    }

    // Longest leading run of equality/$in components.
    var run: usize = 0;
    while (run < n and (infos[run].eq != null or infos[run].in_values != null)) run += 1;
    var lo: ?bson.Value = null;
    var lo_incl = false;
    var hi: ?bson.Value = null;
    var hi_incl = false;
    if (run < n) {
        lo = infos[run].lo;
        lo_incl = infos[run].lo_incl;
        hi = infos[run].hi;
        hi_incl = infos[run].hi_incl;
    }
    const sort_dir = index_provides_sort(ix, run, sort);
    // With no filter to narrow anything down, a full index scan is only
    // worth it when it is what produces the ordering — otherwise scanning
    // the docs map directly is strictly cheaper.
    if (run == 0 and lo == null and hi == null and sort_dir == null) return null;

    // A two-sided range on a multikey index can under-approximate: a doc
    // like {a: [1, 2]} satisfies {a: {$gt: 5, $lt: 25}} with the array for
    // the lower bound (rank 5 > 5) and an element for the upper (1 < 25),
    // so no single entry lies inside (5, 25) — the range scan would miss
    // it. One-sided ranges are safe: whichever candidate satisfies the
    // bound is itself an entry inside it. Equality/$in are unaffected.
    if (ix.multikey and lo != null and hi != null) return null;

    // A sparse index skips documents missing a field; {a: null} would
    // otherwise miss them. Never use a sparse index for a null component.
    if (ix.sparse) {
        for (0..run) |i| {
            if (infos[i].eq) |v| {
                if (v == .null) return null;
            }
            if (infos[i].in_values) |list| {
                for (list) |m| if (m == .null) return null;
            }
        }
        if (lo) |v| if (v == .null) return null;
        if (hi) |v| if (v == .null) return null;
    }

    var counts: [max_index_keys]usize = undefined;
    for (0..run) |i| {
        counts[i] = if (infos[i].eq != null) 1 else infos[i].in_values.?.len;
    }
    var combos: u64 = 1;
    for (0..run) |i| {
        // A single huge $in list (or a product over 100) falls back to a
        // scan; checking the factor first keeps the product from wrapping.
        if (counts[i] > max_combos) return null;
        combos *= counts[i];
        if (combos > max_combos) return null;
    }

    var pl = Plan{
        .index = ix,
        .lookup_keys = .empty,
        .lo = lo,
        .lo_incl = lo_incl,
        .hi = hi,
        .hi_incl = hi_incl,
    };
    // Several lookup keys ($in) concatenate disjoint ranges, whose
    // concatenation is not ordered.
    pl.provides_sort = sort_dir != null and combos == 1;
    pl.backward = sort_dir orelse false;
    errdefer pl.deinit(gpa);
    if (run == 0) {
        const empty = try gpa.alloc(bson.Value, 0);
        errdefer gpa.free(empty);
        try pl.lookup_keys.append(gpa, empty);
    } else {
        var choice: [max_index_keys]usize = undefined;
        @memset(choice[0..run], 0);
        while (true) {
            const key = try gpa.alloc(bson.Value, run);
            errdefer gpa.free(key);
            // An eq component contributes its single value; an $in
            // component the choice-th member. (Never materialize the
            // single-value option as a temporary array: it would dangle.)
            for (0..run) |i| {
                key[i] = if (infos[i].eq) |v| v else infos[i].in_values.?[choice[i]];
            }
            try pl.lookup_keys.append(gpa, key);
            if (!advance_choice(choice[0..run], counts[0..run])) break;
        }
    }
    return pl;
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

const testing = std.testing;

/// Serialize a fabricated doc's pairs to canonical bytes (owned by the
/// caller), since entry generation now reads stored documents as bytes.
fn bytes_of(gpa: std.mem.Allocator, pairs: []const bson.Pair) ![]u8 {
    var out: std.ArrayListUnmanaged(u8) = .empty;
    defer out.deinit(gpa);
    try bson.write_doc(pairs, gpa, &out);
    return out.toOwnedSlice(gpa);
}

/// A fabricated tree document for functions that still parse specs from
/// pairs (parse_spec). Never deinit'd — mirrors the old doc_of.
fn doc_of(pairs: []const bson.Pair) bson.Document {
    return .{ .arena = undefined, .pairs = pairs };
}

/// One data file shared by every test in this file, created on first use.
///
/// Shared rather than per-test because a pager needs an `io` and a temp path,
/// and threading both through twenty tests would bury what each is about. The
/// tests are independent regardless: each Index owns its own pages, and nothing
/// here frees any, so they cannot interfere. Allocated from the page allocator
/// so it is not reported as a leak by whichever test happens to create it.
var test_pager_state: ?struct {
    threaded: *std.Io.Threaded,
    pager: *pgr.Pager,
} = null;

fn test_pager() *pgr.Pager {
    if (test_pager_state) |st| return st.pager;
    const a = std.heap.page_allocator;
    const threaded = a.create(std.Io.Threaded) catch @panic("test pager");
    threaded.* = .init_single_threaded;
    const pg = a.create(pgr.Pager) catch @panic("test pager");
    std.Io.Dir.cwd().deleteFile(threaded.io(), ".zig-cache/index-test.data") catch {};
    pg.* = pgr.Pager.open(a, threaded.io(), ".zig-cache/index-test.data", .{}) catch @panic("test pager");
    test_pager_state = .{ .threaded = threaded, .pager = pg };
    return pg;
}

fn simple_index(
    gpa: std.mem.Allocator,
    pager: *pgr.Pager,
    paths: []const []const u8,
    unique: bool,
    sparse: bool,
) !Index {
    var keys: [max_index_keys]IndexKey = undefined;
    for (paths, 0..) |p, i| keys[i] = .{ .path = p, .descending = false };
    return Index.init(gpa, pager, "test", keys[0..paths.len], unique, sparse, null);
}

/// Look up the documents under `key` and compare with the expected offsets.
/// A leaf record's payload is a slab offset now, so tests identify documents by
/// small distinct numbers rather than by byte-string ids.
fn expect_offs(
    gpa: std.mem.Allocator,
    ix: *const Index,
    key: []const bson.Value,
    expected: []const u64,
) !void {
    var out: std.ArrayListUnmanaged(u64) = .empty;
    defer out.deinit(gpa);
    try ix.lookup_eq(gpa, key, &out);
    std.mem.sort(u64, out.items, {}, std.sort.asc(u64));
    try testing.expectEqualSlices(u64, expected, out.items);
}

/// The tree's entries as (key, id) references, in iteration order.
fn refs_of(gpa: std.mem.Allocator, ix: *const Index, out: *std.ArrayListUnmanaged(EntryRef)) !void {
    var it = ix.iter();
    while (it.next()) |e| try out.append(gpa, e);
}

/// EntryRef ordering: key bytes, then the document's slab offset.
fn ref_lt(a: EntryRef, b: EntryRef) bool {
    const o = std.mem.order(u8, a.key, b.key);
    if (o != .eq) return o == .lt;
    return a.off < b.off;
}

test "entries sort across numeric types and string/null/objectid" {
    const gpa = testing.allocator;
    var ix = try simple_index(gpa, test_pager(), &.{"a"}, false, false);
    defer ix.deinit(gpa);

    const d_int = try bytes_of(gpa, &.{ .{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "a", .value = .{ .int32 = 5 } } });
    defer gpa.free(d_int);
    const d_dbl = try bytes_of(gpa, &.{ .{ .key = "_id", .value = .{ .int32 = 2 } }, .{ .key = "a", .value = .{ .double = 5.0 } } });
    defer gpa.free(d_dbl);
    const d_str = try bytes_of(gpa, &.{ .{ .key = "_id", .value = .{ .int32 = 3 } }, .{ .key = "a", .value = .{ .string = "b" } } });
    defer gpa.free(d_str);
    const d_nul = try bytes_of(gpa, &.{ .{ .key = "_id", .value = .{ .int32 = 4 } }, .{ .key = "a", .value = .null } });
    defer gpa.free(d_nul);
    const d_oid = try bytes_of(gpa, &.{ .{ .key = "_id", .value = .{ .int32 = 5 } }, .{ .key = "a", .value = .{ .object_id = [_]u8{ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11 } } } });
    defer gpa.free(d_oid);
    _ = try ix.add_doc(gpa, d_int, 1, true);
    _ = try ix.add_doc(gpa, d_dbl, 2, true);
    _ = try ix.add_doc(gpa, d_str, 3, true);
    _ = try ix.add_doc(gpa, d_nul, 4, true);
    _ = try ix.add_doc(gpa, d_oid, 5, true);

    // An int64 query finds both the int32 and double entries: compare-equal.
    try expect_offs(gpa, &ix, &.{.{ .int64 = 5 }}, &.{ 1, 2 });
    try expect_offs(gpa, &ix, &.{.null}, &.{4});
    try expect_offs(gpa, &ix, &.{.{ .string = "b" }}, &.{3});
    try expect_offs(gpa, &ix, &.{.{ .object_id = [_]u8{ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11 } }}, &.{5});

    // Full order: null, 5, 5, "b", oid — iteration respects it (equal
    // keys tie-break by id, as the leaf position rule does).
    var refs: std.ArrayListUnmanaged(EntryRef) = .empty;
    defer refs.deinit(gpa);
    try refs_of(gpa, &ix, &refs);
    try testing.expectEqual(@as(usize, 5), refs.items.len);
    try testing.expect(ref_lt(refs.items[0], refs.items[1]));
    try testing.expect(ref_lt(refs.items[1], refs.items[2]));
    try testing.expect(ref_lt(refs.items[2], refs.items[3]));
}

test "missing field is indexed as null; sparse skips the document" {
    const gpa = testing.allocator;
    var ix = try simple_index(gpa, test_pager(), &.{"a"}, false, false);
    defer ix.deinit(gpa);
    const d = try bytes_of(gpa, &.{.{ .key = "_id", .value = .{ .int32 = 1 } }});
    defer gpa.free(d);
    _ = try ix.add_doc(gpa, d, 1, true);
    try testing.expectEqual(@as(usize, 1), ix.count());
    try expect_offs(gpa, &ix, &.{.null}, &.{1});

    var sp = try simple_index(gpa, test_pager(), &.{"a"}, false, true);
    defer sp.deinit(gpa);
    _ = try sp.add_doc(gpa, d, 2, true);
    try testing.expectEqual(@as(usize, 0), sp.count());
}

test "multikey expansion indexes the array and its elements" {
    const gpa = testing.allocator;
    var ix = try simple_index(gpa, test_pager(), &.{"tags"}, false, false);
    defer ix.deinit(gpa);

    const d = try bytes_of(gpa, &.{ .{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "tags", .value = .{ .array = &.{ .{ .string = "a" }, .{ .string = "b" } } } } });
    defer gpa.free(d);
    _ = try ix.add_doc(gpa, d, 1, true);

    // 3 entries: the array itself, "a", "b".
    try testing.expectEqual(@as(usize, 3), ix.count());
    // Element query.
    try expect_offs(gpa, &ix, &.{.{ .string = "a" }}, &.{1});
    try expect_offs(gpa, &ix, &.{.{ .string = "b" }}, &.{1});
    // Whole-array equality — the reason arrays are indexed twice.
    try expect_offs(gpa, &ix, &.{.{ .array = &.{ .{ .string = "a" }, .{ .string = "b" } } }}, &.{1});
    try expect_offs(gpa, &ix, &.{.{ .array = &.{ .{ .string = "b" }, .{ .string = "a" } } }}, &.{});
}

test "per-document dedup keeps {a: [1,1]} under a unique index" {
    const gpa = testing.allocator;
    var ix = try simple_index(gpa, test_pager(), &.{"a"}, true, false);
    defer ix.deinit(gpa);
    const d = try bytes_of(gpa, &.{ .{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "a", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 1 } } } } });
    defer gpa.free(d);
    _ = try ix.add_doc(gpa, d, 1, true);
    // Entries after dedup: the array itself and one element.
    try testing.expectEqual(@as(usize, 2), ix.count());
}

test "parallel arrays are rejected" {
    const gpa = testing.allocator;
    var ix = try simple_index(gpa, test_pager(), &.{ "a", "b" }, false, false);
    defer ix.deinit(gpa);
    const d = try bytes_of(gpa, &.{
        .{ .key = "_id", .value = .{ .int32 = 1 } },
        .{ .key = "a", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 2 } } } },
        .{ .key = "b", .value = .{ .array = &.{ .{ .int32 = 3 }, .{ .int32 = 4 } } } },
    });
    defer gpa.free(d);
    try testing.expectError(error.ParallelArrays, ix.add_doc(gpa, d, 1, true));

    // One array path is fine.
    const ok = try bytes_of(gpa, &.{
        .{ .key = "_id", .value = .{ .int32 = 2 } },
        .{ .key = "a", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 2 } } } },
        .{ .key = "b", .value = .{ .int32 = 3 } },
    });
    defer gpa.free(ok);
    _ = try ix.add_doc(gpa, ok, 2, true);
    try testing.expectEqual(@as(usize, 3), ix.count());
}

test "unique conflict across documents, replace of own entries allowed" {
    const gpa = testing.allocator;
    var ix = try simple_index(gpa, test_pager(), &.{"a"}, true, false);
    defer ix.deinit(gpa);

    const d1 = try bytes_of(gpa, &.{ .{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "a", .value = .{ .int32 = 10 } } });
    defer gpa.free(d1);
    _ = try ix.add_doc(gpa, d1, 1, true);

    const d2 = try bytes_of(gpa, &.{ .{ .key = "_id", .value = .{ .int32 = 2 } }, .{ .key = "a", .value = .{ .int32 = 10 } } });
    defer gpa.free(d2);
    try testing.expectError(error.DuplicateKeyIndex, ix.add_doc(gpa, d2, 2, true));

    // A replace keeps its own key: remove old entries first (the engine's
    // evict_doc does this), then add the new ones.
    const d1b = try bytes_of(gpa, &.{ .{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "a", .value = .{ .int32 = 20 } } });
    defer gpa.free(d1b);
    ix.remove_off(1);
    _ = try ix.add_doc(gpa, d1b, 1, true);
    try expect_offs(gpa, &ix, &.{.{ .int32 = 20 }}, &.{1});
    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, test_pager(), &.{"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, test_pager(), &.{"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
    // docs hashmap is gone, and a prefix match would answer yes for a longer
    // key that merely starts the same way. Compound keys make that concrete --
    // the encoding of {a: 1} is a prefix of the encoding of {a: 1, b: 2}.
    const gpa = testing.allocator;
    var ix = try simple_index(gpa, test_pager(), &.{ "a", "b" }, false, false);
    defer ix.deinit(gpa);

    const d = try bytes_of(gpa, &.{
        .{ .key = "_id", .value = .{ .int32 = 1 } },
        .{ .key = "a", .value = .{ .int32 = 1 } },
        .{ .key = "b", .value = .{ .int32 = 2 } },
    });
    defer gpa.free(d);
    _ = try ix.add_doc(gpa, d, 1, false);

    var enc_full: std.ArrayListUnmanaged(u8) = .empty;
    defer enc_full.deinit(gpa);
    try bson.encode_key(.{ .int32 = 1 }, gpa, &enc_full);
    var enc_prefix: std.ArrayListUnmanaged(u8) = .empty;
    defer enc_prefix.deinit(gpa);
    try enc_prefix.appendSlice(gpa, enc_full.items);
    try bson.encode_key(.{ .int32 = 2 }, gpa, &enc_full);

    // The full two-column key hits.
    try testing.expect(ix.lookup_exact(enc_full.items) != null);
    try testing.expectEqual(@as(u64, 1), ix.lookup_exact(enc_full.items).?);
    // Its proper prefix must not, even though seek lands on the same entry.
    // Mutation check: swapping the eql for cmp_prefix == .eq makes this line
    // return "e1".
    try testing.expect(ix.lookup_exact(enc_prefix.items) == null);
    // And an absent key misses.
    var enc_other: std.ArrayListUnmanaged(u8) = .empty;
    defer enc_other.deinit(gpa);
    try bson.encode_key(.{ .int32 = 9 }, gpa, &enc_other);
    try testing.expect(ix.lookup_exact(enc_other.items) == null);
}

test "range bounds inclusive and exclusive" {
    const gpa = testing.allocator;
    var ix = try simple_index(gpa, test_pager(), &.{"a"}, false, false);
    defer ix.deinit(gpa);
    const docs = [_]struct { off: u64, a: i32 }{
        .{ .off = 1, .a = 1 },
        .{ .off = 2, .a = 2 },
        .{ .off = 3, .a = 3 },
        .{ .off = 4, .a = 4 },
        .{ .off = 5, .a = 5 },
    };
    for (docs) |s| {
        const d = try bytes_of(gpa, &.{ .{ .key = "_id", .value = .{ .int32 = s.a } }, .{ .key = "a", .value = .{ .int32 = s.a } } });
        defer gpa.free(d);
        _ = try ix.add_doc(gpa, d, s.off, true);
    }

    var out: std.ArrayListUnmanaged(u64) = .empty;
    defer out.deinit(gpa);
    // 2 <= a < 4 → {2, 3}
    try ix.lookup_range(gpa, &.{}, .{ .int32 = 2 }, true, .{ .int32 = 4 }, false, &out);
    try testing.expectEqual(@as(usize, 2), out.items.len);
    try testing.expect(out.items[0] == 2 or out.items[0] == 3);
    out.clearRetainingCapacity();
    // 2 < a <= 3 → {3}
    try ix.lookup_range(gpa, &.{}, .{ .int32 = 2 }, false, .{ .int32 = 3 }, true, &out);
    try testing.expectEqual(@as(usize, 1), out.items.len);
    try testing.expectEqual(@as(u64, 3), out.items[0]);
    out.clearRetainingCapacity();
    // a > 4 → {5}
    try ix.lookup_range(gpa, &.{}, .{ .int32 = 4 }, false, null, false, &out);
    try testing.expectEqual(@as(usize, 1), out.items.len);
    try testing.expectEqual(@as(u64, 5), out.items[0]);
}

test "empty index and remove_off" {
    const gpa = testing.allocator;
    var ix = try simple_index(gpa, test_pager(), &.{"a"}, false, false);
    defer ix.deinit(gpa);
    var out: std.ArrayListUnmanaged(u64) = .empty;
    defer out.deinit(gpa);
    try ix.lookup_eq(gpa, &.{.{ .int32 = 1 }}, &out);
    try testing.expectEqual(@as(usize, 0), out.items.len);

    const d = try bytes_of(gpa, &.{ .{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "a", .value = .{ .int32 = 1 } } });
    defer gpa.free(d);
    _ = try ix.add_doc(gpa, d, 1, true);
    ix.remove_off(1);
    try testing.expectEqual(@as(usize, 0), ix.count());
    try ix.lookup_eq(gpa, &.{.{ .int32 = 1 }}, &out);
    try testing.expectEqual(@as(usize, 0), out.items.len);
}

test "compound index prefix search and range on the next key" {
    const gpa = testing.allocator;
    var ix = try simple_index(gpa, test_pager(), &.{ "a", "b" }, false, false);
    defer ix.deinit(gpa);

    const id1: u64 = 1;
    const id2: u64 = 2;
    const id3: u64 = 3;
    const specs = [_]struct { off: u64, a: i32, b: i32 }{
        .{ .off = id1, .a = 1, .b = 2 },
        .{ .off = id2, .a = 1, .b = 3 },
        .{ .off = id3, .a = 2, .b = 1 },
    };
    for (specs) |s| {
        const d = try bytes_of(gpa, &.{
            .{ .key = "_id", .value = .{ .int32 = s.a } },
            .{ .key = "a", .value = .{ .int32 = s.a } },
            .{ .key = "b", .value = .{ .int32 = s.b } },
        });
        defer gpa.free(d);
        _ = try ix.add_doc(gpa, d, s.off, true);
    }

    // Prefix on a only.
    try expect_offs(gpa, &ix, &.{.{ .int32 = 1 }}, &.{ 1, 2 });
    // Full key.
    try expect_offs(gpa, &ix, &.{ .{ .int32 = 1 }, .{ .int32 = 3 } }, &.{2});
    try expect_offs(gpa, &ix, &.{ .{ .int32 = 2 }, .{ .int32 = 1 } }, &.{3});
    // Range on the next key: a == 1 and 2 <= b <= 3.
    var out: std.ArrayListUnmanaged(u64) = .empty;
    defer out.deinit(gpa);
    try ix.lookup_range(gpa, &.{.{ .int32 = 1 }}, .{ .int32 = 2 }, true, .{ .int32 = 3 }, true, &out);
    try testing.expectEqual(@as(usize, 2), out.items.len);
    try testing.expect(out.items[0] == 1 or out.items[0] == 2);
}

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, test_pager(), &.{ "a", "b" }, false, sparse);
        defer by_doc.deinit(gpa);
        var by_scan = try simple_index(gpa, test_pager(), &.{ "a", "b" }, false, sparse);
        defer by_scan.deinit(gpa);

        const n = 120;
        var ids: std.ArrayListUnmanaged(u64) = .empty;
        defer ids.deinit(gpa);
        var arrays: [n][3]bson.Value = undefined;
        var pairs: [n][2]bson.Pair = undefined;
        var docs: [n][]u8 = undefined;

        for (0..n) |i| {
            const id: u64 = @intCast(i + 1);
            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] = try bytes_of(gpa, 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(gpa, false);
        _ = try by_scan.finish_bulk(gpa, false);
        try testing.expectEqual(by_scan.count(), by_doc.count());

        // Remove in a shuffled order, so removals interleave rather than
        // peeling the leaves from one end.
        var order: [n]usize = undefined;
        for (0..n) |i| order[i] = i;
        rand.shuffle(usize, &order);

        var doc_refs: std.ArrayListUnmanaged(EntryRef) = .empty;
        var scan_refs: std.ArrayListUnmanaged(EntryRef) = .empty;
        defer doc_refs.deinit(gpa);
        defer scan_refs.deinit(gpa);
        for (order) |i| {
            by_doc.remove_doc(gpa, docs[i], ids.items[i]);
            by_scan.remove_off(ids.items[i]);

            doc_refs.clearRetainingCapacity();
            scan_refs.clearRetainingCapacity();
            try refs_of(gpa, &by_doc, &doc_refs);
            try refs_of(gpa, &by_scan, &scan_refs);

            testing.expectEqual(scan_refs.items.len, doc_refs.items.len) catch |err| {
                std.debug.print("sparse={} after removing doc {d}: scan left {d}, remove_doc left {d}\n", .{
                    sparse, i, scan_refs.items.len, doc_refs.items.len,
                });
                return err;
            };
            for (doc_refs.items, scan_refs.items) |x, y| {
                try testing.expect(std.mem.eql(u8, x.key, y.key));
                try testing.expectEqual(x.off, y.off);
            }
        }
        for (docs) |b| gpa.free(b);
        try testing.expectEqual(@as(usize, 0), by_doc.count());
    }
}

test "incremental inserts and removals stay identical to a brute-force model" {
    // The bulk-path differential covers packing; this one covers the
    // incremental path: one insert at a time (leaf splits), interleaved
    // with removals (empty-leaf unlinks, one-child internal nodes), all
    // checked against a reference model at every step.
    const gpa = testing.allocator;
    var prng = std.Random.DefaultPrng.init(0x0dd_ba11);
    const rand = prng.random();

    var ix = try simple_index(gpa, test_pager(), &.{ "a", "b" }, false, false);
    defer ix.deinit(gpa);

    var model: std.ArrayListUnmanaged(ModelFact) = .empty;
    defer model.deinit(gpa);
    var live: std.ArrayListUnmanaged(bool) = .empty;
    defer live.deinit(gpa);

    const n = 600;
    for (0..n) |i| {
        // Insert.
        const a = rand.intRangeAtMost(i32, 0, 30);
        const b = rand.intRangeAtMost(i32, 0, 30);
        const id: u64 = @intCast(i + 1);
        const d = try bytes_of(gpa, &.{
            .{ .key = "_id", .value = .{ .int32 = @intCast(i) } },
            .{ .key = "a", .value = .{ .int32 = a } },
            .{ .key = "b", .value = .{ .int32 = b } },
        });
        defer gpa.free(d);
        _ = try ix.add_doc(gpa, d, id, false);
        try model.append(gpa, .{ .a = a, .b = b, .off = id });
        try live.append(gpa, true);

        // Verify equality and one-sided range against the model.
        try verify_model(gpa, &ix, model.items, live.items, rand);
    }
    // Remove a random half in random order.
    var order: std.ArrayListUnmanaged(usize) = .empty;
    defer order.deinit(gpa);
    for (0..n) |i| if (rand.boolean()) try order.append(gpa, i);
    rand.shuffle(usize, order.items);
    for (order.items) |i| {
        const m = model.items[i];
        const d = try bytes_of(gpa, &.{
            .{ .key = "_id", .value = .{ .int32 = @intCast(i) } },
            .{ .key = "a", .value = .{ .int32 = m.a } },
            .{ .key = "b", .value = .{ .int32 = m.b } },
        });
        defer gpa.free(d);
        ix.remove_doc(gpa, d, m.off);
        live.items[i] = false;
        try verify_model(gpa, &ix, model.items, live.items, rand);
    }
}

/// One document's facts in the incremental-mutation differential.
const ModelFact = struct { a: i32, b: i32, off: u64 };

fn verify_model(
    gpa: std.mem.Allocator,
    ix: *const Index,
    model: []const ModelFact,
    live: []const bool,
    rand: std.Random,
) !void {
    // lookup_eq over a random value.
    const a = rand.intRangeAtMost(i32, 0, 30);
    var out: std.ArrayListUnmanaged(u64) = .empty;
    defer out.deinit(gpa);
    try ix.lookup_eq(gpa, &.{.{ .int32 = a }}, &out);
    var expected: usize = 0;
    for (model, 0..) |m, mi| {
        if (live[mi] and m.a == a) expected += 1;
    }
    try testing.expectEqual(expected, out.items.len);

    // lookup_range over a random one-sided range.
    out.clearRetainingCapacity();
    const lo_v = rand.intRangeAtMost(i32, -5, 35);
    const lo_incl = rand.boolean();
    try ix.lookup_range(gpa, &.{.{ .int32 = a }}, .{ .int32 = lo_v }, lo_incl, null, false, &out);
    expected = 0;
    for (model, 0..) |m, mi| {
        if (!live[mi]) continue;
        if (m.a != a) continue;
        if (m.b < lo_v) continue;
        if (m.b == lo_v and !lo_incl) continue;
        expected += 1;
    }
    try testing.expectEqual(expected, out.items.len);
}

test "lookup_range matches a brute-force filter over random data" {
    const gpa = testing.allocator;

    // lookup_range seeks both ends of the range inside the tree instead of
    // scanning the whole equality band. Bounds like that are easy to get
    // subtly wrong at the inclusive/exclusive edges and where the band
    // ends, so check the whole result set against the definition rather
    // than spot-checking.
    var prng = std.Random.DefaultPrng.init(0x5eed_1234);
    const rand = prng.random();

    var ids: std.ArrayListUnmanaged(u64) = .empty;
    defer ids.deinit(gpa);

    var ix = try simple_index(gpa, test_pager(), &.{ "a", "b" }, false, false);
    defer ix.deinit(gpa);

    // Deliberately few distinct values so equal keys, and therefore the
    // boundaries between them, come up constantly.
    const n = 400;
    var facts: [n]struct { a: i32, b: i32 } = undefined;
    for (0..n) |i| {
        const a = rand.intRangeAtMost(i32, 0, 4);
        const b = rand.intRangeAtMost(i32, 0, 9);
        facts[i] = .{ .a = a, .b = b };
        const id: u64 = @intCast(i + 1);
        try ids.append(gpa, id);
        const d = try bytes_of(gpa, &.{
            .{ .key = "_id", .value = .{ .int32 = @intCast(i) } },
            .{ .key = "a", .value = .{ .int32 = a } },
            .{ .key = "b", .value = .{ .int32 = b } },
        });
        defer gpa.free(d);
        try ix.append_doc_entries(gpa, d, id);
    }
    _ = try ix.finish_bulk(gpa, false);

    var out: std.ArrayListUnmanaged(u64) = .empty;
    defer out.deinit(gpa);

    for (0..600) |case| {
        const a = rand.intRangeAtMost(i32, 0, 4);
        const lo_v = rand.intRangeAtMost(i32, -1, 10);
        const hi_v = rand.intRangeAtMost(i32, -1, 10);
        const lo_incl = rand.boolean();
        const hi_incl = rand.boolean();
        const use_lo = rand.boolean();
        const use_hi = rand.boolean();

        out.clearRetainingCapacity();
        try ix.lookup_range(
            gpa,
            &.{.{ .int32 = a }},
            if (use_lo) .{ .int32 = lo_v } else null,
            lo_incl,
            if (use_hi) .{ .int32 = hi_v } else null,
            hi_incl,
            &out,
        );

        var expected: usize = 0;
        for (facts, 0..) |f, i| {
            _ = i;
            if (f.a != a) continue;
            if (use_lo) {
                if (f.b < lo_v) continue;
                if (f.b == lo_v and !lo_incl) continue;
            }
            if (use_hi) {
                if (f.b > hi_v) continue;
                if (f.b == hi_v and !hi_incl) continue;
            }
            expected += 1;
        }
        testing.expectEqual(expected, out.items.len) catch |err| {
            std.debug.print(
                "case {d}: a={d} lo={?d} incl={} hi={?d} incl={}\n",
                .{ case, a, if (use_lo) lo_v else null, lo_incl, if (use_hi) hi_v else null, hi_incl },
            );
            return err;
        };
    }
}

test "TTL spec round-trips through write_spec and compares in spec_equal" {
    const gpa = testing.allocator;
    const spec = doc_of(&.{
        .{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } },
        .{ .key = "name", .value = .{ .string = "expireAt_1" } },
        // A driver sends a plain JS number as a double.
        .{ .key = "expireAfterSeconds", .value = .{ .double = 60.0 } },
    });
    var ix = try parse_spec(gpa, test_pager(), &spec);
    defer ix.deinit(gpa);
    try testing.expectEqual(@as(?i64, 60), ix.ttl);

    // Serialize and reparse: the log/compaction path.
    var bytes: std.ArrayListUnmanaged(u8) = .empty;
    defer bytes.deinit(gpa);
    try ix.write_spec(gpa, &bytes);
    var reparsed_doc = try bson.Document.parse(gpa, bytes.items);
    defer reparsed_doc.deinit();
    try testing.expectEqual(@as(i32, 60), reparsed_doc.get("expireAfterSeconds").?.int32);
    var ix2 = try parse_spec(gpa, test_pager(), &reparsed_doc);
    defer ix2.deinit(gpa);
    try testing.expect(Index.spec_equal(&ix, &ix2));

    // Same name and key, different expiry: not the same spec (the command
    // layer turns this into IndexOptionsConflict).
    const other = doc_of(&.{
        .{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } },
        .{ .key = "name", .value = .{ .string = "expireAt_1" } },
        .{ .key = "expireAfterSeconds", .value = .{ .int32 = 90 } },
    });
    var ix3 = try parse_spec(gpa, test_pager(), &other);
    defer ix3.deinit(gpa);
    try testing.expect(!Index.spec_equal(&ix, &ix3));

    // The largest legal expiry, arriving as an int64, still round-trips as
    // an int32 — the only type this ever emits.
    const big = doc_of(&.{
        .{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } },
        .{ .key = "expireAfterSeconds", .value = .{ .int64 = max_expire_after_seconds } },
    });
    var ix_big = try parse_spec(gpa, test_pager(), &big);
    defer ix_big.deinit(gpa);
    bytes.clearRetainingCapacity();
    try ix_big.write_spec(gpa, &bytes);
    var big_doc = try bson.Document.parse(gpa, bytes.items);
    defer big_doc.deinit();
    try testing.expectEqual(@as(i32, 2147483647), big_doc.get("expireAfterSeconds").?.int32);
    var ix_big2 = try parse_spec(gpa, test_pager(), &big_doc);
    defer ix_big2.deinit(gpa);
    try testing.expectEqual(@as(?i64, max_expire_after_seconds), ix_big2.ttl);

    // No option at all: ttl null, and nothing emitted (old logs reparse
    // unchanged).
    const plain = doc_of(&.{
        .{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } },
    });
    var ix4 = try parse_spec(gpa, test_pager(), &plain);
    defer ix4.deinit(gpa);
    try testing.expect(ix4.ttl == null);
    bytes.clearRetainingCapacity();
    try ix4.write_spec(gpa, &bytes);
    var plain_doc = try bson.Document.parse(gpa, bytes.items);
    defer plain_doc.deinit();
    try testing.expect(plain_doc.get("expireAfterSeconds") == null);
}

test "TTL spec rejects compound keys and bad expireAfterSeconds" {
    const gpa = testing.allocator;
    const compound = doc_of(&.{
        .{ .key = "key", .value = .{ .doc = &.{
            .{ .key = "a", .value = .{ .int32 = 1 } },
            .{ .key = "b", .value = .{ .int32 = 1 } },
        } } },
        .{ .key = "expireAfterSeconds", .value = .{ .int32 = 60 } },
    });
    try testing.expectError(error.TtlOnCompoundIndex, parse_spec(gpa, test_pager(), &compound));

    const bad = [_]bson.Value{
        .{ .int32 = -1 },
        .{ .int64 = -1 },
        .{ .double = -0.5 },
        .{ .double = 1.5 },
        // Past MongoDB's bound, as an int and as a double (1e19 would also
        // overflow the conversion, which the range check runs before).
        .{ .int64 = max_expire_after_seconds + 1 },
        .{ .double = 3e9 },
        .{ .double = 1e19 },
        .{ .double = std.math.inf(f64) },
        .{ .double = std.math.nan(f64) },
        .{ .string = "60" },
        .{ .bool = true },
        .null,
    };
    for (bad) |v| {
        const spec = doc_of(&.{
            .{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } },
            .{ .key = "expireAfterSeconds", .value = v },
        });
        try testing.expectError(error.InvalidExpireAfterSeconds, parse_spec(gpa, test_pager(), &spec));
    }

    // 0 is legal: expire at exactly the stored instant.
    const zero = doc_of(&.{
        .{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } },
        .{ .key = "expireAfterSeconds", .value = .{ .int32 = 0 } },
    });
    var ix = try parse_spec(gpa, test_pager(), &zero);
    defer ix.deinit(gpa);
    try testing.expectEqual(@as(?i64, 0), ix.ttl);
    // The default name still comes from the key pattern.
    try testing.expectEqualStrings("expireAt_1", ix.name);
}

fn str_doc(gpa: std.mem.Allocator, i: usize, fill: u8, len: usize) ![]u8 {
    const s = try gpa.alloc(u8, len);
    defer gpa.free(s);
    @memset(s, fill);
    return bytes_of(gpa, &.{
        .{ .key = "_id", .value = .{ .int32 = @intCast(i) } },
        .{ .key = "s", .value = .{ .string = s } },
    });
}

test "a churned leaf of large keys splits without promoting from an empty half" {
    // remove_record leaves the freed bytes dead at the top of the page, so
    // a leaf holding one or two ~1 KiB keys runs out of room after a few
    // insert/remove cycles while still holding almost nothing. Splitting
    // then would hand the right half zero records and promote whatever the
    // uninitialised slot 0 happened to hold.
    const gpa = testing.allocator;
    var ix = try simple_index(gpa, test_pager(), &.{"s"}, false, false);
    defer ix.deinit(gpa);

    const n = 12;
    var docs: [n][]u8 = undefined;
    var ids: [n]u64 = undefined;
    var made: usize = 0;
    defer for (0..made) |i| {
        gpa.free(docs[i]);
    };
    for (0..n) |i| {
        docs[i] = try str_doc(gpa, i, @intCast('a' + i), 1000);
        ids[i] = @intCast(i + 1);
        made += 1;
    }

    // Keep one entry live while the page fills with dead bytes, then leave
    // two live: a bad split promotes an empty page's slot 0 as the
    // separator, and every later descent then misses the left leaf. The
    // leaf chain would still hold both, so this has to be checked through a
    // lookup, which descends.
    for (0..n - 1) |i| {
        _ = try ix.add_doc(gpa, docs[i], ids[i], false);
        if (i >= 1) ix.remove_doc(gpa, docs[i - 1], ids[i - 1]);
    }
    _ = try ix.add_doc(gpa, docs[n - 1], ids[n - 1], false);

    try testing.expectEqual(@as(usize, 2), ix.count());
    for ([_]usize{ n - 2, n - 1 }) |i| {
        var s: [1000]u8 = undefined;
        @memset(&s, @intCast('a' + i));
        try expect_offs(gpa, &ix, &.{.{ .string = &s }}, &.{ids[i]});
    }
}

test "a split with lopsided record sizes keeps the new record inside its page" {
    // The halves are split by slot count, so all the large records can end
    // up on one side; the record that caused the split is then stored into
    // that half with no room left for it.
    const gpa = testing.allocator;
    var ix = try simple_index(gpa, test_pager(), &.{"s"}, false, false);
    defer ix.deinit(gpa);

    var docs: std.ArrayListUnmanaged([]u8) = .empty;
    var ids: std.ArrayListUnmanaged(u64) = .empty;
    defer {
        for (docs.items) |d| gpa.free(d);
        docs.deinit(gpa);
        ids.deinit(gpa);
    }

    var i: usize = 0;
    // Three ~1 KiB keys, all sorting before the short ones ('a' < 'z') ...
    while (i < 3) : (i += 1) {
        try docs.append(gpa, try str_doc(gpa, i, 'a', 1000 - i));
        try ids.append(gpa, @intCast(i + 1));
    }
    // ... then short keys, filling the page by slot count ...
    while (i < 28) : (i += 1) {
        try docs.append(gpa, try str_doc(gpa, i, 'z', 4));
        try ids.append(gpa, @intCast(i + 1));
    }
    // ... then one more large key, which sorts into the left half.
    try docs.append(gpa, try str_doc(gpa, i, 'a', 996));
    try ids.append(gpa, @intCast(i + 1));

    for (docs.items, ids.items) |d, id| _ = try ix.add_doc(gpa, d, id, false);

    try testing.expectEqual(docs.items.len, ix.count());
    // Every key comes back, in order, exactly once.
    var prev: []const u8 = "";
    var seen: usize = 0;
    var it = ix.iter();
    while (it.next()) |e| : (seen += 1) {
        try testing.expect(std.mem.order(u8, prev, e.key) != .gt);
        prev = e.key;
    }
    try testing.expectEqual(docs.items.len, seen);
}

test "the _id index plan covers equality, ranges and _id sort order" {
    // The implicit _id_ index is a normal Index (keys = [_id: 1]) passed to
    // plan separately from the secondaries. Its encoded keys are canonical,
    // so compare-equal numbers (int32/int64/double) find the same entries,
    // and a full scan of it is the sort planner's order supply for
    // sort({_id: ...}).
    const gpa = testing.allocator;
    var ix = try simple_index(gpa, test_pager(), &.{"_id"}, false, false);
    defer ix.deinit(gpa);
    for (0..5) |i| {
        const d = try bytes_of(gpa, &.{
            .{ .key = "_id", .value = .{ .int32 = @intCast(i + 1) } },
            .{ .key = "v", .value = .{ .int32 = @intCast(i) } },
        });
        defer gpa.free(d);
        const id: u64 = @intCast(i + 1);
        _ = try ix.add_doc(gpa, d, id, false);
    }

    // {_id: 3} → an equality plan whose candidates are just that doc.
    {
        const f = [_]bson.Pair{.{ .key = "_id", .value = .{ .int32 = 3 } }};
        var p = (try plan(gpa, &ix, &.{}, &f, &.{})).?;
        defer p.deinit(gpa);
        try testing.expectEqual(@as(usize, 1), p.key_len());
        var ids: std.ArrayListUnmanaged(u64) = .empty;
        defer ids.deinit(gpa);
        try p.search(gpa, &ids);
        try testing.expectEqual(@as(usize, 1), ids.items.len);
        try testing.expectEqual(@as(u64, 3), ids.items[0]);
    }
    // The same query as an int64 and a double finds the int32 entry.
    {
        const f = [_]bson.Pair{.{ .key = "_id", .value = .{ .int64 = 3 } }};
        var p = (try plan(gpa, &ix, &.{}, &f, &.{})).?;
        defer p.deinit(gpa);
        var ids: std.ArrayListUnmanaged(u64) = .empty;
        defer ids.deinit(gpa);
        try p.search(gpa, &ids);
        try testing.expectEqual(@as(usize, 1), ids.items.len);
        try testing.expectEqual(@as(u64, 3), ids.items[0]);
    }
    // A range on _id yields the band in index order.
    {
        const f = [_]bson.Pair{.{ .key = "_id", .value = .{ .doc = &.{.{ .key = "$gte", .value = .{ .int32 = 3 } }} } }};
        var p = (try plan(gpa, &ix, &.{}, &f, &.{})).?;
        defer p.deinit(gpa);
        var ids: std.ArrayListUnmanaged(u64) = .empty;
        defer ids.deinit(gpa);
        try p.search(gpa, &ids);
        try testing.expectEqual(@as(usize, 3), ids.items.len);
        try testing.expectEqual(@as(u64, 3), ids.items[0]);
        try testing.expectEqual(@as(u64, 5), ids.items[2]);
    }
    // sort({_id: 1}) with no filter: a full index scan that supplies the
    // order — the plan the sort planner needs to stop materializing.
    {
        const sort = [_]query.SortKey{.{ .path = "_id", .descending = false }};
        var p = (try plan(gpa, &ix, &.{}, &.{}, &sort)).?;
        defer p.deinit(gpa);
        try testing.expect(p.provides_sort and !p.backward);
        var ids: std.ArrayListUnmanaged(u64) = .empty;
        defer ids.deinit(gpa);
        try p.search(gpa, &ids);
        try testing.expectEqual(@as(usize, 5), ids.items.len);
        try testing.expectEqual(@as(u64, 1), ids.items[0]);
        try testing.expectEqual(@as(u64, 5), ids.items[4]);
    }
    // sort({_id: -1}): backward.
    {
        const sort = [_]query.SortKey{.{ .path = "_id", .descending = true }};
        var p = (try plan(gpa, &ix, &.{}, &.{}, &sort)).?;
        defer p.deinit(gpa);
        try testing.expect(p.provides_sort and p.backward);
        var ids: std.ArrayListUnmanaged(u64) = .empty;
        defer ids.deinit(gpa);
        try p.search(gpa, &ids);
        try testing.expectEqual(@as(u64, 5), ids.items[0]);
        try testing.expectEqual(@as(u64, 1), ids.items[4]);
    }
    // A filter naming no _id field leaves the _id index unusable.
    {
        const f = [_]bson.Pair{.{ .key = "v", .value = .{ .int32 = 1 } }};
        try testing.expect((try plan(gpa, &ix, &.{}, &f, &.{})) == null);
    }
}

test "planner picks eq run, ranges, and bails on sparse null" {
    const gpa = testing.allocator;
    var ix = try simple_index(gpa, test_pager(), &.{ "a", "b" }, false, false);
    defer ix.deinit(gpa);
    var sp = try simple_index(gpa, test_pager(), &.{ "a", "b" }, false, true);
    defer sp.deinit(gpa);

    // {a: 1, b: 2} → full-key equality.
    {
        const f = [_]bson.Pair{
            .{ .key = "a", .value = .{ .int32 = 1 } },
            .{ .key = "b", .value = .{ .int32 = 2 } },
        };
        var p = (try plan(gpa, null, &.{&ix}, &f, &.{})).?;
        defer p.deinit(gpa);
        try testing.expectEqual(@as(usize, 2), p.key_len());
        try testing.expect(p.lo == null and p.hi == null);
    }
    // {a: 1, b: {$gt: 2}} → equality run of 1 + range on the next key.
    {
        const f = [_]bson.Pair{
            .{ .key = "a", .value = .{ .int32 = 1 } },
            .{ .key = "b", .value = .{ .doc = &.{.{ .key = "$gt", .value = .{ .int32 = 2 } }} } },
        };
        var p = (try plan(gpa, null, &.{&ix}, &f, &.{})).?;
        defer p.deinit(gpa);
        try testing.expectEqual(@as(usize, 1), p.key_len());
        try testing.expect(p.hi == null and p.lo != null and !p.lo_incl);
    }
    // {a: 1} only → prefix run of 1.
    {
        const f = [_]bson.Pair{.{ .key = "a", .value = .{ .int32 = 1 } }};
        var p = (try plan(gpa, null, &.{&ix}, &f, &.{})).?;
        defer p.deinit(gpa);
        try testing.expectEqual(@as(usize, 1), p.key_len());
    }
    // Pure range on the first key → key_len 0 with a bound.
    {
        const f = [_]bson.Pair{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$gte", .value = .{ .int32 = 1 } }} } }};
        var p = (try plan(gpa, null, &.{&ix}, &f, &.{})).?;
        defer p.deinit(gpa);
        try testing.expectEqual(@as(usize, 0), p.key_len());
        try testing.expect(p.lo != null and p.lo_incl);
    }
    // Unusable filter → no plan.
    {
        const f = [_]bson.Pair{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$regex", .value = .{ .string = "^x" } }} } }};
        try testing.expect((try plan(gpa, null, &.{&ix}, &f, &.{})) == null);
        const or_f = [_]bson.Pair{.{ .key = "$or", .value = .{ .array = &.{
            .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} },
        } } }};
        try testing.expect((try plan(gpa, null, &.{&ix}, &or_f, &.{})) == null);
    }
    // Sparse index bails on a null component.
    {
        const f = [_]bson.Pair{.{ .key = "a", .value = .null }};
        try testing.expect((try plan(gpa, null, &.{&sp}, &f, &.{})) == null);
        // Non-sparse is fine with null.
        var p = (try plan(gpa, null, &.{&ix}, &f, &.{})).?;
        defer p.deinit(gpa);
        try testing.expect(p.key_len() == 1);
        // A null inside $in bails too.
        const fin = [_]bson.Pair{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$in", .value = .{ .array = &.{ .{ .int32 = 1 }, .null } } }} } }};
        try testing.expect((try plan(gpa, null, &.{&sp}, &fin, &.{})) == null);
    }
    // $in cartesian product is capped.
    {
        var members: [20]bson.Value = undefined;
        for (0..20) |j| members[j] = .{ .int32 = @intCast(j) };
        const f = [_]bson.Pair{
            .{ .key = "a", .value = .{ .doc = &.{.{ .key = "$in", .value = .{ .array = &members } }} } },
            .{ .key = "b", .value = .{ .doc = &.{.{ .key = "$in", .value = .{ .array = &members } }} } },
        };
        // 20 * 20 = 400 > 100 → fall back to a scan.
        try testing.expect((try plan(gpa, null, &.{&ix}, &f, &.{})) == null);
    }
}