issundb-core 0.1.0-alpha.26

//! In-memory property columns for the read path.
//!
//! `ColumnsCache` holds one typed column per node property name, indexed by a
//! self-contained dense node mapping (the same pattern as the CSR snapshot). It is
//! built lazily from one full scan of the `nodes` sub-database and kept fresh
//! by a post-commit delta: added and updated nodes are re-read individually,
//! node deletion forces a full rebuild because it reshuffles nothing here but
//! invalidates the dense mapping's completeness guarantee.
//!
//! The cache exists so per-row property access in query execution costs a
//! dense-index read instead of an LMDB point lookup plus a full msgpack
//! decode. Values reconstruct exactly what decoding the stored record yields;
//! properties whose values are not uniformly one scalar kind fall back to a
//! `Json` column so no conversion ever changes a value.

use ahash::AHashMap;
use serde_json::Value;

use crate::error::Error;
use crate::schema::{EdgeId, EdgeRecord, NodeId, NodeRecord};
use crate::storage::{Storage, props};

/// Abstracts which LMDB sub-database a column set is built from, so the same
/// columnar machinery serves both node and edge properties. A source knows how
/// to enumerate every entity's decoded properties, re-read one entity, and name
/// the not-found error for its entity kind.
pub(crate) trait ColumnSource {
    /// The entity id type (a `u64` for both nodes and edges).
    type Id: Copy + Eq + std::hash::Hash;

    /// File name of this column set's on-disk cache file, next to the LMDB
    /// files. Each source needs its own, or the node and edge cache files would
    /// overwrite each other.
    #[cfg(feature = "lmdb")]
    const CACHE_FILE: &'static str;

    /// Decode every entity's user properties as JSON, in storage iteration
    /// order. Each item is `(id, props_json)`.
    fn scan_all(storage: &Storage) -> Result<Vec<(Self::Id, Value)>, Error>;

    /// Decode one entity's user properties as JSON through a caller-supplied
    /// transaction; `None` if it no longer exists (deleted between commit and
    /// refresh). This is the single primitive the two fetch forms below share, so
    /// a direct read and a column build cannot decode a record differently.
    fn get_in_txn(
        storage: &Storage,
        rtxn: &crate::storage::RoTxn,
        id: Self::Id,
    ) -> Result<Option<Value>, Error>;

    /// The not-found error for this entity kind.
    fn not_found(id: Self::Id) -> Error;

    /// Re-read one entity's user properties under its own transaction.
    fn fetch_one(storage: &Storage, id: Self::Id) -> Result<Option<Value>, Error> {
        let rtxn = storage.env.read_txn()?;
        Self::get_in_txn(storage, &rtxn, id)
    }

    /// Read many entities under **one** transaction, in input order.
    ///
    /// Both callers need this rather than a loop over [`ColumnSource::fetch_one`].
    /// A transaction per entity would cost one begin/end pair and one reader-slot
    /// acquisition per id, so a gather of a thousand ids paid a thousand of each;
    /// worse, it would observe up to a thousand different commit points, letting
    /// two values of the same row-major gather come from different graph states.
    ///
    /// One call is therefore a single point in time, but note the scope of that: a
    /// *request* spanning several calls is not. `PropColumns::patch` deliberately
    /// chunks, because the alternative is holding every touched entity's decoded
    /// properties in memory at once, so a patch observes one commit point per chunk.
    /// That is sound for a refresh, which converges because each later commit records
    /// its own touched ids, but a caller needing one snapshot across many ids must
    /// pass them in one call.
    fn fetch_many(storage: &Storage, ids: &[Self::Id]) -> Result<Vec<Option<Value>>, Error> {
        let rtxn = storage.env.read_txn()?;
        ids.iter()
            .map(|&id| Self::get_in_txn(storage, &rtxn, id))
            .collect()
    }
}

/// Builds node property columns from the `nodes` sub-database.
pub(crate) struct NodeSource;

impl ColumnSource for NodeSource {
    type Id = NodeId;

    #[cfg(feature = "lmdb")]
    const CACHE_FILE: &'static str = "node_columns.cache";

    fn scan_all(storage: &Storage) -> Result<Vec<(NodeId, Value)>, Error> {
        let rtxn = storage.env.read_txn()?;
        let mut out = Vec::new();
        for entry in storage.nodes.iter(&rtxn)? {
            let (id, bytes) = entry?;
            let rec: NodeRecord = props::decode(bytes)?;
            out.push((id, props::decode(&rec.props)?));
        }
        Ok(out)
    }

    fn get_in_txn(
        storage: &Storage,
        rtxn: &crate::storage::RoTxn,
        id: NodeId,
    ) -> Result<Option<Value>, Error> {
        match storage.nodes.get(rtxn, &id)? {
            Some(bytes) => {
                let rec: NodeRecord = props::decode(bytes)?;
                Ok(Some(props::decode(&rec.props)?))
            }
            None => Ok(None),
        }
    }

    fn not_found(id: NodeId) -> Error {
        Error::NodeNotFound(id)
    }
}

/// Builds edge property columns from the `edges` sub-database.
pub(crate) struct EdgeSource;

impl ColumnSource for EdgeSource {
    type Id = EdgeId;

    #[cfg(feature = "lmdb")]
    const CACHE_FILE: &'static str = "edge_columns.cache";

    fn scan_all(storage: &Storage) -> Result<Vec<(EdgeId, Value)>, Error> {
        let rtxn = storage.env.read_txn()?;
        let mut out = Vec::new();
        for entry in storage.edges.iter(&rtxn)? {
            let (id, bytes) = entry?;
            let rec: EdgeRecord = props::decode(bytes)?;
            out.push((id, props::decode(&rec.props)?));
        }
        Ok(out)
    }

    fn get_in_txn(
        storage: &Storage,
        rtxn: &crate::storage::RoTxn,
        id: EdgeId,
    ) -> Result<Option<Value>, Error> {
        match storage.edges.get(rtxn, &id)? {
            Some(bytes) => {
                let rec: EdgeRecord = props::decode(bytes)?;
                Ok(Some(props::decode(&rec.props)?))
            }
            None => Ok(None),
        }
    }

    fn not_found(id: EdgeId) -> Error {
        Error::EdgeNotFound(id)
    }
}

/// One typed column over dense node indices.
///
/// The serde derives exist for the columns cache file (see [`crate::cache_file`]);
/// `lookup` is skipped there because it is derivable from `dict`, and
/// [`PropColumn::rebuild_lookup`] restores it after a load.
#[derive(serde::Serialize, serde::Deserialize)]
pub(crate) enum PropColumn {
    Int(Vec<Option<i64>>),
    Float(Vec<Option<f64>>),
    Bool(Vec<Option<bool>>),
    /// Dictionary-encoded strings: `idx[dense]` points into `dict`;
    /// `u32::MAX` marks null or missing.
    Str {
        dict: Vec<String>,
        #[serde(skip)]
        lookup: AHashMap<String, u32>,
        idx: Vec<u32>,
    },
    /// Exact-semantics fallback for mixed-kind, array, object, or
    /// out-of-range numeric values.
    Json(Vec<Option<Value>>),
}

const STR_NULL: u32 = u32::MAX;

thread_local! {
    /// Set while a deliberate materialize is running on this thread.
    static MATERIALIZING: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}

#[cfg(any(feature = "lmdb", test))]
fn materializing_columns() -> bool {
    MATERIALIZING.with(|f| f.get())
}

/// Marks the current thread as deliberately building the columns, so the
/// build path stays quiet: the caller asked for the scan and is about to
/// persist its result, which is the opposite of the situation the warning
/// exists to report. Restores the previous value on drop, so a nested call
/// cannot leave the flag raised.
pub(crate) struct MaterializingColumns(bool);

impl MaterializingColumns {
    pub(crate) fn install() -> Self {
        Self(MATERIALIZING.with(|f| f.replace(true)))
    }
}

impl Drop for MaterializingColumns {
    fn drop(&mut self) {
        MATERIALIZING.with(|f| f.set(self.0));
    }
}

/// The slot value in [`IdGroupCodes::codes`] for a node id that does not
/// exist (an allocation hole, or an id past the array).
pub const ID_GROUP_ABSENT: u32 = u32::MAX;

/// Dense group codes of one property, indexed by node id rather than by a
/// request's position. `codes[node_id]` is the node's group code under exact
/// value identity, [`ID_GROUP_ABSENT`] where no such node exists, and `reps`
/// holds one representative value per code. Built once per write generation by
/// [`crate::Graph::node_prop_group_codes_by_id`] and shared, so a grouped
/// aggregation over a bulk row set reads one array cell per row instead of
/// interning one value per row per query.
pub struct IdGroupCodes {
    pub codes: Vec<u32>,
    pub reps: std::sync::Arc<Vec<Value>>,
}

/// The per-generation cache behind [`crate::Graph::node_prop_group_codes_by_id`],
/// one entry per grouped property, discarded whole on any committed write, the
/// same policy as the label-scan cache.
#[derive(Default)]
pub(crate) struct IdGroupCodesCache {
    pub(crate) generation: u64,
    pub(crate) by_prop: AHashMap<String, std::sync::Arc<IdGroupCodes>>,
}

/// A comparison operator for [`PropColumns::cmp_mask`], the typed in-column
/// predicate evaluation behind `Graph::nodes_prop_cmp_mask`.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum PropCmp {
    Eq,
    Ne,
    Lt,
    Le,
    Gt,
    Ge,
}

/// The scalar kind of one JSON value, used to pick or degrade a column type.
#[derive(PartialEq, Clone, Copy)]
enum Kind {
    Null,
    Int,
    Float,
    Bool,
    Str,
    Other,
}

fn kind_of(v: &Value) -> Kind {
    match v {
        Value::Null => Kind::Null,
        Value::Bool(_) => Kind::Bool,
        Value::Number(n) => {
            if n.is_i64() || (n.is_u64() && n.as_i64().is_some()) {
                Kind::Int
            } else if n.is_f64() {
                Kind::Float
            } else {
                Kind::Other
            }
        }
        Value::String(_) => Kind::Str,
        _ => Kind::Other,
    }
}

