feat: add DeleterCtx and deleterWorker to server
This commit is contained in:
dualfroz committed
1 parent
f34107ef56
commit
977e1b0363
1 file changed
+154
+154
@@ -3593,3 +3593,157 @@ test "table_delete survives concurrent flush and compact reshaping the underlyin
|
||||
}
|
||||
}.run;
|
||||
|
||||
const DeleterCtx = struct {
|
||||
allocator: Allocator,
|
||||
addr: std.net.Address,
|
||||
ok: bool = false,
|
||||
};
|
||||
const deleterWorker = struct {
|
||||
fn run(ctx: *DeleterCtx) void {
|
||||
var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return;
|
||||
defer client.close();
|
||||
var seq: i64 = 0;
|
||||
while (seq < poznanRows) : (seq += 1) {
|
||||
const expr = std.fmt.allocPrint(ctx.allocator, "city eq s:poznan and seq eq i:{d}", .{seq}) catch return;
|
||||
defer ctx.allocator.free(expr);
|
||||
const deleted = client.tableDelete("accounts", expr) catch return;
|
||||
if (deleted != 1) return;
|
||||
}
|
||||
ctx.ok = true;
|
||||
}
|
||||
}.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 deleterContext = DeleterCtx{ .allocator = allocator, .addr = server.boundAddress() };
|
||||
const deleterHandle = try std.Thread.spawn(.{}, deleterWorker, .{&deleterContext});
|
||||
|
||||
for (0..readerThreads) |i| readerHandles[i].join();
|
||||
for (0..reshapeThreads) |i| reshapeHandles[i].join();
|
||||
deleterHandle.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(deleterContext.ok);
|
||||
|
||||
// Final sanity check on a fresh connection: the three stable buckets are
|
||||
// still exactly 10 rows each, poznan is completely gone (not just its
|
||||
// index entries - real rows too), and no phantom entry survives under
|
||||
// the "poznan" index bucket for a later put to collide with.
|
||||
var client = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||
defer client.close();
|
||||
|
||||
inline for (.{ "warsaw", "krakow", "gdansk" }) |city| {
|
||||
const buf = try std.fmt.allocPrint(allocator, "city eq s:{s}", .{city});
|
||||
defer allocator.free(buf);
|
||||
const rows = try client.tableSelect(allocator, "accounts", buf, 0, 0, &.{});
|
||||
defer protocol.freeRows(allocator, rows);
|
||||
try std.testing.expectEqual(@as(usize, stableBucketRows), rows.len);
|
||||
}
|
||||
|
||||
const poznanRowsLeft = try client.tableSelect(allocator, "accounts", "city eq s:poznan", 0, 0, &.{});
|
||||
defer protocol.freeRows(allocator, poznanRowsLeft);
|
||||
try std.testing.expectEqual(@as(usize, 0), poznanRowsLeft.len);
|
||||
|
||||
var plan = try client.tableExplain(allocator, "accounts", "city eq s:poznan");
|
||||
defer plan.deinit(allocator);
|
||||
try std.testing.expectEqual(querymod.PlanKind.index_eq, plan.kind);
|
||||
}
|
||||
|
||||
test "server table_select pages a remote result with offset and limit" {
|
||||
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();
|
||||
var pk: i64 = 1;
|
||||
while (pk <= 5) : (pk += 1) try table.put(.{ .int = pk }, &.{});
|
||||
}
|
||||
|
||||
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 page = try client.tableSelect(allocator, "users", "", 1, 2, &.{});
|
||||
defer protocol.freeRows(allocator, page);
|
||||
try std.testing.expectEqual(@as(usize, 2), page.len);
|
||||
try std.testing.expectEqual(@as(i64, 2), page[0].get(querymod.pkField).?.int);
|
||||
try std.testing.expectEqual(@as(i64, 3), page[1].get(querymod.pkField).?.int);
|
||||
|
||||
const rest = try client.tableSelect(allocator, "users", "", 3, 0, &.{});
|
||||
defer protocol.freeRows(allocator, rest);
|
||||
try std.testing.expectEqual(@as(usize, 2), rest.len);
|
||||
try std.testing.expectEqual(@as(i64, 4), rest[0].get(querymod.pkField).?.int);
|
||||
try std.testing.expectEqual(@as(i64, 5), rest[1].get(querymod.pkField).?.int);
|
||||
}
|
||||
|
||||
// `table_explain` reports which strategy `selectPage` would use, but
|
||||
// `table_select`'s own `--order-by` is the one place the network protocol
|
||||
// carries a sort key: `handleTableSelect` forwards `args.orderBy` into
|
||||
// `Table.selectPageOrdered`, which must sort BEFORE windowing by
|
||||
// offset/limit (see query.zig's "applies offset/limit after sorting"
|
||||
// test). This exercises that ordering end to end over a real socket,
|
||||
// including on a field that has an index (`age`) - `selectAllMatches`
|
||||
// still narrows through the index, it just cannot early-stop while a sort
|
||||
// is pending, and the wire round trip must not lose that.
|
||||
test "server table_select orders a remote result by a field before windowing it, ascending and descending" {
|
||||
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("age");
|
||||
// pk order deliberately not age order: pk 1 has the highest age
|
||||
// (50), pk 5 the lowest (10) - a bug that sorted only within pk
|
||||
// scan order would fail every assertion below.
|
||||
var pk: i64 = 1;
|
||||
while (pk <= 5) : (pk += 1) {
|
||||
try table.put(.{ .int = pk }, &.{.{ .name = "age", .value = .{ .int = 60 - pk * 10 } }});
|
||||
}
|
||||
}
|
||||
|
||||
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 asc = try client.tableSelect(allocator, "users", "", 0, 0, &.{.{ .field = "age" }});
|
||||
defer protocol.freeRows(allocator, asc);
|
||||
try std.testing.expectEqual(@as(usize, 5), asc.len);
|
||||
try std.testing.expectEqual(@as(i64, 5), asc[0].get(querymod.pkField).?.int);
|
||||
try std.testing.expectEqual(@as(i64, 10), asc[0].get("age").?.int);
|
||||
try std.testing.expectEqual(@as(i64, 1), asc[4].get(querymod.pkField).?.int);
|
||||
try std.testing.expectEqual(@as(i64, 50), asc[4].get("age").?.int);
|
||||
|
||||
Reference in new issue
Block a user