lib/sql/src/session/relation.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const sql = @import("../root.zig");
3 const staging = @import("staging/root.zig");
4
5 const catalog_mod = sql.catalog;
6 const file = sql.file;
7 const relation_mod = sql.relation;
8 const row = sql.row;
9 const trace = sql.trace;
10 const version = sql.version;
11
12 const Allocator = std.mem.Allocator;
13
14 pub const Error = catalog_mod.Error || version.Error || relation_mod.Error ||
15 staging.Storage.Error || Allocator.Error;
16
17 pub const RelationFlush = struct {
18 commit: file.Commit,
19 relation: version.Hash,
20 database: version.Hash,
21 };
22
23 fn appendPutEdit(allocator: Allocator, edits: *std.ArrayList(relation_mod.Edit), rowid: i64, values: []const row.Value) Error!void {
24 const size = try row.encodedSize(values);
25 const bytes = try allocator.alloc(u8, size);
26 errdefer allocator.free(bytes);
27 _ = try row.encode(bytes, values);
28 try edits.append(allocator, .{ .put = .{
29 .rowid = rowid,
30 .bytes = bytes,
31 } });
32 }
33
34 fn appendUpdateEdit(allocator: Allocator, edits: *std.ArrayList(relation_mod.Edit), rowid: i64, assignments: []const relation_mod.Edit.Assignment) Error!void {
35 const owned = try relation_mod.cloneAssignments(allocator, assignments);
36 errdefer relation_mod.freeAssignments(allocator, owned);
37 try edits.append(allocator, .{ .update = .{
38 .rowid = rowid,
39 .assignments = owned,
40 } });
41 }
42
43 fn appendDeleteEdit(allocator: Allocator, edits: *std.ArrayList(relation_mod.Edit), rowid: i64) Error!void {
44 try edits.append(allocator, .{ .delete = rowid });
45 }
46
47 pub const RelationSession = struct {
48 allocator: Allocator,
49 catalog: catalog_mod.Catalog,
50 name: []u8,
51 schema: catalog_mod.Schema,
52 handle: catalog_mod.RelationHandle,
53 stats: ?catalog_mod.RelationStats,
54 root: version.RelationRoot,
55 edits: std.ArrayList(relation_mod.Edit),
56
57 pub fn open(allocator: Allocator, catalog: *const catalog_mod.Catalog, name: []const u8) Error!RelationSession {
58 const owned_name = try allocator.dupe(u8, name);
59 errdefer allocator.free(owned_name);
60 var state = try catalog.readRelation(allocator, name);
61 errdefer state.deinit();
62 var root = try version.relationRootMaintained(
63 allocator,
64 name,
65 state.schema,
66 &state.handle,
67 state.relationStats(),
68 );
69 errdefer root.deinit();
70 return .{
71 .allocator = allocator,
72 .catalog = catalog.*,
73 .name = owned_name,
74 .schema = state.schema,
75 .handle = state.handle,
76 .stats = state.stats,
77 .root = root,
78 .edits = .empty,
79 };
80 }
81
82 pub fn deinit(self: *RelationSession) void {
83 self.clearEdits();
84 self.edits.deinit(self.allocator);
85 self.root.deinit();
86 if (self.stats) |*stats| stats.deinit();
87 self.handle.deinit();
88 self.allocator.free(self.name);
89 self.* = undefined;
90 }
91
92 pub fn put(self: *RelationSession, rowid: i64, values: []const row.Value) Error!void {
93 const phase = trace.scope("session.relation.put");
94 defer phase.end();
95
96 try appendPutEdit(self.allocator, &self.edits, rowid, values);
97 }
98
99 pub fn putEncoded(self: *RelationSession, rowid: i64, bytes: []const u8) Error!void {
100 const phase = trace.scope("session.relation.put_encoded");
101 defer phase.end();
102
103 _ = try row.View.init(bytes);
104 const owned = try self.allocator.dupe(u8, bytes);
105 errdefer self.allocator.free(owned);
106 try self.edits.append(self.allocator, .{ .put = .{
107 .rowid = rowid,
108 .bytes = owned,
109 } });
110 }
111
112 pub fn update(self: *RelationSession, rowid: i64, assignments: []const relation_mod.Edit.Assignment) Error!void {
113 const phase = trace.scope("session.relation.update");
114 defer phase.end();
115
116 try appendUpdateEdit(self.allocator, &self.edits, rowid, assignments);
117 }
118
119 pub fn delete(self: *RelationSession, rowid: i64) Error!void {
120 const phase = trace.scope("session.relation.delete");
121 defer phase.end();
122
123 try appendDeleteEdit(self.allocator, &self.edits, rowid);
124 }
125
126 pub fn pendingEdits(self: *const RelationSession) usize {
127 return self.edits.items.len;
128 }
129
130 pub fn stagingLimits(self: *const RelationSession) Error!staging.Limits {
131 const demand = try staging.Storage.demandForEdits(self.edits.items);
132 return .{
133 .relations = 1,
134 .edits = demand.edits,
135 .payload_bytes = demand.payload_bytes,
136 .assignments = demand.assignments,
137 };
138 }
139
140 pub fn refresh(self: *RelationSession) Error!void {
141 const phase = trace.scope("session.relation.refresh");
142 defer phase.end();
143
144 var state = try self.catalog.readRelation(self.allocator, self.name);
145 errdefer state.deinit();
146 try state.handle.relation.validateIndexes(self.allocator);
147 var root = try version.relationRootMaintained(
148 self.allocator,
149 self.name,
150 state.schema,
151 &state.handle,
152 state.relationStats(),
153 );
154 errdefer root.deinit();
155
156 self.root.deinit();
157 if (self.stats) |*old_stats| old_stats.deinit();
158 self.handle.deinit();
159 self.schema = state.schema;
160 self.handle = state.handle;
161 self.stats = state.stats;
162 self.root = root;
163 }
164
165 fn clearEdits(self: *RelationSession) void {
166 self.discardEditsFrom(0);
167 self.edits.clearRetainingCapacity();
168 }
169
170 fn discardEditsFrom(self: *RelationSession, start: usize) void {
171 for (self.edits.items[start..]) |edit| switch (edit) {
172 .put => |put_edit| self.allocator.free(put_edit.bytes),
173 .update => |update_edit| relation_mod.freeAssignments(self.allocator, update_edit.assignments),
174 .delete => {},
175 };
176 self.edits.items.len = start;
177 }
178 };
179
180 pub fn finishStaging(session: *RelationSession) void {
181 session.clearEdits();
182 session.edits.deinit(session.allocator);
183 session.* = undefined;
184 }