TsDB2/DB/core/index/time_handler.go

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 瑙ｆ瀽缁撴瀯浣?
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
}