// SPDX-License-Identifier: LGPL-3.0-only
// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®

module Crafter.DB:Storage_impl;
import :Storage;
import :Error;
import :Format;
import :Formatter;
import :File;
import std;

using namespace Crafter::DB;
namespace fs = std::filesystem;

namespace {
    constexpr std::uint64_t maxExtentBytes = std::uint64_t(1) << 30;
    constexpr std::uint64_t willNeedLimit = std::uint64_t(256) << 20;
    constexpr std::uint32_t regionRows = 0;
    constexpr std::uint32_t regionHeap = 1;

    constexpr std::uint64_t RoundUp(std::uint64_t bytes, std::uint64_t granule) noexcept {
        return (bytes + granule - 1) / granule * granule;
    }

    // Extents are aligned to the larger of the format page and the running
    // kernel's page size, so they can be mmapped on 16K and 64K kernels too.
    std::uint64_t Granule() noexcept {
        return std::max<std::uint64_t>(pageSize, Os::PageSize());
    }

    std::span<const std::byte> Bytes(const auto& object) noexcept {
        return std::as_bytes(std::span(&object, 1));
    }

    void FillCommit(CommitRecord& record, std::uint64_t seq, std::uint64_t rowCount, std::uint64_t liveCount, std::uint64_t heapUsed, std::uint64_t heapGarbage, const std::array<std::uint32_t, regionSlots>& extentCount, std::uint64_t fileLength, std::uint64_t flags) noexcept {
        record = CommitRecord{};
        record.commitSeq = seq;
        record.rowCount = rowCount;
        record.liveCount = liveCount;
        record.heapUsed = heapUsed;
        record.heapGarbage = heapGarbage;
        record.extentCount = extentCount;
        record.fileLength = fileLength;
        record.modifiedUnixNanos = Os::UnixNanos();
        record.flags = flags;
        record.crc32c = CommitCrc(record);
    }

    // Picks the newest valid commit slot; nullopt when both are damaged.
    std::optional<std::size_t> ChooseCommit(const FileHeader& header) noexcept {
        const bool valid0 = CommitValid(header.commits[0]);
        const bool valid1 = CommitValid(header.commits[1]);
        if (valid0 && valid1) return header.commits[0].commitSeq >= header.commits[1].commitSeq ? 0 : 1;
        if (valid0) return 0;
        if (valid1) return 1;
        return std::nullopt;
    }

    void ValidateHeader(const FileHeader& header, std::uint64_t fileSize, const fs::path& path) {
        if (header.magic != fileMagic) throw CorruptError(std::format("{}: not a Crafter.DB table (bad magic)", path.string()));
        if (header.formatVersion != formatVersion) throw CorruptError(std::format("{}: file format {} is not supported by this build (format {})", path.string(), header.formatVersion, formatVersion));
        if ((header.formatFlags & formatFlagLittleEndian) == 0) throw CorruptError(std::format("{}: unsupported byte order", path.string()));
        if (header.pageSize < pageSize || header.pageSize % pageSize != 0) throw CorruptError(std::format("{}: extent granule {} is invalid", path.string(), header.pageSize));
        if (header.pageSize % Os::PageSize() != 0) throw IOError(std::format("table was created with {}-byte extents, which cannot be mapped on a system with {}-byte pages", header.pageSize, Os::PageSize()), std::make_error_code(std::errc::not_supported), path);
        if (header.columnCount == 0 || (header.orderedColumn != noColumn && header.orderedColumn >= header.columnCount)) throw CorruptError(std::format("{}: column count or ordered column is invalid", path.string()));
        if (header.headerPages < 2 || std::uint64_t(header.headerPages) * pageSize > fileSize) throw CorruptError(std::format("{}: header page count {} is invalid", path.string(), header.headerPages));
        if (header.regionCount == 0 || header.regionCount > regionSlots) throw CorruptError(std::format("{}: region count {} is invalid", path.string(), header.regionCount));
        if (header.rowSize == 0 || header.rowAlign == 0 || header.rowAlign > 16 || header.rowSize % header.rowAlign != 0) throw CorruptError(std::format("{}: row size {} / alignment {} is invalid", path.string(), header.rowSize, header.rowAlign));
        if (std::uint64_t(header.schemaBlobLength) + pageSize > std::uint64_t(header.headerPages) * pageSize) throw CorruptError(std::format("{}: schema blob does not fit the header pages", path.string()));
    }

