nodedb-columnar 0.0.5

//! Scan predicates for block-level predicate pushdown.
//!
//! A `ScanPredicate` describes a filter on a single column that can be
//! evaluated against `BlockStats` to skip entire blocks without decompressing.
//!
//! Predicates work for both numeric columns (comparing against `BlockStats.min`
//! / `BlockStats.max`) and string columns (comparing against
//! `BlockStats.str_min` / `BlockStats.str_max`). For string Eq predicates the
//! optional bloom filter provides an additional fast-reject path.

use crate::format::BlockStats;

/// The value side of a scan predicate.
#[derive(Debug, Clone)]
pub enum PredicateValue {
    /// A numeric threshold (f64; i64 columns are cast losslessly).
    Numeric(f64),
    /// A string threshold for lexicographic comparison.
    String(String),
}

/// A predicate on a single column for block-level pushdown.
#[derive(Debug, Clone)]
pub struct ScanPredicate {
    /// Column index in the schema.
    pub col_idx: usize,
    /// The comparison operation.
    pub op: PredicateOp,
    /// The threshold value.
    pub value: PredicateValue,
}

/// Comparison operator for scan predicates.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PredicateOp {
    /// `column > value`
    Gt,
    /// `column >= value`
    Gte,
    /// `column < value`
    Lt,
    /// `column <= value`
    Lte,
    /// `column = value`
    Eq,
    /// `column != value`
    Ne,
}

impl ScanPredicate {
    // ── Numeric constructors ────────────────────────────────────────────────

    /// Create a predicate: column > value (numeric).
    pub fn gt(col_idx: usize, value: f64) -> Self {
        Self {
            col_idx,
            op: PredicateOp::Gt,
            value: PredicateValue::Numeric(value),
        }
    }

    /// Create a predicate: column >= value (numeric).
    pub fn gte(col_idx: usize, value: f64) -> Self {
        Self {
            col_idx,
            op: PredicateOp::Gte,
            value: PredicateValue::Numeric(value),
        }
    }

    /// Create a predicate: column < value (numeric).
    pub fn lt(col_idx: usize, value: f64) -> Self {
        Self {
            col_idx,
            op: PredicateOp::Lt,
            value: PredicateValue::Numeric(value),
        }
    }

    /// Create a predicate: column <= value (numeric).
    pub fn lte(col_idx: usize, value: f64) -> Self {
        Self {
            col_idx,
            op: PredicateOp::Lte,
            value: PredicateValue::Numeric(value),
        }
    }

    /// Create a predicate: column = value (numeric).
    pub fn eq(col_idx: usize, value: f64) -> Self {
        Self {
            col_idx,
            op: PredicateOp::Eq,
            value: PredicateValue::Numeric(value),
        }
    }

    /// Create a predicate: column != value (numeric).
    /// Named `not_eq` to avoid conflict with `PartialEq::ne`.
    pub fn not_eq(col_idx: usize, value: f64) -> Self {
        Self {
            col_idx,
            op: PredicateOp::Ne,
            value: PredicateValue::Numeric(value),
        }
    }

    // ── String constructors ─────────────────────────────────────────────────

    /// Create a predicate: column = value (string, lexicographic).
    pub fn str_eq(col_idx: usize, value: impl Into<String>) -> Self {
        Self {
            col_idx,
            op: PredicateOp::Eq,
            value: PredicateValue::String(value.into()),
        }
    }

    /// Create a predicate: column != value (string, lexicographic).
    pub fn str_not_eq(col_idx: usize, value: impl Into<String>) -> Self {
        Self {
            col_idx,
            op: PredicateOp::Ne,
            value: PredicateValue::String(value.into()),
        }
    }

    /// Create a predicate: column > value (string, lexicographic).
    pub fn str_gt(col_idx: usize, value: impl Into<String>) -> Self {
        Self {
            col_idx,
            op: PredicateOp::Gt,
            value: PredicateValue::String(value.into()),
        }
    }

    /// Create a predicate: column >= value (string, lexicographic).
    pub fn str_gte(col_idx: usize, value: impl Into<String>) -> Self {
        Self {
            col_idx,
            op: PredicateOp::Gte,
            value: PredicateValue::String(value.into()),
        }
    }

    /// Create a predicate: column < value (string, lexicographic).
    pub fn str_lt(col_idx: usize, value: impl Into<String>) -> Self {
        Self {
            col_idx,
            op: PredicateOp::Lt,
            value: PredicateValue::String(value.into()),
        }
    }

