From ff764a9697c2d7a67cb35f90492819b8a96ddef4 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Mon, 7 Sep 2026 20:48:30 +0200 Subject: [PATCH] feat: add encodeTableAggregate and TableAggregateArgs to protocol --- src/protocol.zig | 158 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 158 insertions(+) diff --git a/src/protocol.zig b/src/protocol.zig index 52fecc4..e4c55f8 100644 --- a/src/protocol.zig +++ b/src/protocol.zig @@ -1261,3 +1261,161 @@ pub fn freeJoinPairs(allocator: Allocator, pairs: []JoinPair) void { // --- table aggregate --- +/// Wire request for a remote `table aggregate`: a table name, an optional +/// GROUP BY field (empty means no grouping - the network counterpart of the +/// CLI's `*` sentinel for `table aggregate *`, chosen client-side so +/// the wire itself needs no separate mode byte), the list of +/// `aggregate.zig` specs to compute, and an `expr.zig` predicate text +/// filtering rows BEFORE aggregation runs - the same empty-means-every-row +/// convention `table_select`'s `exprText` already uses. +/// `offset`/`limit` window the result GROUPS (not the input rows) when +/// `groupField` is non-empty, the same `join.applyPage`/`Table.selectPage` +/// convention as `table_join`'s wire format - 0/0 means unlimited. Ignored +/// server-side when `groupField` is empty (`runGlobal` always produces +/// exactly one result row, nothing to page). +/// `orderBy` is a `query.OrderBy` sort key list (possibly empty) applied +/// server-side to the result GROUPS via `aggregate.sortGroupsBy`, after +/// HAVING narrows them and before `offset`/`limit` window them - same +/// convention and wire shape (count byte + descending byte + length-prefixed +/// field name, per column) as `encodeTableSelect`'s `orderBy`. Each +/// `orderBy[i].field` must name either `groupField` or one of `specs`' +/// `specLabel`s (validated server-side by `sortGroupsBy`, not here). +/// Meaningless (and ignored server-side) for the ungrouped `groupField.len +/// == 0` path, same as `offset`/`limit` already are - `runGlobal` always +/// produces exactly one result row, nothing to order. +/// +/// `havingText` is a second, independent `expr.zig` predicate text +/// (empty means no HAVING) filtering the result GROUPS AFTER aggregation, +/// evaluated server-side before `offset`/`limit` window them - the wire +/// counterpart of the CLI's `having ` suffix. Encoded length-prefixed +/// (not "rest of payload" like `exprText`) precisely because `exprText` +/// still needs to occupy that final slot. +pub fn encodeTableAggregate(allocator: Allocator, table: []const u8, groupField: []const u8, specs: []const aggregatemod.AggSpec, offset: u32, limit: u32, orderBy: []const querymod.OrderBy, havingText: []const u8, exprText: []const u8) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + try putLenPrefixed(w, table); + try putLenPrefixed(w, groupField); + try w.putU32(@intCast(specs.len)); + for (specs) |spec| { + try w.byte(@intFromEnum(spec.kind)); + try putLenPrefixed(w, spec.field); + } + try w.putU32(offset); + try w.putU32(limit); + try w.byte(@intCast(orderBy.len)); + for (orderBy) |ob| { + try w.byte(if (ob.descending) 1 else 0); + try putLenPrefixed(w, ob.field); + } + try putLenPrefixed(w, havingText); + try w.slice(exprText); + return buf.toOwnedSlice(); +} + +pub const TableAggregateArgs = struct { + table: []const u8, + groupField: []const u8, + /// Borrows `field` slices from the decoded payload but owns the slice + /// itself - caller frees with `allocator.free(args.specs)`, same shape + /// as `decodeBatch`'s returned list. + specs: []aggregatemod.AggSpec, + offset: u32 = 0, + limit: u32 = 0, + /// Empty means no ORDER BY - see `encodeTableAggregate`'s doc. Borrows + /// `field` slices from the decoded payload but owns the slice itself, + /// same convention as `specs`/`TableSelectArgs.orderBy`. + orderBy: []const querymod.OrderBy = &.{}, + /// Empty means no HAVING clause - see `encodeTableAggregate`'s doc. + havingText: []const u8, + exprText: []const u8, +}; + +pub fn decodeTableAggregate(allocator: Allocator, payload: []const u8) !TableAggregateArgs { + var c = Cursor{ .data = payload }; + const table = try c.takeLenPrefixed(); + const groupField = try c.takeLenPrefixed(); + const count = try c.takeU32(); + var specs = try std.ArrayList(aggregatemod.AggSpec).initCapacity(allocator, count); + errdefer specs.deinit(); + var i: u32 = 0; + while (i < count) : (i += 1) { + const kindByte = try c.takeByte(); + const kind = std.meta.intToEnum(aggregatemod.AggKind, kindByte) catch return error.Malformed; + const field = try c.takeLenPrefixed(); + specs.appendAssumeCapacity(.{ .kind = kind, .field = field }); + } + const offset = try c.takeU32(); + const limit = try c.takeU32(); + const orderCount = try c.takeByte(); + var orderBy = try std.ArrayList(querymod.OrderBy).initCapacity(allocator, orderCount); + errdefer orderBy.deinit(); + var j: u8 = 0; + while (j < orderCount) : (j += 1) { + const descending = try c.takeByte(); + const field = try c.takeLenPrefixed(); + orderBy.appendAssumeCapacity(.{ .field = field, .descending = descending != 0 }); + } + const havingText = try c.takeLenPrefixed(); + const exprText = c.data[c.pos..]; + return .{ .table = table, .groupField = groupField, .specs = try specs.toOwnedSlice(), .offset = offset, .limit = limit, .orderBy = try orderBy.toOwnedSlice(), .havingText = havingText, .exprText = exprText }; +} + +fn putEncodedValue(allocator: Allocator, w: encode.ByteWriter, value: rowmod.Value) !void { + const encoded = try rowmod.encodeValue(allocator, value); + defer allocator.free(encoded); + try putLenPrefixed(w, encoded); +} + +fn takeDecodedValue(allocator: Allocator, c: *Cursor) !rowmod.Value { + const raw = try c.takeLenPrefixed(); + return rowmod.decodeValue(allocator, raw) catch return error.Malformed; +} + +/// Encodes a `runGlobal` (no `GROUP BY`) result: exactly `specs.len` values, +/// same order as the request's specs. +pub fn encodeAggGlobalResult(allocator: Allocator, values: []const rowmod.Value) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + try w.putU32(@intCast(values.len)); + for (values) |v| try putEncodedValue(allocator, w, v); + return buf.toOwnedSlice(); +} + +/// Decodes a `runGlobal` result into independently owned `Value`s - callers +/// must free with `freeAggValues`. +pub fn decodeAggGlobalResult(allocator: Allocator, payload: []const u8) ![]rowmod.Value { + var c = Cursor{ .data = payload }; + const count = try c.takeU32(); + var list = try std.ArrayList(rowmod.Value).initCapacity(allocator, count); + errdefer { + for (list.items) |v| v.free(allocator); + list.deinit(); + } + var i: u32 = 0; + while (i < count) : (i += 1) list.appendAssumeCapacity(try takeDecodedValue(allocator, &c)); + return list.toOwnedSlice(); +} + +pub fn freeAggValues(allocator: Allocator, values: []rowmod.Value) void { + for (values) |v| v.free(allocator); + allocator.free(values); +} + +/// Encodes a grouped `run` result: one entry per `GroupRow` (the group-key +/// value, then exactly `specs.len` result values), in the same first-seen +/// order `GroupBy.finish()` produced them. +pub fn encodeAggGroupRows(allocator: Allocator, groups: []const aggregatemod.GroupRow) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + try w.putU32(@intCast(groups.len)); + for (groups) |g| { + try putEncodedValue(allocator, w, g.group); + try w.putU32(@intCast(g.values.len)); + for (g.values) |v| try putEncodedValue(allocator, w, v); + } + return buf.toOwnedSlice(); +} +