package main

// ============================================================================
// 索引内容一致性 —— 页归属只能看出"页放错了地方", 看不出"页没放错、内容却错位"。
//
// 对照 SQLite 的 PRAGMA integrity_check (逐条比对"索引项 ↔ 表行") 与
// PG 的 amcheck (校验 B+ 树键序 / 分隔键), 这里补上三条:
//
//	(1) 键序单调        条目键必须按引擎的查找前提有序 —— 见下方"检查范围"。
//	(2) 悬空二级索引    二级索引条目尾 8B 是 pkID, 必须能在本表数据页里回表到
//	                   "存活行"; 否则就是指向已不存在行的索引项 (删行 / 重建
//	                   索引漏删的典型残留)。
//	(3) 索引指向非数据页 行存储索引(主键 / 时序)条目尾 4B 是 DataPageID,
//	                   该页类型必须真的是数据页 (DataIndex=2)。
//	(4) 反向缺失        每个"存活行"都必须在每个索引里留下自己的条目。
//	                   前三条只证明"索引里没有废条目", 证明不了"该有的条目都在":
//	                   更新时"删旧键/插新键"只做了一半、或建索引漏了一批行,
//	                   只要不留废条目就完全静默 —— 必须反过来从行出发查一次。
//
// (4) 需要从行字节反算"这一行在某索引里该有的键", 所以复刻了引擎的取键函数
// DB/core/index_meta_page.go:extractIndexKey (按 FieldIDs 顺序定长拼接分量字段,
// 行内偏移按该行自己的 schemaVer 沿 sv.FieldIndices 推)。返回 nil 的语义也一致:
// 分量列在该行版本里缺失 → 引擎本来就跳过这一行, 我们也不能判它缺失。
//
// 条目布局逐条对照引擎 handler, 不做推测:
//	Primary (btree)   叶 [pkID(8B BE)][DataPageID(4B BE)]          el=12
//	Time    (扁平链)  叶 [BeginTime(8B)][LastTime(8B)][PageID(4B)] el=20
//	                           —— timeindex.findDataPageByTime
//	BTree/Str/Compose 叶 [key(keyLen)][pkID(8B BE)]                 el=keyLen+8
//	                           —— btree.SBT* (Compose 处理器底层即 SBT 二级 B+ 树)
//	Hash    (桶链)    条目 [key(keyLen)][pkID(8B BE, 最高位=删除标记)]
//	                           —— hash_handler 桶页 [PageHdr(20)][BucketID(4)][Count(4)]
//
// "有序"这条前提是被引擎的查找算法强行要求的, 不是我们的审美:
//	btree 按分隔键下行            → 叶层全局有序;
//	SBT (BTree/Str/Compose) 同构  → 叶层全局有序;
//	timeindex.findDataPageByTime 对页内条目二分 → 页内按 BeginTime 有序。
// 因此检查范围按"算法真正依赖的范围"划定:
//	B+ 树 (Primary/BTree/Str/Compose): 页内 + 跨页叶链都查;
//	Time:                              只查页内 (分段链按时间有序追加, 页内二分定位);
//	Hash:                              桶内无序, 不查。
//
// 注意: 时序条目的 LastTime 会随追加而增长, 所以键只取 BeginTime 前 8 字节。
// ============================================================================

import (
	"cmp"
	"bytes"
	"encoding/binary"
	"fmt"
	"hash/maphash"
	"runtime"
	"slices"
	"sync"
)

// ===== 并行设施 =====

// runParallel 并发跑 n 个互不相干的任务 (参数是任务下标)。
// 并发度 = min(n, NumCPU); n<=1 时直接顺序跑, 不额外起 goroutine。
func runParallel(n int, fn func(int)) {
	if n <= 0 {
		return
	}
	w := runtime.NumCPU()
	if w > n {
		w = n
	}
	if w <= 1 {
		for i := 0; i < n; i++ {
			fn(i)
		}
		return
	}
	var wg sync.WaitGroup
	jobs := make(chan int, n)
	for i := 0; i < n; i++ {
		jobs <- i
	}
	close(jobs)
	wg.Add(w)
	for k := 0; k < w; k++ {
		go func() {
			defer wg.Done()
			for i := range jobs {
				fn(i)
			}
		}()
	}
	wg.Wait()
}

// ===== 条目存在性集合 =====

// pairSet 二级索引的"条目存在集"。
//
// 原来用 map[string]bool, 键是整条条目字节。千万行级别下这就是 4000 万次字符串
// 分配 —— 既吃内存又让 GC 反复扫描 (string 含指针, 整个 map 都在 GC 的扫描范围内)。
// 换成定长指纹: 键不含指针, GC 不用管, 也不再为每条分配字符串。
//
// 指纹取 64 位而非 128 位: Go map 的桶内存按 sizeof(key) 线性增长, [2]uint64 桶
// 每桶约 152B / 装 6.5 条 ≈ 23B/条, 换成 uint64 后约 88B / 6.5 ≈ 13.5B/条 ——
// 三个千万级二级索引合计省下约 300MB。代价是理论上会指纹碰撞 (表现为"漏报一条
// missing-entry"): 一千万条时单次查询碰撞率约 1e-7, 对审计结论无实质影响。
type pairSet map[uint64]bool

