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

package btree

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

// ======================== 二级 B+ 树索引操作 ========================
// 叶节点   [key(keyLen) + PKID(8B)]                entryLen = keyLen + 8
// 内节点   [key(keyLen) + PKID(8B) + childPID(4B)] entryLen = keyLen + 12
//
// 全树条目按复合键 (key, pkID) 全局有序; 内部条目的前 keyLen+8 字节 = 其子树的
// 最小复合键 [key|pkID] (分隔键), 末 4 字节 = 子页号。复合分隔键让 (key, pkID)
// 的点路由精确成立: 删除/插入一次下降即命中目标叶, 不必在重复 key 跨越的兄弟
// 子树间逐棵试探 (旧格式分隔键只有 key, Age 每值 2 万条跨几十个叶时删除退化为
// 平均半个重复区段的页加载, 实测 ~2.5ms/行)。
// 分裂: 统一 50/50 (不做顺序优化, 二级索引键可能非自增)
// 合并: InnoDB 风格 — 删除后检查当前页+兄弟页能否放进一页

// ----- 节点 I/O -----

func sbtNew(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.IndexTypeBTree
	info.EntryLen = uint16(entryLen)
	info.Count = 0
	if isLeaf {
		info.IsLeaf = 1
	}
	return &btNode{buf: buf, info: info, entries: buf[page.IndexHeadSize:]}
}

func sbtLoad(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:]}
}

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

// ----- 条目操作 -----

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

func sbtLeafKey(e []byte) []byte          { return e[:len(e)-8] }
func sbtPKID(e []byte) uint64             { return binary.BigEndian.Uint64(e[len(e)-8:]) }
func sbtIntKey(e []byte) []byte           { return e[:len(e)-12] }                       // 内部条目的 key 部分
func sbtIntPKID(e []byte) uint64          { return binary.BigEndian.Uint64(e[len(e)-12:]) } // 内部条目的 pkID 部分
func sbtChildPID(e []byte) uint32         { return binary.BigEndian.Uint32(e[len(e)-4:]) }
func sbtSetChildPID(e []byte, pid uint32) { binary.BigEndian.PutUint32(e[len(e)-4:], pid) }

func sbtCmp(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
		}
	}
	return 0
}

// sbtCmpKeyPK 复合键 (key, pkID) 的全序比较: 先比 key, 再比 pkID。
func sbtCmpKeyPK(keyA []byte, pkA uint64, keyB []byte, pkB uint64) int {
	if c := sbtCmp(keyA, keyB); c != 0 {
		return c
	}
	switch {
	case pkA < pkB:
		return -1
	case pkA > pkB:
		return 1
	}
	return 0
}

// sbtNodeMinKey 返回节点最左条目的复合键 [key|pkID] (长度 keyLen+8)。
// 叶节点取第一个条目本身 (叶条目就是 [key|pkID]); 内部节点取第一条目的前
// EntryLen-4 字节 (即其子树的最小复合键)。该值用于维护
// "父条目复合键 == 子树最小复合键" 不变式。
func sbtNodeMinKey(n *btNode) []byte {
	if n == nil || n.info.Count == 0 {
		return nil
	}
	e := sbtEntry(n, 0)
	if n.isLeaf() {
		return e
	}
	return e[:len(e)-4]
}

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

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

// sbtUnlinkSubtreeLeaves 把以 pid 为根的子树所覆盖的那一段叶子, 从叶子双向链表中整体摘除。
// 仅在该子树已被确认 "完全为空" 时调用, 用于避免删除收尾阶段残留空叶子导致
// "叶子链长度 != DFS 可达叶子数"。
func sbtUnlinkSubtreeLeaves(handle page.HandleDb, pid uint32) {
	first := sbtFirstLeaf(handle, pid)
	if first == 0 {
		return
	}
	last := sbtLastLeaf(handle, pid)
	fl := sbtLoad(handle, first)
	ll := sbtLoad(handle, last)
	if fl == nil || ll == nil {
		return
	}
	pre := fl.info.Hdr.PrePage
	next := ll.info.Hdr.NextPage
	if pre != 0 {
		pr := sbtLoad(handle, pre)
		if pr != nil {
			pr.info.Hdr.NextPage = next
			sbtSave(handle, pr)
		}
	}
	if next != 0 {
		nl := sbtLoad(handle, next)
		if nl != nil {
			nl.info.Hdr.PrePage = pre
			sbtSave(handle, nl)
		}
	}
}

// ----- 内节点路由 -----

func sbtRoute(n *btNode, key []byte) uint32 {
	cnt := int(n.info.Count)
	if cnt <= 0 {
		return 0 // 防御: 空节点不应作为路由节点
	}
	// 二分查找: 找最后一个 entry.key <= key 的位置
	left, right := 0, cnt-1
	result := 0
	for left <= right {
		mid := (left + right) / 2
		if sbtCmp(sbtIntKey(sbtEntry(n, mid)), key) <= 0 {
			result = mid
			left = mid + 1 // 继续向右找
		} else {
			right = mid - 1
		}
	}
	return sbtChildPID(sbtEntry(n, result))
}

func sbtChildIdx(n *btNode, key []byte) int {
	cnt := int(n.info.Count)
	if cnt <= 0 {
		return 0 // 防御: 空节点不应作为路由节点
	}
	// 二分查找: 找最后一个 entry.key <= key 的位置 (只按 key 路由)
	left, right := 0, cnt-1
	result := 0
	for left <= right {
		mid := (left + right) / 2
		if sbtCmp(sbtIntKey(sbtEntry(n, mid)), key) <= 0 {
			result = mid
			left = mid + 1 // 继续向右找
		} else {
			right = mid - 1
		}
	}
	return result
}

// sbtChildIdxPK 复合键路由: 找最后一个 "子树最小复合键 <= (key, pkID)" 的子树下标。
//
// 全树按 (key, pkID) 全局有序时, 各子树在复合键空间上是连续区间, 该路由唯一
// 确定目标条目所在 (或应当插入) 的子树 —— 插入与 (key,pkID) 精确删除都用它,
// 一次下降即可命中, 无需向左试探兄弟子树。
func sbtChildIdxPK(n *btNode, key []byte, pkID uint64) int {
	cnt := int(n.info.Count)
	if cnt <= 0 {
		return 0 // 防御: 空节点不应作为路由节点
	}
	left, right := 0, cnt-1
	result := 0
	for left <= right {
		mid := (left + right) / 2
		e := sbtEntry(n, mid)
		if sbtCmpKeyPK(sbtIntKey(e), sbtIntPKID(e), key, pkID) <= 0 {
			result = mid
			left = mid + 1
		} else {
			right = mid - 1
		}
	}
	return result
}

