TsDB2/DB/core/index/time_handler.go

package index

import (
	"TsDB2/DB/core/index/datapage"
	timeindex "TsDB2/DB/core/index/timedatapage"
	"TsDB2/DB/core/page"
	"TsDB2/DB/core/util"
	"encoding/binary"
	"errors"
	"unsafe"
)

func init() {
	Compose.RegisterHandler(page.IndexTypeTime, &TimeIndexHandler{})
}

// ======================== IEntry: 时间索引条目 ========================

// IndexTimeEntryLen 固定长度 20
const IndexTimeEntryLen = 20

// IndexTimeEntry 时间索引条目：固定 20 字节
type IndexTimeEntry struct {
	BeginTime int64  // 8
	LastTime  int64  // 8
	PageID    uint32 // 4
}

func (e *IndexTimeEntry) DataEntry() IDateEntry {
	return &timeindex.TimeDataPage{}
}

func (e *IndexTimeEntry) EntryLength() uint16 {
	return IndexTimeEntryLen
}

// Compare 比较两个索引的 BeginTime（给二分查找用）
func (e *IndexTimeEntry) Compare(midEntry []byte, key []byte) int {
	timeMid := int64(binary.BigEndian.Uint64(midEntry[:8]))
	timeKey := int64(binary.BigEndian.Uint64(key[:8]))
	return util.CompareInt64(timeMid, timeKey)
}

// Serialize 序列化为 20 字节
func (e *IndexTimeEntry) Serialize(ie IEntry) []byte {
	if en, ok := ie.(*IndexTimeEntry); ok {
		buf := make([]byte, IndexTimeEntryLen)
		binary.BigEndian.PutUint64(buf[:8], uint64(en.BeginTime))
		binary.BigEndian.PutUint64(buf[8:16], uint64(en.LastTime))
		binary.BigEndian.PutUint32(buf[16:20], en.PageID)
		return buf
	}
	return nil
}

// Deserialize 从 []byte 解析结构体
func (e *IndexTimeEntry) Deserialize(buf []byte) IEntry {
	return &IndexTimeEntry{
		BeginTime: int64(binary.BigEndian.Uint64(buf[:8])),
		LastTime:  int64(binary.BigEndian.Uint64(buf[8:16])),
		PageID:    binary.BigEndian.Uint32(buf[16:20]),
	}
}

// Insert 向索引块尾部追加一个条目
func (e *IndexTimeEntry) Insert(entries *[]byte, data []byte) error {
	*entries = append(*entries, data...)
	return nil
}

// Search 根据时间戳查找 PageID
func (e *IndexTimeEntry) Search(entries []byte, key []byte) (uint32, bool) {
	pos, found := util.BinarySearch(entries, IndexTimeEntryLen, key, e.Compare)
	if !found {
		return 0, false
	}
	start := pos * IndexTimeEntryLen
	entryBuf := entries[start : start+IndexTimeEntryLen]
	pageID := binary.BigEndian.Uint32(entryBuf[16:20])
	return pageID, true
}

// ======================== IndexHandler: 时序索引处理器 ========================

// TimeIndexHandler — 时序索引 (page.IndexTypeTime)
// 扁平索引页链 + TimeDataPage
type TimeIndexHandler struct{}

