package btree import ( "bytes" "encoding/binary" "errors" "fmt" "unsafe" "git.ycl.pub/yl32940/TsDB2/DB/core/page" ) // ======================== 二级 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 { return bytes.Compare(a, b) } // 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 }