TsDB2/demo/unique_index_test/main.go

package main

// 唯一索引 (IndexMeta.Flag 的 IndexFlagUnique 位) 的四条不变式。
//
//  1. 同一索引键只允许一行 —— 重复插入 / 改出重复都必须报 ErrDuplicatedKey;
//  2. NULL 免除 —— 任一分量为 NULL 的行不进该索引, 因此多行 NULL 共存;
//  3. 冲突必须在【任何写入之前】拦下 —— 被拒的行不能在行存储或别的索引里留下痕迹,
//     被拒的更新不能让旧行变成"一半新一半旧";
//  4. 建索引要校验存量 —— 已有重复值时唯一索引根本建不出来 (报错并摘掉元数据)。
//
// 第 3 条是这里最花篇幅的部分: 唯一性是"跨行"的约束, 而写入是逐索引进行的, 天然
// 存在"查完才写"与"边写边查"两种实现 —— 后者在冲突时会把前几个索引写坏。
//
// 约束的执行者是【索引】, 列上的 types.FlagUnique 只是声明; 判定方式是写入前拿整条
// 索引键探一次, 命中"别的行"即冲突 (命中本行自己的条目不算, 否则更新自身原值都会失败)。
//
// 运行: cd demo/unique_index_test && go run main.go

import (
	"errors"
	"fmt"
	"os"

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

const dbPath = "unique_index_test.db"

var passed, failed int

// check 记一笔断言结果。
func check(name string, ok bool, detail string) {
	if ok {
		passed++
		fmt.Printf("  ✅ %s\n", name)
		return
	}
	failed++
	fmt.Printf("  ❌ %s — %s\n", name, detail)
}

// checkErr 断言 err == nil; 返回是否通过 (setup 失败就不必往下走)。
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)
	}
}

// idxErr 把 CreateIndex 的 (meta, err) 收敛成 err。
func idxErr(_ *core.IndexMeta, err error) error { return err }

// nv 构造一个 FieldValue; value 传 nil 即"显式写 NULL"。
func nv(name string, value interface{}) types.FieldValue {
	return types.FieldValue{FieldName: []byte(name), Value: value}
}

// nvLookup 在结果行里找字段, 返回 (值, 是否存在)。
func nvLookup(row []types.FieldValue, name string) (interface{}, bool) {
	for _, fv := range row {
		if string(fv.FieldName) == name {
			return fv.Value, true
		}
	}
	return nil, false
}

// checkUint 断言结果行里 field = want。
func checkUint(name string, row []types.FieldValue, field string, want uint64) {
	v, ok := nvLookup(row, field)
	if !ok {
		check(name, false, fmt.Sprintf("字段 %s 缺失 (want %d)", field, want))
		return
	}
	got, ok := v.(uint64)
	check(name, ok && got == want, fmt.Sprintf("字段 %s = %v (%T), want %d", field, v, v, want))
}

// checkAbsent 断言字段没有出现在结果里 (NULL 的既定表示: 不 append)。
func checkAbsent(name string, row []types.FieldValue, field string) {
	v, ok := nvLookup(row, field)
	check(name, !ok, fmt.Sprintf("字段 %s 应为 NULL 而不出现, 实得 %v", field, v))
}

// checkBytes 断言结果行里 field 的字节内容 = want。
func checkBytes(name string, row []types.FieldValue, field string, want string) {
	v, ok := nvLookup(row, field)
	if !ok {
		check(name, false, fmt.Sprintf("字段 %s 缺失 (want %q)", field, want))
		return
	}
	b, ok := v.([]byte)
	check(name, ok && string(b) == want, fmt.Sprintf("字段 %s = %v (%T), want %q", field, v, v, want))
}

// checkCount 断言某列等值命中 want 行 (走该列的索引, 顺便验证索引本身是准的)。
func checkCount(name string, tbl *core.Table, field string, v uint64, want int64) {
	n, err := tbl.Query().Where(core.Eq(field, v, types.FieldType(0))).Count()
	check(name, err == nil && n == want,
		fmt.Sprintf("%s=%d 命中 %d 行 (err=%v), want %d", field, v, n, err, want))
}

