tuitab 0.8.2

//! Writing sheet edits back into the SQLite / DuckDB file they came from.
//!
//! Fixtures are built here rather than checked in: a binary database in `test_data/`
//! would have to be regenerated by hand every time a column is added to one of these
//! tests, and it hides what the test is actually about.

use std::path::{Path, PathBuf};
use tuitab::data::io::db_write::{self, StmtKind, TableSource, Val};
use tuitab::data::io::{load_duckdb_table_full, load_sqlite_table_full};

/// Scratch directory for fixtures.  Deliberately inside the project, not `$TMPDIR`.
fn scratch(name: &str) -> PathBuf {
    let dir = Path::new(env!("CARGO_MANIFEST_DIR"))
        .join("tmp")
        .join("db-write-tests");
    std::fs::create_dir_all(&dir).unwrap();
    let path = dir.join(name);
    let _ = std::fs::remove_file(&path);
    path
}

/// `users`: an integer key, a text column, a numeric column, and a text column holding
/// both a NULL and an empty string so the two can be told apart.
fn sqlite_fixture(name: &str) -> PathBuf {
    let path = scratch(name);
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, score INTEGER, note TEXT);
         INSERT INTO users VALUES (1, 'ann',  10, NULL);
         INSERT INTO users VALUES (2, 'bob',  20, '');
         INSERT INTO users VALUES (3, 'cara', 30, 'hi');
         INSERT INTO users VALUES (4, 'dan',  40, 'yo');
         CREATE TABLE other (k TEXT);
         INSERT INTO other VALUES ('untouched');",
    )
    .unwrap();
    path
}

fn duckdb_fixture(name: &str) -> PathBuf {
    let path = scratch(name);
    let conn = duckdb::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, score INTEGER, note TEXT);
         INSERT INTO users VALUES (1, 'ann',  10, NULL);
         INSERT INTO users VALUES (2, 'bob',  20, '');
         INSERT INTO users VALUES (3, 'cara', 30, 'hi');
         INSERT INTO users VALUES (4, 'dan',  40, 'yo');
         CREATE TABLE other (k TEXT);
         INSERT INTO other VALUES ('untouched');",
    )
    .unwrap();
    path
}

fn open_sqlite(path: &Path) -> (tuitab::data::dataframe::DataFrame, TableSource) {
    let (df, src) = load_sqlite_table_full(path, "users").unwrap();
    (df, src.expect("users is addressable by rowid"))
}

/// The statements a plan would show, in order.
fn shown_sql(plan: &db_write::WritePlan) -> Vec<String> {
    plan.stmts.iter().map(|s| s.display.clone()).collect()
}

fn rows_of(path: &Path, sql: &str) -> Vec<Vec<String>> {
    let conn = rusqlite::Connection::open(path).unwrap();
    let mut stmt = conn.prepare(sql).unwrap();
    let n = stmt.column_count();
    let mut out = Vec::new();
    let mut rows = stmt.query([]).unwrap();
    while let Some(row) = rows.next().unwrap() {
        out.push(
            (0..n)
                .map(|i| match row.get::<_, rusqlite::types::Value>(i).unwrap() {
                    rusqlite::types::Value::Null => "<null>".to_string(),
                    rusqlite::types::Value::Integer(v) => v.to_string(),
                    rusqlite::types::Value::Real(v) => v.to_string(),
                    rusqlite::types::Value::Text(v) => v,
                    rusqlite::types::Value::Blob(_) => "<blob>".to_string(),
                })
                .collect(),
        );
    }
    out
}

// ── Loading ───────────────────────────────────────────────────────────────────────

#[test]
fn a_null_and_an_empty_string_survive_loading_as_different_values() {
    let path = sqlite_fixture("load-nulls.sqlite");
    let (df, _) = open_sqlite(&path);
    let note = df.column_index("note").unwrap();
    assert!(df.is_null_physical(0, note), "row 1 note was NULL");
    assert!(
        !df.is_null_physical(1, note),
        "row 2 note was an empty string"
    );
    assert_eq!(df.get_physical(1, note), "");
    assert_eq!(df.get_editable(0, note), "\\N");
}

#[test]
fn the_row_identifier_is_captured_but_never_becomes_a_column() {
    let path = sqlite_fixture("load-rowid.sqlite");
    let (df, src) = open_sqlite(&path);
    assert_eq!(
        df.columns
            .iter()
            .map(|c| c.name.as_str())
            .collect::<Vec<_>>(),
        ["id", "name", "score", "note"]
    );
    assert_eq!(src.key_col, "rowid");
    assert_eq!(
        df.db_rows.as_ref().unwrap().ids,
        vec![Some(1), Some(2), Some(3), Some(4)]
    );
}

#[test]
fn declared_types_come_from_the_schema() {
    let path = sqlite_fixture("load-types.sqlite");
    let (_, src) = open_sqlite(&path);
    let decl: Vec<_> = src.columns.iter().map(|c| c.decl.name()).collect();
    assert_eq!(decl, ["integer", "text", "integer", "text"]);
}

// ── Statement generation ──────────────────────────────────────────────────────────

#[test]
fn an_untouched_sheet_produces_no_statements() {
    let path = sqlite_fixture("gen-empty.sqlite");
    let (df, src) = open_sqlite(&path);
    let plan = db_write::build_plan(&src, &df).unwrap();
    assert!(plan.is_empty(), "{:?}", plan.stmts);
    assert_eq!(plan.summary(), "no changes");
}

#[test]
fn one_edit_becomes_one_update_addressed_by_rowid() {
    let path = sqlite_fixture("gen-one.sqlite");
    let (mut df, src) = open_sqlite(&path);
    let name = df.column_index("name").unwrap();
    df.set_cell(2, name, "CARA".to_string()).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.stmts.len(), 1);
    assert_eq!(plan.updates, 1);
    let stmt = &plan.stmts[0];
    assert_eq!(stmt.kind, StmtKind::Update);
    assert_eq!(
        stmt.sql,
        r#"UPDATE "users" SET "name" = ? WHERE "rowid" = ?"#
    );
    assert_eq!(stmt.params, vec![Val::Text("CARA".into()), Val::Int(3)]);
    assert_eq!(
        stmt.display,
        r#"UPDATE "users" SET "name" = 'CARA' WHERE "rowid" = 3"#
    );
}

#[test]
fn the_same_value_across_rows_collapses_into_a_single_statement() {
    let path = sqlite_fixture("gen-bulk.sqlite");
    let (mut df, src) = open_sqlite(&path);
    let name = df.column_index("name").unwrap();
    let rows: std::collections::HashSet<usize> = [0, 1, 3].into_iter().collect();
    df.set_cells_bulk(&rows, name, "same".to_string()).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.stmts.len(), 1, "{:?}", plan.stmts);
    assert_eq!(plan.updates, 3);
    assert_eq!(
        plan.stmts[0].display,
        r#"UPDATE "users" SET "name" = 'same' WHERE "rowid" IN (1, 2, 4)"#
    );
}

#[test]
fn two_columns_of_one_row_become_a_single_update() {
    let path = sqlite_fixture("gen-two-cols.sqlite");
    let (mut df, src) = open_sqlite(&path);
    let name = df.column_index("name").unwrap();
    let score = df.column_index("score").unwrap();
    df.set_cell(0, name, "ANN".to_string()).unwrap();
    df.set_cell(0, score, "99".to_string()).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.stmts.len(), 1);
    assert_eq!(
        plan.stmts[0].display,
        r#"UPDATE "users" SET "name" = 'ANN', "score" = 99 WHERE "rowid" = 1"#
    );
}