impl PropColumn {
    /// Build the tightest column for `values` (one slot per dense index).
    fn from_values(values: Vec<Option<Value>>) -> Self {
        let mut kind = Kind::Null;
        for v in values.iter().flatten() {
            let k = kind_of(v);
            if k == Kind::Null {
                continue;
            }
            if kind == Kind::Null {
                kind = k;
            } else if kind != k {
                kind = Kind::Other;
                break;
            }
        }
        match kind {
            Kind::Int => Self::Int(
                values
                    .into_iter()
                    .map(|v| v.and_then(|v| v.as_i64()))
                    .collect(),
            ),
            Kind::Float => Self::Float(
                values
                    .into_iter()
                    .map(|v| v.and_then(|v| v.as_f64()))
                    .collect(),
            ),
            Kind::Bool => Self::Bool(
                values
                    .into_iter()
                    .map(|v| v.and_then(|v| v.as_bool()))
                    .collect(),
            ),
            Kind::Str => {
                let mut dict = Vec::new();
                let mut lookup: AHashMap<String, u32> = AHashMap::new();
                let mut idx = Vec::with_capacity(values.len());
                for v in values {
                    match v {
                        Some(Value::String(s)) => idx.push(intern(&mut dict, &mut lookup, s)),
                        _ => idx.push(STR_NULL),
                    }
                }
                Self::Str { dict, lookup, idx }
            }
            // All-null columns are stored as Json so a later patch of any kind
            // fits without a degrade.
            Kind::Null | Kind::Other => Self::Json(
                values
                    .into_iter()
                    .map(|v| v.filter(|v| !v.is_null()))
                    .collect(),
            ),
        }
    }

    pub(crate) fn len(&self) -> usize {
        match self {
            Self::Int(v) => v.len(),
            Self::Float(v) => v.len(),
            Self::Bool(v) => v.len(),
            Self::Str { idx, .. } => idx.len(),
            Self::Json(v) => v.len(),
        }
    }

    /// Grow the column with nulls to cover `len` dense slots.
    fn grow(&mut self, len: usize) {
        match self {
            Self::Int(v) => v.resize(len, None),
            Self::Float(v) => v.resize(len, None),
            Self::Bool(v) => v.resize(len, None),
            Self::Str { idx, .. } => idx.resize(len, STR_NULL),
            Self::Json(v) => v.resize(len, None),
        }
    }

    /// Clear one slot to null.
    fn clear(&mut self, dense: usize) {
        match self {
            Self::Int(v) => v[dense] = None,
            Self::Float(v) => v[dense] = None,
            Self::Bool(v) => v[dense] = None,
            Self::Str { idx, .. } => idx[dense] = STR_NULL,
            Self::Json(v) => v[dense] = None,
        }
    }

    /// Set one slot, degrading the column to `Json` when the value's kind does
    /// not match the column type.
    fn set(&mut self, dense: usize, value: Value) {
        match (&mut *self, kind_of(&value)) {
            (_, Kind::Null) => self.clear(dense),
            (Self::Int(v), Kind::Int) => v[dense] = value.as_i64(),
            (Self::Float(v), Kind::Float) => v[dense] = value.as_f64(),
            (Self::Bool(v), Kind::Bool) => v[dense] = value.as_bool(),
            (Self::Str { dict, lookup, idx }, Kind::Str) => {
                if let Value::String(s) = value {
                    idx[dense] = intern(dict, lookup, s);
                }
            }
            (Self::Json(v), _) => v[dense] = Some(value),
            _ => {
                self.degrade_to_json();
                self.set(dense, value);
            }
        }
    }

    /// Convert a typed column to the `Json` fallback in place, exactly.
    fn degrade_to_json(&mut self) {
        let json = |col: &Self| -> Vec<Option<Value>> {
            (0..col.len()).map(|d| col.get_json_opt(d)).collect()
        };
        *self = Self::Json(json(self));
    }

    /// Whether the slot at `dense` holds a non-null value, without
    /// materializing it (a string slot would otherwise clone its dictionary
    /// entry). Backs the grouped-degree kernel's `count(v.prop)` null filter.
    pub(crate) fn is_present(&self, dense: usize) -> bool {
        match self {
            Self::Int(v) => v[dense].is_some(),
            Self::Float(v) => v[dense].is_some(),
            Self::Bool(v) => v[dense].is_some(),
            Self::Str { idx, .. } => idx[dense] != STR_NULL,
            Self::Json(v) => v[dense].is_some(),
        }
    }

    /// Restore the interning `lookup` from `dict` after a cache file load, which
    /// skips it as derivable. A no-op on every other variant.
    ///
    /// Cache files are an LMDB-only structure, so without that feature nothing
    /// loads a column set from bytes and this has no caller.
    #[cfg(feature = "lmdb")]
    pub(crate) fn rebuild_lookup(&mut self) {
        if let Self::Str { dict, lookup, .. } = self {
            *lookup = dict
                .iter()
                .enumerate()
                .map(|(i, s)| (s.clone(), i as u32))
                .collect();
        }
    }

    /// The value at `dense`, or `None` for null/missing.
    pub(crate) fn get_json_opt(&self, dense: usize) -> Option<Value> {
        match self {
            Self::Int(v) => v[dense].map(Value::from),
            Self::Float(v) => v[dense].map(Value::from),
            Self::Bool(v) => v[dense].map(Value::from),
            Self::Str { dict, idx, .. } => match idx[dense] {
                STR_NULL => None,
                i => Some(Value::String(dict[i as usize].clone())),
            },
            Self::Json(v) => v[dense].clone(),
        }
    }
}

fn intern(dict: &mut Vec<String>, lookup: &mut AHashMap<String, u32>, s: String) -> u32 {
    if let Some(&i) = lookup.get(&s) {
        return i;
    }
    let i = dict.len() as u32;
    dict.push(s.clone());
    lookup.insert(s, i);
    i
}

/// Lazily computed distribution statistics over one typed column's non-null
/// values: bounds, an equi-depth histogram, and the most common values.
pub(crate) struct PropStats {
    pub(crate) min: Value,
    pub(crate) max: Value,
    pub(crate) histogram: crate::histogram::Histogram,
    /// Up to [`MCV_LIMIT`] `(value, row_count)` pairs, most frequent first.
    pub(crate) mcvs: Vec<(Value, u64)>,
}

const MCV_LIMIT: usize = 8;
const HISTOGRAM_BUCKETS: usize = 10;

impl PropStats {
    /// Estimated fraction of non-null rows equal to `value`: the exact share
    /// when `value` is a most-common value, the histogram's uniform-in-bucket
    /// estimate otherwise.
    pub(crate) fn equality_selectivity(&self, value: &Value) -> f64 {
        for (v, count) in &self.mcvs {
            if v == value {
                return *count as f64 / self.histogram.total_rows as f64;
            }
        }
        self.histogram.estimate_equality_selectivity(value)
    }
}

/// The materialized column set with its own dense entity mapping. Generic over
/// the [`ColumnSource`] that supplies the entities (nodes or edges).
pub(crate) struct PropColumns<S: ColumnSource> {
    pub(crate) id_to_dense: AHashMap<S::Id, u32>,
    pub(crate) dense_to_id: Vec<S::Id>,
    pub(crate) cols: AHashMap<String, PropColumn>,
    /// Per-property stats, computed on first access through [`prop_stats`]
    /// and invalidated wholesale by [`patch`] (a patch clears the touched
    /// rows in every column, so every property's distribution may change).
    /// `None` is cached for columns with no usable stats (`Json` fallback
    /// columns and all-null columns).
    stats: AHashMap<String, Option<PropStats>>,
}

/// The cache file's view of a column set. Compiled only with `lmdb`, because the
/// files live beside the LMDB database and nothing else reads or writes them.
#[cfg(feature = "lmdb")]
impl<S: ColumnSource<Id = u64>> PropColumns<S> {
    /// Assemble a column set from a cache file's payload: the dense mapping's
    /// inverse and every string column's interning table are derived rather
    /// than stored, and the statistics start empty exactly as a built set's
    /// do. `None` when any column's length disagrees with the dense mapping,
    /// which marks the payload as not describing one entity set.
    pub(crate) fn from_cache_file(
        dense_to_id: Vec<u64>,
        cols: Vec<(String, PropColumn)>,
    ) -> Option<Self> {
        let n = dense_to_id.len();
        let mut map: AHashMap<String, PropColumn> = AHashMap::with_capacity(cols.len());
        for (name, mut col) in cols {
            if col.len() != n {
                return None;
            }
            col.rebuild_lookup();
            map.insert(name, col);
        }
        let id_to_dense = dense_to_id
            .iter()
            .enumerate()
            .map(|(d, &id)| (id, d as u32))
            .collect();
        Some(Self {
            id_to_dense,
            dense_to_id,
            cols: map,
            stats: AHashMap::new(),
        })
    }

    /// The cache file payload view of this column set, the dense mapping and
    /// the columns in name order, so a save is deterministic.
    pub(crate) fn cache_file_parts(&self) -> (&Vec<u64>, Vec<(&String, &PropColumn)>) {
        let mut cols: Vec<(&String, &PropColumn)> = self.cols.iter().collect();
        cols.sort_by_key(|(name, _)| *name);
        (&self.dense_to_id, cols)
    }
}

impl<S: ColumnSource> PropColumns<S> {
    /// Build columns for every property name present, from one full scan.
    fn build(storage: &Storage) -> Result<Self, Error> {
        Ok(Self::from_items(S::scan_all(storage)?))
    }

    /// Build columns over exactly `items`, rather than over every entity.
    ///
    /// The whole-graph build goes through here too, so a partial set is grouped by
    /// the same code. That equivalence is what lets a small grouped read skip the
    /// full scan: the only thing a narrower population can change is the inferred
    /// column kind, and it can only make it *more* specific (fewer distinct kinds
    /// present), never less. Every specific arm yields the same representative
    /// `Value` and the same grouping identity the `Json` fallback would: `Int`
    /// re-wraps through `Value::from(i64)`, `Float` keys on `to_bits`, which the
    /// fallback's shortest-roundtrip string matches, and `Str` clones the same
    /// string. An out-of-range `u64` is `Kind::Other` in `kind_of`, so no arm can
    /// silently null a value the fallback would have kept.
    pub(crate) fn from_items(items: Vec<(S::Id, Value)>) -> Self {
        Self::from_items_for(items, None)
    }

    /// [`PropColumns::from_items`] restricted to one property.
    ///
    /// A grouped read touches exactly one property, so columnarizing the rest is
    /// pure waste: over entities carrying thirty properties it decoded and built
    /// thirty columns to use one, once per group-by expression. `None` keeps every
    /// property, which is what the shared whole-graph build needs.
    pub(crate) fn from_items_for(items: Vec<(S::Id, Value)>, only: Option<&str>) -> Self {
        let n = items.len();
        let mut dense_to_id = Vec::with_capacity(n);
        let mut id_to_dense: AHashMap<S::Id, u32> = AHashMap::with_capacity(n);
        for (i, (id, _)) in items.iter().enumerate() {
            dense_to_id.push(*id);
            id_to_dense.insert(*id, i as u32);
        }

        let mut values: AHashMap<String, Vec<Option<Value>>> = AHashMap::new();
        for (dense, (_, json)) in items.into_iter().enumerate() {
            if let Value::Object(map) = json {
                for (k, v) in map {
                    if only.is_some_and(|want| want != k) {
                        continue;
                    }
                    let col = values.entry(k).or_insert_with(|| vec![None; n]);
                    col[dense] = Some(v);
                }
            }
        }
        let cols: AHashMap<String, PropColumn> = values
            .into_iter()
            .map(|(k, v)| (k, PropColumn::from_values(v)))
            .collect();
        Self {
            id_to_dense,
            dense_to_id,
            cols,
            stats: AHashMap::new(),
        }
    }

