kore_fileformat 1.3.3

/// Phase 3: Vectorized Query Execution
/// Advanced SQL query processing with optimizations for columnar data:
/// - WHERE clause filtering with predicate pushdown
/// - SELECT column projection (lazy evaluation)
/// - GROUP BY with vectorized aggregations
/// - Index-aware chunk skipping via Bloom filters & stats

use std::collections::HashMap;

#[derive(Debug, Clone)]
pub enum Predicate {
    Equals(String, PredicateValue),
    NotEquals(String, PredicateValue),
    GreaterThan(String, f64),
    LessThan(String, f64),
    GreaterOrEqual(String, f64),
    LessOrEqual(String, f64),
    In(String, Vec<String>),
    Like(String, String),
    And(Box<Predicate>, Box<Predicate>),
    Or(Box<Predicate>, Box<Predicate>),
    Not(Box<Predicate>),
}

#[derive(Debug, Clone)]
pub enum PredicateValue {
    String(String),
    Number(f64),
    Bool(bool),
}

#[derive(Debug, Clone)]
pub enum AggregationFunc {
    Count,
    Sum(String),
    Avg(String),
    Min(String),
    Max(String),
    CountDistinct(String),
}

#[derive(Debug, Clone)]
pub struct GroupByClause {
    pub columns: Vec<String>,
    pub aggregations: Vec<(String, AggregationFunc)>,
}

#[derive(Debug, Clone)]
pub struct SelectStatement {
    pub columns: Vec<String>,
    pub from_table: String,
    pub where_clause: Option<Predicate>,
    pub group_by: Option<GroupByClause>,
    pub limit: Option<usize>,
}

pub struct QueryPlanner {
    estimated_selectivity: f64,
}

impl QueryPlanner {
    pub fn new() -> Self {
        Self {
            estimated_selectivity: 1.0,
        }
    }

    /// Parse WHERE clause into structured predicates
    pub fn parse_where_clause(where_str: &str) -> Option<Predicate> {
        let tokens = Self::tokenize(where_str);
        if tokens.is_empty() {
            return None;
        }
        Self::parse_or(&tokens, 0).map(|(pred, _)| pred)
    }

    fn tokenize(input: &str) -> Vec<String> {
        input
            .split(|c: char| c.is_whitespace())
            .filter(|s| !s.is_empty())
            .map(|s| s.to_string())
            .collect()
    }

    fn parse_or(tokens: &[String], mut pos: usize) -> Option<(Predicate, usize)> {
        let (mut left, new_pos) = Self::parse_and(tokens, pos)?;
        pos = new_pos;

        while pos < tokens.len() && tokens[pos].to_uppercase() == "OR" {
            pos += 1;
            let (right, new_pos) = Self::parse_and(tokens, pos)?;
            pos = new_pos;
            left = Predicate::Or(Box::new(left), Box::new(right));
        }

        Some((left, pos))
    }

    fn parse_and(tokens: &[String], mut pos: usize) -> Option<(Predicate, usize)> {
        let (mut left, new_pos) = Self::parse_comparison(tokens, pos)?;
        pos = new_pos;

        while pos < tokens.len() && tokens[pos].to_uppercase() == "AND" {
            pos += 1;
            let (right, new_pos) = Self::parse_comparison(tokens, pos)?;
            pos = new_pos;
            left = Predicate::And(Box::new(left), Box::new(right));
        }

        Some((left, pos))
    }

    fn parse_comparison(tokens: &[String], pos: usize) -> Option<(Predicate, usize)> {
        if pos + 2 >= tokens.len() {
            return None;
        }

        let col = tokens[pos].clone();
        let op = tokens[pos + 1].to_uppercase();
        let val_str = tokens[pos + 2].clone();

        let pred = match op.as_str() {
            "=" => Predicate::Equals(col, PredicateValue::String(val_str.trim_matches('\'').to_string())),
            "!=" | "<>" => Predicate::NotEquals(col, PredicateValue::String(val_str.trim_matches('\'').to_string())),
            ">" => Predicate::GreaterThan(col, val_str.parse::<f64>().ok()?),
            "<" => Predicate::LessThan(col, val_str.parse::<f64>().ok()?),
            ">=" => Predicate::GreaterOrEqual(col, val_str.parse::<f64>().ok()?),
            "<=" => Predicate::LessOrEqual(col, val_str.parse::<f64>().ok()?),
            "LIKE" => Predicate::Like(col, val_str.trim_matches('\'').to_string()),
            _ => return None,
        };

        Some((pred, pos + 3))
    }

