mushroomdb-storage 0.6.7

//! View seams: overlay-over-base read paths for topology and columns.
//!
//! `TopologyView` is wired into `GraphView.topo` so all topology reads
//! transparently consult overlay then base.  `ColumnsView` is wired into
//! `GraphView.props` (Task 2) so column reads also consult overlay then base.
//!
//! **Overlay/archive asymmetry for vectors**: `ColumnsView::vector()` returns
//! a zero-copy `&[f64]` slice for archived base data only.  Overlay vectors
//! (written after the last snapshot) live as `Value::List([Value::Float, ...])`
//! in the owned `ColumnStore` and are returned by `ColumnsView::get()` as
//! `ValueRef::Owned(Value::List(...))`.  Callers that need vectors from both
//! sources must fall back to `get()` when `vector()` returns `None`.

// The zero-copy f64 transmute in `ColumnsView::vector` is only sound on
// little-endian targets where `rkyv::Archived<f64>` == `rend::F64_le` has the
// same wire representation as `f64`.  Fail at compile time on BE targets.
#[cfg(not(target_endian = "little"))]
compile_error!("core-storage v8 seam: zero-copy f64 transmute requires a little-endian target");

use crate::columns::{ColumnHandle, ColumnStore};
use crate::edge_props::EdgeProps;
use crate::topology::{Direction, Topology};
use crate::types::Value;
use crate::v8::layout::{ArchivedColumnData, ArchivedColumns, ArchivedCsr, ArchivedEdgeProps};
use std::borrow::Cow;
use std::collections::{BTreeSet, HashMap};

// ---------------------------------------------------------------------------
// TopologyView
// ---------------------------------------------------------------------------

/// Overlay-over-base topology view.
///
/// `overlay` holds post-snapshot WAL-replayed edges (always an owned
/// `Topology`). `base` is the zero-copy archived CSR from the V8 snapshot
/// mmap, or `None` when the store was opened from a V5–V7 snapshot.
///
/// All read methods consult `overlay` first, then `base`.  Writes always go
/// through the WAL into `overlay` (never through this view).
pub struct TopologyView<'a> {
    pub overlay: &'a Topology,
    pub base: Option<&'a ArchivedCsr>,
}

impl<'a> TopologyView<'a> {
    /// Construct a view backed only by owned overlay (no base — V5/V6/V7 path).
    pub fn owned(overlay: &'a Topology) -> Self {
        Self {
            overlay,
            base: None,
        }
    }

    /// Construct a view that merges an mmap'd V8 base with a WAL-replay overlay.
    ///
    /// Used when a V8 snapshot is open: the base holds the snapshot CSR and the
    /// overlay holds only post-snapshot WAL mutations.
    pub fn with_base(overlay: &'a Topology, base: &'a ArchivedCsr) -> Self {
        Self {
            overlay,
            base: Some(base),
        }
    }

    /// Sorted unique neighbors of `v` for `(etype, dir)`.
    ///
    /// When base is present the result is the sorted-unique union of:
    ///   `overlay_neighbors ∪ base_neighbors`
    ///
    /// This is correct for Task 1 (overlay starts empty on V8 open; WAL
    /// replay only adds post-snapshot edges so there is no overlap).
    /// Edge deletions from base are tracked by the overlay; see
    /// `MergedNeighbors` for the planned tombstone subtraction (Task 3).
    pub fn neighbors(&self, etype: u32, dir: Direction, v: u32) -> Cow<'a, [u32]> {
        let overlay_nbrs = self.overlay.neighbors(etype, dir, v);
        let base = match self.base {
            None => return overlay_nbrs,
            Some(b) => b,
        };

        // Locate the etype entry in the sorted archived CSR.
        let base_nbrs = base_neighbors_from_archived(base, etype, dir, v);

        if base_nbrs.is_empty() {
            return overlay_nbrs;
        }

        // Subtract overlay tombstones from base neighbors before merging.
        // When the overlay records an edge deletion for an edge that exists only
        // in the base CSR, `remove_edge` writes a tombstone into the overlay.
        // That tombstone must be filtered out here so deleted edges do not
        // reappear after a V8 snapshot open.
        let filtered_base_nbrs = subtract_tombstones(base_nbrs, etype, dir, v, self.overlay);

