package datapage import ( "TsDB2/DB/core/page" "encoding/binary" "errors" "sync" ) // PageSize 索引页大小 (避免导入 index 包造成循环依赖) const PageSize = 16384 var ( // errRowTooLarge 行装不下当前页, 调用方需换页/重建。 // 注: "未找到"语义统一用 page.ErrNotFound, 便于上层 errors.Is 判定, 不再各自造错误值。 errRowTooLarge = errors.New("row too large, need heap overflow") ) // ============================================================================= // 统一 DataPage — 所有需要独立数据页的索引类型共享 // // 磁盘布局 (权威定义见 page.DataPageHdr): // [PageHdr(20B) + Count(4B) + Min(8B) + Max(8B)] = 40B header // [len(4B) + rowData] ... = 行数据区 // // len 最高位 (0x80000000) 为软删除标志 // Min/Max 存储 rowData 前 8 字节的最小/最大值 (BigEndian 比较) // ============================================================================= // HdrSize 数据页头部大小 (直接复用 page 包的权威布局定义) const HdrSize = page.DataPageHdrSize // DeletedFlag 软删除标志 (长度字段最高位) const DeletedFlag uint32 = 0x80000000 // DataPage 统一数据页 type DataPage struct { handle page.HandleDb pageID uint32 buf []byte count uint32 dataEnd uint32 } // ======================== 页 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 := make([]byte, PageSize) if err := handle.ReadPage(pageID, buf); err != nil { return nil, err } hdr := page.HdrOf(buf) cnt := page.DataHdrOf(buf).Count return &DataPage{ handle: handle, pageID: pageID, buf: buf, count: cnt, dataEnd: hdr.UseSize, }, nil } // Save 持久化到磁盘 func (dp *DataPage) Save() { page.HdrOf(dp.buf).UseSize = dp.dataEnd page.DataHdrOf(dp.buf).Count = dp.count _ = 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 } // IsFull 判断是否有足够空间追加新行 func (dp *DataPage) IsFull(rowSize uint32) bool { return dp.dataEnd+rowSize+4 >= uint32(len(dp.buf)) } // ======================== 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 LE) + rowData] 或纯 rowData, 由调用方决定 func (dp *DataPage) AppendRow(storeRow []byte) { off := dp.dataEnd binary.LittleEndian.PutUint32(dp.buf[off:], uint32(len(storeRow))) off += 4 copy(dp.buf[off:], storeRow) off += uint32(len(storeRow)) dp.dataEnd = off dp.count++ // 更新 Min/Max: 取 storeRow 前 8 字节 dp.updateMinMax(storeRow) } // InsertRowSorted 保持"页内行按 pkID 升序"不变式, 把行插到正确位置。 // 返回 false 表示页内空间不足, 调用方需换页。 // // 为什么不能直接用 AppendRow: FindRowByPK 依赖行按 pkID 升序, 遇到比目标更大的行就 // 提前判定"不存在"。主键由调用方显式给出时 (自定义主键 / BaseModel.ID) 插入顺序与 // pkID 顺序无关, AppendRow 会把小 pkID 排到大 pkID 后面 → 行明明在页里, 却查不到 // (全叶扫描兜底也走同一个 FindRowByPK, 同样漏)。 func (dp *DataPage) InsertRowSorted(storeRow []byte) bool { return dp.InsertRowSortedByOffset(storeRow, 0) } // InsertRowSortedByOffset 按 storeRow 中 keyOff 处的 8 字节 BigEndian 键升序插入。 // // 与 InsertRowSorted 的唯一区别只是"排序键在哪": // - 主键 / 二级索引的行, 键(或 pkID 前缀)在偏移 0; // - 时序行是 [pkID(8B)][schemaVer(1B)][字段...], 时间键(首列)在偏移 9。 // // 两种布局都需要页内有序: 有序的页才能让查找遇到更大键就提前终止, 也让范围扫描 // 能在页内直接二分/截断, 而不是整页逐行过滤。行长度不足以容纳该键时退化为追加。 func (dp *DataPage) InsertRowSortedByOffset(storeRow []byte, keyOff int) bool { need := uint32(4 + len(storeRow)) if dp.dataEnd+need >= uint32(len(dp.buf)) { return 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]) } binary.LittleEndian.PutUint32(dp.buf[ins:], uint32(len(storeRow))) copy(dp.buf[ins+4:], storeRow) dp.dataEnd += need dp.count++ dp.updateMinMax(storeRow) 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) { 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 { continue } if rowLen >= 8 { rowPK := binary.BigEndian.Uint64(dp.buf[rowStart : rowStart+8]) if rowPK == keyPK { result := make([]byte, rowLen) copy(result, dp.buf[rowStart:rowStart+rowLen]) return result, true } if rowPK > keyPK { // 行按 pkID 顺序, 当前行 pkID > 目标, 后续行更大, 提前终止 return nil, false } } } return 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) { 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 nil, false } 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 { result := make([]byte, rowLen) copy(result, dp.buf[rowStart:rowStart+rowLen]) return result, true } } } return nil, false } // DeleteByPK 按 pkID 软删除, 返回是否找到 func (dp *DataPage) DeleteByPK(pkKeyBE []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 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) return true } } } return false } // DeleteByIndexKey 按 rowData 中指定偏移处的 8 字节匹配并软删除 func (dp *DataPage) DeleteByIndexKey(indexKeyBE []byte, storedOffset int) 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 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) return true } } } return false } // 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 } // 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 时, 用本方法在页内腾挪: // - 顺带回收墓碑 (已删除行) 占用的碎片空间 // - 只要整页还装得下, 变长行无需换页 // // 安全性: 行以自描述的 4B 长度前缀顺序排列, 页内偏移不对外暴露 (外部只通过 // B+ 树叶子的 dpID 定位页面后顺序扫描), 因此页内重排不会破坏任何索引。 // // 返回 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]) // 1. 收集活行, 命中目标键的位置替换为新行 live := make([][]byte, 0, dp.count) off := HdrSize replaced := false 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 !replaced && int(rowLen) >= storedOffset+8 && binary.BigEndian.Uint64(dp.buf[rowStart+uint32(storedOffset):rowStart+uint32(storedOffset)+8]) == keyVal { replaced = true live = append(live, newStoreRow) continue } row := make([]byte, rowLen) copy(row, dp.buf[rowStart:rowStart+rowLen]) live = append(live, row) } if !replaced { return page.ErrNotFound } // 2. 预计算总占用, 装不下则告知调用方换页 const rowHdrSize = uint32(4) // 每行 4B 长度前缀 total := HdrSize for _, r := range live { total += rowHdrSize + uint32(len(r)) } if total > uint32(len(dp.buf)) { return errRowTooLarge } // 3. 按原顺序紧凑重写 w := HdrSize for _, r := range live { 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(live)) dp.recomputeMinMax(live) dp.Save() return nil } // AllDeleted 检查是否所有行都已删除 func (dp *DataPage) AllDeleted() bool { off := HdrSize for i := uint32(0); i < dp.count; i++ { rowLen, deleted, rowStart, ok := dp.nextRow(off) if !ok { return true } if !deleted { return false } off = rowStart + rowLen } return true } // 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 } // RewriteRows 用给定行紧凑重写整页 (顺带回收墓碑), 行顺序即入参顺序。 // // 页分裂 / 页内重排都靠它: 一次写入即完成 "丢弃墓碑 + 重排 + 重算 Count/Min/Max"。 // 返回 false 表示总占用超出页容量 (调用方需换页或分裂), 此时页内容保持不变。 func (dp *DataPage) RewriteRows(rows [][]byte) bool { const rowHdr = uint32(4) // 每行 4B 长度前缀 total := HdrSize for _, r := range rows { total += rowHdr + uint32(len(r)) } if total > uint32(len(dp.buf)) { return false } w := HdrSize for _, r := range rows { binary.LittleEndian.PutUint32(dp.buf[w:], uint32(len(r))) w += rowHdr 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(rows)) dp.recomputeMinMax(rows) return true } // AppendRows 依次追加多行: 先整体校验容量, 再逐行写入。 // 任一行装不下时整页保持原样并返回 false —— 不会留下"写了一半"的页。 func (dp *DataPage) AppendRows(rows [][]byte) bool { total := dp.dataEnd for _, r := range rows { total += 4 + uint32(len(r)) } if total > uint32(len(dp.buf)) { 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 } // ======================== 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 }