dev/MultiforaDB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

index: the node arena and overflow slab live in the data file

Browse Source 
The last structures move onto the pager, so the whole engine's storage is now
one mapped file plus the WAL.

Node ids are deliberately *not* page numbers. 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 as well -- COWing one leaf cascades through the leaf level, one internal
node through its whole subtree. An in-RAM id->page table makes the table slot
the single owner of a page number, so COW 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* this milestone removes.

The overflow slab becomes extents too, so `Slot.off` for a spilled record is an
absolute file offset -- the same change documents went through.

--

Two bugs, both found by measuring rather than by reading, and both worth
recording because the second one would have been invisible until the churn gate.

The reservation was a tail mark, and it cannot be: an upsert reserves tree pages
for every index *and* slab room for the document, all before one log append. The
second reserver overwrote the first one's promise and the first one's allocation
then asserted. Caught on a 512 MB load by the tripwire added in the
`reserve_for` commit, which is the entire reason that assert exists. It is a
count now, and the multi-consumer ordering is pinned by a test.

And a reservation was never released. It is scoped to one write -- taken before
the log append so the publish cannot fail -- but a tree reservation covers the
worst case of several splits while a typical insert causes none, so the promise
accumulated by a handful of pages per write and dragged the file up with it. The
data file was **1.89 GB for 512 MB of documents**; releasing the unclaimed
promise at the end of each write brings it to 551 MB, or 1.08x, which is the
extent slack and the node pages.

--

Measured on one harness, 512 MB / 16 KB docs, against the in-RAM engine this
replaces:

  bulk insert throughput      742.6 MB/s -> 736.4 MB/s
  insertOne (sequential)      0.20 ms    -> 0.22 ms
  createIndex({k: 1})         26.8 ms    -> 16.5 ms
  countDocuments({})          2.1 ms     -> 1.1 ms
  findOne({k: 500}) indexed   0.75 ms    -> 0.56 ms
  find({p: range}).count()    6.6 ms     -> 4.6 ms
  aggregate $group by k       5.8 ms     -> 3.8 ms
  updateMany({k: 7}, {$inc})  1.2 ms     -> 1.0 ms

Reads gain from one contiguous mapping; the two write rows are within noise of
flat. RSS is still unchanged and still cannot improve, for the reason given in
the previous commit: every open replays the whole log and rebuilds everything.

The dev harnesses each open their own data file now. `zig build fuzz` caught all
four of them, again.
This commit is contained in:
Alexey Shakhmatov
2026-08-03 20:55:48 +03:00
parent 9dda943f26
commit 2e7f72074f
7 changed files with 293 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/db.zig
View File

