time_handler.go 30 KB

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