feat: extend main with runRemoteImport and JoinCliMode

This commit is contained in:
dualfroz committed 2026-09-07 15:34:10 +02:00
1 parent 7424bb6ed8
commit 51bad5d86e
1 file changed
+173
+173
View File
@@ -2560,6 +2560,179 @@ fn runRemoteExport(allocator: std.mem.Allocator, stdout: anytype, stderr: anytyp
/// there instead of against a local Db - the client-side counterpart of /// there instead of against a local Db - the client-side counterpart of
/// `server.zig`'s `table_import` opcode. Mirrors `runRemoteExport`: the /// `server.zig`'s `table_import` opcode. Mirrors `runRemoteExport`: the
/// source `<path>` is read from THIS machine (the same way `runTxn` reads /// source `<path>` is read from THIS machine (the same way `runTxn` reads
/// a local script file), and its bytes are sent to the server in one
/// frame. `--user`/`--pass`/`--token` authenticate first, needed only
/// against a server started with `serve --auth`.
fn runRemoteImport(allocator: std.mem.Allocator, stdout: anytype, stderr: anytype, addressSpec: []const u8, rest: []const []const u8) !void {
if (rest.len < 1) return fail(stderr, "remote-import needs <table> csv|json <path>");
const address = try parseHostPort(stdout, addressSpec);
var client = cobalt.Client.connect(allocator, address) catch |err| {
try stderr.print("error: could not connect to {s}: {s}\n", .{ addressSpec, @errorName(err) });
std.process.exit(1);
};
defer client.close();
const authFlags = extractAuthFlags(allocator, rest[1..]) catch return fail(stderr, "invalid --user/--pass/--token value");
defer allocator.free(authFlags.rest);
try authenticateIfRequested(&client, stderr, authFlags.auth);
const args = authFlags.rest;
if (args.len < 2) return fail(stderr, "remote-import needs <table> csv|json <path>");
const format = try parseDumpFormat(stderr, args[0]);
const content = std.fs.cwd().readFileAlloc(allocator, args[1], 1 << 30) catch |err| {
try stderr.print("failed to read '{s}': {s}\n", .{ args[1], @errorName(err) });
std.process.exit(1);
};
defer allocator.free(content);
const imported = client.tableImport(rest[0], format, content) catch |err| {
try stderr.print("error: {s}\n", .{@errorName(err)});
std.process.exit(1);
};
try stdout.print("{d}\n", .{imported});
}
const JoinCliMode = enum { inner, left, right, full };
/// Pulls `--left`/`--right`/`--full` (mutually exclusive, last one wins,
/// same as local `table join`'s parsing loop) and `--left-expr E`/
/// `--right-expr E` out of a `remote-join` argument list, same "strip
/// recognized flags, keep the rest" shape as `extractPageFlags`/
/// `extractAuthFlags`. An expr not given defaults to the empty string,
/// meaning "select every row on that side" - the same convention
/// `tableSelect`'s empty `exprText` already uses. Caller owns and frees
/// `rest`.
fn extractJoinFlags(allocator: std.mem.Allocator, args: []const []const u8) !struct {
mode: JoinCliMode,
leftExpr: []const u8,
rightExpr: []const u8,
rest: [][]const u8,
} {
var mode: JoinCliMode = .inner;
var leftExpr: []const u8 = "";
var rightExpr: []const u8 = "";
var rest = std.ArrayList([]const u8).init(allocator);
errdefer rest.deinit();
var i: usize = 0;
while (i < args.len) : (i += 1) {
if (std.mem.eql(u8, args[i], "--left")) {
mode = .left;
} else if (std.mem.eql(u8, args[i], "--right")) {
mode = .right;
} else if (std.mem.eql(u8, args[i], "--full")) {
mode = .full;
} else if (std.mem.eql(u8, args[i], "--left-expr")) {
i += 1;
if (i >= args.len) return error.InvalidJoinFlag;
leftExpr = args[i];
} else if (std.mem.eql(u8, args[i], "--right-expr")) {
i += 1;
if (i >= args.len) return error.InvalidJoinFlag;
rightExpr = args[i];
} else {
try rest.append(args[i]);
}
}
return .{ .mode = mode, .leftExpr = leftExpr, .rightExpr = rightExpr, .rest = try rest.toOwnedSlice() };
}
/// Connects to a remote `cobalt serve` instance and runs `table join`
/// there instead of against a local Db - the client-side counterpart of
/// `server.zig`'s `table_join` opcode, mirroring `runRemoteSelect`'s shape
/// but for all four of `join.zig`'s equi-join variants instead of a plain
/// select. `--user`/`--pass`/`--token` authenticate first, needed only
/// against a server started with `serve --auth`; printing is identical to
/// the local `table join` command (`L <row>` / `R <row>`, or `L (nil)` /
/// `R (nil)` for the unmatched side of an outer join).
fn runRemoteJoin(allocator: std.mem.Allocator, stdout: anytype, stderr: anytype, addressSpec: []const u8, rest: []const []const u8) !void {
const address = try parseHostPort(stdout, addressSpec);
var client = cobalt.Client.connect(allocator, address) catch |err| {
try stderr.print("error: could not connect to {s}: {s}\n", .{ addressSpec, @errorName(err) });
std.process.exit(1);
};
defer client.close();
const authFlags = extractAuthFlags(allocator, rest) catch return fail(stderr, "invalid --user/--pass/--token value");
defer allocator.free(authFlags.rest);
try authenticateIfRequested(&client, stderr, authFlags.auth);
const jf = extractJoinFlags(allocator, authFlags.rest) catch return fail(stderr, "invalid --left-expr/--right-expr value");
defer allocator.free(jf.rest);
const pf = extractPageFlags(allocator, jf.rest) catch return fail(stderr, "invalid --limit/--offset value");
defer allocator.free(pf.rest);
if (pf.rest.len < 4) return fail(stderr, "remote-join needs [--left|--right|--full] <leftTable> <leftField> <rightTable> <rightField> [--left-expr E] [--right-expr E] [--limit N] [--offset N]");
const mode: cobalt.protocol.JoinMode = switch (jf.mode) {
.inner => .inner,
.left => .left,
.right => .right,
.full => .full,
};
const pairs = client.tableJoin(allocator, mode, pf.rest[0], pf.rest[1], pf.rest[2], pf.rest[3], pf.page.offset, pf.page.limit, jf.leftExpr, jf.rightExpr) catch |err| {
try stderr.print("error: {s}\n", .{@errorName(err)});
std.process.exit(1);
};
defer cobalt.protocol.freeJoinPairs(allocator, pairs);
for (pairs) |p| {
if (p.left) |l| {
try stdout.writeAll("L ");
try printRow(stdout, l);
} else {
try stdout.writeAll("L (nil)\n");
}
if (p.right) |r| {
try stdout.writeAll("R ");
try printRow(stdout, r);
} else {
try stdout.writeAll("R (nil)\n");
}
}
}
/// Connects to a remote `cobalt serve` instance and runs `table aggregate`
/// there instead of against a local Db - the client-side counterpart of
/// `server.zig`'s `table_aggregate` opcode, mirroring `runRemoteSelect`'s
/// shape. Same `*` sentinel as local `table aggregate` for "no GROUP BY"
/// (dispatches to `client.tableAggregateGlobal` instead of
/// `tableAggregateGroups`), same comma-separated spec grammar
/// (`parseAggSpecs`), and the same output format as the local command
/// (`printAggLabel`/`printValue`). `--limit`/`--offset` window the result
/// groups server-side (ignored for the `*` global path, same as local
/// `table aggregate`), same convention as `remote-join`.
/// `--user`/`--pass`/`--token` authenticate first, needed only against a
/// server started with `serve --auth`. An optional literal `having <expr>`
/// after the WHERE-style expr (split out by `splitAggFilterArgs`, the same
/// helper the local `table aggregate` handler uses) filters the result
/// groups server-side after aggregation, before `--limit`/`--offset` window
/// them. An optional literal `order <field[:desc]>[,field...]` after that
/// sorts the (HAVING-narrowed) result groups server-side before that same
/// window, ignored for the `*` global path exactly like `--limit`/`--offset`
/// already are - see `Client.tableAggregateGroups`/`tableAggregateGlobal`.
fn runRemoteAggregate(allocator: std.mem.Allocator, stdout: anytype, stderr: anytype, addressSpec: []const u8, rest: []const []const u8) !void {
if (rest.len < 3) return fail(stderr, "remote-aggregate needs <table> <groupField|*> <spec>[,<spec>...] [expr...] [having <expr>] [order <field[:desc]>[,field...]] [--limit N] [--offset N]");
const address = try parseHostPort(stdout, addressSpec);
var client = cobalt.Client.connect(allocator, address) catch |err| {
try stderr.print("error: could not connect to {s}: {s}\n", .{ addressSpec, @errorName(err) });
std.process.exit(1);
};
defer client.close();
const table = rest[0];
const groupField = rest[1];
const specs = parseAggSpecs(allocator, rest[2]) catch return fail(stderr, "invalid aggregate spec (expected count | count:field | sum:field | avg:field | min:field | max:field | distinct:field, comma-separated)");
defer allocator.free(specs);
} }
fn runVerify(allocator: std.mem.Allocator, stdout: anytype, dir: []const u8, count: usize) !void { fn runVerify(allocator: std.mem.Allocator, stdout: anytype, dir: []const u8, count: usize) !void {