// insert 显式指定 pkID 插入一行。
func insert(tbl *core.Table, pkID uint64, values []types.FieldValue) error {
	_, err := tbl.InsertRowWithID(pkID, values)
	return err
}

func main() {
	fmt.Println("==============================================")
	fmt.Println("  唯一索引测试 (IndexMeta.Flag 唯一位 / NULL 免除)")
	fmt.Println("==============================================")

	os.Remove(dbPath)
	os.Remove(dbPath + ".wal")

	db, err := core.Open(dbPath)
	if err != nil {
		panic(err)
	}
	db.SetAutoCommitDelay(0) // 0 = 关闭自动提交, 只在显式 Commit 时落盘

	fmt.Println("\n---- 1. 重复值: 插入 / 更新都必须被拒, 且不留痕迹 ----")
	testRejectsDuplicateInsert(db)

	fmt.Println("\n---- 2. 被拒的更新不能动旧行 (含动态列) ----")
	testRejectedUpdateKeepsOldRow(db)

	fmt.Println("\n---- 3. NULL 免除: 多行 NULL 共存, 非 NULL 仍唯一 ----")
	testNullExemption(db)

	fmt.Println("\n---- 4. 复合唯一键: 约束的是整组分量 ----")
	testCompositeKey(db)

	fmt.Println("\n---- 5. 存量已有重复值时建不出唯一索引 ----")
	testBackfillOnExistingDuplicates(db)

	fmt.Println("\n---- 6. 行存储索引 (主键 / 时序) 不接受唯一约束 ----")
	testRowStoreRejectsUnique(db)

	_ = db.Close()

	fmt.Println()
	fmt.Println("==============================================")
	fmt.Printf("测试结果: %d 通过, %d 失败\n", passed, failed)
	if failed == 0 {
		fmt.Println("🎉 全部通过!")
		os.Remove(dbPath)
		os.Remove(dbPath + ".wal")
	} else {
		fmt.Printf("⚠️  有 %d 项失败 (库文件保留在 %s 便于排查)\n", failed, dbPath)
		os.Exit(1)
	}
}

// uniqTable 建一张 ID/Tag/Grp 三列表: Tag 唯一索引, Grp 普通二级索引。
//
// Grp 这条普通索引是【半写探针】: 若唯一性检查发生在写入循环中途, 顺序在它之前的
// 索引已经写入, 于是 Grp 索引里会留下一个指向不存在行的条目, 按 Grp 一查就是幽灵行。
func uniqTable(db *core.DB) *core.Table {
	tbl, err := db.CreateTable("UNIQIDX", []types.FieldInfo{
		{FieldType: types.FieldType_8byte, FieldName: []byte("ID")},
		{FieldType: types.FieldType_8byte, FieldName: []byte("Tag")},
		{FieldType: types.FieldType_8byte, FieldName: []byte("Grp")},
	})
	fatal("建表 UNIQIDX", err)
	fatal("建主键索引 pk_uniq", idxErr(tbl.CreateIndex("pk_uniq", page.IndexTypePrimary, []byte{0})))
	fatal("建唯一索引 idx_uniq_tag", idxErr(tbl.CreateIndex("idx_uniq_tag", page.IndexTypeBTree, []byte{1}, true)))
	fatal("建普通索引 idx_uniq_grp", idxErr(tbl.CreateIndex("idx_uniq_grp", page.IndexTypeBTree, []byte{2})))
	return tbl
}

// uniqRow ID/Tag/Grp 一行。
func uniqRow(id, tag, grp uint64) []types.FieldValue {
	return []types.FieldValue{nv("ID", id), nv("Tag", tag), nv("Grp", grp)}
}

