TsDB2/DB/core/index/btree/primary.go

package btree

import (
	"TsDB2/DB/core/index/datapage"
	"TsDB2/DB/core/page"
	"encoding/binary"
	"errors"
	"fmt"
	"unsafe"
)

// btNode B+ 树节点 (raw bytes + page.IndexHead 指针, 非 IndexPage 对象)
type btNode struct {
	buf     []byte          // 完整页缓冲区
	info    *page.IndexHead // 指向 buf 头部的 page.IndexHead 指针
	entries []byte          // buf[page.IndexHeadSize:] 条目区域
}

func (n *btNode) isLeaf() bool   { return n.info.IsLeaf == 1 }
func (n *btNode) pageID() uint32 { return n.info.Hdr.PageID }

// btPathEntry 记录 B+ 树 descent 路径
type btPathEntry struct {
	nodePID  uint32
	childIdx int
}

// btSplitThreshold 计算分裂阈值
func btSplitThreshold() int {
	ff := page.BTreeFillFactor
	if ff < 50 {
		ff = 50
	}
	if ff > 100 {
		ff = 100
	}
	return int(page.IndexPageSize) * ff / 100
}

// ======================== 索引页归还 ========================

// btReleasePage 归还一个已确认不可达的索引页 (内部节点或叶子节点)。
//
// 为什么优先走 DeferFreePage 而不是直接 FreePage: 空闲链表是直写盘的 (绕过
// BufferPool), 而本库是"先 WAL 后回写"。删除/合并途中若立即把页写成空闲链表
// 节点, 一旦崩溃或 Rollback, 树结构虽然退回删除前, 页内容却已在盘上变成空页,
// 索引就指向了一张空页。延迟回收只登记, 等 Commit/Close 再真正还链 ——
// 与 heapReleaseRow / deleter 回收旧 DataPage 是同一套时序约定。
//
// 只有调用方没提供 DeferFreePage 时 (demo 里手搓的 HandleDb) 才退回立即回收。
func btReleasePage(handle page.HandleDb, pid uint32) {
	if pid == 0 {
		return
	}
	if handle.DeferFreePage != nil {
		handle.DeferFreePage(pid)
		return
	}
	if handle.FreePage != nil {
		handle.FreePage(pid)
	}
}

// btReleaseSubtree 归还一整棵已确认不可达的子树 (含内部节点与叶子节点)。
// 只用于"整段摘除"的场景 (空子树被摘链 / 旧树被换根), 调用前必须保证该子树
// 已无任何父条目引用、且已从叶子链上摘除。
func btReleaseSubtree(handle page.HandleDb, rootPID uint32) {
	if rootPID == 0 {
		return
	}
	for _, pid := range CollectAllPageIDs(handle, rootPID) {
		btReleasePage(handle, pid)
	}
}

// ======================== 节点 I/O ========================

func btLoad(handle page.HandleDb, pid uint32) *btNode {
	if pid == 0 {
		return nil
	}
	buf := nodeBufGet()
	if err := handle.ReadPage(pid, buf); err != nil {
		nodeBufPut(buf)
		return nil
	}
	info := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
	return &btNode{buf: buf, info: info, entries: buf[page.IndexHeadSize:]}
}

// btLoadNode 加载并校验「这确实是一张自洽的 B+ 节点页」, 不自洽返回 nil。
//
// 为什么必须校验: btLoad 只按偏移解析页头, 既不看页类型、也不检查条目区长度。
// 传入一张已被回收复用过的页 (DataPage / 堆页 / WAL 页) 时, 里面根本不是节点布局:
//
//	IndexHead.Count 落在 DataPage 的行键 Min/Max 上 → 取到 2 万级的大整数;
//	btEntry(n, i) 于是从行数据中间切片, btPID 截出的 4 字节就成了垃圾页号
//	(实测 DROP TABLE 收集到 2463488 = 0x259700, 一个从未分配过的页号)。
//
// 校验三条: 页类型是索引节点、页内自述 PageID 与取页号一致、Count*EntryLen 不越界。
// 回收路径 (DropTable / btReleaseSubtree / GetDataPageIDs) 必须走这个版本 —— 垃圾页号
// 一旦 FreePage, 空闲链头就指向文件里不存在的位置, 下一次 NewPage 会写到几十 GB 偏移。
func btLoadNode(handle page.HandleDb, pid uint32) *btNode {
	if pid == 0 {
		return nil
	}
	n := btLoad(handle, pid)
	if n == nil {
		return nil
	}
	if n.info.Hdr.Type != page.IndexDataIndex || n.info.Hdr.PageID != pid {
		return nil
	}
	el := uint32(n.info.EntryLen)
	if el < 4 {
		return nil
	}
	if uint32(n.info.Count)*el > uint32(len(n.entries)) {
		return nil
	}
	return n
}

func btSave(handle page.HandleDb, n *btNode) {
	_ = handle.SavePage(n.info.Hdr.PageID, n.buf)
}

func btNew(handle page.HandleDb, isLeaf bool, entryLen int) *btNode {
	pid := handle.NewPage()
	buf := nodeBufGet()
	clear(buf) // 池化缓冲可能带旧节点数据, 条目区必须回到"新页=全零"语义
	info := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
	info.Hdr.Type = page.IndexDataIndex
	info.Hdr.PageID = pid
	info.Hdr.UseSize = uint32(page.IndexHeadSize)
	info.IndexType = page.IndexTypePrimary
	info.EntryLen = uint16(entryLen)
	info.Count = 0
	if isLeaf {
		info.IsLeaf = 1
	}
	return &btNode{buf: buf, info: info, entries: buf[page.IndexHeadSize:]}
}

// ======================== 条目操作 ========================

func btEntry(n *btNode, i int) []byte {
	el := int(n.info.EntryLen)
	return n.entries[i*el : (i+1)*el]
}

// ======================== 条目布局 ========================
//
// 两类条目的**键都在前 8 字节、pid 都在偏移 8 处**, 只是长度不同:
//
//	内部节点条目 [key(8B)][childPID(4B BE)]             entryLen = btIntEntryLen  (12)
//	主键叶条目   [key(8B)][dpID(4B BE)][slot(2B BE)]    entryLen = btLeafEntryLen (14)
//
// 叶条目多出来的 slot 是「行号」: datapage 的槽位目录保证活行槽位号稳定
// (见 datapage 的不变式注释), 命中叶后 ReadRowBySlotPK 就能 O(1) 取行, 不必再在页内
// 扫一遍找 pkID。
//
// 因为键与 pid 都落在固定偏移上, btKey/btPID 对两种条目都成立; 但**绝不能用「末 4 字节」
// 当 pid** —— 对 14 字节的叶条目那读出来的是 dpID 低 2 字节 + slot。slot 只能走 btSlot。
const (
	btIntEntryLen  = 12 // 内部节点: [key(8B)][childPID(4B BE)]
	btLeafEntryLen = 14 // 主键叶:   [key(8B)][dpID(4B BE)][slot(2B BE)]
)

func btKey(e []byte) []byte       { return e[:8] }
func btPID(e []byte) uint32       { return binary.BigEndian.Uint32(e[8:12]) }
func btSetPID(e []byte, p uint32) { binary.BigEndian.PutUint32(e[8:12], p) }

// btSlot 读主键叶条目的槽位号 (1 基; 0 = 无效)。仅对叶条目有效。
func btSlot(e []byte) uint16 { return binary.BigEndian.Uint16(e[12:14]) }

// btSetSlot 写主键叶条目的槽位号。仅对叶条目有效。
func btSetSlot(e []byte, s uint16) { binary.BigEndian.PutUint16(e[12:14], s) }

// btChildPID 读一张「内部节点」条目里的子页号。
//
// 上面那套「键在前 8 字节、pid 在偏移 8」只对**主键**条目成立。二级 B+ 树的条目
// 是变长的 (键长随字段类型走), 布局为:
//
//	二级内部 [key(keyLen)][childPID(4B BE)]   entryLen = keyLen+4   → 子页号在**末 4 字节**
//	二级叶子 [key(keyLen)][PKID(8B)]          entryLen = keyLen+8   → 末 8 字节是主键号, 非页号
//
// 所以下探内部节点时绝不能一律用 btPID(偏移 8..12): 对二级树那读的是键值尾部
// (复合键) 或全零填充 —— 实测整棵二级树的内部条目在该处恒为 0, 于是 BFS 只收到
// 根页, 整棵子树被永久漏收 (页泄漏, 文件只涨不缩)。
//
// 判定依据只认页头自述的 IndexType (btLoadNode 已保证它就是本页), 不做任何猜测。
func btChildPID(n *btNode, e []byte) uint32 {
	if page.IndexType(n.info.IndexType) == page.IndexTypePrimary {
		return btPID(e)
	}
	// BTree / 旧 Str: 子页号在条目末 4 字节。
	return sbtChildPID(e)
}

func btCmp(a, b []byte) int {
	ml := len(a)
	if len(b) < ml {
		ml = len(b)
	}
	for i := 0; i < ml; i++ {
		if a[i] < b[i] {
			return -1
		}
		if a[i] > b[i] {
			return 1
		}
	}
	if len(a) < len(b) {
		return -1
	}
	if len(a) > len(b) {
		return 1
	}
	return 0
}

