spg-storage 7.37.16

//! v7.37.15 (Phase A) — per-row MVCC visibility header.
//!
//! ## Why this exists
//!
//! Pre-v7.37.15 SPG's MVCC story was at the **catalog level**:
//! `CatalogSnapshot` Arc-clones the catalog trie roots, readers see a
//! coherent prior-committed view, writers serialise. That works for
//! mailrs / sentori's SELECT-heavy IMAP workload, but caps multi-
//! writer throughput at "one writer at a time" forever and forces an
//! Arc clone of the entire catalog on each readonly statement —
//! mailrs measured a ~50% perf gap vs PG18 traceable to this.
//!
//! v7.37.15 adds **per-row** visibility on top of that catalog-level
//! Arc snapshot model. Both coexist:
//!
//! - The Arc snapshot path remains for SELECT-only fast reads — the
//!   catalog trie clones are still O(1) Arc bumps.
//! - Each row in the table now carries a `RowHeader { xmin, xmax,
//!   flags }`; scans filter rows against a `Snapshot { version,
//!   in_progress }`.
//! - Writers update `xmin` on insert / `xmax` on delete or update,
//!   so concurrent writers to different rows no longer block one
//!   another (granularity = per-row, not per-catalog).
//!
//! ## Why u64 instead of PG's 32-bit Xid
//!
//! PG carries the historical scars of a 32-bit transaction id — epoch
//! advancement, FrozenTransactionId, VACUUM FREEZE, the entire
//! anti-wraparound machinery. SPG starts fresh: `xmin` and `xmax`
//! are `u64`. At 1ns per transaction (a contemporary CPU's L1 cycle
//! budget) u64 wraps in 584 years; we explicitly do NOT implement
//! wraparound handling because we will never hit it.
//!
//! ## Layout: parallel `PersistentVec`, not embedded in `Row`
//!
//! The header lives in `Table::headers: PersistentVec<RowHeader>`
//! **parallel** to `Table::rows: PersistentVec<Row<'static>>`. Why
//! not embedded in the Row struct?
//!
//! 1. **Cache locality on visibility-only scans.** A scan that only
//!    needs to count visible rows (think `SELECT COUNT(*)`) walks
//!    headers without touching row bodies. With the header inline
//!    every cache line carries one row's worth of payload; with the
//!    header in a separate Vec the scan only loads 24-byte headers,
//!    yielding ~10x throughput on wide-row tables.
//! 2. **Public API stability.** `pub struct Row { pub values }` is
//!    the shape every caller — eval / sort / agg / projection —
//!    already pattern-matches. Adding a header field would break
//!    every match arm in the codebase for a field most call sites
//!    don't care about.
//! 3. **Per-row freeze**. The visibility map (per-segment
//!    `all_visible` bitmap) is an `&[bool]` slice over headers —
//!    parallel storage makes the bitmap construction zero-copy.
//!
//! ## Backward compatibility
//!
//! Rows that come from a pre-v7.37.15 envelope (V1-V5) have no
//! header on disk. On load, every such row gets a default
//! `RowHeader::frozen()` (`xmin = 1`, `xmax = 0`,
//! `flags = HEAP_XMIN_FROZEN`). Visibility checks against any
//! valid snapshot return `true` — so old data is fully visible to
//! everyone, matching the pre-v7.37.15 contract.

use core::sync::atomic::{AtomicU64, Ordering};

/// Bit 0 of `RowHeader.flags`: `xmin` is conceptually-`FrozenXid` — the row
/// existed at process start (loaded from a V1-V5 envelope) and is
/// unconditionally visible to every snapshot.
pub const HEAP_XMIN_FROZEN: u8 = 1 << 0;
/// Bit 1: the row is the head of a HOT chain (v7.37.15 Phase D).
/// Hot-tier in-place UPDATE optimisation; cold tier never sets this.
pub const HEAP_HOT_UPDATED: u8 = 1 << 1;
/// Bit 2: the row is a HOT chain non-head element (v7.37.15 Phase D).
pub const HEAP_ONLY_TUPLE: u8 = 1 << 2;
/// Bit 3: the row's `xmax` is conceptually-frozen — the delete is
/// older than any live snapshot, so vacuum may reclaim it on the
/// next pass.
pub const HEAP_XMAX_FROZEN: u8 = 1 << 3;