var pairSeedA = maphash.MakeSeed()

// pairKey 条目字节 -> 64 位指纹 (前向侧: 条目本身就是 键||pkID(8B BE))。
func pairKey(ent []byte) uint64 {
	return maphash.Bytes(pairSeedA, ent)
}

// pairKey2 (键, pkID) -> 64 位指纹, 语义等于 pairKey(键||pkID大端8字节),
// 但不真的拼出中间字节串 —— 反向检查每行每索引都要算一次, 省掉上亿次分配。
func pairKey2(key []byte, pk uint64) uint64 {
	var h maphash.Hash
	h.SetSeed(pairSeedA)
	var b8 [8]byte
	binary.BigEndian.PutUint64(b8[:], pk)
	h.Write(key)
	h.Write(b8[:])
	return h.Sum64()
}

// IndexIssue 一条索引内容一致性问题。
type IndexIssue struct {
	Table  string `json:"table"`
	Index  string `json:"index"`
	IType  string `json:"itype"`
	Kind   string `json:"kind"` // out-of-order / orphan-pk / not-data-page
	Page   uint32 `json:"page"`
	At     int    `json:"at"` // 页内条目序号
	Detail string `json:"detail"`
}

// IndexStat 单个索引的扫描统计。
type IndexStat struct {
	Table   string `json:"table"`
	Index   string `json:"index"`
	IType   string `json:"itype"`
	Entries int    `json:"entries"` // 扫到的条目数
	Issues  int    `json:"issues"`
}

const idxIssueCap = 200

func idxTypeName(it byte) string {
	switch it {
	case IT_TIME:
		return "Time"
	case IT_PRIMARY:
		return "Primary"
	case IT_STR:
		return "Str"
	case IT_COMPOSE:
		return "Compose"
	case IT_BTREE:
		return "BTree"
	case IT_HASH:
		return "Hash"
	}
	return fmt.Sprintf("?%d", it)
}

// idxEntryLenOf 读索引页的条目长度 (IndexHead.EntryLen); 非法时按主键叶条目 (14) 兜底。
func idxEntryLenOf(b []byte) int {
	el := int(le16(b[IX_ELEN : IX_ELEN+2]))
	if el <= 4 {
		el = 14 // v8 主键叶条目 = [key(8)+dpID(4)+slot(2)]
	}
	return el
}

// idxKeyLenOf 该索引在"某页 el 长度"下, 键占的字节数 (0 = 不适用/不检查)。
func idxKeyLenOf(it byte, el int) int {
	switch it {
	case IT_TIME:
		return 8 // 只取 BeginTime: LastTime 会随追加增长, 不是稳定键
	case IT_PRIMARY:
		// 主键叶条目两种布局的键都恰好是前 8 字节 (键 = pkID 的大端编码):
		//   v8: [key(8) + dpID(4) + slot(2)] = 14B
		//   v7: [key(8) + dpID(4)]            = 12B
		// 所以不能再用 el-4 (v8 下会多算进 2 字节 dpID)。
		return 8
	case IT_BTREE, IT_STR, IT_COMPOSE:
		return el - 8
	}
	return 0
}

// rowStoreRefOff 行存储索引条目里 DataPageID 的偏移。
//
// 主键叶条目自 v8 起在 dpID 之后还有 2B 槽位号, dpID 固定在偏移 8..12, 不能用"末 4B";
// 时序条目是 [BeginTime(8) + LastTime(8) + dpID(4)], dpID 仍在末 4B (LastTime 会随追加增长)。
func rowStoreRefOff(it byte, el int) int {
	if it == IT_PRIMARY {
		return 8
	}
	return el - 4
}

// pageChainOf 顺 PageHdr.NextPage 走一条页链 (自带防环)。
func pageChainOf(start uint32) []uint32 {
	var out []uint32
	seen := map[uint32]bool{}
	pb := newPageBuf()
	for pid := start; pid != 0 && !seen[pid]; {
		seen[pid] = true
		out = append(out, pid)
		b, e := pb.page(pid)
		if e != nil {
			break
		}
		pid = le32(b[PH_NEXT : PH_NEXT+4])
	}
	return out
}

// idxLeafChain 返回索引叶页 (按键的逻辑顺序)。
//   - B+ 树 (含 Compose —— 处理器底层即 SBT): 从根沿"每个内部节点的第 0 个子页"
//     下行到最左叶, 再顺 NextPage 走完叶层。
//     引擎的分隔键不变式是"父条目键 == 子树的键下界", 所以第 0 个条目的子页就是最左子树。
//   - 扁平链 (Time): 直接顺 NextPage 走。
//   - Hash: 无叶层, 返回 nil (另走桶链)。
func idxLeafChain(it byte, entry uint32) []uint32 {
	if entry == 0 || it == IT_HASH {
		return nil
	}
	if !isBTreeIdx(it) {
		return pageChainOf(entry)
	}
	pb := newPageBuf()
	cur := entry
	for depth := 0; depth <= 64; depth++ {
		b, e := pb.page(cur)
		if e != nil || len(b) < IDX_HDR || b[0] != PG_IDATA {
			return nil
		}
		if b[IX_LEAF] == 1 {
			return pageChainOf(cur)
		}
		if cn := int(le32(b[IX_CNT : IX_CNT+4])); cn <= 0 {
			return nil
		}
		el := idxEntryLenOf(b)
		if IDX_HDR+el > len(b) {
			return nil
		}
		next := be32(b[IDX_HDR+el-4 : IDX_HDR+el])
		if next == 0 {
			return nil
		}
		cur = next
	}
	return nil
}

