lib/sql/src/statement/cursor.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const sql = @import("../root.zig");
3 const access_mod = @import("access.zig");
4 const ast_mod = @import("ast.zig");
5 const execute_mod = @import("execute.zig");
6 const predicate_mod = @import("predicate.zig");
7 const result_mod = @import("result.zig");
8 const page = sql.page;
9 const catalog_mod = sql.catalog;
10 const trace = sql.trace;
11 const plan = sql.plan;
12 const index_mod = sql.index;
13 const relation_mod = sql.relation;
14 const row = sql.row;
15
16 const Allocator = std.mem.Allocator;
17
18 pub const SingleCursor = struct {
19 bytes: ?[]u8,
20 emitted: bool = false,
21 };
22
23 pub const IndexCursor = struct {
24 scan: index_mod.Scan,
25 covered: bool,
26 };
27
28 pub const SortedCursor = struct {
29 rows: [][]u8,
30 index: usize = 0,
31 };
32
33 pub const CursorState = union(enum) {
34 empty,
35 single: SingleCursor,
36 rowid: relation_mod.Scan,
37 index: IndexCursor,
38 scan: relation_mod.Scan,
39 sorted: SortedCursor,
40 };
41
42 pub const Cursor = struct {
43 allocator: Allocator,
44 relation: *plan.PreparedRelation,
45 reader: ?plan.RelationRead,
46 fields: []const execute_mod.ProjectedColumn,
47 access: access_mod.SelectAccess,
48 predicates: []const ast_mod.Predicate,
49 predicate_values: []row.Value,
50 state: CursorState,
51 window: execute_mod.Window = .{},
52 row_buffer: std.ArrayList(u8) = .empty,
53 table_row: ?[]u8 = null,
54
55 pub fn deinit(self: *Cursor) void {
56 self.releaseTableRow();
57 switch (self.state) {
58 .empty => {},
59 .single => |single| if (single.bytes) |bytes| self.allocator.free(bytes),
60 .rowid => |*scan| scan.deinit(),
61 .index => |*index_cursor| index_cursor.scan.deinit(),
62 .scan => |*scan| scan.deinit(),
63 .sorted => |sorted| {
64 result_mod.freeSelectedRows(self.allocator, sorted.rows);
65 self.allocator.free(sorted.rows);
66 },
67 }
68 if (self.reader) |*reader| reader.deinit();
69 self.row_buffer.deinit(self.allocator);
70 for (self.predicate_values) |value| result_mod.freeValue(self.allocator, value);
71 self.allocator.free(self.predicate_values);
72 self.allocator.free(self.predicates);
73 self.* = undefined;
74 }
75
76 pub fn next(self: *Cursor) ast_mod.Error!?[]const u8 {
77 const phase = trace.scope("statement.cursor.next");
78 defer phase.end();
79
80 self.releaseTableRow();
81 if (self.window.full()) return null;
82 return switch (self.state) {
83 .empty => null,
84 .single => |*single| self.nextSingle(single),
85 .rowid => |*scan| self.nextRowid(scan),
86 .index => |*index_cursor| self.nextIndex(index_cursor),
87 .scan => |*scan| self.nextScan(scan),
88 .sorted => |*sorted| self.nextSorted(sorted),
89 };
90 }
91
92 pub fn nextSingle(self: *Cursor, single: *SingleCursor) ast_mod.Error!?[]const u8 {
93 if (single.emitted) return null;
94 single.emitted = true;
95 if (!self.window.admit()) return null;
96 return single.bytes;
97 }
98
99 pub fn nextSorted(self: *Cursor, sorted: *SortedCursor) ast_mod.Error!?[]const u8 {
100 _ = self;
101 if (sorted.index >= sorted.rows.len) return null;
102 const bytes = sorted.rows[sorted.index];
103 sorted.index += 1;
104 return bytes;
105 }
106
107 pub fn nextRowid(self: *Cursor, scan: *relation_mod.Scan) ast_mod.Error!?[]const u8 {
108 const access = switch (self.access) {
109 .rowid => |rowid_access| rowid_access,
110 else => unreachable,
111 };
112 const rowid = try predicate_mod.rowidFromValue(self.predicate_values[access.predicate_index]);
113 while (try scan.next()) |entry| {
114 if (!predicate_mod.rowidMatches(entry.rowid, access.operator, rowid)) continue;
115 if (!try predicate_mod.rowBytesMatchPredicates(self.predicates, self.predicate_values, &self.relation.handle, entry.rowid, entry.bytes)) continue;
116 if (!self.window.admit()) continue;
117 return try self.tableEntry(entry.rowid, entry.bytes);
118 }
119 return null;
120 }
121
122 pub fn nextIndex(self: *Cursor, index_cursor: *IndexCursor) ast_mod.Error!?[]const u8 {
123 const index_slot = switch (self.access) {
124 .index => |index_access| index_access.index_slot,
125 else => unreachable,
126 };
127 while (try index_cursor.scan.next()) |entry| {
128 if (index_cursor.covered) {
129 var index_values: [catalog_mod.max_columns]row.Value = undefined;
130 var index_scratch: [page.size]u8 = undefined;
131 const values = try predicate_mod.coveredIndexValues(
132 self.predicates,
133 self.predicate_values,
134 &self.relation.handle,
135 index_slot,
136 entry,
137 &index_values,
138 &index_scratch,
139 ) orelse continue;
140 if (!self.window.admit()) continue;
141 return try result_mod.selectedIndexRowInto(
142 self.allocator,
143 &self.row_buffer,
144 self.fields,
145 &self.relation.handle,
146 index_slot,
147 values,
148 entry.rowid,
149 );
150 }
151 const reader = if (self.reader) |*value| value else unreachable;
152 const bytes = (try reader.get(self.allocator, entry.rowid)) orelse continue;
153 self.table_row = bytes;
154 if (!try predicate_mod.rowBytesMatchPredicates(self.predicates, self.predicate_values, &self.relation.handle, entry.rowid, bytes)) {
155 self.releaseTableRow();
156 continue;
157 }
158 if (!self.window.admit()) {
159 self.releaseTableRow();
160 continue;
161 }
162 if (self.fields.len == 0) return bytes;
163 const projected = try result_mod.selectedRowInto(self.allocator, &self.row_buffer, self.fields, entry.rowid, bytes);
164 self.releaseTableRow();
165 return projected;
166 }
167 return null;
168 }
169
170 pub fn nextScan(self: *Cursor, scan: *relation_mod.Scan) ast_mod.Error!?[]const u8 {
171 while (try scan.next()) |entry| {
172 if (!try predicate_mod.rowBytesMatchPredicates(self.predicates, self.predicate_values, &self.relation.handle, entry.rowid, entry.bytes)) continue;
173 if (!self.window.admit()) continue;
174 return try self.tableEntry(entry.rowid, entry.bytes);
175 }
176 return null;
177 }
178
179 pub fn tableEntry(self: *Cursor, rowid: i64, bytes: []const u8) ast_mod.Error![]const u8 {
180 if (self.fields.len == 0) return bytes;
181 return try result_mod.selectedRowInto(self.allocator, &self.row_buffer, self.fields, rowid, bytes);
182 }
183
184 pub fn releaseTableRow(self: *Cursor) void {
185 if (self.table_row) |bytes| {
186 self.allocator.free(bytes);
187 self.table_row = null;
188 }
189 }
190 };