func btRoute(n *btNode, key []byte) uint32 {
	cnt := int(n.info.Count)
	if cnt <= 0 {
		return 0 // 防御: 空节点不应作为路由节点
	}
	// 二分查找: 找最后一个 entry.key <= key 的位置 (与二级 B+ 树统一 <= 约定)
	left, right := 0, cnt-1
	result := 0
	for left <= right {
		mid := (left + right) / 2
		if btCmp(btKey(btEntry(n, mid)), key) <= 0 {
			result = mid
			left = mid + 1 // 继续向右找
		} else {
			right = mid - 1
		}
	}
	return btPID(btEntry(n, result))
}

// maxLeafEntries 单个叶节点最大条目数
func maxLeafEntries(entryLen int) int {
	threshold := btSplitThreshold()
	available := threshold - page.IndexHeadSize
	if available <= 0 {
		return 1
	}
	return available / entryLen
}

// BuildFromSorted 从已排序的叶条目批量构建 B+ 树 (自底向上)
// leafEntries: 每个条目 = [key(8B) + dpID(4B BE) + slot(2B BE)], 已按 key 升序排列
// 返回新树根 pageID
//
// 内存优化: 避免逐行 B+ 树搜索 (每次 btLoad 16KB), 改为顺序创建所有节点
func BuildFromSorted(leafEntries [][]byte, handle page.HandleDb) uint32 {
	if len(leafEntries) == 0 {
		return 0
	}

	el := btLeafEntryLen // [key(8) + dpID(4) + slot(2)]
	maxPerLeaf := maxLeafEntries(el)

	// ---- 构建叶节点层 ----
	type childInfo struct {
		pid      uint32
		firstKey []byte // 该子树第一个 key (用于 > 路由)
	}

	var children []childInfo
	var prevLeafPID uint32

	// 将条目分组到叶节点 (带链表链接)
	for i := 0; i < len(leafEntries); {
		end := i + maxPerLeaf
		if end > len(leafEntries) {
			end = len(leafEntries)
		}
		leaf := btNew(handle, true, el)
		for j := i; j < end; j++ {
			src := leafEntries[j]
			dst := btEntry(leaf, int(leaf.info.Count))
			copy(dst, src)
			leaf.info.Count++
			leaf.info.Hdr.UseSize += uint32(el)
		}
		// 设置叶节点链表链接
		leaf.info.Hdr.PrePage = prevLeafPID
		if prevLeafPID != 0 {
			// 更新前一个叶节点的 NextPage
			// 注意: 前一个叶节点已经保存, 需要重新加载并更新
			// 但我们可以在保存前就设置好 NextPage
		}
		btSave(handle, leaf)
		// 更新前一个叶节点的 NextPage 指向当前叶
		if prevLeafPID != 0 {
			prevLeaf := btLoad(handle, prevLeafPID)
			if prevLeaf != nil {
				prevLeaf.info.Hdr.NextPage = leaf.pageID()
				btSave(handle, prevLeaf)
			}
		}
		prevLeafPID = leaf.pageID()
		children = append(children, childInfo{
			pid:      leaf.pageID(),
			firstKey: leafEntries[i][:8],
		})
		i = end
	}

	if len(children) == 1 {
		return children[0].pid
	}

	// ---- 自底向上构建内部节点 ----
	// 路由约定 (btRoute): 找最后一个 entry.key <= searchKey
	// 内部节点条目: entry[i] = (child[i].firstKey, child[i].pid)
	// 即 entry[i].key = 该子树最左叶的第一个 key
	intEl := btIntEntryLen
	maxPerInt := maxLeafEntries(intEl)

	for len(children) > 1 {
		var parents []childInfo
		for i := 0; i < len(children); {
			end := i + maxPerInt
			if end > len(children) {
				end = len(children)
			}
			node := btNew(handle, false, intEl)
			for j := i; j < end; j++ {
				e := make([]byte, intEl)
				// 分隔键 = 该 child 的第一个 key
				copy(e[:intEl-4], children[j].firstKey)
				btSetPID(e, children[j].pid)
				dst := btEntry(node, int(node.info.Count))
				copy(dst, e)
				node.info.Count++
				node.info.Hdr.UseSize += uint32(intEl)
			}
			btSave(handle, node)
			parents = append(parents, childInfo{
				pid:      node.pageID(),
				firstKey: children[i].firstKey,
			})
			i = end
		}
		children = parents
	}

	return children[0].pid
}

func btChildIdx(n *btNode, key []byte) int {
	cnt := int(n.info.Count)
	// 二分查找: 找最后一个 entry.key <= key 的位置 (与二级 B+ 树统一 <= 约定)
	left, right := 0, cnt-1
	result := 0
	for left <= right {
		mid := (left + right) / 2
		if btCmp(btKey(btEntry(n, mid)), key) <= 0 {
			result = mid
			left = mid + 1 // 继续向右找
		} else {
			right = mid - 1
		}
	}
	return result
}

func btInsEntry(n *btNode, entry []byte, pos int) {
	el := int(n.info.EntryLen)
	cnt := int(n.info.Count)
	// 整段后移一格 (块拷贝): 逐条目循环移位在千条目级叶上每插一条要搬千次 14B,
	// 批量灌叶路径实测占 ~4% CPU; Go 的 copy 是 memmove 语义, 区间重叠安全。
	copy(n.entries[(pos+1)*el:(cnt+1)*el], n.entries[pos*el:cnt*el])
	copy(n.entries[pos*el:(pos+1)*el], entry)
	n.info.Count++
	n.info.Hdr.UseSize += uint32(el)
	if n.isLeaf() {
		btReportUnsorted("insEntry", n)
	}
}

func btRmEntry(n *btNode, pos int) {
	el := int(n.info.EntryLen)
	cnt := int(n.info.Count)
	for i := pos; i < cnt-1; i++ {
		copy(btEntry(n, i), btEntry(n, i+1))
	}
	n.info.Count--
	n.info.Hdr.UseSize -= uint32(el)
	if n.isLeaf() {
		btReportUnsorted("rmEntry", n)
	}
}

// btLeafFind 在叶节点内二分查找等于 key 的条目。
// 叶条目按 key 升序排列 (B+ 树不变式), 主键唯一, 因此可安全二分。
func btLeafFind(n *btNode, key []byte) (int, bool) {
	cnt := int(n.info.Count)
	left, right := 0, cnt-1
	for left <= right {
		mid := (left + right) / 2
		cmp := btCmp(btKey(btEntry(n, mid)), key)
		if cmp == 0 {
			return mid, true
		} else if cmp < 0 {
			left = mid + 1
		} else {
			right = mid - 1
		}
	}
	return 0, false
}

// btLeafPos 返回第一个 key >= 目标 key 的条目下标 (lower_bound), 用于插入定位。
// 二分实现, 语义与线性扫描完全一致 (未命中时返回 cnt)。
func btLeafPos(n *btNode, key []byte) int {
	cnt := int(n.info.Count)
	left, right := 0, cnt
	for left < right {
		mid := (left + right) / 2
		if btCmp(btKey(btEntry(n, mid)), key) < 0 {
			left = mid + 1
		} else {
			right = mid
		}
	}
	return left
}

// btLeftmostKey 获取子树最左叶的第一个 key（用于 promoKey）
func btLeftmostKey(handle page.HandleDb, nodePID uint32) []byte {
	cur := nodePID
	for {
		n := btLoad(handle, cur)
		if n == nil {
			return nil
		}
		if n.isLeaf() {
			if n.info.Count == 0 {
				return nil
			}
			e := btEntry(n, 0)
			key := make([]byte, len(btKey(e)))
			copy(key, btKey(e))
			return key
		}
		// 非叶节点，继续向左下遍历
		cur = btPID(btEntry(n, 0))
	}
}

// btFirstLeaf 沿最左分支下降, 返回以 pid 为根的子树中最左叶子页 ID (找不到返回 0)。
func btFirstLeaf(handle page.HandleDb, pid uint32) uint32 {
	for pid != 0 {
		n := btLoad(handle, pid)
		if n == nil {
			return 0
		}
		if n.isLeaf() {
			return pid
		}
		if n.info.Count == 0 {
			return 0
		}
		pid = btPID(btEntry(n, 0))
	}
	return 0
}

// btLastLeaf 沿最右分支下降, 返回以 pid 为根的子树中最右叶子页 ID (找不到返回 0)。
func btLastLeaf(handle page.HandleDb, pid uint32) uint32 {
	for pid != 0 {
		n := btLoad(handle, pid)
		if n == nil {
			return 0
		}
		if n.isLeaf() {
			return pid
		}
		if n.info.Count == 0 {
			return 0
		}
		pid = btPID(btEntry(n, int(n.info.Count)-1))
	}
	return 0
}

// btUnlinkSubtreeLeaves 把以 pid 为根的子树所覆盖的那一段叶子, 从叶子双向链表中整体摘除。
// 仅在该子树已被确认 "完全为空" 时调用。
func btUnlinkSubtreeLeaves(handle page.HandleDb, pid uint32) {
	first := btFirstLeaf(handle, pid)
	if first == 0 {
		return
	}
	last := btLastLeaf(handle, pid)
	fl := btLoad(handle, first)
	ll := btLoad(handle, last)
	if fl == nil || ll == nil {
		return
	}
	pre := fl.info.Hdr.PrePage
	next := ll.info.Hdr.NextPage
	if pre != 0 {
		pr := btLoad(handle, pre)
		if pr != nil {
			pr.info.Hdr.NextPage = next
			btSave(handle, pr)
		}
	}
	if next != 0 {
		nl := btLoad(handle, next)
		if nl != nil {
			nl.info.Hdr.PrePage = pre
			btSave(handle, nl)
		}
	}
}