/// Sentinel value used for `xmax` when the row has NOT been deleted.
/// PG uses `InvalidTransactionId = 0`; SPG matches.
pub const XMAX_ALIVE: u64 = 0;

/// Sentinel used for `xmin` on rows loaded from a pre-v7.37.15
/// envelope. Any non-zero value < every real transaction id works
/// — we pick 1 (PG uses `FrozenTransactionId = 2`).
pub const XMIN_FROZEN: u64 = 1;

/// Per-row MVCC visibility header.
///
/// 24 bytes after alignment (8 + 8 + 1 + 7 padding). The padding
/// is intentional: a power-of-two stride keeps array indexing
/// cheap and matches the cache-line layout PG uses for
/// `HeapTupleHeaderData`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(C, align(8))]
pub struct RowHeader {
    /// Version that wrote this row (= `TxId` of the inserting tx
    /// at commit time). Compared against the reader's snapshot
    /// version to decide visibility.
    ///
    /// Default `XMIN_FROZEN = 1` on rows loaded from a pre-
    /// v7.37.15 envelope.
    pub xmin: u64,
    /// Version that deleted / updated this row, or `XMAX_ALIVE = 0`
    /// when still alive. UPDATE writes `xmax` on the old row +
    /// `xmin` on the new one inside the same transaction.
    pub xmax: u64,
    /// Bit-packed flags. See module-level constants. The most
    /// common state (`xmin frozen, xmax alive`) sets only
    /// `HEAP_XMIN_FROZEN` so visibility checks can short-circuit
    /// on the `flags & HEAP_XMIN_FROZEN == HEAP_XMIN_FROZEN &&
    /// xmax == 0` fast path.
    pub flags: u8,
}

impl RowHeader {
    /// The canonical "frozen, alive" header. Use on rows loaded
    /// from a pre-v7.37.15 envelope and on rows inserted before
    /// the writer transaction is assigned a TxId.
    #[must_use]
    pub const fn frozen() -> Self {
        Self {
            xmin: XMIN_FROZEN,
            xmax: XMAX_ALIVE,
            flags: HEAP_XMIN_FROZEN,
        }
    }

    /// A header for a row inserted by transaction `xmin`, still
    /// alive. Default for fresh INSERTs.
    #[must_use]
    pub const fn alive(xmin: u64) -> Self {
        Self {
            xmin,
            xmax: XMAX_ALIVE,
            flags: 0,
        }
    }

    /// True iff this header is the all-visible fast path
    /// (frozen + alive). The visibility-map bit is `true` exactly
    /// when EVERY header in a segment satisfies this — letting
    /// scans skip the per-row check entirely for cold segments.
    #[must_use]
    pub const fn is_all_visible_fast(&self) -> bool {
        self.flags & HEAP_XMIN_FROZEN == HEAP_XMIN_FROZEN && self.xmax == XMAX_ALIVE
    }

    /// Was this row deleted? `false` when `xmax == XMAX_ALIVE`.
    #[must_use]
    pub const fn is_deleted(&self) -> bool {
        self.xmax != XMAX_ALIVE
    }
}

impl Default for RowHeader {
    fn default() -> Self {
        Self::frozen()
    }
}

/// Process-wide monotonic version counter shared by every Database
/// instance in this process. `RowHeader.xmin / xmax` values + the
/// reader's `Snapshot.version` all draw from here.
///
/// `u64` so we never wrap. Starts at `XMIN_FROZEN + 1 = 2` so a
/// fresh transaction can never collide with frozen rows.
///
/// Process-wide (not per-Database) so concurrent databases in the
/// same process share a coherent view of "is tx 17 still alive" —
/// the BitSet `Snapshot.in_progress` keys off the same numbering.
static GLOBAL_VERSION: AtomicU64 = AtomicU64::new(XMIN_FROZEN + 1);

