TsDB2/demo/unique_orm_test/main.go

package main

// ORM 侧 `vrtx:"unique"` 的语义链路:
//
//	tag → 列级 types.FlagUnique + 索引级 core.IndexFlagUnique
//	    → 索引写入路径 (InsertRow / UpdateByKey) 上校验
//
// 三条不变式:
//  1. 同一列值只能出现一次 (重复插入报 core.ErrDuplicatedKey);
//  2. NULL 不参与 —— 可空列的多行 NULL 共存 (NULL 行不进唯一索引, 索引里没有条目可撞);
//  3. 模型声明 unique 而库里那个索引不是唯一的, 重开表时必须报错 —— 否则
//     "模型说唯一、写入却允许重复", 要等数据脏了才被发现。
//
// 顺带钉住指针载体: `*int64` 推断出的列型与 `int64` 一致 (指针只是多了"整列为空"
// 这一档), nil 写 NULL、非 nil 解一层指针按值写, 读回仍是 nil / 7 而不是 0 / 7。
//
// 运行: cd demo/unique_orm_test && go run main.go

import (
	"errors"
	"fmt"
	"os"

	"TsDB2/DB"
	"TsDB2/DB/core"
	"TsDB2/DB/core/types"
)

// ============================ 模型 ============================

// ormUniq 唯一约束模型:
//
//	Name  int64  `vrtx:"unique"`          — 非空, 值唯一
//	Email *int64 `vrtx:"unique,nullable"` — 可空, 值唯一, 多行 NULL 共存
//	Group int64  `vrtx:"index"`           — 普通索引 (对照组)
type ormUniq struct {
	DB.BaseModel
	Name  int64  `vrtx:"unique"`
	Email *int64 `vrtx:"unique,nullable"`
	Group int64  `vrtx:"index"`
}

// ormUniqLax 列集合与索引名和 ormUniq 完全一致, 只是都不声明 unique ——
// 先用它建表再用 ormUniq 重开, 就构造出"索引已存在但不是唯一"的场景。
type ormUniqLax struct {
	DB.BaseModel
	Name  int64  `vrtx:"index"`
	Email *int64 `vrtx:"index,nullable"`
	Group int64  `vrtx:"index"`
}

// ormUniqPair 复合唯一键: index:u2 把两列并成一组, unique 只写在 A 上 ——
// 唯一性作用于 (A,B) 整组, 而不是 A 单列。
type ormUniqPair struct {
	DB.BaseModel
	A    int64 `vrtx:"index:u2,unique"`
	B    int64 `vrtx:"index:u2"`
	Note int64 `vrtx:"nullable"`
}

func i64ptr(v int64) *int64 { return &v }

// ============================ 断言工具 ============================

var passed, failed int

// check 记一笔断言结果; 返回是否通过 (便于"前提不成立就不往下走")。
func check(name string, ok bool, detail string) bool {
	if ok {
		passed++
		fmt.Printf("  ✅ %s\n", name)
		return true
	}
	failed++
	fmt.Printf("  ❌ %s — %s\n", name, detail)
	return false
}

// checkErr 断言 err == nil; 返回是否通过。
func checkErr(name string, err error) bool {
	if err != nil {
		check(name, false, err.Error())
		return false
	}
	check(name, true, "")
	return true
}

// checkDup 断言 err 命中 core.ErrDuplicatedKey。
func checkDup(name string, err error) {
	check(name, errors.Is(err, core.ErrDuplicatedKey),
		fmt.Sprintf("err = %v, want ErrDuplicatedKey", err))
}

// fatal 环境问题 (建库 / 建表) 直接中断。
func fatal(what string, err error) {
	if err != nil {
		fmt.Printf("  💥 %s: %v\n", what, err)
		os.Exit(1)
	}
}

// openDemoDB 打开本程序专用的库文件 (先删旧的)。
// 清理时若已有失败项就保留文件, 便于用 analyzer 排查。
func openDemoDB(path string) (*DB.DB, func()) {
	os.Remove(path)
	os.Remove(path + ".wal")
	db, err := DB.Open(path)
	if err != nil {
		panic(err)
	}
	db.SetAutoCommitDelay(0) // 0 = 关闭自动提交, 只在显式 Commit 时落盘
	return db, func() {
		_ = db.Close()
		if failed == 0 {
			os.Remove(path)
			os.Remove(path + ".wal")
		}
	}
}

// coreTbl 拿到底层 core.Table —— ORM 层不暴露索引元数据, 唯一位得从这里看。
func coreTbl(db *DB.DB, name string) *core.Table {
	t, ok := db.CoreDB().FindTable(name)
	if !ok {
		panic("core 里找不到表: " + name)
	}
	return t
}

// findIdx 按名字取出模型里的索引定义。
func findIdx(m *DB.ModelInfo, name string) (DB.IndexDef, bool) {
	for _, idx := range m.Indexes {
		if idx.Name == name {
			return idx, true
		}
	}
	return DB.IndexDef{}, false
}