func (h *TimeIndexHandler) Insert(handle page.HandleDb, entryPage *uint32, indexKey []byte, pkID uint64, rowData []byte, keyLen int) (uint32, error) {
	// 时序表只收时间戳非递减的写入。顺序写入才能保证每个分段覆盖一段互不重叠的
	// 时间区间、段内也天然有序 —— 范围查询才裁剪得动分段, 段内也不必整段逐行过滤。
	//
	// 顺便拿到索引链的尾页: 新分段的条目必须挂到链尾。挂到当前页会把它的 NextPage
	// 覆盖掉, 之前挂上去的索引页从此成为孤儿 —— 那些页里的分段整段查不到 (丢数据)。
	tailIdxPageID, err := h.assertTimeOrder(handle, *entryPage, indexKey)
	if err != nil {
		return 0, err
	}

	dataEntry := h.getDataEntry()
	// 顺序写入 (assertTimeOrder 已保证) 下新行的时间戳是全表最大的, 它归属的新分段
	// 只可能落在链尾页 —— 所以直接从链尾页开始找。
	//
	// 若沿用"永远从首页开始找": 首页的条目位写满之后, 每条新条目都会新建一张只放一条
	// 的索引页, 链无限增长 (而每次写入又要走完整条链去校验顺序), 同时新行会被塞进首页
	// 那个早就写满的旧分段 —— 段划分错乱, 段数暴涨。
	curPageID := *entryPage
	if tailIdxPageID != 0 {
		curPageID = tailIdxPageID
	}
	for curPageID != 0 {
		buf := make([]byte, page.IndexPageSize)
		if err := handle.ReadPage(curPageID, buf); err != nil {
			return 0, err
		}
		hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
		pageEntryLen := int(hdr.EntryLen)
		count := int(hdr.Count)
		idxPageHasRoom := hdr.Hdr.UseSize+uint32(pageEntryLen) <= page.IndexPageSize

		var entries []byte
		if idxPageHasRoom {
			entries = buf[page.IndexHeadSize : page.IndexHeadSize+(count+1)*pageEntryLen]
		} else {
			endOff := page.IndexHeadSize + count*pageEntryLen + 4
			if endOff <= len(buf) {
				entries = buf[page.IndexHeadSize:endOff]
			} else {
				overflow := make([]byte, count*pageEntryLen+4)
				copy(overflow, buf[page.IndexHeadSize:page.IndexHeadSize+count*pageEntryLen])
				entries = overflow
			}
		}

		err := dataEntry.Insert(handle, entries, pageEntryLen, count, indexKey, pkID, rowData)
		if err == nil {
			// 命中已有分段: 该分段条目的 LastTime 可能已被扩展, 需回写索引页。
			// 页内已满时 entries 是独立副本, 要把条目区拷回 buf 再落盘。
			if !idxPageHasRoom {
				copy(buf[page.IndexHeadSize:page.IndexHeadSize+count*pageEntryLen],
					entries[:count*pageEntryLen])
			}
			if err := handle.SavePage(curPageID, buf); err != nil {
				return 0, err
			}
			return *entryPage, nil
		}
		if errors.Is(err, timeindex.ErrNeedNewEntry) {
			dpID := binary.BigEndian.Uint32(entries[count*pageEntryLen : count*pageEntryLen+4])
			if idxPageHasRoom {
				insertEntrySorted(buf, hdr, buildTimeEntry(indexKey, dpID))
				if err := handle.SavePage(curPageID, buf); err != nil {
					return 0, err
				}
				return *entryPage, nil
			}
			return *entryPage, writeToNewIndexPage(handle, entryPage,
				buildTimeEntry(indexKey, dpID), tailIdxPageID)
		}
		curPageID = hdr.Hdr.NextPage
	}
	// 新分段的第一个 DataPage 必须拿到真实行号: 行存储索引的 DataPage 按 pkID
	// 增序排列, 前缀写错会让 FindRowByPK 的"遇到更大 pkID 就提前结束"立刻成立,
	// 整页行号从此不可达 (按行号读/改/删、二级索引回表统统失效)。
	newDPPID, createErr := dataEntry.CreateDataPage(handle, indexKey, pkID, rowData)
	if createErr != nil {
		return 0, createErr
	}
	return *entryPage, writeToNewIndexPage(handle, entryPage, buildTimeEntry(indexKey, newDPPID), tailIdxPageID)
}

// ======================== 顺序写入约束 ========================

// ErrOutOfOrder 时序表只接受时间戳非递减的写入, 更早的时间戳会被拒绝。
//
// 时序索引只有时间这一个维度: 顺序写入时, 每个分段天然覆盖一段互不重叠的时间区间,
// 段内行也天然按时间递增 —— 范围查询才能用 [BeginTime, LastTime] 裁剪掉无关分段。
// 一旦允许乱序, 新行会落进"BeginTime <= ts 的最后一个分段", 而它后面的分段可能覆盖
// 更晚的时间, 于是段区间互相重叠、段内时间也不再单调, 范围查询只能退化成全链 +
// 整段逐行过滤。(确有乱序数据时, 先按时间排序再写, 或改用普通表 + 二级索引。)
var ErrOutOfOrder = errors.New("time series: out-of-order timestamp rejected (must be >= the latest timestamp)")

