feat: add specLabel and RunningAcc to window

This commit is contained in:
dualfroz committed 2026-09-13 20:02:45 +02:00
1 parent 2da97f6a7e
commit cb94bb5b1b
1 file changed
+154
+154
View File
@@ -152,3 +152,157 @@ fn isFrameValue(kind: WindowKind) bool {
return kind == .first_value or kind == .last_value or kind == .nth_value; return kind == .first_value or kind == .last_value or kind == .nth_value;
} }
/// The row field name a `compute()` result column is exposed under (e.g.
/// for CLI/protocol output) - bare `WindowKind` name for the ranking family
/// and for `count`'s `COUNT(*)` form (`field.len == 0`), otherwise
/// `"kind_field"`. Format-identical to `aggregate.zig`'s `specLabel`.
pub fn specLabel(allocator: Allocator, spec: WindowSpec) 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 });
}
/// Streaming per-row accumulator for the running-aggregate family
/// (`sum`/`avg`/`count`/`min`/`max`). Mirrors `aggregate.zig`'s `AccState`
/// (same int-exact/float-fallback dual for `sum`/`avg`, same
/// compare-and-replace for `min`/`max`), minus `count_distinct` (out of
/// scope here - a running "distinct count so far" is a real SQL feature but
/// a separate piece of work). The key behavioral difference from
/// `AccState.take()`: `snapshot()` does NOT transfer ownership of
/// `minv`/`maxv` out of the accumulator, because a running window emits a
/// result after EVERY row, not once at the very end - the accumulator must
/// keep its own copy alive for the next row's snapshot too.
const RunningAcc = struct {
kind: WindowKind,
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 - same encode-as-hashmap-key trick
/// `aggregate.zig`'s `AccState.distinct` uses. Unlike that GROUP BY
/// counterpart, `snapshot()` below never drains this (a running window
/// needs the set to keep growing across the WHOLE partition walk, not
/// just until one final result) - only `deinit` ever frees it.
distinct: std.StringHashMapUnmanaged(void) = .{},
fn observe(self: *RunningAcc, allocator: Allocator, row: *const Row) WindowError!void {
switch (self.kind) {
.count => {
if (self.field.len == 0 or row.get(self.field) != null) self.n += 1;
},
.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);
},
.sum, .avg => {
const v = row.get(self.field) 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);
}
},
.row_number, .rank, .dense_rank, .lag, .lead, .first_value, .last_value, .nth_value => unreachable,
}
}
/// Returns the accumulator's CURRENT state as a freshly owned `Value`
/// (a dupe for `min`/`max`, since the accumulator keeps its own copy
/// alive for subsequent rows - see the type doc comment above).
/// `.count_distinct` is likewise non-draining - `self.distinct.count()`
/// is read, never emptied, so the NEXT row's snapshot still sees every
/// value observed so far (only `deinit` ever frees the set).
fn snapshot(self: *const RunningAcc, allocator: Allocator) Allocator.Error!Value {
return switch (self.kind) {
.count => .{ .int = @intCast(self.n) },
.count_distinct => .{ .int = @intCast(self.distinct.count()) },
.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 => if (self.minv) |v| try v.dupe(allocator) else .{ .none = {} },
.max => if (self.maxv) |v| try v.dupe(allocator) else .{ .none = {} },
.row_number, .rank, .dense_rank, .lag, .lead, .first_value, .last_value, .nth_value => unreachable,
};
}
fn deinit(self: *RunningAcc, 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);
}
};
/// Groups row indices `0..rows.len` by `partitionBy` (empty = a single
/// partition covering every row), preserving first-seen partition order -
/// the window-function analogue of `aggregate.zig`'s `GroupBy` grouping,
/// except every member index is kept instead of collapsed into one result.
/// Multi-column partition keys are built with `row.appendOrdered`
/// (concatenation-safe and unambiguous by construction), the same
/// primitive `index.zig` uses for composite index keys - a row missing one
/// of the `partitionBy` fields contributes `.none` for that column rather
/// than being excluded (unlike `aggregate.zig`'s GROUP BY, which drops rows
/// missing the group field entirely: a window function must still annotate
/// every input row with SOME result, so there is no row to drop it from).
/// Caller owns the returned slice of slices AND each inner slice
/// (`allocator.free` each inner slice, then the outer one).
fn buildPartitions(allocator: Allocator, rows: []const Row, partitionBy: []const []const u8) Allocator.Error![][]usize {
var order: std.ArrayListUnmanaged([]const u8) = .{};
defer {
for (order.items) |k| allocator.free(k);
order.deinit(allocator);
}
var map: std.StringHashMapUnmanaged(std.ArrayListUnmanaged(usize)) = .{};
defer {
var it = map.iterator();
while (it.next()) |e| e.value_ptr.deinit(allocator);
map.deinit(allocator);
}
for (rows, 0..) |*r, i| {
var keyBuf = std.ArrayList(u8).init(allocator);
defer keyBuf.deinit();
for (partitionBy) |f| {
const v = r.get(f) orelse Value{ .none = {} };
try rowmod.appendOrdered(&keyBuf, v);
}