@@ -86,7 +86,7 @@ pub const Collection = struct {
// where serialize_value is not, so int32 1 / int64 1 / double 1.0
// collide as they do in MongoDB -- see the migration note in
// apply_record.
self.id_index = try index.Index.init(gpa, "_id_", &keys, true, false, null);
self.id_index = try index.Index.init(gpa, pager, "_id_", &keys, true, false, null);
return self;
}
@@ -687,6 +687,9 @@ pub const Engine = struct {
b.ix.insert_entries(&b.built, off);
}
stored = true;
// The write is published; anything the reservations above did not claim
// is dead. Leaving it promised would grow the file on every write.
self.pager.release_reservation();
self.note_compact();
}
@@ -771,7 +774,7 @@ pub const Engine = struct {
spec_doc: *const bson.Document,
) !*index.Index {
const coll = try self.get_or_create_collection(db_name, coll_name);
const parsed = try index.parse_spec(self.gpa, spec_doc);
const parsed = try index.parse_spec(self.gpa, self.pager, spec_doc);
// Boxed before anything is built into it, so publishing is a pointer
// append rather than a struct copy. An Index will own a mapping once
// the arena is file-backed, and copying one then would duplicate that
@@ -806,6 +809,7 @@ pub const Engine = struct {
try ix.append_doc_entries(self.gpa, coll.doc_bytes(entry.value_ptr.*), entry.value_ptr.*);
}
_ = try ix.finish_bulk(self.gpa, true);
self.pager.release_reservation();
// Reserve the collection slot, then persist and publish.
try coll.indexes.ensureUnusedCapacity(self.gpa, 1);
@@ -1237,6 +1241,7 @@ pub const Engine = struct {
};
}
// Tolerated, not enforced: the database must always open.
defer self.pager.release_reservation();
if (try ix.finish_bulk(self.gpa, false)) {
std.debug.print(
"multiforadb: WARNING: unique index '{s}' has duplicate keys in existing " ++
@@ -1255,7 +1260,7 @@ pub const Engine = struct {
coll: *Collection,
spec_doc: *const bson.Document,
) !void {
const parsed = try index.parse_spec(self.gpa, spec_doc);
const parsed = try index.parse_spec(self.gpa, self.pager, spec_doc);
const ix = self.gpa.create(index.Index) catch |err| {
var dead = parsed;
dead.deinit(self.gpa);
@@ -1337,6 +1342,7 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an
defer self.gpa.free(doc_bytes);
try coll.slab_reserve(self.gpa, doc_bytes.len);
const off = coll.slab_append(doc_bytes);
self.pager.release_reservation();
try coll.docs.put(self.gpa, id_key, off);
self.live_docs += 1;
key_owned = true;

24
src/fuzz_split.zig
View File

@@ -12,6 +12,7 @@
const std = @import("std");
const bson = @import("bson.zig");
const index = @import("index.zig");
const pgr = @import("pager.zig");
const testing = std.testing;
@@ -32,12 +33,33 @@ const Doc = struct {
live: bool,
};
/// A throwaway data file for this harness. The B+tree's pages live in the data
/// file now, so a standalone harness has to provide one.
/// The Threaded is intentionally leaked: it must outlive the pager's io, and
/// these harnesses are one-shot processes. fuzz_split runs under the testing
/// allocator, which checks for leaks, so it uses the page allocator here.
fn harness_pager(gpa: std.mem.Allocator, name: []const u8) !*pgr.Pager {
const threaded = try std.heap.page_allocator.create(std.Io.Threaded);
threaded.* = .init_single_threaded;
const io = threaded.io();
const path = try std.fmt.allocPrint(std.heap.page_allocator, ".zig-cache/{s}.data", .{name});
std.Io.Dir.cwd().deleteFile(io, path) catch {};
const pg = try gpa.create(pgr.Pager);
pg.* = try pgr.Pager.open(gpa, io, path, .{});
return pg;
}
fn run(seed: u64, ops: usize, max_len: usize) !void {
const gpa = testing.allocator;
var prng = std.Random.DefaultPrng.init(seed);
const rand = prng.random();
var ix = try index.Index.init(gpa, "s_1", &.{.{ .path = "s", .descending = false }}, false, false, null);
const pg = try harness_pager(gpa, "fuzz_split");
defer {
pg.deinit();
gpa.destroy(pg);
}
var ix = try index.Index.init(gpa, pg, "s_1", &.{.{ .path = "s", .descending = false }}, false, false, null);
defer ix.deinit(gpa);
var docs: std.ArrayListUnmanaged(Doc) = .empty;

213
src/index.zig
View File

@@ -42,6 +42,7 @@
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
@@ -97,6 +98,10 @@ pub const BuiltEntries = struct {
// 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;
@@ -190,10 +195,30 @@ pub const Index = struct {
multikey: bool,
// -- the tree ----------------------------------------------------------
nodes: std.ArrayListUnmanaged(Node),
/// Append-only slab of spilled records; referenced by slots, never
/// rewritten or freed (offsets stay valid forever).
overflow: std.ArrayListUnmanaged(u8),
/// 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.
@@ -213,6 +238,7 @@ pub const Index = struct {
pub fn init(
gpa: std.mem.Allocator,
pager: *pgr.Pager,
name: []const u8,
keys: []const IndexKey,
unique: bool,
@@ -226,8 +252,11 @@ pub const Index = struct {
.sparse = sparse,
.ttl = ttl,
.multikey = false,
.nodes = .empty,
.overflow = .empty,
.pager = pager,
.node_pages = .empty,
.ovf_extents = .empty,
.ovf_tail = 0,
.ovf_end = 0,
.staging = .empty,
.root = 0,
.first_leaf = 0,
@@ -249,10 +278,14 @@ pub const Index = struct {
owned_keys[n] = .{ .path = try gpa.dupe(u8, keys[n].path), .descending = keys[n].descending };
}
self.keys = owned_keys;
// nodes[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.nodes.append(gpa, empty_node(0));
try self.nodes.append(gpa, empty_node(1));
// 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;
@@ -262,8 +295,8 @@ pub const Index = struct {
pub fn deinit(self: *Index, gpa: std.mem.Allocator) void {
for (self.staging.items) |e| gpa.free(e.key);
self.staging.deinit(gpa);
self.nodes.deinit(gpa);
self.overflow.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);
@@ -398,7 +431,11 @@ pub const Index = struct {
// 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;
try self.nodes.ensureUnusedCapacity(gpa, @intCast(extra_nodes));
// 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);
}
@@ -415,7 +452,20 @@ pub const Index = struct {
const rec_len: u64 = e.key.len + off_len;
if (rec_len > inline_limit) overflow_bytes += rec_len;
}
try self.overflow.ensureUnusedCapacity(gpa, @intCast(overflow_bytes));
if (overflow_bytes == 0) return;
if (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.
@@ -496,11 +546,13 @@ pub const Index = struct {
/// 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,
) error{ DuplicateKeyIndex, OutOfMemory }!bool {
) !bool {
std.mem.sort(Staged, self.staging.items, {}, staged_less);
var duplicate = false;
if (self.unique and self.staging.items.len >= 2) {
@@ -869,17 +921,23 @@ pub const Index = struct {
/// document can never be live but unindexed -- so panicking is the only
/// honest response.
fn alloc_node(self: *Index) u32 {
assert_msg(self.nodes.items.len < self.nodes.capacity, "node allocation overran reserve_for's bound");
self.nodes.appendAssumeCapacity(empty_node(0));
return @intCast(self.nodes.items.len - 1);
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 {
try self.nodes.append(gpa, empty_node(0));
return @intCast(self.nodes.items.len - 1);
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 -------------------------------------------------------
@@ -910,19 +968,19 @@ pub const Index = struct {
/// The page holding node `id`, for reading.
inline fn page(self: *const Index, id: u32) *const Node {
return &self.nodes.items[id];
return @ptrCast(self.pager.page(self.node_pages.items[id]));
}
/// The page holding node `id`, for writing.
inline fn page_mut(self: *Index, id: u32) *Node {
return &self.nodes.items[id];
return @ptrCast(self.pager.page_mut(self.node_pages.items[id]));
}
/// 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.overflow.items[@intCast(from)..@intCast(to)];
return self.pager.bytes(from, @intCast(to - from));
}
fn get_slot(node_page: *const Node, i: u32) Slot {
@@ -989,13 +1047,13 @@ pub const Index = struct {
// 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.overflow.items.len + rec_len <= self.overflow.capacity, "spilled record overran reserve_overflow's bound");
s.off = self.overflow.items.len;
self.overflow.appendSliceAssumeCapacity(rec.key);
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| {
var buf: [off_len]u8 = undefined;
std.mem.writeInt(u64, &buf, o, .little);
self.overflow.appendSliceAssumeCapacity(&buf);
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 {
@@ -1623,7 +1681,9 @@ pub const SpecError = error{ InvalidIndexSpec, TtlOnCompoundIndex, InvalidExpire
/// Parse {key: {...}, name?, unique?, sparse?, expireAfterSeconds?} from a
/// spec document — the form drivers send and the form the log stores.
pub fn parse_spec(gpa: std.mem.Allocator, spec: *const bson.Document) SpecError!Index {
/// 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,
@@ -1655,13 +1715,13 @@ pub fn parse_spec(gpa: std.mem.Allocator, spec: *const bson.Document) SpecError!
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, nm, keys[0..key_pairs.len], unique, sparse, ttl);
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, name, keys[0..key_pairs.len], unique, sparse, ttl);
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
@@ -2202,15 +2262,40 @@ 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, "test", keys[0..paths.len], unique, sparse, null);
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.
@@ -2244,7 +2329,7 @@ fn ref_lt(a: EntryRef, b: EntryRef) bool {
test "entries sort across numeric types and string/null/objectid" {
const gpa = testing.allocator;
var ix = try simple_index(gpa, &.{"a"}, false, false);
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 } } });
@@ -2282,7 +2367,7 @@ test "entries sort across numeric types and string/null/objectid" {
test "missing field is indexed as null; sparse skips the document" {
const gpa = testing.allocator;
var ix = try simple_index(gpa, &.{"a"}, false, false);
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);
@@ -2290,7 +2375,7 @@ test "missing field is indexed as null; sparse skips the document" {
try testing.expectEqual(@as(usize, 1), ix.count());
try expect_offs(gpa, &ix, &.{.null}, &.{1});
var sp = try simple_index(gpa, &.{"a"}, false, true);
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());
@@ -2298,7 +2383,7 @@ test "missing field is indexed as null; sparse skips the document" {
test "multikey expansion indexes the array and its elements" {
const gpa = testing.allocator;
var ix = try simple_index(gpa, &.{"tags"}, false, false);
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" } } } } });
@@ -2317,7 +2402,7 @@ test "multikey expansion indexes the array and its elements" {
test "per-document dedup keeps {a: [1,1]} under a unique index" {
const gpa = testing.allocator;
var ix = try simple_index(gpa, &.{"a"}, true, false);
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);
@@ -2328,7 +2413,7 @@ test "per-document dedup keeps {a: [1,1]} under a unique index" {
test "parallel arrays are rejected" {
const gpa = testing.allocator;
var ix = try simple_index(gpa, &.{ "a", "b" }, false, false);
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 } },
@@ -2351,7 +2436,7 @@ test "parallel arrays are rejected" {
test "unique conflict across documents, replace of own entries allowed" {
const gpa = testing.allocator;
var ix = try simple_index(gpa, &.{"a"}, true, false);
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 } } });
@@ -2386,7 +2471,7 @@ test "iter_reverse yields every entry in exact reverse order" {
// wrong from the first entry; make descend_last follow `first_child`
// unconditionally and it silently misses everything to the right.
const gpa = testing.allocator;
var ix = try simple_index(gpa, &.{"a"}, false, false);
var ix = try simple_index(gpa, test_pager(), &.{"a"}, false, false);
defer ix.deinit(gpa);
const n = 400;
@@ -2444,7 +2529,7 @@ test "Candidates streams the same offsets a materialized plan would" {
// not here -- this index is not multikey. See Plan.full_scan for why that
// test only reddens when *both* multikey guards are removed.
const gpa = testing.allocator;
var ix = try simple_index(gpa, &.{"a"}, false, false);
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, &.{
@@ -2480,7 +2565,7 @@ test "lookup_exact matches whole keys only" {
// 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, &.{ "a", "b" }, false, false);
var ix = try simple_index(gpa, test_pager(), &.{ "a", "b" }, false, false);
defer ix.deinit(gpa);
const d = try bytes_of(gpa, &.{
@@ -2515,7 +2600,7 @@ test "lookup_exact matches whole keys only" {
test "range bounds inclusive and exclusive" {
const gpa = testing.allocator;
var ix = try simple_index(gpa, &.{"a"}, false, false);
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 },
@@ -2550,7 +2635,7 @@ test "range bounds inclusive and exclusive" {
test "empty index and remove_off" {
const gpa = testing.allocator;
var ix = try simple_index(gpa, &.{"a"}, false, false);
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);
@@ -2568,7 +2653,7 @@ test "empty index and remove_off" {
test "compound index prefix search and range on the next key" {
const gpa = testing.allocator;
var ix = try simple_index(gpa, &.{ "a", "b" }, false, false);
var ix = try simple_index(gpa, test_pager(), &.{ "a", "b" }, false, false);
defer ix.deinit(gpa);
const id1: u64 = 1;
@@ -2614,9 +2699,9 @@ test "remove_doc leaves the index identical to a full scan removal" {
const rand = prng.random();
for ([_]bool{ false, true }) |sparse| {
var by_doc = try simple_index(gpa, &.{ "a", "b" }, false, 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, &.{ "a", "b" }, false, sparse);
var by_scan = try simple_index(gpa, test_pager(), &.{ "a", "b" }, false, sparse);
defer by_scan.deinit(gpa);
const n = 120;
@@ -2706,7 +2791,7 @@ test "incremental inserts and removals stay identical to a brute-force model" {
var prng = std.Random.DefaultPrng.init(0x0dd_ba11);
const rand = prng.random();
var ix = try simple_index(gpa, &.{ "a", "b" }, false, false);
var ix = try simple_index(gpa, test_pager(), &.{ "a", "b" }, false, false);
defer ix.deinit(gpa);
var model: std.ArrayListUnmanaged(ModelFact) = .empty;
@@ -2803,7 +2888,7 @@ test "lookup_range matches a brute-force filter over random data" {
var ids: std.ArrayListUnmanaged(u64) = .empty;
defer ids.deinit(gpa);
var ix = try simple_index(gpa, &.{ "a", "b" }, false, false);
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
@@ -2881,7 +2966,7 @@ test "TTL spec round-trips through write_spec and compares in spec_equal" {
// A driver sends a plain JS number as a double.
.{ .key = "expireAfterSeconds", .value = .{ .double = 60.0 } },
});
var ix = try parse_spec(gpa, &spec);
var ix = try parse_spec(gpa, test_pager(), &spec);
defer ix.deinit(gpa);
try testing.expectEqual(@as(?i64, 60), ix.ttl);
@@ -2892,7 +2977,7 @@ test "TTL spec round-trips through write_spec and compares in spec_equal" {
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, &reparsed_doc);
var ix2 = try parse_spec(gpa, test_pager(), &reparsed_doc);
defer ix2.deinit(gpa);
try testing.expect(Index.spec_equal(&ix, &ix2));
@@ -2903,7 +2988,7 @@ test "TTL spec round-trips through write_spec and compares in spec_equal" {
.{ .key = "name", .value = .{ .string = "expireAt_1" } },
.{ .key = "expireAfterSeconds", .value = .{ .int32 = 90 } },
});
var ix3 = try parse_spec(gpa, &other);
var ix3 = try parse_spec(gpa, test_pager(), &other);
defer ix3.deinit(gpa);
try testing.expect(!Index.spec_equal(&ix, &ix3));
@@ -2913,14 +2998,14 @@ test "TTL spec round-trips through write_spec and compares in spec_equal" {
.{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } },
.{ .key = "expireAfterSeconds", .value = .{ .int64 = max_expire_after_seconds } },
});
var ix_big = try parse_spec(gpa, &big);
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, &big_doc);
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);
@@ -2929,7 +3014,7 @@ test "TTL spec round-trips through write_spec and compares in spec_equal" {
const plain = doc_of(&.{
.{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } },
});
var ix4 = try parse_spec(gpa, &plain);
var ix4 = try parse_spec(gpa, test_pager(), &plain);
defer ix4.deinit(gpa);
try testing.expect(ix4.ttl == null);
bytes.clearRetainingCapacity();
@@ -2948,7 +3033,7 @@ test "TTL spec rejects compound keys and bad expireAfterSeconds" {
} } },
.{ .key = "expireAfterSeconds", .value = .{ .int32 = 60 } },
});
try testing.expectError(error.TtlOnCompoundIndex, parse_spec(gpa, &compound));
try testing.expectError(error.TtlOnCompoundIndex, parse_spec(gpa, test_pager(), &compound));
const bad = [_]bson.Value{
.{ .int32 = -1 },
@@ -2971,7 +3056,7 @@ test "TTL spec rejects compound keys and bad expireAfterSeconds" {
.{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } },
.{ .key = "expireAfterSeconds", .value = v },
});
try testing.expectError(error.InvalidExpireAfterSeconds, parse_spec(gpa, &spec));
try testing.expectError(error.InvalidExpireAfterSeconds, parse_spec(gpa, test_pager(), &spec));
}
// 0 is legal: expire at exactly the stored instant.
@@ -2979,7 +3064,7 @@ test "TTL spec rejects compound keys and bad expireAfterSeconds" {
.{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } },
.{ .key = "expireAfterSeconds", .value = .{ .int32 = 0 } },
});
var ix = try parse_spec(gpa, &zero);
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.
@@ -3003,7 +3088,7 @@ test "a churned leaf of large keys splits without promoting from an empty half"
// 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, &.{"s"}, false, false);
var ix = try simple_index(gpa, test_pager(), &.{"s"}, false, false);
defer ix.deinit(gpa);
const n = 12;
@@ -3043,7 +3128,7 @@ test "a split with lopsided record sizes keeps the new record inside its page" {
// 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, &.{"s"}, false, false);
var ix = try simple_index(gpa, test_pager(), &.{"s"}, false, false);
defer ix.deinit(gpa);
var docs: std.ArrayListUnmanaged([]u8) = .empty;
@@ -3090,7 +3175,7 @@ test "the _id index plan covers equality, ranges and _id sort order" {
// 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, &.{"_id"}, false, false);
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, &.{
@@ -3172,9 +3257,9 @@ test "the _id index plan covers equality, ranges and _id sort order" {
test "planner picks eq run, ranges, and bails on sparse null" {
const gpa = testing.allocator;
var ix = try simple_index(gpa, &.{ "a", "b" }, false, false);
var ix = try simple_index(gpa, test_pager(), &.{ "a", "b" }, false, false);
defer ix.deinit(gpa);
var sp = try simple_index(gpa, &.{ "a", "b" }, false, true);
var sp = try simple_index(gpa, test_pager(), &.{ "a", "b" }, false, true);
defer sp.deinit(gpa);
// {a: 1, b: 2} → full-key equality.

73
src/pager.zig
View File

@@ -173,9 +173,15 @@ pub const Pager = struct {
file_pages: u32,
/// Pages [0, alloc_tail) have been handed out.
alloc_tail: u32,
/// Headroom promised by `reserve_pages`, so allocation after it is
/// infallible. Never below alloc_tail.
reserved_tail: u32,
/// Pages promised by `reserve_pages` and not yet handed out.
///
/// A count rather than a tail mark, because there is more than one consumer:
/// an upsert reserves tree pages for every index *and* slab room for the
/// document, all before the log append. A tail mark cannot express that --
/// the second reserver overwrites the first one's promise, and then the
/// first one's allocation asserts. Which is exactly what happened, on a
/// 512 MB load, with the tripwire in alloc_node catching it.
reserved_pages: u32,
/// True when this file was created by this open (no checkpoint to load).
fresh: bool,
@@ -252,7 +258,7 @@ pub const Pager = struct {
.mapped_pages = 0,
.file_pages = @intCast(existing_len / page_size),
.alloc_tail = page_first_data,
.reserved_tail = page_first_data,
.reserved_pages = 0,
.fresh = created,
.loaded = .{},
.generation = 0,
@@ -272,7 +278,6 @@ pub const Pager = struct {
// Everything already in the file is allocated until a watermark
// narrows it down.
self.alloc_tail = @max(page_first_data, self.file_pages);
self.reserved_tail = self.alloc_tail;
try self.load_watermark();
}
return self;
@@ -332,10 +337,7 @@ pub const Pager = struct {
/// Hand out `n` contiguous pages, growing the file if needed.
pub fn alloc_pages(self: *Pager, n: u32) !u32 {
assert(n > 0);
try self.grow_to(self.alloc_tail + n);
// Growing satisfies the promise `alloc_pages_assume_reserved` checks;
// without this it would assert against a reservation nobody made.
self.reserved_tail = @max(self.reserved_tail, self.alloc_tail + n);
try self.reserve_pages(n);
return self.alloc_pages_assume_reserved(n);
}
@@ -348,15 +350,31 @@ pub const Pager = struct {
/// leaves a sparse file, `ls -l` grows and `du` does not. That is what makes
/// a generous reservation cheap.
pub fn reserve_pages(self: *Pager, n: u32) !void {
try self.grow_to(self.alloc_tail + n);
self.reserved_tail = self.alloc_tail + n;
// Additive: room for what is already promised *plus* this. Two
// consumers reserving before the same log append must both be able to
// rely on their promise.
try self.grow_to(self.alloc_tail + self.reserved_pages + n);
self.reserved_pages += n;
}
/// Drop whatever is still promised but unclaimed.
///
/// A reservation is scoped to one write: it is taken before the log append
/// so the publish afterwards cannot fail, and once the publish is done
/// anything unclaimed is dead. Without this the promise accumulates -- a
/// tree reservation covers the worst case of several splits and a typical
/// insert causes none, so `reserved_pages` grew by a handful per write and
/// dragged the file up with it. It showed as a 1.89 GB data file for 512 MB
/// of documents.
pub fn release_reservation(self: *Pager) void {
self.reserved_pages = 0;
}
/// Hand out `n` pages against a previous `reserve_pages`. Infallible.
pub fn alloc_pages_assume_reserved(self: *Pager, n: u32) u32 {
assert(n > 0);
assert_msg(
self.alloc_tail + n <= self.reserved_tail,
n <= self.reserved_pages,
"page allocation overran reserve_pages' promise",
);
assert_msg(
@@ -365,6 +383,7 @@ pub const Pager = struct {
);
const first = self.alloc_tail;
self.alloc_tail += n;
self.reserved_pages -= n;
return first;
}
@@ -483,7 +502,7 @@ pub const Pager = struct {
self.loaded = wm;
self.generation = wm.generation;
self.alloc_tail = wm.alloc_tail;
self.reserved_tail = wm.alloc_tail;
self.reserved_pages = 0;
// Everything the published image references is off limits to
// writes from here on.
self.stable_pages = wm.alloc_tail;
@@ -830,6 +849,34 @@ test "the file is extended before any page in the range is reachable" {
}
}
test "two consumers reserving before one commit both keep their promise" {
// The reservation is a count, not a tail mark, and this is why. An upsert
// reserves tree pages for every index *and* slab room for the document,
// all before the log append, and both allocations happen after it. With a
// tail mark the second reserver overwrote the first one's promise and the
// first one's allocation then asserted -- which is how this was found, on a
// 512 MB load, by the tripwire in alloc_node.
//
// Mutation check: make reserve_pages assign `alloc_tail + n` instead of
// accumulating, and this goes red.
var threaded: std.Io.Threaded = .init_single_threaded;
defer threaded.deinit();
const io = threaded.io();
var tp = try TmpPager.init(io, 256 << 20);
defer tp.deinit();
const pg = tp.pg();
// Consumer A reserves a few pages, then consumer B reserves a large extent
// and takes it -- exactly the order upsert uses.
try pg.reserve_pages(8);
try pg.reserve_pages(2048);
const b_first = pg.alloc_pages_assume_reserved(2048);
// A's promise must have survived B's reservation *and* B's allocation.
const a_first = pg.alloc_pages_assume_reserved(8);
try testing.expect(a_first >= b_first + 2048);
try testing.expectEqual(@as(u32, 0), pg.reserved_pages);
}
test "a reservation makes the following allocation infallible" {
var threaded: std.Io.Threaded = .init_single_threaded;
defer threaded.deinit();

20
src/spill.zig
View File

@@ -4,6 +4,7 @@
// lookup, delete and iteration.
const std = @import("std");
const index = @import("index.zig");
const pgr = @import("pager.zig");
const bson = @import("bson.zig");
fn doc_of(gpa: std.mem.Allocator, pairs: []const bson.Pair) ![]u8 {
@@ -13,13 +14,26 @@ fn doc_of(gpa: std.mem.Allocator, pairs: []const bson.Pair) ![]u8 {
return out.toOwnedSlice(gpa);
}
/// A throwaway data file for this harness. The B+tree's pages live in the data
/// file now, so a standalone harness has to provide one.
fn harness_pager(gpa: std.mem.Allocator, name: []const u8) !*pgr.Pager {
const threaded = try gpa.create(std.Io.Threaded);
threaded.* = .init_single_threaded;
const io = threaded.io();
const path = try std.fmt.allocPrint(gpa, ".zig-cache/{s}.data", .{name});
std.Io.Dir.cwd().deleteFile(io, path) catch {};
const pg = try gpa.create(pgr.Pager);
pg.* = try pgr.Pager.open(gpa, io, path, .{});
return pg;
}
pub fn main() !void {
const gpa = std.heap.page_allocator;
var prng = std.Random.DefaultPrng.init(0x1234_5678);
const rand = prng.random();
var keys = [_]index.IndexKey{.{ .path = "tag", .descending = false }};
var ix = try index.Index.init(gpa, "tag", &keys, false, false, null);
var ix = try index.Index.init(gpa, try harness_pager(gpa, "spill"), "tag", &keys, false, false, null);
// Keys straddling the spill threshold: inline, exactly at the limit,
// just over, and one very long. Each id is a short static string.
@@ -41,9 +55,9 @@ pub fn main() !void {
ix.count(),
ix.leaf_count,
ix.depth,
ix.overflow.items.len,
ix.ovf_tail,
});
if (ix.overflow.items.len < 100_000) return error.NoSpill;
if (ix.ovf_tail < 100_000) return error.NoSpill;
// Every entry is found by exact key.
for (lens, 0..) |_, i| {

18
src/spill2.zig
View File

@@ -4,6 +4,7 @@
// exact lookups and deletes half of them.
const std = @import("std");
const index = @import("index.zig");
const pgr = @import("pager.zig");
const bson = @import("bson.zig");
fn doc_of(gpa: std.mem.Allocator, pairs: []const bson.Pair) ![]u8 {
@@ -13,11 +14,24 @@ fn doc_of(gpa: std.mem.Allocator, pairs: []const bson.Pair) ![]u8 {
return out.toOwnedSlice(gpa);
}
/// A throwaway data file for this harness. The B+tree's pages live in the data
/// file now, so a standalone harness has to provide one.
fn harness_pager(gpa: std.mem.Allocator, name: []const u8) !*pgr.Pager {
const threaded = try gpa.create(std.Io.Threaded);
threaded.* = .init_single_threaded;
const io = threaded.io();
const path = try std.fmt.allocPrint(gpa, ".zig-cache/{s}.data", .{name});
std.Io.Dir.cwd().deleteFile(io, path) catch {};
const pg = try gpa.create(pgr.Pager);
pg.* = try pgr.Pager.open(gpa, io, path, .{});
return pg;
}
pub fn main() !void {
const gpa = std.heap.page_allocator;
var keys = [_]index.IndexKey{.{ .path = "tag", .descending = false }};
var ix = try index.Index.init(gpa, "tag", &keys, false, false, null);
var ix = try index.Index.init(gpa, try harness_pager(gpa, "spill2"), "tag", &keys, false, false, null);
// 5000 docs, each with a 2 KiB key: spills on every record, forcing
// leaves and internal nodes to hold overflow references.
@@ -49,7 +63,7 @@ pub fn main() !void {
ix.count(),
ix.leaf_count,
ix.depth,
ix.overflow.items.len,
ix.ovf_tail,
});
if (ix.count() != N) return error.Bad;

24
src/stress.zig
View File

@@ -5,6 +5,7 @@
// lookups against a brute-force model throughout.
const std = @import("std");
const index = @import("index.zig");
const pgr = @import("pager.zig");
const bson = @import("bson.zig");
fn doc_of(gpa: std.mem.Allocator, pairs: []const bson.Pair) ![]u8 {
@@ -48,6 +49,19 @@ fn check_range(
}
}
/// A throwaway data file for this harness. The B+tree's pages live in the data
/// file now, so a standalone harness has to provide one.
fn harness_pager(gpa: std.mem.Allocator, name: []const u8) !*pgr.Pager {
const threaded = try gpa.create(std.Io.Threaded);
threaded.* = .init_single_threaded;
const io = threaded.io();
const path = try std.fmt.allocPrint(gpa, ".zig-cache/{s}.data", .{name});
std.Io.Dir.cwd().deleteFile(io, path) catch {};
const pg = try gpa.create(pgr.Pager);
pg.* = try pgr.Pager.open(gpa, io, path, .{});
return pg;
}
pub fn main() !void {
const gpa = std.heap.page_allocator;
var prng = std.Random.DefaultPrng.init(0xBADCAFE);
@@ -61,7 +75,7 @@ pub fn main() !void {
// 1. Bulk build an index over N docs.
var keys = [_]index.IndexKey{ .{ .path = "a", .descending = false }, .{ .path = "b", .descending = false } };
var ix = try index.Index.init(gpa, "ab", &keys, false, false, null);
var ix = try index.Index.init(gpa, try harness_pager(gpa, "stress"), "ab", &keys, false, false, null);
for (0..N) |i| {
const id: u64 = @intCast(i + 1);
try ids.append(gpa, id);
@@ -79,7 +93,7 @@ pub fn main() !void {
ix.count(),
ix.leaf_count,
ix.depth,
ix.nodes.items.len,
ix.node_pages.items.len,
});
if (ix.count() != N) return error.BadCount;
@@ -111,7 +125,7 @@ pub fn main() !void {
ix.count(),
ix.leaf_count,
ix.depth,
ix.nodes.items.len,
ix.node_pages.items.len,
});
if (ix.count() != N + M) return error.BadCount;
for (0..500) |_| {
@@ -141,7 +155,7 @@ pub fn main() !void {
ix.count(),
ix.leaf_count,
ix.depth,
ix.nodes.items.len,
ix.node_pages.items.len,
});
if (ix.count() != (N + M) - (N + M) / 3 - 1) return error.BadCount;
for (0..500) |_| {
@@ -200,7 +214,7 @@ pub fn main() !void {
ix.count(),
ix.leaf_count,
ix.depth,
ix.nodes.items.len,
ix.node_pages.items.len,
});
if (ix.count() != 0) return error.BadCount;
// The drained tree still accepts and finds entries.