// assertTimeOrder 拒绝比"当前最大时间戳"更早的写入, 并返回索引链的尾页 ID
// (新分段的条目要挂到链尾, 见 Insert 里的说明)。
//
// 最大时间戳取自索引页链**最后一个条目**的 LastTime: 条目在链上按 BeginTime 升序,
// 新条目只会被追加到链尾 —— 首页装得下就追加在首页尾部, 装不下就新建一页挂在链尾,
// 所以链尾页的最后一条正是最近一次写入的分段, 它的 LastTime 就是全表最大时间戳。
//
// 成本: 每次写入都要沿链走到尾, 代价与链长 (分段数 / 每页条目数) 成正比。顺序写入下
// 分段增长缓慢, 链通常只有一两页; 若将来链变得很长, 应把最大时间戳持久化进索引元数据。
func (h *TimeIndexHandler) assertTimeOrder(handle page.HandleDb, entryPage uint32, indexKey []byte) (uint32, error) {
	if entryPage == 0 {
		// 空表: 第一个条目自己就是首页, 没有链尾可挂。
		return 0, nil
	}
	maxTS, tail, ok := h.latestTimeKey(handle, entryPage)
	if ok && len(indexKey) >= 8 && int64(binary.BigEndian.Uint64(indexKey[:8])) < maxTS {
		return tail, ErrOutOfOrder
	}
	return tail, nil
}

// latestTimeKey 沿索引页链走到最后一页, 返回 (最后一条目的 LastTime, 尾页 ID, 是否有条目)。
func (h *TimeIndexHandler) latestTimeKey(handle page.HandleDb, entryPage uint32) (int64, uint32, bool) {
	var (
		last  int64
		tail  uint32
		found bool
	)
	for curPID := entryPage; curPID != 0; {
		buf := make([]byte, page.IndexPageSize)
		if err := handle.ReadPage(curPID, buf); err != nil {
			break
		}
		tail = curPID
		hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
		entryLen := int(hdr.EntryLen)
		count := int(hdr.Count)
		if entryLen >= IndexTimeEntryLen && count > 0 {
			// 条目格式: [BeginTime(8B) + LastTime(8B) + DataPageID(4B)]
			entries := buf[page.IndexHeadSize : page.IndexHeadSize+count*entryLen]
			off := (count-1)*entryLen + 8
			last = int64(binary.BigEndian.Uint64(entries[off : off+8]))
			found = true
		}
		curPID = hdr.Hdr.NextPage
	}
	if tail == 0 {
		// 一页都没读成 (异常): 退回挂到首页, 总比让新页变成孤儿好。
		tail = entryPage
	}
	return last, tail, found
}

// pkIDBytes 把行号编成 8 字节 BigEndian —— DataPage 行首前缀与 FindRowByPK 的约定。
func pkIDBytes(pkID uint64) []byte {
	buf := make([]byte, 8)
	binary.BigEndian.PutUint64(buf, pkID)
	return buf
}

// sameTimeKey 判断两个时间索引键是否指向同一时间点 (时序键定长 8 字节)。
// 取不到键时保守返回 false (当作"时间变了", 让调用方走搬移路径)。
func sameTimeKey(a, b []byte) bool {
	if len(a) < 8 || len(b) < 8 {
		return false
	}
	return tiCmp(a[:8], b[:8]) == 0
}

// findDataPageByPK 沿索引条目链定位"包含该行号"的 DataPage。
//
// 时间键只能定位到分段 (一个分段覆盖一段时间, 里面可能有多行), 行号才能定位到行。
func (h *TimeIndexHandler) findDataPageByPK(handle page.HandleDb, entryPage uint32, pkID uint64) (uint32, bool) {
	if entryPage == 0 {
		return 0, false
	}
	pkKeyBE := pkIDBytes(pkID)
	for curPageID := entryPage; curPageID != 0; {
		buf := make([]byte, page.IndexPageSize)
		if err := handle.ReadPage(curPageID, buf); err != nil {
			break
		}
		hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
		entryLen := int(hdr.EntryLen)
		count := int(hdr.Count)
		if entryLen <= 0 || count == 0 {
			curPageID = hdr.Hdr.NextPage
			continue
		}
		entries := buf[page.IndexHeadSize : page.IndexHeadSize+count*entryLen]
		for i := 0; i < count; i++ {
			dpID := binary.BigEndian.Uint32(entries[i*entryLen+entryLen-4:])
			if dpID == 0 {
				continue
			}
			dp, err := datapage.Load(handle, dpID)
			if err != nil {
				continue
			}
			if _, ok := dp.FindRowByPK(pkKeyBE); ok {
				return dpID, true
			}
		}
		curPageID = hdr.Hdr.NextPage
	}
	return 0, false
}