        if filtered_base_nbrs.is_empty() {
            return overlay_nbrs;
        }
        if overlay_nbrs.is_empty() {
            return Cow::Owned(filtered_base_nbrs);
        }
        Cow::Owned(merge_sorted_unique(
            overlay_nbrs.as_ref(),
            &filtered_base_nbrs,
        ))
    }

    /// Total edge count: (base - tombstones) + overlay.
    ///
    /// Tombstones in the overlay represent edges that were present only in the
    /// base CSR and were subsequently deleted via `remove_edge`.  They must be
    /// subtracted from the base count to avoid phantom edge counts.
    pub fn edge_count(&self) -> u64 {
        let ov = self.overlay.edge_count();
        match self.base {
            None => ov,
            Some(b) => {
                let bv = u64::from(b.edge_count);
                // Count tombstoned base edges (out-direction is canonical).
                let tombstones: u64 = self
                    .overlay
                    .out_tombstones
                    .values()
                    .flat_map(|m| m.values())
                    .map(|s| s.len() as u64)
                    .sum();
                bv.saturating_sub(tombstones) + ov
            }
        }
    }

    /// Edge-type ids present in overlay and/or base, sorted ascending.
    pub fn etypes(&self) -> std::vec::IntoIter<u32> {
        match self.base {
            None => {
                // Fast path: collect from overlay (already sorted by Topology::etypes).
                self.overlay.etypes().collect::<Vec<_>>().into_iter()
            }
            Some(base) => {
                let mut set: BTreeSet<u32> = self.overlay.etypes().collect();
                for et in base.etypes.iter() {
                    set.insert(u32::from(et.etype));
                }
                set.into_iter().collect::<Vec<_>>().into_iter()
            }
        }
    }
}

/// Look up neighbors in an archived CSR for `(etype, dir, v)`.
/// Returns an empty Vec if not found.
fn base_neighbors_from_archived(
    base: &ArchivedCsr,
    etype: u32,
    dir: Direction,
    v: u32,
) -> Vec<u32> {
    // Binary search for the etype (etypes are sorted by etype ascending).
    let et_pos = base
        .etypes
        .binary_search_by_key(&etype, |e| u32::from(e.etype));
    let et_entry = match et_pos {
        Ok(i) => &base.etypes[i],
        Err(_) => return Vec::new(),
    };

    let adj = match dir {
        Direction::Out => &et_entry.out_adj,
        Direction::In => &et_entry.in_adj,
    };

    // Binary search for the vertex (rows are sorted by vertex ascending).
    let row_pos = adj.rows.binary_search_by_key(&v, |r| u32::from(r.vertex));
    let row = match row_pos {
        Ok(i) => &adj.rows[i],
        Err(_) => return Vec::new(),
    };

    // Collect neighbor ids (stored as archived u32 LE values).
    row.neighbors.iter().map(|n| u32::from(*n)).collect()
}

/// Merge two sorted-unique slices into a sorted-unique Vec.
fn merge_sorted_unique(a: &[u32], b: &[u32]) -> Vec<u32> {
    let mut out = Vec::with_capacity(a.len() + b.len());
    let mut ai = 0;
    let mut bi = 0;
    while ai < a.len() && bi < b.len() {
        match a[ai].cmp(&b[bi]) {
            std::cmp::Ordering::Less => {
                out.push(a[ai]);
                ai += 1;
            }
            std::cmp::Ordering::Greater => {
                out.push(b[bi]);
                bi += 1;
            }
            std::cmp::Ordering::Equal => {
                out.push(a[ai]);
                ai += 1;
                bi += 1;
            }
        }
    }
    out.extend_from_slice(&a[ai..]);
    out.extend_from_slice(&b[bi..]);
    out
}

/// Remove tombstoned neighbors from `nbrs` (already collected from the base CSR).
///
/// Reads the overlay's tombstone maps for `(etype, dir, v)` and filters out any
/// neighbor that appears in the tombstone set.  Returns the original `nbrs` Vec
/// unchanged (zero allocation) when there are no tombstones.
fn subtract_tombstones(
    nbrs: Vec<u32>,
    etype: u32,
    dir: Direction,
    v: u32,
    overlay: &Topology,
) -> Vec<u32> {
    let tombstones = match dir {
        Direction::Out => overlay.out_tombstones_for(etype, v),
        Direction::In => overlay.in_tombstones_for(etype, v),
    };
    match tombstones {
        None => nbrs,
        Some(t) if t.is_empty() => nbrs,
        Some(t) => nbrs.into_iter().filter(|n| !t.contains(n)).collect(),
    }
}

/// An iterator over the sorted-unique merged neighbor list.
/// Produced by `TopologyView::neighbors`; callers use `.as_ref()`.
///
/// This type alias is exported to satisfy the Task-1 interface contract.
pub type MergedNeighbors<'a> = Cow<'a, [u32]>;

#[cfg(test)]
mod tests {
    use super::*;
    use crate::columns::ColumnStore;
    use crate::idmap::IdMap;
    use crate::interner::Interner;
    use crate::topology::{RemoveEdgeOutcome, Topology};
    use crate::v8::encode::{encode_v8, V8Meta};
    use crate::v8::MappedBase;
    use std::collections::BTreeMap;

