use crate::workdir::Workdir;

macro_rules! sqlp_test {
    ($name:ident, $fun:expr_2021) => {
        mod $name {
            use std::process;

            #[allow(unused_imports)]
            use super::{make_rows, setup};
            use crate::workdir::Workdir;

            #[test]
            fn headers() {
                let wrk = setup(stringify!($name));
                let mut cmd = wrk.command("sqlp");
                cmd.args(&["cities.csv", "places.csv"]);
                $fun(wrk, cmd);
            }
        }
    };
}

fn setup(name: &str) -> Workdir {
    let cities = vec![
        svec!["city", "state"],
        svec!["Boston", "MA"],
        svec!["New York", "NY"],
        svec!["San Francisco", "CA"],
        svec!["Buffalo", "NY"],
    ];
    let places = vec![
        svec!["city", "place"],
        svec!["Boston", "Logan Airport"],
        svec!["Boston", "Boston Garden"],
        svec!["Buffalo", "Ralph Wilson Stadium"],
        svec!["Orlando", "Disney World"],
    ];

    let wrk = Workdir::new(name);
    wrk.create("cities.csv", cities);
    wrk.create("places.csv", places);
    wrk
}

fn make_rows(left_only: bool, rows: Vec<Vec<String>>) -> Vec<Vec<String>> {
    let mut all_rows = vec![];
    if left_only {
        all_rows.push(svec!["city", "state"]);
    } else {
        all_rows.push(svec!["city", "state", "city:places", "place"]);
    }
    all_rows.extend(rows);
    all_rows
}

sqlp_test!(
    sqlp_join_inner,
    |wrk: Workdir, mut cmd: process::Command| {
        cmd.arg("select * from cities inner join places on cities.city = places.city");
        let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
        let expected = make_rows(
            false,
            vec![
                svec!["Boston", "MA", "Boston", "Logan Airport"],
                svec!["Boston", "MA", "Boston", "Boston Garden"],
                svec!["Buffalo", "NY", "Buffalo", "Ralph Wilson Stadium"],
            ],
        );
        assert_eq!(got, expected);
    }
);

sqlp_test!(
    sqlp_right_join,
    |wrk: Workdir, mut cmd: process::Command| {
        cmd.arg("select * from cities right join places on cities.city = places.city");
        let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
        let expected = make_rows(
            false,
            vec![
                svec!["Boston", "MA", "Boston", "Logan Airport"],
                svec!["Boston", "MA", "Boston", "Boston Garden"],
                svec!["Buffalo", "NY", "Buffalo", "Ralph Wilson Stadium"],
                svec!["", "", "Orlando", "Disney World"],
            ],
        );
        assert_eq!(got, expected);
    }
);

sqlp_test!(
    sqlp_join_outer_left,
    |wrk: Workdir, mut cmd: process::Command| {
        cmd.arg("select * from cities left outer join places on cities.city = places.city");
        let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
        let expected = make_rows(
            false,
            vec![
                svec!["Boston", "MA", "Boston", "Logan Airport"],
                svec!["Boston", "MA", "Boston", "Boston Garden"],
                svec!["New York", "NY", "", ""],
                svec!["San Francisco", "CA", "", ""],
                svec!["Buffalo", "NY", "Buffalo", "Ralph Wilson Stadium"],
            ],
        );
        assert_eq!(got, expected);
    }
);

sqlp_test!(
    sqlp_join_outer_full,
    |wrk: Workdir, mut cmd: process::Command| {
        cmd.arg("select * from cities full outer join places on cities.city = places.city");
        let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
        let expected1 = vec![
            svec!["city", "state", "city:places", "place"],
            svec!["Boston", "MA", "Boston", "Logan Airport"],
            svec!["Boston", "MA", "Boston", "Boston Garden"],
            svec!["Buffalo", "NY", "Buffalo", "Ralph Wilson Stadium"],
            svec!["", "", "Orlando", "Disney World"],
            svec!["New York", "NY", "", ""],
            svec!["San Francisco", "CA", "", ""],
        ];
        let expected2 = vec![
            svec!["city", "state", "city:places", "place"],
            svec!["Boston", "MA", "Boston", "Logan Airport"],
            svec!["Boston", "MA", "Boston", "Boston Garden"],
            svec!["Buffalo", "NY", "Buffalo", "Ralph Wilson Stadium"],
            svec!["", "", "Orlando", "Disney World"],
            svec!["San Francisco", "CA", "", ""],
            svec!["New York", "NY", "", ""],
        ];
        assert!(got == expected1 || got == expected2);
    }
);