// buildTimeEntry 构造 20 字节时间索引条目 [BeginTime(8) + LastTime(8) + DataPageID(4)]。
// LastTime 初始等于 BeginTime, 之后随该分段被追加更大的时间戳而增长。
func buildTimeEntry(indexKey []byte, dpID uint32) []byte {
	entry := make([]byte, IndexTimeEntryLen)
	copy(entry[:8], indexKey[:8])
	copy(entry[8:16], indexKey[:8])
	binary.BigEndian.PutUint32(entry[16:20], dpID)
	return entry
}

func (h *TimeIndexHandler) Search(handle page.HandleDb, entryPage uint32, indexKey []byte, keyLen int, resolve RowResolver) ([]byte, bool) {
	// 时序索引自己就是行数据所在, 不需要回表
	dataEntry := h.getDataEntry()
	curPageID := entryPage
	for curPageID != 0 {
		buf := make([]byte, page.IndexPageSize)
		if err := handle.ReadPage(curPageID, buf); err != nil {
			break
		}
		hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
		entries := buf[page.IndexHeadSize : page.IndexHeadSize+int(hdr.Count)*int(hdr.EntryLen)]
		entryLen := int(hdr.EntryLen)
		rowData, found := dataEntry.Search(handle, entries, entryLen, indexKey)
		if found {
			return rowData, true
		}
		curPageID = hdr.Hdr.NextPage
	}
	return nil, false
}

func (h *TimeIndexHandler) Delete(handle page.HandleDb, entryPage *uint32, indexKey []byte, pkID uint64, keyLen int) error {
	// 行号已知 → 按行号删除"这一行"。
	//
	// 同一时间戳下允许有多行 (同一时刻的多个测点/多次事件), 只按时间键删只会删掉
	// 该时刻的第一行, 其余行变成"看得见却删不掉"的幽灵数据。行号是每行唯一的,
	// 按它定位才能删到调用方指定的那一行。
	if pkID != 0 {
		dpID, ok := h.findDataPageByPK(handle, *entryPage, pkID)
		if !ok {
			return page.ErrNotFound
		}
		if !h.DeleteDataPageByPK(handle, dpID, pkIDBytes(pkID)) {
			return page.ErrNotFound
		}
		// 有意不回收"整页已删空"的 DataPage: 索引条目仍指向它, 立即还链会让该条目
		// 指向一张可能被别的索引复用的页 (全表扫描会把别人的行当成本表的行读出来)。
		// 空页留在原地无害, 之后往同一分段插入时会被继续复用。
		return nil
	}

	dataEntry := h.getDataEntry()
	curPageID := *entryPage
	for curPageID != 0 {
		buf := make([]byte, page.IndexPageSize)
		if err := handle.ReadPage(curPageID, buf); err != nil {
			return err
		}
		hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
		entries := buf[page.IndexHeadSize : page.IndexHeadSize+int(hdr.Count)*int(hdr.EntryLen)]
		entryLen := int(hdr.EntryLen)
		err := dataEntry.Delete(handle, entries, entryLen, indexKey)
		if err == nil {
			return nil
		}
		curPageID = hdr.Hdr.NextPage
	}
	return page.ErrNotFound
}

// IsRowStore 时序索引直接承载行数据
func (h *TimeIndexHandler) IsRowStore() bool { return true }

