MVCC: row versions and snapshots

// Page reads every line pointer on page 0 of the accounts table with pageinspect. It sees all
// row versions, including ones that no snapshot can see any more.
func Page(ctx context.Context, conn *pgx.Conn) ([]Version, error) {
  rows, err := conn.Query(ctx, queries["heap_page"])
  if err != nil {
    return nil, fmt.Errorf("read heap page: %w", err)
  }
  return pgx.CollectRows(rows, func(r pgx.CollectableRow) (Version, error) {
    var v Version
    var flags, off, mask, mask2 int
    var balance []byte
    if err := r.Scan(&v.LP, &flags, &off, &v.Xmin, &v.Xmax, &v.Ctid, &mask, &mask2,
      &v.XminStatus, &v.XmaxStatus, &balance); err != nil {
      return v, fmt.Errorf("scan line pointer: %w", err)
    }
    v.State = []string{Unused, Normal, Redirect, Dead}[flags]
    if v.State == Redirect {
      v.To = off // a redirect stores the target line pointer in lp_off
    }
    if v.Xmax == "0" {
      v.Xmax = ""
    }
    if mask&heapXminFrozen == heapXminFrozen {
      v.XminStatus = "frozen"
    }
    v.HotUpdated = mask2&heapHotUpdated != 0
    v.HeapOnly = mask2&heapOnlyTuple != 0
    if len(balance) == 4 {
      b := int(int32(binary.LittleEndian.Uint32(balance)))
      v.Balance = &b
    }
    return v, nil
  })
}

// checkScenario asserts the claims the sheet makes about each run.
func checkScenario(t *testing.T, id string, f []mvcc.Frame) {
  t.Helper()
  lp := func(fr mvcc.Frame, n int) mvcc.Version {
    for _, v := range fr.Page {
      if v.LP == n {
        return v
      }
    }
    return mvcc.Version{State: "missing"}
  }
  switch id {
  case "hot":
    upd := f[3]
    if v1, v2 := lp(upd, 1), lp(upd, 2); v1.Xmax == "" || v1.Xmax != v2.Xmin || !v1.HotUpdated || !v2.HeapOnly || v1.Ctid != "(0,2)" {
      t.Errorf("after the update: %+v and %+v, want a HOT chain (0,1) -> (0,2) with xmax = xmin", v1, v2)
    }
    if len(upd.Index) != 1 {
      t.Errorf("a HOT update added an index entry: %v", upd.Index)
    }
    if r := f[4]; r.Views[0].Balance != 100 || r.Views[0].Ctid != "(0,1)" {
      t.Errorf("the reader saw %+v during the update, want 100 from (0,1)", r.Views[0])
    }
    if r := f[6]; r.Views[0].Balance != 100 || r.Views[1].Balance != 150 {
      t.Errorf("after the commit: session 1 saw %d and session 2 saw %d, want 100 and 150", r.Views[0].Balance, r.Views[1].Balance)
    }
    if v := lp(f[7], 1); v.State != mvcc.Normal {
      t.Errorf("VACUUM with session 1 open left line pointer 1 %s, want the old version kept", v.State)
    }
    if v := lp(f[9], 1); v.State != mvcc.Redirect || v.To != 2 {
      t.Errorf("VACUUM after session 1 ended left line pointer 1 as %+v, want a redirect to 2", v)
    }
    if v := lp(f[10], 2); v.XminStatus != "frozen" {
      t.Errorf("VACUUM (FREEZE) left %+v, want a frozen xmin", v)
    }
  case "rollback":
    if v := lp(f[3], 2); v.XminStatus != "aborted" {
      t.Errorf("after ROLLBACK the new version is %+v, want an aborted xmin", v)
    }
    if r := f[4]; r.Views[0].Balance != 100 {
      t.Errorf("after ROLLBACK session 1 saw %d, want 100", r.Views[0].Balance)
    }
    // VACUUM frees the slot. A free slot at the end of the line pointer array is cut off.
    if v := lp(f[5], 2); v.State != mvcc.Unused && v.State != "missing" {
      t.Errorf("VACUUM left the aborted version as %s, want it freed", v.State)
    }
  case "indexed":
    if v := lp(f[0], 1); v.HotUpdated || len(f[0].Index) != 2 {
      t.Errorf("an update of an indexed column: %+v with index %v, want no HOT and 2 entries", v, f[0].Index)
    }
    if len(f[1].Index) != 1 {
      t.Errorf("after VACUUM the index has %v, want 1 entry", f[1].Index)
    }
  }
}

// The claims on the sheet.
if upd.Dead != 100000 || upd.Bytes < load.Bytes*19/10 {
  t.Errorf("one update of every row: %d dead, %d bytes; want 100,000 dead and about twice %d bytes", upd.Dead, upd.Bytes, load.Bytes)
}
if vac.Dead != 0 || vac.Bytes != upd.Bytes {
  t.Errorf("VACUUM: %d dead and %d bytes, want 0 dead and the size unchanged at %d", vac.Dead, vac.Bytes, upd.Bytes)
}
if again := at("update-again"); again.Bytes != vac.Bytes {
  t.Errorf("the second update grew the table from %d to %d bytes; want it to reuse the freed space", vac.Bytes, again.Bytes)
}
if held := at("vacuum-held"); held.Dead != 100000 {
  t.Errorf("VACUUM with an open snapshot left %d dead versions, want all 100,000 kept", held.Dead)
}
if grown := at("update-held-again"); grown.Dead != 200000 || grown.Bytes < at("vacuum-held").Bytes*14/10 {
  t.Errorf("a second update with the snapshot open: %d dead, %d bytes; want 200,000 dead and a bigger table", grown.Dead, grown.Bytes)
}
if free := at("vacuum-free"); free.Dead != 0 {
  t.Errorf("VACUUM after the snapshot closed left %d dead versions, want 0", free.Dead)
}
if full := at("full"); full.Bytes != load.Bytes {
  t.Errorf("VACUUM FULL left %d bytes, want the loaded size %d", full.Bytes, load.Bytes)
}