package executor

import (
	"strings"
	"testing"

	"github.com/danfragoso/pizzasql-next/pkg/storage"
)

// transactional DDL: CREATE TABLE and its rows are visible to the transaction
// that issued them, invisible to other sessions, and discarded by ROLLBACK.
func TestTransactionalDDLCreateTableRollback(t *testing.T) {
	_, schema, table := newTestDB(t)
	writer := newExec(schema, table)
	observer := newExec(schema, table)

	execMust(t, writer, "BEGIN")
	execMust(t, writer, "CREATE TABLE staged (id INTEGER PRIMARY KEY, v TEXT)")
	execMust(t, writer, "INSERT INTO staged VALUES (1, 'one')")

	// Read-your-writes inside the transaction.
	if res := execMust(t, writer, "SELECT v FROM staged WHERE id = 1"); res.RowCount != 1 || res.Rows[0][0] != "one" {
		t.Fatalf("staged table read = %v", res.Rows)
	}
	// Another session must not see the staged table.
	if _, err := execSQL(observer, "SELECT * FROM staged"); err == nil {
		t.Fatal("observer saw a table created in another session's uncommitted transaction")
	}

	execMust(t, writer, "ROLLBACK")

	if _, err := execSQL(writer, "SELECT * FROM staged"); err == nil {
		t.Fatal("staged table survived ROLLBACK")
	}
	if schema.TableExists("staged") {
		t.Fatal("durable catalog contains a rolled-back table")
	}
	if _, err := schema.GetSchema("staged"); err == nil {
		t.Fatal("durable catalog still resolves the rolled-back table")
	}
	// DDL and its rows are one unit: the staged INSERT left no durable rows.
	if rows, err := table.Select("staged", nil); err == nil && len(rows) > 0 {
		t.Fatalf("rolled-back table left %d durable rows", len(rows))
	}
}

// transactional DDL: DROP TABLE is undone by ROLLBACK, rows included.
func TestTransactionalDDLDropTableRollback(t *testing.T) {
	_, schema, table := newTestDB(t)
	e := newExec(schema, table)
	execMust(t, e, "CREATE TABLE ledger (id INTEGER PRIMARY KEY, v TEXT)")
	execMust(t, e, "INSERT INTO ledger VALUES (1, 'a'), (2, 'b')")

	execMust(t, e, "BEGIN")
	execMust(t, e, "DROP TABLE ledger")
	if _, err := execSQL(e, "SELECT * FROM ledger"); err == nil {
		t.Fatal("dropped table is still visible to the transaction")
	}
	execMust(t, e, "ROLLBACK")

	res := execMust(t, e, "SELECT id, v FROM ledger ORDER BY id")
	if res.RowCount != 2 {
		t.Fatalf("rows after rollback = %v", res.Rows)
	}
	// The catalog and the rows must both be intact after the rollback.
	if !schema.TableExists("ledger") {
		t.Fatal("durable catalog lost the table")
	}
	row, err := table.GetByPK("ledger", "1")
	if err != nil || row["v"] != "a" {
		t.Fatalf("durable row after rollback: %v %v", row, err)
	}
}

// transactional DDL: ALTER TABLE variants are undone by ROLLBACK.
func TestTransactionalDDLAlterRollback(t *testing.T) {
	_, schema, table := newTestDB(t)
	e := newExec(schema, table)
	execMust(t, e, "CREATE TABLE people (id INTEGER PRIMARY KEY, name TEXT)")
	execMust(t, e, "INSERT INTO people VALUES (1, 'ada')")

	execMust(t, e, "BEGIN")
	execMust(t, e, "ALTER TABLE people ADD COLUMN age INTEGER DEFAULT 30")
	// Read-your-writes: the new column exists for this session and carries the
	// backfilled default.
	res := execMust(t, e, "SELECT age FROM people WHERE id = 1")
	if res.RowCount != 1 || res.Rows[0][0] != int64(30) {
		t.Fatalf("staged ALTER ADD COLUMN read = %v", res.Rows)
	}
	execMust(t, e, "ALTER TABLE people RENAME COLUMN name TO full_name")
	if res := execMust(t, e, "SELECT full_name FROM people"); res.Rows[0][0] != "ada" {
		t.Fatalf("staged RENAME COLUMN read = %v", res.Rows)
	}
	execMust(t, e, "ALTER TABLE people RENAME TO humans")
	if res := execMust(t, e, "SELECT full_name FROM humans"); res.Rows[0][0] != "ada" {
		t.Fatalf("staged RENAME TABLE read = %v", res.Rows)
	}
	execMust(t, e, "ROLLBACK")

	res = execMust(t, e, "SELECT id, name FROM people")
	if res.RowCount != 1 || res.Rows[0][1] != "ada" {
		t.Fatalf("rows after rollback = %v", res.Rows)
	}
	if _, err := execSQL(e, "SELECT age FROM people"); err == nil {
		t.Fatal("rolled-back ADD COLUMN is still visible")
	}
	if _, err := execSQL(e, "SELECT * FROM humans"); err == nil {
		t.Fatal("rolled-back RENAME TABLE is still visible")
	}
	if _, err := execSQL(e, "SELECT * FROM people"); err != nil {
		t.Fatalf("original table name lost by ROLLBACK: %v", err)
	}
}

