chronix-query 0.5.0

//! Encoded-domain predicate pushdown — evaluates predicates against column
//! statistics without decoding the actual data.
//!
//! [`EncodedDomainEvaluator`] uses per-row-group `min`/`max` statistics
//! stored in the segment metadata to determine whether a predicate can be
//! satisfied or excluded without reading and decoding the column bytes.
//! This can reduce CPU usage by up to 80% on filtered queries over sorted data.

use chronix_engine::index::{ZoneMapPredicate, ZoneMapResult};

/// Evaluates column predicates against row-group statistics
/// to avoid unnecessary decoding.
#[derive(Debug, Clone)]
pub struct EncodedDomainEvaluator {
    predicates: Vec<ColumnPredicate>,
}

/// A predicate on a specific named column.
#[derive(Debug, Clone)]
pub struct ColumnPredicate {
    /// Column name this predicate applies to.
    pub column: String,
    /// The zone-map predicate (comparison operator + value).
    pub predicate: ZoneMapPredicate,
}

/// Result of evaluating a row group's statistics against all predicates.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RowGroupVerdict {
    /// All predicates are satisfied by the statistics — the row group
    /// can be emitted without decoding the values column.
    AllMatch,
    /// At least one predicate definitely excludes this row group —
    /// skip it entirely (zero decoding needed).
    Skip,
    /// Statistics are inconclusive — the row group must be decoded
    /// and filtered row-by-row.
    Decode,
}

impl EncodedDomainEvaluator {
    /// Create a new evaluator with the given set of column predicates.
    #[must_use]
    pub fn new(predicates: Vec<ColumnPredicate>) -> Self {
        Self { predicates }
    }

    /// Evaluate a single row group's column statistics.
    ///
    /// `stats_fn` is called with each predicate's column name and should
    /// return `Some((min, max))` if statistics are available for that
    /// column in this row group, or `None` if unknown.
    ///
    /// # Returns
    ///
    /// - [`RowGroupVerdict::Skip`] if any predicate definitively excludes
    ///   the row group.
    /// - [`RowGroupVerdict::AllMatch`] if all predicates definitively match.
    ///   **Note:** `AllMatch` means every row satisfies the filter,
    ///   so filter evaluation can be skipped. However, the row group still
    ///   must be decoded if any filter column appears in the SELECT
    ///   projection (callers must check this separately). When stats are
    ///   unavailable for any predicate column, the verdict is downgraded
    ///   to `Decode` because we cannot prove all-match.
    /// - [`RowGroupVerdict::Decode`] otherwise.
    #[must_use]
    pub fn evaluate<F>(&self, stats_fn: F) -> RowGroupVerdict
    where
        F: Fn(&str) -> Option<(f64, f64)>,
    {
        if self.predicates.is_empty() {
            return RowGroupVerdict::Decode;
        }

        let mut all_match = true;

        for pred in &self.predicates {
            let Some((min, max)) = stats_fn(&pred.column) else {
                // No stats available — can't prune
                all_match = false;
                continue;
            };

            match pred.predicate.evaluate(min, max) {
                ZoneMapResult::NoneMatch => return RowGroupVerdict::Skip,
                ZoneMapResult::AllMatch => { /* this predicate matches fully */ }
                ZoneMapResult::MaybeMatch => {
                    all_match = false;
                }
            }
        }

        if all_match {
            RowGroupVerdict::AllMatch
        } else {
            RowGroupVerdict::Decode
        }
    }