// ======================== B+ 树搜索 ========================

// PKSearchStatus 主键查找的三态结果。
//
// 为什么必须区分"未命中"与"失败": B+ 树不变量成立时, 正常下降到达叶子且键不在,
// 就能断言【全树都不存在该键】—— 键是全局有序的, 该键只会落在被路由到的那片叶子里。
// 旧实现把这两种情况都塞进 found=false, 调用方 (SearchByPK) 只能一律兜底全叶扫描,
// 于是"查一个不存在的主键"是 O(叶数), 而插入查重 (rowStoreHasPK) 每行都要查一次
// 不存在的键 —— 显式主键的批量导入因此退化成 O(N²) (实测 100 万行 ≈1 小时跑不完)。
type PKSearchStatus uint8

const (
	// PKFail 下降中断 (页读不出 / 空路由节点): 树结构异常, 结果未知, 允许兜底扫描。
	PKFail PKSearchStatus = iota
	// PKHit 命中, dpID 与 slot 有效。
	PKHit
	// PKMiss 正常到达叶子且键不在: 键一定不存在于整棵树。
	PKMiss
)

// BTreePKSearchEx 定位主键 (三态版), 语义见 PKSearchStatus。
func BTreePKSearchEx(rootPID uint32, key []byte, handle page.HandleDb) (uint32, uint16, PKSearchStatus) {
	if rootPID == 0 {
		return 0, 0, PKMiss // 空树: 键必不存在
	}
	cur := rootPID
	for {
		n := btLoad(handle, cur)
		if n == nil {
			return 0, 0, PKFail
		}
		if n.isLeaf() {
			pos, ok := btLeafFind(n, key)
			if !ok {
				return 0, 0, PKMiss
			}
			e := btEntry(n, pos)
			return btPID(e), btSlot(e), PKHit
		}
		next := btRoute(n, key)
		if next == 0 {
			return 0, 0, PKFail // 空路由节点: 结构异常, 不再下探
		}
		cur = next
	}
}

// BTreePKSearch 定位主键, 返回 (dpID, slot, ok)。
//
// slot 是 datapage 的槽位号 (1 基), 直接来自叶条目 —— 拿到后用
// ReadRowBySlotPK(dpID, slot, key) 就能 O(1) 取行, 不必再在页内按 pkID 扫一遍
// (叶条目的键就是 pkID, 见 PrimaryHandler.Insert)。
//
// ok=false 同时覆盖"未命中"与"路由失败"; 需要区分二者的调用方 (SearchByPK 等
// 热路径) 请改用 BTreePKSearchEx。本函数留给测试与演示代码。
func BTreePKSearch(rootPID uint32, key []byte, handle page.HandleDb) (uint32, uint16, bool) {
	dpID, slot, st := BTreePKSearchEx(rootPID, key, handle)
	return dpID, slot, st == PKHit
}

// ======================== B+ 树叶子节点定位 ========================

// BTreeFindLeftmostLeaf 从根节点路由到最左叶节点，返回叶节点 pageID
func BTreeFindLeftmostLeaf(rootPID uint32, handle page.HandleDb) uint32 {
	cur := rootPID
	for {
		n := btLoad(handle, cur)
		if n == nil {
			return 0
		}
		if n.isLeaf() {
			return n.info.Hdr.PageID
		}
		cur = btPID(btEntry(n, 0))
	}
}

// BTreeFindRightmostLeaf 从根节点路由到最右叶节点，返回叶节点 pageID
func BTreeFindRightmostLeaf(rootPID uint32, handle page.HandleDb) uint32 {
	cur := rootPID
	for {
		n := btLoad(handle, cur)
		if n == nil {
			return 0
		}
		if n.isLeaf() {
			return n.info.Hdr.PageID
		}
		cur = btPID(btEntry(n, int(n.info.Count)-1))
	}
}

// ======================== B+ 树范围扫描 ========================

// BTreePKRangeScan 主键 B+ 树范围扫描
func BTreePKRangeScan(rootPID uint32, handle page.HandleDb,
	startKey, endKey []byte, startInc, endInc bool, reverse bool,
	callback func(row []byte) bool) error {
	if rootPID == 0 {
		return nil
	}
	if !reverse {
		// 正向: 路由到 startKey 所在叶, 沿 NextPage 遍历
		cur := rootPID
		for {
			n := btLoad(handle, cur)
			if n == nil {
				return nil
			}
			if n.isLeaf() {
				break
			}
			cur = btRoute(n, startKey)
		}
		for cur != 0 {
			n := btLoad(handle, cur)
			if n == nil {
				break
			}
			for i := 0; i < int(n.info.Count); i++ {
				entry := btEntry(n, i)
				key := btKey(entry)
				if endKey != nil {
					cmp := btCmp(key, endKey)
					if cmp > 0 || (cmp == 0 && !endInc) {
						return nil
					}
				}
				if startKey != nil {
					cmp := btCmp(key, startKey)
					if cmp < 0 || (cmp == 0 && !startInc) {
						continue
					}
				}
				// 叶条目直接给出 (dpID, slot): O(1) 取行, 不再在页内按 pkID 扫。
				dp, err := datapage.Load(handle, btPID(entry))
				if err != nil {
					continue
				}
				if row, ok := dp.ReadRowBySlotPK(uint32(btSlot(entry)), key); ok {
					if !callback(row) {
						return nil
					}
				}
			}
			cur = n.info.Hdr.NextPage
		}
	} else {
		// 反向: 路由到 endKey 所在叶(或最右叶), 沿 PrePage 遍历
		cur := rootPID
		if endKey != nil {
			for {
				n := btLoad(handle, cur)
				if n == nil {
					return nil
				}
				if n.isLeaf() {
					break
				}
				cur = btRoute(n, endKey)
			}
		} else {
			cur = BTreeFindRightmostLeaf(rootPID, handle)
		}
		for cur != 0 {
			n := btLoad(handle, cur)
			if n == nil {
				break
			}
			for i := int(n.info.Count) - 1; i >= 0; i-- {
				entry := btEntry(n, i)
				key := btKey(entry)
				if startKey != nil {
					cmp := btCmp(key, startKey)
					if cmp < 0 || (cmp == 0 && !startInc) {
						return nil
					}
				}
				if endKey != nil {
					cmp := btCmp(key, endKey)
					if cmp > 0 || (cmp == 0 && !endInc) {
						continue
					}
				}
				// 反向同理: 叶条目给出 (dpID, slot), O(1) 取行。
				dp, err := datapage.Load(handle, btPID(entry))
				if err != nil {
					continue
				}
				if row, ok := dp.ReadRowBySlotPK(uint32(btSlot(entry)), key); ok {
					if !callback(row) {
						return nil
					}
				}
			}
			cur = n.info.Hdr.PrePage
		}
	}
	return nil
}

// ======================== B+ 树插入 ========================