    // Reads header, commit and schema blob; shared by Inspect and open. The
    // header and the first schema page are fetched in one pread, which is the
    // whole description for every table with fewer than ~170 columns.
    void ReadDescription(const File& file, std::uint64_t size, FileHeader& header, std::size_t& chosen, SchemaInfo& schema) {
        if (size < 2 * pageSize) throw CorruptError(std::format("{}: file is smaller than the header", file.Path().string()));
        std::vector<std::byte> pages(2 * pageSize);
        file.ReadAt(0, pages);
        std::memcpy(&header, pages.data(), sizeof(header));
        ValidateHeader(header, size, file.Path());
        const std::optional<std::size_t> slot = ChooseCommit(header);
        if (!slot) throw CorruptError(std::format("{}: both commit records are damaged", file.Path().string()));
        chosen = *slot;
        std::span<const std::byte> blob;
        std::vector<std::byte> rest;
        if (header.schemaBlobLength <= pageSize) {
            blob = std::span(pages).subspan(pageSize, header.schemaBlobLength);
        } else {
            rest.resize(header.schemaBlobLength);
            file.ReadAt(pageSize, rest);
            blob = rest;
        }
        if (!SchemaInfo::Parse(blob, header.rowAlign, schema)) throw CorruptError(std::format("{}: schema blob is malformed", file.Path().string()));
        if (schema.rowSize != header.rowSize || schema.columns.size() != header.columnCount) throw CorruptError(std::format("{}: schema blob disagrees with the header", file.Path().string()));
    }

    // Reads a scalar column as the variant alternative its kind implies.
    std::optional<ScalarValue> ReadScalar(const std::byte* row, const ColumnDescriptor& column) noexcept {
        const std::byte* at = row + column.offset;
        switch (column.kind) {
            case Kind::Flags: case Kind::Bool: case Kind::UInt8: { std::uint8_t v; std::memcpy(&v, at, 1); return ScalarValue(std::uint64_t(v)); }
            case Kind::UInt16: { std::uint16_t v; std::memcpy(&v, at, 2); return ScalarValue(std::uint64_t(v)); }
            case Kind::UInt32: { std::uint32_t v; std::memcpy(&v, at, 4); return ScalarValue(std::uint64_t(v)); }
            case Kind::UInt64: { std::uint64_t v; std::memcpy(&v, at, 8); return ScalarValue(v); }
            case Kind::Int8: { std::int8_t v; std::memcpy(&v, at, 1); return ScalarValue(std::int64_t(v)); }
            case Kind::Int16: { std::int16_t v; std::memcpy(&v, at, 2); return ScalarValue(std::int64_t(v)); }
            case Kind::Int32: { std::int32_t v; std::memcpy(&v, at, 4); return ScalarValue(std::int64_t(v)); }
            case Kind::Int64: case Kind::TimePoint: case Kind::Duration: {
                if (column.size == 8) { std::int64_t v; std::memcpy(&v, at, 8); return ScalarValue(v); }
                if (column.size == 4) { std::int32_t v; std::memcpy(&v, at, 4); return ScalarValue(std::int64_t(v)); }
                return std::nullopt;
            }
            case Kind::Float32: { float v; std::memcpy(&v, at, 4); return ScalarValue(double(v)); }
            case Kind::Float64: { double v; std::memcpy(&v, at, 8); return ScalarValue(v); }
            case Kind::String: case Kind::Blob: case Kind::FixedBytes: return std::nullopt;
        }
        return std::nullopt;
    }

