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

index/db/commands: fold duplicated index logic into single definitions

Browse Source 
Cleanup pass over the secondary-index feature.

add_doc is now the one entry-commit path. create_index and
build_all_indexes each hand-rolled build -> check_unique -> reserve ->
insert, and had already drifted on whether multikey is set before or
after the unique check; add_doc gained an enforce_unique flag so the
rebuild path keeps its tolerate-and-warn behavior. reserve_for and
insert_entries are now the only way the engine touches Index.entries.

One definition each for: prefix comparison and the prefix binary
searches (prefix_order + std.sort), the cartesian-product odometer
(advance_choice), the spec pair list (write_spec builds on spec_pairs,
so the log format and the listIndexes reply share one schema), the _id
clause parser (plan_id reuses analyze_clause), key-pattern direction
(index.descending, which desc_dir already disagreed with on non-numeric
values), option truthiness (query.truthy), the E11000 message, and
index-removal-by-name (Collection.find_index/remove_index). Key-pattern
matching moved out of the dispatcher into index.find_by_key_pattern.

Dead or redundant: ParallelArraysError, the unread `dropped` counter,
insert_entries' discarded gpa, a third pass computing multikey, the
has_id/is_id_index flag pair, Plan.key_len (always lookup_keys[0].len,
now a method), first_match_consumed (now stages = stages[1..]).

Cheaper hot paths: remove_id compacts in one pass instead of an
orderedRemove per hit; Plan.search skips the sort/dedupe when neither
multikey nor multiple lookup keys can produce a repeat; the _id fast
path reuses one scratch key buffer (bson.write_serialized_value); the
plan loop uses the bound collection instead of re-resolving it through
two hash lookups per candidate.

Behavior is unchanged except that dropping plan_id's fixed 16-clause
buffer enables the _id fast path on filters that previously exceeded it.
This commit is contained in:
Alexey Shakhmatov
2026-08-02 13:40:42 +03:00
parent 0d264c6c57
commit 7482042f34
5 changed files with 260 additions and 334 deletions
Download Patch File Download Diff File Expand all files Collapse all files

143
src/commands.zig
View File