func testRejectsDuplicateInsert(db *core.DB) {
	tbl := uniqTable(db)

	checkErr("插入 row1 (Tag=10)", insert(tbl, 1, uniqRow(1, 10, 100)))

	// 重复 Tag → 拒绝
	checkDup("重复 Tag 插入被拒", insert(tbl, 2, uniqRow(2, 10, 200)))

	// 被拒的行不能留下痕迹: 行存储里没有, 普通索引里也不能有幽灵条目。
	_, err := tbl.FindByPK(2)
	check("被拒的行不在行存储里 (冲突必须早于任何写入)", err != nil, "FindByPK(2) 竟然找到了行")
	checkCount("被拒的行没在普通索引里留幽灵条目", tbl, "Grp", 200, 0)
	if n, err := tbl.Query().Count(); err != nil || n != 1 {
		check("表里仍只有 1 行", false, fmt.Sprintf("行数 = %d (err=%v), want 1", n, err))
	} else {
		check("表里仍只有 1 行", true, "")
	}

	// ===== 更新: 改成已被占用的值 → 拒绝, 且原值原封不动 =====
	checkErr("把 row1 的 Tag 改成 20 (释放 10)", tbl.UpdateRowByPK(1, []types.FieldValue{nv("Tag", uint64(20))}))
	checkErr("Tag=10 释放后可以插入", insert(tbl, 2, uniqRow(2, 10, 200)))
	checkDup("把 row2 的 Tag 改成 20 (撞 row1) 被拒", tbl.UpdateRowByPK(2, []types.FieldValue{nv("Tag", uint64(20))}))

	row2, err := tbl.FindByPK(2)
	if checkErr("读回 row2", err) {
		checkUint("被拒后 row2.Tag 仍是 10 (不能删了旧的没写新的)", row2, "Tag", 10)
		checkUint("被拒后 row2.Grp 也没动", row2, "Grp", 200)
	}

	// 改成自己的原值不算冲突 (命中的是本行自己的索引条目)
	checkErr("改回自身原值不该报冲突", tbl.UpdateRowByPK(2, []types.FieldValue{nv("Tag", uint64(10))}))
	// 换成没人占用的值 → 应成功, 且旧键要从索引里消失
	checkErr("把 row2 的 Tag 改成 30", tbl.UpdateRowByPK(2, []types.FieldValue{nv("Tag", uint64(30))}))
	checkCount("离开的旧键已从索引里摘掉", tbl, "Tag", 10, 0)
	checkCount("新键已进索引", tbl, "Tag", 30, 1)

	// 全表走一遍: 值必须与索引一致
	rows, err := tbl.Query().Find()
	if !checkErr("全表扫描", err) {
		return
	}
	tags := map[uint64]int{}
	for _, r := range rows {
		v, _ := nvLookup(r, "Tag")
		tags[v.(uint64)]++
	}
	check("全表 Tag 分布 = {20:1, 30:1}", len(rows) == 2 && tags[20] == 1 && tags[30] == 1,
		fmt.Sprintf("实际 %v (%d 行)", tags, len(rows)))
}