    /// A pre-V9 base carries its table inside each `Str` column and no shared
    /// section; the resolution rule must fall back to that copy.
    ///
    /// The encoder cannot write a V8 snapshot any more, so the legacy shape is
    /// built directly: this is the only place the pre-V9 read path is exercised
    /// with real strings (`golden_v8.bin` holds a single Int property).
    #[test]
    fn a_pre_v9_column_resolves_through_its_own_table() {
        use crate::v8::layout::{ColumnData, ColumnsData, FieldEntry};

        let legacy = ColumnsData {
            fields: vec![FieldEntry {
                name: "tag".to_string(),
                col: ColumnData::Str {
                    ids: vec![2, 0, 1],
                    present: vec![0b111],
                    // The whole table, copied into the column: exactly what
                    // every V5-V8 snapshot wrote for every string column.
                    strings: vec!["alpha".into(), "beta".into(), "gamma".into()],
                },
            }],
        };
        let bytes = rkyv::api::high::to_bytes::<rkyv::rancor::Error>(&legacy).expect("rkyv encode");
        let archived =
            rkyv::access::<crate::v8::layout::ArchivedColumnsData, rkyv::rancor::Error>(&bytes)
                .expect("rkyv access");

        let overlay = ColumnStore::new();
        // `with_base` leaves `strings: None` — the pre-V9 case.
        let view = ColumnsView::with_base(&overlay, archived);
        assert!(view.strings.is_none(), "a pre-V9 base has no shared table");
        for (id, want) in [(0u32, "gamma"), (1, "alpha"), (2, "beta")] {
            match view.get(id, "tag") {
                Some(ValueRef::Owned(Value::Str(got))) => assert_eq!(got, want, "node {id}"),
                other => panic!("node {id}: expected Str({want}), got {other:?}"),
            }
        }

        // The materialising path resolves the same way with `shared: None`.
        let store = crate::v8::encode::archived_to_columnstore(archived, None);
        assert_eq!(store.get(0, "tag"), Some(&Value::Str("gamma".into())));
        assert_eq!(store.get(1, "tag"), Some(&Value::Str("alpha".into())));
        assert_eq!(store.get(2, "tag"), Some(&Value::Str("beta".into())));
    }

    fn tiny_meta() -> V8Meta {
        use std::collections::HashMap;
        V8Meta {
            labels: vec![],
            edge_props: crate::edge_props::EdgeProps::new(),
            rule_defs: vec![],
            provenance: BTreeMap::new(),
            rule_tripped: BTreeMap::new(),
            rule_fires: BTreeMap::new(),
            ivf_bytes: Vec::new(),
            view_defs: vec![],
            wal_truncated: false,
            hnsw: BTreeMap::new(),
            last_change: HashMap::new(),
        }
    }

    /// Tombstone subtraction: base CSR has edge A→B for etype E.  The overlay
    /// records a deletion of A→B via `remove_edge` (tombstone).  After merging,
    /// B must NOT appear in `neighbors(E, Out, A)`, and A must NOT appear in
    /// `neighbors(E, In, B)`.
    #[test]
    fn neighbors_with_deletions_subtracts_from_base() {
        // Encode a V8 snapshot containing a single edge A (id=0) → B (id=1).
        let etype = 7u32;
        let a = 0u32;
        let b = 1u32;

        let mut base_topo = Topology::new();
        base_topo.add_edge(etype, a, b);

        let mut ids = IdMap::new();
        ids.get_or_insert("A");
        ids.get_or_insert("B");

        let meta = tiny_meta();
        let mut snap_bytes = Vec::new();
        encode_v8(
            None,
            None,
            None,
            None,
            None,
            &base_topo,
            &ColumnStore::new(),
            &ids,
            &Interner::new(),
            &meta,
            &mut snap_bytes,
        )
        .expect("encode_v8");

        // Map the snapshot bytes to get an ArchivedCsr (zero-copy base).
        let mapped = MappedBase::from_bytes(snap_bytes).expect("from_bytes");
        let archived_csr = mapped.topology().expect("topology section");

        // The overlay starts empty; call remove_edge for the base-only edge.
        // Since the edge is not in the overlay, this records a tombstone.
        let mut overlay = Topology::new();
        let outcome = overlay.remove_edge(etype, a, b);
        assert_eq!(
            outcome,
            RemoveEdgeOutcome::TombstonedBase,
            "edge was base-only, not in overlay"
        );

        // Build the merged view.
        let view = TopologyView {
            overlay: &overlay,
            base: Some(archived_csr),
        };

        // Out-direction: B must NOT appear.
        let out_nbrs = view.neighbors(etype, Direction::Out, a);
        assert!(
            !out_nbrs.contains(&b),
            "tombstoned edge A→B must not appear in Out neighbors; got {out_nbrs:?}"
        );

        // In-direction: A must NOT appear.
        let in_nbrs = view.neighbors(etype, Direction::In, b);
        assert!(
            !in_nbrs.contains(&a),
            "tombstoned edge A→B must not appear in In neighbors of B; got {in_nbrs:?}"
        );

        // Verify: an un-tombstoned edge is still visible.
        let c = 2u32;
        let mut base_topo2 = Topology::new();
        base_topo2.add_edge(etype, a, b);
        base_topo2.add_edge(etype, a, c);
        let mut snap2 = Vec::new();
        let mut ids2 = IdMap::new();
        ids2.get_or_insert("A");
        ids2.get_or_insert("B");
        ids2.get_or_insert("C");
        encode_v8(
            None,
            None,
            None,
            None,
            None,
            &base_topo2,
            &ColumnStore::new(),
            &ids2,
            &Interner::new(),
            &tiny_meta(),
            &mut snap2,
        )
        .expect("encode_v8 2");
        let mapped2 = MappedBase::from_bytes(snap2).expect("from_bytes 2");
        let archived_csr2 = mapped2.topology().expect("topology section 2");

        let mut overlay2 = Topology::new();
        overlay2.remove_edge(etype, a, b); // only tombstone B
        let view2 = TopologyView {
            overlay: &overlay2,
            base: Some(archived_csr2),
        };
        let out_nbrs2 = view2.neighbors(etype, Direction::Out, a);
        assert!(
            !out_nbrs2.contains(&b),
            "B must be hidden by tombstone; got {out_nbrs2:?}"
        );
        assert!(
            out_nbrs2.contains(&c),
            "C must still be visible (not tombstoned); got {out_nbrs2:?}"
        );
    }
}

