odbc2parquet 11.0.1

use std::{
    fs::File,
    io::{ErrorKind, Write},
    path::{Path, PathBuf},
    str,
    sync::{Arc, Once},
};

use assert_cmd::{assert::Assert, cargo::cargo_bin_cmd, Command};
use bytes::Bytes;
use odbc_api::{
    buffers::TextRowSet, environment, sys::AttrConnectionPooling, BindParamDesc, Connection,
    ConnectionOptions, Cursor, Environment, IntoParameter,
};
use parquet::{
    column::writer::ColumnWriter,
    data_type::{ByteArray, FixedLenByteArray},
    file::{
        properties::WriterProperties, reader::FileReader, serialized_reader::SerializedFileReader,
        writer::SerializedFileWriter,
    },
    schema::parser::parse_message_type,
};
use predicates::{ord::eq, str::contains};
use tempfile::{tempdir, NamedTempFile, TempDir};

const MSSQL: &str = "Driver={ODBC Driver 18 for SQL Server};\
    Server=localhost;\
    UID=SA;\
    PWD=My@Test@Password1;\
    TrustServerCertificate=yes;";

const POSTGRES: &str = "Driver={PostgreSQL UNICODE};\
    Server=localhost;\
    Port=5432;\
    Database=test;\
    Uid=test;\
    Pwd=test;";

fn env() -> &'static Environment {
    Once::new().call_once(|| {
        // Enable connection pools for faster test execution.
        //
        // # Safety
        //
        // This is safe because it is called once in the entire process. Therefore, no race to this
        // setting can occur. It is also called before initializing the environment, making it
        // certain that the newly created environment knows it is supposed to initialize the
        // connection pools.
        unsafe {
            Environment::set_connection_pooling(AttrConnectionPooling::OnePerDriver)
                .expect("ODBC manager must be able to initialize connection pools");
        }
    });
    environment().unwrap()
}

// Use the parquet-read tool to verify values. It can be installed with `cargo install parquet`.
fn parquet_read_out(file: &str) -> Assert {
    let mut cmd = Command::new("parquet-read");
    cmd.args([file]).assert().success()
}

/// Assertions on the output of parquet_schema.
fn parquet_schema_out(file: &str) -> Assert {
    let mut cmd = Command::new("parquet-schema");
    cmd.args([file]).assert().success()
}

/// Query MSSQL database, yet do not specify username and password in the connection string, but
/// pass them as separate command line options.
#[test]
fn append_user_and_password_to_connection_string() {
    // Setup table for test
    let table_name = "AppendUserAndPasswordToConnectionString";
    TableMssql::new(table_name, &["VARCHAR(10)"]);

    // Connection string without user name and password.
    let connection_string = "Driver={ODBC Driver 18 for SQL Server};Server=localhost;\
        TrustServerCertificate=yes;";
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name}");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            "--connection-string",
            connection_string,
            "--user",
            "SA",
            "--password",
            "My@Test@Password1",
            out_str,
            &query,
        ])
        .assert()
        .success();
}

#[test]
fn insert() {
    roundtrip("insert.par", "odbc2parquet_insert").success();
}

#[test]
fn insert_empty_document() {
    roundtrip("empty_document.par", "odbc2parquet_empty_document").success();
}

#[test]
fn nullable_parquet_buffers() {
    // Setup table for test
    let table_name = "NullableParquetBuffers";
    let mut table = TableMssql::new(table_name, &["VARCHAR(10)"]);
    table.insert_rows_as_text(&[[Some("Hello")], [None], [Some("World")], [None]]);

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name}");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    let expected = "\
        {a: \"Hello\"}\n\
        {a: null}\n\
        {a: \"World\"}\n\
        {a: null}\n\
    ";
    parquet_read_out(out_str).stdout(eq(expected));
}

#[test]
fn foobar_connection_string() {
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Tempfile path must be utf8");

    let mut cmd = cargo_bin_cmd!();
    cmd.args([
        "-vvvv",
        "query",
        "-c",
        "foobar",
        out_str,
        "SELECT * FROM [uk-500$]",
    ])
    .assert()
    .failure()
    .code(1);
}

#[test]
fn should_give_good_error_if_specifying_directory_for_output() {
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // This SHOULD be a path to a file to create, but in this test, we specify a directory.
    let out_path = out_dir.path();
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Tempfile path must be utf8");

    let mut cmd = cargo_bin_cmd!();
    cmd.args(["query", "-c", MSSQL, out_str, "SELECT 42 AS A"])
        .assert()
        .stderr(contains("Could not create output file '"))
        .failure()
        .code(1);
}

#[test]
fn parameters_in_query() {
    // Setup table for test
    let table_name = "ParametersInQuery";
    let mut table = TableMssql::new(table_name, &["VARCHAR(10)", "INTEGER"]);
    table.insert_rows_as_text(&[["Wrong", "5"], ["Right", "42"]]);

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a,b FROM {table_name} WHERE b=?");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
            "42",
        ])
        .assert()
        .success();

    let expected = "\
        {a: \"Right\", b: 42}\n\
    ";
    parquet_read_out(out_str).stdout(eq(expected));
}

#[test]
fn should_error_on_truncation_utf_8() {
    // Setup table for test
    let table_name = "ErrorOnTruncationUtf8";
    let mut table = TableMssql::new(table_name, &["VARCHAR(10)"]);
    table.insert_rows_as_text(&[["0123456789"]]);

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name}");

    let assertion = cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--encoding",
            "system",
            "--connection-string",
            MSSQL,
            "--column-length-limit",
            "5",
            &query,
        ])
        .assert();

    // For UTF-8 the MSSQL driver does return a length indicator.
    let expectation = "A field exceeds the maximum element length of a column buffer. You can use \
        the `--column-length-limit` option to increase the maximum element size of columns. The \
        driver indicated an actual length of 10. The error occurred for column a.";
    assertion.failure().stderr(contains(expectation));
}

#[test]
fn should_error_on_truncation_utf_16() {
    // Setup table for test
    let table_name = "ErrorOnTruncationUtf16";
    let mut table = TableMssql::new(table_name, &["VARCHAR(10)"]);
    table.insert_rows_as_text(&[["0123456789"]]);

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name}");

    let assertion = cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--encoding",
            "Utf16",
            "--connection-string",
            MSSQL,
            "--column-length-limit",
            "5",
            &query,
        ])
        .assert();

    // For UTF-16 the MSSQL driver does not return a length indicator.
    let expectation = "A field exceeds the maximum element length of a column buffer. You can use \
        the `--column-length-limit` option to increase the maximum element size of columns. Sadly \
        the driver did not return a length indicator for the value, so you will have to guess its \
        actual length. The error occurred for column a.";
    assertion.failure().stderr(contains(expectation));
}

#[test]
fn should_error_on_truncation_for_sequential_fetch() {
    // Setup table for test
    let table_name = "ErrorOnTruncationForSequentialFetch";
    let mut table = TableMssql::new(table_name, &["VARCHAR(10)"]);
    table.insert_rows_as_text(&[["0123456789"]]);

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name}");

    let assertion = cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            "--sequential-fetching",
            "--column-length-limit",
            "5",
            &query,
        ])
        .assert();

    #[cfg(target_os = "windows")]
    let expectation = "A field exceeds the maximum element length of a column buffer. You can \
        use the `--column-length-limit` option to increase the maximum element size of columns. \
        Sadly the driver did not return a length indicator for the value, so you will have to \
        guess its actual length. The error occurred for column a.";
    #[cfg(not(target_os = "windows"))]
    let expectation = "A field exceeds the maximum element length of a column buffer. You can \
        use the `--column-length-limit` option to increase the maximum element size of columns. \
        The driver indicated an actual length of 10. The error occurred for column a.";
    assertion.failure().stderr(contains(expectation));
}

#[test]
fn should_allow_specifying_explicit_compression_level() {
    // Setup table for test
    let table_name = "ShouldAllowSpecifyingExplicitCompressionLevel";
    let mut table = TableMssql::new(table_name, &["VARCHAR(10)"]);
    table.insert_rows_as_text(&[["Hello"], ["World"]]);
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name}");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            "--connection-string",
            MSSQL,
            "--column-compression-default",
            "gzip",
            "--column-compression-level-default",
            "9",
            out_str,
            &query,
            "42",
        ])
        .assert()
        .success();
}

#[test]
fn query_sales() {
    // Setup table for test
    let table_name = "QuerySales";
    let mut table = TableMssql::new(table_name, &["DATE", "TIME(7)", "INT", "DECIMAL(10,2)"]);
    table.insert_rows_as_text(&[
        ["2020-09-09", "00:05:34", "54", "9.99"],
        ["2020-09-10", "12:05:32", "54", "9.99"],
        ["2020-09-10", "14:05:32", "34", "2.00"],
        ["2020-09-11", "06:05:12", "12", "-1.50"],
    ]);
    let expected_values = "\
        {a: 2020-09-09, b: 334000000000, c: 54, d: 9.99}\n\
        {a: 2020-09-10, b: 43532000000000, c: 54, d: 9.99}\n\
        {a: 2020-09-10, b: 50732000000000, c: 34, d: 2.00}\n\
        {a: 2020-09-11, b: 21912000000000, c: 12, d: -1.50}\n\
    ";
    let query = format!("SELECT a,b,c,d FROM {table_name} ORDER BY id");
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Tempfile path must be utf8");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    parquet_read_out(out_str).stdout(eq(expected_values));
}

