use std::{borrow::ToOwned, cmp, process};

use newline_converter::dos2unix;

use crate::workdir::Workdir;

macro_rules! stats_tests {
    ($name:ident, $field:expr_2021, $rows:expr_2021, $expect:expr_2021) => {
        stats_tests!($name, $field, $rows, $expect, false, true);
    };
    ($name:ident, $field:expr_2021, $rows:expr_2021, $expect:expr_2021, $nulls:expr_2021, $infer_dates:expr_2021) => {
        mod $name {
            use super::test_stats;

            stats_test_headers!($name, $field, $rows, $expect, $nulls, $infer_dates);
            stats_test_no_headers!($name, $field, $rows, $expect, $nulls, $infer_dates);
        }
    };
}

macro_rules! stats_no_infer_dates_tests {
    ($name:ident, $field:expr_2021, $rows:expr_2021, $expect:expr_2021) => {
        stats_tests!($name, $field, $rows, $expect, false, false);
    };
    ($name:ident, $field:expr_2021, $rows:expr_2021, $expect:expr_2021, $nulls:expr_2021, $infer_dates:expr_2021) => {
        mod $name {
            use super::test_stats;

            stats_test_headers!($name, $field, $rows, $expect, $nulls, $infer_dates);
            stats_test_no_headers!($name, $field, $rows, $expect, $nulls, $infer_dates);
        }
    };
}

macro_rules! stats_test_headers {
    ($name:ident, $field:expr_2021, $rows:expr_2021, $expect:expr_2021) => {
        stats_test_headers!($name, $field, $rows, $expect, false, true);
    };
    ($name:ident, $field:expr_2021, $rows:expr_2021, $expect:expr_2021, $nulls:expr_2021, $infer_dates:expr_2021) => {
        #[test]
        fn headers_no_index() {
            let name = concat!(stringify!($name), "_headers_no_index");
            test_stats(
                name,
                $field,
                $rows,
                $expect,
                true,
                false,
                $nulls,
                $infer_dates,
            );
        }

        #[test]
        fn headers_index() {
            let name = concat!(stringify!($name), "_headers_index");
            test_stats(
                name,
                $field,
                $rows,
                $expect,
                true,
                true,
                $nulls,
                $infer_dates,
            );
        }
    };
}

macro_rules! stats_test_no_headers {
    ($name:ident, $field:expr_2021, $rows:expr_2021, $expect:expr_2021) => {
        stats_test_no_headers!($name, $field, $rows, $expect, false, true);
    };
    ($name:ident, $field:expr_2021, $rows:expr_2021, $expect:expr_2021, $nulls:expr_2021, $infer_dates:expr_2021) => {
        #[test]
        fn no_headers_no_index() {
            let name = concat!(stringify!($name), "_no_headers_no_index");
            test_stats(
                name,
                $field,
                $rows,
                $expect,
                false,
                false,
                $nulls,
                $infer_dates,
            );
        }

        #[test]
        fn no_headers_index() {
            let name = concat!(stringify!($name), "_no_headers_index");
            test_stats(
                name,
                $field,
                $rows,
                $expect,
                false,
                true,
                $nulls,
                $infer_dates,
            );
        }
    };
}

#[allow(clippy::too_many_arguments)]
fn test_stats<S>(
    name: S,
    field: &str,
    rows: &[&str],
    expected: &str,
    headers: bool,
    use_index: bool,
    nulls: bool,
    infer_dates: bool,
) where
    S: ::std::ops::Deref<Target = str>,
{
    let (wrk, mut cmd) = setup(name, rows, headers, use_index, nulls, infer_dates);
    let field_val = get_field_value(&wrk, &mut cmd, field).unwrap_or_default();
    // Only compare the first few bytes since floating point arithmetic
    // can mess with exact comparisons.
    // when field = skewness, we're comparing a long sequence of the quartile columns,
    // that's why we use 40, if not, its a single column, and we need to compare only
    // the first 15 characters just in case its a float
    let len = cmp::min(
        if field == "skewness" { 40 } else { 15 },
        cmp::min(field_val.len(), expected.len()),
    );
    assert_eq!(&field_val[0..len], &expected[0..len]);
}

fn setup<S>(
    name: S,
    rows: &[&str],
    headers: bool,
    use_index: bool,
    nulls: bool,
    infer_dates: bool,
) -> (Workdir, process::Command)
where
    S: ::std::ops::Deref<Target = str>,
{
    let wrk = Workdir::new(&name);
    let mut data: Vec<Vec<String>> = rows.iter().map(|&s| vec![s.to_owned()]).collect();
    if headers {
        data.insert(0, svec!["header"]);
    }
    if use_index {
        wrk.create_indexed("in.csv", data);
    } else {
        wrk.create("in.csv", data);
    }

    let mut cmd = wrk.command("stats");
    cmd.arg("in.csv");
    if !headers {
        cmd.arg("--no-headers");
    }
    if nulls {
        cmd.arg("--nulls");
    }
    if infer_dates {
        cmd.arg("--infer-dates").arg("--dates-whitelist").arg("all");
    }

    (wrk, cmd)
}

fn get_field_value(
    wrk: &Workdir,
    cmd: &mut process::Command,
    field: &str,
) -> Result<String, String> {
    if field == "median" {
        cmd.arg("--median");
    }
    if field == "quartiles" {
        cmd.arg("--quartiles");
    }
    if field == "cardinality" {
        cmd.arg("--cardinality");
    }
    if field == "zero_padded_numeric" {
        cmd.arg("--zero-padded-numeric");
    }
    if field == "mode" || field == "antimode" {
        cmd.arg("--mode");
    }
    if field == "infer_dates" {
        cmd.arg("--infer-dates");
    }

    let mut rows: Vec<Vec<String>> = wrk.read_stdout(cmd);
    if rows.is_empty() {
        return Err(format!("Empty stats for command '{cmd:?}'."));
    }
    let headers = rows.remove(0);
    let mut sequence: Vec<&str> = vec![];
    for row in &rows {
        for (h, val) in headers.iter().zip(row.iter()) {
            match field {
                "quartiles" => match &**h {
                    "lower_outer_fence" | "lower_inner_fence" | "q1" | "q2_median" | "q3"
                    | "iqr" | "upper_inner_fence" | "upper_outer_fence" => {
                        sequence.push(val);
                    },
                    "skewness" => {
                        sequence.push(val);
                        return Ok(sequence.join(","));
                    },
                    _ => {},
                },
                _ => {
                    if &**h == field {
                        return Ok(val.clone());
                    }
                },
            }
        }
    }
    Err(format!(
        "Could not find field '{field}' in headers '{headers:?}' for command '{cmd:?}'."
    ))
}

stats_tests!(stats_infer_string, "type", &["a"], "String");
stats_tests!(stats_infer_int, "type", &["1"], "Integer");
stats_tests!(stats_infer_float, "type", &["1.2"], "Float");
stats_tests!(stats_infer_null, "type", &[""], "NULL");
stats_tests!(stats_infer_date, "type", &["1968-06-27"], "Date");
stats_tests!(stats_infer_date_unix_epoch, "type", &["1970-01-01"], "Date");
stats_tests!(
    stats_infer_date_unix_epoch_2,
    "type",
    &["1970-01-01 00:00:00"],
    "Date"
);
stats_tests!(
    stats_infer_date_unix_epoch_3,
    "type",
    &["1970-01-01 00:00:00 UTC"],
    "Date"
);
stats_tests!(
    stats_infer_date_unix_epoch_4,
    "type",
    &["1970-01-01 00:00:00.000 UTC"],
    "Date"
);
stats_tests!(
    stats_infer_date_unix_epoch_5,
    "type",
    &["1970-01-01 00:00:00.000000 UTC"],
    "Date"
);
stats_tests!(
    stats_infer_date_unix_epoch_6,
    "type",
    &["1970-01-01 00:00:00.000000000 UTC"],
    "Date"
);
stats_tests!(
    stats_infer_date_unix_epoch_7,
    "type",
    &["1970-01-01 00:00:01 UTC"],
    "DateTime"
);
stats_no_infer_dates_tests!(stats_infer_nodate, "type", &["1968-06-27"], "String");
stats_tests!(
    stats_infer_datetime,
    "type",
    &["1968-06-27 12:30:01"],
    "DateTime"
);
stats_no_infer_dates_tests!(
    stats_infer_nodatetime,
    "type",
    &["1968-06-27 12:30:01"],
    "String"
);
stats_tests!(stats_infer_string_null, "type", &["a", ""], "String");
stats_tests!(stats_infer_int_null, "type", &["1", ""], "Integer");
stats_tests!(stats_infer_float_null, "type", &["1.2", ""], "Float");
stats_tests!(
    stats_infer_date_null,
    "type",
    &["June 27, 1968", ""],
    "Date"
);
stats_no_infer_dates_tests!(
    stats_infer_no_infer_dates_null,
    "type",
    &["June 27, 1968", ""],
    "String"
);
stats_tests!(
    stats_infer_datetime_null,
    "type",
    &["June 27, 1968 12:30:00 UTC", ""],
    "DateTime"
);
stats_no_infer_dates_tests!(
    stats_infer_nodatetime_null,
    "type",
    &["June 27, 1968 12:30:00 UTC", ""],
    "String"
);
stats_tests!(stats_infer_null_string, "type", &["", "a"], "String");
stats_tests!(stats_infer_null_int, "type", &["", "1"], "Integer");
stats_tests!(stats_infer_null_float, "type", &["", "1.2"], "Float");
stats_tests!(
    stats_infer_null_date,
    "type",
    &["", "September 17, 2012 at 10:09am PST"],
    "Date"
);
stats_no_infer_dates_tests!(
    stats_infer_null_nodate,
    "type",
    &["", "September 17, 2012 at 10:09am PST"],
    "String"
);
stats_tests!(
    stats_infer_date_datetime,
    "type",
    &["September 11, 2001", "September 17, 2012 at 10:09am PST"],
    "DateTime"
);
stats_no_infer_dates_tests!(
    stats_infer_nodate_nodatetime,
    "type",
    &["September 11, 2001", "September 17, 2012 at 10:09am PST"],
    "String"
);
stats_tests!(stats_infer_int_string, "type", &["1", "a"], "String");
stats_tests!(stats_infer_string_int, "type", &["a", "1"], "String");
stats_tests!(stats_infer_int_float, "type", &["1", "1.2"], "Float");
stats_tests!(stats_infer_float_int, "type", &["1.2", "1"], "Float");
stats_tests!(
    stats_infer_null_int_float_string,
    "type",
    &["", "1", "1.2", "a"],
    "String"
);
stats_tests!(
    stats_infer_date_string,
    "type",
    &["1968-06-27", "abcde"],
    "String"
);
stats_tests!(
    stats_infer_string_date,
    "type",
    &["wxyz", "1968-06-27"],
    "String"
);

stats_tests!(stats_no_mean, "mean", &["a"], "");
stats_tests!(stats_no_stddev, "stddev", &["a"], "");
stats_tests!(stats_no_variance, "variance", &["a"], "");
stats_tests!(stats_no_median, "median", &["a"], "");
stats_tests!(stats_no_quartiles, "quartiles", &["a"], ",,,,,");
stats_tests!(stats_no_mode, "mode", &["a", "b"], "N/A");
stats_tests!(
    stats_multiple_modes,
    "mode",
    &["a", "a", "b", "b", "c", "d", "e", "e"],
    "a|b|e|3|1"
);
stats_tests!(
    stats_multiple_modes_num,
    "mode",
    &["5", "5", "33", "33", "42", "17", "99", "99"],
    "33|5|99|3|1"
);
stats_tests!(
    stats_multiple_antimodes,
    "antimode",
    &["a", "a", "b", "b", "c", "d", "e", "e"],
    "c|d|2|1"
);
stats_tests!(
    stats_multiple_antimodes_num,
    "antimode",
    &["5", "5", "33", "33", "42", "17", "98", "99", "99"],
    "17|42|98|3|1"
);
stats_tests!(
    stats_range,
    "range",
    &["a", "a", "b", "b", "c", "d", "e", "e"],
    ""
);
stats_tests!(
    stats_range_num,
    "range",
    &["5", "5", "33", "33", "42", "17", "98", "99", "99"],
    "94"
);
stats_tests!(
    stats_sparsity,
    "sparsity",
    &["5", "5", "33", "33", "42", "17", "98", "99", "99", ""],
    "0.1"
);

stats_tests!(stats_null_mean, "mean", &[""], "");
stats_tests!(stats_null_stddev, "stddev", &[""], "");
stats_tests!(stats_null_variance, "variance", &[""], "");
stats_tests!(stats_null_median, "median", &[""], "");
stats_tests!(stats_null_quartiles, "quartiles", &[""], ",,,,,");
stats_tests!(stats_null_mode, "mode", &[""], "N/A");
stats_tests!(stats_null_antimode, "antimode", &[""], "*ALL");
stats_tests!(stats_null_range, "range", &[""], "N/A");
stats_tests!(stats_null_sparsity, "sparsity", &[""], "1.0");

stats_tests!(stats_includenulls_null_mean, "mean", &[""], "", true, false);
stats_tests!(
    stats_includenulls_null_stddev,
    "stddev",
    &[""],
    "",
    true,
    false
);
stats_tests!(
    stats_includenulls_null_variance,
    "variance",
    &[""],
    "",
    true,
    false
);
stats_tests!(
    stats_includenulls_null_median,
    "median",
    &[""],
    "",
    true,
    false
);
stats_tests!(
    stats_includenulls_null_quartiles,
    "quartiles",
    &[""],
    ",,,,,",
    true,
    false
);
stats_tests!(
    stats_includenulls_null_mode,
    "mode",
    &[""],
    "N/A",
    true,
    false
);
stats_tests!(
    stats_includenulls_null_antimode,
    "antimode",
    &[""],
    "*ALL",
    true,
    false
);
stats_tests!(
    stats_includenulls_null_range,
    "range",
    &[""],
    "N/A",
    true,
    false
);
stats_tests!(
    stats_includenulls_null_sparsity,
    "sparsity",
    &[""],
    "1.0",
    true,
    false
);

stats_tests!(
    stats_includenulls_mean,
    "mean",
    &["5", "", "15", "10"],
    "7.5",
    true,
    false
);

stats_tests!(stats_sum_integers, "sum", &["1", "2"], "3");
stats_tests!(stats_sum_floats, "sum", &["1.5", "2.8"], "4.3");
stats_tests!(stats_sum_mixed1, "sum", &["1.5", "2"], "3.5");
stats_tests!(stats_sum_mixed2, "sum", &["2", "1.5"], "3.5");
stats_tests!(stats_sum_mixed3, "sum", &["1.5", "hi", "2.8"], "4.3");
stats_tests!(stats_sum_nulls1, "sum", &["1", "", "2"], "3");
stats_tests!(stats_sum_nulls2, "sum", &["", "1", "2"], "3");
stats_tests!(
    stats_sum_overflow,
    "sum",
    &[
        "9223372036854775807", // i64::MAX
        "1",
        "2"
    ],
    "*OVERFLOW*"
);
stats_tests!(
    stats_sum_negative_overflow,
    "sum",
    &[
        "-9223372036854775808", // i64::MIN
        "-1",
        "-2"
    ],
    "*UNDERFLOW*"
);

stats_tests!(stats_min, "min", &["2", "1.1"], "1.1");
stats_tests!(stats_max, "max", &["2", "1.1"], "2");
// stats_tests!(stats_min_mix, "min", &["1", "2", "a", "1.1", "b"], "1.1");
stats_tests!(stats_max_mix, "max", &["2", "a", "1.1"], "a");
stats_tests!(stats_min_null, "min", &["", "2", "1.1"], "1.1");
stats_tests!(stats_max_null, "max", &["2", "1.1", ""], "2");

stats_tests!(stats_len_min, "min_length", &["aa", "a"], "1");
stats_tests!(stats_len_max, "max_length", &["a", "aa"], "2");
stats_tests!(stats_len_min_null, "min_length", &["", "aa", "a"], "0");
stats_tests!(stats_len_max_null, "max_length", &["a", "aa", ""], "2");

stats_tests!(stats_mean, "mean", &["5", "15", "10"], "10");
stats_tests!(stats_stddev, "stddev", &["1", "2", "3"], "0.8165");
stats_tests!(stats_variance, "variance", &["3", "5", "7", "9", "11"], "8");
stats_tests!(stats_mean_null, "mean", &["", "5", "15", "10"], "10");
stats_tests!(stats_stddev_null, "stddev", &["1", "2", "3", ""], "0.8165");
stats_tests!(
    stats_variance_null,
    "variance",
    &["3", "5", "7", "9", "", "10"],
    "6"
);
stats_tests!(stats_mean_mix, "mean", &["5", "15.1", "9.9"], "10");
stats_tests!(stats_stddev_mix, "stddev", &["1", "2.1", "2.9"], "0.7789");
stats_tests!(
    stats_variance_mix,
    "variance",
    &["1.5", "2", "2.5", "3"],
    "0.3125"
);

stats_tests!(stats_cardinality, "cardinality", &["a", "b", "a"], "2");
stats_no_infer_dates_tests!(
    stats_zero_padded_numeric_zip,
    "zero_padded_numeric",
    &["10001", "07306", "90210"],
    "true"
);
stats_no_infer_dates_tests!(
    stats_zero_padded_numeric_all_padded,
    "zero_padded_numeric",
    &["01", "02", "03"],
    "true"
);
// Real-world zero-padded *decimal code* systems cited by @kulnor in PR #3938.
// Every value has a leading-zero integer part, so qsv now infers these as String (to preserve the
// padding, like zip codes) and the column is flagged. See stats_zero_padded_numeric_infers_string.
stats_no_infer_dates_tests!(
    stats_zero_padded_numeric_icd9,
    "zero_padded_numeric",
    &["007.1", "008.45", "038.11"],
    "true"
);
stats_no_infer_dates_tests!(
    stats_zero_padded_numeric_dewey,
    "zero_padded_numeric",
    &["004.67", "005.13", "025.04"],
    "true"
);
stats_no_infer_dates_tests!(
    stats_zero_padded_numeric_nibrs,
    "zero_padded_numeric",
    &["01.10", "09.10", "01.01"],
    "true"
);
stats_no_infer_dates_tests!(
    stats_zero_padded_numeric_cepa,
    "zero_padded_numeric",
    &["01.12", "05.10", "06.01"],
    "true"
);
stats_no_infer_dates_tests!(
    stats_zero_padded_numeric_transport_linear_ref,
    "zero_padded_numeric",
    &["03.40", "02.30", "02.03"],
    "true"
);
stats_no_infer_dates_tests!(
    stats_zero_padded_numeric_hs_tariff,
    "zero_padded_numeric",
    &["01.01", "07.10", "06.10"],
    "true"
);
stats_tests!(stats_mode, "mode", &["a", "b", "a"], "a,1,2");
stats_tests!(stats_mode_null, "mode", &["", "a", "b", "a"], "a,1,2");
stats_tests!(stats_antimode, "antimode", &["a", "b", "a"], "b,1,1");
stats_tests!(
    stats_antimode_null,
    "antimode",
    &["", "a", "b", "a"],
    "NULL|b|2|1"
);
stats_tests!(stats_median, "median", &["1", "2", "3"], "2");
stats_tests!(stats_median_null, "median", &["", "1", "2", "3"], "2");
stats_tests!(stats_median_even, "median", &["1", "2", "3", "4"], "2.5");
stats_tests!(
    stats_median_even_null,
    "median",
    &["", "1", "2", "3", "4"],
    "2.5"
);
stats_tests!(stats_median_mix, "median", &["1", "2.5", "3"], "2.5");
stats_tests!(
    stats_quartiles,
    "quartiles",
    &["1", "2", "3"],
    "-5,-2,1,2,3,2,6,9"
);
stats_tests!(
    stats_quartiles_null,
    "quartiles",
    &["", "1", "2", "3"],
    "-5,-2,1,2,3,2,6,9"
);
stats_tests!(
    stats_quartiles_even,
    "quartiles",
    &["1", "2", "3", "4"],
    "-4.5,-1.5,1.5,2.5,3.5,2,6.5,9.5"
);
stats_tests!(
    stats_quartiles_even_null,
    "quartiles",
    &["", "1", "2", "3", "4"],
    "-4.5,-1.5,1.5,2.5,3.5,2,6.5,9.5"
);
stats_tests!(
    stats_quartiles_mix,
    "quartiles",
    &["1", "2.0", "3", "4"],
    "-4.5,-1.5,1.5,2.5,3.5,2,6.5,9.5"
);
stats_tests!(stats_quartiles_null_empty, "quartiles", &[""], "");

stats_tests!(stats_nullcount, "nullcount", &["", "1", "2"], "1");
stats_tests!(stats_nullcount_none, "nullcount", &["a", "1", "2"], "0");
stats_tests!(
    stats_nullcount_spacenotnull,
    "nullcount",
    &[" ", "1", "2"],
    "0"
);
stats_tests!(stats_nullcount_all, "nullcount", &["", "", ""], "3");
stats_no_infer_dates_tests!(
    stats_noinfer_null_nodate2,
    "type",
    &["", "September 17, 2012 at 10:09am PST"],
    "String",
    true,
    false
);
stats_no_infer_dates_tests!(
    stats_infer_null_nodate2,
    "type",
    &["", "September 17, 2012 at 10:09am PST"],
    "DateTime",
    true,
    true
);
stats_tests!(
    stats_infer_date_datetime2,
    "type",
    &["September 11, 2001", "September 17, 2012 at 10:09am PST"],
    "DateTime",
    false,
    true
);
stats_no_infer_dates_tests!(
    stats_infer_date_datetime3,
    "type",
    &["9-11", "September 17, 2012 at 10:09am PST"],
    "String",
    false,
    true
);

#[test]
fn stats_prefer_dmy() {
    let wrk = Workdir::new("stats_prefer_dmy");
    let test_file = wrk.load_test_file("boston311-dmy-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-dates")
        .arg("--prefer-dmy")
        .args(["--dates-whitelist", "_dT"])
        .arg(test_file);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    wrk.create("in2.csv", got);

    // removed stddev_length, variance_length, cv_length, variance, geometric_mean, harmonic_mean,
    // stddev, sem & cv columns as its causing flaky CI test for float values
    let mut cmd = wrk.command("select");
    cmd.arg("!/stddev_length|variance_length|cv_length|variance|geometric_mean|harmonic_mean|stddev|sem|cv/")
        .arg("in2.csv");

    let got2: String = wrk.stdout(&mut cmd);
    let expected2 = wrk.load_test_resource("boston311-100-stats.csv");

    assert_eq!(dos2unix(&got2), dos2unix(&expected2).trim_end());
}

#[test]
fn stats_prefer_mdy() {
    let wrk = Workdir::new("stats_prefer_mdy");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-dates")
        .args(["--dates-whitelist", "_dt"])
        .arg(test_file);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    wrk.create("in2.csv", got);

    // removed stddev_length, variance_length, cv_length, variance, geometric_mean, harmonic_mean,
    // stddev, sem & cv columns as its causing flaky CI test for float values
    let mut cmd = wrk.command("select");
    cmd.arg("!/stddev_length|variance_length|cv_length|variance|geometric_mean|harmonic_mean|stddev|sem|cv/")
        .arg("in2.csv");

    let got2: String = wrk.stdout(&mut cmd);

    let expected2 = wrk.load_test_resource("boston311-100-stats.csv");

    assert_eq!(dos2unix(&got2), dos2unix(&expected2).trim_end());
}