// hashBucketPages 枚举 Hash 索引的全部桶页: 元数据页 -> 桶目录链 -> 每槽 head -> 桶溢出链。
func hashBucketPages(metaPID uint32) []uint32 {
	// 目录页与桶页各用一只缓冲: 桶页是嵌在目录页扫描循环里读的, 共用一只会把
	// 目录页内容冲掉 (后面还要从它取 NextPage)。
	dirBuf := newPageBuf()
	bktBuf := newPageBuf()
	meta, e := dirBuf.page(metaPID)
	if e != nil || len(meta) < HM_KEYLEN+4 || le32(meta[HM_MAGIC:HM_MAGIC+4]) != HASH_MAGIC {
		return nil
	}
	var out []uint32
	dirSeen := map[uint32]bool{}
	bucketSeen := map[uint32]bool{}
	limit := PHDR + ((PGSZ-PHDR)/HASH_SLOT)*HASH_SLOT
	for dir := le32(meta[HM_DIR : HM_DIR+4]); dir != 0 && !dirSeen[dir]; {
		dirSeen[dir] = true
		b, e := dirBuf.page(dir)
		if e != nil {
			break
		}
		for o := PHDR; o+HASH_SLOT <= limit; o += HASH_SLOT {
			p := le32(b[o : o+4])
			for p != 0 && !bucketSeen[p] {
				bucketSeen[p] = true
				out = append(out, p)
				bb, e := bktBuf.page(p)
				if e != nil {
					break
				}
				p = le32(bb[PH_NEXT : PH_NEXT+4])
			}
		}
		dir = le32(b[PH_NEXT : PH_NEXT+4])
	}
	return out
}

// eachRow 遍历一页里的存活行, 回调 (pkID, 行数据页内偏移, 行数据长度)。
// 行布局对照 index/datapage.nextRow: [rawLen(4B LE)][storeRow],
// rawLen 最高位 = 软删除; storeRow[:8] = pkID (BigEndian), 其后
// rowData = [schemaVer(1B)][列0 W0 B][列结束偏移表 2N][列1..N-1 内容][NULL 位图]
// —— 正好是引擎 extractIndexKey 的入参 (列位置由行内偏移表给出)。
func eachRow(b []byte, fn func(pk uint64, off, ln int)) {
	if len(b) < DP_HDR || b[0] != PG_DATA {
		return
	}
	end := int(le32(b[PH_USE : PH_USE+4]))
	if end > len(b) || end < DP_HDR {
		end = len(b)
	}
	for off := DP_HDR; off+4 <= end; {
		raw := le32(b[off : off+4])
		rl := int(raw &^ 0x80000000)
		start := off + 4
		if rl <= 0 || start+rl > end {
			break
		}
		if raw&0x80000000 == 0 && rl >= 8 {
			fn(binary.BigEndian.Uint64(b[start:start+8]), start+8, rl-8)
		}
		off = start + rl
	}
}

// collectLiveRows 读一张表的全部数据页, 收集"存活行"的 pkID 集合。
//
// 只收 pkID, 不再额外记行数据位置 (原 map[uint64]liveRow 在千万行级别要占几百 MB,
// 且整体落入 GC 扫描范围): 反向缺失检查现在直接按数据页现读现算, 见 checkReverseMissing。
func collectLiveRows(pages []uint32) map[uint64]bool {
	// 先探前 64 页估出行密度, 给 map 一个靠谱的初始容量。
	// 千万级 map 反复扩容 (每轮都要把已有条目全部重新哈希) 的开销远大于这次探测。
	pb := newPageBuf()
	hint := 0
	probe := len(pages)
	if probe > 64 {
		probe = 64
	}
	for i := 0; i < probe; i++ {
		if b, e := pb.page(pages[i]); e == nil {
			eachRow(b, func(uint64, int, int) { hint++ })
		}
	}
	if probe > 0 && len(pages) > probe {
		hint = hint * len(pages) / probe
	}

	set := make(map[uint64]bool, hint)
	for _, pid := range pages {
		b, e := pb.page(pid)
		if e != nil {
			continue
		}
		eachRow(b, func(pk uint64, off, ln int) { set[pk] = true })
	}
	return set
}

// idxIssueMu 保护 R 上的计数器与问题列表。
// 前向/反向检查现在是并行跑的: 每个索引自己的 st 由该任务独占, 但汇总器 R 共享。
// 只在"发现问题"这条冷路径上加锁, 热路径 (逐条扫描) 不加锁。
var idxIssueMu sync.Mutex