// BTreePKInsert 插入主键索引条目 (key = pkID 的 8 字节 BigEndian 编码)。
//
// slot 是 datapage 槽位号 (1 基), 来自 InsertRowSortedSlot / AppendRowSlot —— 叶条目把它
// 一并存下, 之后的查找/范围扫描就能 O(1) 取行 (槽位不变式见 datapage 注释)。
//
// 返回 (newRoot, dupDPID, dup, err): dup=true 表示 key 已存在 (主键冲突), dupDPID 是
// 已存在行所在的 DataPage。
func BTreePKInsert(rootPID uint32, key []byte, dpID uint32, slot uint16, handle page.HandleDb) (uint32, uint32, bool, error) {
	if rootPID == 0 {
		leaf := btNew(handle, true, btLeafEntryLen)
		e := make([]byte, btLeafEntryLen)
		copy(e[:8], key)
		btSetPID(e, dpID)
		btSetSlot(e, slot)
		btInsEntry(leaf, e, 0)
		btSave(handle, leaf)
		newRoot := leaf.pageID()
		leaf.release()
		return newRoot, 0, false, nil
	}
	var path []btPathEntry
	cur := rootPID
	for {
		n := btLoad(handle, cur)
		if n == nil {
			return rootPID, 0, false, errors.New("load failed")
		}
		if n.isLeaf() {
			pos, found := btLeafFind(n, key)
			if found {
				foundPID := btPID(btEntry(n, pos))
				n.release()
				return rootPID, foundPID, true, nil
			}
			el := btLeafEntryLen
			ne := make([]byte, el)
			copy(ne[:8], key)
			btSetPID(ne, dpID)
			btSetSlot(ne, slot)
			ip := btLeafPos(n, key)
			if int(n.info.Hdr.UseSize)+el <= btSplitThreshold() {
				btInsEntry(n, ne, ip)
				btSave(handle, n)
				if ip == 0 {
					// 新键成为本叶最小键: 沿路径向上同步 "父条目键 == 子树最小键"
					for i := len(path) - 1; i >= 0; i-- {
						p := path[i]
						pb := btLoad(handle, p.nodePID)
						if pb == nil || p.childIdx < 0 || p.childIdx >= int(pb.info.Count) {
							break
						}
						copy(btEntry(pb, p.childIdx)[:8], key)
						btSave(handle, pb)
						pb.release()
						if p.childIdx != 0 {
							break
						}
					}
				}
				n.release()
				return rootPID, 0, false, nil
			}
			cnt := int(n.info.Count)
			total := cnt + 1
			all := make([]byte, total*el)
			if ip > 0 {
				copy(all, n.entries[:ip*el])
			}
			copy(all[ip*el:], ne)
			if cnt > ip {
				copy(all[(ip+1)*el:], n.entries[ip*el:cnt*el])
			}
			// 分裂点: 左叶保留 [0,mid), 右叶保留 [mid,total)。
			// 主键顺序追加(自增)时走 SequentialRight, mid = total-1 (右叶只留 1 条), 不再对半分。
			mid := splitPoint(total, PKSplitCfg, ip == cnt)
			// <= 路由约定: 左子叶所有 key 必须严格小于右子叶第一个 key (promoKey)
			// 如果 all[mid-1].key == all[mid].key, 右移 mid 直到分隔
			for mid < total-1 {
				if btCmp(btKey(all[(mid-1)*el:mid*el]), btKey(all[mid*el:(mid+1)*el])) < 0 {
					break
				}
				mid++
			}
			rl := btNew(handle, true, el)
			rc := total - mid
			copy(rl.entries, all[mid*el:total*el])
			rl.info.Count = uint32(rc)
			rl.info.Hdr.UseSize = uint32(page.IndexHeadSize) + uint32(rc*el)
			rl.info.Hdr.NextPage = n.info.Hdr.NextPage
			rl.info.Hdr.PrePage = n.info.Hdr.PageID
			n.info.Hdr.NextPage = rl.info.Hdr.PageID
			copy(n.entries, all[:mid*el])
			n.info.Count = uint32(mid)
			n.info.Hdr.UseSize = uint32(page.IndexHeadSize) + uint32(mid*el)
			btSave(handle, n)
			btSave(handle, rl)
			if rl.info.Hdr.NextPage != 0 {
				nl := btLoad(handle, rl.info.Hdr.NextPage)
				if nl != nil {
					nl.info.Hdr.PrePage = rl.info.Hdr.PageID
					btSave(handle, nl)
					nl.release()
				}
			}
			// 使用右子树最左叶的第一个 key 作为 promoKey
			pk := btLeftmostKey(handle, rl.info.Hdr.PageID)
			nr, err := btPromote(handle, rootPID, path, n.info.Hdr.PageID, pk, rl.info.Hdr.PageID)
			n.release()
			rl.release()
			return nr, 0, false, err
		}
		ci := btChildIdx(n, key)
		cpid := btPID(btEntry(n, ci))
		path = append(path, btPathEntry{nodePID: cur, childIdx: ci})
		n.release()
		cur = cpid
	}
}

func btPromote(handle page.HandleDb, rootPID uint32, path []btPathEntry, leftPID uint32, promoKey []byte, rightPID uint32) (uint32, error) {
	if len(path) == 0 {
		el := btIntEntryLen
		nr := btNew(handle, false, el)
		leftKey := btLeftmostKey(handle, leftPID)
		le := make([]byte, el)
		copy(le[:8], leftKey)
		btSetPID(le, leftPID)
		re := make([]byte, el)
		copy(re[:8], promoKey)
		btSetPID(re, rightPID)
		btInsEntry(nr, le, 0)
		btInsEntry(nr, re, 1)
		btSave(handle, nr)
		newRoot := nr.pageID()
		nr.release()
		return newRoot, nil
	}
	pi := len(path) - 1
	parent := btLoad(handle, path[pi].nodePID)
	if parent == nil {
		return rootPID, errors.New("parent load failed")
	}
	ci := path[pi].childIdx
	el := int(parent.info.EntryLen)
	// ⭐ FIX: 叶子分裂后, 左子树的最小 key 可能已变化
	// 必须更新 parent[ci].key = 左子树的新最小 key
	newLeftKey := btLeftmostKey(handle, leftPID)
	if newLeftKey == nil {
		parent.release()
		return rootPID, errors.New("leftmost key failed")
	}
	// 更新 parent[ci] 的 key 和 PID (内部节点条目: 键在偏移 0..8, 子页号在 8..12)
	copy(btEntry(parent, ci)[:8], newLeftKey)
	btSetPID(btEntry(parent, ci), leftPID)
	// 更新 key 后, parent[ci..] 可能不再有序
	// 需要将 parent[ci] 冒泡到正确位置, 然后插入 (promoKey, rightPID)
	// 1. 冒泡 parent[ci] 到正确位置 (如果 newLeftKey > parent[ci+1].key)
	for i := ci; i+1 < int(parent.info.Count); i++ {
		if btCmp(btKey(btEntry(parent, i)), btKey(btEntry(parent, i+1))) <= 0 {
			break // 已有序
		}
		// 交换 parent[i] 和 parent[i+1]
		tmp := make([]byte, el)
		copy(tmp, btEntry(parent, i))
		copy(btEntry(parent, i), btEntry(parent, i+1))
		copy(btEntry(parent, i+1), tmp)
	}
	// 2. 在排序位置插入 (promoKey, rightPID)
	newPos := int(parent.info.Count)
	for i := ci; i < int(parent.info.Count); i++ {
		if btCmp(btKey(btEntry(parent, i)), promoKey) >= 0 {
			newPos = i
			break
		}
	}
	for i := int(parent.info.Count); i > newPos; i-- {
		copy(btEntry(parent, i), btEntry(parent, i-1))
	}
	ne := make([]byte, el)
	copy(ne[:8], promoKey)
	btSetPID(ne, rightPID)
	copy(btEntry(parent, newPos), ne)
	parent.info.Count++
	parent.info.Hdr.UseSize += uint32(el)
	for i := 1; i < int(parent.info.Count); i++ {
		if btPID(btEntry(parent, i-1)) == btPID(btEntry(parent, i)) {
			println("!!! DUP PID in promote parent=", int(parent.info.Hdr.PageID), "at", i, "pid=", int(btPID(btEntry(parent, i))))
		}
	}
	if newPos > 0 && btCmp(btKey(btEntry(parent, newPos-1)), btKey(btEntry(parent, newPos))) > 0 {
		println("!!! promote unsorted at newPos=", newPos, "parent=", int(parent.info.Hdr.PageID))
	}
	if int(parent.info.Hdr.UseSize)+el <= btSplitThreshold() {
		btSave(handle, parent)
		parent.release()
		return rootPID, nil
	}
	cnt := int(parent.info.Count)
	// 分裂点: 左节点保留 [0,mid), 右节点保留 [mid,cnt)。
	// 顺序追加(新条目落在最右)时沿用主键的 SequentialRight 策略。
	mid := splitPoint(cnt, PKSplitCfg, newPos == cnt-1)
	rn := btNew(handle, false, el)
	rc := cnt - mid
	copy(rn.entries, parent.entries[mid*el:cnt*el])
	rn.info.Count = uint32(rc)
	rn.info.Hdr.UseSize = uint32(page.IndexHeadSize) + uint32(rc*el)
	parent.info.Count = uint32(mid)
	parent.info.Hdr.UseSize = uint32(page.IndexHeadSize) + uint32(mid*el)
	btSave(handle, parent)
	btSave(handle, rn)
	pk2 := make([]byte, 8)
	copy(pk2, btKey(btEntry(rn, 0)))
	ret, err := btPromote(handle, rootPID, path[:pi], parent.info.Hdr.PageID, pk2, rn.info.Hdr.PageID)
	parent.release()
	rn.release()
	return ret, err
}

// ======================== B+ 树删除 (递归 + 合并/借位) ========================

func BTreePKDelete(rootPID uint32, key []byte, handle page.HandleDb) (uint32, uint32, error) {
	if rootPID == 0 {
		return 0, 0, page.ErrNotFound
	}
	delDP, err := btDeleteRec(rootPID, key, nil, handle)
	if err != nil {
		return rootPID, delDP, err
	}
	// 收尾归一化:
	//  - 根为内部节点且只剩 1 个子节点 -> 用该子节点作为新根 (继续向上收缩)
	//  - 根为内部节点且已无条目 (整棵树被删空) -> 换成一个全新的空叶子,
	//    避免留下 "空内部节点" 这种会让查找读到脏数据的状态
	// 归一化过程中被跳过的旧根 (降高时旧根整体退役; 删空时旧根被换成新建空叶)。
	// 收集后在本函数末尾才归还, 避免 btNew 在同一轮里把这些页又分配回来。
	var retiredRoots []uint32
	for {
		n := btLoad(handle, rootPID)
		if n == nil || n.isLeaf() {
			break
		}
		if n.info.Count == 1 {
			childPID := btPID(btEntry(n, 0))
			if childPID == 0 {
				break
			}
			retiredRoots = append(retiredRoots, rootPID)
			rootPID = childPID
			continue
		}
		if n.info.Count == 0 {
			// 换成的必须是「空叶子」, 条目长度用叶布局 (14B), 不能用内部节点的 EntryLen
			leaf := btNew(handle, true, btLeafEntryLen)
			btSave(handle, leaf)
			retiredRoots = append(retiredRoots, rootPID)
			rootPID = leaf.pageID()
		}
		break
	}
	// 只归还退役的旧根自身: Count==1 时其子树仍是新根 (可达), 绝不能按子树回收;
	// Count==0 时该节点已是空内部节点, 其子页在上面的递归里已随空子树摘链归还。
	for _, pid := range retiredRoots {
		btReleasePage(handle, pid)
	}
	return rootPID, delDP, nil
}

