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

package datapage

import (
	"TsDB2/DB/core/page"
	"encoding/binary"
	"errors"
	"sync"
	"unsafe"
)

// PageSize 索引页大小 (避免导入 index 包造成循环依赖)
const PageSize = 16384

var (
	errNotFound    = errors.New("key not found")
	errRowTooLarge = errors.New("row too large, need heap overflow")
)

// =============================================================================
// 统一 DataPage — 所有需要独立数据页的索引类型共享
//
// 磁盘布局:
//   [PageHdr(18B) + Count(4B) + Min(8B) + Max(8B)]  = 38B header
//   [len(4B) + rowData] ...                          = 行数据区
//
// len 最高位 (0x80000000) 为软删除标志
// Min/Max 存储 rowData 前 8 字节的最小/最大值 (BigEndian 比较)
// =============================================================================

// HdrSize 数据页头部大小: PageHdr(18) + Count(4) + Min(8) + Max(8) = 38
const HdrSize = uint32(unsafe.Sizeof(page.PageHdr{})) + 4 + 8 + 8

// 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.PageHdr)(unsafe.Pointer(&buf[0]))
	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.PageHdr)(unsafe.Pointer(&buf[0]))
	cnt := *(*uint32)(unsafe.Pointer(&buf[unsafe.Sizeof(page.PageHdr{})]))
	return &DataPage{
		handle: handle, pageID: pageID, buf: buf,
		count: cnt, dataEnd: hdr.UseSize,
	}, nil
}

// Save 持久化到磁盘
func (dp *DataPage) Save() {
	(*page.PageHdr)(unsafe.Pointer(&dp.buf[0])).UseSize = dp.dataEnd
	*(*uint32)(unsafe.Pointer(&dp.buf[unsafe.Sizeof(page.PageHdr{})])) = 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 维护 ========================

// minOffset Min 字段在 buf 中的偏移
func minOffset() uint32 { return uint32(unsafe.Sizeof(page.PageHdr{})) + 4 }

// maxOffset Max 字段在 buf 中的偏移
func maxOffset() uint32 { return uint32(unsafe.Sizeof(page.PageHdr{})) + 4 + 8 }

// getMin 读取当前 Min 值
func (dp *DataPage) getMin() uint64 {
	return binary.BigEndian.Uint64(dp.buf[minOffset():])
}

// getMax 读取当前 Max 值
func (dp *DataPage) getMax() uint64 {
	return binary.BigEndian.Uint64(dp.buf[maxOffset():])
}

// 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])
	if dp.count <= 1 {
		// 第一行 (或刚从 0 变 1), 直接设置
		binary.BigEndian.PutUint64(dp.buf[minOffset():], val)
		binary.BigEndian.PutUint64(dp.buf[maxOffset():], val)
		return
	}
	curMin := dp.getMin()
	curMax := dp.getMax()
	if val < curMin {
		binary.BigEndian.PutUint64(dp.buf[minOffset():], val)
	}
	if val > curMax {
		binary.BigEndian.PutUint64(dp.buf[maxOffset():], val)
	}
}

// ======================== 行操作 ========================

// 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)
}

// FindRowByPK 按 pkID 查找 (pkKeyBE 为 BigEndian, 匹配 storedRow[0:8])
// 优化: 行按 pkID 顺序存储, 找到 pkID > target 时提前终止
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
}

// 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 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 errNotFound
}

// 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
}

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

// 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
	}
	// 时序索引: 时间戳在 rowData 前 8 字节, 即 storedRow 偏移 8
	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 errors.New("row not found")
	}
	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, 8, 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 errors.New("row not found")
	}
	dp, err := Load(handle, dpID)
	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
}

func (d *IDateEntryDataPage) CreateDataPage(handle page.HandleDb, key []byte, rowData []byte) (uint32, error) {
	dp := 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
}