package index import ( "TsDB2/DB/core/index/datapage" timeindex "TsDB2/DB/core/index/timedatapage" "TsDB2/DB/core/page" "TsDB2/DB/core/util" "encoding/binary" "strings" "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 瑙f瀽缁撴瀯浣? 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) { dataEntry := h.getDataEntry() curPageID := *entryPage 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 { return *entryPage, nil } if strings.Contains(err.Error(), "need new index entry") { dpID := binary.BigEndian.Uint32(entries[count*pageEntryLen : count*pageEntryLen+4]) if idxPageHasRoom { newEntry := make([]byte, pageEntryLen) copy(newEntry, indexKey) binary.BigEndian.PutUint32(newEntry[pageEntryLen-4:], dpID) insertEntrySorted(buf, hdr, newEntry) _ = handle.SavePage(curPageID, buf) return *entryPage, nil } newEntry := make([]byte, pageEntryLen) copy(newEntry, indexKey) binary.BigEndian.PutUint32(newEntry[pageEntryLen-4:], dpID) return *entryPage, writeToNewIndexPage(handle, entryPage, newEntry, curPageID) } curPageID = hdr.Hdr.NextPage } newDPPID, createErr := dataEntry.CreateDataPage(handle, indexKey, rowData) if createErr != nil { return 0, createErr } entryLen := keyLen + 4 newEntry := make([]byte, entryLen) copy(newEntry, indexKey) binary.BigEndian.PutUint32(newEntry[entryLen-4:], newDPPID) return *entryPage, writeToNewIndexPage(handle, entryPage, newEntry, 0) } func (h *TimeIndexHandler) Search(handle page.HandleDb, entryPage uint32, indexKey []byte, keyLen int) ([]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, keyLen int) error { 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 } func (h *TimeIndexHandler) Update(handle page.HandleDb, entryPage *uint32, oldKey []byte, rowData []byte, pkID uint64, keyLen int) error { 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 strings.Contains(err.Error(), "not found") { curPageID = hdr.Hdr.NextPage continue } return err } return page.ErrNotFound } 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 } 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 鏍煎紡: [PageHdr(18B) + Count(4B) + Min(8B) + Max(8B)] + [len(4B) + rowData]... hdrSize := uint32(unsafe.Sizeof(page.PageHdr{})) + 4 + 8 + 8 count := *(*uint32)(unsafe.Pointer(&buf[unsafe.Sizeof(page.PageHdr{})])) dataEnd := (*page.PageHdr)(unsafe.Pointer(&buf[0])).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 } hdrSize := uint32(unsafe.Sizeof(page.PageHdr{})) + 4 + 8 + 8 count := *(*uint32)(unsafe.Pointer(&buf[unsafe.Sizeof(page.PageHdr{})])) dataEnd := (*page.PageHdr)(unsafe.Pointer(&buf[0])).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 } hdrSize := uint32(unsafe.Sizeof(page.PageHdr{})) + 4 + 8 + 8 count := *(*uint32)(unsafe.Pointer(&buf[unsafe.Sizeof(page.PageHdr{})])) dataEnd := (*page.PageHdr)(unsafe.Pointer(&buf[0])).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] 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 !callback(row) { return nil } } } else { for k := len(rows) - 1; k >= 0; k-- { if !callback(rows[k]) { return nil } } } } } return nil } // 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 }