diff --git a/src/commands.zig b/src/commands.zig index a19d7fe..2cfdda0 100644 --- a/src/commands.zig +++ b/src/commands.zig @@ -62,6 +62,21 @@ pub const ErrorCode = enum(i32) { invalid_uuid = 207, idl_failed_to_parse = 40414, idl_unknown_field = 40415, + // Aggregation codes, measured against mongod 8.3.7 rather than recalled -- + // the four-and-five-digit ones are `Location` codes, which mongod names + // after the number rather than after a symbol. + // + // These are what an unimplemented construct answers with, which is the same + // choice `location_unrecognized_stage` already made for `$addFields`: this + // server reports what it does not implement using MongoDB's own code for + // "no such thing", because a code MongoDB never emits would break the + // error-code parity every milestone is held to. The message names the + // construct, so the answer is a bug report rather than a wrong number. + invalid_pipeline_operator = 168, + location_unknown_group_operator = 15952, + location_group_needs_id = 15955, + location_accumulator_not_object = 40234, + location_one_accumulator = 40238, }; /// Which lock (if any) a command needs on the engine. Contract: only @@ -2331,8 +2346,64 @@ fn count_only_pipeline(reply: *wire.Reply, stages: []const bson.Value) !?CountSh return .{ .filter = filter, .id_value = id_value, .accs = accs.items }; } -/// Minimal $group: supports `_id` of null/literal/"$field" and `$sum` -/// accumulators (constant or "$field"). +/// The whole vocabulary this engine can evaluate: a field path, or a constant. +/// +/// There is no expression evaluator (PLAN amendment A6 puts one in M2.5), and +/// until there is, anything else has to be *refused*. It used to fall through +/// to a zero: `{$avg: "$x"}` answered `0` with `ok: 1`, and so did `$max` and +/// `$push`, and a compound `_id` collapsed every document into one group keyed +/// by the unevaluated expression. A wrong number that reports success is worse +/// than an error, because nobody files it. +const GroupExpr = union(enum) { + path: []const u8, + constant: bson.Value, +}; + +/// Classify `v`, or answer the client and return null. +/// +/// `what` names the position for the message -- mongod's own messages name the +/// field, and a refusal that does not say what it refused is only half an +/// improvement on a silent zero. +fn classify_expr(reply: *wire.Reply, v: bson.Value, what: []const u8) !?GroupExpr { + const detail = switch (v) { + .string => |str| { + // "$x" is a path; "x" is the string itself. This is the only place + // that distinction is made now, where it used to be open-coded at + // each use and disagree between them. + if (str.len > 0 and str[0] == '$') return GroupExpr{ .path = str[1..] }; + return GroupExpr{ .constant = v }; + }, + // An operator document is the one shape mongod also refuses, and 168 is + // the code it uses, naming the operator. A compound expression + // (`{a: "$x"}`) mongod would happily evaluate -- so the code is the + // same and the message says what is actually true here instead. + .doc => |d| if (d.len > 0 and d[0].key.len > 0 and d[0].key[0] == '$') + try std.fmt.allocPrint(reply.arena_alloc(), "Unrecognized expression '{s}'", .{d[0].key}) + else + try std.fmt.allocPrint( + reply.arena_alloc(), + "{s} must be a field path or a constant: this server evaluates no expressions", + .{what}, + ), + .array => try std.fmt.allocPrint( + reply.arena_alloc(), + "{s} must be a field path or a constant: this server evaluates no expressions", + .{what}, + ), + else => return GroupExpr{ .constant = v }, + }; + try reply.put_error(@intFromEnum(ErrorCode.invalid_pipeline_operator), "InvalidPipelineOperator", detail); + return null; +} + +/// One output field of a `$group`, with its argument already classified. +const Accumulator = struct { + key: []const u8, + arg: GroupExpr, +}; + +/// Minimal $group: `_id` of a constant or "$field", and `$sum` accumulators +/// over the same. Everything else is refused rather than answered. fn run_group( ctx: *Context, reply: *wire.Reply, @@ -2342,15 +2413,58 @@ fn run_group( ) !?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"); + try reply.put_error( + @intFromEnum(ErrorCode.location_group_needs_id), + "Location15955", + "a group specification must include an _id", + ); return null; }; + const id_class = (try classify_expr(reply, id_expr, "the _id of a $group")) orelse return null; - var accs: std.ArrayListUnmanaged(bson.Pair) = .empty; + // Every accumulator is validated before a single document is read, so a + // pipeline that cannot be answered is refused rather than half-answered. + var accs: std.ArrayListUnmanaged(Accumulator) = .empty; defer accs.deinit(arena); for (group_pairs) |p| { if (std.mem.eql(u8, p.key, "_id")) continue; - try accs.append(arena, p); + const spec = switch (p.value) { + .doc => |d| d, + else => { + const detail = try std.fmt.allocPrint( + arena, + "The field '{s}' must be an accumulator object", + .{p.key}, + ); + try reply.put_error(@intFromEnum(ErrorCode.location_accumulator_not_object), "Location40234", detail); + return null; + }, + }; + if (spec.len != 1) { + const detail = try std.fmt.allocPrint( + arena, + "The field '{s}' must specify one accumulator", + .{p.key}, + ); + try reply.put_error(@intFromEnum(ErrorCode.location_one_accumulator), "Location40238", detail); + return null; + } + if (!std.mem.eql(u8, spec[0].key, "$sum")) { + // `$avg`, `$max`, `$push` and the rest are real MongoDB operators + // that this server does not implement; see the note on the error + // codes for why they are reported as unknown rather than with an + // invented code. They used to answer `0`. + const detail = try std.fmt.allocPrint( + arena, + "unknown group operator '{s}'", + .{spec[0].key}, + ); + try reply.put_error(@intFromEnum(ErrorCode.location_unknown_group_operator), "Location15952", detail); + return null; + } + const what = try std.fmt.allocPrint(arena, "the argument of '{s}'", .{p.key}); + const arg = (try classify_expr(reply, spec[0].value, what)) orelse return null; + try accs.append(arena, .{ .key = p.key, .arg = arg }); } const Group = struct { @@ -2373,9 +2487,9 @@ fn run_group( 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] == '$') (try query_path_value_bytes(walk_arena.allocator(), doc, s[1..])) orelse .null else id_expr, - else => id_expr, + const id_value: bson.Value = switch (id_class) { + .path => |path| (try query_path_value_bytes(walk_arena.allocator(), doc, path)) orelse .null, + .constant => |v| v, }; id_key_buf.clearRetainingCapacity(); try bson.write_value(id_value, ctx.gpa, &id_key_buf); @@ -2392,29 +2506,19 @@ fn run_group( gop.value_ptr.* = .{ .id_value = id_value, .sums = sums }; } for (accs.items, 0..) |acc, i| { - var expr = acc.value; - // Unwrap {$sum: } accumulator documents. - if (expr == .doc) { - if (bson.get_pair(expr.doc, "$sum")) |inner| { - expr = inner; - } else continue; - } - const term: f64 = switch (expr) { + // A `$sum` over something that is not a number contributes nothing, + // which is MongoDB's rule and not a stand-in for an unimplemented + // one: `{$sum: "$name"}` over strings really is zero. + const v: bson.Value = switch (acc.arg) { + .path => |path| (try query_path_value_bytes(walk_arena.allocator(), doc, path)) orelse .null, + .constant => |c| c, + }; + gop.value_ptr.sums[i] += switch (v) { .int32 => |n| @floatFromInt(n), .int64 => |n| @floatFromInt(n), .double => |n| n, - .string => |s| if (s.len > 0 and s[0] == '$') - 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, - else => 0, - } - else - 0, else => 0, }; - gop.value_ptr.sums[i] += term; } } @@ -3872,6 +3976,147 @@ test "aggregate $sort without a preceding $group sorts and frees correctly" { } } +test "$group refuses what it cannot compute instead of answering zero" { + // The failure this closes was not a missing feature, it was a wrong number + // reported as success. `{$avg: "$x"}` answered `0`, and so did `$max` and + // `$push`; a compound `_id` collapsed every document into one group keyed + // by the unevaluated expression; `{$literal: 1}` came back echoed. Six of + // eight probed pipelines answered `ok: 1` with a wrong result. An + // unrecognised stage is a bug report; an `$avg` that returns `0` is a + // corrupted report nobody files. + // + // Every code and codeName below was read off mongod 8.3.7, not recalled, + // and this test is where they are pinned. + // + // Mutation check: drop any one arm of the validation in `run_group` and + // the matching row answers `ok: 1` again. + var threaded = std.Io.Threaded.init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var tdb = try TestDb.init(io); + defer tdb.deinit(); + try dispatch_insert(&tdb, io, "g", &.{ + .{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "k", .value = .{ .string = "a" } }, .{ .key = "x", .value = .{ .int32 = 10 } } } }, + .{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 2 } }, .{ .key = "k", .value = .{ .string = "a" } }, .{ .key = "x", .value = .{ .int32 = 20 } } } }, + }); + + const Case = struct { name: []const u8, group: []const bson.Pair, code: i32 }; + const path_x = bson.Value{ .string = "$x" }; + const sum_one = bson.Value{ .doc = &.{.{ .key = "$sum", .value = .{ .int32 = 1 } }} }; + const cases = [_]Case{ + .{ + .name = "$avg", + .group = &.{ + .{ .key = "_id", .value = .{ .string = "$k" } }, + .{ .key = "v", .value = .{ .doc = &.{.{ .key = "$avg", .value = path_x }} } }, + }, + .code = 15952, + }, + .{ + .name = "$push", + .group = &.{ + .{ .key = "_id", .value = .{ .string = "$k" } }, + .{ .key = "v", .value = .{ .doc = &.{.{ .key = "$push", .value = path_x }} } }, + }, + .code = 15952, + }, + .{ + .name = "an accumulator that is not a document", + .group = &.{ + .{ .key = "_id", .value = .{ .string = "$k" } }, + .{ .key = "v", .value = path_x }, + }, + .code = 40234, + }, + .{ + .name = "two operators in one accumulator", + .group = &.{ + .{ .key = "_id", .value = .{ .string = "$k" } }, + .{ .key = "v", .value = .{ .doc = &.{ + .{ .key = "$sum", .value = path_x }, + .{ .key = "$max", .value = path_x }, + } } }, + }, + .code = 40238, + }, + .{ + .name = "no _id", + .group = &.{.{ .key = "v", .value = sum_one }}, + .code = 15955, + }, + .{ + .name = "a compound _id", + .group = &.{ + .{ .key = "_id", .value = .{ .doc = &.{.{ .key = "k", .value = .{ .string = "$k" } }} } }, + .{ .key = "v", .value = sum_one }, + }, + .code = 168, + }, + .{ + .name = "an operator expression in _id", + .group = &.{ + .{ .key = "_id", .value = .{ .doc = &.{.{ .key = "$literal", .value = .{ .int32 = 1 } }} } }, + .{ .key = "v", .value = sum_one }, + }, + .code = 168, + }, + .{ + .name = "an expression argument to $sum", + .group = &.{ + .{ .key = "_id", .value = .null }, + .{ .key = "v", .value = .{ .doc = &.{.{ .key = "$sum", .value = .{ .doc = &.{ + .{ .key = "$multiply", .value = .{ .array = &.{ path_x, .{ .int32 = 2 } } } }, + } } }} } }, + }, + .code = 168, + }, + }; + + var ctx = tdb.ctx(io); + for (cases) |c| { + const stages = [_]bson.Value{.{ .doc = &.{.{ .key = "$group", .value = .{ .doc = c.group } }} }}; + var msg = try parse_fake_msg("aggregate", .{ .string = "g" }, &.{ + .{ .key = "pipeline", .value = .{ .array = &stages } }, + .{ .key = "cursor", .value = .{ .doc = &.{} } }, + }); + defer msg.deinit(); + var reply = wire.Reply.init(testing.allocator); + defer reply.deinit(); + try dispatch(&ctx, &msg, &reply); + const ok = bson.get_pair(reply.pairs.items, "ok").?.double; + testing.expectEqual(@as(f64, 0.0), ok) catch |err| { + std.debug.print(" {s}: answered ok:1\n", .{c.name}); + return err; + }; + const code = bson.get_pair(reply.pairs.items, "code").?.int32; + testing.expectEqual(c.code, code) catch |err| { + std.debug.print(" {s}: code {d}, wanted {d}\n", .{ c.name, code, c.code }); + return err; + }; + } + + // And the one shape that *is* implemented still answers, so the refusals + // above are a fence and not a wall. + const good = [_]bson.Value{.{ .doc = &.{.{ .key = "$group", .value = .{ .doc = &.{ + .{ .key = "_id", .value = .{ .string = "$k" } }, + .{ .key = "v", .value = .{ .doc = &.{.{ .key = "$sum", .value = path_x }} } }, + } } }} }}; + var msg = try parse_fake_msg("aggregate", .{ .string = "g" }, &.{ + .{ .key = "pipeline", .value = .{ .array = &good } }, + .{ .key = "cursor", .value = .{ .doc = &.{} } }, + }); + 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 cur = bson.get_pair(reply.pairs.items, "cursor").?; + const batch = bson.get_pair(cur.doc, "firstBatch").?.array; + try testing.expectEqual(@as(usize, 1), batch.len); + try testing.expectEqual(@as(i32, 30), bson.get_pair(batch[0].doc, "v").?.int32); +} + test "count_only_pipeline accepts only shapes a count can answer" { // The fast path skips materializing documents, so mis-accepting a // pipeline would silently return a wrong aggregate rather than a slow