#[test]
fn stats_zero_padded_numeric_columns() {
    let wrk = Workdir::new("stats_zero_padded_numeric_columns");
    wrk.create(
        "in.csv",
        vec![
            svec![
                "zip",
                "plain_int",
                "mixed",
                "price",
                "icd9",
                "frac",
                "trailing",
                "mixed_float"
            ],
            svec!["10001", "1", "07306", "1.5", "007.1", "0.5", "7.10", "7.1"],
            svec![
                "07306", "2", "Main St", "2.5", "008.45", "0.25", "8.20", "007.1"
            ],
            svec![
                "90210", "3", "12345", "3.5", "038.11", "0.75", "9.30", "3.5"
            ],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--zero-padded-numeric").arg("in.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout_on_success(&mut cmd);

    let header = &got[0];
    let field_idx = header.iter().position(|h| h == "field").unwrap();
    let zpn_idx = header
        .iter()
        .position(|h| h == "zero_padded_numeric")
        .expect("zero_padded_numeric column should be present");

    let mut zpn: std::collections::HashMap<String, String> = std::collections::HashMap::new();
    for row in &got[1..] {
        zpn.insert(row[field_idx].clone(), row[zpn_idx].clone());
    }

    // zip: String with leading zeros, every value all-digit -> flagged
    assert_eq!(zpn["zip"], "true");
    // plain_int: inferred Integer (no leading zero) -> not flagged
    assert_eq!(zpn["plain_int"], "");
    // mixed: String but "Main St" is not all-digit -> not flagged
    assert_eq!(zpn["mixed"], "");
    // price: Float, all plain (no leading-zero padding) -> not flagged
    assert_eq!(zpn["price"], "");
    // icd9: every value a zero-padded decimal code -> inferred String to preserve padding ->
    // flagged
    assert_eq!(zpn["icd9"], "true");
    // frac: Float, ordinary fractions (single 0 before the dot) -> not flagged
    assert_eq!(zpn["frac"], "");
    // trailing: Float, pure trailing-zero codes (7.10) are out of scope -> not flagged
    assert_eq!(zpn["trailing"], "");
    // mixed_float: a zero-padded value (007.1) widens the column to String -> flagged
    assert_eq!(zpn["mixed_float"], "true");
}

#[test]
fn stats_zero_padded_numeric_gating() {
    let wrk = Workdir::new("stats_zero_padded_numeric_gating");
    wrk.create("in.csv", vec![svec!["zip"], svec!["07306"], svec!["10001"]]);

    // absent from default stats output
    let mut cmd = wrk.command("stats");
    cmd.arg("in.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout_on_success(&mut cmd);
    assert!(!got[0].iter().any(|h| h == "zero_padded_numeric"));

    // present (and true) with --everything
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything").arg("in.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout_on_success(&mut cmd);
    let zpn_idx = got[0]
        .iter()
        .position(|h| h == "zero_padded_numeric")
        .expect("zero_padded_numeric present with --everything");
    assert_eq!(got[1][zpn_idx], "true");
}

// Documents the known limitations of the leading-zero heuristic for the real-world codes
// @kulnor cited in PR #3938. These are intentionally NOT flagged.
#[test]
fn stats_zero_padded_numeric_real_world_limitations() {
    let wrk = Workdir::new("stats_zero_padded_numeric_real_world_limitations");
    wrk.create(
        "in.csv",
        vec![
            svec![
                "hs_multidot",
                "alphanumeric",
                "trailing_only",
                "ordinary_frac"
            ],
            // 0601.10.00: multi-dot HS code -> not a Float, stays String, but not all-digit
            // (has dots) -> fails the integer/String all-digit path too.
            svec!["0601.10.00", "09.0A", "7.10", "0.5"],
            svec!["0102.21.00", "01.1B", "8.20", "0.25"],
            svec!["0710.00.00", "02.0C", "9.30", "0.75"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--zero-padded-numeric").arg("in.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout_on_success(&mut cmd);

    let header = &got[0];
    let field_idx = header.iter().position(|h| h == "field").unwrap();
    let type_idx = header.iter().position(|h| h == "type").unwrap();
    let zpn_idx = header
        .iter()
        .position(|h| h == "zero_padded_numeric")
        .expect("zero_padded_numeric column should be present");

    // map field -> (inferred type, zero_padded_numeric)
    let mut got_map: std::collections::HashMap<String, (String, String)> =
        std::collections::HashMap::new();
    for row in &got[1..] {
        got_map.insert(
            row[field_idx].clone(),
            (row[type_idx].clone(), row[zpn_idx].clone()),
        );
    }

    // multi-dot HS codes don't parse as Float -> inferred String, but not all-digit (dots) so the
    // integer/String path doesn't flag them either.
    assert_eq!(
        got_map["hs_multidot"],
        ("String".to_string(), String::new())
    );
    // alphanumeric codes (09.0A) aren't numeric at all -> inferred String, not flagged
    assert_eq!(
        got_map["alphanumeric"],
        ("String".to_string(), String::new())
    );
    // pure trailing-zero codes parse as Float but have no leading-zero padding -> not flagged
    assert_eq!(
        got_map["trailing_only"],
        ("Float".to_string(), String::new())
    );
    // ordinary fractions (single 0 before the dot) parse as Float -> not flagged
    assert_eq!(
        got_map["ordinary_frac"],
        ("Float".to_string(), String::new())
    );
}

// Zero-padded floats (ICD-9 / Dewey / HS decimal codes) are inferred as String to preserve their
// leading zeros, mirroring how zero-padded integers (zip codes) are already kept as text. This is
// the data-integrity default and is NOT gated behind --zero-padded-numeric.
#[test]
fn stats_zero_padded_numeric_infers_string() {
    let wrk = Workdir::new("stats_zero_padded_numeric_infers_string");
    wrk.create(
        "in.csv",
        vec![
            svec!["icd9", "plain_float"],
            svec!["007.1", "1.5"],
            svec!["008.45", "2.5"],
            svec!["038.11", "3.5"],
        ],
    );

    // type inference flips to String even without --zero-padded-numeric
    let mut cmd = wrk.command("stats");
    cmd.arg("--typesonly").arg("in.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout_on_success(&mut cmd);

    let header = &got[0];
    let field_idx = header.iter().position(|h| h == "field").unwrap();
    let type_idx = header.iter().position(|h| h == "type").unwrap();
    let mut types: std::collections::HashMap<String, String> = std::collections::HashMap::new();
    for row in &got[1..] {
        types.insert(row[field_idx].clone(), row[type_idx].clone());
    }

    // zero-padded decimal codes -> String (padding preserved)
    assert_eq!(types["icd9"], "String");
    // plain floats are unaffected -> still Float
    assert_eq!(types["plain_float"], "Float");
}

#[test]
fn stats_rounding() {
    let wrk = Workdir::new("stats_rounding");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.args(["--everything"])
        .args(["--round", "8"])
        .arg(test_file);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    wrk.create("in2.csv", got);

    // removed stddev_length, variance_length, cv_length, variance, geometric_mean, harmonic_mean,
    // stddev, sem & cv columns as its causing flaky CI test for float values
    let mut cmd = wrk.command("select");
    cmd.arg("!/stddev_length|variance_length|cv_length|variance|geometric_mean|harmonic_mean|stddev|sem|cv/")
        .arg("in2.csv");

    let got2: String = wrk.stdout(&mut cmd);
    let expected2 = wrk.load_test_resource("boston311-100-everything-8places-stats.csv");

    assert_eq!(dos2unix(&got2), dos2unix(&expected2).trim_end());
}

#[test]
fn stats_no_rounding() {
    let wrk = Workdir::new("stats_no_rounding");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .args(["--round", "9999"])
        .arg(test_file);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    wrk.create("in2.csv", got);

    let got2: String = wrk.stdout(&mut cmd);
    let expected2 = wrk.load_test_resource("boston311-100-everything-norounding-stats.csv");

    // this should NOT BE EQUAL as floats are not rounded, and comparing floats is not reliable
    assert_ne!(dos2unix(&got2), dos2unix(&expected2).trim_end());
}

#[test]
fn stats_no_date_inference() {
    let wrk = Workdir::new("stats_no_date_inference");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.args(["--everything"]).arg(test_file);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    wrk.create("in2.csv", got);

    // removed stddev_length, variance_length, cv_length, variance, geometric_mean, harmonic_mean,
    // stddev, sem & cv columns as its causing flaky CI test for float values
    let mut cmd = wrk.command("select");
    cmd.arg("!/stddev_length|variance_length|cv_length|variance|geometric_mean|harmonic_mean|stddev|sem|cv/")
        .arg("in2.csv");

    let got2: String = wrk.stdout(&mut cmd);
    let expected2 = wrk.load_test_resource("boston311-100-everything-nodate-stats.csv");

    assert_eq!(dos2unix(&got2), dos2unix(&expected2).trim_end());
}

#[test]
fn stats_with_date_inference() {
    let wrk = Workdir::new("stats_with_date_inference");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg(test_file)
        .arg("--infer-dates")
        .args(["--dates-whitelist", "all"]);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    wrk.create("in2.csv", got);

    // removed stddev_length, variance_length, cv_length, variance, geometric_mean, harmonic_mean,
    // stddev, sem & cv columns as its causing flaky CI test for float values
    let mut cmd = wrk.command("select");
    cmd.arg("!/stddev_length|variance_length|cv_length|variance|geometric_mean|harmonic_mean|stddev|sem|cv/")
        .arg("in2.csv");

    let got2: String = wrk.stdout(&mut cmd);
    let expected2 = wrk.load_test_resource("boston311-100-everything-date-stats.csv");

    assert_eq!(dos2unix(&got2), dos2unix(&expected2).trim_end());
}

#[test]
fn stats_with_date_inference_default_whitelist() {
    let wrk = Workdir::new("stats_with_date_inference_default_whitelist");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.args(["--everything"])
        .arg(test_file)
        .arg("--infer-dates");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    wrk.create("in2.csv", got);

    // removed stddev_length, variance_length, cv_length, variance, geometric_mean, harmonic_mean,
    // stddev, sem & cv columns as its causing flaky CI test for float values
    let mut cmd = wrk.command("select");
    cmd.arg("!/stddev_length|variance_length|cv_length|variance|geometric_mean|harmonic_mean|stddev|sem|cv/")
        .arg("in2.csv");

    let got2: String = wrk.stdout(&mut cmd);
    let expected2 =
        wrk.load_test_resource("boston311-100-everything-inferdates-defaultwhitelist-stats.csv");

    assert_eq!(dos2unix(&got2), dos2unix(&expected2).trim_end());
}

#[test]
fn stats_with_date_inference_variance_stddev() {
    let wrk = Workdir::new("stats_with_date_inference_variance_stddev");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg(test_file)
        .arg("--infer-dates")
        .args(["--dates-whitelist", "all"]);

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    wrk.create("in2.csv", got);

    let got2: String = wrk.stdout(&mut cmd);
    let expected2 =
        wrk.load_test_resource("boston311-100-everything-date-stats-variance-stddev.csv");

    assert_eq!(dos2unix(&got2), dos2unix(&expected2).trim_end());
}

#[test]
fn stats_with_date_type() {
    let wrk = Workdir::new("stats_with_date_type");
    let test_file = wrk.load_test_file("boston311-100-notime.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg(test_file)
        .arg("--infer-dates")
        .args(["--dates-whitelist", "all"]);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    wrk.create("in2.csv", got);

    // removed stddev_length, variance_length, cv_length, variance, geometric_mean, harmonic_mean,
    // stddev, sem & cv columns as its causing flaky CI test for float values
    let mut cmd = wrk.command("select");
    cmd.arg("!/stddev_length|variance_length|cv_length|variance|geometric_mean|harmonic_mean|stddev|sem|cv/")
        .arg("in2.csv");

    let got2: String = wrk.stdout(&mut cmd);
    let expected2 = wrk.load_test_resource("boston311-100-everything-datenotime-stats.csv");

    assert_eq!(dos2unix(&got2), dos2unix(&expected2).trim_end());
}

#[test]
fn stats_typesonly() {
    let wrk = Workdir::new("stats_typesonly");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.args(["--typesonly"]).arg(test_file);

    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-100-typesonly-stats.csv");

    assert_eq!(dos2unix(&got), dos2unix(&expected).trim_end());
}

#[test]
fn stats_typesonly_with_dates() {
    let wrk = Workdir::new("stats_typesonly_with_dates");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.args(["--typesonly"])
        .arg("--infer-dates")
        .args(["--dates-whitelist", "all"])
        .arg(test_file);

    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-100-typesonly-withdates-stats.csv");

    assert_eq!(dos2unix(&got), dos2unix(&expected).trim_end());
}

#[test]
fn stats_typesonly_dates_whitelist_sniff() {
    let wrk = Workdir::new("stats_typesonly_dates_whitelist_sniff");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.args(["--typesonly"])
        .arg("--infer-dates")
        .args(["--dates-whitelist", "sniff"])
        .arg(test_file);

    let got: String = wrk.stdout(&mut cmd);

    // sniff should detect open_dt, target_dt, closed_dt as DateTime columns
    // which matches the "all" whitelist output
    let expected = wrk.load_test_resource("boston311-100-typesonly-withdates-stats.csv");

    assert_eq!(dos2unix(&got), dos2unix(&expected).trim_end());
}

// The "sniff" dates-whitelist is resolved to concrete date columns and cached with provenance
// (flag_dates_whitelist_raw == "sniff") so a subsequent "sniff" run on an unchanged file reuses
// the sniffed whitelist instead of re-sniffing. Verify both the cached provenance and that the
// warm run produces byte-identical output.
#[test]
fn stats_dates_whitelist_sniff_cache_provenance() {
    use std::path::Path;

    use serde_json::Value;

    let wrk = Workdir::new("stats_dates_whitelist_sniff_cache_provenance");
    let test_file = wrk.load_test_file("boston311-100.csv");

    // first (cold) run with the "sniff" whitelist; force caching
    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-dates")
        .args(["--dates-whitelist", "sniff"])
        .args(["--cache-threshold", "1"])
        .arg(test_file.clone());
    let cold: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    // the cache sidecar must record the resolved date columns AND that they were sniff-derived.
    // Parse the JSON and assert on fields (rather than raw substrings) so the test is robust to
    // serialization formatting changes.
    let sidecar = wrk.path("boston311-100.stats.csv.json");
    assert!(Path::new(&sidecar).exists());
    let json: Value = serde_json::from_str(&std::fs::read_to_string(&sidecar).unwrap()).unwrap();

    // provenance: the original, unresolved whitelist value
    assert_eq!(json["flag_dates_whitelist_raw"], "sniff");
    // resolved whitelist holds the sniffed DateTime columns, not the "sniff" keyword
    let resolved = json["flag_dates_whitelist"].as_str().unwrap();
    assert_ne!(resolved, "sniff");
    assert!(
        resolved.contains("open_dt"),
        "resolved whitelist was {resolved:?}"
    );

    // second (warm) run on the unchanged file must reuse the cached whitelist (no re-sniff)
    // and produce byte-identical output
    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-dates")
        .args(["--dates-whitelist", "sniff"])
        .args(["--cache-threshold", "1"])
        .arg(test_file);
    let warm: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    assert_eq!(cold, warm);
}

// Regression: a stats cache built WITHOUT --infer-dates stores the unresolved literal "sniff"
// whitelist (since "sniff" is the --dates-whitelist default and resolution is gated on
// --infer-dates). A later `--infer-dates --dates-whitelist sniff` run on the unchanged file must
// NOT reuse that literal keyword — doing so would skip sniffing and leave "sniff" as the
// whitelist, breaking date inference. It must sniff and infer dates correctly.
#[test]
fn stats_dates_whitelist_sniff_no_reuse_of_non_infer_cache() {
    let wrk = Workdir::new("stats_dates_whitelist_sniff_no_reuse_of_non_infer_cache");
    let test_file = wrk.load_test_file("boston311-100.csv");

    // build a cache WITHOUT --infer-dates (stores literal "sniff", flag_infer_dates=false)
    let mut cmd = wrk.command("stats");
    cmd.args(["--cache-threshold", "1"]).arg(test_file.clone());
    wrk.run(&mut cmd);

    // now run WITH --infer-dates on the unchanged file; date inference must still work
    let mut cmd = wrk.command("stats");
    cmd.args(["--typesonly"])
        .arg("--infer-dates")
        .args(["--dates-whitelist", "sniff"])
        .args(["--cache-threshold", "1"])
        .arg(test_file);
    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-100-typesonly-withdates-stats.csv");
    assert_eq!(dos2unix(&got), dos2unix(&expected).trim_end());
}

#[test]
fn stats_typesonly_cache_threshold_zero() {
    use std::path::Path;

    let wrk = Workdir::new("stats_typesonly_cache_threshold_zero");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.args(["--typesonly"])
        .arg("--infer-dates")
        .args(["--dates-whitelist", "all"])
        .args(["--cache-threshold", "0"])
        .arg(test_file);

    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-100-typesonly-withdates-stats.csv");

    assert_eq!(dos2unix(&got), dos2unix(&expected).trim_end());

    // check that the stats cache files were NOT created
    assert!(!Path::new(&wrk.path("boston311-100.stats.csv")).exists());
    assert!(!Path::new(&wrk.path("boston311-100.stats.csv.json")).exists());
}

#[test]
fn stats_typesonly_cache() {
    let wrk = Workdir::new("stats_typesonly_cache");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.args(["--typesonly"])
        .arg("--infer-dates")
        .args(["--dates-whitelist", "all"])
        .arg(test_file);

    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-100-typesonly-withdates-stats.csv");

    assert_eq!(dos2unix(&got), dos2unix(&expected).trim_end());
}

#[test]
fn stats_cache() {
    use std::path::Path;

    let wrk = Workdir::new("stats_cache");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-dates")
        .args(["--dates-whitelist", "all"])
        // set cache threshold to 1 to force cache creation
        .args(["--cache-threshold", "1"])
        .arg(test_file);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    wrk.create("in2.csv", got);

    // removed variance, stddev, sem & cv columns as its causing flaky CI test for float values
    let mut cmd = wrk.command("select");
    cmd.arg("!/variance|geometric_mean|harmonic_mean|stddev|sem|cv/")
        .arg("in2.csv");

    let got2: String = wrk.stdout(&mut cmd);
    let expected2 = wrk.load_test_resource("boston311-100-stats.csv");

    assert_eq!(dos2unix(&got2), dos2unix(&expected2).trim_end());

    // check that the stats cache files were created
    assert!(Path::new(&wrk.path("boston311-100.stats.csv")).exists());
    assert!(Path::new(&wrk.path("boston311-100.stats.csv.json")).exists());
}

#[test]
#[ignore = "temporarily ignore while tblshooting fingerprint hash and cache_threshold"]
fn stats_cache_negative_threshold() {
    use std::path::Path;

    let wrk = Workdir::new("stats_cache_negative_threshold");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-dates")
        .arg("--dates-whitelist")
        .arg("all")
        // set cache threshold to -10240 to set autoindex_size to 10 kb
        // and to force cache creation
        .args(["-c", "-10240"])
        .arg(test_file.clone());

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    // the index file SHOULD have been created as the input file size > 10 kb
    assert!(Path::new(&format!("{test_file}.idx")).exists());

    wrk.create("in2.csv", got);

    // removed variance, stddev, sem & cv columns as its causing flaky CI test for float values
    let mut cmd = wrk.command("select");
    cmd.arg("!/variance|geometric_mean|harmonic_mean|stddev|sem|cv/")
        .arg("in2.csv");

    let got2: String = wrk.stdout(&mut cmd);
    let expected2 = wrk.load_test_resource("boston311-100-stats.csv");

    assert_eq!(dos2unix(&got2), dos2unix(&expected2).trim_end());

    // check that the stats cache files were created
    assert!(Path::new(&wrk.path("boston311-100.stats.csv")).exists());
    assert!(Path::new(&wrk.path("boston311-100.stats.csv.json")).exists());
}

#[test]
fn stats_cache_negative_threshold_unmet() {
    use std::path::Path;

    let wrk = Workdir::new("stats_cache_negative_threshold_unmet");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-dates")
        .args(["--dates-whitelist", "all"])
        // set cache threshold to -51200 to set autoindex_size to 50 kb
        // and to force cache creation
        .args(["--cache-threshold", "-51200"])
        .arg(test_file.clone());

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    // the index file SHOULD NOT have been created as the input file < 50 kb
    assert!(!Path::new(&format!("{test_file}.idx")).exists());

    wrk.create("in2.csv", got);

    // removed variance, stddev, sem & cv columns as its causing flaky CI test for float values
    let mut cmd = wrk.command("select");
    cmd.arg("!/variance|geometric_mean|harmonic_mean|stddev|sem|cv/")
        .arg("in2.csv");

    let got2: String = wrk.stdout(&mut cmd);
    let expected2 = wrk.load_test_resource("boston311-100-stats.csv");

    assert_eq!(dos2unix(&got2), dos2unix(&expected2).trim_end());

    // check that the stats cache files were created
    assert!(Path::new(&wrk.path("boston311-100.stats.csv")).exists());
    assert!(Path::new(&wrk.path("boston311-100.stats.csv.json")).exists());
}

#[test]
#[ignore = "temporarily ignore while tblshooting fingerprint hash and cache_threshold"]
fn stats_cache_negative_threshold_five() {
    use std::path::Path;

    let wrk = Workdir::new("stats_cache_negative_threshold_five");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-dates")
        .arg("--dates-whitelist")
        .arg("all")
        // set cache threshold to -10245 to set autoindex_size to 10 kb
        // this creates an index file, and then autodeletes it AND the stats cache files
        .args(["-c", "-10245"])
        .arg(test_file.clone());

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    // the index file WAS CREATED as the input file is > 10k
    // but the index file WAS DELETED after stats exits as the threshold was negative
    // and ends with a 5
    assert!(!Path::new(&format!("{test_file}.idx")).exists());

    wrk.create("in2.csv", got);

    // removed variance, stddev, sem & cv columns as its causing flaky CI test for float values
    let mut cmd = wrk.command("select");
    cmd.arg("!/variance|geometric_mean|harmonic_mean|stddev|sem|cv/")
        .arg("in2.csv");

    let got2: String = wrk.stdout(&mut cmd);
    let expected2 = wrk.load_test_resource("boston311-100-stats.csv");

    assert_eq!(dos2unix(&got2), dos2unix(&expected2).trim_end());

    // check that the stats cache files were created
    assert!(!Path::new(&wrk.path("boston311-100.stats.csv")).exists());
    assert!(!Path::new(&wrk.path("boston311-100.stats.csv.json")).exists());
}

#[test]
fn stats_antimodes_len_500() {
    let wrk = Workdir::new("stats_antimodes_len_500");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("stats");
    cmd.env("QSV_ANTIMODES_LEN", "500")
        .args(["--everything"])
        .arg(test_file);

    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-100-antimodes-len500-stats.csv");

    assert_eq!(dos2unix(&got), dos2unix(&expected).trim_end());
}

#[test]
fn stats_infer_boolean_1_0() {
    let wrk = Workdir::new("stats_infer_boolean_1_0");
    let test_file = wrk.load_test_file("boston311-10-boolean-1or0.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean").arg(test_file);

    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-10-boolean-1or0-stats.csv");

    assert_eq!(dos2unix(&got), dos2unix(&expected).trim_end());
}

#[test]
fn stats_infer_boolean_t_f() {
    let wrk = Workdir::new("stats_infer_boolean_t_f");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean").arg(test_file);

    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-10-boolean-tf-stats.csv");

    assert_eq!(dos2unix(&got), dos2unix(&expected).trim_end());
}

#[test]
fn stats_infer_boolean_true_false() {
    let wrk = Workdir::new("stats_infer_boolean_true_false");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg("true:false")
        .arg(test_file);

    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-10-boolean-tf-stats.csv");

    assert_eq!(dos2unix(&got), dos2unix(&expected).trim_end());
}

#[test]
fn stats_infer_boolean_true_false_error() {
    let wrk = Workdir::new("stats_infer_boolean_true_false_error");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg("true:falsy")
        .arg(test_file);

    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-10-boolean-tf-stats.csv");

    assert_ne!(dos2unix(&got), dos2unix(&expected).trim_end());
}

#[test]
fn stats_infer_boolean_true_false_error_pattern_mismatch() {
    let wrk = Workdir::new("stats_infer_boolean_true_false_error_pattern_mismatch");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg("true:no")
        .arg(test_file);

    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-10-boolean-tf-stats.csv");

    assert_ne!(dos2unix(&got), dos2unix(&expected).trim_end());
}

#[test]
fn stats_typesonly_infer_boolean_t_f() {
    let wrk = Workdir::new("stats_typesonly_infer_boolean_t_f");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--typesonly").arg("--infer-boolean").arg(test_file);

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-10-typesonly-boolean-tf-stats.csv");

    assert_eq!(dos2unix(&got), dos2unix(&expected).trim_end());
}

#[test]
fn stats_typesonly_infer_boolean_t_f_infer_dates() {
    let wrk = Workdir::new("stats_typesonly_infer_boolean_t_f_infer_dates");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--typesonly")
        .arg("--infer-boolean")
        .arg("--infer-dates")
        .arg(test_file);

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-10-typesonly-boolean-tf-inferdates-stats.csv");

    assert_eq!(dos2unix(&got), dos2unix(&expected).trim_end());
}

#[test]
fn stats_infer_boolean_invalid_pattern() {
    let wrk = Workdir::new("stats_infer_boolean_invalid_pattern");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg(":false,yep:,1:0")
        .arg(test_file);

    wrk.assert_err(&mut cmd);
}

#[test]
fn stats_is_ascii() {
    let wrk = Workdir::new("stats_is_ascii");
    let test_file = wrk.load_test_file("boston311-100-with-nonascii.csv");
    let mut cmd = wrk.command("stats");
    cmd.arg(test_file).arg("--force");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    wrk.create("in2.csv", got);

    // removed stddev_length, variance_length, cv_length, variance, geometric_mean, harmonic_mean,
    // stddev, sem & cv columns as its causing flaky CI test for float values
    let mut cmd = wrk.command("select");
    cmd.arg("!/stddev_length|variance_length|cv_length|variance|geometric_mean|harmonic_mean|stddev|sem|cv/")
        .arg("in2.csv");

    let got2: String = wrk.stdout(&mut cmd);
    let expected2 = wrk.load_test_resource("boston311-100-with-nonascii-stats.csv");

    assert_eq!(dos2unix(&got2), dos2unix(&expected2).trim_end());
}

#[test]
fn stats_everything_utf8_japanese_issue817() {
    let wrk = Workdir::new("stats_everything_utf8_japanese");
    let test_file = wrk.load_test_file("utf8-japanesedata.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--everything").arg(test_file);

    wrk.assert_success(&mut cmd);
    // TODO: for now, let's just make sure it doesn't crash
    // comparing utf8 output is a bit tricky, with git line endings
    // and other things

    // let got: String = wrk.stdout(&mut cmd);
    // let expected = wrk.load_test_resource("utf8-japanesedata-stats-everything.csv");
    // assert_eq!(dos2unix(&got).trim_end(), dos2unix(&expected).trim_end());
}

#[test]
fn stats_leading_zero_handling() {
    let wrk = Workdir::new("stats_leading_zero_handling");

    wrk.create(
        "data.csv",
        vec![
            svec!["col1", "col2", "col3"],
            svec!["1", "4321", "01"],
            svec!["2", "3210", "02"],
            svec!["3", "2101", "03"],
            svec!["4", "1012", "04"],
            svec!["5", "0", "10"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--typesonly").arg("data.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["field", "type"],
        svec!["col1", "Integer"],
        svec!["col2", "Integer"],
        svec!["col3", "String"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn stats_zero_cv() {
    let wrk = Workdir::new("stats_zero_cv");

    wrk.create(
        "data.csv",
        vec![
            svec!["col1", "col2", "col3", "col4"],
            svec!["1", "-10", "-100.0", "1000"],
            svec!["2", "-5", "-20.05", "825"],
            svec!["3", "0", "0.0", "10"],
            svec!["4", "5", "20.05", "-900"],
            svec!["5", "10", "100.0", "0"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec![
            "field",
            "type",
            "is_ascii",
            "sum",
            "min",
            "max",
            "range",
            "sort_order",
            "sortiness",
            "min_length",
            "max_length",
            "sum_length",
            "avg_length",
            "stddev_length",
            "variance_length",
            "cv_length",
            "mean",
            "sem",
            "geometric_mean",
            "harmonic_mean",
            "stddev",
            "variance",
            "cv",
            "nullcount",
            "n_negative",
            "n_zero",
            "n_positive",
            "max_precision",
            "sparsity"
        ],
        svec![
            "col1",
            "Integer",
            "",
            "15",
            "1",
            "5",
            "4",
            "Ascending",
            "1",
            "",
            "",
            "",
            "",
            "",
            "",
            "",
            "3",
            "0.6325",
            "2.6052",
            "2.1898",
            "1.4142",
            "2",
            "47.1405",
            "0",
            "0",
            "0",
            "5",
            "",
            "0"
        ],
        svec![
            "col2",
            "Integer",
            "",
            "0",
            "-10",
            "10",
            "20",
            "Ascending",
            "1",
            "",
            "",
            "",
            "",
            "",
            "",
            "",
            "0",
            "3.1623",
            "",
            "",
            "7.0711",
            "50",
            "",
            "0",
            "2",
            "1",
            "2",
            "",
            "0"
        ],
        svec![
            "col3",
            "Float",
            "",
            "0",
            "-100.0",
            "100.0",
            "200",
            "Ascending",
            "1",
            "",
            "",
            "",
            "",
            "",
            "",
            "",
            "0",
            "28.8472",
            "",
            "",
            "64.5043",
            "4160.801",
            "",
            "0",
            "2",
            "1",
            "2",
            "2",
            "0"
        ],
        svec![
            "col4", "Integer", "", "935", "-900", "1000", "1900", "Unsorted", "-0.5", "", "", "",
            "", "", "", "", "187", "304.3603", "", "", "680.5703", "463176", "363.9414", "0", "1",
            "1", "3", "", "0"
        ],
    ];
    assert_eq!(got, expected);
}

#[test]
fn stats_output_tab_delimited() {
    let wrk = Workdir::new("stats_output_tab_delimited");

    wrk.create(
        "data.csv",
        vec![
            svec!["col1", "col2", "col3"],
            svec!["1", "4321", "01"],
            svec!["2", "3210", "02"],
            svec!["3", "2101", "03"],
            svec!["4", "1012", "04"],
            svec!["5", "0", "10"],
        ],
    );

    let out_file = wrk.path("output.tab").to_string_lossy().to_string();

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv").args(["--output", &out_file]);

    wrk.assert_success(&mut cmd);

    let got = std::fs::read_to_string(out_file).unwrap();
    let expected = r#"field	type	is_ascii	sum	min	max	range	sort_order	sortiness	min_length	max_length	sum_length	avg_length	stddev_length	variance_length	cv_length	mean	sem	geometric_mean	harmonic_mean	stddev	variance	cv	nullcount	n_negative	n_zero	n_positive	max_precision	sparsity
col1	Integer		15	1	5	4	Ascending	1								3	0.6325	2.6052	2.1898	1.4142	2	47.1405	0	0	0	5		0
col2	Integer		10644	0	4321	4321	Descending	-1								2128.8	685.6979	0		1533.267	2350907.76	72.0249	0	0	1	4		0
col3	String	true		01	10		Ascending	1	2	2	10	2	0	0	0								0					0
"#;
    assert_eq!(got, expected);
}

// Regression test: when --output is TSV, the .stats.csv cache must remain
// comma-delimited so downstream commands (moarstats, schema, etc.) can read it.
#[test]
fn stats_cache_always_csv_when_output_is_tsv() {
    let wrk = Workdir::new("stats_cache_always_csv_when_output_is_tsv");

    wrk.create(
        "data.csv",
        vec![
            svec!["name", "age"],
            svec!["Alice", "30"],
            svec!["Bob", "25"],
        ],
    );

    let out_file = wrk.path("output.tsv").to_string_lossy().to_string();

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .args(["--output", &out_file])
        .args(["--cache-threshold", "1"]);

    wrk.assert_success(&mut cmd);

    // the output file should be tab-delimited
    let output_content = std::fs::read_to_string(&out_file).unwrap();
    assert!(
        output_content.contains('\t'),
        "output.tsv should be tab-delimited"
    );
    // verify the output is valid stats TSV by checking header fields
    let tsv_first_line = output_content.lines().next().unwrap();
    let tsv_fields: Vec<&str> = tsv_first_line.split('\t').collect();
    assert_eq!(tsv_fields[0], "field", "first TSV header should be 'field'");
    assert_eq!(tsv_fields[1], "type", "second TSV header should be 'type'");

    // the cache file must be comma-delimited (NOT tab-delimited)
    let cache_path = wrk.path("data.stats.csv");
    let cache_content = std::fs::read_to_string(&cache_path).unwrap();
    let first_line = cache_content.lines().next().unwrap();
    assert!(
        first_line.contains(','),
        "stats cache should be comma-delimited, got: {first_line}"
    );
    assert!(
        !first_line.contains('\t'),
        "stats cache must NOT be tab-delimited, got: {first_line}"
    );
    // verify cache has expected stats headers
    let cache_fields: Vec<&str> = first_line.split(',').collect();
    assert_eq!(
        cache_fields[0], "field",
        "first cache header should be 'field'"
    );
    assert_eq!(
        cache_fields[1], "type",
        "second cache header should be 'type'"
    );
}

// Regression test: --output with SSV extension should produce semicolon-delimited
// output while keeping the cache comma-delimited.
#[test]
fn stats_cache_always_csv_when_output_is_ssv() {
    let wrk = Workdir::new("stats_cache_always_csv_when_output_is_ssv");

    wrk.create(
        "data.csv",
        vec![
            svec!["name", "age"],
            svec!["Alice", "30"],
            svec!["Bob", "25"],
        ],
    );

    let out_file = wrk.path("output.ssv").to_string_lossy().to_string();

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .args(["--output", &out_file])
        .args(["--cache-threshold", "1"]);

    wrk.assert_success(&mut cmd);

    // the output file should be semicolon-delimited with valid stats headers
    let output_content = std::fs::read_to_string(&out_file).unwrap();
    assert!(
        output_content.contains(';'),
        "output.ssv should be semicolon-delimited"
    );
    let ssv_first_line = output_content.lines().next().unwrap();
    let ssv_fields: Vec<&str> = ssv_first_line.split(';').collect();
    assert_eq!(ssv_fields[0], "field", "first SSV header should be 'field'");
    assert_eq!(ssv_fields[1], "type", "second SSV header should be 'type'");

    // the cache file must still be comma-delimited
    let cache_path = wrk.path("data.stats.csv");
    let cache_content = std::fs::read_to_string(&cache_path).unwrap();
    let first_line = cache_content.lines().next().unwrap();
    assert!(
        first_line.contains(','),
        "stats cache should be comma-delimited, got: {first_line}"
    );
    assert!(
        !first_line.contains(';'),
        "stats cache must NOT be semicolon-delimited, got: {first_line}"
    );
}

// Regression test: --output with .csv.sz (Snappy-compressed, comma delimiter)
// should produce a Snappy-compressed output while the cache remains plain
// uncompressed CSV.
#[test]
fn stats_cache_always_csv_when_output_is_snappy() {
    let wrk = Workdir::new("stats_cache_always_csv_when_output_is_snappy");

    wrk.create(
        "data.csv",
        vec![
            svec!["name", "age"],
            svec!["Alice", "30"],
            svec!["Bob", "25"],
        ],
    );

    let out_file = wrk.path("output.csv.sz").to_string_lossy().to_string();

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .args(["--output", &out_file])
        .args(["--cache-threshold", "1"]);

    wrk.assert_success(&mut cmd);

    // the output file should exist and NOT be readable as plain text
    // (it's Snappy-compressed)
    let output_bytes = std::fs::read(&out_file).unwrap();
    assert!(
        !output_bytes.is_empty(),
        "output.csv.sz should not be empty"
    );
    // Snappy framing format starts with 0xff 0x06 0x00 0x00
    assert_eq!(
        output_bytes[0], 0xff,
        "output.csv.sz should start with Snappy magic byte"
    );

    // decompress and verify the output is valid CSV stats
    let mut cmd = wrk.command("snappy");
    cmd.arg("decompress").arg(&out_file);

    let got: String = wrk.stdout(&mut cmd);
    let first_line = got.lines().next().unwrap();
    let fields: Vec<&str> = first_line.split(',').collect();
    assert_eq!(
        fields[0], "field",
        "decompressed output first header should be 'field'"
    );
    assert_eq!(
        fields[1], "type",
        "decompressed output second header should be 'type'"
    );

    // the cache file must be plain uncompressed CSV
    let cache_path = wrk.path("data.stats.csv");
    let cache_content = std::fs::read_to_string(&cache_path).unwrap();
    let cache_first_line = cache_content.lines().next().unwrap();
    let cache_fields: Vec<&str> = cache_first_line.split(',').collect();
    assert_eq!(
        cache_fields[0], "field",
        "cache first header should be 'field'"
    );
    assert_eq!(
        cache_fields[1], "type",
        "cache second header should be 'type'"
    );
    // ensure cache is NOT Snappy-compressed (readable as UTF-8 means it's plain text)
    assert!(
        cache_content.starts_with("field,"),
        "cache should be plain CSV starting with 'field,', got: {}",
        &cache_content[..cache_content.len().min(50)]
    );
}

// Regression test: --output with .tsv.sz (Snappy + non-comma delimiter)
// should still write a comma-delimited cache file.
#[test]
fn stats_cache_always_csv_when_output_is_tsv_sz() {
    let wrk = Workdir::new("stats_cache_always_csv_when_output_is_tsv_sz");

    wrk.create(
        "data.csv",
        vec![
            svec!["name", "age"],
            svec!["Alice", "30"],
            svec!["Bob", "25"],
        ],
    );

    let out_file = wrk.path("output.tsv.sz").to_string_lossy().to_string();

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .args(["--output", &out_file])
        .args(["--cache-threshold", "1"]);

    wrk.assert_success(&mut cmd);

    // decompress and verify the output is tab-delimited
    let mut cmd = wrk.command("snappy");
    cmd.arg("decompress").arg(&out_file);

    let got: String = wrk.stdout(&mut cmd);
    assert!(
        got.contains('\t'),
        "decompressed output.tsv.sz should be tab-delimited, got: {got}"
    );
    // verify decompressed output has valid stats headers
    let tsv_first_line = got.lines().next().unwrap();
    let tsv_fields: Vec<&str> = tsv_first_line.split('\t').collect();
    assert_eq!(
        tsv_fields[0], "field",
        "first decompressed TSV header should be 'field'"
    );
    assert_eq!(
        tsv_fields[1], "type",
        "second decompressed TSV header should be 'type'"
    );

    // the cache file must still be comma-delimited
    let cache_path = wrk.path("data.stats.csv");
    let cache_content = std::fs::read_to_string(&cache_path).unwrap();
    let first_line = cache_content.lines().next().unwrap();
    assert!(
        first_line.contains(','),
        "stats cache should be comma-delimited, got: {first_line}"
    );
    assert!(
        !first_line.contains('\t'),
        "stats cache must NOT be tab-delimited, got: {first_line}"
    );
}

#[test]
fn stats_output_ssv_delimited() {
    let wrk = Workdir::new("stats_output_ssv_delimited");

    wrk.create(
        "data.csv",
        vec![
            svec!["col1", "col2", "col3"],
            svec!["1", "4321", "01"],
            svec!["2", "3210", "02"],
            svec!["3", "2101", "03"],
            svec!["4", "1012", "04"],
            svec!["5", "0", "10"],
        ],
    );

    let out_file = wrk.path("output.ssv").to_string_lossy().to_string();

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv").args(["--output", &out_file]);

    wrk.assert_success(&mut cmd);

    let got = std::fs::read_to_string(out_file).unwrap();
    let expected = r#"field;type;is_ascii;sum;min;max;range;sort_order;sortiness;min_length;max_length;sum_length;avg_length;stddev_length;variance_length;cv_length;mean;sem;geometric_mean;harmonic_mean;stddev;variance;cv;nullcount;n_negative;n_zero;n_positive;max_precision;sparsity
col1;Integer;;15;1;5;4;Ascending;1;;;;;;;;3;0.6325;2.6052;2.1898;1.4142;2;47.1405;0;0;0;5;;0
col2;Integer;;10644;0;4321;4321;Descending;-1;;;;;;;;2128.8;685.6979;0;;1533.267;2350907.76;72.0249;0;0;1;4;;0
col3;String;true;;01;10;;Ascending;1;2;2;10;2;0;0;0;;;;;;;;0;;;;;0
"#;
    assert_eq!(got, expected);
}

#[test]
fn stats_output_csvsz_delimited() {
    let wrk = Workdir::new("stats_output_csvsz_delimited");

    wrk.create(
        "data.csv",
        vec![
            svec!["col1", "col2", "col3"],
            svec!["1", "4321", "01"],
            svec!["2", "3210", "02"],
            svec!["3", "2101", "03"],
            svec!["4", "1012", "04"],
            svec!["5", "0", "10"],
        ],
    );

    let out_file = wrk.path("output.csv.sz").to_string_lossy().to_string();

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv").args(["--output", &out_file]);

    wrk.assert_success(&mut cmd);

    let mut cmd = wrk.command("snappy");
    cmd.arg("decompress").arg(out_file.clone());

    let got: String = wrk.stdout(&mut cmd);
    let expected = r#"field,type,is_ascii,sum,min,max,range,sort_order,sortiness,min_length,max_length,sum_length,avg_length,stddev_length,variance_length,cv_length,mean,sem,geometric_mean,harmonic_mean,stddev,variance,cv,nullcount,n_negative,n_zero,n_positive,max_precision,sparsity
col1,Integer,,15,1,5,4,Ascending,1,,,,,,,,3,0.6325,2.6052,2.1898,1.4142,2,47.1405,0,0,0,5,,0
col2,Integer,,10644,0,4321,4321,Descending,-1,,,,,,,,2128.8,685.6979,0,,1533.267,2350907.76,72.0249,0,0,1,4,,0
col3,String,true,,01,10,,Ascending,1,2,2,10,2,0,0,0,,,,,,,,0,,,,,0"#;
    assert_eq!(got, expected);
}

mod stats_infer_nothing {
    // Only test CSV data with headers.
    // Empty CSV data with no headers won't produce any statistical analysis.
    use super::test_stats;
    stats_test_headers!(stats_infer_nothing, "type", &[], "NULL");
}

mod stats_zero_cardinality {
    use super::test_stats;
    stats_test_headers!(stats_zero_cardinality, "cardinality", &[], "0");
}

mod stats_zero_mode {
    use super::test_stats;
    stats_test_headers!(stats_zero_mode, "mode", &[], "N/A");
}

mod stats_zero_mean {
    use super::test_stats;
    stats_test_headers!(stats_zero_mean, "mean", &[], "");
}

mod stats_zero_median {
    use super::test_stats;
    stats_test_headers!(stats_zero_median, "median", &[], "");
}

mod stats_zero_quartiles {
    use super::test_stats;
    stats_test_headers!(stats_zero_quartiles, "quartiles", &[], ",,,,,");
}

mod stats_header_fields {
    use super::test_stats;
    stats_test_headers!(stats_header_field_name, "field", &["a"], "header");
    stats_test_no_headers!(stats_header_no_field_name, "field", &["a"], "0");
}

#[test]
fn stats_vis_whitespace() {
    let wrk = Workdir::new("stats_vis_whitespace");

    // Create test data with various types of whitespace
    wrk.create(
        "data.csv",
        vec![
            svec!["col1", "col2", "col3"],
            svec!["value\t", "\tvalue", "\tvalue\t"], // Tabs
            svec!["value\r", "\rvalue", "\rvalue\r"], // Carriage returns
            svec!["value\n", "\nvalue", "\nvalue\n"], // Line feeds
            svec!["value\n", "\nvalue", "\nvalue\n"], // Line feeds repeat
            svec!["no_whitespace", "also_none", "clean"],
            svec![
                "the spaces in this field are visible as normal spaces",
                "    ",
                "the trailing spaces are left alone   "
            ],
            svec![
                "z obscure whitespace \u{000B} \u{000C} \u{0009} \u{0085} \u{200E} \u{200F} \
                 \u{2028} \u{2029} are also visible",
                "",
                "          "
            ],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--vis-whitespace")
        .arg("--everything")
        .arg("data.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    // Create expected output with visualized whitespace using the exact markers
    #[rustfmt::skip]
    let expected = vec![
        svec!["field", "type", "is_ascii", "sum", "min", "max", "range", "sort_order", "sortiness", "min_length", "max_length", "sum_length", "avg_length", "stddev_length", "variance_length", "cv_length", "mean", "sem", "geometric_mean", "harmonic_mean", "stddev", "variance", "cv", "nullcount", "n_negative", "n_zero", "n_positive", "max_precision", "sparsity", "mad", "lower_outer_fence", "lower_inner_fence", "q1", "q2_median", "q3", "iqr", "upper_inner_fence", "upper_outer_fence", "skewness", "cardinality", "uniqueness_ratio", "mode", "mode_count", "mode_occurrences", "antimode", "antimode_count", "antimode_occurrences", "percentiles", "zero_padded_numeric"],
        svec!["col1", "String", "false", "", "no_whitespace", "z obscure whitespace 《⋮》 《␌》 《→》 《␤》 《␎》 《␏》 《␊》 《␍》 are also visible", "", "Unsorted", "0.3333", "6", "62", "152", "21.7143", "22.883", "523.6327", "1.0538", "", "", "", "", "", "", "", "0", "", "", "", "", "0", "", "", "", "", "", "", "", "", "", "", "6", "0.8571", "value《¶》", "1", "2", "no_whitespace|the spaces in this field are visible as normal spaces|value《→》|value《⏎》|z obscure whitespa...", "5", "1", "", ""],
        svec!["col2", "String", "true", "", "《→》value", "also_none", "", "Unsorted", "0.2", "0", "9", "37", "5.2857", "2.5475", "6.4898", "0.482", "", "", "", "", "", "", "", "1", "", "", "", "", "0.1429", "", "", "", "", "", "", "", "", "", "", "6", "0.8571", "《¶》value", "1", "2", "NULL|《→》value|《⏎》value|    |also_none", "5", "1", "", ""],
        svec!["col3", "String", "true", "", "《→》value《→》", "the trailing spaces are left alone   ", "", "Unsorted", "0.3333", "5", "37", "80", "11.4286", "10.5269", "110.8163", "0.9211", "", "", "", "", "", "", "", "0", "", "", "", "", "0", "", "", "", "", "", "", "", "", "", "", "6", "0.8571", "《¶》value《¶》", "1", "2", "《→》value《→》|《⏎》value《⏎》|          |clean|the trailing spaces are left alone   ", "5", "1", "", ""]];
    //
    assert_eq!(got, expected);
}

#[test]
fn stats_percentiles() {
    let wrk = Workdir::new("stats_percentiles");
    wrk.create(
        "data.csv",
        vec![
            svec!["n"],
            svec!["1"],
            svec!["2"],
            svec!["3"],
            svec!["4"],
            svec!["5"],
            svec!["6"],
            svec!["7"],
            svec!["8"],
            svec!["9"],
            svec!["10"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("10,25,50,75,90");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let expected = vec![
        svec![
            "field",
            "type",
            "is_ascii",
            "sum",
            "min",
            "max",
            "range",
            "sort_order",
            "sortiness",
            "min_length",
            "max_length",
            "sum_length",
            "avg_length",
            "stddev_length",
            "variance_length",
            "cv_length",
            "mean",
            "sem",
            "geometric_mean",
            "harmonic_mean",
            "stddev",
            "variance",
            "cv",
            "nullcount",
            "n_negative",
            "n_zero",
            "n_positive",
            "max_precision",
            "sparsity",
            "percentiles",
        ],
        svec![
            "n",
            "Integer",
            "",
            "55",
            "1",
            "10",
            "9",
            "Ascending",
            "1",
            "",
            "",
            "",
            "",
            "",
            "",
            "",
            "5.5",
            "0.9083",
            "4.5287",
            "3.4142",
            "2.8723",
            "8.25",
            "52.2233",
            "0",
            "0",
            "0",
            "10",
            "",
            "0",
            "10: 1|25: 3|50: 5|75: 8|90: 9",
        ],
    ];

    assert_eq!(got, expected);
}

#[test]
fn stats_percentiles_floats() {
    let wrk = Workdir::new("stats_percentiles");
    wrk.create(
        "data.csv",
        vec![
            svec!["n"],
            svec!["1"],
            svec!["2"],
            svec!["3"],
            svec!["4"],
            svec!["5"],
            svec!["6"],
            svec!["7"],
            svec!["8"],
            svec!["9"],
            svec!["10"],
            svec!["11"],
            svec!["12"],
            svec!["13"],
            svec!["14"],
            svec!["15"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("10.5,25.25,50.75,75.6,90.1");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let expected = vec![
        svec![
            "field",
            "type",
            "is_ascii",
            "sum",
            "min",
            "max",
            "range",
            "sort_order",
            "sortiness",
            "min_length",
            "max_length",
            "sum_length",
            "avg_length",
            "stddev_length",
            "variance_length",
            "cv_length",
            "mean",
            "sem",
            "geometric_mean",
            "harmonic_mean",
            "stddev",
            "variance",
            "cv",
            "nullcount",
            "n_negative",
            "n_zero",
            "n_positive",
            "max_precision",
            "sparsity",
            "percentiles",
        ],
        svec![
            "n",
            "Integer",
            "",
            "120",
            "1",
            "15",
            "14",
            "Ascending",
            "1",
            "",
            "",
            "",
            "",
            "",
            "",
            "",
            "8",
            "1.1155",
            "6.4234",
            "4.5205",
            "4.3205",
            "18.6667",
            "54.0062",
            "0",
            "0",
            "0",
            "15",
            "",
            "0",
            "10.5: 2|25.25: 4|50.75: 8|75.6: 12|90.1: 14"
        ],
    ];

    //
    assert_eq!(got, expected);
}

#[test]
fn stats_percentiles_with_dates() {
    let wrk = Workdir::new("stats_percentiles_dates");
    wrk.create(
        "data.csv",
        vec![
            svec!["date"],
            svec!["2020-01-01"],
            svec!["2020-02-01"],
            svec!["2020-03-01"],
            svec!["2020-04-01"],
            svec!["2020-05-01"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("25,50,75")
        .arg("--infer-dates");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let expected = vec![
        svec![
            "field",
            "type",
            "is_ascii",
            "sum",
            "min",
            "max",
            "range",
            "sort_order",
            "sortiness",
            "min_length",
            "max_length",
            "sum_length",
            "avg_length",
            "stddev_length",
            "variance_length",
            "cv_length",
            "mean",
            "sem",
            "geometric_mean",
            "harmonic_mean",
            "stddev",
            "variance",
            "cv",
            "nullcount",
            "n_negative",
            "n_zero",
            "n_positive",
            "max_precision",
            "sparsity",
            "percentiles",
        ],
        svec![
            "date",
            "Date",
            "",
            "",
            "2020-01-01",
            "2020-05-01",
            "121",
            "Ascending",
            "1",
            "",
            "",
            "",
            "",
            "",
            "",
            "",
            "2020-03-01",
            "19.10099",
            "18322.55022",
            "18322.50044",
            "42.71112",
            "1824.24",
            "0.2331",
            "0",
            "",
            "",
            "",
            "",
            "0",
            "25: 2020-02-01|50: 2020-03-01|75: 2020-04-01",
        ],
    ];

    assert_eq!(got, expected);
}

#[test]
fn stats_percentiles_with_nulls() {
    let wrk = Workdir::new("stats_percentiles_nulls");
    wrk.create(
        "data.csv",
        vec![
            svec!["n"],
            svec!["1"],
            svec![""],
            svec!["3"],
            svec![""],
            svec!["5"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("25,50,75");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let expected = vec![
        svec![
            "field",
            "type",
            "is_ascii",
            "sum",
            "min",
            "max",
            "range",
            "sort_order",
            "sortiness",
            "min_length",
            "max_length",
            "sum_length",
            "avg_length",
            "stddev_length",
            "variance_length",
            "cv_length",
            "mean",
            "sem",
            "geometric_mean",
            "harmonic_mean",
            "stddev",
            "variance",
            "cv",
            "nullcount",
            "n_negative",
            "n_zero",
            "n_positive",
            "max_precision",
            "sparsity",
            "percentiles",
        ],
        svec![
            "n",
            "Integer",
            "",
            "9",
            "1",
            "5",
            "4",
            "Ascending",
            "1",
            "",
            "",
            "",
            "",
            "",
            "",
            "",
            "3",
            "0.9428",
            "2.4662",
            "1.9565",
            "1.633",
            "2.6667",
            "54.4331",
            "2",
            "0",
            "0",
            "3",
            "",
            "0.4",
            "25: 1|50: 3|75: 5",
        ],
    ];

    assert_eq!(got, expected);
}

#[test]
fn stats_percentiles_mixed_types() {
    let wrk = Workdir::new("stats_percentiles_mixed");
    wrk.create(
        "data.csv",
        vec![
            svec!["mixed"],
            svec!["1"],
            svec!["2.5"],
            svec!["abc"],
            svec!["3"],
            svec!["4.7"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("25,50,75");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let expected = vec![
        svec![
            "field",
            "type",
            "is_ascii",
            "sum",
            "min",
            "max",
            "range",
            "sort_order",
            "sortiness",
            "min_length",
            "max_length",
            "sum_length",
            "avg_length",
            "stddev_length",
            "variance_length",
            "cv_length",
            "mean",
            "sem",
            "geometric_mean",
            "harmonic_mean",
            "stddev",
            "variance",
            "cv",
            "nullcount",
            "n_negative",
            "n_zero",
            "n_positive",
            "max_precision",
            "sparsity",
            "percentiles",
        ],
        svec![
            "mixed", "String", "true", "", "1", "abc", "", "Unsorted", "0.5", "1", "3", "11",
            "2.2", "0.9798", "0.96", "0.4454", "", "", "", "", "", "", "", "0", "", "", "", "",
            "0", "",
        ],
    ];

    assert_eq!(got, expected);
}

#[test]
fn stats_percentiles_edge_cases() {
    let wrk = Workdir::new("stats_percentiles_edge");
    wrk.create(
        "data.csv",
        vec![
            svec!["n"],
            svec!["0.0000001"],
            svec!["1000000"],
            svec!["-999999"],
            svec!["0.0000099"],
            svec!["999999"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("10,25,50,75,90");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let expected = vec![
        svec![
            "field",
            "type",
            "is_ascii",
            "sum",
            "min",
            "max",
            "range",
            "sort_order",
            "sortiness",
            "min_length",
            "max_length",
            "sum_length",
            "avg_length",
            "stddev_length",
            "variance_length",
            "cv_length",
            "mean",
            "sem",
            "geometric_mean",
            "harmonic_mean",
            "stddev",
            "variance",
            "cv",
            "nullcount",
            "n_negative",
            "n_zero",
            "n_positive",
            "max_precision",
            "sparsity",
            "percentiles",
        ],
        svec![
            "n",
            "Float",
            "",
            "1000000",
            "-999999.0",
            "1000000.0",
            "1999999",
            "Unsorted",
            "0.5",
            "",
            "",
            "",
            "",
            "",
            "",
            "",
            "200000",
            "334663.7716",
            "",
            "",
            "748330.9428",
            "559999199999.6",
            "374.1655",
            "0",
            "1",
            "0",
            "4",
            "4",
            "0",
            "10: -999999|25: 0|50: 0|75: 999999|90: 1000000"
        ],
    ];

    assert_eq!(got, expected);
}

#[test]
fn stats_percentiles_custom_list() {
    let wrk = Workdir::new("stats_percentiles_custom");
    wrk.create(
        "data.csv",
        vec![
            svec!["n"],
            svec!["1"],
            svec!["2"],
            svec!["3"],
            svec!["4"],
            svec!["5"],
            svec!["6"],
            svec!["7"],
            svec!["8"],
            svec!["9"],
            svec!["10"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("1,5,33.3,66.6,95,99");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let expected = vec![
        svec![
            "field",
            "type",
            "is_ascii",
            "sum",
            "min",
            "max",
            "range",
            "sort_order",
            "sortiness",
            "min_length",
            "max_length",
            "sum_length",
            "avg_length",
            "stddev_length",
            "variance_length",
            "cv_length",
            "mean",
            "sem",
            "geometric_mean",
            "harmonic_mean",
            "stddev",
            "variance",
            "cv",
            "nullcount",
            "n_negative",
            "n_zero",
            "n_positive",
            "max_precision",
            "sparsity",
            "percentiles",
        ],
        svec![
            "n",
            "Integer",
            "",
            "55",
            "1",
            "10",
            "9",
            "Ascending",
            "1",
            "",
            "",
            "",
            "",
            "",
            "",
            "",
            "5.5",
            "0.9083",
            "4.5287",
            "3.4142",
            "2.8723",
            "8.25",
            "52.2233",
            "0",
            "0",
            "0",
            "10",
            "",
            "0",
            "1: 1|5: 1|33.3: 4|66.6: 7|95: 10|99: 10",
        ],
    ];

    assert_eq!(got, expected);
}

#[test]
fn stats_percentiles_single_value() {
    let wrk = Workdir::new("stats_percentiles_single");
    wrk.create("data.csv", vec![svec!["n"], svec!["42"]]);

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("25,50,75");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let expected = vec![
        svec![
            "field",
            "type",
            "is_ascii",
            "sum",
            "min",
            "max",
            "range",
            "sort_order",
            "sortiness",
            "min_length",
            "max_length",
            "sum_length",
            "avg_length",
            "stddev_length",
            "variance_length",
            "cv_length",
            "mean",
            "sem",
            "geometric_mean",
            "harmonic_mean",
            "stddev",
            "variance",
            "cv",
            "nullcount",
            "n_negative",
            "n_zero",
            "n_positive",
            "max_precision",
            "sparsity",
            "percentiles",
        ],
        svec![
            "n",
            "Integer",
            "",
            "42",
            "42",
            "42",
            "0",
            "Unsorted",
            "0",
            "",
            "",
            "",
            "",
            "",
            "",
            "",
            "42",
            "0",
            "42",
            "42",
            "0",
            "0",
            "0",
            "0",
            "0",
            "0",
            "1",
            "",
            "0",
            "25: 42|50: 42|75: 42",
        ],
    ];

    assert_eq!(got, expected);
}

#[test]
fn stats_percentiles_deciles_lowercase() {
    let wrk = Workdir::new("stats_percentiles_deciles_lowercase");
    wrk.create(
        "data.csv",
        vec![
            svec!["n"],
            svec!["4"],
            svec!["8"],
            svec!["1"],
            svec!["5"],
            svec!["3"],
            svec!["6"],
            svec!["7"],
            svec!["9"],
            svec!["2"],
            svec!["10"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("deciles");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    // Find the percentiles column value
    let headers = &got[0];
    let percentiles_idx = headers
        .iter()
        .position(|h| h == "percentiles")
        .expect("percentiles column should exist");
    let percentiles_value = &got[1][percentiles_idx];

    // Deciles should expand to: 10,20,30,40,50,60,70,80,90 (i.e., the 10th, 20th, ..., 90th
    // percentiles; 9 values) Verify we have 9 percentile values separated by |
    let percentile_count = percentiles_value.matches('|').count() + 1;
    assert_eq!(
        percentile_count, 9,
        "Deciles should produce 9 percentile values"
    );
    // Verify the values are reasonable (should contain values from the dataset)
    assert!(
        percentiles_value.contains("1"),
        "Should contain first value"
    );
    assert!(percentiles_value.contains("9"), "Should contain last value");
}

#[test]
fn stats_percentiles_deciles_uppercase() {
    let wrk = Workdir::new("stats_percentiles_deciles_uppercase");
    wrk.create(
        "data.csv",
        vec![
            svec!["n"],
            svec!["1"],
            svec!["2"],
            svec!["3"],
            svec!["4"],
            svec!["5"],
            svec!["6"],
            svec!["7"],
            svec!["8"],
            svec!["9"],
            svec!["10"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("DECILES");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let headers = &got[0];
    let percentiles_idx = headers
        .iter()
        .position(|h| h == "percentiles")
        .expect("percentiles column should exist");
    let percentiles_value = &got[1][percentiles_idx];

    // Should work the same as lowercase - verify structure
    let percentile_count = percentiles_value.matches('|').count() + 1;
    assert_eq!(
        percentile_count, 9,
        "Deciles should produce 9 percentile values"
    );
}

#[test]
fn stats_percentiles_deciles_mixed_case() {
    let wrk = Workdir::new("stats_percentiles_deciles_mixed_case");
    wrk.create(
        "data.csv",
        vec![
            svec!["n"],
            svec!["1"],
            svec!["2"],
            svec!["3"],
            svec!["4"],
            svec!["5"],
            svec!["6"],
            svec!["7"],
            svec!["8"],
            svec!["9"],
            svec!["10"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("DeCiLeS");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let headers = &got[0];
    let percentiles_idx = headers
        .iter()
        .position(|h| h == "percentiles")
        .expect("percentiles column should exist");
    let percentiles_value = &got[1][percentiles_idx];

    // Should work the same as lowercase (case-insensitive) - verify structure
    let percentile_count = percentiles_value.matches('|').count() + 1;
    assert_eq!(
        percentile_count, 9,
        "Deciles should produce 9 percentile values"
    );
}

#[test]
fn stats_percentiles_quintiles_lowercase() {
    let wrk = Workdir::new("stats_percentiles_quintiles_lowercase");
    wrk.create(
        "data.csv",
        vec![
            svec!["n"],
            svec!["1"],
            svec!["2"],
            svec!["3"],
            svec!["4"],
            svec!["5"],
            svec!["6"],
            svec!["7"],
            svec!["8"],
            svec!["9"],
            svec!["10"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("quintiles");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let headers = &got[0];
    let percentiles_idx = headers
        .iter()
        .position(|h| h == "percentiles")
        .expect("percentiles column should exist");
    let percentiles_value = &got[1][percentiles_idx];

    // Quintiles should expand to: 20,40,60,80 (4 values)
    // Verify we have 4 percentile values separated by |
    let percentile_count = percentiles_value.matches('|').count() + 1;
    assert_eq!(
        percentile_count, 4,
        "Quintiles should produce 4 percentile values"
    );
    // Verify the values are reasonable
    assert!(
        percentiles_value.contains("2") || percentiles_value.contains("4"),
        "Should contain reasonable values"
    );
}

#[test]
fn stats_percentiles_quintiles_uppercase() {
    let wrk = Workdir::new("stats_percentiles_quintiles_uppercase");
    wrk.create(
        "data.csv",
        vec![
            svec!["n"],
            svec!["1"],
            svec!["2"],
            svec!["3"],
            svec!["4"],
            svec!["5"],
            svec!["6"],
            svec!["7"],
            svec!["8"],
            svec!["9"],
            svec!["10"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("QUINTILES");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let headers = &got[0];
    let percentiles_idx = headers
        .iter()
        .position(|h| h == "percentiles")
        .expect("percentiles column should exist");
    let percentiles_value = &got[1][percentiles_idx];

    // Should work the same as lowercase - verify structure
    let percentile_count = percentiles_value.matches('|').count() + 1;
    assert_eq!(
        percentile_count, 4,
        "Quintiles should produce 4 percentile values"
    );
}

#[test]
fn stats_percentiles_quintiles_mixed_case() {
    let wrk = Workdir::new("stats_percentiles_quintiles_mixed_case");
    wrk.create(
        "data.csv",
        vec![
            svec!["n"],
            svec!["1"],
            svec!["2"],
            svec!["3"],
            svec!["4"],
            svec!["5"],
            svec!["6"],
            svec!["7"],
            svec!["8"],
            svec!["9"],
            svec!["10"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("QuInTiLeS");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let headers = &got[0];
    let percentiles_idx = headers
        .iter()
        .position(|h| h == "percentiles")
        .expect("percentiles column should exist");
    let percentiles_value = &got[1][percentiles_idx];

    // Should work the same as lowercase (case-insensitive) - verify structure
    let percentile_count = percentiles_value.matches('|').count() + 1;
    assert_eq!(
        percentile_count, 4,
        "Quintiles should produce 4 percentile values"
    );
}

#[test]
fn stats_percentiles_deciles_with_more_values() {
    let wrk = Workdir::new("stats_percentiles_deciles_more_values");
    wrk.create(
        "data.csv",
        vec![
            svec!["n"],
            svec!["1"],
            svec!["2"],
            svec!["3"],
            svec!["4"],
            svec!["5"],
            svec!["6"],
            svec!["7"],
            svec!["8"],
            svec!["9"],
            svec!["10"],
            svec!["11"],
            svec!["12"],
            svec!["13"],
            svec!["14"],
            svec!["15"],
            svec!["16"],
            svec!["17"],
            svec!["18"],
            svec!["19"],
            svec!["20"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("deciles");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let headers = &got[0];
    let percentiles_idx = headers
        .iter()
        .position(|h| h == "percentiles")
        .expect("percentiles column should exist");
    let percentiles_value = &got[1][percentiles_idx];

    // Deciles: 10,20,30,40,50,60,70,80,90 (9 values)
    // Verify we have 9 percentile values
    let percentile_count = percentiles_value.matches('|').count() + 1;
    assert_eq!(
        percentile_count, 9,
        "Deciles should produce 9 percentile values"
    );
    // Verify values are from the dataset range
    assert!(
        percentiles_value.contains("2") || percentiles_value.contains("1"),
        "Should contain values from dataset"
    );
    assert!(
        percentiles_value.contains("20")
            || percentiles_value.contains("18")
            || percentiles_value.contains("19"),
        "Should contain values near end of dataset"
    );
}

#[test]
fn stats_percentiles_quintiles_with_more_values() {
    let wrk = Workdir::new("stats_percentiles_quintiles_more_values");
    wrk.create(
        "data.csv",
        vec![
            svec!["n"],
            svec!["1"],
            svec!["2"],
            svec!["3"],
            svec!["4"],
            svec!["5"],
            svec!["6"],
            svec!["7"],
            svec!["8"],
            svec!["9"],
            svec!["10"],
            svec!["11"],
            svec!["12"],
            svec!["13"],
            svec!["14"],
            svec!["15"],
            svec!["16"],
            svec!["17"],
            svec!["18"],
            svec!["19"],
            svec!["20"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("quintiles");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let headers = &got[0];
    let percentiles_idx = headers
        .iter()
        .position(|h| h == "percentiles")
        .expect("percentiles column should exist");
    let percentiles_value = &got[1][percentiles_idx];

    // Quintiles: 20,40,60,80 (4 values)
    // Verify we have 4 percentile values
    let percentile_count = percentiles_value.matches('|').count() + 1;
    assert_eq!(
        percentile_count, 4,
        "Quintiles should produce 4 percentile values"
    );
    // Verify values are from the dataset range
    assert!(
        percentiles_value.contains("4") || percentiles_value.contains("8"),
        "Should contain reasonable values"
    );
}

#[test]
fn stats_percentiles_regular_list_still_works() {
    let wrk = Workdir::new("stats_percentiles_regular_list");
    wrk.create(
        "data.csv",
        vec![
            svec!["n"],
            svec!["1"],
            svec!["2"],
            svec!["3"],
            svec!["4"],
            svec!["5"],
            svec!["6"],
            svec!["7"],
            svec!["8"],
            svec!["9"],
            svec!["10"],
        ],
    );

    // Regular percentile list should still work (not a special value)
    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("25,50,75");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let headers = &got[0];
    let percentiles_idx = headers
        .iter()
        .position(|h| h == "percentiles")
        .expect("percentiles column should exist");
    let percentiles_value = &got[1][percentiles_idx];

    // Regular list should work: 25,50,75 -> 3|5|8
    assert_eq!(percentiles_value, "25: 3|50: 5|75: 8");
}

#[test]
fn stats_percentiles_invalid_list() {
    let wrk = Workdir::new("stats_percentiles_invalid_list");
    wrk.create(
        "data.csv",
        vec![svec!["n"], svec!["1"], svec!["2"], svec!["3"]],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("10,twenty,30,40,50,60,70,80,90,100");
    wrk.assert_err(&mut cmd);
}

#[test]
fn stats_infer_boolean_prefix_pattern() {
    let wrk = Workdir::new("stats_infer_boolean_prefix_pattern");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg("t*:f*")
        .arg(test_file);

    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-10-boolean-tf-stats.csv");

    assert_eq!(dos2unix(&got), dos2unix(&expected).trim_end());
}

#[test]
fn stats_infer_boolean_multiple_patterns() {
    let wrk = Workdir::new("stats_infer_boolean_multiple_patterns");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg("true:false,t*:f*,y*:n*")
        .arg(test_file);

    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-10-boolean-tf-stats.csv");

    assert_eq!(dos2unix(&got), dos2unix(&expected).trim_end());
}

#[test]
fn stats_infer_boolean_case_insensitive() {
    let wrk = Workdir::new("stats_infer_boolean_case_insensitive");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg("TRUE:FALSE")
        .arg(test_file);

    let got: String = wrk.stdout(&mut cmd);

    let expected = wrk.load_test_resource("boston311-10-boolean-tf-stats.csv");

    assert_eq!(dos2unix(&got), dos2unix(&expected).trim_end());
}

#[test]
fn stats_infer_boolean_long_patterns() {
    let wrk = Workdir::new("stats_infer_boolean_long_patterns");

    // Create test data with values that won't match the truthy:falsy pattern
    wrk.create(
        "data.csv",
        vec![
            svec!["col1"],
            svec!["true"],
            svec!["false"],
            svec!["true"],
            svec!["false"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg("truthy:falsy")
        .arg("data.csv");

    let got: String = wrk.stdout(&mut cmd);

    // Should not be inferred as boolean since values don't match pattern
    assert!(!got.contains("Boolean"));
    assert!(got.contains("String"));
}

#[test]
fn stats_infer_boolean_cardinality_three() {
    let wrk = Workdir::new("stats_infer_boolean_cardinality_three");

    // Create a test file with 3 distinct values that would match boolean patterns
    wrk.create(
        "data.csv",
        vec![
            svec!["col1"],
            svec!["true"],
            svec!["truthy"],
            svec!["false"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg("true*:false*")
        .arg("data.csv");

    let got: String = wrk.stdout(&mut cmd);

    // Should not be inferred as boolean because cardinality is 3
    assert!(!got.contains("Boolean"));
    assert!(got.contains("String"));
}

#[test]
fn stats_infer_boolean_empty_pattern() {
    let wrk = Workdir::new("stats_infer_boolean_empty_pattern");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg("")
        .arg(test_file);

    // This should fail with an error
    wrk.assert_err(&mut cmd);
}

#[test]
fn stats_infer_boolean_missing_colon() {
    let wrk = Workdir::new("stats_infer_boolean_missing_colon");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg("truefalse")
        .arg(test_file);

    // This should fail with an error
    wrk.assert_err(&mut cmd);
}

#[test]
fn stats_infer_boolean_missing_true_pattern() {
    let wrk = Workdir::new("stats_infer_boolean_missing_true_pattern");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg(":false")
        .arg(test_file);

    // This should fail with an error
    wrk.assert_err(&mut cmd);
}

#[test]
fn stats_infer_boolean_missing_false_pattern() {
    let wrk = Workdir::new("stats_infer_boolean_missing_false_pattern");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg("true:")
        .arg(test_file);

    // This should fail with an error
    wrk.assert_err(&mut cmd);
}

#[test]
fn stats_infer_boolean_bare_asterisk_rejected() {
    // A bare "*" has an empty prefix, which would match every value.
    let wrk = Workdir::new("stats_infer_boolean_bare_asterisk_rejected");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg("*:0")
        .arg(&test_file);
    wrk.assert_err(&mut cmd);

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg("1:*")
        .arg(&test_file);
    wrk.assert_err(&mut cmd);
}

#[test]
fn stats_infer_boolean_identical_patterns_rejected() {
    let wrk = Workdir::new("stats_infer_boolean_identical_patterns_rejected");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg("y:y")
        .arg(test_file);
    wrk.assert_err(&mut cmd);
}

#[test]
fn stats_infer_boolean_extra_colon_rejected() {
    let wrk = Workdir::new("stats_infer_boolean_extra_colon_rejected");
    let test_file = wrk.load_test_file("boston311-10-boolean-tf.csv");

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--boolean-patterns")
        .arg("true:false:extra")
        .arg(test_file);
    wrk.assert_err(&mut cmd);
}

#[test]
fn stats_cache_invalidates_on_percentile_list_change() {
    // Regression: with --everything, the cache short-circuit must compare
    // flag_percentile_list. Otherwise, switching --percentile-list returns
    // stale percentiles from the cache.
    let wrk = Workdir::new("stats_cache_invalidates_on_percentile_list_change");
    wrk.create(
        "data.csv",
        vec![
            svec!["v"],
            svec!["1"],
            svec!["2"],
            svec!["3"],
            svec!["4"],
            svec!["5"],
            svec!["6"],
            svec!["7"],
            svec!["8"],
            svec!["9"],
            svec!["10"],
        ],
    );

    // First run: cache --everything with default percentile list.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg("--percentiles")
        .arg("--cache-threshold")
        .arg("1")
        .arg("data.csv");
    wrk.assert_success(&mut cmd);

    // Second run: same cache file, but a different --percentile-list.
    // Must NOT reuse the cached row.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("25,75")
        .arg("data.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    // For values 1..=10 (nearest-rank): default list "5,10,40,60,90,95" yields
    // "1|1|4|6|9|10"; the requested "25,75" yields "3|8". A stale cache would
    // return the default list's value. Either way, the result must differ from
    // the default and reflect the two requested percentiles (separator-split
    // gives exactly 2 fields).
    let header = &got[0];
    let pct_idx = header
        .iter()
        .position(|h| h == "percentiles")
        .expect("percentiles column present");
    let pct_value = &got[1][pct_idx];
    let parts: Vec<&str> = pct_value.split('|').collect();
    assert_eq!(
        parts.len(),
        2,
        "expected 2 percentiles for --percentile-list 25,75, got {pct_value:?}"
    );
}

#[test]
fn stats_cache_invalidates_on_weight_change() {
    // Regression: --weight writes to <stem>.stats.weighted.csv regardless of which
    // weight column is used. The --everything cache short-circuit must compare
    // flag_weight or different weight columns will silently share a cache.
    let wrk = Workdir::new("stats_cache_invalidates_on_weight_change");
    wrk.create(
        "data.csv",
        vec![
            svec!["v", "w1", "w2"],
            svec!["1", "1", "10"],
            svec!["2", "1", "1"],
            svec!["3", "1", "1"],
            svec!["4", "1", "1"],
            svec!["5", "1", "1"],
        ],
    );

    // First run: cache with --weight w1 (uniform weights => weighted mean = 3).
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg("--weight")
        .arg("w1")
        .arg("--cache-threshold")
        .arg("1")
        .arg("data.csv");
    wrk.assert_success(&mut cmd);

    // Second run: --weight w2. The first record dominates so the weighted mean
    // must shift toward 1, not stay at the cached 3.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg("--weight")
        .arg("w2")
        .arg("data.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    let header = &got[0];
    let field_idx = header.iter().position(|h| h == "field").unwrap();
    let mean_idx = header.iter().position(|h| h == "mean").unwrap();
    let v_row = got
        .iter()
        .skip(1)
        .find(|r| r[field_idx] == "v")
        .expect("row for column v");
    let mean: f64 = v_row[mean_idx].parse().expect("numeric mean");
    // With weights [10,1,1,1,1] over values [1..=5], weighted mean = 24/14 ≈ 1.714.
    assert!(
        (mean - 24.0 / 14.0).abs() < 1e-3,
        "weighted mean should reflect new weight column, got {mean}"
    );
}

#[test]
fn stats_cache_invalidates_on_select_change() {
    // Regression: with --everything, the cache short-circuit (in `run`) must
    // compare flag_select. Otherwise the cached output (computed for the first
    // run's selection) is served for a different --select.
    let wrk = Workdir::new("stats_cache_invalidates_on_select_change");
    wrk.create(
        "data.csv",
        vec![
            svec!["a", "b", "c", "d"],
            svec!["1", "10", "100", "1000"],
            svec!["2", "20", "200", "2000"],
            svec!["3", "30", "300", "3000"],
        ],
    );

    // First run: cache --everything --select a,b
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .args(["--select", "a,b"])
        .args(["--cache-threshold", "1"])
        .arg("data.csv");
    wrk.assert_success(&mut cmd);

    // Second run: --select c,d. Cache must be invalidated.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .args(["--select", "c,d"])
        .arg("data.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    let header = &got[0];
    let field_idx = header.iter().position(|h| h == "field").unwrap();
    let fields: Vec<&str> = got.iter().skip(1).map(|r| r[field_idx].as_str()).collect();
    assert_eq!(
        fields.len(),
        2,
        "expected 2 rows for select c,d, got {fields:?}"
    );
    assert!(
        fields.contains(&"c") && fields.contains(&"d"),
        "expected stats for columns c,d but got {fields:?}"
    );
}

#[test]
fn stats_cache_invalidates_on_round_change() {
    // Regression: --round affects the formatted values stored in the cache CSV.
    // The --everything cache short-circuit must compare flag_round.
    let wrk = Workdir::new("stats_cache_invalidates_on_round_change");
    wrk.create(
        "data.csv",
        vec![
            svec!["v"],
            svec!["1.123456"],
            svec!["2.234567"],
            svec!["3.345678"],
        ],
    );

    // First run: cache with --round 5 (5 decimal places).
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .args(["--round", "5"])
        .args(["--cache-threshold", "1"])
        .arg("data.csv");
    wrk.assert_success(&mut cmd);

    // Second run: --round 1. Stale cache would still show 5-decimal values.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .args(["--round", "1"])
        .arg("data.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    let header = &got[0];
    let mean_idx = header.iter().position(|h| h == "mean").unwrap();
    let mean = &got[1][mean_idx];
    let decimals = mean.split('.').nth(1).map_or(0, str::len);
    assert!(
        decimals <= 1,
        "expected ≤1 decimal place under --round 1, got mean={mean:?}"
    );
}

#[test]
fn stats_cache_invalidates_on_typesonly_change() {
    // The --everything cache CSV holds the full stats schema; --typesonly
    // expects a much smaller schema. The cache short-circuit must compare
    // flag_typesonly, otherwise the full row leaks through.
    let wrk = Workdir::new("stats_cache_invalidates_on_typesonly_change");
    wrk.create(
        "data.csv",
        vec![svec!["v"], svec!["1"], svec!["2"], svec!["3"]],
    );

    // First run: --everything caches full schema.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .args(["--cache-threshold", "1"])
        .arg("data.csv");
    let everything_cols = wrk.read_stdout::<Vec<Vec<String>>>(&mut cmd)[0].len();
    assert!(everything_cols > 5, "sanity: --everything has many cols");

    // Second run: --typesonly should use the typesonly schema, not the cached row.
    let mut cmd = wrk.command("stats");
    cmd.arg("--typesonly").arg("data.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    assert!(
        got[0].len() < everything_cols,
        "--typesonly must not serve the --everything cache; got {} cols, expected < \
         {everything_cols}",
        got[0].len()
    );
}

#[test]
fn stats_cache_invalidates_on_infer_boolean_change() {
    // --infer-boolean changes the type column. The cache short-circuit must
    // compare flag_infer_boolean.
    let wrk = Workdir::new("stats_cache_invalidates_on_infer_boolean_change");
    wrk.create(
        "data.csv",
        vec![
            svec!["b"],
            svec!["true"],
            svec!["false"],
            svec!["true"],
            svec!["false"],
        ],
    );

    // First run: --everything caches with non-Boolean type for column b.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .args(["--cache-threshold", "1"])
        .arg("data.csv");
    wrk.assert_success(&mut cmd);

    // Second run: --infer-boolean must reclassify b as Boolean.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg("--infer-boolean")
        .arg("data.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    let header = &got[0];
    let type_idx = header.iter().position(|h| h == "type").unwrap();
    assert_eq!(
        got[1][type_idx], "Boolean",
        "--infer-boolean must recompute and report Boolean, got {:?}",
        got[1][type_idx]
    );
}

#[test]
fn stats_cache_invalidates_on_boolean_patterns_change() {
    // --infer-boolean uses --boolean-patterns to decide which 2-cardinality
    // values map to Boolean. Changing the patterns must invalidate the cache.
    let wrk = Workdir::new("stats_cache_invalidates_on_boolean_patterns_change");
    wrk.create(
        "data.csv",
        vec![
            svec!["b"],
            svec!["yes"],
            svec!["no"],
            svec!["yes"],
            svec!["no"],
        ],
    );

    // First run: patterns "true:false" — "yes"/"no" do NOT match, so type is String.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg("--infer-boolean")
        .args(["--boolean-patterns", "true:false"])
        .args(["--cache-threshold", "1"])
        .arg("data.csv");
    wrk.assert_success(&mut cmd);

    // Second run: patterns "yes:no" — must reclassify column b as Boolean.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg("--infer-boolean")
        .args(["--boolean-patterns", "yes:no"])
        .arg("data.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    let header = &got[0];
    let type_idx = header.iter().position(|h| h == "type").unwrap();
    assert_eq!(
        got[1][type_idx], "Boolean",
        "changing --boolean-patterns must invalidate the cache and recompute as Boolean, got {:?}",
        got[1][type_idx]
    );
}

#[test]
fn stats_cache_invalidates_on_vis_whitespace_change() {
    // --vis-whitespace changes how whitespace-bearing min/max strings are
    // serialized in the stats CSV. The cache short-circuit must compare it.
    let wrk = Workdir::new("stats_cache_invalidates_on_vis_whitespace_change");
    wrk.create(
        "data.csv",
        vec![svec!["s"], svec!["a\tb"], svec!["c d"], svec!["e\nf"]],
    );

    // First run: no --vis-whitespace, raw whitespace in min/max.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .args(["--cache-threshold", "1"])
        .arg("data.csv");
    let baseline: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let header = &baseline[0];
    let max_idx = header.iter().position(|h| h == "max").unwrap();
    let baseline_max = baseline[1][max_idx].clone();

    // Second run: --vis-whitespace must produce a different (escaped) max
    // value, which means the cache was not reused.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg("--vis-whitespace")
        .arg("data.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let max_visible = got[1][max_idx].clone();

    assert_ne!(
        baseline_max, max_visible,
        "--vis-whitespace must change the serialized max; cache appears stale \
         (baseline={baseline_max:?}, vis-whitespace={max_visible:?})"
    );
}

#[test]
fn stats_force_recompute() {
    // --force must skip the cache and recompute from the input. We verify by
    // tampering with the cache CSV in place and confirming --force ignores it.
    use std::path::Path;
    let wrk = Workdir::new("stats_force_recompute");
    wrk.create(
        "data.csv",
        vec![svec!["v"], svec!["1"], svec!["2"], svec!["3"]],
    );

    // Build the cache.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .args(["--cache-threshold", "1"])
        .arg("data.csv");
    wrk.assert_success(&mut cmd);

    let cache_csv = wrk.path("data.stats.csv");
    assert!(
        Path::new(&cache_csv).exists(),
        "cache CSV should exist after first run"
    );

    // Tamper: write a wrong-but-well-formed CSV. Without --force this would be served.
    std::fs::write(&cache_csv, "field,type\nv,Tampered\n").unwrap();

    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg("--force")
        .args(["--cache-threshold", "1"])
        .arg("data.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    let header = &got[0];
    let type_idx = header.iter().position(|h| h == "type").unwrap();
    assert_eq!(
        got[1][type_idx], "Integer",
        "--force must recompute; tampered cache leaked through. Got {:?}",
        got[1][type_idx]
    );
}

#[test]
fn stats_cache_corrupt_json_recomputes() {
    // If <stem>.stats.csv.json is unparsable, `run` should silently delete the
    // sidecar files and recompute (it logs a warning). The next run must succeed
    // and rewrite a parseable JSON.
    use std::path::Path;
    let wrk = Workdir::new("stats_cache_corrupt_json_recomputes");
    wrk.create(
        "data.csv",
        vec![svec!["v"], svec!["1"], svec!["2"], svec!["3"]],
    );

    // Build the cache.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .args(["--cache-threshold", "1"])
        .arg("data.csv");
    wrk.assert_success(&mut cmd);

    let cache_json = wrk.path("data.stats.csv.json");
    assert!(Path::new(&cache_json).exists());

    // Corrupt the JSON sidecar.
    std::fs::write(&cache_json, b"{not valid json").unwrap();

    // Second run: must succeed (recover by recomputing), not panic.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .args(["--cache-threshold", "1"])
        .arg("data.csv");
    wrk.assert_success(&mut cmd);

    // Cache JSON should be re-written and parseable.
    let json = std::fs::read_to_string(&cache_json).unwrap();
    assert!(
        json.trim_start().starts_with('{') && json.contains("\"qsv_version\""),
        "cache JSON should be regenerated, got: {json:?}"
    );
}

#[test]
fn stats_parallel_vs_sequential_equivalence() {
    // Lock in `Commute::merge` correctness: deterministic, integer-valued
    // fields must agree across --jobs 1 (sequential) and --jobs 4 (parallel).
    // Floating-point fields (mean/variance) are excluded since their merge
    // order can produce tiny ulp differences after rounding.
    let wrk = Workdir::new("stats_parallel_vs_sequential_equivalence");
    let mut data = vec![svec!["v"]];
    for i in 1..=200i64 {
        data.push(vec![format!("{i}")]);
    }
    wrk.create_indexed("data.csv", data);

    let extract = |rows: &[Vec<String>], cols: &[&str]| -> Vec<String> {
        let header = &rows[0];
        cols.iter()
            .map(|c| {
                let i = header
                    .iter()
                    .position(|h| h == c)
                    .unwrap_or_else(|| panic!("column {c} not present"));
                rows[1][i].clone()
            })
            .collect()
    };

    // Deterministic integer-valued columns common to --everything output.
    let cols = [
        "type",
        "min",
        "max",
        "sum",
        "nullcount",
        "n_negative",
        "n_zero",
        "n_positive",
        "max_precision",
        "cardinality",
    ];

    // Sequential: --jobs 1, --cache-threshold 0 to prevent cache reuse on rerun.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .args(["--jobs", "1"])
        .args(["--cache-threshold", "0"])
        .arg("data.csv");
    let seq: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    // Parallel: --jobs 4 over the same indexed file, also no cache reuse.
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .args(["--jobs", "4"])
        .args(["--cache-threshold", "0"])
        .arg("data.csv");
    let par: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    assert_eq!(
        extract(&seq, &cols),
        extract(&par, &cols),
        "deterministic integer fields must match across parallel/sequential"
    );
}

#[test]
fn stats_cache_threshold_zero_removes_stats_cache() {
    // With --cache-threshold 0, `run` deletes both the stats CSV and the JSON
    // sidecar at the end of the run. We pre-seed an orphan sidecar so the test
    // exercises the active-deletion path, not just the never-written path.
    use std::path::Path;
    let wrk = Workdir::new("stats_cache_threshold_zero_removes_stats_cache");
    wrk.create(
        "data.csv",
        vec![svec!["v"], svec!["1"], svec!["2"], svec!["3"]],
    );

    // Pre-seed an orphan JSON sidecar from a hypothetical prior run.
    std::fs::write(wrk.path("data.stats.csv.json"), b"{}").unwrap();

    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .args(["--cache-threshold", "0"])
        .arg("data.csv");
    wrk.assert_success(&mut cmd);

    assert!(
        !Path::new(&wrk.path("data.stats.csv")).exists(),
        "stats cache CSV must not exist when --cache-threshold 0"
    );
    assert!(
        !Path::new(&wrk.path("data.stats.csv.json")).exists(),
        "stats cache JSON must not exist when --cache-threshold 0 (orphan sidecar must be cleaned \
         up)"
    );
}

#[test]
fn stats_issue_2668_semicolon_separator() {
    let wrk = Workdir::new("stats_issue_2668_semicolon_separator");
    wrk.create("data.csv", vec![svec!["h1;h2;h3;h4"], svec!["1;2;3;4"]]);

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv");

    // should not fail, and just treat the data as a single column
    // as the default delimiter is comma, and the separator is semicolon
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec![
            "field",
            "type",
            "is_ascii",
            "sum",
            "min",
            "max",
            "range",
            "sort_order",
            "sortiness",
            "min_length",
            "max_length",
            "sum_length",
            "avg_length",
            "stddev_length",
            "variance_length",
            "cv_length",
            "mean",
            "sem",
            "geometric_mean",
            "harmonic_mean",
            "stddev",
            "variance",
            "cv",
            "nullcount",
            "n_negative",
            "n_zero",
            "n_positive",
            "max_precision",
            "sparsity"
        ],
        svec![
            "h1;h2;h3;h4",
            "String",
            "true",
            "",
            "1;2;3;4",
            "1;2;3;4",
            "",
            "Unsorted",
            "0",
            "7",
            "7",
            "7",
            "7",
            "0",
            "0",
            "0",
            "",
            "",
            "",
            "",
            "",
            "",
            "",
            "0",
            "",
            "",
            "",
            "",
            "0"
        ],
    ];

    assert_eq!(got, expected);
}

#[test]
fn stats_string_max_length() {
    let wrk = Workdir::new("stats_string_max_length");

    // Create test data with long strings
    wrk.create(
        "data.csv",
        vec![
            svec!["col1", "col2"],
            svec!["short", "very_short"],
            svec!["medium_length_string", "medium_length_string"],
            svec![
                "this_is_a_very_long_string_that_should_be_truncated",
                "another_very_long_string_that_should_be_truncated"
            ],
        ],
    );

    // Run stats with QSV_STATS_STRING_MAX_LENGTH set to 10
    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv").env("QSV_STATS_STRING_MAX_LENGTH", "10");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    // Print the output for debugging
    // println!("Output with QSV_STATS_STRING_MAX_WIDTH=10:");
    // for row in &got {
    //     println!("{:?}", row);
    // }

    // Find the min and max values in the output
    let mut min_value = String::new();
    let mut max_value = String::new();

    // Find the row for col1
    for row in &got {
        if !row.is_empty() && row[0] == "col1" {
            // The min and max values are in columns 4 and 5 (0-indexed)
            if row.len() > 5 {
                min_value = row[4].clone();
                max_value = row[5].clone();
            }
            break;
        }
    }

    // Check that the long string was truncated
    assert_eq!(min_value, "medium_len...");
    assert_eq!(max_value, "this_is_a_...");

    // Run stats with QSV_STATS_STRING_MAX_LENGTH set to 20
    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv").env("QSV_STATS_STRING_MAX_LENGTH", "20");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    // Find the min and max values in the output
    let mut min_value = String::new();
    let mut max_value = String::new();

    // Find the row for col1
    for row in &got {
        if !row.is_empty() && row[0] == "col1" {
            // The min and max values are in columns 4 and 5 (0-indexed)
            if row.len() > 5 {
                min_value = row[4].clone();
                max_value = row[5].clone();
            }
            break;
        }
    }

    // Check that the long string was truncated with a longer width
    assert_eq!(min_value, "medium_length_string");
    assert_eq!(max_value, "this_is_a_very_long_...");

    // Run stats without the environment variable
    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    // Find the min and max values in the output
    let mut min_value = String::new();
    let mut max_value = String::new();

    // Find the row for col1
    for row in &got {
        if !row.is_empty() && row[0] == "col1" {
            // The min and max values are in columns 4 and 5 (0-indexed)
            if row.len() > 5 {
                min_value = row[4].clone();
                max_value = row[5].clone();
            }
            break;
        }
    }

    // Check that the long string was not truncated
    assert_eq!(min_value, "medium_length_string");
    assert_eq!(
        max_value,
        "this_is_a_very_long_string_that_should_be_truncated"
    );
}

#[test]
fn stats_memory_aware_chunking_dynamic() {
    let wrk = Workdir::new("stats_memory_aware_chunking_dynamic");
    let test_file = wrk.load_test_file("boston311-100.csv");

    // Create index to enable parallel processing
    let mut cmd = wrk.command("index");
    cmd.arg(test_file.clone());
    wrk.assert_success(&mut cmd);

    // Run stats with QSV_STATS_CHUNK_MEMORY_MB=0 (dynamic sizing)
    // and --everything to trigger non-streaming stats
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .env("QSV_STATS_CHUNK_MEMORY_MB", "0")
        .arg(test_file);

    // Verify stats computation succeeds and output is correct
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());
    // Verify we have headers
    assert!(got.len() > 1);
}

#[test]
fn stats_memory_aware_chunking_fixed_limit() {
    let wrk = Workdir::new("stats_memory_aware_chunking_fixed_limit");
    let test_file = wrk.load_test_file("boston311-100.csv");

    // Create index to enable parallel processing
    let mut cmd = wrk.command("index");
    cmd.arg(test_file.clone());
    wrk.assert_success(&mut cmd);

    // Run stats with QSV_STATS_CHUNK_MEMORY_MB set to 100MB (fixed limit)
    // and --everything to trigger non-streaming stats
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .env("QSV_STATS_CHUNK_MEMORY_MB", "100")
        .arg(test_file);

    // Verify stats computation succeeds with fixed memory limit
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());
    assert!(got.len() > 1);
}

#[test]
fn stats_memory_aware_chunking_unset_streaming() {
    let wrk = Workdir::new("stats_memory_aware_chunking_unset_streaming");
    let test_file = wrk.load_test_file("boston311-100.csv");

    // Create index to enable parallel processing
    let mut cmd = wrk.command("index");
    cmd.arg(test_file.clone());
    wrk.assert_success(&mut cmd);

    // Run stats without QSV_STATS_CHUNK_MEMORY_MB and without --everything
    // This should use CPU-based chunking (streaming stats only)
    let mut cmd = wrk.command("stats");
    cmd.arg(test_file);

    // Verify stats computation succeeds
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());
    assert!(got.len() > 1);
}

#[test]
fn stats_memory_aware_chunking_unset_non_streaming() {
    let wrk = Workdir::new("stats_memory_aware_chunking_unset_non_streaming");
    let test_file = wrk.load_test_file("boston311-100.csv");

    // Create index to enable parallel processing
    let mut cmd = wrk.command("index");
    cmd.arg(test_file.clone());
    wrk.assert_success(&mut cmd);

    // Run stats without QSV_STATS_CHUNK_MEMORY_MB but with --cardinality
    // This should automatically enable dynamic memory-aware chunking
    let mut cmd = wrk.command("stats");
    cmd.arg("--cardinality").arg(test_file);

    // Verify stats computation succeeds
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());
    assert!(got.len() > 1);
}

#[test]
#[ignore = "Creates a 20gb file which takes a long time to run"] // TODO: fix this test
fn stats_auto_index_creation_on_oom() {
    use std::path::Path;

    let wrk = Workdir::new("stats_auto_index_creation_on_oom");

    // Create a larger CSV file to increase chance of triggering OOM check
    // We'll create a file with many rows to make it large enough
    // Target size: at least 10MB to trigger OOM check on most systems
    let mut data = vec![svec!["col1", "col2", "col3", "col4", "col5"]];
    // Create ~10 million rows with reasonably sized data (~200 bytes per row * 10 million rows = 20
    // GB)
    for i in 0..10_000_000 {
        data.push(vec![
            format!("value_{}_with_some_padding_to_make_it_larger", i),
            format!("another_value_{}_with_more_data", i),
            format!("data_{}", i),
            format!("field_{}_content", i),
            format!("final_field_{}_with_additional_text", i),
        ]);
    }
    let test_file = wrk.path("large_data.csv");
    wrk.create("large_data.csv", data);

    // Verify index does not exist initially
    // Use the same path construction as util::idx_path (appends .idx to full path)
    let index_path = format!("{}.idx", test_file.display());
    let index_file = Path::new(&index_path);
    if index_file.exists() {
        std::fs::remove_file(index_file).unwrap();
    }
    assert!(!index_file.exists(), "Index should not exist initially");

    // Simulate OOM by setting QSV_FREEMEMORY_HEADROOM_PCT very high (90%)
    // and using --memcheck to force conservative mode (stricter check)
    // This will make mem_file_check fail for sequential processing
    // Request non-streaming stats to trigger the OOM check
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg("--memcheck")
        .env("QSV_FREEMEMORY_HEADROOM_PCT", "90")
        .arg(test_file.clone());

    // Verify stats computation succeeds (should auto-create index)
    // and verify stats output is correct
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());
    assert!(got.len() > 1);

    // Verify index file was created
    // Note: This test may be environment-dependent - on systems with very large RAM,
    // the file might still pass the memory check. The important thing is that
    // the auto-index creation logic exists and works when needed.
    assert!(
        index_file.exists(),
        "Index file should be created automatically when mem_file_check fails"
    );
}

// --- OOM auto-fallback to DataSketches estimators ---
//
// The following tests verify the layered OOM-fallback behavior: when
// mem_file_check returns OutOfMemory, stats automatically enables approx
// quantile/cardinality methods (where flag conflicts allow) in addition to
// the existing auto-index fallback. Like stats_auto_index_creation_on_oom
// above, these need a multi-GB file to deterministically trigger OOM, so
// they are #[ignore]'d by default and must be run with `--ignored`.
//
// The shared `build_large_oom_csv` helper lives in `tests/workdir.rs` so the
// row count, column shape, and padding stay in sync with the parallel
// frequency tests; see `frequency_oom_*` in tests/test_frequency.rs.

#[test]
#[ignore = "Requires a multi-GB file to trigger OOM via mem_file_check"]
fn stats_oom_auto_enables_approx_sketches() {
    let (wrk, test_file) =
        crate::workdir::build_large_oom_csv("stats_oom_auto_enables_approx_sketches");
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg("--memcheck")
        .env("QSV_FREEMEMORY_HEADROOM_PCT", "90")
        .arg(test_file);
    let stderr = wrk.stderr_on_success(&mut cmd);
    assert!(
        stderr.contains("auto-enabling DataSketches estimators"),
        "stderr should mention the auto-enable wwarn, got: {stderr}"
    );
    assert!(
        stderr.contains("--quantile-method approx"),
        "stderr should mention quantile auto-enable, got: {stderr}"
    );
    assert!(
        stderr.contains("--cardinality-method approx"),
        "stderr should mention cardinality auto-enable, got: {stderr}"
    );
}

#[test]
#[ignore = "Requires a multi-GB file to trigger OOM via mem_file_check"]
fn stats_oom_skips_approx_quantiles_with_weight() {
    let (wrk, test_file) =
        crate::workdir::build_large_oom_csv("stats_oom_skips_approx_quantiles_with_weight");
    // Weight column is col1 (all distinct strings — qsv will treat parse failures
    // as default weight 1.0; this is acceptable for the test because we only care
    // that --weight is *set*, which is what suppresses the quantile auto-enable).
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg("--memcheck")
        .arg("--weight")
        .arg("col1")
        .env("QSV_FREEMEMORY_HEADROOM_PCT", "90")
        .arg(test_file);
    let stderr = wrk.stderr_on_success(&mut cmd);
    assert!(
        !stderr.contains("--quantile-method approx"),
        "quantile sketch should be suppressed when --weight is set, got: {stderr}"
    );
    assert!(
        stderr.contains("--cardinality-method approx"),
        "cardinality sketch should still auto-enable, got: {stderr}"
    );
}

#[test]
#[ignore = "Requires a multi-GB file to trigger OOM via mem_file_check"]
fn stats_oom_skips_approx_cardinality_with_infer_boolean() {
    let (wrk, test_file) = crate::workdir::build_large_oom_csv(
        "stats_oom_skips_approx_cardinality_with_infer_boolean",
    );
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .arg("--memcheck")
        .arg("--infer-boolean")
        .env("QSV_FREEMEMORY_HEADROOM_PCT", "90")
        .arg(test_file);
    let stderr = wrk.stderr_on_success(&mut cmd);
    assert!(
        !stderr.contains("--cardinality-method approx"),
        "cardinality sketch should be suppressed under --infer-boolean, got: {stderr}"
    );
    assert!(
        stderr.contains("--quantile-method approx"),
        "quantile sketch should still auto-enable, got: {stderr}"
    );
}

#[test]
fn stats_auto_index_creation_skipped_if_indexed() {
    use std::path::Path;

    let wrk = Workdir::new("stats_auto_index_creation_skipped_if_indexed");
    let test_file = wrk.load_test_file("boston311-100.csv");

    // Create index first
    let mut cmd = wrk.command("index");
    cmd.arg(test_file.clone());
    wrk.assert_success(&mut cmd);

    let index_file = Path::new(&test_file).with_extension("csv.idx");
    assert!(index_file.exists(), "Index should exist");

    // Simulate OOM condition
    // With an existing index, parallel processing will be used and mem_file_check is skipped
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .env("QSV_FREEMEMORY_HEADROOM_PCT", "90")
        .arg(test_file);

    // Verify stats computation succeeds (should use existing index, not create new one)
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());
    assert!(got.len() > 1);
}

#[test]
fn stats_auto_index_creation_skipped_for_stdin() {
    use std::io::Write;

    let wrk = Workdir::new("stats_auto_index_creation_skipped_for_stdin");

    // Create a small CSV file
    wrk.create(
        "data.csv",
        vec![
            svec!["col1", "col2"],
            svec!["1", "a"],
            svec!["2", "b"],
            svec!["3", "c"],
        ],
    );

    // Read the file content
    let file_content = wrk.read_to_string("data.csv").unwrap();

    // Run stats with stdin input and non-streaming stats
    let mut cmd = wrk.command("stats");
    cmd.arg("--everything")
        .env("QSV_FREEMEMORY_HEADROOM_PCT", "90")
        .stdin(process::Stdio::piped())
        .stdout(process::Stdio::piped());

    // Write to stdin
    let mut child = cmd.spawn().unwrap();
    {
        let mut stdin = child.stdin.take().unwrap();
        stdin.write_all(file_content.as_bytes()).unwrap();
    }

    // Verify stats computation succeeds (should not attempt to create index for stdin)
    let output = child.wait_with_output().unwrap();
    assert!(
        output.status.success(),
        "Stats should succeed for stdin input"
    );
}

#[test]
fn stats_memory_aware_chunking_empty_samples() {
    let wrk = Workdir::new("stats_memory_aware_chunking_empty_samples");

    // Create a CSV file with very few records (< 1000, so sampling may be incomplete)
    let mut data = vec![svec!["col1", "col2"]];
    for i in 0..10 {
        data.push(vec![format!("{}", i), format!("value{}", i)]);
    }
    wrk.create("data.csv", data);

    // Create index to enable parallel processing
    let mut cmd = wrk.command("index");
    cmd.arg("data.csv");
    wrk.assert_success(&mut cmd);

    // Run stats with QSV_STATS_CHUNK_MEMORY_MB=0 (dynamic) and non-streaming stats
    let mut cmd = wrk.command("stats");
    cmd.arg("--cardinality")
        .env("QSV_STATS_CHUNK_MEMORY_MB", "0")
        .arg("data.csv");

    // Verify stats computation succeeds (should fallback to CPU-based chunking)
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());
    assert!(got.len() > 1);
}

#[test]
fn stats_memory_aware_chunking_small_records() {
    let wrk = Workdir::new("stats_memory_aware_chunking_small_records");

    // Create a CSV file with very small records (single character fields)
    let mut data = vec![svec!["a", "b", "c"]];
    for _ in 0..100 {
        data.push(svec!["x", "y", "z"]);
    }
    wrk.create("data.csv", data);

    // Create index to enable parallel processing
    let mut cmd = wrk.command("index");
    cmd.arg("data.csv");
    wrk.assert_success(&mut cmd);

    // Run stats with QSV_STATS_CHUNK_MEMORY_MB set to 1MB (small fixed value)
    // and non-streaming stats
    let mut cmd = wrk.command("stats");
    cmd.arg("--cardinality")
        .env("QSV_STATS_CHUNK_MEMORY_MB", "1")
        .arg("data.csv");

    // Verify stats computation succeeds (chunk size calculation should handle small records)
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());
    assert!(got.len() > 1);
}

#[test]
fn stats_memory_aware_chunking_no_samples_fallback() {
    let wrk = Workdir::new("stats_memory_aware_chunking_no_samples_fallback");

    // Create a minimal CSV file
    wrk.create(
        "data.csv",
        vec![svec!["col1"], svec!["1"], svec!["2"], svec!["3"]],
    );

    // Create index to enable parallel processing
    let mut cmd = wrk.command("index");
    cmd.arg("data.csv");
    wrk.assert_success(&mut cmd);

    // Run stats with QSV_STATS_CHUNK_MEMORY_MB=0 (dynamic) and non-streaming stats
    // With very few records, sampling may not work, so should fallback to CPU-based chunking
    let mut cmd = wrk.command("stats");
    cmd.arg("--cardinality")
        .env("QSV_STATS_CHUNK_MEMORY_MB", "0")
        .arg("data.csv");

    // Verify stats computation succeeds (should fallback to CPU-based chunking)
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());
    assert!(got.len() > 1);
}

// ============================================================================
// Weighted Statistics Tests
// ============================================================================

#[test]
fn stats_weighted_mean_simple() {
    let wrk = Workdir::new("stats_weighted_mean_simple");
    // Values: [1, 2, 3], Weights: [1, 2, 3]
    // Weighted mean = (1*1 + 2*2 + 3*3) / (1+2+3) = 14/6 = 2.3333...
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["1", "1"],
            svec!["2", "2"],
            svec!["3", "3"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let headers = &got[0];
    let value_row = &got[1];

    // Find mean column
    let mean_idx = headers.iter().position(|h| h == "mean").unwrap();
    let mean_val: f64 = value_row[mean_idx].parse().unwrap();

    // Weighted mean should be approximately 2.3333
    assert!(
        (mean_val - 2.3333333333333335).abs() < 0.0001,
        "Expected weighted mean ~2.333, got {}",
        mean_val
    );
}

#[test]
fn stats_weighted_mean_vs_unweighted() {
    let wrk = Workdir::new("stats_weighted_mean_vs_unweighted");
    // Same values, but with different weights
    // Unweighted mean of [1, 2, 3] = 2.0
    // Weighted mean with weights [1, 1, 10] = (1*1 + 2*1 + 3*10) / 12 = 33/12 = 2.75
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["1", "1"],
            svec!["2", "1"],
            svec!["3", "10"],
        ],
    );

    // Test unweighted
    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv");
    let got_unweighted: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let headers = &got_unweighted[0];
    let value_row = &got_unweighted[1];
    let mean_idx = headers.iter().position(|h| h == "mean").unwrap();
    let unweighted_mean: f64 = value_row[mean_idx].parse().unwrap();
    assert!((unweighted_mean - 2.0).abs() < 0.0001);

    // Test weighted
    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");
    let got_weighted: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let value_row = &got_weighted[1];
    let weighted_mean: f64 = value_row[mean_idx].parse().unwrap();
    assert!(
        (weighted_mean - 2.75).abs() < 0.0001,
        "Expected weighted mean ~2.75, got {}",
        weighted_mean
    );
}

#[test]
fn stats_weighted_stddev() {
    let wrk = Workdir::new("stats_weighted_stddev");
    // Values: [1, 2, 3], Weights: [1, 1, 1] (should match unweighted)
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["1", "1"],
            svec!["2", "1"],
            svec!["3", "1"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");

    // Check command succeeds
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(got.len() > 1, "Should have at least one data row");

    let headers = &got[0];
    let value_row = &got[1];

    let stddev_idx = headers
        .iter()
        .position(|h| h == "stddev")
        .expect("Should have stddev column");
    let stddev_str = &value_row[stddev_idx];

    // Check if stddev is empty or NaN
    assert!(!stddev_str.is_empty(), "Stddev should not be empty");
    let stddev_val: f64 = stddev_str.parse().expect("Stddev should be a valid number");

    // With equal weights [1, 1, 1], weighted variance uses sample variance (n-1 denominator)
    // Variance = ((1-2)^2 + (2-2)^2 + (3-2)^2) / (3-1) = (1 + 0 + 1) / 2 = 1.0
    // Stddev = sqrt(1.0) = 1.0
    assert!(
        (stddev_val - 1.0).abs() < 0.01,
        "Expected stddev ~1.0 (sample variance with equal weights), got {}",
        stddev_val
    );
}

#[test]
fn stats_weighted_median() {
    let wrk = Workdir::new("stats_weighted_median");
    // Values: [1, 2, 3], Weights: [1, 2, 1]
    // Cumulative weights: [1, 3, 4], median at 50% = 2.0
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["1", "1"],
            svec!["2", "2"],
            svec!["3", "1"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight")
        .arg("weight")
        .arg("--median")
        .arg("data.csv");

    // Check command succeeds
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(
        got.len() > 1,
        "Should have at least one data row, got {} rows",
        got.len()
    );

    let headers = &got[0];
    let value_row = &got[1];

    let median_idx = headers
        .iter()
        .position(|h| h == "median")
        .expect("Should have median column");
    let median_val: f64 = value_row[median_idx]
        .parse()
        .expect("Median should be a valid number");

    // Weighted median should be 2.0
    assert!(
        (median_val - 2.0).abs() < 0.0001,
        "Expected weighted median 2.0, got {}",
        median_val
    );
}

#[test]
fn stats_weighted_quartiles() {
    let wrk = Workdir::new("stats_weighted_quartiles");
    // Values: [1, 2, 3, 4, 5], Weights: [1, 1, 2, 1, 1]
    // Total weight = 6, Q1 at 25% = 1.5, Q2 (median) at 50% = 3.0, Q3 at 75% = 3.5
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["1", "1"],
            svec!["2", "1"],
            svec!["3", "2"],
            svec!["4", "1"],
            svec!["5", "1"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight")
        .arg("weight")
        .arg("--quartiles")
        .arg("data.csv");

    // Check command succeeds
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(
        got.len() > 1,
        "Should have at least one data row, got {} rows",
        got.len()
    );

    let headers = &got[0];
    let value_row = &got[1];

    let q1_idx = headers
        .iter()
        .position(|h| h == "q1")
        .expect("Should have q1 column");
    let q2_idx = headers
        .iter()
        .position(|h| h == "q2_median")
        .expect("Should have q2_median column");
    let q3_idx = headers
        .iter()
        .position(|h| h == "q3")
        .expect("Should have q3 column");

    let q1: f64 = value_row[q1_idx]
        .parse()
        .expect("Q1 should be a valid number");
    let q2: f64 = value_row[q2_idx]
        .parse()
        .expect("Q2 should be a valid number");
    let q3: f64 = value_row[q3_idx]
        .parse()
        .expect("Q3 should be a valid number");

    // Q2 (median) should be 3.0
    assert!((q2 - 3.0).abs() < 0.1, "Expected Q2 ~3.0, got {}", q2);
    // Q1 and Q3 should be reasonable values
    assert!(
        q1 < q2 && q2 < q3,
        "Quartiles should be ordered: Q1={}, Q2={}, Q3={}",
        q1,
        q2,
        q3
    );
}

#[test]
fn stats_weighted_mad() {
    let wrk = Workdir::new("stats_weighted_mad");
    // Values: [1, 2, 3], Weights: [1, 2, 1]
    // Weighted median = 2.0
    // Absolute deviations: [|1-2|, |2-2|, |3-2|] = [1, 0, 1] with weights [1, 2, 1]
    // Weighted MAD = median of [1, 0, 1] with weights [1, 2, 1] = 0.0 (since 0 has weight 2)
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["1", "1"],
            svec!["2", "2"],
            svec!["3", "1"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight")
        .arg("weight")
        .arg("--mad")
        .arg("data.csv");

    // Check command succeeds
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(
        got.len() > 1,
        "Should have at least one data row, got {} rows",
        got.len()
    );

    let headers = &got[0];
    let value_row = &got[1];

    let mad_idx = headers
        .iter()
        .position(|h| h == "mad")
        .expect("Should have mad column");
    let mad_val: f64 = value_row[mad_idx]
        .parse()
        .expect("MAD should be a valid number");

    // Weighted MAD should be 0.0 (median of absolute deviations [1, 0, 1] with weights [1, 2, 1])
    // Cumulative weights for deviations: [1, 3, 4], median at 50% = 2.0, which corresponds to 0
    assert!(
        (mad_val - 0.0).abs() < 0.0001,
        "Expected weighted MAD ~0.0, got {}",
        mad_val
    );
}

#[test]
fn stats_weighted_percentiles() {
    let wrk = Workdir::new("stats_weighted_percentiles");
    // Values: [1, 2, 3, 4, 5], Weights: [1, 1, 2, 1, 1]
    // Total weight = 6
    // 10th percentile: 0.1 * 6 = 0.6 -> value 1 (cumulative weight reaches 1 at value 1)
    // 50th percentile (median): 0.5 * 6 = 3.0 -> value 3 (cumulative weight reaches 4 at value 3)
    // 90th percentile: 0.9 * 6 = 5.4 -> value 5 (cumulative weight reaches 6 at value 5)
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["1", "1"],
            svec!["2", "1"],
            svec!["3", "2"],
            svec!["4", "1"],
            svec!["5", "1"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight")
        .arg("weight")
        .arg("--percentiles")
        .arg("--percentile-list")
        .arg("10,50,90")
        .arg("data.csv");

    // Check command succeeds
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(
        got.len() > 1,
        "Should have at least one data row, got {} rows",
        got.len()
    );

    let headers = &got[0];
    let value_row = &got[1];

    let percentiles_idx = headers
        .iter()
        .position(|h| h == "percentiles")
        .expect("Should have percentiles column");
    let percentiles_str = &value_row[percentiles_idx];

    // Parse percentiles string (format: "10: 1.0|50: 3.0|90: 5.0")
    let percentile_parts: Vec<&str> = percentiles_str.split('|').collect();
    assert_eq!(
        percentile_parts.len(),
        3,
        "Should have 3 percentiles, got {}",
        percentile_parts.len()
    );

    // Extract values from "10: 1.0" format
    let p10_val: f64 = percentile_parts[0]
        .split(':')
        .nth(1)
        .expect("Should have value after colon")
        .trim()
        .parse()
        .expect("P10 should be a valid number");
    let p50_val: f64 = percentile_parts[1]
        .split(':')
        .nth(1)
        .expect("Should have value after colon")
        .trim()
        .parse()
        .expect("P50 should be a valid number");
    let p90_val: f64 = percentile_parts[2]
        .split(':')
        .nth(1)
        .expect("Should have value after colon")
        .trim()
        .parse()
        .expect("P90 should be a valid number");

    // Verify percentile values
    assert!(
        (p10_val - 1.0).abs() < 0.1,
        "Expected 10th percentile ~1.0, got {}",
        p10_val
    );
    assert!(
        (p50_val - 3.0).abs() < 0.1,
        "Expected 50th percentile (median) ~3.0, got {}",
        p50_val
    );
    assert!(
        (p90_val - 5.0).abs() < 0.1,
        "Expected 90th percentile ~5.0, got {}",
        p90_val
    );
    // Verify ordering
    assert!(
        p10_val <= p50_val && p50_val <= p90_val,
        "Percentiles should be ordered: P10={}, P50={}, P90={}",
        p10_val,
        p50_val,
        p90_val
    );
}

#[test]
fn stats_weighted_geometric_mean() {
    let wrk = Workdir::new("stats_weighted_geometric_mean");
    // Values: [2, 8, 32], Weights: [1, 1, 1]
    // Weighted geometric mean = exp((1*ln(2) + 1*ln(8) + 1*ln(32)) / 3)
    // = exp((ln(2) + ln(8) + ln(32)) / 3) = exp(ln(2*8*32) / 3) = exp(ln(512) / 3)
    // = exp(ln(512^(1/3))) = 512^(1/3) = 8.0
    // With equal weights, should match unweighted geometric mean
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["2", "1"],
            svec!["8", "1"],
            svec!["32", "1"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");

    // Check command succeeds
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(got.len() > 1, "Should have at least one data row");

    let headers = &got[0];
    let value_row = &got[1];

    let geom_mean_idx = headers
        .iter()
        .position(|h| h == "geometric_mean")
        .expect("Should have geometric_mean column");
    let geom_mean_str = &value_row[geom_mean_idx];

    // Check if geometric_mean is empty or NaN
    assert!(
        !geom_mean_str.is_empty(),
        "Geometric mean should not be empty"
    );
    let geom_mean_val: f64 = geom_mean_str
        .parse()
        .expect("Geometric mean should be a valid number");

    // Expected: (2 * 8 * 32)^(1/3) = 512^(1/3) ≈ 8.0
    assert!(
        (geom_mean_val - 8.0).abs() < 0.01,
        "Expected geometric mean ~8.0, got {}",
        geom_mean_val
    );
}

#[test]
fn stats_weighted_geometric_mean_unequal_weights() {
    let wrk = Workdir::new("stats_weighted_geometric_mean_unequal_weights");
    // Values: [2, 8], Weights: [1, 3]
    // Weighted geometric mean = exp((1*ln(2) + 3*ln(8)) / 4)
    // = exp((ln(2) + 3*ln(8)) / 4) = exp((ln(2) + ln(8^3)) / 4)
    // = exp(ln(2 * 512) / 4) = exp(ln(1024) / 4) = 1024^(1/4) ≈ 5.66
    wrk.create(
        "data.csv",
        vec![svec!["value", "weight"], svec!["2", "1"], svec!["8", "3"]],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");

    // Check command succeeds
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(got.len() > 1, "Should have at least one data row");

    let headers = &got[0];
    let value_row = &got[1];

    let geom_mean_idx = headers
        .iter()
        .position(|h| h == "geometric_mean")
        .expect("Should have geometric_mean column");
    let geom_mean_str = &value_row[geom_mean_idx];

    assert!(
        !geom_mean_str.is_empty(),
        "Geometric mean should not be empty"
    );
    let geom_mean_val: f64 = geom_mean_str
        .parse()
        .expect("Geometric mean should be a valid number");

    // Expected: exp((ln(2) + 3*ln(8)) / 4) = exp(ln(2 * 512) / 4) = 1024^(1/4) ≈ 5.66
    let expected = (2.0_f64 * 8.0_f64.powi(3)).powf(1.0 / 4.0);
    assert!(
        (geom_mean_val - expected).abs() < 0.01,
        "Expected geometric mean ~{}, got {}",
        expected,
        geom_mean_val
    );
}

#[test]
fn stats_weighted_harmonic_mean() {
    let wrk = Workdir::new("stats_weighted_harmonic_mean");
    // Values: [2, 4, 8], Weights: [1, 1, 1]
    // Weighted harmonic mean = 3 / (1/2 + 1/4 + 1/8) = 3 / (4/8 + 2/8 + 1/8)
    // = 3 / (7/8) = 3 * 8/7 = 24/7 ≈ 3.4286
    // With equal weights, should match unweighted harmonic mean
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["2", "1"],
            svec!["4", "1"],
            svec!["8", "1"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");

    // Check command succeeds
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(got.len() > 1, "Should have at least one data row");

    let headers = &got[0];
    let value_row = &got[1];

    let harm_mean_idx = headers
        .iter()
        .position(|h| h == "harmonic_mean")
        .expect("Should have harmonic_mean column");
    let harm_mean_str = &value_row[harm_mean_idx];

    // Check if harmonic_mean is empty or NaN
    assert!(
        !harm_mean_str.is_empty(),
        "Harmonic mean should not be empty"
    );
    let harm_mean_val: f64 = harm_mean_str
        .parse()
        .expect("Harmonic mean should be a valid number");

    // Expected: 3 / (1/2 + 1/4 + 1/8) = 3 / (7/8) = 24/7 ≈ 3.4286
    let expected = 3.0 / (1.0 / 2.0 + 1.0 / 4.0 + 1.0 / 8.0);
    assert!(
        (harm_mean_val - expected).abs() < 0.01,
        "Expected harmonic mean ~{}, got {}",
        expected,
        harm_mean_val
    );
}

#[test]
fn stats_weighted_harmonic_mean_unequal_weights() {
    let wrk = Workdir::new("stats_weighted_harmonic_mean_unequal_weights");
    // Values: [2, 8], Weights: [1, 3]
    // Weighted harmonic mean = (1 + 3) / (1/2 + 3/8) = 4 / (4/8 + 3/8)
    // = 4 / (7/8) = 4 * 8/7 = 32/7 ≈ 4.5714
    wrk.create(
        "data.csv",
        vec![svec!["value", "weight"], svec!["2", "1"], svec!["8", "3"]],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");

    // Check command succeeds
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(got.len() > 1, "Should have at least one data row");

    let headers = &got[0];
    let value_row = &got[1];

    let harm_mean_idx = headers
        .iter()
        .position(|h| h == "harmonic_mean")
        .expect("Should have harmonic_mean column");
    let harm_mean_str = &value_row[harm_mean_idx];

    assert!(
        !harm_mean_str.is_empty(),
        "Harmonic mean should not be empty"
    );
    let harm_mean_val: f64 = harm_mean_str
        .parse()
        .expect("Harmonic mean should be a valid number");

    // Expected: (1 + 3) / (1/2 + 3/8) = 4 / (7/8) = 32/7 ≈ 4.5714
    let expected = (1.0 + 3.0) / (1.0 / 2.0 + 3.0 / 8.0);
    assert!(
        (harm_mean_val - expected).abs() < 0.01,
        "Expected harmonic mean ~{}, got {}",
        expected,
        harm_mean_val
    );
}

#[test]
fn stats_weighted_geometric_mean_zero_or_negative() {
    let wrk = Workdir::new("stats_weighted_geometric_mean_zero_or_negative");
    // Geometric mean requires positive values
    // Test with zero and negative values - should handle gracefully
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["-1", "1"],
            svec!["0", "1"],
            svec!["2", "1"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");

    // Check command succeeds
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(got.len() > 1, "Should have at least one data row");

    let headers = &got[0];
    let value_row = &got[1];

    let geom_mean_idx = headers
        .iter()
        .position(|h| h == "geometric_mean")
        .expect("Should have geometric_mean column");
    let geom_mean_str = &value_row[geom_mean_idx];

    // Geometric mean should only consider positive values (2)
    // So it should be exp(ln(2) / 1) = 2.0
    // But if negative/zero values cause issues, it might be NaN or empty
    if !geom_mean_str.is_empty() && geom_mean_str != "NaN" {
        let geom_mean_val: f64 = geom_mean_str.parse().unwrap_or(f64::NAN);
        // Should only consider positive value (2), so result should be 2.0
        assert!(
            (geom_mean_val - 2.0).abs() < 0.01 || geom_mean_val.is_nan(),
            "Geometric mean should be ~2.0 (only positive values) or NaN, got {}",
            geom_mean_val
        );
    }
}

#[test]
fn stats_weighted_harmonic_mean_zero() {
    let wrk = Workdir::new("stats_weighted_harmonic_mean_zero");
    // Harmonic mean requires non-zero values
    // Test with zero values - should handle gracefully
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["0", "1"],
            svec!["2", "1"],
            svec!["4", "1"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");

    // Check command succeeds
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(got.len() > 1, "Should have at least one data row");

    let headers = &got[0];
    let value_row = &got[1];

    let harm_mean_idx = headers
        .iter()
        .position(|h| h == "harmonic_mean")
        .expect("Should have harmonic_mean column");
    let harm_mean_str = &value_row[harm_mean_idx];

    // Harmonic mean should only consider non-zero values (2, 4)
    // So it should be 2 / (1/2 + 1/4) = 2 / (3/4) = 8/3 ≈ 2.667
    // But if zero values cause issues, it might be NaN or empty
    if !harm_mean_str.is_empty() && harm_mean_str != "NaN" {
        let harm_mean_val: f64 = harm_mean_str.parse().unwrap_or(f64::NAN);
        let expected = 2.0 / (1.0 / 2.0 + 1.0 / 4.0);
        assert!(
            (harm_mean_val - expected).abs() < 0.01 || harm_mean_val.is_nan(),
            "Harmonic mean should be ~{} (only non-zero values) or NaN, got {}",
            expected,
            harm_mean_val
        );
    }
}

#[test]
fn stats_weighted_modes() {
    let wrk = Workdir::new("stats_weighted_modes");
    // Values: ["a", "b", "a", "c", "b"], Weights: [1, 2, 3, 1, 1]
    // Weighted frequencies:
    //   "a": 1 + 3 = 4 (max weight - mode)
    //   "b": 2 + 1 = 3
    //   "c": 1 (min weight - antimode)
    // Mode should be "a" with weight 4
    // Antimode should be "c" with weight 1
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["a", "1"],
            svec!["b", "2"],
            svec!["a", "3"],
            svec!["c", "1"],
            svec!["b", "1"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight")
        .arg("weight")
        .arg("--mode")
        .arg("data.csv");

    // Check command succeeds
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(
        got.len() > 1,
        "Should have at least one data row, got {} rows",
        got.len()
    );

    let headers = &got[0];
    let value_row = &got[1];

    let mode_idx = headers
        .iter()
        .position(|h| h == "mode")
        .expect("Should have mode column");
    let mode_count_idx = headers
        .iter()
        .position(|h| h == "mode_count")
        .expect("Should have mode_count column");
    let mode_occurrences_idx = headers
        .iter()
        .position(|h| h == "mode_occurrences")
        .expect("Should have mode_occurrences column");
    let antimode_idx = headers
        .iter()
        .position(|h| h == "antimode")
        .expect("Should have antimode column");
    let antimode_count_idx = headers
        .iter()
        .position(|h| h == "antimode_count")
        .expect("Should have antimode_count column");
    let antimode_occurrences_idx = headers
        .iter()
        .position(|h| h == "antimode_occurrences")
        .expect("Should have antimode_occurrences column");

    let mode_str = &value_row[mode_idx];
    let mode_count: usize = value_row[mode_count_idx]
        .parse()
        .expect("mode_count should be a valid number");
    let mode_occurrences: u32 = value_row[mode_occurrences_idx]
        .parse()
        .expect("mode_occurrences should be a valid number");
    let antimode_str = &value_row[antimode_idx];
    let antimode_count: usize = value_row[antimode_count_idx]
        .parse()
        .expect("antimode_count should be a valid number");
    let antimode_occurrences: u32 = value_row[antimode_occurrences_idx]
        .parse()
        .expect("antimode_occurrences should be a valid number");

    // Mode should be "a" with weight 4
    assert_eq!(mode_str, "a", "Expected mode 'a', got '{}'", mode_str);
    assert_eq!(mode_count, 1, "Expected mode_count 1, got {}", mode_count);
    assert_eq!(
        mode_occurrences, 4,
        "Expected mode_occurrences 4 (weight), got {}",
        mode_occurrences
    );

    // Antimode should be "c" with weight 1
    assert_eq!(
        antimode_str, "c",
        "Expected antimode 'c', got '{}'",
        antimode_str
    );
    assert_eq!(
        antimode_count, 1,
        "Expected antimode_count 1, got {}",
        antimode_count
    );
    assert_eq!(
        antimode_occurrences, 1,
        "Expected antimode_occurrences 1 (weight), got {}",
        antimode_occurrences
    );
}

#[test]
fn stats_weighted_modes_multiple() {
    let wrk = Workdir::new("stats_weighted_modes_multiple");
    // Values: ["x", "y", "x", "z", "y"], Weights: [2, 3, 2, 1, 3]
    // Weighted frequencies:
    //   "x": 2 + 2 = 4 (tied for max weight - mode)
    //   "y": 3 + 3 = 6 (max weight - mode)
    //   "z": 1 (min weight - antimode)
    // Modes should be ["x", "y"] or ["y", "x"] with weight 6 (or 4)
    // Antimode should be "z" with weight 1
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["x", "2"],
            svec!["y", "3"],
            svec!["x", "2"],
            svec!["z", "1"],
            svec!["y", "3"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight")
        .arg("weight")
        .arg("--mode")
        .arg("data.csv");

    // Check command succeeds
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(
        got.len() > 1,
        "Should have at least one data row, got {} rows",
        got.len()
    );

    let headers = &got[0];
    let value_row = &got[1];

    let mode_idx = headers
        .iter()
        .position(|h| h == "mode")
        .expect("Should have mode column");
    let mode_count_idx = headers
        .iter()
        .position(|h| h == "mode_count")
        .expect("Should have mode_count column");
    let antimode_idx = headers
        .iter()
        .position(|h| h == "antimode")
        .expect("Should have antimode column");

    let mode_str = &value_row[mode_idx];
    let mode_count: usize = value_row[mode_count_idx]
        .parse()
        .expect("mode_count should be a valid number");
    let antimode_str = &value_row[antimode_idx];

    // Mode should be "y" (or "x" and "y" if both have max weight)
    // Since "y" has weight 6 and "x" has weight 4, "y" should be the mode
    // But if there's a tie, both could be modes
    assert!(
        mode_str.contains("y"),
        "Expected mode to contain 'y', got '{}'",
        mode_str
    );
    // Mode count should be at least 1
    assert!(
        mode_count >= 1,
        "Expected mode_count >= 1, got {}",
        mode_count
    );

    // Antimode should be "z"
    assert_eq!(
        antimode_str, "z",
        "Expected antimode 'z', got '{}'",
        antimode_str
    );
}

#[test]
fn stats_weighted_modes_cardinality() {
    let wrk = Workdir::new("stats_weighted_modes_cardinality");
    // Test that cardinality works with weighted modes
    // Values: ["a", "b", "c"], Weights: [1, 2, 3]
    // Cardinality should be 3 (3 unique values)
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["a", "1"],
            svec!["b", "2"],
            svec!["c", "3"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight")
        .arg("weight")
        .arg("--cardinality")
        .arg("--mode")
        .arg("data.csv");

    // Check command succeeds
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(
        got.len() > 1,
        "Should have at least one data row, got {} rows",
        got.len()
    );

    let headers = &got[0];
    let value_row = &got[1];

    let cardinality_idx = headers
        .iter()
        .position(|h| h == "cardinality")
        .expect("Should have cardinality column");
    let cardinality: u64 = value_row[cardinality_idx]
        .parse()
        .expect("cardinality should be a valid number");

    // Cardinality should be 3 (3 unique values)
    assert_eq!(
        cardinality, 3,
        "Expected cardinality 3, got {}",
        cardinality
    );
}

#[test]
fn stats_weighted_missing_weight_column() {
    let wrk = Workdir::new("stats_weighted_missing_weight_column");
    wrk.create("data.csv", vec![svec!["value"], svec!["1"], svec!["2"]]);

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("nonexistent").arg("data.csv");

    let output = cmd.output().unwrap();
    assert!(
        !output.status.success(),
        "Should fail when weight column doesn't exist"
    );
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        stderr.contains("not found"),
        "Error should mention column not found"
    );
}

#[test]
fn stats_weighted_invalid_weights_default_to_one() {
    let wrk = Workdir::new("stats_weighted_invalid_weights_default_to_one");
    // Invalid/empty weights default to 1.0, but valid weights are used with their actual values.
    // With weights [1.0 (defaulted from "invalid"), 2 (valid), 1.0 (defaulted from "")],
    // weighted mean = (1*1.0 + 2*2 + 3*1.0) / (1.0 + 2 + 1.0) = 8/4 = 2.0
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["1", "invalid"],
            svec!["2", "2"],
            svec!["3", ""], // empty weight
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let headers = &got[0];
    let value_row = &got[1];

    let mean_idx = headers.iter().position(|h| h == "mean").unwrap();
    let mean_val: f64 = value_row[mean_idx].parse().unwrap();

    assert!(
        (mean_val - 2.0).abs() < 0.01,
        "Expected mean ~2.0 with weights [1.0, 2, 1.0], got {}",
        mean_val
    );
}

#[test]
fn stats_weighted_excludes_weight_column() {
    let wrk = Workdir::new("stats_weighted_excludes_weight_column");
    wrk.create(
        "data.csv",
        vec![svec!["value", "weight"], svec!["1", "1"], svec!["2", "2"]],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");

    // Check command succeeds
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(got.len() > 1, "Should have at least one data row");

    let headers = &got[0];

    // Weight column should not appear in stats output headers (which are stat column names)
    assert!(
        !headers.contains(&"weight".to_string()),
        "Weight column should be excluded from stats"
    );

    // Check the "field" column in data rows to ensure "value" is present and "weight" is not
    let field_idx = headers
        .iter()
        .position(|h| h == "field")
        .expect("Should have 'field' column");
    let mut found_value = false;
    for row in got.iter().skip(1) {
        if let Some(field_val) = row.get(field_idx) {
            assert_ne!(
                field_val, "weight",
                "Weight column should not appear in stats"
            );
            if field_val == "value" {
                found_value = true;
            }
        }
    }
    assert!(found_value, "Value column should be in stats");
}

#[test]
fn stats_weighted_parallel_processing() {
    let wrk = Workdir::new("stats_weighted_parallel_processing");

    // Create a larger dataset to test parallel processing
    let mut data = vec![svec!["value", "weight"]];
    for i in 1..=100 {
        data.push(vec![format!("{}", i), format!("{}", i % 10 + 1)]);
    }
    wrk.create("data.csv", data);

    // Create index to enable parallel processing
    let mut cmd = wrk.command("index");
    cmd.arg("data.csv");
    wrk.assert_success(&mut cmd);

    // Run weighted stats with parallel processing
    let mut cmd = wrk.command("stats");
    cmd.arg("--weight")
        .arg("weight")
        .arg("--everything")
        .arg("data.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty());

    let headers = &got[0];
    let value_row = &got[1];

    // Verify weighted statistics are computed
    let mean_idx = headers.iter().position(|h| h == "mean").unwrap();
    let mean_val: f64 = value_row[mean_idx].parse().unwrap();
    assert!(mean_val > 0.0 && mean_val < 100.0);
}

#[test]
fn stats_weighted_cache_filename() {
    let wrk = Workdir::new("stats_weighted_cache_filename");
    wrk.create(
        "data.csv",
        vec![svec!["value", "weight"], svec!["1", "1"], svec!["2", "2"]],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");
    wrk.assert_success(&mut cmd);

    // Check that weighted cache file is created
    assert!(
        wrk.path("data.stats.weighted.csv").exists(),
        "Weighted stats cache file should exist"
    );
    assert!(
        !wrk.path("data.stats.csv").exists(),
        "Unweighted stats cache file should not exist when using --weight"
    );
}

#[test]
fn stats_weighted_vs_unweighted_cache_separation() {
    let wrk = Workdir::new("stats_weighted_vs_unweighted_cache_separation");
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["1", "1"],
            svec!["2", "2"],
            svec!["3", "3"],
        ],
    );

    // Run unweighted stats
    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv");
    wrk.assert_success(&mut cmd);
    assert!(wrk.path("data.stats.csv").exists());

    // Run weighted stats
    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");
    wrk.assert_success(&mut cmd);
    assert!(wrk.path("data.stats.weighted.csv").exists());

    // Both cache files should exist
    assert!(wrk.path("data.stats.csv").exists());
    assert!(wrk.path("data.stats.weighted.csv").exists());
}

#[test]
fn stats_weighted_all_zero_weights() {
    let wrk = Workdir::new("stats_weighted_all_zero_weights");
    // All weights are zero - should handle gracefully
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["1", "0"],
            svec!["2", "0"],
            svec!["3", "0"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");

    // Command should succeed even with all zero weights
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");
    assert!(
        got.len() > 1,
        "Should have at least one data row, got {} rows",
        got.len()
    );

    let headers = &got[0];
    let value_row = &got[1];

    // With all zero weights, mean/stddev should be empty or zero
    let mean_idx = headers.iter().position(|h| h == "mean");
    if let Some(idx) = mean_idx {
        let mean_str = &value_row[idx];
        // Mean should be empty or zero when all weights are zero
        assert!(
            mean_str.is_empty() || mean_str == "0",
            "Mean should be empty or zero with all zero weights, got '{}'",
            mean_str
        );
    }
}

#[test]
fn stats_weighted_negative_weights() {
    let wrk = Workdir::new("stats_weighted_negative_weights");
    // Negative weights should be ignored
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["1", "-1"], // negative weight - should be ignored
            svec!["2", "2"],  // positive weight - should be used
            svec!["3", "-5"], // negative weight - should be ignored
            svec!["4", "1"],  // positive weight - should be used
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let headers = &got[0];
    let value_row = &got[1];

    let mean_idx = headers.iter().position(|h| h == "mean").unwrap();
    let mean_val: f64 = value_row[mean_idx].parse().unwrap();

    // Only values 2 and 4 should contribute (weights 2 and 1)
    // Weighted mean = (2*2 + 4*1) / (2+1) = 8/3 ≈ 2.67
    assert!(
        (mean_val - 2.67).abs() < 0.1,
        "Expected weighted mean ~2.67 (only positive weights), got {}",
        mean_val
    );
}

#[test]
fn stats_weighted_mixed_zero_and_nonzero_weights() {
    let wrk = Workdir::new("stats_weighted_mixed_zero_and_nonzero_weights");
    // Mix of zero and non-zero weights - only non-zero should contribute
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["10", "0"], // zero weight - should be ignored
            svec!["20", "2"], // non-zero weight - should be used
            svec!["30", "0"], // zero weight - should be ignored
            svec!["40", "3"], // non-zero weight - should be used
            svec!["50", "0"], // zero weight - should be ignored
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let headers = &got[0];
    let value_row = &got[1];

    let mean_idx = headers.iter().position(|h| h == "mean").unwrap();
    let mean_val: f64 = value_row[mean_idx].parse().unwrap();

    // Only values 20 and 40 should contribute (weights 2 and 3)
    // Weighted mean = (20*2 + 40*3) / (2+3) = 160/5 = 32.0
    assert!(
        (mean_val - 32.0).abs() < 0.1,
        "Expected weighted mean ~32.0 (only non-zero weights), got {}",
        mean_val
    );
}

#[test]
fn stats_weighted_nan_weights() {
    let wrk = Workdir::new("stats_weighted_nan_weights");
    // NaN weights should be handled (likely defaulted to 1.0 or ignored)
    // Note: CSV parsing might convert "NaN" to a string, so we test with invalid numeric strings
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["1", "invalid"],    // invalid weight - should default to 1.0
            svec!["2", "2"],          // valid weight
            svec!["3", "notanumber"], // invalid weight - should default to 1.0
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");

    // Command should succeed
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");

    let headers = &got[0];
    let value_row = &got[1];

    let mean_idx = headers.iter().position(|h| h == "mean").unwrap();
    let mean_val: f64 = value_row[mean_idx].parse().unwrap();

    // Invalid weights default to 1.0, so weights are effectively [1, 2, 1]
    // Weighted mean = (1*1 + 2*2 + 3*1) / (1+2+1) = 8/4 = 2.0
    assert!(
        (mean_val - 2.0).abs() < 0.1,
        "Expected weighted mean ~2.0 (invalid weights default to 1.0), got {}",
        mean_val
    );
}

#[test]
fn stats_weighted_infinity_weights() {
    let wrk = Workdir::new("stats_weighted_infinity_weights");
    // Very large weights that might cause overflow
    // Using scientific notation for very large numbers
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["1", "1e100"], // very large weight
            svec!["2", "1e100"], // very large weight
            svec!["3", "1"],     // normal weight
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight").arg("weight").arg("data.csv");

    // Command should succeed without overflow
    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");

    let headers = &got[0];
    let value_row = &got[1];

    let mean_idx = headers.iter().position(|h| h == "mean").unwrap();
    let mean_str = &value_row[mean_idx];

    // With very large weights, the mean should be dominated by values 1 and 2
    // Weighted mean ≈ (1*1e100 + 2*1e100 + 3*1) / (1e100 + 1e100 + 1) ≈ 1.5
    // But due to floating point precision, it might be exactly 1.5 or close to it
    if !mean_str.is_empty() {
        let mean_val: f64 = mean_str.parse().unwrap();
        assert!(
            (1.0..=2.0).contains(&mean_val),
            "Mean should be between 1.0 and 2.0 with very large weights, got {}",
            mean_val
        );
    }
}

#[test]
fn stats_weighted_zero_weights_with_modes() {
    let wrk = Workdir::new("stats_weighted_zero_weights_with_modes");
    // Test modes/antimodes with zero weights
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["a", "0"], // zero weight
            svec!["b", "2"], // non-zero weight
            svec!["a", "0"], // zero weight
            svec!["c", "1"], // non-zero weight
            svec!["b", "0"], // zero weight
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight")
        .arg("weight")
        .arg("--mode")
        .arg("data.csv");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let headers = &got[0];
    let value_row = &got[1];

    let mode_idx = headers.iter().position(|h| h == "mode").unwrap();
    let mode_str = &value_row[mode_idx];

    // Only "b" (weight 2) and "c" (weight 1) should contribute
    // Mode should be "b" with weight 2
    assert_eq!(
        mode_str, "b",
        "Expected mode 'b' (weight 2), got '{}'",
        mode_str
    );
}

#[test]
fn stats_weighted_negative_weights_with_quartiles() {
    let wrk = Workdir::new("stats_weighted_negative_weights_with_quartiles");
    // Test quartiles with negative weights (should be ignored)
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["1", "-1"], // negative weight - ignored
            svec!["2", "2"],  // positive weight
            svec!["3", "-5"], // negative weight - ignored
            svec!["4", "3"],  // positive weight
            svec!["5", "1"],  // positive weight
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight")
        .arg("weight")
        .arg("--quartiles")
        .arg("data.csv");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let headers = &got[0];
    let value_row = &got[1];

    let q2_idx = headers.iter().position(|h| h == "q2_median").unwrap();
    let q2_str = &value_row[q2_idx];

    // Quartiles should be computed (negative weights are now filtered out)
    assert!(
        !q2_str.is_empty(),
        "Q2 median should not be empty when valid weights exist"
    );

    let q2_val: f64 = q2_str.parse().expect("Q2 should be a valid number");

    // Only values 2, 4, 5 should contribute (weights 2, 3, 1)
    // Total weight = 6, median at 50% = 3.0
    // Cumulative: 2 (weight 2) = 2, 4 (weight 3) = 5, 5 (weight 1) = 6
    // Median should be 4 (reaches/exceeds 3.0 at value 4)
    assert!(
        (q2_val - 4.0).abs() < 0.1,
        "Expected weighted median ~4.0 (negative weights ignored), got {}",
        q2_val
    );
}

#[test]
fn stats_weighted_mixed_edge_cases() {
    let wrk = Workdir::new("stats_weighted_mixed_edge_cases");
    // Test combination of edge cases: zero, negative, invalid, and valid weights
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["10", "0"],       // zero weight - ignored
            svec!["20", "-2"],      // negative weight - ignored
            svec!["30", "invalid"], // invalid weight - defaults to 1.0
            svec!["40", "3"],       // valid positive weight
            svec!["50", ""],        // empty weight - defaults to 1.0
            svec!["60", "2"],       // valid positive weight
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--weight")
        .arg("weight")
        .arg("--mad")
        .arg("data.csv");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert!(!got.is_empty(), "Stats output should not be empty");

    let headers = &got[0];
    let value_row = &got[1];

    let mean_idx = headers.iter().position(|h| h == "mean").unwrap();
    let mean_val: f64 = value_row[mean_idx].parse().unwrap();

    // Only values 30, 40, 50, 60 should contribute (weights 1, 3, 1, 2)
    // Weighted mean = (30*1 + 40*3 + 50*1 + 60*2) / (1+3+1+2) = 320/7 ≈ 45.71
    assert!(
        (mean_val - 45.71).abs() < 0.5,
        "Expected weighted mean ~45.71 (mixed edge cases), got {}",
        mean_val
    );
}

#[test]
fn stats_json_file_level_metadata() {
    use std::path::Path;

    use serde_json::Value;

    let wrk = Workdir::new("stats_json_file_level_metadata");
    let test_file = wrk.load_test_file("boston311-100.csv");

    // Run stats to populate the cache fields
    let mut dataset_stats_cmd = wrk.command("stats");
    dataset_stats_cmd
        .args(["--cache-threshold", "1"])
        .arg(&test_file);

    wrk.run(&mut dataset_stats_cmd);

    // Read the JSON metadata file
    let json_path = wrk.path("boston311-100.stats.csv.json");
    assert!(Path::new(&json_path).exists(), "JSON file should exist");

    let json_content = std::fs::read_to_string(&json_path).unwrap();
    let json: Value = serde_json::from_str(&json_content).unwrap();

    // Verify field_count is a positive integer (boston311-100.csv has 29 columns)
    let field_count = json.get("field_count").expect("field_count should exist");
    assert!(field_count.is_u64(), "field_count should be a number");
    assert_eq!(
        field_count.as_u64().unwrap(),
        29,
        "field_count should be 29"
    );

    // Verify filesize_bytes is a positive integer
    let filesize_bytes = json
        .get("filesize_bytes")
        .expect("filesize_bytes should exist");
    assert!(filesize_bytes.is_u64(), "filesize_bytes should be a number");
    assert!(
        filesize_bytes.as_u64().unwrap() > 0,
        "filesize_bytes should be positive"
    );

    // Verify hash object exists and has BLAKE3 key
    let hash = json.get("hash").expect("hash should exist");
    assert!(hash.is_object(), "hash should be an object");
    let blake3 = hash.get("BLAKE3").expect("hash.BLAKE3 should exist");
    assert!(blake3.is_string(), "hash.BLAKE3 should be a string");
    let blake3_str = blake3.as_str().unwrap();
    assert!(!blake3_str.is_empty(), "hash.BLAKE3 should not be empty");
    assert_eq!(
        blake3_str.len(),
        64,
        "BLAKE3 hash should be 64 hex characters"
    );
}

#[test]
fn stats_json_arg_input_format() {
    use std::path::Path;

    use serde_json::Value;

    let wrk = Workdir::new("stats_json_arg_input_format");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut stats_cmd = wrk.command("stats");
    stats_cmd.args(["--cache-threshold", "1"]).arg(&test_file);

    wrk.run(&mut stats_cmd);

    let json_path = wrk.path("boston311-100.stats.csv.json");
    assert!(Path::new(&json_path).exists(), "JSON file should exist");

    let json_content = std::fs::read_to_string(&json_path).unwrap();
    let json: Value = serde_json::from_str(&json_content).unwrap();

    // Verify arg_input doesn't contain "Some(" wrapper
    let arg_input = json.get("arg_input").expect("arg_input should exist");
    assert!(arg_input.is_string(), "arg_input should be a string");
    let arg_input_str = arg_input.as_str().unwrap();
    assert!(
        !arg_input_str.contains("Some("),
        "arg_input should not contain 'Some('"
    );
    assert!(
        arg_input_str.ends_with("boston311-100.csv"),
        "arg_input should be a plain path"
    );

    // Verify flag_delimiter doesn't contain "None"
    let flag_delimiter = json
        .get("flag_delimiter")
        .expect("flag_delimiter should exist");
    assert!(
        flag_delimiter.is_string(),
        "flag_delimiter should be a string"
    );
    let flag_delimiter_str = flag_delimiter.as_str().unwrap();
    assert!(
        !flag_delimiter_str.contains("None"),
        "flag_delimiter should not contain 'None'"
    );
}

#[test]
fn stats_json_backward_compat() {
    use std::path::Path;

    use serde_json::Value;

    let wrk = Workdir::new("stats_json_backward_compat");
    let test_file = wrk.load_test_file("boston311-100.csv");

    // First, create a cache file
    let mut cmd = wrk.command("stats");
    cmd.args(["--cache-threshold", "1"]).arg(&test_file);
    wrk.run(&mut cmd);

    let json_path = wrk.path("boston311-100.stats.csv.json");
    assert!(Path::new(&json_path).exists(), "JSON file should exist");

    // Read and modify the JSON to simulate an older cache file without new fields
    let json_content = std::fs::read_to_string(&json_path).unwrap();
    let mut json: Value = serde_json::from_str(&json_content).unwrap();

    // Remove the new fields to simulate an older cache file
    if let Value::Object(ref mut map) = json {
        map.remove("field_count");
        map.remove("filesize_bytes");
        map.remove("hash");
        // Change qsv_version to force cache invalidation and recomputation
        map.insert(
            "qsv_version".to_string(),
            Value::String("0.0.0".to_string()),
        );
    }

    // Write back the modified JSON
    std::fs::write(&json_path, serde_json::to_string_pretty(&json).unwrap()).unwrap();

    // Run stats again - the cache comparison will fail due to qsv_version mismatch,
    // triggering recomputation. This tests that older cache files without the new fields
    // don't cause crashes during the comparison logic.
    let mut recompute_stats_cmd = wrk.command("stats");
    recompute_stats_cmd
        .args(["--cache-threshold", "1"])
        .arg(&test_file);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut recompute_stats_cmd);
    assert!(!got.is_empty(), "Stats should produce output");

    // Verify the JSON file now has the new fields again
    let json_content = std::fs::read_to_string(&json_path).unwrap();
    let json: Value = serde_json::from_str(&json_content).unwrap();
    assert!(
        json.get("field_count").is_some(),
        "field_count should exist after recomputation"
    );
    assert!(
        json.get("filesize_bytes").is_some(),
        "filesize_bytes should exist after recomputation"
    );
    assert!(
        json.get("hash").is_some(),
        "hash should exist after recomputation"
    );
}

#[test]
fn stats_jsonl_tsv_delimiter() {
    use std::path::Path;

    let wrk = Workdir::new("stats_jsonl_tsv_delimiter");
    wrk.create(
        "data.csv",
        vec![
            svec!["name", "value"],
            svec!["a", "1"],
            svec!["b", "2"],
            svec!["c", "3"],
        ],
    );

    let output_tsv = wrk.path("output.tsv");
    let mut cmd = wrk.command("stats");
    cmd.arg("--stats-jsonl")
        .args(["--output", output_tsv.to_str().unwrap()])
        .arg("data.csv");

    wrk.assert_success(&mut cmd);

    // verify the TSV output file was created and is tab-delimited
    assert!(output_tsv.exists(), "TSV output file should exist");
    let tsv_content = std::fs::read_to_string(&output_tsv).unwrap();
    let first_line = tsv_content.lines().next().unwrap();
    assert!(
        first_line.contains('\t'),
        "TSV output should be tab-delimited, got: {first_line}"
    );

    // verify the stats data jsonl file was created
    // the jsonl file is based on the input filestem: data.stats.csv.data.jsonl
    let jsonl_path = wrk.path("data.stats.csv.data.jsonl");
    assert!(
        Path::new(&jsonl_path).exists(),
        "stats data jsonl file should exist at {jsonl_path:?} - naming convention may have changed",
    );

    // verify the jsonl file contains valid JSON lines
    let jsonl_content = std::fs::read_to_string(&jsonl_path).unwrap();
    assert!(
        !jsonl_content.is_empty(),
        "stats data jsonl file should not be empty"
    );
    for line in jsonl_content.lines() {
        assert!(
            serde_json::from_str::<serde_json::Value>(line).is_ok(),
            "each line in jsonl should be valid JSON: {line}"
        );
    }
}

// --- --jsonl / --pretty-json stdout output mode tests --------------------------------

fn jsonl_fixture() -> Vec<Vec<String>> {
    vec![
        svec!["id", "name", "score"],
        svec!["1", "Al", "10.5"],
        svec!["2", "Bo", "20.0"],
        svec!["3", "Cy", ""],
    ]
}

#[test]
fn stats_jsonl_stdout() {
    let wrk = Workdir::new("stats_jsonl_stdout");
    wrk.create("data.csv", jsonl_fixture());

    let mut cmd = wrk.command("stats");
    cmd.arg("--cache-threshold")
        .arg("0")
        .arg("--jsonl")
        .arg("data.csv");
    let got: String = wrk.stdout(&mut cmd);

    // one NDJSON object per column, each a valid JSON object with the metadata fields
    let lines: Vec<&str> = got.lines().collect();
    assert_eq!(lines.len(), 3, "expected one JSON object per column");
    let fields: Vec<String> = lines
        .iter()
        .map(|line| {
            let v: serde_json::Value = serde_json::from_str(line)
                .unwrap_or_else(|_| panic!("each line should be valid JSON: {line}"));
            v.get("field")
                .and_then(serde_json::Value::as_str)
                .unwrap()
                .to_string()
        })
        .collect();
    assert_eq!(fields, vec!["id", "name", "score"]);

    // metadata fields the CSV->tojsonl pipe would lose are present
    let first: serde_json::Value = serde_json::from_str(lines[0]).unwrap();
    assert_eq!(first.get("type").unwrap().as_str().unwrap(), "Integer");
    let name_obj: serde_json::Value = serde_json::from_str(lines[1]).unwrap();
    assert_eq!(name_obj.get("is_ascii").unwrap().as_bool().unwrap(), true);
}

#[test]
fn stats_pretty_json_stdout() {
    let wrk = Workdir::new("stats_pretty_json_stdout");
    wrk.create("data.csv", jsonl_fixture());

    let mut cmd = wrk.command("stats");
    cmd.arg("--cache-threshold")
        .arg("0")
        .arg("--pretty-json")
        .arg("data.csv");
    let got: String = wrk.stdout(&mut cmd);

    // a single pretty-printed JSON array of per-column objects
    let arr: serde_json::Value = serde_json::from_str(&got).unwrap();
    let arr = arr.as_array().expect("output should be a JSON array");
    assert_eq!(arr.len(), 3);
    let fields: Vec<&str> = arr
        .iter()
        .map(|v| v.get("field").unwrap().as_str().unwrap())
        .collect();
    assert_eq!(fields, vec!["id", "name", "score"]);
    // "pretty" => the raw text is multi-line
    assert!(got.lines().count() > 3, "pretty-json should be multi-line");
}

#[test]
fn stats_jsonl_matches_stats_jsonl_sidecar() {
    // --jsonl stdout must be byte-identical to what --stats-jsonl writes to its sidecar.
    let wrk = Workdir::new("stats_jsonl_matches_sidecar");
    wrk.create("data.csv", jsonl_fixture());

    // 1. produce the sidecar via --stats-jsonl
    let mut sidecar_cmd = wrk.command("stats");
    sidecar_cmd
        .arg("--force")
        .arg("--stats-jsonl")
        .args(["--output", wrk.path("out.csv").to_str().unwrap()])
        .arg("data.csv");
    wrk.assert_success(&mut sidecar_cmd);
    let sidecar = std::fs::read_to_string(wrk.path("data.stats.csv.data.jsonl")).unwrap();

    // 2. produce the stdout NDJSON via --jsonl
    let mut jsonl_cmd = wrk.command("stats");
    jsonl_cmd
        .arg("--force")
        .arg("--cache-threshold")
        .arg("0")
        .arg("--jsonl")
        .arg("data.csv");
    let stdout: String = wrk.stdout(&mut jsonl_cmd);

    // stdout has no trailing newline (wrk.stdout trims); compare line-by-line
    let sidecar_lines: Vec<&str> = sidecar.lines().collect();
    let stdout_lines: Vec<&str> = stdout.lines().collect();
    assert_eq!(sidecar_lines, stdout_lines);
}

#[test]
fn stats_jsonl_stdin() {
    let wrk = Workdir::new("stats_jsonl_stdin");
    wrk.create("data.csv", jsonl_fixture());

    let file_content = wrk.read_to_string("data.csv").unwrap();

    let mut cmd = wrk.command("stats");
    cmd.arg("--cache-threshold")
        .arg("0")
        .arg("--jsonl")
        .stdin(process::Stdio::piped())
        .stdout(process::Stdio::piped());

    let mut child = cmd.spawn().unwrap();
    {
        use std::io::Write;
        let mut stdin = child.stdin.take().unwrap();
        stdin.write_all(file_content.as_bytes()).unwrap();
    }
    let out = child.wait_with_output().unwrap();
    assert!(out.status.success(), "stdin --jsonl should succeed");
    let got = String::from_utf8(out.stdout).unwrap();
    let lines: Vec<&str> = got.lines().collect();
    assert_eq!(lines.len(), 3, "expected one JSON object per column");
    for line in &lines {
        assert!(
            serde_json::from_str::<serde_json::Value>(line).is_ok(),
            "each line should be valid JSON: {line}"
        );
    }
}

#[test]
fn stats_jsonl_output_file() {
    let wrk = Workdir::new("stats_jsonl_output_file");
    wrk.create("data.csv", jsonl_fixture());

    let out_path = wrk.path("stats.jsonl");
    let mut cmd = wrk.command("stats");
    cmd.arg("--cache-threshold")
        .arg("0")
        .arg("--jsonl")
        .args(["--output", out_path.to_str().unwrap()])
        .arg("data.csv");
    wrk.assert_success(&mut cmd);

    let content = std::fs::read_to_string(&out_path).unwrap();
    let lines: Vec<&str> = content.lines().collect();
    assert_eq!(
        lines.len(),
        3,
        "expected one JSON object per column in file"
    );
    for line in &lines {
        assert!(
            serde_json::from_str::<serde_json::Value>(line).is_ok(),
            "each line should be valid JSON: {line}"
        );
    }
}

#[test]
fn stats_jsonl_conflicts_with_pretty_json() {
    let wrk = Workdir::new("stats_jsonl_conflicts_pretty");
    wrk.create("data.csv", jsonl_fixture());

    let mut cmd = wrk.command("stats");
    cmd.arg("--jsonl").arg("--pretty-json").arg("data.csv");
    wrk.assert_err(&mut cmd);
    let stderr = wrk.output_stderr(&mut cmd);
    assert!(
        stderr.contains("cannot be combined"),
        "expected mutual-exclusivity error, got: {stderr}"
    );
}

#[test]
fn stats_jsonl_conflicts_with_stats_jsonl() {
    let wrk = Workdir::new("stats_jsonl_conflicts_stats_jsonl");
    wrk.create("data.csv", jsonl_fixture());

    let mut cmd = wrk.command("stats");
    cmd.arg("--jsonl").arg("--stats-jsonl").arg("data.csv");
    wrk.assert_err(&mut cmd);
    let stderr = wrk.output_stderr(&mut cmd);
    assert!(
        stderr.contains("cannot be combined"),
        "expected mutual-exclusivity error, got: {stderr}"
    );
}

#[test]
fn stats_jsonl_reused_cache_with_cache_threshold_zero() {
    // Regression: when a valid stats cache is reused, currstats_filename points at the
    // cache file. With --cache-threshold 0, the cleanup block deletes that cache before
    // the output stage runs. --jsonl must still emit the stats (it reads from a temp
    // copy taken before deletion), not fail trying to reopen a removed file.
    let wrk = Workdir::new("stats_jsonl_reused_cache_threshold_zero");
    wrk.create("data.csv", jsonl_fixture());

    // 1. create a valid stats cache (force caching)
    let mut seed = wrk.command("stats");
    seed.arg("--cache-threshold")
        .arg("1")
        .args(["--output", wrk.path("seed.csv").to_str().unwrap()])
        .arg("data.csv");
    wrk.assert_success(&mut seed);
    assert!(
        wrk.path("data.stats.csv").exists(),
        "stats cache should exist after seeding"
    );

    // 2. reuse the cache while --cache-threshold 0 tears it down
    let mut cmd = wrk.command("stats");
    cmd.arg("--jsonl")
        .arg("--cache-threshold")
        .arg("0")
        .arg("data.csv");
    let got: String = wrk.stdout(&mut cmd);

    let lines: Vec<&str> = got.lines().collect();
    assert_eq!(lines.len(), 3, "expected one JSON object per column");
    for line in &lines {
        assert!(
            serde_json::from_str::<serde_json::Value>(line).is_ok(),
            "each line should be valid JSON: {line}"
        );
    }
    // --cache-threshold 0 semantics preserved: the cache is gone afterward
    assert!(
        !wrk.path("data.stats.csv").exists(),
        "stats cache should be deleted by --cache-threshold 0"
    );
}

// --- --quantile-method approx (t-digest) tests --------------------------------------

/// Build a uniform 1..=N column. With enough rows, t-digest's rank error stays small
/// (~1%), so we can assert tighter envelopes than the small-N smoke test.
fn approx_quartiles_fixture(n: usize) -> Vec<Vec<String>> {
    let mut rows: Vec<Vec<String>> = Vec::with_capacity(n + 1);
    rows.push(vec!["value".to_string()]);
    for i in 1..=n {
        rows.push(vec![i.to_string()]);
    }
    rows
}

#[test]
fn stats_quantile_method_approx_quartiles_within_envelope() {
    // With N=10_000 uniform integers 1..=10_000 the exact quartiles are q1=2500,
    // q2=5000, q3=7500. T-digest with default k=200 should land inside ±2% of those.
    let wrk = Workdir::new("stats_quantile_method_approx_quartiles_within_envelope");
    wrk.create("data.csv", approx_quartiles_fixture(10_000));

    let mut cmd = wrk.command("stats");
    // Pin --jobs 1 because t-digest merge is associative but not chunk-count-invariant;
    // CI runs with arbitrary thread counts would otherwise produce flaky digests.
    cmd.arg("--quartiles")
        .arg("--quantile-method")
        .arg("approx")
        .arg("--jobs")
        .arg("1")
        .arg("data.csv");

    // Single-invocation capture: avoids the wrk.assert_success + wrk.read_stdout
    // double-run pattern (the second run's exit status would not be asserted).
    let output = cmd.output().unwrap();
    assert!(
        output.status.success(),
        "stats command failed: stderr = {}",
        String::from_utf8_lossy(&output.stderr)
    );
    let stdout = String::from_utf8_lossy(&output.stdout);
    let mut lines = stdout.lines();
    let headers: Vec<&str> = lines
        .next()
        .expect("stats output should have a header row")
        .split(',')
        .collect();
    let value_row: Vec<&str> = lines
        .next()
        .expect("stats output should have at least one data row")
        .split(',')
        .collect();
    let q1: f64 = value_row[headers.iter().position(|h| *h == "q1").unwrap()]
        .parse()
        .unwrap();
    let q2: f64 = value_row[headers.iter().position(|h| *h == "q2_median").unwrap()]
        .parse()
        .unwrap();
    let q3: f64 = value_row[headers.iter().position(|h| *h == "q3").unwrap()]
        .parse()
        .unwrap();

    // ~1% rank error per t-digest; allow 2% on values for safety. Approx error is
    // typically much smaller in the tails than the middle.
    let tol = 0.02_f64;
    assert!(
        (q1 - 2500.0).abs() / 2500.0 < tol,
        "approx q1 = {q1} not within 2% of 2500"
    );
    assert!(
        (q2 - 5000.0).abs() / 5000.0 < tol,
        "approx q2 = {q2} not within 2% of 5000"
    );
    assert!(
        (q3 - 7500.0).abs() / 7500.0 < tol,
        "approx q3 = {q3} not within 2% of 7500"
    );
}

#[test]
fn stats_quantile_method_approx_with_weight_is_rejected() {
    let wrk = Workdir::new("stats_quantile_method_approx_with_weight_is_rejected");
    wrk.create(
        "data.csv",
        vec![svec!["value", "weight"], svec!["1", "1"], svec!["2", "1"]],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--quartiles")
        .arg("--quantile-method")
        .arg("approx")
        .arg("--weight")
        .arg("weight")
        .arg("data.csv");

    let output = cmd.output().unwrap();
    assert!(
        !output.status.success(),
        "Should fail when --quantile-method approx is combined with --weight"
    );
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        stderr.contains("approx") && stderr.contains("--weight"),
        "Error should explain the approx+weight incompatibility, got: {stderr}"
    );
}

#[test]
fn stats_quantile_method_approx_disables_mad() {
    // --mad with --quantile-method approx should print a warning and drop the MAD column
    // from the output (MAD is not computable from a t-digest).
    let wrk = Workdir::new("stats_quantile_method_approx_disables_mad");
    wrk.create("data.csv", approx_quartiles_fixture(100));

    let mut cmd = wrk.command("stats");
    cmd.arg("--mad")
        .arg("--quantile-method")
        .arg("approx")
        .arg("--jobs")
        .arg("1")
        .arg("data.csv");

    // Capture both stdout and stderr from a single invocation so the warning we assert on
    // and the headers we assert on are guaranteed to come from the same run (avoids
    // running the command twice — `wrk.read_stdout(&mut cmd)` would re-execute it).
    let output = cmd.output().unwrap();
    assert!(
        output.status.success(),
        "approx + --mad should still succeed"
    );
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        stderr.contains("does not support MAD"),
        "Should warn that MAD is disabled under approx, got: {stderr}"
    );

    // Parse stdout from the captured bytes (no second invocation).
    let stdout = String::from_utf8_lossy(&output.stdout);
    let headers: Vec<&str> = stdout
        .lines()
        .next()
        .expect("stats output should have a header row")
        .split(',')
        .collect();
    assert!(
        !headers.contains(&"mad"),
        "mad column should be omitted when approx disables MAD, got headers: {headers:?}"
    );
}

#[test]
fn stats_quantile_method_invalid_value_is_rejected() {
    let wrk = Workdir::new("stats_quantile_method_invalid_value_is_rejected");
    wrk.create("data.csv", vec![svec!["value"], svec!["1"]]);

    let mut cmd = wrk.command("stats");
    cmd.arg("--quantile-method").arg("nonsense").arg("data.csv");

    let output = cmd.output().unwrap();
    assert!(
        !output.status.success(),
        "Should fail on invalid --quantile-method"
    );
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        stderr.contains("Invalid --quantile-method"),
        "Error should mention invalid --quantile-method, got: {stderr}"
    );
}

#[test]
fn stats_quantile_method_exact_is_default_and_byte_identical() {
    // Sanity: omitting --quantile-method should produce the same output as
    // explicitly passing --quantile-method exact. This guards against a
    // regression where the default ever silently changes.
    let wrk = Workdir::new("stats_quantile_method_exact_is_default_and_byte_identical");
    wrk.create("data.csv", approx_quartiles_fixture(50));

    let mut cmd_default = wrk.command("stats");
    cmd_default.arg("--quartiles").arg("data.csv");
    let default_out: Vec<Vec<String>> = wrk.read_stdout(&mut cmd_default);

    let mut cmd_exact = wrk.command("stats");
    cmd_exact
        .arg("--quartiles")
        .arg("--quantile-method")
        .arg("exact")
        .arg("data.csv");
    let exact_out: Vec<Vec<String>> = wrk.read_stdout(&mut cmd_exact);

    assert_eq!(
        default_out, exact_out,
        "Default --quantile-method must match explicit 'exact'"
    );
}

#[test]
fn stats_quantile_method_approx_percentiles_empty_numeric_column() {
    // Locks the contract that an all-string column under --percentiles + --quantile-method
    // approx produces an empty `percentiles` cell (no numeric values were ever fed to the
    // t-digest, so it stays empty and the reader returns None — matching the exact path).
    let wrk = Workdir::new("stats_quantile_method_approx_percentiles_empty_numeric_column");
    wrk.create(
        "data.csv",
        vec![
            svec!["value"],
            svec!["alpha"],
            svec!["beta"],
            svec!["gamma"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--percentiles")
        .arg("--quantile-method")
        .arg("approx")
        .arg("--jobs")
        .arg("1")
        .arg("data.csv");

    // Single-invocation capture: avoids the wrk.assert_success + wrk.read_stdout
    // double-run pattern (the second run's exit status would not be asserted).
    let output = cmd.output().unwrap();
    assert!(
        output.status.success(),
        "stats command failed: stderr = {}",
        String::from_utf8_lossy(&output.stderr)
    );
    let stdout = String::from_utf8_lossy(&output.stdout);
    let mut lines = stdout.lines();
    let headers: Vec<&str> = lines
        .next()
        .expect("stats output should have a header row")
        .split(',')
        .collect();
    let row: Vec<&str> = lines
        .next()
        .expect("stats output should have at least one data row")
        .split(',')
        .collect();
    let pct_idx = headers
        .iter()
        .position(|h| *h == "percentiles")
        .expect("percentiles column should exist");
    assert_eq!(
        row[pct_idx], "",
        "percentiles cell should be empty for an all-string column under approx, got: {:?}",
        row[pct_idx]
    );
}

// --- --mode-cardinality-cap tests ---------------------------------------------------

#[test]
fn stats_mode_cardinality_cap_default_is_unbounded() {
    // The default (cap=0) must produce byte-identical output to omitting the flag.
    let wrk = Workdir::new("stats_mode_cardinality_cap_default_is_unbounded");
    wrk.create(
        "data.csv",
        vec![
            svec!["id"],
            svec!["1"],
            svec!["1"],
            svec!["2"],
            svec!["2"],
            svec!["3"],
        ],
    );

    let mut cmd_default = wrk.command("stats");
    cmd_default
        .arg("--cardinality")
        .arg("--mode")
        .arg("data.csv");
    let default_out: Vec<Vec<String>> = wrk.read_stdout(&mut cmd_default);

    let mut cmd_zero = wrk.command("stats");
    cmd_zero
        .arg("--cardinality")
        .arg("--mode")
        .arg("--mode-cardinality-cap")
        .arg("0")
        .arg("data.csv");
    let zero_out: Vec<Vec<String>> = wrk.read_stdout(&mut cmd_zero);

    assert_eq!(
        default_out, zero_out,
        "Default behavior must be byte-identical to --mode-cardinality-cap 0"
    );
}

#[test]
fn stats_mode_cardinality_cap_drops_high_cardinality_column() {
    // Cap at 3 with 10 rows must trip the dropper. Output should contain the
    // sentinel values and NOT the normal mode/cardinality fields.
    let wrk = Workdir::new("stats_mode_cardinality_cap_drops_high_cardinality_column");
    let mut rows: Vec<Vec<String>> = vec![vec!["id".to_string()]];
    for i in 1..=10_usize {
        rows.push(vec![i.to_string()]);
    }
    wrk.create("data.csv", rows);

    let mut cmd = wrk.command("stats");
    cmd.arg("--cardinality")
        .arg("--mode")
        .arg("--mode-cardinality-cap")
        .arg("3")
        .arg("--jobs")
        .arg("1") // pin to deterministic single-thread merge for the assertion
        .arg("data.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let headers = &got[0];
    let row = &got[1];
    let card_idx = headers
        .iter()
        .position(|h| h == "cardinality")
        .expect("cardinality column should exist");
    let mode_idx = headers
        .iter()
        .position(|h| h == "mode")
        .expect("mode column should exist");
    let antimode_idx = headers
        .iter()
        .position(|h| h == "antimode")
        .expect("antimode column should exist");

    assert_eq!(
        row[card_idx], ">=3",
        "cardinality should be the sentinel >=<cap>"
    );
    assert_eq!(
        row[mode_idx], "*HIGH_CARDINALITY",
        "mode should be the *HIGH_CARDINALITY sentinel"
    );
    assert_eq!(
        row[antimode_idx], "*HIGH_CARDINALITY",
        "antimode should be the *HIGH_CARDINALITY sentinel"
    );
}

#[test]
fn stats_mode_cardinality_cap_preserves_low_cardinality_column() {
    // Cap large enough that a low-cardinality column does NOT trip the dropper.
    let wrk = Workdir::new("stats_mode_cardinality_cap_preserves_low_cardinality_column");
    wrk.create(
        "data.csv",
        vec![
            svec!["category"],
            svec!["A"],
            svec!["A"],
            svec!["B"],
            svec!["B"],
            svec!["C"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--cardinality")
        .arg("--mode")
        .arg("--mode-cardinality-cap")
        .arg("1000")
        .arg("data.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let headers = &got[0];
    let row = &got[1];
    let card_idx = headers.iter().position(|h| h == "cardinality").unwrap();

    assert_eq!(
        row[card_idx], "3",
        "cardinality should be exact (3) when cap is not tripped, got {:?}",
        row[card_idx]
    );
    assert!(
        !row.iter().any(|c| c.contains("HIGH_CARDINALITY")),
        "no row should contain the HIGH_CARDINALITY sentinel when cap is not tripped: {row:?}"
    );
}

#[test]
fn stats_mode_cardinality_cap_weighted_path() {
    // Exercises the cap's weighted-modes branch (HashMap::len() bound). With cap=3 and a
    // column whose unique-value count exceeds 3, the weighted tracker is dropped and
    // the sentinels fire — same as the unweighted path.
    let wrk = Workdir::new("stats_mode_cardinality_cap_weighted_path");
    wrk.create(
        "data.csv",
        vec![
            svec!["value", "weight"],
            svec!["a", "1"],
            svec!["b", "1"],
            svec!["c", "1"],
            svec!["d", "1"],
            svec!["e", "1"],
            svec!["f", "1"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--cardinality")
        .arg("--mode")
        .arg("--weight")
        .arg("weight")
        .arg("--mode-cardinality-cap")
        .arg("3")
        .arg("--jobs")
        .arg("1")
        .arg("data.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let headers = &got[0];
    let row = &got[1];
    let card_idx = headers
        .iter()
        .position(|h| h == "cardinality")
        .expect("cardinality column should exist");
    let mode_idx = headers
        .iter()
        .position(|h| h == "mode")
        .expect("mode column should exist");

    assert_eq!(
        row[card_idx], ">=3",
        "weighted-mode cap should produce >=<cap> cardinality, got: {:?}",
        row[card_idx]
    );
    assert_eq!(
        row[mode_idx], "*HIGH_CARDINALITY",
        "weighted-mode cap should produce *HIGH_CARDINALITY mode, got: {:?}",
        row[mode_idx]
    );
}

#[test]
fn stats_mode_cardinality_cap_multi_chunk_post_merge_check() {
    // Exercises the post-merge cap re-check: when individual chunks each stay below the
    // cap but the merged result crosses it. Forces parallel chunking by indexing the
    // file (parallel_stats requires an index) and using the default --jobs.
    let wrk = Workdir::new("stats_mode_cardinality_cap_multi_chunk_post_merge_check");
    let mut rows: Vec<Vec<String>> = vec![vec!["id".to_string()]];
    for i in 1..=200_usize {
        rows.push(vec![i.to_string()]);
    }
    wrk.create("data.csv", rows);

    // Build the index so parallel_stats kicks in (qsv routes to sequential_stats when
    // the index is missing, even at default --jobs).
    let mut idx_cmd = wrk.command("index");
    idx_cmd.arg("data.csv");
    wrk.assert_success(&mut idx_cmd);

    let mut cmd = wrk.command("stats");
    cmd.arg("--cardinality")
        .arg("--mode")
        .arg("--mode-cardinality-cap")
        .arg("50")
        .arg("data.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let row = &got[1];
    let headers = &got[0];
    let card_idx = headers.iter().position(|h| h == "cardinality").unwrap();

    // 200 rows, cap=50: even if parallel chunks each stay near the cap individually,
    // the merged total must trip the post-merge re-check and emit the sentinel.
    assert_eq!(
        row[card_idx], ">=50",
        "post-merge cap re-check should fire across chunks, got: {:?}",
        row[card_idx]
    );
}

#[test]
fn stats_mode_cardinality_cap_cardinality_only_emits_two_fields() {
    // When --cardinality is set but --mode is not, the dropped path must emit exactly
    // two sentinel fields (cardinality + uniqueness_ratio) and NOT the six mode/antimode
    // fields. Locks the field-count contract.
    let wrk = Workdir::new("stats_mode_cardinality_cap_cardinality_only_emits_two_fields");
    let mut rows: Vec<Vec<String>> = vec![vec!["id".to_string()]];
    for i in 1..=10_usize {
        rows.push(vec![i.to_string()]);
    }
    wrk.create("data.csv", rows);

    let mut cmd = wrk.command("stats");
    cmd.arg("--cardinality")
        .arg("--mode-cardinality-cap")
        .arg("3")
        .arg("--jobs")
        .arg("1")
        .arg("data.csv");
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let headers = &got[0];
    let row = &got[1];

    // cardinality should be present and = ">=3"
    let card_idx = headers
        .iter()
        .position(|h| h == "cardinality")
        .expect("cardinality column should exist");
    assert_eq!(
        row[card_idx], ">=3",
        "cardinality should be the sentinel, got: {:?}",
        row[card_idx]
    );

    // mode columns should NOT be in the headers when --mode is omitted (and --everything
    // is not set). This guards against the dropped path accidentally emitting 8 fields
    // instead of 2.
    assert!(
        !headers.iter().any(|h| h == "mode"),
        "mode column should not be present when --mode is not set: {headers:?}"
    );
    assert!(
        !headers.iter().any(|h| h == "antimode"),
        "antimode column should not be present when --mode is not set: {headers:?}"
    );
}

/// Build a fixture with `unique` distinct strings each repeated `reps` times. The true
/// cardinality is exactly `unique`; we use this to gauge HLL's relative error.
fn approx_cardinality_fixture(unique: usize, reps: usize) -> Vec<Vec<String>> {
    let mut rows: Vec<Vec<String>> = Vec::with_capacity(unique * reps + 1);
    rows.push(vec!["value".to_string()]);
    for _ in 0..reps {
        for i in 0..unique {
            rows.push(vec![format!("v{i}")]);
        }
    }
    rows
}

#[test]
fn stats_cardinality_method_approx_within_envelope() {
    // 10_000 distinct strings × 10 reps = 100_000 rows, true cardinality = 10_000.
    // HLL with lg_k=12 has ~1.5% RSE; allow 3% slack for safety.
    let wrk = Workdir::new("stats_cardinality_method_approx_within_envelope");
    wrk.create("data.csv", approx_cardinality_fixture(10_000, 10));

    let mut cmd = wrk.command("stats");
    // Pin --jobs 1 for reproducibility (HLL union is associative + order-invariant,
    // but we want exact reproducibility regardless).
    cmd.arg("--cardinality")
        .arg("--cardinality-method")
        .arg("approx")
        .arg("--jobs")
        .arg("1")
        .arg("data.csv");

    let output = cmd.output().unwrap();
    assert!(
        output.status.success(),
        "stats command failed: stderr = {}",
        String::from_utf8_lossy(&output.stderr)
    );
    let stdout = String::from_utf8_lossy(&output.stdout);
    let mut lines = stdout.lines();
    let headers: Vec<&str> = lines
        .next()
        .expect("stats output should have a header row")
        .split(',')
        .collect();
    let value_row: Vec<&str> = lines
        .next()
        .expect("stats output should have at least one data row")
        .split(',')
        .collect();
    let card: f64 = value_row[headers.iter().position(|h| *h == "cardinality").unwrap()]
        .parse()
        .expect("cardinality should be a plain integer under approx (no '>=' sentinel)");

    let tol = 0.03_f64;
    assert!(
        (card - 10_000.0).abs() / 10_000.0 < tol,
        "approx cardinality = {card} not within 3% of 10_000"
    );
}

#[test]
fn stats_cardinality_method_approx_no_cap_sentinel() {
    // Under approx, the --mode-cardinality-cap should NOT cause the cardinality column
    // to emit ">=<cap>" — HLL provides an estimate at fixed memory regardless of cap.
    let wrk = Workdir::new("stats_cardinality_method_approx_no_cap_sentinel");
    wrk.create("data.csv", approx_cardinality_fixture(10_000, 2));

    let mut cmd = wrk.command("stats");
    cmd.arg("--cardinality")
        .arg("--cardinality-method")
        .arg("approx")
        .arg("--mode-cardinality-cap")
        .arg("100")
        .arg("--jobs")
        .arg("1")
        .arg("data.csv");

    let output = cmd.output().unwrap();
    assert!(output.status.success());
    let stdout = String::from_utf8_lossy(&output.stdout);
    let mut lines = stdout.lines();
    let headers: Vec<&str> = lines.next().unwrap().split(',').collect();
    let row: Vec<&str> = lines.next().unwrap().split(',').collect();
    let card_str = row[headers.iter().position(|h| *h == "cardinality").unwrap()];
    assert!(
        !card_str.starts_with(">="),
        "approx cardinality should never emit '>=cap' sentinel, got: {card_str}"
    );
    let card: f64 = card_str.parse().expect("cardinality should be numeric");
    assert!(
        card > 5_000.0,
        "approx cardinality should reflect true ~10_000 unique values, got: {card}"
    );
}

#[test]
fn stats_cardinality_method_approx_with_infer_boolean_falls_back_to_exact() {
    // --infer-boolean needs cardinality == 2 to be exact, so run() forces exact mode
    // and emits a warning. The cardinality output for a known boolean column must be
    // exactly 2, with no '>=' sentinel and no approximation.
    let wrk = Workdir::new("stats_cardinality_method_approx_with_infer_boolean_falls_back");
    wrk.create(
        "data.csv",
        vec![
            svec!["flag"],
            svec!["yes"],
            svec!["no"],
            svec!["yes"],
            svec!["no"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-boolean")
        .arg("--cardinality-method")
        .arg("approx")
        .arg("--jobs")
        .arg("1")
        .arg("data.csv");

    let output = cmd.output().unwrap();
    assert!(
        output.status.success(),
        "stats command failed: stderr = {}",
        String::from_utf8_lossy(&output.stderr)
    );

    let stdout = String::from_utf8_lossy(&output.stdout);
    let mut lines = stdout.lines();
    let headers: Vec<&str> = lines.next().unwrap().split(',').collect();
    let row: Vec<&str> = lines.next().unwrap().split(',').collect();
    let card: u64 = row[headers.iter().position(|h| *h == "cardinality").unwrap()]
        .parse()
        .unwrap();
    assert_eq!(card, 2, "boolean column should report exact cardinality 2");
    let typ = row[headers.iter().position(|h| *h == "type").unwrap()];
    assert_eq!(typ, "Boolean", "type should infer as Boolean, got: {typ}");
}

#[test]
fn stats_cardinality_method_invalid_value_is_rejected() {
    let wrk = Workdir::new("stats_cardinality_method_invalid_value_is_rejected");
    wrk.create("data.csv", vec![svec!["value"], svec!["1"], svec!["2"]]);

    let mut cmd = wrk.command("stats");
    cmd.arg("--cardinality")
        .arg("--cardinality-method")
        .arg("bogus")
        .arg("data.csv");

    let output = cmd.output().unwrap();
    assert!(
        !output.status.success(),
        "should fail on invalid --cardinality-method value"
    );
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        stderr.contains("--cardinality-method") && stderr.contains("bogus"),
        "error should mention the bad flag and value, got: {stderr}"
    );
}

// --- big-endian platform-rejection tests ---------------------------------------------
//
// Apache DataSketches does not support big-endian targets, so the qsv source
// gates t-digest / HyperLogLog behind `#[cfg(not(target_endian = "big"))]`
// and `run()` rejects `--quantile-method approx` / `--cardinality-method approx`
// at the top of the command with a clear platform error message naming s390x.
// These tests are only compiled on big-endian targets; on little-endian they
// would always fail (the flags work normally there) and the existing
// `*_method_approx_*` tests already cover that path.
//
// Names intentionally avoid the `stats_*_method_approx_` prefix so the s390x
// CI `--skip` filters do not drop them.

#[cfg(target_endian = "big")]
#[test]
fn stats_big_endian_quantile_method_approx_rejected() {
    let wrk = Workdir::new("stats_big_endian_quantile_method_approx_rejected");
    wrk.create("data.csv", vec![svec!["value"], svec!["1"], svec!["2"]]);

    let mut cmd = wrk.command("stats");
    cmd.arg("--quartiles")
        .arg("--quantile-method")
        .arg("approx")
        .arg("data.csv");

    let output = cmd.output().unwrap();
    assert!(
        !output.status.success(),
        "--quantile-method approx should be rejected on big-endian"
    );
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        stderr.contains("--quantile-method approx")
            && stderr.contains("requires a little-endian target")
            && stderr.contains("s390x"),
        "error should name the flag, the platform constraint, and s390x; got: {stderr}"
    );
}

#[cfg(target_endian = "big")]
#[test]
fn stats_big_endian_cardinality_method_approx_rejected() {
    let wrk = Workdir::new("stats_big_endian_cardinality_method_approx_rejected");
    wrk.create("data.csv", vec![svec!["value"], svec!["1"], svec!["2"]]);

    let mut cmd = wrk.command("stats");
    cmd.arg("--cardinality")
        .arg("--cardinality-method")
        .arg("approx")
        .arg("data.csv");

    let output = cmd.output().unwrap();
    assert!(
        !output.status.success(),
        "--cardinality-method approx should be rejected on big-endian"
    );
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        stderr.contains("--cardinality-method approx")
            && stderr.contains("requires a little-endian target")
            && stderr.contains("s390x"),
        "error should name the flag, the platform constraint, and s390x; got: {stderr}"
    );
}

// Regression tests for issue #3855: when a column's inferred type widens to
// String mid-stream (e.g. --infer-dates Date→String, Integer→String,
// Float→String, DateTime→String), the string-level stats (min/max,
// min_length/max_length/sum_length/avg_length, stddev_length/variance_length/
// cv_length, sort_order/sortiness) must reflect ALL rows, not just the
// post-widen tail.

/// Helper: run `stats` on the given multi-column CSV, return the row for
/// `field_name` indexed by stats-output header.
fn widen_get_row(
    wrk: &Workdir,
    cmd: &mut process::Command,
    field_name: &str,
) -> std::collections::HashMap<String, String> {
    let mut rows: Vec<Vec<String>> = wrk.read_stdout(cmd);
    let headers = rows.remove(0);
    let row = rows
        .into_iter()
        .find(|r| r.first().map(String::as_str) == Some(field_name))
        .unwrap_or_else(|| panic!("no stats row for field {field_name}"));
    headers.into_iter().zip(row).collect()
}

#[test]
fn stats_widen_date_to_string_issue3855() {
    let wrk = Workdir::new("stats_widen_date_to_string_issue3855");
    wrk.create(
        "data.csv",
        vec![
            svec!["ID", "txt", "date1", "date2"],
            svec!["1", "abc", "2026-05-12", "2026-05-13"],
            svec!["2", "def", "2026-05-11", "NA"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-dates")
        .args(["--dates-whitelist", "all"])
        .arg("data.csv");

    let row = widen_get_row(&wrk, &mut cmd, "date2");
    assert_eq!(row.get("type").map(String::as_str), Some("String"));
    // lex min/max across the post-widen "NA" *and* the earlier "2026-05-13"
    assert_eq!(row.get("min").map(String::as_str), Some("2026-05-13"));
    assert_eq!(row.get("max").map(String::as_str), Some("NA"));
    assert_eq!(row.get("min_length").map(String::as_str), Some("2"));
    assert_eq!(row.get("max_length").map(String::as_str), Some("10"));
    assert_eq!(row.get("sum_length").map(String::as_str), Some("12"));
    assert_eq!(row.get("avg_length").map(String::as_str), Some("6"));
    // online_len-derived length distribution must include both samples
    // (lengths 10 and 2). If a regression re-gates `online_len.add` under
    // `t == TString`, only "NA" (length 2) would be added and these would
    // all become 0 (single-sample distribution).
    assert_eq!(row.get("stddev_length").map(String::as_str), Some("4"));
    assert_eq!(row.get("variance_length").map(String::as_str), Some("16"));
    assert_eq!(row.get("cv_length").map(String::as_str), Some("0.6667"));
    // sort_order is lexical (not chronological) once the column has widened
    // to String: "2026-05-13" < "NA" because digit '2' (0x32) < 'N' (0x4E).
    assert_eq!(row.get("sort_order").map(String::as_str), Some("Ascending"));
}

#[test]
fn stats_widen_integer_to_string_issue3855() {
    let wrk = Workdir::new("stats_widen_integer_to_string_issue3855");
    wrk.create(
        "data.csv",
        vec![svec!["ID", "mixed"], svec!["1", "1"], svec!["2", "abc"]],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv");

    let row = widen_get_row(&wrk, &mut cmd, "mixed");
    assert_eq!(row.get("type").map(String::as_str), Some("String"));
    // earlier "1" must contribute: lex min="1" (vs post-widen-only min="abc")
    assert_eq!(row.get("min").map(String::as_str), Some("1"));
    assert_eq!(row.get("max").map(String::as_str), Some("abc"));
    assert_eq!(row.get("min_length").map(String::as_str), Some("1"));
    assert_eq!(row.get("max_length").map(String::as_str), Some("3"));
    assert_eq!(row.get("sum_length").map(String::as_str), Some("4"));
    assert_eq!(row.get("avg_length").map(String::as_str), Some("2"));
    // online_len-derived length distribution over both samples (1 and 3).
    // Pre-fix regression would only see length 3 → stddev/variance/cv = 0.
    assert_eq!(row.get("stddev_length").map(String::as_str), Some("1"));
    assert_eq!(row.get("variance_length").map(String::as_str), Some("1"));
    assert_eq!(row.get("cv_length").map(String::as_str), Some("0.5"));
}

#[test]
fn stats_widen_float_to_string_issue3855() {
    let wrk = Workdir::new("stats_widen_float_to_string_issue3855");
    wrk.create(
        "data.csv",
        vec![svec!["ID", "mixed"], svec!["1", "1.5"], svec!["2", "abc"]],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv");

    let row = widen_get_row(&wrk, &mut cmd, "mixed");
    assert_eq!(row.get("type").map(String::as_str), Some("String"));
    // earlier "1.5" must contribute
    assert_eq!(row.get("min").map(String::as_str), Some("1.5"));
    assert_eq!(row.get("max").map(String::as_str), Some("abc"));
    assert_eq!(row.get("min_length").map(String::as_str), Some("3"));
    assert_eq!(row.get("max_length").map(String::as_str), Some("3"));
    assert_eq!(row.get("sum_length").map(String::as_str), Some("6"));
    assert_eq!(row.get("avg_length").map(String::as_str), Some("3"));
    // Both samples have equal length 3 → distribution stats are exactly 0
    // both pre-fix and post-fix here, but the sum_length=6 assertion above
    // already catches the regression for this case.
    assert_eq!(row.get("stddev_length").map(String::as_str), Some("0"));
    assert_eq!(row.get("variance_length").map(String::as_str), Some("0"));
    assert_eq!(row.get("cv_length").map(String::as_str), Some("0"));
}

#[test]
fn stats_widen_datetime_to_string_issue3855() {
    let wrk = Workdir::new("stats_widen_datetime_to_string_issue3855");
    wrk.create(
        "data.csv",
        vec![
            svec!["ID", "t"],
            svec!["1", "2026-05-12T10:00:00Z"],
            svec!["2", "oops"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("--infer-dates")
        .args(["--dates-whitelist", "all"])
        .arg("data.csv");

    let row = widen_get_row(&wrk, &mut cmd, "t");
    assert_eq!(row.get("type").map(String::as_str), Some("String"));
    // earlier datetime contributes to lex/length stats
    assert_eq!(
        row.get("min").map(String::as_str),
        Some("2026-05-12T10:00:00Z")
    );
    assert_eq!(row.get("max").map(String::as_str), Some("oops"));
    assert_eq!(row.get("min_length").map(String::as_str), Some("4"));
    assert_eq!(row.get("max_length").map(String::as_str), Some("20"));
    assert_eq!(row.get("sum_length").map(String::as_str), Some("24"));
    assert_eq!(row.get("avg_length").map(String::as_str), Some("12"));
    // online_len-derived length distribution over both samples (20 and 4).
    // Pre-fix regression would only see length 4 → stddev/variance/cv = 0.
    assert_eq!(row.get("stddev_length").map(String::as_str), Some("8"));
    assert_eq!(row.get("variance_length").map(String::as_str), Some("64"));
    assert_eq!(row.get("cv_length").map(String::as_str), Some("0.6667"));
}

#[test]
fn stats_widen_no_regression_pure_integer_issue3855() {
    // Sanity: a pure-Integer column (no widening) must still emit empty
    // length-stat fields — confirms the show(TString) gating still skips
    // length-stat emission for non-String final types.
    let wrk = Workdir::new("stats_widen_no_regression_pure_integer_issue3855");
    wrk.create(
        "data.csv",
        vec![
            svec!["ID", "nums"],
            svec!["1", "10"],
            svec!["2", "20"],
            svec!["3", "30"],
        ],
    );

    let mut cmd = wrk.command("stats");
    cmd.arg("data.csv");

    let row = widen_get_row(&wrk, &mut cmd, "nums");
    assert_eq!(row.get("type").map(String::as_str), Some("Integer"));
    assert_eq!(row.get("min_length").map(String::as_str), Some(""));
    assert_eq!(row.get("max_length").map(String::as_str), Some(""));
    assert_eq!(row.get("sum_length").map(String::as_str), Some(""));
    assert_eq!(row.get("avg_length").map(String::as_str), Some(""));
    // show() gates these on final typ == TString, so pure-Integer columns
    // emit empty fields even though the trackers now collect data.
    assert_eq!(row.get("stddev_length").map(String::as_str), Some(""));
    assert_eq!(row.get("variance_length").map(String::as_str), Some(""));
    assert_eq!(row.get("cv_length").map(String::as_str), Some(""));
}

// issue #1417 regression: `stats` checks `rconfig.format_error` directly (ignoring
// skip_format_check). An explicit QSV_SKIP_FORMAT_CHECK opt-out must therefore not
// leave a format_error set on a failed `.zip` conversion, so the bad zip is read as
// raw bytes instead of being rejected.
#[test]
fn stats_invalid_zip_with_skip_format_check_does_not_error() {
    let wrk = Workdir::new("stats_invalid_zip_with_skip_format_check_does_not_error");
    wrk.create_from_string("bad.zip", "header\nvalue\n");
    let mut cmd = wrk.command("stats");
    cmd.env("QSV_SKIP_FORMAT_CHECK", "1").arg("bad.zip");
    wrk.assert_success(&mut cmd);
}
