TsDB2/DB/Table.go

package DB

import (
	"TsDB2/DB/page"
	"TsDB2/DB/types"
	"TsDB2/DB/util"
	"encoding/binary"
	"strings"
	"unsafe"
)

type RtFieldOffset struct {
	PageID uint32 // 页号
	Offset uint16 // 开始偏移
}

type FieldOffset struct {
	PageID      uint32 // 页号
	StartOffset uint16 // 开始偏移
	EndOffset   uint16 // 相对开始偏移
}
type TableInfo struct {
	Name             [TableNameSize]byte
	Flag             byte   // 表的状态是否被删除
	FieldCount       byte   // 当前字段的总数量（包含已删除，永不减少）byte=255上限
	ActiveFieldCount byte   // 实际活跃字段数量（不包含已删除）
	TotalRows        uint64 // 当前表行的总数量

	FirstHeap uint32 // 堆链首页
	LastHeap  uint32 // 堆链尾页

	FirstData uint32 // 数据链首页
	LastData  uint32 // 数据链尾

	RowSize           uint16 // 单条记录大小（字节）
	FieldAllocateSize uint16 // 当前列总分配大小

	IndexCount   byte   // 这张表有几个索引
	FirstIndexID uint32 // 本表第一个索引的元数据位置
	LastIndexID  uint32 // 本表最后一个索引的元数据位置

	FirstOverflow uint32 // 字段溢出页首页（0表示无溢出）
	LastOverflow  uint32 // 字段溢出页尾页
}

// ReadFieldsChain 读取字段链（支持溢出页）
func (info *TableInfo) ReadFieldsChain(handle page.HandleDb, curPageID uint32, startOff uint16) ([]types.FieldInfo, FieldOffset) {
	var fields []types.FieldInfo
	total := int(info.FieldCount)
	readCount := 0
	readBlock := func(buf []byte, blockStart uint16) (*RtFieldOffset, uint32) {
		blockCount := int(buf[blockStart])
		if blockCount == 0 {
			return nil, 1
		}

		off := uint32(blockStart + 1)

		for i := 0; i < blockCount; i++ {
			typ := buf[off]
			nameLen := buf[off+1]
			name := make([]byte, nameLen)
			copy(name, buf[off+2:off+2+uint32(nameLen)])

			fields = append(fields, types.FieldInfo{
				FieldType: types.FieldType(typ),
				FieldName: name,
			})
			off += 2 + uint32(nameLen)
		}

		readCount += blockCount
		if readCount < total {
			var filedOffset = (*RtFieldOffset)(unsafe.Pointer(&buf[off]))
			return filedOffset, off - uint32(blockStart)
		}
		return nil, off - uint32(blockStart)
	}
	mainBuf := make([]byte, PageSize)
	_ = handle.ReadPage(curPageID, mainBuf)
	fieldOffset, offset := readBlock(mainBuf, startOff)

	if fieldOffset == nil {
		return fields, FieldOffset{
			PageID:      curPageID,
			StartOffset: startOff,
			EndOffset:   uint16(offset),
		}
	}
	currentPage := fieldOffset.PageID
	currentOffset := uint32(fieldOffset.Offset)
	var curFieldOffset = fieldOffset
	for currentPage != 0 && readCount < total {
		hp, err := LoadHeapPage(handle, currentPage)
		if err != nil {
			break
		}

		fieldOffset, currentOffset = readBlock(hp.Get(), fieldOffset.Offset)
		if readCount == total {
			break
		}
		currentPage = fieldOffset.PageID
		curFieldOffset = fieldOffset
	}

	return fields, FieldOffset{
		PageID:      curFieldOffset.PageID,
		StartOffset: curFieldOffset.Offset,
		EndOffset:   uint16(currentOffset),
	}
}

type Table struct {
	info             *TableInfo
	Field            []types.FieldInfo
	handle           page.HandleDb
	indexMeta        *IndexMetaPage
	tablePageID      uint32      // 表ID
	tableStartOffset uint32      // 表所在页的偏移
	curOffset        FieldOffset // 当前Field可以写的偏移位置
}

