znippy_plugin_git/

index_layout.rs

//! One git object index, **three Arrow IPC layouts**, measured against each other.
//!
//! ## The question
//!
//! A git `.pack` carries no index at all. Git derives several *separate* files
//! from it — `.idx` (oid → offset), `.rev` (offset → length), `.bitmap`,
//! `.midx` — and two of the four facts a server actually needs on the hot path
//! are in **none** of them:
//!
//! | fact | where git keeps it |
//! |---|---|
//! | oid | `.idx` |
//! | byte extent (offset, len) | offset in `.idx`, length only via `.rev` or the next offset |
//! | **type** | nowhere — an entry-header parse per object |
//! | **uncompressed size** | nowhere — the same entry-header parse |
//! | **delta base** | nowhere — the same entry-header parse, plus a varint |
//!
//! ## `delta_base` is an OFFSET, and the arms are named for what they were
//!
//! PLAN §13 settles it: `delta_base` is a `u64` **archive offset**, not a row
//! ordinal into this table. An ordinal would be half the width and would index
//! the very table it sits in — and it would break **silently** the first time
//! the table is rebuilt in a different order, which §14 guarantees will happen
//! because the whole index is a droppable cache. An offset is order-independent
//! because the pack it addresses never moves; `PackWalk::rebased` shifts it with
//! one addition and no table to keep in step. `0` is the sentinel for *no base*:
//! archive offset 0 is a pack's `PACK` magic, so no entry can start there.
//!
//! It joined the schema after the three arms were named and measured, so the
//! names now count **sections and tables, which is the variable under test** —
//! not facts. [`FourTables`] is four IPC sections, [`OneTableFourColumns`] is
//! one; both carry **five** payload columns today, and [`PackedPayload`] still
//! carries one. Every ns figure quoted below was taken on the four-fact,
//! 25-byte schema and is marked where that matters.
//!
//! So `cat-file --batch-check` over a pack is a varint decode per object, and a
//! `have` negotiation of 1000 oids is 1000 of them. Consolidating all four facts
//! into one index is the point of this module. The open question is what
//! *shape* that index should have, and this module answers it by building three
//! shapes behind [`ObjectIndex`] and benchmarking them
//! (`examples/index_layout_bench.rs`).
//!
//! | arm | payload columns | buffers a full-row fetch touches |
//! |---|---:|---:|
//! | [`FourTables`] | 5, in four IPC sections | 5 |
//! | [`OneTableFourColumns`] | 5, in one IPC section | 5 |
//! | [`PackedPayload`] | **1**, in one IPC section | **1** |
//!
//! The first two were the original experiment, and they **tied at every size,
//! batch size and hit rate**. The reason is visible in the table: Arrow is
//! columnar, so "one table with five columns" is *not* row-contiguous. Both
//! arms keep `offset`, `len`, `object_type`, `uncompressed_size` and
//! `delta_base` in five separate buffers whose bases are megabytes apart
//! (measured: 1 737 664 bytes between the `offset` and `size` buffers at
//! 100 000 rows), so a full-row fetch costs five unrelated strides either way.
//! **The number of tables was never the variable. The number of columns is** —
//! which is what the third arm changes, and it is the only one that moves the
//! number.
//!
//! ## What is deliberately shared, so the measurement is of the layout
//!
//! All three arms resolve `oid → ordinal` through the **same** [`crate::oid_index`]
//! `stree` (LAW 5: reuse, do not twin — and a third oid index written by
//! accident would have made the arms incomparable anyway). The two columnar arms
//! share one `ColumnarPayload` holding every read path, so they cannot drift
//! apart in the hot loop and the only thing between them is IPC framing.
//!
//! ## Zero-copy is asserted, not assumed
//!
//! All three arms decode with `StreamDecoder::with_require_alignment(true)`,
//! which makes arrow **error** rather than silently allocate-and-copy when a
//! buffer is misaligned. A layout that quietly copied its columns out of the IPC
//! bytes on open would be a different (and much slower to build) thing than the
//! one being claimed, so the reader refuses to be that thing.
//! [`ObjectIndex::ipc_bytes`] plus each arm's `column_is_inside_ipc` let a test
//! prove each array's data pointer actually lies inside the IPC buffer.
//!
//! # Measured
//!
//! oden, 32 cores, 2026-08-07, `--release --no-default-features`, sha1 oids,
//! 100 000 lookups per cell, 5 runs per cell, 4 sizes × 3 hit rates × 7 access
//! paths + 2 column scans. `/proc/loadavg` 0.96–2.98 (1-min) across the three
//! sweeps — the first rotation started at 2.98, the other two at 1.53 and 1.35,
//! and the reported figures are the geometric mean of all three, so no arm sat
//! disproportionately in the busier one.
//!
//! Position is cancelled by **rotation**: the sweep is run three times with the
//! arm order rotated, so each arm occupies each position exactly once. This
//! matters — in the earlier two-arm work the arm timed second was measurably
//! 1.1% slower at N ≥ 1e6 whichever layout it was, which is enough to invent a
//! result out of nothing.
//!
//! **Noise band: median 8.9% run-to-run spread, p90 32.3%** over 270 cells
//! (excluding the 1e3 scans, whose ns-per-row is below timer resolution). The
//! p90 is carried almost entirely by the 100 000-object rows, where the index
//! fits in L3; at 1e3 and 4e6 the bands are 2–20%. **Nothing below is claimed
//! unless it clears its own cell's band.**
//!
//! ## First: the harness really does read all four facts
//!
//! This has to be established before any of the rest means anything. A harness
//! that resolved the oid and threw the row away would measure the `stree` and
//! nothing else, and would produce a tie no matter what the layouts did.
//!
//! [`ObjectIndex::ordinals_batch`] exists for exactly this: it resolves
//! `oid → ordinal` and stops, touching no payload byte, and it is the same code
//! in all three arms. Subtracting it from the full-row path isolates the payload
//! gather — the only part any of these layouts can change:
//!
//! | objects | mix | stree floor | A payload | B payload | **C payload** | C/B |
//! |---:|---|---:|---:|---:|---:|---:|
//! | 1e3 | 100% hit | 50.3 | 6.3 | 6.1 | **1.5** | 0.24 |
//! | 1e3 | 10% hit | 39.1 | 5.3 | 5.1 | **1.8** | 0.34 |
//! | 1e5 | 50% hit | 58.6 | 15.0 | 14.0 | **10.6** | 0.76 |
//! | 1e6 | 100% hit | 195.8 | 45.0 | 46.2 | **27.7** | 0.60 |
//! | 1e6 | 10% hit | 113.4 | 11.4 | 11.1 | **7.2** | 0.65 |
//! | 4e6 | 100% hit | 247.2 | 39.2 | 43.1 | **29.8** | 0.69 |
//! | 4e6 | 50% hit | 199.6 | 30.9 | 27.9 | **20.9** | 0.75 |
//! | 4e6 | 10% hit | 141.3 | 25.1 | 21.7 | **17.7** | 0.81 |
//!
//! ns per lookup, batch 1000, position-cancelled. The payload column is nowhere
//! near zero and it differs between arms, so the facts are being read. It is
//! **9–28% of a full-row lookup**; the shared `stree` is the other 72–91%.
//!
//! ## The owner was right: one column IS faster to fetch than four
//!
//! On the part of the work the layout controls, `PackedPayload` cuts the payload
//! gather by **19–76%**, and by **25–40% at every size from 1e6 up**. One
//! 25-byte stride against four strides megabytes apart, exactly as predicted.
//! The single cell that does not show it (1e5, 100% hit, C/B 1.06) sits in the
//! noisiest regime in the sweep, band 44%.
//!
//! Those cells are the **four-fact** schema. `delta_base` makes the packed
//! record 33 bytes and gives the columnar arms a fifth stride, so the mechanism
//! points the same way and the magnitudes are stale until re-measured. See
//! "What the fifth column costs" below for the measurement that was actually
//! taken after the column landed.
//!
//! **But the win is capped by Amdahl**, and this is the honest headline: the
//! payload gather is only 9–28% of a lookup, so a 31% cut in it is a **4–8% cut
//! end to end**. At 4e6 objects, 100% hit, position-cancelled `C/B`:
//!
//! | path | A ns | B ns | C ns | C/B | band | |
//! |---|---:|---:|---:|---:|---:|---|
//! | ordinals only (the floor) | 247.2 | 245.1 | 244.6 | 0.998 | 4.9% | tie, as it must be |
//! | full row, serial `lookup` | 464.8 | 471.8 | 435.2 | **0.922** | 7.7% | **C faster** |
//! | full row, batch 1 | 824.2 | 811.9 | 757.4 | 0.933 | 11.4% | inside band |
//! | full row, batch 100 | 298.6 | 299.0 | 283.4 | **0.948** | 5.2% | **C faster** |
//! | full row, batch 1000 | 286.4 | 288.3 | 274.4 | **0.952** | 2.8% | **C faster** |
//! | full row, batch 10000 | 284.4 | 283.3 | 271.5 | 0.959 | 4.8% | inside band |
//! | extents only, batch 1000 | 261.5 | 263.0 | 264.2 | 1.005 | 4.0% | tie |
//!
//! Three of the four full-row cells clear their band; the direction is the same
//! in all seven and at every size. At 1e3 the win is larger and cleaner (C/B
//! 0.898–0.931 at batch ≥ 100, bands 1.8–2.8%) because there the `stree` floor
//! is small enough not to swamp it.
//!
//! **`extents` is a tie, and that is the mechanism confirming itself.** A
//! partial-row fetch of two facts lets the columnar arms read two columns
//! instead of four, while the packed arm reads the same 25 bytes it always
//! reads. The advantage is exactly proportional to how much of the row you want,
//! and at half a row it is gone.
//!
//! ## What the fifth column costs
//!
//! **Footprint: exactly 8 bytes per object, in every arm.** This is arithmetic,
//! not an estimate, and the byte-exact guards below hold it there: the packed
//! record goes 25 → 33 B, and each columnar arm gains one `u64` buffer of `8n`
//! bytes. 800 kB per 100 000 objects; 32 MB at 4e6. The only *relative* change
//! between arms is that `PackedPayload` now saves four per-column IPC buffers
//! instead of three — MEASURED 50 560 B at 100 000 rows against the 38 016 B it
//! saved before (`the_packed_arm_holds_the_same_payload_but_a_smaller_section`,
//! which asserts the scaling law rather than the constant).
//!
//! **Latency: NOT re-measured, and no claim is made.** Two reasons, and the
//! second is the one that will still be true tomorrow.
//!
//! Load was the first: the harness refuses above 1-minute load 4.0 and oden was
//! carrying another agent's build at 9–26 for most of this change. That one
//! cleared on its own (it fell to 2.1), so it is not the reason this is still
//! unmeasured.
//!
//! **A before/after of a znippy change cannot be built in this tree, and that is
//! structural.** A before/after needs the old and the new source resolved at the
//! same time, which needs two znippy checkouts, and there is nowhere to put the
//! second one — MEASURED by trying all three placements:
//!
//! | where the second checkout goes | what happens |
//! |---|---|
//! | anywhere under `~/git` | `package collision in the lockfile: znippy-common v0.9.13 (…/znippy-pre-deltabase) and znippy-common v0.9.13 (…/znippy) are different` |
//! | anywhere outside it | `failed to read …/znippy-zoomies/lbzip2/Cargo.toml` |
//!
//! The cycle is the cause, not the placement: znippy's `xtask` and `tests`
//! depend on `../../nornir`, and `nornir` depends back on
//! `../znippy/znippy-common` — an absolute reference to *the* canonical
//! checkout. So a second checkout drags the canonical one into its own resolve
//! and two different copies of one version land in one lockfile. Move it out of
//! `~/git` to escape that and it loses the `../../znippy-zoomies/…`
//! path dependencies instead. Symlinking the siblings back does not help: cargo
//! canonicalises, so `nornir` resolves to the real one and points at the real
//! znippy again.
//!
//! Anyone wanting this number should therefore plan on **one checkout, timed
//! twice** — build the bench, `git checkout` the other revision of these files
//! in place, build again — and should not spend the hour discovering the above.
//! That mutates a shared checkout for as long as it takes to compile, which is
//! why it was not done here while another agent was working in this crate.
//!
//! What can be said without measuring, and is deliberately weaker than a
//! number: the payload gather is 9–28% of a lookup, a columnar full-row fetch
//! goes from four strides to five, and the packed record goes from ~1.4 cache
//! lines to ~1.5. Both arms get slightly worse and the packed arm's *relative*
//! advantage on full-row fetches should narrow a little. An 8-byte-per-row
//! effect is in any case inside the 8.9% median / 32.3% p90 band this sweep
//! already reports, so the honest expectation is a null result rather than a
//! regression. **Re-run `examples/index_layout_bench.rs` on a quiet box before
//! quoting any of the ns tables above as current.**
//!
//! ## …and it loses the scans, by a lot
//!
//! ns per row, full column scan, position-cancelled:
//!
//! | objects | scan | A | B | **C** | C/B |
//! |---:|---|---:|---:|---:|---:|
//! | 4e6 | `sum_uncompressed` (uses 8 B of every row) | 0.36 | 0.36 | **1.19** | **3.3×** |
//! | 4e6 | `count_type` (uses 1 B of every row) | 0.09 | 0.09 | **1.32** | **14.4×** |
//! | 1e6 | `count_type` | 0.09 | 0.09 | **1.27** | **14.0×** |
//! | 1e5 | `count_type` | 0.09 | 0.09 | **0.51** | **5.5×** |
//!
//! The 4e6 rows clear their bands (4.5% and 34.7%) with room to spare, and the
//! direction is identical at every size. This is the trade stated in
//! [`PackedPayload`]'s docs, landing exactly where predicted: `count_type` drags
//! 25 bytes through cache to use one, so it pays ~14× for the privilege of the
//! 5% it won on full-row fetches. A quota gate or a type histogram over a
//! 4-million-object repository is 5 ms on the columnar arms and 5 ms × 3–14 on
//! this one.
//!
//! ## The other figures
//!
//! | | FourTables | OneTableFourColumns | PackedPayload |
//! |---|---:|---:|---:|
//! | resident (IPC + stree), 4e6 | 326.635 MiB | 326.634 MiB | 325.203 MiB |
//! | build, 4e6 objects | 3.956 s | 4.052 s | 3.975 s |
//!
//! * **A vs B is still a tie**, now confirmed with a harness that provably reads
//!   the payload: `A/B` ranges 0.96–1.06 and every cell is inside its band. Four
//!   sections cost exactly **1048 bytes more than one, constant at every size** —
//!   the framing of three extra IPC streams, 0.0003% at 4e6.
//! * **`PackedPayload` is the smallest on disk**, by three per-column IPC
//!   buffers: 38 016 B at 100 000 rows, 75 456 B at 200 000, i.e. `3 · n/8`. Its
//!   *payload* is byte-identical (25 = 8 + 8 + 1 + 8).
//! * **Build is a tie across all three** — within 2.5% at 4e6, inside the band.
//!   Build is the oid sort and the `stree`, not the framing or the packing.
//!
//! ## What this means for the shape of the index
//!
//! Neither the packed arm nor the columnar ones is right everywhere, and the
//! measurement says which is which rather than leaving it to taste:
//!
//! * The negotiation path (`want`/`have`, full row) is **5% faster packed**, and
//!   the ceiling on that is the `stree`, not the payload — the next real win on
//!   that path is in the oid step, not here.
//! * The wire path (`extents`) is **indifferent**.
//! * Analytics (quota gates, type histograms) are **3–14× slower packed**.
//! * The packed arm gives up being four typed Arrow columns. A
//!   `FixedSizeBinary(25)` is opaque to DuckDB / Polars / DataFusion, which can
//!   query the other two arms straight off the IPC bytes with no consumer code.
//!   That is a real cost and it is not a performance one.
//!
//! ## The stree header was the next experiment, and it was a negative result
//!
//! What this sweep *can* say, which the two-arm version could not, is that
//! **the oid step is where the remaining time is**: the `ordinals_batch` floor is
//! 72–91% of every full-row lookup, and it is identical in all three arms. The
//! floor itself grows 39 ns → 247 ns from 1e3 to 4e6 objects, which is cache and
//! TLB behaviour in the `stree`, not in any payload column.
//!
//! That pointed at `oid_index`'s 24-byte header (24 mod 64 = 24, so the key
//! array is off-line by construction). It has now been built and measured —
//! [`crate::oid_index::OidLayout`] and `examples/oid_align_bench.rs`, three arms
//! separating the header offset from the allocator. **A 64-aligned keyspace
//! takes 5.2% off cache-misses, reproducibly, and does not move the clock at
//! all**: 0 of 24 cells clear their band, and 0 of 6 clear it again at 4e6 with
//! the band tightened to 5%. The header is not where the oid step's time is, and
//! the reason is instructive — `stree`'s internal nodes live in the tree's own
//! `Vec`, not in this section, so the header could only ever move the leaf
//! touch, and the pipelined batch walk was already hiding it. The full numbers
//! and the mechanism are in [`crate::oid_index`].

