From 977e1b0363cf4b79884f27d824f4d9856dc786a9 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Fri, 11 Sep 2026 20:16:02 +0200 Subject: [PATCH] feat: add DeleterCtx and deleterWorker to server --- src/server.zig | 154 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 154 insertions(+) diff --git a/src/server.zig b/src/server.zig index 33e8c23..8e3dc52 100644 --- a/src/server.zig +++ b/src/server.zig @@ -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); +