feat: extend main with runRepl and runTxn
This commit is contained in:
dualfroz committed
1 parent
c164c23f4e
commit
601a264b78
1 file changed
+173
+173
@@ -2041,6 +2041,179 @@ fn runTable(allocator: std.mem.Allocator, stdout: anytype, stderr: anytype, db:
|
|||||||
const results = cobalt.window.computeMany(allocator, rows, partitionBy, orderBy, specs, frame) catch |err| return fail(stderr, @errorName(err));
|
const results = cobalt.window.computeMany(allocator, rows, partitionBy, orderBy, specs, frame) catch |err| return fail(stderr, @errorName(err));
|
||||||
defer cobalt.window.freeManyResults(allocator, results);
|
defer cobalt.window.freeManyResults(allocator, results);
|
||||||
|
|
||||||
|
// Display/page in the SAME partition-then-order walk `computeMany`
|
||||||
|
// used internally, not raw scan order - see `table window`'s help
|
||||||
|
// text and `window.orderedIndices`'s doc comment for why.
|
||||||
|
const displayOrder = cobalt.window.orderedIndices(allocator, rows, partitionBy, orderBy) catch |err| return fail(stderr, @errorName(err));
|
||||||
|
defer allocator.free(displayOrder);
|
||||||
|
|
||||||
|
const total = displayOrder.len;
|
||||||
|
const start = @min(flags.page.offset, total);
|
||||||
|
const end = if (flags.page.limit == 0) total else @min(start + flags.page.limit, total);
|
||||||
|
var i: usize = start;
|
||||||
|
while (i < end) : (i += 1) {
|
||||||
|
const idx = displayOrder[i];
|
||||||
|
for (rows[idx].fields.items, 0..) |f, j| {
|
||||||
|
if (j > 0) try stdout.writeAll("\t");
|
||||||
|
try stdout.print("{s}=", .{f.name});
|
||||||
|
try printValue(stdout, f.value);
|
||||||
|
}
|
||||||
|
if (rows[idx].fields.items.len > 0) try stdout.writeAll("\t");
|
||||||
|
for (specs, 0..) |spec, specIdx| {
|
||||||
|
if (specIdx > 0) try stdout.writeAll("\t");
|
||||||
|
try printWindowLabel(stdout, spec);
|
||||||
|
try stdout.writeAll("=");
|
||||||
|
try printValue(stdout, results[idx][specIdx]);
|
||||||
|
}
|
||||||
|
try stdout.writeAll("\n");
|
||||||
|
}
|
||||||
|
} else if (std.mem.eql(u8, sub, "export")) {
|
||||||
|
if (args.len < 3) return fail(stderr, "table export needs <table> csv|json <path>");
|
||||||
|
var table = try cobalt.Table.open(db, allocator, args[0]);
|
||||||
|
defer table.deinit();
|
||||||
|
const n = if (std.mem.eql(u8, args[1], "csv"))
|
||||||
|
cobalt.dump.exportCsv(allocator, &table, args[2]) catch |err| return fail(stderr, @errorName(err))
|
||||||
|
else if (std.mem.eql(u8, args[1], "json"))
|
||||||
|
cobalt.dump.exportJson(allocator, &table, args[2]) catch |err| return fail(stderr, @errorName(err))
|
||||||
|
else
|
||||||
|
return fail(stderr, "table export format must be csv or json");
|
||||||
|
try stdout.print("{d}\n", .{n});
|
||||||
|
} else if (std.mem.eql(u8, sub, "import")) {
|
||||||
|
if (args.len < 3) return fail(stderr, "table import needs <table> csv|json <path>");
|
||||||
|
var table = try cobalt.Table.open(db, allocator, args[0]);
|
||||||
|
defer table.deinit();
|
||||||
|
const n = if (std.mem.eql(u8, args[1], "csv"))
|
||||||
|
cobalt.dump.importCsv(allocator, &table, args[2]) catch |err| return fail(stderr, @errorName(err))
|
||||||
|
else if (std.mem.eql(u8, args[1], "json"))
|
||||||
|
cobalt.dump.importJson(allocator, &table, args[2]) catch |err| return fail(stderr, @errorName(err))
|
||||||
|
else
|
||||||
|
return fail(stderr, "table import format must be csv or json");
|
||||||
|
try stdout.print("{d}\n", .{n});
|
||||||
|
} else {
|
||||||
|
return fail(stderr, "unknown table operation");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Runs a script of `put|del <table> <pk> [field=value ...]` lines as one
|
||||||
|
/// atomic Transaction via TableTxn: every row/index mutation across every
|
||||||
|
/// line becomes visible together on commit, or none of it does. This is the
|
||||||
|
/// only place in the CLI multi-row atomicity is reachable at all - `table
|
||||||
|
/// put` above writes one row directly to the Db per call, with no atomicity
|
||||||
|
/// across rows.
|
||||||
|
/// `repl [--script <file>]`: keeps `db` open across many commands instead
|
||||||
|
/// of the usual one-process-per-command CLI shape, so `begin`/`commit`/
|
||||||
|
/// `rollback` can bracket an interactive `cobalt.Repl` transaction - see
|
||||||
|
/// `repl.zig` for the command grammar and why that is more than `txn
|
||||||
|
/// <script-file>` with extra steps. With `--script <file>` reads from the
|
||||||
|
/// file non-interactively (no prompt, same as `txn`); otherwise reads from
|
||||||
|
/// stdin with a `cobalt>`/`cobalt(txn)>` prompt.
|
||||||
|
fn runRepl(allocator: std.mem.Allocator, stdout: anytype, stderr: anytype, db: *cobalt.Db, rest: []const []const u8) !void {
|
||||||
|
var session = cobalt.Repl.init(allocator, db);
|
||||||
|
defer session.deinit();
|
||||||
|
|
||||||
|
if (rest.len >= 2 and std.mem.eql(u8, rest[0], "--script")) {
|
||||||
|
const contents = std.fs.cwd().readFileAlloc(allocator, rest[1], 16 * 1024 * 1024) catch |err| {
|
||||||
|
try stderr.print("failed to read script '{s}': {s}\n", .{ rest[1], @errorName(err) });
|
||||||
|
std.process.exit(1);
|
||||||
|
};
|
||||||
|
defer allocator.free(contents);
|
||||||
|
var stream = std.io.fixedBufferStream(contents);
|
||||||
|
try session.runFromReader(stream.reader(), stdout, false);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
const stdinReader = std.io.getStdIn().reader();
|
||||||
|
try session.runFromReader(stdinReader, stdout, true);
|
||||||
|
}
|
||||||
|
|
||||||
|
fn runTxn(allocator: std.mem.Allocator, stdout: anytype, stderr: anytype, db: *cobalt.Db, rest: []const []const u8) !void {
|
||||||
|
if (rest.len < 1) return fail(stderr, "txn needs <script-file>");
|
||||||
|
|
||||||
|
const contents = std.fs.cwd().readFileAlloc(allocator, rest[0], 16 * 1024 * 1024) catch |err| {
|
||||||
|
try stderr.print("failed to read script '{s}': {s}\n", .{ rest[0], @errorName(err) });
|
||||||
|
std.process.exit(1);
|
||||||
|
};
|
||||||
|
defer allocator.free(contents);
|
||||||
|
|
||||||
|
var txn = cobalt.Transaction.begin(db);
|
||||||
|
defer txn.deinit();
|
||||||
|
|
||||||
|
var tables = std.StringHashMap(cobalt.TableTxn).init(allocator);
|
||||||
|
defer {
|
||||||
|
var it = tables.valueIterator();
|
||||||
|
while (it.next()) |t| t.deinit();
|
||||||
|
tables.deinit();
|
||||||
|
}
|
||||||
|
|
||||||
|
var applied: usize = 0;
|
||||||
|
var lineNo: usize = 0;
|
||||||
|
var lines = std.mem.splitScalar(u8, contents, '\n');
|
||||||
|
while (lines.next()) |rawLine| {
|
||||||
|
lineNo += 1;
|
||||||
|
const line = std.mem.trim(u8, rawLine, " \t\r");
|
||||||
|
if (line.len == 0 or line[0] == '#') continue;
|
||||||
|
|
||||||
|
var tok = std.mem.tokenizeScalar(u8, line, ' ');
|
||||||
|
const verb = tok.next() orelse continue;
|
||||||
|
const tableName = tok.next() orelse {
|
||||||
|
try stderr.print("line {d}: missing table name\n", .{lineNo});
|
||||||
|
std.process.exit(2);
|
||||||
|
};
|
||||||
|
const pkText = tok.next() orelse {
|
||||||
|
try stderr.print("line {d}: missing primary key\n", .{lineNo});
|
||||||
|
std.process.exit(2);
|
||||||
|
};
|
||||||
|
const pk = parseValue(pkText) catch {
|
||||||
|
try stderr.print("line {d}: invalid primary key '{s}'\n", .{ lineNo, pkText });
|
||||||
|
std.process.exit(2);
|
||||||
|
};
|
||||||
|
|
||||||
|
const gop = try tables.getOrPut(tableName);
|
||||||
|
if (!gop.found_existing) {
|
||||||
|
gop.key_ptr.* = tableName;
|
||||||
|
gop.value_ptr.* = cobalt.TableTxn.open(&txn, allocator, tableName) catch |err| {
|
||||||
|
try stderr.print("line {d}: cannot open table '{s}': {s}\n", .{ lineNo, tableName, @errorName(err) });
|
||||||
|
std.process.exit(2);
|
||||||
|
};
|
||||||
|
}
|
||||||
|
const table = gop.value_ptr;
|
||||||
|
|
||||||
|
if (std.mem.eql(u8, verb, "put")) {
|
||||||
|
var fields = std.ArrayList(Field).init(allocator);
|
||||||
|
defer fields.deinit();
|
||||||
|
while (tok.next()) |raw| {
|
||||||
|
const f = parseField(raw) catch {
|
||||||
|
try stderr.print("line {d}: invalid field '{s}'\n", .{ lineNo, raw });
|
||||||
|
std.process.exit(2);
|
||||||
|
};
|
||||||
|
try fields.append(f);
|
||||||
|
}
|
||||||
|
table.put(pk, fields.items) catch |err| {
|
||||||
|
try stderr.print("line {d}: put failed: {s}\n", .{ lineNo, @errorName(err) });
|
||||||
|
std.process.exit(1);
|
||||||
|
};
|
||||||
|
} else if (std.mem.eql(u8, verb, "del")) {
|
||||||
|
table.delete(pk) catch |err| {
|
||||||
|
try stderr.print("line {d}: del failed: {s}\n", .{ lineNo, @errorName(err) });
|
||||||
|
std.process.exit(1);
|
||||||
|
};
|
||||||
|
} else {
|
||||||
|
try stderr.print("line {d}: unknown verb '{s}' (expected put/del)\n", .{ lineNo, verb });
|
||||||
|
std.process.exit(2);
|
||||||
|
}
|
||||||
|
applied += 1;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (applied == 0) return fail(stderr, "txn script had no operations");
|
||||||
|
|
||||||
|
txn.commit() catch |err| {
|
||||||
|
try stderr.print("transaction failed, nothing applied: {s}\n", .{@errorName(err)});
|
||||||
|
std.process.exit(1);
|
||||||
|
};
|
||||||
|
try stdout.print("OK: {d} operation(s) committed atomically\n", .{applied});
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Dispatches `replicate <replica-dir> [--once] [--interval-ms <n>]`: opens
|
||||||
}
|
}
|
||||||
|
|
||||||
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 {
|
||||||
|
|||||||
Reference in new issue
Block a user