nornir 0.5.3

//! 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).

#![allow(dead_code)]

use std::sync::Arc;

use znippy_zoomies::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
    }
}

/// 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_prefetch, _MM_HINT_T0};
                _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);
    }

    #[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)));
        }
    }
}