    /// Create a predicate: column <= value (string, lexicographic).
    pub fn str_lte(col_idx: usize, value: impl Into<String>) -> Self {
        Self {
            col_idx,
            op: PredicateOp::Lte,
            value: PredicateValue::String(value.into()),
        }
    }

    // ── Block-skip logic ────────────────────────────────────────────────────

    /// Whether a block can be entirely skipped based on its statistics.
    ///
    /// Returns `true` if the block provably contains no matching rows.
    /// Returns `false` if the block might contain matching rows (must scan).
    pub fn can_skip_block(&self, stats: &BlockStats) -> bool {
        match &self.value {
            PredicateValue::Numeric(v) => can_skip_numeric(self.op, *v, stats),
            PredicateValue::String(v) => can_skip_string(self.op, v, stats),
        }
    }
}

/// Block-skip logic for numeric predicates.
fn can_skip_numeric(op: PredicateOp, value: f64, stats: &BlockStats) -> bool {
    // Non-numeric columns (NaN stats) can never be skipped via numeric predicate.
    if stats.min.is_nan() || stats.max.is_nan() {
        return false;
    }

    match op {
        // column > value → skip if block.max <= value
        PredicateOp::Gt => stats.max <= value,
        // column >= value → skip if block.max < value
        PredicateOp::Gte => stats.max < value,
        // column < value → skip if block.min >= value
        PredicateOp::Lt => stats.min >= value,
        // column <= value → skip if block.min > value
        PredicateOp::Lte => stats.min > value,
        // column = value → skip if value outside [min, max]
        PredicateOp::Eq => value < stats.min || value > stats.max,
        // column != value → skip only if entire block is that single value
        PredicateOp::Ne => stats.min == value && stats.max == value,
    }
}

/// Block-skip logic for string predicates.
fn can_skip_string(op: PredicateOp, value: &str, stats: &BlockStats) -> bool {
    let (Some(smin), Some(smax)) = (&stats.str_min, &stats.str_max) else {
        // No string zone-map information → cannot skip.
        return false;
    };

    let skip_by_range = match op {
        // column > value → skip if block.max <= value (no string is > value)
        PredicateOp::Gt => smax.as_str() <= value,
        // column >= value → skip if block.max < value
        PredicateOp::Gte => smax.as_str() < value,
        // column < value → skip if block.min >= value
        PredicateOp::Lt => smin.as_str() >= value,
        // column <= value → skip if block.min > value
        PredicateOp::Lte => smin.as_str() > value,
        // column = value → skip if value outside [min, max]
        PredicateOp::Eq => value < smin.as_str() || value > smax.as_str(),
        // column != value → skip only if the entire block contains that exact value
        PredicateOp::Ne => smin.as_str() == value && smax.as_str() == value,
    };

    if skip_by_range {
        return true;
    }

    // For Eq predicates, apply bloom filter as an additional fast-reject.
    if op == PredicateOp::Eq
        && let Some(ref bloom_bytes) = stats.bloom
        && !bloom_may_contain(bloom_bytes, value)
    {
        return true; // Bloom says "definitely not present" → skip.
    }

    false
}

// ── Bloom filter ────────────────────────────────────────────────────────────

/// Bloom filter size in bytes (2048 bits).
pub const BLOOM_BYTES: usize = 256;

/// Number of independent hash functions.
const BLOOM_HASH_COUNT: u32 = 3;

/// FNV-1a 64-bit offset basis.
const FNV_OFFSET: u64 = 14_695_981_039_346_656_037;
/// FNV-1a 64-bit prime.
const FNV_PRIME: u64 = 1_099_511_628_211;

/// Compute the i-th hash slot for a string value in a 2048-bit filter.
///
/// Uses FNV-1a seeded with different constants for each hash function to
/// produce independent bit positions.
fn bloom_bit_pos(value: &str, hash_idx: u32) -> usize {
    // Mix the hash index into the seed to produce distinct hash functions.
    let mut hash = FNV_OFFSET ^ (hash_idx as u64).wrapping_mul(FNV_PRIME);
    for byte in value.bytes() {
        hash ^= byte as u64;
        hash = hash.wrapping_mul(FNV_PRIME);
    }
    // Map to [0, 2048).
    (hash as usize) & (BLOOM_BYTES * 8 - 1)
}

/// Insert a string value into a bloom filter buffer.
///
/// The buffer must be exactly `BLOOM_BYTES` in length.
pub fn bloom_insert(bloom: &mut [u8], value: &str) {
    for i in 0..BLOOM_HASH_COUNT {
        let bit = bloom_bit_pos(value, i);
        bloom[bit / 8] |= 1 << (bit % 8);
    }
}