// ResolveRow 遍历索引项链定位各个 TimeDataPage, 按 pkID 取行
func (h *TimeIndexHandler) ResolveRow(handle page.HandleDb, entryPage uint32, pkID uint64, cache *[]DPRange) ([]byte, bool) {
	pkKeyBE := make([]byte, 8)
	binary.BigEndian.PutUint64(pkKeyBE, pkID)
	curPageID := entryPage
	for curPageID != 0 {
		buf := make([]byte, page.IndexPageSize)
		if err := handle.ReadPage(curPageID, buf); err != nil {
			break
		}
		hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
		entryLen := int(hdr.EntryLen)
		count := int(hdr.Count)
		if entryLen <= 0 || count == 0 {
			curPageID = hdr.Hdr.NextPage
			continue
		}
		entries := buf[page.IndexHeadSize : page.IndexHeadSize+count*entryLen]
		for i := 0; i < count; i++ {
			dpID := binary.BigEndian.Uint32(entries[i*entryLen+entryLen-4:])
			if dpID == 0 {
				continue
			}
			dp, err := datapage.Load(handle, dpID)
			if err != nil {
				continue
			}
			if row, ok := dp.FindRowByPK(pkKeyBE); ok {
				return row, true
			}
		}
		curPageID = hdr.Hdr.NextPage
	}
	return nil, false
}

func (h *TimeIndexHandler) Update(handle page.HandleDb, entryPage *uint32, oldKey, newKey, rowData []byte, pkID uint64, keyLen int) error {
	// 行号已知 → 按行号改写"这一行"; 时间键相同的多行因此能各自独立更新。
	if pkID != 0 {
		return h.updateByPK(handle, entryPage, oldKey, newKey, rowData, pkID)
	}

	dataEntry := h.getDataEntry()
	curPageID := *entryPage
	for curPageID != 0 {
		buf := make([]byte, page.IndexPageSize)
		if err := handle.ReadPage(curPageID, buf); err != nil {
			return err
		}
		hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
		entries := buf[page.IndexHeadSize : page.IndexHeadSize+int(hdr.Count)*int(hdr.EntryLen)]
		entryLen := int(hdr.EntryLen)
		err := dataEntry.Update(handle, entries, entryLen, oldKey, pkID, rowData)
		if err == nil {
			return nil
		}
		if errors.Is(err, page.ErrNotFound) {
			curPageID = hdr.Hdr.NextPage
			continue
		}
		return err
	}
	return page.ErrNotFound
}

// updateByPK 按行号精确定位并改写一行。
//
// 分三种情形:
//  1. 时间键没变 → 分段内原地改写; 行变长装不下时先在页内紧凑重建;
//  2. 时间键变了 → 行必须换到新时间段: 按行号删掉旧行, 再用同一个行号重新插入;
//  3. 页内重建仍装不下 → 同 2, 搬走重插 (不改变行号)。
func (h *TimeIndexHandler) updateByPK(handle page.HandleDb, entryPage *uint32, oldKey, newKey, rowData []byte, pkID uint64) error {
	if len(newKey) < 8 {
		// 时间索引的键必须是 8 字节时间戳; 取不到键说明行本身不可解析, 不能猜着改。
		return page.ErrNotFound
	}
	pkKeyBE := pkIDBytes(pkID)
	dpID, ok := h.findDataPageByPK(handle, *entryPage, pkID)
	if !ok {
		return page.ErrNotFound
	}

	storeRow := make([]byte, 8+len(rowData))
	copy(storeRow[:8], pkKeyBE)
	copy(storeRow[8:], rowData)

	move := func() error {
		if !h.DeleteDataPageByPK(handle, dpID, pkKeyBE) {
			return page.ErrNotFound
		}
		_, err := h.Insert(handle, entryPage, newKey, pkID, rowData, len(newKey))
		return err
	}

	if !sameTimeKey(oldKey, newKey) {
		// 先验顺序再删旧行: move 是"先删后插", 顺序不合法要在这里就拦下 ——
		// 否则旧行已删、新行被拒, 这一行就凭空消失了。
		if _, err := h.assertTimeOrder(handle, *entryPage, newKey); err != nil {
			return err
		}
		return move()
	}
	if h.UpdateDataPageByPK(handle, dpID, pkKeyBE, storeRow) {
		return nil
	}
	// 原地装不下 (行长变大): 页内重排, 把墓碑的碎片空间收回来。
	if dp, err := datapage.Load(handle, dpID); err == nil {
		if err := dp.RebuildWith(pkKeyBE, storeRow); err == nil {
			return nil
		}
	}
	// 整页都放不下 → 搬到新页, 行号保持不变。
	return move()
}

