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

storage: byte documents in a per-collection slab (roadmap item 4)

Browse Source 
Documents live as canonical BSON bytes in a segmented per-collection slab
(fixed 8 MiB segments keep capacity slack under one segment); the docs map
holds flat offsets that stay valid across segment growth, and removed
documents leave garbage bytes until compaction rewrites. The per-document
ArenaAllocator and its second full Pair-tree copy are gone.

The matcher walks the stored bytes directly, skipping by length any field
the filter does not name (a new bson byte-walker: element_key, skip_value,
read_value with borrowed leaves, get_at, and a borrowed spine parse). The
byte matcher is differential-tested against the tree matcher on a corpus
and shares its operator logic. Stored documents are never materialized on
the scan path or in aggregate $match; $group reads group keys and sums
straight off the bytes. Sort, projection, findAndModify, updates and
index entry generation use a borrowed spine into the slab (or the byte
collector, which also replaced collect_values in build_entries). The
compaction threshold now counts uncompressed data volume, since a
compressed log would otherwise never trigger.

Measured (tests/e2e/results/phase5.txt): server RSS 1979 -> 539 MB (2.4x
smaller than MongoDB; phase1 baseline 2.0 GB), range-scan 22.5 -> ~12 ms
(parity, best run faster than MongoDB), proj 4.1 -> 3.4 ms, createIndex
parity. Verified: unit suite in all three modes with zero leaks, the
crash pair, e2e6, and the stress/spill programs.
This commit is contained in:
Alexey Shakhmatov
2026-08-02 22:15:07 +03:00
parent b4585106f1
commit 570900a6ef
12 changed files with 985 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files

189
src/commands.zig
View File