// ----- 叶节点操作 -----

func sbtInsEntry(n *btNode, entry []byte, pos int) {
	el := int(n.info.EntryLen)
	cnt := int(n.info.Count)
	// 整段后移一格 (块拷贝): 逐条目循环移位在千条目级叶上每插一条要搬千次,
	// 批量灌叶路径实测占 ~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)
}

func sbtRmEntry(n *btNode, pos int) {
	el := int(n.info.EntryLen)
	cnt := int(n.info.Count)
	for i := pos; i < cnt-1; i++ {
		copy(sbtEntry(n, i), sbtEntry(n, i+1))
	}
	n.info.Count--
	n.info.Hdr.UseSize -= uint32(el)
}

func sbtLeafFind(n *btNode, key []byte) (int, bool) {
	el := int(n.info.EntryLen)
	cnt := int(n.info.Count)
	if cnt == 0 {
		return 0, false
	}
	left, right := 0, cnt-1
	for left <= right {
		mid := (left + right) / 2
		cmp := sbtCmp(sbtEntry(n, mid)[:el-8], key)
		if cmp == 0 {
			return mid, true
		} else if cmp < 0 {
			left = mid + 1
		} else {
			right = mid - 1
		}
	}
	return left, false
}

func sbtLeafPos(n *btNode, key []byte) int {
	pos, _ := sbtLeafFind(n, key)
	return pos
}

// sbtLeafLowerBound 返回第一个 key >= 目标 key 的条目下标 (leftmost lower_bound)。
//
// 二级索引允许重复 key, 同一 key 的条目在叶内是连续区段。普通二分 (sbtLeafFind)
// 命中相等 key 时会立即返回, 落点可能位于区段中部; 若调用方只从落点向右扫描,
// 就会漏掉落点左侧的重复条目。定位"相同 key 区段"的起点必须用本函数。
func sbtLeafLowerBound(n *btNode, key []byte) int {
	el := int(n.info.EntryLen)
	cnt := int(n.info.Count)
	left, right := 0, cnt
	for left < right {
		mid := (left + right) / 2
		if sbtCmp(sbtEntry(n, mid)[:el-8], key) < 0 {
			left = mid + 1
		} else {
			right = mid
		}
	}
	return left
}

// sbtLeafLowerBoundPK 返回第一个复合键 (key, pkID) >= 目标的条目下标。
//
// 叶内条目按复合键有序 (重复 key 的条目按 pkID 排列), 因此插入定位与
// (key, pkID) 精确查找都应使用本函数, 而不是"key 落点 + 同 key 区段线性扫描"。
func sbtLeafLowerBoundPK(n *btNode, key []byte, pkID uint64) int {
	cnt := int(n.info.Count)
	left, right := 0, cnt
	for left < right {
		mid := (left + right) / 2
		e := sbtEntry(n, mid)
		if sbtCmpKeyPK(sbtLeafKey(e), sbtPKID(e), key, pkID) < 0 {
			left = mid + 1
		} else {
			right = mid
		}
	}
	return left
}

// sbtLeafHasKey 判断叶节点中是否存在等于 key 的条目 (基于 lower_bound 判定)
func sbtLeafHasKey(n *btNode, key []byte) bool {
	el := int(n.info.EntryLen)
	pos := sbtLeafLowerBound(n, key)
	return pos < int(n.info.Count) && sbtCmp(sbtEntry(n, pos)[:el-8], key) == 0
}

// sbtLeftmostLeafForKey 返回"可能包含 key"的最左叶子页 ID。
//
// 内部节点的分隔键是子树的"最小键", sbtRoute 取"最后一个 分隔键<=key 的子树"。
// 当同一 key 的条目因叶分裂而跨越相邻两叶 (左叶尾键 == 右叶首键) 时,
// sbtRoute 只会落到右侧那一叶, 从而漏掉位于左叶尾部的重复条目。
// 因此在路由落点的基础上, 沿叶子前驱链向左回溯: 只要前驱叶的最大键 >= key,
// 该前驱叶就仍可能含有 key, 必须一并纳入扫描范围。
func sbtLeftmostLeafForKey(rootPID uint32, key []byte, handle page.HandleDb) uint32 {
	pid := rootPID
	for pid != 0 {
		n := sbtLoad(handle, pid)
		if n == nil {
			return 0
		}
		if n.isLeaf() {
			break
		}
		pid = sbtRoute(n, key)
	}
	for pid != 0 {
		n := sbtLoad(handle, pid)
		if n == nil || n.info.Count == 0 {
			return 0
		}
		pre := n.info.Hdr.PrePage
		if pre == 0 {
			return pid
		}
		pn := sbtLoad(handle, pre)
		if pn == nil || pn.info.Count == 0 || pn.info.IsLeaf == 0 {
			return pid
		}
		el := int(n.info.EntryLen)
		lastKey := sbtEntry(pn, int(pn.info.Count)-1)[:el-8]
		if sbtCmp(lastKey, key) < 0 {
			return pid
		}
		pid = pre
	}
	return 0
}

// ======================== 按 key+pkID 精确删除 ========================

