lib/sql/src/profiling/bulk.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const bench = @import("bench");
3 const sql = @import("sql");
4 const sys = @import("sys");
5
6 pub const small_row_count = 1_024;
7 pub const medium_row_count = 8_192;
8 pub const large_row_count = 32_768;
9 pub const delta_seed_count = 32_768;
10 pub const delta_change_count = 256;
11
12 const insert_source = "INSERT INTO items (rowid, ordinal, name, payload) VALUES (?1, ?2, ?3, ?4)";
13
14 pub fn addTo(suite: *bench.Suite) !void {
15 try suite.add("sql bulk insert individual flush 1k", loadBench(.individual, small_row_count), .{});
16 try suite.add("sql bulk insert staged write 1k", loadBench(.staged, small_row_count), .{});
17 try suite.add("sql bulk insert staged write 8k", loadBench(.staged, medium_row_count), .{});
18 try suite.add("sql bulk insert staged write 32k", loadBench(.staged, large_row_count), .{});
19 try suite.add("sql bulk insert session put 1k", loadBench(.session, small_row_count), .{});
20 try suite.add("sql bulk insert session put 8k", loadBench(.session, medium_row_count), .{});
21 try suite.add("sql bulk insert session put 32k", loadBench(.session, large_row_count), .{});
22 try suite.add("sql delta insert staged write 32k+256", deltaBench(.staged), .{});
23 try suite.add("sql delta insert session put 32k+256", deltaBench(.session), .{});
24 }
25
26 const Strategy = enum {
27 individual,
28 staged,
29 session,
30
31 fn label(self: Strategy) []const u8 {
32 return switch (self) {
33 .individual => "individual",
34 .staged => "staged",
35 .session => "session",
36 };
37 }
38 };
39
40 const Load = struct {
41 wal_frames: usize,
42 flushes: usize,
43 checkpoints: usize,
44 };
45
46 fn bulkHeader() sql.wal.Header {
47 return .{
48 .sequence = 81,
49 .salt = .{ .first = 0x6c75_4b42, .second = 0x424b_756c },
50 };
51 }
52
53 fn LoadBenchType(comptime strategy: Strategy, comptime rows: usize) type {
54 return struct {
55 fn run(sample_allocator: std.mem.Allocator) void {
56 const load = runLoad(strategy, rows, sample_allocator);
57 std.mem.doNotOptimizeAway(load.wal_frames +% load.flushes +% load.checkpoints);
58 }
59 };
60 }
61
62 fn loadBench(comptime strategy: Strategy, comptime rows: usize) *const fn (std.mem.Allocator) void {
63 return LoadBenchType(strategy, rows).run;
64 }
65
66 fn runLoad(comptime strategy: Strategy, comptime rows: usize, sample_allocator: std.mem.Allocator) Load {
67 const workload = comptime strategy.label() ++ std.fmt.comptimePrint(" {d}", .{rows});
68
69 var tmp = std.testing.tmpDir(.{});
70 defer tmp.cleanup();
71
72 var database = sql.FileDatabase.openForTesting(sample_allocator, tmp.dir, .{
73 .paths = .{ .database = "bulk.db", .wal = "bulk.wal" },
74 .header = bulkHeader(),
75 }) catch |err| std.debug.panic("bulk database open failed: {t}", .{err});
76 defer database.deinit();
77 database.reserve(.{
78 .wal_frames = walReserve(strategy, rows),
79 }) catch |err| std.debug.panic("bulk database reserve failed: {t}", .{err});
80
81 var catalog = sql.Catalog.open(&database, .{}) catch |err| std.debug.panic("bulk catalog open failed: {t}", .{err});
82 var connection: sql.Connection = undefined;
83 {
84 const prepare_phase = bench.phaseAt("sql.bulk.load.prepare", @src());
85 prepare_phase.setName(workload);
86 defer prepare_phase.end();
87 connection = bulkConnection(sample_allocator, &catalog);
88 executeBulkCatalogStatement(sample_allocator, &connection, "CREATE TABLE items (ordinal, name, payload)");
89 }
90 defer connection.deinit();
91
92 const start_frames = database.pager.frameCount();
93 var checkpoints: usize = 0;
94 const flushes = switch (strategy) {
95 .individual => loadIndividual(&database, &connection, sample_allocator, rows, workload, &checkpoints),
96 .staged => loadStaged(&connection, sample_allocator, rows, workload),
97 .session => loadSession(&connection, &catalog, sample_allocator, rows, workload),
98 };
99
100 {
101 const sync_phase = bench.phaseAt("sql.bulk.load.sync", @src());
102 sync_phase.setName(workload);
103 defer sync_phase.end();
104 database.syncWal() catch |err| std.debug.panic("bulk wal sync failed: {t}", .{err});
105 }
106
107 verifyLoaded(&connection, sample_allocator, rows);
108 return .{
109 .wal_frames = database.pager.frameCount() - start_frames,
110 .flushes = flushes,
111 .checkpoints = checkpoints,
112 };
113 }
114
115 fn loadIndividual(database: *sql.FileDatabase, connection: *sql.Connection, sample_allocator: std.mem.Allocator, comptime rows: usize, comptime workload: []const u8, checkpoints: *usize) usize {
116 var insert = connection.prepare(sample_allocator, insert_source) catch |err| std.debug.panic("bulk individual prepare failed: {t}", .{err});
117 defer insert.deinit();
118
119 const rows_phase = bench.phaseAt("sql.bulk.load.rows", @src());
120 rows_phase.setName(workload);
121 defer rows_phase.end();
122
123 var index: usize = 0;
124 while (index < rows) : (index += 1) {
125 bindRow(&insert, index);
126 var result = executeIndividualRow(database, connection, sample_allocator, &insert, index, checkpoints);
127 result.deinit(sample_allocator);
128 bench.coz.progressNamed("sql.bulk.load.row");
129 }
130 return rows;
131 }
132
133 fn executeIndividualRow(
134 database: *sql.FileDatabase,
135 connection: *sql.Connection,
136 sample_allocator: std.mem.Allocator,
137 insert: *sql.PreparedStatement,
138 index: usize,
139 checkpoints: *usize,
140 ) sql.StatementResult {
141 return connection.executePrepared(insert, sample_allocator, .{
142 .durability = .buffered,
143 }) catch |err| switch (err) {
144 error.WalLimitExceeded => {
145 _ = database.checkpoint(.{ .restart_header = bulkHeader() }) catch |checkpoint_err|
146 std.debug.panic("bulk individual checkpoint failed: {t}", .{checkpoint_err});
147 checkpoints.* += 1;
148 bindRow(insert, index);
149 return connection.executePrepared(insert, sample_allocator, .{
150 .durability = .buffered,
151 }) catch |retry_err| std.debug.panic("bulk individual retry failed: {t}", .{retry_err});
152 },
153 else => std.debug.panic("bulk individual execute failed: {t}", .{err}),
154 };
155 }
156
157 fn loadStaged(connection: *sql.Connection, sample_allocator: std.mem.Allocator, comptime rows: usize, comptime workload: []const u8) usize {
158 var insert = connection.prepare(sample_allocator, insert_source) catch |err| std.debug.panic("bulk staged prepare failed: {t}", .{err});
159 defer insert.deinit();
160 const limits = stagedLimits(0, rows);
161 var workspace = sql.DatabaseWrite.Workspace.allocate(
162 sample_allocator,
163 limits,
164 ) catch |err| std.debug.panic("bulk staged workspace failed: {t}", .{err});
165 defer workspace.deallocate(sample_allocator);
166 var write = connection.beginWrite(&workspace, sample_allocator, .{
167 .limits = limits,
168 .durability = .buffered,
169 }) catch |err| std.debug.panic("bulk staged begin failed: {t}", .{err});
170 defer write.deinit();
171
172 {
173 const rows_phase = bench.phaseAt("sql.bulk.load.rows", @src());
174 rows_phase.setName(workload);
175 defer rows_phase.end();
176 var index: usize = 0;
177 while (index < rows) : (index += 1) {
178 bindRow(&insert, index);
179 var result = connection.executePrepared(&insert, sample_allocator, .{
180 .durability = .buffered,
181 .write = &write,
182 }) catch |err| std.debug.panic("bulk staged execute failed: {t}", .{err});
183 result.deinit(sample_allocator);
184 bench.coz.progressNamed("sql.bulk.load.row");
185 }
186 }
187
188 const flush_phase = bench.phaseAt("sql.bulk.load.flush", @src());
189 flush_phase.setName(workload);
190 defer flush_phase.end();
191 var flush = write.flush() catch |err| std.debug.panic("bulk staged flush failed: {t}", .{err});
192 flush.deinit();
193 return 1;
194 }
195
196 fn loadSession(connection: *sql.Connection, catalog: *const sql.Catalog, sample_allocator: std.mem.Allocator, comptime rows: usize, comptime workload: []const u8) usize {
197 var relation_session = sql.RelationSession.open(sample_allocator, catalog, "items") catch |err| std.debug.panic("bulk session open failed: {t}", .{err});
198 var relation_session_live = true;
199 defer if (relation_session_live) relation_session.deinit();
200
201 {
202 const rows_phase = bench.phaseAt("sql.bulk.load.rows", @src());
203 rows_phase.setName(workload);
204 defer rows_phase.end();
205 var index: usize = 0;
206 while (index < rows) : (index += 1) {
207 var text_buffer: [24]u8 = undefined;
208 const name = rowName(&text_buffer, index);
209 relation_session.put(@intCast(index + 1), &.{
210 .{ .integer = @intCast(index) },
211 .{ .text = name },
212 .{ .integer = @intCast(index % 13) },
213 }) catch |err| std.debug.panic("bulk session put failed: {t}", .{err});
214 bench.coz.progressNamed("sql.bulk.load.row");
215 }
216 }
217
218 const flush_phase = bench.phaseAt("sql.bulk.load.flush", @src());
219 flush_phase.setName(workload);
220 defer flush_phase.end();
221 const limits = relation_session.stagingLimits() catch |err|
222 std.debug.panic("bulk session limits failed: {t}", .{err});
223 var workspace = sql.DatabaseWrite.Workspace.allocate(
224 sample_allocator,
225 limits,
226 ) catch |err| std.debug.panic("bulk session workspace failed: {t}", .{err});
227 defer workspace.deallocate(sample_allocator);
228 var write = connection.beginWrite(&workspace, sample_allocator, .{
229 .limits = limits,
230 .durability = .buffered,
231 }) catch |err| std.debug.panic("bulk session begin failed: {t}", .{err});
232 defer write.deinit();
233 write.stageRelation(&relation_session) catch |err| std.debug.panic("bulk session stage failed: {t}", .{err});
234 relation_session_live = false;
235 var flush = write.flush() catch |err| std.debug.panic("bulk session flush failed: {t}", .{err});
236 flush.deinit();
237 return 1;
238 }
239
240 const DeltaLoad = struct {
241 seed_ns: u64,
242 delta_ns: u64,
243 };
244
245 fn DeltaBenchType(comptime strategy: Strategy) type {
246 return struct {
247 fn run(sample_allocator: std.mem.Allocator) void {
248 const load = runDelta(strategy, delta_seed_count, delta_change_count, sample_allocator);
249 std.mem.doNotOptimizeAway(load.seed_ns +% load.delta_ns);
250 }
251 };
252 }
253
254 fn deltaBench(comptime strategy: Strategy) *const fn (std.mem.Allocator) void {
255 return DeltaBenchType(strategy).run;
256 }
257
258 fn runDelta(comptime strategy: Strategy, comptime seed_rows: usize, comptime delta_rows: usize, sample_allocator: std.mem.Allocator) DeltaLoad {
259 const workload = comptime "delta " ++ strategy.label() ++ std.fmt.comptimePrint(" {d}+{d}", .{ seed_rows, delta_rows });
260
261 var tmp = std.testing.tmpDir(.{});
262 defer tmp.cleanup();
263
264 var database = sql.FileDatabase.openForTesting(sample_allocator, tmp.dir, .{
265 .paths = .{ .database = "delta.db", .wal = "delta.wal" },
266 .header = .{
267 .sequence = 82,
268 .salt = .{ .first = 0x6c65_4464, .second = 0x6444_656c },
269 },
270 }) catch |err| std.debug.panic("delta database open failed: {t}", .{err});
271 defer database.deinit();
272 database.reserve(.{
273 .wal_frames = seed_rows / 8 + 1_024,
274 }) catch |err| std.debug.panic("delta database reserve failed: {t}", .{err});
275
276 var catalog = sql.Catalog.open(&database, .{}) catch |err| std.debug.panic("delta catalog open failed: {t}", .{err});
277 var connection = bulkConnection(sample_allocator, &catalog);
278 defer connection.deinit();
279 executeBulkCatalogStatement(sample_allocator, &connection, "CREATE TABLE items (ordinal, name, payload)");
280 const workspace_limits = stagedLimits(0, seed_rows);
281 var workspace = sql.DatabaseWrite.Workspace.allocate(
282 sample_allocator,
283 workspace_limits,
284 ) catch |err| std.debug.panic("delta workspace failed: {t}", .{err});
285 defer workspace.deallocate(sample_allocator);
286
287 const seed_start = bench.nowNs();
288 {
289 const seed_phase = bench.phaseAt("sql.delta.seed", @src());
290 seed_phase.setName(workload);
291 defer seed_phase.end();
292 seedRows(&connection, &catalog, &workspace, sample_allocator, seed_rows);
293 }
294 const seed_ns = bench.elapsedNs(seed_start);
295
296 const delta_start = bench.nowNs();
297 {
298 const changes_phase = bench.phaseAt("sql.delta.changes", @src());
299 changes_phase.setName(workload);
300 defer changes_phase.end();
301 switch (strategy) {
302 .staged => applyDeltaStaged(&connection, &workspace, sample_allocator, seed_rows, delta_rows),
303 .session => applyDeltaSession(&connection, &catalog, &workspace, sample_allocator, seed_rows, delta_rows),
304 .individual => unreachable,
305 }
306 }
307 const delta_ns = bench.elapsedNs(delta_start);
308
309 database.syncWal() catch |err| std.debug.panic("delta wal sync failed: {t}", .{err});
310 verifyLoaded(&connection, sample_allocator, seed_rows + delta_rows);
311 return .{
312 .seed_ns = seed_ns,
313 .delta_ns = delta_ns,
314 };
315 }
316
317 fn seedRows(
318 connection: *sql.Connection,
319 catalog: *const sql.Catalog,
320 workspace: *sql.DatabaseWrite.Workspace,
321 sample_allocator: std.mem.Allocator,
322 comptime rows: usize,
323 ) void {
324 var relation_session = sql.RelationSession.open(sample_allocator, catalog, "items") catch |err| std.debug.panic("delta seed open failed: {t}", .{err});
325 var relation_session_live = true;
326 defer if (relation_session_live) relation_session.deinit();
327
328 var index: usize = 0;
329 while (index < rows) : (index += 1) {
330 var text_buffer: [24]u8 = undefined;
331 const name = rowName(&text_buffer, index);
332 relation_session.put(@intCast(index + 1), &.{
333 .{ .integer = @intCast(index) },
334 .{ .text = name },
335 .{ .integer = @intCast(index % 13) },
336 }) catch |err| std.debug.panic("delta seed put failed: {t}", .{err});
337 }
338
339 const limits = relation_session.stagingLimits() catch |err|
340 std.debug.panic("delta seed limits failed: {t}", .{err});
341 var write = connection.beginWrite(workspace, sample_allocator, .{
342 .limits = limits,
343 .durability = .buffered,
344 }) catch |err| std.debug.panic("delta seed begin failed: {t}", .{err});
345 defer write.deinit();
346 write.stageRelation(&relation_session) catch |err| std.debug.panic("delta seed stage failed: {t}", .{err});
347 relation_session_live = false;
348 var flush = write.flush() catch |err| std.debug.panic("delta seed flush failed: {t}", .{err});
349 flush.deinit();
350 }
351
352 fn applyDeltaStaged(
353 connection: *sql.Connection,
354 workspace: *sql.DatabaseWrite.Workspace,
355 sample_allocator: std.mem.Allocator,
356 comptime seed_rows: usize,
357 comptime delta_rows: usize,
358 ) void {
359 var insert = connection.prepare(sample_allocator, insert_source) catch |err| std.debug.panic("delta staged prepare failed: {t}", .{err});
360 defer insert.deinit();
361 var write = connection.beginWrite(workspace, sample_allocator, .{
362 .limits = stagedLimits(seed_rows, delta_rows),
363 .durability = .buffered,
364 }) catch |err| std.debug.panic("delta staged begin failed: {t}", .{err});
365 defer write.deinit();
366
367 var index: usize = seed_rows;
368 while (index < seed_rows + delta_rows) : (index += 1) {
369 bindRow(&insert, index);
370 var result = connection.executePrepared(&insert, sample_allocator, .{
371 .durability = .buffered,
372 .write = &write,
373 }) catch |err| std.debug.panic("delta staged execute failed: {t}", .{err});
374 result.deinit(sample_allocator);
375 bench.coz.progressNamed("sql.delta.change");
376 }
377
378 var flush = write.flush() catch |err| std.debug.panic("delta staged flush failed: {t}", .{err});
379 flush.deinit();
380 }
381
382 fn applyDeltaSession(
383 connection: *sql.Connection,
384 catalog: *const sql.Catalog,
385 workspace: *sql.DatabaseWrite.Workspace,
386 sample_allocator: std.mem.Allocator,
387 comptime seed_rows: usize,
388 comptime delta_rows: usize,
389 ) void {
390 var relation_session = sql.RelationSession.open(sample_allocator, catalog, "items") catch |err| std.debug.panic("delta session open failed: {t}", .{err});
391 var relation_session_live = true;
392 defer if (relation_session_live) relation_session.deinit();
393
394 var index: usize = seed_rows;
395 while (index < seed_rows + delta_rows) : (index += 1) {
396 var text_buffer: [24]u8 = undefined;
397 const name = rowName(&text_buffer, index);
398 relation_session.put(@intCast(index + 1), &.{
399 .{ .integer = @intCast(index) },
400 .{ .text = name },
401 .{ .integer = @intCast(index % 13) },
402 }) catch |err| std.debug.panic("delta session put failed: {t}", .{err});
403 bench.coz.progressNamed("sql.delta.change");
404 }
405
406 const limits = relation_session.stagingLimits() catch |err|
407 std.debug.panic("delta session limits failed: {t}", .{err});
408 var write = connection.beginWrite(workspace, sample_allocator, .{
409 .limits = limits,
410 .durability = .buffered,
411 }) catch |err| std.debug.panic("delta session begin failed: {t}", .{err});
412 defer write.deinit();
413 write.stageRelation(&relation_session) catch |err| std.debug.panic("delta session stage failed: {t}", .{err});
414 relation_session_live = false;
415 var flush = write.flush() catch |err| std.debug.panic("delta session flush failed: {t}", .{err});
416 flush.deinit();
417 }
418
419 fn verifyLoaded(connection: *sql.Connection, sample_allocator: std.mem.Allocator, comptime rows: usize) void {
420 var select = connection.prepare(sample_allocator, "SELECT ordinal FROM items WHERE rowid = ?1") catch |err| std.debug.panic("bulk verify prepare failed: {t}", .{err});
421 defer select.deinit();
422 select.bind(1, .{ .integer = rows }) catch |err| std.debug.panic("bulk verify bind failed: {t}", .{err});
423 var result = connection.executePrepared(&select, sample_allocator, .{ .durability = .buffered }) catch |err| std.debug.panic("bulk verify execute failed: {t}", .{err});
424 defer result.deinit(sample_allocator);
425 const bytes = result.nextRow() orelse @panic("bulk verify missing row");
426 const view = sql.RowView.init(bytes) catch |err| std.debug.panic("bulk verify row view failed: {t}", .{err});
427 const ordinal = (view.column(0) catch |err| std.debug.panic("bulk verify column failed: {t}", .{err})).integer;
428 if (ordinal != rows - 1) @panic("bulk verify ordinal mismatch");
429 }
430
431 fn bindRow(insert: *sql.PreparedStatement, index: usize) void {
432 var text_buffer: [24]u8 = undefined;
433 const name = rowName(&text_buffer, index);
434 insert.bind(1, .{ .integer = @intCast(index + 1) }) catch |err| std.debug.panic("bulk bind rowid failed: {t}", .{err});
435 insert.bind(2, .{ .integer = @intCast(index) }) catch |err| std.debug.panic("bulk bind ordinal failed: {t}", .{err});
436 insert.bind(3, .{ .text = name }) catch |err| std.debug.panic("bulk bind name failed: {t}", .{err});
437 insert.bind(4, .{ .integer = @intCast(index % 13) }) catch |err| std.debug.panic("bulk bind payload failed: {t}", .{err});
438 }
439
440 fn rowName(buffer: *[24]u8, index: usize) []const u8 {
441 return std.fmt.bufPrint(buffer, "name-{d:0>8}", .{index}) catch |err| std.debug.panic("bulk row name failed: {t}", .{err});
442 }
443
444 fn stagedLimits(start: usize, count: usize) sql.DatabaseWrite.Limits {
445 var payload_bytes: usize = 0;
446 for (start..start + count) |index| {
447 var text_buffer: [24]u8 = undefined;
448 const name = rowName(&text_buffer, index);
449 const size = sql.row.encodedSize(&.{
450 .{ .integer = @intCast(index) },
451 .{ .text = name },
452 .{ .integer = @intCast(index % 13) },
453 }) catch |err| std.debug.panic("bulk staging capacity failed: {t}", .{err});
454 payload_bytes = std.math.add(usize, payload_bytes, size) catch
455 @panic("bulk staging capacity overflow");
456 }
457 return .{
458 .relations = 1,
459 .edits = count,
460 .payload_bytes = payload_bytes,
461 .assignments = 0,
462 };
463 }
464
465 fn walReserve(comptime strategy: Strategy, comptime rows: usize) usize {
466 return switch (strategy) {
467 .individual => 4_096,
468 .staged, .session => rows / 8 + 512,
469 };
470 }
471
472 fn bulkDatabaseSession(allocator: std.mem.Allocator, catalog: *const sql.Catalog) sql.DatabaseSession {
473 var root = sql.version.databaseRootMaintained(
474 allocator,
475 catalog,
476 sql.ConflictRoot.empty().hash,
477 ) catch |err| std.debug.panic("bulk database root failed: {t}", .{err});
478 const commit = sql.Commit.init(root.hash, &.{});
479 const root_hash = root.hash;
480 return sql.DatabaseSession.initWithRoot(allocator, sql.branch.checkout(.{
481 .name = "main",
482 .target = commit.hash,
483 }, root_hash), &root);
484 }
485
486 fn bulkConnection(allocator: std.mem.Allocator, catalog: *const sql.Catalog) sql.Connection {
487 return sql.Connection.init(catalog.*, bulkDatabaseSession(allocator, catalog));
488 }
489
490 fn executeBulkCatalogStatement(allocator: std.mem.Allocator, connection: *sql.Connection, source: []const u8) void {
491 var result = connection.execute(allocator, source, .{ .durability = .buffered }) catch |err| std.debug.panic("bulk catalog statement failed: {t}", .{err});
492 result.deinit(allocator);
493 }
494
495 const ScalingRun = struct {
496 strategy: Strategy,
497 rows: usize,
498 };
499
500 const scaling_runs = [_]ScalingRun{
501 .{ .strategy = .individual, .rows = small_row_count },
502 .{ .strategy = .individual, .rows = medium_row_count },
503 .{ .strategy = .staged, .rows = small_row_count },
504 .{ .strategy = .staged, .rows = medium_row_count },
505 .{ .strategy = .staged, .rows = large_row_count },
506 .{ .strategy = .session, .rows = small_row_count },
507 .{ .strategy = .session, .rows = medium_row_count },
508 .{ .strategy = .session, .rows = large_row_count },
509 };
510
511 fn reportScaling(allocator: std.mem.Allocator, comptime strategy: Strategy, comptime rows: usize) void {
512 var counting = bench.CountingAllocator.init(allocator);
513 const start = bench.nowNs();
514 const load = runLoad(strategy, rows, counting.allocator());
515 const wall = bench.elapsedNs(start);
516 bench.stdout("{s:<12} {d:>6} {d:>10.2}ms {d:>9.1} {d:>7} {d:>10} {d:>6} {d:>10} {d:>12}\n", .{
517 strategy.label(),
518 rows,
519 @as(f64, @floatFromInt(wall)) / 1e6,
520 @as(f64, @floatFromInt(wall)) / @as(f64, @floatFromInt(rows)),
521 load.flushes,
522 load.wal_frames,
523 load.checkpoints,
524 counting.counts.alloc_count,
525 counting.counts.alloc_bytes,
526 });
527 }
528
529 test "bench: sql bulk load scaling" {
530 const allocator = sys.allocator.benchmarkAllocator();
531 const filter = try sys.env.getOwned(allocator, "BENCH_FILTER");
532 defer if (filter) |value| allocator.free(value);
533 if (filter) |needle| {
534 if (needle.len != 0 and std.mem.indexOf(u8, "sql bulk load scaling", needle) == null) return;
535 }
536
537 bench.stdout("\nsql bulk load scaling\n", .{});
538 bench.stdout("strategy rows wall ns/row flushes wal frames ckpts allocs alloc bytes\n", .{});
539 bench.stdout("--------------------------------------------------------------------------------------------------\n", .{});
540 inline for (scaling_runs) |run| {
541 reportScaling(allocator, run.strategy, run.rows);
542 }
543 bench.stdout("\n", .{});
544 }
545
546 const delta_runs = [_]Strategy{ .staged, .session };
547
548 fn reportDelta(allocator: std.mem.Allocator, comptime strategy: Strategy) void {
549 const load = runDelta(strategy, delta_seed_count, delta_change_count, allocator);
550 bench.stdout("{s:<12} {d:>6}+{d:<4} {d:>10.2}ms seed {d:>12.2}ms delta {d:>12.1} ns/change\n", .{
551 strategy.label(),
552 delta_seed_count,
553 delta_change_count,
554 @as(f64, @floatFromInt(load.seed_ns)) / 1e6,
555 @as(f64, @floatFromInt(load.delta_ns)) / 1e6,
556 @as(f64, @floatFromInt(load.delta_ns)) / @as(f64, @floatFromInt(delta_change_count)),
557 });
558 }
559
560 test "bench: sql delta apply scaling" {
561 const allocator = sys.allocator.benchmarkAllocator();
562 const filter = try sys.env.getOwned(allocator, "BENCH_FILTER");
563 defer if (filter) |value| allocator.free(value);
564 if (filter) |needle| {
565 if (needle.len != 0 and std.mem.indexOf(u8, "sql delta apply scaling", needle) == null) return;
566 }
567
568 bench.stdout("\nsql delta apply scaling\n", .{});
569 bench.stdout("strategy seed+delta seed delta per change\n", .{});
570 bench.stdout("---------------------------------------------------------------------------\n", .{});
571 inline for (delta_runs) |strategy| {
572 reportDelta(allocator, strategy);
573 }
574 bench.stdout("\n", .{});
575 }