package core

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

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

type FieldOffset struct {
	PageID      uint32 // 页号
	StartOffset uint16 // 开始偏移
	EndOffset   uint16 // 相对开始偏移
}

// SchemaVersion 表的一个 schema 版本定义
type SchemaVersion struct {
	Version      byte
	FieldIndices []byte // 每个字段在 masterFields 中的索引
}
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 // 字段溢出页尾页
}

// readSchema 从 TablePage 字段定义链读取完整的 schema 版本信息 (支持溢出页)
// 新格式: [activeVer(1B)] [masterCount(1B)] [type+len+name]×masterCount [verCount(1B)] [verDefs...]
func readSchema(handle page.HandleDb, pageID uint32, startOff uint16, allocSize uint16) (byte, []types.FieldInfo, []SchemaVersion, FieldOffset) {
	// 收集所有可用数据字节 (跟随溢出页链)
	var data []byte
	curPage := pageID
	curOff := startOff
	firstPage := true
	writePageID := pageID
	writeOff := startOff

	for {
		buf := make([]byte, PageSize)
		if firstPage {
			_ = handle.ReadPage(curPage, buf)
			firstPage = false
		} else {
			hp, err := LoadHeapPage(handle, curPage)
			if err != nil {
				break
			}
			buf = hp.Get()
		}

		// 计算当前块可用数据结束位置 (留 6B 给 jump)
		endOff := curOff
		if firstPage {
			endOff = curOff + allocSize - 6
		} else {
			endOff = curOff + uint16(allocSize) - 6
		}
		if int(endOff) > len(buf) {
			endOff = uint16(len(buf)) - 6
		}

		// 读取从 curOff 到 endOff 的数据
		if int(curOff) < len(buf) {
			chunk := make([]byte, int(endOff)-int(curOff))
			copy(chunk, buf[curOff:endOff])
			data = append(data, chunk...)
		}

		// 检查 jump (最后 6B)
		jumpOff := endOff
		if int(jumpOff)+6 > len(buf) {
			break
		}
		jump := (*RtFieldOffset)(unsafe.Pointer(&buf[jumpOff]))
		if jump.PageID == 0 {
			writePageID = curPage
			writeOff = jumpOff
			break
		}
		writePageID = curPage
		writeOff = jumpOff
		curPage = jump.PageID
		curOff = jump.Offset
	}

	if len(data) < 2 {
		return 0, nil, nil, FieldOffset{PageID: writePageID, StartOffset: writeOff, EndOffset: 0}
	}

	pos := 0
	activeVer := data[pos]
	pos++
	masterCount := int(data[pos])
	pos++

	// 读取 master fields
	var masterFields []types.FieldInfo
	for i := 0; i < masterCount && pos+2 <= len(data); i++ {
		typ := types.FieldType(data[pos])
		nameLen := int(data[pos+1])
		pos += 2
		if pos+nameLen > len(data) {
			break
		}
		name := make([]byte, nameLen)
		copy(name, data[pos:pos+nameLen])
		pos += nameLen
		masterFields = append(masterFields, types.FieldInfo{FieldType: typ, FieldName: name})
	}

	// 读取版本定义
	var versions []SchemaVersion
	if pos < len(data) {
		verCount := int(data[pos])
		pos++
		for i := 0; i < verCount && pos < len(data); i++ {
			verNo := data[pos]
			pos++
			if pos >= len(data) {
				break
			}
			fieldCount := int(data[pos])
			pos++
			indices := make([]byte, fieldCount)
			if pos+fieldCount <= len(data) {
				copy(indices, data[pos:pos+fieldCount])
				pos += fieldCount
			}
			versions = append(versions, SchemaVersion{Version: verNo, FieldIndices: indices})
		}
	}

	return activeVer, masterFields, versions, FieldOffset{
		PageID:      writePageID,
		StartOffset: writeOff,
		EndOffset:   0,
	}
}

type Table struct {
	info             *TableInfo
	Field            []types.FieldInfo // master registry (所有字段,含已删除,删除标记 0x80)
	handle           page.HandleDb
	indexMeta        *IndexMetaPage
	tablePageID      uint32      // 表ID
	tableStartOffset uint32      // 表所在页的偏移
	curOffset        FieldOffset // 当前Field定义可写的偏移位置

	// schema 版本管理
	activeVersion byte
	versions      []SchemaVersion

	// 写操作通知回调 (防抖自动提交)
	onWrite func()
}

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

// TotalRows 返回表的总行数
func (table *Table) TotalRows() uint64 { return table.info.TotalRows }

// TablePageID 返回表页 ID (调试用)
func (table *Table) TablePageID() uint32 { return table.tablePageID }