// idxFields 索引的分量列表 (单列索引只有 FieldID, 复合索引看 FieldIDs)。
func idxFields(d DB.IndexDef) []byte {
	if len(d.FieldIDs) > 0 {
		return d.FieldIDs
	}
	return []byte{d.FieldID}
}

func main() {
	fmt.Println("==============================================")
	fmt.Println("  ORM 唯一约束测试 (vrtx:\"unique\")")
	fmt.Println("==============================================")

	fmt.Println("\n---- 1. 标签布线: tag → 列标志 / 索引唯一位 ----")
	testTagWiring()

	fmt.Println("\n---- 2. 端到端: 写入 / 读回 / 重开表 ----")
	testEndToEnd()

	fmt.Println("\n---- 3. 库里已有同名非唯一索引时必须报错 ----")
	testReopenWithStaleIndex()

	fmt.Println()
	fmt.Println("==============================================")
	fmt.Printf("测试结果: %d 通过, %d 失败\n", passed, failed)
	if failed == 0 {
		fmt.Println("🎉 全部通过!")
	} else {
		fmt.Printf("⚠️  有 %d 项失败\n", failed)
		os.Exit(1)
	}
}

// testTagWiring 先钉住"tag → 列标志 / 索引唯一标记"的映射, 不碰磁盘。
func testTagWiring() {
	m, err := DB.ParseModel(&ormUniq{})
	fatal("ParseModel(ormUniq)", err)
	cols := map[string]DB.FieldDef{}
	for _, f := range m.Fields {
		cols[f.ColName] = f
	}

	check("Name 列带 FlagUnique", cols["Name"].Flag&types.FlagUnique != 0,
		fmt.Sprintf("Flag = %#02x, 缺 FlagUnique", cols["Name"].Flag))
	check("Email 列带 FlagUnique|FlagNullable",
		cols["Email"].Flag&types.FlagUnique != 0 && cols["Email"].Flag&types.FlagNullable != 0,
		fmt.Sprintf("Flag = %#02x, 需要 FlagUnique|FlagNullable", cols["Email"].Flag))
	check("Group 列没有 FlagUnique", cols["Group"].Flag&types.FlagUnique == 0,
		fmt.Sprintf("Flag = %#02x", cols["Group"].Flag))
	// 指针载体与值载体的列型必须一致: 指针只是多了"整列为空"这一档。
	check("*int64 与 int64 推断出同一个列型", cols["Email"].FieldType == cols["Name"].FieldType,
		fmt.Sprintf("*int64 → %d, int64 → %d", cols["Email"].FieldType, cols["Name"].FieldType))

	// 索引侧: 唯一标记落在索引上 (约束的执行者), Name 与 Email 各一个。
	//
	// 这里按【列】找索引, 而不是按名字: ParseModel 此刻还不知道表名, 自动索引名里的
	// 表名部分要等 UseTable 才定下来 (所以名字在端到端那段里核对)。
	uniqByCol := map[string]bool{}
	for _, idx := range m.Indexes {
		fids := idxFields(idx)
		if len(fids) != 1 {
			check("每个索引各 1 个分量", false, fmt.Sprintf("索引 %s 分量数 = %d", idx.Name, len(fids)))
			continue
		}
		uniqByCol[m.AllFields[fids[0]].ColName] = idx.Unique
	}
	for _, col := range []string{"Name", "Email"} {
		check("列 "+col+" 的索引带唯一位", uniqByCol[col], "没有唯一标记")
	}
	check("列 Group 的索引不带唯一位", !uniqByCol["Group"], "被标成了唯一")

	// ===== 复合唯一: unique 写在 A 上, 约束的是 (A,B) 整组 =====
	pm, err := DB.ParseModel(&ormUniqPair{})
	fatal("ParseModel(ormUniqPair)", err)
	u2, ok := findIdx(pm, "u2")
	if !check("复合索引 u2 已生成", ok, "模型里没有 u2") {
		return
	}
	check("u2 是唯一索引 (unique 只写在 A 上, 唯一性属于整组)", u2.Unique, "u2 不是唯一索引")
	check("u2 有 2 个分量", len(idxFields(u2)) == 2,
		fmt.Sprintf("分量数 = %d, want 2", len(idxFields(u2))))

	pc := map[string]DB.FieldDef{}
	for _, f := range pm.Fields {
		pc[f.ColName] = f
	}
	// 位只反映"这一列 tag 里写了 unique": A 有, B 没有。
	//
	// 刻意【不】往 B 上补齐: (A,B) 的唯一键不等于 B 唯一, 给 B 打上唯一位就成了
	// "这一列单独唯一"的假话; 而照位反推 tag 的工具会给 B 补一个 unique, 那在 ORM 里
	// 意味着"给 B 单独建一个唯一索引", 凭空多出一条约束。
	check("A 列带 FlagUnique", pc["A"].Flag&types.FlagUnique != 0,
		fmt.Sprintf("Flag = %#02x", pc["A"].Flag))
	check("B 列没有 FlagUnique (唯一性是整组的)", pc["B"].Flag&types.FlagUnique == 0,
		fmt.Sprintf("Flag = %#02x", pc["B"].Flag))
	check("Note 列带 FlagNullable", pc["Note"].Flag&types.FlagNullable != 0,
		fmt.Sprintf("Flag = %#02x", pc["Note"].Flag))
}