use std::sync::Arc;

use anyhow::{Result, anyhow, bail};
use znippy_common::arrow::array::{
    Array, FixedSizeBinaryArray, UInt8Array, UInt64Array,
};
use znippy_common::arrow::buffer::{Buffer, MutableBuffer, ScalarBuffer};
use znippy_common::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use znippy_common::arrow::ipc::reader::StreamDecoder;
use znippy_common::arrow::ipc::writer::StreamWriter;
use znippy_common::arrow::record_batch::RecordBatch;

use crate::object::GitHashKind;
use crate::oid_index::{GitOidIndex, OidEntry, OidLayout, build_section_with_layout};

// ── the four facts ────────────────────────────────────────────────────────────

/// Object type, in git's own **pack entry** encoding.
///
/// This is not [`crate::object::GitObjectKind`] and must not be merged with it:
/// that enum is the four *loose* object types, which is all a canonical
/// `"<type> <size>\0…"` header can express. A packed object also comes in the
/// two delta forms. Owned by the `git-storage-trait` contract; re-exported here
/// so `crate::index_layout::ObjType` stays a valid path.
pub use git_storage_trait::ObjType;

/// One object, as the caller hands it to [`ObjectIndex::build`]. Order is
/// irrelevant — every implementation sorts by oid and derives ordinals itself.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IndexEntry {
    /// 20 bytes (sha1) or 32 (sha256). All entries in one index must agree.
    pub oid: Vec<u8>,
    /// Start of the object's bytes within the archive.
    pub offset: u64,
    /// Length of those bytes as stored (compressed / delta-encoded).
    pub len: u64,
    pub obj_type: ObjType,
    /// Inflated, post-delta-resolution size. The fact `.idx` and `.rev`
    /// together still cannot answer.
    pub uncompressed_size: u64,
    /// `objects.delta_base` — the **archive offset** of the entry this one
    /// deltas against, `0` for none (PLAN §13).
    ///
    /// An offset and not an ordinal, on purpose: it is in the same coordinate
    /// space as [`offset`](Self::offset), so it survives a rebuild of this table
    /// in any order and a rebase of the pack is one addition
    /// ([`crate::pack_walk::PackWalk::rebased`]). `0` cannot collide with a real
    /// base because archive offset 0 is a pack's `PACK` magic. A `REF_DELTA`
    /// records `0` until its oid is resolved to an offset —
    /// [`obj_type`](Self::obj_type) is what distinguishes that from *no base*.
    pub delta_base: u64,
}

/// What a lookup resolves to: the ordinal plus all five facts.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct IndexRow {
    /// Row ordinal — the oid-lexicographic rank, and the join key `FourTables`
    /// uses across its four sections.
    pub ordinal: u32,
    pub offset: u64,
    pub len: u64,
    pub obj_type: ObjType,
    pub uncompressed_size: u64,
    /// See [`IndexEntry::delta_base`]. An archive offset, `0` for none — never
    /// an ordinal, and in particular never *this* row's ordinal space.
    pub delta_base: u64,
}

// ── the trait ─────────────────────────────────────────────────────────────────

/// A git object index over Apache Arrow IPC.
///
/// `build` is `where Self: Sized`, so it stays out of the vtable and
/// `Box<dyn ObjectIndex>` still works — which is how the bench drives both arms
/// through one loop.
pub trait ObjectIndex: Send + Sync {
    fn build(entries: &[IndexEntry]) -> Result<Self>
    where
        Self: Sized;

    /// Resolve one oid. `None` for an absent oid **and** for an oid of the
    /// wrong width.
    fn lookup(&self, oid: &[u8]) -> Option<IndexRow>;

    /// The batch path, and the one that matters: `have` negotiation sends up to
    /// 1000 oids at a time, the push connectivity check sends thousands.
    /// Positional — `out[i]` answers `oids[i]`.
    ///
    /// This is the **full-row** access pattern: all five facts of each object.
    fn lookup_batch(&self, oids: &[&[u8]]) -> Vec<Option<IndexRow>>;

    /// oid → ordinal and **stop**. No payload column is touched at all.
    ///
    /// Identical in every implementation, because every implementation shares
    /// the same `stree`. It is here as the **floor**: whatever this costs is
    /// what no payload layout can remove, and `lookup_batch` minus this is the
    /// only part any of these layouts can change. Without it a three-way tie is
    /// unreadable — it could mean the layouts are equivalent, or it could mean
    /// the harness never read a payload byte.
    fn ordinals_batch(&self, oids: &[&[u8]]) -> Vec<Option<u32>>;

    /// The **partial-row** access pattern: byte extent only, two of the five
    /// facts. This is `extents(&[oid])`, what the wire path actually asks for
    /// when it is about to copy bytes out of a pack.
    fn extents_batch(&self, oids: &[&[u8]]) -> Vec<Option<(u64, u64)>>;

    /// Full **column scan**: total uncompressed bytes over every object. The
    /// quota gate. Reads one 8-byte fact per row and nothing else.
    fn sum_uncompressed(&self) -> u64;

    /// Full **column scan** with a predicate: how many objects of this type.
    /// Reads one *byte* per row and nothing else — the most column-shaped
    /// query there is, and the one a row-packed layout should lose worst.
    fn count_type(&self, t: ObjType) -> usize;

    fn name(&self) -> &'static str;

    fn len(&self) -> usize;

    fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Total bytes of Arrow IPC this index holds resident. Four sections or
    /// one, this counts the same payload, so it is comparable across arms.
    fn ipc_bytes(&self) -> usize;

    /// IPC bytes plus the `stree` oid section — everything the index keeps
    /// alive, excluding the handful of `Arc`'d schema/metadata allocations.
    fn resident_bytes(&self) -> usize;

    // ── the durable form ─────────────────────────────────────────────────────
    //
    // **The projection is the durable index, and redb is its tail** — the
    // pattern the read stack was named for and, until 2026-09-04, did not
    // have. `ObjectReadStack` used to keep EVERY row in redb for ever and
    // re-derive this projection from a full B-tree scan on every open (a
    // 382 B/object RAM floor, measured 744–966 on the kernel mirror), while
    // the redb file grew to 208.5 B/object for 61 B of content — 2.9 GB
    // written, 4 GiB allocated, for 14 M objects. The three methods below are
    // what lets the stack persist the projection instead, fold the redb rows
    // into it, and empty redb of what it folded — so redb holds only what
    // arrived since the last fold, which is the whole of what a B-tree is
    // good for here.

    /// Whether this arm has a single-buffer form the stack can write down and
    /// read back with [`from_persisted`](Self::from_persisted). An arm that
    /// answers `false` is never persisted: the stack then keeps every row in
    /// redb and rebuilds from it, which is the pre-2026-09-04 behaviour, kept
    /// as a correct fallback rather than made an error.
    ///
    /// A `Sized` fn and not an associated const: a const would make the trait
    /// not dyn-compatible, and the layout benches hold `&dyn ObjectIndex`.
    fn persists() -> bool
    where
        Self: Sized,
    {
        false
    }

    /// **The bytes this index is persisted as** — one self-contained Arrow IPC
    /// stream holding one record batch, or `None` for an arm with no such form.
    fn persisted_bytes(&self) -> Option<&[u8]> {
        None
    }

    /// Reconstruct this index from a buffer [`persisted_bytes`](Self::persisted_bytes)
    /// produced — typically a memory-mapped `objects.index`, so every array is a
    /// view into the page cache rather than a copy of it. Refuses a buffer this
    /// arm did not write (a different arm's layout, a different format stamp).
    fn from_persisted(buf: Buffer) -> Result<Self>
    where
        Self: Sized,
    {
        let _ = buf;
        bail!("this index layout has no persisted form")
    }

    /// The raw oid at `ordinal`. Every arm keeps a `FixedSizeBinary` oid
    /// column, so this is one bounds-checked slice, copied out — owned rather
    /// than borrowed so that a stack holding its projection behind a lock can
    /// answer it too. Positional, like Arrow's own `value(i)`: an ordinal past
    /// `len()` (or, on a stack, a tail-tagged one) is a programming error and
    /// panics rather than answering.
    fn oid_at(&self, ordinal: u32) -> Vec<u8>;

    /// Every fact of the row at `ordinal`, as the entry that would rebuild it.
    /// The fold's merge reads the projection back through this.
    fn entry_at(&self, ordinal: u32) -> IndexEntry;

    /// `(offset, len)` of the row at `ordinal` — positional, no oid, no
    /// allocation. The crash-recovery diff scans the whole projection through
    /// this, so it must cost a column read and nothing more.
    fn extent_at(&self, ordinal: u32) -> (u64, u64);

    /// **Every oid, back to back, in ordinal order** — `len() × oid_len` bytes.
    /// This IS the ordinal space the derived tables address (`__gunnar_reach__`
    /// bitmaps, the commit graph's token), and on a persisted arm it is the oid
    /// column's own bytes: the thing `objects.ordinal` used to cache in a file
    /// of its own, with a header and an epoch to validate it, is a contiguous
    /// buffer this index already holds. The default is the positional loop; the
    /// columnar arms hand back their column.
    fn oids_raw(&self) -> Vec<u8> {
        let n = self.len() as u32;
        let mut out = Vec::new();
        for o in 0..n {
            out.extend_from_slice(&self.oid_at(o));
        }
        out
    }

    /// All rows, in ordinal (= oid) order. `len()` allocations of `oid`, which
    /// is the price of handing `build` a slice; the fold pays it once per fold
    /// and nothing on the read path calls it.
    fn entries(&self) -> Vec<IndexEntry> {
        (0..self.len() as u32).map(|o| self.entry_at(o)).collect()
    }
}

/// Schema-metadata key under which a persisted index stamps
/// [`PERSISTED_FORMAT`]. A buffer without it, or with another value, is not
/// this crate's index and [`ObjectIndex::from_persisted`] refuses it.
pub const PERSISTED_FORMAT_KEY: &str = "znippy.index.format";
/// Schema-metadata key naming the arm that wrote the buffer, so one arm cannot
/// misread another's bytes as its own.
pub const PERSISTED_LAYOUT_KEY: &str = "znippy.index.layout";
/// **Bump in the same commit as any change to what a persisted column means.**
/// The Arrow schema already versions the column *set* — a renamed or missing
/// column fails `col()` — so this exists for the change a schema cannot see: the
/// same six names carrying different semantics. `1` is the layout in use when
/// persistence was introduced (2026-09-04).
pub const PERSISTED_FORMAT: u64 = 1;