// ---------------------------------------------------------------------------
// ColumnsView and ValueRef
// ---------------------------------------------------------------------------

/// A borrowed or materialized column value.
///
/// `Borrowed` is returned when the value comes from the owned overlay
/// (`&'a Value` into the `ColumnStore`).  `Owned` is returned when the value
/// is materialized from the archived base section (e.g. a decoded scalar or a
/// float list reconstructed from a `Vector` column).
#[derive(Debug)]
pub enum ValueRef<'a> {
    Borrowed(&'a Value),
    Owned(Value),
}

impl<'a> ValueRef<'a> {
    /// Convert to an owned `Value`, cloning if borrowed.
    pub fn into_value(self) -> Value {
        match self {
            Self::Borrowed(v) => v.clone(),
            Self::Owned(v) => v,
        }
    }

    /// Borrow the inner value.
    pub fn as_value(&self) -> &Value {
        match self {
            Self::Borrowed(v) => v,
            Self::Owned(v) => v,
        }
    }
}

impl<'a> PartialEq<Value> for ValueRef<'a> {
    fn eq(&self, other: &Value) -> bool {
        self.as_value() == other
    }
}

impl<'a> PartialEq for ValueRef<'a> {
    fn eq(&self, other: &Self) -> bool {
        self.as_value() == other.as_value()
    }
}

/// Memoized decode of `Mixed` base columns, owned by the `MappedBase`.
///
/// Every typed column in a V8 base (`Int`, `Float`, `Bool`, `Str`, `Vector`) is
/// indexed by node id, so reading one node's value is O(1) against the mmap.  A
/// `Mixed` column is not: it is a single bincode blob holding *every* node's
/// value for that field, and `List`- and `Map`-valued properties always land
/// there (see `Column::promote` in `columns.rs`).  Decoding it per lookup makes
/// a scan over N nodes cost N full-column decodes.
///
/// The columns section is immutable for the life of the mapping — post-snapshot
/// writes go to the overlay, which `ColumnsView::get` consults first — so each
/// field is decoded once and reused.  The decoded map is no larger than the
/// `ColumnStore` representation the same data occupies when no snapshot exists,
/// so this trades no more memory than a WAL-only open of the same store.
///
/// A blob that fails to decode memoizes as empty, which reads back as "no value
/// for this field" — the same result the uncached path gives for the same bytes.
#[derive(Default)]
pub struct MixedCache {
    /// An `RwLock` rather than a `Mutex`: after the first read of a field every
    /// later one is a shared-lock hash lookup, so concurrent readers of a
    /// `SharedDb` do not queue behind each other on the common path.
    decoded: std::sync::RwLock<HashMap<String, std::sync::Arc<HashMap<u32, Value>>>>,
}