    // cmp_less that also accepts doubles (std::cmp_less is integers only).
    template<class A, class B>
    bool CmpLessPartial(const A& x, const B& y) noexcept {
        if constexpr (std::is_floating_point_v<A> || std::is_floating_point_v<B>) return static_cast<double>(x) < static_cast<double>(y);
        else return std::cmp_less(x, y);
    }

    bool ScalarLess(const ScalarValue& a, const ScalarValue& b) noexcept {
        return std::visit([](const auto& x, const auto& y) { return CmpLessPartial(x, y); }, a, b);
    }

    std::uint32_t VarAlign(const ColumnDescriptor& column) noexcept {
        return column.kind == Kind::Blob && column.paramNum != 0 ? column.paramNum : 1;
    }
}

// --- creation ---------------------------------------------------------------

fs::path TableFile::TempPath(const fs::path& path) {
    std::array<std::byte, 6> random{};
    Os::RandomBytes(random);
    std::string suffix;
    for (const std::byte b : random) suffix += std::format("{:02x}", static_cast<std::uint8_t>(b));
    return fs::path(std::format("{}.tmp-{}", path.string(), suffix));
}

fs::path TableFile::BackupPath(const fs::path& path, std::uint32_t version) {
    return fs::path(std::format("{}.v{}.bak", path.string(), version));
}

void TableFile::Replace(const fs::path& path, const fs::path& temp) {
    Os::Rename(temp, path);
    Os::SyncDirectory(path.parent_path());
}

void TableFile::Create(const fs::path& path, const CreateParams& params, const OpenOptions& options) {
    const std::vector<std::byte> blob = params.schema.Serialize();
    const std::uint64_t granule = Granule();
    FileHeader header{};
    header.magic = fileMagic;
    header.formatVersion = formatVersion;
    header.headerPages = static_cast<std::uint16_t>(1 + RoundUp(blob.size(), pageSize) / pageSize);
    header.formatFlags = formatFlagLittleEndian;
    header.pageSize = static_cast<std::uint32_t>(granule);
    header.rowSize = params.schema.rowSize;
    header.rowAlign = params.schema.rowAlign;
    header.columnCount = static_cast<std::uint16_t>(params.schema.columns.size());
    header.orderedColumn = params.orderedColumn;
    header.regionCount = regionCount;
    header.schemaVersion = params.schemaVersion;
    header.schemaFingerprint = params.fingerprint;
    header.schemaBlobLength = static_cast<std::uint32_t>(blob.size());
    header.minExtentBytes = minExtentBytes;
    header.createdUnixNanos = Os::UnixNanos();
    Os::RandomBytes(std::as_writable_bytes(std::span(header.fileUuid)));
    header.migratedFromVersion = params.migratedFromVersion;

    std::array<std::uint32_t, regionSlots> extentCount{};
    const std::uint64_t headerBytes = std::uint64_t(header.headerPages) * pageSize;
    std::uint64_t fileLength = RoundUp(headerBytes, granule);
    header.extents[regionRows][0] = { fileLength, std::max<std::uint64_t>(RoundUp(minExtentBytes, granule), RoundUp(params.initialRowBytes, granule)) };
    extentCount[regionRows] = 1;
    fileLength += header.extents[regionRows][0].length;
    if (params.schema.HasHeap()) {
        header.extents[regionHeap][0] = { fileLength, std::max<std::uint64_t>(RoundUp(minExtentBytes, granule), RoundUp(params.initialHeapBytes, granule)) };
        extentCount[regionHeap] = 1;
        fileLength += header.extents[regionHeap][0].length;
    }
    FillCommit(header.commits[0], 1, 0, 0, 0, 0, extentCount, fileLength, commitFlagCleanClose);

    if (options.createDirectories && !path.parent_path().empty()) Os::CreateDirectories(path.parent_path(), options.directoryPermissions);
    const fs::path temp = TempPath(path);
    try {
        File file = File::CreateExclusive(temp, options.filePermissions);
        file.WriteAt(0, Bytes(header));
        file.WriteAt(pageSize, blob);
        file.Allocate(headerBytes, fileLength - headerBytes);
        file.Sync();
        file.Close();
        // Never replace: two processes creating the same table at once must
        // end up sharing one file, not each writing to its own inode.
        if (!Os::RenameNoReplace(temp, path)) {
            Os::Remove(temp);
            throw IOError("table already exists", std::make_error_code(std::errc::file_exists), path);
        }
        Os::SyncDirectory(path.parent_path());
    } catch (...) {
        Os::Remove(temp);
        throw;
    }
}

