TsDB2/DB/core/index/compose_handler.go

package index

import (
	"TsDB2/DB/core/index/composepage"
	"TsDB2/DB/core/page"
	"TsDB2/DB/core/util"
	"encoding/binary"
	"unsafe"
)

func init() {
	Compose.RegisterHandler(page.IndexTypeCompose, &ComposeChainHandler{})
}

// ======================== IEntry: 澶嶅悎绱㈠紩鏉＄洰 ========================

// ComposeEntry 閫氱敤澶嶅悎绱㈠紩缁撴瀯浣?
// Keys: 瀛樺偍 N 涓储寮曢敭 (uint64)
// PKID: 涓婚敭ID锛?瀛楄妭锛夛紝閫氳繃 B+ 鏍戣В鏋愭暟鎹綅缃?
type ComposeEntry struct {
	Keys []uint64 // 浠绘剰澶氫釜绱㈠紩閿紙1涓€?涓€?涓?..閮芥敮鎸侊級
	PKID uint64   // 涓婚敭ID (8瀛楄妭)
}

func (e *ComposeEntry) DataEntry() IDateEntry {
	return &composepage.ComposeEntryPage{}
}

// EntryLength 鑷姩璁＄畻闀垮害锛?*閿暟閲?+ 8 (PK_ID)
func (e *ComposeEntry) EntryLength() uint16 {
	return uint16(len(e.Keys)*8 + 8)
}

// Compare 閫氱敤澶嶅悎绱㈠紩姣旇緝閫昏緫锛氳嚜鍔ㄦ寜椤哄簭姣旇緝鎵€鏈夐敭锛堜笉鍚湯灏?PK_ID锛?
func (e *ComposeEntry) Compare(midEntry []byte, key []byte) int {
	entryLen := int(e.EntryLength())
	keyCount := (entryLen - 8) / 8
	for i := 0; i < keyCount; i++ {
		offset := i * 8
		f1 := binary.BigEndian.Uint64(midEntry[offset : offset+8])
		f2 := binary.BigEndian.Uint64(key[offset : offset+8])
		cmp := util.CompareUint64(f1, f2)
		if cmp != 0 {
			return cmp
		}
	}
	return 0
}

// Serialize 鑷姩搴忓垪鍖?N 涓敭 + PK_ID
func (e *ComposeEntry) Serialize(ie IEntry) []byte {
	ent, ok := ie.(*ComposeEntry)
	if !ok {
		return nil
	}
	keyCount := len(ent.Keys)
	buf := make([]byte, keyCount*8+8)
	for i, k := range ent.Keys {
		binary.BigEndian.PutUint64(buf[i*8:(i+1)*8], k)
	}
	binary.BigEndian.PutUint64(buf[keyCount*8:], ent.PKID)
	return buf
}

// Deserialize 鑷姩鍙嶅簭鍒楀寲 N 涓敭 + PK_ID
func (e *ComposeEntry) Deserialize(buf []byte) IEntry {
	keyCount := (len(buf) - 8) / 8
	keys := make([]uint64, 0, keyCount)
	for i := 0; i < keyCount; i++ {
		keys = append(keys, binary.BigEndian.Uint64(buf[i*8:(i+1)*8]))
	}
	return &ComposeEntry{
		Keys: keys,
		PKID: binary.BigEndian.Uint64(buf[keyCount*8:]),
	}
}

// Insert 鎻掑叆绱㈠紩
func (e *ComposeEntry) Insert(entries *[]byte, data []byte) error {
	*entries = append(*entries, data...)
	return nil
}

// Search 浜屽垎鏌ユ壘锛堥€氱敤 N 瀛楁澶嶅悎绱㈠紩锛?
func (e *ComposeEntry) Search(entries []byte, key []byte) (uint32, bool) {
	entryLen := int(e.EntryLength())
	pos, found := util.BinarySearch(entries, entryLen, key, e.Compare)
	if !found {
		return 0, false
	}
	return pos, true
}

// ======================== IndexHandler: 澶嶅悎浜岀骇绱㈠紩澶勭悊鍣?========================

// ComposeChainHandler 鈥?澶嶅悎浜岀骇绱㈠紩 (page.IndexTypeCompose)
// 鎵佸钩绱㈠紩椤甸摼, 鏉＄洰 [composeKey(N*8B) + PKID(8B)]
type ComposeChainHandler struct{}

