feat: add handleTableAggregate to server

src/server.zig

@@ -1243,3 +1243,143 @@ pub const Server = struct {
         };
         defer right.deinit();
 
+        var leftParsed: ?exprmod.Parsed = null;
+        defer if (leftParsed) |*p| p.deinit();
+        if (args.leftExprText.len > 0) {
+            leftParsed = exprmod.parse(self.allocator, args.leftExprText) catch |err| {
+                return writeErr(writer, .bad_request, @errorName(err));
+            };
+        }
+        const leftPred: ?querymod.Predicate = if (leftParsed) |p| p.predicate else null;
+
+        var rightParsed: ?exprmod.Parsed = null;
+        defer if (rightParsed) |*p| p.deinit();
+        if (args.rightExprText.len > 0) {
+            rightParsed = exprmod.parse(self.allocator, args.rightExprText) catch |err| {
+                return writeErr(writer, .bad_request, @errorName(err));
+            };
+        }
+        const rightPred: ?querymod.Predicate = if (rightParsed) |p| p.predicate else null;
+
+        var pairs = std.ArrayList(protocol.JoinPair).init(self.allocator);
+        defer {
+            for (pairs.items) |*p| {
+                if (p.left) |*l| l.deinit();
+                if (p.right) |*r| r.deinit();
+            }
+            pairs.deinit();
+        }
+
+        const page = querymod.Page{ .offset = args.offset, .limit = args.limit };
+
+        switch (args.mode) {
+            .inner => {
+                const joined = joinmod.innerJoin(self.allocator, &left, args.leftField, leftPred, &right, args.rightField, rightPred) catch |err| {
+                    return writeErr(writer, .internal, @errorName(err));
+                };
+                const paged = joinmod.applyPage(joinmod.JoinedRow, self.allocator, joined, page) catch |err| {
+                    return writeErr(writer, .internal, @errorName(err));
+                };
+                defer self.allocator.free(paged);
+                for (paged) |p| try pairs.append(.{ .left = p.left, .right = p.right });
+            },
+            .left => {
+                const joined = joinmod.leftOuterJoin(self.allocator, &left, args.leftField, leftPred, &right, args.rightField, rightPred) catch |err| {
+                    return writeErr(writer, .internal, @errorName(err));
+                };
+                const paged = joinmod.applyPage(joinmod.OuterJoinedRow, self.allocator, joined, page) catch |err| {
+                    return writeErr(writer, .internal, @errorName(err));
+                };
+                defer self.allocator.free(paged);
+                for (paged) |p| try pairs.append(.{ .left = p.left, .right = p.right });
+            },
+            .right => {
+                const joined = joinmod.rightOuterJoin(self.allocator, &left, args.leftField, leftPred, &right, args.rightField, rightPred) catch |err| {
+                    return writeErr(writer, .internal, @errorName(err));
+                };
+                const paged = joinmod.applyPage(joinmod.RightOuterJoinedRow, self.allocator, joined, page) catch |err| {
+                    return writeErr(writer, .internal, @errorName(err));
+                };
+                defer self.allocator.free(paged);
+                for (paged) |p| try pairs.append(.{ .left = p.left, .right = p.right });
+            },
+            .full => {
+                const joined = joinmod.fullOuterJoin(self.allocator, &left, args.leftField, leftPred, &right, args.rightField, rightPred) catch |err| {
+                    return writeErr(writer, .internal, @errorName(err));
+                };
+                const paged = joinmod.applyPage(joinmod.FullOuterJoinedRow, self.allocator, joined, page) catch |err| {
+                    return writeErr(writer, .internal, @errorName(err));
+                };
+                defer self.allocator.free(paged);
+                for (paged) |p| try pairs.append(.{ .left = p.left, .right = p.right });
+            },
+        }
+
+        const out = protocol.encodeJoinedRows(self.allocator, pairs.items) catch |err| {
+            return writeErr(writer, .internal, @errorName(err));
+        };
+        defer self.allocator.free(out);
+        try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), out);
+    }
+
+    /// Executes a `table_aggregate` request: opens the named table over the
+    /// same `*Db` the rest of the server serves, filters rows with the same
+    /// `expr.zig` grammar `table_select`/`table_join` already use (empty
+    /// text selects every row), then hands the filtered rows to
+    /// `aggregate.zig` - `runGlobal` when `args.groupField` is empty (the
+    /// network counterpart of the CLI's `*` sentinel for "no GROUP BY"),
+    /// `run` otherwise. When `args.havingText` is non-empty, it is parsed
+    /// with the same `expr.zig` grammar and applied AFTER aggregation but
+    /// BEFORE paging - `globalMatchesHaving` on the `runGlobal` path (an
+    /// excluded single result comes back as zero values, not an error) or
+    /// `filterHaving` on the grouped path - then, on the grouped path only,
+    /// a non-empty `args.orderBy` sorts the HAVING-narrowed groups via
+    /// `aggregate.sortGroupsBy` (an unknown order field is `bad_request`,
+    /// same treatment as a malformed HAVING expr) - then windowed by
+    /// `args.offset`/`args.limit` through `aggregate.applyPageGroups` (0/0 =
+    /// unlimited, ignored for the `runGlobal` path since that always
+    /// produces at most one result row, same as `args.orderBy`). Two distinct wire shapes come back
+    /// (`encodeAggGlobalResult` vs `encodeAggGroupRows`) because the client
+    /// already knows which one it asked for (it is the one that chose
+    /// `groupField`), so no extra discriminator byte is needed on the wire -
+    /// the client just decodes with whichever matches its own request, the
+    /// same reasoning `Client.tableAggregateGlobal`/`tableAggregateGroups`
+    /// document. A bad table name, malformed expr, or a spec referencing a
+    /// non-numeric field (`error.NonNumericField`/`error.Overflow` from
+    /// `aggregate.zig`) is a client mistake (`bad_request`), not a server
+    /// fault.
+    fn handleTableAggregate(self: *Server, payload: []const u8, writer: anytype) !void {
+        const args = protocol.decodeTableAggregate(self.allocator, payload) catch {
+            return writeErr(writer, .bad_request, "malformed table_aggregate payload");
+        };
+        defer self.allocator.free(args.specs);
+        defer self.allocator.free(args.orderBy);
+
+        var table = querymod.Table.open(self.db, self.allocator, args.table) catch |err| {
+            return writeErr(writer, .bad_request, @errorName(err));
+        };
+        defer table.deinit();
+
+        var parsed: ?exprmod.Parsed = null;
+        defer if (parsed) |*p| p.deinit();
+        if (args.exprText.len > 0) {
+            parsed = exprmod.parse(self.allocator, args.exprText) catch |err| {
+                return writeErr(writer, .bad_request, @errorName(err));
+            };
+        }
+        const pred: ?querymod.Predicate = if (parsed) |p| p.predicate else null;
+
+        var havingParsed: ?exprmod.Parsed = null;
+        defer if (havingParsed) |*p| p.deinit();
+        if (args.havingText.len > 0) {
+            havingParsed = exprmod.parse(self.allocator, args.havingText) catch |err| {
+                return writeErr(writer, .bad_request, @errorName(err));
+            };
+        }
+        const havingPred: ?querymod.Predicate = if (havingParsed) |p| p.predicate else null;
+
+        const rows = table.select(pred) catch |err| {
+            return writeErr(writer, .internal, @errorName(err));
+        };
+        defer querymod.freeRows(self.allocator, rows);
+