package datapage import ( "TsDB2/DB/core/page" "encoding/binary" "errors" "sort" "sync" ) // PageSize 索引页大小 (避免导入 index 包造成循环依赖) const PageSize = 16384 var ( // errRowTooLarge 行装不下当前页, 调用方需换页/重建。 // 注: "未找到"语义统一用 page.ErrNotFound, 便于上层 errors.Is 判定, 不再各自造错误值。 errRowTooLarge = errors.New("row too large, need heap overflow") ) // dpBufPool 复用 DataPage 的 16KB 页缓冲。 // 背景: 插入路径每行至少 Load 一次主数据页, 旧实现每次 make(16KB) —— // 灌数时贡献 ~10% 的总分配量。协议与 btree 包的 nodeBufPool 相同: // 只在调用方能证明 DataPage 已死的位置 release, 其余路径照旧分配。 var dpBufPool = sync.Pool{New: func() any { b := make([]byte, PageSize); return &b }} // Release 把页缓冲归还池中。调用方必须保证此后不再访问该 DataPage // (buf 的任何视图都作废)。导出供 index 包的插入热路径使用。 func (dp *DataPage) Release() { if dp == nil || len(dp.buf) != PageSize || cap(dp.buf) < PageSize { return } b := dp.buf dpBufPool.Put(&b) } // ============================================================================= // 统一 DataPage — 所有需要独立数据页的索引类型共享 // // 磁盘布局 (权威定义见 page.DataPageHdr): // // [PageHdr(20B) + Count(4B) + Min(8B) + Max(8B)] = 40B header // [len(4B) + rowData] ... = 行数据区 (行区密度不变) // ... 空闲区 ... // [slot 目录: slotCount × 2B, 从页尾向前生长] // [slotCount(2B, 页尾最后 2 字节)] // // len 最高位 (0x80000000) 为软删除标志 (行内墓碑) // Min/Max 存储 rowData 前 8 字节的最小/最大值 (BigEndian 比较) // // 槽位目录 (Slot): // - slot i (1 基) 存放在偏移 PageSize-2-2*i 处, 值 = 该行 4B 长度前缀的页内偏移; // 0 表示空槽 (行已被整理页物理移除, 槽位可复用)。 // - 槽位号稳定: 整理页 (Compact) 只搬动行、改写偏移, 绝不改变槽位号 —— // 因此主键 B+ 树叶条目里的 (dpID, slot) 在整理后依然有效。 // - 槽位从页尾向前生长, 新增槽位不会移动既有槽位, 无需 memmove。 // // 为什么需要槽位: 删除/更新采用「行内墓碑 + 写新行」后, 页内会长期存在失效字节, // 必须能在不重编号的前提下整理页; 而主键索引又希望命中后 O(1) 取行而非页内扫描。 // // ============================================================================= // HdrSize 数据页头部大小 (直接复用 page 包的权威布局定义) const HdrSize = page.DataPageHdrSize // DeletedFlag 软删除标志 (长度字段最高位) const DeletedFlag uint32 = 0x80000000 // 槽位目录参数 const ( slotEntrySize = 2 // 每条槽位 2B: 行起始偏移 (uint16 LE); 0 = 空槽 slotCountSize = 2 // 页尾最后 2B: 槽位总数 (uint16 LE) // maxSlotCount 槽位目录理论上限 (行区至少要能放下 4B 长度前缀) maxSlotCount = (PageSize - HdrSize - slotCountSize) / (slotEntrySize + 4) ) // CompactDeadRatio 整理页阈值: 页内失效 (墓碑) 字节占已用行区字节的比例。 // 超过该比例时 Compact / MaybeCompact 会把活行紧凑重排并回收墓碑空间。 // 这是一个可配置参数, 默认 0.4 (40%); 设为 <=0 可关闭自动整理。 var CompactDeadRatio = 0.4 // SetCompactDeadRatio 设置整理页阈值 (供上层配置, 例如按表 / 按库调整)。 func SetCompactDeadRatio(r float64) { CompactDeadRatio = r } // DataPage 统一数据页 type DataPage struct { handle page.HandleDb pageID uint32 buf []byte count uint32 // 行区内物理行数 (含墓碑) == 占用中的槽位数 dataEnd uint32 // 行区结束偏移 (= 空闲区起点) slotCount uint32 // 槽位目录容量 (含空槽) deadBytes uint32 // 失效字节数 (4B 前缀 + 墓碑行数据) —— 页头 Dead 的内存镜像 liveCount uint32 // 未删除行数 —— 页头 LiveCnt 的内存镜像 } // ======================== 页 I/O ======================== // New 创建一个新的空数据页 func New(handle page.HandleDb) *DataPage { pid := handle.NewPage() buf := make([]byte, PageSize) hdr := page.HdrOf(buf) hdr.Type = page.DataIndex hdr.PageID = pid hdr.UseSize = HdrSize return &DataPage{handle: handle, pageID: pid, buf: buf, count: 0, dataEnd: HdrSize} } // Load 从磁盘加载指定页 func Load(handle page.HandleDb, pageID uint32) (*DataPage, error) { buf := *dpBufPool.Get().(*[]byte) if err := handle.ReadPage(pageID, buf); err != nil { dpBufPool.Put(&buf) return nil, err } hdr := page.HdrOf(buf) dh := page.DataHdrOf(buf) cnt := dh.Count dataEnd := hdr.UseSize // 槽位目录容量存在页尾; 必须校验 "目录不侵入行区", 否则一律按无槽位处理 // (被行级 WAL 直接改写的页可能留下垃圾尾字节)。 sc := uint32(binary.LittleEndian.Uint16(buf[PageSize-slotCountSize:])) if sc > maxSlotCount || uint32(PageSize)-slotCountSize-sc*slotEntrySize < dataEnd { sc = 0 binary.LittleEndian.PutUint16(buf[PageSize-slotCountSize:], 0) } // LiveCnt/Dead 由所有写路径增量维护并随页落盘 (见 page.DataPageHdr), // 这里直接读页头 —— 不再全页走链重算 (旧行为让每次 Load 都是 O(页内行数), // 对所有查询路径收税, 无条件 COUNT(*) 更是被它拖回逐行)。 dp := &DataPage{ handle: handle, pageID: pageID, buf: buf, count: cnt, dataEnd: dataEnd, slotCount: sc, deadBytes: dh.Dead, liveCount: dh.LiveCnt, } return dp, nil } // Save 持久化到磁盘 func (dp *DataPage) Save() { page.HdrOf(dp.buf).UseSize = dp.dataEnd dh := page.DataHdrOf(dp.buf) dh.Count = dp.count dh.LiveCnt = dp.liveCount dh.Dead = dp.deadBytes _ = dp.handle.SavePage(dp.pageID, dp.buf) } // PageID 返回当前页 ID func (dp *DataPage) PageID() uint32 { return dp.pageID } // Count 返回行区内物理行数 (含墓碑) func (dp *DataPage) Count() uint32 { return dp.count } // SlotCount 返回槽位目录容量 (含空槽) func (dp *DataPage) SlotCount() uint32 { return dp.slotCount } // DeadBytes 返回页内失效 (墓碑) 字节数 func (dp *DataPage) DeadBytes() uint32 { return dp.deadBytes } // DeadRatio 返回失效字节占已用行区的比例 (0~1) func (dp *DataPage) DeadRatio() float64 { used := dp.dataEnd - HdrSize if used == 0 { return 0 } return float64(dp.deadBytes) / float64(used) } // freeLimit 行区允许增长到的上限 (为"新增一个槽位"预留目录空间) func (dp *DataPage) freeLimit() uint32 { return uint32(PageSize) - slotCountSize - (dp.slotCount+1)*slotEntrySize } // canFit 判断行区再放入 need 字节 (含 4B 前缀) 是否仍不侵入槽位目录 func (dp *DataPage) canFit(need uint32) bool { return dp.dataEnd+need <= dp.freeLimit() } // IsFull 判断是否有足够空间追加新行 (rowSize 约定: 不含 4B 长度前缀) func (dp *DataPage) IsFull(rowSize uint32) bool { return dp.dataEnd+rowSize+4 > dp.freeLimit() } // ======================== 槽位目录 ======================== // slotPos 槽位 i (1 基) 在页内的字节偏移。 func (dp *DataPage) slotPos(slotID uint32) uint32 { return uint32(PageSize) - slotCountSize - slotID*slotEntrySize } // getSlot 读取槽位存储的行起始偏移 (0 = 空槽) func (dp *DataPage) getSlot(slotID uint32) uint32 { p := dp.slotPos(slotID) return uint32(binary.LittleEndian.Uint16(dp.buf[p:])) } // putSlot 写入槽位 func (dp *DataPage) putSlot(slotID, off uint32) { p := dp.slotPos(slotID) binary.LittleEndian.PutUint16(dp.buf[p:], uint16(off)) } // setSlotCount 写入槽位目录容量 (页尾) func (dp *DataPage) setSlotCount(n uint32) { binary.LittleEndian.PutUint16(dp.buf[PageSize-slotCountSize:], uint16(n)) } // allocSlot 为一个位于 off 的行分配槽位: 优先复用空槽, 否则扩容目录。 // 返回 0 表示目录已满 (正常情况下 canFit 已挡住)。 func (dp *DataPage) allocSlot(off uint32) uint32 { for i := uint32(1); i <= dp.slotCount; i++ { if dp.getSlot(i) == 0 { dp.putSlot(i, off) return i } } if dp.slotCount >= maxSlotCount { return 0 } dp.slotCount++ dp.setSlotCount(dp.slotCount) dp.putSlot(dp.slotCount, off) return dp.slotCount } // shiftSlots 把偏移 >= from 的槽位整体后移 delta (行区内插入行导致)。 func (dp *DataPage) shiftSlots(from, delta uint32) { for i := uint32(1); i <= dp.slotCount; i++ { o := dp.getSlot(i) if o != 0 && o >= from { dp.putSlot(i, o+delta) } } } // SlotOffset 返回槽位指向的行起始偏移 (0 = 空槽)。供诊断/测试使用。 func (dp *DataPage) SlotOffset(slotID uint32) uint32 { if slotID == 0 || slotID > dp.slotCount { return 0 } return dp.getSlot(slotID) } // ======================== 失效字节统计 / 整理页 ======================== // Compact 整理页: 按原有页内顺序紧凑重排活行, 回收墓碑占用的空间并释放其槽位。 // 槽位号保持不变 (活行改写偏移, 墓碑槽位置 0), 因此 B+ 树叶条目无需改动。 // 返回是否真正做了整理。 func (dp *DataPage) Compact() bool { type ent struct { slot uint32 off uint32 } ents := make([]ent, 0, dp.slotCount) for i := uint32(1); i <= dp.slotCount; i++ { if o := dp.getSlot(i); o != 0 { ents = append(ents, ent{i, o}) } } if len(ents) == 0 { dp.deadBytes = 0 dp.liveCount = 0 return false } // 必须按页内物理偏移排序重排, 否则会打乱 "行按 pkID/键升序" 的不变式。 sort.Slice(ents, func(a, b int) bool { return ents[a].off < ents[b].off }) liveRows := make([][]byte, 0, len(ents)) liveSlots := make([]uint32, 0, len(ents)) dead := 0 for _, e := range ents { rowLen, deleted, rowStart, ok := dp.nextRow(e.off) if !ok { dp.putSlot(e.slot, 0) // 坏槽位: 释放 continue } if deleted { dp.putSlot(e.slot, 0) dead++ continue } row := make([]byte, rowLen) copy(row, dp.buf[rowStart:rowStart+rowLen]) liveRows = append(liveRows, row) liveSlots = append(liveSlots, e.slot) } if dead == 0 { dp.deadBytes = 0 return false } w := HdrSize for i, r := range liveRows { dp.putSlot(liveSlots[i], w) binary.LittleEndian.PutUint32(dp.buf[w:], uint32(len(r))) w += 4 copy(dp.buf[w:], r) w += uint32(len(r)) } for i := w; i < dp.dataEnd; i++ { dp.buf[i] = 0 } dp.dataEnd = w dp.count = uint32(len(liveRows)) dp.deadBytes = 0 dp.liveCount = uint32(len(liveRows)) dp.recomputeMinMax(liveRows) return true } // MaybeCompact 失效字节占比超过 CompactDeadRatio 时整理页。 func (dp *DataPage) MaybeCompact() bool { if CompactDeadRatio <= 0 || dp.deadBytes == 0 { return false } if dp.DeadRatio() < CompactDeadRatio { return false } return dp.Compact() } // CompactForInsert 在"即将为本页装不下的行新开一页"之前调用: 按配置 (CompactDeadRatio) // 就地整理本页, 把墓碑占用的空间还给活行。返回是否真的整理了 (整理后本页已落盘)。 // // 为什么放在这里而不是删除路径上: 删除时整理是"顺手做一件可能没用的事"(代价大、收益未知), // 而此刻的整理目的非常明确 —— 若整理后装得下, 就省下了一次开新页 (以及随之而来的 // 页面碎片: 每页只放少量行、扫描要读更多页)。所以本方法要求"整理确实能把 need 塞进去", // 否则宁可让调用方开新页, 也不白搬一趟 (整理是 O(页内行数) 的排序 + 拷贝)。 // // 阈值仍由 CompactDeadRatio 决定 (<=0 即关闭自动整理, 见 MaybeCompact): 配置为激进值, // 整理就更容易发生; 配置为 0, 行为完全退回"装不下就开新页"。 func (dp *DataPage) CompactForInsert(need uint32) bool { // 没有墓碑可回收, 或把墓碑全还给活行也仍然装不下 → 不值得整理 if dp.deadBytes == 0 || dp.dataEnd-dp.deadBytes+need > dp.freeLimit() { return false } if !dp.MaybeCompact() { return false } // Compact 只改内存缓冲; 且它保持槽位号不变, 所以先落盘绝不会让 B+ 树叶条目指错行。 dp.Save() return true } // ======================== Min/Max 维护 ======================== // getMin 读取当前 Min 值 func (dp *DataPage) getMin() uint64 { return page.DataHdrOf(dp.buf).GetMin() } // getMax 读取当前 Max 值 func (dp *DataPage) getMax() uint64 { return page.DataHdrOf(dp.buf).GetMax() } // MinPK 返回页内最小 pkID (Min 字段的公开访问) func (dp *DataPage) MinPK() uint64 { return dp.getMin() } // MaxPK 返回页内最大 pkID (Max 字段的公开访问) func (dp *DataPage) MaxPK() uint64 { return dp.getMax() } // updateMinMax 追加行后更新 Min/Max (取 rowData 前 8 字节 BigEndian 比较) func (dp *DataPage) updateMinMax(rowData []byte) { if len(rowData) < 8 { return } val := binary.BigEndian.Uint64(rowData[:8]) dh := page.DataHdrOf(dp.buf) if dp.count <= 1 { // 第一行 (或刚从 0 变 1), 直接设置 dh.SetMin(val) dh.SetMax(val) return } if val < dh.GetMin() { dh.SetMin(val) } if val > dh.GetMax() { dh.SetMax(val) } } // recomputeMinMax 依据活行重算 Min/Max (取每行前 8 字节 BigEndian) func (dp *DataPage) recomputeMinMax(rows [][]byte) { var minV, maxV uint64 first := true for _, r := range rows { if len(r) < 8 { continue } v := binary.BigEndian.Uint64(r[:8]) if first { minV, maxV, first = v, v, false continue } if v < minV { minV = v } if v > maxV { maxV = v } } if first { minV, maxV = 0, 0 } dh := page.DataHdrOf(dp.buf) dh.SetMin(minV) dh.SetMax(maxV) } // ======================== 行操作 ======================== // AppendRow 追加一行 (不返回槽位)。 // storeRow 格式: [pkID(8B BE) + rowData] 或纯 rowData, 由调用方决定 func (dp *DataPage) AppendRow(storeRow []byte) { dp.AppendRowSlot(storeRow) } // AppendRowSlot 追加一行并返回其槽位号 (0 表示装不下)。 func (dp *DataPage) AppendRowSlot(storeRow []byte) uint32 { need := uint32(4 + len(storeRow)) if !dp.canFit(need) { return 0 } off := dp.dataEnd binary.LittleEndian.PutUint32(dp.buf[off:], uint32(len(storeRow))) copy(dp.buf[off+4:], storeRow) dp.dataEnd = off + need dp.count++ dp.liveCount++ dp.updateMinMax(storeRow) return dp.allocSlot(off) } // InsertRowSorted 保持"页内行按 pkID 升序"不变式, 把行插到正确位置。 // 返回 false 表示页内空间不足, 调用方需换页。 // // 为什么不能直接用 AppendRow: FindRowByPK 依赖行按 pkID 升序, 遇到比目标更大的行就 // 提前判定"不存在"。主键由调用方显式给出时 (自定义主键 / BaseModel.ID) 插入顺序与 // pkID 顺序无关, AppendRow 会把小 pkID 排到大 pkID 后面 → 行明明在页里, 却查不到 // (全叶扫描兜底也走同一个 FindRowByPK, 同样漏)。 func (dp *DataPage) InsertRowSorted(storeRow []byte) bool { _, ok := dp.InsertRowSortedSlot(storeRow) return ok } // InsertRowSortedSlot 同 InsertRowSorted, 额外返回新行的槽位号。 func (dp *DataPage) InsertRowSortedSlot(storeRow []byte) (uint32, bool) { return dp.insertRowSortedByOffsetSlot(storeRow, 0) } // InsertRowSortedByOffset 按 storeRow 中 keyOff 处的 8 字节 BigEndian 键升序插入。 // // 与 InsertRowSorted 的唯一区别只是"排序键在哪": // - 主键 / 二级索引的行, 键(或 pkID 前缀)在偏移 0; // - 时序行是 [pkID(8B)][schemaVer(1B)][字段...], 时间键(首列)在偏移 9。 // // 两种布局都需要页内有序: 有序的页才能让查找遇到更大键就提前终止, 也让范围扫描 // 能在页内直接二分/截断, 而不是整页逐行过滤。行长度不足以容纳该键时退化为追加。 func (dp *DataPage) InsertRowSortedByOffset(storeRow []byte, keyOff int) bool { _, ok := dp.insertRowSortedByOffsetSlot(storeRow, keyOff) return ok } func (dp *DataPage) insertRowSortedByOffsetSlot(storeRow []byte, keyOff int) (uint32, bool) { need := uint32(4 + len(storeRow)) if !dp.canFit(need) { return 0, false } hasKey := keyOff >= 0 && len(storeRow) >= keyOff+8 var key uint64 if hasKey { key = binary.BigEndian.Uint64(storeRow[keyOff : keyOff+8]) } // 找到第一个排序键大于 key 的行, 插在它前面; 没有则追加到末尾 ins := dp.dataEnd if hasKey { off := HdrSize for i := uint32(0); i < dp.count; i++ { rowLen, _, rowStart, ok := dp.nextRow(off) if !ok { break } if rowLen >= uint32(keyOff+8) && binary.BigEndian.Uint64(dp.buf[rowStart+uint32(keyOff):rowStart+uint32(keyOff)+8]) > key { ins = off break } off = rowStart + rowLen } } if ins < dp.dataEnd { copy(dp.buf[ins+need:dp.dataEnd+need], dp.buf[ins:dp.dataEnd]) dp.shiftSlots(ins, need) // 行整体后移: 槽位偏移同步 } binary.LittleEndian.PutUint32(dp.buf[ins:], uint32(len(storeRow))) copy(dp.buf[ins+4:], storeRow) dp.dataEnd += need dp.count++ dp.liveCount++ dp.updateMinMax(storeRow) return dp.allocSlot(ins), true } // ======================== 按槽位读写 ======================== // ReadRowBySlot 按槽位号读取行数据副本 (墓碑/空槽返回 false)。 func (dp *DataPage) ReadRowBySlot(slotID uint32) ([]byte, bool) { return dp.readRowBySlot(slotID, nil) } // ReadRowBySlotPK 按槽位号读取行, 并校验行首 8B 是否等于 pkKeyBE (防御陈旧槽位)。 func (dp *DataPage) ReadRowBySlotPK(slotID uint32, pkKeyBE []byte) ([]byte, bool) { if len(pkKeyBE) < 8 { return nil, false } return dp.readRowBySlot(slotID, pkKeyBE) } func (dp *DataPage) readRowBySlot(slotID uint32, pkKeyBE []byte) ([]byte, bool) { if slotID == 0 || slotID > dp.slotCount { return nil, false } off := dp.getSlot(slotID) if off == 0 { return nil, false } rowLen, deleted, rowStart, ok := dp.nextRow(off) if !ok || deleted || rowLen == 0 { return nil, false } if pkKeyBE != nil { if rowLen < 8 || binary.BigEndian.Uint64(dp.buf[rowStart:rowStart+8]) != binary.BigEndian.Uint64(pkKeyBE[:8]) { return nil, false } } result := make([]byte, rowLen) copy(result, dp.buf[rowStart:rowStart+rowLen]) return result, true } // DeleteBySlot 对指定槽位上的行打墓碑 (行内直接标记删除), 返回是否成功。 // 失效字节随之累加, 供整理页阈值判断。 func (dp *DataPage) DeleteBySlot(slotID uint32) bool { if slotID == 0 || slotID > dp.slotCount { return false } off := dp.getSlot(slotID) if off == 0 { return false } rowLen, deleted, _, ok := dp.nextRow(off) if !ok || deleted { return false } binary.LittleEndian.PutUint32(dp.buf[off:], rowLen|DeletedFlag) dp.deadBytes += 4 + rowLen dp.liveCount-- return true } // ======================== 查找 ======================== // FindRowByPK 按 pkID 查找 (pkKeyBE 为 BigEndian, 匹配 storedRow[0:8]) // 优化: 行按 pkID 顺序存储, 找到 pkID > target 时提前终止 // // ⚠️ 仅当页内行确实按 pkID 升序时才能用: 提前终止会把"页里明明有"的行判成不存在。 // 时序 DataPage 的行按时间键排序 (见 InsertRowSortedByOffset), pkID 无序, 必须走 // FindRowByPKLinear。 func (dp *DataPage) FindRowByPK(pkKeyBE []byte) ([]byte, bool) { _, row, ok := dp.findRowByPKSlot(pkKeyBE) return row, ok } // FindRowByPKSlot 同 FindRowByPK, 额外返回该行的槽位号 (便于后续按槽位标删)。 func (dp *DataPage) FindRowByPKSlot(pkKeyBE []byte) (uint32, []byte, bool) { return dp.findRowByPKSlot(pkKeyBE) } func (dp *DataPage) findRowByPKSlot(pkKeyBE []byte) (uint32, []byte, bool) { keyPK := binary.BigEndian.Uint64(pkKeyBE[:8]) for i := uint32(1); i <= dp.slotCount; i++ { off := dp.getSlot(i) if off == 0 { continue } rowLen, deleted, rowStart, ok := dp.nextRow(off) if !ok || deleted || rowLen < 8 { continue } rowPK := binary.BigEndian.Uint64(dp.buf[rowStart : rowStart+8]) if rowPK == keyPK { result := make([]byte, rowLen) copy(result, dp.buf[rowStart:rowStart+rowLen]) return i, result, true } } return 0, nil, false } // FindRowByPKLinear 按 pkID 全页扫描查找, 不做"遇到更大行号即提前结束"的优化。 // 页内行不按 pkID 有序时 (时序 DataPage 按时间键排序) 必须用本方法, 否则会漏行。 func (dp *DataPage) FindRowByPKLinear(pkKeyBE []byte) ([]byte, bool) { keyPK := binary.BigEndian.Uint64(pkKeyBE[:8]) off := HdrSize for i := uint32(0); i < dp.count; i++ { rowLen, deleted, rowStart, ok := dp.nextRow(off) if !ok { return nil, false } off = rowStart + rowLen if deleted || rowLen < 8 { continue } if binary.BigEndian.Uint64(dp.buf[rowStart:rowStart+8]) == keyPK { result := make([]byte, rowLen) copy(result, dp.buf[rowStart:rowStart+rowLen]) return result, true } } return nil, false } // FindRowByIndexKey 按 rowData 中指定偏移处的 8 字节匹配 // indexKeyBE 为 BigEndian 编码的索引键 // storedOffset 为 storedRow 中比较的起始偏移 func (dp *DataPage) FindRowByIndexKey(indexKeyBE []byte, storedOffset int) ([]byte, bool) { _, row, ok := dp.findRowByIndexKeySlot(indexKeyBE, storedOffset) return row, ok } // FindRowByIndexKeySlot 同 FindRowByIndexKey, 额外返回槽位号。 func (dp *DataPage) FindRowByIndexKeySlot(indexKeyBE []byte, storedOffset int) (uint32, []byte, bool) { return dp.findRowByIndexKeySlot(indexKeyBE, storedOffset) } func (dp *DataPage) findRowByIndexKeySlot(indexKeyBE []byte, storedOffset int) (uint32, []byte, bool) { keyVal := binary.BigEndian.Uint64(indexKeyBE[:8]) for i := uint32(1); i <= dp.slotCount; i++ { off := dp.getSlot(i) if off == 0 { continue } rowLen, deleted, rowStart, ok := dp.nextRow(off) if !ok || deleted { continue } if int(rowLen) >= storedOffset+8 { rowVal := binary.BigEndian.Uint64(dp.buf[rowStart+uint32(storedOffset) : rowStart+uint32(storedOffset)+8]) if rowVal == keyVal { result := make([]byte, rowLen) copy(result, dp.buf[rowStart:rowStart+rowLen]) return i, result, true } } } return 0, nil, false } // ======================== 软删除 ======================== // DeleteByPK 按 pkID 软删除 (行内直接打墓碑), 返回是否找到 func (dp *DataPage) DeleteByPK(pkKeyBE []byte) bool { _, ok := dp.DeleteByPKSlot(pkKeyBE) return ok } // DeleteByPKSlot 同 DeleteByPK, 额外返回被删除行的槽位号。 func (dp *DataPage) DeleteByPKSlot(pkKeyBE []byte) (uint32, bool) { keyPK := binary.BigEndian.Uint64(pkKeyBE[:8]) off := HdrSize for i := uint32(0); i < dp.count; i++ { rowLen, deleted, rowStart, ok := dp.nextRow(off) if !ok { return 0, false } lenOff := off off = rowStart + rowLen if deleted { continue } if rowLen >= 8 { rowPK := binary.BigEndian.Uint64(dp.buf[rowStart : rowStart+8]) if rowPK == keyPK { binary.LittleEndian.PutUint32(dp.buf[lenOff:], rowLen|DeletedFlag) dp.deadBytes += 4 + rowLen dp.liveCount-- return dp.slotOfOffset(lenOff), true } } } return 0, false } // DeleteByIndexKey 按 rowData 中指定偏移处的 8 字节匹配并软删除 func (dp *DataPage) DeleteByIndexKey(indexKeyBE []byte, storedOffset int) bool { _, ok := dp.DeleteByIndexKeySlot(indexKeyBE, storedOffset) return ok } // DeleteByIndexKeySlot 同 DeleteByIndexKey, 额外返回被删除行的槽位号。 func (dp *DataPage) DeleteByIndexKeySlot(indexKeyBE []byte, storedOffset int) (uint32, bool) { keyVal := binary.BigEndian.Uint64(indexKeyBE[:8]) off := HdrSize for i := uint32(0); i < dp.count; i++ { rowLen, deleted, rowStart, ok := dp.nextRow(off) if !ok { return 0, false } lenOff := off off = rowStart + rowLen if deleted { continue } if int(rowLen) >= storedOffset+8 { rowVal := binary.BigEndian.Uint64(dp.buf[rowStart+uint32(storedOffset) : rowStart+uint32(storedOffset)+8]) if rowVal == keyVal { binary.LittleEndian.PutUint32(dp.buf[lenOff:], rowLen|DeletedFlag) dp.deadBytes += 4 + rowLen dp.liveCount-- return dp.slotOfOffset(lenOff), true } } } return 0, false } // slotOfOffset 反查行前缀偏移对应的槽位号 (0 = 无)。 func (dp *DataPage) slotOfOffset(off uint32) uint32 { for i := uint32(1); i <= dp.slotCount; i++ { if dp.getSlot(i) == off { return i } } return 0 } // ======================== 原地更新 / 页内重建 ======================== // UpdateByPK 按 pkID 原地更新行数据 (newStoreRow 格式同 AppendRow) func (dp *DataPage) UpdateByPK(pkKeyBE []byte, newStoreRow []byte) error { keyPK := binary.BigEndian.Uint64(pkKeyBE[:8]) off := HdrSize for i := uint32(0); i < dp.count; i++ { rowLen, deleted, rowStart, ok := dp.nextRow(off) if !ok { break } off = rowStart + rowLen if deleted { continue } if rowLen >= 8 { rowPK := binary.BigEndian.Uint64(dp.buf[rowStart : rowStart+8]) if rowPK == keyPK { if uint32(len(newStoreRow)) <= rowLen { copy(dp.buf[rowStart:], newStoreRow) for j := uint32(len(newStoreRow)); j < rowLen; j++ { dp.buf[rowStart+j] = 0 } dp.Save() return nil } return errRowTooLarge } } } return page.ErrNotFound } // UpdateBySlot 按槽位号原地更新行数据 (newStoreRow 格式同 AppendRow)。 // // 槽位号来自主键叶条目, 定位是 O(1); 新行长度不超过旧行时直接覆盖, 否则返回 // errRowTooLarge, 由调用方转 RebuildWithSlot / 换页。 // 不变式 I2: 槽位号不变, 只改行内字节。 func (dp *DataPage) UpdateBySlot(slotID uint32, newStoreRow []byte) error { if slotID == 0 || slotID > dp.slotCount { return page.ErrNotFound } off := dp.getSlot(slotID) if off == 0 { return page.ErrNotFound } rowLen, deleted, rowStart, ok := dp.nextRow(off) if !ok || deleted { return page.ErrNotFound } if uint32(len(newStoreRow)) > rowLen { return errRowTooLarge } copy(dp.buf[rowStart:], newStoreRow) for j := uint32(len(newStoreRow)); j < rowLen; j++ { dp.buf[rowStart+j] = 0 } dp.Save() return nil } // FindAndUpdateByIndexKey 按 rowData 偏移匹配找到行后, 用 newStoreRow 原地更新 func (dp *DataPage) FindAndUpdateByIndexKey(indexKeyBE []byte, storedOffset int, newStoreRow []byte) error { keyVal := binary.BigEndian.Uint64(indexKeyBE[:8]) off := HdrSize for i := uint32(0); i < dp.count; i++ { rowLen, deleted, rowStart, ok := dp.nextRow(off) if !ok { break } off = rowStart + rowLen if deleted { continue } if int(rowLen) >= storedOffset+8 { rowVal := binary.BigEndian.Uint64(dp.buf[rowStart+uint32(storedOffset) : rowStart+uint32(storedOffset)+8]) if rowVal == keyVal { if uint32(len(newStoreRow)) <= rowLen { copy(dp.buf[rowStart:], newStoreRow) for j := uint32(len(newStoreRow)); j < rowLen; j++ { dp.buf[rowStart+j] = 0 } dp.Save() return nil } return errRowTooLarge } } } return page.ErrNotFound } // RebuildWith 页内重建 (repack): 保持活行原有顺序紧凑重排, 并把 pkKeyBE 对应的行 // 原地替换为 newStoreRow, 允许该行**变长**。 // // 用途: schema 演进 (AddField) 后旧行的 schemaVer 与行长都会变化, UpdateByPK 因 // "新行长 > 旧行长" 返回 errRowTooLarge 时, 用本方法在页内腾挪: // - 顺带回收墓碑 (已删除行) 占用的碎片空间 // - 只要整页还装得下, 变长行无需换页 // // 安全性 (不变式 I2): 活行的槽位号保持不变 (只改写其指向的偏移), 墓碑槽位被释放, // 因此页内重排不会破坏主键 B+ 树叶条目里的 (dpID, slot)。 // // 返回 page.ErrNotFound 表示目标行不在本页; 返回 errRowTooLarge 表示整页装不下, 调用方需换页。 func (dp *DataPage) RebuildWith(pkKeyBE []byte, newStoreRow []byte) error { return dp.RebuildWithIndexKey(pkKeyBE, 0, newStoreRow) } // RebuildWithIndexKey 同 RebuildWith, 但按 storedRow 中 storedOffset 处的 8 字节匹配 // (副索引/时序行布局: 见 FindAndUpdateByIndexKey) func (dp *DataPage) RebuildWithIndexKey(indexKeyBE []byte, storedOffset int, newStoreRow []byte) error { if len(indexKeyBE) < 8 { return page.ErrNotFound } keyVal := binary.BigEndian.Uint64(indexKeyBE[:8]) return dp.rebuildCore(func(_ uint32, row []byte) bool { return len(row) >= storedOffset+8 && binary.BigEndian.Uint64(row[storedOffset:storedOffset+8]) == keyVal }, newStoreRow) } // RebuildWithSlot 按槽位号定位并走页内重建, 语义同 RebuildWith, 但**不做键匹配** —— // 槽位号来自主键叶条目, 已经是 O(1) 定位, 省掉整页扫键。 // // 注意: 调用方应先确认该槽位确实是目标行 (叶条目本身就是这个保证), 否则会替换到别的行。 func (dp *DataPage) RebuildWithSlot(slotID uint32, newStoreRow []byte) error { if slotID == 0 || slotID > dp.slotCount { return page.ErrNotFound } return dp.rebuildCore(func(slot uint32, _ []byte) bool { return slot == slotID }, newStoreRow) } // rebuildCore 页内重建的公共实现: 按物理顺序收集活行, 把第一个 match 命中的行替换为 // newStoreRow, 释放墓碑槽位后紧凑重写。 // // 不变式 (必须守住): 活行的槽位号不变 —— 主键 B+ 树叶条目存的就是 (dpID, slot)。 func (dp *DataPage) rebuildCore(match func(slot uint32, row []byte) bool, newStoreRow []byte) error { // 1. 按页内物理顺序收集 (槽位, 偏移), 保证重排后仍与原来同序 type ent struct { slot uint32 off uint32 } ents := make([]ent, 0, dp.slotCount) for i := uint32(1); i <= dp.slotCount; i++ { if o := dp.getSlot(i); o != 0 { ents = append(ents, ent{i, o}) } } sort.Slice(ents, func(a, b int) bool { return ents[a].off < ents[b].off }) liveRows := make([][]byte, 0, len(ents)) liveSlots := make([]uint32, 0, len(ents)) deadSlots := make([]uint32, 0, len(ents)) replaced := false for _, e := range ents { rowLen, deleted, rowStart, ok := dp.nextRow(e.off) if !ok || deleted { deadSlots = append(deadSlots, e.slot) continue } row := make([]byte, rowLen) copy(row, dp.buf[rowStart:rowStart+rowLen]) if !replaced && match(e.slot, row) { replaced = true liveRows = append(liveRows, newStoreRow) liveSlots = append(liveSlots, e.slot) continue } liveRows = append(liveRows, row) liveSlots = append(liveSlots, e.slot) } if !replaced { return page.ErrNotFound } // 2. 预计算总占用, 装不下则告知调用方换页 (目录容量不变) const rowHdrSize = uint32(4) // 每行 4B 长度前缀 total := HdrSize for _, r := range liveRows { total += rowHdrSize + uint32(len(r)) } if total > uint32(PageSize)-slotCountSize-dp.slotCount*slotEntrySize { return errRowTooLarge } // 3. 按原顺序紧凑重写, 同步槽位 for _, s := range deadSlots { dp.putSlot(s, 0) } w := HdrSize for i, r := range liveRows { dp.putSlot(liveSlots[i], w) binary.LittleEndian.PutUint32(dp.buf[w:], uint32(len(r))) w += rowHdrSize copy(dp.buf[w:], r) w += uint32(len(r)) } for i := w; i < dp.dataEnd; i++ { dp.buf[i] = 0 } dp.dataEnd = w dp.count = uint32(len(liveRows)) dp.deadBytes = 0 // 墓碑已全部物理清除 dp.liveCount = uint32(len(liveRows)) dp.recomputeMinMax(liveRows) dp.Save() return nil } // ======================== 全页视图 ======================== // AllDeleted 检查是否所有行都已删除 (O(1), 读页头镜像的存活计数) func (dp *DataPage) AllDeleted() bool { return dp.liveCount == 0 } // CountLiveRows 统计本页未删除的行数 (O(1), 读页头镜像的存活计数)。 // // 为无条件 COUNT(*) 服务: 计数由所有写路径增量维护并随页落盘 // (page.DataPageHdr.LiveCnt), 不必逐行物化、也不必逐行走链。 func (dp *DataPage) CountLiveRows() int { return int(dp.liveCount) } // GetAllRows 返回所有未删除的行 func (dp *DataPage) GetAllRows() [][]byte { var result [][]byte off := HdrSize for i := uint32(0); i < dp.count; i++ { rowLen, deleted, rowStart, ok := dp.nextRow(off) if !ok { break } off = rowStart + rowLen if deleted { continue } row := make([]byte, rowLen) copy(row, dp.buf[rowStart:rowStart+rowLen]) result = append(result, row) } return result } // GetAllRowsPooled 同 GetAllRows, 但从 sync.Pool 获取缓冲区 // 调用方处理完后必须将每个 []byte 归还给 pool func (dp *DataPage) GetAllRowsPooled(pool *sync.Pool) [][]byte { var result [][]byte off := HdrSize for i := uint32(0); i < dp.count; i++ { rowLen, deleted, rowStart, ok := dp.nextRow(off) if !ok { break } off = rowStart + rowLen if deleted { continue } // 从池中获取缓冲区, 避免 make 分配 bp := pool.Get().(*[]byte) buf := *bp copy(buf[:rowLen], dp.buf[rowStart:rowStart+rowLen]) result = append(result, buf[:rowLen]) } return result } // GetAllRowsReversed 返回所有未删除行,顺序反转(最后一行在前) func (dp *DataPage) GetAllRowsReversed() [][]byte { rows := dp.GetAllRows() for i, j := 0, len(rows)-1; i < j; i, j = i+1, j-1 { rows[i], rows[j] = rows[j], rows[i] } return rows } // GetAllRowsWithSlots 返回所有未删除行及其槽位号 (顺序 = 页内物理顺序)。 func (dp *DataPage) GetAllRowsWithSlots() ([]uint32, [][]byte) { var slots []uint32 var rows [][]byte off := HdrSize for i := uint32(0); i < dp.count; i++ { rowLen, deleted, rowStart, ok := dp.nextRow(off) if !ok { break } slotOff := off off = rowStart + rowLen if deleted { continue } row := make([]byte, rowLen) copy(row, dp.buf[rowStart:rowStart+rowLen]) rows = append(rows, row) slots = append(slots, dp.slotOfOffset(slotOff)) } return slots, rows } // 注: 原 RewriteRows (用给定行紧凑重写整页, 槽位目录整体重新编号) 已删除 —— // 它把第 i 行重新编到槽位 i+1, 直接违反槽位不变式 I2: 主键叶条目存的是 (dpID, slot), // 一旦重编号, 所有叶条目就指向别的行。 // // 需要「紧凑重排 / 回收墓碑」时用 Compact 或 MaybeCompact (二者都保留活行的槽位号); // 需要「替换某一行且允许变长」时用 RebuildWith / RebuildWithSlot。 // AppendRows 依次追加多行: 先整体校验容量, 再逐行写入。 // 任一行装不下时整页保持原样并返回 false —— 不会留下"写了一半"的页。 func (dp *DataPage) AppendRows(rows [][]byte) bool { total := dp.dataEnd for _, r := range rows { total += 4 + uint32(len(r)) } if total > dp.freeLimit() { return false } for _, r := range rows { dp.AppendRow(r) } return true } // ======================== 内部辅助 ======================== // nextRow 解析下一行: 返回 (数据长度, 是否已删除, 数据起始偏移, 是否有效) func (dp *DataPage) nextRow(off uint32) (rowLen uint32, deleted bool, rowStart uint32, ok bool) { if off+4 > dp.dataEnd { return 0, false, 0, false } rawLen := binary.LittleEndian.Uint32(dp.buf[off:]) deleted = rawLen&DeletedFlag != 0 rowLen = rawLen & ^DeletedFlag rowStart = off + 4 if rowStart+rowLen > dp.dataEnd { return 0, false, 0, false } return rowLen, deleted, rowStart, true } // ======================== 裸缓冲辅助 (WAL 行级恢复) ======================== // RawAppendRowSlot 在裸页缓冲上追加一行并分配槽位 (供 WAL 行级恢复使用)。 // 语义与 DataPage.AppendRow 一致, 但不构造对象; 返回是否成功。 func RawAppendRowSlot(buf []byte, storeRow []byte) bool { if len(buf) < int(HdrSize) { return false } dh := page.DataHdrOf(buf) dataEnd := dh.UseSize if dataEnd < HdrSize { dataEnd = HdrSize } sc := uint32(binary.LittleEndian.Uint16(buf[PageSize-slotCountSize:])) if sc > maxSlotCount || uint32(PageSize)-slotCountSize-sc*slotEntrySize < dataEnd { sc = 0 } need := uint32(4 + len(storeRow)) if dataEnd+need > uint32(PageSize)-slotCountSize-(sc+1)*slotEntrySize { return false // 放不下 (含为新增槽位预留的目录空间) } binary.LittleEndian.PutUint32(buf[dataEnd:], uint32(len(storeRow))) copy(buf[dataEnd+4:], storeRow) off := dataEnd // 分配槽位: 复用空槽, 否则扩容 slot := uint32(0) for i := uint32(1); i <= sc; i++ { p := uint32(PageSize) - slotCountSize - i*slotEntrySize if binary.LittleEndian.Uint16(buf[p:]) == 0 { binary.LittleEndian.PutUint16(buf[p:], uint16(off)) slot = i break } } if slot == 0 { if sc >= maxSlotCount { return false } sc++ binary.LittleEndian.PutUint16(buf[PageSize-slotCountSize:], uint16(sc)) p := uint32(PageSize) - slotCountSize - sc*slotEntrySize binary.LittleEndian.PutUint16(buf[p:], uint16(off)) } dh.Count++ dh.LiveCnt++ dh.UseSize = dataEnd + need // Min/Max 维护 (与 updateMinMax 同规则) if len(storeRow) >= 8 { val := binary.BigEndian.Uint64(storeRow[:8]) if dh.Count <= 1 { dh.SetMin(val) dh.SetMax(val) } else { if val < dh.GetMin() { dh.SetMin(val) } if val > dh.GetMax() { dh.SetMax(val) } } } return true } // ======================== IDateEntry 接口实现 ======================== // 供时序索引等扁平链模式使用 // IDateEntryDataPage 包装 DataPage 使其实现 IDateEntry 接口 // FindDataPage 回调负责从 indexEntries 中找到目标 DataPageID type IDateEntryDataPage struct { FindDataPage func(indexEntries []byte, entryLen int, count int, key []byte) uint32 } func (d *IDateEntryDataPage) Insert(handle page.HandleDb, indexEntries []byte, entryLen int, count int, key []byte, pkID uint64, rowData []byte) error { dpID := d.FindDataPage(indexEntries, entryLen, count, key) var dp *DataPage if dpID == 0 { dp = New(handle) } else { var err error dp, err = Load(handle, dpID) if err != nil { return err } } if dp.IsFull(uint32(8 + len(rowData))) { dp.Save() dp = New(handle) } storeRow := make([]byte, 8+len(rowData)) binary.BigEndian.PutUint64(storeRow[:8], pkID) copy(storeRow[8:], rowData) dp.AppendRow(storeRow) dp.Save() return nil } func (d *IDateEntryDataPage) Search(handle page.HandleDb, indexEntries []byte, entryLen int, key []byte) ([]byte, bool) { n := len(indexEntries) / entryLen dpID := d.FindDataPage(indexEntries, entryLen, n, key) if dpID == 0 { return nil, false } dp, err := Load(handle, dpID) if err != nil { return nil, false } // 时序行布局: [pkID(8)][schemaVer(1)][字段...], 时间戳是首列 → storedRow 偏移 9 return dp.FindRowByIndexKey(key, 9) } func (d *IDateEntryDataPage) Update(handle page.HandleDb, indexEntries []byte, entryLen int, key []byte, pkID uint64, rowData []byte) error { n := len(indexEntries) / entryLen dpID := d.FindDataPage(indexEntries, entryLen, n, key) if dpID == 0 { return page.ErrNotFound } dp, err := Load(handle, dpID) if err != nil { return err } storeRow := make([]byte, 8+len(rowData)) binary.BigEndian.PutUint64(storeRow[:8], pkID) copy(storeRow[8:], rowData) return dp.FindAndUpdateByIndexKey(key, 9, storeRow) } func (d *IDateEntryDataPage) Delete(handle page.HandleDb, indexEntries []byte, entryLen int, key []byte) error { n := len(indexEntries) / entryLen dpID := d.FindDataPage(indexEntries, entryLen, n, key) if dpID == 0 { return page.ErrNotFound } dp, err := Load(handle, dpID) if err != nil { return err } deleted := dp.DeleteByIndexKey(key, 9) if !deleted { return page.ErrNotFound } if dp.AllDeleted() { if handle.FreePage != nil { handle.FreePage(dp.pageID) } } else { dp.Save() } return nil } func (d *IDateEntryDataPage) CreateDataPage(handle page.HandleDb, key []byte, pkID uint64, rowData []byte) (uint32, error) { dp := New(handle) storeRow := make([]byte, 8+len(rowData)) binary.BigEndian.PutUint64(storeRow[:8], pkID) copy(storeRow[8:], rowData) dp.AppendRow(storeRow) dp.Save() return dp.pageID, nil }