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

package datapage

import (
	"git.ycl.pub/yl32940/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() {
		// 延迟还链, 理由同 TimeDataPage.Delete
		handle.DeferFreeDataPage(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
}