Inspection TableFile::Inspect(const fs::path& path) {
    const File file = File::Open(path, false);
    Inspection result;
    std::size_t chosen = 0;
    ReadDescription(file, file.Size(), result.header, chosen, result.schema);
    result.commit = result.header.commits[chosen];
    return result;
}

void TableFile::RemoveStaleTemps(const fs::path& path) noexcept {
    try {
        const std::string prefix = std::format("{}.tmp-", path.filename().string());
        const fs::path directory = path.parent_path().empty() ? fs::path(".") : path.parent_path();
        for (const fs::directory_entry& entry : fs::directory_iterator(directory)) {
            if (!entry.path().filename().string().starts_with(prefix)) continue;
            try {
                File probe = File::Open(entry.path(), true);
                probe.LockExclusive();   // held by a live writer -> LockedError -> keep it
                Os::Remove(entry.path());
            } catch (...) {
            }
        }
    } catch (...) {
    }
}

// --- open -------------------------------------------------------------------

TableFile::TableFile(const fs::path& path, const OpenOptions& openOptions) : options(openOptions), writable(!openOptions.readOnly), ownerProcess(Os::ProcessId()) {
    file = File::Open(path, writable);
    if (writable) file.LockExclusive();
    else file.LockShared();

    const std::uint64_t size = file.Size();
    ReadDescription(file, size, header, activeSlot, schema);
    current = header.commits[activeSlot];
    if (current.fileLength > size) throw CorruptError(std::format("{}: committed length {} exceeds the file size {}", path.string(), current.fileLength, size));
    if (current.fileLength < std::uint64_t(header.headerPages) * pageSize) throw CorruptError(std::format("{}: committed length is inside the header", path.string()));
    for (std::size_t r = 0; r < regionSlots; r++) {
        if (current.extentCount[r] > maxExtents || (r >= header.regionCount && current.extentCount[r] != 0)) throw CorruptError(std::format("{}: extent count of region {} is invalid", path.string(), r));
    }

    rows.Reserve(options.reserveBytes);
    heapRegion.Reserve(options.reserveBytes);
    const auto mapRegion = [&](MappedRegion& region, std::uint32_t index) {
        for (std::uint32_t e = 0; e < current.extentCount[index]; e++) {
            const Extent extent = header.extents[index][e];
            if (extent.length == 0 || extent.length % header.pageSize != 0 || extent.fileOffset % header.pageSize != 0 || extent.fileOffset < std::uint64_t(header.headerPages) * pageSize || extent.length > current.fileLength || extent.fileOffset > current.fileLength - extent.length) {
                throw CorruptError(std::format("{}: extent {} of region {} lies outside the committed file", path.string(), e, index));
            }
            region.Map(file, extent.fileOffset, extent.length, writable);
        }
    };
    mapRegion(rows, regionRows);
    mapRegion(heapRegion, regionHeap);
    // One madvise pair per region now; extents added later are advised as
    // they are mapped.
    if (options.excludeFromCoreDumps) {
        rows.ExcludeFromCoreDumps();
        heapRegion.ExcludeFromCoreDumps();
    }
    if (current.rowCount > rows.Mapped() / header.rowSize) throw CorruptError(std::format("{}: {} committed rows do not fit the mapped row extents", path.string(), current.rowCount));
    if (current.heapUsed > heapRegion.Mapped()) throw CorruptError(std::format("{}: committed heap length exceeds the mapped heap extents", path.string()));

    extentCount = current.extentCount;
    fileLength = current.fileLength;
    publishedRows.store(current.rowCount, std::memory_order_relaxed);
    publishedLive.store(current.liveCount, std::memory_order_relaxed);
    publishedHeap.store(current.heapUsed, std::memory_order_relaxed);
    heapGarbage = current.heapGarbage;
    heap = Heap{ heapRegion.Base(), &publishedHeap };
    writer.Attach(this, heapRegion.Base(), current.heapUsed, heapRegion.Mapped());
    for (std::size_t c = 0; c < schema.columns.size(); c++) {
        if (schema.columns[c].kind == Kind::String || schema.columns[c].kind == Kind::Blob) varColumns.push_back(c);
    }

    // An unclean close means tombstones (and orphaned heap bytes) may have
    // reached the disk after the last commit; recount so the "no tombstones"
    // scan fast path and the garbage figure stay honest.
    if ((current.flags & commitFlagCleanClose) == 0) RecountLive();
    tombstones.store(publishedLive.load(std::memory_order_relaxed) != publishedRows.load(std::memory_order_relaxed), std::memory_order_relaxed);

    if (options.prefetch) {
        const std::uint64_t rowBytes = current.rowCount * header.rowSize;
        if (rowBytes < willNeedLimit) rows.WillNeed(rowBytes);
        if (current.heapUsed < willNeedLimit) heapRegion.WillNeed(current.heapUsed);
    }

    if (writable) {
        // Uncommitted growth from an earlier crash: drop it so extents we add
        // start exactly at the committed end.
        if (size > current.fileLength) file.Truncate(current.fileLength);
        headerPages.Map(file, 0, pageSize, true);
        // Mark the file open: a crash from here on must recount on next open.
        WriteCommit(false);
        RemoveStaleTemps(path);
    }
}

