tiny.sql.DatabaseSession

1pub const DatabaseSession = struct {2    allocator: Allocator,3    checkout: branch.Checkout,4    staged_relations: std.ArrayList(StagedRelation) = .empty,5    working_root: version.DatabaseRoot,6    pending_edits_base: version.Hash,7    pending_edits: std.StringArrayHashMapUnmanaged(std.ArrayList(i64)) = .empty,8    active_write: ?ActiveWrite = null,9    write_generation: u64 = 0,1011    pub fn initWithRoot(session_allocator: Allocator, checkout: branch.Checkout, root: *version.DatabaseRoot) DatabaseSession {12        std.debug.assert(version.same(root.hash, checkout.working.working));13        const working_root = root.*;14        root.* = undefined;15        return .{16            .allocator = session_allocator,17            .checkout = checkout,18            .working_root = working_root,19            .pending_edits_base = working_root.hash,20        };21    }2223    pub fn deinit(self: *DatabaseSession) void {24        self.discardWrite();25        self.discardStagedRelations();26        self.clearPendingEdits();27        self.pending_edits.deinit(self.allocator);28        self.working_root.deinit();29        self.* = undefined;30    }3132    pub fn reinitWithRoot(33        self: *DatabaseSession,34        checkout: branch.Checkout,35        root: *version.DatabaseRoot,36    ) void {37        const session_allocator = self.allocator;38        const write_generation = self.write_generation;39        self.deinit();40        self.* = initWithRoot(session_allocator, checkout, root);41        self.write_generation = write_generation;42    }4344    pub fn applyRoot(self: *DatabaseSession, root: *version.DatabaseRoot) void {45        self.adoptWorkingRoot(root);46        self.clearPendingEdits();47    }4849    pub fn createRelation(self: *DatabaseSession, allocator: Allocator, catalog: *const catalog_mod.Catalog, definition: catalog_mod.RelationDefinition, options: file.CommitOptions) DatabaseError!CatalogFlush {50        const phase = trace.scope("session.database.create_relation");51        defer phase.end();5253        const commit_value = try catalog.createRelation(allocator, definition, options);54        var root = try self.refreshCatalogRoot(catalog);55        var root_live = true;56        errdefer if (root_live) root.deinit();57        const flush = CatalogFlush{58            .commit = commit_value,59            .database = root.hash,60        };61        self.applyRoot(&root);62        root_live = false;63        return flush;64    }6566    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 {67        const phase = trace.scope("session.database.create_index");68        defer phase.end();6970        const commit_value = try catalog.createIndex(allocator, table_name, definition, options);71        var root = try self.refreshCatalogRoot(catalog);72        var root_live = true;73        errdefer if (root_live) root.deinit();74        const flush = CatalogFlush{75            .commit = commit_value,76            .database = root.hash,77        };78        self.applyRoot(&root);79        root_live = false;80        return flush;81    }8283    pub fn analyzeRelation(self: *DatabaseSession, allocator: Allocator, catalog: *const catalog_mod.Catalog, name: []const u8, options: file.CommitOptions) DatabaseError!CatalogFlush {84        const phase = trace.scope("session.database.analyze_relation");85        defer phase.end();8687        const commit_value = try catalog.analyzeRelation(allocator, name, options);88        var root = try self.refreshCatalogRoot(catalog);89        var root_live = true;90        errdefer if (root_live) root.deinit();91        const flush = CatalogFlush{92            .commit = commit_value,93            .database = root.hash,94        };95        self.applyRoot(&root);96        root_live = false;97        return flush;98    }99100    pub fn dropRelation(self: *DatabaseSession, allocator: Allocator, catalog: *const catalog_mod.Catalog, name: []const u8, options: file.CommitOptions) DatabaseError!CatalogFlush {101        const phase = trace.scope("session.database.drop_relation");102        defer phase.end();103104        const commit_value = try catalog.dropRelation(allocator, name, options);105        var root = try self.refreshCatalogRoot(catalog);106        var root_live = true;107        errdefer if (root_live) root.deinit();108        const flush = CatalogFlush{109            .commit = commit_value,110            .database = root.hash,111        };112        self.applyRoot(&root);113        root_live = false;114        return flush;115    }116117    pub fn beginWrite(118        self: *DatabaseSession,119        workspace: *DatabaseWrite.Workspace,120        flush_allocator: Allocator,121        limits: DatabaseWrite.Limits,122        options: file.CommitOptions,123    ) DatabaseError!DatabaseWrite {124        const phase = trace.scope("session.database.begin_write");125        defer phase.end();126127        if (self.active_write != null) return error.WriteSessionActive;128        std.debug.assert(self.staged_relations.items.len == 0);129        const generation = std.math.add(u64, self.write_generation, 1) catch130            return error.CapacityOverflow;131        const storage = try workspace.acquire(limits);132        self.write_generation = generation;133        self.active_write = .{134            .generation = generation,135            .workspace = workspace,136            .flush_allocator = flush_allocator,137            .options = options,138            .storage = storage,139        };140        return .{141            .session = self,142            .generation = generation,143        };144    }145146    fn discardWrite(self: *DatabaseSession) void {147        if (self.active_write) |*active_write| {148            active_write.workspace.release(&active_write.storage);149            self.active_write = null;150        }151    }152153    fn stageRelation(154        self: *DatabaseSession,155        storage: *staging.Storage,156        relation: *relation_session.RelationSession,157    ) DatabaseError!void {158        const phase = trace.scope("session.database.stage_relation");159        defer phase.end();160161        const demand = try staging.Storage.demandForEdits(relation.edits.items);162        try storage.ensure(demand);163        if (self.stagedRelationIndex(relation.name)) |relation_index| {164            const staged = &self.staged_relations.items[relation_index];165            if (!version.same(staged.root.hash, relation.root.hash)) {166                return error.StagedRelationRootMismatch;167            }168            storage.appendEditsAssumeCapacity(relation_index, relation.edits.items);169            relation.deinit();170            return;171        }172        try storage.ensureRelation();173        try self.staged_relations.ensureUnusedCapacity(self.allocator, 1);174        const staged = StagedRelation.takeMetadata(relation);175        self.staged_relations.appendAssumeCapacity(staged);176        const relation_index = storage.registerRelation();177        std.debug.assert(relation_index == self.staged_relations.items.len - 1);178        storage.appendEditsAssumeCapacity(relation_index, relation.edits.items);179        relation_session.finishStaging(relation);180    }181182    fn flushStagedRelations(183        self: *DatabaseSession,184        flush_allocator: Allocator,185        options: file.CommitOptions,186        storage: *staging.Storage,187    ) DatabaseError!DatabaseFlush {188        const phase = trace.scope("session.database.flush_staged_relations");189        defer phase.end();190191        if (self.staged_relations.items.len == 0) return error.NoStagedRelation;192        errdefer self.discardStagedRelations();193194        const relation_flushes = try flush_allocator.alloc(relation_session.RelationFlush, self.staged_relations.items.len);195        var flush_count: usize = 0;196        errdefer flush_allocator.free(relation_flushes);197198        const replacements = try flush_allocator.alloc(version.RelationEntry, self.staged_relations.items.len);199        defer flush_allocator.free(replacements);200201        storage.prepareOrdered();202        for (203            self.staged_relations.items,204            relation_flushes,205            replacements,206            0..,207        ) |*relation, *flush, *replacement, relation_index| {208            const edits = storage.orderedFor(relation_index);209            try self.checkStagedBase(relation);210            try self.recordPendingEdits(relation.name, edits);211            const root_flush = try relation.flushRoot(flush_allocator, options, edits);212            replacement.* = .{213                .name = relation.name,214                .hash = root_flush.hash,215            };216            flush.* = .{217                .commit = root_flush.commit,218                .relation = root_flush.hash,219                .database = undefined,220            };221            flush_count += 1;222        }223224        var root = try version.databaseRootReplacingEntries(self.allocator, &self.working_root, replacements[0..flush_count]);225        var root_live = true;226        errdefer if (root_live) root.deinit();227        const database = root.hash;228        for (relation_flushes) |*flush| flush.database = database;229        self.adoptWorkingRoot(&root);230        root_live = false;231        self.discardStagedRelations();232        return .{233            .allocator = flush_allocator,234            .relations = relation_flushes[0..flush_count],235            .database = database,236        };237    }238239    pub fn pendingRelations(self: *const DatabaseSession) usize {240        return self.staged_relations.items.len;241    }242243    pub fn workingRoot(self: *const DatabaseSession) *const version.DatabaseRoot {244        return &self.working_root;245    }246247    pub fn discardStagedRelations(self: *DatabaseSession) void {248        for (self.staged_relations.items) |*relation| relation.deinit();249        self.staged_relations.deinit(self.allocator);250        self.staged_relations = .empty;251    }252253    pub fn stage(self: *DatabaseSession) void {254        self.checkout = self.checkout.stage();255    }256257    pub fn commit(self: *DatabaseSession, history: *history_mod.History) DatabaseError!version.Hash {258        if (!self.checkout.working.hasStaged()) return error.NoStagedRoot;259        const root = self.checkout.working.staged;260        const commit_hash = try history.commitBranch(self.checkout.name, root);261        try self.advance(commit_hash, root);262        return commit_hash;263    }264265    pub fn advance(self: *DatabaseSession, commit_hash: version.Hash, root: version.Hash) DatabaseError!void {266        if (!version.same(self.working_root.hash, root)) return error.InvalidHistory;267        self.checkout = self.checkout.advance(commit_hash, root);268        self.clearPendingEdits();269    }270271    /// Tells a publication whether it can write one relation as a list of272    /// changed rows on top of an earlier root by reporting which rows of that273    /// relation have changed since the last commit. The function answers only274    /// when the caller's base hash is the root the pending edits were recorded275    /// against, so the edits are known to cover exactly the distance from that276    /// base, and it returns nothing when the base differs or when the named277    /// relation has no recorded edits, and the caller then writes the whole278    /// relation. The returned rowids borrow the session's own list, which the279    /// next commit clears.280    pub fn pendingEditRowids(281        self: *const DatabaseSession,282        base: version.Hash,283        name: []const u8,284    ) ?[]const i64 {285        if (!version.same(base, self.pending_edits_base)) return null;286        const list = self.pending_edits.get(name) orelse return null;287        return list.items;288    }289290    fn recordPendingEdits(self: *DatabaseSession, name: []const u8, edits: []const relation_mod.Edit) DatabaseError!void {291        const list = try self.pendingEditList(name);292        for (edits) |edit| {293            const rowid = switch (edit) {294                .put => |put_edit| put_edit.rowid,295                .update => |update_edit| update_edit.rowid,296                .delete => |rowid| rowid,297            };298            try list.append(self.allocator, rowid);299        }300    }301302    fn pendingEditList(self: *DatabaseSession, name: []const u8) DatabaseError!*std.ArrayList(i64) {303        if (self.pending_edits.getPtr(name)) |list| return list;304        const owned_name = try self.allocator.dupe(u8, name);305        errdefer self.allocator.free(owned_name);306        const slot = try self.pending_edits.getOrPut(self.allocator, owned_name);307        std.debug.assert(!slot.found_existing);308        slot.key_ptr.* = owned_name;309        slot.value_ptr.* = .empty;310        return slot.value_ptr;311    }312313    fn clearPendingEdits(self: *DatabaseSession) void {314        self.pending_edits_base = self.working_root.hash;315        for (self.pending_edits.keys(), self.pending_edits.values()) |name, *list| {316            self.allocator.free(name);317            list.deinit(self.allocator);318        }319        self.pending_edits.clearRetainingCapacity();320    }321322    fn refreshCatalogRoot(self: *DatabaseSession, catalog: *const catalog_mod.Catalog) relation_session.Error!version.DatabaseRoot {323        return try version.databaseRootMaintained(324            self.allocator,325            catalog,326            self.working_root.conflicts,327        );328    }329330    fn checkStagedBase(self: *const DatabaseSession, relation: *const StagedRelation) DatabaseError!void {331        const base_hash = self.workingEntryHash(relation.name) orelse return error.StagedRelationRootMismatch;332        if (!version.same(base_hash, relation.root.hash)) return error.StagedRelationRootMismatch;333    }334335    fn workingEntryHash(self: *const DatabaseSession, name: []const u8) ?version.Hash {336        for (self.working_root.entries) |entry| {337            if (std.mem.eql(u8, entry.name, name)) return entry.hash;338        }339        return null;340    }341342    fn stagedRelation(self: *DatabaseSession, name: []const u8) ?*StagedRelation {343        const index = self.stagedRelationIndex(name) orelse return null;344        return &self.staged_relations.items[index];345    }346347    fn stagedRelationIndex(self: *const DatabaseSession, name: []const u8) ?usize {348        for (self.staged_relations.items, 0..) |*relation, index| {349            if (std.mem.eql(u8, relation.name, name)) return index;350        }351        return null;352    }353354    fn adoptWorkingRoot(self: *DatabaseSession, root: *version.DatabaseRoot) void {355        const hash = root.hash;356        self.working_root.deinit();357        self.working_root = root.*;358        root.* = undefined;359        self.checkout = self.checkout.withWorking(hash);360    }361};

1pub const DatabaseSession = session.DatabaseSession;