impl MixedCache {
    /// Return the decoded form of the `Mixed` column `field`, decoding `blob`
    /// on the first call and returning the memo on every later call.
    ///
    /// `blob` must be the bytes of that field's column in the base this cache
    /// belongs to; the mapping is immutable, so the pairing is fixed.
    pub fn get_or_decode(&self, field: &str, blob: &[u8]) -> std::sync::Arc<HashMap<u32, Value>> {
        {
            let guard = self.decoded.read().expect("mixed column cache poisoned");
            if let Some(hit) = guard.get(field) {
                return std::sync::Arc::clone(hit);
            }
        }
        // Decoded with no lock held: a concurrent miss on the same field decodes
        // twice and the first insert wins, which costs a little work but never
        // returns different bytes — the blob cannot change.
        let decoded: std::sync::Arc<HashMap<u32, Value>> =
            std::sync::Arc::new(bincode::deserialize(blob).unwrap_or_default());
        let mut guard = self.decoded.write().expect("mixed column cache poisoned");
        std::sync::Arc::clone(
            guard
                .entry(field.to_string())
                .or_insert_with(|| std::sync::Arc::clone(&decoded)),
        )
    }
}

/// A base snapshot's columns section together with the shared string table
/// that resolves its string ids.
///
/// The two travel as a pair because from V9 on neither is complete alone: the
/// columns hold `sid` indices and only the table holds the vocabulary.
/// `strings` is `None` for a pre-V9 base, whose columns still carry their own
/// copies.
#[derive(Copy, Clone)]
pub struct BaseColumns<'a> {
    pub cols: &'a ArchivedColumns,
    pub strings: Option<&'a crate::v8::layout::ArchivedStringTable>,
}

/// Overlay-over-base column store view.
///
/// Reads consult the owned overlay `ColumnStore` first; if the field/node is
/// absent in the overlay, the archived base section is checked.  Writes always
/// go through the WAL into the owned overlay (never through this view).
///
/// **Column handle**: `ColumnsView::column()` returns an overlay-backed
/// `ColumnHandle`.  Base values are NOT visible through the column handle —
/// only through the `get()` path.  This is acceptable because:
///   (a) base is `None` for V5–V7 stores, which have no mmap'd section at all
///       (their whole snapshot is materialised into the overlay at open);
///   (b) callers that need a fused-scan path with base values should use
///       `get()` directly.
///
/// **String columns come in two shapes.** In a V5–V8 base every
/// `ArchivedColumnData::Str` carries its own full copy of the string table and
/// `strings` below is `None`. From V9 on the table is written once as section
/// 12, every column's own copy is empty, and `strings` holds it. `get()`
/// resolves with one rule: if the shared table is present it is the table;
/// otherwise the column's own is.
#[derive(Copy, Clone)]
pub struct ColumnsView<'a> {
    pub overlay: &'a ColumnStore,
    pub base: Option<&'a crate::v8::layout::ArchivedColumns>,
    /// Memo for `Mixed` base columns; see [`MixedCache`].  `None` means every
    /// `Mixed` read decodes the whole column, which is correct but O(column).
    pub mixed: Option<&'a MixedCache>,
    /// The snapshot's one shared string table (section 12).  `None` for a
    /// pre-V9 snapshot, where the column's own `strings` is authoritative.
    pub strings: Option<&'a crate::v8::layout::ArchivedStringTable>,
}

impl<'a> ColumnsView<'a> {
    /// Overlay-only constructor (V5–V7 and the no-base V8 path).
    pub fn owned(overlay: &'a ColumnStore) -> Self {
        Self {
            overlay,
            base: None,
            mixed: None,
            strings: None,
        }
    }

    /// Construct a view that merges an mmap'd V8 base with a WAL-replay overlay.
    ///
    /// Used when a V8 snapshot is open: `overlay` starts empty (or holds only
    /// post-snapshot mutations) and `base` is the zero-copy archived columns
    /// section from the V8 mmap.
    ///
    /// Prefer [`ColumnsView::with_base_cached`] wherever the owning
    /// [`MappedBase`](crate::v8::MappedBase) is in reach: without a cache every
    /// read of a `Mixed` column decodes that whole column.
    pub fn with_base(overlay: &'a ColumnStore, base: &'a ArchivedColumns) -> Self {
        Self {
            overlay,
            base: Some(base),
            mixed: None,
            strings: None,
        }
    }

    /// Same as [`ColumnsView::with_base`], but reads of `Mixed` columns are
    /// served from `mixed` instead of decoding the column on every lookup.
    pub fn with_base_cached(
        overlay: &'a ColumnStore,
        base: &'a ArchivedColumns,
        mixed: &'a MixedCache,
    ) -> Self {
        Self {
            overlay,
            base: Some(base),
            mixed: Some(mixed),
            strings: None,
        }
    }

    /// Attach the snapshot's shared string table (V9 section 12).
    ///
    /// Pass `None` for a pre-V9 base: every `ColumnData::Str` there carries its
    /// own copy of the table and that copy stays authoritative.  Pass the table
    /// from the same [`MappedBase`](crate::v8::MappedBase) the `base` columns
    /// came from — a table from a different snapshot would resolve string ids
    /// against the wrong vocabulary.
    pub fn with_shared_strings(
        mut self,
        strings: Option<&'a crate::v8::layout::ArchivedStringTable>,
    ) -> Self {
        self.strings = strings;
        self
    }