// transactional DDL: CREATE INDEX is transaction-scoped.
func TestTransactionalDDLCreateIndexRollback(t *testing.T) {
	_, schema, table := newTestDB(t)
	e := newExec(schema, table)
	execMust(t, e, "CREATE TABLE items (id INTEGER PRIMARY KEY, sku TEXT)")
	execMust(t, e, "INSERT INTO items VALUES (1, 'a'), (2, 'b')")

	execMust(t, e, "BEGIN")
	execMust(t, e, "CREATE UNIQUE INDEX idx_items_sku ON items (sku)")
	if !e.session.IndexExists("idx_items_sku") {
		t.Fatal("staged index missing from the transaction's catalog")
	}
	if schema.IndexExists("idx_items_sku") {
		t.Fatal("staged index leaked into the committed catalog")
	}
	execMust(t, e, "ROLLBACK")

	if schema.IndexExists("idx_items_sku") {
		t.Fatal("rolled-back index survived in the durable catalog")
	}
	// The index name is reusable, which proves the catalog write was discarded.
	execMust(t, e, "CREATE INDEX idx_items_sku ON items (sku)")
}

// transactional DDL: another session keeps reading the committed catalog until
// the DDL commit publishes it.
func TestTransactionalDDLTwoSessionVisibility(t *testing.T) {
	_, schema, table := newTestDB(t)
	writer := newExec(schema, table)
	observer := newExec(schema, table)

	execMust(t, writer, "CREATE TABLE base (id INTEGER PRIMARY KEY, v TEXT)")
	execMust(t, observer, "SELECT * FROM base")
	execMust(t, writer, "INSERT INTO base VALUES (1, 'committed')")

	execMust(t, writer, "BEGIN")
	execMust(t, writer, "CREATE TABLE extra (id INTEGER PRIMARY KEY)")
	execMust(t, writer, "ALTER TABLE base ADD COLUMN note TEXT DEFAULT 'new'")
	execMust(t, writer, "CREATE INDEX idx_base_note ON base (note)")

	// The observer sees neither the new table nor the new column, and it still
	// reads the committed rows.
	if _, err := execSQL(observer, "SELECT * FROM extra"); err == nil {
		t.Fatal("observer saw staged CREATE TABLE before COMMIT")
	}
	if _, err := execSQL(observer, "SELECT note FROM base"); err == nil {
		t.Fatal("observer saw a staged column before COMMIT")
	}
	if res := execMust(t, observer, "SELECT v FROM base"); res.RowCount != 1 || res.Rows[0][0] != "committed" {
		t.Fatalf("observer read = %v", res.Rows)
	}

	execMust(t, writer, "COMMIT")

	if _, err := execSQL(observer, "SELECT * FROM extra"); err != nil {
		t.Fatalf("observer did not see committed CREATE TABLE: %v", err)
	}
	// The observer's executor resynchronizes its analyzer catalog from the
	// bumped catalog version.
	res := execMust(t, observer, "SELECT note FROM base")
	if res.RowCount != 1 || res.Rows[0][0] != "new" {
		t.Fatalf("observer row after COMMIT = %v", res.Rows)
	}
	if !schema.IndexExists("idx_base_note") {
		t.Fatal("committed index missing from the durable catalog")
	}
}

