From 601a264b78fabd8c7bf277cefd03d5972a90a324 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Mon, 7 Sep 2026 15:14:35 +0200 Subject: [PATCH] feat: extend main with runRepl and runTxn --- src/main.zig | 173 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 173 insertions(+) diff --git a/src/main.zig b/src/main.zig index 543a0c8..2eceb06 100644 --- a/src/main.zig +++ b/src/main.zig @@ -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)); 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 csv|json "); + 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
csv|json "); + 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
[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 ]`: 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 +/// ` with extra steps. With `--script ` 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 "); + + 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 [--once] [--interval-ms ]`: opens } fn runVerify(allocator: std.mem.Allocator, stdout: anytype, dir: []const u8, count: usize) !void {