@@ -6,6 +6,7 @@ const builtin = @import("builtin");
const bson = @import("bson.zig");
const wire = @import("wire.zig");
const db = @import("db.zig");
const Collection = db.Collection;
const query = @import("query.zig");
const update = @import("update.zig");
const index = @import("index.zig");
@@ -569,7 +570,7 @@ fn cmd_find(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
// reply with one batch, so only the magnitude matters.
const limit: usize = @abs(int_value(msg.body.get("limit")) orelse 0);
var matched: std.ArrayListUnmanaged(*const bson.Document) = .empty;
var matched: std.ArrayListUnmanaged(u64) = .empty;
defer matched.deinit(ctx.gpa);
// Documents needed to fill the page, counting the skipped prefix; 0
// means unbounded.
@@ -583,19 +584,28 @@ fn cmd_find(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
var index_sorted = false;
_ = try scan_sorted(ctx, db_name, coll_name, filter, page_end, &matched, sort_keys, &index_sorted);
// Sorting and emitting need the documents as trees; materialize the
// matched page into the reply arena (the slab itself is never copied).
const coll = ctx.engine.get_collection(db_name, coll_name) orelse return;
// Lives in the reply arena; freed with it.
var tree_docs: std.ArrayListUnmanaged(*const bson.Document) = .empty;
const arena = reply.arena_alloc();
for (matched.items) |off| {
try tree_docs.append(arena, try doc_tree(arena, coll, off));
}
if (sort_keys.len > 0 and !index_sorted) {
// Selecting the page is much cheaper than ordering everything when
// the page is a small fraction of the matches. Above that fraction
// the heap's bookkeeping stops paying for itself.
if (page_end > 0 and page_end *| 4 <= matched.items.len) {
try query.sort_docs_top_k(reply.arena_alloc(), matched.items, sort_keys, page_end);
if (page_end > 0 and page_end *| 4 <= tree_docs.items.len) {
try query.sort_docs_top_k(arena, tree_docs.items, sort_keys, page_end);
} else {
try query.sort_docs(reply.arena_alloc(), matched.items, sort_keys);
try query.sort_docs(arena, tree_docs.items, sort_keys);
}
}
const rest = if (skip < matched.items.len) matched.items[skip..] else &.{};
const rest = if (skip < tree_docs.items.len) tree_docs.items[skip..] else &.{};
const page = if (limit > 0 and limit < rest.len) rest[0..limit] else rest;
try emit_docs(reply, db_name, coll_name, proj_pairs, page);
try emit_docs_tree(reply, db_name, coll_name, proj_pairs, page);
try reply.put_ok();
}
@@ -615,7 +625,7 @@ fn scan_matching(
coll_name: []const u8,
filter: []const bson.Pair,
limit: usize,
out: ?*std.ArrayListUnmanaged(*const bson.Document),
out: ?*std.ArrayListUnmanaged(u64),
) !usize {
return scan_sorted(ctx, db_name, coll_name, filter, limit, out, &.{}, null);
}
@@ -630,7 +640,7 @@ fn scan_sorted(
coll_name: []const u8,
filter: []const bson.Pair,
limit: usize,
out: ?*std.ArrayListUnmanaged(*const bson.Document),
out: ?*std.ArrayListUnmanaged(u64),
sort: []const query.SortKey,
sorted: ?*bool,
) !usize {
@@ -641,7 +651,6 @@ fn scan_sorted(
// out to supply the ordering.
var lim: usize = if (sort.len == 0) limit else 0;
const coll = ctx.engine.get_collection(db_name, coll_name) orelse return 0;
const filter_doc = bson.Document{ .arena = undefined, .pairs = filter };
var n: usize = 0;
// Index plan (the implicit _id_ index first, then the secondaries):
@@ -656,9 +665,9 @@ fn scan_sorted(
if (sorted) |flag| flag.* = plan.provides_sort;
if (plan.provides_sort) lim = limit;
for (ids.items) |id| {
const doc = coll.docs.get(id) orelse continue;
if (!try query.matches(ctx.gpa, &filter_doc, doc)) continue;
if (out) |list| try list.append(ctx.gpa, doc);
const off = coll.docs.get(id) orelse continue;
if (!try query.matches_bytes(ctx.gpa, filter, coll.doc_bytes(off))) continue;
if (out) |list| try list.append(ctx.gpa, off);
n += 1;
if (lim != 0 and n >= lim) break;
}
@@ -667,7 +676,7 @@ fn scan_sorted(
var it = coll.docs.iterator();
while (it.next()) |entry| {
if (!try query.matches(ctx.gpa, &filter_doc, entry.value_ptr.*)) continue;
if (!try query.matches_bytes(ctx.gpa, filter, coll.doc_bytes(entry.value_ptr.*))) continue;
if (out) |list| try list.append(ctx.gpa, entry.value_ptr.*);
n += 1;
if (lim != 0 and n >= lim) break;
@@ -675,7 +684,17 @@ fn scan_sorted(
return n;
}
fn emit_docs(reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, proj_pairs: ?[]const bson.Pair, docs: []const *const bson.Document) !void {
/// A stored document (a slab offset) materialized as a borrowed spine in
/// `arena`: keys and leaf values point into the slab's stable bytes, only
/// the pair/value skeleton is allocated. The arena owns the skeleton, so
/// the result is never deinit'd — the reply arena frees it with the reply.
fn doc_tree(arena: std.mem.Allocator, coll: *const Collection, off: u64) !*const bson.Document {
const doc = try arena.create(bson.Document);
doc.* = bson.Document{ .arena = undefined, .pairs = try bson.spine(arena, coll.doc_bytes(off)) };
return doc;
}
fn emit_docs_tree(reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, proj_pairs: ?[]const bson.Pair, docs: []const *const bson.Document) !void {
const values = try reply.arena_alloc().alloc(bson.Value, docs.len);
for (docs, 0..) |d, i| {
values[i] = try project_doc(reply, d, proj_pairs);
@@ -717,7 +736,7 @@ fn cmd_update(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
const multi = bool_arg(spec.get("multi")) orelse false;
const upsert = bool_arg(spec.get("upsert")) orelse false;
var matched: std.ArrayListUnmanaged(*const bson.Document) = .empty;
var matched: std.ArrayListUnmanaged(u64) = .empty;
defer matched.deinit(ctx.gpa);
_ = try scan_matching(ctx, db_name, coll_name, q, if (multi) 0 else 1, &matched);
@@ -739,9 +758,11 @@ fn cmd_update(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
}
n_matched += @intCast(matched.items.len);
for (matched.items) |doc| {
const coll = ctx.engine.get_collection(db_name, coll_name) orelse return;
for (matched.items) |off| {
// Work on a copy: the log write must precede any visible change,
// and a rejected update must not corrupt the stored document.
const doc = try doc_tree(reply.arena_alloc(), coll, off);
const copy = try clone_doc(reply, doc);
update.apply(copy, &.{ .arena = undefined, .pairs = u_doc }) catch |err| switch (err) {
error.ImmutableId, error.InvalidUpdate => return bad_value(reply, "bad update"),
@@ -792,12 +813,16 @@ fn cmd_delete(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
const q = doc_arg(spec.get("q")) orelse return bad_value(reply, "delete spec requires q");
const limit = int_value(spec.get("limit")) orelse 1;
var matched: std.ArrayListUnmanaged(*const bson.Document) = .empty;
var matched: std.ArrayListUnmanaged(u64) = .empty;
defer matched.deinit(ctx.gpa);
_ = try scan_matching(ctx, db_name, coll_name, q, if (limit == 1) 1 else 0, &matched);
for (matched.items) |doc| {
const id = doc.get("_id") orelse continue;
if (try ctx.engine.remove_by_id(db_name, coll_name, id)) n_deleted += 1;
if (ctx.engine.get_collection(db_name, coll_name)) |coll| {
var id_arena = std.heap.ArenaAllocator.init(ctx.gpa);
defer id_arena.deinit();
for (matched.items) |off| {
const id = (try bson.get_at(id_arena.allocator(), coll.doc_bytes(off), "_id")) orelse continue;
if (try ctx.engine.remove_by_id(db_name, coll_name, id)) n_deleted += 1;
}
}
}
@@ -820,16 +845,21 @@ fn cmd_find_and_modify(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !v
if (remove and do_update) return bad_value(reply, "remove and update are mutually exclusive");
if (!remove and !do_update) return bad_value(reply, "must specify update or remove");
var matched: std.ArrayListUnmanaged(*const bson.Document) = .empty;
var matched: std.ArrayListUnmanaged(u64) = .empty;
defer matched.deinit(ctx.gpa);
// Without a sort, only the first match is ever used.
_ = try scan_matching(ctx, db_name, coll_name, q, if (sort_keys.len > 0) 0 else 1, &matched);
const arena = reply.arena_alloc();
const coll = ctx.engine.get_collection(db_name, coll_name) orelse return;
// findAndModify reads and rewrites the document, so materialize the
// (usually tiny) match set as trees in the reply arena.
var matched_docs: std.ArrayListUnmanaged(*const bson.Document) = .empty;
for (matched.items) |off| try matched_docs.append(arena, try doc_tree(arena, coll, off));
if (sort_keys.len > 0) {
try query.sort_docs(reply.arena_alloc(), matched.items, sort_keys);
try query.sort_docs(arena, matched_docs.items, sort_keys);
}
const arena = reply.arena_alloc();
const target = if (matched.items.len > 0) matched.items[0] else null;
const target = if (matched_docs.items.len > 0) matched_docs.items[0] else null;
// Each branch decides what the reply says; the tail below emits it once.
var n: i32 = 0;
@@ -927,28 +957,33 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
one[0] = doc;
docs = one;
}
try emit_docs(reply, db_name, coll_name, null, docs);
try emit_docs_tree(reply, db_name, coll_name, null, docs);
return reply.put_ok();
}
// The pipeline operates on a stream of documents; each stage transforms
// the current window [start, end) of `stream`, and $group replaces the
// stream entirely (so $sort/$limit after it apply to the groups).
var stream: std.ArrayListUnmanaged(*const bson.Document) = .empty;
defer stream.deinit(ctx.gpa);
// the current window [start, end). Before $group the stream holds slab
// offsets (matched in place, never materialized); $group replaces it
// with generated group documents, so the stream flips to tree form.
var offs: std.ArrayListUnmanaged(u64) = .empty;
defer offs.deinit(ctx.gpa);
var trees: std.ArrayListUnmanaged(*const bson.Document) = .empty;
defer trees.deinit(ctx.gpa);
var in_trees = false;
const coll = ctx.engine.get_collection(db_name, coll_name) orelse return;
// A leading $match is pushed down into an indexed candidate scan; the
// stage is then dropped from the pipeline so it is not applied twice.
if (stages.len > 0 and stages[0] == .doc and stages[0].doc.len > 0 and std.mem.eql(u8, stages[0].doc[0].key, "$match")) {
const filter = doc_arg(stages[0].doc[0].value) orelse return bad_value(reply, "$match requires a document");
_ = try scan_matching(ctx, db_name, coll_name, filter, 0, &stream);
_ = try scan_matching(ctx, db_name, coll_name, filter, 0, &offs);
stages = stages[1..];
} else if (ctx.engine.get_collection(db_name, coll_name)) |coll| {
} else {
var it = coll.docs.iterator();
while (it.next()) |entry| try stream.append(ctx.gpa, entry.value_ptr.*);
while (it.next()) |entry| try offs.append(ctx.gpa, entry.value_ptr.*);
}
var start: usize = 0;
var end: usize = stream.items.len;
var end: usize = offs.items.len;
var count_stage: ?[]const u8 = null;
var proj_pairs: ?[]const bson.Pair = null;
@@ -961,22 +996,46 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
const stage_name = stage[0].key;
if (std.mem.eql(u8, stage_name, "$match")) {
const filter = doc_arg(stage[0].value) orelse return bad_value(reply, "$match requires a document");
var kept: std.ArrayListUnmanaged(*const bson.Document) = .empty;
defer kept.deinit(ctx.gpa);
for (stream.items[start..end]) |d| {
if (try query.matches(ctx.gpa, &.{ .arena = undefined, .pairs = filter }, d)) {
try kept.append(ctx.gpa, d);
if (!in_trees) {
var kept: std.ArrayListUnmanaged(u64) = .empty;
defer kept.deinit(ctx.gpa);
for (offs.items[start..end]) |off| {
if (try query.matches_bytes(ctx.gpa, filter, coll.doc_bytes(off))) {
try kept.append(ctx.gpa, off);
}
}
offs.deinit(ctx.gpa);
offs = kept;
kept = .empty;
} else {
var kept: std.ArrayListUnmanaged(*const bson.Document) = .empty;
defer kept.deinit(ctx.gpa);
for (trees.items[start..end]) |d| {
if (try query.matches(ctx.gpa, &.{ .arena = undefined, .pairs = filter }, d)) {
try kept.append(ctx.gpa, d);
}
}
trees.deinit(ctx.gpa);
trees = kept;
kept = .empty;
}
stream.deinit(ctx.gpa);
stream = kept;
kept = .empty;
start = 0;
end = stream.items.len;
end = (if (in_trees) trees.items.len else offs.items.len);
} else if (std.mem.eql(u8, stage_name, "$sort")) {
const keys = try parse_sort_keys(reply, stage[0].value);
if (keys.len > 0) {
try query.sort_docs(reply.arena_alloc(), stream.items[start..end], keys);
const arena = reply.arena_alloc();
if (!in_trees) {
// Sorting needs the values; materialize and switch the
// stream to tree form for the rest of the pipeline.
var all: std.ArrayListUnmanaged(*const bson.Document) = .empty;
for (offs.items) |off| try all.append(arena, try doc_tree(arena, coll, off));
trees.deinit(ctx.gpa);
trees = all;
all = .empty;
in_trees = true;
}
try query.sort_docs(arena, trees.items[start..end], keys);
}
} else if (std.mem.eql(u8, stage_name, "$skip")) {
const n = try stage_count(reply, stage[0].value, "$skip") orelse return;
@@ -988,13 +1047,17 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
proj_pairs = doc_arg(stage[0].value);
} else if (std.mem.eql(u8, stage_name, "$group")) {
const gp = doc_arg(stage[0].value) orelse return bad_value(reply, "$group requires a document");
const grouped_opt = try run_group(ctx, reply, gp, stream.items[start..end]);
const grouped = grouped_opt orelse return;
const grouped_opt = try run_group(ctx, reply, coll, gp, offs.items[start..end]);
var grouped = grouped_opt orelse return;
// Group results replace the stream: later stages see groups.
stream.deinit(ctx.gpa);
stream = grouped;
offs.deinit(ctx.gpa);
offs = .empty;
trees.deinit(ctx.gpa);
trees = grouped;
grouped = .empty;
in_trees = true;
start = 0;
end = stream.items.len;
end = trees.items.len;
} else if (std.mem.eql(u8, stage_name, "$count")) {
count_stage = switch (stage[0].value) {
.string => |s| s,
@@ -1006,16 +1069,22 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
}
}
const slice = stream.items[start..end];
if (count_stage) |name| {
const len = if (in_trees) trees.items[start..end].len else offs.items[start..end].len;
const c = try reply.arena_alloc().alloc(bson.Pair, 1);
c[0] = .{ .key = try reply.arena_alloc().dupe(u8, name), .value = .{ .int32 = @intCast(slice.len) } };
c[0] = .{ .key = try reply.arena_alloc().dupe(u8, name), .value = .{ .int32 = @intCast(len) } };
const values = try reply.arena_alloc().alloc(bson.Value, 1);
values[0] = .{ .doc = c };
try reply.put("cursor", .{ .doc = try cursor_doc(reply, 0, try format_namespace(reply, db_name, coll_name), "firstBatch", values) });
} else {
try emit_docs(reply, db_name, coll_name, proj_pairs, slice);
const arena = reply.arena_alloc();
if (in_trees) {
try emit_docs_tree(reply, db_name, coll_name, proj_pairs, trees.items[start..end]);
} else {
var page: std.ArrayListUnmanaged(*const bson.Document) = .empty;
for (offs.items[start..end]) |off| try page.append(arena, try doc_tree(arena, coll, off));
try emit_docs_tree(reply, db_name, coll_name, proj_pairs, page.items);
}
}
try reply.put_ok();
}
@@ -1087,7 +1156,7 @@ fn count_only_pipeline(reply: *wire.Reply, stages: []const bson.Value) !?CountSh
/// Minimal $group: supports `_id` of null/literal/"$field" and `$sum`
/// accumulators (constant or "$field").
fn run_group(ctx: *Context, reply: *wire.Reply, group_pairs: []const bson.Pair, docs: []const *const bson.Document) !?std.ArrayListUnmanaged(*const bson.Document) {
fn run_group(ctx: *Context, reply: *wire.Reply, coll: *const Collection, group_pairs: []const bson.Pair, docs: []const u64) !?std.ArrayListUnmanaged(*const bson.Document) {
const arena = reply.arena_alloc();
const id_expr = bson.get_pair(group_pairs, "_id") orelse {
try bad_value(reply, "$group requires _id");
@@ -1116,9 +1185,13 @@ fn run_group(ctx: *Context, reply: *wire.Reply, group_pairs: []const bson.Pair,
var id_key_buf: std.ArrayListUnmanaged(u8) = .empty;
defer id_key_buf.deinit(ctx.gpa);
for (docs) |doc| {
// Byte-walk materializations (nested group keys) live here.
var walk_arena = std.heap.ArenaAllocator.init(ctx.gpa);
defer walk_arena.deinit();
for (docs) |off| {
const doc = coll.doc_bytes(off);
const id_value: bson.Value = switch (id_expr) {
.string => |s| if (s.len > 0 and s[0] == '$') query_path_value(doc, s[1..]) orelse .null else id_expr,
.string => |s| if (s.len > 0 and s[0] == '$') (try query_path_value_bytes(walk_arena.allocator(), doc, s[1..])) orelse .null else id_expr,
else => id_expr,
};
id_key_buf.clearRetainingCapacity();
@@ -1148,7 +1221,7 @@ fn run_group(ctx: *Context, reply: *wire.Reply, group_pairs: []const bson.Pair,
.int64 => |n| @floatFromInt(n),
.double => |n| n,
.string => |s| if (s.len > 0 and s[0] == '$')
switch (query_path_value(doc, s[1..]) orelse .null) {
switch ((try query_path_value_bytes(walk_arena.allocator(), doc, s[1..])) orelse .null) {
.int32 => |n| @floatFromInt(n),
.int64 => |n| @floatFromInt(n),
.double => |n| n,
@@ -1190,11 +1263,11 @@ fn run_group(ctx: *Context, reply: *wire.Reply, group_pairs: []const bson.Pair,
}
/// Resolve a simple "$field" path expression inside a document.
fn query_path_value(doc: *const bson.Document, path: []const u8) ?bson.Value {
fn query_path_value_bytes(gpa: std.mem.Allocator, bytes: []const u8, path: []const u8) !?bson.Value {
var cur: bson.Value = undefined;
var it = std.mem.splitScalar(u8, path, '.');
const first = it.next() orelse return null;
cur = bson.get_pair(doc.pairs, first) orelse return null;
cur = (try bson.get_at(gpa, bytes, first)) orelse return null;
while (it.next()) |seg| {
cur = switch (cur) {
.doc => |pairs| bson.get_pair(pairs, seg) orelse return null,