func btCheckSorted(n *btNode) (int, bool) {
	cnt := int(n.info.Count)
	for i := 1; i < cnt; i++ {
		if btCmp(btKey(btEntry(n, i-1)), btKey(btEntry(n, i))) > 0 {
			return i, true
		}
	}
	return -1, false
}

// btDebugSorted 叶有序性校验开关。开着时 btInsEntry/btRmEntry 每次条目变更
// 都整叶 O(count) 扫一遍 —— 排障脚手架, 批量灌叶路径上占 ~9% CPU, 生产关闭。
// 排障时改回 true 即可恢复"乱序即打印"的行为。
const btDebugSorted = false

func btReportUnsorted(tag string, n *btNode) {
	if !btDebugSorted || n == nil || !n.isLeaf() {
		return
	}
	if bad, ok := btCheckSorted(n); ok {
		el := int(n.info.EntryLen)
		prev := btEntry(n, bad-1)
		cur := btEntry(n, bad)
		println("!!! UNSORTED", tag, "pid=", int(n.info.Hdr.PageID), "cnt=", int(n.info.Count),
			"el=", el, "bad=", bad, "prev=", int(btKey(prev)[7]), "cur=", int(btKey(cur)[7]))
	}
}

func btDeleteRec(nodePID uint32, key []byte, path []btPathEntry, handle page.HandleDb) (uint32, error) {
	node := btLoad(handle, nodePID)
	if node == nil {
		return nodePID, page.ErrNotFound
	}
	el := int(node.info.EntryLen)
	if node.isLeaf() {
		pos, ok := btLeafFind(node, key)
		if !ok {
			return nodePID, page.ErrNotFound
		}
		delDP := btPID(btEntry(node, pos))
		btRmEntry(node, pos)
		if node.info.Count == 0 && len(path) == 0 {
			btSave(handle, node)
			return delDP, nil
		}
		btSave(handle, node)
		if len(path) > 0 && node.info.Count > 0 {
			btTryMerge(handle, path, node, el)
		}
		return delDP, nil
	}
	ci := btChildIdx(node, key)
	childPID := btPID(btEntry(node, ci))
	newPath := make([]btPathEntry, len(path)+1)
	copy(newPath, path)
	newPath[len(path)] = btPathEntry{nodePID: nodePID, childIdx: ci}
	delDP, err := btDeleteRec(childPID, key, newPath, handle)
	if err != nil {
		return delDP, err
	}
	node = btLoad(handle, nodePID)
	if node == nil {
		return delDP, errors.New("reload node failed")
	}
	if node.info.Count > 0 {
		// 仅当原 child 条目仍在本节点的 ci 位置时才处理
		// (若该条目已被合并移除, 则说明子树已被兄弟吸收, 无需再处理)
		if ci >= 0 && ci < int(node.info.Count) && btPID(btEntry(node, ci)) == childPID {
			child := btLoad(handle, childPID)
			if child != nil {
				if child.info.Count == 0 {
					// 注意: 传的是**子节点**的条目长度, 不是本节点的 el —— 叶条目 (14B:
					// key8+dpID4+slot2) 与内部条目 (12B) 长度不同, 借位/合并时按父节点的 12B
					// 去搬运叶条目会截掉 slot 字节, 让被搬的行失效。
					btHandleUnderflow(node, ci, int(child.info.EntryLen), handle)
					// 借位/合并会直接改写本节点并保存, 必须重新加载
					node = btLoad(handle, nodePID)
					if node == nil {
						return delDP, errors.New("reload node failed after underflow")
					}
				} else {
					// 维护不变式: 父条目键 == 子树最小键
					// (删除叶子的最小键后, 父条目键会变旧)
					copy(btEntry(node, ci)[:8], btKey(btEntry(child, 0)))
				}
			}
		}
	}
	// 归一化: 内部节点若只剩一个 "已空" 的子节点 (没有兄弟可合并), 则该节点本身
	// 也变空。此时必须把该空子树整段从叶子链摘除并让本节点变为空 (Count=0),
	// 由上层继续向上收缩。否则会残留 "空内部节点", 破坏路由/查找。
	if !node.isLeaf() && node.info.Count == 1 {
		c0 := btPID(btEntry(node, 0))
		cn := btLoad(handle, c0)
		if cn != nil && cn.info.Count == 0 {
			btUnlinkSubtreeLeaves(handle, c0)
			node.info.Count = 0
			node.info.Hdr.UseSize = uint32(page.IndexHeadSize)
			btSave(handle, node)
			// c0 子树整段摘链后不再被任何父条目或叶子链引用, 连同其内部节点
			// 一并归还 —— 否则每删空一棵子树就漏掉一批索引页。
			btReleaseSubtree(handle, c0)
			return delDP, nil
		}
	}
	if node.info.Count == 0 {
		btSave(handle, node)
		return delDP, nil
	}
	btSave(handle, node)
	return delDP, nil
}

func btTryMerge(handle page.HandleDb, path []btPathEntry, node *btNode, el int) {
	if len(path) == 0 {
		return
	}
	nodePID := node.info.Hdr.PageID
	pp := path[len(path)-1].nodePID
	parent := btLoad(handle, pp)
	if parent == nil {
		return
	}
	ci := -1
	for i := 0; i < int(parent.info.Count); i++ {
		if btPID(btEntry(parent, i)) == node.info.Hdr.PageID {
			ci = i
			break
		}
	}
	if ci < 0 {
		return
	}
	if ci > 0 {
		leftPID := btPID(btEntry(parent, ci-1))
		left := btLoad(handle, leftPID)
		if left != nil && left.isLeaf() {
			combined := uint32(page.IndexHeadSize) + (left.info.Count+node.info.Count)*uint32(el)
			threshold := uint32(btSplitThreshold())
			if combined <= threshold {
				if leftPID == node.info.Hdr.PageID {
					println("!!! SELF tryMerge ci>0 pid=", int(leftPID))
				}
				rc := int(node.info.Count)
				copy(left.entries[int(left.info.Count)*el:], node.entries[:rc*el])
				left.info.Count += node.info.Count
				left.info.Hdr.UseSize += uint32(rc * el)
				left.info.Hdr.NextPage = node.info.Hdr.NextPage
				btSave(handle, left)
				btReportUnsorted("tryMergeIntoLeft", left)
				if left.info.Hdr.NextPage != 0 {
					nl := btLoad(handle, left.info.Hdr.NextPage)
					if nl != nil {
						nl.info.Hdr.PrePage = leftPID
						btSave(handle, nl)
					}
				}
				btRmEntry(parent, ci)
				btSave(handle, parent)
				// node 的条目已全部并入 left, 且指向 node 的父条目已被摘除,
				// 叶子链也已改接到 left 的 NextPage —— node 页至此彻底不可达。
				btReleasePage(handle, nodePID)
				return
			}
		}
	}
	if ci < int(parent.info.Count)-1 {
		rightPID := btPID(btEntry(parent, ci+1))
		right := btLoad(handle, rightPID)
		if right != nil && right.isLeaf() {
			combined := uint32(page.IndexHeadSize) + (node.info.Count+right.info.Count)*uint32(el)
			threshold := uint32(btSplitThreshold())
			if combined <= threshold {
				if rightPID == node.info.Hdr.PageID {
					println("!!! SELF tryMerge right pid=", int(rightPID))
				}
				rc := int(right.info.Count)
				copy(node.entries[int(node.info.Count)*el:], right.entries[:rc*el])
				node.info.Count += right.info.Count
				node.info.Hdr.UseSize += uint32(rc * el)
				node.info.Hdr.NextPage = right.info.Hdr.NextPage
				btSave(handle, node)
				btReportUnsorted("tryMergeAbsorbRight", node)
				if node.info.Hdr.NextPage != 0 {
					nl := btLoad(handle, node.info.Hdr.NextPage)
					if nl != nil {
						nl.info.Hdr.PrePage = node.info.Hdr.PageID
						btSave(handle, nl)
					}
				}
				btRmEntry(parent, ci+1)
				btSave(handle, parent)
				// right 的条目已全部并入 node, 且指向 right 的父条目已摘除,
				// 叶子链也已改接到 node 的 NextPage —— right 页至此彻底不可达。
				btReleasePage(handle, rightPID)
				return
			}
		}
	}
}