func (h *TimeIndexHandler) GetDataPageIDs(handle page.HandleDb, entryPage uint32) []uint32 {
	if entryPage == 0 {
		return nil
	}
	var ids []uint32
	curPageID := entryPage
	for curPageID != 0 {
		buf := make([]byte, page.IndexPageSize)
		if err := handle.ReadPage(curPageID, buf); err != nil {
			break
		}
		hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
		entryLen := int(hdr.EntryLen)
		count := int(hdr.Count)
		entries := buf[page.IndexHeadSize : page.IndexHeadSize+count*entryLen]
		// 时间索引条目格式: [BeginTime(8B) + LastTime(8B) + DataPageID(4B)]
		for i := 0; i < count; i++ {
			dpID := binary.BigEndian.Uint32(entries[i*entryLen+entryLen-4:])
			if dpID != 0 {
				ids = append(ids, dpID)
			}
		}
		curPageID = hdr.Hdr.NextPage
	}
	return ids
}

// CollectPageIDs 汇报时序索引当前占用的全部页: 索引条目链上的每一页, 加上每个
// 时间段指向的 DataPage (行真正落在这里)。
//
// 时序索引的页是"扁平 IndexHead 链", 既不是 B+ 节点也不是哈希桶, 所以自己枚举才准确。
// 不能指望 btree.CollectAllPageIDs: 它把页当 B+ 节点解析, 这里 IsLeaf=1 只会返回链首
// 一页 —— 单页可容 817 段, 实测插到 6 万行仍压在一页内, 所以链页尚未见泄漏; 但行数
// 继续增长 (每页约 400~500 段) 后超出部分就会成为不可达页。
// 数据页此前是靠 btree.GetDataPageIDs 顺带收回的: 它把 NextPage 当链指针, 又把每个
// 条目末 4 字节当 DataPageID, 恰好与 [BeginTime/LastTime/DataPageID] 的布局撞对。
// 这里改为显式调用自己的 GetDataPageIDs, 不再依赖这种巧合。
func (h *TimeIndexHandler) CollectPageIDs(handle page.HandleDb, entryPage uint32) []uint32 {
	if entryPage == 0 {
		return nil
	}
	var ids []uint32
	seen := make(map[uint32]bool)
	for curPageID := entryPage; curPageID != 0 && !seen[curPageID]; {
		seen[curPageID] = true
		ids = append(ids, curPageID)
		buf := make([]byte, page.IndexPageSize)
		if err := handle.ReadPage(curPageID, buf); err != nil {
			break
		}
		hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
		curPageID = hdr.Hdr.NextPage
	}
	for _, dpID := range h.GetDataPageIDs(handle, entryPage) {
		if dpID != 0 && !seen[dpID] {
			seen[dpID] = true
			ids = append(ids, dpID)
		}
	}
	return ids
}

func (h *TimeIndexHandler) ReadDataPageRows(handle page.HandleDb, dpID uint32) [][]byte {
	dp, err := datapage.Load(handle, dpID)
	if err != nil {
		return nil
	}
	return dp.GetAllRows()
}

func (h *TimeIndexHandler) getDataEntry() IDateEntry {
	handler, _ := Compose.GetEntryHandler(page.IndexTypeTime)
	return handler.DataEntry()
}

// GetDataPageIDs 已存在于上方 (L145)

// SearchDataPageByPK 在指定 DataPage 中按 pkID(LE) 查找行
func (h *TimeIndexHandler) SearchDataPageByPK(handle page.HandleDb, dpID uint32, pkKeyBE []byte) ([]byte, bool) {
	buf := make([]byte, page.IndexPageSize)
	if err := handle.ReadPage(dpID, buf); err != nil {
		return nil, false
	}
	// DataPage 格式: [page.DataPageHdr(40B)] + [len(4B) + rowData]...
	dh := page.DataHdrOf(buf)
	hdrSize := page.DataPageHdrSize
	count := dh.Count
	dataEnd := dh.UseSize
	keyPK := binary.BigEndian.Uint64(pkKeyBE[:8])
	off := hdrSize
	for i := uint32(0); i < count; i++ {
		if off+4 > dataEnd {
			break
		}
		rawLen := binary.LittleEndian.Uint32(buf[off:])
		deleted := rawLen&0x80000000 != 0
		rowLen := rawLen & 0x7FFFFFFF
		off += 4
		if off+rowLen > dataEnd {
			break
		}
		if !deleted && rowLen >= 8 {
			rowPK := binary.BigEndian.Uint64(buf[off : off+8])
			if rowPK == keyPK {
				result := make([]byte, rowLen)
				copy(result, buf[off:off+rowLen])
				return result, true
			}
		}
		off += rowLen
	}
	return nil, false
}