// SBTDeleteWithPKID 删除指定 key + pkID 的条目（处理重复 key 场景）。
//
// 采用与 sbtDeleteRec 完全一致的递归下溢处理: 叶被删空时由父层 sbtHandleUnderflow
// 借位/合并, 从而不会残留 "空叶可达" (否则会破坏 叶子链 == DFS 可达叶 的不变式,
// 并让 sbtRoute 落到空叶导致查找失败)。
func SBTDeleteWithPKID(rootPID uint32, key []byte, pkID uint64, handle page.HandleDb) (uint32, error) {
	if rootPID == 0 {
		return 0, page.ErrNotFound
	}
	found, err := sbtDeleteRecPK(rootPID, key, pkID, nil, handle)
	if err != nil {
		return rootPID, err
	}
	if !found {
		return rootPID, page.ErrNotFound
	}
	// 收尾归一化: 与 SBTDelete 一致
	newRoot := rootPID
	// 归一化过程中被跳过的旧根 (降高时旧根整体退役; 删空时旧根被换成新建空叶)。
	// 收集后在本函数末尾才归还, 避免 sbtNew 在同一轮里把这些页又分配回来。
	var retiredRoots []uint32
	for {
		n := sbtLoad(handle, newRoot)
		if n == nil || n.isLeaf() {
			break
		}
		if n.info.Count == 1 {
			childPID := sbtChildPID(sbtEntry(n, 0))
			if childPID == 0 {
				break
			}
			retiredRoots = append(retiredRoots, newRoot)
			newRoot = childPID
			continue
		}
		if n.info.Count == 0 {
			// n 必为内部节点 (循环开头已 break 叶子): 新建空叶要用叶条目长度
			// (内部 EntryLen keyLen+12 - 4 = keyLen+8)
			leaf := sbtNew(handle, true, int(n.info.EntryLen)-4)
			sbtSave(handle, leaf)
			retiredRoots = append(retiredRoots, newRoot)
			newRoot = leaf.pageID()
		}
		break
	}
	// 只归还退役的旧根自身: Count==1 时其子树仍是新根 (可达), 绝不能按子树回收;
	// Count==0 时该节点已是空内部节点, 其子页在上面的递归里已随空子树摘链归还。
	for _, pid := range retiredRoots {
		btReleasePage(handle, pid)
	}
	return newRoot, nil
}

// sbtDeleteRecPK 按 (key, pkID) 精确删除, 并复用 sbtDeleteRec 的下溢/合并/归一化流程。
//
// 内部节点的分隔键是子树的最小复合键 (key, pkID), 子树在复合键空间上是连续区间,
// 因此 sbtChildIdxPK 一次路由即唯一确定目标条目所在的子树 —— 不再需要旧实现
// "从 key 路由落点向左逐棵子树试探" 的兜底 (重复 key 跨几十个叶时那是平均半个
// 重复区段的页加载)。
func sbtDeleteRecPK(nodePID uint32, key []byte, pkID uint64, path []btPathEntry, handle page.HandleDb) (bool, error) {
	node := sbtLoad(handle, nodePID)
	if node == nil {
		return false, nil
	}
	el := int(node.info.EntryLen)
	if node.isLeaf() {
		return sbtDeleteLeafPK(node, key, pkID, path, handle), nil
	}
	if node.info.Count == 0 {
		return false, nil
	}
	ci := sbtChildIdxPK(node, key, pkID)
	childPID := sbtChildPID(sbtEntry(node, ci))
	newPath := make([]btPathEntry, len(path)+1)
	copy(newPath, path)
	newPath[len(path)] = btPathEntry{nodePID: nodePID, childIdx: ci}
	found, err := sbtDeleteRecPK(childPID, key, pkID, newPath, handle)
	if err != nil {
		return false, err
	}
	if !found {
		return false, nil
	}
	// 命中并已在该子树内删除: 处理本层的下溢/最小键/归一化 (与 sbtDeleteRec 相同)
	node = sbtLoad(handle, nodePID)
	if node == nil {
		return true, errors.New("sbt reload node failed")
	}
	if node.info.Count > 0 && ci < int(node.info.Count) && sbtChildPID(sbtEntry(node, ci)) == childPID {
		child := sbtLoad(handle, childPID)
		if child != nil {
			if child.info.Count == 0 {
				sbtHandleUnderflow(node, ci, handle)
				node = sbtLoad(handle, nodePID)
				if node == nil {
					return true, errors.New("sbt reload node failed after underflow")
				}
			} else {
				// 维护不变式: 父条目复合键 == 子树最小复合键 (删除叶子最小键后父键会变旧)
				copy(sbtEntry(node, ci)[:el-4], sbtNodeMinKey(child))
			}
		}
	}
	if !node.isLeaf() && node.info.Count == 1 {
		c0 := sbtChildPID(sbtEntry(node, 0))
		cn := sbtLoad(handle, c0)
		if cn != nil && cn.info.Count == 0 {
			sbtUnlinkSubtreeLeaves(handle, c0)
			node.info.Count = 0
			node.info.Hdr.UseSize = uint32(page.IndexHeadSize)
			sbtSave(handle, node)
			// c0 子树整段摘链后不再被任何父条目或叶子链引用, 连同其内部节点
			// 一并归还 —— 否则每删空一棵子树就漏掉一批索引页。
			btReleaseSubtree(handle, c0)
			return true, nil
		}
	}
	sbtSave(handle, node)
	return true, nil
}

// sbtDeleteLeafPK 在叶内删除指定 (key, pkID); 命中返回 true。
// 下溢处理与 sbtDeleteRec 的叶分支保持一致。
func sbtDeleteLeafPK(node *btNode, key []byte, pkID uint64, path []btPathEntry, handle page.HandleDb) bool {
	el := int(node.info.EntryLen)
	cnt := int(node.info.Count)
	// 叶内条目按复合键有序: 复合下界一次定位, 无需同 key 区段线性扫描
	pos := sbtLeafLowerBoundPK(node, key, pkID)
	if pos >= cnt {
		return false
	}
	entry := sbtEntry(node, pos)
	if sbtCmp(sbtLeafKey(entry), key) != 0 || sbtPKID(entry) != pkID {
		return false
	}
	sbtRmEntry(node, pos)
	if node.info.Count == 0 && len(path) == 0 {
		sbtSave(handle, node)
		return true
	}
	sbtSave(handle, node)
	if len(path) > 0 && node.info.Count > 0 {
		sbtTryMerge(handle, path, node, el)
	}
	return true
}

// ======================== 查找所有相同 key 的 pkID ========================