#[test]
fn query_sales_sequentially() {
    // Setup table for test
    let table_name = "QuerySalesSequentially";
    let mut table = TableMssql::new(table_name, &["DATE", "TIME(7)", "INT", "DECIMAL(10,2)"]);
    table.insert_rows_as_text(&[
        ["2020-09-09", "00:05:34", "54", "9.99"],
        ["2020-09-10", "12:05:32", "54", "9.99"],
        ["2020-09-10", "14:05:32", "34", "2.00"],
        ["2020-09-11", "06:05:12", "12", "-1.50"],
    ]);
    let expected_values = "\
        {a: 2020-09-09, b: 334000000000, c: 54, d: 9.99}\n\
        {a: 2020-09-10, b: 43532000000000, c: 54, d: 9.99}\n\
        {a: 2020-09-10, b: 50732000000000, c: 34, d: 2.00}\n\
        {a: 2020-09-11, b: 21912000000000, c: 12, d: -1.50}\n\
    ";
    let query = format!("SELECT a,b,c,d FROM {table_name} ORDER BY id");
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Tempfile path must be utf8");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            "--sequential-fetching",
            &query,
        ])
        .assert()
        .success();

    parquet_read_out(out_str).stdout(eq(expected_values));
}

#[test]
fn query_decimals() {
    // Setup table for test
    let table_name = "QueryDecimals";
    let mut table = TableMssql::new(
        table_name,
        &[
            "NUMERIC(3,2) NOT NULL",
            "DECIMAL(3,2) NOT NULL",
            "DECIMAL(3,0) NOT NULL",
            "DECIMAL(10,0) NOT NULL",
        ],
    );
    table.insert_rows_as_text(&[["1.23", "1.23", "3", "1234567890"]]);
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a,b,c,d FROM {table_name};");
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    let expected_values = "{a: 1.23, b: 1.23, c: 3., d: 1234567890.}\n";
    parquet_read_out(out_str).stdout(eq(expected_values));

    parquet_schema_out(out_str).stdout(contains(
        "message schema {\n  REQUIRED INT32 a (DECIMAL(3,2));\n  \
                REQUIRED INT32 b (DECIMAL(3,2));\n  \
                REQUIRED INT32 c (DECIMAL(3,0));\n  \
                REQUIRED INT64 d (DECIMAL(10,0));\n\
            }",
    ));
}

/// Produce output for downstream artefacts like polars which lack support for decimal. In effect
/// logical type decimal should not show up in the output
#[test]
fn query_decimals_avoid_decimal() {
    // Setup table for test
    let table_name = "QueryDecimalsAvoidDecimal";
    let mut table = TableMssql::new(
        table_name,
        &[
            "NUMERIC(3,2) NOT NULL",
            "DECIMAL(3,2) NOT NULL",
            "DECIMAL(3,0) NOT NULL",
            "DECIMAL(10,0) NOT NULL",
        ],
    );
    table.insert_rows_as_text(&[["1.23", "1.23", "3", "1234567890"]]);
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a,b,c,d FROM {table_name};");
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            "--avoid-decimal",
            &query,
        ])
        .assert()
        .success();

    let expected_values = "{a: \"1.23\", b: \"1.23\", c: 3, d: 1234567890}\n";
    parquet_read_out(out_str).stdout(eq(expected_values));

    parquet_schema_out(out_str).stdout(contains(
        "message schema {\n  \
                REQUIRED BYTE_ARRAY a (UTF8);\n  \
                REQUIRED BYTE_ARRAY b (UTF8);\n  \
                REQUIRED INT32 c (INTEGER(32,true));\n  \
                REQUIRED INT64 d (INTEGER(64,true));\n\
            }",
    ));
}

/// Combination of avoid-decimal and int64-not-supported by driver. E.g. querying Decimal columns
/// from Oracle and using the output in polars
#[test]
fn query_decimals_avoid_decimal_int64_not_supported_by_driver() {
    // Setup table for test
    let table_name = "QueryDecimalsAvoidDecimalInt64NotSupportedByDriver";
    let mut table = TableMssql::new(table_name, &["DECIMAL(10,0) NOT NULL"]);
    table.insert_rows_as_text(&[["1234567890"]]);
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name};");
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            "--avoid-decimal",
            "--driver-does-not-support-64bit-integers",
            &query,
        ])
        .assert()
        .success();

    let expected_values = "{a: 1234567890}\n";
    parquet_read_out(out_str).stdout(eq(expected_values));

    parquet_schema_out(out_str).stdout(contains(
        "message schema {\n  \
                REQUIRED INT64 a (INTEGER(64,true));\n\
            }",
    ));
}

#[test]
fn query_decimals_optional() {
    // Setup table for test
    let table_name = "QueryDecimalsOptional";
    let mut table = TableMssql::new(
        table_name,
        &["NUMERIC(3,2)", "DECIMAL(3,2)", "DECIMAL(3,0)"],
    );
    table.insert_rows_as_text(&[
        [Some("1.23"), Some("1.23"), Some("123")],
        [None, None, None],
        [Some("4.56"), Some("4.56"), Some("456")],
    ]);
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a,b,c FROM {table_name} ORDER BY id;");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    let expected_values =
        "{a: 1.23, b: 1.23, c: 123.}\n{a: null, b: null, c: null}\n{a: 4.56, b: 4.56, c: 456.}\n";
    parquet_read_out(out_str).stdout(eq(expected_values));
}

/// Query a numeric/decimal which could be represented is i64 as text instead to accommodate missing
/// driver support for i64
#[test]
fn query_large_numeric_as_text() {
    // Setup table for test
    let table_name = "QueryLargeNumericAsText";
    let mut table = TableMssql::new(table_name, &["NUMERIC(10,0) NOT NULL"]);
    table.insert_rows_as_text(&[["1234567890"]]);
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name};");
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            "--driver-does-not-support-64bit-integers",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    parquet_read_out(out_str).stdout(eq("{a: 1234567890.}\n"));
    parquet_schema_out(out_str).stdout(contains("{\n  REQUIRED INT64 a (DECIMAL(10,0));\n}"));
}

#[test]
fn query_numeric_13_3() {
    // Setup table for test
    let table_name = "QueryNumeric13_3";
    let mut table = TableMssql::new(table_name, &["NUMERIC(13,3) NOT NULL"]);
    table.insert_rows_as_text(&[["-1234567890.123"]]);
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");
    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    let expected_values = "{a: -1234567890.123}\n";
    parquet_read_out(out_str).stdout(eq(expected_values));

    parquet_schema_out(out_str).stdout(contains("{\n  REQUIRED INT64 a (DECIMAL(13,3));\n}"));
}

#[test]
fn query_numeric_33_3() {
    // Setup table for test
    let table_name = "QueryNumeric33_3";
    let mut table = TableMssql::new(table_name, &["NUMERIC(33,3) NOT NULL"]);
    table.insert_rows_as_text(&[["-123456789012345678901234567890.123"]]);
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");
    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    let expected_values = "{a: -123456789012345678901234567890.123}\n";
    parquet_read_out(out_str).stdout(eq(expected_values));

    parquet_schema_out(out_str).stdout(contains(
        "{\n  REQUIRED FIXED_LEN_BYTE_ARRAY (14) a (DECIMAL(33,3));\n}",
    ));
}

#[test]
fn query_timestamp_with_timezone_mssql() {
    // Setup table for test
    let table_name = "QueryTimestampWithTimezone";
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    // ODBC data type: SqlDataType(-155), column_size: 34, decimal_digits: 7
    setup_empty_table_mssql(&conn, table_name, &["DATETIMEOFFSET"]).unwrap();
    let insert = format!(
        "INSERT INTO {table_name}
        (a)
        VALUES
        ('2022-09-07 16:04:12.1234567 +02:00');"
    );
    conn.execute(&insert, (), None).unwrap();
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");
    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    let expected_values = "{a: 1662559452123456700}\n";
    parquet_read_out(out_str).stdout(eq(expected_values));

    parquet_schema_out(out_str).stdout(contains("OPTIONAL INT64 a (TIMESTAMP(NANOS,true));"));
}

#[test]
fn query_timestamp_mssql_precision_7() {
    // Setup table for test
    let table_name = "QueryTimestampMssqlPrecision7";
    let mut table = TableMssql::new(table_name, &["DATETIME2(7)"]);
    table.insert_rows_as_text(&[["2022-09-07 16:04:12.1234567"]]);
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");
    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    let expected_values = "{a: 1662566652123456700}\n";
    parquet_read_out(out_str).stdout(eq(expected_values));

    parquet_schema_out(out_str).stdout(contains("OPTIONAL INT64 a (TIMESTAMP(NANOS,false));"));
}

#[test]
fn query_unsigned_tinyint() {
    // Setup table for test
    let table_name = "QueryUnsignedTinyInt";
    let mut table = TableMssql::new(table_name, &["TINYINT"]);
    table.insert_rows_as_text(&[["255"]]);
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");
    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    let expected_values = "{a: 255}\n";
    parquet_read_out(out_str).stdout(eq(expected_values));

    parquet_schema_out(out_str).stdout(contains("OPTIONAL INT32 a (INTEGER(8,false));"));
}

#[test]
fn should_error_if_timestamp_is_out_of_range() {
    // Setup table for test
    let table_name = "ShouldErrorIfTimestampIsOutOfRange";
    let mut table = TableMssql::new(table_name, &["DATETIME2(7)"]);
    table.insert_rows_as_text(&[["2700-01-01 00:00:00"]]);
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");
    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .failure()
        .stderr(contains(
            "Invalid timestamp: 2700-01-01 00:00:00. The valid range for timestamps with \
            nano seconds precision is between 1677-09-21 00:12:44 and 2262-04-11 \
            23:47:16.854775807. Other timestamps can not be represented in parquet. To mitigate \
            this you could downcast the precision in the query or convert the column to text.",
        ));
}

#[test]
fn should_correctly_fetch_upper_bound_timestamp() {
    // Setup table for test
    let table_name = "ShouldCorrectlyFetchUpperBoundTimestamp";
    let mut table = TableMssql::new(table_name, &["DATETIME2(7)"]);
    // Due to precision the last 7 nanoseconds are excluded from the upper bound.
    table.insert_rows_as_text(&[["2262-04-11 23:47:16.854775807"]]);
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");
    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    // Binary of the expectation is all `1`, except for the first (sign) and the last three digits
    // (precision of MsSQL is not nanoseconds, but 7 digits).
    let expected_values = "{a: 9223372036854775800}\n";
    parquet_read_out(out_str).stdout(eq(expected_values));
}

