datapage.go 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158
  1. package timeindex
  2. import (
  3. "TsDB2/DB/core/index/datapage"
  4. "TsDB2/DB/core/page"
  5. "encoding/binary"
  6. "errors"
  7. )
  8. // Row 时序数据行
  9. type Row struct {
  10. TS int64 // 时间戳
  11. Flag byte // 标记/状态
  12. Data []byte // 真实数据
  13. }
  14. type TimeEntry struct {
  15. Hdr page.PageHdr
  16. BeginTime int64 // 本页最小时间
  17. EndTime int64 // 本页最大时间
  18. Count uint32 // 本页记录总数
  19. }
  20. type TimeDataPage struct {
  21. TimeEntry TimeEntry
  22. Rows []Row
  23. }
  24. // ======================== IDateEntry 接口实现 ========================
  25. func (t *TimeDataPage) Insert(handle page.HandleDb, indexEntries []byte, entryLen int, count int, key []byte, pkID uint64, rowData []byte) error {
  26. dataPageID := findDataPageByTime(indexEntries, entryLen, count, key)
  27. var dp *datapage.DataPage
  28. if dataPageID == 0 {
  29. dp = datapage.New(handle)
  30. } else {
  31. var err error
  32. dp, err = datapage.Load(handle, dataPageID)
  33. if err != nil {
  34. return err
  35. }
  36. }
  37. if dp.IsFull(uint32(8 + len(rowData))) {
  38. dp.Save()
  39. dp = datapage.New(handle)
  40. }
  41. storeRow := make([]byte, 8+len(rowData))
  42. binary.BigEndian.PutUint64(storeRow[:8], pkID)
  43. copy(storeRow[8:], rowData)
  44. dp.AppendRow(storeRow)
  45. dp.Save()
  46. if dp.PageID() != dataPageID {
  47. if count*entryLen+4 <= len(indexEntries) {
  48. binary.BigEndian.PutUint32(indexEntries[count*entryLen:], dp.PageID())
  49. }
  50. return errNeedNewEntry
  51. }
  52. return nil
  53. }
  54. func (t *TimeDataPage) Search(handle page.HandleDb, indexEntries []byte, entryLen int, key []byte) ([]byte, bool) {
  55. n := len(indexEntries) / entryLen
  56. dataPageID := findDataPageByTime(indexEntries, entryLen, n, key)
  57. if dataPageID == 0 {
  58. return nil, false
  59. }
  60. dp, err := datapage.Load(handle, dataPageID)
  61. if err != nil {
  62. return nil, false
  63. }
  64. return dp.FindRowByIndexKey(key, 9)
  65. }
  66. func (t *TimeDataPage) CreateDataPage(handle page.HandleDb, key []byte, rowData []byte) (uint32, error) {
  67. dp := datapage.New(handle)
  68. storeRow := make([]byte, 8+len(rowData))
  69. if len(rowData) >= 8 {
  70. copy(storeRow[:8], rowData[:8])
  71. }
  72. copy(storeRow[8:], rowData)
  73. dp.AppendRow(storeRow)
  74. dp.Save()
  75. return dp.PageID(), nil
  76. }
  77. func (t *TimeDataPage) Update(handle page.HandleDb, indexEntries []byte, entryLen int, key []byte, pkID uint64, rowData []byte) error {
  78. n := len(indexEntries) / entryLen
  79. dataPageID := findDataPageByTime(indexEntries, entryLen, n, key)
  80. if dataPageID == 0 {
  81. return errors.New("row not found")
  82. }
  83. dp, err := datapage.Load(handle, dataPageID)
  84. if err != nil {
  85. return err
  86. }
  87. storeRow := make([]byte, 8+len(rowData))
  88. binary.BigEndian.PutUint64(storeRow[:8], pkID)
  89. copy(storeRow[8:], rowData)
  90. return dp.FindAndUpdateByIndexKey(key, 8, storeRow)
  91. }
  92. func (t *TimeDataPage) Delete(handle page.HandleDb, indexEntries []byte, entryLen int, key []byte) error {
  93. n := len(indexEntries) / entryLen
  94. dataPageID := findDataPageByTime(indexEntries, entryLen, n, key)
  95. if dataPageID == 0 {
  96. return errors.New("row not found")
  97. }
  98. dp, err := datapage.Load(handle, dataPageID)
  99. if err != nil {
  100. return err
  101. }
  102. deleted := dp.DeleteByIndexKey(key, 8)
  103. if !deleted {
  104. return errors.New("row not found")
  105. }
  106. if dp.AllDeleted() {
  107. if handle.FreePage != nil {
  108. handle.FreePage(dp.PageID())
  109. }
  110. } else {
  111. dp.Save()
  112. }
  113. return nil
  114. }
  115. // ======================== 内部辅助 ========================
  116. var errNeedNewEntry = errors.New("need new index entry")
  117. // ErrRowTooLarge 行数据太大,需要堆页溢出处理
  118. var ErrRowTooLarge = errors.New("row too large, need heap overflow")
  119. func findDataPageByTime(entries []byte, entryLen int, n int, key []byte) uint32 {
  120. if n == 0 {
  121. return 0
  122. }
  123. // 时间索引条目格式: BeginTime(8) + LastTime(8) + PageID(4)
  124. result := 0
  125. left, right := 0, n-1
  126. for left <= right {
  127. mid := (left + right) / 2
  128. entryBegin := int64(binary.BigEndian.Uint64(entries[mid*entryLen:]))
  129. keyTime := int64(binary.BigEndian.Uint64(key[:8]))
  130. if entryBegin <= keyTime {
  131. result = mid
  132. left = mid + 1
  133. } else {
  134. right = mid - 1
  135. }
  136. }
  137. // PageID 在条目最后 4 字节
  138. off := result*entryLen + entryLen - 4
  139. return binary.BigEndian.Uint32(entries[off:])
  140. }