// testRejectedUpdateKeepsOldRow 被拒的更新必须"要么全改, 要么全不动"。
//
// Body 是动态列 (堆块), 它正是这条不变式最脆弱的地方: serializeRow 会就地覆写旧行的
// 堆块 (heapRewrite) 甚至 heapUnref 掉旧块, 而索引冲突是在那之后才由 UpdateByKey 发现。
// 若预检不放序列化之前, 这里读回的 Body 就会是新值 (极端情况是已释放的块)。
func testRejectedUpdateKeepsOldRow(db *core.DB) {
	tbl, err := db.CreateTable("UNIQDYN", []types.FieldInfo{
		{FieldType: types.FieldType_8byte, FieldName: []byte("ID")},
		{FieldType: types.FieldType_8byte, FieldName: []byte("Tag")},
		{FieldType: types.FieldType_Dynamicbyte, FieldName: []byte("Body")},
	})
	fatal("建表 UNIQDYN", err)
	fatal("建主键索引", idxErr(tbl.CreateIndex("pk_uniqdyn", page.IndexTypePrimary, []byte{0})))
	fatal("建唯一索引", idxErr(tbl.CreateIndex("idx_uniqdyn_tag", page.IndexTypeBTree, []byte{1}, true)))

	checkErr("插入 row1 (Tag=10, Body=aaa)", insert(tbl, 1, []types.FieldValue{
		nv("ID", uint64(1)), nv("Tag", uint64(10)), nv("Body", []byte("aaa")),
	}))
	// row2 的 Body 刻意留长: 这次被拒的更新里新值更短, 会命中 heapRewrite 的原地覆写
	// 分支 (oldLen >= newLen) —— 若预检晚于 serializeRow, 旧行的堆块内容会立刻被改成
	// 新值, 读回来就是脏的 (新值更长时走换块, 表现为悬空引用, 要等那块内存被重新分配
	// 出去才露出来 —— 更难测, 所以这里选"变短")。
	checkErr("插入 row2 (Tag=20, Body=bbbbb)", insert(tbl, 2, []types.FieldValue{
		nv("ID", uint64(2)), nv("Tag", uint64(20)), nv("Body", []byte("bbbbb")),
	}))

	// 同时改 Tag (撞车) 与 Body: 冲突必须让整次更新作废, Body 也得原样。
	checkDup("同时改 Tag(撞车) 与 Body 被拒",
		tbl.UpdateRowByPK(2, []types.FieldValue{nv("Tag", uint64(10)), nv("Body", []byte("c"))}))

	row2, err := tbl.FindByPK(2)
	if !checkErr("读回 row2", err) {
		return
	}
	checkUint("被拒后 row2.Tag 仍是 20", row2, "Tag", 20)
	checkBytes("被拒后 row2.Body 仍是 bbbbb (没被就地覆写)", row2, "Body", "bbbbb")
}

func testNullExemption(db *core.DB) {
	tbl, err := db.CreateTable("UNIQNULL", []types.FieldInfo{
		{FieldType: types.FieldType_8byte, FieldName: []byte("ID")},
		{FieldType: types.FieldType_8byte, FieldName: []byte("Note"), Flag: types.FlagNullable},
	})
	fatal("建表 UNIQNULL", err)
	fatal("建主键索引", idxErr(tbl.CreateIndex("pk_uniqnull", page.IndexTypePrimary, []byte{0})))
	fatal("建唯一索引", idxErr(tbl.CreateIndex("idx_uniqnull_note", page.IndexTypeBTree, []byte{1}, true)))

	// 三行 NULL: 唯一索引一条条目都没有 → 随便多少行都行
	for i := uint64(1); i <= 3; i++ {
		checkErr(fmt.Sprintf("插入第 %d 行 (Note=NULL)", i),
			insert(tbl, i, []types.FieldValue{nv("ID", i), nv("Note", nil)}))
	}
	checkCount("NULL 不等于 0 (Note=0 命中 0 行)", tbl, "Note", 0, 0)

	// 非 NULL 值仍然唯一: 第一行 7 成功, 第二行 7 拒绝
	checkErr("插入 Note=7", insert(tbl, 4, []types.FieldValue{nv("ID", uint64(4)), nv("Note", uint64(7))}))
	checkDup("第二个 Note=7 被拒",
		insert(tbl, 5, []types.FieldValue{nv("ID", uint64(5)), nv("Note", uint64(7))}))

	// NULL 行改成 7 → 也撞车 (NULL 行不在索引里, 但 7 已被占)
	checkDup("NULL → 7 被拒 (7 已被占)",
		tbl.UpdateRowByPK(1, []types.FieldValue{nv("Note", uint64(7))}))
	row1, err := tbl.FindByPK(1)
	if checkErr("读回 row1", err) {
		checkAbsent("被拒后 row1.Note 仍是 NULL", row1, "Note")
	}

	// 持值行改成 NULL → 成功, 索引条目必须同步摘掉
	checkErr("7 → NULL 成功", tbl.UpdateRowByPK(4, []types.FieldValue{nv("Note", nil)}))
	checkCount("值改成 NULL 后索引条目已摘掉", tbl, "Note", 7, 0)
	checkErr("7 腾出来后新的 7 可以进",
		insert(tbl, 6, []types.FieldValue{nv("ID", uint64(6)), nv("Note", uint64(7))}))
	checkCount("Note=7 命中 1 行", tbl, "Note", 7, 1)
}

