TsDB2/DB/core/index/datapage/datapage.go

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
}