wardenclyffe 0.1.1

use crate::csv_source::Row;
use std::time::Instant;
use crate::{datasource::DataSource, parser::parse_expr, eval::eval};

pub fn execute(path: &str, filter: Option<&str>, selects: Vec<&str>) {
    let start = Instant::now();

    let mut source = DataSource::open(path);
    let expr = filter.map(parse_expr);

    let mut scanned = 0usize;
    let mut matched = 0usize;

    match &mut source {
        DataSource::Csv(csv) => {
            for row in csv.rows() {
                scanned += 1;

                // apply filter
                if let Some(e) = &expr {
                    if !eval(e, &row) {
                        continue;
                    }
                }

                matched += 1;

                // projection
                if selects.len() == 1 && selects[0] == "*" {
                    println!("{:?}", row);
                } else {
                    print_projection(&row, &selects);
                }
            }
        }

        DataSource::Parquet(_pq) => {
            println!("Parquet support coming in next phase!");
        }
    }

    let elapsed = start.elapsed();
    let ms = elapsed.as_secs_f64() * 1000.0;

    eprintln!(
        "\nScanned {} rows, matched {} rows in {:.3} ms",
        scanned, matched, ms
    );
}

// projection helper
// fn print_projection(row: &crate::datasource::Row, selects: &[&str]) {
fn print_projection(row: &Row, selects: &[&str]) {
    let mut out = Vec::new();

    for s in selects {
        match *s {
            "age" => out.push(row.age.map(|x| x.to_string()).unwrap_or_default()),
            "name" => out.push(row.name.clone().unwrap_or_default()),
            "country" => out.push(row.country.clone().unwrap_or_default()),
            "salary" => out.push(row.salary.map(|x| x.to_string()).unwrap_or_default()),
            "*" => out.push(format!("{:?}", row)),
            _ => out.push(String::from("NULL")), // unknown column
        }
    }

    println!("{}", out.join(", "));
}