feat: add sortPartitionByOrder and Ctx to window

src/window.zig

@@ -306,3 +306,188 @@ fn buildPartitions(allocator: Allocator, rows: []const Row, partitionBy: []const
             try rowmod.appendOrdered(&keyBuf, v);
         }
 
+        if (map.getPtr(keyBuf.items)) |list| {
+            try list.append(allocator, i);
+            continue;
+        }
+        const key = try allocator.dupe(u8, keyBuf.items);
+        errdefer allocator.free(key);
+        var list: std.ArrayListUnmanaged(usize) = .{};
+        errdefer list.deinit(allocator);
+        try list.append(allocator, i);
+        // `map.putAssumeCapacity`/`order.appendAssumeCapacity` below are
+        // INFALLIBLE (capacity reserved just above) and back to back with
+        // no fallible statement between them - a bug caught by
+        // `computeMany`'s exhaustive OOM sweep test: with plain fallible
+        // `map.put` then `order.append`, a failure in `order.append` AFTER
+        // `map.put` already succeeded left `key`/`list` reachable from BOTH
+        // this iteration's still-armed `errdefer list.deinit(allocator)`
+        // AND the function-level `map` cleanup `defer` above (which iterates
+        // every entry, including the one just inserted) - both would then
+        // free the SAME `list` backing array. Reserving capacity for both
+        // `map` and `order` up front and using the assume-capacity variants
+        // means nothing fallible can happen between the two inserts, so
+        // there is no window for a failure to leave `key`/`list` owned by
+        // one structure while their local errdefers are still armed.
+        try map.ensureUnusedCapacity(allocator, 1);
+        try order.ensureUnusedCapacity(allocator, 1);
+        map.putAssumeCapacity(key, list);
+        order.appendAssumeCapacity(key);
+    }
+
+    const result = try allocator.alloc([]usize, order.items.len);
+    var filled: usize = 0;
+    errdefer {
+        for (result[0..filled]) |p| allocator.free(p);
+        allocator.free(result);
+    }
+    for (order.items, 0..) |key, idx| {
+        const list = map.getPtr(key).?;
+        result[idx] = try list.toOwnedSlice(allocator);
+        filled += 1;
+    }
+    return result;
+}
+
+/// Stable in-place sort of a partition's row INDICES by `orderBy` - the
+/// window-function analogue of `query.zig`'s `sortRowsBy`/`aggregate.zig`'s
+/// `sortGroupsBy`, but sorting `usize` indices into `rows` rather than the
+/// rows themselves (so the caller's row order and `results`' index
+/// alignment are never disturbed). `orderBy.len == 0` is a no-op, leaving
+/// `indices` in first-seen (input) order.
+fn sortPartitionByOrder(rows: []const Row, indices: []usize, orderBy: []const querymod.OrderBy) void {
+    if (orderBy.len == 0) return;
+    const Ctx = struct {
+        rows: []const Row,
+        orderBy: []const querymod.OrderBy,
+
+        fn lessThan(ctx: @This(), a: usize, b: usize) bool {
+            for (ctx.orderBy) |ob| {
+                const av = ctx.rows[a].get(ob.field) orelse Value{ .none = {} };
+                const bv = ctx.rows[b].get(ob.field) orelse Value{ .none = {} };
+                const cmp = rowmod.compare(av, bv);
+                if (cmp == .eq) continue;
+                return if (ob.descending) cmp == .gt else cmp == .lt;
+            }
+            return false;
+        }
+    };
+    std.sort.block(usize, indices, Ctx{ .rows = rows, .orderBy = orderBy }, Ctx.lessThan);
+}
+
+/// Whether `rows[curIdx]` ties `rows[prevIdx]` on every `orderBy` column -
+/// the tie-group boundary `rank`/`dense_rank` need to decide whether to
+/// advance. Only meaningful once `indices` has already been sorted by the
+/// same `orderBy`.
+fn tiesWithPrev(rows: []const Row, orderBy: []const querymod.OrderBy, prevIdx: usize, curIdx: usize) bool {
+    for (orderBy) |ob| {
+        const av = rows[curIdx].get(ob.field) orelse Value{ .none = {} };
+        const bv = rows[prevIdx].get(ob.field) orelse Value{ .none = {} };
+        if (rowmod.compare(av, bv) != .eq) return false;
+    }
+    return true;
+}
+
+/// `buildPartitions` followed by sorting every partition in place by
+/// `orderBy` - the shared prep step `compute()` and `orderedIndices()` both
+/// need, factored out so the flattened display order `orderedIndices`
+/// exposes is GUARANTEED to match the partition walk `compute()` uses
+/// internally (computed once, not two independently-written sorts that
+/// could drift apart).
+fn partitionedAndSorted(allocator: Allocator, rows: []const Row, partitionBy: []const []const u8, orderBy: []const querymod.OrderBy) Allocator.Error![][]usize {
+    const partitions = try buildPartitions(allocator, rows, partitionBy);
+    for (partitions) |indices| sortPartitionByOrder(rows, indices, orderBy);
+    return partitions;
+}
+
+/// Returns the row-index walk order `compute()` uses internally, flattened:
+/// every partition's member indices (each sorted by `orderBy`), partitions
+/// concatenated in first-seen order. `compute()`'s own results stay aligned
+/// to `rows`' ORIGINAL order regardless of this - this exists for callers
+/// (namely `table window`'s CLI) that want to PAGE or DISPLAY results in
+/// the same order the window computation itself walked, rather than raw
+/// scan order, which is what plain `[start..end]` slicing of `rows` would
+/// give instead. Caller owns the returned slice.
+pub fn orderedIndices(allocator: Allocator, rows: []const Row, partitionBy: []const []const u8, orderBy: []const querymod.OrderBy) Allocator.Error![]usize {
+    const partitions = try partitionedAndSorted(allocator, rows, partitionBy, orderBy);
+    defer {
+        for (partitions) |p| allocator.free(p);
+        allocator.free(partitions);
+    }
+    const out = try allocator.alloc(usize, rows.len);
+    var pos: usize = 0;
+    for (partitions) |indices| {
+        @memcpy(out[pos..][0..indices.len], indices);
+        pos += indices.len;
+    }
+    return out;
+}
+
+/// Computes one window-function result `Value` per input row, returned in
+/// the SAME order as `rows` (never reorders the caller's data) - the window
+/// counterpart of `aggregate.zig`'s `run`, which collapses rows into one
+/// per group instead of annotating every one. `partitionBy` splits `rows`
+/// into independent groups exactly like GROUP BY (first-seen order, empty
+/// = a single partition covering every row); `orderBy` decides the
+/// ranking/running-total walk WITHIN each partition.
+///
+/// Frame semantics follow standard SQL defaults, decided entirely by
+/// whether `orderBy` is empty:
+///   - `orderBy.len == 0`: `rank`/`dense_rank` treat the whole partition as
+///     one tie (every row gets 1); `row_number` still assigns 1..n in
+///     first-seen order, since it needs no ordering key to number rows.
+///     The running-aggregate family uses the whole-partition frame
+///     (`UNBOUNDED PRECEDING .. UNBOUNDED FOLLOWING`) - every row in the
+///     partition sees the SAME final total, mirroring `SUM(x) OVER
+///     (PARTITION BY p)` with no `ORDER BY` inside the `OVER` clause.
+///   - `orderBy.len > 0`: `frame` picks the running-aggregate family's frame
+///     along the sorted sequence. `.rows` walks one physical row at a time
+///     (`ROWS UNBOUNDED PRECEDING .. CURRENT ROW`) - every row gets its own
+///     individual running value, even a row tied with its predecessor.
+///     `.range` (SQL's actual default, `RANGE UNBOUNDED PRECEDING ..
+///     CURRENT ROW`) instead advances the accumulator by a whole tie group
+///     at once and stamps every row in that group with the SAME value (the
+///     group's combined total, not each member's own position within it) -
+///     the same tie test `rank`/`dense_rank` already use
+///     (`tiesWithPrev`), just driving accumulation instead of a rank
+///     counter. `frame` is ignored (both variants agree) when `orderBy` is
+///     empty, and rejected outright (`error.RangeNotApplicable`) for the
+///     ranking family - see `WindowError.RangeNotApplicable`.
+///
+/// `lag`/`lead` (`spec.offset`/`spec.default`) ignore `frame` entirely (also
+/// `error.RangeNotApplicable` under `.range`, same as the ranking family -
+/// SQL does not accept a frame clause on either) and read a fixed physical
+/// position `spec.offset` rows before (`lag`) or after (`lead`) each row
+/// within the sorted partition sequence (first-seen order when `orderBy` is
+/// empty) - unlike the running-aggregate family, a tie on `orderBy` does NOT
+/// collapse two rows into one lookup target; each remains its own distinct
+/// position. A row whose target position falls outside the partition gets
+/// `spec.default` (SQL NULL, `.none`, unless the caller set one).
+///
+/// `first_value`/`last_value` DO respect `frame`, unlike `lag`/`lead`: the
+/// frame's start is always the partition's first row regardless of
+/// `FrameMode` (so `first_value` is constant across a whole partition), but
+/// the frame's END for `last_value` follows the running-aggregate family's
+/// same split - `.rows` ends the frame at the CURRENT row (so `last_value`
+/// degenerates to that row's own field, SQL's well-known gotcha), `.range`
+/// ends it at the current row's whole tie group, and no `orderBy` at all
+/// means the whole-partition frame (so `last_value` is the last row in
+/// first-seen order, same value for every row). Requires `field` like the
+/// running-aggregate family (`error.MissingField` if omitted).
+///
+/// `nth_value` shares that same frame (start/end identical to
+/// `first_value`/`last_value`) but reads the `spec.offset`-th row (1-based)
+/// counted from the frame's start rather than always the first or always the
+/// last; a position past the frame's current end gets `spec.default`.
+///
+/// Caller owns the returned slice; free each `Value` (`.free(allocator)`)
+/// and the slice itself, or call `freeResults`.
+pub fn compute(allocator: Allocator, rows: []const Row, partitionBy: []const []const u8, orderBy: []const querymod.OrderBy, spec: WindowSpec, frame: FrameMode) WindowError![]Value {
+    const partitions = try partitionedAndSorted(allocator, rows, partitionBy, orderBy);
+    defer {
+        for (partitions) |p| allocator.free(p);
+        allocator.free(partitions);
+    }
+    return computeOneSpec(allocator, rows, partitions, orderBy, spec, frame);
+}
+