| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929 |
- package index
- import (
- "git.ycl.pub/yl32940/TsDB2/DB/core/index/datapage"
- timeindex "git.ycl.pub/yl32940/TsDB2/DB/core/index/timedatapage"
- "git.ycl.pub/yl32940/TsDB2/DB/core/page"
- "git.ycl.pub/yl32940/TsDB2/DB/core/util"
- "encoding/binary"
- "errors"
- "sort"
- "unsafe"
- )
- func init() {
- Compose.RegisterHandler(page.IndexTypeTime, &TimeIndexHandler{})
- }
- // ======================== IEntry: 时间索引条目 ========================
- // IndexTimeEntryLen 固定长度 20
- const IndexTimeEntryLen = 20
- // IndexTimeEntry 时间索引条目:固定 20 字节
- type IndexTimeEntry struct {
- BeginTime int64 // 8
- LastTime int64 // 8
- PageID uint32 // 4
- }
- func (e *IndexTimeEntry) DataEntry() IDateEntry {
- return &timeindex.TimeDataPage{}
- }
- func (e *IndexTimeEntry) EntryLength() uint16 {
- return IndexTimeEntryLen
- }
- // Compare 比较两个索引的 BeginTime(给二分查找用)
- func (e *IndexTimeEntry) Compare(midEntry []byte, key []byte) int {
- timeMid := int64(binary.BigEndian.Uint64(midEntry[:8]))
- timeKey := int64(binary.BigEndian.Uint64(key[:8]))
- return util.CompareInt64(timeMid, timeKey)
- }
- // Serialize 序列化为 20 字节
- func (e *IndexTimeEntry) Serialize(ie IEntry) []byte {
- if en, ok := ie.(*IndexTimeEntry); ok {
- buf := make([]byte, IndexTimeEntryLen)
- binary.BigEndian.PutUint64(buf[:8], uint64(en.BeginTime))
- binary.BigEndian.PutUint64(buf[8:16], uint64(en.LastTime))
- binary.BigEndian.PutUint32(buf[16:20], en.PageID)
- return buf
- }
- return nil
- }
- // Deserialize 从 []byte 解析结构体
- func (e *IndexTimeEntry) Deserialize(buf []byte) IEntry {
- return &IndexTimeEntry{
- BeginTime: int64(binary.BigEndian.Uint64(buf[:8])),
- LastTime: int64(binary.BigEndian.Uint64(buf[8:16])),
- PageID: binary.BigEndian.Uint32(buf[16:20]),
- }
- }
- // Insert 向索引块尾部追加一个条目
- func (e *IndexTimeEntry) Insert(entries *[]byte, data []byte) error {
- *entries = append(*entries, data...)
- return nil
- }
- // Search 根据时间戳查找 PageID
- func (e *IndexTimeEntry) Search(entries []byte, key []byte) (uint32, bool) {
- pos, found := util.BinarySearch(entries, IndexTimeEntryLen, key, e.Compare)
- if !found {
- return 0, false
- }
- start := pos * IndexTimeEntryLen
- entryBuf := entries[start : start+IndexTimeEntryLen]
- pageID := binary.BigEndian.Uint32(entryBuf[16:20])
- return pageID, true
- }
- // ======================== IndexHandler: 时序索引处理器 ========================
- // TimeIndexHandler — 时序索引 (page.IndexTypeTime)
- // 扁平索引页链 + TimeDataPage
- type TimeIndexHandler struct{}
- func (h *TimeIndexHandler) Insert(handle page.HandleDb, entryPage *uint32, indexKey []byte, pkID uint64, rowData []byte, keyLen int) (uint32, error) {
- // 时序表只收时间戳非递减的写入。顺序写入才能保证每个分段覆盖一段互不重叠的
- // 时间区间、段内也天然有序 —— 范围查询才裁剪得动分段, 段内也不必整段逐行过滤。
- //
- // 顺便拿到索引链的尾页: 新分段的条目必须挂到链尾。挂到当前页会把它的 NextPage
- // 覆盖掉, 之前挂上去的索引页从此成为孤儿 —— 那些页里的分段整段查不到 (丢数据)。
- tailIdxPageID, err := h.assertTimeOrder(handle, *entryPage, indexKey)
- if err != nil {
- return 0, err
- }
- dataEntry := h.getDataEntry()
- // 顺序写入 (assertTimeOrder 已保证) 下新行的时间戳是全表最大的, 它归属的新分段
- // 只可能落在链尾页 —— 所以直接从链尾页开始找。
- //
- // 若沿用"永远从首页开始找": 首页的条目位写满之后, 每条新条目都会新建一张只放一条
- // 的索引页, 链无限增长 (而每次写入又要走完整条链去校验顺序), 同时新行会被塞进首页
- // 那个早就写满的旧分段 —— 段划分错乱, 段数暴涨。
- curPageID := *entryPage
- if tailIdxPageID != 0 {
- curPageID = tailIdxPageID
- }
- for curPageID != 0 {
- buf := make([]byte, page.IndexPageSize)
- if err := handle.ReadPage(curPageID, buf); err != nil {
- return 0, err
- }
- hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
- pageEntryLen := int(hdr.EntryLen)
- count := int(hdr.Count)
- idxPageHasRoom := hdr.Hdr.UseSize+uint32(pageEntryLen) <= page.IndexPageSize
- var entries []byte
- if idxPageHasRoom {
- entries = buf[page.IndexHeadSize : page.IndexHeadSize+(count+1)*pageEntryLen]
- } else {
- endOff := page.IndexHeadSize + count*pageEntryLen + 4
- if endOff <= len(buf) {
- entries = buf[page.IndexHeadSize:endOff]
- } else {
- overflow := make([]byte, count*pageEntryLen+4)
- copy(overflow, buf[page.IndexHeadSize:page.IndexHeadSize+count*pageEntryLen])
- entries = overflow
- }
- }
- err := dataEntry.Insert(handle, entries, pageEntryLen, count, indexKey, pkID, rowData)
- if err == nil {
- // 命中已有分段: 该分段条目的 LastTime 可能已被扩展, 需回写索引页。
- // 页内已满时 entries 是独立副本, 要把条目区拷回 buf 再落盘。
- if !idxPageHasRoom {
- copy(buf[page.IndexHeadSize:page.IndexHeadSize+count*pageEntryLen],
- entries[:count*pageEntryLen])
- }
- if err := handle.SavePage(curPageID, buf); err != nil {
- return 0, err
- }
- return *entryPage, nil
- }
- if errors.Is(err, timeindex.ErrNeedNewEntry) {
- dpID := binary.BigEndian.Uint32(entries[count*pageEntryLen : count*pageEntryLen+4])
- if idxPageHasRoom {
- insertEntrySorted(buf, hdr, buildTimeEntry(indexKey, dpID))
- if err := handle.SavePage(curPageID, buf); err != nil {
- return 0, err
- }
- return *entryPage, nil
- }
- return *entryPage, writeToNewIndexPage(handle, entryPage,
- buildTimeEntry(indexKey, dpID), tailIdxPageID)
- }
- curPageID = hdr.Hdr.NextPage
- }
- // 新分段的第一个 DataPage 必须拿到真实行号: 行存储索引的 DataPage 按 pkID
- // 增序排列, 前缀写错会让 FindRowByPK 的"遇到更大 pkID 就提前结束"立刻成立,
- // 整页行号从此不可达 (按行号读/改/删、二级索引回表统统失效)。
- newDPPID, createErr := dataEntry.CreateDataPage(handle, indexKey, pkID, rowData)
- if createErr != nil {
- return 0, createErr
- }
- return *entryPage, writeToNewIndexPage(handle, entryPage, buildTimeEntry(indexKey, newDPPID), tailIdxPageID)
- }
- // ======================== 顺序写入约束 ========================
- // ErrOutOfOrder 时序表只接受时间戳非递减的写入, 更早的时间戳会被拒绝。
- //
- // 时序索引只有时间这一个维度: 顺序写入时, 每个分段天然覆盖一段互不重叠的时间区间,
- // 段内行也天然按时间递增 —— 范围查询才能用 [BeginTime, LastTime] 裁剪掉无关分段。
- // 一旦允许乱序, 新行会落进"BeginTime <= ts 的最后一个分段", 而它后面的分段可能覆盖
- // 更晚的时间, 于是段区间互相重叠、段内时间也不再单调, 范围查询只能退化成全链 +
- // 整段逐行过滤。(确有乱序数据时, 先按时间排序再写, 或改用普通表 + 二级索引。)
- var ErrOutOfOrder = errors.New("time series: out-of-order timestamp rejected (must be >= the latest timestamp)")
- // assertTimeOrder 拒绝比"当前最大时间戳"更早的写入, 并返回索引链的尾页 ID
- // (新分段的条目要挂到链尾, 见 Insert 里的说明)。
- //
- // 最大时间戳取自索引页链**最后一个条目**的 LastTime: 条目在链上按 BeginTime 升序,
- // 新条目只会被追加到链尾 —— 首页装得下就追加在首页尾部, 装不下就新建一页挂在链尾,
- // 所以链尾页的最后一条正是最近一次写入的分段, 它的 LastTime 就是全表最大时间戳。
- //
- // 成本: 每次写入都要沿链走到尾, 代价与链长 (分段数 / 每页条目数) 成正比。顺序写入下
- // 分段增长缓慢, 链通常只有一两页; 若将来链变得很长, 应把最大时间戳持久化进索引元数据。
- func (h *TimeIndexHandler) assertTimeOrder(handle page.HandleDb, entryPage uint32, indexKey []byte) (uint32, error) {
- if entryPage == 0 {
- // 空表: 第一个条目自己就是首页, 没有链尾可挂。
- return 0, nil
- }
- maxTS, tail, ok := h.latestTimeKey(handle, entryPage)
- if ok && len(indexKey) >= 8 && int64(binary.BigEndian.Uint64(indexKey[:8])) < maxTS {
- return tail, ErrOutOfOrder
- }
- return tail, nil
- }
- // latestTimeKey 沿索引页链走到最后一页, 返回 (最后一条目的 LastTime, 尾页 ID, 是否有条目)。
- func (h *TimeIndexHandler) latestTimeKey(handle page.HandleDb, entryPage uint32) (int64, uint32, bool) {
- var (
- last int64
- tail uint32
- found bool
- )
- for curPID := entryPage; curPID != 0; {
- buf := make([]byte, page.IndexPageSize)
- if err := handle.ReadPage(curPID, buf); err != nil {
- break
- }
- tail = curPID
- hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
- entryLen := int(hdr.EntryLen)
- count := int(hdr.Count)
- if entryLen >= IndexTimeEntryLen && count > 0 {
- // 条目格式: [BeginTime(8B) + LastTime(8B) + DataPageID(4B)]
- entries := buf[page.IndexHeadSize : page.IndexHeadSize+count*entryLen]
- off := (count-1)*entryLen + 8
- last = int64(binary.BigEndian.Uint64(entries[off : off+8]))
- found = true
- }
- curPID = hdr.Hdr.NextPage
- }
- if tail == 0 {
- // 一页都没读成 (异常): 退回挂到首页, 总比让新页变成孤儿好。
- tail = entryPage
- }
- return last, tail, found
- }
- // pkIDBytes 把行号编成 8 字节 BigEndian —— DataPage 行首前缀与 FindRowByPK 的约定。
- func pkIDBytes(pkID uint64) []byte {
- buf := make([]byte, 8)
- binary.BigEndian.PutUint64(buf, pkID)
- return buf
- }
- // sameTimeKey 判断两个时间索引键是否指向同一时间点 (时序键定长 8 字节)。
- // 取不到键时保守返回 false (当作"时间变了", 让调用方走搬移路径)。
- func sameTimeKey(a, b []byte) bool {
- if len(a) < 8 || len(b) < 8 {
- return false
- }
- return tiCmp(a[:8], b[:8]) == 0
- }
- // findDataPageByPK 沿索引条目链定位"包含该行号"的 DataPage。
- //
- // 时间键只能定位到分段 (一个分段覆盖一段时间, 里面可能有多行), 行号才能定位到行。
- func (h *TimeIndexHandler) findDataPageByPK(handle page.HandleDb, entryPage uint32, pkID uint64) (uint32, bool) {
- if entryPage == 0 {
- return 0, false
- }
- pkKeyBE := pkIDBytes(pkID)
- for curPageID := entryPage; curPageID != 0; {
- buf := make([]byte, page.IndexPageSize)
- if err := handle.ReadPage(curPageID, buf); err != nil {
- break
- }
- hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
- entryLen := int(hdr.EntryLen)
- count := int(hdr.Count)
- if entryLen <= 0 || count == 0 {
- curPageID = hdr.Hdr.NextPage
- continue
- }
- entries := buf[page.IndexHeadSize : page.IndexHeadSize+count*entryLen]
- for i := 0; i < count; i++ {
- dpID := binary.BigEndian.Uint32(entries[i*entryLen+entryLen-4:])
- if dpID == 0 {
- continue
- }
- dp, err := datapage.Load(handle, dpID)
- if err != nil {
- continue
- }
- if _, ok := dp.FindRowByPK(pkKeyBE); ok {
- return dpID, true
- }
- }
- curPageID = hdr.Hdr.NextPage
- }
- return 0, false
- }
- // buildTimeEntry 构造 20 字节时间索引条目 [BeginTime(8) + LastTime(8) + DataPageID(4)]。
- // LastTime 初始等于 BeginTime, 之后随该分段被追加更大的时间戳而增长。
- func buildTimeEntry(indexKey []byte, dpID uint32) []byte {
- entry := make([]byte, IndexTimeEntryLen)
- copy(entry[:8], indexKey[:8])
- copy(entry[8:16], indexKey[:8])
- binary.BigEndian.PutUint32(entry[16:20], dpID)
- return entry
- }
- func (h *TimeIndexHandler) Search(handle page.HandleDb, entryPage uint32, indexKey []byte, keyLen int, resolve RowResolver) ([]byte, bool) {
- // 时序索引自己就是行数据所在, 不需要回表
- dataEntry := h.getDataEntry()
- curPageID := entryPage
- for curPageID != 0 {
- buf := make([]byte, page.IndexPageSize)
- if err := handle.ReadPage(curPageID, buf); err != nil {
- break
- }
- hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
- entries := buf[page.IndexHeadSize : page.IndexHeadSize+int(hdr.Count)*int(hdr.EntryLen)]
- entryLen := int(hdr.EntryLen)
- rowData, found := dataEntry.Search(handle, entries, entryLen, indexKey)
- if found {
- return rowData, true
- }
- curPageID = hdr.Hdr.NextPage
- }
- return nil, false
- }
- func (h *TimeIndexHandler) Delete(handle page.HandleDb, entryPage *uint32, indexKey []byte, pkID uint64, keyLen int) error {
- // 行号已知 → 按行号删除"这一行"。
- //
- // 同一时间戳下允许有多行 (同一时刻的多个测点/多次事件), 只按时间键删只会删掉
- // 该时刻的第一行, 其余行变成"看得见却删不掉"的幽灵数据。行号是每行唯一的,
- // 按它定位才能删到调用方指定的那一行。
- if pkID != 0 {
- dpID, ok := h.findDataPageByPK(handle, *entryPage, pkID)
- if !ok {
- return page.ErrNotFound
- }
- if !h.DeleteDataPageByPK(handle, dpID, pkIDBytes(pkID)) {
- return page.ErrNotFound
- }
- // 有意不回收"整页已删空"的 DataPage: 索引条目仍指向它, 立即还链会让该条目
- // 指向一张可能被别的索引复用的页 (全表扫描会把别人的行当成本表的行读出来)。
- // 空页留在原地无害, 之后往同一分段插入时会被继续复用。
- return nil
- }
- dataEntry := h.getDataEntry()
- curPageID := *entryPage
- for curPageID != 0 {
- buf := make([]byte, page.IndexPageSize)
- if err := handle.ReadPage(curPageID, buf); err != nil {
- return err
- }
- hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
- entries := buf[page.IndexHeadSize : page.IndexHeadSize+int(hdr.Count)*int(hdr.EntryLen)]
- entryLen := int(hdr.EntryLen)
- err := dataEntry.Delete(handle, entries, entryLen, indexKey)
- if err == nil {
- return nil
- }
- curPageID = hdr.Hdr.NextPage
- }
- return page.ErrNotFound
- }
- // IsRowStore 时序索引直接承载行数据
- func (h *TimeIndexHandler) IsRowStore() bool { return true }
- // ResolveRow 遍历索引项链定位各个 TimeDataPage, 按 pkID 取行
- func (h *TimeIndexHandler) ResolveRow(handle page.HandleDb, entryPage uint32, pkID uint64, cache *[]DPRange) ([]byte, bool) {
- pkKeyBE := make([]byte, 8)
- binary.BigEndian.PutUint64(pkKeyBE, pkID)
- curPageID := entryPage
- for curPageID != 0 {
- buf := make([]byte, page.IndexPageSize)
- if err := handle.ReadPage(curPageID, buf); err != nil {
- break
- }
- hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
- entryLen := int(hdr.EntryLen)
- count := int(hdr.Count)
- if entryLen <= 0 || count == 0 {
- curPageID = hdr.Hdr.NextPage
- continue
- }
- entries := buf[page.IndexHeadSize : page.IndexHeadSize+count*entryLen]
- for i := 0; i < count; i++ {
- dpID := binary.BigEndian.Uint32(entries[i*entryLen+entryLen-4:])
- if dpID == 0 {
- continue
- }
- dp, err := datapage.Load(handle, dpID)
- if err != nil {
- continue
- }
- if row, ok := dp.FindRowByPK(pkKeyBE); ok {
- return row, true
- }
- }
- curPageID = hdr.Hdr.NextPage
- }
- return nil, false
- }
- // ResolveMany 批量回表: 见 index.BatchRowResolver 接口契约。
- //
- // 与逐个 ResolveRow (每个 pkID 沿整条分段链走一遍) 不同: 这里沿分段链【一趟】
- // 收集, 每页先用 MinPK/MaxPK 与剩余目标区间预检 (无交则整页跳过), 有交才扫页内行,
- // 行 pkID 在升序目标里二分定位。O(链页数 + 触碰页行数·log N), 对"大表 + 多点回表"
- // 把 O(N·段数·页内扫) 压下来。
- //
- // 刻意不假设页内行序与链序 (乱序时间戳插入会让两者发散): 目标集本身升序,
- // 二分对任意页序都正确。
- //
- // 与主键版同一安全约定: 分批"链走查 + 物化"完成后才调 sink —— 回调里的
- // Update/Delete (段改写/链调整) 不会踩坏正在推进的游标。
- func (h *TimeIndexHandler) ResolveMany(handle page.HandleDb, entryPage uint32,
- sortedPKIDs []uint64, sink func(pkID uint64, storeRow []byte) bool) {
- if entryPage == 0 || len(sortedPKIDs) == 0 {
- return
- }
- type hit struct {
- pkID uint64
- row []byte
- }
- for start := 0; start < len(sortedPKIDs); start += resolveChunk {
- end := start + resolveChunk
- if end > len(sortedPKIDs) {
- end = len(sortedPKIDs)
- }
- wanted := sortedPKIDs[start:end]
- wi := 0 // 下一个未满足的目标下标
- var batch []hit
- curPageID := entryPage
- for curPageID != 0 && wi < len(wanted) {
- buf := make([]byte, page.IndexPageSize)
- if err := handle.ReadPage(curPageID, buf); err != nil {
- break
- }
- hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
- entryLen := int(hdr.EntryLen)
- count := int(hdr.Count)
- if entryLen <= 0 || count == 0 {
- curPageID = hdr.Hdr.NextPage
- continue
- }
- entries := buf[page.IndexHeadSize : page.IndexHeadSize+count*entryLen]
- for i := 0; i < count && wi < len(wanted); i++ {
- dpID := binary.BigEndian.Uint32(entries[i*entryLen+entryLen-4:])
- if dpID == 0 {
- continue
- }
- dp, err := datapage.Load(handle, dpID)
- if err != nil {
- continue
- }
- // Min/Max 预检: 本页与 [wanted[wi], wanted[last]] 无交则整页跳过
- if dp.MaxPK() < wanted[wi] || dp.MinPK() > wanted[len(wanted)-1] {
- continue
- }
- for _, storeRow := range dp.GetAllRows() {
- if len(storeRow) < 8 {
- continue
- }
- pk := binary.BigEndian.Uint64(storeRow[:8])
- if pk < wanted[wi] {
- continue
- }
- // 行 pkID 在剩余目标里二分
- j := sort.Search(len(wanted)-wi, func(k int) bool { return wanted[wi+k] >= pk }) + wi
- if j < len(wanted) && wanted[j] == pk {
- batch = append(batch, hit{pkID: pk, row: storeRow})
- // 该目标已收集, 推进 wi (重复 wanted 由调用方去重)
- wi = j + 1
- }
- }
- }
- curPageID = hdr.Hdr.NextPage
- }
- // 本批走查完成, 现在才触发上层回调
- for _, h := range batch {
- if !sink(h.pkID, h.row) {
- return
- }
- }
- }
- }
- func (h *TimeIndexHandler) Update(handle page.HandleDb, entryPage *uint32, oldKey, newKey, rowData []byte, pkID uint64, keyLen int) error {
- // 行号已知 → 按行号改写"这一行"; 时间键相同的多行因此能各自独立更新。
- if pkID != 0 {
- return h.updateByPK(handle, entryPage, oldKey, newKey, rowData, pkID)
- }
- dataEntry := h.getDataEntry()
- curPageID := *entryPage
- for curPageID != 0 {
- buf := make([]byte, page.IndexPageSize)
- if err := handle.ReadPage(curPageID, buf); err != nil {
- return err
- }
- hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
- entries := buf[page.IndexHeadSize : page.IndexHeadSize+int(hdr.Count)*int(hdr.EntryLen)]
- entryLen := int(hdr.EntryLen)
- err := dataEntry.Update(handle, entries, entryLen, oldKey, pkID, rowData)
- if err == nil {
- return nil
- }
- if errors.Is(err, page.ErrNotFound) {
- curPageID = hdr.Hdr.NextPage
- continue
- }
- return err
- }
- return page.ErrNotFound
- }
- // updateByPK 按行号精确定位并改写一行。
- //
- // 分三种情形:
- // 1. 时间键没变 → 分段内原地改写; 行变长装不下时先在页内紧凑重建;
- // 2. 时间键变了 → 行必须换到新时间段: 按行号删掉旧行, 再用同一个行号重新插入;
- // 3. 页内重建仍装不下 → 同 2, 搬走重插 (不改变行号)。
- func (h *TimeIndexHandler) updateByPK(handle page.HandleDb, entryPage *uint32, oldKey, newKey, rowData []byte, pkID uint64) error {
- if len(newKey) < 8 {
- // 时间索引的键必须是 8 字节时间戳; 取不到键说明行本身不可解析, 不能猜着改。
- return page.ErrNotFound
- }
- pkKeyBE := pkIDBytes(pkID)
- dpID, ok := h.findDataPageByPK(handle, *entryPage, pkID)
- if !ok {
- return page.ErrNotFound
- }
- storeRow := make([]byte, 8+len(rowData))
- copy(storeRow[:8], pkKeyBE)
- copy(storeRow[8:], rowData)
- move := func() error {
- if !h.DeleteDataPageByPK(handle, dpID, pkKeyBE) {
- return page.ErrNotFound
- }
- _, err := h.Insert(handle, entryPage, newKey, pkID, rowData, len(newKey))
- return err
- }
- if !sameTimeKey(oldKey, newKey) {
- // 先验顺序再删旧行: move 是"先删后插", 顺序不合法要在这里就拦下 ——
- // 否则旧行已删、新行被拒, 这一行就凭空消失了。
- if _, err := h.assertTimeOrder(handle, *entryPage, newKey); err != nil {
- return err
- }
- return move()
- }
- if h.UpdateDataPageByPK(handle, dpID, pkKeyBE, storeRow) {
- return nil
- }
- // 原地装不下 (行长变大): 页内重排, 把墓碑的碎片空间收回来。
- if dp, err := datapage.Load(handle, dpID); err == nil {
- if err := dp.RebuildWith(pkKeyBE, storeRow); err == nil {
- return nil
- }
- }
- // 整页都放不下 → 搬到新页, 行号保持不变。
- return move()
- }
- func (h *TimeIndexHandler) GetDataPageIDs(handle page.HandleDb, entryPage uint32) []uint32 {
- if entryPage == 0 {
- return nil
- }
- var ids []uint32
- curPageID := entryPage
- for curPageID != 0 {
- buf := make([]byte, page.IndexPageSize)
- if err := handle.ReadPage(curPageID, buf); err != nil {
- break
- }
- hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
- entryLen := int(hdr.EntryLen)
- count := int(hdr.Count)
- entries := buf[page.IndexHeadSize : page.IndexHeadSize+count*entryLen]
- // 时间索引条目格式: [BeginTime(8B) + LastTime(8B) + DataPageID(4B)]
- for i := 0; i < count; i++ {
- dpID := binary.BigEndian.Uint32(entries[i*entryLen+entryLen-4:])
- if dpID != 0 {
- ids = append(ids, dpID)
- }
- }
- curPageID = hdr.Hdr.NextPage
- }
- return ids
- }
- // CollectPageIDs 汇报时序索引当前占用的全部页: 索引条目链上的每一页, 加上每个
- // 时间段指向的 DataPage (行真正落在这里)。
- //
- // 时序索引的页是"扁平 IndexHead 链", 既不是 B+ 节点也不是哈希桶, 所以自己枚举才准确。
- // 不能指望 btree.CollectAllPageIDs: 它把页当 B+ 节点解析, 这里 IsLeaf=1 只会返回链首
- // 一页 —— 单页可容 817 段, 实测插到 6 万行仍压在一页内, 所以链页尚未见泄漏; 但行数
- // 继续增长 (每页约 400~500 段) 后超出部分就会成为不可达页。
- // 数据页此前是靠 btree.GetDataPageIDs 顺带收回的: 它把 NextPage 当链指针, 又把每个
- // 条目末 4 字节当 DataPageID, 恰好与 [BeginTime/LastTime/DataPageID] 的布局撞对。
- // 这里改为显式调用自己的 GetDataPageIDs, 不再依赖这种巧合。
- func (h *TimeIndexHandler) CollectPageIDs(handle page.HandleDb, entryPage uint32) []uint32 {
- if entryPage == 0 {
- return nil
- }
- var ids []uint32
- seen := make(map[uint32]bool)
- for curPageID := entryPage; curPageID != 0 && !seen[curPageID]; {
- seen[curPageID] = true
- ids = append(ids, curPageID)
- buf := make([]byte, page.IndexPageSize)
- if err := handle.ReadPage(curPageID, buf); err != nil {
- break
- }
- hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
- curPageID = hdr.Hdr.NextPage
- }
- for _, dpID := range h.GetDataPageIDs(handle, entryPage) {
- if dpID != 0 && !seen[dpID] {
- seen[dpID] = true
- ids = append(ids, dpID)
- }
- }
- return ids
- }
- func (h *TimeIndexHandler) ReadDataPageRows(handle page.HandleDb, dpID uint32) [][]byte {
- dp, err := datapage.Load(handle, dpID)
- if err != nil {
- return nil
- }
- return dp.GetAllRows()
- }
- // ReadDataPageLiveCount 返回数据页中未删除的行数 (不物化行)。
- // 与 PrimaryHandler 同款: 时序表的无条件 COUNT(*) 因此走 O(页数) 快路径,
- // 而不是把每页每行读出来只为取 len。
- func (h *TimeIndexHandler) ReadDataPageLiveCount(handle page.HandleDb, dpID uint32) int {
- dp, err := datapage.Load(handle, dpID)
- if err != nil {
- return 0
- }
- return dp.CountLiveRows()
- }
- func (h *TimeIndexHandler) getDataEntry() IDateEntry {
- handler, _ := Compose.GetEntryHandler(page.IndexTypeTime)
- return handler.DataEntry()
- }
- // GetDataPageIDs 已存在于上方 (L145)
- // SearchDataPageByPK 在指定 DataPage 中按 pkID(LE) 查找行
- func (h *TimeIndexHandler) SearchDataPageByPK(handle page.HandleDb, dpID uint32, pkKeyBE []byte) ([]byte, bool) {
- buf := make([]byte, page.IndexPageSize)
- if err := handle.ReadPage(dpID, buf); err != nil {
- return nil, false
- }
- // DataPage 格式: [page.DataPageHdr(40B)] + [len(4B) + rowData]...
- dh := page.DataHdrOf(buf)
- hdrSize := page.DataPageHdrSize
- count := dh.Count
- dataEnd := dh.UseSize
- keyPK := binary.BigEndian.Uint64(pkKeyBE[:8])
- off := hdrSize
- for i := uint32(0); i < count; i++ {
- if off+4 > dataEnd {
- break
- }
- rawLen := binary.LittleEndian.Uint32(buf[off:])
- deleted := rawLen&0x80000000 != 0
- rowLen := rawLen & 0x7FFFFFFF
- off += 4
- if off+rowLen > dataEnd {
- break
- }
- if !deleted && rowLen >= 8 {
- rowPK := binary.BigEndian.Uint64(buf[off : off+8])
- if rowPK == keyPK {
- result := make([]byte, rowLen)
- copy(result, buf[off:off+rowLen])
- return result, true
- }
- }
- off += rowLen
- }
- return nil, false
- }
- // DeleteDataPageByPK 在指定 DataPage 中按 pkID(LE) 软删除行, 返回是否找到
- func (h *TimeIndexHandler) DeleteDataPageByPK(handle page.HandleDb, dpID uint32, pkKeyBE []byte) bool {
- buf := make([]byte, page.IndexPageSize)
- if err := handle.ReadPage(dpID, buf); err != nil {
- return false
- }
- dh := page.DataHdrOf(buf)
- hdrSize := page.DataPageHdrSize
- count := dh.Count
- dataEnd := dh.UseSize
- keyPK := binary.BigEndian.Uint64(pkKeyBE[:8])
- off := hdrSize
- for i := uint32(0); i < count; i++ {
- if off+4 > dataEnd {
- break
- }
- rawLen := binary.LittleEndian.Uint32(buf[off:])
- deleted := rawLen&0x80000000 != 0
- rowLen := rawLen & 0x7FFFFFFF
- off += 4
- if off+rowLen > dataEnd {
- break
- }
- if !deleted && rowLen >= 8 {
- rowPK := binary.BigEndian.Uint64(buf[off : off+8])
- if rowPK == keyPK {
- binary.LittleEndian.PutUint32(buf[off-4:], rowLen|0x80000000)
- // 直写路径绕过 datapage.DataPage, 页头的存活/失效统计在这里同步维护
- dh.LiveCnt--
- dh.Dead += 4 + rowLen
- _ = handle.SavePage(dpID, buf)
- return true
- }
- }
- off += rowLen
- }
- return false
- }
- // UpdateDataPageByPK 在指定 DataPage 中按 pkID(LE) 原地更新行, 返回是否找到
- func (h *TimeIndexHandler) UpdateDataPageByPK(handle page.HandleDb, dpID uint32, pkKeyBE []byte, newStoreRow []byte) bool {
- buf := make([]byte, page.IndexPageSize)
- if err := handle.ReadPage(dpID, buf); err != nil {
- return false
- }
- dh := page.DataHdrOf(buf)
- hdrSize := page.DataPageHdrSize
- count := dh.Count
- dataEnd := dh.UseSize
- keyPK := binary.BigEndian.Uint64(pkKeyBE[:8])
- off := hdrSize
- for i := uint32(0); i < count; i++ {
- if off+4 > dataEnd {
- break
- }
- rawLen := binary.LittleEndian.Uint32(buf[off:])
- deleted := rawLen&0x80000000 != 0
- rowLen := rawLen & 0x7FFFFFFF
- off += 4
- if off+rowLen > dataEnd {
- break
- }
- if !deleted && rowLen >= 8 {
- rowPK := binary.BigEndian.Uint64(buf[off : off+8])
- if rowPK == keyPK {
- if uint32(len(newStoreRow)) <= rowLen {
- copy(buf[off:], newStoreRow)
- for j := uint32(len(newStoreRow)); j < rowLen; j++ {
- buf[off+j] = 0
- }
- _ = handle.SavePage(dpID, buf)
- return true
- }
- return false // row too large
- }
- }
- off += rowLen
- }
- return false
- }
- // RangeScan 时序索引范围扫描
- // startKey/endKey: BigEndian 编码的时间戳
- // 遍历 IndexPage 条目链 对时间范围与查询范围重叠的条目加载 DataPage 并回调每行
- func (h *TimeIndexHandler) RangeScan(handle page.HandleDb, entryPage uint32,
- startKey, endKey []byte, startInclusive, endInclusive bool, reverse bool,
- callback func(row []byte) bool) error {
- if entryPage == 0 {
- return nil
- }
- // 收集所有索引页数据
- type pageInfo struct {
- entries []byte
- entryLen int
- count int
- }
- var pages []pageInfo
- curPageID := entryPage
- for curPageID != 0 {
- buf := make([]byte, page.IndexPageSize)
- if err := handle.ReadPage(curPageID, buf); err != nil {
- break
- }
- hdr := (*page.IndexHead)(unsafe.Pointer(&buf[0]))
- entryLen := int(hdr.EntryLen)
- count := int(hdr.Count)
- entries := make([]byte, count*entryLen)
- copy(entries, buf[page.IndexHeadSize:page.IndexHeadSize+count*entryLen])
- pages = append(pages, pageInfo{entries: entries, entryLen: entryLen, count: count})
- curPageID = hdr.Hdr.NextPage
- }
- // 确定遍历顺序
- pageStart, pageEnd, pageStep := 0, len(pages)-1, 1
- if reverse {
- pageStart, pageEnd, pageStep = len(pages)-1, 0, -1
- }
- visited := make(map[uint32]bool)
- for p := pageStart; ; p += pageStep {
- if pageStep > 0 && p > pageEnd {
- break
- }
- if pageStep < 0 && p < pageEnd {
- break
- }
- pg := &pages[p]
- if pg.entryLen < IndexTimeEntryLen {
- continue // 防御: 旧库/损坏页的条目长度不足 20B, 跳过以免越界
- }
- entryStart, entryEnd, entryStep := 0, pg.count-1, 1
- if reverse {
- entryStart, entryEnd, entryStep = pg.count-1, 0, -1
- }
- for i := entryStart; ; i += entryStep {
- if entryStep > 0 && i > entryEnd {
- break
- }
- if entryStep < 0 && i < entryEnd {
- break
- }
- off := i * pg.entryLen
- beginTime := pg.entries[off : off+8]
- lastTime := pg.entries[off+8 : off+16]
- dpID := binary.BigEndian.Uint32(pg.entries[off+pg.entryLen-4:])
- if dpID == 0 || visited[dpID] {
- continue
- }
- if endKey != nil {
- cmp := tiCmp(beginTime, endKey)
- if cmp > 0 || (cmp == 0 && !endInclusive) {
- continue
- }
- }
- if startKey != nil {
- cmp := tiCmp(lastTime, startKey)
- if cmp < 0 || (cmp == 0 && !startInclusive) {
- continue
- }
- }
- visited[dpID] = true
- dp, dpErr := datapage.Load(handle, dpID)
- if dpErr != nil {
- continue
- }
- rows := dp.GetAllRows()
- if !reverse {
- for _, row := range rows {
- if !timeRowInRange(row, startKey, endKey, startInclusive, endInclusive) {
- continue
- }
- if !callback(row) {
- return nil
- }
- }
- } else {
- for k := len(rows) - 1; k >= 0; k-- {
- if !timeRowInRange(rows[k], startKey, endKey, startInclusive, endInclusive) {
- continue
- }
- if !callback(rows[k]) {
- return nil
- }
- }
- }
- }
- }
- return nil
- }
- // timeRowInRange 判断时序数据行的时间是否落在 [startKey, endKey] 区间内。
- // 行内布局: [pkID(8B)] + [schemaVer(1B)] + 字段..., 时间键位于偏移 9(8B BE)。
- // 由于分段 DataPage 覆盖一段时间跨度, 段内可能混有区间外的点, 必须逐行过滤,
- // 否则范围查询会把整段的点都返回 (含越界点)。
- func timeRowInRange(row, startKey, endKey []byte, startInclusive, endInclusive bool) bool {
- const tsOff = 9
- if len(row) < tsOff+8 {
- // 行过短无法解析时间, 保守返回 true, 交由上层判断
- return true
- }
- if startKey == nil && endKey == nil {
- return true
- }
- ts := row[tsOff : tsOff+8]
- if startKey != nil {
- cmp := tiCmp(ts, startKey)
- if cmp < 0 || (cmp == 0 && !startInclusive) {
- return false
- }
- }
- if endKey != nil {
- cmp := tiCmp(ts, endKey)
- if cmp > 0 || (cmp == 0 && !endInclusive) {
- return false
- }
- }
- return true
- }
- // tiCmp 字节切片比较
- func tiCmp(a, b []byte) int {
- ml := len(a)
- if len(b) < ml {
- ml = len(b)
- }
- for i := 0; i < ml; i++ {
- if a[i] < b[i] {
- return -1
- }
- if a[i] > b[i] {
- return 1
- }
- }
- if len(a) < len(b) {
- return -1
- }
- if len(a) > len(b) {
- return 1
- }
- return 0
- }
|