#[test]
fn should_correctly_fetch_lower_bound_timestamp() {
    // Setup table for test
    let table_name = "ShouldCorrectlyFetchLowerBound";
    let mut table = TableMssql::new(table_name, &["DATETIME2(7)"]);
    // Due to precision the last 7 nanoseconds are excluded from the lower bound, so we round up
    table.insert_rows_as_text(&[["1677-09-21 00:12:44"]]);
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");
    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    // Binary of the expectation is all `1`, execpt for the first (sign) and the last three digits
    // (precision of MsSQL is not nano seconds, but 7 digits).
    let expected_values = "{a: -9223372036000000000}\n";
    parquet_read_out(out_str).stdout(eq(expected_values));
}

#[test]
fn query_timestamp_ms_with_timezone_mssql() {
    // Setup table for test
    let table_name = "QueryTimestampMsWithTimezone";
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    // ODBC data type: SqlDataType(-155), decimal_digits: 3
    setup_empty_table_mssql(&conn, table_name, &["DATETIMEOFFSET(3)"]).unwrap();
    let insert = format!(
        "INSERT INTO {table_name}
        (a)
        VALUES
        ('2022-09-07 16:04:12 +02:00');"
    );
    conn.execute(&insert, (), None).unwrap();
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");
    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    let expected_values = "{a: 2022-09-07 14:04:12.000 +00:00}\n";
    parquet_read_out(out_str).stdout(eq(expected_values));

    parquet_schema_out(out_str).stdout(contains("OPTIONAL INT64 a (TIMESTAMP(MILLIS,true));"));
}

#[test]
fn query_time_mssql() {
    // Setup table for test
    let table_name = "QueryTime";
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["TIME"]).unwrap();
    let insert = format!(
        "INSERT INTO {table_name}
        (a)
        VALUES
        ('16:04:12');"
    );
    conn.execute(&insert, (), None).unwrap();
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");
    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    let expected_values = "{a: 57852000000000}\n";
    parquet_read_out(out_str).stdout(eq(expected_values));

    parquet_schema_out(out_str).stdout(contains("OPTIONAL INT64 a (TIME(NANOS,false));"));
}

#[test]
fn query_time_0_mssql() {
    // Setup table for test
    let table_name = "QueryTime0";
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["TIME(0)"]).unwrap();
    let insert = format!(
        "INSERT INTO {table_name}
        (a)
        VALUES
        ('16:04:12');"
    );
    conn.execute(&insert, (), None).unwrap();
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");
    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    // We can not use parquet read, as it uses the record API, which does not allow for the TIME type.
    // let expected_values = "{a: 57852000000000}\n";
    // parquet_read_out(out_str).stdout(eq(expected_values));

    parquet_schema_out(out_str).stdout(contains("OPTIONAL INT32 a (TIME(MILLIS,false));"));
}

#[test]
fn query_timestamp_with_timezone_postgres() {
    // Setup table for test
    let table_name = "QueryTimestampWithTimezone";
    let conn = env()
        .connect_with_connection_string(POSTGRES, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_pg(&conn, table_name, &["TIMESTAMPTZ"]).unwrap();
    let insert = format!(
        "INSERT INTO {table_name}
        (a)
        VALUES
        ('2022-09-07 16:04:12 +02:00');"
    );
    conn.execute(&insert, (), None).unwrap();
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");
    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            POSTGRES,
            &query,
        ])
        .assert()
        .success();

    let expected_values = "{a: 2022-09-07 14:04:12.000000 +00:00}\n";
    parquet_read_out(out_str).stdout(eq(expected_values));

    parquet_schema_out(out_str).stdout(contains("OPTIONAL INT64 a (TIMESTAMP(MICROS,false));"));
}

#[test]
fn query_timestamp_postgres() {
    // Setup table for test
    let table_name = "QueryTimestamp";
    let conn = env()
        .connect_with_connection_string(POSTGRES, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_pg(&conn, table_name, &["TIMESTAMP"]).unwrap();
    let insert = format!(
        "INSERT INTO {table_name}
        (a)
        VALUES
        ('2022-09-07 16:04:12');"
    );
    conn.execute(&insert, (), None).unwrap();
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");
    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            POSTGRES,
            &query,
        ])
        .assert()
        .success();

    let expected_values = "{a: 2022-09-07 16:04:12.000000 +00:00}\n";
    parquet_read_out(out_str).stdout(eq(expected_values));

    parquet_schema_out(out_str).stdout(contains("OPTIONAL INT64 a (TIMESTAMP(MICROS,false));"));
}

#[test]
fn query_all_the_types() {
    // Setup table for test
    let table_name = "AllTheTypes";
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(
        &conn,
        table_name,
        &[
            "CHAR(5) NOT NULL",
            "NUMERIC(3,2) NOT NULL",
            "DECIMAL(3,2) NOT NULL",
            "INTEGER NOT NULL",
            "SMALLINT NOT NULL",
            "FLOAT(3) NOT NULL",
            "REAL NOT NULL",
            "DOUBLE PRECISION NOT NULL",
            "VARCHAR(100) NOT NULL",
            "DATE NOT NULL",
            "TIME NOT NULL",
            "DATETIME NOT NULL",
        ],
    )
    .unwrap();
    let insert = format!(
        "INSERT INTO {table_name}
        (a,b,c,d,e,f,g,h,i,j,k,l)
        VALUES
        ('abcde', 1.23, 1.23, 42, 42, 1.23, 1.23, 1.23, 'Hello, World!', '2020-09-16', '03:54:12', '2020-09-16 03:54:12');"
    );
    conn.execute(&insert, (), None).unwrap();

    let expected_values = "{\
        a: \"abcde\", \
        b: 1.23, \
        c: 1.23, \
        d: 42, \
        e: 42, \
        f: 1.23, \
        g: 1.23, \
        h: 1.23, \
        i: \"Hello, World!\", \
        j: 2020-09-16, \
        k: 14052000000000, \
        l: 2020-09-16 03:54:12.000 +00:00\
    }\n";

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a,b,c,d,e,f,g,h,i,j,k,l FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    parquet_read_out(out_str).stdout(eq(expected_values));
}

#[test]
fn query_bits() {
    // Setup table for test
    let table_name = "QueryBits";
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["BIT"]).unwrap();
    let insert = format!(
        "INSERT INTO {table_name}
        (a)
        VALUES
        (0), (1), (NULL), (1), (0);"
    );
    conn.execute(&insert, (), None).unwrap();

    let expected_values = "\
        {a: false}\n\
        {a: true}\n\
        {a: null}\n\
        {a: true}\n\
        {a: false}\n\
    ";

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    parquet_read_out(out_str).stdout(eq(expected_values));
}

#[test]
fn query_doubles() {
    // Setup table for test
    let table_name = "QueryDoubles";
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DOUBLE PRECISION NOT NULL"]).unwrap();
    let insert = format!(
        "INSERT INTO {table_name}
        (a)
        VALUES
        (0.1), (2.3);"
    );
    conn.execute(&insert, (), None).unwrap();

    let expected_values = "\
        {a: 0.1}\n\
        {a: 2.3}\n\
    ";

    let expected_schema = "REQUIRED DOUBLE a;";

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    parquet_read_out(out_str).stdout(eq(expected_values));

    // Also verify schema to ensure f64 is chosen and not f32
    parquet_schema_out(out_str).stdout(contains(expected_schema));
}

/// Should not create a file if the query comes back empty `--no-empty-file` is set.
#[test]
fn query_comes_back_with_no_rows() {
    // Setup table for test
    let table_name = "QueryComesBackWithNoRows";
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DOUBLE PRECISION NOT NULL"]).unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            "--connection-string",
            MSSQL,
            "--no-empty-file",
            out_str,
            &query,
        ])
        .assert()
        .success();

    assert_eq!(
        ErrorKind::NotFound,
        File::open(out_str).err().unwrap().kind()
    );
}

/// Should not create a file if the query comes back empty `--no-empty-file` is set. This should
/// still work, even in the presence of (potential) file splitting.
///
/// This test has been created after a bug report. Original implementation of `--no-empty-file`
/// deleted the file after the fact. Yet it tried to delete the base path without taking the suffix
/// from file splitting into account.
#[test]
fn no_empty_file_works_with_split_files() {
    // Setup table for test
    let table_name = "NoEmptyFileWorksWithSplitFiles";
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DOUBLE PRECISION NOT NULL"]).unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            "--connection-string",
            MSSQL,
            "--row-groups-per-file",
            "1",
            "--no-empty-file",
            out_str,
            &query,
        ])
        .assert()
        .success();

    assert_eq!(
        ErrorKind::NotFound,
        File::open(out_str).err().unwrap().kind()
    );
}

/// If the query does not come back empty, of course a file should be created, even if the
/// `--no-empty-file` flag is set.
#[test]
fn emit_file_despite_no_empty_file_set() {
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = "SELECT 42 as a";

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            "--connection-string",
            MSSQL,
            "--no-empty-file",
            out_str,
            query,
        ])
        .assert()
        .success();

    let expected = "{a: 42}\n";
    parquet_read_out(out_str).stdout(eq(expected));
}

/// Should read query from standard input if "-" is provided as query text.
#[test]
fn read_query_from_stdin() {
    // Setup table for test
    let table_name = "ReadQueryFromStdin";
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["INT"]).unwrap();
    let insert = format!(
        "INSERT INTO {table_name}
        (a)
        VALUES
        (54),
        (54),
        (34),
        (12);"
    );
    conn.execute(&insert, (), None).unwrap();

    let expected_values = "\
        {a: 54}\n\
        {a: 54}\n\
        {a: 34}\n\
        {a: 12}\n\
    ";
    let query = format!("SELECT a FROM {table_name} ORDER BY id");
    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Tempfile path must be utf8");

    cargo_bin_cmd!()
        .args(["-vvvv", "query", out_str, "--connection-string", MSSQL, "-"])
        .write_stdin(query)
        .assert()
        .success();

    parquet_read_out(out_str).stdout(eq(expected_values));
}