// SBTSearchAll 查找相同 key 下的所有 pkID
func SBTSearchAll(rootPID uint32, key []byte, handle page.HandleDb) ([]uint64, bool) {
	if rootPID == 0 {
		return nil, false
	}
	var result []uint64
	for pid := sbtLeftmostLeafForKey(rootPID, key, handle); pid != 0; {
		n := sbtLoad(handle, pid)
		if n == nil {
			return nil, false
		}
		el := int(n.info.EntryLen)
		cnt := int(n.info.Count)
		// 用 leftmost lower_bound 定位相同 key 区段起点, 避免漏掉左侧重复条目
		pos := sbtLeafLowerBound(n, key)
		if pos < cnt {
			if sbtCmp(sbtEntry(n, pos)[:el-8], key) != 0 {
				break // 已越过相同 key 区段
			}
			// 收集本叶中所有相同 key 的 pkID
			for i := pos; i < cnt; i++ {
				entry := sbtEntry(n, i)
				if sbtCmp(entry[:el-8], key) != 0 {
					break
				}
				result = append(result, sbtPKID(entry))
			}
		}
		pid = n.info.Hdr.NextPage
	}
	if len(result) == 0 {
		return nil, false
	}
	return result, true
}

// ======================== 搜索 ========================

func SBTSearch(rootPID uint32, key []byte, handle page.HandleDb) (uint64, bool) {
	if rootPID == 0 {
		return 0, false
	}
	// 从左起第一个可能含 key 的叶子开始, 沿 NextPage 右扫, 返回最左(最小 pkID)的条目
	for pid := sbtLeftmostLeafForKey(rootPID, key, handle); pid != 0; {
		n := sbtLoad(handle, pid)
		if n == nil {
			return 0, false
		}
		el := int(n.info.EntryLen)
		cnt := int(n.info.Count)
		pos := sbtLeafLowerBound(n, key)
		if pos < cnt {
			if sbtCmp(sbtEntry(n, pos)[:el-8], key) == 0 {
				return sbtPKID(sbtEntry(n, pos)), true
			}
			return 0, false // 已越过相同 key 区段
		}
		pid = n.info.Hdr.NextPage
	}
	return 0, false
}

// ======================== 范围扫描 ========================

func SBTRangeScan(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 {
		cur := rootPID
		if startKey != nil {
			// 必须用 sbtLeftmostLeafForKey 定位"可能含 startKey 的最左叶"。
			// sbtRoute 找的是"最后一个分隔键 <= startKey 的子树", 落点是该 key 区段的
			// 最右叶; 当同一 key 因叶分裂跨越相邻两叶 (左叶尾键 == 右叶首键) 时,
			// 左侧叶子里的重复条目会被整段跳过 —— 等值查询 (startKey == endKey)
			// 表现为只返回最右叶那部分: 实测每键 2500 条只回 256 条, 每键 1000 条只回 184 条。
			// 全范围扫描 (startKey == nil) 不受影响, 故只有带边界的查询出错。
			cur = sbtLeftmostLeafForKey(rootPID, startKey, handle)
		} else {
			for {
				n := sbtLoad(handle, cur)
				if n == nil {
					return nil
				}
				if n.isLeaf() {
					break
				}
				// startKey 为空时, 显式路由到最左叶 (跟随第一个子节点)
				cur = sbtChildPID(sbtEntry(n, 0))
			}
		}
		for cur != 0 {
			n := sbtLoad(handle, cur)
			if n == nil {
				break
			}
			if n.info.Hdr.Type != page.IndexDataIndex {
				return fmt.Errorf("btree: 正向范围扫描遇到非索引页 pid=%d type=%d, 叶子链已损坏",
					n.info.Hdr.PageID, n.info.Hdr.Type)
			}
			el := int(n.info.EntryLen)
			for i := 0; i < int(n.info.Count); i++ {
				entry := sbtEntry(n, i)
				key := entry[:el-8]
				if endKey != nil {
					cmp := sbtCmp(key, endKey)
					if cmp > 0 || (cmp == 0 && !endInc) {
						return nil
					}
				}
				if startKey != nil {
					cmp := sbtCmp(key, startKey)
					if cmp < 0 || (cmp == 0 && !startInc) {
						continue
					}
				}
				pkID := entry[el-8:]
				if !callback(pkID) {
					return nil
				}
			}
			cur = n.info.Hdr.NextPage
		}
	} else {
		// 反向: 路由到 endKey 或最右叶, 沿 PrePage 遍历
		cur := rootPID
		if endKey != nil {
			for {
				n := sbtLoad(handle, cur)
				if n == nil {
					return nil
				}
				if n.isLeaf() {
					break
				}
				cur = sbtRoute(n, endKey)
			}
		} else {
			for {
				n := sbtLoad(handle, cur)
				if n == nil {
					return nil
				}
				if n.isLeaf() {
					break
				}
				cur = sbtChildPID(sbtEntry(n, int(n.info.Count)-1))
			}
		}
		for cur != 0 {
			n := sbtLoad(handle, cur)
			if n == nil {
				break
			}
			if n.info.Hdr.Type != page.IndexDataIndex {
				return fmt.Errorf("btree: 反向范围扫描遇到非索引页 pid=%d type=%d, 叶子链已损坏",
					n.info.Hdr.PageID, n.info.Hdr.Type)
			}
			el := int(n.info.EntryLen)
			for i := int(n.info.Count) - 1; i >= 0; i-- {
				entry := sbtEntry(n, i)
				key := entry[:el-8]
				if startKey != nil {
					cmp := sbtCmp(key, startKey)
					if cmp < 0 || (cmp == 0 && !startInc) {
						return nil
					}
				}
				if endKey != nil {
					cmp := sbtCmp(key, endKey)
					if cmp > 0 || (cmp == 0 && !endInc) {
						continue
					}
				}
				pkID := entry[el-8:]
				if !callback(pkID) {
					return nil
				}
			}
			cur = n.info.Hdr.PrePage
		}
	}
	return nil
}

// ======================== 插入 ========================

