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 };