TableFile::~TableFile() {
    try {
        Close();
    } catch (...) {
    }
}

void TableFile::Close() {
    if (closed) return;
    closed = true;
    // A forked child inherited this object; its counters describe the parent's
    // state at fork time and must never reach the file.
    if (writable && file.IsOpen() && OwnedByThisProcess()) Flush(true);
    headerPages.Release();
    rows.Release();
    heapRegion.Release();
    file.Close();
}

bool TableFile::OwnedByThisProcess() const noexcept {
    return Os::ProcessId() == ownerProcess;
}

void TableFile::RequireWritable() const {
    if (!writable) throw ReadOnlyError(std::format("{}: table is open read-only", file.Path().string()));
    if (closed) throw Error(std::format("{}: table is closed", file.Path().string()));
    if (!OwnedByThisProcess()) throw Error(std::format("{}: table was opened by another process (inherited through fork)", file.Path().string()));
}

TableStats TableFile::Stats() const noexcept {
    return { SlotCount(), LiveCount(), HeapUsed(), heapGarbage, fileLength, current.commitSeq };
}

bool TableFile::Deleted(const std::byte* row) noexcept {
    return (std::atomic_ref<std::uint8_t>(*const_cast<std::uint8_t*>(reinterpret_cast<const std::uint8_t*>(row))).load(std::memory_order_relaxed) & rowFlagDeleted) != 0;
}

void TableFile::RecountLive() noexcept {
    const std::uint64_t count = publishedRows.load(std::memory_order_relaxed);
    std::uint64_t live = 0;
    std::uint64_t referenced = 0;
    for (std::uint64_t i = 0; i < count; i++) {
        const std::byte* row = Slot(i);
        if (Deleted(row)) continue;
        live++;
        referenced += HeapBytesOf(row);
    }
    publishedLive.store(live, std::memory_order_relaxed);
    // Everything in the heap that no live row references is garbage: this
    // also covers bytes orphaned by Update/Set before the crash.
    const std::uint64_t used = publishedHeap.load(std::memory_order_relaxed);
    heapGarbage = used > referenced ? used - referenced : 0;
}