/// Regression test for #869: every file should hold the configured number of
/// row groups, including the first.
#[test]
fn split_files_on_num_row_groups() {
    // Given
    let table_name = "SplitFilesOnNumRowGroups";
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["INTEGER"]).unwrap();
    let insert = format!("INSERT INTO {table_name} (A) VALUES(1),(2),(3),(4),(5)");
    conn.execute(&insert, (), None).unwrap();

    let out_dir = tempdir().unwrap();
    let out_path = out_dir.path().join("out.par");
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name}");

    // When
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            "--batch-size-row",
            "1",
            "--row-groups-per-file",
            "2",
            &query,
        ])
        .assert()
        .success();

    // Then
    let num_row_groups_in = |file_name| {
        let path = out_dir.path().join(file_name);
        let file = File::open(path).unwrap();
        SerializedFileReader::new(file)
            .unwrap()
            .metadata()
            .num_row_groups()
    };
    // The first file must hold the configured number of row groups (2).
    assert_eq!(2, num_row_groups_in("out_01.par"));
    // The same goes for the second file.
    assert_eq!(2, num_row_groups_in("out_02.par"));
    // For the third file, only one row group is left.
    assert_eq!(1, num_row_groups_in("out_03.par"));
}

/// Verify naming of the files is with successive numbers starting from 1 to 3 with split files and
/// `--no-empty-file` flag set. This was messed up, with a refactoring once and file names started
/// with `2` instead of `1``.
#[test]
fn file_name_no_empty_file_and_split_files() {
    // Setup table for test
    let table_name = "FileNameNoEmptyFileAndSplitFiles";
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["INTEGER"]).unwrap();
    let insert = format!("INSERT INTO {table_name} (A) VALUES(1),(2),(3)");
    conn.execute(&insert, (), None).unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name}");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            "--no-empty-file",
            "--batch-size-row",
            "1",
            "--row-groups-per-file",
            "1",
            &query,
        ])
        .assert()
        .success();

    // Expect one file per row in table (3)

    parquet_read_out(out_dir.path().join("out_01.par").to_str().unwrap());
    parquet_read_out(out_dir.path().join("out_02.par").to_str().unwrap());
    parquet_read_out(out_dir.path().join("out_03.par").to_str().unwrap());
}

#[test]
fn split_files_on_size_limit() {
    // Setup table for test
    let table_name = "SplitFilesOnSizeLimit";
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["INTEGER"]).unwrap();
    let insert = format!("INSERT INTO {table_name} (A) VALUES(1),(2),(3)");
    conn.execute(&insert, (), None).unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name}");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            "--batch-size-row",
            "1",
            "--file-size-threshold",
            "1B",
            &query,
        ])
        .assert()
        .success();

    // Expect one file per row in table (3)

    parquet_read_out(out_dir.path().join("out_01.par").to_str().unwrap());
    parquet_read_out(out_dir.path().join("out_02.par").to_str().unwrap());
    parquet_read_out(out_dir.path().join("out_03.par").to_str().unwrap());
}

#[test]
fn configurable_suffix_length() {
    // Setup table for test
    let table_name = "ConfigurableSuffixLength";
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["INTEGER"]).unwrap();
    let insert = format!("INSERT INTO {table_name} (A) VALUES(1)");
    conn.execute(&insert, (), None).unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name}");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            "--batch-size-row",
            "1",
            "--file-size-threshold",
            "1B",
            "--suffix-length",
            "4",
            &query,
        ])
        .assert()
        .success();

    // Expect one file per row in table (3)

    parquet_read_out(out_dir.path().join("out_0001.par").to_str().unwrap());
}

#[test]
fn varbinary_column() {
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();

    setup_empty_table_mssql(&conn, "VarbinaryColumn", &["VARBINARY(10)"]).unwrap();
    conn.execute(
        "INSERT INTO VarbinaryColumn (a) Values \
        (CONVERT(Binary(5), 'Hello')),\
        (CONVERT(Binary(5), 'World')),\
        (NULL)",
        (),
        None,
    )
    .unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = "SELECT a FROM VarbinaryColumn;";

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            query,
        ])
        .assert()
        .success();

    let expected = "{a: [72, 101, 108, 108, 111]}\n{a: [87, 111, 114, 108, 100]}\n{a: null}\n";

    parquet_read_out(out_str).stdout(eq(expected));
}

/// Since VARCHARMAX reports a size of 0. odbc2parquet should detect this and use the global column
/// length limit instead.
#[test]
fn query_varchar_max() {
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    let table_name = "QueryVarcharMax";

    setup_empty_table_mssql(&conn, table_name, &["VARCHAR(MAX)"]).unwrap();
    conn.execute(
        &format!("INSERT INTO {table_name} (a) Values ('Hello, World!');"),
        (),
        None,
    )
    .unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name};");

    // VARCHAR(max) has size 0. => Column is ignored and file would be empty and schemaless
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            "--connection-string",
            MSSQL,
            out_str,
            &query,
        ])
        .assert()
        .success();

    parquet_read_out(out_str).stdout(eq("{a: \"Hello, World!\"}\n"));
}

/// Introduced after discovering a bug, that columns were not ignored on windows.
///
/// Since VARCHARMAX reports a size of 0, it will be ignored, resulting in an output file with no
/// columns. Yet odbc2parquet should detect this and give the user an error instead.
#[test]
fn query_varchar_max_utf16() {
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    let table_name = "QueryVarcharMaxUtf16";

    setup_empty_table_mssql(&conn, table_name, &["VARCHAR(MAX)"]).unwrap();
    conn.execute(
        &format!("INSERT INTO {table_name} (a) Values ('Hello, World!');"),
        (),
        None,
    )
    .unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name};");

    // VARCHAR(max) has size 0. => Column is ignored and file would be empty and schemaless
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            "--encoding",
            "Utf16",
            "--connection-string",
            MSSQL,
            out_str,
            &query,
        ])
        .assert()
        .success();

    parquet_read_out(out_str).stdout(eq("{a: \"Hello, World!\"}\n"));
}

#[test]
fn binary_column() {
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();

    setup_empty_table_mssql(&conn, "BinaryColumn", &["BINARY(5)"]).unwrap();
    conn.execute(
        "INSERT INTO BinaryColumn (a) Values \
        (CONVERT(Binary(5), 'Hello')),\
        (CONVERT(Binary(5), 'World')),\
        (NULL)",
        (),
        None,
    )
    .unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = "SELECT a FROM BinaryColumn;";

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            query,
        ])
        .assert()
        .success();

    let expected = "{a: [72, 101, 108, 108, 111]}\n{a: [87, 111, 114, 108, 100]}\n{a: null}\n";
    parquet_read_out(out_str).stdout(eq(expected));
}

/// The prefer-varbinary flag must enforce mapping of binary columns to BYTE_ARRAY instead of
/// FIXED_LEN_BYTE_ARRAY.
#[test]
fn prefer_varbinary() {
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();

    let table_name = "PreferVarbinary";

    setup_empty_table_mssql(&conn, table_name, &["BINARY(5)"]).unwrap();
    conn.execute(
        &format!(
            "INSERT INTO {table_name} (a) Values \
        (CONVERT(Binary(5), 'Hello')),\
        (CONVERT(Binary(5), 'World')),\
        (NULL)"
        ),
        (),
        None,
    )
    .unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--prefer-varbinary",
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();

    let expected = "{a: [72, 101, 108, 108, 111]}\n{a: [87, 111, 114, 108, 100]}\n{a: null}\n";
    parquet_read_out(out_str).stdout(eq(expected));

    parquet_schema_out(out_str).stdout(contains("OPTIONAL BYTE_ARRAY a;"));
}

/// Strings with interior nuls should be written into parquet file as they are.
#[test]
fn interior_nul_in_varchar() {
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, "InteriorNul", &["VARCHAR(10)"]).unwrap();

    conn.execute(
        "INSERT INTO InteriorNul (a) VALUES (?);",
        &"a\0b".into_parameter(),
        None,
    )
    .unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = "SELECT a FROM InteriorNul;";

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            query,
        ])
        .assert()
        .success();

    let expected = "{a: \"a\0b\"}\n";

    parquet_read_out(out_str).stdout(eq(expected));
}

/// Fixed size NCHAR column on database is not truncated then value is passed into a narrow buffer.
#[test]
#[cfg(not(target_os = "windows"))] // Windows does not use UTF-8 as default system encoding
fn nchar_not_truncated() {
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    let table_name = "NCharNotTruncated";
    setup_empty_table_mssql(&conn, table_name, &["NCHAR(1)"]).unwrap();

    conn.execute(
        &format!("INSERT INTO {} (a) VALUES (?);", table_name),
        &"Ü".into_parameter(),
        None,
    )
    .unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = &format!("SELECT a FROM {};", table_name);

    cargo_bin_cmd!()
        .args(&[
            "-vvvv",
            "query",
            "--encoding",
            "system",
            "--connection-string",
            MSSQL,
            out_str,
            query,
        ])
        .assert()
        .success();

    let expected = "{a: \"Ü\"}\n";

    parquet_read_out(out_str).stdout(eq(expected));
}

/// Test non ASCII character with system encoding
#[test]
#[cfg(not(target_os = "windows"))] // Windows does not use UTF-8 as default system encoding
fn system_encoding() {
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    let table_name = "SystemEncoding";
    setup_empty_table_mssql(&conn, table_name, &["VARCHAR(10)"]).unwrap();

    conn.execute(
        &format!("INSERT INTO {} (a) VALUES (?);", table_name),
        &"Ü".into_parameter(),
        None,
    )
    .unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = &format!("SELECT a FROM {};", table_name);

    cargo_bin_cmd!()
        .args(&[
            "-vvvv",
            "query",
            "--encoding",
            "system",
            "--connection-string",
            MSSQL,
            out_str,
            query,
        ])
        .assert()
        .success();

    let expected = "{a: \"Ü\"}\n";

    parquet_read_out(out_str).stdout(eq(expected));
}