func SBTInsert(rootPID uint32, entry []byte, handle page.HandleDb) (uint32, error) {
	el := len(entry)
	if rootPID == 0 {
		leaf := sbtNew(handle, true, el)
		sbtInsEntry(leaf, entry, 0)
		sbtSave(handle, leaf)
		newRoot := leaf.pageID()
		leaf.release()
		return newRoot, nil
	}
	var path []btPathEntry
	cur := rootPID
	for {
		n := sbtLoad(handle, cur)
		if n == nil {
			return rootPID, errors.New("sbt load failed")
		}
		if n.isLeaf() {
			key := entry[:el-8]
			pkID := sbtPKID(entry)
			// 叶内条目按复合键 (key, pkID) 有序: 复合下界一次定位插入位置
			ip := sbtLeafLowerBoundPK(n, key, pkID)
			if int(n.info.Hdr.UseSize)+el <= btSplitThreshold() {
				sbtInsEntry(n, entry, ip)
				sbtSave(handle, n)
				if ip == 0 {
					// 新条目成为本叶最小复合键: 沿路径向上同步
					// "父条目复合键 == 子树最小复合键"
					for i := len(path) - 1; i >= 0; i-- {
						p := path[i]
						pb := sbtLoad(handle, p.nodePID)
						if pb == nil || p.childIdx < 0 || p.childIdx >= int(pb.info.Count) {
							break
						}
						pbe := sbtEntry(pb, p.childIdx)
						copy(pbe[:len(pbe)-4], entry) // entry 本身就是 [key|pkID]
						sbtSave(handle, pb)
						pb.release()
						if p.childIdx != 0 {
							break
						}
					}
				}
				n.release()
				return rootPID, 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:], entry)
			if cnt > ip {
				copy(all[(ip+1)*el:], n.entries[ip*el:cnt*el])
			}
			// 分裂点: 左叶保留 [0,mid), 右叶保留 [mid,total)
			mid := splitPoint(total, SBTSplitCfg, false)
			rl := sbtNew(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)
			sbtSave(handle, n)
			sbtSave(handle, rl)
			if rl.info.Hdr.NextPage != 0 {
				nl := sbtLoad(handle, rl.info.Hdr.NextPage)
				if nl != nil {
					nl.info.Hdr.PrePage = rl.info.Hdr.PageID
					sbtSave(handle, nl)
					nl.release()
				}
			}
			// 提升的分隔复合键 = 右叶首条目 [key|pkID] (rl 已在内存, 无需再下降)
			pkc := append([]byte(nil), sbtEntry(rl, 0)...)
			nr, err := sbtPromote(handle, rootPID, path, n.info.Hdr.PageID, pkc, rl.info.Hdr.PageID, el)
			n.release()
			rl.release()
			return nr, err
		}
		// 插入同样按复合键路由: 子树在复合键空间上是连续区间, key 路由可能
		// 落到重复 key 区段右半的子树, 会破坏全局 (key, pkID) 有序性
		ci := sbtChildIdxPK(n, entry[:el-8], sbtPKID(entry))
		cpid := sbtChildPID(sbtEntry(n, ci))
		path = append(path, btPathEntry{nodePID: cur, childIdx: ci})
		n.release()
		cur = cpid
	}
}

// sbtPromote 把分裂出的右节点 (及其最小复合键 promoComposite) 插入父层, 必要时
// 递归分裂。leafEl 是叶条目长度 keyLen+8; 内部条目长度恒为 leafEl+4
// ([key | pkID(8) | childPID(4)])。promoComposite 长度 = leafEl ([key|pkID])。
func sbtPromote(handle page.HandleDb, rootPID uint32, path []btPathEntry, leftPID uint32, promoComposite []byte, rightPID uint32, leafEl int) (uint32, error) {
	iel := leafEl + 4
	promoKey := promoComposite[:len(promoComposite)-8]
	promoPK := binary.BigEndian.Uint64(promoComposite[len(promoComposite)-8:])
	if len(path) == 0 {
		nr := sbtNew(handle, false, iel)
		ln := sbtLoad(handle, leftPID)
		if ln == nil {
			nr.release()
			return rootPID, errors.New("sbt promote: left load failed")
		}
		le := make([]byte, iel)
		copy(le[:iel-4], sbtNodeMinKey(ln)) // leftPID 可能是叶或内部节点, 取其子树最小复合键
		ln.release()
		sbtSetChildPID(le, leftPID)
		re := make([]byte, iel)
		copy(re[:iel-4], promoComposite)
		sbtSetChildPID(re, rightPID)
		sbtInsEntry(nr, le, 0)
		sbtInsEntry(nr, re, 1)
		sbtSave(handle, nr)
		newRoot := nr.pageID()
		nr.release()
		return newRoot, nil
	}
	pi := len(path) - 1
	parent := sbtLoad(handle, path[pi].nodePID)
	if parent == nil {
		return rootPID, errors.New("sbt parent load failed")
	}
	ci := path[pi].childIdx
	cnt := int(parent.info.Count)
	pos := cnt
	for i := ci + 1; i < cnt; i++ {
		pe := sbtEntry(parent, i)
		if sbtCmpKeyPK(sbtIntKey(pe), sbtIntPKID(pe), promoKey, promoPK) >= 0 {
			pos = i
			break
		}
	}
	for i := cnt; i > pos; i-- {
		copy(sbtEntry(parent, i), sbtEntry(parent, i-1))
	}
	ne := make([]byte, iel)
	copy(ne[:iel-4], promoComposite)
	sbtSetChildPID(ne, rightPID)
	copy(sbtEntry(parent, pos), ne)
	parent.info.Count++
	parent.info.Hdr.UseSize += uint32(iel)
	if int(parent.info.Hdr.UseSize)+iel <= btSplitThreshold() {
		sbtSave(handle, parent)
		parent.release()
		return rootPID, nil
	}
	cnt = int(parent.info.Count)
	mid := splitPoint(cnt, SBTSplitCfg, false)
	// 分裂点取 mid: 左半保留 [0,mid), 右半保留 [mid,cnt)。
	// 不能丢掉 entry[mid] (否则它指向的整棵子树会不可达)。
	rn := sbtNew(handle, false, iel)
	rc := cnt - mid
	if rc > 0 {
		copy(rn.entries, parent.entries[mid*iel:cnt*iel])
	}
	rn.info.Count = uint32(rc)
	rn.info.Hdr.UseSize = uint32(page.IndexHeadSize) + uint32(rc*iel)
	parent.info.Count = uint32(mid)
	parent.info.Hdr.UseSize = uint32(page.IndexHeadSize) + uint32(mid*iel)
	sbtSave(handle, parent)
	sbtSave(handle, rn)
	// 提升的分隔复合键 = 右节点首条目的复合键 (其子树的最小复合键)
	e0 := sbtEntry(rn, 0)
	pk2 := make([]byte, iel-4)
	copy(pk2, e0[:iel-4])
	ret, err := sbtPromote(handle, rootPID, path[:pi], parent.info.Hdr.PageID, pk2, rn.info.Hdr.PageID, leafEl)
	parent.release()
	rn.release()
	return ret, err
}

