TsDB2/DB/core/index/sbt_handler.go

package index

import (
	"bytes"
	"encoding/binary"
	"sort"

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

// sbtSecondary — SBT 二级索引处理器基座: BTreeHandler / ComposeChainHandler 共享的
// 全部 IndexHandler 方法与批量写入路径。
//
// 两个二级索引处理器对 IndexHandler 的实现逐字等价 (条目都是 [key(keyLen) +
// PKID(8B BE)], 全部委托 btree.SBT*), 于是收敛到这一份嵌入实现; 各类型只保留
// 自己的索引类型语义与独有方法 (如 BTreeHandler.SearchAll)。
type sbtSecondary struct{}

// 嵌入基座后两个处理器的方法集在编译期锁定: IndexHandler 全数命中,
// BatchInserter 随之命中 —— IndexMetaPage.InsertRows 的批量分发对复合索引同样生效。
var (
	_ IndexHandler  = (*sbtSecondary)(nil)
	_ BatchInserter = (*sbtSecondary)(nil)
)

func (h *sbtSecondary) 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
}

// InsertBatch 批量写入二级索引条目。
//
// 先把整批条目按复合键 (key, pkID) 排序, 再交给 SBTBulkInsert 做"顺序扫描式"
// 灌叶 —— 排序批量对 B+ 树就是最优写入形态: 相邻条目几乎总落同一张叶, 树被
// 自左向右扫一遍, 每张叶加载/落盘各一次; 而逐条路径是每条从根随机下降 + 整页
// 回写。IndexKey 为 nil 的行不进本索引 (NULL 分量), 直接跳过。
func (h *sbtSecondary) InsertBatch(handle page.HandleDb, entryPage *uint32, rows []BatchRow) error {
	// 收集非 nil 键的条目 (定长 [key|pkID], 键长取首个非 nil)
	n := 0
	keyLen := 0
	for i := range rows {
		if rows[i].IndexKey != nil {
			if n == 0 {
				keyLen = len(rows[i].IndexKey)
			}
			n++
		}
	}
	if n == 0 {
		return nil
	}
	el := keyLen + 8
	flat := make([]byte, n*el)
	order := make([]int, 0, n)
	idx := 0
	for i := range rows {
		r := &rows[i]
		if r.IndexKey == nil {
			continue
		}
		e := flat[idx*el : idx*el+el]
		copy(e[:keyLen], r.IndexKey)
		binary.BigEndian.PutUint64(e[keyLen:], r.PKID)
		order = append(order, idx)
		idx++
	}
	// 定长条目 [key|pkID] 的字节序 == 复合键 (key, pkID) 全序, 直接按字节排序
	sort.Slice(order, func(a, b int) bool {
		return bytes.Compare(flat[order[a]*el:order[a]*el+el], flat[order[b]*el:order[b]*el+el]) < 0
	})
	sortedFlat := make([]byte, n*el)
	for j, oi := range order {
		copy(sortedFlat[j*el:j*el+el], flat[oi*el:oi*el+el])
	}

	rootPID := *entryPage
	newRoot, err := btree.SBTBulkInsert(rootPID, sortedFlat, el, handle)
	if err != nil {
		return err
	}
	if newRoot != *entryPage {
		*entryPage = newRoot
	}
	return nil
}

func (h *sbtSecondary) 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 *sbtSecondary) Delete(handle page.HandleDb, entryPage *uint32, indexKey []byte, pkID uint64, keyLen int) error {
	rootPID := *entryPage
	if rootPID == 0 {
		return page.ErrNotFound
	}
	// 二级索引允许重复 key, 用 (key, pkID) 精确定位本行的条目。
	// pkID 未知(为 0)时退回按 key 搜索一个条目。
	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 *sbtSecondary) Update(handle page.HandleDb, entryPage *uint32, oldKey, newKey, rowData []byte, pkID uint64, keyLen int) error {
	// 二级索引条目只存 [key + pkID], 不含行数据:
	//   键未变 → 无事可做 (过去"先删再插"会让条目在重复 key 区段里错位, 属于多余动作)
	//   键变了 → 按 (oldKey, pkID) 精确删旧条目, 再插新条目
	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
}

// IsRowStore 二级索引不承载行数据
func (h *sbtSecondary) IsRowStore() bool { return false }

// ResolveRow 二级索引无法按 pkID 取行, 由调用方的回表函数负责
func (h *sbtSecondary) ResolveRow(handle page.HandleDb, entryPage uint32, pkID uint64, cache *[]DPRange) ([]byte, bool) {
	return nil, false
}

func (h *sbtSecondary) GetDataPageIDs(handle page.HandleDb, entryPage uint32) []uint32 {
	// 二级索引没有 DataPage, 返回空
	return nil
}

func (h *sbtSecondary) ReadDataPageRows(handle page.HandleDb, dpID uint32) [][]byte {
	// 二级索引没有 DataPage
	return nil
}

func (h *sbtSecondary) 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)
}