feat: add decodeAggGroupRows and freeAggGroupRows to protocol

This commit is contained in:
dualfroz committed 2026-09-07 20:53:02 +02:00
1 parent ff764a9697
commit 272b55d669
1 file changed
+151
+151
View File
@@ -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,
};