/// Test non ASCII character with utf16 encoding
#[test]
fn utf_16_encoding() {
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    let table_name = "Utf16Encoding";
    setup_empty_table_mssql(&conn, table_name, &["VARCHAR(10)"]).unwrap();

    conn.execute(
        &format!("INSERT INTO {table_name} (a) VALUES (?);"),
        &"Ü".into_parameter(),
        None,
    )
    .unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = &format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            "--encoding",
            "utf16",
            "--connection-string",
            MSSQL,
            out_str,
            query,
        ])
        .assert()
        .success();

    let expected = "{a: \"Ü\"}\n";

    parquet_read_out(out_str).stdout(eq(expected));
}

/// See: <https://github.com/pacman82/odbc2parquet/issues/862>
///
/// Then querying under windows (implying UTF-16 encoding) a column with value
/// "Colt Telecom España S.A." was followed by a column if max_length with 18 bytes in UTF-8
/// encoding. This let to a panic due to a violation of the String invariant, because that would
/// split the reused buffer in the middle of the UTF-8 character "ñ" (which is 2 bytes in
/// UTF-8).
#[test]
fn issue_862() {
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    let table_name = "Issue862";
    setup_empty_table_mssql(&conn, table_name, &["VARCHAR(50)"]).unwrap();

    conn.execute(
        &format!("INSERT INTO {table_name} (a) VALUES (?),(?);"),
        (
            &"Colt Telecom España S.A.".into_parameter(),
            // 6 characters will may take up to 18 bytes in UTF-8, these don't but it will still
            // reproduce the issue.
            &"123456".into_parameter(),
        ),
        None,
    )
    .unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = &format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            "--encoding",
            "utf16",
            "--connection-string",
            MSSQL,
            out_str,
            query,
        ])
        .assert()
        .success();

    let expected = "{a: \"Colt Telecom España S.A.\"}\n{a: \"123456\"}\n";

    parquet_read_out(out_str).stdout(eq(expected));
}

/// Test non ASCII character with automatic codec detection
#[test]
fn auto_encoding() {
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    let table_name = "AutoEncoding";
    setup_empty_table_mssql(&conn, table_name, &["VARCHAR(1)"]).unwrap();

    conn.execute(
        &format!("INSERT INTO {table_name} (a) VALUES (?);"),
        &"Ü".into_parameter(),
        None,
    )
    .unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = &format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            "--encoding",
            "auto",
            "--connection-string",
            MSSQL,
            out_str,
            query,
        ])
        .assert()
        .success();

    let expected = "{a: \"Ü\"}\n";

    parquet_read_out(out_str).stdout(eq(expected));
}

#[test]
pub fn insert_32_bit_integer() {
    let table_name = "Insert32BitInteger";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["INTEGER"]).unwrap();
    // Prepare file
    let message_type = "
        message schema {
            REQUIRED INT32 a;
        }
    ";
    let input = TmpParquetFile::with_1_dim(message_type, &[Some(42i32), Some(5), Some(1)]);
    let input_path = input.path_as_str();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path,
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);
    assert_eq!("42\n5\n1", actual);
}

#[test]
pub fn insert_optional_32_bit_integer() {
    let table_name = "InsertOptional32BitInteger";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["INTEGER"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            OPTIONAL INT32 a;
        }
    ";
    let input = TmpParquetFile::with_1_dim(message_type, &[Some(42i32), None, Some(1)]);
    let input_path = input.path_as_str();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path,
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("42\nNULL\n1", actual);
}

#[test]
pub fn insert_64_bit_integer() {
    let table_name = "Insert64BitInteger";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["INTEGER"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            REQUIRED INT64 a;
        }
    ";
    let input = TmpParquetFile::with_1_dim(message_type, &[Some(-42i64), Some(5), Some(1)]);
    let input_path = input.path_as_str();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path,
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("-42\n5\n1", actual);
}

/// See issue: <https://github.com/pacman82/odbc2parquet/issues/665>
///
/// Underlying issue for the bug has been the assumption the buffer used to write into parquet is
/// always the same length as the source buffer, which happens usually in the last row group which
/// may have less rows, because the total number of rows is not a divisor of max row group size.
#[test]
pub fn insert_file_with_last_row_group_of_less_size() {
    let table_name = "InsertFileWithLastRowGroupOfLessSize";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["INTEGER"]).unwrap();

    // Prepare file

    // A temporary directory, to be removed at the end of the test.
    let tmp_dir = tempdir().unwrap();
    // The name of the input parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let input_path = tmp_dir.path().join("input.par");

    let message_type = "
        message schema {
            REQUIRED INT64 a;
        }
    ";

    let schema = Arc::new(parse_message_type(message_type).unwrap());
    let props = Arc::new(WriterProperties::builder().build());
    let file = File::create(&input_path).unwrap();
    let mut writer = SerializedFileWriter::new(file, schema, props).unwrap();
    let mut row_group_writer = writer.next_row_group().unwrap();
    let mut col_writer = row_group_writer.next_column().unwrap().unwrap();
    i64::write_batch(col_writer.untyped(), &[1, 2], None);
    col_writer.close().unwrap();
    row_group_writer.close().unwrap();
    let mut row_group_writer = writer.next_row_group().unwrap();
    let mut col_writer = row_group_writer.next_column().unwrap().unwrap();
    i64::write_batch(col_writer.untyped(), &[1], None);
    col_writer.close().unwrap();
    row_group_writer.close().unwrap();
    writer.close().unwrap();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("1\n2\n1", actual);
}

#[test]
pub fn insert_optional_64_bit_integer() {
    let table_name = "InsertOptional64BitInteger";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["BIGINT"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            OPTIONAL INT64 a;
        }
    ";
    let input = TmpParquetFile::with_1_dim(message_type, &[Some(-42i64), None, Some(1)]);
    let input_path = input.path_as_str();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path,
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("-42\nNULL\n1", actual);
}

#[test]
pub fn insert_utf8() {
    let table_name = "InsertUtf8";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["VARCHAR(50)"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            REQUIRED BYTE_ARRAY a (UTF8);
        }
    ";
    let text: ByteArray = "Hello, World!".into();
    let input = TmpParquetFile::with_1_dim(
        message_type,
        &[
            Some(text),
            Some("Hallo, Welt!".into()),
            Some("Bonjour, Monde!".into()),
        ],
    );
    let input_path = input.path_as_str();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path,
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("Hello, World!\nHallo, Welt!\nBonjour, Monde!", actual);
}

#[test]
pub fn insert_optional_utf8() {
    let table_name = "InsertOptionalUtf8";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["VARCHAR(50)"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            OPTIONAL BYTE_ARRAY a (UTF8);
        }
    ";
    let text: ByteArray = "Hello, World!".into();
    let input = TmpParquetFile::with_1_dim(
        message_type,
        &[Some(text), None, Some("Hallo, Welt!".into())],
    );
    let input_path = input.path_as_str();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path,
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("Hello, World!\nNULL\nHallo, Welt!", actual);
}

#[test]
pub fn insert_utf16() {
    let table_name = "InsertUtf16";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["VARCHAR(50)"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            REQUIRED BYTE_ARRAY a (UTF8);
        }
    ";
    let text: ByteArray = "Hello, World!".into();
    let input = TmpParquetFile::with_1_dim(
        message_type,
        &[
            Some(text),
            Some("Hallo, Welt!".into()),
            Some("Bonjour, Monde!".into()),
        ],
    );
    let input_path = input.path_as_str();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            "--encoding",
            "Utf16",
            input_path,
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("Hello, World!\nHallo, Welt!\nBonjour, Monde!", actual);
}

#[test]
pub fn insert_optional_utf16() {
    let table_name = "InsertOptionalUtf16";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["VARCHAR(50)"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            OPTIONAL BYTE_ARRAY a (UTF8);
        }
    ";
    let text: ByteArray = "Hello, World!".into();
    let input = TmpParquetFile::with_1_dim(
        message_type,
        &[Some(text), None, Some("Hallo, Welt!".into())],
    );
    let input_path = input.path_as_str();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            "--encoding",
            "Utf16",
            input_path,
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("Hello, World!\nNULL\nHallo, Welt!", actual);
}