func (table *Table) CreateIndex(idxName string, typ page.IndexType, fields []byte) (*IndexMeta, error) {
	return table.indexMeta.CreateIndex(idxName, typ, fields)
}
func (table *Table) DropIndex(idx string) error { return table.indexMeta.DropIndex(idx) }

// findFieldIndex 查找字段索引
func (table *Table) findFieldIndex(name string) int {
	for i, field := range table.Field {
		if string(field.FieldName) == name {
			return i
		}
	}
	return -1
}

// calcFieldOffset 计算字段在行中的偏移
func (table *Table) calcFieldOffset(fieldIdx int) uint16 {
	var offset uint16 = 0
	for i := 0; i < fieldIdx; i++ {
		offset += types.GetSize(table.Field[i].FieldType)
	}
	return offset
}

// compareValue 比较字段值是否相等
func (table *Table) compareValue(row []byte, offset uint16, typ types.FieldType, value interface{}) bool {
	switch typ {
	case types.FieldType_1byte:
		return row[offset] == value.(byte)
	case types.FieldType_2byte:
		stored := binary.BigEndian.Uint16(row[offset:])
		return stored == uint16(value.(int16))
	case types.FieldType_4byte:
		stored := binary.BigEndian.Uint32(row[offset:])
		return stored == uint32(value.(int32))
	case types.FieldType_8byte:
		stored := binary.BigEndian.Uint64(row[offset:])
		return stored == value.(uint64)
	case types.FieldType_256byte:
		stored := strings.TrimRight(string(row[offset:offset+256]), "\x00")
		return stored == string(value.([]byte))
	}
	return false
}

// compareGreater 比较字段值是否大于给定值
func (table *Table) compareGreater(row []byte, offset uint16, typ types.FieldType, value interface{}) bool {
	switch typ {
	case types.FieldType_8byte:
		stored := binary.BigEndian.Uint64(row[offset:])
		return stored > value.(uint64)
	case types.FieldType_4byte:
		stored := binary.BigEndian.Uint32(row[offset:])
		return stored > uint32(value.(int32))
	}
	return false
}

// compareLess 比较字段值是否小于给定值
func (table *Table) compareLess(row []byte, offset uint16, typ types.FieldType, value interface{}) bool {
	switch typ {
	case types.FieldType_8byte:
		stored := binary.BigEndian.Uint64(row[offset:])
		return stored < value.(uint64)
	}
	return false
}

// compareGreaterOrEqual 比较字段值是否大于等于给定值
func (table *Table) compareGreaterOrEqual(row []byte, offset uint16, typ types.FieldType, value interface{}) bool {
	return table.compareGreater(row, offset, typ, value) || table.compareValue(row, offset, typ, value)
}

// compareLessOrEqual 比较字段值是否小于等于给定值
func (table *Table) compareLessOrEqual(row []byte, offset uint16, typ types.FieldType, value interface{}) bool {
	return table.compareLess(row, offset, typ, value) || table.compareValue(row, offset, typ, value)
}

func (table *Table) GetFieldCount() byte          { return table.info.FieldCount }
func (table *Table) GetActiveFieldCount() byte    { return table.info.ActiveFieldCount }
func (table *Table) GetFieldAllocateSize() uint16 { return table.info.FieldAllocateSize }
func (table *Table) GetFirstOverflow() uint32     { return table.info.FirstOverflow }
func (table *Table) GetLastOverflow() uint32      { return table.info.LastOverflow }
func (table *Table) GetFirstData() uint32         { return table.info.FirstData }

func (table *Table) GetFields() []types.FieldInfo {
	return table.Field
}

func (table *Table) findFieldByName(name []byte) *types.FieldInfo {
	target := util.BytesToString(name)
	for i := range table.Field {
		if util.BytesToString(table.Field[i].FieldName) == target {
			return &table.Field[i]
		}
	}
	return nil
}

