tiny.sql.Connection

1pub const Connection = struct {2    catalog: catalog_mod.Catalog,3    session: session_mod.DatabaseSession,4    recovery_required: bool = false,56    pub fn init(catalog: catalog_mod.Catalog, session: session_mod.DatabaseSession) Connection {7        return .{8            .catalog = catalog,9            .session = session,10        };11    }1213    pub fn create(allocator: Allocator, database: *file.Database, history: *history_mod.History, options: Options) Error!Connection {14        var catalog = try catalog_mod.Catalog.open(database, options.catalog);15        var value = try version.databaseValue(16            allocator,17            &catalog,18            version.ConflictRoot.empty().hash,19        );20        var value_live = true;21        errdefer if (value_live) value.deinit();22        const root_commit = version.Commit.init(value.root.hash, &.{});23        try database.syncWal();24        try publishDatabaseValue(history, &value);25        try history.putCommit(root_commit);26        _ = try history.createBranch(options.branch, root_commit.hash);27        const checkout_value = try history.checkoutBranch(options.branch);28        var root = value.intoRoot();29        value_live = false;30        const session = session_mod.DatabaseSession.initWithRoot(allocator, checkout_value, &root);31        return Connection.init(catalog, session);32    }3334    pub fn openLocal(allocator: Allocator, database: *file.Database, options: Options) Error!Connection {35        const catalog = try catalog_mod.Catalog.open(database, options.catalog);36        var root = try version.databaseRootMaintained(37            allocator,38            &catalog,39            version.ConflictRoot.empty().hash,40        );41        errdefer root.deinit();42        const root_commit = version.Commit.init(root.hash, &.{});43        const checkout_value = branch_mod.checkout(.{44            .name = options.branch,45            .target = root_commit.hash,46        }, root.hash);47        return Connection.init(48            catalog,49            session_mod.DatabaseSession.initWithRoot(allocator, checkout_value, &root),50        );51    }5253    pub fn open(allocator: Allocator, database: *file.Database, history: *history_mod.History, options: Options) Error!Connection {54        try recoverFastForward(allocator, database, history, options.catalog);55        const catalog = try catalog_mod.Catalog.open(database, options.catalog);56        const checkout_value = try history.checkoutBranch(options.branch);57        var target = try history.databaseRoot(allocator, checkout_value.working.working);58        defer target.deinit();59        try history.validateConflictRoot(target.conflicts);60        var root = try version.databaseRootMaintained(allocator, &catalog, target.conflicts);61        errdefer root.deinit();62        const live_checkout = checkout_value.withWorking(root.hash);63        return Connection.init(catalog, session_mod.DatabaseSession.initWithRoot(allocator, live_checkout, &root));64    }6566    pub fn adoptRebuiltHistory(self: *Connection, history: *history_mod.History, branch_name: []const u8) Error!void {67        try self.ensureUsable();68        std.debug.assert(history.len() == 0);69        const allocator = self.session.allocator;70        var value = try version.databaseValue(71            allocator,72            &self.catalog,73            version.ConflictRoot.empty().hash,74        );75        var value_live = true;76        errdefer if (value_live) value.deinit();77        const root_commit = version.Commit.init(value.root.hash, &.{});78        try self.catalog.database.syncWal();79        try publishDatabaseValue(history, &value);80        try history.putCommit(root_commit);81        _ = try history.createBranch(branch_name, root_commit.hash);82        const checkout_value = try history.checkoutBranch(branch_name);83        var root = value.intoRoot();84        value_live = false;85        self.session.reinitWithRoot(checkout_value, &root);86    }8788    pub fn deinit(self: *Connection) void {89        self.session.deinit();90        self.* = undefined;91    }9293    pub fn lastRowId(self: *const Connection, allocator: Allocator, table_name: []const u8) Error!?i64 {94        try self.ensureUsable();95        var handle = try self.catalog.openRelation(allocator, table_name);96        defer handle.deinit();97        return try handle.relation.lastRowId();98    }99100    pub fn prepare(self: *const Connection, allocator: Allocator, source: []const u8) statement_mod.Error!statement_mod.Prepared {101        try self.ensureUsable();102        return try statement_mod.prepare(&self.catalog, allocator, source);103    }104105    pub fn execute(self: *Connection, allocator: Allocator, source: []const u8, options: ExecuteOptions) statement_mod.Error!statement_mod.Result {106        var prepared = try self.prepare(allocator, source);107        defer prepared.deinit();108        return try self.executePrepared(&prepared, allocator, options);109    }110111    pub fn executePrepared(self: *Connection, prepared: *statement_mod.Prepared, allocator: Allocator, options: ExecuteOptions) statement_mod.Error!statement_mod.Result {112        try self.ensureUsable();113        return try prepared.execute(allocator, options.statement(&self.session));114    }115116    /// Opens a cursor over the rows `prepared` selects in `target`.117    pub fn openCursor(118        self: *Connection,119        target: *statement_mod.Cursor,120        prepared: *statement_mod.Prepared,121        allocator: Allocator,122    ) statement_mod.Error!void {123        try self.ensureUsable();124        try prepared.openCursor(target, allocator);125    }126127    pub fn beginWrite(128        self: *Connection,129        workspace: *session_mod.DatabaseWrite.Workspace,130        flush_allocator: Allocator,131        options: WriteOptions,132    ) session_mod.DatabaseError!session_mod.DatabaseWrite {133        try self.ensureUsable();134        return try self.session.beginWrite(135            workspace,136            flush_allocator,137            options.limits,138            options.commit(),139        );140    }141142    pub fn stage(self: *Connection) Error!void {143        try self.ensureUsable();144        self.session.stage();145    }146147    pub fn commit(self: *Connection, history: *history_mod.History) Error!version.Hash {148        try self.ensureUsable();149        if (!self.session.checkout.working.hasStaged()) return error.NoStagedRoot;150        try self.catalog.database.syncWal();151        try self.publishStagedValue(history, self.session.checkout.working.staged);152        return try self.session.commit(history);153    }154155    pub fn commitLocal(self: *Connection) Error!version.Hash {156        try self.ensureUsable();157        if (!self.session.checkout.working.hasStaged()) return error.NoStagedRoot;158        const root = self.session.checkout.working.staged;159        const parents = [_]version.Hash{self.session.checkout.head};160        const commit_value = version.Commit.init(root, &parents);161        try self.session.advance(commit_value.hash, root);162        return commit_value.hash;163    }164165    pub fn mergeCommit(self: *Connection, history: *history_mod.History, theirs: version.Hash) Error!version.Hash {166        try self.ensureUsable();167        const root = self.session.checkout.working.working;168        try self.catalog.database.syncWal();169        try self.publishStagedValue(history, root);170        const commit_hash = try history.mergeCommitBranch(self.session.checkout.name, root, theirs);171        try self.session.advance(commit_hash, root);172        return commit_hash;173    }174175    pub fn createBranch(self: *const Connection, history: *history_mod.History, name: []const u8) history_mod.Error!version.Ref {176        try self.ensureUsable();177        return try history.createBranch(name, self.session.checkout.head);178    }179180    pub fn checkoutBranch(self: *Connection, allocator: Allocator, history: *const history_mod.History, name: []const u8) Error!void {181        try self.ensureUsable();182        const checkout_value = try history.checkoutBranch(name);183        var value = try history.databaseValue(allocator, checkout_value.working.working);184        var value_live = true;185        errdefer if (value_live) value.deinit();186        try history.validateConflictRoot(value.root.conflicts);187        try self.materializeDatabaseValue(allocator, &value, .{ .durability = .synced });188        var root = value.intoRoot();189        value_live = false;190        self.session.reinitWithRoot(checkout_value, &root);191    }192193    pub fn fastForwardBranch(self: *Connection, allocator: Allocator, history: *history_mod.History, target: version.Hash) Error!void {194        try self.ensureUsable();195        var next_ref = (try history.ref(self.session.checkout.name)) orelse return error.RefNotFound;196        const baseline = try self.validateFastForwardBaseline(allocator, history, next_ref);197        const entries = try history.commitEntries(allocator);198        defer allocator.free(entries);199        try branch_mod.fastForwardRef(allocator, entries, &next_ref, target);200        const commit_value = try history.commitValue(target);201        var value = try history.databaseValue(allocator, commit_value.root);202        var value_live = true;203        errdefer if (value_live) value.deinit();204        try history.validateConflictRoot(value.root.conflicts);205        var plan = try DatabaseMaterialization.init(self, allocator, &value);206        defer plan.deinit();207        try self.coordinateFastForward(208            allocator,209            history,210            next_ref,211            baseline,212            &value,213            &value_live,214            &plan,215        );216    }217218    fn validateFastForwardBaseline(219        self: *Connection,220        allocator: Allocator,221        history: *const history_mod.History,222        current_ref: version.Ref,223    ) Error!version.Hash {224        if (!self.catalog.database.walSynced()) return error.WorkingSetChanged;225        if (!version.same(current_ref.target, self.session.checkout.head)) return error.RefChanged;226        const working = self.session.checkout.working;227        if (!version.same(working.base, working.working) or228            !version.same(working.base, working.staged) or229            !version.same(working.base, self.session.workingRoot().hash) or230            self.session.pendingRelations() != 0)231        {232            return error.WorkingSetChanged;233        }234        const current_commit = try history.commitValue(self.session.checkout.head);235        if (!version.same(current_commit.root, working.base)) return error.WorkingSetChanged;236        var live_root = try version.databaseRootMaintained(237            allocator,238            &self.catalog,239            self.session.workingRoot().conflicts,240        );241        defer live_root.deinit();242        if (!version.same(live_root.hash, working.base)) return error.WorkingSetChanged;243        return working.base;244    }245246    fn coordinateFastForward(247        self: *Connection,248        allocator: Allocator,249        history: *history_mod.History,250        next_ref: version.Ref,251        baseline: version.Hash,252        value: *version.DatabaseValue,253        value_live: *bool,254        plan: *DatabaseMaterialization,255    ) Error!void {256        try self.catalog.database.syncWal();257        const savepoint = try self.catalog.database.savepoint();258        try self.catalog.database.beginCoordinator();259        var coordinator_active = true;260        errdefer if (coordinator_active and !self.catalog.database.requiresRecovery()) {261            self.catalog.database.endCoordinator();262        };263        var update = history.beginFastForward(264            next_ref.name,265            self.session.checkout.head,266            next_ref.target,267        ) catch |err| {268            if (err == error.RecoveryRequired) {269                self.poisonFastForward(history);270                return error.RecoveryRequired;271            }272            self.catalog.database.endCoordinator();273            coordinator_active = false;274            return err;275        };276        plan.apply(self, .{ .durability = .synced }) catch |err| {277            try self.restoreFastForward(history, &update, savepoint, baseline);278            coordinator_active = false;279            return err;280        };281        self.verifyMaterializedDatabaseRoot(allocator, &value.root) catch |err| {282            try self.restoreFastForward(history, &update, savepoint, baseline);283            coordinator_active = false;284            return err;285        };286        update.commit() catch |err| {287            if (err == error.RecoveryRequired) {288                self.poisonFastForward(history);289                return error.RecoveryRequired;290            }291            try self.restoreFastForward(history, &update, savepoint, baseline);292            coordinator_active = false;293            return err;294        };295        const checkout_value = branch_mod.checkout(next_ref, value.root.hash);296        var root = value.intoRoot();297        value_live.* = false;298        self.session.reinitWithRoot(checkout_value, &root);299        try self.completeFastForward(history, &update);300        self.catalog.database.endCoordinator();301        coordinator_active = false;302    }303304    fn verifyMaterializedDatabaseRoot(305        self: *Connection,306        allocator: Allocator,307        target: *const version.DatabaseRoot,308    ) Error!void {309        var names = try self.catalog.relationNames(allocator);310        defer names.deinit();311        if (names.names.len != target.entries.len) return error.InvalidHistory;312        for (names.names) |name| {313            const entry = relationEntry(target.entries, name) orelse return error.InvalidHistory;314            var state = try self.catalog.readRelation(allocator, name);315            defer state.deinit();316            const live = try version.relationKey(name, &state.handle, state.relationStats());317            if (!version.same(live.hash, entry.hash)) return error.InvalidHistory;318        }319        var verified = try version.databaseRootMaintained(320            allocator,321            &self.catalog,322            target.conflicts,323        );324        defer verified.deinit();325        if (!version.same(verified.hash, target.hash)) return error.InvalidHistory;326    }327328    fn restoreFastForward(329        self: *Connection,330        history: *history_mod.History,331        update: *history_mod.FastForwardUpdate,332        savepoint: file.Savepoint,333        baseline: version.Hash,334    ) Error!void {335        self.rollbackFastForward(update, savepoint, baseline) catch {336            self.poisonFastForward(history);337            return error.RecoveryRequired;338        };339    }340341    fn completeFastForward(342        self: *Connection,343        history: *history_mod.History,344        update: *history_mod.FastForwardUpdate,345    ) Error!void {346        update.complete() catch |err| {347            if (err != error.RecoveryRequired) update.complete() catch {348                self.poisonFastForward(history);349                return error.RecoveryRequired;350            } else {351                self.poisonFastForward(history);352                return error.RecoveryRequired;353            }354        };355    }356357    fn rollbackFastForward(358        self: *Connection,359        update: *history_mod.FastForwardUpdate,360        savepoint: file.Savepoint,361        baseline: version.Hash,362    ) Error!void {363        try self.catalog.database.restore(savepoint);364        try self.catalog.database.syncWal();365        var restored = try version.databaseRootMaintained(366            self.session.allocator,367            &self.catalog,368            self.session.workingRoot().conflicts,369        );370        defer restored.deinit();371        if (!version.same(restored.hash, baseline)) return error.RecoveryRequired;372        try update.abort();373        try update.complete();374        self.catalog.database.endCoordinator();375    }376377    fn poisonFastForward(self: *Connection, history: *history_mod.History) void {378        self.recovery_required = true;379        self.catalog.database.poison();380        history.poison();381    }382383    fn ensureUsable(self: *const Connection) error{RecoveryRequired}!void {384        if (self.recovery_required or self.catalog.database.requiresRecovery()) {385            return error.RecoveryRequired;386        }387    }388389    pub fn mergeBase(self: *const Connection, allocator: Allocator, history: *const history_mod.History, theirs: []const u8) history_mod.Error!?version.Hash {390        try self.ensureUsable();391        const theirs_ref = (try history.ref(theirs)) orelse return error.RefNotFound;392        const entries = try history.commitEntries(allocator);393        defer allocator.free(entries);394        return try branch_mod.mergeBase(allocator, entries, self.session.checkout.head, theirs_ref.target);395    }396397    pub fn relationView(self: *const Connection, allocator: Allocator, name: []const u8) Error!RelationView {398        try self.ensureUsable();399        var state = try self.catalog.readRelation(allocator, name);400        errdefer state.handle.deinit();401        defer if (state.stats) |*relation_stats| relation_stats.deinit();402        var root = try version.relationRoot(403            allocator,404            name,405            state.schema,406            &state.handle,407            state.relationStats(),408        );409        errdefer root.deinit();410        return .{411            .allocator = allocator,412            .root = root,413            .handle = state.handle,414        };415    }416417    pub fn databaseView(self: *const Connection, allocator: Allocator, history: *const history_mod.History, names: []const []const u8) Error!DatabaseView {418        try self.ensureUsable();419        var conflicts = try history.conflictArtifacts(allocator, self.session.workingRoot().conflicts);420        errdefer conflicts.deinit();421422        const relations = try allocator.alloc(RelationView, names.len);423        errdefer allocator.free(relations);424        var relation_count: usize = 0;425        errdefer for (relations[0..relation_count]) |*relation_view| relation_view.deinit();426427        const snapshots = try allocator.alloc(diff_mod.RelationSnapshot, names.len);428        errdefer allocator.free(snapshots);429430        for (names, relations, snapshots) |name, *relation_view, *snapshot| {431            relation_view.* = try self.relationView(allocator, name);432            relation_count += 1;433            snapshot.* = relation_view.snapshot();434        }435436        return .{437            .allocator = allocator,438            .relations = relations,439            .snapshots = snapshots,440            .conflicts = conflicts,441        };442    }443444    pub fn databaseViewAtCommit(self: *const Connection, allocator: Allocator, history: *const history_mod.History, commit_hash: version.Hash) Error!DatabaseView {445        try self.ensureUsable();446        var root = try history.commitDatabaseRoot(allocator, commit_hash);447        defer root.deinit();448        return try databaseViewFromRoot(allocator, history, root);449    }450451    pub fn diffRelation(self: *const Connection, allocator: Allocator, name: []const u8, other: diff_mod.RelationSnapshot) Error!diff_mod.RelationDiff {452        try self.ensureUsable();453        var local = try self.relationView(allocator, name);454        defer local.deinit();455        return try diff_mod.relation(allocator, local.snapshot(), other);456    }457458    pub fn mergeDatabase(self: *Connection, allocator: Allocator, history: *history_mod.History, base: merge_mod.DatabaseSnapshot, theirs: merge_mod.DatabaseSnapshot, options: MergeOptions) Error!merge_mod.DatabaseMerge {459        try self.ensureUsable();460        var ours = try self.materializedWorkingValue(allocator);461        defer ours.deinit();462        var ours_conflicts = try history.conflictArtifacts(allocator, ours.root.conflicts);463        defer ours_conflicts.deinit();464        var result = try merge_mod.mergeDatabase(allocator, base, .{465            .value = &ours,466            .conflicts = .{467                .root = ours_conflicts.root,468                .artifacts = ours_conflicts.artifacts,469            },470        }, theirs);471        errdefer result.deinit();472        try self.materializeDatabaseMerge(allocator, history, &result, options.commit);473        var root = try result.value.root.clone(self.session.allocator);474        self.session.applyRoot(&root);475        return result;476    }477478    pub fn mergeBranch(self: *Connection, allocator: Allocator, history: *history_mod.History, theirs: []const u8, options: MergeOptions) Error!merge_mod.DatabaseMerge {479        try self.ensureUsable();480        const theirs_ref = (try history.ref(theirs)) orelse return error.RefNotFound;481        const entries = try history.commitEntries(allocator);482        defer allocator.free(entries);483        const base_commit = (try branch_mod.mergeBase(allocator, entries, self.session.checkout.head, theirs_ref.target)) orelse return error.NoMergeBase;484        var base_view = try self.databaseViewAtCommit(allocator, history, base_commit);485        defer base_view.deinit();486        var theirs_view = try self.databaseViewAtCommit(allocator, history, theirs_ref.target);487        defer theirs_view.deinit();488        return try self.mergeDatabase(allocator, history, base_view.snapshot(), theirs_view.snapshot(), options);489    }490491    fn replaceConflictRoot(self: *Connection, conflicts: version.ConflictRoot) Error!version.Hash {492        var root = try version.DatabaseRoot.initSorted(self.session.allocator, self.session.workingRoot().entries, conflicts);493        self.session.applyRoot(&root);494        return (try self.workingRoot());495    }496497    pub fn conflictArtifacts(self: *const Connection, allocator: Allocator, history: *const history_mod.History) Error!history_mod.ConflictArtifacts {498        try self.ensureUsable();499        return try history.conflictArtifacts(allocator, self.session.workingRoot().conflicts);500    }501502    pub fn resolveConflicts(self: *Connection, allocator: Allocator, history: *history_mod.History, resolved: []const version.Hash) Error!version.Hash {503        try self.ensureUsable();504        if (resolved.len == 0) return try self.workingRoot();505        var entries = try history.conflictEntries(allocator, self.session.workingRoot().conflicts);506        defer entries.deinit();507        for (resolved) |hash| {508            if (!hasConflictHash(entries.entries, hash)) return error.ConflictNotFound;509        }510511        var remaining: std.ArrayList(version.ConflictEntry) = .empty;512        defer remaining.deinit(allocator);513        for (entries.entries) |entry| {514            if (!hasConflictHashValue(resolved, entry.hash)) try remaining.append(allocator, entry);515        }516517        const root = try history.putConflictRoot(remaining.items);518        return try self.replaceConflictRoot(root);519    }520521    pub fn checkout(self: *const Connection) Error!@import("branch.zig").Checkout {522        try self.ensureUsable();523        return self.session.checkout;524    }525526    pub fn workingRoot(self: *const Connection) Error!version.Hash {527        try self.ensureUsable();528        return self.session.checkout.working.working;529    }530531    fn relationViewFromRoot(allocator: Allocator, history: *const history_mod.History, root_hash: version.Hash) Error!RelationView {532        var root = try history.relationRoot(allocator, root_hash);533        errdefer root.deinit();534        const rows = try history.relationRows(allocator, root_hash);535        errdefer version.freeRelationRows(allocator, rows);536        return .{537            .allocator = allocator,538            .root = root,539            .rows = rows,540        };541    }542543    fn databaseViewFromRoot(allocator: Allocator, history: *const history_mod.History, root: version.DatabaseRoot) Error!DatabaseView {544        var conflicts = try history.conflictArtifacts(allocator, root.conflicts);545        errdefer conflicts.deinit();546547        const relations = try allocator.alloc(RelationView, root.entries.len);548        errdefer allocator.free(relations);549        var relation_count: usize = 0;550        errdefer for (relations[0..relation_count]) |*relation_view| relation_view.deinit();551552        const snapshots = try allocator.alloc(diff_mod.RelationSnapshot, root.entries.len);553        errdefer allocator.free(snapshots);554555        for (root.entries, relations, snapshots) |entry, *relation_view, *snapshot| {556            relation_view.* = try relationViewFromRoot(allocator, history, entry.hash);557            relation_count += 1;558            snapshot.* = relation_view.snapshot();559        }560561        return .{562            .allocator = allocator,563            .relations = relations,564            .snapshots = snapshots,565            .conflicts = conflicts,566        };567    }568569    fn materializeDatabaseValue(self: *Connection, allocator: Allocator, value: *const version.DatabaseValue, options: file.CommitOptions) Error!void {570        var plan = try DatabaseMaterialization.init(self, allocator, value);571        defer plan.deinit();572        try plan.apply(self, options);573    }574575    fn materializeDatabaseMerge(576        self: *Connection,577        allocator: Allocator,578        history: *history_mod.History,579        result: *const merge_mod.DatabaseMerge,580        options: file.CommitOptions,581    ) Error!void {582        if (!version.same(result.value.root.conflicts, result.conflict_root.hash)) {583            return error.InvalidHistory;584        }585        try result.persistConflicts(history);586        try history.validateConflictRoot(result.value.root.conflicts);587        try self.materializeDatabaseValue(allocator, &result.value, options);588        try self.verifyMaterializedDatabaseRoot(allocator, &result.value.root);589    }590591    fn publishStagedValue(self: *Connection, history: *history_mod.History, root: version.Hash) Error!void {592        const allocator = self.session.allocator;593        const working = self.session.workingRoot();594        if (!version.same(working.hash, root)) return error.InvalidHistory;595        try history.validateConflictRoot(working.conflicts);596        var live_root = try version.databaseRootMaintained(597            allocator,598            &self.catalog,599            working.conflicts,600        );601        defer live_root.deinit();602        if (!version.same(live_root.hash, working.hash)) return error.InvalidHistory;603604        var batch = try history.beginWriteBatch();605        errdefer batch.deinit();606607        var base_root: ?version.DatabaseRoot = null;608        defer if (base_root) |*owned| owned.deinit();609        base_root = history.commitDatabaseRoot(allocator, self.session.checkout.head) catch null;610611        for (working.entries) |entry| {612            const base = self.publicationBase(&base_root, entry.name);613            const need = try history.relationRowsNeed(allocator, entry.hash, base);614615            var state = try self.catalog.readRelation(allocator, entry.name);616            defer state.deinit();617            const live_key = try version.relationKey(618                entry.name,619                &state.handle,620                state.relationStats(),621            );622            if (!version.same(live_key.hash, entry.hash)) return error.InvalidHistory;623            if (need == .none and history.hasRelationRoot(entry.hash)) continue;624625            var relation_root = try version.relationRootMaintained(626                allocator,627                entry.name,628                state.schema,629                &state.handle,630                state.relationStats(),631            );632            defer relation_root.deinit();633            try history.putRelationRoot(relation_root);634635            switch (need) {636                .none => {},637                .full => {638                    var writer = try history.beginRelationRows(entry.hash);639                    defer writer.deinit();640                    try streamRelationRows(&writer, allocator, &state.handle, null);641                    try writer.finish();642                },643                .suffix => |plan| {644                    var writer = try history.beginRelationRowsSuffix(entry.hash, base.?, plan);645                    defer writer.deinit();646                    try streamRelationRows(&writer, allocator, &state.handle, plan.boundary);647                    try writer.finish();648                },649            }650        }651        try history.putDatabaseRoot(working.*);652        try batch.finish();653    }654655    fn streamRelationRows(656        writer: *history_mod.RelationRowsWriter,657        allocator: Allocator,658        handle: *const catalog_mod.RelationHandle,659        boundary: ?i64,660    ) Error!void {661        if (boundary) |value| {662            if (value == std.math.maxInt(i64)) return;663        }664        const start: ?i64 = if (boundary) |value| value + 1 else null;665        var scan: relation_mod.Scan = undefined;666        try handle.relation.scan(&scan, allocator, start, null);667        defer scan.deinit();668        while (try scan.next()) |row_entry| try writer.append(row_entry.rowid, row_entry.bytes);669    }670671    fn publicationBase(self: *const Connection, base_root: *const ?version.DatabaseRoot, name: []const u8) ?history_mod.IncrementalBase {672        const root = base_root.* orelse return null;673        const edited = self.session.pendingEditRowids(root.hash, name) orelse return null;674        for (root.entries) |entry| {675            if (std.mem.eql(u8, entry.name, name)) return .{ .root = entry.hash, .edited = edited };676        }677        return null;678    }679680    pub fn materializedWorkingValue(self: *Connection, allocator: Allocator) Error!version.DatabaseValue {681        try self.ensureUsable();682        const working = self.session.workingRoot();683        return try version.databaseValue(allocator, &self.catalog, working.conflicts);684    }685};

1pub const Connection = connection.Connection;