feat: add MutatorCtx and mutatorWorker to server
This commit is contained in:
dualfroz committed
1 parent
0beb78c942
commit
b056b0901e
1 file changed
+159
+159
@@ -4856,3 +4856,162 @@ test "table_aggregate survives concurrent flush and compact reshaping the underl
|
|||||||
if (g.values[1].int != expectedSum(g.group.text)) return;
|
if (g.values[1].int != expectedSum(g.group.text)) return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const global = client.tableAggregateGlobal(ctx.allocator, "orders", &specs, "", "") catch return;
|
||||||
|
defer protocol.freeAggValues(ctx.allocator, global);
|
||||||
|
if (global[0].int != rowCount) return;
|
||||||
|
if (global[1].int != 465) return;
|
||||||
|
}
|
||||||
|
ctx.ok = true;
|
||||||
|
}
|
||||||
|
}.run;
|
||||||
|
|
||||||
|
const MutatorCtx = struct {
|
||||||
|
allocator: Allocator,
|
||||||
|
addr: std.net.Address,
|
||||||
|
thread: usize,
|
||||||
|
ok: bool = false,
|
||||||
|
};
|
||||||
|
const mutatorWorker = struct {
|
||||||
|
fn run(ctx: *MutatorCtx) void {
|
||||||
|
var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return;
|
||||||
|
defer client.close();
|
||||||
|
var i: usize = 0;
|
||||||
|
while (i < mutatorIters) : (i += 1) {
|
||||||
|
var buf: [32]u8 = undefined;
|
||||||
|
const key = std.fmt.bufPrint(&buf, "agg-churn-{d}-{d}", .{ ctx.thread, i }) catch return;
|
||||||
|
client.set(key, "x") catch return;
|
||||||
|
client.flush() catch return;
|
||||||
|
client.compact() catch 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 mutatorContexts: [mutatorThreads]MutatorCtx = undefined;
|
||||||
|
var mutatorHandles: [mutatorThreads]std.Thread = undefined;
|
||||||
|
for (0..mutatorThreads) |i| {
|
||||||
|
mutatorContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress(), .thread = i };
|
||||||
|
mutatorHandles[i] = try std.Thread.spawn(.{}, mutatorWorker, .{&mutatorContexts[i]});
|
||||||
|
}
|
||||||
|
|
||||||
|
for (0..readerThreads) |i| readerHandles[i].join();
|
||||||
|
for (0..mutatorThreads) |i| mutatorHandles[i].join();
|
||||||
|
for (0..readerThreads) |i| try std.testing.expect(readerContexts[i].ok);
|
||||||
|
for (0..mutatorThreads) |i| try std.testing.expect(mutatorContexts[i].ok);
|
||||||
|
|
||||||
|
// Final sanity check on a fresh connection: after all the concurrent
|
||||||
|
// churn, the aggregate results are still exactly what they were before
|
||||||
|
// the server ever started - the churn wrote only unrelated
|
||||||
|
// `agg-churn-*` keys and reshaped physical sstable layout, never touched
|
||||||
|
// `orders` values.
|
||||||
|
var client = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||||
|
defer client.close();
|
||||||
|
const specs = [_]aggregatemod.AggSpec{.{ .field = "", .kind = .count }};
|
||||||
|
const finalGroups = try client.tableAggregateGroups(allocator, "orders", "city", &specs, 0, 0, &.{}, "", "");
|
||||||
|
defer protocol.freeAggGroupRows(allocator, finalGroups);
|
||||||
|
try std.testing.expectEqual(@as(usize, 3), finalGroups.len);
|
||||||
|
for (finalGroups) |g| try std.testing.expectEqual(expectedCount(g.group.text), g.values[0].int);
|
||||||
|
}
|
||||||
|
|
||||||
|
fn seedJoinTables(allocator: Allocator, db: *Db) !void {
|
||||||
|
var customers = try querymod.Table.open(db, allocator, "customers");
|
||||||
|
defer customers.deinit();
|
||||||
|
try customers.put(.{ .int = 1 }, &.{.{ .name = "name", .value = .{ .text = "alice" } }});
|
||||||
|
try customers.put(.{ .int = 2 }, &.{.{ .name = "name", .value = .{ .text = "bob" } }});
|
||||||
|
try customers.put(.{ .int = 3 }, &.{.{ .name = "name", .value = .{ .text = "carol" } }});
|
||||||
|
|
||||||
|
var orders = try querymod.Table.open(db, allocator, "orders");
|
||||||
|
defer orders.deinit();
|
||||||
|
try orders.put(.{ .int = 101 }, &.{
|
||||||
|
.{ .name = "customer_id", .value = .{ .int = 1 } },
|
||||||
|
.{ .name = "total", .value = .{ .int = 50 } },
|
||||||
|
});
|
||||||
|
try orders.put(.{ .int = 102 }, &.{
|
||||||
|
.{ .name = "customer_id", .value = .{ .int = 1 } },
|
||||||
|
.{ .name = "total", .value = .{ .int = 30 } },
|
||||||
|
});
|
||||||
|
try orders.put(.{ .int = 103 }, &.{
|
||||||
|
.{ .name = "customer_id", .value = .{ .int = 999 } },
|
||||||
|
.{ .name = "total", .value = .{ .int = 999 } },
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
test "server table_join runs a remote inner join, using an index when available" {
|
||||||
|
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();
|
||||||
|
try seedJoinTables(allocator, &db);
|
||||||
|
|
||||||
|
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 noIndex = try client.tableJoin(allocator, .inner, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "");
|
||||||
|
defer protocol.freeJoinPairs(allocator, noIndex);
|
||||||
|
try std.testing.expectEqual(@as(usize, 2), noIndex.len);
|
||||||
|
for (noIndex) |p| {
|
||||||
|
try std.testing.expect(p.left != null);
|
||||||
|
try std.testing.expect(p.right != null);
|
||||||
|
try std.testing.expectEqualStrings("alice", p.left.?.get("name").?.text);
|
||||||
|
}
|
||||||
|
|
||||||
|
{
|
||||||
|
var orders = try querymod.Table.open(&db, allocator, "orders");
|
||||||
|
defer orders.deinit();
|
||||||
|
try orders.createIndex("customer_id");
|
||||||
|
}
|
||||||
|
const withIndex = try client.tableJoin(allocator, .inner, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "");
|
||||||
|
defer protocol.freeJoinPairs(allocator, withIndex);
|
||||||
|
try std.testing.expectEqual(@as(usize, 2), withIndex.len);
|
||||||
|
}
|
||||||
|
|
||||||
|
test "server table_join runs left, right, and full outer variants with predicates" {
|
||||||
|
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();
|
||||||
|
try seedJoinTables(allocator, &db);
|
||||||
|
|
||||||
|
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 left = try client.tableJoin(allocator, .left, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "");
|
||||||
|
defer protocol.freeJoinPairs(allocator, left);
|
||||||
|
try std.testing.expectEqual(@as(usize, 4), left.len);
|
||||||
|
var carolUnmatched = false;
|
||||||
|
for (left) |p| {
|
||||||
|
if (p.left) |l| {
|
||||||
|
if (std.mem.eql(u8, l.get("name").?.text, "carol")) {
|
||||||
|
try std.testing.expect(p.right == null);
|
||||||
|
carolUnmatched = true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
try std.testing.expect(carolUnmatched);
|
||||||
|
|
||||||
Reference in new issue
Block a user