// ======================== 删除 ========================

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

// sbtDeleteRec 按 key 删除 (删除该 key 的一条条目), 并做下溢/合并/归一化。
func sbtDeleteRec(nodePID uint32, key []byte, path []btPathEntry, handle page.HandleDb) (uint32, error) {
	node := sbtLoad(handle, nodePID)
	if node == nil {
		return nodePID, page.ErrNotFound
	}
	el := int(node.info.EntryLen)
	if node.isLeaf() {
		pos, ok := sbtLeafFind(node, key)
		if !ok {
			return nodePID, page.ErrNotFound
		}
		sbtRmEntry(node, pos)
		if node.info.Count == 0 && len(path) == 0 {
			sbtSave(handle, node)
			return nodePID, nil
		}
		sbtSave(handle, node)
		if len(path) > 0 && node.info.Count > 0 {
			sbtTryMerge(handle, path, node, el)
		}
		return nodePID, nil
	}
	ci := sbtChildIdx(node, key)
	childPID := sbtChildPID(sbtEntry(node, ci))
	newPath := make([]btPathEntry, len(path)+1)
	copy(newPath, path)
	newPath[len(path)] = btPathEntry{nodePID: nodePID, childIdx: ci}
	if _, err := sbtDeleteRec(childPID, key, newPath, handle); err != nil {
		return nodePID, err
	}
	// 递归过程中子节点可能触发合并, 而合并会单独加载并保存本节点。
	// 这里必须重新加载, 否则后面用过期副本保存会把合并结果覆盖掉 (数据重复/条目丢失)。
	node = sbtLoad(handle, nodePID)
	if node == nil {
		return nodePID, errors.New("sbt reload node failed")
	}
	if node.info.Count > 0 {
		// 仅当原 child 条目仍在本节点的 ci 位置时才处理
		// (若该条目已被合并移除, 则说明子树已被兄弟吸收, 无需再处理)
		if ci >= 0 && ci < int(node.info.Count) && sbtChildPID(sbtEntry(node, ci)) == childPID {
			child := sbtLoad(handle, childPID)
			if child != nil {
				if child.info.Count == 0 {
					sbtHandleUnderflow(node, ci, handle)
					// 借位/合并会直接改写本节点并保存, 必须重新加载
					node = sbtLoad(handle, nodePID)
					if node == nil {
						return nodePID, errors.New("sbt reload node failed after underflow")
					}
				} else {
					// 维护不变式: 父条目复合键 == 子树最小复合键
					// (删除叶子的最小键后, 父条目键会变旧)
					copy(sbtEntry(node, ci)[:el-4], sbtNodeMinKey(child))
				}
			}
		}
	}
	// 归一化: 内部节点若只剩一个 "已空" 的子节点 (没有兄弟可合并), 则该节点本身
	// 也变空。此时必须把该空子树整段从叶子链摘除并让本节点变为空 (Count=0),
	// 由上层继续向上收缩。否则会残留 "空内部节点", 破坏路由/查找。
	if !node.isLeaf() && node.info.Count == 1 {
		c0 := sbtChildPID(sbtEntry(node, 0))
		cn := sbtLoad(handle, c0)
		if cn != nil && cn.info.Count == 0 {
			sbtUnlinkSubtreeLeaves(handle, c0)
			node.info.Count = 0
			node.info.Hdr.UseSize = uint32(page.IndexHeadSize)
			sbtSave(handle, node)
			// c0 子树整段摘链后不再被任何父条目或叶子链引用, 连同其内部节点
			// 一并归还 —— 否则每删空一棵子树就漏掉一批索引页。
			btReleaseSubtree(handle, c0)
			return nodePID, nil
		}
	}
	if node.info.Count == 0 {
		sbtSave(handle, node)
		return nodePID, nil
	}
	sbtSave(handle, node)
	return nodePID, nil
}

func sbtTryMerge(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 := sbtLoad(handle, pp)
	if parent == nil {
		return
	}
	ci := -1
	for i := 0; i < int(parent.info.Count); i++ {
		if sbtChildPID(sbtEntry(parent, i)) == node.info.Hdr.PageID {
			ci = i
			break
		}
	}
	if ci < 0 {
		return
	}
	if ci > 0 {
		leftPID := sbtChildPID(sbtEntry(parent, ci-1))
		left := sbtLoad(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 {
				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
				sbtSave(handle, left)
				if left.info.Hdr.NextPage != 0 {
					nl := sbtLoad(handle, left.info.Hdr.NextPage)
					if nl != nil {
						nl.info.Hdr.PrePage = leftPID
						sbtSave(handle, nl)
					}
				}
				sbtRmEntry(parent, ci)
				sbtSave(handle, parent)
				// node 的条目已全部并入 left, 且指向 node 的父条目已被摘除,
				// 叶子链也已改接到 left 的 NextPage —— node 页至此彻底不可达。
				btReleasePage(handle, nodePID)
				return
			}
		}
	}
	if ci < int(parent.info.Count)-1 {
		rightPID := sbtChildPID(sbtEntry(parent, ci+1))
		right := sbtLoad(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 {
				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
				sbtSave(handle, node)
				if node.info.Hdr.NextPage != 0 {
					nl := sbtLoad(handle, node.info.Hdr.NextPage)
					if nl != nil {
						nl.info.Hdr.PrePage = node.info.Hdr.PageID
						sbtSave(handle, nl)
					}
				}
				sbtRmEntry(parent, ci+1)
				sbtSave(handle, parent)
				// right 的条目已全部并入 node, 且指向 right 的父条目已摘除,
				// 叶子链也已改接到 node 的 NextPage —— right 页至此彻底不可达。
				btReleasePage(handle, rightPID)
				return
			}
		}
	}
}