    /// Batch evaluation: given a list of row groups with their stats,
    /// produce a verdict for each.
    ///
    /// `row_group_stats` is a slice of closures or a single closure that
    /// takes `(row_group_index, column_name)` and returns stats.
    #[must_use]
    pub fn evaluate_batch<F>(&self, num_row_groups: usize, stats_fn: F) -> Vec<RowGroupVerdict>
    where
        F: Fn(usize, &str) -> Option<(f64, f64)>,
    {
        (0..num_row_groups)
            .map(|rg| self.evaluate(|col| stats_fn(rg, col)))
            .collect()
    }
}

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

    fn make_evaluator(col: &str, op: ZoneMapOp, value: f64) -> EncodedDomainEvaluator {
        EncodedDomainEvaluator::new(vec![ColumnPredicate {
            column: col.to_string(),
            predicate: ZoneMapPredicate { op, value },
        }])
    }

    #[test]
    fn skip_when_predicate_excludes() {
        // Predicate: value > 100, row group has max=50
        let eval = make_evaluator("value", ZoneMapOp::Gt, 100.0);
        let verdict = eval.evaluate(|_col| Some((10.0, 50.0)));
        assert_eq!(verdict, RowGroupVerdict::Skip);
    }

    #[test]
    fn all_match_when_predicate_fully_satisfied() {
        // Predicate: value > 5, row group has min=10, max=100
        let eval = make_evaluator("value", ZoneMapOp::Gt, 5.0);
        let verdict = eval.evaluate(|_col| Some((10.0, 100.0)));
        assert_eq!(verdict, RowGroupVerdict::AllMatch);
    }

    #[test]
    fn decode_when_predicate_partially_matches() {
        // Predicate: value > 50, row group has min=10, max=100
        let eval = make_evaluator("value", ZoneMapOp::Gt, 50.0);
        let verdict = eval.evaluate(|_col| Some((10.0, 100.0)));
        assert_eq!(verdict, RowGroupVerdict::Decode);
    }

    #[test]
    fn decode_when_no_stats_available() {
        let eval = make_evaluator("value", ZoneMapOp::Gt, 50.0);
        let verdict = eval.evaluate(|_col| None);
        assert_eq!(verdict, RowGroupVerdict::Decode);
    }

    #[test]
    fn empty_predicates_returns_decode() {
        let eval = EncodedDomainEvaluator::new(vec![]);
        let verdict = eval.evaluate(|_col| Some((10.0, 100.0)));
        assert_eq!(verdict, RowGroupVerdict::Decode);
    }

    #[test]
    fn multiple_predicates_all_match() {
        // value > 5 AND value < 200
        let eval = EncodedDomainEvaluator::new(vec![
            ColumnPredicate {
                column: "value".to_string(),
                predicate: ZoneMapPredicate {
                    op: ZoneMapOp::Gt,
                    value: 5.0,
                },
            },
            ColumnPredicate {
                column: "value".to_string(),
                predicate: ZoneMapPredicate {
                    op: ZoneMapOp::Lt,
                    value: 200.0,
                },
            },
        ]);
        // Row group: [10, 100] — both predicates fully satisfied
        let verdict = eval.evaluate(|_col| Some((10.0, 100.0)));
        assert_eq!(verdict, RowGroupVerdict::AllMatch);
    }

    #[test]
    fn multiple_predicates_one_excludes() {
        // value > 5 AND value < 8
        let eval = EncodedDomainEvaluator::new(vec![
            ColumnPredicate {
                column: "value".to_string(),
                predicate: ZoneMapPredicate {
                    op: ZoneMapOp::Gt,
                    value: 5.0,
                },
            },
            ColumnPredicate {
                column: "value".to_string(),
                predicate: ZoneMapPredicate {
                    op: ZoneMapOp::Lt,
                    value: 8.0,
                },
            },
        ]);
        // Row group: [10, 100] — second predicate excludes (min=10 >= 8)
        let verdict = eval.evaluate(|_col| Some((10.0, 100.0)));
        assert_eq!(verdict, RowGroupVerdict::Skip);
    }

    #[test]
    fn batch_evaluation() {
        let eval = make_evaluator("value", ZoneMapOp::Gt, 50.0);
        let stats = [
            (10.0, 40.0),  // all below 50 → Skip
            (60.0, 100.0), // all above 50 → AllMatch
            (30.0, 80.0),  // straddles 50 → Decode
        ];
        let verdicts = eval.evaluate_batch(3, |rg, _col| Some(stats[rg]));
        assert_eq!(verdicts[0], RowGroupVerdict::Skip);
        assert_eq!(verdicts[1], RowGroupVerdict::AllMatch);
        assert_eq!(verdicts[2], RowGroupVerdict::Decode);
    }

    #[test]
    fn skip_entire_sorted_column() {
        // Simulates a segment with sorted column: 5 row groups, each
        // with non-overlapping ranges. Query: value == 75.
        let eval = make_evaluator("value", ZoneMapOp::Eq, 75.0);
        let rg_stats = [
            (0.0, 20.0),   // Skip
            (20.0, 40.0),  // Skip
            (40.0, 60.0),  // Skip
            (60.0, 80.0),  // Decode (75 could be in here)
            (80.0, 100.0), // Skip
        ];
        let verdicts = eval.evaluate_batch(5, |rg, _col| Some(rg_stats[rg]));
        assert_eq!(
            verdicts
                .iter()
                .filter(|v| **v == RowGroupVerdict::Skip)
                .count(),
            4
        );
        assert_eq!(verdicts[3], RowGroupVerdict::Decode);
    }
}