Table.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769
  1. package core
  2. import (
  3. "TsDB2/DB/core/page"
  4. "TsDB2/DB/core/types"
  5. "TsDB2/DB/core/util"
  6. "encoding/binary"
  7. "strings"
  8. "unsafe"
  9. )
  10. type RtFieldOffset struct {
  11. PageID uint32 // 页号
  12. Offset uint16 // 开始偏移
  13. }
  14. type FieldOffset struct {
  15. PageID uint32 // 页号
  16. StartOffset uint16 // 开始偏移
  17. EndOffset uint16 // 相对开始偏移
  18. }
  19. // SchemaVersion 表的一个 schema 版本定义
  20. type SchemaVersion struct {
  21. Version byte
  22. FieldIndices []byte // 每个字段在 masterFields 中的索引
  23. }
  24. type TableInfo struct {
  25. Name [TableNameSize]byte
  26. Flag byte // 表的状态是否被删除
  27. FieldCount byte // 当前字段的总数量(包含已删除,永不减少)byte=255上限
  28. ActiveFieldCount byte // 实际活跃字段数量(不包含已删除)
  29. TotalRows uint64 // 当前表行的总数量
  30. FirstHeap uint32 // 堆链首页
  31. LastHeap uint32 // 堆链尾页
  32. FirstData uint32 // 数据链首页
  33. LastData uint32 // 数据链尾
  34. RowSize uint16 // 单条记录大小(字节)
  35. FieldAllocateSize uint16 // 当前列总分配大小
  36. IndexCount byte // 这张表有几个索引
  37. FirstIndexID uint32 // 本表第一个索引的元数据位置
  38. LastIndexID uint32 // 本表最后一个索引的元数据位置
  39. FirstOverflow uint32 // 字段溢出页首页(0表示无溢出)
  40. LastOverflow uint32 // 字段溢出页尾页
  41. }
  42. // readSchema 从 TablePage 字段定义链读取完整的 schema 版本信息 (支持溢出页)
  43. // 新格式: [activeVer(1B)] [masterCount(1B)] [type+len+name]×masterCount [verCount(1B)] [verDefs...]
  44. func readSchema(handle page.HandleDb, pageID uint32, startOff uint16, allocSize uint16) (byte, []types.FieldInfo, []SchemaVersion, FieldOffset) {
  45. // 收集所有可用数据字节 (跟随溢出页链)
  46. var data []byte
  47. curPage := pageID
  48. curOff := startOff
  49. firstPage := true
  50. writePageID := pageID
  51. writeOff := startOff
  52. for {
  53. buf := make([]byte, PageSize)
  54. if firstPage {
  55. _ = handle.ReadPage(curPage, buf)
  56. firstPage = false
  57. } else {
  58. hp, err := LoadHeapPage(handle, curPage)
  59. if err != nil {
  60. break
  61. }
  62. buf = hp.Get()
  63. }
  64. // 计算当前块可用数据结束位置 (留 6B 给 jump)
  65. endOff := curOff
  66. if firstPage {
  67. endOff = curOff + allocSize - 6
  68. } else {
  69. endOff = curOff + uint16(allocSize) - 6
  70. }
  71. if int(endOff) > len(buf) {
  72. endOff = uint16(len(buf)) - 6
  73. }
  74. // 读取从 curOff 到 endOff 的数据
  75. if int(curOff) < len(buf) {
  76. chunk := make([]byte, int(endOff)-int(curOff))
  77. copy(chunk, buf[curOff:endOff])
  78. data = append(data, chunk...)
  79. }
  80. // 检查 jump (最后 6B)
  81. jumpOff := endOff
  82. if int(jumpOff)+6 > len(buf) {
  83. break
  84. }
  85. jump := (*RtFieldOffset)(unsafe.Pointer(&buf[jumpOff]))
  86. if jump.PageID == 0 {
  87. writePageID = curPage
  88. writeOff = jumpOff
  89. break
  90. }
  91. writePageID = curPage
  92. writeOff = jumpOff
  93. curPage = jump.PageID
  94. curOff = jump.Offset
  95. }
  96. if len(data) < 2 {
  97. return 0, nil, nil, FieldOffset{PageID: writePageID, StartOffset: writeOff, EndOffset: 0}
  98. }
  99. pos := 0
  100. activeVer := data[pos]
  101. pos++
  102. masterCount := int(data[pos])
  103. pos++
  104. // 读取 master fields
  105. var masterFields []types.FieldInfo
  106. for i := 0; i < masterCount && pos+2 <= len(data); i++ {
  107. typ := types.FieldType(data[pos])
  108. nameLen := int(data[pos+1])
  109. pos += 2
  110. if pos+nameLen > len(data) {
  111. break
  112. }
  113. name := make([]byte, nameLen)
  114. copy(name, data[pos:pos+nameLen])
  115. pos += nameLen
  116. masterFields = append(masterFields, types.FieldInfo{FieldType: typ, FieldName: name})
  117. }
  118. // 读取版本定义
  119. var versions []SchemaVersion
  120. if pos < len(data) {
  121. verCount := int(data[pos])
  122. pos++
  123. for i := 0; i < verCount && pos < len(data); i++ {
  124. verNo := data[pos]
  125. pos++
  126. if pos >= len(data) {
  127. break
  128. }
  129. fieldCount := int(data[pos])
  130. pos++
  131. indices := make([]byte, fieldCount)
  132. if pos+fieldCount <= len(data) {
  133. copy(indices, data[pos:pos+fieldCount])
  134. pos += fieldCount
  135. }
  136. versions = append(versions, SchemaVersion{Version: verNo, FieldIndices: indices})
  137. }
  138. }
  139. return activeVer, masterFields, versions, FieldOffset{
  140. PageID: writePageID,
  141. StartOffset: writeOff,
  142. EndOffset: 0,
  143. }
  144. }
  145. type Table struct {
  146. info *TableInfo
  147. Field []types.FieldInfo // master registry (所有字段,含已删除,删除标记 0x80)
  148. handle page.HandleDb
  149. indexMeta *IndexMetaPage
  150. tablePageID uint32 // 表ID
  151. tableStartOffset uint32 // 表所在页的偏移
  152. curOffset FieldOffset // 当前Field定义可写的偏移位置
  153. // schema 版本管理
  154. activeVersion byte
  155. versions []SchemaVersion
  156. // 写操作通知回调 (防抖自动提交)
  157. onWrite func()
  158. }
  159. func (table *Table) CreateIndex(idxName string, typ page.IndexType, fields []byte) (*IndexMeta, error) {
  160. return table.indexMeta.CreateIndex(idxName, typ, fields)
  161. }
  162. func (table *Table) DropIndex(idx string) error { return table.indexMeta.DropIndex(idx) }
  163. // TotalRows 返回表的总行数
  164. func (table *Table) TotalRows() uint64 { return table.info.TotalRows }
  165. // TablePageID 返回表页 ID (调试用)
  166. func (table *Table) TablePageID() uint32 { return table.tablePageID }
  167. // TableStartOffset 返回表信息在页中的偏移 (调试用)
  168. func (table *Table) TableStartOffset() uint32 { return table.tableStartOffset }
  169. // IndexMeta 返回索引元数据页 (调试用)
  170. func (table *Table) IndexMeta() *IndexMetaPage { return table.indexMeta }
  171. // SetOnWrite 设置写操作通知回调 (防抖自动提交)
  172. func (table *Table) SetOnWrite(fn func()) { table.onWrite = fn }
  173. // findFieldIndex 查找字段在 master registry 中的索引
  174. func (table *Table) findFieldIndex(name string) int {
  175. for i, field := range table.Field {
  176. if string(field.FieldName) == name {
  177. return i
  178. }
  179. }
  180. return -1
  181. }
  182. // getFieldSize 获取字段大小 (去除已删除标记 0x80)
  183. func getFieldSize(ft types.FieldType) uint16 {
  184. return types.GetSize(ft & 0x7F)
  185. }
  186. // getVersion 获取指定版本的 schema 定义
  187. func (table *Table) getVersion(ver byte) *SchemaVersion {
  188. for i := range table.versions {
  189. if table.versions[i].Version == ver {
  190. return &table.versions[i]
  191. }
  192. }
  193. return nil
  194. }
  195. // getVersionedOffset 计算某行中指定 master 字段索引在行数据中的偏移
  196. // rowData[0] = schemaVer, 后续字段按该版本的 FieldIndices 顺序排列
  197. func (table *Table) getVersionedOffset(rowData []byte, masterFieldIdx int) (uint16, bool) {
  198. if len(rowData) < 1 {
  199. return 0, false
  200. }
  201. ver := rowData[0]
  202. sv := table.getVersion(ver)
  203. if sv == nil {
  204. return 0, false
  205. }
  206. offset := uint16(1) // 跳过 schemaVer 字节
  207. for _, idx := range sv.FieldIndices {
  208. if int(idx) == masterFieldIdx {
  209. return offset, true
  210. }
  211. offset += getFieldSize(table.Field[idx].FieldType)
  212. }
  213. return 0, false
  214. }
  215. // getActiveFieldTypes 返回当前活跃版本的字段类型列表 (不含已删除)
  216. func (table *Table) getActiveFieldTypes() []types.FieldInfo {
  217. sv := table.getVersion(table.activeVersion)
  218. if sv == nil {
  219. return nil
  220. }
  221. fields := make([]types.FieldInfo, len(sv.FieldIndices))
  222. for i, idx := range sv.FieldIndices {
  223. fields[i] = table.Field[idx]
  224. }
  225. return fields
  226. }
  227. // calcRowDataSize 计算当前活跃版本的行数据大小 (含 schemaVer 1B)
  228. func (table *Table) calcRowDataSize() int {
  229. size := 1 // schemaVer
  230. sv := table.getVersion(table.activeVersion)
  231. if sv != nil {
  232. for _, idx := range sv.FieldIndices {
  233. size += int(getFieldSize(table.Field[idx].FieldType))
  234. }
  235. }
  236. return size
  237. }
  238. // compareValue 比较字段值是否相等
  239. func (table *Table) compareValue(row []byte, offset uint16, typ types.FieldType, value interface{}) bool {
  240. switch typ {
  241. case types.FieldType_1byte:
  242. return row[offset] == value.(byte)
  243. case types.FieldType_2byte:
  244. stored := binary.BigEndian.Uint16(row[offset:])
  245. return stored == uint16(value.(int16))
  246. case types.FieldType_4byte:
  247. stored := binary.BigEndian.Uint32(row[offset:])
  248. return stored == uint32(value.(int32))
  249. case types.FieldType_8byte:
  250. stored := binary.BigEndian.Uint64(row[offset:])
  251. return stored == value.(uint64)
  252. default:
  253. // 所有定长字符串类型 (16/32/64/128/256 byte)
  254. size := int(types.GetSize(typ))
  255. if size > 0 && int(offset)+size <= len(row) {
  256. stored := strings.TrimRight(string(row[offset:int(offset)+size]), "\x00")
  257. var target string
  258. switch v := value.(type) {
  259. case []byte:
  260. target = strings.TrimRight(string(v), "\x00")
  261. case string:
  262. target = v
  263. default:
  264. return false
  265. }
  266. return stored == target
  267. }
  268. }
  269. return false
  270. }
  271. // compareGreater 比较字段值是否大于给定值
  272. func (table *Table) compareGreater(row []byte, offset uint16, typ types.FieldType, value interface{}) bool {
  273. switch typ {
  274. case types.FieldType_8byte:
  275. stored := binary.BigEndian.Uint64(row[offset:])
  276. return stored > value.(uint64)
  277. case types.FieldType_4byte:
  278. stored := binary.BigEndian.Uint32(row[offset:])
  279. return stored > uint32(value.(int32))
  280. case types.FieldType_2byte:
  281. stored := binary.BigEndian.Uint16(row[offset:])
  282. return stored > uint16(value.(int16))
  283. }
  284. return false
  285. }
  286. // compareLess 比较字段值是否小于给定值
  287. func (table *Table) compareLess(row []byte, offset uint16, typ types.FieldType, value interface{}) bool {
  288. switch typ {
  289. case types.FieldType_8byte:
  290. stored := binary.BigEndian.Uint64(row[offset:])
  291. return stored < value.(uint64)
  292. case types.FieldType_4byte:
  293. stored := binary.BigEndian.Uint32(row[offset:])
  294. return stored < uint32(value.(int32))
  295. case types.FieldType_2byte:
  296. stored := binary.BigEndian.Uint16(row[offset:])
  297. return stored < uint16(value.(int16))
  298. }
  299. return false
  300. }
  301. // compareGreaterOrEqual 比较字段值是否大于等于给定值
  302. func (table *Table) compareGreaterOrEqual(row []byte, offset uint16, typ types.FieldType, value interface{}) bool {
  303. return table.compareGreater(row, offset, typ, value) || table.compareValue(row, offset, typ, value)
  304. }
  305. // compareLessOrEqual 比较字段值是否小于等于给定值
  306. func (table *Table) compareLessOrEqual(row []byte, offset uint16, typ types.FieldType, value interface{}) bool {
  307. return table.compareLess(row, offset, typ, value) || table.compareValue(row, offset, typ, value)
  308. }
  309. func (table *Table) GetFieldCount() byte { return table.info.FieldCount }
  310. func (table *Table) GetActiveFieldCount() byte { return table.info.ActiveFieldCount }
  311. // GetActiveFieldNames 返回当前活跃版本中包含的所有字段名 (不含已删除)
  312. func (table *Table) GetActiveFieldNames() []string {
  313. sv := table.getVersion(table.activeVersion)
  314. if sv == nil {
  315. return nil
  316. }
  317. names := make([]string, 0, len(sv.FieldIndices))
  318. for _, idx := range sv.FieldIndices {
  319. ft := table.Field[idx].FieldType
  320. if ft&0x80 != 0 {
  321. continue // 跳过已删除
  322. }
  323. names = append(names, string(table.Field[idx].FieldName))
  324. }
  325. return names
  326. }
  327. // GetActiveFieldType 返回指定字段的 FieldType (基于 master registry)
  328. func (table *Table) GetActiveFieldType(name string) (types.FieldType, bool) {
  329. for i, f := range table.Field {
  330. if string(f.FieldName) == name && f.FieldType&0x80 == 0 {
  331. _ = i
  332. return f.FieldType, true
  333. }
  334. }
  335. return 0, false
  336. }
  337. func (table *Table) GetFieldAllocateSize() uint16 { return table.info.FieldAllocateSize }
  338. func (table *Table) GetFirstOverflow() uint32 { return table.info.FirstOverflow }
  339. func (table *Table) GetLastOverflow() uint32 { return table.info.LastOverflow }
  340. func (table *Table) GetFirstData() uint32 { return table.info.FirstData }
  341. func (table *Table) GetFields() []types.FieldInfo {
  342. return table.Field
  343. }
  344. func (table *Table) findFieldByName(name []byte) *types.FieldInfo {
  345. target := util.BytesToString(name)
  346. for i := range table.Field {
  347. if util.BytesToString(table.Field[i].FieldName) == target {
  348. return &table.Field[i]
  349. }
  350. }
  351. return nil
  352. }
  353. // serializeSchema 将当前 schema 序列化为字节流
  354. func (table *Table) serializeSchema() []byte {
  355. var data []byte
  356. data = append(data, table.activeVersion)
  357. data = append(data, byte(len(table.Field)))
  358. for _, f := range table.Field {
  359. data = append(data, byte(f.FieldType))
  360. data = append(data, byte(len(f.FieldName)))
  361. data = append(data, f.FieldName...)
  362. }
  363. data = append(data, byte(len(table.versions)))
  364. for _, sv := range table.versions {
  365. data = append(data, sv.Version)
  366. data = append(data, byte(len(sv.FieldIndices)))
  367. data = append(data, sv.FieldIndices...)
  368. }
  369. return data
  370. }
  371. // writeSchema 将 schema 数据写入 TablePage 字段定义区域 (支持溢出页)
  372. func (table *Table) writeSchema() error {
  373. data := table.serializeSchema()
  374. allocSize := int(table.info.FieldAllocateSize)
  375. usablePerBlock := allocSize - 6 // 每块留 6B 给 jump
  376. written := 0
  377. curPageID := table.tablePageID
  378. curStart := uint16(table.tableStartOffset + uint32(unsafe.Sizeof(*table.info)))
  379. isFirst := true
  380. for written < len(data) {
  381. buf := make([]byte, PageSize)
  382. if isFirst {
  383. _ = table.handle.ReadPage(curPageID, buf)
  384. isFirst = false
  385. } else {
  386. hp, err := LoadHeapPage(table.handle, curPageID)
  387. if err != nil {
  388. return err
  389. }
  390. buf = hp.Get()
  391. }
  392. remaining := len(data) - written
  393. toWrite := remaining
  394. if toWrite > usablePerBlock {
  395. toWrite = usablePerBlock
  396. }
  397. copy(buf[curStart:curStart+uint16(toWrite)], data[written:written+toWrite])
  398. written += toWrite
  399. if written < len(data) {
  400. // 需要溢出到新页
  401. jumpOff := curStart + uint16(usablePerBlock)
  402. if table.info.LastOverflow == 0 {
  403. heap := NewHeapPage(table.handle, page.DataOverflowPage)
  404. table.info.FirstOverflow = heap.Info.Hdr.PageID
  405. table.info.LastOverflow = heap.Info.Hdr.PageID
  406. jump := RtFieldOffset{PageID: heap.Info.Hdr.PageID, Offset: uint16(unsafe.Sizeof(page.PageHdr{}))}
  407. *(*RtFieldOffset)(unsafe.Pointer(&buf[jumpOff])) = jump
  408. _ = table.handle.SavePage(curPageID, buf)
  409. curPageID = heap.Info.Hdr.PageID
  410. curStart = uint16(unsafe.Sizeof(page.PageHdr{}))
  411. } else {
  412. jump := RtFieldOffset{PageID: table.info.LastOverflow, Offset: uint16(unsafe.Sizeof(page.PageHdr{}))}
  413. *(*RtFieldOffset)(unsafe.Pointer(&buf[jumpOff])) = jump
  414. _ = table.handle.SavePage(curPageID, buf)
  415. curPageID = table.info.LastOverflow
  416. curStart = uint16(unsafe.Sizeof(page.PageHdr{}))
  417. }
  418. } else {
  419. // 写完,清零 jump
  420. jumpOff := curStart + uint16(toWrite)
  421. if int(jumpOff)+6 <= len(buf) {
  422. for i := 0; i < 6; i++ {
  423. buf[jumpOff+uint16(i)] = 0
  424. }
  425. }
  426. _ = table.handle.SavePage(curPageID, buf)
  427. }
  428. }
  429. table.curOffset.PageID = curPageID
  430. table.curOffset.StartOffset = curStart
  431. return nil
  432. }
  433. func (table *Table) AddField(newField types.FieldInfo) error {
  434. existing := table.findFieldByName(newField.FieldName)
  435. if existing != nil {
  436. // 字段存在且活跃 → 报错
  437. if existing.FieldType&0x80 == 0 {
  438. return ErrFieldExists
  439. }
  440. // 字段已删除 (0x80) → 复活它
  441. existing.FieldType = newField.FieldType // 恢复类型 (清除 0x80)
  442. table.info.ActiveFieldCount++
  443. // 找到该字段在 master registry 中的索引
  444. revivedIdx := -1
  445. for i := range table.Field {
  446. if &table.Field[i] == existing {
  447. revivedIdx = i
  448. break
  449. }
  450. }
  451. // 构建新版本: 旧版本字段 + 复活的字段索引
  452. sv := table.getVersion(table.activeVersion)
  453. newIndices := make([]byte, 0, len(sv.FieldIndices)+1)
  454. if sv != nil {
  455. newIndices = append(newIndices, sv.FieldIndices...)
  456. }
  457. newIndices = append(newIndices, byte(revivedIdx))
  458. table.activeVersion++
  459. table.versions = append(table.versions, SchemaVersion{
  460. Version: table.activeVersion,
  461. FieldIndices: newIndices,
  462. })
  463. if err := table.writeSchema(); err != nil {
  464. return err
  465. }
  466. table.saveTableInfo()
  467. return nil
  468. }
  469. // 全新字段: 追加到 master registry
  470. table.Field = append(table.Field, newField)
  471. table.info.FieldCount++
  472. table.info.ActiveFieldCount++
  473. // 构建新版本: 旧版本字段 + 新字段索引
  474. sv := table.getVersion(table.activeVersion)
  475. newIndices := make([]byte, 0, len(sv.FieldIndices)+1)
  476. if sv != nil {
  477. newIndices = append(newIndices, sv.FieldIndices...)
  478. }
  479. newIndices = append(newIndices, byte(len(table.Field)-1))
  480. table.activeVersion++
  481. table.versions = append(table.versions, SchemaVersion{
  482. Version: table.activeVersion,
  483. FieldIndices: newIndices,
  484. })
  485. if err := table.writeSchema(); err != nil {
  486. return err
  487. }
  488. table.saveTableInfo()
  489. return nil
  490. }
  491. func (table *Table) saveTableInfo() {
  492. buf := make([]byte, PageSize)
  493. if err := table.handle.ReadPage(table.tablePageID, buf); err != nil {
  494. return
  495. }
  496. info := (*TableInfo)(unsafe.Pointer(&buf[table.tableStartOffset]))
  497. info.FirstOverflow = table.info.FirstOverflow
  498. info.LastOverflow = table.info.LastOverflow
  499. info.FieldCount = table.info.FieldCount
  500. info.ActiveFieldCount = table.info.ActiveFieldCount
  501. info.TotalRows = table.info.TotalRows
  502. table.handle.SavePage(table.tablePageID, buf)
  503. }
  504. // =============================================================================
  505. // 删除字段(软删除 + 新版本)
  506. // =============================================================================
  507. func (table *Table) DropField(fieldName string) error {
  508. // 在 master registry 中查找
  509. masterIdx := -1
  510. for i, f := range table.Field {
  511. if util.IsHighBitSet(byte(f.FieldType)) {
  512. continue // 已删除
  513. }
  514. if string(f.FieldName) == fieldName {
  515. masterIdx = i
  516. break
  517. }
  518. }
  519. if masterIdx < 0 {
  520. return ErrFieldNotFound
  521. }
  522. // 标记删除
  523. table.Field[masterIdx].FieldType |= 0x80
  524. table.info.ActiveFieldCount--
  525. // 构建新版本: 跳过被删字段
  526. sv := table.getVersion(table.activeVersion)
  527. newIndices := make([]byte, 0, len(sv.FieldIndices))
  528. if sv != nil {
  529. for _, idx := range sv.FieldIndices {
  530. if int(idx) != masterIdx {
  531. newIndices = append(newIndices, idx)
  532. }
  533. }
  534. }
  535. table.activeVersion++
  536. table.versions = append(table.versions, SchemaVersion{
  537. Version: table.activeVersion,
  538. FieldIndices: newIndices,
  539. })
  540. if err := table.writeSchema(); err != nil {
  541. return err
  542. }
  543. table.saveTableInfo()
  544. return nil
  545. }
  546. // ======================== 行数据序列化/反序列化 ========================
  547. // serializeRow 将 FieldValue 切片序列化为行数据字节 (不含 pkID 前缀)
  548. // 新格式: [schemaVer(1B)] [field0_data] [field1_data] ...
  549. func (table *Table) serializeRow(values []types.FieldValue) []byte {
  550. rowDataSize := table.calcRowDataSize()
  551. rowData := make([]byte, rowDataSize)
  552. rowData[0] = table.activeVersion // 写入当前 schema 版本
  553. sv := table.getVersion(table.activeVersion)
  554. if sv == nil {
  555. return rowData
  556. }
  557. // 按版本字段顺序写入, offset 从 1 开始 (跳过 schemaVer)
  558. offset := uint16(1)
  559. for _, masterIdx := range sv.FieldIndices {
  560. ft := table.Field[masterIdx].FieldType
  561. // 在 values 中查找对应字段
  562. for _, fv := range values {
  563. if table.findFieldIndex(string(fv.FieldName)) == int(masterIdx) {
  564. writeFieldValue(rowData[offset:], ft, fv.Value)
  565. break
  566. }
  567. }
  568. offset += getFieldSize(ft)
  569. }
  570. return rowData
  571. }
  572. func writeFieldValue(buf []byte, ft types.FieldType, value interface{}) {
  573. switch ft {
  574. case types.FieldType_1byte:
  575. buf[0] = value.(byte)
  576. case types.FieldType_2byte:
  577. binary.BigEndian.PutUint16(buf, uint16(value.(int16)))
  578. case types.FieldType_4byte:
  579. binary.BigEndian.PutUint32(buf, uint32(value.(int32)))
  580. case types.FieldType_8byte:
  581. binary.BigEndian.PutUint64(buf, value.(uint64))
  582. default:
  583. // 所有字符串/定长类型
  584. if b, ok := value.([]byte); ok {
  585. copy(buf, b)
  586. }
  587. }
  588. }
  589. // deserializeRow 将行数据字节反序列化为 FieldValue 切片 (不含 pkID 前缀)
  590. // 行格式: [schemaVer(1B)] [field_data...] — 按行自身的版本解析
  591. func (table *Table) deserializeRow(rowData []byte) []types.FieldValue {
  592. if len(rowData) < 1 {
  593. return nil
  594. }
  595. ver := rowData[0]
  596. sv := table.getVersion(ver)
  597. if sv == nil {
  598. return nil // 未知版本, 无法解析
  599. }
  600. var result []types.FieldValue
  601. offset := uint16(1) // 跳过 schemaVer
  602. for _, idx := range sv.FieldIndices {
  603. f := table.Field[idx]
  604. // 跳过已删除字段 (master registry 中标记 0x80)
  605. if !util.IsHighBitSet(byte(f.FieldType)) {
  606. size := getFieldSize(f.FieldType)
  607. if int(offset)+int(size) <= len(rowData) {
  608. val := readFieldValue(rowData[offset:], f.FieldType, int(size))
  609. result = append(result, types.FieldValue{FieldName: f.FieldName, Value: val})
  610. }
  611. }
  612. offset += getFieldSize(table.Field[idx].FieldType)
  613. }
  614. return result
  615. }
  616. func readFieldValue(buf []byte, ft types.FieldType, size int) interface{} {
  617. switch ft {
  618. case types.FieldType_1byte:
  619. return buf[0]
  620. case types.FieldType_2byte:
  621. return int16(binary.BigEndian.Uint16(buf))
  622. case types.FieldType_4byte:
  623. return int32(binary.BigEndian.Uint32(buf))
  624. case types.FieldType_8byte:
  625. return binary.BigEndian.Uint64(buf)
  626. default:
  627. // 所有字符串/定长类型
  628. result := make([]byte, size)
  629. copy(result, buf[:size])
  630. return result
  631. }
  632. }
  633. // ======================== 高层 CRUD ========================
  634. // InsertRow 插入一行数据
  635. func (table *Table) InsertRow(values []types.FieldValue) (uint64, error) {
  636. pkID := table.info.TotalRows + 1
  637. rowData := table.serializeRow(values)
  638. if err := table.indexMeta.InsertRow(table.Field, pkID, rowData, table.activeVersion, table.versions); err != nil {
  639. return 0, err
  640. }
  641. table.info.TotalRows++
  642. table.saveTableInfo()
  643. if table.onWrite != nil {
  644. table.onWrite()
  645. }
  646. return pkID, nil
  647. }
  648. // SearchByIndex 通过索引搜索一行
  649. func (table *Table) SearchByIndex(idxName string, key []byte) ([]types.FieldValue, error) {
  650. storeRow, found := table.indexMeta.SearchByIndex(idxName, key, table.Field)
  651. if !found {
  652. return nil, ErrFieldNotFound
  653. }
  654. // storeRow = [pkID(8B LE) + rowData]
  655. if len(storeRow) < 8 {
  656. return nil, ErrFieldNotFound
  657. }
  658. rowData := storeRow[8:]
  659. return table.deserializeRow(rowData), nil
  660. }
  661. // DeleteRow 通过索引删除一行
  662. func (table *Table) DeleteRow(idxName string, key []byte) error {
  663. if err := table.indexMeta.DeleteByKey(idxName, key, table.Field, table.activeVersion, table.versions); err != nil {
  664. return err
  665. }
  666. table.info.TotalRows--
  667. table.saveTableInfo()
  668. if table.onWrite != nil {
  669. table.onWrite()
  670. }
  671. return nil
  672. }
  673. // UpdateRow 通过索引更新一行
  674. func (table *Table) UpdateRow(idxName string, key []byte, values []types.FieldValue) error {
  675. newRowData := table.serializeRow(values)
  676. // 找到 pkID
  677. storeRow, found := table.indexMeta.SearchByIndex(idxName, key, table.Field)
  678. if !found {
  679. return ErrFieldNotFound
  680. }
  681. pkID := binary.BigEndian.Uint64(storeRow[:8])
  682. if err := table.indexMeta.UpdateByKey(idxName, key, newRowData, pkID, table.Field, table.activeVersion, table.versions); err != nil {
  683. return err
  684. }
  685. if table.onWrite != nil {
  686. table.onWrite()
  687. }
  688. return nil
  689. }