diff --git a/PLAN.md b/PLAN.md index f0bcc77..2959b07 100644 --- a/PLAN.md +++ b/PLAN.md @@ -394,6 +394,69 @@ per-collection gate judges what reclamation left. Same six rounds, 256 MB. Numbers and reproduction in `tests/e2e/results/m0-gates.txt` under `[M1.1]` and `[M1.2]`. +### Amendment A6 — M2 splits in two, and its gate did not exist (amends the M2 row) + +The M2 design review (`docs/M2_DESIGN_REVIEW.md`) found that the milestone's +gate names a corpus that is not there. `mongodb/specifications` has no +aggregation suite: the thirteen `aggregate-*.json` files this project runs live +inside `crud` and test the aggregate *command* — cursor and `batchSize`, +`readConcern` routing, the write semantics of `$out`/`$merge`, `collation`, +`let`. They touch stages barely and expressions not at all; `$lookup`, +`$unwind`, `$facet`, `$addFields` and `$replaceRoot` appear nowhere in the +pinned corpus. A literal "green" is unreachable besides: all 19 skips are +version- or topology-gated and none can pass on a standalone, so the gate is +stated as *0 fail* rather than *all pass*. + +So one milestone name was covering two milestones, and they are split. + +**M2 is the command surface**, gated on the corpus that already exists: `$out`, +`$merge`, `db.aggregate()`, collation, `let`. That is 8 of the 13 current +failures in the write stages alone. It is small and fully measurable today. + +**M2.5 is the engine**, gated on a corpus this project has to write, in the +unified format the runner already reads, with every expectation measured +against mongod 8.3.7 rather than recalled — the discipline that corrected three +assumptions in M1's session work. The tiers, in dependency order: the +expression evaluator, then a per-stage document iterator (the stream is a +materialized window today, which cannot express 1→N), then the accumulators. +`$unwind` is in the first cut because it falls out of the iterator for free. +`$lookup` is out: a stage that takes a second collection's lock while the +pipeline holds one is a lock-order question that should not ride along. +`$facet` is out: sub-pipeline execution should be allowed to settle after the +iterator exists. + +**Sequencing: command surface first, engine second** — chosen deliberately, and +with the cost stated rather than glossed. The engine is what gives wrong +answers *now*: measured against a live server, `$avg`, `$max` and `$push` all +return `0`, a compound `_id` collapses every document into one group keyed by +the unevaluated expression, `$literal` is echoed back, and a `$match` after a +`$project` still sees the projected-away field. Six of eight probed pipelines +answer `ok: 1` with a wrong result; only `$addFields` fails honestly. Doing the +command surface first leaves those alive for a further milestone. + +*Measured after Tier 0 landed, and it corrects this amendment:* pricing the 13 +failures one by one shows **6 of them cannot turn green in M2 at all** — three +need `$listLocalSessions` (M4) *and* `$addFields` (M2.5), two need the +expression engine, one needs a collation. The gate reads "0 fail among the +seven reachable", not "0 fail". And the seven that remain, `$out` and `$merge`, +need a lock-model decision this amendment did not anticipate: they write to a +collection the pipeline is not reading, which the dispatch contract, the static +lock table and `Collection.lock`'s one-at-a-time invariant all stand against. +It is the same objection that kept `$lookup` out of M2.5's first cut, and the +review failed to apply it to the write stages. Both options are set out in +`docs/M2_DESIGN_REVIEW.md` §7; nothing past Tier 0 is implemented until one is +chosen. + +**Which is why M2 carries the refusals.** Every construct the engine does not +implement stops answering `0` and starts answering an error, with the code +measured against mongod: unknown accumulators, non-path expressions where a +path is expected, `$project` anywhere but last. This is the same move +`expectEvents` made before the free list in M1 — it will push the scorecard +*down*, and that is the point. An unrecognised stage is a bug report; an `$avg` +that returns `0` is a corrupted report nobody files. Without this the sequencing +above would be trading a measurable milestone for a year of silent wrong +answers, and the trade is only acceptable because the lie is removed first. + --- ## 3. Milestones and gates @@ -402,7 +465,8 @@ Numbers and reproduction in `tests/e2e/results/m0-gates.txt` under `[M1.1]` and |---|---|---|---| | M0 | **mmap + WAL foundation** | data file format, page/extent allocator, mmap slab + B+tree arena, copy-on-write + page free list (A1/A2), watermark/replay, checkpoint (= compaction repurposed), leaf payload → slab offset (A3), drop docs hashmap, churn measurement | D7 (6 items) | | M1 | **Cursors + wire polish** | getMore / killCursors / batchSize; server-side cursor state with idle timeout; sessions plumbing (lsid accepted) as drivers send it; hello advertisement updates; **`moreToCome` on requests** (see the bug below); command-monitoring assertions in the spec runner | crud spec suite green; e2e green | -| M2 | **Aggregation expansion** | pipeline stages and expression engine (tiered scope defined at M2 design review) | aggregate spec suite green | +| M2 | **The `aggregate` command surface** | `$out` and `$merge` (7 of the 13 failures), and refusing every pipeline construct the engine does not implement instead of answering `0` (amendment A6). The other 6 failures are blocked on M2.5, M4 and M8 — see `docs/M2_DESIGN_REVIEW.md` §7 | `aggregate-*.json`: 0 fail among the 7 reachable cases | +| M2.5 | **The aggregation engine** | expression evaluator, per-stage document iterator, the accumulators, `$unwind`; `$lookup`/`$facet` explicitly out of the first cut (amendment A6) | a purpose-built stage corpus, every expectation measured against mongod | | M3 | **Update operators + index types** | $setOnInsert, $addToSet, $mul, $min/$max, $pop, $pullAll, $currentDate, pipeline updates; partial + hashed indexes | remaining crud coverage; e2e3/e2e4 green | | M4 | **Sessions + transactions** | logical sessions, snapshot isolation on the mmap engine, write concern at commit | sessions + transactions spec suites green | | M5 | **Change streams** | change feed + resume tokens (likely log-seq based), getMore integration | change-streams spec suite green | @@ -969,8 +1033,13 @@ has to be its own commit with its own re-recorded scorecard. taking a `session` argument, and no `lsid` assertion. The value here is protocol hygiene, not a number — and after Stage 1 a wrongly-refused command would have shown up as a changed event stream rather than silently. -- **M2 aggregation**: stage/expression tiers, which spec-test files are - the gate, whether $lookup/$unwind/facet make the first cut. +- **M2 aggregation** — *settled, see amendment A6 and `docs/M2_DESIGN_REVIEW.md`.* + The three questions this entry held are answered: the tiers, the gate (the + named one does not exist), and the first cut ($unwind in, $lookup and $facet + out). What the review did *not* settle and the engine milestone must: + `$out`/`$merge` durability semantics, whether the expression evaluator is + shared with M3's pipeline updates, and whether `allowDiskUse` has to stop + being a lie. - **M4 transactions**: snapshot isolation over mmap (COW vs undo), read concern snapshot, conflict → TransientTransactionError semantics, retryable-writes interplay. diff --git a/docs/M2_DESIGN_REVIEW.md b/docs/M2_DESIGN_REVIEW.md new file mode 100644 index 0000000..9be3cca --- /dev/null +++ b/docs/M2_DESIGN_REVIEW.md @@ -0,0 +1,359 @@ +# M2 design review — aggregation expansion + +PLAN §3 gives M2 one line of scope ("pipeline stages and expression engine, +tiered scope defined at M2 design review") and one line of gate ("aggregate +spec suite green"), and §6 lists the three questions it deferred: the +stage/expression tiers, which spec files are the gate, and whether +`$lookup`/`$unwind`/`$facet` make the first cut. + +This document answers them, and reports one finding that has to be settled +before any of the rest is worth discussing. + +Everything below is measured against the M1 tip (`a748a3d`) unless it says +otherwise. Nothing here is recalled. + +--- + +## 1. The gate named in the plan does not exist + +`mongodb/specifications` has no aggregation suite. The full list of `source/` +directories at the pinned commit is: `auth`, `bson-*`, `causal-consistency`, +`change-streams`, `client-side-encryption`, `collation`, +`collection-management`, `command-logging-and-monitoring`, `compression`, +`connection-*`, `crud`, `gridfs`, `index-management`, `load-balancers`, +`logging`, `read-write-concern`, `retryable-*`, `run-command`, `sessions`, +`server-discovery-and-monitoring`, `server-selection`, `transactions`, +`unified-test-format`, `versioned-api`, and a handful of others. There is no +`aggregation`. + +The thirteen `aggregate-*.json` files we do run live inside `crud`, and they +test the aggregate **command surface** — cursor and `batchSize`, `readConcern` +and `readPreference` routing, the write semantics of `$out`/`$merge`, +`collation`, `let`, `allowDiskUse`, `rawData`. They barely touch stages, and +they touch expressions not at all: `$lookup`, `$unwind`, `$facet`, +`$addFields`, `$replaceRoot` do not appear anywhere in the pinned corpus. + +So "aggregate spec suite green" is not a gate for the milestone the scope line +describes. It is a gate for a much smaller milestone that happens to share the +command name. **Whichever option §4 picks, PLAN's M2 gate line has to be +rewritten**, and that is the first decision this review asks for. + +Where the current corpus stands, for reference: + +``` +aggregate-*.json overall 9 pass 13 fail 19 skip +``` + +| failure | cases | +|---|---| +| `$merge` unimplemented | 5 | +| `$out` unimplemented | 3 | +| `db.aggregate()` — `{aggregate: 1}`, no collection name | 3 | +| `collation` in aggregate | 1 | +| `let` | 1 | + +All 19 skips are environmental -- 10 distinct requirements, some covering a +whole file at once (`needs topology replicaset`, `needs server >= 5.0`, `needs +server >= 8.2.0`, `needs server <= 4.2.99`). Not one can turn green on a +standalone, so a literal reading of "green" is unreachable for that reason +alone, and the gate has to be stated as *0 fail* rather than *all pass*. + +--- + +## 2. Where the implementation actually is + +**Seven stages**, all in one `if/else if` chain in `cmd_aggregate`: +`$match`, `$sort`, `$skip`, `$limit`, `$project`, `$group`, `$count`. +Anything else answers `Location40324`, with the code verified against a real +mongod. + +(`$set` and `$unset` exist in the tree as **update operators** in +`src/update.zig`. They are not pipeline stages. An earlier count of nine +stages in this project's notes was wrong.) + +**There is no expression engine.** `$field` path resolution is open-coded in +two places inside `run_group`, and nowhere else. `$sum` is the only +accumulator. + +That last sentence is the important one, because of *how* the missing pieces +fail. Probed against a live server on three documents +(`{g:'a',x:10}, {g:'a',x:20}, {g:'b',x:30}`): + +| pipeline | answer | correct answer | +|---|---|---| +| `$group: {_id:'$g', s:{$sum:'$x'}}` | `[{a,30},{b,30}]` | ✅ | +| `$group: {_id:'$g', a:{$avg:'$x'}}` | `[{a,0},{b,0}]` | `[{a,15},{b,30}]` | +| `$group: {_id:'$g', m:{$max:'$x'}}` | `[{a,0},{b,0}]` | `[{a,20},{b,30}]` | +| `$group: {_id:'$g', p:{$push:'$x'}}` | `[{a,0},{b,0}]` | arrays | +| `$group: {_id:{g:'$g'}, n:{$sum:1}}` | one group, `_id` = `{"g":"$g"}` | two groups | +| `$project:{g:1}` then `$match:{x:10}` | matches | matches nothing | +| `$group: {_id:{$literal:1}, ...}` | `_id` = `{"$literal":1}` | `_id` = `1` | +| `$addFields` | `Location40324` | (unimplemented) | + +**Six of those eight are silent wrong answers with `ok: 1`.** Only +`$addFields` fails honestly. That is the shape of the risk in this area and it +should drive the milestone: an unrecognised stage is a bug report, an `$avg` +that returns `0` is a corrupted report that nobody files. It is also precisely +the class the M1 spec-runner work was about — and the corpus cannot see any of +it, because the corpus has no stage tests. + +**The pipeline model will not carry the milestone as it stands.** The stream is +a materialized window `[start, end)` over either slab offsets or generated +documents, and each stage moves the bounds. That works for the filtering and +reordering stages and nothing else: + +- `$project` is not a stage transform at all. It sets a variable that is + applied once, at emit — so only the last `$project` in a pipeline has any + effect, and a `$match` after a `$project` sees the unprojected document. +- `$unwind` is 1→N. There is no way to express that in a window over the input. +- `$lookup` needs a second collection under a second lock, inside a stage. +- `$facet` needs the same stream fanned into several independent sub-pipelines. +- `allowDiskUse` is accepted and ignored. `$group` and `$sort` are unbounded in + memory; the three passing `allowdiskuse` cases pass because the option is + parsed, not because anything spills. `src/spill.zig` exists and is used by + index builds, not here. + +So M2 is not "add stages to the chain". It is: build an expression evaluator, +change the stream model to a per-stage document iterator, and then stages +become cheap. Anything that adds stages before those two exist adds more +silent wrong answers. + +--- + +## 3. What that implies for tiers + +The dependency order is forced, not chosen: + +**Tier 0 — stop lying (no new features).** Refuse what is not implemented +instead of answering `0`: unknown accumulators, non-path expressions where a +path is expected, `$project` used anywhere but last. Every one of these becomes +an error with a measured mongod code. This is small, and it converts six silent +wrong answers into six honest ones. It is also the only tier that makes the +later ones measurable, for the same reason Stage 1 preceded Stage 3 in M1. + +**Tier 1 — the expression engine.** A real evaluator over `bson.Value`: +field paths, `$literal`, comparison and boolean operators, arithmetic, +`$cond`/`$ifNull`/`$switch`, string and date operators. Everything above +depends on it, including `$project`'s computed fields and `$group`'s `_id`. + +**Tier 2 — the stage iterator.** Replace the window with a per-stage pull +model, so a stage can emit more or fewer documents than it consumed. `$project` +becomes a real transform. Carries `$addFields`/`$set`, `$unset`, +`$replaceRoot`, `$unwind`, `$sortByCount`. + +**Tier 3 — the accumulators.** `$avg`, `$min`, `$max`, `$first`, `$last`, +`$push`, `$addToSet`, `$count`, `$mergeObjects`. Cheap once Tier 1 exists. + +**Tier 4 — the expensive stages.** `$lookup` (second collection, lock order), +`$facet` (sub-pipelines), `$unionWith`, `$graphLookup`. Each is a design +question of its own. + +**Tier 5 — the write stages.** `$out`, `$merge`. These are not aggregation +work at all — they are a write path that happens to be spelled as a stage, with +their own durability and atomicity questions. They are also 8 of the 13 current +`aggregate-*` failures. + +**Orthogonal, and cheap:** `db.aggregate()` (`{aggregate: 1}`), 3 failures. +**Orthogonal, and not cheap:** `allowDiskUse` meaning something, which is a +memory-bound question for `$group`/`$sort` and belongs with whichever tier +first makes those able to hold a lot. + +--- + +## 4. Gate options + +Each option is stated with what it would actually prove and what it costs. + +### Option A — "the aggregate command surface" + +Gate: the 13 `aggregate-*.json` files in the crud corpus, minus the 13 +environmental skips. Scope: Tier 5 + `db.aggregate()` + collation + `let`. + +- **Proves:** the aggregate *command* behaves — cursors, read concern, write + stages, database-level form. +- **Does not prove:** any stage or expression is correct. All six silent wrong + answers survive this gate untouched. +- **Cost:** small. Two write stages and three small fixes. +- **Honest name:** this is not "aggregation expansion". If chosen, M2 should be + renamed and the aggregation engine deferred to its own milestone. + +### Option B — a hand-built stage corpus, modelled on mongod's own + +Gate: a new `tests/spec/aggregate/` corpus in the unified format the runner +already reads, with cases ported from `mongod`'s `jstests/aggregation` and +every expected result **measured against mongod 8.3.7**, the way the cursor and +`lsid` error codes were. + +- **Proves:** exactly what the milestone claims to build, at whatever tier + depth the corpus is written to. +- **Cost:** the corpus is the milestone's second deliverable and a real one — + but the runner, the entity model, the matchers and `expectEvents` all exist + already, so the marginal cost is writing cases, not machinery. +- **Risk, stated plainly:** a corpus we author is a corpus we can accidentally + write to match our own bugs. The mitigation is the one this project already + uses: every expectation measured against a real mongod before it is + committed, never recalled. + +### Option C — differential against mongod + +Gate: a harness that runs the same pipelines against MultiforaDB and mongod +8.3.7 and diffs the results, over generated pipelines and a fixed corpus. + +- **Proves:** the most, by a distance, and it finds the silent-wrong-answer + class by construction rather than by someone thinking to test for it. +- **Cost:** mongod becomes a test dependency (it already is one for + measurement, but not for the gate). Result ordering and floating-point + formatting need normalising, and `$group` output order is genuinely + unspecified. +- **Note:** this is a *harness*, not a corpus, so it composes with B rather + than competing with it — B gives regression cases with committed + expectations, C finds new ones. + +### Option D — keep the plan's line and reinterpret "green" + +Gate: `aggregate-*.json` at 22 pass / 0 fail / 13 skip, and declare the skips +out of scope. + +- **Proves:** the same as A. +- **Why it is listed:** only to be rejected explicitly. It reads as if the + aggregation milestone were gated on aggregation tests, and it is not. If A is + the decision, say A. + +--- + +## 5. Decision + +**Taken: the command surface first (Option A as M2), the engine second (M2.5), +with Tier 0 folded into M2.** Recorded as PLAN amendment A6; the §3 rows are +rewritten and §6's deferred entry now points here. + +The cost of that order is on the record: the engine is what gives wrong answers +today, and this leaves them alive for a further milestone. Tier 0 is what makes +the trade acceptable rather than merely convenient — every unimplemented +construct stops answering `0` and starts answering an error, so the wrong +answers become visible bug reports instead of quiet ones while the command +surface is built. It is not optional under this sequencing. + +The recommendation the review arrived at independently is kept below, since a +decision is easier to revisit when the argument it overrode is still legible. + +## 5.1 Recommendation as written before the decision + +**Tier 0 first, alone, before any scoping decision is final** — the same +argument that put `expectEvents` before the free list in M1. Six silent wrong +answers are six results nobody can measure past, and turning them into errors +is a day's work that changes what every later number means. It will move the +scorecard *down*, and that is the point. + +**Then B as the gate, with C as the harness that feeds it.** Option A is a real +piece of work and should happen — the `$merge`/`$out` failures are 8 of 13 — +but it should be named for what it is and probably belongs in M6 (admin/ops) or +its own small milestone, not as the gate for an engine. + +**Tiers 1→3 as M2 proper**, with Tier 4 explicitly out of the first cut. That +answers §6's third question: **`$unwind` yes** (it falls out of Tier 2 for +free), **`$lookup` and `$facet` no** — `$lookup` because a stage that takes a +second collection lock while the pipeline holds one is a lock-order design +question this milestone should not be carrying, and `$facet` because it needs +sub-pipeline execution that Tier 2's iterator should be allowed to settle +first. + +**One thing to decide that this review cannot decide alone:** whether +`allowDiskUse` must mean something by the end of M2. It is currently accepted +and ignored, which is the same category of lie as `$avg` returning `0` — a +client that passes it believes it is protected. If the answer is yes, the +memory bound on `$group`/`$sort` joins Tier 2 rather than trailing it. + +--- + +## 6. What this review does not cover + +- `$out`/`$merge` durability semantics (atomic swap? logged? crash behaviour?). + Needed before Tier 5 is scoped. +- Whether the expression engine is shared with `update`'s pipeline-update form, + which M3 will need. Probably yes; it changes where the code lives. +- Index use inside a pipeline beyond the existing leading-`$match` pushdown. + +--- + +## 7. Post-decision measurement: the gate is not reachable either + +Written after Tier 0 landed, when the 13 failures were priced one by one +instead of counted. Section 1 established that the *corpus* named in the plan +was the wrong instrument; this establishes that the *target* set on the +corpus we kept is unreachable too. Both were found by looking rather than by +assuming, and this one only by opening each case. + +| cases | what they need | milestone | +|---|---|---| +| 5 `aggregate-merge.json` | `$merge` with a plain `into` | M2, blocked below | +| 2 `aggregate-out.json` | `$out` with a plain target | M2, blocked below | +| 3 `db-aggregate*.json` | `$listLocalSessions` **and** `$addFields` | M4 + M2.5 | +| 2 `aggregate-let.json` | `let`, `$$var`, `$expr` | M2.5 | +| 1 `aggregate-collation.json` | a case-insensitive collation | M8 | + +So **6 of 13 can never turn green in M2**, whatever is built, and the gate has +to read *0 fail among the seven this milestone can reach*, with the other six +named and attributed. Note what the three `db.aggregate()` cases really need: +implementing `{aggregate: 1}` moves none of them, because each then fails on +`$listLocalSessions` instead. The design review priced that line "orthogonal, +and cheap"; it was cheap and it was worth nothing. + +### And the seven need a decision this review has no authority to take + +`$out` and `$merge` write to a collection the pipeline is not reading. Three +things in the current design stand against that, and none of them is a detail: + +- `aggregate` is declared `.kind = .read`, and the dispatch contract is + explicit that "only `.write` commands may call engine mutation functions". +- dispatch acquires locks from a **static table keyed on the command name**, + before the handler runs, for the one collection named by the command. +- `Collection.lock`'s own comment states the invariant that decides this: + *never more than one collection lock at a time*. Taking the target's + exclusive lock while holding the source's shared one breaks it directly. + +This is the same shape as the objection that kept `$lookup` out of M2.5's first +cut — "a stage that takes a second collection's lock while the pipeline holds +one is a lock-order question that should not ride along" — and the review +should have applied it to the write stages in the same breath. It did not, +which is the review's own error and is recorded here rather than quietly +corrected. + +Two ways out, both real: + +**(a) A pipeline that ends in a write stage is a write command.** Compute the +effective lock shape from the pipeline rather than the name: catalog exclusive, +no collection lock from dispatch, and the handler takes source-shared → +release → target-exclusive in that order. Heavier than mongod, and it +serializes such an aggregate against the whole catalog, but it keeps every +existing invariant intact and is honest about what the command is. + +**(b) The write stages run in the dispatch epilogue,** after every lock is +released, next to the commit and the checkpoint that already live there. The +handler stashes what to write. Cheaper and better-behaved under concurrency, +but it amends the `.read`/`.write` contract, so the contract's comment has to +change with it and say why. + +Either way the durability question is open and belongs with whichever is +chosen: mongod's `$out` replaces the target collection *atomically*, and this +engine has no cross-collection atomicity. A pipeline that fails after writing +half its output must not leave the target half-replaced. + +**Decision: (b), the epilogue.** Taken after the options were set out. The +handler computes the output under the locks it already holds and leaves it in +`Context.pending_write`; the epilogue applies it once every lock is released, +beside the commit and the checkpoint that already live there. The +`.read`/`.write` contract's comment was amended to say so and why. + +Measured after: `aggregate-*.json` went 9 pass / 13 fail to 16 pass / 6 fail, +and the whole corpus 194/97/196 to 201/90/196. The seven that moved are exactly +the seven priced as reachable above, and the six that remain are exactly the +six attributed to M2.5, M4 and M8. + +**The durability question is answered honestly rather than solved.** `$out` is +*not* atomic here: it drops the target and inserts, so a crash in between +leaves part of the new output where mongod would leave the whole of the old. +The shape that fixes it is write-to-temp-and-rename, and this server has no +rename command. Recorded in `apply_pending_write`'s own comment as well, since +that is where somebody will be standing when it matters. + diff --git a/src/commands.zig b/src/commands.zig index a19d7fe..225470e 100644 --- a/src/commands.zig +++ b/src/commands.zig @@ -23,6 +23,36 @@ pub const Context = struct { client_desc: []const u8, engine: *db.Engine, server_start: std.Io.Timestamp, + /// What an aggregation's last stage asked to be written, and where. + /// + /// `$out` and `$merge` write to a collection the pipeline is not reading, + /// and three things in this file stand against doing that inside the + /// handler: `aggregate` is a `.read` command, dispatch takes locks from a + /// static table before the handler runs, and `Collection.lock` allows only + /// one collection lock at a time. So the handler computes the documents + /// under the locks it has and leaves them here; the epilogue applies them + /// once every lock is released, next to the commit and the checkpoint that + /// already live there. + /// + /// Cleared at the top of every dispatch, so a request can never inherit the + /// one before it. + pending_write: ?PendingWrite = null, +}; + +/// A write an aggregation pipeline asked the epilogue to perform. +pub const PendingWrite = struct { + db: []const u8, + coll: []const u8, + /// Documents in the reply's arena, which outlives the epilogue. + docs: []const *const bson.Document, + mode: enum { + /// `$out`: the target holds the pipeline's output and nothing else. + replace, + /// `$merge`: each document replaces the one with its `_id`, or is + /// inserted. The default `whenMatched`/`whenNotMatched` pair, which is + /// the only one this server implements. + merge, + }, }; pub const ErrorCode = enum(i32) { @@ -62,6 +92,25 @@ 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_project_empty = 51272, + location_project_mixed = 31254, + location_project_unknown_expression = 31325, + location_write_stage_not_last = 40601, + 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 @@ -184,6 +233,10 @@ pub fn dispatch(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void { // the next write that needs the catalog exclusive to create a collection // blocks forever. It presented as an unrelated client-side timeout one // command later. + // Nothing carries over: a handler that errors before it sets one must not + // leave the previous command's write to be applied below. + ctx.pending_write = null; + var ns: ?struct { db: []const u8, coll: []const u8 } = null; if (cmd.locks.coll != .none) { const db_name = msg.db_name() orelse @@ -224,6 +277,26 @@ pub fn dispatch(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void { coll = null; if (catalog_held) ctx.engine.unlock_catalog(cmd.locks.catalog == .exclusive); catalog_held = false; + // Here, and only here: no lock is held, so taking the target's is not a + // second one. See `Context.pending_write`. + if (ctx.pending_write) |pending| { + ctx.pending_write = null; + apply_pending_write(ctx, pending) catch |err| { + // The pipeline's own answer is already in `reply`; replace it with + // the failure, because a client told `ok: 1` would believe the + // collection had been written. The pairs go, the arena stays -- + // resetting it would free the very strings this message is built + // from. + const detail = try std.fmt.allocPrint( + reply.arena_alloc(), + "the pipeline's output could not be written to {s}.{s}: {s}", + .{ pending.db, pending.coll, @errorName(err) }, + ); + reply.pairs.clearRetainingCapacity(); + try reply.put_error(@intFromEnum(ErrorCode.operation_failed), "OperationFailed", detail); + }; + try ctx.engine.commit(); + } if (cmd.locks.coll == .exclusive) { // Durability (seal + fsync) coalesces across concurrent writers. A // commit error deliberately wins over the handler's captured `result`: @@ -2140,7 +2213,6 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void { var start: usize = 0; var end: usize = offs.items.len; var count_stage: ?[]const u8 = null; - var proj_pairs: ?[]const bson.Pair = null; for (stages) |stage_v| { const stage = switch (stage_v) { @@ -2214,10 +2286,40 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void { const n = try stage_count(reply, stage[0].value, "$limit") orelse return; end = @min(end, start + n); } else if (std.mem.eql(u8, stage_name, "$project")) { - proj_pairs = doc_arg(stage[0].value); + const pp = doc_arg(stage[0].value) orelse return bad_value(reply, "$project requires a document"); + if (try refuse_unprojectable(reply, pp)) return; + // Applied here rather than remembered for the emit. It used to set + // a variable that only the last `$project` in a pipeline could win + // and that no later stage could see -- so a `$match` after a + // `$project` still matched on a field the projection had removed. + const arena = reply.arena_alloc(); + var projected: std.ArrayListUnmanaged(*const bson.Document) = .empty; + errdefer projected.deinit(ctx.gpa); + try projected.ensureTotalCapacity(ctx.gpa, end - start); + if (in_trees) { + for (trees.items[start..end]) |d| { + projected.appendAssumeCapacity(try projected_tree(arena, d, pp)); + } + } else { + for (offs.items[start..end]) |off| { + projected.appendAssumeCapacity(try projected_tree(arena, try doc_tree(arena, coll, off), pp)); + } + } + offs.deinit(ctx.gpa); + offs = .empty; + trees.deinit(ctx.gpa); + trees = projected; + projected = .empty; + in_trees = true; + start = 0; + end = trees.items.len; } 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, coll, gp, offs.items[start..end]); + const src: Stream = if (in_trees) + .{ .docs = trees.items[start..end] } + else + .{ .offsets = offs.items[start..end] }; + const grouped_opt = try run_group(ctx, reply, coll, gp, src); var grouped = grouped_opt orelse return; // Group results replace the stream: later stages see groups. offs.deinit(ctx.gpa); @@ -2228,6 +2330,39 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void { in_trees = true; start = 0; end = trees.items.len; + } else if (std.mem.eql(u8, stage_name, "$out") or std.mem.eql(u8, stage_name, "$merge")) { + const is_out = std.mem.eql(u8, stage_name, "$out"); + // MongoDB requires either to be last, and this server needs it too: + // the stage does not produce a stream for a later one to read. + if (!std.mem.eql(u8, stage_name, stages[stages.len - 1].doc[0].key)) { + const detail = try std.fmt.allocPrint( + reply.arena_alloc(), + "{s} can only be the final stage in the pipeline", + .{stage_name}, + ); + return reply.put_error(@intFromEnum(ErrorCode.location_write_stage_not_last), "Location40601", detail); + } + const target = (try write_stage_target(reply, db_name, stage[0].value, is_out)) orelse return; + const arena = reply.arena_alloc(); + var out_docs: std.ArrayListUnmanaged(*const bson.Document) = .empty; + if (in_trees) { + for (trees.items[start..end]) |d| try out_docs.append(arena, d); + } else { + for (offs.items[start..end]) |off| try out_docs.append(arena, try doc_tree(arena, coll, off)); + } + // Handed to the epilogue rather than written here: see + // `Context.pending_write`. The documents live in the reply's arena, + // which outlives it. + ctx.pending_write = .{ + .db = target.db, + .coll = target.coll, + .docs = out_docs.items, + .mode = if (is_out) .replace else .merge, + }; + // Both stages answer an empty cursor, as mongod does: the output + // went to a collection, not to the client. + try emit_first_batch(ctx, reply, db_name, coll_name, null, &.{}, batch_size); + return reply.put_ok(); } else if (std.mem.eql(u8, stage_name, "$count")) { count_stage = switch (stage[0].value) { .string => |s| s, @@ -2255,12 +2390,12 @@ fn cmd_aggregate(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void { const arena = reply.arena_alloc(); if (in_trees) { const window = trees.items[start..end]; - try emit_first_batch(ctx, reply, db_name, coll_name, proj_pairs, window, batch_size); + try emit_first_batch(ctx, reply, db_name, coll_name, null, window, batch_size); } 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)); const window = page.items; - try emit_first_batch(ctx, reply, db_name, coll_name, proj_pairs, window, batch_size); + try emit_first_batch(ctx, reply, db_name, coll_name, null, window, batch_size); } } try reply.put_ok(); @@ -2331,26 +2466,358 @@ 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"). +/// Refuse a `$project` this engine cannot carry out, answering the client. +/// Returns true when it did. +/// +/// The stage takes inclusion and exclusion flags and nothing else. A computed +/// field (`{y: {$literal: 5}}`) needs the expression evaluator that does not +/// exist yet, and a nested spec (`{a: {b: 1}}`) needs a narrowing +/// `query.project` does not do. Both used to be read as *falsy*, which put the +/// whole projection into its exclusion branch: `{$project: {y: {$literal: 5}}}` +/// returned every document with `y` removed, where mongod adds a computed `y`. +/// Measured on a live server, not inferred. +/// +/// Codes read off mongod 8.3.7. It refuses the empty and the mixed forms too, +/// so those two are parity rather than a limitation of this server. +fn refuse_unprojectable(reply: *wire.Reply, pp: []const bson.Pair) !bool { + if (pp.len == 0) { + try reply.put_error( + @intFromEnum(ErrorCode.location_project_empty), + "Location51272", + "Invalid $project :: caused by :: projection specification must have at least one field", + ); + return true; + } + var include: ?bool = null; + for (pp) |p| { + switch (p.value) { + .bool, .int32, .int64, .double => {}, + else => { + // A `$`-led document is an expression mongod evaluates and + // names in its own message; anything else is a nested spec, + // where the honest message is the one this server can stand + // behind. + const detail = if (p.value == .doc and p.value.doc.len > 0 and + p.value.doc[0].key.len > 0 and p.value.doc[0].key[0] == '$') + try std.fmt.allocPrint( + reply.arena_alloc(), + "Invalid $project :: caused by :: Unknown expression {s}", + .{p.value.doc[0].key}, + ) + else + try std.fmt.allocPrint( + reply.arena_alloc(), + "Invalid $project :: caused by :: field '{s}' must be an inclusion or " ++ + "exclusion flag: this server projects no computed or nested fields", + .{p.key}, + ); + try reply.put_error(@intFromEnum(ErrorCode.location_project_unknown_expression), "Location31325", detail); + return true; + }, + } + // `_id` is the one field that may be excluded from an inclusion + // projection, so it never decides which kind this is. + if (std.mem.eql(u8, p.key, "_id")) continue; + const flag = query.truthy(p.value); + if (include) |want| { + if (want != flag) { + const detail = try std.fmt.allocPrint( + reply.arena_alloc(), + "Invalid $project :: caused by :: Cannot do {s} on field {s} in {s} projection", + .{ + if (flag) "inclusion" else "exclusion", + p.key, + if (want) "inclusion" else "exclusion", + }, + ); + try reply.put_error(@intFromEnum(ErrorCode.location_project_mixed), "Location31254", detail); + return true; + } + } else include = flag; + } + return false; +} + +/// One document put through a projection, as a tree the rest of the pipeline +/// can read. This is what makes `$project` a stage rather than a note about how +/// to print the answer. +fn projected_tree( + arena: std.mem.Allocator, + doc: *const bson.Document, + pp: []const bson.Pair, +) !*const bson.Document { + var out: std.ArrayListUnmanaged(bson.Pair) = .empty; + try query.project(arena, doc, &.{ .arena = undefined, .pairs = pp }, &out); + const projected = try arena.create(bson.Document); + projected.* = .{ .arena = undefined, .pairs = out.items }; + return projected; +} + +/// Where a `$out` or `$merge` writes, or null once the client has been told +/// why not. +/// +/// `$out` takes a collection name or `{db, coll}`; `$merge` takes `into` in +/// either of those shapes. Everything past that -- `whenMatched`, +/// `whenNotMatched`, `on`, `let` -- selects behaviour this server does not +/// have, so it is refused rather than ignored: a `whenMatched: "fail"` that +/// silently merged would be the same lie Tier 0 spent three commits removing. +fn write_stage_target( + reply: *wire.Reply, + db_name: []const u8, + v: bson.Value, + is_out: bool, +) !?struct { db: []const u8, coll: []const u8 } { + var spec = v; + if (!is_out) { + const d = doc_arg(v) orelse { + // mongod's IDL parser answers for the whole stage document. + try reply.put_error( + @intFromEnum(ErrorCode.idl_failed_to_parse), + "IDLFailedToParse", + "BSON field '$merge.into' is missing but a required field", + ); + return null; + }; + for (d) |p| { + if (std.mem.eql(u8, p.key, "into")) continue; + const detail = try std.fmt.allocPrint( + reply.arena_alloc(), + "$merge does not support '{s}' on this server: only the default " ++ + "whenMatched/whenNotMatched behaviour is implemented", + .{p.key}, + ); + try reply.put_error(@intFromEnum(ErrorCode.idl_unknown_field), "IDLUnknownField", detail); + return null; + } + spec = bson.get_pair(d, "into") orelse { + try reply.put_error( + @intFromEnum(ErrorCode.idl_failed_to_parse), + "IDLFailedToParse", + "BSON field '$merge.into' is missing but a required field", + ); + return null; + }; + } + switch (spec) { + .string => |name| return .{ .db = db_name, .coll = name }, + .doc => |d| { + const coll = str_arg(bson.get_pair(d, "coll")) orelse { + try bad_value(reply, "the target of a write stage needs a coll"); + return null; + }; + return .{ .db = str_arg(bson.get_pair(d, "db")) orelse db_name, .coll = coll }; + }, + else => { + try bad_value(reply, "the target of a write stage must be a string or a document"); + return null; + }, + } +} + +/// Apply what an aggregation's last stage asked for. The caller holds no lock. +/// +/// Not atomic, and that has to be said out loud: mongod replaces an `$out` +/// target atomically, and this engine has no cross-collection atomicity and no +/// rename to build one out of. A crash between the drop and the last insert +/// leaves the target holding part of the new output where MongoDB would leave +/// the whole of the old. Recorded in `docs/M2_DESIGN_REVIEW.md` as the open +/// half of this decision rather than papered over: the shape that fixes it is +/// write-to-temp-and-rename, and rename is a command this server does not have. +fn apply_pending_write(ctx: *Context, pending: PendingWrite) !void { + try ctx.engine.lock(); + defer ctx.engine.unlock(); + if (pending.mode == .replace) { + // `$out` means "the target holds this and nothing else". + _ = ctx.engine.drop_collection(pending.db, pending.coll) catch |err| switch (err) { + error.NamespaceNotFound => {}, + else => return err, + }; + } + for (pending.docs) |d| { + // The pairs belong to the reply's arena; this Document is only a + // carrier, so its own arena is empty and frees nothing that matters. + var doc: bson.Document = .{ .arena = std.heap.ArenaAllocator.init(ctx.gpa), .pairs = d.pairs }; + defer doc.arena.deinit(); + // `$out` writes into a collection it has just emptied, so every write + // is an insert; `$merge`'s default pair is "replace the document with + // this `_id`, or insert it", which is what `replace` already means. + _ = try ctx.engine.replace(pending.db, pending.coll, &doc, ctx.oid_gen); + } +} + +/// Where a pipeline stage reads its input. +/// +/// A pipeline starts as slab offsets -- matched and reordered in place, never +/// materialized -- and flips to generated documents the moment a stage +/// produces something that is not in the slab. Both forms are real and a stage +/// that reads only one of them reads the wrong list. +/// +/// `$group` used to take `[]const u64` and be handed `offs.items[start..end]` +/// unconditionally, with `start`/`end` set from whichever list was live. After +/// a stage that materializes, `offs` is empty and the bounds are the tree +/// count, so a second `$group` sliced an empty list with a non-zero end and +/// panicked the server. Any client could send it. +const Stream = union(enum) { + offsets: []const u64, + docs: []const *const bson.Document, + + fn len(self: Stream) usize { + return switch (self) { + .offsets => |o| o.len, + .docs => |d| d.len, + }; + } +}; + +/// A dotted path resolved against a document tree, the counterpart of +/// `query_path_value_bytes` for the materialized half of a stream. +fn path_in_pairs(pairs: []const bson.Pair, path: []const u8) ?bson.Value { + var it = std.mem.splitScalar(u8, path, '.'); + var cur = bson.get_pair(pairs, it.next() orelse return null) orelse return null; + while (it.next()) |seg| { + cur = switch (cur) { + .doc => |p| bson.get_pair(p, seg) orelse return null, + else => return null, + }; + } + return cur; +} + +/// One item of a stream, resolved along `path`, whichever form the stream is +/// in. The slab side stays byte-walked: `$group` over a million documents does +/// not build a million trees to read one field. +fn stream_path( + gpa: std.mem.Allocator, + coll: *const Collection, + src: Stream, + i: usize, + path: []const u8, +) !?bson.Value { + return switch (src) { + .offsets => |o| try query_path_value_bytes(gpa, coll.doc_bytes(o[i]), path), + .docs => |d| path_in_pairs(d[i].pairs, path), + }; +} + +/// 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, coll: *const Collection, group_pairs: []const bson.Pair, - docs: []const u64, + src: Stream, ) !?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 { @@ -2371,11 +2838,11 @@ fn run_group( // 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] == '$') (try query_path_value_bytes(walk_arena.allocator(), doc, s[1..])) orelse .null else id_expr, - else => id_expr, + var i: usize = 0; + while (i < src.len()) : (i += 1) { + const id_value: bson.Value = switch (id_class) { + .path => |path| (try stream_path(walk_arena.allocator(), coll, src, i, path)) orelse .null, + .constant => |v| v, }; id_key_buf.clearRetainingCapacity(); try bson.write_value(id_value, ctx.gpa, &id_key_buf); @@ -2391,30 +2858,20 @@ fn run_group( @memset(sums, 0); 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) { + for (accs.items, 0..) |acc, a| { + // 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 stream_path(walk_arena.allocator(), coll, src, i, path)) orelse .null, + .constant => |c| c, + }; + gop.value_ptr.sums[a] += 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; } } @@ -2430,13 +2887,13 @@ fn run_group( const npairs = 1 + accs.items.len; const pairs = try arena.alloc(bson.Pair, npairs); pairs[0] = .{ .key = "_id", .value = try bson.copy_value(arena, entry.value_ptr.id_value) }; - for (accs.items, 0..) |acc, i| { - const sum: f64 = entry.value_ptr.sums[i]; + for (accs.items, 0..) |acc, a| { + const sum: f64 = entry.value_ptr.sums[a]; const sum_value: bson.Value = if (sum == @floor(sum) and sum <= 2_147_483_647 and sum >= -2_147_483_648) .{ .int32 = @intFromFloat(sum) } else .{ .double = sum }; - pairs[1 + i] = .{ .key = acc.key, .value = sum_value }; + pairs[1 + a] = .{ .key = acc.key, .value = sum_value }; } const doc = try arena.create(bson.Document); doc.* = bson.Document{ .arena = undefined, .pairs = pairs }; @@ -3872,6 +4329,418 @@ test "aggregate $sort without a preceding $group sorts and frees correctly" { } } +test "$out and $merge write through the epilogue" { + // The write stages are the reason `Context.pending_write` exists. They + // write to a collection the pipeline is not reading, and doing that inside + // the handler would take a second collection lock while the first is held + // -- which `Collection.lock`'s own comment forbids. So the handler computes + // and the epilogue writes, with nothing held. + // + // Mutation check: apply the write inside the `$out` branch instead of + // stashing it. Deadlocks rather than fails, which is the argument for the + // epilogue in one line. + 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, "src", &.{ + .{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "x", .value = .{ .int32 = 1 } } } }, + .{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 2 } }, .{ .key = "x", .value = .{ .int32 = 2 } } } }, + }); + try dispatch_insert(&tdb, io, "dst", &.{ + .{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 9 } }, .{ .key = "old", .value = .{ .bool = true } } } }, + }); + var ctx = tdb.ctx(io); + + const run = struct { + fn go(c: *Context, stages: []const bson.Value) !wire.Reply { + var reply = wire.Reply.init(testing.allocator); + errdefer reply.deinit(); + var msg = try parse_fake_msg("aggregate", .{ .string = "src" }, &.{ + .{ .key = "pipeline", .value = .{ .array = stages } }, + .{ .key = "cursor", .value = .{ .doc = &.{} } }, + }); + defer msg.deinit(); + try dispatch(c, &msg, &reply); + return reply; + } + }.go; + + // $out replaces the target outright: the pre-existing document is gone. + { + const stages = [_]bson.Value{.{ .doc = &.{.{ .key = "$out", .value = .{ .string = "dst" } }} }}; + var reply = try run(&ctx, &stages); + defer reply.deinit(); + try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply.pairs.items, "ok").?.double); + } + { + const coll = ctx.engine.get_collection("test", "dst").?; + try testing.expectEqual(@as(u64, 2), coll.doc_count); + } + + // $merge keeps what it does not name and replaces what it does. + try dispatch_insert(&tdb, io, "dst", &.{ + .{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 9 } }, .{ .key = "old", .value = .{ .bool = true } } } }, + }); + { + const stages = [_]bson.Value{.{ .doc = &.{.{ .key = "$merge", .value = .{ .doc = &.{ + .{ .key = "into", .value = .{ .string = "dst" } }, + } } }} }}; + var reply = try run(&ctx, &stages); + defer reply.deinit(); + try testing.expectEqual(@as(f64, 1.0), bson.get_pair(reply.pairs.items, "ok").?.double); + } + { + const coll = ctx.engine.get_collection("test", "dst").?; + try testing.expectEqual(@as(u64, 3), coll.doc_count); + } + + // And the three shapes that are refused rather than half-honoured. + const Case = struct { name: []const u8, stages: []const bson.Value, code: i32 }; + const out_then_match = [_]bson.Value{ + .{ .doc = &.{.{ .key = "$out", .value = .{ .string = "dst" } }} }, + .{ .doc = &.{.{ .key = "$match", .value = .{ .doc = &.{} } }} }, + }; + const merge_bare = [_]bson.Value{.{ .doc = &.{.{ .key = "$merge", .value = .{ .doc = &.{} } }} }}; + const merge_when = [_]bson.Value{.{ .doc = &.{.{ .key = "$merge", .value = .{ .doc = &.{ + .{ .key = "into", .value = .{ .string = "dst" } }, + .{ .key = "whenMatched", .value = .{ .string = "fail" } }, + } } }} }}; + const cases = [_]Case{ + .{ .name = "$out is not last", .stages = &out_then_match, .code = 40601 }, + .{ .name = "$merge without into", .stages = &merge_bare, .code = 40414 }, + .{ .name = "$merge with whenMatched", .stages = &merge_when, .code = 40415 }, + }; + for (cases) |c| { + var reply = try run(&ctx, c.stages); + defer reply.deinit(); + testing.expectEqual(@as(f64, 0.0), bson.get_pair(reply.pairs.items, "ok").?.double) catch |err| { + std.debug.print(" {s}: answered ok:1\n", .{c.name}); + return err; + }; + try testing.expectEqual(c.code, bson.get_pair(reply.pairs.items, "code").?.int32); + // And nothing was written: a refused stage leaves the target alone. + try testing.expectEqual(@as(u64, 3), ctx.engine.get_collection("test", "dst").?.doc_count); + } +} + +test "$project is a stage, not a note about how to print the answer" { + // It used to set a variable applied once, at the emit. Three consequences, + // all measured on a live server before this changed: only the *last* + // `$project` in a pipeline had any effect; a `$match` after one still saw + // the field it had removed; and `{y: {$literal: 5}}` read as falsy, which + // flipped the whole projection into its exclusion branch and returned every + // document minus `y`. + // + // Every answer below is byte-identical to mongod 8.3.7 on the same input. + // + // Mutation check: move the projection back to the emit -- pass `pp` to + // `emit_first_batch` instead of transforming the stream -- and the first + // two cases go red. + 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, "pj", &.{ + .{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "g", .value = .{ .string = "a" } }, .{ .key = "x", .value = .{ .int32 = 10 } } } }, + .{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 2 } }, .{ .key = "g", .value = .{ .string = "b" } }, .{ .key = "x", .value = .{ .int32 = 20 } } } }, + }); + var ctx = tdb.ctx(io); + + const keep_g = bson.Value{ .doc = &.{.{ .key = "$project", .value = .{ .doc = &.{ + .{ .key = "g", .value = .{ .int32 = 1 } }, + } } }} }; + + // A $match after a $project cannot see what the projection removed. + { + const match_x = bson.Value{ .doc = &.{.{ .key = "$match", .value = .{ .doc = &.{ + .{ .key = "x", .value = .{ .int32 = 10 } }, + } } }} }; + const stages = [_]bson.Value{ keep_g, match_x }; + var msg = try parse_fake_msg("aggregate", .{ .string = "pj" }, &.{ + .{ .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 cur = bson.get_pair(reply.pairs.items, "cursor").?; + try testing.expectEqual(@as(usize, 0), bson.get_pair(cur.doc, "firstBatch").?.array.len); + } + + // And a $group after one reads the projected document, not the stored one. + { + const group_g = bson.Value{ .doc = &.{.{ .key = "$group", .value = .{ .doc = &.{ + .{ .key = "_id", .value = .null }, + .{ .key = "n", .value = .{ .doc = &.{.{ .key = "$sum", .value = .{ .int32 = 1 } }} } }, + .{ .key = "x", .value = .{ .doc = &.{.{ .key = "$sum", .value = .{ .string = "$x" } }} } }, + } } }} }; + const stages = [_]bson.Value{ keep_g, group_g }; + var msg = try parse_fake_msg("aggregate", .{ .string = "pj" }, &.{ + .{ .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 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, 2), bson.get_pair(batch[0].doc, "n").?.int32); + // `x` was projected away, so summing it is summing nothing. + try testing.expectEqual(@as(i32, 0), bson.get_pair(batch[0].doc, "x").?.int32); + } + + // The three shapes mongod refuses, refused with its codes. + const Case = struct { name: []const u8, spec: []const bson.Pair, code: i32 }; + const cases = [_]Case{ + .{ .name = "empty", .spec = &.{}, .code = 51272 }, + .{ + .name = "mixed inclusion and exclusion", + .spec = &.{ + .{ .key = "g", .value = .{ .int32 = 1 } }, + .{ .key = "x", .value = .{ .int32 = 0 } }, + }, + .code = 31254, + }, + .{ + .name = "a computed field", + .spec = &.{.{ .key = "y", .value = .{ .doc = &.{.{ .key = "$literal", .value = .{ .int32 = 5 } }} } }}, + .code = 31325, + }, + .{ + .name = "a nested spec", + .spec = &.{.{ .key = "a", .value = .{ .doc = &.{.{ .key = "b", .value = .{ .int32 = 1 } }} } }}, + .code = 31325, + }, + }; + for (cases) |c| { + const stages = [_]bson.Value{.{ .doc = &.{.{ .key = "$project", .value = .{ .doc = c.spec } }} }}; + var msg = try parse_fake_msg("aggregate", .{ .string = "pj" }, &.{ + .{ .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); + testing.expectEqual(@as(f64, 0.0), bson.get_pair(reply.pairs.items, "ok").?.double) catch |err| { + std.debug.print(" {s}: answered ok:1\n", .{c.name}); + return err; + }; + testing.expectEqual(c.code, bson.get_pair(reply.pairs.items, "code").?.int32) catch |err| { + std.debug.print(" {s}: wrong code\n", .{c.name}); + return err; + }; + } +} + +test "a pipeline may group what an earlier stage generated" { + // A remote crash, reachable by any client with a two-stage pipeline and no + // authentication in front of it: `$group` took `[]const u64` and was handed + // `offs.items[start..end]` whatever the stream was made of. After a stage + // that materializes -- another `$group`, and now `$project` -- `offs` is + // empty while the bounds count trees, so the slice ran off an empty list + // and panicked the server thread. + // + // "index out of bounds: index 2, len 0", measured against the binary at the + // previous commit before this was written. + // + // Mutation check: hand `run_group` `.{ .offsets = offs.items[start..end] }` + // unconditionally, and this aborts the run rather than failing it -- which + // is exactly why it was worth a test rather than a code read. + 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, "gg", &.{ + .{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 1 } }, .{ .key = "k", .value = .{ .string = "a" } } } }, + .{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 2 } }, .{ .key = "k", .value = .{ .string = "a" } } } }, + .{ .doc = &.{ .{ .key = "_id", .value = .{ .int32 = 3 } }, .{ .key = "k", .value = .{ .string = "b" } } } }, + }); + + // Group by key, then count the groups: the second $group reads what the + // first one generated, which lives in no slab. + const by_k = bson.Value{ .doc = &.{.{ .key = "$group", .value = .{ .doc = &.{ + .{ .key = "_id", .value = .{ .string = "$k" } }, + .{ .key = "n", .value = .{ .doc = &.{.{ .key = "$sum", .value = .{ .int32 = 1 } }} } }, + } } }} }; + const count_groups = bson.Value{ .doc = &.{.{ .key = "$group", .value = .{ .doc = &.{ + .{ .key = "_id", .value = .null }, + .{ .key = "groups", .value = .{ .doc = &.{.{ .key = "$sum", .value = .{ .int32 = 1 } }} } }, + // And a path that only resolves against the *generated* document, + // which is what makes this more than a bounds check. + .{ .key = "docs", .value = .{ .doc = &.{.{ .key = "$sum", .value = .{ .string = "$n" } }} } }, + } } }} }; + const stages = [_]bson.Value{ by_k, count_groups }; + + var ctx = tdb.ctx(io); + var msg = try parse_fake_msg("aggregate", .{ .string = "gg" }, &.{ + .{ .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); + + 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, 2), bson.get_pair(batch[0].doc, "groups").?.int32); + try testing.expectEqual(@as(i32, 3), bson.get_pair(batch[0].doc, "docs").?.int32); +} + +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 diff --git a/tests/spec/scorecard.txt b/tests/spec/scorecard.txt index e690acb..e73dc88 100644 --- a/tests/spec/scorecard.txt +++ b/tests/spec/scorecard.txt @@ -18,16 +18,16 @@ # hasServerConnectionId, and maxTimeMS in an expected command (CSOT rewrites # it -- the only assertion this runner declines to make). -total 194 pass 97 fail 196 skip 175 files 0 errored +total 201 pass 90 fail 196 skip 175 files 0 errored # per-file: name pass fail skip aggregate-allowdiskuse.json 3 0 0 aggregate-collation.json 0 1 0 aggregate-let.json 0 2 2 aggregate-merge-errorResponse.json 0 0 1 -aggregate-merge.json 0 5 0 +aggregate-merge.json 5 0 0 aggregate-out-readConcern.json 0 0 4 -aggregate-out.json 0 2 0 +aggregate-out.json 2 0 0 aggregate-rawdata.json 1 0 1 aggregate-write-readPreference.json 0 0 4 aggregate.json 5 0 2 @@ -202,16 +202,9 @@ aggregate-collation.json FAIL Aggregate with collation aggregate: expected 1 ele aggregate-let.json SKIP Aggregate with let option needs server >= 5.0 aggregate-let.json FAIL Aggregate with let option unsupported (server-side error) aggregate: expected an error, the operation succeeded aggregate-let.json SKIP Aggregate to collection with let option needs server >= 5.0 -aggregate-let.json FAIL Aggregate to collection with let option unsupported (server-side error) aggregate: error message "Unrecognized pipeline stage name: '$out'" does not contain "unrecognized field 'let'" +aggregate-let.json FAIL Aggregate to collection with let option unsupported (server-side error) aggregate: expected an error, the operation succeeded aggregate-merge-errorResponse.json SKIP aggregate $merge DuplicateKey error is accessible needs server >= 5.1 -aggregate-merge.json FAIL Aggregate with $merge MongoServerError: Unrecognized pipeline stage name: '$merge' -aggregate-merge.json FAIL Aggregate with $merge and batch size of 0 MongoServerError: Unrecognized pipeline stage name: '$merge' -aggregate-merge.json FAIL Aggregate with $merge and majority readConcern MongoServerError: Unrecognized pipeline stage name: '$merge' -aggregate-merge.json FAIL Aggregate with $merge and local readConcern MongoServerError: Unrecognized pipeline stage name: '$merge' -aggregate-merge.json FAIL Aggregate with $merge and available readConcern MongoServerError: Unrecognized pipeline stage name: '$merge' aggregate-out-readConcern.json SKIP * needs topology replicaset/sharded -aggregate-out.json FAIL Aggregate with $out MongoServerError: Unrecognized pipeline stage name: '$out' -aggregate-out.json FAIL Aggregate with $out and batch size of 0 MongoServerError: Unrecognized pipeline stage name: '$out' aggregate-rawdata.json SKIP Aggregate with rawData option needs server >= 8.2.0 aggregate-write-readPreference.json SKIP * needs topology replicaset/sharded/load-balanced aggregate.json SKIP aggregate with a document comment - pre 4.4 needs server <= 4.2.99 @@ -261,7 +254,7 @@ bulkWrite-updateOne-pipeline.json FAIL UpdateOne in bulk write using pipelines M bulkWrite-updateOne-rawdata.json SKIP BulkWrite updateOne with rawData option needs server >= 8.2.0 bulkWrite-updateOne-rawdata.json FAIL BulkWrite updateOne with rawData option on less than 8.2.0 - ignore argument MongoBulkWriteError: update spec requires u bulkWrite-updateOne-sort.json SKIP BulkWrite updateOne with sort option needs server >= 8.0 -bypassDocumentValidation.json FAIL Aggregate with $out passes bypassDocumentValidation: false MongoServerError: Unrecognized pipeline stage name: '$out' +bypassDocumentValidation.json FAIL Aggregate with $out passes bypassDocumentValidation: false events client0[0].command.bypassDocumentValidation: missing from actual bypassDocumentValidation.json FAIL BulkWrite passes bypassDocumentValidation: false events client0[0].command.bypassDocumentValidation: missing from actual bypassDocumentValidation.json FAIL FindOneAndReplace passes bypassDocumentValidation: false events client0[0].command.bypassDocumentValidation: missing from actual bypassDocumentValidation.json FAIL FindOneAndUpdate passes bypassDocumentValidation: false events client0[0].command.bypassDocumentValidation: missing from actual