    /// Statistics for `prop`, computed on first access and cached until the
    /// next patch or rebuild. `None` when the property has no column, the
    /// column is the `Json` fallback, or it holds no non-null values.
    pub(crate) fn prop_stats(&mut self, prop: &str) -> Option<&PropStats> {
        if !self.stats.contains_key(prop) {
            let computed = self.cols.get(prop).and_then(compute_prop_stats);
            self.stats.insert(prop.to_string(), computed);
        }
        self.stats.get(prop).and_then(|s| s.as_ref())
    }

    /// Gather `props` for each id in `ids`, row-major: `out[i][j]` is the
    /// value of `props[j]` on `ids[i]`. Each id resolves to its dense index
    /// once, so the per-cell cost is one typed column read. A missing property
    /// (or a property name with no column) reads as `Value::Null`; a missing
    /// node is an error, matching the per-row executor path.
    pub(crate) fn props_table(
        &self,
        ids: &[S::Id],
        props: &[&str],
    ) -> Result<Vec<Vec<Value>>, Error> {
        let cols: Vec<Option<&PropColumn>> = props.iter().map(|p| self.cols.get(*p)).collect();
        let mut out = Vec::with_capacity(ids.len());
        for &id in ids {
            let dense = *self.id_to_dense.get(&id).ok_or_else(|| S::not_found(id))? as usize;
            out.push(
                cols.iter()
                    .map(|c| c.and_then(|c| c.get_json_opt(dense)).unwrap_or(Value::Null))
                    .collect(),
            );
        }
        Ok(out)
    }

    /// Gather one property for each id in `ids`: `out[i]` is the value of
    /// `prop` on `ids[i]`. The single-property form of `props_table`,
    /// returning one flat vector so the gather does not pay one row vector
    /// allocation per id. Same semantics: a missing property (or a property
    /// name with no column) reads as `Value::Null`; a missing node is an
    /// error.
    pub(crate) fn prop_column(&self, ids: &[S::Id], prop: &str) -> Result<Vec<Value>, Error> {
        let col = self.cols.get(prop);
        let mut out = Vec::with_capacity(ids.len());
        for &id in ids {
            let dense = *self.id_to_dense.get(&id).ok_or_else(|| S::not_found(id))? as usize;
            out.push(
                col.and_then(|c| c.get_json_opt(dense))
                    .unwrap_or(Value::Null),
            );
        }
        Ok(out)
    }

    /// Assign one dense group code per id under exact value identity of
    /// `prop`, plus one representative value per code (the first occurrence).
    /// Null and missing values share one code whose representative is
    /// `Value::Null`. Two ids get the same code exactly when decoding their
    /// records yields equal property values, which for scalar JSON values is
    /// also serialization equality, so grouping by code matches grouping by
    /// the serialized value. A missing node is an error.
    ///
    /// On a typed column the per-row cost is one dense-index read plus one
    /// native-keyed intern (for the dictionary-encoded string column, a plain
    /// array index); no `Value` is built per row.
    pub(crate) fn group_codes(
        &self,
        ids: &[S::Id],
        prop: &str,
    ) -> Result<(Vec<u32>, Vec<Value>), Error> {
        let mut codes = Vec::with_capacity(ids.len());
        let mut reps: Vec<Value> = Vec::new();

        let Some(col) = self.cols.get(prop) else {
            // No such column: every (existing) id is one null group.
            for &id in ids {
                if !self.id_to_dense.contains_key(&id) {
                    return Err(S::not_found(id));
                }
            }
            if !ids.is_empty() {
                reps.push(Value::Null);
                codes.resize(ids.len(), 0);
            }
            return Ok((codes, reps));
        };

        // Null gets its code lazily so an all-present column never spends one.
        let mut null_code: Option<u32> = None;
        let mut intern_null = |reps: &mut Vec<Value>| -> u32 {
            *null_code.get_or_insert_with(|| {
                reps.push(Value::Null);
                (reps.len() - 1) as u32
            })
        };

        // Sizing the interning map up front spares a near-unique key column (a
        // grouped count keyed by an id-like property) the rehashes of growing a
        // map to one entry per row; the cap bounds the transient for a bulk
        // request that turns out to have few groups.
        let seen_capacity = ids.len().min(1 << 20);

        match col {
            PropColumn::Int(v) => {
                let mut seen: AHashMap<i64, u32> = AHashMap::with_capacity(seen_capacity);
                for &id in ids {
                    let dense =
                        *self.id_to_dense.get(&id).ok_or_else(|| S::not_found(id))? as usize;
                    codes.push(match v[dense] {
                        None => intern_null(&mut reps),
                        Some(n) => *seen.entry(n).or_insert_with(|| {
                            reps.push(Value::from(n));
                            (reps.len() - 1) as u32
                        }),
                    });
                }
            }
            PropColumn::Float(v) => {
                // Keyed by bit pattern: JSON numbers cannot be NaN, and the
                // shortest-roundtrip formatting is injective on f64, so bit
                // identity is serialization identity.
                let mut seen: AHashMap<u64, u32> = AHashMap::with_capacity(seen_capacity);
                for &id in ids {
                    let dense =
                        *self.id_to_dense.get(&id).ok_or_else(|| S::not_found(id))? as usize;
                    codes.push(match v[dense] {
                        None => intern_null(&mut reps),
                        Some(f) => *seen.entry(f.to_bits()).or_insert_with(|| {
                            reps.push(Value::from(f));
                            (reps.len() - 1) as u32
                        }),
                    });
                }
            }
            PropColumn::Bool(v) => {
                let mut seen: [Option<u32>; 2] = [None, None];
                for &id in ids {
                    let dense =
                        *self.id_to_dense.get(&id).ok_or_else(|| S::not_found(id))? as usize;
                    codes.push(match v[dense] {
                        None => intern_null(&mut reps),
                        Some(b) => *seen[b as usize].get_or_insert_with(|| {
                            reps.push(Value::from(b));
                            (reps.len() - 1) as u32
                        }),
                    });
                }
            }
            PropColumn::Str { dict, idx, .. } => {
                // The dictionary index is already a dense value identity; the
                // per-row work is two array reads.
                let mut dict_code: Vec<u32> = vec![u32::MAX; dict.len()];
                for &id in ids {
                    let dense =
                        *self.id_to_dense.get(&id).ok_or_else(|| S::not_found(id))? as usize;
                    codes.push(match idx[dense] {
                        STR_NULL => intern_null(&mut reps),
                        i => {
                            if dict_code[i as usize] == u32::MAX {
                                reps.push(Value::String(dict[i as usize].clone()));
                                dict_code[i as usize] = (reps.len() - 1) as u32;
                            }
                            dict_code[i as usize]
                        }
                    });
                }
            }
            PropColumn::Json(v) => {
                // Mixed kinds: key by the serialized value, the exact group
                // identity the executor's string-keyed fold uses.
                let mut seen: AHashMap<String, u32> = AHashMap::with_capacity(seen_capacity);
                for &id in ids {
                    let dense =
                        *self.id_to_dense.get(&id).ok_or_else(|| S::not_found(id))? as usize;
                    codes.push(match &v[dense] {
                        None => intern_null(&mut reps),
                        Some(val) => *seen.entry(val.to_string()).or_insert_with(|| {
                            reps.push(val.clone());
                            (reps.len() - 1) as u32
                        }),
                    });
                }
            }
        }
        Ok((codes, reps))
    }

    /// Evaluate `value <op> rhs` for each id's value of `prop` directly against
    /// the typed column, one keep flag per id, without materializing a `Value`
    /// per row. `None` declines: the column is the `Json` fallback, whose mixed
    /// kinds need the boxed comparison. A nonexistent entity is an error, as it
    /// is for the boxed gather.
    ///
    /// The semantics mirrored here are Cypher's scalar comparison, the outcome a
    /// filter keeps a row on. Three rules cover every case a typed column can
    /// hold. A null or missing value fails every operator, `Ne` included,
    /// because both comparison forms evaluate a null operand to null, which is
    /// not TRUE. Same-kind values compare natively, with a mixed int and float
    /// pair going through `f64` exactly as the boxed path's `as_f64` fallback
    /// does. A non-null value against a constant of any other kind (a string
    /// against a number, or any array, object, or NaN-sentinel constant) is
    /// unequal but unordered, so it passes `Ne` and fails everything else.
    pub(crate) fn cmp_mask(
        &self,
        ids: &[S::Id],
        prop: &str,
        op: PropCmp,
        rhs: &Value,
    ) -> Result<Option<Vec<bool>>, Error> {
        use std::cmp::Ordering;

        let Some(col) = self.cols.get(prop) else {
            // No such column: every value is null, and null fails every operator.
            for &id in ids {
                if !self.id_to_dense.contains_key(&id) {
                    return Err(S::not_found(id));
                }
            }
            return Ok(Some(vec![false; ids.len()]));
        };

        let keeps = |ord: Ordering| match op {
            PropCmp::Eq => ord == Ordering::Equal,
            PropCmp::Ne => ord != Ordering::Equal,
            PropCmp::Lt => ord == Ordering::Less,
            PropCmp::Le => ord != Ordering::Greater,
            PropCmp::Gt => ord == Ordering::Greater,
            PropCmp::Ge => ord != Ordering::Less,
        };
        // A null constant fails every operator; a kind-mismatched one passes
        // only `Ne`. Both are per-row constants, so the row test reduces to
        // presence.
        let mismatch_keeps = if rhs.is_null() {
            false
        } else {
            op == PropCmp::Ne
        };

        let dense_of = |id: S::Id| -> Result<usize, Error> {
            Ok(*self.id_to_dense.get(&id).ok_or_else(|| S::not_found(id))? as usize)
        };

        let mask = match col {
            PropColumn::Int(v) => match rhs {
                Value::Number(n) => {
                    if let Some(c) = n.as_i64() {
                        ids.iter()
                            .map(|&id| Ok(v[dense_of(id)?].is_some_and(|x| keeps(x.cmp(&c)))))
                            .collect::<Result<Vec<bool>, Error>>()?
                    } else if let Some(c) = n.as_f64() {
                        ids.iter()
                            .map(|&id| {
                                Ok(v[dense_of(id)?]
                                    .is_some_and(|x| (x as f64).partial_cmp(&c).is_some_and(keeps)))
                            })
                            .collect::<Result<Vec<bool>, Error>>()?
                    } else {
                        self.presence_mask(ids, |d| v[d].is_some(), mismatch_keeps)?
                    }
                }
                _ => self.presence_mask(ids, |d| v[d].is_some(), mismatch_keeps)?,
            },
            PropColumn::Float(v) => match rhs {
                // `as_f64` is how the boxed comparison reads either numeric
                // kind, so both constant kinds funnel through it here too.
                Value::Number(n) => {
                    if let Some(c) = n.as_f64() {
                        ids.iter()
                            .map(|&id| {
                                Ok(v[dense_of(id)?]
                                    .is_some_and(|x| x.partial_cmp(&c).is_some_and(keeps)))
                            })
                            .collect::<Result<Vec<bool>, Error>>()?
                    } else {
                        self.presence_mask(ids, |d| v[d].is_some(), mismatch_keeps)?
                    }
                }
                _ => self.presence_mask(ids, |d| v[d].is_some(), mismatch_keeps)?,
            },
            PropColumn::Bool(v) => match rhs {
                Value::Bool(c) => ids
                    .iter()
                    .map(|&id| Ok(v[dense_of(id)?].is_some_and(|x| keeps(x.cmp(c)))))
                    .collect::<Result<Vec<bool>, Error>>()?,
                _ => self.presence_mask(ids, |d| v[d].is_some(), mismatch_keeps)?,
            },
            PropColumn::Str { dict, idx, .. } => match rhs {
                Value::String(c) => {
                    // One comparison per distinct dictionary entry, one array
                    // read per row.
                    let pass: Vec<bool> = dict.iter().map(|s| keeps(s.as_str().cmp(c))).collect();
                    ids.iter()
                        .map(|&id| {
                            Ok(match idx[dense_of(id)?] {
                                STR_NULL => false,
                                i => pass[i as usize],
                            })
                        })
                        .collect::<Result<Vec<bool>, Error>>()?
                }
                _ => self.presence_mask(ids, |d| idx[d] != STR_NULL, mismatch_keeps)?,
            },
            PropColumn::Json(_) => return Ok(None),
        };
        Ok(Some(mask))
    }