// transactional DDL: COMMIT publishes every staged statement atomically and the
// result reloads from durable storage.
func TestTransactionalDDLCommitPublishesAndReloads(t *testing.T) {
	pool, schema, table := newTestDB(t)
	e := newExec(schema, table)
	execMust(t, e, "CREATE TABLE orders (id INTEGER PRIMARY KEY, total INTEGER)")
	execMust(t, e, "INSERT INTO orders VALUES (1, 10)")

	execMust(t, e, "BEGIN")
	execMust(t, e, "ALTER TABLE orders ADD COLUMN region TEXT DEFAULT 'eu'")
	execMust(t, e, "CREATE TABLE shipments (id INTEGER PRIMARY KEY, order_id INTEGER)")
	execMust(t, e, "INSERT INTO shipments VALUES (100, 1)")
	execMust(t, e, "CREATE INDEX idx_orders_region ON orders (region)")
	execMust(t, e, "COMMIT")

	// Reload the catalog from durable state with fresh managers over the same KV
	// server: the commit must have written schema, index, and rows together.
	reloadedSchemas := storage.NewSchemaManager(pool, "testdb")
	reloadedTables := storage.NewTableManager(pool, reloadedSchemas, "testdb")

	orders, err := reloadedSchemas.GetSchema("orders")
	if err != nil {
		t.Fatal(err)
	}
	if _, ok := orders.GetColumn("region"); !ok {
		t.Fatal("committed ADD COLUMN missing after reload")
	}
	if !reloadedSchemas.IndexExists("idx_orders_region") {
		t.Fatal("committed index missing after reload")
	}
	if !reloadedSchemas.TableExists("shipments") {
		t.Fatal("committed CREATE TABLE missing after reload")
	}
	orderRow, err := reloadedTables.GetByPK("orders", "1")
	if err != nil {
		t.Fatal(err)
	}
	if orderRow["region"] != "eu" {
		t.Fatalf("backfilled default after reload = %v", orderRow)
	}
	shipment, err := reloadedTables.GetByPK("shipments", "100")
	if err != nil || shipment["order_id"] != int64(1) {
		t.Fatalf("committed insert after reload: %v %v", shipment, err)
	}
}

// transactional DDL: DML and DDL commit or roll back as one unit.
func TestTransactionalDDLWithDML(t *testing.T) {
	_, schema, table := newTestDB(t)
	e := newExec(schema, table)
	execMust(t, e, "CREATE TABLE events (id INTEGER PRIMARY KEY, kind TEXT)")

	execMust(t, e, "BEGIN")
	execMust(t, e, "CREATE TABLE archive (id INTEGER PRIMARY KEY, kind TEXT)")
	execMust(t, e, "INSERT INTO archive VALUES (1, 'old')")
	execMust(t, e, "INSERT INTO events VALUES (1, 'new')")
	execMust(t, e, "ALTER TABLE events ADD COLUMN archived INTEGER DEFAULT 0")
	execMust(t, e, "ROLLBACK")

	if _, err := execSQL(e, "SELECT * FROM archive"); err == nil {
		t.Fatal("rollback kept a table created with DML in the same transaction")
	}
	if res := execMust(t, e, "SELECT COUNT(*) FROM events"); res.Rows[0][0] != int64(0) {
		t.Fatalf("rollback kept a DML row: %v", res.Rows)
	}
	if _, err := execSQL(e, "SELECT archived FROM events"); err == nil {
		t.Fatal("rollback kept a staged column")
	}

	// The same sequence commits as one unit.
	execMust(t, e, "BEGIN")
	execMust(t, e, "CREATE TABLE archive (id INTEGER PRIMARY KEY, kind TEXT)")
	execMust(t, e, "INSERT INTO archive VALUES (1, 'old')")
	execMust(t, e, "INSERT INTO events VALUES (1, 'new')")
	execMust(t, e, "ALTER TABLE events ADD COLUMN archived INTEGER DEFAULT 0")
	execMust(t, e, "COMMIT")

	if res := execMust(t, e, "SELECT COUNT(*) FROM archive"); res.Rows[0][0] != int64(1) {
		t.Fatalf("committed CREATE TABLE rows = %v", res.Rows)
	}
	if res := execMust(t, e, "SELECT archived FROM events WHERE id = 1"); res.Rows[0][0] != int64(0) {
		t.Fatalf("committed ADD COLUMN default = %v", res.Rows)
	}
}

