diff --git a/src/server.zig b/src/server.zig index 969b56b..88f6a85 100644 --- a/src/server.zig +++ b/src/server.zig @@ -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; } + 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); +