// --- growth -----------------------------------------------------------------

void TableFile::AddExtent(std::uint32_t region, std::uint64_t minimumBytes) {
    RequireWritable();
    const std::uint32_t count = extentCount[region];
    if (count >= maxExtents) throw IOError("table has reached its maximum number of extents; Compact() it", std::make_error_code(std::errc::file_too_large), file.Path());
    const std::uint64_t granule = header.pageSize;
    const std::uint64_t previous = count == 0 ? 0 : header.extents[region][count - 1].length;
    const std::uint64_t length = std::max({ RoundUp(minExtentBytes, granule), std::min(previous * 2, maxExtentBytes), RoundUp(minimumBytes, granule) });
    const Extent extent{ fileLength, length };
    file.Allocate(extent.fileOffset, extent.length);
    MappedRegion& target = region == regionRows ? rows : heapRegion;
    target.Map(file, extent.fileOffset, extent.length, true);
    header.extents[region][count] = extent;
    std::memcpy(headerPages.Data() + __builtin_offsetof(FileHeader, extents) + (region * maxExtents + count) * sizeof(Extent), &extent, sizeof(extent));
    extentCount[region] = count + 1;
    fileLength = extent.fileOffset + extent.length;
}

void TableFile::EnsureHeapCapacity(std::uint64_t bytes) {
    while (heapRegion.Mapped() < bytes) AddExtent(regionHeap, bytes - heapRegion.Mapped());
    writer.SetCapacity(heapRegion.Mapped());
}

void TableFile::Arena::Grow(std::uint64_t neededCapacity) {
    owner->EnsureHeapCapacity(neededCapacity);
}

// --- writes -----------------------------------------------------------------

std::byte* TableFile::BeginAppend(std::uint64_t count) {
    RequireWritable();
    const std::uint64_t slots = publishedRows.load(std::memory_order_relaxed);
    const std::uint64_t needed = (slots + count) * header.rowSize;
    while (rows.Mapped() < needed) AddExtent(regionRows, needed - rows.Mapped());
    return rows.Base() + slots * header.rowSize;
}

void TableFile::CommitAppend(std::uint64_t count) noexcept {
    publishedHeap.store(writer.Cursor(), std::memory_order_release);
    publishedLive.fetch_add(count, std::memory_order_relaxed);
    publishedRows.fetch_add(count, std::memory_order_release);
}

void TableFile::AbortAppend() noexcept {
    writer.Rewind(publishedHeap.load(std::memory_order_relaxed));
}

std::byte* TableFile::MutableSlot(std::uint64_t index) {
    RequireWritable();
    return rows.Base() + index * header.rowSize;
}

bool TableFile::MarkDeleted(std::uint64_t index) {
    RequireWritable();
    std::byte* row = rows.Base() + index * header.rowSize;
    // The flag goes up before the byte, so a scan that starts after this call
    // checks every row.
    tombstones.store(true, std::memory_order_release);
    const std::uint8_t previous = std::atomic_ref<std::uint8_t>(*reinterpret_cast<std::uint8_t*>(row)).fetch_or(rowFlagDeleted, std::memory_order_relaxed);
    if ((previous & rowFlagDeleted) != 0) return false;
    publishedLive.fetch_sub(1, std::memory_order_relaxed);
    heapGarbage += HeapBytesOf(row);
    return true;
}

std::uint64_t TableFile::HeapBytesOf(const std::byte* row) const noexcept {
    std::uint64_t bytes = 0;
    for (const std::size_t c : varColumns) {
        const ColumnDescriptor& column = schema.columns[c];
        Var var;
        std::memcpy(&var, row + column.offset, sizeof(var));
        if (!VarInline(var.length, VarAlign(column))) bytes += var.length;
    }
    return bytes;
}

