linkrs-storage 0.1.0

//! Column Statistics Snapshot Reader
//!
//! Exposes per-column storage statistics to the query optimizer as immutable
//! snapshots. Data sources are the storage-side zone maps and persisted
//! column stats, so reading a snapshot costs O(chunk count) rather than
//! O(row count) and never materializes rows.
//!
//! The trait methods are defaulted so test stubs and partial engines only
//! implement what they can serve; consumers must fall back conservatively
//! when a snapshot is unavailable.

use std::sync::Arc;

use linkrs_core::Value;

/// Immutable point-in-time statistics for one property column of a vertex
/// tag or an edge type.
#[derive(Debug, Clone, Default)]
pub struct ColumnStatsSnapshot {
    /// Live row estimate for the owning relation (tag / edge type).
    pub row_count: u64,
    /// Number of null values, when tracked by the storage layer.
    pub null_count: Option<u64>,
    /// Distinct value count excluding nulls, as an HLL union estimate over
    /// available registers. Retained when any contributing table tracks it.
    pub distinct_count: Option<u64>,
    /// HLL registers backing the distinct estimate when available.
    pub hll: Option<crate::stats::HyperLogLog>,
    /// Global minimum over non-null values (zone-map bounds are conservative:
    /// they only widen after writes, never shrink).
    pub min_value: Option<Value>,
    /// Global maximum over non-null values.
    pub max_value: Option<Value>,
}

/// Immutable point-in-time cardinality for one vertex tag or edge type.
///
/// Derived from live versus allocated slot counts, so optimizers get a
/// table-level row estimate without reassembling it from column snapshots.
/// Counts come from shard-inconsistent reads; use them for sizing and plan
/// costing, never as a strongly consistent census. Never persisted;
/// recomputed per timestamp.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct TableCardinalitySnapshot {
    /// Live rows visible at the snapshot timestamp.
    pub live_rows: u64,
    /// Allocated slots including deleted-but-unreclaimed holes.
    pub allocated_slots: u64,
    /// Shards contributing to the counts.
    pub shard_count: usize,
}

impl TableCardinalitySnapshot {
    /// Deleted-but-unreclaimed slots.
    pub fn hole_count(&self) -> u64 {
        self.allocated_slots.saturating_sub(self.live_rows)
    }

    /// Hole rate `1 - live / allocated`, zero for empty tables.
    pub fn hole_rate(&self) -> f64 {
        if self.allocated_slots == 0 || self.live_rows >= self.allocated_slots {
            0.0
        } else {
            1.0 - (self.live_rows as f64 / self.allocated_slots as f64)
        }
    }

    /// Whether the snapshot carries any rows at all.
    pub fn is_empty(&self) -> bool {
        self.live_rows == 0 && self.allocated_slots == 0
    }

    /// Merge `other` into `self`: counts and shard counts add up, the hole
    /// rate is always recomputed from the merged totals, never averaged.
    pub fn absorb(&mut self, other: &TableCardinalitySnapshot) {
        self.live_rows += other.live_rows;
        self.allocated_slots += other.allocated_slots;
        self.shard_count += other.shard_count;
    }
}

impl ColumnStatsSnapshot {
    /// Whether this snapshot carries any usable bound for selectivity
    /// estimation.
    pub fn has_envelope(&self) -> bool {
        self.min_value.is_some() || self.max_value.is_some()
    }

    /// Merge `other` into `self` so that `self` covers the union of both
    /// snapshots' rows:
    /// - row counts add up;
    /// - min/max bounds widen using the numeric-aware comparison shared with
    ///   pushed-predicate evaluation;
    /// - null counts sum over the tables that track them (missing tracking
    ///   is treated as no contribution);
    /// - distinct counts merge HLL registers into a union estimate when both
    ///   sides carry HLL state; when only numeric estimates exist without
    ///   registers the merge degrades to `None` instead of summing (a sum
    ///   would be a loose upper bound that double-counts overlap).
    pub fn absorb(&mut self, other: &ColumnStatsSnapshot) {
        self.row_count += other.row_count;
        if let Some(n) = other.null_count {
            *self.null_count.get_or_insert(0) += n;
        }
        match (self.hll.clone(), other.hll.clone()) {
            (Some(mut a), Some(b)) => {
                a.merge(&b);
                self.distinct_count = Some(a.estimate());
                self.hll = Some(a);
            }
            (Some(a), None) => {
                self.distinct_count = Some(a.estimate());
                self.hll = Some(a);
            }
            (None, Some(b)) => {
                self.distinct_count = Some(b.estimate());
                self.hll = Some(b);
            }
            (None, None) => {
                self.distinct_count = None;
                self.hll = None;
            }
        }
        merge_min(&mut self.min_value, other.min_value.clone());
        merge_max(&mut self.max_value, other.max_value.clone());
    }