/// The schema metadata every persisted arm writes.
fn persisted_metadata(layout: &str) -> std::collections::HashMap<String, String> {
    [
        (PERSISTED_FORMAT_KEY.to_string(), PERSISTED_FORMAT.to_string()),
        (PERSISTED_LAYOUT_KEY.to_string(), layout.to_string()),
    ]
    .into_iter()
    .collect()
}

/// Refuse a decoded batch whose schema was not stamped by `layout` at
/// [`PERSISTED_FORMAT`]. The error names what was found, because "wrong
/// format" with no value behind it sends the reader to a hex dump.
fn check_persisted_stamp(batch: &RecordBatch, layout: &str) -> Result<()> {
    let meta = batch.schema_ref().metadata();
    match meta.get(PERSISTED_FORMAT_KEY).map(String::as_str) {
        Some(v) if v == PERSISTED_FORMAT.to_string() => {}
        other => bail!(
            "persisted index carries {PERSISTED_FORMAT_KEY}={other:?}, this build reads \
             {PERSISTED_FORMAT}"
        ),
    }
    match meta.get(PERSISTED_LAYOUT_KEY).map(String::as_str) {
        Some(v) if v == layout => Ok(()),
        other => bail!(
            "persisted index was written by layout {other:?}, this projection is {layout}"
        ),
    }
}

// ── shared machinery: both arms are built from exactly this ───────────────────

/// Column names, one place, so the two arms cannot drift apart on spelling.
pub const COL_OID: &str = "oid";
pub const COL_OFFSET: &str = "offset";
pub const COL_LEN: &str = "len";
pub const COL_TYPE: &str = "object_type";
pub const COL_SIZE: &str = "uncompressed_size";
pub const COL_DELTA_BASE: &str = "delta_base";

/// The six Arrow arrays that carry the five facts, in oid-lexicographic order.
/// The byte extent is one fact in two columns because that is what it is —
/// `(offset, len)` — and splitting it lets a scan of just the offsets stay
/// contiguous.
pub struct Columns {
    pub oid: FixedSizeBinaryArray,
    pub offset: UInt64Array,
    pub len: UInt64Array,
    pub obj_type: UInt8Array,
    pub size: UInt64Array,
    /// Archive offset of the base entry, `0` for none. Same coordinate space as
    /// `offset`, which is what makes a whole-column comparison against the
    /// `offset` column meaningful (a delta's base is an entry in this table).
    pub delta_base: UInt64Array,
}

/// Sort indices of `entries` into oid-lexicographic order, which is also the
/// ordinal order and the order `oid_index::build_section` puts its keys in.
///
/// Ties on the full oid are impossible in a well-formed index and are checked
/// for in [`validate`] rather than being silently deduplicated.
fn oid_order(entries: &[IndexEntry]) -> Vec<u32> {
    let mut order: Vec<u32> = (0..entries.len() as u32).collect();
    order.sort_unstable_by(|&a, &b| entries[a as usize].oid.cmp(&entries[b as usize].oid));
    order
}

/// Reject the two things that would make an index silently wrong: mixed oid
/// widths, and a duplicate oid (which would give one object two ordinals and
/// make the two arms disagree on which one a lookup returns).
fn validate(entries: &[IndexEntry], order: &[u32]) -> Result<GitHashKind> {
    let Some(first) = entries.first() else {
        // An empty index is legitimate (an empty push) and its hash kind is
        // arbitrary; sha1 keeps the section 20-byte shaped.
        return Ok(GitHashKind::Sha1);
    };
    let hash = match first.oid.len() {
        20 => GitHashKind::Sha1,
        32 => GitHashKind::Sha256,
        n => bail!("oid width {n} is neither sha1 (20) nor sha256 (32)"),
    };
    for e in entries {
        if e.oid.len() != hash.oid_len() {
            bail!(
                "mixed oid widths: {} and {} in one index",
                hash.oid_len(),
                e.oid.len()
            );
        }
    }
    for w in order.windows(2) {
        if entries[w[0] as usize].oid == entries[w[1] as usize].oid {
            bail!(
                "duplicate oid {} — one object cannot hold two ordinals",
                hex::encode(&entries[w[0] as usize].oid)
            );
        }
    }
    Ok(hash)
}

/// The `stree` oid keyspace, shared verbatim by both arms.
///
/// `lookup_row` is set to the ordinal here. That is not a redefinition of the
/// field: [`crate::oid_index::build_section`] is the lower-level API and its
/// contract lets the caller choose what row an oid points at. This index's rows
/// *are* its ordinals, and both arms use the ordinal, so nothing is lost.
pub struct OidResolver {
    tree: GitOidIndex,
    section_bytes: usize,
}

impl OidResolver {
    fn build(entries: &[IndexEntry], order: &[u32], hash: GitHashKind) -> Result<Self> {
        Self::build_with(entries, order, hash, OidLayout::default())
    }

    /// [`build`](Self::build) with the section's cache-line layout chosen by the
    /// caller. Only `PackedPayload::build_with_oid_layout` uses it; see there.
    fn build_with(
        entries: &[IndexEntry],
        order: &[u32],
        hash: GitHashKind,
        layout: OidLayout,
    ) -> Result<Self> {
        let oid_entries: Vec<OidEntry> = order
            .iter()
            .enumerate()
            .map(|(rank, &i)| OidEntry {
                oid: entries[i as usize].oid.clone(),
                lookup_row: rank as u64,
                ordinal: rank as u32,
            })
            .collect();
        let section = build_section_with_layout(&oid_entries, hash, layout)?;
        let section_bytes = section.len();
        // `parse_as`, not `parse`: `Compact64Alloc` is a reader-side allocation
        // choice that the section's bytes cannot express.
        Ok(Self { tree: GitOidIndex::parse_as(section, layout)?, section_bytes })
    }

    /// The resolver over an already-sorted oid column — a persisted index being
    /// read back. Row `i` of the column IS ordinal `i`, which is what
    /// [`from_persisted`](ObjectIndex::from_persisted) relies on and what
    /// `build` guaranteed when it wrote the column in `oid_order`.
    fn from_oid_column(oid: &FixedSizeBinaryArray, hash: GitHashKind) -> Result<Self> {
        let oid_entries: Vec<OidEntry> = (0..oid.len())
            .map(|i| OidEntry { oid: oid.value(i).to_vec(), lookup_row: i as u64, ordinal: i as u32 })
            .collect();
        let layout = OidLayout::default();
        let section = build_section_with_layout(&oid_entries, hash, layout)?;
        let section_bytes = section.len();
        Ok(Self { tree: GitOidIndex::parse_as(section, layout)?, section_bytes })
    }

    #[inline]
    fn ordinal(&self, oid: &[u8]) -> Option<u32> {
        self.tree.lookup(oid).map(|h| h.ordinal)
    }

    #[inline]
    fn ordinals(&self, oids: &[&[u8]]) -> Vec<Option<u32>> {
        self.tree
            .lookup_batch(oids)
            .into_iter()
            .map(|h| h.map(|h| h.ordinal))
            .collect()
    }
}

/// Materialise the five arrays from `entries` in `order`.
///
/// Both arms call this and nothing else, so any difference the bench reports is
/// a difference of IPC packing, not of array construction.
pub fn columns(entries: &[IndexEntry], order: &[u32], oid_len: usize) -> Columns {
    let n = order.len();
    let mut offset: Vec<u64> = Vec::with_capacity(n);
    let mut len: Vec<u64> = Vec::with_capacity(n);
    let mut ty: Vec<u8> = Vec::with_capacity(n);
    let mut size: Vec<u64> = Vec::with_capacity(n);
    let mut delta_base: Vec<u64> = Vec::with_capacity(n);
    for &i in order {
        let e = &entries[i as usize];
        offset.push(e.offset);
        len.push(e.len);
        ty.push(e.obj_type.code());
        size.push(e.uncompressed_size);
        delta_base.push(e.delta_base);
    }
    Columns {
        oid: oid_column(entries, order, oid_len),
        offset: UInt64Array::new(ScalarBuffer::from(offset), None),
        len: UInt64Array::new(ScalarBuffer::from(len), None),
        obj_type: UInt8Array::new(ScalarBuffer::from(ty), None),
        size: UInt64Array::new(ScalarBuffer::from(size), None),
        delta_base: UInt64Array::new(ScalarBuffer::from(delta_base), None),
    }
}

/// The oid column alone. Every arm stores it, and no arm packs it into the
/// payload: the oid is the *key* the stree resolves against, not one of the
/// facts a resolved lookup fetches, so putting it next to the payload would
/// widen every payload read for a field nobody reads at lookup time.
pub fn oid_column(entries: &[IndexEntry], order: &[u32], oid_len: usize) -> FixedSizeBinaryArray {
    let mut bytes: Vec<u8> = Vec::with_capacity(order.len() * oid_len);
    for &i in order {
        bytes.extend_from_slice(&entries[i as usize].oid);
    }
    FixedSizeBinaryArray::new(oid_len as i32, Buffer::from_vec(bytes), None)
}

/// The five facts of one object, packed little-endian into `PACKED_LEN` bytes.
///
/// ```text
///   0  u64 LE  offset               8
///   8  u64 LE  len                  8
///  16  u8      object type code     1
///  17  u64 LE  uncompressed_size    8
///  25  u64 LE  delta_base           8
///                                  ══
///                                  33
/// ```
///
/// Fixed order, fixed width, no padding, no nulls. The type byte sits between
/// the extent and the sizes rather than after them so the record is 33 bytes
/// instead of 40 — the whole point of the arm is how few bytes a full-row fetch
/// drags through cache, and 33 against 40 is 18% of that budget.
///
/// **This was 25 bytes before `delta_base` (PLAN §13) joined the schema.** The
/// column is appended rather than inserted so the first four fields keep the
/// offsets every reader here and every hand-decode in the tests already uses;
/// the width is asserted byte for byte by
/// `the_packed_record_is_the_documented_33_bytes`.
pub const PACKED_LEN: usize = 33;
const P_OFFSET: usize = 0;
const P_LEN: usize = 8;
const P_TYPE: usize = 16;
const P_SIZE: usize = 17;
const P_DELTA_BASE: usize = 25;

/// One `FixedSizeBinary(PACKED_LEN)` element per object, in `order`.
pub fn packed_column(entries: &[IndexEntry], order: &[u32]) -> FixedSizeBinaryArray {
    let mut bytes: Vec<u8> = Vec::with_capacity(order.len() * PACKED_LEN);
    for &i in order {
        let e = &entries[i as usize];
        bytes.extend_from_slice(&e.offset.to_le_bytes());
        bytes.extend_from_slice(&e.len.to_le_bytes());
        bytes.push(e.obj_type.code());
        bytes.extend_from_slice(&e.uncompressed_size.to_le_bytes());
        bytes.extend_from_slice(&e.delta_base.to_le_bytes());
    }
    debug_assert_eq!(bytes.len(), order.len() * PACKED_LEN);
    FixedSizeBinaryArray::new(PACKED_LEN as i32, Buffer::from_vec(bytes), None)
}

#[inline]
fn le64(r: &[u8], at: usize) -> u64 {
    u64::from_le_bytes(r[at..at + 8].try_into().unwrap())
}

fn oid_field(oid_len: usize) -> Field {
    Field::new(COL_OID, DataType::FixedSizeBinary(oid_len as i32), false)
}

/// Serialise one batch as a self-contained Arrow IPC **stream**.
fn to_ipc(batch: &RecordBatch) -> Result<Vec<u8>> {
    let mut out = Vec::with_capacity(64 + batch.get_array_memory_size());
    {
        let mut w = StreamWriter::try_new(&mut out, batch.schema_ref())
            .map_err(|e| anyhow!("ipc writer: {e}"))?;
        w.write(batch).map_err(|e| anyhow!("ipc write: {e}"))?;
        w.finish().map_err(|e| anyhow!("ipc finish: {e}"))?;
    }
    Ok(out)
}

/// Decode an IPC stream **zero-copy**, and return the owning [`Buffer`] with it.
///
/// The `Vec<u8>` a writer produces is only 1-byte aligned, so it is first copied
/// once into a `MutableBuffer` (64-byte aligned). After that every array in the
/// batch is a view into `buffer` — enforced by `with_require_alignment(true)`,
/// which errors instead of quietly re-allocating.
fn decode_ipc(bytes: &[u8]) -> Result<(Buffer, RecordBatch)> {
    let mut mb = MutableBuffer::with_capacity(bytes.len());
    mb.extend_from_slice(bytes);
    decode_ipc_buffer(mb.into())
}

/// The half of [`decode_ipc`] that does not control the buffer's alignment,
/// split out so a test can hand it a deliberately misaligned one and prove
/// `with_require_alignment(true)` is load-bearing rather than decoration.
fn decode_ipc_buffer(owner: Buffer) -> Result<(Buffer, RecordBatch)> {
    // A persisted index is a file: through the IPC guard first, so hostile bytes
    // are an `Err` and never an arrow-ipc panic or abort. It reads the metadata
    // only and copies nothing.
    znippy_common::ipc_guard::check_stream(owner.as_slice())
        .map_err(|e| anyhow!("ipc decode: {e}"))?;
    let mut cursor = owner.clone();
    let mut decoder = StreamDecoder::new().with_require_alignment(true);
    let mut found: Option<RecordBatch> = None;
    while !cursor.is_empty() {
        match decoder.decode(&mut cursor).map_err(|e| anyhow!("ipc decode: {e}"))? {
            Some(b) if found.is_none() => found = Some(b),
            Some(_) => bail!("index section holds more than one record batch"),
            None => {}
        }
    }
    decoder.finish().map_err(|e| anyhow!("ipc unfinished: {e}"))?;
    found.ok_or_else(|| anyhow!("index section holds no record batch"))
        .map(|b| (owner, b))
}