void TableFile::WriteCommit(bool cleanClose) {
    const std::size_t slot = activeSlot == 0 ? 1 : 0;
    CommitRecord record;
    FillCommit(record, current.commitSeq + 1, publishedRows.load(std::memory_order_relaxed), publishedLive.load(std::memory_order_relaxed), publishedHeap.load(std::memory_order_relaxed), heapGarbage, extentCount, fileLength, cleanClose ? commitFlagCleanClose : 0);
    std::memcpy(headerPages.Data() + __builtin_offsetof(FileHeader, commits) + slot * sizeof(CommitRecord), &record, sizeof(record));
    headerPages.Sync();
    current = record;
    header.commits[slot] = record;
    activeSlot = slot;
}

void TableFile::Flush(bool cleanClose) {
    if (!writable || (closed && !file.IsOpen())) return;
    if (!OwnedByThisProcess()) throw Error(std::format("{}: table was opened by another process (inherited through fork)", file.Path().string()));
    // Data first: the commit record must never describe rows or heap bytes
    // that are not yet on disk.
    file.SyncData();
    WriteCommit(cleanClose);
}

// --- compaction -------------------------------------------------------------

void TableFile::CompactInto(TableFile& target) const {
    target.RequireWritable();
    if (target.RowSize() != RowSize() || target.Schema().columns.size() != schema.columns.size()) throw Error("CompactInto: target table has a different schema");
    const std::uint64_t slots = SlotCount();
    const std::uint64_t live = LiveCount();
    if (live == 0) return;
    std::byte* out = target.BeginAppend(live);
    HeapWriter& arena = target.Writer();
    std::uint64_t written = 0;
    for (std::uint64_t i = 0; i < slots && written < live; i++) {
        const std::byte* row = Slot(i);
        if (Deleted(row)) continue;
        std::memcpy(out, row, header.rowSize);
        for (const std::size_t c : varColumns) {
            const ColumnDescriptor& column = schema.columns[c];
            if ((column.flags & columnFlagOptional) != 0 && column.presenceOffset != noColumn) {
                // The payload of a disengaged optional carries no value: write
                // an empty handle rather than carrying stale bytes forward.
                std::uint8_t present;
                std::memcpy(&present, out + column.presenceOffset, 1);
                if (present == 0) {
                    const Var empty{};
                    std::memcpy(out + column.offset, &empty, sizeof(empty));
                    continue;
                }
            }
            Var var;
            std::memcpy(&var, out + column.offset, sizeof(var));
            if (VarInline(var.length, VarAlign(column))) continue;
            const std::span<const std::byte> bytes = heap.Read(var.tailOrOffset, var.length);
            var.tailOrOffset = arena.Write(bytes, VarAlign(column));
            std::memcpy(out + column.offset, &var, sizeof(var));
        }
        out += header.rowSize;
        written++;
    }
    target.CommitAppend(written);
}

// --- verification -----------------------------------------------------------

VerifyResult TableFile::Verify() const {
    VerifyResult result;
    const std::uint64_t slots = SlotCount();
    const std::uint64_t heapLimit = HeapUsed();
    const std::uint16_t ordered = header.orderedColumn;
    std::optional<ScalarValue> previousKey;
    std::uint64_t live = 0;
    for (std::uint64_t i = 0; i < slots; i++) {
        const std::byte* row = Slot(i);
        std::uint8_t flags;
        std::memcpy(&flags, row, 1);
        if ((flags & ~rowFlagDeleted) != 0) result.problems.push_back(std::format("row {}: unknown flag bits {:#x}", i, flags));
        if ((flags & rowFlagDeleted) == 0) live++;
        for (std::size_t c = 1; c < schema.columns.size(); c++) {
            const ColumnDescriptor& column = schema.columns[c];
            if ((column.flags & columnFlagPresence) != 0) {
                std::uint8_t present;
                std::memcpy(&present, row + column.offset, 1);
                if (present > 1) result.problems.push_back(std::format("row {}: presence byte of '{}' is {}", i, schema.Name(c), present));
            }
            if (column.kind == Kind::String || column.kind == Kind::Blob) {
                Var var;
                std::memcpy(&var, row + column.offset, sizeof(var));
                if (!VarInline(var.length, VarAlign(column)) && (var.tailOrOffset > heapLimit || var.length > heapLimit - var.tailOrOffset)) {
                    result.problems.push_back(std::format("row {}: '{}' points at heap [{}, +{}) beyond the committed {} bytes", i, schema.Name(c), var.tailOrOffset, var.length, heapLimit));
                }
            }
        }
        if (ordered != noColumn && ordered < schema.columns.size()) {
            const std::optional<ScalarValue> key = ReadScalar(row, schema.columns[ordered]);
            if (key && previousKey && ScalarLess(*key, *previousKey)) result.problems.push_back(std::format("row {}: '{}' is out of order", i, schema.Name(ordered)));
            if (key) previousKey = key;
        }
        if (result.problems.size() > 1000) {
            result.problems.push_back("... further problems omitted");
            return result;
        }
    }
    if (live != LiveCount()) result.problems.push_back(std::format("live count {} does not match the {} rows without a tombstone", LiveCount(), live));
    return result;
}