func addIdxIssue(R *AuditResult, st *IndexStat, is IndexIssue) {
	idxIssueMu.Lock()
	defer idxIssueMu.Unlock()
	switch is.Kind {
	case "out-of-order":
		R.IdxBadOrder++
	case "orphan-pk":
		R.IdxOrphanPK++
	case "not-data-page":
		R.IdxNotDataPage++
	case "missing-row":
		R.IdxMissingRow++
	case "missing-entry":
		R.IdxMissingEntry++
	}
	st.Issues++
	if len(R.IdxIssues) < idxIssueCap {
		R.IdxIssues = append(R.IdxIssues, is)
	} else {
		R.IdxIssueTrunc = true
	}
}

// checkIndexContent 逐表逐索引做内容一致性检查, 结果写回 R。
func checkIndexContent(R *AuditResult, w *walker, ts []TInfo) {
	// 未恢复的 WAL 说明盘上不是最终状态, 内容比对会系统性误报 —— 明确跳过并说明。
	if R.Reach.Wal > 0 {
		R.IdxSkipped = "检测到未恢复的 WAL 链, 盘上不是最终状态 —— 索引内容一致性校验已跳过 (先让引擎跑完恢复)"
		return
	}

	// 预聚合: 表名 -> 该表的数据页。
	// 数据页可能由两条路径认领: 数据链 (owner = 表名) 与行存储索引叶 (owner = 表名.索引名),
	// ownerTable 归一后取并集, 保证"存活行 pkID 集合"完整。
	dataPages := map[string][]uint32{}
	for pid, o := range w.owner {
		if w.role[pid] == "data" {
			tn := ownerTable(o)
			dataPages[tn] = append(dataPages[tn], pid)
		}
	}

	for i := range ts {
		t := &ts[i]
		if len(t.Idxs) == 0 {
			continue
		}

		// 行存储索引决定"行数据"的物理布局: 主键 -> DataPage (布局已知);
		// 时序 -> TimeDataPage (布局不同, 本文件不解析)。时序表下"存活行集合"
		// 会算成空集, 拿空集去比会把整表二级索引条目全判成悬空 —— 必须整体跳过。
		rowStore := byte(0)
		for _, idx := range t.Idxs {
			if isRowStore(idx.Type) {
				if rowStore == 0 || idx.Type == IT_PRIMARY {
					rowStore = idx.Type
				}
			}
		}
		primaryOK := rowStore == IT_PRIMARY

		// 存活行 pkID 集合 (建一次, 全索引复用): 前向扫索引时用来判"悬空 pk"。
		// 行数据本体不再另外收集 —— 反向检查按数据页现读现算 (见 checkReverseMissing)。
		var live map[uint64]bool
		if primaryOK {
			live = collectLiveRows(dataPages[t.Name])
		}

		ctx := make(map[string]*idxCtx, len(t.Idxs))
		primaryName := ""
		// 每个索引都有自己的 st/col (live 只读), 互不相干 —— 并行扫。
		// emp 那种一张表 4 个千万级索引的情形, 这一步就是整个审计最耗时的部分,
		// 顺序跑等于把多核晾着。
		type fwdTask struct {
			idx IInfo
			st  *IndexStat
			col *idxEntries
		}
		tasks := make([]fwdTask, len(t.Idxs))
		for k := range t.Idxs {
			idx := t.Idxs[k]
			st := &IndexStat{Table: t.Name, Index: idx.Name, IType: idxTypeName(idx.Type)}
			col := &idxEntries{it: idx.Type}
			switch {
			case idx.Type == IT_PRIMARY:
				col.pks = map[uint64]bool{}
			case idx.Type != IT_TIME:
				col.pairs = pairSet{}
			}
			tasks[k] = fwdTask{idx: idx, st: st, col: col}
		}
		runParallel(len(tasks), func(k int) {
			tk := tasks[k]
			if tk.idx.Type == IT_HASH {
				checkHashEntries(R, t.Name, tk.idx, tk.st, live, tk.col)
			} else {
				checkOneIndex(R, t.Name, tk.idx, tk.st, live, tk.col)
			}
		})
		for _, tk := range tasks {
			if tk.idx.Type == IT_PRIMARY {
				primaryName = tk.idx.Name
			}
			ctx[tk.idx.Name] = &idxCtx{info: tk.idx, stat: tk.st, set: tk.col}
		}

		// (4) 反向缺失: 行 -> 索引项。必须在前向扫完、集合齐了之后再做。
		// 主键方向 (行有没有掉出 B+ 树) 只要 pkID 集合, 没有二级索引的表 (只有主键,
		// 恰恰是数据量最大的那类) 也必须查 —— 那是最致命的一种缺失。
		if primaryOK && primaryName != "" {
			checkReverseMissing(R, t.Name, t, dataPages[t.Name], primaryName, ctx)
		} else if !primaryOK {
			R.IdxNotes = append(R.IdxNotes,
				fmt.Sprintf("%s: 行存储不是主键索引 (索引类型 %s, 行在 TimeDataPage), 反向缺失校验已跳过",
					t.Name, idxTypeName(rowStore)))
		}

		// 统计汇总放在最后: 反向检查也会往 st.Issues 上加。
		for _, idx := range t.Idxs {
			c := ctx[idx.Name]
			if c == nil {
				continue
			}
			R.IdxEntries += c.stat.Entries
			R.IdxStats = append(R.IdxStats, *c.stat)
		}
	}
	// 并行收集时问题的追加顺序随调度而变, 排一遍让报告可复现。
	slices.SortStableFunc(R.IdxIssues, func(a, b IndexIssue) int {
		if a.Table != b.Table {
			return cmp.Compare(a.Table, b.Table)
		}
		if a.Index != b.Index {
			return cmp.Compare(a.Index, b.Index)
		}
		if a.Page != b.Page {
			return cmp.Compare(a.Page, b.Page)
		}
		if a.At != b.At {
			return cmp.Compare(a.At, b.At)
		}
		return cmp.Compare(a.Kind, b.Kind)
	})
}