@@ -336,18 +336,17 @@ fn cmd_create_indexes(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !vo
// {_id: 1} is the implicit index: an idempotent no-op. Any other
// secondary index touching _id is rejected.
var has_id = false;
var is_id_index = false;
for (key_pairs) |p| {
if (std.mem.eql(u8, p.key, "_id")) has_id = true;
}
if (key_pairs.len == 1 and has_id) {
if (has_id) {
const only = key_pairs[0].value;
is_id_index = (only == .int32 and only.int32 == 1) or
const is_id_index = key_pairs.len == 1 and ((only == .int32 and only.int32 == 1) or
(only == .int64 and only.int64 == 1) or
(only == .double and only.double == 1.0);
(only == .double and only.double == 1.0));
if (is_id_index) continue;
return bad_value(reply, "cannot create a secondary index on the _id field");
}
if (is_id_index) continue;
if (has_id) return bad_value(reply, "cannot create a secondary index on the _id field");
const name = bson.get_pair(spec, "name") orelse bson.Value.null;
if (name == .string and std.mem.eql(u8, name.string, "_id_")) {
return bad_value(reply, "cannot create index with name '_id_'");
@@ -359,11 +358,7 @@ fn cmd_create_indexes(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !vo
error.IndexOptionsConflict => return reply.put_error(@intFromEnum(ErrorCode.index_options_conflict), "IndexOptionsConflict", "index already exists with a different specification"),
error.DuplicateKeyIndex => {
const ix_name = if (name == .string) name.string else "index";
const msg_text = try std.fmt.allocPrint(
reply.arena_alloc(),
"E11000 duplicate key error collection: {s}.{s} index: {s} dup key: {s}",
.{ db_name, coll_name, ix_name, try render_spec_key(reply, key_pairs) },
);
const msg_text = try e11000_message(reply, db_name, coll_name, ix_name, try render_spec_key(reply, key_pairs));
return reply.put_error(@intFromEnum(ErrorCode.duplicate_key), "DuplicateKey", msg_text);
},
error.ParallelArrays => return bad_value(reply, "cannot index parallel arrays"),
@@ -418,23 +413,18 @@ fn cmd_drop_indexes(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void
const n_indexes_was: i32 = @intCast(coll.indexes.items.len + 1);
const arg = msg.body.get("index") orelse return bad_value(reply, "dropIndexes requires index");
var dropped: usize = 0;
if (arg == .string and std.mem.eql(u8, arg.string, "*")) {
// Drop every secondary index. Copy the names first: each drop
// mutates the collection's index list.
var names: std.ArrayListUnmanaged([]const u8) = .empty;
defer names.deinit(ctx.gpa);
for (coll.indexes.items) |ix| try names.append(ctx.gpa, ix.name);
for (names.items) |nm| {
if (try ctx.engine.drop_index(db_name, coll_name, nm)) dropped += 1;
}
for (names.items) |nm| _ = try ctx.engine.drop_index(db_name, coll_name, nm);
} else if (arg == .string) {
if (std.mem.eql(u8, arg.string, "_id_")) {
return invalid_arg(reply, "cannot drop the _id index");
}
if (try ctx.engine.drop_index(db_name, coll_name, arg.string)) {
dropped += 1;
} else {
if (!try ctx.engine.drop_index(db_name, coll_name, arg.string)) {
return reply.put_error(@intFromEnum(ErrorCode.index_not_found), "IndexNotFound", "index not found with name");
}
} else if (arg == .doc) {
@@ -444,23 +434,9 @@ fn cmd_drop_indexes(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void
.doc => |p| p,
else => return bad_value(reply, "dropIndexes index must be a name, key document, or '*'"),
};
var target: ?[]const u8 = null;
for (coll.indexes.items) |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 != (desc_dir(kp.value) < 0)) {
match = false;
break;
}
}
if (match) {
target = ix.name;
break;
}
}
const nm = target orelse return reply.put_error(@intFromEnum(ErrorCode.index_not_found), "IndexNotFound", "index not found with key pattern");
if (try ctx.engine.drop_index(db_name, coll_name, nm)) dropped += 1;
const target = index.find_by_key_pattern(coll.indexes.items, key_pairs) orelse
return reply.put_error(@intFromEnum(ErrorCode.index_not_found), "IndexNotFound", "index not found with key pattern");
_ = try ctx.engine.drop_index(db_name, coll_name, target.name);
} else {
return bad_value(reply, "dropIndexes index must be a name, key document, or '*'");
}
@@ -469,13 +445,14 @@ fn cmd_drop_indexes(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void
try reply.put_ok();
}
fn desc_dir(v: bson.Value) i32 {
return switch (v) {
.int32 => |n| if (n < 0) -1 else 1,
.int64 => |n| if (n < 0) -1 else 1,
.double => |n| if (n < 0) -1 else 1,
else => 1,
};
/// The E11000 text drivers parse. One definition for both create-time and
/// write-time conflicts; they differ only in how the dup key is rendered.
fn e11000_message(reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, index_name: []const u8, key_text: []const u8) ![]const u8 {
return std.fmt.allocPrint(
reply.arena_alloc(),
"E11000 duplicate key error collection: {s}.{s} index: {s} dup key: {s}",
.{ db_name, coll_name, index_name, key_text },
);
}
/// Render the key pattern with placeholder values for a createIndexes
@@ -588,20 +565,14 @@ fn scan_matching(
// queried value's compare-equivalence class is serialization-ambiguous
// (see index.plan_id).
if (index.plan_id(filter)) |id_plan| {
var keys: std.ArrayListUnmanaged([]u8) = .empty;
defer {
for (keys.items) |k| ctx.gpa.free(k);
keys.deinit(ctx.gpa);
}
// One scratch key, rebuilt per value: a key is never needed past its
// own lookup.
var key: std.ArrayListUnmanaged(u8) = .empty;
defer key.deinit(ctx.gpa);
for (id_plan.values) |v| {
const key = try bson.serialize_value(ctx.gpa, v);
keys.append(ctx.gpa, key) catch |err| {
ctx.gpa.free(key);
return err;
};
}
for (keys.items) |key| {
const doc = coll.docs.get(key) orelse continue;
key.clearRetainingCapacity();
try bson.write_serialized_value(v, ctx.gpa, &key);
const doc = coll.docs.get(key.items) orelse continue;
if (!try query.matches(ctx.gpa, &filter_doc, doc)) continue;
if (out) |list| try list.append(ctx.gpa, doc);
n += 1;
@@ -619,7 +590,7 @@ fn scan_matching(
defer ids.deinit(ctx.gpa);
try plan.search(ctx.gpa, &ids);
for (ids.items) |id| {
const doc = ctx.engine.get_doc(db_name, coll_name, id) orelse continue;
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);
n += 1;
@@ -846,7 +817,7 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
const coll_name = str_arg(msg.body.get("aggregate")) orelse return bad_value(reply, "aggregate requires a collection name");
const pipeline_value = msg.body.get("pipeline") orelse return bad_value(reply, "aggregate requires pipeline");
const stages = switch (pipeline_value) {
var stages = switch (pipeline_value) {
.array => |a| a,
else => return bad_value(reply, "pipeline must be an array"),
};
@@ -857,12 +828,11 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
var stream: std.ArrayListUnmanaged(*const bson.Document) = .empty;
defer stream.deinit(ctx.gpa);
// A leading $match is pushed down into an indexed candidate scan; the
// stage is then consumed so it is not applied a second time.
var first_match_consumed = false;
// 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);
first_match_consumed = true;
stages = stages[1..];
} else if (ctx.engine.get_collection(db_name, coll_name)) |coll| {
var it = coll.docs.iterator();
while (it.next()) |entry| try stream.append(ctx.gpa, entry.value_ptr.*);
@@ -874,10 +844,6 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
var proj_pairs: ?[]const bson.Pair = null;
for (stages) |stage_v| {
if (first_match_consumed) {
first_match_consumed = false;
continue;
}
const stage = switch (stage_v) {
.doc => |pairs| pairs,
else => return bad_value(reply, "pipeline stages must be documents"),
@@ -1214,7 +1180,6 @@ fn bad_value(reply: *wire.Reply, msg: []const u8) !void {
/// rejected unique-index write) when the conflict came from a secondary
/// index; otherwise it is the _id_ index.
fn duplicate_key_message(ctx: *Context, reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, doc: *const bson.Document) ![]const u8 {
const arena = reply.arena_alloc();
var index_name: []const u8 = "_id_";
var key_text: []const u8 = undefined;
if (ctx.engine.dup_index) |name| {
@@ -1223,40 +1188,34 @@ fn duplicate_key_message(ctx: *Context, reply: *wire.Reply, db_name: []const u8,
} else {
key_text = try serialize_value_compact(reply, doc.get("_id") orelse bson.Value.null);
}
return std.fmt.allocPrint(
arena,
"E11000 duplicate key error collection: {s}.{s} index: {s} dup key: {s}",
.{ db_name, coll_name, index_name, key_text },
);
return e11000_message(reply, db_name, coll_name, index_name, key_text);
}
/// Render the dup key of a secondary index from the offending document: the
/// document's values for the index key pattern.
fn render_dup_key(ctx: *Context, reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, index_name: []const u8, doc: *const bson.Document) ![]const u8 {
const arena = reply.arena_alloc();
if (ctx.engine.get_collection(db_name, coll_name)) |coll| {
for (coll.indexes.items) |*ix| {
if (std.mem.eql(u8, ix.name, index_name)) {
var out: std.ArrayListUnmanaged(u8) = .empty;
defer out.deinit(arena);
try out.append(arena, '{');
for (ix.keys, 0..) |k, i| {
if (i > 0) try out.appendSlice(arena, ", ");
try out.appendSlice(arena, k.path);
try out.appendSlice(arena, ": ");
var values: std.ArrayListUnmanaged(bson.Value) = .empty;
defer values.deinit(arena);
try query.collect_values(arena, doc.pairs, k.path, &values, 0);
const v: bson.Value = if (values.items.len > 0) values.items[0] else .null;
try out.appendSlice(arena, try serialize_value_compact(reply, v));
}
try out.append(arena, '}');
return out.toOwnedSlice(arena);
}
}
const coll = ctx.engine.get_collection(db_name, coll_name) orelse
// Index not found (defensive): fall back to the document's _id.
return serialize_value_compact(reply, doc.get("_id") orelse bson.Value.null);
const ix = coll.find_index(index_name) orelse
return serialize_value_compact(reply, doc.get("_id") orelse bson.Value.null);
var out: std.ArrayListUnmanaged(u8) = .empty;
defer out.deinit(arena);
try out.append(arena, '{');
for (ix.keys, 0..) |k, i| {
if (i > 0) try out.appendSlice(arena, ", ");
try out.appendSlice(arena, k.path);
try out.appendSlice(arena, ": ");
var values: std.ArrayListUnmanaged(bson.Value) = .empty;
defer values.deinit(arena);
try query.collect_values(arena, doc.pairs, k.path, &values, 0);
const v: bson.Value = if (values.items.len > 0) values.items[0] else .null;
try out.appendSlice(arena, try serialize_value_compact(reply, v));
}
// Index not found (defensive): fall back to the document's _id.
return serialize_value_compact(reply, doc.get("_id") orelse bson.Value.null);
try out.append(arena, '}');
return out.toOwnedSlice(arena);
}
fn duplicate_key_error(ctx: *Context, reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, doc: *const bson.Document) !void {