TsDB2/demo/verify_hash_widekey/main.go

// Hash 宽键 (字符串) 索引专项测试 — 验证分裂/合并阈值随桶页容量自适应
//
// 背景
// ---------------------------------------------------------------------------
// 分裂阈值原先固定为 400 条/桶。窄键时这没问题: int64 键的条目 16B, 单桶页可容
// 1022 条, 400 条远未写满。但宽键的桶页容量会骤降 —— maxlen:64 的列条目 72B,
// 单页仅容 227 条; 默认 string (256 字节) 只有 61 条。仍按固定 400 分裂, 平均每桶
// 就要摊成 6~7 页溢出链, 等值查找退化成 O(链长), 与哈希索引的卖点相悖。
// 现改为 min(400, bucketCap*3/4), 合并阈值 = 分裂阈值/4 (保持 400/100 的滞后比)。
//
// 本测试的判据
// ---------------------------------------------------------------------------
// 同一张表、同样的 3000 行, 放两个哈希索引, 只让键宽不同, 用「分裂进度」区分行为:
//
//	idx_tag  maxlen:64 → keyLen=64 → EntryLen=72  → 桶页容量 227  → 阈值 min(400,170)=170
//	                   → 16*170 = 2720 行开始分裂 → 3000 行时必然分裂 2 次 → 桶数 18
//	idx_grp  int64     → keyLen=8  → EntryLen=16  → 桶页容量 1022 → 阈值 400 (上限生效)
//	                   → 16*400 = 6400 行才分裂   → 3000 行时分裂 0 次 → 桶数 16
//
// 若阈值被改回固定 400, idx_tag 的 splitPointer 会退成 0, 核心断言立即失败。
//
// 元数据读取: 演示程序在 core 包外, 拿不到 hash_handler.go 里的未导出偏移常量,
// 因此这里按同一份布局直接读 .db 文件 (页 → 文件偏移: 4096 + (pid-1)*16384)。
// 读之前必须先 db.Commit(): WAL 提交会把脏页 WritePageDirect 落盘并清空缓存。
package main

import (
	"git.ycl.pub/yl32940/TsDB2/DB"
	"git.ycl.pub/yl32940/TsDB2/DB/core"
	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
	"encoding/binary"
	"fmt"
	"os"
	"strings"
	"time"
)

// WideRec 一张表里塞两个哈希索引, 只让键宽不同:
//   - Tag: maxlen:64 → FieldType_String (64B) → hash 索引键 64 字节
//   - Grp: int64     → FieldType_Uint64  (8B)  → hash 索引键  8 字节
type WideRec struct {
	DB.BaseModel
	Tag string `vrtx:"index:idx_tag,type:hash,maxlen:64"`
	Grp int64  `vrtx:"index:idx_grp,type:hash"`
	Pay int64
}

const (
	rowCount   = 3000 // 插满这么多行后看分裂进度
	deleteRows = 2500 // 再删掉这么多行, 看桶合并收缩
	grpMod     = 50   // Grp = i % 50 → 每组恰好 rowCount/grpMod 行
)

// ---------------------------------------------------------------------------
// 哈希索引元数据页布局 (与 DB/core/index/hash_handler.go 的 hashMeta*Off 一致)
//
//	[0,32)   page.IndexHead
//	[32,36)  magic
//	[36,40)  level
//	[40,44)  splitPointer
//	[44,48)  totalEntries
//	[48,52)  dirStartPID
//	[52,56)  dirPageCount
//	[56,60)  keyLen
//
// ---------------------------------------------------------------------------
const (
	offMagic     = page.IndexHeadSize + 0
	offLevel     = page.IndexHeadSize + 4
	offSplit     = page.IndexHeadSize + 8
	offTotal     = page.IndexHeadSize + 12
	offDirStart  = page.IndexHeadSize + 16
	offDirPages  = page.IndexHeadSize + 20
	offKeyLen    = page.IndexHeadSize + 24
	hashPageHdr  = 28                                                               // 桶页头: PageHdr(20) + BucketID(4) + Count(4), next 复用 PageHdr.NextPage
	hashMagicVal = uint32('T')<<24 | uint32('H')<<16 | uint32('H')<<8 | uint32('3') // "THH3"
	pkIDLen      = 8
)

// hashMeta 哈希索引元数据页 (只解析测试要断言的字段)
type hashMeta struct {
	Magic        uint32
	Level        uint32
	SplitPointer uint32
	TotalEntries uint32
	DirStartPID  uint32
	DirPageCount uint32
	KeyLen       uint32
}