// TableStartOffset 返回表信息在页中的偏移 (调试用)
func (table *Table) TableStartOffset() uint32 { return table.tableStartOffset }

// IndexMeta 返回索引元数据页 (调试用)
func (table *Table) IndexMeta() *IndexMetaPage { return table.indexMeta }

// SetOnWrite 设置写操作通知回调 (防抖自动提交)
func (table *Table) SetOnWrite(fn func()) { table.onWrite = fn }

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

// getFieldSize 获取字段大小 (去除已删除标记 0x80)
func getFieldSize(ft types.FieldType) uint16 {
	return types.GetSize(ft & 0x7F)
}

// getVersion 获取指定版本的 schema 定义
func (table *Table) getVersion(ver byte) *SchemaVersion {
	for i := range table.versions {
		if table.versions[i].Version == ver {
			return &table.versions[i]
		}
	}
	return nil
}

// getVersionedOffset 计算某行中指定 master 字段索引在行数据中的偏移
// rowData[0] = schemaVer, 后续字段按该版本的 FieldIndices 顺序排列
func (table *Table) getVersionedOffset(rowData []byte, masterFieldIdx int) (uint16, bool) {
	if len(rowData) < 1 {
		return 0, false
	}
	ver := rowData[0]
	sv := table.getVersion(ver)
	if sv == nil {
		return 0, false
	}
	offset := uint16(1) // 跳过 schemaVer 字节
	for _, idx := range sv.FieldIndices {
		if int(idx) == masterFieldIdx {
			return offset, true
		}
		offset += getFieldSize(table.Field[idx].FieldType)
	}
	return 0, false
}

// getActiveFieldTypes 返回当前活跃版本的字段类型列表 (不含已删除)
func (table *Table) getActiveFieldTypes() []types.FieldInfo {
	sv := table.getVersion(table.activeVersion)
	if sv == nil {
		return nil
	}
	fields := make([]types.FieldInfo, len(sv.FieldIndices))
	for i, idx := range sv.FieldIndices {
		fields[i] = table.Field[idx]
	}
	return fields
}

// calcRowDataSize 计算当前活跃版本的行数据大小 (含 schemaVer 1B)
func (table *Table) calcRowDataSize() int {
	size := 1 // schemaVer
	sv := table.getVersion(table.activeVersion)
	if sv != nil {
		for _, idx := range sv.FieldIndices {
			size += int(getFieldSize(table.Field[idx].FieldType))
		}
	}
	return size
}

// 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)
	default:
		// 所有定长字符串类型 (16/32/64/128/256 byte)
		size := int(types.GetSize(typ))
		if size > 0 && int(offset)+size <= len(row) {
			stored := strings.TrimRight(string(row[offset:int(offset)+size]), "\x00")
			var target string
			switch v := value.(type) {
			case []byte:
				target = strings.TrimRight(string(v), "\x00")
			case string:
				target = v
			default:
				return false
			}
			return stored == target
		}
	}
	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))
	case types.FieldType_2byte:
		stored := binary.BigEndian.Uint16(row[offset:])
		return stored > uint16(value.(int16))
	}
	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)
	case types.FieldType_4byte:
		stored := binary.BigEndian.Uint32(row[offset:])
		return stored < uint32(value.(int32))
	case types.FieldType_2byte:
		stored := binary.BigEndian.Uint16(row[offset:])
		return stored < uint16(value.(int16))
	}
	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 }

// GetActiveFieldNames 返回当前活跃版本中包含的所有字段名 (不含已删除)
func (table *Table) GetActiveFieldNames() []string {
	sv := table.getVersion(table.activeVersion)
	if sv == nil {
		return nil
	}
	names := make([]string, 0, len(sv.FieldIndices))
	for _, idx := range sv.FieldIndices {
		ft := table.Field[idx].FieldType
		if ft&0x80 != 0 {
			continue // 跳过已删除
		}
		names = append(names, string(table.Field[idx].FieldName))
	}
	return names
}

// GetActiveFieldType 返回指定字段的 FieldType (基于 master registry)
func (table *Table) GetActiveFieldType(name string) (types.FieldType, bool) {
	for i, f := range table.Field {
		if string(f.FieldName) == name && f.FieldType&0x80 == 0 {
			_ = i
			return f.FieldType, true
		}
	}
	return 0, false
}
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
}