// checkOneIndex 校验单个非 Hash 索引。
// col != nil 时顺带把"条目存在性"记进去, 供反向缺失检查复用 —— 不重复扫一遍盘。
func checkOneIndex(R *AuditResult, table string, idx IInfo, st *IndexStat, live map[uint64]bool, col *idxEntries) {
	it := idx.Type
	crossPage := isBTreeIdx(it) // 只有 B+ 树保证叶层全局有序 (见文件头)

	var inPage []byte // 页内前一条键 (每页重置)
	var lastOfPrevPage []byte
	leafBuf := newPageBuf() // 叶页迭代缓冲
	refBuf := newPageBuf()  // 条目引用页缓冲 (与叶页分开, 否则会把正在遍历的叶页冲掉)
	lastRef := uint32(0)    // 连续条目常指向同一数据页 —— 记住上一次, 免重复读
	lastRefRead := false
	lastRefBad := false
	lastRefType := byte(0)
	var lastRefPage []byte // 上一次读到的引用页内容 (复用, 顺带校验主键 slot)
	for _, lp := range idxLeafChain(it, idx.EP) {
		b, e := leafBuf.page(lp)
		if e != nil || len(b) < IDX_HDR || b[0] != PG_IDATA {
			continue
		}
		el := idxEntryLenOf(b)
		cn := int(le32(b[IX_CNT : IX_CNT+4]))
		keyLen := idxKeyLenOf(it, el)
		if cn <= 0 || el <= 4 || keyLen <= 0 || keyLen > el {
			continue
		}
		inPage = inPage[:0]
		for k := 0; k < cn; k++ {
			o := IDX_HDR + k*el
			if o+el > len(b) {
				break
			}
			ent := b[o : o+el]
			st.Entries++
			key := ent[:keyLen]

			// 顺带收集条目存在性 (反向缺失用)。
			if col != nil {
				if it == IT_PRIMARY {
					col.pks[binary.BigEndian.Uint64(key)] = true
				} else if col.pairs != nil {
					// TIME 索引不参与反向缺失校验 (条目里的 LastTime 随追加增长、不是稳定键),
					// 因此创建 idxEntries 时故意没有为它分配 pairs —— 这里必须判空,
					// 否则往 nil map 写入会 panic (曾导致打开 orm_full.db 时分析器进程崩溃)。
					col.pairs[pairKey(ent)] = true // 条目字节本身就是 key||pkID(8B BE)
				}
			}

			// (1) 键序。
			switch {
			case len(inPage) > 0:
				if bytes.Compare(inPage, key) > 0 {
					addIdxIssue(R, st, IndexIssue{
						Table: table, Index: idx.Name, IType: idxTypeName(it),
						Kind: "out-of-order", Page: lp, At: k,
						Detail: fmt.Sprintf("页内键序回退: 前一条 %s > 本条 %s", hexKey(inPage), hexKey(key)),
					})
				}
			case crossPage && len(lastOfPrevPage) > 0:
				if bytes.Compare(lastOfPrevPage, key) > 0 {
					addIdxIssue(R, st, IndexIssue{
						Table: table, Index: idx.Name, IType: idxTypeName(it),
						Kind: "out-of-order", Page: lp, At: k,
						Detail: fmt.Sprintf("跨页键序回退: 上一叶末条 %s > 本叶首条 %s", hexKey(lastOfPrevPage), hexKey(key)),
					})
				}
			}
			inPage = append(inPage[:0], key...)

			// (3) 行存储索引: 条目里的 DataPageID 必须真的指向数据页。
			if isRowStore(it) {
				refOff := rowStoreRefOff(it, el)
				if ref := be32(ent[refOff : refOff+4]); ref != 0 {
					if !lastRefRead || ref != lastRef {
						rb, e := refBuf.page(ref)
						lastRef, lastRefRead = ref, true
						lastRefBad = e == nil && len(rb) > 0 && rb[0] != PG_DATA
						lastRefPage = rb
						if lastRefBad {
							lastRefType = rb[0]
						}
					}
					if lastRefBad {
						addIdxIssue(R, st, IndexIssue{
							Table: table, Index: idx.Name, IType: idxTypeName(it),
							Kind: "not-data-page", Page: lp, At: k,
							Detail: fmt.Sprintf("条目引用的页 %d 实际类型是 %s, 不是数据页",
								ref, pageTypeName(int(lastRefType))),
						})
					}
					// (3b) v8 主键叶条目多带 2B 槽位号: 必须落在该数据页的槽位目录范围内,
					// 否则 {dpID, slot} 会解析成别的行 (或空槽)。
					//
					// 端序要分清: 条目里的 slot 是**大端** (引擎 btree.btSlot/btSetSlot 用
					// BigEndian); 而数据页尾部的槽位目录容量是**小端** (datapage 用 LittleEndian)。
					// 这里曾把 slot 也按小端读 → 槽位 1 被读成 256, 全表主键条目集体误报 bad-slot。
					if it == IT_PRIMARY && !lastRefBad && el >= 14 && len(lastRefPage) >= 2 {
						slot := int(binary.BigEndian.Uint16(ent[12:14]))
						sc := int(le16(lastRefPage[len(lastRefPage)-2:]))
						if slot < 1 || slot > sc {
							addIdxIssue(R, st, IndexIssue{
								Table: table, Index: idx.Name, IType: idxTypeName(it),
								Kind: "bad-slot", Page: lp, At: k,
								Detail: fmt.Sprintf("条目的槽位号 %d 超出数据页 %d 的槽位目录范围 (1..%d)",
									slot, ref, sc),
							})
						}
					}
				}
				continue
			}

			// (2) 二级索引: 尾 8B 是 pkID, 必须能在本表回表到存活行。
			if live != nil {
				pk := binary.BigEndian.Uint64(ent[el-8 : el])
				if !live[pk] {
					addIdxIssue(R, st, IndexIssue{
						Table: table, Index: idx.Name, IType: idxTypeName(it),
						Kind: "orphan-pk", Page: lp, At: k,
						Detail: fmt.Sprintf("二级索引条目指向 pkID=%d 的行, 但该表数据页里没有这一行", pk),
					})
				}
			}
		}
		if crossPage {
			lastOfPrevPage = append(lastOfPrevPage[:0], inPage...)
		}
	}
}