    /// Estimate how many rows pass the filter
    pub fn estimate_selectivity(&mut self, predicate: &Predicate) -> f64 {
        let sel = match predicate {
            Predicate::Equals(_, _) => 0.01,
            Predicate::GreaterThan(_, _) => 0.5,
            Predicate::LessThan(_, _) => 0.5,
            Predicate::In(_, vals) => vals.len() as f64 * 0.01,
            Predicate::Like(_, _) => 0.1,
            Predicate::And(left, right) => {
                self.estimate_selectivity(left) * self.estimate_selectivity(right)
            }
            Predicate::Or(left, right) => {
                let l = self.estimate_selectivity(left);
                let r = self.estimate_selectivity(right);
                l + r - (l * r)
            }
            Predicate::Not(inner) => 1.0 - self.estimate_selectivity(inner),
            _ => 0.1,
        };
        self.estimated_selectivity = sel;
        sel
    }
}

pub struct RowFilter;

impl RowFilter {
    /// Evaluate predicate against a row
    pub fn matches(predicate: &Predicate, row: &HashMap<String, String>) -> bool {
        match predicate {
            Predicate::Equals(col, val) => {
                row.get(col).map_or(false, |v| {
                    match val {
                        PredicateValue::String(s) => v == s,
                        _ => false,
                    }
                })
            }
            Predicate::GreaterThan(col, threshold) => {
                row.get(col)
                    .and_then(|v| v.parse::<f64>().ok())
                    .map_or(false, |v| v > *threshold)
            }
            Predicate::LessThan(col, threshold) => {
                row.get(col)
                    .and_then(|v| v.parse::<f64>().ok())
                    .map_or(false, |v| v < *threshold)
            }
            Predicate::GreaterOrEqual(col, threshold) => {
                row.get(col)
                    .and_then(|v| v.parse::<f64>().ok())
                    .map_or(false, |v| v >= *threshold)
            }
            Predicate::LessOrEqual(col, threshold) => {
                row.get(col)
                    .and_then(|v| v.parse::<f64>().ok())
                    .map_or(false, |v| v <= *threshold)
            }
            Predicate::Like(col, pattern) => {
                row.get(col).map_or(false, |v| Self::like_match(v, pattern))
            }
            Predicate::And(left, right) => {
                Self::matches(left, row) && Self::matches(right, row)
            }
            Predicate::Or(left, right) => {
                Self::matches(left, row) || Self::matches(right, row)
            }
            Predicate::Not(inner) => !Self::matches(inner, row),
            _ => true,
        }
    }

    fn like_match(text: &str, pattern: &str) -> bool {
        let parts: Vec<&str> = pattern.split('%').collect();
        match parts.len() {
            1 => text == parts[0],
            2 => {
                if parts[0].is_empty() {
                    text.ends_with(parts[1])
                } else if parts[1].is_empty() {
                    text.starts_with(parts[0])
                } else {
                    text.starts_with(parts[0]) && text.ends_with(parts[1])
                }
            }
            _ => {
                let mut pos = 0;
                for (i, part) in parts.iter().enumerate() {
                    if i == 0 && !part.is_empty() {
                        if !text.starts_with(part) {
                            return false;
                        }
                        pos = part.len();
                    } else if i == parts.len() - 1 && !part.is_empty() {
                        if !text[pos..].ends_with(part) {
                            return false;
                        }
                    } else if !part.is_empty() {
                        match text[pos..].find(part) {
                            Some(offset) => pos += offset + part.len(),
                            None => return false,
                        }
                    }
                }
                true
            }
        }
    }
}