// Transactional DDL: a rejected statement leaves the transaction's prior schema
// intact, including DDL staged before it.
func TestTransactionalDDLFailureLeavesPriorSchema(t *testing.T) {
	_, schema, table := newTestDB(t)
	e := newExec(schema, table)
	execMust(t, e, "CREATE TABLE parent (id INTEGER PRIMARY KEY)")
	execMust(t, e, "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id))")
	execMust(t, e, "INSERT INTO parent VALUES (1)")
	execMust(t, e, "INSERT INTO child VALUES (10, 1)")

	execMust(t, e, "BEGIN")
	execMust(t, e, "CREATE TABLE scratch (id INTEGER PRIMARY KEY)")
	// Duplicate primary key inside the staged table fails the statement.
	if _, err := execSQL(e, "INSERT INTO scratch VALUES (1), (1)"); err == nil {
		t.Fatal("duplicate primary key accepted")
	}
	// Dropping a referenced parent is rejected.
	if _, err := execSQL(e, "DROP TABLE parent"); err == nil {
		t.Fatal("dropped a table another table references")
	}
	// Adding a NOT NULL column without a default to a populated table fails.
	if _, err := execSQL(e, "ALTER TABLE child ADD COLUMN required TEXT NOT NULL"); err == nil {
		t.Fatal("NOT NULL column without default accepted on a populated table")
	}

	// Earlier staged DDL is still visible and staged DML is intact.
	if res := execMust(t, e, "SELECT COUNT(*) FROM scratch"); res.Rows[0][0] != int64(0) {
		t.Fatalf("scratch rows = %v", res.Rows)
	}
	if res := execMust(t, e, "SELECT COUNT(*) FROM child"); res.Rows[0][0] != int64(1) {
		t.Fatalf("child rows after failed statements = %v", res.Rows)
	}
	execMust(t, e, "COMMIT")

	if !schema.TableExists("scratch") {
		t.Fatal("committed staged table missing after a failed statement")
	}
	if schema.TableExists("scratch") && schema.TableExists("nonexistent") {
		t.Fatal("unreachable")
	}
}

// Transactional DDL: a multi-statement migration rolls back as one unit.
func TestTransactionalDDLMultiStatementMigrationRollback(t *testing.T) {
	_, schema, table := newTestDB(t)
	e := newExec(schema, table)
	execMust(t, e, "CREATE TABLE accounts (id INTEGER PRIMARY KEY, balance INTEGER)")

	execMust(t, e, "BEGIN")
	execMust(t, e, "CREATE TABLE audit (id INTEGER PRIMARY KEY, account_id INTEGER)")
	execMust(t, e, "ALTER TABLE accounts ADD COLUMN currency TEXT DEFAULT 'EUR'")
	execMust(t, e, "CREATE UNIQUE INDEX idx_accounts_currency ON accounts (currency)")
	execMust(t, e, "INSERT INTO accounts VALUES (1, 100, 'EUR')")
	execMust(t, e, "INSERT INTO audit VALUES (1, 1)")
	execMust(t, e, "ROLLBACK")

	if schema.TableExists("audit") || schema.IndexExists("idx_accounts_currency") {
		t.Fatal("migration objects survived ROLLBACK")
	}
	accounts, err := schema.GetSchema("accounts")
	if err != nil {
		t.Fatal(err)
	}
	if _, ok := accounts.GetColumn("currency"); ok {
		t.Fatal("migration column survived ROLLBACK")
	}
}

// Transactional DDL: a UNIQUE index staged by the transaction is enforced by
// the commit's validation, so duplicate values never become durable.
func TestTransactionalDDLStagedUniqueIndexValidation(t *testing.T) {
	_, schema, table := newTestDB(t)
	e := newExec(schema, table)
	execMust(t, e, "CREATE TABLE accounts (id INTEGER PRIMARY KEY, balance INTEGER)")
	execMust(t, e, "INSERT INTO accounts VALUES (1, 100)")

	execMust(t, e, "BEGIN")
	execMust(t, e, "ALTER TABLE accounts ADD COLUMN currency TEXT DEFAULT 'EUR'")
	execMust(t, e, "CREATE UNIQUE INDEX idx_accounts_currency ON accounts (currency)")
	execMust(t, e, "INSERT INTO accounts (id, balance) VALUES (2, 200)")
	// Both rows now carry the backfilled default 'EUR', so the staged UNIQUE
	// index must reject the transaction at commit.
	if _, err := execSQL(e, "COMMIT"); err == nil || !strings.Contains(err.Error(), "UNIQUE constraint failed") {
		t.Fatalf("commit with duplicate staged UNIQUE values = %v", err)
	}
	if schema.IndexExists("idx_accounts_currency") {
		t.Fatal("failed commit published the staged index")
	}
	if res := execMust(t, e, "SELECT COUNT(*) FROM accounts"); res.Rows[0][0] != int64(1) {
		t.Fatalf("failed commit published staged rows: %v", res.Rows)
	}
}

