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

package timeindex

import (
	"TsDB2/DB/core/index/datapage"
	"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 := 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
		}
	}

	if dp.IsFull(uint32(8 + len(rowData))) {
		dp.Save()
		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()

	if dp.PageID() != dataPageID {
		if count*entryLen+4 <= len(indexEntries) {
			binary.BigEndian.PutUint32(indexEntries[count*entryLen:], dp.PageID())
		}
		return errNeedNewEntry
	}
	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)
}

func (t *TimeDataPage) CreateDataPage(handle page.HandleDb, key []byte, rowData []byte) (uint32, error) {
	dp := datapage.New(handle)
	storeRow := make([]byte, 8+len(rowData))
	if len(rowData) >= 8 {
		copy(storeRow[:8], rowData[:8])
	}
	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 errors.New("row not found")
	}
	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)
	return dp.FindAndUpdateByIndexKey(key, 8, storeRow)
}

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 errors.New("row not found")
	}
	dp, err := datapage.Load(handle, dataPageID)
	if err != nil {
		return err
	}
	deleted := dp.DeleteByIndexKey(key, 8)
	if !deleted {
		return errors.New("row not found")
	}
	if dp.AllDeleted() {
		if handle.FreePage != nil {
			handle.FreePage(dp.PageID())
		}
	} else {
		dp.Save()
	}
	return nil
}

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

var errNeedNewEntry = errors.New("need new index entry")

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

func findDataPageByTime(entries []byte, entryLen int, n int, key []byte) uint32 {
	if n == 0 {
		return 0
	}
	// 时间索引条目格式: BeginTime(8) + LastTime(8) + PageID(4)
	result := 0
	left, right := 0, n-1
	for left <= right {
		mid := (left + right) / 2
		entryBegin := int64(binary.BigEndian.Uint64(entries[mid*entryLen:]))
		keyTime := int64(binary.BigEndian.Uint64(key[:8]))
		if entryBegin <= keyTime {
			result = mid
			left = mid + 1
		} else {
			right = mid - 1
		}
	}
	// PageID 在条目最后 4 字节
	off := result*entryLen + entryLen - 4
	return binary.BigEndian.Uint32(entries[off:])
}