package main

// NULL 位图 (types.NullBitmapLen) 的两条主线:
//
//  1. 可空列的写入 / 读取语义, 以及 NULL 参与比较时的三值逻辑 —— NULL 不匹配任何
//     谓词 (不等于 0, 也不算小于), 也不进二级索引;
//  2. schema 演进 (AddField 跨 8 位边界) 后, 各版本旧行仍按自身位图长度正确解析。
//
// 位图放在行尾是硬约束 (时序索引按固定偏移读首列), 因此这里对"字段值"逐列比对,
// 而不是只看行数: 位图位置 / 长度算错一格, 最先坏掉的就是中间列的值。
//
// 运行: cd demo/verify_null_bitmap && go run main.go

import (
	"fmt"
	"os"

	"git.ycl.pub/yl32940/TsDB2/DB/core"
	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
)

const dbPath = "null_bitmap_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; 返回是否通过。
func checkErr(name string, err error) bool {
	if err != nil {
		check(name, false, err.Error())
		return false
	}
	check(name, true, "")
	return true
}

// 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))
}

// checkAllUint 逐列比对一组期望值。
func checkAllUint(prefix string, row []types.FieldValue, want map[string]uint64) {
	for name, w := range want {
		checkUint(fmt.Sprintf("%s.%s = %d", prefix, name, w), row, name, w)
	}
}

// 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))
}

func main() {
	fmt.Println("==============================================")
	fmt.Println("  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. 可空列: 写 NULL → 不参与比较 → 不进索引 ----")
	testNullableColumn(db)

	fmt.Println("\n---- 2. schema 演进: 8 列 → 9 列 → 17 列, 位图按行版本算 ----")
	testSchemaEvolutionAcross8(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)
	}
}

func testNullableColumn(db *core.DB) {
	tbl, err := db.CreateTable("NULLABLE", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("ID")},
		{FieldType: types.FieldType_Uint64, FieldName: []byte("Tag")},
		{FieldType: types.FieldType_Uint64, FieldName: []byte("Note"), Flag: types.FlagNullable},
	})
	fatal("建表 NULLABLE", err)
	fatal("建主键索引", idxErr(tbl.CreateIndex("pk_nullable", page.IndexTypePrimary, []byte{0})))
	// Note 上建二级索引: 用来验证"该列为 NULL 的行不进这个索引"。
	fatal("建二级索引 idx_nullable_note", idxErr(tbl.CreateIndex("idx_nullable_note", page.IndexTypeBTree, []byte{2})))

	pk1, err := tbl.InsertRow([]types.FieldValue{nv("ID", uint64(1)), nv("Tag", uint64(10)), nv("Note", nil)})
	fatal("插入 NULL 行", err)
	pk2, err := tbl.InsertRow([]types.FieldValue{nv("ID", uint64(2)), nv("Tag", uint64(20)), nv("Note", uint64(30))})
	fatal("插入持值行", err)

	// ===== 读取: NULL 字段不出现, 同行的其它字段完好 =====
	row1, err := tbl.FindByPK(pk1)
	if checkErr("读回 NULL 行", err) {
		checkUint("NULL 行的 Tag 完好", row1, "Tag", 10)
		checkAbsent("NULL 行的 Note 不出现", row1, "Note")
	}
	row2, err := tbl.FindByPK(pk2)
	if checkErr("读回持值行", err) {
		checkUint("持值行的 Note = 30", row2, "Note", 30)
	}

	// ===== 比较: NULL 不匹配任何谓词 (SQL 三值逻辑) =====
	// 这三条同时覆盖"索引路径" (Eq 走 Note 索引) 与"全表过滤":
	//   - 若把 NULL 当零字节比: Eq(Note,0) 会命中 NULL 行;
	//   - 若索引把 NULL 行的键当成 8 个零字节: Eq(Note,0) 同样会命中。
	checkCount("Count(Note=0) = 0 (NULL 不等于 0)", tbl, "Note", 0, 0)
	checkCount("Count(Note=30) = 1", tbl, "Note", 30, 1)
	n, err := tbl.Query().Where(core.Lt("Note", uint64(100), types.FieldType(0))).Count()
	check("Count(Note<100) = 1 (NULL 不算小于)", err == nil && n == 1,
		fmt.Sprintf("命中 %d 行 (err=%v), want 1", n, err))

	// ===== 更新: 有值 → NULL, 再读回 =====
	checkErr("把 row2 的 Note 改成 NULL", tbl.UpdateRowByPK(pk2, []types.FieldValue{nv("Note", nil)}))
	row2, err = tbl.FindByPK(pk2)
	if checkErr("改 NULL 后读回 row2", err) {
		checkUint("未提供的列沿用旧值 (Tag = 20)", row2, "Tag", 20)
		checkAbsent("row2.Note 现在是 NULL", row2, "Note")
	}
	checkCount("改成 NULL 后陈旧索引条目已摘掉", tbl, "Note", 30, 0)
}

