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