// testCompositeKey 复合唯一索引: 唯一性作用于【整组分量】。
//
// 与单列版的差别正是最容易写错的地方:
//   - 只有一个分量不同就是不同的键 (10,200) 与 (10,100) 可以共存;
//   - 整组相同才算撞车, 且撞车判定要用【整个键】去探, 漏掉尾部分量就会把
//     "A 相同"误判成重复, 白白拒掉合法数据;
//   - NULL 免除同样按"任一分量为 NULL"生效: (10,NULL) 的行整行不进索引, 可以有多行。
func testCompositeKey(db *core.DB) {
	tbl, err := db.CreateTable("UNIQCOMP", []types.FieldInfo{
		{FieldType: types.FieldType_8byte, FieldName: []byte("ID")},
		{FieldType: types.FieldType_8byte, FieldName: []byte("A")},
		{FieldType: types.FieldType_8byte, FieldName: []byte("B"), Flag: types.FlagNullable},
	})
	fatal("建表 UNIQCOMP", err)
	fatal("建主键索引", idxErr(tbl.CreateIndex("pk_uniqcomp", page.IndexTypePrimary, []byte{0})))
	m, err := tbl.CreateIndex("idx_uniqcomp_ab", page.IndexTypeBTree, []byte{1, 2}, true)
	fatal("建复合唯一索引", err)
	check("复合索引带着唯一位 (2 个分量)", m.Unique() && m.FieldCnt == 2,
		fmt.Sprintf("Unique=%v FieldCnt=%d, want true/2", m.Unique(), m.FieldCnt))

	ab := func(id uint64, a uint64, b interface{}) []types.FieldValue {
		return []types.FieldValue{nv("ID", id), nv("A", a), nv("B", b)}
	}

	// 尾部分量不同 → 不同的键, 都该进得去
	checkErr("插入 (A=10,B=100)", insert(tbl, 1, ab(1, 10, uint64(100))))
	checkErr("插入 (A=10,B=200): 与 (10,100) 是不同键", insert(tbl, 2, ab(2, 10, uint64(200))))
	checkErr("插入 (A=20,B=100): 与 (10,100) 是不同键", insert(tbl, 3, ab(3, 20, uint64(100))))
	// 整组相同 → 撞车
	checkDup("重复 (10,100) 被拒", insert(tbl, 4, ab(4, 10, uint64(100))))
	// 任一分量 NULL → 整行不进索引, 多行共存
	checkErr("插入 (A=10,B=NULL)", insert(tbl, 4, ab(4, 10, nil)))
	checkErr("第二行 (A=10,B=NULL) 不冲突 (NULL 免除)", insert(tbl, 5, ab(5, 10, nil)))

	// 更新到已被占用的整组键 → 拒绝; 改 A 或 B 任一即可避开
	checkDup("更新成 (10,100) 被拒",
		tbl.UpdateRowByPK(3, []types.FieldValue{nv("A", uint64(10)), nv("B", uint64(100))}))
	row3, err := tbl.FindByPK(3)
	if checkErr("读回 row3", err) {
		checkUint("被拒后 row3.A 仍是 20", row3, "A", 20)
		checkUint("被拒后 row3.B 仍是 100", row3, "B", 100)
	}
	// 只改 B: (20,300) 无人占用 → 成功
	checkErr("更新 (20,100) → (20,300)",
		tbl.UpdateRowByPK(3, []types.FieldValue{nv("B", uint64(300))}))

	// A=10 走这个复合索引只能找到 2 行: 另外两行 B 是 NULL, 整行压根不在索引里。
	// 这正是"NULL 免除"的物理表现 —— 它们只能靠全表扫描 / 主键找到。
	checkCount("A=10 经复合索引命中 2 行 (B=NULL 的两行不在索引里)", tbl, "A", 10, 2)

	rows, err := tbl.Query().Find()
	if !checkErr("全表扫描", err) {
		return
	}
	check("总行数 = 5 (两次被拒的插入/更新都没留下痕迹)", len(rows) == 5,
		fmt.Sprintf("行数 = %d, want 5", len(rows)))
}