    /// The kind-mismatch mask, which is `keeps` for a present value and false
    /// for a null or missing one.
    fn presence_mask(
        &self,
        ids: &[S::Id],
        present: impl Fn(usize) -> bool,
        keeps: bool,
    ) -> Result<Vec<bool>, Error> {
        ids.iter()
            .map(|&id| {
                let dense = *self.id_to_dense.get(&id).ok_or_else(|| S::not_found(id))? as usize;
                Ok(keeps && present(dense))
            })
            .collect()
    }

    /// Re-read `touched` node records and patch their slots in place. New
    /// nodes extend the dense mapping; new property names start a new column.
    fn patch(&mut self, storage: &Storage, touched: &[S::Id]) -> Result<(), Error> {
        // A patch clears the touched rows in every column before re-setting
        // the present properties, so every cached distribution may be stale,
        // including ones whose property the new records no longer carry.
        if !touched.is_empty() {
            self.stats.clear();
        }
        // Read in chunks, not all at once. One transaction per entity would charge
        // the next refresh a begin/end pair for each, but one gather over the whole
        // list would hold every touched entity's decoded properties in memory at
        // once, and that list is unbounded: a million-node batch under a single
        // `Graph::update` accumulates a million ids. Chunking keeps the transient
        // proportional to the chunk while still amortizing the transaction.
        for chunk in touched.chunks(PATCH_CHUNK) {
            let fetched = S::fetch_many(storage, chunk)?;
            for (&id, json) in chunk.iter().zip(fetched) {
                let json = match json {
                    Some(j) => j,
                    // Deleted between commit and refresh; deletion also sets
                    // force_full, so this patch run's result is discarded anyway.
                    None => continue,
                };
                let dense = match self.id_to_dense.get(&id) {
                    Some(&d) => d as usize,
                    None => {
                        let d = self.dense_to_id.len();
                        self.dense_to_id.push(id);
                        self.id_to_dense.insert(id, d as u32);
                        d
                    }
                };
                let n = self.dense_to_id.len();
                for col in self.cols.values_mut() {
                    col.grow(n);
                    col.clear(dense);
                }
                if let Value::Object(map) = json {
                    for (k, v) in map {
                        let col = self.cols.entry(k).or_insert_with(|| {
                            let mut c = PropColumn::Json(Vec::new());
                            c.grow(n);
                            c
                        });
                        col.grow(n);
                        col.set(dense, v);
                    }
                }
            }
        }
        Ok(())
    }
}

/// Sorted non-null values of a typed column. `None` for the `Json` fallback,
/// whose mixed kinds have no total order to summarize.
fn sorted_non_null_values(col: &PropColumn) -> Option<Vec<Value>> {
    let mut vals: Vec<Value> = match col {
        PropColumn::Int(v) => v.iter().flatten().map(|&x| Value::from(x)).collect(),
        // NaN is excluded: it is unordered, and a NaN cell fails every
        // comparison the estimates model, so leaving it out keeps bounds
        // and histogram mass conservative.
        PropColumn::Float(v) => v
            .iter()
            .flatten()
            .filter(|x| !x.is_nan())
            .map(|&x| Value::from(x))
            .collect(),
        PropColumn::Bool(v) => v.iter().flatten().map(|&x| Value::Bool(x)).collect(),
        PropColumn::Str { dict, idx, .. } => idx
            .iter()
            .filter(|&&i| i != STR_NULL)
            .map(|&i| Value::String(dict[i as usize].clone()))
            .collect(),
        PropColumn::Json(_) => return None,
    };
    vals.sort_unstable_by(|a, b| {
        crate::histogram::compare_values(a, b).unwrap_or(std::cmp::Ordering::Equal)
    });
    Some(vals)
}

fn compute_prop_stats(col: &PropColumn) -> Option<PropStats> {
    let vals = sorted_non_null_values(col)?;
    let (min, max) = match (vals.first(), vals.last()) {
        (Some(mn), Some(mx)) => (mn.clone(), mx.clone()),
        _ => return None,
    };
    let histogram = crate::histogram::Histogram::build(&vals, HISTOGRAM_BUCKETS);

    let mut runs: Vec<(Value, u64)> = Vec::new();
    for v in &vals {
        match runs.last_mut() {
            Some((last, count)) if last == v => *count += 1,
            _ => runs.push((v.clone(), 1)),
        }
    }
    // Descending by count, and stable, so equal counts keep the ascending value order
    // `vals` arrived in. That order is observable through `mcvs`.
    runs.sort_by_key(|(_, count)| std::cmp::Reverse(*count));
    runs.truncate(MCV_LIMIT);

    Some(PropStats {
        min,
        max,
        histogram,
        mcvs: runs,
    })
}

/// Pending entity mutations the columns have not absorbed yet.
struct ColumnsDelta<Id> {
    touched: Vec<Id>,
    force_full: bool,
}

impl<Id> Default for ColumnsDelta<Id> {
    fn default() -> Self {
        Self {
            touched: Vec::new(),
            force_full: false,
        }
    }
}

/// Thread-safe lazy holder for [`PropColumns`], fed post-commit by the write
/// path and refreshed on read access. Generic over the [`ColumnSource`] so one
/// instance serves nodes and another serves edges.
pub(crate) struct ColumnsCache<S: ColumnSource> {
    columns: parking_lot::RwLock<Option<PropColumns<S>>>,
    pending: parking_lot::Mutex<ColumnsDelta<S::Id>>,
    /// Point reads served straight from storage while the columns were absent.
    /// See [`ColumnsCache::note_direct_reads`].
    direct_reads: std::sync::atomic::AtomicU64,
}

/// Direct point reads tolerated before building the columns anyway. A single
/// point lookup must not pay for a full scan, but a row pipeline that reads one
/// property per row would pay the per-read decode forever, so the reads
/// amortize the build after this many.
const DIRECT_READ_BUILD_THRESHOLD: u64 = 4096;

/// Entities re-read per transaction when patching. Bounds the transient memory a
/// *patch* holds: the pending touched list is unbounded (one `Graph::update` may
/// insert millions of nodes), so a single gather over all of it would decode every
/// one of their property maps at once.
///
/// This bounds the patch path only. A `force_full` refresh takes
/// `PropColumns::build` instead, whose `scan_all` still returns a decoded property
/// map for every entity in the graph before `from_items` runs, so the same
/// transient is paid there. Since a node deletion in a batch sets `force_full`, a
/// large batch containing one can still take that path; bounding it needs a
/// streaming build, which is a larger change than this constant.
const PATCH_CHUNK: usize = 4096;

/// Largest gather served straight from storage while the columns are absent.
/// Above this the request is bulk enough that one scan is the cheaper way to
/// answer it, since the scan decodes each record exactly once either way.
const SMALL_GATHER_MAX: usize = 1024;

impl<S: ColumnSource> Default for ColumnsCache<S> {
    fn default() -> Self {
        Self {
            columns: parking_lot::RwLock::new(None),
            pending: parking_lot::Mutex::new(ColumnsDelta::default()),
            direct_reads: std::sync::atomic::AtomicU64::new(0),
        }
    }
}

impl<S: ColumnSource<Id = u64>> ColumnsCache<S> {
    /// A full build, served from the columns cache file when a fresh one exists.
    /// The cache file is generation-checked against storage, so a load and a
    /// build are indistinguishable to the caller; any mismatch or damage falls
    /// through to the scan.
    fn build_or_load(storage: &Storage) -> Result<PropColumns<S>, Error> {
        #[cfg(feature = "lmdb")]
        if let Some(cols) = crate::cache_file::load_columns::<S>(storage) {
            return Ok(cols);
        }
        // Falling through means a full scan of every entity record, which is the
        // expensive half of this structure: measured on a 2.4 M-node graph, the
        // build peaked at 13.4 GB against 4.0 GB for loading a cache file, and
        // took roughly three times as long. A process that reaches here has no
        // usable file, so it pays that on its first bulk property read and again
        // on every restart. Saying so is the only warning a server surface gets:
        // neither REST nor MCP exposes a way to build the columns, so an operator
        // who does not know this is happening has no way to notice it either.
        //
        // A deliberate `materialize_*` call is about to write the file, so it is
        // not warned about; see `MaterializingColumns`. The advice only makes
        // sense where a cache file can exist at all, so the in-memory backend
        // stays silent: it never persists, and a reopen sees an empty graph.
        #[cfg(feature = "lmdb")]
        if !materializing_columns() {
            tracing::warn!(
                entity = std::any::type_name::<S>(),
                "building property columns from a full scan because no current \
                 cache file exists; this repeats on every process start. Run \
                 `materialize-columns` in the CLI, or \
                 `materialize_property_columns()` or \
                 `materialize_edge_property_columns()` from Python, to persist \
                 them",
            );
        }
        PropColumns::build(storage)
    }
    /// Record an added or updated entity. Called post-commit.
    pub(crate) fn record_touched(&self, id: S::Id) {
        let mut p = self.pending.lock();
        if !p.force_full {
            p.touched.push(id);
        }
    }