/// Allocate the next transaction id / row version. Caller stores
/// it in the row's `xmin` (for insert) or `xmax` (for delete /
/// update). Threadsafe; no lock involved.
#[must_use]
pub fn next_version() -> u64 {
    GLOBAL_VERSION.fetch_add(1, Ordering::AcqRel)
}

/// Read the current version cursor without advancing. Snapshots
/// use this as their `Snapshot.version`.
#[must_use]
pub fn current_version() -> u64 {
    GLOBAL_VERSION.load(Ordering::Acquire)
}

/// v7.38 — recover the version cursor past a version read off a durable
/// image. `GLOBAL_VERSION` lives in process memory and restarts at
/// `XMIN_FROZEN + 1`, but rows persisted by an earlier process carry the
/// versions *that* process allocated. A fresh process must not hand out a
/// version any restored row already uses, and — because `Snapshot::visible`
/// rejects `xmin > version` as "written by a future transaction" — must take
/// snapshots at a version above every restored `xmin`, or committed rows
/// silently vanish from reads. The same applies to `xmax`: a delete that
/// looks like the future would resurrect the deleted row.
///
/// This is the version-cursor twin of the `next_rowid` recovery in
/// `codec::read_mvcc_header_appendix`, and mirrors PG recovering `nextXid`
/// from `pg_control` rather than restarting the counter at zero.
///
/// `XMAX_ALIVE` (0) carries no version and is ignored.
pub fn observe_persisted_version(v: u64) {
    if v == XMAX_ALIVE {
        return;
    }
    GLOBAL_VERSION.fetch_max(v.saturating_add(1), Ordering::AcqRel);
}

/// v7.37.15 (Phase C.1) — stable per-relation row identity.
///
/// ## Why a stable id, separate from the physical index
///
/// Pre-Phase-C a row was addressed by its **physical index** into
/// `Table::rows`. That index is invalidated the moment a delete /
/// vacuum compacts the survivor vec — every surviving row after the
/// hole shifts down. Physical indices therefore cannot serve as:
///
/// 1. a **row-lock key** (Phase C.4: `(RelId, RowId)` must survive
///    concurrent compaction while a lock is held),
/// 2. a **HOT-chain pointer** (Phase D: chain head → new version
///    must not dangle after vacuum),
/// 3. a **WAL redo identity** (Epic W: `RowChange` UPDATE/DELETE
///    must name the row by a key that survives replay, not a slot
///    that shifted — closing the position-fragility caveat on the
///    `RowChange` doc).
///
/// `RowId` is per-relation, monotonic, and **never reused**. It
/// lives in `Table::rowids: PersistentVec<RowId>` parallel to
/// `rows` / `headers`, so `rowids[i]` is the stable id of the row
/// physically at slot `i`. Compaction rebuilds all three vecs
/// together, so the id travels with the row while the slot shifts.
///
/// Phase C.1 introduces the id additively (allocated + kept
/// lock-step, but indices still address by physical slot); later
/// phases migrate index locators, the lock table, and the WAL to
/// address by `RowId`.
///
/// `u64`, never wraps (same rationale as `xmin`/`xmax`). Starts at
/// 1 per relation; 0 is reserved as an "unassigned" sentinel.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
#[repr(transparent)]
pub struct RowId(pub u64);

impl RowId {
    /// The "unassigned" sentinel. A real appended row never gets 0;
    /// the per-relation allocator starts at 1.
    pub const UNASSIGNED: RowId = RowId(0);
}

