diff --git a/src/aggregate.zig b/src/aggregate.zig new file mode 100644 index 0000000..6af3ab7 --- /dev/null +++ b/src/aggregate.zig @@ -0,0 +1,1338 @@ +const std = @import("std"); +const rowmod = @import("row.zig"); +const querymod = @import("query.zig"); + +const Allocator = std.mem.Allocator; +const Value = rowmod.Value; +const Row = rowmod.Row; +const Page = querymod.Page; +const Predicate = querymod.Predicate; + +/// GROUP BY / aggregate engine over already-materialized `Row` slices (the +/// same currency `join.zig` uses): callers filter and fetch rows through +/// `Table.select`/`Table.scanAll` first, then feed them here one at a time. +/// Keeping this decoupled from `query.zig`'s `Table` (rather than adding a +/// method on it) mirrors the existing split between the storage-facing +/// `Table` and the purely row-shaped `join.zig`. +pub const AggKind = enum { + /// Counts rows. `field.len == 0` counts every observed row + /// (`COUNT(*)`); otherwise only rows where `field` is present + /// (`COUNT(field)`). + count, + sum, + avg, + min, + max, + /// Counts distinct non-none values of `field` (SQL's + /// `COUNT(DISTINCT field)`). Always requires a `field` - there is no + /// bare `COUNT(DISTINCT *)` form, mirroring `sum`/`avg`/`min`/`max`. + count_distinct, +}; + +pub const AggSpec = struct { + /// Ignored for `.count` with `COUNT(*)` semantics (pass `""`). + field: []const u8, + kind: AggKind, +}; + +pub const AggregateError = error{ + /// `.sum`/`.avg` observed a field holding a non-numeric `Value` + /// (anything but `.int`/`.float`). + NonNumericField, + /// The running integer sum of a `.sum`/`.avg` accumulator would not fit + /// in `i64`. Never silently wraps or drops precision by switching to + /// float behind the caller's back. + Overflow, + /// `sortGroupsBy` was asked to order by a field that names neither the + /// GROUP BY key nor any `specLabel` of the aggregate results - the + /// ORDER BY counterpart of `expr.parse`'s "malformed input is an error, + /// not a guess" convention `filterHaving` already applies to HAVING. + UnknownOrderField, +} || Allocator.Error; + +/// One row of `GroupBy.finish()`'s result: `group` is the distinct group-key +/// value, `values` holds one result `Value` per input `AggSpec`, same order. +pub const GroupRow = struct { + group: Value, + values: []Value, + + pub fn deinit(self: *GroupRow, allocator: Allocator) void { + self.group.free(allocator); + for (self.values) |v| v.free(allocator); + allocator.free(self.values); + } +}; + +fn freeGroupRows(allocator: Allocator, rows: []GroupRow) void { + for (rows) |*r| r.deinit(allocator); + allocator.free(rows); +} + +/// Trims a `GroupBy.finish()`/`run()` result to a `query.zig` `Page` window +/// (0 = unlimited, the same convention `Table.selectPage`/`join.applyPage` +/// use) - the GROUP BY counterpart of pagination for `table_aggregate`/ +/// `remote-aggregate`. Grouping has already had to materialize every +/// distinct group to know how many there are, so - exactly like +/// `join.applyPage` - there is no cheaper way to honor `offset`/`limit` +/// than windowing the already-complete result; this always runs after +/// `run()` returns, never instead of it. Not a generic +/// `join.applyPage(GroupRow, ...)` call because `GroupRow.deinit` takes an +/// `allocator` argument (it owns heap payloads on `.group`/`.values`) +/// where `join.applyPage`'s `T.deinit()` takes none. +/// +/// Takes ownership of `rows`: frees the backing slice, `.deinit()`s every +/// row that falls outside `[offset, offset+limit)`, and returns a freshly +/// allocated slice holding only the kept rows in their original (first-seen) +/// order. Callers own the returned slice and the rows in it exactly as they +/// would have owned `rows`. +pub fn applyPageGroups(allocator: Allocator, rows: []GroupRow, page: Page) ![]GroupRow { + const total = rows.len; + const start = @min(page.offset, total); + const end = if (page.limit == 0) total else @min(start + page.limit, total); + + for (rows[0..start]) |*r| r.deinit(allocator); + for (rows[end..total]) |*r| r.deinit(allocator); + + const kept = end - start; + const result = allocator.alloc(GroupRow, kept) catch |err| { + for (rows[start..end]) |*r| r.deinit(allocator); + allocator.free(rows); + return err; + }; + @memcpy(result, rows[start..end]); + allocator.free(rows); + return result; +} + +/// The row field name a `GroupRow`/global aggregate result is exposed under +/// for HAVING and CLI output purposes: bare `AggKind` name for `field.len == +/// 0` (`COUNT(*)`), otherwise `"kind_field"`. Must stay in sync with +/// `main.zig`'s `printAggLabel`, which prints this same format directly to a +/// writer without allocating - `filterHaving`/`globalMatchesHaving` need an +/// owned string instead because it becomes a `Row` field name, so this is a +/// separate (but format-identical) helper rather than a shared one. +pub fn specLabel(allocator: Allocator, spec: AggSpec) Allocator.Error![]u8 { + if (spec.field.len == 0) return allocator.dupe(u8, @tagName(spec.kind)); + return std.fmt.allocPrint(allocator, "{s}_{s}", .{ @tagName(spec.kind), spec.field }); +} + +/// Builds the synthetic probe `Row` a HAVING predicate is evaluated against: +/// one field per `spec` (named via `specLabel`) holding that spec's result +/// `Value` for this group, plus (when `groupField.len > 0`) one field named +/// `groupField` holding the group's own key - so a HAVING clause can +/// reference an aggregate result (`count`, `sum_score`, ...) or the group key +/// itself through the exact same `expr.zig` grammar `WHERE` already uses. +/// Does not take ownership of `groupValue`/`values` - `Row.set` dupes. +fn buildHavingProbe(allocator: Allocator, groupField: []const u8, groupValue: ?Value, specs: []const AggSpec, values: []const Value) !Row { + var probe = Row.init(allocator); + errdefer probe.deinit(); + if (groupValue) |gv| try probe.set(groupField, gv); + for (specs, values) |spec, v| { + const label = try specLabel(allocator, spec); + defer allocator.free(label); + try probe.set(label, v); + } + return probe; +} + +/// Filters an already-grouped `run()` result by a HAVING predicate - the +/// GROUP BY counterpart of `Predicate.evaluate` on plain rows, applied AFTER +/// aggregation instead of before it (the same WHERE/HAVING split SQL makes). +/// Takes ownership of `groups` (same convention as `applyPageGroups`): every +/// group that fails `having` is `.deinit()`-ed here, the rest are moved +/// (first-seen order preserved) into a freshly allocated result slice. +/// Callers typically run this between `run()` and `applyPageGroups` - HAVING +/// narrows which groups exist before LIMIT/OFFSET windows them, exactly like +/// SQL evaluates `HAVING` before `LIMIT`. +pub fn filterHaving(allocator: Allocator, groups: []GroupRow, groupField: []const u8, specs: []const AggSpec, having: Predicate) AggregateError![]GroupRow { + var kept: std.ArrayListUnmanaged(GroupRow) = .{}; + var i: usize = 0; + // `groups[0..i)` is always fully accounted for by this point: each one + // either moved into `kept` (accepted) or was already `.deinit()`-ed + // in-loop (rejected) - touching it again here would double-free. Only + // `groups[i..]` (the current index, mid-evaluation when this errdefer + // fires, plus everything not yet reached) is still owned by `groups` + // and needs disposing, the same "always dispose of the input on every + // path" contract `applyPageGroups` upholds for its own OOM case. + errdefer { + for (kept.items) |*g| g.deinit(allocator); + kept.deinit(allocator); + for (groups[i..]) |*g| g.deinit(allocator); + allocator.free(groups); + } + + while (i < groups.len) : (i += 1) { + var probe = try buildHavingProbe(allocator, groupField, groups[i].group, specs, groups[i].values); + defer probe.deinit(); + if (having.evaluate(&probe)) { + try kept.append(allocator, groups[i]); + } else { + groups[i].deinit(allocator); + } + } + allocator.free(groups); + return kept.toOwnedSlice(allocator); +} + +/// HAVING for the ungrouped (`runGlobal`, CLI's `*` sentinel) aggregate path: +/// evaluates `having` against the same kind of synthetic probe row +/// `filterHaving` builds, minus any group-key field (there is no group). +/// Reports only whether the single global result should be kept - unlike +/// `filterHaving` it never frees `values`, since `runGlobal` always produces +/// exactly one result and the caller (which already owns it via +/// `freeGlobalResult`) decides whether to print it or not. +pub fn globalMatchesHaving(allocator: Allocator, specs: []const AggSpec, values: []const Value, having: Predicate) AggregateError!bool { + var probe = try buildHavingProbe(allocator, "", null, specs, values); + defer probe.deinit(); + return having.evaluate(&probe); +} + +/// Which part of a `GroupRow` a single `OrderBy` column reads from - +/// resolved once per column by `resolveSortKeys` rather than re-matching +/// `ob.field` against `groupField`/every `specLabel` on every comparison +/// `sortGroupsBy`'s sort makes. +const SortKey = union(enum) { + group, + value: usize, +}; + +/// Resolves each `order[i].field` to either the group key itself +/// (`groupField`, e.g. `city`) or the index of the `specs` entry whose +/// `specLabel` matches (e.g. `sum_score`) - the same two-namespace lookup +/// `buildHavingProbe` exposes to a HAVING predicate, done once up front +/// instead of building a probe `Row` per comparison. An `order[i].field` +/// matching neither is `error.UnknownOrderField` rather than silently +/// sorting as if every row tied on that column. Caller owns the returned +/// slice. +fn resolveSortKeys(allocator: Allocator, groupField: []const u8, specs: []const AggSpec, order: []const querymod.OrderBy) AggregateError![]SortKey { + var keys = try allocator.alloc(SortKey, order.len); + errdefer allocator.free(keys); + for (order, 0..) |ob, i| { + if (groupField.len > 0 and std.mem.eql(u8, ob.field, groupField)) { + keys[i] = .group; + continue; + } + var found: ?usize = null; + for (specs, 0..) |spec, j| { + const label = try specLabel(allocator, spec); + defer allocator.free(label); + if (std.mem.eql(u8, label, ob.field)) { + found = j; + break; + } + } + keys[i] = .{ .value = found orelse return error.UnknownOrderField }; + } + return keys; +} + +/// Stable in-place sort of `groups` by `order` - the GROUP BY counterpart of +/// `query.zig`'s `sortRowsBy`, applied to already-aggregated `GroupRow`s +/// instead of plain `Row`s so an ORDER BY column can name either the group +/// key (`city`) or an aggregate result (`sum_score`), the same two things a +/// HAVING predicate can reference. Columns compare left to right exactly +/// like `sortRowsBy`/SQL's `ORDER BY a, b`: the first entry decides unless +/// it ties, in which case the next breaks it. `order.len == 0` is a no-op +/// (the common case - most `table aggregate` calls have no ORDER BY). +/// Does not take ownership of `groups` - unlike `filterHaving`/ +/// `applyPageGroups`, this never frees or reallocates any group, only +/// reorders the slice in place, so callers keep using their own `defer +/// freeGroupRows(...)` around the call exactly as if it were never made. +/// Typically run after `filterHaving` and before `applyPageGroups` - ORDER +/// BY sees the HAVING-narrowed set and decides what LIMIT/OFFSET's window +/// actually keeps, the same clause order SQL evaluates. +pub fn sortGroupsBy(allocator: Allocator, groups: []GroupRow, groupField: []const u8, specs: []const AggSpec, order: []const querymod.OrderBy) AggregateError!void { + if (order.len == 0) return; + const keys = try resolveSortKeys(allocator, groupField, specs, order); + defer allocator.free(keys); + + const Ctx = struct { + order: []const querymod.OrderBy, + keys: []const SortKey, + + fn lessThan(ctx: @This(), a: GroupRow, b: GroupRow) bool { + for (ctx.order, ctx.keys) |ob, key| { + const av = switch (key) { + .group => a.group, + .value => |j| a.values[j], + }; + const bv = switch (key) { + .group => b.group, + .value => |j| b.values[j], + }; + const cmp = rowmod.compare(av, bv); + if (cmp == .eq) continue; + return if (ob.descending) cmp == .gt else cmp == .lt; + } + return false; + } + }; + std.sort.block(GroupRow, groups, Ctx{ .order = order, .keys = keys }, Ctx.lessThan); +} + +const AccState = struct { + kind: AggKind, + field: []const u8, + n: u64 = 0, + sumInt: i64 = 0, + sumFloat: f64 = 0, + allInt: bool = true, + minv: ?Value = null, + maxv: ?Value = null, + /// `.count_distinct` only: encoded-`Value` bytes seen so far, one entry + /// per distinct value (owns each key, same encode-as-hashmap-key trick + /// `GroupBy.observe` uses for the group key itself). + distinct: std.StringHashMapUnmanaged(void) = .{}, + + fn numericFieldValue(self: *const AccState, row: *const Row) ?Value { + return row.get(self.field); + } + + fn observe(self: *AccState, allocator: Allocator, row: *const Row) AggregateError!void { + switch (self.kind) { + .count => { + if (self.field.len == 0) { + self.n += 1; + } else if (row.get(self.field) != null) { + self.n += 1; + } + }, + .sum, .avg => { + const v = self.numericFieldValue(row) orelse return; + switch (v) { + .int => |i| { + if (self.allInt) { + self.sumInt = std.math.add(i64, self.sumInt, i) catch return error.Overflow; + } else { + self.sumFloat += @floatFromInt(i); + } + }, + .float => |f| { + if (self.allInt) { + // First float observed after an int-only run: + // migrate the exact integer sum so far into the + // float accumulator once, then continue in float. + self.sumFloat = @floatFromInt(self.sumInt); + self.allInt = false; + } + self.sumFloat += f; + }, + else => return error.NonNumericField, + } + self.n += 1; + }, + .min => { + const v = row.get(self.field) orelse return; + if (self.minv == null or rowmod.compare(v, self.minv.?) == .lt) { + if (self.minv) |old| old.free(allocator); + self.minv = try v.dupe(allocator); + } + }, + .max => { + const v = row.get(self.field) orelse return; + if (self.maxv == null or rowmod.compare(v, self.maxv.?) == .gt) { + if (self.maxv) |old| old.free(allocator); + self.maxv = try v.dupe(allocator); + } + }, + .count_distinct => { + const v = row.get(self.field) orelse return; + const key = try rowmod.encodeValue(allocator, v); + const res = try self.distinct.getOrPut(allocator, key); + if (res.found_existing) allocator.free(key); + }, + } + } + + /// Produces the final result `Value`, transferring ownership of any + /// heap payload (`.min`/`.max` on text/bytes fields) out of this + /// accumulator - a subsequent `deinit` sees `null` and frees nothing, + /// so the same payload is never freed twice. `.count_distinct` instead + /// frees its whole tracking set right here (the result is a plain + /// `.int`, nothing to hand ownership of), which is why this needs + /// `allocator` - both call sites (`GroupBy.finish`, `runGlobal`) skip a + /// separate `deinit` call after `take()` entirely, so this is the only + /// place that set is ever freed on the success path. + fn take(self: *AccState, allocator: Allocator) Value { + return switch (self.kind) { + .count => .{ .int = @intCast(self.n) }, + .sum => if (self.allInt) .{ .int = self.sumInt } else .{ .float = self.sumFloat }, + .avg => blk: { + if (self.n == 0) break :blk .{ .none = {} }; + const total: f64 = if (self.allInt) @floatFromInt(self.sumInt) else self.sumFloat; + break :blk .{ .float = total / @as(f64, @floatFromInt(self.n)) }; + }, + .min => blk: { + const v = self.minv orelse Value{ .none = {} }; + self.minv = null; + break :blk v; + }, + .max => blk: { + const v = self.maxv orelse Value{ .none = {} }; + self.maxv = null; + break :blk v; + }, + .count_distinct => blk: { + const result = Value{ .int = @intCast(self.distinct.count()) }; + var it = self.distinct.keyIterator(); + while (it.next()) |k| allocator.free(k.*); + self.distinct.deinit(allocator); + self.distinct = .{}; + break :blk result; + }, + }; + } + + fn deinit(self: *AccState, allocator: Allocator) void { + if (self.minv) |v| v.free(allocator); + if (self.maxv) |v| v.free(allocator); + var it = self.distinct.keyIterator(); + while (it.next()) |k| allocator.free(k.*); + self.distinct.deinit(allocator); + } +}; + +const Bundle = struct { + groupValue: Value, + accs: []AccState, + + fn deinit(self: *Bundle, allocator: Allocator) void { + self.groupValue.free(allocator); + for (self.accs) |*a| a.deinit(allocator); + allocator.free(self.accs); + } +}; + +/// Streaming GROUP BY accumulator: call `observe` once per candidate row +/// (already filtered by whatever `Predicate` the caller wants applied), then +/// `finish` to materialize one `GroupRow` per distinct group-key value seen, +/// in first-seen order (stable, not sorted - callers that want sorted +/// output can sort the returned slice by `.group` using `row.compare`). +pub const GroupBy = struct { + allocator: Allocator, + groupField: []const u8, + specs: []const AggSpec, + map: std.StringHashMapUnmanaged(*Bundle) = .{}, + order: std.ArrayListUnmanaged([]const u8) = .{}, + + pub fn init(allocator: Allocator, groupField: []const u8, specs: []const AggSpec) GroupBy { + return .{ .allocator = allocator, .groupField = groupField, .specs = specs }; + } + + /// Safe to call after `finish()` (which empties the internal maps) or on + /// a `GroupBy` that was never `finish()`-ed (e.g. `observe` returned an + /// error partway through a batch) - either way, exactly what is still + /// owned gets freed once. + pub fn deinit(self: *GroupBy) void { + var it = self.map.iterator(); + while (it.next()) |entry| { + self.allocator.free(entry.key_ptr.*); + entry.value_ptr.*.deinit(self.allocator); + self.allocator.destroy(entry.value_ptr.*); + } + self.map.deinit(self.allocator); + self.order.deinit(self.allocator); + self.map = .{}; + self.order = .{}; + } + + /// Rows missing `groupField` entirely are excluded from every group + /// (no "null group" bucket) - the same omission `Table.createIndex`'s + /// backfill applies to rows missing an indexed field. + pub fn observe(self: *GroupBy, row: *const Row) AggregateError!void { + const groupVal = row.get(self.groupField) orelse return; + const key = try rowmod.encodeValue(self.allocator, groupVal); + + if (self.map.get(key)) |bundle| { + self.allocator.free(key); + for (bundle.accs) |*a| try a.observe(self.allocator, row); + return; + } + + // Scoped to this block so the errdefers below only guard the + // allocation itself: once `bundle` is handed to `self.map.put` + // just after, ownership has moved there, and a later failure (the + // final observe loop) must NOT free it out from under the map - + // an errdefer spanning the whole function would do exactly that + // and leave a dangling pointer in `self.map`. + const bundle = blk: { + const b = try self.allocator.create(Bundle); + errdefer self.allocator.destroy(b); + const accs = try self.allocator.alloc(AccState, self.specs.len); + errdefer self.allocator.free(accs); + for (self.specs, 0..) |spec, i| accs[i] = .{ .kind = spec.kind, .field = spec.field }; + const dupedGroup = try groupVal.dupe(self.allocator); + b.* = .{ .groupValue = dupedGroup, .accs = accs }; + break :blk b; + }; + + try self.map.put(self.allocator, key, bundle); + try self.order.append(self.allocator, key); + for (bundle.accs) |*a| try a.observe(self.allocator, row); + } + + /// Consumes the accumulated groups into an owned `[]GroupRow` and + /// leaves `self` empty (a subsequent `deinit()` is then a no-op). + pub fn finish(self: *GroupBy) AggregateError![]GroupRow { + var results = try self.allocator.alloc(GroupRow, self.order.items.len); + var filled: usize = 0; + errdefer freeGroupRows(self.allocator, results[0..filled]); + + for (self.order.items) |key| { + const bundle = self.map.get(key).?; + var values = try self.allocator.alloc(Value, bundle.accs.len); + errdefer self.allocator.free(values); + for (bundle.accs, 0..) |*a, j| values[j] = a.take(self.allocator); + results[filled] = .{ .group = bundle.groupValue, .values = values }; + filled += 1; + } + + var it = self.map.iterator(); + while (it.next()) |entry| { + self.allocator.free(entry.key_ptr.*); + self.allocator.free(entry.value_ptr.*.accs); + self.allocator.destroy(entry.value_ptr.*); + } + self.map.deinit(self.allocator); + self.order.deinit(self.allocator); + self.map = .{}; + self.order = .{}; + + return results; + } +}; + +/// Convenience wrapper: groups `rows` by `groupField`, applies `specs`, and +/// returns the finished result in one call. +pub fn run(allocator: Allocator, rows: []const Row, groupField: []const u8, specs: []const AggSpec) AggregateError![]GroupRow { + var gb = GroupBy.init(allocator, groupField, specs); + errdefer gb.deinit(); + for (rows) |*r| try gb.observe(r); + return gb.finish(); +} + +/// A single aggregate row over all of `rows` with no grouping (SQL's +/// `SELECT COUNT(*), SUM(x) FROM t` with no `GROUP BY`) - always produces +/// exactly one result, even for zero input rows (every accumulator still +/// reports its identity result: `count` 0, `sum` 0, `avg`/`min`/`max` +/// `.none`). +pub fn runGlobal(allocator: Allocator, rows: []const Row, specs: []const AggSpec) AggregateError![]Value { + var accs = try allocator.alloc(AccState, specs.len); + defer allocator.free(accs); + for (specs, 0..) |spec, i| accs[i] = .{ .kind = spec.kind, .field = spec.field }; + errdefer for (accs) |*a| a.deinit(allocator); + + for (rows) |*r| { + for (accs) |*a| try a.observe(allocator, r); + } + + var values = try allocator.alloc(Value, specs.len); + for (accs, 0..) |*a, i| values[i] = a.take(allocator); + return values; +} + +pub fn freeGlobalResult(allocator: Allocator, values: []Value) void { + for (values) |v| v.free(allocator); + allocator.free(values); +} + +fn makeRow(allocator: Allocator, fields: []const rowmod.Field) !Row { + var r = Row.init(allocator); + for (fields) |f| try r.set(f.name, f.value); + return r; +} + +fn findGroup(rows: []const GroupRow, group: Value) ?*const GroupRow { + for (rows) |*r| { + if (Value.eql(r.group, group)) return r; + } + return null; +} + +test "count sum avg min max group by text field" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + const data = [_]struct { city: []const u8, score: i64 }{ + .{ .city = "warsaw", .score = 10 }, + .{ .city = "warsaw", .score = 30 }, + .{ .city = "krakow", .score = 5 }, + }; + for (data) |d| { + try rows.append(try makeRow(allocator, &.{ + .{ .name = "city", .value = .{ .text = d.city } }, + .{ .name = "score", .value = .{ .int = d.score } }, + })); + } + + const specs = [_]AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "score", .kind = .sum }, + .{ .field = "score", .kind = .avg }, + .{ .field = "score", .kind = .min }, + .{ .field = "score", .kind = .max }, + }; + const groups = try run(allocator, rows.items, "city", &specs); + defer freeGroupRows(allocator, groups); + + try std.testing.expectEqual(@as(usize, 2), groups.len); + + const warsaw = findGroup(groups, .{ .text = "warsaw" }).?; + try std.testing.expectEqual(@as(i64, 2), warsaw.values[0].int); + try std.testing.expectEqual(@as(i64, 40), warsaw.values[1].int); + try std.testing.expectApproxEqAbs(@as(f64, 20), warsaw.values[2].float, 1e-9); + try std.testing.expectEqual(@as(i64, 10), warsaw.values[3].int); + try std.testing.expectEqual(@as(i64, 30), warsaw.values[4].int); + + const krakow = findGroup(groups, .{ .text = "krakow" }).?; + try std.testing.expectEqual(@as(i64, 1), krakow.values[0].int); + try std.testing.expectEqual(@as(i64, 5), krakow.values[1].int); +} + +test "sum and avg skip rows missing the field, count star still counts them" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + try rows.append(try makeRow(allocator, &.{ + .{ .name = "team", .value = .{ .text = "a" } }, + .{ .name = "points", .value = .{ .int = 7 } }, + })); + // No "points" field on this row - must not participate in sum/avg, but + // must still be counted by COUNT(*). + try rows.append(try makeRow(allocator, &.{ + .{ .name = "team", .value = .{ .text = "a" } }, + })); + + const specs = [_]AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "points", .kind = .count }, + .{ .field = "points", .kind = .sum }, + .{ .field = "points", .kind = .avg }, + }; + const groups = try run(allocator, rows.items, "team", &specs); + defer freeGroupRows(allocator, groups); + + try std.testing.expectEqual(@as(usize, 1), groups.len); + try std.testing.expectEqual(@as(i64, 2), groups[0].values[0].int); // COUNT(*) + try std.testing.expectEqual(@as(i64, 1), groups[0].values[1].int); // COUNT(points) + try std.testing.expectEqual(@as(i64, 7), groups[0].values[2].int); // SUM(points) + try std.testing.expectApproxEqAbs(@as(f64, 7), groups[0].values[3].float, 1e-9); // AVG(points) +} + +test "avg on a group where the field is always missing reports none" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + try rows.append(try makeRow(allocator, &.{.{ .name = "team", .value = .{ .text = "a" } }})); + + const specs = [_]AggSpec{.{ .field = "points", .kind = .avg }}; + const groups = try run(allocator, rows.items, "team", &specs); + defer freeGroupRows(allocator, groups); + + try std.testing.expectEqual(@as(usize, 1), groups.len); + try std.testing.expectEqual(rowmod.ValueTag.none, @as(rowmod.ValueTag, groups[0].values[0])); +} + +test "sum stays exact integer until a float value forces it to switch" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + try rows.append(try makeRow(allocator, &.{ + .{ .name = "g", .value = .{ .int = 1 } }, + .{ .name = "x", .value = .{ .int = 3 } }, + })); + try rows.append(try makeRow(allocator, &.{ + .{ .name = "g", .value = .{ .int = 1 } }, + .{ .name = "x", .value = .{ .int = 4 } }, + })); + + const specs = [_]AggSpec{.{ .field = "x", .kind = .sum }}; + { + const groups = try run(allocator, rows.items, "g", &specs); + defer freeGroupRows(allocator, groups); + // All-int input: exact int result, not a float with rounding risk. + try std.testing.expectEqual(rowmod.ValueTag.int, @as(rowmod.ValueTag, groups[0].values[0])); + try std.testing.expectEqual(@as(i64, 7), groups[0].values[0].int); + } + + try rows.append(try makeRow(allocator, &.{ + .{ .name = "g", .value = .{ .int = 1 } }, + .{ .name = "x", .value = .{ .float = 0.5 } }, + })); + { + const groups = try run(allocator, rows.items, "g", &specs); + defer freeGroupRows(allocator, groups); + try std.testing.expectEqual(rowmod.ValueTag.float, @as(rowmod.ValueTag, groups[0].values[0])); + try std.testing.expectApproxEqAbs(@as(f64, 7.5), groups[0].values[0].float, 1e-9); + } +} + +test "sum reports overflow instead of silently wrapping" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + try rows.append(try makeRow(allocator, &.{ + .{ .name = "g", .value = .{ .int = 1 } }, + .{ .name = "x", .value = .{ .int = std.math.maxInt(i64) } }, + })); + try rows.append(try makeRow(allocator, &.{ + .{ .name = "g", .value = .{ .int = 1 } }, + .{ .name = "x", .value = .{ .int = 1 } }, + })); + + const specs = [_]AggSpec{.{ .field = "x", .kind = .sum }}; + var gb = GroupBy.init(allocator, "g", &specs); + defer gb.deinit(); + try gb.observe(&rows.items[0]); + try std.testing.expectError(error.Overflow, gb.observe(&rows.items[1])); +} + +test "sum on a non-numeric field errors without leaking accumulated state" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + try rows.append(try makeRow(allocator, &.{ + .{ .name = "g", .value = .{ .int = 1 } }, + .{ .name = "x", .value = .{ .text = "nope" } }, + })); + + const specs = [_]AggSpec{.{ .field = "x", .kind = .sum }}; + var gb = GroupBy.init(allocator, "g", &specs); + defer gb.deinit(); + try std.testing.expectError(error.NonNumericField, gb.observe(&rows.items[0])); +} + +test "min and max use the total order across owned text values" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + const names = [_][]const u8{ "delta", "alpha", "charlie", "bravo" }; + for (names) |n| { + try rows.append(try makeRow(allocator, &.{ + .{ .name = "g", .value = .{ .int = 1 } }, + .{ .name = "name", .value = .{ .text = n } }, + })); + } + + const specs = [_]AggSpec{ + .{ .field = "name", .kind = .min }, + .{ .field = "name", .kind = .max }, + }; + const groups = try run(allocator, rows.items, "g", &specs); + defer freeGroupRows(allocator, groups); + + try std.testing.expectEqualStrings("alpha", groups[0].values[0].text); + try std.testing.expectEqualStrings("delta", groups[0].values[1].text); +} + +test "count_distinct counts unique field values once per group, repeats collapsed" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + const data = [_]struct { team: []const u8, tag: []const u8 }{ + .{ .team = "a", .tag = "red" }, + .{ .team = "a", .tag = "blue" }, + .{ .team = "a", .tag = "red" }, // repeat, must not double count + .{ .team = "b", .tag = "green" }, + }; + for (data) |d| { + try rows.append(try makeRow(allocator, &.{ + .{ .name = "team", .value = .{ .text = d.team } }, + .{ .name = "tag", .value = .{ .text = d.tag } }, + })); + } + + const specs = [_]AggSpec{.{ .field = "tag", .kind = .count_distinct }}; + const groups = try run(allocator, rows.items, "team", &specs); + defer freeGroupRows(allocator, groups); + + const a = findGroup(groups, .{ .text = "a" }).?; + try std.testing.expectEqual(@as(i64, 2), a.values[0].int); + const b = findGroup(groups, .{ .text = "b" }).?; + try std.testing.expectEqual(@as(i64, 1), b.values[0].int); +} + +test "count_distinct ignores rows missing the field and never counts none as a value" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + try rows.append(try makeRow(allocator, &.{ + .{ .name = "team", .value = .{ .text = "a" } }, + .{ .name = "tag", .value = .{ .text = "red" } }, + })); + // No "tag" field on this row - must not create a phantom distinct value. + try rows.append(try makeRow(allocator, &.{ + .{ .name = "team", .value = .{ .text = "a" } }, + })); + + const specs = [_]AggSpec{.{ .field = "tag", .kind = .count_distinct }}; + const groups = try run(allocator, rows.items, "team", &specs); + defer freeGroupRows(allocator, groups); + + try std.testing.expectEqual(@as(usize, 1), groups.len); + try std.testing.expectEqual(@as(i64, 1), groups[0].values[0].int); +} + +test "count_distinct on zero observed values reports zero, not none" { + const allocator = std.testing.allocator; + const rows: []const Row = &.{}; + const specs = [_]AggSpec{.{ .field = "tag", .kind = .count_distinct }}; + const values = try runGlobal(allocator, rows, &specs); + defer freeGlobalResult(allocator, values); + + try std.testing.expectEqual(@as(i64, 0), values[0].int); +} + +test "count_distinct propagates OutOfMemory without leaking already-tracked distinct values" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + const tags = [_][]const u8{ "red", "blue", "green" }; + for (tags) |t| { + try rows.append(try makeRow(allocator, &.{ + .{ .name = "g", .value = .{ .text = "x" } }, + .{ .name = "tag", .value = .{ .text = t } }, + })); + } + + var failing = std.testing.FailingAllocator.init(allocator, .{ .fail_index = 6 }); + const specs = [_]AggSpec{.{ .field = "tag", .kind = .count_distinct }}; + var gb = GroupBy.init(failing.allocator(), "g", &specs); + defer gb.deinit(); + + var sawError = false; + for (rows.items) |*r| { + gb.observe(r) catch { + sawError = true; + break; + }; + } + try std.testing.expect(sawError); +} + +test "rows missing the group field are excluded from every group" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + try rows.append(try makeRow(allocator, &.{.{ .name = "team", .value = .{ .text = "a" } }})); + try rows.append(try makeRow(allocator, &.{})); // no "team" field at all + + const specs = [_]AggSpec{.{ .field = "", .kind = .count }}; + const groups = try run(allocator, rows.items, "team", &specs); + defer freeGroupRows(allocator, groups); + + try std.testing.expectEqual(@as(usize, 1), groups.len); + try std.testing.expectEqual(@as(i64, 1), groups[0].values[0].int); +} + +test "group order is first-seen, not sorted" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + const order = [_][]const u8{ "z", "a", "m" }; + for (order) |g| { + try rows.append(try makeRow(allocator, &.{.{ .name = "g", .value = .{ .text = g } }})); + } + + const specs = [_]AggSpec{.{ .field = "", .kind = .count }}; + const groups = try run(allocator, rows.items, "g", &specs); + defer freeGroupRows(allocator, groups); + + try std.testing.expectEqual(@as(usize, 3), groups.len); + try std.testing.expectEqualStrings("z", groups[0].group.text); + try std.testing.expectEqualStrings("a", groups[1].group.text); + try std.testing.expectEqualStrings("m", groups[2].group.text); +} + +test "runGlobal aggregates every row into a single result with no grouping" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + var i: i64 = 1; + while (i <= 4) : (i += 1) { + try rows.append(try makeRow(allocator, &.{.{ .name = "x", .value = .{ .int = i } }})); + } + + const specs = [_]AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "x", .kind = .sum }, + .{ .field = "x", .kind = .max }, + }; + const values = try runGlobal(allocator, rows.items, &specs); + defer freeGlobalResult(allocator, values); + + try std.testing.expectEqual(@as(i64, 4), values[0].int); + try std.testing.expectEqual(@as(i64, 10), values[1].int); + try std.testing.expectEqual(@as(i64, 4), values[2].int); +} + +test "runGlobal on zero rows still reports identity results, not an error" { + const allocator = std.testing.allocator; + const rows: []const Row = &.{}; + const specs = [_]AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "x", .kind = .sum }, + .{ .field = "x", .kind = .max }, + }; + const values = try runGlobal(allocator, rows, &specs); + defer freeGlobalResult(allocator, values); + + try std.testing.expectEqual(@as(i64, 0), values[0].int); + try std.testing.expectEqual(@as(i64, 0), values[1].int); + try std.testing.expectEqual(rowmod.ValueTag.none, @as(rowmod.ValueTag, values[2])); +} + +test "GroupBy propagates OutOfMemory without leaking any accumulated group" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + const groupNames = [_][]const u8{ "a", "b", "c" }; + for (groupNames) |g| { + try rows.append(try makeRow(allocator, &.{ + .{ .name = "g", .value = .{ .text = g } }, + .{ .name = "x", .value = .{ .int = 1 } }, + })); + } + + var failing = std.testing.FailingAllocator.init(allocator, .{ .fail_index = 6 }); + const specs = [_]AggSpec{.{ .field = "x", .kind = .sum }}; + var gb = GroupBy.init(failing.allocator(), "g", &specs); + defer gb.deinit(); + + var sawError = false; + for (rows.items) |*r| { + gb.observe(r) catch { + sawError = true; + break; + }; + } + try std.testing.expect(sawError); +} + +fn fourGroupRows(allocator: Allocator) ![]GroupRow { + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + const groupNames = [_][]const u8{ "a", "b", "c", "d" }; + for (groupNames) |g| { + try rows.append(try makeRow(allocator, &.{.{ .name = "g", .value = .{ .text = g } }})); + } + const specs = [_]AggSpec{.{ .field = "", .kind = .count }}; + return run(allocator, rows.items, "g", &specs); +} + +fn firstLettersOf(rows: []const GroupRow) [4]u8 { + var out: [4]u8 = undefined; + for (rows, 0..) |r, i| out[i] = r.group.text[0]; + return out; +} + +test "applyPageGroups with offset and limit keeps a middle window in first-seen order" { + const allocator = std.testing.allocator; + const groups = try fourGroupRows(allocator); + try std.testing.expectEqual(@as(usize, 4), groups.len); + + const paged = try applyPageGroups(allocator, groups, .{ .offset = 1, .limit = 2 }); + defer freeGroupRows(allocator, paged); + + try std.testing.expectEqual(@as(usize, 2), paged.len); + try std.testing.expectEqualStrings("b", paged[0].group.text); + try std.testing.expectEqualStrings("c", paged[1].group.text); +} + +test "applyPageGroups with limit 0 returns every group, matching unpaged run" { + const allocator = std.testing.allocator; + const groups = try fourGroupRows(allocator); + const paged = try applyPageGroups(allocator, groups, .{}); + defer freeGroupRows(allocator, paged); + + try std.testing.expectEqualSlices(u8, &.{ 'a', 'b', 'c', 'd' }, &firstLettersOf(paged)); +} + +test "applyPageGroups with offset past the end returns nothing" { + const allocator = std.testing.allocator; + const groups = try fourGroupRows(allocator); + const paged = try applyPageGroups(allocator, groups, .{ .offset = 50 }); + defer freeGroupRows(allocator, paged); + + try std.testing.expectEqual(@as(usize, 0), paged.len); +} + +test "applyPageGroups with limit larger than the remaining groups stops at the end" { + const allocator = std.testing.allocator; + const groups = try fourGroupRows(allocator); + const paged = try applyPageGroups(allocator, groups, .{ .offset = 3, .limit = 100 }); + defer freeGroupRows(allocator, paged); + + try std.testing.expectEqual(@as(usize, 1), paged.len); + try std.testing.expectEqualStrings("d", paged[0].group.text); +} + +test "applyPageGroups propagates OutOfMemory from its own result allocation without leaking any group" { + const allocator = std.testing.allocator; + const groups = try fourGroupRows(allocator); + try std.testing.expectEqual(@as(usize, 4), groups.len); + + var failing = std.testing.FailingAllocator.init(allocator, .{ .fail_index = 0 }); + try std.testing.expectError(error.OutOfMemory, applyPageGroups(failing.allocator(), groups, .{ .offset = 1, .limit = 2 })); + // No explicit assertion beyond this: `testing.allocator`'s leak detector + // at process exit is the real check - every one of the 4 `GroupRow`s (2 + // kept by the window, 2 trimmed outside it) must have been + // `.deinit(allocator)`-ed on the error path, and the `rows` backing + // slice itself freed, or this test fails on leak. Every `GroupRow` here + // was built by `run()` using the real `testing.allocator`, so the error + // path frees through that same allocator regardless of which allocator + // `applyPageGroups` was asked to allocate its OWN result slice with - + // the exact isolation `join.zig`'s analogous OOM test documents. +} + +fn threeCityGroups(allocator: Allocator) ![]GroupRow { + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + const data = [_]struct { city: []const u8, score: i64 }{ + .{ .city = "warsaw", .score = 10 }, + .{ .city = "warsaw", .score = 30 }, + .{ .city = "krakow", .score = 5 }, + .{ .city = "gdansk", .score = 100 }, + .{ .city = "gdansk", .score = 100 }, + .{ .city = "gdansk", .score = 100 }, + }; + for (data) |d| { + try rows.append(try makeRow(allocator, &.{ + .{ .name = "city", .value = .{ .text = d.city } }, + .{ .name = "score", .value = .{ .int = d.score } }, + })); + } + const specs = [_]AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "score", .kind = .sum }, + }; + return run(allocator, rows.items, "city", &specs); +} + +test "filterHaving keeps only groups whose aggregate result matches the predicate" { + const allocator = std.testing.allocator; + const groups = try threeCityGroups(allocator); + // warsaw: count=2 sum=40, krakow: count=1 sum=5, gdansk: count=3 sum=300 + const specs = [_]AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "score", .kind = .sum }, + }; + const having = Predicate{ .gt = .{ .field = "count", .value = .{ .int = 1 } } }; + const kept = try filterHaving(allocator, groups, "city", &specs, having); + defer freeGroupRows(allocator, kept); + + try std.testing.expectEqual(@as(usize, 2), kept.len); + try std.testing.expectEqualStrings("warsaw", kept[0].group.text); + try std.testing.expectEqualStrings("gdansk", kept[1].group.text); +} + +test "filterHaving can reference the underlying aggregate field name (sum_field)" { + const allocator = std.testing.allocator; + const groups = try threeCityGroups(allocator); + const specs = [_]AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "score", .kind = .sum }, + }; + const having = Predicate{ .gte = .{ .field = "sum_score", .value = .{ .int = 100 } } }; + const kept = try filterHaving(allocator, groups, "city", &specs, having); + defer freeGroupRows(allocator, kept); + + try std.testing.expectEqual(@as(usize, 1), kept.len); + try std.testing.expectEqualStrings("gdansk", kept[0].group.text); +} + +test "filterHaving can also reference the group key itself, not just an aggregate result" { + const allocator = std.testing.allocator; + const groups = try threeCityGroups(allocator); + const specs = [_]AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "score", .kind = .sum }, + }; + const having = Predicate{ .ne = .{ .field = "city", .value = .{ .text = "krakow" } } }; + const kept = try filterHaving(allocator, groups, "city", &specs, having); + defer freeGroupRows(allocator, kept); + + try std.testing.expectEqual(@as(usize, 2), kept.len); + try std.testing.expectEqualStrings("warsaw", kept[0].group.text); + try std.testing.expectEqualStrings("gdansk", kept[1].group.text); +} + +test "filterHaving that rejects every group returns an empty, still-valid slice" { + const allocator = std.testing.allocator; + const groups = try threeCityGroups(allocator); + const specs = [_]AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "score", .kind = .sum }, + }; + const having = Predicate{ .gt = .{ .field = "count", .value = .{ .int = 1000 } } }; + const kept = try filterHaving(allocator, groups, "city", &specs, having); + defer freeGroupRows(allocator, kept); + + try std.testing.expectEqual(@as(usize, 0), kept.len); +} + +test "filterHaving composes with applyPageGroups: HAVING narrows before LIMIT/OFFSET windows" { + const allocator = std.testing.allocator; + const groups = try threeCityGroups(allocator); + const specs = [_]AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "score", .kind = .sum }, + }; + // Only warsaw and gdansk survive HAVING; a limit of 1 then keeps just + // the first of those two (warsaw, first-seen), not the first of all + // three original groups. + const having = Predicate{ .gt = .{ .field = "count", .value = .{ .int = 1 } } }; + const filtered = try filterHaving(allocator, groups, "city", &specs, having); + const paged = try applyPageGroups(allocator, filtered, .{ .limit = 1 }); + defer freeGroupRows(allocator, paged); + + try std.testing.expectEqual(@as(usize, 1), paged.len); + try std.testing.expectEqualStrings("warsaw", paged[0].group.text); +} + +test "filterHaving propagates OutOfMemory without leaking any group, kept or rejected" { + const allocator = std.testing.allocator; + const groups = try threeCityGroups(allocator); + const specs = [_]AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "score", .kind = .sum }, + }; + const having = Predicate{ .gt = .{ .field = "count", .value = .{ .int = 1 } } }; + + var failing = std.testing.FailingAllocator.init(allocator, .{ .fail_index = 0 }); + try std.testing.expectError(error.OutOfMemory, filterHaving(failing.allocator(), groups, "city", &specs, having)); + // Leak detector on `std.testing.allocator` at process exit is the real + // check here (same pattern as `applyPageGroups`'s analogous OOM test): + // every one of the 3 groups `threeCityGroups` built with the real + // allocator must still get freed on this early failure, even though + // `filterHaving` itself was asked to allocate its OWN bookkeeping + // (`kept`) through the failing allocator. +} + +test "globalMatchesHaving reports true or false without freeing the caller's values" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + var i: i64 = 1; + while (i <= 5) : (i += 1) { + try rows.append(try makeRow(allocator, &.{.{ .name = "x", .value = .{ .int = i } }})); + } + const specs = [_]AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "x", .kind = .sum }, + }; + const values = try runGlobal(allocator, rows.items, &specs); + defer freeGlobalResult(allocator, values); + + const passing = Predicate{ .eq = .{ .field = "count", .value = .{ .int = 5 } } }; + try std.testing.expect(try globalMatchesHaving(allocator, &specs, values, passing)); + + const failing = Predicate{ .gt = .{ .field = "sum_x", .value = .{ .int = 1000 } } }; + try std.testing.expect(!try globalMatchesHaving(allocator, &specs, values, failing)); +} + +test "sortGroupsBy orders by the group key ascending by default" { + const allocator = std.testing.allocator; + const groups = try threeCityGroups(allocator); + defer freeGroupRows(allocator, groups); + + const order = [_]querymod.OrderBy{.{ .field = "city" }}; + try sortGroupsBy(allocator, groups, "city", &.{ + .{ .field = "", .kind = .count }, + .{ .field = "score", .kind = .sum }, + }, &order); + + try std.testing.expectEqualStrings("gdansk", groups[0].group.text); + try std.testing.expectEqualStrings("krakow", groups[1].group.text); + try std.testing.expectEqualStrings("warsaw", groups[2].group.text); +} + +test "sortGroupsBy orders by the group key descending when asked" { + const allocator = std.testing.allocator; + const groups = try threeCityGroups(allocator); + defer freeGroupRows(allocator, groups); + + const order = [_]querymod.OrderBy{.{ .field = "city", .descending = true }}; + try sortGroupsBy(allocator, groups, "city", &.{ + .{ .field = "", .kind = .count }, + .{ .field = "score", .kind = .sum }, + }, &order); + + try std.testing.expectEqualStrings("warsaw", groups[0].group.text); + try std.testing.expectEqualStrings("krakow", groups[1].group.text); + try std.testing.expectEqualStrings("gdansk", groups[2].group.text); +} + +test "sortGroupsBy orders by an aggregate result label, not just the group key" { + const allocator = std.testing.allocator; + const groups = try threeCityGroups(allocator); + defer freeGroupRows(allocator, groups); + const specs = [_]AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "score", .kind = .sum }, + }; + + // warsaw sum=40, krakow sum=5, gdansk sum=300 - ascending by sum_score + // is a completely different order than ascending by city. + const order = [_]querymod.OrderBy{.{ .field = "sum_score" }}; + try sortGroupsBy(allocator, groups, "city", &specs, &order); + + try std.testing.expectEqualStrings("krakow", groups[0].group.text); + try std.testing.expectEqualStrings("warsaw", groups[1].group.text); + try std.testing.expectEqualStrings("gdansk", groups[2].group.text); +} + +test "sortGroupsBy breaks ties left to right across multiple order columns" { + const allocator = std.testing.allocator; + var rows = std.ArrayList(Row).init(allocator); + defer { + for (rows.items) |*r| r.deinit(); + rows.deinit(); + } + // Every group has count == 1 (a full tie on the first order column), so + // only the second column (the group key itself) can decide the order - + // built out of first-seen order d,b,a,c specifically so an + // alphabetical result proves the second column actually ran, rather + // than the sort coincidentally matching input order. + const groupNames = [_][]const u8{ "d", "b", "a", "c" }; + for (groupNames) |g| { + try rows.append(try makeRow(allocator, &.{.{ .name = "g", .value = .{ .text = g } }})); + } + const specs = [_]AggSpec{.{ .field = "", .kind = .count }}; + const groups = try run(allocator, rows.items, "g", &specs); + defer freeGroupRows(allocator, groups); + + const order = [_]querymod.OrderBy{ + .{ .field = "count" }, + .{ .field = "g" }, + }; + try sortGroupsBy(allocator, groups, "g", &specs, &order); + + try std.testing.expectEqualStrings("a", groups[0].group.text); + try std.testing.expectEqualStrings("b", groups[1].group.text); + try std.testing.expectEqualStrings("c", groups[2].group.text); + try std.testing.expectEqualStrings("d", groups[3].group.text); +} + +test "sortGroupsBy with an empty order list leaves groups in first-seen order" { + const allocator = std.testing.allocator; + const groups = try threeCityGroups(allocator); + defer freeGroupRows(allocator, groups); + + try sortGroupsBy(allocator, groups, "city", &.{ + .{ .field = "", .kind = .count }, + .{ .field = "score", .kind = .sum }, + }, &.{}); + + try std.testing.expectEqualStrings("warsaw", groups[0].group.text); + try std.testing.expectEqualStrings("krakow", groups[1].group.text); + try std.testing.expectEqualStrings("gdansk", groups[2].group.text); +} + +test "sortGroupsBy rejects a field that names neither the group key nor a known aggregate label" { + const allocator = std.testing.allocator; + const groups = try threeCityGroups(allocator); + defer freeGroupRows(allocator, groups); + + const order = [_]querymod.OrderBy{.{ .field = "not_a_real_field" }}; + try std.testing.expectError(error.UnknownOrderField, sortGroupsBy(allocator, groups, "city", &.{ + .{ .field = "", .kind = .count }, + .{ .field = "score", .kind = .sum }, + }, &order)); +} + +test "sortGroupsBy composes with filterHaving and applyPageGroups: ORDER BY reorders before LIMIT/OFFSET windows" { + const allocator = std.testing.allocator; + const groups = try threeCityGroups(allocator); + const specs = [_]AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "score", .kind = .sum }, + }; + // HAVING keeps only warsaw (count=2, sum=40) and gdansk (count=3, + // sum=300); descending sum_score then puts gdansk first even though + // warsaw was first-seen - a limit of 1 must keep gdansk, not warsaw. + const having = Predicate{ .gt = .{ .field = "count", .value = .{ .int = 1 } } }; + const filtered = try filterHaving(allocator, groups, "city", &specs, having); + const order = [_]querymod.OrderBy{.{ .field = "sum_score", .descending = true }}; + try sortGroupsBy(allocator, filtered, "city", &specs, &order); + const paged = try applyPageGroups(allocator, filtered, .{ .limit = 1 }); + defer freeGroupRows(allocator, paged); + + try std.testing.expectEqual(@as(usize, 1), paged.len); + try std.testing.expectEqualStrings("gdansk", paged[0].group.text); +} + +test "sortGroupsBy propagates OutOfMemory without leaking or freeing the caller's groups" { + const allocator = std.testing.allocator; + const groups = try threeCityGroups(allocator); + defer freeGroupRows(allocator, groups); + const specs = [_]AggSpec{ + .{ .field = "", .kind = .count }, + .{ .field = "score", .kind = .sum }, + }; + + var failing = std.testing.FailingAllocator.init(allocator, .{ .fail_index = 0 }); + const order = [_]querymod.OrderBy{.{ .field = "city" }}; + try std.testing.expectError(error.OutOfMemory, sortGroupsBy(failing.allocator(), groups, "city", &specs, &order)); + + // `sortGroupsBy` never took ownership of `groups` (unlike + // `filterHaving`), so the outer `defer freeGroupRows(allocator, ...)` + // above must still see every group intact and still first-seen order + // (untouched, not partially sorted) - both the leak detector on + // `std.testing.allocator` and this ordering check are the real + // assertions here. + try std.testing.expectEqualStrings("warsaw", groups[0].group.text); + try std.testing.expectEqualStrings("krakow", groups[1].group.text); + try std.testing.expectEqualStrings("gdansk", groups[2].group.text); +} diff --git a/src/auth.zig b/src/auth.zig new file mode 100644 index 0000000..59d8fba --- /dev/null +++ b/src/auth.zig @@ -0,0 +1,450 @@ +const std = @import("std"); +const dbmod = @import("db.zig"); + +const Allocator = std.mem.Allocator; +const Db = dbmod.Db; +const Hmac256 = std.crypto.auth.hmac.sha2.HmacSha256; + +/// Username/password authentication and coarse-grained role authorization, +/// layered over the same Db as tables/indexes/collections under its own +/// namespace tag 'A' (row.zig/index.zig use 'R'/'I'/'M', collections.zig +/// uses 'X', so no keyspace ever collides): +/// "A" "U\0" username -> UserRecord{role, rounds, salt[16], hash[32]} +/// Passwords are never stored: only a per-user salted PBKDF2-HMAC-SHA256 +/// hash, using `std.crypto.pwhash.pbkdf2` - the same primitive the standard +/// library itself ships for exactly this purpose, not a hand-rolled one. +pub const nsTag: u8 = 'A'; +const userPrefixBytes = [_]u8{ nsTag, 'U', 0 }; +const saltLen = 16; +const hashLen = 32; +const recordLen = 1 + 4 + saltLen + hashLen; + +pub const AuthError = error{ + UserNotFound, + UserExists, + InvalidCredentials, + Malformed, +}; + +/// Roles form a total order, not a set of independent flags: readwrite +/// implies readonly, admin implies both. `atLeast` is the only check any +/// caller (server.zig's dispatch) ever needs. +pub const Role = enum(u8) { + readonly = 0, + readwrite = 1, + admin = 2, + + pub fn fromByte(b: u8) !Role { + return std.meta.intToEnum(Role, b) catch error.Malformed; + } + + pub fn parse(text: []const u8) !Role { + if (std.mem.eql(u8, text, "readonly")) return .readonly; + if (std.mem.eql(u8, text, "readwrite")) return .readwrite; + if (std.mem.eql(u8, text, "admin")) return .admin; + return error.Malformed; + } + + pub fn name(self: Role) []const u8 { + return switch (self) { + .readonly => "readonly", + .readwrite => "readwrite", + .admin => "admin", + }; + } + + pub fn atLeast(self: Role, min: Role) bool { + return @intFromEnum(self) >= @intFromEnum(min); + } +}; + +/// Recommended real-world cost factor for interactive login (OWASP's floor +/// for PBKDF2-HMAC-SHA256 as of 2023). A constant rather than a value +/// baked into `hashPassword` so tests can pass a tiny round count and stay +/// fast, while `main.zig` and any other real caller default to this. +pub const recommendedRounds: u32 = 100_000; + +fn userKey(allocator: Allocator, username: []const u8) ![]u8 { + const out = try allocator.alloc(u8, userPrefixBytes.len + username.len); + @memcpy(out[0..userPrefixBytes.len], &userPrefixBytes); + @memcpy(out[userPrefixBytes.len..], username); + return out; +} + +const UserRecord = struct { + role: Role, + rounds: u32, + salt: [saltLen]u8, + hash: [hashLen]u8, + + fn encode(self: UserRecord) [recordLen]u8 { + var out: [recordLen]u8 = undefined; + out[0] = @intFromEnum(self.role); + std.mem.writeInt(u32, out[1..5], self.rounds, .little); + @memcpy(out[5 .. 5 + saltLen], &self.salt); + @memcpy(out[5 + saltLen .. recordLen], &self.hash); + return out; + } + + fn decode(data: []const u8) !UserRecord { + if (data.len != recordLen) return error.Malformed; + var rec: UserRecord = undefined; + rec.role = try Role.fromByte(data[0]); + rec.rounds = std.mem.readInt(u32, data[1..5], .little); + @memcpy(&rec.salt, data[5 .. 5 + saltLen]); + @memcpy(&rec.hash, data[5 + saltLen .. recordLen]); + return rec; + } +}; + +fn randomSalt() [saltLen]u8 { + var salt: [saltLen]u8 = undefined; + std.crypto.random.bytes(&salt); + return salt; +} + +fn hashPassword(password: []const u8, salt: [saltLen]u8, rounds: u32) ![hashLen]u8 { + var out: [hashLen]u8 = undefined; + try std.crypto.pwhash.pbkdf2(&out, password, &salt, rounds, Hmac256); + return out; +} + +/// Fixed salt/rounds used to cost a lookup against a username that does not +/// exist, so `verify` spends roughly the same CPU whether the account is +/// missing or the password was merely wrong - closing the most obvious +/// timing side channel for username enumeration over the network (the +/// key comparison itself still only runs on the "user exists" path, since +/// there is nothing real to compare against otherwise). +const dummySalt = [_]u8{0} ** saltLen; +const dummyRounds: u32 = 4096; + +pub const UserInfo = struct { + username: []u8, + role: Role, + + pub fn deinit(self: UserInfo, allocator: Allocator) void { + allocator.free(self.username); + } +}; + +pub fn freeUserList(allocator: Allocator, users: []UserInfo) void { + for (users) |u| u.deinit(allocator); + allocator.free(users); +} + +/// Username/password/role management layered over a `*Db`. Every method +/// touches only keys under `userPrefixBytes`; unrelated data in the same +/// Db (tables, collections, raw KV) is never read or written. Not +/// synchronized on its own, same as every other module built over Db - +/// server.zig serializes access with its own connection-handling mutex. +pub const Store = struct { + db: *Db, + + pub fn init(db: *Db) Store { + return .{ .db = db }; + } + + pub fn createUser(self: Store, allocator: Allocator, username: []const u8, password: []const u8, role: Role, rounds: u32) !void { + const key = try userKey(allocator, username); + defer allocator.free(key); + if (try self.db.get(allocator, key)) |existing| { + allocator.free(existing); + return error.UserExists; + } + const salt = randomSalt(); + const record = UserRecord{ + .role = role, + .rounds = rounds, + .salt = salt, + .hash = try hashPassword(password, salt, rounds), + }; + const encoded = record.encode(); + try self.db.put(key, &encoded); + } + + fn mustGet(self: Store, allocator: Allocator, key: []const u8) !UserRecord { + const raw = try self.db.get(allocator, key) orelse return error.UserNotFound; + defer allocator.free(raw); + return UserRecord.decode(raw); + } + + pub fn setPassword(self: Store, allocator: Allocator, username: []const u8, newPassword: []const u8, rounds: u32) !void { + const key = try userKey(allocator, username); + defer allocator.free(key); + var record = try self.mustGet(allocator, key); + record.rounds = rounds; + record.salt = randomSalt(); + record.hash = try hashPassword(newPassword, record.salt, rounds); + const encoded = record.encode(); + try self.db.put(key, &encoded); + } + + pub fn setRole(self: Store, allocator: Allocator, username: []const u8, role: Role) !void { + const key = try userKey(allocator, username); + defer allocator.free(key); + var record = try self.mustGet(allocator, key); + record.role = role; + const encoded = record.encode(); + try self.db.put(key, &encoded); + } + + pub fn deleteUser(self: Store, allocator: Allocator, username: []const u8) !void { + const key = try userKey(allocator, username); + defer allocator.free(key); + _ = try self.mustGet(allocator, key); + try self.db.delete(key); + } + + /// Verifies `password` for `username`, returning the stored role on + /// success. See `dummySalt`/`dummyRounds` for why a missing user still + /// pays a full PBKDF2 pass before failing. The final comparison uses + /// `std.crypto.utils.timingSafeEql` rather than `std.mem.eql`, since an + /// early-exit byte comparison on the hash would leak how many leading + /// bytes matched to a network attacker timing repeated attempts. + pub fn verify(self: Store, allocator: Allocator, username: []const u8, password: []const u8) !Role { + const key = try userKey(allocator, username); + defer allocator.free(key); + const raw = try self.db.get(allocator, key) orelse { + _ = hashPassword(password, dummySalt, dummyRounds) catch {}; + return error.InvalidCredentials; + }; + defer allocator.free(raw); + const record = try UserRecord.decode(raw); + const candidate = try hashPassword(password, record.salt, record.rounds); + if (!std.crypto.utils.timingSafeEql([hashLen]u8, candidate, record.hash)) return error.InvalidCredentials; + return record.role; + } + + /// Whether at least one admin account exists - `main.zig` uses this to + /// refuse turning on server-side auth enforcement until an operator has + /// created a way back in, so enabling `--auth` can never lock everyone + /// out of a fresh database. + pub fn hasAnyAdmin(self: Store, allocator: Allocator) !bool { + var it = try self.db.scanPrefix(allocator, &userPrefixBytes); + defer it.deinit(); + while (try it.next()) |entry| { + const record = UserRecord.decode(entry.value) catch continue; + if (record.role == .admin) return true; + } + return false; + } + + /// Returns every user with their role, independently owned - free with + /// `freeUserList`. Order follows key order (lexicographic by username). + pub fn listUsers(self: Store, allocator: Allocator) ![]UserInfo { + var it = try self.db.scanPrefix(allocator, &userPrefixBytes); + defer it.deinit(); + var out = std.ArrayList(UserInfo).init(allocator); + errdefer { + for (out.items) |u| u.deinit(allocator); + out.deinit(); + } + while (try it.next()) |entry| { + const record = UserRecord.decode(entry.value) catch continue; + const uname = entry.key[userPrefixBytes.len..]; + try out.append(.{ .username = try allocator.dupe(u8, uname), .role = record.role }); + } + return out.toOwnedSlice(); + } +}; + +fn testDb(allocator: Allocator, dir: std.fs.Dir) !Db { + const path = try dir.realpathAlloc(allocator, "."); + defer allocator.free(path); + return Db.open(allocator, path, .{}); +} + +test "role ordering is a total order across readonly/readwrite/admin" { + try std.testing.expect(Role.readwrite.atLeast(.readonly)); + try std.testing.expect(Role.admin.atLeast(.readonly)); + try std.testing.expect(Role.admin.atLeast(.readwrite)); + try std.testing.expect(Role.admin.atLeast(.admin)); + try std.testing.expect(!Role.readonly.atLeast(.readwrite)); + try std.testing.expect(!Role.readwrite.atLeast(.admin)); +} + +test "Role.parse rejects an unknown role name" { + try std.testing.expectEqual(Role.admin, try Role.parse("admin")); + try std.testing.expectError(error.Malformed, Role.parse("superuser")); +} + +test "createUser then verify accepts the correct password" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var db = try testDb(allocator, tmp.dir); + defer db.close(); + + const store = Store.init(&db); + try store.createUser(allocator, "alice", "hunter2", .readwrite, 16); + + const role = try store.verify(allocator, "alice", "hunter2"); + try std.testing.expectEqual(Role.readwrite, role); +} + +test "verify rejects a wrong password without revealing which part was wrong" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var db = try testDb(allocator, tmp.dir); + defer db.close(); + + const store = Store.init(&db); + try store.createUser(allocator, "alice", "hunter2", .readonly, 16); + + try std.testing.expectError(error.InvalidCredentials, store.verify(allocator, "alice", "wrong")); +} + +test "verify rejects an unknown username the same way as a wrong password" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var db = try testDb(allocator, tmp.dir); + defer db.close(); + + const store = Store.init(&db); + try std.testing.expectError(error.InvalidCredentials, store.verify(allocator, "ghost", "anything")); +} + +test "createUser rejects a duplicate username" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var db = try testDb(allocator, tmp.dir); + defer db.close(); + + const store = Store.init(&db); + try store.createUser(allocator, "alice", "hunter2", .readonly, 16); + try std.testing.expectError(error.UserExists, store.createUser(allocator, "alice", "different", .admin, 16)); +} + +test "setPassword replaces the password and invalidates the old one" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var db = try testDb(allocator, tmp.dir); + defer db.close(); + + const store = Store.init(&db); + try store.createUser(allocator, "alice", "old-pass", .readonly, 16); + try store.setPassword(allocator, "alice", "new-pass", 16); + + try std.testing.expectError(error.InvalidCredentials, store.verify(allocator, "alice", "old-pass")); + try std.testing.expectEqual(Role.readonly, try store.verify(allocator, "alice", "new-pass")); +} + +test "setPassword on a missing user reports UserNotFound" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var db = try testDb(allocator, tmp.dir); + defer db.close(); + + const store = Store.init(&db); + try std.testing.expectError(error.UserNotFound, store.setPassword(allocator, "ghost", "x", 16)); +} + +test "setRole changes role without disturbing the password" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var db = try testDb(allocator, tmp.dir); + defer db.close(); + + const store = Store.init(&db); + try store.createUser(allocator, "alice", "hunter2", .readonly, 16); + try store.setRole(allocator, "alice", .admin); + + try std.testing.expectEqual(Role.admin, try store.verify(allocator, "alice", "hunter2")); +} + +test "deleteUser removes the account so verify and mustGet both fail" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var db = try testDb(allocator, tmp.dir); + defer db.close(); + + const store = Store.init(&db); + try store.createUser(allocator, "alice", "hunter2", .readonly, 16); + try store.deleteUser(allocator, "alice"); + + try std.testing.expectError(error.InvalidCredentials, store.verify(allocator, "alice", "hunter2")); + try std.testing.expectError(error.UserNotFound, store.deleteUser(allocator, "alice")); +} + +test "hasAnyAdmin reflects whether any created user carries the admin role" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var db = try testDb(allocator, tmp.dir); + defer db.close(); + + const store = Store.init(&db); + try std.testing.expect(!(try store.hasAnyAdmin(allocator))); + + try store.createUser(allocator, "alice", "hunter2", .readwrite, 16); + try std.testing.expect(!(try store.hasAnyAdmin(allocator))); + + try store.createUser(allocator, "root", "toor", .admin, 16); + try std.testing.expect(try store.hasAnyAdmin(allocator)); +} + +test "listUsers returns every account with its role" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var db = try testDb(allocator, tmp.dir); + defer db.close(); + + const store = Store.init(&db); + try store.createUser(allocator, "alice", "a-pass", .readonly, 16); + try store.createUser(allocator, "bob", "b-pass", .readwrite, 16); + try store.createUser(allocator, "root", "r-pass", .admin, 16); + + const users = try store.listUsers(allocator); + defer freeUserList(allocator, users); + try std.testing.expectEqual(@as(usize, 3), users.len); + + var sawAdmin = false; + for (users) |u| { + if (std.mem.eql(u8, u.username, "root")) { + try std.testing.expectEqual(Role.admin, u.role); + sawAdmin = true; + } + } + try std.testing.expect(sawAdmin); +} + +test "users survive a Db close and reopen" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + + { + var db = try testDb(allocator, tmp.dir); + defer db.close(); + const store = Store.init(&db); + try store.createUser(allocator, "alice", "hunter2", .admin, 16); + } + + var db = try testDb(allocator, tmp.dir); + defer db.close(); + const store = Store.init(&db); + try std.testing.expectEqual(Role.admin, try store.verify(allocator, "alice", "hunter2")); +} + +test "a malformed stored record is reported as Malformed, not a crash" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var db = try testDb(allocator, tmp.dir); + defer db.close(); + + const key = try userKey(allocator, "alice"); + defer allocator.free(key); + try db.put(key, "too short to be a real record"); + + const store = Store.init(&db); + try std.testing.expectError(error.Malformed, store.verify(allocator, "alice", "anything")); +} diff --git a/src/backup.zig b/src/backup.zig new file mode 100644 index 0000000..d43a934 --- /dev/null +++ b/src/backup.zig @@ -0,0 +1,527 @@ +const std = @import("std"); +const db_mod = @import("db.zig"); +const encode = @import("encode.zig"); + +const Allocator = std.mem.Allocator; +const Db = db_mod.Db; +const Snapshot = db_mod.Snapshot; +const WriteBatch = db_mod.WriteBatch; + +// Portable logical backup format for cobalt. Dumps a consistent, point-in-time +// view of the live keyspace (as returned by Snapshot.scan - tombstones already +// resolved) into a self-describing file that can be replayed into any Db. +// +// Layout: +// header: magic("CBLB" 4B) | version(u32) | seq(u64) +// entry*: keyLen(u32) | valueLen(u32) | key | value | entryCrc(u32) +// entryCrc = crc32(keyLen le bytes || valueLen le bytes || key || value) +// footer: magic("CBLE" 4B) | entryCount(u64) | fileCrc(u32) +// fileCrc = crc32 of every byte written before the footer magic +// (header + all entries), rolling-accumulated while writing. +// +// Restore validates the whole file (magics, per-entry crc, trailing count, +// whole-file crc) before touching the target Db, so a corrupt or truncated +// backup never applies a partial write. + +const headerMagic = "CBLB"; +const footerMagic = "CBLE"; +const formatVersion: u32 = 1; + +pub const BackupError = error{ + BadMagic, + UnsupportedVersion, + Truncated, + Corrupt, + CountMismatch, +}; + +pub const BackupInfo = struct { + seq: u64, + entryCount: u64, +}; + +const RollingCrc = struct { + hasher: std.hash.Crc32, + + fn init() RollingCrc { + return .{ .hasher = std.hash.Crc32.init() }; + } + + fn feed(self: *RollingCrc, bytes: []const u8) void { + self.hasher.update(bytes); + } + + fn value(self: RollingCrc) u32 { + var copy = self.hasher; + return copy.final(); + } +}; + +fn writeAllCrc(writer: anytype, roll: *RollingCrc, bytes: []const u8) !void { + try writer.writeAll(bytes); + roll.feed(bytes); +} + +fn putU32Crc(writer: anytype, roll: *RollingCrc, v: u32) !void { + var b: [4]u8 = undefined; + std.mem.writeInt(u32, &b, v, encode.endian); + try writeAllCrc(writer, roll, &b); +} + +fn putU64Crc(writer: anytype, roll: *RollingCrc, v: u64) !void { + var b: [8]u8 = undefined; + std.mem.writeInt(u64, &b, v, encode.endian); + try writeAllCrc(writer, roll, &b); +} + +/// Writes a full, consistent backup of `snapshot` to `path` (created/truncated). +/// Returns the number of entries written. +pub fn dumpToFile(allocator: Allocator, snapshot: Snapshot, path: []const u8) !u64 { + const file = try std.fs.cwd().createFile(path, .{ .truncate = true }); + defer file.close(); + + var bufferedWriter = std.io.bufferedWriter(file.writer()); + const writer = bufferedWriter.writer(); + + var roll = RollingCrc.init(); + try writeAllCrc(writer, &roll, headerMagic); + try putU32Crc(writer, &roll, formatVersion); + try putU64Crc(writer, &roll, snapshot.seq); + + var iter = try snapshot.scan(allocator, null, null); + defer iter.deinit(); + + var count: u64 = 0; + var entryScratch = std.ArrayList(u8).init(allocator); + defer entryScratch.deinit(); + + while (try iter.next()) |entry| { + entryScratch.clearRetainingCapacity(); + var lenBuf: [8]u8 = undefined; + std.mem.writeInt(u32, lenBuf[0..4], @intCast(entry.key.len), encode.endian); + std.mem.writeInt(u32, lenBuf[4..8], @intCast(entry.value.len), encode.endian); + try entryScratch.appendSlice(&lenBuf); + try entryScratch.appendSlice(entry.key); + try entryScratch.appendSlice(entry.value); + const entryCrc = std.hash.Crc32.hash(entryScratch.items); + + try writeAllCrc(writer, &roll, entryScratch.items); + try putU32Crc(writer, &roll, entryCrc); + count += 1; + } + + try writer.writeAll(footerMagic); + var countBuf: [8]u8 = undefined; + std.mem.writeInt(u64, &countBuf, count, encode.endian); + try writer.writeAll(&countBuf); + + const fileCrc = roll.value(); + var crcBuf: [4]u8 = undefined; + std.mem.writeInt(u32, &crcBuf, fileCrc, encode.endian); + try writer.writeAll(&crcBuf); + + try bufferedWriter.flush(); + return count; +} + +const ParsedEntry = struct { + key: []const u8, + value: []const u8, +}; + +/// One pass over a fully-loaded backup buffer: validates structure end to end +/// (magics, per-entry crc, declared count, whole-file crc) without allocating +/// per-entry copies. Returns the entries slice (positions into `data`) plus +/// backup metadata, or a BackupError describing the first problem found. +const ParsedBackup = struct { + seq: u64, + entries: []ParsedEntry, +}; + +fn parse(allocator: Allocator, data: []const u8) !ParsedBackup { + if (data.len < headerMagic.len + 4 + 8) return BackupError.Truncated; + if (!std.mem.eql(u8, data[0..4], headerMagic)) return BackupError.BadMagic; + + const version = encode.readU32(data, 4); + if (version != formatVersion) return BackupError.UnsupportedVersion; + const seq = encode.readU64(data, 8); + + var pos: usize = 16; + var entries = std.ArrayList(ParsedEntry).init(allocator); + errdefer entries.deinit(); + + while (true) { + if (pos + footerMagic.len <= data.len and std.mem.eql(u8, data[pos .. pos + 4], footerMagic)) { + break; + } + if (pos + 8 > data.len) return BackupError.Truncated; + const keyLen = encode.readU32(data, pos); + const valueLen = encode.readU32(data, pos + 4); + const entryStart = pos; + const keyStart = pos + 8; + const valueStart = keyStart + keyLen; + const valueEnd = valueStart + valueLen; + if (valueEnd + 4 > data.len) return BackupError.Truncated; + + const entryBytes = data[entryStart..valueEnd]; + const storedCrc = encode.readU32(data, valueEnd); + const actualCrc = std.hash.Crc32.hash(entryBytes); + if (storedCrc != actualCrc) return BackupError.Corrupt; + + try entries.append(.{ .key = data[keyStart..valueStart], .value = data[valueStart..valueEnd] }); + pos = valueEnd + 4; + } + + if (pos + 4 + 8 + 4 > data.len) return BackupError.Truncated; + const declaredCount = encode.readU64(data, pos + 4); + if (declaredCount != entries.items.len) return BackupError.CountMismatch; + + const storedFileCrc = encode.readU32(data, pos + 4 + 8); + const actualFileCrc = std.hash.Crc32.hash(data[0..pos]); + if (storedFileCrc != actualFileCrc) return BackupError.Corrupt; + + return .{ .seq = seq, .entries = try entries.toOwnedSlice() }; +} + +/// Reads and structurally validates a backup file without applying it, +/// returning its metadata (seq, entry count). Useful to verify a backup +/// before trusting it for disaster recovery. +pub fn inspectFile(allocator: Allocator, path: []const u8) !BackupInfo { + const data = try std.fs.cwd().readFileAlloc(allocator, path, 1 << 34); + defer allocator.free(data); + const parsed = try parse(allocator, data); + defer allocator.free(parsed.entries); + return .{ .seq = parsed.seq, .entryCount = parsed.entries.len }; +} + +/// Restores every entry from the backup at `path` into `db`, in chunks of +/// `chunkSize` applied atomically via WriteBatch. The whole file is validated +/// (magics, per-entry and whole-file crc, declared count) before any chunk is +/// applied, so a corrupt backup fails without mutating the target database. +/// Existing keys in `db` outside the backup are left untouched; keys present +/// in both are overwritten with the backup's value. +pub fn restoreFromFile(allocator: Allocator, db: *Db, path: []const u8, chunkSize: usize) !BackupInfo { + const data = try std.fs.cwd().readFileAlloc(allocator, path, 1 << 34); + defer allocator.free(data); + + const parsed = try parse(allocator, data); + defer allocator.free(parsed.entries); + + var i: usize = 0; + while (i < parsed.entries.len) { + const end = @min(i + chunkSize, parsed.entries.len); + var batch = WriteBatch.init(allocator); + defer batch.deinit(); + for (parsed.entries[i..end]) |entry| { + try batch.put(entry.key, entry.value); + } + try db.apply(&batch); + i = end; + } + + return .{ .seq = parsed.seq, .entryCount = parsed.entries.len }; +} + +test "backup round trip: dump then restore into a fresh db reproduces every key" { + const allocator = std.testing.allocator; + const srcDir = "test-backup-src-1"; + const dstDir = "test-backup-dst-1"; + const backupPath = "test-backup-1.cblb"; + std.fs.cwd().deleteTree(srcDir) catch {}; + std.fs.cwd().deleteTree(dstDir) catch {}; + defer std.fs.cwd().deleteTree(srcDir) catch {}; + defer std.fs.cwd().deleteTree(dstDir) catch {}; + defer std.fs.cwd().deleteFile(backupPath) catch {}; + + var src = try Db.open(allocator, srcDir, .{}); + defer src.close(); + + var keyBuf: [16]u8 = undefined; + var valBuf: [16]u8 = undefined; + var i: usize = 0; + while (i < 200) : (i += 1) { + const key = try std.fmt.bufPrint(&keyBuf, "k{d:0>6}", .{i}); + const value = try std.fmt.bufPrint(&valBuf, "v{d:0>6}", .{i}); + try src.put(key, value); + } + // overwrite one, delete another: backup must reflect resolved state. + try src.put("k000005", "updated"); + try src.delete("k000010"); + + const snap = try src.snapshot(); + const count = try dumpToFile(allocator, snap, backupPath); + try std.testing.expectEqual(@as(u64, 199), count); + + var dst = try Db.open(allocator, dstDir, .{}); + defer dst.close(); + const info = try restoreFromFile(allocator, &dst, backupPath, 32); + try std.testing.expectEqual(@as(u64, 199), info.entryCount); + + i = 0; + while (i < 200) : (i += 1) { + const key = try std.fmt.bufPrint(&keyBuf, "k{d:0>6}", .{i}); + const got = try dst.get(allocator, key); + if (i == 10) { + try std.testing.expect(got == null); + continue; + } + defer allocator.free(got.?); + if (i == 5) { + try std.testing.expectEqualStrings("updated", got.?); + } else { + const expected = try std.fmt.bufPrint(&valBuf, "v{d:0>6}", .{i}); + try std.testing.expectEqualStrings(expected, got.?); + } + } +} + +test "backup of empty db round trips to zero entries" { + const allocator = std.testing.allocator; + const srcDir = "test-backup-src-empty"; + const dstDir = "test-backup-dst-empty"; + const backupPath = "test-backup-empty.cblb"; + std.fs.cwd().deleteTree(srcDir) catch {}; + std.fs.cwd().deleteTree(dstDir) catch {}; + defer std.fs.cwd().deleteTree(srcDir) catch {}; + defer std.fs.cwd().deleteTree(dstDir) catch {}; + defer std.fs.cwd().deleteFile(backupPath) catch {}; + + var src = try Db.open(allocator, srcDir, .{}); + defer src.close(); + const snap = try src.snapshot(); + const count = try dumpToFile(allocator, snap, backupPath); + try std.testing.expectEqual(@as(u64, 0), count); + + const info = try inspectFile(allocator, backupPath); + try std.testing.expectEqual(@as(u64, 0), info.entryCount); + + var dst = try Db.open(allocator, dstDir, .{}); + defer dst.close(); + _ = try restoreFromFile(allocator, &dst, backupPath, 32); + var iter = try dst.scan(allocator, null, null); + defer iter.deinit(); + try std.testing.expect((try iter.next()) == null); +} + +test "backup is point-in-time: writes after snapshot are excluded" { + const allocator = std.testing.allocator; + const srcDir = "test-backup-src-pit"; + const backupPath = "test-backup-pit.cblb"; + std.fs.cwd().deleteTree(srcDir) catch {}; + defer std.fs.cwd().deleteTree(srcDir) catch {}; + defer std.fs.cwd().deleteFile(backupPath) catch {}; + + var src = try Db.open(allocator, srcDir, .{}); + defer src.close(); + try src.put("before", "1"); + const snap = try src.snapshot(); + try src.put("after", "2"); + + const count = try dumpToFile(allocator, snap, backupPath); + try std.testing.expectEqual(@as(u64, 1), count); + + const info = try inspectFile(allocator, backupPath); + try std.testing.expectEqual(snap.seq, info.seq); +} + +test "restore into db with pre-existing disjoint keys merges without touching them" { + const allocator = std.testing.allocator; + const srcDir = "test-backup-src-merge"; + const dstDir = "test-backup-dst-merge"; + const backupPath = "test-backup-merge.cblb"; + std.fs.cwd().deleteTree(srcDir) catch {}; + std.fs.cwd().deleteTree(dstDir) catch {}; + defer std.fs.cwd().deleteTree(srcDir) catch {}; + defer std.fs.cwd().deleteTree(dstDir) catch {}; + defer std.fs.cwd().deleteFile(backupPath) catch {}; + + var src = try Db.open(allocator, srcDir, .{}); + defer src.close(); + try src.put("from-backup", "yes"); + const count = try dumpToFile(allocator, try src.snapshot(), backupPath); + try std.testing.expectEqual(@as(u64, 1), count); + + var dst = try Db.open(allocator, dstDir, .{}); + defer dst.close(); + try dst.put("already-here", "kept"); + + _ = try restoreFromFile(allocator, &dst, backupPath, 32); + + const kept = try dst.get(allocator, "already-here"); + defer allocator.free(kept.?); + try std.testing.expectEqualStrings("kept", kept.?); + + const restored = try dst.get(allocator, "from-backup"); + defer allocator.free(restored.?); + try std.testing.expectEqualStrings("yes", restored.?); +} + +test "restore chunking applies every entry across multiple write batches" { + const allocator = std.testing.allocator; + const srcDir = "test-backup-src-chunks"; + const dstDir = "test-backup-dst-chunks"; + const backupPath = "test-backup-chunks.cblb"; + std.fs.cwd().deleteTree(srcDir) catch {}; + std.fs.cwd().deleteTree(dstDir) catch {}; + defer std.fs.cwd().deleteTree(srcDir) catch {}; + defer std.fs.cwd().deleteTree(dstDir) catch {}; + defer std.fs.cwd().deleteFile(backupPath) catch {}; + + var src = try Db.open(allocator, srcDir, .{}); + defer src.close(); + var keyBuf: [16]u8 = undefined; + var i: usize = 0; + while (i < 97) : (i += 1) { + const key = try std.fmt.bufPrint(&keyBuf, "c{d:0>6}", .{i}); + try src.put(key, "x"); + } + _ = try dumpToFile(allocator, try src.snapshot(), backupPath); + + var dst = try Db.open(allocator, dstDir, .{}); + defer dst.close(); + const info = try restoreFromFile(allocator, &dst, backupPath, 10); + try std.testing.expectEqual(@as(u64, 97), info.entryCount); + + var found: usize = 0; + var iter = try dst.scan(allocator, null, null); + defer iter.deinit(); + while (try iter.next()) |_| found += 1; + try std.testing.expectEqual(@as(usize, 97), found); +} + +test "corrupted entry payload is rejected before any restore write happens" { + const allocator = std.testing.allocator; + const srcDir = "test-backup-src-corrupt"; + const dstDir = "test-backup-dst-corrupt"; + const backupPath = "test-backup-corrupt.cblb"; + std.fs.cwd().deleteTree(srcDir) catch {}; + std.fs.cwd().deleteTree(dstDir) catch {}; + defer std.fs.cwd().deleteTree(srcDir) catch {}; + defer std.fs.cwd().deleteTree(dstDir) catch {}; + defer std.fs.cwd().deleteFile(backupPath) catch {}; + + var src = try Db.open(allocator, srcDir, .{}); + defer src.close(); + try src.put("intact", "value-one"); + try src.put("victim", "value-two"); + _ = try dumpToFile(allocator, try src.snapshot(), backupPath); + + { + const file = try std.fs.cwd().openFile(backupPath, .{ .mode = .read_write }); + defer file.close(); + // flip a byte squarely inside the entry region (well past the 16 byte header). + try file.seekTo(24); + var byte: [1]u8 = undefined; + _ = try file.read(&byte); + try file.seekTo(24); + try file.writeAll(&.{byte[0] ^ 0xFF}); + } + + var dst = try Db.open(allocator, dstDir, .{}); + defer dst.close(); + const result = restoreFromFile(allocator, &dst, backupPath, 32); + try std.testing.expectError(BackupError.Corrupt, result); + + // nothing from the corrupt backup should have been applied: validation + // happens over the whole in-memory buffer before the first WriteBatch. + var iter = try dst.scan(allocator, null, null); + defer iter.deinit(); + try std.testing.expect((try iter.next()) == null); +} + +test "truncated file is rejected as Truncated, not a crash" { + const allocator = std.testing.allocator; + const srcDir = "test-backup-src-trunc"; + const dstDir = "test-backup-dst-trunc"; + const backupPath = "test-backup-trunc.cblb"; + std.fs.cwd().deleteTree(srcDir) catch {}; + std.fs.cwd().deleteTree(dstDir) catch {}; + defer std.fs.cwd().deleteTree(srcDir) catch {}; + defer std.fs.cwd().deleteTree(dstDir) catch {}; + defer std.fs.cwd().deleteFile(backupPath) catch {}; + + var src = try Db.open(allocator, srcDir, .{}); + defer src.close(); + try src.put("a", "b"); + _ = try dumpToFile(allocator, try src.snapshot(), backupPath); + + { + const data = try std.fs.cwd().readFileAlloc(allocator, backupPath, 1 << 20); + defer allocator.free(data); + const truncated = data[0 .. data.len - 5]; + const file = try std.fs.cwd().createFile(backupPath, .{ .truncate = true }); + defer file.close(); + try file.writeAll(truncated); + } + + var dst = try Db.open(allocator, dstDir, .{}); + defer dst.close(); + const result = restoreFromFile(allocator, &dst, backupPath, 32); + try std.testing.expectError(BackupError.Truncated, result); +} + +test "bad magic header is rejected without reading the rest of the file" { + const allocator = std.testing.allocator; + const backupPath = "test-backup-badmagic.cblb"; + defer std.fs.cwd().deleteFile(backupPath) catch {}; + + const file = try std.fs.cwd().createFile(backupPath, .{ .truncate = true }); + { + defer file.close(); + try file.writeAll("XXXX" ++ [_]u8{0} ** 12); + } + + const result = inspectFile(allocator, backupPath); + try std.testing.expectError(BackupError.BadMagic, result); +} + +test "unsupported version is rejected" { + const allocator = std.testing.allocator; + const backupPath = "test-backup-badversion.cblb"; + defer std.fs.cwd().deleteFile(backupPath) catch {}; + + const file = try std.fs.cwd().createFile(backupPath, .{ .truncate = true }); + { + defer file.close(); + var buf: [16]u8 = undefined; + @memcpy(buf[0..4], headerMagic); + std.mem.writeInt(u32, buf[4..8], 99, encode.endian); + std.mem.writeInt(u64, buf[8..16], 0, encode.endian); + try file.writeAll(&buf); + } + + const result = inspectFile(allocator, backupPath); + try std.testing.expectError(BackupError.UnsupportedVersion, result); +} + +test "declared entry count mismatch is caught even if per-entry crcs are valid" { + const allocator = std.testing.allocator; + const srcDir = "test-backup-src-countmismatch"; + const backupPath = "test-backup-countmismatch.cblb"; + std.fs.cwd().deleteTree(srcDir) catch {}; + defer std.fs.cwd().deleteTree(srcDir) catch {}; + defer std.fs.cwd().deleteFile(backupPath) catch {}; + + var src = try Db.open(allocator, srcDir, .{}); + defer src.close(); + try src.put("a", "1"); + try src.put("b", "2"); + _ = try dumpToFile(allocator, try src.snapshot(), backupPath); + + { + const data = try std.fs.cwd().readFileAlloc(allocator, backupPath, 1 << 20); + defer allocator.free(data); + // footer is the last 4(magic)+8(count)+4(crc)=16 bytes; corrupt count only, + // which necessarily also breaks the whole-file crc, but count is checked first. + const countPos = data.len - 12; + var mutated = try allocator.dupe(u8, data); + defer allocator.free(mutated); + std.mem.writeInt(u64, mutated[countPos..][0..8], 999, encode.endian); + const file = try std.fs.cwd().createFile(backupPath, .{ .truncate = true }); + defer file.close(); + try file.writeAll(mutated); + } + + const result = inspectFile(allocator, backupPath); + try std.testing.expectError(BackupError.CountMismatch, result); +} diff --git a/src/client.zig b/src/client.zig new file mode 100644 index 0000000..1217124 --- /dev/null +++ b/src/client.zig @@ -0,0 +1,703 @@ +const std = @import("std"); +const protocol = @import("protocol.zig"); +const dbmod = @import("db.zig"); +const rowmod = @import("row.zig"); +const authmod = @import("auth.zig"); +const ttlmod = @import("ttl.zig"); +const collmod = @import("collections.zig"); +const aggregatemod = @import("aggregate.zig"); +const windowmod = @import("window.zig"); +const querymod = @import("query.zig"); + +const Allocator = std.mem.Allocator; + +pub const ClientError = error{ + ServerBadRequest, + ServerInternal, + ServerConflict, + ServerRateLimited, + ServerUnauthorized, + ServerForbidden, + ServerUserNotFound, + ServerKeyNotFound, + ServerCollectionNotFound, + UnexpectedStatus, +}; + +/// Result of a remote `table export`: the row count `dump.zig`'s exporter +/// reported plus the rendered bytes, independently allocated so they +/// outlive the response frame they arrived in. +pub const RemoteExport = struct { + rowCount: u64, + content: []u8, + + pub fn deinit(self: RemoteExport, allocator: Allocator) void { + allocator.free(self.content); + } +}; + +fn statusToError(status: protocol.Status) ClientError { + return switch (status) { + .bad_request => error.ServerBadRequest, + .internal => error.ServerInternal, + .conflict => error.ServerConflict, + .rate_limited => error.ServerRateLimited, + .unauthorized => error.ServerUnauthorized, + .forbidden => error.ServerForbidden, + else => error.UnexpectedStatus, + }; +} + +/// A blocking client for the wire protocol in `protocol.zig`. Each call +/// writes one request frame and reads exactly one response frame, so a +/// `Client` must not be shared across threads without external locking. +pub const Client = struct { + allocator: Allocator, + stream: std.net.Stream, + /// The session token from the last successful `authenticate` call, if + /// the server had `enableSessions` on (see server.zig) - null on a + /// server without sessions, or before any successful `authenticate`. + /// Owned by the `Client`, freed on `close()` and on every subsequent + /// `authenticate`/`authenticateWithToken` call. + token: ?[]u8 = null, + + pub fn connect(allocator: Allocator, address: std.net.Address) !Client { + const stream = try std.net.tcpConnectToAddress(address); + return .{ .allocator = allocator, .stream = stream }; + } + + pub fn close(self: *Client) void { + if (self.token) |t| self.allocator.free(t); + self.stream.close(); + } + + fn setToken(self: *Client, payload: []const u8) !void { + if (self.token) |old| self.allocator.free(old); + self.token = if (payload.len > 0) try self.allocator.dupe(u8, payload) else null; + } + + /// The session token captured by the last successful `authenticate`, + /// if any (see `token` above). Callers that want to reconnect later + /// without a password should dupe this before the `Client` closes. + pub fn sessionToken(self: *Client) ?[]const u8 { + return self.token; + } + + fn roundTrip(self: *Client, op: protocol.Op, payload: []const u8) !protocol.Frame { + try protocol.writeFrame(self.stream.writer(), @intFromEnum(op), payload); + return protocol.readFrame(self.allocator, self.stream.reader()); + } + + /// Authenticates this connection against a server that has auth + /// enabled (see `server.zig`'s `enableAuth`). Every request on this + /// stream after a successful call is dispatched under the role + /// returned by the server's `auth.Store.verify`; a server without auth + /// enabled always answers `ok` regardless of the credentials given. If + /// the server also has sessions enabled (`enableSessions`), the + /// response carries a fresh token captured into `sessionToken()` - + /// otherwise `sessionToken()` reads null afterwards, same as before + /// this call was ever made. + pub fn authenticate(self: *Client, username: []const u8, password: []const u8) !void { + const payload = try protocol.encodeAuth(self.allocator, username, password); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.auth, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + try self.setToken(frame.payload); + } + + /// Authenticates this connection using a token obtained from another + /// connection's `sessionToken()`, instead of a username/password pair + /// (see server.zig's `auth_token` opcode) - skips the password + /// entirely and resumes the same role until the session expires. A + /// server without sessions enabled accepts any token unconditionally, + /// matching `authenticate`'s behavior against a server without auth. + pub fn authenticateWithToken(self: *Client, token: []const u8) !void { + const payload = try protocol.encodeAuthToken(self.allocator, token); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.auth_token, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + } + + pub fn ping(self: *Client) !void { + var frame = try self.roundTrip(.ping, ""); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + } + + pub fn get(self: *Client, allocator: Allocator, key: []const u8) !?[]u8 { + var frame = try self.roundTrip(.get, key); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + return switch (status) { + .ok => try allocator.dupe(u8, frame.payload), + .not_found => null, + else => statusToError(status), + }; + } + + pub fn set(self: *Client, key: []const u8, value: []const u8) !void { + const payload = try protocol.encodeSet(self.allocator, key, value); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.set, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + } + + pub fn delete(self: *Client, key: []const u8) !void { + var frame = try self.roundTrip(.del, key); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + } + + pub fn scan(self: *Client, allocator: Allocator, args: protocol.ScanArgs) ![]protocol.OwnedEntry { + const payload = try protocol.encodeScan(self.allocator, args); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.scan, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return protocol.decodeEntries(allocator, frame.payload); + } + + pub fn prefix(self: *Client, allocator: Allocator, prefixBytes: []const u8, limit: u32) ![]protocol.OwnedEntry { + const payload = try protocol.encodePrefix(self.allocator, prefixBytes, limit); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.prefix, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return protocol.decodeEntries(allocator, frame.payload); + } + + pub fn stats(self: *Client) !dbmod.Stats { + var frame = try self.roundTrip(.stats, ""); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return protocol.decodeStats(frame.payload); + } + + /// Fetches the server's combined Prometheus text export (its own + /// request/rate-limit/auth/session counters from `Server.registerMetrics` + /// plus `Db.stats()`, in the one payload `Server.handleMetrics` builds) - + /// the network counterpart of the CLI's local-only `stats --prom`. + /// Returned text is independently owned; free with `allocator.free`. + pub fn metricsText(self: *Client, allocator: Allocator) ![]u8 { + var frame = try self.roundTrip(.metrics, ""); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return allocator.dupe(u8, frame.payload); + } + + pub fn flush(self: *Client) !void { + var frame = try self.roundTrip(.flush, ""); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + } + + pub fn compact(self: *Client) !void { + var frame = try self.roundTrip(.compact, ""); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + } + + pub fn batch(self: *Client, ops: []const protocol.BatchOp) !void { + const payload = try protocol.encodeBatch(self.allocator, ops); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.batch, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + } + + /// Remote `table explain`: the wire counterpart of the CLI's local + /// `table explain`, reporting which strategy `query.Table.explain` + /// says a matching `tableSelect` would take against the server's + /// `*Db`, without running the query. `exprText` is the same `expr.zig` + /// grammar as `tableSelect`'s (empty means no predicate, i.e. always a + /// full scan). Returned `Plan` owns its `indexName` - free with + /// `Plan.deinit`. + pub fn tableExplain(self: *Client, allocator: Allocator, table: []const u8, exprText: []const u8) !querymod.Plan { + const payload = try protocol.encodeTableExplain(self.allocator, table, exprText); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.table_explain, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return protocol.decodePlan(allocator, frame.payload); + } + + /// Remote `table update`: the wire counterpart of the CLI's local + /// `table update`, applying `sets` to every row matching `exprText` + /// (empty means every row, the same convention `tableSelect`/ + /// `tableExplain` use) against the server's `*Db` via + /// `query.Table.updateWhere`. Returns the number of rows updated. + pub fn tableUpdate(self: *Client, table: []const u8, sets: []const rowmod.Field, exprText: []const u8) !u64 { + const payload = try protocol.encodeTableUpdate(self.allocator, table, sets, exprText); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.table_update, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return protocol.decodeTtlSweepCount(frame.payload); + } + + /// Remote `table delete`: the wire counterpart of the CLI's local + /// `table delete`, removing every row matching `exprText` (empty means + /// every row, the same convention `tableUpdate`/`tableExplain` use) + /// against the server's `*Db` via `query.Table.deleteWhere`. Returns the + /// number of rows deleted. + pub fn tableDelete(self: *Client, table: []const u8, exprText: []const u8) !u64 { + const payload = try protocol.encodeTableDelete(self.allocator, table, exprText); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.table_delete, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return protocol.decodeTtlSweepCount(frame.payload); + } + + /// Remote `table export`: the wire counterpart of the CLI's local + /// `table export csv|json `, rendering the whole table on + /// the server and returning both the row count and the rendered bytes + /// in one round trip. `content` is a fresh allocation independent of + /// the response frame - free it with `RemoteExport.deinit`. + pub fn tableExport(self: *Client, allocator: Allocator, table: []const u8, format: protocol.DumpFormat) !RemoteExport { + const payload = try protocol.encodeTableExport(self.allocator, table, format); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.table_export, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + const result = try protocol.decodeExportResult(frame.payload); + const content = try allocator.dupe(u8, result.content); + return .{ .rowCount = result.rowCount, .content = content }; + } + + /// Remote `table import`: the wire counterpart of the CLI's local + /// `table import
csv|json `. `content` is data the + /// caller already read into memory (e.g. from a local file) and sends + /// to the server in one frame; the server runs it through `dump.zig`'s + /// importer there. Returns the number of rows imported. + pub fn tableImport(self: *Client, table: []const u8, format: protocol.DumpFormat, content: []const u8) !u64 { + const payload = try protocol.encodeTableImport(self.allocator, table, format, content); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.table_import, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return protocol.decodeTtlSweepCount(frame.payload); + } + + /// Remote `table select`: `exprText` is the same `expr.zig` predicate + /// grammar the local CLI's `table select` accepts (empty selects every + /// row). `offset`/`limit` page the result the same way + /// `query.Table.selectPage` does locally (0 limit = unlimited), so a + /// large remote result does not have to come back in a single frame. + /// `orderBy`, when non-empty, sorts server-side via + /// `Table.selectPageOrdered` before that page window applies (one or + /// more columns applied left to right, ties on earlier columns broken + /// by later ones) - the network counterpart of the local CLI's `table + /// select --order-by`. Returned rows are independently owned - free + /// with `protocol.freeRows`. + pub fn tableSelect(self: *Client, allocator: Allocator, table: []const u8, exprText: []const u8, offset: u32, limit: u32, orderBy: []const querymod.OrderBy) ![]rowmod.Row { + const payload = try protocol.encodeTableSelect(self.allocator, table, exprText, offset, limit, orderBy); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.table_select, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return protocol.decodeRows(allocator, frame.payload); + } + + /// Remote `table join`: the wire counterpart of the CLI's local `table + /// join [--left|--right|--full]`, running whichever of `join.zig`'s four + /// equi-join variants `mode` names against the server's `*Db` instead of + /// a locally opened one. `leftExprText`/`rightExprText` are the same + /// `expr.zig` predicate grammar as `tableSelect`'s `exprText` (empty + /// selects every row on that side), filtering each table before the + /// join runs - the network counterpart of `innerJoin`'s + /// `leftPred`/`rightPred` parameters. `offset`/`limit` window the + /// already-joined result server-side (0/0 = unlimited), the same + /// convention `tableSelect` uses, via `join.applyPage`. Returned pairs + /// are independently owned - free with `protocol.freeJoinPairs`. + pub fn tableJoin( + self: *Client, + allocator: Allocator, + mode: protocol.JoinMode, + leftTable: []const u8, + leftField: []const u8, + rightTable: []const u8, + rightField: []const u8, + offset: u32, + limit: u32, + leftExprText: []const u8, + rightExprText: []const u8, + ) ![]protocol.JoinPair { + const payload = try protocol.encodeTableJoin(self.allocator, mode, leftTable, leftField, rightTable, rightField, offset, limit, leftExprText, rightExprText); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.table_join, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return protocol.decodeJoinedRows(allocator, frame.payload); + } + + /// Remote `table aggregate
* [expr...]` (no `GROUP BY`): + /// the wire counterpart of the CLI's local `table aggregate` with a `*` + /// group field, calling `aggregate.runGlobal` server-side over rows + /// filtered by `exprText` (same `expr.zig` grammar, empty selects every + /// row, as `tableSelect`). `havingText` (empty = no HAVING) filters the + /// single result server-side via `aggregate.globalMatchesHaving` - + /// returns zero values (not an error) when it excludes the only row. + /// Otherwise always returns exactly `specs.len` values in the same order + /// as `specs` - free with `protocol.freeAggValues`. + pub fn tableAggregateGlobal(self: *Client, allocator: Allocator, table: []const u8, specs: []const aggregatemod.AggSpec, havingText: []const u8, exprText: []const u8) ![]rowmod.Value { + const payload = try protocol.encodeTableAggregate(self.allocator, table, "", specs, 0, 0, &.{}, havingText, exprText); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.table_aggregate, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return protocol.decodeAggGlobalResult(allocator, frame.payload); + } + + /// Remote `table aggregate
[expr...] + /// [having ] [--limit N] [--offset N]`: the wire counterpart of + /// the CLI's local `table aggregate` with a real GROUP BY field, calling + /// `aggregate.run` server-side, then (when `havingText` is non-empty) + /// `aggregate.filterHaving` to narrow the groups, then windowing what + /// remains server-side with `aggregate.applyPageGroups` (0/0 = + /// unlimited, the same convention `tableJoin`'s `offset`/`limit` use - + /// HAVING always runs before the window, same as SQL). `orderBy` + /// (possibly empty) sorts the HAVING-narrowed groups server-side via + /// `aggregate.sortGroupsBy` before that window, the same + /// HAVING-then-ORDER-BY-then-LIMIT clause order SQL evaluates - each + /// `orderBy[i].field` must name either `groupField` or one of `specs`' + /// result labels (see `aggregate.specLabel`). One `GroupRow` per + /// distinct group value kept by HAVING and the window, in `orderBy` + /// order when non-empty, otherwise first-seen order (not sorted), same + /// as the local library call - free with `protocol.freeAggGroupRows`. + pub fn tableAggregateGroups(self: *Client, 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) ![]aggregatemod.GroupRow { + const payload = try protocol.encodeTableAggregate(self.allocator, table, groupField, specs, offset, limit, orderBy, havingText, exprText); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.table_aggregate, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return protocol.decodeAggGroupRows(allocator, frame.payload); + } + + /// Remote `table window
[expr...] [partition + /// [,field...]] [order [,field...]] [--limit N] + /// [--offset N]`: the wire counterpart of the CLI's local `table + /// window`, calling `window.compute` server-side over rows filtered by + /// `exprText` (same `expr.zig` grammar, empty selects every row). + /// Unlike `tableAggregateGroups`, this never collapses rows - every + /// matching row keeps its own `protocol.WindowRow` - but `offset`/`limit` + /// still window WHICH rows come back, walked in the same + /// partition-then-order display sequence `window.orderedIndices` + /// produces server-side (0/0 means unlimited, same convention as every + /// other paginated opcode). Returns one `WindowRow` per surviving row, in + /// that display order - free with `protocol.freeWindowRows`. + pub fn tableWindow(self: *Client, allocator: Allocator, table: []const u8, specs: []const windowmod.WindowSpec, partitionBy: []const []const u8, orderBy: []const querymod.OrderBy, offset: u32, limit: u32, exprText: []const u8, windowFrame: windowmod.FrameMode) ![]protocol.WindowRow { + const payload = try protocol.encodeTableWindow(self.allocator, table, specs, partitionBy, orderBy, offset, limit, exprText, windowFrame); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.table_window, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return protocol.decodeWindowRows(allocator, frame.payload); + } + + /// Remote `subscribe [,channel...]`: registers this connection + /// with the server's `pubsub.Hub` and blocks until the server's ack + /// frame comes back, so a `publish` issued on a different connection + /// right after this returns is guaranteed to reach this subscription - + /// see `server.zig`'s `handleSubscribe` doc comment for why the ack + /// exists at all. After this call, `self` must only ever be used with + /// `nextMessage` - the server never answers another request opcode on + /// this same connection again (a real, documented protocol + /// simplification, not an oversight: see `handleSubscribe`'s doc + /// comment on why a subscribed connection is receive-only). Returns the + /// `pubsub.Hub`-assigned subscriber id, the only handle a later + /// `unsubscribe` call - necessarily made through a DIFFERENT `Client`, + /// since this one can never send another request - can use to name this + /// subscription. + pub fn subscribe(self: *Client, channels: []const []const u8) !u64 { + const payload = try protocol.encodeSubscribe(self.allocator, channels); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.subscribe, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + const ack = try protocol.decodeSubscribeAck(frame.payload); + return ack.subscriberId; + } + + /// Remote `unsubscribe [,channel...]`: the + /// client-side counterpart of `server.zig`'s `unsubscribe` opcode. Sent + /// over an ordinary connection (this one may be reused for other + /// requests afterward, unlike the subscribe connection it targets) to + /// drop a subset of `subscriberId`'s channels - or, if it names every + /// channel that subscriber still has left, to end it outright (see + /// `pubsub.Hub.unsubscribe`'s doc comment). Returns how many of + /// `channels` were actually removed. + pub fn unsubscribe(self: *Client, subscriberId: u64, channels: []const []const u8) !u32 { + const payload = try protocol.encodeUnsubscribe(self.allocator, subscriberId, channels); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.unsubscribe, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return protocol.decodeUnsubscribeResult(frame.payload); + } + + /// Reads the next pushed message on a connection that already called + /// `subscribe`, or `null` once the server closes the connection (the + /// same "no more frames coming" signal `roundTrip` treats as an error + /// everywhere else, but is the routine end of a subscription here). + /// Blocks indefinitely otherwise - there is no polling variant, exactly + /// like `Subscriber.waitMessage` server-side. Returns an independently + /// owned `protocol.PubSubMessage` - free with its own `deinit`. + pub fn nextMessage(self: *Client, allocator: Allocator) !?protocol.PubSubMessage { + var frame = protocol.readFrame(self.allocator, self.stream.reader()) catch |err| switch (err) { + error.EndOfStream => return null, + else => return err, + }; + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return try protocol.decodePubSubMessage(allocator, frame.payload); + } + + /// Remote `publish `: delivers `message` to every + /// subscriber currently registered on `channel` across every connection + /// the server has, returning how many subscribers it reached (0 is a + /// normal outcome for a channel nobody is listening to, not an error - + /// see `pubsub.Hub.publish`'s doc comment). Any connection may publish, + /// whether or not it has ever subscribed to anything itself. + pub fn publish(self: *Client, channel: []const u8, message: []const u8) !u64 { + const payload = try protocol.encodePublish(self.allocator, channel, message); + defer self.allocator.free(payload); + var frame = try self.roundTrip(.publish, payload); + defer frame.deinit(); + const status = try protocol.Status.fromByte(frame.tag); + if (status != .ok) return statusToError(status); + return protocol.decodePublishResult(frame.payload); + } + + /// Remote `txn