time_handler.go 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833
  1. package index
  2. import (
  3. "TsDB2/DB/core/index/datapage"
  4. timeindex "TsDB2/DB/core/index/timedatapage"
  5. "TsDB2/DB/core/page"
  6. "TsDB2/DB/core/util"
  7. "encoding/binary"
  8. "errors"
  9. "unsafe"
  10. )
  11. func init() {
  12. Compose.RegisterHandler(page.IndexTypeTime, &TimeIndexHandler{})
  13. }
  14. // ======================== IEntry: 时间索引条目 ========================
  15. // IndexTimeEntryLen 固定长度 20
  16. const IndexTimeEntryLen = 20
  17. // IndexTimeEntry 时间索引条目:固定 20 字节
  18. type IndexTimeEntry struct {
  19. BeginTime int64 // 8
  20. LastTime int64 // 8
  21. PageID uint32 // 4
  22. }
  23. func (e *IndexTimeEntry) DataEntry() IDateEntry {
  24. return &timeindex.TimeDataPage{}
  25. }
  26. func (e *IndexTimeEntry) EntryLength() uint16 {
  27. return IndexTimeEntryLen
  28. }
  29. // Compare 比较两个索引的 BeginTime(给二分查找用)
  30. func (e *IndexTimeEntry) Compare(midEntry []byte, key []byte) int {
  31. timeMid := int64(binary.BigEndian.Uint64(midEntry[:8]))
  32. timeKey := int64(binary.BigEndian.Uint64(key[:8]))
  33. return util.CompareInt64(timeMid, timeKey)
  34. }
  35. // Serialize 序列化为 20 字节
  36. func (e *IndexTimeEntry) Serialize(ie IEntry) []byte {
  37. if en, ok := ie.(*IndexTimeEntry); ok {
  38. buf := make([]byte, IndexTimeEntryLen)
  39. binary.BigEndian.PutUint64(buf[:8], uint64(en.BeginTime))
  40. binary.BigEndian.PutUint64(buf[8:16], uint64(en.LastTime))
  41. binary.BigEndian.PutUint32(buf[16:20], en.PageID)
  42. return buf
  43. }
  44. return nil
  45. }
  46. // Deserialize 从 []byte 解析结构体
  47. func (e *IndexTimeEntry) Deserialize(buf []byte) IEntry {
  48. return &IndexTimeEntry{
  49. BeginTime: int64(binary.BigEndian.Uint64(buf[:8])),
  50. LastTime: int64(binary.BigEndian.Uint64(buf[8:16])),
  51. PageID: binary.BigEndian.Uint32(buf[16:20]),
  52. }
  53. }
  54. // Insert 向索引块尾部追加一个条目
  55. func (e *IndexTimeEntry) Insert(entries *[]byte, data []byte) error {
  56. *entries = append(*entries, data...)
  57. return nil
  58. }
  59. // Search 根据时间戳查找 PageID
  60. func (e *IndexTimeEntry) Search(entries []byte, key []byte) (uint32, bool) {
  61. pos, found := util.BinarySearch(entries, IndexTimeEntryLen, key, e.Compare)
  62. if !found {
  63. return 0, false
  64. }
  65. start := pos * IndexTimeEntryLen
  66. entryBuf := entries[start : start+IndexTimeEntryLen]
  67. pageID := binary.BigEndian.Uint32(entryBuf[16:20])
  68. return pageID, true
  69. }
  70. // ======================== IndexHandler: 时序索引处理器 ========================
  71. // TimeIndexHandler — 时序索引 (page.IndexTypeTime)
  72. // 扁平索引页链 + TimeDataPage
  73. type TimeIndexHandler struct{}
  74. func (h *TimeIndexHandler) Insert(handle page.HandleDb, entryPage *uint32, indexKey []byte, pkID uint64, rowData []byte, keyLen int) (uint32, error) {
  75. // 时序表只收时间戳非递减的写入。顺序写入才能保证每个分段覆盖一段互不重叠的
  76. // 时间区间、段内也天然有序 —— 范围查询才裁剪得动分段, 段内也不必整段逐行过滤。
  77. //
  78. // 顺便拿到索引链的尾页: 新分段的条目必须挂到链尾。挂到当前页会把它的 NextPage
  79. // 覆盖掉, 之前挂上去的索引页从此成为孤儿 —— 那些页里的分段整段查不到 (丢数据)。
  80. tailIdxPageID, err := h.assertTimeOrder(handle, *entryPage, indexKey)
  81. if err != nil {
  82. return 0, err
  83. }
  84. dataEntry := h.getDataEntry()
  85. // 顺序写入 (assertTimeOrder 已保证) 下新行的时间戳是全表最大的, 它归属的新分段
  86. // 只可能落在链尾页 —— 所以直接从链尾页开始找。
  87. //
  88. // 若沿用"永远从首页开始找": 首页的条目位写满之后, 每条新条目都会新建一张只放一条
  89. // 的索引页, 链无限增长 (而每次写入又要走完整条链去校验顺序), 同时新行会被塞进首页
  90. // 那个早就写满的旧分段 —— 段划分错乱, 段数暴涨。
  91. curPageID := *entryPage
  92. if tailIdxPageID != 0 {
  93. curPageID = tailIdxPageID
  94. }
  95. for curPageID != 0 {
  96. buf := make([]byte, page.IndexPageSize)
  97. if err := handle.ReadPage(curPageID, buf); err != nil {
  98. return 0, err
  99. }
  100. hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
  101. pageEntryLen := int(hdr.EntryLen)
  102. count := int(hdr.Count)
  103. idxPageHasRoom := hdr.Hdr.UseSize+uint32(pageEntryLen) <= page.IndexPageSize
  104. var entries []byte
  105. if idxPageHasRoom {
  106. entries = buf[page.IndexHeadSize : page.IndexHeadSize+(count+1)*pageEntryLen]
  107. } else {
  108. endOff := page.IndexHeadSize + count*pageEntryLen + 4
  109. if endOff <= len(buf) {
  110. entries = buf[page.IndexHeadSize:endOff]
  111. } else {
  112. overflow := make([]byte, count*pageEntryLen+4)
  113. copy(overflow, buf[page.IndexHeadSize:page.IndexHeadSize+count*pageEntryLen])
  114. entries = overflow
  115. }
  116. }
  117. err := dataEntry.Insert(handle, entries, pageEntryLen, count, indexKey, pkID, rowData)
  118. if err == nil {
  119. // 命中已有分段: 该分段条目的 LastTime 可能已被扩展, 需回写索引页。
  120. // 页内已满时 entries 是独立副本, 要把条目区拷回 buf 再落盘。
  121. if !idxPageHasRoom {
  122. copy(buf[page.IndexHeadSize:page.IndexHeadSize+count*pageEntryLen],
  123. entries[:count*pageEntryLen])
  124. }
  125. if err := handle.SavePage(curPageID, buf); err != nil {
  126. return 0, err
  127. }
  128. return *entryPage, nil
  129. }
  130. if errors.Is(err, timeindex.ErrNeedNewEntry) {
  131. dpID := binary.BigEndian.Uint32(entries[count*pageEntryLen : count*pageEntryLen+4])
  132. if idxPageHasRoom {
  133. insertEntrySorted(buf, hdr, buildTimeEntry(indexKey, dpID))
  134. if err := handle.SavePage(curPageID, buf); err != nil {
  135. return 0, err
  136. }
  137. return *entryPage, nil
  138. }
  139. return *entryPage, writeToNewIndexPage(handle, entryPage,
  140. buildTimeEntry(indexKey, dpID), tailIdxPageID)
  141. }
  142. curPageID = hdr.Hdr.NextPage
  143. }
  144. // 新分段的第一个 DataPage 必须拿到真实行号: 行存储索引的 DataPage 按 pkID
  145. // 增序排列, 前缀写错会让 FindRowByPK 的"遇到更大 pkID 就提前结束"立刻成立,
  146. // 整页行号从此不可达 (按行号读/改/删、二级索引回表统统失效)。
  147. newDPPID, createErr := dataEntry.CreateDataPage(handle, indexKey, pkID, rowData)
  148. if createErr != nil {
  149. return 0, createErr
  150. }
  151. return *entryPage, writeToNewIndexPage(handle, entryPage, buildTimeEntry(indexKey, newDPPID), tailIdxPageID)
  152. }
  153. // ======================== 顺序写入约束 ========================
  154. // ErrOutOfOrder 时序表只接受时间戳非递减的写入, 更早的时间戳会被拒绝。
  155. //
  156. // 时序索引只有时间这一个维度: 顺序写入时, 每个分段天然覆盖一段互不重叠的时间区间,
  157. // 段内行也天然按时间递增 —— 范围查询才能用 [BeginTime, LastTime] 裁剪掉无关分段。
  158. // 一旦允许乱序, 新行会落进"BeginTime <= ts 的最后一个分段", 而它后面的分段可能覆盖
  159. // 更晚的时间, 于是段区间互相重叠、段内时间也不再单调, 范围查询只能退化成全链 +
  160. // 整段逐行过滤。(确有乱序数据时, 先按时间排序再写, 或改用普通表 + 二级索引。)
  161. var ErrOutOfOrder = errors.New("time series: out-of-order timestamp rejected (must be >= the latest timestamp)")
  162. // assertTimeOrder 拒绝比"当前最大时间戳"更早的写入, 并返回索引链的尾页 ID
  163. // (新分段的条目要挂到链尾, 见 Insert 里的说明)。
  164. //
  165. // 最大时间戳取自索引页链**最后一个条目**的 LastTime: 条目在链上按 BeginTime 升序,
  166. // 新条目只会被追加到链尾 —— 首页装得下就追加在首页尾部, 装不下就新建一页挂在链尾,
  167. // 所以链尾页的最后一条正是最近一次写入的分段, 它的 LastTime 就是全表最大时间戳。
  168. //
  169. // 成本: 每次写入都要沿链走到尾, 代价与链长 (分段数 / 每页条目数) 成正比。顺序写入下
  170. // 分段增长缓慢, 链通常只有一两页; 若将来链变得很长, 应把最大时间戳持久化进索引元数据。
  171. func (h *TimeIndexHandler) assertTimeOrder(handle page.HandleDb, entryPage uint32, indexKey []byte) (uint32, error) {
  172. if entryPage == 0 {
  173. // 空表: 第一个条目自己就是首页, 没有链尾可挂。
  174. return 0, nil
  175. }
  176. maxTS, tail, ok := h.latestTimeKey(handle, entryPage)
  177. if ok && len(indexKey) >= 8 && int64(binary.BigEndian.Uint64(indexKey[:8])) < maxTS {
  178. return tail, ErrOutOfOrder
  179. }
  180. return tail, nil
  181. }
  182. // latestTimeKey 沿索引页链走到最后一页, 返回 (最后一条目的 LastTime, 尾页 ID, 是否有条目)。
  183. func (h *TimeIndexHandler) latestTimeKey(handle page.HandleDb, entryPage uint32) (int64, uint32, bool) {
  184. var (
  185. last int64
  186. tail uint32
  187. found bool
  188. )
  189. for curPID := entryPage; curPID != 0; {
  190. buf := make([]byte, page.IndexPageSize)
  191. if err := handle.ReadPage(curPID, buf); err != nil {
  192. break
  193. }
  194. tail = curPID
  195. hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
  196. entryLen := int(hdr.EntryLen)
  197. count := int(hdr.Count)
  198. if entryLen >= IndexTimeEntryLen && count > 0 {
  199. // 条目格式: [BeginTime(8B) + LastTime(8B) + DataPageID(4B)]
  200. entries := buf[page.IndexHeadSize : page.IndexHeadSize+count*entryLen]
  201. off := (count-1)*entryLen + 8
  202. last = int64(binary.BigEndian.Uint64(entries[off : off+8]))
  203. found = true
  204. }
  205. curPID = hdr.Hdr.NextPage
  206. }
  207. if tail == 0 {
  208. // 一页都没读成 (异常): 退回挂到首页, 总比让新页变成孤儿好。
  209. tail = entryPage
  210. }
  211. return last, tail, found
  212. }
  213. // pkIDBytes 把行号编成 8 字节 BigEndian —— DataPage 行首前缀与 FindRowByPK 的约定。
  214. func pkIDBytes(pkID uint64) []byte {
  215. buf := make([]byte, 8)
  216. binary.BigEndian.PutUint64(buf, pkID)
  217. return buf
  218. }
  219. // sameTimeKey 判断两个时间索引键是否指向同一时间点 (时序键定长 8 字节)。
  220. // 取不到键时保守返回 false (当作"时间变了", 让调用方走搬移路径)。
  221. func sameTimeKey(a, b []byte) bool {
  222. if len(a) < 8 || len(b) < 8 {
  223. return false
  224. }
  225. return tiCmp(a[:8], b[:8]) == 0
  226. }
  227. // findDataPageByPK 沿索引条目链定位"包含该行号"的 DataPage。
  228. //
  229. // 时间键只能定位到分段 (一个分段覆盖一段时间, 里面可能有多行), 行号才能定位到行。
  230. func (h *TimeIndexHandler) findDataPageByPK(handle page.HandleDb, entryPage uint32, pkID uint64) (uint32, bool) {
  231. if entryPage == 0 {
  232. return 0, false
  233. }
  234. pkKeyBE := pkIDBytes(pkID)
  235. for curPageID := entryPage; curPageID != 0; {
  236. buf := make([]byte, page.IndexPageSize)
  237. if err := handle.ReadPage(curPageID, buf); err != nil {
  238. break
  239. }
  240. hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
  241. entryLen := int(hdr.EntryLen)
  242. count := int(hdr.Count)
  243. if entryLen <= 0 || count == 0 {
  244. curPageID = hdr.Hdr.NextPage
  245. continue
  246. }
  247. entries := buf[page.IndexHeadSize : page.IndexHeadSize+count*entryLen]
  248. for i := 0; i < count; i++ {
  249. dpID := binary.BigEndian.Uint32(entries[i*entryLen+entryLen-4:])
  250. if dpID == 0 {
  251. continue
  252. }
  253. dp, err := datapage.Load(handle, dpID)
  254. if err != nil {
  255. continue
  256. }
  257. if _, ok := dp.FindRowByPK(pkKeyBE); ok {
  258. return dpID, true
  259. }
  260. }
  261. curPageID = hdr.Hdr.NextPage
  262. }
  263. return 0, false
  264. }
  265. // buildTimeEntry 构造 20 字节时间索引条目 [BeginTime(8) + LastTime(8) + DataPageID(4)]。
  266. // LastTime 初始等于 BeginTime, 之后随该分段被追加更大的时间戳而增长。
  267. func buildTimeEntry(indexKey []byte, dpID uint32) []byte {
  268. entry := make([]byte, IndexTimeEntryLen)
  269. copy(entry[:8], indexKey[:8])
  270. copy(entry[8:16], indexKey[:8])
  271. binary.BigEndian.PutUint32(entry[16:20], dpID)
  272. return entry
  273. }
  274. func (h *TimeIndexHandler) Search(handle page.HandleDb, entryPage uint32, indexKey []byte, keyLen int, resolve RowResolver) ([]byte, bool) {
  275. // 时序索引自己就是行数据所在, 不需要回表
  276. dataEntry := h.getDataEntry()
  277. curPageID := entryPage
  278. for curPageID != 0 {
  279. buf := make([]byte, page.IndexPageSize)
  280. if err := handle.ReadPage(curPageID, buf); err != nil {
  281. break
  282. }
  283. hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
  284. entries := buf[page.IndexHeadSize : page.IndexHeadSize+int(hdr.Count)*int(hdr.EntryLen)]
  285. entryLen := int(hdr.EntryLen)
  286. rowData, found := dataEntry.Search(handle, entries, entryLen, indexKey)
  287. if found {
  288. return rowData, true
  289. }
  290. curPageID = hdr.Hdr.NextPage
  291. }
  292. return nil, false
  293. }
  294. func (h *TimeIndexHandler) Delete(handle page.HandleDb, entryPage *uint32, indexKey []byte, pkID uint64, keyLen int) error {
  295. // 行号已知 → 按行号删除"这一行"。
  296. //
  297. // 同一时间戳下允许有多行 (同一时刻的多个测点/多次事件), 只按时间键删只会删掉
  298. // 该时刻的第一行, 其余行变成"看得见却删不掉"的幽灵数据。行号是每行唯一的,
  299. // 按它定位才能删到调用方指定的那一行。
  300. if pkID != 0 {
  301. dpID, ok := h.findDataPageByPK(handle, *entryPage, pkID)
  302. if !ok {
  303. return page.ErrNotFound
  304. }
  305. if !h.DeleteDataPageByPK(handle, dpID, pkIDBytes(pkID)) {
  306. return page.ErrNotFound
  307. }
  308. // 有意不回收"整页已删空"的 DataPage: 索引条目仍指向它, 立即还链会让该条目
  309. // 指向一张可能被别的索引复用的页 (全表扫描会把别人的行当成本表的行读出来)。
  310. // 空页留在原地无害, 之后往同一分段插入时会被继续复用。
  311. return nil
  312. }
  313. dataEntry := h.getDataEntry()
  314. curPageID := *entryPage
  315. for curPageID != 0 {
  316. buf := make([]byte, page.IndexPageSize)
  317. if err := handle.ReadPage(curPageID, buf); err != nil {
  318. return err
  319. }
  320. hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
  321. entries := buf[page.IndexHeadSize : page.IndexHeadSize+int(hdr.Count)*int(hdr.EntryLen)]
  322. entryLen := int(hdr.EntryLen)
  323. err := dataEntry.Delete(handle, entries, entryLen, indexKey)
  324. if err == nil {
  325. return nil
  326. }
  327. curPageID = hdr.Hdr.NextPage
  328. }
  329. return page.ErrNotFound
  330. }
  331. // IsRowStore 时序索引直接承载行数据
  332. func (h *TimeIndexHandler) IsRowStore() bool { return true }
  333. // ResolveRow 遍历索引项链定位各个 TimeDataPage, 按 pkID 取行
  334. func (h *TimeIndexHandler) ResolveRow(handle page.HandleDb, entryPage uint32, pkID uint64, cache *[]DPRange) ([]byte, bool) {
  335. pkKeyBE := make([]byte, 8)
  336. binary.BigEndian.PutUint64(pkKeyBE, pkID)
  337. curPageID := entryPage
  338. for curPageID != 0 {
  339. buf := make([]byte, page.IndexPageSize)
  340. if err := handle.ReadPage(curPageID, buf); err != nil {
  341. break
  342. }
  343. hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
  344. entryLen := int(hdr.EntryLen)
  345. count := int(hdr.Count)
  346. if entryLen <= 0 || count == 0 {
  347. curPageID = hdr.Hdr.NextPage
  348. continue
  349. }
  350. entries := buf[page.IndexHeadSize : page.IndexHeadSize+count*entryLen]
  351. for i := 0; i < count; i++ {
  352. dpID := binary.BigEndian.Uint32(entries[i*entryLen+entryLen-4:])
  353. if dpID == 0 {
  354. continue
  355. }
  356. dp, err := datapage.Load(handle, dpID)
  357. if err != nil {
  358. continue
  359. }
  360. if row, ok := dp.FindRowByPK(pkKeyBE); ok {
  361. return row, true
  362. }
  363. }
  364. curPageID = hdr.Hdr.NextPage
  365. }
  366. return nil, false
  367. }
  368. func (h *TimeIndexHandler) Update(handle page.HandleDb, entryPage *uint32, oldKey, newKey, rowData []byte, pkID uint64, keyLen int) error {
  369. // 行号已知 → 按行号改写"这一行"; 时间键相同的多行因此能各自独立更新。
  370. if pkID != 0 {
  371. return h.updateByPK(handle, entryPage, oldKey, newKey, rowData, pkID)
  372. }
  373. dataEntry := h.getDataEntry()
  374. curPageID := *entryPage
  375. for curPageID != 0 {
  376. buf := make([]byte, page.IndexPageSize)
  377. if err := handle.ReadPage(curPageID, buf); err != nil {
  378. return err
  379. }
  380. hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
  381. entries := buf[page.IndexHeadSize : page.IndexHeadSize+int(hdr.Count)*int(hdr.EntryLen)]
  382. entryLen := int(hdr.EntryLen)
  383. err := dataEntry.Update(handle, entries, entryLen, oldKey, pkID, rowData)
  384. if err == nil {
  385. return nil
  386. }
  387. if errors.Is(err, page.ErrNotFound) {
  388. curPageID = hdr.Hdr.NextPage
  389. continue
  390. }
  391. return err
  392. }
  393. return page.ErrNotFound
  394. }
  395. // updateByPK 按行号精确定位并改写一行。
  396. //
  397. // 分三种情形:
  398. // 1. 时间键没变 → 分段内原地改写; 行变长装不下时先在页内紧凑重建;
  399. // 2. 时间键变了 → 行必须换到新时间段: 按行号删掉旧行, 再用同一个行号重新插入;
  400. // 3. 页内重建仍装不下 → 同 2, 搬走重插 (不改变行号)。
  401. func (h *TimeIndexHandler) updateByPK(handle page.HandleDb, entryPage *uint32, oldKey, newKey, rowData []byte, pkID uint64) error {
  402. if len(newKey) < 8 {
  403. // 时间索引的键必须是 8 字节时间戳; 取不到键说明行本身不可解析, 不能猜着改。
  404. return page.ErrNotFound
  405. }
  406. pkKeyBE := pkIDBytes(pkID)
  407. dpID, ok := h.findDataPageByPK(handle, *entryPage, pkID)
  408. if !ok {
  409. return page.ErrNotFound
  410. }
  411. storeRow := make([]byte, 8+len(rowData))
  412. copy(storeRow[:8], pkKeyBE)
  413. copy(storeRow[8:], rowData)
  414. move := func() error {
  415. if !h.DeleteDataPageByPK(handle, dpID, pkKeyBE) {
  416. return page.ErrNotFound
  417. }
  418. _, err := h.Insert(handle, entryPage, newKey, pkID, rowData, len(newKey))
  419. return err
  420. }
  421. if !sameTimeKey(oldKey, newKey) {
  422. // 先验顺序再删旧行: move 是"先删后插", 顺序不合法要在这里就拦下 ——
  423. // 否则旧行已删、新行被拒, 这一行就凭空消失了。
  424. if _, err := h.assertTimeOrder(handle, *entryPage, newKey); err != nil {
  425. return err
  426. }
  427. return move()
  428. }
  429. if h.UpdateDataPageByPK(handle, dpID, pkKeyBE, storeRow) {
  430. return nil
  431. }
  432. // 原地装不下 (行长变大): 页内重排, 把墓碑的碎片空间收回来。
  433. if dp, err := datapage.Load(handle, dpID); err == nil {
  434. if err := dp.RebuildWith(pkKeyBE, storeRow); err == nil {
  435. return nil
  436. }
  437. }
  438. // 整页都放不下 → 搬到新页, 行号保持不变。
  439. return move()
  440. }
  441. func (h *TimeIndexHandler) GetDataPageIDs(handle page.HandleDb, entryPage uint32) []uint32 {
  442. if entryPage == 0 {
  443. return nil
  444. }
  445. var ids []uint32
  446. curPageID := entryPage
  447. for curPageID != 0 {
  448. buf := make([]byte, page.IndexPageSize)
  449. if err := handle.ReadPage(curPageID, buf); err != nil {
  450. break
  451. }
  452. hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
  453. entryLen := int(hdr.EntryLen)
  454. count := int(hdr.Count)
  455. entries := buf[page.IndexHeadSize : page.IndexHeadSize+count*entryLen]
  456. // 时间索引条目格式: [BeginTime(8B) + LastTime(8B) + DataPageID(4B)]
  457. for i := 0; i < count; i++ {
  458. dpID := binary.BigEndian.Uint32(entries[i*entryLen+entryLen-4:])
  459. if dpID != 0 {
  460. ids = append(ids, dpID)
  461. }
  462. }
  463. curPageID = hdr.Hdr.NextPage
  464. }
  465. return ids
  466. }
  467. // CollectPageIDs 汇报时序索引当前占用的全部页: 索引条目链上的每一页, 加上每个
  468. // 时间段指向的 DataPage (行真正落在这里)。
  469. //
  470. // 时序索引的页是"扁平 IndexHead 链", 既不是 B+ 节点也不是哈希桶, 所以自己枚举才准确。
  471. // 不能指望 btree.CollectAllPageIDs: 它把页当 B+ 节点解析, 这里 IsLeaf=1 只会返回链首
  472. // 一页 —— 单页可容 817 段, 实测插到 6 万行仍压在一页内, 所以链页尚未见泄漏; 但行数
  473. // 继续增长 (每页约 400~500 段) 后超出部分就会成为不可达页。
  474. // 数据页此前是靠 btree.GetDataPageIDs 顺带收回的: 它把 NextPage 当链指针, 又把每个
  475. // 条目末 4 字节当 DataPageID, 恰好与 [BeginTime/LastTime/DataPageID] 的布局撞对。
  476. // 这里改为显式调用自己的 GetDataPageIDs, 不再依赖这种巧合。
  477. func (h *TimeIndexHandler) CollectPageIDs(handle page.HandleDb, entryPage uint32) []uint32 {
  478. if entryPage == 0 {
  479. return nil
  480. }
  481. var ids []uint32
  482. seen := make(map[uint32]bool)
  483. for curPageID := entryPage; curPageID != 0 && !seen[curPageID]; {
  484. seen[curPageID] = true
  485. ids = append(ids, curPageID)
  486. buf := make([]byte, page.IndexPageSize)
  487. if err := handle.ReadPage(curPageID, buf); err != nil {
  488. break
  489. }
  490. hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
  491. curPageID = hdr.Hdr.NextPage
  492. }
  493. for _, dpID := range h.GetDataPageIDs(handle, entryPage) {
  494. if dpID != 0 && !seen[dpID] {
  495. seen[dpID] = true
  496. ids = append(ids, dpID)
  497. }
  498. }
  499. return ids
  500. }
  501. func (h *TimeIndexHandler) ReadDataPageRows(handle page.HandleDb, dpID uint32) [][]byte {
  502. dp, err := datapage.Load(handle, dpID)
  503. if err != nil {
  504. return nil
  505. }
  506. return dp.GetAllRows()
  507. }
  508. func (h *TimeIndexHandler) getDataEntry() IDateEntry {
  509. handler, _ := Compose.GetEntryHandler(page.IndexTypeTime)
  510. return handler.DataEntry()
  511. }
  512. // GetDataPageIDs 已存在于上方 (L145)
  513. // SearchDataPageByPK 在指定 DataPage 中按 pkID(LE) 查找行
  514. func (h *TimeIndexHandler) SearchDataPageByPK(handle page.HandleDb, dpID uint32, pkKeyBE []byte) ([]byte, bool) {
  515. buf := make([]byte, page.IndexPageSize)
  516. if err := handle.ReadPage(dpID, buf); err != nil {
  517. return nil, false
  518. }
  519. // DataPage 格式: [page.DataPageHdr(40B)] + [len(4B) + rowData]...
  520. dh := page.DataHdrOf(buf)
  521. hdrSize := page.DataPageHdrSize
  522. count := dh.Count
  523. dataEnd := dh.UseSize
  524. keyPK := binary.BigEndian.Uint64(pkKeyBE[:8])
  525. off := hdrSize
  526. for i := uint32(0); i < count; i++ {
  527. if off+4 > dataEnd {
  528. break
  529. }
  530. rawLen := binary.LittleEndian.Uint32(buf[off:])
  531. deleted := rawLen&0x80000000 != 0
  532. rowLen := rawLen & 0x7FFFFFFF
  533. off += 4
  534. if off+rowLen > dataEnd {
  535. break
  536. }
  537. if !deleted && rowLen >= 8 {
  538. rowPK := binary.BigEndian.Uint64(buf[off : off+8])
  539. if rowPK == keyPK {
  540. result := make([]byte, rowLen)
  541. copy(result, buf[off:off+rowLen])
  542. return result, true
  543. }
  544. }
  545. off += rowLen
  546. }
  547. return nil, false
  548. }
  549. // DeleteDataPageByPK 在指定 DataPage 中按 pkID(LE) 软删除行, 返回是否找到
  550. func (h *TimeIndexHandler) DeleteDataPageByPK(handle page.HandleDb, dpID uint32, pkKeyBE []byte) bool {
  551. buf := make([]byte, page.IndexPageSize)
  552. if err := handle.ReadPage(dpID, buf); err != nil {
  553. return false
  554. }
  555. dh := page.DataHdrOf(buf)
  556. hdrSize := page.DataPageHdrSize
  557. count := dh.Count
  558. dataEnd := dh.UseSize
  559. keyPK := binary.BigEndian.Uint64(pkKeyBE[:8])
  560. off := hdrSize
  561. for i := uint32(0); i < count; i++ {
  562. if off+4 > dataEnd {
  563. break
  564. }
  565. rawLen := binary.LittleEndian.Uint32(buf[off:])
  566. deleted := rawLen&0x80000000 != 0
  567. rowLen := rawLen & 0x7FFFFFFF
  568. off += 4
  569. if off+rowLen > dataEnd {
  570. break
  571. }
  572. if !deleted && rowLen >= 8 {
  573. rowPK := binary.BigEndian.Uint64(buf[off : off+8])
  574. if rowPK == keyPK {
  575. binary.LittleEndian.PutUint32(buf[off-4:], rowLen|0x80000000)
  576. // 直写路径绕过 datapage.DataPage, 页头的存活/失效统计在这里同步维护
  577. dh.LiveCnt--
  578. dh.Dead += 4 + rowLen
  579. _ = handle.SavePage(dpID, buf)
  580. return true
  581. }
  582. }
  583. off += rowLen
  584. }
  585. return false
  586. }
  587. // UpdateDataPageByPK 在指定 DataPage 中按 pkID(LE) 原地更新行, 返回是否找到
  588. func (h *TimeIndexHandler) UpdateDataPageByPK(handle page.HandleDb, dpID uint32, pkKeyBE []byte, newStoreRow []byte) bool {
  589. buf := make([]byte, page.IndexPageSize)
  590. if err := handle.ReadPage(dpID, buf); err != nil {
  591. return false
  592. }
  593. dh := page.DataHdrOf(buf)
  594. hdrSize := page.DataPageHdrSize
  595. count := dh.Count
  596. dataEnd := dh.UseSize
  597. keyPK := binary.BigEndian.Uint64(pkKeyBE[:8])
  598. off := hdrSize
  599. for i := uint32(0); i < count; i++ {
  600. if off+4 > dataEnd {
  601. break
  602. }
  603. rawLen := binary.LittleEndian.Uint32(buf[off:])
  604. deleted := rawLen&0x80000000 != 0
  605. rowLen := rawLen & 0x7FFFFFFF
  606. off += 4
  607. if off+rowLen > dataEnd {
  608. break
  609. }
  610. if !deleted && rowLen >= 8 {
  611. rowPK := binary.BigEndian.Uint64(buf[off : off+8])
  612. if rowPK == keyPK {
  613. if uint32(len(newStoreRow)) <= rowLen {
  614. copy(buf[off:], newStoreRow)
  615. for j := uint32(len(newStoreRow)); j < rowLen; j++ {
  616. buf[off+j] = 0
  617. }
  618. _ = handle.SavePage(dpID, buf)
  619. return true
  620. }
  621. return false // row too large
  622. }
  623. }
  624. off += rowLen
  625. }
  626. return false
  627. }
  628. // RangeScan 时序索引范围扫描
  629. // startKey/endKey: BigEndian 编码的时间戳
  630. // 遍历 IndexPage 条目链 对时间范围与查询范围重叠的条目加载 DataPage 并回调每行
  631. func (h *TimeIndexHandler) RangeScan(handle page.HandleDb, entryPage uint32,
  632. startKey, endKey []byte, startInclusive, endInclusive bool, reverse bool,
  633. callback func(row []byte) bool) error {
  634. if entryPage == 0 {
  635. return nil
  636. }
  637. // 收集所有索引页数据
  638. type pageInfo struct {
  639. entries []byte
  640. entryLen int
  641. count int
  642. }
  643. var pages []pageInfo
  644. curPageID := entryPage
  645. for curPageID != 0 {
  646. buf := make([]byte, page.IndexPageSize)
  647. if err := handle.ReadPage(curPageID, buf); err != nil {
  648. break
  649. }
  650. hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
  651. entryLen := int(hdr.EntryLen)
  652. count := int(hdr.Count)
  653. entries := make([]byte, count*entryLen)
  654. copy(entries, buf[page.IndexHeadSize:page.IndexHeadSize+count*entryLen])
  655. pages = append(pages, pageInfo{entries: entries, entryLen: entryLen, count: count})
  656. curPageID = hdr.Hdr.NextPage
  657. }
  658. // 确定遍历顺序
  659. pageStart, pageEnd, pageStep := 0, len(pages)-1, 1
  660. if reverse {
  661. pageStart, pageEnd, pageStep = len(pages)-1, 0, -1
  662. }
  663. visited := make(map[uint32]bool)
  664. for p := pageStart; ; p += pageStep {
  665. if pageStep > 0 && p > pageEnd {
  666. break
  667. }
  668. if pageStep < 0 && p < pageEnd {
  669. break
  670. }
  671. pg := &pages[p]
  672. if pg.entryLen < IndexTimeEntryLen {
  673. continue // 防御: 旧库/损坏页的条目长度不足 20B, 跳过以免越界
  674. }
  675. entryStart, entryEnd, entryStep := 0, pg.count-1, 1
  676. if reverse {
  677. entryStart, entryEnd, entryStep = pg.count-1, 0, -1
  678. }
  679. for i := entryStart; ; i += entryStep {
  680. if entryStep > 0 && i > entryEnd {
  681. break
  682. }
  683. if entryStep < 0 && i < entryEnd {
  684. break
  685. }
  686. off := i * pg.entryLen
  687. beginTime := pg.entries[off : off+8]
  688. lastTime := pg.entries[off+8 : off+16]
  689. dpID := binary.BigEndian.Uint32(pg.entries[off+pg.entryLen-4:])
  690. if dpID == 0 || visited[dpID] {
  691. continue
  692. }
  693. if endKey != nil {
  694. cmp := tiCmp(beginTime, endKey)
  695. if cmp > 0 || (cmp == 0 && !endInclusive) {
  696. continue
  697. }
  698. }
  699. if startKey != nil {
  700. cmp := tiCmp(lastTime, startKey)
  701. if cmp < 0 || (cmp == 0 && !startInclusive) {
  702. continue
  703. }
  704. }
  705. visited[dpID] = true
  706. dp, dpErr := datapage.Load(handle, dpID)
  707. if dpErr != nil {
  708. continue
  709. }
  710. rows := dp.GetAllRows()
  711. if !reverse {
  712. for _, row := range rows {
  713. if !timeRowInRange(row, startKey, endKey, startInclusive, endInclusive) {
  714. continue
  715. }
  716. if !callback(row) {
  717. return nil
  718. }
  719. }
  720. } else {
  721. for k := len(rows) - 1; k >= 0; k-- {
  722. if !timeRowInRange(rows[k], startKey, endKey, startInclusive, endInclusive) {
  723. continue
  724. }
  725. if !callback(rows[k]) {
  726. return nil
  727. }
  728. }
  729. }
  730. }
  731. }
  732. return nil
  733. }
  734. // timeRowInRange 判断时序数据行的时间是否落在 [startKey, endKey] 区间内。
  735. // 行内布局: [pkID(8B)] + [schemaVer(1B)] + 字段..., 时间键位于偏移 9(8B BE)。
  736. // 由于分段 DataPage 覆盖一段时间跨度, 段内可能混有区间外的点, 必须逐行过滤,
  737. // 否则范围查询会把整段的点都返回 (含越界点)。
  738. func timeRowInRange(row, startKey, endKey []byte, startInclusive, endInclusive bool) bool {
  739. const tsOff = 9
  740. if len(row) < tsOff+8 {
  741. // 行过短无法解析时间, 保守返回 true, 交由上层判断
  742. return true
  743. }
  744. if startKey == nil && endKey == nil {
  745. return true
  746. }
  747. ts := row[tsOff : tsOff+8]
  748. if startKey != nil {
  749. cmp := tiCmp(ts, startKey)
  750. if cmp < 0 || (cmp == 0 && !startInclusive) {
  751. return false
  752. }
  753. }
  754. if endKey != nil {
  755. cmp := tiCmp(ts, endKey)
  756. if cmp > 0 || (cmp == 0 && !endInclusive) {
  757. return false
  758. }
  759. }
  760. return true
  761. }
  762. // tiCmp 字节切片比较
  763. func tiCmp(a, b []byte) int {
  764. ml := len(a)
  765. if len(b) < ml {
  766. ml = len(b)
  767. }
  768. for i := 0; i < ml; i++ {
  769. if a[i] < b[i] {
  770. return -1
  771. }
  772. if a[i] > b[i] {
  773. return 1
  774. }
  775. }
  776. if len(a) < len(b) {
  777. return -1
  778. }
  779. if len(a) > len(b) {
  780. return 1
  781. }
  782. return 0
  783. }