pub struct GroupByExecutor;

impl GroupByExecutor {
    /// Execute vectorized GROUP BY aggregation
    pub fn execute(
        rows: Vec<HashMap<String, String>>,
        group_cols: &[String],
        aggs: &[(String, AggregationFunc)],
    ) -> Vec<HashMap<String, String>> {
        // Partition rows into groups
        let mut groups: HashMap<String, Vec<HashMap<String, String>>> = HashMap::new();

        for row in rows {
            let key = group_cols
                .iter()
                .map(|col| row.get(col).cloned().unwrap_or_default())
                .collect::<Vec<_>>()
                .join("|");
            groups.entry(key).or_insert_with(Vec::new).push(row);
        }

        // Aggregate each group
        let mut results = Vec::new();
        for (key, group) in groups {
            let mut result = HashMap::new();

            // Set group keys
            let key_parts: Vec<&str> = key.split('|').collect();
            for (i, col) in group_cols.iter().enumerate() {
                if i < key_parts.len() {
                    result.insert(col.clone(), key_parts[i].to_string());
                }
            }

            // Calculate aggregations
            for (alias, agg_func) in aggs {
                let agg_val = match agg_func {
                    AggregationFunc::Count => group.len().to_string(),
                    AggregationFunc::Sum(col) => {
                        let sum: f64 = group
                            .iter()
                            .filter_map(|row| row.get(col).and_then(|v| v.parse::<f64>().ok()))
                            .sum();
                        sum.to_string()
                    }
                    AggregationFunc::Avg(col) => {
                        let values: Vec<f64> = group
                            .iter()
                            .filter_map(|row| row.get(col).and_then(|v| v.parse::<f64>().ok()))
                            .collect();
                        if !values.is_empty() {
                            (values.iter().sum::<f64>() / values.len() as f64).to_string()
                        } else {
                            "0".to_string()
                        }
                    }
                    AggregationFunc::Min(col) => {
                        group
                            .iter()
                            .filter_map(|row| row.get(col).cloned())
                            .min()
                            .unwrap_or_default()
                    }
                    AggregationFunc::Max(col) => {
                        group
                            .iter()
                            .filter_map(|row| row.get(col).cloned())
                            .max()
                            .unwrap_or_default()
                    }
                    AggregationFunc::CountDistinct(col) => {
                        let distinct: std::collections::HashSet<_> = group
                            .iter()
                            .filter_map(|row| row.get(col).cloned())
                            .collect();
                        distinct.len().to_string()
                    }
                };
                result.insert(alias.clone(), agg_val);
            }

            results.push(result);
        }

        results
    }
}

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

    #[test]
    fn test_parse_where_clause() {
        let pred = QueryPlanner::parse_where_clause("age > 30 AND city = NYC");
        assert!(pred.is_some());
    }

    #[test]
    fn test_row_filter_equals() {
        let mut row = HashMap::new();
        row.insert("name".to_string(), "Alice".to_string());

        let pred = Predicate::Equals("name".to_string(), PredicateValue::String("Alice".to_string()));
        assert!(RowFilter::matches(&pred, &row));
    }

    #[test]
    fn test_like_matching() {
        assert!(RowFilter::like_match("hello", "hel%"));
        assert!(RowFilter::like_match("hello", "%llo"));
        assert!(!RowFilter::like_match("hello", "world%"));
    }

    #[test]
    fn test_group_by_execution() {
        let mut row1 = HashMap::new();
        row1.insert("category".to_string(), "A".to_string());
        row1.insert("amount".to_string(), "100".to_string());

        let mut row2 = HashMap::new();
        row2.insert("category".to_string(), "A".to_string());
        row2.insert("amount".to_string(), "50".to_string());

        let rows = vec![row1, row2];
        let groups = vec!["category".to_string()];
        let aggs = vec![("total".to_string(), AggregationFunc::Sum("amount".to_string()))];

        let results = GroupByExecutor::execute(rows, &groups, &aggs);
        assert_eq!(results.len(), 1);
    }
}