// serializeSchema 将当前 schema 序列化为字节流
func (table *Table) serializeSchema() []byte {
	var data []byte
	data = append(data, table.activeVersion)
	data = append(data, byte(len(table.Field)))
	for _, f := range table.Field {
		data = append(data, byte(f.FieldType))
		data = append(data, byte(len(f.FieldName)))
		data = append(data, f.FieldName...)
	}
	data = append(data, byte(len(table.versions)))
	for _, sv := range table.versions {
		data = append(data, sv.Version)
		data = append(data, byte(len(sv.FieldIndices)))
		data = append(data, sv.FieldIndices...)
	}
	return data
}

// writeSchema 将 schema 数据写入 TablePage 字段定义区域 (支持溢出页)
func (table *Table) writeSchema() error {
	data := table.serializeSchema()
	allocSize := int(table.info.FieldAllocateSize)
	usablePerBlock := allocSize - 6 // 每块留 6B 给 jump

	written := 0
	curPageID := table.tablePageID
	curStart := uint16(table.tableStartOffset + uint32(unsafe.Sizeof(*table.info)))
	isFirst := true

	for written < len(data) {
		buf := make([]byte, PageSize)
		if isFirst {
			_ = table.handle.ReadPage(curPageID, buf)
			isFirst = false
		} else {
			hp, err := LoadHeapPage(table.handle, curPageID)
			if err != nil {
				return err
			}
			buf = hp.Get()
		}

		remaining := len(data) - written
		toWrite := remaining
		if toWrite > usablePerBlock {
			toWrite = usablePerBlock
		}

		copy(buf[curStart:curStart+uint16(toWrite)], data[written:written+toWrite])
		written += toWrite

		if written < len(data) {
			// 需要溢出到新页
			jumpOff := curStart + uint16(usablePerBlock)
			if table.info.LastOverflow == 0 {
				heap := NewHeapPage(table.handle, page.DataOverflowPage)
				table.info.FirstOverflow = heap.Info.Hdr.PageID
				table.info.LastOverflow = heap.Info.Hdr.PageID
				jump := RtFieldOffset{PageID: heap.Info.Hdr.PageID, Offset: uint16(unsafe.Sizeof(page.PageHdr{}))}
				*(*RtFieldOffset)(unsafe.Pointer(&buf[jumpOff])) = jump
				_ = table.handle.SavePage(curPageID, buf)
				curPageID = heap.Info.Hdr.PageID
				curStart = uint16(unsafe.Sizeof(page.PageHdr{}))
			} else {
				jump := RtFieldOffset{PageID: table.info.LastOverflow, Offset: uint16(unsafe.Sizeof(page.PageHdr{}))}
				*(*RtFieldOffset)(unsafe.Pointer(&buf[jumpOff])) = jump
				_ = table.handle.SavePage(curPageID, buf)
				curPageID = table.info.LastOverflow
				curStart = uint16(unsafe.Sizeof(page.PageHdr{}))
			}
		} else {
			// 写完，清零 jump
			jumpOff := curStart + uint16(toWrite)
			if int(jumpOff)+6 <= len(buf) {
				for i := 0; i < 6; i++ {
					buf[jumpOff+uint16(i)] = 0
				}
			}
			_ = table.handle.SavePage(curPageID, buf)
		}
	}

	table.curOffset.PageID = curPageID
	table.curOffset.StartOffset = curStart
	return nil
}

func (table *Table) AddField(newField types.FieldInfo) error {
	existing := table.findFieldByName(newField.FieldName)
	if existing != nil {
		// 字段存在且活跃 → 报错
		if existing.FieldType&0x80 == 0 {
			return ErrFieldExists
		}
		// 字段已删除 (0x80) → 复活它
		existing.FieldType = newField.FieldType // 恢复类型 (清除 0x80)
		table.info.ActiveFieldCount++

		// 找到该字段在 master registry 中的索引
		revivedIdx := -1
		for i := range table.Field {
			if &table.Field[i] == existing {
				revivedIdx = i
				break
			}
		}

		// 构建新版本: 旧版本字段 + 复活的字段索引
		sv := table.getVersion(table.activeVersion)
		newIndices := make([]byte, 0, len(sv.FieldIndices)+1)
		if sv != nil {
			newIndices = append(newIndices, sv.FieldIndices...)
		}
		newIndices = append(newIndices, byte(revivedIdx))

		table.activeVersion++
		table.versions = append(table.versions, SchemaVersion{
			Version:      table.activeVersion,
			FieldIndices: newIndices,
		})

		if err := table.writeSchema(); err != nil {
			return err
		}
		table.saveTableInfo()
		return nil
	}

	// 全新字段: 追加到 master registry
	table.Field = append(table.Field, newField)
	table.info.FieldCount++
	table.info.ActiveFieldCount++

	// 构建新版本: 旧版本字段 + 新字段索引
	sv := table.getVersion(table.activeVersion)
	newIndices := make([]byte, 0, len(sv.FieldIndices)+1)
	if sv != nil {
		newIndices = append(newIndices, sv.FieldIndices...)
	}
	newIndices = append(newIndices, byte(len(table.Field)-1))

	table.activeVersion++
	table.versions = append(table.versions, SchemaVersion{
		Version:      table.activeVersion,
		FieldIndices: newIndices,
	})

	if err := table.writeSchema(); err != nil {
		return err
	}
	table.saveTableInfo()
	return nil
}

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
	info.TotalRows = table.info.TotalRows
	table.handle.SavePage(table.tablePageID, buf)
}