// checkHashEntries 校验 Hash 索引桶条目 (桶内无序, 只查悬空 pkID)。
func checkHashEntries(R *AuditResult, table string, idx IInfo, st *IndexStat, live map[uint64]bool, col *idxEntries) {
	meta, e := readPage(idx.EP)
	if e != nil || len(meta) < HM_KEYLEN+4 || le32(meta[HM_MAGIC:HM_MAGIC+4]) != HASH_MAGIC {
		return
	}
	keyLen := int(le32(meta[HM_KEYLEN : HM_KEYLEN+4]))
	el := keyLen + 8
	if keyLen <= 0 || el > PGSZ-HASH_BUCKET_HDR {
		return // 元数据不可信: 不猜 (页归属 / 交叉引用已覆盖结构问题)
	}
	pb := newPageBuf()
	for _, bp := range hashBucketPages(idx.EP) {
		b, e := pb.page(bp)
		if e != nil || len(b) < HASH_BUCKET_HDR {
			continue
		}
		n := 0
		if us := int(le32(b[PH_USE : PH_USE+4])); us >= HASH_BUCKET_HDR && us <= len(b) {
			n = (us - HASH_BUCKET_HDR) / el
		} else if c := int(le32(b[24:28])); c > 0 {
			n = c // 退化: 用桶页 Count 推算
		}
		for k := 0; k < n; k++ {
			o := HASH_BUCKET_HDR + k*el
			if o+el > len(b) {
				break
			}
			st.Entries++
			pk := binary.BigEndian.Uint64(b[o+keyLen : o+el])
			if pk&(1<<63) != 0 {
				continue // 已删除条目 (hashDeletedFlag)
			}
			if col != nil {
				col.pairs[pairKey2(b[o:o+keyLen], pk)] = true
			}
			if live != nil && !live[pk] {
				addIdxIssue(R, st, IndexIssue{
					Table: table, Index: idx.Name, IType: idxTypeName(idx.Type),
					Kind: "orphan-pk", Page: bp, At: k,
					Detail: fmt.Sprintf("Hash 条目指向 pkID=%d 的行, 但该表数据页里没有这一行", pk),
				})
			}
		}
	}
}

// hexKey 把条目键按十六进制显示 (最多 16 字节)。
func hexKey(k []byte) string {
	if len(k) > 16 {
		k = k[:16]
	}
	return "0x" + fmt.Sprintf("%x", k)
}

// ============================================================================
// 反向缺失 (行 -> 索引项)
// ============================================================================

// idxEntries 一个索引里"已存在哪些条目"。
// 前向扫盘时顺手记下, 反向检查直接查表, 不再扫第二遍。
type idxEntries struct {
	it    byte
	pks   map[uint64]bool // 主键索引: 条目键就是 pkID
	pairs pairSet         // 二级索引: key||pkID(8B BE) 的 64 位指纹集
}

// idxCtx 一个索引的三件套 (元信息 / 统计 / 条目集合)。
type idxCtx struct {
	info IInfo
	stat *IndexStat
	set  *idxEntries
}

// keyScratch 复算索引键时的复用暂存: 分量表与结果缓冲都复用, 避免"每行每索引"
// 都分配 seg/key 两份切片 (千万行级别下这是 GB 级垃圾, 堆剖析里就落在这一处)。
type keyScratch struct {
	seg [][]byte
	key []byte
}