fn col<T: Array + Clone + 'static>(batch: &RecordBatch, name: &str) -> Result<T> {
    batch
        .column_by_name(name)
        .ok_or_else(|| anyhow!("no `{name}` column"))?
        .as_any()
        .downcast_ref::<T>()
        .cloned()
        .ok_or_else(|| anyhow!("`{name}` has an unexpected type"))
}

/// True when `array`'s value bytes lie inside `ipc` — i.e. the decode really was
/// zero-copy and not an aligned re-allocation behind our back.
fn inside(ipc: &Buffer, values: &[u8]) -> bool {
    if values.is_empty() {
        // Nothing to point at; a zero-length column cannot disprove anything.
        return true;
    }
    let base = ipc.as_ptr() as usize;
    let p = values.as_ptr() as usize;
    p >= base && p + values.len() <= base + ipc.len()
}

// ── shared read paths for the two columnar arms ───────────────────────────────

/// The six arrays and **every** read path over them.
///
/// Both columnar arms hold one of these and neither has a hot path of its own,
/// so they cannot drift apart (LAW 5) and the only thing the bench can see
/// between them is how many IPC streams the arrays were framed into. That the
/// two then tie is a fact about Arrow, not about this struct: `FourTables` and
/// `OneTableFourColumns` both have **five payload columns in five separate
/// buffers**, and a full-row fetch touches five distant strides either way. The
/// number of *tables* was never the variable. The number of *columns* is, which
/// is what [`PackedPayload`] changes.
struct ColumnarPayload {
    oid: FixedSizeBinaryArray,
    offset: UInt64Array,
    len: UInt64Array,
    obj_type: UInt8Array,
    size: UInt64Array,
    delta_base: UInt64Array,
}

impl ColumnarPayload {
    /// Five separate buffer reads, five strides.
    #[inline]
    fn row_at(&self, ordinal: u32) -> IndexRow {
        let i = ordinal as usize;
        IndexRow {
            ordinal,
            offset: self.offset.value(i),
            len: self.len.value(i),
            // A code the writer cannot produce cannot appear here: the column is
            // built from `ObjType`, so the round trip is total. `Blob` is the
            // only defensible fallback and it is unreachable in a section this
            // crate wrote.
            obj_type: ObjType::from_code(self.obj_type.value(i)).unwrap_or(ObjType::Blob),
            uncompressed_size: self.size.value(i),
            delta_base: self.delta_base.value(i),
        }
    }

    /// Two of the five buffers.
    #[inline]
    fn extent_at(&self, ordinal: u32) -> (u64, u64) {
        let i = ordinal as usize;
        (self.offset.value(i), self.len.value(i))
    }

    /// The oid column's bytes, contiguous, in ordinal order. One copy out of
    /// the (possibly mapped) buffer — the column is never sliced here, so its
    /// value buffer is exactly `len × width` from offset 0, but the loop over
    /// `value(i)` is kept rather than trusting that, because a wrong ordinal
    /// space is believed by every bitmap built on it.
    fn oids_raw(&self) -> Vec<u8> {
        let n = self.oid.len();
        let w = self.oid.value_length() as usize;
        let mut out = Vec::with_capacity(n * w);
        for i in 0..n {
            out.extend_from_slice(self.oid.value(i));
        }
        out
    }

    /// All six, as the entry that would rebuild this row. The fold's merge
    /// path; never the read path.
    fn entry_at(&self, ordinal: u32) -> IndexEntry {
        let i = ordinal as usize;
        IndexEntry {
            oid: self.oid.value(i).to_vec(),
            offset: self.offset.value(i),
            len: self.len.value(i),
            obj_type: ObjType::from_code(self.obj_type.value(i)).unwrap_or(ObjType::Blob),
            uncompressed_size: self.size.value(i),
            delta_base: self.delta_base.value(i),
        }
    }

    /// One contiguous `u64` run, start to end — the case columnar is for.
    fn sum_uncompressed(&self) -> u64 {
        self.size.values().iter().copied().fold(0u64, u64::wrapping_add)
    }

    /// One contiguous `u8` run: `n` bytes read to answer a question about `n`
    /// rows, which is the least memory traffic any layout can do here.
    fn count_type(&self, t: ObjType) -> usize {
        let c = t.code();
        self.obj_type.values().iter().filter(|&&x| x == c).count()
    }

    fn inside(&self, oid_ipc: &Buffer, extent_ipc: &Buffer, type_ipc: &Buffer, size_ipc: &Buffer) -> bool {
        inside(oid_ipc, self.oid.value_data())
            && inside(extent_ipc, self.offset.values().inner().as_slice())
            && inside(extent_ipc, self.len.values().inner().as_slice())
            && inside(extent_ipc, self.delta_base.values().inner().as_slice())
            && inside(type_ipc, self.obj_type.values().inner().as_slice())
            && inside(size_ipc, self.size.values().inner().as_slice())
    }
}

/// The three schemas layout A frames its facts into, plus the oid schema.
///
/// `delta_base` rides in the **extent** section rather than in a fifth one, and
/// the reason is what the section means: it is *where the bytes are*, and a
/// delta base is an address in exactly the same coordinate space as `offset` —
/// the two are compared against each other by every guard that proves the base
/// resolves. Keeping it here also keeps layout A what the benchmark named it:
/// **four IPC sections** against layout B's one. The section count is the
/// variable under test, and adding a fifth section would have changed the
/// experiment rather than the schema.
fn extent_schema() -> SchemaRef {
    Arc::new(Schema::new(vec![
        Field::new(COL_OFFSET, DataType::UInt64, false),
        Field::new(COL_LEN, DataType::UInt64, false),
        Field::new(COL_DELTA_BASE, DataType::UInt64, false),
    ]))
}

// ── Impl A — FourTables ───────────────────────────────────────────────────────

/// Four Arrow IPC sections — oid, byte extent, type, size — joined by row
/// ordinal.
///
/// This is git's own shape: `.idx`, `.rev` and the missing type/size files, each
/// an independent artifact. A lookup resolves the oid to an ordinal and then
/// indexes three further sections at that ordinal.
///
/// The name counts **sections**, which is this arm's whole difference from
/// [`OneTableFourColumns`]. It carries five payload columns: `delta_base` joined
/// the schema afterwards and shares the extent section, for the reason given on
/// [`extent_schema`].
pub struct FourTables {
    oids: OidResolver,
    /// Owning IPC buffers, kept alive because every array below is a view into
    /// one of them. Four buffers, four allocations, four distinct page runs.
    ipc: [Buffer; 4],
    cols: ColumnarPayload,
    rows: usize,
}

impl FourTables {
    /// Section order: oid, extent, type, size.
    pub fn ipc_section_lens(&self) -> [usize; 4] {
        [self.ipc[0].len(), self.ipc[1].len(), self.ipc[2].len(), self.ipc[3].len()]
    }

    /// The oid column, straight out of its own section — proof that "four
    /// tables" still gives a contiguous scannable column.
    pub fn oid_column(&self) -> &FixedSizeBinaryArray {
        &self.cols.oid
    }

    /// Every column's bytes lie inside the section they were decoded from.
    pub fn column_is_inside_ipc(&self) -> bool {
        self.cols.inside(&self.ipc[0], &self.ipc[1], &self.ipc[2], &self.ipc[3])
    }
}

impl ObjectIndex for FourTables {
    fn build(entries: &[IndexEntry]) -> Result<Self> {
        let order = oid_order(entries);
        let hash = validate(entries, &order)?;
        let oid_len = hash.oid_len();
        let c = columns(entries, &order, oid_len);
        let oids = OidResolver::build(entries, &order, hash)?;

        let s_oid = Arc::new(Schema::new(vec![oid_field(oid_len)]));
        let s_type = Arc::new(Schema::new(vec![Field::new(COL_TYPE, DataType::UInt8, false)]));
        let s_size = Arc::new(Schema::new(vec![Field::new(COL_SIZE, DataType::UInt64, false)]));

        let b_oid = RecordBatch::try_new(s_oid, vec![Arc::new(c.oid)])?;
        let b_extent = RecordBatch::try_new(
            extent_schema(),
            vec![Arc::new(c.offset), Arc::new(c.len), Arc::new(c.delta_base)],
        )?;
        let b_type = RecordBatch::try_new(s_type, vec![Arc::new(c.obj_type)])?;
        let b_size = RecordBatch::try_new(s_size, vec![Arc::new(c.size)])?;

        let (ipc_oid, r_oid) = decode_ipc(&to_ipc(&b_oid)?)?;
        let (ipc_extent, r_extent) = decode_ipc(&to_ipc(&b_extent)?)?;
        let (ipc_type, r_type) = decode_ipc(&to_ipc(&b_type)?)?;
        let (ipc_size, r_size) = decode_ipc(&to_ipc(&b_size)?)?;

        Ok(Self {
            oids,
            ipc: [ipc_oid, ipc_extent, ipc_type, ipc_size],
            cols: ColumnarPayload {
                oid: col(&r_oid, COL_OID)?,
                offset: col(&r_extent, COL_OFFSET)?,
                len: col(&r_extent, COL_LEN)?,
                obj_type: col(&r_type, COL_TYPE)?,
                size: col(&r_size, COL_SIZE)?,
                delta_base: col(&r_extent, COL_DELTA_BASE)?,
            },
            rows: order.len(),
        })
    }

    fn lookup(&self, oid: &[u8]) -> Option<IndexRow> {
        self.oids.ordinal(oid).map(|o| self.cols.row_at(o))
    }

    fn lookup_batch(&self, oids: &[&[u8]]) -> Vec<Option<IndexRow>> {
        self.oids.ordinals(oids).into_iter().map(|o| o.map(|o| self.cols.row_at(o))).collect()
    }

    fn ordinals_batch(&self, oids: &[&[u8]]) -> Vec<Option<u32>> {
        self.oids.ordinals(oids)
    }

    fn extents_batch(&self, oids: &[&[u8]]) -> Vec<Option<(u64, u64)>> {
        self.oids.ordinals(oids).into_iter().map(|o| o.map(|o| self.cols.extent_at(o))).collect()
    }

    fn sum_uncompressed(&self) -> u64 {
        self.cols.sum_uncompressed()
    }

    fn count_type(&self, t: ObjType) -> usize {
        self.cols.count_type(t)
    }

    fn name(&self) -> &'static str {
        "FourTables"
    }

    fn oid_at(&self, ordinal: u32) -> Vec<u8> {
        self.cols.oid.value(ordinal as usize).to_vec()
    }

    fn entry_at(&self, ordinal: u32) -> IndexEntry {
        self.cols.entry_at(ordinal)
    }

    fn extent_at(&self, ordinal: u32) -> (u64, u64) {
        self.cols.extent_at(ordinal)
    }

    fn oids_raw(&self) -> Vec<u8> {
        self.cols.oids_raw()
    }

    fn len(&self) -> usize {
        self.rows
    }

    fn ipc_bytes(&self) -> usize {
        self.ipc.iter().map(|b| b.len()).sum()
    }

    fn resident_bytes(&self) -> usize {
        self.ipc_bytes() + self.oids.section_bytes
    }
}

// ── Impl B — OneTableFourColumns ──────────────────────────────────────────────

/// One Arrow IPC section, one table, the facts one column each.
///
/// Arrow is columnar, so this is **not** row-contiguous: `offset` is its own
/// buffer, `len` is its own buffer, and a full-row fetch still touches five
/// strides that are megabytes apart. What it saves over [`FourTables`] is the
/// framing of three IPC streams and nothing else — which is exactly what the
/// measurement found.
///
/// The name counts the columns it had when it was named and measured; since
/// `delta_base` (PLAN §13) there are five payload columns here and in
/// [`FourTables`] alike, so the two are still separated by section count and
/// nothing else.
pub struct OneTableFourColumns {
    oids: OidResolver,
    ipc: Buffer,
    cols: ColumnarPayload,
    rows: usize,
}

impl OneTableFourColumns {
    /// What this arm writes under [`PERSISTED_LAYOUT_KEY`], and the only value
    /// [`ObjectIndex::from_persisted`] accepts there.
    const LAYOUT_NAME: &'static str = "OneTableFourColumns";

    /// The one constructor both `build` and `from_persisted` end in, so the
    /// column wiring is spelled once.
    fn from_decoded(oids: OidResolver, ipc: Buffer, r: &RecordBatch, rows: usize) -> Result<Self> {
        Ok(Self {
            oids,
            ipc,
            cols: ColumnarPayload {
                oid: col(r, COL_OID)?,
                offset: col(r, COL_OFFSET)?,
                len: col(r, COL_LEN)?,
                obj_type: col(r, COL_TYPE)?,
                size: col(r, COL_SIZE)?,
                delta_base: col(r, COL_DELTA_BASE)?,
            },
            rows,
        })
    }

    pub fn oid_column(&self) -> &FixedSizeBinaryArray {
        &self.cols.oid
    }

    /// All five arrays are views into the one IPC buffer.
    pub fn column_is_inside_ipc(&self) -> bool {
        self.cols.inside(&self.ipc, &self.ipc, &self.ipc, &self.ipc)
    }

    /// **The bytes this index would be persisted as** — the single Arrow IPC
    /// stream every column is a view into, exactly as [`to_ipc`] framed it.
    ///
    /// [`ObjectIndex::ipc_bytes`] reports this buffer's *length*, which is
    /// enough to compare two layouts' footprints and not enough to write one
    /// down. `examples/tail_write_bench.rs` needs the bytes themselves: its
    /// no-tail arm has to rewrite the whole index on every push, and a
    /// measurement of that write is only honest if the bytes going to the file
    /// are the index's own and not a stand-in of the same size.
    ///
    /// A borrowed slice and not a `Vec`: the buffer is already resident and
    /// aligned, and handing out a copy would put an allocation inside the very
    /// write this exists to measure.
    pub fn ipc_slice(&self) -> &[u8] {
        self.ipc.as_slice()
    }
}