// testBackfillOnExistingDuplicates 存量已有重复值时, 唯一索引必须建不出来。
//
// 硬建下去会得到一个"违反自身约束"的索引 (同一键挂两行), 点查可能返回其中任意一行,
// 而索引看起来完全正常 —— 这类坏索引无法从外部察觉, 只能在建立时拦下。
func testBackfillOnExistingDuplicates(db *core.DB) {
	tbl, err := db.CreateTable("UNIQBAK", []types.FieldInfo{
		{FieldType: types.FieldType_8byte, FieldName: []byte("ID")},
		{FieldType: types.FieldType_8byte, FieldName: []byte("Tag")},
	})
	fatal("建表 UNIQBAK", err)
	fatal("建主键索引", idxErr(tbl.CreateIndex("pk_uniqbak", page.IndexTypePrimary, []byte{0})))

	// 先插两行同值 (此时还没有唯一索引, 允许)
	checkErr("插入 row1 (Tag=10)", insert(tbl, 1, []types.FieldValue{nv("ID", uint64(1)), nv("Tag", uint64(10))}))
	checkErr("插入 row2 (Tag=10)", insert(tbl, 2, []types.FieldValue{nv("ID", uint64(2)), nv("Tag", uint64(10))}))

	_, err = tbl.CreateIndex("idx_uniqbak_tag", page.IndexTypeBTree, []byte{1}, true)
	checkDup("存量重复时建唯一索引被拒", err)
	// 失败的索引不能留在元数据里 (半个索引比没有索引更危险: 查询会选中它)
	_, found := tbl.IndexMeta().FindIndex("idx_uniqbak_tag")
	check("建立失败的唯一索引没留在元数据里", !found, "索引仍在元数据里")

	// 把重复行删掉之后就该能建成功
	checkErr("删掉 row2", tbl.DeleteRowByPK(2))
	fatal("清掉重复后建唯一索引", idxErr(tbl.CreateIndex("idx_uniqbak_tag", page.IndexTypeBTree, []byte{1}, true)))
	// 建好之后约束立即生效
	checkDup("回填后插入重复值被拒", insert(tbl, 3, []types.FieldValue{nv("ID", uint64(3)), nv("Tag", uint64(10))}))
}

// testRowStoreRejectsUnique 行存储索引不接受唯一约束 —— 必须报错而不是静默忽略。
//
// 主键本就唯一, 时序索引的多行同刻是【刻意允许】的。静默忽略这个标记最坏: 调用方
// 以为约束生效了, 直到插入同刻第二行成功才发现不对, 那时数据已经写进去了。
func testRowStoreRejectsUnique(db *core.DB) {
	tbl, err := db.CreateTable("UNIQRS", []types.FieldInfo{
		{FieldType: types.FieldType_8byte, FieldName: []byte("ID")},
	})
	fatal("建表 UNIQRS", err)

	_, err = tbl.CreateIndex("pk_uniqrs", page.IndexTypePrimary, []byte{0}, true)
	check("行存储索引标唯一被拒 (ErrUniqueRowStore)", errors.Is(err, core.ErrUniqueRowStore),
		fmt.Sprintf("err = %v, want ErrUniqueRowStore", err))

	// 非唯一仍然可以建
	fatal("普通主键索引", idxErr(tbl.CreateIndex("pk_uniqrs", page.IndexTypePrimary, []byte{0})))
	m, found := tbl.IndexMeta().FindIndex("pk_uniqrs")
	check("普通主键索引没被标成唯一", found && !m.Unique(), fmt.Sprintf("found=%v", found))
}