feat: grow server
This commit is contained in:
dualfroz committed
1 parent
c71d01bd7e
commit
6fb0088939
1 file changed
+157
+157
@@ -2968,3 +2968,160 @@ test "table_update survives concurrent flush and compact reshaping a composite i
|
||||
}
|
||||
}.run;
|
||||
|
||||
var readerContexts: [readerThreads]ReaderCtx = undefined;
|
||||
var readerHandles: [readerThreads]std.Thread = undefined;
|
||||
for (0..readerThreads) |i| {
|
||||
readerContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress() };
|
||||
readerHandles[i] = try std.Thread.spawn(.{}, readerWorker, .{&readerContexts[i]});
|
||||
}
|
||||
var reshapeContexts: [reshapeThreads]ReshapeCtx = undefined;
|
||||
var reshapeHandles: [reshapeThreads]std.Thread = undefined;
|
||||
for (0..reshapeThreads) |i| {
|
||||
reshapeContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress(), .thread = i };
|
||||
reshapeHandles[i] = try std.Thread.spawn(.{}, reshapeWorker, .{&reshapeContexts[i]});
|
||||
}
|
||||
var updaterContext = UpdaterCtx{ .allocator = allocator, .addr = server.boundAddress() };
|
||||
const updaterHandle = try std.Thread.spawn(.{}, updaterWorker, .{&updaterContext});
|
||||
|
||||
for (0..readerThreads) |i| readerHandles[i].join();
|
||||
for (0..reshapeThreads) |i| reshapeHandles[i].join();
|
||||
updaterHandle.join();
|
||||
for (0..readerThreads) |i| try std.testing.expect(readerContexts[i].ok);
|
||||
for (0..reshapeThreads) |i| try std.testing.expect(reshapeContexts[i].ok);
|
||||
try std.testing.expect(updaterContext.ok);
|
||||
|
||||
// Final sanity check on a fresh connection: the updater alternated
|
||||
// krakow/30 and gdansk/40, so the last krakow/30 write was iteration 18
|
||||
// (last even below 20) and the last gdansk/40 write was iteration 19
|
||||
// (last odd) - the other four (city, age) combos were never touched by
|
||||
// any predicate and must still read epoch 0.
|
||||
var client = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||
defer client.close();
|
||||
|
||||
for (combos) |combo| {
|
||||
const expr = try std.fmt.allocPrint(allocator, "city eq s:{s} and age eq i:{d}", .{ combo.city, combo.age });
|
||||
defer allocator.free(expr);
|
||||
const rows = try client.tableSelect(allocator, "accounts", expr, 0, 0, &.{});
|
||||
defer protocol.freeRows(allocator, rows);
|
||||
try std.testing.expectEqual(@as(usize, 5), rows.len);
|
||||
const wantEpoch: i64 = if (std.mem.eql(u8, combo.city, "krakow") and combo.age == 30)
|
||||
18
|
||||
else if (std.mem.eql(u8, combo.city, "gdansk") and combo.age == 40)
|
||||
19
|
||||
else
|
||||
0;
|
||||
for (rows) |row| try std.testing.expectEqual(wantEpoch, row.get("epoch").?.int);
|
||||
|
||||
// The composite index must still be the strategy actually used
|
||||
// after all this churn, not a fallback to a full scan.
|
||||
var plan = try client.tableExplain(allocator, "accounts", expr);
|
||||
defer plan.deinit(allocator);
|
||||
try std.testing.expectEqual(querymod.PlanKind.composite_eq, plan.kind);
|
||||
}
|
||||
}
|
||||
|
||||
test "server table_delete removes matching rows over real TCP, keeping an index in sync" {
|
||||
const allocator = std.testing.allocator;
|
||||
var tmp = std.testing.tmpDir(.{});
|
||||
defer tmp.cleanup();
|
||||
const path = try tmp.dir.realpathAlloc(allocator, ".");
|
||||
defer allocator.free(path);
|
||||
|
||||
var db = try testDb(allocator, path);
|
||||
defer db.close();
|
||||
|
||||
{
|
||||
var table = try querymod.Table.open(&db, allocator, "users");
|
||||
defer table.deinit();
|
||||
try table.createIndex("city");
|
||||
try table.put(.{ .int = 1 }, &.{.{ .name = "city", .value = .{ .text = "warsaw" } }});
|
||||
try table.put(.{ .int = 2 }, &.{.{ .name = "city", .value = .{ .text = "warsaw" } }});
|
||||
try table.put(.{ .int = 3 }, &.{.{ .name = "city", .value = .{ .text = "krakow" } }});
|
||||
}
|
||||
|
||||
const address = try std.net.Address.parseIp("127.0.0.1", 0);
|
||||
var server = try Server.init(allocator, &db, address);
|
||||
try server.start();
|
||||
defer server.stop();
|
||||
|
||||
const client_mod = @import("client.zig");
|
||||
var client = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||
defer client.close();
|
||||
|
||||
const deleted = try client.tableDelete("users", "city eq s:warsaw");
|
||||
try std.testing.expectEqual(@as(u64, 2), deleted);
|
||||
|
||||
// The index entries for both deleted rows are gone, not just the rows -
|
||||
// a stale entry here would surface as a phantom in a later select.
|
||||
const gone = try client.tableSelect(allocator, "users", "city eq s:warsaw", 0, 0, &.{});
|
||||
defer protocol.freeRows(allocator, gone);
|
||||
try std.testing.expectEqual(@as(usize, 0), gone.len);
|
||||
|
||||
// The row not matched by the predicate (pk 3, krakow) was left untouched.
|
||||
const untouched = try client.tableSelect(allocator, "users", "city eq s:krakow", 0, 0, &.{});
|
||||
defer protocol.freeRows(allocator, untouched);
|
||||
try std.testing.expectEqual(@as(usize, 1), untouched.len);
|
||||
|
||||
// A fresh row can reuse the "warsaw" value without colliding with a
|
||||
// leftover index entry from the deleted rows.
|
||||
{
|
||||
var table = try querymod.Table.open(&db, allocator, "users");
|
||||
defer table.deinit();
|
||||
try table.put(.{ .int = 4 }, &.{.{ .name = "city", .value = .{ .text = "warsaw" } }});
|
||||
}
|
||||
const reused = try client.tableSelect(allocator, "users", "city eq s:warsaw", 0, 0, &.{});
|
||||
defer protocol.freeRows(allocator, reused);
|
||||
try std.testing.expectEqual(@as(usize, 1), reused.len);
|
||||
try std.testing.expectEqual(@as(i64, 4), reused[0].get(querymod.pkField).?.int);
|
||||
}
|
||||
|
||||
test "server table_delete with an empty expr removes every row" {
|
||||
const allocator = std.testing.allocator;
|
||||
var tmp = std.testing.tmpDir(.{});
|
||||
defer tmp.cleanup();
|
||||
const path = try tmp.dir.realpathAlloc(allocator, ".");
|
||||
defer allocator.free(path);
|
||||
|
||||
var db = try testDb(allocator, path);
|
||||
defer db.close();
|
||||
|
||||
{
|
||||
var table = try querymod.Table.open(&db, allocator, "counters");
|
||||
defer table.deinit();
|
||||
try table.put(.{ .int = 1 }, &.{.{ .name = "n", .value = .{ .int = 5 } }});
|
||||
try table.put(.{ .int = 2 }, &.{.{ .name = "n", .value = .{ .int = 7 } }});
|
||||
}
|
||||
|
||||
const address = try std.net.Address.parseIp("127.0.0.1", 0);
|
||||
var server = try Server.init(allocator, &db, address);
|
||||
try server.start();
|
||||
defer server.stop();
|
||||
|
||||
const client_mod = @import("client.zig");
|
||||
var client = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||
defer client.close();
|
||||
|
||||
const deleted = try client.tableDelete("counters", "");
|
||||
try std.testing.expectEqual(@as(u64, 2), deleted);
|
||||
|
||||
const everyone = try client.tableSelect(allocator, "counters", "", 0, 0, &.{});
|
||||
defer protocol.freeRows(allocator, everyone);
|
||||
try std.testing.expectEqual(@as(usize, 0), everyone.len);
|
||||
}
|
||||
|
||||
test "server table_delete rejects a malformed expr with bad_request, not a crash" {
|
||||
const allocator = std.testing.allocator;
|
||||
var tmp = std.testing.tmpDir(.{});
|
||||
defer tmp.cleanup();
|
||||
const path = try tmp.dir.realpathAlloc(allocator, ".");
|
||||
defer allocator.free(path);
|
||||
|
||||
var db = try testDb(allocator, path);
|
||||
defer db.close();
|
||||
|
||||
{
|
||||
var table = try querymod.Table.open(&db, allocator, "users");
|
||||
defer table.deinit();
|
||||
try table.put(.{ .int = 1 }, &.{});
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user