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

package timeindex

import (
	"git.ycl.pub/yl32940/TsDB2/DB/core/index/datapage"
	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"encoding/binary"
	"errors"
)

// Row 时序数据行
type Row struct {
	TS   int64  // 时间戳
	Flag byte   // 标记/状态
	Data []byte // 真实数据
}

type TimeEntry struct {
	Hdr       page.PageHdr
	BeginTime int64  // 本页最小时间
	EndTime   int64  // 本页最大时间
	Count     uint32 // 本页记录总数
}

type TimeDataPage struct {
	TimeEntry TimeEntry
	Rows      []Row
}

// ======================== IDateEntry 接口实现 ========================

func (t *TimeDataPage) Insert(handle page.HandleDb, indexEntries []byte, entryLen int, count int, key []byte, pkID uint64, rowData []byte) error {
	dataPageID, entryIdx := findDataPageByTime(indexEntries, entryLen, count, key)

	var dp *datapage.DataPage
	if dataPageID == 0 {
		dp = datapage.New(handle)
	} else {
		var err error
		dp, err = datapage.Load(handle, dataPageID)
		if err != nil {
			return err
		}
	}

	storeRow := make([]byte, 8+len(rowData))
	binary.BigEndian.PutUint64(storeRow[:8], pkID)
	copy(storeRow[8:], rowData)

	// 用"按 pkID 有序插入"而不是追加: FindRowByPK 依赖页内行按 pkID 升序, 一旦遇到
	// 比目标更大的行就提前判定"不存在"。正常写入时 pkID 单增, 有序插入等价于追加;
	// 但"改时间键把行搬到别的时间段"时, 搬过去的行号可能小于目标页里已有的行号 ——
	// 追加会打乱这页的顺序, 之后整页的行都按行号查不到 (二级索引回表也跟着失效)。
	if !dp.InsertRowSorted(storeRow) {
		// 开新页之前先按配置就地整理一次: 墓碑占比达到 CompactDeadRatio 时把墓碑空间还给
		// 活行, 常常就够放下这一行 —— 少开一页, 也少一个时间段分段。
		// 整理成功且重试写入成功 → 留在本页 (末尾统一 Save, 页 ID 不变即"追加进已有分段");
		// 否则 (没墓碑可回收 / 整理了也装不下) 新开一页, 由调用方为它新增一个时间段条目。
		need := uint32(len(storeRow)) + 4
		if !dp.CompactForInsert(need) || !dp.InsertRowSorted(storeRow) {
			dp.Save()
			dp = datapage.New(handle)
			if !dp.InsertRowSorted(storeRow) {
				return ErrRowTooLarge
			}
		}
	}
	dp.Save()

	if dp.PageID() != dataPageID {
		if count*entryLen+4 <= len(indexEntries) {
			binary.BigEndian.PutUint32(indexEntries[count*entryLen:], dp.PageID())
		}
		return ErrNeedNewEntry
	}

	// 追加进已有分段: 按需扩展该条目的 LastTime, 使时间范围扫描不遗漏本分段。
	if entryIdx >= 0 && entryLen >= 20 {
		lastOff := entryIdx*entryLen + 8
		if lastOff+8 <= len(indexEntries) {
			newLast := int64(binary.BigEndian.Uint64(key[:8]))
			if newLast > int64(binary.BigEndian.Uint64(indexEntries[lastOff:lastOff+8])) {
				binary.BigEndian.PutUint64(indexEntries[lastOff:lastOff+8], uint64(newLast))
			}
		}
	}
	return nil
}

func (t *TimeDataPage) Search(handle page.HandleDb, indexEntries []byte, entryLen int, key []byte) ([]byte, bool) {
	n := len(indexEntries) / entryLen
	dataPageID, _ := findDataPageByTime(indexEntries, entryLen, n, key)
	if dataPageID == 0 {
		return nil, false
	}
	dp, err := datapage.Load(handle, dataPageID)
	if err != nil {
		return nil, false
	}
	return dp.FindRowByIndexKey(key, 9)
}