// bucketCount 当前可用桶数 = 2^Level + SplitPointer (同 hashMetaData.bucketCount)
func (m hashMeta) bucketCount() uint32 { return (uint32(1) << m.Level) + m.SplitPointer }

// bucketCap 单桶页可容纳条目数 = (页大小 - 桶页头) / (keyLen + 8) (同 hashMetaData.bucketCap)
func (m hashMeta) bucketCap() uint32 {
	entryLen := m.KeyLen + pkIDLen
	if entryLen == 0 {
		return 0
	}
	return (uint32(page.IndexPageSize) - hashPageHdr) / entryLen
}

// splitThreshold 复刻 hashSplitThreshold: min(400, bucketCap*3/4)
func (m hashMeta) splitThreshold() uint32 {
	if t := m.bucketCap() * 3 / 4; t < 400 {
		return t
	}
	return 400
}

// mergeThreshold 复刻 hashMergeThreshold: 分裂阈值 / 4
func (m hashMeta) mergeThreshold() uint32 {
	if t := m.splitThreshold() / 4; t >= 1 {
		return t
	}
	return 1
}

// readHashMeta 绕过缓冲池直接读 .db 文件, 解析哈希索引元数据页。
// 页 → 文件偏移的换算见 DB/core/File.go: PageSize4K + (pageID-1)*len(buf)。
func readHashMeta(path string, pid uint32) (hashMeta, error) {
	f, err := os.Open(path)
	if err != nil {
		return hashMeta{}, err
	}
	defer f.Close()

	buf := make([]byte, core.PageSize)
	offset := int64(core.PageSize4K) + int64(pid-1)*int64(core.PageSize)
	if _, err := f.ReadAt(buf, offset); err != nil {
		return hashMeta{}, fmt.Errorf("读页 %d (文件偏移 %d): %w", pid, offset, err)
	}

	m := hashMeta{
		Magic:        binary.LittleEndian.Uint32(buf[offMagic:]),
		Level:        binary.LittleEndian.Uint32(buf[offLevel:]),
		SplitPointer: binary.LittleEndian.Uint32(buf[offSplit:]),
		TotalEntries: binary.LittleEndian.Uint32(buf[offTotal:]),
		DirStartPID:  binary.LittleEndian.Uint32(buf[offDirStart:]),
		DirPageCount: binary.LittleEndian.Uint32(buf[offDirPages:]),
		KeyLen:       binary.LittleEndian.Uint32(buf[offKeyLen:]),
	}
	if m.Magic != hashMagicVal {
		return m, fmt.Errorf("页 %d 的 magic=%#x (期望 %#x): 该页不是哈希元数据页, "+
			"说明页偏移换算或布局假设有误", pid, m.Magic, hashMagicVal)
	}
	return m, nil
}

// entryPageOf 从索引元信息取哈希索引的元数据页 ID。
// 注意: CreateIndex 时 EntryPage 为 0, 首次 Insert 才分配, 因此必须在插入之后调用。
func entryPageOf(tbl *core.Table, idxName string) (uint32, error) {
	m, ok := tbl.IndexMeta().FindIndex(idxName)
	if !ok {
		return 0, fmt.Errorf("索引 %q 不存在", idxName)
	}
	if m.EntryPage == 0 {
		return 0, fmt.Errorf("索引 %q 的 EntryPage 仍为 0 (需先插入至少一行触发初始化)", idxName)
	}
	return m.EntryPage, nil
}

// tagOf 从回表结果里取 Tag 字段并去掉定长编码的尾部 0 填充。
func tagOf(vals []types.FieldValue) string {
	for _, f := range vals {
		if string(f.FieldName) != "Tag" {
			continue
		}
		if b, ok := f.Value.([]byte); ok {
			return strings.TrimRight(string(b), "\x00")
		}
	}
	return ""
}

// ---------------------------------------------------------------------------
// 测试脚手架
// ---------------------------------------------------------------------------

var (
	dbPath         string
	db             *DB.DB
	passed, failed int
)

func check(msg string, ok bool, detail ...string) {
	if ok {
		passed++
		fmt.Printf("  [PASS] %s\n", msg)
		return
	}
	failed++
	d := ""
	if len(detail) > 0 {
		d = " — " + detail[0]
	}
	fmt.Printf("  [FAIL] %s%s\n", msg, d)
}

func section(t string) { fmt.Printf("\n========== %s\n", t) }

func summary() {
	section("测试总结")
	fmt.Printf("  总计 %d 项, 通过 %d, 失败 %d\n", passed+failed, passed, failed)
	if failed == 0 {
		fmt.Println("  [OK] 宽键哈希索引工作正常, 分裂阈值确实随桶页容量收敛")
	} else {
		fmt.Println("  [FAIL] 存在失败项, 见上")
	}
}

