TsDB2/DB/driver/p1_features_test.go

package vrtxsql

// P1 功能测试: ALTER TABLE ADD/DROP COLUMN、SHOW TABLES/COLUMNS、DESCRIBE、
// TRUNCATE、DISTINCT、LIMIT offset,count、SQLSTATE、DSN mode=ro/cache=private。

import (
	"database/sql"
	"errors"
	"fmt"
	"path/filepath"
	"sort"
	"testing"
)

// sqlState 提取错误的 SQLSTATE; 不是 *SQLError 返回 ""。
func sqlState(err error) string {
	var se *SQLError
	if errors.As(err, &se) {
		return se.State
	}
	return ""
}

// TestP1AlterTable ADD/DROP COLUMN: 新列对旧行为 NULL, DROP 后列消失。
func TestP1AlterTable(t *testing.T) {
	db := openTestDB(t)
	if _, err := db.Exec(`CREATE TABLE t (id BIGINT PRIMARY KEY, v INT)`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO t (id, v) VALUES (?, ?)`, 1, 100); err != nil {
		t.Fatal(err)
	}

	// ADD COLUMN: 旧行的新列应为 NULL
	if _, err := db.Exec(`ALTER TABLE t ADD COLUMN nickname VARCHAR(20)`); err != nil {
		t.Fatalf("add column: %v", err)
	}
	var nick sql.NullString
	if err := db.QueryRow(`SELECT nickname FROM t WHERE id = ?`, 1).Scan(&nick); err != nil {
		t.Fatalf("read new column: %v", err)
	}
	if nick.Valid {
		t.Fatalf("旧行新列应为 NULL, 实际 %q", nick.String)
	}
	// 新列可写可读
	if _, err := db.Exec(`UPDATE t SET nickname = ? WHERE id = ?`, "neo", 1); err != nil {
		t.Fatalf("update new column: %v", err)
	}
	if err := db.QueryRow(`SELECT nickname FROM t WHERE id = ?`, 1).Scan(&nick); err != nil {
		t.Fatal(err)
	}
	if !nick.Valid || nick.String != "neo" {
		t.Fatalf("新列写入读回不对: %+v", nick)
	}

	// DROP COLUMN: v 列消失
	if _, err := db.Exec(`ALTER TABLE t DROP COLUMN v`); err != nil {
		t.Fatalf("drop column: %v", err)
	}
	rs, err := db.Query(`SELECT * FROM t`)
	if err != nil {
		t.Fatal(err)
	}
	cols, _ := rs.Columns()
	if len(cols) != 2 { // id + nickname
		t.Fatalf("DROP 后应剩 2 列, 实际 %d (%v)", len(cols), cols)
	}
	rs.Close()
	if _, err := db.Query(`SELECT v FROM t`); err == nil {
		t.Fatal("查询已删除列应报错")
	} else if st := sqlState(err); st != "42S22" {
		t.Fatalf("已删除列错误状态=%q, 想要 42S22", st)
	}
}

// TestP1ShowAndDescribe SHOW TABLES / SHOW COLUMNS / DESCRIBE 结果集。
func TestP1ShowAndDescribe(t *testing.T) {
	db := openTestDB(t)
	if _, err := db.Exec(`CREATE TABLE t (
		id BIGINT PRIMARY KEY,
		name VARCHAR(32) NOT NULL,
		age INT
	)`); err != nil {
		t.Fatal(err)
	}

	// SHOW TABLES: 单列, 含 t
	rs, err := db.Query(`SHOW TABLES`)
	if err != nil {
		t.Fatalf("show tables: %v", err)
	}
	found := false
	for rs.Next() {
		var nm []byte
		if err := rs.Scan(&nm); err != nil {
			t.Fatal(err)
		}
		if string(nm) == "t" {
			found = true
		}
	}
	rs.Close()
	if !found {
		t.Fatal("SHOW TABLES 未列出 t")
	}

	// SHOW COLUMNS FROM t: 六列, 三行; id 行 Key=PRI、name 行 Null=NO
	checkColumns := func(query string) {
		c, err := db.Query(query)
		if err != nil {
			t.Fatalf("%s: %v", query, err)
		}
		defer c.Close()
		cols, _ := c.Columns()
		if len(cols) != 6 {
			t.Fatalf("%s 应返回 6 列, 实际 %d", query, len(cols))
		}
		rows := 0
		for c.Next() {
			var field, typ, null, key, def, extra []byte
			if err := c.Scan(&field, &typ, &null, &key, &def, &extra); err != nil {
				t.Fatal(err)
			}
			switch string(field) {
			case "id":
				if string(key) != "PRI" {
					t.Fatalf("id Key=%q, 想要 PRI", key)
				}
			case "name":
				if string(null) != "NO" {
					t.Fatalf("name Null=%q, 想要 NO", null)
				}
				if string(typ) != "varchar(32)" {
					t.Fatalf("name Type=%q, 想要 varchar(32)", typ)
				}
			}
			rows++
		}
		if rows != 3 {
			t.Fatalf("%s 应描述 3 列, 实际 %d", query, rows)
		}
	}
	checkColumns(`SHOW COLUMNS FROM t`)
	checkColumns(`DESCRIBE t`)
	checkColumns(`DESC t`)
}

// TestP1Truncate 清空数据但保留结构, 清空后可继续插入。
func TestP1Truncate(t *testing.T) {
	db := openTestDB(t)
	if _, err := db.Exec(`CREATE TABLE t (id BIGINT PRIMARY KEY, v INT)`); err != nil {
		t.Fatal(err)
	}
	for i := 1; i <= 5; i++ {
		if _, err := db.Exec(`INSERT INTO t (id, v) VALUES (?, ?)`, i, i*10); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := db.Exec(`TRUNCATE TABLE t`); err != nil {
		t.Fatalf("truncate: %v", err)
	}
	var cnt int
	if err := db.QueryRow(`SELECT COUNT(*) FROM t`).Scan(&cnt); err != nil {
		t.Fatal(err)
	}
	if cnt != 0 {
		t.Fatalf("TRUNCATE 后应 0 行, 实际 %d", cnt)
	}
	// 表结构仍在, 可继续插入
	if _, err := db.Exec(`INSERT INTO t (id, v) VALUES (?, ?)`, 100, 1); err != nil {
		t.Fatalf("TRUNCATE 后插入失败: %v", err)
	}
}

// TestP1Distinct 投影列去重。
func TestP1Distinct(t *testing.T) {
	db := openTestDB(t)
	if _, err := db.Exec(`CREATE TABLE t (id BIGINT PRIMARY KEY, cat VARCHAR(10))`); err != nil {
		t.Fatal(err)
	}
	cats := []string{"a", "b", "a", "c", "b"}
	for i, c := range cats {
		if _, err := db.Exec(`INSERT INTO t (id, cat) VALUES (?, ?)`, i+1, c); err != nil {
			t.Fatal(err)
		}
	}
	rs, err := db.Query(`SELECT DISTINCT cat FROM t`)
	if err != nil {
		t.Fatalf("distinct: %v", err)
	}
	defer rs.Close()
	var got []string
	for rs.Next() {
		var c []byte
		if err := rs.Scan(&c); err != nil {
			t.Fatal(err)
		}
		got = append(got, string(c))
	}
	sort.Strings(got)
	if len(got) != 3 || got[0] != "a" || got[1] != "b" || got[2] != "c" {
		t.Fatalf("DISTINCT 结果 %v, 想要 [a b c]", got)
	}
}

// TestP1LimitOffsetComma LIMIT offset,count 语法: 跳过 offset 行取 count 行。
func TestP1LimitOffsetComma(t *testing.T) {
	db := openTestDB(t)
	if _, err := db.Exec(`CREATE TABLE t (id BIGINT PRIMARY KEY)`); err != nil {
		t.Fatal(err)
	}
	for i := 1; i <= 10; i++ {
		if _, err := db.Exec(`INSERT INTO t (id) VALUES (?)`, i); err != nil {
			t.Fatal(err)
		}
	}
	rs, err := db.Query(`SELECT id FROM t ORDER BY id LIMIT 2,3`)
	if err != nil {
		t.Fatalf("limit n,m: %v", err)
	}
	defer rs.Close()
	var got []int64
	for rs.Next() {
		var id int64
		if err := rs.Scan(&id); err != nil {
			t.Fatal(err)
		}
		got = append(got, id)
	}
	want := []int64{3, 4, 5}
	if len(got) != 3 || got[0] != want[0] || got[1] != want[1] || got[2] != want[2] {
		t.Fatalf("LIMIT 2,3 结果 %v, 想要 %v", got, want)
	}
}

// TestP1SQLState 关键错误的 SQLSTATE 映射。
func TestP1SQLState(t *testing.T) {
	db := openTestDB(t)
	if _, err := db.Exec(`CREATE TABLE t (id BIGINT PRIMARY KEY, v INT)`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO t (id) VALUES (?)`, 1); err != nil {
		t.Fatal(err)
	}

	// 表不存在 → 42S02
	_, err := db.Query(`SELECT * FROM nope`)
	if st := sqlState(err); st != "42S02" {
		t.Fatalf("表不存在 state=%q, 想要 42S02 (err=%v)", st, err)
	}
	// 主键重复 → 23000
	_, err = db.Exec(`INSERT INTO t (id) VALUES (?)`, 1)
	if st := sqlState(err); st != "23000" {
		t.Fatalf("主键重复 state=%q, 想要 23000 (err=%v)", st, err)
	}
	// 语法错误 → 42000
	_, err = db.Exec(`SELEC bad`)
	if st := sqlState(err); st != "42000" {
		t.Fatalf("语法错误 state=%q, 想要 42000 (err=%v)", st, err)
	}
	// 列不存在 → 42S22
	_, err = db.Query(`SELECT nope FROM t`)
	if st := sqlState(err); st != "42S22" {
		t.Fatalf("列不存在 state=%q, 想要 42S22 (err=%v)", st, err)
	}
}

