feat: add encodeBatch and DecodedBatchOp to protocol
This commit is contained in:
1 parent
aad2726772
commit
e9b06f45d3
1 file changed
+162
@@ -953,3 +953,165 @@ pub const BatchOp = union(enum) {
|
|||||||
const batchKindPut: u8 = 0;
|
const batchKindPut: u8 = 0;
|
||||||
const batchKindDel: u8 = 1;
|
const batchKindDel: u8 = 1;
|
||||||
|
|
||||||
|
pub fn encodeBatch(allocator: Allocator, ops: []const BatchOp) ![]u8 {
|
||||||
|
var buf = std.ArrayList(u8).init(allocator);
|
||||||
|
errdefer buf.deinit();
|
||||||
|
const w = encode.ByteWriter.init(&buf);
|
||||||
|
try w.putU32(@intCast(ops.len));
|
||||||
|
for (ops) |op| {
|
||||||
|
switch (op) {
|
||||||
|
.put => |kv| {
|
||||||
|
try w.byte(batchKindPut);
|
||||||
|
try putLenPrefixed(w, kv.key);
|
||||||
|
try putLenPrefixed(w, kv.value);
|
||||||
|
},
|
||||||
|
.del => |key| {
|
||||||
|
try w.byte(batchKindDel);
|
||||||
|
try putLenPrefixed(w, key);
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return buf.toOwnedSlice();
|
||||||
|
}
|
||||||
|
|
||||||
|
pub const DecodedBatchOp = union(enum) {
|
||||||
|
put: struct { key: []const u8, value: []const u8 },
|
||||||
|
del: []const u8,
|
||||||
|
};
|
||||||
|
|
||||||
|
pub fn decodeBatch(allocator: Allocator, payload: []const u8) ![]DecodedBatchOp {
|
||||||
|
var c = Cursor{ .data = payload };
|
||||||
|
const count = try c.takeU32();
|
||||||
|
var list = try std.ArrayList(DecodedBatchOp).initCapacity(allocator, count);
|
||||||
|
errdefer list.deinit();
|
||||||
|
var i: u32 = 0;
|
||||||
|
while (i < count) : (i += 1) {
|
||||||
|
const kind = try c.takeByte();
|
||||||
|
switch (kind) {
|
||||||
|
batchKindPut => {
|
||||||
|
const key = try c.takeLenPrefixed();
|
||||||
|
const value = try c.takeLenPrefixed();
|
||||||
|
list.appendAssumeCapacity(.{ .put = .{ .key = key, .value = value } });
|
||||||
|
},
|
||||||
|
batchKindDel => {
|
||||||
|
const key = try c.takeLenPrefixed();
|
||||||
|
list.appendAssumeCapacity(.{ .del = key });
|
||||||
|
},
|
||||||
|
else => return error.Malformed,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return list.toOwnedSlice();
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- table select ---
|
||||||
|
|
||||||
|
/// Wire request for a remote `table select`: a table name, the raw text of
|
||||||
|
/// an `expr.zig` predicate expression (empty text selects every row, the
|
||||||
|
/// same convention `query.Table.select(null)` uses locally), an
|
||||||
|
/// offset/limit page window using the same "0 = unlimited" convention as
|
||||||
|
/// `ScanArgs.limit` above - so a large remote result can be paged instead
|
||||||
|
/// of always coming back in one frame bounded only by `maxFrameBytes` - and
|
||||||
|
/// a `query.OrderBy` sort key list (possibly empty) applied server-side via
|
||||||
|
/// `Table.selectPageOrdered` before that page window. The order-by count
|
||||||
|
/// byte (and each field name/direction pair behind it) sit BEFORE
|
||||||
|
/// `exprText` rather than after it, because `exprText` is decoded as
|
||||||
|
/// "everything left in the payload" - anything meant to survive decoding
|
||||||
|
/// has to come first. A single byte bounds the count to 255 columns, far
|
||||||
|
/// more than any real query needs.
|
||||||
|
pub fn encodeTableSelect(allocator: Allocator, table: []const u8, exprText: []const u8, offset: u32, limit: u32, orderBy: []const querymod.OrderBy) ![]u8 {
|
||||||
|
var buf = std.ArrayList(u8).init(allocator);
|
||||||
|
errdefer buf.deinit();
|
||||||
|
const w = encode.ByteWriter.init(&buf);
|
||||||
|
try putLenPrefixed(w, table);
|
||||||
|
try w.putU32(offset);
|
||||||
|
try w.putU32(limit);
|
||||||
|
try w.byte(@intCast(orderBy.len));
|
||||||
|
for (orderBy) |ob| {
|
||||||
|
try w.byte(if (ob.descending) 1 else 0);
|
||||||
|
try putLenPrefixed(w, ob.field);
|
||||||
|
}
|
||||||
|
try w.slice(exprText);
|
||||||
|
return buf.toOwnedSlice();
|
||||||
|
}
|
||||||
|
|
||||||
|
pub const TableSelectArgs = struct {
|
||||||
|
table: []const u8,
|
||||||
|
exprText: []const u8,
|
||||||
|
offset: u32 = 0,
|
||||||
|
limit: u32 = 0,
|
||||||
|
orderBy: []const querymod.OrderBy = &.{},
|
||||||
|
};
|
||||||
|
|
||||||
|
/// `table`, `exprText`, and every `orderBy[i].field` are borrowed slices
|
||||||
|
/// into `payload` - the same convention `decodeCollOp` uses. Only the
|
||||||
|
/// `orderBy` array itself is owned by the caller (free with
|
||||||
|
/// `allocator.free`); never its contents.
|
||||||
|
pub fn decodeTableSelect(allocator: Allocator, payload: []const u8) !TableSelectArgs {
|
||||||
|
var c = Cursor{ .data = payload };
|
||||||
|
const table = try c.takeLenPrefixed();
|
||||||
|
const offset = try c.takeU32();
|
||||||
|
const limit = try c.takeU32();
|
||||||
|
const count = try c.takeByte();
|
||||||
|
var list = try std.ArrayList(querymod.OrderBy).initCapacity(allocator, count);
|
||||||
|
errdefer list.deinit();
|
||||||
|
var i: u8 = 0;
|
||||||
|
while (i < count) : (i += 1) {
|
||||||
|
const descending = try c.takeByte();
|
||||||
|
const field = try c.takeLenPrefixed();
|
||||||
|
list.appendAssumeCapacity(.{ .field = field, .descending = descending != 0 });
|
||||||
|
}
|
||||||
|
const exprText = c.data[c.pos..];
|
||||||
|
return .{ .table = table, .exprText = exprText, .offset = offset, .limit = limit, .orderBy = try list.toOwnedSlice() };
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Encodes a `table select` result set as a length-prefixed list of rows,
|
||||||
|
/// each row itself encoded with `Row.encode` (the same self-describing
|
||||||
|
/// field format a Table stores on disk).
|
||||||
|
pub fn encodeRows(allocator: Allocator, rows: []const rowmod.Row) ![]u8 {
|
||||||
|
var buf = std.ArrayList(u8).init(allocator);
|
||||||
|
errdefer buf.deinit();
|
||||||
|
const w = encode.ByteWriter.init(&buf);
|
||||||
|
try w.putU32(@intCast(rows.len));
|
||||||
|
for (rows) |*r| {
|
||||||
|
const encoded = try r.encode(allocator);
|
||||||
|
defer allocator.free(encoded);
|
||||||
|
try putLenPrefixed(w, encoded);
|
||||||
|
}
|
||||||
|
return buf.toOwnedSlice();
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Decodes a `table select` result set into independently owned Rows -
|
||||||
|
/// callers must free each with `Row.deinit` (see `freeRows` below).
|
||||||
|
pub fn decodeRows(allocator: Allocator, payload: []const u8) ![]rowmod.Row {
|
||||||
|
var c = Cursor{ .data = payload };
|
||||||
|
const count = try c.takeU32();
|
||||||
|
var list = try std.ArrayList(rowmod.Row).initCapacity(allocator, count);
|
||||||
|
errdefer {
|
||||||
|
for (list.items) |*r| r.deinit();
|
||||||
|
list.deinit();
|
||||||
|
}
|
||||||
|
var i: u32 = 0;
|
||||||
|
while (i < count) : (i += 1) {
|
||||||
|
const raw = try c.takeLenPrefixed();
|
||||||
|
const row = rowmod.Row.decode(allocator, raw) catch return error.Malformed;
|
||||||
|
list.appendAssumeCapacity(row);
|
||||||
|
}
|
||||||
|
return list.toOwnedSlice();
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn freeRows(allocator: Allocator, rows: []rowmod.Row) void {
|
||||||
|
for (rows) |*r| r.deinit();
|
||||||
|
allocator.free(rows);
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- table join ---
|
||||||
|
|
||||||
|
/// Which of `join.zig`'s four equi-join variants a `table_join` request
|
||||||
|
/// wants - the wire counterpart of the CLI's `table join`
|
||||||
|
/// `[--left|--right|--full]` flags (no flag means `.inner`).
|
||||||
|
pub const JoinMode = enum(u8) {
|
||||||
|
inner = 0,
|
||||||
|
left = 1,
|
||||||
|
right = 2,
|
||||||
|
full = 3,
|
||||||
|
|
||||||
Reference in new issue
Block a user