#[test]
pub fn insert_bool() {
    let table_name = "InsertBool";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["BIT"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            REQUIRED BOOLEAN a;
        }
    ";
    let input = TmpParquetFile::with_1_dim(message_type, &[Some(true), Some(false), Some(false)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("1\n0\n0", actual);
}

#[test]
pub fn insert_optional_bool() {
    let table_name = "InsertOptionalBool";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["BIT"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            OPTIONAL BOOLEAN a;
        }
    ";
    let input = TmpParquetFile::with_1_dim(message_type, &[Some(true), None, Some(false)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("1\nNULL\n0", actual);
}

#[test]
pub fn insert_f32() {
    let table_name = "InsertF32";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["FLOAT"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            REQUIRED FLOAT a;
        }
    ";
    let input = TmpParquetFile::with_1_dim(message_type, &[Some(1.2f32), Some(3.4), Some(5.6)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(
        "1.2000000476837158\n3.4000000953674316\n5.5999999046325684",
        actual
    );
}

#[test]
pub fn insert_optional_f32() {
    let table_name = "InsertOptionalF32";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["FLOAT"]).unwrap();
    let message_type = "
        message schema {
            OPTIONAL FLOAT a;
        }
    ";
    let input = TmpParquetFile::with_1_dim(message_type, &[Some(1.2f32), None, Some(3.4)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("1.2000000476837158\nNULL\n3.4000000953674316", actual);
}

#[test]
pub fn insert_f64() {
    let table_name = "InsertF64";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["FLOAT(53)"]).unwrap();
    let message_type = "
        message schema {
            REQUIRED DOUBLE a;
        }
    ";

    let input = TmpParquetFile::with_1_dim(message_type, &[Some(1.2f64), Some(3.4), Some(5.6)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("1.2\n3.3999999999999999\n5.5999999999999996", actual);
}

#[test]
pub fn insert_optional_f64() {
    let table_name = "InsertOptionalF64";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["FLOAT(53)"]).unwrap();
    let message_type = "
        message schema {
            OPTIONAL DOUBLE a;
        }
    ";

    let input = TmpParquetFile::with_1_dim(message_type, &[Some(1.2f64), None, Some(3.4)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("1.2\nNULL\n3.3999999999999999", actual);
}

#[test]
pub fn insert_date() {
    let table_name = "InsertDate";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DATE"]).unwrap();
    let message_type = "
        message schema {
            REQUIRED INT32 a (DATE);
        }
    ";

    let input = TmpParquetFile::with_1_dim(message_type, &[Some(0i32), Some(365), Some(18695)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("1970-01-01\n1971-01-01\n2021-03-09", actual);
}

#[test]
pub fn insert_optional_date() {
    let table_name = "InsertOptionalDate";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["Date"]).unwrap();

    let message_type = "
        message schema {
            OPTIONAL INT32 a (DATE);
        }
    ";

    let input = TmpParquetFile::with_1_dim(message_type, &[Some(0i32), None, Some(18695)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("1970-01-01\nNULL\n2021-03-09", actual);
}

#[test]
pub fn insert_time_ms() {
    let table_name = "InsertTimeMs";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["TIME(3)"]).unwrap();
    let message_type = "
        message schema {
            REQUIRED INT32 a (TIME_MILLIS);
        }
    ";
    // Total number of milli seconds since midnight
    let input = TmpParquetFile::with_1_dim(
        message_type,
        &[Some(0i32), Some(3_600_000), Some(82_800_000)],
    );
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("00:00:00.000\n01:00:00.000\n23:00:00.000", actual);
}

#[test]
pub fn insert_optional_time_ms() {
    let table_name = "InsertOptionalTimeMs";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["TIME(3)"]).unwrap();

    let message_type = "
        message schema {
            OPTIONAL INT32 a (TIME_MILLIS);
        }
    ";

    // Total number of milli seconds since midnight
    let input = TmpParquetFile::with_1_dim(message_type, &[Some(0i32), None, Some(82_800_000)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("00:00:00.000\nNULL\n23:00:00.000", actual);
}

#[test]
pub fn insert_time_us() {
    let table_name = "InsertTimeUs";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["TIME(6)"]).unwrap();
    let message_type = "
        message schema {
            REQUIRED INT64 a (TIME_MICROS);
        }
    ";

    // Total number of microseconds since midnight
    let input = TmpParquetFile::with_1_dim(
        message_type,
        &[Some(0i64), Some(3_600_000_000), Some(82_800_000_000)],
    );
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("00:00:00.000000\n01:00:00.000000\n23:00:00.000000", actual);
}

#[test]
pub fn insert_optional_time_us() {
    let table_name = "InsertOptionalTimeUs";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["TIME(6)"]).unwrap();
    let message_type = "
        message schema {
            OPTIONAL INT64 a (TIME_MICROS);
        }
    ";

    // Total number of microseconds since midnight
    let input = TmpParquetFile::with_1_dim(message_type, &[Some(0i64), None, Some(82_800_000_000)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("00:00:00.000000\nNULL\n23:00:00.000000", actual);
}

#[test]
pub fn insert_decimal_from_i32() {
    let table_name = "InsertDecimalFromI32";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DECIMAL(9,2)"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            REQUIRED INT32 a (DECIMAL(9,2));
        }
    ";
    let input = TmpParquetFile::with_1_dim(message_type, &[Some(0i32), Some(123456789), Some(-42)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(".00\n1234567.89\n-.42", actual);
}

#[test]
pub fn insert_decimal_from_i32_optional() {
    let table_name = "InsertDecimalFromI32Optional";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DECIMAL(9,2)"]).unwrap();
    let message_type = "
        message schema {
            OPTIONAL INT32 a (DECIMAL(9,2));
        }
    ";
    let input = TmpParquetFile::with_1_dim(message_type, &[Some(0i32), None, Some(-42)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(".00\nNULL\n-.42", actual);
}

#[test]
pub fn insert_decimal_from_i64() {
    let table_name = "InsertDecimalFromI64";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DECIMAL(9,2)"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            REQUIRED INT64 a (DECIMAL(9,2));
        }
    ";

    let input = TmpParquetFile::with_1_dim(message_type, &[Some(0i64), Some(123456789), Some(-42)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(".00\n1234567.89\n-.42", actual);
}

#[test]
pub fn insert_decimal_from_i64_optional() {
    let table_name = "InsertDecimalFromI64Optional";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DECIMAL(9,2)"]).unwrap();

    // Prepare file
    let message_type = "
            message schema {
                OPTIONAL INT64 a (DECIMAL(9,2));
            }
        ";

    let input = TmpParquetFile::with_1_dim(message_type, &[Some(0i64), None, Some(-42)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(".00\nNULL\n-.42", actual);
}

#[test]
pub fn insert_timestamp_ms() {
    let table_name = "InsertTimestampMs";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DATETIME2"]).unwrap();
    let message_type = "
        message schema {
            REQUIRED INT64 a (TIMESTAMP_MILLIS);
        }
    ";
    // Total number of milli seconds since unix epoch
    let input =
        TmpParquetFile::with_1_dim(message_type, &[Some(0i64), Some(1), Some(1616367053000)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(
        "1970-01-01 00:00:00.0000000\n1970-01-01 00:00:00.0010000\n2021-03-21 22:50:53.0000000",
        actual
    );
}

#[test]
pub fn insert_timestamp_ms_optional() {
    let table_name = "InsertTimestampMsOptional";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DATETIME2"]).unwrap();

    // Prepare file

    let message_type = "
        message schema {
            OPTIONAL INT64 a (TIMESTAMP_MILLIS);
        }
    ";

    // Total number of milli seconds since unix epoch
    let input = TmpParquetFile::with_1_dim(message_type, &[Some(0i64), None, Some(1616367053000)]);
    let input_path = input.path.clone();

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input_path.to_str().unwrap(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(
        "1970-01-01 00:00:00.0000000\nNULL\n2021-03-21 22:50:53.0000000",
        actual
    );
}

#[test]
pub fn insert_timestamp_us() {
    let table_name = "InsertTimestampUs";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DATETIME2"]).unwrap();
    // Prepare file
    let message_type = "
        message schema {
            REQUIRED INT64 a (TIMESTAMP_MICROS);
        }
    ";
    // Total number of microseconds since unix epoch
    let input =
        TmpParquetFile::with_1_dim(message_type, &[Some(0i64), Some(1), Some(1616367053000000)]);

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input.path_as_str(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(
        "1970-01-01 00:00:00.0000000\n1970-01-01 00:00:00.0000010\n2021-03-21 22:50:53.0000000",
        actual
    );
}

#[test]
pub fn insert_timestamp_us_optional() {
    let table_name = "InsertTimestampUsOptional";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DATETIME2"]).unwrap();
    let message_type = "
        message schema {
            OPTIONAL INT64 a (TIMESTAMP_MICROS);
        }
    ";
    // Total number of microseconds since unix epoch
    let input =
        TmpParquetFile::with_1_dim(message_type, &[Some(0i64), None, Some(1616367053000000)]);

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input.path_as_str(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(
        "1970-01-01 00:00:00.0000000\nNULL\n2021-03-21 22:50:53.0000000",
        actual
    );
}

#[test]
pub fn insert_binary() {
    let table_name = "InsertBinary";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["VARBINARY(50)"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            REQUIRED BYTE_ARRAY a;
        }
    ";

    let text: ByteArray = "Hello, World!".into();
    let input = TmpParquetFile::with_1_dim(
        message_type,
        &[
            Some(text),
            Some("Hallo, Welt!".into()),
            Some("Bonjour, Monde!".into()),
        ],
    );

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input.path_as_str(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(
        "48656C6C6F2C20576F726C6421\n48616C6C6F2C2057656C7421\n426F6E6A6F75722C204D6F6E646521",
        actual
    );
}

#[test]
pub fn insert_binary_optional() {
    let table_name = "InsertBinaryOptional";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["VARBINARY(50)"]).unwrap();

    // Prepare file

    let message_type = "
        message schema {
            OPTIONAL BYTE_ARRAY a;
        }
    ";

    let text: ByteArray = "Hello, World!".into();
    let input = TmpParquetFile::with_1_dim(
        message_type,
        &[Some(text), None, Some("Hallo, Welt!".into())],
    );

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input.path_as_str(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(
        "48656C6C6F2C20576F726C6421\nNULL\n48616C6C6F2C2057656C7421",
        actual
    );
}

#[test]
pub fn insert_fixed_len_binary() {
    let table_name = "InsertFixedLenBinary";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["BINARY(13)"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            REQUIRED FIXED_LEN_BYTE_ARRAY(13) a;
        }
    ";

    let to_fixed_len_byte_array = |input: &str| -> FixedLenByteArray {
        let ba: ByteArray = input.into();
        ba.into()
    };

    let text: FixedLenByteArray = to_fixed_len_byte_array("Hello, World!");
    let input = TmpParquetFile::with_1_dim(
        message_type,
        &[
            Some(text),
            Some(to_fixed_len_byte_array("Hallo, Welt!!")),
            Some(to_fixed_len_byte_array("0123456789012")),
        ],
    );

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input.path_as_str(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(
        "48656C6C6F2C20576F726C6421\n48616C6C6F2C2057656C742121\n30313233343536373839303132",
        actual
    );
}

#[test]
pub fn insert_fixed_len_binary_optional() {
    let table_name = "InsertFixedLenBinaryOptional";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["BINARY(13)"]).unwrap();

    // Prepare file

    let message_type = "
        message schema {
            OPTIONAL FIXED_LEN_BYTE_ARRAY(13) a;
        }
    ";

    let to_fixed_len_byte_array = |input: &str| -> FixedLenByteArray {
        let ba: ByteArray = input.into();
        ba.into()
    };

    let text: FixedLenByteArray = to_fixed_len_byte_array("Hello, World!");
    let input = TmpParquetFile::with_1_dim(
        message_type,
        &[
            Some(text),
            None,
            Some(to_fixed_len_byte_array("0123456789012")),
        ],
    );

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input.path_as_str(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(
        "48656C6C6F2C20576F726C6421\nNULL\n30313233343536373839303132",
        actual
    );
}

#[test]
pub fn insert_decimal_from_binary() {
    let table_name = "InsertDecimalFromBinary";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DECIMAL(9,2)"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            REQUIRED BYTE_ARRAY a (DECIMAL(9,2));
        }
    ";
    let zero: ByteArray = vec![0u8].into();
    let input = TmpParquetFile::with_1_dim(
        message_type,
        &[
            Some(zero),
            Some(vec![1].into()),
            Some(vec![1, 0, 0].into()),
            Some(vec![255, 255, 255].into()),
            Some(vec![255].into()),
        ],
    );

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input.path_as_str(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(".00\n.01\n655.36\n-.01\n-.01", actual);
}

#[test]
pub fn insert_decimal_from_binary_optional() {
    let table_name = "InsertDecimalFromBinaryOptional";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DECIMAL(9,2)"]).unwrap();
    // Prepare file
    let message_type = "
        message schema {
            OPTIONAL BYTE_ARRAY a (DECIMAL(9,2));
        }
    ";
    let zero: ByteArray = vec![0u8].into();
    let input = TmpParquetFile::with_1_dim(message_type, &[Some(zero), None, Some(vec![1].into())]);

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input.path_as_str(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(".00\nNULL\n.01", actual);
}

#[test]
pub fn insert_decimal_from_fixed_binary() {
    let table_name = "InsertDecimalFromFixedBinary";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DECIMAL(5,2)"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            REQUIRED FIXED_LEN_BYTE_ARRAY(3) a (DECIMAL(5,2));
        }
    ";
    let to_fixed_len_byte_array = |input: Vec<u8>| -> FixedLenByteArray {
        let ba: ByteArray = input.into();
        ba.into()
    };
    let input = TmpParquetFile::with_1_dim(
        message_type,
        &[
            Some(to_fixed_len_byte_array(vec![0, 0, 0])),
            Some(to_fixed_len_byte_array(vec![0, 0, 1])),
            Some(to_fixed_len_byte_array(vec![1, 0, 0])),
            Some(to_fixed_len_byte_array(vec![255, 255, 255])),
        ],
    );

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input.path_as_str(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(".00\n.01\n655.36\n-.01", actual);
}

#[test]
pub fn insert_decimal_from_fixed_binary_optional() {
    let table_name = "InsertDecimalFromFixedBinaryOptional";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["DECIMAL(5,2)"]).unwrap();
    let message_type = "
        message schema {
            OPTIONAL FIXED_LEN_BYTE_ARRAY(3) a (DECIMAL(5,2));
        }
    ";
    let to_fixed_len_byte_array = |input: Vec<u8>| -> FixedLenByteArray {
        let ba: ByteArray = input.into();
        ba.into()
    };
    let input = TmpParquetFile::with_1_dim(
        message_type,
        &[
            Some(to_fixed_len_byte_array(vec![0, 0, 1])),
            None,
            Some(to_fixed_len_byte_array(vec![255, 255, 255])),
        ],
    );

    // Insert file into table
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            input.path_as_str(),
            table_name,
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!(".01\nNULL\n-.01", actual);
}

/// Verify that we can use the `exec` command to insert values into a table
#[test]
fn basic_use_exec_for_insert() {
    let table_name = "InsertUsingExec";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["INTEGER", "INTEGER"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            OPTIONAL INT32 a;
        }
    ";
    let input = TmpParquetFile::with_1_dim(message_type, &[Some(1i32), None, Some(2)]);

    // When insert values into table using exec
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "exec",
            "--connection-string",
            MSSQL,
            input.path_as_str(),
            &format!("INSERT INTO {table_name} (a) VALUES (?a?)"),
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("1\nNULL\n2", actual);
}

/// Verify that named arguments work in exec command, even if they appear in a different order from
/// the columns in the parquet file
#[test]
fn exec_with_switched_order() {
    let table_name = "ExecWithSwitchedOrder";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["INTEGER", "INTEGER"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            OPTIONAL INT32 a;
            OPTIONAL INT32 b;
        }
    ";
    let input = TmpParquetFile::with_2_dim(message_type, &[Some(1i32)], &[Some(2i32)]);

    // When insert values into table using exec
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "exec",
            "--connection-string",
            MSSQL,
            input.path_as_str(),
            &format!("INSERT INTO {table_name} (b, a) VALUES (?b?, ?a?)"),
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a, b FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("1,2", actual);
}

/// Verify that named arguments work even if they appear in multiple positions
#[test]
fn exec_with_multiple_placeholder_mentions() {
    let table_name = "ExecWithMultiplePlaceholderMentions";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["INTEGER", "VARCHAR(50)", "INTEGER"]).unwrap();

    // Prepare file
    let message_type = "
        message schema {
            OPTIONAL INT32 a;
            OPTIONAL BYTE_ARRAY b (UTF8);
        }
    ";
    let text: ByteArray = "Hello".into();
    let input = TmpParquetFile::with_2_dim(message_type, &[Some(1i32)], &[Some(text)]);

    // When insert values into table using exec
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "exec",
            "--connection-string",
            MSSQL,
            input.path_as_str(),
            &format!("INSERT INTO {table_name} (a, b, c) VALUES (?a?, ?b?, ?a?)"),
        ])
        .assert()
        .success();

    // Query table and check for expected result
    let query = format!("SELECT a, b, c FROM {table_name} ORDER BY Id");
    let cursor = conn.execute(&query, (), None).unwrap().unwrap();
    let actual = cursor_to_string(cursor);

    assert_eq!("1,Hello,1", actual);
}

/// Write query output to stdout
#[test]
pub fn write_query_result_to_stdout() {
    // Given
    let table_name = "WriteQueryResultToStdout";
    // Prepare table
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["INTEGER"]).unwrap();
    conn.execute(
        &format!("INSERT INTO {table_name} (a) VALUES (?)"),
        [42i32, 5, 64].as_slice(),
        None,
    )
    .unwrap();

    // When

    // Query table and write contents to stdout
    let query = format!("SELECT a FROM {table_name} ORDER BY id");
    let command = cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            "--connection-string",
            MSSQL,
            "-", // Use `-` to explicitly write to stdout
            &query,
        ])
        .assert()
        .success();

    // Then
    let output = &command.get_output().stdout;
    assert!(!output.is_empty());

    // Write captured contents of stdout to temporary file so we can inspect it with read parquet
    let mut output_file = NamedTempFile::new().unwrap();
    output_file.write_all(output).unwrap();

    let expected = "{a: 42}\n";

    let output_path = output_file.path().to_str().unwrap();
    parquet_read_out(output_path).stdout(eq(expected));
}

/// Write query output to stdout
#[test]
pub fn reject_writing_to_stdout_and_file_size_limit() {
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            "--connection-string",
            "FakeConnectionString",
            "--file-size-threshold",
            "1GiB",
            "-", // Use `-` to explicitly write to stdout
            "SELECT a FROM FakeTableName ORDER BY id",
        ])
        .assert()
        .failure()
        .stderr(contains(
            "file-size-threshold conflicts with specifying stdout ('-') as output.",
        ));
}

#[test]
fn write_statistics_for_text_columns() {
    // Setup table for test
    let table_name = "WriteStatisticsForTextColumns";
    let mut table = TableMssql::new(table_name, &["VARCHAR(10)"]);
    table.insert_rows_as_text(&[["aaa"], ["zzz"]]);
    let query = format!("SELECT a FROM {table_name}");

    let command = cargo_bin_cmd!()
        .args([
            "query",
            "--connection-string",
            MSSQL,
            "-", // Use `-` to explicitly write to stdout
            &query,
        ])
        .assert()
        .success();

    // Then
    let bytes = Bytes::from(command.get_output().stdout.clone());
    let reader = SerializedFileReader::new(bytes).unwrap();
    let stats = reader
        .metadata()
        .row_group(0)
        .column(0)
        .statistics()
        .unwrap();
    assert_eq!(
        "aaa",
        str::from_utf8(stats.min_bytes_opt().unwrap()).unwrap()
    );
    assert_eq!(
        "zzz",
        str::from_utf8(stats.max_bytes_opt().unwrap()).unwrap()
    );
}

/// This did not work in earlier versions there we set the batch write size of the parquet writer to
/// the ODBC batch size.
#[test]
#[ignore = "Takes too long to run"]
fn query_4097_bits() {
    let num_bits = 4097;

    // Setup table for test
    let table_name = "Query4097Bits";
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    setup_empty_table_mssql(&conn, table_name, &["BIT"]).unwrap();

    // Insert 4097 bits "false" (default constructed) into the table
    let insert = format!(
        "INSERT INTO {table_name}
        (a)
        VALUES
        (?);"
    );
    let desc = BindParamDesc::bit(false);
    let mut parameter_buffer = conn
        .prepare(&insert)
        .unwrap()
        .into_column_inserter(num_bits, [desc])
        .unwrap();
    parameter_buffer.set_num_rows(num_bits);
    parameter_buffer.execute().unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let query = format!("SELECT a FROM {table_name};");

    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            out_str,
            "--connection-string",
            MSSQL,
            &query,
        ])
        .assert()
        .success();
}

/// A parquet file which will remove itself after it leaves scope. Intended for provinding test
/// fixtures for tests which require a parquet file as input.
struct TmpParquetFile {
    // A temporary directory, to be removed at the end of the test.
    _tmp_dir: TempDir,
    path: PathBuf,
}

impl TmpParquetFile {
    /// Creates a new temporary parquet file.
    pub fn with_1_dim<T>(message_type: &str, values: &[Option<T>]) -> Self
    where
        T: WriteToCw + Clone,
    {
        let values = ColumnDataImpl::new(values);
        Self::new(message_type, &[&values])
    }

    pub fn with_2_dim<T, U>(
        message_type: &str,
        values_a: &[Option<T>],
        values_b: &[Option<U>],
    ) -> Self
    where
        T: WriteToCw + Clone,
        U: WriteToCw + Clone,
    {
        let column_a = ColumnDataImpl::new(values_a);
        let column_b = ColumnDataImpl::new(values_b);
        Self::new(message_type, &[&column_a, &column_b])
    }

    pub fn new(message_type: &str, values: &[&dyn ColumnData]) -> Self {
        let tmp_dir = tempdir().expect("Must be able to create temporary directory");
        let path = tmp_dir.path().join("input.par");
        write_values_to_file(message_type, &path, values);
        TmpParquetFile {
            _tmp_dir: tmp_dir,
            path,
        }
    }

    /// Returns the path to the parquet file.
    pub fn path_as_str(&self) -> &str {
        self.path
            .to_str()
            .expect("Temporary file path must be utf8")
    }
}

/// Writes a parquet file with one row group and one column.
fn write_values_to_file(message_type: &str, input_path: &Path, columns: &[&dyn ColumnData]) {
    let schema = Arc::new(parse_message_type(message_type).unwrap());
    let props = Arc::new(WriterProperties::builder().build());
    let file = File::create(input_path).unwrap();
    let mut writer = SerializedFileWriter::new(file, schema, props).unwrap();
    let mut row_group_writer = writer.next_row_group().unwrap();

    for col in columns {
        let mut col_writer = row_group_writer.next_column().unwrap().unwrap();
        col.write(&mut col_writer.untyped());
        col_writer.close().unwrap();
    }

    row_group_writer.close().unwrap();
    writer.close().unwrap();
}

const COLUMN_NAMES: &[&str] = &["a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l"];

pub trait ColumnData {
    fn write(&self, column_writer: &mut ColumnWriter);
}

struct ColumnDataImpl<T> {
    values: Vec<T>,
    def_levels: Vec<i16>,
}

impl<T> ColumnDataImpl<T> {
    /// Creates a new instance of `ColumnDataImpl`.
    pub fn new(input: &[Option<T>]) -> Self
    where
        T: Clone,
    {
        let def_levels = input
            .iter()
            .map(|opt| if opt.is_some() { 1i16 } else { 0 })
            .collect::<Vec<_>>();
        let values = input
            .into_iter()
            .cloned()
            .filter_map(|opt| opt)
            .collect::<Vec<_>>();

        ColumnDataImpl { values, def_levels }
    }
}

impl<T> ColumnData for ColumnDataImpl<T>
where
    T: WriteToCw,
{
    fn write(&self, column_writer: &mut ColumnWriter) {
        T::write_batch(column_writer, &self.values, Some(&self.def_levels));
    }
}

/// Sets up a table in the mssql database and allows us to fill it with data. Column names are given
/// automatically a,b,c, etc.
pub struct TableMssql<'a, const NUM_COLUMNS: usize> {
    pub name: &'a str,
    pub conn: Connection<'a>,
}

impl<'a, const NUM_COLUMNS: usize> TableMssql<'a, NUM_COLUMNS> {
    pub fn new(name: &'a str, column_types: &'a [&'a str; NUM_COLUMNS]) -> Self {
        let conn = env()
            .connect_with_connection_string(
                MSSQL,
                ConnectionOptions {
                    login_timeout_sec: Some(5),
                    ..Default::default()
                },
            )
            .expect("Must be able to connect to MSSQL database.");
        setup_empty_table_mssql(&conn, name, column_types)
            .expect("Must be able to setup empty table.");
        TableMssql { name, conn }
    }

    pub fn insert_rows_as_text<'b, C>(&mut self, content: &[[C; NUM_COLUMNS]])
    where
        C: Into<Option<&'b str>> + Copy,
    {
        let statement = self.insert_statement(NUM_COLUMNS);
        // Insert everything in one go => capacity == length of array
        let capacity = content.len();
        let max_str_len = (0..NUM_COLUMNS)
            .map(|col_index| {
                (0..content.len())
                    .map(|row_index| {
                        let opt: Option<&str> = content[row_index][col_index].into();
                        opt.map(|s| s.len()).unwrap_or(0)
                    })
                    .max()
                    .unwrap_or(0)
            })
            .collect::<Vec<_>>();
        let mut inserter = self
            .conn
            .prepare(&statement)
            .unwrap()
            .into_text_inserter(capacity, max_str_len)
            .unwrap();

        for (row_index, row) in content.iter().enumerate() {
            for (column_index, element) in row.iter().enumerate() {
                let element: Option<&str> = (*element).into();
                let element = element.map(|s| s.as_bytes());
                inserter
                    .column_mut(column_index)
                    .set_cell(row_index, element);
            }
        }

        inserter.set_num_rows(content.len());
        inserter.execute().unwrap();
    }

    fn insert_statement(&self, number_of_columns: usize) -> String {
        // A string like e.g. "a,b,c"
        let columns = COLUMN_NAMES[..number_of_columns].join(",");
        let placeholders = vec!["?"; number_of_columns].join(",");
        format!(
            "INSERT INTO {} ({}) VALUES ({})",
            self.name, columns, placeholders
        )
    }
}

/// Creates the table and assures it is empty. Columns are named a,b,c, etc.
pub fn setup_empty_table_mssql(
    conn: &Connection,
    table_name: &str,
    column_types: &[&str],
) -> Result<(), odbc_api::Error> {
    let identity = "int IDENTITY(1,1)";
    setup_empty_table(table_name, column_types, conn, identity)
}

fn setup_empty_table(
    table_name: &str,
    column_types: &[&str],
    conn: &Connection,
    identity: &str,
) -> Result<(), odbc_api::Error> {
    let drop_table = &format!("DROP TABLE IF EXISTS {table_name}");
    let cols = column_types
        .iter()
        .zip(COLUMN_NAMES)
        .map(|(ty, name)| format!("{name} {ty}"))
        .collect::<Vec<_>>()
        .join(", ");
    let create_table = format!("CREATE TABLE {table_name} (id {identity},{cols});");
    conn.execute(drop_table, (), None)?;
    conn.execute(&create_table, (), None)?;
    Ok(())
}

/// Creates the table and assures it is empty (adapted for PostgreSQL). Columns are named a,b,c, etc.
pub fn setup_empty_table_pg(
    conn: &Connection,
    table_name: &str,
    column_types: &[&str],
) -> Result<(), odbc_api::Error> {
    let identity = "SERIAL PRIMARY KEY";
    setup_empty_table(table_name, column_types, conn, identity)
}

/// Test helper using two commands to roundtrip parquet to and from a data source.
///
/// # Parameters
///
/// * `file`: File used in the roundtrip. Table schema is currently hardcoded.
/// * `table_name`: Each test must use its unique table name, to avoid race conditions with other
///   tests.
fn roundtrip(file: &'static str, table_name: &str) -> Assert {
    // Setup table for test. We use the table name only in this test.
    let conn = env()
        .connect_with_connection_string(MSSQL, ConnectionOptions::default())
        .unwrap();
    conn.execute(&format!("DROP TABLE IF EXISTS {table_name}"), (), None)
        .unwrap();
    conn.execute(
        &format!("CREATE TABLE {table_name} (country VARCHAR(255), population BIGINT);"),
        (),
        None,
    )
    .unwrap();

    // A temporary directory, to be removed at the end of the test.
    let out_dir = tempdir().unwrap();
    // The name of the output parquet file we are going to write. Since it is in a temporary
    // directory it will not outlive the end of the test.
    let out_path = out_dir.path().join("out.par");
    // We need to pass the output path as a string argument.
    let out_str = out_path.to_str().expect("Temporary file path must be utf8");

    let in_path = format!("tests/{file}");

    // Insert parquet
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "insert",
            "--connection-string",
            MSSQL,
            &in_path,
            table_name,
        ])
        .assert()
        .success();

    // Query as parquet
    cargo_bin_cmd!()
        .args([
            "-vvvv",
            "query",
            "--connection-string",
            MSSQL,
            out_str,
            &format!("SELECT country, population FROM {table_name} ORDER BY population;"),
        ])
        .assert()
        .success();

    let expectation = String::from_utf8(
        std::process::Command::new("parquet-read")
            .args(&[&in_path][..])
            .output()
            .unwrap()
            .stdout,
    )
    .unwrap();

    parquet_read_out(out_str).stdout(expectation)
}

/// Consumes a cursor and generates a CSV string from the result set.
fn cursor_to_string(mut cursor: impl Cursor) -> String {
    let batch_size = 20;
    let mut buffer = TextRowSet::for_cursor(batch_size, &mut cursor, None).unwrap();
    let mut row_set_cursor = cursor.bind_buffer(&mut buffer).unwrap();

    let mut text = String::new();

    while let Some(row_set) = row_set_cursor.fetch().unwrap() {
        for row_index in 0..row_set.num_rows() {
            if row_index != 0 {
                text.push('\n');
            }
            for col_index in 0..row_set.num_cols() {
                if col_index != 0 {
                    text.push(',');
                }
                text.push_str(
                    row_set
                        .at_as_str(col_index, row_index)
                        .unwrap()
                        .unwrap_or("NULL"),
                );
            }
        }
    }

    text
}

trait WriteToCw: Sized {
    fn write_batch(col_writer: &mut ColumnWriter, values: &[Self], def_levels: Option<&[i16]>);
}

macro_rules! impl_write_to_cw {
    ($type:ty , $variant:ident) => {
        impl WriteToCw for $type {
            fn write_batch(
                col_writer: &mut ColumnWriter,
                values: &[Self],
                def_levels: Option<&[i16]>,
            ) {
                match col_writer {
                    ColumnWriter::$variant(ref mut cw) => {
                        cw.write_batch(values, def_levels, None).unwrap();
                    }
                    _ => panic!("Unexpected Column Writer type"),
                }
            }
        }
    };
}

impl_write_to_cw!(bool, BoolColumnWriter);
impl_write_to_cw!(i32, Int32ColumnWriter);
impl_write_to_cw!(i64, Int64ColumnWriter);
impl_write_to_cw!(f32, FloatColumnWriter);
impl_write_to_cw!(f64, DoubleColumnWriter);
impl_write_to_cw!(ByteArray, ByteArrayColumnWriter);
impl_write_to_cw!(FixedLenByteArray, FixedLenByteArrayColumnWriter);