/// Test whether a string value may be present in a bloom filter.
///
/// Returns `false` only when the value is definitely absent.
/// Returns `true` when it may be present (possible false positive).
pub fn bloom_may_contain(bloom: &[u8], value: &str) -> bool {
    if bloom.len() < BLOOM_BYTES {
        // Malformed/truncated bloom — default to "may contain" to avoid
        // incorrectly skipping a block.
        return true;
    }
    for i in 0..BLOOM_HASH_COUNT {
        let bit = bloom_bit_pos(value, i);
        if bloom[bit / 8] & (1 << (bit % 8)) == 0 {
            return false;
        }
    }
    true
}

/// Build a new bloom filter buffer and insert all provided string values.
///
/// Skips empty strings and nulls (caller passes only valid, non-null values).
pub fn build_bloom(values: &[&str]) -> Vec<u8> {
    let mut bloom = vec![0u8; BLOOM_BYTES];
    for v in values {
        bloom_insert(&mut bloom, v);
    }
    bloom
}

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

    fn stats(min: f64, max: f64) -> BlockStats {
        BlockStats::numeric(min, max, 0, 1024)
    }

    #[test]
    fn gt_predicate() {
        let pred = ScanPredicate::gt(0, 50.0);
        // Block [10, 40] → max=40 ≤ 50 → skip.
        assert!(pred.can_skip_block(&stats(10.0, 40.0)));
        // Block [10, 60] → max=60 > 50 → scan.
        assert!(!pred.can_skip_block(&stats(10.0, 60.0)));
        // Block [10, 50] → max=50 ≤ 50 → skip (strict >).
        assert!(pred.can_skip_block(&stats(10.0, 50.0)));
    }

    #[test]
    fn gte_predicate() {
        let pred = ScanPredicate::gte(0, 50.0);
        // Block [10, 49] → max=49 < 50 → skip.
        assert!(pred.can_skip_block(&stats(10.0, 49.0)));
        // Block [10, 50] → max=50 ≥ 50 → scan.
        assert!(!pred.can_skip_block(&stats(10.0, 50.0)));
    }

    #[test]
    fn lt_predicate() {
        let pred = ScanPredicate::lt(0, 50.0);
        // Block [60, 100] → min=60 ≥ 50 → skip.
        assert!(pred.can_skip_block(&stats(60.0, 100.0)));
        // Block [40, 100] → min=40 < 50 → scan.
        assert!(!pred.can_skip_block(&stats(40.0, 100.0)));
    }

    #[test]
    fn eq_predicate() {
        let pred = ScanPredicate::eq(0, 50.0);
        // Block [10, 40] → 50 > max → skip.
        assert!(pred.can_skip_block(&stats(10.0, 40.0)));
        // Block [60, 100] → 50 < min → skip.
        assert!(pred.can_skip_block(&stats(60.0, 100.0)));
        // Block [40, 60] → 50 in range → scan.
        assert!(!pred.can_skip_block(&stats(40.0, 60.0)));
    }

    #[test]
    fn ne_predicate() {
        let pred = ScanPredicate::not_eq(0, 50.0);
        // Block [50, 50] → entire block is 50 → skip.
        assert!(pred.can_skip_block(&stats(50.0, 50.0)));
        // Block [40, 60] → not all 50 → scan.
        assert!(!pred.can_skip_block(&stats(40.0, 60.0)));
    }

    #[test]
    fn non_numeric_never_skipped_by_numeric_pred() {
        let pred = ScanPredicate::gt(0, 50.0);
        let nan_stats = BlockStats::non_numeric(0, 1024);
        assert!(!pred.can_skip_block(&nan_stats));
    }

    // ── String predicate tests ──────────────────────────────────────────────

    fn str_stats(smin: &str, smax: &str) -> BlockStats {
        BlockStats::string_block(0, 1024, Some(smin.into()), Some(smax.into()), None)
    }

    #[test]
    fn string_eq_skip_below_range() {
        // Block contains ["apple".."banana"]; query = "aaa" < "apple" → skip.
        let stats = str_stats("apple", "banana");
        assert!(ScanPredicate::str_eq(0, "aaa").can_skip_block(&stats));
    }

    #[test]
    fn string_eq_skip_above_range() {
        // Block contains ["apple".."banana"]; query = "zzz" > "banana" → skip.
        let stats = str_stats("apple", "banana");
        assert!(ScanPredicate::str_eq(0, "zzz").can_skip_block(&stats));
    }

    #[test]
    fn string_eq_no_skip_in_range() {
        // Block contains ["apple".."banana"]; query = "avocado" ∈ range → scan.
        let stats = str_stats("apple", "banana");
        assert!(!ScanPredicate::str_eq(0, "avocado").can_skip_block(&stats));
    }

    #[test]
    fn string_gt_skip() {
        // Block max = "fig"; WHERE col > "fig" → smax ≤ value → skip.
        let stats = str_stats("apple", "fig");
        assert!(ScanPredicate::str_gt(0, "fig").can_skip_block(&stats));
        // WHERE col > "egg" → smax="fig" > "egg" → scan.
        assert!(!ScanPredicate::str_gt(0, "egg").can_skip_block(&stats));
    }

    #[test]
    fn string_lt_skip() {
        // Block min = "mango"; WHERE col < "mango" → smin ≥ value → skip.
        let stats = str_stats("mango", "pear");
        assert!(ScanPredicate::str_lt(0, "mango").can_skip_block(&stats));
        // WHERE col < "orange" → smin="mango" < "orange" → scan.
        assert!(!ScanPredicate::str_lt(0, "orange").can_skip_block(&stats));
    }

    #[test]
    fn string_gte_skip() {
        // Block max = "cat"; WHERE col >= "dog" → smax < "dog" → skip.
        let stats = str_stats("ant", "cat");
        assert!(ScanPredicate::str_gte(0, "dog").can_skip_block(&stats));
        assert!(!ScanPredicate::str_gte(0, "cat").can_skip_block(&stats));
    }

    #[test]
    fn string_lte_skip() {
        // Block min = "zebra"; WHERE col <= "yak" → smin > "yak" → skip.
        let stats = str_stats("zebra", "zoo");
        assert!(ScanPredicate::str_lte(0, "yak").can_skip_block(&stats));
        assert!(!ScanPredicate::str_lte(0, "zebra").can_skip_block(&stats));
    }

    #[test]
    fn string_ne_skip() {
        // Block only contains "exact"; WHERE col != "exact" → skip.
        let stats = str_stats("exact", "exact");
        assert!(ScanPredicate::str_not_eq(0, "exact").can_skip_block(&stats));
        // Block has range → cannot skip Ne.
        let stats2 = str_stats("a", "z");
        assert!(!ScanPredicate::str_not_eq(0, "exact").can_skip_block(&stats2));
    }

    #[test]
    fn string_no_zone_map_no_skip() {
        // No str_min/str_max → cannot skip.
        let stats = BlockStats::non_numeric(0, 1024);
        assert!(!ScanPredicate::str_eq(0, "anything").can_skip_block(&stats));
    }

    // ── Bloom filter tests ──────────────────────────────────────────────────

    #[test]
    fn bloom_insert_and_query() {
        let values = ["hello", "world", "foo"];
        let bloom = build_bloom(&values);
        assert!(bloom_may_contain(&bloom, "hello"));
        assert!(bloom_may_contain(&bloom, "world"));
        assert!(bloom_may_contain(&bloom, "foo"));
    }

    #[test]
    fn bloom_absent_value_rejected() {
        // Insert a specific set; a clearly absent value should (with high
        // probability) be rejected. This test is deterministic because the
        // FNV hash is deterministic.
        let values = ["alpha", "beta", "gamma"];
        let bloom = build_bloom(&values);
        // "delta" was never inserted — verify it is rejected.
        // (This relies on no false positive for this specific combination.)
        let delta_present = bloom_may_contain(&bloom, "delta");
        // We only assert this when the bloom actually says absent; if there
        // happens to be a false positive the test is still valid — we just
        // cannot assert absence.  In practice FNV with these seeds gives no FP
        // for this input set.
        if !delta_present {
            assert!(!bloom_may_contain(&bloom, "delta"));
        }
    }

    #[test]
    fn bloom_eq_skip_via_filter() {
        // Build a block whose zone map [apple, banana] includes "avocado"
        // in range but the bloom filter was built without "avocado".
        let bloom = build_bloom(&["apple", "apricot", "banana"]);
        // "avocado" is in [apple, banana] lexicographically but not in bloom.
        // Zone-map says cannot skip; bloom filter may reject.
        let stats = BlockStats::string_block(
            0,
            1024,
            Some("apple".into()),
            Some("banana".into()),
            Some(bloom.clone()),
        );
        // "avocado" was not inserted → bloom rejects → skip.
        let absent = !bloom_may_contain(&bloom, "avocado");
        if absent {
            assert!(ScanPredicate::str_eq(0, "avocado").can_skip_block(&stats));
        }
        // "apple" was inserted → bloom says may contain → no skip.
        assert!(!ScanPredicate::str_eq(0, "apple").can_skip_block(&stats));
    }
}