func (h *ComposeChainHandler) Insert(handle page.HandleDb, entryPage *uint32, indexKey []byte, pkID uint64, rowData []byte, keyLen int) (uint32, error) {
	dataEntry := h.getDataEntry()
	curPageID := *entryPage
	lastIdxPageID := uint32(0)
	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
		if !idxPageHasRoom {
			lastIdxPageID = curPageID
			curPageID = hdr.Hdr.NextPage
			continue
		}
		entries := buf[page.IndexHeadSize : page.IndexHeadSize+(count+1)*pageEntryLen]
		err := dataEntry.Insert(handle, entries, pageEntryLen, count, indexKey, pkID, rowData)
		if err == nil {
			hdr.Count++
			hdr.Hdr.UseSize += uint32(pageEntryLen)
			_ = handle.SavePage(curPageID, buf)
			return *entryPage, nil
		}
		lastIdxPageID = curPageID
		curPageID = hdr.Hdr.NextPage
	}
	entryLen := keyLen + 8
	newEntry := make([]byte, entryLen)
	copy(newEntry[:keyLen], indexKey)
	binary.BigEndian.PutUint64(newEntry[keyLen:], pkID)
	return *entryPage, writeToNewIndexPage(handle, entryPage, newEntry, lastIdxPageID)
}

func (h *ComposeChainHandler) 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 *ComposeChainHandler) Delete(handle page.HandleDb, entryPage *uint32, indexKey []byte, keyLen int) error {
	return h.deleteEntry(handle, entryPage, indexKey)
}

func (h *ComposeChainHandler) Update(handle page.HandleDb, entryPage *uint32, oldKey []byte, rowData []byte, pkID uint64, keyLen int) error {
	// 浜岀骇绱㈠紩鏇存柊鐢?IndexMetaPage.UpdateByKey 缁熶竴澶勭悊
	return page.ErrNotFound
}

func (h *ComposeChainHandler) GetDataPageIDs(handle page.HandleDb, entryPage uint32) []uint32 {
	return nil
}

func (h *ComposeChainHandler) ReadDataPageRows(handle page.HandleDb, dpID uint32) [][]byte {
	return nil
}

// RangeScan 澶嶅悎浜岀骇绱㈠紩鑼冨洿鎵弿
// 鏉＄洰鏍煎紡: [key(keyLen) + pkID(8B BE)]
// 鍥炶皟鏀跺埌 8B BigEndian pkID
func (h *ComposeChainHandler) 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 pageData struct {
		entries []byte
		el      int
		count   int
	}
	var pages []pageData
	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]))
		el := int(hdr.EntryLen)
		count := int(hdr.Count)
		entries := make([]byte, count*el)
		copy(entries, buf[page.IndexHeadSize:page.IndexHeadSize+count*el])
		pages = append(pages, pageData{entries: entries, el: el, count: count})
		curPageID = hdr.Hdr.NextPage
	}
	if !reverse {
		for p := 0; p < len(pages); p++ {
			pg := &pages[p]
			for i := 0; i < pg.count; i++ {
				entry := pg.entries[i*pg.el : (i+1)*pg.el]
				key := entry[:pg.el-8]
				if endKey != nil {
					cmp := composeCmp(key, endKey)
					if cmp > 0 || (cmp == 0 && !endInclusive) {
						return nil
					}
				}
				if startKey != nil {
					cmp := composeCmp(key, startKey)
					if cmp < 0 || (cmp == 0 && !startInclusive) {
						continue
					}
				}
				pkID := entry[pg.el-8:]
				if !callback(pkID) {
					return nil
				}
			}
		}
	} else {
		for p := len(pages) - 1; p >= 0; p-- {
			pg := &pages[p]
			for i := pg.count - 1; i >= 0; i-- {
				entry := pg.entries[i*pg.el : (i+1)*pg.el]
				key := entry[:pg.el-8]
				if startKey != nil {
					cmp := composeCmp(key, startKey)
					if cmp < 0 || (cmp == 0 && !startInclusive) {
						return nil
					}
				}
				if endKey != nil {
					cmp := composeCmp(key, endKey)
					if cmp > 0 || (cmp == 0 && !endInclusive) {
						continue
					}
				}
				pkID := entry[pg.el-8:]
				if !callback(pkID) {
					return nil
				}
			}
		}
	}
	return nil
}

// composeCmp 瀛楄妭鍒囩墖姣旇緝
func composeCmp(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
}

func (h *ComposeChainHandler) deleteEntry(handle page.HandleDb, entryPage *uint32, indexKey []byte) 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 {
			hdr.Hdr.UseSize -= uint32(entryLen)
			_ = handle.SavePage(curPageID, buf)
			return nil
		}
		curPageID = hdr.Hdr.NextPage
	}
	return page.ErrNotFound
}

func (h *ComposeChainHandler) getDataEntry() IDateEntry {
	handler, _ := Compose.GetEntryHandler(page.IndexTypeCompose)
	return handler.DataEntry()
}