// DeleteDataPageByPK 在指定 DataPage 中按 pkID(LE) 软删除行, 返回是否找到
func (h *TimeIndexHandler) DeleteDataPageByPK(handle page.HandleDb, dpID uint32, pkKeyBE []byte) bool {
	buf := make([]byte, page.IndexPageSize)
	if err := handle.ReadPage(dpID, buf); err != nil {
		return false
	}
	dh := page.DataHdrOf(buf)
	hdrSize := page.DataPageHdrSize
	count := dh.Count
	dataEnd := dh.UseSize
	keyPK := binary.BigEndian.Uint64(pkKeyBE[:8])
	off := hdrSize
	for i := uint32(0); i < count; i++ {
		if off+4 > dataEnd {
			break
		}
		rawLen := binary.LittleEndian.Uint32(buf[off:])
		deleted := rawLen&0x80000000 != 0
		rowLen := rawLen & 0x7FFFFFFF
		off += 4
		if off+rowLen > dataEnd {
			break
		}
		if !deleted && rowLen >= 8 {
			rowPK := binary.BigEndian.Uint64(buf[off : off+8])
			if rowPK == keyPK {
				binary.LittleEndian.PutUint32(buf[off-4:], rowLen|0x80000000)
				_ = handle.SavePage(dpID, buf)
				return true
			}
		}
		off += rowLen
	}
	return false
}

// UpdateDataPageByPK 在指定 DataPage 中按 pkID(LE) 原地更新行, 返回是否找到
func (h *TimeIndexHandler) UpdateDataPageByPK(handle page.HandleDb, dpID uint32, pkKeyBE []byte, newStoreRow []byte) bool {
	buf := make([]byte, page.IndexPageSize)
	if err := handle.ReadPage(dpID, buf); err != nil {
		return false
	}
	dh := page.DataHdrOf(buf)
	hdrSize := page.DataPageHdrSize
	count := dh.Count
	dataEnd := dh.UseSize
	keyPK := binary.BigEndian.Uint64(pkKeyBE[:8])
	off := hdrSize
	for i := uint32(0); i < count; i++ {
		if off+4 > dataEnd {
			break
		}
		rawLen := binary.LittleEndian.Uint32(buf[off:])
		deleted := rawLen&0x80000000 != 0
		rowLen := rawLen & 0x7FFFFFFF
		off += 4
		if off+rowLen > dataEnd {
			break
		}
		if !deleted && rowLen >= 8 {
			rowPK := binary.BigEndian.Uint64(buf[off : off+8])
			if rowPK == keyPK {
				if uint32(len(newStoreRow)) <= rowLen {
					copy(buf[off:], newStoreRow)
					for j := uint32(len(newStoreRow)); j < rowLen; j++ {
						buf[off+j] = 0
					}
					_ = handle.SavePage(dpID, buf)
					return true
				}
				return false // row too large
			}
		}
		off += rowLen
	}
	return false
}