// TestP1ReadOnlyDSN mode=ro: 可读不可写 (写报 25008); cache=private 独立实例持久化。
func TestP1ReadOnlyDSN(t *testing.T) {
	path := filepath.Join(t.TempDir(), "ro.db")

	// 先写数据再关闭
	w, err := sql.Open("vrtx", path)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := w.Exec(`CREATE TABLE t (id BIGINT PRIMARY KEY, v INT)`); err != nil {
		t.Fatal(err)
	}
	if _, err := w.Exec(`INSERT INTO t (id, v) VALUES (?, ?)`, 1, 10); err != nil {
		t.Fatal(err)
	}
	if err := w.Close(); err != nil {
		t.Fatal(err)
	}

	// 只读打开
	ro, err := sql.Open("vrtx", path+"?mode=ro")
	if err != nil {
		t.Fatal(err)
	}
	defer ro.Close()
	var cnt int
	if err := ro.QueryRow(`SELECT COUNT(*) FROM t`).Scan(&cnt); err != nil {
		t.Fatalf("只读 SELECT 失败: %v", err)
	}
	if cnt != 1 {
		t.Fatalf("只读读到 %d 行, 想要 1", cnt)
	}
	_, err = ro.Exec(`INSERT INTO t (id, v) VALUES (?, ?)`, 2, 20)
	if st := sqlState(err); st != "25008" {
		t.Fatalf("只读写入 state=%q, 想要 25008 (err=%v)", st, err)
	}
	if err := ro.Close(); err != nil {
		t.Fatal(err)
	}

	// cache=private: 独立实例, 数据持久化、可继续写
	pv, err := sql.Open("vrtx", path+"?cache=private")
	if err != nil {
		t.Fatal(err)
	}
	if _, err := pv.Exec(`INSERT INTO t (id, v) VALUES (?, ?)`, 2, 20); err != nil {
		t.Fatalf("private 实例写入: %v", err)
	}
	if err := pv.QueryRow(`SELECT COUNT(*) FROM t`).Scan(&cnt); err != nil {
		t.Fatal(err)
	}
	if cnt != 2 {
		t.Fatalf("private 实例应读到 2 行, 实际 %d", cnt)
	}
	pv.Close()

	// 关闭后普通连接能看到 private 写入 (独立实例也正确落盘)
	plain, err := sql.Open("vrtx", path)
	if err != nil {
		t.Fatal(err)
	}
	defer plain.Close()
	if err := plain.QueryRow(`SELECT COUNT(*) FROM t`).Scan(&cnt); err != nil {
		t.Fatal(err)
	}
	if cnt != 2 {
		t.Fatalf("持久化后应 2 行, 实际 %d (%s)", cnt, fmt.Sprint(cnt))
	}
}

