TsDB2/DB/core/index/compose_handler.go

package index

import (
	"bytes"
	"encoding/binary"

	"git.ycl.pub/yl32940/TsDB2/DB/core/index/btree"
	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
)

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

// ======================== IndexHandler: 复合二级索引处理器 ========================

// ComposeChainHandler — 复合二级索引 (page.IndexTypeCompose)
// 底层即 B+ 树二级索引 (与 BTreeHandler 共享 btree.SBT* 实现):
// Insert/Search/Delete/RangeScan 全部委托 SBT, 条目格式 [key(keyLen) + PKID(8B BE)],
// keyLen = 各分量按真实宽度定长拼接的总长 (分量不限 8 字节, 与 BTree 复合键一致)。
type ComposeChainHandler struct{}

func (h *ComposeChainHandler) Insert(handle page.HandleDb, entryPage *uint32, indexKey []byte, pkID uint64, rowData []byte, keyLen int) (uint32, error) {
	entry := make([]byte, keyLen+8)
	copy(entry[:keyLen], indexKey)
	binary.BigEndian.PutUint64(entry[keyLen:], pkID)
	rootPID := *entryPage
	newRoot, err := btree.SBTInsert(rootPID, entry, handle)
	if err != nil {
		return 0, err
	}
	*entryPage = newRoot
	return newRoot, nil
}

func (h *ComposeChainHandler) Search(handle page.HandleDb, entryPage uint32, indexKey []byte, keyLen int, resolve RowResolver) ([]byte, bool) {
	rootPID := entryPage
	if rootPID == 0 {
		return nil, false
	}
	pkID, found := btree.SBTSearch(rootPID, indexKey[:keyLen], handle)
	if !found {
		return nil, false
	}
	if resolve != nil {
		if row, ok := resolve(pkID); ok {
			return row, true
		}
		return nil, false
	}
	pkBuf := make([]byte, 8)
	binary.BigEndian.PutUint64(pkBuf, pkID)
	return pkBuf, true
}

func (h *ComposeChainHandler) Delete(handle page.HandleDb, entryPage *uint32, indexKey []byte, pkID uint64, keyLen int) error {
	rootPID := *entryPage
	if rootPID == 0 {
		return page.ErrNotFound
	}
	if pkID == 0 {
		pk, found := btree.SBTSearch(rootPID, indexKey[:keyLen], handle)
		if !found {
			return page.ErrNotFound
		}
		pkID = pk
	}
	newRoot, err := btree.SBTDeleteWithPKID(rootPID, indexKey[:keyLen], pkID, handle)
	if err != nil {
		return err
	}
	*entryPage = newRoot
	return nil
}

func (h *ComposeChainHandler) Update(handle page.HandleDb, entryPage *uint32, oldKey, newKey, rowData []byte, pkID uint64, keyLen int) error {
	if bytes.Equal(oldKey, newKey) {
		return nil
	}
	if oldKey != nil {
		if err := h.Delete(handle, entryPage, oldKey, pkID, len(oldKey)); err != nil && err != page.ErrNotFound {
			return err
		}
	}
	_, err := h.Insert(handle, entryPage, newKey, pkID, rowData, len(newKey))
	return err
}

func (h *ComposeChainHandler) IsRowStore() bool { return false }

func (h *ComposeChainHandler) ResolveRow(handle page.HandleDb, entryPage uint32, pkID uint64, cache *[]DPRange) ([]byte, bool) {
	return nil, false
}

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

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

func (h *ComposeChainHandler) RangeScan(handle page.HandleDb, entryPage uint32,
	startKey, endKey []byte, startInclusive, endInclusive bool, reverse bool,
	callback func(row []byte) bool) error {
	return btree.SBTRangeScan(entryPage, handle, startKey, endKey, startInclusive, endInclusive, reverse, callback)
}