// CreateDataPage 为新的时间段建第一个 DataPage, 并把首行写入。
//
// 旧实现把 rowData[:8] 当成行号写进了页首 —— rowData 是 [schemaVer + 字段...],
// 前 8 字节是"schemaVer + 时间戳的高 7 字节", 根本不是行号。于是新建分段的
// 第一行带了一个极大的假行号, FindRowByPK 一比较就提前结束, 整页行号全部不可达
// (按行号读/改/删、二级索引回表统统失效)。这里直接用调用方给的真实 pkID。
func (t *TimeDataPage) CreateDataPage(handle page.HandleDb, key []byte, pkID uint64, rowData []byte) (uint32, error) {
	dp := datapage.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
}

func (t *TimeDataPage) Update(handle page.HandleDb, indexEntries []byte, entryLen int, key []byte, pkID uint64, rowData []byte) error {
	n := len(indexEntries) / entryLen
	dataPageID, _ := findDataPageByTime(indexEntries, entryLen, n, key)
	if dataPageID == 0 {
		return page.ErrNotFound
	}
	dp, err := datapage.Load(handle, dataPageID)
	if err != nil {
		return err
	}
	storeRow := make([]byte, 8+len(rowData))
	binary.BigEndian.PutUint64(storeRow[:8], pkID)
	copy(storeRow[8:], rowData)
	// 时序行内布局: [pkID(8B)] + [schemaVer(1B)] + 字段..., 时间键位于偏移 9
	if err := dp.FindAndUpdateByIndexKey(key, 9, storeRow); err != nil {
		// schema 演进 (AddField) 后旧行变长, 原地覆盖装不下 → 页内重建支持变长
		return dp.RebuildWithIndexKey(key, 9, storeRow)
	}
	return nil
}

func (t *TimeDataPage) Delete(handle page.HandleDb, indexEntries []byte, entryLen int, key []byte) error {
	n := len(indexEntries) / entryLen
	dataPageID, _ := findDataPageByTime(indexEntries, entryLen, n, key)
	if dataPageID == 0 {
		return page.ErrNotFound
	}
	dp, err := datapage.Load(handle, dataPageID)
	if err != nil {
		return err
	}
	// 时序行内布局: [pkID(8B)] + [schemaVer(1B)] + 字段..., 时间键位于偏移 9
	deleted := dp.DeleteByIndexKey(key, 9)
	if !deleted {
		return page.ErrNotFound
	}
	if dp.AllDeleted() {
		// 延迟还链: 时序条目仍映射该时间段, 立即还链会让复用后的页被
		// 扫描/回表读到 (与 time_handler.go 按 pkID 删除路径的约定对齐)。
		handle.DeferFreeDataPage(dp.PageID())
	} else {
		dp.Save()
	}
	return nil
}

// ======================== 内部辅助 ========================

// ErrNeedNewEntry 当前分段页已满, 调用方需要为这个 DataPage 新增一个时间段条目。
// 导出为哨兵错误: 上层用 errors.Is 判定, 不再靠匹配错误字符串。
var ErrNeedNewEntry = errors.New("need new index entry")

// ErrRowTooLarge 行数据太大，需要堆页溢出处理
var ErrRowTooLarge = errors.New("row too large, need heap overflow")

// findDataPageByTime 二分查找 BeginTime <= key 的最后一个分段,
// 返回 (DataPageID, 条目下标)。未命中时返回 (0, -1)。
func findDataPageByTime(entries []byte, entryLen int, n int, key []byte) (uint32, int) {
	if n == 0 {
		return 0, -1
	}
	// 时间索引条目格式: BeginTime(8) + LastTime(8) + PageID(4)
	result := -1
	left, right := 0, n-1
	keyTime := int64(binary.BigEndian.Uint64(key[:8]))
	for left <= right {
		mid := (left + right) / 2
		entryBegin := int64(binary.BigEndian.Uint64(entries[mid*entryLen:]))
		if entryBegin <= keyTime {
			result = mid
			left = mid + 1
		} else {
			right = mid - 1
		}
	}
	if result < 0 {
		return 0, -1
	}
	// PageID 在条目最后 4 字节
	off := result*entryLen + entryLen - 4
	return binary.BigEndian.Uint32(entries[off:]), result
}