#[test]
fn the_null_literal_writes_a_real_null_and_an_emptied_text_cell_writes_an_empty_string() {
    let path = sqlite_fixture("gen-null.sqlite");
    let (mut df, src) = open_sqlite(&path);
    let note = df.column_index("note").unwrap();
    df.set_cell(2, note, "\\N".to_string()).unwrap();
    df.set_cell(3, note, String::new()).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    let shown: Vec<&str> = plan.stmts.iter().map(|s| s.display.as_str()).collect();
    assert!(
        shown
            .iter()
            .any(|s| s.contains(r#""note" = NULL WHERE "rowid" = 3"#)),
        "{:?}",
        shown
    );
    assert!(
        shown
            .iter()
            .any(|s| s.contains(r#""note" = '' WHERE "rowid" = 4"#)),
        "{:?}",
        shown
    );
}

#[test]
fn an_emptied_numeric_cell_becomes_null_rather_than_failing() {
    let path = sqlite_fixture("gen-empty-num.sqlite");
    let (mut df, src) = open_sqlite(&path);
    let score = df.column_index("score").unwrap();
    df.set_cell(0, score, String::new()).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.stmts[0].params[0], Val::Null);
}

#[test]
fn a_value_that_does_not_fit_the_column_stops_the_save_before_any_sql_exists() {
    let path = sqlite_fixture("gen-badtype.sqlite");
    let (mut df, src) = open_sqlite(&path);
    let score = df.column_index("score").unwrap();
    df.set_cell(1, score, "abc".to_string()).unwrap();

    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("score"), "{}", err);
    assert!(
        err.contains("row 2"),
        "names the row the user sees: {}",
        err
    );
    assert!(err.contains("not an integer"), "{}", err);
}

#[test]
fn quotes_in_values_and_column_names_survive() {
    let path = scratch("gen-quotes.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(r#"CREATE TABLE t ("we""ird" TEXT); INSERT INTO t VALUES ('x');"#)
        .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    df.set_cell(0, 0, "O'Brien".to_string()).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(
        plan.stmts[0].display,
        r#"UPDATE "t" SET "we""ird" = 'O''Brien' WHERE "rowid" = 1"#
    );
    db_write::apply(&src, &plan).unwrap();
    assert_eq!(
        rows_of(&path, r#"SELECT "we""ird" FROM t"#)[0][0],
        "O'Brien"
    );
}

#[test]
fn deleting_rows_produces_deletes_and_editing_a_deleted_row_does_not_also_update_it() {
    let path = sqlite_fixture("gen-delete.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.record_deleted_rows([1usize, 2usize]);

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.deletes, 2);
    assert_eq!(plan.updates, 0);
    assert_eq!(
        plan.stmts[0].display,
        r#"DELETE FROM "users" WHERE "rowid" IN (2, 3)"#
    );
}

// ── Blocking ──────────────────────────────────────────────────────────────────────

#[test]
fn a_frame_that_lost_its_row_identity_cannot_be_written_back() {
    let path = sqlite_fixture("block-identity.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.db_rows = None;
    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("row identity was lost"), "{}", err);
    assert!(err.contains("Save to a different file"), "{}", err);
}

/// A percentage column stores the value divided by 100 and a currency column stores a
/// bare float — those are ways of *showing* a number, and writing one back would put
/// `0.42` where the user is looking at `42%`.
#[test]
fn retyping_to_a_display_format_is_refused_with_a_reason() {
    let path = sqlite_fixture("block-retype.sqlite");
    let (mut df, src) = open_sqlite(&path);
    let score = df.column_index("score").unwrap();
    df.columns[score].db_retype = Some(tuitab::types::ColumnType::Percentage);

    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("display format"), "{}", err);
    assert!(err.contains("rescale"), "{}", err);
    assert!(err.contains("Save to a different file"), "{}", err);
}

// ── Schema changes ────────────────────────────────────────────────────────────────

#[test]
fn a_loaded_column_knows_which_database_column_it_is() {
    let path = sqlite_fixture("schema-origin.sqlite");
    let (df, _) = open_sqlite(&path);
    let origins: Vec<_> = df
        .columns
        .iter()
        .map(|c| c.db_origin.as_deref().unwrap_or("<none>"))
        .collect();
    assert_eq!(origins, ["id", "name", "score", "note"]);
    assert!(df.columns.iter().all(|c| c.db_retype.is_none()));
}

#[test]
fn dropping_a_column_becomes_a_drop_column() {
    let path = sqlite_fixture("schema-drop.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.drop_column(3).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.schema, 1);
    assert_eq!(plan.stmts[0].kind, StmtKind::Schema);
    assert_eq!(
        plan.stmts[0].display,
        r#"ALTER TABLE "users" DROP COLUMN "note""#
    );
}

#[test]
fn renaming_a_column_becomes_a_rename_column() {
    let path = sqlite_fixture("schema-rename.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.rename_column(1, "nom").unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.schema, 1);
    assert_eq!(
        plan.stmts[0].display,
        r#"ALTER TABLE "users" RENAME COLUMN "name" TO "nom""#
    );
}

/// The point of tagging a column with where it came from rather than journalling what
/// happened to it: two renames are still one column, so they are one statement.
#[test]
fn renaming_twice_is_a_single_statement() {
    let path = sqlite_fixture("schema-rename2.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.rename_column(1, "nom").unwrap();
    df.rename_column(1, "handle").unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.schema, 1, "{:?}", plan.stmts);
    assert_eq!(
        plan.stmts[0].display,
        r#"ALTER TABLE "users" RENAME COLUMN "name" TO "handle""#
    );
}

#[test]
fn renaming_a_column_back_to_its_own_name_is_nothing() {
    let path = sqlite_fixture("schema-rename-back.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.rename_column(1, "nom").unwrap();
    df.rename_column(1, "name").unwrap();

    assert!(db_write::build_plan(&src, &df).unwrap().is_empty());
}

#[test]
fn an_added_column_becomes_an_add_column_plus_its_values() {
    let path = sqlite_fixture("schema-add.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.insert_empty_column(4, "tier").unwrap();
    let tier = df.column_index("tier").unwrap();
    df.set_cell(0, tier, "gold".to_string()).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.schema, 1);
    assert_eq!(
        plan.stmts[0].display,
        r#"ALTER TABLE "users" ADD COLUMN "tier" TEXT"#
    );
    // Every row of a new column is a change, and identical values collapse.
    let updates: Vec<&str> = plan
        .stmts
        .iter()
        .filter(|s| s.kind == StmtKind::Update)
        .map(|s| s.display.as_str())
        .collect();
    assert_eq!(updates.len(), 2, "{:?}", updates);
    assert!(
        updates.iter().any(|s| s.contains("'gold'")),
        "{:?}",
        updates
    );
    assert!(
        updates
            .iter()
            .any(|s| s.contains(r#""tier" = '' WHERE "rowid" IN (2, 3, 4)"#)),
        "{:?}",
        updates
    );
}

#[test]
fn a_column_added_and_then_dropped_leaves_nothing_behind() {
    let path = sqlite_fixture("schema-add-drop.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.insert_empty_column(4, "tier").unwrap();
    let tier = df.column_index("tier").unwrap();
    df.drop_column(tier).unwrap();

    assert!(db_write::build_plan(&src, &df).unwrap().is_empty());
}

#[test]
fn renaming_a_column_added_in_this_session_is_an_add_not_a_rename() {
    let path = sqlite_fixture("schema-add-rename.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.insert_empty_column(4, "tmp").unwrap();
    let tmp = df.column_index("tmp").unwrap();
    df.rename_column(tmp, "tier").unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    let schema: Vec<&str> = plan
        .stmts
        .iter()
        .filter(|s| s.kind == StmtKind::Schema)
        .map(|s| s.display.as_str())
        .collect();
    assert_eq!(schema, [r#"ALTER TABLE "users" ADD COLUMN "tier" TEXT"#]);
}

/// `zr`/`zg` force the column to String as a side effect of replacing text in it. That
/// is not a type the user asked for, and it must never reach the database as one.
#[test]
fn a_find_and_replace_produces_no_schema_statement() {
    let path = sqlite_fixture("schema-replace.sqlite");
    let (mut df, src) = open_sqlite(&path);
    let name = df.column_index("name").unwrap();
    df.col_replace(name, "ann", "ANN", true).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.schema, 0, "{:?}", plan.stmts);
    assert_eq!(plan.updates, 1);
}

/// Pinning is how you keep a column in sight while scrolling. It is not a migration.
#[test]
fn pinning_a_column_produces_no_sql_at_all() {
    let path = sqlite_fixture("schema-pin.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.toggle_pin_column(2).unwrap();
    df.toggle_pin_column(3).unwrap();

    assert!(db_write::build_plan(&src, &df).unwrap().is_empty());
    assert_eq!(
        df.columns
            .iter()
            .map(|c| c.name.as_str())
            .collect::<Vec<_>>(),
        ["id", "name", "score", "note"],
        "the frame must not have been reordered"
    );
}

#[test]
fn unpinning_restores_nothing_because_nothing_moved() {
    let path = sqlite_fixture("schema-unpin.sqlite");
    let (mut df, _) = open_sqlite(&path);
    let before: Vec<String> = df.columns.iter().map(|c| c.name.clone()).collect();
    for i in [1, 3] {
        df.toggle_pin_column(i).unwrap();
    }
    for i in [1, 3] {
        df.toggle_pin_column(i).unwrap();
    }
    let after: Vec<String> = df.columns.iter().map(|c| c.name.clone()).collect();
    assert_eq!(before, after);
    assert!(df.columns.iter().all(|c| !c.pinned));
}

#[test]
fn statements_are_ordered_drop_then_rename_then_add_then_rows() {
    let path = sqlite_fixture("schema-order.sqlite");
    let (mut df, src) = open_sqlite(&path);
    // Drop `note`, rename `score` into the freed name, add a fresh `score`, edit a row.
    df.drop_column(3).unwrap();
    df.rename_column(2, "note").unwrap();
    df.insert_empty_column(3, "score").unwrap();
    df.set_cell(0, 1, "ANN".to_string()).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    let shape: Vec<&str> = plan
        .stmts
        .iter()
        .map(|s| s.display.split_whitespace().nth(3).unwrap_or(""))
        .collect();
    assert_eq!(&shape[..3], ["DROP", "RENAME", "ADD"], "{:?}", plan.stmts);
    assert!(
        plan.stmts[3..].iter().all(|s| s.kind != StmtKind::Schema),
        "rows come last"
    );
}

#[test]
fn swapping_two_column_names_goes_through_a_scratch_name() {
    let path = sqlite_fixture("schema-swap-names.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.rename_column(1, "tmp").unwrap();
    df.rename_column(2, "name").unwrap();
    df.rename_column(1, "score").unwrap();

    // No single ALTER can express a swap, so the plan parks one of the two names.
    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.schema, 3, "{:?}", shown_sql(&plan));
    assert!(
        shown_sql(&plan).iter().any(|s| s.contains("__tuitab_swap")),
        "{:?}",
        shown_sql(&plan)
    );

    db_write::apply(&src, &plan).unwrap();

    // The two columns have traded names, and the values stayed with their columns.
    assert_eq!(
        rows_of(
            &path,
            "SELECT name FROM pragma_table_info('users') ORDER BY cid"
        ),
        vec![
            vec!["id".to_string()],
            vec!["score".to_string()],
            vec!["name".to_string()],
            vec!["note".to_string()],
        ]
    );
    assert_eq!(
        rows_of(&path, "SELECT score, name FROM users ORDER BY id LIMIT 1"),
        vec![vec!["ann".to_string(), "10".to_string()]]
    );
}

// ── Rebuilding the table ──────────────────────────────────────────────────────────

#[test]
fn reordering_columns_rebuilds_the_table_and_keeps_every_row() {
    let path = sqlite_fixture("rebuild-reorder.sqlite");
    let before = rows_of(&path, "SELECT id, name, score, note FROM users ORDER BY id");
    let (mut df, src) = open_sqlite(&path);
    df.swap_columns(1, 2).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert!(plan.rebuild);
    let shown: Vec<&str> = plan.stmts.iter().map(|s| s.display.as_str()).collect();
    assert!(
        shown[0].starts_with(r#"CREATE TABLE "users__tuitab_rebuild""#),
        "{:?}",
        shown
    );
    assert!(
        shown.iter().any(|s| s.contains("DROP TABLE")),
        "{:?}",
        shown
    );

    db_write::apply(&src, &plan).unwrap();

    // Same rows, same rowids, new column order.
    assert_eq!(
        rows_of(
            &path,
            "SELECT name FROM pragma_table_info('users') ORDER BY cid"
        )
        .into_iter()
        .map(|r| r[0].clone())
        .collect::<Vec<_>>(),
        ["id", "score", "name", "note"]
    );
    assert_eq!(
        before,
        rows_of(&path, "SELECT id, name, score, note FROM users ORDER BY id")
    );
    assert_eq!(
        rows_of(&path, "SELECT rowid FROM users ORDER BY rowid")
            .into_iter()
            .map(|r| r[0].clone())
            .collect::<Vec<_>>(),
        ["1", "2", "3", "4"]
    );
}

#[test]
fn a_rebuild_preserves_rowids_across_gaps() {
    let path = sqlite_fixture("rebuild-rowids.sqlite");
    rusqlite::Connection::open(&path)
        .unwrap()
        .execute_batch("DELETE FROM users WHERE id = 2")
        .unwrap();

    let (mut df, src) = open_sqlite(&path);
    df.swap_columns(0, 1).unwrap();
    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();

    assert_eq!(
        rows_of(&path, "SELECT rowid FROM users ORDER BY rowid")
            .into_iter()
            .map(|r| r[0].clone())
            .collect::<Vec<_>>(),
        ["1", "3", "4"],
        "the gap left by the deleted row must survive"
    );
}

#[test]
fn a_rebuild_recreates_the_tables_indexes_and_triggers() {
    let path = scratch("rebuild-objects.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, a TEXT, b TEXT);
         INSERT INTO t VALUES (1, 'x', 'p'), (2, 'y', 'q');
         CREATE INDEX idx_a ON t(a);
         CREATE TRIGGER trg AFTER UPDATE ON t BEGIN SELECT 1; END;",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    df.swap_columns(1, 2).unwrap();
    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();

    let objects: Vec<String> = rows_of(
        &path,
        "SELECT type || ' ' || name FROM sqlite_master WHERE tbl_name = 't' ORDER BY name",
    )
    .into_iter()
    .map(|r| r[0].clone())
    .collect();
    assert!(
        objects.contains(&"index idx_a".to_string()),
        "{:?}",
        objects
    );
    assert!(
        objects.contains(&"trigger trg".to_string()),
        "{:?}",
        objects
    );
    assert_eq!(rows_of(&path, "PRAGMA integrity_check")[0][0], "ok");
}

#[test]
fn sqlite_changes_a_column_type_by_rebuilding() {
    let path = scratch("rebuild-retype.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, n TEXT);
         INSERT INTO t VALUES (1, '10'), (2, '20');",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    df.columns[1].db_retype = Some(tuitab::types::ColumnType::Integer);

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert!(plan.rebuild);
    db_write::apply(&src, &plan).unwrap();

    assert_eq!(
        rows_of(
            &path,
            "SELECT type FROM pragma_table_info('t') WHERE name = 'n'"
        )
        .first()
        .map(|r| r[0].clone())
        .unwrap_or_default(),
        "INTEGER"
    );
    assert_eq!(
        rows_of(&path, "SELECT typeof(n) FROM t ORDER BY id")[0][0],
        "integer"
    );
}

/// Where a *new* column sits in the sheet says nothing about where it goes in the
/// table: `ADD COLUMN` appends, as it does in every other tool, and rebuilding a table
/// to place a column that did not exist a moment ago is not a trade worth making. The
/// sheet reloads after the save and shows where it actually landed.
#[test]
fn adding_a_column_in_the_middle_does_not_rebuild_the_table() {
    let path = sqlite_fixture("rebuild-mixed.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.drop_column(3).unwrap();
    df.insert_empty_column(1, "tier").unwrap();
    df.set_cell(0, 1, "gold".to_string()).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert!(!plan.rebuild, "{:?}", plan.stmts);
    db_write::apply(&src, &plan).unwrap();

    let cols: Vec<String> = rows_of(
        &path,
        "SELECT name FROM pragma_table_info('users') ORDER BY cid",
    )
    .into_iter()
    .map(|r| r[0].clone())
    .collect();
    assert_eq!(cols, ["id", "name", "score", "tier"]);
    assert_eq!(
        rows_of(&path, "SELECT tier, name FROM users ORDER BY id")[0],
        ["gold", "ann"]
    );
}

/// Moving an existing column, on the other hand, is a request about the table.
#[test]
fn a_rebuild_carries_a_dropped_column_and_a_reorder_in_one_go() {
    let path = sqlite_fixture("rebuild-mixed2.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.drop_column(3).unwrap();
    df.swap_columns(1, 2).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert!(plan.rebuild);
    db_write::apply(&src, &plan).unwrap();

    let cols: Vec<String> = rows_of(
        &path,
        "SELECT name FROM pragma_table_info('users') ORDER BY cid",
    )
    .into_iter()
    .map(|r| r[0].clone())
    .collect();
    assert_eq!(cols, ["id", "score", "name"]);
    assert_eq!(
        rows_of(&path, "SELECT name, score FROM users ORDER BY id")[0],
        ["ann", "10"]
    );
}

#[test]
fn a_table_referenced_by_a_foreign_key_refuses_the_rebuild() {
    let path = scratch("rebuild-fk.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, a TEXT, b TEXT);
         INSERT INTO t VALUES (1, 'x', 'p');
         CREATE TABLE child (id INTEGER, t_id INTEGER REFERENCES t(id));",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    df.swap_columns(1, 2).unwrap();

    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("foreign key into it"), "{}", err);
    assert!(err.contains("child"), "{}", err);
}

#[test]
fn a_table_a_view_is_built_on_refuses_the_rebuild() {
    let path = scratch("rebuild-view.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, a TEXT, b TEXT);
         INSERT INTO t VALUES (1, 'x', 'p');
         CREATE VIEW v AS SELECT a FROM t;",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    df.swap_columns(1, 2).unwrap();

    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("view 'v' is built on it"), "{}", err);
}

#[test]
fn a_table_with_a_check_constraint_refuses_the_rebuild() {
    let path = scratch("rebuild-check.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, a TEXT, n INTEGER CHECK (n > 0));
         INSERT INTO t VALUES (1, 'x', 5);",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    df.swap_columns(1, 2).unwrap();

    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("definition uses CHECK"), "{}", err);
}

#[test]
fn a_table_with_a_unique_constraint_refuses_the_rebuild() {
    let path = scratch("rebuild-unique.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, a TEXT UNIQUE, b TEXT);
         INSERT INTO t VALUES (1, 'x', 'p');",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    df.swap_columns(1, 2).unwrap();

    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("UNIQUE or PRIMARY KEY constraint"), "{}", err);
}

#[test]
fn a_rebuild_that_fails_leaves_the_table_exactly_as_it_was() {
    let path = sqlite_fixture("rebuild-fail.sqlite");
    let before = rows_of(&path, "SELECT id, name, score, note FROM users ORDER BY id");
    let (mut df, src) = open_sqlite(&path);
    df.swap_columns(1, 2).unwrap();
    let mut plan = db_write::build_plan(&src, &df).unwrap();
    plan.stmts.push(db_write::Stmt {
        sql: "SELECT this_is_not_valid_sql(".to_string(),
        display: "SELECT this_is_not_valid_sql(".to_string(),
        params: Vec::new(),
        kind: StmtKind::Schema,
    });

    assert!(db_write::apply(&src, &plan).is_err());
    assert_eq!(
        before,
        rows_of(&path, "SELECT id, name, score, note FROM users ORDER BY id"),
        "the table must be exactly as it was"
    );
    assert_eq!(
        rows_of(
            &path,
            "SELECT COUNT(*) FROM sqlite_master WHERE name LIKE '%rebuild%'"
        )[0][0],
        "0",
        "no scratch table left behind"
    );
}

#[test]
fn duckdb_reorders_by_rebuilding_after_the_row_changes() {
    let path = duckdb_fixture("rebuild-reorder.duckdb");
    let (mut df, src) = load_duckdb_table_full(&path, "users").unwrap();
    let src = src.unwrap();
    df.set_cell(0, 1, "ANN".to_string()).unwrap();
    df.swap_columns(1, 2).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert!(plan.rebuild);
    // The row change has to land before the rebuild: DuckDB renumbers rowids.
    let update_at = plan
        .stmts
        .iter()
        .position(|s| s.kind == StmtKind::Update)
        .unwrap();
    let create_at = plan
        .stmts
        .iter()
        .position(|s| s.display.starts_with("CREATE TABLE"))
        .unwrap();
    assert!(update_at < create_at, "{:?}", plan.stmts);

    db_write::apply(&src, &plan).unwrap();

    let (after, _) = load_duckdb_table_full(&path, "users").unwrap();
    let names: Vec<&str> = after.columns.iter().map(|c| c.name.as_str()).collect();
    assert_eq!(names, ["id", "score", "name", "note"]);
    assert_eq!(after.get_physical(0, 2), "ANN");
}

// ── Schema changes reaching the database ──────────────────────────────────────────

#[test]
fn an_added_column_reaches_the_database_with_its_values() {
    let path = sqlite_fixture("apply-add.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.insert_empty_column(4, "tier").unwrap();
    let tier = df.column_index("tier").unwrap();
    df.set_cell(0, tier, "gold".to_string()).unwrap();

    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();

    let rows = rows_of(&path, "SELECT id, tier FROM users ORDER BY id");
    assert_eq!(rows[0], ["1", "gold"]);
    assert_eq!(rows[1], ["2", ""]);
}

#[test]
fn a_dropped_column_is_gone_and_the_rest_keep_their_values() {
    let path = sqlite_fixture("apply-drop.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.drop_column(3).unwrap();

    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();

    assert_eq!(
        rows_of(&path, "SELECT id, name, score FROM users ORDER BY id")[0],
        ["1", "ann", "10"]
    );
    assert!(
        rows_of(&path, "SELECT name FROM pragma_table_info('users')")
            .iter()
            .all(|r| r[0] != "note")
    );
}

#[test]
fn a_renamed_column_keeps_its_data_and_an_edit_lands_in_it() {
    let path = sqlite_fixture("apply-rename.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.rename_column(1, "nom").unwrap();
    df.set_cell(0, 1, "ANN".to_string()).unwrap();

    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();

    let rows = rows_of(&path, "SELECT nom FROM users ORDER BY id");
    assert_eq!(rows[0][0], "ANN");
    assert_eq!(rows[1][0], "bob");
}

#[test]
fn a_schema_change_and_a_row_change_land_in_one_transaction() {
    let path = sqlite_fixture("apply-schema-and-rows.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.insert_empty_column(4, "tier").unwrap();
    let tier = df.column_index("tier").unwrap();
    df.set_cell(0, tier, "gold".to_string()).unwrap();
    df.record_deleted_rows([3usize]);

    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();

    let rows = rows_of(&path, "SELECT id, tier FROM users ORDER BY id");
    assert_eq!(rows.len(), 3);
    assert_eq!(rows[0], ["1", "gold"]);
}

#[test]
fn a_failed_schema_change_leaves_the_table_exactly_as_it_was() {
    let path = sqlite_fixture("apply-schema-fail.sqlite");
    let before = rows_of(&path, "SELECT id, name, score, note FROM users ORDER BY id");
    let (mut df, src) = open_sqlite(&path);
    df.rename_column(1, "nom").unwrap();
    let mut plan = db_write::build_plan(&src, &df).unwrap();
    // A statement the engine will reject, after the rename has already been queued.
    plan.stmts.push(db_write::Stmt {
        sql: r#"ALTER TABLE "users" DROP COLUMN "nope""#.to_string(),
        display: r#"ALTER TABLE "users" DROP COLUMN "nope""#.to_string(),
        params: Vec::new(),
        kind: StmtKind::Schema,
    });

    assert!(db_write::apply(&src, &plan).is_err());
    assert_eq!(
        before,
        rows_of(&path, "SELECT id, name, score, note FROM users ORDER BY id"),
        "the rename must have rolled back with the failure"
    );
}

#[test]
fn duckdb_changes_a_column_type_natively() {
    let path = duckdb_fixture("apply-retype.duckdb");
    let (mut df, src) = load_duckdb_table_full(&path, "users").unwrap();
    let src = src.unwrap();
    let note = df.column_index("note").unwrap();
    df.columns[note].db_retype = Some(tuitab::types::ColumnType::Integer);

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(
        plan.stmts[0].display,
        r#"ALTER TABLE "users" ALTER COLUMN "note" TYPE BIGINT"#
    );

    // The fixture has 'hi'/'yo' in `note`, which is not a number — the engine refuses
    // and the transaction rolls back, which is the honest outcome.
    assert!(db_write::apply(&src, &plan).is_err());

    // With numbers in it, the same change goes through.
    duckdb::Connection::open(&path)
        .unwrap()
        .execute_batch("UPDATE users SET note = '7'")
        .unwrap();
    let (df2, src2) = load_duckdb_table_full(&path, "users").unwrap();
    let src2 = src2.unwrap();
    let mut df2 = df2;
    df2.columns[note].db_retype = Some(tuitab::types::ColumnType::Integer);
    db_write::apply(&src2, &db_write::build_plan(&src2, &df2).unwrap()).unwrap();

    let (after, _) = load_duckdb_table_full(&path, "users").unwrap();
    assert_eq!(after.get_physical(0, note), "7");
}

#[test]
fn duckdb_adds_and_drops_columns() {
    let path = duckdb_fixture("apply-schema.duckdb");
    let (mut df, src) = load_duckdb_table_full(&path, "users").unwrap();
    let src = src.unwrap();
    df.drop_column(3).unwrap();
    df.insert_empty_column(3, "tier").unwrap();
    df.set_cell(0, 3, "gold".to_string()).unwrap();

    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();

    let (after, _) = load_duckdb_table_full(&path, "users").unwrap();
    let names: Vec<&str> = after.columns.iter().map(|c| c.name.as_str()).collect();
    assert_eq!(names, ["id", "name", "score", "tier"]);
    assert_eq!(after.get_physical(0, 3), "gold");
}

// ── Execution ─────────────────────────────────────────────────────────────────────

#[test]
fn an_edit_reaches_the_database_and_leaves_every_other_row_alone() {
    let path = sqlite_fixture("apply-edit.sqlite");
    let before = rows_of(&path, "SELECT id, name, score, note FROM users ORDER BY id");
    let (mut df, src) = open_sqlite(&path);
    let name = df.column_index("name").unwrap();
    df.set_cell(2, name, "CARA".to_string()).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    db_write::apply(&src, &plan).unwrap();

    let after = rows_of(&path, "SELECT id, name, score, note FROM users ORDER BY id");
    assert_eq!(after[2][1], "CARA");
    for i in [0, 1, 3] {
        assert_eq!(before[i], after[i], "row {} was not touched", i);
    }
    // The NULL is still a NULL, not an empty string.
    assert_eq!(after[0][3], "<null>");
    assert_eq!(after[1][3], "");
}

#[test]
fn other_tables_are_left_exactly_as_they_were() {
    let path = sqlite_fixture("apply-other.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.set_cell(0, 1, "changed".to_string()).unwrap();
    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();
    assert_eq!(rows_of(&path, "SELECT k FROM other")[0][0], "untouched");
}

#[test]
fn deletes_and_inserts_reach_the_database() {
    let path = sqlite_fixture("apply-rows.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.record_deleted_rows([0usize]);
    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();
    let ids = rows_of(&path, "SELECT id FROM users ORDER BY id");
    assert_eq!(
        ids.iter().map(|r| r[0].as_str()).collect::<Vec<_>>(),
        ["2", "3", "4"]
    );
}

#[test]
fn a_table_that_changed_underneath_is_refused_and_nothing_is_written() {
    let path = sqlite_fixture("apply-drift.sqlite");
    let (mut df, src) = open_sqlite(&path);
    let name = df.column_index("name").unwrap();
    df.set_cell(2, name, "CARA".to_string()).unwrap();
    let plan = db_write::build_plan(&src, &df).unwrap();

    // Someone else edits the same row in the meantime.
    rusqlite::Connection::open(&path)
        .unwrap()
        .execute_batch("UPDATE users SET name = 'elsewhere' WHERE id = 3")
        .unwrap();

    let err = db_write::apply(&src, &plan).unwrap_err().to_string();
    assert!(err.contains("changed since it was opened"), "{}", err);
    let after = rows_of(&path, "SELECT name FROM users ORDER BY id");
    assert_eq!(after[2][0], "elsewhere", "the write must not have happened");
}

#[test]
fn a_type_error_aborts_before_the_transaction_opens() {
    let path = sqlite_fixture("apply-badtype.sqlite");
    let before = rows_of(&path, "SELECT id, name, score, note FROM users ORDER BY id");
    let (mut df, src) = open_sqlite(&path);
    let name = df.column_index("name").unwrap();
    let score = df.column_index("score").unwrap();
    df.set_cell(0, name, "fine".to_string()).unwrap();
    df.set_cell(1, score, "abc".to_string()).unwrap();

    assert!(db_write::build_plan(&src, &df).is_err());
    let after = rows_of(&path, "SELECT id, name, score, note FROM users ORDER BY id");
    assert_eq!(before, after, "the good edit must not have leaked through");
}

/// The trap this design exists to avoid: a filter shrinks `row_order` exactly the way a
/// deletion does, so anything diffing `row_order` would delete the filtered-out rows.
#[test]
fn filtering_rows_out_of_view_never_deletes_them() {
    let path = sqlite_fixture("apply-filter.sqlite");
    let (mut df, src) = open_sqlite(&path);
    // Keep only physical row 2, as a drill-down filter would.
    df.row_order = std::sync::Arc::new(vec![2]);
    df.original_order = df.row_order.clone();
    let name = df.column_index("name").unwrap();
    df.set_cell(2, name, "CARA".to_string()).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.deletes, 0, "{:?}", plan.stmts);
    assert_eq!(plan.updates, 1);

    db_write::apply(&src, &plan).unwrap();
    assert_eq!(rows_of(&path, "SELECT COUNT(*) FROM users")[0][0], "4");
}

#[test]
fn copying_to_a_new_file_keeps_every_table_and_leaves_the_original_alone() {
    let path = sqlite_fixture("copy-src.sqlite");
    let dest = scratch("copy-dest.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.set_cell(0, 1, "copied".to_string()).unwrap();
    let plan = db_write::build_plan(&src, &df).unwrap();

    db_write::copy_db(&src, &dest).unwrap();
    db_write::apply(&src.at(&dest), &plan).unwrap();

    assert_eq!(
        rows_of(&dest, "SELECT name FROM users ORDER BY id")[0][0],
        "copied"
    );
    assert_eq!(
        rows_of(&path, "SELECT name FROM users ORDER BY id")[0][0],
        "ann"
    );
    assert_eq!(rows_of(&dest, "SELECT k FROM other")[0][0], "untouched");
}

// ── DuckDB ────────────────────────────────────────────────────────────────────────

#[test]
fn duckdb_edits_reach_the_database() {
    let path = duckdb_fixture("apply-edit.duckdb");
    let (mut df, src) = load_duckdb_table_full(&path, "users").unwrap();
    let src = src.expect("users is addressable by rowid");
    let name = df.column_index("name").unwrap();
    df.set_cell(2, name, "CARA".to_string()).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    db_write::apply(&src, &plan).unwrap();

    let (after, _) = load_duckdb_table_full(&path, "users").unwrap();
    assert_eq!(after.get_physical(2, name), "CARA");
    assert_eq!(after.get_physical(0, name), "ann");
    assert!(after.is_null_physical(0, 3), "the NULL is still a NULL");
    assert_eq!(after.get_physical(1, 3), "");
}

#[test]
fn duckdb_refuses_a_table_that_changed_underneath() {
    let path = duckdb_fixture("apply-drift.duckdb");
    let (mut df, src) = load_duckdb_table_full(&path, "users").unwrap();
    let src = src.unwrap();
    df.set_cell(2, 1, "CARA".to_string()).unwrap();
    let plan = db_write::build_plan(&src, &df).unwrap();

    duckdb::Connection::open(&path)
        .unwrap()
        .execute_batch("UPDATE users SET name = 'elsewhere' WHERE id = 3")
        .unwrap();

    let err = db_write::apply(&src, &plan).unwrap_err().to_string();
    assert!(err.contains("changed since it was opened"), "{}", err);
}

#[test]
fn duckdb_copies_to_a_new_file() {
    let path = duckdb_fixture("copy-src.duckdb");
    let dest = scratch("copy-dest.duckdb");
    let (mut df, src) = load_duckdb_table_full(&path, "users").unwrap();
    let src = src.unwrap();
    df.set_cell(0, 1, "copied".to_string()).unwrap();
    let plan = db_write::build_plan(&src, &df).unwrap();

    db_write::copy_db(&src, &dest).unwrap();
    db_write::apply(&src.at(&dest), &plan).unwrap();

    let (copied, _) = load_duckdb_table_full(&dest, "users").unwrap();
    assert_eq!(copied.get_physical(0, 1), "copied");
    let (original, _) = load_duckdb_table_full(&path, "users").unwrap();
    assert_eq!(original.get_physical(0, 1), "ann");
    assert!(load_duckdb_table_full(&dest, "other").is_ok());
}

/// Floats are where a careless drift check falls over: SQLite renders REAL 1000.0 as
/// "1000.0" while Rust's `f64::to_string` gives "1000", so a check that read the value
/// back any other way than the loader does would report drift on every untouched row.
#[test]
fn a_real_column_does_not_report_phantom_drift() {
    let path = scratch("real-drift.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT, salary REAL);
         INSERT INTO t VALUES (1, 'ann', 1000.0), (2, 'bob', 2000.5), (3, 'cara', 0.1);",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    let name = df.column_index("name").unwrap();
    df.set_cell(0, name, "ANN".to_string()).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    db_write::apply(&src, &plan).expect("an untouched REAL column must not look like drift");
    assert_eq!(
        rows_of(&path, "SELECT name FROM t ORDER BY id")[0][0],
        "ANN"
    );
    assert_eq!(
        rows_of(&path, "SELECT salary FROM t ORDER BY id")[1][0],
        "2000.5"
    );
}

/// Editing a REAL column writes a number, not a quoted string.
#[test]
fn a_number_is_written_as_a_number() {
    let path = scratch("real-write.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, salary REAL); INSERT INTO t VALUES (1, 1.0);",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    df.set_cell(0, 1, "12.5".to_string()).unwrap();
    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.stmts[0].params[0], Val::Real(12.5));
    db_write::apply(&src, &plan).unwrap();
    assert_eq!(
        rows_of(&path, "SELECT typeof(salary), salary FROM t")[0],
        ["real", "12.5"]
    );
}

/// A generated column can be read but never written, and SQLite says so up front.
#[test]
fn a_generated_column_is_refused_before_the_engine_sees_it() {
    let path = scratch("generated.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (a INTEGER, b INTEGER GENERATED ALWAYS AS (a * 2) VIRTUAL);
         INSERT INTO t (a) VALUES (5);",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    assert!(src.column("b").unwrap().generated, "xinfo marks it");
    df.set_cell(0, 1, "99".to_string()).unwrap();

    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("generated by the database"), "{}", err);
}

/// Indexes, views and unrelated tables are not part of what a writeback touches.
#[test]
fn the_shape_of_the_database_is_unchanged_by_a_write() {
    let path = scratch("schema-intact.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE employees (id INTEGER PRIMARY KEY, name TEXT NOT NULL, dept TEXT, salary REAL);
         INSERT INTO employees VALUES (1,'Ann','eng',1000.0),(2,'Bob','eng',2000.5);
         CREATE INDEX idx_dept ON employees(dept);
         CREATE VIEW eng AS SELECT * FROM employees WHERE dept='eng';
         CREATE TABLE audit (msg TEXT);
         INSERT INTO audit VALUES ('do not touch');",
    )
    .unwrap();
    drop(conn);

    let schema_of = |p: &Path| -> Vec<String> {
        rows_of(
            p,
            "SELECT type || ' ' || name FROM sqlite_master ORDER BY name",
        )
        .into_iter()
        .map(|r| r[0].clone())
        .collect()
    };
    let before = schema_of(&path);

    let (mut df, src) = load_sqlite_table_full(&path, "employees").unwrap();
    let src = src.unwrap();
    df.set_cell(0, 1, "Anna".to_string()).unwrap();
    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();

    assert_eq!(before, schema_of(&path), "schema must be untouched");
    assert_eq!(
        rows_of(&path, "SELECT msg FROM audit")[0][0],
        "do not touch"
    );
    assert_eq!(
        rows_of(&path, "SELECT name FROM eng ORDER BY id")[0][0],
        "Anna"
    );
}

/// A row added after load has no key yet, so it is inserted rather than updated.
#[test]
fn an_added_row_is_inserted() {
    let path = sqlite_fixture("apply-insert.sqlite");
    let (mut df, src) = open_sqlite(&path);

    // What a paste does: rows on the end of the frame, with no database row behind them.
    // Matching dtypes: `id` and `score` are declared INTEGER and load as Int64.
    let added = polars::prelude::DataFrame::new(
        1,
        vec![
            polars::prelude::Column::new("id".into(), vec![Some(99i64)]),
            polars::prelude::Column::new("name".into(), vec![Some("eve")]),
            polars::prelude::Column::new("score".into(), vec![Some(50i64)]),
            polars::prelude::Column::new("note".into(), vec![None::<&str>]),
        ],
    )
    .unwrap();
    let height = df.df.height();
    df.df.vstack_mut(&added).unwrap();
    std::sync::Arc::make_mut(&mut df.row_order).push(height);
    std::sync::Arc::make_mut(&mut df.original_order).push(height);
    df.record_added_rows(1);

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.inserts, 1);
    assert_eq!(plan.updates, 0);
    assert_eq!(
        plan.stmts[0].display,
        r#"INSERT INTO "users" ("id", "name", "score", "note") VALUES (99, 'eve', 50, NULL)"#
    );

    db_write::apply(&src, &plan).unwrap();
    let rows = rows_of(&path, "SELECT id, name, note FROM users ORDER BY id");
    assert_eq!(rows.len(), 5);
    assert_eq!(rows[4], ["99", "eve", "<null>"]);
}

/// After a write the sheet is reloaded, so saving again must find nothing to do.
///
/// Without the reload an INSERT would run twice — the frame still holds a row with no
/// key, and the key the database just assigned is unknown here.
#[test]
fn saving_twice_does_not_apply_the_same_change_again() {
    let path = sqlite_fixture("apply-twice.sqlite");
    let (mut df, src) = open_sqlite(&path);
    let name = df.column_index("name").unwrap();
    df.set_cell(0, name, "ANN".to_string()).unwrap();
    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();

    // What the app does after a successful write.
    let (reloaded_df, reloaded_src) = load_sqlite_table_full(&path, "users").unwrap();
    let plan = db_write::build_plan(&reloaded_src.unwrap(), &reloaded_df).unwrap();
    assert!(
        plan.is_empty(),
        "a reloaded sheet has nothing to write: {:?}",
        plan.stmts
    );
}

/// Saving a database elsewhere copies the whole file and asks nothing, which is only
/// safe while "elsewhere" is empty.
#[test]
fn copying_onto_an_existing_file_is_refused_rather_than_overwriting_it() {
    let path = sqlite_fixture("copy-guard-src.sqlite");
    let dest = sqlite_fixture("copy-guard-dest.sqlite");
    let existing = rows_of(&dest, "SELECT k FROM other");

    let (df, src) = open_sqlite(&path);
    let err = db_write::copy_db(&src, &dest).unwrap_err().to_string();
    assert!(err.contains("already exists"), "{}", err);
    assert_eq!(
        rows_of(&dest, "SELECT k FROM other"),
        existing,
        "the destination must be untouched"
    );
    drop(df);
}

/// The NULL sentinel belongs to database sheets.  A CSV has no null to mean, and a
/// JSON sheet writes into its document tree first — storing the two characters
/// literally there while the table showed NULL would be worse than not offering it.
#[test]
fn the_null_sentinel_is_inert_on_a_sheet_that_did_not_come_from_a_database() {
    let mut df = tuitab::data::io::load_file(Path::new("test_data/sample.csv"), None).unwrap();
    assert!(df.db_rows.is_none());
    let name = df.column_index("name").unwrap();
    df.set_cell(0, name, "\\N".to_string()).unwrap();
    assert_eq!(
        df.get_physical(0, name),
        "\\N",
        "stored verbatim, not turned into NULL"
    );
    assert!(!df.is_null_physical(0, name));
    assert_eq!(df.get_editable(0, name), "\\N");
}

// ── Refusing what the engine would refuse anyway, but with a sentence ─────────────

#[test]
fn dropping_the_primary_key_column_is_refused() {
    let path = sqlite_fixture("preflight-pk.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.drop_column(0).unwrap();

    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("primary key"), "{}", err);
    assert!(err.contains("Save to a different file"), "{}", err);
}

#[test]
fn dropping_an_indexed_column_is_refused_and_names_the_index() {
    let path = scratch("preflight-index.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, email TEXT, note TEXT);
         INSERT INTO t VALUES (1, 'a@b.c', 'x');
         CREATE INDEX idx_email ON t(email);",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    let email = df.column_index("email").unwrap();
    df.drop_column(email).unwrap();

    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("idx_email"), "{}", err);
    assert!(err.contains("Drop the index first"), "{}", err);
    // Nothing was attempted, so the column is still there.
    assert_eq!(rows_of(&path, "SELECT email FROM t")[0][0], "a@b.c");
}

#[test]
fn dropping_a_column_a_view_depends_on_is_refused() {
    let path = scratch("preflight-view.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, dept TEXT, note TEXT);
         INSERT INTO t VALUES (1, 'eng', 'x');
         CREATE VIEW eng AS SELECT id FROM t WHERE dept = 'eng';",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    let dept = df.column_index("dept").unwrap();
    df.drop_column(dept).unwrap();

    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("view 'eng'"), "{}", err);
}

/// A column whose name merely contains another column's name is not a reference.
#[test]
fn a_similarly_named_column_does_not_trip_the_dependency_scan() {
    let path = scratch("preflight-substring.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, dept TEXT, dept_code TEXT);
         INSERT INTO t VALUES (1, 'eng', 'E1');
         CREATE VIEW v AS SELECT dept_code FROM t;",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    let dept = df.column_index("dept").unwrap();
    df.drop_column(dept).unwrap();

    let plan = db_write::build_plan(&src, &df).expect("dept_code is not dept");
    assert_eq!(plan.schema, 1);
}

/// The value-level drift check compares cells; a column appearing or vanishing behind
/// tuitab's back needs its own look.
#[test]
fn a_table_whose_columns_changed_underneath_is_refused() {
    let path = sqlite_fixture("shape-drift.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.rename_column(1, "nom").unwrap();
    let plan = db_write::build_plan(&src, &df).unwrap();

    rusqlite::Connection::open(&path)
        .unwrap()
        .execute_batch("ALTER TABLE users ADD COLUMN surprise TEXT")
        .unwrap();

    let err = db_write::apply(&src, &plan).unwrap_err().to_string();
    assert!(err.contains("columns of 'users' changed"), "{}", err);
    // The rename must not have happened.
    assert_eq!(
        rows_of(&path, "SELECT name FROM users ORDER BY id")[0][0],
        "ann"
    );
}

/// A computed column is named after the expression that made it — `=score * 2` — which
/// is a legal identifier only in quotes.
#[test]
fn a_computed_column_is_added_under_its_quoted_name() {
    let path = sqlite_fixture("schema-computed.sqlite");
    let (mut df, src) = open_sqlite(&path);
    let expr = tuitab::data::expression::Expr::parse("score * 2").unwrap();
    df.add_computed_column("=score * 2", &expr, 2).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(
        plan.stmts[0].display,
        // INTEGER: `score` is declared INTEGER and now loads as one, so doubling it
        // stays whole and the new column is told the same thing.
        r#"ALTER TABLE "users" ADD COLUMN "=score * 2" INTEGER"#
    );

    db_write::apply(&src, &plan).unwrap();
    let rows = rows_of(&path, r#"SELECT "=score * 2" FROM users ORDER BY id"#);
    assert_eq!(rows[0][0], "20");
    assert_eq!(rows[3][0], "80");
}

/// Splitting a column adds several, all of them plain text.
#[test]
fn split_columns_are_added_as_text() {
    let path = scratch("schema-split.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, city TEXT);
         INSERT INTO t VALUES (1, 'Berlin/DE'), (2, 'Paris/FR');",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    let city = df.column_index("city").unwrap();
    df.col_split(city, "/").unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.schema, 2, "{:?}", plan.stmts);
    db_write::apply(&src, &plan).unwrap();

    let rows = rows_of(&path, r#"SELECT "city.1", "city.2" FROM t ORDER BY id"#);
    assert_eq!(rows[0], ["Berlin", "DE"]);
    assert_eq!(rows[1], ["Paris", "FR"]);
}

/// A realistic file: NOT NULL, an index, a view, a trigger and a bystander table.
/// Adding a column and editing a row must leave every one of them alone.
#[test]
fn a_real_schema_survives_an_add_column_untouched() {
    let path = scratch("real-schema.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE employees (id INTEGER PRIMARY KEY, name TEXT NOT NULL, dept TEXT, salary REAL);
         INSERT INTO employees VALUES (1,'Ann','eng',1000.0),(2,'Bob','eng',2000.5);
         CREATE INDEX idx_dept ON employees(dept);
         CREATE VIEW eng AS SELECT * FROM employees WHERE dept='eng';
         CREATE TRIGGER trg AFTER UPDATE ON employees BEGIN SELECT 1; END;
         CREATE TABLE audit (msg TEXT);
         INSERT INTO audit VALUES ('do not touch');",
    )
    .unwrap();
    drop(conn);

    let others = |p: &Path| -> Vec<String> {
        rows_of(
            p,
            "SELECT type || ' ' || name FROM sqlite_master \
             WHERE name NOT IN ('employees') ORDER BY name",
        )
        .into_iter()
        .map(|r| r[0].clone())
        .collect()
    };
    let before = others(&path);

    let (mut df, src) = load_sqlite_table_full(&path, "employees").unwrap();
    let src = src.unwrap();
    df.insert_empty_column(4, "level").unwrap();
    df.set_cell(0, 4, "senior".to_string()).unwrap();
    df.set_cell(0, 1, "Anna".to_string()).unwrap();

    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();

    assert_eq!(
        before,
        others(&path),
        "index, view and trigger must survive"
    );
    assert_eq!(
        rows_of(&path, "SELECT msg FROM audit")[0][0],
        "do not touch"
    );
    assert_eq!(
        rows_of(&path, "SELECT name, level FROM employees ORDER BY id")[0],
        ["Anna", "senior"]
    );
    // The REAL column was never touched, and is still a REAL.
    assert_eq!(
        rows_of(&path, "SELECT typeof(salary) FROM employees ORDER BY id")[1][0],
        "real"
    );
    assert_eq!(rows_of(&path, "PRAGMA integrity_check")[0][0], "ok");
}

/// `INT` and `INTEGER` are the same storage class. Pressing `t` on a column that
/// already holds integers must not be read as a request to change its type — on SQLite
/// that would refuse the whole save.
#[test]
fn retyping_a_column_to_the_type_it_already_has_is_not_a_change() {
    let path = scratch("retype-noop.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INT PRIMARY KEY, n INT, s VARCHAR(20));
         INSERT INTO t VALUES (1, 10, 'a');",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    df.columns[1].db_retype = Some(tuitab::types::ColumnType::Integer);
    df.columns[2].db_retype = Some(tuitab::types::ColumnType::String);

    let plan = db_write::build_plan(&src, &df).expect("no type actually changed");
    assert!(plan.is_empty(), "{:?}", plan.stmts);
}

/// The DuckDB rebuild runs *after* the ALTERs, so the table it copies from already
/// answers to the new names. Copying by the load-time name would read a column that no
/// longer exists.
#[test]
fn duckdb_renaming_and_reordering_together_keeps_the_data() {
    let path = duckdb_fixture("rebuild-rename-reorder.duckdb");
    let (mut df, src) = load_duckdb_table_full(&path, "users").unwrap();
    let src = src.unwrap();
    df.rename_column(1, "nom").unwrap();
    df.swap_columns(1, 2).unwrap();

    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();

    let (after, _) = load_duckdb_table_full(&path, "users").unwrap();
    let names: Vec<&str> = after.columns.iter().map(|c| c.name.as_str()).collect();
    assert_eq!(names, ["id", "score", "nom", "note"]);
    let nom = after.column_index("nom").unwrap();
    assert_eq!(after.get_physical(0, nom), "ann");
    assert_eq!(after.get_physical(3, nom), "dan");
}

/// …and a column added in the same save already holds its values by then, so leaving it
/// out of the copy would commit a table full of NULLs.
#[test]
fn duckdb_adding_a_column_and_reordering_keeps_the_new_values() {
    let path = duckdb_fixture("rebuild-add-reorder.duckdb");
    let (mut df, src) = load_duckdb_table_full(&path, "users").unwrap();
    let src = src.unwrap();
    df.insert_empty_column(4, "tier").unwrap();
    let tier = df.column_index("tier").unwrap();
    df.set_cell(0, tier, "gold".to_string()).unwrap();
    df.swap_columns(1, 2).unwrap();

    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();

    let (after, _) = load_duckdb_table_full(&path, "users").unwrap();
    let tier = after.column_index("tier").unwrap();
    assert_eq!(
        after.get_physical(0, tier),
        "gold",
        "the added column must not arrive empty"
    );
    assert_eq!(after.get_physical(1, tier), "");
}

/// An index spells its column the way it was named when the index was made; a rebuild
/// replays that SQL verbatim, so a rename underneath it has to be refused rather than
/// blowing up mid-transaction.
#[test]
fn renaming_a_column_an_index_names_refuses_the_rebuild() {
    let path = scratch("rebuild-rename-index.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, a TEXT, b TEXT);
         INSERT INTO t VALUES (1, 'x', 'p');
         CREATE INDEX idx_a ON t(a);",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    df.rename_column(1, "alpha").unwrap();
    df.swap_columns(1, 2).unwrap();

    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("idx_a"), "{}", err);
    assert!(err.contains("being renamed"), "{}", err);
    assert_eq!(rows_of(&path, "SELECT a FROM t")[0][0], "x");
}

/// Renaming a column an index names is fine on its own — that path is an ALTER, and
/// SQLite rewrites the index itself.
#[test]
fn renaming_a_column_an_index_names_is_fine_without_a_rebuild() {
    let path = scratch("rename-index-ok.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, a TEXT);
         INSERT INTO t VALUES (1, 'x');
         CREATE INDEX idx_a ON t(a);",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    df.rename_column(1, "alpha").unwrap();

    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();
    assert_eq!(rows_of(&path, "SELECT alpha FROM t")[0][0], "x");
    assert!(
        rows_of(&path, "SELECT sql FROM sqlite_master WHERE name = 'idx_a'")[0][0]
            .contains("alpha")
    );
}

// ── Assumptions the create path is built on ───────────────────────────────────────

/// A blank sheet is a frame with columns and no rows. Adding the first row has to work
/// on exactly that, and it must take the dtypes from the frame rather than assume text.
#[test]
fn a_row_can_be_added_to_a_frame_that_has_columns_but_no_rows() {
    use polars::prelude::*;
    let pdf = polars::prelude::DataFrame::new(
        0,
        vec![
            Column::new("a".into(), Vec::<String>::new()),
            Column::new("n".into(), Vec::<i64>::new()),
        ],
    )
    .unwrap();
    let one = polars::prelude::DataFrame::new(
        1,
        vec![
            Series::full_null("a".into(), 1, &DataType::String).into(),
            Series::full_null("n".into(), 1, &DataType::Int64).into(),
        ],
    )
    .unwrap();

    let mut pdf = pdf;
    pdf.vstack_mut(&one)
        .expect("a zero-row frame accepts a matching row");
    assert_eq!(pdf.height(), 1);
    assert!(matches!(
        pdf.columns()[1].get(0),
        Ok(polars::prelude::AnyValue::Null)
    ));
}

/// Inserts are chunked into multi-row VALUES, so both engines have to bind a parameter
/// list wider than one row.
#[test]
fn both_engines_bind_a_multi_row_values_list() {
    let path = scratch("multirow.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch("CREATE TABLE t (a TEXT, n INTEGER)")
        .unwrap();
    let vals = [
        Val::Text("x".into()),
        Val::Int(1),
        Val::Text("y".into()),
        Val::Int(2),
        Val::Null,
        Val::Int(3),
    ];
    conn.execute(
        "INSERT INTO t (a, n) VALUES (?,?),(?,?),(?,?)",
        rusqlite::params_from_iter(vals.iter()),
    )
    .unwrap();
    drop(conn);
    assert_eq!(rows_of(&path, "SELECT a, n FROM t ORDER BY n").len(), 3);
    assert_eq!(rows_of(&path, "SELECT a FROM t ORDER BY n")[2][0], "<null>");

    let dpath = scratch("multirow.duckdb");
    let dconn = duckdb::Connection::open(&dpath).unwrap();
    dconn
        .execute_batch("CREATE TABLE t (a TEXT, n INTEGER)")
        .unwrap();
    dconn
        .execute(
            "INSERT INTO t (a, n) VALUES (?,?),(?,?),(?,?)",
            duckdb::params_from_iter(vals.iter()),
        )
        .unwrap();
    let n: i64 = dconn
        .query_row("SELECT COUNT(*) FROM t", [], |r| r.get(0))
        .unwrap();
    assert_eq!(n, 3);
}

// ── Creating a table from a sheet ─────────────────────────────────────────────────

/// A sheet the way a user would have built one: typed columns, a null, an empty string.
fn sheet_to_create() -> tuitab::data::dataframe::DataFrame {
    use tuitab::data::column::ColumnMeta;
    use tuitab::types::ColumnType;
    let pdf = polars::prelude::DataFrame::new(
        3,
        vec![
            polars::prelude::Column::new("sku".into(), vec![Some("A-1"), Some(""), None]),
            polars::prelude::Column::new("qty".into(), vec![Some(12i64), Some(0), None]),
            polars::prelude::Column::new("price".into(), vec![Some(9.99f64), Some(1.5), None]),
        ],
    )
    .unwrap();
    let mut metas = vec![
        ColumnMeta::new("sku".to_string()),
        ColumnMeta::new("qty".to_string()),
        ColumnMeta::new("price".to_string()),
    ];
    metas[1].col_type = ColumnType::Integer;
    metas[2].col_type = ColumnType::Float;
    tuitab::data::dataframe::DataFrame::from_parts(pdf, metas)
}

#[test]
fn a_created_table_declares_the_types_the_sheet_shows() {
    let path = scratch("create-types.sqlite");
    let df = sheet_to_create();
    db_write::create_table(db_write::DbKind::Sqlite, &path, "inventory", &df).unwrap();

    let decls: Vec<Vec<String>> = rows_of(
        &path,
        "SELECT name, type FROM pragma_table_info('inventory') ORDER BY cid",
    );
    assert_eq!(decls[0], ["sku", "TEXT"]);
    assert_eq!(decls[1], ["qty", "INTEGER"]);
    assert_eq!(decls[2], ["price", "REAL"]);

    // …and stores them as those types, not as display strings.
    assert_eq!(
        rows_of(
            &path,
            "SELECT typeof(qty), typeof(price) FROM inventory LIMIT 1"
        )[0],
        ["integer", "real"]
    );
}

#[test]
fn a_created_table_keeps_null_and_empty_string_apart() {
    let path = scratch("create-nulls.sqlite");
    db_write::create_table(
        db_write::DbKind::Sqlite,
        &path,
        "inventory",
        &sheet_to_create(),
    )
    .unwrap();

    let skus = rows_of(&path, "SELECT sku FROM inventory ORDER BY rowid");
    assert_eq!(skus[0][0], "A-1");
    assert_eq!(skus[1][0], "", "an empty string stays an empty string");
    assert_eq!(skus[2][0], "<null>", "a NULL stays NULL");
    assert_eq!(
        rows_of(&path, "SELECT qty FROM inventory ORDER BY rowid")[2][0],
        "<null>"
    );
}

#[test]
fn a_created_table_is_written_in_the_order_the_sheet_shows() {
    let path = scratch("create-order.sqlite");
    let mut df = sheet_to_create();
    df.row_order = std::sync::Arc::new(vec![2, 0]);

    db_write::create_table(db_write::DbKind::Sqlite, &path, "inventory", &df).unwrap();
    let skus = rows_of(&path, "SELECT sku FROM inventory ORDER BY rowid");
    assert_eq!(skus.len(), 2, "only the visible rows");
    assert_eq!(skus[0][0], "<null>");
    assert_eq!(skus[1][0], "A-1");
}

#[test]
fn duckdb_creates_a_typed_table_too() {
    let path = scratch("create.duckdb");
    db_write::create_table(
        db_write::DbKind::DuckDb,
        &path,
        "inventory",
        &sheet_to_create(),
    )
    .unwrap();

    let (df, src) = load_duckdb_table_full(&path, "inventory").unwrap();
    assert!(
        src.is_some(),
        "a created table is addressable straight away"
    );
    assert_eq!(df.visible_row_count(), 3);
    let src = src.unwrap();
    let decls: Vec<&str> = src.columns.iter().map(|c| c.decl_raw.as_str()).collect();
    assert_eq!(decls, ["VARCHAR", "BIGINT", "DOUBLE"]);
}

#[test]
fn creating_into_an_existing_database_adds_a_table_and_leaves_the_others() {
    let path = sqlite_fixture("create-alongside.sqlite");
    db_write::create_table(
        db_write::DbKind::Sqlite,
        &path,
        "inventory",
        &sheet_to_create(),
    )
    .unwrap();

    assert_eq!(rows_of(&path, "SELECT COUNT(*) FROM users")[0][0], "4");
    assert_eq!(rows_of(&path, "SELECT k FROM other")[0][0], "untouched");
    assert_eq!(rows_of(&path, "SELECT COUNT(*) FROM inventory")[0][0], "3");
}

#[test]
fn creating_over_an_existing_table_drops_it_first() {
    let path = scratch("create-replace.sqlite");
    db_write::create_table(db_write::DbKind::Sqlite, &path, "t", &sheet_to_create()).unwrap();

    let (plan, src) =
        db_write::create_plan(db_write::DbKind::Sqlite, &path, "t", &sheet_to_create()).unwrap();
    assert!(plan.create && plan.rebuild);
    assert_eq!(plan.stmts[0].display, r#"DROP TABLE "t""#);

    db_write::apply(&src, &plan).unwrap();
    assert_eq!(rows_of(&path, "SELECT COUNT(*) FROM t")[0][0], "3");
}

#[test]
fn a_cell_the_column_type_cannot_hold_stops_the_create_and_leaves_no_file() {
    use tuitab::data::column::ColumnMeta;
    let path = scratch("create-badtype.sqlite");
    let pdf = polars::prelude::DataFrame::new(
        2,
        vec![polars::prelude::Column::new(
            "qty".into(),
            vec!["12", "n/a"],
        )],
    )
    .unwrap();
    let mut metas = vec![ColumnMeta::new("qty".to_string())];
    metas[0].col_type = tuitab::types::ColumnType::Integer;
    let df = tuitab::data::dataframe::DataFrame::from_parts(pdf, metas);

    let err = db_write::create_table(db_write::DbKind::Sqlite, &path, "t", &df)
        .unwrap_err()
        .to_string();
    assert!(err.contains("qty"), "{}", err);
    assert!(
        err.contains("row 2"),
        "names the row the user sees: {}",
        err
    );
    assert!(
        !path.exists(),
        "a failed create must not leave a file behind"
    );
}

#[test]
fn inserts_are_chunked_but_every_row_lands() {
    use tuitab::data::column::ColumnMeta;
    let path = scratch("create-chunked.sqlite");
    let n = 1200;
    let pdf = polars::prelude::DataFrame::new(
        n,
        vec![
            polars::prelude::Column::new("a".into(), (0..n as i64).collect::<Vec<_>>()),
            polars::prelude::Column::new("b".into(), vec!["x"; n]),
        ],
    )
    .unwrap();
    let mut metas = vec![
        ColumnMeta::new("a".to_string()),
        ColumnMeta::new("b".to_string()),
    ];
    metas[0].col_type = tuitab::types::ColumnType::Integer;
    let df = tuitab::data::dataframe::DataFrame::from_parts(pdf, metas);

    let (plan, src) = db_write::create_plan(db_write::DbKind::Sqlite, &path, "t", &df).unwrap();
    assert_eq!(plan.inserts, n);
    assert!(plan.stmts.len() < 20, "{} statements", plan.stmts.len());
    db_write::apply(&src, &plan).unwrap();
    assert_eq!(rows_of(&path, "SELECT COUNT(*) FROM t")[0][0], "1200");
}

#[test]
fn a_created_sqlite_table_can_be_reopened_and_edited() {
    let path = scratch("create-roundtrip.sqlite");
    db_write::create_table(
        db_write::DbKind::Sqlite,
        &path,
        "inventory",
        &sheet_to_create(),
    )
    .unwrap();

    let (mut df, src) = load_sqlite_table_full(&path, "inventory").unwrap();
    let src = src.expect("a created table has rowid identity");
    df.set_cell(0, 0, "B-2".to_string()).unwrap();
    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();

    assert_eq!(
        rows_of(&path, "SELECT sku FROM inventory ORDER BY rowid")[0][0],
        "B-2"
    );
}

#[test]
fn a_sheet_with_no_columns_cannot_become_a_table() {
    let path = scratch("create-nocols.sqlite");
    let df = tuitab::data::dataframe::DataFrame::empty();
    let err = db_write::create_table(db_write::DbKind::Sqlite, &path, "t", &df)
        .unwrap_err()
        .to_string();
    assert!(err.contains("no columns"), "{}", err);
}

/// The identifier scan must look at identifiers, not at any text that happens to
/// contain a keyword.
#[test]
fn a_column_named_after_a_keyword_does_not_block_a_rebuild() {
    let path = scratch("rebuild-keyword-name.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, checked TEXT, generated_at TEXT);
         INSERT INTO t VALUES (1, 'yes', '2020');",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    df.swap_columns(1, 2).unwrap();

    let plan = db_write::build_plan(&src, &df).expect("'checked' is not a CHECK constraint");
    assert!(plan.rebuild);
    db_write::apply(&src, &plan).unwrap();
    assert_eq!(
        rows_of(
            &path,
            "SELECT name FROM pragma_table_info('t') ORDER BY cid"
        )
        .into_iter()
        .map(|r| r[0].clone())
        .collect::<Vec<_>>(),
        ["id", "generated_at", "checked"]
    );
}

// ── Declared types reach the frame ────────────────────────────────────────────────

/// Without this a database column is a string column, and `score > 100` is a
/// comparison polars refuses outright — numeric filtering over a database does not
/// work at all.
#[test]
fn a_declared_integer_column_loads_as_an_integer() {
    let path = sqlite_fixture("typed-int.sqlite");
    let (df, _) = open_sqlite(&path);
    let types: Vec<_> = df.columns.iter().map(|c| c.col_type.name()).collect();
    assert_eq!(types, ["integer", "string", "integer", "string"]);
}

#[test]
fn a_numeric_filter_over_a_database_compares_numerically() {
    use tuitab::data::filter::{Clause, Operand, PredOp, Predicate};
    let path = scratch("typed-filter.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, score INTEGER);
         INSERT INTO t VALUES (1, 20), (2, 1000), (3, 3);",
    )
    .unwrap();
    drop(conn);

    let (df, _) = load_sqlite_table_full(&path, "t").unwrap();
    let matched = tuitab::data::filter::matching_rows(
        &df,
        &[Clause::One(Predicate {
            col: "score".to_string(),
            op: PredOp::Gt,
            value: Operand::Literal(tuitab::data::expression::Value::Number(100.0)),
        })],
    )
    .expect("a declared INTEGER column compares as a number");
    // Lexically, "20" and "3" both beat "100"; numerically only 1000 does.
    assert_eq!(matched, vec![1]);
}

/// SQLite lets an INTEGER column hold text. Such a column is still worth reading, so
/// the cast is best-effort per column rather than fatal.
#[test]
fn a_declared_integer_column_holding_text_stays_text() {
    let path = scratch("typed-mixed.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, n INTEGER);
         INSERT INTO t VALUES (1, 10);
         INSERT INTO t VALUES (2, 'not a number');",
    )
    .unwrap();
    drop(conn);

    let (df, src) = load_sqlite_table_full(&path, "t").unwrap();
    assert!(src.is_some(), "the table is still readable");
    let n = df.column_index("n").unwrap();
    assert_eq!(df.columns[n].col_type.name(), "string");
    assert_eq!(df.get_physical(1, n), "not a number");
}

/// SQLite has no boolean type: a BOOLEAN column stores the integers 1 and 0. Casting
/// it to a polars Boolean would render `true` where the re-read database says `1`, and
/// every drift check would fire on rows nobody touched.
#[test]
fn a_declared_boolean_column_stays_text_and_round_trips() {
    let path = scratch("typed-bool.sqlite");
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, ok BOOLEAN, name TEXT);
         INSERT INTO t VALUES (1, 1, 'a'), (2, 0, 'b');",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_sqlite_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    let ok = df.column_index("ok").unwrap();
    assert_eq!(df.columns[ok].col_type.name(), "string");
    assert_eq!(df.get_physical(0, ok), "1");

    // Editing an unrelated column must not report drift on the boolean one.
    df.set_cell(0, 2, "A".to_string()).unwrap();
    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap()).unwrap();
    assert_eq!(rows_of(&path, "SELECT ok FROM t ORDER BY id")[0][0], "1");
}

/// DuckDB renders a DOUBLE 1.0 as "1.0" and Rust's f64::to_string gives "1". The drift
/// check has to compare the number, not its spelling, or every whole-numbered double
/// reports as changed.
#[test]
fn a_whole_numbered_double_does_not_look_like_drift_in_duckdb() {
    let path = scratch("typed-double.duckdb");
    let conn = duckdb::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE t (id INTEGER PRIMARY KEY, r DOUBLE, big DOUBLE, name TEXT);
         INSERT INTO t VALUES (1, 1.0, 1e20, 'a'), (2, 2000.5, 0.1, 'b');",
    )
    .unwrap();
    drop(conn);

    let (mut df, src) = load_duckdb_table_full(&path, "t").unwrap();
    let src = src.unwrap();
    let name = df.column_index("name").unwrap();
    df.set_cell(0, name, "A".to_string()).unwrap();

    db_write::apply(&src, &db_write::build_plan(&src, &df).unwrap())
        .expect("untouched doubles must not read as drift");

    let (after, _) = load_duckdb_table_full(&path, "t").unwrap();
    assert_eq!(after.get_physical(0, name), "A");
    assert_eq!(after.get_physical(1, 1), "2000.5");
}

/// The value the user typed has to survive being wrong: a plain cast would turn it
/// into a silent NULL, losing what they can see on screen.
#[test]
fn a_value_the_column_cannot_hold_is_kept_and_refused_at_save() {
    let path = sqlite_fixture("typed-badedit.sqlite");
    let (mut df, src) = open_sqlite(&path);
    let score = df.column_index("score").unwrap();
    df.set_cell(0, score, "abc".to_string()).unwrap();

    assert_eq!(df.get_physical(0, score), "abc", "not silently nulled");
    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("not an integer"), "{}", err);
}

// ── Views ─────────────────────────────────────────────────────────────────────────

#[test]
fn a_view_is_listed_and_readable_but_not_writable() {
    for (name, duck) in [("views.sqlite", false), ("views.duckdb", true)] {
        let path = scratch(name);
        let ddl = "CREATE TABLE t (id INTEGER PRIMARY KEY, n INTEGER);
                   INSERT INTO t VALUES (1, 10), (2, 30);
                   CREATE VIEW big AS SELECT id, n FROM t WHERE n > 20;";
        if duck {
            duckdb::Connection::open(&path)
                .unwrap()
                .execute_batch(ddl)
                .unwrap();
        } else {
            rusqlite::Connection::open(&path)
                .unwrap()
                .execute_batch(ddl)
                .unwrap();
        }

        let listed = tuitab::data::io::db_containers(&path).unwrap();
        let view = listed
            .iter()
            .find(|c| c.name == "big")
            .unwrap_or_else(|| panic!("{}: the view is missing from the listing", name));
        assert!(view.view, "{}", name);
        assert_eq!(view.rows, None, "{}: a view must not be counted", name);
        assert_eq!(view.columns, 2, "{}", name);
        assert!(
            view.sql.as_deref().unwrap_or("").contains("CREATE"),
            "{}",
            name
        );

        let (df, src) = if duck {
            load_duckdb_table_full(&path, "big").unwrap()
        } else {
            load_sqlite_table_full(&path, "big").unwrap()
        };
        assert_eq!(df.visible_row_count(), 1, "{}: the view reads", name);
        assert!(src.is_none(), "{}: a view has no row identity", name);
    }
}

// ── Binary columns ────────────────────────────────────────────────────────────────

/// `docs`: one text column and one BLOB, so an edit to each can be planned separately.
fn sqlite_blob_fixture(name: &str) -> PathBuf {
    let path = scratch(name);
    let conn = rusqlite::Connection::open(&path).unwrap();
    conn.execute_batch(
        "CREATE TABLE docs (id INTEGER PRIMARY KEY, label TEXT, body BLOB);
         INSERT INTO docs VALUES (1, 'first',  x'0102030405');
         INSERT INTO docs VALUES (2, 'second', x'ff00ff');",
    )
    .unwrap();
    path
}

#[test]
fn a_blob_reads_as_its_size_so_a_swap_of_another_size_is_noticed() {
    let path = sqlite_blob_fixture("blob-size.sqlite");
    let (df, src) = load_sqlite_table_full(&path, "docs").unwrap();
    let src = src.unwrap();
    let body = df.column_index("body").unwrap();
    assert_eq!(df.get_physical(0, body), "[BLOB 5 bytes]");

    // Edit the text column, then let something else replace the blob underneath.
    let mut df = df;
    let label = df.column_index("label").unwrap();
    df.set_cell(0, label, "renamed".to_string()).unwrap();
    let plan = db_write::build_plan(&src, &df).unwrap();
    rusqlite::Connection::open(&path)
        .unwrap()
        .execute_batch("UPDATE docs SET body = x'0102030405060708' WHERE id = 1")
        .unwrap();

    let err = db_write::apply(&src, &plan).unwrap_err().to_string();
    assert!(err.contains("changed since it was opened"), "{}", err);
}

#[test]
fn editing_a_blob_column_is_refused_while_its_neighbour_still_saves() {
    let path = sqlite_blob_fixture("blob-edit.sqlite");
    let (mut df, src) = load_sqlite_table_full(&path, "docs").unwrap();
    let src = src.unwrap();
    let label = df.column_index("label").unwrap();
    let body = df.column_index("body").unwrap();

    // The neighbour on its own is an ordinary save.
    df.set_cell(0, label, "renamed".to_string()).unwrap();
    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.updates, 1);
    db_write::apply(&src, &plan).unwrap();

    // The blob is a rendering; writing it back would store the description.
    let (mut df, src) = load_sqlite_table_full(&path, "docs").unwrap();
    let src = src.unwrap();
    df.set_cell(0, body, "hello".to_string()).unwrap();
    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("binary data"), "{}", err);

    // And the bytes are still the bytes.
    assert_eq!(
        rows_of(&path, "SELECT hex(body), label FROM docs WHERE id = 1"),
        vec![vec!["0102030405".to_string(), "renamed".to_string()]]
    );
}

#[test]
fn a_duckdb_blob_column_does_not_make_every_save_look_like_drift() {
    let path = scratch("blob-duck.duckdb");
    {
        let conn = duckdb::Connection::open(&path).unwrap();
        conn.execute_batch(
            "CREATE TABLE docs (id INTEGER PRIMARY KEY, label TEXT, body BLOB);
             INSERT INTO docs VALUES (1, 'first', '\\x01\\x02\\x03'::BLOB);
             INSERT INTO docs VALUES (2, 'second', '\\xff\\x00'::BLOB);",
        )
        .unwrap();
    }
    let (mut df, src) = load_duckdb_table_full(&path, "docs").unwrap();
    let src = src.unwrap();
    let label = df.column_index("label").unwrap();
    df.set_cell(0, label, "renamed".to_string()).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    // The drift check reads blobs back through a different path than the loader did; if
    // the two disagree, every save on a table with a BLOB column fails here.
    db_write::apply(&src, &plan).unwrap();
}

// ── The rest of what the plan left open ───────────────────────────────────────────

#[test]
fn a_plan_of_nothing_but_inserts_still_checks_the_table_it_was_built_against() {
    let path = sqlite_fixture("insert-shape.sqlite");
    let (mut df, src) = open_sqlite(&path);
    df.insert_empty_row(4).unwrap();
    let name = df.column_index("name").unwrap();
    df.set_cell(4, name, "eve".to_string()).unwrap();

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.inserts, 1);
    assert_eq!(plan.updates, 0);

    // Somebody else changes the shape of the table between the plan and the write.
    rusqlite::Connection::open(&path)
        .unwrap()
        .execute_batch("ALTER TABLE users DROP COLUMN note")
        .unwrap();

    let err = db_write::apply(&src, &plan).unwrap_err().to_string();
    assert!(err.contains("columns of 'users' changed"), "{}", err);
    assert_eq!(
        rows_of(&path, "SELECT COUNT(*) FROM users"),
        vec![vec!["4".to_string()]],
        "nothing was inserted"
    );
}

#[test]
fn the_drift_message_does_not_blame_tuitab() {
    let path = sqlite_fixture("drift-wording.sqlite");
    let (mut df, src) = open_sqlite(&path);
    let name = df.column_index("name").unwrap();
    df.set_cell(0, name, "ANN".to_string()).unwrap();
    let plan = db_write::build_plan(&src, &df).unwrap();

    rusqlite::Connection::open(&path)
        .unwrap()
        .execute_batch("UPDATE users SET name = 'someone else' WHERE id = 1")
        .unwrap();

    let err = db_write::apply(&src, &plan).unwrap_err().to_string();
    assert!(err.contains("something else has written to it"), "{}", err);
    assert!(!err.contains("tuitab"), "{}", err);
}

#[test]
fn reordering_a_duckdb_table_with_a_generated_column_is_refused() {
    let path = scratch("duck-generated.duckdb");
    {
        let conn = duckdb::Connection::open(&path).unwrap();
        conn.execute_batch(
            "CREATE TABLE users (id INTEGER, a INTEGER, b INTEGER GENERATED ALWAYS AS (a * 2));
             INSERT INTO users (id, a) VALUES (1, 10), (2, 20);",
        )
        .unwrap();
    }
    let (mut df, src) = load_duckdb_table_full(&path, "users").unwrap();
    let src = src.unwrap();
    df.swap_columns(0, 1).unwrap();

    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("GENERATED"), "{}", err);
}

#[test]
fn a_find_and_replace_cancels_a_type_the_user_had_assigned() {
    let path = sqlite_fixture("retype-then-replace.sqlite");
    let (mut df, src) = open_sqlite(&path);
    // `t` → Integer on a TEXT column, then `zr` puts it back to text.
    df.set_column_type(3, tuitab::types::ColumnType::String)
        .unwrap();
    df.columns[3].db_retype = Some(tuitab::types::ColumnType::Integer);
    df.col_replace(3, "hi", "there", true).unwrap();

    assert!(df.columns[3].db_retype.is_none());
    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.schema, 0, "{:?}", shown_sql(&plan));
    assert!(!plan.rebuild);
}

#[test]
fn asking_whether_a_table_exists_does_not_read_the_database() {
    let path = sqlite_fixture("table-exists.sqlite");
    assert!(db_write::table_exists(
        db_write::DbKind::Sqlite,
        &path,
        "users"
    ));
    assert!(db_write::table_exists(
        db_write::DbKind::Sqlite,
        &path,
        "other"
    ));
    assert!(!db_write::table_exists(
        db_write::DbKind::Sqlite,
        &path,
        "missing"
    ));
    // A view counts: creating a table over one would fail either way.
    rusqlite::Connection::open(&path)
        .unwrap()
        .execute_batch("CREATE VIEW peek AS SELECT id FROM users")
        .unwrap();
    assert!(db_write::table_exists(
        db_write::DbKind::Sqlite,
        &path,
        "peek"
    ));

    let missing = scratch("table-exists-absent.sqlite");
    assert!(!db_write::table_exists(
        db_write::DbKind::Sqlite,
        &missing,
        "users"
    ));
}

#[test]
fn a_plan_too_large_to_show_says_how_many_it_is_hiding() {
    let path = scratch("display-cap.sqlite");
    {
        let conn = rusqlite::Connection::open(&path).unwrap();
        conn.execute_batch("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
            .unwrap();
        let tx = conn.unchecked_transaction().unwrap();
        for i in 0..2500 {
            tx.execute(
                "INSERT INTO users (id, name) VALUES (?1, ?2)",
                rusqlite::params![i, format!("n{}", i)],
            )
            .unwrap();
        }
        tx.commit().unwrap();
    }
    let (mut df, src) = open_sqlite(&path);
    let name = df.column_index("name").unwrap();
    // Every row a different value, so each UPDATE is its own statement.
    for row in 0..2500 {
        df.set_cell(row, name, format!("x{}", row)).unwrap();
    }

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.stmts.len(), 2500);
    assert_eq!(plan.hidden_stmts(), 500);
    assert!(!plan.stmts[1999].display.is_empty());
    assert!(plan.stmts[2000].display.is_empty());
    // The statements it cannot show still run.
    db_write::apply(&src, &plan).unwrap();
    assert_eq!(
        rows_of(&path, "SELECT name FROM users WHERE id = 2400"),
        vec![vec!["x2400".to_string()]]
    );
}

#[test]
fn a_duckdb_file_named_db_is_opened_as_duckdb() {
    let path = scratch("disguised.db");
    {
        let conn = duckdb::Connection::open(&path).unwrap();
        conn.execute_batch("CREATE TABLE users (id INTEGER, name TEXT); INSERT INTO users VALUES (1, 'ann'); CHECKPOINT;")
            .unwrap();
    }
    assert_eq!(
        db_write::kind_for_path(&path),
        db_write::DbKind::DuckDb,
        "the header, not the extension"
    );
    let listed = tuitab::data::io::db_containers(&path).unwrap();
    assert_eq!(listed.len(), 1);
    assert_eq!(listed[0].name, "users");

    // And a SQLite file keeps being SQLite whatever it is called.
    let sqlite = scratch("disguised.duckdb");
    rusqlite::Connection::open(&sqlite)
        .unwrap()
        .execute_batch("CREATE TABLE t (a INTEGER)")
        .unwrap();
    assert_eq!(db_write::kind_for_path(&sqlite), db_write::DbKind::Sqlite);

    // A file that does not exist yet has only its name to go on.
    let fresh = scratch("brand-new.db");
    assert_eq!(db_write::kind_for_path(&fresh), db_write::DbKind::Sqlite);
}

#[test]
fn a_save_waits_for_another_writer_instead_of_failing() {
    let path = sqlite_fixture("busy-wait.sqlite");
    let (mut df, src) = open_sqlite(&path);
    let name = df.column_index("name").unwrap();
    df.set_cell(0, name, "ANN".to_string()).unwrap();
    let plan = db_write::build_plan(&src, &df).unwrap();

    // Another connection holds the write lock for a moment.
    let blocker = rusqlite::Connection::open(&path).unwrap();
    blocker.execute_batch("BEGIN IMMEDIATE").unwrap();
    let handle = std::thread::spawn(move || {
        std::thread::sleep(std::time::Duration::from_millis(400));
        blocker.execute_batch("COMMIT").unwrap();
    });

    db_write::apply(&src, &plan).unwrap();
    handle.join().unwrap();
    assert_eq!(
        rows_of(&path, "SELECT name FROM users WHERE id = 1"),
        vec![vec!["ANN".to_string()]]
    );
}

/// DuckDB lets one process at a time near a file and has no timeout to wait with, so
/// the only thing to get right is the sentence.  It takes a second process to provoke:
/// within one process DuckDB shares the open database rather than refusing.
#[test]
fn a_duckdb_file_held_by_another_process_says_so_plainly() {
    let path = scratch("duck-second-writer.duckdb");
    if std::env::var("TUITAB_DUCK_LOCK_CHILD").is_ok() {
        let _held = duckdb::Connection::open(&path).unwrap();
        std::thread::sleep(std::time::Duration::from_secs(3));
        return;
    }
    {
        let conn = duckdb::Connection::open(&path).unwrap();
        conn.execute_batch(
            "CREATE TABLE users (id INTEGER, name TEXT); INSERT INTO users VALUES (1, 'ann'); CHECKPOINT;",
        )
        .unwrap();
    }

    let mut child = std::process::Command::new(std::env::current_exe().unwrap())
        .args([
            "a_duckdb_file_held_by_another_process_says_so_plainly",
            "--exact",
        ])
        .env("TUITAB_DUCK_LOCK_CHILD", "1")
        .stdout(std::process::Stdio::null())
        .spawn()
        .unwrap();
    std::thread::sleep(std::time::Duration::from_millis(1200));

    let err = match load_duckdb_table_full(&path, "users") {
        Ok(_) => panic!("the child holds the file; the open should have been refused"),
        Err(e) => e.to_string(),
    };
    let _ = child.wait();

    assert!(err.contains("open in another program"), "{}", err);
    assert!(
        !err.contains("Conflicting lock"),
        "raw engine text: {}",
        err
    );
}

// ── Types the user assigns with `t` ───────────────────────────────────────────────

/// SQLite stores BOOLEAN as the integers 1 and 0, so a Boolean column renders `true`
/// where the table says `1`.  That is a difference in clothing, not in value.
#[test]
fn showing_a_boolean_column_as_boolean_plans_nothing() {
    let path = scratch("retype-bool.sqlite");
    {
        let conn = rusqlite::Connection::open(&path).unwrap();
        conn.execute_batch(
            "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, active BOOLEAN);
             INSERT INTO users VALUES (1, 'ann', 1), (2, 'bob', 0), (3, 'cara', 1);",
        )
        .unwrap();
    }
    let (mut df, src) = open_sqlite(&path);
    let active = df.column_index("active").unwrap();
    df.set_column_type(active, tuitab::types::ColumnType::Boolean)
        .unwrap();
    df.columns[active].db_retype = Some(tuitab::types::ColumnType::Boolean);

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert!(plan.is_empty(), "{:?}", shown_sql(&plan));

    // …and a real edit in that column still reaches the table.
    df.set_cell(0, active, "false".to_string()).unwrap();
    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.updates, 1, "{:?}", shown_sql(&plan));
    db_write::apply(&src, &plan).unwrap();
    assert_eq!(
        rows_of(&path, "SELECT active FROM users ORDER BY id"),
        vec![
            vec!["0".to_string()],
            vec!["0".to_string()],
            vec!["1".to_string()]
        ]
    );
}

/// A Date read out of a TEXT column that holds a time is a *reading*: the time is gone
/// from the frame, and writing the frame back would take it out of the table too.
#[test]
fn showing_a_text_column_of_timestamps_as_a_date_writes_nothing() {
    let path = scratch("retype-date-lossy.sqlite");
    {
        let conn = rusqlite::Connection::open(&path).unwrap();
        conn.execute_batch(
            "CREATE TABLE users (id INTEGER PRIMARY KEY, seen TEXT);
             INSERT INTO users VALUES (1, '2024-01-01 09:30:00'), (2, '2024-02-03 17:05:00');",
        )
        .unwrap();
    }
    let (mut df, src) = open_sqlite(&path);
    let seen = df.column_index("seen").unwrap();
    df.set_column_type(seen, tuitab::types::ColumnType::Date)
        .unwrap();
    df.columns[seen].db_retype = Some(tuitab::types::ColumnType::Date);
    assert_eq!(
        df.get_physical(0, seen),
        "2024-01-01",
        "the frame truncated"
    );

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert!(plan.is_empty(), "{:?}", shown_sql(&plan));
    assert_eq!(plan.warnings.len(), 1, "{:?}", plan.warnings);
    assert!(plan.warnings[0].contains("will not be written"));

    db_write::apply(&src, &plan).unwrap();
    assert_eq!(
        rows_of(&path, "SELECT seen FROM users ORDER BY id"),
        vec![
            vec!["2024-01-01 09:30:00".to_string()],
            vec!["2024-02-03 17:05:00".to_string()]
        ],
        "the time is still in the table"
    );
}

/// The lossless half of the same rule: plain dates round-trip, so nothing is planned and
/// nothing is refused either.
#[test]
fn showing_a_text_column_of_plain_dates_as_a_date_plans_nothing() {
    let path = scratch("retype-date-clean.sqlite");
    {
        let conn = rusqlite::Connection::open(&path).unwrap();
        conn.execute_batch(
            "CREATE TABLE users (id INTEGER PRIMARY KEY, seen TEXT);
             INSERT INTO users VALUES (1, '2024-01-01'), (2, '2024-02-03');",
        )
        .unwrap();
    }
    let (mut df, src) = open_sqlite(&path);
    let seen = df.column_index("seen").unwrap();
    df.set_column_type(seen, tuitab::types::ColumnType::Date)
        .unwrap();
    df.columns[seen].db_retype = Some(tuitab::types::ColumnType::Date);

    let plan = db_write::build_plan(&src, &df).unwrap();
    assert!(plan.is_empty(), "{:?}", shown_sql(&plan));
    assert!(plan.warnings.is_empty(), "{:?}", plan.warnings);
}

// ── Replacing a table ─────────────────────────────────────────────────────────────

#[test]
fn replacing_a_table_says_what_the_drop_takes_with_it() {
    let path = sqlite_fixture("replace-warnings.sqlite");
    rusqlite::Connection::open(&path)
        .unwrap()
        .execute_batch(
            "CREATE INDEX ix_users_name ON users(name);
             CREATE TRIGGER audit AFTER INSERT ON users BEGIN
                 INSERT INTO other VALUES ('added');
             END;
             CREATE VIEW big AS SELECT id FROM users WHERE score > 15;",
        )
        .unwrap();

    let (df, _) = open_sqlite(&path);
    let (plan, _) = db_write::create_plan(db_write::DbKind::Sqlite, &path, "users", &df).unwrap();
    assert!(plan.rebuild);
    let said = plan.warnings.join(" | ");
    assert!(
        said.contains("index 'ix_users_name' will be lost"),
        "{}",
        said
    );
    assert!(said.contains("trigger 'audit' will be lost"), "{}", said);
    assert!(
        said.contains("view 'big' is built on this table"),
        "{}",
        said
    );
}

#[test]
fn replacing_a_table_something_points_at_is_refused() {
    let path = scratch("replace-fk.sqlite");
    {
        let conn = rusqlite::Connection::open(&path).unwrap();
        conn.execute_batch(
            "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT);
             INSERT INTO users VALUES (1, 'ann');
             CREATE TABLE orders (id INTEGER PRIMARY KEY, who INTEGER REFERENCES users(id));",
        )
        .unwrap();
    }
    let (df, _) = load_sqlite_table_full(&path, "users").unwrap();
    let err = match db_write::create_plan(db_write::DbKind::Sqlite, &path, "users", &df) {
        Ok(_) => panic!("replacing a table an order points at should be refused"),
        Err(e) => e.to_string(),
    };
    assert!(err.contains("foreign key into 'users'"), "{}", err);
}

// ── Reading, when reading goes wrong ──────────────────────────────────────────────

/// A table whose catalogue says nothing about it is an error, not a table of
/// unconstrained text columns: inventing the metadata switches off every check built on
/// it — NOT NULL, DEFAULT, the generated-column skip and the declared typing.
#[test]
fn a_table_that_is_not_there_does_not_come_back_as_untyped_columns() {
    let path = sqlite_fixture("meta-missing.sqlite");
    let err = match load_sqlite_table_full(&path, "nosuchtable") {
        Ok(_) => panic!("a table that does not exist should not load"),
        Err(e) => e.to_string(),
    };
    assert!(!err.is_empty());

    // And the real table still comes back with its constraints intact.
    let (_, src) = load_sqlite_table_full(&path, "users").unwrap();
    let src = src.unwrap();
    assert!(
        src.column("id").unwrap().pk,
        "the primary key was invented away"
    );
}

/// A locked database is not a view.  The narrow fallback is what keeps a transient
/// failure from turning a writable table into a read-only sheet.
#[test]
fn a_failure_that_is_not_about_rowid_is_reported_as_itself() {
    let path = scratch("not-a-database.sqlite");
    std::fs::write(&path, b"this is not a database at all, not even close").unwrap();
    match load_sqlite_table_full(&path, "users") {
        Ok((_, src)) => panic!(
            "a broken file must not read as a source: {:?}",
            src.is_none()
        ),
        Err(e) => {
            let text = e.to_string();
            assert!(!text.contains("view"), "{}", text);
        }
    }
}

// ── DuckDB's rebuild preflight, brought level with SQLite's ───────────────────────

#[test]
fn a_duckdb_table_a_view_is_built_on_refuses_the_rebuild() {
    let path = duckdb_fixture("duck-view-rebuild.duckdb");
    {
        let conn = duckdb::Connection::open(&path).unwrap();
        conn.execute_batch("CREATE VIEW big AS SELECT id, name FROM users WHERE score > 15")
            .unwrap();
    }
    let (mut df, src) = load_duckdb_table_full(&path, "users").unwrap();
    let src = src.unwrap();
    df.swap_columns(1, 2).unwrap();

    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("view 'big' is built on it"), "{}", err);
}

/// A missing table whose *name* contains `rowid` reports itself as missing.
///
/// Both engines name the table in the message — `no such table: rowid_map`,
/// `Table with name rowid_map does not exist!` — so a fallback keyed on the bare word
/// `rowid` matches it.  Today the second read fails the same way and the error still
/// comes out right, so this pins the property rather than catching a live defect: it is
/// what would break if the fallback ever swallowed the error it retries.
#[test]
fn a_missing_table_named_after_rowid_is_not_mistaken_for_a_view() {
    let sqlite = sqlite_fixture("rowid-name.sqlite");
    match load_sqlite_table_full(&sqlite, "rowid_map") {
        Ok(_) => panic!("a table that does not exist must not load"),
        Err(e) => assert!(e.to_string().contains("no such table"), "{}", e),
    }

    let duck = duckdb_fixture("rowid-name.duckdb");
    match load_duckdb_table_full(&duck, "rowid_map") {
        Ok(_) => panic!("a table that does not exist must not load"),
        Err(e) => assert!(
            e.to_string().to_lowercase().contains("does not exist"),
            "{}",
            e
        ),
    }
}

/// DuckDB has no flag for a computed column, so it is read out of the stored DDL — and
/// once it is read, the ordinary per-column refusals start working on DuckDB too.
#[test]
fn a_duckdb_generated_column_is_read_only_like_sqlites() {
    let path = scratch("duck-generated-col.duckdb");
    {
        let conn = duckdb::Connection::open(&path).unwrap();
        conn.execute_batch(
            "CREATE TABLE users (id INTEGER, name TEXT, \"twice it\" INTEGER \
                 GENERATED ALWAYS AS (id * 2), tier TEXT DEFAULT 'basic');
             INSERT INTO users (id, name) VALUES (1, 'ann'), (2, 'bob');",
        )
        .unwrap();
    }
    let (mut df, src) = load_duckdb_table_full(&path, "users").unwrap();
    let src = src.unwrap();
    assert!(
        src.column("twice it").unwrap().generated,
        "the generated column was not recognised"
    );
    assert!(
        !src.column("tier").unwrap().generated,
        "a DEFAULT was mistaken for a generated column"
    );

    // Editing it is refused by name…
    let twice = df.column_index("twice it").unwrap();
    df.set_cell(0, twice, "99".to_string()).unwrap();
    let err = db_write::build_plan(&src, &df).unwrap_err().to_string();
    assert!(err.contains("generated by the database"), "{}", err);

    // …and a new row does not try to write it.
    let (mut df, src) = load_duckdb_table_full(&path, "users").unwrap();
    let src = src.unwrap();
    let name = df.column_index("name").unwrap();
    let id = df.column_index("id").unwrap();
    df.insert_empty_row(2).unwrap();
    df.set_cell(2, id, "3".to_string()).unwrap();
    df.set_cell(2, name, "cara".to_string()).unwrap();
    let plan = db_write::build_plan(&src, &df).unwrap();
    assert_eq!(plan.inserts, 1);
    assert!(
        !shown_sql(&plan).iter().any(|s| s.contains("twice it")),
        "{:?}",
        shown_sql(&plan)
    );
    db_write::apply(&src, &plan).unwrap();
}