// sbtHandleUnderflow 处理空/下溢子节点: 先尝试向兄弟借位, 否则与兄弟合并。
// 不接收 el 参数 —— child/兄弟与父节点的条目长度可能不同 (叶 keyLen+8,
// 内部 keyLen+12), 一律从各节点自身的 EntryLen 取。
func sbtHandleUnderflow(parent *btNode, childIdx int, handle page.HandleDb) {
	childPID := sbtChildPID(sbtEntry(parent, childIdx))
	child := sbtLoad(handle, childPID)
	if child == nil {
		return
	}
	if childIdx > 0 {
		leftPID := sbtChildPID(sbtEntry(parent, childIdx-1))
		leftSib := sbtLoad(handle, leftPID)
		if leftSib != nil && int(leftSib.info.Count) > 1 {
			borrowed := append([]byte(nil), sbtEntry(leftSib, int(leftSib.info.Count)-1)...)
			sbtRmEntry(leftSib, int(leftSib.info.Count)-1)
			// 左兄弟的最后一条被移到 child 首部
			sbtInsEntry(child, borrowed, 0)
			// child 的最小复合键变成 borrowed; 左兄弟最小复合键不变
			pe := sbtEntry(parent, childIdx)
			copy(pe[:len(pe)-4], sbtNodeMinKey(child))
			pe = sbtEntry(parent, childIdx-1)
			copy(pe[:len(pe)-4], sbtNodeMinKey(leftSib))
			sbtSave(handle, leftSib)
			sbtSave(handle, child)
			sbtSave(handle, parent)
			return
		}
	}
	if childIdx < int(parent.info.Count)-1 {
		rightPID := sbtChildPID(sbtEntry(parent, childIdx+1))
		rightSib := sbtLoad(handle, rightPID)
		if rightSib != nil && int(rightSib.info.Count) > 1 {
			borrowed := append([]byte(nil), sbtEntry(rightSib, 0)...)
			sbtRmEntry(rightSib, 0)
			// 右兄弟的首条被追加到 child 尾部: child 的最小复合键不变
			sbtInsEntry(child, borrowed, int(child.info.Count))
			pe := sbtEntry(parent, childIdx)
			copy(pe[:len(pe)-4], sbtNodeMinKey(child))
			// 右兄弟失去首条后最小复合键变化, 更新其父条目
			pe = sbtEntry(parent, childIdx+1)
			copy(pe[:len(pe)-4], sbtNodeMinKey(rightSib))
			sbtSave(handle, rightSib)
			sbtSave(handle, child)
			sbtSave(handle, parent)
			return
		}
	}
	if childIdx > 0 {
		sbtMergeChildren(handle, parent, childIdx-1)
	} else if childIdx < int(parent.info.Count)-1 {
		sbtMergeChildren(handle, parent, childIdx)
	}
}

// sbtMergeChildren 把 sepIdx+1 指向的右子树并入 sepIdx 指向的左子树。
// 条目长度从 left 节点自身的 EntryLen 取 (左右同为叶或同为内部节点)。
func sbtMergeChildren(handle page.HandleDb, parent *btNode, sepIdx int) {
	leftPID := sbtChildPID(sbtEntry(parent, sepIdx))
	rightPID := sbtChildPID(sbtEntry(parent, sepIdx+1))
	left := sbtLoad(handle, leftPID)
	right := sbtLoad(handle, rightPID)
	if left == nil || right == nil {
		return
	}
	// 注意: 本实现的父条目键是 "子树最小键的副本", 不是经典 B+ 树里独立的分隔键。
	// 因此合并内部节点时, 直接把右兄弟的条目追加到左兄弟即可,
	// 绝不能把 parent[sepIdx] 下放 (那会写出一条指向左兄弟自身的自引用条目)。
	el := int(left.info.EntryLen)
	rc := int(right.info.Count)
	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
	}
	sbtSave(handle, left)
	if left.isLeaf() && left.info.Hdr.NextPage != 0 {
		nl := sbtLoad(handle, left.info.Hdr.NextPage)
		if nl != nil {
			nl.info.Hdr.PrePage = leftPID
			sbtSave(handle, nl)
		}
	}
	sbtRmEntry(parent, sepIdx+1)
	sbtSetChildPID(sbtEntry(parent, sepIdx), leftPID)
	// 左节点若原本是空节点, 合并后其最小复合键变为右兄弟的最小复合键, 必须同步
	// 父条目, 否则会出现 "父键 != 子树最小复合键", 破坏路由。
	if left.info.Count > 0 {
		pe := sbtEntry(parent, sepIdx)
		copy(pe[:len(pe)-4], sbtNodeMinKey(left))
	}
	sbtSave(handle, parent)
	// right 的条目已全部并入 left, 父条目 sepIdx+1 已摘除 (叶子链在上面的
	// NextPage/PrePage 重接中已绕过 right) —— right 子树至此彻底不可达。
	btReleasePage(handle, rightPID)
}

// SBTBuildFromSorted 从已排序的条目批量构建二级 B+ 树 (自底向上)
// entries: 每个条目 = [key(keyLen) + pkID(8B)], 必须按复合键 (key, pkID) 升序排列
// (只按 key 排序不够: 复合分隔键的路由/二分要求叶内同 key 条目也按 pkID 有序)
// 返回新树根 pageID
//
// 内存优化: 避免逐行 B+ 树搜索 (每次 sbtLoad 16KB), 改为顺序创建所有节点
func SBTBuildFromSorted(entries [][]byte, handle page.HandleDb) uint32 {
	if len(entries) == 0 {
		return 0
	}

	el := len(entries[0]) // entry len: keyLen + 8
	keyLen := el - 8
	maxPerLeaf := maxLeafEntries(el)

	type childInfo struct {
		pid      uint32
		firstKey []byte // 该子树最小 key (keyLen 字节)
		firstPK  uint64 // 该子树最小 pkID
	}

	var children []childInfo
	var prevLeafPID uint32

	// 构建叶节点 (带链表链接)
	for i := 0; i < len(entries); {
		end := i + maxPerLeaf
		if end > len(entries) {
			end = len(entries)
		}
		leaf := sbtNew(handle, true, el)
		for j := i; j < end; j++ {
			dst := sbtEntry(leaf, int(leaf.info.Count))
			copy(dst, entries[j])
			leaf.info.Count++
			leaf.info.Hdr.UseSize += uint32(el)
		}
		leaf.info.Hdr.PrePage = prevLeafPID
		sbtSave(handle, leaf)
		if prevLeafPID != 0 {
			prevLeaf := sbtLoad(handle, prevLeafPID)
			if prevLeaf != nil {
				prevLeaf.info.Hdr.NextPage = leaf.pageID()
				sbtSave(handle, prevLeaf)
			}
		}
		prevLeafPID = leaf.pageID()
		fk := make([]byte, keyLen)
		copy(fk, entries[i][:keyLen])
		children = append(children, childInfo{pid: leaf.pageID(), firstKey: fk, firstPK: sbtPKID(entries[i])})
		i = end
	}

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

	// 自底向上构建内部节点
	// 路由约定 (sbtRoute/sbtChildIdx): 找最后一个 分隔复合键 <= 搜索复合键 的子树
	// 内部节点条目: [child.firstKey | child.firstPK | child.pid]
	// 即 entry[i] 的前 keyLen+8 字节 = child[i] 子树的最小复合键
	intEl := el + 4 // 内部条目 = [key | pkID(8) | childPID(4)]
	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 := sbtNew(handle, false, intEl)
			for j := i; j < end; j++ {
				e := make([]byte, intEl)
				// entry[j] 前 keyLen+8 字节 = 该 child 子树的最小复合键
				copy(e[:keyLen], children[j].firstKey)
				binary.BigEndian.PutUint64(e[keyLen:keyLen+8], children[j].firstPK)
				sbtSetChildPID(e, children[j].pid)
				dst := sbtEntry(node, int(node.info.Count))
				copy(dst, e)
				node.info.Count++
				node.info.Hdr.UseSize += uint32(intEl)
			}
			sbtSave(handle, node)
			parents = append(parents, childInfo{
				pid:      node.pageID(),
				firstKey: children[i].firstKey,
				firstPK:  children[i].firstPK,
			})
			i = end
		}
		children = parents
	}

	return children[0].pid
}

