src/stree_log.rs
//! Ragnar-fast: a static search-tree index over the warehouse commit/version log.
//!
//! The warehouse is an append-only, SHA-keyed log: every fact is a row tagged
//! with a monotone key (a snapshot id / `ts_micros` version). The two questions
//! a time-traveller asks are
//!
//! * **exact** — "the row *for* version `V`" (`at_exact`), and
//! * **as-of / point-in-time** — "the row *in effect at* version/time `V`",
//! i.e. the greatest key `<= V` (`as_of`) — the predecessor / floor.
//!
//! The naive path answers both with a linear scan of the whole log
//! ([`LinearScan`]). This module answers them in `O(log n)` with a **static
//! search tree**, immutable once built and read lock-free — the same shape
//! [`skade`'s snapshot-id `StaticIndex`] uses, here generalised to carry an
//! arbitrary payload and to serve the *floor* (as-of) query the linear scan does.
//!
//! Two Ragnar-Groot-Koerkamp static trees back it:
//!
//! * **exact** hits reuse [`znippy_zoomies::stree::STree64`] verbatim — Ragnar's
//! cache-line-packed implicit S(+)-tree, a handful of branchless SIMD probes
//! with no hashing and no locks. (See that module's header for the full
//! attribution — the algorithm is entirely Ragnar's.)
//! * **as-of** (floor) hits descend an **Eytzinger-layout** array of the same
//! sorted keys: the in-order = sorted BST packed breadth-first, so a floor
//! descent is a straight-line loop of cache-friendly, prefetchable probes.
//! `STree64` only reports *exact* membership, so the predecessor variant the
//! as-of query needs is built here rather than bolted onto the dependency.
//!
//! Correctness law: keys are immutable version ids; the index is built once from
//! a consistent read of the log and replaced wholesale (never mutated in place),
//! so a lookup can only be *stale* (miss a newer row), never *wrong*. When a key
//! repeats (a superseding append), **last writer wins** — the entry with the
//! greatest original position in the input — matching the warehouse's
//! append-only "a new row supersedes" rule. [`LinearScan`] resolves ties the
//! same way, so the two agree row-for-row (the differential test below).
use std::sync::Arc;
use crate::stree::STree64;
/// Immutable, read-lock-free static index over a version/commit log keyed by an
/// ascending `i64` (a snapshot id or `ts_micros`). Built once, queried many
/// times; a rebuild replaces the whole structure.
pub struct RagnarLog<V> {
/// Sorted, de-duplicated keys (ascending), parallel to `entries`.
keys: Arc<[i64]>,
/// Payloads in the same order as `keys`.
entries: Arc<[V]>,
/// Ragnar `STree64` over `keys` for `O(log n)` **exact** membership. `None`
/// only for the empty log (which answers every query `None`).
exact: Option<STree64>,
/// Eytzinger (breadth-first BST) layout of `keys`, 1-indexed (slot 0 unused),
/// for the branchless **floor** descent that serves `as_of`.
eytz_keys: Vec<i64>,
/// `eytz_slot -> index into keys/entries` for the same node.
eytz_idx: Vec<u32>,
min_key: i64,
max_key: i64,
}
impl<V> std::fmt::Debug for RagnarLog<V> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RagnarLog")
.field("len", &self.entries.len())
.field("min_key", &self.min_key)
.field("max_key", &self.max_key)
.finish()
}
}
impl<V> RagnarLog<V> {
/// Build the index from `(key, payload)` rows in any order. Sorts ascending
/// by key and collapses duplicate keys **last-writer-wins** (the row with the
/// greatest original position survives). Returns an empty index for empty
/// input (every query then answers `None`).
pub fn build(rows: Vec<(i64, V)>) -> Self {
// Tag with original position, then sort by (key asc, orig desc) so that
// `dedup_by_key` — which keeps the *first* of each run — keeps the
// greatest original position: last writer wins.
let mut tagged: Vec<(i64, usize, V)> = rows
.into_iter()
.enumerate()
.map(|(i, (k, v))| (k, i, v))
.collect();
tagged.sort_by(|a, b| a.0.cmp(&b.0).then(b.1.cmp(&a.1)));
tagged.dedup_by_key(|t| t.0);
let keys: Vec<i64> = tagged.iter().map(|t| t.0).collect();
let entries: Vec<V> = tagged.into_iter().map(|t| t.2).collect();
if keys.is_empty() {
return Self {
keys: Arc::from(Vec::new()),
entries: Arc::from(Vec::new()),
exact: None,
eytz_keys: vec![i64::MIN], // slot 0 only
eytz_idx: vec![0],
min_key: 0,
max_key: 0,
};
}
let min_key = keys[0];
let max_key = *keys.last().unwrap();
let exact = Some(STree64::new(&keys));
let (eytz_keys, eytz_idx) = build_eytzinger(&keys);
Self {
keys: keys.into(),
entries: entries.into(),
exact,
eytz_keys,
eytz_idx,
min_key,
max_key,
}
}
/// Number of distinct keys in the log.
#[inline]
pub fn len(&self) -> usize {
self.entries.len()
}
/// True when the log is empty.
#[inline]
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
/// Lowest key covered (meaningless when empty).
#[inline]
pub fn min_key(&self) -> i64 {
self.min_key
}
/// Highest key covered (meaningless when empty).
#[inline]
pub fn max_key(&self) -> i64 {
self.max_key
}
/// **Exact** lookup: the payload written *for* version `key`, or `None`.
/// Routes through Ragnar's `STree64` (`O(log n)`, branchless SIMD probes).
#[inline]
pub fn at_exact(&self, key: i64) -> Option<&V> {
let idx = self.exact.as_ref()?.find_exact(key)?;
debug_assert_eq!(self.keys.get(idx).copied(), Some(key));
self.entries.get(idx)
}
/// **As-of** (point-in-time) lookup: the payload in effect at version `key`,
/// i.e. the entry with the greatest key `<= key` (the floor / predecessor),
/// or `None` when `key` precedes the whole log. Eytzinger floor descent —
/// `O(log n)` cache-friendly probes.
#[inline]
pub fn as_of(&self, key: i64) -> Option<&V> {
let sorted = self.floor_index(key)?;
self.entries.get(sorted)
}
/// The number of entries whose key is `<= key` — the replay **cut** for an
/// as-of-time fold (events `0..boundary_leq(t)` are those at or before `t`).
/// `O(log n)` via the same floor descent.
#[inline]
pub fn boundary_leq(&self, key: i64) -> usize {
match self.floor_index(key) {
Some(i) => i + 1,
None => 0,
}
}
/// Index into `keys`/`entries` of the floor (greatest key `<= q`), or `None`.
#[inline]
fn floor_index(&self, q: i64) -> Option<usize> {
let n = self.keys.len();
if n == 0 || q < self.min_key {
return None;
}
let mut k = 1usize;
let mut best = u32::MAX;
while k <= n {
// Prefetch both children's cache line before we branch — the classic
// Eytzinger latency-hiding trick (both are one load away regardless
// of which way we go).
prefetch(&self.eytz_keys, 2 * k);
if self.eytz_keys[k] <= q {
best = self.eytz_idx[k];
k = 2 * k + 1; // look right for a bigger key still <= q
} else {
k = 2 * k; // too big; look left
}
}
(best != u32::MAX).then_some(best as usize)
}
/// The sorted key/entry pair at sorted index `i` — for callers that want to
/// slice the log around an `as_of`/`boundary_leq` answer.
#[inline]
pub fn get(&self, i: usize) -> Option<(i64, &V)> {
Some((*self.keys.get(i)?, self.entries.get(i)?))
}
/// The de-duplicated, ascending key slice (for range slicing by a caller).
#[inline]
pub fn keys(&self) -> &[i64] {
&self.keys
}
/// **Batched as-of** — the whole reason this type lives beside `STree64`.
///
/// One floor descent at a time serialises on memory: every level is a
/// dependent load, so the CPU stalls on a cache miss it cannot overlap with
/// anything. `P` *independent* descents advanced in lockstep turn those
/// stalls into parallelism — the same software-pipelining
/// [`crate::stree::STree64::lookup_batch_pipeline`] applies to the exact
/// lookup, here for the predecessor query.
///
/// `P` is a tuning knob, not a constant of nature; ~16–32 is the usual
/// plateau and the bench sweep decides. Returns one answer per query, in
/// query order.
pub fn as_of_batch<const P: usize>(&self, queries: &[i64]) -> Vec<Option<&V>> {
self.floor_index_batch::<P>(queries)
.into_iter()
.map(|i| i.and_then(|i| self.entries.get(i)))
.collect()
}
/// Batched [`boundary_leq`](Self::boundary_leq) — the replay cut per query.
pub fn boundary_leq_batch<const P: usize>(&self, queries: &[i64]) -> Vec<usize> {
self.floor_index_batch::<P>(queries)
.into_iter()
.map(|i| i.map_or(0, |i| i + 1))
.collect()
}
/// Batched [`at_exact`](Self::at_exact).
///
/// Note this does **not** yet route through `STree64`'s pipelined batch: that
/// API is `(&[i64], &[u8]) -> Vec<Option<usize>>` over an mmap'd payload
/// region, while this log holds its entries as a plain `Arc<[V]>`. Wiring the
/// two is worth doing and is deliberately left as a separate step rather than
/// faked here — an unbatched loop that *claims* to be batched would be worse
/// than one that says it is not.
pub fn at_exact_batch(&self, queries: &[i64]) -> Vec<Option<&V>> {
queries.iter().map(|&q| self.at_exact(q)).collect()
}
/// `P` independent Eytzinger floor descents advanced in lockstep.
///
/// Each iteration prefetches every live descent's next cache line *before*
/// any of them branch, so the misses overlap instead of queueing. The tree is
/// a complete BST of fixed depth, so all descents finish in the same number
/// of steps and no per-lane bookkeeping is needed.
fn floor_index_batch<const P: usize>(&self, queries: &[i64]) -> Vec<Option<usize>> {
let n = self.keys.len();
let mut out = Vec::with_capacity(queries.len());
if n == 0 {
out.resize(queries.len(), None);
return out;
}
for chunk in queries.chunks(P.max(1)) {
let mut k = [1usize; P];
let mut best = [u32::MAX; P];
// Depth of the implicit complete tree: every lane needs the same
// number of steps, so one shared loop bound is exact, not a bound.
let depth = usize::BITS - n.leading_zeros();
for _ in 0..depth {
for (lane, _) in chunk.iter().enumerate() {
if k[lane] <= n {
prefetch(&self.eytz_keys, 2 * k[lane]);
}
}
for (lane, &q) in chunk.iter().enumerate() {
if k[lane] > n {
continue;
}
if self.eytz_keys[k[lane]] <= q {
best[lane] = self.eytz_idx[k[lane]];
k[lane] = 2 * k[lane] + 1;
} else {
k[lane] = 2 * k[lane];
}
}
}
for (lane, &q) in chunk.iter().enumerate() {
out.push(
(q >= self.min_key && best[lane] != u32::MAX).then_some(best[lane] as usize),
);
}
}
out
}
}
/// Recursively fill an Eytzinger (breadth-first complete-BST) layout from a
/// sorted slice: an in-order walk of the implicit tree visits `sorted` in order.
/// Returns `(eytz_keys, eytz_idx)`, both 1-indexed (slot 0 is an unused sentinel).
fn build_eytzinger(sorted: &[i64]) -> (Vec<i64>, Vec<u32>) {
let n = sorted.len();
let mut ek = vec![i64::MIN; n + 1];
let mut ei = vec![0u32; n + 1];
let mut pos = 0usize;
fill(1, n, sorted, &mut ek, &mut ei, &mut pos);
(ek, ei)
}
fn fill(k: usize, n: usize, sorted: &[i64], ek: &mut [i64], ei: &mut [u32], pos: &mut usize) {
if k > n {
return;
}
fill(2 * k, n, sorted, ek, ei, pos);
ek[k] = sorted[*pos];
ei[k] = *pos as u32;
*pos += 1;
fill(2 * k + 1, n, sorted, ek, ei, pos);
}
/// Prefetch the cache line holding `slot` (and its sibling) into L1. No-op off
/// x86_64 or when the slot is out of range; purely a latency hint.
#[inline(always)]
fn prefetch(eytz: &[i64], slot: usize) {
#[cfg(target_arch = "x86_64")]
{
if slot < eytz.len() {
// SAFETY: bounds-checked; `_mm_prefetch` only reads TLB/cache state.
unsafe {
use std::arch::x86_64::{_MM_HINT_T0, _mm_prefetch};
_mm_prefetch::<_MM_HINT_T0>(eytz.as_ptr().add(slot) as *const i8);
}
}
}
#[cfg(not(target_arch = "x86_64"))]
{
let _ = (eytz, slot);
}
}
/// The naive baseline: keeps the raw log and answers every query with a full
/// linear scan. This is the path the static index replaces; it exists so the
/// fast path can be proven correct against it (differential test) and measured
/// against it (HEAVY bench). Ties resolve **last-writer-wins**, identically to
/// [`RagnarLog::build`], so the two agree row-for-row.
pub struct LinearScan<V> {
rows: Vec<(i64, usize, V)>, // (key, original position, payload)
}
impl<V> LinearScan<V> {
/// Wrap the raw `(key, payload)` rows — no sorting, no index.
pub fn build(rows: Vec<(i64, V)>) -> Self {
let rows = rows
.into_iter()
.enumerate()
.map(|(i, (k, v))| (k, i, v))
.collect();
Self { rows }
}
/// Exact lookup by full scan (last writer among equal keys wins).
pub fn at_exact(&self, key: i64) -> Option<&V> {
let mut best: Option<&(i64, usize, V)> = None;
for r in &self.rows {
if r.0 == key && best.map(|b| r.1 > b.1).unwrap_or(true) {
best = Some(r);
}
}
best.map(|r| &r.2)
}
/// As-of (floor) lookup by full scan: greatest key `<= key`, last writer wins.
pub fn as_of(&self, key: i64) -> Option<&V> {
let mut best: Option<&(i64, usize, V)> = None;
for r in &self.rows {
if r.0 > key {
continue;
}
let take = match best {
None => true,
Some(b) => r.0 > b.0 || (r.0 == b.0 && r.1 > b.1),
};
if take {
best = Some(r);
}
}
best.map(|r| &r.2)
}
/// Count of distinct keys `<= key` — the replay cut, by scan.
pub fn boundary_leq(&self, key: i64) -> usize {
let mut seen: Vec<i64> = self
.rows
.iter()
.filter(|r| r.0 <= key)
.map(|r| r.0)
.collect();
seen.sort_unstable();
seen.dedup();
seen.len()
}
}
#[cfg(test)]
mod tests {
use super::*;
// A small deterministic pseudo-random generator (no dev-dep needed).
struct Rng(u64);
impl Rng {
fn next(&mut self) -> u64 {
// xorshift64*
let mut x = self.0;
x ^= x >> 12;
x ^= x << 25;
x ^= x >> 27;
self.0 = x;
x.wrapping_mul(0x2545_F491_4F6C_DD1D)
}
fn below(&mut self, n: i64) -> i64 {
(self.next() % n as u64) as i64
}
}
#[test]
fn empty_log_answers_none() {
let idx: RagnarLog<u32> = RagnarLog::build(vec![]);
assert!(idx.is_empty());
assert_eq!(idx.at_exact(5), None);
assert_eq!(idx.as_of(5), None);
assert_eq!(idx.boundary_leq(5), 0);
}
/// The batched floor descent must answer **exactly** what the one-at-a-time
/// descent answers — same values, same order, for every query including ones
/// below the log, above it, on a key, and between keys. A batched fast path
/// that silently differs from its scalar twin is the failure this
/// constellation keeps deleting, so it is pinned rather than trusted.
///
/// Sweeps several `P` to prove the answer does not depend on the lane count,
/// and uses a query count that is deliberately NOT a multiple of any `P` so
/// the ragged final chunk is exercised.
#[test]
fn batched_as_of_matches_scalar_as_of() {
let rows: Vec<(i64, String)> = (0..500i64).map(|i| (i * 7, format!("v{i}"))).collect();
let idx = RagnarLog::build(rows);
// Below the log, on keys, between keys, and past the end — 1001 queries,
// coprime-ish with every P below so the last chunk is always partial.
let queries: Vec<i64> = (-3..998i64).map(|q| q * 3 - 1).collect();
assert_eq!(queries.len(), 1001);
let scalar: Vec<Option<&String>> = queries.iter().map(|&q| idx.as_of(q)).collect();
let cuts: Vec<usize> = queries.iter().map(|&q| idx.boundary_leq(q)).collect();
assert_eq!(
idx.as_of_batch::<1>(&queries),
scalar,
"P=1 differs from scalar"
);
assert_eq!(
idx.as_of_batch::<4>(&queries),
scalar,
"P=4 differs from scalar"
);
assert_eq!(
idx.as_of_batch::<16>(&queries),
scalar,
"P=16 differs from scalar"
);
assert_eq!(
idx.as_of_batch::<32>(&queries),
scalar,
"P=32 differs from scalar"
);
assert_eq!(
idx.boundary_leq_batch::<16>(&queries),
cuts,
"batched cut differs"
);
// Non-vacuity: the sweep must contain both misses and hits, or it proves
// nothing. (A test that only ever sees `None` would pass on a stub.)
assert!(
scalar.iter().any(|v| v.is_none()),
"no query fell below the log"
);
assert!(scalar.iter().any(|v| v.is_some()), "no query hit the log");
}
/// An empty log must not panic on the batched path either — the scalar
/// version has its own test and the batch one takes a different branch.
#[test]
fn batched_as_of_on_empty_log() {
let idx: RagnarLog<String> = RagnarLog::build(vec![]);
assert_eq!(idx.as_of_batch::<8>(&[1, 2, 3]), vec![None, None, None]);
assert_eq!(idx.boundary_leq_batch::<8>(&[1, 2, 3]), vec![0, 0, 0]);
}
#[test]
fn floor_and_exact_on_a_hand_log() {
// keys 10,20,30 → entries "a","b","c"
let idx = RagnarLog::build(vec![(30, "c"), (10, "a"), (20, "b")]);
assert_eq!(idx.len(), 3);
assert_eq!(idx.min_key(), 10);
assert_eq!(idx.max_key(), 30);
// exact
assert_eq!(idx.at_exact(10), Some(&"a"));
assert_eq!(idx.at_exact(20), Some(&"b"));
assert_eq!(idx.at_exact(30), Some(&"c"));
assert_eq!(idx.at_exact(15), None);
// as-of (floor)
assert_eq!(idx.as_of(9), None);
assert_eq!(idx.as_of(10), Some(&"a"));
assert_eq!(idx.as_of(25), Some(&"b"));
assert_eq!(idx.as_of(30), Some(&"c"));
assert_eq!(idx.as_of(99), Some(&"c"));
// boundary (replay cut)
assert_eq!(idx.boundary_leq(9), 0);
assert_eq!(idx.boundary_leq(10), 1);
assert_eq!(idx.boundary_leq(25), 2);
assert_eq!(idx.boundary_leq(999), 3);
}
#[test]
fn last_writer_wins_on_duplicate_keys() {
// key 20 written twice; the later row (position 3) must win.
let rows = vec![(10, "a"), (20, "b_old"), (30, "c"), (20, "b_new")];
let idx = RagnarLog::build(rows.clone());
assert_eq!(idx.len(), 3, "duplicate key collapses");
assert_eq!(idx.at_exact(20), Some(&"b_new"));
assert_eq!(idx.as_of(25), Some(&"b_new"));
// Linear baseline agrees.
let lin = LinearScan::build(rows);
assert_eq!(lin.at_exact(20), Some(&"b_new"));
assert_eq!(lin.as_of(25), Some(&"b_new"));
}
/// The load-bearing differential test: over a large, dense, duplicate-heavy
/// random log, the static index returns the SAME answer as the linear scan
/// for exact, as-of and boundary — across the whole query space.
#[test]
fn differential_ragnar_matches_linear() {
let mut rng = Rng(0x9E37_79B9_7F4A_7C15);
let n = 50_000;
// Keys drawn from a space smaller than n so duplicates (supersedes) abound.
let key_space = 20_000i64;
let rows: Vec<(i64, u64)> = (0..n).map(|_| (rng.below(key_space), rng.next())).collect();
let ragnar = RagnarLog::build(rows.clone());
let linear = LinearScan::build(rows);
// Probe every possible key plus out-of-range sentinels below and above.
let mut mismatches = 0u64;
for q in -3..=(key_space + 3) {
if ragnar.at_exact(q) != linear.at_exact(q) {
mismatches += 1;
}
if ragnar.as_of(q) != linear.as_of(q) {
mismatches += 1;
}
if ragnar.boundary_leq(q) != linear.boundary_leq(q) {
mismatches += 1;
}
}
assert_eq!(mismatches, 0, "static index diverged from linear scan");
assert_eq!(ragnar.min_key(), 0.max(ragnar.min_key()));
}
/// Sparse, wide-key log (snapshot-id style, no duplicates): every present key
/// hits exactly, every gap floors to its predecessor.
#[test]
fn differential_sparse_snapshot_ids() {
let mut rng = Rng(0xDEAD_BEEF_CAFE_1234);
let n = 10_000;
let mut keys: Vec<i64> = Vec::with_capacity(n);
let mut k = 1_000_000i64;
for _ in 0..n {
k += 1 + rng.below(1_000); // strictly increasing, gappy
keys.push(k);
}
let rows: Vec<(i64, i64)> = keys.iter().map(|&k| (k, k * 2)).collect();
let ragnar = RagnarLog::build(rows.clone());
let linear = LinearScan::build(rows);
for _ in 0..50_000 {
let q = 1_000_000 + rng.below(n as i64 * 1_100);
assert_eq!(ragnar.at_exact(q), linear.at_exact(q));
assert_eq!(ragnar.as_of(q), linear.as_of(q));
assert_eq!(ragnar.boundary_leq(q), linear.boundary_leq(q));
}
// Exact hits on real keys.
for &kk in keys.iter().step_by(97) {
assert_eq!(ragnar.at_exact(kk), Some(&(kk * 2)));
}
}
}