diff --git a/src/protocol.zig b/src/protocol.zig index e4c55f8..235d4c7 100644 --- a/src/protocol.zig +++ b/src/protocol.zig @@ -1419,3 +1419,154 @@ pub fn encodeAggGroupRows(allocator: Allocator, groups: []const aggregatemod.Gro return buf.toOwnedSlice(); } +/// Decodes a grouped `run` result into independently owned `GroupRow`s - +/// callers must free each with `GroupRow.deinit`, then the slice itself +/// (see `freeAggGroupRows`). +pub fn decodeAggGroupRows(allocator: Allocator, payload: []const u8) ![]aggregatemod.GroupRow { + var c = Cursor{ .data = payload }; + const count = try c.takeU32(); + var list = try std.ArrayList(aggregatemod.GroupRow).initCapacity(allocator, count); + errdefer { + for (list.items) |*g| g.deinit(allocator); + list.deinit(); + } + var i: u32 = 0; + while (i < count) : (i += 1) { + const group = try takeDecodedValue(allocator, &c); + // Armed for the rest of THIS iteration's block only - Zig disarms an + // errdefer once the block it was declared in exits normally, and + // `appendAssumeCapacity` below is that block's last statement, so a + // later iteration's failure can never reach back and double-free an + // already-appended `group` (the exact class of bug `aggregate.zig`'s + // `GroupBy.observe` hit and fixed - see its comment for the details). + errdefer group.free(allocator); + const valueCount = try c.takeU32(); + var values = try allocator.alloc(rowmod.Value, valueCount); + errdefer allocator.free(values); + var filled: u32 = 0; + errdefer for (values[0..filled]) |v| v.free(allocator); + var j: u32 = 0; + while (j < valueCount) : (j += 1) { + values[j] = try takeDecodedValue(allocator, &c); + filled += 1; + } + list.appendAssumeCapacity(.{ .group = group, .values = values }); + } + return list.toOwnedSlice(); +} + +pub fn freeAggGroupRows(allocator: Allocator, groups: []aggregatemod.GroupRow) void { + for (groups) |*g| g.deinit(allocator); + allocator.free(groups); +} + +// --- table window --- + +/// Wire request for a remote `table window`: a table name, the LIST of +/// `window.zig` `WindowSpec`s to compute (one or more columns sharing the +/// same `PARTITION BY`/`ORDER BY`/`frame` - `window.computeMany`'s shape, +/// mirroring `table_aggregate`'s `specs` list), the `PARTITION BY`/`ORDER +/// BY` field lists (either possibly empty, same "no partition"/"no order" +/// meaning as local `table window`'s `partitionBy`/`orderBy`), `offset`/ +/// `limit` windowing the PRINTED rows in the same partition-then-order walk +/// `window.orderedIndices` produces (not raw scan order, not `computeMany`'s +/// per-row results, which stay aligned 1:1 with every matching row +/// regardless of paging - see `handleTableWindow`'s doc), and an `expr.zig` +/// predicate text filtering rows BEFORE annotation (same +/// empty-means-every-row convention every other `table_*` opcode's +/// `exprText` already uses). `orderBy` reuses `querymod.OrderBy`'s wire +/// shape (descending byte + length-prefixed field name) exactly like +/// `encodeTableAggregate`'s. `specs` is a byte count followed by +/// `(kind byte, length-prefixed field)` pairs, same shape `partitionBy` +/// already uses for its own count-prefixed field list. `frame` is a single +/// byte (`@intFromEnum(windowmod.FrameMode)`) written right before +/// `exprText` - after every other fixed-size field but before the +/// unbounded, unprefixed trailing text, same placement +/// `encodeTableAggregate`'s `havingText` uses relative to ITS trailing +/// `exprText`. +/// Writes a `WindowSpec`'s `offset`/`default` fields, shared by `lag`/`lead` +/// (offset = rows back/forward from the current row) and `nth_value` +/// (offset = 1-based position within the frame) - both pairs share the same +/// wire shape, so one writer covers all three `WindowKind`s. `default` uses +/// the tag+payload shape `row.zig`'s `encodeValue` does, but written +/// directly (not through `putEncodedValue`, which allocates a throwaway +/// buffer via `rowmod.encodeValue` only to copy it right back out) since a +/// `WindowSpec.default` is always a small scalar or short borrowed string, +/// never worth a heap round trip. +fn putLagLeadExtra(w: encode.ByteWriter, offset: u64, default: rowmod.Value) !void { + try w.putU64(offset); + try w.byte(@intFromEnum(@as(rowmod.ValueTag, default))); + switch (default) { + .none => {}, + .boolean => |v| try w.byte(if (v) 1 else 0), + .int => |v| try w.putU64(@bitCast(v)), + .float => |v| try w.putU64(@bitCast(v)), + .text, .bytes => |s| try putLenPrefixed(w, s), + } +} + +/// Reads back what `putLagLeadExtra` wrote. `.text`/`.bytes` are returned as +/// slices BORROWED from `c`'s backing payload (never allocated) - the same +/// convention `WindowSpec.field`/`TableWindowArgs.partitionBy` already use, +/// safe because the payload buffer outlives the whole request handler. +fn takeLagLeadExtra(c: *Cursor) !struct { offset: u64, default: rowmod.Value } { + const offset = try c.takeU64(); + const tagByte = try c.takeByte(); + const tag = std.meta.intToEnum(rowmod.ValueTag, tagByte) catch return error.Malformed; + const default: rowmod.Value = switch (tag) { + .none => .{ .none = {} }, + .boolean => .{ .boolean = (try c.takeByte()) != 0 }, + .int => .{ .int = @bitCast(try c.takeU64()) }, + .float => .{ .float = @bitCast(try c.takeU64()) }, + .text => .{ .text = try c.takeLenPrefixed() }, + .bytes => .{ .bytes = try c.takeLenPrefixed() }, + }; + return .{ .offset = offset, .default = default }; +} + +pub fn encodeTableWindow(allocator: Allocator, table: []const u8, specs: []const windowmod.WindowSpec, partitionBy: []const []const u8, orderBy: []const querymod.OrderBy, offset: u32, limit: u32, exprText: []const u8, frame: windowmod.FrameMode) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + try putLenPrefixed(w, table); + try w.byte(@intCast(specs.len)); + for (specs) |spec| { + try w.byte(@intFromEnum(spec.kind)); + try putLenPrefixed(w, spec.field); + if (spec.kind == .lag or spec.kind == .lead or spec.kind == .nth_value) try putLagLeadExtra(w, spec.offset, spec.default); + } + try w.byte(@intCast(partitionBy.len)); + for (partitionBy) |field| try putLenPrefixed(w, field); + try w.byte(@intCast(orderBy.len)); + for (orderBy) |ob| { + try w.byte(if (ob.descending) 1 else 0); + try putLenPrefixed(w, ob.field); + } + try w.putU32(offset); + try w.putU32(limit); + try w.byte(@intFromEnum(frame)); + try w.slice(exprText); + return buf.toOwnedSlice(); +} + +pub const TableWindowArgs = struct { + table: []const u8, + /// Borrows field-name slices from the decoded payload but owns the + /// slice itself, same convention as `partitionBy`/`orderBy` below. + /// Always non-empty for a well-formed request (`decodeTableWindow` + /// itself does not enforce this - `handleTableWindow` rejects an empty + /// list as `bad_request`, same as `table_aggregate`'s empty-specs + /// check). + specs: []const windowmod.WindowSpec = &.{}, + /// Borrows field-name slices from the decoded payload but owns the + /// slice itself, same convention as `TableAggregateArgs.orderBy`. Empty + /// means "one implicit partition" (the whole result set), same as local + /// `table window`'s default. + partitionBy: []const []const u8 = &.{}, + orderBy: []const querymod.OrderBy = &.{}, + offset: u32 = 0, + limit: u32 = 0, + frame: windowmod.FrameMode = .rows, + exprText: []const u8, +}; +