func btHandleUnderflow(parent *btNode, childIdx int, el int, handle page.HandleDb) {
	childPID := btPID(btEntry(parent, childIdx))
	child := btLoad(handle, childPID)
	if child == nil {
		return
	}
	if childIdx > 0 {
		leftPID := btPID(btEntry(parent, childIdx-1))
		leftSib := btLoad(handle, leftPID)
		if leftSib != nil && int(leftSib.info.Count) > 1 {
			borrowed := make([]byte, el)
			copy(borrowed, btEntry(leftSib, int(leftSib.info.Count)-1))
			btRmEntry(leftSib, int(leftSib.info.Count)-1)
			btInsEntry(child, borrowed, 0)
			copy(btEntry(parent, childIdx)[:8], btKey(btEntry(child, 0)))
			btSave(handle, leftSib)
			btReportUnsorted("borrowLeft-child", child)
			btReportUnsorted("borrowLeft-leftSib", leftSib)
			btSave(handle, child)
			btSave(handle, parent)
			return
		}
	}
	if childIdx < int(parent.info.Count)-1 {
		rightPID := btPID(btEntry(parent, childIdx+1))
		rightSib := btLoad(handle, rightPID)
		if rightSib != nil && int(rightSib.info.Count) > 1 {
			borrowed := make([]byte, el)
			copy(borrowed, btEntry(rightSib, 0))
			btRmEntry(rightSib, 0)
			// 右兄弟的首条被追加到 child 尾部: child 的最小键不变
			btInsEntry(child, borrowed, int(child.info.Count))
			copy(btEntry(parent, childIdx)[:8], btKey(btEntry(child, 0)))
			// 右兄弟失去首条后最小键变化, 必须同步其父条目
			copy(btEntry(parent, childIdx+1)[:8], btKey(btEntry(rightSib, 0)))
			btSave(handle, rightSib)
			btReportUnsorted("borrowRight-child", child)
			btReportUnsorted("borrowRight-rightSib", rightSib)
			btSave(handle, child)
			btSave(handle, parent)
			return
		}
	}
	if childIdx > 0 {
		btMergeChildren(handle, parent, childIdx-1, el)
	} else if childIdx < int(parent.info.Count)-1 {
		btMergeChildren(handle, parent, childIdx, el)
	}
}

func btMergeChildren(handle page.HandleDb, parent *btNode, sepIdx int, el int) {
	leftPID := btPID(btEntry(parent, sepIdx))
	rightPID := btPID(btEntry(parent, sepIdx+1))
	left := btLoad(handle, leftPID)
	right := btLoad(handle, rightPID)
	if left == nil || right == nil {
		return
	}
	// 注意: 本实现的父条目键是 "子树最小键的副本", 不是经典 B+ 树里独立的分隔键。
	// 因此合并内部节点时直接把右兄弟的条目追加到左兄弟即可,
	// 绝不能把 parent[sepIdx] 下放 (那会写出一条指向左兄弟自身的自引用条目)。
	rc := int(right.info.Count)
	if leftPID == rightPID {
		println("!!! SELF btMergeChildren pid=", int(leftPID), "sepIdx=", sepIdx)
	}
	copy(left.entries[int(left.info.Count)*el:], right.entries[:rc*el])
	left.info.Count += right.info.Count
	left.info.Hdr.UseSize += uint32(rc * el)
	if left.isLeaf() {
		left.info.Hdr.NextPage = right.info.Hdr.NextPage
	}
	btSave(handle, left)
	btReportUnsorted("mergeChildren-left", left)
	if left.isLeaf() && left.info.Hdr.NextPage != 0 {
		nl := btLoad(handle, left.info.Hdr.NextPage)
		if nl != nil {
			nl.info.Hdr.PrePage = leftPID
			btSave(handle, nl)
		}
	}
	btRmEntry(parent, sepIdx+1)
	btSetPID(btEntry(parent, sepIdx), leftPID)
	// 左节点若原本是空节点, 合并后其最小键变为右兄弟的最小键, 必须同步父条目键,
	// 否则会出现 "父键 != 子树最小键", 破坏路由。
	if left.info.Count > 0 {
		copy(btEntry(parent, sepIdx)[:8], btKey(btEntry(left, 0)))
	}
	btSave(handle, parent)
	// right 的条目已全部并入 left, 父条目 sepIdx+1 已摘除 (叶子链在上面的
	// NextPage/PrePage 重接中已绕过 right) —— right 子树至此彻底不可达。
	btReleasePage(handle, rightPID)
}

// ======================== DataPage 操作 ========================

// FindSpaceInLeaf 在叶节点中找到有空间的 DataPage
func FindSpaceInLeaf(handle page.HandleDb, rootPID uint32, rowSize uint32, key []byte) (uint32, bool) {
	cur := rootPID
	for {
		n := btLoad(handle, cur)
		if n == nil {
			return 0, false
		}
		if n.isLeaf() {
			pos := btLeafPos(n, key)
			if pos >= int(n.info.Count) {
				pos = int(n.info.Count) - 1
			}
			if pos < 0 {
				return 0, false
			}
			dpID := btPID(btEntry(n, pos))
			dp, err := datapage.Load(handle, dpID)
			if err == nil && !dp.IsFull(rowSize) {
				return dpID, true
			}
			if n.info.Hdr.PrePage != 0 {
				prevN := btLoad(handle, n.info.Hdr.PrePage)
				if prevN != nil && prevN.isLeaf() && prevN.info.Count > 0 {
					prevDPID := btPID(btEntry(prevN, int(prevN.info.Count)-1))
					prevDP, err := datapage.Load(handle, prevDPID)
					if err == nil && !prevDP.IsFull(rowSize) {
						return prevDPID, true
					}
				}
			}
			if n.info.Hdr.NextPage != 0 {
				nextN := btLoad(handle, n.info.Hdr.NextPage)
				if nextN != nil && nextN.isLeaf() && nextN.info.Count > 0 {
					nextDPID := btPID(btEntry(nextN, 0))
					nextDP, err := datapage.Load(handle, nextDPID)
					if err == nil && !nextDP.IsFull(rowSize) {
						return nextDPID, true
					}
				}
			}
			return 0, false
		}
		cur = btRoute(n, key)
	}
}

// LastLeafDPID 路由到目标叶子, 返回其最后一个条目指向的 DataPage ID —— 即"当前正在
// 追加的那一页"。不判断是否装得下: 调用方拿到它之后可以按配置整理一次再决定是否新开页
// (见 datapage.CompactForInsert)。
//
// 只检查路由目标叶子, 不跨叶子查找 (保证 B+ 树不变性)。
func LastLeafDPID(handle page.HandleDb, rootPID uint32, key []byte) (uint32, bool) {
	cur := rootPID
	for {
		n := btLoad(handle, cur)
		if n == nil {
			return 0, false
		}
		if n.isLeaf() {
			if n.info.Count == 0 {
				return 0, false
			}
			return btPID(btEntry(n, int(n.info.Count)-1)), true
		}
		cur = btRoute(n, key)
	}
}

// FindLastLeafDP 路由到目标叶子, 检查最后一个条目的 DataPage 是否有空间
// 只检查路由目标叶子, 不跨叶子查找 (保证 B+ 树不变性)
func FindLastLeafDP(handle page.HandleDb, rootPID uint32, rowSize uint32, key []byte) (uint32, bool) {
	dpID, ok := LastLeafDPID(handle, rootPID, key)
	if !ok {
		return 0, false
	}
	dp, err := datapage.Load(handle, dpID)
	if err != nil || dp.IsFull(rowSize) {
		return 0, false
	}
	return dpID, true
}

// GetDataPageIDs 返回所有 DataPage ID (去重, 用于全表扫描)
func GetDataPageIDs(handle page.HandleDb, rootPID uint32) []uint32 {
	if rootPID == 0 {
		return nil
	}
	var dpIDs []uint32
	seen := make(map[uint32]bool)
	walked := make(map[uint32]bool) // 叶链成环保护: 坏 next 会让下面这个 for 永远转不完
	cur := rootPID
	for {
		n := btLoadNode(handle, cur)
		if n == nil {
			break
		}
		if n.isLeaf() {
			break
		}
		cur = btChildPID(n, btEntry(n, 0))
	}
	for cur != 0 {
		if walked[cur] {
			break
		}
		walked[cur] = true
		n := btLoadNode(handle, cur)
		if n == nil {
			break
		}
		for i := 0; i < int(n.info.Count); i++ {
			dpID := btPID(btEntry(n, i))
			if dpID == 0 || seen[dpID] {
				continue
			}
			// 合法性兜底: 条目错位解析出来的"页号"绝不能进入回收列表
			if handle.ValidPage != nil && !handle.ValidPage(dpID) {
				continue
			}
			seen[dpID] = true
			dpIDs = append(dpIDs, dpID)
		}
		cur = n.info.Hdr.NextPage
	}
	return dpIDs
}

// FindDataPageIDByPK 定位存放 pkID=pkKeyBE 那一行的 DataPage ID。
// 用途: 当行因 schema 演进变长、页内已装不下 (RebuildWith 返回 errRowTooLarge) 时,
// 需要先定位旧页才能把行搬到新页并改写 B+ 树叶条目。
//
// 实现: 沿叶链逐叶二分找键 (叶键就是 pkID), 命中即拿到 (dpID, slot), 再用
// ReadRowBySlotPK 校验该槽位确实是这一行 (陈旧槽位防御) 才返回。比旧实现「每叶每条
// 目都 Load 一次 DataPage 再整页扫 pkID」少掉全部 DataPage 全页扫描。
func FindDataPageIDByPK(rootPID uint32, handle page.HandleDb, pkKeyBE []byte) (uint32, bool) {
	if rootPID == 0 {
		return 0, false
	}
	cur := btFirstLeaf(handle, rootPID)
	for cur != 0 {
		n := btLoad(handle, cur)
		if n == nil {
			break
		}
		if pos, ok := btLeafFind(n, pkKeyBE); ok {
			e := btEntry(n, pos)
			dpID := btPID(e)
			if dp, err := datapage.Load(handle, dpID); err == nil {
				if _, ok := dp.ReadRowBySlotPK(uint32(btSlot(e)), pkKeyBE); ok {
					return dpID, true
				}
			}
		}
		cur = n.info.Hdr.NextPage
	}
	return 0, false
}