// --- DynamicRow -------------------------------------------------------------

bool DynamicRow::Deleted() const noexcept {
    return TableFile::Deleted(row);
}

std::optional<std::size_t> DynamicRow::Find(std::string_view name) const noexcept {
    for (std::size_t i = 0; i < schema->columns.size(); i++) {
        if (schema->Name(i) == name) return i;
    }
    return std::nullopt;
}

bool DynamicRow::Present(std::size_t index) const noexcept {
    const ColumnDescriptor& column = schema->columns[index];
    if ((column.flags & columnFlagOptional) == 0 || column.presenceOffset == noColumn) return true;
    std::uint8_t present;
    std::memcpy(&present, row + column.presenceOffset, 1);
    return present != 0;
}

std::optional<ScalarValue> DynamicRow::Scalar(std::size_t index) const noexcept {
    return ReadScalar(row, schema->columns[index]);
}

std::optional<std::span<const std::byte>> DynamicRow::Bytes(std::size_t index) const {
    const ColumnDescriptor& column = schema->columns[index];
    if (column.kind == Kind::FixedBytes) return std::span<const std::byte>(row + column.offset, column.size);
    if (column.kind != Kind::String && column.kind != Kind::Blob) return std::nullopt;
    Var var;
    std::memcpy(&var, row + column.offset, sizeof(var));
    if (VarInline(var.length, VarAlign(column))) return std::span<const std::byte>(row + column.offset + 4, var.length);
    return heap->Read(var.tailOrOffset, var.length);
}

std::optional<std::string_view> DynamicRow::Text(std::size_t index) const {
    const ColumnDescriptor& column = schema->columns[index];
    if (column.kind != Kind::String && column.kind != Kind::FixedBytes) return std::nullopt;
    const std::optional<std::span<const std::byte>> bytes = Bytes(index);
    if (!bytes) return std::nullopt;
    std::string_view text(reinterpret_cast<const char*>(bytes->data()), bytes->size());
    if (column.kind == Kind::FixedBytes) text = text.substr(0, text.find('\0'));
    return text;
}

std::string DynamicRow::Describe() const {
    std::string out;
    for (std::size_t i = 1; i < schema->columns.size(); i++) {
        const ColumnDescriptor& column = schema->columns[i];
        if ((column.flags & columnFlagPresence) != 0) continue;
        if (!out.empty()) out += ", ";
        out += schema->Name(i);
        out += '=';
        if (!Present(i)) {
            out += "null";
            continue;
        }
        if (const std::optional<std::string_view> text = Text(i)) {
            out += std::format("\"{}\"", *text);
        } else if (const std::optional<ScalarValue> scalar = Scalar(i)) {
            std::visit([&](const auto& v) { out += std::format("{}", v); }, *scalar);
        } else if (const std::optional<std::span<const std::byte>> bytes = Bytes(i)) {
            out += std::format("<{} bytes>", bytes->size());
        } else {
            out += '?';
        }
    }
    return out;
}