// TestP1Vacuum VACUUM 语句: 整库重建后行数据原样、文件可继续读写。
// 顺带验证三种典型调用形态: 裸语句 / 多行空白 / 大小写混合。
func TestP1Vacuum(t *testing.T) {
	db := openTestDB(t)
	if _, err := db.Exec(`CREATE TABLE t (id BIGINT PRIMARY KEY, v INT)`); err != nil {
		t.Fatal(err)
	}
	for i := 1; i <= 10; i++ {
		if _, err := db.Exec(`INSERT INTO t (id, v) VALUES (?, ?)`, i, i*10); err != nil {
			t.Fatal(err)
		}
	}
	// 删掉一半再 VACUUM: 有垃圾可收, 更接近真实场景
	if _, err := db.Exec(`DELETE FROM t WHERE id > ?`, 5); err != nil {
		t.Fatal(err)
	}

	res, err := db.Exec(`VACUUM`)
	if err != nil {
		t.Fatalf("vacuum: %v", err)
	}
	if n, err := res.RowsAffected(); err != nil || n != 5 {
		t.Fatalf("RowsAffected 应为复制的 5 行, 实际 %d (%v)", n, err)
	}

	// 行数据原样
	var cnt int
	if err := db.QueryRow(`SELECT COUNT(*) FROM t`).Scan(&cnt); err != nil {
		t.Fatal(err)
	}
	if cnt != 5 {
		t.Fatalf("VACUUM 后应 5 行, 实际 %d", cnt)
	}
	var v int
	if err := db.QueryRow(`SELECT v FROM t WHERE id = ?`, 3).Scan(&v); err != nil {
		t.Fatal(err)
	}
	if v != 30 {
		t.Fatalf("行数据应为 30, 实际 %d", v)
	}

	// 重建后可继续读写
	if _, err := db.Exec(`INSERT INTO t (id, v) VALUES (?, ?)`, 100, 1); err != nil {
		t.Fatalf("VACUUM 后插入失败: %v", err)
	}
	// 大小写/空白形态
	if _, err := db.Exec("  vacuum  "); err != nil {
		t.Fatalf("小写带空白 VACUUM: %v", err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM t`).Scan(&cnt); err != nil {
		t.Fatal(err)
	}
	if cnt != 6 {
		t.Fatalf("第二次 VACUUM 后应 6 行, 实际 %d", cnt)
	}
}

// TestP1VacuumInTxn 活跃事务中 VACUUM 必须被拒 (引擎层 ErrTxnActive 透传)。
func TestP1VacuumInTxn(t *testing.T) {
	db := openTestDB(t)
	if _, err := db.Exec(`CREATE TABLE t (id BIGINT PRIMARY KEY, v INT)`); err != nil {
		t.Fatal(err)
	}
	tx, err := db.Begin()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := tx.Exec(`INSERT INTO t (id, v) VALUES (?, ?)`, 1, 1); err != nil {
		t.Fatal(err)
	}
	if _, err := tx.Exec(`VACUUM`); err == nil {
		t.Fatal("事务中 VACUUM 应被拒绝")
	}
	if err := tx.Rollback(); err != nil {
		t.Fatal(err)
	}
	// 回滚后 VACUUM 恢复可用
	if _, err := db.Exec(`VACUUM`); err != nil {
		t.Fatalf("回滚后 VACUUM: %v", err)
	}
}