impl ObjectIndex for OneTableFourColumns {
    fn build(entries: &[IndexEntry]) -> Result<Self> {
        let order = oid_order(entries);
        let hash = validate(entries, &order)?;
        let oid_len = hash.oid_len();
        let c = columns(entries, &order, oid_len);
        let oids = OidResolver::build(entries, &order, hash)?;

        // The stamp rides inside the IPC bytes, so the persisted file needs no
        // framing of its own and `from_persisted` can refuse a foreign buffer
        // by reading the schema it already had to decode.
        let schema: SchemaRef = Arc::new(Schema::new_with_metadata(
            vec![
                oid_field(oid_len),
                Field::new(COL_OFFSET, DataType::UInt64, false),
                Field::new(COL_LEN, DataType::UInt64, false),
                Field::new(COL_TYPE, DataType::UInt8, false),
                Field::new(COL_SIZE, DataType::UInt64, false),
                Field::new(COL_DELTA_BASE, DataType::UInt64, false),
            ],
            persisted_metadata(Self::LAYOUT_NAME),
        ));
        let batch = RecordBatch::try_new(
            schema,
            vec![
                Arc::new(c.oid),
                Arc::new(c.offset),
                Arc::new(c.len),
                Arc::new(c.obj_type),
                Arc::new(c.size),
                Arc::new(c.delta_base),
            ],
        )?;
        let (ipc, r) = decode_ipc(&to_ipc(&batch)?)?;
        Self::from_decoded(oids, ipc, &r, order.len())
    }

    fn persists() -> bool {
        true
    }

    fn persisted_bytes(&self) -> Option<&[u8]> {
        Some(self.ipc.as_slice())
    }

    /// The buffer is decoded in place — `decode_ipc_buffer`, not `decode_ipc`,
    /// so a memory-mapped file stays mapped and every column is a view into
    /// it — and the `stree` is rebuilt over its oid column. The stree is the
    /// one thing this does not read back: its section is a reader-side
    /// structure `GitOidIndex::parse_as` derives, and building it is O(n) in
    /// the oid count against the O(n log n) sort-and-encode a rebuild from
    /// redb pays, with no B-tree walk in front of it.
    fn from_persisted(buf: Buffer) -> Result<Self> {
        let (ipc, r) = decode_ipc_buffer(buf)?;
        check_persisted_stamp(&r, Self::LAYOUT_NAME)?;
        let oid: FixedSizeBinaryArray = col(&r, COL_OID)?;
        let hash = match oid.value_length() {
            20 => GitHashKind::Sha1,
            32 => GitHashKind::Sha256,
            w => bail!("persisted oid column is {w} bytes wide, expected 20 or 32"),
        };
        let oids = OidResolver::from_oid_column(&oid, hash)?;
        let rows = r.num_rows();
        Self::from_decoded(oids, ipc, &r, rows)
    }

    fn oid_at(&self, ordinal: u32) -> Vec<u8> {
        self.cols.oid.value(ordinal as usize).to_vec()
    }

    fn entry_at(&self, ordinal: u32) -> IndexEntry {
        self.cols.entry_at(ordinal)
    }

    fn extent_at(&self, ordinal: u32) -> (u64, u64) {
        self.cols.extent_at(ordinal)
    }

    fn oids_raw(&self) -> Vec<u8> {
        self.cols.oids_raw()
    }

    fn lookup(&self, oid: &[u8]) -> Option<IndexRow> {
        self.oids.ordinal(oid).map(|o| self.cols.row_at(o))
    }

    fn lookup_batch(&self, oids: &[&[u8]]) -> Vec<Option<IndexRow>> {
        self.oids.ordinals(oids).into_iter().map(|o| o.map(|o| self.cols.row_at(o))).collect()
    }

    fn ordinals_batch(&self, oids: &[&[u8]]) -> Vec<Option<u32>> {
        self.oids.ordinals(oids)
    }

    fn extents_batch(&self, oids: &[&[u8]]) -> Vec<Option<(u64, u64)>> {
        self.oids.ordinals(oids).into_iter().map(|o| o.map(|o| self.cols.extent_at(o))).collect()
    }

    fn sum_uncompressed(&self) -> u64 {
        self.cols.sum_uncompressed()
    }

    fn count_type(&self, t: ObjType) -> usize {
        self.cols.count_type(t)
    }

    fn name(&self) -> &'static str {
        "OneTableFourColumns"
    }

    fn len(&self) -> usize {
        self.rows
    }

    fn ipc_bytes(&self) -> usize {
        self.ipc.len()
    }

    fn resident_bytes(&self) -> usize {
        self.ipc_bytes() + self.oids.section_bytes
    }
}

// ── Impl C — PackedPayload ────────────────────────────────────────────────────

/// **One** payload column: all five facts of one object packed adjacently into
/// a single `FixedSizeBinary(33)` element.
///
/// This is the arm the first two were both missing. `FourTables` and
/// `OneTableFourColumns` differ in how many *tables* they use and agree in
/// having **five payload columns**, so a full-row fetch costs five buffer reads
/// at five unrelated addresses in both. Here it costs one:
///
/// ```text
/// per row, byte for byte (little-endian, no padding, no nulls):
///   0  u64  offset
///   8  u64  len
///  16  u8   object type code (1 commit, 2 tree, 3 blob, 4 tag, 6 ofs-delta, 7 ref-delta)
///  17  u64  uncompressed_size
///  25  u64  delta_base — archive offset of the base entry, 0 for none
///  33  ── next row
/// ```
///
/// **It was 24 bytes narrower before `delta_base` (PLAN §13).** A 33-byte record
/// spans one 64-byte cache line 49% of the time and two the rest, so a full-row
/// fetch is ~1.5 lines against the five the columnar arms touch — where at 25
/// bytes it was ~1.4 against four. The trade moved slightly, in both directions
/// at once, which is why the numbers below were re-taken rather than scaled.
///
/// The oid stays in its own column. It is the key the `stree` resolves
/// *against*, not one of the facts a resolved lookup fetches, so packing it in
/// would grow every payload read by 20 bytes nobody reads at lookup time.
///
/// ## The trade, stated before it is measured
///
/// * **Full-row access should win** — one stride instead of four.
/// * **Column scans should lose, and lose badly.** `count_type` reads one byte
///   per row; the columnar arms stream `n` bytes to answer it, this arm drags
///   `33n` through cache to use `n`. `sum_uncompressed` reads 8 of every 33
///   instead of 8 of every 8.
/// * It is **no longer typed Arrow columns**. A `FixedSizeBinary(33)` is
///   opaque to DuckDB / Polars / DataFusion, which can query the other two arms
///   directly off the IPC bytes. That is a real cost and it is not a
///   performance one.
pub struct PackedPayload {
    oids: OidResolver,
    ipc: Buffer,
    oid: FixedSizeBinaryArray,
    payload: FixedSizeBinaryArray,
    rows: usize,
}

/// Column name of the single packed payload column.
pub const COL_PACKED: &str = "packed";

impl PackedPayload {
    /// [`ObjectIndex::build`] with the `stree` section's [`OidLayout`] chosen
    /// explicitly — the whole apparatus of the alignment experiment
    /// (`examples/oid_align_bench.rs`).
    ///
    /// It lives on this arm alone rather than on the trait because
    /// `ordinals_batch` is **the same code in all three arms** (they share one
    /// [`OidResolver`]), so a second arm would add a second Arrow payload to
    /// keep resident and would measure nothing new. This is the arm the payload
    /// sweep picked.
    pub fn build_with_oid_layout(entries: &[IndexEntry], layout: OidLayout) -> Result<Self> {
        let order = oid_order(entries);
        let hash = validate(entries, &order)?;
        let oid_len = hash.oid_len();
        let oid_arr = oid_column(entries, &order, oid_len);
        let packed = packed_column(entries, &order);
        let oids = OidResolver::build_with(entries, &order, hash, layout)?;
        Self::assemble(oids, oid_arr, packed, oid_len, order.len())
    }

    /// The cache-line phase of this index's `stree` keyspace — 0 for
    /// [`OidLayout::Aligned64`]. The bench asserts on it before quoting a
    /// number, because two arms that landed on the same phase would be one arm
    /// measured twice.
    pub fn keyspace_phase(&self) -> usize {
        self.oids.tree.keyspace_phase()
    }

    /// Everything after the oid resolver: framing the two columns into one IPC
    /// stream and decoding it back zero-copy. Shared with `build` so the two
    /// cannot drift (LAW 5) — the only difference between them must be the
    /// `stree` layout, or the experiment is measuring two indexes.
    fn assemble(
        oids: OidResolver,
        oid_arr: FixedSizeBinaryArray,
        packed: FixedSizeBinaryArray,
        oid_len: usize,
        rows: usize,
    ) -> Result<Self> {
        let schema: SchemaRef = Arc::new(Schema::new(vec![
            oid_field(oid_len),
            Field::new(COL_PACKED, DataType::FixedSizeBinary(PACKED_LEN as i32), false),
        ]));
        let batch = RecordBatch::try_new(schema, vec![Arc::new(oid_arr), Arc::new(packed)])?;
        let (ipc, r) = decode_ipc(&to_ipc(&batch)?)?;
        Ok(Self { oids, ipc, oid: col(&r, COL_OID)?, payload: col(&r, COL_PACKED)?, rows })
    }

    pub fn oid_column(&self) -> &FixedSizeBinaryArray {
        &self.oid
    }

    pub fn column_is_inside_ipc(&self) -> bool {
        inside(&self.ipc, self.oid.value_data()) && inside(&self.ipc, self.payload.value_data())
    }

    /// One buffer read, one stride.
    #[inline]
    fn row_at(&self, ordinal: u32) -> IndexRow {
        let r = self.payload.value(ordinal as usize);
        IndexRow {
            ordinal,
            offset: le64(r, P_OFFSET),
            len: le64(r, P_LEN),
            obj_type: ObjType::from_code(r[P_TYPE]).unwrap_or(ObjType::Blob),
            uncompressed_size: le64(r, P_SIZE),
            delta_base: le64(r, P_DELTA_BASE),
        }
    }

    /// The same one buffer read — a partial row costs a packed layout exactly
    /// what a full row costs, which is half the point of the trade.
    #[inline]
    fn extent_at(&self, ordinal: u32) -> (u64, u64) {
        let r = self.payload.value(ordinal as usize);
        (le64(r, P_OFFSET), le64(r, P_LEN))
    }
}

impl ObjectIndex for PackedPayload {
    fn build(entries: &[IndexEntry]) -> Result<Self> {
        let order = oid_order(entries);
        let hash = validate(entries, &order)?;
        let oid_len = hash.oid_len();
        let oid_arr = oid_column(entries, &order, oid_len);
        let packed = packed_column(entries, &order);
        let oids = OidResolver::build(entries, &order, hash)?;
        Self::assemble(oids, oid_arr, packed, oid_len, order.len())
    }

    fn lookup(&self, oid: &[u8]) -> Option<IndexRow> {
        self.oids.ordinal(oid).map(|o| self.row_at(o))
    }

    fn lookup_batch(&self, oids: &[&[u8]]) -> Vec<Option<IndexRow>> {
        self.oids.ordinals(oids).into_iter().map(|o| o.map(|o| self.row_at(o))).collect()
    }

    fn ordinals_batch(&self, oids: &[&[u8]]) -> Vec<Option<u32>> {
        self.oids.ordinals(oids)
    }

    fn extents_batch(&self, oids: &[&[u8]]) -> Vec<Option<(u64, u64)>> {
        self.oids.ordinals(oids).into_iter().map(|o| o.map(|o| self.extent_at(o))).collect()
    }

    /// Strided: 8 useful bytes out of every 25 touched.
    fn sum_uncompressed(&self) -> u64 {
        let d = self.payload.value_data();
        let mut acc = 0u64;
        let mut i = P_SIZE;
        while i + 8 <= d.len() {
            acc = acc.wrapping_add(le64(d, i));
            i += PACKED_LEN;
        }
        acc
    }

    /// Strided: 1 useful byte out of every 25 touched.
    fn count_type(&self, t: ObjType) -> usize {
        let c = t.code();
        let d = self.payload.value_data();
        let mut n = 0usize;
        let mut i = P_TYPE;
        while i < d.len() {
            if d[i] == c {
                n += 1;
            }
            i += PACKED_LEN;
        }
        n
    }

    fn name(&self) -> &'static str {
        "PackedPayload"
    }

    fn oid_at(&self, ordinal: u32) -> Vec<u8> {
        self.oid.value(ordinal as usize).to_vec()
    }

    fn entry_at(&self, ordinal: u32) -> IndexEntry {
        let r = self.row_at(ordinal);
        IndexEntry {
            oid: self.oid.value(ordinal as usize).to_vec(),
            offset: r.offset,
            len: r.len,
            obj_type: r.obj_type,
            uncompressed_size: r.uncompressed_size,
            delta_base: r.delta_base,
        }
    }

    fn extent_at(&self, ordinal: u32) -> (u64, u64) {
        // The inherent `extent_at` (one 33-byte stride, two fields out of it).
        PackedPayload::extent_at(self, ordinal)
    }

    fn oids_raw(&self) -> Vec<u8> {
        let n = self.oid.len();
        let mut out = Vec::with_capacity(n * self.oid.value_length() as usize);
        for i in 0..n {
            out.extend_from_slice(self.oid.value(i));
        }
        out
    }

    fn len(&self) -> usize {
        self.rows
    }

    fn ipc_bytes(&self) -> usize {
        self.ipc.len()
    }

    fn resident_bytes(&self) -> usize {
        self.ipc_bytes() + self.oids.section_bytes
    }
}