/// v7.37.15 (Phase C.1) — stable per-catalog relation identity.
///
/// ## Why a stable id, separate from the table's `Vec` position
///
/// A `Catalog` stores tables in a `Vec<Table>`; a `DROP TABLE`
/// removes one, shifting every later table's position down. So the
/// physical `tables[i]` index cannot key:
///
/// 1. the **row-lock table** — Phase C.4 keys locks by
///    `(RelId, RowId)`; a lock held across a concurrent `DROP TABLE`
///    of an *unrelated* table must keep naming the same relation,
/// 2. the **`RelationStore` shard map** — Phase C.5 splits the
///    single catalog latch into a `DashMap<RelId, _>` of per-relation
///    locks; the key must survive catalog mutation,
/// 3. a **replication relation mapping** — Epic R maps a change to
///    its relation by a stable id, not a shifting slot.
///
/// `RelId` is per-catalog, monotonic, and **never reused** even after
/// the table is dropped, so a stale lock / redo reference is
/// detectable rather than silently aliasing a table that reused the
/// slot. It pairs with [`RowId`] to form the `(RelId, RowId)` tuple
/// identity Phase C.4's lock table needs.
///
/// Introduced additively in Phase C.1: assigned at `CREATE TABLE`
/// and stored on the table, but nothing consumes it yet. `u64`,
/// never wraps; 0 is the `RelId::UNASSIGNED` sentinel, real ids start
/// at 1.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
#[repr(transparent)]
pub struct RelId(pub u64);

impl RelId {
    /// The "unassigned" sentinel. A table created through
    /// `Catalog::create_table` always gets a real id ≥ 1; a bare
    /// `Table::new` (test helpers, interim construction) starts
    /// unassigned until the catalog stamps it.
    pub const UNASSIGNED: RelId = RelId(0);
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn frozen_header_is_all_visible() {
        let h = RowHeader::frozen();
        assert_eq!(h.xmin, XMIN_FROZEN);
        assert_eq!(h.xmax, XMAX_ALIVE);
        assert!(h.is_all_visible_fast());
        assert!(!h.is_deleted());
    }

    #[test]
    fn alive_header_is_not_fast_visible_until_frozen() {
        let h = RowHeader::alive(7);
        assert_eq!(h.xmin, 7);
        assert_eq!(h.xmax, XMAX_ALIVE);
        // Not on the fast path because HEAP_XMIN_FROZEN not set.
        // Visibility against a snapshot is decided by the
        // snapshot-aware path in the engine, not by this fast
        // check.
        assert!(!h.is_all_visible_fast());
        assert!(!h.is_deleted());
    }

    #[test]
    fn version_counter_is_monotonic() {
        let a = next_version();
        let b = next_version();
        let c = next_version();
        assert!(a < b);
        assert!(b < c);
        // Reading does not advance.
        let d = current_version();
        assert!(d >= c);
        let e = current_version();
        assert_eq!(d, e);
    }

    #[test]
    fn version_counter_starts_above_frozen() {
        // First-ever next_version() must return at least
        // XMIN_FROZEN + 1 so a fresh transaction can never collide
        // with a frozen row's xmin.
        let v = current_version();
        assert!(v > XMIN_FROZEN);
    }

    #[test]
    fn deleted_row_header_reports_deletion() {
        let mut h = RowHeader::alive(7);
        assert!(!h.is_deleted());
        h.xmax = 13;
        assert!(h.is_deleted());
    }

    #[test]
    fn observe_persisted_version_advances_cursor_past_restored_rows() {
        // v7.38 — a restored row's version must never look like the future to a
        // snapshot this process takes, or `Snapshot::visible` drops it. The
        // cursor is process-global and monotonic, so observing a large restored
        // version must leave `current_version()` strictly above it.
        let restored_xmin = 1_000_000_007u64;
        observe_persisted_version(restored_xmin);
        assert!(
            current_version() > restored_xmin,
            "cursor must sit above every restored version"
        );
        // Idempotent: observing an older version never rewinds the cursor.
        let after = current_version();
        observe_persisted_version(42);
        assert_eq!(current_version(), after);
        // XMAX_ALIVE is the "not deleted" sentinel, not a version.
        observe_persisted_version(XMAX_ALIVE);
        assert_eq!(current_version(), after);
    }
}