    /// Look up a property value for `(id, field)`: overlay first, then base.
    ///
    /// Returns `ValueRef::Borrowed` for overlay hits (zero allocation) and
    /// `ValueRef::Owned` for base hits (materialises from archived data).
    pub fn get(&self, id: u32, field: &str) -> Option<ValueRef<'_>> {
        // Overlay first.
        if let Some(v) = self.overlay.get(id, field) {
            return Some(ValueRef::Borrowed(v));
        }
        // Prop tombstone check: a RemoveProp for a base-only value records a
        // tombstone so that subsequent reads correctly see the value as absent.
        if self.overlay.is_tombstoned(id, field) {
            return None;
        }
        // Base fallback.
        let base = self.base?;
        let field_entry = base.fields.iter().find(|e| e.name.as_str() == field)?;
        match &field_entry.col {
            ArchivedColumnData::Int { data, present } => {
                if !archived_bitmap_test(present.as_slice(), id) {
                    return None;
                }
                let idx = id as usize;
                if idx >= data.len() {
                    return None;
                }
                Some(ValueRef::Owned(Value::Int(i64::from(data[idx]))))
            }
            ArchivedColumnData::Float { data, present } => {
                if !archived_bitmap_test(present.as_slice(), id) {
                    return None;
                }
                let idx = id as usize;
                if idx >= data.len() {
                    return None;
                }
                Some(ValueRef::Owned(Value::Float(f64::from(data[idx]))))
            }
            ArchivedColumnData::Bool { data, present } => {
                if !archived_bitmap_test(present.as_slice(), id) {
                    return None;
                }
                let idx = id as usize;
                if idx >= data.len() {
                    return None;
                }
                Some(ValueRef::Owned(Value::Bool(data[idx] != 0)))
            }
            ArchivedColumnData::Str {
                ids,
                present,
                strings,
            } => {
                if !archived_bitmap_test(present.as_slice(), id) {
                    return None;
                }
                let idx = id as usize;
                if idx >= ids.len() {
                    return None;
                }
                let sid = u32::from(ids[idx]) as usize;
                // The resolution rule: if the shared table is present it is the
                // table; otherwise the column's own `strings` is.
                let table = match self.strings {
                    Some(shared) => &shared.strings,
                    None => strings,
                };
                if sid >= table.len() {
                    return None;
                }
                Some(ValueRef::Owned(Value::Str(table[sid].as_str().to_string())))
            }
            ArchivedColumnData::Mixed(blob) => match self.mixed {
                // One decode per field for the life of the mapping. The value
                // is cloned out, so the `Arc` is released with this expression.
                Some(cache) => cache
                    .get_or_decode(field, blob.as_slice())
                    .get(&id)
                    .cloned()
                    .map(ValueRef::Owned),
                None => {
                    let map: HashMap<u32, Value> = bincode::deserialize(blob.as_slice()).ok()?;
                    map.get(&id).cloned().map(ValueRef::Owned)
                }
            },
            ArchivedColumnData::Vector { dim, data, present } => {
                if !archived_bitmap_test(present.as_slice(), id) {
                    return None;
                }
                let dim_val = u32::from(*dim) as usize;
                let start = id as usize * dim_val;
                let end = start + dim_val;
                if end > data.len() {
                    return None;
                }
                let floats: Vec<Value> = data[start..end]
                    .iter()
                    .map(|f| Value::Float(f64::from(*f)))
                    .collect();
                Some(ValueRef::Owned(Value::List(floats)))
            }
        }
    }

    /// Return the raw `f64` vector for `(id, field)` from the archived base.
    ///
    /// Returns `None` when:
    ///   - there is no base (V5–V7 or V8 with empty base),
    ///   - the field is not a `Vector` column in the base,
    ///   - node `id` has no value for the field, or
    ///   - the overlay has a value for `(id, field)` (overlay takes priority,
    ///     but that value is accessible via `get()` as a `Value::List`).
    ///
    /// Returns `Cow::Borrowed` when the archived data happens to be 8-byte
    /// aligned (zero-copy fast path), or `Cow::Owned` when it is not
    /// (element-wise copy via `read_unaligned`).
    ///
    /// **Overlay/archive asymmetry**: vectors written after the last snapshot
    /// live in the overlay as `Value::List([Value::Float, ...])`.  Callers
    /// that need `&[f64]` for overlay vectors must call `get()` and convert
    /// manually (`Value::List` → iterate `Value::Float` elements).
    pub fn vector(&self, id: u32, field: &str) -> Option<Cow<'_, [f64]>> {
        // Overlay takes priority: if the overlay has data for (id, field), we
        // do not fall through to base — but we cannot return &[f64] from a
        // Value::List.  See the doc comment above for the asymmetry.
        if self.overlay.get(id, field).is_some() {
            return None;
        }
        // Prop tombstone check: a RemoveProp for a base-only vector records a
        // tombstone so that subsequent reads correctly see the value as absent.
        if self.overlay.is_tombstoned(id, field) {
            return None;
        }
        let base = self.base?;
        let field_entry = base.fields.iter().find(|e| e.name.as_str() == field)?;
        let (dim, data, present) = match &field_entry.col {
            ArchivedColumnData::Vector { dim, data, present } => (dim, data, present),
            _ => return None,
        };
        if !archived_bitmap_test(present.as_slice(), id) {
            return None;
        }
        let dim_val = u32::from(*dim) as usize;
        if dim_val == 0 {
            return None;
        }
        let start = id as usize * dim_val;
        let end = start + dim_val;
        let archived_slice = data.as_slice();
        if end > archived_slice.len() {
            return None;
        }
        let chunk = &archived_slice[start..end];
        // `rend::F64_le` has align(1), so rkyv does not guarantee 8-byte alignment.
        // Fast path: if the pointer is 8-byte aligned, a zero-copy transmute is sound
        // because the LE compile guard above ensures `Archived<f64>` == `rend::F64_le`
        // has the same wire bytes as `f64` on this target.
        // Slow path: copy each element via `read_unaligned` to avoid UB from a
        // misaligned `&f64` reference — correct and safe on all supported targets.
        let ptr = chunk.as_ptr() as *const f64;
        let result = if ptr.align_offset(std::mem::align_of::<f64>()) == 0 {
            // SAFETY: LE compile guard (line 17-18) ensures same bit representation.
            // Alignment verified above.  Slice length matches chunk.len() elements.
            Cow::Borrowed(unsafe { std::slice::from_raw_parts(ptr, chunk.len()) })
        } else {
            let vec: Vec<f64> = (0..chunk.len())
                .map(|i|
                    // SAFETY: ptr+i is within chunk (bounds checked above).
                    // read_unaligned handles the misaligned access correctly.
                    unsafe { std::ptr::read_unaligned(ptr.add(i)) })
                .collect();
            Cow::Owned(vec)
        };
        Some(result)
    }

    /// Return all field names visible through this view: overlay ∪ base,
    /// deduplicated and sorted.
    ///
    /// Does not filter per-node tombstones — the caller should call `get(id,
    /// field)` for each field, which applies tombstone masking.  Fields that
    /// are tombstoned for a given node will return `None` from `get()`.
    pub fn field_names(&self) -> Vec<String> {
        let mut seen = BTreeSet::new();
        for f in self.overlay.fields() {
            seen.insert(f.to_string());
        }
        if let Some(base) = self.base {
            for e in base.fields.iter() {
                seen.insert(e.name.as_str().to_string());
            }
        }
        seen.into_iter().collect()
    }

    /// Return a pre-resolved column handle for `field`.
    ///
    /// The handle is backed by the overlay only.  Base values are NOT visible
    /// through the returned handle — use `get()` for overlay-then-base reads.
    /// This matches the fused-scan hot path in exec.rs which handles V5–V7 and
    /// V8 post-snapshot overlay data; base reads happen via `get()`.
    pub fn column(&self, field: &str) -> ColumnHandle<'_> {
        self.overlay.column(field)
    }
}