// rowVarOffMask 偏移表项的"偏移"掩码: bit15 是"该列是堆引用"标记, 低 15 位才是偏移
// (同引擎 types.RowVarOffMask)。
const rowVarOffMask = 0x7FFF

// padToWidth 把行内实际字节补零到档位宽度。
//
// v9 行内字符串列按实际长度存储, 而索引键按 FieldIDs 顺序【定长】拼接 (见 extract);
// 不补零算出的键长度不一, 与引擎的定长键不是同一份编码, 反向检查会满屏误报。
// 语义同引擎 types.PadToWidth: 已等宽时原样返回。
func padToWidth(raw []byte, width int) []byte {
	if len(raw) == width {
		return raw
	}
	buf := make([]byte, width)
	copy(buf, raw)
	return buf
}

// extract 复刻引擎 DB/core/index_meta_page.go:extractIndexKey。
//
// rowData = [schemaVer(1B)][列0 W0 B][列结束偏移表 2N][列1..N-1 内容][NULL 位图];
// 键 = 按 FieldIDs 顺序定长拼接各分量字段 (补零到档位宽度)。列位置一律由行内偏移表
// 给出, 行内偏移按该行自己的 schemaVer 取 (见函数内注释)。
//
// 返回 nil 与引擎同义 —— "这一行在该版本里凑不齐分量列", 引擎会跳过它, 因此
// 反向检查也必须跳过, 否则会把合法跳过的行误报成缺失。
//
// 返回的切片由 scratch 持有, 下次调用即被覆盖 —— 调用方必须立即用掉, 不得留存。
func (s *keyScratch) extract(rowData []byte, idx IInfo, t *TInfo) []byte {
	n := len(idx.Flds)
	if n == 0 || len(rowData) < 1 {
		return nil
	}
	if n > 16 { // 引擎 FieldIDs 上限 16
		return nil
	}
	ver := rowData[0]
	var sv *VInfo
	for i := range t.Versions {
		if t.Versions[i].Ver == ver {
			sv = &t.Versions[i]
			break
		}
	}
	if sv == nil {
		return nil
	}

	if cap(s.seg) < n {
		s.seg = make([][]byte, n)
	}
	seg := s.seg[:n]
	for i := range seg {
		seg[i] = nil // 清掉上一行的残留, 否则结果键会把旧分量算进来
	}

	// v9 行布局 (与引擎 types/rowvar.go 逐字节一致, 这里独立复刻):
	//   [schemaVer 1B][列0 W0 B][列结束偏移表 2N][列1..N-1 内容][NULL 位图 ⌈N/8⌉B]
	//   - 列 0 恒占 W0 字节 (档位宽度, 不足补零), W0 由 schema 决定 —— 正因可静态算出,
	//     解析端才不必先猜列 0 多长; 时序表 W0 == 8 ⇒ 时间键恒在绝对偏移 9。
	//   - 表项 0 = 变长内容区起点 (= 1+W0+2N), 表项 i(i≥1) = 列 i 内容结束偏移;
	//     列 i(i≥1) 起点 = 表项 i-1, 列 0 结束恒为 1+W0。位图起点 = 表末项。
	//   - 行内字符串列按【实际长度】存储, 拼键前必须补零回档位宽度 —— 索引键是定长拼接,
	//     不补零算出的键就与引擎不是同一份编码, 会满屏误报。
	// 任一处校验不过 → 返回 nil, 与引擎 RowParseOffsets 失败时跳过该行完全同义。
	rowN := len(sv.Indices)
	if rowN == 0 {
		return nil
	}
	col0Idx := sv.Indices[0]
	if col0Idx >= len(t.MasterFields) || t.MasterFields[col0Idx].Sz <= 0 {
		return nil
	}
	tblBase := 1 + t.MasterFields[col0Idx].Sz // 偏移表基址 == 列 0 结束偏移
	header := tblBase + 2*rowN                // 变长内容区起点 == 表项 0
	if header > len(rowData) {
		return nil
	}
	if int(binary.BigEndian.Uint16(rowData[tblBase:tblBase+2])) != header {
		return nil // 表项 0 必须正好是变长内容区起点 (被零填充 / 冲掉都会在这里暴露)
	}
	bitmapLen := (rowN + 7) / 8
	prev := header
	for i := 0; i < rowN; i++ {
		off := int(binary.BigEndian.Uint16(rowData[tblBase+2*i:tblBase+2*i+2])) & rowVarOffMask
		if off < prev || off > len(rowData) {
			return nil
		}
		prev = off
	}
	if prev+bitmapLen > len(rowData) {
		return nil
	}
	bitmap := rowData[prev : prev+bitmapLen]

	need := n
	for i, fi := range sv.Indices {
		if fi >= len(t.MasterFields) {
			return nil
		}
		size := t.MasterFields[fi].Sz
		if size <= 0 { // 未知类型: 宽度算不出来, 不猜
			return nil
		}
		start := 1
		if i > 0 {
			start = int(binary.BigEndian.Uint16(rowData[tblBase+2*(i-1):tblBase+2*i])) & rowVarOffMask
		}
		end := tblBase // 列 0 结束恒为 1+W0 == 表基址
		if i > 0 {
			end = int(binary.BigEndian.Uint16(rowData[tblBase+2*i:tblBase+2*i+2])) & rowVarOffMask
		}
		for k := 0; k < n; k++ {
			if idx.Flds[k] != fi {
				continue
			}
			if bitmap[i>>3]&(1<<uint(i&7)) != 0 {
				return nil // 分量是 NULL → 该行不进本索引
			}
			if end < start || end > len(rowData) {
				return nil
			}
			seg[k] = padToWidth(rowData[start:end], size)
			need--
			break
		}
		if need == 0 {
			break
		}
	}
	if need != 0 {
		return nil
	}

	total := 0
	for k := 0; k < n; k++ {
		total += len(seg[k])
	}
	if cap(s.key) < total {
		s.key = make([]byte, 0, total)
	}
	key := s.key[:0]
	for k := 0; k < n; k++ {
		key = append(key, seg[k]...)
	}
	s.key = key
	return key
}