// =============================================================================
// 删除字段（软删除 + 新版本）
// =============================================================================
func (table *Table) DropField(fieldName string) error {
	// 在 master registry 中查找
	masterIdx := -1
	for i, f := range table.Field {
		if util.IsHighBitSet(byte(f.FieldType)) {
			continue // 已删除
		}
		if string(f.FieldName) == fieldName {
			masterIdx = i
			break
		}
	}
	if masterIdx < 0 {
		return ErrFieldNotFound
	}

	// 标记删除
	table.Field[masterIdx].FieldType |= 0x80
	table.info.ActiveFieldCount--

	// 构建新版本: 跳过被删字段
	sv := table.getVersion(table.activeVersion)
	newIndices := make([]byte, 0, len(sv.FieldIndices))
	if sv != nil {
		for _, idx := range sv.FieldIndices {
			if int(idx) != masterIdx {
				newIndices = append(newIndices, idx)
			}
		}
	}

	table.activeVersion++
	table.versions = append(table.versions, SchemaVersion{
		Version:      table.activeVersion,
		FieldIndices: newIndices,
	})

	if err := table.writeSchema(); err != nil {
		return err
	}
	table.saveTableInfo()
	return nil
}

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

// serializeRow 将 FieldValue 切片序列化为行数据字节 (不含 pkID 前缀)
// 新格式: [schemaVer(1B)] [field0_data] [field1_data] ...
func (table *Table) serializeRow(values []types.FieldValue) []byte {
	rowDataSize := table.calcRowDataSize()
	rowData := make([]byte, rowDataSize)
	rowData[0] = table.activeVersion // 写入当前 schema 版本

	sv := table.getVersion(table.activeVersion)
	if sv == nil {
		return rowData
	}

	// 按版本字段顺序写入, offset 从 1 开始 (跳过 schemaVer)
	offset := uint16(1)
	for _, masterIdx := range sv.FieldIndices {
		ft := table.Field[masterIdx].FieldType
		// 在 values 中查找对应字段
		for _, fv := range values {
			if table.findFieldIndex(string(fv.FieldName)) == int(masterIdx) {
				writeFieldValue(rowData[offset:], ft, fv.Value)
				break
			}
		}
		offset += getFieldSize(ft)
	}
	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 前缀)
// 行格式: [schemaVer(1B)] [field_data...] — 按行自身的版本解析
func (table *Table) deserializeRow(rowData []byte) []types.FieldValue {
	if len(rowData) < 1 {
		return nil
	}
	ver := rowData[0]
	sv := table.getVersion(ver)
	if sv == nil {
		return nil // 未知版本, 无法解析
	}

	var result []types.FieldValue
	offset := uint16(1) // 跳过 schemaVer
	for _, idx := range sv.FieldIndices {
		f := table.Field[idx]
		// 跳过已删除字段 (master registry 中标记 0x80)
		if !util.IsHighBitSet(byte(f.FieldType)) {
			size := getFieldSize(f.FieldType)
			if int(offset)+int(size) <= len(rowData) {
				val := readFieldValue(rowData[offset:], f.FieldType, int(size))
				result = append(result, types.FieldValue{FieldName: f.FieldName, Value: val})
			}
		}
		offset += getFieldSize(table.Field[idx].FieldType)
	}
	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, table.activeVersion, table.versions); err != nil {
		return 0, err
	}
	table.info.TotalRows++
	table.saveTableInfo()
	if table.onWrite != nil {
		table.onWrite()
	}
	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, table.activeVersion, table.versions); err != nil {
		return err
	}
	table.info.TotalRows--
	table.saveTableInfo()
	if table.onWrite != nil {
		table.onWrite()
	}
	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])
	if err := table.indexMeta.UpdateByKey(idxName, key, newRowData, pkID, table.Field, table.activeVersion, table.versions); err != nil {
		return err
	}
	if table.onWrite != nil {
		table.onWrite()
	}
	return nil
}