    /// Attach an HLL estimator and refresh the distinct estimate from it.
    pub fn set_hll(&mut self, hll: crate::stats::HyperLogLog) {
        self.distinct_count = Some(hll.estimate());
        self.hll = Some(hll);
    }
}

/// Keep the smaller of the two optional bounds, using the same numeric-aware
/// comparison as pushed-predicate evaluation so merged envelopes stay
/// consistent with zone-map pruning semantics.
pub(crate) fn merge_min(cur: &mut Option<Value>, candidate: Option<Value>) {
    let Some(v) = candidate else { return };
    match cur {
        Some(c)
            if crate::vertex::column_store::compare_values(c, &v)
                != std::cmp::Ordering::Greater => {}
        _ => *cur = Some(v),
    }
}

/// Keep the larger of the two optional bounds; see [`merge_min`].
pub(crate) fn merge_max(cur: &mut Option<Value>, candidate: Option<Value>) {
    let Some(v) = candidate else { return };
    match cur {
        Some(c)
            if crate::vertex::column_store::compare_values(c, &v) != std::cmp::Ordering::Less => {}
        _ => *cur = Some(v),
    }
}

/// Reader capability for optimizer-facing column statistics snapshots.
///
/// `relation` addressing follows the catalog naming: a tag name for vertex
/// columns and an edge type name for edge columns.
pub trait ColumnStatsReader: Send + Sync {
    /// Snapshot of one property column of `tag` inside `space`.
    fn vertex_column_stats(
        &self,
        _space: &str,
        _tag: &str,
        _column: &str,
    ) -> Option<Arc<ColumnStatsSnapshot>> {
        None
    }

    /// Snapshot of one property column of `edge_type` inside `space`.
    fn edge_column_stats(
        &self,
        _space: &str,
        _edge_type: &str,
        _column: &str,
    ) -> Option<Arc<ColumnStatsSnapshot>> {
        None
    }

    /// Table-level cardinality of vertex tag `tag` inside `space`.
    fn vertex_table_stats(
        &self,
        _space: &str,
        _tag: &str,
    ) -> Option<Arc<TableCardinalitySnapshot>> {
        None
    }

    /// Table-level cardinality of edge type `edge_type` inside `space`.
    ///
    /// Served by the engine from per-partition snapshot counts merged with
    /// [`TableCardinalitySnapshot::absorb`]; absent only when the edge type
    /// has no partitions. Costing falls back to live edge counts when
    /// unavailable.
    fn edge_table_stats(
        &self,
        _space: &str,
        _edge_type: &str,
    ) -> Option<Arc<TableCardinalitySnapshot>> {
        None
    }

