commands: createIndexes/listIndexes/dropIndexes + planner wiring
Adds the three driver commands, parameterized E11000 messages (engine dup_index carries the index name into writeErrors), the scan_matching planner wiring (_id fast path → index plan → scan) and the first-$match aggregate pushdown. The equivalence test (mixed-type corpus x 27 filters, non-sparse and sparse indexes) drove out three real bugs: the two-bound range under-approximation on multikey indexes (fall back to scan), a dangling single-value option array in the planner, and update-time unique violations now reporting writeErrors instead of corrupting state.
This commit is contained in:
737
src/commands.zig
737
src/commands.zig
@@ -8,6 +8,7 @@ const wire = @import("wire.zig");
|
||||
const db = @import("db.zig");
|
||||
const query = @import("query.zig");
|
||||
const update = @import("update.zig");
|
||||
const index = @import("index.zig");
|
||||
|
||||
pub const Context = struct {
|
||||
gpa: std.mem.Allocator,
|
||||
@@ -24,11 +25,13 @@ pub const ErrorCode = enum(i32) {
|
||||
bad_value = 2,
|
||||
invalid_argument = 72,
|
||||
namespace_not_found = 26,
|
||||
index_not_found = 27,
|
||||
duplicate_key = 11000,
|
||||
namespace_exists = 48,
|
||||
failed_to_parse = 9,
|
||||
internal_error = 1,
|
||||
invalid_pipeline_operator = 40324,
|
||||
index_options_conflict = 85,
|
||||
};
|
||||
|
||||
/// Which lock (if any) a command needs on the engine. Contract: only
|
||||
@@ -72,10 +75,13 @@ const command_table = [_]Command{
|
||||
.{ .name = "create", .kind = .write, .handler = cmd_create },
|
||||
.{ .name = "drop", .kind = .write, .handler = cmd_drop },
|
||||
.{ .name = "dropDatabase", .kind = .write, .handler = cmd_drop_database },
|
||||
.{ .name = "createIndexes", .kind = .write, .handler = cmd_create_indexes },
|
||||
.{ .name = "dropIndexes", .kind = .write, .handler = cmd_drop_indexes },
|
||||
.{ .name = "insert", .kind = .write, .handler = cmd_insert },
|
||||
.{ .name = "update", .kind = .write, .handler = cmd_update },
|
||||
.{ .name = "delete", .kind = .write, .handler = cmd_delete },
|
||||
.{ .name = "findAndModify", .kind = .write, .handler = cmd_find_and_modify },
|
||||
.{ .name = "listIndexes", .kind = .read, .handler = cmd_list_indexes },
|
||||
};
|
||||
|
||||
pub fn dispatch(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
|
||||
@@ -299,6 +305,195 @@ fn cmd_drop_database(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !voi
|
||||
try reply.put_ok();
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Indexes
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
fn cmd_create_indexes(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
|
||||
const db_name = msg.db_name() orelse return invalid_arg(reply, "createIndexes requires $db");
|
||||
const coll_name = str_arg(msg.body.get("createIndexes")) orelse return bad_value(reply, "createIndexes requires a collection name");
|
||||
const indexes = switch (msg.body.get("indexes") orelse return bad_value(reply, "createIndexes requires indexes")) {
|
||||
.array => |a| a,
|
||||
else => return bad_value(reply, "indexes must be an array"),
|
||||
};
|
||||
|
||||
const existed = ctx.engine.get_collection(db_name, coll_name) != null;
|
||||
const coll = try ctx.engine.get_or_create_collection(db_name, coll_name);
|
||||
const num_before: i32 = @intCast(coll.indexes.items.len + 1); // + the _id_ index
|
||||
|
||||
for (indexes) |spec_v| {
|
||||
const spec = switch (spec_v) {
|
||||
.doc => |p| p,
|
||||
else => return bad_value(reply, "indexes must be documents"),
|
||||
};
|
||||
const key_value = bson.get_pair(spec, "key") orelse return bad_value(reply, "index spec requires key");
|
||||
const key_pairs = switch (key_value) {
|
||||
.doc => |p| p,
|
||||
else => return bad_value(reply, "key must be a document"),
|
||||
};
|
||||
if (key_pairs.len == 0) return bad_value(reply, "cannot create index with an empty key");
|
||||
|
||||
// {_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) {
|
||||
const only = key_pairs[0].value;
|
||||
is_id_index = (only == .int32 and only.int32 == 1) or
|
||||
(only == .int64 and only.int64 == 1) or
|
||||
(only == .double and only.double == 1.0);
|
||||
}
|
||||
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_'");
|
||||
}
|
||||
|
||||
const spec_doc = bson.Document{ .arena = undefined, .pairs = spec };
|
||||
_ = ctx.engine.create_index(db_name, coll_name, &spec_doc) catch |err| switch (err) {
|
||||
error.InvalidIndexSpec => return bad_value(reply, "invalid index spec"),
|
||||
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) },
|
||||
);
|
||||
return reply.put_error(@intFromEnum(ErrorCode.duplicate_key), "DuplicateKey", msg_text);
|
||||
},
|
||||
error.ParallelArrays => return bad_value(reply, "cannot index parallel arrays"),
|
||||
else => return err,
|
||||
};
|
||||
}
|
||||
|
||||
try reply.put("createdCollectionAutomatically", .{ .bool = !existed });
|
||||
try reply.put("numIndexesBefore", .{ .int32 = num_before });
|
||||
try reply.put("numIndexesAfter", .{ .int32 = @intCast(coll.indexes.items.len + 1) });
|
||||
try reply.put_ok();
|
||||
}
|
||||
|
||||
fn cmd_list_indexes(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
|
||||
const db_name = msg.db_name() orelse return invalid_arg(reply, "listIndexes requires $db");
|
||||
const coll_name = str_arg(msg.body.get("listIndexes")) orelse return bad_value(reply, "listIndexes requires a collection name");
|
||||
const coll = ctx.engine.get_collection(db_name, coll_name) orelse
|
||||
return reply.put_error(@intFromEnum(ErrorCode.namespace_not_found), "NamespaceNotFound", "ns not found");
|
||||
|
||||
// The _id_ index first, then the secondaries.
|
||||
const n = coll.indexes.items.len + 1;
|
||||
const values = try reply.arena_alloc().alloc(bson.Value, n);
|
||||
const id_pairs = try reply.arena_alloc().alloc(bson.Pair, 2);
|
||||
id_pairs[0] = .{ .key = "v", .value = .{ .int32 = 2 } };
|
||||
id_pairs[1] = .{ .key = "key", .value = .{ .doc = &.{.{ .key = "_id", .value = .{ .int32 = 1 } }} } };
|
||||
values[0] = .{ .doc = try index_pairs_append(reply, id_pairs, "_id_") };
|
||||
for (coll.indexes.items, 0..) |*ix, i| {
|
||||
// The pairs live in the reply arena (freed with it); the values
|
||||
// array below references them.
|
||||
var pairs: std.ArrayListUnmanaged(bson.Pair) = .empty;
|
||||
try ix.spec_pairs(reply.arena_alloc(), &pairs);
|
||||
values[1 + i] = .{ .doc = pairs.items };
|
||||
}
|
||||
try reply.put("cursor", .{ .doc = try cursor_doc(reply, 0, try format_namespace(reply, db_name, coll_name), "firstBatch", values) });
|
||||
try reply.put_ok();
|
||||
}
|
||||
|
||||
/// The _id_ index entry: {v, key: {_id: 1}, name: "_id_"}.
|
||||
fn index_pairs_append(reply: *wire.Reply, pairs: []const bson.Pair, name: []const u8) ![]const bson.Pair {
|
||||
const arena = reply.arena_alloc();
|
||||
const with_name = try arena.alloc(bson.Pair, pairs.len + 1);
|
||||
@memcpy(with_name[0..pairs.len], pairs);
|
||||
with_name[pairs.len] = .{ .key = "name", .value = .{ .string = try arena.dupe(u8, name) } };
|
||||
return with_name;
|
||||
}
|
||||
|
||||
fn cmd_drop_indexes(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
|
||||
const db_name = msg.db_name() orelse return invalid_arg(reply, "dropIndexes requires $db");
|
||||
const coll_name = str_arg(msg.body.get("dropIndexes")) orelse return bad_value(reply, "dropIndexes requires a collection name");
|
||||
const coll = ctx.engine.get_collection(db_name, coll_name) orelse
|
||||
return reply.put_error(@intFromEnum(ErrorCode.namespace_not_found), "NamespaceNotFound", "ns not found");
|
||||
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;
|
||||
}
|
||||
} 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 {
|
||||
return reply.put_error(@intFromEnum(ErrorCode.index_not_found), "IndexNotFound", "index not found with name");
|
||||
}
|
||||
} else if (arg == .doc) {
|
||||
// A key document: drop the index with a matching key pattern.
|
||||
const key_value = bson.get_pair(arg.doc, "key") orelse bson.Value{ .doc = arg.doc };
|
||||
const key_pairs = switch (key_value) {
|
||||
.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;
|
||||
} else {
|
||||
return bad_value(reply, "dropIndexes index must be a name, key document, or '*'");
|
||||
}
|
||||
|
||||
try reply.put("nIndexesWas", .{ .int32 = n_indexes_was });
|
||||
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,
|
||||
};
|
||||
}
|
||||
|
||||
/// Render the key pattern with placeholder values for a createIndexes
|
||||
/// duplicate-key error (no specific document is involved).
|
||||
fn render_spec_key(reply: *wire.Reply, key_pairs: []const bson.Pair) ![]const u8 {
|
||||
const arena = reply.arena_alloc();
|
||||
var out: std.ArrayListUnmanaged(u8) = .empty;
|
||||
defer out.deinit(arena);
|
||||
try out.append(arena, '{');
|
||||
for (key_pairs, 0..) |p, i| {
|
||||
if (i > 0) try out.appendSlice(arena, ", ");
|
||||
try out.appendSlice(arena, p.key);
|
||||
try out.appendSlice(arena, ": ?");
|
||||
}
|
||||
try out.append(arena, '}');
|
||||
return out.toOwnedSlice(arena);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// CRUD
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -318,11 +513,11 @@ fn cmd_insert(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
|
||||
inserted += 1;
|
||||
} else |err| {
|
||||
switch (err) {
|
||||
error.DuplicateKey => {
|
||||
error.DuplicateKey, error.DuplicateKeyIndex => {
|
||||
const e = try reply.arena_alloc().alloc(bson.Pair, 3);
|
||||
e[0] = .{ .key = "index", .value = .{ .int32 = @intCast(i) } };
|
||||
e[1] = .{ .key = "code", .value = .{ .int32 = @intFromEnum(ErrorCode.duplicate_key) } };
|
||||
e[2] = .{ .key = "errmsg", .value = .{ .string = try duplicate_key_message(reply, db_name, coll_name, doc) } };
|
||||
e[2] = .{ .key = "errmsg", .value = .{ .string = try duplicate_key_message(ctx, reply, db_name, coll_name, doc) } };
|
||||
try write_errors.append(reply.arena_alloc(), .{ .doc = e });
|
||||
},
|
||||
else => return err,
|
||||
@@ -370,7 +565,13 @@ fn cmd_find(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
|
||||
/// Collect the documents in `db_name.coll_name` matching `filter`, stopping
|
||||
/// after `limit` matches (0 = unlimited). Returns the number matched; `out`
|
||||
/// may be null when only the count is wanted. Every command that scans a
|
||||
/// collection goes through here, so an index would only need one call site.
|
||||
/// collection goes through here, so an index only needs this one call site.
|
||||
///
|
||||
/// Candidate generation order: the _id_ fast path (docs map lookup), then a
|
||||
/// secondary-index plan, then a plain scan. Every candidate is re-checked
|
||||
/// with the unchanged filter, so an index that over-approximates is merely
|
||||
/// slow — never wrong. With no index created and no usable _id clause, the
|
||||
/// plain-scan path is the only one reached.
|
||||
fn scan_matching(
|
||||
ctx: *Context,
|
||||
db_name: []const u8,
|
||||
@@ -382,6 +583,51 @@ fn scan_matching(
|
||||
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;
|
||||
|
||||
// _id_ fast path: the docs map is the _id index. Skipped when the
|
||||
// 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);
|
||||
}
|
||||
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;
|
||||
if (!try query.matches(ctx.gpa, &filter_doc, doc)) continue;
|
||||
if (out) |list| try list.append(ctx.gpa, doc);
|
||||
n += 1;
|
||||
if (limit != 0 and n >= limit) break;
|
||||
}
|
||||
return n;
|
||||
}
|
||||
|
||||
// Secondary-index plan: candidates in index order, re-filtered. The
|
||||
// returned ids alias the docs map keys, valid under the read lock.
|
||||
if (try index.plan(ctx.gpa, coll.indexes.items, filter)) |p| {
|
||||
var plan = p;
|
||||
defer plan.deinit(ctx.gpa);
|
||||
var ids: std.ArrayListUnmanaged([]const u8) = .empty;
|
||||
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;
|
||||
if (!try query.matches(ctx.gpa, &filter_doc, doc)) continue;
|
||||
if (out) |list| try list.append(ctx.gpa, doc);
|
||||
n += 1;
|
||||
if (limit != 0 and n >= limit) break;
|
||||
}
|
||||
return n;
|
||||
}
|
||||
|
||||
var it = coll.docs.iterator();
|
||||
while (it.next()) |entry| {
|
||||
if (!try query.matches(ctx.gpa, &filter_doc, entry.value_ptr.*)) continue;
|
||||
@@ -421,6 +667,8 @@ fn cmd_update(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
|
||||
var n_modified: i64 = 0;
|
||||
var upserted: std.ArrayListUnmanaged(bson.Pair) = .empty;
|
||||
defer upserted.deinit(reply.arena_alloc());
|
||||
var write_errors: std.ArrayListUnmanaged(bson.Value) = .empty;
|
||||
defer write_errors.deinit(reply.arena_alloc());
|
||||
|
||||
for (specs, 0..) |*spec, si| {
|
||||
const q = doc_arg(spec.get("q")) orelse return bad_value(reply, "update spec requires q");
|
||||
@@ -436,7 +684,7 @@ fn cmd_update(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
|
||||
if (upsert) {
|
||||
const new_doc = try build_upsert_doc(reply, q, u_doc);
|
||||
ctx.engine.insert(db_name, coll_name, new_doc, ctx.oid_gen) catch |err| switch (err) {
|
||||
error.DuplicateKey => return duplicate_key_error(reply, db_name, coll_name, new_doc),
|
||||
error.DuplicateKey, error.DuplicateKeyIndex => return duplicate_key_error(ctx, reply, db_name, coll_name, new_doc),
|
||||
else => return err,
|
||||
};
|
||||
const id = new_doc.get("_id") orelse bson.Value.null;
|
||||
@@ -458,13 +706,28 @@ fn cmd_update(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
|
||||
error.ImmutableId, error.InvalidUpdate => return bad_value(reply, "bad update"),
|
||||
else => return err,
|
||||
};
|
||||
try ctx.engine.replace(db_name, coll_name, copy, ctx.oid_gen);
|
||||
ctx.engine.replace(db_name, coll_name, copy, ctx.oid_gen) catch |err| switch (err) {
|
||||
error.DuplicateKey, error.DuplicateKeyIndex => {
|
||||
const e = try reply.arena_alloc().alloc(bson.Pair, 3);
|
||||
e[0] = .{ .key = "index", .value = .{ .int32 = @intCast(si) } };
|
||||
e[1] = .{ .key = "code", .value = .{ .int32 = @intFromEnum(ErrorCode.duplicate_key) } };
|
||||
e[2] = .{ .key = "errmsg", .value = .{ .string = try duplicate_key_message(ctx, reply, db_name, coll_name, copy) } };
|
||||
try write_errors.append(reply.arena_alloc(), .{ .doc = e });
|
||||
continue;
|
||||
},
|
||||
else => return err,
|
||||
};
|
||||
n_modified += 1;
|
||||
}
|
||||
}
|
||||
|
||||
try reply.put("n", .{ .int32 = @intCast(n_matched) });
|
||||
try reply.put("nModified", .{ .int32 = @intCast(n_modified) });
|
||||
if (write_errors.items.len > 0) {
|
||||
const arr = try reply.arena_alloc().alloc(bson.Value, write_errors.items.len);
|
||||
@memcpy(arr, write_errors.items);
|
||||
try reply.put("writeErrors", .{ .array = arr });
|
||||
}
|
||||
if (upserted.items.len > 0) {
|
||||
const arr = try reply.arena_alloc().alloc(bson.Value, upserted.items.len);
|
||||
for (upserted.items, 0..) |u, i| arr[i] = u.value;
|
||||
@@ -533,7 +796,7 @@ fn cmd_find_and_modify(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !v
|
||||
const u_doc = doc_arg(msg.body.get("update")) orelse return bad_value(reply, "update must be a document");
|
||||
const new_doc = try build_upsert_doc(reply, q, u_doc);
|
||||
ctx.engine.insert(db_name, coll_name, new_doc, ctx.oid_gen) catch |err| switch (err) {
|
||||
error.DuplicateKey => return duplicate_key_error(reply, db_name, coll_name, new_doc),
|
||||
error.DuplicateKey, error.DuplicateKeyIndex => return duplicate_key_error(ctx, reply, db_name, coll_name, new_doc),
|
||||
else => return err,
|
||||
};
|
||||
n = 1;
|
||||
@@ -593,7 +856,14 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
|
||||
// stream entirely (so $sort/$limit after it apply to the groups).
|
||||
var stream: std.ArrayListUnmanaged(*const bson.Document) = .empty;
|
||||
defer stream.deinit(ctx.gpa);
|
||||
if (ctx.engine.get_collection(db_name, coll_name)) |coll| {
|
||||
// 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;
|
||||
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;
|
||||
} 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.*);
|
||||
}
|
||||
@@ -604,6 +874,10 @@ 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"),
|
||||
@@ -936,18 +1210,57 @@ fn bad_value(reply: *wire.Reply, msg: []const u8) !void {
|
||||
}
|
||||
|
||||
/// The E11000 text, shared by the top-level error reply and the per-document
|
||||
/// `writeErrors` entries of a batch insert.
|
||||
fn duplicate_key_message(reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, doc: *const bson.Document) ![]const u8 {
|
||||
const key = try serialize_value_compact(reply, doc.get("_id") orelse bson.Value.null);
|
||||
/// `writeErrors` entries of a batch insert. Uses engine.dup_index (set by a
|
||||
/// 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| {
|
||||
index_name = name;
|
||||
key_text = try render_dup_key(ctx, reply, db_name, coll_name, name, doc);
|
||||
} else {
|
||||
key_text = try serialize_value_compact(reply, doc.get("_id") orelse bson.Value.null);
|
||||
}
|
||||
return std.fmt.allocPrint(
|
||||
reply.arena_alloc(),
|
||||
"E11000 duplicate key error collection: {s}.{s} index: _id_ dup key: {s}",
|
||||
.{ db_name, coll_name, key },
|
||||
arena,
|
||||
"E11000 duplicate key error collection: {s}.{s} index: {s} dup key: {s}",
|
||||
.{ db_name, coll_name, index_name, key_text },
|
||||
);
|
||||
}
|
||||
|
||||
fn duplicate_key_error(reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, doc: *const bson.Document) !void {
|
||||
const msg_text = try duplicate_key_message(reply, db_name, coll_name, doc);
|
||||
/// 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
// Index not found (defensive): fall back to the document's _id.
|
||||
return serialize_value_compact(reply, doc.get("_id") orelse bson.Value.null);
|
||||
}
|
||||
|
||||
fn duplicate_key_error(ctx: *Context, reply: *wire.Reply, db_name: []const u8, coll_name: []const u8, doc: *const bson.Document) !void {
|
||||
const msg_text = try duplicate_key_message(ctx, reply, db_name, coll_name, doc);
|
||||
return reply.put_error(@intFromEnum(ErrorCode.duplicate_key), "DuplicateKey", msg_text);
|
||||
}
|
||||
|
||||
@@ -1223,3 +1536,397 @@ test "concurrent insert/find commands on a threaded Io" {
|
||||
try dispatch(&ctx, &msg, &reply);
|
||||
try testing.expectEqual(total, bson.get_pair(reply.pairs.items, "n").?.int32);
|
||||
}
|
||||
|
||||
// -- index command tests ----------------------------------------------------
|
||||
|
||||
/// Dispatch an insert of the given documents (each a bson.Value .doc).
|
||||
fn dispatch_insert(tdb: *TestDb, io: std.Io, coll_name: []const u8, docs: []const bson.Value) !void {
|
||||
var ctx = tdb.ctx(io);
|
||||
var msg = try parse_fake_msg("insert", .{ .string = coll_name }, &.{
|
||||
.{ .key = "documents", .value = .{ .array = docs } },
|
||||
});
|
||||
defer msg.deinit();
|
||||
var reply = wire.Reply.init(testing.allocator);
|
||||
defer reply.deinit();
|
||||
try dispatch(&ctx, &msg, &reply);
|
||||
try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply.pairs.items, "ok").?.double);
|
||||
}
|
||||
|
||||
/// Dispatch createIndexes for one spec.
|
||||
fn dispatch_create_index(tdb: *TestDb, io: std.Io, coll_name: []const u8, spec: bson.Value) !void {
|
||||
var ctx = tdb.ctx(io);
|
||||
const specs = [_]bson.Value{spec};
|
||||
var msg = try parse_fake_msg("createIndexes", .{ .string = coll_name }, &.{
|
||||
.{ .key = "indexes", .value = .{ .array = &specs } },
|
||||
});
|
||||
defer msg.deinit();
|
||||
var reply = wire.Reply.init(testing.allocator);
|
||||
defer reply.deinit();
|
||||
try dispatch(&ctx, &msg, &reply);
|
||||
try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply.pairs.items, "ok").?.double);
|
||||
}
|
||||
|
||||
/// Dispatch find and append the serialized _id of every result to `out`.
|
||||
fn dispatch_find_ids(tdb: *TestDb, io: std.Io, coll_name: []const u8, filter: []const bson.Pair, out: *std.ArrayListUnmanaged([]u8)) !void {
|
||||
var ctx = tdb.ctx(io);
|
||||
var msg = try parse_fake_msg("find", .{ .string = coll_name }, &.{
|
||||
.{ .key = "filter", .value = .{ .doc = filter } },
|
||||
});
|
||||
defer msg.deinit();
|
||||
var reply = wire.Reply.init(testing.allocator);
|
||||
defer reply.deinit();
|
||||
try dispatch(&ctx, &msg, &reply);
|
||||
try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply.pairs.items, "ok").?.double);
|
||||
const cursor = bson.get_pair(reply.pairs.items, "cursor") orelse return error.TestUnexpectedResult;
|
||||
const batch = switch (bson.get_pair(cursor.doc, "firstBatch") orelse return error.TestUnexpectedResult) {
|
||||
.array => |a| a,
|
||||
else => return error.TestUnexpectedResult,
|
||||
};
|
||||
for (batch) |d| {
|
||||
const idv = bson.get_pair(d.doc, "_id") orelse continue;
|
||||
try out.append(testing.allocator, try bson.serialize_value(testing.allocator, idv));
|
||||
}
|
||||
std.mem.sort([]u8, out.items, {}, less_u8);
|
||||
}
|
||||
|
||||
fn less_u8(_: void, a: []u8, b: []u8) bool {
|
||||
return std.mem.order(u8, a, b) == .lt;
|
||||
}
|
||||
|
||||
test "createIndexes, listIndexes, dropIndexes, and idempotent re-create" {
|
||||
var threaded: std.Io.Threaded = .init_single_threaded;
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
var tdb = try TestDb.init(io);
|
||||
defer tdb.deinit();
|
||||
|
||||
// createIndexes builds the index immediately.
|
||||
{
|
||||
var ctx = tdb.ctx(io);
|
||||
const specs = [_]bson.Value{.{ .doc = &.{
|
||||
.{ .key = "key", .value = .{ .doc = &.{.{ .key = "email", .value = .{ .int32 = 1 } }} } },
|
||||
.{ .key = "name", .value = .{ .string = "email_1" } },
|
||||
} }};
|
||||
var msg = try parse_fake_msg("createIndexes", .{ .string = "users" }, &.{
|
||||
.{ .key = "indexes", .value = .{ .array = &specs } },
|
||||
});
|
||||
defer msg.deinit();
|
||||
var reply = wire.Reply.init(testing.allocator);
|
||||
defer reply.deinit();
|
||||
try dispatch(&ctx, &msg, &reply);
|
||||
try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply.pairs.items, "ok").?.double);
|
||||
try testing.expectEqual(@as(i64, 1), bson.get_pair(reply.pairs.items, "numIndexesBefore").?.int32);
|
||||
try testing.expectEqual(@as(i64, 2), bson.get_pair(reply.pairs.items, "numIndexesAfter").?.int32);
|
||||
}
|
||||
|
||||
// Idempotent re-create of the same spec.
|
||||
try dispatch_create_index(&tdb, io, "users", .{ .doc = &.{
|
||||
.{ .key = "key", .value = .{ .doc = &.{.{ .key = "email", .value = .{ .int32 = 1 } }} } },
|
||||
.{ .key = "name", .value = .{ .string = "email_1" } },
|
||||
} });
|
||||
|
||||
// listIndexes: _id_ first, then the secondary.
|
||||
{
|
||||
var ctx = tdb.ctx(io);
|
||||
var msg = try parse_fake_msg("listIndexes", .{ .string = "users" }, &.{});
|
||||
defer msg.deinit();
|
||||
var reply = wire.Reply.init(testing.allocator);
|
||||
defer reply.deinit();
|
||||
try dispatch(&ctx, &msg, &reply);
|
||||
try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply.pairs.items, "ok").?.double);
|
||||
const cursor = bson.get_pair(reply.pairs.items, "cursor").?;
|
||||
const batch = bson.get_pair(cursor.doc, "firstBatch").?.array;
|
||||
try testing.expectEqual(@as(usize, 2), batch.len);
|
||||
try testing.expectEqualStrings("_id_", bson.get_pair(batch[0].doc, "name").?.string);
|
||||
try testing.expectEqualStrings("email_1", bson.get_pair(batch[1].doc, "name").?.string);
|
||||
const key_pairs = bson.get_pair(batch[1].doc, "key").?.doc;
|
||||
try testing.expect(bson.get_pair(key_pairs, "email") != null);
|
||||
}
|
||||
|
||||
// listIndexes on a missing collection: NamespaceNotFound.
|
||||
{
|
||||
var ctx = tdb.ctx(io);
|
||||
var msg = try parse_fake_msg("listIndexes", .{ .string = "nope" }, &.{});
|
||||
defer msg.deinit();
|
||||
var reply = wire.Reply.init(testing.allocator);
|
||||
defer reply.deinit();
|
||||
try dispatch(&ctx, &msg, &reply);
|
||||
try testing.expectEqual(@as(i32, 26), bson.get_pair(reply.pairs.items, "code").?.int32);
|
||||
}
|
||||
|
||||
// dropIndexes("*") removes the secondary but keeps _id_.
|
||||
{
|
||||
var ctx = tdb.ctx(io);
|
||||
var msg = try parse_fake_msg("dropIndexes", .{ .string = "users" }, &.{
|
||||
.{ .key = "index", .value = .{ .string = "*" } },
|
||||
});
|
||||
defer msg.deinit();
|
||||
var reply = wire.Reply.init(testing.allocator);
|
||||
defer reply.deinit();
|
||||
try dispatch(&ctx, &msg, &reply);
|
||||
try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply.pairs.items, "ok").?.double);
|
||||
try testing.expectEqual(@as(i64, 2), bson.get_pair(reply.pairs.items, "nIndexesWas").?.int32);
|
||||
}
|
||||
try testing.expectEqual(@as(usize, 0), tdb.engine.get_collection("test", "users").?.indexes.items.len);
|
||||
|
||||
// dropIndexes("_id_") errors.
|
||||
{
|
||||
var ctx = tdb.ctx(io);
|
||||
var msg = try parse_fake_msg("dropIndexes", .{ .string = "users" }, &.{
|
||||
.{ .key = "index", .value = .{ .string = "_id_" } },
|
||||
});
|
||||
defer msg.deinit();
|
||||
var reply = wire.Reply.init(testing.allocator);
|
||||
defer reply.deinit();
|
||||
try dispatch(&ctx, &msg, &reply);
|
||||
try testing.expectEqual(@as(i32, 72), bson.get_pair(reply.pairs.items, "code").?.int32);
|
||||
}
|
||||
}
|
||||
|
||||
test "unique index constraint returns 11000 through insert and update" {
|
||||
var threaded: std.Io.Threaded = .init_single_threaded;
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
var tdb = try TestDb.init(io);
|
||||
defer tdb.deinit();
|
||||
|
||||
try dispatch_create_index(&tdb, io, "users", .{ .doc = &.{
|
||||
.{ .key = "key", .value = .{ .doc = &.{.{ .key = "email", .value = .{ .int32 = 1 } }} } },
|
||||
.{ .key = "name", .value = .{ .string = "email_1" } },
|
||||
.{ .key = "unique", .value = .{ .bool = true } },
|
||||
} });
|
||||
|
||||
const docs = [_]bson.Value{
|
||||
.{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "email", .value = .{ .string = "a@x.io" } } } },
|
||||
.{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 2 } }, .{ .key = "email", .value = .{ .string = "a@x.io" } } } },
|
||||
.{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 3 } }, .{ .key = "email", .value = .{ .string = "b@x.io" } } } },
|
||||
};
|
||||
var ctx = tdb.ctx(io);
|
||||
var msg = try parse_fake_msg("insert", .{ .string = "users" }, &.{
|
||||
.{ .key = "documents", .value = .{ .array = &docs } },
|
||||
});
|
||||
defer msg.deinit();
|
||||
var reply = wire.Reply.init(testing.allocator);
|
||||
defer reply.deinit();
|
||||
try dispatch(&ctx, &msg, &reply);
|
||||
// 2 inserted, 1 writeError with code 11000 naming the index.
|
||||
try testing.expectEqual(@as(i64, 2), bson.get_pair(reply.pairs.items, "n").?.int32);
|
||||
const errors = bson.get_pair(reply.pairs.items, "writeErrors").?.array;
|
||||
try testing.expectEqual(@as(usize, 1), errors.len);
|
||||
try testing.expectEqual(@as(i64, 11000), bson.get_pair(errors[0].doc, "code").?.int32);
|
||||
const errmsg = bson.get_pair(errors[0].doc, "errmsg").?.string;
|
||||
try testing.expect(std.mem.indexOf(u8, errmsg, "email_1") != null);
|
||||
try testing.expect(std.mem.indexOf(u8, errmsg, "E11000") != null);
|
||||
|
||||
// An update that collides is reported as a writeError with code 11000.
|
||||
const updates = [_]bson.Value{.{ .doc = &.{
|
||||
.{ .key = "q", .value = .{ .doc = &.{.{ .key = "_id", .value = .{ .int32 = 3 } }} } },
|
||||
.{ .key = "u", .value = .{ .doc = &.{.{ .key = "$set", .value = .{ .doc = &.{.{ .key = "email", .value = .{ .string = "a@x.io" } }} } }} } },
|
||||
} }};
|
||||
var msg2 = try parse_fake_msg("update", .{ .string = "users" }, &.{
|
||||
.{ .key = "updates", .value = .{ .array = &updates } },
|
||||
});
|
||||
defer msg2.deinit();
|
||||
var reply2 = wire.Reply.init(testing.allocator);
|
||||
defer reply2.deinit();
|
||||
try dispatch(&ctx, &msg2, &reply2);
|
||||
try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply2.pairs.items, "ok").?.double);
|
||||
try testing.expectEqual(@as(i64, 0), bson.get_pair(reply2.pairs.items, "nModified").?.int32);
|
||||
const up_errs = bson.get_pair(reply2.pairs.items, "writeErrors").?.array;
|
||||
try testing.expectEqual(@as(usize, 1), up_errs.len);
|
||||
try testing.expectEqual(@as(i64, 11000), bson.get_pair(up_errs[0].doc, "code").?.int32);
|
||||
}
|
||||
|
||||
|
||||
/// Free a list of serialized ids (each element is gpa-owned).
|
||||
fn free_id_list(gpa: std.mem.Allocator, list: *std.ArrayListUnmanaged([]u8)) void {
|
||||
for (list.items) |id| gpa.free(id);
|
||||
list.deinit(gpa);
|
||||
}
|
||||
|
||||
/// Free the serialized ids and reset the list, keeping capacity.
|
||||
fn clear_id_list(gpa: std.mem.Allocator, list: *std.ArrayListUnmanaged([]u8)) void {
|
||||
for (list.items) |id| gpa.free(id);
|
||||
list.clearRetainingCapacity();
|
||||
}
|
||||
|
||||
test "indexed queries are equivalent to scans over a mixed corpus" {
|
||||
var threaded: std.Io.Threaded = .init_single_threaded;
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
var tdb = try TestDb.init(io);
|
||||
defer tdb.deinit();
|
||||
|
||||
// Corpus deliberately mixes numeric _id encodings (int32 1 and int64 1
|
||||
// compare equal but hash differently), arrays, nested docs, missing
|
||||
// fields, explicit nulls, and duplicate values.
|
||||
const corpus = [_]bson.Value{
|
||||
.{ .doc = &.{
|
||||
.{ .key = "_id", .value = .{ .int32 = 1 } },
|
||||
.{ .key = "a", .value = .{ .int32 = 10 } },
|
||||
.{ .key = "b", .value = .{ .string = "x" } },
|
||||
.{ .key = "tags", .value = .{ .array = &.{ .{ .string = "a" }, .{ .string = "b" } } } },
|
||||
} },
|
||||
.{ .doc = &.{
|
||||
.{ .key = "_id", .value = .{ .int64 = 1 } },
|
||||
.{ .key = "a", .value = .{ .int32 = 20 } },
|
||||
.{ .key = "b", .value = .{ .string = "y" } },
|
||||
.{ .key = "tags", .value = .{ .array = &.{} } },
|
||||
} },
|
||||
.{ .doc = &.{
|
||||
.{ .key = "_id", .value = .{ .int32 = 3 } },
|
||||
.{ .key = "a", .value = .{ .double = 30.0 } },
|
||||
.{ .key = "b", .value = .null },
|
||||
.{ .key = "tags", .value = .{ .array = &.{ .{ .string = "a" } } } },
|
||||
} },
|
||||
.{ .doc = &.{
|
||||
.{ .key = "_id", .value = .{ .string = "s4" } },
|
||||
.{ .key = "a", .value = .{ .int32 = 10 } },
|
||||
.{ .key = "b", .value = .{ .string = "z" } },
|
||||
.{ .key = "tags", .value = .{ .array = &.{ .{ .int32 = 10 }, .{ .int32 = 20 } } } },
|
||||
} },
|
||||
.{ .doc = &.{
|
||||
.{ .key = "_id", .value = .{ .int32 = 5 } },
|
||||
.{ .key = "a", .value = .null },
|
||||
.{ .key = "b", .value = .{ .string = "x" } },
|
||||
} },
|
||||
.{ .doc = &.{
|
||||
.{ .key = "_id", .value = .{ .int32 = 6 } },
|
||||
.{ .key = "a", .value = .{ .doc = &.{.{ .key = "n", .value = .{ .int32 = 1 } }} } },
|
||||
.{ .key = "b", .value = .{ .string = "w" } },
|
||||
} },
|
||||
.{ .doc = &.{
|
||||
.{ .key = "_id", .value = .{ .int32 = 7 } },
|
||||
.{ .key = "a", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 2 } } } },
|
||||
.{ .key = "b", .value = .{ .string = "q" } },
|
||||
} },
|
||||
.{ .doc = &.{
|
||||
.{ .key = "_id", .value = .{ .int32 = 8 } },
|
||||
.{ .key = "c", .value = .{ .int32 = 99 } },
|
||||
} },
|
||||
.{ .doc = &.{
|
||||
.{ .key = "_id", .value = .{ .int32 = 9 } },
|
||||
.{ .key = "a", .value = .{ .int32 = 40 } },
|
||||
.{ .key = "b", .value = .{ .array = &.{ .{ .int32 = 5 }, .{ .int32 = 6 } } } },
|
||||
} },
|
||||
};
|
||||
try dispatch_insert(&tdb, io, "eq", &corpus);
|
||||
|
||||
const filters = [_]struct { pairs: []const bson.Pair }{
|
||||
.{ .pairs = &.{} },
|
||||
.{ .pairs = &.{.{ .key = "a", .value = .{ .int32 = 10 } }} },
|
||||
.{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$eq", .value = .{ .int32 = 20 } }} } }} },
|
||||
.{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$gt", .value = .{ .int32 = 15 } }} } }} },
|
||||
.{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$gte", .value = .{ .int32 = 20 } }} } }} },
|
||||
.{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$lt", .value = .{ .int32 = 30 } }} } }} },
|
||||
.{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$lte", .value = .{ .int32 = 30 } }} } }} },
|
||||
.{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$in", .value = .{ .array = &.{ .{ .int32 = 10 }, .{ .int32 = 30 } } } }} } }} },
|
||||
.{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{ .{ .key = "$gt", .value = .{ .int32 = 5 } }, .{ .key = "$lt", .value = .{ .int32 = 25 } } } }} } },
|
||||
.{ .pairs = &.{.{ .key = "b", .value = .{ .string = "x" } }} },
|
||||
.{ .pairs = &.{.{ .key = "b", .value = .null }} },
|
||||
.{ .pairs = &.{ .{ .key = "a", .value = .{ .int32 = 10 } }, .{ .key = "b", .value = .{ .string = "x" } } } },
|
||||
.{ .pairs = &.{.{ .key = "tags", .value = .{ .string = "a" } }} },
|
||||
.{ .pairs = &.{.{ .key = "tags", .value = .{ .array = &.{ .{ .string = "a" }, .{ .string = "b" } } } }} },
|
||||
.{ .pairs = &.{.{ .key = "tags", .value = .{ .doc = &.{.{ .key = "$all", .value = .{ .array = &.{ .{ .string = "a" } } } }} } }} },
|
||||
.{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$regex", .value = .{ .string = "^1" } }} } }} },
|
||||
.{ .pairs = &.{.{ .key = "c", .value = .{ .doc = &.{.{ .key = "$exists", .value = .{ .bool = false } }} } }} },
|
||||
.{ .pairs = &.{.{ .key = "a", .value = .null }} },
|
||||
.{ .pairs = &.{.{ .key = "_id", .value = .{ .int32 = 1 } }} },
|
||||
.{ .pairs = &.{.{ .key = "_id", .value = .{ .string = "s4" } }} },
|
||||
.{ .pairs = &.{.{ .key = "_id", .value = .{ .doc = &.{.{ .key = "$in", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .string = "s4" } } } }} } }} },
|
||||
.{ .pairs = &.{.{ .key = "$and", .value = .{ .array = &.{
|
||||
.{ .doc = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$gte", .value = .{ .int32 = 10 } }} } }} },
|
||||
.{ .doc = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$lte", .value = .{ .int32 = 30 } }} } }} },
|
||||
} } }} },
|
||||
.{ .pairs = &.{.{ .key = "a", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 2 } } } }} },
|
||||
.{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$gt", .value = .{ .int32 = 100 } }} } }} },
|
||||
.{ .pairs = &.{.{ .key = "$or", .value = .{ .array = &.{
|
||||
.{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 10 } }} },
|
||||
.{ .doc = &.{.{ .key = "b", .value = .{ .string = "y" } }} },
|
||||
} } }} },
|
||||
.{ .pairs = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "$gt", .value = .null }} } }} },
|
||||
.{ .pairs = &.{ .{ .key = "a", .value = .{ .int32 = 10 } }, .{ .key = "tags", .value = .{ .int32 = 10 } } } },
|
||||
};
|
||||
|
||||
// First with a compound (a, b) index, then after dropping it — and the
|
||||
// _id fast path is exercised by the _id filters in both runs.
|
||||
try dispatch_create_index(&tdb, io, "eq", .{ .doc = &.{
|
||||
.{ .key = "key", .value = .{ .doc = &.{
|
||||
.{ .key = "a", .value = .{ .int32 = 1 } },
|
||||
.{ .key = "b", .value = .{ .int32 = 1 } },
|
||||
} } },
|
||||
.{ .key = "name", .value = .{ .string = "a_1_b_1" } },
|
||||
} });
|
||||
|
||||
// Run every filter through the index (the (a, b) plan plus the _id fast
|
||||
// path), then drop the index and require identical results from the
|
||||
// scan. The corpus's int32/int64 _id pair exercises the fast-path guard
|
||||
// (numbers fall back to a scan in both runs).
|
||||
var indexed_results: std.ArrayListUnmanaged(std.ArrayListUnmanaged([]u8)) = .empty;
|
||||
defer {
|
||||
for (indexed_results.items) |*l| free_id_list(testing.allocator, l);
|
||||
indexed_results.deinit(testing.allocator);
|
||||
}
|
||||
for (filters) |f| {
|
||||
var list: std.ArrayListUnmanaged([]u8) = .empty;
|
||||
errdefer free_id_list(testing.allocator, &list);
|
||||
try dispatch_find_ids(&tdb, io, "eq", f.pairs, &list);
|
||||
try indexed_results.append(testing.allocator, list);
|
||||
}
|
||||
{
|
||||
var ctx = tdb.ctx(io);
|
||||
var msg = try parse_fake_msg("dropIndexes", .{ .string = "eq" }, &.{
|
||||
.{ .key = "index", .value = .{ .string = "*" } },
|
||||
});
|
||||
defer msg.deinit();
|
||||
var reply = wire.Reply.init(testing.allocator);
|
||||
defer reply.deinit();
|
||||
try dispatch(&ctx, &msg, &reply);
|
||||
}
|
||||
var scanned: std.ArrayListUnmanaged([]u8) = .empty;
|
||||
defer free_id_list(testing.allocator, &scanned);
|
||||
for (filters, 0..) |f, fi| {
|
||||
clear_id_list(testing.allocator, &scanned);
|
||||
try dispatch_find_ids(&tdb, io, "eq", f.pairs, &scanned);
|
||||
const indexed = indexed_results.items[fi];
|
||||
if (scanned.items.len != indexed.items.len) std.debug.print("MISMATCH(ab) filter {d}: scan={d} idx={d}\n", .{ fi, scanned.items.len, indexed.items.len });
|
||||
try testing.expectEqual(scanned.items.len, indexed.items.len);
|
||||
for (scanned.items, indexed.items) |a, b| try testing.expectEqualSlices(u8, a, b);
|
||||
}
|
||||
|
||||
// Same corpus with a sparse (a, b) index: the sparse/null bail keeps
|
||||
// {a: null} and null-range filters correct.
|
||||
try dispatch_create_index(&tdb, io, "eq", .{ .doc = &.{
|
||||
.{ .key = "key", .value = .{ .doc = &.{
|
||||
.{ .key = "a", .value = .{ .int32 = 1 } },
|
||||
.{ .key = "b", .value = .{ .int32 = 1 } },
|
||||
} } },
|
||||
.{ .key = "name", .value = .{ .string = "a_1_b_1_sparse" } },
|
||||
.{ .key = "sparse", .value = .{ .bool = true } },
|
||||
} });
|
||||
for (indexed_results.items) |*l| free_id_list(testing.allocator, l);
|
||||
indexed_results.clearRetainingCapacity();
|
||||
for (filters) |f| {
|
||||
var list: std.ArrayListUnmanaged([]u8) = .empty;
|
||||
errdefer free_id_list(testing.allocator, &list);
|
||||
try dispatch_find_ids(&tdb, io, "eq", f.pairs, &list);
|
||||
try indexed_results.append(testing.allocator, list);
|
||||
}
|
||||
{
|
||||
var ctx = tdb.ctx(io);
|
||||
var msg = try parse_fake_msg("dropIndexes", .{ .string = "eq" }, &.{
|
||||
.{ .key = "index", .value = .{ .string = "*" } },
|
||||
});
|
||||
defer msg.deinit();
|
||||
var reply = wire.Reply.init(testing.allocator);
|
||||
defer reply.deinit();
|
||||
try dispatch(&ctx, &msg, &reply);
|
||||
}
|
||||
for (filters, 0..) |f, fi| {
|
||||
clear_id_list(testing.allocator, &scanned);
|
||||
try dispatch_find_ids(&tdb, io, "eq", f.pairs, &scanned);
|
||||
const indexed = indexed_results.items[fi];
|
||||
try testing.expectEqual(scanned.items.len, indexed.items.len);
|
||||
for (scanned.items, indexed.items) |a, b| try testing.expectEqualSlices(u8, a, b);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user