    pub(crate) fn record_touched_many(&self, ids: &[S::Id]) {
        let mut p = self.pending.lock();
        if !p.force_full {
            p.touched.extend_from_slice(ids);
        }
    }

    /// Record a deletion (which reshuffles the dense mapping). Called
    /// post-commit.
    pub(crate) fn record_force_full(&self) {
        let mut p = self.pending.lock();
        p.force_full = true;
        p.touched.clear();
    }

    /// Run `f` against fresh columns, building or patching them first if the
    /// cache is stale or absent.
    pub(crate) fn with_fresh<T>(
        &self,
        storage: &Storage,
        f: impl FnOnce(&PropColumns<S>) -> T,
    ) -> Result<T, Error> {
        loop {
            {
                let guard = self.columns.read();
                if let Some(cols) = guard.as_ref() {
                    let p = self.pending.lock();
                    if p.touched.is_empty() && !p.force_full {
                        drop(p);
                        return Ok(f(cols));
                    }
                }
            }
            let mut guard = self.columns.write();
            let delta = std::mem::take(&mut *self.pending.lock());
            let absorbed = match guard.as_mut() {
                Some(cols) if !delta.force_full => cols.patch(storage, &delta.touched),
                _ => Self::build_or_load(storage).map(|cols| {
                    *guard = Some(cols);
                    self.forget_direct_reads();
                }),
            };
            self.recover_if_failed(absorbed)?;
            // Loop back to the fast path so a delta that landed during the
            // rebuild is also absorbed before serving.
        }
    }

    /// Turn a failed absorb into a pending full rebuild instead of a silently
    /// stale cache.
    ///
    /// The delta was already taken out of the shared buffer before the absorb ran,
    /// so propagating the error on its own would drop that work: the buffer would
    /// be empty, the columns would still hold pre-write values, and every later
    /// read would take the fast path and serve them indefinitely. One transient
    /// LMDB error would become permanently wrong property reads.
    ///
    /// The recovery has to be a full rebuild rather than re-queueing the taken ids,
    /// because `patch` applies per entity and may have failed partway, leaving the
    /// columns in a state no id list describes.
    fn recover_if_failed(&self, outcome: Result<(), Error>) -> Result<(), Error> {
        if outcome.is_err() {
            self.record_force_full();
        }
        outcome
    }

    /// Whether the columns are already materialized. A caller that can serve
    /// itself from storage uses this to avoid *causing* a build.
    pub(crate) fn is_built(&self) -> bool {
        self.columns.read().is_some()
    }

    /// Forget the accumulated direct reads. Called whenever the columns are
    /// installed or dropped, because the counter asks "have enough direct reads
    /// piled up to justify a build?" and that question restarts with each column
    /// lifetime.
    ///
    /// Without the reset the counter was monotonic for the life of the process, so
    /// once any 4096 reads had amortized one build the small-request escape hatch
    /// was disabled forever. A later `force_full` that dropped the columns then
    /// left a single point read paying a full node scan to rebuild them, which is
    /// the exact cost the escape hatch exists to avoid.
    fn forget_direct_reads(&self) {
        self.direct_reads
            .store(0, std::sync::atomic::Ordering::Relaxed);
    }

    /// Record `n` reads served without the columns, reporting whether the
    /// accumulated direct reads now justify building them.
    fn note_direct_reads(&self, n: usize) -> bool {
        let n = n.max(1) as u64;
        self.direct_reads
            .fetch_add(n, std::sync::atomic::Ordering::Relaxed)
            + n
            >= DIRECT_READ_BUILD_THRESHOLD
    }

    /// Whether a gather over `ids` entities should be served straight from
    /// storage instead of by building every column.
    ///
    /// True only while the columns are absent, the request is small, and the
    /// accumulated direct reads have not yet amortized a build. Building is one
    /// full entity scan, so it is the wrong answer to a request for a handful of
    /// entities however the caller phrased it: the vectorized executor gathers
    /// even a one-row projection through the bulk API, and that is what made the
    /// first property-touching query on a large graph pay a full scan.
    pub(crate) fn should_serve_directly(&self, ids: usize) -> bool {
        ids <= SMALL_GATHER_MAX && !self.is_built() && !self.note_direct_reads(ids)
    }