func cleanup() {
	if db != nil {
		_ = db.Close()
		db = nil
	}
	// [KEEP-DB] 保留数据库文件供 Analyzer 分析, 屏蔽运行结束后的清理删除
	// _ = os.Remove(dbPath)
	// _ = os.Remove(dbPath + ".wal")
}

// fatal 前置条件不成立时立即收尾退出 —— 后续断言建立在错误前提上没有意义。
func fatal(msg string, err error) {
	failed++
	fmt.Printf("  [FAIL] %s: %v\n", msg, err)
	summary()
	cleanup()
	os.Exit(1)
}

func main() {
	dbPath = fmt.Sprintf("hash_widekey_%d.db", time.Now().UnixNano())
	_ = os.Remove(dbPath)
	_ = os.Remove(dbPath + ".wal")

	fmt.Println("============================================================")
	fmt.Println("  Hash 宽键索引测试 — 分裂阈值随桶页容量自适应")
	fmt.Println("============================================================")

	d, err := DB.Open(dbPath)
	if err != nil {
		fatal("打开数据库", err)
	}
	db = d
	db.SetAutoCommitDelay(0)

	tbl, err := DB.UseTable[WideRec](db, "wide")
	if err != nil {
		fatal("UseTable[WideRec]", err)
	}
	coreTbl, ok := db.CoreDB().FindTable("wide")
	if !ok {
		fatal("FindTable(wide)", fmt.Errorf("表不存在"))
	}

	// ============================================================
	section(fmt.Sprintf("1. 插入 %d 行 (同表两个哈希索引, 仅键宽不同)", rowCount))

	t0 := time.Now()
	for i := 1; i <= rowCount; i++ {
		if err := tbl.Insert(&WideRec{
			Tag: fmt.Sprintf("TAG%05d", i),
			Grp: int64(i % grpMod),
			Pay: int64(i),
		}); err != nil {
			check(fmt.Sprintf("Insert 第 %d 行", i), false, err.Error())
			break
		}
	}
	if err := db.Commit(); err != nil {
		fatal("Commit", err)
	}
	cnt, _ := tbl.Count()
	check(fmt.Sprintf("插入 %d 行, Count=%d, 耗时 %.2fs", rowCount, cnt, time.Since(t0).Seconds()),
		cnt == int64(rowCount))

	// ============================================================
	section("2. 分裂进度: 宽键已分裂, 窄键未分裂")

	tagPID, err := entryPageOf(coreTbl, "idx_tag")
	if err != nil {
		fatal("取 idx_tag EntryPage", err)
	}
	grpPID, err := entryPageOf(coreTbl, "idx_grp")
	if err != nil {
		fatal("取 idx_grp EntryPage", err)
	}

	tagMeta, err := readHashMeta(dbPath, tagPID)
	if err != nil {
		fatal("读 idx_tag 元数据页", err)
	}
	grpMeta, err := readHashMeta(dbPath, grpPID)
	if err != nil {
		fatal("读 idx_grp 元数据页", err)
	}
	fmt.Printf("  [info] idx_tag EntryPage=%d level=%d split=%d total=%d keyLen=%d 目录页=%d\n",
		tagPID, tagMeta.Level, tagMeta.SplitPointer, tagMeta.TotalEntries, tagMeta.KeyLen, tagMeta.DirPageCount)
	fmt.Printf("  [info] idx_grp EntryPage=%d level=%d split=%d total=%d keyLen=%d 目录页=%d\n",
		grpPID, grpMeta.Level, grpMeta.SplitPointer, grpMeta.TotalEntries, grpMeta.KeyLen, grpMeta.DirPageCount)

	// --- 键宽/容量必须与理论推导一致, 否则后面的阈值推论不成立 ---
	check(fmt.Sprintf("idx_tag keyLen=%d → EntryLen=%d → 桶页容量=%d (期望 64/72/227)",
		tagMeta.KeyLen, tagMeta.KeyLen+pkIDLen, tagMeta.bucketCap()),
		tagMeta.KeyLen == 64 && tagMeta.bucketCap() == 227)
	check(fmt.Sprintf("idx_grp keyLen=%d → EntryLen=%d → 桶页容量=%d (期望 8/16/1022)",
		grpMeta.KeyLen, grpMeta.KeyLen+pkIDLen, grpMeta.bucketCap()),
		grpMeta.KeyLen == 8 && grpMeta.bucketCap() == 1022)

	check(fmt.Sprintf("idx_tag 分裂阈值=%d (min(400, 227*3/4)=170), 合并阈值=%d",
		tagMeta.splitThreshold(), tagMeta.mergeThreshold()),
		tagMeta.splitThreshold() == 170 && tagMeta.mergeThreshold() == 42)
	check(fmt.Sprintf("idx_grp 分裂阈值=%d (1022*3/4=766 > 400, 上限生效), 合并阈值=%d",
		grpMeta.splitThreshold(), grpMeta.mergeThreshold()),
		grpMeta.splitThreshold() == 400 && grpMeta.mergeThreshold() == 100)

	// --- 核心断言: 同样的行数, 宽键分裂、窄键不分裂 ---
	// 宽键: 桶数从 16 起, 阈值 170 → 第 2721 行触发第 1 次分裂 (桶→17, 阈值 2890),
	//       第 2891 行触发第 2 次 (桶→18, 阈值 3060), 之后到 3000 行不再分裂。
	check(fmt.Sprintf("idx_tag %d 行 → level=%d splitPointer=%d 桶数=%d (期望 4/2/18)",
		rowCount, tagMeta.Level, tagMeta.SplitPointer, tagMeta.bucketCount()),
		tagMeta.Level == 4 && tagMeta.SplitPointer == 2 && tagMeta.bucketCount() == 18)
	// 窄键: 16*400 = 6400 行才分裂, 3000 行远未到, 桶数仍是下限 16。
	check(fmt.Sprintf("idx_grp %d 行 → level=%d splitPointer=%d 桶数=%d (期望 4/0/16)",
		rowCount, grpMeta.Level, grpMeta.SplitPointer, grpMeta.bucketCount()),
		grpMeta.Level == 4 && grpMeta.SplitPointer == 0 && grpMeta.bucketCount() == 16)

	check(fmt.Sprintf("两个索引活跃条目数均为 %d", rowCount),
		tagMeta.TotalEntries == rowCount && grpMeta.TotalEntries == rowCount)

	// 桶目录: 18 个槽位远小于单页 4091 槽, 目录必须还在一页内
	check(fmt.Sprintf("idx_tag 桶目录页数=%d (18 个槽位应仍在一页内)", tagMeta.DirPageCount),
		tagMeta.DirPageCount == 1)

	fmt.Printf("\n  [说明] 若阈值仍是固定 400: idx_tag 要 16*400=6400 行才分裂,\n" +
		"         本次 3000 行时 splitPointer 会是 0 —— 上面那条断言即为回归护栏\n")

	// ============================================================
	section("3. 宽键哈希索引正确性 (全量点查 / 分组计数 / 范围扫描)")

	miss := 0
	for i := 1; i <= rowCount; i++ {
		want := fmt.Sprintf("TAG%05d", i)
		key, err := coreTbl.BuildIndexKey("idx_tag", []byte(want))
		if err != nil {
			check("BuildIndexKey", false, err.Error())
			break
		}
		vals, err := coreTbl.SearchByIndex("idx_tag", key)
		if err != nil || len(vals) == 0 || tagOf(vals) != want {
			miss++
		}
	}
	check(fmt.Sprintf("%d 个 64B 键全部点查命中且内容一致 (丢失 %d)", rowCount, miss), miss == 0)

	perGrp := int64(rowCount / grpMod)
	grpBad := 0
	for g := int64(0); g < grpMod; g++ {
		n, err := tbl.Where("Grp", "=", g).Count()
		if err != nil || n != perGrp {
			grpBad++
		}
	}
	check(fmt.Sprintf("%d 个 Grp 分组的计数全部 = %d", grpMod, perGrp), grpBad == 0)

	rangeN, err := tbl.Where("Grp", ">=", int64(10)).Where("Grp", "<=", int64(19)).Count()
	check(fmt.Sprintf("Grp 范围 [10,19] Count=%d (期望 %d; 哈希不保序, 靠全桶扫描+键过滤)",
		rangeN, 10*perGrp), err == nil && rangeN == 10*perGrp)

	// ============================================================
	section(fmt.Sprintf("4. 删除 %d 行 → 桶合并收缩", deleteRows))

	delFail := 0
	for i := uint64(1); i <= uint64(deleteRows); i++ {
		if err := tbl.Delete(i); err != nil {
			delFail++
		}
	}
	if err := db.Commit(); err != nil {
		fatal("Commit", err)
	}
	left, _ := tbl.Count()
	check(fmt.Sprintf("删除 %d 行 (失败 %d), 剩余 Count=%d", deleteRows, delFail, left),
		delFail == 0 && left == int64(rowCount-deleteRows))

	tagMeta2, err := readHashMeta(dbPath, tagPID)
	if err != nil {
		fatal("删除后读 idx_tag 元数据页", err)
	}
	fmt.Printf("  [info] 删除后 idx_tag level=%d split=%d total=%d 桶数=%d\n",
		tagMeta2.Level, tagMeta2.SplitPointer, tagMeta2.TotalEntries, tagMeta2.bucketCount())
	// 合并阈值 42: 桶数 18 时 T<756 合并一次(→17), 17 时 T<714 再合并一次(→16),
	// 此后桶数已到下限 hashMinBucketCount=16, 不再合并。最终 500 行 / 16 桶。
	check(fmt.Sprintf("删除后 idx_tag total=%d splitPointer=%d 桶数=%d (期望 %d/0/16)",
		tagMeta2.TotalEntries, tagMeta2.SplitPointer, tagMeta2.bucketCount(), rowCount-deleteRows),
		tagMeta2.TotalEntries == rowCount-deleteRows &&
			tagMeta2.SplitPointer == 0 && tagMeta2.bucketCount() == 16)

	// ============================================================
	section("5. 关闭重开 (持久化)")

	if err := db.Close(); err != nil {
		fatal("Close", err)
	}
	db = nil

	// 关库后读盘最确定: 此时不可能有未落盘的脏页
	fileMeta, err := readHashMeta(dbPath, tagPID)
	if err != nil {
		fatal("关库后读 idx_tag 元数据页", err)
	}
	check(fmt.Sprintf("关库后磁盘上 idx_tag splitPointer=%d total=%d, 与关库前一致",
		fileMeta.SplitPointer, fileMeta.TotalEntries),
		fileMeta.SplitPointer == tagMeta2.SplitPointer &&
			fileMeta.TotalEntries == tagMeta2.TotalEntries)

	db2, err := DB.Open(dbPath)
	if err != nil {
		fatal("重开数据库", err)
	}
	db = db2
	db.SetAutoCommitDelay(0)
	tbl2, err := DB.UseTable[WideRec](db, "wide")
	if err != nil {
		fatal("重开 UseTable[WideRec]", err)
	}
	coreTbl2, ok := db.CoreDB().FindTable("wide")
	if !ok {
		fatal("重开 FindTable(wide)", fmt.Errorf("表不存在"))
	}

	pid2, err := entryPageOf(coreTbl2, "idx_tag")
	if err != nil {
		fatal("重开取 idx_tag EntryPage", err)
	}
	check(fmt.Sprintf("重开后 idx_tag EntryPage=%d (关闭前 %d)", pid2, tagPID), pid2 == tagPID)

	reopMeta, err := readHashMeta(dbPath, pid2)
	if err != nil {
		fatal("重开后读 idx_tag 元数据页", err)
	}
	check(fmt.Sprintf("重开后 level=%d splitPointer=%d total=%d 与关闭前一致",
		reopMeta.Level, reopMeta.SplitPointer, reopMeta.TotalEntries),
		reopMeta.Level == tagMeta2.Level &&
			reopMeta.SplitPointer == tagMeta2.SplitPointer &&
			reopMeta.TotalEntries == tagMeta2.TotalEntries)

	cnt2, _ := tbl2.Count()
	check(fmt.Sprintf("重开后 Count=%d (期望 %d)", cnt2, rowCount-deleteRows),
		cnt2 == int64(rowCount-deleteRows))

	// 重开后仍要能按宽键点查 (存活的行是 PK 2501..3000)
	reopenMiss := 0
	for i := deleteRows + 1; i <= rowCount; i++ {
		want := fmt.Sprintf("TAG%05d", i)
		key, err := coreTbl2.BuildIndexKey("idx_tag", []byte(want))
		if err != nil {
			reopenMiss++
			continue
		}
		vals, err := coreTbl2.SearchByIndex("idx_tag", key)
		if err != nil || tagOf(vals) != want {
			reopenMiss++
		}
	}
	check(fmt.Sprintf("重开后 %d 个存活行的宽键点查全部命中 (丢失 %d)", rowCount-deleteRows, reopenMiss),
		reopenMiss == 0)

	// 已删的行不该还能查到
	goneKey, _ := coreTbl2.BuildIndexKey("idx_tag", []byte("TAG00001"))
	_, goneErr := coreTbl2.SearchByIndex("idx_tag", goneKey)
	check("重开后已删除的 TAG00001 查不到", goneErr != nil)

	// ============================================================
	summary()
	cleanup()
	if failed > 0 {
		os.Exit(1)
	}
}