// ── deterministic workload generation, shared by tests and the bench ──────────

/// splitmix64. Seeded, so a bench run is reproducible and a historized series is
/// comparable at all.
pub struct Rng(pub u64);

impl Rng {
    pub fn next_u64(&mut self) -> u64 {
        self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15);
        let mut z = self.0;
        z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
        z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
        z ^ (z >> 31)
    }

    pub fn fill(&mut self, out: &mut [u8]) {
        for c in out.chunks_mut(8) {
            let w = self.next_u64().to_le_bytes();
            let n = c.len();
            c.copy_from_slice(&w[..n]);
        }
    }
}

/// `n` distinct entries with uniformly random oids.
///
/// Uniform randomness is the property under test, not a convenience: real oids
/// are hashes and share no prefixes, so a generator with structure would give
/// the stree an unrepresentative keyspace.
///
/// **Every `OfsDelta` entry carries a non-zero `delta_base`** — the offset of
/// the entry immediately before it, which is a real row of the same workload —
/// and every other entry carries `0`. LAW 2's identity-value trap is the whole
/// reason: a generator that left `delta_base` at `0` everywhere would let a
/// layout that never wrote the column, or read it from the wrong stride, pass
/// every comparison in this module. It is derived from `out.last()` rather than
/// from `rng`, so oids, offsets, lengths and types are bit-identical to what
/// this generator produced before the column existed and no earlier figure is
/// invalidated by the workload changing under it.
pub fn synthetic_entries(n: usize, oid_len: usize, seed: u64) -> Vec<IndexEntry> {
    let mut rng = Rng(seed);
    let mut seen = std::collections::HashSet::with_capacity(n * 2);
    let mut out: Vec<IndexEntry> = Vec::with_capacity(n);
    let mut off = 12u64; // past a pack header
    while out.len() < n {
        let mut oid = vec![0u8; oid_len];
        rng.fill(&mut oid);
        if !seen.insert(oid.clone()) {
            continue;
        }
        // Length distribution roughly like a real pack: mostly small, a long
        // tail. The exact shape does not affect a point lookup, but it keeps
        // the u64 columns from being all-identical, which would let a compressor
        // or a branch predictor flatter one arm over the other.
        let len = 32 + (rng.next_u64() % 4096);
        let ty = ObjType::ALL[(rng.next_u64() % 6) as usize];
        // An ofs-delta's base is the entry before it: always an offset that some
        // row of this same workload really starts at, never 0, and never a
        // forward reference. A ref-delta's base is named by oid and has no
        // offset until it is resolved, so it records the 0 sentinel — as does
        // every whole object.
        let delta_base = match (ty, out.last()) {
            (ObjType::OfsDelta, Some(prev)) => prev.offset,
            _ => 0,
        };
        out.push(IndexEntry {
            oid,
            offset: off,
            len,
            obj_type: ty,
            uncompressed_size: len * (1 + rng.next_u64() % 5),
            delta_base,
        });
        off += len;
    }
    out
}


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

    /// A persisted index is a file on disk: hostile bytes handed to
    /// `from_persisted` are an `Err`, never a panic or an abort.
    #[test]
    #[ignore = "run as a child by a_hostile_persisted_index_is_an_error"]
    fn hostile_child() {
        if crate::hostile_inputs::child_case().is_none() {
            return;
        }
        let entries = synthetic_entries(3, 20, 7);
        let ix = OneTableFourColumns::build(&entries).unwrap();
        let valid = ix.persisted_bytes().unwrap().to_vec();
        assert!(OneTableFourColumns::from_persisted(Buffer::from(valid.clone())).is_ok());
        crate::hostile_inputs::sweep(&valid, |_, b| {
            let mut mb = MutableBuffer::with_capacity(b.len());
            mb.extend_from_slice(b);
            let _ = OneTableFourColumns::from_persisted(mb.into());
            let _ = decode_ipc(b);
        });
    }

    #[test]
    fn a_hostile_persisted_index_is_an_error() {
        crate::hostile_inputs::run_in_child("index_layout::tests::hostile_child", "persisted");
    }

    /// All three arms over the same entries. Every semantic test runs against
    /// the triple, because "the three layouts are one index" is the claim the
    /// whole comparison rests on.
    fn arms(entries: &[IndexEntry]) -> (FourTables, OneTableFourColumns, PackedPayload) {
        (
            FourTables::build(entries).expect("A builds"),
            OneTableFourColumns::build(entries).expect("B builds"),
            PackedPayload::build(entries).expect("C builds"),
        )
    }

    /// The core correctness claim: **the three layouts are the same index**.
    ///
    /// Asserted on applied output — the five facts of every row, plus the
    /// ordinal — for every present oid, for absent oids, and through the
    /// serial, batch, extent-only and ordinal-only paths. A row-count or an
    /// `is_some()` check would pass for three indexes that disagreed on every
    /// value.
    ///
    /// Seen RED by changing `ColumnarPayload::row_at` to read `self.len.value(i)`
    /// for `offset`: "layouts disagree on 80313d462e6994e9e819193e6fdbd5f8390e92e3
    ///   left: IndexRow { ordinal: 259, offset: 344, len: 344, .. }
    ///  right: IndexRow { ordinal: 259, offset: 12, len: 344, .. }".
    /// Restored.
    ///
    /// Seen RED a second time, for the packed arm specifically, by swapping the
    /// `len` and `uncompressed_size` writes in `packed_column` only, so the
    /// writer and the reader disagree about the record layout:
    /// "C disagrees with A on 80313d462e6994e9e819193e6fdbd5f8390e92e3
    ///   left: IndexRow { offset: 12, len: 344, uncompressed_size: 1720 }
    ///  right: IndexRow { offset: 12, len: 1720, uncompressed_size: 344 }".
    /// Restored. Note that A and B stayed green throughout — only a guard that
    /// compares the packed arm against them can see this class of bug.
    ///
    /// Seen RED a third time, when `delta_base` was added, by dropping
    /// [`FourTables`] read `COL_LEN` into `ColumnarPayload::delta_base` — the
    /// single most plausible way to wire a new column in wrong, since both are
    /// `u64` columns of the same section: "A and B disagree on
    /// 80313d462e6994e9e819193e6fdbd5f8390e92e3 / left: IndexRow { ordinal: 259,
    /// offset: 12, len: 344, obj_type: Tree, uncompressed_size: 1720,
    /// **delta_base: 344** } / right: IndexRow { … **delta_base: 0** }". B and C
    /// were right and A was wrong, and only the row-for-row comparison said so.
    /// Restored.
    #[test]
    fn the_three_layouts_return_identical_rows() {
        for &oid_len in &[20usize, 32] {
            let entries = synthetic_entries(500, oid_len, 0xA11CE);
            let (a, b, c) = arms(&entries);
            assert_eq!(a.len(), 500);
            assert_eq!(b.len(), 500);
            assert_eq!(c.len(), 500);

            // Not a blind guard: the workload has to carry the column at
            // something other than its identity value, or every arm agreeing on
            // `delta_base: 0` would prove nothing about `delta_base` at all.
            let with_base = entries.iter().filter(|e| e.delta_base != 0).count();
            assert!(
                with_base >= 50,
                "only {with_base} of 500 entries carry a non-zero delta_base — this guard would \
                 sit on the identity value and could not see a column that was never written"
            );

            let mut hits = 0usize;
            for e in &entries {
                let ra = a.lookup(&e.oid).unwrap_or_else(|| {
                    panic!("{} missed {}", a.name(), hex::encode(&e.oid))
                });
                let rb = b.lookup(&e.oid).unwrap_or_else(|| {
                    panic!("{} missed {}", b.name(), hex::encode(&e.oid))
                });
                let rc = c.lookup(&e.oid).unwrap_or_else(|| {
                    panic!("{} missed {}", c.name(), hex::encode(&e.oid))
                });
                assert_eq!(ra, rb, "A and B disagree on {}", hex::encode(&e.oid));
                assert_eq!(
                    ra,
                    rc,
                    "C disagrees with A on {}",
                    hex::encode(&e.oid)
                );
                // …and all three agree with the input, not just with each other.
                assert_eq!(rc.offset, e.offset);
                assert_eq!(rc.len, e.len);
                assert_eq!(rc.obj_type, e.obj_type);
                assert_eq!(rc.uncompressed_size, e.uncompressed_size);
                assert_eq!(
                    rc.delta_base,
                    e.delta_base,
                    "delta_base did not survive C for {}",
                    hex::encode(&e.oid)
                );
                assert_eq!(ra.delta_base, e.delta_base, "delta_base did not survive A");
                assert_eq!(rb.delta_base, e.delta_base, "delta_base did not survive B");
                hits += 1;
            }
            assert_eq!(hits, 500, "every entry must resolve");

            // Absent oids: same answer everywhere, and that answer is None.
            let absent = synthetic_entries(200, oid_len, 0xBEEF_0000);
            for e in &absent {
                assert_eq!(a.lookup(&e.oid), None);
                assert_eq!(b.lookup(&e.oid), None);
                assert_eq!(c.lookup(&e.oid), None);
            }

            // Every batch path, on a mix, positionally.
            let mut q: Vec<&[u8]> = Vec::new();
            for (i, e) in entries.iter().enumerate() {
                q.push(&e.oid);
                if i < absent.len() {
                    q.push(&absent[i].oid);
                }
            }
            let ba = a.lookup_batch(&q);
            let bb = b.lookup_batch(&q);
            let bc = c.lookup_batch(&q);
            assert_eq!(ba.len(), q.len());
            assert_eq!(ba, bb, "A/B batch paths disagree at oid_len {oid_len}");
            assert_eq!(ba, bc, "A/C batch paths disagree at oid_len {oid_len}");
            let n_hits = ba.iter().filter(|r| r.is_some()).count();
            assert_eq!(n_hits, 500, "expected exactly the 500 present oids to hit");
            for (i, r) in ba.iter().enumerate() {
                assert_eq!(*r, c.lookup(q[i]), "C batch/serial disagree at {i}");
            }

            // The partial-row and ordinal-only paths agree with the full row.
            let ea = a.extents_batch(&q);
            let ec = c.extents_batch(&q);
            assert_eq!(ea, ec, "extent paths disagree");
            let oa = a.ordinals_batch(&q);
            let oc = c.ordinals_batch(&q);
            assert_eq!(oa, oc, "ordinal paths disagree");
            for i in 0..q.len() {
                assert_eq!(ea[i], ba[i].map(|r| (r.offset, r.len)), "extent != row at {i}");
                assert_eq!(oa[i], ba[i].map(|r| r.ordinal), "ordinal != row at {i}");
            }
        }
    }

    /// **`delta_base` is an offset that locates a row of this very table**, in
    /// all three arms — which is the property an ordinal could not have.
    ///
    /// Read back through the index (never from the input entries), every
    /// non-zero `delta_base` is looked up against the set of `offset`s the index
    /// itself reports, and it has to land on one, earlier in the archive than
    /// the delta that names it. `offset` and `delta_base` are the same
    /// coordinate space, and this is the assertion that says so.
    ///
    /// Seen RED by making `synthetic_entries` write `prev.offset + 1` as the
    /// delta base — a value one byte off a real entry boundary, which is
    /// precisely how an off-by-one in a rebase or in a varint would look:
    /// "A: delta_base 4601 of ordinal 1256 lands on no entry offset in this
    /// index". Restored to `prev.offset`.
    ///
    /// Seen RED a second time, on the ordinal question itself, by writing the
    /// base's *row index* into the column instead of its offset:
    /// "A: delta_base 2 of ordinal 1256 lands on no entry offset in this index"
    /// — a small number that is a perfectly valid row address and an invalid
    /// archive address, which is §13's decision failing loudly rather than
    /// silently. Restored.
    #[test]
    fn delta_base_locates_a_real_entry_in_every_arm() {
        let entries = synthetic_entries(2000, 20, 0xDE17A);
        let (a, b, c) = arms(&entries);

        // The offsets this index reports, taken from the index and not from the
        // input: `lookup` of every stored oid.
        let offsets: std::collections::HashSet<u64> =
            entries.iter().filter_map(|e| a.lookup(&e.oid)).map(|r| r.offset).collect();
        assert_eq!(offsets.len(), 2000, "every row must report a distinct offset");

        let arms: [(&str, &dyn ObjectIndex); 3] = [("A", &a), ("B", &b), ("C", &c)];
        let mut checked = 0usize;
        for (name, idx) in arms {
            let mut with_base = 0usize;
            for e in &entries {
                let row = idx.lookup(&e.oid).expect("stored oid resolves");
                if row.delta_base == 0 {
                    assert_ne!(
                        row.obj_type,
                        ObjType::OfsDelta,
                        "{name}: an ofs-delta with no recorded base is a row that cannot be \
                         resolved at all"
                    );
                    continue;
                }
                with_base += 1;
                assert!(
                    offsets.contains(&row.delta_base),
                    "{name}: delta_base {} of ordinal {} lands on no entry offset in this index",
                    row.delta_base,
                    row.ordinal
                );
                assert!(
                    row.delta_base < row.offset,
                    "{name}: delta_base {} is not earlier in the archive than the delta at {}",
                    row.delta_base,
                    row.offset
                );
            }
            assert!(
                with_base > 100,
                "{name}: only {with_base} of 2000 rows carried a base — nothing was proven"
            );
            checked += with_base;
        }
        assert!(checked > 300, "three arms must each have checked real bases");
    }

    /// The two column scans return the same answer from all three arms, and
    /// that answer is the one computed straight from the input entries.
    ///
    /// This is the guard that makes the scan half of the benchmark meaningful:
    /// a strided reader that skipped rows would be very fast and very wrong.
    ///
    /// Seen RED by advancing `PackedPayload::sum_uncompressed` by `PACKED_LEN + 8`
    /// instead of `PACKED_LEN`: "C sum_uncompressed: left 8826373087920709042,
    /// right 12349770". The stride slipped out of phase, so it summed bytes
    /// straddling field boundaries — a fast, wrong scan that no timing
    /// assertion would ever have noticed. Restored.
    #[test]
    fn column_scans_agree_across_arms_and_with_the_input() {
        let entries = synthetic_entries(2000, 20, 0x5CA7);
        let (a, b, c) = arms(&entries);

        let want: u64 = entries.iter().map(|e| e.uncompressed_size).sum();
        assert_eq!(a.sum_uncompressed(), want, "A sum_uncompressed");
        assert_eq!(b.sum_uncompressed(), want, "B sum_uncompressed");
        assert_eq!(c.sum_uncompressed(), want, "C sum_uncompressed");
        assert!(want > 0, "the workload must actually carry sizes");

        let mut total = 0usize;
        for t in ObjType::ALL {
            let want = entries.iter().filter(|e| e.obj_type == t).count();
            assert_eq!(a.count_type(t), want, "A count_type({})", t.as_str());
            assert_eq!(b.count_type(t), want, "B count_type({})", t.as_str());
            assert_eq!(c.count_type(t), want, "C count_type({})", t.as_str());
            assert!(want > 0, "{} must appear in the workload", t.as_str());
            total += want;
        }
        assert_eq!(total, entries.len(), "the six types must partition the index");
    }

    /// The packed record is 33 bytes laid out exactly as documented, read
    /// straight out of the IPC buffer rather than through the accessor that is
    /// under test. A layout that changed silently would still round-trip
    /// through its own reader; this pins it to the bytes.
    ///
    /// Seen RED by moving the type byte from offset 16 to offset 24 in
    /// `packed_column`: "row 0 byte 16 is 0 but the type code is 6". Restored.
    ///
    /// Seen RED again for the fifth field, by writing `e.offset` where
    /// `packed_column` writes `e.delta_base`: "row 0 bytes 25..33 must be the
    /// delta base / left: 368 / right: 12". Restored. Writing the *offset*
    /// there is the failure mode worth catching: it is a plausible archive
    /// offset, it is even a real entry boundary, so nothing downstream would
    /// look odd — only a comparison against the input entry tells the two
    /// apart.
    #[test]
    fn the_packed_record_is_the_documented_33_bytes() {
        let entries = synthetic_entries(64, 20, 0x9F);
        let c = PackedPayload::build(&entries).unwrap();
        assert_eq!(PACKED_LEN, 33);
        assert_eq!(PACKED_LEN, 8 + 8 + 1 + 8 + 8, "offset+len+type+size+delta_base");

        let mut sorted: Vec<&IndexEntry> = entries.iter().collect();
        sorted.sort_by(|x, y| x.oid.cmp(&y.oid));
        let d = c.payload.value_data();
        assert_eq!(d.len(), 64 * PACKED_LEN, "payload column is not 33 bytes per row");
        assert!(
            sorted.iter().filter(|e| e.delta_base != 0).count() >= 4,
            "the fixture must carry real delta bases or bytes 25..33 are all zero"
        );

        for (i, e) in sorted.iter().enumerate() {
            let r = &d[i * PACKED_LEN..(i + 1) * PACKED_LEN];
            assert_eq!(
                u64::from_le_bytes(r[0..8].try_into().unwrap()),
                e.offset,
                "row {i} bytes 0..8 must be the offset"
            );
            assert_eq!(
                u64::from_le_bytes(r[8..16].try_into().unwrap()),
                e.len,
                "row {i} bytes 8..16 must be the len"
            );
            assert_eq!(
                r[16],
                e.obj_type.code(),
                "row {i} byte 16 is {} but the type code is {}",
                r[16],
                e.obj_type.code()
            );
            assert_eq!(
                u64::from_le_bytes(r[17..25].try_into().unwrap()),
                e.uncompressed_size,
                "row {i} bytes 17..25 must be the uncompressed size"
            );
            assert_eq!(
                u64::from_le_bytes(r[25..33].try_into().unwrap()),
                e.delta_base,
                "row {i} bytes 25..33 must be the delta base"
            );
        }
    }

    /// The whole premise of the third arm, asserted as a byte count rather than
    /// as an argument: a full-row fetch touches **one** payload buffer of 33
    /// bytes per row, where the columnar arms touch five buffers whose bases are
    /// megabytes apart.
    ///
    /// Seen RED by asserting the columnar arm's payload buffers were within 25
    /// bytes of each other: "offset and size buffers are 1737664 bytes apart,
    /// not adjacent" at 100 000 rows. That distance is the fact the packed arm
    /// exists to change. Restored to assert the true distances.
    #[test]
    fn a_packed_row_is_one_stride_and_a_columnar_row_is_five() {
        let entries = synthetic_entries(100_000, 20, 0x0FF5);
        let (_, b, c) = arms(&entries);

        // Columnar: five payload buffers, and consecutive facts of ONE row are
        // ~800 kB apart because each column is 100000 × 8 bytes long.
        let base = |s: &[u8]| s.as_ptr() as usize;
        let off = base(b.cols.offset.values().inner().as_slice());
        let len = base(b.cols.len.values().inner().as_slice());
        let size = base(b.cols.size.values().inner().as_slice());
        let dbase = base(b.cols.delta_base.values().inner().as_slice());
        assert!(
            len.abs_diff(off) >= 800_000,
            "offset and len columns are only {} bytes apart — this arm is supposed to be \
             columnar",
            len.abs_diff(off)
        );
        assert!(size.abs_diff(off) >= 800_000);
        assert!(
            dbase.abs_diff(off) >= 800_000 && dbase.abs_diff(size) >= 800_000,
            "delta_base is {} bytes from offset and {} from size — the fifth fact must be its \
             own stride, not a field inside another column",
            dbase.abs_diff(off),
            dbase.abs_diff(size)
        );

        // Packed: the five facts of one row are inside 33 consecutive bytes.
        let d = c.payload.value_data();
        assert_eq!(d.len(), 100_000 * PACKED_LEN);
        let row7 = &d[7 * PACKED_LEN..8 * PACKED_LEN];
        let r = c.lookup(&entries.iter().min_by_key(|e| e.oid.clone()).unwrap().oid);
        assert!(r.is_some(), "the lexicographically first oid must resolve");
        // Every fact of row 7 is readable from those 33 bytes alone.
        let from_bytes = IndexRow {
            ordinal: 7,
            offset: u64::from_le_bytes(row7[0..8].try_into().unwrap()),
            len: u64::from_le_bytes(row7[8..16].try_into().unwrap()),
            obj_type: ObjType::from_code(row7[16]).unwrap(),
            uncompressed_size: u64::from_le_bytes(row7[17..25].try_into().unwrap()),
            delta_base: u64::from_le_bytes(row7[25..33].try_into().unwrap()),
        };
        assert_eq!(
            from_bytes,
            c.row_at(7),
            "one 33-byte slice must carry the whole row"
        );
    }

    /// The ordinal is the oid-lexicographic rank in **all three** arms, and it
    /// is the join key `FourTables` relies on. Asserted against an
    /// independently sorted copy of the input rather than against another arm.
    ///
    /// Seen RED by making `oid_order` sort by `offset` instead of by `oid`:
    /// "A ordinal is not the rank — left: 206, right: 0" on the first row.
    /// Restored.
    #[test]
    fn ordinals_are_the_oid_lexicographic_rank_in_every_arm() {
        let entries = synthetic_entries(300, 20, 7);
        let (a, b, c) = arms(&entries);
        let mut sorted: Vec<&IndexEntry> = entries.iter().collect();
        sorted.sort_by(|x, y| x.oid.cmp(&y.oid));
        for (rank, e) in sorted.iter().enumerate() {
            let ra = a.lookup(&e.oid).unwrap();
            let rb = b.lookup(&e.oid).unwrap();
            let rc = c.lookup(&e.oid).unwrap();
            assert_eq!(ra.ordinal as usize, rank, "A ordinal is not the rank");
            assert_eq!(rb.ordinal as usize, rank, "B ordinal is not the rank");
            assert_eq!(rc.ordinal as usize, rank, "C ordinal is not the rank");
            // The oid column of each arm, read at that ordinal, is that oid.
            assert_eq!(a.oid_column().value(rank), e.oid.as_slice());
            assert_eq!(b.oid_column().value(rank), e.oid.as_slice());
            assert_eq!(c.oid_column().value(rank), e.oid.as_slice());
        }
    }

    /// Zero-copy, asserted by pointer containment rather than by hope, in all
    /// three arms.
    ///
    /// `decode_ipc` uses `with_require_alignment(true)`, so a misaligned buffer
    /// is an error, not a silent re-allocation. This proves the positive side:
    /// each array's bytes are *inside* the IPC buffer it came from.
    ///
    /// Seen RED by swapping `decode_ipc_buffer`'s `StreamDecoder` for the
    /// high-level `StreamReader` over a `Cursor` (which owns its own
    /// allocations): "FourTables columns were copied out of IPC". Restored.
    ///
    /// A weaker edit did **not** turn it red and is worth recording: replacing
    /// the `MutableBuffer` copy with `Buffer::from_vec(bytes.to_vec())` left the
    /// test green, because glibc happened to hand back an 8-aligned allocation
    /// for a ~57 kB `Vec`. Alignment by luck is not alignment, which is what
    /// [`require_alignment_is_load_bearing`] pins down separately.
    #[test]
    fn every_column_is_a_zero_copy_view_into_its_ipc_section() {
        let entries = synthetic_entries(1000, 32, 11);
        let (a, b, c) = arms(&entries);
        assert!(a.column_is_inside_ipc(), "FourTables columns were copied out of IPC");
        assert!(b.column_is_inside_ipc(), "OneTable columns were copied out of IPC");
        assert!(c.column_is_inside_ipc(), "PackedPayload columns were copied out of IPC");

        // And the payload really is there: 1000 × 32 oid bytes + 1000 × 8 for
        // each of offset/len/size/delta_base + 1000 × 1 type = 65000 bytes
        // minimum.
        assert!(
            b.ipc_bytes() >= 65_000,
            "one-table section is {} bytes, too small to hold the payload",
            b.ipc_bytes()
        );
        assert!(a.ipc_bytes() >= 65_000);
        assert!(c.ipc_bytes() >= 65_000);
    }

    /// `with_require_alignment(true)` is the thing that makes "zero-copy" a
    /// contract instead of a hope: with it off, arrow silently allocates a fresh
    /// aligned buffer and copies, and the only symptom is that opening a large
    /// index got slower and doubled its peak memory.
    ///
    /// The same IPC bytes are decoded twice — once from a 64-byte-aligned
    /// buffer, once from the identical bytes sitting at offset 4 of the same
    /// allocation. The first must succeed and be a view; the second must be
    /// **refused**.
    ///
    /// Seen RED by dropping `.with_require_alignment(true)` from
    /// `decode_ipc_buffer`: the misaligned decode returned `Ok` and the test
    /// panicked with "a misaligned buffer must be refused, not silently
    /// copied". Restored.
    #[test]
    fn require_alignment_is_load_bearing() {
        let entries = synthetic_entries(64, 20, 77);
        let order = oid_order(&entries);
        let c = columns(&entries, &order, 20);
        let batch = RecordBatch::try_new(
            extent_schema(),
            vec![Arc::new(c.offset), Arc::new(c.len), Arc::new(c.delta_base)],
        )
        .unwrap();
        let ipc = to_ipc(&batch).unwrap();

        // Aligned: 64 bytes of padding then the stream, sliced back to the
        // stream's start — the pointer is 64-aligned.
        let mut mb = MutableBuffer::with_capacity(64 + ipc.len());
        mb.extend_from_slice(&[0u8; 64]);
        mb.extend_from_slice(&ipc);
        let padded: Buffer = mb.into();
        let (owner, ok) = decode_ipc_buffer(padded.slice(64)).expect("aligned decode must work");
        assert_eq!(ok.num_rows(), 64, "the aligned decode must yield all 64 rows");
        assert!(inside(&owner, ok.column(0).to_data().buffers()[0].as_slice()));

        // Misaligned by 4: byte-identical stream, 4-byte-offset pointer.
        let mut mb = MutableBuffer::with_capacity(4 + ipc.len());
        mb.extend_from_slice(&[0u8; 4]);
        mb.extend_from_slice(&ipc);
        let skewed: Buffer = mb.into();
        let err = match decode_ipc_buffer(skewed.slice(4)) {
            Ok(_) => panic!("a misaligned buffer must be refused, not silently copied"),
            Err(e) => e.to_string(),
        };
        assert!(
            err.contains("Misaligned"),
            "the refusal must name the misalignment, got: {err}"
        );
    }

    /// Four sections cost strictly more IPC framing than one, and the amount is
    /// asserted as a byte count, not as "greater than".
    ///
    /// Each Arrow IPC stream carries a schema message, a record-batch message
    /// header and an end-of-stream marker. Three extra copies of that is the
    /// entire structural overhead of layout A, and on a tiny index it dominates.
    ///
    /// Seen RED by asserting `>= 900`: "expected ≥900 bytes of extra framing
    /// for three extra sections, got 728 (A=4704 B=3976)". The bound below is
    /// the one the measurement supports — 728 bytes, ~243 per extra section,
    /// which is the schema + batch-header + EOS framing and not payload.
    #[test]
    fn four_sections_pay_three_extra_ipc_frames() {
        let entries = synthetic_entries(64, 20, 3);
        let (a, b, _) = arms(&entries);
        let extra = a.ipc_bytes() as i64 - b.ipc_bytes() as i64;
        assert!(
            (600..1200).contains(&extra),
            "expected 600..1200 bytes of extra framing for three extra sections, got {extra} \
             (A={} B={})",
            a.ipc_bytes(),
            b.ipc_bytes()
        );
        // The four sections are separately sized and the oid one is the biggest.
        let lens = a.ipc_section_lens();
        assert_eq!(lens.len(), 4);
        assert!(lens[0] > lens[2], "oid section {} must exceed type section {}", lens[0], lens[2]);
    }

    /// The packed arm stores the same *payload* bytes — 33 per row against
    /// 8 + 8 + 1 + 8 + 8 — so nothing in the timing comparison is a
    /// memory-footprint effect in disguise. Its IPC **section** is nonetheless
    /// smaller, and the amount is asserted as a scaling law rather than as a
    /// constant, at two sizes, so it is attributed rather than hand-waved.
    ///
    /// Four fewer columns is four fewer per-column IPC buffers, and the
    /// saving grows linearly with row count at about `4 · n/8` bytes. It was
    /// `3 · n/8` — 38 016 B at 100 000 rows — before `delta_base` gave the
    /// columnar arm a fifth payload column. That is the only footprint
    /// difference between the three arms.
    ///
    /// Seen RED by asserting the two sections were within 4096 bytes of each
    /// other: "packed payload is 4525512 bytes against 4563528 columnar", which
    /// is what sent me to count buffers instead of guessing. Seen RED a second
    /// time by asserting the saving was *constant* across the two sizes:
    /// "saving did not scale: 38016 at 100000 rows, 75456 at 200000". Seen RED
    /// a third time by leaving the `3 · n/8` bound in place after the fifth
    /// column landed: "at 100000 rows the saving is 50560 B, not the ~37500 B
    /// that four fewer columns accounts for" — 50 560 B is `4 · n/8` plus
    /// framing, so the bound is counting real buffers rather than tracking a
    /// moving target.
    #[test]
    fn the_packed_arm_holds_the_same_payload_but_a_smaller_section() {
        let mut savings = Vec::new();
        for &n in &[100_000usize, 200_000] {
            let entries = synthetic_entries(n, 20, 0xBEE5);
            let (_, b, c) = arms(&entries);

            // The payload itself is byte-for-byte the same size.
            let payload_bytes = n * (8 + 8 + 1 + 8 + 8);
            assert_eq!(c.payload.value_data().len(), payload_bytes);
            assert_eq!(c.payload.value_data().len(), n * PACKED_LEN);

            let saving = b.ipc_bytes() as i64 - c.ipc_bytes() as i64;
            assert!(
                saving > 0,
                "the two-column section ({}) must not be larger than the six-column one ({})",
                c.ipc_bytes(),
                b.ipc_bytes()
            );
            // Four fewer columns, at roughly n/8 bytes of per-column buffer each.
            let expect = 4 * (n as i64) / 8;
            assert!(
                (expect..expect + 4096).contains(&saving),
                "at {n} rows the saving is {saving} B, not the ~{expect} B that four fewer \
                 columns accounts for"
            );
            savings.push(saving);
        }
        assert!(
            savings[1] > savings[0] * 3 / 2,
            "saving did not scale: {} at 100000 rows, {} at 200000",
            savings[0],
            savings[1]
        );
    }

    /// LAW 2 — the 8-byte prefix collision, carried through all three arms.
    ///
    /// Two oids sharing their first eight bytes share one stree key, so the tree
    /// alone cannot tell them apart. Every arm must still return each oid's own
    /// four facts, and a third oid on the same prefix that was never inserted
    /// must miss.
    ///
    /// Seen RED by replacing `OidResolver::ordinal`'s verified `tree.lookup`
    /// with the unverified `tree.candidate_run(key_for_oid(oid)).start` — i.e.
    /// trusting the 8-byte prefix: "collision resolved to the wrong row —
    /// left: 1000, right: 2000". Both colliding oids came back as the first
    /// candidate. Restored.
    #[test]
    fn a_prefix_collision_resolves_to_the_right_row_in_every_arm() {
        let prefix = [0xde, 0xad, 0xbe, 0xef, 0x01, 0x02, 0x03, 0x04];
        // The two colliding entries carry *different* delta bases, so a column
        // read at the wrong ordinal shows up here too and not only in `offset`.
        let mk = |last: u8, off: u64, ty: ObjType| {
            let mut oid = vec![0u8; 20];
            oid[..8].copy_from_slice(&prefix);
            oid[8] = last;
            IndexEntry {
                oid,
                offset: off,
                len: 10,
                obj_type: ty,
                uncompressed_size: off * 2,
                delta_base: off / 2,
            }
        };
        let mut entries = vec![mk(0xaa, 1000, ObjType::Commit), mk(0xbb, 2000, ObjType::OfsDelta)];
        entries.extend(synthetic_entries(200, 20, 99));
        let (a, b, c) = arms(&entries);

        for (last, off, ty) in [(0xaau8, 1000u64, ObjType::Commit), (0xbb, 2000, ObjType::OfsDelta)]
        {
            let oid = mk(last, off, ty).oid;
            let ra = a.lookup(&oid).expect("colliding oid must resolve in A");
            let rb = b.lookup(&oid).expect("colliding oid must resolve in B");
            let rc = c.lookup(&oid).expect("colliding oid must resolve in C");
            assert_eq!(ra, rb);
            assert_eq!(ra, rc);
            assert_eq!(ra.offset, off, "collision resolved to the wrong row");
            assert_eq!(ra.obj_type, ty);
            assert_eq!(ra.delta_base, off / 2, "collision resolved to the wrong delta base");
        }
        let never = mk(0xcc, 0, ObjType::Blob).oid;
        assert_eq!(a.lookup(&never), None, "unstored oid on a colliding prefix must miss");
        assert_eq!(b.lookup(&never), None);
        assert_eq!(c.lookup(&never), None);
    }

    /// All six pack type codes survive the round trip in every arm, including
    /// the two delta kinds that are the reason this fact is stored at all.
    ///
    /// Seen RED by mapping `ObjType::RefDelta` to code 6 in `code()`, colliding
    /// with `OfsDelta`: "type did not survive the column — left: OfsDelta,
    /// right: RefDelta". Restored.
    #[test]
    fn all_six_pack_types_round_trip_including_the_deltas() {
        let entries: Vec<IndexEntry> = ObjType::ALL
            .iter()
            .enumerate()
            .map(|(i, &t)| {
                let mut oid = vec![0u8; 32];
                oid[0] = i as u8 * 17;
                oid[31] = i as u8;
                IndexEntry {
                    oid,
                    offset: 100 + i as u64,
                    len: 5,
                    obj_type: t,
                    uncompressed_size: 900 + i as u64,
                    // Only the two delta codes carry a base, which is the
                    // relationship the type column exists to express.
                    delta_base: match t {
                        ObjType::OfsDelta => 100,
                        _ => 0,
                    },
                }
            })
            .collect();
        let (a, b, c) = arms(&entries);
        let mut seen: Vec<ObjType> = Vec::new();
        for e in &entries {
            let ra = a.lookup(&e.oid).unwrap();
            assert_eq!(ra, b.lookup(&e.oid).unwrap());
            assert_eq!(ra, c.lookup(&e.oid).unwrap());
            assert_eq!(ra.obj_type, e.obj_type, "type did not survive the column");
            seen.push(ra.obj_type);
        }
        assert_eq!(seen, ObjType::ALL.to_vec(), "all six codes must be distinct");
        assert_eq!(ObjType::from_code(0), None);
        assert_eq!(ObjType::from_code(5), None, "git leaves 5 unused; it must not be mapped");
        assert_eq!(ObjType::from_code(8), None);
    }

    /// An empty index and a one-row index are clean, not panics, in every arm.
    ///
    /// Seen RED by replacing `validate`'s empty-input `Ok(GitHashKind::Sha1)`
    /// with `bail!("an empty index")`: "A builds: an empty index" — every arm
    /// failed to construct at all. Restored.
    #[test]
    fn degenerate_sizes_are_clean_in_every_arm() {
        let (a, b, c) = arms(&[]);
        assert!(a.is_empty() && b.is_empty() && c.is_empty());
        let probe = vec![7u8; 20];
        assert_eq!(a.lookup(&probe), None);
        assert_eq!(b.lookup(&probe), None);
        assert_eq!(c.lookup(&probe), None);
        assert_eq!(c.lookup_batch(&[&probe[..]]), vec![None]);
        assert_eq!(c.extents_batch(&[&probe[..]]), vec![None]);
        // A scan over an empty index is 0, not a panic or a wrap.
        assert_eq!(c.sum_uncompressed(), 0);
        assert_eq!(c.count_type(ObjType::Blob), 0);

        let one = synthetic_entries(1, 20, 5);
        let (a, b, c) = arms(&one);
        assert_eq!(a.lookup(&one[0].oid).unwrap(), b.lookup(&one[0].oid).unwrap());
        assert_eq!(a.lookup(&one[0].oid).unwrap(), c.lookup(&one[0].oid).unwrap());
        assert_eq!(c.lookup(&one[0].oid).unwrap().offset, one[0].offset);
        assert_eq!(c.sum_uncompressed(), one[0].uncompressed_size);
    }

    /// A wrong-width oid is a miss, never a panic and never a wrong row —
    /// the query side of P-4.
    ///
    /// The width check inside `oid_index::lookup` is only a fast path; what
    /// actually makes this true is the **full-oid compare**, because
    /// `key_for_oid` zero-pads and an 8-byte truncation of a stored oid
    /// therefore lands on that oid's key. Seen RED by the same
    /// unverified-`candidate_run` edit as
    /// [`a_prefix_collision_resolves_to_the_right_row_in_every_arm`]:
    /// "a 8-byte oid must miss — left: Some(IndexRow { ordinal: 47, offset: 12,
    /// len: 1505, obj_type: RefDelta, uncompressed_size: 3010 }), right: None".
    /// Restored.
    #[test]
    fn a_wrong_width_oid_misses_in_every_arm() {
        let entries = synthetic_entries(64, 20, 21);
        let (a, b, c) = arms(&entries);
        let short = &entries[0].oid[..8];
        let mut long = entries[0].oid.clone();
        long.extend_from_slice(&[0u8; 12]);
        for q in [short, &long[..]] {
            assert_eq!(a.lookup(q), None, "a {}-byte oid must miss", q.len());
            assert_eq!(b.lookup(q), None);
            assert_eq!(c.lookup(q), None);
        }
        assert_eq!(a.lookup_batch(&[short, &long]), vec![None, None]);
        assert_eq!(b.lookup_batch(&[short, &long]), vec![None, None]);
        assert_eq!(c.lookup_batch(&[short, &long]), vec![None, None]);
    }

    /// A duplicate oid, or a mix of sha1 and sha256, is refused at build.
    /// Silently keeping one of two identical oids would give the arms different
    /// ordinals for the same object and make every later comparison a lie.
    ///
    /// Seen RED by short-circuiting the duplicate scan in `validate` with
    /// `if false && …`: the build succeeded and the test panicked with
    /// "a duplicate oid must be refused at build". Restored.
    #[test]
    fn build_refuses_duplicates_and_mixed_widths() {
        fn why<T>(r: Result<T>, what: &str) -> String {
            match r {
                Ok(_) => panic!("{what} must be refused at build"),
                Err(e) => e.to_string(),
            }
        }
        let mut dup = synthetic_entries(4, 20, 1);
        dup.push(dup[0].clone());
        let err = why(FourTables::build(&dup), "a duplicate oid");
        assert!(err.contains("duplicate oid"), "{err}");
        assert!(OneTableFourColumns::build(&dup).is_err());
        assert!(PackedPayload::build(&dup).is_err());

        let mut mixed = synthetic_entries(4, 20, 2);
        mixed.push(synthetic_entries(1, 32, 3).pop().unwrap());
        let err = why(PackedPayload::build(&mixed), "a mixed-width index");
        assert!(err.contains("mixed oid widths"), "{err}");
        assert!(FourTables::build(&mixed).is_err());
        assert!(OneTableFourColumns::build(&mixed).is_err());
    }

    /// Every arm is `Send + Sync`, which is what lets a server share one index
    /// across connection handlers. A compile-time assertion, plus an actual
    /// cross-thread lookup so it is not only a type-level claim.
    ///
    /// Seen RED by dropping `Send + Sync` from the `ObjectIndex` supertrait
    /// list: "error[E0277]: `dyn index_layout::ObjectIndex` cannot be shared
    /// between threads safely" and the same for `Send` — the crate stopped
    /// compiling. Restored.
    #[test]
    fn an_index_can_be_shared_across_threads() {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<FourTables>();
        assert_send_sync::<OneTableFourColumns>();
        assert_send_sync::<PackedPayload>();

        let entries = synthetic_entries(256, 20, 42);
        let a: Arc<dyn ObjectIndex> = Arc::new(FourTables::build(&entries).unwrap());
        let b: Arc<dyn ObjectIndex> = Arc::new(OneTableFourColumns::build(&entries).unwrap());
        let c: Arc<dyn ObjectIndex> = Arc::new(PackedPayload::build(&entries).unwrap());
        let oids: Vec<Vec<u8>> = entries.iter().map(|e| e.oid.clone()).collect();
        let mut handles = Vec::new();
        for idx in [a, b, c] {
            let oids = oids.clone();
            handles.push(std::thread::spawn(move || {
                let refs: Vec<&[u8]> = oids.iter().map(|o| o.as_slice()).collect();
                idx.lookup_batch(&refs).iter().filter(|r| r.is_some()).count()
            }));
        }
        for h in handles {
            assert_eq!(h.join().unwrap(), 256, "every oid must resolve off-thread");
        }
    }
}