// ---------------------------------------------------------------------------
// EdgePropsView
// ---------------------------------------------------------------------------

/// Overlay-over-base edge-property view.
///
/// After a V8 snapshot open the base edge props live in the zero-copy
/// `ArchivedEdgeProps` section.  Only post-snapshot changes are in `overlay`.
/// Tombstones in `overlay` mask base entries for deleted edges.
///
/// This type is `Copy` (two references) and intended to be passed by value.
#[derive(Copy, Clone)]
pub struct EdgePropsView<'a> {
    pub overlay: &'a EdgeProps,
    pub base: Option<&'a ArchivedEdgeProps>,
}

impl<'a> EdgePropsView<'a> {
    /// Overlay-only constructor (no V8 base, or V8 with no prior snapshot).
    pub fn owned(overlay: &'a EdgeProps) -> Self {
        Self {
            overlay,
            base: None,
        }
    }

    /// Overlay + archived base constructor.
    pub fn with_base(overlay: &'a EdgeProps, base: &'a ArchivedEdgeProps) -> Self {
        Self {
            overlay,
            base: Some(base),
        }
    }

    /// Look up a field value for `(etype, src, dst)`.
    ///
    /// 1. If tombstoned in overlay → `None` (deletion masking).
    /// 2. If present in overlay map → return overlay value (owned clone).
    /// 3. Binary-search the archived base (zero-copy decode on hit).
    /// 4. `None` if absent everywhere.
    pub fn get(&self, etype: u32, src: u32, dst: u32, field: &str) -> Option<Value> {
        // Tombstone check — masks both overlay and base entries.
        if self.overlay.is_tombstoned(etype, src, dst) {
            return None;
        }
        // Overlay lookup.
        if let Some(v) = self.overlay.get(etype, src, dst, field) {
            return Some(v.clone());
        }
        // Archive fallback.
        let base = self.base?;
        let entry = base.entries.binary_search_by(|e| {
            let ke = u32::from(e.etype);
            let ks = u32::from(e.src);
            let kd = u32::from(e.dst);
            (ke, ks, kd).cmp(&(etype, src, dst))
        });
        let entry = match entry {
            Ok(i) => &base.entries[i],
            Err(_) => return None,
        };
        let props: std::collections::BTreeMap<String, Value> =
            bincode::deserialize(entry.props_blob.as_slice()).ok()?;
        props.get(field).cloned()
    }
}