// Transactional DDL: validations run against the staged catalog.
func TestTransactionalDDLValidationsUseStagedSchema(t *testing.T) {
	_, schema, table := newTestDB(t)
	e := newExec(schema, table)

	execMust(t, e, "BEGIN")
	execMust(t, e, "CREATE TABLE parents (a INTEGER, b INTEGER, PRIMARY KEY (a, b))")
	execMust(t, e, "CREATE TABLE kids (id INTEGER PRIMARY KEY, pa INTEGER, pb INTEGER, FOREIGN KEY (pa, pb) REFERENCES parents(a, b))")
	execMust(t, e, "INSERT INTO parents VALUES (1, 2)")
	execMust(t, e, "INSERT INTO kids VALUES (10, 1, 2)")
	// Composite primary key: the same pair collides, a different pair does not.
	if _, err := execSQL(e, "INSERT INTO parents VALUES (1, 2)"); err == nil {
		t.Fatal("duplicate composite primary key accepted")
	}
	execMust(t, e, "INSERT INTO parents VALUES (3, 4)")
	// CHECK added by a staged ALTER TABLE is enforced by staged writes.
	execMust(t, e, "ALTER TABLE kids ADD COLUMN age INTEGER CHECK (age >= 0)")
	execMust(t, e, "INSERT INTO kids VALUES (12, 3, 4, 5)")
	if _, err := execSQL(e, "INSERT INTO kids VALUES (13, 3, 4, -1)"); err == nil {
		t.Fatal("staged CHECK constraint not enforced")
	}
	// NOT NULL added by a staged ALTER TABLE is enforced by staged writes.
	execMust(t, e, "ALTER TABLE kids ADD COLUMN tag TEXT NOT NULL DEFAULT 't'")
	if _, err := execSQL(e, "INSERT INTO kids (id, pa, pb, age, tag) VALUES (14, 3, 4, 1, NULL)"); err == nil {
		t.Fatal("staged NOT NULL constraint not enforced")
	}
	execMust(t, e, "COMMIT")

	// The staged constraints are durable and still enforced.
	if _, err := execSQL(e, "INSERT INTO kids VALUES (15, 3, 4, -1, 'x')"); err == nil {
		t.Fatal("committed CHECK constraint not enforced")
	}

	// A composite foreign key declared by staged DDL is validated at commit
	// against the transaction's final state.
	execMust(t, e, "BEGIN")
	execMust(t, e, "INSERT INTO kids (id, pa, pb) VALUES (11, 1, 9)")
	if _, err := execSQL(e, "COMMIT"); err == nil || !strings.Contains(err.Error(), "FOREIGN KEY constraint failed") {
		t.Fatalf("commit with a dangling staged composite foreign key = %v", err)
	}
	if res := execMust(t, e, "SELECT COUNT(*) FROM kids WHERE id = 11"); res.Rows[0][0] != int64(0) {
		t.Fatalf("failed commit published the dangling row: %v", res.Rows)
	}
	if _, err := execSQL(e, "INSERT INTO kids (id, pa, pb) VALUES (16, 8, 9)"); err == nil {
		t.Fatal("committed FOREIGN KEY not enforced")
	}
}

// Transactional DDL: cascades and composite keys keep working across a commit.
func TestTransactionalDDLCascadesAcrossCommit(t *testing.T) {
	_, schema, table := newTestDB(t)
	e := newExec(schema, table)

	execMust(t, e, "BEGIN")
	execMust(t, e, "CREATE TABLE parents (id INTEGER PRIMARY KEY)")
	execMust(t, e, "CREATE TABLE kids (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parents(id) ON DELETE CASCADE)")
	execMust(t, e, "INSERT INTO parents VALUES (1)")
	execMust(t, e, "INSERT INTO kids VALUES (10, 1)")
	execMust(t, e, "COMMIT")

	if res := execMust(t, e, "SELECT COUNT(*) FROM kids"); res.Rows[0][0] != int64(1) {
		t.Fatalf("kids = %v", res.Rows)
	}
	execMust(t, e, "DELETE FROM parents WHERE id = 1")
	if res := execMust(t, e, "SELECT COUNT(*) FROM kids"); res.Rows[0][0] != int64(0) {
		t.Fatalf("cascade did not delete the child row: %v", res.Rows)
	}
}

