index/commands: let an index supply the sort order

A sort ordered every match before discarding all but one page, even when
an index already held the candidates in exactly that order. The planner
now recognizes that case and the scan streams the page straight out.

An index provides the sort when the sort keys line up with the components
after the equality-pinned prefix (those are fixed to one value each, so
they do not affect the order of what follows) and every direction agrees
uniformly -- all the same way round, or all opposite, since the array can
only be read forwards or backwards. Multikey indexes are excluded: they
emit a document once per indexed value, so their order is not an order on
documents. So is an $in, whose disjoint ranges concatenate unordered.

Entries already come out of the array in key order, so forward scans were
sorted all along; what destroyed it was the dedupe pass sorting by id.
The conditions above are exactly the ones under which that pass is
skipped, so ordered output needs only reversing for a backward scan.

evaluate_index gains a plan shape it did not have: a full index scan when
the ordering is the reason to use it. Without that, find({}).sort(...)
was unreachable -- an empty filter yields no clauses and the planner bailed
before looking at any index. It is guarded to only appear when the sort is
satisfied, since otherwise scanning the docs map directly is cheaper.

The early stop had to move into the scan, which is the only place that
knows whether the order came from an index: a limit is a valid page
boundary without a sort, or with one an index supplies, and otherwise
means nothing. Getting this wrong the other way -- limiting first and
re-scanning -- would have doubled the work for every unindexed sort.

  find({}).sort({k: 1}).limit(20) over 65,536 x 16 KB documents:
    4.0ms -> 1.0ms
  sort({_id: -1}) is unchanged at 4.3ms: nothing ordered covers _id yet.

Checked against the definition rather than by example: eleven query shapes
-- forward, backward, skip, unlimited, filtered, equality-prefixed both
directions, compound, directions the index cannot serve, and a range --
each compared through the real driver against the page of the equivalent
full sort.

Verified: 78 unit tests under ReleaseFast and ReleaseSafe, e2e
29/16/17/3/2, the crash pair, e2e6 72/72.

src/commands.zig

@@ -577,12 +577,13 @@ fn cmd_find(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
         const skip_usize = std.math.cast(usize, skip) orelse break :blk 0;
         break :blk skip_usize +| limit;
     };
-    // Stop scanning once the page is filled. Only sound without a sort,
-    // which has to see every match before it can tell which ones the page
-    // contains.
-    _ = try scan_matching(ctx, db_name, coll_name, filter, if (sort_keys.len > 0) 0 else page_end, &matched);
+    // An index whose order already is the requested one lets the scan stop
+    // at the page boundary and skip sorting entirely. Otherwise a sort has
+    // to see every match before it can tell which ones the page contains.
+    var index_sorted = false;
+    _ = try scan_sorted(ctx, db_name, coll_name, filter, page_end, &matched, sort_keys, &index_sorted);
 
-    if (sort_keys.len > 0) {
+    if (sort_keys.len > 0 and !index_sorted) {
         // Selecting the page is much cheaper than ordering everything when
         // the page is a small fraction of the matches. Above that fraction
         // the heap's bookkeeping stops paying for itself.
@@ -616,6 +617,29 @@ fn scan_matching(
     limit: usize,
     out: ?*std.ArrayListUnmanaged(*const bson.Document),
 ) !usize {
+    return scan_sorted(ctx, db_name, coll_name, filter, limit, out, &.{}, null);
+}
+
+/// `scan_matching` plus the option of having an index produce the ordering.
+/// When `sorted` is given it reports whether the candidates came out in
+/// `sort` order, in which case the caller must not sort them again — and
+/// `limit` is then a genuine early stop rather than an arbitrary subset.
+fn scan_sorted(
+    ctx: *Context,
+    db_name: []const u8,
+    coll_name: []const u8,
+    filter: []const bson.Pair,
+    limit: usize,
+    out: ?*std.ArrayListUnmanaged(*const bson.Document),
+    sort: []const query.SortKey,
+    sorted: ?*bool,
+) !usize {
+    if (sorted) |flag| flag.* = false;
+    // Stopping early is only meaningful when the candidates come out in the
+    // order the caller asked for. Without a sort any subset of that size is
+    // a valid page; with one, the limit is honoured only if an index turns
+    // out to supply the ordering.
+    var lim: usize = if (sort.len == 0) limit else 0;
     const coll = ctx.engine.get_collection(db_name, coll_name) orelse return 0;
     const filter_doc = bson.Document{ .arena = undefined, .pairs = filter };
     var n: usize = 0;
@@ -637,25 +661,27 @@ fn scan_matching(
             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;
+            if (lim != 0 and n >= lim) 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| {
+    if (try index.plan(ctx.gpa, coll.indexes.items, filter, sort)) |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);
+        if (sorted) |flag| flag.* = plan.provides_sort;
+        if (plan.provides_sort) lim = limit;
         for (ids.items) |id| {
             const doc = coll.docs.get(id) orelse continue;
             if (!try query.matches(ctx.gpa, &filter_doc, doc)) continue;
             if (out) |list| try list.append(ctx.gpa, doc);
             n += 1;
-            if (limit != 0 and n >= limit) break;
+            if (lim != 0 and n >= lim) break;
         }
         return n;
     }
@@ -665,7 +691,7 @@ fn scan_matching(
         if (!try query.matches(ctx.gpa, &filter_doc, entry.value_ptr.*)) continue;
         if (out) |list| try list.append(ctx.gpa, entry.value_ptr.*);
         n += 1;
-        if (limit != 0 and n >= limit) break;
+        if (lim != 0 and n >= lim) break;
     }
     return n;
 }