package index import ( "TsDB2/DB/core/index/datapage" timeindex "TsDB2/DB/core/index/timedatapage" "TsDB2/DB/core/page" "TsDB2/DB/core/util" "encoding/binary" "errors" "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 } 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() } 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) _ = 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 }