// pageLoc 输出用的" (数据页 N)" 后缀; 页号 0 表示未知。
func pageLoc(page uint32) string {
	if page == 0 {
		return ""
	}
	return fmt.Sprintf(" (数据页 %d)", page)
}

// checkReverseMissing 反向缺失: 每个存活行都必须在每个索引里留下条目。
//
// 这一步补的是前三条补不上的那一半。前三条全是"索引项 -> 行"方向, 它们能证明
// "索引里没有废条目", 但证明不了"该有的条目都在": 一行整体从某个索引里掉了
// (更新时删旧键/插新键只做了一半、重建索引漏了一批行), 只要不留废条目,
// 前三条一个都不会响 —— 必须从行出发反向查一次。
//
// 直接按数据页并行扫描: 不再预建 map[uint64]liveRow (千万行级别要占几百 MB, 且整体
// 落入 GC 扫描范围)。行数据本就躺在数据页里, 这里现读现算; 每只 worker 只多占
// 一只 16KB 页缓冲 + 一份 keyScratch。
func checkReverseMissing(R *AuditResult, table string, t *TInfo, dataPages []uint32,
	primaryName string, ctx map[string]*idxCtx) {
	if len(dataPages) == 0 {
		return
	}
	// 反向检查是纯读 (各索引条目集在前向阶段已全部定型) + 计数回报, 按数据页切块并行;
	// 问题汇总由 addIdxIssue 里的锁保证安全。
	wk := runtime.NumCPU()
	if wk > len(dataPages) {
		wk = len(dataPages)
	}
	chunk := (len(dataPages) + wk - 1) / wk
	var wg sync.WaitGroup
	for k := 0; k < wk; k++ {
		lo := k * chunk
		hi := lo + chunk
		if lo >= len(dataPages) {
			break
		}
		if hi > len(dataPages) {
			hi = len(dataPages)
		}
		wg.Add(1)
		go func(dps []uint32) {
			defer wg.Done()
			pb := newPageBuf()
			var sc keyScratch
			for _, page := range dps {
				revScanPage(R, table, t, page, primaryName, ctx, pb, &sc)
			}
		}(dataPages[lo:hi])
	}
	wg.Wait()
}

// revScanPage 反向缺失检查的单页。
func revScanPage(R *AuditResult, table string, t *TInfo, page uint32, primaryName string,
	ctx map[string]*idxCtx, pb *pageBuf, sc *keyScratch) {
	b, e := pb.page(page)
	if e != nil || len(b) < DP_HDR || b[0] != PG_DATA {
		return
	}
	loc := pageLoc(page)
	eachRow(b, func(pk uint64, off, ln int) {
		// (4a) 主键索引里必须能找到这一行自己。
		if c := ctx[primaryName]; c != nil && c.set != nil && c.set.pks != nil {
			if !c.set.pks[pk] {
				addIdxIssue(R, c.stat, IndexIssue{
					Table: table, Index: primaryName, IType: idxTypeName(c.info.Type),
					Kind: "missing-row", Page: page, At: -1,
					Detail: fmt.Sprintf("存活行 pkID=%d%s 在主键索引里没有条目 —— "+
						"这行已从 B+ 树掉出去, 按主键和任何索引都查不到它", pk, loc),
				})
			}
		}

		// (4b) 每个二级索引里都必须有该行的条目 (键用引擎同款算法复算)。
		rowData := b[off : off+ln]
		for _, idx := range t.Idxs {
			if isRowStore(idx.Type) {
				continue
			}
			c := ctx[idx.Name]
			if c == nil || c.set == nil || c.set.pairs == nil {
				continue
			}
			key := sc.extract(rowData, idx, t)
			if key == nil {
				continue // 引擎同样会跳过这一行 (分量列缺失 / 版本不符)
			}
			if !c.set.pairs[pairKey2(key, pk)] {
				addIdxIssue(R, c.stat, IndexIssue{
					Table: table, Index: idx.Name, IType: idxTypeName(idx.Type),
					Kind: "missing-entry", Page: page, At: -1,
					Detail: fmt.Sprintf("存活行 pkID=%d%s 在本索引里没有条目 (该有的键 %s) —— "+
						"按这个索引查这一行会查不到", pk, loc, hexKey(key)),
				})
			}
		}
	})
}