func (table *Table) writeField(buf []byte, f types.FieldInfo) (*RtFieldOffset, bool) {
	allocSize := int(table.info.FieldAllocateSize)
	// 🔥 核心修复：动态获取当前缓冲区的真实长度（自动适配主页/溢出块）
	bufLen := len(buf)
	fieldSize := 2 + len(f.FieldName)
	// 基础校验：块起始位置不能超出缓冲区
	if int(table.curOffset.StartOffset) >= bufLen {
		return nil, false
	}

	if allocSize-6 < int(table.curOffset.EndOffset)+fieldSize {
		jump := (*RtFieldOffset)(unsafe.Pointer(&buf[table.curOffset.EndOffset+table.curOffset.StartOffset]))
		return jump, false
	}
	blockStart := table.curOffset.StartOffset
	blockFieldCount := buf[blockStart]
	off := table.curOffset.EndOffset + table.curOffset.StartOffset
	buf[off] = byte(f.FieldType)
	off++
	buf[off] = uint8(len(f.FieldName))
	off++
	copy(buf[off:], f.FieldName)
	// 字段计数+1
	buf[blockStart] = blockFieldCount + 1
	table.curOffset.EndOffset += uint16(len(f.FieldName)) + 2
	return nil, true
}

func (table *Table) AddField(newField types.FieldInfo) error {
	// 1. 基础校验 完整保留
	if int(table.info.FieldCount)+1 >= MaxFieldSize {
		return ErrInvalidFieldCount
	}
	if table.findFieldByName(newField.FieldName) != nil {
		return ErrFieldExists
	}

	allocSize := uint32(table.info.FieldAllocateSize)

	targetPageID := table.curOffset.PageID
	mainBlockEnd := uint32(0)

	var (
		ok          bool
		fieldOffset *RtFieldOffset
		heap        *HeapPage
		err         error
	)
	if heap, err = LoadHeapPage(table.handle, targetPageID); err != nil {
		return err
	}
	mainBuf := heap.Get()
	for {

		if fieldOffset, ok = table.writeField(mainBuf, newField); ok {
			_ = table.handle.SavePage(targetPageID, mainBuf)
			table.addFieldToMem(newField)
			return nil
		}
		if heap.Info.Hdr.Type != page.DataOverflowPage {
			if table.info.LastOverflow == 0 {
				heap = NewHeapPage(table.handle, page.DataOverflowPage)
				if table.info.FirstOverflow == 0 {
					table.info.FirstOverflow = heap.Info.Hdr.PageID
				}
				table.info.LastOverflow = heap.Info.Hdr.PageID
			}
			if heap == nil {
				if fieldOffset.PageID > 0 {
					heap, err = LoadHeapPage(table.handle, fieldOffset.PageID)
					if err != nil {
						return err
					}
				} else {
					heap, err = LoadHeapPage(table.handle, table.info.LastOverflow)
					if err != nil {
						return err
					}
				}
			}
		}
		// 申请堆页空间
		targetPageID, mainBlockEnd, err = heap.Alloc(allocSize)
		if err != nil {
			return err
		}
		table.writeJump(mainBuf, targetPageID, mainBlockEnd)
		_ = table.handle.SavePage(table.curOffset.PageID, mainBuf)

		mainBuf = heap.Get()
		table.curOffset.PageID = targetPageID
		table.curOffset.StartOffset = uint16(mainBlockEnd)
		table.curOffset.EndOffset = 1 //让出长度1个字节
		table.info.LastOverflow = heap.Info.Hdr.PageID
		table.saveTableInfo()
	}
}