// testEndToEnd 唯一约束在真实的写入 / 读取 / 重开表链路上都必须成立。
func testEndToEnd() {
	db, cleanup := openDemoDB("unique_orm.db")
	defer cleanup()

	tbl, err := DB.UseTable[ormUniq](db, "uniq")
	fatal("UseTable[ormUniq]", err)

	// 列标志与索引标记都要真的落到 core schema 上 (否则约束只是纸面声明)。
	f, ok := coreTbl(db, "uniq").GetActiveFieldFlag("Name")
	check("core schema 的 Name 列带 FlagUnique", ok && f&types.FlagUnique != 0,
		fmt.Sprintf("Flag = %#02x (ok=%v), 缺 FlagUnique", f, ok))
	for _, name := range []string{"idx_uniq_Name", "idx_uniq_Email"} {
		m, found := coreTbl(db, "uniq").IndexMeta().FindIndex(name)
		check("core 里 "+name+" 是唯一索引", found && m.Unique(), fmt.Sprintf("found=%v", found))
	}

	// ===== NULL 免除: 多行 NULL 共存 =====
	for i := int64(1); i <= 3; i++ {
		checkErr(fmt.Sprintf("插入第 %d 行 (Email=NULL)", i),
			tbl.Insert(&ormUniq{Name: i, Email: nil, Group: 100}))
	}

	// ===== 非 NULL 值唯一 =====
	checkErr("插入 Email=7", tbl.Insert(&ormUniq{Name: 4, Email: i64ptr(7), Group: 100}))
	checkDup("第二个 Email=7 被拒", tbl.Insert(&ormUniq{Name: 5, Email: i64ptr(7), Group: 100}))
	// Name 也是唯一列 (非主键): 重复 Name 同样被拒
	checkDup("重复 Name 被拒", tbl.Insert(&ormUniq{Name: 1, Email: nil, Group: 100}))

	// ===== 读回: 可空列按指针载体原样往返 =====
	rows, err := tbl.Where("Email", "=", int64(7)).Find()
	if checkErr("查 Email=7", err) {
		if len(rows) != 1 {
			check("Email=7 命中 1 行", false, fmt.Sprintf("命中 %d 行", len(rows)))
		} else {
			check("Email=7 命中 1 行且指针解回 7",
				rows[0].Email != nil && *rows[0].Email == 7 && rows[0].Name == 4,
				fmt.Sprintf("Email = %v, Name = %d", rows[0].Email, rows[0].Name))
		}
	}
	// NULL 行读回必须是 nil 指针 (不是 0): 0 是"读到了 0", nil 才是"没有值"。
	nulls, err := tbl.Where("Name", "=", int64(1)).Find()
	if checkErr("查 Name=1", err) {
		if len(nulls) != 1 {
			check("Name=1 命中 1 行", false, fmt.Sprintf("命中 %d 行", len(nulls)))
		} else {
			check("NULL 读回是 nil 指针 (不是 0)", nulls[0].Email == nil,
				fmt.Sprintf("Email = %v", nulls[0].Email))
		}
	}

	// ===== 重开表: 约束必须还在 (它随索引元数据落盘) =====
	fatal("Commit", db.Commit())
	tbl2, err := DB.UseTable[ormUniq](db, "uniq")
	if !checkErr("重开表 UseTable[ormUniq]", err) {
		return
	}
	checkDup("重开后插入重复值仍被拒", tbl2.Insert(&ormUniq{Name: 6, Email: i64ptr(7), Group: 100}))
	checkErr("重开后 NULL 行仍然可以进 (约束没把 NULL 也管上)",
		tbl2.Insert(&ormUniq{Name: 6, Email: nil, Group: 100}))
}

// testReopenWithStaleIndex 库里已有同名非唯一索引时, 声明 unique 必须报错。
func testReopenWithStaleIndex() {
	db, cleanup := openDemoDB("unique_orm_stale.db")
	defer cleanup()

	// 先用"不声明 unique"的模型建表 (索引名与分量和 ormUniq 完全一致)
	_, err := DB.UseTable[ormUniqLax](db, "uniq")
	fatal("UseTable[ormUniqLax]", err)
	m, found := coreTbl(db, "uniq").IndexMeta().FindIndex("idx_uniq_Name")
	check("前提: idx_uniq_Name 已存在且非唯一", found && !m.Unique(), fmt.Sprintf("found=%v", found))
	fatal("Commit", db.Commit())

	// 换成声明了 unique 的模型重开: 约束无法生效, 必须当场报错
	_, err = DB.UseTable[ormUniq](db, "uniq")
	check("模型声明 unique 而库里索引不唯一 → 重开报错", err != nil, "竟然静默沿用了非唯一索引")
}