lib/sql/src/session/database.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const sql = @import("../root.zig");
3 const relation_session = @import("relation.zig");
4 const staging = @import("staging/root.zig");
5
6 const branch = sql.branch;
7 const catalog_mod = sql.catalog;
8 const file = sql.file;
9 const history_mod = sql.history;
10 const relation_mod = sql.relation;
11 const row = sql.row;
12 const trace = sql.trace;
13 const version = sql.version;
14
15 const Allocator = std.mem.Allocator;
16
17 const DatabaseStageError = error{
18 NoStagedRelation,
19 NoStagedRoot,
20 StagedRelationRootMismatch,
21 WriteSessionActive,
22 WorkspaceBusy,
23 };
24
25 pub const DatabaseError = relation_session.Error || history_mod.Error || DatabaseStageError;
26
27 pub const CatalogFlush = struct {
28 commit: catalog_mod.Commit,
29 database: version.Hash,
30 };
31
32 pub const DatabaseFlush = struct {
33 allocator: Allocator,
34 relations: []relation_session.RelationFlush,
35 database: version.Hash,
36
37 pub fn deinit(self: *DatabaseFlush) void {
38 if (self.relations.len != 0) self.allocator.free(self.relations);
39 self.* = undefined;
40 }
41
42 pub fn onlyRelation(self: *const DatabaseFlush) relation_session.RelationFlush {
43 std.debug.assert(self.relations.len == 1);
44 return self.relations[0];
45 }
46 };
47
48 const RelationRootFlush = struct {
49 commit: file.Commit,
50 hash: version.Hash,
51 };
52
53 const ActiveWrite = struct {
54 generation: u64,
55 workspace: *staging.Workspace,
56 flush_allocator: Allocator,
57 options: file.CommitOptions,
58 storage: staging.Storage,
59 };
60
61 pub const DatabaseSession = struct {
62 allocator: Allocator,
63 checkout: branch.Checkout,
64 staged_relations: std.ArrayList(StagedRelation) = .empty,
65 working_root: version.DatabaseRoot,
66 pending_edits_base: version.Hash,
67 pending_edits: std.StringArrayHashMapUnmanaged(std.ArrayList(i64)) = .empty,
68 active_write: ?ActiveWrite = null,
69 write_generation: u64 = 0,
70
71 pub fn initWithRoot(session_allocator: Allocator, checkout: branch.Checkout, root: *version.DatabaseRoot) DatabaseSession {
72 std.debug.assert(version.same(root.hash, checkout.working.working));
73 const working_root = root.*;
74 root.* = undefined;
75 return .{
76 .allocator = session_allocator,
77 .checkout = checkout,
78 .working_root = working_root,
79 .pending_edits_base = working_root.hash,
80 };
81 }
82
83 pub fn deinit(self: *DatabaseSession) void {
84 self.discardWrite();
85 self.discardStagedRelations();
86 self.clearPendingEdits();
87 self.pending_edits.deinit(self.allocator);
88 self.working_root.deinit();
89 self.* = undefined;
90 }
91
92 pub fn reinitWithRoot(
93 self: *DatabaseSession,
94 checkout: branch.Checkout,
95 root: *version.DatabaseRoot,
96 ) void {
97 const session_allocator = self.allocator;
98 const write_generation = self.write_generation;
99 self.deinit();
100 self.* = initWithRoot(session_allocator, checkout, root);
101 self.write_generation = write_generation;
102 }
103
104 pub fn applyRoot(self: *DatabaseSession, root: *version.DatabaseRoot) void {
105 self.adoptWorkingRoot(root);
106 self.clearPendingEdits();
107 }
108
109 pub fn createRelation(self: *DatabaseSession, allocator: Allocator, catalog: *const catalog_mod.Catalog, definition: catalog_mod.RelationDefinition, options: file.CommitOptions) DatabaseError!CatalogFlush {
110 const phase = trace.scope("session.database.create_relation");
111 defer phase.end();
112
113 const commit_value = try catalog.createRelation(allocator, definition, options);
114 var root = try self.refreshCatalogRoot(catalog);
115 var root_live = true;
116 errdefer if (root_live) root.deinit();
117 const flush = CatalogFlush{
118 .commit = commit_value,
119 .database = root.hash,
120 };
121 self.applyRoot(&root);
122 root_live = false;
123 return flush;
124 }
125
126 pub fn createIndex(self: *DatabaseSession, allocator: Allocator, catalog: *const catalog_mod.Catalog, table_name: []const u8, definition: catalog_mod.IndexDefinition, options: file.CommitOptions) DatabaseError!CatalogFlush {
127 const phase = trace.scope("session.database.create_index");
128 defer phase.end();
129
130 const commit_value = try catalog.createIndex(allocator, table_name, definition, options);
131 var root = try self.refreshCatalogRoot(catalog);
132 var root_live = true;
133 errdefer if (root_live) root.deinit();
134 const flush = CatalogFlush{
135 .commit = commit_value,
136 .database = root.hash,
137 };
138 self.applyRoot(&root);
139 root_live = false;
140 return flush;
141 }
142
143 pub fn analyzeRelation(self: *DatabaseSession, allocator: Allocator, catalog: *const catalog_mod.Catalog, name: []const u8, options: file.CommitOptions) DatabaseError!CatalogFlush {
144 const phase = trace.scope("session.database.analyze_relation");
145 defer phase.end();
146
147 const commit_value = try catalog.analyzeRelation(allocator, name, options);
148 var root = try self.refreshCatalogRoot(catalog);
149 var root_live = true;
150 errdefer if (root_live) root.deinit();
151 const flush = CatalogFlush{
152 .commit = commit_value,
153 .database = root.hash,
154 };
155 self.applyRoot(&root);
156 root_live = false;
157 return flush;
158 }
159
160 pub fn dropRelation(self: *DatabaseSession, allocator: Allocator, catalog: *const catalog_mod.Catalog, name: []const u8, options: file.CommitOptions) DatabaseError!CatalogFlush {
161 const phase = trace.scope("session.database.drop_relation");
162 defer phase.end();
163
164 const commit_value = try catalog.dropRelation(allocator, name, options);
165 var root = try self.refreshCatalogRoot(catalog);
166 var root_live = true;
167 errdefer if (root_live) root.deinit();
168 const flush = CatalogFlush{
169 .commit = commit_value,
170 .database = root.hash,
171 };
172 self.applyRoot(&root);
173 root_live = false;
174 return flush;
175 }
176
177 pub fn beginWrite(
178 self: *DatabaseSession,
179 workspace: *DatabaseWrite.Workspace,
180 flush_allocator: Allocator,
181 limits: DatabaseWrite.Limits,
182 options: file.CommitOptions,
183 ) DatabaseError!DatabaseWrite {
184 const phase = trace.scope("session.database.begin_write");
185 defer phase.end();
186
187 if (self.active_write != null) return error.WriteSessionActive;
188 std.debug.assert(self.staged_relations.items.len == 0);
189 const generation = std.math.add(u64, self.write_generation, 1) catch
190 return error.CapacityOverflow;
191 const storage = try workspace.acquire(limits);
192 self.write_generation = generation;
193 self.active_write = .{
194 .generation = generation,
195 .workspace = workspace,
196 .flush_allocator = flush_allocator,
197 .options = options,
198 .storage = storage,
199 };
200 return .{
201 .session = self,
202 .generation = generation,
203 };
204 }
205
206 fn discardWrite(self: *DatabaseSession) void {
207 if (self.active_write) |*active_write| {
208 active_write.workspace.release(&active_write.storage);
209 self.active_write = null;
210 }
211 }
212
213 fn stageRelation(
214 self: *DatabaseSession,
215 storage: *staging.Storage,
216 relation: *relation_session.RelationSession,
217 ) DatabaseError!void {
218 const phase = trace.scope("session.database.stage_relation");
219 defer phase.end();
220
221 const demand = try staging.Storage.demandForEdits(relation.edits.items);
222 try storage.ensure(demand);
223 if (self.stagedRelationIndex(relation.name)) |relation_index| {
224 const staged = &self.staged_relations.items[relation_index];
225 if (!version.same(staged.root.hash, relation.root.hash)) {
226 return error.StagedRelationRootMismatch;
227 }
228 storage.appendEditsAssumeCapacity(relation_index, relation.edits.items);
229 relation.deinit();
230 return;
231 }
232 try storage.ensureRelation();
233 try self.staged_relations.ensureUnusedCapacity(self.allocator, 1);
234 const staged = StagedRelation.takeMetadata(relation);
235 self.staged_relations.appendAssumeCapacity(staged);
236 const relation_index = storage.registerRelation();
237 std.debug.assert(relation_index == self.staged_relations.items.len - 1);
238 storage.appendEditsAssumeCapacity(relation_index, relation.edits.items);
239 relation_session.finishStaging(relation);
240 }
241
242 fn flushStagedRelations(
243 self: *DatabaseSession,
244 flush_allocator: Allocator,
245 options: file.CommitOptions,
246 storage: *staging.Storage,
247 ) DatabaseError!DatabaseFlush {
248 const phase = trace.scope("session.database.flush_staged_relations");
249 defer phase.end();
250
251 if (self.staged_relations.items.len == 0) return error.NoStagedRelation;
252 errdefer self.discardStagedRelations();
253
254 const relation_flushes = try flush_allocator.alloc(relation_session.RelationFlush, self.staged_relations.items.len);
255 var flush_count: usize = 0;
256 errdefer flush_allocator.free(relation_flushes);
257
258 const replacements = try flush_allocator.alloc(version.RelationEntry, self.staged_relations.items.len);
259 defer flush_allocator.free(replacements);
260
261 storage.prepareOrdered();
262 for (
263 self.staged_relations.items,
264 relation_flushes,
265 replacements,
266 0..,
267 ) |*relation, *flush, *replacement, relation_index| {
268 const edits = storage.orderedFor(relation_index);
269 try self.checkStagedBase(relation);
270 try self.recordPendingEdits(relation.name, edits);
271 const root_flush = try relation.flushRoot(flush_allocator, options, edits);
272 replacement.* = .{
273 .name = relation.name,
274 .hash = root_flush.hash,
275 };
276 flush.* = .{
277 .commit = root_flush.commit,
278 .relation = root_flush.hash,
279 .database = undefined,
280 };
281 flush_count += 1;
282 }
283
284 var root = try version.databaseRootReplacingEntries(self.allocator, &self.working_root, replacements[0..flush_count]);
285 var root_live = true;
286 errdefer if (root_live) root.deinit();
287 const database = root.hash;
288 for (relation_flushes) |*flush| flush.database = database;
289 self.adoptWorkingRoot(&root);
290 root_live = false;
291 self.discardStagedRelations();
292 return .{
293 .allocator = flush_allocator,
294 .relations = relation_flushes[0..flush_count],
295 .database = database,
296 };
297 }
298
299 pub fn pendingRelations(self: *const DatabaseSession) usize {
300 return self.staged_relations.items.len;
301 }
302
303 pub fn workingRoot(self: *const DatabaseSession) *const version.DatabaseRoot {
304 return &self.working_root;
305 }
306
307 pub fn discardStagedRelations(self: *DatabaseSession) void {
308 for (self.staged_relations.items) |*relation| relation.deinit();
309 self.staged_relations.deinit(self.allocator);
310 self.staged_relations = .empty;
311 }
312
313 pub fn stage(self: *DatabaseSession) void {
314 self.checkout = self.checkout.stage();
315 }
316
317 pub fn commit(self: *DatabaseSession, history: *history_mod.History) DatabaseError!version.Hash {
318 if (!self.checkout.working.hasStaged()) return error.NoStagedRoot;
319 const root = self.checkout.working.staged;
320 const commit_hash = try history.commitBranch(self.checkout.name, root);
321 try self.advance(commit_hash, root);
322 return commit_hash;
323 }
324
325 pub fn advance(self: *DatabaseSession, commit_hash: version.Hash, root: version.Hash) DatabaseError!void {
326 if (!version.same(self.working_root.hash, root)) return error.InvalidHistory;
327 self.checkout = self.checkout.advance(commit_hash, root);
328 self.clearPendingEdits();
329 }
330
331 /// Tells a publication whether it can write one relation as a list of
332 /// changed rows on top of an earlier root by reporting which rows of that
333 /// relation have changed since the last commit. The function answers only
334 /// when the caller's base hash is the root the pending edits were recorded
335 /// against, so the edits are known to cover exactly the distance from that
336 /// base, and it returns nothing when the base differs or when the named
337 /// relation has no recorded edits, and the caller then writes the whole
338 /// relation. The returned rowids borrow the session's own list, which the
339 /// next commit clears.
340 pub fn pendingEditRowids(
341 self: *const DatabaseSession,
342 base: version.Hash,
343 name: []const u8,
344 ) ?[]const i64 {
345 if (!version.same(base, self.pending_edits_base)) return null;
346 const list = self.pending_edits.get(name) orelse return null;
347 return list.items;
348 }
349
350 fn recordPendingEdits(self: *DatabaseSession, name: []const u8, edits: []const relation_mod.Edit) DatabaseError!void {
351 const list = try self.pendingEditList(name);
352 for (edits) |edit| {
353 const rowid = switch (edit) {
354 .put => |put_edit| put_edit.rowid,
355 .update => |update_edit| update_edit.rowid,
356 .delete => |rowid| rowid,
357 };
358 try list.append(self.allocator, rowid);
359 }
360 }
361
362 fn pendingEditList(self: *DatabaseSession, name: []const u8) DatabaseError!*std.ArrayList(i64) {
363 if (self.pending_edits.getPtr(name)) |list| return list;
364 const owned_name = try self.allocator.dupe(u8, name);
365 errdefer self.allocator.free(owned_name);
366 const slot = try self.pending_edits.getOrPut(self.allocator, owned_name);
367 std.debug.assert(!slot.found_existing);
368 slot.key_ptr.* = owned_name;
369 slot.value_ptr.* = .empty;
370 return slot.value_ptr;
371 }
372
373 fn clearPendingEdits(self: *DatabaseSession) void {
374 self.pending_edits_base = self.working_root.hash;
375 for (self.pending_edits.keys(), self.pending_edits.values()) |name, *list| {
376 self.allocator.free(name);
377 list.deinit(self.allocator);
378 }
379 self.pending_edits.clearRetainingCapacity();
380 }
381
382 fn refreshCatalogRoot(self: *DatabaseSession, catalog: *const catalog_mod.Catalog) relation_session.Error!version.DatabaseRoot {
383 return try version.databaseRootMaintained(
384 self.allocator,
385 catalog,
386 self.working_root.conflicts,
387 );
388 }
389
390 fn checkStagedBase(self: *const DatabaseSession, relation: *const StagedRelation) DatabaseError!void {
391 const base_hash = self.workingEntryHash(relation.name) orelse return error.StagedRelationRootMismatch;
392 if (!version.same(base_hash, relation.root.hash)) return error.StagedRelationRootMismatch;
393 }
394
395 fn workingEntryHash(self: *const DatabaseSession, name: []const u8) ?version.Hash {
396 for (self.working_root.entries) |entry| {
397 if (std.mem.eql(u8, entry.name, name)) return entry.hash;
398 }
399 return null;
400 }
401
402 fn stagedRelation(self: *DatabaseSession, name: []const u8) ?*StagedRelation {
403 const index = self.stagedRelationIndex(name) orelse return null;
404 return &self.staged_relations.items[index];
405 }
406
407 fn stagedRelationIndex(self: *const DatabaseSession, name: []const u8) ?usize {
408 for (self.staged_relations.items, 0..) |*relation, index| {
409 if (std.mem.eql(u8, relation.name, name)) return index;
410 }
411 return null;
412 }
413
414 fn adoptWorkingRoot(self: *DatabaseSession, root: *version.DatabaseRoot) void {
415 const hash = root.hash;
416 self.working_root.deinit();
417 self.working_root = root.*;
418 root.* = undefined;
419 self.checkout = self.checkout.withWorking(hash);
420 }
421 };
422
423 const StagedRelation = struct {
424 allocator: Allocator,
425 catalog: catalog_mod.Catalog,
426 name: []u8,
427 schema: catalog_mod.Schema,
428 stats: ?catalog_mod.RelationStats,
429 root: version.RelationRoot,
430
431 fn takeMetadata(relation: *relation_session.RelationSession) StagedRelation {
432 const staged = StagedRelation{
433 .allocator = relation.allocator,
434 .catalog = relation.catalog,
435 .name = relation.name,
436 .schema = relation.schema,
437 .stats = relation.stats,
438 .root = relation.root,
439 };
440 relation.handle.deinit();
441 return staged;
442 }
443
444 fn deinit(self: *StagedRelation) void {
445 self.root.deinit();
446 if (self.stats) |*stats| stats.deinit();
447 self.allocator.free(self.name);
448 self.* = undefined;
449 }
450
451 fn flushRoot(
452 self: *StagedRelation,
453 flush_allocator: Allocator,
454 options: file.CommitOptions,
455 edits: []const relation_mod.Edit,
456 ) DatabaseError!RelationRootFlush {
457 const phase = trace.scope("session.relation.flush_root");
458 defer phase.end();
459
460 var handle = try self.catalog.openRelation(self.allocator, self.name);
461 defer handle.deinit();
462 if (!try self.matchesCurrentRoot(&handle, options.validate_indexes)) return error.StagedRelationRootMismatch;
463
464 const commit = try handle.relation.applyEdits(flush_allocator, edits, options);
465 if (options.validate_indexes) try handle.relation.validateIndexes(self.allocator);
466 const changed = edits.len != 0;
467 if (changed and self.stats != null) _ = try self.catalog.clearRelationStats(self.allocator, self.name, options);
468 const root_stats: ?*const catalog_mod.RelationStats = if (changed) null else if (self.stats) |*relation_stats| relation_stats else null;
469 const relation_key = try version.relationKey(self.name, &handle, root_stats);
470 return .{
471 .commit = commit,
472 .hash = relation_key.hash,
473 };
474 }
475
476 fn matchesCurrentRoot(self: *const StagedRelation, handle: *const catalog_mod.RelationHandle, validate_indexes: bool) DatabaseError!bool {
477 if (validate_indexes) try handle.relation.validateIndexes(self.allocator);
478 const relation_key = try version.relationKey(self.name, handle, if (self.stats) |*relation_stats| relation_stats else null);
479 return version.same(relation_key.hash, self.root.hash);
480 }
481 };
482
483 pub const DatabaseWrite = struct {
484 pub const Limits = staging.Limits;
485 pub const Capacity = staging.Capacity;
486 pub const Demand = staging.Storage.Demand;
487 pub const StagedEdits = staging.Storage.Iterator;
488 pub const Workspace = staging.Workspace;
489
490 session: *DatabaseSession,
491 generation: u64,
492 active: bool = true,
493
494 pub fn deinit(self: *DatabaseWrite) void {
495 self.discard();
496 self.* = undefined;
497 }
498
499 pub fn stageRelation(self: *DatabaseWrite, relation: *relation_session.RelationSession) DatabaseError!void {
500 std.debug.assert(self.active);
501 try self.session.stageRelation(&self.state().storage, relation);
502 }
503
504 pub fn ensure(self: *const DatabaseWrite, demand: Demand) DatabaseError!void {
505 std.debug.assert(self.active);
506 try self.stateConst().storage.ensure(demand);
507 }
508
509 pub fn stagedRelationRoot(self: *DatabaseWrite, name: []const u8) ?*const version.RelationRoot {
510 _ = self.state();
511 const staged = self.session.stagedRelation(name) orelse return null;
512 return &staged.root;
513 }
514
515 pub fn stagedRelationEdits(self: *DatabaseWrite, name: []const u8) ?StagedEdits {
516 std.debug.assert(self.active);
517 const relation_index = self.session.stagedRelationIndex(name) orelse return null;
518 return self.state().storage.iterator(relation_index);
519 }
520
521 pub fn stagePut(self: *DatabaseWrite, name: []const u8, rowid: i64, values: []const row.Value) DatabaseError!void {
522 const phase = trace.scope("session.database.stage_put");
523 defer phase.end();
524
525 std.debug.assert(self.active);
526 const relation_index = self.session.stagedRelationIndex(name) orelse return error.NoStagedRelation;
527 try self.state().storage.appendPut(relation_index, rowid, values);
528 }
529
530 pub fn stageUpdate(self: *DatabaseWrite, name: []const u8, rowid: i64, assignments: []const relation_mod.Edit.Assignment) DatabaseError!void {
531 const phase = trace.scope("session.database.stage_update");
532 defer phase.end();
533
534 std.debug.assert(self.active);
535 const relation_index = self.session.stagedRelationIndex(name) orelse return error.NoStagedRelation;
536 try self.state().storage.appendUpdate(relation_index, rowid, assignments);
537 }
538
539 pub fn stageDelete(self: *DatabaseWrite, name: []const u8, rowid: i64) DatabaseError!void {
540 const phase = trace.scope("session.database.stage_delete");
541 defer phase.end();
542
543 std.debug.assert(self.active);
544 const relation_index = self.session.stagedRelationIndex(name) orelse return error.NoStagedRelation;
545 try self.state().storage.appendDelete(relation_index, rowid);
546 }
547
548 pub fn flush(self: *DatabaseWrite) DatabaseError!DatabaseFlush {
549 std.debug.assert(self.active);
550 const active_write = self.state();
551 const flush_value = try self.session.flushStagedRelations(
552 active_write.flush_allocator,
553 active_write.options,
554 &active_write.storage,
555 );
556 active_write.workspace.release(&active_write.storage);
557 self.session.active_write = null;
558 self.active = false;
559 return flush_value;
560 }
561
562 pub fn flushRelation(self: *DatabaseWrite, relation: *relation_session.RelationSession) DatabaseError!relation_session.RelationFlush {
563 std.debug.assert(self.active);
564 try self.stageRelation(relation);
565 var flush_value = try self.flush();
566 defer flush_value.deinit();
567 return flush_value.onlyRelation();
568 }
569
570 pub fn pendingRelations(self: *const DatabaseWrite) usize {
571 _ = self.stateConst();
572 return self.session.pendingRelations();
573 }
574
575 pub fn discard(self: *DatabaseWrite) void {
576 if (!self.active) return;
577 if (self.session.active_write) |*active_write| {
578 if (active_write.generation == self.generation) {
579 self.session.discardStagedRelations();
580 self.session.discardWrite();
581 }
582 }
583 self.active = false;
584 }
585
586 fn state(self: *DatabaseWrite) *ActiveWrite {
587 std.debug.assert(self.active);
588 const active_write = &self.session.active_write.?;
589 std.debug.assert(active_write.generation == self.generation);
590 return active_write;
591 }
592
593 fn stateConst(self: *const DatabaseWrite) *const ActiveWrite {
594 std.debug.assert(self.active);
595 const active_write = &self.session.active_write.?;
596 std.debug.assert(active_write.generation == self.generation);
597 return active_write;
598 }
599 };
600
601 test "pending edit name allocation failure leaves ownership empty" {
602 const testing = std.testing;
603 var failing = testing.FailingAllocator.init(testing.allocator, .{});
604 var database_session = DatabaseSession{
605 .allocator = failing.allocator(),
606 .checkout = undefined,
607 .working_root = undefined,
608 .pending_edits_base = undefined,
609 };
610 defer database_session.pending_edits.deinit(database_session.allocator);
611 try database_session.pending_edits.ensureUnusedCapacity(
612 database_session.allocator,
613 1,
614 );
615 failing.fail_index = failing.alloc_index;
616 try testing.expectError(
617 error.OutOfMemory,
618 database_session.recordPendingEdits("session_records", &.{}),
619 );
620 try testing.expect(failing.has_induced_failure);
621 try testing.expectEqual(@as(usize, 0), database_session.pending_edits.count());
622 }