// ScanAllLeavesForPK 扫描所有叶子查找 pkID(LE)。
// 这是 BTreePKSearch 之外的兜底路径: 树结构异常 (路由坏掉) 时仍能捞到行。
func ScanAllLeavesForPK(rootPID uint32, handle page.HandleDb, pkKeyBE []byte) ([]byte, bool) {
	if rootPID == 0 {
		return nil, false
	}
	cur := btFirstLeaf(handle, rootPID)
	for cur != 0 {
		n := btLoad(handle, cur)
		if n == nil {
			break
		}
		// 叶内二分: 键即 pkID, 命中就直接拿到 (dpID, slot)
		if pos, ok := btLeafFind(n, pkKeyBE); ok {
			e := btEntry(n, pos)
			if dp, err := datapage.Load(handle, btPID(e)); err == nil {
				if rowData, ok := dp.ReadRowBySlotPK(uint32(btSlot(e)), pkKeyBE); ok {
					return rowData, true
				}
			}
		}
		cur = n.info.Hdr.NextPage
	}
	return nil, false
}

// ScanAndDelete 扫描所有叶子按 indexKey 删除
func ScanAndDelete(rootPID uint32, handle page.HandleDb, indexKey, pkKeyBE []byte) error {
	if rootPID == 0 {
		return page.ErrNotFound
	}
	cur := rootPID
	for {
		n := btLoad(handle, cur)
		if n == nil {
			return page.ErrNotFound
		}
		if n.isLeaf() {
			break
		}
		cur = btPID(btEntry(n, 0))
	}
	for cur != 0 {
		n := btLoad(handle, cur)
		if n == nil {
			break
		}
		for i := 0; i < int(n.info.Count); i++ {
			e := btEntry(n, i)
			// 叶键就是 pkID: 只处理键匹配的条目, 其余条目所在的页根本不用碰
			if btCmp(btKey(e), indexKey) != 0 {
				continue
			}
			dp, err := datapage.Load(handle, btPID(e))
			if err == nil {
				slotID := uint32(btSlot(e))
				deleted := false
				if _, ok := dp.ReadRowBySlotPK(slotID, indexKey); ok {
					// 槽位命中 → O(1) 删除
					deleted = dp.DeleteBySlot(slotID)
				} else if dp.DeleteByIndexKey(indexKey, 8) {
					deleted = true
				} else if dp.DeleteByPK(pkKeyBE) {
					deleted = true
				}
				if deleted {
					if dp.AllDeleted() {
						newRoot, _, err := BTreePKDelete(rootPID, indexKey, handle)
						if err != nil {
							return err
						}
						_ = newRoot
					} else {
						dp.Save()
					}
					return nil
				}
			}
		}
		cur = n.info.Hdr.NextPage
	}
	return page.ErrNotFound
}

// ScanAndDeleteByPK 扫描所有叶子按 pkID 删除
func ScanAndDeleteByPK(rootPID uint32, handle page.HandleDb, pkKeyBE []byte) error {
	if rootPID == 0 {
		return page.ErrNotFound
	}
	cur := rootPID
	for {
		n := btLoad(handle, cur)
		if n == nil {
			return page.ErrNotFound
		}
		if n.isLeaf() {
			break
		}
		cur = btPID(btEntry(n, 0))
	}
	for cur != 0 {
		n := btLoad(handle, cur)
		if n == nil {
			break
		}
		for i := 0; i < int(n.info.Count); i++ {
			e := btEntry(n, i)
			if btCmp(btKey(e), pkKeyBE) != 0 {
				continue
			}
			dp, err := datapage.Load(handle, btPID(e))
			if err == nil {
				slotID := uint32(btSlot(e))
				deleted := false
				if _, ok := dp.ReadRowBySlotPK(slotID, pkKeyBE); ok {
					deleted = dp.DeleteBySlot(slotID)
				} else if dp.DeleteByPK(pkKeyBE) {
					deleted = true
				}
				if deleted {
					if dp.AllDeleted() {
						newRoot, _, err := BTreePKDelete(rootPID, pkKeyBE, handle)
						if err != nil {
							return err
						}
						_ = newRoot
					} else {
						dp.Save()
					}
					return nil
				}
			}
		}
		cur = n.info.Hdr.NextPage
	}
	return page.ErrNotFound
}

// ScanAndUpdate 扫描所有叶子按 pkID 更新
func ScanAndUpdate(rootPID uint32, handle page.HandleDb, pkKeyBE, storeRowData []byte) error {
	if rootPID == 0 {
		return page.ErrNotFound
	}
	cur := rootPID
	for {
		n := btLoad(handle, cur)
		if n == nil {
			return page.ErrNotFound
		}
		if n.isLeaf() {
			break
		}
		cur = btPID(btEntry(n, 0))
	}
	for cur != 0 {
		n := btLoad(handle, cur)
		if n == nil {
			break
		}
		for i := 0; i < int(n.info.Count); i++ {
			e := btEntry(n, i)
			if btCmp(btKey(e), pkKeyBE) != 0 {
				continue
			}
			dp, err := datapage.Load(handle, btPID(e))
			if err == nil {
				slotID := uint32(btSlot(e))
				if err := dp.UpdateBySlot(slotID, storeRowData); err == nil {
					return nil
				} else if err := dp.RebuildWithSlot(slotID, storeRowData); err == nil {
					// 行长变化 (schema 演进后旧行补新字段) → 页内重建, 支持变长
					return nil
				}
			}
		}
		cur = n.info.Hdr.NextPage
	}
	return page.ErrNotFound
}

// BTreeValidate 检查 B+ 树内部节点条目排序
func BTreeValidate(handle page.HandleDb, rootPID uint32) []string {
	var issues []string
	if rootPID == 0 {
		return issues
	}
	var check func(pid uint32, depth int)
	check = func(pid uint32, depth int) {
		n := btLoad(handle, pid)
		if n == nil {
			issues = append(issues, fmt.Sprintf("d=%d: load fail pid=%d", depth, pid))
			return
		}
		cnt := int(n.info.Count)
		if n.isLeaf() {
			for i := 1; i < cnt; i++ {
				pk := btKey(btEntry(n, i-1))
				ck := btKey(btEntry(n, i))
				if btCmp(pk, ck) >= 0 {
					issues = append(issues, fmt.Sprintf("d=%d leaf=%d: unsorted [%d]>=[%d]", depth, pid, i-1, i))
				}
			}
			return
		}
		for i := 1; i < cnt; i++ {
			pk := btKey(btEntry(n, i-1))
			ck := btKey(btEntry(n, i))
			if btCmp(pk, ck) > 0 {
				issues = append(issues, fmt.Sprintf("d=%d int=%d: UNSORTED [%d]>[%d]", depth, pid, i-1, i))
			}
		}
		for i := 0; i < cnt; i++ {
			childPID := btPID(btEntry(n, i))
			check(childPID, depth+1)
		}
	}
	check(rootPID, 0)
	return issues
}

// BTreeTraceSearch 搜索并打印路由路径
func BTreeTraceSearch(rootPID uint32, key []byte, handle page.HandleDb) {
	if rootPID == 0 {
		fmt.Println("Tree is empty")
		return
	}
	cur := rootPID
	depth := 0
	for {
		n := btLoad(handle, cur)
		if n == nil {
			fmt.Printf("  d=%d: LOAD FAIL pid=%d\n", depth, cur)
			return
		}
		cnt := int(n.info.Count)
		if n.isLeaf() {
			fk := uint64(0)
			lk := uint64(0)
			if cnt > 0 {
				fk = binary.BigEndian.Uint64(btKey(btEntry(n, 0)))
				lk = binary.BigEndian.Uint64(btKey(btEntry(n, cnt-1)))
			}
			pos, found := btLeafFind(n, key)
			fmt.Printf("  d=%d LEAF pid=%d cnt=%d first=%d last=%d pos=%d found=%v\n",
				depth, cur, cnt, fk, lk, pos, found)
			return
		}
		fmt.Printf("  d=%d INT pid=%d cnt=%d [", depth, cur, cnt)
		for i := 0; i < cnt; i++ {
			k := binary.BigEndian.Uint64(btKey(btEntry(n, i)))
			p := btPID(btEntry(n, i))
			if i > 0 {
				fmt.Printf(", ")
			}
			if i < 3 || i >= cnt-2 {
				fmt.Printf("(k=%d,p=%d)", k, p)
			} else if i == 3 {
				fmt.Printf("...")
			}
		}
		fmt.Printf("]\n")
		ci := btChildIdx(n, key)
		fmt.Printf("    btChildIdx(%d) = %d\n", binary.BigEndian.Uint64(key), ci)
		cur = btPID(btEntry(n, ci))
		depth++
	}
}