/// Test whether bit `id` is set in an archived bitmap (slice of `Archived<u64>`).
fn archived_bitmap_test(words: &[rkyv::Archived<u64>], id: u32) -> bool {
    let word = id as usize / 64;
    let bit = id as usize % 64;
    if word >= words.len() {
        return false;
    }
    (u64::from(words[word]) >> bit) & 1 == 1
}

#[cfg(test)]
mod value_ref_tests {
    use super::ValueRef;
    use crate::types::Value;

    fn borrowed(v: &Value) -> ValueRef<'_> {
        ValueRef::Borrowed(v)
    }

    fn owned(v: Value) -> ValueRef<'static> {
        ValueRef::Owned(v)
    }

    // All Value variants compared through ValueRef vs owned Value.
    #[test]
    fn value_ref_cross_type_equivalence_battery() {
        use std::collections::BTreeMap;
        let cases: Vec<Value> = vec![
            Value::Int(0),
            Value::Int(i64::MIN),
            Value::Int(i64::MAX),
            Value::Float(0.0),
            Value::Float(f64::NAN), // NaN != NaN, so this tests the false branch
            Value::Float(1.5),
            Value::Bool(true),
            Value::Bool(false),
            Value::Str("hello".into()),
            Value::Str("".into()),
            Value::List(vec![Value::Float(1.0), Value::Float(2.0)]),
            Value::List(vec![]),
            Value::Map(BTreeMap::new()),
            Value::Map({
                let mut m = BTreeMap::new();
                m.insert("k".to_string(), Value::Int(1));
                m
            }),
        ];

        for val in &cases {
            // NaN must be handled first: IEEE 754 NaN != NaN for every comparison
            // variant.  All four pairs must assert_ne, then skip the rest of the loop.
            if let Value::Float(f) = val {
                if f.is_nan() {
                    let b = borrowed(val);
                    let o = owned(val.clone());
                    assert_ne!(b, *val, "NaN: Borrowed(v) must not equal v");
                    assert_ne!(o, *val, "NaN: Owned(v) must not equal v");
                    assert_ne!(b, borrowed(val), "NaN: Borrowed == Borrowed must be false");
                    assert_ne!(o, owned(val.clone()), "NaN: Owned == Owned must be false");
                    continue;
                }
            }

            let b = borrowed(val);
            let o = owned(val.clone());

            // Borrowed == owned Value
            assert_eq!(b, *val, "Borrowed(v) == v failed for {val:?}");
            // Owned == owned Value
            assert_eq!(o, *val, "Owned(v) == v failed for {val:?}");
            // Borrowed == Borrowed
            assert_eq!(b, borrowed(val), "Borrowed == Borrowed failed for {val:?}");
            // Owned == Owned
            assert_eq!(o, owned(val.clone()), "Owned == Owned failed for {val:?}");
            // Borrowed == Owned
            assert_eq!(b, o, "Borrowed == Owned failed for {val:?}");
        }
    }

    #[test]
    fn value_ref_cross_type_not_equal() {
        // Different variant types must not compare equal.
        let int_val = Value::Int(1);
        let float_val = Value::Float(1.0);
        assert_ne!(
            borrowed(&int_val),
            borrowed(&float_val),
            "Int(1) must not equal Float(1.0)"
        );
        assert_ne!(
            owned(Value::Bool(true)),
            owned(Value::Int(1)),
            "Bool(true) must not equal Int(1)"
        );
        assert_ne!(
            owned(Value::Map(std::collections::BTreeMap::new())),
            owned(Value::Str("".into())),
            "Map(empty) must not equal Str(empty)"
        );
    }

    #[test]
    fn value_ref_into_value_roundtrip() {
        let val = Value::Str("round-trip".into());
        let b = borrowed(&val);
        assert_eq!(b.into_value(), val);

        let o = owned(Value::Int(42));
        assert_eq!(o.into_value(), Value::Int(42));
    }
}