    /// Monotonic data-version stamp: changes whenever data has been written,
    /// independent of schema changes. Used by the optimizer's statistics
    /// cache to detect stale estimates after DML.
    ///
    /// Contract: `0` means "no writes observed / unknown" and is reserved;
    /// engines backed by real data must return values that strictly increase
    /// across DML (they may start at any value > 0). Consumers must treat a
    /// return value of `0` as "cacheability unknown" rather than as a valid
    /// epoch. Returns 0 when unknown.
    fn stats_epoch(&self) -> u64 {
        0
    }
}

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

    fn snap(row_count: u64, min: Option<Value>, max: Option<Value>) -> ColumnStatsSnapshot {
        ColumnStatsSnapshot {
            row_count,
            null_count: None,
            distinct_count: None,
            hll: None,
            min_value: min,
            max_value: max,
        }
    }

    fn hll_with(values: &[i32]) -> crate::stats::HyperLogLog {
        let mut hll = crate::stats::HyperLogLog::new();
        for v in values {
            hll.add_value(&Value::Int(*v));
        }
        hll
    }

    #[test]
    fn absorb_widens_bounds_across_mixed_numeric_kinds() {
        let mut acc = snap(10, Some(Value::Int(1)), Some(Value::BigInt(100)));
        acc.null_count = Some(2);
        acc.set_hll(hll_with(&(0..100).collect::<Vec<_>>()));

        let mut other = snap(5, Some(Value::Double(-0.5)), Some(Value::SmallInt(50)));
        other.null_count = Some(1);
        other.set_hll(hll_with(&(100..150).collect::<Vec<_>>()));
        acc.absorb(&other);

        assert_eq!(acc.row_count, 15);
        assert_eq!(acc.null_count, Some(3));
        // HLL union estimate tracks the true union instead of summing.
        let est = acc.distinct_count.unwrap() as f64;
        assert!((est - 150.0).abs() / 150.0 < 0.2, "estimate={}", est);
        assert_eq!(acc.min_value, Some(Value::Double(-0.5)));
        assert_eq!(acc.max_value, Some(Value::BigInt(100)));
    }

    #[test]
    fn absorb_drops_partial_distinct_counts() {
        let mut acc = snap(4, None, None);
        acc.set_hll(hll_with(&[1, 2, 3]));
        let expected = acc.distinct_count;
        acc.absorb(&snap(2, None, None));
        assert_eq!(acc.distinct_count, expected);
    }

    #[test]
    fn absorb_without_hll_registers_degrades_to_unknown() {
        // Bare numeric estimates without registers must not be summed.
        let mut acc = snap(4, None, None);
        acc.distinct_count = Some(9);
        let mut other = snap(2, None, None);
        other.distinct_count = Some(3);
        acc.absorb(&other);
        assert_eq!(acc.distinct_count, None);
    }

    #[test]
    fn absorb_keeps_null_tracking_when_only_one_side_tracks() {
        let mut acc = snap(1, None, None);
        acc.null_count = None;
        let mut other = snap(1, None, None);
        other.null_count = Some(1);
        acc.absorb(&other);
        assert_eq!(acc.null_count, Some(1));

        // And the reverse order keeps the tracked contribution too.
        let mut acc = snap(1, None, None);
        acc.null_count = Some(1);
        acc.absorb(&snap(1, None, None));
        assert_eq!(acc.null_count, Some(1));
    }

    #[test]
    fn merge_helpers_compare_cross_type_numerically() {
        let mut min = Some(Value::Int(-2));
        merge_min(&mut min, Some(Value::Double(-2.5)));
        assert_eq!(min, Some(Value::Double(-2.5)));

        let mut max = Some(Value::Float(1.5));
        merge_max(&mut max, Some(Value::SmallInt(2)));
        assert_eq!(max, Some(Value::SmallInt(2)));

        // Non-numeric kinds fall back to native Value ordering.
        let mut max = Some(Value::string("P99"));
        merge_max(&mut max, Some(Value::string("P100")));
        assert_eq!(max, Some(Value::string("P99")));
    }

    #[test]
    fn table_cardinality_absorb_sums_and_recomputes_hole_rate() {
        let mut acc = TableCardinalitySnapshot {
            live_rows: 70,
            allocated_slots: 100,
            shard_count: 2,
        };
        let other = TableCardinalitySnapshot {
            live_rows: 30,
            allocated_slots: 50,
            shard_count: 1,
        };
        acc.absorb(&other);
        assert_eq!(acc.live_rows, 100);
        assert_eq!(acc.allocated_slots, 150);
        assert_eq!(acc.shard_count, 3);
        assert_eq!(acc.hole_count(), 50);
        let rate = acc.hole_rate();
        assert!((rate - (1.0 - 100.0 / 150.0)).abs() < 1e-9);
        assert!(!acc.is_empty());
        let empty = TableCardinalitySnapshot::default();
        assert!(empty.is_empty());
        assert_eq!(empty.hole_rate(), 0.0);
    }
}