//// writeJump：在【字段结束位置】写入6字节跳转，而不是固定块尾
//// 完全匹配你 ReadFieldsChain 的读取方式
//func (table *Table) writeJump(buf []byte, blockStart uint32, nextPageID uint32, nextOff uint32) {
//	// 1. 读取块首字段数量
//	fieldCount := buf[blockStart]
//	off := blockStart + 1
//
//	// 2. 跳过所有字段 → 找到【字段结束的位置】（你读取时的偏移）
//	for i := 0; i < int(fieldCount); i++ {
//		nameLen := buf[off+1]
//		off += 2 + uint32(nameLen)
//	}
//	// 3. 【核心】在字段结束位置写入6字节跳转（你读的就是这个位置！）
//	jump := FieldOffset{
//		PageID:      nextPageID,
//		StartOffset: uint16(nextOff), // 严格转uint16，6字节总大小
//	}
//	// 直接内存写入，和你读取的代码完全对齐
//	*(*FieldOffset)(unsafe.Pointer(&buf[off])) = jump
//}

func (table *Table) writeJump(buf []byte, nextPageID uint32, nextOff uint32) {
	offset := table.curOffset.EndOffset + table.curOffset.StartOffset
	jump := FieldOffset{
		PageID:      nextPageID,
		StartOffset: uint16(nextOff), // 严格转uint16，6字节总大小
	}
	// 直接内存写入，和你读取的代码完全对齐
	*(*FieldOffset)(unsafe.Pointer(&buf[offset])) = jump
}

func (table *Table) addFieldToMem(f types.FieldInfo) {
	table.Field = append(table.Field, f)
	table.info.FieldCount++
	table.info.ActiveFieldCount++
	table.saveTableInfo()
}

func (table *Table) saveTableInfo() {
	buf := make([]byte, PageSize)
	if err := table.handle.ReadPage(table.tablePageID, buf); err != nil {
		return
	}
	info := (*TableInfo)(unsafe.Pointer(&buf[table.tableStartOffset]))
	info.FirstOverflow = table.info.FirstOverflow
	info.LastOverflow = table.info.LastOverflow
	info.FieldCount = table.info.FieldCount
	info.ActiveFieldCount = table.info.ActiveFieldCount
	table.handle.SavePage(table.tablePageID, buf)
}

// =============================================================================
// 删除字段（软删除）
// =============================================================================
func (table *Table) DropField(fieldName string) error {
	allocSize := uint32(table.info.FieldAllocateSize)
	infoSize := uint32(unsafe.Sizeof(table.info))
	fieldStart := table.tableStartOffset + infoSize
	hdrSize := uint32(unsafe.Sizeof(page.PageHdr{}))

	buf := make([]byte, PageSize)
	if err := table.handle.ReadPage(table.tablePageID, buf); err != nil {
		return err
	}

	searchBlock := func(buf []byte, start uint32) bool {
		cnt := buf[start]
		off := start + 1
		for i := byte(0); i < cnt; i++ {
			if off+2 > PageSize {
				return false
			}
			ft := types.FieldType(buf[off])
			nl := buf[off+1]
			ds := 2 + uint32(nl)
			if off+ds > start+allocSize {
				return false
			}
			if util.IsHighBitSet(byte(ft)) {
				off += ds
				continue
			}
			name := util.BytesToString(buf[off+2 : off+2+uint32(nl)])
			if name == fieldName {
				buf[off] |= 0x80
				return true
			}
			off += ds
		}
		return false
	}

	if searchBlock(buf, fieldStart) {
		table.handle.SavePage(table.tablePageID, buf)
		table.info.ActiveFieldCount--
		table.saveTableInfo()
		return nil
	}

	curPage := table.info.FirstOverflow
	for curPage != 0 {
		buf := make([]byte, PageSize)
		if err := table.handle.ReadPage(curPage, buf); err != nil {
			return err
		}
		if searchBlock(buf, hdrSize) {
			table.handle.SavePage(curPage, buf)
			table.info.ActiveFieldCount--
			table.saveTableInfo()
			return nil
		}
		curPage = (*page.PageHdr)(unsafe.Pointer(&buf[0])).NextPage
	}

	return ErrFieldNotFound
}

// ======================== 行数据序列化/反序列化 ========================

