compose_handler.go 8.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302
  1. package index
  2. import (
  3. "TsDB2/DB/core/index/composepage"
  4. "TsDB2/DB/core/page"
  5. "TsDB2/DB/core/util"
  6. "encoding/binary"
  7. "unsafe"
  8. )
  9. func init() {
  10. Compose.RegisterHandler(page.IndexTypeCompose, &ComposeChainHandler{})
  11. }
  12. // ======================== IEntry: 澶嶅悎绱㈠紩鏉$洰 ========================
  13. // ComposeEntry 閫氱敤澶嶅悎绱㈠紩缁撴瀯浣?
  14. // Keys: 瀛樺偍 N 涓储寮曢敭 (uint64)
  15. // PKID: 涓婚敭ID锛?瀛楄妭锛夛紝閫氳繃 B+ 鏍戣В鏋愭暟鎹綅缃?
  16. type ComposeEntry struct {
  17. Keys []uint64 // 浠绘剰澶氫釜绱㈠紩閿紙1涓€?涓€?涓?..閮芥敮鎸侊級
  18. PKID uint64 // 涓婚敭ID (8瀛楄妭)
  19. }
  20. func (e *ComposeEntry) DataEntry() IDateEntry {
  21. return &composepage.ComposeEntryPage{}
  22. }
  23. // EntryLength 鑷姩璁$畻闀垮害锛?*閿暟閲?+ 8 (PK_ID)
  24. func (e *ComposeEntry) EntryLength() uint16 {
  25. return uint16(len(e.Keys)*8 + 8)
  26. }
  27. // Compare 閫氱敤澶嶅悎绱㈠紩姣旇緝閫昏緫锛氳嚜鍔ㄦ寜椤哄簭姣旇緝鎵€鏈夐敭锛堜笉鍚湯灏?PK_ID锛?
  28. func (e *ComposeEntry) Compare(midEntry []byte, key []byte) int {
  29. entryLen := int(e.EntryLength())
  30. keyCount := (entryLen - 8) / 8
  31. for i := 0; i < keyCount; i++ {
  32. offset := i * 8
  33. f1 := binary.BigEndian.Uint64(midEntry[offset : offset+8])
  34. f2 := binary.BigEndian.Uint64(key[offset : offset+8])
  35. cmp := util.CompareUint64(f1, f2)
  36. if cmp != 0 {
  37. return cmp
  38. }
  39. }
  40. return 0
  41. }
  42. // Serialize 鑷姩搴忓垪鍖?N 涓敭 + PK_ID
  43. func (e *ComposeEntry) Serialize(ie IEntry) []byte {
  44. ent, ok := ie.(*ComposeEntry)
  45. if !ok {
  46. return nil
  47. }
  48. keyCount := len(ent.Keys)
  49. buf := make([]byte, keyCount*8+8)
  50. for i, k := range ent.Keys {
  51. binary.BigEndian.PutUint64(buf[i*8:(i+1)*8], k)
  52. }
  53. binary.BigEndian.PutUint64(buf[keyCount*8:], ent.PKID)
  54. return buf
  55. }
  56. // Deserialize 鑷姩鍙嶅簭鍒楀寲 N 涓敭 + PK_ID
  57. func (e *ComposeEntry) Deserialize(buf []byte) IEntry {
  58. keyCount := (len(buf) - 8) / 8
  59. keys := make([]uint64, 0, keyCount)
  60. for i := 0; i < keyCount; i++ {
  61. keys = append(keys, binary.BigEndian.Uint64(buf[i*8:(i+1)*8]))
  62. }
  63. return &ComposeEntry{
  64. Keys: keys,
  65. PKID: binary.BigEndian.Uint64(buf[keyCount*8:]),
  66. }
  67. }
  68. // Insert 鎻掑叆绱㈠紩
  69. func (e *ComposeEntry) Insert(entries *[]byte, data []byte) error {
  70. *entries = append(*entries, data...)
  71. return nil
  72. }
  73. // Search 浜屽垎鏌ユ壘锛堥€氱敤 N 瀛楁澶嶅悎绱㈠紩锛?
  74. func (e *ComposeEntry) Search(entries []byte, key []byte) (uint32, bool) {
  75. entryLen := int(e.EntryLength())
  76. pos, found := util.BinarySearch(entries, entryLen, key, e.Compare)
  77. if !found {
  78. return 0, false
  79. }
  80. return pos, true
  81. }
  82. // ======================== IndexHandler: 澶嶅悎浜岀骇绱㈠紩澶勭悊鍣?========================
  83. // ComposeChainHandler 鈥?澶嶅悎浜岀骇绱㈠紩 (page.IndexTypeCompose)
  84. // 鎵佸钩绱㈠紩椤甸摼, 鏉$洰 [composeKey(N*8B) + PKID(8B)]
  85. type ComposeChainHandler struct{}
  86. func (h *ComposeChainHandler) Insert(handle page.HandleDb, entryPage *uint32, indexKey []byte, pkID uint64, rowData []byte, keyLen int) (uint32, error) {
  87. dataEntry := h.getDataEntry()
  88. curPageID := *entryPage
  89. lastIdxPageID := uint32(0)
  90. for curPageID != 0 {
  91. buf := make([]byte, page.IndexPageSize)
  92. if err := handle.ReadPage(curPageID, buf); err != nil {
  93. return 0, err
  94. }
  95. hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
  96. pageEntryLen := int(hdr.EntryLen)
  97. count := int(hdr.Count)
  98. idxPageHasRoom := hdr.Hdr.UseSize+uint32(pageEntryLen) <= page.IndexPageSize
  99. if !idxPageHasRoom {
  100. lastIdxPageID = curPageID
  101. curPageID = hdr.Hdr.NextPage
  102. continue
  103. }
  104. entries := buf[page.IndexHeadSize : page.IndexHeadSize+(count+1)*pageEntryLen]
  105. err := dataEntry.Insert(handle, entries, pageEntryLen, count, indexKey, pkID, rowData)
  106. if err == nil {
  107. hdr.Count++
  108. hdr.Hdr.UseSize += uint32(pageEntryLen)
  109. _ = handle.SavePage(curPageID, buf)
  110. return *entryPage, nil
  111. }
  112. lastIdxPageID = curPageID
  113. curPageID = hdr.Hdr.NextPage
  114. }
  115. entryLen := keyLen + 8
  116. newEntry := make([]byte, entryLen)
  117. copy(newEntry[:keyLen], indexKey)
  118. binary.BigEndian.PutUint64(newEntry[keyLen:], pkID)
  119. return *entryPage, writeToNewIndexPage(handle, entryPage, newEntry, lastIdxPageID)
  120. }
  121. func (h *ComposeChainHandler) Search(handle page.HandleDb, entryPage uint32, indexKey []byte, keyLen int) ([]byte, bool) {
  122. dataEntry := h.getDataEntry()
  123. curPageID := entryPage
  124. for curPageID != 0 {
  125. buf := make([]byte, page.IndexPageSize)
  126. if err := handle.ReadPage(curPageID, buf); err != nil {
  127. break
  128. }
  129. hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
  130. entries := buf[page.IndexHeadSize : page.IndexHeadSize+int(hdr.Count)*int(hdr.EntryLen)]
  131. entryLen := int(hdr.EntryLen)
  132. rowData, found := dataEntry.Search(handle, entries, entryLen, indexKey)
  133. if found {
  134. return rowData, true
  135. }
  136. curPageID = hdr.Hdr.NextPage
  137. }
  138. return nil, false
  139. }
  140. func (h *ComposeChainHandler) Delete(handle page.HandleDb, entryPage *uint32, indexKey []byte, keyLen int) error {
  141. return h.deleteEntry(handle, entryPage, indexKey)
  142. }
  143. func (h *ComposeChainHandler) Update(handle page.HandleDb, entryPage *uint32, oldKey []byte, rowData []byte, pkID uint64, keyLen int) error {
  144. // 浜岀骇绱㈠紩鏇存柊鐢?IndexMetaPage.UpdateByKey 缁熶竴澶勭悊
  145. return page.ErrNotFound
  146. }
  147. func (h *ComposeChainHandler) GetDataPageIDs(handle page.HandleDb, entryPage uint32) []uint32 {
  148. return nil
  149. }
  150. func (h *ComposeChainHandler) ReadDataPageRows(handle page.HandleDb, dpID uint32) [][]byte {
  151. return nil
  152. }
  153. // RangeScan 澶嶅悎浜岀骇绱㈠紩鑼冨洿鎵弿
  154. // 鏉$洰鏍煎紡: [key(keyLen) + pkID(8B BE)]
  155. // 鍥炶皟鏀跺埌 8B BigEndian pkID
  156. func (h *ComposeChainHandler) RangeScan(handle page.HandleDb, entryPage uint32,
  157. startKey, endKey []byte, startInclusive, endInclusive bool, reverse bool,
  158. callback func(row []byte) bool) error {
  159. if entryPage == 0 {
  160. return nil
  161. }
  162. type pageData struct {
  163. entries []byte
  164. el int
  165. count int
  166. }
  167. var pages []pageData
  168. curPageID := entryPage
  169. for curPageID != 0 {
  170. buf := make([]byte, page.IndexPageSize)
  171. if err := handle.ReadPage(curPageID, buf); err != nil {
  172. break
  173. }
  174. hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
  175. el := int(hdr.EntryLen)
  176. count := int(hdr.Count)
  177. entries := make([]byte, count*el)
  178. copy(entries, buf[page.IndexHeadSize:page.IndexHeadSize+count*el])
  179. pages = append(pages, pageData{entries: entries, el: el, count: count})
  180. curPageID = hdr.Hdr.NextPage
  181. }
  182. if !reverse {
  183. for p := 0; p < len(pages); p++ {
  184. pg := &pages[p]
  185. for i := 0; i < pg.count; i++ {
  186. entry := pg.entries[i*pg.el : (i+1)*pg.el]
  187. key := entry[:pg.el-8]
  188. if endKey != nil {
  189. cmp := composeCmp(key, endKey)
  190. if cmp > 0 || (cmp == 0 && !endInclusive) {
  191. return nil
  192. }
  193. }
  194. if startKey != nil {
  195. cmp := composeCmp(key, startKey)
  196. if cmp < 0 || (cmp == 0 && !startInclusive) {
  197. continue
  198. }
  199. }
  200. pkID := entry[pg.el-8:]
  201. if !callback(pkID) {
  202. return nil
  203. }
  204. }
  205. }
  206. } else {
  207. for p := len(pages) - 1; p >= 0; p-- {
  208. pg := &pages[p]
  209. for i := pg.count - 1; i >= 0; i-- {
  210. entry := pg.entries[i*pg.el : (i+1)*pg.el]
  211. key := entry[:pg.el-8]
  212. if startKey != nil {
  213. cmp := composeCmp(key, startKey)
  214. if cmp < 0 || (cmp == 0 && !startInclusive) {
  215. return nil
  216. }
  217. }
  218. if endKey != nil {
  219. cmp := composeCmp(key, endKey)
  220. if cmp > 0 || (cmp == 0 && !endInclusive) {
  221. continue
  222. }
  223. }
  224. pkID := entry[pg.el-8:]
  225. if !callback(pkID) {
  226. return nil
  227. }
  228. }
  229. }
  230. }
  231. return nil
  232. }
  233. // composeCmp 瀛楄妭鍒囩墖姣旇緝
  234. func composeCmp(a, b []byte) int {
  235. ml := len(a)
  236. if len(b) < ml {
  237. ml = len(b)
  238. }
  239. for i := 0; i < ml; i++ {
  240. if a[i] < b[i] {
  241. return -1
  242. }
  243. if a[i] > b[i] {
  244. return 1
  245. }
  246. }
  247. if len(a) < len(b) {
  248. return -1
  249. }
  250. if len(a) > len(b) {
  251. return 1
  252. }
  253. return 0
  254. }
  255. func (h *ComposeChainHandler) deleteEntry(handle page.HandleDb, entryPage *uint32, indexKey []byte) error {
  256. dataEntry := h.getDataEntry()
  257. curPageID := *entryPage
  258. for curPageID != 0 {
  259. buf := make([]byte, page.IndexPageSize)
  260. if err := handle.ReadPage(curPageID, buf); err != nil {
  261. return err
  262. }
  263. hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
  264. entries := buf[page.IndexHeadSize : page.IndexHeadSize+int(hdr.Count)*int(hdr.EntryLen)]
  265. entryLen := int(hdr.EntryLen)
  266. err := dataEntry.Delete(handle, entries, entryLen, indexKey)
  267. if err == nil {
  268. hdr.Hdr.UseSize -= uint32(entryLen)
  269. _ = handle.SavePage(curPageID, buf)
  270. return nil
  271. }
  272. curPageID = hdr.Hdr.NextPage
  273. }
  274. return page.ErrNotFound
  275. }
  276. func (h *ComposeChainHandler) getDataEntry() IDateEntry {
  277. handler, _ := Compose.GetEntryHandler(page.IndexTypeCompose)
  278. return handler.DataEntry()
  279. }