// RangeScan 时序索引范围扫描
// startKey/endKey: BigEndian 编码的时间戳
// 遍历 IndexPage 条目链 对时间范围与查询范围重叠的条目加载 DataPage 并回调每行
func (h *TimeIndexHandler) RangeScan(handle page.HandleDb, entryPage uint32,
	startKey, endKey []byte, startInclusive, endInclusive bool, reverse bool,
	callback func(row []byte) bool) error {
	if entryPage == 0 {
		return nil
	}
	// 收集所有索引页数据
	type pageInfo struct {
		entries  []byte
		entryLen int
		count    int
	}
	var pages []pageInfo
	curPageID := entryPage
	for curPageID != 0 {
		buf := make([]byte, page.IndexPageSize)
		if err := handle.ReadPage(curPageID, buf); err != nil {
			break
		}
		hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
		entryLen := int(hdr.EntryLen)
		count := int(hdr.Count)
		entries := make([]byte, count*entryLen)
		copy(entries, buf[page.IndexHeadSize:page.IndexHeadSize+count*entryLen])
		pages = append(pages, pageInfo{entries: entries, entryLen: entryLen, count: count})
		curPageID = hdr.Hdr.NextPage
	}

	// 确定遍历顺序
	pageStart, pageEnd, pageStep := 0, len(pages)-1, 1
	if reverse {
		pageStart, pageEnd, pageStep = len(pages)-1, 0, -1
	}

	visited := make(map[uint32]bool)
	for p := pageStart; ; p += pageStep {
		if pageStep > 0 && p > pageEnd {
			break
		}
		if pageStep < 0 && p < pageEnd {
			break
		}
		pg := &pages[p]
		if pg.entryLen < IndexTimeEntryLen {
			continue // 防御: 旧库/损坏页的条目长度不足 20B, 跳过以免越界
		}
		entryStart, entryEnd, entryStep := 0, pg.count-1, 1
		if reverse {
			entryStart, entryEnd, entryStep = pg.count-1, 0, -1
		}
		for i := entryStart; ; i += entryStep {
			if entryStep > 0 && i > entryEnd {
				break
			}
			if entryStep < 0 && i < entryEnd {
				break
			}
			off := i * pg.entryLen
			beginTime := pg.entries[off : off+8]
			lastTime := pg.entries[off+8 : off+16]
			dpID := binary.BigEndian.Uint32(pg.entries[off+pg.entryLen-4:])
			if dpID == 0 || visited[dpID] {
				continue
			}
			if endKey != nil {
				cmp := tiCmp(beginTime, endKey)
				if cmp > 0 || (cmp == 0 && !endInclusive) {
					continue
				}
			}
			if startKey != nil {
				cmp := tiCmp(lastTime, startKey)
				if cmp < 0 || (cmp == 0 && !startInclusive) {
					continue
				}
			}
			visited[dpID] = true
			dp, dpErr := datapage.Load(handle, dpID)
			if dpErr != nil {
				continue
			}
			rows := dp.GetAllRows()
			if !reverse {
				for _, row := range rows {
					if !timeRowInRange(row, startKey, endKey, startInclusive, endInclusive) {
						continue
					}
					if !callback(row) {
						return nil
					}
				}
			} else {
				for k := len(rows) - 1; k >= 0; k-- {
					if !timeRowInRange(rows[k], startKey, endKey, startInclusive, endInclusive) {
						continue
					}
					if !callback(rows[k]) {
						return nil
					}
				}
			}
		}
	}
	return nil
}

// timeRowInRange 判断时序数据行的时间是否落在 [startKey, endKey] 区间内。
// 行内布局: [pkID(8B)] + [schemaVer(1B)] + 字段..., 时间键位于偏移 9(8B BE)。
// 由于分段 DataPage 覆盖一段时间跨度, 段内可能混有区间外的点, 必须逐行过滤,
// 否则范围查询会把整段的点都返回 (含越界点)。
func timeRowInRange(row, startKey, endKey []byte, startInclusive, endInclusive bool) bool {
	const tsOff = 9
	if len(row) < tsOff+8 {
		// 行过短无法解析时间, 保守返回 true, 交由上层判断
		return true
	}
	if startKey == nil && endKey == nil {
		return true
	}
	ts := row[tsOff : tsOff+8]
	if startKey != nil {
		cmp := tiCmp(ts, startKey)
		if cmp < 0 || (cmp == 0 && !startInclusive) {
			return false
		}
	}
	if endKey != nil {
		cmp := tiCmp(ts, endKey)
		if cmp > 0 || (cmp == 0 && !endInclusive) {
			return false
		}
	}
	return true
}

// tiCmp 字节切片比较
func tiCmp(a, b []byte) int {
	ml := len(a)
	if len(b) < ml {
		ml = len(b)
	}
	for i := 0; i < ml; i++ {
		if a[i] < b[i] {
			return -1
		}
		if a[i] > b[i] {
			return 1
		}
	}
	if len(a) < len(b) {
		return -1
	}
	if len(a) > len(b) {
		return 1
	}
	return 0
}