    /// Like [`ColumnsCache::with_fresh`], but with the mutable access the lazy
    /// statistics cache needs, and it never builds an absent cache: returns
    /// `Ok(None)` when the columns do not exist yet.
    ///
    /// This is for advisory readers, the optimizer's selectivity estimates and
    /// the zone-map prune, whose answers only weight a choice. Building for them
    /// costs one full entity scan on the first query that so much as mentions a
    /// property, which dominates cold-start latency; declining to answer leaves
    /// the caller on its default weight, which is always sound.
    pub(crate) fn with_existing_mut<T>(
        &self,
        storage: &Storage,
        f: impl FnOnce(&mut PropColumns<S>) -> T,
    ) -> Result<Option<T>, Error> {
        // Checked under the read lock so the common absent case does not
        // serialize against concurrent gathers.
        if self.columns.read().is_none() {
            return Ok(None);
        }
        let mut guard = self.columns.write();
        if guard.is_none() {
            return Ok(None);
        }
        loop {
            let delta = std::mem::take(&mut *self.pending.lock());
            if delta.touched.is_empty() && !delta.force_full {
                if let Some(cols) = guard.as_mut() {
                    return Ok(Some(f(cols)));
                }
            }
            match guard.as_mut() {
                // Patching is per touched entity, so an advisory reader can
                // afford to bring existing columns up to date. A failure here must
                // leave a full rebuild pending rather than drop the taken delta;
                // see `recover_if_failed`.
                Some(cols) if !delta.force_full => {
                    self.recover_if_failed(cols.patch(storage, &delta.touched))?
                }
                // A node deletion invalidates the dense mapping and forces a
                // full rebuild, which is the one refresh an advisory reader must
                // not pay for: dropping the columns and declining leaves the
                // rebuild to the next gather. Rebuilding here would make a
                // delete-then-query session pay one full scan per query, the
                // exact cost this reader exists to avoid.
                _ => {
                    *guard = None;
                    // The columns are gone, so small requests must be servable
                    // directly again rather than forced to rebuild them.
                    self.forget_direct_reads();
                    return Ok(None);
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use serde_json::json;
    use tempfile::TempDir;

    use super::{ColumnsCache, DIRECT_READ_BUILD_THRESHOLD, NodeSource};
    use crate::{Graph, error::Error};

    /// The materialize guard suppresses the no-cache warning for its own thread
    /// and restores whatever was set before, including under nesting.
    ///
    /// The warning itself is a single `tracing::warn!` and is not asserted here:
    /// capturing it would mean a subscriber in dev-dependencies for one line.
    /// What can actually break is this guard, since a leaked `true` would silence
    /// the warning for every later query on the thread, which is exactly the
    /// case it exists to report.
    #[test]
    fn the_materialize_guard_suppresses_and_restores() {
        assert!(!super::materializing_columns(), "quiet by default");
        {
            let _outer = super::MaterializingColumns::install();
            assert!(super::materializing_columns());
            {
                let _inner = super::MaterializingColumns::install();
                assert!(super::materializing_columns());
            }
            assert!(
                super::materializing_columns(),
                "the inner guard must restore the outer one's value, not the default",
            );
        }
        assert!(
            !super::materializing_columns(),
            "the guard must not leak past its scope",
        );
    }

    /// A small grouped read must not build every column, and must produce exactly
    /// what the built columns would.
    ///
    /// Grouping used to always build, so a grouped count over a handful of group
    /// nodes paid one full node scan on a cold graph. The ephemeral path has to
    /// agree with the shared one on the cases where a narrower population infers a
    /// different column kind: `mixed` is a `Json` column over the whole graph but
    /// `Int` over an int-only subset, and both must group and represent identically.
    #[test]
    fn small_grouped_read_avoids_the_build_and_agrees_with_it() {
        let (_dir, g) = open_tmp();
        let ints: Vec<_> = (0..4)
            .map(|i| {
                g.add_node("N", &json!({ "mixed": i % 2, "s": "x" }))
                    .unwrap()
            })
            .collect();
        // These make the whole-graph `mixed` column a Json fallback, while any
        // subset drawn from `ints` alone infers `Int`.
        g.add_node("N", &json!({ "mixed": 1.5 })).unwrap();
        g.add_node("N", &json!({ "mixed": "one" })).unwrap();
        // A node with no `mixed` at all, so the null group is exercised.
        let bare = g.add_node("N", &json!({ "s": "y" })).unwrap();

        let mut ask = ints.clone();
        ask.push(bare);

        assert!(!g.prop_columns.is_built(), "cold to start");
        let direct = g.node_prop_group_codes(&ask, "mixed").unwrap();
        assert!(
            !g.prop_columns.is_built(),
            "a small grouped read must not build every column"
        );

        // Force the shared columns, then ask again: same codes, same
        // representatives, even though the column kind differs between the two
        // populations.
        g.prop_columns
            .with_fresh(&g.storage, |_| ())
            .expect("build the shared columns");
        assert!(g.prop_columns.is_built());
        let built = g.node_prop_group_codes(&ask, "mixed").unwrap();

        assert_eq!(direct.0, built.0, "group codes must agree");
        assert_eq!(direct.1, built.1, "representative values must agree");
        // Sanity: 0 and 1 are distinct groups and the missing value is its own.
        assert_eq!(direct.0.len(), 5);
        assert_eq!(direct.1.len(), 3);

        // A node that does not exist is still an error on the direct path.
        assert!(g.node_prop_group_codes(&[999_999], "mixed").is_err());
    }

    /// A failed absorb must leave a full rebuild pending. The delta is taken out
    /// of the shared buffer before the absorb runs, so propagating the error
    /// alone would drop that work and leave the columns serving pre-write values
    /// forever, with an empty buffer telling every later reader they are current.
    #[test]
    fn a_failed_absorb_leaves_a_full_rebuild_pending() {
        let cache: ColumnsCache<NodeSource> = ColumnsCache::default();
        cache.record_touched(7);

        let outcome = cache.recover_if_failed(Err(Error::Corrupt("simulated absorb failure")));

        assert!(outcome.is_err(), "the error still propagates to the caller");
        let pending = cache.pending.lock();
        assert!(
            pending.force_full,
            "a failed absorb must queue a full rebuild, not vanish"
        );
        assert!(
            pending.touched.is_empty(),
            "a full rebuild supersedes the per-entity list"
        );
    }

    /// A successful absorb must leave the buffer alone, so the recovery path
    /// cannot cost a rebuild on the happy path.
    #[test]
    fn a_successful_absorb_queues_nothing() {
        let cache: ColumnsCache<NodeSource> = ColumnsCache::default();

        assert!(cache.recover_if_failed(Ok(())).is_ok());

        let pending = cache.pending.lock();
        assert!(!pending.force_full);
        assert!(pending.touched.is_empty());
    }

    fn open_tmp() -> (TempDir, Graph) {
        let dir = TempDir::new().unwrap();
        let g = Graph::open(dir.path(), 1).unwrap();
        (dir, g)
    }

    /// Materialize the property columns. No reader builds them unconditionally any
    /// more: the advisory statistics never do, and both a small gather and a small
    /// grouped read are served without them, so a test that needs the shared
    /// columns present has to ask for them directly.
    fn materialize_columns(g: &Graph) {
        g.prop_columns
            .with_fresh(&g.storage, |_| ())
            .expect("materialize the property columns");
    }

    #[test]
    fn typed_values_round_trip_exactly() {
        let (_dir, g) = open_tmp();
        let a = g
            .add_node(
                "N",
                &json!({ "i": 42, "f": 1.5, "s": "hello", "b": true, "arr": [1, 2] }),
            )
            .unwrap();

        assert_eq!(g.node_prop_json(a, "i").unwrap(), Some(json!(42)));
        assert_eq!(g.node_prop_json(a, "f").unwrap(), Some(json!(1.5)));
        assert_eq!(g.node_prop_json(a, "s").unwrap(), Some(json!("hello")));
        assert_eq!(g.node_prop_json(a, "b").unwrap(), Some(json!(true)));
        assert_eq!(g.node_prop_json(a, "arr").unwrap(), Some(json!([1, 2])));
    }

    #[test]
    fn missing_property_is_null_and_missing_node_is_none() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &json!({ "x": 1 })).unwrap();
        assert_eq!(
            g.node_prop_json(a, "nope").unwrap(),
            Some(serde_json::Value::Null)
        );
        assert_eq!(g.node_prop_json(a + 999, "x").unwrap(), None);
    }

    /// The id-indexed group codes must induce the same partition and the same
    /// representative values as the per-id form, serve repeats from the cache,
    /// and track writes.
    #[test]
    fn id_indexed_group_codes_agree_with_the_per_id_form() {
        let (_dir, g) = open_tmp();
        let ids = vec![
            g.add_node("N", &json!({ "v": 1 })).unwrap(),
            g.add_node("N", &json!({ "v": 2 })).unwrap(),
            g.add_node("N", &json!({ "v": 1 })).unwrap(),
            g.add_node("N", &json!({})).unwrap(),
        ];
        materialize_columns(&g);

        let by_id = g.node_prop_group_codes_by_id("v").unwrap();
        let (codes, reps) = g.node_prop_group_codes(&ids, "v").unwrap();

        // Same partition, compared through the representative values.
        let dense_reps: Vec<_> = ids
            .iter()
            .map(|&id| by_id.reps[by_id.codes[id as usize] as usize].clone())
            .collect();
        let per_id_reps: Vec<_> = ids
            .iter()
            .zip(&codes)
            .map(|(_, &c)| reps[c as usize].clone())
            .collect();
        assert_eq!(dense_reps, per_id_reps);
        assert_eq!(dense_reps, vec![json!(1), json!(2), json!(1), json!(null)]);

        // A repeat with no intervening write serves the cached array.
        let again = g.node_prop_group_codes_by_id("v").unwrap();
        assert!(std::sync::Arc::ptr_eq(&by_id, &again));

        // A write invalidates, and the new node appears.
        let e = g.add_node("N", &json!({ "v": 2 })).unwrap();
        let fresh = g.node_prop_group_codes_by_id("v").unwrap();
        assert!(!std::sync::Arc::ptr_eq(&by_id, &fresh));
        assert_eq!(fresh.reps[fresh.codes[e as usize] as usize], json!(2));
    }

    /// The typed comparison mask must match what boxing every value and
    /// comparing under Cypher scalar semantics would produce: null and missing
    /// values fail every operator, same-kind values compare natively, and a
    /// kind-mismatched non-null value passes only `Ne`.
    #[test]
    fn typed_cmp_mask_follows_cypher_scalar_semantics() {
        use super::PropCmp::*;
        let (_dir, g) = open_tmp();
        let n30 = g
            .add_node("N", &json!({ "age": 30, "city": "oslo" }))
            .unwrap();
        let n40 = g
            .add_node("N", &json!({ "age": 40, "city": "rome" }))
            .unwrap();
        let n25 = g
            .add_node("N", &json!({ "age": 25, "city": "bern" }))
            .unwrap();
        let bare = g.add_node("N", &json!({ "city": "oslo" })).unwrap();
        materialize_columns(&g);
        let ids = [n30, n40, n25, bare];

        let mask = |prop: &str, op, rhs: serde_json::Value| {
            g.nodes_prop_cmp_mask(&ids, prop, op, &rhs)
                .unwrap()
                .expect("typed column must answer")
        };

        assert_eq!(mask("age", Ge, json!(30)), vec![true, true, false, false]);
        assert_eq!(mask("age", Le, json!(30)), vec![true, false, true, false]);
        assert_eq!(mask("age", Lt, json!(30)), vec![false, false, true, false]);
        assert_eq!(mask("age", Gt, json!(30)), vec![false, true, false, false]);
        assert_eq!(mask("age", Eq, json!(30)), vec![true, false, false, false]);
        assert_eq!(mask("age", Ne, json!(30)), vec![false, true, true, false]);

        // Cross-kind numeric compares go through f64, as the boxed path does.
        assert_eq!(
            mask("age", Ge, json!(30.5)),
            vec![false, true, false, false]
        );
        assert_eq!(
            mask("age", Eq, json!(30.0)),
            vec![true, false, false, false]
        );

        // Dictionary strings compare per distinct dictionary entry.
        assert_eq!(
            mask("city", Eq, json!("oslo")),
            vec![true, false, false, true]
        );
        assert_eq!(
            mask("city", Gt, json!("bern")),
            vec![true, true, false, true]
        );

        // A kind-mismatched constant keeps a non-null row only under `Ne`.
        assert_eq!(
            mask("age", Ne, json!("thirty")),
            vec![true, true, true, false]
        );
        assert_eq!(mask("age", Eq, json!("thirty")), vec![false; 4]);
        assert_eq!(mask("age", Lt, json!("thirty")), vec![false; 4]);
        assert_eq!(
            mask("age", Ne, json!([1, 2])),
            vec![true, true, true, false]
        );

        // A null constant fails every operator on every row.
        assert_eq!(mask("age", Ne, serde_json::Value::Null), vec![false; 4]);
        assert_eq!(mask("age", Eq, serde_json::Value::Null), vec![false; 4]);

        // A property with no column at all reads as null everywhere.
        assert_eq!(mask("nope", Ne, json!(1)), vec![false; 4]);

        // A nonexistent node is an error, exactly as the boxed gather reports it.
        assert!(
            g.nodes_prop_cmp_mask(&[bare + 999], "age", Eq, &json!(1))
                .is_err()
        );
    }

    /// Float columns and integer constants must compare exactly as the boxed
    /// path's `as_f64` fallback does, and boolean columns order false before
    /// true.
    #[test]
    fn typed_cmp_mask_covers_float_and_bool_columns() {
        use super::PropCmp::*;
        let (_dir, g) = open_tmp();
        let a = g
            .add_node("N", &json!({ "score": 1.5, "flag": true }))
            .unwrap();
        let b = g
            .add_node("N", &json!({ "score": 2.0, "flag": false }))
            .unwrap();
        materialize_columns(&g);
        let ids = [a, b];

        let mask = |prop: &str, op, rhs: serde_json::Value| {
            g.nodes_prop_cmp_mask(&ids, prop, op, &rhs)
                .unwrap()
                .expect("typed column must answer")
        };

        assert_eq!(mask("score", Gt, json!(1.5)), vec![false, true]);
        assert_eq!(mask("score", Ge, json!(2)), vec![false, true]);
        assert_eq!(mask("score", Eq, json!(2)), vec![false, true]);
        assert_eq!(mask("flag", Eq, json!(true)), vec![true, false]);
        assert_eq!(mask("flag", Lt, json!(true)), vec![false, true]);
        assert_eq!(mask("flag", Ne, json!(1)), vec![true, true]);
    }

    /// A `Json` fallback column declines rather than approximating, and a small
    /// request on a cold graph declines rather than building every column.
    #[test]
    fn typed_cmp_mask_declines_json_columns_and_small_cold_requests() {
        use super::PropCmp::*;
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &json!({ "mixed": 1 })).unwrap();
        let b = g.add_node("N", &json!({ "mixed": "one" })).unwrap();

        assert!(!g.prop_columns.is_built(), "cold to start");
        assert_eq!(
            g.nodes_prop_cmp_mask(&[a, b], "mixed", Eq, &json!(1))
                .unwrap(),
            None,
            "a small request on a cold graph must decline, not build"
        );
        assert!(!g.prop_columns.is_built());

        materialize_columns(&g);
        assert_eq!(
            g.nodes_prop_cmp_mask(&[a, b], "mixed", Eq, &json!(1))
                .unwrap(),
            None,
            "a mixed-kind column must decline"
        );
    }

    #[test]
    fn mixed_kind_property_keeps_exact_values() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &json!({ "v": 1 })).unwrap();
        let b = g.add_node("N", &json!({ "v": "one" })).unwrap();
        assert_eq!(g.node_prop_json(a, "v").unwrap(), Some(json!(1)));
        assert_eq!(g.node_prop_json(b, "v").unwrap(), Some(json!("one")));
    }

    #[test]
    fn update_node_is_visible_and_can_remove_and_degrade() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &json!({ "x": 1, "y": 2 })).unwrap();
        assert_eq!(g.node_prop_json(a, "x").unwrap(), Some(json!(1)));

        // x changes kind (degrade), y disappears (cleared slot).
        g.update_node(a, &json!({ "x": "now a string" })).unwrap();
        assert_eq!(
            g.node_prop_json(a, "x").unwrap(),
            Some(json!("now a string"))
        );
        assert_eq!(
            g.node_prop_json(a, "y").unwrap(),
            Some(serde_json::Value::Null)
        );
    }

    #[test]
    fn nodes_added_after_first_build_are_visible() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &json!({ "x": 1 })).unwrap();
        assert_eq!(g.node_prop_json(a, "x").unwrap(), Some(json!(1)));

        let b = g.add_node("N", &json!({ "x": 2, "fresh": "yes" })).unwrap();
        assert_eq!(g.node_prop_json(b, "x").unwrap(), Some(json!(2)));
        assert_eq!(g.node_prop_json(b, "fresh").unwrap(), Some(json!("yes")));
        // The new property name reads as null on the older node.
        assert_eq!(
            g.node_prop_json(a, "fresh").unwrap(),
            Some(serde_json::Value::Null)
        );
    }

    #[test]
    fn delete_node_forces_rebuild() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &json!({ "x": 1 })).unwrap();
        let b = g.add_node("N", &json!({ "x": 2 })).unwrap();
        assert_eq!(g.node_prop_json(a, "x").unwrap(), Some(json!(1)));

        g.delete_node(a).unwrap();
        assert_eq!(g.node_prop_json(a, "x").unwrap(), None);
        assert_eq!(g.node_prop_json(b, "x").unwrap(), Some(json!(2)));
    }

    #[test]
    fn props_table_gathers_rows_in_input_order() {
        let (_dir, g) = open_tmp();
        let a = g
            .add_node("N", &json!({ "name": "ada", "age": 36, "city": "london" }))
            .unwrap();
        let b = g
            .add_node("N", &json!({ "name": "bob", "age": 4 }))
            .unwrap();

        // Duplicate ids are allowed and each occurrence gets its own row.
        let table = g
            .node_props_json_table(&[b, a, b], &["name", "age", "city"])
            .unwrap();
        assert_eq!(
            table,
            vec![
                vec![json!("bob"), json!(4), serde_json::Value::Null],
                vec![json!("ada"), json!(36), json!("london")],
                vec![json!("bob"), json!(4), serde_json::Value::Null],
            ]
        );

        // An unknown property name yields a null column, not an error.
        let table = g.node_props_json_table(&[a], &["nope"]).unwrap();
        assert_eq!(table, vec![vec![serde_json::Value::Null]]);

        // Empty inputs are fine.
        assert!(g.node_props_json_table(&[], &["name"]).unwrap().is_empty());
        assert_eq!(
            g.node_props_json_table(&[a], &[]).unwrap(),
            vec![Vec::<serde_json::Value>::new()]
        );
    }

    #[test]
    fn group_codes_match_value_identity() {
        let (_dir, g) = open_tmp();
        // One mixed-kind property (a Json column): 1, "1", 1.0, true, and a
        // missing value must each get their own group; equal values share.
        let vals = [
            json!({ "v": 1 }),
            json!({ "v": "1" }),
            json!({ "v": 1.0 }),
            json!({ "v": true }),
            json!({}),
            json!({ "v": 1 }),
            json!({ "v": "1" }),
        ];
        let ids: Vec<_> = vals.iter().map(|p| g.add_node("N", p).unwrap()).collect();

        let (codes, reps) = g.node_prop_group_codes(&ids, "v").unwrap();
        assert_eq!(codes.len(), ids.len());
        // Equal values share a code; the representative is the value itself.
        assert_eq!(codes[0], codes[5]);
        assert_eq!(codes[1], codes[6]);
        let distinct: std::collections::HashSet<u32> = codes.iter().copied().collect();
        assert_eq!(distinct.len(), 5);
        for (i, &c) in codes.iter().enumerate() {
            let expected = vals[i].get("v").cloned().unwrap_or(serde_json::Value::Null);
            assert_eq!(reps[c as usize], expected, "representative for row {i}");
        }
    }

    #[test]
    fn group_codes_cover_typed_columns_and_unknown_props() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &json!({ "s": "x", "i": 7 })).unwrap();
        let b = g.add_node("N", &json!({ "s": "y", "i": 7 })).unwrap();
        let c = g.add_node("N", &json!({ "s": "x" })).unwrap();

        // Dictionary-encoded string column: same string, same code.
        let (codes, reps) = g.node_prop_group_codes(&[a, b, c, a], "s").unwrap();
        assert_eq!(codes[0], codes[2]);
        assert_eq!(codes[0], codes[3]);
        assert_ne!(codes[0], codes[1]);
        assert_eq!(reps[codes[0] as usize], json!("x"));

        // Int column with a null slot.
        let (codes, reps) = g.node_prop_group_codes(&[a, b, c], "i").unwrap();
        assert_eq!(codes[0], codes[1]);
        assert_ne!(codes[0], codes[2]);
        assert_eq!(reps[codes[2] as usize], serde_json::Value::Null);

        // Unknown property: every row is one null group.
        let (codes, reps) = g.node_prop_group_codes(&[a, b], "nope").unwrap();
        assert_eq!(codes, vec![0, 0]);
        assert_eq!(reps, vec![serde_json::Value::Null]);

        // Missing node is an error, like the table gather.
        assert!(g.node_prop_group_codes(&[a + 999], "s").is_err());
    }

    #[test]
    fn prop_column_gathers_in_input_order() {
        let (_dir, g) = open_tmp();
        let a = g
            .add_node("N", &json!({ "name": "ada", "age": 36 }))
            .unwrap();
        let b = g.add_node("N", &json!({ "name": "bob" })).unwrap();

        // Duplicate ids are allowed; a missing property reads as null.
        let col = g.node_prop_json_column(&[b, a, b], "age").unwrap();
        assert_eq!(
            col,
            vec![serde_json::Value::Null, json!(36), serde_json::Value::Null]
        );

        // An unknown property name yields a null column, not an error.
        let col = g.node_prop_json_column(&[a, b], "nope").unwrap();
        assert_eq!(col, vec![serde_json::Value::Null; 2]);

        // Empty input is fine; a missing node is an error, like the table.
        assert!(g.node_prop_json_column(&[], "age").unwrap().is_empty());
        let err = g.node_prop_json_column(&[a + 999], "age").unwrap_err();
        assert!(matches!(err, crate::error::Error::NodeNotFound(id) if id == a + 999));
    }

    #[test]
    fn props_table_errors_on_missing_node() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &json!({ "x": 1 })).unwrap();
        let err = g.node_props_json_table(&[a, a + 999], &["x"]).unwrap_err();
        assert!(matches!(err, crate::error::Error::NodeNotFound(id) if id == a + 999));
    }

    #[test]
    fn props_table_sees_committed_writes_immediately() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &json!({ "x": 1 })).unwrap();
        let table = g.node_props_json_table(&[a], &["x"]).unwrap();
        assert_eq!(table, vec![vec![json!(1)]]);

        g.update_node(a, &json!({ "x": 2 })).unwrap();
        let table = g.node_props_json_table(&[a], &["x"]).unwrap();
        assert_eq!(table, vec![vec![json!(2)]]);
    }

    #[test]
    fn batch_transaction_writes_are_visible() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &json!({ "x": 1 })).unwrap();
        assert_eq!(g.node_prop_json(a, "x").unwrap(), Some(json!(1)));

        let b = g
            .update(|txn| {
                txn.update_node(a, &json!({ "x": 10 }))?;
                txn.add_node("N", &json!({ "x": 20 }))
            })
            .unwrap();
        assert_eq!(g.node_prop_json(a, "x").unwrap(), Some(json!(10)));
        assert_eq!(g.node_prop_json(b, "x").unwrap(), Some(json!(20)));
    }

    #[test]
    fn prop_column_min_max_bounds() {
        let (_dir, g) = open_tmp();
        let _a = g
            .add_node(
                "N",
                &json!({ "age": 30, "weight": 70.5, "active": true, "name": "ada" }),
            )
            .unwrap();
        let _b = g
            .add_node(
                "N",
                &json!({ "age": 40, "weight": 80.2, "active": false, "name": "bob" }),
            )
            .unwrap();
        let c = g
            .add_node(
                "N",
                &json!({ "age": 20, "weight": 60.1, "active": true, "name": "charlie" }),
            )
            .unwrap();

        materialize_columns(&g);
        let (min_age, max_age) = g.node_prop_min_max("age").unwrap().unwrap();
        assert_eq!(min_age, json!(20));
        assert_eq!(max_age, json!(40));

        let (min_w, max_w) = g.node_prop_min_max("weight").unwrap().unwrap();
        assert_eq!(min_w, json!(60.1));
        assert_eq!(max_w, json!(80.2));

        let (min_act, max_act) = g.node_prop_min_max("active").unwrap().unwrap();
        assert_eq!(min_act, json!(false));
        assert_eq!(max_act, json!(true));

        let (min_name, max_name) = g.node_prop_min_max("name").unwrap().unwrap();
        assert_eq!(min_name, json!("ada"));
        assert_eq!(max_name, json!("charlie"));

        // An unknown property has no statistics.
        assert!(g.node_prop_min_max("nope").unwrap().is_none());

        // An update invalidates the cached statistics: 20 is gone and 50 is
        // the new maximum.
        g.update_node(c, &json!({ "age": 50, "weight": 90.0 }))
            .unwrap();
        let (min_age, max_age) = g.node_prop_min_max("age").unwrap().unwrap();
        assert_eq!(min_age, json!(30));
        assert_eq!(max_age, json!(50));
    }

    /// A point read must agree with the column read for every representable
    /// value, since the two now serve the same call depending only on whether
    /// the columns happen to exist.
    #[test]
    fn direct_point_read_agrees_with_the_column_read() {
        let (_dir, g) = open_tmp();
        let props = json!({
            "i": 42,
            "neg": -7,
            "big": i64::MAX,
            "f": 1.5,
            "whole_f": 3.0,
            "b": true,
            "s": "hello",
            "empty_s": "",
            "null": serde_json::Value::Null,
            "list": [1, 2, 3],
            "obj": { "k": "v" },
        });
        let n = g.add_node("N", &props).unwrap();
        let keys: Vec<&str> = props
            .as_object()
            .unwrap()
            .keys()
            .map(String::as_str)
            .chain(std::iter::once("absent"))
            .collect();

        // Columns absent: these go straight to storage.
        assert!(!g.prop_columns.is_built());
        let direct: Vec<_> = keys
            .iter()
            .map(|k| g.node_prop_json(n, k).unwrap())
            .collect();

        // Force the columns, then read the same values through them.
        materialize_columns(&g);
        assert!(g.prop_columns.is_built());
        let columnar: Vec<_> = keys
            .iter()
            .map(|k| g.node_prop_json(n, k).unwrap())
            .collect();

        for ((k, d), c) in keys.iter().zip(direct).zip(columnar) {
            assert_eq!(d, c, "property {k} disagrees between the two read paths");
        }
    }

    #[test]
    fn direct_point_read_reports_a_missing_node_as_none() {
        let (_dir, g) = open_tmp();
        assert!(!g.prop_columns.is_built());
        assert_eq!(g.node_prop_json(9999, "x").unwrap(), None);
    }

    /// The advisory statistics must not build the columns: that build is the
    /// dominant cold-start cost, and an absent estimate only costs plan quality.
    #[test]
    fn advisory_statistics_do_not_build_the_columns() {
        let (_dir, g) = open_tmp();
        g.add_node("N", &json!({ "age": 30 })).unwrap();

        assert_eq!(g.node_prop_min_max("age").unwrap(), None);
        assert_eq!(
            g.estimate_equality_selectivity("age", &json!(30)).unwrap(),
            None
        );
        assert_eq!(
            g.estimate_range_selectivity("age", Some(&json!(0)), None)
                .unwrap(),
            None
        );
        assert!(
            !g.prop_columns.is_built(),
            "an advisory read built the columns"
        );

        // Once a gather has built them, the same readers answer.
        materialize_columns(&g);
        assert_eq!(
            g.node_prop_min_max("age").unwrap(),
            Some((json!(30), json!(30)))
        );
        assert!(
            g.estimate_equality_selectivity("age", &json!(30))
                .unwrap()
                .is_some()
        );
    }

    /// A node deletion forces a full rebuild, and an advisory reader must not
    /// pay for it: it drops the stale columns and declines, leaving the rebuild
    /// to the next gather.
    #[test]
    fn a_pending_deletion_makes_the_advisory_readers_decline() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &json!({ "age": 10 })).unwrap();
        g.add_node("N", &json!({ "age": 20 })).unwrap();
        materialize_columns(&g);
        assert!(g.node_prop_min_max("age").unwrap().is_some());

        g.delete_node(a).unwrap();
        assert_eq!(g.node_prop_min_max("age").unwrap(), None);
        assert!(
            !g.prop_columns.is_built(),
            "the advisory read paid for a full rebuild"
        );

        // A gather rebuilds them, and the statistics then reflect the deletion.
        materialize_columns(&g);
        assert_eq!(
            g.node_prop_min_max("age").unwrap(),
            Some((json!(20), json!(20)))
        );
    }

    /// Sustained point reads must fall back to building, so a row pipeline that
    /// reads one property per row does not pay the per-read decode forever.
    #[test]
    fn sustained_direct_reads_amortize_into_a_build() {
        let (_dir, g) = open_tmp();
        let n = g.add_node("N", &json!({ "age": 30 })).unwrap();
        for _ in 0..DIRECT_READ_BUILD_THRESHOLD {
            assert_eq!(g.node_prop_json(n, "age").unwrap(), Some(json!(30)));
        }
        assert!(
            g.prop_columns.is_built(),
            "the columns never built despite {DIRECT_READ_BUILD_THRESHOLD} direct reads"
        );
        assert_eq!(g.node_prop_json(n, "age").unwrap(), Some(json!(30)));
    }

    #[test]
    fn prop_stats_refresh_when_update_removes_the_property() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &json!({ "age": 10 })).unwrap();
        let _b = g.add_node("N", &json!({ "age": 99 })).unwrap();
        materialize_columns(&g);
        let (_, max_age) = g.node_prop_min_max("age").unwrap().unwrap();
        assert_eq!(max_age, json!(99));

        // The new record no longer carries `age` at all, so the key is absent
        // from the patched property map; the stats must still refresh.
        g.update_node(a + 1, &json!({ "renamed": 1 })).unwrap();
        let (min_age, max_age) = g.node_prop_min_max("age").unwrap().unwrap();
        assert_eq!(min_age, json!(10));
        assert_eq!(max_age, json!(10));
    }

    #[test]
    fn equality_selectivity_uses_most_common_values() {
        let (_dir, g) = open_tmp();
        for _ in 0..90 {
            g.add_node("N", &json!({ "team": "blue" })).unwrap();
        }
        for i in 0..10 {
            g.add_node("N", &json!({ "team": format!("t{i}") }))
                .unwrap();
        }
        materialize_columns(&g);
        let sel = g
            .estimate_equality_selectivity("team", &json!("blue"))
            .unwrap()
            .unwrap();
        assert!((sel - 0.9).abs() < 1e-9, "got {sel}");
        // A value outside the column's bounds estimates to zero.
        let sel = g
            .estimate_equality_selectivity("team", &json!("zzz"))
            .unwrap()
            .unwrap();
        assert_eq!(sel, 0.0);
        // No statistics exist for an unknown property.
        assert!(
            g.estimate_equality_selectivity("nope", &json!(1))
                .unwrap()
                .is_none()
        );
    }

    #[test]
    fn range_selectivity_estimates_fraction() {
        let (_dir, g) = open_tmp();
        for i in 0..100 {
            g.add_node("N", &json!({ "age": i })).unwrap();
        }
        materialize_columns(&g);
        let sel = g
            .estimate_range_selectivity("age", Some(&json!(50)), None)
            .unwrap()
            .unwrap();
        assert!((sel - 0.5).abs() < 0.05, "got {sel}");
        let sel = g
            .estimate_range_selectivity("age", None, Some(&json!(1000)))
            .unwrap()
            .unwrap();
        assert!((sel - 1.0).abs() < 1e-9, "got {sel}");
    }

    // ------------------------------------------------------------------
    // Edge property columns
    // ------------------------------------------------------------------

    #[test]
    fn edge_typed_values_round_trip_and_missing_semantics() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &()).unwrap();
        let b = g.add_node("N", &()).unwrap();
        let e = g
            .add_edge(
                a,
                b,
                "E",
                &json!({ "i": 7, "s": "hit", "f": 1.5, "b": true }),
            )
            .unwrap();

        assert_eq!(g.edge_prop_json(e, "i").unwrap(), Some(json!(7)));
        assert_eq!(g.edge_prop_json(e, "s").unwrap(), Some(json!("hit")));
        assert_eq!(g.edge_prop_json(e, "f").unwrap(), Some(json!(1.5)));
        assert_eq!(g.edge_prop_json(e, "b").unwrap(), Some(json!(true)));
        // Missing property is null; nonexistent edge is None.
        assert_eq!(
            g.edge_prop_json(e, "nope").unwrap(),
            Some(serde_json::Value::Null)
        );
        assert_eq!(g.edge_prop_json(e + 999, "i").unwrap(), None);
    }

    #[test]
    fn edge_table_and_column_gather_in_input_order() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &()).unwrap();
        let b = g.add_node("N", &()).unwrap();
        let e1 = g.add_edge(a, b, "E", &json!({ "pa": 1, "hr": 1 })).unwrap();
        let e2 = g.add_edge(a, b, "E", &json!({ "pa": 1 })).unwrap();

        let table = g
            .edge_props_json_table(&[e2, e1, e2], &["pa", "hr"])
            .unwrap();
        assert_eq!(
            table,
            vec![
                vec![json!(1), serde_json::Value::Null],
                vec![json!(1), json!(1)],
                vec![json!(1), serde_json::Value::Null],
            ]
        );
        let col = g.edge_prop_json_column(&[e2, e1], "hr").unwrap();
        assert_eq!(col, vec![serde_json::Value::Null, json!(1)]);
        // A nonexistent edge is an error, like the node gather.
        assert!(g.edge_props_json_table(&[e1 + 999], &["pa"]).is_err());
    }

    #[test]
    fn edge_group_codes_match_value_identity() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &()).unwrap();
        let b = g.add_node("N", &()).unwrap();
        let e1 = g.add_edge(a, b, "E", &json!({ "hand": "L" })).unwrap();
        let e2 = g.add_edge(a, b, "E", &json!({ "hand": "R" })).unwrap();
        let e3 = g.add_edge(a, b, "E", &json!({ "hand": "L" })).unwrap();

        let (codes, reps) = g.edge_prop_group_codes(&[e1, e2, e3], "hand").unwrap();
        assert_eq!(codes[0], codes[2]);
        assert_ne!(codes[0], codes[1]);
        assert_eq!(reps[codes[0] as usize], json!("L"));
        assert_eq!(reps[codes[1] as usize], json!("R"));
    }

    #[test]
    fn edge_update_is_visible_and_delete_forces_rebuild() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &()).unwrap();
        let b = g.add_node("N", &()).unwrap();
        let e1 = g.add_edge(a, b, "E", &json!({ "w": 1 })).unwrap();
        let e2 = g.add_edge(a, b, "E", &json!({ "w": 2 })).unwrap();
        assert_eq!(g.edge_prop_json(e1, "w").unwrap(), Some(json!(1)));

        g.update_edge(e1, &json!({ "w": 10 })).unwrap();
        assert_eq!(g.edge_prop_json(e1, "w").unwrap(), Some(json!(10)));

        g.delete_edge(e1).unwrap();
        // The surviving edge is still readable after the forced rebuild.
        assert_eq!(g.edge_prop_json(e2, "w").unwrap(), Some(json!(2)));
        assert_eq!(g.edge_prop_json(e1, "w").unwrap(), None);
    }

    #[test]
    fn edge_columns_built_in_batch_transaction_are_visible() {
        let (_dir, g) = open_tmp();
        let (a, b) = g
            .update(|txn| {
                let a = txn.add_node("N", &())?;
                let b = txn.add_node("N", &())?;
                Ok((a, b))
            })
            .unwrap();
        let e = g
            .update(|txn| txn.add_edge(a, b, "E", &json!({ "k": 42 })))
            .unwrap();
        assert_eq!(g.edge_prop_json(e, "k").unwrap(), Some(json!(42)));
    }

    /// Deleting a node cascades to its incident edges, so those edges must drop
    /// out of the edge property columns too. Without the forced rebuild a deleted
    /// edge stays readable through `edge_prop_json`.
    #[test]
    fn delete_node_cascade_invalidates_edge_columns() {
        let (_dir, g) = open_tmp();
        let a = g.add_node("N", &()).unwrap();
        let b = g.add_node("N", &()).unwrap();
        let e = g.add_edge(a, b, "E", &json!({ "w": 7 })).unwrap();
        // Materialize the edge columns.
        assert_eq!(g.edge_prop_json(e, "w").unwrap(), Some(json!(7)));

        // Deleting a cascades the deletion of edge e.
        g.delete_node(a).unwrap();
        assert!(g.get_edge(e).unwrap().is_none(), "edge is gone from LMDB");
        assert_eq!(
            g.edge_prop_json(e, "w").unwrap(),
            None,
            "a cascade-deleted edge must not remain readable via the columns"
        );
    }
}