// Transactional DDL: a concurrent committed DDL that conflicts with the staged
// change fails the commit with a serialization error instead of corrupting the
// catalog.
func TestTransactionalDDLConcurrentConflict(t *testing.T) {
	_, schema, table := newTestDB(t)
	first := newExec(schema, table)
	second := newExec(schema, table)

	execMust(t, first, "BEGIN")
	execMust(t, first, "CREATE TABLE races (id INTEGER PRIMARY KEY)")

	// A second session creates the same table and commits first.
	execMust(t, second, "CREATE TABLE races (id INTEGER PRIMARY KEY, v TEXT)")

	if _, err := execSQL(first, "COMMIT"); err == nil {
		t.Fatal("conflicting DDL commit succeeded")
	} else if !strings.Contains(err.Error(), "serialization") {
		t.Fatalf("conflicting DDL commit error = %v, want a serialization error", err)
	}

	// The committed definition wins and stays intact.
	stored, err := schema.GetSchema("races")
	if err != nil {
		t.Fatal(err)
	}
	if _, ok := stored.GetColumn("v"); !ok {
		t.Fatalf("concurrent committed schema was overwritten: %+v", stored.Columns)
	}
}

// Transactional DDL: a savepoint rollback also reverts staged DDL.
func TestTransactionalDDLSavepointRollback(t *testing.T) {
	_, schema, table := newTestDB(t)
	e := newExec(schema, table)
	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")

	execMust(t, e, "BEGIN")
	execMust(t, e, "CREATE TABLE before_sp (id INTEGER PRIMARY KEY)")
	execMust(t, e, "SAVEPOINT sp1")
	execMust(t, e, "CREATE TABLE after_sp (id INTEGER PRIMARY KEY)")
	execMust(t, e, "ALTER TABLE t ADD COLUMN v TEXT DEFAULT 'x'")
	execMust(t, e, "ROLLBACK TO SAVEPOINT sp1")

	if _, err := execSQL(e, "SELECT * FROM after_sp"); err == nil {
		t.Fatal("DDL after the savepoint survived ROLLBACK TO SAVEPOINT")
	}
	if _, err := execSQL(e, "SELECT v FROM t"); err == nil {
		t.Fatal("ALTER after the savepoint survived ROLLBACK TO SAVEPOINT")
	}
	if _, err := execSQL(e, "SELECT * FROM before_sp"); err != nil {
		t.Fatalf("DDL before the savepoint was lost: %v", err)
	}
	execMust(t, e, "COMMIT")

	if !schema.TableExists("before_sp") || schema.TableExists("after_sp") {
		t.Fatal("committed savepoint result is wrong")
	}
}

// transactional DDL: DROP INDEX is undone by ROLLBACK and stays invisible to
// other sessions until COMMIT.
func TestTransactionalDDLDropIndexRollback(t *testing.T) {
	_, schema, table := newTestDB(t)
	writer := newExec(schema, table)
	observer := newExec(schema, table)
	execMust(t, writer, "CREATE TABLE docs (id INTEGER PRIMARY KEY, slug TEXT)")
	execMust(t, writer, "CREATE UNIQUE INDEX idx_docs_slug ON docs (slug)")
	execMust(t, writer, "INSERT INTO docs VALUES (1, 'a')")

	execMust(t, writer, "BEGIN")
	execMust(t, writer, "DROP INDEX idx_docs_slug")
	if writer.session.IndexExists("idx_docs_slug") {
		t.Fatal("staged DROP INDEX is still visible to the transaction")
	}
	if !observer.session.IndexExists("idx_docs_slug") {
		t.Fatal("staged DROP INDEX became visible to another session before COMMIT")
	}
	execMust(t, writer, "ROLLBACK")
	if !schema.IndexExists("idx_docs_slug") {
		t.Fatal("rollback did not restore the dropped index")
	}

	// The duplicate still fails, proving the unique index survived intact.
	if _, err := execSQL(writer, "INSERT INTO docs VALUES (2, 'a')"); err == nil {
		t.Fatal("unique index lost its enforcement after a rolled-back DROP INDEX")
	}

	execMust(t, writer, "BEGIN")
	execMust(t, writer, "DROP INDEX idx_docs_slug")
	execMust(t, writer, "COMMIT")
	if schema.IndexExists("idx_docs_slug") {
		t.Fatal("committed DROP INDEX did not reach the catalog")
	}
	execMust(t, observer, "INSERT INTO docs VALUES (3, 'a')")
}