#[test]
fn sqlp_join_cross() {
    let wrk = Workdir::new("join_cross");
    wrk.create(
        "letters.csv",
        vec![svec!["h1", "h2"], svec!["a", "b"], svec!["c", "d"]],
    );
    wrk.create(
        "numbers.csv",
        vec![svec!["h3", "h4"], svec!["1", "2"], svec!["3", "4"]],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.args(["letters.csv", "numbers.csv"])
        .arg("select * from letters cross join numbers");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["h1", "h2", "h3", "h4"],
        svec!["a", "b", "1", "2"],
        svec!["a", "b", "3", "4"],
        svec!["c", "d", "1", "2"],
        svec!["c", "d", "3", "4"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_join_same_colname_1820() {
    let wrk = Workdir::new("sqlp_join_same_colname_1820");
    wrk.create("one.csv", vec![svec!["id", "data"], svec!["1", "open"]]);
    wrk.create("two.csv", vec![svec!["id", "data"], svec!["1", "closed"]]);

    let mut cmd = wrk.command("sqlp");
    cmd.args(["one.csv", "two.csv"])
        .arg("SELECT _t_1.id, _t_2.data FROM _t_1 JOIN _t_2 ON _t_1.id = _t_2.id");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![svec!["id", "data"], svec!["1", "closed"]];
    assert_eq!(got, expected);
}

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

    let mut cmd = wrk.command("sqlp");

    // we quote "boston311-100" as contains a hyphen in its name, which is a special character
    // in SQL, so we need to make it a quoted identifier
    cmd.arg(&test_file)
        .arg(r#"select ward, count(*) as cnt from "boston311-100" group by ward order by cnt desc, ward asc"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["ward", "cnt"],
        svec!["Ward 3", "10"],
        svec!["Ward 6", "7"],
        svec!["Ward 1", "6"],
        svec!["3", "5"],
        svec!["Ward 20", "5"],
        svec!["Ward 4", "5"],
        svec!["Ward 5", "5"],
        svec!["14", "4"],
        svec!["Ward 13", "4"],
        svec!["Ward 16", "4"],
        svec!["Ward 18", "3"],
        svec!["Ward 19", "3"],
        svec!["Ward 7", "3"],
        svec!["Ward 8", "3"],
        svec!["03", "2"],
        svec!["17", "2"],
        svec!["22", "2"],
        svec!["Ward 11", "2"],
        svec!["Ward 21", "2"],
        svec![" ", "1"],
        svec!["01", "1"],
        svec!["02", "1"],
        svec!["04", "1"],
        svec!["06", "1"],
        svec!["07", "1"],
        svec!["1", "1"],
        svec!["10", "1"],
        svec!["16", "1"],
        svec!["18", "1"],
        svec!["19", "1"],
        svec!["21", "1"],
        svec!["7", "1"],
        svec!["8", "1"],
        svec!["9", "1"],
        svec!["Ward 10", "1"],
        svec!["Ward 12", "1"],
        svec!["Ward 14", "1"],
        svec!["Ward 15", "1"],
        svec!["Ward 17", "1"],
        svec!["Ward 2", "1"],
        svec!["Ward 22", "1"],
        svec!["Ward 9", "1"],
    ];
    assert_eq!(got, expected);
}

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

    let mut cmd = wrk.command("sqlp");

    cmd.arg(&test_file)
        .arg(r#"select ward, count(*) as cnt from "boston311-100" group by ward order by all"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["ward", "cnt"],
        svec![" ", "1"],
        svec!["01", "1"],
        svec!["02", "1"],
        svec!["03", "2"],
        svec!["04", "1"],
        svec!["06", "1"],
        svec!["07", "1"],
        svec!["1", "1"],
        svec!["10", "1"],
        svec!["14", "4"],
        svec!["16", "1"],
        svec!["17", "2"],
        svec!["18", "1"],
        svec!["19", "1"],
        svec!["21", "1"],
        svec!["22", "2"],
        svec!["3", "5"],
        svec!["7", "1"],
        svec!["8", "1"],
        svec!["9", "1"],
        svec!["Ward 1", "6"],
        svec!["Ward 10", "1"],
        svec!["Ward 11", "2"],
        svec!["Ward 12", "1"],
        svec!["Ward 13", "4"],
        svec!["Ward 14", "1"],
        svec!["Ward 15", "1"],
        svec!["Ward 16", "4"],
        svec!["Ward 17", "1"],
        svec!["Ward 18", "3"],
        svec!["Ward 19", "3"],
        svec!["Ward 2", "1"],
        svec!["Ward 20", "5"],
        svec!["Ward 21", "2"],
        svec!["Ward 22", "1"],
        svec!["Ward 3", "10"],
        svec!["Ward 4", "5"],
        svec!["Ward 5", "5"],
        svec!["Ward 6", "7"],
        svec!["Ward 7", "3"],
        svec!["Ward 8", "3"],
        svec!["Ward 9", "1"],
    ];
    assert_eq!(got, expected);
}

// Verify the SQL-standard FILTER (WHERE …) aggregate modifier added in
// Polars PR https://github.com/pola-rs/polars/pull/27564 works through
// sqlp. The data + assertions mirror that PR's docstring example so the
// expected numbers are easy to cross-check: for group "aa" the
// unfiltered SUM(x) is 105.375 and SUM(x) FILTER (WHERE y > 250) is
// 44.875 (only the rows where y > 250 contribute); for "bb" the
// unfiltered SUM(x) is -43.75 and the filtered SUM is 6.5.
#[test]
fn sqlp_aggregate_filter() {
    let wrk = Workdir::new("sqlp_aggregate_filter");
    wrk.create(
        "data.csv",
        vec![
            svec!["grp", "x", "y"],
            svec!["aa", "60.5", "100"],
            svec!["bb", "-50.25", "200"],
            svec!["aa", "42.75", "300"],
            svec!["bb", "-3.5", "400"],
            svec!["aa", "2.125", "500"],
            svec!["bb", "10.0", "600"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv").arg(
        "SELECT grp, SUM(x) AS sum_x, SUM(x) FILTER (WHERE y > 250) AS sum_x_when_y_gt_250, \
         COUNT(*) FILTER (WHERE y > 250) AS n_y_gt_250 FROM data GROUP BY grp ORDER BY grp",
    );

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["grp", "sum_x", "sum_x_when_y_gt_250", "n_y_gt_250"],
        svec!["aa", "105.375", "44.875", "2"],
        svec!["bb", "-43.75", "6.5", "2"],
    ];
    assert_eq!(got, expected);
}

#[test]
// #[ignore = "temporarily disable due to a bug in polars aliasing"]
fn sqlp_boston311_groupby_orderby_with_table_alias() {
    let wrk = Workdir::new("sqlp_boston311_groupby_orderby");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("sqlp");

    // we use _t_1 alias as "boston311-100" contains a hyphen in its name, which is a special
    // character in SQL
    cmd.arg(&test_file)
        .arg("select ward, count(*) as cnt from _t_1 group by ward order by cnt desc, ward asc");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["ward", "cnt"],
        svec!["Ward 3", "10"],
        svec!["Ward 6", "7"],
        svec!["Ward 1", "6"],
        svec!["3", "5"],
        svec!["Ward 20", "5"],
        svec!["Ward 4", "5"],
        svec!["Ward 5", "5"],
        svec!["14", "4"],
        svec!["Ward 13", "4"],
        svec!["Ward 16", "4"],
        svec!["Ward 18", "3"],
        svec!["Ward 19", "3"],
        svec!["Ward 7", "3"],
        svec!["Ward 8", "3"],
        svec!["03", "2"],
        svec!["17", "2"],
        svec!["22", "2"],
        svec!["Ward 11", "2"],
        svec!["Ward 21", "2"],
        svec![" ", "1"],
        svec!["01", "1"],
        svec!["02", "1"],
        svec!["04", "1"],
        svec!["06", "1"],
        svec!["07", "1"],
        svec!["1", "1"],
        svec!["10", "1"],
        svec!["16", "1"],
        svec!["18", "1"],
        svec!["19", "1"],
        svec!["21", "1"],
        svec!["7", "1"],
        svec!["8", "1"],
        svec!["9", "1"],
        svec!["Ward 10", "1"],
        svec!["Ward 12", "1"],
        svec!["Ward 14", "1"],
        svec!["Ward 15", "1"],
        svec!["Ward 17", "1"],
        svec!["Ward 2", "1"],
        svec!["Ward 22", "1"],
        svec!["Ward 9", "1"],
    ];
    assert_eq!(got, expected);
}

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

    let mut cmd = wrk.command("sqlp");

    cmd.arg(&test_file)
        .args(["--wnull-value", "Not Specified"])
        .arg(
            "select location_street_name, location_zipcode from _t_1 where location_zipcode is \
             null order by location_street_name limit 5",
        );

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["location_street_name", "location_zipcode"],
        svec!["INTERSECTION Asticou Rd & Washington St", "Not Specified"],
        svec![
            "INTERSECTION Charles River Plz & Cambridge St",
            "Not Specified"
        ],
        svec!["INTERSECTION Columbia Rd & E Cottage St", "Not Specified"],
        svec!["INTERSECTION E Canton St & Albany St", "Not Specified"],
        svec![
            "INTERSECTION Gallivan Blvd & Washington St",
            "Not Specified"
        ],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_decimal_comma_issue_1050() {
    let wrk = Workdir::new("sqlp_decimal_comma_issue_1050");
    let test_file = wrk.load_test_file("progetti_sample_10.csv");

    let mut cmd = wrk.command("sqlp");

    cmd.arg(&test_file)
        .arg("--decimal-comma")
        .args(["--delimiter", ";"])
        .arg("select COD_LOCALE_PROGETTO, FINANZ_UE from _t_1");

    let got: String = wrk.stdout(&mut cmd);
    let expected = r#"COD_LOCALE_PROGETTO;FINANZ_UE
1AGCOE1627;6328008,44
1AGCOE2113;344203,2
1AGCOE645;491260,0
1AGCOE705;522811,32
1AGCOE706;460463,85
1ANPALINP-001;141566507,74
1ANPALINP-CLP-00002;47325000,0
1ANPALINP-CLP-00003;78450000,0
1ANPALINP-CLP-00004;67500000,0
1ANPALVAO1C001;6416,62"#;
    assert_eq!(got, expected);
}

#[test]
fn sqlp_null_aware_equality_checks() {
    let wrk = Workdir::new("sqlp_null_aware_equality_checks");
    wrk.create(
        "test_null.csv",
        vec![
            svec!["a", "b"],
            svec!["1", "1"],
            svec!["", ""],
            svec!["3", "3"],
            svec!["6", "4"],
            svec!["5", ""],
        ],
    );

    let mut cmd = wrk.command("sqlp");

    cmd.arg("test_null.csv")
        .args(["--wnull-value", "NULL"])
        .arg(
            r#"SELECT (a = b) as "1_eq_unaware",
           (a != b) as "2_neq_unaware", 
           (a <=> b) as "3_eq_aware", 
           (a IS NOT DISTINCT FROM b) as "4_eq_aware", 
           (a IS DISTINCT FROM b) as "5_neq_aware" 
         FROM test_null"#,
        );

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec![
            "1_eq_unaware",
            "2_neq_unaware",
            "3_eq_aware",
            "4_eq_aware",
            "5_neq_aware"
        ],
        svec!["true", "false", "true", "true", "false"],
        svec!["NULL", "NULL", "true", "true", "false"],
        svec!["true", "false", "true", "true", "false"],
        svec!["false", "true", "false", "false", "true"],
        svec!["NULL", "NULL", "false", "false", "true"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_rnull_values_wnull_value() {
    let wrk = Workdir::new("sqlp_rnull_values_wnull_value");
    wrk.create(
        "test_null.csv",
        vec![
            svec!["a", "b"],
            svec!["1", "Nothing"],
            svec!["2", "NA"],
            svec!["3", "Dunno"],
            svec!["4", "4"],
            svec!["5", ""],
            svec!("6", "6"),
            svec!("7", "DUNNO"),
        ],
    );

    let mut cmd = wrk.command("sqlp");

    cmd.arg("test_null.csv")
        .args(["--rnull-values", "Nothing,NA,<empty string>,Dunno"])
        .args(["--wnull-value", "NULL"])
        .arg("SELECT * FROM test_null");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["a", "b"],
        svec!["1", "NULL"],
        svec!["2", "NULL"],
        svec!["3", "NULL"],
        svec!["4", "4"],
        svec!["5", "NULL"],
        svec!["6", "6"],
        svec!["7", "DUNNO"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_regex_operators() {
    let wrk = Workdir::new("sqlp_regex_operators");
    wrk.create(
        "test_regex.csv",
        vec![
            svec!["n", "sval"],
            svec!["1", "ABC"],
            svec!["2", "abc"],
            svec!["3", "000"],
            svec!["4", "A0C"],
            svec!["5", "a0c"],
        ],
    );

    // ~ operator - contains pattern (case-sensitive)
    let mut cmd = wrk.command("sqlp");
    cmd.arg("test_regex.csv")
        .arg(r#"SELECT * FROM test_regex WHERE sval ~ '\d'"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["n", "sval"],
        svec!["3", "000"],
        svec!["4", "A0C"],
        svec!["5", "a0c"],
    ];
    assert_eq!(got, expected);

    // ~* operator - contains pattern (case-insensitive)
    let mut cmd = wrk.command("sqlp");
    cmd.arg("test_regex.csv")
        .arg(r#"SELECT * FROM test_regex WHERE sval ~* '^a0'"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![svec!["n", "sval"], svec!["4", "A0C"], svec!["5", "a0c"]];
    assert_eq!(got, expected);

    // !~ operator - does not contain pattern (case-sensitive)
    let mut cmd = wrk.command("sqlp");
    cmd.arg("test_regex.csv")
        .arg(r#"SELECT * FROM test_regex WHERE sval !~ '^a0'"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["n", "sval"],
        svec!["1", "ABC"],
        svec!["2", "abc"],
        svec!["3", "000"],
        svec!["4", "A0C"],
    ];
    assert_eq!(got, expected);

    // !~* operator - does not contain pattern (case-insensitive)
    let mut cmd = wrk.command("sqlp");
    cmd.arg("test_regex.csv")
        .arg(r#"SELECT * FROM test_regex WHERE sval !~* '^a0'"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["n", "sval"],
        svec!["1", "ABC"],
        svec!["2", "abc"],
        svec!["3", "000"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_regexp_like() {
    let wrk = Workdir::new("sqlp_regexp_like");
    wrk.create(
        "test_regexp_like.csv",
        vec![
            svec!["scol"],
            svec!["abcde"],
            svec!["abc"],
            svec!["a"],
            svec![""],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test_regexp_like.csv")
        .arg(r#"SELECT scol FROM test_regexp_like where REGEXP_LIKE(scol,'(C|D)','i')"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![svec!["scol"], svec!["abcde"], svec!["abc"]];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_string_functions() {
    let wrk = Workdir::new("sqlp_string_functions");
    wrk.create(
        "test_strings.csv",
        vec![
            svec!["scol"],
            svec!["abcdE"],
            svec!["abc"],
            svec!["    abc"],
            svec!["a"],
            svec!["b"],
        ],
    );

    // starts_with
    let mut cmd = wrk.command("sqlp");
    cmd.arg("test_strings.csv")
        .arg(r#"SELECT starts_with(scol, 'a') from test_strings"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["scol"],
        svec!["true"],
        svec!["true"],
        svec!["false"],
        svec!["true"],
        svec!["false"],
    ];
    assert_eq!(got, expected);

    // ends_with
    let mut cmd = wrk.command("sqlp");
    cmd.arg("test_strings.csv")
        .arg(r#"SELECT ends_with(scol, 'c') from test_strings"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["scol"],
        svec!["false"],
        svec!["true"],
        svec!["true"],
        svec!["false"],
        svec!["false"],
    ];
    assert_eq!(got, expected);

    // left
    let mut cmd = wrk.command("sqlp");
    cmd.arg("test_strings.csv")
        .arg(r#"SELECT left(scol, 3) from test_strings"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["scol"],
        svec!["abc"],
        svec!["abc"],
        svec!["   "],
        svec!["a"],
        svec!["b"],
    ];
    assert_eq!(got, expected);

    // substr
    let mut cmd = wrk.command("sqlp");
    cmd.arg("test_strings.csv")
        .arg(r#"SELECT substr(scol, 2, 2) from test_strings"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["scol"],
        svec!["bc"],
        svec!["bc"],
        svec!["  "],
        svec![""],
        svec![""],
    ];
    assert_eq!(got, expected);

    // upper
    let mut cmd = wrk.command("sqlp");
    cmd.arg("test_strings.csv")
        .arg(r#"SELECT upper(scol) from test_strings"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["scol"],
        svec!["ABCDE"],
        svec!["ABC"],
        svec!["    ABC"],
        svec!["A"],
        svec!["B"],
    ];
    assert_eq!(got, expected);

    // lower
    let mut cmd = wrk.command("sqlp");
    cmd.arg("test_strings.csv")
        .arg(r#"SELECT lower(scol) from test_strings"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["scol"],
        svec!["abcde"],
        svec!["abc"],
        svec!["    abc"],
        svec!["a"],
        svec!["b"],
    ];
    assert_eq!(got, expected);

    // length
    let mut cmd = wrk.command("sqlp");
    cmd.arg("test_strings.csv")
        .arg(r#"SELECT length(scol) from test_strings"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["scol"],
        svec!["5"],
        svec!["3"],
        svec!["7"],
        svec!["1"],
        svec!["1"],
    ];
    assert_eq!(got, expected);

    // octet_length
    let mut cmd = wrk.command("sqlp");
    cmd.arg("test_strings.csv")
        .arg(r#"SELECT octet_length(scol) from test_strings"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["scol"],
        svec!["5"],
        svec!["3"],
        svec!["7"],
        svec!["1"],
        svec!["1"],
    ];
    assert_eq!(got, expected);
}

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file)
        .arg("--try-parsedates")
        .arg(
            "select ward, cast(avg(closed_dt - open_dt) as float) as avg_tat from _t_1 where \
             case_status = 'Closed' group by ward order by avg_tat desc, ward asc",
        )
        .arg("--ignore-errors");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["ward", "avg_tat"],
        svec!["Ward 11", "4847760000000.0"],
        svec!["01", "4818270000000.0"],
        svec!["Ward 13", "1518365750000.0"],
        svec!["Ward 15", "1278926000000.0"],
        svec!["Ward 21", "878446000000.0"],
        svec!["Ward 14", "618933000000.0"],
        svec!["Ward 3", "437691444444.0"],
        svec!["Ward 5", "411909500000.0"],
        svec!["Ward 20", "367233000000.0"],
        svec!["9", "353495000000.0"],
        svec!["Ward 18", "249882000000.0"],
        svec!["19", "212566000000.0"],
        svec!["Ward 4", "112872600000.0"],
        svec!["Ward 1", "107850666666.0"],
        svec!["Ward 10", "104110000000.0"],
        svec!["16", "93557000000.0"],
        svec!["Ward 19", "84164000000.0"],
        svec!["10", "79101000000.0"],
        svec!["21", "77717000000.0"],
        svec!["7", "74611000000.0"],
        svec!["17", "70117500000.0"],
        svec!["3", "68836600000.0"],
        svec!["Ward 9", "64097000000.0"],
        svec!["Ward 12", "62930000000.0"],
        svec!["Ward 6", "54770166666.0"],
        svec!["Ward 7", "38346333333.0"],
        svec!["Ward 8", "32767500000.0"],
        svec!["03", "29810500000.0"],
        svec!["07", "25328000000.0"],
        svec!["22", "23919000000.0"],
        svec!["14", "20786500000.0"],
        svec!["Ward 22", "13524000000.0"],
        svec!["1", "9469000000.0"],
        svec!["06", "5290000000.0"],
        svec!["Ward 16", "4533666666.0"],
        svec!["8", "1757000000.0"],
        svec!["02", "1650000000.0"],
        svec!["18", "507000000.0"],
    ];
    assert_eq!(got, expected);
}

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file)
        .arg("--try-parsedates")
        .args(["--float-precision", "3"])
        .arg(
            "select ward, cast(avg(closed_dt - open_dt) as float) as avg_tat from _t_1 where \
             case_status = 'Closed' group by ward order by avg_tat desc, ward asc limit 5",
        )
        .arg("--ignore-errors");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["ward", "avg_tat"],
        svec!["Ward 11", "4847760000000.000"],
        svec!["01", "4818270000000.000"],
        svec!["Ward 13", "1518365750000.000"],
        svec!["Ward 15", "1278926000000.000"],
        svec!["Ward 21", "878446000000.000"],
    ];
    assert_eq!(got, expected);
}

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file)
        .arg("--try-parsedates")
        .args(["--datetime-format", "%a %Y-%m-%d %H:%M:%S"])
        .arg("select closed_dt, open_dt from _t_1 where case_status = 'Closed' limit 5")
        .arg("--ignore-errors");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["closed_dt", "open_dt"],
        svec!["Wed 2022-01-19 11:42:16", "Wed 2022-01-19 11:18:00"],
        svec!["Sun 2022-01-09 06:43:06", "Sat 2022-01-08 12:54:49"],
        svec!["Mon 2022-01-10 08:42:23", "Sat 2022-01-01 00:16:00"],
        svec!["Thu 2022-01-20 08:45:03", "Thu 2022-01-20 08:07:49"],
        svec!["Thu 2022-01-20 08:45:12", "Thu 2022-01-20 08:15:45"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_comments() {
    let wrk = Workdir::new("sqlp_comments");
    wrk.create(
        "comments.csv",
        vec![
            svec!["# test file to see how comments work", ""],
            svec!["# this is another comment before the header", ""],
            svec!["# DATA DICTIONARY", ""],
            svec!["# column1 - alphabetic; id of the column", ""],
            svec!["# column2 - numeric; just a number", ""],
            svec!["column1", "column2"],
            svec!["a", "1"],
            svec!["#b", "2"],
            svec!["c", "3"],
            svec!["#d - this row is corrupted skip", "extra col2"],
            svec!["e", "5"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.env("QSV_COMMENT_CHAR", "#");
    cmd.arg("comments.csv")
        .arg("select column1, column2 from comments order by column2 desc");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["column1", "column2"],
        svec!["e", "5"],
        svec!["c", "3"],
        svec!["a", "1"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_compress() {
    let wrk = Workdir::new("sqlp_compress");
    wrk.create(
        "data.csv",
        vec![
            svec!["column1", "column2"],
            svec!["a", "1"],
            svec!["c", "3"],
            svec!["e", "5"],
        ],
    );

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg("select column1, column2 from data order by column2 desc")
        .args(["-o", &out_file]);

    wrk.assert_success(&mut cmd);

    let mut cmd_2 = wrk.command("snappy"); // DevSkim: ignore DS126858
    cmd_2.arg("decompress").arg(out_file); // DevSkim: ignore DS126858

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd_2); // DevSkim: ignore DS126858
    let expected = vec![
        svec!["column1", "column2"],
        svec!["e", "5"],
        svec!["c", "3"],
        svec!["a", "1"],
    ];

    assert_eq!(got, expected);
}

#[test]
// #[ignore = "temporarily disable due to a bug in polars aliasing"]
fn sqlp_boston311_sql_script() {
    let wrk = Workdir::new("sqlp_boston311_sql_script");
    let test_file = wrk.load_test_file("boston311-100.csv");

    wrk.create_from_string(
        "test.sql",
        r#"create table temp_table as select * from "boston311-100" where ontime = 'OVERDUE';
create table temp_table2 as select * from temp_table limit 10;
-- we already got what we needed from temp_table into temp_table2
-- so we can truncate temp_table. Otherwise, the memory taken by temp_table
-- won't be released until the end of the script
truncate temp_table;
select ward,count(*) as cnt from temp_table2 group by ward order by cnt desc, ward asc;"#,
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file).arg("test.sql");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["ward", "cnt"],
        svec!["Ward 3", "2"],
        svec![" ", "1"],
        svec!["04", "1"],
        svec!["3", "1"],
        svec!["Ward 13", "1"],
        svec!["Ward 17", "1"],
        svec!["Ward 19", "1"],
        svec!["Ward 21", "1"],
        svec!["Ward 6", "1"],
    ];

    assert_eq!(got, expected);
}

#[test]
// #[ignore = "temporarily disable due to a bug in polars aliasing"]
fn sqlp_boston311_sql_script_json() {
    let wrk = Workdir::new("sqlp_boston311_sql_script_json");
    let test_file = wrk.load_test_file("boston311-100.csv");

    wrk.create_from_string(
        "test.sql",
        r#"create table temp_table as select * from "boston311-100" where ontime = 'OVERDUE';
create table temp_table2 as select * from temp_table limit 10;
select ward,count(*) as cnt from temp_table2 group by ward order by cnt desc, ward asc;"#,
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file)
        .arg("test.sql")
        .args(["--format", "json"]);

    let got: String = wrk.stdout(&mut cmd);
    let expected = r#"[{"ward":"Ward 3","cnt":2},{"ward":" ","cnt":1},{"ward":"04","cnt":1},{"ward":"3","cnt":1},{"ward":"Ward 13","cnt":1},{"ward":"Ward 17","cnt":1},{"ward":"Ward 19","cnt":1},{"ward":"Ward 21","cnt":1},{"ward":"Ward 6","cnt":1}]"#;

    assert_eq!(got, expected);
}

#[test]
// #[ignore = "temporarily disable due to a bug in polars aliasing"]
fn sqlp_boston311_sql_script_jsonl() {
    let wrk = Workdir::new("sqlp_boston311_sql_script_jsonl");
    let test_file = wrk.load_test_file("boston311-100.csv");

    wrk.create_from_string(
        "test.sql",
        r#"create table temp_table as select * from "boston311-100" where ontime = 'OVERDUE';
create table temp_table2 as select * from temp_table limit 10;
select ward,count(*) as cnt from temp_table2 group by ward order by cnt desc, ward asc;"#,
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file)
        .arg("test.sql")
        .args(["--format", "jsonl"]);

    let got: String = wrk.stdout(&mut cmd);
    let expected = r#"{"ward":"Ward 3","cnt":2}
{"ward":" ","cnt":1}
{"ward":"04","cnt":1}
{"ward":"3","cnt":1}
{"ward":"Ward 13","cnt":1}
{"ward":"Ward 17","cnt":1}
{"ward":"Ward 19","cnt":1}
{"ward":"Ward 21","cnt":1}
{"ward":"Ward 6","cnt":1}"#;

    assert_eq!(got, expected);
}

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

    wrk.create_from_string(
        "test.sql",
        r#"create table temp_table as select * from "boston311-100" where ontime = 'OVERDUE';
create table temp_table2 as select * from temp_table limit 10;
select ward,count(*) as cnt from temp_table2 group by ward order by cnt desc, ward asc;"#,
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file)
        .arg("test.sql")
        .args(["--format", "jsonl", "--cache-schema"]);

    let got: String = wrk.stdout(&mut cmd);
    let expected = r#"{"ward":"Ward 3","cnt":2}
{"ward":" ","cnt":1}
{"ward":"04","cnt":1}
{"ward":"3","cnt":1}
{"ward":"Ward 13","cnt":1}
{"ward":"Ward 17","cnt":1}
{"ward":"Ward 19","cnt":1}
{"ward":"Ward 21","cnt":1}
{"ward":"Ward 6","cnt":1}"#;

    assert_eq!(got, expected);
    assert!(wrk.path("boston311-100.csv.pschema.json").exists());
    let boston311_schema =
        std::fs::read_to_string(wrk.path("boston311-100.csv.pschema.json")).unwrap();
    assert_eq!(
        boston311_schema,
        r#"{
  "fields": {
    "case_enquiry_id": "UInt64",
    "open_dt": {
      "Datetime": [
        "Milliseconds",
        null
      ]
    },
    "target_dt": {
      "Datetime": [
        "Milliseconds",
        null
      ]
    },
    "closed_dt": {
      "Datetime": [
        "Milliseconds",
        null
      ]
    },
    "ontime": "String",
    "case_status": "String",
    "closure_reason": "String",
    "case_title": "String",
    "subject": "String",
    "reason": "String",
    "type": "String",
    "queue": "String",
    "department": "String",
    "submittedphoto": "String",
    "closedphoto": "String",
    "location": "String",
    "fire_district": "String",
    "pwd_district": "String",
    "city_council_district": "String",
    "police_district": "String",
    "neighborhood": "String",
    "neighborhood_services_district": "String",
    "ward": "String",
    "precinct": "String",
    "location_street_name": "String",
    "location_zipcode": "String",
    "latitude": "Float32",
    "longitude": "Float32",
    "source": "String"
  },
  "metadata": null
}"#
    );
}

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

    wrk.create_from_string(
        "test.sql",
        "select latitude,longitude from 'boston311-100' limit 10",
    );

    wrk.create_from_string(
        "boston311-100.csv.pschema.json",
        r#"{
        "fields": {
          "case_enquiry_id": "Int64",
          "open_dt": "String",
          "target_dt": "String",
          "closed_dt": "String",
          "ontime": "String",
          "case_status": "String",
          "closure_reason": "String",
          "case_title": "String",
          "subject": "String",
          "reason": "String",
          "type": "String",
          "queue": "String",
          "department": "String",
          "submittedphoto": "String",
          "closedphoto": "String",
          "location": "String",
          "fire_district": "String",
          "pwd_district": "String",
          "city_council_district": "String",
          "police_district": "String",
          "neighborhood": "String",
          "neighborhood_services_district": "String",
          "ward": "String",
          "precinct": "String",
          "location_street_name": "String",
          "location_zipcode": "String",
          "latitude": {"Decimal" : [10, 3]},
          "longitude": {"Decimal" : [10, 6]},
          "source": "String"
        },"metadata": null
      }"#,
    );

    assert!(wrk.path("boston311-100.csv").exists());
    assert!(wrk.path("boston311-100.csv.pschema.json").exists());

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file)
        .arg("test.sql")
        .args(["--format", "jsonl", "--cache-schema"]);

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);
    let expected = r#"{"latitude":"42.359","longitude":"-71.058700"}
{"latitude":"42.363","longitude":"-71.056600"}
{"latitude":"42.288","longitude":"-71.133000"}
{"latitude":"42.340","longitude":"-71.080300"}
{"latitude":"42.374","longitude":"-71.059900"}
{"latitude":"42.359","longitude":"-71.070000"}
{"latitude":"42.312","longitude":"-71.115200"}
{"latitude":"42.361","longitude":"-71.063800"}
{"latitude":"42.360","longitude":"-71.063400"}
{"latitude":"42.349","longitude":"-71.081100"}"#;

    assert_eq!(got, expected);
}

#[test]
// #[ignore = "temporarily disable due to a bug in polars aliasing"]
fn sqlp_boston311_cte_script() {
    let wrk = Workdir::new("sqlp_boston311_cte");
    let test_file = wrk.load_test_file("boston311-100.csv");

    wrk.create_from_string(
        "test.sql",
        r#"with boston311_roxbury as (select * from "boston311-100" where neighborhood = 'Roxbury')
select ward,count(*) as cnt from boston311_roxbury group by ward order by cnt desc, ward asc;"#,
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file).arg("test.sql");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["ward", "cnt"],
        svec!["Ward 11", "2"],
        svec!["Ward 13", "2"],
        svec!["Ward 8", "2"],
        svec!["14", "1"],
        svec!["Ward 12", "1"],
    ];

    assert_eq!(got, expected);
}

#[test]
// #[ignore = "temporarily disable due to a bug in polars aliasing"]
fn sqlp_boston311_cte() {
    let wrk = Workdir::new("sqlp_boston311_cte");
    let test_file = wrk.load_test_file("boston311-100.csv");

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file).arg(
        r#"with boston311_roxbury as (select * from "boston311-100" where neighborhood = 'Roxbury')
    select ward,count(*) as cnt from boston311_roxbury group by ward order by cnt desc, ward asc;"#,
    );

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["ward", "cnt"],
        svec!["Ward 11", "2"],
        svec!["Ward 13", "2"],
        svec!["Ward 8", "2"],
        svec!["14", "1"],
        svec!["Ward 12", "1"],
    ];

    assert_eq!(got, expected);
}

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file).arg(
        r#"with boston311_roxbury as (select * from read_csv('boston311-100.csv.gz') where neighborhood = 'Roxbury')
    select ward,count(*) as cnt from boston311_roxbury group by ward order by cnt desc, ward asc;"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["ward", "cnt"],
        svec!["Ward 11", "2"],
        svec!["Ward 13", "2"],
        svec!["Ward 8", "2"],
        svec!["14", "1"],
        svec!["Ward 12", "1"],
    ];

    assert_eq!(got, expected);
}

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file).arg(
        r#"SELECT case_enquiry_id, 
           CASE closed_dt is null and case_title ~* 'graffiti' 
              WHEN True THEN 'Yes' 
              WHEN False THEN 'No' 
              ELSE 'N/A'
           END as graffiti_related
           from _t_1
           where case_status = 'Open'"#,
    );

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["case_enquiry_id", "graffiti_related"],
        svec!["101004143000", "No"],
        svec!["101004155594", "No"],
        svec!["101004154423", "No"],
        svec!["101004141848", "No"],
        svec!["101004113313", "No"],
        svec!["101004113751", "Yes"],
        svec!["101004113902", "Yes"],
        svec!["101004113473", "No"],
        svec!["101004113604", "No"],
        svec!["101004114154", "Yes"],
        svec!["101004114383", "No"],
        svec!["101004114795", "Yes"],
        svec!["101004118346", "Yes"],
        svec!["101004115302", "No"],
        svec!["101004115066", "No"],
    ];

    assert_eq!(got, expected);
}

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file).arg(
        r#"SELECT case_enquiry_id, 
           CASE closed_dt is null and case_title ~* 'graffiti' 
              WHEN True THEN 'Yes' 
              WHEN False THEN 'No' 
              ELSE 'N/A'
           END as graffiti_related
           from read_csv('boston311-100.csv.zlib')
           where case_status = 'Open'"#,
    );

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["case_enquiry_id", "graffiti_related"],
        svec!["101004143000", "No"],
        svec!["101004155594", "No"],
        svec!["101004154423", "No"],
        svec!["101004141848", "No"],
        svec!["101004113313", "No"],
        svec!["101004113751", "Yes"],
        svec!["101004113902", "Yes"],
        svec!["101004113473", "No"],
        svec!["101004113604", "No"],
        svec!["101004114154", "Yes"],
        svec!["101004114383", "No"],
        svec!["101004114795", "Yes"],
        svec!["101004118346", "Yes"],
        svec!["101004115302", "No"],
        svec!["101004115066", "No"],
    ];

    assert_eq!(got, expected);
}

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file)
        .arg(
            r#"SELECT case_enquiry_id, 
           CASE closed_dt is null and case_title ~* 'graffiti' 
              WHEN True THEN 'Yes' 
              WHEN False THEN 'No' 
              ELSE 'N/A'
           END as graffiti_related
           from _t_1
           where case_status = 'Open'"#,
        )
        .arg("--streaming");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["case_enquiry_id", "graffiti_related"],
        svec!["101004143000", "No"],
        svec!["101004155594", "No"],
        svec!["101004154423", "No"],
        svec!["101004141848", "No"],
        svec!["101004113313", "No"],
        svec!["101004113751", "Yes"],
        svec!["101004113902", "Yes"],
        svec!["101004113473", "No"],
        svec!["101004113604", "No"],
        svec!["101004114154", "Yes"],
        svec!["101004114383", "No"],
        svec!["101004114795", "Yes"],
        svec!["101004118346", "Yes"],
        svec!["101004115302", "No"],
        svec!["101004115066", "No"],
    ];

    assert_eq!(got, expected);
}

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file).arg(
        r#"SELECT case_enquiry_id, 
           CASE 
              WHEN case_title ~* 'graffiti' THEN 'Graffitti' 
              WHEN case_title ~* 'vehicle' THEN 'Vehicle'
              WHEN case_title ~* 'sidewalk' THEN 'Sidewalk'
              ELSE 'Something else'
           END as topic
           from _t_1
           where case_status = 'Open'"#,
    );

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["case_enquiry_id", "topic"],
        svec!["101004143000", "Something else"],
        svec!["101004155594", "Something else"],
        svec!["101004154423", "Sidewalk"],
        svec!["101004141848", "Something else"],
        svec!["101004113313", "Something else"],
        svec!["101004113751", "Graffitti"],
        svec!["101004113902", "Graffitti"],
        svec!["101004113473", "Sidewalk"],
        svec!["101004113604", "Something else"],
        svec!["101004114154", "Graffitti"],
        svec!["101004114383", "Something else"],
        svec!["101004114795", "Graffitti"],
        svec!["101004118346", "Graffitti"],
        svec!["101004115302", "Vehicle"],
        svec!["101004115066", "Sidewalk"],
    ];

    assert_eq!(got, expected);
}

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_file).arg(
        r#"SELECT case_enquiry_id, 
           CASE 
              WHEN case_title ~* 'graffiti' THEN 'Graffitti' 
              WHEN case_title ~* 'vehicle' THEN 'Vehicle'
              WHEN case_title ~* 'sidewalk' THEN 'Sidewalk'
              ELSE 'Something else'
           END as topic
           from read_csv('boston311-100.csv.zst')
           where case_status = 'Open'"#,
    );

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["case_enquiry_id", "topic"],
        svec!["101004143000", "Something else"],
        svec!["101004155594", "Something else"],
        svec!["101004154423", "Sidewalk"],
        svec!["101004141848", "Something else"],
        svec!["101004113313", "Something else"],
        svec!["101004113751", "Graffitti"],
        svec!["101004113902", "Graffitti"],
        svec!["101004113473", "Sidewalk"],
        svec!["101004113604", "Something else"],
        svec!["101004114154", "Graffitti"],
        svec!["101004114383", "Something else"],
        svec!["101004114795", "Graffitti"],
        svec!["101004118346", "Graffitti"],
        svec!["101004115302", "Vehicle"],
        svec!["101004115066", "Sidewalk"],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_literal_pattern_match() {
    let wrk = Workdir::new("sqlp_literal_pattern_match");
    wrk.create(
        "test.csv",
        vec![
            svec!["idx", "val"],
            svec!["0", "ABC"],
            svec!["1", "abc"],
            svec!["2", "000"],
            svec!["3", "A0C"],
            svec!["4", "a0c"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv")
        .arg(r#"SELECT * FROM test WHERE val NOT REGEXP '.*c$'"#);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["idx", "val"],
        svec!["0", "ABC"],
        svec!["2", "000"],
        svec!["3", "A0C"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_expression_pattern_match() {
    let wrk = Workdir::new("sqlp_expression_pattern_match");
    wrk.create(
        "test.csv",
        vec![
            svec!["idx", "val", "pat"],
            svec!["0", "ABC", "^A"],
            svec!["1", "abc", "^A"],
            svec!["2", "000", "^A"],
            svec!["3", "A0C", r#"[AB]\d.*$"#,],
            svec!["4", "a0c", ".*xxx$"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv")
        .arg("SELECT idx, val FROM test WHERE val REGEXP pat");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![svec!["idx", "val"], svec!["0", "ABC"], svec!["3", "A0C"]];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_sql_join_on_subquery() {
    let wrk = Workdir::new("sqlp_sql_join_on_subquery");
    wrk.create(
        "test.csv",
        vec![
            svec!["idx", "val"],
            svec!["0", "ABC"],
            svec!["1", "abc"],
            svec!["2", "000"],
            svec!["3", "A0C"],
            svec!["4", "a0c"],
        ],
    );

    wrk.create(
        "test2.csv",
        vec![
            svec!["idx", "val"],
            svec!["0", "ABC"],
            svec!["1", "abc"],
            svec!["2", "000"],
            svec!["3", "A0C"],
            svec!["4", "a0c"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv").arg("test2.csv").arg(
        "SELECT * FROM test t1 JOIN (SELECT idx, val FROM test2 WHERE idx > 2) t2 ON t1.idx = \
         t2.idx",
    );

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["idx", "val", "idx:t2", "val:t2"],
        svec!["3", "A0C", "3", "A0C"],
        svec!["4", "a0c", "4", "a0c"],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_sql_join_on_literal_string_comparison() {
    let wrk = Workdir::new("sqlp_sql_join_on_literal_string_comparison");
    wrk.create(
        "test.csv",
        vec![
            svec!["name", "role"],
            svec!["alice", "admin"],
            svec!["bob", "user"],
            svec!["adam", "admin"],
            svec!["charlie", "user"],
        ],
    );

    wrk.create(
        "test2.csv",
        vec![
            svec!["name", "dept"],
            svec!["alice", "IT"],
            svec!["bob", "HR"],
            svec!["charlie", "IT"],
            svec!["adam", "SEC"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv").arg("test2.csv").arg(
        "SELECT t1.name, t1.role, t2.dept FROM test t1 INNER JOIN test2 t2 ON t1.name = t2.name \
         AND t1.role = 'admin' ORDER BY t1.name",
    );

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["name", "role", "dept"],
        svec!["adam", "admin", "SEC"],
        svec!["alice", "admin", "IT"],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_sql_join_on_expression_comparison() {
    let wrk = Workdir::new("sqlp_sql_join_on_expression_comparison");
    wrk.create(
        "test.csv",
        vec![
            svec!["code", "value"],
            svec!["HELLO", "100"],
            svec!["WORLD", "200"],
            svec!["FOO", "300"],
        ],
    );

    wrk.create(
        "test2.csv",
        vec![
            svec!["code", "value"],
            svec!["hello", "-1.2345"],
            svec!["world", "4.5678"],
            svec!["bar", "2.2222"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv")
        .arg("test2.csv")
        .arg(
            "SELECT t1.code AS code1, t2.code AS code2, (t1.value * t2.value) AS VAL FROM test t1 \
             INNER JOIN test2 t2 ON LOWER(t1.code) = t2.code",
        )
        .args(["--float-precision", "2"]);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["code1", "code2", "VAL"],
        svec!["HELLO", "hello", "-123.45"],
        svec!["WORLD", "world", "913.56"],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_sql_from_subquery() {
    let wrk = Workdir::new("sqlp_sql_from_subquery");
    wrk.create(
        "test.csv",
        vec![
            svec!["idx", "val"],
            svec!["0", "ABC"],
            svec!["1", "abc"],
            svec!["2", "000"],
            svec!["3", "A0C"],
            svec!["4", "a0c"],
        ],
    );

    wrk.create(
        "test2.csv",
        vec![
            svec!["idx", "val"],
            svec!["0", "ABC"],
            svec!["1", "abc"],
            svec!["2", "000"],
            svec!["3", "A0C"],
            svec!["4", "a0c"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv").arg("test2.csv").arg(
        "SELECT * FROM (SELECT idx, val FROM test WHERE idx > 2) t1 JOIN test2 t2 ON t1.idx = \
         t2.idx",
    );

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["idx", "val", "idx:t2", "val:t2"],
        svec!["3", "A0C", "3", "A0C"],
        svec!["4", "a0c", "4", "a0c"],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_sql_tsv() {
    let wrk = Workdir::new("sqlp_sql_tsv");
    wrk.create(
        "test.csv",
        vec![
            svec!["idx", "val"],
            svec!["0", "ABC"],
            svec!["1", "abc"],
            svec!["2", "000"],
            svec!["3", "A0C"],
            svec!["4", "a0c"],
        ],
    );

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv")
        .arg("SELECT * FROM test")
        .args(["--output", &output_file]);

    wrk.assert_success(&mut cmd);

    let got = wrk.read_to_string(&output_file).unwrap();

    let expected = "idx\tval\n0\tABC\n1\tabc\n2\t000\n3\tA0C\n4\ta0c\n";

    assert_eq!(got, expected);
}

#[test]
fn sqlp_sql_ssv() {
    let wrk = Workdir::new("sqlp_sql_ssv");
    wrk.create(
        "test.csv",
        vec![
            svec!["idx", "val"],
            svec!["0", "ABC"],
            svec!["1", "abc"],
            svec!["2", "000"],
            svec!["3", "A0C"],
            svec!["4", "a0c"],
        ],
    );

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv")
        .arg("SELECT * FROM test")
        .args(["--output", &output_file]);

    wrk.assert_success(&mut cmd);

    let got = wrk.read_to_string(&output_file).unwrap();

    let expected = "idx;val\n0;ABC\n1;abc\n2;000\n3;A0C\n4;a0c\n";

    assert_eq!(got, expected);
}

#[test]
fn sqlp_sql_ssv_output() {
    let wrk = Workdir::new("sqlp_sql_ssv_output");
    wrk.create_with_delim(
        "test.ssv",
        vec![
            svec!["idx", "val"],
            svec!["0", "ABC"],
            svec!["1", "abc"],
            svec!["2", "000"],
            svec!["3", "A0C"],
            svec!["4", "a0c"],
        ],
        b';',
    );

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.ssv")
        .arg("SELECT * FROM test")
        .args(["--output", &output_file]);

    wrk.assert_success(&mut cmd);

    let got = wrk.read_to_string(&output_file).unwrap();

    let expected = "idx;val\n0;ABC\n1;abc\n2;000\n3;A0C\n4;a0c\n";

    assert_eq!(got, expected);
}

#[test]
fn sqlp_issue2014() {
    let wrk = Workdir::new("sqlp_issue2014");
    wrk.create_with_delim(
        "test.ssv",
        vec![
            svec!["id", "item", "price"],
            svec!["0", "wallet", "9.99"],
            svec!["1", "comb", "1.39"],
            svec!["2", "pencil", "0.49"],
        ],
        b';',
    );

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.ssv")
        .arg("SELECT * FROM test")
        .args(["--output", &output_file]);

    wrk.assert_success(&mut cmd);

    let mut cmd = wrk.command("snappy");
    cmd.arg("decompress").arg(output_file.clone());
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        ["id;item;price"],
        ["0;wallet;9.99"],
        ["1;comb;1.39"],
        ["2;pencil;0.49"],
    ];
    assert_eq!(got, expected);

    let mut cmd = wrk.command("slice");
    cmd.arg(output_file.clone());

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["id", "item", "price"],
        svec!["0", "wallet", "9.99"],
        svec!["1", "comb", "1.39"],
        svec!["2", "pencil", "0.49"],
    ];

    assert_eq!(got, expected);

    let mut cmd = wrk.command("headers");
    cmd.arg(output_file);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![["1   id"], ["2   item"], ["3   price"]];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_binary_functions() {
    let wrk = Workdir::new("sqlp_sql_binary_functions");
    wrk.create("dummy.csv", vec![svec!["dummy"], svec!["0"]]);

    let mut cmd = wrk.command("sqlp");
    cmd.arg("dummy.csv")
        .arg(
            r#"
        SELECT *,
          -- bit strings
          b''                 AS b0,
          b'1001'             AS b1,
          b'11101011'         AS b2,
          b'1111110100110010' AS b3,
          -- hex strings
          x''                 AS x0,
          x'FF'               AS x1,
          x'4142'             AS x2,
          x'DeadBeef'         AS x3,
        FROM dummy
        "#,
        )
        .args(["--format", "parquet"]);

    wrk.assert_success(&mut cmd);
}

#[test]

fn sqlp_length_fns() {
    let wrk = Workdir::new("sqlp_sql_length_fns");
    wrk.create(
        "test.csv",
        vec![svec!["words"], svec!["Cafe"], svec![""], svec!["東京"]],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv").arg(
        r#"
        SELECT
              words,
              LENGTH(words) AS n_chrs1,
              CHAR_LENGTH(words) AS n_chrs2,
              CHARACTER_LENGTH(words) AS n_chrs3,
              OCTET_LENGTH(words) AS n_bytes,
              BIT_LENGTH(words) AS n_bits
            FROM test
"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec![
            "words", "n_chrs1", "n_chrs2", "n_chrs3", "n_bytes", "n_bits"
        ],
        svec!["Cafe", "4", "4", "4", "4", "32"],
        // as the rnull-values is not set, the empty string is not converted to NULL
        // so the output is all empty strings
        svec!["", "", "", "", "", ""],
        svec!["東京", "2", "2", "2", "6", "48"],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_length_fns_rnull_set_to_null() {
    let wrk = Workdir::new("sqlp_sql_length_fns_rnull_set_to_null");
    wrk.create(
        "test.csv",
        vec![svec!["words"], svec!["Cafe"], svec![""], svec!["東京"]],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv")
        .arg(
            r#"
        SELECT
              words,
              LENGTH(words) AS n_chrs1,
              CHAR_LENGTH(words) AS n_chrs2,
              CHARACTER_LENGTH(words) AS n_chrs3,
              OCTET_LENGTH(words) AS n_bytes,
              BIT_LENGTH(words) AS n_bits
            FROM test
"#,
        )
        .args(["--rnull-values", "NULL"]);

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec![
            "words", "n_chrs1", "n_chrs2", "n_chrs3", "n_bytes", "n_bits"
        ],
        svec!["Cafe", "4", "4", "4", "4", "32"],
        // as the rnull-values is set to NULL, the empty string is converted to NULL
        // so the output of all the length functions is zero
        svec!["", "0", "0", "0", "0", "0"],
        svec!["東京", "2", "2", "2", "6", "48"],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_control_flow() {
    let wrk = Workdir::new("sqlp_control_flow");
    wrk.create(
        "test.csv",
        vec![
            svec!["x", "y", "z"],
            svec!["1", "5", "3"],
            svec!["", "4", "4"],
            svec!["2", "", ""],
            svec!["3", "3", "3"],
            svec!["", "", "6"],
            svec!["4", "2", ""],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv").arg(
        r#"
        SELECT
          COALESCE(x,y,z) as "coalsc",
          NULLIF(x, y) as "nullif x_y",
          NULLIF(y, z) as "nullif y_z",
          IFNULL(x, y) as "ifnull x_y",
          IFNULL(y,-1) as "inullf y_z",
          COALESCE(x, NULLIF(y,z)) as "both",
          IF(x = y, 'eq', 'ne') as "x_eq_y",
        FROM test
"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec![
            "coalsc",
            "nullif x_y",
            "nullif y_z",
            "ifnull x_y",
            "inullf y_z",
            "both",
            "x_eq_y"
        ],
        svec!["1", "1", "5", "1", "5", "1", "ne"],
        svec!["4", "", "", "4", "4", "", "ne"],
        svec!["2", "2", "", "2", "-1", "2", "ne"],
        svec!["3", "", "", "3", "3", "3", "eq"],
        svec!["6", "", "", "", "-1", "", "ne"],
        svec!["4", "4", "2", "4", "2", "4", "ne"],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_div_sign() {
    let wrk = Workdir::new("sqlp_div_sign");
    wrk.create(
        "test.csv",
        vec![
            svec!["a", "b"],
            svec!["10.0", "-100.5"],
            svec!["20.0", "7.0"],
            svec!["30.0", "2.5"],
            svec!["40.0", ""],
            svec!["50.0", "-3.14"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv").arg(
        r#"
        SELECT
            a / b AS a_div_b,
            a // b AS a_floordiv_b,
            SIGN(b) AS b_sign,
        FROM test
"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["a_div_b", "a_floordiv_b", "b_sign"],
        svec!["-0.09950248756218906", "-1", "-1.0"],
        svec!["2.857142857142857", "2", "1.0"],
        svec!["12.0", "12", "1.0"],
        svec!["", "", ""],
        svec!["-15.92356687898089", "-16", "-1.0"],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_string_replace() {
    let wrk = Workdir::new("sqlp_string_replace");
    wrk.create(
        "test.csv",
        vec![
            svec!["words"],
            svec!["Yemeni coffee is the best coffee"],
            svec![""],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv")
        .arg(
            r#"
        SELECT
        REPLACE(
          REPLACE(words, 'coffee', 'tea'),
          'Yemeni',
          'English breakfast'
        )
        FROM test
"#,
        )
        .args(["--rnull-values", "NULL"]);

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["words"],
        svec!["English breakfast tea is the best tea"],
        svec![""],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_compound_join_basic() {
    let wrk = Workdir::new("sqlp_compound_join_basic");
    wrk.create(
        "test1.csv",
        vec![
            svec!["a", "b"],
            svec!["1", "1"],
            svec!["2", "3"],
            svec!["3", "4"],
            svec!["4", "4"],
            svec!["5", "5"],
        ],
    );

    wrk.create(
        "test2.csv",
        vec![
            svec!["a", "b"],
            svec!["1", "0"],
            svec!["2", "3"],
            svec!["3", "4"],
            svec!["4", "5"],
            svec!["5", "6"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.args(["test1.csv", "test2.csv"])
        .arg(
            r#"
        SELECT * FROM test1
         INNER JOIN test2 ON test1.a = test2.a AND test1.b = test2.b
"#,
        )
        .arg("--truncate-ragged-lines");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["a", "b", "a:test2", "b:test2"],
        svec!["2", "3", "2", "3"],
        svec!["3", "4", "3", "4"],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_compound_join_diff_colnames() {
    let wrk = Workdir::new("sqlp_compound_join_diff_colnames");
    wrk.create(
        "test1.csv",
        vec![
            svec!["a", "b"],
            svec!["1", "1"],
            svec!["2", "2"],
            svec!["3", "3"],
            svec!["4", "4"],
            svec!["5", "5"],
        ],
    );

    wrk.create(
        "test2.csv",
        vec![
            svec!["a", "b", "c"],
            svec!["0", "1", "7"],
            svec!["2", "2", "8"],
            svec!["3", "3", "9"],
            svec!["4", "5", "10"],
            svec!["5", "6", "11"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.args(["test1.csv", "test2.csv"])
        .arg(
            r#"
        SELECT * FROM test1
         INNER JOIN test2 ON test1.a = test2.b AND test1.b = test2.a
"#,
        )
        .arg("--truncate-ragged-lines");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["a", "b", "a:test2", "b:test2", "c"],
        svec!["2", "2", "2", "2", "8"],
        svec!["3", "3", "3", "3", "9"],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_compound_join_three_tables() {
    let wrk = Workdir::new("sqlp_compound_join_three_tables");
    wrk.create(
        "test1.csv",
        vec![
            svec!["a", "b"],
            svec!["1", "1"],
            svec!["2", "3"],
            svec!["3", "4"],
            svec!["4", "4"],
            svec!["5", "5"],
        ],
    );

    wrk.create(
        "test2.csv",
        vec![
            svec!["a", "b"],
            svec!["1", "0"],
            svec!["2", "3"],
            svec!["3", "4"],
            svec!["4", "5"],
            svec!["5", "6"],
        ],
    );

    wrk.create(
        "test3.csv",
        vec![
            svec!["a", "b", "c"],
            svec!["1", "0", "0"],
            svec!["2", "3", "3"],
            svec!["3", "4", "4"],
            svec!["4", "5", "5"],
            svec!["5", "6", "6"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.args(["test1.csv", "test2.csv", "test3.csv"])
        .arg(
            r#"
            SELECT * FROM test1
            INNER JOIN test2
                ON test1.a = test2.a AND test1.b = test2.b
            INNER JOIN test3
                ON test1.a = test3.a AND test1.b = test3.b
"#,
        )
        .arg("--truncate-ragged-lines");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["a", "b", "a:test2", "b:test2", "a:test3", "b:test3", "c"],
        svec!["2", "3", "2", "3", "2", "3", "3"],
        svec!["3", "4", "3", "4", "3", "4", "4"],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_string_concat() {
    let wrk = Workdir::new("sqlp_string_concat");
    wrk.create(
        "test.csv",
        vec![
            svec!["x", "y", "z"],
            svec!["a", "d", "1"],
            svec!["", "e", "2"],
            svec!["c", "f", "3"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv").arg(
        r#"
        SELECT
           ("x" || "x" || "y")           AS c0,
           ("x" || "y" || "z")           AS c1,
           CONCAT(("x" + "x"), "y")      AS c2,
           CONCAT("x", "x", "y")         AS c3,
           CONCAT("x", "y", ("z" * 2))   AS c4,
           CONCAT_WS(':', "x", "y", "z") AS c5,
           CONCAT_WS('!', "x", "y", "z") AS c6,
        FROM test
"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["c0", "c1", "c2", "c3", "c4", "c5", "c6"],
        svec!["aad", "ad1", "aad", "aad", "ad2", "a:d:1", "a!d!1"],
        svec!["", "", "e", "e", "e4", "e:2", "e!2"],
        svec!["ccf", "cf3", "ccf", "ccf", "cf6", "c:f:3", "c!f!3"],
    ];

    assert_eq!(got, expected);
}

// #[test]
// fn sqlp_select_1() {
//     let wrk = Workdir::new("sqlp_select_1");
//     wrk.create(
//         "test.csv",
//         vec![
//             svec!["x", "y", "z"],
//             svec!["a", "d", "1"],
//             svec!["", "e", "2"],
//             svec!["c", "f", "3"],
//         ],
//     );

//     let mut cmd = wrk.command("sqlp");
//     cmd.arg("test.csv").arg("SELECT 1 from _t_1");

//     wrk.assert_success(&mut cmd);

//     let got = wrk.output_stderr(&mut cmd);
//     let expected = "(3, 1)";

//     assert!(got.starts_with(expected));
// }

#[test]
fn sqlp_string_right_reverse() {
    let wrk = Workdir::new("sqlp_string_right_reverse");
    wrk.create(
        "test.csv",
        vec![
            svec!["txt"],
            svec!["abcde"],
            svec!["abc"],
            svec!["a"],
            svec![""],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv").arg(
        r#"
        SELECT
           LEFT(txt,2) AS "l",
           RIGHT(txt,2) AS "r",
           REVERSE(txt) AS "rev"
        FROM test
"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["l", "r", "rev"],
        svec!["ab", "de", "edcba"],
        svec!["ab", "bc", "cba"],
        svec!["a", "a", "a"],
        svec!["", "", ""],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_modulo() {
    let wrk = Workdir::new("sqlp_modulo");
    wrk.create(
        "test.csv",
        vec![
            svec!["a", "b", "c", "d"],
            svec!["1.5", "6", "11", "16.5"],
            svec!["", "7", "12", "17.0"],
            svec!["3.0", "8", "13", "18.5"],
            svec!["4333333333", "9", "14", ""],
            svec!["5.0", "10", "15", "20.0"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv").arg(
        r#"
        SELECT
            cast(a as float) % 2 AS a2,
            cast(b as float)  % 3 AS b3,
            MOD(cast(c as float), 4) AS c4,
            MOD(cast(d as float), 5.5) AS d55
        FROM test
"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["a2", "b3", "c4", "d55"],
        svec!["1.5", "0.0", "3.0", "0.0"],
        svec!["", "1.0", "0.0", "0.5"],
        svec!["1.0", "2.0", "1.0", "2.0"],
        svec!["1.0", "0.0", "2.0", ""],
        svec!["1.0", "1.0", "3.0", "3.5"],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_try_cast() {
    let wrk = Workdir::new("sqlp_try_cast");
    wrk.create(
        "test.csv",
        vec![
            svec!["foo", "bar"],
            svec!["65432", "1999-12-31"],
            svec!["101010", "N/A"],
            svec!["-3333", "2024-01-01"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv").arg(
        r#"
        SELECT
            try_cast(foo as uint2),
            try_cast(bar as DATE)
        FROM test
"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["foo", "bar"],
        svec!["65432", "1999-12-31"],
        svec!["", ""],
        svec!["", "2024-01-01"],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_stddev_variance() {
    let wrk = Workdir::new("sqlp_stddev_variance");
    wrk.create(
        "test.csv",
        vec![
            svec!["v1", "v2", "v3", "v4"],
            svec!["-1.0", "5.5", "-10", "-100"],
            svec!["0.0", "0.0", "", "0.0"],
            svec!["1.0", "3.0", "10", "-50.0"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("test.csv").arg(
        r#"
        SELECT
            STDEV(v1) AS "v1_std",
            STDDEV(v2) AS "v2_std",
            STDEV_SAMP(v3) AS "v3_std",
            STDDEV_SAMP(v4) AS "v4_std",
            VAR(v1) AS "v1_var",
            VARIANCE(v2) AS "v2_var",
            VARIANCE(v3) AS "v3_var",
            VAR_SAMP(v4) AS "v4_var"
        FROM test
"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec![
            "v1_std", "v2_std", "v3_std", "v4_std", "v1_var", "v2_var", "v3_var", "v4_var"
        ],
        svec![
            "1.0",
            "2.753785273643051",
            "14.142135623730951",
            "50.0",
            "1.0",
            "7.583333333333334",
            "200.0",
            "2500.0"
        ],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_string_position() {
    let wrk = Workdir::new("sqlp_string_position");
    wrk.create(
        "cities.csv",
        vec![
            svec!["city"],
            svec!["Dubai"],
            svec!["Abu Dhabi"],
            svec!["Sharjah"],
            svec!["Al Ain"],
            svec!["Ajman"],
            svec!["Ras Al Khaimah"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("cities.csv").arg(
        r#"
        SELECT
        POSITION('a' IN city) AS a_lc1,
        POSITION('A' IN city) AS a_uc1,
        STRPOS(city,'a') AS a_lc2,
        STRPOS(city,'A') AS a_uc2,
      FROM cities
"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["a_lc1", "a_uc1", "a_lc2", "a_uc2"],
        svec!["4", "0", "4", "0"],
        svec!["7", "1", "7", "1"],
        svec!["3", "0", "3", "0"],
        svec!["0", "1", "0", "1"],
        svec!["4", "1", "4", "1"],
        svec!["2", "5", "2", "5"],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_date_part() {
    let wrk = Workdir::new("sqlp_date_part");
    wrk.create(
        "datestbl.csv",
        vec![svec!["datecol"], svec!["20-01-2012 10:30:20"]],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("datestbl.csv")
        .arg(
            r#"
        SELECT 
            DATE_PART('isoyear', datecol) as c1,
            DATE_PART('month', datecol) as c2,
            DATE_PART('day', datecol) as c3,
            DATE_PART('hour', datecol) as c4,
            DATE_PART('minute', datecol) as c5,
            DATE_PART('second', datecol) as c6,
            DATE_PART('millisecond', datecol) as c61,
            DATE_PART('microsecond', datecol) as c62,
            DATE_PART('nanosecond', datecol) as c63,
            DATE_PART('isoweek', datecol) as c7,
            DATE_PART('dayofyear', datecol) as c8,
            DATE_PART('dayofweek', datecol) as c9,
            DATE_PART('time', datecol) as c10,
            DATE_PART('decade', datecol) as c11,
            DATE_PART('century', datecol) as c12,
            DATE_PART('millennium', datecol) as c13,
            DATE_PART('quarter', datecol) as c14,
      FROM datestbl
"#,
        )
        .arg("--try-parsedates");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec![
            "c1", "c2", "c3", "c4", "c5", "c6", "c61", "c62", "c63", "c7", "c8", "c9", "c10",
            "c11", "c12", "c13", "c14"
        ],
        svec![
            "2012",
            "1",
            "20",
            "10",
            "30",
            "20",
            "20000.0",
            "20000000.0",
            "20000000000.0",
            "3",
            "20",
            "5",
            "10:30:20.000000000",
            "201",
            "21",
            "3",
            "1"
        ],
    ];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_date_part_tz() {
    let wrk = Workdir::new("sqlp_date_part_tz");
    wrk.create(
        "datestbl.csv",
        vec![
            svec!["datecol"],
            svec!["2012-01-20T10:30:20-05:00"],
            svec!["2012-11-05T10:30:20.1234Z"],
            svec!["1999-03-15T15:30:20.1234+01:00"],
            svec!["1978-08-05T23:30:20.1234-01:00"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("datestbl.csv")
        .arg(
            r#"
        SELECT
            DATE_PART('isoyear', datecol) as c1,
            DATE_PART('month', datecol) as c2,
            DATE_PART('day', datecol) as c3,
            DATE_PART('hour', datecol) as c4,
            DATE_PART('minute', datecol) as c5,
            DATE_PART('second', datecol) as c6,
            DATE_PART('millisecond', datecol) as c61,
            DATE_PART('microsecond', datecol) as c62,
            DATE_PART('nanosecond', datecol) as c63,
            DATE_PART('isoweek', datecol) as c7,
            DATE_PART('dayofyear', datecol) as c8,
            DATE_PART('dayofweek', datecol) as c9,
            DATE_PART('time', datecol) as c10,
            DATE_PART('decade', datecol) as c11,
            DATE_PART('century', datecol) as c12,
            DATE_PART('millennium', datecol) as c13,
            DATE_PART('quarter', datecol) as c14,
            DATE_PART('timezone', datecol) as c15,
            DATE_PART('epoch', datecol) as c16,
      FROM datestbl
"#,
        )
        .arg("--try-parsedates");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec![
            "c1", "c2", "c3", "c4", "c5", "c6", "c61", "c62", "c63", "c7", "c8", "c9", "c10",
            "c11", "c12", "c13", "c14", "c15", "c16"
        ],
        svec![
            "2012",
            "1",
            "20",
            "15",
            "30",
            "20",
            "20000.0",
            "20000000.0",
            "20000000000.0",
            "3",
            "20",
            "5",
            "15:30:20.000000000",
            "201",
            "21",
            "3",
            "1",
            "0",
            "1327073420.0"
        ],
        svec![
            "2012",
            "11",
            "5",
            "10",
            "30",
            "20",
            "20123.4",
            "20123400.0",
            "20123400000.0",
            "45",
            "310",
            "1",
            "10:30:20.123400000",
            "201",
            "21",
            "3",
            "4",
            "0",
            "1352111420.1234"
        ],
        svec![
            "1999",
            "3",
            "15",
            "14",
            "30",
            "20",
            "20123.4",
            "20123400.0",
            "20123400000.0",
            "11",
            "74",
            "1",
            "14:30:20.123400000",
            "199",
            "20",
            "2",
            "1",
            "0",
            "921508220.1234"
        ],
        svec![
            "1978",
            "8",
            "6",
            "0",
            "30",
            "20",
            "20123.4",
            "20123400.0",
            "20123400000.0",
            "31",
            "218",
            "0",
            "00:30:20.123400000",
            "197",
            "20",
            "2",
            "3",
            "0",
            "271211420.1234"
        ],
    ];

    assert_eq!(got, expected);
}

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg("SKIP_INPUT").arg(
        r#"
        SELECT 
            DATE('2021-03-15') as c1,
            DATE('2021-03-15 10:30:20', '%Y-%m-%d %H:%M:%S') as c2,
            DATE('03-15-2021 10:30:20 AM EST', '%m-%d-%Y %I:%M:%S %p %Z') as c3"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    // this is the documented behavior of the date function
    // use STRFTIME and STRPTIME for more control
    let expected = vec![
        svec!["c1", "c2", "c3"],
        svec!["2021-03-15", "2021-03-15", "2021-03-15"],
    ];

    assert_eq!(got, expected);
}

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

    wrk.create(
        "data.csv",
        vec![
            svec!["dtm", "dt", "tm"],
            svec!["1972-03-06 23:50:03", "1978-07-05", "10:10:10"],
            svec!["1980-09-30 01:25:50", "1969-12-31", "22:33:55"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg(
            r#"
        SELECT
      STRFTIME(dtm,'%m.%d.%Y/%T') AS s_dtm,
      STRFTIME(dt,'%B %d, %Y') AS s_dt,
      STRFTIME(tm,'%S.%M.%H') AS s_tm,
    FROM data"#,
        )
        .arg("--try-parsedates");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["s_dtm", "s_dt", "s_tm"],
        svec!["03.06.1972/23:50:03", "July 05, 1978", "10.10.10"],
        svec!["09.30.1980/01:25:50", "December 31, 1969", "55.33.22"],
    ];

    assert_eq!(got, expected);
}

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

    let test_jsonl = wrk.load_test_file("boston311-10.jsonl");

    let mut cmd = wrk.command("sqlp");
    cmd.arg("SKIP_INPUT").arg(
        format!(
            "SELECT * FROM read_json('{}') WHERE source = 'City Worker App'",
            test_jsonl
        )
        .as_str(),
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec![
            "case_enquiry_id",
            "open_dt",
            "target_dt",
            "closed_dt",
            "ontime",
            "case_status",
            "closure_reason",
            "case_title",
            "subject",
            "reason",
            "type",
            "queue",
            "department",
            "submittedphoto",
            "closedphoto",
            "location",
            "fire_district",
            "pwd_district",
            "city_council_district",
            "police_district",
            "neighborhood",
            "neighborhood_services_district",
            "ward",
            "precinct",
            "location_street_name",
            "location_zipcode",
            "latitude",
            "longitude",
            "source"
        ],
        svec![
            "101004120108",
            "2022-01-08 12:54:49",
            "2022-01-11 08:30:00",
            "2022-01-09 06:43:06",
            "ONTIME",
            "Closed",
            "Case Closed. Closed date : Sun Jan 09 06:43:06 EST 2022 Noted ",
            "CE Collection",
            "Public Works Department",
            "Street Cleaning",
            "CE Collection",
            "PWDx_District 1C: Downtown",
            "PWDx",
            "",
            "",
            "198 W Springfield St  Roxbury  MA  02118",
            "4",
            "1C",
            "7",
            "D4",
            "South End",
            "6",
            "Ward 9",
            "0902",
            "198 W Springfield St",
            "2118",
            "42.3401",
            "-71.0803",
            "City Worker App"
        ],
        svec![
            "101004141354",
            "2022-01-20 08:07:49",
            "2022-01-21 08:30:00",
            "2022-01-20 08:45:03",
            "ONTIME",
            "Closed",
            "Case Closed. Closed date : Thu Jan 20 08:45:03 EST 2022 Noted ",
            "CE Collection",
            "Public Works Department",
            "Street Cleaning",
            "CE Collection",
            "PWDx_District 1B: North End",
            "PWDx",
            "",
            "",
            "21-23 Temple St  Boston  MA  02114",
            "3",
            "1B",
            "1",
            "A1",
            "Beacon Hill",
            "3",
            "Ward 3",
            "0306",
            "21-23 Temple St",
            "2114",
            "42.3606",
            "-71.0638",
            "City Worker App"
        ],
        svec![
            "101004141367",
            "2022-01-20 08:15:45",
            "2022-01-21 08:30:00",
            "2022-01-20 08:45:12",
            "ONTIME",
            "Closed",
            "Case Closed. Closed date : Thu Jan 20 08:45:12 EST 2022 Noted ",
            "CE Collection",
            "Public Works Department",
            "Street Cleaning",
            "CE Collection",
            "PWDx_District 1B: North End",
            "PWDx",
            "",
            "",
            "12 Derne St  Boston  MA  02114",
            "3",
            "1B",
            "1",
            "A1",
            "Beacon Hill",
            "3",
            "Ward 3",
            "0306",
            "12 Derne St",
            "2114",
            "42.3596",
            "-71.0634",
            "City Worker App"
        ],
    ];

    assert_eq!(got, expected);
}

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

    wrk.create(
        "likedata.csv",
        vec![
            svec!["x", "y"],
            svec!["aaa", "abc"],
            svec!["bbb", "b"],
            svec!["a", "aa"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("likedata.csv").arg(
        r#"
        SELECT
            x,
            x ^@ 'a' AS x_starts_with_a,
            x ~~* '%B' AS x_ends_with_b_case_insensitive,
            x ^@ y AS x_starts_with_y,
            x ~~ '%a' AS x_ends_with_a
        FROM likedata"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec![
            "x",
            "x_starts_with_a",
            "x_ends_with_b_case_insensitive",
            "x_starts_with_y",
            "x_ends_with_a"
        ],
        svec!["aaa", "true", "false", "false", "true"],
        svec!["bbb", "false", "true", "true", "false"],
        svec!["a", "true", "false", "false", "true"],
    ];

    assert_eq!(got, expected);
}

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

    wrk.create(
        "data1.csv",
        vec![
            svec!["CharacterID", "FirstName", "LastName"],
            svec!["1", "Jerna Morat", "Gurgeh"],
            svec!["2", "Cheradenine", "Zakalwe"],
            svec!["3", "Byr", "Genar-Hofoen"],
        ],
    );

    wrk.create(
        "data2.csv",
        vec![
            svec!["CharacterID", "Book"],
            svec!["1", "Player of Games"],
            svec!["2", "Use of Weapons"],
            svec!["3", "Excession"],
        ],
    );

    wrk.create(
        "data3.csv",
        vec![
            svec!["CharacterID", "Ship"],
            svec!["1", "Limiting Factor"],
            svec!["2", "Xenophobe"],
            svec!["3", "Grey Area"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.args(["data1.csv", "data2.csv", "data3.csv"]).arg(
        r#"SELECT COLUMNS('^[^:]+$')
  FROM data1
    NATURAL JOIN data2
    NATURAL JOIN data3
  ORDER BY CharacterID"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["CharacterID", "FirstName", "LastName", "Book", "Ship"],
        svec![
            "1",
            "Jerna Morat",
            "Gurgeh",
            "Player of Games",
            "Limiting Factor"
        ],
        svec!["2", "Cheradenine", "Zakalwe", "Use of Weapons", "Xenophobe"],
        svec!["3", "Byr", "Genar-Hofoen", "Excession", "Grey Area"],
    ];

    assert_eq!(got, expected);
}

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

    wrk.create(
        "starlikedata.csv",
        vec![
            svec!["ID", "FirstName", "LastName", "Address", "City"],
            svec!["333", "Bruce", "Wayne", "The Batcave", "Gotham"],
            svec!["666", "Diana", "Prince", "Paradise Island", "Themyscira"],
            svec!["999", "Clark", "Kent", "Fortress of Solitude", "Metropolis"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("starlikedata.csv").arg(
        r#"
        SELECT * ILIKE '%a%e%'
  FROM starlikedata
  ORDER BY FirstName"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["FirstName", "LastName", "Address"],
        svec!["Bruce", "Wayne", "The Batcave"],
        svec!["Clark", "Kent", "Fortress of Solitude"],
        svec!["Diana", "Prince", "Paradise Island"],
    ];

    assert_eq!(got, expected);

    let mut cmd = wrk.command("sqlp");
    cmd.arg("starlikedata.csv").arg(
        r#"
        SELECT * ILIKE '%I%' RENAME (FirstName AS Name) 
  FROM starlikedata
  ORDER BY 3 DESC"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["ID", "Name", "City"],
        svec!["666", "Diana", "Themyscira"],
        svec!["999", "Clark", "Metropolis"],
        svec!["333", "Bruce", "Gotham"],
    ];

    assert_eq!(got, expected);

    let mut cmd = wrk.command("sqlp");
    cmd.arg("starlikedata.csv").arg(
        r#"
        SELECT * EXCLUDE (ID, City, LastName) RENAME FirstName AS Name
  FROM starlikedata
  ORDER BY Name"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["Name", "Address"],
        svec!["Bruce", "The Batcave"],
        svec!["Clark", "Fortress of Solitude"],
        svec!["Diana", "Paradise Island"],
    ];

    assert_eq!(got, expected);
}

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg("SKIP_INPUT")
        .arg("SELECT 1 AS one, '2' AS two, 3.0 AS three");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![svec!["one", "two", "three"], svec!["1", "2", "3.0"]];

    assert_eq!(got, expected);
}

#[test]
fn sqlp_autotab_delim() {
    let wrk = Workdir::new("sqlp_autotab_delim");
    wrk.create_with_delim(
        "cities_array.tsv",
        vec![
            svec!["countryid", "cities"],
            svec!["DB", "Dubai,Abu Dhabi,Sharjah"],
            svec!["IN", "Mumbai,Delhi,Bangalore"],
            svec!["US", "New York,Los Angeles,Chicago"],
            svec!["UK", "London,Birmingham,Manchester"],
            svec!["CN", "Beijing"],
            svec!["RU", "Moscow"],
            svec!["NL", "Amsterdam"],
            svec!["IT", "Rome,Milan,Turin,Naples,Venice"],
        ],
        b'\t',
    );

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg("cities_array.tsv")
        .arg(
            r#"
            SELECT 
                countryid, cities
            FROM cities_array
    "#,
        )
        .args(["--output", &output_file]);

    wrk.assert_success(&mut cmd);

    let got = wrk.read_to_string(&output_file).unwrap();
    let expected = r#"countryid	cities
DB	Dubai,Abu Dhabi,Sharjah
IN	Mumbai,Delhi,Bangalore
US	New York,Los Angeles,Chicago
UK	London,Birmingham,Manchester
CN	Beijing
RU	Moscow
NL	Amsterdam
IT	Rome,Milan,Turin,Naples,Venice
"#;

    assert_eq!(got, expected);
}

#[test]
fn sqlp_dollar_quoting_string_literals() {
    let wrk = Workdir::new("sqlp_dollar_quoting_string_literals");
    wrk.create(
        "cities_array.csv",
        vec![
            svec!["country_id", "description"],
            svec!["DB", "Dubai"],
            svec!["IN", "India"],
            svec!["US", "United States"],
            svec!["UK", "United Kingdom"],
            svec!["CN", "China"],
            svec!["RU", "Russia"],
            svec!["NL", "Netherlands"],
            svec!["IT", "Italy"],
        ],
    );

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

    let mut cmd = wrk.command("sqlp");
    cmd.arg("cities_array.csv")
        .arg(
            r#"SELECT country_id, $$This is a literal $,%,',"$$ AS complex_literal FROM cities_array"#,
        )
        .args(["--output", &output_file]);

    wrk.assert_success(&mut cmd);

    let got = wrk.read_to_string(&output_file).unwrap();
    let expected = r#"country_id	complex_literal
DB	"This is a literal $,%,',"""
IN	"This is a literal $,%,',"""
US	"This is a literal $,%,',"""
UK	"This is a literal $,%,',"""
CN	"This is a literal $,%,',"""
RU	"This is a literal $,%,',"""
NL	"This is a literal $,%,',"""
IT	"This is a literal $,%,',"""
"#;

    assert_eq!(got, expected);
}

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

    // Create left table with partial strings
    wrk.create(
        "left.csv",
        vec![
            svec!["id", "val"],
            svec!["1", "cat"],
            svec!["2", "dog"],
            svec!["3", "bird"],
            svec!["4", "fish"],
        ],
    );

    // Create right table with full strings
    wrk.create(
        "right.csv",
        vec![
            svec!["id", "val"],
            svec!["1", "category"],
            svec!["2", "doghouse"],
            svec!["3", "jailbird"],
            svec!["4", "starfish"],
            svec!["5", "catalog"],
            svec!["6", "hotdog"],
        ],
    );

    // Test STRPOS
    let mut cmd = wrk.command("sqlp");
    cmd.args(["left.csv", "right.csv"]).arg(
        "SELECT l.val as left_val, r.val as right_val FROM left l CROSS JOIN right r WHERE \
         STRPOS(r.val, l.val) > 0 ORDER BY l.val, r.val",
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["left_val", "right_val"],
        svec!["bird", "jailbird"],
        svec!["cat", "catalog"],
        svec!["cat", "category"],
        svec!["dog", "doghouse"],
        svec!["dog", "hotdog"],
        svec!["fish", "starfish"],
    ];
    assert_eq!(got, expected);

    // Test STARTS_WITH
    let mut cmd = wrk.command("sqlp");
    cmd.args(["left.csv", "right.csv"]).arg(
        "SELECT l.val as left_val, r.val as right_val FROM left l CROSS JOIN right r WHERE \
         STARTS_WITH(r.val, l.val) ORDER BY l.val, r.val",
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["left_val", "right_val"],
        svec!["cat", "catalog"],
        svec!["cat", "category"],
        svec!["dog", "doghouse"],
    ];
    assert_eq!(got, expected);

    // Test ENDS_WITH
    let mut cmd = wrk.command("sqlp");
    cmd.args(["left.csv", "right.csv"]).arg(
        "SELECT l.val as left_val, r.val as right_val FROM left l CROSS JOIN right r WHERE \
         ENDS_WITH(r.val, l.val) ORDER BY l.val, r.val",
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["left_val", "right_val"],
        svec!["bird", "jailbird"],
        svec!["dog", "hotdog"],
        svec!["fish", "starfish"],
    ];
    assert_eq!(got, expected);

    // Test reverse STRPOS (left contains right)
    let mut cmd = wrk.command("sqlp");
    cmd.args(["left.csv", "right.csv"]).arg(
        "SELECT l.val as left_val, r.val as right_val FROM left l CROSS JOIN right r WHERE \
         STRPOS(l.val, r.val) > 0 ORDER BY l.val, r.val",
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![svec!["left_val", "right_val"]]; // Empty because no left values contain right values in our test data
    assert_eq!(got, expected);
}

#[test]
fn sqlp_string_to_array() {
    let wrk = Workdir::new("sqlp_string_to_array");
    wrk.create(
        "data.csv",
        vec![
            svec!["id", "cities"],
            svec!["1", "Dubai,Abu Dhabi,Sharjah"],
            svec!["2", "Mumbai,Delhi,Bangalore"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg(
            r#"
        SELECT id, STRING_TO_ARRAY(cities, ',') AS cities FROM data
    "#,
        )
        .args(["--format", "json"]);

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);
    let expected = r#"[{"id":1,"cities":["Dubai","Abu Dhabi","Sharjah"]},{"id":2,"cities":["Mumbai","Delhi","Bangalore"]}]"#;
    assert_eq!(got, expected);
}

#[test]
fn sqlp_split_part() {
    let wrk = Workdir::new("sqlp_split_part");
    wrk.create(
        "data.csv",
        vec![
            svec!["s"],
            svec!["xx,yy,zz"],
            svec!["abc,,xyz,???,hmmm"],
            svec![""],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv").arg(
        r#"
        SELECT
          SPLIT_PART(s,',',1) AS "s+1",
          SPLIT_PART(s,',',3) AS "s+3",
          SPLIT_PART(s,',',-2) AS "s-2",
        FROM data
    "#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["s+1", "s+3", "s-2"],
        svec!["xx", "zz", "yy"],
        svec!["abc", "xyz", "???"],
        svec!["", "", ""],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_corr() {
    let wrk = Workdir::new("sqlp_corr");
    wrk.create(
        "data.csv",
        vec![
            svec!["foo", "bar"],
            svec!["1", "2"],
            svec!["2", "4"],
            svec!["3", "7"],
            svec!["4", "5"],
            svec!["5", "9"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg("SELECT CORR(foo, bar) AS corr FROM data")
        .args(["--float-precision", "6"]);

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![svec!["corr"], svec!["0.877809"]];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_covar_pop() {
    let wrk = Workdir::new("sqlp_covar_pop");
    wrk.create(
        "data.csv",
        vec![
            svec!["foo", "bar"],
            svec!["1", "2"],
            svec!["2", "4"],
            svec!["3", "7"],
            svec!["4", "5"],
            svec!["5", "9"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg("SELECT COVAR(foo, bar) AS covar FROM data")
        .args(["--float-precision", "2"]);

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![svec!["covar"], svec!["3.75"]];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_corr_covar_covar_pop() {
    let wrk = Workdir::new("sqlp_corr_covar_covar_pop");
    wrk.create(
        "data.csv",
        vec![
            svec!["foo", "bar"],
            svec!["1", "2"],
            svec!["2", "4"],
            svec!["3", "7"],
            svec!["4", "5"],
            svec!["5", "9"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg(
            "SELECT CORR(foo, bar) AS corr, COVAR(foo, bar) AS covar, COVAR_POP(foo, bar) AS \
             covar_pop FROM data",
        )
        .args(["--float-precision", "2"]);

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["corr", "covar", "covar_pop"],
        svec!["0.88", "3.75", "3.00"],
    ];
    assert_eq!(got, expected);
}

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

//     wrk.create(
//         "data.csv",
//         vec![
//             svec!["a", "b", "c"],
//             svec!["1", "2", "3"],
//             svec!["4", "5", "6"],
//             svec!["7", "8", "9"],
//         ],
//     );

//     let output_dotfile = wrk.path("output.dot").to_string_lossy().to_string();

//     let mut cmd = wrk.command("sqlp");
//     cmd.env("POLARS_VISUALIZE_PHYSICAL_PLAN", output_dotfile.as_str())
//         .arg("data.csv")
//         .arg(
//             r#"
//         SELECT a, b, c
//         FROM data
//         WHERE a > 2
//         ORDER BY a DESC
//     "#,
//         )
//         .arg("--streaming");

//     wrk.assert_success(&mut cmd);

//     assert!(std::path::Path::new(&output_dotfile).exists());

//     let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
//     let expected = vec![
//         svec!["a", "b", "c"],
//         svec!["7", "8", "9"],
//         svec!["4", "5", "6"],
//     ];

//     assert_eq!(got, expected);

//     let got_dot = wrk.read_to_string(&output_dotfile).unwrap();
//     let expected_dot = r#"digraph {
//     "Projection" -> "Filter";
//     "Filter" -> "CsvScan";
// }"#;

//     assert_eq!(got_dot, expected_dot);
// }

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

    // Create test data with decimal commas
    let test_data = vec![
        svec!["id", "value"],
        svec!["1", "100,50"],
        svec!["2", "200,75"],
    ];
    wrk.create("data.csv", test_data);

    // Test 1: --decimal-comma with comma delimiter should fail
    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg("--decimal-comma")
        .arg("select * from _t_1");

    wrk.assert_err(&mut cmd);
    let got = wrk.output_stderr(&mut cmd);
    assert!(got.contains("Using --decimal-comma with a comma separator is invalid"));

    // Test 2: --decimal-comma with semicolon delimiter should succeed
    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg("--decimal-comma")
        .args(["--delimiter", ";"])
        .arg("select * from _t_1");

    wrk.assert_success(&mut cmd);
    let got: String = wrk.stdout(&mut cmd);
    let expected = "id,value\n\"1,\"\"100,50\"\"\"\n\"2,\"\"200,75\"\"\"";
    assert_eq!(got, expected);

    // Test 3: --decimal-comma with tab delimiter should succeed
    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg("--decimal-comma")
        .args(["--delimiter", "\t"])
        .arg("select * from _t_1");

    wrk.assert_success(&mut cmd);
    let got: String = wrk.stdout(&mut cmd);
    let expected = "id,value\n\"1,\"\"100,50\"\"\"\n\"2,\"\"200,75\"\"\"";
    assert_eq!(got, expected);

    // Test 4: --decimal-comma with pipe delimiter should succeed
    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg("--decimal-comma")
        .args(["--delimiter", "|"])
        .arg("select * from _t_1");

    wrk.assert_success(&mut cmd);
    let got: String = wrk.stdout(&mut cmd);
    let expected = "id,value\n\"1,\"\"100,50\"\"\"\n\"2,\"\"200,75\"\"\"";
    assert_eq!(got, expected);
}

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

    // Create test data with decimal commas in TSV format
    let test_data = vec![
        svec!["id", "value"],
        svec!["1", "100,50"],
        svec!["2", "200,75"],
    ];
    wrk.create_with_delim("data.tsv", test_data, b'\t');

    // Test: --decimal-comma with TSV file (tab delimiter) should succeed
    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.tsv")
        .arg("--decimal-comma")
        .arg("select * from _t_1");

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["id", "value"],
        svec!["1", "100,5"],
        svec!["2", "200,75"],
    ];
    assert_eq!(got, expected);
}

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

    // Create test data with decimal commas in SSV format
    let test_data = vec![
        svec!["id", "value"],
        svec!["1", "100,50"],
        svec!["2", "200,75"],
    ];
    wrk.create_with_delim("data.ssv", test_data, b';');

    // Test: --decimal-comma with SSV file (semicolon delimiter) should succeed
    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.ssv")
        .arg("--decimal-comma")
        .arg("select * from _t_1");

    wrk.assert_success(&mut cmd);
    let got: String = wrk.stdout(&mut cmd);
    let expected = "id,value\n1,\"100,5\"\n2,\"200,75\"";
    assert_eq!(got, expected);
}

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

    // Create test data with decimal commas
    let test_data = vec![
        svec!["id", "value"],
        svec!["1", "100,50"],
        svec!["2", "200,75"],
    ];
    wrk.create_with_delim("data.csv", test_data, b';');

    // Test: --decimal-comma with output file should validate the output delimiter
    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg("--decimal-comma")
        .args(["--delimiter", ";"])
        .args(["--output", "output.csv"])
        .arg("select * from _t_1");

    wrk.assert_success(&mut cmd);
    let got: String = wrk.read_to_string("output.csv").unwrap();
    let expected = "id;value\n1;100,5\n2;200,75\n";
    assert_eq!(got, expected);

    // Test: --decimal-comma with output file that would have comma delimiter should fail
    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg("--decimal-comma")
        .args(["--output", "output.csv"])
        .arg("select * from _t_1");

    wrk.assert_err(&mut cmd);
    let got = wrk.output_stderr(&mut cmd);
    assert!(got.contains("Using --decimal-comma with a comma separator is invalid"));
}

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

    // Create test data with decimal commas
    let test_data = vec![
        svec!["id", "value"],
        svec!["1", "100,50"],
        svec!["2", "200,75"],
    ];
    wrk.create_with_delim("data.csv", test_data, b'\t');

    // Test: --decimal-comma with TSV output file should succeed
    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg("--decimal-comma")
        .args(["--delimiter", ";"])
        .args(["--output", "output.tsv"])
        .arg("select * from _t_1");

    wrk.assert_success(&mut cmd);
    let got: String = wrk.read_to_string("output.tsv").unwrap();
    let expected = "\"id\tvalue\"\n\"1\t100,50\"\n\"2\t200,75\"\n";
    assert_eq!(got, expected);
}

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

    // Create test data with decimal commas
    let test_data = vec![
        svec!["id", "value"],
        svec!["1", "100,50"],
        svec!["2", "200,75"],
    ];
    wrk.create_with_delim("data.csv", test_data, b';');

    // Test: --decimal-comma with SSV output file should succeed
    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg("--decimal-comma")
        .args(["--delimiter", ";"])
        .args(["--output", "output.ssv"])
        .arg("select * from _t_1");

    wrk.assert_success(&mut cmd);
    let got: String = wrk.read_to_string("output.ssv").unwrap();
    let expected = "id;value\n1;100,5\n2;200,75\n";
    assert_eq!(got, expected);
}

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

    // Create test data with decimal commas
    let test_data = vec![
        svec!["id", "value"],
        svec!["1", "100,50"],
        svec!["2", "200,75"],
    ];
    wrk.create("data.csv", test_data);

    // Test: --decimal-comma with SKIP_INPUT should succeed since no validation is performed
    let mut cmd = wrk.command("sqlp");
    cmd.arg("SKIP_INPUT")
        .arg("--decimal-comma")
        .args(["--output", "output.csv"])
        .arg("select * from read_csv('data.csv')");

    wrk.assert_success(&mut cmd);
    let got: String = wrk.read_to_string("output.csv").unwrap();
    let expected = "id,value\n1,\"100,50\"\n2,\"200,75\"\n";
    assert_eq!(got, expected);
}

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

    // Create first table with columns "Value" and "Tag"
    wrk.create(
        "df1.csv",
        vec![
            svec!["Value", "Tag"],
            svec!["100", "hello"],
            svec!["200", "foo"],
        ],
    );

    // Create second table with different column names "Number" and "String"
    wrk.create(
        "df2.csv",
        vec![
            svec!["Number", "String"],
            svec!["300", "world"],
            svec!["400", "bar"],
        ],
    );

    // Test: UNION should combine tables positionally, using column names from first table
    let mut cmd = wrk.command("sqlp");
    cmd.args(["df1.csv", "df2.csv"]).arg(
        r#"SELECT u.* FROM (
            SELECT * FROM df1
            UNION
            SELECT * FROM df2
        ) u ORDER BY Value"#,
    );

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["Value", "Tag"],
        svec!["100", "hello"],
        svec!["200", "foo"],
        svec!["300", "world"],
        svec!["400", "bar"],
    ];
    assert_eq!(got, expected);
}

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

    wrk.create("data.csv", vec![svec!["id", "data"], svec!["1", "a,b,c"]]);

    // Test: UNNEST should unnest the array column
    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg("select first(id) as idf, unnest(string_to_array(data, ',')) as value from data");

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["idf", "value"],
        svec!["1", "a"],
        svec!["1", "b"],
        svec!["1", "c"],
    ];
    assert_eq!(got, expected);
}

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

    wrk.create(
        "data.csv",
        vec![
            svec!["id", "category", "value"],
            svec!["1", "A", "20"],
            svec!["2", "A", "10"],
            svec!["3", "A", "25"],
            svec!["4", "B", "10"],
            svec!["5", "B", "40"],
            svec!["6", "B", "25"],
            svec!["7", "C", "35"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv").arg(
        r#"SELECT
            value,
            ROW_NUMBER() OVER (ORDER BY value DESC, category DESC) AS row_num,
            RANK() OVER (ORDER BY value DESC, category DESC) AS rank,
            DENSE_RANK() OVER (ORDER BY value DESC, category DESC) AS dense_rank
        FROM data
        ORDER BY value, id DESC"#,
    );

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["value", "row_num", "rank", "dense_rank"],
        svec!["10", "6", "6", "6"],
        svec!["10", "7", "7", "7"],
        svec!["20", "5", "5", "5"],
        svec!["25", "3", "3", "3"],
        svec!["25", "4", "4", "4"],
        svec!["35", "2", "2", "2"],
        svec!["40", "1", "1", "1"],
    ];
    assert_eq!(got, expected);

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv").arg(
        r#"SELECT
            value,
            ROW_NUMBER() OVER (ORDER BY value) AS row_num,
            RANK() OVER (ORDER BY value) AS rank,
            DENSE_RANK() OVER (ORDER BY value) AS dense_rank
        FROM data
        ORDER BY value, id"#,
    );
    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["value", "row_num", "rank", "dense_rank"],
        svec!["10", "1", "1", "1"],
        svec!["10", "2", "1", "1"],
        svec!["20", "3", "3", "2"],
        svec!["25", "4", "4", "3"],
        svec!["25", "5", "4", "3"],
        svec!["35", "6", "6", "4"],
        svec!["40", "7", "7", "5"],
    ];
    assert_eq!(got, expected);

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv").arg(
        r#"SELECT
            category,
            value,
            ROW_NUMBER() OVER (PARTITION BY category ORDER BY value) AS row_num,
            RANK() OVER (PARTITION BY category ORDER BY value) AS rank,
            DENSE_RANK() OVER (PARTITION BY category ORDER BY value) AS dense
        FROM data
        ORDER BY category, value"#,
    );
    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["category", "value", "row_num", "rank", "dense"],
        svec!["A", "10", "1", "1", "1"],
        svec!["A", "20", "2", "2", "2"],
        svec!["A", "25", "3", "3", "3"],
        svec!["B", "10", "1", "1", "1"],
        svec!["B", "25", "2", "2", "2"],
        svec!["B", "40", "3", "3", "3"],
        svec!["C", "35", "1", "1", "1"],
    ];
    assert_eq!(got, expected);
}

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

    wrk.create(
        "data.csv",
        vec![
            svec!["id", "category", "value"],
            svec!["1", "A", "20"],
            svec!["2", "A", "10"],
            svec!["3", "A", "30"],
            svec!["4", "B", "15"],
            svec!["5", "B", "50"],
            svec!["6", "B", "30"],
            svec!["7", "C", "35"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv").arg(
        r#"SELECT
        category,
        value,
        SUM(value) OVER w AS "w:sum",
        MIN(value) OVER w AS "w:min",
        AVG(value) OVER w AS "w:avg",
      FROM data
      WINDOW w AS (PARTITION BY category ORDER BY value)
      ORDER BY category, value"#,
    );
    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["category", "value", "w:sum", "w:min", "w:avg"],
        svec!["A", "10", "10", "10", "20.0"],
        svec!["A", "20", "30", "10", "20.0"],
        svec!["A", "30", "60", "10", "20.0"],
        svec!["B", "15", "15", "15", "31.666666666666668"],
        svec!["B", "30", "45", "15", "31.666666666666668"],
        svec!["B", "50", "95", "15", "31.666666666666668"],
        svec!["C", "35", "35", "35", "35.0"],
    ];
    assert_eq!(got, expected);

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv").arg(
        r#"SELECT
        category,
        value,
        AVG(value) OVER w1 AS category_avg,
        SUM(value) OVER w2 AS running_value,
        COUNT(*) OVER w3 AS total_count
      FROM data
      WINDOW
        w1 AS (PARTITION BY category),
        w2 AS (ORDER BY value),
        w3 AS ()
      ORDER BY category, value"#,
    );
    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec![
            "category",
            "value",
            "category_avg",
            "running_value",
            "total_count"
        ],
        svec!["A", "10", "20.0", "10", "7"],
        svec!["A", "20", "20.0", "45", "7"],
        svec!["A", "30", "20.0", "75", "7"],
        svec!["B", "15", "31.666666666666668", "25", "7"],
        svec!["B", "30", "31.666666666666668", "105", "7"],
        svec!["B", "50", "31.666666666666668", "190", "7"],
        svec!["C", "35", "35.0", "140", "7"],
    ];
    assert_eq!(got, expected);
}

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

    wrk.create(
        "data.csv",
        vec![
            svec!["a", "b"],
            svec!["first", "1"],
            svec!["first", "1"],
            svec!["third", "42"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv").arg(
        r#"
        SELECT b, ARRAY_TO_STRING("a",', ') AS a2s,
        FROM (
            SELECT b, ARRAY_AGG(a) AS "a"
            FROM data
            GROUP BY b
        ) tbl
        ORDER BY a2s"#,
    );

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["b", "a2s"],
        svec!["1", "first, first"],
        svec!["42", "third"],
    ];
    assert_eq!(got, expected);
}

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

    wrk.create(
        "data.csv",
        vec![
            svec!["id", "data"],
            svec!["1", "a,b,c"],
            svec!["2", ""],
            svec!["3", "b,c,d"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .arg("select id, unnest(string_to_array(data, ',')) as value from data");

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["id", "value"],
        svec!["1", "a"],
        svec!["1", "b"],
        svec!["1", "c"],
        svec!["2", ""],
        svec!["3", "b"],
        svec!["3", "c"],
        svec!["3", "d"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_distinct_basic() {
    let wrk = Workdir::new("sqlp_distinct_basic");
    wrk.create(
        "data.csv",
        vec![
            svec!["category", "subcategory", "value", "status", "score"],
            svec!["A", "x", "100", "active", "10"],
            svec!["A", "x", "100", "active", "20"],
            svec!["B", "y", "200", "active", "30"],
            svec!["B", "y", "200", "inactive", "30"],
            svec!["B", "z", "300", "active", "40"],
            svec!["C", "x", "400", "inactive", "50"],
            svec!["C", "y", "500", "active", "60"],
            svec!["", "x", "600", "active", "70"],
            svec!["", "y", "700", "inactive", "80"],
            svec!["A", "x", "100", "active", "10"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .args(["--rnull-values", "<empty string>"])
        .args(["--wnull-value", "NULL"])
        .arg("SELECT DISTINCT category FROM data ORDER BY category NULLS LAST");

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["category"],
        svec!["A"],
        svec!["B"],
        svec!["C"],
        svec!["NULL"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_distinct_multiple_columns() {
    let wrk = Workdir::new("sqlp_distinct_multiple_columns");
    wrk.create(
        "data.csv",
        vec![
            svec!["category", "subcategory", "value", "status", "score"],
            svec!["A", "x", "100", "active", "10"],
            svec!["A", "x", "100", "active", "20"],
            svec!["B", "y", "200", "active", "30"],
            svec!["B", "y", "200", "inactive", "30"],
            svec!["B", "z", "300", "active", "40"],
            svec!["C", "x", "400", "inactive", "50"],
            svec!["C", "y", "500", "active", "60"],
            svec!["", "x", "600", "active", "70"],
            svec!["", "y", "700", "inactive", "80"],
            svec!["A", "x", "100", "active", "10"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .args(["--rnull-values", "<empty string>"])
        .args(["--wnull-value", "NULL"])
        .arg(
            "SELECT DISTINCT category, subcategory FROM data ORDER BY category NULLS LAST, \
             subcategory",
        );

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["category", "subcategory"],
        svec!["A", "x"],
        svec!["B", "y"],
        svec!["B", "z"],
        svec!["C", "x"],
        svec!["C", "y"],
        svec!["NULL", "x"],
        svec!["NULL", "y"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_distinct_with_nulls() {
    let wrk = Workdir::new("sqlp_distinct_with_nulls");
    wrk.create(
        "data.csv",
        vec![
            svec!["category", "subcategory", "value", "status", "score"],
            svec!["A", "x", "100", "active", "10"],
            svec!["A", "x", "100", "active", "20"],
            svec!["B", "y", "200", "active", "30"],
            svec!["B", "y", "200", "inactive", "30"],
            svec!["B", "z", "300", "active", "40"],
            svec!["C", "x", "400", "inactive", "50"],
            svec!["C", "y", "500", "active", "60"],
            svec!["", "x", "600", "active", "70"],
            svec!["", "y", "700", "inactive", "80"],
            svec!["A", "x", "100", "active", "10"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .args(["--rnull-values", "<empty string>"])
        .args(["--wnull-value", "NULL"])
        .arg("SELECT DISTINCT category FROM data ORDER BY category NULLS FIRST");

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["category"],
        svec!["NULL"],
        svec!["A"],
        svec!["B"],
        svec!["C"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_distinct_all_columns() {
    let wrk = Workdir::new("sqlp_distinct_all_columns");
    wrk.create(
        "data.csv",
        vec![
            svec!["category", "subcategory", "value", "status", "score"],
            svec!["A", "x", "100", "active", "10"],
            svec!["A", "x", "100", "active", "20"],
            svec!["B", "y", "200", "active", "30"],
            svec!["B", "y", "200", "inactive", "30"],
            svec!["B", "z", "300", "active", "40"],
            svec!["C", "x", "400", "inactive", "50"],
            svec!["C", "y", "500", "active", "60"],
            svec!["", "x", "600", "active", "70"],
            svec!["", "y", "700", "inactive", "80"],
            svec!["A", "x", "100", "active", "10"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .args(["--rnull-values", "<empty string>"])
        .args(["--wnull-value", "NULL"])
        .arg(
            "SELECT DISTINCT * FROM data ORDER BY category NULLS LAST, subcategory, value, \
             status, score",
        );

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["category", "subcategory", "value", "status", "score"],
        svec!["A", "x", "100", "active", "10"],
        svec!["A", "x", "100", "active", "20"],
        svec!["B", "y", "200", "active", "30"],
        svec!["B", "y", "200", "inactive", "30"],
        svec!["B", "z", "300", "active", "40"],
        svec!["C", "x", "400", "inactive", "50"],
        svec!["C", "y", "500", "active", "60"],
        svec!["NULL", "x", "600", "active", "70"],
        svec!["NULL", "y", "700", "inactive", "80"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_distinct_on_basic() {
    let wrk = Workdir::new("sqlp_distinct_on_basic");
    wrk.create(
        "data.csv",
        vec![
            svec!["category", "subcategory", "value", "status", "score"],
            svec!["A", "x", "100", "active", "10"],
            svec!["A", "x", "100", "active", "20"],
            svec!["B", "y", "200", "active", "30"],
            svec!["B", "y", "200", "inactive", "30"],
            svec!["B", "z", "300", "active", "40"],
            svec!["C", "x", "400", "inactive", "50"],
            svec!["C", "y", "500", "active", "60"],
            svec!["", "x", "600", "active", "70"],
            svec!["", "y", "700", "inactive", "80"],
            svec!["A", "x", "100", "active", "10"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .args(["--rnull-values", "<empty string>"])
        .args(["--wnull-value", "NULL"])
        .arg(
            "SELECT DISTINCT ON (category) category, value, status FROM data ORDER BY category \
             NULLS LAST, value DESC",
        );

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["category", "value", "status"],
        svec!["A", "100", "active"],
        svec!["B", "300", "active"],
        svec!["C", "500", "active"],
        svec!["NULL", "700", "inactive"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_distinct_on_multiple_keys() {
    let wrk = Workdir::new("sqlp_distinct_on_multiple_keys");
    wrk.create(
        "data.csv",
        vec![
            svec!["category", "subcategory", "value", "status", "score"],
            svec!["A", "x", "100", "active", "10"],
            svec!["A", "x", "100", "active", "20"],
            svec!["B", "y", "200", "active", "30"],
            svec!["B", "y", "200", "inactive", "30"],
            svec!["B", "z", "300", "active", "40"],
            svec!["C", "x", "400", "inactive", "50"],
            svec!["C", "y", "500", "active", "60"],
            svec!["", "x", "600", "active", "70"],
            svec!["", "y", "700", "inactive", "80"],
            svec!["A", "x", "100", "active", "10"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .args(["--rnull-values", "<empty string>"])
        .args(["--wnull-value", "NULL"])
        .arg(
            "SELECT DISTINCT ON (category, subcategory) * FROM data ORDER BY category NULLS LAST, \
             subcategory, score DESC",
        );

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["category", "subcategory", "value", "status", "score"],
        svec!["A", "x", "100", "active", "20"],
        svec!["B", "y", "200", "active", "30"],
        svec!["B", "z", "300", "active", "40"],
        svec!["C", "x", "400", "inactive", "50"],
        svec!["C", "y", "500", "active", "60"],
        svec!["NULL", "x", "600", "active", "70"],
        svec!["NULL", "y", "700", "inactive", "80"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_distinct_on_with_nulls() {
    let wrk = Workdir::new("sqlp_distinct_on_with_nulls");
    wrk.create(
        "data.csv",
        vec![
            svec!["category", "subcategory", "value", "status", "score"],
            svec!["A", "x", "100", "active", "10"],
            svec!["A", "x", "100", "active", "20"],
            svec!["B", "y", "200", "active", "30"],
            svec!["B", "y", "200", "inactive", "30"],
            svec!["B", "z", "300", "active", "40"],
            svec!["C", "x", "400", "inactive", "50"],
            svec!["C", "y", "500", "active", "60"],
            svec!["", "x", "600", "active", "70"],
            svec!["", "y", "700", "inactive", "80"],
            svec!["A", "x", "100", "active", "10"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .args(["--rnull-values", "<empty string>"])
        .args(["--wnull-value", "NULL"])
        .arg("SELECT DISTINCT ON (category) * FROM data ORDER BY category NULLS LAST, value DESC");

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["category", "subcategory", "value", "status", "score"],
        svec!["A", "x", "100", "active", "10"],
        svec!["B", "z", "300", "active", "40"],
        svec!["C", "y", "500", "active", "60"],
        svec!["NULL", "y", "700", "inactive", "80"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_distinct_on_expression() {
    // Note: DISTINCT ON only supports column names, not expressions.
    // This test verifies that DISTINCT ON works with a computed column in ORDER BY
    // by using a column alias approach instead.
    let wrk = Workdir::new("sqlp_distinct_on_expression");
    wrk.create(
        "data.csv",
        vec![
            svec!["category", "subcategory", "value", "status", "score"],
            svec!["A", "x", "100", "active", "10"],
            svec!["A", "x", "100", "active", "20"],
            svec!["B", "y", "200", "active", "30"],
            svec!["B", "y", "200", "inactive", "30"],
            svec!["B", "z", "300", "active", "40"],
            svec!["C", "x", "400", "inactive", "50"],
            svec!["C", "y", "500", "active", "60"],
            svec!["", "x", "600", "active", "70"],
            svec!["", "y", "700", "inactive", "80"],
            svec!["A", "x", "100", "active", "10"],
        ],
    );

    // Since DISTINCT ON doesn't support expressions, we test with a column name
    // but use an expression in ORDER BY to verify ordering works correctly
    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .args(["--rnull-values", "<empty string>"])
        .args(["--wnull-value", "NULL"])
        .arg(
            "SELECT DISTINCT ON (category) category, value FROM data ORDER BY category NULLS \
             LAST, value DESC",
        );

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["category", "value"],
        svec!["A", "100"],
        svec!["B", "300"],
        svec!["C", "500"],
        svec!["NULL", "700"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_distinct_with_aggregation() {
    let wrk = Workdir::new("sqlp_distinct_with_aggregation");
    wrk.create(
        "data.csv",
        vec![
            svec!["category", "subcategory", "value", "status", "score"],
            svec!["A", "x", "100", "active", "10"],
            svec!["A", "x", "100", "active", "20"],
            svec!["B", "y", "200", "active", "30"],
            svec!["B", "y", "200", "inactive", "30"],
            svec!["B", "z", "300", "active", "40"],
            svec!["C", "x", "400", "inactive", "50"],
            svec!["C", "y", "500", "active", "60"],
            svec!["", "x", "600", "active", "70"],
            svec!["", "y", "700", "inactive", "80"],
            svec!["A", "x", "100", "active", "10"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .args(["--rnull-values", "<empty string>"])
        .args(["--wnull-value", "NULL"])
        .arg(
            "SELECT DISTINCT category, COUNT(*) as cnt FROM data GROUP BY category ORDER BY \
             category NULLS LAST",
        );

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["category", "cnt"],
        svec!["A", "3"],
        svec!["B", "3"],
        svec!["C", "2"],
        svec!["NULL", "2"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_distinct_on_complex_ordering() {
    let wrk = Workdir::new("sqlp_distinct_on_complex_ordering");
    wrk.create(
        "data.csv",
        vec![
            svec!["category", "subcategory", "value", "status", "score"],
            svec!["A", "x", "100", "active", "10"],
            svec!["A", "x", "100", "active", "20"],
            svec!["B", "y", "200", "active", "30"],
            svec!["B", "y", "200", "inactive", "30"],
            svec!["B", "z", "300", "active", "40"],
            svec!["C", "x", "400", "inactive", "50"],
            svec!["C", "y", "500", "active", "60"],
            svec!["", "x", "600", "active", "70"],
            svec!["", "y", "700", "inactive", "80"],
            svec!["A", "x", "100", "active", "10"],
        ],
    );

    let mut cmd = wrk.command("sqlp");
    cmd.arg("data.csv")
        .args(["--rnull-values", "<empty string>"])
        .args(["--wnull-value", "NULL"])
        .arg(
            "SELECT DISTINCT ON (category) category, value, status FROM data ORDER BY category \
             NULLS LAST, value DESC, status ASC",
        );

    wrk.assert_success(&mut cmd);
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["category", "value", "status"],
        svec!["A", "100", "active"],
        svec!["B", "300", "active"],
        svec!["C", "500", "active"],
        svec!["NULL", "700", "inactive"],
    ];
    assert_eq!(got, expected);
}

#[cfg(not(feature = "datapusher_plus"))]
#[test]
fn sqlp_datetime_schema_inference() {
    let wrk = Workdir::new("sqlp_datetime_schema_inference");

    // Create a CSV with datetime columns
    wrk.create(
        "datetimes.csv",
        vec![
            svec!["id", "name", "created_at", "updated_at", "amount"],
            svec![
                "1",
                "Alice",
                "2024-01-15 10:30:00",
                "2024-02-01 14:00:00",
                "100.50"
            ],
            svec![
                "2",
                "Bob",
                "2024-03-20 09:15:00",
                "2024-04-10 16:45:00",
                "200.75"
            ],
            svec![
                "3",
                "Charlie",
                "2024-05-05 08:00:00",
                "2024-06-15 12:30:00",
                "300.25"
            ],
        ],
    );

    // Step 1: Generate stats cache with --infer-dates so DateTime type is detected
    let mut cmd = wrk.command("stats");
    cmd.arg("datetimes.csv")
        .arg("--infer-dates")
        .args(["--dates-whitelist", "all"])
        .arg("--cardinality")
        .arg("--stats-jsonl")
        .args(["--cache-threshold", "1"]);
    wrk.assert_success(&mut cmd);

    // Verify stats detected DateTime columns
    let stats_output: String = wrk.stdout(&mut cmd);
    assert!(stats_output.contains("created_at,DateTime"));
    assert!(stats_output.contains("updated_at,DateTime"));

    // Step 2: Generate Polars schema from stats cache
    let mut cmd = wrk.command("schema");
    cmd.arg("--polars").arg("datetimes.csv");
    wrk.assert_success(&mut cmd);

    // Verify the pschema.json file was created and contains Datetime type
    assert!(wrk.path("datetimes.csv.pschema.json").exists());
    let schema_json = std::fs::read_to_string(wrk.path("datetimes.csv.pschema.json")).unwrap();
    assert!(
        schema_json.contains(r#""Datetime""#),
        "Schema should contain Datetime type, got: {schema_json}"
    );
    assert!(
        schema_json.contains(r#""Milliseconds""#),
        "Schema should contain Milliseconds time unit, got: {schema_json}"
    );

    // Step 3: Use sqlp with the cached schema to verify it parses datetimes correctly
    let mut cmd = wrk.command("sqlp");
    cmd.arg("datetimes.csv")
        .arg("SELECT id, name, created_at, updated_at FROM datetimes LIMIT 1")
        .arg("--cache-schema");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["id", "name", "created_at", "updated_at"],
        svec![
            "1",
            "Alice",
            "2024-01-15T10:30:00.000",
            "2024-02-01T14:00:00.000"
        ],
    ];
    assert_eq!(got, expected);
}

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

    // Create a CSV with datetime columns
    wrk.create(
        "events.csv",
        vec![
            svec!["id", "event_name", "event_time", "score"],
            svec!["1", "login", "2024-06-15 08:30:00", "95.5"],
            svec!["2", "logout", "2024-06-15 17:00:00", "88.0"],
        ],
    );

    // Manually create a pschema.json with Datetime type
    wrk.create_from_string(
        "events.csv.pschema.json",
        r#"{
  "fields": {
    "id": "UInt8",
    "event_name": "String",
    "event_time": {
      "Datetime": [
        "Milliseconds",
        null
      ]
    },
    "score": "Float32"
  },
  "metadata": null
}"#,
    );

    // sqlp should use the manually provided schema and parse datetimes
    let mut cmd = wrk.command("sqlp");
    cmd.arg("events.csv")
        .arg("SELECT id, event_name, event_time, score FROM events")
        .arg("--cache-schema");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["id", "event_name", "event_time", "score"],
        svec!["1", "login", "2024-06-15T08:30:00.000", "95.5"],
        svec!["2", "logout", "2024-06-15T17:00:00.000", "88.0"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_alias_no_substring_collision() {
    // Regression: `_t_1` must not match the prefix of `_t_10`. Before the
    // word-boundary regex fix, the naive `String::replace` would rewrite
    // `_t_10` into `"d1"0`, producing a SQL parse error.
    let wrk = Workdir::new("sqlp_alias_no_substring_collision");

    // Create 10 input files so we have aliases _t_1 .. _t_10.
    for i in 1..=10 {
        wrk.create(
            &format!("d{i}.csv"),
            vec![vec!["v".to_string()], vec![format!("row_{i}")]],
        );
    }

    let mut cmd = wrk.command("sqlp");
    for i in 1..=10 {
        cmd.arg(format!("d{i}.csv"));
    }
    cmd.arg("select v from _t_10");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![svec!["v"], svec!["row_10"]];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_alias_preserves_literal_with_alias_substring() {
    // Regression: a string literal containing an alias-shaped token (e.g.
    // `_t_10_note`) must survive the rewrite untouched, because `\d+` is
    // greedy and the next char (`_`) does not form a word boundary with `0`.
    // (Alias text inside literals delimited only by quotes is *not* protected
    // — that is a documented limitation of the regex-based approach.)
    let wrk = Workdir::new("sqlp_alias_preserves_literal_with_alias_substring");
    wrk.create("d1.csv", vec![svec!["v"], svec!["one"]]);

    let mut cmd = wrk.command("sqlp");
    cmd.arg("d1.csv")
        .arg("select '_t_10_note' as k, v from _t_1");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![svec!["k", "v"], svec!["_t_10_note", "one"]];
    assert_eq!(got, expected);
}

#[test]
fn sqlp_invalid_format_errors() {
    // Regression: an invalid --format value should fail loudly instead of
    // silently falling back to CSV.
    let wrk = Workdir::new("sqlp_invalid_format_errors");
    wrk.create("d.csv", vec![svec!["v"], svec!["1"]]);

    let mut cmd = wrk.command("sqlp");
    cmd.args(["d.csv", "select * from d", "--format", "jsonn"]);

    wrk.assert_err(&mut cmd);
}