// testSchemaEvolutionAcross8 8 列(1B 位图) → 9 列(2B) → 17 列(3B),
// 每个版本的旧行永远按自己的位图长度解析; 高位下标 (8 / 16) 的 NULL 也必须落在正确的字节。
func testSchemaEvolutionAcross8(db *core.DB) {
	// v1: ID + A..G (G 可空) = 8 列 → 位图 1 字节
	fields := []types.FieldInfo{{FieldType: types.FieldType_Uint64, FieldName: []byte("ID")}}
	for c := 'A'; c <= 'F'; c++ {
		fields = append(fields, types.FieldInfo{FieldType: types.FieldType_Uint64, FieldName: []byte(string(c))})
	}
	fields = append(fields, types.FieldInfo{
		FieldType: types.FieldType_Uint64, FieldName: []byte("G"), Flag: types.FlagNullable,
	})
	if len(fields) != 8 {
		fatal("v1 应为 8 列", fmt.Errorf("实得 %d 列", len(fields)))
	}
	tbl, err := db.CreateTable("EVOLVE", fields)
	fatal("建表 EVOLVE", err)
	fatal("建主键索引", idxErr(tbl.CreateIndex("pk_evolve", page.IndexTypePrimary, []byte{0})))

	// v1 行: G = NULL (位下标 7, 落在第 1 个字节的最高位)
	pk1, err := tbl.InsertRow([]types.FieldValue{
		nv("ID", uint64(1)), nv("A", uint64(11)), nv("B", uint64(12)), nv("C", uint64(13)),
		nv("D", uint64(14)), nv("E", uint64(15)), nv("F", uint64(16)), nv("G", nil),
	})
	fatal("插入 v1 行 (G=NULL)", err)

	// v2: 加 H (可空) → 9 列 → 位图 2 字节
	fatal("加列 H", tbl.AddField(types.FieldInfo{
		FieldType: types.FieldType_Uint64, FieldName: []byte("H"), Flag: types.FlagNullable,
	}))
	pk2, err := tbl.InsertRow([]types.FieldValue{
		nv("ID", uint64(2)), nv("A", uint64(21)), nv("B", uint64(22)), nv("C", uint64(23)),
		nv("D", uint64(24)), nv("E", uint64(25)), nv("F", uint64(26)), nv("G", uint64(27)), nv("H", nil),
	})
	fatal("插入 v2 行 (H=NULL)", err)
	pk3, err := tbl.InsertRow([]types.FieldValue{
		nv("ID", uint64(3)), nv("A", uint64(31)), nv("B", uint64(32)), nv("C", uint64(33)),
		nv("D", uint64(34)), nv("E", uint64(35)), nv("F", uint64(36)), nv("G", uint64(37)), nv("H", uint64(38)),
	})
	fatal("插入 v2 行 (H=val)", err)

	// v3: 再加 I..P (P 可空) → 17 列 → 位图 3 字节
	for _, name := range []string{"I", "J", "K", "L", "M", "N", "O"} {
		fatal("加列 "+name, tbl.AddField(types.FieldInfo{FieldType: types.FieldType_Uint64, FieldName: []byte(name)}))
	}
	fatal("加列 P", tbl.AddField(types.FieldInfo{
		FieldType: types.FieldType_Uint64, FieldName: []byte("P"), Flag: types.FlagNullable,
	}))

	v3Vals := []types.FieldValue{
		nv("ID", uint64(4)), nv("A", uint64(41)), nv("B", uint64(42)), nv("C", uint64(43)),
		nv("D", uint64(44)), nv("E", uint64(45)), nv("F", uint64(46)), nv("G", uint64(47)),
		nv("H", uint64(48)), nv("I", uint64(49)), nv("J", uint64(50)), nv("K", uint64(51)),
		nv("L", uint64(52)), nv("M", uint64(53)), nv("N", uint64(54)), nv("O", uint64(55)), nv("P", nil),
	}
	pk4, err := tbl.InsertRow(v3Vals)
	fatal("插入 v3 行 (P=NULL)", err)
	v3Vals2 := make([]types.FieldValue, len(v3Vals))
	copy(v3Vals2, v3Vals)
	for i := range v3Vals2 {
		if string(v3Vals2[i].FieldName) == "P" {
			v3Vals2[i] = nv("P", uint64(58))
		}
	}
	pk5, err := tbl.InsertRow(v3Vals2)
	fatal("插入 v3 行 (P=val)", err)

	// ===== 逐版本读回, 逐列比对值 (位图长度 / 位置算错都会在这里暴露) =====
	row1, err := tbl.FindByPK(pk1)
	if checkErr("读回 v1 行", err) {
		checkAllUint("v1", row1, map[string]uint64{
			"ID": 1, "A": 11, "B": 12, "C": 13, "D": 14, "E": 15, "F": 16,
		})
		checkAbsent("v1.G 是位下标 7 的 NULL", row1, "G") // 位下标 7
		checkAbsent("v1 还没有 H 列", row1, "H")
	}

	row2, err := tbl.FindByPK(pk2)
	if checkErr("读回 v2 行 (H=NULL)", err) {
		checkUint("v2.G = 27", row2, "G", 27)
		checkAbsent("v2.H 是位下标 8 的 NULL (第 2 个字节 bit0)", row2, "H")
	}

	row3, err := tbl.FindByPK(pk3)
	if checkErr("读回 v2 行 (H=val)", err) {
		checkUint("v2.G = 37", row3, "G", 37)
		checkUint("v2.H = 38", row3, "H", 38)
	}

	row4, err := tbl.FindByPK(pk4)
	if checkErr("读回 v3 行 (P=NULL)", err) {
		checkAllUint("v3", row4, map[string]uint64{
			"G": 47, "H": 48, "I": 49, "J": 50, "K": 51, "L": 52, "M": 53, "N": 54, "O": 55,
		})
		checkAbsent("v3.P 是位下标 16 的 NULL (第 3 个字节 bit0)", row4, "P")
	}

	row5, err := tbl.FindByPK(pk5)
	if checkErr("读回 v3 行 (P=val)", err) {
		checkUint("v3.O = 55", row5, "O", 55)
		checkUint("v3.P = 58", row5, "P", 58)
	}

	// 全表扫一遍: 版本混杂时行数不能丢, 也不能因某版本解析出错而少行。
	rows, err := tbl.Query().Find()
	check("全表 5 行 (版本混杂也不丢行)", err == nil && len(rows) == 5,
		fmt.Sprintf("行数 = %d (err=%v), want 5", len(rows), err))

	// 位图长度必须按行版本算: 这里显式核对三个版本各自的字节数。
	for _, tc := range []struct{ n, bitmapBytes int }{{8, 1}, {9, 2}, {17, 3}} {
		got := types.NullBitmapLen(tc.n)
		check(fmt.Sprintf("NullBitmapLen(%d) = %d 字节", tc.n, tc.bitmapBytes), got == tc.bitmapBytes,
			fmt.Sprintf("实得 %d", got))
	}
}