// transactional DDL: a renamed table stays writable inside the transaction and
// the staged rows move with it at commit.
func TestTransactionalDDLRenameTableReadYourWrites(t *testing.T) {
	_, schema, table := newTestDB(t)
	e := newExec(schema, table)
	execMust(t, e, "CREATE TABLE old_name (id INTEGER PRIMARY KEY, v TEXT)")
	execMust(t, e, "INSERT INTO old_name VALUES (1, 'kept')")

	execMust(t, e, "BEGIN")
	execMust(t, e, "ALTER TABLE old_name RENAME TO new_name")
	execMust(t, e, "INSERT INTO new_name VALUES (2, 'staged')")
	if res := execMust(t, e, "SELECT v FROM new_name ORDER BY id"); res.RowCount != 2 {
		t.Fatalf("rows after staged rename = %v", res.Rows)
	}
	execMust(t, e, "COMMIT")

	if schema.TableExists("old_name") {
		t.Fatal("renamed table is still addressable under its old name")
	}
	res := execMust(t, e, "SELECT id, v FROM new_name ORDER BY id")
	if res.RowCount != 2 || res.Rows[0][1] != "kept" || res.Rows[1][1] != "staged" {
		t.Fatalf("rows after committed rename = %v", res.Rows)
	}
	// The durable rows moved rather than being copied.
	if rows, err := table.Select("old_name", nil); err == nil && len(rows) > 0 {
		t.Fatalf("old data prefix still holds %d rows", len(rows))
	}
}

// transactional DDL: another session keeps reading a dropped table's rows until
// the drop commits.
func TestTransactionalDDLDropTableTwoSessions(t *testing.T) {
	_, schema, table := newTestDB(t)
	writer := newExec(schema, table)
	observer := newExec(schema, table)
	execMust(t, writer, "CREATE TABLE doomed (id INTEGER PRIMARY KEY, v TEXT)")
	execMust(t, writer, "INSERT INTO doomed VALUES (1, 'still here')")

	execMust(t, writer, "BEGIN")
	execMust(t, writer, "DROP TABLE doomed")
	if res := execMust(t, observer, "SELECT v FROM doomed"); res.RowCount != 1 || res.Rows[0][0] != "still here" {
		t.Fatalf("observer rows while the drop is uncommitted = %v", res.Rows)
	}
	execMust(t, writer, "COMMIT")

	if _, err := execSQL(observer, "SELECT * FROM doomed"); err == nil {
		t.Fatal("observer still reads a dropped table after COMMIT")
	}
	if rows, err := table.Select("doomed", nil); err == nil && len(rows) > 0 {
		t.Fatalf("dropped table left %d durable rows", len(rows))
	}
}

// transactional DDL: a write that commits after the DDL statement but before the
// DDL commit makes the commit fail with a serialization error rather than
// silently discarding the concurrent write.
func TestTransactionalDDLConcurrentWriteConflicts(t *testing.T) {
	_, schema, table := newTestDB(t)
	writer := newExec(schema, table)
	other := newExec(schema, table)
	execMust(t, writer, "CREATE TABLE shared (id INTEGER PRIMARY KEY, v TEXT)")
	execMust(t, writer, "INSERT INTO shared VALUES (1, 'a')")

	execMust(t, writer, "BEGIN")
	execMust(t, writer, "ALTER TABLE shared ADD COLUMN note TEXT DEFAULT 'x'")

	// Another session writes the table between the DDL statement and the commit.
	execMust(t, other, "INSERT INTO shared VALUES (2, 'b')")

	if _, err := execSQL(writer, "COMMIT"); err == nil {
		t.Fatal("commit over a concurrent write succeeded")
	} else if !strings.Contains(err.Error(), "serialization") {
		t.Fatalf("commit error = %v, want a serialization error", err)
	}
	// The concurrent row is intact and the column was never published.
	if res := execMust(t, other, "SELECT COUNT(*) FROM shared"); res.Rows[0][0] != int64(2) {
		t.Fatalf("concurrent rows after failed commit = %v", res.Rows)
	}
	if _, err := execSQL(other, "SELECT note FROM shared"); err == nil {
		t.Fatal("failed commit published the staged column")
	}
}