// CollectAllPageIDs 收集 B+ 树所有页 ID (包括内部节点和叶子节点)
// 用于 DropTable 时释放所有 B+ 树占用的页
func CollectAllPageIDs(handle page.HandleDb, rootPID uint32) []uint32 {
	if rootPID == 0 {
		return nil
	}
	var pages []uint32
	seen := map[uint32]bool{} // 去重 + 成环保护: 同一子页被多个父条目引用会让 BFS 无限膨胀
	// BFS 遍历所有节点
	queue := []uint32{rootPID}
	for len(queue) > 0 {
		pid := queue[0]
		queue = queue[1:]
		if seen[pid] {
			continue
		}
		seen[pid] = true
		pages = append(pages, pid)
		// 必须用校验版加载: 非 B+ 页被强行按偏移解析时, Count 会是行键这类大整数,
		// btPID 随之吐出从未分配过的垃圾页号 —— 它们会被 FreePage 直接写进空闲链。
		n := btLoadNode(handle, pid)
		if n == nil {
			continue
		}
		if !n.isLeaf() {
			// 内部节点: 收集所有子节点
			for i := 0; i < int(n.info.Count); i++ {
				childPID := btChildPID(n, btEntry(n, i))
				if childPID == 0 {
					continue
				}
				if handle.ValidPage != nil && !handle.ValidPage(childPID) {
					continue
				}
				queue = append(queue, childPID)
			}
		}
	}
	return pages
}

// CollectLeafEntries 返回整棵树所有叶子条目 (副本), 顺序为先序遍历。
//
// 主键索引的每条叶子条目形如 [key(8)][dpID(4)][slot(2)] (见 btLeafEntryLen), 即「每行一条」。
// 该函数供诊断与回归测试使用: 统计条目数并与存活行数比对, 就能发现删除后残留在树里的
// 陈旧条目 (它们指向随后被回收复用的页, 是 not-data-page / freeLive 的根源)。
func CollectLeafEntries(handle page.HandleDb, rootPID uint32) [][]byte {
	if rootPID == 0 {
		return nil
	}
	var out [][]byte
	queue := []uint32{rootPID}
	seen := map[uint32]bool{}
	for len(queue) > 0 {
		pid := queue[0]
		queue = queue[1:]
		if seen[pid] {
			continue
		}
		seen[pid] = true
		n := btLoad(handle, pid)
		if n == nil {
			continue
		}
		if n.isLeaf() {
			for i := 0; i < int(n.info.Count); i++ {
				e := btEntry(n, i)
				cp := make([]byte, len(e))
				copy(cp, e)
				out = append(out, cp)
			}
			continue
		}
		for i := 0; i < int(n.info.Count); i++ {
			// 与 CollectAllPageIDs 同理: 内部条目子页号偏移随索引类型而变。
			childPID := btChildPID(n, btEntry(n, i))
			if childPID != 0 {
				queue = append(queue, childPID)
			}
		}
	}
	return out
}

// ======================== 主键叶批量插入 ========================

// PKBulkItem 主键 B+ 树叶条目原料: [pkID(8B BE)] → (数据页, 槽位)。
type PKBulkItem struct {
	Key  [8]byte // pkID BigEndian
	DPID uint32
	Slot uint16
}

// btDescendLeaf 从 rootPID 沿 key 下降到路由目标叶 (持有节点缓冲, 调用方负责 release)。
func btDescendLeaf(handle page.HandleDb, rootPID uint32, key []byte) *btNode {
	cur := rootPID
	for {
		n := btLoad(handle, cur)
		if n == nil {
			return nil
		}
		if n.isLeaf() {
			return n
		}
		cur = btRoute(n, key)
		n.release()
	}
}

// BTreePKBulkInsert 批量插入主键叶条目。items 必须按 Key 升序 (调用方保证;
// 自增主键的批量插入天然满足)。
//
// 与逐条 BTreePKInsert 的区别只有"叶节点驻内存": 排序批量下目标叶单调右移,
// 驻留叶加载一次、页内定位插入、装满才落盘 —— 逐条路径每条都要"下降加载整页 +
// 整页回写" (16KB×2 memcpy/条), 批量路径摊到 ~每千条一次。
//
// 正确性关键 (与 SBTBulkInsert 同一套约定): 驻留叶的可用区间是 [叶首键, 上界),
// 上界 = 右邻叶首键 (叶链 NextPage)。判定"该不该留在当前叶"必须用上界 ——
// 用"大于叶末键"会在条目越过右邻叶首键时仍往左叶追加, 破坏父层路由。
// 单条 BTreePKInsert 兜底只发生在三种边角: 叶满要分裂、条目成为叶最小键
// (父键同步)、链尾之后还有更大键 —— 分裂/promote/父键同步全套复用既有逻辑。
//
// 返回新根 (与 BTreePKInsert 一致, 树未改根时等于 rootPID)。
func BTreePKBulkInsert(rootPID uint32, items []PKBulkItem, handle page.HandleDb) (uint32, error) {
	if len(items) == 0 {
		return rootPID, nil
	}
	el := btLeafEntryLen
	th := btSplitThreshold()
	var leaf *btNode
	var upper []byte // 驻留叶键上界 (右邻叶首键的副本); nil = 最右叶 (上界无穷)

	// upperOf 计算叶的上界。副本必须复制出来 —— 右邻叶还回缓冲池后内存会被复用。
	// 空叶 (健康树上不应出现) 一律当"无上界", 交给后续单条兜底修正。
	upperOf := func(n *btNode) []byte {
		if n.info.Hdr.NextPage == 0 {
			return nil
		}
		nl := btLoad(handle, n.info.Hdr.NextPage)
		if nl == nil || nl.info.Count == 0 {
			if nl != nil {
				nl.release()
			}
			return nil
		}
		up := make([]byte, 8)
		copy(up, btKey(btEntry(nl, 0)))
		nl.release()
		return up
	}

	// flushLeaf 把驻留叶落盘 (单条兜底 / 批次结束前必须先做: 兜底路径会自己 btLoad
	// 这张页, 不落盘就会读到池里的陈旧副本, 驻留期间的插入全丢)。
	flushLeaf := func() {
		if leaf != nil {
			btSave(handle, leaf)
			if rootPID == 0 {
				rootPID = leaf.pageID()
			}
			leaf.release()
			leaf = nil
		}
	}

	// descend 按 key 从根下降到路由目标叶, 并取上界。
	descend := func(key []byte) error {
		n := btDescendLeaf(handle, rootPID, key)
		if n == nil {
			return errors.New("pk bulk insert: load failed")
		}
		leaf = n
		upper = upperOf(n)
		return nil
	}

	// stepTo 沿叶链从 from 走到第一张非空叶 (或链尾) 并驻留。
	stepTo := func(from uint32) error {
		cur := from
		for {
			nd := btLoad(handle, cur)
			if nd == nil {
				return errors.New("pk bulk insert: chain load failed")
			}
			if nd.info.Count > 0 || nd.info.Hdr.NextPage == 0 {
				leaf = nd
				upper = upperOf(nd)
				return nil
			}
			cur = nd.info.Hdr.NextPage // 空叶没有承载键, 直接跳过
			nd.release()
		}
	}

	for i := range items {
		it := &items[i]
		if leaf == nil {
			if rootPID == 0 {
				leaf = btNew(handle, true, btLeafEntryLen)
				upper = nil
			} else if err := descend(it.Key[:]); err != nil {
				return rootPID, err
			}
		}
		// 越过驻留叶上界: 目标叶在右侧。落盘驻留叶后沿叶链右移 (比重新下降省
		// 2/3 页加载); 链走尽了重新下降, 让路由把后续大键送进(新的)最右叶。
		for upper != nil && btCmp(it.Key[:], upper) >= 0 {
			next := leaf.info.Hdr.NextPage
			flushLeaf()
			if next == 0 {
				break
			}
			if err := stepTo(next); err != nil {
				return rootPID, err
			}
		}
		if leaf == nil {
			if err := descend(it.Key[:]); err != nil {
				return rootPID, err
			}
		}

		if int(leaf.info.Hdr.UseSize)+el > th {
			// 叶满: 落盘驻留叶, 单条兜底 (分裂/promote/叶链维护全套)
			flushLeaf()
			nr, _, _, err := BTreePKInsert(rootPID, it.Key[:], it.DPID, it.Slot, handle)
			if err != nil {
				return rootPID, err
			}
			rootPID = nr
			continue
		}
		cnt := int(leaf.info.Count)
		ip, _ := btLeafFind(leaf, it.Key[:])
		if ip == 0 && cnt > 0 {
			// 新条目将成为叶最小键: "父条目键 == 子树最小键"不变式需要沿路径
			// 向上同步, 交给单条路径 (整批有序, 只可能发生在批首条目)
			flushLeaf()
			nr, _, _, err := BTreePKInsert(rootPID, it.Key[:], it.DPID, it.Slot, handle)
			if err != nil {
				return rootPID, err
			}
			rootPID = nr
			continue
		}
		e := make([]byte, el)
		copy(e[:8], it.Key[:])
		btSetPID(e, it.DPID)
		btSetSlot(e, it.Slot)
		btInsEntry(leaf, e, ip)
	}
	flushLeaf()
	return rootPID, nil
}