// ======================== 二级叶批量插入 ========================

// SBTBulkInsert 批量插入二级索引条目。entries 是 [key|pkID] 定长条目的有序串
// (按复合键 (key, pkID) 升序, 调用方排序), el 为条目长度 (keyLen+8)。
//
// 思路: 排序批量对叶做"顺序扫描式"填充 —— 驻留叶加载一次, 条目在页内定位插入;
// 目标叶单调右移 (整批有序), 沿叶链走而不是逐条从根下降。每张叶加载/落盘各一次,
// 单条 SBTInsert 兜底只发生在三种边角: 叶满要分裂、条目成为叶最小键 (父键同步)、
// 链尾之后还有更大键 —— 分裂/提升/同步全套复用既有逻辑。
//
// 正确性关键: 驻留叶的可用区间是 [叶首条目, 上界), 上界 = 右邻叶首条目 (叶链
// NextPage)。判定"该不该留在当前叶"必须用上界, 不能用"大于叶末条"—— 后者会在
// 条目越过右邻叶首键时仍然往左叶追加, 破坏父层路由 (该条目从此按键查不到)。
// 批量键与既有数据交错 (少基数键的翻页灌数) 恰好整批落进叶中段, 走页内定位插入,
// 不再退化成逐条单插。
func SBTBulkInsert(rootPID uint32, entries []byte, el int, handle page.HandleDb) (uint32, error) {
	if el <= 12 || len(entries)%el != 0 {
		return rootPID, errors.New("sbt bulk insert: bad entry layout")
	}
	n := len(entries) / el
	if n == 0 {
		return rootPID, nil
	}
	th := btSplitThreshold()
	var leaf *btNode
	var upper []byte // 驻留叶复合键上界 (右邻叶首条目的副本); nil = 最右叶 (上界无穷)

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

	flushLeaf := func() {
		if leaf != nil {
			sbtSave(handle, leaf)
			if rootPID == 0 {
				rootPID = leaf.pageID()
			}
			leaf.release()
			leaf = nil
		}
	}

	// descend 按复合键从根下降到 ent 所属叶 (与 SBTInsert 的路由约定一致), 并取上界。
	descend := func(ent []byte) error {
		cur := rootPID
		for {
			nd := sbtLoad(handle, cur)
			if nd == nil {
				return errors.New("sbt bulk insert: load failed")
			}
			if nd.isLeaf() {
				leaf = nd
				upper = upperOf(nd)
				return nil
			}
			ci := sbtChildIdxPK(nd, ent[:el-8], sbtPKID(ent))
			cur = sbtChildPID(sbtEntry(nd, ci))
			nd.release()
		}
	}

	// stepTo 沿叶链从 from 走到第一张非空叶 (或链尾) 并驻留。
	// 调用方随后用上界复查, ent 仍越界就再走一步。
	stepTo := func(from uint32) error {
		cur := from
		for {
			nd := sbtLoad(handle, cur)
			if nd == nil {
				return errors.New("sbt 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 := 0; i < n; i++ {
		ent := entries[i*el : (i+1)*el]
		if leaf == nil {
			if rootPID == 0 {
				leaf = sbtNew(handle, true, el)
				upper = nil
			} else if err := descend(ent); err != nil {
				return rootPID, err
			}
		}
		// 越过驻留叶上界: 目标叶在右侧。落盘驻留叶后沿叶链右移 (比重新下降省
		// 2/3 页加载); 链走尽了重新下降, 让路由把后续大键送进(新的)最右叶。
		for upper != nil && sbtCmp(ent, 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(ent); err != nil {
				return rootPID, err
			}
		}

		if int(leaf.info.Hdr.UseSize)+el > th {
			// 叶满: 落盘驻留叶, 单条兜底 (分裂/提升/叶链维护全套)
			flushLeaf()
			nr, err := SBTInsert(rootPID, ent, handle)
			if err != nil {
				return rootPID, err
			}
			rootPID = nr
			continue
		}
		cnt := int(leaf.info.Count)
		ip := sbtLeafLowerBoundPK(leaf, ent[:el-8], sbtPKID(ent))
		if ip == 0 && cnt > 0 {
			// 新条目将成为叶最小复合键: "父条目复合键 == 子树最小复合键"不变式
			// 需要沿路径向上同步, 交给单条路径 (整批有序, 只可能发生在批首条目)
			flushLeaf()
			nr, err := SBTInsert(rootPID, ent, handle)
			if err != nil {
				return rootPID, err
			}
			rootPID = nr
			continue
		}
		sbtInsEntry(leaf, ent, ip)
	}
	flushLeaf()
	return rootPID, nil
}