// serializeRow 将 FieldValue 切片序列化为行数据字节 (不含 pkID 前缀)
func (table *Table) serializeRow(values []types.FieldValue) []byte {
	rowSize := int(table.info.RowSize)
	rowData := make([]byte, rowSize)
	for _, fv := range values {
		idx := table.findFieldIndex(string(fv.FieldName))
		if idx < 0 {
			continue
		}
		off := table.calcFieldOffset(idx)
		ft := table.Field[idx].FieldType
		writeFieldValue(rowData[off:], ft, fv.Value)
	}
	return rowData
}

func writeFieldValue(buf []byte, ft types.FieldType, value interface{}) {
	switch ft {
	case types.FieldType_1byte:
		buf[0] = value.(byte)
	case types.FieldType_2byte:
		binary.BigEndian.PutUint16(buf, uint16(value.(int16)))
	case types.FieldType_4byte:
		binary.BigEndian.PutUint32(buf, uint32(value.(int32)))
	case types.FieldType_8byte:
		binary.BigEndian.PutUint64(buf, value.(uint64))
	default:
		// 所有字符串/定长类型
		if b, ok := value.([]byte); ok {
			copy(buf, b)
		}
	}
}

// deserializeRow 将行数据字节反序列化为 FieldValue 切片 (不含 pkID 前缀)
func (table *Table) deserializeRow(rowData []byte) []types.FieldValue {
	var result []types.FieldValue
	for i, f := range table.Field {
		if util.IsHighBitSet(byte(f.FieldType)) {
			continue // 跳过已删除字段
		}
		off := table.calcFieldOffset(i)
		size := types.GetSize(f.FieldType)
		if int(off)+int(size) > len(rowData) {
			continue
		}
		val := readFieldValue(rowData[off:], f.FieldType, int(size))
		result = append(result, types.FieldValue{FieldName: f.FieldName, Value: val})
	}
	return result
}

func readFieldValue(buf []byte, ft types.FieldType, size int) interface{} {
	switch ft {
	case types.FieldType_1byte:
		return buf[0]
	case types.FieldType_2byte:
		return int16(binary.BigEndian.Uint16(buf))
	case types.FieldType_4byte:
		return int32(binary.BigEndian.Uint32(buf))
	case types.FieldType_8byte:
		return binary.BigEndian.Uint64(buf)
	default:
		// 所有字符串/定长类型
		result := make([]byte, size)
		copy(result, buf[:size])
		return result
	}
}

// ======================== 高层 CRUD ========================

// InsertRow 插入一行数据
func (table *Table) InsertRow(values []types.FieldValue) (uint64, error) {
	pkID := table.info.TotalRows + 1
	rowData := table.serializeRow(values)
	if err := table.indexMeta.InsertRow(table.Field, pkID, rowData); err != nil {
		return 0, err
	}
	table.info.TotalRows++
	table.saveTableInfo()
	return pkID, nil
}

// SearchByIndex 通过索引搜索一行
func (table *Table) SearchByIndex(idxName string, key []byte) ([]types.FieldValue, error) {
	storeRow, found := table.indexMeta.SearchByIndex(idxName, key, table.Field)
	if !found {
		return nil, ErrFieldNotFound
	}
	// storeRow = [pkID(8B LE) + rowData]
	if len(storeRow) < 8 {
		return nil, ErrFieldNotFound
	}
	rowData := storeRow[8:]
	return table.deserializeRow(rowData), nil
}

// DeleteRow 通过索引删除一行
func (table *Table) DeleteRow(idxName string, key []byte) error {
	if err := table.indexMeta.DeleteByKey(idxName, key, table.Field); err != nil {
		return err
	}
	table.info.TotalRows--
	table.saveTableInfo()
	return nil
}

// UpdateRow 通过索引更新一行
func (table *Table) UpdateRow(idxName string, key []byte, values []types.FieldValue) error {
	newRowData := table.serializeRow(values)
	// 找到 pkID
	storeRow, found := table.indexMeta.SearchByIndex(idxName, key, table.Field)
	if !found {
		return ErrFieldNotFound
	}
	pkID := binary.BigEndian.Uint64(storeRow[:8])
	return table.indexMeta.UpdateByKey(idxName, key, newRowData, pkID, table.Field)
}